传统微信客服机器人靠关键词匹配回复，问题很明显：用户换个说法就答不上来，话术库越堆越臃肿，维护成本比请客服还高。大模型普及后，这件事有了更简单的解法——把Webhook 收到的用户消息直接交给大模型，生成回复后再通过微信接口发回去。

本文基于 WTAPI 实现微信侧收发，大模型部分使用你自己的模型服务（OpenAI 兼容接口），两侧分工，代码可直接跑。

一、工作链路

官方文档给出的 AI 客服标准链路：

用户发消息 → Webhook 推送 AddMsg
           → 幂等去重 + 组装上下文
           → 调用大模型生成回复
           → postText 回发微信

二、准备工作

1. WTAPI：控制台「Token回调」页获取 X-finder-TOKEN，「微信实例」扫码登录拿到 appId，配置 Webhook 回调地址；
2. 大模型：自备模型服务的 API Key 和接口地址（任何 OpenAI 兼容格式的服务均可）。

API 基础地址：`WTAPI框架

三、完整代码（Python Flask）

import json
import threading
from collections import defaultdict, deque

import requests
from flask import Flask, request
from openai import OpenAI

# ========== WTAPI 官方配置 ==========
API_BASE_URL = "https://wx.chuapi.com"
FINDER_TOKEN = "YOUR_TOKEN"
JWT = "YOUR_JWT"                 # 部分环境需要，以官方文档为准
APP_ID = "YOUR_APPID"

HEADERS = {
    "X-finder-TOKEN": FINDER_TOKEN,
    "Authorization": f"Bearer {JWT}",
    "Content-Type": "application/json",
}

# ========== 大模型配置（自备，替换为你的服务商） ==========
llm = OpenAI(
    api_key="YOUR_LLM_KEY",
    base_url="https://你的大模型服务商/v1"
)
SYSTEM_PROMPT = (
    "你是品牌官方客服，回答友好、简洁。"
    "不确定的问题不要编造，引导用户回复【人工】。"
)

# 多轮对话上下文（生产环境建议替换为 Redis，按 wxid 隔离并设过期时间）
history = defaultdict(lambda: deque(maxlen=10))

# ========== 微信发消息：官方 postText ==========
def post_text(to_wxid, content):
    requests.post(
        f"{API_BASE_URL}/finder/v2/api/message/postText",
        headers=HEADERS,
        data=json.dumps({
            "appId": APP_ID,
            "toWxid": to_wxid,
            "content": content
        }),
        timeout=10
    )

# ========== 调用大模型 ==========
def ask_llM(user_wxid, text):
    history[user_wxid].append({"role": "user", "content": text})
    try:
        resp = llm.chat.completions.create(
            model="YOUR_MODEL",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                *history[user_wxid],
            ],
            timeout=20
        )
        reply = resp.choices[0].message.content
        history[user_wxid].append({"role": "assistant", "content": reply})
        return reply
    except Exception:
        # 降级：模型超时/报错时回退固定话术，避免用户无人应答
        return "客服暂时繁忙，请稍后再试，或回复【人工】联系客服～"

# ========== Webhook 回调 ==========
app = Flask(__name__)
_seen = set()   # 官方去重键：Appid + Data.NewMsgId

@app.route("/wtapi/callback", methods=["POST"])
def callback():
    msg = request.get_json(silent=True) or {}
    # 官方规则：3秒内返回响应，大模型耗时，必须异步
    if msg.get("TypeName") == "AddMsg":
        threading.Thread(target=handle, args=(msg,)).start()
    return "", 200

def handle(msg):
    d = msg.get("Data", {})

    key = f"{msg.get('Appid')}:{d.get('NewMsgId')}"
    if key in _seen:
        return
    _seen.add(key)

    if d.get("MsgType") != 1:    # 只处理文本消息
        return

    from_user = d.get("FromUserName", {}).get("string", "")
    content = d.get("Content", {}).get("string", "").strip()
    if not from_user or not content:
        return

    # FromUserName 以 @chatroom 结尾为群消息，按需过滤
    # 转人工
    if content in ("人工", "转人工", "真人"):
        post_text(from_user, "正在为你转接人工客服，请稍候～")
        # 可在此处向客服群推送通知
        return

    reply = ask_llM(from_user, content)
    post_text(from_user, reply)

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

四、五个工程要点

1. 必须异步处理。 官方要求回调 3 秒内响应，大模型生成回复往往要 5-15 秒，同步处理必然超时重推。示例中收到消息立即返回 200，模型调用在线程中完成。

2. 幂等去重不能省。 按官方规则 Appid + Data.NewMsgId 去重，否则一次重推，大模型会把同一句话回答两遍。

3. 多轮上下文按 wxid 隔离。 用 deque(maxlen=10) 只保留最近 5 轮，既支持上下文理解，又避免 token 无限增长。生产环境建议用 Redis 并设置过期时间，用户长时间不说话自动清空。

4. 必须有降级方案。 模型超时、限流、报错时回退固定话术（示例已包含），绝不能让用户发了消息石沉大海。

5. 转人工要留出口。 在系统提示词里约束模型"不确定就引导转人工"，同时代码层识别"人工"关键词——AI 解决 80% 重复问题，人工聚焦复杂问题，成本和体验才能平衡。

五、两个可扩展方向

• 知识库问答：把产品文档、常见问题做成向量检索，用户提问时先检索相关内容，连同问题一起交给大模型（RAG），回答准确度会显著提高，还能避免模型编造；
• 群客服：群消息 FromUserName 以 @chatroom 结尾，可设置"仅被 @ 机器人才响应"，避免机器人在群里对每条消息都回复造成骚扰。