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

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))