그래프 빌더 API

그래프 빌더 API (Graph Builder API)

그래프 빌더 API는 병렬 실행 그래프를 구성하기 위한 강력한 빌더 패턴을 제공해요. 기존의 BaseNode 기반 그래프 API도 여전히 사용할 수 있고(빌더 API와 상호 운용됨), 이에 대한 문서는 메인 그래프 문서에 있어요.

출처: 문서

본문

pydantic-graph의 그래프 빌더 API는 다음을 제공해요:

  • 스텝 노드(Step nodes) - 비동기 함수 실행
  • 결정 노드(Decision nodes) - 조건부 분기
  • 스프레드 연산(Spread operations) - 반복 가능한 데이터의 병렬 처리
  • 브로드캐스트 연산(Broadcast operations) - 동일한 데이터를 여러 병렬 경로로 전송
  • 조인 노드와 리듀서(Join nodes and Reducers) - 병렬 실행 결과 집계

이 API는 병렬성, 라우팅, 데이터 집계를 선언적으로 제어하고 싶은 고급 워크플로우를 위해 설계되었어요.

설치 (Installation)

그래프 빌더 API는 pydantic-graph에 포함되어 있어요:

pip install pydantic-graph

또는 pydantic-ai의 일부로:

pip install pydantic-ai

빠른 시작 (Quick Start)

시작하기 위한 간단한 예제예요:

from dataclasses import dataclass

from pydantic_graph import GraphBuilder, StepContext


@dataclass
class CounterState:
    """State for tracking a counter value."""

    value: int = 0


async def main():
    # Create a graph builder with state and output types
    g = GraphBuilder(state_type=CounterState, output_type=int)

    # Define steps using the decorator
    @g.step
    async def increment(ctx: StepContext[CounterState, None, None]) -> int:
        """Increment the counter and return its value."""
        ctx.state.value += 1
        return ctx.state.value

    @g.step
    async def double_it(ctx: StepContext[CounterState, None, int]) -> int:
        """Double the input value."""
        return ctx.inputs * 2

    # Add edges connecting the nodes
    g.add(
        g.edge_from(g.start_node).to(increment),
        g.edge_from(increment).to(double_it),
        g.edge_from(double_it).to(g.end_node),
    )

    # Build and run the graph
    graph = g.build()
    state = CounterState()
    result = await graph.run(state=state)
    print(f'Result: {result}')
    #> Result: 2
    print(f'Final state: {state.value}')
    #> Final state: 1

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

핵심 개념 (Key Concepts)

GraphBuilder

GraphBuilder는 그래프 구성의 주요 진입점이에요. 다음 타입에 대해 제네릭(generic)입니다:

  • StateT - 모든 노드가 공유하는 변경 가능한 상태 타입
  • DepsT - 노드에 주입되는 의존성 타입
  • InputT - 그래프의 초기 입력 타입
  • OutputT - 그래프의 최종 출력 타입

스텝 (Steps)

스텝은 각 노드에서 수행할 실제 작업을 정의하는 @g.step으로 데코레이트된 비동기 함수예요. 스텝은 다음에 접근할 수 있는 StepContext를 받아요:

  • ctx.state - 변경 가능한 그래프 상태
  • ctx.deps - 주입된 의존성
  • ctx.inputs - 이 스텝의 입력 데이터

엣지 (Edges)

엣지는 노드 간의 연결을 정의해요. 빌더는 엣지를 만드는 여러 방법을 제공합니다:

  • g.add() - 하나 이상의 엣지 경로 추가
  • g.add_edge() - 두 노드 사이에 단순 엣지 추가
  • g.edge_from() - 복잡한 엣지 경로 구축 시작

시작·종료 노드 (Start and End Nodes)

모든 그래프에는 다음이 있어요:

더 복잡한 예제 (A More Complex Example)

맵 연산으로 병렬 실행을 보여주는 예제예요:

from dataclasses import dataclass

from pydantic_graph import GraphBuilder, StepContext, reduce_list_append


@dataclass
class ProcessingState:
    """State for tracking processing metrics."""

    items_processed: int = 0


async def main():
    g = GraphBuilder(
        state_type=ProcessingState,
        input_type=list[int],
        output_type=list[int],
    )

    @g.step
    async def square(ctx: StepContext[ProcessingState, None, int]) -> int:
        """Square a number and track that we processed it."""
        ctx.state.items_processed += 1
        return ctx.inputs * ctx.inputs

    # Create a join to collect results
    collect_results = g.join(reduce_list_append, initial_factory=list[int])

    # Build the graph with map operation
    g.add(
        g.edge_from(g.start_node).map().to(square),
        g.edge_from(square).to(collect_results),
        g.edge_from(collect_results).to(g.end_node),
    )

    graph = g.build()
    state = ProcessingState()
    result = await graph.run(state=state, inputs=[1, 2, 3, 4, 5])

    print(f'Results: {sorted(result)}')
    #> Results: [1, 4, 9, 16, 25]
    print(f'Items processed: {state.items_processed}')
    #> Items processed: 5

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

이 예제에서:

  1. 시작 노드가 정수 리스트를 받아요
  2. .map() 연산이 각 항목을 square 스텝의 별도 병렬 실행으로 펼쳐요
  3. 모든 결과는 reduce_list_append로 다시 모아져요
  4. 조인된 결과가 종료 노드로 흘러가요

다음 단계 (Next Steps)

각 기능의 상세 문서를 살펴보세요:

  • 스텝 - 스텝 노드와 실행 컨텍스트
  • 조인 - 조인 노드와 리듀서 패턴
  • 결정 - 조건부 분기 구현
  • 병렬 실행 - broadcasting과 mapping 마스터하기

고급 실행 제어 (Advanced Execution Control)

기본적인 graph.run() 메서드 외에도, 빌더 API는 그래프 실행을 세밀하게 제어하는 방법을 제공해요.

단계별 실행 (Step-by-Step Execution)

graph.iter()로 그래프를 한 번에 한 스텝씩 실행할 수 있어요:

from dataclasses import dataclass

from pydantic_graph import GraphBuilder, StepContext


@dataclass
class CounterState:
    value: int = 0


async def main():
    g = GraphBuilder(state_type=CounterState, output_type=int)

    @g.step
    async def increment(ctx: StepContext[CounterState, None, None]) -> int:
        ctx.state.value += 1
        return ctx.state.value

    @g.step
    async def double_it(ctx: StepContext[CounterState, None, int]) -> int:
        return ctx.inputs * 2

    g.add(
        g.edge_from(g.start_node).to(increment),
        g.edge_from(increment).to(double_it),
        g.edge_from(double_it).to(g.end_node),
    )

    graph = g.build()
    state = CounterState()

    # Use iter() for step-by-step execution
    async with graph.iter(state=state) as graph_run:
        print(f'Initial state: {state.value}')
        #> Initial state: 0

        # Advance execution step by step
        async for event in graph_run:
            print(f'{state.value=} | {event=}')
            #> state.value=0 | event=[GraphTask(node_id='increment', inputs=None)]
            #> state.value=1 | event=[GraphTask(node_id='double_it', inputs=1)]
            #> state.value=1 | event=[GraphTask(node_id='__end__', inputs=2)]
            #> state.value=1 | event=EndMarker(_value=2)
            if graph_run.output is not None:
                print(f'Final output: {graph_run.output}')
                #> Final output: 2
                break

(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)

GraphRun 객체는 다음을 제공해요:

  • 비동기 반복: 실행 이벤트 순회
  • next_task 속성: 다가오는 작업 검사
  • output 속성: 그래프 완료 여부 확인 및 최종 출력 가져오기
  • next() 메서드: 선택적 값 주입으로 실행 수동 진행

그래프 시각화 (Visualizing Graphs)

graph.render()으로 그래프 구조의 Mermaid 다이어그램을 생성할 수 있어요:

from dataclasses import dataclass

from pydantic_graph import GraphBuilder, StepContext


@dataclass
class SimpleState:
    pass


g = GraphBuilder(state_type=SimpleState, output_type=str)

@g.step
async def step_a(ctx: StepContext[SimpleState, None, None]) -> int:
    return 10

@g.step
async def step_b(ctx: StepContext[SimpleState, None, int]) -> str:
    return f'Result: {ctx.inputs}'

g.add(
    g.edge_from(g.start_node).to(step_a),
    g.edge_from(step_a).to(step_b),
    g.edge_from(step_b).to(g.end_node),
)

graph = g.build()

# Generate a Mermaid diagram
mermaid_diagram = graph.render(title='My Graph', direction='LR')
print(mermaid_diagram)
"""
---
title: My Graph
---
stateDiagram-v2
  direction LR
  step_a
  step_b

  [*] --> step_a
  step_a --> step_b
  step_b --> [*]
"""

렌더링된 다이어그램은 문서, 노트북, 또는 Mermaid 문법을 지원하는 어떤 도구에서든 표시할 수 있어요.

원래 API와의 비교 (Comparison with Original API)

원래 그래프 API(메인 그래프 페이지에 문서화됨)는 BaseNode 서브클래스 기반의 클래스 지향 접근법을 사용해요. 빌더 API는 데코레이트된 함수로 된 빌더 패턴을 사용하는데, 이는 다음을 제공합니다:

장점:

  • 단순 워크플로우에 더 간결한 문법
  • map/broadcast로 병렬성에 대한 명시적 제어
  • 흔한 집계 패턴을 위한 네이티브 리듀서
  • 복잡한 데이터 흐름을 더 쉽게 시각화

트레이드오프:

  • 빌더 패턴에 대한 이해 필요
  • 덜 객체지향적이고, 더 함수형 스타일

두 API 모두 완전히 지원되며, 필요하면 함께 통합할 수도 있어요.

영속성과 재개 가능성 (Persistence and Resumability)

그래프 상태는 스냅샷되지 않아요

그래프 빌더 API도 원래 Graph API도 그래프 상태를 스냅샷하지 않아요. pydantic_graph.persistence는 병렬 실행에서 일관된 스냅샷을 만드는 것의 복잡성 때문에 V2에서 제거됐어요. 에이전트 실행 상태를 저장·재개·포크하려면 Storage에 설명된 Harness의 StepPersistence 기능을 사용하세요. 크래시와 재시작을 넘어 전체 실행을 유지하려면 영속 실행(durable execution) 엔진을 사용해요.

더 알아보기 (Learn more)

  • 스텝 — 스텝 노드와 실행 컨텍스트.
  • 조인 — 조인 노드와 리듀서 패턴.
  • 결정 — 조건부 분기.
  • 병렬 실행 — broadcasting과 mapping.