最近帮几个品牌方接智能客服场景是微信服务号和企业微信同时接入大模型做自动回复。一开始大家觉得这事很简单——拿到用户消息拼个 prompt丢给大模型拿回结果回复不就行了实际做起来完全不是这样。消息刚发过来时微信/企微要求接入服务器在几秒内必须应答而大模型接口的响应普遍要一两秒甚至更长加上网络抖动和上下文拼装同步调用经常被平台判定超时。一旦超时平台会重试推送用户那边看到的就是机器人“已读不回”或者同一个问题被重复处理好几遍。除了超时还有高并发问题早上九点推送一条活动几百上千个用户几乎同时发消息回调洪峰瞬间碾过来一台服务器直接被打挂。所以我把这些诉求收敛成一个独立组件——自建 AI 智能客服网关。它不是简单的“转发消息给大模型”而是一个集接入、会话、排队、限流、上报、推送于一体的中间层。这篇文章就把我从零搭建这个网关的完整架构和核心代码写出来包含了踩坑过程、设计取舍和压测数据希望能给正在做类似“微信/企微 大模型”客服系统的人一些参考。1. 为什么客服接入大模型之前要先有一个自建网关先讲清楚一个核心观点网关不是给架构“添乱”而是把平台对接和模型调用这两个本来就不该耦合的东西拆开。1.1 平台回调的 5 秒硬约束注定了不能用同步调用微信服务号、订阅号的服务器配置里平台会持续推送用户消息到我们配置的 URL并且等待响应。公众号回调的等待时间非常短企微的语义也类似——你需要在极短时间内告诉平台“我收到了”否则平台会认为你挂了触发重试甚至停用回调。而大模型的调用链路有多长一条用户消息过来你要做会话历史的拼装、命中知识库的检索、系统提示词的注入、模型厂商 API 的调用运气好一两秒遇到模型排队可能三四秒。这个耗时放在回调的同步链路里是扛不住的。比较好的做法是回调入口只做验签、解密、校验幂等然后立刻返回 200真正的大模型调用放到异步任务里回复通过客服消息接口主动推送回去。这个“快速回包 异步处理 主动推送”的模式是整个网关的第一块基石。很多人上来就调大模型 API然后发现回调一直超时就是这个原因。1.2 网关是唯一能把“业务接入”和“模型接入”解耦的位置品牌方之间差异极大。有的要服务号客服有的用企微有的两个都要模型侧有的用通用大模型有的已经微调过有的接的是私有大模型。如果把这些都写在一个业务逻辑里那每换一个渠道、每换一家模型都要动核心代码。网关把渠道抽象成“上游”把模型抽象成“下游”。上游只管收到消息、推回消息下游只管把文本、工具调用返回给网关。中间的会话管理、上下文拼接、限流、鉴权、日志全部在网关内部处理。这样渠道接入和模型接入就能各自独立演进换模型不影响上游逻辑加渠道也不影响模型侧设计。1.3 限流、熔断、审计这些“配件”也需要集中在网关做AI 客服和传统客服最大的区别是每次回答都要消耗真实成本而且模型供应商有严格的 QPS每秒请求数限制。如果不做控制一个用户反复刷消息可能把整月预算刷掉还可能触发模型厂商限流导致正常用户也得不到回复。网关是控制这些风险的最佳位置。可以在网关层做用户的单会话限流、品牌方的配额控制、对模型供应商的熔断保护还可以把每次请求的用户 ID、消息、回复、耗时、费用流水落库方便后续做审计和对账。这些逻辑如果散落在业务代码里最后会变成一团乱麻。2. 从用户消息到 AI 回复一条完整链路上有哪些必须存在的节点在写代码之前先把我服务的这个大流量链路完整拆一遍。这个图我画了几遍后基本固定下来整体并不复杂但每个节点都有它存在的理由。用户发消息 │ ▼ [接入层] 验签 → 解密 → 幂等校验 → 写队列 → 立刻返回200 │ ▼ [会话状态层] 从 Redis 取会话上下文 / 创建新会话 │ ▼ [异步 Worker] 组装 prompt → 查知识库/RAG → 调用模型 → 拿回复 │ ▼ [消息回写层] 调微信/企微客服消息接口推送回复 │ ▼ [状态回写] 更新会话上下文、写审计日志、更新统计指标2.1 接入层验签、解密、回包这三件事的顺序不要搞错微信和企微的回调都带签名信息。服务号用的是 signature、timestamp、nonce企微在此基础上多了 msg_signature 做整体校验。验签不通过的消息直接丢弃防止伪造请求打进来。企微如果是加密模式还要对消息体做 AES 解密拿到明文 XML 之后再解析出 msgtype 和内容。这个顺序我一开始搞反过先解密后验签结果遇到非法数据直接解密报错还把异常堆栈回给了平台。正确顺序一定是先验签验过了再解密解密后做格式解析。验签不通过就当噪音处理不要浪费任何计算资源。回包也很关键。接入层应该用一个非常轻的 handler 来接收请求决不能在里面做数据库查询或者模型调用。我在实际项目里接入层只做四件事验签、解密、把消息推给队列、返回“success 或空串”。所有 IO 型重活都放到后面异步做。2.2 会话状态层低延迟的前提是拿到上下文不费劲AI 客服需要多轮上下文但微信/企微的消息回调本身是无状态的它不会告诉你“这是同一会话的第几条消息”。所以网关必须自己维护会话 ID 和上下文历史。会话 ID 的生成规则要根据渠道来定服务号可以用用户的 OpenID 作为天然会话标识企微可以用外部联系人 ID 或者客户群的会话 ID。不能把所有用户放在一个全局上下文里否则用户 A 的问题会被用户 B 的聊天记录污染这是新手最容易犯的错。上下文存储我放在 Redis用的 Hash 结构key 是 session_idfield 是历史消息序列化后的字符串。读的时候一次性取出最近 N 轮写的时候追加新轮次并裁剪掉旧的。为什么不放 MySQL因为每一次用户消息都要做一次“读历史 写新历史”的操作MySQL 在高并发场景下既慢又容易被锁拖垮Redis 的纯内存操作能把这次读写的耗时段压到几毫秒。2.3 异步队列与 Worker把 5 秒问题交给主动推送解决接入层收到消息后把规范化后的消息体推进队列然后直接返回。Worker 从队列里捞消息执行“取上下文 → 组装 Prompt → 调模型 → 回写上下文 → 推送回复”这些重操作。队列选型上单机场景可以用 Redis Stream 或者 List分布式场景可以考虑上专业消息队列。我用 Redis Stream 的原因很简单它支持消费者组可以多个 Worker 并发消费消息有持久化能力机器重启后不会直接丢失自带 ACK 机制处理失败还能重新投递。对于客服这种每天几万到几十万条消息的规模这个量级用 Redis Stream 完全够用不必要一开始就上重型队列。2.4 模型网关层路由、RAG、工具的编排位置真正调用大模型的不是业务代码而是网关里的一个模型路由模块。它内部根据配置决定当前请求走哪家模型、用哪个 Prompt 模板、是否要带知识库结果。我把它拆成三个子模块模型路由按渠道、品牌方配置、会话类型选择模型实例支持多模型灰度。RAG 检索先从向量库/知识库检索相关内容把命中结果拼进上下文减少模型乱编的可能。工具调用如果用户要查订单、查物流、转人工模型需要返回结构化工具调用网关解析后执行并再次回填给模型。这三块如果全揉在 Worker 里代码会变得很难维护。用独立模块组织后每块都能单独测试。3. “加了一层网关为什么反而更快”低延迟的关键设计有人会问加了一个中间层多了一次网络跳转和队列传递怎么还可能更快关键在于用户感知的延迟不是回调响应时间而是“从发消息到收到回复”的总时长。网关的价值在于让纯计算和等待模型返回的过程与平台限时脱钩同时通过复用和缓存减少每个环节的耗时。3.1 用户感知延迟的构成与优化目标一条消息从发出到收到 AI 回复由四段组成用户消息到达接入层的时间、队列排队时间、模型调用耗时、消息推送耗时。接入层因为只做验签解密和写队列延迟通常在几十毫秒级别队列排队时间取决于消费速度正常情况下在几十毫秒模型调用是绝对大头可能占掉 80% 以上的耗时推送走平台客服接口又是几百毫秒。所以我们优化的核心对象不是接入层那几十毫秒而是“排队时间”和“模型耗时”。排队时间靠多 Worker 水平扩展来压模型耗时靠上下文压缩、命中缓存、以及和模型厂商的连接复用来解决。网关恰好在所有这些优化点上都有位置可以下手。3.2 端到端连接复用从框架到上游 SDK 都要调这一步非常容易被忽略。Python 里用 requests 直连大模型接口每次请求都要新建 TCP 连接光握手就要几十毫秒在高并发下还会造成端口和句柄泄漏。换成 httpx 并开启连接池后TCP 握手只发生一次后续请求走复用连接模型调用耗时能省下不少。对 Redis 的连接也一样。每来一条消息都重新建连是不可接受的需要在服务启动时初始化 Redis 连接池后续操作复用连接。对模型服务商的 SDK也要检查它是否内置了连接池没有的话就用自己的 httpx 封装。这一轮改完整体 p95 延迟通常能下降 20% 到 30%。3.3 上下文压缩与热点缓存让模型少算让用户少等大模型处理时间跟输入长度强相关。如果每次都把最近 20 轮完整对话发给模型不仅慢还费钱。我实际的做法是保留最近 6 轮对话原文更早的历史用“摘要优先、原文兜底”的方式压缩。比如每天上午的会话会先总结成一段话拼接时优先用摘要。这样既能保持上下文连贯又不会让输入过长。热点缓存也很有用。很多用户问的问题是重复的比如“怎么退换货”“发货多久能到”。我会在网关里做一层语义缓存把用户问题向量化后和近期命中过的高频问题做相似度匹配相似度超过阈值就直接用缓存答案回复不再调用模型。这个方案能把模型调用量降下来一大截延迟也从秒级变成毫秒级。注意只对标准问答做缓存个性化问题不能误命中。3.4 “非流式通道”下的体验补偿策略微信和企微的下发机制不支持像网页聊天那样逐字流式输出所以用户拿到的一定是完整回复。这带来一个体验问题模型思考时间越长用户越觉得“机器人卡了”。我的做法是当上游消息确认是人工可见会话时立刻先推一条“已收到正在思考”的占位提示模型返回完整回复后再把正式答案推过去。这样用户感知到的响应是即时的焦虑感会大幅降低。企微场景还可以用到“正在输入”之类的状态接口虽然不能完全模拟真人打字但能明显改善体验。这块属于锦上添花但做与不做用户口碑差很多。4. 高并发改造实录我踩过的串行、限流与队列堆积的坑低延迟解决的是“单条消息快不快”高并发解决的是“一堆消息来的时候系统还稳不稳”。这两个目标有时会打架需要做一些很具体的取舍。4.1 接入层无状态化扩容的前提也是回滚的前提接入层最忌讳把状态放在进程内存里。比如用进程内变量保存用户会话一旦启动多实例消息会被负载均衡打到不同机器上上下文就找不到了。接入层必须做到无状态所有状态放 Redis所有配置放配置中心每个实例只承担“接收与分发”谁来处理都一样。这样就能在流量高峰前把接入层扩容到多实例由负载均衡统一分发。扩容后不需要迁移任何数据回滚也一样简单。我在一次活动前把接入层从 2 个实例扩到 10 个全程改动只是调整了负载均衡的后端节点。4.2 同一会话消息串行化不能因为并发把上下文搞乱高并发下的一个隐蔽问题是用户快速连发两条消息两个 Worker 同时消费到同一个会话各自去读上下文各自调模型最后后发的那条先返回上下文反而写乱了。必须让同一个 session_id 的消息在任意时刻只被一个 Worker 处理。实现方式是在消费前对 session_id 加一个 Redis 锁或者用 Redis 的 incr 生成自增序列号只有拿到的序号是当前会话期望序号时才能处理。我实际采用的是“分布式锁 会话序号”双重保证锁保证不并发序号保证不越序。这样同一个用户的多条消息即使同时进来也能依次处理、不乱序。4.3 限流与分级熔断保护自己也要保护大模型服务商限流不能只做一层。我在网关里做了三层限流用户级限流单个用户每分钟最多 N 次模型调用防止恶意刷消息。品牌方级限流每个品牌方有单独的配额互不影响。模型供应商级限流全局控制对某个模型 API 的总 QPS避免触发上游限流。如果检测到模型供应商的接口错误率连续升高熔断器打开后续请求直接走兜底话术“系统繁忙请稍后再试”而不是继续打到已经快挂掉的上游。等错误率恢复熔断器再自动关闭。这个设计让网关在模型供应商故障时不会跟着一起崩溃。各层限流参数我放在配置中心里随时可调不用发版。4.4 压测结果与容量估算网关搭建完我做了两轮压测一轮用 wrk 只压接入层一轮用脚本模拟完整链路。单台 4 核 8G 实例的接入层纯验签回包大概能扛到每秒 3000 个请求完整链路因为有模型调用耗时单 Worker 实际吞吐大约每秒 30 到 50 条消息。吞吐瓶颈在模型侧不在网关层。不同环节的容量规划如下表所示这是我按业务预估整理的环节目标容量实际规划扩容方式接入层1000 QPS3 个实例起步水平加实例Redis 会话层每秒 2000 次读写单实例 AOF分片集群异步 Worker200 条/秒10 个 Worker加消费者模型调用50 QPS对接多家模型路由分流5. 核心代码实现FastAPI 网关骨架与关键链路下面直接给出我在生产环境使用的网关骨架语言用的是 Python FastAPI异步模型处理Redis 做会话与队列。代码只保留核心链路一些与业务强相关的字段做了简化。5.1 回调入口验签与快速回包FastAPI 接口接收平台回调。微信和企微的验签算法略有差异但入口结构类似。这里以企业微信加密模式为例公众号去掉解密部分即可核心是“验签 → 解密 → 投递队列 → 立刻返回”。from fastapi import FastAPI, Request, Response import hashlib import json import time app FastAPI() app.api_route(/wecom/callback, methods[GET, POST]) async def wecom_callback(request: Request): params request.query_params # 1. 验签企业微信 msg_signature / 公众号 signature if not verify_signature(params): return Response(contentinvalid, status_code403) # 2. 解密消息体企业微信 AES 解密 plaintext decrypt_message(await request.body(), params) # 3. 转为统一消息结构 msg parse_message(plaintext, channelwecom) # 4. 幂等判断同一 msg_id 不重复投递 if not await is_duplicated(msg.msg_id): await push_to_queue(msg) # 5. 快速回包表示接收成功 return Response(contentsuccess) def verify_signature(params) - bool: # 用 token timestamp nonce echostr 做 sha1 校验 # 企微还要加上 msg_signature 的整体校验 token get_config(wecom_token) elements [token, params[timestamp], params[nonce]] elements.sort() sha hashlib.sha1(.join(elements).encode()).hexdigest() return sha params.get(msg_signature) or sha params.get(signature)注意验签失败返回 403不要返回 200 空包这样平台可以感知到接入异常。企微首次配置 URL 时会带一个 echostr这个值必须原样返回配置才能通过所以 GET 请求和 POST 请求要分别处理。5.2 消息写入队列与幂等控制统一消息结构后把消息推到 Redis Stream 里。这里我用了消息 ID 做幂等避免平台重试导致同一消息被消费两次。import aioredis redis await aioredis.from_url(redis://localhost:6379) async def push_to_queue(msg: dict): # 用 redis stream 实现可靠投递 await redis.xadd( ai:msg:queue, { msg_id: msg[msg_id], session_id: msg[session_id], content: msg[content], channel: msg[channel], user_id: msg[user_id] }, maxlen1000000 ) async def is_duplicated(msg_id: str) - bool: # 用 SET 做短期去重 result await redis.set(fai:dup:{msg_id}, 1, nxTrue, ex3600) return not result幂等键的过期时间设得不能太短平台重试窗口通常有好几分钟我设了 1 小时基本覆盖所有重试场景。另外 Redis Stream 的 maxlen 要设置否则队列无限增长会占满内存。5.3 Worker 消费与模型调用编排Worker 是异步常驻任务消费消息、取上下文、组装 Prompt、调用模型、回写上下文、触发推送。核心流程如下async def worker_loop(): while True: # 从 stream 消费消息支持消费者组 entries await redis.xreadgroup( ai:consumers, worker-1, {ai:msg:queue: }, count10, block2000 ) for _, msgs in entries: for msg_id, data in msgs: await process_message(json.loads(data[msg])) await redis.xack(ai:msg:queue, ai:consumers, msg_id) async def process_message(msg: dict): session_id msg[session_id] # 1. 会话粒度加锁保证同 session 串行 async with session_lock(session_id): # 2. 读取最近的上下文Redis Hash 裁剪 context await load_context(session_id, max_rounds6) # 3. 组装 prompt必要时触发 RAG 检索 prompt build_prompt(context, msg[content]) # 4. 调用模型httpx 连接池复用设置超时 answer await call_llm(prompt, timeout5) # 5. 回写上下文 await save_context(session_id, msg[content], answer) # 6. 调用平台客服接口下发 await send_customer_service_message(msg, answer)为什么这段要加锁而不是简单依赖 Redis 单线程因为加锁保护的是“读上下文→写上下文”这个复合操作如果不锁住两个 Worker 同时读到旧上下文后写的数据会把先写的数据覆盖掉。这个锁用 SET NX 过期时间实现即可锁的粒度是 session_id不会对全局产生大范围阻塞。5.4 主动推送与错误回落Worker 拿到模型回复后调用微信/企微的客服消息接口下发。微信公众号的客服消息接口有 48 小时主动会话限制企微的应用消息则要求用户处于可触达的会话范围内这块需要按渠道分别适配。如果模型调用失败或者整体链路超时我会直接推送一条兜底文案避免用户发出去的消息石沉大海。同时把失败事件写入日志方便后续排查。async def send_customer_service_message(msg: dict, answer: str): if msg[channel] wechat: url https://api.weixin.qq.com/cgi-bin/message/custom/send?access_token await get_access_token() payload { touser: msg[user_id], msgtype: text, text: {content: answer[:2000]}, } elif msg[channel] wecom: url https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token await get_access_token() payload { touser: msg[user_id], msgtype: text, text: {content: answer[:2000]}, agentid: get_config(wecom_agent_id), } async with httpx.AsyncClient() as client: resp await client.post(url, jsonpayload) resp.raise_for_status()注意 access_token 必须做全局缓存和刷新微信的 access_token 有效期只有 2 小时而且接口有调用频率限制频繁刷新会报 45009。我用 Redis 保存 token 并在过期前 5 分钟主动刷新这套逻辑单独封装避免多处并发刷新导致 token 失效。5.5 配置与启动脚本配置管理我用 pydantic-settings 从环境变量读取避免把密钥写进代码。下面是一个简化版本from pydantic_settings import BaseSettings class Settings(BaseSettings): wecom_token: str wecom_encoding_aes_key: str wecom_agent_id: str wx_app_id: str wx_app_secret: str redis_url: str redis://localhost:6379 llm_api_key: str llm_base_url: str llm_model: str gpt-4o-mini max_context_rounds: int 6 user_rate_limit: int 10 class Config: env_file .env settings Settings()启动 Worker 用一个简单的入口脚本生产上用 Supervisor 或容器编排托管。一个进程跑 FastAPI 接入层另一个进程跑 Worker两者可以分开扩容。# 接入层 uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 # Worker 进程 python worker.py6. 线上运行的经验如何让网关在长期运行中保持稳定代码能跑通只是第一步线上长期稳定运行才是门槛。这块我积累了一些很实在的经验挑几个最有用的说一下。6.1 常见故障与定位手段线上最容易出问题的点往往不在网关本身而在平台侧和模型侧。我把最常见的故障整理成了表定位时可以直接照着查。故障现象根因定位手段回调一直验签失败token 配置不一致或负载均衡改了 query 参数打印收到的完整 URL比对平台配置用户收不到回复access_token 过期被并发刷新检查 Redis 里的 token 时间戳加固刷新锁消息被重复处理平台重试 幂等键设置过短查看日志中的重复 msg_id延长幂等时间延迟突然飙高Redis Stream 消费积压Worker 数量不够监控队列长度及时扩容 Worker模型返回大量报错供应商限流或模型服务不稳定看熔断器状态调低全局 QPS 或切换到备用模型6.2 监控指标真正该盯的不是 CPU而是这五个网关的核心指标是队列深度、P95 模型耗时、推送失败率、会话锁等待时延和熔断器状态。抓取这五个指标比盯着 CPU 和内存更有用。队列深度如果持续上涨说明消费能力不够要么加 Worker要么模型调用需要优化。P95 模型耗时模型变慢了用户体感会立刻变差要能看出来是供应商问题还是 Prompt 变长导致的。推送失败率微信/企微客服接口偶发失败需要配合重试策略。会话锁等待时延锁等待过长说明同一个 session 的消息串行化做得太重影响体验。熔断器状态一旦打开就要立即告警因为它意味着模型侧已经出问题。这些指标我全部推到 Prometheus再配 Grafana 看板告警走企业微信机器人。每个品牌方单独打点出了问题能快速定位到具体是哪个流量入口。6.3 发布与灰度AI 客服底下的模型随时要换换错怎么办AI 客服和传统服务最大的不同是模型切换等于直接改变所有用户感知。不能一个配置发上去就全局生效必须支持灰度。我在模型路由里加了灰度权重配置可以按比例把流量切到新模型上。比如先切 5%观察用户投诉率、回复采纳率、错误率稳定后再逐渐放大到 10%、30%、50%最后全量。整个过程中如果发现异常把灰度比例调回 0 即可代码不用动。另外每次模型变更前我会准备好一批回归测试用例把常见问题固定下来发布后统一跑一遍看回复质量是否达到预期。这比在线上靠用户反馈来发现问题要快很多。构建这套微信/企微 AI 智能客服网关最核心的收获是不要把“接大模型”当作一个 API 调用问题而要把它当作一个完整的系统设计问题。回调约束决定了异步架构高并发决定了状态外置稳定性决定了必须有熔断和灰度这些都是平时写单个接口时不会遇到的。代码我给的是可运行的骨架真正要落地还需要根据业务渠道、模型服务商和流量规模做适配但整体链路和设计思路是可以直接拿来参考的。