最近在做一个智能客服系统的升级项目之前那套基于关键词和规则引擎的老系统实在是有点力不从心了。用户问题稍微复杂点或者换个说法它就“听不懂”了要么答非所问要么直接转人工体验很差。正好在研究大语言模型LLM的应用就决定用 LangChain 来重构一套目标是打造一个既能理解复杂意图又能基于我们内部知识库准确回答的智能客服。整个过程从架构设计到最终上线踩了不少坑也积累了一些实战经验在这里和大家分享一下。传统客服系统无论是基于关键词匹配还是规则树其核心问题在于“僵化”。它只能处理预设好的路径和问题对于自然语言中大量的同义替换、上下文依赖和隐含意图束手无策。维护成本也高业务一变动规则库就得大改。而 LLM 的出现带来了转机它拥有强大的语言理解和生成能力能够处理开放域的对话。但直接调用 LLM API 就像用一把万能钥匙开所有锁力量很大但不够精准尤其是在需要结合特定领域知识比如公司产品文档、客服话术时容易产生“幻觉”胡说八道。这就引出了我们的技术选型为什么是 LangChain直接调用 OpenAI 或国内大模型的 API 当然可以但你需要自己处理很多“胶水”逻辑如何把用户问题和知识库关联起来如何管理多轮对话的历史如何将复杂的任务拆解成多个 LLM 调用LangChain 的价值就在于它提供了“Chain”链和“Agent”代理这些高层抽象把检索、记忆、提示词模板、LLM调用等模块标准化并连接起来。它让开发者能像搭积木一样构建应用专注于业务逻辑而非底层通信极大地提升了开发效率和系统的可维护性。对于我们这个项目模块化设计至关重要它允许我们对意图识别、知识检索、对话生成等环节进行独立优化和迭代。接下来我详细拆解一下核心实现部分的几个关键环节。构建本地知识库向量检索FAISS这是保证回答准确性的基石。我们所有的产品手册、常见问题解答FAQ、历史工单记录都被处理成知识库。流程是先用文本分割器RecursiveCharacterTextSplitter将文档切成大小适中的片段chunk然后使用嵌入模型Embedding Model如text-embedding-ada-002或开源的BGE模型将每个文本块转换为向量最后存入 FAISS 索引。FAISS 是一个高效的向量相似度搜索库能快速从海量知识片段中找出与用户问题最相关的几个。这里的一个优化点是 chunk 的策略单纯按字符数切分可能会把完整的句子或段落打断。我们采用了重叠分割并尝试了按标点、按语义等多种分割器最终选择了一种结合句子和固定长度的混合模式保证了上下文片段的相对完整性。设计带降级Fallback机制的对话链直接让 LLM 去回答所有问题风险很高比如遇到敏感问题、超出知识范围的问题或者模型本身服务不稳定时。我们的对话链ConversationChain设计了一个分层处理流程。首先用户输入会经过一个意图分类器实际上是一个简单的 LLM 调用 固定提示词判断问题属于“知识库问答”、“闲聊”还是“敏感/无效问题”。如果是“知识库问答”则触发 RAG检索增强生成流程先检索相关知识片段然后将“问题知识片段”组合成提示词交给 LLM 生成答案。如果检索结果的相关度得分低于阈值或 LLM 生成失败则触发 fallback转向一个更保守的、预设好的通用回答模板或者直接引导用户转人工。这个链式设计确保了服务的鲁棒性。搭建异步 FastAPI 服务为了应对高并发客服请求我们使用 FastAPI 构建了异步 Web 服务。核心是一个 POST 接口接收用户 ID 和消息返回 AI 回复。异步async/await在这里非常关键因为 LLM 调用和向量检索都是 I/O 密集型操作异步可以避免在等待模型响应时阻塞线程极大提升吞吐量。我们使用langchain的异步支持如acall来调用模型。服务端还维护了一个对话记忆池使用ConversationBufferWindowMemory并限制窗口大小根据用户 ID 来存储和获取历史对话实现多轮上下文。下面是一些关键模块的代码示例遵循了比较清晰的风格。带缓存的知识库加载器为了避免每次服务启动都重新生成向量库我们实现了一个带 TTL生存时间的缓存层。缓存键由知识库文件路径和嵌入模型名称的哈希生成。from datetime import datetime, timedelta from typing import Optional, Dict, Any import hashlib import pickle import faiss from langchain.vectorstores import FAISS from langchain.embeddings import OpenAIEmbeddings class CachedVectorStore: 带TTL缓存的知识库向量存储加载器 def __init__(self, cache_dir: str ./vector_cache, ttl_hours: int 24): self.cache_dir Path(cache_dir) self.cache_dir.mkdir(exist_okTrue) self.ttl timedelta(hoursttl_hours) self._cache_meta: Dict[str, datetime] {} def _get_cache_key(self, doc_paths: list, embedding_model: str) - str: 生成缓存键 content -.join(sorted(doc_paths)) embedding_model return hashlib.md5(content.encode()).hexdigest() def load_or_create( self, doc_paths: list, embedding_model: OpenAIEmbeddings, force_rebuild: bool False ) - FAISS: 加载或创建向量存储 cache_key self._get_cache_key(doc_paths, embedding_model.model_name) cache_file self.cache_dir / f{cache_key}.pkl meta_file self.cache_dir / f{cache_key}.meta # 检查缓存是否有效 if not force_rebuild and cache_file.exists() and meta_file.exists(): with open(meta_file, rb) as f: cache_time pickle.load(f) if datetime.now() - cache_time self.ttl: with open(cache_file, rb) as f: return pickle.load(f) # 重建向量库 # ... (文档加载、分割、嵌入、创建FAISS索引的代码) vector_store self._build_vectorstore(doc_paths, embedding_model) # 保存缓存 with open(cache_file, wb) as f: pickle.dump(vector_store, f) with open(meta_file, wb) as f: pickle.dump(datetime.now(), f) return vector_store敏感词过滤中间件在请求进入核心业务逻辑前我们增加了一个中间件进行输入过滤。from fastapi import Request, HTTPException import re class SensitiveWordFilterMiddleware: 敏感信息过滤中间件 def __init__(self, sensitive_patterns: list): # 可以配置正则表达式模式用于过滤手机号、身份证号、辱骂词汇等 self.patterns [re.compile(p, re.IGNORECASE) for p in sensitive_patterns] async def filter_request(self, request: Request): 过滤请求中的文本内容 body await request.json() user_input body.get(message, ) for pattern in self.patterns: if pattern.search(user_input): # 记录日志并返回安全回复 self._log_sensitive_attempt(request, user_input) raise HTTPException( status_code400, detail您的输入包含不合适的内容请重新表述。 ) # 也可以进行替换操作如脱敏手机号 cleaned_input self._mask_sensitive_info(user_input) body[message] cleaned_input # 修改request的_body这里需要一些hack实际中可能将清洗后的数据放入request.state return body def _mask_sensitive_info(self, text: str) - str: 脱敏个人信息例如将手机号替换为*** phone_pattern re.compile(r1[3-9]\d{9}) return phone_pattern.sub(***, text)对话状态管理使用 Pydantic 模型来严格定义对话状态的数据结构便于验证和序列化。from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class DialogueTurn(BaseModel): 对话轮次模型 role: str # user or assistant content: str timestamp: datetime Field(default_factorydatetime.now) class ConversationState(BaseModel): 对话状态模型 user_id: str turns: List[DialogueTurn] Field(default_factorylist) created_at: datetime Field(default_factorydatetime.now) updated_at: datetime Field(default_factorydatetime.now) def add_turn(self, role: str, content: str): 添加一轮对话 self.turns.append(DialogueTurn(rolerole, contentcontent)) self.updated_at datetime.now() def get_recent_context(self, max_turns: int 5) - str: 获取最近N轮对话作为上下文 recent_turns self.turns[-max_turns*2:] # 考虑一问一答 context \n.join([f{turn.role}: {turn.content} for turn in recent_turns]) return context系统搭建好后要上线生产环境还有一系列考量。压力测试方案我们使用 Locust 编写了压力测试脚本模拟用户并发提问。关键点在于模拟真实的对话流而不仅仅是单次调用。from locust import HttpUser, task, between class ChatbotUser(HttpUser): wait_time between(1, 3) # 用户思考时间 def on_start(self): self.session_id fuser_{random.randint(1000, 9999)} task def send_message(self): payload { user_id: self.session_id, message: random.choice([怎么退款, 产品保修期多久, 介绍一下你们的核心功能]) } with self.client.post(/chat, jsonpayload, catch_responseTrue) as response: if response.status_code 200: response.success() else: response.failure(fStatus: {response.status_code})通过测试我们找到了服务的瓶颈最初是数据库连接数并据此优化了连接池和异步处理逻辑。对话日志的脱敏存储所有对话日志必须脱敏后存储。我们定义了一个日志处理器在写入数据库或日志文件前会自动调用前面提到的敏感信息过滤和脱敏方法将手机号、邮箱等个人信息替换为哈希值或标记。同时日志中只存储对话的元数据如对话ID、时间戳、意图分类和脱敏后的文本原始用户输入在内存中处理完毕后即丢弃。模型冷启动优化服务刚启动时加载大模型和向量库可能耗时几十秒。为了不影响第一个用户的体验我们在服务启动脚本中加入了“预热”环节在健康检查通过前主动用几个简单问题触发一次完整的流程让模型和缓存都准备就绪。另外对于嵌入模型等相对静态的组件我们将其常驻内存而不是每次请求都加载。在实战中我们也总结了一些避坑指南。避免 Prompt 注入永远不要将未经处理的用户输入直接拼接到给 LLM 的指令中。我们实现了一个sanitize函数它会将用户输入中的引号、换行符等进行转义或者更彻底地采用严格的模板将用户输入放在特定的、明确的“用户问题”字段中与系统指令物理隔离。处理长上下文窗口即使模型支持长上下文一次性输入过多的检索结果也会增加成本并可能稀释关键信息。我们采用了“重排序”策略先用向量检索召回 Top-K 个片段比如10个然后用一个更轻量级的交叉编码器模型或基于 LLM 的评估器对这 K 个片段进行相关性重排序只选择 Top-N 个比如3个最相关的片段送入最终生成环节。异步 IO 导致的会话状态混乱在高并发下如果对话状态管理不当可能会出现用户 A 的消息读到了用户 B 的历史。关键在于确保会话状态的访问是线程/协程安全的。我们为每个user_id维护了一个独立的记忆对象并且使用异步锁asyncio.Lock来保护对同一用户状态的并发写操作虽然大部分情况下同一个用户的请求是串行的但这是一个安全防护。整个项目做下来基于 LangChain 的智能客服确实在灵活性和准确性上比传统系统有了质的飞跃。RAG 技术让回答有了依据模块化设计让后续的维护和扩展比如增加一个查询订单状态的工具变得非常清晰。不过在追求更精准回答的同时模型的响应延迟和计算成本也成了需要持续权衡的问题。如何平衡模型精度与响应延迟这可能没有标准答案。我们的策略是分级处理对简单、高频问题尝试用更小的模型或甚至缓存标准答案对复杂问题才启用完整的 RAG 流程和大模型。同时持续监控耗时指标优化检索速度和提示词效率在用户体验和成本之间寻找那个最佳平衡点。这条路还在继续希望这些实践经验对大家有所帮助。