---

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

---

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

去年 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 模式	支持	支持	不支持
数据合规	需跨境	需跨境	境内机房，过保测评

结论：

1. 出海业务→GPT-4，函数调用稳；
2. 预算敏感→国产模型，延迟低；
3. 中间路线→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 只需要：

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

---

结语

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

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

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

---