Agent 短期记忆

Agent 在执行多轮对话时，需要记住之前的上下文。通过 Checkpointer 机制，Agent 可以实现短期记忆（Memory），将对话历史或状态保存在内存或数据库中。

create_agent API 文档

langgraph.checkpoint API 文档

短期记忆实现

通过给 create_agent 传入 checkpointer 参数，可以使 Agent 具备短期记忆能力。调用时需要传入带有 thread_id 的 config，以便区分不同用户的对话线程。

1. 内存存储 (MemorySaver)

使用 MemorySaver 可以将记忆保存在内存中，适合测试或无需持久化的场景。

from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI

# 1. 定义模型
model = ChatOpenAI(model="gpt-4o")

# 2. 创建基于内存的 Checkpointer
memory = MemorySaver()

# 3. 创建带有短期记忆的 Agent
agent = create_agent(
    model,
    tools=[],
    checkpointer=memory
)

# 4. 调用时传入配置 (区分不同对话线程)
config = {"configurable": {"thread_id": "1"}}

# 第一轮对话
response1 = agent.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫 Alice"}]},
    config=config
)
print(response1["messages"][-1].content)

# 第二轮对话 (Agent 会记得之前的内容)
response2 = agent.invoke(
    {"messages": [{"role": "user", "content": "我的名字是什么？"}]},
    config=config
)
print(response2["messages"][-1].content) # 输出: 你的名字是 Alice

2. 持久化存储 (PostgresSaver)

在生产环境中，我们需要将记忆持久化。可以使用 PostgresSaver 将状态保存在 PostgreSQL 数据库中。 首先需要安装依赖：

pip install langgraph-checkpoint-postgres psycopg
# 也可以选择安装 psycopg[binary] 或 psycopg[c]

PostgresSaver API 文档

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_openai import ChatOpenAI

# 数据库连接 URI
DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"

# 使用上下文管理器连接并初始化数据库
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 首次使用时需调用 setup() 创建相关表
    checkpointer.setup()
    
    model = ChatOpenAI(model="gpt-4o")
    
    agent = create_agent(
        model,
        tools=[],
        checkpointer=checkpointer
    )
    
    config = {"configurable": {"thread_id": "user_123"}}
    
    response = agent.invoke(
        {"messages": [{"role": "user", "content": "记住我最喜欢的颜色是蓝色"}]},
        config=config
    )

**注意：**如果手动创建数据库连接并传给 PostgresSaver，必须设置 autocommit=True 和 row_factory=dict_row（from psycopg.rows import dict_row），否则会导致表无法正确创建或抛出字典访问异常。

Token 优化方案

随着对话的进行，messages 列表会越来越长，可能超出模型 Token 限制或增加调用成本。需要对长对话上下文进行裁剪或摘要。

1. 消息裁剪 (trim_messages)

trim_messages 是一种快速且简单的方案。它可以根据最大 Token 数或消息数量截断旧消息，同时保证保留 System Prompt。

trim_messages API 文档

from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")

# 假设已有历史消息 messages

# 裁剪历史记录，只保留最后的 45 个 token，同时保留 system 提示词
trimmed_messages = trim_messages(
    messages,
    max_tokens=45,
    strategy="last",               # 保留最新的消息
    token_counter=model,           # 使用模型计算 token
    start_on="human",              # 确保截断后的历史记录以 HumanMessage 开头
    include_system=True,           # 始终保留首个 SystemMessage
    allow_partial=False            # 不允许截断单条消息的部分内容
)

**优化技巧：**为了在调用频繁的地方获得更快的速度，可以将 token_counter 设为 "approximate" 以使用近似计算，或者设为 len 直接按消息条数来进行裁剪。

2. 对话摘要 (SummarizationMiddleware)

除了直接抛弃旧消息，更好的方式是使用中间件在 Token 或消息数达到阈值时，自动将旧消息总结为一段摘要。这样可以在大幅节省 Token 的同时，让模型依然了解之前的沟通内容。

SummarizationMiddleware API 文档

from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver
from langchain.agents.middleware import SummarizationMiddleware
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o")
memory = MemorySaver()

agent = create_agent(
    model=model,
    tools=[],
    checkpointer=memory,
    middleware=[
        SummarizationMiddleware(
            model=model,                 # 负责生成摘要的语言模型
            trigger=("messages", 3),     # 触发条件：当消息数量达到 3 条时进行摘要
            keep=("messages", 2)         # 保留策略：保留最新的 2 条消息原样，更早的压缩进摘要
        )
    ]
)

# 调用时同上，中间件会在后台自动接管消息摘要
config = {"configurable": {"thread_id": "1"}}
# agent.invoke(...)
最近更新:
Contributors: kingmusi