멀티 에이전트 협업 아키텍처 실전
설계 패턴부터 프로덕션까지（2026）

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class PipelineState(TypedDict):
    query: str
    retrieved_docs: str
    analysis: str
    final_report: str

def retrieval_agent(state: PipelineState):
    docs = search_knowledge_base(state["query"])
    return {"retrieved_docs": docs}

def analysis_agent(state: PipelineState):
    result = llm.invoke(f"다음 내용을 분석하세요: {state['retrieved_docs']}")
    return {"analysis": result.content}

def writer_agent(state: PipelineState):
    report = llm.invoke(f"분석 결과를 바탕으로 보고서를 작성하세요: {state['analysis']}")
    return {"final_report": report.content}

builder = StateGraph(PipelineState)
builder.add_node("retriever", retrieval_agent)
builder.add_node("analyzer", analysis_agent)
builder.add_node("writer", writer_agent)
builder.add_edge(START, "retriever")
builder.add_edge("retriever", "analyzer")
builder.add_edge("analyzer", "writer")
builder.add_edge("writer", END)
pipeline = builder.compile()

from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ResearchState(TypedDict):
    query: str
    research_results: Annotated[list, operator.add]
    final_synthesis: str

def supervisor(state: ResearchState):
    subtasks = [
        {"query": state["query"], "source": "academic"},
        {"query": state["query"], "source": "industry"},
        {"query": state["query"], "source": "news"},
    ]
    return [Send("research_worker", task) for task in subtasks]

def research_worker(state: dict):
    result = search_by_source(state["query"], state["source"])
    return {"research_results": [result]}

def synthesizer(state: ResearchState):
    combined = "\n".join(state["research_results"])
    synthesis = llm.invoke(f"다음 연구 결과를 종합하세요: {combined}")
    return {"final_synthesis": synthesis.content}

builder = StateGraph(ResearchState)
builder.add_node("research_worker", research_worker)
builder.add_node("synthesizer", synthesizer)
builder.add_conditional_edges(START, supervisor, ["research_worker"])
builder.add_edge("research_worker", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()

KEYWORD_ROUTING = {
    "코드": "code_agent", "code": "code_agent",
    "검색": "search_agent", "조회": "search_agent",
    "데이터": "data_agent",
}

def supervisor_with_fast_path(state):
    query = state["query"].lower()
    for keyword, agent_name in KEYWORD_ROUTING.items():
        if keyword in query:
            return {"next": agent_name}
    routing_prompt = f"""
    사용자 요청: {state['query']}
    사용 가능 Agent: code_agent, search_agent, data_agent
    가장 적합한 Agent 이름만 반환하세요.
    """
    decision = llm.invoke(routing_prompt)
    return {"next": decision.content.strip()}

import autogen

reviewer_1 = autogen.AssistantAgent(
    name="SecurityReviewer",
    system_message="코드 보안 취약점에 집중하는 보안 전문가입니다."
)
reviewer_2 = autogen.AssistantAgent(
    name="PerformanceReviewer",
    system_message="코드 효율과 리소스 사용에 집중하는 성능 전문가입니다."
)
human_proxy = autogen.UserProxyAgent(
    name="CodeAuthor",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=2,
    is_termination_msg=lambda x: "APPROVED" in x.get("content", "")
)
groupchat = autogen.GroupChat(
    agents=[human_proxy, reviewer_1, reviewer_2],
    messages=[],
    max_round=6
)
manager = autogen.GroupChatManager(groupchat=groupchat)

from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("data-agent-mcp")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_customer_db",
            description="고객 DB 조회, ID·이름·이메일로 검색 지원",
            inputSchema={
                "type": "object",
                "properties": {
                    "field": {"type": "string", "enum": ["id", "name", "email"]},
                    "value": {"type": "string"}
                },
                "required": ["field", "value"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_customer_db":
        result = db.query(arguments["field"], arguments["value"])
        return [TextContent(type="text", text=str(result))]

import httpx

async def discover_and_delegate(agent_url: str, task: str):
    card_response = await httpx.get(f"{agent_url}/.well-known/agent.json")
    agent_card = card_response.json()
    available_skills = [s["id"] for s in agent_card["skills"]]
    if "web_research" not in available_skills:
        raise ValueError(f"Agent {agent_card['name']}은 web_research 스킬을 지원하지 않습니다")
    payload = {
        "jsonrpc": "2.0",
        "method": "message/send",
        "id": "task-001",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": task}]
            }
        }
    }
    response = await httpx.post(agent_card["url"], json=payload)
    return response.json()

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import interrupt

with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/agentdb") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "user-session-12345"}}
    result = graph.invoke({"query": "Q2 실적 분석"}, config)

def high_risk_action_agent(state):
    proposed_action = plan_action(state)
    human_decision = interrupt({
        "proposed_action": proposed_action,
        "risk_level": "HIGH",
        "message": "이 작업은 프로덕션 DB를 수정합니다. 실행을 확인해 주세요"
    })
    if human_decision["approved"]:
        return execute_action(proposed_action)
    return {"status": "cancelled", "reason": human_decision.get("reason")}

import time
from functools import wraps

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = "CLOSED"
        self.last_failure_time = None

    def __call__(self, func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if self.state == "OPEN":
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = "HALF_OPEN"
                else:
                    raise Exception("Circuit breaker OPEN - Agent 일시 사용 불가")
            try:
                result = await func(*args, **kwargs)
                if self.state == "HALF_OPEN":
                    self.state = "CLOSED"
                    self.failure_count = 0
                return result
            except Exception:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if self.failure_count >= self.failure_threshold:
                    self.state = "OPEN"
                raise
        return wrapper

class TokenBudgetManager:
    def __init__(self, total_budget: int = 100_000):
        self.total_budget = total_budget
        self.used_tokens = 0
        self.agent_usage = {}

    def check_budget(self, agent_name: str, estimated_tokens: int) -> bool:
        remaining = self.total_budget - self.used_tokens
        if estimated_tokens > remaining:
            raise BudgetExceededException(
                f"Agent {agent_name}이 {estimated_tokens} tokens를 요청했으나 "
                f"잔여 예산은 {remaining} tokens뿐입니다"
            )
        return True

    def record_usage(self, agent_name: str, actual_tokens: int):
        self.used_tokens += actual_tokens
        self.agent_usage[agent_name] = self.agent_usage.get(agent_name, 0) + actual_tokens

from opentelemetry import trace
import uuid

tracer = trace.get_tracer("multi-agent-system")

def traced_agent_call(agent_name: str, task: dict, correlation_id: str = None):
    if not correlation_id:
        correlation_id = str(uuid.uuid4())
    with tracer.start_as_current_span(f"agent.{agent_name}") as span:
        span.set_attribute("agent.name", agent_name)
        span.set_attribute("correlation.id", correlation_id)
        span.set_attribute("task.type", task.get("type", "unknown"))
        try:
            result = agent_registry[agent_name].run(task)
            span.set_attribute("agent.tokens_used", result.get("tokens", 0))
            span.set_attribute("agent.status", "success")
            return result
        except Exception as e:
            span.set_attribute("agent.status", "error")
            span.set_attribute("error.message", str(e))
            raise

MONITORING_METRICS = {
    "task_success_rate": "엔드투엔드 작업 완료율(목표: >85%)",
    "e2e_latency_p95": "P95 엔드투엔드 지연(목표: <30s)",
    "total_cost_per_task": "작업당 평균 토큰 비용",
    "agent_error_rate": "Agent별 오류율(목표: <5%)",
    "agent_retry_count": "재시도 횟수(높을수록 조사 필요)",
    "tool_call_budget_usage": "도구 호출 횟수/예산 비율",
    "output_quality_score": "출력 품질 점수",
    "goal_alignment_score": "목표 정합성 점수",
    "hallucination_rate": "환각 탐지율",
}

def evaluate_agent_output(original_task: str, agent_output: str) -> dict:
    evaluation_prompt = f"""
    원본 작업: {original_task}
    Agent 출력: {agent_output}
    작업 완료도, 정확성, 관련성, 환각 탐지 4개 차원으로 1~5점을 매기고
    JSON으로 반환하세요: {{"completeness": x, "accuracy": x, "relevance": x,
    "hallucination_detected": true/false, "comments": "..."}}
    """
    evaluation = llm.invoke(evaluation_prompt)
    return json.loads(evaluation.content)

def validate_agent_output(output: dict, schema: dict) -> bool:
    jsonschema.validate(output, schema)
    if output.get("confidence_score", 1.0) < 0.7:
        raise LowConfidenceError(f"Agent 출력 신뢰도 부족: {output['confidence_score']}")
    required_fields = schema.get("required", [])
    missing = [f for f in required_fields if not output.get(f)]
    if missing:
        raise MissingFieldsError(f"출력에 필수 필드 누락: {missing}")
    return True

MAX_ITERATIONS = 10
MAX_TOOL_CALLS_PER_AGENT = 20
MAX_TOTAL_TOKENS = 50_000

class ProductionGuardrails:
    def validate_input(self, user_input: str) -> str:
        if len(user_input) > 10000:
            raise InputTooLongError("입력이 10000자 제한을 초과했습니다")
        injection_patterns = ["ignore previous instructions", "forget everything"]
        for pattern in injection_patterns:
            if pattern.lower() in user_input.lower():
                raise PromptInjectionError("잠재적 프롬프트 인젝션 탐지")
        return user_input.strip()

    def validate_output(self, output: str) -> str:
        output = self.pii_filter.redact(output)
        if self.content_classifier.is_harmful(output):
            raise HarmfulContentError("출력에 유해 콘텐츠 포함")
        return output

builder.add_node("supervisor", supervisor_node, defer=True)

작업에 명확한 선형 의존 단계가 있나요?
├─ 예 → 하위 작업을 병렬 실행할 수 있나요?
│        ├─ 아니오 → 【순차 파이프라인】
│        └─ 예 → 【병렬 팬아웃 + 순차 파이프라인 하이브리드】
└─ 아니오 → 의사결정 권한을 가진 Agent가 있나요?
         ├─ 예 → 규모가 커서 하위 팀이 필요한가요?
         │        ├─ 아니오 → 【Supervisor-Worker 계층 패턴】
         │        └─ 예 → 【계층형(Supervisors of Supervisors)】
         └─ 아니오 → 장시간 비동기 작업인가요?
                  ├─ 예 → 【블랙보드 아키텍처】
                  └─ 아니오 → Agent 수가 ≤ 5인가요?
                           ├─ 예 → 【Swarm(종료 조건 설정 필수)】
                           └─ 아니오 → 【계층 패턴으로 재분해 검토】