从0到1：基于LLM搭建智能客服系统的工程实践与架构演进

从0到1基于LLM搭建智能客服系统的工程实践与架构演进最近在做一个项目需要把传统的、基于规则和关键词匹配的客服系统升级一下。老系统大家懂的都懂用户问个稍微复杂点的问题或者换个说法机器人就懵了要么答非所问要么直接转人工维护那堆规则库更是让人头大。正好大语言模型LLM火起来了它的自然语言理解和多轮对话能力简直是客服场景的“天选之子”于是决定动手从零开始基于LLM搭建一个更智能的客服系统。这篇笔记就记录下整个过程中的思考、踩过的坑和一些实用的代码片段。一、 为什么是LLM告别“人工智障”客服传统的规则引擎客服其核心是“模式匹配”。我们需要预先穷举用户可能的问题和关键词然后配置对应的回答。这种方式的局限性非常明显泛化能力差用户提问方式千变万化“怎么退款”和“钱不要了怎么处理”在规则引擎里可能就是两个完全不同的配置项需要大量维护。无法处理上下文多轮对话是噩梦。用户问“我的订单”系统回复订单号用户接着问“状态呢”规则引擎很难知道这个“状态”指的是上一个订单的状态。冷启动成本高每增加一个业务领域或产品功能都需要投入大量人力梳理话术、编写规则。体验生硬回答通常是固定的模板缺乏灵活性和“人情味”。而LLM的出现恰好解决了这些痛点。它通过在海量文本上训练获得的“知识”和强大的语义理解能力注意力机制/Attention Mechanism能够理解用户意图即使表达方式不同也能捕捉到核心诉求。维持对话状态通过对话历史上下文理解当前问题的所指。生成自然回复回复不再是冷冰冰的模板而是更接近人类的自然语言。当然LLM不是银弹直接调用通用模型做客服会遇到知识更新不及时、可能产生“幻觉”胡编乱造以及成本较高等问题。这就需要我们引入合适的工程架构。二、 技术选型Fine-tuning 还是 RAG确定了用LLM接下来就要选择如何让它具备我们特定的客服知识。主流有两种路径微调Fine-tuning和检索增强生成Retrieval-Augmented Generation, RAG。Fine-tuning微调做法使用我们自己的客服对话数据在基础LLM上进行额外的训练让模型“学习”我们的业务知识和回复风格。优点效果可能更精准回复风格高度可控推理Inference阶段速度快只需调用一次模型。缺点成本高训练需要大量计算资源和数据知识更新麻烦每次更新知识都要重新训练或增量训练可能存在灾难性遗忘。RAG检索增强生成做法将产品文档、FAQ等知识库拆分成片段建立向量数据库。用户提问时先从向量库中检索出最相关的知识片段然后将“问题知识片段”一起作为提示Prompt交给LLM让它基于给定的知识生成回答。优点知识更新容易直接更新向量库即可成本相对较低无需训练只需调用可解释性强答案来源于检索到的文档能有效缓解模型幻觉。缺点效果依赖于检索质量整体链路更长检索生成可能引入不相关信息。对于我们这个项目业务知识如产品功能、政策会频繁更新且初期没有足够的高质量对话数据用于训练。因此RAG路线成为了更合适的选择。它让我们能快速启动并随着业务发展灵活调整知识库。至于基座模型的选择我们综合考量了效果、成本和API稳定性。初期实验了GPT-3.5-Turbo和GPT-4GPT-4效果显著更好但成本高昂。对于大部分常见客服场景GPT-3.5-Turbo在搭配精心设计的Prompt和RAG后已经能提供令人满意的回答因此将其作为主力模型。对于少数复杂或关键的对话可以设计路由策略分流到GPT-4进行处理。三、 核心实现从Prompt到状态管理1. 异步对话处理框架客服系统需要处理高并发请求异步Async是必须的。下面是一个基于asyncio和aiohttp的简易对话处理框架包含了错误重试和基础限流。import asyncio import aiohttp from typing import Optional, Dict, Any from tenacity import retry, stop_after_attempt, wait_exponential import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class AsyncLLMClient: def __init__(self, api_key: str, base_url: str, max_retries: int 3): self.api_key api_key self.base_url base_url self.max_retries max_retries # 简单的令牌桶限流器参数 self.rate_limit_semaphore asyncio.Semaphore(10) # 控制并发数 retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) async def _call_api(self, session: aiohttp.ClientSession, payload: Dict[str, Any]) - Optional[Dict[str, Any]]: 调用LLM API内置重试机制 headers {Authorization: fBearer {self.api_key}, Content-Type: application/json} try: async with session.post(f{self.base_url}/chat/completions, jsonpayload, headersheaders) as response: if response.status 429: logger.warning(Rate limited, retrying...) raise Exception(Rate limit exceeded) # 触发重试 response.raise_for_status() return await response.json() except aiohttp.ClientError as e: logger.error(fHTTP error occurred: {e}) raise except Exception as e: logger.error(fUnexpected error: {e}) raise async def chat_completion(self, messages: list, model: str gpt-3.5-turbo) - Optional[str]: 主要的对话完成方法带有限流 async with self.rate_limit_semaphore: async with aiohttp.ClientSession() as session: payload { model: model, messages: messages, temperature: 0.7, # 控制创造性 max_tokens: 500, } result await self._call_api(session, payload) if result and choices in result and len(result[choices]) 0: return result[choices][0][message][content] return None # 使用示例 async def main(): client AsyncLLMClient(api_keyyour-api-key, base_urlhttps://api.openai.com/v1) messages [{role: user, content: 你好请问如何办理退款}] response await client.chat_completion(messages) print(response) if __name__ __main__: asyncio.run(main())2. 带实体识别的Prompt模板设计单纯的用户问题知识库检索还不够。我们需要在Prompt中明确指令让模型扮演好客服角色并结构化地输出信息。这里我们设计了一个包含系统指令、检索知识和对话历史的模板并引导模型识别关键实体。def build_customer_service_prompt(user_query: str, retrieved_knowledge: str, conversation_history: list) - list: 构建客服Prompt。 :param user_query: 用户当前问题 :param retrieved_knowledge: 从向量库检索到的相关知识 :param conversation_history: 之前的对话记录格式为 [{role:user/assistant, content:...}, ...] :return: 符合OpenAI API格式的messages列表 system_message 你是一个专业的客服助手。请根据以下提供的“相关知识”和“对话历史”专业、友好、准确地回答用户问题。 回答要求 1. 严格基于“相关知识”进行回答如果知识中没有明确信息请如实告知用户“暂时无法处理该问题建议联系人工客服”。 2. 回答需简洁明了重点突出。 3. 如果用户问题中涉及订单号、用户名、手机号等具体信息请在回答中确认这些信息例如“关于您的订单[订单号]...”但不要虚构任何信息。 4. 保持热情、乐于助人的语气。 knowledge_section f【相关知识】\n{retrieved_knowledge}\n if retrieved_knowledge else 【相关知识】\n暂无相关信息。\n # 构建消息列表 messages [{role: system, content: system_message}] # 加入历史对话注意上下文长度限制这里可以截断最老的记录 for msg in conversation_history[-6:]: # 只保留最近3轮对话6条消息 messages.append(msg) # 加入当前查询和知识 user_message_content knowledge_section f用户问题{user_query} messages.append({role: user, content: user_message_content}) return messages3. 对话状态机的Redis存储实现为了支持多轮对话我们需要维护对话状态上下文。Redis因其高性能和丰富的数据结构成为理想选择。这里我们为每个会话Session存储一个消息列表。import json import redis from typing import List, Dict, Any class DialogueStateManager: def __init__(self, redis_client: redis.Redis, session_ttl: int 1800): self.redis redis_client self.session_ttl session_ttl # 会话过期时间单位秒 def _get_key(self, session_id: str) - str: return fcs:dialogue:{session_id} def save_turn(self, session_id: str, role: str, content: str): 保存一轮对话用户或助理的发言 key self._get_key(session_id) message {role: role, content: content} # 使用Redis列表存储对话历史 self.redis.rpush(key, json.dumps(message, ensure_asciiFalse)) self.redis.expire(key, self.session_ttl) # 每次更新都刷新TTL def get_history(self, session_id: str, max_turns: int 10) - List[Dict[str, Any]]: 获取指定会话的最近对话历史 key self._get_key(session_id) # 获取列表最后 max_turns*2 条消息因为一轮对话包含user和assistant两条 stored self.redis.lrange(key, -max_turns*2, -1) history [json.loads(msg) for msg in stored] return history def clear_history(self, session_id: str): 清空某个会话的历史例如问题解决后 key self._get_key(session_id) self.redis.delete(key) # 使用示例 # redis_client redis.Redis(hostlocalhost, port6379, db0) # state_mgr DialogueStateManager(redis_client) # session_id user_12345 # state_mgr.save_turn(session_id, user, 我想退款) # history state_mgr.get_history(session_id) # print(history)四、 生产环境下的关键考量1. 性能与延迟测试上线前我们模拟了不同QPS每秒查询率下的压力。结果发现在QPS达到20左右时平均响应延迟开始显著上升主要瓶颈在于LLM API的调用延迟和网络IO。我们通过以下方式优化异步并发与连接池如上文代码所示使用异步请求和连接池复用。分级缓存对高频通用问题如“营业时间”的答案进行缓存直接返回减少对LLM的调用。响应流式输出对于长回答采用流式Streaming响应让用户能尽快看到开头部分提升体验感知。2. 安全、合规与敏感词过滤直接使用LLM存在生成不当内容的风险必须增加防护层。输入输出过滤在调用LLM前对用户输入进行敏感词过滤在返回给用户前对模型输出再次进行过滤和合规性检查。我们结合了关键词匹配和一个小型的文本分类模型来实现。审核兜底对于模型置信度低或者涉及高风险主题如财务、隐私的回答自动标记并转交人工审核或直接引导至人工客服。五、 避坑指南那些我们踩过的“坑”1. 上下文长度限制与优化LLM有上下文窗口限制如GPT-3.5-Turbo的16K。对话历史太长会装不下导致最早的上下文丢失。策略我们采用“滑动窗口”法只保留最近N轮对话。更精细的策略是利用LLM本身对长历史进行总结Summarization将冗长的历史压缩成一段“背景摘要”再与最近几轮对话一起送入模型。代码提示在get_history方法中控制max_turns或者在build_customer_service_prompt函数中对conversation_history进行截断。2. 应对模型“幻觉”模型有时会自信地编造不存在的信息。RAG是首要防线强制模型基于检索到的知识回答并在Prompt中明确指令见上文模板。置信度评估让模型在回复的同时输出一个对答案确定性的自评分数例如0-1对于低分回答进行特殊处理如提示“信息可能不准确”或转人工。关键信息校验对于模型回复中提到的具体数据、日期、条款编号等尝试与知识库源文档进行二次匹配校验。3. 成本控制与Token计数API调用成本与输入输出的Token数量直接相关。精简Prompt优化系统指令避免冗长。知识库检索时只返回最相关的1-3个片段而非大段全文。监控与告警在代码中统计每次请求的输入/输出Token数并汇总到监控系统。设置每日/每月的Token消耗预算和告警阈值。缓存如前所述对标准问答进行缓存是降低成本最有效的手段之一。六、 开放思考如何评估对话质量系统搭建好了但怎么知道它是不是真的“智能”呢人工一条条看效率太低。我们尝试了以下几种评估维度人工评估抽样对话从“准确性”、“有用性”、“流畅性”等多个维度打分。这是黄金标准但成本高。自动指标检索相关性评估检索到的知识片段与用户问题的匹配度。答案与知识的一致性通过NLI自然语言推理模型判断生成的答案是否与提供的知识矛盾。用户反馈设置“回答是否解决您的问题”的点赞/点踩按钮收集直接反馈。一个更工程化的方法是A/B测试可以将流量随机分为两组A组使用新版的LLM客服B组使用旧版规则引擎或不同配置的LLM模型。然后对比关键业务指标如问题解决率用户在一轮对话后未再提问的比例。人工转接率对话中用户请求转人工的比例。用户满意度评分对话结束后的调查评分。平均对话轮次解决一个问题所需的平均交互次数。通过严谨的A/B测试我们可以用数据驱动决策持续优化Prompt、检索策略和模型选择让智能客服真正越用越聪明。整个从0到1的搭建过程就像在组装一个精密的仪器。LLM提供了强大的“大脑”但需要RAG作为“记忆库”需要稳健的异步框架作为“神经系统”需要状态管理和安全过滤作为“免疫系统”。虽然过程中遇到了上下文管理、幻觉、成本等各种挑战但看到系统能流畅地处理一个个真实的用户问题那种成就感还是非常足的。希望这篇笔记里的思路和代码片段能给你的LLM应用实践带来一些参考。