背景痛点：规则引擎为何扛不住“长对话+多意图”
过去两年，我维护的客服系统一直用“正则+关键词”硬编码。用户问一句“我昨天买的手机能退吗？顺便把充电器也退了”，规则引擎先匹配“退货”，再匹配“充电器”，结果两条流程互斥，机器人直接宕机。更尴尬的是，当用户追问“那运费谁出？”时，系统把上下文全丢，只能从头再来。
长对话里，用户常一次抛 3~4 个意图，还会随时跳回之前的话题。规则引擎的 if-else 树很快变成“意大利面条”，维护成本指数级上升。我们统计过，对话轮次超过 5 轮，准确率从 92% 跌到 54%，人工接管率飙升。痛定思痛，决定把大语言模型（LLM）搬上生产线。
技术选型：Fine-tuning vs Prompt Engineering vs RAG
| 维度 | 全参数微调 | 纯 Prompt Engineering | RAG（检索增强生成） |
|---|---|---|---|
| 时延 | 高（需加载 7B/13B 模型） | 低（直接调 API） | 中（+向量检索 50~150 ms） |
| 训练成本 | 高（A100×2，3 天） | 0 | 低（仅训 Embedding） |
| 效果 | 垂直领域最佳 | 依赖提示词，易幻觉 | 可解释、可溯源 |
| 数据需求 | 万级标注 | 几十例提示 | 千级文档即可 |
客服场景要“一周上线、成本可控、答案可溯源”，我们最终采用“Prompt Engineering + RAG”双轨：
- 高频简单问法 → 用 Prompt 模板，200 ms 内返回；
- 长尾复杂问题 → 走 RAG，召回产品手册片段再生成，准确率提升 40%，而成本只有微调的 1/8。
核心实现：Flask+LangChain 端到端代码
1. 系统架构
- API 网关：Flask + Gunicorn，负责鉴权、限流、日志；
- 对话引擎：LangChain 的
ConversationBufferWindowMemory+ 自定义状态机； - 意图模型：BERT-base 微调，输出 32 类意图；
- 知识检索：Milvus 向量库，召回 Top5 段落。
2. Flask 网关层（PEP8 带类型标注）
# app.py from flask import Flask, request, Response from typing import Dict, Any import json from chat_service import ChatService # 下层 LangChain 封装 app = Flask(__name__) chat_service = ChatService() @app.post("/api/v1/chat") def chat() -> Response: """统一聊天接口，返回 SSE 流式事件""" try: user_id: str = request.json["user_id"] query: str = request.json["query"] stream = chat_service.stream_chat(user_id, query) return Response(stream, mimetype="text/event-stream") except KeyError as e: return Response(f"Missing key: {e}", status=400)3. LangChain 对话状态跟踪
# chat_service.py from langchain.memory import ConversationBufferWindowMemory from langchain.chains import ConversationalRetrievalChain from langchain.llms import OpenAI from langchain.prompts import PromptTemplate class ChatService: def __init__(self, k: int = 5) -> None: self.memory = ConversationBufferWindowMemory(k=k) self.retriever = self._build_retriever() # Milvus 连接 self.chain = ConversationalRetrievalChain.from_llm( llm=OpenAI(temperature=0.1), retriever=self.retriever, memory=self.memory, combine_docs_chain_kwargs={ "prompt": self._build_prompt_template() } ) def stream_chat(self, user_id: str, query: str): # 先跑意图分类 intent: str = self._predict_intent(query) if intent == "order_cancel": # 状态机跳转 self.memory.save_context({"question": query}, {"answer": "已为您取消订单"}) else: for chunk in self.chain.stream({"question": query}): yield f"data: {json.dumps(chunk)}\n\n"4. BERT 意图微调（含数据清洗）
# intent_train.py import pandas as pd from sklearn.model_selection import train_test_split from transformers import BertTokenizer, BertForSequenceClassification, Trainer, TrainingArguments from torch.utils.data import Dataset import torch class IntentDataset(Dataset): def __init__(self, texts, labels): self.encodings = tokenizer(texts, truncation=True, padding=True, max_length=64) self.labels = labels def __getitem__(self, idx): item = {key: torch.tensor(val[idx]) for key, val in self.encodings.items()} item["labels"] = torch.tensor(self.labels[idx]) return item def __len__(self): return len(self.labels) # 1. 数据清洗：去掉少于 3 字的噪音 df = pd.read_csv("raw_chat.csv") df = df[df["text"].str.len() > 2] df["label"] = df["intent"].astype("category").cat.codes train, val = train_test_split(df, test_size=0.1, random_state=42) tokenizer = BertTokenizer.from_pretrained("bert-base-chinese") model = BertForSequenceClassification.from_pretrained("bert-base-chinese", num_labels=32) training_args = TrainingArguments( output_dir="./intent_model", per_device_train_batch_size=64, num_train_epochs=3, evaluation_strategy="epoch", save_strategy="epoch", logging_dir="./logs", ) trainer = Trainer( model=model, args=training_args, train_dataset=IntentDataset(train["text"].tolist(), train["label"].tolist()), eval_dataset=IntentDataset(val["text"].tolist(), val["label"].tolist()), ) trainer.train()生产考量：GPU 省内存与合规过滤
1. 对话历史压缩算法
长对话直接把 4k token 塞进 GPU，显存瞬间爆炸。我们实现滑动窗口摘要：
- 每轮用 LLM 把前 3 轮压缩成 50 字摘要；
- 摘要 + 最近 2 轮原始消息再送进模型；
- 显存占用从 14 GB 降到 5 GB，首响时间 <800 ms。
def compress_history(messages: list, llm) -> str: if len(messages) <= 3: return "\n".join(messages) summary = llm(f"把以下对话用50字总结：{chr(10).join(messages[:-3])}") return summary + "\n" + "\n".join(messages[-3:])2. 敏感词过滤中间件
双校验：正则速判 + 词库精判，支持热更新。
# middleware.py import re import os from typing import List class SensitiveFilter: def __init__(self, dict_path: str = "sensitive.txt"): with open(dict_path, encoding="utf8") as f: self.words: List[str] = [w.strip() for w in f if w.strip()] self.regex = re.compile("|".join(map(re.escape, self.words))) def mask(self, text: str) -> str: if self.regex.search(text): for w in self.words: text = text.replace(w, "*" * len(w)) return text在 Flask 里加before_request钩子，全部过一遍再进 LLM，审计无忧。
避坑指南：幻觉、超时、幂等
1. 避免 LLM 幻觉的 3 种校验
- 知识边界卡尺：RAG 召回为空时，强制回复“暂无资料，转人工”；
- 数字格式化：让 LLM 输出 JSON，正则校验字段存在与类型；
- 双模型投票：主模型生成 + 小模型打分，低于 0.7 阈值就降权。
2. 对话超时重试的幂等性
用户网络抖动，同一条消息可能重发 3 次。我们在 Redis 里以user_id + md5(query)做幂等键，TTL 60 秒，重复请求直接返回缓存结果，既防重复扣费，又保持体验一致。
代码规范小结
- 所有函数加类型标注与
__future__ import annotations； - 关键路径包
try/except，日志用structlog输出 JSON； - 单元测试覆盖 >80%，CI 流水线强制
black + flake8检查，不过不让合并。
延伸思考：多语言改造方案
要在东南亚市场复制一套泰语/越南语客服，无需重新训大模型，只需三步：
- 意图数据翻译：用 Azure Translator API 把 2 万条中文样本翻成目标语言，再人工抽检 10% 纠偏；
- Embedding 对齐：选 multilingual-mpnet，语义空间天然跨语言，向量库无需重建；
- 生成提示切换：在 Prompt 里加
"Please answer in Thai."，LLM 自动输出泰语，实测 BLEU > 45。
整个改造周期 5 人日，成本不到首次投入的 15%，真正做到“一次建库，多语言复用”。
把系统灰度上线两周，机器人独立解决率从 58% 提到 81%，平均对话轮次减少 1.8 轮，GPU 利用率稳定在 65% 左右。LLM 不是万能，但用对架构、踩完坑，它确实能让客服团队晚上 10 点准时下班。下一步想把语音通道也接进来，让机器人直接“开口”，继续折腾。
