JARVIS/development-doc/plan/agent-update/phase-5-advanced-features.md

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/内存/网络)
• 文件系统配额
• 安全策略配置

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持久化存储
• 支持历史查询
• 数据导出/归档

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补充：

• 重要信息提取
• 跨会话上下文复用
• 知识更新机制
• 遗忘策略

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计数
• 成本估算
• 告警机制
• 使用报告

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的插件系统

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. 遗忘曲线 — 不是无限存储

   • 模拟生物遗忘，不是简单删除
   • 基于重要性动态计算衰减率

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天	遗忘边界，需要特殊唤醒

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)
        )

触发机制：

# 定时任务：每天凌晨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 为可选特性，根据实际需求选择性实施。