rag系列文章目录

---

前言

众所周知，大模型是无状态的。但是基于大模型的agent一般是有状态的，也就是它有记忆功能。在AI Agent框架中，Memory机制是核心组件之一，它赋予Agent持续学习和上下文感知的能力，使其能够像人类一样基于历史交互进行决策。

---

一、Memory机制作用

上下文保持（Context Preservation）
存储对话历史、工具调用记录等，确保多轮交互中Agent能理解用户意图的连贯性。例如，用户问“上海天气如何？”后追问“明天呢？”，Agent需记忆前一轮的“上海”地理位置

状态持久化（State Persistence）
支持断点续传：通过Checkpoint机制（如MemorySaver或PostgresSaver）将运行时状态序列化存储，崩溃后可从最近状态恢复，避免重复计算。
典型场景：长时间运行的任务（如代码生成）中途中断后，可从上次保存的步骤继续。

个性化适配（Personalization）
记忆用户偏好或历史行为（如常查询的城市），实现定制化响应。

协作与隔离（Multi-agent Coordination）
通过thread_id隔离不同用户/任务的记忆，避免数据混淆。例如，多用户对话中，每个用户的会话历史独立存储。

资源优化（Token Efficiency）
动态修剪或摘要历史消息（如trim_messages），避免上下文窗口溢出或token浪费。例如仅保留最近3条消息或按token数截断。

二、memory分类

短期记忆：
本质：就是上下文窗口内的 对话历史，存储在 KV Cache 或者文本 buffer 里。
使用场景
• 聊天机器人（连续对话）
• 任务执行中，需要记住上一轮用户的指令

长期记忆:
本质：把对话内容、用户信息等存入一个外部数据库（向量库/知识库），通过 检索 机制在需要时取出。
使用场景
• 长期陪伴型智能体（记住用户的兴趣、习惯）
• 多会话场景（跨会话记忆，不仅限于一次对话）
• Agent 需要跨天/跨周执行任务（如知识管理、个人助理）

类型	存储方式	实现原理	适用场景
短期记忆	State / Buffer / Cache	把对话历史拼接进 prompt	连续对话、任务短期上下文
长期记忆	向量库 / 数据库	embedding + 检索 / 摘要存储	长期个性化、多会话、跨时空任务

三、langgraph实践

• 无记忆情况

from typing_extensions import TypedDict, Annotated
from typing import Any
from langchain_core.messages import HumanMessage, AnyMessage, SystemMessage
from langchain_groq import ChatGroq
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_conditionfrom dotenv import load_dotenvload_dotenv()# Define the state
class MessagesState(TypedDict):messages: Annotated[list[AnyMessage], add_messages]# Tool: Person Information
def get_person_details(person_name: str) -> str:"""Retrieve details of a person."""return f"{person_name} is a DevOps Engineer."# Tool: Location Information
def get_person_location(person_name: str) -> str:"""Retrieve location of a person."""return f"{person_name} lives in Bangalore."# Initialize the Groq chat model
llm = ChatOpenAI(base_url="",model="gpt-4o-mini",api_key="",temperature=0.0)# Bind tools to the Groq model
tools = [get_person_details, get_person_location]
llm_with_tools = llm.bind_tools(tools)# System message
sys_msg = SystemMessage(content="You are a helpful assistant that provides accurate responses based on the given tools.")# Define the assistant node
def assistant(state: MessagesState):return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}# Build the graph
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))# Adding Edges
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")# Example without memory
react_graph = builder.compile()# Test interaction without memory
messages = [HumanMessage(content="Tell me about Harsha?she is girl")]
result = react_graph.invoke({"messages": messages})
# for message in result['messages']:
#     message.pretty_print()# Test interaction without memory
messages = [HumanMessage(content="Is she a Devops Enginer?")]
result = react_graph.invoke({"messages": messages})
for message in result['messages']:message.pretty_print()

• 有记忆情况

from typing_extensions import TypedDict, Annotated
from typing import Any
from langchain_core.messages import HumanMessage, AnyMessage, SystemMessage
from langchain_groq import ChatGroq
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaverfrom dotenv import load_dotenvload_dotenv()# Define the state
class MessagesState(TypedDict):messages: Annotated[list[AnyMessage], add_messages]# Tool: Person Information
def get_person_details(person_name: str) -> str:"""Retrieve details of a person."""return f"{person_name} is a DevOps Engineer."# Tool: Location Information
def get_person_location(person_name: str) -> str:"""Retrieve location of a person."""return f"{person_name} lives in Bangalore."# Initialize the Groq chat model
# llm = ChatGroq(model="llama-3.3-70b-versatile")
llm = ChatOpenAI(base_url="",model="gpt-4o-mini",api_key="",temperature=0.0)
# Bind tools to the Groq model
tools = [get_person_details, get_person_location]
llm_with_tools = llm.bind_tools(tools)# System message
sys_msg = SystemMessage(content="You are a helpful assistant that provides accurate responses based on the given tools.")
memory = MemorySaver() # Define the assistant node
def assistant(state: MessagesState):return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}# Build the graph
builder = StateGraph(MessagesState)
builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))# Adding Edge
builder.add_edge(START, "assistant")
builder.add_conditional_edges("assistant", tools_condition)
builder.add_edge("tools", "assistant")
react_graph = builder.compile(checkpointer=memory)# Test interaction with memory
# Thread ID for maintaining context
config = {"configurable": {"thread_id": "1"}}messages = [HumanMessage(content="Tell me about Harsha?she is girl")]
result = react_graph.invoke({"messages": messages}, config)messages = [HumanMessage(content="Is she a Devops Enginer?")]
result = react_graph.invoke({"messages": messages}, config)
for message in result['messages']:message.pretty_print()

这里根据上面的问题，agent能够自动推理出，第二个问题的主体人。

---

总结

类似人脑一样，memory机制是agent智能化的基石。当前，memory机制也有一些挑战，比如记忆泄露（未清理的旧数据可能导致存储膨胀（需设置TTL））、隐私合规（GDPR要求用户数据可删除）等。

参考文档