swarms GraphWorkflow 实战指南用图工作流编排多 Agent 非线性协作【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms图工作流Graph Workflow是 swarms 框架中用于编排复杂、非线性多 Agent 交互的核心机制Agent 是图上的节点Agent 之间的关系构成有向边由此可以表达条件路径、并行分支、基于中间结果动态路由、扇出fan-out/扇入fan-in等远超线性管道的能力。本文以 examples/multi_agent/graphworkflow_examples/README.md 为骨架结合 graph_workflow.py 源码与配套示例带你从零掌握GraphWorkflow的建图、编译、执行、校验、可视化、序列化与并行化并给出可复制运行的完整代码。一、认识 Graph Workflow从线性管道到有向图编排1.1 为什么需要图工作流传统多 Agent 编排多采用顺序管道Sequential WorkflowAgent 依次执行、前一个的输出直接作为后一个的输入。这类模型结构简单但无法表达条件路径根据中间结果决定后续走向并行分支多个 Agent 同时处理互不依赖的子任务动态汇聚多个分支的输出在某个 Agent 处汇合再统一处理多轮迭代将上一轮终点结果回注到下一轮起点实现循环改进。Graph Workflow 正是为这些场景设计图中节点之间的边显式声明了数据依赖关系执行引擎依据拓扑序自动分层调度从而把谁先谁后、谁能并行交给图结构决定。1.2 目录结构与示例总览仓库中 graphworkflow_examples 目录 集中收录了图工作流示例README 列出的核心文件包括文件主题graph_workflow_basic.py最基础的链式图工作流graph_workflow_example.py完整工作流 JSON 导出advanced_graph_workflow.py高级图模式graph_workflow_validation.py工作流结构校验graph_workflow_composition.py图组合子图嵌套graph_workflow_serialization.py序列化与反序列化graph_workflow_fanout_fanin.py扇出/扇入模式graph_workflow_checkpointing.py检查点续跑graph_workflow_streaming.py流式输出test_enhanced_json_export.pyJSON 导出测试test_graph_workflow_caching.py编译缓存测试test_graphviz_visualization.pyGraphviz 可视化测试test_parallel_processing_example.py并行处理测试目录内还有两个子目录graph/核心图工具示例基础与完整示例的简版变体rustworkx_examples/基于 rustworkx 高性能后端的系列示例含性能基准、错误处理、大规模工作流等 10 个脚本与测试 test_graph_workflow_rustworkx.py。底层实现集中在 swarms/structs/graph_workflow.py约 3700 行对应单元测试位于 tests/structs/test_graph_workflow.py。二、核心概念与底层模型2.1 Node、Edge、NodeType 三元组在 graph_workflow.py 中图的三要素被建模为三个类Node节点id、type、agent、metadata。type取自NodeType枚举取值为AGENT普通 Agent或SUBGRAPH嵌套的 GraphWorkflow。节点 ID 优先取 Agent 的agent_name取不到时需显式指定否则抛ValueError。Edge边source、target、metadata。Edge.from_nodes支持三种入参——Node对象、带agent_name属性的 Agent 对象、或直接的字符串 ID兼容性很强。GraphBackend图后端抽象基类定义add_node/add_edge/in_degree/out_degree/predecessors/reverse/topological_generations/simple_cycles/descendants/is_dag/adjacency等统一接口让上层逻辑与具体图库解耦。2.2 双后端设计NetworkX 与 Rustworkx图后端有两种实现通过GraphWorkflow(backend...)参数选择默认networkxNetworkXBackend基于networkx.DiGraph()功能完整、生态成熟无需额外安装RustworkxBackend基于 Rust 实现的rustworkx.PyDiGraph()内部用整数索引、对外暴露字符串节点 ID。执行速度更快适合大规模图但需要pip install rustworkx未安装时会自动回退到 networkx 并输出警告graph_workflow.py#L798-L811。两个后端都做了对环的容错topological_generations()在遇到环时不会直接抛错而是先用 Kahn 算法分层可排序部分再把环内剩余节点作为最后一层追加保证编译与执行不会因环而崩溃见 graph_workflow.py#L240-L272 与 graph_workflow.py#L432-L467。此外simple_cycles()会先做 O(VE) 的无环短路判断避免在无环图上付出指数级枚举代价。2.3 GraphWorkflow 主类与关键参数GraphWorkflow.__init__graph_workflow.py#L762-L916的核心参数参数默认值说明nameGraph-Workflow-01工作流名称用于日志与可视化标题description预置描述工作流描述nodes/edgesNone可直接传入预构建的节点字典与边列表entry_points/end_points[]入口/出口节点 ID 列表编译时若为空会自动推导max_loops1整图循环执行轮数1 时每轮终点输出会回注到下一轮起点taskNone默认任务run()不传参时使用auto_compileTrue构造后自动编译拓扑分层、校验、缓存执行计划verboseFalse是否输出详细日志backendnetworkx图后端选择checkpoint_dirNone检查点目录启用断点续跑on_node_completeNone每节点完成的回调max_parallel_nodesNone最大并行节点数不设置时默认max(1, int(CPU核数 * 0.95))三、快速上手从零搭建一个链式图工作流graph_workflow_basic.py 演示了最简洁的三 Agent 链式流程核心步骤如下from swarms.structs.graph_workflow import GraphWorkflow from swarms.structs.agent import Agent agent_one Agent( agent_nameresearch_agent, model_namegpt-5.4, agent_descriptionAgent responsible for gathering and summarizing research information., ) agent_two Agent( agent_nameresearch_agent_two, model_namegpt-5.4, agent_descriptionAgent that analyzes the research data provided and processes insights., ) agent_three Agent( agent_nameresearch_agent_three, model_namegpt-5.4, agent_descriptionAgent tasked with structuring analysis into a final report or output., ) # 创建工作流 workflow GraphWorkflow(nameBasic Example, verboseTrue) # 批量添加节点也可逐个 add_node workflow.add_nodes([agent_one, agent_two, agent_three]) # 用 agent_name 建立有向依赖research_agent - research_agent_two - research_agent_three workflow.add_edge(research_agent, research_agent_two) workflow.add_edge(research_agent_two, research_agent_three) workflow.visualize() # 生成 Graphviz 可视化图 workflow.compile() # 拓扑分层 预校验 缓存执行计划 results workflow.run(Complete a simple task) print(results) # Dict[node_id - output]执行结果run()返回以节点 ID 为键、Agent 输出为值的字典如{research_agent: ..., research_agent_two: ..., research_agent_three: ...}。模型名以仓库实际可用模型为准示例中使用gpt-5.4可替换为任意已配置的模型。四、掌握建图 API链、扇出、扇入与并行全连接4.1 逐个建图与批量建图add_node(agent, **kwargs)graph_workflow.py#L1189添加一个 Agent 或嵌套 GraphWorkflow。节点 ID 自动取自agent_name若重复会抛ValueError每次结构变更会触发_invalidate_compilation()使编译缓存失效保证数据一致性。add_nodes(agents, batch_size10)graph_workflow.py#L1238批量并发添加内部按batch_size分批每批内并行提交、批次间严格同步适合几十上百个 Agent 的场景。4.2 四类边的快捷方式在单条add_edge之外框架提供了三种批量连边 API见 graph_workflow.py#L1360-L1468API模式场景add_edges_from_source(source, targets)扇出 fan-out一个 Agent 的输出并行分发到多个下游add_edges_to_target(sources, target)扇入 fan-in多个上游输出汇聚到同一 Agentadd_parallel_chain(sources, targets)全连接 mesh多源多目标的全连接并行网格示例来自源码 docstring# 扇出数据收集者的输出同时发给三位分析师 workflow.add_edges_from_source( DataCollector, [TechnicalAnalyst, FundamentalAnalyst, SentimentAnalyst], ) # 扇入三位分析师的结果汇聚到综合 Agent workflow.add_edges_to_target( [TechnicalAnalyst, FundamentalAnalyst, SentimentAnalyst], SynthesisAgent, ) # 全连接两个数据源并行喂给三个分析师 workflow.add_parallel_chain( [DataCollector1, DataCollector2], [Analyst1, Analyst2, Analyst3], )4.3 入口与出口点的自动推导入口点无入边节点与出口点无出边节点可手工用set_entry_points/set_end_points指定也可调用auto_set_entry_points()graph_workflow.py#L1661和auto_set_end_points()graph_workflow.py#L1691自动推导。当所有节点都有入边疑似成环时会输出告警日志。五、编译机制拓扑分层与执行计划缓存compile()graph_workflow.py#L939是执行前的核心预处理做了四件事自动补齐入口/出口点拓扑分层调用topological_generations()把节点划分为若干层同一层内节点彼此无依赖天然可并行结构校验通过一次遍历得到的前驱/后继邻接表检查孤立节点、不可达节点、环等问题错误只记日志不中断严格校验请用validate(raise_on_errorTrue)冻结执行计划把每层节点连同其 Agent、类型、名称固化成_execution_plan元组列表避免运行期热循环中的字典查找。编译结果会被缓存。max_loops 1时编译只执行一次、多轮循环复用同一份结构这正是test_graph_workflow_caching.py验证的行为。任何图结构变更都会触发_invalidate_compilation()主动失效缓存。六、执行模型run()的并行调度与多轮迭代6.1 分层并行调度run(task, img, on_node_complete, streaming_callback, **kwargs)graph_workflow.py#L1899采用按层并行的调度策略计算最宽层的节点数创建共享线程池max_workers min(配置值, 最宽层大小)避免在深层图上反复建池对每一层先为所有节点预构建 prompt_build_prompt再把每个 Agent 调用封装为零参数可调用对象统一提交线程池单节点失败不会拖垮整层_safe_output会把异常转换为[ERROR] Agent xxx failed: ...字符串保证下游 Agent 仍有输入可用graph_workflow.py#L1876-L1893。Prompt 组装规则_build_promptgraph_workflow.py#L1737非首层节点把每个前驱的输出作为独立user消息传入主 prompt 要求 Agent 审查前驱的工作、验证发现并在此基础上扩展多轮迭代的非首层入口节点会收到上一轮终点输出prompt 指示其审阅上一轮结果、修正或扩充首层首节点直接执行原任务。6.2 多轮循环max_loops 1当max_loops 1时整图执行多轮每轮的终点输出会作为附加上下文注入下一轮的入口节点实现初稿 → 复审 → 精修的迭代式生成。返回结构也有变化结果字典中既有{node_id}_loop_{n}形式的逐轮结果也有最终轮以纯node_id为键的结果。6.3 异步与流式arun(...)graph_workflow.py#L1827异步包装内部通过asyncio.to_thread复用同步run适合放入事件循环streaming_callback(node_id, token)逐 token 流式回调可在生成过程中实时转发对应 graph_workflow_streaming.pyon_node_complete(node_id, output)每节点完成即回调早于整层完成run()参数优先于构造时实例级回调。七、子图组合把工作流当作节点NodeType.SUBGRAPH允许把另一个GraphWorkflow作为节点嵌入父图实现模块化复用对应 graph_workflow_composition.pysubflow GraphWorkflow(nameinner-research, auto_compileTrue) subflow.add_nodes([agent_a, agent_b]) subflow.add_edge(agent_a, agent_b) outer GraphWorkflow(nameouter, auto_compileTrue) outer.add_node(agent_entry) # 普通 Agent 节点 outer.add_node(subflow) # 嵌套子图节点类型自动判定为 SUBGRAPH outer.add_edge(agent_entry.agent_name, subflow.name) outer.run(...)执行子图节点时框架会把前驱消息与当前 prompt 拼接到一起作为子图的任务传入子图内部再按自己的拓扑并行执行见 graph_workflow.py#L2116-L2158 的_make_call分支。启用父图检查点时子图会自动在父检查点目录下建立自己的子目录实现嵌套断点续跑。八、校验在运行前发现结构问题validate(auto_fixFalse, raise_on_errorFalse)graph_workflow.py#L3519返回包含is_valid、errors、warnings、fixed、cycles的字典report workflow.validate(auto_fixTrue, raise_on_errorTrue) print(report[is_valid], report[errors], report[cycles])auto_fixTrue自动补齐缺失的入口/出口点并把不可达节点追加为入口点raise_on_errorTrueis_validFalse时直接抛ValueError校验与编译共用同一套_structural_checks保证两者结论不会漂移校验模式额外枚举所有简单环编译模式为性能只用 O(VE) 的是否存在环判断。九、可视化、序列化与持久化9.1 Graphviz 可视化visualize(formatpng, viewTrue, enginedot, show_summaryFalse)graph_workflow.py#L2430把工作流渲染成图支持png/svg/pdf/dot格式布局引擎可选dot/neato/fdp/sfdp/twopi/circo自动检测扇出/扇入节点并用不同颜色样式高亮入口/出口节点也有独立样式输出文件名形如{safe_name}_visualization_{uuid}.png。示例输出见 graph_workflow_example.png 与 Graph-Workflow-01_visualization需要先pip install graphviz未安装会抛ImportError。9.2 序列化与 JSON 导出to_dict()graph_workflow.py#L2883与to_json()graph_workflow.py#L3008把工作流结构节点、边、入口/出口、参数导出为字典或 JSON 字符串便于存储、分享与 CI 断言见 graph_workflow_example.py 与 test_enhanced_json_export.pyfrom_json(json_str)graph_workflow.py#L3247与load(path)graph_workflow.py#L3380反序列化恢复工作流配合save(path)/save_spec(path)/save_to_file(path)/load_from_file(path)完成持久化闭环对应 graph_workflow_serialization.py。十、检查点续跑断点恢复与容错设置checkpoint_dir后run()会在每层完成后把该层输出写入{task_key}_layer_{layer_idx}.jsontask_key为任务的 SHA-256 前缀保证跨进程稳定。再次运行时若发现某层检查点已存在直接加载恢复并跳过该层实现故障恢复长任务中断后无需从头执行分层断点按拓扑层粒度保存恢复粒度精细子图嵌套断点子图在父目录下独立保存。参考 graph_workflow_checkpointing.py 与 v11 示例 graph_workflow_checkpoint.py。十一、Rustworkx 高性能后端与大规模场景rustworkx_examples 子目录 提供了 10 个进阶示例覆盖01_basic_usage.py/02_backend_comparison.py两后端 API 一致性与差异对比03_fan_out_fan_in_patterns.py扇出/扇入模式在 rustworkx 上的实现05_performance_benchmark.py性能基准测试验证 Rust 后端的吞吐优势06_error_handling.py/09_workflow_validation.py错误处理与校验07_large_scale_workflow.py/08_parallel_chain_example.py大规模与并行链场景10_real_world_scenario.py真实业务场景落地。启用方式只需workflow GraphWorkflow(namescaled, backendrustworkx, max_parallel_nodes32)未安装 rustworkx 时会自动回退 networkx 并告警因此生产环境可直接声明backendrustworkx而无需担心缺失依赖崩溃。rustworkx 后端用整数索引加双向映射表实现 O(1) 节点查找、原生拓扑分层与后继遍历见 graph_workflow.py#L311-L570配合max_parallel_nodes显式控制并行度适合节点规模大、扇出扇入密集的工作流。十二、完整实战一个研究 → 分析 → 综合报告三步工作流综合以上 API给出一个可直接复制运行的完整示例改编自 graph_workflow_example.pyfrom swarms import Agent, GraphWorkflow from swarms.prompts.multi_agent_collab_prompt import MULTI_AGENT_COLLAB_PROMPT_TWO # 三个协作 Agent agent1 Agent( agent_nameResearchAgent1, model_namegpt-5.4, max_loops1, system_promptMULTI_AGENT_COLLAB_PROMPT_TWO, ) agent2 Agent( agent_nameResearchAgent2, model_namegpt-5.4, max_loops1, system_promptMULTI_AGENT_COLLAB_PROMPT_TWO, ) agent3 Agent( agent_nameSynthesisAgent, model_namegpt-5.4, max_loops1, ) # 建图两个研究 Agent 并行汇聚到综合 Agent workflow GraphWorkflow( nameResearch Workflow, descriptionResearching arbitrage trading strategies for altcoins, auto_compileTrue, ) workflow.add_nodes([agent1, agent2, agent3]) workflow.add_edges_to_target( [ResearchAgent1, ResearchAgent2], SynthesisAgent, ) report workflow.validate() assert report[is_valid], report[errors] workflow.visualize() results workflow.run( taskWhat are the best arbitrage trading strategies for altcoins? Give me research papers and articles on the topic. ) for agent_name, result in results.items(): print(f{agent_name}: {result[:200]}...)要点回顾add_nodes批量注册 Agent → 用add_edges_to_target建立扇入汇聚 →validate()在运行前兜底 →run()按拓扑层并行执行两个研究 Agent 同层并行其输出作为前驱消息喂给SynthesisAgent。十三、最佳实践与注意事项ID 稳定性边关系使用agent_name字符串命名需唯一且稳定避免与Agent实例解耦后失联善用编译缓存结构不变时compile()只会执行一次max_loops多轮场景收益最大每次改图都会自动失效缓存无需手工清理先校验后执行生产环境在run()前调用validate(auto_fixTrue)把孤立节点、环、不可达节点问题暴露在运行前并行度控制max_parallel_nodes显式限制同层并发避免突发并发打爆 API 配额默认按 CPU 核数 * 0.95 计算失败隔离单 Agent 失败降级为[ERROR]字符串而非中断整图可在on_node_complete回调中感知并记录异常长任务开检查点耗时工作流设置checkpoint_dir中断后可分层续跑大数据量用 rustworkx节点规模大、扇出扇入密集时切换backendrustworkx并参考 rustworkx_examples 的基准与大规模示例评估收益可视化辅助排错visualize()生成的图中入口/出口、扇出/扇入节点均有专属样式是快速定位拓扑设计问题的有效手段。结语Graph Workflow 把多 Agent 协作从线性接力升级为有向图调度以GraphWorkflow为核心用Node/Edge建模拓扑由双后端NetworkX / Rustworkx承载图运算以拓扑分层实现层内并行再叠加多轮迭代、子图组合、检查点、流式回调与序列化能力即可编排条件路径、并行分支、汇聚与循环改进等复杂场景。建议从 graph_workflow_basic.py 起步逐步尝试 graph_workflow_fanout_fanin.py、graph_workflow_composition.py 与 rustworkx_examples 系列并在 tests/structs/test_graph_workflow.py 中查看对应行为约束。【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/17 6:33:42 · 阅读 1.2k swarms GraphWorkflow 实战指南：用图工作流编排多 Agent 非线性协作
swarms GraphWorkflow 实战指南用图工作流编排多 Agent 非线性协作【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms图工作流Graph Workflow是 swarms 框架中用于编排复杂、非线性多 Agent 交互的核心机制Agent 是图上的节点Agent 之间的关系构成有向边由此可以表达条件路径、并行分支、基于中间结果动态路由、扇出fan-out/扇入fan-in等远超线性管道的能力。本文以 examples/multi_agent/graphworkflow_examples/README.md 为骨架结合 graph_workflow.py 源码与配套示例带你从零掌握GraphWorkflow的建图、编译、执行、校验、可视化、序列化与并行化并给出可复制运行的完整代码。一、认识 Graph Workflow从线性管道到有向图编排1.1 为什么需要图工作流传统多 Agent 编排多采用顺序管道Sequential WorkflowAgent 依次执行、前一个的输出直接作为后一个的输入。这类模型结构简单但无法表达条件路径根据中间结果决定后续走向并行分支多个 Agent 同时处理互不依赖的子任务动态汇聚多个分支的输出在某个 Agent 处汇合再统一处理多轮迭代将上一轮终点结果回注到下一轮起点实现循环改进。Graph Workflow 正是为这些场景设计图中节点之间的边显式声明了数据依赖关系执行引擎依据拓扑序自动分层调度从而把谁先谁后、谁能并行交给图结构决定。1.2 目录结构与示例总览仓库中 graphworkflow_examples 目录 集中收录了图工作流示例README 列出的核心文件包括文件主题graph_workflow_basic.py最基础的链式图工作流graph_workflow_example.py完整工作流 JSON 导出advanced_graph_workflow.py高级图模式graph_workflow_validation.py工作流结构校验graph_workflow_composition.py图组合子图嵌套graph_workflow_serialization.py序列化与反序列化graph_workflow_fanout_fanin.py扇出/扇入模式graph_workflow_checkpointing.py检查点续跑graph_workflow_streaming.py流式输出test_enhanced_json_export.pyJSON 导出测试test_graph_workflow_caching.py编译缓存测试test_graphviz_visualization.pyGraphviz 可视化测试test_parallel_processing_example.py并行处理测试目录内还有两个子目录graph/核心图工具示例基础与完整示例的简版变体rustworkx_examples/基于 rustworkx 高性能后端的系列示例含性能基准、错误处理、大规模工作流等 10 个脚本与测试 test_graph_workflow_rustworkx.py。底层实现集中在 swarms/structs/graph_workflow.py约 3700 行对应单元测试位于 tests/structs/test_graph_workflow.py。二、核心概念与底层模型2.1 Node、Edge、NodeType 三元组在 graph_workflow.py 中图的三要素被建模为三个类Node节点id、type、agent、metadata。type取自NodeType枚举取值为AGENT普通 Agent或SUBGRAPH嵌套的 GraphWorkflow。节点 ID 优先取 Agent 的agent_name取不到时需显式指定否则抛ValueError。Edge边source、target、metadata。Edge.from_nodes支持三种入参——Node对象、带agent_name属性的 Agent 对象、或直接的字符串 ID兼容性很强。GraphBackend图后端抽象基类定义add_node/add_edge/in_degree/out_degree/predecessors/reverse/topological_generations/simple_cycles/descendants/is_dag/adjacency等统一接口让上层逻辑与具体图库解耦。2.2 双后端设计NetworkX 与 Rustworkx图后端有两种实现通过GraphWorkflow(backend...)参数选择默认networkxNetworkXBackend基于networkx.DiGraph()功能完整、生态成熟无需额外安装RustworkxBackend基于 Rust 实现的rustworkx.PyDiGraph()内部用整数索引、对外暴露字符串节点 ID。执行速度更快适合大规模图但需要pip install rustworkx未安装时会自动回退到 networkx 并输出警告graph_workflow.py#L798-L811。两个后端都做了对环的容错topological_generations()在遇到环时不会直接抛错而是先用 Kahn 算法分层可排序部分再把环内剩余节点作为最后一层追加保证编译与执行不会因环而崩溃见 graph_workflow.py#L240-L272 与 graph_workflow.py#L432-L467。此外simple_cycles()会先做 O(VE) 的无环短路判断避免在无环图上付出指数级枚举代价。2.3 GraphWorkflow 主类与关键参数GraphWorkflow.__init__graph_workflow.py#L762-L916的核心参数参数默认值说明nameGraph-Workflow-01工作流名称用于日志与可视化标题description预置描述工作流描述nodes/edgesNone可直接传入预构建的节点字典与边列表entry_points/end_points[]入口/出口节点 ID 列表编译时若为空会自动推导max_loops1整图循环执行轮数1 时每轮终点输出会回注到下一轮起点taskNone默认任务run()不传参时使用auto_compileTrue构造后自动编译拓扑分层、校验、缓存执行计划verboseFalse是否输出详细日志backendnetworkx图后端选择checkpoint_dirNone检查点目录启用断点续跑on_node_completeNone每节点完成的回调max_parallel_nodesNone最大并行节点数不设置时默认max(1, int(CPU核数 * 0.95))三、快速上手从零搭建一个链式图工作流graph_workflow_basic.py 演示了最简洁的三 Agent 链式流程核心步骤如下from swarms.structs.graph_workflow import GraphWorkflow from swarms.structs.agent import Agent agent_one Agent( agent_nameresearch_agent, model_namegpt-5.4, agent_descriptionAgent responsible for gathering and summarizing research information., ) agent_two Agent( agent_nameresearch_agent_two, model_namegpt-5.4, agent_descriptionAgent that analyzes the research data provided and processes insights., ) agent_three Agent( agent_nameresearch_agent_three, model_namegpt-5.4, agent_descriptionAgent tasked with structuring analysis into a final report or output., ) # 创建工作流 workflow GraphWorkflow(nameBasic Example, verboseTrue) # 批量添加节点也可逐个 add_node workflow.add_nodes([agent_one, agent_two, agent_three]) # 用 agent_name 建立有向依赖research_agent - research_agent_two - research_agent_three workflow.add_edge(research_agent, research_agent_two) workflow.add_edge(research_agent_two, research_agent_three) workflow.visualize() # 生成 Graphviz 可视化图 workflow.compile() # 拓扑分层 预校验 缓存执行计划 results workflow.run(Complete a simple task) print(results) # Dict[node_id - output]执行结果run()返回以节点 ID 为键、Agent 输出为值的字典如{research_agent: ..., research_agent_two: ..., research_agent_three: ...}。模型名以仓库实际可用模型为准示例中使用gpt-5.4可替换为任意已配置的模型。四、掌握建图 API链、扇出、扇入与并行全连接4.1 逐个建图与批量建图add_node(agent, **kwargs)graph_workflow.py#L1189添加一个 Agent 或嵌套 GraphWorkflow。节点 ID 自动取自agent_name若重复会抛ValueError每次结构变更会触发_invalidate_compilation()使编译缓存失效保证数据一致性。add_nodes(agents, batch_size10)graph_workflow.py#L1238批量并发添加内部按batch_size分批每批内并行提交、批次间严格同步适合几十上百个 Agent 的场景。4.2 四类边的快捷方式在单条add_edge之外框架提供了三种批量连边 API见 graph_workflow.py#L1360-L1468API模式场景add_edges_from_source(source, targets)扇出 fan-out一个 Agent 的输出并行分发到多个下游add_edges_to_target(sources, target)扇入 fan-in多个上游输出汇聚到同一 Agentadd_parallel_chain(sources, targets)全连接 mesh多源多目标的全连接并行网格示例来自源码 docstring# 扇出数据收集者的输出同时发给三位分析师 workflow.add_edges_from_source( DataCollector, [TechnicalAnalyst, FundamentalAnalyst, SentimentAnalyst], ) # 扇入三位分析师的结果汇聚到综合 Agent workflow.add_edges_to_target( [TechnicalAnalyst, FundamentalAnalyst, SentimentAnalyst], SynthesisAgent, ) # 全连接两个数据源并行喂给三个分析师 workflow.add_parallel_chain( [DataCollector1, DataCollector2], [Analyst1, Analyst2, Analyst3], )4.3 入口与出口点的自动推导入口点无入边节点与出口点无出边节点可手工用set_entry_points/set_end_points指定也可调用auto_set_entry_points()graph_workflow.py#L1661和auto_set_end_points()graph_workflow.py#L1691自动推导。当所有节点都有入边疑似成环时会输出告警日志。五、编译机制拓扑分层与执行计划缓存compile()graph_workflow.py#L939是执行前的核心预处理做了四件事自动补齐入口/出口点拓扑分层调用topological_generations()把节点划分为若干层同一层内节点彼此无依赖天然可并行结构校验通过一次遍历得到的前驱/后继邻接表检查孤立节点、不可达节点、环等问题错误只记日志不中断严格校验请用validate(raise_on_errorTrue)冻结执行计划把每层节点连同其 Agent、类型、名称固化成_execution_plan元组列表避免运行期热循环中的字典查找。编译结果会被缓存。max_loops 1时编译只执行一次、多轮循环复用同一份结构这正是test_graph_workflow_caching.py验证的行为。任何图结构变更都会触发_invalidate_compilation()主动失效缓存。六、执行模型run()的并行调度与多轮迭代6.1 分层并行调度run(task, img, on_node_complete, streaming_callback, **kwargs)graph_workflow.py#L1899采用按层并行的调度策略计算最宽层的节点数创建共享线程池max_workers min(配置值, 最宽层大小)避免在深层图上反复建池对每一层先为所有节点预构建 prompt_build_prompt再把每个 Agent 调用封装为零参数可调用对象统一提交线程池单节点失败不会拖垮整层_safe_output会把异常转换为[ERROR] Agent xxx failed: ...字符串保证下游 Agent 仍有输入可用graph_workflow.py#L1876-L1893。Prompt 组装规则_build_promptgraph_workflow.py#L1737非首层节点把每个前驱的输出作为独立user消息传入主 prompt 要求 Agent 审查前驱的工作、验证发现并在此基础上扩展多轮迭代的非首层入口节点会收到上一轮终点输出prompt 指示其审阅上一轮结果、修正或扩充首层首节点直接执行原任务。6.2 多轮循环max_loops 1当max_loops 1时整图执行多轮每轮的终点输出会作为附加上下文注入下一轮的入口节点实现初稿 → 复审 → 精修的迭代式生成。返回结构也有变化结果字典中既有{node_id}_loop_{n}形式的逐轮结果也有最终轮以纯node_id为键的结果。6.3 异步与流式arun(...)graph_workflow.py#L1827异步包装内部通过asyncio.to_thread复用同步run适合放入事件循环streaming_callback(node_id, token)逐 token 流式回调可在生成过程中实时转发对应 graph_workflow_streaming.pyon_node_complete(node_id, output)每节点完成即回调早于整层完成run()参数优先于构造时实例级回调。七、子图组合把工作流当作节点NodeType.SUBGRAPH允许把另一个GraphWorkflow作为节点嵌入父图实现模块化复用对应 graph_workflow_composition.pysubflow GraphWorkflow(nameinner-research, auto_compileTrue) subflow.add_nodes([agent_a, agent_b]) subflow.add_edge(agent_a, agent_b) outer GraphWorkflow(nameouter, auto_compileTrue) outer.add_node(agent_entry) # 普通 Agent 节点 outer.add_node(subflow) # 嵌套子图节点类型自动判定为 SUBGRAPH outer.add_edge(agent_entry.agent_name, subflow.name) outer.run(...)执行子图节点时框架会把前驱消息与当前 prompt 拼接到一起作为子图的任务传入子图内部再按自己的拓扑并行执行见 graph_workflow.py#L2116-L2158 的_make_call分支。启用父图检查点时子图会自动在父检查点目录下建立自己的子目录实现嵌套断点续跑。八、校验在运行前发现结构问题validate(auto_fixFalse, raise_on_errorFalse)graph_workflow.py#L3519返回包含is_valid、errors、warnings、fixed、cycles的字典report workflow.validate(auto_fixTrue, raise_on_errorTrue) print(report[is_valid], report[errors], report[cycles])auto_fixTrue自动补齐缺失的入口/出口点并把不可达节点追加为入口点raise_on_errorTrueis_validFalse时直接抛ValueError校验与编译共用同一套_structural_checks保证两者结论不会漂移校验模式额外枚举所有简单环编译模式为性能只用 O(VE) 的是否存在环判断。九、可视化、序列化与持久化9.1 Graphviz 可视化visualize(formatpng, viewTrue, enginedot, show_summaryFalse)graph_workflow.py#L2430把工作流渲染成图支持png/svg/pdf/dot格式布局引擎可选dot/neato/fdp/sfdp/twopi/circo自动检测扇出/扇入节点并用不同颜色样式高亮入口/出口节点也有独立样式输出文件名形如{safe_name}_visualization_{uuid}.png。示例输出见 graph_workflow_example.png 与 Graph-Workflow-01_visualization需要先pip install graphviz未安装会抛ImportError。9.2 序列化与 JSON 导出to_dict()graph_workflow.py#L2883与to_json()graph_workflow.py#L3008把工作流结构节点、边、入口/出口、参数导出为字典或 JSON 字符串便于存储、分享与 CI 断言见 graph_workflow_example.py 与 test_enhanced_json_export.pyfrom_json(json_str)graph_workflow.py#L3247与load(path)graph_workflow.py#L3380反序列化恢复工作流配合save(path)/save_spec(path)/save_to_file(path)/load_from_file(path)完成持久化闭环对应 graph_workflow_serialization.py。十、检查点续跑断点恢复与容错设置checkpoint_dir后run()会在每层完成后把该层输出写入{task_key}_layer_{layer_idx}.jsontask_key为任务的 SHA-256 前缀保证跨进程稳定。再次运行时若发现某层检查点已存在直接加载恢复并跳过该层实现故障恢复长任务中断后无需从头执行分层断点按拓扑层粒度保存恢复粒度精细子图嵌套断点子图在父目录下独立保存。参考 graph_workflow_checkpointing.py 与 v11 示例 graph_workflow_checkpoint.py。十一、Rustworkx 高性能后端与大规模场景rustworkx_examples 子目录 提供了 10 个进阶示例覆盖01_basic_usage.py/02_backend_comparison.py两后端 API 一致性与差异对比03_fan_out_fan_in_patterns.py扇出/扇入模式在 rustworkx 上的实现05_performance_benchmark.py性能基准测试验证 Rust 后端的吞吐优势06_error_handling.py/09_workflow_validation.py错误处理与校验07_large_scale_workflow.py/08_parallel_chain_example.py大规模与并行链场景10_real_world_scenario.py真实业务场景落地。启用方式只需workflow GraphWorkflow(namescaled, backendrustworkx, max_parallel_nodes32)未安装 rustworkx 时会自动回退 networkx 并告警因此生产环境可直接声明backendrustworkx而无需担心缺失依赖崩溃。rustworkx 后端用整数索引加双向映射表实现 O(1) 节点查找、原生拓扑分层与后继遍历见 graph_workflow.py#L311-L570配合max_parallel_nodes显式控制并行度适合节点规模大、扇出扇入密集的工作流。十二、完整实战一个研究 → 分析 → 综合报告三步工作流综合以上 API给出一个可直接复制运行的完整示例改编自 graph_workflow_example.pyfrom swarms import Agent, GraphWorkflow from swarms.prompts.multi_agent_collab_prompt import MULTI_AGENT_COLLAB_PROMPT_TWO # 三个协作 Agent agent1 Agent( agent_nameResearchAgent1, model_namegpt-5.4, max_loops1, system_promptMULTI_AGENT_COLLAB_PROMPT_TWO, ) agent2 Agent( agent_nameResearchAgent2, model_namegpt-5.4, max_loops1, system_promptMULTI_AGENT_COLLAB_PROMPT_TWO, ) agent3 Agent( agent_nameSynthesisAgent, model_namegpt-5.4, max_loops1, ) # 建图两个研究 Agent 并行汇聚到综合 Agent workflow GraphWorkflow( nameResearch Workflow, descriptionResearching arbitrage trading strategies for altcoins, auto_compileTrue, ) workflow.add_nodes([agent1, agent2, agent3]) workflow.add_edges_to_target( [ResearchAgent1, ResearchAgent2], SynthesisAgent, ) report workflow.validate() assert report[is_valid], report[errors] workflow.visualize() results workflow.run( taskWhat are the best arbitrage trading strategies for altcoins? Give me research papers and articles on the topic. ) for agent_name, result in results.items(): print(f{agent_name}: {result[:200]}...)要点回顾add_nodes批量注册 Agent → 用add_edges_to_target建立扇入汇聚 →validate()在运行前兜底 →run()按拓扑层并行执行两个研究 Agent 同层并行其输出作为前驱消息喂给SynthesisAgent。十三、最佳实践与注意事项ID 稳定性边关系使用agent_name字符串命名需唯一且稳定避免与Agent实例解耦后失联善用编译缓存结构不变时compile()只会执行一次max_loops多轮场景收益最大每次改图都会自动失效缓存无需手工清理先校验后执行生产环境在run()前调用validate(auto_fixTrue)把孤立节点、环、不可达节点问题暴露在运行前并行度控制max_parallel_nodes显式限制同层并发避免突发并发打爆 API 配额默认按 CPU 核数 * 0.95 计算失败隔离单 Agent 失败降级为[ERROR]字符串而非中断整图可在on_node_complete回调中感知并记录异常长任务开检查点耗时工作流设置checkpoint_dir中断后可分层续跑大数据量用 rustworkx节点规模大、扇出扇入密集时切换backendrustworkx并参考 rustworkx_examples 的基准与大规模示例评估收益可视化辅助排错visualize()生成的图中入口/出口、扇出/扇入节点均有专属样式是快速定位拓扑设计问题的有效手段。结语Graph Workflow 把多 Agent 协作从线性接力升级为有向图调度以GraphWorkflow为核心用Node/Edge建模拓扑由双后端NetworkX / Rustworkx承载图运算以拓扑分层实现层内并行再叠加多轮迭代、子图组合、检查点、流式回调与序列化能力即可编排条件路径、并行分支、汇聚与循环改进等复杂场景。建议从 graph_workflow_basic.py 起步逐步尝试 graph_workflow_fanout_fanin.py、graph_workflow_composition.py 与 rustworkx_examples 系列并在 tests/structs/test_graph_workflow.py 中查看对应行为约束。【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考