1. 项目概述为什么我们需要一个带记忆的智能客服最近在折腾AI应用开发的朋友估计没少听人提起LangChain。它确实是个好东西把大模型LLM和外部工具、数据连接起来的活它帮你封装好了大半。但说实话跟着官方教程跑通一个“Hello World”级别的Agent后我总觉得差点意思。最大的痛点就是这Agent怎么跟金鱼似的说完上句就忘了下句用户多问两句或者换个方式问同一个问题它要么答非所问要么又把之前说过的话重复一遍用户体验瞬间跌到谷底。所以我决定动手搞一个真正能用的、带“历史记忆”的智能客服Agent。这不仅仅是给对话加个上下文窗口那么简单。想象一下真实的客服场景用户可能先问“我的订单发货了吗”然后隔了五分钟又问“物流到哪了”。一个合格的客服Agent必须能记住“用户有一个待发货订单”这个事实并在后续对话中主动关联。更进一步它还需要记住自己之前给用户推荐过什么产品、用户表达过什么偏好甚至处理过什么投诉。这种跨越多轮对话的“状态记忆”才是智能客服的灵魂。这个项目就是要把LangChain提供的记忆模块、工具调用和Agent执行逻辑像搭积木一样组合起来构建一个具备实用级记忆能力的对话系统。我们不止于调用API更要深入理解记忆是如何产生、存储、检索和被利用的并解决其中必然会遇到的性能、成本和一致性问题。如果你也受够了“健忘”的AI助手想亲手打造一个更懂事的客服机器人那这篇从零开始的实战记录或许能给你一些直接的参考。2. 核心架构设计记忆模块如何与Agent协同工作在动手写代码之前我们先得把架构想清楚。一个带记忆的Agent核心是处理好“思考-行动-观察-记忆”这个循环。LangChain在这方面提供了丰富的组件但选择太多反而容易让人迷茫。我的设计思路是分层处理将不同类型的记忆放在不同的“存储层”让Agent各取所需。2.1 记忆的三种类型与存储策略首先我们必须区分清楚客服场景下需要的几种记忆对话历史记忆这是最基础的就是按顺序记录用户和AI的每一轮对话。LangChain的ConversationBufferMemory或ConversationBufferWindowMemory就能搞定。但全量存储很快会让上下文爆炸所以通常我会用ConversationSummaryMemory它让LLM定期对之前的对话进行摘要只把摘要存入记忆大大节省了token。对于客服场景我推荐混合使用最近几轮对话用BufferWindowMemory保持细节更早的对话则用SummaryMemory压缩成要点。实体事实记忆这是记忆系统的“数据库”。比如用户说“我叫张三”、“我的订单号是12345”、“我喜欢黑色”。这些是客观事实需要被精准地存储和检索。ConversationEntityMemory可以自动从对话中提取实体如人名、订单号、产品名及其关系但它更适合自由对话。对于结构化的客服数据我更喜欢用VectorStoreRetrieverMemory。具体做法是每当对话中产生一个关键事实例如通过工具调用查询到了订单状态我就用一个小型的嵌入模型比如text-embedding-3-small把这个事实转换成向量然后存入像ChromaDB或FAISS这样的向量数据库。下次用户提问时Agent可以先检索相关的历史事实作为补充上下文。这比把一堆事实文本全塞进提示词要高效和准确得多。Agent状态记忆这是最容易被忽略但至关重要的部分。它记录的是Agent自身的“工作状态”。例如当前是否正在执行一个多步骤任务比如“退货流程”走到了哪一步上一个工具调用的结果是什么用户是否已经验证了身份这类记忆通常是结构化的我习惯用简单的键值对存储比如Python的字典或者更持久化地用SQLite。LangChain的AgentExecutor本身会维护一个intermediate_steps列表这就是一种状态记忆我们可以扩展它。我的架构选择是用ConversationSummaryBufferMemory管理对话流用VectorStoreRetrieverMemory作为核心事实知识库再用一个自定义的StateMemory类来维护Agent的工作状态。这三者通过一个自定义的MemoryManager类进行统一调度。2.2 Agent执行链的定制化改造默认的initialize_agent虽然方便但对我们这种需要精细控制记忆流入流出的场景来说太“黑盒”了。我选择使用AgentExecutor配合自定义的Agent类来构建执行链。核心在于重写Agent的_take_next_step方法或使用LCEL语法构建更灵活的链。我们需要在每一步之前主动从MemoryManager中获取三类记忆并拼接到给LLM的提示词中。在每一步之后根据LLM的输出和工具执行的结果决定哪些信息需要被写回到哪类记忆里。例如当工具返回“订单12345已发货”时这个结果不仅要返回给用户还要被MemoryManager提取关键实体订单号12345状态“已发货”生成向量存入向量数据库。同时ConversationSummaryBufferMemory会记录这轮完整的对话。如果这是一个多步骤流程的完成StateMemory则更新状态为“已完成”。注意这里有一个关键的性能权衡。每次调用LLM前都检索向量库会增加延迟。我的经验是可以设置一个触发机制仅当用户问题中包含明确的指代如“它”、“那个”、“我的订单”或关键词时才触发向量检索。这需要对用户query做一个轻量级的意图识别。3. 环境搭建与核心组件实现理论说再多不如一行代码。我们开始动手搭建。首先确保你的Python环境在3.8以上。3.1 基础依赖安装与模型选择pip install langchain langchain-community langchain-openai chromadb tiktoken这里我选择 OpenAI 的模型作为核心LLM因为它API稳定工具调用Function Calling能力强大这对Agent至关重要。向量数据库我选ChromaDB因为它轻量、易用适合快速原型和中小规模应用。当然你也可以替换为开源的模型如通过Ollama部署的Qwen和向量库如FAISS。import os from langchain_openai import ChatOpenAI from langchain_community.embeddings import OpenAIEmbeddings # 设置你的OpenAI API Key os.environ[OPENAI_API_KEY] your-api-key-here # 初始化LLM。gpt-3.5-turbo性价比高gpt-4-turbo效果更好但贵。 llm ChatOpenAI(modelgpt-3.5-turbo-0125, temperature0) # 初始化嵌入模型用于生成向量 embeddings OpenAIEmbeddings(modeltext-embedding-3-small)3.2 实现分层记忆管理器这是整个项目的核心。我们来创建MemoryManager。from langchain.memory import ConversationSummaryBufferMemory, VectorStoreRetrieverMemory from langchain_community.vectorstores import Chroma from langchain.schema import Document from typing import List, Dict, Any import json class StateMemory: 自定义的Agent状态记忆 def __init__(self): self.state {} def get_state(self, key: str, defaultNone): return self.state.get(key, default) def set_state(self, key: str, value: Any): self.state[key] value # 这里可以添加持久化逻辑比如存入SQLite # self._save_to_db() def clear_state(self, key: str): self.state.pop(key, None) class MemoryManager: def __init__(self, llm, embeddings, vector_store_path./chroma_db): # 1. 对话摘要记忆 self.conversation_memory ConversationSummaryBufferMemory( llmllm, max_token_limit1000, # 控制摘要记忆的token上限 memory_keychat_history, return_messagesTrue ) # 2. 向量存储事实记忆 # 初始化或加载Chroma向量库 self.vectorstore Chroma( persist_directoryvector_store_path, embedding_functionembeddings ) # 创建检索器设置相似度检索的前K个结果 retriever self.vectorstore.as_retriever(search_kwargs{k: 3}) self.fact_memory VectorStoreRetrieverMemory(retrieverretriever) # 3. Agent状态记忆 self.state_memory StateMemory() # 缓存上一次的检索结果避免重复检索 self._last_query self._last_facts [] def get_context_for_agent(self, current_input: str) - str: 为Agent组装完整的记忆上下文 context_parts [] # 获取对话历史已由LangChain格式化为字符串或消息列表 chat_history self.conversation_memory.load_memory_variables({})[chat_history] # 通常我们需要将其转换为字符串格式 chat_history_str \n.join([f{msg.type}: {msg.content} for msg in chat_history]) # **关键优化动态事实检索** # 不是每次都要检索当输入包含指代或明确查询时再检索 if self._should_retrieve_facts(current_input): facts self.fact_memory.load_memory_variables({prompt: current_input})[history] self._last_query current_input self._last_facts facts else: facts self._last_facts # 获取当前Agent状态例如当前在处理什么流程 current_process self.state_memory.get_state(current_process, 常规咨询) # 组装上下文 context_parts.append(f## 当前对话历史最近摘要:\n{chat_history_str}) if facts: context_parts.append(f## 相关历史事实:\n{facts}) context_parts.append(f## Agent当前状态: 处于「{current_process}」流程中。) return \n\n.join(context_parts) def _should_retrieve_facts(self, query: str) - bool: 简单的启发式规则判断是否需要检索事实 trigger_words [我的, 上次, 之前, 它, 那个, 订单, 产品, 物流] # 检查是否有指代性词语或关键实体词 if any(word in query for word in trigger_words): return True # 或者如果这是一个全新的、与上次完全不同的话题简单实现计算词袋重叠度这里简化 if not self._last_query or len(set(query.split()) set(self._last_query.split())) 1: return True return False def save_interaction(self, user_input: str, agent_response: str, tool_outputs: List[Dict] None): 保存一轮完整的交互到记忆系统 # 1. 保存到对话记忆LangChain会自动处理 self.conversation_memory.save_context({input: user_input}, {output: agent_response}) # 2. 从交互中提取事实存入向量记忆这是重点 if tool_outputs: for output in tool_outputs: # 假设工具输出是结构化的例如 {order_id: 12345, status: shipped} # 我们可以将其转换为文本描述并存储 fact_text self._extract_fact_from_tool_output(output) if fact_text: # 使用一个唯一的ID例如工具名时间戳 doc_id ftool_fact_{hash(fact_text)} self.vectorstore.add_documents([Document(page_contentfact_text, metadata{source: tool, id: doc_id})]) # 3. 根据交互内容可能更新Agent状态 # 例如如果agent_response包含“开始退货流程”则更新状态 if 退货流程 in agent_response: self.state_memory.set_state(current_process, 退货处理) def _extract_fact_from_tool_output(self, output: Dict) - str: 从工具输出中提取需要长期记忆的事实文本。 这是一个需要根据你的工具具体返回格式来定制的函数。 # 示例如果输出包含订单信息 if order_id in output and status in output: return f订单 {output[order_id]} 的状态是 {output[status]}。 # 示例如果输出包含用户信息 if user_name in output: return f用户姓名是 {output[user_name]}。 return None这个MemoryManager类扮演了记忆中枢的角色。get_context_for_agent方法会在每次Agent思考前被调用它负责收集所有相关记忆并格式化。save_interaction方法则在每轮对话后调用负责将有价值的信息写回不同的记忆存储中。3.3 定义客服工具集没有工具的Agent只是聊天机器人。智能客服需要能真正“做事”。我们定义几个典型的客服工具。from langchain.tools import tool from typing import Optional tool def query_order_status(order_id: str) - str: 根据订单号查询订单状态。 # 这里应该是连接你的订单数据库的代码。我们模拟一下。 order_database { 12345: {status: 已发货, 物流单号: SF123456789, 商品: 智能音箱}, 67890: {status: 待付款, 商品: 无线耳机} } if order_id in order_database: info order_database[order_id] return f订单 {order_id} 状态{info[status]}。商品{info[商品]}。 (f物流单号{info[物流单号]}。 if 物流单号 in info else ) else: return f未找到订单号 {order_id} 的信息。 tool def lookup_return_policy(product_category: str) - str: 查询某类商品的退货政策。 policy_db { 电子产品: 支持7天无理由退货需商品完好、配件齐全。, 服装: 支持7天无理由退换货需吊牌未拆、未洗涤。, 生鲜: 非质量问题不支持退货如有质量问题请提供照片。 } return policy_db.get(product_category, 通用政策请联系人工客服咨询具体退货流程。) tool def escalate_to_human_agent(reason: str) - str: 将复杂问题转接给人工客服。 # 模拟创建一个工单或发送通知 ticket_id fTICKET-{int(time.time())} # 在实际应用中这里可能是调用CRM系统API return f您的问题已转接工单号{ticket_id}。人工客服将尽快通过电话或在线消息与您联系。转接原因{reason}工具的定义使用了LangChain的tool装饰器这能让LangChain自动识别工具的输入参数和描述并生成适合LLM函数调用的格式。4. 构建并运行带记忆的Agent执行器现在我们把LLM、记忆管理器、工具组合起来创建最终的Agent。4.1 组装Agent执行链我们不使用高层的initialize_agent而是用更底层的create_react_agent来获得更多控制权。ReActReasoning Acting是让Agent“思考-行动”的经典范式。from langchain.agents import create_react_agent, AgentExecutor from langchain.prompts import PromptTemplate from langchain.schema import SystemMessage # 1. 定义系统提示词这是Agent的“人设”和核心指令 system_prompt SystemMessage(content你是一个专业的智能客服助手名字叫“小智”。 你的职责是准确、友好地解答用户关于订单、产品、售后政策的问题并可以协助处理简单的退货、查询流程。 你必须严格遵守以下规则 1. 首先仔细倾听用户的问题并结合对话历史和已知事实进行理解。 2. 如果问题明确且你有对应的工具请毫不犹豫地使用工具。 3. 如果工具返回了结果你需要向用户清晰、完整地解释结果。 4. 如果问题超出你的能力范围如需要主观判断、涉及复杂纠纷请主动使用 escalate_to_human_agent 工具转接人工。 5. 在对话中要自然地引用已知信息例如“根据您之前的订单...”让用户感觉你记得他。 永远保持礼貌和专业。 ) # 2. 创建ReAct Agent提示词模板 # LangChain有内置的ReAct模板但我们自定义一下以融入记忆上下文 prompt_template PromptTemplate.from_template( {system_prompt} ## 当前对话背景信息 {agent_scratchpad} ## 历史记忆与当前状态 {memory_context} ## 工具 {tools} ## 用户当前问题 {input} ## 你的思考过程请一步步推理决定是使用工具还是直接回答 ) # 3. 初始化记忆管理器 memory_manager MemoryManager(llmllm, embeddingsembeddings) # 4. 准备工具列表 tools [query_order_status, lookup_return_policy, escalate_to_human_agent] # 5. 创建Agent agent create_react_agent(llm, tools, prompt_template) # 6. 创建Agent执行器并传入我们自定义的记忆处理逻辑 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 打开详细日志方便调试 handle_parsing_errorsTrue, # 优雅处理LLM输出解析错误 max_iterations5, # 防止Agent陷入死循环 )4.2 实现带记忆注入的执行循环AgentExecutor默认不会调用我们的MemoryManager。我们需要包装一下执行过程。def run_agent_with_memory(user_input: str) - str: 执行一轮带完整记忆的Agent对话 # 步骤1从记忆管理器中获取当前对话的上下文 memory_context memory_manager.get_context_for_agent(user_input) # 步骤2准备Agent的输入将记忆上下文作为 agent_scratchpad 的一部分传入 # 注意这里需要根据你使用的prompt模板调整输入键 agent_input { input: user_input, memory_context: memory_context, system_prompt: system_prompt.content, tools: \n.join([f- {tool.name}: {tool.description} for tool in tools]), agent_scratchpad: # 初始为空由AgentExecutor在执行中填充 } # 步骤3执行Agent try: response agent_executor.invoke(agent_input) agent_output response[output] # 获取Agent执行过程中产生的中间步骤工具调用记录 intermediate_steps response.get(intermediate_steps, []) except Exception as e: agent_output f抱歉处理您的请求时出现了点问题{e} intermediate_steps [] # 步骤4将本轮交互保存到记忆系统 # 我们需要从 intermediate_steps 中提取工具的输出 tool_outputs [] for action, observation in intermediate_steps: if hasattr(action, tool): tool_outputs.append({tool: action.tool, output: observation}) memory_manager.save_interaction(user_input, agent_output, tool_outputs) # 步骤5返回最终回复给用户 return agent_output4.3 进行端到端对话测试让我们模拟一个多轮对话看看记忆是否生效。# 模拟对话 conversation [ 你好我想查一下我的订单12345发货了没, 物流到哪了, # 这里没有提订单号考验记忆 那我另一个订单67890呢, 电子产品比如耳机退货政策是什么 ] print(【客服小智】您好我是智能客服小智请问有什么可以帮您) for user_msg in conversation: print(f\n【用户】{user_msg}) reply run_agent_with_memory(user_msg) print(f【小智】{reply}) time.sleep(1) # 模拟一点延迟预期的理想输出第一轮调用query_order_status返回已发货信息并将“订单12345已发货”存入事实记忆。第二轮用户问“物流到哪了”。MemoryManager的_should_retrieve_facts会触发从向量库中检索到“订单12345已发货”的事实并关联物流单号。Agent可能会说“根据您的订单12345它已发货物流单号是SF123456789您可以用这个单号在官网查询具体位置。” 这体现了事实记忆的关联能力。第三轮用户问另一个订单。Agent会调用工具查询67890并更新记忆。第四轮用户问退货政策。Agent调用lookup_return_policy并可能结合之前对话中提到的“耳机”属于电子产品来给出更精准的回答。5. 性能优化与生产环境考量一个能跑通的Demo和一个能用的生产系统之间隔着无数个坑。以下是几个必须考虑的优化点。5.1 记忆检索的精度与效率平衡问题每次对话都检索向量库延迟高、成本高Embedding API调用费且可能引入不相关噪音。优化方案分层缓存对最近N轮对话的事实直接缓存在内存中如LRU Cache避免频繁查询向量库。混合检索结合关键词检索BM25和向量检索Embedding。先用关键词快速过滤出候选集再用向量做精排。LangChain的EnsembleRetriever可以做到这一点。元数据过滤在存入向量库时为每个事实打上丰富的元数据标签如user_id、session_id、fact_type“order”, “user_info”, “complaint”。检索时优先过滤当前用户和会话相关的数据大幅提升精度和速度。# 改进的save_interaction片段添加元数据 def save_interaction(self, user_input: str, agent_response: str, session_id: str, user_id: Optional[str] None, tool_outputs: List[Dict] None): # ... 其他逻辑 ... if fact_text: metadata { source: tool, session_id: session_id, timestamp: time.time(), fact_type: self._infer_fact_type(fact_text) # 推断事实类型 } if user_id: metadata[user_id] user_id self.vectorstore.add_documents([Document(page_contentfact_text, metadatametadata)])5.2 长期记忆的压缩与遗忘机制问题向量库会无限膨胀旧的不相关记忆会影响检索质量。优化方案基于时间的衰减为记忆条目添加时间戳和“访问频率/最近访问时间”字段。定期清理过于陈旧或长期未被触及的记忆。重要性评分让LLM对提取出的事实进行重要性评分例如1-5分。低分的事实可以被归档或删除。这可以在_extract_fact_from_tool_output步骤后增加一个LLM调用来实现。摘要式归档对于同一主题如同一笔订单的多个事实定期如会话结束时让LLM生成一个总结性段落存入长期记忆并删除原始的琐碎事实。5.3 状态管理的复杂性与持久化问题简单的键值对StateMemory在复杂、多步骤的客服流程如退货、投诉中会变得难以维护。优化方案状态机模式为每个主要的客服流程定义明确的状态机例如退货流程申请-审核-寄回-验货-退款。StateMemory中存储当前状态和上下文。Agent根据状态决定下一步该调用什么工具或询问什么信息。持久化到数据库将状态记忆存入如Redis快速或PostgreSQL可靠。确保用户中途离开后回来还能继续之前的流程。与对话记忆联动当Agent状态改变时可以主动在对话中插入一条系统消息如“系统提示已进入退货申请流程下一步需要您提供退货原因”让LLM的回复更符合当前流程。5.4 工具调用的稳定性增强问题LLM可能生成不合规的工具参数或在不该调用工具时调用。优化方案参数验证与后处理在工具被调用前对LLM生成的参数进行类型验证和范围检查。例如order_id必须是数字或特定格式。工具描述优化精心编写工具的description和args_schema这是引导LLM正确使用工具的最有效手段。描述要清晰、无歧义并包含示例。设置置信度阈值如果Agent对“是否使用工具”或“使用哪个工具”的置信度不高可以通过让LLM输出置信度分数或解析其思考过程的确定性来判断可以设定为不调用工具转而要求用户澄清。6. 常见问题排查与调试技巧在实际搭建过程中你肯定会遇到各种奇怪的问题。这里记录几个我踩过的坑和解决方法。6.1 Agent陷入循环或拒绝使用工具现象Agent不停地“思考”但就是不调用工具或者说“我没有这个功能”。排查检查提示词系统提示词是否明确指令它“请使用工具”ReAct模板是否包含了Thought:Action:Observation:的明确格式LLM可能不理解你期望它输出的格式。检查工具描述工具的函数名和描述是否清晰易懂LLM是根据描述来决定是否调用的。尝试把描述写得更像自然语言任务例如将“query_order_status”描述为“当用户询问订单状态时使用此工具查询最新信息”。开启Verbose模式这是最重要的调试手段。设置AgentExecutor(verboseTrue)你会看到LLM每一步的完整思考链Chain of Thought。看看它卡在哪一步是没理解问题还是格式输出错误。降低Temperature在测试阶段将LLM的temperature设为0或接近0使其输出更确定、更可预测减少随机性带来的干扰。6.2 记忆检索不到或检索错误信息现象明明之前说过但Agent好像不记得。排查检查向量入库save_interaction方法是否被正确调用工具输出的信息是否被成功提取并转换成了Document对象可以在存入后立刻做一次相似度搜索测试。检查检索策略_should_retrieve_facts的逻辑是否太严格或太宽松打印出每次的检索触发条件和检索到的文本看看是否符合预期。Embedding模型问题不同的Embedding模型对同一句话的向量表示差异很大。确保存入和检索使用的是同一个模型。对于中文场景text-embedding-3-small对英文优化更好可以考虑专门的中文Embedding模型如M3E、BGE。元数据过滤如果你使用了元数据检查检索时是否传入了正确的过滤条件如session_id。6.3 上下文长度超限与Token成本控制现象对话进行到后面越来越慢甚至API报错“上下文超长”。排查与优化使用摘要记忆ConversationSummaryBufferMemory是必须的它能将长篇历史压缩成简短摘要。精简事实记忆存入向量库的事实文本要尽可能简洁、信息密度高。避免存入整句对话而是提取核心事实三元组主体关系客体。选择性上下文注入不要在每次提示词中都注入全部记忆。MemoryManager.get_context_for_agent应该只返回最相关的部分。相关性可以由检索分数阈值来控制。监控Token使用使用tiktoken库计算每次请求的token数并设置告警。对于长上下文模型如GPT-4-128k也要关注成本。6.4 处理模糊或冲突的用户输入现象用户说“它坏了”Agent无法理解“它”指代什么。策略指代消解在将用户输入送入Agent前可以先用一个轻量级的NLP模型或规则尝试将代词替换为上一轮对话中提到的实体。例如上一轮在讨论“订单12345”那么这一轮的“它”可以替换为“订单12345”。主动澄清如果指代消解失败或者检索到多个可能实体最好的策略是让Agent主动询问。例如“您指的是之前提到的订单12345还是其他商品” 这比猜错了再补救体验好得多。搭建一个真正智能、实用的带记忆客服Agent是一个不断迭代和调优的过程。从基础的记忆模块拼接到复杂的性能优化和异常处理每一步都需要结合具体的业务场景进行设计。这个项目骨架为你提供了一个坚实的起点但真正的挑战和乐趣在于如何让它在你自己的业务数据和服务流程中变得越来越“聪明”和“可靠”。