LangChain：短期记忆的访问方式

使用 Tools(工具) 访问和修改智能体的短期记忆。可以设置在模型调用前，或调用后访问记忆。

访问记忆

可以通过多种方式访问和修改智能体的短期记忆（状态）

访问方式	说明	主要用途
工具（Tools）	在工具函数中通过 ToolRuntime 读取或修改状态	工具需要访问用户信息 或持久化中间结果
提示（Prompt）	在中间件中访问状态， 构建动态提示词	根据对话历史或自定义状态 动态生成系统提示
Before Model	在模型调用前访问和修改状态	消息预处理、修剪、添加上下文
After Model	在模型调用后访问和修改状态	响应验证、敏感词过滤、后处理

工具：读取短期记忆

在工具中通过 runtime 参数（类型为 ToolRuntime）访问短期记忆。runtime 参数对工具签名是隐藏的（模型看不到它），但工具可以通过它访问状态。

# 导入所需模块
from langchain.agents import create_agent, AgentState
from langchain.tools import tool, ToolRuntime

# 定义自定义状态，包含user_id字段
class CustomState(AgentState):
    user_id: str

# 使用tool装饰器定义工具函数
@tool
def get_user_info(runtime: ToolRuntime) -> str:
    """
    查找用户信息。
    runtime参数是隐式传递的，不会被模型感知。
    """
    # 从运行时状态中获取user_id
    user_id = runtime.state["user_id"]
    # 根据user_id返回对应的用户信息
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

# 创建智能体，配置自定义状态
agent = create_agent(
    model="gpt-5-nano",           # 指定的模型
    tools=[get_user_info],        # 工具列表
    state_schema=CustomState,     # 自定义状态模式
)

# 调用智能体，传入user_id
result = agent.invoke({
    "messages": "look up user information",  # 用户消息
    "user_id": "user_123"                    # 自定义状态字段
})

# 打印智能体的响应内容
print(result["messages"][-1].content)  # 输出: User is John Smith.

代码说明：ToolRuntime是LangChain提供的运行时对象，它封装了当前执行的上下文信息。工具可以通过runtime.state访问智能体的当前状态，从而读取自定义字段。这种设计的优点是模型不需要知道状态的存在，工具可以“悄悄地”获取需要的信息。

• @tool 装饰器将普通函数转换为 LangChain 工具，可供代理调用。
• ToolRuntime 类型的 runtime 参数不会暴露给 LLM，但工具函数可以从中读取状态信息。
• 这种方式让工具能够访问对话上下文，而无需每次都将状态作为参数传给工具。

工具：写入短期记忆

在执行过程中修改代理的短期记忆（状态），可以直接从工具中返回状态更新。这对于持久化中间结果或使信息可供后续工具或提示访问非常有用。

from langchain.tools import tool, ToolRuntime
from langchain.agents import create_agent, AgentState
from langgraph.types import Command
from pydantic import BaseModel
from langchain.messages import ToolMessage

# 定义自定义状态（包含用户姓名）
class CustomState(AgentState):
    user_name: str

# 定义自定义上下文（包含用户ID）
class CustomContext(BaseModel):
    user_id: str

# 定义工具：查找并更新用户信息
@tool
def update_user_info(runtime: ToolRuntime[CustomContext, CustomState]) -> Command:
    """
    查找并更新用户信息。
    使用Command返回状态更新。
    """
    # 从上下文中获取用户ID
    user_id = runtime.context.user_id
    # 根据用户ID确定用户名
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    
    # 返回Command对象，同时更新状态和消息历史
    return Command(update={
        "user_name": name,      # 更新状态中的user_name字段
        "messages": [           # 同时添加工具响应消息
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=runtime.tool_call_id
            )
        ]
    })

# 定义工具：问候用户
@tool
def greet(runtime: ToolRuntime[CustomContext, CustomState]) -> str | Command:
    """
    在找到用户信息后用于问候用户。
    """
    # 从运行时状态中获取用户名
    user_name = runtime.state.get("user_name", None)
    
    if user_name is None:
        # 如果用户名尚未获取，返回Command要求先获取
        return Command(
            update={
                "messages": [
                    ToolMessage(
                        "Call update_user_info first to get and update the user's name.",
                        tool_call_id=runtime.tool_call_id
                    )
                ]
            }
        )
    
    # 如果已有用户名，返回问候语
    return f"Hello {user_name}!"

# 创建智能体
agent = create_agent(
    model="gpt-5-nano",                     # 指定的模型
    tools=[update_user_info, greet],        # 两个工具
    state_schema=CustomState,               # 自定义状态模式
    context_schema=CustomContext,           # 自定义上下文模式
)

# 调用智能体，传入自定义上下文
agent.invoke(
    {"messages": [{"role": "user", "content": "greet the user"}]},
    context=CustomContext(user_id="user_123"),
)

代码说明：通过Command(update={...})可以同时更新状态和消息历史。ToolRuntime是一个泛型类，ToolRuntime[CustomContext, CustomState]表示运行时包含指定类型的上下文和状态，这样可以获得更好的类型提示和代码补全支持。两个工具的配合展示了状态如何在工具之间共享：update_user_info写入user_name，greet读取它。

• Command(update={...}) 是 LangGraph 中用于返回状态更新的对象，可以同时修改状态的多个字段。
• ToolRuntime[CustomContext, CustomState] 使用了泛型类型注解，分别指定上下文和状态的类型。
• 这种模式允许工具链中的后续工具访问先前工具产生的结果，实现更复杂的工作流。

提示：访问短期记忆

在中间件（middleware）中访问短期记忆，可以根据对话历史或自定义状态字段创建动态提示词。

from langchain.agents import create_agent
from typing import TypedDict
from langchain.agents.middleware import dynamic_prompt, ModelRequest

# 定义自定义上下文类型
class CustomContext(TypedDict):
    user_name: str

# 定义一个示例工具函数
def get_weather(city: str) -> str:
    """获取城市的天气信息。"""
    return f"The weather in {city} is always sunny!"

# 使用dynamic_prompt装饰器创建动态系统提示
@dynamic_prompt
def dynamic_system_prompt(request: ModelRequest) -> str:
    """
    根据运行时上下文动态生成系统提示。
    模型每次调用前会执行此函数以获取当前提示。
    """
    # 从运行时上下文中获取用户名
    user_name = request.runtime.context["user_name"]
    # 构建个性化的系统提示
    system_prompt = f"You are a helpful assistant. Address the user as {user_name}."
    return system_prompt

# 创建智能体
agent = create_agent(
    model="gpt-5-nano",                       # 指定的模型
    tools=[get_weather],                      # 工具列表
    middleware=[dynamic_system_prompt],       # 动态提示中间件
    context_schema=CustomContext,             # 自定义上下文模式
)

# 调用智能体，传入用户名上下文
result = agent.invoke(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    context=CustomContext(user_name="John Smith"),
)

# 打印完整的消息历史
for msg in result["messages"]:
    msg.pretty_print()

输出：

================================ Human Message =================================

What is the weather in SF?
================================== Ai Message ==================================
Tool Calls:
  get_weather (call_WFQlOGn4b2yoJrv7cih342FG)
 Call ID: call_WFQlOGn4b2yoJrv7cih342FG
  Args:
    city: San Francisco
================================= Tool Message =================================
Name: get_weather

The weather in San Francisco is always sunny!
================================== Ai Message ==================================

Hi John Smith, the weather in San Francisco is always sunny!

代码说明：@dynamic_prompt装饰器允许在模型调用前动态生成系统提示。此处根据用户名生成个性化问候，使模型能够自然地以用户的名字称呼对方。运行时上下文中的user_name被嵌入到系统提示中，而不是作为普通消息发送，这样用户名字就被固定在系统提示中，不会被后续对话覆盖。

• @dynamic_prompt 装饰器允许在模型调用前动态生成系统提示。
• 生成的提示词会追加到原始系统提示之后，用于指导模型行为。
• 这种方式可以根据对话状态或外部上下文动态调整模型的行为，比如根据用户名个性化回复。

Before Model：模型调用前访问记忆

在 @before_model 中间件中访问短期记忆，在模型调用之前处理消息：

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
from typing import Any

# 使用before_model装饰器定义消息修剪中间件
@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """
    仅保留最后几条消息以适应上下文窗口。
    该函数在模型调用前执行，确保输入不超限。
    """
    messages = state["messages"]
    
    # 消息数量不超过3条时无需修改
    if len(messages) <= 3:
        return None
    
    # 保留第一条消息和最新消息
    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages
    
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 移除所有现有消息
            *new_messages                           # 添加新消息列表
        ]
    }

# 创建智能体
agent = create_agent(
    "gpt-5-nano",                          # 指定的模型
    tools=[],                               # 工具列表
    middleware=[trim_messages],             # 修剪中间件
    checkpointer=InMemorySaver(),           # 内存检查点器
)

# 配置线程ID
config: RunnableConfig = {"configurable": {"thread_id": "1"}}

# 多轮对话示例
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)

# 验证智能体是否还记得名字
final_response = agent.invoke({"messages": "what's my name?"}, config)
final_response["messages"][-1].pretty_print()

代码说明：@before_model中间件在模型被调用之前执行，最适合进行消息修剪、内容过滤或提示增强等操作，因为这些修改直接影响模型接收到的输入。此示例的修剪策略与之前相同：保留第一条消息和最新的几条消息。

After Model：模型调用后访问记忆

在 @after_model 中间件中访问短期记忆，在模型调用之后处理消息：

from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langgraph.runtime import Runtime

# 定义 after_model 中间件
@after_model
def validate_response(state: AgentState, runtime: Runtime) -> dict | None:
    """移除包含敏感词的响应消息"""
    STOP_WORDS = ["password", "secret"]   # 敏感词列表
    last_message = state["messages"][-1]

    # 检查最后一条消息是否包含敏感词
    if any(word in last_message.content for word in STOP_WORDS):
        return {"messages": [RemoveMessage(id=last_message.id)]}  # 删除该消息
    return None

# 创建代理
agent = create_agent(
    model="gpt-5-nano",
    tools=[],
    middleware=[validate_response],   # 添加响应验证中间件
    checkpointer=InMemorySaver(),
)

代码说明：

• @before_model 中间件在模型调用前执行，适合做消息预处理、修剪、添加上下文等操作。
• @after_model 中间件在模型调用后执行，适合做响应验证、敏感词过滤、日志记录等操作。
• 两者都可以访问和修改代理的状态，但执行时机不同，应根据需求选择合适的中间件类型。

总结和建议

短期记忆 vs 长期记忆

维度	短期记忆	长期记忆
作用范围	单个线程/对话内	跨多个线程和会话
生命周期	对话结束即消亡	持久化存储，永久保留
典型内容	对话历史、临时上下文	用户偏好、知识库、配置文件
实现方式	检查点器（内存/数据库）	向量数据库、知识图谱、外部存储

如果需要在不同对话之间记住用户特定或应用程序级别的数据，应使用长期记忆（Long-term Memory）。

上下文窗口管理策略对比

策略	优点	缺点	适用场景
修剪（Trim）	简单高效，成本低	可能丢失重要信息	对话以最近内容为主， 旧消息相关性低
删除（Delete）	精准控制	需要人工决定删除哪些消息	需要移除敏感内容 或重置对话
摘要（Summarize）	保留语义信息， 压缩效果好	需要额外的 LLM 调用， 成本较高	对话内容复杂， 需要长期保持上下文语义

生产环境建议

1. 选择合适的检查点器：开发阶段使用 InMemorySaver 便于调试；生产环境务必使用 PostgresSaver 等数据库后端，确保状态持久化。
2. 合理设置线程 ID：为每个独立的对话会话分配唯一的 thread_id，可以轻松实现多会话隔离和管理。
3. 主动管理上下文窗口：对于长对话场景，应结合修剪或摘要策略，主动控制上下文大小，避免超出 LLM 的限制。
4. 使用轻量级模型生成摘要：如果采用摘要策略，建议使用更小、更便宜的模型（如 gpt-5.4-mini）来生成摘要，以降低成本。
5. 工具中更新状态需谨慎：通过工具修改状态时，注意不要意外删除或覆盖重要的对话历史。