langgraph-supervisor에서 마이그레이션하기

langgraph-supervisor 패키지는 더 이상 적극적으로 유지보수되지 않아요. 대신 subagents 패턴—즉 메인 에이전트가 전문화된 워커들을 도구처럼 호출하는 방식—을 쓰는 걸 권장합니다. 이 가이드에서는 create_supervisor를 create_agent 기반 코드로 옮기는 방법을, interrupt와 외부 API 콜백을 쓰는 설정까지 포함해 설명할게요.

출처: 공식문서

변경 요약

langgraph-supervisor	권장 대체
워커를 그래프 노드로 갖는 `create_supervisor`	subagent를 `@tool` 함수로 감싼 `create_agent`
메시지 히스토리를 위한 `output_mode`	툴 래퍼에서 subagent 출력을 형식화 (subagent outputs 참고)
커스텀 라우팅을 위한 `create_handoff_tool`	`subagent.invoke(...)`를 호출하는 커스텀 `@tool`
중첩 supervisor (supervisor의 supervisor)	다른 subagent를 호출하는 `@tool`로 감싼 subagent

기본 마이그레이션

langgraph-supervisor에서는 워커가 그래프 노드였고 supervisor가 handoff 도구로 라우팅했어요.

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent

research_agent = create_react_agent(
    model=model,
    tools=[web_search],
    name="research_expert",
    prompt="You are a research expert.",
)

math_agent = create_react_agent(
    model=model,
    tools=[add, multiply],
    name="math_expert",
    prompt="You are a math expert.",
)

workflow = create_supervisor(
    [research_agent, math_agent],
    model=model,
    prompt="Route research questions to research_expert and math to math_expert.",
)
app = workflow.compile(checkpointer=checkpointer)

subagents 패턴으로 옮기면, 각 워커를 메인 에이전트의 툴로 감싸면 돼요.

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

research_agent = create_agent(
    model=model,
    tools=[web_search],
    system_prompt="You are a research expert.",
)

math_agent = create_agent(
    model=model,
    tools=[add, multiply],
    system_prompt="You are a math expert.",
)


@tool("research_expert", description="Research expert for current events and web lookups.")
def call_research_agent(query: str) -> str:
    result = research_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content


@tool("math_expert", description="Math expert for calculations.")
def call_math_agent(query: str) -> str:
    result = math_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content


supervisor = create_agent(
    model=model,
    tools=[call_research_agent, call_math_agent],
    system_prompt=(
        "Route research questions to research_expert and math to math_expert."
    ),
    checkpointer=InMemorySaver(),
)

전체 워크스루는 subagents로 개인 비서 만들기를 참고하세요.

interrupt와 resume 흐름 마이그레이션

langgraph-supervisor에서 자주 쓰는 설정 중 하나는, 워커 에이전트의 툴 안에서 interrupt를 호출해 외부 서비스가 끝날 때까지 실행을 멈추는 방식이에요.

# Before: create_supervisor with a subgraph node
#
# Supervisor (create_supervisor)
#   └── ResearchAgent (subgraph node)
#         └── preview_tool
#               ├── fire_external_api()      # kicks off async job
#               ├── result = interrupt(...)  # pauses graph, waits for callback
#               └── render_results(result)   # runs after resume

subagents 패턴에서도 같은 흐름이 그대로 동작해요. subagent 툴 안의 interrupt는 툴로 감싸진 create_agent 레이어를 타고 올라가 가장 바깥 그래프까지 전파됩니다. 외부 콜백은 여전히 Command(resume=result)로 재개할 수 있어요.

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt


@tool
def preview_tool(document_id: str) -> str:
    """Run an async enrichment preview and wait for results."""
    job_id = fire_external_api(document_id)
    result = interrupt({"job_id": job_id, "status": "pending"})
    return render_results(result)


research_agent = create_agent(
    model=model,
    tools=[preview_tool],
    system_prompt="You are a research agent.",
)

@tool("research_agent", description="Research and enrichment tasks.")
def call_research_agent(query: str) -> str:
    result = research_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

supervisor = create_agent(
    model=model,
    tools=[call_research_agent],
    system_prompt="Delegate research tasks to research_agent.",
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "1"}}

from langgraph.types import Command

# Invoke — preview_tool calls interrupt() and the graph pauses
response = supervisor.invoke(
    {"messages": [{"role": "user", "content": "Preview enrichment for doc-123"}]},
    config=config,
)
# response contains __interrupt__

# External service completes and calls back into your app
supervisor.invoke(Command(resume=external_result), config=config)

interrupt 전파 요건

interrupt가 중첩된 create_agent 레이어를 타고 올라오려면 두 가지 규칙을 지켜야 해요.

가장 바깥 그래프에만 checkpointer로 compile. subagent에는 checkpointer=...를 두지 말고, 그렇게 하면 per-invocation persistence를 쓰면서 런타임에 부모의 checkpointer를 상속받아요.
configurable에 thread_id 전달. 바깥 invoke()/stream_events() 호출에 thread_id가 있어야 그래프가 체크포인트하고 재개할 수 있어요.

이 규칙은 임의의 깊이로 중첩한 설정에도 동일하게 적용돼요. 예를 들어 커스텀 StateGraph가 바깥, 중간에 create_agent supervisor, 안쪽에 다시 create_agent subagent가 있어도 같은 메커니즘을 따라요.

Custom StateGraph (outer, with checkpointer)
  └── prospecting_agent (create_agent, no checkpointer)
        └── call_powerup_agent tool → powerup_agent.invoke(...)
              └── powerup_agent (create_agent, no checkpointer)
                    └── preview_tool → interrupt(...)

preview_tool이 interrupt를 호출하면 예외가 두 create_agent 레이어를 타고 올라가 바깥 StateGraph의 invoke 결과에 __interrupt__로 나타나요. 기존 Command(resume=result) 콜백 경로는 그대로 동작합니다.

커스텀 StateGraph를 써야 할 때

결정적(deterministic) 단계와 에이전트 단계를 섞어야 한다면 커스텀 StateGraph를 쓰세요. 예를 들어 고정 라우팅·검증·외부 API 호출 같은 걸 create_agent 노드와 함께 두어야 할 때요.

중첩 supervisor 마이그레이션

langgraph-supervisor는 supervisor를 compile해서 다른 create_supervisor에 넘기는 식으로 다단계 계층을 지원했어요. subagents 패턴에서는 두 가지 선택지가 있어요.

단일 supervisor로 평탄화 — 리프 에이전트마다 툴 하나씩. 각 워커가 독립적일 때 가장 간단해요.
툴 호출 중첩 — 중간 단계 조정이 필요할 때. 중간 계층 에이전트(자체 subagent 툴을 가진 create_agent)를 최상위 supervisor의 툴로 감싸요.

from langchain.agents import create_agent
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver

# Middle-tier agent with its own subagents
billing_team = create_agent(
    model=model,
    tools=[call_refunds_agent, call_invoices_agent],
    system_prompt="Coordinate billing specialists.",
)

@tool("billing_team", description="Handle billing, refunds, and invoices.")
def call_billing_team(query: str) -> str:
    result = billing_team.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].content

# Top-level supervisor
top_supervisor = create_agent(
    model=model,
    tools=[call_billing_team, call_support_agent],
    system_prompt="Route billing to billing_team and general support to support_agent.",
    checkpointer=InMemorySaver(),
)

정적 subgraph 탐색, 계층별 checkpoint namespace, 계층 간 공유 상태 키가 필요하다면 커스텀 StateGraph를 subgraph 노드와 함께 쓰세요.

메시지 히스토리 옵션 마이그레이션

create_supervisor의 output_mode는 워커 메시지가 대화 히스토리에 어떻게 나타날지 제어했어요.

full_history: 워커 에이전트의 모든 메시지 포함.
last_message: 워커의 최종 응답만 포함.

subagents 패턴에서는 이 동작을 툴 래퍼에서 제어해요. last_message 동작은 마지막 메시지만 반환하고, full_history 동작은 전체 대화를 형식화한 요약을 반환하면 돼요. Subagent outputs에서 supervisor에 추가 상태를 전달하는 패턴을 볼 수 있어요.

더 알아보기 (Learn more)

Subagents — 패턴 개요와 설계 결정
subagents로 개인 비서 만들기 — 단계별 supervisor 튜토리얼
Use subgraphs — subgraph persistence, interrupts, 상태 검사
Interrupts — 그래프 실행 일시정지·재개
LangGraph v1 마이그레이션 가이드 — create_react_agent → create_agent