[langgraph] Multi-agent

[langgraph] Multi-agent

원문 출처: https://langchain-ai.github.io/langgraph/agents/multi-agent/

하나의 에이전트가 여러 도메인에 특화되거나 많은 도구를 관리해야 할 때는 한계가 있습니다. 이를 해결하기 위해, 에이전트를 더 작고 독립적인 여러 에이전트로 분할하고, 이들을 조합해 멀티 에이전트 시스템을 만들 수 있습니다.

멀티 에이전트 시스템에서는 에이전트 간 통신이 필요합니다. 이때 handoff(핸드오프)라는 프리미티브를 사용해, 어떤 에이전트에게 제어권을 넘기고 어떤 데이터를 전달할지 지정합니다.

대표적인 멀티 에이전트 아키텍처는 다음과 같습니다:

• Supervisor(감독자): 개별 에이전트들을 중앙의 supervisor(감독자) 에이전트가 조율합니다. supervisor는 모든 통신 흐름과 작업 할당을 제어하며, 현재 컨텍스트와 작업 요구에 따라 어떤 에이전트를 호출할지 결정합니다.
• Swarm(군집): 에이전트들이 각자의 전문성에 따라 동적으로 제어권을 넘깁니다. 시스템은 마지막으로 활성화된 에이전트를 기억해, 이후 상호작용에서 해당 에이전트와 대화를 이어갑니다.

---

Supervisor(감독자)

langgraph-supervisor 라이브러리를 사용해 supervisor 기반 멀티 에이전트 시스템을 만들 수 있습니다.

pip install langgraph-supervisor

API Reference: ChatOpenAI | create_react_agent | create_supervisor

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor

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="openai:gpt-4o",
    tools=[book_flight],
    prompt="You are a flight booking assistant",
    name="flight_assistant"
)

hotel_assistant = create_react_agent(
    model="openai:gpt-4o",
    tools=[book_hotel],
    prompt="You are a hotel booking assistant",
    name="hotel_assistant"
)

supervisor = create_supervisor(
    agents=[flight_assistant, hotel_assistant],
    model=ChatOpenAI(model="gpt-4o"),
    prompt=(
        "You manage a hotel booking assistant and a"
        "flight booking assistant. Assign work to them."
    )
).compile()

for chunk in supervisor.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
            }
        ]
    }
):
    print(chunk)
    print("\n")

---

Swarm(군집)

langgraph-swarm 라이브러리를 사용해 swarm 기반 멀티 에이전트 시스템을 만들 수 있습니다.

pip install langgraph-swarm

API Reference: create_react_agent | create_swarm | create_handoff_tool

from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_swarm, create_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"
)

swarm = create_swarm(
    agents=[flight_assistant, hotel_assistant],
    default_active_agent="flight_assistant"
).compile()

for chunk in swarm.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
            }
        ]
    }
):
    print(chunk)
    print("\n")

---

Handoffs(핸드오프)

멀티 에이전트 상호작용에서 흔히 사용되는 패턴이 handoffs(핸드오프)입니다. 한 에이전트가 다른 에이전트에게 제어권을 넘기는 방식입니다. handoff는 다음을 지정할 수 있습니다:

• destination: 제어권을 넘길 대상 에이전트
• payload: 해당 에이전트에게 전달할 정보

이 방식은 supervisor(감독자) 구조와 swarm(군집) 구조 모두에서 사용됩니다.

핸드오프를 구현하려면:

1. 제어권을 넘기는 특수 도구를 만듭니다.

   def transfer_to_bob():
    """bob 에이전트로 제어권 전달"""
    return Command(
        goto="bob",  # 이동할 에이전트(노드) 이름
        update={"messages": [...]},  # 전달할 데이터
        graph=Command.PARENT,  # 부모 그래프의 에이전트 노드로 이동
    )

2. 각 에이전트에 handoff 도구를 추가합니다.

   flight_assistant = create_react_agent(
    ..., tools=[book_flight, transfer_to_hotel_assistant]
   )
   hotel_assistant = create_react_agent(
    ..., tools=[book_hotel, transfer_to_flight_assistant]
   )

3. 개별 에이전트를 노드로 포함하는 부모 그래프를 정의합니다.

   from langgraph.graph import StateGraph, MessagesState
   multi_agent_graph = (
    StateGraph(MessagesState)
    .add_node(flight_assistant)
    .add_node(hotel_assistant)
    ...
   )

전체 예시:

API Reference: tool | InjectedToolCallId | create_react_agent | InjectedState | StateGraph | START | 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],
        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={"messages": state["messages"] + [tool_message]},
            graph=Command.PARENT,
        )
    return handoff_tool

# Handoffs
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"
            }
        ]
    }
):
    print(chunk)
    print("\n")

이 handoff 구현은 각 에이전트가 전체 메시지 히스토리를 입력으로 받고, 자신의 내부 메시지 히스토리를 전체 시스템의 메시지 히스토리에 출력하는 구조를 가정합니다.

더 다양한 handoff 커스터마이징은 LangGraph supervisor, swarm 문서를 참고하세요.