Claw-Code Agent 架构分析与工作流程（Python 实现版）

Claw-Code Agent 架构分析与工作流程（Python 实现版）

一、Agent 系统概述

Claw Code 的 Python 实现是一个元数据驱动、模拟执行、单线程顺序处理的 Agent 原型系统。与 Rust 版本的生产级多线程实现不同，Python 版本专注于：

  1. 架构验证：通过镜像 TypeScript 源码的命令和工具元数据，验证移植可行性
  2. 教学参考：提供清晰的代码结构，便于理解 Agent harness 设计模式
  3. 快速原型：使用 Python 的动态特性快速迭代和测试

核心差异：

  • ❌ 无真实子 Agent 创建：Python 版本没有实现真正的多线程子 Agent
  • ✅ 元数据镜像：完整镜像了原始 TS 代码的命令/工具清单
  • ✅ 模拟执行：通过 ExecutionRegistry 模拟命令和工具的调用
  • ✅ 路由匹配：实现了基于关键词的 prompt 路由机制

二、核心架构设计

2.1 Python 实现的 Agent 层级

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
┌─────────────────────────────────────────────┐
│         CLI / REPL (main.py)                │
│  - argparse 命令行解析                       │
│  - 20+ 子命令（summary, bootstrap, etc.）   │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│       PortRuntime (runtime.py)              │
│  - route_prompt() 路由匹配                  │
│  - bootstrap_session() 会话引导             │
│  - run_turn_loop() 转环执行                 │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│    QueryEnginePort (query_engine.py)        │
│  - submit_message() 消息提交                │
│  - stream_submit_message() 流式输出         │
│  - persist_session() 会话持久化             │
└──────────────────┬──────────────────────────┘
                   │
                   ▼
┌─────────────────────────────────────────────┐
│   ExecutionRegistry (execution_registry.py) │
│  - MirroredCommand 镜像命令                 │
│  - MirroredTool 镜像工具                    │
│  - 模拟执行（无真实 API 调用）               │
└─────────────────────────────────────────────┘

2.2 关键组件对比

组件 Python 实现 Rust 实现 状态
Agent 创建 ❌ 未实现 ✅ execute_agent() + 后台线程 Python 仅镜像元数据
子 Agent 执行 ❌ 无线程模型 ✅ spawn_agent_job() Python 为单线程
工具白名单 ❌ 无限制逻辑 ✅ allowed_tools_for_subagent() Python 未实现
会话管理 ✅ QueryEnginePort ✅ ConversationRuntime 两者都实现
路由匹配 ✅ route_prompt() ❌ 无此概念 Python 特有
元数据镜像 ✅ PORTED_COMMANDS ❌ 硬编码 Python 从 JSON 加载
执行注册表 ✅ ExecutionRegistry ✅ ToolExecutor trait 设计相似
流式输出 ✅ Generator yield ✅ SSE 流 都支持

三、Agent 工作流程详解

3.1 主流程：从用户输入到子 Agent 执行

步骤 1：用户触发 Agent 工具

用户在主 Agent 对话中发出指令，主 Agent 决定调用 Agent 工具：

1
2
3
4
5
6
7
8
9
10
{
  "tool": "Agent",
  "input": {
    "description": "审计代码分支",
    "prompt": "检查测试用例和未完成的工作",
    "subagent_type": "Explore",
    "name": "ship-audit",
    "model": "claude-sonnet-4-6"
  }
}

步骤 2：Agent 工具处理（execute_tool）

文件: rust/crates/tools/src/lib.rs:1260-1422

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
fn execute_agent(input_value: &Value) -> Result<String, String> {
    // 1. 解析输入参数
    let input: AgentInput = serde_json::from_value(input_value.clone())?;
    
    // 2. 标准化 subagent_type（支持别名映射）
    let normalized_subagent_type = normalize_subagent_type(input.subagent_type.as_deref());
    // "explorer" → "Explore", "planner" → "Plan", etc.
    
    // 3. 生成唯一 agent_id 和时间戳
    let agent_id = generate_agent_id();
    let created_at = iso8601_now();
    
    // 4. 构建系统提示（包含 subagent_type 特定指令）
    let system_prompt = build_agent_system_prompt(&normalized_subagent_type)?;
    // 添加："You are a background sub-agent of type `Explore`. 
    //        Work only on the delegated task..."
    
    // 5. 根据 subagent_type 确定允许的工具集
    let allowed_tools = allowed_tools_for_subagent(&normalized_subagent_type);
    // Explore: [read_file, glob_search, grep_search, WebFetch, ...]
    // Plan:    [read_file, TodoWrite, StructuredOutput, ...]
    // Verification: [bash, read_file, PowerShell, ...]
    
    // 6. 创建输出文件和 manifest 文件
    let output_file = format!("{}/{}.md", agent_store_dir, agent_id);
    let manifest_file = format!("{}/{}.manifest.json", agent_store_dir, agent_id);
    
    // 7. 写入初始任务描述
    std::fs::write(&output_file, format!("# Agent Task\n..."));
    
    // 8. 创建 AgentOutput manifest
    let manifest = AgentOutput {
        agent_id,
        name: agent_name,
        description: input.description,
        subagent_type: Some(normalized_subagent_type),
        model: Some(model),
        status: "running".to_string(),
        output_file: output_file.clone(),
        manifest_file: manifest_file.clone(),
        created_at: created_at.clone(),
        started_at: Some(created_at),
        completed_at: None,
        error: None,
    };
    
    // 9. 持久化 manifest
    write_agent_manifest(&manifest)?;
    
    // 10. 构建 AgentJob
    let job = AgentJob {
        manifest: manifest.clone(),
        prompt: input.prompt,
        system_prompt,
        allowed_tools,
    };
    
    // 11. 启动后台线程
    spawn_agent_job(job)?;
    
    // 12. 立即返回 manifest（不等待子 Agent 完成）
    Ok(serde_json::to_string_pretty(&manifest)?)
}

步骤 3：后台线程启动（spawn_agent_job）

文件: rust/crates/tools/src/lib.rs:1424-1449

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
fn spawn_agent_job(job: AgentJob) -> Result<(), String> {
    // 1. 创建命名线程（便于调试和监控）
    let thread_name = format!("clawd-agent-{}", job.manifest.agent_id);
    
    std::thread::Builder::new()
        .name(thread_name)
        .spawn(move || {
            // 2. 捕获 panic，防止子线程崩溃影响主进程
            let result = std::panic::catch_unwind(
                std::panic::AssertUnwindSafe(|| run_agent_job(&job))
            );
            
            // 3. 处理执行结果
            match result {
                Ok(Ok(())) => {
                    // 成功完成，状态已在 run_agent_job 中持久化
                }
                Ok(Err(error)) => {
                    // 运行时错误，更新 manifest 为 failed
                    persist_agent_terminal_state(
                        &job.manifest, "failed", None, Some(error)
                    );
                }
                Err(_) => {
                    // Panic，记录异常
                    persist_agent_terminal_state(
                        &job.manifest, "failed", None, 
                        Some("sub-agent thread panicked".to_string())
                    );
                }
            }
        })
        .map(|_| ())
        .map_err(|error| error.to_string())
}

步骤 4：子 Agent 执行（run_agent_job）

文件: rust/crates/tools/src/lib.rs:1451-1458

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
fn run_agent_job(job: &AgentJob) -> Result<(), String> {
    // 1. 构建子 Agent 的 ConversationRuntime
    let mut runtime = build_agent_runtime(job)?
        .with_max_iterations(DEFAULT_AGENT_MAX_ITERATIONS);
    
    // 2. 执行单轮对话（无用户交互）
    let summary = runtime.run_turn(job.prompt.clone(), None)?;
    
    // 3. 提取最终回复文本
    let final_text = final_assistant_text(&summary);
    
    // 4. 持久化完成状态和结果
    persist_agent_terminal_state(
        &job.manifest, "completed", Some(final_text.as_str()), None
    )
}

步骤 5：构建子 Agent 运行时（build_agent_runtime）

文件: rust/crates/tools/src/lib.rs:1460-1478

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
fn build_agent_runtime(
    job: &AgentJob,
) -> Result<ConversationRuntime<AnthropicRuntimeClient, SubagentToolExecutor>, String> {
    // 1. 确定模型（默认 claude-opus-4-6）
    let model = job.manifest.model.clone()
        .unwrap_or_else(|| DEFAULT_AGENT_MODEL.to_string());
    
    // 2. 创建 API 客户端（带工具白名单）
    let api_client = AnthropicRuntimeClient::new(
        model, 
        job.allowed_tools.clone()
    )?;
    
    // 3. 创建受限的工具执行器
    let tool_executor = SubagentToolExecutor::new(job.allowed_tools.clone());
    
    // 4. 构建 ConversationRuntime
    ConversationRuntime::new(
        Session::new(),              // 全新的空会话（隔离）
        api_client,                  // 独立的 API 客户端
        tool_executor,               // 受限工具执行器
        agent_permission_policy(),   // DangerFullAccess（子 Agent 无需确认）
        job.system_prompt.clone(),   // 包含 subagent_type 指令
    )
}

步骤 6：工具执行限制（SubagentToolExecutor）

文件: rust/crates/tools/src/lib.rs:1769-1788

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
struct SubagentToolExecutor {
    allowed_tools: BTreeSet<String>,
}

impl ToolExecutor for SubagentToolExecutor {
    fn execute(&mut self, tool_name: &str, input: &str) -> Result<String, ToolError> {
        // 1. 检查工具是否在白名单中
        if !self.allowed_tools.contains(tool_name) {
            return Err(ToolError::new(format!(
                "tool `{tool_name}` is not enabled for this sub-agent"
            )));
        }
        
        // 2. 委托给实际的工具实现
        execute_tool(tool_name, &serde_json::from_str(input)?)
            .map_err(|e| ToolError::new(e))
    }
}

3.2 子 Agent 类型与工具映射

文件: rust/crates/tools/src/lib.rs:1503-1582

Sub-agent Type 用途 允许的工具 禁止的工具
Explore 代码探索、信息收集 read_file, glob_search, grep_search, WebFetch, WebSearch, ToolSearch, Skill, StructuredOutput bash, write_file, edit_file, Agent
Plan 任务规划、待办管理 Explore 工具 + TodoWrite, SendUserMessage bash, write_file, edit_file, Agent
Verification 测试验证、CI 检查 bash, read_file, grep_search, WebFetch, WebSearch, ToolSearch, TodoWrite, StructuredOutput, SendUserMessage, PowerShell write_file, edit_file, Agent
general-purpose 通用任务（默认） 几乎所有工具（18+）除了 Agent Agent（防止递归爆炸）
claw-code-guide 文档查询 只读工具 + StructuredOutput, SendUserMessage 所有写操作、bash
statusline-setup 配置修改 bash, read_file, write_file, edit_file, glob_search, grep_search, ToolSearch Agent, WebFetch

关键规则：

  • 所有子 Agent 都不能调用 Agent 工具：防止无限递归创建
  • Explore/Plan 不能执行 bash：确保安全性
  • Verification 可以执行 bash 但不能写文件：只读验证

四、Agent 交互机制

4.1 主 Agent 与子 Agent 的通信

单向委托模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
主 Agent                          子 Agent
    │                                │
    │  1. 调用 Agent 工具             │
    │  ──────────────────────────►   │
    │     (description, prompt,      │
    │      subagent_type, model)     │
    │                                │
    │  2. 立即返回 manifest          │
    │  ◄──────────────────────────   │
    │     (agent_id, status=running) │
    │                                │
    │  3. 后台线程执行               │
    │                                │──► 执行任务
    │                                │──► 调用允许的工具
    │                                │──► 生成结果
    │                                │
    │  4. 持久化结果                 │
    │     (output_file, manifest)    │
    │                                │
    │  5. 主 Agent 读取结果          │
    │     (通过 read_file 或轮询)    │
    │                                │

特点：

  • 异步非阻塞：主 Agent 调用后立即返回，不等待子 Agent 完成
  • 文件系统通信：通过 output_file 和 manifest_file 交换数据
  • 无直接内存共享：完全隔离，通过文件系统解耦

结果查询方式

主 Agent 可以通过以下方式获取子 Agent 结果：

  1. 轮询 manifest 文件：

    1
    2
    3
    4
    5
    6
    7
    8
    # 伪代码
    while True:
        manifest = json.load(open(f"{agent_id}.manifest.json"))
        if manifest["status"] in ["completed", "failed"]:
            break
        time.sleep(1)

    result = open(manifest["output_file"]).read()

  2. 直接读取输出文件：

    1
    2
    # 子 Agent 持续追加结果到 output_file
    cat .clawd-agents/{agent_id}.md

  3. 使用内置工具查询（未来扩展）：

    1
    2
    3
    4
    {
      "tool": "AgentStatus",
      "input": {"agent_id": "abc123"}
    }

4.2 子 Agent 之间的交互

当前设计：子 Agent 之间不直接通信

每个子 Agent 都是独立、隔离的执行单元：

  • 独立的 Session（无共享对话历史）
  • 独立的 ConversationRuntime
  • 独立的线程
  • 只能通过文件系统间接通信（例如读写同一文件）

未来可能的扩展：

  • Agent 协调器：主 Agent 作为协调者，收集多个子 Agent 的结果并整合
  • 共享状态存储：通过数据库或 Redis 实现跨 Agent 状态共享
  • 消息队列：使用 MQ 实现异步消息传递

五、并行执行机制

5.1 何时创建子 Agent

主 Agent 在以下场景会创建子 Agent：

  1. 复杂任务分解：

    • 用户请求：”分析整个代码库的安全漏洞”
    • 主 Agent 创建多个 Explore 子 Agent，分别检查不同模块

  2. 长时间运行任务：

    • 用户请求：”运行完整的测试套件并修复失败用例”
    • 主 Agent 创建 Verification 子 Agent，避免阻塞主对话

  3. 并行信息收集：

    • 用户请求：”比较三个不同框架的性能”
    • 主 Agent 并行创建 3 个 Explore 子 Agent

  4. 隔离危险操作：

    • 用户请求：”删除所有临时文件”
    • 主 Agent 创建 general-purpose 子 Agent 执行，限制影响范围

5.2 何时可以并行运行

所有子 Agent 都可以并行运行，因为：

  1. 独立线程：每个子 Agent 在独立的 OS 线程中运行

    1
    2
    3
    std::thread::Builder::new()
        .name(format!("clawd-agent-{}", agent_id))
        .spawn(move || { ... })

  2. 无共享状态：

    • 每个子 Agent 有独立的 Session
    • 独立的 ConversationRuntime
    • 独立的 API 连接

  3. 文件系统原子操作：

    • 每个子 Agent 写入不同的 output_file 和 manifest_file
    • 使用 append 模式追加结果，避免竞争

  4. 工具执行隔离：

    • 每个子 Agent 有自己的 SubagentToolExecutor
    • 工具白名单在创建时确定，运行时不可变

5.3 并行示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// 主 Agent 同时创建 3 个子 Agent
let agent1 = execute_tool("Agent", json!({
    "description": "探索前端代码",
    "prompt": "分析 src/frontend 目录",
    "subagent_type": "Explore"
}));

let agent2 = execute_tool("Agent", json!({
    "description": "探索后端代码",
    "prompt": "分析 src/backend 目录",
    "subagent_type": "Explore"
}));

let agent3 = execute_tool("Agent", json!({
    "description": "探索测试代码",
    "prompt": "分析 tests 目录",
    "subagent_type": "Explore"
}));

// 三个子 Agent 在各自线程中并行执行
// 主 Agent 可以继续其他工作或等待结果

时间线：

1
2
3
4
5
6
7
8
T0: 主 Agent 创建 agent1 ──► 线程1 启动
T1: 主 Agent 创建 agent2 ──► 线程2 启动
T2: 主 Agent 创建 agent3 ──► 线程3 启动
T3-T10: 三个线程并行执行
T11: agent1 完成，写入结果
T12: agent2 完成，写入结果
T13: agent3 完成，写入结果
T14: 主 Agent 读取所有结果并整合

5.4 并行度限制

当前无显式限制，但受以下因素制约：

  1. 系统资源：

    • CPU 核心数
    • 内存限制
    • 文件描述符数量

  2. API 速率限制：

    • Anthropic API 的并发请求限制
    • Token 配额消耗速度

  3. 工具冲突：

    • 多个子 Agent 同时写同一文件可能导致竞争
    • bash 命令的资源竞争（端口、文件锁等）

建议最佳实践：

  • 同时运行的子 Agent 不超过 5-10 个
  • 对于 I/O 密集型任务（文件读取），可以更高并行度
  • 对于 CPU 密集型任务（代码分析），限制并行度

六、核心文件清单

6.1 Rust 实现（生产级）

文件 路径 行数 关键内容
tools/lib.rs rust/crates/tools/src/lib.rs 4241 Agent 工具实现、SubagentToolExecutor、allowed_tools_for_subagent、spawn_agent_job、run_agent_job
conversation.rs rust/crates/runtime/src/conversation.rs 973 ConversationRuntime、run_turn、工具执行循环、Hook 集成
session.rs rust/crates/runtime/src/session.rs ~400 Session 数据结构、消息持久化
permissions.rs rust/crates/runtime/src/permissions.rs ~230 PermissionPolicy、PermissionMode、授权逻辑

关键代码段索引

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
// Agent 工具入口
// rust/crates/tools/src/lib.rs:1260-1422
fn execute_agent(input_value: &Value) -> Result<String, String>

// 后台线程启动
// rust/crates/tools/src/lib.rs:1424-1449
fn spawn_agent_job(job: AgentJob) -> Result<(), String>

// 子 Agent 执行逻辑
// rust/crates/tools/src/lib.rs:1451-1458
fn run_agent_job(job: &AgentJob) -> Result<(), String>

// 构建子 Agent 运行时
// rust/crates/tools/src/lib.rs:1460-1478
fn build_agent_runtime(job: &AgentJob) -> Result<ConversationRuntime<...>, String>

// 工具白名单映射
// rust/crates/tools/src/lib.rs:1503-1582
fn allowed_tools_for_subagent(subagent_type: &str) -> BTreeSet<String>

// 受限工具执行器
// rust/crates/tools/src/lib.rs:1769-1788
impl ToolExecutor for SubagentToolExecutor

// 主对话循环
// rust/crates/runtime/src/conversation.rs:170-283
pub fn run_turn(&mut self, user_input: impl Into<String>, ...) -> Result<TurnSummary, RuntimeError>

6.2 Python 实现（原型参考）

文件 路径 说明
runtime.py src/runtime.py PortRuntime 类，模拟 Agent 路由和执行
query_engine.py src/query_engine.py QueryEnginePort，会话管理和转环逻辑
execution_registry.py src/execution_registry.py 命令和工具的镜像执行注册表

注意：Python 实现主要是元数据和模拟，没有真正的多线程子 Agent 执行。

6.3 配置文件

文件 说明
.clawd-agents/ 子 Agent 输出目录（gitignored）
CLAWD_AGENT_STORE 环境变量，指定 Agent 存储路径（默认 .clawd-agents/）

七、数据流图

7.1 Agent 创建数据流

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
用户输入
  │
  ▼
主 Agent (ConversationRuntime)
  │
  ├─► 决定调用 Agent 工具
  │
  ▼
execute_agent()
  │
  ├─► 解析输入 (AgentInput)
  ├─► 标准化 subagent_type
  ├─► 生成 agent_id, timestamp
  ├─► 构建 system_prompt
  ├─► 查询 allowed_tools_for_subagent()
  ├─► 创建 output_file, manifest_file
  ├─► 写入初始任务描述
  ├─► 序列化 AgentOutput manifest
  ├─► 构建 AgentJob
  │
  ▼
spawn_agent_job(AgentJob)
  │
  ├─► std::thread::Builder::new()
  ├─► 设置线程名称 "clawd-agent-{id}"
  ├─► spawn(move || { ... })
  │
  ▼
[新线程开始]
  │
  ├─► catch_unwind { run_agent_job(&job) }
  │
  ▼
run_agent_job(&job)
  │
  ├─► build_agent_runtime(job)
  │     ├─► AnthropicRuntimeClient::new(model, allowed_tools)
  │     ├─► SubagentToolExecutor::new(allowed_tools)
  │     └─► ConversationRuntime::new(...)
  │
  ├─► runtime.run_turn(prompt, None)
  │     ├─► session.messages.push(user_message)
  │     ├─► loop {
  │     │     ├─► api_client.stream(request)
  │     │     ├─► 解析 AssistantEvent
  │     │     ├─► 遇到 ToolUse → tool_executor.execute()
  │     │     │                      ├─► 检查 allowed_tools
  │     │     │                      └─► 执行工具
  │     │     ├─► 无更多工具 → break
  │     │   }
  │     └─► 返回 TurnSummary
  │
  ├─► 提取 final_text
  ├─► persist_agent_terminal_state("completed", final_text)
  │     ├─► append_agent_output(output_file, result)
  │     └─► write_agent_manifest(updated_manifest)
  │
  ▼
[线程结束]

7.2 状态持久化流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
Agent 生命周期状态机：

created ──► running ──► completed
                     └─► failed

文件状态：
1. created:
   - output_file: "# Agent Task\n..."
   - manifest: { status: "running", started_at: "...", completed_at: null }

2. running:
   - output_file: 持续追加中间结果
   - manifest: 保持不变

3. completed:
   - output_file: 追加 "\n## Result\n\n- status: completed\n\n### Final response\n\n..."
   - manifest: { status: "completed", completed_at: "...", error: null }

4. failed:
   - output_file: 追加 "\n## Result\n\n- status: failed\n\n### Error\n\n..."
   - manifest: { status: "failed", completed_at: "...", error: "..." }

八、错误处理与容错

8.1 错误类型

错误类型 发生位置 处理方式
输入解析错误 execute_agent 返回 Err，不创建子 Agent
线程创建失败 spawn_agent_job 更新 manifest 为 failed，返回 Err
Panic 子线程内部 catch_unwind 捕获，记录 “sub-agent thread panicked”
运行时错误 run_agent_job 捕获 Err，更新 manifest 为 failed
工具执行失败 SubagentToolExecutor 返回 ToolError，记录到会话消息
API 错误 AnthropicRuntimeClient 返回 RuntimeError，终止子 Agent

8.2 Panic 恢复

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
std::thread::Builder::new()
    .spawn(move || {
        let result = std::panic::catch_unwind(
            std::panic::AssertUnwindSafe(|| run_agent_job(&job))
        );
        
        match result {
            Ok(Ok(())) => {}  // 成功
            Ok(Err(error)) => {
                // 运行时错误
                persist_agent_terminal_state(&job.manifest, "failed", None, Some(error));
            }
            Err(_) => {
                // Panic
                persist_agent_terminal_state(
                    &job.manifest, "failed", None, 
                    Some("sub-agent thread panicked".to_string())
                );
            }
        }
    })

保证：即使子 Agent panic，也不会影响主进程或其他子 Agent。

8.3 超时控制

当前实现：

  • 子 Agent 通过 DEFAULT_AGENT_MAX_ITERATIONS 限制最大迭代次数
  • 单个工具调用可通过 timeout 参数限制（如 bash 工具）

未来改进：

  • 添加整体执行超时（如 30 分钟）
  • 支持用户自定义超时

九、性能优化

9.1 线程池 vs 每任务一线程

当前设计：每创建一个子 Agent 就启动一个新线程

优点：

  • 实现简单
  • 完全隔离
  • 易于调试（线程名称包含 agent_id）

缺点：

  • 大量子 Agent 时线程开销大
  • 无线程复用

优化方向：

  • 引入线程池（如 tokio::task::spawn）
  • 使用 async/await 替代阻塞线程

9.2 API 连接复用

当前设计：每个子 Agent 创建独立的 AnthropicClient

优化方向：

  • 共享 HTTP 客户端连接池
  • 复用 TCP 连接

9.3 文件系统优化

当前设计：每次追加都打开/关闭文件

优化方向：

  • 保持文件句柄打开直到完成
  • 使用缓冲区减少 I/O 次数

十、安全考虑

10.1 工具权限隔离

风险 缓解措施
子 Agent 滥用 bash Explore/Plan 类型禁止 bash
递归创建 Agent 所有子 Agent 禁止调用 Agent 工具
文件覆盖 每个子 Agent 写入独立的 output_file
资源耗尽 限制 max_iterations，建议限制并行度

10.2 沙箱建议

当前：子 Agent 与主 Agent 运行在同一进程，共享文件系统权限

增强方案：

  • 使用容器（Docker）隔离子 Agent
  • 使用 seccomp 限制系统调用
  • 使用 namespaces 隔离网络和 PID

十一、调试与监控

11.1 线程名称

每个子 Agent 线程都有唯一名称：

1
clawd-agent-{agent_id}

查看运行中的子 Agent：

1
ps aux | grep clawd-agent

11.2 Manifest 文件

每个子 Agent 都有 manifest 文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
  "agentId": "a1b2c3d4",
  "name": "ship-audit",
  "description": "审计代码分支",
  "subagentType": "Explore",
  "model": "claude-sonnet-4-6",
  "status": "completed",
  "outputFile": ".clawd-agents/a1b2c3d4.md",
  "manifestFile": ".clawd-agents/a1b2c3d4.manifest.json",
  "createdAt": "2026-04-01T10:00:00Z",
  "startedAt": "2026-04-01T10:00:00Z",
  "completedAt": "2026-04-01T10:05:30Z",
  "error": null
}

11.3 输出文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# Agent Task

- id: a1b2c3d4
- name: ship-audit
- description: 审计代码分支
- subagent_type: Explore
- created_at: 2026-04-01T10:00:00Z

## Prompt

检查测试用例和未完成的工作

## Result

- status: completed

### Final response

我发现了 3 个失败的测试用例和 2 个未完成的 TODO...

十二、扩展与定制

12.1 添加新的 Sub-agent Type

步骤：

  1. 在 allowed_tools_for_subagent 中添加新类型：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    fn allowed_tools_for_subagent(subagent_type: &str) -> BTreeSet<String> {
        let tools = match subagent_type {
            // ... 现有类型
            "CodeReview" => vec![
                "read_file",
                "grep_search",
                "StructuredOutput",
            ],
            _ => vec![...],
        };
        tools.into_iter().map(str::to_string).collect()
    }

  2. 在 normalize_subagent_type 中添加别名：

    1
    2
    3
    4
    5
    6
    fn normalize_subagent_type(subagent_type: Option<&str>) -> String {
        match subagent_type.map(str::trim).unwrap_or_default() {
            "codereview" | "review" => "CodeReview".to_string(),
            // ...
        }
    }

  3. 更新系统提示模板（可选）：

    1
    2
    3
    fn build_agent_system_prompt(subagent_type: &str) -> Result<Vec<String>, String> {
        // 根据 subagent_type 添加特定指令
    }

12.2 自定义工具白名单

可以在创建 Agent 时动态指定工具：

1
2
3
4
5
6
7
8
{
  "tool": "Agent",
  "input": {
    "description": "自定义任务",
    "prompt": "执行特定操作",
    "allowed_tools": ["read_file", "bash", "write_file"]
  }
}

（需要扩展 AgentInput 结构以支持此功能）

12.3 结果回调

未来可以添加 webhook 回调：

1
2
3
4
5
6
7
8
{
  "tool": "Agent",
  "input": {
    "description": "异步任务",
    "prompt": "处理数据",
    "callback_url": "https://example.com/webhook"
  }
}

子 Agent 完成后 POST 结果到 callback_url。

十三、总结

13.1 核心设计原则

  1. 隔离性：每个子 Agent 独立线程、独立会话、独立工具集
  2. 异步性：主 Agent 不阻塞，立即返回 manifest
  3. 安全性：基于 subagent_type 的工具白名单，禁止递归创建
  4. 可观测性：完整的 manifest 和 output 文件记录
  5. 容错性：Panic 捕获、错误持久化、不影响主进程

13.2 适用场景

✅ 适合：

  • 并行信息收集（多个 Explore Agent）
  • 长时间运行任务（后台验证、测试）
  • 危险操作隔离（文件删除、系统修改）
  • 任务分解与编排（主 Agent 协调多个子 Agent）

❌ 不适合：

  • 需要实时交互的任务
  • 高度依赖共享状态的任务
  • 超低延迟要求的场景

13.3 未来演进方向

  1. 线程池优化：减少线程创建开销
  2. Agent 间通信：消息队列或共享状态存储
  3. 动态工具授权：运行时调整工具白名单
  4. 优先级调度：高优先级 Agent 优先执行
  5. 资源配额：限制每个 Agent 的 Token/CPU/内存使用
  6. 可视化监控：Web UI 查看所有运行中的 Agent

文档版本：1.0
最后更新：2026-04-01
基于代码版本：rust/crates/tools v0.1.0

Claw-Code Agent 架构分析与工作流程（Python 实现版）
http://example.com/2026/04/12/claw-code-python/
作者
Kevin
发布于
2026年4月12日
许可协议