1. 从“记忆”的混沌到“状态”的清晰一个LangChain老兵的三年踩坑史如果你在过去三年里深度使用过LangChain来构建AI应用尤其是那些需要“记住”对话历史的聊天机器人或智能助手那么“记忆”Memory这个概念大概率曾让你感到困惑甚至抓狂。它无处不在却又难以捉摸它看似简单实现起来却陷阱重重。从最早的ConversationBufferMemory到后来五花八门的ConversationSummaryMemory、ConversationBufferWindowMemory再到各种向量存储记忆、知识图谱记忆……我以及我身边的许多开发者都曾在这片“记忆森林”里迷失过方向。问题的核心在于早期的LangChain将“记忆”视为一个独立的、可插拔的“模块”。你选择一个记忆类配置它然后把它“挂”到链Chain或代理Agent上。听起来很美好对吧但实际操作中你会发现一系列令人头疼的问题记忆的存储和读取时机模糊不清不同记忆方案之间的状态管理混乱在复杂的多步Agent工作流中记忆的传递和更新更是成了一团乱麻。我们常常在代码里写满了memory.save_context(...)和memory.load_memory_variables(...)却依然无法保证在正确的步骤拿到正确的历史信息。这种设计本质上是一种“过程式”的、基于副作用的状态管理它违背了现代应用开发中“状态显式、数据流清晰”的最佳实践。经过长达三年的实践、试错以及至少八种不同记忆方案的折腾LangChain社区包括其核心团队终于迎来了一次根本性的范式转变。这次转变的标志就是从传统的“Memory模块”思维转向了基于LangChain Expression Language (LCEL)和Runnable的“状态”State驱动模型。而最新推出的createAgent等高级API正是这一新范式的集大成者。这次重构不是简单的API改名而是一次从哲学到实践层面的清晰化革命。本文将带你回顾这段从混乱走向清晰的历程拆解新范式的核心原理并分享如何利用createAgent等新工具构建真正可靠、可维护的AI应用记忆系统。2. 旧时代的“八种方案”为何记忆会成为技术债重灾区在深入新世界之前我们必须理解旧世界为何如此令人痛苦。所谓“八种方案”并非一个精确的数字而是形容当时记忆方案碎片化、选择困难的现状。每一种方案都试图解决特定问题但又引入了新的复杂度。2.1 记忆方案的“八宗罪”ConversationBufferMemory对话缓冲记忆最基础也最直接把整个对话历史以字符串形式保存在内存中。问题显而易见对话越长消耗的上下文窗口Context Window就越大最终会触及模型Token限制的天花板且效率低下。ConversationSummaryMemory对话摘要记忆为了解决长对话问题它定期调用LLM对历史进行摘要。这带来了新的问题摘要的准确性、摘要的时机何时触发、以及摘要过程中信息的丢失。更糟糕的是摘要本身也是一次LLM调用增加了成本和延迟。ConversationBufferWindowMemory对话缓冲窗口记忆只保留最近K轮对话。这虽然控制了长度但粗暴地丢弃了早期可能关键的信息对于需要长期记忆的会话场景是致命的。ConversationKGMemory对话知识图谱记忆尝试将对话内容提取为实体和关系存储为知识图谱。理念先进但实现复杂对LLM的提取能力依赖度高且查询效率与图谱设计的优劣强相关不够通用。VectorStoreRetrieverMemory向量存储检索记忆将历史对话片段转换为向量存入向量数据库如Chroma、Pinecone。需要回忆时通过当前查询的向量进行相似性检索。这解决了长尾记忆问题但引入了向量数据库的运维成本并且“检索”行为不一定能精准召回最相关的历史片段存在信息遗漏或噪声。CombinedMemory组合记忆为了兼顾长短记忆出现了将多种记忆组合起来的方案。例如用BufferWindowMemory记近期对话用VectorStoreRetrieverMemory记长期知识。这听起来是终极方案但实际上让状态管理复杂了数倍你需要管理多个记忆源的保存、加载和冲突解决策略。自定义记忆类当内置方案无法满足时开发者被迫继承BaseMemory实现自己的逻辑。这虽然灵活但意味着你需要深入理解LangChain内部的状态流转机制极易写出脆弱且难以调试的代码。完全外部状态管理有些开发者彻底放弃LangChain的Memory模块自己用数据库如Redis、SQLite或全局变量来管理对话状态然后在每次调用链时手动拼接历史。这带来了极致的控制力但也完全失去了LangChain提供的抽象和便利性相当于“重新发明轮子”。2.2 混乱的根源隐式状态与副作用所有这些方案的共同痛点在于它们都依赖于隐式的状态管理和副作用。在旧模式中记忆对象Memory Object通常是一个全局或半全局的单例或者被绑定在链对象内部。当链执行时它会在内部某个地方“悄悄地”调用memory.save_context()将输入输出保存起来。下一次执行时又“悄悄地”从内存中加载这些变量。这个过程对开发者是不透明的。这种模式会引发一系列连锁问题调试地狱你无法清晰地追踪“记忆到底在何时、何地被修改”。当Agent执行多步工具调用时你很难确定在哪一步之后记忆被更新成了什么样子。并发与一致性灾难在Web服务器等并发环境下多个请求共享同一个记忆对象会导致数据错乱。虽然可以通过为每个会话创建新的记忆实例来缓解但这又增加了对象管理的复杂度。组合困难当你试图将多个带有记忆的链组合成一个更复杂的链时记忆的传递和合并规则变得模糊不清。每个子链可能都有自己的记忆逻辑如何让它们协同工作测试困难由于记忆状态是隐式且持久的为链编写单元测试非常棘手。你需要在测试前精心设置记忆状态并在测试后清理测试用例之间容易相互干扰。正是这些深层次的痛点催生了LangChain底层架构的重构需求。社区意识到需要一种更声明式、更函数式、状态流更清晰的方式来管理记忆和整个应用的状态。3. 范式转移从“Memory模块”到“State流”LangChain的重构核心是LangChain Expression Language (LCEL)。LCEL引入了一个革命性的概念一切皆Runnable。一个链、一个工具、一个模型甚至一个简单的字符串处理函数都可以被包装成Runnable。而Runnable的核心特点是它们通过invoke()或stream()方法接受一个输入字典并返回一个输出字典。这个过程是纯函数式的或至少是显式的没有隐式的副作用。在这个新范式下“记忆”不再是一个特殊的、有状态的、会“偷偷”做事的模块。“记忆”被还原为其本质上一轮或前几轮对话的“输入”和“输出”。这些输入输出只不过是应用状态State的一部分。3.1 LCEL与Runnable状态即数据流让我们看一个LCEL的简单示例理解状态是如何流动的from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # 1. 定义组件它们都是Runnable prompt ChatPromptTemplate.from_template(你是一个助手。历史对话{history}\n用户{input}) model ChatOpenAI(modelgpt-4) output_parser StrOutputParser() # 2. 使用管道操作符 | 组合成一个链它也是一个Runnable chain prompt | model | output_parser # 3. 调用链。状态输入被明确地传递进去。 # 注意这里我们手动管理了history这个状态。 state { input: 今天的天气怎么样, history: 用户你好。助手你好我是AI助手。 } response chain.invoke(state) print(response)在这个例子中history作为状态的一部分被明确地传入链中。没有任何隐藏的记忆对象在后台操作。链的执行过程是可预测、可追溯的state-prompt-model-output_parser-response。那么如何实现“记忆”功能呢答案就是编写一个Runnable它的职责是“管理状态”——即在每次调用后将本次的输入和输出合并到历史状态中并传递给下一次调用。这个Runnable可以是一个简单的函数也可以是一个复杂的子链。3.2 构建状态管理Runnable以RunnableWithMessageHistory为例LangChain认识到了这种模式的通用性因此提供了开箱即用的工具来简化这个过程最典型的就是RunnableWithMessageHistory。它的工作原理如下核心链Core Chain你首先需要定义一个不关心“如何获取历史”的核心链。这个链的输入期望包含input和history或messages等字段。历史管理函数Get Session History你提供一个函数这个函数能根据一个session_id返回一个专门用于存储该会话消息列表的对象通常是ChatMessageHistory。RunnableWithMessageHistory这个包装器负责协调。当你调用它时它根据session_id调用你的函数获取到该会话的历史存储对象。它将存储对象中的历史消息history取出与本次的input合并组装成完整的输入字典传递给核心链。核心链执行并产生输出通常是AI的回复消息。包装器将本次的input用户消息和outputAI消息追加到会话的历史存储对象中。返回输出。这个过程完美体现了新范式的思想状态显式历史消息作为输入数据的一部分清晰可见。副作用可控修改历史存储数据库是这个包装器明确声明的职责而不是链的隐藏行为。易于测试你可以轻松模拟Get Session History函数提供固定的历史消息进行测试。并发安全通过session_id隔离不同会话的状态天然支持并发。from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory from langchain_openai import ChatOpenAI # 模拟一个存储会话历史的字典 store {} def get_session_history(session_id: str) - BaseChatMessageHistory: if session_id not in store: store[session_id] ChatMessageHistory() return store[session_id] # 核心链只需要关心如何处理带有messages的输入 prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。), MessagesPlaceholder(variable_namemessages), # 动态注入历史消息 ]) core_chain prompt | ChatOpenAI(modelgpt-4) | StrOutputParser() # 用 RunnableWithMessageHistory 包装核心链 chain_with_history RunnableWithMessageHistory( core_chain, get_session_history, input_messages_keyinput, # 用户最新输入对应的键 history_messages_keymessages, # 历史消息在核心链输入中对应的键 ) # 使用方式 config {configurable: {session_id: user-123}} # 明确指定会话ID response chain_with_history.invoke({input: 你好}, configconfig) print(response) # 助手回复 # 此时store[“user-123”] 中已经保存了用户消息和助手消息 response2 chain_with_history.invoke({input: 我叫小明。}, configconfig) # 这次调用时get_session_history会返回包含上一次对话的历史从而实现记忆。这个例子清晰地展示了记忆功能是如何通过一个专用的、状态管理型的Runnable以声明式、数据流驱动的方式实现的。混乱的“Memory模块”被拆解为一个存储抽象BaseChatMessageHistory、一个状态获取函数get_session_history和一个协调器RunnableWithMessageHistory。每个部分的职责都无比清晰。4.createAgent新范式下的智能体构建最佳实践如果说RunnableWithMessageHistory解决了简单对话链的记忆问题那么createAgent则是将这一范式应用于更复杂的智能体Agent场景的终极答案。智能体通常包含工具调用、多步推理、条件分支等复杂逻辑其状态管理比简单对话链要复杂得多。4.1createAgent的核心哲学createAgent函数通常从langchain.agents导入的设计理念是提供一个高度结构化、基于状态流的方式来构建智能体。它强制你思考以下几个关键部分工具Tools智能体可以调用的函数列表。提示词Prompt指导智能体行为的系统提示和用户指令模板。LLM驱动智能体推理的大语言模型。状态流State Flow这是最关键的部分。你需要定义一个State类型通常使用TypedDict明确描述智能体在每一步推理中所处的完整状态。这个状态通常包括input: 用户的最新问题。chat_history: 历史对话消息列表。intermediate_steps: 到目前为止智能体已经执行过的工具工具输出对列表。这是实现“多步思考”记忆的核心。agent_scratchpad: 一个临时区域用于存储智能体当前步骤的思考过程通常通过提示词模板自动构建。输出解析器负责解析LLM的输出判断是直接回复用户还是调用某个工具。通过明确定义State整个智能体的工作流程就变成了一场围绕这个状态字典的“游戏”。每个步骤一次LLM调用都接收当前状态然后输出一个动作回复或调用工具这个动作会被执行其结果被更新到状态中例如将工具调用和结果追加到intermediate_steps然后状态被送入下一步。4.2 实战用createAgent构建一个带记忆的查询助手让我们构建一个能记住对话历史并能进行多步工具调用的智能体。假设我们有一个查询天气和新闻的工具。from typing import List, TypedDict, Annotated, Union from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain.agents import create_react_agent from langchain.agents.format_scratchpad import format_log_to_messages from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder import operator # --- 第1步定义状态State--- class AgentState(TypedDict): input: str # 用户输入 chat_history: List[BaseMessage] # 完整的对话历史 intermediate_steps: Annotated[List[tuple], operator.add] # 工具调用历史。Annotated用于定义如何合并多个步骤的状态。 # LangGraph等框架利用这个注解来知道如何更新此字段。 # --- 第2步定义工具 --- tool def get_weather(city: str) - str: 获取指定城市的天气信息。 # 模拟实现 return f{city}的天气是晴朗25摄氏度。 tool def get_latest_news(topic: str) - str: 获取关于某个主题的最新新闻摘要。 # 模拟实现 return f关于{topic}的最新新闻AI技术取得新突破。 tools [get_weather, get_latest_news] # --- 第3步定义提示词明确要求使用历史 --- prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以查询天气和新闻。请充分利用对话历史来理解上下文。 如果你需要调用工具请严格按照格式输出。 对话历史{chat_history} 当前问题{input} 你之前的步骤如果有{agent_scratchpad}), ]) # --- 第4步组装智能体 --- llm ChatOpenAI(modelgpt-4, temperature0) # 将工具绑定到LLM使其知道工具的描述 llm_with_tools llm.bind_tools(tools) # 创建智能体执行链 agent ( { input: lambda x: x[input], chat_history: lambda x: x[chat_history], # 将 intermediate_steps 格式化为适合放入提示词的消息 agent_scratchpad: lambda x: format_log_to_messages(x[intermediate_steps]), } | prompt | llm_with_tools | ReActSingleInputOutputParser() # 解析LLM输出为 AgentAction 或 AgentFinish ) # 现在agent 是一个Runnable它接收 AgentState 并输出 Action 或 Finish。 # 要运行它我们需要一个执行循环或使用LangGraph来编排。上面的代码定义了一个智能体的“大脑”agent。它接收一个包含input、chat_history、intermediate_steps的完整状态经过提示词填充、LLM推理、输出解析最终决定下一步做什么。为了运行这个智能体我们需要一个执行循环。在简单场景下可以手动编写from langchain_core.agents import AgentFinish, AgentAction def run_agent(user_input: str, current_state: AgentState) - AgentState: # 1. 调用智能体获取决策 decision agent.invoke(current_state) # 2. 处理决策 if isinstance(decision, AgentFinish): # 智能体决定结束返回最终答案 new_message AIMessage(contentdecision.return_values[output]) new_state { input: , # 本轮处理完毕 chat_history: current_state[chat_history] [HumanMessage(contentuser_input), new_message], intermediate_steps: current_state[intermediate_steps], } return new_state elif isinstance(decision, AgentAction): # 智能体决定调用工具 # 3. 执行工具 tool_to_use {t.name: t for t in tools}[decision.tool] observation tool_to_use.invoke(decision.tool_input) # 4. 更新状态将本次动作观察对加入 intermediate_steps new_intermediate_steps current_state[intermediate_steps] [(decision, observation)] new_state { input: user_input, # 用户问题仍未完全解决保持 chat_history: current_state[chat_history], intermediate_steps: new_intermediate_steps, } return new_state else: raise ValueError(f未知的决策类型{type(decision)}) # 初始化状态 state: AgentState { input: 北京天气怎么样然后告诉我一些科技新闻。, chat_history: [], intermediate_steps: [], } # 执行第一轮 print(用户, state[input]) new_state run_agent(state[input], state) # 假设第一轮决策是调用天气工具并更新了状态。 # 在真实循环中我们会检查 new_state 中的 intermediate_steps 是否包含未完成的动作 # 并继续用更新后的状态调用 agent直到返回 AgentFinish。这个手动循环清晰地展示了状态是如何一步步演进的。chat_history保存了完整的对话记录intermediate_steps保存了工具调用的轨迹。智能体的每一次“思考”都基于这个完整的、显式的状态。4.3 使用LangGraph进行专业级编排手动循环对于理解原理很好但对于生产环境太简陋。LangGraph库正是为了编排这种基于状态的、可能带有循环和分支的工作流而生的。createAgent产生的Runnable可以无缝集成到LangGraph中。from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from typing import Literal from langchain_core.agents import AgentFinish # 定义更丰富的状态类型供LangGraph使用 class GraphState(TypedDict): input: str chat_history: List[BaseMessage] intermediate_steps: List[tuple] # LangGraph需要知道最后输出的内容 final_output: str # 工具执行器 tool_executor ToolExecutor(tools) def should_continue(state: GraphState) - Literal[tools, __end__]: 根据agent的输出来决定下一步是调用工具还是结束。 decision agent.invoke(state) if isinstance(decision, AgentFinish): # 如果是最终结果存入状态并结束 return __end__ else: # 否则将动作存入状态并进入工具调用节点 # 注意这里需要把动作也传递下去一种方法是在状态里加字段或者用全局变量。 # 更规范的做法是使用LangGraph的“边”来传递数据。这里为简化我们假设能处理。 # 实际使用中通常会设计一个call_tool节点来处理。 return tools # 构建图的工作流此处为概念展示完整代码需定义节点和边 # 1. agent_node: 调用我们上面定义的agent输出决策。 # 2. tools_node: 如果决策是调用工具则执行工具并将结果追加到intermediate_steps。 # 3. 条件边从agent_node出来根据should_continue函数决定去tools_node还是END。 # 4. 从tools_node出来总是回到agent_node进行下一步思考。通过LangGraph我们可以图形化地定义智能体的工作流清晰地看到状态如何在“思考节点”和“工具节点”之间流转、更新。这比旧的、基于AgentExecutor的黑盒执行方式要透明和可控得多。5. 重构后的清晰世界经验、技巧与避坑指南经过这次范式重构我们在构建带记忆的AI应用时思路应该彻底转变。以下是我在实践中总结的核心经验和避坑点。5.1 新范式下的核心设计原则状态驱动数据流优先设计应用时首先定义你的State。思考清楚你的应用在运行过程中需要维护哪些信息用户输入、AI回复、工具调用记录、中间计算结果等。把这些信息都放进一个状态字典里。整个应用就是一个接收旧状态、返回新状态的函数或Runnable管道。存储与计算分离BaseChatMessageHistory及其实现如ChatMessageHistory,RedisChatMessageHistory只负责数据的持久化存储CRUD。而RunnableWithMessageHistory或你的自定义Runnable负责在“计算”前后从存储中读取状态、更新状态。这符合单一职责原则。会话隔离靠ID并发安全通过唯一的session_id或thread_id,conversation_id来保证。你的状态获取函数根据这个ID返回对应的历史存储对象。这是实现多用户、多对话并发的基石。拥抱LCEL和Runnable尽可能使用|操作符来组合你的逻辑。LCEL不仅让代码更简洁更重要的是它提供了强大的原语如RunnableLambda包装自定义函数、RunnableParallel并行执行、RunnableBranch条件分支让你能以声明式的方式构建复杂的数据流。5.2 常见陷阱与解决方案陷阱一状态定义过于庞大或模糊问题在State中定义了太多字段或者字段含义不清导致状态更新逻辑复杂且容易出错。解决方案遵循最小化原则。只定义工作流必需的字段。使用TypedDict提供类型提示。对于复杂的嵌套结构可以考虑使用Pydantic模型来定义State以获得更好的验证和IDE支持。陷阱二在错误的地方修改状态问题在某个Runnable的内部函数中直接修改了传入的状态字典产生了不可预料的副作用。解决方案牢记函数式编程的“不可变性”思想。一个Runnable应该根据输入计算输出而不是修改输入。状态的更新应该由专门的状态管理节点如RunnableWithMessageHistory或LangGraph中显式的tools_node来完成。在LCEL中你可以用RunnableLambda来创建纯函数式的状态转换。陷阱三历史消息格式不一致问题chat_history里混用了HumanMessage、AIMessage、SystemMessage但在提示词模板中却错误地引用导致LLM接收到的历史信息混乱。解决方案标准化你的消息格式。通常完整的对话历史就是List[BaseMessage]。在提示词中使用MessagesPlaceholder(variable_namechat_history)来整体注入。确保你保存到历史存储和从历史存储读取的格式是一致的。陷阱四工具调用历史intermediate_steps处理不当问题intermediate_steps的格式不符合format_log_to_messages或你自定义解析函数的期望导致无法正确地将工具调用历史转换成LLM能理解的提示词部分agent_scratchpad。解决方案intermediate_steps的标准格式是List[Tuple[AgentAction, str]]其中AgentAction包含工具名和输入str是工具执行后的观察结果。使用LangChain提供的format_log_to_messages、format_to_openai_function_messages等辅助函数来确保格式正确。如果自定义工具请确保其输出是字符串。陷阱五忽略状态序列化问题在Web应用中你需要将状态在多次HTTP请求间传递。如果状态中包含不可JSON序列化的对象如某些自定义类实例会导致错误。解决方案状态字典中的值应尽量使用基本类型str, int, list, dict或可序列化的Pydantic模型。对于像BaseMessage这样的对象LangChain通常提供了to_dict()和from_dict()方法。在持久化或网络传输前将状态序列化为JSON或字典使用时再反序列化。5.3 性能与扩展性考量历史存储后端的选择开发/轻量级使用内存字典ChatMessageHistory或本地SQLite通过SQLChatMessageHistory。生产环境使用RedisChatMessageHistory高性能支持TTL过期或PostgresChatMessageHistory关系型易于查询分析。选择取决于你的基础设施和访问模式。历史消息的裁剪与摘要即使使用了状态流过长的chat_history仍然会消耗大量Token。策略包括滑动窗口在get_session_history函数中返回时只返回最近N条消息。动态摘要维护一个“摘要”字段在状态中。当历史超过一定长度时调用一个LLM摘要链生成摘要然后用“摘要近期消息”作为新的历史。这比旧的ConversationSummaryMemory更可控因为摘要的触发和存储完全由你决定。向量检索与长期记忆对于需要从大量历史中检索相关片段的场景依然可以结合向量存储。实现方式是在你的状态流中加入一个“检索”节点。该节点根据当前input和chat_history去向量库查询相关文档并将查询结果作为一个新的字段如retrieved_docs加入到状态中供后续的LLM调用使用。这实现了长期记忆与短期对话记忆的分离与融合。从混乱的“Memory模块”到清晰的“State流”这不仅是LangChain API的一次升级更是构建复杂、可靠AI应用在方法论上的一次成熟。它要求开发者从“魔术般的效果”思维转向“精确的工程化”思维。理解并掌握这套基于LCEL和显式状态管理的新范式意味着你能真正驾驭AI应用的记忆与上下文构建出行为可预测、易于调试和维护的智能系统。这三年踩过的坑终将化为清晰道路上的基石。 LangChain记忆系统重构：从Memory模块到State流的技术演进
1. 从“记忆”的混沌到“状态”的清晰一个LangChain老兵的三年踩坑史如果你在过去三年里深度使用过LangChain来构建AI应用尤其是那些需要“记住”对话历史的聊天机器人或智能助手那么“记忆”Memory这个概念大概率曾让你感到困惑甚至抓狂。它无处不在却又难以捉摸它看似简单实现起来却陷阱重重。从最早的ConversationBufferMemory到后来五花八门的ConversationSummaryMemory、ConversationBufferWindowMemory再到各种向量存储记忆、知识图谱记忆……我以及我身边的许多开发者都曾在这片“记忆森林”里迷失过方向。问题的核心在于早期的LangChain将“记忆”视为一个独立的、可插拔的“模块”。你选择一个记忆类配置它然后把它“挂”到链Chain或代理Agent上。听起来很美好对吧但实际操作中你会发现一系列令人头疼的问题记忆的存储和读取时机模糊不清不同记忆方案之间的状态管理混乱在复杂的多步Agent工作流中记忆的传递和更新更是成了一团乱麻。我们常常在代码里写满了memory.save_context(...)和memory.load_memory_variables(...)却依然无法保证在正确的步骤拿到正确的历史信息。这种设计本质上是一种“过程式”的、基于副作用的状态管理它违背了现代应用开发中“状态显式、数据流清晰”的最佳实践。经过长达三年的实践、试错以及至少八种不同记忆方案的折腾LangChain社区包括其核心团队终于迎来了一次根本性的范式转变。这次转变的标志就是从传统的“Memory模块”思维转向了基于LangChain Expression Language (LCEL)和Runnable的“状态”State驱动模型。而最新推出的createAgent等高级API正是这一新范式的集大成者。这次重构不是简单的API改名而是一次从哲学到实践层面的清晰化革命。本文将带你回顾这段从混乱走向清晰的历程拆解新范式的核心原理并分享如何利用createAgent等新工具构建真正可靠、可维护的AI应用记忆系统。2. 旧时代的“八种方案”为何记忆会成为技术债重灾区在深入新世界之前我们必须理解旧世界为何如此令人痛苦。所谓“八种方案”并非一个精确的数字而是形容当时记忆方案碎片化、选择困难的现状。每一种方案都试图解决特定问题但又引入了新的复杂度。2.1 记忆方案的“八宗罪”ConversationBufferMemory对话缓冲记忆最基础也最直接把整个对话历史以字符串形式保存在内存中。问题显而易见对话越长消耗的上下文窗口Context Window就越大最终会触及模型Token限制的天花板且效率低下。ConversationSummaryMemory对话摘要记忆为了解决长对话问题它定期调用LLM对历史进行摘要。这带来了新的问题摘要的准确性、摘要的时机何时触发、以及摘要过程中信息的丢失。更糟糕的是摘要本身也是一次LLM调用增加了成本和延迟。ConversationBufferWindowMemory对话缓冲窗口记忆只保留最近K轮对话。这虽然控制了长度但粗暴地丢弃了早期可能关键的信息对于需要长期记忆的会话场景是致命的。ConversationKGMemory对话知识图谱记忆尝试将对话内容提取为实体和关系存储为知识图谱。理念先进但实现复杂对LLM的提取能力依赖度高且查询效率与图谱设计的优劣强相关不够通用。VectorStoreRetrieverMemory向量存储检索记忆将历史对话片段转换为向量存入向量数据库如Chroma、Pinecone。需要回忆时通过当前查询的向量进行相似性检索。这解决了长尾记忆问题但引入了向量数据库的运维成本并且“检索”行为不一定能精准召回最相关的历史片段存在信息遗漏或噪声。CombinedMemory组合记忆为了兼顾长短记忆出现了将多种记忆组合起来的方案。例如用BufferWindowMemory记近期对话用VectorStoreRetrieverMemory记长期知识。这听起来是终极方案但实际上让状态管理复杂了数倍你需要管理多个记忆源的保存、加载和冲突解决策略。自定义记忆类当内置方案无法满足时开发者被迫继承BaseMemory实现自己的逻辑。这虽然灵活但意味着你需要深入理解LangChain内部的状态流转机制极易写出脆弱且难以调试的代码。完全外部状态管理有些开发者彻底放弃LangChain的Memory模块自己用数据库如Redis、SQLite或全局变量来管理对话状态然后在每次调用链时手动拼接历史。这带来了极致的控制力但也完全失去了LangChain提供的抽象和便利性相当于“重新发明轮子”。2.2 混乱的根源隐式状态与副作用所有这些方案的共同痛点在于它们都依赖于隐式的状态管理和副作用。在旧模式中记忆对象Memory Object通常是一个全局或半全局的单例或者被绑定在链对象内部。当链执行时它会在内部某个地方“悄悄地”调用memory.save_context()将输入输出保存起来。下一次执行时又“悄悄地”从内存中加载这些变量。这个过程对开发者是不透明的。这种模式会引发一系列连锁问题调试地狱你无法清晰地追踪“记忆到底在何时、何地被修改”。当Agent执行多步工具调用时你很难确定在哪一步之后记忆被更新成了什么样子。并发与一致性灾难在Web服务器等并发环境下多个请求共享同一个记忆对象会导致数据错乱。虽然可以通过为每个会话创建新的记忆实例来缓解但这又增加了对象管理的复杂度。组合困难当你试图将多个带有记忆的链组合成一个更复杂的链时记忆的传递和合并规则变得模糊不清。每个子链可能都有自己的记忆逻辑如何让它们协同工作测试困难由于记忆状态是隐式且持久的为链编写单元测试非常棘手。你需要在测试前精心设置记忆状态并在测试后清理测试用例之间容易相互干扰。正是这些深层次的痛点催生了LangChain底层架构的重构需求。社区意识到需要一种更声明式、更函数式、状态流更清晰的方式来管理记忆和整个应用的状态。3. 范式转移从“Memory模块”到“State流”LangChain的重构核心是LangChain Expression Language (LCEL)。LCEL引入了一个革命性的概念一切皆Runnable。一个链、一个工具、一个模型甚至一个简单的字符串处理函数都可以被包装成Runnable。而Runnable的核心特点是它们通过invoke()或stream()方法接受一个输入字典并返回一个输出字典。这个过程是纯函数式的或至少是显式的没有隐式的副作用。在这个新范式下“记忆”不再是一个特殊的、有状态的、会“偷偷”做事的模块。“记忆”被还原为其本质上一轮或前几轮对话的“输入”和“输出”。这些输入输出只不过是应用状态State的一部分。3.1 LCEL与Runnable状态即数据流让我们看一个LCEL的简单示例理解状态是如何流动的from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # 1. 定义组件它们都是Runnable prompt ChatPromptTemplate.from_template(你是一个助手。历史对话{history}\n用户{input}) model ChatOpenAI(modelgpt-4) output_parser StrOutputParser() # 2. 使用管道操作符 | 组合成一个链它也是一个Runnable chain prompt | model | output_parser # 3. 调用链。状态输入被明确地传递进去。 # 注意这里我们手动管理了history这个状态。 state { input: 今天的天气怎么样, history: 用户你好。助手你好我是AI助手。 } response chain.invoke(state) print(response)在这个例子中history作为状态的一部分被明确地传入链中。没有任何隐藏的记忆对象在后台操作。链的执行过程是可预测、可追溯的state-prompt-model-output_parser-response。那么如何实现“记忆”功能呢答案就是编写一个Runnable它的职责是“管理状态”——即在每次调用后将本次的输入和输出合并到历史状态中并传递给下一次调用。这个Runnable可以是一个简单的函数也可以是一个复杂的子链。3.2 构建状态管理Runnable以RunnableWithMessageHistory为例LangChain认识到了这种模式的通用性因此提供了开箱即用的工具来简化这个过程最典型的就是RunnableWithMessageHistory。它的工作原理如下核心链Core Chain你首先需要定义一个不关心“如何获取历史”的核心链。这个链的输入期望包含input和history或messages等字段。历史管理函数Get Session History你提供一个函数这个函数能根据一个session_id返回一个专门用于存储该会话消息列表的对象通常是ChatMessageHistory。RunnableWithMessageHistory这个包装器负责协调。当你调用它时它根据session_id调用你的函数获取到该会话的历史存储对象。它将存储对象中的历史消息history取出与本次的input合并组装成完整的输入字典传递给核心链。核心链执行并产生输出通常是AI的回复消息。包装器将本次的input用户消息和outputAI消息追加到会话的历史存储对象中。返回输出。这个过程完美体现了新范式的思想状态显式历史消息作为输入数据的一部分清晰可见。副作用可控修改历史存储数据库是这个包装器明确声明的职责而不是链的隐藏行为。易于测试你可以轻松模拟Get Session History函数提供固定的历史消息进行测试。并发安全通过session_id隔离不同会话的状态天然支持并发。from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory from langchain_openai import ChatOpenAI # 模拟一个存储会话历史的字典 store {} def get_session_history(session_id: str) - BaseChatMessageHistory: if session_id not in store: store[session_id] ChatMessageHistory() return store[session_id] # 核心链只需要关心如何处理带有messages的输入 prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。), MessagesPlaceholder(variable_namemessages), # 动态注入历史消息 ]) core_chain prompt | ChatOpenAI(modelgpt-4) | StrOutputParser() # 用 RunnableWithMessageHistory 包装核心链 chain_with_history RunnableWithMessageHistory( core_chain, get_session_history, input_messages_keyinput, # 用户最新输入对应的键 history_messages_keymessages, # 历史消息在核心链输入中对应的键 ) # 使用方式 config {configurable: {session_id: user-123}} # 明确指定会话ID response chain_with_history.invoke({input: 你好}, configconfig) print(response) # 助手回复 # 此时store[“user-123”] 中已经保存了用户消息和助手消息 response2 chain_with_history.invoke({input: 我叫小明。}, configconfig) # 这次调用时get_session_history会返回包含上一次对话的历史从而实现记忆。这个例子清晰地展示了记忆功能是如何通过一个专用的、状态管理型的Runnable以声明式、数据流驱动的方式实现的。混乱的“Memory模块”被拆解为一个存储抽象BaseChatMessageHistory、一个状态获取函数get_session_history和一个协调器RunnableWithMessageHistory。每个部分的职责都无比清晰。4.createAgent新范式下的智能体构建最佳实践如果说RunnableWithMessageHistory解决了简单对话链的记忆问题那么createAgent则是将这一范式应用于更复杂的智能体Agent场景的终极答案。智能体通常包含工具调用、多步推理、条件分支等复杂逻辑其状态管理比简单对话链要复杂得多。4.1createAgent的核心哲学createAgent函数通常从langchain.agents导入的设计理念是提供一个高度结构化、基于状态流的方式来构建智能体。它强制你思考以下几个关键部分工具Tools智能体可以调用的函数列表。提示词Prompt指导智能体行为的系统提示和用户指令模板。LLM驱动智能体推理的大语言模型。状态流State Flow这是最关键的部分。你需要定义一个State类型通常使用TypedDict明确描述智能体在每一步推理中所处的完整状态。这个状态通常包括input: 用户的最新问题。chat_history: 历史对话消息列表。intermediate_steps: 到目前为止智能体已经执行过的工具工具输出对列表。这是实现“多步思考”记忆的核心。agent_scratchpad: 一个临时区域用于存储智能体当前步骤的思考过程通常通过提示词模板自动构建。输出解析器负责解析LLM的输出判断是直接回复用户还是调用某个工具。通过明确定义State整个智能体的工作流程就变成了一场围绕这个状态字典的“游戏”。每个步骤一次LLM调用都接收当前状态然后输出一个动作回复或调用工具这个动作会被执行其结果被更新到状态中例如将工具调用和结果追加到intermediate_steps然后状态被送入下一步。4.2 实战用createAgent构建一个带记忆的查询助手让我们构建一个能记住对话历史并能进行多步工具调用的智能体。假设我们有一个查询天气和新闻的工具。from typing import List, TypedDict, Annotated, Union from langchain_core.messages import BaseMessage, HumanMessage, AIMessage from langchain_core.tools import tool from langchain.agents import create_react_agent from langchain.agents.format_scratchpad import format_log_to_messages from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder import operator # --- 第1步定义状态State--- class AgentState(TypedDict): input: str # 用户输入 chat_history: List[BaseMessage] # 完整的对话历史 intermediate_steps: Annotated[List[tuple], operator.add] # 工具调用历史。Annotated用于定义如何合并多个步骤的状态。 # LangGraph等框架利用这个注解来知道如何更新此字段。 # --- 第2步定义工具 --- tool def get_weather(city: str) - str: 获取指定城市的天气信息。 # 模拟实现 return f{city}的天气是晴朗25摄氏度。 tool def get_latest_news(topic: str) - str: 获取关于某个主题的最新新闻摘要。 # 模拟实现 return f关于{topic}的最新新闻AI技术取得新突破。 tools [get_weather, get_latest_news] # --- 第3步定义提示词明确要求使用历史 --- prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以查询天气和新闻。请充分利用对话历史来理解上下文。 如果你需要调用工具请严格按照格式输出。 对话历史{chat_history} 当前问题{input} 你之前的步骤如果有{agent_scratchpad}), ]) # --- 第4步组装智能体 --- llm ChatOpenAI(modelgpt-4, temperature0) # 将工具绑定到LLM使其知道工具的描述 llm_with_tools llm.bind_tools(tools) # 创建智能体执行链 agent ( { input: lambda x: x[input], chat_history: lambda x: x[chat_history], # 将 intermediate_steps 格式化为适合放入提示词的消息 agent_scratchpad: lambda x: format_log_to_messages(x[intermediate_steps]), } | prompt | llm_with_tools | ReActSingleInputOutputParser() # 解析LLM输出为 AgentAction 或 AgentFinish ) # 现在agent 是一个Runnable它接收 AgentState 并输出 Action 或 Finish。 # 要运行它我们需要一个执行循环或使用LangGraph来编排。上面的代码定义了一个智能体的“大脑”agent。它接收一个包含input、chat_history、intermediate_steps的完整状态经过提示词填充、LLM推理、输出解析最终决定下一步做什么。为了运行这个智能体我们需要一个执行循环。在简单场景下可以手动编写from langchain_core.agents import AgentFinish, AgentAction def run_agent(user_input: str, current_state: AgentState) - AgentState: # 1. 调用智能体获取决策 decision agent.invoke(current_state) # 2. 处理决策 if isinstance(decision, AgentFinish): # 智能体决定结束返回最终答案 new_message AIMessage(contentdecision.return_values[output]) new_state { input: , # 本轮处理完毕 chat_history: current_state[chat_history] [HumanMessage(contentuser_input), new_message], intermediate_steps: current_state[intermediate_steps], } return new_state elif isinstance(decision, AgentAction): # 智能体决定调用工具 # 3. 执行工具 tool_to_use {t.name: t for t in tools}[decision.tool] observation tool_to_use.invoke(decision.tool_input) # 4. 更新状态将本次动作观察对加入 intermediate_steps new_intermediate_steps current_state[intermediate_steps] [(decision, observation)] new_state { input: user_input, # 用户问题仍未完全解决保持 chat_history: current_state[chat_history], intermediate_steps: new_intermediate_steps, } return new_state else: raise ValueError(f未知的决策类型{type(decision)}) # 初始化状态 state: AgentState { input: 北京天气怎么样然后告诉我一些科技新闻。, chat_history: [], intermediate_steps: [], } # 执行第一轮 print(用户, state[input]) new_state run_agent(state[input], state) # 假设第一轮决策是调用天气工具并更新了状态。 # 在真实循环中我们会检查 new_state 中的 intermediate_steps 是否包含未完成的动作 # 并继续用更新后的状态调用 agent直到返回 AgentFinish。这个手动循环清晰地展示了状态是如何一步步演进的。chat_history保存了完整的对话记录intermediate_steps保存了工具调用的轨迹。智能体的每一次“思考”都基于这个完整的、显式的状态。4.3 使用LangGraph进行专业级编排手动循环对于理解原理很好但对于生产环境太简陋。LangGraph库正是为了编排这种基于状态的、可能带有循环和分支的工作流而生的。createAgent产生的Runnable可以无缝集成到LangGraph中。from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from typing import Literal from langchain_core.agents import AgentFinish # 定义更丰富的状态类型供LangGraph使用 class GraphState(TypedDict): input: str chat_history: List[BaseMessage] intermediate_steps: List[tuple] # LangGraph需要知道最后输出的内容 final_output: str # 工具执行器 tool_executor ToolExecutor(tools) def should_continue(state: GraphState) - Literal[tools, __end__]: 根据agent的输出来决定下一步是调用工具还是结束。 decision agent.invoke(state) if isinstance(decision, AgentFinish): # 如果是最终结果存入状态并结束 return __end__ else: # 否则将动作存入状态并进入工具调用节点 # 注意这里需要把动作也传递下去一种方法是在状态里加字段或者用全局变量。 # 更规范的做法是使用LangGraph的“边”来传递数据。这里为简化我们假设能处理。 # 实际使用中通常会设计一个call_tool节点来处理。 return tools # 构建图的工作流此处为概念展示完整代码需定义节点和边 # 1. agent_node: 调用我们上面定义的agent输出决策。 # 2. tools_node: 如果决策是调用工具则执行工具并将结果追加到intermediate_steps。 # 3. 条件边从agent_node出来根据should_continue函数决定去tools_node还是END。 # 4. 从tools_node出来总是回到agent_node进行下一步思考。通过LangGraph我们可以图形化地定义智能体的工作流清晰地看到状态如何在“思考节点”和“工具节点”之间流转、更新。这比旧的、基于AgentExecutor的黑盒执行方式要透明和可控得多。5. 重构后的清晰世界经验、技巧与避坑指南经过这次范式重构我们在构建带记忆的AI应用时思路应该彻底转变。以下是我在实践中总结的核心经验和避坑点。5.1 新范式下的核心设计原则状态驱动数据流优先设计应用时首先定义你的State。思考清楚你的应用在运行过程中需要维护哪些信息用户输入、AI回复、工具调用记录、中间计算结果等。把这些信息都放进一个状态字典里。整个应用就是一个接收旧状态、返回新状态的函数或Runnable管道。存储与计算分离BaseChatMessageHistory及其实现如ChatMessageHistory,RedisChatMessageHistory只负责数据的持久化存储CRUD。而RunnableWithMessageHistory或你的自定义Runnable负责在“计算”前后从存储中读取状态、更新状态。这符合单一职责原则。会话隔离靠ID并发安全通过唯一的session_id或thread_id,conversation_id来保证。你的状态获取函数根据这个ID返回对应的历史存储对象。这是实现多用户、多对话并发的基石。拥抱LCEL和Runnable尽可能使用|操作符来组合你的逻辑。LCEL不仅让代码更简洁更重要的是它提供了强大的原语如RunnableLambda包装自定义函数、RunnableParallel并行执行、RunnableBranch条件分支让你能以声明式的方式构建复杂的数据流。5.2 常见陷阱与解决方案陷阱一状态定义过于庞大或模糊问题在State中定义了太多字段或者字段含义不清导致状态更新逻辑复杂且容易出错。解决方案遵循最小化原则。只定义工作流必需的字段。使用TypedDict提供类型提示。对于复杂的嵌套结构可以考虑使用Pydantic模型来定义State以获得更好的验证和IDE支持。陷阱二在错误的地方修改状态问题在某个Runnable的内部函数中直接修改了传入的状态字典产生了不可预料的副作用。解决方案牢记函数式编程的“不可变性”思想。一个Runnable应该根据输入计算输出而不是修改输入。状态的更新应该由专门的状态管理节点如RunnableWithMessageHistory或LangGraph中显式的tools_node来完成。在LCEL中你可以用RunnableLambda来创建纯函数式的状态转换。陷阱三历史消息格式不一致问题chat_history里混用了HumanMessage、AIMessage、SystemMessage但在提示词模板中却错误地引用导致LLM接收到的历史信息混乱。解决方案标准化你的消息格式。通常完整的对话历史就是List[BaseMessage]。在提示词中使用MessagesPlaceholder(variable_namechat_history)来整体注入。确保你保存到历史存储和从历史存储读取的格式是一致的。陷阱四工具调用历史intermediate_steps处理不当问题intermediate_steps的格式不符合format_log_to_messages或你自定义解析函数的期望导致无法正确地将工具调用历史转换成LLM能理解的提示词部分agent_scratchpad。解决方案intermediate_steps的标准格式是List[Tuple[AgentAction, str]]其中AgentAction包含工具名和输入str是工具执行后的观察结果。使用LangChain提供的format_log_to_messages、format_to_openai_function_messages等辅助函数来确保格式正确。如果自定义工具请确保其输出是字符串。陷阱五忽略状态序列化问题在Web应用中你需要将状态在多次HTTP请求间传递。如果状态中包含不可JSON序列化的对象如某些自定义类实例会导致错误。解决方案状态字典中的值应尽量使用基本类型str, int, list, dict或可序列化的Pydantic模型。对于像BaseMessage这样的对象LangChain通常提供了to_dict()和from_dict()方法。在持久化或网络传输前将状态序列化为JSON或字典使用时再反序列化。5.3 性能与扩展性考量历史存储后端的选择开发/轻量级使用内存字典ChatMessageHistory或本地SQLite通过SQLChatMessageHistory。生产环境使用RedisChatMessageHistory高性能支持TTL过期或PostgresChatMessageHistory关系型易于查询分析。选择取决于你的基础设施和访问模式。历史消息的裁剪与摘要即使使用了状态流过长的chat_history仍然会消耗大量Token。策略包括滑动窗口在get_session_history函数中返回时只返回最近N条消息。动态摘要维护一个“摘要”字段在状态中。当历史超过一定长度时调用一个LLM摘要链生成摘要然后用“摘要近期消息”作为新的历史。这比旧的ConversationSummaryMemory更可控因为摘要的触发和存储完全由你决定。向量检索与长期记忆对于需要从大量历史中检索相关片段的场景依然可以结合向量存储。实现方式是在你的状态流中加入一个“检索”节点。该节点根据当前input和chat_history去向量库查询相关文档并将查询结果作为一个新的字段如retrieved_docs加入到状态中供后续的LLM调用使用。这实现了长期记忆与短期对话记忆的分离与融合。从混乱的“Memory模块”到清晰的“State流”这不仅是LangChain API的一次升级更是构建复杂、可靠AI应用在方法论上的一次成熟。它要求开发者从“魔术般的效果”思维转向“精确的工程化”思维。理解并掌握这套基于LCEL和显式状态管理的新范式意味着你能真正驾驭AI应用的记忆与上下文构建出行为可预测、易于调试和维护的智能系统。这三年踩过的坑终将化为清晰道路上的基石。