一、整体链路

给微信机器人接上大模型，架构上只比普通自动回复多一个环节：收到消息后不返回固定话术，而是先调一次大模型接口，拿到生成的回答再发回去。

用户发消息
    │
    ▼
平台 Webhook 回调 ──► 你的服务器（入队，立刻返回 {"ret": 200}）
                            │
                            ▼
                     消费者取出消息
                            │
                     调用大模型 API 生成回答
                            │
                     调用 postText 把回答发回去

三个模块各司其职：回调只管收、队列只管排队、消费者管“AI 生成 + 发送”。AI 接口调用耗时长（1~10 秒），所以绝对不能在回调里同步调 AI，必须异步。

二、准备工作

需要两样东西：

1. 微信机器人侧：Token、appId（通过 getLoginQrCode 扫码登录获得）
2. 大模型侧：任意一家 OpenAI 兼容接口的 API Key（DeepSeek、通义、Kimi 等都提供兼容接口）

三、回调接收：只入队，不处理

from flask import Flask, request, jsonify
import threading, queue, time, random, requests

app = Flask(__name__)

# 消息队列
msg_queue = queue.Queue()

@app.route("/callback", methods=["POST"])
def callback():
    try:
        msg = request.get_json()
    except Exception:
        return jsonify({"ret": 200})

    # 只处理文本消息
    if msg and msg.get("msgType") == 1:
        msg_queue.put(msg)

    # 必须 5 秒内返回这个格式，否则平台重推
    return jsonify({"ret": 200})

要点回顾：chatRoomId 有值是群消息、为空是私聊；返回体必须是 {"ret": 200}。

四、调用大模型生成回答

以 OpenAI 兼容接口为例，封装一个生成函数：

LLM_URL = "https://api.deepseek.com/v1/chat/completions"  # 换成你用的大模型地址
LLM_KEY = "sk-你的大模型Key"

def ask_llm(user_content: str) -> str:
    try:
        resp = requests.post(
            LLM_URL,
            headers={"Authorization": f"Bearer {LLM_KEY}"},
            json={
                "model": "deepseek-chat",   # 按你实际使用的模型填写
                "messages": [
                    {"role": "system", "content": "你是一个友好的助手，回答简洁，不超过200字。"},
                    {"role": "user", "content": user_content}
                ],
                "max_tokens": 500,
                "temperature": 0.7
            },
            timeout=30
        )
        data = resp.json()
        return data["choices"][0]["message"]["content"].strip()
    except Exception as e:
        print(f"LLM调用失败: {e}")
        return None

失败时返回 None，消费者里降级为固定话术，不能让消息黑洞。

五、消费者：生成 + 发送 + 限频

APP_ID = "你的appId"
BASE_URL = "https://wx.chuapi.com"
TOKEN = "你的X-finder-TOKEN"

def send_text(to_wxid: str, content: str) -> bool:
    resp = requests.post(
        f"{BASE_URL}/finder/v2/api/message/postText",
        headers={
            "Content-Type": "application/json",
            "X-finder-TOKEN": TOKEN
        },
        json={
            "appId": APP_ID,
            "toWxid": to_wxid,
            "content": content
        },
        timeout=10
    )
    return resp.json().get("ret") == 200

def consumer():
    while True:
        msg = msg_queue.get()
        try:
            # 群消息回复到群，私聊回复给发送人
            to_wxid = msg.get("chatRoomId") or msg["fromUser"]

            # 调大模型生成回答
            answer = ask_llm(msg["content"])
            if answer is None:
                answer = "系统繁忙，请稍后再试～"

            send_text(to_wxid, answer)
        finally:
            # 随机间隔 3~8 秒：AI 响应本身有耗时，间隔可以比纯自动回复短
            time.sleep(random.uniform(3, 8))

threading.Thread(target=consumer, daemon=True).start()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

六、加多轮对话记忆

上面的版本每条消息都是独立的一问一答。真实场景里用户会连续追问，需要给每个会话维护上下文。用字典按会话 ID 存最近几轮：

from collections import defaultdict, deque

# 每个会话保留最近 6 条消息（3轮对话）
session_history = defaultdict(lambda: deque(maxlen=6))

def get_session_id(msg) -> str:
    # 群聊按群+发送人区分会话，私聊按发送人
    if msg.get("chatRoomId"):
        return f"{msg['chatRoomId']}_{msg['fromUser']}"
    return msg["fromUser"]

def ask_llm_with_history(msg) -> str:
    session_id = get_session_id(msg)
    history = session_history[session_id]

    # 组装带上下文的 messages
    messages = [
        {"role": "system", "content": "你是一个友好的助手，回答简洁，不超过200字。"}
    ]
    messages.extend(history)
    messages.append({"role": "user", "content": msg["content"]})

    resp = requests.post(
        LLM_URL,
        headers={"Authorization": f"Bearer {LLM_KEY}"},
        json={
            "model": "deepseek-chat",
            "messages": messages,
            "max_tokens": 500
        },
        timeout=30
    )
    answer = resp.json()["choices"][0]["message"]["content"].strip()

    # 本轮对话写回历史
    history.append({"role": "user", "content": msg["content"]})
    history.append({"role": "assistant", "content": answer})

    return answer

生产环境把 session_history 换成 Redis（key 设过期时间，比如 30 分钟），避免重启丢上下文、内存无限涨。

七、群聊场景：只响应被@的消息

群里如果每条消息都调 AI，一是费钱，二是刷屏容易被投诉。标准做法是只响应@机器人的消息：

def should_reply_in_group(msg) -> bool:
    content = msg["content"]
    # @消息的内容里通常带 @昵称 前缀，按你账号昵称匹配
    if "机器人昵称" in content:
        return True
    return False

# 消费者里
if msg.get("chatRoomId"):
    if not should_reply_in_group(msg):
        continue  # 群消息且没@我，跳过

私聊则可以全量响应（或加黑名单过滤）。

八、几个必须控制的点

风险点	后果	控制方式
AI 响应超时	回复延迟几十秒	timeout 设 30 秒，超时降级话术
消息积压	排队回复越来越慢	消费者可开 2~3 个线程，但发送仍串行
上下文过长	token 费用暴涨	每会话只留最近 3 轮
高频群刷屏	触发风控/被踢	群聊只响应@消息 + 随机间隔
敏感内容	AI 输出违规内容	system prompt 里约束 + 输出关键词过滤

九、完整数据流回顾

用户: "帮我写个请假条"
    ↓ Webhook (msgType=1, fromUser=wxid_xxx, content="帮我写个请假条")
回调服务: 校验 → 去重 → 入队 → 返回 {"ret": 200}
    ↓ queue
消费者: 取消息 → 组装带历史的 messages → 调 LLM
    ↓ LLM 返回 "好的，以下是一份请假条模板……"
消费者: postText(appId, toWxid, answer) → ret==200 发送成功
    ↓ sleep 3~8秒
下一条消息……

十、小结

接入 AI 大模型没有想象中复杂：回调收消息 → 队列解耦 → 消费者调 LLM → postText 发回答，四步就是一个能用的微信 AI 机器人。多轮记忆、群聊@过滤、超时降级这三件事加上，体验就和商业产品差不多了。机器人侧的接口细节以官方文档为准；大模型侧选任何 OpenAI 兼容接口都可以，代码几乎不用改。如果不想自己维护协议层，选一个 HTTP + Webhook 形式的机器人 API（比如 WTAPI 这类方案）套上本文模板即可。

参考资料

• 接口定义与参数说明文档：weiti.apifox.cn