企业官网建设流程全解析

在实际的多 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 系统通常包含以下三个层次：

1. 短期记忆 (Short-term Memory)

   • 定义：存储当前对话或任务执行过程中产生的瞬时信息，生命周期极短，通常与单次推理或几次连续的交互绑定。
   • 类比：类似于计算机的 CPU 缓存或人类的工作记忆。
   • 内容：最新的用户输入、模型生成的思考过程、工具调用的临时结果、当前对话的上下文窗口。
   • 存储：通常存在于内存中，结构简单，访问速度极快。
   • 管理：采用滑动窗口或 FIFO（先进先出）队列机制，当容量达到上限时，自动淘汰最旧的记忆。

2. 中期记忆 (Mid-term Memory)

   • 定义：存储单次会话（Session）范围内的关键信息。会话是指从用户开始一个明确任务到任务结束或用户主动结束的完整周期。
   • 类比：类似于一个项目的工作区，项目结束后大部分内容可以归档或丢弃。
   • 内容：会话的目标、已完成的子任务、会话中产生的临时事实和决策、用户在本会话中表达的偏好。
   • 存储：可以持久化到数据库或文件系统中，但通常与会话 ID 绑定，会话结束后可被清理或转入长期记忆。
   • 管理：基于会话 ID 进行创建、查询和销毁。

3. 长期记忆 (Long-term Memory)

   • 定义：存储跨越多次会话的、具有持久价值的核心信息。这些信息构成了 Agent 的“知识”或“个性”。
   • 类比：类似于个人的长期知识库或数据库。
   • 内容：用户画像（如姓名、常用需求）、历史交互的摘要或关键结论、学到的技能或规则、需要长期跟踪的任务状态。
   • 存储：必须持久化到可靠的存储中，如关系型数据库、向量数据库（用于基于语义的检索）。
   • 管理：支持复杂的 CRUD 操作、语义检索、定期更新和归档。

1.3 记忆在 Agent 协作中的流转

记忆并非静态存储。在一个典型的多 Agent 请求处理流程中，记忆会动态流转：

1. 用户发起请求。
2. 路由或协调 Agent从长期记忆中检索与该用户相关的历史信息和偏好。
3. 结合当前请求，形成增强的上下文，传递给专业 Agent（如代码生成 Agent、数据分析 Agent）。
4. 专业 Agent 在处理过程中，会利用短期记忆保持思考连贯性，并将关键步骤和结果写入中期记忆（关联本次会话）。
5. 任务完成后，协调 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

安装基础依赖（如果使用 Pydantic）：

pip install pydantic

2.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_factory=lambda: str(uuid4())) content: Dict[str, Any] # 记忆内容，用字典存储灵活的数据 metadata: Dict[str, Any] = Field(default_factory=dict) # 元数据，如来源、类型、重要性 created_at: datetime = Field(default_factory=datetime.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_factory=list) # 用于分类检索的标签 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(maxlen=maxlen) 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(maxlen=self._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(parents=True, exist_ok=True) 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( id=row[0], session_id=row[1], content=self._json_to_dict(row[2]), metadata=self._json_to_dict(row[3]), created_at=datetime.fromisoformat(row[4]), last_accessed_at=datetime.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(parents=True, exist_ok=True) 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 1=1"] 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(f"AND 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( id=row[0], user_id=row[1], content=self._json_to_dict(row[2]), metadata=self._json_to_dict(row[3]), tags=tags, created_at=datetime.fromisoformat(row[5]), last_accessed_at=datetime.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(maxlen=20) # 短期记忆容量设为20 self.mid_term = MidTermMemory() self.long_term = LongTermMemory() def add_short_term(self, content: dict, metadata: Optional[dict] = None) -> str: """添加短期记忆""" item = ShortTermMemoryItem(content=content, metadata=metadata or {}) return self.short_term.add(item) def add_mid_term(self, content: dict, metadata: Optional[dict] = None) -> str: """添加中期记忆，自动关联当前会话""" item = MidTermMemoryItem( session_id=self.session_id, content=content, metadata=metadata 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_id=self.user_id, content=content, tags=tags or [], metadata=metadata 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(query=query, user_id=self.user_id, limit=3) 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, limit=10) 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 context

4.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": f"Context: {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_id=user_id, session_id=session_id) print(f"初始化记忆管理器: User={user_id}, Session={session_id}") # 2. 创建 Agent agent = SimpleAgent(name="ReportAssistant", memory_manager=memory_manager) # 3. 模拟第一次交互（无历史记忆） print("\n--- 第一次交互 ---") response1 = agent.process_request("我想生成上个月的销售报告。") print(f"Agent 回复: {response1}") # 4. 模拟第二次交互（短期记忆生效） print("\n--- 第二次交互 ---") response2 = agent.process_request("对，用柱状图展示。") print(f"Agent 回复: {response2}") # 5. 模拟第三次交互（长期记忆开始记录偏好） print("\n--- 第三次交互 ---") response3 = agent.process_request("我比较喜欢用图表，看起来直观。") print(f"Agent 回复: {response3}") # 6. 模拟新会话中的交互（长期记忆生效） print("\n--- 新会话交互（长期记忆测试）---") new_session_id = "session_analysis_202311" memory_manager_new = MemoryManager(user_id=user_id, session_id=new_session_id) agent_new = SimpleAgent(name="AnalysisAssistant", memory_manager=memory_manager_new) response4 = agent_new.process_request("帮我分析一下数据趋势。") print(f"新 Agent 回复: {response4}") # 7. 手动检查记忆存储 print("\n--- 记忆存储检查 ---") print("长期记忆条目（用户偏好）:") long_mems = memory_manager.long_term.search(query="图表", user_id=user_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 三层记忆系统演示 === 初始化记忆管理器: User=user_123, Session=session_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', ...}

这个演示验证了：

1. 短期记忆：Agent 在第二次交互时，能“看到”第一次交互的内容（通过recent_short_term_memories）。
2. 中期记忆：当用户做出确认（“对，用柱状图展示”）时，该决策被记录到中期记忆，并在后续同会话交互中可用。
3. 长期记忆：用户表达的偏好（“喜欢用图表”）被提炼并存储到长期记忆。当新会话开始时，新的 Agent 能检索到这个长期偏好，从而提供更个性化的服务。

6. 常见问题排查与生产环境建议

将 Memmy 系统应用到真实项目时，你会遇到一系列工程化挑战。以下是一些常见问题及其排查思路。

6.1 记忆检索不准确或返回空

6.2 性能与扩展性问题

6.3 数据一致性与安全性

7. 最佳实践与扩展方向

基于上述实现和问题，以下是部署 Memmy 类系统时需要遵循的最佳实践和可能的扩展方向。

7.1 生产环境部署清单

1. 存储选型：
   • 短期记忆：使用高性能内存存储，如 Redis，并设置合理的 TTL（生存时间）。
   • 中期记忆：使用关系型数据库（如 PostgreSQL）或文档数据库（如 MongoDB），按会话 ID 分片或建立索引。
   • 长期记忆：结合使用关系型数据库（存结构化元数据）和向量数据库（如 ChromaDB, Pinecone, Weaviate）进行语义检索。
2. 记忆摘要与压缩：不要存储原始冗长的对话。使用 LLM 对一段对话或任务结果进行总结，将摘要存入中长期记忆。这能极大减少存储压力并提升检索质量。
3. 记忆生命周期管理：
   • 实现定期清理任务，删除过期的中期记忆（如 30 天前的会话）。
   • 对长期记忆设置归档策略，将低频访问的记忆移至冷存储。
4. 可观测性：为记忆的读写操作添加详细的日志和指标（Metrics），监控记忆命中率、写入延迟、存储容量等，便于问题排查和性能优化。
5. 版本化与回滚：重要的长期记忆（如用户配置、学到的规则）应考虑版本化，允许在出错时回滚到之前的版本。

7.2 扩展方向：从 Memmy 到智能体操作系统

基础的 Memmy 系统可以沿多个方向深化：

1. 记忆向量化与语义检索：这是最重要的扩展。使用嵌入模型（如 text-embedding-ada-002）将记忆内容转换为向量，存入向量数据库。检索时，将用户查询也向量化，进行相似度搜索，从而找到语义相关而非仅仅关键词匹配的记忆。
2. 记忆关联与图谱：建立记忆条目之间的关系。例如，记忆 A（“用户询问产品X”）和记忆 B（“用户购买了产品X”）可以关联起来。这有助于 Agent 进行更复杂的推理。可以使用图数据库（如 Neo4j）来存储这种关系。
3. 记忆重要性评分与遗忘机制：模仿人类的遗忘曲线，为记忆条目设计重要性评分算法。评分可基于访问频率、用户显式反馈、与其他记忆的关联度等。低分记忆可自动降级或清理。
4. 多模态记忆：支持存储和检索文本、图像、音频等多种格式的记忆。例如，用户上传的参考图片、语音指令都可以成为记忆的一部分。
5. 记忆共享与权限：在多个 Agent 协作的场景中，设计记忆的共享机制。某些记忆可以被工作区内的所有 Agent 访问，而有些则仅限于特定 Agent。这需要一套权限模型。

实现一个健壮的记忆系统是构建具有持续性和个性化能力 AI Agent 的关键一步。Memmy 三层架构提供了一个清晰的起点，但真正的挑战在于如何根据你的具体业务场景，在性能、准确性、成本和复杂性之间找到最佳平衡点。建议从最小可行产品开始，优先实现向量化检索和记忆摘要，这两项能带来最显著的体验提升，再逐步迭代更复杂的功能。