[langgraph] ReAct 스타일 에이전트로 구조화된 출력을 리턴하는 방법

LangGraph 공식문서를 번역한 내용입니다. 필요한 경우 부연 설명을 추가하였고 이해하기 쉽게 예제를 일부 변경하였습니다. 문제가 되면 삭제하겠습니다.

https://langchain-ai.github.io/langgraph/how-tos/react-agent-structured-output/

에이전트의 출력을 구조화된 형식으로 반환해야 할 때가 있을 수 있다. 예를 들어, 에이전트의 출력이 다른 다운스트림 소프트웨어에서 사용되는 경우, 에이전트가 호출될 때마다 동일한 구조화된 형식으로 출력을 제공하도록 하면 일관성을 유지할 수 있다.

여기서는 툴 호출 에이전트의 출력을 구조화된 형식으로 강제하는 두 가지 방법을 알아보자. 기본 ReAct 에이전트(모델 노드와 툴 호출 노드)를 사용하며, 마지막에는 사용자에게 응답을 포맷팅하는 세 번째 노드가 추가된다. 두 가지 방법 모두 아래 다이어그램에 표시된 동일한 그래프 구조를 사용하지만, 내부 메커니즘은 다르다.

첫 번째 방법은 에이전트 노드에서 사용할 추가 도구로 원하는 출력을 바인딩하여 툴 호출 에이전트가 구조화된 출력을 강제하도록 하는 것이다. 기본 ReAct 에이전트와는 달리, 이 경우 에이전트 노드는 툴과 END 사이를 선택하는 것이 아니라 호출할 특정 툴을 선택한다. 예상 흐름은 에이전트 노드의 LLM이 먼저 액션 툴을 선택하고, 액션 툴 출력값을 받은 후 응답 툴을 호출한다. 이후 응답 툴은 에이전트 노드의 툴 호출 인수를 구조화하는 응답 노드로 라우팅된다.

장점과 단점

장점
이 방식의 장점은 LLM을 하나만 사용하므로 비용과 대기 시간이 절감된다는 점이다.

단점
그러나 단점은 단일 LLM이 항상 원하는 툴을 호출한다고 보장할 수 없다는 점이다. bind_tools를 사용할 때 tool_choice를 any로 설정하면 LLM이 매번 최소한 하나의 툴을 선택하도록 강제할 수 있지만, 이것이 완벽한 해결책은 아니다. 또 다른 단점은 에이전트가 여러 툴을 호출할 수 있다는 점이다. 따라서 라우팅 함수에서 이를 명시적으로 확인해야 한다. OpenAI를 사용하는 경우, parallell_tool_calling=False로 설정하여 한 번에 하나의 툴만 호출되도록 설정할 수 있다.

두 번째 방법은 사용자에게 응답하기 위해 두 번째 LLM(이 예에서는 model_with_structured_output)을 사용하는 것이다.

이 경우 기본 ReAct 에이전트를 정상적으로 정의하지만, 에이전트 노드는 툴 노드와 대화를 종료하는 선택지 사이에서 선택하는 대신 툴 노드와 응답 노드 사이에서 선택하게 된다. 응답 노드는 구조화된 출력을 사용하는 두 번째 LLM을 포함하며, 호출되면 직접 사용자에게 반환된다. 이 방법은 사용자에게 응답하기 전에 하나의 추가 단계를 추가한 기본 ReAct 방식으로 생각할 수 있다.

장점과 단점

장점
이 방법의 주요 장점은 .with_structured_output이 LLM에서 예상대로 작동하는 한 구조화된 출력을 보장한다는 점이다.

단점
단점은 사용자에게 응답하기 전에 추가적인 LLM 호출이 필요하다는 점으로, 이로 인해 비용과 대기 시간이 증가할 수 있다. 또한, 에이전트 노드 LLM에 원하는 출력 스키마에 대한 정보를 제공하지 않으면, 에이전트 LLM이 올바른 출력 스키마로 답변하기 위해 필요한 툴을 호출하지 못할 위험이 있다.

두 옵션 모두 동일한 그래프 구조를 따른다(위의 다이어그램 참조). 둘 다 기본 ReAct 아키텍처의 정확한 복제본이며, 대화 종료 전에 응답 노드가 추가된 형태이다.

준비

우선, 필요한 패키지를 설치하자.

pip install langgraph langchain_openai

모델, 도구, 그래프 상태 정의

이제 출력 구조를 정의하고, 그래프 상태를 설정하며, 사용할 툴과 모델을 정의해 보자.

구조화된 출력을 사용하려면 LangChain의 with_structured_output 메서드를 사용할 것이다. 이 메서드에 대한 자세한 내용은 여기에서 확인할 수 있다.

이 예제에서는 날씨를 찾는 단일 툴을 사용하고, 사용자에게 구조화된 날씨 응답을 반환할 것이다.

from typing import Literal

from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState
from pydantic import BaseModel, Field

load_dotenv()


class WeatherResponse(BaseModel):
    """Respond to the user with this"""

    temperature: float = Field(description="The temperature in fahrenheit")
    wind_directon: str = Field(
        description="The direction of the wind in abbreviated form"
    )
    wind_speed: float = Field(description="The speed of the wind in km/h")


class AgentState(MessagesState):
    final_response: WeatherResponse


@tool
def get_weather(city: Literal["서울", "부산"]):
    """Use this to get weather information."""
    if city == "서울":
        return "서울은 구름이 많고 북동풍이 불고 있고 기온은 30도이다."
    elif city == "부산":
        return "부산은 남동풍이 불고 있고 32도이고 맑다."
    else:
        raise AssertionError("Unknown city")


tools = [get_weather]

model = ChatOpenAI(model="gpt-4o-mini")

model_with_tools = model.bind_tools(tools)
model_with_structured_output = model.with_structured_output(WeatherResponse)

옵션 1: 출력을 도구로 바인딩

단일 LLM 옵션을 사용하는 방법을 알아보자.

그래프 정의

그래프 정의는 위의 예와 매우 유사하며, 유일한 차이점은 응답 노드에서 더 이상 LLM을 호출하지 않는 대신, 이미 get_weather 툴을 포함한 LLM에 WeatherResponse 툴을 바인딩한다는 점이다.

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

tools = [get_weather, WeatherResponse]

model_with_response_tool = model.bind_tools(tools, tool_choice="any")


def call_model(state: AgentState):
    response = model_with_response_tool.invoke(state["messages"])
    return {"messages": [response]}


def respond(state: AgentState):
    response = WeatherResponse(**state["messages"][-1].tool_calls[0]["args"])
    return {"final_response": response}


def should_continue(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    if (
        len(last_message.tool_calls) == 1
        and last_message.tool_calls[0]["name"] == "WeatherResponse"
    ):
        return "respond"
    else:
        return "continue"


# Define a new graph
workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("respond", respond)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "respond": "respond",
    },
)

workflow.add_edge("tools", "agent")
workflow.add_edge("respond", END)
graph = workflow.compile()

사용법

이제 의도한 대로 동작하는지 확인해보자.

answer = graph.invoke(input={"messages": [("human", "서울 날씨 어때?")]})[
    "final_response"
]

print(answer)

temperature=30.0 wind_directon='NE' wind_speed=8.0

의도한 대로 WeatherResponse를 리턴했다.

옵션 2: 2 LLM

두 번째 LLM이 구조화된 출력을 강제하도록 사용하는 방법을 알아보자.

그래프 정의

from typing import Literal

from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState
from pydantic import BaseModel, Field

load_dotenv()


class WeatherResponse(BaseModel):
    """Respond to the user with this"""

    temperature: float = Field(description="The temperature in celsius")
    wind_directon: str = Field(
        description="The direction of the wind in abbreviated form"
    )
    wind_speed: float = Field(description="The speed of the wind in km/h")


class AgentState(MessagesState):
    final_response: WeatherResponse


@tool
def get_weather(city: Literal["서울", "부산"]):
    """Use this to get weather information."""
    if city == "서울":
        return "서울은 구름이 많고 5 mph 북동풍이 불고 있고 기온은 30도이다."
    elif city == "부산":
        return "부산은 5 mph 남동풍이 불고 있고 32도이고 맑다."
    else:
        raise AssertionError("Unknown city")


tools = [get_weather]

model = ChatOpenAI(model="gpt-4o-mini")

model_with_tools = model.bind_tools(tools)
model_with_structured_output = model.with_structured_output(WeatherResponse)

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage


def call_model(state: AgentState):
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}


def respond(state: AgentState):
    response = model_with_structured_output.invoke(
        [HumanMessage(content=state["messages"][-2].content)]
    )
    return {"final_response": response}


def should_continue(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    if not last_message.tool_calls:
        return "respond"
    else:
        return "continue"


workflow = StateGraph(AgentState)

workflow.add_node("agent", call_model)
workflow.add_node("respond", respond)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "respond": "respond",
    },
)

workflow.add_edge("tools", "agent")
workflow.add_edge("respond", END)
graph = workflow.compile()

사용법

기대한 대로 구조화된 출력으로 나타나는지 확인할 수 있다.

answer = graph.invoke(input={"messages": [("human", "서울 날씨 어때?")]})[
    "final_response"
]

print(answer)

temperature=30.0 wind_directon='NE' wind_speed=8.0

에이전트는 예상대로 WeatherResponse 객체를 반환했다. 이제 이 에이전트를 더 복잡한 소프트웨어 스택에서 사용하더라도, 에이전트의 출력이 다음 단계에서 기대하는 형식과 일치하지 않을까 걱정할 필요가 없다.

LangGraph 참고 자료

• Controllability
  • 병렬 실행 브랜치 생성하는 방법
  • map-reduce 브랜치를 병렬 실행하는 방법
  • 그래프 재귀 한도를 제한하는 방법
• Persistence
  • 그래프에 스레드 수준 영속성 추가하는 방법
  • 서브그래프에 스레드 수준 영속성을 추가하는 방법
  • 그래프에 여러 스레드 간 영속성을 추가하는 방법
  • Postgres 영속성 체크포인터 사용하는 방법
• Memory
  • 대화 이력을 관리하는 방법
  • 메시지 삭제하는 방법
  • 대화 이력을 요약하는 방법
• Human-in-the-loop
  • 중단 지점 추가하는 방법
  • 동적 중단 지점을 추가하는 방법
  • 그래프 상태를 수정하는 방법
  • 사용자 입력을 대기하는 방법
  • 그래프의 과거 상태를 보고 수정하는 방법
  • 도구 호출을 리뷰하는 방법
• Streaming
  • 그래프의 전체 상태를 스트리밍하는 방법
  • 그래프의 상태 업데이트를 스트리밍하는 방법
  • 그래프에서 LLM 토큰을 스트리밍하는 방법
  • LLM 토큰을 스트리밍하는 방법 (LangChain을 사용하지 않고)
  • 커스텀 데이터를 스트리밍하는 방법
  • 동시에 여러 스트리밍 모드를 구성하는 방법
  • 도구 내에 이벤트를 스트리밍하는 방법
  • 도구 내에 이벤트를 스트리밍하는 방법 (LangChain을 사용하지 않고)
  • 최종 노드에서 스트리밍하는 방법
  • 서브그래프에서 스트리밍하는 방법
  • 지원하지 않는 모델에서 스트리밍을 비활성화하는 방법
• Tool calling
  • ToolNode를 사용하여 도구를 호출하는 방법
  • 도구 호출 시 에러 처리하는 방법
  • 도구에 런타임 값을 전달하는 방법
  • 도구에 구성 전달하는 방법
  • 많은 수의 도구를 처리하는 방법
• Subgraphs
  • 서브 그래프를 추가하고 사용하는 방법
  • 서브 그래프에서 상태를 조회하고 업데이트하는 방법
  • 서브 그래프의 입/출력값을 변환하는 방법
• State Management
  • 상태에 pydantic 모델을 사용하는 방법
  • 그래프의 입출력 스키마를 정의하는 방법
  • 노드 간 비공개 상태 전달하는 방법
• Other
  • 그래프를 비동기로 실행하는 방법
  • 그래프를 시각화하는 방법
  • 그래프에 런타임 구성을 하는 방법
  • 노드 재시도 정책 추가하는 방법
  • ReAct 스타일 에이전트로 구조화된 출력을 리턴하는 방법
  • LangSmith에서 그래프 실행 시 실행 ID, 태그와 메타데이터를 전달하는 방법
  • 재귀 횟수에 도달하기 전에 상태를 리턴하는 방법
  • AutoGen, CrewAI 외 다른 프레임워크와 LangGraph 통합하는 방법
• Prebuilt ReAct Agent
  • ReAct 에이전트를 사용하는 방법
  • ReAct 에이전트에 메모리 추가하는 방법
  • ReAct 에이전트에 사용자 정의 시스템 프롬프트 추가하는 방법
  • ReAct 에이전트에 사람이 개입하는 프로세스를 추가하는 방법
  • 처음부터 ReAct 에이전트를 만드는 방법