DeepSeek V4.1 Flash 接入 飞书 完整指南

1每日 tokens：200 × 5 × (300 + 500) = 800,000 tokens
2每日成本（V4.1 Flash）：300×200×5/1M×0.8 + 500×200×5/1M×2.4 = ¥0.24 + ¥1.20 = ¥1.44
3每月成本（30 天）：约 ¥43
4换成某闭源旗舰 A：约 ¥1,080/月

1❌ 需要跨几十条历史消息做复杂推理的群（激活参数太小，长链条推理会掉链子）
2❌ 要机器人写生产级代码并直接合并（8B 激活对边界场景考虑不足，必须人工 review）
3❌ 要求稳定输出严格 JSON 给下游系统消费（闭源旗舰的结构化输出更可靠，或加 response_format + 重试）
4❌ 合规要求数据不出内网的场景（这时该用开源权重本地部署，而不是调公有云 API）

1Webhook 地址：
2https://open.feishu.cn/open-apis/bot/v2/hook/8a1f2c3d-4b5e-6f70-8192-a3b4c5d6e7f8
3
4密钥（secret）：
5Xy7KpQ2mNv9RtWzA

1import hmac
2import hashlib
3import base64
4import time
5
6def gen_sign(secret: str, timestamp: int | None = None) -> tuple[int, str]:
7    """生成飞书自定义机器人加签所需的 timestamp 与 sign。
8
9    注意三件事：
10    1. timestamp 必须是秒级（10 位），不要用 int(time.time() * 1000)
11    2. 待签名串是 timestamp + 换行符 + secret，顺序不能反
12    3. HMAC 的 key 是空字符串 b""，secret 是参与签名的原文
13    """
14    timestamp = timestamp or int(time.time())   # 秒级，不是毫秒
15    string_to_sign = f"{timestamp}\n{secret}"   # 拼接原材料
16    hmac_code = hmac.new(
17        b"",                                    # 空字符串作为密钥
18        string_to_sign.encode("utf-8"),
19        hashlib.sha256,
20    ).digest()
21    sign = hmac_code.hex()                      # 十六进制，不是 base64
22    return timestamp, sign

1import httpx
2
3WEBHOOK = "https://open.feishu.cn/open-apis/bot/v2/hook/8a1f2c3d-4b5e-6f70-8192-a3b4c5d6e7f8"
4SECRET = "Xy7KpQ2mNv9RtWzA"
5
6def send_text(text: str) -> dict:
7    ts, sign = gen_sign(SECRET)
8    payload = {
9        "timestamp": str(ts),   # 字符串形式的秒级时间戳
10        "sign": sign,           # 十六进制签名
11        "msg_type": "text",
12        "content": {"text": text},
13    }
14    r = httpx.post(WEBHOOK, json=payload, timeout=10)
15    return r.json()
16    # 成功：{"code": 0, "msg": "success", "data": {}}
17    # 失败：{"code": 19021, "msg": "sign match fail or timestamp is not within one hour"}

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

1# mac / linux
2echo 'export DEEPSEEK_API_KEY="sk-ds-v4-1f-7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c"' >> ~/.bashrc
3source ~/.bashrc
4
5# windows powershell
6setx DEEPSEEK_API_KEY "sk-ds-v4-1f-7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2c"

1Base URL：https://api.deepseek.com/v1
2Model ID：deepseek-v4.1-flash

1feishu-deepseek-bot/
2├── .env              # 存放密钥，务必加进 .gitignore
3├── .gitignore
4├── bot.py            # 主程序
5└── card.py           # 卡片 JSON 构造（下一步展开）

1import os
2import json
3import time
4import hmac
5import hashlib
6import httpx
7from openai import OpenAI
8
9# ---------- 凭证：全部走环境变量 ----------
10WEBHOOK = os.environ["FEISHU_WEBHOOK"]     # https://open.feishu.cn/open-apis/bot/v2/hook/xxx
11SECRET  = os.environ["FEISHU_SECRET"]      # 加签密钥
12DS_KEY  = os.environ["DEEPSEEK_API_KEY"]
13
14# ---------- DeepSeek 客户端 ----------
15ds = OpenAI(
16    api_key=DS_KEY,
17    base_url="https://api.deepseek.com/v1",   # 注意带 /v1
18    timeout=30.0,
19    max_retries=2,                            # 网络抖动自动重试 2 次
20)
21MODEL = "deepseek-v4.1-flash"
22
23# ---------- 系统提示词：决定机器人在群里的「人设」 ----------
24SYSTEM_PROMPT = """你是飞书群里的 AI 助手，底层模型是 DeepSeek V4.1 Flash。
25回答要求：
261. 中文，简洁，能用列表就用列表，控制在 300 字以内
272. 涉及代码时给出完整可运行片段，并标注语言
283. 不确定就说不确定，不要编造版本号、价格、接口地址
294. 如果用户的问题需要公司内网数据，直接说明你拿不到
30"""
31
32def ask_deepseek(question: str, max_tokens: int = 1024) -> tuple[str, int]:
33    """返回 (回答文本, 总 token 消耗)"""
34    resp = ds.chat.completions.create(
35        model=MODEL,
36        messages=[
37            {"role": "system", "content": SYSTEM_PROMPT},
38            {"role": "user", "content": question},
39        ],
40        temperature=0.3,     # 群问答要稳，不要发散
41        max_tokens=max_tokens,
42        stream=False,
43    )
44    return resp.choices[0].message.content, resp.usage.total_tokens
45
46def gen_sign(secret: str) -> tuple[str, str]:
47    ts = str(int(time.time()))             # 秒
48    sign = hmac.new(b"", f"{ts}\n{secret}".encode(), hashlib.sha256).hexdigest()
49    return ts, sign
50
51def push(msg_type: str, content: dict) -> dict:
52    """统一推送入口，msg_type 支持 text / post / interactive"""
53    ts, sign = gen_sign(SECRET)
54    body = {"timestamp": ts, "sign": sign, "msg_type": msg_type, "content": content}
55    r = httpx.post(WEBHOOK, json=body, timeout=10)
56    return r.json()
57
58if __name__ == "__main__":
59    q = "用一句话解释什么是 MoE 架构"
60    answer, tokens = ask_deepseek(q)
61    print(push("text", {"text": f"Q：{q}\n\nA：{answer}\n\n（消耗 {tokens} tokens）"}))

1pip install openai httpx python-dotenv
2python bot.py
3# 群里应该收到一条消息，控制台打印 {'code': 0, 'msg': 'success', 'data': {}}

1{
2  "msg_type": "interactive",
3  "card": {
4    "config": {
5      "wide_screen_mode": true,      // true 时卡片在桌面端占满宽度，长回答必开
6      "enable_forward": true         // 允许群成员转发这张卡
7    },
8    "header": {
9      "template": "blue",            // 必须是内置色号，不能写 #1677ff
10      "title": {
11        "tag": "plain_text",
12        "content": "DeepSeek V4.1 Flash · 回答"
13      },
14      "subtitle": {
15        "tag": "plain_text",
16        "content": "耗时 1.8s · 消耗 742 tokens · 模型 V4.1 Flash"
17      }
18    },
19    "elements": [
20      {
21        "tag": "div",                // div = 一段 Markdown 文本
22        "text": {
23          "tag": "lark_md",          // lark_md 支持 Markdown 语法
24          "content": "**你的问题：** MoE 架构是什么？"
25        }
26      },
27      { "tag": "hr" },               // 分隔线
28      {
29        "tag": "div",
30        "text": {
31          "tag": "lark_md",
32          "content": "MoE（Mixture of Experts）把大模型拆成多个「专家」子网络……"
33        }
34      },
35      {
36        "tag": "note",               // note = 底部灰色小字
37        "elements": [
38          { "tag": "plain_text", "content": "本回答由 AI 生成，请自行核实关键信息" }
39        ]
40      }
41    ],
42    "actions": [
43      {
44        "tag": "action",
45        "actions": [
46          {
47            "tag": "button",
48            "text": { "tag": "plain_text", "content": "重新生成" },
49            "type": "primary",       // primary / default / danger
50            "value": { "action": "regenerate", "qid": "9527" }  // 点击后原样回传
51          },
52          {
53            "tag": "button",
54            "text": { "tag": "plain_text", "content": "复制全文" },
55            "type": "default",
56            "value": { "action": "copy", "qid": "9527" }
57          },
58          {
59            "tag": "select_static",  // 下拉选择
60            "placeholder": { "tag": "plain_text", "content": "转给谁？" },
61            "value": { "action": "assign" },
62            "options": [
63              { "text": { "tag": "plain_text", "content": "转给值班同事" }, "value": "duty" },
64              { "text": { "tag": "plain_text", "content": "转成工单" }, "value": "ticket" }
65            ]
66          }
67        ]
68      }
69    ]
70  }
71}

1def build_answer_card(question: str, answer: str, elapsed: float, tokens: int) -> dict:
2    return {
3        "config": {"wide_screen_mode": True, "enable_forward": True},
4        "header": {
5            "template": "blue",
6            "title": {"tag": "plain_text", "content": "DeepSeek V4.1 Flash · 回答"},
7            "subtitle": {"tag": "plain_text", "content": f"耗时 {elapsed:.1f}s · {tokens} tokens"},
8        },
9        "elements": [
10            {"tag": "div", "text": {"tag": "lark_md", "content": f"**你的问题**\n{question}"}},
11            {"tag": "hr"},
12            {"tag": "div", "text": {"tag": "lark_md", "content": answer}},
13            {"tag": "note", "elements": [
14                {"tag": "plain_text", "content": "AI 生成内容，请核实后再使用"}
15            ]},
16        ],
17        "actions": [{
18            "tag": "action",
19            "actions": [
20                {"tag": "button", "text": {"tag": "plain_text", "content": "重新生成"},
21                 "type": "primary", "value": {"action": "regenerate"}},
22                {"tag": "button", "text": {"tag": "plain_text", "content": "复制全文"},
23                 "type": "default", "value": {"action": "copy"}},
24            ],
25        }],
26    }
27
28# 推送（注意 card 放在 content 里）
29push("interactive", build_answer_card(q, answer, 1.8, 742))

1> **注意**：群自定义机器人的 Webhook **不支持更新已发出的消息**。要做「先占位后更新」，必须走方式 B 的自建应用，用 `PATCH /open-apis/im/v1/messages/{message_id}` 覆盖原卡片。
2
3```python
4def send_placeholder(chat_id: str) -> str:
5    """先发一张灰色「思考中」卡片，返回 message_id 用于后续更新"""
6    card = {
7        "config": {"wide_screen_mode": True},
8        "header": {"template": "grey",
9                   "title": {"tag": "plain_text", "content": "DeepSeek 思考中…"}},
10        "elements": [{"tag": "div",
11                      "text": {"tag": "lark_md", "content": "正在调用 V4.1 Flash，请稍候"}}],
12    }
13    r = httpx.post(
14        SEND_URL,
15        params={"receive_id_type": "chat_id"},
16        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
17        json={"receive_id": chat_id, "msg_type": "interactive",
18              "content": json.dumps(card, ensure_ascii=False)},
19        timeout=10,
20    ).json()
21    return r["data"]["message_id"]   # 拿到 id 才能更新
22
23def update_card(message_id: str, card: dict) -> dict:
24    """用 PATCH 覆盖原卡片，用户看到的是同一条消息「长出」了答案"""
25    return httpx.patch(
26        f"{SEND_URL}/{message_id}",
27        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
28        json={"content": json.dumps(card, ensure_ascii=False)},
29        timeout=10,
30    ).json()

1App ID：     cli_a1b2c3d4e5f6g7h8
2App Secret： Xy7KpQ2mNv9RtWzA1B2c3D4e5F6g7H8

1import time
2import threading
3import httpx
4
5APP_ID = os.environ["FEISHU_APP_ID"]
6APP_SECRET = os.environ["FEISHU_APP_SECRET"]
7TOKEN_URL = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
8
9_token_cache = {"token": None, "expire_at": 0}
10_lock = threading.Lock()
11
12def get_tenant_access_token() -> str:
13    """带缓存的 token 获取：过期前 5 分钟刷新，避免每次请求都打鉴权接口。
14
15    为什么必须缓存：飞书对鉴权接口有独立限流，实测每 60 秒超过 20 次
16    就会被拒绝（表现为获取 token 返回非 0 code），进而导致全站发消息失败。
17    """
18    now = time.time()
19    if _token_cache["token"] and _token_cache["expire_at"] - now > 300:
20        return _token_cache["token"]
21
22    with _lock:   # 加锁，避免并发时多个线程同时刷新
23        if _token_cache["token"] and _token_cache["expire_at"] - time.time() > 300:
24            return _token_cache["token"]
25        r = httpx.post(TOKEN_URL, json={"app_id": APP_ID, "app_secret": APP_SECRET}, timeout=10)
26        data = r.json()
27        if data.get("code") != 0:
28            raise RuntimeError(f"获取 token 失败：{data}")
29        _token_cache["token"] = data["tenant_access_token"]
30        _token_cache["expire_at"] = now + data["expire"]   # expire 单位是秒，通常 7200
31        return _token_cache["token"]

1SEND_URL = "https://open.feishu.cn/open-apis/im/v1/messages"
2
3def send_to_chat(chat_id: str, text: str, receive_id_type: str = "chat_id"):
4    token = get_tenant_access_token()
5    r = httpx.post(
6        SEND_URL,
7        params={"receive_id_type": receive_id_type},   # chat_id / open_id / union_id / email
8        headers={"Authorization": f"Bearer {token}"},
9        json={
10            "receive_id": chat_id,        # 群聊传 oc_xxx，单聊传用户的 ou_xxx
11            "msg_type": "interactive",    # text / post / interactive / image / file
12            "content": json.dumps(build_answer_card(...), ensure_ascii=False),  # 注意：必须是 JSON 字符串
13        },
14        timeout=10,
15    )
16    return r.json()

1pip install lark-oapi

1import lark_oapi as lark
2from lark_oapi.api.im.v1 import *
3
4def on_receive_message(data: P2ImMessageReceiveV1) -> None:
5    """收到用户消息时触发"""
6    msg = data.event.message
7    # 只处理文本消息，忽略图片、表情、系统消息
8    if msg.message_type != "text":
9        return
10    import json as _json
11    question = _json.loads(msg.content).get("text", "").strip()
12    if not question:
13        return
14
15    answer, tokens = ask_deepseek(question)          # 复用步骤四的函数
16    card = build_answer_card(question, answer, 0, tokens)
17
18    # 用 message_id 做「回复」，消息会挂在原消息下面，群里不会刷屏
19    client.im.v1.message.reply(ReplyMessageRequest.builder()
20        .message_id(msg.message_id)
21        .request_body(ReplyMessageRequestBody.builder()
22            .msg_type("interactive")
23            .content(_json.dumps(card, ensure_ascii=False))
24            .build())
25        .build())
26
27# 注册事件处理器并启动长连接
28event_handler = lark.EventDispatcherHandler.builder("", "") \
29    .register_p2_im_message_receive_v1(on_receive_message) \
30    .build()
31
32client = lark.ws.Client(
33    os.environ["FEISHU_APP_ID"],
34    os.environ["FEISHU_APP_SECRET"],
35    event_handler=event_handler,
36    log_level=lark.LogLevel.INFO,
37)
38client.start()   # 阻塞运行，Ctrl+C 退出

1@app.post("/feishu/event")
2async def feishu_event(req: Request):
3    body = await req.json()
4
5    # 1. 首次配置时的 URL 校验
6    if "challenge" in body:
7        return {"challenge": body["challenge"]}
8
9    # 2. 正式事件：如果配了 Encrypt Key，body 是密文
10    if "encrypt" in body:
11        body = decrypt(body["encrypt"], os.environ["FEISHU_ENCRYPT_KEY"])
12
13    # 3. 处理消息事件
14    if body.get("header", {}).get("event_type") == "im.message.receive_v1":
15        asyncio.create_task(handle_message(body["event"]))   # 异步处理，别阻塞
16
17    return {"code": 0}   # 必须立刻返回，否则飞书会重试 3 次

1import base64, hashlib, json
2from Crypto.Cipher import AES   # pip install pycryptodome
3
4def decrypt(encrypt_b64: str, encrypt_key: str) -> dict:
5    key = hashlib.sha256(encrypt_key.encode()).digest()[:32]   # 32 字节密钥
6    raw = base64.b64decode(encrypt_b64)
7    iv, cipher_text = raw[:16], raw[16:]                       # 前 16 字节是 IV
8    cipher = AES.new(key, AES.MODE_CBC, iv)
9    plain = cipher.decrypt(cipher_text)
10    plain = plain[:-plain[-1]]                                 # 去掉 PKCS7 填充
11    return json.loads(plain.decode("utf-8"))

1第 1 天：只把自己加进「可用范围」，单聊自测
2第 2 天：拉一个 3 人的测试群，验证 @、私聊、卡片按钮
3第 3 天：放开到一个 20 人的小群，观察 token 消耗和响应时长
4第 5 天：全公司放开，同时在群里置顶一条「机器人使用说明」

1- [x] 密钥全部在环境变量里，git status 看不到 .env
2- [x] 连续发 100 条测试消息，没有触发限流
3- [x] 断网 10 秒再恢复，长连接能自动重连
4- [x] 卡片里所有按钮点下去都有正确回传
5- [x] 群里 @ 机器人但没提问题时，机器人回一句引导语而不是报错
6- [x] DeepSeek 后台设了预算告警

1@DeepSeek V4.1 Flash 助手 
2我们的 Python 服务用 SQLAlchemy 2.0，查询接口偶发 
3"QueuePool limit of size 5 overflow 10 reached" 报错。
4请说明：1）这个报错的根因；2）三种修复方案及其取舍；3）推荐哪一种，
5用表格输出，控制在 300 字内。

1@DeepSeek V4.1 Flash 助手
2以下是今天的周会记录（约 1800 字），请整理成：
31）5 条以内的结论；2）待办清单（负责人 + 截止时间，缺失的标「待确认」）；
43）需要升级到老板决策的 1-2 个风险点。
5输出用 Markdown，不要复述原文。

1【P1 告警】prod-order-service 5xx 错误率 8.7%（阈值 1%），
2持续 3 分钟，影响接口：POST /api/v1/orders。
3最近一次发版：2026-04-27 14:02，改动：订单金额计算逻辑。

1@DeepSeek V4.1 Flash 助手
2附件是我们《订单系统接口规范 v3.2》全文（约 6 万字）。请回答：
31）创建订单接口在「优惠券叠加」场景下的字段校验规则是什么？
42）退款接口的幂等键由哪几个字段组成？
53）v3.1 到 v3.2 改了什么？
6请给出原文出处（章节号）。

1阶段 1（入门）：群自定义机器人 + 加签，能往群里推 DeepSeek 的回答      ← 你现在在这里
2阶段 2（进阶）：改发交互式卡片，加「重新生成 / 复制 / 转工单」按钮
3阶段 3（高级）：自建应用 + 长连接事件订阅，机器人能听见并回复
4阶段 4（专家）：卡片流式更新 + Function Calling 接内部系统 + 用量看板
5阶段 5（架构）：Webhook 回调 + 多副本 + 密钥管理服务 + 敏感信息脱敏网关