그래프

그래프 (Graphs)

필요하지 않은데 못 박는 총을 쓰지 마세요

Pydantic AI 에이전트가 망치이고, 멀티 에이전트 워크플로우가 큰 망치라면, 그래프는 못 박는 총(네일 건)이에요:

• 맞아요, 못 박는 총이 망치보다 멋져 보이죠
• 하지만 못 박는 총은 망치보다 준비할 게 훨씬 많아요
• 그리고 못 박는 총이 당신을 더 나은 건축가로 만들어 주지 않아요. 네일 건을 든 건축가로 만들 뿐이죠
• 마지막으로 (이 비유를 애써 끌고 가는 감이 있지만) 망치 같은 중세 도구와 타입 없는 파이썬을 좋아한다면, 네일 건이나 우리의 그래프 접근법을 좋아하지 않을 거예요. (하지만 반대로 파이썬의 타입 힌트를 좋아하지 않는다면, 어차피 Pydantic AI를 떠나서 장난감 에이전트 프레임워크를 쓰고 있을 거예요. 행운을 빌어요. 그리고 그게 필요하다는 걸 깨닫게 되면 제 큰 망치를 빌려 가도 좋아요)

요컨대, 그래프는 강력한 도구지만 모든 작업에 맞는 도구는 아니에요. 진행하기 전에 다른 멀티 에이전트 접근법을 먼저 고려해 주세요.

그래프 기반 접근법이 좋은 생각인지 확신이 없다면, 아마 불필요한 것일 수 있어요.

그래프와 유한 상태 머신(FSM)은 복잡한 워크플로우를 모델링·실행·제어·시각화하는 강력한 추상화예요.

Pydantic AI와 함께, 우리는 pydantic-graph를 개발했어요. 노드와 엣지를 타입 힌트로 정의하는 파이썬용 비동기 그래프 및 상태 머신 라이브러리죠.

이 라이브러리는 Pydantic AI의 일부로 개발되지만, pydantic-ai에 대한 의존성이 없어 순수 그래프 기반 상태 머신 라이브러리로 볼 수 있어요. Pydantic AI를 쓰는지, 아니면 GenAI로 빌드하는지와 무관하게 유용하게 쓸 수 있을 거예요.

pydantic-graph는 제어 흐름이 어려운 부분일 때 잡아 봐요. 명시적 상태, 분기, 재개 가능한 전환이 타입화되고 다이어그램으로 그려지길 원할 때죠. 이 라이브러리는 파이썬 제네릭과 타입 힌트에 기대요. 대부분의 에이전트에서는 일반 파이썬과 멀티 에이전트 패턴이 더 짧은 길이에요.

출처: 문서

본문

설치 (Installation)

pydantic-graph는 pydantic-ai의 필수 의존성이고, pydantic-ai-slim의 선택 의존성이에요. 자세한 내용은 설치 안내를 참고하세요. 직접 설치할 수도 있어요:

pip install pydantic-graph

uv add pydantic-graph

그래프 타입 (Graph Types)

pydantic-graph는 몇 가지 핵심 컴포넌트로 구성돼요:

GraphRunContext

GraphRunContext — 그래프 실행의 컨텍스트로, Pydantic AI의 RunContext와 비슷해요. 그래프의 상태와 의존성(deps)을 담고 있으며, 노드가 실행될 때 노드에 전달됩니다.

GraphRunContext는 사용되는 그래프의 상태 타입인 StateT에 대해 제네릭(generic)이에요.

End

End — 그래프 실행이 끝나야 함을 나타내는 반환 값이에요.

End는 사용되는 그래프의 반환 타입인 RunEndT에 대해 제네릭입니다.

노드 (Nodes)

BaseNode의 서브클래스가 그래프에서 실행될 노드를 정의해요.

일반적으로 dataclass인 노드는 대개 다음으로 구성됩니다:

• 노드를 호출할 때 필요/선택적인 파라미터를 담는 필드
• 노드를 실행하는 비즈니스 로직, run 메서드 안에 있음
• run 메서드의 반환 어노테이션. pydantic-graph가 이를 읽어 노드의 나가는(outgoing) 엣지를 결정해요

노드는 다음에 대해 제네릭입니다:

• 상태(state) — 노드가 포함된 그래프의 상태와 같은 타입이어야 함. StateT는 기본값이 None이라, 상태를 쓰지 않으면 이 제네릭 파라미터를 생략할 수 있어요. 상태가 있는 그래프 참고
• deps — 노드가 포함된 그래프의 deps와 같은 타입이어야 함. DepsT는 기본값이 None이라, deps를 쓰지 않으면 이 제네릭 파라미터를 생략할 수 있어요. 의존성 주입 참고
• 그래프 반환 타입 — 노드가 End를 반환하는 경우에만 적용됨. RunEndT의 기본값은 Never라서 노드가 End를 반환하지 않으면 이 제네릭 파라미터를 생략할 수 있지만, 반환한다면 포함해야 합니다.

그래프의 시작 또는 중간 노드 예제예요. End를 반환하지 않으므로 실행을 끝낼 수 없어요:

from dataclasses import dataclass

from pydantic_graph import BaseNode, GraphRunContext


@dataclass
class MyNode(BaseNode[MyState]):  # (1)
    foo: int  # (2)

    async def run(
        self,
        ctx: GraphRunContext[MyState],  # (3)
    ) -> AnotherNode:  # (4)
        ...
        return AnotherNode()

이 예제의 상태는 MyState(표시 생략)이므로 BaseNode는 MyState로 파라미터화돼요. 이 노드는 실행을 끝낼 수 없어서 RunEndT 제네릭 파라미터는 생략되고 Never로 기본 설정됩니다.

MyNode는 dataclass이고 int 타입의 단일 필드 foo를 가져요.

run 메서드는 다시 상태 MyState로 파라미터화된 GraphRunContext 파라미터를 받아요.

run 메서드의 반환 타입은 AnotherNode(표시 생략)예요. 이는 노드의 나가는 엣지를 결정하는 데 쓰입니다.

foo가 5로 나누어 떨어지면 실행을 선택적으로 끝내도록 MyNode를 확장할 수도 있어요:

from dataclasses import dataclass

from pydantic_graph import BaseNode, End, GraphRunContext


@dataclass
class MyNode(BaseNode[MyState, None, int]):  # (1)
    foo: int

    async def run(
        self,
        ctx: GraphRunContext[MyState],
    ) -> AnotherNode | End[int]:  # (2)
        if self.foo % 5 == 0:
            return End(self.foo)
        else:
            return AnotherNode()

상태뿐 아니라 반환 타입(이 경우 int)으로 노드를 파라미터화해요. 제네릭 파라미터는 위치 전용(positional-only)이라 deps를 나타내는 두 번째 파라미터로 None을 포함해야 합니다.

run 메서드의 반환 타입은 이제 AnotherNode와 End[int]의 유니온이에요. 이렇게 하면 foo가 5로 나누어 떨어질 때 노드가 실행을 끝낼 수 있어요.

그래프 (Graph)

Graph — GraphBuilder가 만드는 실행 가능한 그래프예요. 빌더는 스텝 함수, BaseNode 클래스, 그리고 그들을 연결하는 엣지로 그래프를 조립하는 진입점입니다.

GraphBuilder는 다음에 대해 제네릭이에요:

• state 그래프의 상태 타입, StateT
• deps 그래프의 deps 타입, DepsT
• input 그래프에 전달되는 초기 입력 타입, InputT
• output 그래프가 만드는 최종 출력 타입, OutputT

두 BaseNode 서브클래스로 만든 단순한 그래프 예제예요:

from __future__ import annotations

from dataclasses import dataclass

from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext


@dataclass
class DivisibleBy5(BaseNode[None, None, int]):  # (1)
    foo: int

    async def run(
        self,
        ctx: GraphRunContext,
    ) -> Increment | End[int]:
        if self.foo % 5 == 0:
            return End(self.foo)
        else:
            return Increment(self.foo)


@dataclass
class Increment(BaseNode):  # (2)
    foo: int

    async def run(self, ctx: GraphRunContext) -> DivisibleBy5:
        return DivisibleBy5(self.foo + 1)


g = GraphBuilder(input_type=int, output_type=int)  # (3)


@g.step
async def start(ctx: StepContext[None, None, int]) -> DivisibleBy5:  # (4)
    return DivisibleBy5(ctx.inputs)


g.add(
    g.node(DivisibleBy5),  # (5)
    g.node(Increment),
    g.edge_from(g.start_node).to(start),  # (6)
)

fives_graph = g.build()  # (7)


async def main():
    result = await fives_graph.run(inputs=4)  # (8)
    print(result)
    #> 5

DivisibleBy5 노드는 이 그래프가 상태나 deps를 쓰지 않으므로 상태 파라미터에 None, deps 파라미터에 None으로 파라미터화되고, 실행을 끝낼 수 있으므로 int로 파라미터화돼요.

Increment 노드는 End를 반환하지 않으므로 RunEndT 제네릭 파라미터는 생략되고, 그래프가 상태를 쓰지 않으므로 상태도 생략할 수 있어요.

그래프의 입력·출력 타입을 선언해 GraphBuilder를 만드세요.

초기 입력을 첫 BaseNode로 감싸는 스텝을 정의하세요. 빌더는 실행이 g.start_node를 떠날 때 이 스텝을 호출해요.

빌더가 알도록 각 BaseNode 서브클래스를 g.node()로 등록하세요. 나가는 엣지는 각 노드의 run 반환 타입에서 추론돼요.

시작 노드를 진입 스텝에 연결하세요.

g.build()는 실행 준비가 된 Graph를 반환해요.

graph.run()은 비동기이며 원시 출력 값(End 노드가 반환한 int)을 반환합니다.

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

이 그래프의 mermaid 다이어그램은 print(fives_graph)로, 또는 fives_graph.render()를 호출해 생성할 수 있어요:

stateDiagram-v2
  start
  DivisibleBy5
  state decision <<choice>>
  Increment

  [*] --> start
  start --> DivisibleBy5
  DivisibleBy5 --> decision
  decision --> Increment
  decision --> [*]
  Increment --> DivisibleBy5

상태가 있는 그래프 (Stateful Graphs)

pydantic-graph의 "상태" 개념은 노드가 그래프에서 실행될 때 객체(흔히 dataclass 또는 Pydantic 모델)에 접근하고 변경하는 선택적 방법을 제공해요. 그래프를 생산 라인으로 생각해 보세요. 그러면 상태는 라인을 따라 전달되며 그래프가 실행되는 동안 각 노드가 조립하는 엔진과 같아요.

사용자가 동전을 넣고 살 제품을 선택하는 자판기를 나타내는 그래프 예제예요.

from __future__ import annotations

from dataclasses import dataclass

from rich.prompt import Prompt

from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext


@dataclass
class MachineState:  # (1)
    user_balance: float = 0.0
    product: str | None = None


@dataclass
class InsertCoin(BaseNode[MachineState]):  # (3)
    async def run(self, ctx: GraphRunContext[MachineState]) -> CoinsInserted:  # (14)
        return CoinsInserted(float(Prompt.ask('Insert coins')))  # (4)


@dataclass
class CoinsInserted(BaseNode[MachineState]):
    amount: float  # (5)

    async def run(
        self, ctx: GraphRunContext[MachineState]
    ) -> SelectProduct | Purchase:  # (15)
        ctx.state.user_balance += self.amount  # (6)
        if ctx.state.product is not None:  # (7)
            return Purchase(ctx.state.product)
        else:
            return SelectProduct()


@dataclass
class SelectProduct(BaseNode[MachineState]):
    async def run(self, ctx: GraphRunContext[MachineState]) -> Purchase:
        return Purchase(Prompt.ask('Select product'))


PRODUCT_PRICES = {  # (2)
    'water': 1.25,
    'soda': 1.50,
    'crisps': 1.75,
    'chocolate': 2.00,
}


@dataclass
class Purchase(BaseNode[MachineState, None, None]):  # (16)
    product: str

    async def run(
        self, ctx: GraphRunContext[MachineState]
    ) -> End | InsertCoin | SelectProduct:
        if price := PRODUCT_PRICES.get(self.product):  # (8)
            ctx.state.product = self.product  # (9)
            if ctx.state.user_balance >= price:  # (10)
                ctx.state.user_balance -= price
                return End(None)
            else:
                diff = price - ctx.state.user_balance
                print(f'Not enough money for {self.product}, need {diff:0.2f} more')
                #> Not enough money for crisps, need 0.75 more
                return InsertCoin()  # (11)
        else:
            print(f'No such product: {self.product}, try again')
            return SelectProduct()  # (12)


g = GraphBuilder(state_type=MachineState)  # (13)


@g.step
async def start(ctx: StepContext[MachineState, None, None]) -> InsertCoin:
    return InsertCoin()


g.add(
    g.node(InsertCoin),
    g.node(CoinsInserted),
    g.node(SelectProduct),
    g.node(Purchase),
    g.edge_from(g.start_node).to(start),
)

vending_machine_graph = g.build()


async def main():
    state = MachineState()  # (17)
    await vending_machine_graph.run(state=state)  # (18)
    print(f'purchase successful item={state.product} change={state.user_balance:0.2f}')
    #> purchase successful item=crisps change=0.25

자판기 상태는 사용자의 잔액과 선택한 제품(있을 경우)을 가진 dataclass로 정의돼요.

가격에 매핑된 제품 딕셔너리예요.

InsertCoin 노드. BaseNode는 이 그래프에서 쓰는 상태이므로 MachineState로 파라미터화돼요.

InsertCoin 노드는 사용자에게 동전을 넣으라고 요청해요. 금액을 float로 입력하게 해서 간단하게 유지합니다.

CoinsInserted 노드; 이것 역시 한 필드 amount를 가진 dataclass예요.

넣은 금액으로 사용자 잔액을 갱신하세요.

사용자가 이미 제품을 선택했다면 Purchase로, 아니면 SelectProduct로 갑니다.

Purchase 노드에서, 사용자가 유효한 제품을 입력했다면 제품의 가격을 조회해요.

사용자가 유효한 제품을 입력했다면, SelectProduct를 다시 방문하지 않도록 상태에 제품을 설정해요.

잔액이 제품을 살 만큼 충분하면 구매를 반영하도록 잔액을 조정하고 그래프를 끝내기 위해 End를 반환해요. 실행 반환 타입을 쓰지 않으므로 None으로 End를 호출합니다.

잔액이 부족하면 사용자에게 동전을 더 넣으라고 요청하도록 InsertCoin으로 갑니다.

제품이 유효하지 않으면 사용자에게 제품을 다시 선택하라고 요청하도록 SelectProduct로 갑니다.

GraphBuilder로 MachineState 타입을 선언해 그래프를 빌드하세요. 각 BaseNode 서브클래스는 g.node()로 등록되고, 나가는 엣지는 run 반환 타입에서 추론돼요. start 스텝이 첫 번째 노드를 만듭니다.

노드의 run 메서드의 반환 타입은 노드의 나가는 엣지를 결정하는 데 쓰이므로 중요해요. 이 정보는 mermaid 다이어그램 렌더링에 사용되고, 잘못된 동작을 가능한 한 빨리 감지하도록 런타임에서 강제됩니다.

CoinsInserted의 run 메서드의 반환 타입은 유니온이라 여러 나가는 엣지가 가능해요.

다른 노드와 달리 Purchase는 실행을 끝낼 수 있으므로 RunEndT 제네릭 파라미터를 설정해야 해요. 이 경우 그래프 실행 반환 타입이 None이므로 None입니다.

상태를 초기화하세요. 이는 그래프 실행에 전달되고 그래프가 실행되며 변경됩니다.

초기 상태로 그래프를 실행하세요. 실행할 첫 노드는 g.start_node에 연결한 start 스텝이 결정합니다.

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

이 그래프의 mermaid 다이어그램은 print(vending_machine_graph)로 생성할 수 있어요:

stateDiagram-v2
  start
  InsertCoin
  CoinsInserted
  state decision <<choice>>
  Purchase
  SelectProduct
  state decision_2 <<choice>>

  [*] --> start
  start --> InsertCoin
  InsertCoin --> CoinsInserted
  CoinsInserted --> decision
  decision --> Purchase
  decision --> SelectProduct
  SelectProduct --> Purchase
  Purchase --> decision_2
  decision_2 --> InsertCoin
  decision_2 --> SelectProduct
  decision_2 --> [*]

다이어그램 생성에 대한 자세한 내용은 아래를 참고하세요.

GenAI 예제 (GenAI Example)

지금까지는 Pydantic AI나 GenAI를 전혀 쓰지 않는 그래프 예제를 보여주지 않았어요.

이 예제에서는 한 에이전트가 사용자에게 환영 이메일을 생성하고, 다른 에이전트가 그 이메일에 대한 피드백을 제공해요.

이 그래프는 매우 단순한 구조를 가져요:

---
title: feedback_graph
---
stateDiagram-v2
  [*] --> WriteEmail
  WriteEmail --> Feedback
  Feedback --> WriteEmail
  Feedback --> [*]

from __future__ import annotations as _annotations

from dataclasses import dataclass, field

from pydantic import BaseModel, EmailStr

from pydantic_ai import Agent, ModelMessage, format_as_xml
from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext


@dataclass
class User:
    name: str
    email: EmailStr
    interests: list[str]


@dataclass
class Email:
    subject: str
    body: str


@dataclass
class State:
    user: User
    write_agent_messages: list[ModelMessage] = field(default_factory=list)


email_writer_agent = Agent(
    'google:gemini-3-pro-preview',
    output_type=Email,
    instructions='Write a welcome email to our tech blog.',
)


@dataclass
class WriteEmail(BaseNode[State]):
    email_feedback: str | None = None

    async def run(self, ctx: GraphRunContext[State]) -> Feedback:
        if self.email_feedback:
            prompt = (
                f'Rewrite the email for the user:\n'
                f'{format_as_xml(ctx.state.user)}\n'
                f'Feedback: {self.email_feedback}'
            )
        else:
            prompt = (
                f'Write a welcome email for the user:\n'
                f'{format_as_xml(ctx.state.user)}'
            )

        result = await email_writer_agent.run(
            prompt,
            message_history=ctx.state.write_agent_messages,
        )
        ctx.state.write_agent_messages += result.new_messages()
        return Feedback(result.output)


class EmailRequiresWrite(BaseModel):
    feedback: str


class EmailOk(BaseModel):
    pass


feedback_agent = Agent[object, EmailRequiresWrite | EmailOk](
    'openai:gpt-5.2',
    output_type=EmailRequiresWrite | EmailOk,  # type: ignore
    instructions=(
        'Review the email and provide feedback, email must reference the users specific interests.'
    ),
)


@dataclass
class Feedback(BaseNode[State, None, Email]):
    email: Email

    async def run(
        self,
        ctx: GraphRunContext[State],
    ) -> WriteEmail | End[Email]:
        prompt = format_as_xml({'user': ctx.state.user, 'email': self.email})
        result = await feedback_agent.run(prompt)
        if isinstance(result.output, EmailRequiresWrite):
            return WriteEmail(email_feedback=result.output.feedback)
        else:
            return End(self.email)


g = GraphBuilder(state_type=State, output_type=Email)


@g.step
async def start(ctx: StepContext[State, None, None]) -> WriteEmail:
    return WriteEmail()


g.add(
    g.node(WriteEmail),
    g.node(Feedback),
    g.edge_from(g.start_node).to(start),
)

feedback_graph = g.build()


async def main():
    user = User(
        name='John Doe',
        email='[email protected]',
        interests=['Haskel', 'Lisp', 'Fortran'],
    )
    state = State(user)
    result = await feedback_graph.run(state=state)
    print(result)
    """
    Email(
        subject='Welcome to our tech blog!',
        body='Hello John, Welcome to our tech blog! ...',
    )
    """

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

그래프 반복 (Iterating Over a Graph)

단계별 실행 — 각 작업이 실행될 때 검사, 다음 스텝 오버라이드, 루프를 수동으로 구동 — 을 하려면 graph.run() 대신 graph.iter()를 사용하세요. 반복 모델과 예제는 그래프 빌더 문서의 고급 실행 제어를 참고하세요.

의존성 주입 (Dependency Injection)

Pydantic AI와 마찬가지로 pydantic-graph도 의존성 주입을 지원해요. GraphBuilder에 deps_type을 넘기고, 각 BaseNode 서브클래스를 deps 타입으로 파라미터화한 뒤, run() 안에서 GraphRunContext.deps로 읽습니다(스텝 함수 안에서는 StepContext.deps).

예를 들어, 위의 DivisibleBy5 예제를 수정해 ProcessPoolExecutor로 계산 부하를 별도 프로세스에서 실행해 봅시다(이 예제에서 ProcessPoolExecutor가 실제로 성능을 개선하지는 않는 인위적인 예제예요):

from __future__ import annotations

import asyncio
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass

from pydantic_graph import BaseNode, End, GraphBuilder, GraphRunContext, StepContext


@dataclass
class GraphDeps:
    executor: ProcessPoolExecutor


@dataclass
class DivisibleBy5(BaseNode[None, GraphDeps, int]):
    foo: int

    async def run(
        self,
        ctx: GraphRunContext[None, GraphDeps],
    ) -> Increment | End[int]:
        if self.foo % 5 == 0:
            return End(self.foo)
        else:
            return Increment(self.foo)


@dataclass
class Increment(BaseNode[None, GraphDeps]):
    foo: int

    async def run(self, ctx: GraphRunContext[None, GraphDeps]) -> DivisibleBy5:
        loop = asyncio.get_running_loop()
        compute_result = await loop.run_in_executor(
            ctx.deps.executor,
            self.compute,
        )
        return DivisibleBy5(compute_result)

    def compute(self) -> int:
        return self.foo + 1


g = GraphBuilder(deps_type=GraphDeps, input_type=int, output_type=int)


@g.step
async def start(ctx: StepContext[None, GraphDeps, int]) -> DivisibleBy5:
    return DivisibleBy5(ctx.inputs)


g.add(
    g.node(DivisibleBy5),
    g.node(Increment),
    g.edge_from(g.start_node).to(start),
)

fives_graph = g.build()


async def main():
    with ProcessPoolExecutor() as executor:
        deps = GraphDeps(executor)
        result = await fives_graph.run(inputs=3, deps=deps)
    print(result)
    #> 5

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

Mermaid 다이어그램 (Mermaid Diagrams)

Pydantic Graph는 만들어진 어떤 그래프에 대해서도 mermaid stateDiagram-v2 다이어그램을 렌더링할 수 있어요. mermaid 소스를 얻으려면 graph.render()를 호출(또는 그냥 print(graph))하세요. 레이아웃을 제어하려면 direction('TB', 'LR', 'RL', 'BT')을 전달하세요. 렌더링 옵션 전체는 그래프 빌더 mermaid 섹션을 참고하세요.

더 알아보기 (Learn more)

• 그래프 빌더 — 선언적 스텝/결정/조인 기반 그래프 구성 API.
• 멀티 에이전트 애플리케이션 — 그래프 없이 에이전트를 조합하는 다른 방법.