# Phase 5：高级特性（Advanced Features）

日期：2026-04-03
状态：规划中

---

## 1. 阶段目标

Phase 5 包含一系列**高级特性**，在完成 Phase 1-4 后根据实际需求选择性实施。

这些特性不直接影响核心功能，但可以显著提升系统的**可用性、安全性和可扩展性**。

---

## 2. 特性清单

### 2.1 Full Sandbox 隔离

**目标**：实现完整的Docker级隔离

**Phase 4已做**：Worktree隔离

**Phase 5补充**：
- 完整的容器生命周期管理
- 资源限制(CPU/内存/网络)
- 文件系统配额
- 安全策略配置

```python
class FullSandbox:
    """完整沙箱隔离"""
    
    async def execute(
        self,
        task: Task,
        config: SandboxConfig
    ) -> TaskResult:
        """
        在完整沙箱中执行
        
        特点：
        - 完全隔离的网络
        - 资源限制
        - 持久化存储(可选)
        - 安全策略
        """
        
        # 1. 创建容器
        container = await self.client.containers.run(
            image=config.image,
            detach=True,
            mem_limit=config.memory_limit,
            cpu_period=config.cpu_period,
            network_mode="isolated",  # 完全隔离网络
            volumes=config.volumes,
        )
        
        try:
            # 2. 执行任务
            result = await self._execute_in_container(container, task)
            return result
            
        finally:
            # 3. 清理
            await container.remove(force=True)
```

### 2.2 Persistence 层

**目标**：将Event/Message持久化到数据库

**Phase 1-3已做**：内存trace

**Phase 5补充**：
- Event持久化存储
- Message持久化存储
- 支持历史查询
- 数据导出/归档

```python
class EventPersistence:
    """事件持久化"""
    
    async def save_event(self, event: Event):
        """保存事件到数据库"""
        ...
    
    async def query_events(
        self,
        conversation_id: str,
        event_types: list[str] | None = None,
        start_time: datetime | None = None,
        end_time: datetime | None = None,
        limit: int = 100
    ) -> list[Event]:
        """查询历史事件"""
        ...

class MessagePersistence:
    """消息持久化"""
    
    async def save_message(self, message: AgentMessage):
        """保存消息到数据库"""
        ...
    
    async def get_thread_history(
        self,
        thread_id: str,
        limit: int = 100
    ) -> list[AgentMessage]:
        """获取线程历史"""
        ...
```

### 2.3 Multi-turn Memory

**目标**：支持跨会话的长期记忆

**当前**：每个会话独立memory

**Phase 5补充**：
- 重要信息提取
- 跨会话上下文复用
- 知识更新机制
- 遗忘策略

```python
class MultiTurnMemory:
    """跨会话记忆"""
    
    def extract_important_info(
        self,
        conversation_summary: str,
        user_profile: UserProfile
    ) -> list[MemoryEntry]:
        """从对话中提取重要信息"""
        ...
    
    async def get_relevant_context(
        self,
        current_request: str,
        user_id: str
    ) -> list[MemoryEntry]:
        """获取与当前请求相关的记忆"""
        ...
    
    def update_memory(
        self,
        user_id: str,
        new_info: MemoryEntry
    ):
        """更新记忆"""
        ...
    
    def decay_old_memories(self, user_id: str):
        """遗忘旧记忆"""
        ...
```

### 2.4 Cost Monitoring

**目标**：实时监控Token成本

**Phase 3已有**：Budget模型

**Phase 5补充**：
- 实时Token计数
- 成本估算
- 告警机制
- 使用报告

```python
class CostMonitor:
    """成本监控"""
    
    async def track_usage(
        self,
        conversation_id: str,
        model: str,
        input_tokens: int,
        output_tokens: int
    ):
        """跟踪使用量"""
        ...
    
    async def estimate_cost(
        self,
        conversation_id: str
    ) -> CostEstimate:
        """估算当前会话成本"""
        ...
    
    async def check_budget(
        self,
        user_id: str,
        expected_tokens: int
    ) -> bool:
        """检查预算是否足够"""
        ...
    
    async def send_alert(
        self,
        user_id: str,
        threshold: float
    ):
        """发送告警"""
        ...

# 使用示例
@dataclass
class CostEstimate:
    total_tokens: int
    estimated_cost: float
    breakdown: dict[str, int]  # per-model
    threshold_percent: float  # 相对于用户限额
```

### 2.5 Advanced UI

**目标**：完整的前端协作面板

**Phase 4已有**：API

**Phase 5补充**：
- 实时协作拓扑图
- Agent对话界面
- 任务看板
- 成本仪表盘

```
┌────────────────────────────────────────────────────────────────────┐
│  Jarvis 协作面板                                                    │
├────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  ┌─────────────┐    ┌─────────────────────────────────────────────┐ │
│  │ 拓扑图      │    │ 当前会话                                    │ │
│  │             │    │                                             │ │
│  │ [Master]    │    │ User: 帮我分析这个项目...                    │ │
│  │    │        │    │                                             │ │
│  │ [Coord]    │    │ [Coordinator]: 拆分为3个子任务              │ │
│  │  ┌─┴─┐    │    │   - Task 1: 检索相关知识                    │ │
│  │  │   │    │    │   - Task 2: 执行分析                       │ │
│  │ [W1] [W2] │    │   - Task 3: 汇总报告                       │ │
│  │             │    │                                             │ │
│  │ 点击查看详情 │    │ [Worker-1]: 正在检索...                    │ │
│  └─────────────┘    └─────────────────────────────────────────────┘ │
│                                                                     │
│  ┌─────────────┐    ┌─────────────────────────────────────────────┐ │
│  │ 任务列表    │    │ 成本监控                                    │ │
│  │             │    │                                             │ │
│  │ ☑ Task 1   │    │ Token: 12,345 / 50,000                   │ │
│  │ ◐ Task 2   │    │ 成本: $0.23 / $5.00                       │ │
│  │ ○ Task 3   │    │ [████████████████░░░░░] 24%              │ │
│  └─────────────┘    └─────────────────────────────────────────────┘ │
│                                                                     │
└────────────────────────────────────────────────────────────────────┘
```

### 2.6 Plugin System

**目标**：支持第三方插件扩展

**设计参考**：Claude Code CLI的插件系统

```python
class PluginSystem:
    """插件系统"""
    
    async def load_plugin(self, plugin_path: str) -> Plugin:
        """加载插件"""
        ...
    
    async def execute_plugin(
        self,
        plugin_id: str,
        context: dict
    ) -> Any:
        """执行插件"""
        ...

@dataclass
class Plugin:
    """插件定义"""
    plugin_id: str
    name: str
    version: str
    capabilities: list[str]  # 提供的工具/能力
    hooks: list[str]  # 生命周期钩子
    
    async def execute(self, context: dict) -> Any:
        ...

@dataclass
class PluginManifest:
    """插件清单"""
    tools: list[ToolManifest]
    commands: list[CommandManifest]
    hooks: list[str]
```

### 2.7 TagMemo — 仿生记忆系统

**目标**：实现基于遗忘曲线的智能记忆系统

**设计参考**：VCPToolBox的TagMemo V6/V7 RAG系统

**核心概念**：

1. **LIF神经元模型** — 脉冲传播机制
   - 记忆不是静态存储，而是动态激活
   - 重要记忆获得更高的激活频率

2. **Core Tags vs Normal Tags** — 核心记忆
   - Core Tags：获得1.2-1.4x权重加成
   - 核心记忆有虚拟召回能力

3. **遗忘曲线** — 不是无限存储
   - 模拟生物遗忘，不是简单删除
   - 基于重要性动态计算衰减率

```python
class MemoryImportance(str, Enum):
    """记忆重要性等级"""
    CORE = "core"      # 核心记忆，1.2-1.4x权重
    HIGH = "high"      # 高重要性
    MEDIUM = "medium"  # 中等重要性
    LOW = "low"        # 低重要性，会自然遗忘

@dataclass
class TagMemoEntry:
    """TagMemo记忆条目"""
    entry_id: str
    content: str
    
    # VCPToolBox借鉴
    importance: MemoryImportance = MemoryImportance.MEDIUM
    decay_rate: float = 0.1  # 遗忘率
    last_activated: datetime = field(default_factory=datetime.now)
    activation_count: int = 0  # 激活次数
    
    # EPA模块（可选）
    logic_depth: int = 0  # 逻辑深度
    resonance_score: float = 0.0  # 共振分数

class TagMemoMemory:
    """
    仿生记忆系统
    
    特点：
    - 遗忘曲线模拟
    - 重要性权重
    - 动态激活
    """
    
    async def add_memory(
        self,
        content: str,
        importance: MemoryImportance = MemoryImportance.MEDIUM,
        tags: list[str] | None = None
    ) -> TagMemoEntry:
        """添加记忆"""
        entry = TagMemoEntry(
            entry_id=generate_id(),
            content=content,
            importance=importance,
            decay_rate=self._calculate_decay_rate(importance),
            tags=tags or []
        )
        await self._storage.save(entry)
        return entry
    
    def should_retain(self, entry: TagMemoEntry, days_elapsed: int) -> bool:
        """
        判断记忆是否应该保留
        
        基于动态Beta公式：
        β = σ(L·log(1+R) - S·noise_penalty)
        """
        if entry.importance == MemoryImportance.CORE:
            return True  # 核心记忆永远保留
        
        # 遗忘概率 = 基础衰减 × 时间 × 噪声惩罚
        retention_prob = math.exp(
            -entry.decay_rate * days_elapsed * self._noise_factor
        )
        return random.random() < retention_prob
    
    async def get_relevant_memories(
        self,
        query: str,
        limit: int = 5
    ) -> list[TagMemoEntry]:
        """获取相关记忆（带权重）"""
        candidates = await self._vector_search(query, limit=limit * 2)
        
        # 按重要性权重排序
        weighted = []
        for entry in candidates:
            weight = self._calculate_weight(entry)
            weighted.append((entry, weight))
        
        weighted.sort(key=lambda x: x[1], reverse=True)
        return [e for e, _ in weighted[:limit]]
    
    def _calculate_weight(self, entry: TagMemoEntry) -> float:
        """计算记忆权重"""
        base = 1.0
        
        # 重要性权重
        if entry.importance == MemoryImportance.CORE:
            base *= 1.3
        elif entry.importance == MemoryImportance.HIGH:
            base *= 1.1
        
        # 激活频率奖励
        base *= (1 + math.log(1 + entry.activation_count))
        
        # 时间衰减
        days = (datetime.now() - entry.last_activated).days
        base *= math.exp(-0.01 * days)
        
        return base
```

### 2.8 AgentDream — 仿生梦境系统

**目标**：AI在"睡眠"时自动整理和巩固记忆

**设计参考**：VCPToolBox的AgentDream bijective morphic system

**三层时间记忆涟漪**：

| 时间层 | 范围 | 特点 |
|--------|------|------|
| 短期记忆 | 0-7天 | 高频共振，快速激活 |
| 中期记忆 | 7-90天 | 弱共振，需要触发 |
| 长期记忆 | >90天 | 遗忘边界，需要特殊唤醒 |

```python
class DreamLayer(str, Enum):
    """梦境记忆层"""
    SHORT_TERM = "short_term"   # 0-7天
    MID_TERM = "mid_term"       # 7-90天
    LONG_TERM = "long_term"     # >90天

@dataclass
class DreamMemory:
    """梦境记忆结构"""
    layer: DreamLayer
    resonance_bridges: list[str] = field(default_factory=list)  # 共振桥接
    consolidation_level: float = 0.0  # 巩固程度 0-1

class AgentDreamEngine:
    """
    仿生梦境引擎
    
    功能：
    - 定时触发记忆整理
    - 跨层共振发现
    - 遗忘边界管理
    """
    
    async def dream(self, user_id: str) -> DreamReport:
        """
        执行梦境整理
        
        流程：
        1. 获取近期记忆（0-7天）
        2. 与中期记忆建立共振桥
        3. 评估哪些记忆应该升级/遗忘
        4. 生成梦境叙事报告
        """
        short_term = await self._get_memories(
            user_id, 
            layer=DreamLayer.SHORT_TERM
        )
        mid_term = await self._get_memories(
            user_id, 
            layer=DreamLayer.MID_TERM
        )
        
        # 发现共振
        bridges = await self._discover_resonance_bridges(
            short_term, mid_term
        )
        
        # 评估记忆
        to_promote = []  # 升级到中期
        to_forget = []   # 标记遗忘
        to_consolidate = []  # 巩固
        
        for memory in short_term:
            if memory.activation_count > 10:
                to_promote.append(memory)
            elif memory.activation_count < 2:
                to_forget.append(memory)
            else:
                to_consolidate.append(memory)
        
        # 执行整理
        await self._promote_memories(to_promote)
        await self._apply_forgetting(to_forget)
        await self._consolidate(to_consolidate, bridges)
        
        # 生成梦境报告
        return await self._generate_dream_narrative(
            promoted=to_promote,
            forgotten=to_forget,
            bridges=bridges
        )
    
    async def _discover_resonance_bridges(
        self,
        short_term: list[TagMemoEntry],
        mid_term: list[TagMemoEntry]
    ) -> list[tuple[TagMemoEntry, TagMemoEntry, float]]:
        """发现跨层共振桥接"""
        bridges = []
        
        for s in short_term:
            for m in mid_term:
                similarity = await self._calculate_resonance(s, m)
                if similarity > 0.7:
                    bridges.append((s, m, similarity))
        
        return bridges
    
    async def _generate_dream_narrative(
        self,
        promoted: list,
        forgotten: list,
        bridges: list
    ) -> DreamReport:
        """生成第一人称梦境叙事"""
        narrative = f"""
        【梦境记录】
        
        今夜整理了{len(promoted)}段重要记忆，它们已经被巩固：
        {', '.join(m.content for m in promoted[:3])}
        
        发现了{len(bridges)}条记忆共振：
        {self._describe_bridges(bridges)}
        
        {len(forgotten)}段记忆正在消散...
        """
        
        return DreamReport(
            narrative=narrative,
            promoted_count=len(promoted),
            forgotten_count=len(forgotten),
            bridges_count=len(bridges)
        )
```

**触发机制**：

```python
# 定时任务：每天凌晨3点执行梦境整理
@scheduler.scheduled(cron="0 3 * * *")
async def scheduled_agent_dream():
    """AI睡眠时整理记忆"""
    for user_id in await get_active_users():
        try:
            report = await agent_dream_engine.dream(user_id)
            logger.info(f"Dream complete for {user_id}: {report.summary}")
        except Exception as e:
            logger.error(f"Dream failed for {user_id}: {e}")
```

---

## 3. 实施优先级

| 特性 | 优先级 | 依赖 | 建议实施时机 |
|------|--------|------|--------------|
| Cost Monitoring | 🔴 高 | Phase 3 | 正式上线前 |
| TagMemo | 🟡 中 | Phase 2 | 用户反馈需要更好记忆时 |
| AgentDream | 🟢 低 | Phase 5+TagMemo | 凌晨调度资源时 |
| Persistence | 🟡 中 | Phase 1-3 | 有审计需求时 |
| Multi-turn Memory | 🟡 中 | Phase 1-2 | 用户反馈需要时 |
| Advanced UI | 🟡 中 | Phase 4 | 有前端资源时 |
| Full Sandbox | 🟢 低 | Phase 4 | 有安全需求时 |
| Plugin System | 🟢 低 | Phase 1 | 有社区需求时 |

---

## 4. 风险点

| 风险 | 缓解措施 |
|------|----------|
| 功能蔓延 | 严格控制每个特性的scope |
| 性能影响 | Persistence要考虑异步和索引优化 |
| 成本增加 | Full Sandbox资源限制要明确 |

---

## 5. 验收标准

| 特性 | 验收标准 |
|------|----------|
| Cost Monitoring | 能实时显示Token使用量和估算成本 |
| Persistence | 事件和消息可持久化存储和查询 |
| Multi-turn Memory | 跨会话可复用关键信息 |
| Advanced UI | 有可用的前端协作面板 |
| Full Sandbox | 容器隔离完整，资源限制生效 |
| Plugin System | 插件可加载和执行 |

---

## 6. 本阶段完成后预期结果

完成后，Jarvis 将具备：

- ✅ 完整的成本监控能力
- ✅ 仿生记忆系统（TagMemo）
- ✅ AI梦境整理（AgentDream）
- ✅ 历史数据持久化
- ✅ 跨会话的智能记忆
- ✅ 完整的协作可视化UI
- ✅ 高级隔离执行环境
- ✅ 可扩展的插件系统

**Phase 5 为可选特性，根据实际需求选择性实施。**