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

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

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

멀티 에이전트로 해결:

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

---

멀티 에이전트 주요 패턴

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

패턴 2: 순차 파이프라인

패턴 3: 병렬 처리

---

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로 여러 에이전트 동시 실행

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