커스텀 시작·종료 이벤트(Custom Start and Stop Events)

커스텀 시작·종료 이벤트(Custom Start and Stop Events)

대부분의 워크플로는 Getting Started에서 소개한 기본 StartEvent와 StopEvent로 충분해요. 워크플로 경계 자체에 유용한 스키마가 있을 때, 즉 시작 부분에 타입이 있는 요청 객체가 있거나 끝 부분에 타입이 있는 결과 객체가 필요할 때 커스텀 시작·종료 이벤트를 정의하세요.

출처: 공식문서 - Custom start and stop events

커스텀 StartEvent 사용하기

키워드 인자로 run()을 호출하면 워크플로가 그 인자들로 시작 이벤트를 만들어요. 기본 StartEvent에서는 어떤 추가 필드든 받아들여집니다.

result = await workflow.run(topic="pirates")

작은 입력에는 편리하죠. 프로덕션 코드라면 커스텀 StartEvent가 진입점에 진짜 스키마를 제공하고, 첫 스텝이 실행되기 전에 Pydantic이 누락되거나 잘못된 입력을 검증하게 해 줍니다.

StartEvent를 상속하는 커스텀 클래스를 만드세요.

from workflows.events import StartEvent


class JokeStartEvent(StartEvent):
    topic: str
    tone: str = "funny"

그다음 워크플로를 시작하는 스텝에서 그 이벤트 타입을 사용합니다.

class JokeFlow(Workflow):
    @step
    async def generate_joke(self, ev: JokeStartEvent) -> JokeEvent:
        prompt = f"Write a {ev.tone} joke about {ev.topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

여전히 필드를 키워드 인자로 넘길 수 있어요.

w = JokeFlow(timeout=60)
result = await w.run(topic="pirates", tone="dry")

더 큰 입력 객체라면 start_event를 통해 이벤트 인스턴스를 넘기세요.

start_event = JokeStartEvent(topic="pirates", tone="dry")
w = JokeFlow(timeout=60)
result = await w.run(start_event=start_event)

이벤트는 직렬화 가능한 데이터에 쓰세요. 워크플로가 LLM 클라이언트, 인덱스, DB 커넥션, 파일 핸들을 필요로 한다면 시작 이벤트에 넣지 말고 리소스로 주입하세요. 시작 이벤트는 워크플로를 스냅샷하거나 서버로 제공할 때 직렬화될 수 있는데, 무거운 런타임 객체는 보통 직렬화할 수 없어요.

커스텀 StopEvent 사용하기

내장 StopEvent는 result에 넣은 것을 그대로 돌려줍니다.

return StopEvent(result={"critique": critique, "score": score})

빠른 워크플로에는 괜찮지만, 결과 타입이 Any로 잡혀요. 커스텀 종료 이벤트는 출력 형태를 명시적으로 만들어 줍니다.

StopEvent의 서브클래스를 만드세요.

from workflows.events import StopEvent


class JokeResult(StopEvent):
    joke: str
    critique: str

이제 워크플로에서 StopEvent를 JokeResult로 바꿀 수 있어요.

class JokeFlow(Workflow):
    ...

    @step
    async def critique_joke(self, ev: JokeEvent) -> JokeResult:
        prompt = f"Give a thorough analysis and critique of the following joke: {ev.joke}"
        response = await self.llm.acomplete(prompt)
        return JokeResult(joke=ev.joke, critique=str(response))

스텝이 기본 StopEvent를 반환하면 await workflow.run(...)은 stop_event.result를 돌려줘요. 스텝이 커스텀 StopEvent 서브클래스를 반환하면 await workflow.run(...)은 이벤트 인스턴스 자체를 돌려줍니다.

w = JokeFlow(timeout=60)
result = await w.run(topic="pirates")
print(result.joke)
print(result.critique)

그 덕분에 결과가 타입 체커, 에디터 자동완성, 워크플로 스키마를 검사하는 호출자에게 친숙해져요.

더 알아보기