Kimi K3 接入 钉钉 完整指南：200 万字长文档问答 + 20 条踩坑实测

1❌ 纯短问答（一句「Python 怎么写文件」也用 K3 是浪费，用 moonshot-v1-128k）
2❌ 高并发批量推送（长上下文模型的并发额度更紧）
3❌ 需要极强代码生成的场景（K3 代码能力中上，不如 DeepSeek V4.1 Flash）

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":"Kimi 助手已上线，这是连通性测试"}}'

1import os, requests
2
3MOONSHOT_KEY = os.environ['MOONSHOT_API_KEY']
4BASE = 'https://api.moonshot.cn/v1'
5
6def upload_file(path: str) -> str:
7    """上传文件拿到 file_id，后续多轮问答复用，省 token"""
8    with open(path, 'rb') as f:
9        r = requests.post(
10            f'{BASE}/files',
11            headers={'Authorization': f'Bearer {MOONSHOT_KEY}'},
12            files={'file': (os.path.basename(path), f),
13                   'purpose': ('file-extract',)},   # 这个 purpose 会触发内容抽取
14            timeout=120
15        )
16    r.raise_for_status()
17    return r.json()['id']            # 形如 file-xxxx
18
19def ask_with_file(file_id: str, question: str) -> str:
20    """用 file_id 提问：messages 里只放引用，不重复传全文"""
21    r = requests.post(
22        f'{BASE}/chat/completions',
23        headers={'Authorization': f'Bearer {MOONSHOT_KEY}',
24                 'Content-Type': 'application/json'},
25        json={
26            'model': 'kimi-k3-turbo',
27            'messages': [
28                # system 里塞 file_id，Kimi 会把文件内容当背景知识
29                {'role': 'system',
30                 'content': f'你是文档助手。文档内容见 file-{file_id}，'
31                            f'回答必须基于文档，不确定就说不确定。'},
32                {'role': 'user', 'content': question}
33            ],
34            'temperature': 0.3,
35            'max_tokens': 800
36        },
37        timeout=180
38    )
39    r.raise_for_status()
40    return r.json()['choices'][0]['message']['content']

1def build_cached_messages(file_id: str, question: str):
2    """用 cache_control 标记可缓存前缀"""
3    return [
4        {'role': 'system',
5         'content': f'你是文档助手，文档见 file-{file_id}。',
6         'cache_control': {'type': 'ephemeral'}},   # 这一段会被缓存
7        {'role': 'user', 'content': question}
8    ]

1import threading
2
3def keep_alive(file_id: str, interval: int = 180):
4    def ping():
5        try:
6            ask_with_file(file_id, '好的')
7        except Exception:
8            pass
9        threading.Timer(interval, ping).start()
10    ping()

1{
2  "msgtype": "text",
3  "text": { "content": "Kimi 助手：招标文件解析完成，共 268 页" },
4  "at": { "atMobiles": ["13800138000"], "isAtAll": false }
5}

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "title": "招标文件解析结果",
5    "text": "### Kimi 助手 · 验收标准矛盾检测\n\n- **第三章 3.4**：要求验收周期 <font color=#008000>15 个工作日</font>\n- **第七章 7.2**：要求验收周期 30 个自然日\n- **结论**：<font color=#FF0000>存在矛盾</font>，两者相差近一倍\n\n> 建议以第三章为准，并在答疑阶段向招标方确认\n\n[查看原文 P88](https://doc.example.com/tender#p88)"
6  }
7}

1{
2  "msgtype": "link",
3  "link": {
4    "title": "Kimi 助手：合同解析完成",
5    "text": "128 页合同已解析，检出 4 处风险条款",
6    "picUrl": "https://example.com/contract.png",
7    "messageUrl": "https://doc.example.com/c/881"
8  }
9}

1{
2  "msgtype": "actionCard",
3  "actionCard": {
4    "title": "Kimi 助手：检出 4 处风险条款",
5    "text": "### 合同风险初审\n\n- **5.2**：违约金 30%，超 20% 红线\n- **8.1**：知识产权归属未约定",
6    "hideAvatar": "0",
7    "btnOrientation": "0",
8    "btns": [
9      { "title": "查看原文", "actionURL": "https://doc.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": "Kimi 日报：今日解析 6 份文档", "messageURL": "https://doc.example.com/today", "picURL": "https://example.com/1.png" },
6      { "title": "《招标文件》检出 2 处矛盾", "messageURL": "https://doc.example.com/t/2", "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 send_long_answer(text: str, title: str = 'Kimi 助手', limit: int = 4000):
2    """按段落切分，避免单条超长被截断"""
3    chunks, buf = [], ''
4    for para in text.split('\n\n'):
5        if len(buf) + len(para) + 2 > limit:
6            chunks.append(buf)
7            buf = para
8        else:
9            buf = f'{buf}\n\n{para}' if buf else para
10    if buf:
11        chunks.append(buf)
12
13    for i, chunk in enumerate(chunks, 1):
14        limiter.acquire()                    # 每条都要拿令牌
15        send_markdown(chunk, title=f'{title}（{i}/{len(chunks)}）')

1CorpId:     ding5f4e3d2c1b0a9988
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
21            return self._token

1pip install dingtalk-stream

1import os
2import dingtalk_stream
3from dingtalk_stream import AckMessage
4from openai import OpenAI
5
6kimi = OpenAI(api_key=os.environ['MOONSHOT_API_KEY'],
7              base_url='https://api.moonshot.cn/v1')
8
9def stream_kimi(file_id: str, question: str):
10    """Kimi 支持 SSE 流式，逐块 yield"""
11    resp = kimi.chat.completions.create(
12        model='kimi-k3-turbo',
13        messages=[
14            {'role': 'system', 'content': f'文档见 file-{file_id}，'
15                                          f'回答控制在 400 字，必须引用页码。',
16             'cache_control': {'type': 'ephemeral'}},
17            {'role': 'user', 'content': question}
18        ],
19        stream=True, temperature=0.3, max_tokens=800
20    )
21    for chunk in resp:
22        if chunk.choices and chunk.choices[0].delta.content:
23            yield chunk.choices[0].delta.content
24
25class KimiCardBot(dingtalk_stream.ChatbotHandler):
26    def __init__(self):
27        super().__init__()
28        self.replier = dingtalk_stream.AICardReplier()
29        self.file_ids = {}                    # conversationId -> file_id
30
31    async def process(self, callback: dingtalk_stream.CallbackMessage):
32        incoming = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
33
34        # 用户 @ 机器人并附带文件时，先上传拿到 file_id
35        file_id = self.file_ids.get(incoming.conversation_id)
36        question = incoming.text.content.strip()
37
38        # 1. 先建卡片占位：用户立刻看到标题栏，回调也不超时
39        card = self.replier.create_and_send_card(incoming, title='Kimi 正在读文档…')
40
41        # 2. 流式推送，攒够 20 字再刷，避免闪屏
42        buf = ''
43        for chunk in stream_kimi(file_id, question):
44            buf += chunk
45            if len(buf) >= 20:
46                self.replier.replay_streaming(card.card_id, buf, finished=False)
47
48        # 3. 结束帧（漏了卡片一直转圈）
49        self.replier.replay_streaming(card.card_id, buf, finished=True)
50        return AckMessage.STATUS_OK, 'ok'
51
52credential = dingtalk_stream.Credential(os.environ['DINGTALK_APPKEY'],
53                                        os.environ['DINGTALK_APPSECRET'])
54client = dingtalk_stream.DingTalkStreamClient(credential)
55client.register_callback_handler(dingtalk_stream.chatbot.ChatbotMessage.TOPIC,
56                                 KimiCardBot())
57client.start_forever()

1@Kimi 助手
2
3这是刚收到的招标文件（已上传）。
4请回答：
51. 投标截止时间、开标时间分别是哪天？
62. 对投标人资质有哪些硬性要求？逐条列出
73. 第三章和第七章对验收标准的表述是否一致？不一致请指出差异
84. 有哪些条款对我们明显不利？
9
10每条答案后面标注页码，400 字以内。

1你是战略复盘助手。下面是本季度 23 次周会的完整记录（已上传）。
2
3请输出：
41. 本季度被反复提及但始终没有结论的话题（列出话题 + 出现次数 + 最近一次讨论时间）
52. 上季度定下的 8 项 OKR，哪些在记录中能找到明确进展？哪些完全没被提到？
63. 出现频率最高的 5 个阻塞词（如「等排期」「依赖」「人力不足」），各出现多少次
7
8用 Markdown 表格外的形式输出（钉钉不支持表格），500 字以内。

1你是竞品分析助手。下面是 5 家竞品近半年的产品更新日志合集（已上传）。
2
3请输出：
41. 每家最主要的 3 个功能更新（用「竞品名 - 功能 - 时间」格式）
52. 哪些功能是 3 家以上都在做的？说明这是行业趋势
63. 哪些是我们还没有但 2 家以上已上线的？标为「能力缺口」
74. 有没有哪家的更新日志透露出下一步方向？
8
9600 字以内，不要复述原文。

1你是法务初审助手。下面是完整合同（已上传）。
2
3请输出风险项，每条一行，格式「条款号 | 风险描述 | 建议修改 | 页码」：
4
51. 付款条件：预付款比例是否过高、验收后付款周期是否过长
62. 违约责任：违约金比例是否超过合同额 20%
73. 知识产权归属是否明确
84. 争议解决方式与管辖法院是否合理
95. 是否存在单方面变更条款的权利
10
11无风险输出「LGTM」。

1阶段 1（入门）：群机器人 Webhook + markdown 推送              ← 你现在在这里
2阶段 2（进阶）：/v1/files 上传 + Context Cache + 令牌桶限流
3阶段 3（高级）：企业内部应用 + token 缓存 + AI 卡片流式
4阶段 4（专家）：文档库常驻 + Function Calling + 多模型分流