一、AI客服和“接个大模型”不是一回事

上一篇讲的是把大模型接到微信机器人上做问答。但真拿去做客服业务，会发现光有 LLM 不够：

• 用户问“运费多少”，LLM 可能编一个答案——客服场景答错比不答更糟
• 用户问“我要退款”，AI 唠半天也解决不了——需要转人工
• 领导问“今天多少咨询、机器人解决了几个”——需要统计

所以完整的 AI 客服系统是一个分层架构：消息接入层、意图识别层、应答层（知识库 + LLM）、人工兜底层、数据统计层。这篇文章按层拆解，给出可直接落地的架构和代码骨架。

二、整体架构

                        ┌──────────────────────────────────┐
微信用户 ◄──► 微信机器人API │ Webhook回调 → 队列 → 意图识别     │
                        │         │                        │
                        │    ┌────┴─────┐                  │
                        │    ▼          ▼                  │
                        │ 知识库命中   LLM生成               │
                        │ (FAQ直出)   (开放问题)             │
                        │    └────┬─────┘                  │
                        │         ▼                        │
                        │   postText 发送回答               │
                        │         │                        │
                        │    高危意图/用户要求 ──► 转人工     │
                        └──────────────────────────────────┘

核心原则：能确定答案的走知识库，不确定的走大模型，涉及交易的转人工。

三、第一层：消息接入

回调服务照旧只做三件事：解析、去重、入队，5 秒内返回 {"ret": 200}。

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

app = Flask(__name__)
msg_queue = queue.Queue()
seen = set()  # 生产环境换 Redis

@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:
        # 幂等去重
        fp = hashlib.md5(
            f"{msg.get('createTime')}_{msg.get('fromUser')}_{msg.get('content')}".encode()
        ).hexdigest()
        if fp not in seen:
            seen.add(fp)
            msg_queue.put(msg)

    return jsonify({"ret": 200})

群消息场景里，客服号通常拉了群，要按上一节说的方式只响应@消息，私聊全量响应。

四、第二层：意图识别

意图识别决定这条消息“谁来回”。工程上够用的做法是三级路由：

第一级：正则/关键词精确匹配——覆盖高频确定性意图（查订单、问价格、转人工）。

第二级：FAQ 相似度匹配——用户换了个说法但意思一样（“怎么退货”=“退货流程是什么”）。

第三级：LLM 兜底——前两级都没接住的开放问题。

import re

# 一级意图：规则表，answer 可以是固定话术，也可以指向业务动作
RULES = [
    (r"转人工|人工客服|真人",              {"intent": "HUMAN",   "answer": None}),
    (r"退款|退货",                        {"intent": "REFUND",  "answer": None}),  # 触发业务动作
    (r"(几点|什么时间)(上班|营业)",        {"intent": "HOURS",   "answer": "工作时间：周一至周日 9:00-21:00"}),
]

def classify_by_rule(content: str):
    for pattern, action in RULES:
        if re.search(pattern, content):
            return action
    return None

二级 FAQ 匹配用向量相似度。简化示意（生产用 embedding 接口 + 向量库）：

# FAQ 库：question 标准问法，answer 标准答案
FAQ = [
    {"q": "怎么修改收货地址", "a": "下单后30分钟内可在订单页自助修改，超时请联系客服改地址哦。"},
    {"q": "支持哪些付款方式", "a": "支持微信支付、支付宝、货到付款。"},
    # ... 积累到几百条
]

def classify_by_faq(content: str, threshold: float = 0.82):
    # 伪代码：content 和 FAQ 所有 q 算 embedding 余弦相似度
    best_q, best_score = None, 0
    for item in FAQ:
        score = cosine_sim(embed(content), embed(item["q"]))  # 按你用的embedding服务实现
        if score > best_score:
            best_q, best_score = item["q"], score
    if best_score >= threshold:
        return {"intent": "FAQ", "answer": best_q["a"]}
    return None

五、第三层：应答与转人工

消费者里串起路由逻辑，高危意图触发转人工通知：

BASE_URL = "https://wx.chuapi.com"
TOKEN = "你的X-finder-TOKEN"
APP_ID = "你的appId"
STAFF_WXID = "客服员工的wxid"

import requests

def send_text(to_wxid, content):
    return 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
    ).json().get("ret") == 200

def ask_llm(content):
    # 上一篇的实现，加 system 约束：只回答业务相关问题，不确定就引导转人工
    ...

def consumer():
    while True:
        msg = msg_queue.get()
        try:
            to_wxid = msg.get("chatRoomId") or msg["fromUser"]
            content = msg["content"]

            # 用户点名的会话处于人工模式 → 直接透传，不再走AI
            if human_mode.get(to_wxid):
                forward_to_staff(msg)
                continue

            # 三级路由
            result = classify_by_rule(content) or classify_by_faq(content)

            if result and result["intent"] == "HUMAN":
                human_mode[to_wxid] = True
                send_text(to_wxid, "已为您转接人工客服，请稍候。")
                send_text(STAFF_WXID, f"用户 {msg.get('nickName')}({to_wxid}) 请求人工：\n{content}")
            elif result and result["intent"] == "REFUND":
                send_text(to_wxid, "退款申请已记录，客服会尽快与您确认。")
                send_text(STAFF_WXID, f"[业务]用户 {to_wxid} 申请退款")
            elif result:
                send_text(to_wxid, result["answer"])
            else:
                answer = ask_llm(content)
                send_text(to_wxid, answer or "这个问题我需要人工帮您确认，已为您记录。")
        finally:
            time.sleep(random.uniform(2, 6))

human_mode = {}

def forward_to_staff(msg):
    send_text(STAFF_WXID, f"[人工模式] {msg.get('nickName')}: {msg['content']}")

转人工的最小实现就是把消息转发给员工微信，员工直接用手机回复；进阶做法是员工也通过 postText 接口用机器人号回复，保持会话统一。

六、会话状态管理

AI 客服比闲聊机器人多一类状态：人工/机器模式切换、满意度收集。用 Redis 存会话状态：

# 状态建议
# wxid → {"mode": "ai" | "human", "human_expire": 时间戳}

def should_use_human(session: dict) -> bool:
    if session.get("mode") == "human":
        # 人工模式 30 分钟无消息自动切回 AI
        return time.time() < session.get("human_expire", 0)
    return False

七、数据统计层

客服系统必须有量。消息流过消费者时顺手打点：

指标	统计方式
日咨询量	按 fromUser 去重计数
机器人解决率	(总回复 - 转人工数) / 总数
热点问题	FAQ 未命中 TopN → 反哺知识库
响应时长	入队时间与发送时间的差值

每天把"未命中问题"导出来人工补充进 FAQ 库，知识库是越滚越准的，这是 AI 客服上线后最重要的运营动作。

八、上线检查清单

检查项	标准
回调响应	5 秒内返回 {"ret": 200}，任何分支都不遗漏
发送频率	1 分钟 ≤ 40 条，消费者随机间隔
超时降级	LLM 30 秒超时，降级为引导话术
敏感过滤	涉及金额、承诺类话术一律转人工，不让 LLM 自由发挥
appId 管理	掉线重登传同一 appId，regionId 用账号常用地区
消息去重	createTime + fromUser + content 指纹
告警	队列积压、连续发送失败、掉线事件都要有通知

九、小结

AI 客服 = 机器人 API（收发消息）+ 意图路由（规则/FAQ/LLM 三级）+ 人工兜底 + 数据运营。消息接入层的接口职责很简单：收消息靠 Webhook 回调，发消息靠 postText，复杂度都在你自己业务层的路由和状态管理里。先上线规则 + 转人工保证不出错，再逐步积累 FAQ、放开 LLM 的回答范围，是风险最低的落地路径。接口细节以官方文档为准；如果不想碰协议层，选 WTAPI 这类 HTTP + Webhook 形式的机器人 API，把精力全部放在业务层即可。

参考资料

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