Kimi K3 接入 飞书 完整指南：长文本群问答 + 卡片流式更新双路径实测

1长文部分：10 × (30,000/1M × 2 + 800/1M × 12) = 10 × (0.06 + 0.0096) = ¥0.696/天
2短文部分：50 × (400/1M × 1 + 500/1M × 4)     = 50 × (0.0004 + 0.002) = ¥0.12/天
3合计：约 ¥0.82/天，约 ¥25/月

1❌ 短平快的高频补全类任务（用 k3-8k 或别的便宜模型更划算）
2❌ 需要严格 JSON Schema 给下游消费（建议加后处理校验 + 失败重试）
3❌ 实时信息问答（模型知识有截止，需要自己接检索）
4❌ 超大文件批量处理（单文件超 100MB 或几百页时，自己做切片并行更稳）

1Webhook 地址：
2https://open.feishu.cn/open-apis/bot/v2/hook/7d6c5b4a-3928-1706-5f4e-3d2c1b0a9f8e
3
4密钥（secret）：
5Kp3Lm7Xn2QrT9Yb

1import hmac, hashlib, time
2
3def gen_sign(secret: str) -> tuple[str, str]:
4    """飞书自定义机器人加签。三个易错点：
5    1) timestamp 用秒，不要 int(time.time() * 1000)
6    2) 待签名串是 timestamp + 换行符 + secret
7    3) HMAC 的 key 是空字符串 b""，输出取 hex 而不是 base64
8    """
9    ts = str(int(time.time()))
10    sign = hmac.new(b"", f"{ts}\n{secret}".encode("utf-8"), hashlib.sha256).hexdigest()
11    return ts, sign

1import httpx
2
3WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/7d6c5b4a-3928-1706-5f4e-3d2c1b0a9f8e"
4SECRET = "Kp3Lm7Xn2QrT9Yb"
5
6def push(msg_type: str, content: dict) -> dict:
7    ts, sign = gen_sign(SECRET)
8    body = {"timestamp": ts, "sign": sign, "msg_type": msg_type, "content": content}
9    return httpx.post(WEBHOOK, json=body, timeout=10).json()
10    # 成功 {"code":0,"msg":"success"}；失败 {"code":19021,"msg":"sign match fail..."}
11
12print(push("text", {"text": "Kimi K3 接入测试"}))

1import crypto from 'node:crypto';
2
3const WEBHOOK = 'https://open.feishu.cn/open-apis/bot/v2/hook/7d6c5b4a-3928-1706-5f4e-3d2c1b0a9f8e';
4const SECRET = 'Kp3Lm7Xn2QrT9Yb';
5
6function genSign(secret) {
7  const ts = String(Math.floor(Date.now() / 1000));   // 秒级！不要 Date.now()
8  const sign = crypto
9    .createHmac('sha256', Buffer.alloc(0))             // 空 Buffer 作密钥
10    .update(`${ts}\n${secret}`)
11    .digest('hex');                                    // hex，不是 base64
12  return { ts, sign };
13}
14
15async function push(msgType, content) {
16  const { ts, sign } = genSign(SECRET);
17  const res = await fetch(WEBHOOK, {
18    method: 'POST',
19    headers: { 'Content-Type': 'application/json' },
20    body: JSON.stringify({ timestamp: ts, sign, msg_type: msgType, content }),
21  });
22  return res.json();
23}

1# mac / linux
2echo 'export MOONSHOT_API_KEY="sk-moonshot-xxxxxxxxxxxxxxxx"' >> ~/.bashrc && source ~/.bashrc
3
4# windows powershell
5setx MOONSHOT_API_KEY "sk-moonshot-xxxxxxxxxxxxxxxx"

1Base URL：https://api.moonshot.cn/v1
2Model：   kimi-k3（另有 kimi-k3-thinking / kimi-k3-8k）

1feishu-kimi-bot/
2├── .env          # 密钥，务必加进 .gitignore
3├── .gitignore    # 至少一行：.env
4├── bot.py
5└── card.py

1import os, json, time, hmac, hashlib, httpx
2
3WEBHOOK = os.environ["FEISHU_WEBHOOK"]
4SECRET  = os.environ["FEISHU_SECRET"]
5MS_KEY  = os.environ["MOONSHOT_API_KEY"]
6MS_URL  = "https://api.moonshot.cn/v1/chat/completions"
7
8SYSTEM_PROMPT = """你是飞书群里的长文助手，底层模型是 Kimi K3。
9回答要求：
101. 中文，结构化，控制在 400 字以内，能用列表就用列表
112. 回答文档相关问题时必须标注出处（章节号或页码）
123. 文档里没有的内容直接说「文档中未提及」，不要编造
134. 用户要求对比时，用 Markdown 表格呈现
14"""
15
16def ask_kimi(question: str, model: str = "kimi-k3", file_id: str | None = None,
17             timeout: int = 60) -> tuple[str, int]:
18    """返回 (回答, 总 token 消耗)。file_id 为上传后的文档引用，可为空。"""
19    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
20    if file_id:   # 引用已上传文件：内容不进 messages，省一大笔输入 tokens
21        messages.append({"role": "system",
22                         "content": {"type": "file", "file_id": file_id}})
23    messages.append({"role": "user", "content": question})
24
25    r = httpx.post(
26        MS_URL,
27        headers={"Authorization": f"Bearer {MS_KEY}"},
28        json={"model": model, "messages": messages,
29              "temperature": 0.3, "max_tokens": 1024},
30        timeout=timeout,        # 长文场景必须放宽，默认 10 秒一定超时
31    )
32    data = r.json()
33    if "choices" not in data:
34        raise RuntimeError(f"Kimi 调用失败：{data}")
35    return data["choices"][0]["message"]["content"], data["usage"]["total_tokens"]
36
37def pick_model(question: str, has_file: bool) -> str:
38    """路由：推理类问题走 thinking，长文走 k3，短问答走低价版"""
39    if any(k in question for k in ["为什么", "推算", "是否合理", "矛盾", "风险"]):
40        return "kimi-k3-thinking"
41    if has_file or len(question) > 200:
42        return "kimi-k3"
43    return "kimi-k3-8k"
44
45def gen_sign(secret: str) -> tuple[str, str]:
46    ts = str(int(time.time()))
47    return ts, hmac.new(b"", f"{ts}\n{secret}".encode(), hashlib.sha256).hexdigest()
48
49def push(msg_type: str, content: dict) -> dict:
50    ts, sign = gen_sign(SECRET)
51    return httpx.post(WEBHOOK, json={
52        "timestamp": ts, "sign": sign, "msg_type": msg_type, "content": content
53    }, timeout=10).json()
54
55if __name__ == "__main__":
56    q = "这份需求文档里，支付模块有哪些前置依赖？"
57    model = pick_model(q, has_file=False)
58    answer, tokens = ask_kimi(q, model)
59    print(model, push("text", {"text": f"{answer}\n\n（{model} · {tokens} tokens）"}))

1pip install httpx python-dotenv
2python bot.py
3# 控制台打印 {'code': 0, 'msg': 'success', 'data': {}} 即链路打通

1import httpx
2
3def upload_file(path: str, purpose: str = "file-extract") -> str:
4    """上传到 Kimi 文件接口，返回 file_id。
5    支持 pdf / docx / xlsx / pptx / txt / md 等，单文件建议控制在 100MB 内。
6    """
7    with open(path, "rb") as f:
8        r = httpx.post(
9            "https://api.moonshot.cn/v1/files",
10            headers={"Authorization": f"Bearer {MS_KEY}"},
11            files={"file": (os.path.basename(path), f)},
12            data={"purpose": purpose},
13            timeout=120,
14        )
15    data = r.json()
16    return data["id"]        # 形如 cq1x2y3z4-xxxx-xxxx

1fid = upload_file("./季度产品需求文档.pdf")
2q = "支付模块有哪些前置依赖？请列出依赖方和当前状态，用表格输出。"
3answer, tokens = ask_kimi(q, model="kimi-k3", file_id=fid)

1def download_feishu_file(message_id: str, file_key: str) -> bytes:
2    """用飞书接口下载用户发来的文件"""
3    token = get_tenant_access_token()
4    r = httpx.get(
5        f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}/resources/{file_key}",
6        params={"type": "file"},
7        headers={"Authorization": f"Bearer {token}"},
8        timeout=60,
9    )
10    return r.content      # 二进制内容，落到临时文件后交给 upload_file
11
12def handle_file_question(message_id: str, file_key: str, question: str) -> str:
13    raw = download_feishu_file(message_id, file_key)
14    tmp = f"/tmp/{file_key}.pdf"
15    with open(tmp, "wb") as f:
16        f.write(raw)
17    fid = upload_file(tmp)                       # 转存 Kimi，拿 file_id
18    cache[message_id] = fid                      # 缓存，后续追问直接复用
19    answer, _ = ask_kimi(question, "kimi-k3", file_id=fid)
20    return answer

1{
2  "msg_type": "interactive",
3  "card": {
4    "config": { "wide_screen_mode": true, "enable_forward": true },
5    "header": {
6      "template": "turquoise",
7      "title": { "tag": "plain_text", "content": "Kimi K3 · 文档问答" },
8      "subtitle": { "tag": "plain_text", "content": "季度产品需求文档.pdf · 25.3K tokens · 4.4s" }
9    },
10    "elements": [
11      { "tag": "div", "text": { "tag": "lark_md", "content": "**你的问题**\n支付模块有哪些前置依赖？" } },
12      { "tag": "hr" },
13      { "tag": "div", "text": { "tag": "lark_md", "content": "| 依赖方 | 状态 |\n|---|---|\n| 用户中心 | 已就绪 |" } },
14      { "tag": "note", "elements": [
15        { "tag": "plain_text", "content": "出处：第 3.2 节 支付模块 · AI 生成内容请核实" }
16      ]}
17    ],
18    "actions": [
19      { "tag": "action", "actions": [
20        { "tag": "button", "text": { "tag": "plain_text", "content": "继续追问" },
21          "type": "primary", "value": { "action": "follow_up", "fid": "cq1x2y3z4" } },
22        { "tag": "button", "text": { "tag": "plain_text", "content": "换个模型重答" },
23          "type": "default", "value": { "action": "rethink", "fid": "cq1x2y3z4" } },
24        { "tag": "button", "text": { "tag": "plain_text", "content": "导出到飞书文档" },
25          "type": "default", "value": { "action": "export" } }
26      ]}
27    ]
28  }
29}

1def build_doc_card(question: str, answer: str, doc_name: str,
2                   model: str, elapsed: float, tokens: int, fid: str | None = None) -> dict:
3    return {
4        "config": {"wide_screen_mode": True, "enable_forward": True},
5        "header": {
6            "template": "turquoise",
7            "title": {"tag": "plain_text", "content": "Kimi K3 · 文档问答"},
8            "subtitle": {"tag": "plain_text",
9                         "content": f"{doc_name} · {tokens/1000:.1f}K tokens · {elapsed:.1f}s"},
10        },
11        "elements": [
12            {"tag": "div", "text": {"tag": "lark_md", "content": f"**你的问题**\n{question}"}},
13            {"tag": "hr"},
14            {"tag": "div", "text": {"tag": "lark_md", "content": answer}},
15            {"tag": "note", "elements": [
16                {"tag": "plain_text", "content": f"{model} · AI 生成内容，请核实关键信息"}
17            ]},
18        ],
19        "actions": [{
20            "tag": "action",
21            "actions": [
22                {"tag": "button", "text": {"tag": "plain_text", "content": "继续追问"},
23                 "type": "primary", "value": {"action": "follow_up", "fid": fid}},
24                {"tag": "button", "text": {"tag": "plain_text", "content": "换个模型重答"},
25                 "type": "default", "value": {"action": "rethink", "fid": fid}},
26            ],
27        }],
28    }
29
30push("interactive", build_doc_card(q, answer, "需求文档.pdf", "kimi-k3", 4.4, 25300, fid))

1def stream_answer_to_card(chat_id: str, question: str, fid: str | None = None):
2    # 1. 先发一张灰色占位卡，用户 0.3 秒内就能看到「机器人在干活」
3    message_id = send_placeholder(chat_id)
4
5    # 2. 流式拉取 Kimi 输出
6    buf, last_push, full = "", 0.0, ""
7    with httpx.stream("POST", MS_URL,
8                      headers={"Authorization": f"Bearer {MS_KEY}"},
9                      json={"model": "kimi-k3", "messages": build_messages(question, fid),
10                            "stream": True, "temperature": 0.3},
11                      timeout=120) as resp:
12        for line in resp.iter_lines():
13            if not line.startswith("data: "):
14                continue
15            chunk = line[6:]
16            if chunk == "[DONE]":
17                break
18            delta = json.loads(chunk)["choices"][0]["delta"].get("content", "")
19            full += delta
20
21            # 3. 每 0.8 秒或每累积 120 字更新一次卡片，别一个字一个请求
22            if time.time() - last_push > 0.8 or len(full) - len(buf) > 120:
23                update_card(message_id, build_streaming_card(question, full))
24                buf, last_push = full, time.time()
25
26    # 4. 收尾：换成带按钮的最终卡片
27    update_card(message_id, build_doc_card(question, full, "需求文档.pdf",
28                                           "kimi-k3", 0, len(full)//2, fid))

1App ID：     cli_c3d4e5f6g7h8i9j0
2App Secret： Kp3Lm7Xn2QrT9YbA1b2C3d4E5f6G7h8

1import os, time, threading, httpx
2
3APP_ID = os.environ["FEISHU_APP_ID"]
4APP_SECRET = os.environ["FEISHU_APP_SECRET"]
5FS_TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
6
7_fs = {"token": None, "expire_at": 0}
8_fs_lock = threading.Lock()
9
10def get_tenant_access_token() -> str:
11    """有效期 2 小时，过期前 5 分钟刷新。
12    为什么必须缓存：飞书对鉴权接口有独立限流，高频换取会被拒绝，
13    进而导致所有发消息请求失败。
14    """
15    now = time.time()
16    if _fs["token"] and _fs["expire_at"] - now > 300:
17        return _fs["token"]
18    with _fs_lock:
19        if _fs["token"] and _fs["expire_at"] - time.time() > 300:
20            return _fs["token"]
21        r = httpx.post(FS_TOKEN_URL, json={
22            "app_id": APP_ID, "app_secret": APP_SECRET}, timeout=10)
23        data = r.json()
24        if data.get("code") != 0:
25            raise RuntimeError(f"获取 tenant_access_token 失败：{data}")
26        _fs["token"] = data["tenant_access_token"]
27        _fs["expire_at"] = now + data["expire"]     # 通常 7200 秒
28        return _fs["token"]

1SEND_URL = "https://open.feishu.cn/open-apis/im/v1/messages"
2
3def send_to_chat(chat_id: str, card: dict, receive_id_type: str = "chat_id") -> str:
4    r = httpx.post(SEND_URL, params={"receive_id_type": receive_id_type},
5        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
6        json={"receive_id": chat_id, "msg_type": "interactive",
7              "content": json.dumps(card, ensure_ascii=False)},   # 必须字符串化
8        timeout=10).json()
9    return r["data"]["message_id"]
10
11def send_placeholder(chat_id: str) -> str:
12    card = {"config": {"wide_screen_mode": True},
13            "header": {"template": "grey",
14                       "title": {"tag": "plain_text", "content": "Kimi K3 正在读文档…"}},
15            "elements": [{"tag": "div",
16                          "text": {"tag": "lark_md", "content": "文档解析中，请稍候"}}]}
17    return send_to_chat(chat_id, card)
18
19def update_card(message_id: str, card: dict):
20    """PATCH 覆盖原卡片，用户看到的是同一条消息「长出」答案"""
21    return httpx.patch(f"{SEND_URL}/{message_id}",
22        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
23        json={"content": json.dumps(card, ensure_ascii=False)}, timeout=10).json()

1import os, json as _json
2import lark_oapi as lark
3from lark_oapi.api.im.v1 import *
4
5def on_receive(data: P2ImMessageReceiveV1) -> None:
6    msg = data.event.message
7    # 文本消息
8    if msg.message_type == "text":
9        question = _json.loads(msg.content).get("text", "").strip()
10        if not question:
11            return
12        fid = cache.get(f"fid:{msg.chat_id}")        # 复用群里最近一份文档
13        model = pick_model(question, bool(fid))
14        answer, _ = ask_kimi(question, model, file_id=fid)
15        card = build_doc_card(question, answer, cache.get(f"doc:{msg.chat_id}", "会话"),
16                              model, 0, 0, fid)
17    # 文件消息（pdf / docx 等）
18    elif msg.message_type == "file":
19        content = _json.loads(msg.content)
20        fid = handle_file_question(msg.message_id, content["file_key"], "请先给出 3 条要点摘要")
21        cache[f"fid:{msg.chat_id}"] = fid
22        cache[f"doc:{msg.chat_id}"] = content.get("file_name", "文档")
23        card = {"config": {"wide_screen_mode": True},
24                "header": {"template": "green",
25                           "title": {"tag": "plain_text", "content": "文档已读取"}},
26                "elements": [{"tag": "div",
27                              "text": {"tag": "lark_md", "content": "已读完，直接 @ 我问问题即可"}}]}
28    else:
29        return
30
31    client.im.v1.message.reply(ReplyMessageRequest.builder()
32        .message_id(msg.message_id)
33        .request_body(ReplyMessageRequestBody.builder()
34            .msg_type("interactive")
35            .content(_json.dumps(card, ensure_ascii=False)).build())
36        .build())
37
38handler = lark.EventDispatcherHandler.builder("", "") \
39    .register_p2_im_message_receive_v1(on_receive).build()
40client = lark.ws.Client(os.environ["FEISHU_APP_ID"], os.environ["FEISHU_APP_SECRET"],
41                        event_handler=handler, log_level=lark.LogLevel.INFO)
42client.start()

1@app.post("/feishu/event")
2async def feishu_event(req: Request):
3    body = await req.json()
4    if "challenge" in body:                      # 1. URL 校验，1 秒内返回
5        return {"challenge": body["challenge"]}
6    if "encrypt" in body:                        # 2. 配了 Encrypt Key 时是密文
7        body = decrypt(body["encrypt"], os.environ["FEISHU_ENCRYPT_KEY"])
8    if body.get("header", {}).get("event_type") == "im.message.receive_v1":
9        asyncio.create_task(handle_message(body["event"]))   # 异步，别阻塞
10    return {"code": 0}

1import base64, hashlib
2from Crypto.Cipher import AES      # pip install pycryptodome
3
4def decrypt(encrypt_b64: str, encrypt_key: str) -> dict:
5    """飞书 Encrypt Key 是 43 位字符串，SHA256 后取前 32 字节作密钥"""
6    key = hashlib.sha256(encrypt_key.encode()).digest()[:32]
7    raw = base64.b64decode(encrypt_b64)
8    iv, cipher_text = raw[:16], raw[16:]           # 前 16 字节是 IV
9    plain = AES.new(key, AES.MODE_CBC, iv).decrypt(cipher_text)
10    return _json.loads(plain[:-plain[-1]].decode("utf-8"))    # 去 PKCS7 填充

1第 1 天：可用范围只放自己，单聊自测 + 传一份 10 页 PDF
2第 2 天：拉 3 人测试群，验证 @、私聊、文件上传、卡片按钮
3第 3 天：放开到 20 人小群，观察 token 消耗与响应时长
4第 5 天：全项目组放开，群里置顶「机器人使用说明」

1- [x] 密钥全在环境变量里，git status 看不到 .env
2- [x] 连续发 100 条测试消息，没有触发飞书限流
3- [x] 上传一份 80 页 PDF，超时设置已放宽到 60 秒以上
4- [x] 断网 10 秒再恢复，长连接能自动重连
5- [x] 卡片按钮点下去都有正确回传，file_id 能正确复用
6- [x] Moonshot 后台设了余额告警

1@Kimi K3 长文助手 （先上传《季度产品需求文档.pdf》）
2这份文档里支付模块有哪些前置依赖？请用表格列出依赖方、当前状态、
3负责人，并标注出处章节。

1@Kimi K3 长文助手
2对比这两份竞品分析报告：
31）双方对「中小企业市场」的判断有什么分歧；
42）各自引用的数据口径是否一致；
53）给出 3 条我们可以借鉴的打法。
6要求用 Markdown 表格对比，每条结论标注出自哪份文档。

1@Kimi K3 长文助手
2这是一场 2 小时评审会的转写稿。请输出：
31）会议议题清单（按讨论顺序）；2）每个议题的结论（未达成一致的标「悬而未决」）；
43）所有待办（负责人 + 时间，缺失标「待确认」）；
54）被反复提及但始终没有结论的话题。

1@Kimi K3 长文助手
2审这份合同，只关注我方（乙方）风险：
31）付款条款里有没有对我方不利的约定；
42）违约责任的赔偿上限是多少；
53）验收标准是否明确，有没有「以甲方满意为准」这类模糊表述。
6输出按风险等级排序，标注条款号。

1阶段 1（入门）：群自定义机器人 + 加签，能把 Kimi 的回答推进群里      ← 你现在在这里
2阶段 2（进阶）：/v1/files 上传 + file_id 引用，做长文档问答
3阶段 3（高级）：交互卡片 + 占位卡 + 流式更新，把感知延迟压到 0.4 秒
4阶段 4（专家）：自建应用收文件 + 长连接事件订阅 + 模型路由
5阶段 5（架构）：Webhook 回调 + 多副本 + KMS 密钥管理 + 敏感文档脱敏