DeepSeek V4.1 Flash 接入 钉钉 完整指南

1❌ 跨 30+ 文件的架构评审（需要 35B+ 激活参数的全局推理）
2❌ 生产代码的深度安全审计（安全对齐比 V4 Pro 弱约 4 个百分点）
3❌ 要求严格 JSON Schema 输出的关键业务（结构化输出稳定性弱于 GPT-5）

1import time
2import hmac
3import hashlib
4import base64
5import urllib.parse
6
7def build_signed_url(webhook: str, secret: str) -> str:
8    # 1. 毫秒级时间戳（13 位）
9    timestamp = str(round(time.time() * 1000))
10    # 2. 拼接待签名字符串：timestamp + 换行 + secret
11    string_to_sign = f'{timestamp}\n{secret}'
12    # 3. 以 secret 为密钥做 HMAC-SHA256
13    hmac_code = hmac.new(
14        secret.encode('utf-8'),
15        string_to_sign.encode('utf-8'),
16        digestmod=hashlib.sha256
17    ).digest()
18    # 4. base64 编码
19    sign = base64.b64encode(hmac_code).decode('utf-8')
20    # 5. urlencode —— 这一步最容易漏
21    sign_quoted = urllib.parse.quote_plus(sign)
22    return f'{webhook}&timestamp={timestamp}&sign={sign_quoted}'

1const crypto = require('crypto');
2
3function buildSignedUrl(webhook, secret) {
4  // 毫秒级时间戳
5  const timestamp = Date.now().toString();
6  const stringToSign = `${timestamp}\n${secret}`;
7  const hmacCode = crypto
8    .createHmac('sha256', secret)
9    .update(stringToSign)
10    .digest('base64');
11  const sign = encodeURIComponent(hmacCode); // 等价于 quote_plus
12  return `${webhook}&timestamp=${timestamp}&sign=${sign}`;
13}

1# 查看自己的出口 IP
2curl https://api.ipify.org

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

1{"errcode":0,"errmsg":"ok"}

1import os, json, requests
2
3WEBHOOK = os.environ['DINGTALK_WEBHOOK']   # 不带 timestamp/sign 的原始地址
4SECRET  = os.environ['DINGTALK_SECRET']    # SEC 开头，没开加签就留空串
5
6def send_text(content: str, at_mobiles=None, at_all=False):
7    # 关键词兜底：确保正文一定包含关键词
8    if 'DeepSeek' not in content:
9        content = f'DeepSeek 助手：{content}'
10
11    url = build_signed_url(WEBHOOK, SECRET) if SECRET else WEBHOOK
12
13    payload = {
14        'msgtype': 'text',
15        'text': {'content': content},
16        'at': {
17            'atMobiles': at_mobiles or [],   # 手机号列表，人必须在群里
18            'isAtAll': at_all                # True 时覆盖 atMobiles
19        }
20    }
21    r = requests.post(url, json=payload, timeout=10)
22    result = r.json()
23    if result.get('errcode') != 0:
24        raise RuntimeError(f'钉钉返回错误：{result}')
25    return result
26
27send_text('连通性测试成功', at_mobiles=['13800138000'])

1{
2  "msgtype": "text",
3  "text": { "content": "DeepSeek 助手：构建 #4821 已通过" },
4  "at": { "atMobiles": ["13800138000"], "isAtAll": false }
5}

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "title": "构建结果通知",
5    "text": "### DeepSeek 助手 · 构建完成\n\n- **分支**：`release/2.8`\n- **耗时**：3 分 12 秒\n- **结果**：<font color=#008000>全部通过</font>\n\n> 由 V4.1 Flash 自动汇总 47 条测试日志\n\n[查看流水线](https://ci.example.com/runs/4821)"
6  },
7  "at": { "atMobiles": [], "isAtAll": false }
8}

1{
2  "msgtype": "link",
3  "link": {
4    "title": "DeepSeek 助手：线上告警 P1",
5    "text": "支付网关 5xx 错误率 12%，持续 8 分钟",
6    "picUrl": "https://example.com/alert.png",
7    "messageUrl": "https://grafana.example.com/d/pay/alert"
8  }
9}

1{
2  "msgtype": "actionCard",
3  "actionCard": {
4    "title": "DeepSeek 助手：需要你确认发布",
5    "text": "### 发布确认\n\n版本 **v2.8.1** 已构建完成，请确认是否发布到生产环境。",
6    "singleTitle": "立即发布",
7    "singleURL": "https://deploy.example.com/confirm?run=4821",
8    "btnOrientation": "0"
9  }
10}

1{
2  "msgtype": "actionCard",
3  "actionCard": {
4    "title": "DeepSeek 助手：请选择处理方式",
5    "text": "### 告警：订单服务 RT 升高\n\nV4.1 Flash 分析后给出三个候选动作。",
6    "hideAvatar": "0",
7    "btnOrientation": "0",
8    "btns": [
9      { "title": "查看监控", "actionURL": "https://grafana.example.com/d/order" },
10      { "title": "回滚版本", "actionURL": "https://deploy.example.com/rollback" },
11      { "title": "忽略本次", "actionURL": "https://ops.example.com/mute/4821" }
12    ]
13  }
14}

1{
2  "msgtype": "feedCard",
3  "feedCard": {
4    "links": [
5      { "title": "DeepSeek 日报：今日 3 个 P1 已闭环", "messageURL": "https://ops.example.com/d/1", "picURL": "https://example.com/1.png" },
6      { "title": "支付网关优化上线，RT 降 34%", "messageURL": "https://ops.example.com/d/2", "picURL": "https://example.com/2.png" },
7      { "title": "用户服务新增 2 个慢查询告警", "messageURL": "https://ops.example.com/d/3", "picURL": "https://example.com/3.png" }
8    ]
9  }
10}

1import os, time, json, threading
2import requests
3from collections import deque
4
5DEEPSEEK_KEY = os.environ['DEEPSEEK_API_KEY']
6DEEPSEEK_URL = 'https://api.deepseek.com/v1/chat/completions'
7MODEL = 'deepseek-v4.1-flash'
8
9# ---- 令牌桶：钉钉每个机器人每分钟最多 20 条 ----
10class RateLimiter:
11    def __init__(self, max_calls=20, period=60):
12        self.max_calls, self.period = max_calls, period
13        self.calls = deque()
14        self.lock = threading.Lock()
15
16    def acquire(self):
17        with self.lock:
18            now = time.time()
19            while self.calls and now - self.calls[0] > self.period:
20                self.calls.popleft()
21            if len(self.calls) >= self.max_calls:
22                sleep_for = self.period - (now - self.calls[0]) + 0.5
23                time.sleep(sleep_for)
24                return self.acquire()
25            self.calls.append(now)
26
27limiter = RateLimiter()
28
29# ---- 调 DeepSeek V4.1 Flash ----
30def ask_deepseek(question: str, system: str = '') -> str:
31    messages = []
32    if system:
33        messages.append({'role': 'system', 'content': system})
34    messages.append({'role': 'user', 'content': question})
35
36    r = requests.post(
37        DEEPSEEK_URL,
38        headers={'Authorization': f'Bearer {DEEPSEEK_KEY}'},
39        json={
40            'model': MODEL,
41            'messages': messages,
42            'temperature': 0.3,      # 群聊场景要稳定，别开高
43            'max_tokens': 1024,      # 群里长篇大论没人看
44            'stream': False
45        },
46        timeout=60
47    )
48    r.raise_for_status()
49    return r.json()['choices'][0]['message']['content']
50
51# ---- 发到钉钉 ----
52def send_markdown(text: str, title: str = 'DeepSeek 助手'):
53    if 'DeepSeek' not in text:
54        text = f'DeepSeek 助手\n\n{text}'
55    url = build_signed_url(WEBHOOK, SECRET) if SECRET else WEBHOOK
56    payload = {
57        'msgtype': 'markdown',
58        'markdown': {'title': title, 'text': text}
59    }
60    r = requests.post(url, json=payload, timeout=10)
61    result = r.json()
62    if result.get('errcode') != 0:
63        raise RuntimeError(f'钉钉错误：{result}')
64    return result
65
66def answer_to_group(question: str):
67    limiter.acquire()                      # 先拿令牌，超 20 条自动睡
68    answer = ask_deepseek(
69        question,
70        system='你是钉钉群里的技术助手。回答控制在 300 字以内，用 Markdown，'
71               '先给结论再给步骤，不确定就说不确定。'
72    )
73    send_markdown(f'**Q：** {question}\n\n---\n\n{answer}')
74
75if __name__ == '__main__':
76    answer_to_group('K8s Pod 一直 CrashLoopBackOff，排查思路是什么？')

1pip install dingtalk-stream

1import os, json
2import dingtalk_stream
3from dingtalk_stream import AckMessage
4
5DEEPSEEK_KEY = os.environ['DEEPSEEK_API_KEY']
6MODEL = 'deepseek-v4.1-flash'
7
8def ask_deepseek_stream(question: str):
9    """DeepSeek 支持 SSE：stream=True 后逐块 yield 文本增量"""
10    import requests
11    with requests.post(
12        'https://api.deepseek.com/v1/chat/completions',
13        headers={'Authorization': f'Bearer {DEEPSEEK_KEY}'},
14        json={'model': MODEL,
15              'messages': [{'role': 'user', 'content': question}],
16              'stream': True, 'temperature': 0.3},
17        stream=True, timeout=120
18    ) as r:
19        for line in r.iter_lines():
20            if not line or not line.startswith(b'data: '):
21                continue
22            data = line[6:]
23            if data == b'[DONE]':
24                break
25            delta = json.loads(data)['choices'][0]['delta']
26            if 'content' in delta:
27                yield delta['content']
28
29class DeepSeekCardBot(dingtalk_stream.ChatbotHandler):
30    def __init__(self):
31        super().__init__()
32        self.card_replier = dingtalk_stream.AICardReplier()
33
34    async def process(self, callback: dingtalk_stream.CallbackMessage):
35        incoming = dingtalk_stream.ChatbotMessage.from_dict(callback.data)
36        question = incoming.text.content.strip()
37
38        # 1. 创建一张 AI 卡片（用户立刻看到标题栏，不再干等）
39        card = self.card_replier.create_and_send_card(
40            incoming, title='DeepSeek 助手正在思考…'
41        )
42
43        # 2. 流式推送：每攒够一段就 update 一次卡片
44        buffer = ''
45        for chunk in ask_deepseek_stream(question):
46            buffer += chunk
47            if len(buffer) >= 20:            # 攒 20 字再推，避免刷屏
48                self.card_replier.replay_streaming(card.card_id, buffer, finished=False)
49
50        # 3. 结束帧（漏了这一步卡片会一直转圈）
51        self.card_replier.replay_streaming(card.card_id, buffer, finished=True)
52        return AckMessage.STATUS_OK, 'ok'
53
54# 启动常驻进程
55credential = dingtalk_stream.Credential(
56    os.environ['DINGTALK_APPKEY'], os.environ['DINGTALK_APPSECRET']
57)
58client = dingtalk_stream.DingTalkStreamClient(credential)
59client.register_callback_handler(
60    dingtalk_stream.chatbot.ChatbotMessage.TOPIC, DeepSeekCardBot()
61)
62client.start_forever()

1def stream_answer_to_group(question: str, chunk_interval: float = 1.5):
2    """群机器人伪流式：每 1.5 秒更新一次消息内容"""
3    buffer = ''
4    last_push = 0.0
5    for chunk in ask_deepseek_stream(question):
6        buffer += chunk
7        now = time.time()
8        if now - last_push >= chunk_interval and len(buffer) > 0:
9            limiter.acquire()
10            send_markdown(buffer, title='DeepSeek 助手（生成中…）')
11            last_push = now
12    limiter.acquire()
13    send_markdown(buffer, title='DeepSeek 助手')

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

1import time, threading
2import requests
3
4TOKEN_URL = 'https://oapi.dingtalk.com/gettoken'
5
6class TokenManager:
7    """带提前量 + 线程锁的 access_token 缓存"""
8    def __init__(self, appkey: str, appsecret: str, early: int = 300):
9        self.appkey, self.appsecret = appkey, appsecret
10        self.early = early          # 提前 5 分钟过期，避开边界失效
11        self._token = None
12        self._expire_at = 0
13        self._lock = threading.Lock()
14
15    def get_token(self) -> str:
16        with self._lock:              # 并发下只刷新一次
17            if self._token and time.time() < self._expire_at:
18                return self._token
19
20            r = requests.get(TOKEN_URL, params={
21                'appkey': self.appkey,      # 注意是全小写 appkey
22                'appsecret': self.appsecret
23            }, timeout=10)
24            data = r.json()
25            if data.get('errcode') != 0:
26                raise RuntimeError(f'获取 token 失败：{data}')
27
28            self._token = data['access_token']
29            self._expire_at = time.time() + data['expires_in'] - self.early
30            return self._token
31
32token_mgr = TokenManager(os.environ['DINGTALK_APPKEY'],
33                         os.environ['DINGTALK_APPSECRET'])

1SEND_URL = 'https://oapi.dingtalk.com/topapi/im/chat/scencegroup/message/send_v2'
2
3def send_by_app(open_conversation_id: str, text: str):
4    token = token_mgr.get_token()
5    r = requests.post(
6        SEND_URL,
7        params={'access_token': token},
8        json={
9            'msgtype': 'markdown',
10            'markdown': {'title': 'DeepSeek 助手', 'text': text},
11            'open_conversation_id': open_conversation_id,  # 群的唯一 ID
12            'robot_code': os.environ['DINGTALK_APPKEY']
13        },
14        timeout=10
15    )
16    result = r.json()
17    if result.get('errcode') != 0:
18        raise RuntimeError(f'发送失败：{result}')
19    return result

1@DeepSeek 助手
2
3线上订单服务今天 14:20 开始 RT 从 80ms 涨到 900ms，
4同时 MySQL 慢查询日志里出现大量 SELECT * FROM order_item WHERE order_id IN (...)。
5请用 300 字以内给出：
61. 最可能的 3 个原因，按概率排序
72. 每个原因的验证方法（一条命令或一个 SQL）
83. 临时止血手段

1你是技术团队的日报助手。下面是今天 Git 提交的原始记录：
2
3[粘贴 git log --since="1 day ago" --pretty=format:"%an|%s" 的输出]
4
5请输出：
61. 按人分组，每人 3 条以内，用「动词 + 对象」句式
72. 单独列出「需要关注」的部分（含 fix/hotfix/revert 关键字的提交）
83. 整体用 Markdown，不超过 400 字

1你是 SRE 助手。收到如下 Prometheus 告警，请判断是否为真故障：
2
3名称：HighErrorRate
4实例：payment-gateway-7d9f8c
5表达式：rate(http_requests_total{status=~"5xx"}[5m]) > 0.1
6当前值：0.127
7持续时间：8m
8关联信息：3 分钟前有过一次 release/2.8.1 的发布
9
10请输出：
111. 是真故障还是发布引起（一句话结论）
122. 建议的下一步（一个动作）
133. 是否需要 @ 值班人（是/否）

1你是资深 Go 后端评审。请评审下面的 diff，只输出：
21. 明确的 bug（行号 + 问题描述）
32. 资源泄漏 / 并发安全问题（行号 + 描述）
43. 命名或风格问题（最多 2 条）
5
6不要输出夸奖，不要输出「建议增加测试」这种泛泛建议。
7如果没问题就输出「LGTM」。
8
9diff:
10[粘贴 git diff 内容]

1import itertools, threading
2
3# 方案一：多机器人轮询（推荐，扩容最省事）
4WEBHOOKS = [os.environ[f'DINGTALK_WEBHOOK_{i}'] for i in range(1, 5)]  # 4 个 = 80 条/分钟
5_rr = itertools.cycle(WEBHOOKS)
6def next_webhook():
7    return next(_rr)
8
9# 方案二：同向消息聚合（适合告警场景）
10_pending = []
11
12def enqueue(msg):
13    _pending.append(msg)
14    if len(_pending) == 1:
15        threading.Timer(30.0, flush).start()   # 30 秒窗口聚合
16
17def flush():
18    if not _pending:
19        return
20    combined = '\n\n'.join(_pending)
21    _pending.clear()
22    send_markdown(combined, title='DeepSeek 助手 · 告警聚合')

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