Agent开发，短期和长期记忆 / ai #48

机器人的记忆(由AIJoe生成)

短期记忆

LangGraph框架中的checkpointer，通过一些数据结构来存储State状态中产生的信息，并且在每个task开始时去读取全局的状态。主要通过以下四种方式来实现：

• MemorySaver： 用于实验性质的记忆检查点。
• SqliteSaver / AsyncSqliteSaver： 使用 SQLite 数据库 实现的记忆检查点，适合实验性质和本地工作流程。
• PostgresSaver / AsyncPostgresSaver： 使用 Postgres 数据库实现的高级检查点，适合在生产系统中使用。
• 支持自定义检查点。

checkpointer是memory的一种特定实现，它在执行期间保存图在各个点的状态，使系统能够在中断时从该点恢复。这与 LangGraph 中状态的一般概念不同，后者表示应用程序在任何给定时刻的当前快照。虽然状态是动态的并且随着图形的执行而变化，但checkpointer提供了一种存储和检索历史状态的方法，从而促进更复杂的工作流程和人机交互。

以 MemorySaver 这个实现checkpointer的方法为例，帮助大家理解这个过程。

# 导入检查点
from langgraph.checkpoint.memory import MemorySaver

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)


memory = MemorySaver()
graph_with_memory = builder.compile(checkpointer=memory)   # 在编译图的时候添加检查点
# 当添加了`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如何去做持久化的存储呢？正如我们上面使用的 MemorySaver， 虽然在当前的代码运行环境下可以去指定线程ID，获取到具体的历史信息，但是，一旦我们重启代码环境，则所有的数据都将被抹除。那么一种持久化的方法就是把每个checkpointer存储到本地的数据库中。

长期记忆

有效的记忆管理可以增强代理维护上下文、从过去的经验中学习以及随着时间的推移做出更明智决策的能力。大多数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()