文心一言 ERNIE 4.5 接入 飞书 完整指南

1每日 tokens：150 × 4 × (350 + 450) = 480,000 tokens
2全部走 ernie-4.5-turbo：约 ¥1.07/天，约 ¥32/月
3全部走 ernie-4.5 旗舰：  约 ¥5.3/天，约 ¥160/月
4混合路由（80% turbo + 20% 旗舰）：约 ¥1.9/天，约 ¥58/月

1❌ 重度代码生成与重构（文心的代码能力可用，但不如专攻代码的模型稳）
2❌ 需要超长上下文一次性吞进几十万字（128K 是上限，海量场景要自己做切片检索）
3❌ 要求严格 JSON Schema 给下游系统消费（建议加后处理校验 + 失败重试）
4❌ 强实时信息问答（模型知识有截止，需要接搜索插件或联网能力）

1Webhook 地址：
2https://open.feishu.cn/open-apis/bot/v2/hook/2f9e8d7c-6b5a-4938-8271-6e5d4c3b2a19
3
4密钥（secret）：
5Qw8Zx3Nm5RtYvB1

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/2f9e8d7c-6b5a-4938-8271-6e5d4c3b2a19"
4SECRET = "Qw8Zx3Nm5RtYvB1"
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": "文心一言接入测试"}))

1import crypto from 'node:crypto';
2
3const WEBHOOK = 'https://open.feishu.cn/open-apis/bot/v2/hook/2f9e8d7c-6b5a-4938-8271-6e5d4c3b2a19';
4const SECRET = 'Qw8Zx3Nm5RtYvB1';
5
6function genSign(secret) {
7  const ts = String(Math.floor(Date.now() / 1000));      // 秒级！
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}

1API Key（AK）：    bce-v3/ALTAK-xxxxxxxxxxxxxxxx/xxxxxxxxxxxxxxxxxxxx
2Secret Key（SK）： bce-v3/ALTSK-xxxxxxxxxxxxxxxx/xxxxxxxxxxxxxxxxxxxx

1curl -X POST \
2  'https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=你的AK&client_secret=你的SK'

1{
2  "access_token": "24.be1f4c2d8a3e5f6b7c8d9e0f1a2b3c4d.2592000.1785000000.xxxx",
3  "expires_in": 2592000,
4  "refresh_token": "25.xxxx"
5}

1import os, time, threading, httpx
2
3AK = os.environ["QIANFAN_AK"]
4SK = os.environ["QIANFAN_SK"]
5TOKEN_URL = "https://aip.baidubce.com/oauth/2.0/token"
6
7_tok = {"value": None, "expire_at": 0}
8_lock = threading.Lock()
9
10def get_access_token() -> str:
11    """千帆 access_token 有效期 30 天，但换取接口有限流，务必缓存。
12
13    提前 24 小时刷新，避免长跑的服务在半夜因为 token 恰好过期而全线失败。
14    """
15    now = time.time()
16    if _tok["value"] and _tok["expire_at"] - now > 86400:
17        return _tok["value"]
18    with _lock:
19        if _tok["value"] and _tok["expire_at"] - time.time() > 86400:
20            return _tok["value"]
21        r = httpx.post(TOKEN_URL, params={
22            "grant_type": "client_credentials",
23            "client_id": AK,
24            "client_secret": SK,
25        }, timeout=10)
26        data = r.json()
27        if "access_token" not in data:
28            raise RuntimeError(f"换取 access_token 失败：{data}")
29        _tok["value"] = data["access_token"]
30        _tok["expire_at"] = now + data.get("expires_in", 2592000)
31        return _tok["value"]

1from openai import OpenAI
2
3client = OpenAI(
4    api_key=get_access_token(),                              # 千帆 access_token
5    base_url="https://qianfan.baidubce.com/v2",              # 千帆 v2 OpenAI 兼容地址
6    timeout=30.0,
7    max_retries=2,
8)
9
10resp = client.chat.completions.create(
11    model="ernie-4.5-turbo",        # 或 ernie-4.5 / ernie-4.5-vl / ernie-speed
12    messages=[{"role": "user", "content": "用 100 字介绍飞书自定义机器人"}],
13    temperature=0.3,
14    max_tokens=1024,
15)
16print(resp.choices[0].message.content)

1feishu-ernie-bot/
2├── .env          # AK/SK 与 webhook 密钥，务必加进 .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"]
5QIANFAN_URL = "https://qianfan.baidubce.com/v2/chat/completions"
6
7SYSTEM_PROMPT = """你是飞书群里的 AI 助手，底层模型是文心一言 ERNIE 4.5。
8回答要求：
91. 中文，简洁，控制在 300 字以内，能用列表就用列表
102. 涉及公文/通知时使用规范的中文书面语，不要口语化
113. 不确定就明确说不确定，不要编造政策条款、人名、数字
124. 用户要求改写时，只输出改写结果，不要解释你改了什么
13"""
14
15def ask_ernie(question: str, model: str = "ernie-4.5-turbo") -> tuple[str, int]:
16    """返回 (回答, 总 token 消耗)"""
17    r = httpx.post(
18        QIANFAN_URL,
19        headers={"Authorization": f"Bearer {get_access_token()}"},
20        json={
21            "model": model,
22            "messages": [
23                {"role": "system", "content": SYSTEM_PROMPT},
24                {"role": "user", "content": question},
25            ],
26            "temperature": 0.3,
27            "max_tokens": 1024,
28            "stream": False,
29        },
30        timeout=30,
31    )
32    data = r.json()
33    if "choices" not in data:
34        raise RuntimeError(f"文心一言调用失败：{data}")
35    return data["choices"][0]["message"]["content"], data["usage"]["total_tokens"]
36
37def pick_model(question: str) -> str:
38    """简单路由：长问题或命中关键词时切旗舰，其余走 turbo 省钱"""
39    heavy = any(k in question for k in ["分析", "总结", "起草", "方案", "对比", "复盘"])
40    if heavy or len(question) > 200:
41        return "ernie-4.5"
42    return "ernie-4.5-turbo"
43
44def gen_sign(secret: str) -> tuple[str, str]:
45    ts = str(int(time.time()))
46    return ts, hmac.new(b"", f"{ts}\n{secret}".encode(), hashlib.sha256).hexdigest()
47
48def push(msg_type: str, content: dict) -> dict:
49    ts, sign = gen_sign(SECRET)
50    return httpx.post(WEBHOOK, json={
51        "timestamp": ts, "sign": sign, "msg_type": msg_type, "content": content
52    }, timeout=10).json()
53
54if __name__ == "__main__":
55    q = "帮我把这段话改写成一份正式的通知"
56    model = pick_model(q)
57    answer, tokens = ask_ernie(q, model)
58    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': {}} 即链路打通

1{
2  "msg_type": "interactive",
3  "card": {
4    "config": {
5      "wide_screen_mode": true,     // 桌面端占满宽度，长回答必开
6      "enable_forward": true        // 允许群成员转发这张卡
7    },
8    "header": {
9      "template": "indigo",         // 必须是内置色号，不能写 #4E83FD
10      "title": { "tag": "plain_text", "content": "文心一言 ERNIE 4.5 · 回答" },
11      "subtitle": { "tag": "plain_text", "content": "ernie-4.5-turbo · 耗时 1.4s · 683 tokens" }
12    },
13    "elements": [
14      { "tag": "div", "text": { "tag": "lark_md", "content": "**你的问题**\n帮我把这段话改写成正式通知" } },
15      { "tag": "hr" },
16      { "tag": "div", "text": { "tag": "lark_md", "content": "关于……的通知\n各部门：……" } },
17      {
18        "tag": "note",
19        "elements": [
20          { "tag": "plain_text", "content": "AI 生成内容，正式发布前请人工核对" }
21        ]
22      }
23    ],
24    "actions": [
25      {
26        "tag": "action",
27        "actions": [
28          { "tag": "button", "text": { "tag": "plain_text", "content": "换个语气" },
29            "type": "primary", "value": { "action": "rewrite", "tone": "formal" } },
30          { "tag": "button", "text": { "tag": "plain_text", "content": "再精简一半" },
31            "type": "default", "value": { "action": "shorten" } },
32          { "tag": "button", "text": { "tag": "plain_text", "content": "存进飞书文档" },
33            "type": "default", "value": { "action": "save_doc" } },
34          {
35            "tag": "select_static",
36            "placeholder": { "tag": "plain_text", "content": "转给谁？" },
37            "value": { "action": "assign" },
38            "options": [
39              { "text": { "tag": "plain_text", "content": "转给行政部" }, "value": "hr" },
40              { "text": { "tag": "plain_text", "content": "转给值班同事" }, "value": "duty" }
41            ]
42          }
43        ]
44      }
45    ]
46  }
47}

1def build_answer_card(question: str, answer: str, model: str, elapsed: float, tokens: int) -> dict:
2    return {
3        "config": {"wide_screen_mode": True, "enable_forward": True},
4        "header": {
5            "template": "indigo",
6            "title": {"tag": "plain_text", "content": "文心一言 ERNIE 4.5 · 回答"},
7            "subtitle": {"tag": "plain_text", "content": f"{model} · {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": "rewrite"}},
22                {"tag": "button", "text": {"tag": "plain_text", "content": "再精简一半"},
23                 "type": "default", "value": {"action": "shorten"}},
24            ],
25        }],
26    }
27
28push("interactive", build_answer_card(q, answer, "ernie-4.5-turbo", 1.4, 683))

1import base64
2
3def ask_ernie_with_image(question: str, image_path: str) -> str:
4    b64 = base64.b64encode(open(image_path, "rb").read()).decode()
5    r = httpx.post(
6        "https://qianfan.baidubce.com/v2/chat/completions",
7        headers={"Authorization": f"Bearer {get_access_token()}"},
8        json={
9            "model": "ernie-4.5-vl",
10            "messages": [{
11                "role": "user",
12                "content": [
13                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}},
14                    {"type": "text", "text": question},
15                ],
16            }],
17            "max_tokens": 1024,
18        },
19        timeout=60,
20    )
21    return r.json()["choices"][0]["message"]["content"]

1App ID：     cli_b2c3d4e5f6g7h8i9
2App Secret： Qw8Zx3Nm5RtYvB1C2d3E4f5G6h7I8

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    """飞书 token 有效期 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
23        }, timeout=10)
24        data = r.json()
25        if data.get("code") != 0:
26            raise RuntimeError(f"获取 tenant_access_token 失败：{data}")
27        _fs["token"] = data["tenant_access_token"]
28        _fs["expire_at"] = now + data["expire"]     # 通常 7200 秒
29        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"):
4    r = httpx.post(
5        SEND_URL,
6        params={"receive_id_type": receive_id_type},   # chat_id / open_id / union_id / email
7        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
8        json={
9            "receive_id": chat_id,                     # 群 oc_xxx，单人 ou_xxx
10            "msg_type": "interactive",
11            "content": json.dumps(card, ensure_ascii=False),  # 必须是字符串化的 JSON
12        },
13        timeout=10,
14    )
15    return r.json()

1def send_placeholder(chat_id: str) -> str:
2    card = {
3        "config": {"wide_screen_mode": True},
4        "header": {"template": "grey", "title": {"tag": "plain_text", "content": "文心一言思考中…"}},
5        "elements": [{"tag": "div", "text": {"tag": "lark_md", "content": "正在调用 ERNIE 4.5，请稍候"}}],
6    }
7    r = httpx.post(SEND_URL, params={"receive_id_type": "chat_id"},
8        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
9        json={"receive_id": chat_id, "msg_type": "interactive",
10              "content": json.dumps(card, ensure_ascii=False)}, timeout=10).json()
11    return r["data"]["message_id"]
12
13def update_card(message_id: str, card: dict):
14    return httpx.patch(f"{SEND_URL}/{message_id}",
15        headers={"Authorization": f"Bearer {get_tenant_access_token()}"},
16        json={"content": json.dumps(card, ensure_ascii=False)}, timeout=10).json()

1pip install lark-oapi

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    if msg.message_type != "text":      # 忽略图片、表情、系统消息
8        return
9    question = _json.loads(msg.content).get("text", "").strip()
10    if not question:
11        return
12    model = pick_model(question)
13    answer, tokens = ask_ernie(question, model)
14    card = build_answer_card(question, answer, model, 0, tokens)
15
16    # 用 message_id 做「回复」，消息挂在原消息下面，群里不会刷屏
17    client.im.v1.message.reply(ReplyMessageRequest.builder()
18        .message_id(msg.message_id)
19        .request_body(ReplyMessageRequestBody.builder()
20            .msg_type("interactive")
21            .content(_json.dumps(card, ensure_ascii=False))
22            .build())
23        .build())
24
25handler = lark.EventDispatcherHandler.builder("", "") \
26    .register_p2_im_message_receive_v1(on_receive).build()
27
28client = lark.ws.Client(os.environ["FEISHU_APP_ID"],
29                        os.environ["FEISHU_APP_SECRET"],
30                        event_handler=handler,
31                        log_level=lark.LogLevel.INFO)
32client.start()      # 阻塞运行

1@app.post("/feishu/event")
2async def feishu_event(req: Request):
3    body = await req.json()
4    if "challenge" in body:                     # 1. URL 校验
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}                          # 3. 立刻返回，否则飞书会重试

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 天：可用范围只放自己，单聊自测
2第 2 天：拉 3 人测试群，验证 @、私聊、卡片按钮
3第 3 天：放开到一个 20 人小群，观察 QPS 与 token 消耗
4第 5 天：全部门放开，群里置顶一条「机器人使用说明」

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

1@文心 ERNIE 4.5 助手
2把下面这段口语改成一份正式的部门通知，要求：
31）标题用「关于……的通知」格式；2）正文分「背景 / 要求 / 执行时间」三段；
43）语气正式但不生硬；4）不超过 300 字。
5原文：下周三开始我们这边要搬到 5 楼了，大家记得把自己的东西收好，
6电脑别忘关机，搬家公司九点到，有问题找行政小王。

1@文心 ERNIE 4.5 助手
2以下是今天的季度复盘会记录（约 2200 字），请整理成：
31）5 条以内的结论；2）待办清单（负责人 + 截止时间，缺失的标「待确认」）；
43）需要上级决策的 1-2 个风险点。
5输出用 Markdown，不要复述原文。

1@文心 ERNIE 4.5 助手
2这张截图是后台报的错，请告诉我：
31）错误信息是什么；2）最可能的三个原因；
43）我下一步该点哪里、查什么日志。

1@文心 ERNIE 4.5 助手
2我们公司准备调整年休假规则，请说明：
31）法定年休假的天数是怎么按工龄算的；
42）未休年假折算工资的常见处理方式；
53）哪些做法有合规风险。
6请注明这只是一般性参考，不构成法律意见。

1阶段 1（入门）：群自定义机器人 + 加签，能把文心的回答推进群里      ← 你现在在这里
2阶段 2（进阶）：改发交互式卡片，加「换个语气 / 精简 / 存文档」按钮
3阶段 3（高级）：AK/SK 换 token 的缓存与自动刷新，自建应用 + 长连接事件订阅
4阶段 4（专家）：图片理解 + 卡片流式更新 + Function Calling 接内部系统
5阶段 5（架构）：Webhook 回调 + 多副本 + KMS 密钥管理 + 敏感信息脱敏网关