文心一言 ERNIE 4.5 接入微信完整指南：企业微信群机器人 + 公众号双路径实测

1✅ 中文公文、通知、汇报材料撰写与润色
2✅ 群里收到的截图、票据、表格图片解读（多模态）
3✅ 会议纪要整理与待办抽取
4✅ 政企、国企等需要完整合规资质的场景
5✅ 已有百度智能云账号、能复用账单与发票的团队
6❌ 超长上下文代码库检索（128K 不如 V4.1 Flash 的 1M）
7❌ 极致成本敏感的高频闲聊（同价位有更便宜的）
8❌ 需要严格 JSON Schema 结构化输出的关键链路（建议加校验层）

1curl 'https://aip.baidubce.com/oauth/2.0/token' \
2  -d 'grant_type=client_credentials' \
3  -d 'client_id=你的APIKey' \
4  -d 'client_secret=你的SecretKey'

1{
2  "access_token": "24.be1f0e7f...",
3  "expires_in": 2592000,
4  "refresh_token": "25.xxx"
5}

1export QIANFAN_TOKEN=24.be1f0e7f...
2
3curl https://qianfan.baidubce.com/v2/chat/completions \
4  -H "Content-Type: application/json" \
5  -H "Authorization: Bearer $QIANFAN_TOKEN" \
6  -d '{
7        "model": "ernie-4.5",
8        "messages": [{"role":"user","content":"用一句话介绍你自己"}],
9        "max_tokens": 200
10      }'

1curl 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的key' \
2   -H 'Content-Type: application/json' \
3   -d '{"msgtype":"text","text":{"content":"文心一言 ERNIE 4.5 已就位"}}'

1import time, hmac, hashlib, base64, urllib.parse
2import requests
3
4WEBHOOK_KEY = "693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa"
5SECRET      = "SEC-xxxxxxxxxxxx"      # 机器人 secret，不是 webhook key
6
7def gen_sign(secret: str, timestamp: str) -> str:
8    """timestamp 必须是秒级字符串；中间是单个 LF 换行；最后必须 urlencode。"""
9    string_to_sign = f"{timestamp}\n{secret}"
10    digest = hmac.new(secret.encode(), string_to_sign.encode(), hashlib.sha256).digest()
11    return urllib.parse.quote_plus(base64.b64encode(digest).decode())
12
13def push(content: str, msgtype: str = "text"):
14    ts   = str(int(time.time()))      # 秒级！不是毫秒
15    sign = gen_sign(SECRET, ts)
16    url  = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
17            f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={sign}")
18    r = requests.post(url, json={"msgtype": msgtype, "text": {"content": content}}, timeout=10)
19    result = r.json()
20    if result.get("errcode") != 0:
21        raise RuntimeError(f"推送失败: {result}")
22    return result

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

1# app.py —— pip install flask requests redis
2import os, time, hmac, hashlib, base64, urllib.parse, threading
3import requests
4from flask import Flask, request, jsonify
5
6app = Flask(__name__)
7
8QIANFAN_AK = os.environ["QIANFAN_AK"]        # 千帆 API Key
9QIANFAN_SK = os.environ["QIANFAN_SK"]        # 千帆 Secret Key
10QIANFAN_URL = "https://qianfan.baidubce.com/v2/chat/completions"
11MODEL_ID = "ernie-4.5"                        # 型号；要省钱改成 "ernie-4.5-turbo"
12
13WEBHOOK_KEY = os.environ["WECOM_WEBHOOK_KEY"]
14WEBHOOK_SECRET = os.environ["WECOM_WEBHOOK_SECRET"]
15
16_token_cache = {"token": None, "expire_at": 0}
17_tlock = threading.Lock()
18
19def qianfan_token() -> str:
20    """千帆 access_token 有效期 30 天，提前 1 天刷新，加锁防并发击穿。"""
21    if _token_cache["token"] and time.time() < _token_cache["expire_at"] - 86400:
22        return _token_cache["token"]
23    with _tlock:
24        if _token_cache["token"] and time.time() < _token_cache["expire_at"] - 86400:
25            return _token_cache["token"]
26        r = requests.post(
27            "https://aip.baidubce.com/oauth/2.0/token",
28            data={"grant_type": "client_credentials",
29                  "client_id": QIANFAN_AK, "client_secret": QIANFAN_SK},
30            timeout=10,
31        ).json()
32        if "access_token" not in r:
33            raise RuntimeError(f"换取千帆 token 失败: {r}")
34        _token_cache["token"] = r["access_token"]
35        _token_cache["expire_at"] = time.time() + int(r.get("expires_in", 2592000))
36        return r["access_token"]
37
38SYSTEM_PROMPT = """你是企业微信工作群里的 AI 助手，基于百度文心一言 ERNIE 4.5。
39- 回答控制在 300 字以内
40- 写公文用正式书面语，不写口语化废话
41- 不确定就说不确定，不要编造数据、文件名或人名
42- 输出用 markdown，标题 ##，重点 ** 加粗 **
43"""
44
45def ask_ernie(question: str, image_url=None) -> str:
46    """支持图文混合输入——这是 ERNIE 4.5 相比纯文本模型的优势。"""
47    if image_url:
48        content = [
49            {"type": "text", "text": question},
50            {"type": "image_url", "image_url": {"url": image_url}},
51        ]
52    else:
53        content = question
54    resp = requests.post(
55        QIANFAN_URL,
56        headers={"Authorization": f"Bearer {qianfan_token()}"},
57        json={
58            "model": MODEL_ID,
59            "messages": [
60                {"role": "system", "content": SYSTEM_PROMPT},
61                {"role": "user", "content": content},
62            ],
63            "temperature": 0.3,
64            "max_tokens": 1500,
65        },
66        timeout=60,
67    )
68    resp.raise_for_status()
69    return resp.json()["choices"][0]["message"]["content"]
70
71def gen_sign(secret, ts):
72    s = f"{ts}\n{secret}".encode()
73    return urllib.parse.quote_plus(
74        base64.b64encode(hmac.new(secret.encode(), s, hashlib.sha256).digest()).decode()
75    )
76
77def push_to_wecom(md: str):
78    ts = str(int(time.time()))
79    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
80           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(WEBHOOK_SECRET, ts)}")
81    r = requests.post(url, json={"msgtype": "markdown", "markdown": {"content": md[:4096]}}, timeout=10)
82    if r.json().get("errcode") != 0:
83        app.logger.error("push failed: %s", r.json())
84
85@app.post("/ask")
86def ask():
87    q = (request.json or {}).get("q", "").strip()
88    img = (request.json or {}).get("image_url")
89    if not q:
90        return jsonify(err="empty question"), 400
91    answer = ask_ernie(q, img)
92    push_to_wecom(f"**Q**: {q}\n\n{answer}")
93    return jsonify(ok=True, answer=answer)
94
95if __name__ == "__main__":
96    app.run(host="0.0.0.0", port=8000)

1export QIANFAN_AK=xxx QIANFAN_SK=yyy
2export WECOM_WEBHOOK_KEY=693a91f6-xxxx WECOM_WEBHOOK_SECRET=SEC-xxxx
3python app.py
4
5curl -X POST http://127.0.0.1:8000/ask \
6  -H 'Content-Type: application/json' \
7  -d '{"q":"帮我写一则关于下周三季度总结会的部门通知"}'

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "content": "## 部门通知\n> 发布人：行政部\n\n**会议时间**：下周三 14:00\n**会议地点**：3 号楼大会议室\n**携带材料**：本季度工作总结（纸质 2 份）\n\n<font color=\"warning\">请提前 10 分钟到场</font>"
5  }
6}

1import hashlib
2from flask import Flask, request
3
4app = Flask(__name__)
5TOKEN = "ErnieBot2026"
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    tmp = "".join(sorted([TOKEN, timestamp, nonce]))   # 字典序！顺序错必失败
14    if hashlib.sha1(tmp.encode()).hexdigest() == signature:
15        return echostr                                  # 原样返回，不能加任何字符
16    return "fail", 403

1pip install wechatpy

1from wechatpy.crypto import WeChatCrypto
2from wechatpy.exceptions import InvalidSignatureException
3
4crypto = WeChatCrypto(TOKEN, ENCODING_AES_KEY, APP_ID)
5try:
6    plain_xml = crypto.decrypt_message(body, msg_signature, timestamp, nonce)
7except InvalidSignatureException:
8    return "fail", 403

1用户发消息 → 微信 POST /wechat
2   → ① 丢进队列（线程池 / Celery / Redis）
3   → ② 立刻返回 "success"（3ms 内）      ← 微信满意
4   → ③ 后台调 ERNIE 4.5（慢慢来）
5   → ④ 调客服消息接口主动推送

1import threading, requests
2from flask import Flask, request
3from wechatpy import parse_message
4
5app = Flask(__name__)
6
7@app.post("/wechat")
8def handle():
9    msg = parse_message(request.data)
10    if msg.type != "text":
11        return "success"
12    threading.Thread(target=async_answer, args=(msg.source, msg.content), daemon=True).start()
13    return "success"          # 立刻返回，绝不让微信等
14
15def async_answer(openid, question):
16    answer = ask_ernie(question)          # 慢调用
17    push_cs_msg(openid, answer)           # 客服消息补发
18
19def push_cs_msg(openid, content):
20    token = get_wx_token()                # 必须缓存，见下节
21    requests.post(
22        f"https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token={token}",
23        json={"touser": openid, "msgtype": "text", "text": {"content": content[:2048]}},
24        timeout=10,
25    )

1import threading, time, requests, redis
2r = redis.Redis(host="localhost", port=6379, db=0)
3_lock = threading.Lock()
4
5def get_wx_token() -> str:
6    """公众号 access_token：2 小时过期，重新获取会让上一个立刻失效。"""
7    tok = r.get("wx:token")
8    if tok and r.ttl("wx:token") > 300:
9        return tok.decode()
10    with _lock:                                  # 双重检查 + 锁
11        tok = r.get("wx:token")
12        if tok and r.ttl("wx:token") > 300:
13            return tok.decode()
14        resp = requests.get(
15            "https://api.weixin.qq.com/cgi-bin/token",
16            params={"grant_type": "client_credential",
17                    "appid": APPID, "secret": APPSECRET},
18            timeout=10,
19        ).json()
20        if "access_token" not in resp:
21            raise RuntimeError(f"微信 token 获取失败: {resp}")
22        r.setex("wx:token", int(resp.get("expires_in", 7200)), resp["access_token"])
23        return resp["access_token"]

1你是技术支持助手。下面是用户发来的一张软件报错截图。
2
3请按以下结构输出：
41. **这是什么软件**：从界面特征判断是哪个系统或应用
52. **报错信息**：原样摘录截图中的错误文字（不要改写）
63. **可能原因**：给出 2-3 条，按可能性排序
74. **下一步**：告诉用户第一步该做什么（一句话，可执行）
8
9如果截图中文字模糊无法辨认，明确说「截图分辨率不足，请发原图」，不要猜测。
10输出到 250 字以内。

1请把下面这段口语化口信，改写成一则正式的部门通知：
2
3「下周三下午两点大会议室开季度总结会，让大家带上纸质总结材料，别迟到。」
4
5要求：
6- 使用正式书面语，不出现「大家」「别」「搞定」等口语词
7- 结构：标题 → 通知对象 → 会议时间地点 → 参会要求 → 落款与日期
8- 日期用「2026 年 5 月 20 日（周三）14:00」格式
9- 落款写「行政部」，日期留占位符 [YYYY 年 MM 月 DD 日]
10- 全文 200 字以内

1下面是本次项目周会的转写文本（存在识别错误，请自行修正明显的同音错字）：
2
3【粘贴 2000-4000 字转写文本】
4
5请输出：
6## 会议结论（每条一行，动词开头）
7## 待办事项（表格形式：负责人 | 事项 | 截止时间 | 优先级）
8## 未决问题（需要下次会议继续讨论的）
9## 风险提示
10
11约束：
12- 待办事项的负责人必须从文本中出现的人名里提取，不要编造
13- 截止时间只写文本中明确提到的日期，没提就写「未明确」
14- 转写中的同音错字请修正，但不要改变原意

1阶段 1：企业微信群机器人 + 加签，跑通单向播报        ← 你现在在这里
2阶段 2：公众号服务器配置 + 异步客服消息，实现双向问答
3阶段 3：图文混合输入 + 千帆知识库，做企业文档问答
4阶段 4：多模型路由 + 成本看板，做成团队基础设施