AI Agent 的记忆机制：短期记忆与长期记忆

为什么 Agent 需要记忆？

想象你在和一个健忘的朋友聊天——每句话都要重复上下文，效率极低。AI Agent 也一样，没有记忆就无法处理复杂的多轮任务。

记忆是 Agent 的"经验"，让它能从过去的学习中成长。

记忆的分类

┌─────────────────────────────────────────┐
│              Agent 记忆系统               │
├─────────────────┬───────────────────────┤
│    短期记忆      │       长期记忆         │
│  (Working Memory)│    (Long-term Memory)  │
├─────────────────┼───────────────────────┤
│ • 对话上下文     │ • 用户偏好            │
│ • 当前任务状态   │ • 历史交互记录        │
│ • 临时变量      │ • 知识库             │
│ • 最近 N 轮对话  │ • 学习到的模式        │
└─────────────────┴───────────────────────┘

短期记忆实现

1. ConversationBufferMemory

from langchain.memory import ConversationBufferMemory

# 最简单的记忆：保存所有对话
memory = ConversationBufferMemory()
memory.save_context(
    {"input": "你好，我想订机票"},
    {"output": "好的，请告诉我目的地和日期"}
)

# 查看记忆内容
print(memory.load_memory_variables({}))
# {'history': 'Human: 你好，我想订机票\nAI: 好的，请告诉我目的地和日期'}

2. ConversationBufferWindowMemory

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近 k 轮对话，防止上下文过长
memory = ConversationBufferWindowMemory(k=3)

3. ConversationSummaryMemory

from langchain.memory import ConversationSummaryMemory
from langchain_openai import OpenAI

# 自动总结历史对话
llm = OpenAI(temperature=0)
memory = ConversationSummaryMemory(llm=llm)

memory.save_context(
    {"input": "你好"},
    {"output": "你好！有什么可以帮助你？"}
)
memory.save_context(
    {"input": "我想学习Python"},
    {"output": "太好了！你有多少编程经验？"}
)

print(memory.load_memory_variables({}))
# 输出："人类向AI打招呼，表示想学习Python，AI询问其编程经验"

长期记忆：向量数据库

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.memory import VectorStoreRetrieverMemory

# 初始化向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

# 创建基于向量检索的记忆
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever()
)

# 存储长期记忆
memory.save_context(
    {"input": "用户偏好"},
    {"output": "用户喜欢Python，讨厌Java，对AI感兴趣"}
)

# 检索相关记忆
relevant_memories = memory.load_memory_variables({"input": "推荐编程语言"})
# 返回：用户喜欢Python的信息

记忆系统的架构设计

class AdvancedMemorySystem:
    """高级记忆管理系统"""
    
    def __init__(self):
        # 短期记忆：当前会话
        self.short_term = ConversationBufferWindowMemory(k=5)
        
        # 中期记忆：当天会话摘要
        self.mid_term = ConversationSummaryMemory(llm=llm)
        
        # 长期记忆：向量数据库
        self.long_term = VectorStoreRetrieverMemory(
            retriever=vectorstore.as_retriever()
        )
    
    def remember(self, user_input: str, ai_output: str):
        """保存记忆到各层级"""
        self.short_term.save_context(
            {"input": user_input},
            {"output": ai_output}
        )
        self.mid_term.save_context(
            {"input": user_input},
            {"output": ai_output}
        )
        self.long_term.save_context(
            {"input": user_input},
            {"output": ai_output}
        )
    
    def recall(self, query: str) -> str:
        """检索相关记忆"""
        # 先查短期记忆
        short = self.short_term.load_memory_variables({})
        
        # 再查长期记忆
        long = self.long_term.load_memory_variables({"input": query})
        
        return f"短期记忆：{short}\n长期记忆：{long}"

记忆的遗忘与更新

import time
from datetime import datetime

class TimeDecayMemory:
    """带时间衰减的记忆"""
    
    def __init__(self, decay_rate=0.95):
        self.memories = []
        self.decay_rate = decay_rate
    
    def add(self, content: str, importance: float = 1.0):
        self.memories.append({
            "content": content,
            "importance": importance,
            "timestamp": datetime.now()
        })
    
    def recall(self) -> List[str]:
        """根据时间和重要性召回记忆"""
        now = datetime.now()
        
        weighted_memories = []
        for mem in self.memories:
            # 计算时间衰减
            hours_passed = (now - mem["timestamp"]).total_seconds() / 3600
            time_weight = self.decay_rate ** hours_passed
            
            final_weight = mem["importance"] * time_weight
            weighted_memories.append((final_weight, mem["content"]))
        
        # 按权重排序，返回 Top-K
        weighted_memories.sort(reverse=True)
        return [m[1] for m in weighted_memories[:5]]

实战：个性化推荐 Agent

class PersonalizedAgent:
    def __init__(self):
        self.memory = AdvancedMemorySystem()
        self.user_profile = {}
    
    def chat(self, user_input: str) -> str:
        # 1. 检索相关记忆
        context = self.memory.recall(user_input)
        
        # 2. 构建提示
        prompt = f"""
        用户历史偏好：{context}
        用户画像：{self.user_profile}
        用户输入：{user_input}
        请根据用户历史偏好给出个性化回答。
        """
        
        # 3. 生成回复
        response = llm(prompt)
        
        # 4. 保存记忆
        self.memory.remember(user_input, response)
        
        # 5. 更新用户画像
        self._update_profile(user_input, response)
        
        return response

总结

记忆系统让 Agent 从" Stateless “变成” Stateful "：

• 短期记忆：处理当前会话
• 长期记忆：积累用户偏好
• 向量检索：高效召回相关信息

下一步：学习多 Agent 协作，让多个 Agent 共享记忆、协同工作。