大模型本身是不具备记忆能力的。AI 应用需要记忆(memory)来在多次交互中共享上下文。在 LangGraph 中,您可以添加两种类型的记忆:
Memory会将其所实现的记忆功能分为短期记忆和长期记忆两大类,以适应不同的应用需求。对短期记忆的普遍理解是指那些存储在缓存、内存或程序运行过程中的状态信息。这类信息通常作为整个大模型或Agent的短期记忆,在构建过程中通过数据接口进行临时存储,并在任务完成后直接被清除,就像LangGraph的State状态机制一样。相对地,长期记忆通常涉及将重要数据通过特定方式持久化保存在某种存储容器中,任何程序都可以在这里提取到内容,且不受时间的影响,一般这种常规的实现方法是数据库的本地存储。
LangGraph框架中的checkpointer,通过一些数据结构来存储State状态中产生的信息,并且在每个task开始时去读取全局的状态。主要通过以下四种方式来实现:
SQLite 数据库 实现的记忆检查点,适合实验性质和本地工作流程。Postgres 数据库实现的高级检查点,适合在生产系统中使用。checkpointer是memory的一种特定实现,它在执行期间保存图在各个点的状态,使系统能够在中断时从该点恢复。这与 LangGraph 中状态的一般概念不同,后者表示应用程序在任何给定时刻的当前快照。虽然状态是动态的并且随着图形的执行而变化,但checkpointer提供了一种存储和检索历史状态的方法,从而促进更复杂的工作流程和人机交互。
短期记忆(线程级持久化)使代理能够跟踪多轮对话。
# 导入检查点
from langgraph.checkpoint.memory import InMemorySaver
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
class State(TypedDict):
messages: Annotated[list, add_messages]
def call_model(state: State):
response = llm.invoke(state["messages"])
return {"messages": response}
def translate_message(state: State):
system_prompt = """
Please translate the received text in any language into English as output
"""
messages = state['messages'][-1]
messages = [SystemMessage(content=system_prompt)] + [HumanMessage(content=messages.content)]
response = llm.invoke(messages)
return {"messages": response}
builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_node("translate_message", translate_message)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "translate_message")
builder.add_edge("translate_message", END)
checkpointer = InMemorySaver()
graph_with_memory = builder.compile(checkpointer=checkpointer) # 在编译图的时候添加检查点
# 当添加了`checkpointer`后,在该图执行的每个超级步骤中会自动创建检查点。即每个节点处理其输入并更新状态后,会当前状态将保存为检查点。但如果像普通图一样,仅传入输入的问题是会报错的
# **当增加了`checkpointer`后,需要`Thread`来作为`checkpointer`保存图中每个检查点的唯一标识,而`Thread`(线程)又是通过`thread_id`来标识某个特定执行线程,所以在使用`checkpointer`调用图时,必须指定`thread_id`,指定的方式是作为配置`configurable`的一部分进行声明。** 正确调用的代码就如下所示:
# 这个 thread_id 可以取任意数值
config = {"configurable": {"thread_id": "1"}}
for chunk in graph_with_memory.stream({"messages": ["你好,我叫西山老师"]}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
for chunk in graph_with_memory.stream({"messages": ["请问我叫什么?"]}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()
短期记忆可让应用程序记住单个线程或对话中先前的交互,并且可以随时找到某个对话线程中继续之前的问答。LangGraph 将短期记忆作为代理状态的一部分进行管理,并通过线程范围的检查点进行持久化。此状态通常可以包括对话历史记录以及其他状态数据,例如上传的文件、检索的文档或生成的工件。通过将这些存储在图的状态中,程序可以访问给定对话的完整上下文,同时保持不同线程之间的分离。这就是其现实应用价值的体现。
那么接下来要考虑的是: 既然所实际进行存储的是 Checkpointer, 那么Checkpointer如何去做持久化的存储呢?正如我们上面使用的 InMemorySaver, 虽然在当前的代码运行环境下可以去指定线程ID,获取到具体的历史信息,但是,一旦我们重启代码环境,则所有的数据都将被抹除。那么一种持久化的方法就是把每个checkpointer存储到本地的数据库中。
RE: LangGraph开发实战