멀티 에이전트 패턴 — 에이전트들이 협업하는 구조

왜 여러 에이전트가 필요한가?

단일 에이전트에 모든 도구를 주면 문제가 생깁니다.

flowchart LR
    subgraph SINGLE["단일 에이전트의 한계"]
        A["하나의 에이전트"]
        A --> T1["검색"]
        A --> T2["코드 실행"]
        A --> T3["파일 관리"]
        A --> T4["이메일 발송"]
        A --> T5["DB 조회"]
        A --> T6["API 호출"]
    end
    
    SINGLE --> PROBLEM["도구가 많을수록\n선택 오류 증가\n컨텍스트 낭비"]

멀티 에이전트로 해결:

flowchart TB
    OR["오케스트레이터\n전체 조율"]
    
    OR --> R["리서치 에이전트\n검색·정보 수집"]
    OR --> C["코드 에이전트\n코드 작성·실행"]
    OR --> W["작성 에이전트\n문서·보고서 작성"]
    
    R & C & W --> OR

각 에이전트가 전문화된 도구만 갖고 명확한 역할을 담당합니다.

---

멀티 에이전트 주요 패턴

패턴 1: 오케스트레이터-서브에이전트

flowchart TB
    USER["사용자 요청"] --> ORCH["오케스트레이터\n작업 분배·결과 통합"]
    
    ORCH -->|"서브태스크 1"| A1["리서치 에이전트"]
    ORCH -->|"서브태스크 2"| A2["분석 에이전트"]
    ORCH -->|"서브태스크 3"| A3["작성 에이전트"]
    
    A1 & A2 & A3 -->|"결과 반환"| ORCH
    ORCH --> FINAL["최종 결과"]

패턴 2: 순차 파이프라인

flowchart LR
    A1["수집 에이전트"] -->|"원시 데이터"| A2["분석 에이전트"]
    A2 -->|"분석 결과"| A3["작성 에이전트"]
    A3 -->|"초안"| A4["검토 에이전트"]
    A4 --> FINAL["최종 보고서"]

패턴 3: 병렬 처리

flowchart TB
    SPLIT["작업 분할"] --> A1["에이전트 1\n(태스크 A)"]
    SPLIT --> A2["에이전트 2\n(태스크 B)"]
    SPLIT --> A3["에이전트 3\n(태스크 C)"]
    
    A1 & A2 & A3 -->|"동시 완료"| MERGE["결과 통합"]

---

LangGraph로 오케스트레이터 구현

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 공유 상태
class MultiAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str           # 다음 실행할 에이전트
    research_result: str      # 리서치 결과
    analysis_result: str      # 분석 결과
    final_report: str         # 최종 보고서

---

각 에이전트 노드 구현

from langchain_core.tools import tool

# 리서치 에이전트 전용 도구
@tool
def search_news(topic: str) -> str:
    """최신 뉴스와 정보를 검색합니다."""
    return f"'{topic}' 관련 최신 정보: [검색 결과]"

@tool  
def search_academic(topic: str) -> str:
    """학술 자료와 연구 결과를 검색합니다."""
    return f"'{topic}' 학술 자료: [논문 요약]"

# 분석 에이전트 전용 도구
@tool
def run_statistics(data: str) -> str:
    """데이터 통계 분석을 수행합니다."""
    return f"통계 분석 결과: 평균, 표준편차, 트렌드 분석..."

# 작성 에이전트 전용 도구
@tool
def format_report(content: str, style: str = "markdown") -> str:
    """내용을 보고서 형식으로 정리합니다."""
    return f"[{style}] 형식으로 정리된 보고서"

# 에이전트 노드 함수들
def research_agent(state: MultiAgentState):
    research_llm = llm.bind_tools([search_news, search_academic])
    
    messages = [
        SystemMessage("당신은 리서치 전문가입니다. 정보를 수집하고 요약하세요."),
        HumanMessage(state["messages"][-1].content)
    ]
    
    response = research_llm.invoke(messages)
    return {
        "messages": [response],
        "research_result": response.content,
        "next_agent": "analyst"
    }

def analysis_agent(state: MultiAgentState):
    analysis_llm = llm.bind_tools([run_statistics])
    
    messages = [
        SystemMessage("당신은 데이터 분석가입니다. 수집된 정보를 분석하세요."),
        HumanMessage(f"다음 리서치 결과를 분석해주세요:\n{state['research_result']}")
    ]
    
    response = analysis_llm.invoke(messages)
    return {
        "messages": [response],
        "analysis_result": response.content,
        "next_agent": "writer"
    }

def writing_agent(state: MultiAgentState):
    writing_llm = llm.bind_tools([format_report])
    
    messages = [
        SystemMessage("당신은 전문 작가입니다. 분석 결과를 보고서로 작성하세요."),
        HumanMessage(
            f"리서치: {state['research_result']}\n"
            f"분석: {state['analysis_result']}\n"
            f"위 내용으로 최종 보고서를 작성하세요."
        )
    ]
    
    response = writing_llm.invoke(messages)
    return {
        "messages": [response],
        "final_report": response.content,
        "next_agent": "end"
    }

def orchestrator(state: MultiAgentState):
    """작업 흐름을 제어하는 오케스트레이터"""
    return {"next_agent": state.get("next_agent", "researcher")}

---

그래프 조립

def route_to_next(state: MultiAgentState) -> str:
    return state["next_agent"]

# 그래프 구성
graph = StateGraph(MultiAgentState)

graph.add_node("orchestrator", orchestrator)
graph.add_node("researcher", research_agent)
graph.add_node("analyst", analysis_agent)
graph.add_node("writer", writing_agent)

graph.set_entry_point("orchestrator")

graph.add_conditional_edges(
    "orchestrator",
    route_to_next,
    {
        "researcher": "researcher",
        "analyst": "analyst",
        "writer": "writer",
        "end": END
    }
)

# 각 에이전트 완료 후 오케스트레이터로 복귀
graph.add_edge("researcher", "orchestrator")
graph.add_edge("analyst", "orchestrator")
graph.add_edge("writer", "orchestrator")

multi_agent = graph.compile()

---

실행 및 모니터링

def run_multi_agent(topic: str):
    print(f"\n{'='*50}")
    print(f"작업 시작: {topic}")
    print('='*50)
    
    result = multi_agent.invoke({
        "messages": [HumanMessage(f"'{topic}'에 대한 리서치 보고서를 작성해줘")],
        "next_agent": "researcher",
        "research_result": "",
        "analysis_result": "",
        "final_report": ""
    })
    
    print("\n📋 최종 보고서:")
    print(result["final_report"])
    return result

run_multi_agent("생성형 AI 시장 동향 2024")

---

병렬 에이전트 실행

import asyncio

async def run_parallel_research(topics: list[str]) -> list[str]:
    """여러 주제를 동시에 리서치합니다."""
    tasks = [
        multi_agent.ainvoke({
            "messages": [HumanMessage(f"'{topic}' 간단 리서치")],
            "next_agent": "researcher",
            "research_result": "", "analysis_result": "", "final_report": ""
        })
        for topic in topics
    ]
    
    results = await asyncio.gather(*tasks)
    return [r["research_result"] for r in results]

# 5개 주제 동시 리서치
topics = ["AI 반도체", "자율주행", "양자컴퓨팅", "바이오테크", "우주산업"]
results = asyncio.run(run_parallel_research(topics))

---

에이전트 간 통신 패턴

패턴	설명	적합한 상황
공유 상태	모든 에이전트가 State 읽기/쓰기	긴밀히 협력
메시지 패싱	에이전트 간 직접 메시지	느슨한 결합
이벤트 기반	완료 이벤트 구독	비동기 병렬

---

정리

개념	내용
오케스트레이터	작업 분배·결과 통합하는 조율 에이전트
서브에이전트	특정 역할에 전문화된 에이전트
조건부 엣지	next_agent 상태로 다음 실행 노드 결정
병렬 실행	asyncio.gather로 여러 에이전트 동시 실행

다음 편에서는 실전 프로젝트 — 멀티 에이전트로 자율 리서치 어시스턴트를 완성합니다.