AI Agent 记忆与状态管理教程
SEO 元信息
- 名称：AI Agent 记忆与状态管理教程
- 描述：零基础 AI Agent 记忆与状态管理教程，涵盖 Agent 记忆系统架构、短期长期记忆设计、向量数据库存储、LangGraph 状态管理、RAG 增强记忆、多 Agent 记忆共享等核心技能，适合 AI 开发者系统学习。
- 关键词：AI Agent 记忆, 状态管理, LangGraph, 向量数据库, 长期记忆
- 长尾关键词：AI Agent 记忆系统设计教程, LangGraph 状态管理实战, AI Agent 长期记忆实现方案, 多 Agent 记忆共享开发教程
目录
- 为什么 Agent 需要记忆？
- AI Agent 记忆系统概述
- 对话历史管理策略
- 向量数据库存储长期记忆
- LangChain/LangGraph 记忆模块
- 状态机与状态图设计
- 检索增强记忆（RAG + Memory）
- 多 Agent 间记忆共享
- 记忆持久化与序列化
- 记忆压缩与摘要策略
- 实战：构建有记忆能力的个人助手
- 总结与最佳实践
1. 为什么 Agent 需要记忆？
大语言模型（LLM）本身是无状态的——每次 API 调用都是独立的，模型不会记住上一次对话的内容。这种"金鱼记忆"在构建 AI Agent 时会造成严重问题：
- 任务断裂：用户说"帮我查一下昨天那个项目的进度"，Agent 完全不知道"那个项目"是什么
- 重复劳动：每次对话都要重新解释用户偏好、背景信息
- 无法学习：Agent 无法从过去的交互中积累经验
记忆系统的核心目标就是让 Agent 拥有跨会话的上下文保持能力和持续学习能力。
2. AI Agent 记忆系统概述
借鉴认知科学的理论，AI Agent 的记忆可以分为三个层次：
2.1 短期记忆（Short-term Memory）
对应概念： 工作记忆 / 对话上下文
实现方式： 当前会话的消息列表
生命周期： 会话结束即丢失（或被压缩）
# 短期记忆的本质就是一个消息列表
short_term_memory = [
{"role": "user", "content": "帮我写一个 Python 脚本"},
{"role": "assistant", "content": "好的，请问脚本需要实现什么功能？"},
{"role": "user", "content": "读取 CSV 文件并统计每列的缺失值"},
]
特点：
- 访问速度最快（直接在内存中）
- 容量有限（受模型上下文窗口限制，通常 4K-128K tokens）
- 无需额外存储
- 天然支持对话连贯性
2.2 长期记忆（Long-term Memory）
对应概念： 情景记忆 / 语义记忆
实现方式： 向量数据库 + 外部存储
生命周期： 持久化，跨会话保持
# 长期记忆存储在外部数据库中
long_term_memory = {
"user_preferences": {
"language": "Python",
"code_style": "Google Style Guide",
"response_format": "简洁直接",
},
"past_interactions": [
# 向量化的对话片段，支持语义检索
],
"learned_facts": [
# 用户教给 Agent 的知识
],
}
特点：
- 容量几乎无限
- 需要检索机制（语义搜索、关键词匹配等）
- 访问有延迟（需要查询外部存储）
- 支持跨会话持久化
2.3 工作记忆（Working Memory）
对应概念： 当前任务的临时状态
实现方式： 结构化的状态对象
生命周期： 单次任务执行期间
# 工作记忆是当前任务的上下文
working_memory = {
"current_task": "数据分析",
"step": 2,
"total_steps": 5,
"intermediate_results": {
"data_shape": (1000, 15),
"missing_rate": 0.05,
},
"next_action": "处理缺失值",
}
2.4 三层记忆协作模型
用户输入
│
▼
┌──────────────────────────────────────────┐
│ 工作记忆 │
│ 当前任务状态、中间结果 │
│ ▲ ▲ │
│ │ │ │
│ ┌─────────┘ └──────────┐ │
│ │ │ │
│ ▼ ▼ │
│ 短期记忆 长期记忆 │
│ 对话上下文 历史知识 │
│ (消息列表) (向量数据库) │
└──────────────────────────────────────────┘
3. 对话历史管理策略
3.1 滑动窗口（Sliding Window）
最简单的策略：只保留最近 N 条消息。
from collections import deque
class SlidingWindowMemory:
def __init__(self, max_messages: int = 20):
self.messages = deque(maxlen=max_messages)
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
def get_context(self) -> list[dict]:
return list(self.messages)
# 使用
memory = SlidingWindowMemory(max_messages=20)
memory.add("user", "你好")
memory.add("assistant", "你好！有什么可以帮你的？")
memory.add("user", "帮我写一个排序算法")
context = memory.get_context() # 只包含最近20条
优点： 简单高效，token 消耗可控 缺点： 可能丢失重要上下文，无差别丢弃
3.2 Token 限制策略
基于 token 数量而非消息数量来管理上下文。
import tiktoken
class TokenLimitedMemory:
def __init__(self, max_tokens: int = 4000):
self.messages: list[dict] = []
self.max_tokens = max_tokens
self.encoder = tiktoken.get_encoding("cl100k_base")
def _count_tokens(self, text: str) -> int:
return len(self.encoder.encode(text))
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self._trim()
def _trim(self):
total = sum(self._count_tokens(m["content"]) for m in self.messages)
while total > self.max_tokens and len(self.messages) > 1:
removed = self.messages.pop(0)
total -= self._count_tokens(removed["content"])
def get_context(self) -> list[dict]:
return self.messages
3.3 摘要压缩策略
当对话过长时，用 LLM 对早期对话进行摘要，替换原始消息。
class SummaryMemory:
def __init__(self, llm, max_messages: int = 20):
self.messages: list[dict] = []
self.summary: str = ""
self.llm = llm
self.max_messages = max_messages
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
if len(self.messages) > self.max_messages:
self._compress()
def _compress(self):
# 取前半部分消息进行摘要
old_messages = self.messages[: len(self.messages) // 2]
summary_prompt = "请将以下对话总结为简要摘要，保留关键信息：\n\n"
for msg in old_messages:
summary_prompt += f"{msg['role']}: {msg['content']}\n"
new_summary = self.llm.generate(summary_prompt)
self.summary = f"之前对话摘要：{new_summary}\n" if self.summary else new_summary
# 保留后半部分消息
self.messages = self.messages[len(self.messages) // 2 :]
def get_context(self) -> list[dict]:
context = []
if self.summary:
context.append({"role": "system", "content": self.summary})
context.extend(self.messages)
return context
3.4 重要性评分策略
不是所有消息都同等重要。基于重要性分数来决定保留哪些消息。
def calculate_importance(message: dict) -> float:
"""计算消息的重要性分数（0-1）"""
score = 0.5 # 基础分
content = message["content"].lower()
# 包含决策关键词
if any(kw in content for kw in ["决定", "选择", "确认", "同意"]):
score += 0.2
# 包含具体数据
if any(c.isdigit() for c in content):
score += 0.1
# 用户消息通常比助手消息更重要
if message["role"] == "user":
score += 0.1
# 较长的消息通常包含更多信息
if len(content) > 200:
score += 0.1
return min(score, 1.0)
4. 向量数据库存储长期记忆
4.1 为什么用向量数据库？
传统的关键词搜索无法理解语义。向量数据库通过将文本转换为高维向量，支持语义相似度搜索。
用户查询："那个帮我修电脑的人叫什么"
关键词搜索：找不到（没有"修电脑"的精确匹配）
向量搜索：找到 → "张工上周帮我组装了一台新主机"（语义相似）
4.2 主流向量数据库对比
| 数据库 | 特点 | 适用场景 |
|---|---|---|
| ChromaDB | 轻量、嵌入式、Python 原生 | 本地开发、小型项目 |
| FAISS | Facebook 出品、高性能 | 大规模向量检索 |
| Pinecone | 全托管云服务 | 生产环境、不想运维 |
| Milvus | 开源分布式 | 大规模生产环境 |
| Qdrant | Rust 编写、高性能 | 高性能需求 |
4.3 使用 ChromaDB 实现长期记忆
import chromadb
from chromadb.utils import embedding_functions
class LongTermMemory:
def __init__(self, collection_name: str = "agent_memory"):
self.client = chromadb.PersistentClient(path="./memory_db")
self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key="your-api-key",
model_name="text-embedding-3-small"
)
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=self.embedding_fn,
)
def remember(self, content: str, metadata: dict = None):
"""存储一条记忆"""
import time
memory_id = f"mem_{int(time.time() * 1000)}"
self.collection.add(
ids=[memory_id],
documents=[content],
metadatas=[metadata or {}],
)
return memory_id
def recall(self, query: str, top_k: int = 5) -> list[dict]:
"""根据语义查询检索相关记忆"""
results = self.collection.query(
query_texts=[query],
n_results=top_k,
)
memories = []
for i, doc in enumerate(results["documents"][0]):
memories.append({
"content": doc,
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i],
})
return memories
def forget(self, memory_id: str):
"""删除一条记忆"""
self.collection.delete(ids=[memory_id])
# 使用示例
memory = LongTermMemory()
# 存储记忆
memory.remember(
"用户喜欢用 Python，偏好 Google 代码风格",
{"type": "preference", "timestamp": "2024-01-15"}
)
memory.remember(
"用户的项目叫 DataPipeline，使用 Pandas 和 FastAPI",
{"type": "project", "timestamp": "2024-01-16"}
)
# 检索记忆
results = memory.recall("用户的编程偏好是什么？")
for r in results:
print(f"[相似度: {1 - r['distance']:.2f}] {r['content']}")
4.4 记忆的元数据管理
合理的元数据设计让记忆管理更高效：
metadata_schema = {
"type": "preference | fact | event | instruction",
"timestamp": "ISO 8601 格式",
"source": "对话 | 教学 | 观察",
"importance": "high | medium | low",
"session_id": "关联的会话 ID",
"expires_at": "可选的过期时间",
}
5. LangChain/LangGraph 记忆模块
LangChain 和 LangGraph 提供了丰富的记忆组件，可以快速构建有记忆能力的 Agent。
5.1 LangChain 记忆类型
from langchain.memory import (
ConversationBufferMemory,
ConversationSummaryMemory,
ConversationBufferWindowMemory,
ConversationSummaryBufferMemory,
)
# 1. 完整缓冲记忆 - 保留所有消息
buffer_memory = ConversationBufferMemory(return_messages=True)
# 2. 窗口记忆 - 只保留最近 K 轮
window_memory = ConversationBufferWindowMemory(k=5, return_messages=True)
# 3. 摘要记忆 - 用 LLM 压缩历史
summary_memory = ConversationSummaryMemory(
llm=ChatOpenAI(model="gpt-4o-mini"),
return_messages=True,
)
# 4. 摘要缓冲记忆 - 混合策略（最新的原样保留，旧的摘要）
summary_buffer_memory = ConversationSummaryBufferMemory(
llm=ChatOpenAI(model="gpt-4o-mini"),
max_token_limit=2000,
return_messages=True,
)
5.2 在 Chain 中使用记忆
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
llm = ChatOpenAI(model="gpt-4o-mini")
memory = ConversationBufferMemory(return_messages=True)
conversation = ConversationChain(
llm=llm,
memory=memory,
verbose=True,
)
# 多轮对话自动保持上下文
response1 = conversation.predict(input="你好，我叫张三")
response2 = conversation.predict(input="我是一名数据分析师")
response3 = conversation.predict(input="你还记得我叫什么吗？")
# → "你叫张三"
5.3 LangGraph 中的状态管理
LangGraph 使用图（Graph）结构来管理 Agent 的状态流转，比 LangChain 的 Chain 更灵活：
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
# 定义状态类型
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
user_name: str | None
preferences: dict
# 创建图
graph = StateGraph(AgentState)
llm = ChatOpenAI(model="gpt-4o-mini")
def chatbot_node(state: AgentState):
"""对话节点"""
response = llm.invoke(state["messages"])
return {"messages": [response]}
def extract_info_node(state: AgentState):
"""信息提取节点 - 从对话中提取用户信息"""
last_msg = state["messages"][-1].content
updates = {}
# 简单的规则提取（实际中可用 LLM 提取）
if "我叫" in last_msg:
name = last_msg.split("我叫")[-1].strip().rstrip("。，！?")
updates["user_name"] = name
if updates:
return updates
return {}
# 构建图
graph.add_node("chatbot", chatbot_node)
graph.add_node("extract_info", extract_info_node)
graph.add_edge(START, "extract_info")
graph.add_edge("extract_info", "chatbot")
graph.add_edge("chatbot", END)
# 添加检查点（持久化状态）
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
# 使用
config = {"configurable": {"thread_id": "user-123"}}
# 第一轮对话
result1 = app.invoke(
{"messages": [("user", "你好，我叫李四")]},
config=config,
)
# 第二轮对话 - 状态自动保持
result2 = app.invoke(
{"messages": [("user", "你还记得我叫什么吗？")]},
config=config,
)
6. 状态机与状态图设计
6.1 状态机基础
Agent 的行为可以用有限状态机（FSM）来建模：
from enum import Enum
from typing import Callable
class AgentState(Enum):
IDLE = "idle" # 空闲等待输入
UNDERSTANDING = "understanding" # 理解用户意图
PLANNING = "planning" # 制定执行计划
EXECUTING = "executing" # 执行任务
VERIFYING = "verifying" # 验证结果
RESPONDING = "responding" # 生成回复
ERROR = "error" # 错误处理
class StateMachine:
def __init__(self):
self.state = AgentState.IDLE
self.transitions: dict[tuple, Callable] = {}
self.context: dict = {}
def add_transition(self, from_state, to_state, action):
self.transitions[(from_state, to_state)] = action
def transition(self, to_state, **kwargs):
key = (self.state, to_state)
if key not in self.transitions:
raise ValueError(f"不允许从 {self.state} 转换到 {to_state}")
action = self.transitions[key]
result = action(self.context, **kwargs)
self.state = to_state
return result
# 定义转换
sm = StateMachine()
sm.add_transition(
AgentState.IDLE, AgentState.UNDERSTANDING,
lambda ctx, **kw: {"intent": "analyze", "input": kw.get("user_input")}
)
6.2 LangGraph 状态图
LangGraph 的状态图是生产级的状态管理方案，支持条件分支、循环、并行执行：
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
class WorkflowState(TypedDict):
query: str
intent: str | None
plan: list[str] | None
results: dict | None
response: str | None
retry_count: int
graph = StateGraph(WorkflowState)
def classify_intent(state: WorkflowState):
"""意图分类"""
query = state["query"]
# 用 LLM 分类意图
intent = llm.invoke(f"分类用户意图: {query}").content
return {"intent": intent}
def make_plan(state: WorkflowState):
"""制定执行计划"""
return {"plan": [f"步骤1: 搜索 {state['intent']}", "步骤2: 总结结果"]}
def execute_plan(state: WorkflowState):
"""执行计划"""
return {"results": {"data": "执行结果"}}
def generate_response(state: WorkflowState):
"""生成回复"""
return {"response": f"基于 {state['results']} 生成的回复"}
def should_retry(state: WorkflowState) -> Literal["retry", "end"]:
"""条件路由"""
if state.get("results") is None and state.get("retry_count", 0) < 3:
return "retry"
return "end"
# 构建图
graph.add_node("classify", classify_intent)
graph.add_node("plan", make_plan)
graph.add_node("execute", execute_plan)
graph.add_node("respond", generate_response)
graph.add_edge(START, "classify")
graph.add_edge("classify", "plan")
graph.add_edge("plan", "execute")
graph.add_conditional_edges(
"execute",
should_retry,
{"retry": "execute", "end": "respond"},
)
graph.add_edge("respond", END)
app = graph.compile()
6.3 人机协作状态图
在关键决策点加入人工审批：
from langgraph.types import interrupt, Command
def human_review(state: WorkflowState):
"""人工审批节点"""
result = interrupt({
"question": "请确认以下执行计划是否正确？",
"plan": state["plan"],
})
if result.get("approved"):
return {}
else:
return {"plan": result.get("revised_plan", state["plan"])}
7. 检索增强记忆（RAG + Memory）
7.1 RAG 与记忆的结合
RAG（Retrieval-Augmented Generation）是让 LLM 利用外部知识的标准方法。将 RAG 与记忆系统结合，可以实现更智能的记忆检索：
class RAGMemory:
"""带语义检索的记忆系统"""
def __init__(self, vector_store, llm):
self.vector_store = vector_store
self.llm = llm
self.short_term: list[dict] = [] # 短期记忆（当前对话）
def add_interaction(self, user_msg: str, assistant_msg: str):
"""添加一轮交互到记忆"""
# 1. 添加到短期记忆
self.short_term.append({"role": "user", "content": user_msg})
self.short_term.append({"role": "assistant", "content": assistant_msg})
# 2. 提取关键信息存入长期记忆
summary = self._extract_key_info(user_msg, assistant_msg)
if summary:
self.vector_store.add_texts(
texts=[summary],
metadatas=[{"timestamp": str(datetime.now())}],
)
def _extract_key_info(self, user_msg: str, assistant_msg: str) -> str | None:
"""用 LLM 提取对话中的关键信息"""
prompt = f"""从以下对话中提取值得长期记住的关键信息。
如果没有什么重要的信息需要记住，返回空字符串。
用户: {user_msg}
助手: {assistant_msg}
关键信息:"""
result = self.llm.invoke(prompt).content.strip()
return result if result else None
def get_context(self, current_query: str, max_tokens: int = 4000) -> list[dict]:
"""为当前查询构建上下文（混合短期+检索的长期记忆）"""
context = []
# 1. 检索相关的长期记忆
relevant_memories = self.vector_store.similarity_search(
current_query, k=3
)
if relevant_memories:
memory_text = "\n".join(f"- {m.page_content}" for m in relevant_memories)
context.append({
"role": "system",
"content": f"相关历史记忆：\n{memory_text}",
})
# 2. 添加短期记忆（对话历史）
context.extend(self.short_term[-10:]) # 最近5轮
return context
7.2 混合检索策略
结合语义搜索和关键词搜索，提高记忆检索的准确性：
class HybridRetriever:
def __init__(self, vector_store, keyword_index):
self.vector_store = vector_store
self.keyword_index = keyword_index
def retrieve(self, query: str, top_k: int = 5) -> list[dict]:
# 语义搜索
semantic_results = self.vector_store.similarity_search_with_score(query, k=top_k)
# 关键词搜索
keyword_results = self.keyword_index.search(query, k=top_k)
# 合并排序（RRF - Reciprocal Rank Fusion）
all_results = {}
for rank, (doc, score) in enumerate(semantic_results):
key = doc.page_content
all_results[key] = all_results.get(key, 0) + 1 / (60 + rank)
for rank, doc in enumerate(keyword_results):
key = doc.page_content
all_results[key] = all_results.get(key, 0) + 1 / (60 + rank)
# 按融合分数排序
sorted_results = sorted(all_results.items(), key=lambda x: x[1], reverse=True)
return [{"content": content, "score": score} for content, score in sorted_results[:top_k]]
8. 多 Agent 间记忆共享
8.1 共享记忆总线
当系统中有多个 Agent 协作时，需要一种机制来共享关键信息：
from dataclasses import dataclass, field
from datetime import datetime
import threading
@dataclass
class MemoryEvent:
source: str # 来源 Agent
content: str # 记忆内容
event_type: str # "fact" | "decision" | "observation"
timestamp: datetime = field(default_factory=datetime.now)
tags: list[str] = field(default_factory=list)
class SharedMemoryBus:
"""多 Agent 共享记忆总线"""
def __init__(self):
self._events: list[MemoryEvent] = []
self._subscribers: dict[str, list[callable]] = {}
self._lock = threading.Lock()
def publish(self, event: MemoryEvent):
"""发布一条记忆事件"""
with self._lock:
self._events.append(event)
# 通知订阅者
for callback in self._subscribers.get(event.event_type, []):
callback(event)
for callback in self._subscribers.get("*", []):
callback(event)
def subscribe(self, event_type: str, callback: callable):
"""订阅特定类型的记忆事件"""
if event_type not in self._subscribers:
self._subscribers[event_type] = []
self._subscribers[event_type].append(callback)
def query(self, tags: list[str] = None, source: str = None,
event_type: str = None, limit: int = 10) -> list[MemoryEvent]:
"""查询共享记忆"""
results = self._events
if tags:
results = [e for e in results if any(t in e.tags for t in tags)]
if source:
results = [e for e in results if e.source == source]
if event_type:
results = [e for e in results if e.event_type == event_type]
return results[-limit:]
# 使用示例
bus = SharedMemoryBus()
# 研究 Agent 发布发现
bus.publish(MemoryEvent(
source="research_agent",
content="竞品 X 刚发布了新功能，支持 AI 代码审查",
event_type="observation",
tags=["competitor", "product"],
))
# 写作 Agent 订阅竞品动态
bus.subscribe("observation", lambda e: print(f"[写作Agent] 收到情报: {e.content}"))
8.2 基于 LangGraph 的多 Agent 记忆共享
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
class SharedState(TypedDict):
messages: Annotated[list, lambda x, y: x + y] # 合并消息
research_findings: list[str] # 研究 Agent 的发现
analysis_results: list[str] # 分析 Agent 的结果
final_report: str | None # 最终输出
def research_agent(state: SharedState):
"""研究 Agent - 搜索信息"""
findings = ["发现1: 市场增长20%", "发现2: 用户需求变化"]
return {"research_findings": findings}
def analysis_agent(state: SharedState):
"""分析 Agent - 分析研究结果"""
analysis = f"基于 {len(state['research_findings'])} 条发现的分析"
return {"analysis_results": [analysis]}
def writer_agent(state: SharedState):
"""写作 Agent - 综合所有信息生成报告"""
report = f"报告：{state['research_findings']} + {state['analysis_results']}"
return {"final_report": report}
# 构建多 Agent 工作流
graph = StateGraph(SharedState)
graph.add_node("researcher", research_agent)
graph.add_node("analyst", analysis_agent)
graph.add_node("writer", writer_agent)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)
app = graph.compile()
result = app.invoke({"messages": [], "research_findings": [], "analysis_results": [], "final_report": None})
9. 记忆持久化与序列化
9.1 文件系统持久化
import json
import os
from datetime import datetime
class FileMemoryStore:
def __init__(self, storage_dir: str = "./agent_memory"):
self.storage_dir = storage_dir
os.makedirs(storage_dir, exist_ok=True)
def save(self, agent_id: str, memory: dict):
"""保存 Agent 记忆到文件"""
filepath = os.path.join(self.storage_dir, f"{agent_id}.json")
memory["_last_saved"] = datetime.now().isoformat()
with open(filepath, "w", encoding="utf-8") as f:
json.dump(memory, f, ensure_ascii=False, indent=2)
def load(self, agent_id: str) -> dict:
"""从文件加载 Agent 记忆"""
filepath = os.path.join(self.storage_dir, f"{agent_id}.json")
if not os.path.exists(filepath):
return {}
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
def list_agents(self) -> list[str]:
"""列出所有已保存的 Agent"""
return [
f.replace(".json", "")
for f in os.listdir(self.storage_dir)
if f.endswith(".json")
]
9.2 SQLite 持久化
import sqlite3
import json
class SQLiteMemoryStore:
def __init__(self, db_path: str = "memory.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS memories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
agent_id TEXT NOT NULL,
memory_type TEXT NOT NULL,
content TEXT NOT NULL,
metadata TEXT,
embedding BLOB,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.execute("""
CREATE INDEX IF NOT EXISTS idx_agent_type
ON memories(agent_id, memory_type)
""")
self.conn.commit()
def store(self, agent_id: str, memory_type: str, content: str, metadata: dict = None):
self.conn.execute(
"INSERT INTO memories (agent_id, memory_type, content, metadata) VALUES (?, ?, ?, ?)",
(agent_id, memory_type, content, json.dumps(metadata or {}))
)
self.conn.commit()
def retrieve(self, agent_id: str, memory_type: str = None, limit: int = 100) -> list[dict]:
query = "SELECT * FROM memories WHERE agent_id = ?"
params = [agent_id]
if memory_type:
query += " AND memory_type = ?"
params.append(memory_type)
query += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
cursor = self.conn.execute(query, params)
columns = [desc[0] for desc in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
9.3 LangGraph Checkpoint 持久化
LangGraph 内置了检查点机制，支持多种后端：
from langgraph.checkpoint.memory import MemorySaver # 内存
from langgraph.checkpoint.sqlite import SqliteSaver # SQLite
from langgraph.checkpoint.postgres import PostgresSaver # PostgreSQL
# 内存检查点（开发用）
memory_checkpointer = MemorySaver()
# SQLite 检查点（本地持久化）
sqlite_checkpointer = SqliteSaver.from_conn_string("./checkpoints.db")
# PostgreSQL 检查点（生产环境）
pg_checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db")
# 在编译时指定检查点
app = graph.compile(checkpointer=sqlite_checkpointer)
10. 记忆压缩与摘要策略
10.1 分层压缩
class HierarchicalCompressor:
"""分层记忆压缩器"""
def __init__(self, llm):
self.llm = llm
def compress_conversation(self, messages: list[dict], target_tokens: int = 500) -> str:
"""将对话压缩为摘要"""
if len(messages) <= 3:
return self._format_messages(messages)
# 分组：每3-5轮对话一组
chunks = self._chunk_messages(messages, chunk_size=5)
summaries = []
for chunk in chunks:
summary = self._summarize_chunk(chunk)
summaries.append(summary)
# 如果摘要仍然太长，递归压缩
combined = "\n".join(summaries)
if self._estimate_tokens(combined) > target_tokens:
return self.compress_conversation(
[{"role": "system", "content": combined}],
target_tokens
)
return combined
def _chunk_messages(self, messages: list[dict], chunk_size: int):
return [messages[i:i + chunk_size] for i in range(0, len(messages), chunk_size)]
def _summarize_chunk(self, chunk: list[dict]) -> str:
text = "\n".join(f"{m['role']}: {m['content']}" for m in chunk)
prompt = f"请用1-2句话总结以下对话的关键信息：\n{text}"
return self.llm.invoke(prompt).content
def _format_messages(self, messages: list[dict]) -> str:
return "\n".join(f"{m['role']}: {m['content']}" for m in messages)
def _estimate_tokens(self, text: str) -> int:
return len(text) // 2 # 粗略估计
10.2 增量摘要
不是一次性压缩全部历史，而是增量更新摘要：
class IncrementalSummarizer:
def __init__(self, llm, max_buffer: int = 10):
self.llm = llm
self.max_buffer = max_buffer
self.summary: str = ""
self.buffer: list[dict] = []
def add(self, message: dict):
self.buffer.append(message)
if len(self.buffer) >= self.max_buffer:
self._flush_buffer()
def _flush_buffer(self):
if not self.buffer:
return
new_text = "\n".join(f"{m['role']}: {m['content']}" for m in self.buffer)
if self.summary:
prompt = f"""当前摘要：{self.summary}
新的对话内容：
{new_text}
请更新摘要，融合新旧信息，保持简洁："""
else:
prompt = f"请总结以下对话：\n{new_text}"
self.summary = self.llm.invoke(prompt).content
self.buffer = []
def get_summary(self) -> str:
self._flush_buffer() # 确保缓冲区被处理
return self.summary
11. 实战：构建有记忆能力的个人助手
下面我们把前面学到的知识整合起来，构建一个完整的有记忆能力的个人助手。
11.1 架构设计
用户输入
│
▼
┌─────────────────────────────────┐
│ Personal Assistant │
│ │
│ ┌───────────┐ ┌────────────┐ │
│ │ 短期记忆 │ │ 工作记忆 │ │
│ │ (对话历史) │ │ (任务状态) │ │
│ └─────┬─────┘ └──────┬─────┘ │
│ │ │ │
│ ┌─────┴───────────────┴─────┐ │
│ │ 记忆检索层 │ │
│ │ (语义搜索 + 关键词匹配) │ │
│ └─────────────┬─────────────┘ │
│ │ │
│ ┌─────────────┴─────────────┐ │
│ │ 长期记忆存储 │ │
│ │ (ChromaDB / SQLite) │ │
│ └───────────────────────────┘ │
└─────────────────────────────────┘
11.2 完整代码实现
"""
有记忆能力的个人助手 - 完整实现
"""
from typing import TypedDict, Annotated
from datetime import datetime
import chromadb
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
import json
import uuid
# ========== 1. 定义状态 ==========
class AssistantState(TypedDict):
messages: Annotated[list, add_messages] # 对话历史
user_profile: dict # 用户画像
current_task: dict | None # 当前任务
retrieved_memories: list[str] # 检索到的记忆
metadata: dict # 元数据
# ========== 2. 记忆管理器 ==========
class MemoryManager:
def __init__(self):
self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
self.chroma_client = chromadb.PersistentClient(path="./assistant_memory")
self.collection = self.chroma_client.get_or_create_collection(
name="personal_memories",
)
self.llm = ChatOpenAI(model="gpt-4o-mini")
def store_memory(self, content: str, memory_type: str, metadata: dict = None):
"""存储一条记忆"""
self.collection.add(
ids=[str(uuid.uuid4())],
documents=[content],
metadatas=[{
"type": memory_type,
"timestamp": datetime.now().isoformat(),
**(metadata or {}),
}],
)
def recall(self, query: str, top_k: int = 5) -> list[str]:
"""检索相关记忆"""
results = self.collection.query(
query_texts=[query],
n_results=top_k,
)
return results["documents"][0] if results["documents"] else []
def extract_and_store(self, user_msg: str, assistant_msg: str):
"""从对话中提取关键信息并存储"""
prompt = f"""分析以下对话，提取值得长期记住的信息。
返回 JSON 数组，每项包含 "content" 和 "type" 字段。
type 可选值: "preference"（偏好）, "fact"（事实）, "instruction"（指令）, "event"（事件）
如果没有值得记住的信息，返回空数组 []
用户: {user_msg}
助手: {assistant_msg}
JSON:"""
try:
result = self.llm.invoke(prompt).content
# 解析 JSON
items = json.loads(result)
for item in items:
self.store_memory(item["content"], item["type"])
except (json.JSONDecodeError, Exception):
pass # 解析失败时静默处理
# ========== 3. 定义节点函数 ==========
memory_manager = MemoryManager()
def retrieve_memories(state: AssistantState) -> dict:
"""检索相关记忆"""
if not state["messages"]:
return {"retrieved_memories": []}
last_msg = state["messages"][-1].content
memories = memory_manager.recall(last_msg)
return {"retrieved_memories": memories}
def generate_response(state: AssistantState) -> dict:
"""生成回复"""
llm = ChatOpenAI(model="gpt-4o-mini")
# 构建系统提示
system_parts = ["你是一个有记忆能力的个人助手。"]
# 添加用户画像
if state.get("user_profile"):
profile = state["user_profile"]
system_parts.append(f"用户信息：{json.dumps(profile, ensure_ascii=False)}")
# 添加检索到的记忆
if state.get("retrieved_memories"):
memory_text = "\n".join(f"- {m}" for m in state["retrieved_memories"])
system_parts.append(f"相关记忆：\n{memory_text}")
system_prompt = "\n\n".join(system_parts)
# 调用 LLM
messages = [{"role": "system", "content": system_prompt}] + [
{"role": m.type, "content": m.content} for m in state["messages"]
]
response = llm.invoke(messages)
# 异步存储记忆（在实际应用中可以放到后台线程）
if state["messages"]:
user_msg = state["messages"][-1].content
memory_manager.extract_and_store(user_msg, response.content)
return {"messages": [response]}
# ========== 4. 构建图 ==========
graph = StateGraph(AssistantState)
graph.add_node("retrieve_memories", retrieve_memories)
graph.add_node("generate_response", generate_response)
graph.add_edge(START, "retrieve_memories")
graph.add_edge("retrieve_memories", "generate_response")
graph.add_edge("generate_response", END)
# 持久化检查点
checkpointer = SqliteSaver.from_conn_string("./assistant_checkpoints.db")
app = graph.compile(checkpointer=checkpointer)
# ========== 5. 运行助手 ==========
def chat(user_input: str, thread_id: str = "default"):
"""与助手对话"""
config = {"configurable": {"thread_id": thread_id}}
result = app.invoke(
{"messages": [{"role": "user", "content": user_input}]},
config=config,
)
return result["messages"][-1].content
# 交互循环
if __name__ == "__main__":
print("个人助手已启动（输入 'quit' 退出）")
print("=" * 50)
thread_id = str(uuid.uuid4())
while True:
user_input = input("\n你: ").strip()
if user_input.lower() in ("quit", "exit", "q"):
print("再见！")
break
if not user_input:
continue
try:
response = chat(user_input, thread_id)
print(f"\n助手: {response}")
except Exception as e:
print(f"\n错误: {e}")
11.3 测试场景
# 场景1：告诉助手你的偏好
chat("我叫王明，是一个 Python 开发者，喜欢简洁的代码风格")
# 助手: "你好王明！很高兴认识你。我会记住你的偏好——简洁的 Python 代码风格。"
# 场景2：跨会话记忆
chat("你还记得我的名字吗？")
# 助手: "你叫王明，是一位 Python 开发者，偏好简洁的代码风格。"
# 场景3：基于记忆的个性化回复
chat("帮我写一个文件读取函数")
# 助手: *（会用简洁的 Python 风格编写，因为记住了你的偏好）*
# 场景4：教学记忆
chat("以后给我写代码时都加上类型注解")
# 助手: "好的，我记住了。以后代码都会加上类型注解。"
# （这条指令会被存储为长期记忆）
11.4 运行前准备
# 安装依赖
pip install langgraph langchain-openai langchain-community chromadb
# 设置环境变量
export OPENAI_API_KEY="your-api-key"
12. 总结与最佳实践
核心要点回顾
- 三层记忆架构：短期记忆（对话历史）、长期记忆（向量数据库）、工作记忆（任务状态）各司其职
- 对话管理策略：滑动窗口、Token 限制、摘要压缩、重要性评分，按需选择或组合
- 向量数据库是长期记忆的核心存储，支持语义检索
- LangGraph 提供了生产级的状态管理方案，支持检查点持久化和多 Agent 协作
- RAG + Memory 的结合让记忆检索更智能
- 记忆压缩是控制 token 消耗的关键技术
最佳实践
设计原则：
- 记忆系统应该对用户透明——让用户知道 Agent 记住了什么
- 提供"遗忘"机制——允许用户删除特定记忆
- 区分记忆的生命周期——不是所有信息都值得长期记住
性能优化：
- 批量写入记忆，避免频繁的单条写入
- 定期清理过期或低价值的记忆
- 使用元数据过滤缩小检索范围
安全考虑：
- 记忆中不应存储敏感信息（密码、密钥等）
- 不同用户的记忆应该严格隔离
- 实现记忆的访问控制
扩展建议：
- 为记忆添加时间衰减权重——越近的记忆权重越高
- 实现记忆的主动回忆——在相关话题出现时主动提及
- 支持多模态记忆——图片、文件等非文本信息
本教程由 AI 辅助生成，内容基于 LangChain/LangGraph 等开源项目的公开文档整理，仅供学习参考。
