Claude API 오케스트레이션 패턴 5가지: Python 워크플로우 자동화 실전 구현

2026.04.09 · 36분 읽기

목차

import anthropic

client = anthropic.Anthropic()

# 1단계: 사용자 요청 분류
classify = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=50,
    messages=[{"role": "user", "content": req}],
    system="분류: bug_report | feature_request | question"
)

# 2단계: 분류 결과에 따라 다른 프롬프트 실행
handler = handlers[classify.content[0].text.strip()]
result = handler(req)

이 코드가 Claude API 워크플로우 자동화의 가장 기본 형태다. 사용자 요청을 분류하고, 분류 결과에 따라 다른 처리 로직을 태운다. 단순해 보이지만 이 구조 하나로 팀 내 반복 업무 상당수를 자동화했다.

프론트엔드에서 백엔드로 넘어온 지 3년 됐는데, Claude API 워크플로우 자동화를 본격적으로 쓰기 시작한 건 올해 초부터다. 처음엔 단일 API 호출로 끝나는 간단한 작업만 했는데, 팀에서 처리하는 요청이 복잡해지면서 여러 호출을 엮는 패턴이 필요해졌다. Anthropic 공식 문서에서 소개하는 오케스트레이션 패턴을 하나씩 구현해보면서 정리한 내용이다.

오케스트레이션 패턴이 필요한 이유

LLM API 호출 한 번으로 끝나는 작업은 사실 많지 않다. 실무에서 자동화하려는 워크플로우는 대부분 여러 단계로 나뉜다.

고객 문의가 들어오면 → 분류하고 → 분류에 맞는 처리를 하고 → 결과를 검증한다. 코드 리뷰를 자동화하려면 → 변경된 파일을 파싱하고 → 각 파일을 분석하고 → 결과를 합친다. 이런 흐름을 하나의 프롬프트에 다 넣으면 프롬프트가 비대해지고, 중간에 뭐가 잘못됐는지 디버깅이 안 된다.

에이전틱 시스템 vs 워크플로우
Anthropic은 LLM이 도구를 직접 호출하며 자율적으로 동작하는 걸 “에이전틱 시스템”, 코드로 LLM 호출 순서를 제어하는 걸 “워크플로우”로 구분한다. 이 글에서 다루는 건 후자다.
그래서 나온 게 오케스트레이션 패턴이다. 각 단계를 독립적인 LLM 호출로 분리하고, 그 호출들을 코드로 엮는 방식 ([Claude 시스템 프롬프트 설계 패턴](/claude-system-prompt-design-patterns-guide) 참고). Anthropic 공식 블로그에서 5가지 패턴을 소개하는데, 하나씩 Python으로 구현해보겠다.

패턴 1: 프롬프트 체이닝 — 순차 파이프라인

가장 직관적인 패턴이다. A의 출력을 B의 입력으로 넣는다. 그게 끝이다.

import anthropic
import json

client = anthropic.Anthropic()

def chain_translate_and_summarize(text: str) -> dict:
    """번역 → 요약 → 핵심 키워드 추출 체이닝"""
    
    # Step 1: 번역
    translation = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        system="영어 텍스트를 한국어로 번역한다. 번역문만 출력.",
        messages=[{"role": "user", "content": text}]
    )
    translated = translation.content[0].text
    
    # Step 2: 요약 (번역 결과를 입력으로)
    summary = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        system="주어진 텍스트를 3문장으로 요약한다.",
        messages=[{"role": "user", "content": translated}]
    )
    summarized = summary.content[0].text
    
    # Step 3: 키워드 추출
    keywords = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=200,
        system="텍스트에서 핵심 키워드 5개를 JSON 배열로 출력한다.",
        messages=[{"role": "user", "content": summarized}]
    )
    
    return {
        "translation": translated,
        "summary": summarized,
        "keywords": json.loads(keywords.content[0].text)
    }

체이닝의 핵심: 게이트 검증

단순히 연결만 하면 중간에 이상한 출력이 나와도 다음 단계로 넘어간다. 그래서 각 단계 사이에 검증 로직을 넣는 게 좋다.

def gate_check(output: str, criteria: str) -> bool:
    """중간 출력물 품질 검증"""
    check = client.messages.create(
        model="claude-haiku-3-5-20241022",
        max_tokens=10,
        system=f"다음 기준을 충족하면 'PASS', 아니면 'FAIL'만 출력: {criteria}",
        messages=[{"role": "user", "content": output}]
    )
    return check.content[0].text.strip() == "PASS"

검증에는 claude-haiku-3-5-20241022 같은 가벼운 모델을 쓴다. 비용도 아끼고 속도도 빠르다. 개인적으로 게이트 체크 하나 넣는 것만으로 후속 단계의 품질이 확 올라가더라.

패턴 2: 라우팅 — 요청을 적절한 핸들러로 분기

서두에 보여준 코드가 바로 이 패턴이다. 입력을 분류하고, 분류 결과에 따라 다른 프롬프트/모델을 태운다.

from typing import Callable

def create_router(categories: dict[str, Callable]) -> Callable:
    """분류 기반 라우터 생성"""
    
    category_list = " | ".join(categories.keys())
    
    def route(user_input: str) -> str:
        # 분류 단계 - 가벼운 모델로
        classification = client.messages.create(
            model="claude-haiku-3-5-20241022",
            max_tokens=50,
            system=f"사용자 입력을 다음 중 하나로 분류: {category_list}\n카테고리명만 출력.",
            messages=[{"role": "user", "content": user_input}]
        )
        category = classification.content[0].text.strip()
        
        if category not in categories:
            return categories.get("fallback", lambda x: "분류 실패")(user_input)
        
        return categories[category](user_input)
    
    return route

# 사용 예시
def handle_bug(text):
    resp = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        system="버그 리포트를 분석하고 재현 단계, 예상 원인, 우선순위를 정리한다.",
        messages=[{"role": "user", "content": text}]
    )
    return resp.content[0].text

router = create_router({
    "bug_report": handle_bug,
    "feature_request": handle_feature,
    "question": handle_question,
    "fallback": handle_general,
})

라우팅에서 자주 하는 실수

분류 프롬프트에서 카테고리 설명을 너무 길게 넣으면 오히려 분류 정확도가 떨어진다. "bug_report: 소프트웨어 오작동, 에러 발생, 예상과 다른 동작…" 이런 식으로 설명을 붙이고 싶겠지만, 카테고리명만 나열하는 게 낫다. 모델이 이미 맥락을 잘 잡는다.

라우팅 모델 선택
분류 작업은 Haiku급으로 충분하다. 실제 처리는 Sonnet이나 Opus를 쓰더라도 라우팅 자체는 가벼운 모델로 돌리면 비용을 크게 줄일 수 있다. 분류 정확도가 95% 이상 나온다.
## 패턴 3: 병렬화 — 독립 작업 동시 실행

여러 작업이 서로 의존하지 않으면 동시에 돌리는 게 당연하다. Python asyncio와 Anthropic 비동기 클라이언트를 쓰면 된다.

import asyncio

async_client = anthropic.AsyncAnthropic()

async def parallel_analysis(code: str) -> dict:
    """코드를 여러 관점에서 동시에 분석"""
    
    analyses = {
        "security": "보안 취약점을 분석한다. OWASP Top 10 기준으로 체크.",
        "performance": "성능 이슈를 분석한다. 시간 복잡도, 메모리 사용량 중심.",
        "readability": "가독성을 분석한다. 네이밍, 구조, 주석 관점에서.",
    }
    
    async def analyze(aspect: str, system_prompt: str) -> tuple[str, str]:
        resp = await async_client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            system=system_prompt,
            messages=[{"role": "user", "content": code}]
        )
        return aspect, resp.content[0].text
    
    tasks = [analyze(k, v) for k, v in analyses.items()]
    results = await asyncio.gather(*tasks)
    
    return dict(results)

# 실행
result = asyncio.run(parallel_analysis(some_code))

순차로 3번 호출하면 각 호출이 2~3초씩 걸려서 총 6~9초. 병렬로 돌리면 가장 느린 호출 시간인 2~3초면 끝난다. 체감 차이가 크다.

병렬화의 변형: 투표(Voting)

같은 프롬프트를 여러 번 돌려서 다수결로 결과를 정하는 것도 병렬화의 일종이다.

async def vote(prompt: str, n: int = 3) -> str:
    """같은 질문을 n번 돌려서 가장 많이 나온 답 선택"""
    
    async def single_call():
        resp = await async_client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=100,
            temperature=0.7,
            messages=[{"role": "user", "content": prompt}]
        )
        return resp.content[0].text.strip()
    
    results = await asyncio.gather(*[single_call() for _ in range(n)])
    
    from collections import Counter
    most_common = Counter(results).most_common(1)[0][0]
    return most_common

temperature를 올려서 다양한 응답을 받고, 가장 많이 나온 걸 채택한다. 분류나 판단 작업에서 정확도를 높이는 트릭이다. 다만 비용이 n배로 뛰니까 정말 정확도가 중요한 경우에만 쓴다.

패턴 4: 오케스트레이터-워커

이게 가장 복잡하지만 실무에서 제일 많이 쓰는 패턴이다. 중앙 오케스트레이터가 작업을 분해하고, 각 워커에게 할당한 뒤, 결과를 수집해서 합친다.

import json

def orchestrate_code_review(pr_diff: str) -> str:
    """PR diff를 분석하여 코드 리뷰 수행"""
    
    # 오케스트레이터: 작업 분해
    plan = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1000,
        system="""PR diff를 보고 리뷰할 작업 목록을 JSON으로 출력한다.
형식: {"tasks": [{"file": "파일명", "focus": "리뷰 포인트"}]}""",
        messages=[{"role": "user", "content": pr_diff}]
    )
    
    tasks = json.loads(plan.content[0].text)["tasks"]
    
    # 워커: 각 파일별 리뷰 수행
    reviews = []
    for task in tasks:
        review = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1500,
            system=f"다음 관점에서 코드를 리뷰한다: {task['focus']}",
            messages=[{"role": "user", "content": f"파일: {task['file']}\n\n{pr_diff}"}]
        )
        reviews.append(f"### {task['file']}\n{review.content[0].text}")
    
    # 오케스트레이터: 결과 통합
    synthesis = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=2000,
        system="개별 코드 리뷰 결과를 종합하여 하나의 리뷰 코멘트로 정리한다.",
        messages=[{"role": "user", "content": "\n\n".join(reviews)}]
    )
    
    return synthesis.content[0].text
오케스트레이터 출력 파싱 주의
오케스트레이터가 JSON을 출력하도록 했는데, 가끔 마크다운 코드블록으로 감싸서 내보내는 경우가 있다. `json.loads()` 전에 코드블록 마커를 제거하는 처리를 넣어두는 게 안전하다.
사실 이 패턴을 처음 구현했을 때 워커 호출을 순차로 돌렸다. 당연히 느렸다. 위 코드에서 워커 부분을 `asyncio.gather`로 바꾸면 병렬화 패턴과 합쳐져서 속도가 확 줄어든다. 패턴을 조합하는 게 실무에서는 자연스럽다.

패턴 5: 평가자-최적화 — 자동 품질 개선 루프

이건 좀 재밌는 패턴이다. 생성자가 결과물을 만들고, 평가자가 피드백을 주고, 피드백을 반영해서 다시 만든다. 이걸 품질이 기준을 넘을 때까지 반복한다.

def evaluate_and_optimize(
    task: str, 
    max_iterations: int = 3,
    quality_threshold: float = 0.8
) -> str:
    """생성 → 평가 → 개선 루프"""
    
    current_output = ""
    
    for i in range(max_iterations):
        # 생성 (첫 회차) 또는 개선 (2회차~)
        if i == 0:
            gen_prompt = task
        else:
            gen_prompt = f"원래 작업: {task}\n\n이전 결과:\n{current_output}\n\n피드백:\n{feedback}\n\n피드백을 반영하여 개선한다."
        
        generation = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{"role": "user", "content": gen_prompt}]
        )
        current_output = generation.content[0].text
        
        # 평가
        evaluation = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=500,
            system="""결과물을 평가한다. JSON으로 출력:
{"score": 0.0~1.0, "feedback": "구체적 개선 사항"}""",
            messages=[{"role": "user", "content": f"작업: {task}\n\n결과물:\n{current_output}"}]
        )
        
        eval_result = json.loads(evaluation.content[0].text)
        score = eval_result["score"]
        feedback = eval_result["feedback"]
        
        if score >= quality_threshold:
            break
    
    return current_output

![오케스트레이션 패턴 구조도]({{image:orchestration-patterns|Isometric 3D diagram showing five connected workflow nodes: chain, router, parallel, orchestrator-worker, evaluator-optimizer. Minimal flat design with blue gradient arrows connecting nodes on dark background}})

주의할 점은 max_iterations를 반드시 설정하는 거다. 안 그러면 평가자가 계속 까다로운 피드백을 주면서 무한루프에 빠질 수 있다. 실제로 이것 때문에 API 크레딧을 쓸데없이 날린 적이 있다. 3~5회가 적당하다.

Claude API 워크플로우 자동화 패턴 비교

5가지 패턴을 정리하면 이렇다.

패턴 복잡도 API 호출 수 적합한 작업 비용 효율
프롬프트 체이닝 낮음 2~5회 순차 번역→요약, 데이터 변환 파이프라인 높음
라우팅 낮음 2회 (분류+처리) 고객 문의 분류, 티켓 시스템 높음
병렬화 중간 N회 동시 다관점 분석, 투표 기반 판단 중간
오케스트레이터-워커 높음 2+N회 코드 리뷰, 문서 생성 낮음
평가자-최적화 높음 2×N회 반복 글 작성, 코드 생성 낮음

솔직히 대부분의 경우 프롬프트 체이닝이나 라우팅만으로 충분하다. Anthropic 공식 문서에서도 가능한 한 단순한 패턴부터 시작하라고 권한다. 오케스트레이터-워커나 평가자-최적화는 단순 패턴으로 안 되는 복잡한 작업에서만 꺼내 쓰는 게 맞다.

실전 적용: 팀 슬랙 봇에 라우팅 패턴 붙이기

비교만 해놓으면 와닿지 않으니까 실제로 쓸 법한 예제를 하나 더 잡아본다. 팀에서 슬랙으로 들어오는 요청을 자동 분류하고 처리하는 구조다.

from dataclasses import dataclass

@dataclass
class SlackMessage:
    user: str
    channel: str
    text: str

def build_slack_handler():
    """슬랙 메시지 처리 파이프라인 (라우팅 + 체이닝 조합)"""
    
    def handle_deploy_request(msg: SlackMessage) -> str:
        # 체이닝: 배포 요청 파싱 → 유효성 검증 → 배포 명령 생성
        parsed = client.messages.create(
            model="claude-haiku-3-5-20241022",
            max_tokens=200,
            system="""배포 요청에서 정보를 추출하여 JSON으로 출력:
{"service": "서비스명", "env": "staging|production", "branch": "브랜치명"}""",
            messages=[{"role": "user", "content": msg.text}]
        )
        
        deploy_info = json.loads(parsed.content[0].text)
        
        # production 배포는 추가 확인
        if deploy_info["env"] == "production":
            return f"⚠️ 프로덕션 배포 요청: {deploy_info['service']} ({deploy_info['branch']})\n승인자 멘션이 필요합니다."
        
        return f"✅ 배포 시작: {deploy_info['service']} → {deploy_info['env']} ({deploy_info['branch']})"
    
    def handle_oncall_question(msg: SlackMessage) -> str:
        resp = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1000,
            system="온콜 엔지니어가 참고할 수 있도록 문제 해결 가이드를 제공한다. 런북 형태로.",
            messages=[{"role": "user", "content": msg.text}]
        )
        return resp.content[0].text
    
    router = create_router({
        "deploy_request": lambda text: handle_deploy_request(
            SlackMessage(user="", channel="", text=text)
        ),
        "oncall_question": lambda text: handle_oncall_question(
            SlackMessage(user="", channel="", text=text)
        ),
        "fallback": lambda text: "이 요청은 자동 처리 대상이 아닙니다.",
    })
    
    return router

이건 라우팅과 체이닝을 합친 구조다. 분류 → 분기 → 각 분기 안에서 다시 체이닝. 패턴을 레고처럼 조합하는 게 Claude API 워크플로우 자동화의 핵심이다.

![워크플로우 라우팅 흐름]({{image:workflow-routing|Flowchart diagram showing message input splitting into three routes: deploy request, oncall question, and fallback. Each route has different processing steps. Clean minimal style with soft blue palette on white background}})

에러 핸들링과 비용 관리

코드 예제에선 에러 처리를 다 빼놨는데, 실제로 돌리면 여러 곳에서 터진다.

API 에러 재시도

import time
from anthropic import RateLimitError, APIStatusError

def call_with_retry(func, max_retries=3, base_delay=1.0):
    """지수 백오프 재시도"""
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError:
            delay = base_delay * (2 ** attempt)
            time.sleep(delay)
        except APIStatusError as e:
            if e.status_code >= 500:
                time.sleep(base_delay)
                continue
            raise
    raise Exception(f"{max_retries}회 재시도 실패")

RateLimitError가 제일 자주 나온다. 병렬화 패턴에서 동시에 10개씩 쏘면 거의 확실히 만난다. asyncio.Semaphore로 동시 요청 수를 제한하는 게 좋다.

비용 추적

def track_usage(response) -> dict:
    """API 응답에서 토큰 사용량 추출"""
    usage = response.usage
    # Claude Sonnet 기준 대략적인 비용 (2025년 기준)
    input_cost = usage.input_tokens * 3.0 / 1_000_000
    output_cost = usage.output_tokens * 15.0 / 1_000_000
    return {
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "estimated_cost_usd": round(input_cost + output_cost, 6)
    }
비용 절감 실전 팁
분류, 게이트 체크, 간단한 추출 작업은 Haiku로 돌려라. Sonnet 대비 토큰당 비용이 훨씬 싸다. 실제 “생각”이 필요한 단계만 Sonnet이나 Opus를 쓰면 된다.
오케스트레이터-워커 패턴에서 워커가 5개 돌아가면 API 호출이 최소 7회다(계획 1 + 워커 5 + 통합 1). 평가자-최적화에서 3회 반복하면 6회. 비용이 선형으로 느는 게 아니라 패턴 복잡도에 비례해서 뛴다. 반드시 비용 추적을 달아놓고 돌려야 한다.

Claude API 워크플로우 자동화, 어디까지 해봤나

graph LR
    A[단일 호출] --> B[프롬프트 체이닝]
    B --> C[라우팅]
    C --> D[병렬화]
    D --> E[오케스트레이터-워커]
    E --> F[평가자-최적화]
    
    style A fill:#e8f5e9
    style B fill:#e8f5e9
    style C fill:#fff3e0
    style D fill:#fff3e0
    style E fill:#fce4ec
    style F fill:#fce4ec

위 다이어그램에서 왼쪽이 단순, 오른쪽이 복잡하다. 대부분의 Claude API 워크플로우 자동화는 초록색 영역(단일 호출 ~ 체이닝)에서 시작하는 게 맞다. 노란색(라우팅, 병렬화)은 입력이 다양하거나 속도가 중요할 때. 빨간색(오케스트레이터, 평가자)은 정말 복잡한 작업에서만 필요하다.

![복잡도 단계별 자동화 아키텍처]({{image:automation-architecture|Abstract layered architecture diagram with three tiers: simple, intermediate, complex. Each tier shows connected blocks representing API call patterns. Gradient from green to red indicating complexity. Modern tech illustration style}})

개인적으로 가장 가성비 좋은 건 라우팅이다. 구현 난이도는 낮은데 활용 범위가 넓다. 팀 내 요청 분류, 고객 문의 라우팅, CI 파이프라인에서 실패 원인 분류 등. 한번 만들어두면 카테고리만 추가하면서 확장된다.

Anthropic의 공식 에이전트 빌딩 가이드에서 이런 패턴들을 더 깊이 다루고 있다. 특히 tool use와 결합하면 LLM이 외부 API를 직접 호출하는 에이전틱 시스템도 만들 수 있는데, 그건 워크플로우 자동화의 다음 단계다.

Python SDK 버전 참고
이 글의 코드는 `anthropic` Python SDK 0.52.x 기준이다. `pip install anthropic`으로 설치하면 최신 버전이 들어온다. 비동기 클라이언트(`AsyncAnthropic`)는 0.18 버전부터 지원된다.
Claude API 워크플로우 자동화는 결국 LLM 호출을 코드로 제어하는 거다. 프레임워크 없이 Python만으로 충분히 구현된다. 핵심을 정리하면 이렇다.

단순한 패턴부터 시작하라. 프롬프트 체이닝이나 라우팅으로 안 되는 문제인지 먼저 확인하고, 그래도 안 될 때만 복잡한 패턴을 꺼내라. 그리고 패턴은 조합해서 쓰는 게 자연스럽다. 라우팅 안에 체이닝을 넣고, 오케스트레이터 워커를 병렬로 돌리고. 마지막으로 비용 추적은 처음부터 달아라. 나중에 달겠다고 미루면 청구서 보고 놀란다.

여기서 다룬 건 코드로 흐름을 제어하는 워크플로우 패턴이다. 다음 단계로는 tool use를 붙여서 LLM이 직접 외부 시스템을 호출하는 에이전틱 패턴을 파볼 만하다. MCP(Model Context Protocol) 서버를 만들어서 Claude에 사내 API를 연결하는 것도 꽤 실용적이다. 프롬프트 엔지니어링 자체를 더 깊이 파고 싶다면 시스템 프롬프트 설계부터 잡는 게 순서다.

관련 글