记忆系统与多智能体编排深度分析
---
一、记忆系统 (Memory System) 架构分析
1.1 设计理念
OpenAkita 的记忆系统经过 v2 架构重构，参考了 Mem0、LangMem 等业界最佳实践，形成了独特的三层记忆架构。其核心设计理念是：
> "区分「用户是谁」和「用户要做什么」" —— 只记忆用户的身份、性格、长期偏好，不记忆具体的任务指令。
1.2 记忆类型体系
项目定义了 7 种记忆类型 (types.py)：
class MemoryType(Enum):
FACT = "fact" # 事实性信息
PREFERENCE = "preference" # 用户偏好
SKILL = "skill" # 技能知识
CONTEXT = "context" # 上下文（向后兼容）
RULE = "rule" # 行为规则
ERROR = "error" # 错误经验
PERSONA_TRAIT = "persona_trait" # 人格特征
EXPERIENCE = "experience" # 任务经验教训
配合 4 种优先级 (types.py)：
class MemoryPriority(Enum):
TRANSIENT = "transient" # 临时（1天）
SHORT_TERM = "short_term" # 短期（3天）
LONG_TERM = "long_term" # 长期（30天）
PERMANENT = "permanent" # 永久
以及 3 种作用域：
| 作用域 | 说明 |
|---|
GLOBAL | 全局共享 |
AGENT | Agent 私有 |
SESSION | 会话私有 |
1.3 v2 架构核心组件
┌─────────────────────────────────────────────────────────────────┐
│ MemoryManager (协调器) │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Extractor │ │ Retrieval │ │ UnifiedStore │ │
│ │ (AI提取) │ │ Engine │ │ (SQLite + Search) │ │
│ │ │ │ (多路召回) │ │ │ │
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │Consolidator │ │ VectorStore │ │ SearchBackend │ │
│ │ (记忆合并) │ │ (向量存储) │ │ (FTS5/向量/混合) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
#### 1.3.1 UnifiedStore — 统一存储层
unified_store.py 实现了存储与检索的分离：
- SQLite 主存储：
MemoryStorage 负责结构化数据持久化
- SearchBackend 索引：
FTS5Backend（全文搜索）或向量后端
- 降级机制：当向量后端不可用时，自动回退到 FTS5
class UnifiedStore:
def __init__(self, db_path, search_backend=None, ...):
self.db = MemoryStorage(db_path) # SQLite
self.search = create_search_backend(backend_type, ...)
#### 1.3.2 RetrievalEngine — 多路召回引擎
retrieval.py 实现了复杂的检索策略：
召回通道：
1. 语义搜索 — 基于向量或 FTS5
2. 情节搜索 — 实体/工具名关联
3. 时间搜索 — 最近 N 天
4. 附件搜索 — 文件/媒体关联
排序权重（第 38-41 行）：
W_RELEVANCE = 0.40 # 相关性
W_RECENCY = 0.20 # 时效性
W_IMPORTANCE = 0.20 # 重要性
W_ACCESS = 0.20 # 访问频率
LLM 关键词拆解：使用 QUERY_DECOMPOSE_PROMPT 将自然语言查询转换为搜索关键词。
#### 1.3.3 MemoryExtractor — AI 提取器
extractor.py 负责从对话中提取值得记忆的信息。其核心判断逻辑（第 35-60 行）：
> 问自己"这条信息在一个月后的新对话中还有用吗？"
提取原则：
- ✅ 记录：用户身份、性格偏好、行为规则、技术环境、可复用经验
- ❌ 不记录：具体任务请求、临时指令
1.4 记忆注入策略
manager.py 实现了三层注入：
┌─────────────────────────────────────────┐
│ Scratchpad (工作记忆) │
│ 跨 session 的工作草稿，实时更新 │
├─────────────────────────────────────────┤
│ Core Memory (核心记忆) │
│ 身份、性格、长期偏好，MEMORY.md │
├─────────────────────────────────────────┤
│ Dynamic Memories (动态记忆) │
│ 检索召回的事实、偏好、经验 │
└─────────────────────────────────────────┘
1.5 记忆系统亮点与改进建议
| 亮点 | 改进建议 |
|---|
| 多路召回 + 综合排序 | 可增加用户反馈闭环优化排序 |
| 智能过期机制 (TTL) | 可增加主动回忆/强化机制 |
| 向后兼容 v1 | 长期可考虑移除兼容层 |
| LLM 关键词拆解 | 可增加缓存命中率 |
---
二、多智能体编排 (Multi-Agent Orchestration) 架构分析
2.1 设计目标
OpenAkita 的多智能体系统旨在实现：
- 专业化分工：不同 Agent 擅长不同领域
- 并行协作：一个请求触发多个 Agent 同时工作
- 故障转移：某个 Agent 失败时自动切换
- 状态可视化：实时追踪子 Agent 状态
2.2 核心组件
┌─────────────────────────────────────────────────────────────────┐
│ AgentOrchestrator (编排器) │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ AgentMailbox│ │ AgentHealth│ │ AgentInstancePool │ │
│ │ (消息队列) │ │ (健康度) │ │ (实例管理+空闲回收) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │DelegationLog│ │ Fallback │ │ ProfileStore │ │
│ │ (委托日志) │ │ (故障转移) │ │ (AgentProfile 存储) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
#### 2.2.1 AgentOrchestrator — 中央编排器
orchestrator.py 是整个多智能体系统的核心：
核心职责：
1. 消息路由：根据 session 的 agent_profile_id 分发到对应 Agent
2. 委托管理：支持 Agent 之间的任务委托（有深度限制 MAX_DELEGATION_DEPTH = 5）
3. 超时处理：空闲超时 1200s、硬性超时可配置
4. 健康度追踪：记录每个 Agent 的请求数、成功率、平均延迟
委托请求结构（第 38-46 行）：
@dataclass
class DelegationRequest:
from_agent: str # 委托方
to_agent: str # 被委托方
message: str # 任务描述
session_key: str # 会话标识
depth: int = 0 # 委托深度
#### 2.2.2 AgentMailbox — 异步消息队列
每个 Agent 拥有独立的异步消息队列 (asyncio.Queue)：
class AgentMailbox:
async def send(self, message: dict) -> None:
await self._queue.put(message)
async def receive(self, timeout: float = 300.0) -> dict | None:
# 超时返回 None，支持优雅降级
这种设计支持多 Agent 并行工作，互不阻塞。
#### 2.2.3 AgentFactory — Agent 实例工厂
factory.py 根据 AgentProfile 创建差异化实例：
核心功能：
1. 技能过滤：根据 profile 配置筛选可用技能
2. 提示词注入：添加自定义提示词
3. 名称/图标设置：差异化展示
关键配置（第 28-45 行）：
class SkillsMode(str, Enum):
INCLUSIVE = "inclusive" # 仅含列表中的技能
EXCLUSIVE = "exclusive" # 排除列表中的技能
ALL = "all" # 全部技能
基础系统技能（第 23-27 行）：无论哪种模式，都必须保留的核心技能：
- 规划：
create-plan, update-plan-step, get-plan-status
- 技能发现：
get-skill-info, list-skills
- 文件系统：
run-shell, read-file, write-file
- 记忆：
search-memory, add-memory
#### 2.2.4 AgentProfile — Agent 蓝图
profile.py 定义了 Agent 的配置模板：
@dataclass
class AgentProfile:
id: str
name: str
skills: list[str] # 技能列表
skills_mode: SkillsMode # 技能模式
custom_prompt: str # 自定义提示词
fallback_profile_id: str # 故障转移目标
icon: str # 显示图标
color: str # 主题色
系统预置 Agent：
default — 默认助手tech_expert — 技术专家boyfriend / girlfriend — 情感陪伴jarvis / butler — 管家式服务business / family — 商务/家庭场景
2.3 委托协作流程
用户: "创建一个竞品分析报告"
│
▼
┌──────────────────────────────┐
│ AgentOrchestrator │
│ 解析任务，识别需要: │
│ - 搜索 Agent │
│ - 写作 Agent │
│ - 验证 Agent │
└──────────────────────────────┘
│
├───────────────────┬───────────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│Search │ │Writing │ │Verify │
│Agent │────────▶│Agent │────────▶│Agent │
│(并行) │ │ │ │ │
└─────────┘ └─────────┘ └─────────┘
│ │ │
└───────────────────┴───────────────────┘
│
▼
┌─────────────┐
│ 返回报告给用户 │
└─────────────┘
2.4 健康度与故障处理
AgentHealth 指标（第 52-63 行）：
@dataclass
class AgentHealth:
total_requests: int # 总请求数
successful: int # 成功数
failed: int # 失败数
avg_latency_ms: float # 平均延迟
@property
def success_rate(self) -> float:
return self.successful / max(self.total_requests, 1)
故障转移：当某个 Agent 连续失败时，自动切换到 fallback_profile_id 指定的备用 Agent。
2.5 多智能体系统亮点与改进建议
| 亮点 | 改进建议 |
|---|
| 轻量级 asyncio 设计 | 可增加分布式部署支持 |
| 委托深度限制防死循环 | 可增加更细粒度的环检测 |
| 空闲回收机制 (30min) | 可增加热备份预加载 |
| 健康度实时追踪 | 可增加告警与自动扩缩容 |
| 故障转移机制 | 可增加熔断器模式 |
---
三、两个系统的协同
记忆系统与多智能体编排并非独立运作，而是紧密协作：
3.1 Agent 使用记忆
# 每个 Agent 都可以访问记忆系统
- search-memory: 检索相关记忆
- add-memory: 添加新记忆
3.2 Agent 私有记忆
通过 MemoryScope.AGENT 实现 Agent 私有记忆：
- 不同的 Agent 可以有不同的"人格"
- 共享全局记忆 (
GLOBAL) 实现跨 Agent 协作
3.3 任务记忆传递
当 Agent 之间进行委托时：
- 任务上下文通过
DelegationRequest 传递
- 记忆检索可以包含委托链上下文
---
四、总结
OpenAkita 的记忆系统和多智能体编排都体现了工程化的成熟度：
1. 记忆系统：v2 架构清晰，多路召回策略合理，AI 提取逻辑务实
2. 多智能体编排：轻量级 asyncio 设计，支持并行协作和故障转移
两个系统都遵循了关注点分离原则，组件边界清晰，便于扩展和维护。整体架构在开源 AI Agent 项目中属于较为完善的实现。