LangGraph Tool Calling 챗봇 완성 – ToolNode부터 구조화 출력까지

 

1. LangGraph로 Tool Calling 그래프 완성하기

앞서 bind_tools()로 LLM에 도구를 연결하는 방법을 살펴봤습니다.
이번에는 이것을 LangGraph 그래프로 연결해 실제로 동작하는 챗봇을 만들어 봅니다.

전체 구조는 다음과 같습니다.

START → chatbot ──(tool_calls 있음)──→ tools → chatbot
                └─(tool_calls 없음)──→ END

1.1 State와 chatbot 노드 정의

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_tavily import TavilySearch
from langchain_openai import ChatOpenAI

class State(TypedDict):
    messages: Annotated[list, add_messages]

tool = TavilySearch(max_results=2)
tools = [tool]
llm = ChatOpenAI(model="gpt-4o")
llm_with_tools = llm.bind_tools(tools)

graph_builder = StateGraph(State)

def chatbot(state: State):
    # LLM이 일반 답변 or tool_calls 중 하나를 반환
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)

1.2 BasicToolNode – 도구 실행 노드 직접 구현

ToolNode는 LangGraph에서 도구 호출 결과를 처리하는 노드입니다.
LLM이 반환한 AIMessage.tool_calls를 읽어
각 호출에 맞는 도구를 실행하고, ToolMessage로 변환해 대화 상태에 추가합니다.

내부 동작을 이해하기 위해 직접 구현하면 다음과 같습니다.

import json
from langchain_core.messages import ToolMessage

class BasicToolNode:
    def __init__(self, tools: list) -> None:
        # 도구 이름으로 빠르게 찾을 수 있도록 dict로 관리
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]  # 가장 마지막 메시지(LLM 응답)
        else:
            raise ValueError("No message found in input")

        outputs = []
        for tool_call in message.tool_calls:  # LLM이 요청한 tool_calls 순회
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}

실제 사용 시에는 LangGraph 내장 ToolNode를 사용하면 됩니다.

from langgraph.prebuilt import ToolNode

tool_node = BasicToolNode(tools=[tool])   # 직접 구현
# tool_node = ToolNode(tools=[tool])      # 내장 사용 (권장)

graph_builder.add_node("tools", tool_node)

1.3 route_tools – 조건부 분기

LLM 응답에 tool_calls가 있으면 tools 노드로, 없으면 END로 이동합니다.

def route_tools(state: State):
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state: {state}")

    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return END

graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", END: END},
)

LangGraph 내장 tools_condition을 사용하면 위 route_tools를 직접 구현하지 않아도 됩니다.

1.4 그래프 완성 및 실행

graph_builder.add_edge("tools", "chatbot")  # 도구 실행 후 chatbot으로 복귀
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

스트리밍으로 응답을 출력하는 함수와 대화 루프를 추가하면 챗봇이 완성됩니다.

def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

while True:
    user_input = input("User: ")
    if user_input.lower() in ["quit", "exit", "q"]:
        print("Goodbye!")
        break
    stream_graph_updates(user_input)

---

2. create_react_agent – 한 줄로 ReAct 에이전트 만들기

위처럼 그래프를 직접 구성하는 대신,
create_react_agent 를 사용하면 ReAct 패턴의 에이전트를 한 줄로 완성할 수 있습니다.

ReAct(Reason + Act) 란?
LLM이 "추론(Reason) → 도구 호출(Act) → 결과 반영 → 재추론"을 반복하며
최종 답변에 도달하는 에이전트 패턴입니다.

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch

tool = TavilySearch(max_results=2)
tools = [tool]
llm = ChatOpenAI(model="gpt-4o")

agent = create_react_agent(llm, tools)

실행은 동일하게 invoke()를 사용합니다.

response = agent.invoke({"messages": [{"role": "user", "content": "What is LangGraph?"}]})

create_react_agent는:

• 프롬프트(시스템 메시지), LLM, 도구 목록, 상태 스키마 등을 표준화된 방식으로 감싸고
• chatbot → tools → chatbot 루프를 자동으로 오케스트레이션합니다.
• OpenAI 외에도 다양한 LLM과 호환되며, 스트리밍·메모리 추가도 쉽게 확장할 수 있습니다.

---

3. 구조화된 출력을 사용하는 챗봇

3.1 with_structured_output

LLM의 응답을 Pydantic 모델로 바로 파싱받고 싶을 때 with_structured_output()을 사용합니다.

from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

class MovieResponse(BaseModel):
    title: str = Field(description="영화 제목")
    director: str = Field(description="감독 이름")
    genre: str = Field(description="장르")
    release_year: str = Field(description="개봉 연도")

model = ChatOpenAI(model="gpt-4o")
model_with_structured_output = model.with_structured_output(MovieResponse)

model_with_structured_output.invoke("메르시 영화에 대해 알려주세요")
# → MovieResponse(title='...', director='...', genre='...', release_year='...')

3.2 Union으로 응답 타입 분기

질문 종류에 따라 영화 정보 또는 일반 대화 응답 중 하나를 선택하게 할 수도 있습니다.

from typing import Union

class ConversationalResponse(BaseModel):
    response: str = Field(description="친절하게 대화하듯 답변하는 텍스트")

class FinalResponse(BaseModel):
    final_output: Union[MovieResponse, ConversationalResponse]

structured_llm = model.with_structured_output(FinalResponse)

structured_llm.invoke('메르시 영화에 대해 알려주세요')
# → FinalResponse(final_output=MovieResponse(...))

structured_llm.invoke('메르시 영화에 대해 알려달라고 했잖아요')
# → FinalResponse(final_output=ConversationalResponse(...))

3.3 구조화 출력 + Tool Calling 그래프

구조화 출력과 도구 호출을 결합하면, 도구로 정보를 먼저 가져온 뒤
정해진 형식의 객체로 최종 응답을 반환하는 그래프를 만들 수 있습니다.

커스텀 도구 정의

from langchain_core.tools import tool
from typing import Literal

@tool
def get_movieinfo(movie: Literal['메르시', '인터스텔라']):
    '''아래 두 영화에 대한 정보입니다. 꼭 참고하세요'''
    if movie == '메르시':
        return "메르시는 ..."
    elif movie == '인터스텔라':
        return "인터스텔라는 ..."

State에 final_response 필드 추가

from langgraph.graph import MessagesState

class State(MessagesState):
    final_response: MovieResponse  # 최종 구조화 응답 저장용

MessagesState는 messages: Annotated[list, add_messages] 필드가 이미 포함된
LangGraph 기본 상태 클래스입니다. 이를 상속해 필드를 추가할 수 있습니다.

노드 정의

tools = [get_movieinfo]
model_with_tool = model.bind_tools(tools)

def call_model(state: State):
    response = model_with_tool.invoke(state['messages'])
    return {"messages": [response]}

model_with_structured_output = ChatOpenAI(model="gpt-4o").with_structured_output(MovieResponse)

def respond(state: State):
    # 사용자 원본 질문을 structured output 모델에 전달
    response = model_with_structured_output.invoke(
        [HumanMessage(content=state['messages'][-2].content)]
    )
    return {'final_response': response}

라우팅 – tool_calls 여부로 분기

def should_continue(state: State):
    last_message = state['messages'][-1]
    if not last_message.tool_calls:
        return "respond"   # 도구 호출 없음 → 구조화 응답 생성
    else:
        return "continue"  # 도구 호출 있음 → tools 노드로

그래프 조립

from langgraph.prebuilt import ToolNode

graph_builder = StateGraph(State)
graph_builder.add_node('agent', call_model)
graph_builder.add_node('respond', respond)
graph_builder.add_node('tools', ToolNode(tools))

graph_builder.set_entry_point('agent')
graph_builder.add_conditional_edges(
    'agent', should_continue,
    {"continue": "tools", "respond": "respond"}
)
graph_builder.add_edge("tools", "agent")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()

START → agent ──(tool_calls 있음)──→ tools → agent
              └─(tool_calls 없음)──→ respond → END

실행

answer = graph.invoke(input={"messages": [("human", "메르시 영화에 대해 알려주세요")]})
answer['final_response']
# → MovieResponse(title='메르시', director='...', genre='...', release_year='...')

---

4. arxiv 도구

arxiv는 물리학·수학·컴퓨터 과학 분야의 논문을 무료로 공개·공유하는 플랫폼입니다.

• 저자가 직접 업로드해 심사 전에도 빠르게 공유 가능
• 공식 peer review를 거치지 않을 수 있음
• 주요 분야: 머신러닝·딥러닝, 컴퓨터 비전, 자연어 처리, 물리학, 수학 등
• PDF 다운로드 무료 제공

LangChain에서는 ArxivAPIWrapper로 간편하게 논문을 검색할 수 있습니다.

!pip install arxiv

from langchain_community.utilities import ArxivAPIWrapper

arxiv = ArxivAPIWrapper()

논문 ID로 검색

# "Attention Is All You Need" 논문 (Transformer 원본)
docs = arxiv.run('1706.03762')

# LangChain 관련 논문
docs = arxiv.run('2201.12086')

키워드로 검색

docs = arxiv.run('transformer attention')
print(docs)

ArxivAPIWrapper는 논문의 제목·저자·초록(Abstract) 을 텍스트로 반환합니다.
RAG 파이프라인에서 최신 논문 컨텍스트를 검색하는 도구로 활용할 수 있습니다.

[ 오늘의 정리 ] – LangGraph Tool Calling 챗봇 완성 포인트

• ToolNode는 LLM이 반환한 tool_calls를 읽어 해당 도구를 실행하고, 결과를 ToolMessage로 변환해 상태에 추가하는 노드입니다.
• route_tools(또는 내장 tools_condition) 로 tool_calls 유무에 따라 tools 노드 또는 END로 분기합니다.
• create_react_agent 를 사용하면 위 그래프 구조를 한 줄로 자동 생성할 수 있습니다.
• with_structured_output 은 LLM 응답을 Pydantic 모델로 바로 파싱받는 기능으로, Union 타입을 활용해 응답 유형을 분기할 수 있습니다.
• MessagesState 는 messages 필드가 내장된 기본 상태 클래스로, 상속해 커스텀 필드를 추가하면 구조화 출력 등과 쉽게 결합할 수 있습니다.
• arxiv는 ArxivAPIWrapper로 논문 ID 또는 키워드 검색이 가능하며, RAG 파이프라인의 검색 도구로 활용할 수 있습니다.