大模型驱动的智能客服系统实战：从架构设计到性能优化-编程阁

大模型驱动的智能客服系统实战：从架构设计到性能优化

背景痛点：规则引擎的“天花板”

去年 618 之前，我们内部做过一次“旧系统体检”：

意图识别准确率 72%，遇到口语化表达直接“躺平”；
多轮对话靠 if-else 维护，3000+ 条规则，新人两周才敢改一行；
平均响应 1.8 s，高峰期 CPU 飙到 90%，用户排队 40 s 后流失率 35%。

一句话：规则引擎在“语义”和“规模”两条路上都撞到了天花板。
把大模型搬进客服，不是为了赶时髦，而是为了让“答得上”变成“答得准、答得快”。

技术选型：GPT-4、Claude 与国产模型的“三角权衡”

维度	GPT-4	Claude-3-Sonnet	国产某 130B 模型
中文语感	优	优	更优（训练语料 60%+ 中文）
API 单价 (1k tokens)	$0.03	$0.015	￥0.008
首 token 延迟 (P95)	680 ms	520 ms	380 ms
函数调用/JSON 模式	支持	支持	不支持
数据合规	需跨境	需跨境	境内机房，过保测评

结论：

出海业务→GPT-4，函数调用稳；
预算敏感→国产模型，延迟低；
中间路线→Claude，价格与效果折中。

我们最后把“国产 130B”放进生产，用“Claude”做灰度对照组，随时可切换。

核心实现：FastAPI + 状态机 + 大模型封装

1. 系统架构鸟瞰

用户→网关→FastAPI→对话状态机→LLM 服务→缓存/日志→返回

全部走异步，I/O 耗时 70% 的场景直接原地起飞。

2. 对话状态机设计

状态机只干三件事：

记录“用户上一轮说了啥”；
记录“系统上一轮回了啥”；
记录“当前槽位填充度”。

持久化用 Redis Hash：key=session:{uid}，field=history/slots/turn，TTL=30 min。
崩溃重启后，只要 session 没过期，对话还能接着聊。

3. 关键代码（Python 3.11）

# llm_client.py from typing import List, Dict import httpx, json, os class LLMClient: def __init__(self, endpoint: str, key: str, cache_ttl: int = 300): self.endpoint = endpoint self.key = key self.cache = {} # 生产换 Redis self.ttl = cache_ttl async def chat(self, messages: List[Dict[str, str]], temperature: float = 0.3, top_p: float = 0.85, max_tokens: int = 512) -> str: # 1. 构造请求体 payload = { "model": "chat", "messages": messages, "temperature": temperature, "top_p": top_p, "max_tokens": max_tokens, "stream": False } # 2. 缓存 key = hash(str(messages)) cache_key = str(hash(json.dumps(messages, ensure_ascii=False))) if cache_key in self.cache: return self.cache[cache_key] # 3. 异步调用 async with httpx.AsyncClient(timeout=15) as client: r = await client.post( self.endpoint, headers={"Authorization": f"Bearer {self.key}"}, json=payload ) r.raise_for_status() reply = r.json()["choices"][0]["message"]["content"] # 4. 写缓存 self.cache[cache_key] = reply return reply

# prompt_builder.py from datetime import datetime SYS_PROMPT = """你是电商客服助手，请严格遵守： 1. 回答不超过 80 字； 2. 拒绝讨论政治、暴力、色情内容； 3. 不确定时请转人工。""" def build_prompt(history: List[Dict[str, str]], query: str) -> List[Dict[str, str]]: """把历史对话拼成 OpenAI 格式""" messages = [{"role": "system", "content": SYS_PROMPT}] for turn in history[-6:]: # 只保留最近 3 轮 if turn["role"] == "user": content = turn["content"] # 敏感词过滤 content = sensitive_replace(content) messages.append({"role": "user", "content": content}) else: messages.append({"role": "assistant", "content": turn["content"]}) return messages

# main.py from fastapi import FastAPI, Request from pydantic import BaseModel import uvicorn, redis.asyncio as aioredis app = FastAPI(title="LLM-CS") rdb = aioredis.from_url("redis://localhost:6379/0", decode_responses=True) class Msg(BaseModel): uid: str query: str @app.post("/chat") async def chat(msg: Msg): # 1. 取历史 key = f"session:{msg.uid}" hist = await rdb.hget(key, "history") history = json.loads(hist) if hist else [] # 2. 构造 prompt messages = build_prompt(history, msg.query) # 3. 调大模型 answer = await llm.chat(messages) # 4. 更新历史 history.append({"role": "user", "content": msg.query}) history.append({"role": "assistant", "content": answer}) await rdb.hset(key, "history", json.dumps(history, ensure_ascii=False)) await rdb.expire(key, 1800) return {"answer": answer}

性能优化：把 1.8 s 压到 450 ms

1. 缓存策略

对“高频标准问”做精确匹配缓存，命中率 42%，P99 延迟降 60%。
对“相似问”做语义缓存：用向量模型取 embedding，余弦距离 < 0.92 直接返回答案。
向量库用 Qdrant，单机 100 万条 128 维，延迟 18 ms。

2. 异步 IO & 连接池

FastAPI 默认async def已足够，但大模型侧要注意：

httpx.AsyncClient 复用单例，limit=200，keepalive=30 s；
Redis 连接池 max_connections=100，防止“惊群”。

3. 负载测试

Locust 脚本（节选）：

from locustio import HttpUser, task, between class CSUser(HttpUser): wait_time = between(1, 3) @task def ask(self): self.client.post("/chat", json={"uid": "u{{$random}}", "query": "我的订单怎么还没到货"})

结果（4 核 8 G，单实例）：

RPS 稳态 420
P95 响应 450 ms
CPU 占用 68%
内存 1.2 G

横向扩展到 3 实例 + 轮询，可扛 1200 RPS，满足日常 3 倍峰刺。

避坑指南：踩过的坑，都写成代码

1. 敏感词过滤

不要自己写正则，维护成本爆炸。方案：

开源“敏感词库”+ Double-Array Trie，2 万条词库，单次过滤 < 1 ms；
对英文、拼音、谐音变形，用同音字映射+回文，召回率 96%。

2. 对话超时重试

大模型偶尔 15 s 才吐首 token，用户端不能干等。

网关层设置 5 s 返回“正在思考”，后台异步推 WebSocket；
重试策略：指数退避，最多 2 次，超次转人工。

3. 微调数据集清洗

血泪经验：

去掉“客服你好/在吗”这类无意义开头，减少 18% token；
合并同一用户连续 3 句，防止模型学会“自言自语”；
人工抽检 5%，把“答非所问”样本全部踢掉，否则微调后模型会“自信地胡说”。

延伸思考：RAG 让知识库“活”起来

大模型再强，也记不住实时促销规则。下一步：

把商品文档、订单政策切成 512 token 段落，embedding 入库；
用户问题先走向量召回，Top3 段落拼进 prompt，再让大模型生成答案；
实测知识准确率从 78% → 94%，幻觉下降一半。

如果你已经跑通本文的框架，加 RAG 只需要：

在build_prompt前插入retrieve_docs(query)；
把召回文本放在 system prompt 的“背景知识”区域；
限制 max_tokens，防止上下文爆掉。

结语

把大模型塞进客服，不是“换个接口”那么简单：

选型阶段就要把成本、延迟、合规算清楚；
状态机、缓存、重试，一个缺位就会被流量教做人；
最后还得靠 RAG 把“幻觉”关进笼子。

代码已全部在生产跑了一个大促，日活 30 万无重大事故，响应提升 3 倍，人工会话占比从 42% 压到 11%。
如果你也在旧系统里挣扎，希望这份实战笔记能帮你少走一点弯路。