今天简单聊聊Agent的短期记忆管理，短期记忆大多数时候是指对话过程中产生的上下文信息。
一共3种短期记忆管理策略，内容可能会比较简单，主要是这2天在试玩一些开源的open deepreserach方案，发现langgraph新增的一些方法，用起来比较方便，分享给家人们~
目前的大多数chatbot系统都在做一件事，它们会存储对话历史中的每一句话。现在的大模型，大多数都会使用ChatML的prompt模板，可以天然的完成这个存储过程，自动的拼接到prompt中。
但是，想象一下，当我们和其他人聊天时，似乎并不会记住对话中的每一个字、每一个标点符号。可能只会记住谈话的要点和关键信息。
所以目前这种常见的短期记忆的拼接，会带来三个明显的问题:
Token用量暴增 - 每次对话都要消耗大量token 上下文窗口溢出 - 超出模型的处理上限 响应延迟 - 需要处理过多历史信息
我们可以用langgraph轻松模拟出这样一个chatbot系统。
fromlanggraph.graphimportMessagesState,START,END,StateGraph
defchat_model_node(state:MessagesState):
response=llm.invoke(state["messages"])
return{"messages":response}
builder=StateGraph(MessagesState)
builder.add_node("chat_model_node",chat_model_node)
builder.add_edge(START,"chat_model_node")
builder.add_edge("chat_model_node",END)
graph=builder.compile()
graph_response=graph.invoke({"messages":messages})
formingraph_response["messages"]:
m.pretty_print()
RemoveMessage是langgraph中的一个最简单管理短期记忆的方法。简单来说，它就是只保留最近的N条对话记录。比如:
为了精简信息，这里可以只保留消息列表中的最后三条消息：
fromlangchain_core.messagesimportRemoveMessage
deffilter_messages(state):
messages=state["messages"]
#只保留最后3条消息
messages=[RemoveMessage(m.id)forminmessages[:-3]]
return{"messages":messages}
builder=StateGraph(MessagesState)
builder.add_node("filter_messages",filter_messages)
builder.add_node("chat_model_node",chat_model_node)
builder.add_edge(START,"filter_messages")
builder.add_edge("filter_messages","chat_model_node")
builder.add_edge("chat_model_node",END)
graph=builder.compile()
插入一个filter_messages节点
这种方法简单有效,但可能会丢失一些重要的历史信息。并且每次输入prompt的轮次固定的，可能会超长，也可能会分成短。
因此，trim_messages 方法来了，仍然是只保留一定量的记忆历史，但是会计算出保留的历史信息符合，自行设定的特定tokens大小。
fromlangchain_core.messagesimporttrim_messages
defchat_model_node(state:MessagesState):
messages=trim_messages(
allow_partial=True,#允许消息在中间部分部分拆分；这种方法可能会丢失上下文。
strategy="last",#从哪里开始算，last最后开始
max_tokens=100,#最大tokens数量
token_counter=ChatOpenAI(model="gpt-3.5-turbo"),
state=state["messages"]
)
response=llm.invoke(messages)
return{"messages":response}
builder=StateGraph(MessagesState)
builder.add_node("chat_model_node",chat_model_node)
builder.add_edge(START,"chat_model_node")
builder.add_edge("chat_model_node",END)
graph=builder.compile()
最后是，动态摘要的方法。通过定期对对话历史进行总结,只保留关键信息。
代码也很简单，这里有个技巧是，当每一次积累到K轮才开始总结，避免过多的消耗，增加一个conditional边即可。
defsummarize_conversation(state:State):
summary=state.get("summary","")
ifsummary:
summary_message=(
f"Thisissummaryoftheconversationtodate:{summary}\n\n"
"Extendthesummarybytakingintoaccountthenewmessagesabove:"
)
else:
summary_message="Createasummaryoftheconversationabove:"
messages=state["messages"]+[HumanMessage(content=summary_message)]
response=model.invoke(messages)
delete_messages=[RemoveMessage(id=m.id)forminstate["messages"][:-2]]
return{"summary":response.content,"messages":delete_messages}
defshould_continue(state:State)->Literal["summarize_conversation",END]:
messages=state["messages"]
iflen(messages)>6:
return"summarize_conversation"
returnEND
workflow=StateGraph(State)
workflow.add_node("conversation",call_model)
workflow.add_node(summarize_conversation)
workflow.add_edge(START,"conversation")
workflow.add_conditional_edges(
"conversation",
should_continue,
)
workflow.add_edge("summarize_conversation",END)