背景痛点：传统客服为什么总被吐槽
做客服系统的老同学都知道，规则引擎就像“写死”的 if-else 树：
- 用户说“我要退货”，必须精准命中关键词“退货”，换个“想退掉”就识别失败
- 新增一条意图，得重新写正则、发版、全量回归测试，节奏慢到怀疑人生
- 多轮对话一旦跳步，状态机就“失忆”，用户得把话再重复一遍
简单 Chatbot 用开源 NLU 模型能缓解一部分，但意图泛化能力依旧有限，且维护知识库的成本随业务线性增长。老板一句“降本增效”，团队只能熬夜加规则，最后把脚本跑成“屎山”。
技术选型：GPT-3.5/4、Claude 还是本地化小模型？
先给出一张对比表，方便一眼看懂（数字为线上实测均值，业务不同会有浮动）：
| 维度 | GPT-3.5-turbo | GPT-4 | Claude-3 | 本地化 7B |
|---|---|---|---|---|
| 成本(1k 会话) | 0.2 美元 | 6 美元 | 0.3 美元 | 0.02 美元(电费) |
| 首 token 延迟 | 0.8 s | 2.5 s | 1.2 s | 0.3 s |
| 意图准确率 | 92% | 96% | 94% | 83% |
| 幻觉率 | 8% | 3% | 5% | 12% |
| 中文闲聊友好 | 优 | 优 | 良 | 中 |
结论速记：
- 预算充足、追求“开箱即用”：GPT-4 直接上，幻觉最少
- 要平衡成本与效果：GPT-3.5 + 本地化兜底，95% 场景够用
- 数据必须留在内网：本地化 7B + LoRA 微调，别指望零 shots 就能打平大模型
核心架构：一张图看懂系统分层
- 接入层：Nginx + HTTPS 白名单，防刷
- 服务层：FastAPI 异步接口，负责验签、限流、敏感词过滤
- 状态机层：维护会话生命周期，驱动“闲聊/问答/工单”三态跳转
- LLM 适配层：统一封装 openai/claude/本地模型，可热切换
- 缓存层：Redis 存热点问答、用户画像、对话历史压缩指纹
- 观测层：Prometheus + Grafana，核心指标：P99 延迟、意图置信度、缓存命中率
对话状态机设计
用“状态+上下文槽位”双维度描述，比纯 DAG 更易扩展：
- 状态：Idle / Greeting / Inquire / Handoff / Evaluate
- 槽位：order_id、return_reason、phone、human_requested
转移示例：
Idle ──用户输入──> Inquire（槽位空）
Inquire ──槽位补齐──> Handoff（生成工单）
Handoff ──用户点“转人工”──> Evaluate（满意度）
Evaluate ──评分完成──> Idle
状态图用 Mermaid 维护，上线前跑 2000 组随机回归，保证无死循环。
上下文记忆实现方案
- 滑动窗口：保留最近 6 轮，超出的做“摘要→嵌入”
- 摘要算法：LLM 二次调用，temperature=0.3，输出≤50 字
- 嵌入存储：用 sentence-transformers 转 384 维向量，放 Redis SET，TTL=24h
- 检索：用户新提问先 embedding，Top-3 相关历史摘要拼进 prompt，控制总 token<2k
代码实现：可直接搬走的 Python 片段
以下代码均跑在 Python 3.10，符合 PEP8，关键行给出中文注释。
1. 异步调用 LLM API 的封装类
import asyncio import openai from typing import List, Dict class LLMClient: def __init__(self, model: str = "gpt-3.5-turbo", max_tokens: int = 512): self.model = model self.max_tokens = max_tokens openai.api_key = "sk-xxx" async def achat(self, messages: List[Dict[str, str]]) -> str: loop = asyncio.get_event_loop() # 使用 run_in_executor 把同步 SDK 转成异步 resp = await loop.run_in_executor( None, lambda: openai.ChatCompletion.create( model=self.model, messages=messages, temperature=0.5, max_tokens=self.max_tokens, stop=["用户:", "客服:"] ) ) return resp.choices[0].message.content.strip()2. 对话历史压缩算法
async def compress_history(history: List[str]) -> str: """把多轮对话压成≤50 字摘要，减少后续 token 消耗""" prompt = ( "请将以下对话压缩成 50 字以内的摘要，保留关键信息：\n" + "\n".join(history) ) summary = await LLMClient(max_tokens=60).achat([{"role": "user", "content": prompt}]) return summary3. 敏感词过滤中间件
import re from fastapi import FastAPI, Request, HTTPException app = FastAPI() SENSITIVE = {"反动", "脏话", "广告"} # 实际用 Trie+DFA，效率 O(1) @app.middleware("http") async def filter_sensitive(request: Request, call_next): body = await request.body() text = body.decode("utf-8") if any(w in text for w in SENSITIVE): raise HTTPException(status_code=400, detail="Input contains sensitive words") response = await call_next(request) return response生产考量：让老板晚上能睡踏实
Redis 缓存
- Key 设计：
faq:md5(question) - 命中率目标>60%，写回 TTL=1h，后台定时批量预热
- Key 设计：
令牌桶限流
- 每个 UID 每秒 3 次，突发 10 次，防脚本刷爆账单
- 用 Redis Lua 脚本保证原子性，代码示例略
监控指标
- 平均响应时间≤1.2s，P99≤3s
- 意图识别准确率≥90%，幻觉率≤5%
- 缓存命中率、限流触发次数、GPU 利用率全部接入钉钉告警
避坑指南：血与泪换来的 checklist
模型幻觉
- 在 prompt 里加“若知识库无答案，请直接回复‘暂无相关信息’，勿编造”
- 后置置信度过滤度过滤，<0.85 触发“人工复核”标签
多轮上下文长度优化
- 摘要+滑动窗口双保险，总 token 不超模型上限 75%
- 对超长订单列表，用“只保留最近 3 条”策略，用户可输入“查看更多”再全量拉取
冷启动流量控制
- 上线前灰度 5% 流量，收集 1k 真实日志做离线回放
- 动态调低 temperature 到 0.3，减少“创造力”带来的不确定性
- 开启“答案一致性检测”，同一问题连续 3 次回答不一致自动降级到人工
留给你思考的问题
- 当模型效果与推理成本呈指数级矛盾时，你会优先砍“准确率”还是“延迟”？
- 如果业务突然要求支持粤语、四川话，你的 prompt 工程 + 语音识别 pipeline 会怎么改？
- 在数据无法出境的限制下，本地化 13B 模型要追平 GPT-4 的 96% 准确率，你会从“预训练”还是“后训练”下手？
把实验结果告诉我，一起交流。祝你也能在下一个“618”大促前，把客服机器人从“智障”升级成“智能”，让值班同学安心睡个整觉。
