비동기 워크플로 작성하기(Async Workflows)

비동기 워크플로 작성하기(Async Workflows)

워크플로는 Python의 asyncio 이벤트 루프 위에서 돌아요. 런타임은 협력적 멀티태스킹(cooperative multitasking)을 사용해서, 한 스텝이 await 하고 있는 동안(예: LLM 응답이나 네트워크 호출을 기다리는 동안) 다른 스텝과 워크플로가 진행할 수 있다는 뜻이에요.

Python의 async 프로그래밍이 처음이라면 먼저 Introduction to async Python에서 asyncio, 이벤트 루프, await에 대한 전반적인 개요를 읽어보세요.

스텝은 async def 또는 평범한 def로 정의할 수 있어요. 이 페이지에서는 두 방식이 각각 어떻게 동작하는지, 그리고 이벤트 루프를 막지 않고 블로킹 또는 CPU 집약 작업을 처리하는 방법을 설명합니다.

출처: 공식문서 - Writing async workflows

동기 스텝(async def 대신 def)

워크플로 스텝은 async def 대신 평범한 def 함수로 정의할 수 있어요. 런타임은 동기 스텝을 만나면 asyncio.get_event_loop().run_in_executor()로 함수 전체를 기본 스레드 풀에 자동으로 오프로드해서, 이벤트 루프가 절대 막히지 않게 합니다.

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


class SyncStepWorkflow(Workflow):
    @step
    def fetch_data(self, ev: StartEvent) -> StopEvent:
        # This runs in a thread automatically, so the event loop stays free
        response = requests.get("https://api.example.com/data")
        return StopEvent(result=response.json())

스텝 본문이 전적으로 동기일 때 가장 단순한 선택이에요. 프레임워크가 스레드 오프로딩을 처리하며 스레드 경계를 넘어 contextvars까지 보존해 줍니다.

하지만 async def 스텝 안에서 더 세밀한 제어가 필요한 경우가 있어요. 예를 들어 스텝의 일부만 블로킹일 때, 또는 CPU 집약 작업에 전용 실행기를 쓰고 싶을 때죠. 아래 섹션들이 그 시나리오를 다룹니다.

async 스텝에서의 블로킹 I/O

많은 Python 라이브러리는 동기 API만 제공해요. 데이터베이스 드라이버, HTTP 클라이언트, 파일 시스템 연산, SDK 호출 등이 그렇죠. async def 워크플로 스텝 안에서 이런 것들을 써야 한다면 asyncio.to_thread로 호출을 스레드 풀에 오프로드하세요.

import asyncio
import requests
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent


class BlockingIOWorkflow(Workflow):
    @step
    async def fetch_data(self, ev: StartEvent) -> StopEvent:
        # Bad: this blocks the event loop until the request completes
        # response = requests.get("https://api.example.com/data")

        # Good: run the blocking call in a thread so the event loop stays free
        response = await asyncio.to_thread(
            requests.get, "https://api.example.com/data"
        )
        return StopEvent(result=response.json())

asyncio.to_thread는 기본 ThreadPoolExecutor에 함수를 예약하고 awaitable을 돌려줘요. 블로킹 호출이 별도 스레드에서 실행되는 동안 이벤트 루프는 다른 스텝과 워크플로를 계속 처리합니다.

이 방법은 I/O를 수행하는 모든 동기 라이브러리 호출에 적용돼요. 파일 읽기, DB 질의, 외부 API 호출 등이 그렇죠.

import asyncio
import json
from pathlib import Path
from workflows import Workflow, step
from workflows.events import StartEvent, StopEvent


def read_large_file(path: str) -> dict:
    """A synchronous function that reads and parses a large JSON file."""
    return json.loads(Path(path).read_text())


class FileReaderWorkflow(Workflow):
    @step
    async def process_file(self, ev: StartEvent) -> StopEvent:
        data = await asyncio.to_thread(read_large_file, ev.file_path)
        return StopEvent(result=data)

CPU 집약 작업

CPU 바운드 작업(데이터 변환, 이미지 처리, 수치 계산 등)은 다른 문제를 제기해요. 스레드에서 실행하더라도 CPU 집약 Python 코드는 GIL(전역 인터프리터 락) 때문에 이벤트 루프와 경합할 수 있습니다.

CPU 집약 작업에는 전용의 작은 스레드 풀(또는 프로세스 풀)을 사용해서, 이 작업들이 큐에 쌓여 기본 실행기를 포화시키지 않게 하세요.

import asyncio
from concurrent.futures import ThreadPoolExecutor
from workflows import Workflow, step
from workflows.events import Event, StartEvent, StopEvent


# A small, dedicated pool for CPU-bound work.
# Keeping this small ensures CPU tasks are queued rather than
# overwhelming the system with parallel CPU-bound threads.
cpu_pool = ThreadPoolExecutor(max_workers=2)


def expensive_computation(data: str) -> str:
    """A CPU-intensive operation, e.g. data parsing or transformation."""
    # Simulate heavy work
    result = data
    for _ in range(1_000_000):
        result = result.strip()
    return result


class ComputeEvent(Event):
    data: str


class CPUWorkflow(Workflow):
    @step
    async def start(self, ev: StartEvent) -> ComputeEvent:
        return ComputeEvent(data=ev.input_data)

    @step
    async def compute(self, ev: ComputeEvent) -> StopEvent:
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(cpu_pool, expensive_computation, ev.data)
        return StopEvent(result=result)

I/O 경우와의 핵심 차이점은 다음과 같아요.

• loop.run_in_executor와 명시적 실행기를 쓰세요. asyncio.to_thread 대신 실행기의 크기와 종류를 제어할 수 있어요.
• 풀을 작게 유지하세요. 워커 1~2개의 풀은 CPU 작업이 CPU 시간을 놓고 경쟁하기보다는 큐에 쌓이게 만듭니다. 워크로드와 가용 코어에 맞게 조정하세요.
• GIL 밖에서 돌리려면 ProcessPoolExecutor를 고려하세요. API는 동일하며 실행기 종류만 바꾸면 됩니다.

from concurrent.futures import ProcessPoolExecutor


cpu_pool = ProcessPoolExecutor(max_workers=2)

ProcessPoolExecutor에 제출되는 함수는 피클 가능해야 한다는 점을 기억하세요(최상위 함수여야 하고, 람다나 클로저는 안 됩니다).

요약

핵심 원칙은 단순해요. asyncio 이벤트 루프를 절대 막지 마세요. 완전 동기 스텝은 평범한 def를 쓰고 프레임워크가 스레딩을 처리하게 두세요. async def 스텝 안의 블로킹 호출은 스레드나 프로세스로 오프로드하고 결과를 await 하면 됩니다.

더 알아보기

• 동시 실행(Concurrent Execution) — 여러 스텝을 병렬로 실행하는 패턴.
• 스트리밍 이벤트(Streaming events) — 실행 중 진행 상황을 실시간으로 노출하기.