有效的记忆管理可以增强代理维护上下文、从过去的经验中学习以及随着时间的推移做出更明智决策的能力。大多数AI Agent构建的应用程序都需要记忆来在多个交互中共享上下文。在 LangGraph 中,这种记忆就是通过checkpointer 和 store 来做持久性,从而添加到任何StateGraph中。最常见的用例之一是用它来跟踪对话历史记录,但是也有很大的优化空间,因为随着对话变得越来越长,历史记录会累积并占用越来越多的上下文窗口,导致对大模型的调用更加昂贵和耗时,并且可能会出错。为了防止这种情况发生,我们一般是需要借助一些优化手段去管理对话历史记录,同时更加适配生产环境的PostgresSaver / AsyncPostgresSaver )高级检查点,我们也将随着知识点的进一步补充后,再结合实际的案例进行详细的讲解。
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
checkpointer = InMemorySaver()
store = InMemoryStore()
model = ...
research_agent = ...
math_agent = ...
workflow = create_supervisor(
[research_agent, math_agent],
model=model,
prompt="You are a team supervisor managing a research expert and a math expert.",
)
# Compile with checkpointer/store
graph = workflow.compile(
checkpointer=checkpointer,
store=store
)
config = {"configurable": {"thread_id": "111"}, "user_id": "8"}
async for chunk in graph.astream({"messages": ["你好,介绍一个你自己"]}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
LangGraph 框架中的Memory通过状态管理与检查点与 thread_id 进行绑定,从而隔离不同线程的记忆和状态,防止并发交互之间的干扰,保证每个线程独立运行。总体而言,thread_id 是 LangGraph 中组织和管理记忆的关键机制,可实现有效的状态跟踪和交互连续性。但是,通过thread_id不能实现的是:跨线程的交互。我们虽然可以通过thread_id去中断或者恢复某个对话,但是让不同线程间共享各自的消息,在checkpointer的实现机制下并不能做到,所以有了LangGraph的长期记忆的实现模块。
RE: LangGraph开发实战