이벤트 스트리밍(Streaming events)

이벤트 스트리밍(Streaming events)

워크플로는 끝나는 데 시간이 걸리는 경우가 많아요. 느린 프로바이더를 호출하거나, 분기하거나, 배치에 팬아웃하거나, 사람의 응답을 기다리기도 하죠. 스트리밍을 쓰면 실행이 진행 중인 동안에도 진행 상황을 드러낼 수 있습니다.

스트리밍에는 두 면이 있어요.

방향 API
스텝 내부에서 ctx.write_event_to_stream(...)
워크플로 밖에서 handler.stream_events()

workflow.run(...)은 워크플로를 시작하고 WorkflowHandler를 돌려줘요. 핸들러는 최종 결과에 대해 await 가능하고, 해당 실행의 이벤트 스트림도 소유합니다.

출처: 공식문서 - Streaming events

필요한 의존성을 가져오겠습니다.

import asyncio
from llama_index.llms.openai import OpenAI

from workflows import (
    Workflow,
    Context,
    step,
)
from workflows.events import (
    StartEvent,
    StopEvent,
    Event,
)

간단한 3단계 워크플로를 위한 이벤트들과, 진행 상황을 스트리밍할 이벤트를 하나 만들게요.

class FirstEvent(Event):
    first_output: str


class SecondEvent(Event):
    second_output: str
    response: str


class ProgressEvent(Event):
    msg: str

그리고 이벤트를 보내는 워크플로 클래스를 정의합니다.

class MyWorkflow(Workflow):
    @step
    async def step_one(self, ctx: Context, ev: StartEvent) -> FirstEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="Step one is happening"))
        return FirstEvent(first_output="First step complete.")

    @step
    async def step_two(self, ctx: Context, ev: FirstEvent) -> SecondEvent:
        llm = OpenAI(model="gpt-4o-mini")
        generator = await llm.astream_complete(
            "Please give me the first 3 paragraphs of Moby Dick, a book in the public domain."
        )
        full_resp = ""
        async for response in generator:
            # Allow the workflow to stream this piece of response
            ctx.write_event_to_stream(ProgressEvent(msg=response.delta))
            full_resp += response.delta

        return SecondEvent(
            second_output="Second step complete, full response attached",
            response=full_resp,
        )

    @step
    async def step_three(self, ctx: Context, ev: SecondEvent) -> StopEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="Step three is happening"))
        return StopEvent(result="Workflow complete.")

참고로, OpenAI()는 환경에 OPENAI_API_KEY가 설정되어 있다고 가정합니다. api_key 파라미터로 직접 넘겨줄 수도 있어요.

step_one과 step_three에서는 이벤트 하나를 스트림에 씁니다. step_two에서는 astream_complete로 LLM 응답의 반복 가능한 생성기를 만든 뒤, 최종 응답을 step_three에 돌려주기 전에 LLM이 보내는 데이터 청크마다 이벤트 하나를 생성하죠.

실제로 이 출력을 얻으려면 워크플로를 비동기로 실행하고 이벤트를 들어야 합니다.

async def main():
    w = MyWorkflow(timeout=30, verbose=True)
    handler = w.run(first_input="Start the workflow.")

    async for ev in handler.stream_events():
        if isinstance(ev, ProgressEvent):
            print(ev.msg)

    final_result = await handler
    print("Final result", final_result)


if __name__ == "__main__":
    asyncio.run(main())

run은 워크플로를 백그라운드에 예약합니다. stream_events()는 스트림에 쓰인 모든 이벤트를 생성하며, 스트림이 StopEvent를 전달하면 멈춥니다. 그 후 핸들러를 await 해 최종 결과를 얻어요.

핸들러 스트림은 한 번만 소비할 수 있습니다. 워크플로 이벤트를 여러 클라이언트에 브로드캐스트해야 한다면 애플리케이션에서 스트림을 한 번 소비하고 그 이벤트들을 직접 팬아웃 하세요.

워크플로 종료 처리

워크플로가 비정상적으로 끝나면 await handler에서 예외가 발생하기 전에 특정 StopEvent 서브클래스가 스트림에 게시됩니다.

  • WorkflowTimedOutEvent — 워크플로가 타임아웃을 초과했을 때 게시됩니다. timeout(초)과 active_steps(실행 중이던 스텝 이름 리스트)를 담아요.
  • WorkflowCancelledEvent — 사용자가 워크플로를 취소했을 때 게시됩니다.
  • WorkflowFailedEvent — 재시도를 모두 소진한 뒤 스텝이 영구히 실패했을 때 게시됩니다. step_name, exception, attempts, elapsed_seconds를 담아요.
from workflows.events import (
    WorkflowTimedOutEvent,
    WorkflowCancelledEvent,
    WorkflowFailedEvent,
)


async for ev in handler.stream_events():
    if isinstance(ev, WorkflowTimedOutEvent):
        print(f"Workflow timed out after {ev.timeout}s")
    elif isinstance(ev, WorkflowCancelledEvent):
        print("Workflow was cancelled")
    elif isinstance(ev, WorkflowFailedEvent):
        print(f"Step '{ev.step_name}' failed after {ev.attempts} attempts: {ev.exception}")

더 알아보기