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 默认存储历史消息列表messages；Checkpointer 负责把 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 端点是 https://taotoken.net/api 。
第一步，去控制台创建 API Key。打开 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，登录后在 API Keys 页面新建一个 Key，复制保存。这个 Key 就是后面.env里的TAOTOKEN_API_KEY。
第二步，确认你要用的模型 ID。在模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 可以看到当前可用的模型列表，记下你要用的 Model ID，比如gpt-4o-mini这类。Model ID 要原样填进init_chat_model的model参数。
第三步，写.env文件。在项目根目录创建.env，内容如下：
TAOTOKEN_API_KEY=sk-你的Key TAOTOKEN_BASE_URL=https://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(override=True) model = init_chat_model( model="gpt-4o-mini", model_provider="openai", api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.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(override=True) def get_model(model_id: str = "gpt-4o-mini"): return init_chat_model( model=model_id, model_provider="openai", api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL"), )这样后面所有脚本都from config import get_model，改模型只改一处。
3.2 内存持久化：InMemorySaver 快速验证
内存持久化适合快速测试和调试。核心是InMemorySaver+thread_id：
from 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( model=model, tools=[], checkpointer=checkpointer, ) config = {"configurable": {"thread_id": "1"}} # 第一轮 r1 = agent.invoke( {"messages": [HumanMessage("我叫张三")]}, config=config, ) print("第一轮:", r1["messages"][-1].content) # 第二轮，同一个 thread_id r2 = agent.invoke( {"messages": [HumanMessage("我叫什么？")]}, config=config, ) 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_URL=postgresql://langchain_user:你的密码@你的IP:5432/langchain_db?sslmode=disable然后写持久化脚本：
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( model=model, tools=[], checkpointer=checkpointer, ) config = {"configurable": {"thread_id": "1"}} r1 = agent.invoke( {"messages": [HumanMessage("你好，我是老张")]}, config=config, ) print("第一次:", r1["messages"][-1].content) r2 = agent.invoke( {"messages": [HumanMessage("你好，我是谁？")]}, config=config, ) 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(id=REMOVE_ALL_MESSAGES), first_msg, *recent, ] } agent = create_agent( model=model, middleware=[trim_messages], checkpointer=InMemorySaver(), )删除用@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(id=m.id) for m in messages[:to_delete]]} return None摘要用内置SummarizationMiddleware，把早期历史压缩成摘要，保留语义：
from langchain.agents.middleware import SummarizationMiddleware agent = create_agent( model=model, tools=[], checkpointer=InMemorySaver(), middleware=[ SummarizationMiddleware( model=get_model("gpt-4o-mini"), trigger=[("tokens", 100)], keep=("messages", 2), summary_prompt="对历史消息摘要，消息列表如下\n{messages}", ) ], )参数调优建议：模型上下文窗口 4k 时阈值设 3000，8k 设 6000，16k 设 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( model=get_model(), tools=[], checkpointer=InMemorySaver(), ) 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( model=get_model(), tools=[], checkpointer=InMemorySaver(), middleware=[ SummarizationMiddleware( model=get_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
这一节对照真实报错，逐个排查。
报错1：401 Unauthorized
openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}原因：TAOTOKEN_API_KEY没读到或复制不完整。排查：print(os.getenv("TAOTOKEN_API_KEY"))确认非空；检查.env是否被load_dotenv(override=True)加载；检查 Key 前后是否有空格。如果 Key 正确仍 401，确认base_url是https://taotoken.net/api，不要带 UTM 参数。
报错2：local proxy failed / Connection error
openai.APIConnectionError: Connection error.原因：base_url写错或网络不通。排查：确认TAOTOKEN_BASE_URL=https://taotoken.net/api，末尾无斜杠；用curl https://taotoken.net/api/models -H "Authorization: Bearer $TAOTOKEN_API_KEY"测试连通性。如果 curl 通但 Python 不通，检查是否有环境变量覆盖。
报错3：reading choices / KeyError 'choices'
KeyError: 'choices'原因：响应格式不符合 OpenAI 协议，通常是base_url指向了非兼容端点，或模型 ID 写错导致返回错误结构。排查：确认model_provider="openai"；确认 Model ID 在模型对话页面存在；打印原始响应resp.response_metadata看返回结构。
报错4：OAuth / 认证方式冲突
Error: OAuth authentication is not supported原因：某些 SDK 默认走 OAuth 流程，而 TaoToken 用 API Key 认证。排查：确保init_chat_model传了api_key参数，不要依赖环境变量OPENAI_API_KEY的自动读取，显式传参最稳。
报错5：记忆不生效
Agent 回答“我不知道你的名字”。排查清单：create_agent是否传了checkpointer；invoke是否传了config；两次调用的thread_id是否一致；InMemorySaver是否每次运行都新建（新建会丢历史，改用PostgresSaver）。
报错6：PostgresSaver 连接失败
psycopg.OperationalError: connection failed原因：DATABASE_URL格式错误或数据库未启动。排查：确认格式postgresql://user:pass@host:port/db?sslmode=disable；确认langgraph-checkpoint-postgres已安装；确认setup()已调用。
如果你在排查过程中需要确认模型是否可用，可以去模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite 直接测试；需要管理 Key 去 API Keys 页面 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ；接入细节参考文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
6. 长期编码与 Agent 场景的接入建议
短期记忆跑通后，如果你要做的是长期编码助手或复杂 Agent，建议把接入方式固定下来。核心是三件套：Base URL、Key、Model ID。
Base URL 固定为https://taotoken.net/api，Key 从控制台获取后写入.env，Model ID 按任务选——轻量摘要用便宜模型，主对话用能力强的模型。这样切换模型只改一个参数，记忆层和治理层代码完全不用动。
对于需要长期运行的编码 Agent，建议用PostgresSaver做持久化，配合SummarizationMiddleware做上下文治理。阈值设置参考：主模型上下文窗口的 75% 作为触发点，保留最近 2-4 条消息，摘要用便宜模型生成。这样既能控制成本，又能保证长会话的连贯性。
如果你要跑 Claude Code 这类编码工具，接入配置也是同样的三件套逻辑，Base URL 指向https://taotoken.net/api，Key 用 TaoToken 的 Key，Model ID 按工具要求填。具体接入步骤参考文档页面的对应章节。
最后给一个实用技巧：在开发阶段用InMemorySaver快速迭代，上线前切到PostgresSaver，切换时只需要改checkpointer的实例化那一行，create_agent和invoke的调用方式完全不变。这个抽象设计让记忆层的迁移成本极低，也是 LangChain 1.x 相比 0.x 的一大改进。
