智能客服Agent项目实战：从架构设计到生产环境部署的避坑指南-编程实验室

背景痛点：高并发下的三座大山

去年公司把客服外包团队砍掉，决定自研一套 Agent 智能客服。需求评审会上，老板只丢下一句话：“618 大促峰值 3 万 QPS，会话不能丢，答案要对，成本别超预算。”
真动手才发现，传统 FAQ 机器人根本扛不住：

并发请求：单实例 Flask 在 4C8G 机器上 500 并发就飘红，CPU 一半耗在 JSON 序列化。
会话状态保持：HTTP 无状态，每次请求都要把历史对话重新喂给 LLM，token 费用翻倍，延迟 2 s 起跳。
意图识别准确率：关键词规则在促销季被用户花式口语吊打，同义词“退货”“退钱”“想退”都能让正则崩溃。

一句话：高并发、状态掉线、答非所问，三座大山一起压下来，谁也跑不了。

技术选型：规则引擎 vs 传统 NLP vs LLM

先给三种方案做个“体检”，用同一批 2 万条真实客服日志回放测试，结果如下：

维度	规则引擎	传统 NLP（BERT 微调）	LLM（GPT-3.5）
响应延迟 P99	120 ms	280 ms	1.8 s
单次成本	0	0.0003 $	0.002 $
可解释性	高，规则可见	中，attention 可可视化	低，黑盒
扩展新意图	需写正则	重标数据+重训	改 prompt
多轮上下文	无	需手动拼接	原生支持

结论：

规则引擎适合高频、标准问答，占总量 60 %，放在最前面挡子弹。
LLM 负责长尾、复杂语义，兜底 15 % 的疑难杂症。
传统 NLP 当“备胎”，等标注数据够了再切流量，目前跳过。

核心实现：Flask+Redis 对话状态管理

1. 整体架构

采用事件驱动 + 缓存分层：
Gateway（统一鉴权）→ Dialogue-State-Manager（DSM，本文主角）→ Intent-Router（规则/LLM 分流）→ Answer-Assembler（拼装答案）。
DSM 用 Redis 做“会话粘性/Session Affinity”存储，JWT 做用户身份绑定，保证网关水平扩容时状态不丢失。

2. 代码演示

以下代码可直接docker-compose up跑起来，已含类型注解、异常捕获与 docstring。

# state_manager.py from typing import Optional, Dict import redis import json import time import jwt from flask import Flask, request, g from functools import wraps app = Flask(__name__) POOL = redis.ConnectionPool(host='redis', port=6379, max_connections=50, decode_responses=True) redis_client = redis.Redis(connection_pool=POOL) JWT_SECRET = "change_me_in_prod" class SessionTimeoutError(Exception): """Raised when dialogue timeout.""" def jwt_required(f): """Decorator to validate JWT and inject user_id.""" @wraps(f) def decorated(*args, **kwargs): token = request.headers.get("Authorization", "").removeprefix("Bearer ") try: payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"]) g.user_id = payload["user_id"] except jwt.InvalidTokenError: return {"msg": "invalid token"}, 401 return f(*args, **kwargs) return decorated def timeout_clean(f): """Auto clean expired dialogue.""" @wraps(f) def decorated(*args, **kwargs): key = f"dlg:{g.user_id}" ttl = redis_client.ttl(key) if ttl < 0: redis_client.delete(key) raise SessionTimeoutError("dialogue timeout") return f(*args, **kwargs) return decorated @app.post("/dialogue") @jwt_required @timeout_clean def append_turn(): """Append a new user utterance and return reply.""" body = request.get_json(force=True) msg: str = body["message"] key = f"dlg:{g.user_id}" history: list = json.loads(redis_client.get(key) or "[]") history.append({"role": "user", "content": msg}) # 省略 intent 识别 & 答案生成 bot_reply = {"role": "assistant", "content": "收到，稍后回复"} history.append(bot_reply) pipe = redis_client.pipeline() pipe.set(key, json.dumps(history)) pipe.expire(key, 600) # 10 min TTL pipe.execute() return {"reply": bot_reply["content"]}

3. 装饰器细节

jwt_required把 user_id 注入 g，省得每层都解码。
timeout_clean先检查 TTL，负数表示 key 已过期，直接抛异常，由前端引导重新建立会话，避免脏数据。

性能优化：Redis 连接池与异步 IO

1. 连接池参数

踩坑记录：默认max_connections=10在 2 k QPS 时把连接打满，新建连接耗时 6 ms，直接拖慢 P99。
调优后：

max_connections=50 socket_keepalive=True socket_keepalive_options={TCP_KEEPID: 60, TCP_KEEPINTVL: 30, TCP_KEEPCNT: 3}

保证复用长连接，断网闪断时内核层快速回收。

2. 压测对比

机器：4C8G，Docker 限制 1 G 内存。
脚本：locust，同一批对话，单轮 10 历史句子。

模式	RPS	平均延迟	CPU
Flask+sync	420	240 ms	95 %
FastAPI+async	1800	55 ms	82 %

换成异步后，网络 IO 等待不再占线程，吞吐量翻 4 倍；再往上加节点即可水平扩容。

避坑指南：上线前必读

对话 ID 必须分布式唯一
早期用user_id+timestamp被并发撞车，两条会话串台。改成UUIDv7可排序 + Snowflake，既唯一又方便日志排序。
警惕 LLM token 超限
历史对话拼接后 4 k token 是常态，超过 8 k 直接报错。做法：
- 先统计 token（用tiktoken库），超限时摘要化中间轮次，只保留系统提示与最近 3 轮。
- 兜底策略：返回“文字太长，请简化问题”，比 500 错误体验好。
敏感词过滤放异步
同步正则 1 ms 能搞定，但关键词库 2 万条后飙到 15 ms。改方案：
- 把内容发到 Celery 任务，Redis 里标记“审核中”，先返回“答案生成中”。
- 审核失败再主动推送“消息因违规被隐藏”，避免阻塞主流程。