AI 에이전트로 인프라 비용 10배 줄이기 — Meta Capacity Efficiency 패턴으로 배우는 멀티에이전트 성능 최적화

# 개념적 오케스트레이터 구조 예시 (LangGraph 스타일)
from langgraph.graph import StateGraph, END, START
from typing import TypedDict, Literal
 
class AgentState(TypedDict):
    task: str
    findings: list[str]
    pr_draft: str | None
    route: Literal["offense", "defense"]
 
def router(state: AgentState) -> Literal["offense_agent", "defense_agent"]:
    if state["route"] == "offense":
        return "offense_agent"
    return "defense_agent"
 
def offense_agent(state: AgentState) -> AgentState:
    # SLM으로 최적화 기회 탐색 — 비용 효율적
    opportunities = scan_for_optimization_opportunities(state["task"])
    return {**state, "findings": opportunities}
 
def defense_agent(state: AgentState) -> AgentState:
    # 회귀 감지 → PR 자동 생성
    regression = detect_performance_regression(state["task"])
    pr = generate_fix_pr(regression) if regression else None
    return {**state, "pr_draft": pr}
 
graph = StateGraph(AgentState)
graph.add_node("offense_agent", offense_agent)
graph.add_node("defense_agent", defense_agent)
 
# START 상수를 사용한 조건부 진입점 — 실제 동작하는 패턴
graph.add_conditional_edges(START, router)
graph.add_edge("offense_agent", END)
graph.add_edge("defense_agent", END)

# Inference Layering 라우팅 로직 예시
from enum import Enum
 
class TaskComplexity(Enum):
    SIMPLE = "simple"      # 로그 파싱, 형식 변환
    MODERATE = "moderate"  # 패턴 탐지, 요약
    COMPLEX = "complex"    # 근본 원인 분석, 코드 생성
 
def select_model(complexity: TaskComplexity) -> str:
    # 비용 비율: SIMPLE:MODERATE:COMPLEX ≈ 1:10:150 수준 (시점에 따라 변동)
    routing_table = {
        TaskComplexity.SIMPLE: "llama-3.2-3b",      # CPU 처리
        TaskComplexity.MODERATE: "llama-3.1-8b",     # GPU 처리
        TaskComplexity.COMPLEX: "claude-opus-4-7",   # 프론티어 모델
    }
    return routing_table[complexity]
 
def classify_task(task_description: str) -> TaskComplexity:
    keywords_complex = ["root cause", "generate PR", "architectural decision"]
    keywords_simple = ["parse log", "format", "count", "extract field"]
 
    if any(kw in task_description.lower() for kw in keywords_complex):
        return TaskComplexity.COMPLEX
    elif any(kw in task_description.lower() for kw in keywords_simple):
        return TaskComplexity.SIMPLE
    return TaskComplexity.MODERATE
 
# ⚠️ 주의: 키워드 매칭은 프로토타입 수준입니다.
# 실제 프로덕션에서는 경량 분류 모델을 파인튜닝하거나
# 임베딩 유사도 방식으로 대체하는 것이 훨씬 견고합니다.

# agents/regression_detector.py
import asyncio
import aiohttp
from dataclasses import dataclass
from anthropic import AsyncAnthropic
 
@dataclass
class PerformanceSnapshot:
    service: str
    p99_latency_ms: float
    throughput_rps: float
    error_rate: float
    timestamp: str
 
async def fetch_current_metrics(
    service_name: str,
    prometheus_url: str = "http://prometheus:9090"
) -> PerformanceSnapshot:
    """Prometheus에서 실시간 메트릭 수집"""
    queries = {
        "p99_latency": f'histogram_quantile(0.99, rate(http_request_duration_seconds_bucket{{service="{service_name}"}}[5m])) * 1000',
        "throughput": f'rate(http_requests_total{{service="{service_name}"}}[5m])',
        "error_rate": f'rate(http_requests_total{{service="{service_name}",status=~"5.."}}[5m]) / rate(http_requests_total{{service="{service_name}"}}[5m])',
    }
    results = {}
    async with aiohttp.ClientSession() as session:
        for key, query in queries.items():
            async with session.get(
                f"{prometheus_url}/api/v1/query",
                params={"query": query}
            ) as resp:
                data = await resp.json()
                value = data["data"]["result"]
                results[key] = float(value[0]["value"][1]) if value else 0.0
 
    return PerformanceSnapshot(
        service=service_name,
        p99_latency_ms=results["p99_latency"],
        throughput_rps=results["throughput"],
        error_rate=results["error_rate"],
        timestamp=asyncio.get_event_loop().time().__str__(),
    )
 
async def detect_regression(
    baseline: PerformanceSnapshot,
    current: PerformanceSnapshot,
    client: AsyncAnthropic
) -> dict:
    """
    1단계: 룰 기반 수치 탐지 (비용 무료)
    2단계: 회귀 확인 시에만 프론티어 모델 호출 (비용 최소화)
    """
    regression_signals = []
    if current.p99_latency_ms > baseline.p99_latency_ms * 1.2:
        regression_signals.append(
            f"P99 레이턴시 {baseline.p99_latency_ms:.1f}ms → {current.p99_latency_ms:.1f}ms (20% 초과)"
        )
    if current.error_rate > baseline.error_rate + 0.01:
        regression_signals.append(
            f"에러율 {baseline.error_rate:.2%} → {current.error_rate:.2%}"
        )
 
    if not regression_signals:
        return {"regression_detected": False}
 
    # 회귀 확정 후에만 프론티어 모델 호출
    analysis_prompt = f"""
    서비스 '{current.service}'에서 다음 성능 회귀가 감지되었습니다:
    {chr(10).join(regression_signals)}
 
    가능한 근본 원인과 즉각적인 수정 방향을 제안해주세요.
    응답 형식: {{"root_cause": "...", "fix_suggestion": "...", "severity": "low|medium|high"}}
    """
 
    response = await client.messages.create(
        model="claude-opus-4-7",
        max_tokens=1024,
        messages=[{"role": "user", "content": analysis_prompt}]
    )
 
    return {
        "regression_detected": True,
        "signals": regression_signals,
        "analysis": response.content[0].text
    }

# orchestrator/capacity_workflow.py
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from typing import TypedDict, Annotated
import operator
 
class CapacityState(TypedDict):
    service_name: str
    metrics: dict
    regression_report: str | None
    optimization_opportunities: list[str]
    generated_prs: list[str]
    errors: Annotated[list[str], operator.add]  # 에러는 누적 (덮어쓰기 방지)
 
async def metrics_collector(state: CapacityState) -> CapacityState:
    """Prometheus/CloudWatch에서 현재 메트릭 수집"""
    metrics = await fetch_current_metrics(state["service_name"])
    return {**state, "metrics": metrics}
 
async def regression_detector(state: CapacityState) -> CapacityState:
    """성능 회귀 감지 — SLM 또는 룰 기반"""
    report = await run_regression_analysis(state["metrics"])
    return {**state, "regression_report": report}
 
async def opportunity_scanner(state: CapacityState) -> CapacityState:
    """최적화 기회 스캔 — offense 에이전트"""
    opportunities = await scan_efficiency_opportunities(state["metrics"])
    return {**state, "optimization_opportunities": opportunities}
 
async def pr_generator(state: CapacityState) -> CapacityState:
    """회귀 수정 또는 최적화 PR 자동 생성"""
    prs = []
    if state["regression_report"]:
        prs.append(await generate_regression_fix_pr(state["regression_report"]))
    for opp in state["optimization_opportunities"]:
        prs.append(await generate_optimization_pr(opp))
    return {**state, "generated_prs": prs}
 
def should_generate_pr(state: CapacityState) -> str:
    has_work = state["regression_report"] or state["optimization_opportunities"]
    return "pr_generator" if has_work else END
 
# 그래프 조립
workflow = StateGraph(CapacityState)
workflow.add_node("metrics_collector", metrics_collector)
workflow.add_node("regression_detector", regression_detector)
workflow.add_node("opportunity_scanner", opportunity_scanner)
workflow.add_node("pr_generator", pr_generator)
 
workflow.set_entry_point("metrics_collector")
 
# 같은 소스 노드에서 두 엣지를 추가하면 LangGraph가 팬아웃(fan-out)으로 처리합니다.
# regression_detector와 opportunity_scanner가 동시에 실행됩니다.
workflow.add_edge("metrics_collector", "regression_detector")
workflow.add_edge("metrics_collector", "opportunity_scanner")
 
workflow.add_conditional_edges("regression_detector", should_generate_pr)
workflow.add_conditional_edges("opportunity_scanner", should_generate_pr)
workflow.add_edge("pr_generator", END)
 
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)