🔄 Sub-Agent

"世界上有一种智慧叫做分解。当面对一个庞大而复杂的任务时，聪明的做法不是独自硬扛，而是把它拆成小块，让每个Sub-Agent各司其职。就像一支足球队，前锋、中场、后卫，每个人都有自己的位置和责任。"

📖 什么是 Sub-Agent？

Sub-Agent（子Agent） 是 OpenClaw 中用于任务分解和并行处理的核心机制。它允许一个主 Agent 创建多个独立的子 Agent，每个子 Agent 负责处理任务的一部分，最后主 Agent 汇总所有结果。

🔧 sessions_spawn 详解

基本语法

sessions_spawn(
    task="子Agent的任务描述",
    runtime="subagent",      # 运行时类型：subagent 或 acp
    mode="run",              # 模式：run（一次性）或 session（持久）
    model="gpt-4",           # 可选：指定模型
    timeoutSeconds=300,      # 可选：超时时间
    thread=false             # 可选：是否在线程中运行
)

两种运行模式

🛠️ 实战应用

场景1：内容生产流水线

# 主Agent：协调整个内容生产流程
def produce_blog_post(topic):
    
    # Sub-agent 1: 研究员 - 收集资料
    research = sessions_spawn(
        task=f"深入研究主题：{topic}。搜索最新资讯，整理5-10个关键信息点。",
        runtime="subagent",
        mode="run"
    )
    
    # Sub-agent 2: 大纲撰写员 - 创建文章结构
    outline = sessions_spawn(
        task=f"基于以下研究资料，创建文章大纲：\n{research}\n\n要求：包含引言、3-5个主要章节、结论",
        runtime="subagent",
        mode="run"
    )
    
    # Sub-agent 3: 内容撰写员 - 撰写正文
    draft = sessions_spawn(
        task=f"根据以下大纲撰写完整文章：\n{outline}\n\n风格：专业且易懂，字数2000字左右",
        runtime="subagent",
        mode="run"
    )
    
    # Sub-agent 4: 编辑 - 润色修改
    final = sessions_spawn(
        task=f"编辑以下文章，改善语言表达和结构：\n{draft}",
        runtime="subagent",
        mode="run"
    )
    
    return final

场景2：代码审查系统

# 并行代码审查
def parallel_code_review(code_changes):
    
    # 同时启动多个审查Sub-agent
    reviews = []
    
    # 安全性审查
    security_review = sessions_spawn(
        task=f"审查以下代码的安全漏洞：\n{code_changes}\n\n重点关注：SQL注入、XSS、敏感数据泄露",
        runtime="subagent",
        mode="run"
    )
    reviews.append(security_review)
    
    # 性能审查
    perf_review = sessions_spawn(
        task=f"审查以下代码的性能问题：\n{code_changes}\n\n重点关注：算法复杂度、数据库查询、内存使用",
        runtime="subagent",
        mode="run"
    )
    reviews.append(perf_review)
    
    # 可维护性审查
    maintain_review = sessions_spawn(
        task=f"审查以下代码的可维护性：\n{code_changes}\n\n重点关注：代码风格、注释、模块化程度",
        runtime="subagent",
        mode="run"
    )
    reviews.append(maintain_review)
    
    # 汇总所有审查结果
    return aggregate_reviews(reviews)

场景3：数据分析任务

# 大规模数据分析 - 分片处理
def analyze_large_dataset(dataset_path):
    
    # 将数据集分成多个分片
    shards = split_dataset(dataset_path, num_shards=5)
    
    # 为每个分片创建Sub-agent
    analysis_tasks = []
    for i, shard in enumerate(shards):
        task = sessions_spawn(
            task=f"分析数据分片 {i+1}，计算统计指标：平均值、标准差、分布情况\n数据路径：{shard}",
            runtime="subagent",
            mode="run"
        )
        analysis_tasks.append(task)
    
    # 汇总所有分片的分析结果
    combined = combine_analyses(analysis_tasks)
    
    # 生成最终报告
    report = sessions_spawn(
        task=f"基于以下分析结果生成综合报告：\n{combined}",
        runtime="subagent",
        mode="run"
    )
    
    return report

⏱️ 等待与同步

使用 sessions_yield

# 启动Sub-agent后，主Agent需要等待结果
sub_task = sessions_spawn(
    task="执行耗时任务...",
    runtime="subagent",
    mode="run"
)

# 主Agent结束当前回合，等待Sub-agent结果
sessions_yield(message="已启动子任务，等待完成...")

# 下一回合，主Agent收到Sub-agent的结果
# 继续处理...

🎯 最佳实践

1. 任务粒度

2. 错误处理

try:
    result = sessions_spawn(
        task="可能失败的任务",
        runtime="subagent",
        mode="run",
        timeoutSeconds=60
    )
except TimeoutError:
    # 超时重试或降级
    result = fallback_approach()
except Exception as e:
    # 记录错误
    log_error(e)
    result = None

3. 结果验证

# 验证Sub-agent输出
def validate_result(result, expected_format):
    if not result:
        return False, "空结果"
    
    if not meets_format(result, expected_format):
        return False, "格式不符"
    
    if not meets_quality(result):
        return False, "质量不达标"
    
    return True, "验证通过"

# 使用验证器
is_valid, message = validate_result(sub_result, "json")
if not is_valid:
    # 重新委派或人工介入
    handle_validation_failure()

🚀 进阶技巧

1. 分层Sub-agent

# 三层Sub-agent架构
# Level 1: 协调器
# Level 2: 模块负责人  
# Level 3: 具体执行者

def hierarchical_task(project):
    # Level 1: 项目分解
    modules = sessions_spawn(
        task=f"将项目分解为模块：{project}",
        runtime="subagent",
        mode="run"
    )
    
    results = []
    for module in modules:
        # Level 2: 模块负责人
        module_result = sessions_spawn(
            task=f"负责模块 {module.name}，协调子任务",
            runtime="subagent",
            mode="run"
        )
        results.append(module_result)
    
    return combine_results(results)

2. 动态任务调整

# 根据中间结果调整后续任务
initial_result = sessions_spawn(task="初步分析...", ...)

# 分析结果，决定下一步
if initial_result.confidence < 0.7:
    # 置信度低，增加验证步骤
    validation = sessions_spawn(task="验证初步结果...", ...)
    final_result = reconcile(initial_result, validation)
else:
    final_result = initial_result

🎬 结语

"世界上有一种力量叫做分解。Sub-Agent 不是逃避复杂，而是用智慧征服复杂。当每个小块都被妥善处理，整体就有了坚实的根基。"

掌握 Sub-Agent 的使用，你就掌握了扩展 AI 处理能力的关键。开始设计你的分层 Agent 架构吧！