Cord Trees — 动态任务树编排

在运行时构建协调树。你来决定分解方式，而不是开发者。

灵感来源于 June Kim 的 Cord。

核心概念

不是遵循预定义的工作流，而是分析目标并构建自己的任务树：

目标：评估是否应从 REST 迁移到 GraphQL

你决定：
├── #1 生成：审计当前 REST API 接口
├── #2 生成：研究 GraphQL 的权衡
├── #3 询问：有多少并发用户？（被 #1 阻塞）
├── #4 分支：对比分析（被 #2、#3 阻塞）
└── #5 分支：撰写建议（被 #4 阻塞）

树从你的分析中生成，而不是从硬编码逻辑中生成。

五种原语

1. 生成 — 隔离上下文

子节点只获取其任务提示。从零开始。

python
spawn(
goal=研究 GraphQL 采用模式,
prompt=搜索 REST→GraphQL 迁移的案例研究...,
blocked_by=[] # 可以立即开始
)

使用场景： 任务是自包含的，不需要兄弟上下文。

2. 分支 — 继承上下文

子节点接收所有已完成的兄弟节点结果注入到提示中。

python
fork(
goal=综合发现形成建议,
prompt=基于研究，撰写建议...,
blocked_by=[research-rest, research-graphql, user-scale]
)

使用场景： 需要先前工作的综合、分析或整合。

3. 询问 — 人工引导

暂停等待人工输入。创建一个检查点。

python
ask(
question=你们服务多少并发用户？,
options=[<1K, 1K-10K, 10K-100K, >100K],
blocked_by=[audit-api] # 在审计提供上下文后询问
)

使用场景： 决策需要人工知识或批准。

4. 串行 — 有序序列

子节点按顺序执行。隐式依赖关系。

python
serial([
{goal: 起草报告, type: spawn},
{goal: 审核草稿, type: ask},
{goal: 定稿报告, type: fork}
])

使用场景： 需要严格排序。

5. 目标 — 根节点

顶层目标。你将其分解为子节点。

使用 OpenClaw 实现

将 Cord 原语映射到 OpenClaw 工具：

Cord 原语	OpenClaw 实现
spawn	sessionsspawn(task=prompt, label=id)
fork

sessionsspawn 在任务中注入兄弟结果 |
| ask | 向人工发送消息，等待响应 |
| serial | 顺序生成，每次之间等待 |
| read_tree | 读取状态文件 + subagents list |
| complete | 将结果写入状态文件 |

状态文件

在 cord-state.json 中跟踪树：

json
{
goal: 评估 REST 到 GraphQL 迁移,
nodes: {
#1: {
type: spawn,
goal: 审计 REST API,
status: complete,
result: 47 个端点，12 个嵌套...,
blockedBy: [],
sessionKey: abc123
},
#2: {
type: spawn,
goal: 研究 GraphQL,
status: running,
blockedBy: [],
sessionKey: def456
},
#3: {
type: ask,
goal: 获取用户规模,
status: waiting,
question: 有多少并发用户？,
options: [<1K, 1K-10K, 10K-100K, >100K],
blockedBy: [#1]
},
#4: {
type: fork,
goal: 对比分析,
status: blocked,
blockedBy: [#2, #3]
}
},
nextId: 5
}

工作流

阶段 1：分析目标

阅读目标。思考：

• - 主要组成部分是什么？
• 哪些可以并行运行？
• 哪些有依赖关系？
• 哪里需要人工输入？
• 哪些需要综合（分支）vs 隔离（生成）？

阶段 2：构建初始树

为第一层分解创建节点：

python

初始化状态

state = {
goal: user_goal,
nodes: {},
nextId: 1
}

添加初始节点

add_node(state, type=spawn, goal=研究 A, blockedBy=[]) add_node(state, type=spawn, goal=研究 B, blockedBy=[]) add_node(state, type=fork, goal=综合, blockedBy=[#1, #2])

write(cord-state.json, state)

阶段 3：执行就绪节点

找到就绪的节点（所有 blockedBy 已完成）：

python
def getreadynodes(state):
ready = []
for id, node in state[nodes].items():
if node[status] != blocked:
continue
deps = node[blockedBy]
if all(state[nodes][d][status] == complete for d in deps):
ready.append(id)
return ready

对于每个就绪节点：

如果是生成：
python
sessions_spawn(
task=node[prompt],
label=node_id,
runTimeoutSeconds=600
)
node[status] = running

如果是分支：
python

注入兄弟结果

siblingcontext = collectsibling_results(state, node)
fullprompt = f{node[prompt]}\n\n来自先前工作的上下文：\n{siblingcontext}

sessionsspawn(task=fullprompt, label=node_id)
node[status] = running

如果是询问：
python

向人工发送消息

message(action=send, message=f问题：{node[question]}\n选项：{node[options]})
node[status] = waiting

等待响应，然后用答案标记完成

阶段 4：监控与更新

轮询运行中的代理，在完成时更新状态：

python
while hasrunningor_blocked(state):
# 检查代理状态
agents = subagents(action=list)

for agent in agents:
node = findnodeby_session(state, agent[sessionKey])
if agent[status] == complete:
# 从会话历史获取结果
result = getagentresult(agent)
node[status] = complete
node[result] = result

# 分派新就绪的节点
for nodeid in getready_nodes(state):
dispatchnode(state, nodeid)

save_state(state)
wait(30) # 不要过于频繁地轮询

阶段 5：综合

当所有节点完成时，最终的分支节点产生结果。

动态重构

代理可以在运行时修改自己的子树：

python

子代理意识到需要更多研究

addchildnode(
parent=#2,
type=spawn,
goal=深入探讨性能影响,
blockedBy=[]
)

这就是 Cord 风格编排的强大之处——树基于代理的发现而演变。

生成 vs 分支决策指南

情况	使用
独立研究任务	spawn
不需要兄弟上下文的任务

spawn | | 失败后重新启动成本低 | spawn | | 跨先前工作的综合或分析 | fork | | 最终整合步骤 | fork | | 基于发现构建的任务 | fork |

默认使用生成。 仅在需要上下文继承时使用分支。

人在回路模式

审批关卡

#1 spawn: 起草提案
#2 ask: 批准此提案？ (blocked-by: #1)
#3 fork: 实施已批准的提案 (blocked-by: #2)

澄清

#1 spawn: 初步分析
#2 ask: 我们应该聚焦哪个方向？ (blocked-by: #1)
#3 spawn: 深入探讨选定方向 (blocked-by: #2)

定期检查点

#1 spawn: 阶段 1
#2 ask: 继续到阶段 2？ (blocked-by: #1)
#3 spawn: 阶段 2 (blocked-by: #2)
#4 ask: 继续到阶段 3？ (blocked-by: #3)
...

示例：完整分解

目标：创建全面的竞争对手分析报告

#1 [spawn] 列出前