👥 一个 Agent 不够用?那就来一群!通过子Agent编排,你可以让多个专业 Agent 协作完成复杂任务。本文教你如何使用 OpenClaw 的子Agent系统构建强大的 AI 团队。
子Agent编排是指将复杂任务拆分给多个专业 Agent 并行或串行执行的模式。每个子Agent专注于特定领域,通过协作完成整体目标。
主 Agent 分配任务,子 Agent 执行并汇报
多个 Agent 并行执行,汇总结果
串行执行,前一个的输出是后一个的输入
多个 Agent 讨论,达成共识
# 使用 sessions_spawn 创建子Agent
sessions_spawn(
task="搜索今日 AI 热点新闻",
taskName="news_searcher",
mode="run"
)
# 返回子Agent会话,等待完成
# 等待所有子Agent完成
sessions_yield(message="等待子Agent完成")
# 子Agent完成后会收到通知
# 列出所有子Agent
subagents(action="list")
# 查看子Agent历史
sessions_history(sessionKey="subagent_key")
主 Agent 作为 Leader,分配任务给子 Agent 并汇总结果。
# Leader 分配任务给多个专家
# 1. 创建 SEO 专家
seo_expert = sessions_spawn(
task="分析网站 SEO 状况,检查 meta 标签、死链、sitemap",
taskName="seo_expert"
)
# 2. 创建内容专家
content_expert = sessions_spawn(
task="生成 3 篇高质量博客文章",
taskName="content_expert"
)
# 3. 创建数据专家
data_expert = sessions_spawn(
task="分析过去 7 天的流量数据",
taskName="data_expert"
)
# 4. 等待所有专家完成
sessions_yield(message="等待所有专家完成")
# 5. 汇总结果
seo_report = sessions_history(sessionKey=seo_expert)
content_report = sessions_history(sessionKey=content_expert)
data_report = sessions_history(sessionKey=data_expert)
# 6. 生成综合报告
generate_final_report(seo_report, content_report, data_report)
多个 Agent 并行执行相同任务,取最佳结果或汇总。
# Swarm 模式:多个 Agent 同时搜索
tasks = [
"搜索 OpenClaw 最新动态",
"搜索 AI Agent 行业新闻",
"搜索 MCP 协议进展"
]
# 并行创建多个搜索 Agent
sessions = []
for task in tasks:
session = sessions_spawn(
task=task,
taskName=f"searcher_{len(sessions)}",
mode="run"
)
sessions.append(session)
# 等待所有完成
sessions_yield(message="等待所有搜索完成")
# 汇总搜索结果
all_results = []
for session in sessions:
results = sessions_history(sessionKey=session)
all_results.extend(results)
# 去重和排序
final_results = deduplicate_and_rank(all_results)
串行执行,前一个的输出是后一个的输入。
# Pipeline 模式:数据处理流水线
# Step 1: 数据收集
collector = sessions_spawn(
task="收集竞品网站数据",
taskName="data_collector"
)
sessions_yield()
raw_data = sessions_history(sessionKey=collector)
# Step 2: 数据清洗
cleaner = sessions_spawn(
task=f"清洗和格式化数据: {raw_data}",
taskName="data_cleaner"
)
sessions_yield()
clean_data = sessions_history(sessionKey=cleaner)
# Step 3: 数据分析
analyzer = sessions_spawn(
task=f"分析数据并生成洞察: {clean_data}",
taskName="data_analyzer"
)
sessions_yield()
insights = sessions_history(sessionKey=analyzer)
# Step 4: 生成报告
reporter = sessions_spawn(
task=f"生成可视化报告: {insights}",
taskName="report_generator"
)
sessions_yield()
final_report = sessions_history(sessionKey=reporter)
多个 Agent 讨论,达成共识。
# Debate 模式:技术方案讨论
# 创建不同观点的 Agent
optimist = sessions_spawn(
task="支持使用微服务架构,列出优点",
taskName="optimist"
)
pessimist = sessions_spawn(
task="反对使用微服务架构,列出缺点和风险",
taskName="pessimist"
)
moderator = sessions_spawn(
task="综合双方观点,给出平衡的建议",
taskName="moderator"
)
# 等待讨论完成
sessions_yield()
# 获取最终建议
final_decision = sessions_history(sessionKey=moderator)
| 模式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| Leader Election | 复杂项目管理 | 职责清晰,易于管理 | Leader 成为瓶颈 |
| Swarm | 并行搜索、批量处理 | 速度快,容错性好 | 结果可能不一致 |
| Pipeline | 数据处理、内容生成 | 流程清晰,质量可控 | 延迟较高 |
| Debate | 决策、方案评估 | 多角度分析,减少偏见 | 耗时较长 |
# 处理子Agent失败
try:
session = sessions_spawn(
task="执行可能失败的任务",
taskName="risky_task"
)
sessions_yield()
except Exception as e:
logger.error(f"子Agent失败: {e}")
# 执行降级策略
fallback_strategy()
# 完整的内容运营流程
# 1. 热点追踪 Agent
trend_spotter = sessions_spawn(
task="搜索今日 AI 行业热点",
taskName="trend_spotter"
)
# 2. 内容创作 Agent
content_creator = sessions_spawn(
task="基于热点创作 3 篇博客文章",
taskName="content_creator"
)
# 3. SEO 优化 Agent
seo_optimizer = sessions_spawn(
task="优化文章 SEO,添加 meta 标签",
taskName="seo_optimizer"
)
# 4. 发布 Agent
publisher = sessions_spawn(
task="发布文章到网站并更新 sitemap",
taskName="publisher"
)
# 5. 推广 Agent
promoter = sessions_spawn(
task="将文章分享到社交媒体",
taskName="promoter"
)
# 等待所有完成
sessions_yield()