원문: https://langchain-ai.github.io/langgraph/how-tos/multi_agent/

개요

단일 에이전트는 여러 도메인을 전문화하거나 많은 도구를 관리해야 할 때 어려움을 겪을 수 있습니다. 이를 해결하기 위해 에이전트를 더 작고 독립적인 에이전트로 분리하여 멀티 에이전트 시스템으로 구성할 수 있습니다.

멀티 에이전트 시스템에서 에이전트는 서로 통신해야 합니다. 이들은 핸드오프(Handoff) 를 통해 통신합니다. 핸드오프는 제어권을 넘겨줄 에이전트와 해당 에이전트에게 전송할 페이로드를 설명하는 프리미티브입니다.

이 가이드는 다음 내용을 다룹니다:

멀티 에이전트 시스템 구축을 시작하려면 가장 인기 있는 두 가지 멀티 에이전트 아키텍처인 슈퍼바이저스웜의 LangGraph 미리 빌드된 구현을 확인하세요.

핸드오프

멀티 에이전트 시스템에서 에이전트 간 통신을 설정하려면 핸드오프를 사용할 수 있습니다. 핸드오프는 한 에이전트가 제어권을 다른 에이전트에게 넘기는 패턴입니다. 핸드오프를 통해 다음을 지정할 수 있습니다:

  • 목적지: 이동할 대상 에이전트(예: 이동할 LangGraph 노드의 이름)
  • 페이로드: 해당 에이전트에게 전달할 정보(예: 상태 업데이트)

핸드오프 생성

핸드오프를 구현하려면 에이전트 노드나 도구에서 Command 객체를 반환할 수 있습니다:

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import create_react_agent, 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"Transfer to {agent_name}"
 
    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],  # (1)
        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(  # (2)
            goto=agent_name,  # (3)
            update={"messages": state["messages"] + [tool_message]},  # (4)
            graph=Command.PARENT,  # (5)
        )
 
    return handoff_tool
  1. InjectedState 어노테이션을 사용하여 핸드오프 도구를 호출하는 에이전트의 상태에 접근합니다.
  2. Command 프리미티브를 사용하면 상태 업데이트와 노드 전환을 단일 작업으로 지정할 수 있어 핸드오프 구현에 유용합니다.
  3. 핸드오프할 에이전트 또는 노드의 이름입니다.
  4. 에이전트의 메시지를 가져와서 핸드오프의 일부로 부모의 상태추가합니다. 다음 에이전트는 부모 상태를 볼 수 있습니다.
  5. LangGraph에게 부모 멀티 에이전트 그래프의 에이전트 노드로 이동해야 함을 나타냅니다.

Command를 반환하는 도구를 사용하려면 미리 빌드된 create_react_agent / ToolNode 컴포넌트를 사용하거나, 도구가 반환한 Command 객체를 수집하여 목록으로 반환하는 자체 도구 실행 노드를 구현할 수 있습니다:

def call_tools(state):
    ...
    commands = [tools_by_name[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
    return commands

중요

이 핸드오프 구현은 다음을 가정합니다:

  • 각 에이전트는 멀티 에이전트 시스템의 전체 메시지 히스토리(모든 에이전트에 걸친)를 입력으로 받습니다. 에이전트 입력에 대한 더 많은 제어가 필요한 경우 이 섹션을 참조하세요.
  • 각 에이전트는 자신의 내부 메시지 히스토리를 멀티 에이전트 시스템의 전체 메시지 히스토리로 출력합니다. 에이전트 출력이 추가되는 방식에 대한 더 많은 제어가 필요한 경우 별도의 노드 함수로 에이전트를 래핑하세요:
def call_hotel_assistant(state):
    # 내부 독백을 제외하고 에이전트의 최종 응답 반환
    response = hotel_assistant.invoke(state)
    return {"messages": response["messages"][-1]}

에이전트 입력 제어

Send() 프리미티브를 사용하여 핸드오프 중에 워커 에이전트에게 직접 데이터를 전송할 수 있습니다. 예를 들어, 호출하는 에이전트가 다음 에이전트를 위한 작업 설명을 작성하도록 요청할 수 있습니다:

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, 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(
        # 이것은 호출하는 에이전트에 의해 채워집니다
        task_description: Annotated[
            str,
            "Description of what the next agent should do, including all of the relevant context.",
        ],
        # 이러한 매개변수는 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

핸드오프에서 Send()를 사용하는 전체 예제는 멀티 에이전트 슈퍼바이저 예제를 참조하세요.

멀티 에이전트 시스템 구축

LangGraph로 구축된 모든 에이전트에서 핸드오프를 사용할 수 있습니다. Command를 반환하는 핸드오프 도구를 기본적으로 지원하는 미리 빌드된 에이전트 또는 ToolNode를 사용하는 것을 권장합니다. 다음은 핸드오프를 사용하여 여행 예약을 위한 멀티 에이전트 시스템을 구현하는 방법의 예입니다:

from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, MessagesState
 
def create_handoff_tool(*, agent_name: str, description: str | None = None):
    # 위와 동일한 구현
    ...
    return Command(...)
 
# 핸드오프
transfer_to_hotel_assistant = create_handoff_tool(agent_name="hotel_assistant")
transfer_to_flight_assistant = create_handoff_tool(agent_name="flight_assistant")
 
# 에이전트 정의
flight_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[..., transfer_to_hotel_assistant],
    name="flight_assistant"
)
 
hotel_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[..., transfer_to_flight_assistant],
    name="hotel_assistant"
)
 
# 멀티 에이전트 그래프 정의
multi_agent_graph = (
    StateGraph(MessagesState)
    .add_node(flight_assistant)
    .add_node(hotel_assistant)
    .add_edge(START, "flight_assistant")
    .compile()
)

완전한 예제: 여행 예약을 위한 멀티 에이전트 시스템

from typing import Annotated
from langchain_core.messages import convert_to_messages
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command
 
# 메시지 출력을 위한 헬퍼 함수
def pretty_print_message(message, indent=False):
    pretty_message = message.pretty_repr(html=True)
    if not indent:
        print(pretty_message)
        return
 
    indented = "\n".join("\t" + c for c in pretty_message.split("\n"))
    print(indented)
 
def pretty_print_messages(update, last_message=False):
    is_subgraph = False
    if isinstance(update, tuple):
        ns, update = update
        # 출력에서 부모 그래프 업데이트 건너뛰기
        if len(ns) == 0:
            return
 
        graph_id = ns[-1].split(":")[0]
        print(f"Update from subgraph {graph_id}:")
        print("\n")
        is_subgraph = True
 
    for node_name, node_update in update.items():
        update_label = f"Update from node {node_name}:"
        if is_subgraph:
            update_label = "\t" + update_label
 
        print(update_label)
        print("\n")
 
        messages = convert_to_messages(node_update["messages"])
        if last_message:
            messages = messages[-1:]
 
        for m in messages:
            pretty_print_message(m, indent=is_subgraph)
            print("\n")
 
def create_handoff_tool(*, agent_name: str, description: str | None = None):
    name = f"transfer_to_{agent_name}"
    description = description or f"Transfer to {agent_name}"
 
    @tool(name, description=description)
    def handoff_tool(
        state: Annotated[MessagesState, InjectedState],  # (1)
        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(  # (2)
            goto=agent_name,  # (3)
            update={"messages": state["messages"] + [tool_message]},  # (4)
            graph=Command.PARENT,  # (5)
        )
 
    return handoff_tool
 
# 핸드오프
transfer_to_hotel_assistant = create_handoff_tool(
    agent_name="hotel_assistant",
    description="Transfer user to the hotel-booking assistant.",
)
 
transfer_to_flight_assistant = create_handoff_tool(
    agent_name="flight_assistant",
    description="Transfer user to the flight-booking assistant.",
)
 
# 간단한 에이전트 도구
def book_hotel(hotel_name: str):
    """호텔 예약"""
    return f"Successfully booked a stay at {hotel_name}."
 
def book_flight(from_airport: str, to_airport: str):
    """항공편 예약"""
    return f"Successfully booked a flight from {from_airport} to {to_airport}."
 
# 에이전트 정의
flight_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_flight, transfer_to_hotel_assistant],
    prompt="You are a flight booking assistant",
    name="flight_assistant"
)
 
hotel_assistant = create_react_agent(
    model="anthropic:claude-3-5-sonnet-latest",
    tools=[book_hotel, transfer_to_flight_assistant],
    prompt="You are a hotel booking assistant",
    name="hotel_assistant"
)
 
# 멀티 에이전트 그래프 정의
multi_agent_graph = (
    StateGraph(MessagesState)
    .add_node(flight_assistant)
    .add_node(hotel_assistant)
    .add_edge(START, "flight_assistant")
    .compile()
)
 
# 멀티 에이전트 그래프 실행
for chunk in multi_agent_graph.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
            }
        ]
    },
    subgraphs=True
):
    pretty_print_messages(chunk)
  1. 에이전트의 상태에 접근
  2. Command 프리미티브는 상태 업데이트와 노드 전환을 단일 작업으로 지정할 수 있어 핸드오프 구현에 유용합니다.
  3. 핸드오프할 에이전트 또는 노드의 이름
  4. 에이전트의 메시지를 가져와서 핸드오프의 일부로 부모의 상태추가합니다. 다음 에이전트는 부모 상태를 볼 수 있습니다.
  5. LangGraph에게 부모 멀티 에이전트 그래프의 에이전트 노드로 이동해야 함을 나타냅니다.

멀티턴 대화

사용자는 하나 이상의 에이전트와 멀티턴 대화를 하고 싶어할 수 있습니다. 이를 처리할 수 있는 시스템을 구축하려면 interrupt를 사용하여 사용자 입력을 수집하고 활성 에이전트로 다시 라우팅하는 노드를 만들 수 있습니다.

그런 다음 에이전트를 에이전트 단계를 실행하고 다음 작업을 결정하는 그래프의 노드로 구현할 수 있습니다:

  1. 대화를 계속하기 위해 사용자 입력을 기다리거나, 또는
  2. 핸드오프를 통해 다른 에이전트로 라우팅(또는 루프처럼 자기 자신으로 돌아감)
def human(state) -> Command[Literal["agent", "another_agent"]]:
    """사용자 입력을 수집하는 노드"""
    user_input = interrupt(value="Ready for user input.")
 
    # 활성 에이전트 결정
    active_agent = ...
 
    ...
    return Command(
        update={
            "messages": [{
                "role": "human",
                "content": user_input,
            }]
        },
        goto=active_agent
    )
 
def agent(state) -> Command[Literal["agent", "another_agent", "human"]]:
    # 라우팅/중단 조건은 무엇이든 될 수 있음(예: LLM 도구 호출/구조화된 출력 등)
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    if goto:
        return Command(goto=goto, update={"my_state_key": "my_state_value"})
    else:
        return Command(goto="human")  # 사람 노드로 이동

완전한 예제: 여행 추천을 위한 멀티 에이전트 시스템

이 예제에서는 핸드오프를 통해 서로 통신할 수 있는 여행 어시스턴트 에이전트 팀을 구축합니다.

2개의 에이전트를 만듭니다:

  • travel_advisor: 여행 목적지 추천을 도울 수 있습니다. hotel_advisor에게 도움을 요청할 수 있습니다.
  • hotel_advisor: 호텔 추천을 도울 수 있습니다. travel_advisor에게 도움을 요청할 수 있습니다.
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
 
model = ChatAnthropic(model="claude-3-5-sonnet-latest")
 
class MultiAgentState(MessagesState):
    last_active_agent: str
 
# 여행 어드바이저 도구 및 ReAct 에이전트 정의
travel_advisor_tools = [
    get_travel_recommendations,
    make_handoff_tool(agent_name="hotel_advisor"),
]
 
travel_advisor = create_react_agent(
    model,
    travel_advisor_tools,
    prompt=(
        "You are a general travel expert that can recommend travel destinations (e.g. countries, cities, etc). "
        "If you need hotel recommendations, ask 'hotel_advisor' for help. "
        "You MUST include human-readable response before transferring to another agent."
    ),
)
 
def call_travel_advisor(
    state: MultiAgentState,
) -> Command[Literal["hotel_advisor", "human"]]:
    # 입력/출력을 에이전트로/에이전트에서 변경하는 등의 추가 로직을 추가할 수도 있습니다
    # 참고: 상태의 전체 메시지 히스토리로 ReAct 에이전트를 호출합니다
    response = travel_advisor.invoke(state)
    update = {**response, "last_active_agent": "travel_advisor"}
    return Command(update=update, goto="human")
 
# 호텔 어드바이저 도구 및 ReAct 에이전트 정의
hotel_advisor_tools = [
    get_hotel_recommendations,
    make_handoff_tool(agent_name="travel_advisor"),
]
 
hotel_advisor = create_react_agent(
    model,
    hotel_advisor_tools,
    prompt=(
        "You are a hotel expert that can provide hotel recommendations for a given destination. "
        "If you need help picking travel destinations, ask 'travel_advisor' for help."
        "You MUST include human-readable response before transferring to another agent."
    ),
)
 
def call_hotel_advisor(
    state: MultiAgentState,
) -> Command[Literal["travel_advisor", "human"]]:
    response = hotel_advisor.invoke(state)
    update = {**response, "last_active_agent": "hotel_advisor"}
    return Command(update=update, goto="human")
 
def human_node(
    state: MultiAgentState, config
) -> Command[Literal["hotel_advisor", "travel_advisor", "human"]]:
    """사용자 입력을 수집하는 노드"""
 
    user_input = interrupt(value="Ready for user input.")
    active_agent = state["last_active_agent"]
 
    return Command(
        update={
            "messages": [
                {
                    "role": "human",
                    "content": user_input,
                }
            ]
        },
        goto=active_agent,
    )
 
builder = StateGraph(MultiAgentState)
builder.add_node("travel_advisor", call_travel_advisor)
builder.add_node("hotel_advisor", call_hotel_advisor)
 
# 사람 입력을 수집하는 노드를 추가하며, 이는 활성 에이전트로 다시 라우팅됩니다
builder.add_node("human", human_node)
 
# 항상 일반 여행 어드바이저로 시작합니다
builder.add_edge(START, "travel_advisor")
 
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

이 애플리케이션으로 멀티턴 대화를 테스트해 봅시다:

import uuid
 
thread_config = {"configurable": {"thread_id": str(uuid.uuid4())}}
 
inputs = [
    # 1차 대화
    {
        "messages": [
            {"role": "user", "content": "i wanna go somewhere warm in the caribbean"}
        ]
    },
    # `interrupt`를 사용하므로 Command 프리미티브를 사용하여 재개해야 합니다
    # 2차 대화
    Command(
        resume="could you recommend a nice hotel in one of the areas and tell me which area it is."
    ),
    # 3차 대화
    Command(
        resume="i like the first one. could you recommend something to do near the hotel?"
    ),
]
 
for idx, user_input in enumerate(inputs):
    print()
    print(f"--- Conversation Turn {idx + 1} ---")
    print()
    print(f"User: {user_input}")
    print()
    for update in graph.stream(
        user_input,
        config=thread_config,
        stream_mode="updates",
    ):
        for node_id, value in update.items():
            if isinstance(value, dict) and value.get("messages", []):
                last_message = value["messages"][-1]
                if isinstance(last_message, dict) or last_message.type != "ai":
                    continue
                print(f"{node_id}: {last_message.content}")

실행 결과:

--- Conversation Turn 1 ---

User: {'messages': [{'role': 'user', 'content': 'i wanna go somewhere warm in the caribbean'}]}

travel_advisor: Based on the recommendations, Aruba would be an excellent choice for your Caribbean getaway! Aruba is known as "One Happy Island" and offers:
- Year-round warm weather with consistent temperatures around 82°F (28°C)
- Beautiful white sand beaches like Eagle Beach and Palm Beach
- Clear turquoise waters perfect for swimming and snorkeling
- Minimal rainfall and location outside the hurricane belt
- A blend of Caribbean and Dutch culture
- Great dining options and nightlife
- Various water sports and activities

Would you like me to get some specific hotel recommendations in Aruba for your stay? I can transfer you to our hotel advisor who can help with accommodations.

--- Conversation Turn 2 ---

User: Command(resume='could you recommend a nice hotel in one of the areas and tell me which area it is.')

hotel_advisor: Based on the recommendations, I can suggest two excellent options:

1. The Ritz-Carlton, Aruba
- Located in Palm Beach
- This luxury resort is situated in the vibrant Palm Beach area
- Known for its exceptional service and amenities
- Perfect if you want to be close to dining, shopping, and entertainment
- Features multiple restaurants, a casino, and a world-class spa
- Located on a pristine stretch of Palm Beach

2. Bucuti & Tara Beach Resort
- Located in Eagle Beach
- An adults-only boutique resort on Eagle Beach
- Known for being more intimate and peaceful
- Award-winning for its sustainability practices
- Perfect for a romantic getaway or peaceful vacation
- Located on one of the most beautiful beaches in the Caribbean

Would you like more specific information about either of these properties or their locations?

--- Conversation Turn 3 ---

User: Command(resume='i like the first one. could you recommend something to do near the hotel?')

travel_advisor: Near the Ritz-Carlton in Palm Beach, here are some highly recommended activities:

1. Visit the Palm Beach Plaza Mall
- Just a short walk from the hotel, featuring shopping, dining, and entertainment
2. Try your luck at the Stellaris Casino
- It's right in the Ritz-Carlton
3. Take a sunset sailing cruise
- Many depart from the nearby pier
4. Visit the California Lighthouse
- A scenic landmark just north of Palm Beach
5. Enjoy water sports at Palm Beach:
- Jet skiing
- Parasailing
- Snorkeling
- Stand-up paddleboarding

Would you like more specific information about any of these activities or would you like to know about other options in the area?

미리 빌드된 구현

LangGraph는 가장 인기 있는 두 가지 멀티 에이전트 아키텍처의 미리 빌드된 구현을 제공합니다:

  • 슈퍼바이저 — 개별 에이전트가 중앙 슈퍼바이저 에이전트에 의해 조정됩니다. 슈퍼바이저는 모든 통신 흐름과 작업 위임을 제어하며, 현재 컨텍스트와 작업 요구 사항에 따라 어떤 에이전트를 호출할지 결정합니다. langgraph-supervisor 라이브러리를 사용하여 슈퍼바이저 멀티 에이전트 시스템을 만들 수 있습니다.

  • 스웜 — 에이전트가 전문 분야에 따라 서로에게 동적으로 제어권을 넘깁니다. 시스템은 마지막으로 활성화된 에이전트를 기억하여 후속 상호작용에서 해당 에이전트와 대화를 재개할 수 있도록 합니다. langgraph-swarm 라이브러리를 사용하여 스웜 멀티 에이전트 시스템을 만들 수 있습니다.

참고 자료