基于图的Agent编排：DAG、循环与条件分支实战指南

从去年下半年开始我一直在做Agent相关的项目落地陆陆续续接触了不少框架也踩了不少坑。起初大家做Agent大多是单点能力比如一个模型封装一层提示词丢给业务方就完了。但是等到Agent真正要处理复杂任务——比如多轮信息收集、多工具协同、质量校验不过要重写、答案不好要换条路——你就会发现单纯靠线性链式的代码去编排Agent根本撑不住。这也是为什么“基于图的Agent编排”这个概念越来越热DAG、循环、条件分支这几个词几乎成了Agent框架和编排方向的标配关键词。这篇文章我就拿自己实际开发和重构的经验和大家聊聊基于图的Agent编排到底是什么、为什么非要这么做、以及怎么把DAG、循环、条件分支这些结构真正落地到自己的代码里去。不管你是正在做Agent开发的学习者还是已经在折腾Agent框架的老手只要你遇到过“流程一复杂代码就烂掉”“重复执行没法收敛”“分支逻辑写死没法扩展”这类问题这篇内容应该能帮你把思路重新捋一遍。1. 为什么Agent编排需要“图”而不是“链”1.1 线性编排的局限当工作流不再是一条直线先说说最直接的问题。大部分刚接触Agent开发的人最先学会的编排方式就是“链式调用”一个Agent的输出直接作为下一个Agent的输入代码写出来就是一段接一段的API调用。这种方式在任务链路短、逻辑固定的场景下确实没问题比如“写一段新闻标题→扩写成完整短文→生成摘要→打上标签”一条直线走到底逻辑简单清晰出问题也好排查。但实际项目里一旦增加需求这种线性结构的脆弱性就立刻暴露。举个例子我做一个市场分析Agent时最初设计是用户输入产品名→Agent搜资料→生成分析报告。看起来很简单对吧但真实业务里有很多分支搜到的资料质量不高怎么办需要先追问用户补充信息还是直接改用备用数据源生成的报告有严重事实错误需要回炉重写还是送人工审核不同用户可能只需要分析报告的一部分是全部生成还是按模块按需生成这些问题一旦出现前端可能会因为紧张而影响发挥无法自如表现。你要么把所有逻辑硬编码嵌套在代码里要么就引入状态机去处理各种条件跳转。等到逻辑复杂到一定程度代码就开始失控了——到处都是状态位和if分支加一个新需求要动十几个地方。1.2 图的本质把事情的组织方式还给开发者我个人的体会是链式编排本质上是在“画一条线”而图的编排是在“画一个网”。把Agent的执行关系抽象成图最大的好处是节点和边是解耦的。你可以单独维护每一个Agent节点的能力也可以单独定义节点之间怎么连接。这就好比做菜。链式编排像一份固定菜单从头菜到甜点一道一道按顺序上中间不能换也不能调。图的编排则更像厨房动线洗菜、切配、炒菜、装盘每个工位独立存在哪个工序需要哪个工位的产出按需连接。某个工位出了问题不影响其他工位的运转甚至可以并行开工。图的这种特性对Agent编排带来的直接价值有三点一是任务关系可视化DAG画出来整个Agent协作逻辑一目了然二是节点复用度大幅提升同一个“质量评估Agent”既可以在生成步骤后使用又可以在改写步骤后使用不用复制代码三是执行方式灵活多样可以顺序、可以并行、可以条件跳转、可以循环迭代全部由图的拓扑结构决定。这也是为什么现在主流的Agent框架包括LangGraph、AutoGen、CrewAI这类工具几乎全部转向了图编排的思路。1.3 三种图结构的定位DAG、循环图、条件分支在基于图的Agent编排里DAG、循环、条件分支分别解决不同层面的问题。DAG有向无环图解决的是“一次性、有依赖关系的任务流”怎么组织。比如数据预处理必须在前模型推理在后最后再做结果后处理这就是典型的DAG结构。DAG没有环意味着每个节点最多执行一次非常适合流水线式的Agent协作。循环Cycle解决的则是“反复执行直到满足条件”的场景。Agent在推理过程中经常需要自我反思、反复改写、多轮迭代这时候如果只能用DAG表达你会发现自己陷进一个死胡同——没法让流程回到前面的节点去。所以循环结构的核心价值就是允许Agent工作流“回头”。条件分支Conditional Branch解决的是“下一步走哪条路”的问题。Agent执行过程中的不确定性是天然存在的模型输出的质量、工具调用的结果、外部数据的可达性都会影响接下来的走向。条件分支让Agent可以基于上一步的结果自主决定下一步的行动路径这是Agent区别于普通自动化脚本的关键一点。这三种结构在实际系统中往往是混合使用的。一个复杂的Agent工作流里主流程是DAG中间穿插若干条件分支某些子流程内部又包含循环。理解了它们各自的角色你才能在设计Agent编排时做到游刃有余。2. 核心设计把Agent变成图上的节点2.1 节点的抽象一个Agent节点到底封装了什么要把Agent装进图框架里第一步就是把“Agent”这个模糊的概念拆成一个个可以被执行的节点。我在项目里的做法是定义一个统一的AgentNode接口每个节点必须具备三样东西输入Schema必须明确这个节点需要哪些字段。输出Schema必须明确这个节点执行成功后产出哪些字段。执行函数接收输入执行Agent逻辑返回结构化的输出。很多刚开始做Agent开发的同行容易忽略Schema的设计觉得把整个大模型调用结果传下去就行了省事。但后期你会为这种做法付出惨重代价尤其是要加循环和条件分支的时候。如果你的节点没有明确的结构化输入和输出条件判断根本没有依据——你不知道该拿哪个字段去评分、去判断。我在重构成图编排的时候第一步就是把所有节点的输入输出全部结构化这一步完成后后面所有工作都顺畅了。节点内部的实现完全是黑盒可以是调用大模型的逻辑可以是调用外部API也可以是一个纯函数计算。图编排框架不关心节点内部怎么实现只关心你的输入输出是否匹配。这种设计带来的直接好处就是后期想替换某个Agent的模型、Prompt或者是调用策略只需要改节点内部代码整个图的结构一行都不用动。2.2 边的语义数据流、控制流与状态传递有了节点之后接下来关键在于“边”的理解。很多人觉得图里的边不就是一条带箭头的线吗两个节点连一起A执行完执行B这就完了。但真实做下来你会发现边至少有三种不同的语义数据依赖Data DependencyB节点需要A节点的输出作为输入这是最基础的一种。在DAG里如果B依赖A的结果那么执行顺序上A必须先于B。控制依赖Control DependencyB节点不一定需要A的输出数据但必须等到A执行完才能开始以保证时序正确。比如日志记录节点、审计节点、状态更新节点都属于这一类。条件依赖Conditional DependencyB节点是否执行取决于A节点输出的某些字段。这就是条件分支的关键实现。在实际代码里这三种边往往需要不同的实现策略。数据依赖最直观框架需要做好状态管理让每个节点的输出可以被下游节点通过key来引用。控制依赖可以转化成一个空的数据依赖来统一处理。条件依赖则需要引入专门的Router节点来做判断不能直接在边上写表达式否则图的语义会被弄得很脏。关于状态传递我的实践经验是用一个全局的State对象作为所有节点之间共享的数据总线。每个节点从State里取自己需要的输入字段把输出字段写回State。这种方式看起来简单粗暴但在实际工程中是最容易排错、最容易做可视化追踪的方案。2.3 图执行引擎的调度策略拓扑排序与并行执行有了节点、边和状态之后还需要一个调度引擎来驱动整张图跑起来。对于DAG部分标准做法是拓扑排序——把图中所有节点排成线性顺序保证在执行任何节点之前它的所有前置节点都已经被执行。但这里有一个很容易被忽略的优化点拓扑排序只保证了时序的合法性并没有充分利用并行能力。如果你的DAG里有多个互不依赖的分支理论上它们是可以并行执行的。比如在“生成内容方案→同时拉取资料、分析竞品、构建报告大纲”这种结构里“拉取资料”“分析竞品”“构建大纲”三个节点互不依赖完全可以并发执行让整体耗时从三者之和缩短为三者最大值。我实际做执行引擎的时候用的方案是“入度为零”的拓扑调度策略。维护一个Ready队列初始把入度为0的节点全部放进去然后并发执行队列里的节点每完成一个节点就更新它所指向的下游节点的入度一旦下游节点入度归零就把它加进Ready队列。这套逻辑实现起来不复杂但对执行效率的提升非常明显。需要特别提醒的是循环结构和DAG在调度上有一个根本差异——DAG中每个节点最多执行一次而循环中的节点可能执行多次。所以在设计执行引擎时需要区分节点的“类型”普通节点DAG节点、循环体节点可能在一次图执行中被多次调用、分支路由节点只负责判断下一步去向。3. 实操从零实现一个轻量级Agent编排框架3.1 基础数据结构Graph、Node、Edge的定义说了这么多理论还是得来点能跑的。我先分享一个轻量级的Agent编排框架核心实现大家可以直接参考或者抄走改改。语言用Python依赖只需要标准库加一个Pydantic用于Schema校验。先看基础的数据结构定义from typing import Callable, Any, List, Dict, Optional from enum import Enum from pydantic import BaseModel, Field import asyncio from collections import deque class NodeType(str, Enum): TASK task # 普通任务节点 ROUTER router # 条件路由节点 LOOP loop # 循环控制节点 class AgentNode(BaseModel): node_id: str node_type: NodeType NodeType.TASK handler: Callable None # 实际的执行函数 input_fields: List[str] [] # 需要从State中取哪些字段 output_fields: List[str] [] # 执行完往State中写哪些字段 class Config: arbitrary_types_allowed True class Edge(BaseModel): source: str target: str condition: Optional[Callable] None # 条件函数返回True才执行target class Config: arbitrary_types_allowed True class Graph(BaseModel): nodes: Dict[str, AgentNode] {} edges: List[Edge] []写这个类的时候我特意把condition设计成一个可选的Callable而不是写死的字符串表达式。这样灵活性最大化你要用简单的lambda可以要接复杂的评分逻辑也可以完全取决于你的场景需求。State的定义我用Pydantic的话会更规范但为了简洁先直接用dict。实际生产环境建议用带版本控制的数据结构否则并行写State会出问题这一点后面会在常见问题里继续聊。3.2 条件分支节点执行结果的“岔路口”条件分支的实现在图编排中是一个重点。我的做法是引入一个专门的Router节点它的handler根据输入数据返回一个字符串这个字符串决定接下来走哪条边。举个例子我的Agent工作流里有一个“内容质量评估”节点它会返回一个score字段分数在0到1之间。Router节点拿到score之后如果大于0.7就进入“直接发布”路径否则就进入“人工改写”路径。代码如下async def quality_router(state: Dict[str, Any]) - str: score state.get(quality_score, 0) if score 0.7: return publish elif score 0.4: return revise else: return reject def make_quality_graph() - Graph: gen_node AgentNode( node_idgenerate, handlergenerate_article, input_fields[topic], output_fields[draft, quality_score] ) router AgentNode( node_idquality_router, node_typeNodeType.ROUTER, handlerquality_router, input_fields[quality_score], output_fields[] ) graph Graph( nodes{generate: gen_node, quality_router: router}, edges[ Edge(sourcegenerate, targetquality_router), ] ) return graphRouter的返回值还需要作用在边的条件判断上。具体做法是在执行引擎调度时如果当前节点是一个Router就先执行它拿到路由结果然后根据路由结果筛选出当前分支可用的出边。边上的条件函数可以这样设计——它接收两个参数一个是当前State一个是Router返回的分支标识def edge_condition(state: Dict[str, Any], route_result: Optional[str]) - bool: if route_result is None: return True # 非路由节点无条件放行 return edge.target route_result这个设计的好处是新增一个分支路径不需要修改已有的节点和边只需要添加一条新的边和一个新的处理节点。对Agent开发来说这会让你增加新能力的成本低很多。3.3 循环的实现让流程可以“回头”循环是图编排里稍难啃的一块骨头。和DAG不同循环要求执行引擎有能力把已经执行过的节点再次加入执行队列。我在设计时把循环拆成两个部分来理解循环体要重复执行的节点集合和循环条件决定是否继续循环的节点。一个典型的场景是“多轮迭代优化”。Agent生成一份文案然后交给“评审Agent”评分如果得分低于目标值就回到“改写Agent”重新加工直到得分达标或者达到最大迭代次数。这里“回到改写Agent”就是一个循环。在数据结构的层面循环本质上就是在DAG上增加一条“回头”的边从循环条件节点指回循环体的开始节点。但这里有一个问题如果不加限制DAG的性质就被破坏了拓扑排序的算法不能直接使用。所以我的处理方式是对循环做特殊的执行控制。async def execute_with_loop( graph: Graph, state: Dict[str, Any], loop_start: str, loop_condition: str, max_iterations: int 10 ): iteration 0 while True: iteration 1 if iteration max_iterations: state[loop_terminated] max_iterations break # 执行一轮从loop_start到loop_condition的流程 await execute_scope(graph, state, startloop_start, endloop_condition) # 判断是否继续循环 should_continue state.get(should_loop, False) if not should_continue: state[loop_terminated] condition_met break这个实现虽然简化了不少但核心逻辑是清楚的把循环拆成“执行一轮→检查条件→决定是否再来一轮”的迭代模式。关键的隐藏条件是循环体内节点的State字段读写必须设计好不能让上一轮的旧数据污染下一轮的计算。项目里我用了一个比较土但有效的方法——循环体内所有节点执行前先对State做一次深拷贝备份一轮执行完后如果决定退出循环就保留最终结果如果继续循环就用初始状态叠加部分保留字段再做一遍。3.4 完整示例一个带分支和循环的Agent工作流把上面的所有逻辑拼起来我给大家展示一个可以用来参考的完整示例。这个示例模拟了一个“产品文案智能生成与优化”的Agent工作流包含三个阶段初稿生成、质量评估与分支路由、循环优化。async def generate_draft(state): topic state[topic] response await call_llm(f针对主题『{topic}』写一篇产品推广文案初稿) state[draft] response state[revision_count] 0 return state async def evaluate_quality(state): draft state[draft] score await call_llm_score(draft) state[quality_score] score return state def route_by_quality(state): if state[quality_score] 0.8: return finish else: return rewrite async def rewrite_draft(state): draft state[draft] suggestion state.get(revision_suggestion, 提升文案吸引力和点击率) new_draft await call_llm(f根据建议『{suggestion}』修改文案{draft}) state[draft] new_draft state[revision_count] 1 return state主流程的逻辑是generate_draft执行完→evaluate_quality评分→route_by_quality路由。如果分数合适走finish分支结束如果分数不合适走rewrite分支在rewrite执行完后检查revision_count没超过3次就回到evaluate_quality超过就强制结束。这里最核心的执行顺序是rewrite完必须再评估再决定是否继续而不是rewrite完直接进入下一个环节。这也是循环和普通条件分支容易搞混的地方——很多新手会把循环写成分支结果只改了一次就结束了达不到迭代优化的效果。4. 常见问题与排查技巧实录4.1 死循环图编排最容易翻车的坑我最早做循环Agent时踩过最大的坑就是死循环。一个“自我反思”的Agent在迭代改写文案时因为评分阈值设得太高模型生成的内容始终拿不到理想分数节点陷入无限循环最后直接耗尽API额度。这个教训花了我不少成本。死循环的防范我在工程上总结了三道基本的保障最大迭代次数所有循环都必须有硬性的迭代上限超出即终止哪怕结果不理想也要放出去。可以在State里维护一个iteration_count字段循环每轮自增一次。连续结果变化检测如果连续若干轮的结果与上一轮几乎一样说明模型已经收敛了继续改写没有意义。可以直接终止循环。超时熔断为整张图的执行设置一个全局超时比如总执行时间超过60秒就强制结束。这是最后的兜底方案。另外我要特别提醒一个细节循环条件判断的依据字段必须确保在循环体中会被更新。很多死循环的真正原因不是循环逻辑问题而是循环体内的节点没有把“决定是否继续循环”的那个字段回写到State里导致判断条件永远不变化。这个错误排查起来非常隐蔽因为光看代码可能觉得逻辑没问题实际跑起来就卡死。4.2 状态冲突并行节点共同修改同一字段这个坑是我在尝试并行执行DAG分支节点时发现的。假设你的工作流中有一个节点是“生成三种不同风格的文案”它们并行执行最后都会往State里写入draft字段。三个节点同时写同一个key最终结果就是后写覆盖先写另外两个分支的努力直接白费。解决这个问题我提供了两种方案具体选哪种看你场景方案一是给State的写入加上命名空间前缀。比如三个并行节点分别写draft_v1、draft_v2、draft_v3然后再用一个合并节点从中选取最终版本。这种做法适合并行分支结果独立的场景可追溯性较好。方案二是直接用不可变性数据结构和写时复制Copy-on-Write机制每个节点操作的是State的一个快照版本最终由父节点统一合并。这适合边数较多、层级较深的图结构虽然实现复杂度高一些但对状态一致性的保障更强。4.3 条件分支误判一个问题三条路都踩过条件分支本身的代码逻辑并不难真正难的是条件判断的“依据”是否可靠。我遇到的几次分支误判分析下来基本都出在同一个问题上用来做分支判断的字段值不确定性太大。打个比方你用大模型给内容评分把输出直接parse成float。大模型返回的JSON可能格式不规范、可能数字前后有额外文字、甚至可能直接拒绝评分返回空字符串。如果解析逻辑不健壮router节点就会拿到异常值然后走错分支。应对策略有两个方向。一是在router节点前加一个专门的“数据清洗节点”把大模型输出解析成严格类型解析失败就采用默认值或者重试一次。二是对关键分支的“阈值”留出缓冲区比如判断“是否通过”的阈值是0.7那0.69和0.71之间不要死卡边界最好设置一个0.65到0.75的灰色区域落入灰色区域就不做硬性路由而是走人工兜底或者重试。4.4 调试策略图编排的“可观测性”问题图编排框架用起来爽但调试起来会比线性代码更痛苦。线性代码你打日志就能看到执行到哪一步图结构则可能同时有好几个节点在跑还有条件跳转干扰视线。项目里我积累了一套针对图编排的调试方法分享给大家第一步启动图级日志。每执行一个节点在调度引擎里记录节点的node_id、开始时间、结束时间、输入字段、输出字段落盘到结构化日志里。这样你可以完整还原一次图执行的路径。第二步为每轮图执行生成一个Trace ID。所有节点日志、API调用日志、中间状态变化都带上这个Trace ID排查问题的时候全链路搜索。第三步可视化图执行路径。把图和State的转储成一个JSON快照直接渲染成可视化图表。这套东西搭起来确实要花一些时间但你会发现在Agent开发遇到复杂问题时这套工具比任何代码review都管用。5. 实战经验扩展从能跑到可用、再到好用5.1 从“能跑”到“稳定”生产级图编排的几个关键点如果你已经按照前面的代码把图编排的雏形跑通了恭喜你但这只是刚刚开始。我自己的项目从Demo到真正能稳定对接业务中间又花了大概三倍的时间做工程加固。有几个点我认为格外值得投资状态持久化。生产环境里一张复杂的Agent图可能执行好几分钟中途任何一个节点崩溃都有可能导致整个流程回滚。把State定期持久化到Redis或者数据库里崩溃后可以从最近的快照点恢复执行这个能力对长耗时Agent非常关键。节点级重试与退避。每个节点执行的底座不同比如有的Agent调大模型API有的调外部爬虫不同的服务故障率相差很大。图编排框架应该允许给每个节点单独配置重试次数和退避策略。执行限流。当图里有大量并行节点同时跑的时候如果不加全局限流几十个并发请求可能直接把上游的API打爆。我做过一次压测10个并行分支同时调同一个大模型API结果直接触发了限流整张图全部报错重试最后耗时比串行执行还长。失败节点的降级策略。有些节点失败其实是可接受的比如“摘要生成节点”挂了能不能临时用“截取前500字”来代替把这个降级逻辑做成一个可配置项你可以大大提升图执行的容错率。5.2 明确送分题你未必需要自研图编排框架讲了很多自研的代码和框架实现这里要诚实地说一句如果你的项目已经有条件使用成熟框架不一定要自研。市面上像LangGraph、LlamaIndex Workflow、Temporal这类工具底层都提供了较为成熟的图编排和任务调度能力。但如果你是下面这几种情况我建议可以先自研一个轻量的你的Agent数量不多、依赖关系简单引入一个重量级框架反而增加学习成本。你有非常强的定制化需求比如希望整个执行流程可视化到公司内部平台上。你希望深入理解图编排的底层原理不是为了产出工具而是为了提升能力。自研和用框架的取舍没有绝对标准说到底要看团队情况。我的经验是前两个项目可以用成熟框架积累经验等遇到框架满足不了的场景时再动手自研那时候你对问题的理解会比一开始就闷头写深刻得多。5.3 图编排后续还能扩展的方向最后聊聊图编排后续的扩展方向这些方向我现在已经在逐步尝试效果还不错。动态图目前介绍的图结构是预先定义好的。但在真实的Agent场景中有些节点是动态发现的——比如根据用户输入的不同临时添加一个“查天气Agent”或“查航班Agent”节点进图里。这种动态图对执行引擎的要求更高但也更贴近智能体的本质。图的嵌套与组合把一张已经完成的子图作为一个节点嵌入到另一张更大的图里形成层级化的编排。这种方式对组织大型Agent团队非常有效可以让不同模块的Owner独立开发和测试自己的子图最后再组合起来。基于学习的自动路由条件分支里的Router节点现在大多是基于规则或阈值判断。更进一步的做法是把Router本身交给一个轻量级模型去决策让它根据当前上下文动态决定走哪条分支路径。这种设计会让Agent的自主性提升一个台阶。我在实际使用中发现图编排框架一旦搭好后面扩Agent能力的效率提升是肉眼可见的。以前加一个新功能节点要在一堆if判断里找位置现在只需要往图里加一个node、连两条边整个世界都清爽了。不过还是要提醒大家一句框架是工具真正让Agent好用的还是你对任务拆解的深度和对业务的理解工具解决的是“怎么串起来”的问题而“串什么内容”永远是你的核心设计能力。