子会话与多智能体协作 OpenClaw 的子会话机制让 AI 从"单兵作战"升级为"团队协作"。每个子会话都是一个独立的微型智能体,有自己的上下文、工具和决策能力。本节深入拆解子会话的架构原理、四种协作模式,以及两个完整的实战案例。 你能学到什么 理解子会话的独立性、完整性和隔离性三大特征 掌握并行分工、流水线、专家顾问、辩证审查四种协作模式 能使用 创建子会话并管理其生命周期 通过内容创作团队和智能客服两个案例,学会设计多智能体系统 一、子会话的核心架构 1.1 什么是子会话 子会话是主会话创建的独立执行单元。跟普通的函数调用不同,子会话是一个完整的智能体——它有独立的内存上下文、完整的推理能力、独立的工具权限。
OpenClaw 的子会话机制让 AI 从"单兵作战"升级为"团队协作"。每个子会话都是一个独立的微型智能体,有自己的上下文、工具和决策能力。本节深入拆解子会话的架构原理、四种协作模式,以及两个完整的实战案例。
sessions_spawn 创建子会话并管理其生命周期子会话是主会话创建的独立执行单元。跟普通的函数调用不同,子会话是一个完整的智能体——它有独立的内存上下文、完整的推理能力、独立的工具权限。
四个关键特性:
| 特性 | 含义 | 实际影响 |
|---|---|---|
| 独立性 | 拥有独立的内存上下文 | 不与父会话共享变量状态 |
| 完整性 | 继承主会话的技能和工具 | 能执行主会话能做的一切操作 |
| 隔离性 | 崩溃不影响其他会话 | 一个子会话出错,主流程不受影响 |
| 可控性 | 主会话可监控和引导 | 可以动态调整子会话的执行方向 |
OpenClaw 的会话系统分三层:
主会话是用户直接交互的顶层会话,负责任务分解和结果汇总。子会话由主会话创建,专注于特定子任务。线程会话是子会话内部更细粒度的并行单元,生命周期依附于创建它的子会话。
每个会话都有唯一的层级化标识:
agent:main:qqbot:direct:e6f63319045821b25a5a9fb90f76b0ebagent:main:subagent:ff95203d-4252-4c8d-a38f-add0f99572cd前缀 agent:main 表示所属的智能体系统,中间部分标识渠道类型或子会话类型,最后是 UUID。
💡 调试技巧:排查子会话问题时,会话标识是关键线索。在日志中搜索 subagent 关键字,能看到所有子会话的创建、执行和完成记录。
多个子会话同时处理同一任务的不同部分,最后汇总结果。
适用场景:数据聚合、多源搜索、批量处理。核心优势是速度——三个搜索任务同时跑,总耗时约等于最慢的那个,而不是三者之和。
创建并行子会话的方式:
# 创建 3 个子会话同时搜索不同来源 subagent_1 = await spawn_subagent( task="搜索学术数据库", context={"source": "arxiv, scholarly"} ) subagent_2 = await spawn_subagent( task="搜索技术博客", context={"source": "medium, dev.to"} ) subagent_3 = await spawn_subagent( task="搜索官方文档", context={"source": "docs, github"} ) # 等待所有结果返回 results = await gather(subagent_1, subagent_2, subagent_3) merged = merge_results(results)
子会话按顺序处理,每个会话的输出是下一个的输入。
适用场景:内容创作、数据处理流水线、多阶段分析。核心优势是质量——每个阶段由专门的子会话负责,比一个子会话从头做到尾的质量更高。
# 阶段 1:研究 researcher = await spawn_subagent( task="收集资料", context={"topic": topic} ) research_data = await researcher.result() # 阶段 2:写作 writer = await spawn_subagent( task="撰写初稿", context={"research": research_data} ) draft = await writer.result() # 阶段 3:编辑 editor = await spawn_subagent( task="润色编辑", context={"draft": draft} ) final = await editor.result()
主会话作为协调者,根据问题类型动态创建专家子会话。
适用场景:智能客服、技术咨询、决策支持。核心优势是专业度——不同类型的問題由不同领域的专家处理,而不是让一个通用智能体硬扛所有问题。
专家映射表:
| 问题类型 | 专家角色 | 核心能力 |
|---|---|---|
| 技术问题 | 技术专家 | 编程、架构、调试 |
| 账单咨询 | 账单专家 | 费用、订单、支付 |
| 功能使用 | 使用专家 | 操作指导、最佳实践 |
| Bug 反馈 | 反馈专家 | 问题定位、复现 |
| 功能建议 | 产品专家 | 需求分析、优先级 |
创建持不同观点的子会话进行辩论,再由综合子会话平衡双方论点。
适用场景:决策分析、风险评估、方案对比。核心优势是全面性——强制从正反两面思考,避免单一视角的盲区。
# 正方 pro = await spawn_subagent( task="论证方案优势", context={"topic": topic, "stance": "支持"} ) # 反方 con = await spawn_subagent( task="论证方案风险", context={"topic": topic, "stance": "反对"} ) # 综合 synthesis = await spawn_subagent( task="综合平衡双方论点", context={ "pro_args": await pro.result(), "con_args": await con.result() } )
⚠️ 模式选择不要过度设计:如果你的任务只需要一个子会话就能搞定,不要硬套四种模式。并行分工和流水线是最常用的两种,先把这两种用熟,再根据实际需要引入其他模式。
💡 调试子会话的好方法:开发阶段给每个子会话加上 label 参数(比如 label="研究员"),日志里就能按标签过滤,快速定位是哪个子会话出了问题。
子会话之间不能直接通信,所有信息传递都通过主会话中转。
| 方式 | 方向 | 说明 |
|---|---|---|
| 初始上下文传递 | 主 → 子 | 创建子会话时传入任务和参数 |
| 结果报告 | 子 → 主 | 子会话完成后自动推送结构化结果 |
| 推送式通知 | 子 → 主 | 无需轮询,结果自动推送 |
| 引导(Steering) | 主 → 子 | 动态调整正在运行的子会话方向 |
结果报告的标准格式:
{ "status": "completed", "result": { "data": "..." }, "metadata": { "duration": 45.2, "tokens_used": 3200 } }
主会话可以在子会话运行过程中发送引导消息,改变其执行方向:
await subagent.steer( message="调整搜索重点,关注大语言模型优化方向" )
这在流水线模式中特别有用——如果上游子会话的输出质量不达标,主会话可以引导它重新处理,而不是重新启动一个全新的子会话。
构建一个自动化的内容创作团队,实现从研究到发布的完整流程:
主编(主会话)的职责是协调三个子会话的执行顺序,传递上下文,并在每个阶段完成后检查结果质量。研究员负责从多个来源搜索资料并过滤出高质量内容。写作者基于研究结果构建文章结构并逐节撰写。编辑负责多轮润色(结构优化、语言润色、技术准确性检查、SEO 优化)。
每个子会话完成后都会返回结构化的结果报告,包括产出物、元数据(字数、耗时)和质量评分。主编根据质量评分决定是否需要返工。
💡 增量式优化:不必等研究员完全结束才启动写作者。可以用流式处理——研究员每找到一批资料就推送给写作者,写作者先开始写已有资料支撑的章节,后续资料到了再补充。这样能显著缩短总耗时。
构建一个多层智能客服系统,实现问题分流、专家解答和自动总结:
分流器子会话做三件事:意图识别(用户想干什么)、实体提取(涉及什么对象)、紧急度评估(需要多快响应)。根据分类结果,主会话从专家池中创建对应类型的专家子会话。
专家池支持动态调度——咨询量大时自动创建多个同类型专家实例,咨询量小时减少实例数。每个专家的解答都会记录到交互日志中,用于后续分析和优化路由规则。
⚠️ 权限最小化:不同类型的子会话应该只拥有完成任务所需的最小权限。研究员只需要搜索权限,写作者只需要文件读写权限,发布环节才需要 API 发布权限。这样即使某个子会话被注入恶意指令,损失也是可控的。
并发控制:不要同时创建太多子会话。每个子会话都消耗内存和 API 配额。建议设置并发上限(比如最多 5 个子会话同时运行),用信号量控制:
MAX_CONCURRENT = 5 semaphore = asyncio.Semaphore(MAX_CONCURRENT)
超时管理:给每个子会话设置超时时间,避免某个子会话卡死拖垮整个流程:
result = await asyncio.wait_for( spawn_subagent(task), timeout=300 # 5 分钟超时 )
缓存复用:对于重复性任务(比如同一个主题的多次搜索),用缓存避免重复执行。
