스텝
스텝 (Steps)
스텝(step)은 그래프에서 작업을 수행하는 기본 단위예요. StepContext를 받아 값을 반환하는 비동기 함수입니다.
출처: 문서
본문
스텝 만들기 (Creating Steps)
스텝은 GraphBuilder의 @g.step 데코레이터로 만들 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class MyState:
counter: int = 0
g = GraphBuilder(state_type=MyState, output_type=int)
@g.step
async def increment(ctx: StepContext[MyState, None, None]) -> int:
ctx.state.counter += 1
return ctx.state.counter
g.add(
g.edge_from(g.start_node).to(increment),
g.edge_from(increment).to(g.end_node),
)
graph = g.build()
async def main():
state = MyState()
result = await graph.run(state=state)
print(result)
#> 1
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
스텝 컨텍스트 (Step Context)
모든 스텝 함수는 첫 번째 파라미터로 StepContext를 받아요. 컨텍스트는 다음에 접근할 수 있게 해줍니다:
ctx.state- 변경 가능한 그래프 상태 (타입:StateT)ctx.deps- 주입된 의존성 (타입:DepsT)ctx.inputs- 이 스텝의 입력 데이터 (타입:InputT)
상태 접근하기 (Accessing State)
상태는 그래프의 모든 스텝에서 공유되고 자유롭게 변경할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class AppState:
messages: list[str]
async def main():
g = GraphBuilder(state_type=AppState, output_type=list[str])
@g.step
async def add_hello(ctx: StepContext[AppState, None, None]) -> None:
ctx.state.messages.append('Hello')
@g.step
async def add_world(ctx: StepContext[AppState, None, None]) -> None:
ctx.state.messages.append('World')
@g.step
async def get_messages(ctx: StepContext[AppState, None, None]) -> list[str]:
return ctx.state.messages
g.add(
g.edge_from(g.start_node).to(add_hello),
g.edge_from(add_hello).to(add_world),
g.edge_from(add_world).to(get_messages),
g.edge_from(get_messages).to(g.end_node),
)
graph = g.build()
state = AppState(messages=[])
result = await graph.run(state=state)
print(result)
#> ['Hello', 'World']
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
입력 다루기 (Working with Inputs)
스텝은 입력 데이터를 받아 변환할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(
state_type=SimpleState,
input_type=int,
output_type=str,
)
@g.step
async def double_it(ctx: StepContext[SimpleState, None, int]) -> int:
"""Double the input value."""
return ctx.inputs * 2
@g.step
async def stringify(ctx: StepContext[SimpleState, None, int]) -> str:
"""Convert to a formatted string."""
return f'Result: {ctx.inputs}'
g.add(
g.edge_from(g.start_node).to(double_it),
g.edge_from(double_it).to(stringify),
g.edge_from(stringify).to(g.end_node),
)
graph = g.build()
result = await graph.run(state=SimpleState(), inputs=21)
print(result)
#> Result: 42
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
의존성 주입 (Dependency Injection)
스텝은 ctx.deps를 통해 주입된 의존성에 접근할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class AppState:
pass
@dataclass
class AppDeps:
"""Dependencies injected into the graph."""
multiplier: int
async def main():
g = GraphBuilder(
state_type=AppState,
deps_type=AppDeps,
input_type=int,
output_type=int,
)
@g.step
async def multiply(ctx: StepContext[AppState, AppDeps, int]) -> int:
"""Multiply input by the injected multiplier."""
return ctx.inputs * ctx.deps.multiplier
g.add(
g.edge_from(g.start_node).to(multiply),
g.edge_from(multiply).to(g.end_node),
)
graph = g.build()
deps = AppDeps(multiplier=10)
result = await graph.run(state=AppState(), deps=deps, inputs=5)
print(result)
#> 50
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
스텝 커스터마이징 (Customizing Steps)
커스텀 노드 ID (Custom Node IDs)
기본적으로 스텝 노드 ID는 함수 이름에서 추론돼요. 다음처럼 오버라이드할 수 있습니다:
from pydantic_graph import StepContext
from basic_step import MyState, g
@g.step(node_id='my_custom_id')
async def my_step(ctx: StepContext[MyState, None, None]) -> int:
return 42
# The node ID is now 'my_custom_id' instead of 'my_step'
사람이 읽을 수 있는 라벨 (Human-Readable Labels)
라벨은 다이어그램 생성을 위한 문서화를 제공해요:
from pydantic_graph import StepContext
from basic_step import MyState, g
@g.step(label='Increment the counter')
async def increment(ctx: StepContext[MyState, None, None]) -> int:
ctx.state.counter += 1
return ctx.state.counter
# Access the label programmatically
print(increment.label)
#> Increment the counter
순차 스텝 (Sequential Steps)
여러 스텝을 순차로 체인으로 연결할 수 있어요:
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class MathState:
operations: list[str]
async def main():
g = GraphBuilder(
state_type=MathState,
input_type=int,
output_type=int,
)
@g.step
async def add_five(ctx: StepContext[MathState, None, int]) -> int:
ctx.state.operations.append('add 5')
return ctx.inputs + 5
@g.step
async def multiply_by_two(ctx: StepContext[MathState, None, int]) -> int:
ctx.state.operations.append('multiply by 2')
return ctx.inputs * 2
@g.step
async def subtract_three(ctx: StepContext[MathState, None, int]) -> int:
ctx.state.operations.append('subtract 3')
return ctx.inputs - 3
# Connect steps sequentially
g.add(
g.edge_from(g.start_node).to(add_five),
g.edge_from(add_five).to(multiply_by_two),
g.edge_from(multiply_by_two).to(subtract_three),
g.edge_from(subtract_three).to(g.end_node),
)
graph = g.build()
state = MathState(operations=[])
result = await graph.run(state=state, inputs=10)
print(f'Result: {result}')
#> Result: 27
print(f'Operations: {state.operations}')
#> Operations: ['add 5', 'multiply by 2', 'subtract 3']
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
계산은 (10 + 5) * 2 - 3 = 27이에요.
스트리밍 스텝 (Streaming Steps)
단일 값을 반환하는 일반 스텝 외에도, @g.stream 데코레이터를 사용해 시간에 따라 여러 값을 산출하는 스트리밍 스텝을 만들 수 있어요:
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.stream
async def generate_stream(ctx: StepContext[SimpleState, None, None]):
"""Stream numbers from 1 to 5."""
for i in range(1, 6):
yield i
@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])
g.add(
g.edge_from(g.start_node).to(generate_stream),
# The stream output is an AsyncIterable, so we can map over it
g.edge_from(generate_stream).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())을 추가하세요. 다른 변경은 필요 없어요.)
스트리밍 스텝이 동작하는 방식 (How Streaming Steps Work)
스트리밍 스텝은 시간에 따라 값을 산출하는 AsyncIterable을 반환해요. 스트리밍 스텝의 출력에 .map()을 쓰면, 그래프가 각 산출 값이 준비될 때마다 처리하며 병렬 작업을 동적으로 만들어요. 이는 다음과 같은 경우 특히 유용합니다:
- 응답을 스트리밍하는 API에서 데이터 처리
- 실시간 데이터 피드 처리
- 대용량 데이터셋의 진행형 처리
- 모든 데이터가 준비되기 전에 결과 처리를 시작하고 싶은 모든 시나리오
일반 스텝처럼 스트리밍 스텝도 커스텀 노드 ID와 라벨을 가질 수 있어요:
from pydantic_graph import StepContext
from streaming_step import SimpleState, g
@g.stream(node_id='my_stream', label='Generate numbers progressively')
async def labeled_stream(ctx: StepContext[SimpleState, None, None]):
for i in range(10):
yield i
엣지 구축 편의 메서드 (Edge Building Convenience Methods)
빌더는 흔한 엣지 패턴을 위한 헬퍼 메서드를 제공해요:
add_edge()로 간단한 엣지 (Simple Edges with add_edge())
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class SimpleState:
pass
async def main():
g = GraphBuilder(state_type=SimpleState, output_type=int)
@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]) -> int:
return ctx.inputs + 5
# Using add_edge() for simple connections
g.add_edge(g.start_node, step_a)
g.add_edge(step_a, step_b, label='from a to b')
g.add_edge(step_b, g.end_node)
graph = g.build()
result = await graph.run(state=SimpleState())
print(result)
#> 15
(이 예제를 실행하려면 asyncio를 임포트하고 asyncio.run(main())을 추가하세요. 다른 변경은 필요 없어요.)
타입 안전성 (Type Safety)
그래프 빌더 API는 제네릭(generics)을 통해 강력한 타입 검사를 제공해요. StepContext의 타입 파라미터는 다음을 보장해요:
- 상태 접근이 올바르게 타입화됨
- 의존성이 올바르게 타입화됨
- 입력/출력 타입이 엣지를 가로질러 일치함
from dataclasses import dataclass
from pydantic_graph import GraphBuilder, StepContext
@dataclass
class MyState:
pass
g = GraphBuilder(state_type=MyState, output_type=str)
# Type checker will catch mismatches
@g.step
async def expects_int(ctx: StepContext[MyState, None, int]) -> str:
return str(ctx.inputs)
@g.step
async def returns_str(ctx: StepContext[MyState, None, None]) -> str:
return 'hello'
# This would be a type error - expects_int needs int input, but returns_str outputs str
# g.add(g.edge_from(returns_str).to(expects_int)) # Type error!