微信机器人+AI Agent：如何打造24小时自动客服

一、为什么需要 Agent，接个大模型不够吗

前两篇分别讲了"接大模型"和"AI客服架构"。但真要做7×24小时的自动客服，发现一个尴尬：用户说"帮我查一下我的订单"，大模型只能瞎编——它没有你的业务数据；用户说"帮我把这个产品加入购物车"，大模型也执行不了——它没有调用能力。

AI Agent = 大模型 + 工具调用。Agent 不再只是被动回答问题，而是能：

1. 理解用户意图
2. 选择合适的工具（查订单、改地址、退款申请……）
3. 调用后端业务接口拿到真实数据
4. 把结果组织成自然语言回复用户

微信机器人负责收消息和发消息，Agent 负责中间那条"理解→调用→回复"的链路。两者配合才能真正替代人工。

二、整体架构

微信用户发消息
    │
    ▼
Webhook回调 ──► 消息队列 ──► Agent执行器 ──► 工具层（你的业务API）
                                 │
                                 ▼
                          postText 发送回复

Agent 执行器是核心，它的职责：接收一条消息 → 决定要不要调工具 → 调哪个工具 → 拿到结果后生成最终回复。

三、消息接入：和之前完全一样

回调服务、去重、入队的逻辑不变，直接复用：

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

app = Flask(__name__)
msg_queue = queue.Queue()
seen = set()

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

@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})

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

四、工具定义：告诉 Agent 能做什么

工具是 Agent 的"手脚"，每个工具用一段 JSON 描述清楚：名字、用途、参数。大模型拿到工具列表后，会判断用户消息需要调哪个工具、怎么填参数。

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "query_order",
            "description": "根据订单号查询订单状态、物流信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_no": {
                        "type": "string",
                        "description": "订单号，如 DD20260920001"
                    }
                },
                "required": ["order_no"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "apply_refund",
            "description": "对已支付未发货的订单申请退款",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_no": {"type": "string", "description": "订单号"},
                    "reason":    {"type": "string", "description": "退款原因"}
                },
                "required": ["order_no", "reason"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "change_address",
            "description": "修改未发货订单的收货地址",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_no":  {"type": "string"},
                    "new_address": {"type": "string", "description": "完整收货地址"}
                },
                "required": ["order_no", "new_address"]
            }
        }
    }
]

五、工具实现：对接你的业务后端

每个工具对应一个实际的后端 API 调用，用 Python 封装：

def query_order(order_no: str) -> str:
    """查订单状态——对接你自己的订单系统"""
    try:
        resp = requests.get(
            f"https://your-backend.com/api/order/{order_no}",
            headers={"Authorization": "Bearer 内部密钥"},
            timeout=5
        )
        data = resp.json()
        if data.get("code") != 0:
            return f"未找到订单 {order_no}，请确认订单号是否正确。"
        order = data["data"]
        return (
            f"订单 {order_no} 状态：{order['status']}\n"
            f"物流：{order.get('express', '暂无')} {order.get('tracking_no', '')}\n"
            f"预计送达：{order.get('eta', '未更新')}"
        )
    except Exception as e:
        return f"查询失败：{e}，请稍后再试。"

def apply_refund(order_no: str, reason: str) -> str:
    """申请退款"""
    try:
        resp = requests.post(
            "https://your-backend.com/api/order/refund",
            headers={"Authorization": "Bearer 内部密钥", "Content-Type": "application/json"},
            json={"order_no": order_no, "reason": reason},
            timeout=5
        )
        data = resp.json()
        if data.get("code") == 0:
            return f"退款申请已提交，订单 {order_no} 将在1-3个工作日内处理。"
        return f"退款申请失败：{data.get('msg', '未知原因')}，建议联系人工客服。"
    except Exception as e:
        return f"退款接口异常：{e}"

def change_address(order_no: str, new_address: str) -> str:
    """改地址"""
    ...  # 同上模式

安全红线：写操作（退款、改地址）必须加校验——不能只凭用户一句话就执行。至少要关联用户身份（比如要求先输入手机号验证），或者只允许对用户本人历史订单操作。

六、Agent 执行器：推理 → 调用 → 回复

Agent 的核心逻辑是一个循环：让大模型看用户消息和工具列表，如果大模型决定调工具，就执行工具、把结果喂回去让大模型再判断，直到大模型不再调工具而是给出最终回复。

LLM_URL = "https://api.deepseek.com/v1/chat/completions"
LLM_KEY = "sk-你的大模型Key"

def call_llm(messages, tools, tool_choice="auto"):
    resp = requests.post(
        LLM_URL,
        headers={"Authorization": f"Bearer {LLM_KEY}"},
        json={
            "model": "deepseek-chat",
            "messages": messages,
            "tools": tools,
            "tool_choice": tool_choice,
            "max_tokens": 1000
        },
        timeout=30
    )
    return resp.json()

# 工具名 → 实现函数的映射
TOOL_DISPATCH = {
    "query_order": query_order,
    "apply_refund": apply_refund,
    "change_address": change_address
}

def run_agent(user_content: str) -> str:
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个电商客服Agent。优先使用工具查询真实数据，不要编造。"
                "写操作（退款、改地址）执行前要向用户确认订单号和意图。"
                "不确定的问题请引导转人工，不要猜测。"
            )
        },
        {"role": "user", "content": user_content}
    ]

    # 最多循环 3 轮，防止死循环
    for _ in range(3):
        result = call_llm(messages, TOOLS)
        message = result["choices"][0]["message"]

        # 大模型决定不调工具 → 最终回复
        if not message.get("tool_calls"):
            return message["content"].strip()

        # 大模型决定调工具
        messages.append(message)  # 把 assistant 的 tool_calls 加进历史

        for tool_call in message["tool_calls"]:
            func_name = tool_call["function"]["name"]
            func_args = json.loads(tool_call["function"]["arguments"])

            # 执行工具
            func = TOOL_DISPATCH.get(func_name)
            tool_result = func(**func_args) if func else f"未知工具: {func_name}"

            # 把工具结果喂回大模型
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call["id"],
                "name": func_name,
                "content": tool_result
            })

    # 超过 3 轮兜底
    return "抱歉，我处理这个问题时遇到了困难，建议您转人工客服。"

循环流程示意：

用户: "帮我查一下 DD20260920001 的订单"
  ↓ LLM 判断 → 需要调 query_order
  ↓ 执行 query_order("DD20260920001")
  ↓ 工具返回 "已发货，中通快递 SF123456"
  ↓ 结果喂回 LLM
  ↓ LLM 不再调工具 → 生成自然语言回复：
     "您的订单 DD20260920001 已发货，中通快递 SF123456，预计明天送达。"

七、消费者里串起 Agent

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

            # Agent 执行（含大模型调用 + 工具调用，耗时可能几秒到十几秒）
            reply = run_agent(content)
            send_text(to_wxid, reply)
        finally:
            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)

消费者建议用 2 个线程并行跑 Agent（工具调用本身有 IO 阻塞），但 send_text 仍然必须串行调用，靠队列或互斥锁控制。

八、生产环境必须加的防护

风险	后果	防护
Agent 死循环	同一用户被刷屏	最多 3 轮强制退出 + 超时降级
LLM 编造参数	工具调用失败	工具函数内部做参数校验，失败返回友好提示
并发写操作	两个退款请求同时通过	后端业务接口本身加幂等锁，Agent 层只是调用
用户身份冒用	别人帮你申请退款	写操作前验证身份（手机号后4位 + 短信验证码）
消息积压	Agent 执行慢导致队列堆积	监控队列长度，超阈值告警

九、Agent vs 普通大模型客服

维度	大模型直接回复	Agent + 工具调用
数据来源	训练数据（可能过时、编造）	实时查询你的业务系统
执行能力	只能说话	能触发退款、改地址等操作
确定性	低（开放式问答容易编）	高（有工具就不瞎猜）
实现复杂度	低	中（需要定义工具 + 对接后端）

Agent 的核心价值就是把大模型的"模糊"收窄到工具定义的"精确"范围内。用户问的事如果在工具覆盖内，走工具拿真实数据；不在覆盖内，LLM 兜底或引导转人工。

十、小结

AI Agent 落地微信客服的链路：机器人 API 收消息 → Agent 执行器推理 → 工具调用拿真实数据 → postText 发回复。其中机器人 API 侧的接口职责清晰——Webhook 收、postText 发，不用关心业务逻辑；复杂度全在 Agent 层：定义好工具、写好 system prompt、控制好风险边界。先从 3~5 个高频工具（查订单、转人工、查物流）开始做，跑通后逐步扩充，是风险最低的落地路径。接口细节以官方文档为准；不想自己处理协议层的话，选 WTAPI 这类 HTTP + Webhook 形式的机器人 API，把精力全部放在 Agent 和业务工具上即可。

参考资料

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