커스텀 시작/종료 이벤트 — 워크플로 경계에 스키마 부여하기

커스텀 시작/종료 이벤트 — 워크플로 경계에 스키마 부여하기

대부분의 워크플로는 기본으로 제공되는 StartEvent와 StopEvent로 충분해요. 그런데 워크플로의 경계 자체가 의미 있는 스키마를 가질 때, 즉 시작 지점에 타입이 있는 요청 객체가 있거나 끝 지점에 타입이 있는 결과 객체가 있을 때는 커스텀 시작/종료 이벤트를 정의하면 돼요.

출처: 공식문서

커스텀 StartEvent 사용하기

기본 StartEvent는 작은 입력에 편리해요. 하지만 운영 코드에서는 커스텀 StartEvent가 진입점에 실제 스키마를 제공하고, Pydantic이 첫 스텝이 실행되기 전에 누락되거나 잘못된 입력을 검증해 줘요.

StartEvent를 상속받는 클래스를 만들면 됩니다.

from llama_index.core.workflow.events import StartEvent

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

이 이벤트 타입을 워크플로를 시작하는 스텝에서 사용하고, 실제 실행 때는 이렇게 줘요.

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

이벤트는 직렬화 가능한 데이터에 쓰세요. LLM 클라이언트·인덱스·DB 연결·파일 핸들 같은 무거운 런타임 객체가 필요하다면 시작 이벤트에 얹지 말고 Resource로 주입하세요. 시작 이벤트는 워크플로를 스냅샷하거나 서빙할 때 직렬화되는데, 무거운 런타임 객체는 보통 그럴 수 없거든요.

커스텀 StopEvent 사용하기

내장 StopEvent는 result에 넣은 값을 그대로 돌려줘요. 빠른 워크플로에는 그걸로 충분하지만, result의 타입은 Any라서 결과 형태가 명시적이지 않아요. 커스텀 stop 이벤트는 출력 형태를 분명하게 만들어 줍니다.

StopEvent의 서브클래스를 만들어요.

from llama_index.core.workflow.events import StopEvent

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

그리고 워크플로에서 StopEvent 대신 JokeResult를 돌려주면 돼요. 스텝이 기본 StopEvent를 돌려주면 await workflow.run(...)은 stop_event.result를 반환하지만, 커스텀 StopEvent 서브클래스를 돌려주면 이벤트 인스턴스 자체를 반환해요.

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))

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

타입이 있는 시작 이벤트와 종료 이벤트를 갖추면, 워크플로의 입출력 계약이 코드로 드러나요. 배포된 워크플로를 다른 팀이 호출할 때도 실수 여지가 줄어듭니다.

더 알아보기