让Agent“记住”与“可控”：记忆管理与中间件实战指南

爱昏羔

319人浏览 · 2026-07-10 19:31:07

爱昏羔 · 2026-07-10 19:31:07 发布

让Agent“记住”与“可控”：记忆管理与中间件实战指南

在前两篇博客中，我们为Agent赋予了工具调用能力，并通过MCP解决了工具接入的碎片化问题。一个能调用外部工具、自主解决问题的Agent已经初见雏形。

然而，当我们将Agent投入真实应用场景时，两个问题会立刻浮出水面：

失忆症：Agent处理完一次请求后，就像金鱼一样，完全不记得刚才聊了什么。在一场多轮对话中，它会对你说过的信息“左耳进右耳出”。
失控感：Agent可能在一轮循环中生成数百条消息，或是不假思索地调用高风险的“删库”操作，缺乏有效的监管。

记忆管理和中间件正是LangChain生态为解决这两个核心痛点而设计的“良药”。本文将深入探讨如何让Agent拥有健壮的“短期记忆”，以及如何通过中间件实现对Agent执行过程的精细控制。

1. 记忆管理：让Agent告别“失忆症”

在LangChain中，Agent的记忆通过持久化（Persistence） 机制实现，其核心思想是将Agent在某次对话中的状态（State） 保存下来，以便在后续对话中恢复。

1.1 “失忆”问题：为什么Agent会忘记？

默认情况下，每次调用agent.invoke()都是一个独立的、无状态的执行单元。当这次调用结束时，Agent的所有内部状态（如消息历史、中间步骤等）都会被销毁。因此，在下一次调用中，它是一个全新的、没有任何“过去”的Agent。这种无状态特性在简单的问答中没有问题，但在构建复杂的、需要上下文理解的应用时，就成为了致命的短板。

1.2 Checkpointer：记忆的“存档点”

LangGraph提出了Checkpointer（检查点保存器） 的概念。它像一个游戏中的自动存档点，在Agent执行的每一步之后，自动将当前的状态（包括所有消息、待办事项、临时变量等）保存下来。当Agent再次被调用时，Checkpointer会加载之前保存的“存档”，让Agent从上次结束的地方无缝继续。

Checkpointer有不同的实现，以适应开发和生产环境。

InMemorySaver: 用于开发和测试的内存型存储。Agent状态保存在RAM中，一旦进程重启，所有记忆将丢失。
SqliteSaver: 用于本地开发的文件型存储。状态保存在本地的SQLite数据库中，进程重启后记忆依然存在。
PostgresSaver: 生产环境推荐。使用PostgreSQL数据库存储状态，支持高并发和水平扩展，是构建生产级应用的基石。AWS用户也可使用AgentCoreMemorySaver集成Bedrock服务。

1.3 Thread ID：记忆的“房间号”

有了Checkpointer这个“保险箱”，我们还需要一把钥匙来区分不同用户的“记忆”。这把钥匙就是 thread_id （线程ID）。

想象一下，每个thread_id都代表一个独立的对话“房间”。同一个用户（或会话）使用相同的thread_id，Agent就能加载这个房间里的历史记忆；不同用户使用不同的thread_id，他们的记忆就完全隔离，互不干扰。

在代码中，我们将thread_id作为配置参数传入：

config = {"configurable": {"thread_id": "user-session-123"}}
# 第一次对话，Agent会创建并记住“user-session-123”这个房间的状态
agent.invoke({"messages": [HumanMessage(content="你好，我叫小明")]}, config=config)

# 第二次对话，通过相同的thread_id，Agent能回忆起之前的对话
agent.invoke({"messages": [HumanMessage(content="我叫什么名字？")]}, config=config)
# Agent会回答：“你叫小明”

1.4 记忆的局限：Checkpointer不是万能的

Checkpointer提供了强大的短期记忆（Short-term Memory），但它也有其局限性。

它是线程范围的：所有记忆都依附于特定的thread_id。如果用户开启了一个全新的会话（生成了新的thread_id），Agent将无法访问旧会话的任何信息。对于需要跨会话记忆的长期记忆（如用户偏好、历史事实），LangGraph提供了StoreAPI作为补充。
存储空间会增长：随着对话不断进行，保存的状态会不断累积，可能导致存储空间膨胀和性能下降。生产环境中需要配置合理的清理策略（TTL）或定期归档。
是“快照”，不是“流”：Checkpointer保存的是状态快照，适合回溯和恢复，但不适合用于实时协作场景。

1.5 完整示例：为Agent装上记忆

下面是一个完整的代码示例，演示了如何使用InMemorySaver和thread_id让Agent具备记忆能力。

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 1. 初始化模型
model = ChatOpenAI(model="gpt-4o")

# 2. 创建Checkpointer
checkpointer = InMemorySaver()

# 3. 创建Agent，编译时传入checkpointer
agent = create_react_agent(model, tools=[], checkpointer=checkpointer)

# 4. 定义一个thread_id
config = {"configurable": {"thread_id": "conversation-001"}}

# 5. 第一轮对话
resp1 = agent.invoke(
    {"messages": [HumanMessage(content="你好，我叫Alice，我住在伦敦。")]},
    config=config
)
print(resp1["messages"][-1].content) 
# 输出: 你好Alice，很高兴认识你！

# 6. 第二轮对话，使用相同的thread_id
resp2 = agent.invoke(
    {"messages": [HumanMessage(content="我在哪个城市？")]},
    config=config
)
print(resp2["messages"][-1].content) 
# 输出: 根据你之前说的，你住在伦敦。

注意：如果是生产环境，将InMemorySaver替换为PostgresSaver即可实现持久化。

2. Agent中间件：为“大脑”装上“控制面板”

记忆管理解决了Agent的“失忆”问题，让我们能够构建流畅的对话体验。中间件则解决了另一个核心问题：如何让Agent的行为更可控、更安全、更高效。

2.1 概念：Agent执行流水线上的“过滤器”

中间件（Middleware）是一种强大的设计模式，它在Agent的执行循环（模型调用 -> 工具执行 -> 模型调用 -> ...）中插入钩子（Hooks），允许我们在不影响Agent核心逻辑的前提下，横切（Cross-cutting） 地处理各种关注点，如日志、限流、重试、提示词注入等。

LangChain的create_agent函数支持通过middleware参数传入一个中间件列表。

2.2 消息压缩中间件：应对上下文窗口极限

当对话轮次过多，消息历史可能会超出模型的上下文窗口（Context Window）限制，导致昂贵的长上下文费用或直接报错。SummarizationMiddleware（消息压缩中间件） 正是为此而生。

它的工作原理是：当消息总长度（或总轮数）达到预设的阈值时，自动触发一个摘要过程。它会将较早的消息压缩成一个简短的文本摘要，用这个摘要代替原始消息，从而保证对话上下文始终在模型处理能力之内。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[...],
    middleware=[
        SummarizationMiddleware(
            # 例如：当上下文达到总容量的75%时触发总结
            trigger=("context_fraction", 0.75), 
            # 保留最新的10条消息不进行总结，以保证近期的细节不丢失
            keep=("messages", 10) 
        ),
    ],
)

2.3 人工审核中间件：为高风险操作按下“暂停键”

对于一些关键或高风险操作，我们往往希望在工具执行前加入人工审核环节，这就是HumanInTheLoopMiddleware（人工审核中间件）。当Agent尝试调用被interrupt_on标记的工具时，执行会暂停，并生成一个待审核的请求。系统需要等待管理员通过接口批准、拒绝或编辑该请求后，Agent才能继续执行。

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

# 假设send_email是一个发送邮件的工具
agent = create_agent(
    model="gpt-4o",
    tools=[send_email, ...],
    checkpointer=InMemorySaver(), # HITL依赖于Checkpointer来保存中断状态
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": { # 针对send_email工具
                    "allowed_decisions": ["approve", "edit", "reject"], 
                    # 允许：批准、编辑后执行、拒绝
                }
            }
        )
    ],
)

2.4 示例：组合使用中间件

在实际应用中，我们常常会组合多种中间件以达到最佳效果。比如，在一个金融咨询Agent中，我们既需要通过SummarizationMiddleware控制长对话成本，又需要通过HumanInTheLoopMiddleware保障涉及资金操作的安全性。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware, HumanInTheLoopMiddleware

agent = create_agent(
    model="gpt-4o",
    tools=[check_balance, transfer_funds, ...],
    middleware=[
        # 第一个中间件：处理消息压缩
        SummarizationMiddleware(trigger=("tokens", 100000)),
        # 第二个中间件：为关键操作加入人工审核
        HumanInTheLoopMiddleware(
            interrupt_on={"transfer_funds": {"allowed_decisions": ["approve", "reject"]}}
        ),
        # ... 可以添加更多中间件
    ],
)

总结

记忆管理和中间件是构建生产级AI Agent不可或缺的两大支柱：

记忆管理（Checkpointer + Thread ID） 让Agent拥有了连贯的对话能力，它通过持久化状态，解决了无状态Agent的“失忆”问题，是实现流畅用户体验的基础。
中间件（Middleware） 则为Agent提供了强大的治理能力，通过SummarizationMiddleware和HumanInTheLoopMiddleware等内置中间件，我们可以灵活地控制成本、保障安全并增强系统的健壮性。

掌握这两项技能，意味着你不仅能让Agent“动起来”，更能让它在你设定的轨道上“安全、高效地跑起来”。