워크플로(Workflows)란 무엇인가

워크플로(Workflows)란 무엇인가

워크플로는 애플리케이션의 실행 흐름을 제어하는 이벤트 중심의 단계 기반 방식이에요. 애플리케이션을 여러 개의 **스텝(step)**으로 나누고, 각 스텝이 이벤트를 받아 일을 처리한 다음 다른 이벤트를 돌려주도록 설계하죠. 그 돌려받은 이벤트가, 타입 어노테이션상 그 이벤트를 받아들이는 다음 스텝을 실행시킵니다.

이것이 전부예요. 각 스텝은 LLM을 호출하거나, 검색을 수행하거나, 사람의 입력을 요청하거나, 공유 상태를 업데이트하거나, 작업 묶음을 분산시킬 수 있어요. 이벤트 타입이 워크플로의 간선(edge)을 정의하고, 각 간선 안쪽의 로직은 평범한 Python으로 표현됩니다.

출처: 공식문서 - Introduction to Workflows

왜 워크플로인가

생성형 AI 애플리케이션이 복잡해질수록 데이터의 흐름을 관리하고 실행을 제어하는 일이 어려워져요. 워크플로는 애플리케이션을 더 작고 다루기 쉬운 조각으로 쪼개 이 복잡성을 관리하게 해 줍니다.

이전에는 이런 문제를 DAG(방향성 비순환 그래프)로 풀려는 시도가 있었는데, 워크플로가 극복한 한계가 몇 가지 있어요.

  • 루프나 분기 같은 로직을 그래프의 간선에 인코딩해야 해서 읽고 이해하기 어려웠어요.
  • DAG의 노드 사이에 데이터를 전달할 때 optional 값, 기본값, 어떤 파라미터를 넘길지가 복잡해졌죠.
  • 복잡하고 반복적이며 분기하는 AI 애플리케이션을 만드는 개발자에게 DAG는 자연스럽지 않았어요.

이벤트 기반 패턴과 평범한 Python을 쓰는 워크플로 방식이 이 문제를 해결합니다. 분기는 서로 다른 이벤트 타입을 돌려주는 보통의 if 문이고, 루프는 앞선 스텝이 처리할 이벤트를 돌려주는 스텝이에요. 동시 작업은 list[Event]를 돌려주는 스텝과 그 리스트를 받는 스텝이 짝을 이룹니다. 흐름이 동적으로 변해야 할 때는 Context에서 직접 이벤트를 보낼 수도 있어요.

참고로, 워크플로 라이브러리는 pip install llama-index-workflows로 단독 설치할 수 있지만, llama-index-core에는 워크플로가 이미 포함되어 있어요. llama-index-core나 llama-index를 설치하면 llama_index.core.workflow 경로로 워크플로를 쓸 수 있습니다.

시작하기

워크플로는 async를 일급 시민으로 다루기 때문에, 이 문서는 async 환경에서 실행 중이라고 가정해요. 이미 FastAPI 같은 서버나 노트북에서 작업하고 있다면 await를 자유롭게 쓸 수 있습니다. 직접 Python 스크립트를 실행한다면 async 진입점을 하나 두는 게 관례예요.

async def main():
    w = MyWorkflow(...)
    result = await w.run(...)
    print(result)


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

가장 작고 유용한 형태는 이렇습니다. 뭔가를 생성해서 다른 스텝에 넘기고, 결과를 내면서 멈추는 형태예요.

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

# `pip install llama-index-llms-openai` 필요
from llama_index.llms.openai import OpenAI


class JokeEvent(Event):
    joke: str


class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4.1")

    @step
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        topic = ev.topic
        prompt = f"Write your best joke about {topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

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


w = JokeFlow(timeout=60, verbose=False)
result = await w.run(topic="pirates")
print(str(result))

여기에는 몇 가지 움직이는 조각이 있으니 하나씩 살펴볼게요.

워크플로 이벤트 정의하기

class JokeEvent(Event):
    joke: str

이벤트는 사용자가 정의하는 Pydantic 객체예요. 속성과 부가 메서드를 직접 제어합니다. 이 예에서는 JokeEvent 하나만 쓰고 있어요.

워크플로 클래스 만들기

class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4.1")
    ...

워크플로는 Workflow 클래스를 상속해 구현합니다. 편의상 정적 OpenAI LLM 인스턴스를 붙여 뒀어요.

워크플로 진입점

@step
async def generate_joke(self, ev: StartEvent) -> JokeEvent:
    topic = ev.topic
    prompt = f"Write your best joke about {topic}."
    response = await self.llm.acomplete(prompt)
    return JokeEvent(joke=str(response))

여기가 워크플로의 진입점이에요. 대부분의 이벤트는 사용자가 정의하지만 예외로 두 가지, StartEvent와 StopEvent는 프레임워크가 기본 제공합니다. StartEvent는 초기 워크플로 입력이 어디로 가는지를 나타내요.

StartEvent는 임의의 속성을 담을 수 있는 특별한 객체예요. 위에서 ev.topic으로 접근했는데, 없으면 오류가 나겠죠. 속성이 없을 수도 있는 상황을 오류 없이 처리하려면 ev.get("topic")을 쓸 수도 있어요. 더 타입 안전하게 하려면 StartEvent를 직접 상속하는 방법도 있습니다.

혹시 눈치챘을지 모르지만, 어떤 스텝이 어떤 이벤트를 처리하는지 명시적으로 알려주지 않았어요. 대신 @step 데코레이터가 각 스텝의 입출력 타입을 추론합니다. 이 추론된 타입 덕분에 실행 전에 워크플로가 유효한지 검증해 주기도 해요.

워크플로 종료점

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

여기가 워크플로의 두 번째이자 마지막 스텝이에요. 특별한 StopEvent를 돌려주기 때문에 마지막 스텝임을 알 수 있죠. 워크플로가 StopEvent를 만나면 즉시 멈추고 result 파라미터에 넘긴 값을 돌려줍니다. 이 경우 결과는 문자열이지만 딕셔너리, 리스트, 다른 어떤 객체도 될 수 있어요. 타입 안전성을 위해 StopEvent 클래스를 상속할 수도 있습니다.

워크플로 실행하기

w = JokeFlow(timeout=60, verbose=False)
result = await w.run(topic="pirates")
print(str(result))

마지막으로 워크플로를 만들고 실행해요. 타임아웃(초 단위)이나 verbosity 같은 디버깅용 설정이 있습니다.

.run() 메서드는 워크플로를 시작하고 WorkflowHandler를 돌려줘요. 핸들러는 await 가능하므로 await w.run(...)는 최종 결과를 기다립니다. 실행 중 스트리밍 이벤트가 필요하다면 핸들러를 붙잡아 두세요.

handler = w.run(topic="pirates")
async for ev in handler.stream_events():
    ...
result = await handler

run()에 넘긴 키워드 인자는 워크플로를 시작하기 위해 자동으로 방출되는 특별한 StartEvent의 필드가 됩니다. 위 예에서 topic은 ev.topic으로 스텝에서 접근했죠.

올바른 워크플로 API 고르기

대부분의 워크플로 코드는 타입이 지정된 스텝 입력과 반환을 사용하는 게 좋아요. 검증, 다이어그램, 읽기 쉬운 코드를 얻을 수 있으니까요.

@step
async def retrieve(self, ev: StartEvent) -> Retrieved:
    ...

@step
async def synthesize(self, ev: Retrieved) -> StopEvent:
    ...

작업의 형태에 따라 다른 API를 쓰면 됩니다.

이걸 써라 언제
list[A] 반환 스텝이 유한한 배치를 갖고, 다운스트림 워커가 시작되기 전에 작업을 모두 만들 수 있을 때
list[A] 수신 스텝이 계속 진행되기 전에 결과 전체 배치가 필요할 때
ctx.send_event(...) 이벤트를 점진적으로, 또는 개수를 미리 모르는 채로 방출하거나, 실행 중 워크플로 밖에서 이벤트를 보낼 때
ctx.collect_events(...) ctx.send_event를 썼고, 알려진 이벤트 집합을 직접 기다려야 할 때
ctx.store 스텝들이 실행마다 공유하는 상태가 필요할 때
Resource(...) 직렬화된 상태에 두면 안 되는 클라이언트·인덱스·모델·설정 같은 의존성이 필요할 때

타입 우선 API는 검증하고 시각화하기 쉬워요. 컨텍스트 API는 더 유연하지만 그만큼 북키핑을 직접 관리해야 합니다.

검증

워크플로가 실행되기 전에, 스텝 시그니처가 기술하는 이벤트 그래프를 검증해요. 시작·종료 이벤트가 있는지, 생성된 이벤트에 소비자가 있는지, 소비되는 이벤트에 생성자가 있는지, 그래프에 우발적인 막다른 길이 없는지 확인하죠.

대부분의 검증 실패는 유용한 설계 피드백입니다. 보통 다음 중 하나를 의미해요.

증상 주원인
스텝이 절대 생성되지 않는 이벤트를 소비함 반환 어노테이션이 빠졌거나, ctx.send_event로만 동적으로 보내는 이벤트일 때
스텝이 아무도 소비하지 않는 이벤트를 생성 다음 스텝의 이벤트 타입이 틀렸거나, 분기가 끝나지 않았을 때
워크플로에 종료 이벤트가 없음 도달 가능한 스텝 중 StopEvent(또는 그 서브클래스)를 반환하는 게 없을 때

의도적으로 동적인 워크플로에서는 가능한 한 타입 어노테이션에 담고, 타이밍만 ctx.send_event로 처리하세요. 어떤 스텝이 정적으로 도달 불가능하도록 의도했다면 그 스텝의 검사를 건너뛸 수 있어요.

@step(skip_graph_checks=["reachability"])
async def receive_webhook(self, ev: WebhookEvent) -> StopEvent:
    ...

테스트나 시작 코드에서 workflow.validate()를 직접 호출할 수도 있습니다. 리소스 설정 파일은 기본적으로 검증되고, 리소스 팩토리는 validate(validate_resources=True)를 호출할 때만 풀립니다.

더 알아보기

워크플로 개념에 익숙해지고 싶다면 공식 문서의 예제 노트북을 추천해요.

  • Common Workflow Patterns — 단순한 워크플로로 루프와 상태 관리 같은 흔한 패턴을 익힐 수 있어요.
  • RAG + Reranking — 수집과 질의를 모두 수행하는 꽤 단순한 워크플로로 실제 유스케이스를 구현해요.
  • Citation Query Engine — 검색과 생성 사이에 중간 단계를 넣는 방법을 보여줘요.
  • Corrective RAG — RAG 워크플로 위에 평가 단계 이후 웹 검색을 호출하는 복잡성을 더해요.
  • Utilizing Concurrency — 워크플로에서 스텝의 병렬 실행을 다루는 법을 설명해요.

더 복잡한 에이전트 시나리오(도구 호출, 메모리, 라우팅)에서도 워크플로가 빛을 발합니다. ReAct Agent, Function Calling Agent, Human In The Loop, Durable Workflows 등 다양한 예제가 공식 문서에 준비되어 있어요.