1. 为什么你的 Agent 总是“失忆”从多轮任务断片说起很多人第一次用 LangGraph 搭 Agent 时都会遇到一个很尴尬的场景第一轮告诉它“我叫 Ada帮我查一下北京火锅店”它回答得挺好第二轮你接着问“那第一家店附近有什么酒店”它却像换了个人完全不知道“第一家店”指的是什么。这不是模型笨而是你没有给它装记忆。Agent 的“记忆”其实分两层。一层是短期记忆解决的是同一个对话线程里上下文别丢比如你刚说过的名字、刚推荐过的店、刚确认过的日期。另一层是长期记忆解决的是跨对话、跨会话的知识沉淀比如用户偏好、历史预订、身份信息。短期记忆靠 LangGraph 的 Checkpointer 把图状态按 thread 持久化长期记忆靠 Store 把结构化或非结构化数据按 namespace 存起来需要时再检索回上下文。这篇文章面向的是已经能跑通一个基础 ReAct Agent、但一遇到多轮任务就“断片”的开发者。我会带你从零把短期记忆、长期记忆、语义检索、MCP 工具链、Multi-Agent 协作这几块拼起来最后跑通一个带记忆的酒店预订 Demo。全程代码可复制模型调用统一走 TaoToken 的 OpenAI 兼容接口你只需要把 Base URL、API Key、Model ID 三件套填进去就能跑。先说结论让 AI 拥有记忆并不神秘核心就是“状态存哪里、什么时候读、什么时候写”。LangGraph 已经把这三件事抽象成了 Checkpointer 和 Store 两个组件你只要理解它们的边界剩下的就是按业务写节点。2. TaoToken 前置准备三件套配置与依赖安装在写记忆逻辑之前先把模型调用通道打通。TaoToken 提供 OpenAI 兼容的接口所以 LangChain 的init_chat_model可以直接用不需要额外写适配层。你需要准备三样东西Base URL、API Key、Model ID。Base URL 填https://taotoken.net/apiAPI Key 在控制台的 API Keys 页面创建Model ID 选你账号下可用的对话模型。这三件套后面会在每个代码片段里以BASE_URL、TOKEN、MODEL_NAME三个变量出现你统一替换即可。依赖安装分两块。基础记忆能力只需要 LangGraph 和 LangChainpip install -U langgraph langchain langchain-openai如果要上 PostgreSQL 做持久化再加两个包pip install -U psycopg[binary,pool] langgraph-checkpoint-postgres如果要接 MCP 工具链再加pip install langchain-mcp-adapters langgraph-supervisor如果你打算用 LangMem 做消息摘要还需要pip install -U langmem tiktoken这里有个容易踩的坑langgraph-checkpoint-postgres和langgraph.store.postgres是两个不同的模块前者管 Checkpointer后者管 Store但它们在同一个包里装一次就行。另外 psycopg 一定要带[binary,pool]否则连接池初始化会报ModuleNotFoundError。模型初始化统一写成这样后面所有示例都复用这个model对象from langchain.chat_models import init_chat_model BASE_URL https://taotoken.net/api TOKEN 你的API Key MODEL_NAME 你的Model ID model init_chat_model( modelMODEL_NAME, model_provideropenai, base_urlBASE_URL, api_keyTOKEN, temperature0, )model_provideropenai是关键它告诉 LangChain 用 OpenAI 协议发请求TaoToken 的兼容层会正确解析。如果你这里填错成别的 provider最常见的报错是 401 或者model not found。3. 可复制配置短期记忆 Checkpointer 与长期记忆 Store短期记忆的核心是 Checkpointer。它的作用是在图的每个 super-step 结束后把当前状态快照存到一个 thread 里。你只要在调用时传入相同的thread_idLangGraph 就会自动把历史状态加载回来。开发阶段用InMemorySaver最快生产环境换成PostgresSaver。先看内存版短期记忆的完整配置from langgraph.checkpoint.memory import InMemorySaver from langgraph.prebuilt import create_react_agent checkpointer InMemorySaver() agent create_react_agent( modelmodel, tools[], checkpointercheckpointer, ) config {configurable: {thread_id: 1}} response agent.invoke( {messages: [{role: user, content: 你好我叫ada}]}, config, ) print(response[messages][-1].content) response agent.invoke( {messages: [{role: user, content: 请问你还记得我叫什么名字么}]}, config, ) print(response[messages][-1].content)跑完你会看到同一个thread_id1下Agent 能记住“ada”。如果你把thread_id换成2再问同样的问题它就会说不记得。这就是短期记忆的边界它只在一个 thread 内有效。生产环境把InMemorySaver换成PostgresSaver配置片段如下from langgraph.checkpoint.postgres import PostgresSaver DB_URI postgresql://postgres:postgreslocalhost:5432/postgres?sslmodedisable with PostgresSaver.from_conn_string(DB_URI) as checkpointer: checkpointer.setup() # 用这个 checkpointer 编译图setup()只需要在第一次调用时执行它会自动建checkpoints、checkpoint_writes、checkpoint_blobs、checkpoint_migrations四张表。checkpoints存状态快照checkpoint_writes存真正的消息内容。程序重启后只要thread_id不变历史对话依然能读回来。长期记忆用 Store。它和 Checkpointer 最大的区别是Checkpointer 自动按 thread 存Store 需要你显式put和get而且可以跨 thread 共享。Store 的数据按 namespace 组织namespace 是个元组类似文件系统的文件夹。内存版 Store 配置from langgraph.store.memory import InMemoryStore store InMemoryStore() store.put( (users,), user_123, {name: ada, language: 中文}, )读取时用store.get((users,), user_123)返回的value就是那个字典。如果要让 Store 支持语义检索需要在初始化时传入 embedding 配置store InMemoryStore( index{ embed: custom_embeddings, dims: 2560, } )这里的custom_embeddings需要你自己实现一个继承langchain.embeddings.base.Embeddings的类把embed_documents和embed_query指向你的 embedding 服务。dims要和你的 embedding 模型输出维度一致填错会在写入时报维度不匹配。生产环境用PostgresStore配置和 Checkpointer 类似from langgraph.store.postgres import PostgresStore with PostgresStore.from_conn_string(DB_URI) as store: store.setup() # 用这个 store 编译图把 Checkpointer 和 Store 同时挂到图上Agent 就同时具备了短期和长期记忆能力graph builder.compile(checkpointercheckpointer, storestore)4. 验证请求跑通带记忆的 Multi-Agent 酒店预订案例这一节我们把前面所有组件拼起来做一个真实可跑的 Multi-Agent 案例。场景是用户查北京火锅店然后预订附近酒店最后管理员查询预订信息。整个系统有三个子 Agent用一个 Supervisor 协调。先初始化长期记忆和短期记忆from langgraph.store.memory import InMemoryStore from langgraph.checkpoint.memory import InMemorySaver store InMemoryStore() checkpointer InMemorySaver()搜索助手通过 MCP 接一个搜索工具。MCP 的配置用MultiServerMCPClienttransport 选ssefrom langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent search_client MultiServerMCPClient( { other_search: { url: 你的MCP搜索服务地址, headers: {Authorization: fBearer {TOKEN}}, transport: sse, } } ) search_tools await search_client.get_tools() search_agent create_react_agent( model, search_tools, namesearch_assistant, prompt你是一个能搜索各种信息的助手。, )酒店预订助手负责把预订信息写进 Store。注意这里用了store.put把用户预订记录按user_id存起来from typing import TypedDict from langchain_core.runnables import RunnableConfig class UserInfo(TypedDict): user_id: str hotel_name: str date: str num_guests: int def book_hotel(user_info: UserInfo, config: RunnableConfig): user_id config[configurable].get(user_id) namespace (user_bookings,) user_bookings store.get(namespace, user_id) or [] user_bookings.append(user_info) store.put(namespace, user_id, user_bookings) return f成功为用户 {user_id} 预订了 {user_info[hotel_name]} book_hotel_agent create_react_agent( modelmodel, tools[book_hotel], storestore, namehotel_assistant, prompt你是一个酒店预订助手请直接预订。, )查询助手需要人工介入验证管理员身份用interrupt实现中断from langgraph.types import interrupt from langchain_core.messages import AIMessage def authentication_and_query_node(state, config): admin_input interrupt(请输入管理员id如需退出查询请输入exit) if admin_input exit: result 用户已退出查询。 elif admin_input admin_123: result query_booking_from_store(config) else: result f没有权限查询admin_id 不匹配 (输入为: {admin_input}) return {messages: [AIMessage(contentresult)]}query_booking_from_store从 Store 里读预订记录from langgraph.config import get_store def query_booking_from_store(config: RunnableConfig) - str: store get_store() user_id config[configurable].get(user_id) booking_info store.get((user_bookings,), user_id) if booking_info and booking_info.value: return f已找到预订信息{str(booking_info.value)} return 未找到该用户的预订信息把查询节点编译成子图并给它命名Supervisor 才能调用from langgraph.graph import StateGraph, END query_workflow StateGraph(SubgraphState) query_workflow.add_node(auth_and_query, authentication_and_query_node) query_workflow.set_entry_point(auth_and_query) query_workflow.add_edge(auth_and_query, END) booking_query_subgraph query_workflow.compile(checkpointercheckpointer, storestore) booking_query_subgraph.name booking_info_assistant最后用 Supervisor 把三个 Agent 串起来from langgraph_supervisor import create_supervisor workflow create_supervisor( [search_agent, book_hotel_agent, booking_query_subgraph], modelmodel, prompt( 您是团队主管负责管理信息搜索助手、酒店预订助手、以及用户信息查询助手。 如需搜索信息请交由 search_assistant 处理。 如需预订酒店请交由 hotel_assistant 处理。 如需查询用户预订信息请交由 booking_info_assistant 处理。 注意你每次只能调用一个助手agent ), ) supervisor workflow.compile(checkpointercheckpointer, storestore)验证请求分四步。第一步查火锅店验证短期记忆config {configurable: {thread_id: 1, user_id: user_123}} async for chunk in supervisor.astream( {messages: [(user, 北京最出名的老北京火锅是哪家)]}, config, ): for key, value in chunk.items(): print(fNode: {key}) if value: print(value)第二步接着问“那第一个推荐的火锅店附近有哪些酒店”Supervisor 能记住上文提到的第一家店这就是短期记忆在起作用。第三步预订酒店async for chunk in supervisor.astream( {messages: [(user, 帮我预订北京王府井希尔顿酒店日期2025-11-13到2025-11-14入住人数1)]}, config, ): for key, value in chunk.items(): print(fNode: {key}) if value: print(value)第四步换一个 thread 做管理员查询触发中断config {configurable: {thread_id: 2, user_id: user_123}} async for chunk in supervisor.astream( {messages: [(user, 查询用户预订酒店信息)]}, config, ): for key, value in chunk.items(): if key __interrupt__: print(f中断信息: {value[0].value}) break收到中断后用Command(resumeadmin_123)恢复执行查询助手会从 Store 里读到第三步写入的预订记录返回给 Supervisor。整个链路跑通说明短期记忆、长期记忆、MCP 工具链、Multi-Agent 协作全部生效。5. 本篇常见错排查401、local proxy failed 与 reading choices跑这套代码时报错基本集中在几个固定位置。下面按真实报错逐个拆。401 Unauthorized。这个最常见九成是 API Key 或 Base URL 填错。检查TOKEN是不是从 TaoToken 控制台复制的完整 KeyBASE_URL是不是https://taotoken.net/api注意结尾不要多斜杠。如果 Key 没问题检查model_provider是不是openai。还有一种情况是 Key 有额度但模型 ID 写错报错信息里会带model not found这时候去控制台确认 Model ID 拼写。local proxy failed / connection refused。这个报错通常出现在你本地起了代理但没关或者环境变量里残留了HTTP_PROXY、HTTPS_PROXY。LangChain 发请求时会读这些环境变量导致请求被转发到一个不存在的本地端口。解决办法是清掉这些环境变量或者在代码里显式指定http_client不走代理。另外 PostgreSQL 连接报connection refused是另一回事检查DB_URI里的 host、port、用户名密码以及 postgres 服务是否启动。reading choices of undefined。这个报错说明接口返回的 JSON 结构里没有choices字段通常是服务端返回了错误信息但被当成正常响应解析。打印原始响应就能看到真实原因多数是 401 或 429。在init_chat_model里加max_retries0可以让错误更快暴露方便定位。OAuth / token expired。如果你用的是需要 OAuth 的模型服务token 过期会报这个。TaoToken 的 API Key 是长期有效的一般不会遇到。如果遇到重新在控制台生成一个 Key 替换即可。interrupt 恢复后状态丢失。这个坑在于Command(resume...)必须用同一个thread_id和同一个 checkpointer 实例。如果你在恢复时新建了 checkpointer或者换了 thread_id中断前的状态就找不回来了。另外interrupt只能在有 checkpointer 的图里用没挂 checkpointer 的图调用interrupt会直接报错。Store 语义检索返回空。检查dims是否和 embedding 模型输出维度一致检查embed_documents里对文本的预处理是否和embed_query一致。如果写入时文本被eval处理过查询时也要做同样处理否则向量空间对不上。MCP 工具列表为空。MultiServerMCPClient的url必须是完整的 SSE 端点transport必须是sse。如果服务端要求特定 headerheaders里要带全。工具列表为空时先单独调一次get_tools()打印结果确认连接通了再往下走。6. 语义一致 CTA把记忆能力接到你的真实项目跑通上面的 Demo 之后你会发现记忆机制本身并不复杂难的是把它接到你现有的业务里。我的建议是先从短期记忆入手给你的 Agent 挂上 Checkpointer把thread_id和你的会话 ID 绑定这一步能解决 80% 的多轮断片问题。然后再根据业务需要把用户偏好、历史操作这类跨会话数据迁到 Store 里。模型调用这块TaoToken 的 OpenAI 兼容接口可以直接替换init_chat_model里的三件套不需要改其他代码。如果你还在选模型或者想先验证一下对话效果可以去模型对话页面直接试要长期跑编码类 Agent 或者多轮任务Coding Plan 的额度更适合持续调用API Key 在控制台的 API Keys 页面管理接入细节看接入文档。把 Base URL、Key、Model ID 三件套配好上面所有代码都能直接跑。最后留一个实用技巧Store 的 namespace 设计要提前想清楚。我习惯用(memories, user_id)存用户记忆用(user_bookings, user_id)存业务数据用(preferences, user_id)存偏好。namespace 分得越细后面做语义检索时越不容易串数据。如果你一开始就把所有东西塞进一个 namespace后期迁移会很痛苦。