文心 ERNIE 4.5 接入 钉钉 完整指南：千帆鉴权三件套 + 18 条踩坑实测

1❌ 纯代码生成与重构（ERNIE 4.5 在 HumanEval 类任务上弱于 DeepSeek V4.1 Flash 约 9 个百分点）
2❌ 极低成本的批量推送（它比 Turbo 版贵 4-5 倍）
3❌ 需要英文长文写作的场景（中文强不等于英文强）

1import time, hmac, hashlib, base64, urllib.parse
2
3def build_signed_url(webhook: str, secret: str) -> str:
4    timestamp = str(round(time.time() * 1000))          # 毫秒，13 位
5    string_to_sign = f'{timestamp}\n{secret}'            # 换行符是 \n
6    hmac_code = hmac.new(
7        secret.encode('utf-8'),
8        string_to_sign.encode('utf-8'),
9        digestmod=hashlib.sha256
10    ).digest()
11    sign = base64.b64encode(hmac_code).decode('utf-8')   # base64
12    sign_quoted = urllib.parse.quote_plus(sign)          # urlencode，别漏
13    return f'{webhook}&timestamp={timestamp}&sign={sign_quoted}'

1curl https://api.ipify.org

1import os, time, threading, requests
2
3API_KEY = os.environ['QIANFAN_AK']
4SECRET_KEY = os.environ['QIANFAN_SK']
5TOKEN_URL = ('https://aip.baidubce.com/oauth/2.0/token'
6             '?grant_type=client_credentials'
7             '&client_id={ak}&client_secret={sk}')
8
9class ErnieToken:
10    """旧版 access_token 缓存：官方有效期 30 天，我们 25 天就换"""
11    def __init__(self, ak, sk, refresh_days=25):
12        self.url = TOKEN_URL.format(ak=ak, sk=sk)
13        self.ttl = refresh_days * 86400
14        self._token, self._expire_at, self._lock = None, 0, threading.Lock()
15
16    def get(self):
17        with self._lock:
18            if self._token and time.time() < self._expire_at:
19                return self._token
20            r = requests.post(self.url, timeout=10)
21            data = r.json()
22            if 'access_token' not in data:
23                raise RuntimeError(f'换取失败：{data}')
24            self._token = data['access_token']
25            self._expire_at = time.time() + self.ttl
26            return self._token

1curl 'https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN' \
2  -H 'Content-Type: application/json' \
3  -d '{"msgtype":"text","text":{"content":"文心助手已上线，这是连通性测试"}}'

1import os
2from openai import OpenAI
3
4client = OpenAI(
5    api_key=os.environ['QIANFAN_API_KEY'],                       # bce-v3/... 那串
6    base_url='https://qianfan.baidubce.com/v2',                  # 末尾 /v2 必带
7)
8
9resp = client.chat.completions.create(
10    model='ernie-4.5-turbo-128k',        # 群聊默认用 turbo；复杂任务换 ernie-4.5
11    messages=[
12        {'role': 'system', 'content': '你是钉钉群里的制度助手，回答控制在 300 字，'
13                                      '引用条款时必须给出条款号，不确定就说不确定。'},
14        {'role': 'user', 'content': '差旅报销里，一线城市住宿标准是多少？'}
15    ],
16    temperature=0.3,
17    max_completion_tokens=800,
18    stream=False,
19)
20print(resp.choices[0].message.content)

1import requests
2
3def ask_ernie(question: str, model='ernie-4.5-turbo-128k', system=''):
4    messages = []
5    if system:
6        messages.append({'role': 'system', 'content': system})
7    messages.append({'role': 'user', 'content': question})
8
9    r = requests.post(
10        'https://qianfan.baidubce.com/v2/chat/completions',
11        headers={
12            'Authorization': f'Bearer {os.environ["QIANFAN_API_KEY"]}',  # Bearer 前缀别忘
13            'Content-Type': 'application/json'
14        },
15        json={'model': model, 'messages': messages,
16              'temperature': 0.3, 'max_completion_tokens': 800},
17        timeout=60
18    )
19    body = r.json()
20    if 'error' in body:
21        raise RuntimeError(f'千帆错误：{body["error"]}')
22    return body['choices'][0]['message']['content']

1{
2  "msgtype": "text",
3  "text": { "content": "文心助手：制度问答服务已就绪" },
4  "at": { "atMobiles": ["13800138000"], "isAtAll": false }
5}

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "title": "差旅标准问答",
5    "text": "### 文心助手 · 差旅住宿标准\n\n- **一线城市**：<font color=#008000>600 元/晚</font>\n- **二线城市**：450 元/晚\n- **其他城市**：350 元/晚\n\n> 依据《差旅管理办法》第 3.2 条\n\n[查看完整制度](https://wiki.example.com/travel)"
6  }
7}

1{
2  "msgtype": "link",
3  "link": {
4    "title": "文心助手：制度更新提醒",
5    "text": "《采购管理办法》v3.1 已发布，新增 2 条审批要求",
6    "picUrl": "https://example.com/doc.png",
7    "messageUrl": "https://wiki.example.com/purchase"
8  }
9}

1{
2  "msgtype": "actionCard",
3  "actionCard": {
4    "title": "文心助手：这份合同有 2 处风险条款",
5    "text": "### 合同初审结果\n\n- **第 5.2 条**：违约金比例偏高\n- **第 8.1 条**：缺少知识产权归属约定",
6    "hideAvatar": "0",
7    "btnOrientation": "0",
8    "btns": [
9      { "title": "查看原文", "actionURL": "https://contract.example.com/c/881" },
10      { "title": "转法务", "actionURL": "https://oa.example.com/legal/881" }
11    ]
12  }
13}

1{
2  "msgtype": "feedCard",
3  "feedCard": {
4    "links": [
5      { "title": "文心日报：3 份待审合同", "messageURL": "https://contract.example.com/today", "picURL": "https://example.com/1.png" },
6      { "title": "《差旅办法》v3.1 已发布", "messageURL": "https://wiki.example.com/travel", "picURL": "https://example.com/2.png" }
7    ]
8  }
9}

1import time, threading, itertools, os
2from collections import deque
3
4class RateLimiter:
5    """钉钉每个机器人每分钟最多 20 条"""
6    def __init__(self, max_calls=20, period=60):
7        self.max_calls, self.period = max_calls, period
8        self.calls, self.lock = deque(), threading.Lock()
9
10    def acquire(self):
11        with self.lock:
12            now = time.time()
13            while self.calls and now - self.calls[0] > self.period:
14                self.calls.popleft()
15            if len(self.calls) >= self.max_calls:
16                time.sleep(self.period - (now - self.calls[0]) + 0.5)
17                return self.acquire()
18            self.calls.append(now)
19
20limiter = RateLimiter()
21
22# 方案二：30 秒窗口聚合，把 8 条告警压成 1 条
23_pending, _timer = [], None
24
25def enqueue(msg):
26    global _timer
27    _pending.append(msg)
28    if _timer is None:
29        _timer = threading.Timer(30.0, flush)
30        _timer.start()
31
32def flush():
33    global _timer
34    if not _pending:
35        _timer = None
36        return
37    combined = '\n\n'.join(_pending)
38    _pending.clear()
39    _timer = None
40    limiter.acquire()
41    send_markdown(combined, title='文心助手 · 聚合提醒')

1CorpId:     ding1a2b3c4d5e6f7a8b
2AppKey:     dingoaqxxxxxxxxxxxxxxx
3AppSecret:  xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

1import time, threading, requests
2
3class DingToken:
4    def __init__(self, appkey, appsecret, early=300):
5        self.appkey, self.appsecret, self.early = appkey, appsecret, early
6        self._token, self._expire_at, self._lock = None, 0, threading.Lock()
7
8    def get(self):
9        with self._lock:                       # 并发下只刷新一次
10            if self._token and time.time() < self._expire_at:
11                return self._token
12            r = requests.get('https://oapi.dingtalk.com/gettoken',
13                             params={'appkey': self.appkey,    # 全小写
14                                     'appsecret': self.appsecret},
15                             timeout=10)
16            data = r.json()
17            if data.get('errcode') != 0:
18                raise RuntimeError(f'token 获取失败：{data}')
19            self._token = data['access_token']
20            self._expire_at = time.time() + data['expires_in'] - self.early  # 提前 5 分钟
21            return self._token

1pip install dingtalk-stream

1import os, json
2import dingtalk_stream
3from dingtalk_stream import AckMessage
4from openai import OpenAI
5
6ernie = OpenAI(api_key=os.environ['QIANFAN_API_KEY'],
7               base_url='https://qianfan.baidubce.com/v2')
8
9def stream_ernie(question: str):
10    """ERNIE 4.5 同样支持 SSE 流式"""
11    resp = ernie.chat.completions.create(
12        model='ernie-4.5-turbo-128k',
13        messages=[{'role': 'user', 'content': question}],
14        stream=True, temperature=0.3, max_completion_tokens=800
15    )
16    for chunk in resp:
17        if chunk.choices and chunk.choices[0].delta.content:
18            yield chunk.choices[0].delta.content
19
20class ErnieCardBot(dingtalk_stream.ChatbotHandler):
21    def __init__(self):
22        super().__init__()
23        self.replier = dingtalk_stream.AICardReplier()
24
25    async def process(self, callback: dingtalk_stream.CallbackMessage):
26        incoming = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
27        question = incoming.text.content.strip()
28
29        # 1. 先建卡片占位（用户立刻看到标题栏，回调也不会超时）
30        card = self.replier.create_and_send_card(incoming, title='文心助手正在思考…')
31
32        # 2. 流式推送，攒够 20 字再刷一次，避免闪屏
33        buf = ''
34        for chunk in stream_ernie(question):
35            buf += chunk
36            if len(buf) >= 20:
37                self.replier.replay_streaming(card.card_id, buf, finished=False)
38
39        # 3. 结束帧（漏了卡片会一直转圈）
40        self.replier.replay_streaming(card.card_id, buf, finished=True)
41        return AckMessage.STATUS_OK, 'ok'
42
43credential = dingtalk_stream.Credential(os.environ['DINGTALK_APPKEY'],
44                                        os.environ['DINGTALK_APPSECRET'])
45client = dingtalk_stream.DingTalkStreamClient(credential)
46client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
47                                 ErnieCardBot())
48client.start_forever()

1@文心助手
2
3我在上海出差 4 天，其中一天去苏州开会。
4请问：
51. 住宿标准按哪个城市算？
62. 市内交通补贴怎么算？
73. 需要哪些票据？
8
9请引用《差旅管理办法》的具体条款号，300 字以内。

1你是会议纪要助手。下面是今天产品评审会的语音转写原文（有口语、有重复）。
2
3请输出：
41. 会议主题与参会人（从内容推断，不确定标「待确认」）
52. 已确认的决策（每条一行，用「动词 + 对象」句式）
63. 待办事项：负责人 + 事项 + 截止时间（缺失项标「未指定」）
74. 有争议未决的问题（单独列出）
8
9用 Markdown，不超过 500 字。不要复述讨论过程。
10
11原文：
12[粘贴转写文本]

1你是客服质检员。请评估下面这段客服对话，只输出问题：
2
3检查项：
41. 是否出现承诺无法兑现的表述（如「一定」「保证」）
52. 是否泄露内部信息（工单号以外的内容、内部流程）
63. 是否有情绪化或不耐烦表达
74. 是否遗漏客户明确提出的需求
8
9输出格式：
10- 命中项：序号 + 原话引用 + 建议修改
11- 无问题则输出「通过」
12
13对话：
14[粘贴客服对话]

1你是法务初审助手。请审下面这份采购合同，只输出风险项：
2
31. 付款条件（是否有预付款比例过高、验收后付款周期过长）
42. 违约责任（违约金比例是否超过合同额的 20%）
53. 知识产权归属是否明确
64. 争议解决方式与管辖法院是否合理
7
8输出：每条一行，格式「条款号 | 风险描述 | 建议修改」。
9无风险输出「LGTM」。
10
11合同文本：
12[粘贴合同正文]

1阶段 1（入门）：群机器人 Webhook + markdown 推送        ← 你现在在这里
2阶段 2（进阶）：加签 + 令牌桶限流 + 五种消息类型
3阶段 3（高级）：企业内部应用 + token 缓存 + AI 卡片流式
4阶段 4（专家）：RAG 制度库 + Function Calling + 私有化部署