多角色协作与子流程 Playbook
企业场景里经常会出现“让多个 Agent 协作”的说法。落到工程实现时，真正要设计的通常不是一个新的框架原语，而是四件事：
- 每个角色负责什么判断；
- 角色之间传什么结构化结果；
- 哪些状态和产物共享，哪些必须隔离；
- 父流程如何分派、等待、汇总和回放。
Agently 里更稳的表达是：用角色化 Agent 处理专门判断，用 TriggerFlow sub-flow 隔离子任务，用 Workspace 设计共享/独立状态拓扑，用父流程完成分派、汇总和交付。
什么时候需要多角色协作
先判断一个 Agent 是否已经够用。
| 场景 | 推荐做法 |
|---|---|
| 只是一次分类、摘要、表单抽取 | 一个 AgentExecution + output schema |
| 同一输入要同时经过风险、事实、合规、编辑等不同视角 | 父 TriggerFlow 分派多个角色子流 |
| 每个角色会产生自己的证据、产物、审查记录 | 每个子流绑定独立 Workspace 或独立 collection |
| 多个角色之间需要等待、审批、重跑或合并 | 父流程负责生命周期和结果聚合 |
| 只是想“更智能一点” | 不要先拆多 Agent，先把输出契约和验收规则写清楚 |
多角色协作的价值不在“模型数量变多”，而在把不同责任、不同证据和不同失败边界拆清楚。
推荐结构
Gateway / API
-> parent TriggerFlow execution
-> capture request context
-> to_sub_flow(risk_review_flow)
-> to_sub_flow(fact_check_flow)
-> to_sub_flow(editor_flow)
-> aggregate structured outputs
-> write decision / artifact / checkpoint to Workspace
-> return product-facing result这个结构里，父流程拥有业务主线；子流程拥有角色细节；Workspace 保存可追溯证据；最终输出仍然回到结构化业务字段。
三种状态拓扑
| 拓扑 | 何时用 | 注意事项 |
|---|---|---|
| 一个共享 Workspace | 多个角色处理同一份材料，需要复用证据和产物 | 约定 collection 名称，避免互相覆盖 |
| 每个角色独立 Workspace | 角色之间需要隔离，例如合规审查和编辑生成不能共享草稿 | 父流程只拿结构化 summary / artifact ref |
| 独立 collection + 共享 artifact | 大多数企业场景 | 共享原始材料，角色写入各自 observation / decision |
不要把共享 state 当作协作。协作应该通过明确的 input、output、artifact ref 和 decision 发生。
子流程怎么划分
一个角色子流程应该能独立回答：
- 输入是什么；
- 输出字段是什么；
- 需要哪些 Actions、MCP 或外部资源；
- 哪些中间证据要写入 Workspace；
- 失败、拒绝、需要人工确认时怎么表达。
如果回答不出来，说明它还不是一个稳定角色，只是父流程里的一个普通 chunk。
from agently import Agently, TriggerFlow
def create_role_agent(role_prompt: str):
agent = Agently.create_agent()
agent.define(prompt={"role": role_prompt})
return agent
def build_risk_review_flow():
risk_agent = create_role_agent("你负责识别风险、证据和需要人工确认的点。")
flow = TriggerFlow(name="risk-review")
async def review(data):
result = (
risk_agent
.input(data.input["document"])
.output({
"risk_level": (str, "low / medium / high", True),
"findings": ([str], "风险点", True),
"requires_approval": (bool, "是否需要人工确认", True),
})
.get_result()
)
review_data = await result.async_get_data()
await data.async_set_state("risk_review", review_data)
return review_data
flow.to(review)
return flow父流程再用 to_sub_flow(...) 调用角色子流程，并通过 capture / write_back 控制交接。
parent.to(prepare).to_sub_flow(
build_risk_review_flow(),
capture={"input": "value"},
write_back={
"state.risk_review": "snapshot.risk_review",
},
).to(aggregate)实际项目里，角色 Agent 可以挂接自己的 Actions、Skills 或 Workspace；父流程只要求它返回稳定结构。
结果汇总怎么做
父流程汇总时不要让模型“凭感觉读聊天记录”。更好的输入是每个角色返回的结构化结果：
{
"risk_review": {
"risk_level": "high",
"findings": ["缺少审批记录"],
"requires_approval": true
},
"fact_check": {
"verified": false,
"missing_evidence": ["供应商付款凭证"]
},
"editor": {
"draft_artifact_ref": "workspace://artifacts/report-draft.md"
}
}父流程可以用确定性规则先处理硬约束：只要任一角色要求审批，就进入 pause_for(...)；只有都通过，再进入最终生成或发送。
runtime stream 怎么给用户看
子流程里的 data.async_put_into_stream(...) 会进入父 execution 的 runtime stream。面向产品界面时，建议统一成业务事件：
await data.async_put_into_stream({
"type": "role_status",
"role": "risk_review",
"stage": "done",
"summary": "发现 2 个高风险点，需要人工确认",
})前端或 IM 网关不需要知道子流内部 chunk 名称，只消费 role、stage、summary、artifact_ref 这类稳定字段。
不要这样做
| 误用 | 为什么不稳 |
|---|---|
| 同时启动 5 个 Agent，让它们自由聊天 | 没有交接契约、没有可复核结果、难以回放 |
| 每个角色都能看到全部工具 | 角色越多，越容易越权或误选能力 |
| 用共享全局 state 传递所有中间结果 | 并发 execution 容易互相污染，也难恢复 |
| 把角色输出作为自然语言段落传给下一个角色 | 下游无法做硬规则校验和聚合 |
| 让子流程自己决定最终交付 | 父流程失去业务主线和验收边界 |
验收清单
| 检查项 | 通过标准 |
|---|---|
| 角色边界 | 每个角色有独立职责、输入、输出和失败语义 |
| 交接契约 | 父子流程通过结构化字段、artifact ref 和 state key 交接 |
| 状态拓扑 | 共享与隔离的 Workspace / collection 选择明确 |
| 行动权限 | 角色只看到完成职责所需 Actions / MCP / Skills |
| 过程输出 | 父 execution 的 runtime stream 能展示每个角色进度 |
| 恢复能力 | pause/resume、checkpoint 和人工审批都基于父 execution 管理 |
| 质量证据 | 聚合结果有业务规则校验、模型 judge 或人工确认路径 |