from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator
print("✅ LangGraph 라이브러리 임포트 완료")
# tool 데코레이터는 일반함수를 LLM이 호출 가능한 도구로 변환
@tool
def web_search_tool(query: str) -> str: #쿼리를 스트링으로 쓰고 실행하면 결과가 스트링
'''인터넷에서 실시간 정보를 검색합니다'''
#위 쿼리는 LLM에 설명(정보)를 주는 용도
print(f"🔍 웹 검색: {query}")
try:
url = "https://api.tavily.com/search"
payload = {
"api_key": TAVILY_API_KEY,
"query": query,
"search_depth": "basic",
"max_results": 3
}
response = requests.post(url, json=payload)
response.raise_for_status()
data = response.json()
# 결과를 간단한 텍스트로 변환
results_text = f"검색 결과 for '{query}':\n\n"
for i, result in enumerate(data.get("results", [])[:3], 1):
results_text += f"{i}. {result.get('title')}\n"
results_text += f" {result.get('content')[:150]}...\n\n"
if data.get("answer"):
results_text += f"요약: {data.get('answer')}\n"
return results_text
except Exception as e:
return f"검색 실패: {str(e)}"
@tool
def calculate_tool(expression: str) -> str:
'''수학 계산을 수행합니다. 이 도구는 수식만 넣어주면 계산 결과를 돌려줍니다.
예) '125*45'
또는 '(100+50)/2'처럼 표현식만 써주면 됩니다.
'''
print(f"🧮 계산 실행: '{expression}'")
try:
# 안전한 계산
allowed_names = {'abs': abs ,'round':round,'pow':pow}
result = eval(expression,{'__builtins__' :{}},allowed_names)
return json.dumps({
"expression": expression,
"result": result
}, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": f"계산 실패: {str(e)}"}, ensure_ascii=False)
@tool
def get_current_time_tool() -> str:
'''이 도구를 사용하면 현재 날짜와 시간을 가져옵니다. 매개변수를 넣을 필요는 없습니다.'''
print(f"⏰ 현재 시간 조회")
now = datetime.now()
return json.dumps({
'datetime' : now.strftime('%Y-%m-D %H:%M:%S'),
'date':now.strftime('%Y-%m-%d'),
'time':now.strftime('%H:%M:%S'),
'day_of_week':now.strftime('%A')
}, ensure_ascii=False)
# 도구 리스트
langchain_tools = [web_search_tool, calculate_tool, get_current_time_tool]
print("✅ LangChain 스타일 도구 정의 완료")
for tool in langchain_tools:
print(f" - {tool.name}: {tool.description}")
# ChatOpenAI 모델 생성
llm = ChatOpenAI(
model="gpt-4.1-nano",
temperature=0,
api_key=OPENAI_API_KEY
)
# 모델에 도구 바인딩
llm_with_tools = llm.bind_tools(langchain_tools)
print("✅ LLM 설정 및 도구 바인딩 완료")
print(f"모델: {llm.model_name}")
print(f"바인딩된 도구 수: {len(langchain_tools)}")
위 코드의 실행 결과
✅ LangChain 스타일 도구 정의 완료 - web_search_tool: 인터넷에서 실시간 정보를 검색합니다 - calculate_tool: 수학 계산을 수행합니다. 이 도구는 수식만 넣어주면 계산 결과를 돌려줍니다. 예) '125*45' 또는 '(100+50)/2'처럼 표현식만 써주면 됩니다. - get_current_time_tool: 이 도구를 사용하면 현재 날짜와 시간을 가져옵니다. 매개변수를 넣을 필요는 없습니다.
✅ LLM 설정 및 도구 바인딩 완료 모델: gpt-4.1-nano 바인딩된 도구 수: 3
LangGraph는 LangChain과 달리 분기를 나눠 수행 가능 (일렬로 수행되지 않음)
# 사용자에게 질문을 받고 -> LLM이 생각하고 -> 도구 호출하고 -> 도구 결과 받고
# 도구 실행 결과 + 사용자 질문을 보고 -> 다시 생각하고 -> 최종 답변
# 각 단계마다 정보가 축적 되어야 하니 이 모든 정보를 담는 그릇 이 'STATE'
class AgentState(TypedDict):
"""에이전트의 상태를 정의합니다"""
# messages: 대화 히스토리를 저장하는 리스트
# operator.add를 사용하여 메시지를 누적
# TypedDict : 상태 구조를 명확히 정의
messages : Annotated[Sequence[BaseMessage], operator.add]
# 명확성 , 안정성, 가독성
#Annoatated = 새 메세지가 들어오면 기존 리스트에 추가
print("✅ 그래프 상태 정의 완료")
print("""
AgentState 구조:
- messages: 대화 메시지들의 리스트
- operator.add: 새 메시지를 기존 리스트에 추가
""")
def call_model(state: AgentState):
"""LLM을 호출하는 노드"""
print("🤖 LLM 호출 중...")
messages = state["messages"]
response = llm_with_tools.invoke(messages)
print('응답 타입 : ', type(response))
if hasattr(response, 'tool_calls'):
print(f' 도구 호출 수 : {len(response.tool_calls)}')
return {'messages' : [response]}
def should_continue(state: AgentState):
"""다음 노드를 결정하는 조건부 엣지"""
messages = state["messages"]
last_message = messages[-1]
# 도구 호출이 있으면 "tools"로, 없으면 종료
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
print("🔧 도구 호출 필요 → tools 노드로 이동")
return "tools"
else:
print("✅ 작업 완료 → 종료")
return "end"
print("✅ 노드 함수 정의 완료")
workflow = StateGraph(AgentState)
# 1. 노드 추가 (기능을 수행하는 노드부터 만든다)
workflow.add_node('agent', call_model) #agent는 모델을 뜻함. 위에서 정의한 LLM을 호출하는 함수
workflow.add_node('tools', ToolNode(langchain_tools)) #도구 3개 모음집을 연결
# 2. 시작점 설정
workflow.set_entry_point('agent') #위에서 정의한 agent 노드가 시작점임을 세팅
#시작은 무조건 llm 모델(agent)부터 하지만, 실행 후 결과값은 분기됨
# 조건부 엣지 추가
workflow.add_conditional_edges(
'agent',
should_continue, #결과값을 보고 tools로 갈지, 종료할지 결정하는 함수였음
{'tools': 'tools', # should continue가 도구 호출이 필요하다 라고 아웃풋을 냈다면, tools노드로
'end': END} #완료되면 종료
)
# tools 노드에서 agent로 돌아가는 엣지
workflow.add_edge('tools','agent')
# 그래프 컴파일
app = workflow.compile()
try:
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))
print("✅ 그래프 시각화 완료")
except Exception as e:
print(f"⚠️ 그래프 시각화 실패 (선택사항): {e}")
print("그래프 구조는 정상적으로 작동합니다.")
위 코드 실행 결과
agent - 도구 - 종료 노드 그래프
def run_langgraph_agent(user_input: str):
"""LangGraph 에이전트를 실행하는 함수"""
print("="*60)
print("🚀 LangGraph 에이전트 시작")
print("="*60)
print(f"📝 질문: {user_input}\n")
# 초기 상태
inputs = {"messages": [HumanMessage(content=user_input)]}
# 스트리밍 방식으로 실행
for output in app.stream(inputs):
for key, value in output.items():
print(f"\n--- {key} 노드 ---")
if "messages" in value:
for msg in value["messages"]:
if isinstance(msg, AIMessage):
if msg.content:
print(f"💬 AI: {msg.content[:200]}...")
if hasattr(msg, 'tool_calls') and msg.tool_calls:
for tc in msg.tool_calls:
print(f"🔧 도구 호출: {tc['name']}")
elif isinstance(msg, ToolMessage):
print(f"✅ 도구 결과: {msg.content[:150]}...")
print("\n" + "="*60)
# 최종 응답 반환
final_state = app.invoke(inputs)
final_message = final_state['messages'][-1]
return final_message.content
print("✅ LangGraph 실행 함수 정의 완료")
세팅 완료.
대화를 해보자.
result = run_langgraph_agent("2024 * 35 를 계산하고, 오늘 날씨를 확인해 줘")
print(f"\n💬 최종 답변:\n{result}")
위 코드의 실행 결과
============================================================ 🚀 LangGraph 에이전트 시작 ============================================================
📝질문: 2024 * 35 를 계산하고, 오늘 날씨를 확인해 줘
🤖 LLM 호출 중... 응답 타입 : <class 'langchain_core.messages.ai.AIMessage'> 도구 호출 수 : 2 🔧 도구 호출 필요 → tools 노드로 이동
--- agent 노드 --- 🔧 도구 호출: calculate_tool 🔧 도구 호출: get_current_time_tool 🧮 계산 실행: '2024*35' ⏰ 현재 시간 조회
--- agent 노드 --- 💬 AI: 2024 곱하기 35의 계산 결과는 70,840입니다. 오늘은 2026년 1월 6일 화요일이며, 현재 시간은 오전 9시 10분 41초입니다. 현재 날씨 정보를 원하시면, 어느 지역의 날씨를 확인하고 싶은지 알려주세요....
============================================================ 🤖 LLM 호출 중... 응답 타입 : <class 'langchain_core.messages.ai.AIMessage'> 도구 호출 수 : 2 🔧 도구 호출 필요 → tools 노드로 이동 🧮 계산 실행: '2024*35' ⏰ 현재 시간 조회 🤖 LLM 호출 중... 응답 타입 : <class 'langchain_core.messages.ai.AIMessage'> 도구 호출 수 : 0 ✅ 작업 완료 → 종료
💬 최종 답변: 2024 곱하기 35의 계산 결과는 70,840입니다. 오늘은 2026년 1월 6일 화요일이며, 현재 시간은 오전 9시 10분 43초입니다. 현재 날씨 정보를 확인하려면 어느 지역의 날씨를 알고 싶으신지 알려주세요.
문제는, 기존 대화 내역을 저장하지 못함
대화 내용이 메모리에 저장되지 않음
메모리 저장소를 생성하고 대화 내역을 저장하자
from langgraph.checkpoint.memory import MemorySaver
# 메모리 저장소 생성
memory = MemorySaver()
# 메모리를 사용하는 그래프 컴파일
app_with_memory = workflow.compile(checkpointer=memory)
def run_with_memory(user_input: str, thread_id: str = "1"):
"""메모리를 사용하는 에이전트 실행"""
config = {'configurable' : {'thread_id' : thread_id}}
inputs = {'messages' : [HumanMessage(content=user_input)]}
result = app_with_memory.invoke(inputs, config)
return result["messages"][-1].content
# 연속된 대화 예제
print("\n📝 대화 1:")
response1 = run_with_memory("내 이름은 철수야", thread_id="user_123")
print(f"AI: {response1}")
print("\n📝 대화 2:")
response2 = run_with_memory("내 이름이 뭐였지?", thread_id="user_123")
#user_121 등으로 바꾸면 철수가 아니므로 사용자의 이름을 기억하지 못했으니 알려달라고 함
print(f"AI: {response2}")
