조인과 리듀서
조인과 리듀서 (Joins and Reducers)
조인 노드(join node)는 병렬 실행 경로에서 오는 데이터를 동기화하고 집계해요. 여러 입력을 단일 출력으로 합치기 위해 리듀서(Reducer) 를 사용합니다.
출처: 문서
본문
병렬 실행(broadcasting 또는 mapping)을 쓰면 결과를 모으고 합칠 필요가 자주 생겨요. 조인 노드는 다음 일을 하며 그 역할을 채워요:
- 모든 병렬 작업이 끝나기를 기다림
ReducerFunction으로 출력을 집계- 집계된 결과를 다음 노드로 전달
조인 만들기 (Creating Joins)
리듀서 함수와 초기 값(또는 팩토리)을 GraphBuilder.join에 넘겨 조인을 만듭니다:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class SimpleState:
pass
g = GraphBuilder(state_type=SimpleState, output_type=list[int])
@g.step
async def generate_numbers(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
# Create a join to collect all squared values
collect = g.join(reduce_list_append, initial_factory=list[int])
g.add(
g.edge_from(g.start_node).to(generate_numbers),
g.edge_from(generate_numbers).map().to(square),
g.edge_from(square).to(collect),
g.edge_from(collect).to(g.end_node),
)
graph = g.build()
async def main():
result = await graph.run(state=SimpleState())
print(sorted(result))
#> [1, 4, 9, 16, 25]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
내장 리듀서 (Built-in Reducers)
Pydantic Graph는 몇 가지 일반적인 리듀서 타입을 기본 제공해요:
reduce_list_append
reduce_list_append는 모든 입력을 리스트로 모아요:
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 [10, 20, 30]
@g.step
async def to_string(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),
g.edge_from(generate).map().to(to_string),
g.edge_from(to_string).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())을 추가하세요. 다른 변경은 필요 없어요.)
reduce_list_extend
reduce_list_extend는 반복 가능한(iterable) 항목으로 리스트를 확장해요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_list_extend
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=list[int])
@g.step
async def generate(ctx: StepContext[SimpleState, None, None]) -> list[int]:
return [1, 2, 3]
@g.step
async def create_range(ctx: StepContext[SimpleState, None, int]) -> list[int]:
"""Create a range from 0 to the input value."""
return list(range(ctx.inputs))
collect = g.join(reduce_list_extend, initial_factory=list[int])
g.add(
g.edge_from(g.start_node).to(generate),
g.edge_from(generate).map().to(create_range),
g.edge_from(create_range).to(collect),
g.edge_from(collect).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState())
print(sorted(result))
#> [0, 0, 0, 1, 1, 2]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
reduce_dict_update
reduce_dict_update는 딕셔너리들을 병합해요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_dict_update
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=dict[str, int])
@g.step
async def generate_keys(ctx: StepContext[SimpleState, None, None]) -> list[str]:
return ['apple', 'banana', 'cherry']
@g.step
async def create_entry(ctx: StepContext[SimpleState, None, str]) -> dict[str, int]:
return {ctx.inputs: len(ctx.inputs)}
merge = g.join(reduce_dict_update, initial_factory=dict[str, int])
g.add(
g.edge_from(g.start_node).to(generate_keys),
g.edge_from(generate_keys).map().to(create_entry),
g.edge_from(create_entry).to(merge),
g.edge_from(merge).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState())
result = {k: result[k] for k in sorted(result)} # force deterministic ordering
print(result)
#> {'apple': 5, 'banana': 6, 'cherry': 6}
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
reduce_null
reduce_null은 모든 입력을 버리고 None을 반환해요. 부수 효과(side effect)만 신경 쓸 때 유용합니다:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_null
@dataclass
class CounterState:
total: int = 0
async def main():
g = GraphBuilder(state_type=CounterState, output_type=int)
@g.step
async def generate(ctx: StepContext[CounterState, None, None]) -> list[int]:
return [1, 2, 3, 4, 5]
@g.step
async def accumulate(ctx: StepContext[CounterState, None, int]) -> int:
ctx.state.total += ctx.inputs
return ctx.inputs
# We don't care about the outputs, only the side effect on state
ignore = g.join(reduce_null, initial=None)
@g.step
async def get_total(ctx: StepContext[CounterState, None, None]) -> int:
return ctx.state.total
g.add(
g.edge_from(g.start_node).to(generate),
g.edge_from(generate).map().to(accumulate),
g.edge_from(accumulate).to(ignore),
g.edge_from(ignore).to(get_total),
g.edge_from(get_total).to(g.end_node),
)
graph = g.build()
state = CounterState()
result = await graph.run(state=state)
print(result)
#> 15
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
reduce_sum
reduce_sum은 숫자 값을 합산해요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, reduce_sum
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=int)
@g.step
async def generate(ctx: StepContext[SimpleState, None, None]) -> list[int]:
return [10, 20, 30, 40]
@g.step
async def identity(ctx: StepContext[SimpleState, None, int]) -> int:
return ctx.inputs
sum_join = g.join(reduce_sum, initial=0)
g.add(
g.edge_from(g.start_node).to(generate),
g.edge_from(generate).map().to(identity),
g.edge_from(identity).to(sum_join),
g.edge_from(sum_join).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState())
print(result)
#> 100
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
ReduceFirstValue
ReduceFirstValue는 받은 첫 번째 값을 반환하고 다른 모든 병렬 작업을 취소해요. 첫 번째 성공 결과를 원하는 "경주(race)" 시나리오에 유용합니다:
import asyncio
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, ReduceFirstValue, StepContext
@dataclass
class SimpleState:
tasks_completed: int = 0
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=str)
@g.step
async def generate(ctx: StepContext[SimpleState, None, None]) -> list[int]:
return [1, 12, 13, 14, 15]
@g.step
async def slow_process(ctx: StepContext[SimpleState, None, int]) -> str:
"""Simulate variable processing times."""
# Simulate different delays
await asyncio.sleep(ctx.inputs * 0.1)
ctx.state.tasks_completed += 1
return f'Result from task {ctx.inputs}'
# Use ReduceFirstValue to get the first result and cancel the rest
first_result = g.join(ReduceFirstValue[str](), initial=None, node_id='first_result')
g.add(
g.edge_from(g.start_node).to(generate),
g.edge_from(generate).map().to(slow_process),
g.edge_from(slow_process).to(first_result),
g.edge_from(first_result).to(g.end_node),
)
graph = g.build()
state = SimpleState()
result = await graph.run(state=state)
print(result)
#> Result from task 1
print(f'Tasks completed: {state.tasks_completed}')
#> Tasks completed: ...
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
커스텀 리듀서 (Custom Reducers)
ReducerFunction을 정의해 커스텀 리듀서를 만들 수 있어요:
from pydantic_graph import GraphBuilder, StepContext
def reduce_sum(current: int, inputs: int) -> int:
"""A reducer that sums numbers."""
return current + inputs
async def main():
g = GraphBuilder(output_type=int)
@g.step
async def generate(ctx: StepContext[None, None, None]) -> list[int]:
return [5, 10, 15, 20]
@g.step
async def identity(ctx: StepContext[None, None, int]) -> int:
return ctx.inputs
sum_join = g.join(reduce_sum, initial=0)
g.add(
g.edge_from(g.start_node).to(generate),
g.edge_from(generate).map().to(identity),
g.edge_from(identity).to(sum_join),
g.edge_from(sum_join).to(g.end_node),
)
graph = g.build()
result = await graph.run()
print(result)
#> 50
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
상태에 접근하는 리듀서 (Reducers with State Access)
리듀서는 그래프 상태에 접근하고 수정할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, ReducerContext, StepContext
@dataclass
class MetricsState:
total_count: int = 0
total_sum: int = 0
@dataclass
class ReducedMetrics:
count: int = 0
sum: int = 0
def reduce_metrics_sum(ctx: ReducerContext[MetricsState, None], current: ReducedMetrics, inputs: int) -> ReducedMetrics:
ctx.state.total_count += 1
ctx.state.total_sum += inputs
return ReducedMetrics(count=current.count + 1, sum=current.sum + inputs)
def reduce_metrics_max(current: ReducedMetrics, inputs: ReducedMetrics) -> ReducedMetrics:
return ReducedMetrics(count=max(current.count, inputs.count), sum=max(current.sum, inputs.sum))
async def main():
g = GraphBuilder(state_type=MetricsState, output_type=dict[str, int])
@g.step
async def generate(ctx: StepContext[object, None, None]) -> list[int]:
return [1, 3, 5, 7, 9, 10, 20, 30, 40]
@g.step
async def process_even(ctx: StepContext[MetricsState, None, int]) -> int:
return ctx.inputs * 2
@g.step
async def process_odd(ctx: StepContext[MetricsState, None, int]) -> int:
return ctx.inputs * 3
metrics_even = g.join(reduce_metrics_sum, initial_factory=ReducedMetrics, node_id='metrics_even')
metrics_odd = g.join(reduce_metrics_sum, initial_factory=ReducedMetrics, node_id='metrics_odd')
metrics_max = g.join(reduce_metrics_max, initial_factory=ReducedMetrics, node_id='metrics_max')
g.add(
g.edge_from(g.start_node).to(generate),
# Send even and odd numbers to their respective `process` steps
g.edge_from(generate).map().to(
g.decision()
.branch(g.match(int, matches=lambda x: x % 2 == 0).label('even').to(process_even))
.branch(g.match(int, matches=lambda x: x % 2 == 1).label('odd').to(process_odd))
),
# Reduce metrics for even and odd numbers separately
g.edge_from(process_even).to(metrics_even),
g.edge_from(process_odd).to(metrics_odd),
# Aggregate the max values for each field
g.edge_from(metrics_even).to(metrics_max),
g.edge_from(metrics_odd).to(metrics_max),
# Finish the graph run with the final reduced value
g.edge_from(metrics_max).to(g.end_node),
)
graph = g.build()
state = MetricsState()
result = await graph.run(state=state)
print(f'Result: {result}')
#> Result: ReducedMetrics(count=5, sum=200)
print(f'State total_count: {state.total_count}')
#> State total_count: 9
print(f'State total_sum: {state.total_sum}')
#> State total_sum: 275
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
형제 작업 취소하기 (Canceling Sibling Tasks)
ReducerContext에 접근할 수 있는 리듀서는 ctx.cancel_sibling_tasks()를 호출해 같은 포크(fork)의 다른 모든 병렬 작업을 취소할 수 있어요. 원하는 것을 찾았을 때 조기 종료하려면 유용합니다:
import asyncio
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, ReducerContext, StepContext
@dataclass
class SearchState:
searches_completed: int = 0
def reduce_find_match(ctx: ReducerContext[SearchState, None], current: str | None, inputs: str) -> str | None:
"""Return the first input that contains 'target' and cancel remaining tasks."""
if current is not None:
# We already found a match, ignore subsequent inputs
return current
if 'target' in inputs:
# Found a match! Cancel all other parallel tasks
ctx.cancel_sibling_tasks()
return inputs
return None
async def main():
g = GraphBuilder(state_type=SearchState, output_type=str | None)
@g.step
async def generate_searches(ctx: StepContext[SearchState, None, None]) -> list[str]:
return ['item1', 'item2', 'target_item', 'item4', 'item5']
@g.step
async def search(ctx: StepContext[SearchState, None, str]) -> str:
"""Simulate a search that never finishes for 'item4' and 'item5'."""
if ctx.inputs in {'item4', 'item5'}:
# These searches only ever end by being canceled.
await asyncio.Event().wait()
ctx.state.searches_completed += 1
return ctx.inputs
find_match = g.join(reduce_find_match, initial=None)
g.add(
g.edge_from(g.start_node).to(generate_searches),
g.edge_from(generate_searches).map().to(search),
g.edge_from(search).to(find_match),
g.edge_from(find_match).to(g.end_node),
)
graph = g.build()
state = SearchState()
result = await graph.run(state=state)
print(f'Found: {result}')
#> Found: target_item
print(f'Searches completed: {state.searches_completed}')
#> Searches completed: 3
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
5개 모두가 아니라 3개만 검색이 완료된 것에 주목하세요. 리듀서가 매칭을 찾은 뒤 나머지 작업을 취소했기 때문이에요.
여러 조인 (Multiple Joins)
그래프는 여러 개의 독립적인 조인을 가질 수 있어요:
from dataclasses import dataclass, field
from pydantic_graph import GraphBuilder, StepContext, reduce_list_append
@dataclass
class MultiState:
results: dict[str, list[int]] = field(default_factory=dict)
async def main():
g = GraphBuilder(state_type=MultiState, output_type=dict[str, list[int]])
@g.step
async def source_a(ctx: StepContext[MultiState, None, None]) -> list[int]:
return [1, 2, 3]
@g.step
async def source_b(ctx: StepContext[MultiState, None, None]) -> list[int]:
return [10, 20]
@g.step
async def process_a(ctx: StepContext[MultiState, None, int]) -> int:
return ctx.inputs * 2
@g.step
async def process_b(ctx: StepContext[MultiState, None, int]) -> int:
return ctx.inputs * 3
join_a = g.join(reduce_list_append, initial_factory=list[int], node_id='join_a')
join_b = g.join(reduce_list_append, initial_factory=list[int], node_id='join_b')
@g.step
async def store_a(ctx: StepContext[MultiState, None, list[int]]) -> None:
ctx.state.results['a'] = ctx.inputs
@g.step
async def store_b(ctx: StepContext[MultiState, None, list[int]]) -> None:
ctx.state.results['b'] = ctx.inputs
@g.step
async def combine(ctx: StepContext[MultiState, None, None]) -> dict[str, list[int]]:
return ctx.state.results
g.add(
g.edge_from(g.start_node).to(source_a, source_b),
g.edge_from(source_a).map().to(process_a),
g.edge_from(source_b).map().to(process_b),
g.edge_from(process_a).to(join_a),
g.edge_from(process_b).to(join_b),
g.edge_from(join_a).to(store_a),
g.edge_from(join_b).to(store_b),
g.edge_from(store_a, store_b).to(combine),
g.edge_from(combine).to(g.end_node),
)
graph = g.build()
state = MultiState()
result = await graph.run(state=state)
print(f"Group A: {sorted(result['a'])}")
#> Group A: [2, 4, 6]
print(f"Group B: {sorted(result['b'])}")
#> Group B: [30, 60]
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
조인 노드 커스터마이징 (Customizing Join Nodes)
커스텀 노드 ID (Custom Node IDs)
스텝과 마찬가지로 조인에도 커스텀 ID를 지정할 수 있어요:
from pydantic_graph import reduce_list_append
from basic_join import g
my_join = g.join(reduce_list_append, initial_factory=list[int], node_id='my_custom_join_id')
조인이 동작하는 방식 (How Joins Work)
내부적으로 그래프는 각 병렬 작업이 어느 "포크(fork)"에 속하는지 추적해요. 조인은:
- 자신의 부모 포크(병렬 경로를 만든 포크)를 식별
- 그 포크의 모든 작업이 조인에 도달할 때까지 대기
- 들어오는 각 값에 대해
reduce()호출 - 모든 값을 받으면
finalize()호출 - 최종 결과를 다운스트림 노드로 전달
이렇게 해서 중첩된 병렬 연산에서도 올바른 동기화를 보장해요.