병렬 실행
병렬 실행 (Parallel Execution)
그래프 빌더 API는 병렬 실행을 위한 두 가지 강력한 메커니즘을 제공해요: broadcasting(브로드캐스팅) 과 mapping(매핑) 이에요.
- Broadcasting - 동일한 데이터를 여러 병렬 경로로 보내기
- Spreading - 반복 가능한(iterable) 항목을 병렬 경로로 펼치기(fan out)
둘 다 실행 그래프에 "포크(fork)"를 만들어서, 나중에 조인 노드로 동기화할 수 있어요.
출처: 문서
본문
Broadcasting
브로드캐스팅은 동일한 데이터를 여러 대상에 동시에 보내요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[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_factory=list[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(state=SimpleState())
print(sorted(result))
#> [11, 12, 13]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
세 스텝 모두 동일한 입력 값(10)을 받고 병렬로 실행돼요.
Spreading
스프레딩은 반복 가능한 요소를 펼쳐서 각 요소를 병렬로 처리해요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[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_factory=list[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(state=SimpleState())
print(sorted(result))
#> [1, 4, 9, 16, 25]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
AsyncIterable 스프레딩 (Spreading AsyncIterables)
.map() 연산은 AsyncIterable 값에서도 동작해요. 비동기 이터러블을 매핑하면 그래프가 값이 산출될 때마다 병렬 작업을 동적으로 만들어요. 이는 스트리밍 데이터나 실시간으로 생성되는 데이터를 처리할 때 특히 유용합니다:
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_type=SimpleState, output_type=list[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_factory=list[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(state=SimpleState())
print(sorted(result))
#> [3, 6, 9]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
이렇게 하면 진행형(progressive) 처리가 가능해져요. 나중 결과가 아직 생성되는 동안에도 다운스트림 스텝이 먼저 도착한 결과부터 작업을 시작할 수 있어요.
add_mapping_edge() 사용하기
편의 메서드인 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_type=SimpleState, output_type=list[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 f'Value: {ctx.inputs}'
collect = g.join(reduce_list_append, initial_factory=list[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(state=SimpleState())
print(sorted(result))
#> ['Value: 10', 'Value: 20', 'Value: 30']
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
빈 이터러블 (Empty Iterables)
빈 이터러블을 매핑할 때는 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_type=SimpleState, output_type=list[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_factory=list[int])
g.add(g.edge_from(g.start_node).to(generate_empty))
g.add_mapping_edge(generate_empty, double, downstream_join_id=collect.id)
g.add(
g.edge_from(double).to(collect),
g.edge_from(collect).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState())
print(result)
#> []
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
중첩 병렬 연산 (Nested Parallel Operations)
복잡한 병렬 패턴을 위해 브로드캐스트와 맵을 중첩할 수 있어요:
스프레드 후 브로드캐스트 (Spread then Broadcast)
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[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_factory=list[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(state=SimpleState())
print(sorted(result))
#> [11, 12, 21, 22]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
결과에는 다음이 담겨요:
- 10에서:
10+1=11그리고10+2=12 - 20에서:
20+1=21그리고20+2=22
여러 개의 순차 스프레드 (Multiple Sequential Spreads)
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[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 f'num:{ctx.inputs}'
collect = g.join(reduce_list_append, initial_factory=list[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(state=SimpleState())
print(sorted(result))
#> ['num:1', 'num:2', 'num:3', 'num:4']
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
엣지 라벨 (Edge Labels)
더 나은 문서화를 위해 병렬 엣지에 라벨을 추가할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[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 f'item-{ctx.inputs}'
collect = g.join(reduce_list_append, initial_factory=list[str])
g.add(g.edge_from(g.start_node).to(generate))
g.add_mapping_edge(
generate,
process,
pre_map_label='before map',
post_map_label='after map',
)
g.add(
g.edge_from(process).to(collect),
g.edge_from(collect).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState())
print(sorted(result))
#> ['item-1', 'item-2', 'item-3']
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
병렬 실행에서의 상태 공유 (State Sharing in Parallel Execution)
모든 병렬 작업은 동일한 그래프 상태를 공유해요. 변경(mutation)에는 조심하세요:
from dataclasses import dataclass, field
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class CounterState:
values: list[int] = field(default_factory=list)
async def main():
g = GraphBuilder(state_type=CounterState, output_type=list[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_factory=list[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(state=state)
print(f'Squared: {sorted(result)}')
#> Squared: [1, 4, 9]
print(f'Tracked: {sorted(state.values)}')
#> Tracked: [1, 2, 3]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
엣지 변환 (Edge Transformations)
.transform() 메서드를 사용해 엣지를 따라 흐르는 데이터를 인라인으로 변환할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=str)
@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 f'The 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(state=SimpleState())
print(result)
#> The answer is: 84
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
변환 함수는 현재 입력을 담은 StepContext를 받고, 상태와 의존성에 접근할 수 있어요. 이것은 다음과 같은 경우에 유용합니다:
- 호환되지 않는 스텝 간 데이터 타입 변환
- 복잡한 객체에서 특정 필드 추출
- 전체 스텝을 만들지 않고 간단한 계산 적용
- 라우팅 중 데이터 형식 조정
변환은 체인으로 연결할 수 있고 .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_type=SimpleState, output_type=list[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 f'Processed: {ctx.inputs}'
collect = g.join(reduce_list_append, initial_factory=list[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(state=SimpleState())
print(sorted(result))
#> ['Processed: 10', 'Processed: 20', 'Processed: 30']
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)