멀티에이전트 파이프라인 설계 — Claude Agent SDK 오케스트레이터·서브에이전트 상태 공유와 오류 전파 처리

Orchestrator
  ├── 태스크 분해
  ├── 서브에이전트 A 위임 ──▶ [독립 컨텍스트] ──▶ 결과 반환
  ├── 서브에이전트 B 위임 ──▶ [독립 컨텍스트] ──▶ 결과 반환
  └── 결과 종합

import anthropic
import asyncio
import json
from typing import TypedDict
 
# AsyncAnthropic 사용 — asyncio.gather로 병렬 실행 시 이벤트 루프를 블록하지 않습니다
async_client = anthropic.AsyncAnthropic()
 
# 비용이 낮은 모델로 먼저 시작해 필요에 따라 교체하는 것을 권장합니다
MODEL = "claude-haiku-4-5-20251001"
 
class PipelineState(TypedDict):
    task: str
    plan: str | None
    output: str | None
    evaluation: str | None
    passed: bool
 
async def run_subagent(system_prompt: str, user_message: str) -> str:
    """단일 서브에이전트 실행 — 결과 텍스트만 반환"""
    response = await async_client.messages.create(
        model=MODEL,
        max_tokens=4096,
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}],
    )
    return response.content[0].text
 
async def planner_agent(task: str) -> str:
    return await run_subagent(
        system_prompt="당신은 작업 구조를 설계하는 플래너입니다. 목표와 단계별 접근법을 명확히 서술하세요.",
        user_message=f"다음 태스크를 분석하고 실행 계획을 작성하세요:\n\n{task}",
    )
 
async def generator_agent(task: str, plan: str) -> str:
    return await run_subagent(
        system_prompt="당신은 실행 에이전트입니다. 주어진 계획대로 결과물을 생성하세요.",
        user_message=f"태스크: {task}\n\n계획:\n{plan}\n\n위 계획에 따라 결과물을 작성하세요.",
    )
 
async def evaluator_agent(task: str, output: str) -> tuple[bool, str]:
    result = await run_subagent(
        system_prompt=(
            "당신은 독립적인 품질 검증 에이전트입니다. "
            "결과물이 태스크 목표를 충족하는지 판단하고, "
            "반드시 다음 JSON 형식으로만 응답하세요:\n"
            '{"status": "pass" 또는 "fail", "reason": "판단 이유"}'
        ),
        user_message=f"태스크: {task}\n\n결과물:\n{output}\n\n품질을 검증하세요.",
    )
    try:
        data = json.loads(result)
        passed = data.get("status") == "pass"
    except json.JSONDecodeError:
        passed = False
    return passed, result
 
async def orchestrator(task: str) -> PipelineState:
    state: PipelineState = {
        "task": task,
        "plan": None,
        "output": None,
        "evaluation": None,
        "passed": False,
    }
 
    print("[오케스트레이터] 플래너 서브에이전트 실행 중...")
    state["plan"] = await planner_agent(task)
 
    print("[오케스트레이터] 제너레이터 서브에이전트 실행 중...")
    state["output"] = await generator_agent(task, state["plan"])
 
    print("[오케스트레이터] 평가자 서브에이전트 실행 중...")
    passed, evaluation = await evaluator_agent(task, state["output"])
    state["evaluation"] = evaluation
    state["passed"] = passed
 
    return state
 
async def main():
    result = await orchestrator(
        "Python으로 간단한 할 일 관리 CLI 도구를 설계하고 구현하세요."
    )
    print(f"\n최종 결과 — PASS: {result['passed']}")
 
asyncio.run(main())

import asyncio
import json
import time
from typing import Any
 
import anthropic
 
async_client = anthropic.AsyncAnthropic()
MODEL = "claude-haiku-4-5-20251001"
 
# 인메모리 KV 스토어 (실제 환경에서는 Redis 등으로 대체)
_store: dict[str, Any] = {}
 
def kv_set(key: str, value: Any) -> None:
    _store[key] = {"value": value, "timestamp": time.time()}
 
def kv_get(key: str) -> Any | None:
    entry = _store.get(key)
    return entry["value"] if entry else None
 
async def compliance_check_agent(run_id: str, domain: str, task: str) -> dict:
    """도메인별 컴플라이언스 검사 서브에이전트"""
    response = await async_client.messages.create(
        model=MODEL,
        max_tokens=512,
        system=(
            f"당신은 {domain} 컴플라이언스 전문가입니다. "
            "주어진 거래 데이터를 검토하고 이슈 여부를 판단하세요. "
            '반드시 JSON으로만 응답하세요: {"status": "pass" 또는 "fail", "issues": ["이슈 목록"]}'
        ),
        messages=[{"role": "user", "content": f"거래 데이터:\n{task}"}],
    )
 
    try:
        parsed = json.loads(response.content[0].text)
    except json.JSONDecodeError:
        parsed = {"status": "error", "issues": ["응답 파싱 실패"]}
 
    result = {
        "domain": domain,
        "status": parsed.get("status", "error"),
        "issues": parsed.get("issues", []),
        "checked_at": time.time(),
    }
    kv_set(f"{run_id}:compliance:{domain}", result)
    print(f"  [{domain}] 검사 완료 → KV 저장")
    return result
 
async def report_agent(run_id: str, domains: list[str]) -> str:
    """KV에서 결과를 수집해 최종 리포트 생성"""
    results = []
    for domain in domains:
        result = kv_get(f"{run_id}:compliance:{domain}")
        if result:
            results.append(result)
 
    passed = sum(1 for r in results if r["status"] == "pass")
    report = (
        f"컴플라이언스 검사 결과: {passed}/{len(results)} 통과\n"
        + "\n".join(f"- {r['domain']}: {r['status']}" for r in results)
    )
    return report
 
async def financial_compliance_orchestrator(transaction_data: str) -> str:
    run_id = f"run_{int(time.time())}"
    domains = ["AML", "KYC", "FATF"]
 
    print(f"[오케스트레이터] 병렬 컴플라이언스 검사 시작 (run_id={run_id})")
 
    # 독립적인 검사를 병렬로 실행
    await asyncio.gather(
        *[
            compliance_check_agent(run_id, domain, transaction_data)
            for domain in domains
        ]
    )
 
    print("[오케스트레이터] 리포트 에이전트 실행")
    return await report_agent(run_id, domains)
 
asyncio.run(financial_compliance_orchestrator("거래 데이터 샘플"))

import asyncio
import time
from enum import Enum
 
class CircuitOpenError(RuntimeError):
    """Circuit Breaker가 OPEN 상태일 때 발생하는 전용 예외"""
    pass
 
class CircuitState(Enum):
    CLOSED = "closed"        # 정상 운전
    OPEN = "open"            # 트립됨, 요청 차단
    HALF_OPEN = "half_open"  # 복구 탐색 중
 
class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time: float | None = None
        self.state = CircuitState.CLOSED
 
    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return False
        return time.time() - self.last_failure_time >= self.recovery_timeout
 
    async def call(self, coro):
        if self.state == CircuitState.OPEN:
            if self._should_attempt_reset():
                self.state = CircuitState.HALF_OPEN
                print("[Circuit Breaker] HALF_OPEN — 복구 시도 중")
            else:
                raise CircuitOpenError("Circuit Breaker OPEN: 요청 차단됨")
 
        try:
            result = await coro
            self.failure_count = 0
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                print("[Circuit Breaker] CLOSED — 복구 완료")
            return result
 
        except CircuitOpenError:
            raise
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
                print(
                    f"[Circuit Breaker] OPEN — {self.failure_count}회 실패, "
                    f"{self.recovery_timeout}초 후 재시도 가능"
                )
            raise e
 
async def agent_with_retry(
    agent_fn,
    breaker: CircuitBreaker,
    max_retries: int = 3,
    base_delay: float = 1.0,
):
    """지수 백오프 재시도 + Circuit Breaker 조합"""
    for attempt in range(max_retries):
        try:
            return await breaker.call(agent_fn())
        except CircuitOpenError:
            raise  # Circuit Breaker 트립은 재시도하지 않음
        except Exception:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"  재시도 {attempt + 1}/{max_retries}, {delay:.1f}초 대기")
            await asyncio.sleep(delay)