langgraph의 공식문서를 번역해 놓은 자료입니다. 이해하기 쉽게 예제는 변경하였습니다. 또한 필요한 경우 부연 설명을 추가하였습니다. 문제가 되면 삭제하겠습니다.
Part 2: 도구(Tool)로 챗봇 강화하기
챗봇이 "memory"로 답변할 수 없는 질문을 처리하기 위해, 웹 검색 도구를 사용할 것이다. 이 도구를 사용하여 챗봇이 관련 정보를 찾고 더 나은 응답을 제공할 수 있도록 할 수 있다.
다음 단계는 웹 검색 도구를 챗봇에 추가하는 것이다. 이렇게 하면 사용자가 묻는 질문에 대해 더 풍부하고 정확한 정보를 제공할 수 있게 된다.
준비작업
시작하기 전에 필요한 패키지가 설치되어 있고 API 키가 설정되어 있는지 확인하자.
먼저, Tavily Search Engine을 사용하기 위해 필요한 패키지를 설치하고, TAVILY_API_KEY를 설정하자.
pip install tavily-python.env 파일에 TAVILY_API_KEY를 추가하자.
TAVILY_API_KEY=tvly-xxxx
tavily는 LLM을 활용한 검색엔진으로 실시간 정보를 제공한다. 월 1,000회 API 호출은 무료로 사용할 수 있다.https://app.tavily.com/home에 접속해서 API Keys 메뉴에서 새로운 API를 발급하면 된다.
다음으로 툴을 정의하자.
from dotenv import load_dotenv
from langchain_community.tools.tavily_search import TavilySearchResults
load_dotenv()
tool = TavilySearchResults(max_results=2)
tools = [tool]
result = tool.invoke("What's a 'node' in LangGraph?")
print(result)실행결과
[
{
'url': 'https://medium.com/@cplog/introduction-to-langgraph-a-beginners-guide-14f9be027141',
'content': 'Nodes: Nodes are the building blocks of your LangGraph ...'
},
{
'url': 'https://www.datacamp.com/tutorial/langgraph-tutorial',
'content': "In LangGraph, each node represents an LLM agent, ..."
}
]실행결과에 나오는 데이터는 챗봇이 질문에 답변하는 데 사용할 수 있는 페이지 요약이다.
다음으로 그래프를 정의하자. 아래 내용은 이전과 동일하지만, LLM에 bind_tools를 추가했다. 이를 통해 LLM은 검색 엔진을 사용하고 싶을 때 올바른 JSON 형식을 사용하도록 알 수 있다.
from typing import Annotated
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)
llm = ChatOpenAI(model="gpt-3.5-turbo")
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
graph_builder.add_node("chatbot", chatbot)다음으로, 도구가 호출되었을 때 실제로 실행할 수 있는 함수를 생성해야 한다. 이를 위해 새로운 노드에 도구를 추가한다.
아래에서는 BasicToolNode를 구현하여 상태의 가장 최근 메시지를 확인하고, 메시지에 tool_calls가 포함되어 있는 경우 도구를 호출한다. 이 과정은 Anthropic, OpenAI, Google Gemini 및 기타 여러 LLM 제공업체에서 사용할 수 있는 LLM의 tool_calling 지원에 의존한다.
이 후에는 LangGraph의 미리 만들어진 ToolNode로 이 기능을 대체할 예정이지만, 처음에는 직접 구축해 보는 것이 좋다.
import json
from langchain_core.messages import ToolMessage
class BasicToolNode:
"""마지막 AI 메시지에서 요청된 도구를 실행하는 노드"""
def __init__(self, tools: list) -> None:
self.tools_by_name = {tool.name: tool for tool in tools}
def __call__(self, inputs: dict):
if messages := inputs.get("messages", []):
message = messages[-1]
else:
raise ValueError("No message found in input")
outputs = []
for tool_call in message.tool_calls:
# tool_call: {'name': 'tavily_search_results_json', 'args': {'query': '오늘 서울 날씨가 어때?'}, 'id': 'call_mDnvuCd8YHmp7XaVK7qYTAAW', 'type': 'tool_call'}
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"],
)
)
# outputs: [ToolMessage(content='[{"url": "https://news.sbs.co.kr/news/endPage.do?news_id=N1007850214", "content": "..."}, {"url": "https://www.yna.co.kr/view/AKR20241024078800009", "content": "..."}]', name='tavily_search_results_json', tool_call_id='call_mDnvuCd8YHmp7XaVK7qYTAAW')]
return {"messages": outputs}
tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)도구 노드를 추가했고, 이제 conditional_edges를 정의할 수 있다.
엣지는 제어 흐름을 한 노드에서 다음 노드로 라우팅하는 역할을 한다. conditional edge는 일반적으로 현재 그래프 상태에 따라 다른 노드로 라우팅하기 위해 "if" 문을 포함한다. 이러한 함수는 현재 그래프 상태를 수신하고, 다음에 호출할 노드의 문자열 또는 문자열 목록을 반환한다.
아래에서는 route_tools라는 라우터 함수를 정의하여 챗봇의 출력에서 tool_calls를 확인한다. 이 함수를 그래프에 제공하기 위해 add_conditional_edges를 호출하여 챗봇 노드가 완료될 때마다 이 함수를 확인하여 다음에 어디로 갈지를 판단하도록 한다.
조건은 도구 호출이 있는 경우 도구로 라우팅하고, 그렇지 않은 경우 END로 라우팅한다.
여기서는 chatbot에서 llm_with_tools을 호출하기 때문에 무조건 tool을 호출한다.
이후에는 더 간결하게 만들기 위해 미리 만들어진 tools_condition으로 이 기능을 대체할 예정이지만, 처음에는 직접 구현하여 이해를 돕는 것이 좋다.
tools_condition은 langchain.prebuilt에 있는 함수이며 아래 route_tools와 동일한 역할을 하는 함수이다. 여기서는 tools_condition의 내부동작이 어떻게 구성되어 있는지 보여주려고 직접 구현을 해봤고, 이후에는 tools_condition을 직접 사용하는 것으로 되어 있다.
from typing import Literal
def route_tools(
state: State,
):
"""
마지막 메시지에 도구 호출이 있으면 ToolNode로 라우팅하기 위해 conditional_edge에서 사용한다.
그렇지 않으면 종료로 라우팅한다.
"""
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" # 도구를 사용해야 할 때 "tools"를 반환한다.
return END
# tools_condition 함수는 챗봇이 도구를 사용해야 하는지 여부를 판단하여,
# 도구를 사용해야 할 때는 "tools"를 반환하고, 직접 응답해도 될 때는 "END"를 반환한다.
# 이 조건부 라우팅은 메인 에이전트 루프를 정의한다.
graph_builder.add_conditional_edges(
"chatbot",
route_tools,
# 다음 dict는 그래프에 조건의 출력을 특정 노드로 해석하도록 지시할 수 있다.
# 기본적으로는 identity 함수로 설정되지만, "tools" 이외의 다른 이름을 가진 노드를 사용하려면 dict 값을 변경할 수 있다.
# 예: "tools": "my_tools" {"tools": "tools", END: END},
)
# 도구가 호출될 때마다 우리는 다음 단계를 결정하기 위해 다시 챗봇으로 돌아간다.
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()conditional edge는 단일 노드에서 시작한다는 점에 주목하자. 이는 그래프에 "챗봇 노드가 실행될 때마다 도구를 호출하면 'tools'로 가고, 그렇지 않으면 루프를 종료하라"고 지시하는 것이다.
미리 만들어진 tools_condition과 마찬가지로, 함수에서 도구 호출이 없을 경우 END 문자열을 반환한다. 그래프가 END로 전환되면 더 이상 수행할 작업이 없으며 실행을 중단한다. 조건이 END를 반환할 수 있기 때문에 이번에는 종료 지점을 명시적으로 설정할 필요가 없다. 그래프는 이미 종료할 수 있는 방법을 가지고 있다.
이제 우리가 구축한 그래프를 시각화해보자. 아래 함수는 이 튜토리얼에 중요하지 않은 추가 의존성을 필요로 한다.
from IPython.display import Image, display
try:
display(
Image(
graph.get_graph().draw_mermaid_png(
output_file_path="./chatbot_with_tool.png"
)
)
)
except Exception:
# This requires some extra dependencies and is optional
pass
이제 우리는 봇에게 훈련 데이터 외부의 질문을 할 수 있다.
def stream_graph_updates(user_input: str):
messages = []
for event in graph.stream({"messages": [("user", user_input)]}):
for value in event.values():
messages.append(value["messages"][-1].content)
return "\n".join(messages)
question = "오늘 서울 날씨가 어때?"
result = stream_graph_updates(question)
print(f"result: {result}")실행결과
[{"url": "https://www.accuweather.com/ko/kr/seoul/1-226081_30_al/hourly-weather-forecast/1-226081_30_al", "content": "..."}, {"url": "https://www.weather.go.kr/w/index.do", "content": "..."}]
서울의 오늘 날씨에 대한 정보입니다:
- AccuWeather에 따르면, 오늘 오후 3시부터 11시까지 비가 오고, 기온은 65°F에서 82°F 사이로 예상됩니다.
- 기상청에 따르면, 서울은 내일 새벽 2시부터 오후 12시까지 맑은 날씨가 기대되며, 밤에는 안개가 발생할 수 있습니다. 현재 기온은 14°C이며, 습도는 24%입니다.
