如果你管着几个做客服的微信号，大概率见过这种场景：晚高峰咨询涌进来，人工回不过来，客户等不到回复转身就去问别家了。上了大模型之后问题也没解决——AI 回答很快，但消息在微信里进不来、回不出去，AI 根本没机会干活。
问题从来不在 AI，在通道：**微信个人号没有官方对外 API，消息怎么拿出来、回复怎么写回去，才是整件事的地基。**这篇把微信接 AI 客服这件事一次讲完：路线怎么选、架构怎么分层、代码怎么写，最后给一份限流、去重、兜底齐全的完整脚本。
一、四条路线，一张表先给结论
| 路线 | 原理 | 优点 | 硬伤 |
|---|---|---|---|
| 桌面自动化 | 模拟点击 PC 微信 | 零成本起步 | UI 一变就失效，挂机电脑跑不了量 |
| 逆向 Hook | 注入微信客户端 | 功能深 | 封号风险挂在头顶，版本更新即失效 |
| 开源协议端 | 自部署开源项目 | 数据全在本地 | 登录态、掉线、协议维护全自己扛 |
| 接口服务 | 平台侧维护协议，输出标准 HTTP API | 拿 appId 就能调，掉线有回调通知 | 依赖服务商稳定性 |
客服场景的判断很直接：客户消息一分钟都断不得，掉线恢复能力是第一优先级——这一项上，桌面自动化和逆向方案基本出局，自建开源要有专职运维才玩得动。多数团队的合理选择是接口服务：协议层的脏活由平台侧处理，你的精力全在业务和 AI 上。
二、架构：三层，各管各的
接 AI 客服的系统分三层，分层的意义是任何一层出问题不动另外两层：
- 通道层：微信消息的收发。我走的是 WTAPI——基于 iPad/Mac 协议的微信机器人接口服务，把消息、好友、群、朋友圈这些能力做成标准 HTTP API（下文简称“平台”）。收消息靠 Webhook 回调推送，发消息调
postText一个接口。 - 业务层：你写的机器人本体——收消息、去重、限流、组装上下文、调 AI、回消息。这篇的重点。
- AI 层：大模型或编排平台（Dify、Coze 都行），负责生成回答。换 AI 供应商只动这一层，通道代码一行不改。
三、完整代码
依赖 Flask 和 requests，业务层的四个关键件——去重、限流、异步、兜底——全在代码里：
pip install flask requests
# -*- coding: utf-8 -*-
# WTAPI 接口平台 | 文档: https://weiti.apifox.cn
# 本文依赖:
# - 发消息: postText (appId / toWxid / content, toWxid=filehelper 可自测)
# - 收消息: Webhook 回调 AddMsg (3秒内应答, 去重键=Appid+Data.NewMsgId)
# 下一步: 文档第一屏有「5分钟完成第一次调用」, 先发通一条再联调全链路
# 官网: www.chuapi.com
import time
import queue
import hashlib
import threading
import requests
from flask import Flask, request
WTAPI_BASE = "https://wx.chuapi.com"
WTAPI_TOKEN = "你的X-finder-TOKEN"
APP_ID = "你的appId"
HEADERS = {"X-finder-TOKEN": WTAPI_TOKEN, "Content-Type": "application/json"}
AI_URL = "你的大模型接口地址" # OpenAI 兼容接口均可
AI_KEY = "你的Key"
WORKERS = 8 # AI平均耗时(秒) x 峰值每秒消息数
TTL = 600 # 去重键保留 10 分钟
RATE_LIMIT = 20 # 每分钟回复上限, 自控红线
app = Flask(__name__)
inbox: queue.Queue = queue.Queue(maxsize=2000)
seen: dict = {}
minute_counter = {"minute": 0, "count": 0}
lock = threading.Lock()
def gc_seen() -> None:
"""去重表过期清理, 不做这步内存只涨不降。"""
now = time.time()
for k in [k for k, t in seen.items() if now - t > TTL]:
seen.pop(k, None)
def under_rate_limit() -> bool:
"""滑动计数限流: 超了宁可排队, 不超发。"""
with lock:
m = int(time.time() // 60)
if minute_counter["minute"] != m:
minute_counter["minute"], minute_counter["count"] = m, 0
if minute_counter["count"] >= RATE_LIMIT:
return False
minute_counter["count"] += 1
return True
def ai_answer(text: str, user_key: str) -> str:
"""大模型调用, OpenAI 兼容接口。超时和异常由调用方兜底。"""
resp = requests.post(AI_URL, json={
"model": "你的模型名",
"messages": [{"role": "user", "content": text}],
}, headers={"Authorization": f"Bearer {AI_KEY}"}, timeout=30)
return resp.json()["choices"][0]["message"]["content"]
def reply_text(to_wxid: str, content: str) -> None:
body = {"appId": APP_ID, "toWxid": to_wxid, "content": content}
requests.post(f"{WTAPI_BASE}/finder/v2/api/message/postText",
json=body, headers=HEADERS, timeout=10)
def worker() -> None:
while True:
from_wxid, text = inbox.get()
try:
answer = ai_answer(text, hashlib.md5(from_wxid.encode()).hexdigest())
except Exception:
answer = "您好，系统繁忙，人工客服会尽快回复您~" # 兜底: 不能哑
reply_text(from_wxid, answer)
inbox.task_done()
@app.route("/webhook", methods=["POST"])
def webhook():
data = request.get_json(silent=True) or {}
if data.get("TypeName") != "AddMsg":
return "ok", 200
msg = data.get("Data", {})
mid = msg.get("NewMsgId")
gc_seen()
if mid in seen: # 平台会重复推送, 去重键=Appid+NewMsgId
return "ok", 200
seen[mid] = time.time()
from_wxid = msg.get("FromUserName", {}).get("string", "")
content = msg.get("Content", {}).get("string", "")
if from_wxid.endswith("@chatroom") or not content:
return "ok", 200 # 群消息本期只记录, 客服先走私聊
if under_rate_limit():
inbox.put_nowait((from_wxid, content))
else:
print(f"[限流] 本分钟已达上限, 积压: {content[:20]}")
return "ok", 200 # 3 秒应答, 永远不被业务堵住
if __name__ == "__main__":
for _ in range(WORKERS):
threading.Thread(target=worker, daemon=True).start()
app.run(host="0.0.0.0", port=5000)
（平台侧接口均为核实过的可用写法；ai_answer 按你的大模型供应商替换请求体结构，参数以各家 API 文档为准。）
四、四个件，一个都不能少
去重：平台会重复推送同一条消息，去重键是 Appid + NewMsgId。没有去重，客户会收到重复回复——客服场景里这比不回复还伤。
限流：RATE_LIMIT 是自控红线，不是平台限制。AI 客服的回复频率本身就是风控项，每分钟上限写进配置，高峰宁可让客户稍等，不超发。
异步：回调要求 3 秒内应答，AI 再慢也不能堵在应答前面——队列加 worker 池是最低配解法，worker 数按“AI 平均耗时 × 峰值每秒消息数”算。
兜底：AI 接口超时、限流、异常，任何一条路径都不能让客户的消息石沉大海。兜底文案指向人工，是客服场景的标准动作。
五、落地三步
- 验证通道：控制台注册后按文档第一屏的“5 分钟完成第一次调用”，先给文件传输助手发通一条消息。
- 小流量试跑：脚本上线后先用一两个测试号咨询，盯三个数——回复延迟、重复回复（应为零）、限流触发次数。
- 逐步放量：三个数都稳了再接真实客服号，群消息、多媒体消息等进阶场景按文档模块逐个加。
六、合规边界，先说清楚
自动化能力用于客户主动咨询后的服务响应这类场景是它的正当用途；主动营销、批量群发、诱导分享是被平台打击的行为，也是封号的主因。接口的使用频率、内容边界以平台官方文档为准——机器人是客服，不是营销炮台，这条守不住，上面所有工程投入都是给封号倒计时。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2600_96923743/article/details/169561883
