智谱 GLM-5.2 接入微信完整指南：企业微信群机器人 + 公众号双路径实测

1✅ 需要联网查最新信息（行业动态、竞品、政策法规）
2✅ 长文档摘要（合同、标书、技术白皮书）
3✅ 工具型机器人（查系统、建工单、改配置）
4✅ 代码与日志分析（GLM 系列代码能力一直在线）
5✅ 国内直连且需要合规数据留在境内的团队
6❌ 极致成本敏感的高频闲聊（输入 ¥1.0 不是最低档）
7❌ 超长代码库整体检索（200K 不如 V4.1 Flash 的 1M）
8❌ 需要语音输入输出的场景（GLM-5.2 不做音频）

1export ZHIPU_KEY=xxxx.yyyy
2
3curl https://open.bigmodel.cn/api/paas/v4/chat/completions \
4  -H "Authorization: Bearer $ZHIPU_KEY" \
5  -H "Content-Type: application/json" \
6  -d '{
7        "model": "glm-5.2",
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":"智谱 GLM-5.2 已就位"}}'

1import time, hmac, hashlib, base64, urllib.parse
2import requests
3
4WEBHOOK_KEY = "693a91f6-7xxx-4bc4-97a0-0ec2sifa5aaa"
5SECRET      = "SEC-xxxxxxxxxxxx"
6
7def gen_sign(secret: str, timestamp: str) -> str:
8    """秒级 timestamp + 单个 LF 换行 + secret → HMAC-SHA256 → base64 → urlencode。"""
9    raw = f"{timestamp}\n{secret}"
10    digest = hmac.new(secret.encode(), raw.encode(), hashlib.sha256).digest()
11    return urllib.parse.quote_plus(base64.b64encode(digest).decode())
12
13def push(payload: dict):
14    ts = str(int(time.time()))
15    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
16           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(SECRET, ts)}")
17    r = requests.post(url, json=payload, timeout=10).json()
18    if r.get("errcode") != 0:
19        raise RuntimeError(f"推送失败: {r}")
20    return r

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(crypto.createHmac('sha256', secret).update(raw).digest('base64'));
10}
11
12function push(payload) {
13  const ts = Math.floor(Date.now() / 1000).toString();
14  const url = `https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=${WEBHOOK_KEY}&timestamp=${ts}&sign=${genSign(SECRET, ts)}`;
15  const body = JSON.stringify(payload);
16  return new Promise((resolve, reject) => {
17    const req = https.request(url, {
18      method: 'POST',
19      headers: { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(body) }
20    }, (res) => {
21      let d = ''; res.on('data', (c) => (d += c));
22      res.on('end', () => resolve(JSON.parse(d)));
23    });
24    req.on('error', reject); req.write(body); req.end();
25  });
26}

1# app.py —— pip install flask requests
2import os, time, json, hmac, hashlib, base64, urllib.parse
3import requests
4from flask import Flask, request, jsonify
5
6app = Flask(__name__)
7
8ZHIPU_KEY = os.environ["ZHIPU_API_KEY"]          # 形如 xxxx.yyyy，整串填
9ZHIPU_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
10MODEL_ID  = "glm-5.2"                             # 型号
11
12WEBHOOK_KEY    = os.environ["WECOM_WEBHOOK_KEY"]
13WEBHOOK_SECRET = os.environ["WECOM_WEBHOOK_SECRET"]
14
15SYSTEM_PROMPT = """你是企业微信工作群的 AI 助手，基于智谱 GLM-5.2。
16- 回答控制在 350 字以内
17- 遇到「最近」「最新」「新规」「多少钱」这类需要时效性的问题，必须调用联网搜索
18- 引用信息时标注来源，不要编造文件名、链接或数据
19- 输出用 markdown，标题 ##，重点 ** 加粗 **
20"""
21
22# 内置联网搜索工具：模型自己判断要不要调用
23WEB_SEARCH_TOOL = {
24    "type": "web_search",
25    "web_search": {
26        "enable": True,
27        "search_result": True,      # 把检索原文也返回，便于溯源
28    },
29}
30
31def ask_glm(question: str, enable_search: bool = False) -> str:
32    body = {
33        "model": MODEL_ID,
34        "messages": [
35            {"role": "system", "content": SYSTEM_PROMPT},
36            {"role": "user", "content": question},
37        ],
38        "temperature": 0.3,
39        "max_tokens": 1800,
40        "timeout": 90 if enable_search else 30,
41    }
42    if enable_search:
43        body["tools"] = [WEB_SEARCH_TOOL]          # 联网会显著变慢，超时放宽
44
45    resp = requests.post(
46        ZHIPU_URL,
47        headers={"Authorization": f"Bearer {ZHIPU_KEY}"},
48        json=body,
49        timeout=120,
50    )
51    resp.raise_for_status()
52    data = resp.json()
53    msg = data["choices"][0]["message"]
54    # 有工具调用时，正文可能在 content 里，也可能在 tool_calls 之后
55    return msg.get("content") or ""
56
57def need_search(q: str) -> bool:
58    """简单的关键词预判：命中就开联网，避免每次都白等 7 秒。"""
59    triggers = ["最近", "最新", "新规", "今天", "昨天", "本周", "政策", "多少钱", "多少钱", "什么时候发布"]
60    return any(t in q for t in triggers)
61
62def gen_sign(secret, ts):
63    raw = f"{ts}\n{secret}".encode()
64    return urllib.parse.quote_plus(
65        base64.b64encode(hmac.new(secret.encode(), raw, hashlib.sha256).digest()).decode()
66    )
67
68def push_markdown(md: str):
69    ts = str(int(time.time()))
70    url = (f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send"
71           f"?key={WEBHOOK_KEY}&timestamp={ts}&sign={gen_sign(WEBHOOK_SECRET, ts)}")
72    return requests.post(url, json={"msgtype": "markdown", "markdown": {"content": md[:4096]}}, timeout=10).json()
73
74@app.post("/ask")
75def ask():
76    q = ((request.json or {}).get("q") or "").strip()
77    if not q:
78        return jsonify(err="empty question"), 400
79    use_search = need_search(q)
80    answer = ask_glm(q, enable_search=use_search)
81    tag = "（已联网）" if use_search else ""
82    push_markdown(f"**Q**: {q}{tag}\n\n{answer}")
83    return jsonify(ok=True, answer=answer, searched=use_search)
84
85if __name__ == "__main__":
86    app.run(host="0.0.0.0", port=8000)

1# 第一步：定义你的工具
2MY_TOOLS = [
3    {
4        "type": "function",
5        "function": {
6            "name": "query_order",
7            "description": "根据订单号查询订单状态与金额",
8            "parameters": {
9                "type": "object",
10                "properties": {
11                    "order_id": {"type": "string", "description": "订单号，形如 ORD-2026-001"}
12                },
13                "required": ["order_id"]
14            },
15        },
16    }
17]
18
19def query_order(order_id: str) -> dict:
20    """你的真实业务逻辑，这里用假数据演示。"""
21    return {"order_id": order_id, "status": "已发货", "amount": 299.0, "eta": "2026-05-15"}
22
23def ask_with_tools(question: str, max_rounds: int = 3) -> str:
24    """最多 3 轮，防止模型陷入死循环反复调工具。"""
25    messages = [
26        {"role": "system", "content": SYSTEM_PROMPT},
27        {"role": "user", "content": question},
28    ]
29    for _ in range(max_rounds):
30        resp = requests.post(
31            ZHIPU_URL,
32            headers={"Authorization": f"Bearer {ZHIPU_KEY}"},
33            json={"model": MODEL_ID, "messages": messages, "tools": MY_TOOLS, "temperature": 0.2},
34            timeout=60,
35        ).json()
36        msg = resp["choices"][0]["message"]
37        calls = msg.get("tool_calls")
38
39        if not calls:                     # 没有工具调用 → 直接是最终答案
40            return msg.get("content", "")
41
42        messages.append(msg)              # 把「我要调工具」这条 assistant 消息回传
43        for call in calls:
44            args = json.loads(call["function"]["arguments"])
45            result = query_order(**args)  # 执行真实业务
46            messages.append({             # 把工具结果以 tool 角色回传
47                "role": "tool",
48                "tool_call_id": call["id"],
49                "content": json.dumps(result, ensure_ascii=False),
50            })
51    return "工具调用轮次超限，请简化你的问题。"

1{
2  "msgtype": "markdown",
3  "markdown": {
4    "content": "## 行业动态速览\n> 来源：模型联网检索，共 3 条\n\n**1. 数据合规新要求**\n主管部门发布新规，要求...\n\n**2. ...**\n\n<font color=\"comment\">检索时间 2026-05-12 10:30</font>"
5  }
6}

1import hashlib
2from flask import Flask, request
3
4app = Flask(__name__)
5TOKEN = "GLMBot2026"
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      # 封装了 WXBizMsgCrypt，别自己抄官方 demo

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   → ③ 后台调 GLM-5.2（含联网与工具调用，慢慢来）
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_glm(question, enable_search=need_search(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", "appid": APPID, "secret": APPSECRET},
17            timeout=10,
18        ).json()
19        if "access_token" not in resp:
20            raise RuntimeError(f"微信 token 获取失败: {resp}")
21        r.setex("wx:token", int(resp.get("expires_in", 7200)), resp["access_token"])
22        return resp["access_token"]

1请联网检索过去 24 小时国内「人工智能行业监管」相关的动态，然后输出：
2
3## 今日要点（3 条，每条 60 字以内）
4## 影响解读（对我们业务有什么影响，2 条）
5## 需要关注（需要谁跟进，1 条）
6
7约束：
8- 只输出检索到的真实信息，不要推测或补充常识性内容
9- 每条要点末尾用括号标注来源机构
10- 如果检索不到足够内容，直接说「今日无重大动态」
11- 总长度 300 字以内

1下面是一份技术外包合同全文（约 4 万字）。请输出：
2
3## 合同主体与标的（甲方、乙方、服务内容、合同金额）
4## 关键时间节点（交付日期、验收期、付款节点）
5## 风险条款（对我方不利的条款，逐条列出并说明风险）
6## 待确认项（合同中未明确或表述模糊的地方）
7
8约束：
9- 金额、日期必须原文摘录，不要改写或换算
10- 风险条款必须引用原文原句，并说明为什么有风险
11- 如果某项在合同中找不到，写「合同未约定」

1下面是生产环境的一段报错日志（Java 服务）：
2
3【粘贴 40-80 行日志】
4
5请输出：
61. **根因**：一句话说清最可能的根本原因
72. **证据链**：引用日志中的具体行，说明你是怎么推出来的
83. **次要可能**：1-2 条备选原因
94. **排查下一步**：最该先做的 2 个动作
10
11不要复述日志原文，直接给结论。300 字以内。

1阶段 1：企业微信群机器人 + 加签，跑通单向播报        ← 你现在在这里
2阶段 2：公众号服务器配置 + 异步客服消息，实现双向问答
3阶段 3：接入 web_search 联网工具，做时效性问答
4阶段 4：Function Calling 接内部系统，做成工具型机器人