1. 从一次“失忆”事故说起LangChain 短期记忆到底解决什么问题如果你用 LangChain 写过对话 Agent大概率遇到过这种尴尬第一轮告诉它“我叫张三”第二轮问“我叫什么”它一本正经地回答“我不知道你的名字”。这不是模型笨而是大模型本身是无状态的——每次agent.invoke()都是一次全新的开始它不会自动记住上一轮说过什么。LangChain 短期记忆Short-term Memory就是专门解决这个问题的组件。它的作用范围是单个对话线程Thread核心职责是「存储历史交互信息」和「在下次请求时把历史喂回模型」。你可以把它理解成 RPG 游戏里的自动存档你不需要手动保存系统在关键节点自动记录下次进入游戏时从上次的存档点继续。这篇内容适合三类人正在用 LangChain 1.x 搭多轮对话的开发者、被上下文窗口撑爆 token 的工程师、以及想把会话持久化到数据库但不知道从哪下手的同学。我会从零开始把「State Checkpointer Thread ID」这套短期记忆三件套讲透再结合 TaoToken 统一 Key/API 通道完成模型接入最后交付可复制的配置片段、上下文裁剪参数调优以及通过实际对话轮次验证记忆读写与治理效果。先说结论LangChain 1.x 的短期记忆 State会话内部状态 Checkpointer持久化机制 Thread ID会话作用域。State 默认存储历史消息列表messagesCheckpointer 负责把 State 作为检查点持久化保存Thread ID 用于唯一标识 State运行时按thread_id读写 State 快照。三者缺一不可少任何一个记忆都会失效。我试过只加checkpointer不加config结果 Agent 依然“失忆”排查了半天才发现是thread_id没传。所以下面每个环节我都会把「为什么」和「怎么验证」讲清楚避免你踩同样的坑。2. TaoToken 前置准备统一 Key 与 API 通道接入在写记忆代码之前先把模型接入通道打通。LangChain 的init_chat_model支持自定义base_url和api_key这意味着你可以用 TaoToken 的统一 Key 来调用不同厂商的模型切换模型时只改model参数不用动其他代码。TaoToken 的定位是统一模型接入通道一个 Key 走通多家模型兼容 OpenAI 风格的接口协议。对 LangChain 来说只要base_url指向兼容端点init_chat_model就能正常工作。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。第一步去控制台创建 API Key。打开 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 登录后在 API Keys 页面新建一个 Key复制保存。这个 Key 就是后面.env里的TAOTOKEN_API_KEY。第二步确认你要用的模型 ID。在模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 可以看到当前可用的模型列表记下你要用的 Model ID比如gpt-4o-mini这类。Model ID 要原样填进init_chat_model的model参数。第三步写.env文件。在项目根目录创建.env内容如下TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api注意base_url不要带末尾斜杠也不要加 UTM 参数保持干净的 API 端点。LangChain 内部会拼接/chat/completions路径多一个斜杠可能导致 404。第四步安装依赖。LangChain 1.x 的包结构和 0.x 差别较大建议用虚拟环境pip install langchain langchain-openai langgraph langgraph-checkpoint-postgres python-dotenv这里langgraph是必须的因为 LangChain 1.x 的 Agent 构建在 LangGraph 图结构之上短期记忆的 Checkpointer 也来自langgraph.checkpoint。langgraph-checkpoint-postgres是后面做持久化用的先装上。第五步验证接入。写一个最小脚本确认 Key 和 Base URL 能通from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os load_dotenv(overrideTrue) model init_chat_model( modelgpt-4o-mini, model_provideropenai, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), ) resp model.invoke(用一句话介绍你自己) print(resp.content)能打印出模型回复说明通道打通。如果报 401检查 Key 是否复制完整如果报连接错误检查base_url是否写成了https://taotoken.net/api。这一步过了再往下写记忆逻辑。3. 可复制配置State Checkpointer Thread ID 三件套这一节是核心我会给出完整的可复制配置片段包括环境变量、Agent 创建、内存持久化和 PostgreSQL 持久化两套方案。3.1 环境变量与模型初始化先建一个config.py把模型初始化抽出来复用# config.py from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os load_dotenv(overrideTrue) def get_model(model_id: str gpt-4o-mini): return init_chat_model( modelmodel_id, model_provideropenai, api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL), )这样后面所有脚本都from config import get_model改模型只改一处。3.2 内存持久化InMemorySaver 快速验证内存持久化适合快速测试和调试。核心是InMemorySaverthread_idfrom langchain.agents import create_agent from langchain.messages import HumanMessage from langgraph.checkpoint.memory import InMemorySaver from config import get_model model get_model() checkpointer InMemorySaver() agent create_agent( modelmodel, tools[], checkpointercheckpointer, ) config {configurable: {thread_id: 1}} # 第一轮 r1 agent.invoke( {messages: [HumanMessage(我叫张三)]}, configconfig, ) print(第一轮:, r1[messages][-1].content) # 第二轮同一个 thread_id r2 agent.invoke( {messages: [HumanMessage(我叫什么)]}, configconfig, ) print(第二轮:, r2[messages][-1].content)预期输出第二轮会回答“你叫张三”。如果回答“我不知道”检查三件事checkpointer是否传入create_agent、config是否传入invoke、两次调用的thread_id是否一致。thread_id是记忆管理的核心开关。同一个thread_id共享记忆不同thread_id完全隔离。你可以用user_alice、user_bob做多用户隔离也可以用task_coding、task_docs做同一用户的不同任务隔离。3.3 外部持久化PostgresSaver 生产级方案InMemorySaver只存在内存里进程结束数据就丢。生产环境要用外部存储这里选 PostgreSQL。先准备数据库连接串DATABASE_URLpostgresql://langchain_user:你的密码你的IP:5432/langchain_db?sslmodedisable然后写持久化脚本from langchain.agents import create_agent from langchain.messages import HumanMessage from langgraph.checkpoint.postgres import PostgresSaver from config import get_model import os model get_model() DB_URL os.getenv(DATABASE_URL) with PostgresSaver.from_conn_string(DB_URL) as checkpointer: checkpointer.setup() # 首次运行建表重复执行不会重建 agent create_agent( modelmodel, tools[], checkpointercheckpointer, ) config {configurable: {thread_id: 1}} r1 agent.invoke( {messages: [HumanMessage(你好我是老张)]}, configconfig, ) print(第一次:, r1[messages][-1].content) r2 agent.invoke( {messages: [HumanMessage(你好我是谁)]}, configconfig, ) print(第二次:, r2[messages][-1].content)setup()底层是Create IF Not Exists重复执行安全。执行后会创建四张表checkpoints主表存每个 thread 的检查点快照、checkpoint_blobs存不适合内联的复杂 channel 值、checkpoint_writes存中间写入、checkpoint_migrations迁移版本表。关键区别InMemorySaver每次运行新建 Saver 会丢历史PostgresSaver只要thread_id一致即使进程重启、重新创建 Saver历史状态依然能加载。这就是持久化的意义。3.4 记忆治理上下文裁剪参数调优历史消息无限增长会带来三个问题token 消耗增加、可能超过模型上下文窗口、响应变慢。治理策略有三种裁剪、删除、摘要。裁剪用before_model中间件在模型调用前裁剪消息列表from langchain.agents import create_agent, AgentState from langchain.agents.middleware import before_model from langchain.messages import RemoveMessage from langgraph.graph.message import REMOVE_ALL_MESSAGES from langgraph.checkpoint.memory import InMemorySaver from langgraph.runtime import Runtime from typing import Any from config import get_model model get_model() before_model def trim_messages(state: AgentState, runtime: Runtime) - dict[str, Any] | None: messages state[messages] if len(messages) 3: return None first_msg messages[0] recent messages[-3:] if len(messages) % 2 0 else messages[-4:] return { messages: [ RemoveMessage(idREMOVE_ALL_MESSAGES), first_msg, *recent, ] } agent create_agent( modelmodel, middleware[trim_messages], checkpointerInMemorySaver(), )删除用after_model在模型调用后永久移除旧消息from langchain.agents.middleware import after_model from langchain.messages import RemoveMessage after_model def delete_old_messages(state: AgentState, runtime: Runtime) - dict | None: messages state[messages] if len(messages) 5: to_delete len(messages) - 5 return {messages: [RemoveMessage(idm.id) for m in messages[:to_delete]]} return None摘要用内置SummarizationMiddleware把早期历史压缩成摘要保留语义from langchain.agents.middleware import SummarizationMiddleware agent create_agent( modelmodel, tools[], checkpointerInMemorySaver(), middleware[ SummarizationMiddleware( modelget_model(gpt-4o-mini), trigger[(tokens, 100)], keep(messages, 2), summary_prompt对历史消息摘要消息列表如下\n{messages}, ) ], )参数调优建议模型上下文窗口 4k 时阈值设 30008k 设 600016k 设 12000留余量给工具调用和系统提示。摘要触发频率要监控频繁触发就提高阈值从不触发就降低阈值。4. 验证请求用实际对话轮次确认记忆读写配置写完必须验证。我设计一个四轮对话测试覆盖记忆写入、读取、隔离和治理四个维度。4.1 基础记忆读写验证from langchain.agents import create_agent from langchain.messages import HumanMessage from langgraph.checkpoint.memory import InMemorySaver from config import get_model agent create_agent( modelget_model(), tools[], checkpointerInMemorySaver(), ) config {configurable: {thread_id: verify_1}} # 轮次1写入姓名 r1 agent.invoke({messages: [HumanMessage(我叫张三)]}, config) print(轮次1:, r1[messages][-1].content) # 轮次2读取姓名 r2 agent.invoke({messages: [HumanMessage(我叫什么)]}, config) print(轮次2:, r2[messages][-1].content) # 轮次3追问上一轮问题 r3 agent.invoke({messages: [HumanMessage(我刚才问了什么)]}, config) print(轮次3:, r3[messages][-1].content)预期轮次2 回答“张三”轮次3 回答“你刚才问了我叫什么”。如果轮次3 答不上来说明历史消息没有正确累积检查checkpointer是否生效。4.2 线程隔离验证config_a {configurable: {thread_id: user_alice}} config_b {configurable: {thread_id: user_bob}} agent.invoke({messages: [HumanMessage(我叫 Alice)]}, config_a) agent.invoke({messages: [HumanMessage(我叫 Bob)]}, config_b) ra agent.invoke({messages: [HumanMessage(我叫什么)]}, config_a) rb agent.invoke({messages: [HumanMessage(我叫什么)]}, config_b) print(Alice 线程:, ra[messages][-1].content) # 应回答 Alice print(Bob 线程:, rb[messages][-1].content) # 应回答 Bob两个线程互不干扰这是多用户场景的基础。4.3 查看 State 快照用agent.get_state(config)可以查看当前线程的完整状态from rich import print as rprint state agent.get_state(config) rprint(state.values[messages])你会看到messages列表里累积了所有 HumanMessage 和 AIMessage每条消息带id、content、usage_metadata等字段。usage_metadata里的input_tokens和output_tokens可以用来监控 token 消耗判断是否需要触发治理。4.4 治理效果验证用摘要中间件跑一段长对话观察 State 里的消息是否被压缩from langchain.agents.middleware import SummarizationMiddleware agent create_agent( modelget_model(), tools[], checkpointerInMemorySaver(), middleware[ SummarizationMiddleware( modelget_model(gpt-4o-mini), trigger[(tokens, 100)], keep(messages, 2), summary_prompt对历史消息摘要消息列表如下\n{messages}, ) ], ) config {configurable: {thread_id: summary_test}} long_text 我是一名工程师这里是一段很长的废话... * 20 agent.invoke({messages: [HumanMessage(long_text)]}, config) r agent.invoke({messages: [HumanMessage(请总结一下我的信息)]}, config) final_state agent.get_state(config) for msg in final_state.values[messages]: print(type(msg).__name__, :, str(msg.content)[:80])预期State 里第一条消息变成Here is a summary of the conversation to date: ...的摘要消息原始长文本被替换。这说明治理生效token 被压缩。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节对照真实报错逐个排查。报错1401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因TAOTOKEN_API_KEY没读到或复制不完整。排查print(os.getenv(TAOTOKEN_API_KEY))确认非空检查.env是否被load_dotenv(overrideTrue)加载检查 Key 前后是否有空格。如果 Key 正确仍 401确认base_url是https://taotoken.net/api不要带 UTM 参数。报错2local proxy failed / Connection erroropenai.APIConnectionError: Connection error.原因base_url写错或网络不通。排查确认TAOTOKEN_BASE_URLhttps://taotoken.net/api末尾无斜杠用curl https://taotoken.net/api/models -H Authorization: Bearer $TAOTOKEN_API_KEY测试连通性。如果 curl 通但 Python 不通检查是否有环境变量覆盖。报错3reading choices / KeyError choicesKeyError: choices原因响应格式不符合 OpenAI 协议通常是base_url指向了非兼容端点或模型 ID 写错导致返回错误结构。排查确认model_provideropenai确认 Model ID 在模型对话页面存在打印原始响应resp.response_metadata看返回结构。报错4OAuth / 认证方式冲突Error: OAuth authentication is not supported原因某些 SDK 默认走 OAuth 流程而 TaoToken 用 API Key 认证。排查确保init_chat_model传了api_key参数不要依赖环境变量OPENAI_API_KEY的自动读取显式传参最稳。报错5记忆不生效Agent 回答“我不知道你的名字”。排查清单create_agent是否传了checkpointerinvoke是否传了config两次调用的thread_id是否一致InMemorySaver是否每次运行都新建新建会丢历史改用PostgresSaver。报错6PostgresSaver 连接失败psycopg.OperationalError: connection failed原因DATABASE_URL格式错误或数据库未启动。排查确认格式postgresql://user:passhost:port/db?sslmodedisable确认langgraph-checkpoint-postgres已安装确认setup()已调用。如果你在排查过程中需要确认模型是否可用可以去模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 直接测试需要管理 Key 去 API Keys 页面 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 接入细节参考文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。6. 长期编码与 Agent 场景的接入建议短期记忆跑通后如果你要做的是长期编码助手或复杂 Agent建议把接入方式固定下来。核心是三件套Base URL、Key、Model ID。Base URL 固定为https://taotoken.net/apiKey 从控制台获取后写入.envModel ID 按任务选——轻量摘要用便宜模型主对话用能力强的模型。这样切换模型只改一个参数记忆层和治理层代码完全不用动。对于需要长期运行的编码 Agent建议用PostgresSaver做持久化配合SummarizationMiddleware做上下文治理。阈值设置参考主模型上下文窗口的 75% 作为触发点保留最近 2-4 条消息摘要用便宜模型生成。这样既能控制成本又能保证长会话的连贯性。如果你要跑 Claude Code 这类编码工具接入配置也是同样的三件套逻辑Base URL 指向https://taotoken.net/apiKey 用 TaoToken 的 KeyModel ID 按工具要求填。具体接入步骤参考文档页面的对应章节。最后给一个实用技巧在开发阶段用InMemorySaver快速迭代上线前切到PostgresSaver切换时只需要改checkpointer的实例化那一行create_agent和invoke的调用方式完全不变。这个抽象设计让记忆层的迁移成本极低也是 LangChain 1.x 相比 0.x 的一大改进。