결정 노드
결정 노드 (Decision Nodes)
결정 노드(decision node)는 그래프를 흐르는 데이터의 타입이나 값에 따라 그래프에 조건부 분기(conditional branching)를 만들어 주는 기능이에요. 하향식으로 분기를 하나씩 평가해서 조건에 맞는 경로를 따라가는 구조죠.
출처: 문서
본문
결정 노드는 들어오는 데이터를 평가하고, 다음 기준에 따라 다른 분기로 라우팅해요:
- 타입 매칭(
isinstance사용) - 리터럴 값 매칭
- 커스텀 조건 함수(predicate function)
첫 번째로 매칭된 분기가 선택되며, 이는 패턴 매칭이나 if-elif-else 체인과 비슷해요.
결정 만들기 (Creating Decisions)
g.decision()으로 결정 노드를 만들고, g.match()로 분기를 추가합니다:
from dataclasses import dataclass
from typing import Literal
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
path_taken: str | None = None
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def choose_path(ctx: StepContext[DecisionState, None, None]) -> Literal['left', 'right']:
return 'left'
@g.step
async def left_path(ctx: StepContext[DecisionState, None, object]) -> str:
ctx.state.path_taken = 'left'
return 'Went left'
@g.step
async def right_path(ctx: StepContext[DecisionState, None, object]) -> str:
ctx.state.path_taken = 'right'
return 'Went right'
g.add(
g.edge_from(g.start_node).to(choose_path),
g.edge_from(choose_path).to(
g.decision()
.branch(g.match(TypeExpression[Literal['left']]).to(left_path))
.branch(g.match(TypeExpression[Literal['right']]).to(right_path))
),
g.edge_from(left_path, right_path).to(g.end_node),
)
graph = g.build()
state = DecisionState()
result = await graph.run(state=state)
print(result)
#> Went left
print(state.path_taken)
#> left
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
타입 매칭 (Type Matching)
일반 파이썬 타입으로 타입을 매칭해요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def return_int(ctx: StepContext[DecisionState, None, None]) -> int:
return 42
@g.step
async def handle_int(ctx: StepContext[DecisionState, None, int]) -> str:
return f'Got int: {ctx.inputs}'
@g.step
async def handle_str(ctx: StepContext[DecisionState, None, str]) -> str:
return f'Got str: {ctx.inputs}'
g.add(
g.edge_from(g.start_node).to(return_int),
g.edge_from(return_int).to(
g.decision()
.branch(g.match(int).to(handle_int))
.branch(g.match(str).to(handle_str))
),
g.edge_from(handle_int, handle_str).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> Got int: 42
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
유니온 타입 매칭 (Matching Union Types)
유니온 같은 더 복잡한 타입 표현식에는 TypeExpression을 써야 해요. 파이썬 타입 시스템은 유니온 타입을 런타임 값으로 직접 사용하는 걸 허용하지 않거든요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def return_value(ctx: StepContext[DecisionState, None, None]) -> int | str:
"""Returns either an int or a str."""
return 42
@g.step
async def handle_number(ctx: StepContext[DecisionState, None, int | float]) -> str:
return f'Got number: {ctx.inputs}'
@g.step
async def handle_text(ctx: StepContext[DecisionState, None, str]) -> str:
return f'Got text: {ctx.inputs}'
g.add(
g.edge_from(g.start_node).to(return_value),
g.edge_from(return_value).to(
g.decision()
# Use TypeExpression for union types
.branch(g.match(TypeExpression[int | float]).to(handle_number))
.branch(g.match(str).to(handle_text))
),
g.edge_from(handle_number, handle_text).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> Got number: 42
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
참고
TypeExpression은 유니온(int | str)이나 Literal처럼 런타임 type 객체로 유효하지 않은 복잡한 타입 표현식에서만 필요해요. int, str, 커스텀 클래스 같은 단순 타입은 g.match()에 직접 넘기면 됩니다.
PEP 747에서 도입된 TypeForm 클래스가 언젠가 이 우회 방법의 필요성을 없애줄 거예요.
커스텀 매처 (Custom Matchers)
matches 파라미터로 커스텀 매칭 로직을 제공해요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def return_number(ctx: StepContext[DecisionState, None, None]) -> int:
return 7
@g.step
async def even_path(ctx: StepContext[DecisionState, None, int]) -> str:
return f'{ctx.inputs} is even'
@g.step
async def odd_path(ctx: StepContext[DecisionState, None, int]) -> str:
return f'{ctx.inputs} is odd'
g.add(
g.edge_from(g.start_node).to(return_number),
g.edge_from(return_number).to(
g.decision()
.branch(g.match(TypeExpression[int], matches=lambda x: x % 2 == 0).to(even_path))
.branch(g.match(TypeExpression[int], matches=lambda x: x % 2 == 1).to(odd_path))
),
g.edge_from(even_path, odd_path).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> 7 is odd
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
분기 우선순위 (Branch Priority)
분기는 추가된 순서대로 평가돼요. 처음으로 매칭된 분기가 선택됩니다:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def return_value(ctx: StepContext[DecisionState, None, None]) -> int:
return 10
@g.step
async def branch_a(ctx: StepContext[DecisionState, None, int]) -> str:
return 'Branch A'
@g.step
async def branch_b(ctx: StepContext[DecisionState, None, int]) -> str:
return 'Branch B'
g.add(
g.edge_from(g.start_node).to(return_value),
g.edge_from(return_value).to(
g.decision()
.branch(g.match(TypeExpression[int], matches=lambda x: x >= 5).to(branch_a))
.branch(g.match(TypeExpression[int], matches=lambda x: x >= 0).to(branch_b))
),
g.edge_from(branch_a, branch_b).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> Branch A
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
두 분기 모두 10과 매칭될 수 있지만, Branch A가 먼저라서 선택돼요.
포괄 분기 (Catch-All Branches)
object나 Any를 사용해 어떤 값이든 잡는 포괄 분기를 만들 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def return_value(ctx: StepContext[DecisionState, None, None]) -> int:
return 100
@g.step
async def catch_all(ctx: StepContext[DecisionState, None, object]) -> str:
return f'Caught: {ctx.inputs}'
g.add(
g.edge_from(g.start_node).to(return_value),
g.edge_from(return_value).to(g.decision().branch(g.match(TypeExpression[object]).to(catch_all))),
g.edge_from(catch_all).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> Caught: 100
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
중첩 결정 (Nested Decisions)
복잡한 조건 로직을 위해 결정을 중첩할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def get_number(ctx: StepContext[DecisionState, None, None]) -> int:
return 15
@g.step
async def is_positive(ctx: StepContext[DecisionState, None, int]) -> int:
return ctx.inputs
@g.step
async def is_negative(ctx: StepContext[DecisionState, None, int]) -> str:
return 'Negative'
@g.step
async def small_positive(ctx: StepContext[DecisionState, None, int]) -> str:
return 'Small positive'
@g.step
async def large_positive(ctx: StepContext[DecisionState, None, int]) -> str:
return 'Large positive'
g.add(
g.edge_from(g.start_node).to(get_number),
g.edge_from(get_number).to(
g.decision()
.branch(g.match(TypeExpression[int], matches=lambda x: x > 0).to(is_positive))
.branch(g.match(TypeExpression[int], matches=lambda x: x <= 0).to(is_negative))
),
g.edge_from(is_positive).to(
g.decision()
.branch(g.match(TypeExpression[int], matches=lambda x: x < 10).to(small_positive))
.branch(g.match(TypeExpression[int], matches=lambda x: x >= 10).to(large_positive))
),
g.edge_from(is_negative, small_positive, large_positive).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> Large positive
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
라벨이 있는 분기 (Branching with Labels)
문서화와 다이어그램 생성을 위해 분기에 라벨을 추가할 수 있어요:
from dataclasses import dataclass
from typing import Literal
from pydantic_graph import GraphBuilder, StepContext, TypeExpression
@dataclass
class DecisionState:
pass
async def main():
g = GraphBuilder(state_type=DecisionState, output_type=str)
@g.step
async def choose(ctx: StepContext[DecisionState, None, None]) -> Literal['a', 'b']:
return 'a'
@g.step
async def path_a(ctx: StepContext[DecisionState, None, object]) -> str:
return 'Path A'
@g.step
async def path_b(ctx: StepContext[DecisionState, None, object]) -> str:
return 'Path B'
g.add(
g.edge_from(g.start_node).to(choose),
g.edge_from(choose).to(
g.decision()
.branch(g.match(TypeExpression[Literal['a']]).label('Take path A').to(path_a))
.branch(g.match(TypeExpression[Literal['b']]).label('Take path B').to(path_b))
),
g.edge_from(path_a, path_b).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=DecisionState())
print(result)
#> Path A
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)