LangGraph Supervisor 예제

https://langchain-ai.github.io/langgraph/tutorials/multi_agent/agent_supervisor/

Agent Supervisor

자세한 내용은 위 공식문서 확인

 

 

해당 예제에는 아래 표와 같이 3가지 방식으로 supervisor를 구현하고 있음

구분	langgraph-supervisor	Supervisor from scratch	Delegation tasks
구현 방식	함수 호출으로 구성	Graph API + primitives 직접 구현	handoff tool 커스터마이징, Send/forward 도입
설정/커스터마이징 난이도	낮음, 기본 구성 제공	높음, 모든 로직 직접 정의	중간, tool 정의만 잘 하면 됨
제어 수준	보통 수준	매우 세밀한 제어 가능	transfer granularity 높이며 메시지 흐름 직접 정의 가능
추천 대상	빠르게 supervisor 구현하고 싶은 경우	복잡한 흐름이나 유연한 라우팅이 필요한 경우	task delegation이나 handoff 조정이 필요한 경우

정리

간편하게 multi-agent supervisor가 필요하면 → langgraph-supervisor 활용 (방법 2)

보다 정교한 agent 흐름 제어가 필요하면 → Graph API로 직접 구현 (방법 3)

특정 단계에 delegate나 forward 메시지가 중요하면 → delegation 로직 추가 (방법 4)

내용 설명

langgraph-supervisor 이용: 미리 준비된 Supervisor

• create_supervisor라는 함수 하나로 구현 가능 → 코드 간결
• 내부에서 handoff tool과 supervisor agent 생성/관리 모두 처리
• 장점: 빠르고 직관적, 기본 레벨 multi-agent 구축에 적합
• 주요 설정 옵션:
  • output_mode: 메시지 기록 방식 선택 (full_history, last_message)
  • handoff_tool_prefix, add_handoff_messages 등으로 handoff 동작 커스터마이징 가능
• 확장성: supervisor 계층화 가능 (multi-level hierarchical)

Supervisor from scratch: 직접 손수 구현

• StateGraph + Command/Send 같은 primitives로 명시적으로 구성
• agent 간 graf node, message 흐름, routing logic 직접 설정
• 장점: handoff 과정, 메시지 구조, 상태 전달 등을 훨씬 세밀하게 제어 가능
• 필요 작업:
  • 각 agent 노드 정의 → MessagesState 사용
  • Command로 어떤 agent에게 handoff할지 지정
  • Send로 payload 직접 추가
• 적합 대상: 복잡하거나 정교한 흐름이 필요한 커스텀 multi-agent 시스템

Delegation tasks: 위임(task delegation) 기법

• supervisor나 agent가 하위 agent에게 task 요청, 결과 수집하는 고급 패턴
• 예시: create_handoff_tool를 Send 포함해서 task 설명 직접 작성하게 유도
• 또는 create_forward_message_tool로 하위 agent의 답변을 supervisor 없이 바로 최종 출력까지 전달
• 장점: task granularity를 조절하거나 요약/전달 방식 정의 가능
• 특히 도메인 특정 handoff나 역할 분리된 비즈니스 로직에 유리

 

예제 코드

1. Create worker agents

Research agen

웹서치를 위한 도구 생성

from langchain_tavily import TavilySearch

web_search = TavilySearch(max_results=3)
web_search_results = web_search.invoke("who is the mayor of NYC?")

print(web_search_results["results"][0]["content"])

웹서치 도구를 사용하는 agent생성

from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[web_search],
    prompt=(
        "You are a research agent.\n\n"
        "INSTRUCTIONS:\n"
        "- Assist ONLY with research-related tasks, DO NOT do any math\n"
        "- After you're done with your tasks, respond to the supervisor directly\n"
        "- Respond ONLY with the results of your work, do NOT include ANY other text."
    ),
    name="research_agent",
)

agent 실행

for chunk in research_agent.stream(
    {"messages": [{"role": "user", "content": "who is the mayor of NYC?"}]}
):
    pretty_print_messages(chunk)

Math agent

수학 작업을 하는 tool과 agent생성

def add(a: float, b: float):
    """Add two numbers."""
    return a + b


def multiply(a: float, b: float):
    """Multiply two numbers."""
    return a * b


def divide(a: float, b: float):
    """Divide two numbers."""
    return a / b


math_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[add, multiply, divide],
    prompt=(
        "You are a math agent.\n\n"
        "INSTRUCTIONS:\n"
        "- Assist ONLY with math-related tasks\n"
        "- After you're done with your tasks, respond to the supervisor directly\n"
        "- Respond ONLY with the results of your work, do NOT include ANY other text."
    ),
    name="math_agent",
)

Math agent실행

for chunk in math_agent.stream(
    {"messages": [{"role": "user", "content": "what's (3 + 5) x 7"}]}
):
    pretty_print_messages(chunk)

2. Create supervisor with langgraph-supervisor

prebuilt를 사용하여 supervisor생성

from langgraph_supervisor import create_supervisor
from langchain.chat_models import init_chat_model

supervisor = create_supervisor(
    model=init_chat_model("openai:gpt-4.1"),
    agents=[research_agent, math_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research-related tasks to this agent\n"
        "- a math agent. Assign math-related tasks to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    add_handoff_back_messages=True,
    output_mode="full_history",
).compile()

두 에이전트(검색 수학)가 모두 필요한 쿼리로 실행

 for chunk in supervisor.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "find US and New York state GDP in 2024. what % of US GDP was New York state?",
            }
        ]
    },
):
    pretty_print_messages(chunk, last_message=True)

final_message_history = chunk["supervisor"]["messages"]

3. Create supervisor from scratch

에이전트 커뮤니케이션 설정

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command


def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ) -> Command:
        tool_message = {
            "role": "tool",
            "content": f"Successfully transferred to {agent_name}",
            "name": name,
            "tool_call_id": tool_call_id,
        }
        return Command(
            goto=agent_name,  
            update={**state, "messages": state["messages"] + [tool_message]},  
            graph=Command.PARENT,  
        )

    return handoff_tool


# Handoffs
assign_to_research_agent = create_handoff_tool(
    agent_name="research_agent",
    description="Assign task to a researcher agent.",
)

assign_to_math_agent = create_handoff_tool(
    agent_name="math_agent",
    description="Assign task to a math agent.",
)

supervisor 에이전트 생성

supervisor_agent = create_react_agent(
    model="openai:gpt-4.1",
    tools=[assign_to_research_agent, assign_to_math_agent],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research-related tasks to this agent\n"
        "- a math agent. Assign math-related tasks to this agent\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    name="supervisor",
)

multi-agent graph 생성

from langgraph.graph import END

# Define the multi-agent supervisor graph
supervisor = (
    StateGraph(MessagesState)
    # NOTE: `destinations` is only needed for visualization and doesn't affect runtime behavior
    .add_node(supervisor_agent, destinations=("research_agent", "math_agent", END))
    .add_node(research_agent)
    .add_node(math_agent)
    .add_edge(START, "supervisor")
    # always return back to the supervisor
    .add_edge("research_agent", "supervisor")
    .add_edge("math_agent", "supervisor")
    .compile()
)

실행

for chunk in supervisor.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "find US and New York state GDP in 2024. what % of US GDP was New York state?",
            }
        ]
    },
):
    pretty_print_messages(chunk, last_message=True)

final_message_history = chunk["supervisor"]["messages"]

4. Create delegation tasks

from langgraph.types import Send


def create_task_description_handoff_tool(
    *, agent_name: str, description: str | None = None
):
    name = f"transfer_to_{agent_name}"
    description = description or f"Ask {agent_name} for help."

    @tool(name, description=description)
    def handoff_tool(
        # this is populated by the supervisor LLM
        task_description: Annotated[
            str,
            "Description of what the next agent should do, including all of the relevant context.",
        ],
        # these parameters are ignored by the LLM
        state: Annotated[MessagesState, InjectedState],
    ) -> Command:
        task_description_message = {"role": "user", "content": task_description}
        agent_input = {**state, "messages": [task_description_message]}
        return Command(
            goto=[Send(agent_name, agent_input)],
            graph=Command.PARENT,
        )

    return handoff_tool


assign_to_research_agent_with_description = create_task_description_handoff_tool(
    agent_name="research_agent",
    description="Assign task to a researcher agent.",
)

assign_to_math_agent_with_description = create_task_description_handoff_tool(
    agent_name="math_agent",
    description="Assign task to a math agent.",
)

supervisor_agent_with_description = create_react_agent(
    model="openai:gpt-4.1",
    tools=[
        assign_to_research_agent_with_description,
        assign_to_math_agent_with_description,
    ],
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent. Assign research-related tasks to this assistant\n"
        "- a math agent. Assign math-related tasks to this assistant\n"
        "Assign work to one agent at a time, do not call agents in parallel.\n"
        "Do not do any work yourself."
    ),
    name="supervisor",
)

supervisor_with_description = (
    StateGraph(MessagesState)
    .add_node(
        supervisor_agent_with_description, destinations=("research_agent", "math_agent")
    )
    .add_node(research_agent)
    .add_node(math_agent)
    .add_edge(START, "supervisor")
    .add_edge("research_agent", "supervisor")
    .add_edge("math_agent", "supervisor")
    .compile()
)

for chunk in supervisor.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "find US and New York state GDP in 2024. what % of US GDP was New York state?",
            }
        ]
    },
    subgraphs=True,
):
    pretty_print_messages(chunk, last_message=True)