Claude驱动的动态工作流:从任务拆解到结果合成

在当前的AI工程领域,多Agent协作已经成为提升任务效率和质量的重要方向。就在不久前,围绕Loop Engineering的讨论还在行业内盛行,而如今Graph Engineering的概念已经迅速崛起,成为新的技术热点。近期技术分享中提到,Loop Engineering已经逐渐被Graph Engineering所取代,后者通过更灵活的任务编排方式,实现了更高效的协作。
动手前的四个核心判断
Graph并非Loop的简单升级,而是对Loop的组织方式进行了重构。相较于单个Loop,Graph需要消耗更多的Token资源,也带来了更高的协调开销,当出现问题时,排查难度也会显著提升。因此在动手搭建Graph工程之前,需要先明确四个关键问题,确保投入是值得的。
一、任务是否能拆分为明确的角色分工?如果无法清晰定义每个参与者的职责边界,那么强行拆分只会增加不必要的成本,此时继续使用单个Loop反而更加高效。
二、是否存在真正可以并行执行的子任务?如果所有子任务都存在严格的线性依赖,那么Graph不仅不会提升效率,反而会因为额外的协调开销降低整体性能。只有当存在独立并行的工作时,Graph的优势才能体现出来。
三、单个Agent的上下文是否能承载全部任务背景?如果单个Agent可以处理所有的任务信息,那么拆分任务只会浪费资源。拆分的核心目的是为了释放上下文空间,而非为了形式上的复杂。
四、失败后的分支跳转成本是否在可承受范围内?如果没有明确的失败处理机制,Graph可能会在不可见的情况下出现卡顿或执行异常。需要提前规划好重试耗尽后的处理路径,比如退回上一步、切换备用节点或转由人工介入。
还有一个比以上四个问题更重要的附加题:你是否已经拥有了一个稳定运行的单体Loop?如果没有,那么暂时不要尝试搭建Graph。Graph是Loop的组织方式,而非替代品,只有当单体Loop运行稳定后,才能考虑通过Graph进一步优化协作效率。
哪些团队适合尝试Graph Engineering?
已经成功稳定运行至少一个单体Loop的团队,任务中存在明确可拆分的角色(比如调研、撰写、审核等),并且愿意接受更高的Token成本来换取更高的任务质量和并行执行效率。
哪些场景不适合Graph Engineering?
尚未让单体Loop稳定运行的个人开发者、任务存在强线性依赖无法拆分、瓶颈在于协调开销而非单节点能力的团队,都暂时不适合引入Graph Engineering。总体而言,当前大多数团队的单体Loop尚未跑稳,因此还不需要考虑Graph的搭建。
Graph工程的四个核心构件
一个可正常运行的Graph工作流,可以拆分为四个可独立验证的核心部分:
节点:图的最小执行单元。每个节点都是一个独立运行的Agent或确定性执行步骤,只处理单一任务,只认可明确的输入并输出标准的结果。如果一个节点无法明确判断任务完成的标准,那么它就不是一个合格的节点,只是隐藏的依赖项。
边:定义数据流转与执行依赖。边并非简单的“执行顺序”,而是明确的数据传递承诺:上游节点输出特定格式的数据,下游节点按照该格式进行消费。边可以分为顺序边、条件边和并行边,边留给模型判断的空间越多,Graph的灵活性越强,但排查难度也会越高。
共享状态:全局数据交互载体。下游节点需要使用的字段,必须由上游节点明确写入。共享状态的设计会倒逼团队梳理清楚任务中的所有环节,发现那些尚未被明确考虑的细节。
失败路由:异常处理机制。当一个节点的重试次数耗尽后,控制权应该如何转移?是退回上一步、切换到备用节点,还是转由人工处理?没有失败路由的Graph,只是一张静态流程图,而非可正常运行的系统。
Graph工程的实操步骤
第一步:厘清节点与边的核心边界
一张Agent工作流图本质上只有两个核心元素:节点和边。节点是独立的执行单元,有明确的输入输出;边则定义了数据如何从一个节点流转到下一个节点,而非简单的执行顺序。很多人容易犯的错误是将“然后”当成边,比如“先总结文件,再查询天气”,这两个步骤之间其实没有数据依赖,只是线性执行顺序,并非真正的边。只有当数据确实从上游节点流转到下游节点时,才存在真正的边。
当你将一个Agent写成“先做A,再做B,再做C”时,其实已经构建了一张退化的Graph,只是一条不分岔的单链。这种线性执行的方式虽然简单,但存在明显的缺陷:如果其中一个节点出现故障,后续所有节点都无法正常执行。Graph工程的第一项能力,就是重新梳理这条单链,移除那些没有数据依赖的虚假边,将其拆分为可并行执行的独立节点,再通过合并节点整合结果。
第二步:为节点和边制定契约
一个无法被推理的节点,无法融入并行执行的体系。解决这个问题的方法是为每个节点制定明确的契约:输入有边界,输出有标准,只处理单一任务。输入必须是显式传递的,不能依赖节点从共享空间自动获取;输出则应该是可校验的结构化数据,这样下游节点可以直接使用,无需额外解析。
在工作流中,这种契约可以通过JSON Schema强制实现。为Agent调用配置Schema后,返回的结果会自动经过校验,格式错误时会自动重试,避免返回自由文本导致后续处理困难。这也是“可直接接入Graph的节点”和“仅人工可解析的节点”的核心区别。
// 一个符合契约的节点:输入有边界,输出经过校验,仅处理单一任务
const ITEM = {
type: 'object', additionalProperties: false,
properties: {
title: { type: 'string' },
url: { type: 'string' },
impact: { type: 'string', enum: ['high', 'medium', 'low'] },
},
required: ['title', 'url', 'impact'],
};
const result = await agent(source.prompt, {
label: `research:${source.key}`,
schema: ITEM, // 强制返回校验后的结构化数据
agentType: 'general-purpose',
});
// result现在是下游节点可信任的格式,无需手动解析
边同样可以看作是一种数据契约:上游节点承诺输出特定格式的数据,下游节点按照该格式进行消费。通过数据名称而非执行顺序来命名边,可以更清晰地判断边是否真实存在,也可以在不改变数据格式的前提下替换节点,不会破坏整个Graph的结构。
第三步:构建扇出、扇入与菱形拓扑
扇出:一次性派发多个并行任务
当拥有多个独立节点时,不要将它们按顺序执行,而是通过并行执行来提升效率。在工作流中,可以使用parallel()方法:将任务数组传入,为每个任务分配一个子Agent,所有任务并发执行,最后将结果集合一次性返回。
需要注意两个关键细节:一是parallel()会等待所有任务完成后再返回结果,二是失败的任务会被解析为null,不会拖垮整个批次,因此需要对结果进行.filter(Boolean)过滤掉失败的节点。并发数可以根据系统资源进行调整,超出限制的任务会自动排队执行。
phase('Research');
// 多个数据源,并行执行任务
const raw = await parallel(
SOURCES.map(s => () => agent(s.prompt, {
label: `research:${s.key}`,
phase: 'Research',
schema: ITEM_SCHEMA, // 每个节点返回校验后的JSON
agentType: 'general-purpose',
}))
);
const collected = raw.filter(Boolean); // 过滤失败的任务
扇入:在屏障处合并结果
并行派发任务后,需要有节点来合并结果。合并节点可以一次性获取所有上游结果,处理需要全集信息才能完成的任务,比如跨数据源去重、按影响力排序、当结果为空时提前终止流程。这是Graph中唯一值得等待屏障的场景,因为只有合并节点需要全部前置结果才能执行。
// 边可以通过普通JS实现,无需Agent,零Token消耗
const flat = collected.flatMap(c => c.items);
log(`Collected ${flat.length} items`);
phase('Curate');
// 合并节点需要全部结果才能执行去重和排序
const curated = await agent(
`Dedupe and rank these by impact:\n${JSON.stringify(flat)}`,
{ phase: 'Curate', schema: CURATED_SCHEMA },
);
菱形拓扑:拆分-执行-合并
将扇出和扇入结合起来,就得到了最常用的Graph拓扑:菱形结构。一个节点负责拆分任务,多个节点并行执行,最后一个节点合并结果。这种拓扑结构广泛应用于市场扫描、依赖审计、代码评审等场景,只需要更换数据源和提示词,就可以复用这套骨架。
其标准流程可以概括为:派发任务→归约压缩→合成结果。通过这种方式,可以将复杂的任务拆解为可并行的子任务,大幅提升执行效率。
第四步:路由、验证与节点隔离
动态路由:根据节点结果选择执行路径
并非所有的Graph都是固定的,有些执行路径需要根据节点的执行结果动态选择。路由节点会检查上游结果,决定后续的执行路径,比如根据工单分类分配到不同的处理节点,或者根据代码变更的风险程度选择快速评审或完整审计。在工作流中,路由逻辑可以通过普通的if/switch语句实现,判断依据是上游节点校验后的结果,确保控制流的可靠性。
// 路由节点:Agent负责分类,代码负责选择路径
const { severity } = await agent(
`Classify this diff's risk:\n${diff}`,
{ schema: { type: 'object', properties: { severity: { enum: ['low', 'high'] } }, required: ['severity'] } },
);
let review;
if (severity === 'high') {
// 高风险:并行完整审计
review = await parallel(FILES.map(f => () => agent(`Audit ${f}`)));
} else {
// 低风险:快速评审
review = await agent(`Quick review of ${diff}`);
}
结果验证:为Graph增加确定性
Graph的核心优势不仅在于并行执行,更在于围绕结果构建的确定性。验证器节点会在结果放行到下游之前,尝试推翻该结果,只有通过验证的结果才能进入后续流程。常见的验证模式有三种:
- 对抗式验证:为每个发现分配多个独立的怀疑者,专门反驳该发现,多数未被驳倒的结果才算可靠。
- 多视角验证:让每个验证者从不同角度检查结果,比如正确性、安全性、可复现性等,覆盖越全面,越容易发现隐藏的问题。
- 评委制:从多个角度生成多个方案,通过并行的评委打分选择最优方案,再整合其他方案的亮点。
节点隔离:避免失败扩散
在单链执行中,一个节点失败会导致整个链条停摆,但在Graph中,失败应该被限制在单个节点内。通过parallel()执行的任务中,失败的节点会被过滤掉,不会影响其他节点的执行。同时,需要设计每个合并节点能够容忍缺失的输入,而非假设所有上游节点都能成功执行。
更隐蔽的风险是节点之间的资源冲突,比如多个Agent并行写入文件时可能出现碰撞。解决方案是使用隔离机制,比如为每个Agent分配独立的工作区,在沙盒中完成任务后再合并结果,只在真正需要并行写入时使用这种机制,避免不必要的开销。
第五步:循环、模型分层与拓扑优化
收敛式循环:处理未知规模的任务
当任务规模未知时,比如漏洞排查可能会发现新的问题,此时需要引入受控的循环。但需要注意避免无限循环,确保循环能够收敛。标准的收敛写法是“持续执行直到连续多轮没有新发现”,同时需要对所有见过的结果进行去重,避免重复处理相同的内容。
const seen = new Set();
const confirmed = [];
let dry = 0;
while (dry < 2) { // 连续两轮无新发现则停止
const found = (await parallel(FINDERS.map(f => () => agent(f.prompt, { schema: BUGS })))).filter(Boolean).flatMap(r => r.bugs);
const fresh = found.filter(b => !seen.has(key(b)));
if (!fresh.length) { dry++; continue; }
dry = 0;
fresh.forEach(b => seen.add(key(b))); // 对所有见过的结果去重
// 每个新发现都需要经过多视角验证
const judged = await parallel(fresh.map(b => () => parallel(['correctness', 'security', 'repro'].map(lens => () => agent(`Judge "${b.desc}" via ${lens} — real?`, { schema: VERDICT }))).then(v => ({ b, real: v.filter(Boolean).filter(x => x.real).length >= 2 })))));
confirmed.push(...judged.filter(v => v.real).map(v => v.b));
}
模型分层:优化Token成本
并非所有节点都需要使用最强的模型。有些节点处理的是有明确边界的重复性任务,比如抽取字段、分类工单,可以使用低成本模型;而需要判断力的节点,比如合成报告、验证结果,则需要使用高成本的优质模型。通过在单个Agent调用中指定model选项,可以为不同节点分配不同的模型,在不影响效果的前提下大幅降低Token成本。
拓扑优化:平衡成本与延迟
Graph的拓扑结构直接决定了运行成本和延迟。最常见的选择是使用pipeline()还是parallel():parallel()会等待所有任务完成后再进入下一阶段,而pipeline()则让每个任务独立依次执行,无需等待全部完成,更快的任务可以提前结束。默认情况下应该优先使用pipeline(),只有当需要全部前置结果时才使用parallel()屏障,避免不必要的延迟。
第六步:让Agent自动生成Graph
对于无法提前规划的任务,可以使用动态工作流让Agent自动生成Graph。只需要描述任务目标,Agent会自动编写编排脚本,拆解任务、分配子Agent、合并结果,最终生成适配当前任务的Graph。
有三种使用方式:一是在Prompt中提及“workflow”,Agent会自动为任务编写编排脚本;二是运行已存储或内置的工作流,其内置了完整的执行骨架:定范围→并行搜索→抓取→对抗式验证→合成;三是开启相关功能,Agent会为会话中的每个任务自动规划工作流,执行完成后可以保存脚本以便后续复用。
› Run a workflow to audit every route under src/routes/ for missing auth. Spawn one agent per route file, then verify each finding before reporting.
● Agent wrote an orchestration script · launching in background…
/workflows — auth-audit · running ✓
Scope 1/1 2.1k tok
Fan-out 18/18 one agent per route file
Verify 11/18 3-vote skeptics per finding…
Synthesize 0/1 waiting on verify
// 会话保持响应,任务在后台继续执行
总结
近年来,多Agent协作的技术杠杆一直聚焦在单体Loop的优化上,比如更可靠的验证机制、更稳定的退出条件、更清晰的状态管理。而如今,如何将这些Loop进行合理的编排连接,已经成为新的技术护城河。Graph Engineering不仅是技术形式的升级,更是多Agent协作模式的进阶,能够帮助团队在复杂任务中实现更高的效率和质量。

