ARCH · MULTI-AGENT

멀티에이전트 토폴로지

🌐 4가지 구조 비교 👔 Supervisor 패턴 🔗 에이전트 간 통신

Star, Pipeline, Hierarchical, Mesh 토폴로지를 비교하고, LangGraph로 Supervisor 패턴과 에이전트 네트워크를 구현합니다.

4가지 멀티에이전트 토폴로지

그림 1. 멀티에이전트 토폴로지 4가지 — 복잡도와 유연성 트레이드오프

토폴로지	확장성	실패 격리	구현 복잡도	적합 케이스
Star	중간	중간	낮음	단순 위임, Supervisor 패턴
Pipeline	높음	낮음	낮음	ETL, 순차 처리 워크플로우
Hierarchical	매우 높음	높음	높음	대규모 엔터프라이즈 시스템
Mesh	중간	높음	매우 높음	협업 필요 특수 케이스

Supervisor 패턴 구현

가장 실용적인 패턴입니다. Supervisor LLM이 사용자 요청을 분석하고, 적합한 Worker Agent에게 위임합니다. LangGraph에서는 conditional_edges로 라우팅을 표현합니다.

pythonsupervisor.py — LangGraph Supervisor 패턴

from langgraph.graph import StateGraph, END, MessagesState
from langchain_anthropic import ChatAnthropic
from typing import Literal
from pydantic import BaseModel

WORKERS = ["researcher", "coder", "analyst", "writer"]

class RouterOutput(BaseModel):
    next: Literal["researcher", "coder", "analyst", "writer", "FINISH"]
    reason: str

supervisor_llm = ChatAnthropic(model="claude-opus-4-7")

async def supervisor_node(state: MessagesState) -> dict:
    """다음에 호출할 Worker 또는 FINISH 결정"""
    worker_desc = """
- researcher: 웹 검색, 정보 수집
- coder: 코드 작성, 실행, 디버깅
- analyst: 데이터 분석, 시각화
- writer: 보고서, 문서 작성"""

    structured_llm = supervisor_llm.with_structured_output(RouterOutput)
    result = await structured_llm.ainvoke([{
        "role": "system",
        "content": f"""당신은 팀의 Supervisor입니다. 대화를 보고 다음 Worker를 선택하거나 완료(FINISH)하세요.

Workers:{worker_desc}

완료 조건: 사용자 요청이 완전히 처리된 경우"""
    }, *state["messages"]])

    from langchain_core.messages import HumanMessage
    return {"messages": [HumanMessage(
        content=f"[Supervisor → {result.next}]: {result.reason}",
        name="supervisor"
    )], "next": result.next}

def route_to_worker(state) -> str:
    return state.get("next", "FINISH")

# Worker Agent 팩토리
def create_worker_node(name: str, system_prompt: str):
    worker_llm = ChatAnthropic(model="claude-haiku-4-5-20251001")
    worker_llm_with_tools = worker_llm.bind_tools(get_tools_for(name))

    async def worker_node(state: MessagesState) -> dict:
        response = await worker_llm_with_tools.ainvoke(
            state["messages"],
            system=system_prompt
        )
        from langchain_core.messages import AIMessage
        return {"messages": [AIMessage(
            content=response.content,
            name=name
        )]}
    return worker_node

# ─── 그래프 구성 ─────────────────────────────────────
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class SupervisorState(MessagesState):
    next: str

builder = StateGraph(SupervisorState)
builder.add_node("supervisor",  supervisor_node)
builder.add_node("researcher",  create_worker_node("researcher", "웹 검색 전문가"))
builder.add_node("coder",       create_worker_node("coder",       "Python 개발자"))
builder.add_node("analyst",    create_worker_node("analyst",    "데이터 분석가"))
builder.add_node("writer",     create_worker_node("writer",     "테크니컬 라이터"))

builder.set_entry_point("supervisor")
builder.add_conditional_edges("supervisor", route_to_worker, {
    "researcher": "researcher",
    "coder":       "coder",
    "analyst":    "analyst",
    "writer":     "writer",
    "FINISH":     END
})
for worker in WORKERS:
    builder.add_edge(worker, "supervisor")

supervisor_graph = builder.compile()

에이전트 간 통신 설계 원칙

원칙 1

메시지 기반 통신

Agent 간에 직접 함수 호출보다 메시지(dict/JSON)로 통신하세요. 의존성이 줄고, Agent를 독립적으로 교체할 수 있습니다.

원칙 2

컨텍스트 최소 전달

Agent에게 전체 이력이 아닌 해당 작업에 필요한 최소한의 컨텍스트만 전달합니다. 토큰 비용과 컨텍스트 혼란을 줄입니다.

원칙 3

결과 표준화

모든 Worker Agent의 반환 형식을 통일합니다. Supervisor가 결과 파싱에 LLM을 낭비하지 않도록 구조화된 출력을 사용합니다.

원칙 4

실패 전파 제어

한 Worker의 실패가 전체 파이프라인을 멈추지 않도록 Supervisor가 오류를 처리하고 대체 Worker를 선택하거나 부분 결과로 계속합니다.

Tool Use 고급 ↑ 목차 Agent 평가