一、长短期记忆混合存储架构方案流程
前言：需要先按照短期记忆的持久性和数据特点做分层，划分哪些需要长期存储那些不需要，这样更合理，当然本文并没有过多讲解，可以参考文章013RAG上下文工程-如何预防上下文窗口爆掉？来具体了解短期记忆有哪些，如何识别处理。（必读）
L0（窗口层）:当前会话（短期记忆）中的关键用户特征信息---来自于短期记忆即当前用户会话/当前上下文（一般是最近的N轮对话）
L1（摘要层）:当前N轮会话（短期记忆）前的所有会话
L2（会话全局摘要）：当前会话的所有信息
混合模式架构图（主流）混合记忆（主流Agent架构），大多数实用Agent会将两者结合：
工作流程：
用户输入 → 从长期记忆中检索相关记忆 → 合并到短期记忆（当前上下文）
Agent执行推理/动作 → 将新的观察结果暂存于短期记忆
当短期记忆积累到阈值或任务结束 → 触发记忆整理 → 将重要片段写入长期记忆
代表框架：
LangChain 的
Memory模块（提供ConversationBufferMemory短期，VectorStoreRetrieverMemory长期）AutoGPT / BabyAGI 的任务队列 + 向量记忆
MemGPT：显式模拟操作系统中的分页机制，在无限上下文与有限窗口间调度记忆
流程3详解（又可以分为主动触发和被动触发）
二、主被动触发写入长期记忆
1. 主动触发（Agent自主决策写入）
定义：Agent根据当前任务重要性、用户指令或自身判断，主动决定将某些短期记忆片段存入长期记忆。
实现方式：
关键点检测：模型在推理时判断某个信息（如用户偏好、重要事实、任务结果）值得长期保存，调用
save_to_long_term_memory()工具。指令驱动：用户明确说“记住这个”或“把...保存下来”。
元认知提示：在System Prompt中要求模型“当你认为某个信息对后续任务重要时，主动记住它”。
特点：
优点：智能、按需存储，避免冗余
缺点：依赖模型判断能力，可能漏掉重要信息或存储噪声
典型场景：个人助理Agent（记住用户习惯）、任务规划Agent（记住成功的策略）
伪代码实例
class ActiveMemoryAgent:
def process(self, user_input):
# 模型推理时可能输出一个特殊的"memory_action"
response = llm.generate(
user_input,
system_prompt="""
如果你认为某个信息对长期有帮助，输出：
MEMORIZE: {内容}
然后再回复用户。
"""
)
if "MEMORIZE:" in response:
content = extract_memorize_content(response)
vector = embed(content)
long_term_db.insert(vector, content)
# 再从短期中移除已存储的部分
self.short_term = filter_out_remembered(self.short_term)2. 被动触发（系统规则自动写入）
定义：由系统预设规则自动触发记忆整理，Agent本身不参与决策。
常见触发条件：
容量阈值：短期记忆超过N条或K个tokens
时间阈值：每M分钟/轮次自动整理
任务边界：调用
task_complete()或进入空闲状态会话结束：用户关闭对话或切换话题
实现方式：
固定策略：直接调用总结算法（如LLM summarization、textrank）
滑动窗口：将最早的一半短期记忆压缩后存入长期
优先级队列：根据信息重要性评分（如TF-IDF、注意力分数）只保留高分片段
特点：
优点：简单可靠，保证长期记忆不丢失核心信息
缺点：可能存储重复或低价值内容，增加检索负担
典型场景：客服机器人（记录完整会话）、代码辅助Agent（存储调试历史）
伪代码实例
class PassiveMemoryAgent:
short_term = [] # 最多保留20条
def add_interaction(self, user_msg, agent_msg):
self.short_term.append((user_msg, agent_msg))
if len(self.short_term) >= 20: # 阈值触发
self._trigger_memory_consolidation()
def _trigger_memory_consolidation(self):
# 被动整理：取前10条，生成摘要
old_memories = self.short_term[:10]
summary = llm.summarize(old_memories)
vector = embed(summary)
long_term_db.insert(vector, summary)
# 移除已整理的部分
self.short_term = self.short_term[10:]混合模式（主流方案实现）大多数成熟的Agent框架会结合两者：
三、主被动触发技术对比总结
四、实际框架中的应用
LangChain：
ConversationSummaryMemory（被动，对话轮数触发）、VectorStoreRetrieverMemory（通常主动调用add_documents）MemGPT：类似操作系统分页，当上下文满时被动触发页面换出，但Agent也可主动调用
core_memory_append。AutoGPT：任务执行过程中主动保存重要结果，任务结束后被动压缩整个会话。
五、主被动触发总结
核心记忆（用户画像、关键决策）→ 主动触发
过程记忆（对话历史、调试日志）→ 被动触发
六、代码实现
基于LangChain的混合记忆Agent完整实现
以下是一个完整的、可直接运行的混合记忆Agent实现，包含主动触发（Agent自主决定存储）和被动触发（系统阈值自动整理）两种长期记忆写入机制。
1. 环境准备
pip install langchain langchain-community langchain-openai chromadb faiss-cpu tiktoken2. 完整代码实现
"""
混合记忆Agent - 同时支持主动触发和被动触发的长期记忆
"""
import json
import hashlib
from typing import Dict, Any, List, Optional
from datetime import datetime
from dataclasses import dataclass, field
from enum import Enum
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.memory import ConversationBufferWindowMemory, VectorStoreRetrieverMemory
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate
from langchain_community.vectorstores import Chroma
from langchain_core.tools import tool
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
# ============================================================
# 第一部分：记忆触发类型定义
# ============================================================
class TriggerType(Enum):
"""记忆触发类型"""
ACTIVE = "active" # 主动触发：Agent自主决策写入
PASSIVE = "passive" # 被动触发：系统阈值自动写入
@dataclass
class MemoryEntry:
"""记忆条目"""
content: str
trigger_type: TriggerType
timestamp: datetime
importance_score: float = 0.5 # 重要性评分，用于遗忘机制
# ============================================================
# 第二部分：混合记忆管理器
# ============================================================
class HybridMemoryManager:
"""
混合记忆管理器
- 管理短期记忆（滑动窗口）
- 管理长期记忆（向量数据库）
- 实现主动和被动两种写入机制
"""
def __init__(
self,
short_term_window_size: int = 10, # 短期记忆保留轮数
passive_trigger_threshold: int = 8, # 被动触发阈值（超过此轮数触发整理）
long_term_k: int = 3, # 长期记忆检索数量
persist_directory: str = "./hybrid_memory_db"
):
# 1. 短期记忆（滑动窗口）
self.short_term_memory = ConversationBufferWindowMemory(
memory_key="chat_history",
input_key="input",
k=short_term_window_size,
return_messages=True
)
# 2. 长期记忆（向量数据库）
self.embeddings = OpenAIEmbeddings()
self.vector_store = Chroma(
collection_name="long_term_memory",
embedding_function=self.embeddings,
persist_directory=persist_directory
)
retriever = self.vector_store.as_retriever(search_kwargs={"k": long_term_k})
self.long_term_memory = VectorStoreRetrieverMemory(
retriever=retriever,
memory_key="long_term_context"
)
# 3. 记忆管理配置
self.short_term_window_size = short_term_window_size
self.passive_trigger_threshold = passive_trigger_threshold
# 4. 记忆跟踪
self.conversation_count = 0
self.important_keywords = ["记住", "我的名字是", "我喜欢", "我偏好", "important", "remember"]
print(f"✅ 混合记忆管理器初始化完成")
print(f" - 短期记忆窗口: {short_term_window_size}轮")
print(f" - 被动触发阈值: {passive_trigger_threshold}轮")
print(f" - 长期记忆存储: {persist_directory}")
def _is_important_content(self, content: str) -> bool:
"""判断内容是否重要（用于主动触发决策）"""
content_lower = content.lower()
for keyword in self.important_keywords:
if keyword.lower() in content_lower:
return True
return False
def _generate_summary(self, conversations: List[tuple]) -> str:
"""生成对话摘要（用于被动触发时的记忆压缩）"""
if not conversations:
return ""
# 简化版摘要：直接拼接，实际可用LLM生成精炼摘要
summary_parts = []
for user_msg, ai_msg in conversations:
if len(user_msg) > 100:
user_msg = user_msg[:100] + "..."
if len(ai_msg) > 100:
ai_msg = ai_msg[:100] + "..."
summary_parts.append(f"用户: {user_msg}\nAI: {ai_msg}")
return "\n---\n".join(summary_parts)
def active_save(self, content: str, metadata: Dict = None) -> bool:
"""
主动触发：Agent主动决定保存重要信息到长期记忆
"""
print(f"\n🔵 [主动触发] 正在保存重要信息到长期记忆...")
print(f" 内容: {content[:100]}...")
try:
# 添加时间戳和元数据
enriched_content = f"[记忆时间: {datetime.now().isoformat()}] {content}"
# 保存到向量数据库
# VectorStoreRetrieverMemory 使用 save_context 方法
self.long_term_memory.save_context(
{"input": content},
{"output": f"已记住: {content[:50]}..."}
)
print(f" ✅ 主动保存成功")
return True
except Exception as e:
print(f" ❌ 主动保存失败: {e}")
return False
def passive_consolidate(self) -> bool:
"""
被动触发：当短期记忆达到阈值时，自动整理并压缩到长期记忆
"""
print(f"\n🟢 [被动触发] 开始记忆整理与压缩...")
try:
# 1. 获取当前短期记忆中的对话历史
current_memory = self.short_term_memory.load_memory_variables({})
chat_history = current_memory.get("chat_history", [])
if len(chat_history) < self.passive_trigger_threshold:
print(f" 当前记忆轮数 {len(chat_history)} < 阈值 {self.passive_trigger_threshold}，跳过整理")
return False
# 2. 提取需要压缩的对话（最早的50%）
compress_count = len(chat_history) // 2
to_compress = chat_history[:compress_count]
# 3. 转换为可存储格式并生成摘要
conversations = []
for i in range(0, len(to_compress), 2):
if i+1 < len(to_compress):
user_msg = to_compress[i].content if hasattr(to_compress[i], 'content') else str(to_compress[i])
ai_msg = to_compress[i+1].content if hasattr(to_compress[i+1], 'content') else str(to_compress[i+1])
conversations.append((user_msg, ai_msg))
if conversations:
summary = self._generate_summary(conversations)
# 4. 存储到长期记忆
self.long_term_memory.save_context(
{"input": f"[会话摘要] {summary}"},
{"output": f"已压缩存储 {len(conversations)} 轮对话"}
)
print(f" ✅ 被动整理完成: 压缩了 {len(conversations)} 轮对话到长期记忆")
return True
except Exception as e:
print(f" ❌ 被动整理失败: {e}")
return False
def load_context(self, query: str) -> str:
"""
加载相关记忆（短期+长期）
"""
# 1. 加载短期记忆
short_term = self.short_term_memory.load_memory_variables({})
short_term_context = short_term.get("chat_history", [])
# 2. 加载长期记忆（基于当前查询）
long_term = self.long_term_memory.load_memory_variables({"prompt": query})
long_term_context = long_term.get("long_term_context", "")
return {
"short_term": short_term_context,
"long_term": long_term_context
}
def add_interaction(self, user_input: str, ai_response: str, force_active: bool = False):
"""
添加一次交互到短期记忆，并检查是否需要触发记忆整理
"""
# 1. 保存到短期记忆
self.short_term_memory.save_context(
{"input": user_input},
{"output": ai_response}
)
self.conversation_count += 1
# 2. 主动触发检测：如果用户明确要求记住，或内容包含重要关键词
if force_active or self._is_important_content(user_input):
self.active_save(user_input)
# 3. 被动触发检测：如果短期记忆轮数超过阈值
current_memory = self.short_term_memory.load_memory_variables({})
chat_history = current_memory.get("chat_history", [])
if len(chat_history) >= self.passive_trigger_threshold:
self.passive_consolidate()
def get_memory_stats(self) -> Dict:
"""获取记忆统计信息"""
current_memory = self.short_term_memory.load_memory_variables({})
chat_history = current_memory.get("chat_history", [])
return {
"short_term_count": len(chat_history),
"short_term_max": self.short_term_window_size,
"passive_threshold": self.passive_trigger_threshold,
"total_interactions": self.conversation_count
}
# ============================================================
# 第三部分：带记忆的Agent（使用ReAct架构）
# ============================================================
def create_memory_agent():
"""
创建一个带有混合记忆的ReAct Agent
Agent可以主动调用工具来保存重要记忆
"""
# 初始化记忆管理器
memory_manager = HybridMemoryManager(
short_term_window_size=10,
passive_trigger_threshold=6, # 每6轮对话触发一次被动整理
persist_directory="./hybrid_memory_db"
)
# 定义主动记忆工具（Agent可主动调用）
@tool
def remember_important_info(info: str) -> str:
"""
主动记住重要信息。
当用户分享重要信息（如姓名、偏好、重要事实）时，调用此工具保存。
Args:
info: 需要记住的重要信息内容
"""
success = memory_manager.active_save(info)
if success:
return f"✅ 已记住: {info[:100]}..."
return "❌ 记住失败"
@tool
def recall_memories(query: str) -> str:
"""
回忆相关的历史记忆。
当需要回忆用户之前分享的信息时，调用此工具查询。
Args:
query: 想要回忆的关键词或问题
"""
context = memory_manager.load_context(query)
long_term = context.get("long_term", "")
if long_term:
return f"📖 相关历史记忆:\n{long_term}"
return "没有找到相关的历史记忆"
# 定义其他工具
@tool
def get_current_time() -> str:
"""获取当前时间"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 工具列表
tools = [remember_important_info, recall_memories, get_current_time]
# 创建Prompt模板（包含记忆使用指引）
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个智能AI助手，拥有长期记忆和短期记忆能力。
【记忆系统说明】
1. 短期记忆：自动保留最近10轮对话，用于维持对话连贯性
2. 长期记忆：持久化存储重要信息，跨会话有效
3. 系统会自动整理和压缩旧对话到长期记忆（被动触发）
【主动记忆工具使用指南】
- 当用户分享重要信息时（姓名、偏好、事实），使用 remember_important_info 工具主动保存
- 当需要回忆之前的信息时，使用 recall_memories 工具查询
- 重要信息包括：用户姓名、喜好、偏好设置、重要日期、个人特征等
【记忆使用原则】
- 主动保存用户明确要求记住的内容
- 主动保存你认为对后续对话有价值的信息
- 回复时充分利用短期和长期记忆中的信息
当前时间: {current_time}
"""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# 初始化LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
# 创建Agent
agent = create_react_agent(llm, tools, prompt)
# 创建Agent执行器（带记忆注入）
class MemoryAgentExecutor:
def __init__(self, agent, tools, memory_manager):
self.agent = agent
self.tools = {t.name: t for t in tools}
self.memory_manager = memory_manager
def run(self, user_input: str) -> str:
# 1. 加载相关记忆（注入到对话上下文）
context = self.memory_manager.load_context(user_input)
# 2. 构建输入
from langchain.agents import AgentExecutor
executor = AgentExecutor(
agent=self.agent,
tools=list(self.tools.values()),
verbose=True,
handle_parsing_errors=True
)
# 3. 执行Agent
result = executor.invoke({
"input": user_input,
"current_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"chat_history": context["short_term"]
})
response = result.get("output", "")
# 4. 保存交互到记忆（会触发主动/被动检查）
self.memory_manager.add_interaction(user_input, response)
return response
def get_stats(self):
return self.memory_manager.get_memory_stats()
return MemoryAgentExecutor(agent, tools, memory_manager)
# ============================================================
# 第四部分：演示运行
# ============================================================
def main():
print("=" * 60)
print("🤖 混合记忆Agent演示")
print("=" * 60)
print("\n特性说明:")
print(" 🔵 主动触发: Agent检测到重要信息时主动保存")
print(" 🟢 被动触发: 对话超过6轮时自动整理压缩")
print("-" * 60)
# 创建Agent
agent = create_memory_agent()
# 演示对话
conversations = [
"你好！我叫张三，是一名软件工程师。",
"我平时喜欢喝美式咖啡，不加糖。",
"今天天气怎么样？", # 普通问题，不触发主动保存
"我下个月要去北京出差，有什么推荐吗？",
"对了，记住我喜欢靠窗的座位。", # 主动保存
"你还记得我的名字吗？",
"我的咖啡偏好是什么？",
"我有个重要的会议在周五下午3点。",
"你能提醒我那个会议吗？", # 主动保存
"我们聊了这么多，你还记得多少关于我的信息？"
]
for i, user_input in enumerate(conversations, 1):
print(f"\n{'='*60}")
print(f"📝 第{i}轮对话")
print(f"👤 用户: {user_input}")
response = agent.run(user_input)
print(f"🤖 AI: {response}")
# 显示记忆统计
if i % 3 == 0:
stats = agent.get_stats()
print(f"\n📊 [记忆统计] 短期记忆: {stats['short_term_count']}/{stats['short_term_max']}轮 | 总交互: {stats['total_interactions']}次")
print("\n" + "=" * 60)
print("✅ 演示完成")
print("=" * 60)
if __name__ == "__main__":
main()3. 代码结构解析
核心组件说明
主动触发机制
# Agent可调用的工具
@tool
def remember_important_info(info: str) -> str:
"""Agent主动决定保存重要信息"""
memory_manager.active_save(info)被动触发机制
def add_interaction(self, user_input: str, ai_response: str):
# 每次交互后检查短期记忆长度
if len(chat_history) >= self.passive_trigger_threshold:
self.passive_consolidate() # 自动整理压缩4. 运行效果预期
🔵 [主动触发] 正在保存重要信息到长期记忆...
内容: 我叫张三，是一名软件工程师...
✅ 主动保存成功
...
🟢 [被动触发] 开始记忆整理与压缩...
✅ 被动整理完成: 压缩了 3 轮对话到长期记忆
📊 [记忆统计] 短期记忆: 6/10轮 | 总交互: 6次这个实现完全满足你的需求：主动触发（Agent通过工具调用）+ 被动触发（系统阈值自动整理）的混合记忆架构。
