ch1에서는 LangGraph를 사용한 기본 챗봇을, ch2에서는 특정 상황에서 도구(검색 엔진)를 자동 호출하는 흐름을 살펴봤습니다. 하지만, ch2에서 우리는 LangGraph가 미리 만들어놓은 ToolNode & tools_condition 을 그대로 썼죠. 이번 ch3에서는 이러한 편의 기능을 직접 구현해봄으로써, LangGraph의 노드 정의와 조건부 라우팅(conditional_edges)을 보다 깊이 이해해보겠습니다.
1. 준비 사항
- 가상환경 & 패키지 설치
python -m venv .venv
source .venv/bin/activate
pip install -U langgraph langsmith langchain_anthropic tavily-python langchain_community- API Key 설정
- Anthropic:
export ANTHROPIC_API_KEY="..." - Tavily:
export TAVILY_API_KEY="..."
- Anthropic:
- 기본 개념
- LangGraph에서 노드(Node) 는 파이썬 함수(또는 객체의
__call__)로 구현되며, - “START -> (여러 노드) -> END” 형태의 흐름을 edges로 연결한다.
- conditional_edges를 통해 조건부 분기를 제어할 수 있다.
- LangGraph에서 노드(Node) 는 파이썬 함수(또는 객체의
3. 전체 코드
아래 예시는 Anthropic Claude와 TavilySearchResults를 연동한 LangGraph 챗봇으로, LLM이 “검색이 필요하다”고 판단할 때, 직접 만든 BasicToolNode가 실행되어 웹 검색을 수행합니다.
import os
import json
# Anthropic Claude
from langchain_anthropic import ChatAnthropic
# LangGraph
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
# Tavily 검색 툴 (웹 검색)
from langchain_community.tools.tavily_search import TavilySearchResults
# typing
from typing import Annotated
from typing_extensions import TypedDict
# langchain-core에서 ToolMessage (툴 호출 결과를 담아 LLM에 전달하기 위함)
from langchain_core.messages import ToolMessage
###############################################################################
# 1) 상태(State) 정의
###############################################################################
class State(TypedDict):
# messages: 사용자/AI 메시지를 누적 저장
messages: Annotated[list, add_messages]
###############################################################################
# 2) 그래프 빌더 생성
###############################################################################
graph_builder = StateGraph(State)
###############################################################################
# 3) LLM 및 툴 초기화
###############################################################################
# - TavilySearchResults: 검색어를 넣으면 JSON 형식으로 결과 반환
search_tool = TavilySearchResults(
max_results=2,
tavily_api_key=os.environ.get("TAVILY_API_KEY") # 환경변수에서 가져오기
)
# - Anthropic Claude
llm = ChatAnthropic(
model="claude-3-5-sonnet-20240620",
anthropic_api_key=os.environ.get("ANTHROPIC_API_KEY")
)
# - LLM에 툴 목록을 "bind"하여,
# LLM이 tool_calls(JSON) 형태로 특정 툴을 호출할 수 있음
llm_with_tools = llm.bind_tools([search_tool])
###############################################################################
# 4) 노드: chatbot
###############################################################################
def chatbot_node(state: State):
"""
1) state["messages"]를 LLM에게 전달
2) LLM이 필요 시 tool_calls를 생성 (search_tool 등)
3) 새 메시지를 반환 -> LangGraph가 append
"""
ai_resp = llm_with_tools.invoke(state["messages"])
return {"messages": [ai_resp]}
###############################################################################
# 5) 노드: BasicToolNode (직접 구현)
###############################################################################
# - LangGraph가 기본 제공하는 ToolNode가 아니라,
# LLM이 호출하겠다고 선언한 tool_calls를 직접 처리하는 로직
class BasicToolNode:
"""
LLM의 마지막 메시지에 tool_calls가 있으면,
그에 따라 실제 툴을 실행하고 ToolMessage를 생성한다.
"""
def __init__(self, tools: list):
# tools_by_name = {툴_이름: 툴_인스턴스}
self.tools_by_name = {t.name: t for t in tools}
def __call__(self, inputs: dict):
# inputs에는 "messages" 리스트가 들어있음
if messages := inputs.get("messages", []):
last_msg = messages[-1] # 가장 최근 AI 메시지
else:
raise ValueError("No messages found in input.")
outputs = []
# AI 메시지에 tool_calls가 있는지 확인
if hasattr(last_msg, "tool_calls"):
for call_info in last_msg.tool_calls:
tool_name = call_info["name"]
args = call_info["args"]
# 툴 이름에 맞춰 실제 invoke
if tool_name in self.tools_by_name:
result = self.tools_by_name[tool_name].invoke(args)
# 검색 결과(등등)를 ToolMessage 형태로 담아 반환
# -> chatbot 노드에서 이어받아 최종 답변에 활용
outputs.append(
ToolMessage(
content=json.dumps(result),
name=tool_name,
tool_call_id=call_info["id"],
)
)
return {"messages": outputs}
###############################################################################
# 6) 노드 등록
###############################################################################
graph_builder.add_node("chatbot", chatbot_node)
tool_node = BasicToolNode([search_tool])
graph_builder.add_node("tools", tool_node)
###############################################################################
# 7) 조건부 라우팅
###############################################################################
# - chatbot 노드가 실행된 후, LLM이 도구를 호출했는지 여부를 판단해
# "tools" 노드로 갈지, "END"로 갈지 결정
def route_tools(state: State):
"""
chatbot 노드가 끝난 직후 실행되는 함수
- state["messages"][-1] 에 tool_calls가 있으면 'tools'
- 없으면 'END'
"""
if messages := state.get("messages", []):
last_msg = messages[-1]
# tool_calls가 있고, 갯수가 > 0이면 'tools'
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools"
# 그렇지 않으면 'END'
return END
# - add_conditional_edges("chatbot", route_tools, {"tools":"tools", END:END})
graph_builder.add_conditional_edges(
"chatbot",
route_tools,
{
"tools": "tools",
END: END
}
)
# - tools -> chatbot: 툴 실행 후 다시 chatbot 노드로 돌아가서 최종 답변을 완성
graph_builder.add_edge("tools", "chatbot")
# - START -> chatbot
graph_builder.add_edge(START, "chatbot")
###############################################################################
# 8) 그래프 컴파일
###############################################################################
graph = graph_builder.compile()
################################################################################
# 그래프 시각화를 위한 ASCII 아트 생성
################################################################################
# - graph.get_graph().draw_ascii()를 호출하여 그래프 구조를 ASCII 아트로 표현
# - 노드와 엣지의 연결 관계를 시각적으로 보여줌
ascii_graph = graph.get_graph().draw_ascii()
print(ascii_graph)
###############################################################################
# 9) 메인 함수
###############################################################################
def main():
print("==== [ch3] Custom BasicToolNode Demo ====")
print("Type 'quit' or 'q' to exit.\n")
while True:
user_input = input("User: ")
if user_input.lower() in ["quit", "q", "exit"]:
print("Goodbye!")
break
# 그래프를 단계별로 실행
# "messages": [("user", user_input)] => 대화 목록에 새 사용자 메시지 추가
events = graph.stream({"messages": [("user", user_input)]})
# 실행 중 발생하는 모든 노드의 업데이트를 받아 메시지가 생성되면 출력
for event in events:
for node_output in event.values():
if "messages" in node_output:
last_msg = node_output["messages"][-1]
# AIMessage 혹은 ToolMessage
print("Assistant:", last_msg.content)
if __name__ == "__main__":
main()4. 코드 해설
chatbot_node
state["messages"]를 LLM(Claude)에 전달해 새 응답을 생성.- Claude가 “검색이 필요하다”고 판단하면 응답에
tool_calls를 담아 돌려준다.
BasicToolNode (직접 구현)
- LangGraph의 빌트인
ToolNode와 유사한 역할을 하지만, 커스텀 로직이 필요할 경우 이처럼 직접 짤 수 있다. - 마지막 AI 메시지(
messages[-1])에tool_calls가 있으면, 해당 툴을invoke(args)로 실행. - 그 결과를 ToolMessage로 만들어 대화에 추가.
- 한 번 검색을 수행하면, 다음 턴에 다시
chatbot_node로 돌아와 최종 답변이 완성되는 구조.
조건부 라우팅 (route_tools)
- chatbot 노드가 끝난 뒤 실행되는 함수로,
- 메시지에
tool_calls가 있으면"tools", 없으면"END"로 분기. - 이후
"tools" -> "chatbot"으로 돌아와 LLM이 새로운 정보를 활용할 수 있게 된다.
메인 함수
- 사용자 입력 ->
graph.stream(...)-> 순차적으로 노드가 실행되며 메시지를 추가 -> 결과 출력 - 사용자가 “quit”를 입력하기 전까지 반복.
5. 실행 예시
$ python ch3_chatbot_with_tools_using_customnode.py
+-----------+
| __start__ |
+-----------+
*
*
*
+---------+
| chatbot |
+---------+
. .
.. ..
. .
+-------+ +---------+
| tools | | __end__ |
+-------+ +---------+
==== [ch3] Custom BasicToolNode Demo ====
Type 'quit' or 'q' to exit.
User: 안녕 Claude
Assistant: 안녕하세요! 어떻게 도와드릴까요?
User: LangGraph가 뭐야?
Assistant: [{'text': "Let me look that up for you...", 'type': 'text'},
{'id': 'toolu_01A...', 'input': {'query': 'LangGraph'}, ...}]
Assistant: [{"url": "https://github.com/langchain-ai/langgraph", "content": "LangGraph is a library for ..."}]
Assistant: LangGraph는 ....- Claude가 “검색해야겠다”고 판단하면,
tool_callsJSON이 AI 메시지에 들어 있음 →BasicToolNode가 search_tool을 실제로 호출 → 그 결과를 ToolMessage로 만들어 대화에 추가 → 다시 chatbot_node에서 최종 요약/답변을 준다.
6. add_conditional_edges 좀 더 살펴보기
노드(Node)가 완료된 뒤, 현재 상태(State)를 분석하여 다음에 어디로 갈지 분기(조건부 라우팅) 하고 싶을 때 사용합니다. 즉, “조건에 따라 서로 다른 노드로 이동”하는 흐름을 간단히 정의할 수 있습니다.
- 일반적인 add_edge("A", "B")는 “A 후에 무조건 B”로 연결되지만,
- add_conditional_edges("A", condition_fn, {...})는 “A 후에 condition_fn의 결과값에 따라 여러 갈래”로 분기할 수 있습니다.
함수 문법
StateGraph.add_conditional_edges(
from_node: str,
condition: Callable[[Any], str | list[str]],
mapping: dict[str, str] = {},
...
) -> Nonefrom_node: str
- 어떤 노드가 끝난 뒤 이 조건부 로직을 적용할지 지정.
- 예:
"chatbot"→ chatbot 노드가 실행을 마치면condition함수를 호출하여 분기 결정.
condition: Callable[[Any], str | list[str]]
- 실제 분기 로직이 들어 있는 함수(“조건 함수”).
- 그래프의 현재 State(예:
TypedDict)를 입력으로 받아, 다음으로 이동할 노드 이름(문자열)을 반환. - 하나의 문자열을 반환하면 “단일 분기”, 문자열 리스트를 반환하면 “여러 노드로 병렬 분기”도 가능.
- 예시:
def route_tools(state: State) -> str:
last_msg = state["messages"][-1]
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
return "tools"
return "END"mapping: dict[str, str]
condition함수가 반환한 문자열(또는 문자열 중 하나)을 실제 노드 이름 혹은 특수 키(END)에 매핑.- 예:
{ "tools": "tools", "end": END }condition이"tools"를 반환하면 실제로tools노드로 이동."end"를 반환하면 그래프 종료(END).
- 만약
condition이"my_tools"같은 임의 문자열을 반환하고,mapping에{"my_tools": "tools"}라고 작성하면, 실제로는 "tools" 노드로 이동시키는 식으로 커스터마이징 가능. - 보통은
"tools"→"tools","END"→END처럼 간단하게 1:1 매핑하는 경우가 많습니다.
리턴(반환) 값
add_conditional_edges자체는 그래프에 조건부 엣지를 추가하고None을 반환합니다.- 이후 그래프가
from_node를 마친 시점에 내부적으로condition(state)를 호출해, 반환값을mapping으로 해석하여 다음 노드를 결정하게 됩니다.
7. 마무리 및 확장
- 직접 구현한 ToolNode를 통해 LangGraph에서 다양한 모듈(예: 검색, DB 조회, 사내 API 호출 등)을 자유롭게 연결할 수 있습니다.
- 함수형으로 만들 수도 있고, 이번처럼 클래스(
__call__)로 만들어 노드 로직을 더 체계적으로 관리할 수도 있습니다. - 조건부 라우팅 역시 직접
route_tools함수를 짜보면, LangGraph가 제공하는 편의 함수(tools_condition)의 내부 동작을 파악하는 데 도움이 됩니다.
이상으로, ch3: LangGraph에서 직접 만든 ToolNode로 검색 툴을 연동해보는 방법을 소개해드렸습니다!