第二十章：Agent 的记忆与状态管理

第二十章：Agent 的记忆与状态管理#

        mindmap
  root((Agent记忆与状态))
    为什么需要记忆
      对话连贯性
      经验积累
      个性化服务
    记忆分类
      感知记忆
      短期记忆
      长期记忆
      情景记忆
      语义记忆
      程序记忆
    向量数据库
      Chroma
      Pinecone
      Weaviate
      Milvus
    记忆生命周期
      存储
      检索
      遗忘
      整合
    状态管理
      有限状态机
      状态图
      检查点

“没有记忆的 Agent 就像没有笔记本的侦探——每次都要从头开始。”

20.1 为什么 Agent 需要记忆#

LLM 本身是无状态的——每次调用都是独立的，不记得之前的对话。Agent 需要记忆系统来：

  • 维持对话连贯性：记住用户之前说了什么

  • 积累经验：从过去的成功和失败中学习

  • 个性化服务：记住用户的偏好和习惯

  • 长期任务：跨多次交互完成复杂任务

20.2 记忆的分类#

Agent 记忆系统
├── 感知记忆（Sensory Memory）
│   └── 当前输入的原始信息
├── 短期记忆（Short-term / Working Memory）
│   └── 当前对话的上下文窗口
├── 长期记忆（Long-term Memory）
│   ├── 情景记忆（Episodic Memory）
│   │   └── 过去的具体经历和对话
│   ├── 语义记忆（Semantic Memory）
│   │   └── 知识、事实、概念
│   └── 程序记忆（Procedural Memory）
│       └── 技能、习惯、工作流程

各类记忆的实现#

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class Memory:
    content: str
    memory_type: str  # "episodic", "semantic", "procedural"
    timestamp: datetime = field(default_factory=datetime.utcnow)
    importance: float = 0.5  # 0-1
    access_count: int = 0
    last_accessed: Optional[datetime] = None
    metadata: dict = field(default_factory=dict)

class MemorySystem:
    def __init__(self):
        self.short_term: list[dict] = []  # 当前对话
        self.long_term: list[Memory] = []  # 持久化记忆
        self.max_short_term = 20  # 短期记忆容量
    
    def add_to_short_term(self, message: dict):
        """添加到短期记忆（对话历史）"""
        self.short_term.append(message)
        if len(self.short_term) > self.max_short_term:
            # 溢出的内容总结后存入长期记忆
            overflow = self.short_term[:5]
            self.consolidate(overflow)
            self.short_term = self.short_term[5:]
    
    def add_to_long_term(self, content: str, memory_type: str, importance: float = 0.5):
        """添加到长期记忆"""
        memory = Memory(
            content=content,
            memory_type=memory_type,
            importance=importance
        )
        self.long_term.append(memory)
    
    def consolidate(self, messages: list[dict]):
        """将短期记忆整合为长期记忆"""
        summary = self.summarize(messages)
        self.add_to_long_term(summary, "episodic", importance=0.3)
    
    def recall(self, query: str, top_k: int = 5) -> list[Memory]:
        """根据查询检索相关记忆"""
        # 使用向量相似度搜索
        relevant = self.vector_search(query, top_k)
        # 更新访问信息
        for memory in relevant:
            memory.access_count += 1
            memory.last_accessed = datetime.utcnow()
        return relevant
    
    def forget(self, threshold: float = 0.1):
        """遗忘不重要的记忆"""
        self.long_term = [
            m for m in self.long_term
            if m.importance > threshold or m.access_count > 3
        ]

20.3 向量数据库在记忆系统中的应用#

主流向量数据库对比#

数据库

类型

特点

适用场景

Chroma

嵌入式

轻量、Python 原生

原型、小项目

Pinecone

云服务

全托管、高性能

生产环境

Weaviate

自托管/云

GraphQL API、多模态

企业级

Milvus

自托管

高性能、分布式

大规模部署

pgvector

PostgreSQL 扩展

与现有 PG 集成

已有 PG 的项目

使用 Chroma 构建记忆#

import chromadb
from chromadb.utils import embedding_functions

# 初始化
ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="your-key",
    model_name="text-embedding-3-small"
)

client = chromadb.PersistentClient(path="./agent_memory")

# 创建不同类型的记忆集合
episodic = client.get_or_create_collection(
    name="episodic_memory",
    embedding_function=ef
)

semantic = client.get_or_create_collection(
    name="semantic_memory",
    embedding_function=ef
)

# 存储记忆
episodic.add(
    documents=["用户喜欢用 Python 写后端，偏好 FastAPI 框架"],
    metadatas=[{"type": "preference", "importance": 0.8, "date": "2026-03-01"}],
    ids=["mem_001"]
)

# 检索记忆
results = episodic.query(
    query_texts=["用户的技术偏好是什么？"],
    n_results=5
)
print(results["documents"])

20.4 记忆的存储、检索、遗忘与整合#

记忆生命周期#

新信息 → 评估重要性 → 存储
                         ↓
                    定期整合（合并相似记忆）
                         ↓
                    检索使用（更新访问频率）
                         ↓
                    遗忘（清理低价值记忆）

重要性评估#

async def assess_importance(content: str, context: str) -> float:
    """使用 LLM 评估记忆的重要性"""
    prompt = f"""
    请评估以下信息的重要性（0-1分）：
    
    信息：{content}
    上下文：{context}
    
    评估标准：
    - 用户明确表达的偏好：0.8-1.0
    - 项目相关的技术决策：0.7-0.9
    - 一般性对话内容：0.2-0.4
    - 临时性信息：0.1-0.2
    
    只返回一个数字。
    """
    score = await llm.generate(prompt)
    return float(score.strip())

20.5 状态管理#

有限状态机（FSM）#

from enum import Enum

class AgentState(Enum):
    IDLE = "idle"
    THINKING = "thinking"
    USING_TOOL = "using_tool"
    WAITING_FOR_USER = "waiting_for_user"
    ERROR = "error"

class AgentFSM:
    def __init__(self):
        self.state = AgentState.IDLE
        self.transitions = {
            AgentState.IDLE: [AgentState.THINKING],
            AgentState.THINKING: [AgentState.USING_TOOL, AgentState.WAITING_FOR_USER, AgentState.IDLE],
            AgentState.USING_TOOL: [AgentState.THINKING, AgentState.ERROR],
            AgentState.WAITING_FOR_USER: [AgentState.THINKING],
            AgentState.ERROR: [AgentState.IDLE, AgentState.THINKING],
        }
    
    def transition(self, new_state: AgentState):
        if new_state in self.transitions.get(self.state, []):
            self.state = new_state
        else:
            raise ValueError(f"Invalid transition: {self.state} → {new_state}")

LangGraph 检查点#

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 创建带检查点的图
memory = MemorySaver()
graph = StateGraph(AgentState)

# ... 定义节点和边 ...

app = graph.compile(checkpointer=memory)

# 运行时指定 thread_id，自动保存和恢复状态
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [user_message]}, config)

# 下次调用同一 thread_id，自动恢复之前的状态
result2 = app.invoke({"messages": [another_message]}, config)

20.6 实战：构建带长期记忆的 Agent#

"""完整示例：带长期记忆的个人助手 Agent"""
import chromadb
from openai import OpenAI
from datetime import datetime

class MemoryAgent:
    def __init__(self):
        self.llm = OpenAI()
        self.chroma = chromadb.PersistentClient(path="./memory_db")
        self.memories = self.chroma.get_or_create_collection("agent_memories")
        self.conversation_history = []
    
    def remember(self, content: str, importance: float = 0.5):
        """存储新记忆"""
        self.memories.add(
            documents=[content],
            metadatas=[{
                "importance": importance,
                "timestamp": datetime.now().isoformat(),
                "access_count": 0
            }],
            ids=[f"mem_{datetime.now().timestamp()}"]
        )
    
    def recall(self, query: str, n: int = 5) -> list[str]:
        """检索相关记忆"""
        if self.memories.count() == 0:
            return []
        results = self.memories.query(query_texts=[query], n_results=min(n, self.memories.count()))
        return results["documents"][0] if results["documents"] else []
    
    def chat(self, user_message: str) -> str:
        """与用户对话，自动使用记忆"""
        # 1. 检索相关记忆
        relevant_memories = self.recall(user_message)
        memory_context = "\n".join(f"- {m}" for m in relevant_memories) if relevant_memories else "无相关记忆"
        
        # 2. 构建 Prompt
        system_prompt = f"""你是一个有记忆的个人助手。

相关记忆：
{memory_context}

请基于记忆和当前对话回答用户。如果用户分享了重要信息（偏好、事实、决定），
请在回答末尾用 [REMEMBER: ...] 标记需要记住的内容。"""
        
        self.conversation_history.append({"role": "user", "content": user_message})
        
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                *self.conversation_history[-10:]  # 最近10轮对话
            ]
        )
        
        reply = response.choices[0].message.content
        
        # 3. 提取需要记忆的内容
        if "[REMEMBER:" in reply:
            memory_content = reply.split("[REMEMBER:")[1].split("]")[0].strip()
            self.remember(memory_content, importance=0.8)
            reply = reply.split("[REMEMBER:")[0].strip()
        
        self.conversation_history.append({"role": "assistant", "content": reply})
        return reply

# 使用
agent = MemoryAgent()
print(agent.chat("我叫 Walter，是一名 Python 开发者"))
# → "你好 Walter！很高兴认识你..." [自动记住名字和职业]

print(agent.chat("我最近在学 Rust"))
# → "学 Rust 是个好选择！作为 Python 开发者..." [记住了之前的信息]

20.7 本章小结#

记忆系统是 Agent 从"一次性工具"进化为"持续助手"的关键。通过合理设计短期记忆、长期记忆和状态管理，Agent 可以提供更个性化、更连贯的服务。

核心要点：

  1. 分层记忆：不同类型的信息用不同的记忆机制

  2. 向量检索：用语义相似度而非关键词匹配来检索记忆

  3. 遗忘机制：不是所有信息都值得记住

  4. 状态持久化：使用检查点保存和恢复 Agent 状态

思考题

  1. Agent 的记忆系统和人类的记忆有什么相似和不同？

  2. 如何处理记忆中的错误信息（用户纠正了之前的说法）？

  3. 记忆系统的隐私问题如何解决？

目录