멀티 에이전트 시스템

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

개요

에이전트는 LLM을 활용하여 애플리케이션의 제어 흐름을 결정하는 시스템입니다. 이러한 시스템을 개발하다 보면 시간이 지남에 따라 복잡성이 증가하여 관리와 확장이 어려워질 수 있습니다. 예를 들어, 다음과 같은 문제가 발생할 수 있습니다:

• 에이전트가 너무 많은 도구를 보유하여 다음에 호출할 도구를 제대로 선택하지 못함
• 단일 에이전트가 추적하기에는 컨텍스트가 너무 복잡함
• 시스템에 여러 전문 분야가 필요함(예: 계획 수립자, 연구원, 수학 전문가 등)

이러한 문제를 해결하기 위해 애플리케이션을 여러 개의 작고 독립적인 에이전트로 분리하고 이들을 멀티 에이전트 시스템으로 구성할 수 있습니다. 이러한 독립적인 에이전트는 프롬프트와 LLM 호출만큼 간단할 수도 있고, ReAct 에이전트처럼 복잡할 수도 있습니다.

멀티 에이전트 시스템을 사용하는 주요 이점은 다음과 같습니다:

• 모듈성: 개별 에이전트로 분리하면 에이전트 시스템의 개발, 테스트 및 유지 관리가 더욱 쉬워집니다.
• 전문성: 특정 도메인에 집중하는 전문 에이전트를 만들 수 있어 전체 시스템 성능 향상에 도움이 됩니다.
• 제어: 에이전트 간 통신 방식을 명시적으로 제어할 수 있습니다(함수 호출에 의존하는 대신).

멀티 에이전트 아키텍처

멀티 에이전트 시스템에서 에이전트를 연결하는 방법은 여러 가지가 있습니다:

• 네트워크: 각 에이전트가 다른 모든 에이전트와 통신할 수 있습니다. 모든 에이전트가 다음에 호출할 에이전트를 결정할 수 있습니다.
• 슈퍼바이저: 각 에이전트가 단일 슈퍼바이저 에이전트와 통신합니다. 슈퍼바이저 에이전트가 다음에 호출할 에이전트를 결정합니다.
• 슈퍼바이저(도구 호출): 이것은 슈퍼바이저 아키텍처의 특수한 경우입니다. 개별 에이전트를 도구로 표현할 수 있습니다. 이 경우 슈퍼바이저 에이전트는 도구 호출 LLM을 사용하여 어떤 에이전트 도구를 호출할지, 그리고 해당 에이전트에게 전달할 인수를 결정합니다.
• 계층적: 슈퍼바이저의 슈퍼바이저가 있는 멀티 에이전트 시스템을 정의할 수 있습니다. 이는 슈퍼바이저 아키텍처의 일반화로, 더 복잡한 제어 흐름을 허용합니다.
• 사용자 정의 멀티 에이전트 워크플로우: 각 에이전트가 에이전트의 하위 집합과만 통신합니다. 흐름의 일부는 결정론적이며, 일부 에이전트만 다음에 호출할 다른 에이전트를 결정할 수 있습니다.

핸드오프

멀티 에이전트 아키텍처에서 에이전트는 그래프 노드로 표현될 수 있습니다. 각 에이전트 노드는 단계를 실행하고 실행을 종료할지 또는 다른 에이전트로 라우팅할지 결정합니다(자기 자신으로 라우팅하여 루프로 실행할 수도 있음). 멀티 에이전트 상호작용의 일반적인 패턴은 핸드오프로, 한 에이전트가 제어권을 다른 에이전트에게 넘깁니다. 핸드오프를 통해 다음을 지정할 수 있습니다:

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

LangGraph에서 핸드오프를 구현하려면 에이전트 노드가 제어 흐름과 상태 업데이트를 모두 결합할 수 있는 Command 객체를 반환할 수 있습니다:

def agent(state) -> Command[Literal["agent", "another_agent"]]:
    # 라우팅/중단 조건은 무엇이든 될 수 있음(예: LLM 도구 호출/구조화된 출력 등)
    goto = get_next_agent(...)  # 'agent' / 'another_agent'
    return Command(
        # 다음에 호출할 에이전트 지정
        goto=goto,
        # 그래프 상태 업데이트
        update={"my_state_key": "my_state_value"}
    )

각 에이전트 노드가 그 자체로 그래프인(즉, 서브그래프) 더 복잡한 시나리오에서는 에이전트 서브그래프 중 하나의 노드가 다른 에이전트로 이동하려고 할 수 있습니다. 예를 들어, 두 개의 에이전트 alice와 bob(부모 그래프의 서브그래프 노드)이 있고 alice가 bob으로 이동해야 하는 경우, Command 객체에서 graph=Command.PARENT를 설정할 수 있습니다:

def some_node_inside_alice(state):
    return Command(
        goto="bob",
        update={"my_state_key": "my_state_value"},
        # 이동할 그래프 지정(기본값은 현재 그래프)
        graph=Command.PARENT,
    )

참고

Command(graph=Command.PARENT)를 사용하여 통신하는 서브그래프의 시각화를 지원해야 하는 경우, 노드 함수를 Command 어노테이션으로 래핑해야 합니다:

이렇게 하는 대신:

builder.add_node(alice)

이렇게 해야 합니다:

def call_alice(state) -> Command[Literal["bob"]]:
    return alice.invoke(state)
 
builder.add_node("alice", call_alice)

도구로서의 핸드오프

가장 일반적인 에이전트 유형 중 하나는 도구 호출 에이전트입니다. 이러한 유형의 에이전트의 경우, 핸드오프를 도구 호출로 래핑하는 것이 일반적인 패턴입니다:

from langchain_core.tools import tool
 
@tool
def transfer_to_bob():
    """bob에게 전송합니다."""
    return Command(
        # 이동할 에이전트(노드)의 이름
        goto="bob",
        # 에이전트에게 전송할 데이터
        update={"my_state_key": "my_state_value"},
        # LangGraph에게 부모 그래프의 에이전트 노드로
        # 이동해야 함을 나타냄
        graph=Command.PARENT,
    )

이것은 제어 흐름 외에도 상태 업데이트가 포함된 도구에서 그래프 상태를 업데이트하는 특수한 경우입니다.

중요

Command를 반환하는 도구를 사용하려면 미리 빌드된 create_react_agent / ToolNode 컴포넌트를 사용하거나, 자체 로직을 구현할 수 있습니다:

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

이제 다양한 멀티 에이전트 아키텍처를 자세히 살펴보겠습니다.

네트워크

이 아키텍처에서 에이전트는 그래프 노드로 정의됩니다. 각 에이전트는 다른 모든 에이전트와 통신할 수 있으며(다대다 연결), 다음에 호출할 에이전트를 결정할 수 있습니다. 이 아키텍처는 에이전트의 명확한 계층 구조나 에이전트를 호출해야 하는 특정 순서가 없는 문제에 적합합니다.

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END
 
model = ChatOpenAI()
 
def agent_1(state: MessagesState) -> Command[Literal["agent_2", "agent_3", END]]:
    # 상태의 관련 부분을 LLM에 전달하여(예: state["messages"])
    # 다음에 호출할 에이전트를 결정할 수 있음. 일반적인 패턴은
    # 구조화된 출력으로 모델을 호출하는 것(예: "next_agent" 필드가 있는 출력 반환 강제)
    response = model.invoke(...)
    # LLM의 결정에 따라 에이전트 중 하나로 라우팅하거나 종료
    # LLM이 "__end__"를 반환하면 그래프가 실행을 종료함
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )
 
def agent_2(state: MessagesState) -> Command[Literal["agent_1", "agent_3", END]]:
    response = model.invoke(...)
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )
 
def agent_3(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]:
    ...
    return Command(
        goto=response["next_agent"],
        update={"messages": [response["content"]]},
    )
 
builder = StateGraph(MessagesState)
builder.add_node(agent_1)
builder.add_node(agent_2)
builder.add_node(agent_3)
 
builder.add_edge(START, "agent_1")
network = builder.compile()

슈퍼바이저

이 아키텍처에서는 에이전트를 노드로 정의하고, 다음에 어떤 에이전트 노드를 호출해야 할지 결정하는 슈퍼바이저 노드(LLM)를 추가합니다. 슈퍼바이저의 결정에 따라 적절한 에이전트 노드로 실행을 라우팅하기 위해 Command를 사용합니다. 이 아키텍처는 여러 에이전트를 병렬로 실행하거나 맵-리듀스 패턴을 사용하는 데에도 적합합니다.

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState, START, END
 
model = ChatOpenAI()
 
def supervisor(state: MessagesState) -> Command[Literal["agent_1", "agent_2", END]]:
    # 상태의 관련 부분을 LLM에 전달하여(예: state["messages"])
    # 다음에 호출할 에이전트를 결정할 수 있음. 일반적인 패턴은
    # 구조화된 출력으로 모델을 호출하는 것(예: "next_agent" 필드가 있는 출력 반환 강제)
    response = model.invoke(...)
    # 슈퍼바이저의 결정에 따라 에이전트 중 하나로 라우팅하거나 종료
    # 슈퍼바이저가 "__end__"를 반환하면 그래프가 실행을 종료함
    return Command(goto=response["next_agent"])
 
def agent_1(state: MessagesState) -> Command[Literal["supervisor"]]:
    # 상태의 관련 부분을 LLM에 전달하고(예: state["messages"])
    # 추가 로직을 추가할 수 있음(다른 모델, 사용자 정의 프롬프트, 구조화된 출력 등)
    response = model.invoke(...)
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )
 
def agent_2(state: MessagesState) -> Command[Literal["supervisor"]]:
    response = model.invoke(...)
    return Command(
        goto="supervisor",
        update={"messages": [response]},
    )
 
builder = StateGraph(MessagesState)
builder.add_node(supervisor)
builder.add_node(agent_1)
builder.add_node(agent_2)
 
builder.add_edge(START, "supervisor")
 
supervisor = builder.compile()

슈퍼바이저(도구 호출)

슈퍼바이저 아키텍처의 이 변형에서는 서브 에이전트를 호출하는 슈퍼바이저 에이전트를 정의합니다. 서브 에이전트는 슈퍼바이저에게 도구로 노출되며, 슈퍼바이저 에이전트는 어떤 에이전트 도구를 호출할지와 해당 에이전트에게 전달할 인수를 결정합니다. 슈퍼바이저 에이전트는 도구를 호출하는 LLM이 중지하기로 결정할 때까지 while 루프에서 실행되는 표준 구현을 따릅니다.

from typing import Annotated
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import InjectedState, create_react_agent
 
model = ChatOpenAI()
 
# 이것은 도구로 호출될 에이전트 함수
# InjectedState 어노테이션을 통해 도구에 상태를 전달할 수 있음을 주목
def agent_1(state: Annotated[dict, InjectedState]):
    # 상태의 관련 부분을 LLM에 전달하고(예: state["messages"])
    # 추가 로직을 추가할 수 있음(다른 모델, 사용자 정의 프롬프트, 구조화된 출력 등)
    response = model.invoke(...)
    # LLM 응답을 문자열로 반환(예상되는 도구 응답 형식)
    # 이것은 미리 빌드된 create_react_agent(슈퍼바이저)에 의해
    # 자동으로 ToolMessage로 변환됨
    return response.content
 
def agent_2(state: Annotated[dict, InjectedState]):
    response = model.invoke(...)
    return response.content
 
tools = [agent_1, agent_2]
# 도구 호출 기능이 있는 슈퍼바이저를 구축하는 가장 간단한 방법은
# 미리 빌드된 ReAct 에이전트 그래프를 사용하는 것
# 도구 호출 LLM 노드(즉, 슈퍼바이저)와 도구 실행 노드로 구성됨
supervisor = create_react_agent(model, tools)

계층적

시스템에 더 많은 에이전트를 추가하면 슈퍼바이저가 모든 에이전트를 관리하기 너무 어려워질 수 있습니다. 슈퍼바이저가 다음에 호출할 에이전트에 대해 잘못된 결정을 내리기 시작하거나, 단일 슈퍼바이저가 추적하기에는 컨텍스트가 너무 복잡해질 수 있습니다. 즉, 애초에 멀티 에이전트 아키텍처를 채택하게 된 것과 동일한 문제가 발생하게 됩니다.

이를 해결하기 위해 시스템을 계층적으로 설계할 수 있습니다. 예를 들어, 개별 슈퍼바이저가 관리하는 전문화된 에이전트 팀을 별도로 만들고, 최상위 슈퍼바이저가 팀을 관리하도록 할 수 있습니다.

from typing import Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command
 
model = ChatOpenAI()
 
# 팀 1 정의(위의 단일 슈퍼바이저 예제와 동일)
def team_1_supervisor(state: MessagesState) -> Command[Literal["team_1_agent_1", "team_1_agent_2", END]]:
    response = model.invoke(...)
    return Command(goto=response["next_agent"])
 
def team_1_agent_1(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:
    response = model.invoke(...)
    return Command(goto="team_1_supervisor", update={"messages": [response]})
 
def team_1_agent_2(state: MessagesState) -> Command[Literal["team_1_supervisor"]]:
    response = model.invoke(...)
    return Command(goto="team_1_supervisor", update={"messages": [response]})
 
team_1_builder = StateGraph(Team1State)
team_1_builder.add_node(team_1_supervisor)
team_1_builder.add_node(team_1_agent_1)
team_1_builder.add_node(team_1_agent_2)
team_1_builder.add_edge(START, "team_1_supervisor")
team_1_graph = team_1_builder.compile()
 
# 팀 2 정의(위의 단일 슈퍼바이저 예제와 동일)
class Team2State(MessagesState):
    next: Literal["team_2_agent_1", "team_2_agent_2", "__end__"]
 
def team_2_supervisor(state: Team2State):
    ...
 
def team_2_agent_1(state: Team2State):
    ...
 
def team_2_agent_2(state: Team2State):
    ...
 
team_2_builder = StateGraph(Team2State)
...
team_2_graph = team_2_builder.compile()
 
# 최상위 슈퍼바이저 정의
 
builder = StateGraph(MessagesState)
def top_level_supervisor(state: MessagesState) -> Command[Literal["team_1_graph", "team_2_graph", END]]:
    # 상태의 관련 부분을 LLM에 전달하여(예: state["messages"])
    # 다음에 호출할 팀을 결정할 수 있음. 일반적인 패턴은
    # 구조화된 출력으로 모델을 호출하는 것(예: "next_team" 필드가 있는 출력 반환 강제)
    response = model.invoke(...)
    # 슈퍼바이저의 결정에 따라 팀 중 하나로 라우팅하거나 종료
    # 슈퍼바이저가 "__end__"를 반환하면 그래프가 실행을 종료함
    return Command(goto=response["next_team"])
 
builder = StateGraph(MessagesState)
builder.add_node(top_level_supervisor)
builder.add_node("team_1_graph", team_1_graph)
builder.add_node("team_2_graph", team_2_graph)
builder.add_edge(START, "top_level_supervisor")
builder.add_edge("team_1_graph", "top_level_supervisor")
builder.add_edge("team_2_graph", "top_level_supervisor")
graph = builder.compile()

사용자 정의 멀티 에이전트 워크플로우

이 아키텍처에서는 개별 에이전트를 그래프 노드로 추가하고 에이전트가 호출되는 순서를 사용자 정의 워크플로우에서 미리 정의합니다. LangGraph에서 워크플로우는 두 가지 방식으로 정의할 수 있습니다:

• 명시적 제어 흐름(일반 에지): LangGraph를 사용하면 일반 그래프 에지를 통해 애플리케이션의 제어 흐름(즉, 에이전트가 통신하는 순서)을 명시적으로 정의할 수 있습니다. 이것은 위의 아키텍처 중 가장 결정론적인 변형으로, 다음에 호출될 에이전트를 미리 항상 알고 있습니다.

• 동적 제어 흐름(Command): LangGraph에서 LLM이 애플리케이션 제어 흐름의 일부를 결정하도록 할 수 있습니다. 이는 Command를 사용하여 달성할 수 있습니다. 이것의 특수한 경우가 슈퍼바이저 도구 호출 아키텍처입니다. 이 경우 슈퍼바이저 에이전트를 구동하는 도구 호출 LLM이 도구(에이전트)가 호출되는 순서를 결정합니다.

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START
 
model = ChatOpenAI()
 
def agent_1(state: MessagesState):
    response = model.invoke(...)
    return {"messages": [response]}
 
def agent_2(state: MessagesState):
    response = model.invoke(...)
    return {"messages": [response]}
 
builder = StateGraph(MessagesState)
builder.add_node(agent_1)
builder.add_node(agent_2)
# 흐름을 명시적으로 정의
builder.add_edge(START, "agent_1")
builder.add_edge("agent_1", "agent_2")

통신 및 상태 관리

멀티 에이전트 시스템을 구축할 때 가장 중요한 것은 에이전트가 통신하는 방식을 파악하는 것입니다.

에이전트가 통신하는 일반적이고 보편적인 방법은 메시지 목록을 통하는 것입니다. 이는 다음과 같은 질문을 제기합니다:

• 에이전트가 핸드오프를 통해 통신하는가 아니면 도구 호출을 통해 통신하는가?
• 한 에이전트에서 다음 에이전트로 어떤 메시지가 전달되는가?
• 핸드오프가 메시지 목록에서 어떻게 표현되는가?
• 서브 에이전트의 상태를 어떻게 관리하는가?

또한 더 복잡한 에이전트를 다루거나 개별 에이전트 상태를 멀티 에이전트 시스템 상태와 별도로 유지하려는 경우, 다른 상태 스키마를 사용해야 할 수 있습니다.

핸드오프 vs 도구 호출

에이전트 간에 전달되는 “페이로드”는 무엇일까요? 위에서 논의한 대부분의 아키텍처에서 에이전트는 핸드오프를 통해 통신하고 핸드오프 페이로드의 일부로 그래프 상태를 전달합니다. 특히 에이전트는 그래프 상태의 일부로 메시지 목록을 전달합니다. 도구 호출을 사용하는 슈퍼바이저의 경우, 페이로드는 도구 호출 인수입니다.

에이전트 간 메시지 전달

에이전트가 통신하는 가장 일반적인 방법은 공유 상태 채널, 일반적으로 메시지 목록을 통하는 것입니다. 이는 에이전트가 공유하는 상태에 항상 적어도 하나의 채널(키)이 있다고 가정합니다(예: messages). 공유 메시지 목록을 통해 통신할 때 추가로 고려해야 할 사항이 있습니다: 에이전트가 사고 과정의 전체 히스토리를 공유해야 하는가, 아니면 최종 결과만 공유해야 하는가?

전체 사고 과정 공유

에이전트는 다른 모든 에이전트와 사고 과정의 전체 히스토리(즉, “스크래치패드”)를 공유할 수 있습니다. 이 “스크래치패드”는 일반적으로 메시지 목록처럼 보입니다. 전체 사고 과정을 공유하는 이점은 다른 에이전트가 더 나은 결정을 내리는 데 도움이 될 수 있고 시스템 전체의 추론 능력을 향상시킬 수 있다는 것입니다. 단점은 에이전트의 수와 복잡성이 증가함에 따라 “스크래치패드”가 빠르게 커지고 메모리 관리를 위한 추가 전략이 필요할 수 있다는 것입니다.

최종 결과만 공유

에이전트는 자체 비공개 “스크래치패드”를 가지고 최종 결과만 나머지 에이전트와 공유할 수 있습니다. 이 접근 방식은 많은 에이전트가 있거나 더 복잡한 에이전트가 있는 시스템에 더 적합할 수 있습니다. 이 경우 다른 상태 스키마로 에이전트를 정의해야 합니다.

도구로 호출되는 에이전트의 경우, 슈퍼바이저가 도구 스키마를 기반으로 입력을 결정합니다. 또한 LangGraph는 런타임에 개별 도구에 상태를 전달할 수 있으므로, 필요한 경우 하위 에이전트가 부모 상태에 액세스할 수 있습니다.

메시지에 에이전트 이름 표시

특히 긴 메시지 히스토리의 경우, 특정 AI 메시지가 어떤 에이전트로부터 온 것인지 표시하는 것이 도움이 될 수 있습니다. 일부 LLM 제공자(예: OpenAI)는 메시지에 name 매개변수를 추가하는 것을 지원합니다. 이를 사용하여 메시지에 에이전트 이름을 첨부할 수 있습니다. 지원되지 않는 경우, 에이전트 이름을 메시지 콘텐츠에 수동으로 삽입하는 것을 고려할 수 있습니다(예: <agent>alice</agent><message>alice의 메시지</message>).

메시지 히스토리에서 핸드오프 표현

핸드오프는 일반적으로 전용 핸드오프 도구를 호출하는 LLM을 통해 수행됩니다. 이것은 다음 에이전트(LLM)에게 전달되는 도구 호출이 있는 AI 메시지로 표현됩니다. 대부분의 LLM 제공자는 해당 도구 메시지 없이 도구 호출이 있는 AI 메시지를 받는 것을 지원하지 않습니다.

따라서 두 가지 옵션이 있습니다:

1. 메시지 목록에 추가 도구 메시지를 추가합니다(예: “에이전트 X로 성공적으로 전송됨”).
2. 도구 호출이 있는 AI 메시지를 제거합니다.

실제로는 대부분의 개발자가 옵션 (1)을 선택합니다.

서브 에이전트의 상태 관리

일반적인 관행은 공유 메시지 목록에서 여러 에이전트가 통신하지만 목록에 최종 메시지만 추가하는 것입니다. 이는 모든 중간 메시지(예: 도구 호출)가 이 목록에 저장되지 않음을 의미합니다.

이러한 메시지를 저장하여 나중에 이 특정 서브 에이전트가 호출될 때 다시 전달할 수 있도록 하려면 어떻게 해야 할까요?

이를 달성하는 두 가지 고수준 접근 방식이 있습니다:

1. 이러한 메시지를 공유 메시지 목록에 저장하되, 서브 에이전트 LLM에 전달하기 전에 목록을 필터링합니다. 예를 들어, 다른 에이전트의 모든 도구 호출을 필터링하도록 선택할 수 있습니다.
2. 서브 에이전트의 그래프 상태에 각 에이전트에 대한 별도의 메시지 목록(예: alice_messages)을 저장합니다. 이것이 메시지 히스토리가 어떻게 보이는지에 대한 그들의 “뷰”가 됩니다.

다른 상태 스키마 사용

에이전트가 다른 에이전트와 다른 상태 스키마를 가져야 할 수 있습니다. 예를 들어, 검색 에이전트는 쿼리와 검색된 문서만 추적하면 될 수 있습니다. LangGraph에서 이를 달성하는 두 가지 방법이 있습니다:

• 별도의 상태 스키마를 가진 서브그래프 에이전트를 정의합니다. 서브그래프와 부모 그래프 간에 공유되는 상태 키(채널)가 없는 경우, 부모 그래프가 서브그래프와 통신하는 방법을 알 수 있도록 입력/출력 변환을 추가하는 것이 중요합니다.
• 전체 그래프 상태 스키마와 구별되는 비공개 입력 상태 스키마를 가진 에이전트 노드 함수를 정의합니다. 이를 통해 특정 에이전트를 실행하는 데만 필요한 정보를 전달할 수 있습니다.