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/api，API Key 在控制台的 API Keys 页面创建，Model ID 选你账号下可用的对话模型。这三件套后面会在每个代码片段里以BASE_URL、TOKEN、MODEL_NAME三个变量出现，你统一替换即可。
依赖安装分两块。基础记忆能力只需要 LangGraph 和 LangChain：
pip 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( model=MODEL_NAME, model_provider="openai", base_url=BASE_URL, api_key=TOKEN, temperature=0, )model_provider="openai"是关键，它告诉 LangChain 用 OpenAI 协议发请求，TaoToken 的兼容层会正确解析。如果你这里填错成别的 provider，最常见的报错是 401 或者model not found。
3. 可复制配置：短期记忆 Checkpointer 与长期记忆 Store
短期记忆的核心是 Checkpointer。它的作用是在图的每个 super-step 结束后，把当前状态快照存到一个 thread 里。你只要在调用时传入相同的thread_id，LangGraph 就会自动把历史状态加载回来。开发阶段用InMemorySaver最快，生产环境换成PostgresSaver。
先看内存版短期记忆的完整配置：
from langgraph.checkpoint.memory import InMemorySaver from langgraph.prebuilt import create_react_agent checkpointer = InMemorySaver() agent = create_react_agent( model=model, tools=[], checkpointer=checkpointer, ) 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_id="1"下，Agent 能记住“ada”。如果你把thread_id换成"2"再问同样的问题，它就会说不记得。这就是短期记忆的边界：它只在一个 thread 内有效。
生产环境把InMemorySaver换成PostgresSaver，配置片段如下：
from langgraph.checkpoint.postgres import PostgresSaver DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable" 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(checkpointer=checkpointer, store=store)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 的配置用MultiServerMCPClient，transport 选sse：
from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent search_client = MultiServerMCPClient( { "other_search": { "url": "你的MCP搜索服务地址", "headers": {"Authorization": f"Bearer {TOKEN}"}, "transport": "sse", } } ) search_tools = await search_client.get_tools() search_agent = create_react_agent( model, search_tools, name="search_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( model=model, tools=[book_hotel], store=store, name="hotel_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(content=result)]}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(checkpointer=checkpointer, store=store) 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], model=model, prompt=( "您是团队主管，负责管理信息搜索助手、酒店预订助手、以及用户信息查询助手。" "如需搜索信息，请交由 search_assistant 处理。" "如需预订酒店，请交由 hotel_assistant 处理。" "如需查询用户预订信息，请交由 booking_info_assistant 处理。" "注意，你每次只能调用一个助手agent！" ), ) supervisor = workflow.compile(checkpointer=checkpointer, store=store)验证请求分四步。第一步查火锅店，验证短期记忆：
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(f"Node: '{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(f"Node: '{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(resume="admin_123")恢复执行，查询助手会从 Store 里读到第三步写入的预订记录，返回给 Supervisor。整个链路跑通，说明短期记忆、长期记忆、MCP 工具链、Multi-Agent 协作全部生效。
5. 本篇常见错排查：401、local proxy failed 与 reading choices
跑这套代码时，报错基本集中在几个固定位置。下面按真实报错逐个拆。
401 Unauthorized。这个最常见，九成是 API Key 或 Base URL 填错。检查TOKEN是不是从 TaoToken 控制台复制的完整 Key，BASE_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_retries=0可以让错误更快暴露，方便定位。
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"。如果服务端要求特定 header，headers里要带全。工具列表为空时，先单独调一次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，后期迁移会很痛苦。
