精通LangGraph-Memory

文摘 2025-01-14 12:50 四川

LangGraph 可让您轻松管理图表中的对话记忆。这些操作指南展示了如何为此实施不同的策略。

管理对话历史记录

持久性最常见的用例之一是使用它来跟踪对话历史记录。它使继续对话变得容易。然而，随着对话越来越长，这个对话历史记录会累积起来并占用越来越多的上下文窗口。

这通常是不受欢迎的，因为它会导致对 LLM 的调用更昂贵、更长时间，并且可能会出错。为了防止这种情况发生，您可能需要管理对话历史记录。

注意：本指南重点介绍如何在 LangGraph 中执行此操作，您可以完全自定义此操作方式。如果您想要更现成的解决方案，可以查看 LangChain message 中提供的对应的功能

先准备一个简单的graph agent

from typing import Literalfrom langchain_anthropic import ChatAnthropicfrom langchain_core.tools import toolfrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import MessagesState, StateGraph, START, ENDfrom langgraph.prebuilt import ToolNodememory = MemorySaver()@tooldef search(query: str):    """Call to surf the web."""    return "It's sunny in San Francisco, but you better look out if you're a Gemini 😈."tools = [search]tool_node = ToolNode(tools)model = ChatAnthropic(model="claude-3-haiku-20240307")bound_model = model.bind_tools(tools)def should_continue(state: MessagesState):    last_message = state["messages"][-1]    if not last_message.tool_calls:        return END    return "action"# Define the function that calls the modeldef call_model(state: MessagesState):    response = bound_model.invoke(state["messages"])    return {"messages": response}workflow = StateGraph(MessagesState)workflow.add_node("agent", call_model)workflow.add_node("action", tool_node)workflow.add_edge(START, "agent")workflow.add_conditional_edges(    "agent",    should_continue,    ["action", END],)workflow.add_edge("action", "agent")app = workflow.compile(checkpointer=memory)

然后调用一下

from langchain_core.messages import HumanMessageconfig = {"configurable": {"thread_id": "2"}}input_message = HumanMessage(content="你好我是桐人")for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):    event["messages"][-1].pretty_print()input_message = HumanMessage(content="我叫什么")for event in app.stream({"messages": [input_message]}, config, stream_mode="values"):    event["messages"][-1].pretty_print()

================================ Human Message =================================
你好我是桐人================================== Ai Message ==================================
你好！桐人是一个角色名字吗？来自某个游戏或者动画作品吧？很高兴认识你，有什么我可以帮助你的吗？================================ Human Message =================================
我叫什么================================== Ai Message ==================================
您刚才告诉我您的名字是“桐人”。请问您还需要了解或确认其他信息吗？

过滤消息

为防止对话历史记录爆炸，最直接的做法是在将消息列表传递给 LLM 之前对其进行过滤。

这涉及两个部分：定义一个函数来过滤消息，然后将其添加到graph中。请参阅下面的示例，其中定义了一个非常简单的filter_messages函数，然后使用它。

上面的代码基本不变，只改了两个地方

def filter_messages(messages: list):    # 这是一个非常简单的帮助函数，它只使用最后一条消息    return messages[-1:]
def call_model(state: MessagesState):    # 注意这里：我们在调用模型之前过滤了消息    messages = filter_messages(state["messages"])    response = bound_model.invoke(messages)    return {"messages": response}

================================ Human Message =================================
你好我是桐人================================== Ai Message ==================================
你好！桐人这个角色是来自哪部作品呢？如果你有关于他的问题或者想讨论的内容，都可以和我分享哦。================================ Human Message =================================
我叫什么================================== Ai Message ==================================
你是Qwen，一个由阿里云开发的助手。但我现在应该称呼你什么呢？请告诉我你的名字吧！

可以看到，因为我们过滤消息的时候只保留了最后一条消息，因此LLM不知道上下文，因此，不知道我叫什么。

下面是langchain开箱即用的消息过滤和裁剪链接，读者可以去试一下：

https://python.langchain.com/docs/how_to/filter_messages/

https://python.langchain.com/docs/how_to/trim_messages/

如何删除消息

图表的常见状态之一是消息列表。通常您只向该状态添加消息。但是，有时您可能想要删除消息（通过直接修改状态或作为图表的一部分）。为此，您可以使用修饰符RemoveMessage。

关键思想是每个状态键都有一个reducer key。此key指定如何将更新组合到状态。默认MessagesState有一个messages key，该键的 Reducer 接受这些RemoveMessage修饰符。然后，该 Reducer 使用这些修饰符RemoveMessage从键中删除消息。

因此请注意，仅仅因为您的图形状态有一个消息列表的键，并不意味着此RemoveMessage修饰符将起作用。您还必须有一个reducer知道如何使用它的定义。

注意：许多模型都要求消息列表遵循某些规则。例如，有些模型要求消息以一条user消息开头，而有些模型则要求所有带有工具调用的消息后面都跟一条工具消息。删除消息时，您需要确保不违反这些规则。

基本代码不变，还是上面的graph 流程。

首先，我们将介绍如何手动删除消息。让我们看一下线程的当前状态：

[HumanMessage(content='你好我是桐人', additional_kwargs={}, response_metadata={}, id='7054cbf0-e714-4b4c-b065-6a9eb75b7a2d'), AIMessage(content='你好！桐人是一个来自轻小说《 Sword Art Online （刀剑神域）》的角色，对吧？你对这部作品或者角色有什么感兴趣的话题或问题想要讨论吗？', additional_kwargs={}, response_metadata={'model': 'qwen2.5:7b', 'created_at': '2025-01-07T03:40:50.799446Z', 'done': True, 'done_reason': 'stop', 'total_duration': 2063392667, 'load_duration': 24295500, 'prompt_eval_count': 155, 'prompt_eval_duration': 247000000, 'eval_count': 40, 'eval_duration': 1788000000, 'message': {'role': 'assistant', 'content': '你好！桐人是一个来自轻小说《 Sword Art Online （刀剑神域）》的角色，对吧？你对这部作品或者角色有什么感兴趣的话题或问题想要讨论吗？', 'images': None, 'tool_calls': None}}, id='run-d68ff793-1a32-49b1-a769-5b6e59c94357-0', usage_metadata={'input_tokens': 155, 'output_tokens': 40, 'total_tokens': 195}), HumanMessage(content='我叫什么', additional_kwargs={}, response_metadata={}, id='0a6eef9b-ddde-4a4d-90d3-414ddeda2ca9'), AIMessage(content='您刚才提到的名字是“桐人”。请问您现在想让我称呼您为什么名字呢？如果您有其他偏好，请告诉我。', additional_kwargs={}, response_metadata={'model': 'qwen2.5:7b', 'created_at': '2025-01-07T03:40:52.339773Z', 'done': True, 'done_reason': 'stop', 'total_duration': 1438884083, 'load_duration': 27642042, 'prompt_eval_count': 207, 'prompt_eval_duration': 163000000, 'eval_count': 28, 'eval_duration': 1242000000, 'message': {'role': 'assistant', 'content': '您刚才提到的名字是“桐人”。请问您现在想让我称呼您为什么名字呢？如果您有其他偏好，请告诉我。', 'images': None, 'tool_calls': None}}, id='run-e066f578-da75-4ad2-805e-b6fcbf5a7e2c-0', usage_metadata={'input_tokens': 207, 'output_tokens': 28, 'total_tokens': 235})]

我们可以调用update_state并传入第一条消息的 ID。这将删除该消息。

from langchain_core.messages import RemoveMessageapp.update_state(config, {"messages": RemoveMessage(id=messages[0].id)})messages = app.get_state(config).values["messages"]print(messages)

[AIMessage(content='你好！桐人是一个来自轻小说《刀剑神域》的角色，对吧？你有什么问题或者想讨论的内容吗？', additional_kwargs={}, response_metadata={'model': 'qwen2.5:7b', 'created_at': '2025-01-07T03:43:39.974483Z', 'done': True, 'done_reason': 'stop', 'total_duration': 1445857708, 'load_duration': 27660333, 'prompt_eval_count': 155, 'prompt_eval_duration': 138000000, 'eval_count': 29, 'eval_duration': 1276000000, 'message': {'role': 'assistant', 'content': '你好！桐人是一个来自轻小说《刀剑神域》的角色，对吧？你有什么问题或者想讨论的内容吗？', 'images': None, 'tool_calls': None}}, id='run-cbbaeb2b-9bcb-43a7-9e02-fa86aea86924-0', usage_metadata={'input_tokens': 155, 'output_tokens': 29, 'total_tokens': 184}), HumanMessage(content='我叫什么', additional_kwargs={}, response_metadata={}, id='d3f38c66-0c7e-4f8e-97ce-108cc57ea47a'), AIMessage(content='您刚才提到您的名字是“桐人”。如果您有任何其他问题或需要帮助的事情，请告诉我。', additional_kwargs={}, response_metadata={'model': 'qwen2.5:7b', 'created_at': '2025-01-07T03:43:41.18703Z', 'done': True, 'done_reason': 'stop', 'total_duration': 1142763958, 'load_duration': 12822958, 'prompt_eval_count': 196, 'prompt_eval_duration': 161000000, 'eval_count': 22, 'eval_duration': 964000000, 'message': {'role': 'assistant', 'content': '您刚才提到您的名字是“桐人”。如果您有任何其他问题或需要帮助的事情，请告诉我。', 'images': None, 'tool_calls': None}}, id='run-4b322f02-3689-45a8-92d8-11783e7d7869-0', usage_metadata={'input_tokens': 196, 'output_tokens': 22, 'total_tokens': 218})]

我们可以验证第一条消息已被删除。

以编程的方式删除消息

我们还可以从图表内部以编程方式删除消息。在这里，我们将修改图表以在图表运行结束时删除所有旧消息（超过 3 条消息之前的消息）。

from langchain_core.messages import RemoveMessagefrom langgraph.graph import END

def delete_messages(state):    messages = state["messages"]    if len(messages) > 3:        return {"messages": [RemoveMessage(id=m.id) for m in messages[:-3]]}

# 我们要修改逻辑调用 delete_messages不是立刻enddef should_continue(state: MessagesState) -> Literal["action", "delete_messages"]:    last_message = state["messages"][-1]    if not last_message.tool_calls:        return "delete_messages"    return "action"

# Define a new graphworkflow = StateGraph(MessagesState)workflow.add_node("agent", call_model)workflow.add_node("action", tool_node)
# 新定义的nodeworkflow.add_node(delete_messages)

workflow.add_edge(START, "agent")workflow.add_conditional_edges(    "agent",    should_continue,)workflow.add_edge("action", "agent")
# 这里添加了一条新的边workflow.add_edge("delete_messages", END)app = workflow.compile(checkpointer=memory)

相当于在最后END之前调用了删除消息，当然了我们可以自定义其它删除的node和时机。这里结果就不做展示了。

构建对话摘要

这通常是不受欢迎的，因为它会导致对 LLM 的调用更昂贵、更长时间，并且可能会出错。解决这个问题的一种方法是创建迄今为止的对话摘要，这跟上面的过滤消息和删除消息的思路稍微有所不同。并将其与过去的 N 条消息一起使用。

这将涉及几个步骤：检查对话是否太长（可以通过检查消息数量或消息长度来完成）
如果是，则创建摘要（需要提示）
然后删除除最后 N 条消息之外的所有消息

其中很大一部分是删除旧消息。

from typing import Literalfrom langchain_anthropic import ChatAnthropicfrom langchain_core.messages import SystemMessage, RemoveMessage, HumanMessagefrom langgraph.checkpoint.memory import MemorySaverfrom langgraph.graph import MessagesState, StateGraph, START, ENDmemory = MemorySaver()# 我们将添加一个`summary`属性（除了`messages`键之外，MessagesState已经有了）class State(MessagesState):    summary: strmodel = ChatAnthropic(model_name="claude-3-haiku-20240307")# 定义调用模型的逻辑def call_model(state: State):    # 如果存在摘要，我们将其作为系统消息添加进去    summary = state.get("summary", "")    if summary:        system_message = f"Summary of conversation earlier: {summary}"        messages = [SystemMessage(content=system_message)] + state["messages"]    else:        messages = state["messages"]    response = model.invoke(messages)    return {"messages": [response]}# 我们现在定义确定是否结束或总结对话的逻辑def should_continue(state: State) -> Literal["summarize_conversation", END]:    """Return the next node to execute."""    messages = state["messages"]    # 如果消息超过六条，我们就总结对话    if len(messages) > 6:        return "summarize_conversation"    # 否则我们就结束    return ENDdef summarize_conversation(state: State):    # 首先，我们总结对话    summary = state.get("summary", "")    if summary:        # 如果已经存在摘要，我们使用不同的系统提示来总结它        summary_message = (            f"This is summary of the conversation to date: {summary}\n\n"            "Extend the summary by taking into account the new messages above:"        )    else:        summary_message = "Create a summary of the conversation above:"    messages = state["messages"] + [HumanMessage(content=summary_message)]    response = model.invoke(messages)    # 我们现在需要删除不再想显示的消息    # 我将删除除最后两条消息之外的所有消息，但您可以更改此设置    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]    return {"summary": response.content, "messages": delete_messages}# Define a new graphworkflow = StateGraph(State)# 定义对话节点和总结节点workflow.add_node("conversation", call_model)workflow.add_node(summarize_conversation)# 将入口点设置为对话workflow.add_edge(START, "conversation")# We now add a conditional edgeworkflow.add_conditional_edges(    # 首先，我们定义开始节点。我们使用`conversation`。    # 这意味着这些是在调用`conversation`节点之后采取的边。    "conversation",    # 接下来，我们传入将确定下一个调用的函数。    should_continue,)# 我们现在从`summarize_conversation`到END添加了一个普通边。# 这意味着在调用`summarize_conversation`之后，我们直接ENDworkflow.add_edge("summarize_conversation", END)app = workflow.compile(checkpointer=memory)

这个代码比较清晰，这里不再实际运行，你可以动手试试！

另外跨thread的存储，我们在之前的持久化的部分已经讲过了，这里就不重复了，基本上就是需要添加一个store。

如果想执行memory的语义搜索也可以使用相关的embeddings 和store实现，同样可以参考之前的持久化部分。

半夏决明

读书，摄影，随笔