다중 에이전트 워크플로우 구축하기

워크플로우는 본질적으로 계산 그래프다 — 단계(노드)와 엣지(연결)를 담는 컨테이너. n8n이나 Flowise 같은 드래그 앤 드롭 AI 툴의 내부가 바로 이렇게 만들어진다.

이 그래프 기반 접근은 강력한 역량을 가능케 한다.

• 검증 — 런타임 이전에 실행 경로를 확인한다.
• 시각화 — 시스템 동작에 대한 명료한 표현.
• 결정론적 실행 — 예측 가능한 결과.
• 병렬 처리 — 의존성이 허용하는 한 동시 실행.
• 조건부 로직 — 단계 출력이나 공유 상태에 기반한 동적 라우팅.

# 단계들을 워크플로우로 조합
workflow = (Workflow(metadata=WorkflowMetadata(name="Math Pipeline"))
            .chain(double_step, add_ten_step))

# 스트리밍 관측 가능성과 함께 실행
runner = WorkflowRunner()
async for event in runner.run_stream(workflow, {"value": 3}):
    print(event)   # 3 → 6 → 16

단계는 워크플로우의 근본 빌딩 블록 — 입력 데이터를 출력 데이터로 변환하는 단일 계산 단위다. 모든 단계는 공통 베이스 BaseStep으로부터 확장된다.

class NumberInput(BaseModel):
    value: int

class NumberOutput(BaseModel):
    result: int

async def double_number(input_data: NumberInput,
                        context: Context) -> NumberOutput:
    return NumberOutput(result=input_data.value * 2)

double_step = FunctionStep(
    step_id="double",
    input_type=NumberInput,
    output_type=NumberOutput,
    func=double_number)

Pydantic 모델이 단계 사이를 흐르는 데이터의 계약을 정의한다. 워크플로우는 실행 전에 double_step의 NumberOutput이 add_ten_step의 입력 타입과 일치하는지 검증한다. BaseStep은 인프라 관심사(실행 생애 주기·상태 추적·오류 처리)를, FunctionStep 같은 구체적 구현은 도메인 로직을 다룬다.

엣지는 단계 간 제어 흐름·조건을 정의한다. 세 가지 패턴을 다룬다 — 순차, 조건부, 병렬.

• 조건부 엣지 — 출력 기반 조건은 검증·품질 관리 패턴을, 상태 기반 조건은 구성 주도 동작과 동적 적응을 가능케 한다.
• 병렬 실행 패턴 — DAG 러너는 자동 팬-인 감지를 사용한다. 동일 단계로 향하는 여러 엣지가 "always" 조건을 가질 때, 러너는 모든 의존성이 완료될 때까지 기다린다.

WorkflowRunner는 정교한 의존성 해결을 구현한다 — 팬인(모든 입력을 기다림)과 조건부(유효한 경로 중 어느 하나를 기다림) 로직. 자원 한계 내 동시 단계 실행, 교착 상태 탐지, 우아한 오류 복구를 처리하면서 디버깅·모니터링을 위한 스트리밍 관측 가능성을 제공한다.

워크플로우는 공유·버전 관리·UI 통합을 위해 JSON으로의 완전한 직렬화를 지원한다. 자동 체크포인팅은 실패 이후 실행 재개를 가능케 하며, 구조 해시 검증이 안전한 재개를 보장한다.