Claude Code 源码泄露完整技术分析

泄露时间: 2026-03-31
代码规模: 50万行 TypeScript，2000+ 文件
GitHub: instructkr/claude-code (6.4万 stars)

Phase 1: 协同机制深挖分析报告

生成时间: 2026-04-01
源码状态: TypeScript 源码已移除，基于架构快照分析
关键子系统: AgentTool (19 模块)、Coordinator (1 模块)、Hooks/ToolPermission (104 模块)

一、AgentTool 架构分析 ⭐⭐⭐⭐⭐

1.1 模块清单

基于 tools_snapshot.json，AgentTool 包含 19 个核心模块：


tools/AgentTool/
├── AgentTool.tsx          # 主入口 - 工具定义与执行
├── UI.tsx                 # React 终端渲染组件
├── agentColorManager.ts   # 颜色编码管理（区分不同代理）
├── agentDisplay.ts        # 显示逻辑（终端输出格式化）
├── agentMemory.ts         # 代理记忆系统 ⭐
├── agentMemorySnapshot.ts # 记忆快照（持久化与恢复）
├── agentToolUtils.ts      # 工具辅助函数
├── constants.ts           # 常量定义
├── forkSubagent.ts        # 分离子代理 ⭐⭐⭐⭐⭐
├── loadAgentsDir.ts       # 加载代理目录
├── prompt.ts              # 提示词生成
├── resumeAgent.ts         # 恢复代理执行 ⭐
├── runAgent.ts            # 运行代理核心逻辑
├── builtInAgents.ts       # 内置代理注册
│
└── built-in/              # 内置代理实现（7 种）
    ├── claudeCodeGuideAgent.ts  # Claude Code 指导代理
    ├── exploreAgent.ts          # 探索代理（代码库探索）
    ├── planAgent.ts             # 规划代理（任务规划）
    ├── generalPurposeAgent.ts   # 通用代理（灵活任务）
    ├── verificationAgent.ts     # 验证代理（结果检查）
    ├── statuslineSetup.ts       # 状态栏设置代理
    └── builtInAgents.ts         # 内置代理汇总

1.2 核心机制分析

1.2.1 forkSubagent - 子代理生成 ⭐⭐⭐⭐⭐

设计模式: 主代理 → 分支子代理 → 异步执行 → 结果聚合

typescript
// 伪代码（基于架构快照推断）
interface ForkSubagentParams {
  agentType: 'explore' | 'plan' | 'verify' | 'general';
  task: string;
  context: SessionContext;
  memorySnapshot?: AgentMemorySnapshot;
}

async function forkSubagent(params: ForkSubagentParams) {
  // 1. 创建隔离的子会话
  const sessionId = generateSessionId();
  
  // 2. 注入记忆快照（如果有）
  if (params.memorySnapshot) {
    await restoreMemory(sessionId, params.memorySnapshot);
  }
  
  // 3. 配置工具权限（继承父代理）
  const permissionContext = inheritPermissions(params.context);
  
  // 4. 启动异步执行
  const agentProcess = spawnAgentProcess({
    sessionId,
    agentType: params.agentType,
    task: params.task,
    permissionContext
  });
  
  // 5. 返回会话 ID 用于后续查询
  return {
    sessionId,
    status: 'running',
    color: assignAgentColor() // 颜色编码区分
  };
}

Python 调度器可复用:

python
# 直接对应 sessions_spawn
def fork_subagent(self, task: str, agent_type: str = "general"):
    return sessions_spawn(
        runtime="subagent",
        task=task,
        agentId=agent_type,
        mode="run"  # one-shot 执行
    )

1.2.2 agentMemory - 代理记忆系统 ⭐⭐⭐⭐

设计理念: 子代理拥有独立记忆，可与主代理共享关键信息


记忆层级结构：
┌─────────────────────────────────┐
│ 主代理记忆 (SessionMemory)      │
│  - 项目背景                     │
│  - 用户偏好                     │
│  - 工作历史                     │
└─────────────────────────────────┘
         ↓ 注入关键信息
┌─────────────────────────────────┐
│ 子代理记忆 (AgentMemory)        │
│  - 任务上下文                   │
│  - 中间结果                     │
│  - 发现的知识                   │
└─────────────────────────────────┘
         ↓ 快照保存
┌─────────────────────────────────┐
│ 记忆快照 (AgentMemorySnapshot)  │
│  - 可恢复的状态                 │
│  - 跨会话持久化                 │
└─────────────────────────────────┘

关键模块交互:

typescript
// agentMemory.ts - 记忆管理
interface AgentMemory {
  store(key: string, value: any): void;
  retrieve(key: string): any;
  snapshot(): AgentMemorySnapshot;
  restore(snapshot: AgentMemorySnapshot): void;
}

// agentMemorySnapshot.ts - 快照持久化
interface AgentMemorySnapshot {
  sessionId: string;
  timestamp: number;
  memoryEntries: Map<string, any>;
  toStorageFormat(): string; // JSON 序列化
  fromStorageFormat(data: string): AgentMemorySnapshot;
}

Python 实现参考:

python
class AgentMemory:
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.entries = {}
    
    def store(self, key: str, value: Any):
        self.entries[key] = {
            "value": value,
            "timestamp": datetime.now()
        }
    
    def snapshot(self) -> AgentMemorySnapshot:
        return AgentMemorySnapshot(
            session_id=self.session_id,
            entries=self.entries.copy()
        )

class AgentMemorySnapshot:
    def to_json(self) -> str:
        return json.dumps({
            "session_id": self.session_id,
            "entries": self.entries,
            "timestamp": datetime.now().isoformat()
        })
    
    @classmethod
    def from_json(cls, data: str):
        parsed = json.loads(data)
        return cls(
            session_id=parsed["session_id"],
            entries=parsed["entries"]
        )

1.2.3 resumeAgent - 代理恢复机制 ⭐

设计模式: 从快照恢复 → 继续执行 → 完成任务

typescript
// resumeAgent.ts - 恢复代理执行
async function resumeAgent(sessionId: string) {
  // 1. 加载会话历史
  const sessionHistory = loadSessionHistory(sessionId);
  
  // 2. 恢复记忆快照
  const snapshot = await loadMemorySnapshot(sessionId);
  const memory = AgentMemory.restore(snapshot);
  
  // 3. 重建上下文
  const context = rebuildContext(sessionHistory, memory);
  
  // 4. 继续执行
  return runAgent(context);
}

Python 调度器参考:

python
def resume_agent(self, session_id: str):
    # 对应 sessions_spawn 的 resumeSessionId 参数
    return sessions_spawn(
        runtime="acp",
        resumeSessionId=session_id,
        task="继续执行任务"
    )

1.3 内置代理类型分析

代理类型	文件	核心功能	Python 对应
exploreAgent	built-in/exploreAgent.ts	代码库探索、文件搜索、依赖分析	`subagent` + GlobTool/GrepTool
planAgent	built-in/planAgent.ts	任务规划、方案设计、步骤分解	`subagent` + 独立思考模式
verificationAgent	built-in/verificationAgent.ts	结果验证、代码检查、测试运行	`subagent` + BashTool 测试
generalPurposeAgent	built-in/generalPurposeAgent.ts	通用任务执行、灵活适配	`subagent` 默认模式
claudeCodeGuideAgent	built-in/claudeCodeGuideAgent.ts	Claude Code 使用指导	技能系统
statuslineSetup	built-in/statuslineSetup.ts	状态栏配置	UI 配置

1.4 颜色编码系统

agentColorManager.ts - 区分不同代理的终端显示


代理颜色分配：
┌─────────────────────────────────┐
│ 主代理    - 默认颜色（白色）     │
│ 探索代理  - 蓝色 (#3B82F6)       │
│ 规划代理  - 绿色 (#10B981)       │
│ 验证代理  - 紫色 (#8B5CF6)       │
│ 通用代理  - 黄色 (#F59E0B)       │
└─────────────────────────────────┘

Python 实现:

python
class AgentColorManager:
    COLORS = {
        "main": "\033[37m",      # 白色
        "explore": "\033[34m",   # 蓝色
        "plan": "\033[32m",      # 绿色
        "verify": "\033[35m",    # 紫色
        "general": "\033[33m"    # 黄色
    }
    
    def get_color(self, agent_type: str) -> str:
        return self.COLORS.get(agent_type, self.COLORS["main"])

二、Coordinator 模式分析 ⭐⭐⭐⭐⭐

2.1 架构快照

文件: coordinator/coordinatorMode.ts (单文件，精简设计)

子系统: coordinator (1 个模块)

2.2 设计模式

主从架构: 主 Claude + 多工人 Agent 并行调度


┌──────────────────────────────────────────────────────┐
│                    Coordinator                       │
│  ┌────────────────────────────────────────────────┐ │
│  │ 主 Claude (决策者)                              │ │
│  │  - 分析任务                                     │ │
│  │  - 分配子任务                                   │ │
│  │  - 聚合结果                                     │ │
│  │  - 最终决策                                     │ │
│  └────────────────────────────────────────────────┘ │
│                                                       │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐          │
│  │ Agent 1  │  │ Agent 2  │  │ Agent 3  │          │
│  │ (探索)   │  │ (规划)   │  │ (验证)   │          │
│  │ 异步执行 │  │ 异步执行 │  │ 异步执行 │          │
│  └──────────┘  └──────────┘  └──────────┘          │
│       ↓             ↓             ↓                  │
│  ┌──────────────────────────────────────────────┐  │
│  │ 结果聚合                                      │  │
│  │  - 探索结果 + 规划方案 + 验证报告            │  │
│  └──────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────┘

2.3 权限钩子集成

hooks/toolPermission/handlers/coordinatorHandler.ts

typescript
// coordinatorHandler.ts - Coordinator 权限控制
interface CoordinatorPermission {
  maxAgents: number;        // 最大子代理数量
  allowedAgentTypes: string[]; // 允许的代理类型
  resourceLimits: {
    maxTokens: number;
    maxTime: number;
  };
}

function handleCoordinatorPermission(
  context: PermissionContext,
  params: CoordinatorParams
): PermissionResult {
  // 1. 检查代理数量限制
  if (params.agentCount > context.limits.maxAgents) {
    return { allowed: false, reason: "超过最大代理数量" };
  }
  
  // 2. 检查代理类型权限
  for (const agentType of params.agentTypes) {
    if (!context.limits.allowedAgentTypes.includes(agentType)) {
      return { allowed: false, reason: `不允许的代理类型: ${agentType}` };
    }
  }
  
  // 3. 检查资源限制
  if (params.estimatedTokens > context.limits.maxTokens) {
    return { allowed: false, reason: "超过 Token 限制" };
  }
  
  return { allowed: true };
}

2.4 Python 调度器实现参考 ⭐⭐⭐⭐⭐

python
class Coordinator:
    """多代理协调器 - Claude Code Coordinator 模式实现"""
    
    def __init__(self, max_agents: int = 5):
        self.max_agents = max_agents
        self.active_agents = {}
        self.results_queue = []
    
    async def dispatch_task(self, task: str) -> dict:
        """分发任务到多个子代理"""
        
        # 1. 分析任务，确定需要的代理类型
        agent_types = self._analyze_task_requirements(task)
        
        # 2. 并行启动子代理
        sessions = []
        for agent_type in agent_types:
            session = sessions_spawn(
                runtime="subagent",
                task=self._create_subtask(task, agent_type),
                agentId=agent_type,
                mode="run"
            )
            sessions.append(session)
            self.active_agents[session.session_id] = {
                "type": agent_type,
                "status": "running"
            }
        
        # 3. 等待所有代理完成
        results = await self._collect_results(sessions)
        
        # 4. 聚合结果
        return self._aggregate_results(results)
    
    def _analyze_task_requirements(self, task: str) -> List[str]:
        """分析任务，确定需要的代理类型"""
        # 简单规则：复杂任务需要探索+规划+验证
        if "重构" in task or "架构" in task:
            return ["explore", "plan", "verify"]
        elif "实现" in task:
            return ["explore", "general"]
        else:
            return ["general"]
    
    async def _collect_results(self, sessions: List) -> List[dict]:
        """收集所有子代理结果"""
        results = []
        for session in sessions:
            # 轮询直到完成
            while True:
                status = subagents(action="list", recent_minutes=1)
                if session.session_id not in status["active"]:
                    break
                await asyncio.sleep(5)
            
            # 获取结果
            history = sessions_history(session.session_id)
            results.append({
                "session_id": session.session_id,
                "output": history[-1]["content"],
                "agent_type": self.active_agents[session.session_id]["type"]
            })
        
        return results
    
    def _aggregate_results(self, results: List[dict]) -> dict:
        """聚合多代理结果"""
        aggregated = {
            "exploration": [],
            "planning": [],
            "verification": [],
            "general": []
        }
        
        for result in results:
            agent_type = result["agent_type"]
            aggregated[agent_type].append(result["output"])
        
        return aggregated

可复用点标注:

✅ fork_subagent - 直接使用 sessions_spawn
✅ 颜色管理 - Python 终端颜色编码
✅ 结果聚合 - sessions_history + subagents
⚠️ 记忆快照 - 需要自己实现（参考 AgentMemory）
⚠️ 权限钩子 - 需要自己实现（参考 coordinatorHandler）

三、权限系统分析 ⭐⭐⭐⭐

3.1 Hooks 子系统架构

模块数量: 104 个文件

关键目录:


hooks/
├── toolPermission/                    # 工具权限管理 ⭐⭐⭐⭐⭐
│   ├── PermissionContext.ts          # 权限上下文
│   ├── handlers/                      # 权限处理器
│   │   ├── coordinatorHandler.ts     # Coordinator 权限
│   │   ├── interactiveHandler.ts     # 交互式权限
│   │   └── swarmWorkerHandler.ts     # 群组工人权限
│   └── permissionLogging.ts          # 权限日志
│
├── notifs/                            # 通知系统
│   ├── useAutoModeUnavailableNotification.tsx
│   ├── useDeprecationWarningNotification.tsx
│   ├── useRateLimitWarningNotification.tsx
│   ├── usePluginInstallationStatus.tsx
│   └── useMcpConnectivityStatus.tsx
│
├── fileSuggestions.ts                 # 文件建议
├── unifiedSuggestions.ts              # 统一建议系统
└── useAfterFirstRender.ts             # 渲染钩子

3.2 PermissionContext 设计

typescript
// PermissionContext.ts
interface PermissionContext {
  // 用户权限配置
  userPermissions: {
    allowedTools: string[];
    deniedTools: string[];
    restrictedPrefixes: string[];
  };
  
  // 会话权限继承
  sessionInheritance: {
    parentSessionId?: string;
    inheritedPermissions: string[];
  };
  
  // 动态权限调整
  dynamicAdjustment: {
    temporaryAllow: string[];
    temporaryDeny: string[];
    expiryTime: number;
  };
  
  // 权限检查方法
  checkPermission(toolName: string): PermissionResult;
  requestPermission(toolName: string, reason: string): Promise<boolean>;
}

3.3 权限处理器模式

三种权限处理器:

coordinatorHandler.ts - 多代理权限

typescript
- 代理数量限制
- 代理类型白名单
- 资源配额控制

interactiveHandler.ts - 交互式权限

typescript
- 用户实时授权
- 临时权限授予
- 权限过期机制

swarmWorkerHandler.ts - 群组工人权限

typescript
- 批量任务权限
- 并发限制
- 任务队列管理

3.4 Python 权限系统参考

python
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta

@dataclass
class PermissionContext:
    """权限上下文 - Claude Code PermissionContext 参考实现"""
    
    allowed_tools: List[str]
    denied_tools: List[str]
    restricted_prefixes: List[str]
    
    # 动态权限
    temporary_allow: List[str] = []
    temporary_deny: List[str] = []
    expiry_time: Optional[datetime] = None
    
    def check_permission(self, tool_name: str) -> dict:
        """检查工具权限"""
        
        # 1. 检查永久拒绝
        if tool_name in self.denied_tools:
            return {
                "allowed": False,
                "reason": f"工具 {tool_name} 在拒绝列表中"
            }
        
        # 2. 检查前缀限制
        for prefix in self.restricted_prefixes:
            if tool_name.startswith(prefix):
                return {
                    "allowed": False,
                    "reason": f"工具 {tool_name} 匹配限制前缀 {prefix}"
                }
        
        # 3. 检查临时权限（如果未过期）
        if self.expiry_time and datetime.now() < self.expiry_time:
            if tool_name in self.temporary_deny:
                return {"allowed": False, "reason": "临时拒绝"}
            if tool_name in self.temporary_allow:
                return {"allowed": True}
        
        # 4. 检查允许列表
        if tool_name in self.allowed_tools:
            return {"allowed": True}
        
        # 5. 默认：需要用户授权
        return {
            "allowed": False,
            "reason": "需要用户授权",
            "requires_approval": True
        }
    
    def request_permission(
        self,
        tool_name: str,
        reason: str,
        duration_minutes: int = 60
    ) -> bool:
        """请求用户授权临时权限"""
        
        # 实际应用中这里应该调用用户确认机制
        # 比如 AskUserQuestionTool
        
        granted = self._ask_user(f"是否允许使用工具 {tool_name}? 原因: {reason}")
        
        if granted:
            self.temporary_allow.append(tool_name)
            self.expiry_time = datetime.now() + timedelta(minutes=duration_minutes)
        
        return granted
    
    def _ask_user(self, question: str) -> bool:
        """用户确认（实际实现）"""
        # 这里应该集成 OpenClaw 的 AskUserQuestionTool
        # 或者通过 message 工具询问用户
        pass


class CoordinatorPermissionHandler:
    """Coordinator 权限处理器"""
    
    def __init__(self, max_agents: int = 5, max_tokens: int = 100000):
        self.max_agents = max_agents
        self.max_tokens = max_tokens
        self.allowed_agent_types = ["explore", "plan", "verify", "general"]
    
    def handle(self, params: dict, context: PermissionContext) -> dict:
        """处理 Coordinator 权限请求"""
        
        # 1. 检查代理数量
        if params.get("agent_count", 0) > self.max_agents:
            return {
                "allowed": False,
                "reason": f"超过最大代理数量限制 ({self.max_agents})"
            }
        
        # 2. 检查代理类型
        for agent_type in params.get("agent_types", []):
            if agent_type not in self.allowed_agent_types:
                return {
                    "allowed": False,
                    "reason": f"不允许的代理类型: {agent_type}"
                }
        
        # 3. 检查资源配额
        if params.get("estimated_tokens", 0) > self.max_tokens:
            return {
                "allowed": False,
                "reason": f"超过 Token 配额限制 ({self.max_tokens})"
            }
        
        return {"allowed": True}

四、可复用设计清单 ⭐⭐⭐⭐⭐

4.1 直接可复用（无需修改）

设计点	Claude Code 实现	Python 调度器对应	复用难度
forkSubagent	AgentTool.tsx → forkSubagent.ts	`sessions_spawn(runtime="subagent")`	⭐ 极简单
颜色管理	agentColorManager.ts	Python 终端 ANSI 颜色	⭐ 极简单
结果聚合	AgentTool.tsx → runAgent.ts	`sessions_history` + `subagents list`	⭐ 极简单
恢复机制	resumeAgent.ts	`sessions_spawn(resumeSessionId=...)`	⭐ 极简单

4.2 需要适配实现

设计点	Claude Code 实现	Python 实现思路	复用难度
AgentMemory	agentMemory.ts + agentMemorySnapshot.ts	自定义 Memory 类 + JSON 持久化	⭐⭐ 中等
PermissionContext	hooks/toolPermission/PermissionContext.ts	自定义权限类 + OpenClaw 集成	⭐⭐ 中等
CoordinatorHandler	coordinatorHandler.ts	自定义权限处理器	⭐⭐ 中等

4.3 架构级参考

设计点	Claude Code 架构	Python 调度器架构参考	复用难度
主从模式	Coordinator + Workers	主调度器 + 多 subagent	⭐⭐⭐ 需架构设计
记忆层级	SessionMemory → AgentMemory → Snapshot	三层记忆架构设计	⭐⭐⭐ 需架构设计
权限钩子	hooks/toolPermission/*handlers	权限系统钩子设计	⭐⭐⭐ 需架构设计

4.4 Python 调度器改造建议 ⭐⭐⭐⭐⭐

基于 Claude Code 协同机制的 Python 调度器架构:


python-dispatcher/
├── coordinator/                   # Coordinator 模式
│   ├── __init__.py
│   ├── coordinator.py            # 主协调器
│   ├── agent_manager.py          # 代理管理
│   └── result_aggregator.py      # 结果聚合
│
├── memory/                        # 记忆系统
│   ├── __init__.py
│   ├── agent_memory.py           # 代理记忆
│   ├── memory_snapshot.py        # 记忆快照
│   └── session_memory.py         # 会话记忆（继承 SessionMemory）
│
├── permissions/                   # 权限系统
│   ├── __init__.py
│   ├── context.py                # PermissionContext
│   ├── handlers/
│   │   ├── coordinator_handler.py
│   │   ├── interactive_handler.py
│   │   └── swarm_handler.py
│   └── logging.py                # 权限日志
│
├── agents/                        # 内置代理
│   ├── __init__.py
│   ├── explore_agent.py          # 探索代理
│   ├── plan_agent.py             # 规划代理
│   ├── verify_agent.py           # 验证代理
│   └── general_agent.py          # 通用代理
│
├── ui/                            # 终端 UI
│   ├── __init__.py
│   ├── color_manager.py          # 颜色管理
│   └── display.py                # 显示逻辑
│
└── main.py                        # CLI 入口（参考 Phase 2）

五、关键发现总结

5.1 Claude Code 协同机制精髓

异步子代理执行 - forkSubagent + resumeAgent 双机制
颜色编码区分 - agentColorManager 终端友好
记忆快照恢复 - agentMemorySnapshot 跨会话持久
权限钩子控制 - toolPermission 多层级管理
主从并行架构 - Coordinator 高效调度

5.2 Python 调度器可借鉴点

直接使用 OpenClaw 工具:
- sessions_spawn - 对应 forkSubagent
- sessions_history - 对应结果聚合
- subagents - 对应代理管理
需要自己实现:
- AgentMemory 类 - 记忆管理
- PermissionContext - 权限系统
- Coordinator 类 - 多代理调度逻辑
架构级改造:
- 三层记忆架构（会话 → 代理 → 快照）
- 权限钩子系统（多处理器模式）
- CLI 页面入口 + while loop（Phase 2 分析）

六、下一步建议

6.1 立即可实施

Coordinator 基础版本 - 基于 sessions_spawn 实现简单调度
颜色管理 - Python ANSI 颜色编码
权限基础类 - PermissionContext 骨架

6.2 需要深度开发

AgentMemory 系统 - 记忆快照完整实现
权限钩子系统 - 多处理器集成
内置代理 - explore/plan/verify 代理逻辑

6.3 架构级重构

Python 调度器目录结构 - 参考 4.4 建议
CLI 入口改造 - Phase 2 分析（Ink 渲染 + while loop）
记忆系统集成 - Phase 3 分析（KAIROS + autoDream）

报告状态: Phase 1 完成
下一步: Phase 2 - CLI 架构深挖
生成时间: 2026-04-01 10:15

Phase 2: CLI 架构深挖分析报告

生成时间: 2026-04-01
源码状态: TypeScript 源码已移除，基于架构快照分析
关键子系统: Entrypoints (8 模块)、CLI (19 模块)、Screens (3 模块)、Components (389 模块)、State (6 模块)

一、Ink 渲染框架分析 ⭐⭐⭐⭐⭐

1.1 技术栈核心

Claude Code 使用 React 19 + Ink 实现终端 UI

Ink 特性:

React 组件模型 → 终端渲染
声明式 UI 编程
Flexbox 布局系统（类似 CSS）
异步渲染支持

架构快照证据:


components/ (389 个 React 组件)
├── App.tsx                    # 主应用组件
├── AgentProgressLine.tsx      # 代理进度显示
├── CoordinatorAgentStatus.tsx # Coordinator 状态组件
├── REPL.tsx                   # REPL 主界面（screens）
└── ... 385+ 更多组件

1.2 Ink 渲染原理

typescript
// 伪代码（基于 React + Ink 架构推断）
import { render } from 'ink';
import { App } from './components/App';

function startCLI() {
  // 1. 创建 React 应用
  const app = render(<App />);
  
  // 2. Ink 渲染到终端
  // 3. 处理键盘事件
  // 4. 管理组件生命周期
}

关键组件分析:

1.2.1 App.tsx - 主应用容器

typescript
// components/App.tsx 伪代码
interface AppProps {
  initialMode: 'repl' | 'doctor' | 'resume';
  sessionId?: string;
}

function App({ initialMode, sessionId }: AppProps) {
  const [currentScreen, setScreen] = useState(initialMode);
  const [appState, setAppState] = useAppStateStore();
  
  // 路由逻辑
  return (
    <Box flexDirection="column">
      {currentScreen === 'repl' && <REPLScreen />}
      {currentScreen === 'doctor' && <DoctorScreen />}
      {currentScreen === 'resume' && <ResumeConversation sessionId={sessionId} />}
    </Box>
  );
}

1.2.2 REPL.tsx - REPL 主界面 ⭐⭐⭐⭐⭐

typescript
// screens/REPL.tsx 伪代码
function REPLScreen() {
  const [messages, addMessage] = useMessages();
  const [isProcessing, setIsProcessing] = useState(false);
  
  // 输入处理循环
  const handleInput = async (input: string) => {
    setIsProcessing(true);
    
    // 1. 调用 Claude API
    const response = await claudeAPI.sendMessage(input);
    
    // 2. 渲染响应
    addMessage(response);
    
    // 3. 执行工具（如果有）
    if (response.toolUse) {
      await executeTool(response.toolUse);
    }
    
    setIsProcessing(false);
  };
  
  return (
    <Box flexDirection="column">
      {/* 消息历史 */}
      <MessagesDisplay messages={messages} />
      
      {/* 进度显示 */}
      {isProcessing && <AgentProgressLine />}
      
      {/* 输入框 */}
      <InputBox onSubmit={handleInput} />
    </Box>
  );
}

1.2.3 Ink 组件库

389 个组件的设计模式:


components/
├── 基础组件
│   ├── BaseTextInput.tsx      # 基础文本输入
│   ├── ConfigurableShortcutHint.tsx  # 快捷键提示
│   └── ClickableImageRef.tsx  # 可点击图片引用
│
├── 对话框组件
│   ├── ApproveApiKey.tsx      # API Key 授权
│   ├── AutoModeOptInDialog.tsx  # 自动模式确认
│   ├── BypassPermissionsModeDialog.tsx  # 绕过权限确认
│   ├── CostThresholdDialog.tsx  # 成本阈值警告
│   └── BridgeDialog.tsx       # 桥接对话框
│
├── 状态显示组件
│   ├── AgentProgressLine.tsx  # 代理进度线
│   ├── CoordinatorAgentStatus.tsx  # Coordinator 状态
│   ├── AwsAuthStatusBox.tsx   # AWS 认证状态
│   └── ContextVisualization.tsx  # 上下文可视化
│
├── 交互组件
│   ├── AutoUpdater.tsx        # 自动更新
│   ├── AutoUpdaterWrapper.tsx # 更新包装器
│   └── ConsoleOAuthFlow.tsx   # OAuth 流程
│
└── 特殊组件
    ├── Buddy/CompanionSprite.tsx  # BUDDY 精灵渲染 ⭐
    └── ClaudeInChromeOnboarding.tsx  # Chrome 集成引导

1.3 Python 终端 UI 参考 ⭐⭐⭐⭐

Ink 对应的 Python 方案:

方案	特性	对应 Ink 能力
Rich	富文本渲染、进度条、表格	部分对应（无 React 模型）
Textual	TUI 框架、组件化、CSS 布局	⭐⭐⭐⭐ 高度对应
PyInquirer	交互式问答	对应对话框组件
asciimatics	动画效果	对应 Buddy 精灵

推荐方案: Textual (最佳 Ink 对应)

python
# Textual 示例（对应 REPL.tsx）
from textual.app import App, ComposeResult
from textual.widgets import Input, Static
from textual.containers import Vertical

class REPLApp(App):
    """REPL 主界面 - 对应 screens/REPL.tsx"""
    
    CSS = """
    Vertical {
        layout: vertical;
    }
    .message-history {
        height: 80%;
    }
    .input-box {
        height: 20%;
    }
    """
    
    def compose(self) -> ComposeResult:
        return Vertical(
            Static(id="message-history", classes="message-history"),
            Input(placeholder="输入任务...", id="input-box", classes="input-box")
        )
    
    def on_input_submitted(self, event: Input.Submitted):
        """处理输入 - 对应 handleInput"""
        input_text = event.value
        
        # 1. 显示用户输入
        self.query_one("#message-history").mount(
            Static(f"用户: {input_text}")
        )
        
        # 2. 调用调度器
        result = self.dispatcher.process(input_text)
        
        # 3. 显示响应
        self.query_one("#message-history").mount(
            Static(f"Claude: {result}")
        )

二、入口循环模式分析 ⭐⭐⭐⭐⭐

2.1 While Loop 架构

entrypoints/cli.tsx - 主入口文件

typescript
// entrypoints/cli.tsx 伪代码（推断）
import { render } from 'ink';
import { App } from './components/App';
import { PortRuntime } from './runtime';

async function main() {
  // 1. 初始化运行时
  const runtime = await PortRuntime.initialize();
  
  // 2. 启动 Ink 应用
  const { unmount } = render(<App runtime={runtime} />);
  
  // 3. While Loop 模式
  while (runtime.isActive) {
    // a. 渲染 UI（Ink 自动处理）
    // b. 处理用户输入（事件驱动）
    // c. 调用工具执行
    // d. 更新状态
    
    await runtime.waitForNextTurn();
  }
  
  // 4. 清理资源
  unmount();
  runtime.shutdown();
}

2.2 Turn Loop 实现

Python 移植版参考 (src/runtime.py):

python
# src/runtime.py (移植版伪代码)
class PortRuntime:
    """Claude Code Runtime - while loop 核心"""
    
    def __init__(self):
        self.is_active = True
        self.session_store = SessionStore()
        self.tool_pool = assemble_tool_pool()
    
    def run_turn_loop(self, prompt: str, max_turns: int = 3):
        """Turn Loop 实现 - 对应 while loop"""
        
        results = []
        for turn in range(max_turns):
            # 1. 处理输入
            message = self.process_input(prompt)
            
            # 2. 调用 Claude API
            response = self.claude_api.send(message)
            
            # 3. 执行工具（如果有）
            if response.tool_use:
                tool_result = self.execute_tool(
                    response.tool_use.name,
                    response.tool_use.params
                )
                
                # 4. 工具结果反馈
                message = self.create_tool_result_message(tool_result)
                
                # 5. 继续下一轮
                prompt = message
            
            # 6. 记录结果
            results.append({
                "turn": turn + 1,
                "output": response.content,
                "stop_reason": response.stop_reason
            })
            
            # 7. 判断是否结束
            if response.stop_reason == "end_turn":
                break
        
        return results
    
    def execute_tool(self, name: str, params: dict):
        """工具执行 - 对应 ToolPool"""
        tool = self.tool_pool.get(name)
        return tool.call(params)

2.3 Python 调度器改造参考 ⭐⭐⭐⭐⭐

基于 Claude Code 的 Python 调度器架构:

python
# python-dispatcher/main.py (改造方案)
import asyncio
from textual.app import App
from dispatcher import Coordinator

class DispatcherApp(App):
    """调度器主入口 - 对应 entrypoints/cli.tsx"""
    
    async def main(self):
        """主循环 - while loop 模式"""
        
        # 1. 初始化 Coordinator
        coordinator = Coordinator(max_agents=5)
        
        # 2. 启动 Textual UI
        await self.run_async()
        
        # 3. While Loop
        while coordinator.is_active:
            # a. 从 UI 获取任务
            task = await self.ui.get_next_task()
            
            if not task:
                # 等待下一个任务
                await asyncio.sleep(1)
                continue
            
            # b. 分发任务
            result = await coordinator.dispatch_task(task)
            
            # c. 显示结果
            await self.ui.display_result(result)
            
            # d. 更新状态
            coordinator.update_state(result)

关键改造点:

Claude Code	Python 调度器	改造难度
Ink + React	Textual	⭐⭐⭐ 中等
cli.tsx while loop	asyncio while loop	⭐⭐ 简单
PortRuntime.run_turn_loop	Coordinator.dispatch_task	⭐⭐ 简单
ToolPool.execute_tool	sessions_spawn + subagents	⭐ 极简单

三、CLI 页面入口设计 ⭐⭐⭐⭐

3.1 Screens 架构

3 个主要页面:


screens/
├── Doctor.tsx             # 诊断页面（环境检查）
├── REPL.tsx               # REPL 主界面 ⭐⭐⭐⭐⭐
└── ResumeConversation.tsx # 恢复会话页面

3.2 页面路由模式

typescript
// components/App.tsx 路由逻辑（推断）
function App() {
  const [currentScreen, setScreen] = useState('repl');
  
  // 命令式路由
  const routes = {
    '/doctor': DoctorScreen,
    '/repl': REPLScreen,
    '/resume': ResumeConversationScreen
  };
  
  return (
    <Box>
      {routes[currentScreen]}
    </Box>
  );
}

// 命令触发路由
function handleSlashCommand(command: string) {
  if (command === '/doctor') {
    setScreen('doctor');
  } else if (command === '/resume') {
    setScreen('resume');
  }
}

3.3 Python 页面实现参考

python
# python-dispatcher/screens/screens.py
from textual.screen import Screen
from textual.widgets import Static, Button

class DoctorScreen(Screen):
    """诊断页面 - 对应 Doctor.tsx"""
    
    def compose(self):
        return Vertical(
            Static("环境诊断..."),
            Button("开始检查", id="start-check")
        )
    
    def on_button_pressed(self, event: Button.Pressed):
        if event.button.id == "start-check":
            # 执行环境检查
            results = self.run_diagnosis()
            self.query_one(Static).update(results)

class REPLScreen(Screen):
    """REPL 主界面 - 对应 REPL.tsx ⭐⭐⭐⭐⭐"""
    
    def compose(self):
        return Vertical(
            Static(id="history", classes="history"),
            Input(placeholder="输入任务...", id="input"),
            Static(id="status", classes="status")
        )
    
    def on_input_submitted(self, event: Input.Submitted):
        task = event.value
        
        # 显示用户输入
        self.query_one("#history").mount(
            Static(f"[用户] {task}")
        )
        
        # 调用调度器
        result = self.app.coordinator.dispatch_task(task)
        
        # 显示响应
        self.query_one("#history").mount(
            Static(f"[Claude] {result.output}")
        )

class ResumeConversationScreen(Screen):
    """恢复会话页面 - 对应 ResumeConversation.tsx"""
    
    def compose(self):
        return Vertical(
            Static("选择要恢复的会话..."),
            # 会话列表
            ListView(id="session-list")
        )
    
    def on_list_view_selected(self, event: ListView.Selected):
        session_id = event.item.id
        
        # 加载会话
        session = load_session(session_id)
        
        # 切换到 REPL 页面
        self.app.push_screen("repl", session=session)

3.4 CLI Handlers 架构

19 个 CLI 处理器:


cli/handlers/
├── agents.ts              # 代理管理处理器
├── auth.ts                # 认证处理器
├── autoMode.ts            # 自动模式处理器
├── mcp.tsx                # MCP 处理器
├── plugins.ts             # 插件处理器
└── util.tsx               # 工具处理器

Python 处理器对应:

python
# python-dispatcher/handlers/handlers.py
class CLIHandlers:
    """CLI 处理器集合"""
    
    def handle_agents(self, args):
        """代理管理 - 对应 handlers/agents.ts"""
        if args.action == "list":
            return subagents(action="list")
        elif args.action == "kill":
            return subagents(action="kill", target=args.session_id)
    
    def handle_auth(self, args):
        """认证管理 - 对应 handlers/auth.ts"""
        # 实际应用中应该调用 OAuth 流程
        pass
    
    def handle_auto_mode(self, args):
        """自动模式 - 对应 handlers/autoMode.ts"""
        # 切换自动模式
        self.coordinator.set_auto_mode(args.enabled)

四、状态管理分析 ⭐⭐⭐⭐

4.1 AppState 架构

6 个状态模块:


state/
├── AppState.tsx           # 应用状态定义 ⭐⭐⭐⭐⭐
├── AppStateStore.ts       # 状态存储
├── onChangeAppState.ts    # 状态变更监听
├── selectors.ts           # 状态选择器
├── store.ts               # Redux-like Store
└── teammateViewHelpers.ts # 团队视图辅助

4.2 AppState 定义

typescript
// state/AppState.tsx 伪代码（推断）
interface AppState {
  // 会话状态
  session: {
    id: string;
    messages: Message[];
    inputTokens: number;
    outputTokens: number;
  };
  
  // 代理状态
  agents: {
    active: AgentInfo[];
    completed: AgentResult[];
  };
  
  // UI 状态
  ui: {
    currentScreen: 'repl' | 'doctor' | 'resume';
    isLoading: boolean;
    notifications: Notification[];
  };
  
  // 权限状态
  permissions: {
    context: PermissionContext;
    pendingRequests: PermissionRequest[];
  };
  
  // 资源状态
  resources: {
    tokenUsage: number;
    cost: number;
    rateLimit: RateLimitInfo;
  };
}

4.3 Python 状态管理参考

python
# python-dispatcher/state/state.py
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime

@dataclass
class AppState:
    """应用状态 - 对应 AppState.tsx"""
    
    # 会话状态
    session_id: str
    messages: List[dict] = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    
    # 代理状态
    active_agents: List[dict] = field(default_factory=list)
    completed_agents: List[dict] = field(default_factory=list)
    
    # UI 状态
    current_screen: str = "repl"
    is_loading: bool = False
    notifications: List[dict] = field(default_factory=list)
    
    # 权限状态
    permission_context: Optional['PermissionContext'] = None
    pending_requests: List[dict] = field(default_factory=list)
    
    # 资源状态
    token_usage: int = 0
    cost: float = 0.0
    rate_limit: Optional[dict] = None
    
    def to_json(self) -> str:
        """序列化为 JSON"""
        return json.dumps(self.__dict__, default=str)
    
    @classmethod
    def from_json(cls, data: str) -> 'AppState':
        """从 JSON 反序列化"""
        parsed = json.loads(data)
        return cls(**parsed)


class AppStateStore:
    """状态存储 - 对应 AppStateStore.ts"""
    
    def __init__(self):
        self.state = AppState(session_id="default")
        self.listeners = []
    
    def get_state(self) -> AppState:
        """获取当前状态"""
        return self.state
    
    def update_state(self, updates: dict):
        """更新状态"""
        for key, value in updates.items():
            if hasattr(self.state, key):
                setattr(self.state, key, value)
        
        # 触发监听器
        self.notify_listeners(updates)
    
    def subscribe(self, listener):
        """订阅状态变更"""
        self.listeners.append(listener)
    
    def notify_listeners(self, changes: dict):
        """通知所有监听器"""
        for listener in self.listeners:
            listener(changes)

五、Python 调度器改造完整架构 ⭐⭐⭐⭐⭐

5.1 目录结构设计


python-dispatcher/
├── main.py                     # CLI 入口（对应 entrypoints/cli.tsx）
│
├── screens/                    # 页面组件（对应 screens/）
│   ├── __init__.py
│   ├── repl_screen.py         # REPL 主界面 ⭐
│   ├── doctor_screen.py       # 诊断页面
│   └── resume_screen.py       # 恢复会话
│
├── components/                 # UI 组件（对应 components/）
│   ├── __init__.py
│   ├── message_display.py     # 消息显示
│   ├── agent_status.py        # 代理状态
│   ├── input_box.py           # 输入框
│   └── notification.py        # 通知组件
│
├── state/                      # 状态管理（对应 state/）
│   ├── __init__.py
│   ├── app_state.py           # AppState 定义
│   ├── store.py               # 状态存储
│   └── selectors.py           # 状态选择器
│
├── handlers/                   # CLI 处理器（对应 cli/handlers/）
│   ├── __init__.py
│   ├── agents_handler.py      # 代理管理
│   ├── auth_handler.py        # 认证处理
│   └── auto_mode_handler.py   # 自动模式
│
├── coordinator/                # Coordinator 模式（Phase 1）
│   ├── __init__.py
│   ├── coordinator.py         # 主协调器
│   ├── agent_manager.py       # 代理管理
│   └── result_aggregator.py   # 结果聚合
│
├── memory/                     # 记忆系统（Phase 3）
│   ├── __init__.py
│   ├── agent_memory.py        # 代理记忆
│   └── memory_snapshot.py     # 记忆快照
│
├── permissions/                # 权限系统（Phase 1）
│   ├── __init__.py
│   ├── context.py             # PermissionContext
│   └── handlers/
│       ├── coordinator_handler.py
│       └── interactive_handler.py
│
├── runtime/                    # 运行时（对应 PortRuntime）
│   ├── __init__.py
│   ├── dispatcher.py          # 主调度器
│   └── tool_pool.py           # 工具池
│
└── config/                     # 配置文件
    ├── __init__.py
    └── settings.py            # 全局设置

5.2 Main 入口实现

python
# python-dispatcher/main.py
import asyncio
from textual.app import App
from screens.repl_screen import REPLScreen
from screens.doctor_screen import DoctorScreen
from screens.resume_screen import ResumeScreen
from coordinator.coordinator import Coordinator
from state.store import AppStateStore

class DispatcherCLI(App):
    """调度器 CLI - 对应 entrypoints/cli.tsx ⭐⭐⭐⭐⭐"""
    
    SCREENS = {
        "repl": REPLScreen,
        "doctor": DoctorScreen,
        "resume": ResumeScreen
    }
    
    CSS = """
    .main-container {
        layout: vertical;
        background: $surface;
    }
    """
    
    def __init__(self):
        super().__init__()
        
        # 初始化核心组件
        self.coordinator = Coordinator(max_agents=5)
        self.store = AppStateStore()
        self.is_active = True
    
    async def on_mount(self):
        """启动应用 - 对应 while loop 入口"""
        
        # 1. 初始化运行时
        await self.initialize_runtime()
        
        # 2. 启动主循环
        await self.run_main_loop()
    
    async def initialize_runtime(self):
        """初始化运行时"""
        
        # 加载配置
        self.config = self.load_config()
        
        # 初始化权限
        self.permissions = PermissionContext(
            allowed_tools=self.config.allowed_tools,
            denied_tools=self.config.denied_tools
        )
        
        # 注册工具
        self.tool_pool = self.assemble_tool_pool()
    
    async def run_main_loop(self):
        """主循环 - while loop 模式 ⭐⭐⭐⭐⭐"""
        
        while self.is_active:
            # a. 等待用户输入（UI 自动处理）
            # b. 处理任务（事件驱动）
            
            # 检查后台任务
            await self.check_background_tasks()
            
            # 更新状态
            await self.update_ui_state()
            
            # 等待下一个事件
            await asyncio.sleep(0.1)
    
    async def check_background_tasks(self):
        """检查后台代理任务"""
        
        # 获取活跃代理
        active = subagents(action="list", recent_minutes=5)
        
        # 更新状态
        self.store.update_state({
            "active_agents": active["running"]
        })
        
        # 检查完成的代理
        for agent in active["completed"]:
            # 获取结果
            history = sessions_history(agent["session_id"])
            
            # 聚合结果
            self.coordinator.collect_result(agent["session_id"], history)
    
    async def update_ui_state(self):
        """更新 UI 状态"""
        
        # 获取当前状态
        state = self.store.get_state()
        
        # 更新 UI（如果需要）
        if self.current_screen == "repl":
            screen = self.query_one(REPLScreen)
            screen.update_state(state)
    
    def switch_screen(self, screen_name: str):
        """切换页面 - 对应路由逻辑"""
        self.push_screen(screen_name)
    
    async def process_task(self, task: str):
        """处理任务"""
        
        # 1. 更新加载状态
        self.store.update_state({"is_loading": True})
        
        # 2. 分发任务
        result = await self.coordinator.dispatch_task(task)
        
        # 3. 显示结果
        if self.current_screen == "repl":
            screen = self.query_one(REPLScreen)
            screen.display_result(result)
        
        # 4. 更新状态
        self.store.update_state({
            "is_loading": False,
            "messages": self.store.state.messages + [result]
        })
    
    async def on_exit(self):
        """退出应用"""
        self.is_active = False
        
        # 清理资源
        await self.coordinator.shutdown()
        
        # 保存会话
        self.save_session()


def main():
    """主入口 - 对应 main() 函数"""
    
    # 1. 解析命令行参数
    import argparse
    parser = argparse.ArgumentParser(description="Python 调度器")
    parser.add_argument("--screen", default="repl", help="初始页面")
    parser.add_argument("--session", help="恢复会话 ID")
    args = parser.parse_args()
    
    # 2. 启动应用
    app = DispatcherCLI()
    
    # 3. 设置初始页面
    app.switch_screen(args.screen)
    
    # 4. 运行（对应 render(<App />) + while loop）
    app.run()


if __name__ == "__main__":
    main()

5.3 关键设计对比

Claude Code	Python 调度器	实现工具
React + Ink	Textual	`textual.app.App`
entrypoints/cli.tsx	main.py	`main()` + `DispatcherCLI`
screens/*.tsx	screens/*.py	`Screen` 类
components/*.tsx	components/*.py	`Widget` 类
state/AppState.tsx	state/app_state.py	`@dataclass AppState`
PortRuntime	DispatcherCLI	`run_main_loop()`
while loop	asyncio while	`while self.is_active`

六、可复用设计清单 ⭐⭐⭐⭐⭐

6.1 直接可复用

设计点	Claude Code	Python 实现	复用难度
页面路由	App.tsx 路由逻辑	Textual SCREENS	⭐ 极简单
While Loop	cli.tsx 主循环	asyncio while loop	⭐ 极简单
AppState	AppState.tsx 定义	@dataclass 类	⭐ 极简单
状态订阅	onChangeAppState.ts	Store.subscribe()	⭐⭐ 简单

6.2 需要适配实现

设计点	Claude Code	Python 实现	复用难度
Ink 组件	389 React 组件	Textual Widget 组件	⭐⭐⭐ 中等
CLI Handlers	19 处理器	Python 处理器类	⭐⭐ 简单
状态选择器	selectors.ts	Python 选择器函数	⭐⭐ 简单

6.3 架构级参考

设计点	Claude Code	Python 调度器	复用难度
组件化架构	389 组件模块化	Textual 组件系统设计	⭐⭐⭐⭐ 高
事件驱动	Ink 事件系统	Textual 事件系统	⭐⭐⭐ 中等
异步渲染	React 异步渲染	Textual async support	⭐⭐⭐ 中等

七、下一步建议

7.1 立即可实施

Textual 基础框架 - 安装并创建基础 App
REPL 页面 - 实现基础 REPLScreen
While Loop - 实现 asyncio 主循环
AppState - 定义基础状态类

7.2 需要深度开发

组件库 - 对应 389 组件的 Python 实现
CLI Handlers - 完整的处理器系统
状态管理 - Store + Selectors 完整实现

7.3 架构级重构

目录结构重组 - 参考 5.1 设计
Textual 组件系统 - 设计可复用组件库
事件系统集成 - 与 Coordinator 权限钩子联动

报告状态: Phase 2 完成
下一步: Phase 3 - 记忆系统深挖
生成时间: 2026-04-01 10:30

Phase 3: 记忆系统深挖分析报告

生成时间: 2026-04-01
源码状态: TypeScript 源码已移除，基于架构快照分析
关键子系统: Memdir (8 模块)、Services (130 模块)、AgentMemory (2 模块)

一、KAIROS 架构分析 ⭐⭐⭐⭐⭐

1.1 记忆系统层级

Claude Code 记忆架构: 三层记忆体系


┌─────────────────────────────────────────────┐
│ 第一层: SessionMemory (会话记忆)           │
│  - 当前会话上下文                           │
│  - 实时对话历史                             │
│  - Token 计数                               │
│  - 临时状态                                 │
│  位置: services/SessionMemory/             │
│  生命周期: 会话内                           │
└─────────────────────────────────────────────┘
            ↓ 提取关键信息 (extractMemories)
┌─────────────────────────────────────────────┐
│ 第二层: AgentMemory (代理记忆)             │
│  - 任务上下文                               │
│  - 发现的知识                               │
│  - 中间结果                                 │
│  - 快照持久化                               │
│  位置: tools/AgentTool/agentMemory.ts       │
│  生命周期: 跨会话（可恢复）                 │
└─────────────────────────────────────────────┘
            ↓ 夜间整理 (autoDream)
┌─────────────────────────────────────────────┐
│ 第三层: PersistentMemory (持久记忆)        │
│  - 用户偏好                                 │
│  - 项目背景                                 │
│  - 工作习惯                                 │
│  - 长期知识                                 │
│  位置: memdir/                              │
│  生命周期: 永久存储                         │
└─────────────────────────────────────────────┘

1.2 Memdir 子系统分析

8 个核心模块:


memdir/
├── memdir.ts                 # 记忆目录管理主入口 ⭐⭐⭐⭐⭐
├── findRelevantMemories.ts   # 语义检索相关记忆 ⭐⭐⭐⭐
├── memoryScan.ts             # 记忆扫描与索引
├── memoryAge.ts              # 记忆时效性管理
├── memoryTypes.ts            # 记忆类型定义 ⭐⭐⭐⭐⭐
├── paths.ts                  # 记忆存储路径
├── teamMemPaths.ts           # 团队记忆路径
└── teamMemPrompts.ts         # 团队记忆提示词

1.3 MemoryTypes 定义 ⭐⭐⭐⭐⭐

typescript
// memdir/memoryTypes.ts 伪代码（推断）
interface MemoryType {
  // 记忆类型枚举
  type: 'project' | 'user' | 'session' | 'knowledge' | 'habit';
  
  // 记忆内容
  content: {
    key: string;
    value: any;
    metadata: {
      created_at: number;
      last_accessed: number;
      access_count: number;
      importance: number;  // 0-1 重要性评分
      tags: string[];
    };
  };
  
  // 记忆来源
  source: {
    session_id: string;
    agent_id?: string;
    extraction_method: 'manual' | 'auto' | 'dream';
  };
  
  // 记忆关联
  associations: {
    related_memories: string[];  // 关联记忆 ID
    contexts: string[];          // 适用上下文
  };
}

// 记忆类型枚举
enum MemoryCategory {
  PROJECT = 'project',     // 项目背景（代码库结构、技术栈）
  USER = 'user',           // 用户偏好（编码风格、常用工具）
  SESSION = 'session',     // 会话上下文（当前任务状态）
  KNOWLEDGE = 'knowledge', // 知识沉淀（最佳实践、解决方案）
  HABIT = 'habit'          // 工作习惯（常用命令、快捷键）
}

1.4 Memdir.ts - 记忆目录管理

typescript
// memdir/memdir.ts 伪代码（推断）
class Memdir {
  // 记忆存储目录
  private memoryDir: string;
  private projectMemDir: string;
  private userMemDir: string;
  
  // 记忆索引
  private memoryIndex: Map<string, MemoryEntry>;
  
  // 记忆检索
  async findRelevantMemories(query: string, limit: number = 10) {
    // 1. 语义检索
    const embeddings = await this.embedQuery(query);
    
    // 2. 向量检索
    const results = await this.vectorSearch(embeddings, limit);
    
    // 3. 时效性过滤
    const freshResults = this.filterByAge(results, maxAgeDays=30);
    
    // 4. 重要性排序
    const rankedResults = this.rankByImportance(freshResults);
    
    return rankedResults;
  }
  
  // 记忆存储
  async storeMemory(memory: MemoryType) {
    // 1. 选择存储路径
    const path = this.resolvePath(memory.type);
    
    // 2. 序列化记忆
    const serialized = this.serializeMemory(memory);
    
    // 3. 写入文件
    await this.writeMemoryFile(path, memory.key, serialized);
    
    // 4. 更新索引
    this.updateIndex(memory);
  }
  
  // 记忆老化
  async ageMemories() {
    // 1. 扫描所有记忆
    const allMemories = await this.scanAllMemories();
    
    // 2. 计算时效性
    for (const memory of allMemories) {
      const ageScore = this.calculateAgeScore(memory);
      
      // 3. 标记过期记忆
      if (ageScore < 0.3) {
        await this.markForDeletion(memory);
      }
    }
  }
}

1.5 FindRelevantMemories - 语义检索 ⭐⭐⭐⭐

typescript
// memdir/findRelevantMemories.ts 伪代码
async function findRelevantMemories(
  query: string,
  context: SessionContext,
  options: SearchOptions = {}
): Promise<MemoryMatch[]> {
  
  // 1. 查询嵌入
  const queryEmbedding = await embedText(query);
  
  // 2. 上下文嵌入（增强检索）
  const contextEmbedding = await embedContext(context);
  
  // 3. 组合向量
  const combinedVector = combineVectors(queryEmbedding, contextEmbedding);
  
  // 4. 向量检索
  const candidates = await vectorDB.search(combinedVector, {
    limit: options.limit || 10,
    threshold: options.threshold || 0.7
  });
  
  // 5. 过滤与排序
  const filtered = candidates
    .filter(c => filterByAge(c, options.maxAgeDays))
    .filter(c => filterByContext(c, context))
    .sort((a, b) => b.importance - a.importance);
  
  // 6. 返回结果
  return filtered.map(c => ({
    memory: c.memory,
    score: c.score,
    relevance: calculateRelevance(c, query)
  }));
}

二、夜间做梦机制分析 ⭐⭐⭐⭐⭐

2.1 AutoDream 服务架构

services/autoDream/ 子系统

设计理念: 模拟人类睡眠时的记忆整理过程


日间交互 → 碎片记忆积累
    ↓
午夜触发 autoDream 服务
    ↓
┌─────────────────────────────┐
│ 1. 提取碎片记忆             │
│    extractMemories          │
├─────────────────────────────┤
│ 2. 分类与整理               │
│    classifyMemories         │
├─────────────────────────────┤
│ 3. 清除无用突触             │
│    pruneUnused              │
├─────────────────────────────┤
│ 4. 巩固重要信息             │
│    consolidate              │
├─────────────────────────────┤
│ 5. 压缩上下文窗口           │
│    compressContext          │
└─────────────────────────────┘
    ↓
长期记忆沉淀 → memdir 存储

2.2 AutoDream 核心流程

typescript
// services/autoDream/autoDream.ts 伪代码（推断）
class AutoDreamService {
  // 做梦触发条件
  private dreamTrigger: {
    time: "02:00",  // 午夜 2 点
    minMemories: 100,  // 最少 100 条碎片记忆
    maxAge: 7 * 24 * 3600  // 7 天内的记忆
  };
  
  // 启动做梦
  async startDream() {
    // 1. 检查触发条件
    if (!this.shouldDream()) {
      return;
    }
    
    // 2. 收集碎片记忆
    const fragments = await this.collectMemoryFragments();
    
    // 3. 提取关键信息
    const extracted = await this.extractMemories(fragments);
    
    // 4. 分类整理
    const classified = await this.classifyMemories(extracted);
    
    // 5. 清除无用
    const pruned = await this.pruneUnused(classified);
    
    // 6. 巩固重要
    const consolidated = await this.consolidate(pruned);
    
    // 7. 压缩上下文
    await this.compressContext(consolidated);
    
    // 8. 持久化
    await this.persistToMemdir(consolidated);
  }
  
  // 收集碎片记忆
  private async collectMemoryFragments(): Promise<MemoryFragment[]> {
    // 1. 查询最近会话
    const recentSessions = await queryRecentSessions({
      since: this.dreamTrigger.maxAge
    });
    
    // 2. 提取会话记忆
    const fragments = [];
    for (const session of recentSessions) {
      const sessionMemories = await extractSessionMemories(session);
      fragments.push(...sessionMemories);
    }
    
    return fragments;
  }
  
  // 提取关键信息
  private async extractMemories(fragments: MemoryFragment[]): Promise<ExtractedMemory[]> {
    const extracted = [];
    
    for (const fragment of fragments) {
      // 1. LLM 分析提取
      const analysis = await claudeAPI.analyze({
        prompt: "提取这段对话中的关键信息、用户偏好、项目背景",
        input: fragment.content
      });
      
      // 2. 结构化记忆
      const memory = {
        type: determineMemoryType(analysis),
        content: analysis.keyInfo,
        importance: analysis.importanceScore,
        source: {
          session_id: fragment.session_id,
          extraction_method: 'dream'
        }
      };
      
      extracted.push(memory);
    }
    
    return extracted;
  }
  
  // 分类整理
  private async classifyMemories(memories: ExtractedMemory[]): Promise<ClassifiedMemory[]> {
    return memories.map(memory => ({
      ...memory,
      category: classifyMemoryCategory(memory.content),
      tags: extractTags(memory.content),
      associations: findAssociations(memory, memories)
    }));
  }
  
  // 清除无用突触
  private async pruneUnused(memories: ClassifiedMemory[]): Promise<ClassifiedMemory[]> {
    // 1. 计算重要性衰减
    const decayedMemories = memories.map(memory => ({
      ...memory,
      importance: decayImportance(memory.importance, memory.last_accessed)
    }));
    
    // 2. 过滤低重要性
    return decayedMemories.filter(memory => memory.importance > 0.3);
  }
  
  // 巩固重要信息
  private async consolidate(memories: ClassifiedMemory[]): Promise<ConsolidatedMemory[]> {
    // 1. 合并相似记忆
    const merged = mergeSimilarMemories(memories);
    
    // 2. 强化高频记忆
    const reinforced = merged.map(memory => ({
      ...memory,
      importance: reinforceImportance(memory.importance, memory.access_count)
    }));
    
    // 3. 创建记忆关联网络
    const networked = createMemoryNetwork(reinforced);
    
    return networked;
  }
  
  // 压缩上下文窗口
  private async compressContext(memories: ConsolidatedMemory[]): Promise<void> {
    // 1. 生成记忆摘要
    const summary = await generateMemorySummary(memories);
    
    // 2. 压缩存储
    await compressAndStore(summary);
  }
}

2.3 ExtractMemories 服务 ⭐⭐⭐⭐

typescript
// services/extractMemories/extractMemories.ts 伪代码
interface ExtractionResult {
  memories: ExtractedMemory[];
  patterns: UserPattern[];
  insights: Insight[];
}

async function extractMemories(
  sessionHistory: SessionHistory,
  options: ExtractionOptions = {}
): Promise<ExtractionResult> {
  
  // 1. 分析对话历史
  const analysis = await analyzeSession(sessionHistory);
  
  // 2. 提取关键信息
  const keyInfo = await extractKeyInfo(analysis);
  
  // 3. 识别用户模式
  const patterns = await identifyPatterns(sessionHistory);
  
  // 4. 生成洞察
  const insights = await generateInsights(analysis, patterns);
  
  // 5. 结构化记忆
  const memories = await structureMemories(keyInfo, patterns);
  
  return {
    memories,
    patterns,
    insights
  };
}

// 提取关键信息
async function extractKeyInfo(analysis: SessionAnalysis): Promise<KeyInfo[]> {
  // 使用 LLM 提取
  const prompt = `
    分析以下对话内容，提取：
    1. 项目背景信息（技术栈、架构、目录结构）
    2. 用户偏好（编码风格、常用工具、命名习惯）
    3. 知识点（最佳实践、解决方案、踩坑经验）
    4. 当前任务状态（未完成事项、待办任务）
    
    对话内容：
    ${analysis.conversation}
  `;
  
  const response = await claudeAPI.sendMessage(prompt);
  
  return parseKeyInfo(response);
}

// 识别用户模式
async function identifyPatterns(history: SessionHistory): Promise<UserPattern[]> {
  // 1. 分析命令频率
  const commandFrequency = analyzeCommandFrequency(history);
  
  // 2. 分析工具使用频率
  const toolFrequency = analyzeToolFrequency(history);
  
  // 3. 分析编码风格
  const codingStyle = analyzeCodingStyle(history);
  
  // 4. 分析工作时间分布
  const timeDistribution = analyzeTimeDistribution(history);
  
  return [
    {
      type: 'command_pattern',
      data: commandFrequency,
      confidence: calculateConfidence(commandFrequency)
    },
    {
      type: 'tool_pattern',
      data: toolFrequency,
      confidence: calculateConfidence(toolFrequency)
    },
    {
      type: 'coding_style',
      data: codingStyle,
      confidence: calculateConfidence(codingStyle)
    },
    {
      type: 'time_pattern',
      data: timeDistribution,
      confidence: calculateConfidence(timeDistribution)
    }
  ];
}

三、AgentMemory 系统分析 ⭐⭐⭐⭐

3.1 AgentMemory 模块

tools/AgentTool/agentMemory.ts

typescript
// agentMemory.ts 伪代码（推断）
class AgentMemory {
  private sessionId: string;
  private memoryStore: Map<string, MemoryEntry>;
  private importanceThreshold: number = 0.5;
  
  constructor(sessionId: string) {
    this.sessionId = sessionId;
    this.memoryStore = new Map();
  }
  
  // 存储记忆
  store(key: string, value: any, metadata?: MemoryMetadata) {
    const entry: MemoryEntry = {
      key,
      value,
      metadata: {
        created_at: Date.now(),
        last_accessed: Date.now(),
        access_count: 0,
        importance: metadata?.importance || 0.5,
        tags: metadata?.tags || []
      },
      source: {
        session_id: this.sessionId,
        agent_id: metadata?.agent_id
      }
    };
    
    this.memoryStore.set(key, entry);
  }
  
  // 检索记忆
  retrieve(key: string): MemoryEntry | undefined {
    const entry = this.memoryStore.get(key);
    
    if (entry) {
      // 更新访问记录
      entry.metadata.last_accessed = Date.now();
      entry.metadata.access_count++;
    }
    
    return entry;
  }
  
  // 搜索记忆
  search(query: string): MemoryEntry[] {
    const results = [];
    
    for (const [key, entry] of this.memoryStore.entries()) {
      if (matchesQuery(entry, query)) {
        results.push(entry);
      }
    }
    
    // 按重要性排序
    return results.sort((a, b) => b.metadata.importance - a.metadata.importance);
  }
  
  // 创建快照
  snapshot(): AgentMemorySnapshot {
    return {
      sessionId: this.sessionId,
      timestamp: Date.now(),
      entries: Array.from(this.memoryStore.entries()),
      summary: this.generateSummary()
    };
  }
  
  // 从快照恢复
  restore(snapshot: AgentMemorySnapshot) {
    this.sessionId = snapshot.sessionId;
    this.memoryStore = new Map(snapshot.entries);
    
    // 更新恢复时间
    for (const entry of this.memoryStore.values()) {
      entry.metadata.last_accessed = Date.now();
    }
  }
  
  // 生成摘要
  private generateSummary(): string {
    const importantEntries = Array.from(this.memoryStore.values())
      .filter(e => e.metadata.importance > this.importanceThreshold);
    
    return summarizeEntries(importantEntries);
  }
}

3.2 AgentMemorySnapshot 模块 ⭐⭐⭐⭐

tools/AgentTool/agentMemorySnapshot.ts

typescript
// agentMemorySnapshot.ts 伪代码（推断）
interface AgentMemorySnapshot {
  sessionId: string;
  timestamp: number;
  entries: [string, MemoryEntry][];
  summary: string;
  metadata: {
    total_entries: number;
    total_importance: number;
    compression_ratio?: number;
  };
}

class AgentMemorySnapshotManager {
  private snapshotDir: string;
  
  // 保存快照
  async saveSnapshot(snapshot: AgentMemorySnapshot): Promise<string> {
    // 1. 序列化
    const serialized = JSON.stringify(snapshot);
    
    // 2. 压缩（可选）
    const compressed = await compress(serialized);
    
    // 3. 写入文件
    const filename = `${snapshot.sessionId}_${snapshot.timestamp}.json`;
    const filepath = path.join(this.snapshotDir, filename);
    
    await fs.writeFile(filepath, compressed || serialized);
    
    return filepath;
  }
  
  // 加载快照
  async loadSnapshot(filepath: string): Promise<AgentMemorySnapshot> {
    // 1. 读取文件
    const data = await fs.readFile(filepath);
    
    // 2. 解压（如果需要）
    const decompressed = await decompress(data);
    
    // 3. 解析 JSON
    return JSON.parse(decompressed || data.toString());
  }
  
  // 列出快照
  async listSnapshots(sessionId?: string): Promise<SnapshotInfo[]> {
    const files = await fs.readdir(this.snapshotDir);
    
    const snapshots = files
      .filter(f => f.endsWith('.json'))
      .filter(f => sessionId ? f.startsWith(sessionId) : true)
      .map(f => parseSnapshotFilename(f));
    
    return snapshots.sort((a, b) => b.timestamp - a.timestamp);
  }
  
  // 删除过期快照
  async pruneOldSnapshots(maxAge: number = 7 * 24 * 3600) {
    const now = Date.now();
    const snapshots = await this.listSnapshots();
    
    for (const snapshot of snapshots) {
      if (now - snapshot.timestamp > maxAge) {
        await fs.unlink(path.join(this.snapshotDir, snapshot.filename));
      }
    }
  }
}

四、Python 记忆系统实现参考 ⭐⭐⭐⭐⭐

4.1 记忆层级架构

python
# python-dispatcher/memory/memory_system.py
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime, timedelta
import json
import os

@dataclass
class MemoryEntry:
    """记忆条目 - 对应 MemoryEntry"""
    
    key: str
    value: Any
    created_at: datetime = field(default_factory=datetime.now)
    last_accessed: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    importance: float = 0.5
    tags: List[str] = field(default_factory=list)
    source_session_id: Optional[str] = None
    source_agent_id: Optional[str] = None


@dataclass
class MemorySnapshot:
    """记忆快照 - 对应 AgentMemorySnapshot"""
    
    session_id: str
    timestamp: datetime
    entries: Dict[str, MemoryEntry]
    summary: str
    
    def to_json(self) -> str:
        """序列化为 JSON"""
        return json.dumps({
            "session_id": self.session_id,
            "timestamp": self.timestamp.isoformat(),
            "entries": {
                k: {
                    "key": v.key,
                    "value": v.value,
                    "created_at": v.created_at.isoformat(),
                    "last_accessed": v.last_accessed.isoformat(),
                    "access_count": v.access_count,
                    "importance": v.importance,
                    "tags": v.tags,
                    "source_session_id": v.source_session_id,
                    "source_agent_id": v.source_agent_id
                }
                for k, v in self.entries.items()
            },
            "summary": self.summary
        }, ensure_ascii=False)
    
    @classmethod
    def from_json(cls, data: str) -> 'MemorySnapshot':
        """从 JSON 反序列化"""
        parsed = json.loads(data)
        
        entries = {}
        for k, v in parsed["entries"].items():
            entries[k] = MemoryEntry(
                key=v["key"],
                value=v["value"],
                created_at=datetime.fromisoformat(v["created_at"]),
                last_accessed=datetime.fromisoformat(v["last_accessed"]),
                access_count=v["access_count"],
                importance=v["importance"],
                tags=v["tags"],
                source_session_id=v.get("source_session_id"),
                source_agent_id=v.get("source_agent_id")
            )
        
        return cls(
            session_id=parsed["session_id"],
            timestamp=datetime.fromisoformat(parsed["timestamp"]),
            entries=entries,
            summary=parsed["summary"]
        )


class AgentMemory:
    """代理记忆系统 - 对应 AgentMemory.ts ⭐⭐⭐⭐⭐"""
    
    def __init__(self, session_id: str):
        self.session_id = session_id
        self.entries: Dict[str, MemoryEntry] = {}
        self.importance_threshold = 0.5
    
    def store(self, key: str, value: Any, importance: float = 0.5, tags: List[str] = None, agent_id: str = None):
        """存储记忆"""
        
        entry = MemoryEntry(
            key=key,
            value=value,
            importance=importance,
            tags=tags or [],
            source_session_id=self.session_id,
            source_agent_id=agent_id
        )
        
        self.entries[key] = entry
    
    def retrieve(self, key: str) -> Optional[MemoryEntry]:
        """检索记忆"""
        
        entry = self.entries.get(key)
        
        if entry:
            # 更新访问记录
            entry.last_accessed = datetime.now()
            entry.access_count += 1
        
        return entry
    
    def search(self, query: str) -> List[MemoryEntry]:
        """搜索记忆"""
        
        results = []
        
        for entry in self.entries.values():
            if self._matches_query(entry, query):
                results.append(entry)
        
        # 按重要性排序
        return sorted(results, key=lambda e: e.importance, reverse=True)
    
    def snapshot(self) -> MemorySnapshot:
        """创建快照"""
        
        return MemorySnapshot(
            session_id=self.session_id,
            timestamp=datetime.now(),
            entries=self.entries.copy(),
            summary=self._generate_summary()
        )
    
    def restore(self, snapshot: MemorySnapshot):
        """从快照恢复"""
        
        self.session_id = snapshot.session_id
        self.entries = snapshot.entries.copy()
        
        # 更新恢复时间
        for entry in self.entries.values():
            entry.last_accessed = datetime.now()
    
    def _matches_query(self, entry: MemoryEntry, query: str) -> bool:
        """匹配查询"""
        
        # 简单匹配：key 或 value 包含查询词
        query_lower = query.lower()
        
        if query_lower in entry.key.lower():
            return True
        
        if isinstance(entry.value, str) and query_lower in entry.value.lower():
            return True
        
        # 标签匹配
        if any(query_lower in tag.lower() for tag in entry.tags):
            return True
        
        return False
    
    def _generate_summary(self) -> str:
        """生成摘要"""
        
        important_entries = [
            e for e in self.entries.values()
            if e.importance > self.importance_threshold
        ]
        
        if not important_entries:
            return "无重要记忆"
        
        summary_lines = []
        for entry in important_entries[:10]:  # 最多 10 条
            summary_lines.append(f"- {entry.key}: {str(entry.value)[:100]}")
        
        return "\n".join(summary_lines)


class MemorySnapshotManager:
    """快照管理器 - 对应 AgentMemorySnapshotManager ⭐⭐⭐⭐"""
    
    def __init__(self, snapshot_dir: str = "~/.openclaw/memory/snapshots"):
        self.snapshot_dir = os.path.expanduser(snapshot_dir)
        os.makedirs(self.snapshot_dir, exist_ok=True)
    
    def save_snapshot(self, snapshot: MemorySnapshot) -> str:
        """保存快照"""
        
        filename = f"{snapshot.session_id}_{snapshot.timestamp.strftime('%Y%m%d_%H%M%S')}.json"
        filepath = os.path.join(self.snapshot_dir, filename)
        
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(snapshot.to_json())
        
        return filepath
    
    def load_snapshot(self, filepath: str) -> MemorySnapshot:
        """加载快照"""
        
        with open(filepath, 'r', encoding='utf-8') as f:
            data = f.read()
        
        return MemorySnapshot.from_json(data)
    
    def list_snapshots(self, session_id: str = None) -> List[Dict]:
        """列出快照"""
        
        files = os.listdir(self.snapshot_dir)
        
        snapshots = []
        for f in files:
            if not f.endswith('.json'):
                continue
            
            if session_id and not f.startswith(session_id):
                continue
            
            # 解析文件名
            parts = f.replace('.json', '').split('_')
            if len(parts) >= 3:
                snapshots.append({
                    "session_id": parts[0],
                    "timestamp": datetime.strptime(f"{parts[1]}_{parts[2]}", "%Y%m%d_%H%M%S"),
                    "filename": f
                })
        
        return sorted(snapshots, key=lambda s: s["timestamp"], reverse=True)
    
    def prune_old_snapshots(self, max_age_days: int = 7):
        """删除过期快照"""
        
        now = datetime.now()
        snapshots = self.list_snapshots()
        
        for snapshot in snapshots:
            age = (now - snapshot["timestamp"]).days
            
            if age > max_age_days:
                filepath = os.path.join(self.snapshot_dir, snapshot["filename"])
                os.unlink(filepath)


class PersistentMemory:
    """持久记忆系统 - 对应 memdir ⭐⭐⭐⭐⭐"""
    
    def __init__(self, memory_dir: str = "~/.openclaw/memory/persistent"):
        self.memory_dir = os.path.expanduser(memory_dir)
        os.makedirs(self.memory_dir, exist_ok=True)
        
        self.index: Dict[str, MemoryEntry] = {}
        self._load_index()
    
    def store_persistent(self, memory_type: str, key: str, value: Any, importance: float = 0.7):
        """存储持久记忆"""
        
        # 选择存储路径
        subdir = os.path.join(self.memory_dir, memory_type)
        os.makedirs(subdir, exist_ok=True)
        
        # 创建记忆条目
        entry = MemoryEntry(
            key=key,
            value=value,
            importance=importance,
            tags=[memory_type],
            created_at=datetime.now()
        )
        
        # 写入文件
        filepath = os.path.join(subdir, f"{key}.json")
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(json.dumps({
                "key": key,
                "value": value,
                "importance": importance,
                "created_at": entry.created_at.isoformat(),
                "tags": entry.tags
            }, ensure_ascii=False))
        
        # 更新索引
        self.index[f"{memory_type}/{key}"] = entry
    
    def find_relevant(self, query: str, limit: int = 10) -> List[MemoryEntry]:
        """查找相关记忆 - 对应 findRelevantMemories.ts"""
        
        # 简单文本搜索（实际应用中应该使用向量检索）
        results = []
        
        for entry in self.index.values():
            if self._matches_query(entry, query):
                results.append(entry)
        
        # 按重要性排序
        results.sort(key=lambda e: e.importance, reverse=True)
        
        return results[:limit]
    
    def _load_index(self):
        """加载索引"""
        
        for memory_type in os.listdir(self.memory_dir):
            subdir = os.path.join(self.memory_dir, memory_type)
            
            if not os.path.isdir(subdir):
                continue
            
            for filename in os.listdir(subdir):
                if not filename.endswith('.json'):
                    continue
                
                filepath = os.path.join(subdir, filename)
                
                try:
                    with open(filepath, 'r', encoding='utf-8') as f:
                        data = json.load(f)
                    
                    entry = MemoryEntry(
                        key=data["key"],
                        value=data["value"],
                        importance=data["importance"],
                        tags=data["tags"],
                        created_at=datetime.fromisoformat(data["created_at"])
                    )
                    
                    self.index[f"{memory_type}/{entry.key}"] = entry
                except Exception as e:
                    print(f"加载记忆失败: {filepath} - {e}")
    
    def _matches_query(self, entry: MemoryEntry, query: str) -> bool:
        """匹配查询"""
        
        query_lower = query.lower()
        
        # key 匹配
        if query_lower in entry.key.lower():
            return True
        
        # value 匹配
        if isinstance(entry.value, str) and query_lower in entry.value.lower():
            return True
        
        # tags 匹配
        if any(query_lower in tag.lower() for tag in entry.tags):
            return True
        
        return False

4.2 AutoDream 服务实现

python
# python-dispatcher/memory/auto_dream.py
import asyncio
from datetime import datetime, timedelta
from typing import List, Dict
import json

class AutoDreamService:
    """自动梦境服务 - 对应 autoDream ⭐⭐⭐⭐⭐"""
    
    def __init__(
        self,
        trigger_time: str = "02:00",
        min_memories: int = 100,
        max_age_days: int = 7
    ):
        self.trigger_time = trigger_time
        self.min_memories = min_memories
        self.max_age_days = max_age_days
        
        self.persistent_memory = PersistentMemory()
        self.snapshot_manager = MemorySnapshotManager()
    
    async def start_dream_loop(self):
        """启动做梦循环"""
        
        while True:
            # 检查是否应该做梦
            if self.should_dream():
                await self.start_dream()
            
            # 等待到下一个触发时间
            await self.wait_until_next_trigger()
    
    def should_dream(self) -> bool:
        """检查触发条件"""
        
        # 1. 检查时间
        now = datetime.now()
        trigger_hour, trigger_minute = map(int, self.trigger_time.split(':'))
        
        if now.hour != trigger_hour:
            return False
        
        # 2. 检查碎片记忆数量
        recent_sessions = self._query_recent_sessions()
        
        if len(recent_sessions) < self.min_memories:
            return False
        
        return True
    
    async def start_dream(self):
        """执行做梦流程"""
        
        print(f"[AutoDream] 开始做梦 - {datetime.now()}")
        
        # 1. 收集碎片记忆
        fragments = await self.collect_memory_fragments()
        
        print(f"[AutoDream] 收集到 {len(fragments)} 条碎片记忆")
        
        # 2. 提取关键信息
        extracted = await self.extract_memories(fragments)
        
        print(f"[AutoDream] 提取到 {len(extracted)} 条关键信息")
        
        # 3. 分类整理
        classified = await self.classify_memories(extracted)
        
        # 4. 清除无用
        pruned = await self.prune_unused(classified)
        
        print(f"[AutoDream] 清除后剩余 {len(pruned)} 条记忆")
        
        # 5. 巩固重要
        consolidated = await self.consolidate(pruned)
        
        # 6. 持久化
        await self.persist_to_memdir(consolidated)
        
        print(f"[AutoDream] 做梦完成 - 持久化 {len(consolidated)} 条记忆")
    
    async def collect_memory_fragments(self) -> List[Dict]:
        """收集碎片记忆"""
        
        # 查询最近会话
        since = datetime.now() - timedelta(days=self.max_age_days)
        
        # 实际应用中应该从 sessions_history 获取
        # 这里简化实现
        
        fragments = []
        
        # 模拟：列出最近的快照
        snapshots = self.snapshot_manager.list_snapshots()
        
        for snapshot_info in snapshots[:20]:  # 最近 20 个会话
            snapshot = self.snapshot_manager.load_snapshot(
                os.path.join(self.snapshot_manager.snapshot_dir, snapshot_info["filename"])
            )
            
            for key, entry in snapshot.entries.items():
                fragments.append({
                    "session_id": snapshot.session_id,
                    "key": key,
                    "value": entry.value,
                    "importance": entry.importance,
                    "timestamp": entry.created_at
                })
        
        return fragments
    
    async def extract_memories(self, fragments: List[Dict]) -> List[Dict]:
        """提取关键信息"""
        
        # 实际应用中应该调用 LLM 分析
        # 这里简化实现：直接过滤高重要性
        
        extracted = []
        
        for fragment in fragments:
            if fragment["importance"] > 0.5:
                extracted.append({
                    "type": self._determine_memory_type(fragment),
                    "key": fragment["key"],
                    "value": fragment["value"],
                    "importance": fragment["importance"],
                    "source_session_id": fragment["session_id"],
                    "extraction_method": "dream"
                })
        
        return extracted
    
    async def classify_memories(self, memories: List[Dict]) -> List[Dict]:
        """分类整理"""
        
        classified = []
        
        for memory in memories:
            classified.append({
                **memory,
                "category": memory["type"],
                "tags": [memory["type"]],
                "associations": []
            })
        
        return classified
    
    async def prune_unused(self, memories: List[Dict]) -> List[Dict]:
        """清除无用突触"""
        
        # 计算重要性衰减
        pruned = []
        
        for memory in memories:
            # 重要性衰减（7 天衰减 30%）
            age_days = (datetime.now() - memory.get("timestamp", datetime.now())).days
            decay_factor = 1 - (age_days / 7) * 0.3
            
            decayed_importance = memory["importance"] * decay_factor
            
            # 过滤低重要性
            if decayed_importance > 0.3:
                memory["importance"] = decayed_importance
                pruned.append(memory)
        
        return pruned
    
    async def consolidate(self, memories: List[Dict]) -> List[Dict]:
        """巩固重要信息"""
        
        # 合并相似记忆（简化实现）
        consolidated = []
        
        # 按类型分组
        by_type = {}
        for memory in memories:
            type_key = memory["type"]
            if type_key not in by_type:
                by_type[type_key] = []
            by_type[type_key].append(memory)
        
        # 每个类型保留最重要的记忆
        for type_key, type_memories in by_type.items():
            # 按重要性排序
            sorted_memories = sorted(type_memories, key=lambda m: m["importance"], reverse=True)
            
            # 保留前 10 条
            for memory in sorted_memories[:10]:
                # 强化重要性
                memory["importance"] = min(memory["importance"] * 1.2, 1.0)
                consolidated.append(memory)
        
        return consolidated
    
    async def persist_to_memdir(self, memories: List[Dict]):
        """持久化到 memdir"""
        
        for memory in memories:
            self.persistent_memory.store_persistent(
                memory_type=memory["type"],
                key=memory["key"],
                value=memory["value"],
                importance=memory["importance"]
            )
    
    async def wait_until_next_trigger(self):
        """等待到下一个触发时间"""
        
        now = datetime.now()
        trigger_hour, trigger_minute = map(int, self.trigger_time.split(':'))
        
        # 计算下一个触发时间
        next_trigger = now.replace(hour=trigger_hour, minute=trigger_minute, second=0, microsecond=0)
        
        if next_trigger <= now:
            # 如果已经过了今天的触发时间，等待明天
            next_trigger += timedelta(days=1)
        
        # 计算等待秒数
        wait_seconds = (next_trigger - now).total_seconds()
        
        print(f"[AutoDream] 下一次做梦时间: {next_trigger}，等待 {wait_seconds / 3600:.1f} 小时")
        
        await asyncio.sleep(wait_seconds)
    
    def _query_recent_sessions(self) -> List[str]:
        """查询最近会话"""
        
        # 实际应用中应该使用 sessions_list
        # 这里简化实现
        
        return []
    
    def _determine_memory_type(self, fragment: Dict) -> str:
        """确定记忆类型"""
        
        # 简单规则判断
        key = fragment["key"].lower()
        value = str(fragment["value"]).lower()
        
        if "project" in key or "project" in value:
            return "project"
        elif "user" in key or "preference" in value:
            return "user"
        elif "knowledge" in key or "best" in value:
            return "knowledge"
        elif "habit" in key or "pattern" in value:
            return "habit"
        else:
            return "session"


# 启动 AutoDream 服务示例
async def start_auto_dream_service():
    """启动 AutoDream 服务"""
    
    service = AutoDreamService()
    await service.start_dream_loop()


if __name__ == "__main__":
    asyncio.run(start_auto_dream_service())

五、可复用设计清单 ⭐⭐⭐⭐⭐

5.1 直接可复用

设计点	Claude Code	Python 实现	复用难度
MemoryEntry	memoryTypes.ts	@dataclass MemoryEntry	⭐ 极简单
AgentMemory	agentMemory.ts	AgentMemory 类	⭐⭐ 简单
MemorySnapshot	agentMemorySnapshot.ts	MemorySnapshot + Manager	⭐⭐ 简单
三层记忆架构	Session → Agent → Persistent	三层类设计	⭐⭐⭐ 中等

5.2 需要深度实现

设计点	Claude Code	Python 实现	复用难度
AutoDream 服务	autoDream.ts	AutoDreamService + asyncio	⭐⭐⭐⭐ 高
ExtractMemories	extractMemories.ts	LLM 分析 + 结构化	⭐⭐⭐⭐ 高
语义检索	findRelevantMemories.ts	向量检索（需要嵌入模型）	⭐⭐⭐⭐⭐ 极高

5.3 架构级参考

设计点	Claude Code	Python 调度器	复用难度
记忆层级设计	SessionMemory → AgentMemory → Persistent	三层架构设计	⭐⭐⭐ 中等
夜间整理机制	autoDream 定时服务	cron + asyncio 服务	⭐⭐⭐ 中等
快照恢复模式	snapshot → restore	快照持久化 + 恢复	⭐⭐ 简单

六、集成到现有 Harness 系统建议 ⭐⭐⭐⭐

6.1 立即可集成

AgentMemory 基础版本 - 在 Python 调度器中使用
MemorySnapshot - 快照持久化与恢复
PersistentMemory - 简单持久记忆存储

6.2 需要开发集成

AutoDream 定时服务 - 使用 OpenClaw cron 调度

python
# 使用 cron 工具设置定时任务
cron(action="add", job={
    "name": "auto-dream",
    "schedule": {"kind": "cron", "expr": "0 2 * * *"},  # 每天 2:00
    "payload": {"kind": "agentTurn", "message": "执行记忆整理"}
})

ExtractMemories - 集成到会话结束流程

python
# 在会话结束时提取记忆
def on_session_end(session_id: str):
    history = sessions_history(session_id)
    extracted = extract_memories(history)
    agent_memory.restore(extracted)

6.3 高级功能（需要外部支持）

向量检索 - 需要嵌入模型（如 OpenAI Embeddings）
LLM 分析提取 - 需要调用 Claude/GPT API
语义匹配 - 需要向量数据库（如 ChromaDB）

七、下一步建议

7.1 立即可实施

AgentMemory 类 - 实现基础记忆存储
MemorySnapshot - 实现快照持久化
PersistentMemory - 实现持久记忆目录

7.2 需要深度开发

AutoDream 服务 - 集成到 OpenClaw cron
ExtractMemories - 开发记忆提取逻辑
记忆索引 - 实现快速检索机制

7.3 高级优化

向量检索 - 集成嵌入模型
LLM 分析 - 集成智能提取
记忆网络 - 实现关联记忆图谱

报告状态: Phase 3 完成
下一步: Phase 4 - 工具系统深挖
生成时间: 2026-04-01 10:45