LangGraph Supervisor 패턴 프로덕션 적용기 — 여러 AI 에이전트를 오케스트레이션하고 책임을 추적하는 법

Planner Layer  →  Supervisor (무엇을 누구에게?)
Agents Layer   →  Worker Agents (각 전문 도메인 처리)
Tooling Layer  →  Tools / APIs (실제 실행 단위)

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
 
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # reducer 지정
    current_task: str

from langgraph_supervisor import create_handoff_tool
 
handoff_to_math = create_handoff_tool(
    agent_name="math_expert",
    description="수학 계산이 필요할 때 사용"
)

[메시지 히스토리 예시]
User → "57의 제곱근을 계산하고 결과를 한국어로 설명해줘"
Supervisor → [ToolMessage] math_expert로 이전 (tool_call_id: abc123)
math_expert → [AIMessage] "57의 제곱근은 약 7.55입니다"
Supervisor → [ToolMessage] writer_expert로 이전 (tool_call_id: def456)
writer_expert → [AIMessage] "57의 제곱근은 7.55 정도예요. 쉽게 말하면..."

import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
 
# 더미 도구 정의 — 실제로는 외부 API 호출 등으로 교체
@tool
def calculator(expression: str) -> str:
    """수식을 계산합니다. 예: '3 ** 2 + 1'"""
    try:
        return str(eval(expression))  # 실제 프로덕션에서는 안전한 파서 사용
    except Exception as e:
        return f"계산 오류: {e}"
 
@tool
def get_weather(city: str) -> str:
    """도시의 날씨 정보를 반환합니다."""
    # 실제 구현에서는 외부 날씨 API 호출
    return f"{city}의 현재 기온은 22°C, 맑음입니다."
 
llm = ChatOpenAI(model="gpt-4o", api_key=os.environ["OPENAI_API_KEY"])
 
# 수학 전문가 에이전트
math_expert = create_react_agent(
    model=llm,
    tools=[calculator],
    name="math_expert",  # 핸드오프 추적의 키. 공백 없이 snake_case로
    prompt="당신은 수학 전문가입니다. 계산 문제만 처리합니다."
)
 
# 날씨 전문가 에이전트
weather_expert = create_react_agent(
    model=llm,
    tools=[get_weather],
    name="weather_expert",
    prompt="당신은 날씨 정보 전문가입니다."
)
 
# 글쓰기 전문가 에이전트 (도구 없이 LLM만 활용)
writer_expert = create_react_agent(
    model=llm,
    tools=[],
    name="writer_expert",
    prompt="당신은 글쓰기 전문가입니다. 명확하고 친절한 설명을 작성합니다."
)
 
# Supervisor 생성 및 컴파일
supervisor = create_supervisor(
    agents=[math_expert, weather_expert, writer_expert],
    model=llm,
    prompt=(
        "당신은 팀 매니저입니다. "
        "요청을 분석하고 가장 적합한 전문가에게 위임하세요. "
        "복합 요청은 여러 전문가를 순차적으로 활용할 수 있습니다."
    )
).compile()
 
# 실행
result = supervisor.invoke({
    "messages": [
        {"role": "user", "content": "오늘 서울 날씨 알려주고, 기온의 제곱근도 계산해줘"}
    ]
})
 
print(result["messages"][-1].content)

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
 
# AWS Bedrock 환경이라면 아래처럼 LLM만 교체하면 나머지 코드는 동일
# from langchain_aws import ChatBedrock
# llm = ChatBedrock(model_id="anthropic.claude-3-5-sonnet-20241022-v2:0")
 
@tool
def get_invoice(customer_id: str) -> str:
    """고객 청구서를 조회합니다."""
    return f"고객 {customer_id}의 최근 청구액: 99,000원 (2025-04 기준)"
 
@tool
def process_refund(customer_id: str, amount: int) -> str:
    """환불을 처리합니다."""
    return f"고객 {customer_id}에게 {amount}원 환불 처리 완료"
 
@tool
def search_kb(query: str) -> str:
    """지식베이스에서 기술 문서를 검색합니다."""
    return f"'{query}' 관련 문서: docs.example.com/troubleshoot"
 
@tool
def create_priority_ticket(description: str) -> str:
    """긴급 티켓을 생성하고 인간 상담사에게 알립니다."""
    return f"긴급 티켓 #9999 생성 완료. 담당자에게 알림 전송됨."
 
billing_agent = create_react_agent(
    model=llm,
    tools=[get_invoice, process_refund],
    name="billing_agent",
    prompt="청구 관련 문의만 처리합니다. 환불, 청구서 조회 등을 담당합니다."
)
 
technical_agent = create_react_agent(
    model=llm,
    tools=[search_kb],
    name="technical_agent",
    prompt="기술적 문제 해결을 담당합니다. 해결이 어렵다면 에스컬레이션을 권장합니다."
)
 
escalation_agent = create_react_agent(
    model=llm,
    tools=[create_priority_ticket],
    name="escalation_agent",
    prompt="긴급하거나 복잡한 케이스를 인간 상담사에게 연결합니다."
)
 
support_supervisor = create_supervisor(
    agents=[billing_agent, technical_agent, escalation_agent],
    model=llm,
    prompt="""
    고객 지원 매니저입니다. 요청 유형에 따라 적절한 에이전트에 위임하세요:
    - 청구/결제 문의 → billing_agent
    - 기술적 문제 → technical_agent
    - 긴급하거나 동일 문의가 반복되는 경우 → escalation_agent
    항상 고객에게 친절하게 응대하세요.
    """
).compile()

import asyncio
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(model="gpt-4o")
 
class DocumentState(TypedDict):
    document: str
    sections: List[str]
    summaries: List[str]
    final_report: str
 
async def summarize_section(section: str) -> str:
    """섹션 하나를 요약합니다."""
    response = await llm.ainvoke(f"다음 내용을 2문장으로 요약해주세요:\n\n{section}")
    return response.content
 
def split_document(state: DocumentState) -> dict:
    sections = [s.strip() for s in state["document"].split("\n\n") if s.strip()]
    return {"sections": sections}
 
async def process_sections_parallel(state: DocumentState) -> dict:
    # asyncio.gather로 여러 섹션을 병렬 요약
    # LangGraph는 async def 노드를 지원하므로 그대로 await 가능
    tasks = [summarize_section(section) for section in state["sections"]]
    summaries = await asyncio.gather(*tasks)
    return {"summaries": list(summaries)}
 
def merge_results(state: DocumentState) -> dict:
    final = "\n\n".join(
        f"[섹션 {i+1}] {summary}"
        for i, summary in enumerate(state["summaries"])
    )
    return {"final_report": final}
 
# 그래프 구성
graph = StateGraph(DocumentState)
graph.add_node("split", split_document)
graph.add_node("process", process_sections_parallel)
graph.add_node("merge", merge_results)
 
graph.add_edge(START, "split")
graph.add_edge("split", "process")
graph.add_edge("process", "merge")
graph.add_edge("merge", END)
 
pipeline = graph.compile()
 
# 실행 (async 환경에서)
async def main():
    result = await pipeline.ainvoke({
        "document": "첫 번째 섹션 내용...\n\n두 번째 섹션 내용...\n\n세 번째 섹션 내용..."
    })
    print(result["final_report"])