2-1. langgraph_chat_with_tools

LangGraph 챗봇에서 도구 사용하기

챗봇이 자체 '기억'으로 답변할 수 없는 질문을 처리하기 위해 웹 검색 도구를 통합하여 기능을 향상시켜 보겠습니다. 이를 통해 챗봇은 더 관련성 높은 정보를 찾아 더 나은 응답을 제공할 수 있습니다.​

학습 목표

• 챗봇에 도구를 통합하여 기능 확장하기
• Tavily 검색 도구를 사용하여 질문에 답변하기 : 달에 2000건 정도 무료 사용 가능
• LangGraph 노드에서 도구 사용법 학습하기

1. 환경 설정

라이브러리 설치

다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.

pip install -U python-dotenv notebook langgraph langchain_openai tavily-python langchain_community

환경변수 설정

환경변수 파일 .env를 생성하여 다음의 내용을 설정합니다.

OPENAI_API_KEY=본인의_OpenAI_API키
OPENAI_MODEL=gpt-4o-mini
TAVILY_API_KEY=본인의_tavily_api_key

환경변수를 로드하기 위해 Python의 python-dotenv 라이브러리를 사용합니다.

2. 코드 설명

라이브러리 임포트

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict
from typing import Annotated, List
from dotenv import load_dotenv
import os

• ChatOpenAI: OpenAI의 Chat Completion API를 호출하기 위한 클래스입니다.
• StateGraph, START, END: LangGraph의 상태 기반 그래프 구조를 구성하는 데 사용됩니다.
• add_messages: LangGraph의 메시지 관리 기능을 자동화하는 역할을 수행합니다.
• TypedDict, Annotated, List: 상태 정의를 위한 타입 힌트 도구입니다.
• load_dotenv: 환경변수를 로드합니다.

환경변수 로딩

load_dotenv()

openai_model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

도구 정의

Tavily 검색 도구를 설정합니다.

from langchain_community.tools.tavily_search import TavilySearchResults

tool = TavilySearchResults(max_results=2)
tool.invoke("LangGraph에서 '노드'란 무엇인가요?")

• TavilySearchResults: 사용자가 입력한 질의에 대해 최대 2개의 결과를 가져오는 검색 도구를 만들었습니다.

상태(State) 정의

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

• 이 코드는 LangGraph에서 챗봇의 상태(State)를 정의하는 부분입니다.
• 상태는 LangGraph의 핵심 개념입니다. 챗봇이 메시지를 주고받으며 관리할 데이터를 명확히 정의합니다.
• LangGraph는 상태 기반 그래프(StateGraph)를 사용하는 프레임워크로, 각 단계의 상태를 명확히 정의하고, 상태 간에 데이터를 전달하는 방식으로 작동합니다.
• messages는 메시지를 관리하는 리스트로, 자동 메시지 추가 및 관리가 가능합니다.

구성 요소 설명

• class State(TypedDict)
  • Python의 TypedDict를 사용해 명시적인 타입이 있는 딕셔너리를 정의합니다.
  • 이를 통해 상태(State)의 구조와 타입을 명확히 지정하여 코드의 가독성과 안정성을 높입니다.
• messages
  • 상태에 저장되는 주요 데이터입니다. 여기서는 대화 중 주고받은 메시지를 담고 있는 리스트입니다.
  • LangGraph는 대화의 문맥을 유지하기 위해 메시지의 리스트를 관리합니다.
• Annotated[List, add_messages]
  • Annotated 타입 힌트를 사용하여 메시지 리스트가 특정한 규칙(add_messages)을 따르도록 합니다.
  • 여기서 add_messages는 LangGraph에서 제공하는 특별한 주석(annotation)으로, 메시지 리스트를 다룰 때 자동으로 관리(추가)를 돕는 기능을 수행합니다.

이 상태 정의를 통해 챗봇은:

• 대화의 각 단계를 명확히 관리할 수 있습니다.
• 사용자의 메시지와 모델의 응답 메시지를 체계적으로 기록하고 관리합니다.
• 명시적인 타입과 자동 관리 기능을 활용하여 코드의 오류를 최소화하고 유지보수를 쉽게 합니다.

이러한 방식을 통해, LangGraph는 상태 기반의 복잡한 챗봇 애플리케이션에서도 효율적이고 명확한 데이터 관리와 흐름 제어를 가능하게 합니다.

LLM 모델 설정 및 도구 바인딩

llm = ChatOpenAI(model=openai_model)
tools = [tool]
llm_with_tools = llm.bind_tools(tools)

• llm: 사용되는 LLM 모델로, 여기서는 OpenAI의 gpt-4o-mini 모델을 사용합니다.
• llm_with_tools: LLM에 정의한 도구를 바인딩하여 도구 사용이 가능하게 합니다.

챗봇 노드 정의

챗봇이 도구를 이용하여 사용자 메시지에 응답하도록 설정합니다.

def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

• chatbot(state: State): 챗봇 노드 함수로, 상태에서 받은 메시지를 기반으로 도구를 활용하여 응답을 생성하고 상태를 업데이트합니다.

도구 실행 노드 정의

이 클래스는 LLM이 요청한 **도구(tool)**를 실제로 실행하는 역할을 합니다.

from langchain_core.messages import ToolMessage
import json

class ToolNode:
    def __init__(self, tools):
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No messages found in inputs.")
        
        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call['name']].invoke(tool_call['args'])
            tool_message = ToolMessage(
                content=json.dumps(tool_result),
                name=tool_call['name'],
                tool_call_id=tool_call['id'],
            )
            outputs.append(tool_message)

        return {"messages": outputs}
    
tool_node = ToolNode(tools)

① 초기화 메서드 (__init__)

def __init__(self, tools):
    self.tools_by_name = {tool.name: tool for tool in tools}

• tools는 미리 정의된 도구 객체의 리스트입니다.
• 이 도구들을 도구의 이름을 키(key)로 갖는 딕셔너리 형태로 변환하여 저장합니다.
• 이렇게 하는 이유는 이후 특정 도구를 빠르게 찾기 위함입니다.
  • 예: {"search": TavilySearchResults(max_results=2), ...}

② 호출 메서드 (__call__) 이 메서드는 실제로 그래프 내에서 노드가 실행될 때 호출되는 메서드입니다.

입력 메시지 확인

if messages := inputs.get("messages", []):
    message = messages[-1]
else:
    raise ValueError("No messages found in inputs.")

• 입력으로 들어온 상태(inputs)에서 messages를 가져옵니다.
• 가져온 메시지 중 가장 최근 메시지(messages[-1])를 선택합니다.
• 메시지가 없는 경우 명시적으로 오류를 발생시켜 문제를 알립니다.

도구 호출 요청 확인 및 처리

outputs = []
for tool_call in message.tool_calls:

• 가장 최근 메시지(message)에서 도구 호출 요청(tool_calls)을 하나씩 순회합니다.
• 여기서 tool_calls는 LLM이 생성한 메시지 안에 있는 도구 호출 요청 목록입니다.
  • 각 요청에는 도구 이름(name), 호출 인자(args), 그리고 고유 식별자(id)가 있습니다.

실제 도구 실행

tool_result = self.tools_by_name[tool_call['name']].invoke(tool_call['args'])

• 각 도구 호출에서 요청된 도구(tool_call['name'])를 찾아 실행합니다.
• 실행 인자(tool_call['args'])를 도구에 전달하여 결과를 얻습니다.
• 예: Tavily 검색 도구에 사용자가 요청한 검색어를 전달하고 검색 결과를 얻습니다.

도구 실행 결과 메시지 생성

tool_message = ToolMessage(
    content=json.dumps(tool_result),
    name=tool_call['name'],
    tool_call_id=tool_call['id'],
)
outputs.append(tool_message)

• 도구가 실행된 결과(tool_result)를 JSON 형식으로 변환하여 메시지로 만듭니다.
• 이 메시지는 ToolMessage 형식으로, LLM이 결과를 다시 이해하고 이후 단계에서 사용할 수 있도록 구조화합니다.
• 결과 메시지는 다음의 필드를 포함합니다:
  • content: 도구의 실행 결과 (JSON 형태)
  • name: 도구의 이름
  • tool_call_id: 원본 도구 호출 요청의 ID (LLM이 요청과 결과를 매칭하는데 사용됨)

③ 반환 값

return {"messages": outputs}

• 도구 실행 결과로 만들어진 메시지 목록(outputs)을 다시 상태(State)로 반환합니다.
• 이후 챗봇 노드가 이 결과를 받아 LLM 모델이 추가 응답을 생성하는 데 활용합니다.

④ 도구 노드 인스턴스화

tool_node = ToolNode(tools)

• 위에서 정의한 클래스를 실제로 사용할 수 있도록 인스턴스화합니다.
• 이렇게 인스턴스화된 노드를 그래프 내에서 직접 사용할 수 있게 됩니다.

조건부 엣지를 위한 조건 함수 정의

이 코드는 **LangGraph에서 조건부 흐름을 결정하는 조건 함수(tools_condition)**입니다. 이 함수의 역할은 챗봇이 생성한 메시지를 검사하여, 챗봇이 도구 사용을 요청했는지 확인하고, 다음으로 어떤 노드로 갈지 결정하는 것입니다.

def tools_condition(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 to tool_edge: {state}")
    
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return 'tools'
    return END

① 입력값 (state) 검사하기

if isinstance(state, list):
    ai_message = state[-1]

• 먼저, 함수에 입력된 state가 리스트인지 확인합니다.
• 리스트라면, 가장 최근 메시지(state[-1])를 선택합니다.
• 이는 메시지 상태가 리스트 형태로 전달될 수 있는 경우를 대비한 코드입니다.

② 입력 상태가 딕셔너리일 경우 (elif 문)

elif messages := state.get("messages", []):
    ai_message = messages[-1]

• state가 딕셔너리 형태일 경우, "messages" 키를 찾습니다.
• "messages" 키의 값이 존재하고 빈 리스트가 아니면, 가장 최근 메시지를 ai_message로 선택합니다.
• 이때 사용된 := (할당 표현식) 연산자는 메시지 값을 변수에 할당하면서 동시에 조건 평가를 수행합니다.

③ 메시지가 없는 경우 에러 처리

else:
    raise ValueError(f"No messages found in input state to tool_edge: {state}")

• 위 두 조건이 모두 실패한다면, state 안에 메시지가 없다는 것을 의미합니다.
• 이때 명시적으로 오류(ValueError)를 발생시켜 문제를 즉시 알립니다.
• 오류 메시지에는 입력된 상태(state)를 포함하여 디버깅을 쉽게 합니다.

④ 다음 노드를 결정하는 핵심 조건부 판단

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

• 먼저, 최근 메시지(ai_message)가 tool_calls 속성을 가지고 있는지 확인합니다.
  • tool_calls는 LLM이 도구 사용을 요청할 때 메시지에 추가하는 속성입니다.
• 그리고 tool_calls 속성의 길이(len(ai_message.tool_calls))가 0보다 크면, 즉 적어도 하나 이상의 도구 호출 요청이 존재하면 다음으로 진행할 노드를 'tools'로 결정합니다.
  • 이는 챗봇이 도구 사용을 요청했음을 의미합니다.
• 그렇지 않다면(도구 호출 요청이 없다면) 다음 노드로 END를 선택하여 그래프를 종료합니다.

함수가 결정하는 흐름 요약

조건결과 (다음 노드)의미

메시지에 도구 호출(tool_calls) 요청이 있음	'tools' 노드로 이동	도구를 실행할 필요가 있음 (도구 노드 호출)
메시지에 도구 호출 요청이 없음	END로 이동	도구 실행 없이 챗봇 흐름 종료

그래프 구성 및 컴파일

조건 함수 tools_condition을 사용하여 조건부로 도구 노드를 호출합니다.

workflow = StateGraph(State)

workflow.add_node("chatbot", chatbot)
workflow.add_node("tools", tool_node)

workflow.add_conditional_edges("chatbot", tools_condition, {"tools": "tools", END: END})
workflow.add_edge("tools", "chatbot")
workflow.add_edge(START, "chatbot")

graph = workflow.compile()

• 먼저 StateGraph(State)를 통해 상태 기반 그래프를 생성합니다.
• workflow.add_node("chatbot", chatbot)은 그래프에 "chatbot"이라는 이름의 노드를 추가하며, 이 노드는 앞에서 정의한 chatbot 함수를 실행합니다.
• workflow.add_node("tools", tool_node)는 "tools"이라는 이름의 노드를 추가하며, 이 노드는 도구 실행을 관리합니다.
• workflow.add_conditional_edges("chatbot", tools_condition, {"tools": "tools", END: END})은 "chatbot" 노드에서 "tools" 노드로 연결되는 조건부 엣지를 정의합니다. tools_condition이 True일 때만 "tools" 노드로 연결되고, 그렇지 않으면 END 노드로 연결됩니다.
• workflow.add_edge("tool", "chatbot")은 "tool" 노드에서 다시 "chatbot" 노드로 연결하는 엣지를 정의합니다. 이 엣지는 도구 실행 후 결과를 챗봇 노드로 반환하는 역할을 합니다.
• workflow.set_entry_point("chatbot")은 그래프의 시작점을 "chatbot" 노드로 설정합니다. 즉, 챗봇이 처음 실행되는 지점입니다.
• workflow.compile()은 정의한 그래프를 실제로 실행 가능한 형태로 컴파일하여 최적화 상태를 준비합니다.

그래프 시각화

컴파일된 그래프를 이용해 시각화봅니다.

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

LangGraph에서 시각화된 그래프에서 직선 화살표와 점선 화살표는 명확히 구분되는 의미를 가지고 있습니다.

• 직선 화살표 (Solid Arrow)
  • 직선 화살표는 **일반적인 엣지(edge)**를 나타냅니다.
  • 특별한 조건 없이 한 노드에서 다음 노드로의 단순 이동 경로를 의미합니다.
  • 예를 들어, 이미지에서 __start__ 노드에서 chatbot 노드로 이동하거나, tools 노드에서 다시 chatbot으로 되돌아가는 경로는 직선으로 표시되어 있습니다.
• 점선 화살표 (Dashed Arrow)
  • 점선 화살표는 **조건부 엣지(conditional edge)**를 나타냅니다.
  • 특정 조건이 충족될 경우에만 활성화되어 노드가 이동합니다.
  • 즉, 조건에 따라 실행 여부가 결정되는 흐름을 점선으로 나타냅니다.
  • 이미지에서 chatbot 노드에서 tools 노드로 또는 __end__ 노드로 가는 화살표가 점선으로 표시된 이유는, 이 경로가 tools_condition 조건 함수의 결과에 따라 분기되기 때문입니다.
    • tools_condition이 **참(True)**일 경우 → tools 노드로 이동
    • tools_condition이 **거짓(False)**일 경우 → __end__ 노드로 이동 (그래프 종료)

요약 정리

화살표 타입의미예시

직선 (Solid)	조건 없는 일반적인 흐름 (무조건 진행)	__start__ → chatbot
점선 (Dashed)	조건에 따라 진행 여부 결정됨	chatbot → tools 또는 __end__

이러한 구분을 통해 그래프의 흐름과 조건에 따른 실행 구조를 직관적으로 이해할 수 있습니다.

3. 챗봇 실행

사용자가 입력한 메시지를 기반으로 챗봇이 응답을 생성하는 과정입니다. 도구를 사용하여 더욱 정확한 정보를 제공합니다.

user_input = "LangGraph에서 '노드'란 무엇인가요?"
state = {"messages": [HumanMessage(content=user_input)]}
response = graph.invoke(state)

print(response["messages"][-1].content)