DeepSeek V4.1 Flash 接入微信完整指南：企业微信群机器人 + 公众号双路径实测

1✅ 群内短问答（SQL 生成、日志解读、术语解释）—— 便宜是王道
2✅ 定时播报（日报、周报、巡检结果汇总）—— 模板化程度高，8B 足够
3✅ 告警降噪（把一堆报错压缩成人话）—— 中等推理 + 短输出
4✅ FAQ 客服（把知识库切片后检索回答）—— 128K 上下文够塞一本手册
5✅ 公众号自动回复（用户提问长度通常 < 500 字）
6❌ 合同全文逐条风险审查（需要 35B 激活的深度推理）
7❌ 跨几十页 PDF 的精算比对（上 V4 Pro）
8❌ 需要严格 JSON Schema 输出的关键业务（建议加 response_format + 人工兜底）

1curl https://api.deepseek.com/v1/chat/completions \
2  -H "Content-Type: application/json" \
3  -H "Authorization: Bearer $DEEPSEEK_API_KEY" \
4  -d '{
5        "model": "deepseek-v4.1-flash",
6        "messages": [{"role":"user","content":"用一句话说明你是谁"}],
7        "max_tokens": 100
8      }'

1curl 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key' \
2   -H 'Content-Type: application/json' \
3   -d '{
4         "msgtype": "text",
5         "text": {
6           "content": "DeepSeek V4.1 Flash 已就位，Hello World"
7         }
8       }'

1生产环境：加签 + IP 白名单（双保险，缺一不可）
2内网固定 IP：IP 白名单为主，加签为辅
3本地开发调试：先用「自定义关键词」跑通链路，上线前必须换成加签

1import time
2import hmac
3import hashlib
4import base64
5import urllib.parse
6import requests
7
8WEBHOOK_KEY = "693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa"
9SECRET = "SEC-xxxxxxxxxxxx"      # 机器人详情页的 secret，不是 key
10
11def gen_sign(secret: str, timestamp: str) -> str:
12    """生成企业微信机器人加签。
13    注意三件事：timestamp 必须是秒级字符串；中间是单个 \n；最后要 urlencode。"""
14    string_to_sign = f"{timestamp}\n{secret}"
15    hmac_code = hmac.new(
16        secret.encode("utf-8"),
17        string_to_sign.encode("utf-8"),
18        hashlib.sha256
19    ).digest()
20    return urllib.parse.quote_plus(base64.b64encode(hmac_code).decode("utf-8"))
21
22def send_wecom(content: str, msgtype: str = "text"):
23    timestamp = str(int(time.time()))   # 秒级！不是毫秒
24    sign = gen_sign(SECRET, timestamp)
25    url = (
26        f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
27        f"?key={WEBHOOK_KEY}&timestamp={timestamp}&sign={sign}"
28    )
29    payload = {"msgtype": msgtype, "text": {"content": content}}
30    resp = requests.post(url, json=payload, timeout=10)
31    result = resp.json()
32    if result.get("errcode") != 0:
33        raise RuntimeError(f"推送失败: {result}")
34    return result

1const crypto = require('crypto');
2const https = require('https');
3
4const WEBHOOK_KEY = '693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa';
5const SECRET = 'SEC-xxxxxxxxxxxx';
6
7/** 生成签名：timestamp(秒) + '\n' + secret 做 HMAC-SHA256 → base64 → urlencode */
8function genSign(secret, timestamp) {
9  const stringToSign = `${timestamp}\n${secret}`;
10  const hmacCode = crypto
11    .createHmac('sha256', secret)
12    .update(stringToSign)
13    .digest('base64');
14  return encodeURIComponent(hmacCode);
15}
16
17async function sendWecom(content) {
18  const timestamp = Math.floor(Date.now() / 1000).toString(); // 必须除以 1000
19  const sign = genSign(SECRET, timestamp);
20  const url = `https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=${WEBHOOK_KEY}&timestamp=${timestamp}&sign=${sign}`;
21  const body = JSON.stringify({ msgtype: 'text', text: { content } });
22
23  return new Promise((resolve, reject) => {
24    const req = https.request(url, {
25      method: 'POST',
26      headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(body) }
27    }, (res) => {
28      let data = '';
29      res.on('data', (c) => (data += c));
30      res.on('end', () => resolve(JSON.parse(data)));
31    });
32    req.on('error', reject);
33    req.write(body);
34    req.end();
35  });
36}

1import javax.crypto.Mac;
2import javax.crypto.spec.SecretKeySpec;
3import java.net.URLEncoder;
4import java.nio.charset.StandardCharsets;
5import java.util.Base64;
6
7public class WecomSign {
8    public static String genSign(String secret, String timestamp) throws Exception {
9        String stringToSign = timestamp + "\n" + secret;
10        Mac mac = Mac.getInstance("HmacSHA256");
11        mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
12        byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
13        return URLEncoder.encode(Base64.getEncoder().encodeToString(signData), "UTF-8");
14    }
15}

1# app.py  —— 依赖：pip install flask requests
2import os, time, hmac, hashlib, base64, urllib.parse
3import requests
4from flask import Flask, request, jsonify
5
6app = Flask(__name__)
7
8DEEPSEEK_KEY = os.environ["DEEPSEEK_API_KEY"]        # 从环境变量读，绝不硬编码
9DEEPSEEK_URL = "https://api.deepseek.com/v1/chat/completions"
10MODEL_ID     = "deepseek-v4.1-flash"                 # 👈 型号，改这里就换模型
11
12WEBHOOK_KEY  = os.environ["WECOM_WEBHOOK_KEY"]
13WEBHOOK_SECRET = os.environ["WECOM_WEBHOOK_SECRET"]
14
15# 群机器人的系统人设：决定它在群里说话的调性
16SYSTEM_PROMPT = """你是企业微信工作群里的 AI 助手。
17- 回答控制在 300 字以内，群里没人看长文
18- 涉及代码必须给完整可运行片段，用 markdown 围栏包住
19- 不确定就说不确定，不要编造接口名和参数
20- 输出使用 markdown 语法，标题用 ##，重点用 ** 加粗
21"""
22
23def ask_deepseek(question: str) -> str:
24    """调用 DeepSeek V4.1 Flash。temperature 调低保证群里回答稳定。"""
25    resp = requests.post(
26        DEEPSEEK_URL,
27        headers={"Authorization": f"Bearer {DEEPSEEK_KEY}"},
28        json={
29            "model": MODEL_ID,
30            "messages": [
31                {"role": "system", "content": SYSTEM_PROMPT},
32                {"role": "user", "content": question},
33            ],
34            "temperature": 0.3,     # 群播报要稳定，别设 0.8
35            "max_tokens": 1500,      # 群消息太长会被折叠
36            "stream": False,
37        },
38        timeout=30,
39    )
40    resp.raise_for_status()
41    return resp.json()["choices"][0]["message"]["content"]
42
43def push_to_wecom(markdown: str):
44    """推送 markdown 消息到群，带加签。"""
45    timestamp = str(int(time.time()))
46    sign = gen_sign(WEBHOOK_SECRET, timestamp)
47    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
48           f"?key={WEBHOOK_KEY}&timestamp={timestamp}&sign={sign}")
49    payload = {"msgtype": "markdown", "markdown": {"content": markdown[:4096]}}
50    r = requests.post(url, json=payload, timeout=10).json()
51    if r.get("errcode") != 0:
52        app.logger.error("wecom push failed: %s", r)
53    return r
54
55def gen_sign(secret, timestamp):
56    s = f"{timestamp}\n{secret}".encode()
57    return urllib.parse.quote_plus(
58        base64.b64encode(hmac.new(secret.encode(), s, hashlib.sha256).digest()).decode()
59    )
60
61@app.post("/ask")
62def ask():
63    q = (request.json or {}).get("q", "").strip()
64    if not q:
65        return jsonify(err="empty question"), 400
66    answer = ask_deepseek(q)
67    push_to_wecom(f"**Q**: {q}\n\n{answer}")
68    return jsonify(ok=True, answer=answer)
69
70if __name__ == "__main__":
71    app.run(host="0.0.0.0", port=8000)

1export DEEPSEEK_API_KEY=sk-xxx
2export WECOM_WEBHOOK_KEY=693a91f6-xxxx
3export WECOM_WEBHOOK_SECRET=SEC-xxxx
4python app.py
5
6# 另开一个终端
7curl -X POST http://127.0.0.1:8000/ask \
8  -H 'Content-Type: application/json' \
9  -d '{"q":"用 Python 写一个按 URI 聚合 nginx 5xx 日志的脚本"}'

1{
2  "msgtype": "text",
3  "text": {
4    "content": "广州机房 10.0.2.13 磁盘使用率 91%，已触发告警",
5    "mentioned_list": ["@all"],          // @所有人；也可填 userid 数组
6    "mentioned_mobile_list": ["13800138000"]
7  }
8}

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "content": "## 巡检日报 2026-05-07\n> 共 **12** 台主机，异常 <font color=\"warning\">2</font> 台\n\n**TOP 问题**\n1. `10.0.2.13` 磁盘 91%\n2. `10.0.2.44` 内存 swap 持续增长\n\n[查看详情](https://grafana.internal/d/xxx)"
5  }
6}

1{
2  "msgtype": "news",
3  "news": {
4    "articles": [
5      {
6        "title": "DeepSeek V4.1 Flash 接入报告",
7        "description": "200 条真实提问实测，月成本 ¥56",
8        "url": "https://aitoolku.cn/tutorials/deepseek-v4.1-flash-vscode",
9        "picurl": "https://example.com/cover.png"
10      }
11    ]
12  }
13}

1{
2  "msgtype": "template_card",
3  "template_card": {
4    "card_type": "text_notice",
5    "source": { "desc": "DeepSeek 助手", "desc_color": 1 },
6    "main_title": { "title": "线上告警需要你确认", "desc": "P2 · 影响 2 台主机" },
7    "sub_title_text": "10.0.2.13 磁盘 91%，建议清理 /var/log",
8    "card_action": { "type": 1, "url": "https://grafana.internal/d/xxx" }
9  }
10}

1{
2  "msgtype": "file",
3  "file": {
4    "media_id": "3a8asdfasdf..."   // 需先调上传接口拿 media_id，有效期 3 天
5  }
6}

1import hashlib
2from flask import Flask, request
3
4app = Flask(__name__)
5TOKEN = "DeepSeekBot2026"
6
7@app.get("/wechat")
8def verify():
9    signature = request.args.get("signature", "")
10    timestamp = request.args.get("timestamp", "")
11    nonce     = request.args.get("nonce", "")
12    echostr   = request.args.get("echostr", "")
13
14    # 字典序排序后拼接 → sha1
15    tmp = "".join(sorted([TOKEN, timestamp, nonce]))
16    if hashlib.sha1(tmp.encode()).hexdigest() == signature:
17        return echostr          # 原样返回，不能加任何字符
18    return "fail", 403

1pip install wechatpy   # 封装了 WXBizMsgCrypt，省得自己抄官方 demo

1from wechatpy.crypto import WeChatCrypto
2from wechatpy.exceptions import InvalidSignatureException
3
4crypto = WeChatCrypto(TOKEN, ENCODING_AES_KEY, APP_ID)
5
6try:
7    msg = crypto.decrypt_message(encrypted_body, msg_signature, timestamp, nonce)
8    # msg 是解密后的明文 XML
9except InvalidSignatureException:
10    return "fail", 403

1用户发消息
2   ↓
3微信 POST 到你的 /wechat
4   ↓
5① 立刻把消息丢进队列（Redis / Celery / 线程池）
6   ↓
7② 3 毫秒内返回空字符串 "" 或 "success"   ← 占住连接，微信满意
8   ↓
9③ 后台 worker 调 DeepSeek（慢慢来，8 秒也不怕）
10   ↓
11④ 用「客服消息接口」主动推送结果给用户（48 小时内可推）

1import threading, requests
2from flask import Flask, request
3from wechatpy import parse_message
4from wechatpy.replies import TextReply
5
6app = Flask(__name__)
7
8@app.post("/wechat")
9def handle():
10    xml = request.data
11    msg = parse_message(xml)
12    if msg.type != "text":
13        return "success"                      # 非文本消息直接忽略
14
15    openid   = msg.source                     # 用户 openid，推客服消息要用
16    question = msg.content
17
18    # ① 异步线程慢慢跑，不阻塞本次响应
19    threading.Thread(
20        target=async_answer, args=(openid, question), daemon=True
21    ).start()
22
23    # ② 立刻返回，绝不让微信等
24    return "success"
25
26def async_answer(openid: str, question: str):
27    answer = call_deepseek(question)          # 慢调用，随便几秒
28    push_customer_service_msg(openid, answer) # 客服消息接口异步补发
29
30def push_customer_service_msg(openid: str, content: str):
31    token = get_cached_access_token()         # ← 必须缓存！见下节
32    requests.post(
33        f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={token}",
34        json={"touser": openid, "msgtype": "text", "text": {"content": content[:2048]}},
35        timeout=10,
36    )

1import threading, time, requests, redis
2
3APPID, SECRET = "wx1234567890", "你的AppSecret"
4r = redis.Redis(host="localhost", port=6379, db=0)
5_lock = threading.Lock()
6
7def get_cached_access_token() -> str:
8    """提前 300 秒刷新，用分布式锁防止并发击穿。"""
9    token = r.get("wx:access_token")
10    ttl   = r.ttl("wx:access_token")
11    if token and ttl > 300:
12        return token.decode()
13
14    with _lock:                       # 双重检查，拿到锁再看一次
15        token = r.get("wx:access_token")
16        if token and r.ttl("wx:access_token") > 300:
17            return token.decode()
18
19        resp = requests.get(
20            "https://api.weixin.qq.com/cgi-bin/token",
21            params={"grant_type": "client_credential", "appid": APPID, "secret": SECRET},
22            timeout=10,
23        ).json()
24        if "access_token" not in resp:
25            raise RuntimeError(f"获取 token 失败: {resp}")
26        expires = int(resp.get("expires_in", 7200))
27        r.setex("wx:access_token", expires, resp["access_token"])
28        return resp["access_token"]

1from wechatpy.replies import TextReply
2
3reply = TextReply(content=answer, message=msg)
4xml = reply.render()          # 生成微信要求的 XML
5return xml

1<xml>
2  <ToUserName><![CDATA[oAbc123...]]></ToUserName>
3  <FromUserName><![CDATA[gh_abc123...]]></FromUserName>
4  <CreateTime>1746576000</CreateTime>
5  <MsgType><![CDATA[text]]></MsgType>
6  <Content><![CDATA[这是 DeepSeek 的回答]]></Content>
7</xml>

1你是 SRE 值班助手。下面是过去 10 分钟的告警原始记录：
2
3【粘贴 20-50 行告警】
4
5请输出：
61. 用一句话说清楚「发生了什么、影响面多大」
72. 按可能根因归类（最多 3 类），每类给出涉及的主机列表
83. 给出按优先级排序的处理建议（P0 优先）
94. 如果判断为误报或可自愈，明确说明
10
11输出到 300 字以内，用 markdown，主机 IP 用反引号包住。

1表结构：
2- orders(id, user_id, amount, status, created_at)
3- users(id, name, city, vip_level)
4
5请写一条 MySQL 8 查询：统计 2026 年 4 月每个城市的 VIP 用户人均订单金额，
6只保留订单数 >= 5 的城市，按人均金额降序，取前 10。
7
8要求：
9- 只输出 SQL，不要解释（群里要看结果不是看作文）
10- 字段名用中文别名
11- 注意 created_at 有索引，写法要能走索引

1下面是本项目今天的 Git 提交记录、Jira 流转记录和测试通过率：
2
3【粘贴原始数据】
4
5请生成一份给团队看的日报，结构如下：
6## 今日进展（3-5 条，每条一行，动词开头）
7## 风险与阻塞（没有就写「无」）
8## 明日计划（基于未关闭 issue 推断）
9## 需要谁介入（@ 具体负责人）
10
11语气：简洁、不说套话、不写「大家辛苦了」。总长度 250 字以内。

1阶段 1（入门）：企业微信群机器人 + 加签，跑通单向播报        ← 你现在在这里
2阶段 2（进阶）：公众号服务器配置 + 异步客服消息，做双向问答
3阶段 3（高级）：Function Calling 接内部系统，让机器人能查能改
4阶段 4（专家）：多模型路由 + Prompt Caching + 成本看板，做成团队基础设施