在实际的多 Agent 系统开发中一个长期存在的挑战是如何高效、持久地管理对话历史和智能体记忆。当多个 Agent 协作处理复杂任务时它们需要记住上下文、用户偏好、历史决策以及彼此之间的交互。如果每次对话都从零开始或者记忆分散在各个角落不仅会浪费计算资源还会导致对话不连贯、任务执行中断。Memmy 三层记忆系统正是为了解决这类问题而设计的一种架构思路它旨在为多 Agent 对话提供一个统一、分层、可扩展的记忆管理框架。本文面向正在构建或计划构建多 Agent 应用的开发者、架构师以及对 Agent 记忆机制感兴趣的研究者。我们将从零开始解析 Memmy 三层记忆系统的核心概念并通过一个模拟的 Python 项目展示如何实现一个基础版本。你将理解短期、中期、长期记忆在 Agent 协作中的作用掌握记忆的存储、检索与更新机制并学会如何将这套系统集成到你的 Agent 项目中从而提升 Agent 的上下文感知能力和任务连续性。1. 理解 Memmy 三层记忆系统的设计动机与核心架构在深入代码之前我们必须先厘清“记忆”在多 Agent 系统中的含义。这里的记忆并非指生物记忆而是指 Agent 在运行过程中产生、获取并可用于未来决策的数据状态。一个没有记忆的 Agent每次交互都是孤立的而拥有良好记忆系统的 Agent则能表现出连贯性、个性化和更高的任务完成率。1.1 为什么需要分层记忆单一的记忆存储方式例如只保存最近 10 条对话无法满足复杂场景的需求。考虑一个客服 Agent 和一个数据分析 Agent 协作的场景短期记忆需要记住当前对话轮次中用户的即时问题“查询上个月销售额”和 Agent 的即时回复用于维持对话流畅性。中期记忆需要记住本次会话中用户的核心意图“做月度报告”、已执行的操作“已提取销售数据”、以及临时产生的用户偏好“图表要柱状图”这些信息在单次会话内有效。长期记忆需要记住用户的身份信息、历史查询模式、常用的分析维度等这些信息跨越多次会话用于实现个性化服务和避免重复工作。Memmy 系统通过三层结构来区分这些不同粒度和生命周期的记忆实现精细化管理。1.2 Memmy 三层记忆的定义与职责Memmy 系统通常包含以下三个层次短期记忆 (Short-term Memory)定义存储当前对话或任务执行过程中产生的瞬时信息生命周期极短通常与单次推理或几次连续的交互绑定。类比类似于计算机的 CPU 缓存或人类的工作记忆。内容最新的用户输入、模型生成的思考过程、工具调用的临时结果、当前对话的上下文窗口。存储通常存在于内存中结构简单访问速度极快。管理采用滑动窗口或 FIFO先进先出队列机制当容量达到上限时自动淘汰最旧的记忆。中期记忆 (Mid-term Memory)定义存储单次会话Session范围内的关键信息。会话是指从用户开始一个明确任务到任务结束或用户主动结束的完整周期。类比类似于一个项目的工作区项目结束后大部分内容可以归档或丢弃。内容会话的目标、已完成的子任务、会话中产生的临时事实和决策、用户在本会话中表达的偏好。存储可以持久化到数据库或文件系统中但通常与会话 ID 绑定会话结束后可被清理或转入长期记忆。管理基于会话 ID 进行创建、查询和销毁。长期记忆 (Long-term Memory)定义存储跨越多次会话的、具有持久价值的核心信息。这些信息构成了 Agent 的“知识”或“个性”。类比类似于个人的长期知识库或数据库。内容用户画像如姓名、常用需求、历史交互的摘要或关键结论、学到的技能或规则、需要长期跟踪的任务状态。存储必须持久化到可靠的存储中如关系型数据库、向量数据库用于基于语义的检索。管理支持复杂的 CRUD 操作、语义检索、定期更新和归档。1.3 记忆在 Agent 协作中的流转记忆并非静态存储。在一个典型的多 Agent 请求处理流程中记忆会动态流转用户发起请求。路由或协调 Agent从长期记忆中检索与该用户相关的历史信息和偏好。结合当前请求形成增强的上下文传递给专业 Agent如代码生成 Agent、数据分析 Agent。专业 Agent 在处理过程中会利用短期记忆保持思考连贯性并将关键步骤和结果写入中期记忆关联本次会话。任务完成后协调 Agent 可能会将本次会话的精华总结例如“用户完成了月度销售报告偏好使用折线图”写回长期记忆供未来使用。这个流转过程确保了记忆的可用性和时效性是 Memmy 系统价值的关键体现。2. 环境准备与项目结构设计我们将使用 Python 来构建一个简化的 Memmy 三层记忆系统演示项目。这个项目不依赖特定的大模型 API而是聚焦于记忆系统本身的架构与实现。2.1 开发环境与依赖Python 版本: 3.8包管理工具: pip核心依赖:sqlite3(Python 内置): 用于长期记忆的持久化存储演示用生产环境可换用 PostgreSQL, MySQL 等。chromadb或faiss-cpu: 可选用于实现长期记忆的语义检索能力。本文为简化暂不使用向量数据库而是用关键词匹配模拟。pydantic: 用于数据验证和设置管理推荐。uuid: 用于生成唯一的会话 ID 和记忆 ID。首先创建一个新的项目目录并初始化虚拟环境mkdir memmy_agent_system cd memmy_agent_system python -m venv venv # Windows venv\Scripts\activate # Linux/macOS source venv/bin/activate安装基础依赖如果使用 Pydanticpip install pydantic2.2 项目目录结构一个清晰的项目结构有助于管理复杂度。我们的项目结构如下memmy_agent_system/ ├── memmy/ # 核心记忆系统包 │ ├── __init__.py │ ├── memory/ # 记忆层实现 │ │ ├── __init__.py │ │ ├── base.py # 记忆基类 │ │ ├── short_term.py # 短期记忆实现 │ │ ├── mid_term.py # 中期记忆实现 │ │ └── long_term.py # 长期记忆实现 │ ├── agent/ # Agent 相关示例 │ │ ├── __init__.py │ │ └── simple_agent.py # 一个使用记忆系统的简单 Agent │ ├── models.py # 数据模型Pydantic │ └── utils.py # 工具函数如 ID 生成 ├── storage/ # 存储相关如 SQLite 数据库文件 │ └── memory.db # SQLite 数据库文件运行后生成 ├── tests/ # 单元测试 ├── requirements.txt # 项目依赖 ├── config.yaml # 配置文件可选 └── demo.py # 演示主程序这个结构将记忆系统的核心逻辑与 Agent 逻辑分离便于维护和扩展。3. 实现 Memmy 三层记忆系统的核心组件我们将自底向上先定义数据模型和基类再实现每一层记忆。3.1 定义记忆数据模型 (memmy/models.py)使用 Pydantic 定义清晰的数据结构能有效避免后续的类型混乱。from pydantic import BaseModel, Field from typing import Any, Dict, List, Optional from datetime import datetime from uuid import uuid4 class MemoryItem(BaseModel): 记忆条目的基类模型 id: str Field(default_factorylambda: str(uuid4())) content: Dict[str, Any] # 记忆内容用字典存储灵活的数据 metadata: Dict[str, Any] Field(default_factorydict) # 元数据如来源、类型、重要性 created_at: datetime Field(default_factorydatetime.now) last_accessed_at: Optional[datetime] None class Config: json_encoders { datetime: lambda v: v.isoformat() } class ShortTermMemoryItem(MemoryItem): 短期记忆条目可能增加时效性字段 expires_at: Optional[datetime] None # 可选的过期时间 class MidTermMemoryItem(MemoryItem): 中期记忆条目关联会话 session_id: str # 所属会话的唯一标识 class LongTermMemoryItem(MemoryItem): 长期记忆条目可能增加用户关联和标签 user_id: Optional[str] None # 关联的用户ID tags: List[str] Field(default_factorylist) # 用于分类检索的标签 embedding: Optional[List[float]] None # 向量嵌入用于语义检索后续扩展3.2 实现记忆基类与短期记忆 (memmy/memory/base.py,short_term.py)基类 (memmy/memory/base.py)定义通用接口。from abc import ABC, abstractmethod from typing import List, Optional from ..models import MemoryItem class BaseMemory(ABC): 记忆系统的抽象基类 abstractmethod def add(self, item: MemoryItem) - str: 添加一条记忆返回记忆ID pass abstractmethod def get(self, memory_id: str) - Optional[MemoryItem]: 根据ID获取一条记忆 pass abstractmethod def search(self, query: str, limit: int 5) - List[MemoryItem]: 搜索相关记忆不同层实现不同 pass abstractmethod def update(self, memory_id: str, updates: dict) - bool: 更新一条记忆 pass abstractmethod def delete(self, memory_id: str) - bool: 删除一条记忆 pass短期记忆 (memmy/memory/short_term.py)使用内存中的有序字典或双端队列实现容量有限。from collections import deque from typing import List, Optional, Deque from .base import BaseMemory from ..models import ShortTermMemoryItem class ShortTermMemory(BaseMemory): 短期记忆实现基于固定长度的双端队列 (deque) def __init__(self, maxlen: int 10): 初始化短期记忆 :param maxlen: 记忆的最大容量超出时自动移除最旧的记忆 self._storage: Deque[ShortTermMemoryItem] deque(maxlenmaxlen) self._index {} # ID到内存对象的映射用于快速查找 def add(self, item: ShortTermMemoryItem) - str: # 如果队列已满deque 会自动弹出最左边的元素我们需要同步清理 _index if len(self._storage) self._storage.maxlen: removed_item self._storage[0] # 即将被移除的项目 self._index.pop(removed_item.id, None) self._storage.append(item) self._index[item.id] item return item.id def get(self, memory_id: str) - Optional[ShortTermMemoryItem]: return self._index.get(memory_id) def search(self, query: str, limit: int 5) - List[ShortTermMemoryItem]: # 短期记忆搜索简单返回最新的几条或进行简单的内容匹配 # 这里模拟基于关键词的简单搜索 results [] query_lower query.lower() for item in reversed(self._storage): # 从最新开始搜索 # 检查 content 字典的 values 或 metadata 中是否包含查询词 content_str str(item.content).lower() if query_lower in content_str and len(results) limit: results.append(item) return results def update(self, memory_id: str, updates: dict) - bool: item self.get(memory_id) if not item: return False # 更新 content 和 metadata if content in updates: item.content.update(updates[content]) if metadata in updates: item.metadata.update(updates[metadata]) item.last_accessed_at datetime.now() return True def delete(self, memory_id: str) - bool: item self.get(memory_id) if not item: return False # 从 deque 中删除比较低效但短期记忆容量小可以接受。 # 更优方案是使用 OrderedDict这里为演示清晰使用简单方法。 try: # 创建一个新 deque排除要删除的项 new_storage deque(maxlenself._storage.maxlen) for i in self._storage: if i.id ! memory_id: new_storage.append(i) self._storage new_storage self._index.pop(memory_id, None) return True except Exception: return False def get_recent(self, n: int) - List[ShortTermMemoryItem]: 获取最近的 n 条记忆这是短期记忆的常用操作 return list(self._storage)[-n:]3.3 实现中期记忆 (memmy/memory/mid_term.py)中期记忆与会话强绑定我们使用 SQLite 数据库来存储每个表对应一个会话或所有会话存一张表用session_id区分。这里采用单表设计。import sqlite3 from typing import List, Optional from datetime import datetime from pathlib import Path from .base import BaseMemory from ..models import MidTermMemoryItem class MidTermMemory(BaseMemory): 中期记忆实现基于 SQLite按会话隔离 def __init__(self, db_path: str storage/memory.db): self.db_path Path(db_path) self.db_path.parent.mkdir(parentsTrue, exist_okTrue) self._init_db() def _init_db(self): 初始化数据库表 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS mid_term_memory ( id TEXT PRIMARY KEY, session_id TEXT NOT NULL, content TEXT NOT NULL, -- 存储 JSON 字符串 metadata TEXT NOT NULL, -- 存储 JSON 字符串 created_at TIMESTAMP NOT NULL, last_accessed_at TIMESTAMP, INDEX idx_session (session_id), INDEX idx_created (created_at) ) ) conn.commit() conn.close() def _dict_to_json(self, data: dict) - str: import json return json.dumps(data) def _json_to_dict(self, json_str: str) - dict: import json return json.loads(json_str) def add(self, item: MidTermMemoryItem) - str: conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( INSERT INTO mid_term_memory (id, session_id, content, metadata, created_at, last_accessed_at) VALUES (?, ?, ?, ?, ?, ?) , ( item.id, item.session_id, self._dict_to_json(item.content), self._dict_to_json(item.metadata), item.created_at.isoformat(), item.last_accessed_at.isoformat() if item.last_accessed_at else None )) conn.commit() conn.close() return item.id def get(self, memory_id: str) - Optional[MidTermMemoryItem]: conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute(SELECT * FROM mid_term_memory WHERE id ?, (memory_id,)) row cursor.fetchone() conn.close() if not row: return None return self._row_to_item(row) def search(self, query: str, limit: int 5, session_id: Optional[str] None) - List[MidTermMemoryItem]: 搜索中期记忆可以限定在某个会话内 conn sqlite3.connect(self.db_path) cursor conn.cursor() sql SELECT * FROM mid_term_memory WHERE content LIKE ? params [f%{query}%] if session_id: sql AND session_id ? params.append(session_id) sql ORDER BY created_at DESC LIMIT ? params.append(limit) cursor.execute(sql, params) rows cursor.fetchall() conn.close() return [self._row_to_item(row) for row in rows] def get_by_session(self, session_id: str, limit: int 20) - List[MidTermMemoryItem]: 获取指定会话的所有记忆按时间倒序 conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute( SELECT * FROM mid_term_memory WHERE session_id ? ORDER BY created_at DESC LIMIT ? , (session_id, limit)) rows cursor.fetchall() conn.close() return [self._row_to_item(row) for row in rows] def update(self, memory_id: str, updates: dict) - bool: # 实现更新逻辑更新 content/metadata 和 last_accessed_at # 略与短期记忆类似但操作数据库 pass def delete(self, memory_id: str) - bool: conn sqlite3.connect(self.db_path) cursor conn.cursor() cursor.execute(DELETE FROM mid_term_memory WHERE id ?, (memory_id,)) affected cursor.rowcount conn.commit() conn.close() return affected 0 def _row_to_item(self, row) - MidTermMemoryItem: 将数据库行转换为 MidTermMemoryItem 对象 return MidTermMemoryItem( idrow[0], session_idrow[1], contentself._json_to_dict(row[2]), metadataself._json_to_dict(row[3]), created_atdatetime.fromisoformat(row[4]), last_accessed_atdatetime.fromisoformat(row[5]) if row[5] else None )3.4 实现长期记忆 (memmy/memory/long_term.py)长期记忆的实现最为复杂需要考虑高效检索。这里先用 SQLite 实现一个基于标签和关键词的简化版。生产环境强烈建议引入向量数据库如 ChromaDB, Weaviate进行语义检索。import sqlite3 from typing import List, Optional from datetime import datetime from pathlib import Path from .base import BaseMemory from ..models import LongTermMemoryItem class LongTermMemory(BaseMemory): 长期记忆实现基于 SQLite支持标签检索 def __init__(self, db_path: str storage/memory.db): self.db_path Path(db_path) self.db_path.parent.mkdir(parentsTrue, exist_okTrue) self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) cursor conn.cursor() # 主表 cursor.execute( CREATE TABLE IF NOT EXISTS long_term_memory ( id TEXT PRIMARY KEY, user_id TEXT, content TEXT NOT NULL, metadata TEXT NOT NULL, tags TEXT, -- 逗号分隔的标签字符串如 report,monthly,sales created_at TIMESTAMP NOT NULL, last_accessed_at TIMESTAMP, INDEX idx_user (user_id), INDEX idx_created (created_at) ) ) # 为了标签搜索效率可以创建标签关联表这里为简化使用 LIKE 查询 conn.commit() conn.close() def add(self, item: LongTermMemoryItem) - str: conn sqlite3.connect(self.db_path) cursor conn.cursor() tags_str ,.join(item.tags) if item.tags else cursor.execute( INSERT INTO long_term_memory (id, user_id, content, metadata, tags, created_at, last_accessed_at) VALUES (?, ?, ?, ?, ?, ?, ?) , ( item.id, item.user_id, self._dict_to_json(item.content), self._dict_to_json(item.metadata), tags_str, item.created_at.isoformat(), item.last_accessed_at.isoformat() if item.last_accessed_at else None )) conn.commit() conn.close() return item.id def search(self, query: str, limit: int 5, user_id: Optional[str] None, tags: Optional[List[str]] None) - List[LongTermMemoryItem]: 搜索长期记忆支持关键词、用户ID和标签过滤 conn sqlite3.connect(self.db_path) cursor conn.cursor() sql_parts [SELECT * FROM long_term_memory WHERE 11] params [] if query: sql_parts.append(AND (content LIKE ? OR metadata LIKE ?)) params.extend([f%{query}%, f%{query}%]) if user_id: sql_parts.append(AND user_id ?) params.append(user_id) if tags: for tag in tags: sql_parts.append(fAND tags LIKE ?) params.append(f%{tag}%) sql_parts.append(ORDER BY last_accessed_at DESC, created_at DESC LIMIT ?) params.append(limit) sql .join(sql_parts) cursor.execute(sql, params) rows cursor.fetchall() conn.close() return [self._row_to_item(row) for row in rows] # get, update, delete 方法类似中期记忆此处省略具体实现 def _row_to_item(self, row) - LongTermMemoryItem: tags_str row[4] if row[4] else tags tags_str.split(,) if tags_str else [] return LongTermMemoryItem( idrow[0], user_idrow[1], contentself._json_to_dict(row[2]), metadataself._json_to_dict(row[3]), tagstags, created_atdatetime.fromisoformat(row[5]), last_accessed_atdatetime.fromisoformat(row[6]) if row[6] else None )4. 构建统一记忆管理器与集成到 Agent三层记忆组件已经就绪现在需要一个统一的入口来协调它们并展示如何被 Agent 使用。4.1 创建统一记忆管理器 (memmy/memory/manager.py)管理器负责初始化各层记忆并提供高层级的 API。from typing import Optional, List from .short_term import ShortTermMemory from .mid_term import MidTermMemory from .long_term import LongTermMemory from ..models import ShortTermMemoryItem, MidTermMemoryItem, LongTermMemoryItem class MemoryManager: 统一记忆管理器协调三层记忆 def __init__(self, user_id: Optional[str] None, session_id: Optional[str] None): self.user_id user_id self.session_id session_id or str(uuid4()) # 如果没有提供生成一个会话ID self.short_term ShortTermMemory(maxlen20) # 短期记忆容量设为20 self.mid_term MidTermMemory() self.long_term LongTermMemory() def add_short_term(self, content: dict, metadata: Optional[dict] None) - str: 添加短期记忆 item ShortTermMemoryItem(contentcontent, metadatametadata or {}) return self.short_term.add(item) def add_mid_term(self, content: dict, metadata: Optional[dict] None) - str: 添加中期记忆自动关联当前会话 item MidTermMemoryItem( session_idself.session_id, contentcontent, metadatametadata or {} ) return self.mid_term.add(item) def add_long_term(self, content: dict, tags: List[str] None, metadata: Optional[dict] None) - str: 添加长期记忆自动关联当前用户 item LongTermMemoryItem( user_idself.user_id, contentcontent, tagstags or [], metadatametadata or {} ) return self.long_term.add(item) def get_context_for_agent(self, query: str) - dict: 为 Agent 生成增强的上下文。 策略从长期记忆中获取用户相关记忆从中期记忆中获取本次会话记忆 从短期记忆中获取最近几条记忆合并后返回。 context { user_long_term_memories: [], session_mid_term_memories: [], recent_short_term_memories: [] } # 1. 从长期记忆检索基于用户ID和查询关键词 if self.user_id: long_term_memories self.long_term.search(queryquery, user_idself.user_id, limit3) context[user_long_term_memories] [m.content for m in long_term_memories] # 2. 从中期记忆检索基于会话ID mid_term_memories self.mid_term.get_by_session(self.session_id, limit10) context[session_mid_term_memories] [m.content for m in mid_term_memories] # 3. 从短期记忆获取最近几条 short_term_memories self.short_term.get_recent(5) context[recent_short_term_memories] [m.content for m in short_term_memories] return context4.2 创建一个使用记忆的简单 Agent (memmy/agent/simple_agent.py)这个 Agent 模拟一个任务执行者它接收用户请求从记忆管理器中获取上下文然后“处理”请求并将关键信息写入记忆。class SimpleAgent: 一个使用 Memmy 记忆系统的简单 Agent def __init__(self, name: str, memory_manager: MemoryManager): self.name name self.memory memory_manager def process_request(self, user_input: str) - str: 处理用户请求的核心方法。 1. 获取记忆上下文。 2. 基于上下文生成回复。 3. 将关键信息写入记忆。 print(f[Agent {self.name}] 收到请求: {user_input}) # 步骤1获取记忆上下文 context self.memory.get_context_for_agent(user_input) print(f[Agent {self.name}] 上下文: {context}) # 步骤2模拟“思考”和生成回复这里用简单逻辑代替大模型调用 response self._generate_response(user_input, context) # 步骤3写入记忆 self._update_memory(user_input, response, context) return response def _generate_response(self, user_input: str, context: dict) - str: 模拟基于上下文生成回复的逻辑 # 这里可以集成真正的 LLM 调用 # 例如: response llm_client.chat( messages[{role:system, content: fContext: {context}}, ...]) # 为演示我们返回一个固定的回复 long_term_info context.get(user_long_term_memories, []) if long_term_info: return f我知道你之前关心过类似的事情。基于你的历史我的建议是继续推进当前任务。 else: return f这是我第一次处理这类请求。我已记录你的需求{user_input}。 def _update_memory(self, user_input: str, response: str, context: dict): 将交互的关键信息写入各层记忆 # 短期记忆记录本次交互 self.memory.add_short_term({ user_input: user_input, agent_response: response, timestamp: datetime.now().isoformat() }) # 中期记忆记录本次会话的关键决策或事实例如用户确认了某个选项 if confirm in user_input.lower(): self.memory.add_mid_term({ action: user_confirmation, detail: user_input, response: response }, metadata{type: decision}) # 长期记忆如果本次交互包含了值得长期记住的用户偏好或事实 # 例如用户说“我喜欢用图表展示” if 喜欢 in user_input and 图表 in user_input: self.memory.add_long_term({ preference: visualization, detail: 用户偏好使用图表进行展示, learned_from: user_input }, tags[preference, ui], metadata{source: user_input})5. 运行演示与验证现在我们可以编写一个演示脚本 (demo.py) 来验证整个系统的运作。#!/usr/bin/env python3 Memmy 三层记忆系统演示 import sys sys.path.insert(0, .) from memmy.memory.manager import MemoryManager from memmy.agent.simple_agent import SimpleAgent def main(): print( Memmy 三层记忆系统演示 \n) # 模拟用户ID和会话 user_id user_123 session_id session_report_202310 # 1. 初始化记忆管理器关联用户和会话 memory_manager MemoryManager(user_iduser_id, session_idsession_id) print(f初始化记忆管理器: User{user_id}, Session{session_id}) # 2. 创建 Agent agent SimpleAgent(nameReportAssistant, memory_managermemory_manager) # 3. 模拟第一次交互无历史记忆 print(\n--- 第一次交互 ---) response1 agent.process_request(我想生成上个月的销售报告。) print(fAgent 回复: {response1}) # 4. 模拟第二次交互短期记忆生效 print(\n--- 第二次交互 ---) response2 agent.process_request(对用柱状图展示。) print(fAgent 回复: {response2}) # 5. 模拟第三次交互长期记忆开始记录偏好 print(\n--- 第三次交互 ---) response3 agent.process_request(我比较喜欢用图表看起来直观。) print(fAgent 回复: {response3}) # 6. 模拟新会话中的交互长期记忆生效 print(\n--- 新会话交互长期记忆测试---) new_session_id session_analysis_202311 memory_manager_new MemoryManager(user_iduser_id, session_idnew_session_id) agent_new SimpleAgent(nameAnalysisAssistant, memory_managermemory_manager_new) response4 agent_new.process_request(帮我分析一下数据趋势。) print(f新 Agent 回复: {response4}) # 7. 手动检查记忆存储 print(\n--- 记忆存储检查 ---) print(长期记忆条目用户偏好:) long_mems memory_manager.long_term.search(query图表, user_iduser_id) for mem in long_mems: print(f - ID: {mem.id}, 内容: {mem.content}, 标签: {mem.tags}) print(\n中期记忆条目旧会话:) mid_mems memory_manager.mid_term.get_by_session(session_id) for mem in mid_mems: print(f - ID: {mem.id}, 内容: {mem.content}) if __name__ __main__: main()运行演示脚本python demo.py预期你会看到类似以下的输出展示了记忆在不同交互和会话间的流转 Memmy 三层记忆系统演示 初始化记忆管理器: Useruser_123, Sessionsession_report_202310 --- 第一次交互 --- [Agent ReportAssistant] 收到请求: 我想生成上个月的销售报告。 [Agent ReportAssistant] 上下文: {user_long_term_memories: [], session_mid_term_memories: [], recent_short_term_memories: []} Agent 回复: 这是我第一次处理这类请求。我已记录你的需求我想生成上个月的销售报告。。 --- 第二次交互 --- [Agent ReportAssistant] 收到请求: 对用柱状图展示。 [Agent ReportAssistant] 上下文: {user_long_term_memories: [], session_mid_term_memories: [], recent_short_term_memories: [{user_input: 我想生成上个月的销售报告。, agent_response: 这是我第一次处理这类请求。我已记录你的需求\我想生成上个月的销售报告。\。, timestamp: 2023-10-27T10:00:00}]} Agent 回复: 我知道你之前关心过类似的事情。基于你的历史我的建议是继续推进当前任务。 --- 第三次交互 --- [Agent ReportAssistant] 收到请求: 我比较喜欢用图表看起来直观。 [Agent ReportAssistant] 上下文: {user_long_term_memories: [], session_mid_term_memories: [{action: user_confirmation, detail: 对用柱状图展示。, response: 我知道你之前关心过类似的事情。基于你的历史我的建议是继续推进当前任务。}], recent_short_term_memories: [...]} Agent 回复: 这是我第一次处理这类请求。我已记录你的需求我比较喜欢用图表看起来直观。。 --- 新会话交互长期记忆测试--- [Agent AnalysisAssistant] 收到请求: 帮我分析一下数据趋势。 [Agent AnalysisAssistant] 上下文: {user_long_term_memories: [{preference: visualization, detail: 用户偏好使用图表进行展示, learned_from: 我比较喜欢用图表看起来直观。}], session_mid_term_memories: [], recent_short_term_memories: []} Agent 回复: 我知道你之前关心过类似的事情。基于你的历史我的建议是继续推进当前任务。 --- 记忆存储检查 --- 长期记忆条目用户偏好: - ID: xxx, 内容: {preference: visualization, ...}, 标签: [preference, ui] 中期记忆条目旧会话: - ID: yyy, 内容: {action: user_confirmation, ...}这个演示验证了短期记忆Agent 在第二次交互时能“看到”第一次交互的内容通过recent_short_term_memories。中期记忆当用户做出确认“对用柱状图展示”时该决策被记录到中期记忆并在后续同会话交互中可用。长期记忆用户表达的偏好“喜欢用图表”被提炼并存储到长期记忆。当新会话开始时新的 Agent 能检索到这个长期偏好从而提供更个性化的服务。6. 常见问题排查与生产环境建议将 Memmy 系统应用到真实项目时你会遇到一系列工程化挑战。以下是一些常见问题及其排查思路。6.1 记忆检索不准确或返回空问题现象可能原因检查方式处理建议Agent 获取不到预期的历史记忆。1. 记忆条目未成功写入。2. 检索时使用的user_id或session_id不匹配。3. 搜索关键词与存储内容不匹配文本搜索局限性。4. 数据库连接或文件权限问题。1. 检查add方法的返回值记忆ID和日志。2. 核对MemoryManager初始化时的user_id和session_id。3. 直接查询底层存储如 SQLite 数据库看数据是否存在。4. 检查数据库文件路径和读写权限。1. 为写入操作添加日志。2. 确保用户和会话标识在请求间正确传递。3. 考虑使用向量数据库进行语义检索而非简单关键词匹配。4. 使用连接池并添加重试机制。短期记忆似乎“丢失”了较早的对话。短期记忆的maxlen设置过小。检查ShortTermMemory初始化时的maxlen参数。根据对话的平均长度和复杂度调整maxlen。对于复杂任务可能需要 50-100。6.2 性能与扩展性问题问题现象可能原因检查方式处理建议随着记忆条目增多系统响应变慢。1. 数据库表缺乏有效索引。2. 每次请求都进行全表扫描或复杂 JOIN。3. 短期记忆的deque删除操作低效。1. 使用数据库的EXPLAIN命令分析查询计划。2. 监控 API 响应时间定位慢查询。3. 对长期记忆表进行性能压测。1. 为user_id,session_id,created_at,tags等常用查询字段建立索引。2. 对长期记忆进行分库分表或使用专门的向量数据库。3. 考虑使用collections.OrderedDict或lru_cache优化短期记忆的删除操作。记忆内容过大导致存储和传输成本高。记忆的content字段存储了过大的数据如完整文档。检查写入记忆的content数据大小。1. 只存储摘要、关键信息或引用而非完整数据。2. 对大内容进行压缩。3. 将原始数据存储到对象存储如 S3在记忆中只存链接和元数据。6.3 数据一致性与安全性问题现象可能原因检查方式处理建议多个 Agent 实例同时读写同一用户的记忆导致数据错乱。缺乏并发控制。模拟高并发场景检查记忆的最终状态。1. 在数据库层使用事务Transaction。2. 对于关键记忆的更新使用乐观锁或悲观锁。3. 考虑将记忆管理器设计为无状态服务通过外部的分布式锁服务如 Redis协调。敏感信息如个人信息被存入记忆。未对输入数据进行清洗或脱敏。审查所有调用add_short_term、add_mid_term、add_long_term的地方。1. 在记忆写入前增加一个数据过滤或脱敏层。2. 定义清晰的记忆 schema规定哪些信息可以存哪些不可以。3. 提供记忆删除和更新接口以满足数据合规要求如 GDPR 被遗忘权。7. 最佳实践与扩展方向基于上述实现和问题以下是部署 Memmy 类系统时需要遵循的最佳实践和可能的扩展方向。7.1 生产环境部署清单存储选型短期记忆使用高性能内存存储如 Redis并设置合理的 TTL生存时间。中期记忆使用关系型数据库如 PostgreSQL或文档数据库如 MongoDB按会话 ID 分片或建立索引。长期记忆结合使用关系型数据库存结构化元数据和向量数据库如 ChromaDB, Pinecone, Weaviate进行语义检索。记忆摘要与压缩不要存储原始冗长的对话。使用 LLM 对一段对话或任务结果进行总结将摘要存入中长期记忆。这能极大减少存储压力并提升检索质量。记忆生命周期管理实现定期清理任务删除过期的中期记忆如 30 天前的会话。对长期记忆设置归档策略将低频访问的记忆移至冷存储。可观测性为记忆的读写操作添加详细的日志和指标Metrics监控记忆命中率、写入延迟、存储容量等便于问题排查和性能优化。版本化与回滚重要的长期记忆如用户配置、学到的规则应考虑版本化允许在出错时回滚到之前的版本。7.2 扩展方向从 Memmy 到智能体操作系统基础的 Memmy 系统可以沿多个方向深化记忆向量化与语义检索这是最重要的扩展。使用嵌入模型如 text-embedding-ada-002将记忆内容转换为向量存入向量数据库。检索时将用户查询也向量化进行相似度搜索从而找到语义相关而非仅仅关键词匹配的记忆。记忆关联与图谱建立记忆条目之间的关系。例如记忆 A“用户询问产品X”和记忆 B“用户购买了产品X”可以关联起来。这有助于 Agent 进行更复杂的推理。可以使用图数据库如 Neo4j来存储这种关系。记忆重要性评分与遗忘机制模仿人类的遗忘曲线为记忆条目设计重要性评分算法。评分可基于访问频率、用户显式反馈、与其他记忆的关联度等。低分记忆可自动降级或清理。多模态记忆支持存储和检索文本、图像、音频等多种格式的记忆。例如用户上传的参考图片、语音指令都可以成为记忆的一部分。记忆共享与权限在多个 Agent 协作的场景中设计记忆的共享机制。某些记忆可以被工作区内的所有 Agent 访问而有些则仅限于特定 Agent。这需要一套权限模型。实现一个健壮的记忆系统是构建具有持续性和个性化能力 AI Agent 的关键一步。Memmy 三层架构提供了一个清晰的起点但真正的挑战在于如何根据你的具体业务场景在性能、准确性、成本和复杂性之间找到最佳平衡点。建议从最小可行产品开始优先实现向量化检索和记忆摘要这两项能带来最显著的体验提升再逐步迭代更复杂的功能。