분기(Branch)와 루프(Loop)

분기(Branch)와 루프(Loop)

워크플로의 핵심 특징 중 하나는 그래프 기반 접근보다 더 단순하고 유연하게 분기·반복 로직을 쓸 수 있다는 점이에요.

출처: 공식문서 - Branches and loops

워크플로에서의 루프

루프를 만들려면 무작위로 반복하는 LoopingWorkflow를 준비할게요. LoopEvent라는 이름의 이벤트 하나를 사용합니다(이름은 마음대로 바꿔도 돼요).

from workflows.events import Event


class LoopEvent(Event):
    num_loops: int

이제 random을 import 하고 step_one 함수를 반복할지 계속할지를 무작위로 결정하도록 수정해요.

import random
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent


class LoopingWorkflow(Workflow):
    @step
    async def prepare_input(self, ev: StartEvent) -> LoopEvent:
        num_loops = random.randint(0, 10)
        return LoopEvent(num_loops=num_loops)

    @step
    async def loop_step(self, ev: LoopEvent) -> LoopEvent | StopEvent:
        if ev.num_loops <= 0:
            return StopEvent(result="Done looping!")
        return LoopEvent(num_loops=ev.num_loops-1)

적절한 이벤트 타입과 반환 타입을 정의하면, 어떤 스텝에서 어떤 스텝으로든 루프를 만들 수 있어요.

워크플로에서의 분기

루프와 밀접하게 연관된 것이 분기(branch)예요. 앞에서 보았듯 조건에 따라 서로 다른 이벤트를 반환할 수 있죠. 두 개의 서로 다른 경로로 분기하는 워크플로를 살펴볼게요.

import random
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent


class BranchA1Event(Event):
    payload: str


class BranchA2Event(Event):
    payload: str


class BranchB1Event(Event):
    payload: str


class BranchB2Event(Event):
    payload: str


class BranchWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> BranchA1Event | BranchB1Event:
        if random.randint(0, 1) == 0:
            print("Go to branch A")
            return BranchA1Event(payload="Branch A")
        else:
            print("Go to branch B")
            return BranchB1Event(payload="Branch B")

    @step
    async def step_a1(self, ev: BranchA1Event) -> BranchA2Event:
        print(ev.payload)
        return BranchA2Event(payload=ev.payload)

    @step
    async def step_b1(self, ev: BranchB1Event) -> BranchB2Event:
        print(ev.payload)
        return BranchB2Event(payload=ev.payload)

    @step
    async def step_a2(self, ev: BranchA2Event) -> StopEvent:
        print(ev.payload)
        return StopEvent(result="Branch A complete.")

    @step
    async def step_b2(self, ev: BranchB2Event) -> StopEvent:
        print(ev.payload)
        return StopEvent(result="Branch B complete.")

import는 앞과 같지만 이벤트 타입 4개를 새로 만들었어요. start는 무작위로 한 분기를 타고, 각 분기 안의 여러 스텝이 워크플로를 완성합니다. 분기와 루프는 애플리케이션 요구에 맞게 어떤 순서로든 조합할 수 있어요. 이 튜토리얼의 뒤에서는 send_event로 여러 분기를 병렬 실행하고 collect_events로 동기화하는 방법도 배우게 됩니다.

이벤트 서브클래스 라우팅

이벤트 라우팅은 기본적으로 정확히 일치(exact match)해요. ParentEvent로 어노테이션된 스텝은, ChildEvent가 그 클래스를 상속하더라도 자동으로 ChildEvent를 받지 않습니다. 이는 이벤트 모델이 커지면서 우연한 광범위 매칭으로 워크플로가 바뀌는 것을 막아줘요.

부모 이벤트가 정말로 라우팅 계약이라면 스텝 단위로 선택(opt-in)할 수 있습니다.

from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent


class ToolEvent(Event):
    tool_name: str


class SearchEvent(ToolEvent):
    query: str


class ToolWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> SearchEvent:
        return SearchEvent(tool_name="search", query=ev.query)

    @step(accept_event_subclasses=True)
    async def handle_tool(self, ev: ToolEvent) -> StopEvent:
        return StopEvent(result=ev.tool_name)

이 기능은 아껴서 써야 해요. 대부분의 워크플로에서는 구체적인(concrete) 이벤트 타입이 분기 구조를 더 명확하게 만들고 더 나은 검증 오류를 줍니다.

더 알아보기