실전 아키텍처 사례

def route_ticket(ticket: Ticket) -> str:
    """action 기준 라우팅 — 주제가 아닌 요청 행위로 분류"""
    intent = classify_intent(ticket.message)

    if intent.action in ("refund", "cancel", "account_modify"):
        return "human_review"          # write 행위는 자동화 제외
    if intent.confidence < 0.75:
        return "needs_triage"          # 낮은 확신은 사람에게
    if intent.action == "status_inquiry":
        return "auto_status"           # 조회는 바로 처리
    return "policy_retrieval"          # 나머지는 정책 검색 후 응답

THRESHOLDS = {
    "auto_respond": 0.90,   # 자동 응답 — 매우 높은 확신만
    "policy_lookup": 0.75,  # 정책 검색 경유 — 중간 확신
    "human_triage": 0.0,    # 그 외 전부 사람에게
}

def needs_escalation(session: Session) -> bool:
    return (
        session.sentiment_score < -0.6           # 고객 불만 감지
        or session.retry_count >= 3              # 반복 문의
        or session.involves_write_action         # 데이터 변경 행위
        or session.policy_version_mismatch       # 정책 버전 불일치
    )

#!/bin/bash
# evaluate.sh — Evaluator 단계에서 실행
set -euo pipefail

BRANCH="$1"
MAX_RETRIES=3
ATTEMPT=0

git checkout "$BRANCH"

while [ $ATTEMPT -lt $MAX_RETRIES ]; do
  echo "=== 검증 시도 $((ATTEMPT + 1))/$MAX_RETRIES ==="

  # 1. lint
  if ! yarn lint 2> lint_errors.log; then
    echo "LINT_FAIL: fix 요청"
    cat lint_errors.log | agent fix-lint
    ATTEMPT=$((ATTEMPT + 1))
    continue
  fi

  # 2. typecheck
  if ! yarn typecheck 2> type_errors.log; then
    echo "TYPE_FAIL: fix 요청"
    cat type_errors.log | agent fix-types
    ATTEMPT=$((ATTEMPT + 1))
    continue
  fi

  # 3. test
  if ! yarn test 2> test_errors.log; then
    echo "TEST_FAIL: fix 요청"
    cat test_errors.log | agent fix-tests
    ATTEMPT=$((ATTEMPT + 1))
    continue
  fi

  echo "ALL_PASS"
  exit 0
done

echo "MAX_RETRIES_EXCEEDED: human handoff"
exit 1

# watcher.py — git diff 기반 리뷰 트리거
def trigger_review(pr_number: int):
    changed_files = get_changed_files(pr_number)

    risk_score = 0
    for f in changed_files:
        if f.path.startswith("src/api/"):
            risk_score += 3          # API 변경은 고위험
        if f.additions > 200:
            risk_score += 2          # 대규모 변경
        if "migration" in f.path:
            risk_score += 5          # DB 마이그레이션

    if risk_score >= 5:
        assign_reviewer("senior-review-agent", pr_number)
        request_human_approval(pr_number)
    else:
        assign_reviewer("standard-review-agent", pr_number)

def aggregate_evidence(worker_results: list[EvidenceBundle]) -> Report:
    """worker 결과를 claim 단위로 병합하고 충돌을 표시"""
    claim_map: dict[str, list[Evidence]] = {}

    for bundle in worker_results:
        for evidence in bundle.items:
            claim_key = normalize_claim(evidence.claim)
            claim_map.setdefault(claim_key, []).append(evidence)

    report = Report()
    for claim, evidences in claim_map.items():
        if has_conflict(evidences):
            report.add_conflict(claim, evidences)      # 충돌은 명시적으로 표시
        else:
            merged = merge_evidences(evidences)
            report.add_claim(claim, merged)

    report.mark_gaps(find_uncovered_claims(claim_map))  # 빈 구간 표시
    return report

def resolve_conflict(claim: str, evidences: list[Evidence]) -> Evidence:
    """출처 신뢰도 가중치 기반 충돌 해결"""
    SOURCE_WEIGHTS = {
        "official_doc": 1.0,       # 공식 문서 — 최고 신뢰
        "peer_reviewed": 0.9,      # 학술 논문
        "industry_report": 0.7,    # 업계 리포트
        "blog_post": 0.4,          # 블로그
        "forum": 0.2,              # 포럼/커뮤니티
    }

    scored = []
    for ev in evidences:
        weight = SOURCE_WEIGHTS.get(ev.source_type, 0.3)
        recency_bonus = 0.1 if ev.published_within_months(6) else 0.0
        scored.append((ev, weight + recency_bonus))

    scored.sort(key=lambda x: x[1], reverse=True)

    # 최고 점수 근거를 채택하되, 충돌 사실은 보고서에 명시
    winner = scored[0][0]
    winner.conflict_note = f"{len(evidences)}개 출처 중 채택 (신뢰도 {scored[0][1]:.1f})"
    return winner

# dag_definition.py — 스테이지별 의존성과 정책 선언
pipeline = DAG("daily_user_etl", schedule="0 2 * * *")

extract = pipeline.add_stage(
    "extract",
    handler=extract_from_sources,
    retry=RetryPolicy(max_attempts=3, backoff="exponential", base_delay=30),
    timeout=600,
)

transform = pipeline.add_stage(
    "transform",
    handler=transform_records,
    depends_on=[extract],
    retry=RetryPolicy(max_attempts=2, backoff="fixed", base_delay=10),
)

enrich = pipeline.add_stage(
    "enrich",
    handler=enrich_with_external_api,
    depends_on=[extract],
    retry=RetryPolicy(max_attempts=3, backoff="exponential", base_delay=60),
    circuit_breaker=CircuitBreaker(fail_threshold=5, reset_timeout=300),
)

validate = pipeline.add_stage(
    "validate",
    handler=validate_quality,
    depends_on=[transform, enrich],
    retry=RetryPolicy(max_attempts=1),  # 검증은 재시도 불필요
)

load = pipeline.add_stage(
    "load",
    handler=load_to_warehouse,
    depends_on=[validate],
    retry=RetryPolicy(max_attempts=2, backoff="fixed", base_delay=5),
    idempotent=True,  # 중복 실행 안전
)

def handle_partial_failure(run: PipelineRun):
    """실패 스테이지만 재실행 — 성공 스테이지는 checkpoint에서 복원"""
    failed_stages = [s for s in run.stages if s.status == "failed"]
    succeeded_stages = [s for s in run.stages if s.status == "completed"]

    for stage in failed_stages:
        if stage.name == "enrich" and stage.circuit_breaker.is_open:
            # 외부 API 차단 상태 → enrich 건너뛰고 부분 데이터로 진행
            run.mark_degraded(stage, reason="circuit_breaker_open")
            continue

        if stage.retry_exhausted:
            # 재시도 소진 → dead-letter queue로 이동, 사람 알림
            dead_letter.push(stage.failed_records)
            notify_oncall(f"{stage.name} 재시도 소진, {len(stage.failed_records)}건 DLQ 이동")
            continue

        # checkpoint에서 마지막 성공 지점 복원 후 재실행
        stage.restore_checkpoint()
        stage.execute()