从0到1构建智能客服agent：基于LLM的实战架构与避坑指南

咖啡 + Code

340人浏览 · 2026-01-31 01:43:07

咖啡 + Code · 2026-01-31 01:43:07 发布

从0到1构建智能客服agent：基于LLM的实战架构与避坑指南

header

背景痛点：规则引擎的“三座大山”

去年我们团队接手某电商售后系统时，老代码里躺着 1.3 万条正则规则，维护人已经离职，留下一句话：“改一条规则，全站回归 3 天”。
痛点总结如下：

冷启动成本高：新场景需要写正则、画流程图、再写单元测试，平均 2 人/周。
泛化能力差：用户一句“我买的白色 42 码鞋子能换成黑色 43 吗？”需要 7 条规则才能覆盖颜色和尺码的排列组合。
上下文断裂：传统槽位填充只能记住上一轮，用户中途问“运费谁出”再回来，状态机直接懵圈。

于是老板拍板：用 LLM 重构，目标 200 并发、P99<800 ms、意图准确率 ≥92%。

技术选型：Rasa vs Dialogflow vs LLM 自研

维度	Rasa 3.x	Dialogflow CX	LLM + 自研框架
中文预训练	需自己训	谷歌通用模型	可接自研/开源 13B
私有部署
多轮状态	有限	可视化画布	代码级灵活
知识更新	重启服务	后台上传	热加载向量库
成本(月)	4 核 8 G ≈ 1 k	0.006 美元/轮	GPU 推理 ≈ 1.2 k

结论：ToB 场景数据不能出机房，LLM 方案胜出。
架构图如下：

arch

核心链路：网关 → 敏感词异步过滤 → 对话状态机（LangChain）→ 向量检索（FAISS）→ LLM → 后处理 → 超时重试 → 回包。

核心实现

1. 用 LangChain 搭状态机

LangChain 的 ConversationBufferWindowMemory 默认把全历史扔给 LLM，200 轮后 token 爆炸。我们重写了一个 CompressedMemory：

from typing import List, Dict
from langchain.schema import BaseMemory

class CompressedMemory(BaseMemory):
    """滑动窗口 + 摘要压缩，token 控制在 1k 以内"""
    def __init__(self, max_token: int = 1024):
        self.max_token = max_token
        self.history: List[Dict[str, str]] = []

    def save_context(self, inputs: Dict[str, str], outputs: Dict[str, str]) -> None:
        self.history.append({"in": inputs.get("query"), "out": outputs.get("reply")})
        self._compress()

    def _compress(self) -> None:
        while self._token_len() > self.max_token:
            # 弹出最早一轮，保留最近 3 轮
            self.history.pop(0)

    def _token_len(self) -> int:
        return sum(len(m["in"]) + len(m["out"]) for m in self.history)

    def load_memory_variables(self, inputs: Dict[str, str]) -> Dict[str, str]:
        return {"history": "\n".join([f"User:{m['in']}\nBot:{m['out']}" for m in self.history[-5:]])}

把 CompressedMemory 塞进 LLMChain，状态机就瘦身成功。

2. 基于 FAISS 的知识库语义检索

知识库格式：Markdown，每段≤512 字，先拆块再向量化。

import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

class FaissIndex:
    def __init__(self, model_name: str = "shibing624/text2vec-base-chinese"):
        self.encoder = SentenceTransformer(model_name)
        self.index = faiss.IndexFlatIP(768)  # 余弦相似度
        self.text_map = []

    def add_docs(self, docs: List[str]) -> None:
        embeddings = self.encoder.encode(docs, normalize_embeddings=True)
        self.index.add(np.array(embeddings, dtype=np.float32))
        self.text_map.extend(docs)

    def search(self, query: str, topk: int = 3) -> List[str]:
        q = self.encoder.encode([query], normalize_embeddings=True)
        scores, idx = self.index.search(np.array(q, dtype=np.float32), topk)
        return [self.text_map[i] for i in idx[0] if i != -1]

实测 4 核 8 G，10 万条向量，平均检索 18 ms，QPS 400 无压力。

3. 对话历史压缩算法与性能对比

方案	平均 token/轮	意图准确率	P99 延迟
全历史	3.2 k	94.1 %	1.3 s
滑动窗口 5 轮	0.9 k	93.8 %	0.7 s
摘要压缩	0.6 k	92.5 %	0.6 s

权衡后选“滑动窗口 5 轮”，准确率掉 0.3 %，延迟降一半。

生产考量

1. 超时重试机制

LLM 推理偶尔 5 s 才回包，不能让前端空等。用 tenacity 包两层重试：

from tenacity import retry, stop_after_attempt, wait_random_exponential

@retry(wait=wait_random_exponential(multiplier=1, max=10), stop=stop_after_attempt(3))
def llm_generate(prompt: str) -> str:
    resp = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        timeout=8
    )
    return resp.choices[0].message.content

超时阈值 8 s，最多 3 次，失败返回兜底话术“人工客服稍后联系您”。

2. 敏感词过滤异步化

把敏感词检测拆成独立服务，用 asyncio 并行：

import aiohttp, asyncio

async def sensitive_check(text: str) -> bool:
    async with aiohttp.ClientSession() as session:
        async with session.post("http://internal-filter/sensitive", json={"q": text}) as resp:
            result = await resp.json()
            return result.get("hit", False)

主流程里 await asyncio.wait_for(sensitive_check(query), timeout=0.1)，超时就当通过，后续离线审计再处理，保证主链路 P99 不受拖累。