pydantic-graph 并行执行完全指南Broadcasting 与 Spreading 从入门到源码级原理【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-aipydantic-graphGraph Builder API为图工作流提供了两种强大的并行执行机制Broadcasting广播与Spreading展开/映射二者都会在执行图中创建分叉fork并可通过 join 节点 重新汇聚同步。本文以 并行执行官方文档 为主体结合pydantic_graph仓库源码graph_builder.py、paths.py与测试用例系统讲解如何在一次图运行中让多个节点并行处理数据并深入剖析 map、transform、空迭代器处理等边界行为的底层实现。读完本文你将能够熟练编排并行-汇聚的数据流图并理解其背后的 fork/join 执行模型。并行执行总览在 Graph Builder API 中并行执行有两种形态核心区别在于分发的数据是什么机制英文名行为对应 API广播Broadcasting将同一份数据同时发送到多个并行路径g.edge_from(source).to(a, b, c)展开Spreading将可迭代对象中的每个元素分别送往并行路径g.edge_from(source).map().to(step)两种机制都会在执行图中创建 fork分叉随后可以用 join 节点 同步。在源码中fork 由 node.py 中的Fork类表示其中is_map: bool字段区分它是广播分叉还是 map 分叉paths.py 中的BroadcastMarker第 86-98 行与MapMarker第 72-83 行则分别记录这两类分叉在路径上的标记。Broadcasting同份数据多路并行广播将相同的输入值同时送入多个目标节点各节点互不干扰地并行执行。来看官方文档的完整示例from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[int]) g.step async def source(ctx: StepContext[SimpleState, None, None]) - int: return 10 g.step async def add_one(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs 1 g.step async def add_two(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs 2 g.step async def add_three(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs 3 collect g.join(reduce_list_append, initial_factorylist[int]) # Broadcasting: send the value from source to all three steps g.add( g.edge_from(g.start_node).to(source), g.edge_from(source).to(add_one, add_two, add_three), g.edge_from(add_one, add_two, add_three).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [11, 12, 13]运行该示例时需先导入asyncio并加上asyncio.run(main())其余无需改动。关键点在g.edge_from(source).to(add_one, add_two, add_three)这一行to()接受多个目标节点时source的输出值10会被原样拷贝到三条路径add_one、add_two、add_three三个步骤同时拿到10并各自执行最终经collect汇聚得到[11, 12, 13]。从源码看这种多目标边会被构建为一条带BroadcastMarker的路径。运行时 graph_builder.py 的_handle_fork_edges()第 1033-1082 行会为每条路径创建一个携带ForkStackItem(node.id, node_run_id, i)的任务其中i是路径序号用于区分来自同一分叉的不同并行任务。汇聚时 join 节点正是依据这些 fork 栈信息判断哪些任务属于同一个分叉并等待其全部完成。Spreading可迭代对象逐元素并行展开map会把一个可迭代对象摊开让每个元素获得一次独立的并行执行。这是典型的 Map-Reduce 模式from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[int]) g.step async def generate_list(ctx: StepContext[SimpleState, None, None]) - list[int]: return [1, 2, 3, 4, 5] g.step async def square(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs * ctx.inputs collect g.join(reduce_list_append, initial_factorylist[int]) # Spreading: each item in the list gets its own parallel execution g.add( g.edge_from(g.start_node).to(generate_list), g.edge_from(generate_list).map().to(square), g.edge_from(square).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [1, 4, 9, 16, 25].map()只能作用于输出为可迭代类型Iterable[T]或AsyncIterable[T]的路径类型系统会在构建期强制这一约束——见 paths.py 第 231-253 行中map()方法的签名约束。执行时_handle_fork_edges()会遍历可迭代对象为每个元素input_item单独调用_handle_path()并打上带thread_index的 fork 栈标记从而为[1, 2, 3, 4, 5]创建 5 个并行任务每个任务分别计算平方。展开异步可迭代对象AsyncIterable.map()同样支持AsyncIterable。当映射异步可迭代对象时图会在值逐个产出yield的过程中动态创建并行任务非常适合流式数据或边生成边处理的场景import asyncio from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[int]) g.stream async def stream_numbers(ctx: StepContext[SimpleState, None, None]): Stream numbers with delays to simulate real-time data. for i in range(1, 4): await asyncio.sleep(0.05) # Simulate delay yield i g.step async def triple(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs * 3 collect g.join(reduce_list_append, initial_factorylist[int]) g.add( g.edge_from(g.start_node).to(stream_numbers), # Map over the async iterable - tasks created as items are yielded g.edge_from(stream_numbers).map().to(triple), g.edge_from(triple).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [3, 6, 9]这里用g.stream装饰器定义了一个流式步骤见 steps.md 中关于 streaming steps 的说明。_handle_fork_edges()检测到输入是AsyncIterable后会返回一个_GraphTaskAsyncIterable其内部handle_async_iterable()协程在async for循环中逐个 yield 元素并即时为每个元素派发任务graph_builder.py 第 1064-1075 行。这样下游步骤可以在早期结果尚未全部产出时就开始处理实现真正的流水式progressive并行处理。便捷方法add_mapping_edge()如果只需要源节点 → map → 目标节点的简单展开可以用 [add_mapping_edge()][pydantic_graph.graph_builder.GraphBuilder.add_mapping_edge] 一步完成from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[str]) g.step async def generate_numbers(ctx: StepContext[SimpleState, None, None]) - list[int]: return [10, 20, 30] g.step async def stringify(ctx: StepContext[SimpleState, None, int]) - str: return fValue: {ctx.inputs} collect g.join(reduce_list_append, initial_factorylist[str]) g.add(g.edge_from(g.start_node).to(generate_numbers)) g.add_mapping_edge(generate_numbers, stringify) g.add( g.edge_from(stringify).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [Value: 10, Value: 20, Value: 30]add_mapping_edge()的实现位于 graph_builder.py 第 1487-1514 行其完整签名支持四个可选参数def add_mapping_edge( self, source: Source[Iterable[T]], map_to: Destination[T], *, pre_map_label: str | None None, # map 之前的边标签 post_map_label: str | None None, # map 之后的边标签 fork_id: ForkID | None None, # 为本次 map 生成的分叉节点指定 ID downstream_join_id: JoinID | None None, # 下游 join 节点 ID用于空迭代器场景 ) - None:它内部等价于edge_from(source).label(pre_map_label).map(...).label(post_map_label).to(map_to)是链式写法的一个便捷封装。空迭代器的处理downstream_join_id当 map 的输入是空迭代器时join 节点默认不会收到任何值也就永远不会触发。此时可以指定downstream_join_id让 join 仍然执行并以初始值收尾from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[int]) g.step async def generate_empty(ctx: StepContext[SimpleState, None, None]) - list[int]: return [] g.step async def double(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs * 2 collect g.join(reduce_list_append, initial_factorylist[int]) g.add(g.edge_from(g.start_node).to(generate_empty)) g.add_mapping_edge(generate_empty, double, downstream_join_idcollect.id) g.add( g.edge_from(double).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(result) # []这里的downstream_join_idcollect.id是关键。它的底层原理在_handle_fork_edges()中graph_builder.py 第 1044-1055 行当 map 指定了下游 join id 时运行时会在派发任务之前主动为该 join 创建 reducer 状态self.active_reducers.setdefault(...)初始值来自initial_factory()。这样即便 map 的输入为空、没有任何元素到达 joinjoin 也会用初始值[]正常触发并继续向下游流动。对应测试见 test_broadcast_and_spread.py 中的test_map_empty_list与test_map_non_empty_list_with_downstream_join_id。嵌套并行组合广播与展开广播与展开可以任意嵌套组合构建复杂的并行拓扑。先展开再广播from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[int]) g.step async def generate_list(ctx: StepContext[SimpleState, None, None]) - list[int]: return [10, 20] g.step async def add_one(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs 1 g.step async def add_two(ctx: StepContext[SimpleState, None, int]) - int: return ctx.inputs 2 collect g.join(reduce_list_append, initial_factorylist[int]) g.add( g.edge_from(g.start_node).to(generate_list), # Spread the list, then broadcast each item to both steps g.edge_from(generate_list).map().to(add_one, add_two), g.edge_from(add_one, add_two).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [11, 12, 21, 22]注意g.edge_from(generate_list).map().to(add_one, add_two)同时结合了 map 与广播[10, 20]先被展开成两个并行任务每个元素又被广播到add_one和add_two。结果包含 4 项来自1010111与10212来自2020121与20222多重顺序展开from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[str]) g.step async def generate_pairs(ctx: StepContext[SimpleState, None, None]) - list[tuple[int, int]]: return [(1, 2), (3, 4)] g.step async def unpack_pair(ctx: StepContext[SimpleState, None, tuple[int, int]]) - list[int]: return [ctx.inputs[0], ctx.inputs[1]] g.step async def stringify(ctx: StepContext[SimpleState, None, int]) - str: return fnum:{ctx.inputs} collect g.join(reduce_list_append, initial_factorylist[str]) g.add( g.edge_from(g.start_node).to(generate_pairs), # First map: one task per tuple g.edge_from(generate_pairs).map().to(unpack_pair), # Second map: one task per number in each tuple g.edge_from(unpack_pair).map().to(stringify), g.edge_from(stringify).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [num:1, num:2, num:3, num:4]两次.map()串联第一次把[(1, 2), (3, 4)]展开为两个元组任务unpack_pair再把每个元组拆成list[int]第二次 map 继续把[1, 2]、[3, 4]分别展开最终产生 4 个字符串。这说明 map 的粒度完全由每一步返回的可迭代类型决定可以逐层细化数据。test_multiple_sequential_maps与test_map_then_broadcast分别在 test_broadcast_and_spread.py 中验证了这两种嵌套模式。为并行边添加标签并行边支持添加标签便于文档化与后续的 Mermaid 图渲染graph.render()见 index.mdfrom dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[str]) g.step async def generate(ctx: StepContext[SimpleState, None, None]) - list[int]: return [1, 2, 3] g.step async def process(ctx: StepContext[SimpleState, None, int]) - str: return fitem-{ctx.inputs} collect g.join(reduce_list_append, initial_factorylist[str]) g.add(g.edge_from(g.start_node).to(generate)) g.add_mapping_edge( generate, process, pre_map_labelbefore map, post_map_labelafter map, ) g.add( g.edge_from(process).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [item-1, item-2, item-3]标签在源码中由LabelMarker表示paths.py 第 101-109 行它不影响执行语义只用于调试、可视化和文档目的。除了add_mapping_edge的pre_map_label/post_map_label参数链式 API 中也可直接用.label(...)。并行执行中的状态共享所有并行任务共享同一个图状态state。这意味着并行任务对ctx.state的写入是并发且可见的必须谨慎对待状态变更from dataclasses import dataclass, field from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class CounterState: values: list[int] field(default_factorylist) async def main(): g GraphBuilder(state_typeCounterState, output_typelist[int]) g.step async def generate(ctx: StepContext[CounterState, None, None]) - list[int]: return [1, 2, 3] g.step async def track_and_square(ctx: StepContext[CounterState, None, int]) - int: # All parallel tasks mutate the same state ctx.state.values.append(ctx.inputs) return ctx.inputs * ctx.inputs collect g.join(reduce_list_append, initial_factorylist[int]) g.add( g.edge_from(g.start_node).to(generate), g.edge_from(generate).map().to(track_and_square), g.edge_from(track_and_square).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() state CounterState() result await graph.run(statestate) print(fSquared: {sorted(result)}) # Squared: [1, 4, 9] print(fTracked: {sorted(state.values)}) # Tracked: [1, 2, 3]由于三个并行任务共享同一个CounterState最终state.values记录了全部三个输入[1, 2, 3]。注意由于任务是并发执行的values的写入顺序并不保证本文示例输出经过sorted()处理以消除不确定性。如果业务依赖状态的一致性例如计数器累加请自行使用锁或幂等设计或将聚合逻辑下沉到 join 的 reducer 中reducer 可访问ReducerContext并安全地读写 state见 joins.md 中的状态型 reducer 示例。边变换Edge Transformations.transform()除了用步骤节点处理数据还可以在数据沿边流动时用.transform()内联转换数据无需为此创建一个完整步骤from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typestr) g.step async def generate_number(ctx: StepContext[SimpleState, None, None]) - int: return 42 g.step async def format_output(ctx: StepContext[SimpleState, None, str]) - str: return fThe answer is: {ctx.inputs} # Transform the number to a string inline g.add( g.edge_from(g.start_node).to(generate_number), g.edge_from(generate_number).transform(lambda ctx: str(ctx.inputs * 2)).to(format_output), g.edge_from(format_output).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(result) # The answer is: 84transform 函数接收一个带当前输入数据的 [StepContext][pydantic_graph.step.StepContext]可通过ctx.state、ctx.deps访问状态与依赖适合以下场景在不兼容类型的步骤之间做类型转换如上例int→str从复杂对象中提取特定字段应用简单计算而无需新建完整步骤在路由routing过程中适配数据格式在源码中transform 由TransformMarker表示paths.py 第 61-69 行transform()方法第 219-229 行将其追加到路径的工作项列表中。变换的链式组合.transform()可以与.map()、.label()等其他边操作自由链式组合from dataclasses import dataclass from pydantic_graph import GraphBuilder, StepContext, reduce_list_append dataclass class SimpleState: pass async def main(): g GraphBuilder(state_typeSimpleState, output_typelist[str]) g.step async def generate_data(ctx: StepContext[SimpleState, None, None]) - list[dict[str, int]]: return [{value: 10}, {value: 20}, {value: 30}] g.step async def process_number(ctx: StepContext[SimpleState, None, int]) - str: return fProcessed: {ctx.inputs} collect g.join(reduce_list_append, initial_factorylist[str]) g.add( g.edge_from(g.start_node).to(generate_data), # Transform to extract values, then map over them g.edge_from(generate_data) .transform(lambda ctx: [item[value] for item in ctx.inputs]) .label(Extract values) .map() .to(process_number), g.edge_from(process_number).to(collect), g.edge_from(collect).to(g.end_node), ) graph g.build() result await graph.run(stateSimpleState()) print(sorted(result)) # [Processed: 10, Processed: 20, Processed: 30]这里的调用顺序是transform从dict列表中提取value字段组成新列表→label打上Extract values标签→map展开新列表→to送往process_number。注意类型约束.map()要求当前路径的输出类型是迭代器因此transform必须先把数据转换成list[int].map()才能合法展开。底层原理fork 与 join 的同步机制结合 joins.md 中join 如何工作的说明与 graph_builder.py 的运行时实现可以还原并行执行的完整链路分叉Fork执行到带广播/展开语义的路径时_handle_fork_edges()第 1033 行为每条并行路径创建独立GraphTask并在其fork_stack中压入ForkStackItem(fork_id, node_run_id, index)用于标识这份数据来自哪个分叉的第几条路径。任务执行每个任务携带各自的输入广播是同一份拷贝map 是单个元素并发执行步骤函数并行性基于anyio的任务组create_task_group实现。汇聚JoinJoinItem携带join_id与fork_stack流向 join 节点。运行时依据 fork 栈信息解析父分叉_resolve_join_fork_run、parent_forks.py等待该分叉内所有任务到达后对每个输入依次调用 reducer 的reduce()最后调用finalize()生成汇总结果并流向后续节点。空迭代器兜底若指定了downstream_join_idjoin 状态会在派发任务前被主动创建第 1048-1055 行保证空输入时 join 仍以初始值触发。这一机制保证了即使在嵌套并行map 内套广播、多级 map的场景下join 也能正确识别所属分叉并完成同步。测试验证仓库为并行执行提供了系统的测试覆盖可作为理解与验证行为的第一手资料test_broadcast_and_spread.pytest_broadcast_to_multiple_steps多目标广播、test_map_over_list列表展开、test_map_empty_list/test_map_non_empty_list_with_downstream_join_id空迭代器与downstream_join_id、test_nested_broadcasts、test_map_then_broadcast、test_multiple_sequential_maps、test_broadcast_with_different_outputs不同输出类型的广播等test_edge_labels.py验证边的标签含pre_map_label/post_map_label行为test_joins_and_reducers.py覆盖 join 与各 reducer 的汇聚行为test_graph_iteration.py 与 test_graph_execution.py验证并行场景下的逐步执行与整体执行。下一步学习 join 节点与 reducer掌握reduce_list_append、reduce_sum、ReduceFirstValue等内置 reducer 及cancel_sibling_tasks等进阶用法探索 条件分支在并行路径中结合 decision 节点做按值路由查阅 steps 文档深入了解g.step、g.stream与StepContext的完整能力若想从整体上把握 Graph Builder API 的安装、快速上手与执行控制可先阅读 Graph Builder API 总览。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考