豆包 1.5 Pro 接入 钉钉 完整指南：方舟接入点配置 + 21 条踩坑实测

1❌ 复杂代码生成与调试（准确率明显弱于 DeepSeek V4.1 Flash）
2❌ 需要严格结构化输出的业务（JSON Schema 稳定性一般）
3❌ 长文档跨章节推理（32K 上下文不够，256K 版又偏贵）

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 不是 \r\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')
12    return f'{webhook}&timestamp={timestamp}&sign={urllib.parse.quote_plus(sign)}'

1curl https://api.ipify.org

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['ARK_API_KEY'],
6    base_url='https://ark.cn-beijing.volces.com/api/v3',
7)
8
9ENDPOINT = 'ep-20260425143022-a8b7c6'      # 接入点 ID，不是模型名
10
11def ask_doubao(question: str, system: str = '', temperature: float = 0.7):
12    resp = client.chat.completions.create(
13        model=ENDPOINT,                     # ← 这里是 ep-xxx
14        messages=[
15            {'role': 'system', 'content': system or '你是钉钉群里的运营助手，'
16                                                   '回答口语化、控制在 300 字以内。'},
17            {'role': 'user', 'content': question}
18        ],
19        temperature=temperature,            # 文案场景调高，技术场景调低
20        max_tokens=800,
21    )
22    return resp.choices[0].message.content

1import requests
2
3def ask_doubao_raw(question: str):
4    r = requests.post(
5        'https://ark.cn-beijing.volces.com/api/v3/chat/completions',
6        headers={
7            'Authorization': f'Bearer {os.environ["ARK_API_KEY"]}',
8            'Content-Type': 'application/json'
9        },
10        json={'model': ENDPOINT,
11              'messages': [{'role': 'user', 'content': question}],
12              'temperature': 0.7, 'max_tokens': 800},
13        timeout=60
14    )
15    body = r.json()
16    if 'error' in body:
17        raise RuntimeError(f'方舟错误：{body["error"]}')
18    return body['choices'][0]['message']['content']

1THINKING_EP = 'ep-20260425143510-d3e2f1'      # doubao-1.5-thinking-pro 的接入点
2NORMAL_EP   = 'ep-20260425143022-a8b7c6'      # doubao-1.5-pro-32k 的接入点
3
4THINK_KEYWORDS = ('为什么', '原因', '归因', '分析一下', '对比', '怎么判断', '复盘')
5
6def route_endpoint(question: str) -> str:
7    return THINKING_EP if any(k in question for k in THINK_KEYWORDS) else NORMAL_EP
8
9def ask_auto(question: str):
10    resp = client.chat.completions.create(
11        model=route_endpoint(question),
12        messages=[{'role': 'user', 'content': question}],
13        temperature=0.3 if route_endpoint(question) == THINKING_EP else 0.7,
14        max_tokens=1200 if route_endpoint(question) == THINKING_EP else 800,
15    )
16    return resp.choices[0].message.content

1{
2  "msgtype": "text",
3  "text": { "content": "豆包助手：今日文案已生成 12 条，等待审核" },
4  "at": { "atMobiles": ["13800138000"], "isAtAll": false }
5}

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "title": "母婴品类文案 v3",
5    "text": "### 豆包助手 · 三条候选\n\n**1. 温情向**\n> 宝宝的第一口辅食，值得被认真对待\n\n**2. 功能向**\n> 30 秒冲泡，营养不流失\n\n**3. 促销向**\n> 首发 <font color=#FF0000>5 折</font>，仅限前 500 单\n\n[查看全部 12 条](https://doc.example.com/copy/881)"
6  }
7}

1{
2  "msgtype": "link",
3  "link": {
4    "title": "豆包助手：用户反馈周报",
5    "text": "本周 1247 条反馈，负面情绪占比 18%，环比上升 4 个百分点",
6    "picUrl": "https://example.com/report.png",
7    "messageUrl": "https://bi.example.com/feedback/week"
8  }
9}

1{
2  "msgtype": "actionCard",
3  "actionCard": {
4    "title": "豆包助手：检出 3 条高危客诉",
5    "text": "### 需要立刻处理\n\n- **#8821**：物流延误 9 天，用户要求赔付\n- **#8822**：收到破损商品\n- **#8825**：客服态度投诉",
6    "hideAvatar": "0",
7    "btnOrientation": "0",
8    "btns": [
9      { "title": "查看全部", "actionURL": "https://cs.example.com/high" },
10      { "title": "派给值班", "actionURL": "https://cs.example.com/assign" }
11    ]
12  }
13}

1{
2  "msgtype": "feedCard",
3  "feedCard": {
4    "links": [
5      { "title": "豆包日报：今日产出 47 条文案", "messageURL": "https://doc.example.com/today", "picURL": "https://example.com/1.png" },
6      { "title": "客诉周报：负面情绪上升 4pp", "messageURL": "https://bi.example.com/w", "picURL": "https://example.com/2.png" }
7    ]
8  }
9}

1import time, threading
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()

1def push_daily_copies(copies: list, title='豆包助手 · 今日文案'):
2    body = '\n\n'.join(f'**{i}. {c}**' for i, c in enumerate(copies, 1))
3    limiter.acquire()
4    send_markdown(body, title=title)      # 一次发完，不占 12 个名额

1CorpId:     ding2b3c4d5e6f7a8b9c
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
2import dingtalk_stream
3from dingtalk_stream import AckMessage
4
5class DoubaoCardBot(dingtalk_stream.ChatbotHandler):
6    def __init__(self):
7        super().__init__()
8        self.replier = dingtalk_stream.AICardReplier()
9
10    async def process(self, callback: dingtalk_stream.CallbackMessage):
11        incoming = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
12        question = incoming.text.content.strip()
13
14        # 1. 先建卡片占位：用户立刻看到标题栏，回调不超时
15        card = self.replier.create_and_send_card(incoming, title='豆包正在写…')
16
17        # 2. 流式推送，攒够 20 字再刷一次，避免闪屏
18        buf = ''
19        for chunk in stream_doubao(question):
20            buf += chunk
21            if len(buf) >= 20:
22                self.replier.replay_streaming(card.card_id, buf, finished=False)
23
24        # 3. 结束帧（漏了卡片一直转圈）
25        self.replier.replay_streaming(card.card_id, buf, finished=True)
26        return AckMessage.STATUS_OK, 'ok'
27
28def stream_doubao(question: str):
29    ep = route_endpoint(question)
30    resp = client.chat.completions.create(
31        model=ep,
32        messages=[{'role': 'user', 'content': question}],
33        stream=True, temperature=0.3 if ep == THINKING_EP else 0.7,
34        max_tokens=1200 if ep == THINKING_EP else 800
35    )
36    for chunk in resp:
37        if chunk.choices and chunk.choices[0].delta.content:
38            yield chunk.choices[0].delta.content
39
40credential = dingtalk_stream.Credential(os.environ['DINGTALK_APPKEY'],
41                                        os.environ['DINGTALK_APPSECRET'])
42client_stream = dingtalk_stream.DingTalkStreamClient(credential)
43client_stream.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
44                                        DoubaoCardBot())
45client_stream.start_forever()

1tools = [{
2    'type': 'function',
3    'function': {
4        'name': 'query_logistics',
5        'description': '根据订单号查询物流状态',
6        'parameters': {
7            'type': 'object',
8            'properties': {
9                'order_id': {'type': 'string', 'description': '订单号，如 SO20260425001'}
10            },
11            'required': ['order_id']
12        }
13    }
14}]
15
16def ask_with_tools(question: str):
17    resp = client.chat.completions.create(
18        model=NORMAL_EP,
19        messages=[{'role': 'user', 'content': question}],
20        tools=tools,
21        tool_choice='auto',
22        temperature=0.2            # 工具调用场景温度要低
23    )
24    msg = resp.choices[0].message
25
26    if msg.tool_calls:
27        # 模型决定调工具，本地执行后再回传
28        for call in msg.tool_calls:
29            args = json.loads(call.function.arguments)
30            result = query_logistics_internal(args['order_id'])   # 你的内部函数
31
32        follow = client.chat.completions.create(
33            model=NORMAL_EP,
34            messages=[
35                {'role': 'user', 'content': question},
36                msg,
37                {'role': 'tool', 'tool_call_id': call.id, 'content': json.dumps(result)}
38            ],
39            temperature=0.3
40        )
41        return follow.choices[0].message.content
42    return msg.content

1import hashlib
2
3_cache = {}
4
5def normalized_key(q: str) -> str:
6    """归一化：去标点、去空格、转小写，再排序关键词"""
7    import re
8    words = sorted(re.findall(r'[\u4e00-\u9fa5a-zA-Z0-9]+', q.lower()))
9    return hashlib.md5(''.join(words).encode()).hexdigest()
10
11def ask_cached(question: str, ttl=3600):
12    key = normalized_key(question)
13    hit = _cache.get(key)
14    if hit and time.time() - hit[0] < ttl:
15        return hit[1]
16    answer = ask_auto(question)
17    _cache[key] = (time.time(), answer)
18    return answer

1@豆包助手
2
3我要给一款母婴辅食机写电商文案。
4
5产品信息：
6- 30 秒快速蒸煮，保留营养
7- 一键清洗，可拆卸刀头
8- 静音设计，45 分贝
9- 售价 399，首发 5 折，限前 500 单
10
11要求：
121. 写 3 条不同风格的：温情向、功能向、促销向
132. 每条不超过 40 字，口语化，像妈妈之间互相推荐
143. 禁用这些词：不容错过、震撼上市、史上最低、你值得拥有、赶紧下单
154. 促销向必须写清折扣和限制条件
16
17输出格式：风格名 + 文案正文，不要解释。

1你是用户反馈分析助手。下面是今天的 50 条用户评论。
2
3请做三件事：
41. 按类别归类：物流 / 商品质量 / 客服态度 / 价格 / 功能需求 / 其他
52. 每条标注情绪：正面 / 中性 / 负面
63. 挑出需要立刻人工跟进的（负面且涉及赔付、破损、安全）
7
8输出：
9- 先给一行汇总：各类别条数，负面占比
10- 再逐条列出，格式「编号 | 类别 | 情绪 | 15 字内摘要」
11- 最后单独列出需跟进的编号
12
13评论：
14[粘贴 50 条评论]

1@豆包助手
2
3用户问：「我下单 9 天了还没收到，你们是不是虚假发货？」
4用户情绪：愤怒，已在评论区留负面评价
5
6请生成回复话术：
71. 第一版：安抚 + 说明物流规则 + 给出明确时间承诺（100 字内）
82. 第二版：在上一版基础上加补偿方案（50 字内补充）
93. 标注哪些话不能说（如「这不是我们的问题」）
10
11要求：语气真诚不卑微，不用「非常抱歉给您带来不便」这种套话。

1@豆包助手
2
3本周数据：
4- 曝光 128 万（上周 131 万，-2%）
5- 点击率 3.1%（上周 3.8%，-18%）
6- 加购率 12.4%（上周 12.1%，+2%）
7- 支付转化率 4.2%（上周 4.9%，-14%）
8- 客单价 268（上周 271，-1%）
9- 本周改动：主图换了一版、详情页删了 2 屏、新增了一个满 199 减 20
10
11请分析：
121. 最大的问题出在哪一步？给出判断依据
132. 三个最可能的原因，按概率排序
143. 每个原因的验证方法（一个数据切片或一个 A/B 测试设计）
154. 如果只能做一个改动，改什么？
16
17600 字以内，用数据说话。

1# 招式一：非必要不走 Thinking 版（差 3 倍价格）
2model = THINKING_EP if any(k in q for k in THINK_KEYWORDS) else NORMAL_EP
3
4# 招式二：限制 max_tokens 到 600-800（群里没人看长文）
5max_tokens=800
6
7# 招式三：归一化结果缓存（实测命中率 27%）
8answer = ask_cached(question)
9
10# 招式四：批量文案聚合成一条发（不占 20 条/分钟的额度）
11push_daily_copies(copies)

1阶段 1（入门）：群机器人 Webhook + markdown 推送                  ← 你现在在这里
2阶段 2（进阶）：方舟接入点 + 加签 + 令牌桶限流 + 五种消息类型
3阶段 3（高级）：企业内部应用 + Thinking 分流 + AI 卡片流式
4阶段 4（专家）：Function Calling + 结果缓存 + 多接入点配额治理