체크포인터를 사용한 상태 관리하기
상태 영속성
- 여러 대화나 세션의 상태 독립적 관리, 여러 워크플로우 처리
- 그래프 실행 중 노드의 상태 저장
오류 복구
기본 제공 체크포인터 3가지
- BaseCheckpointServer: 추상화 기본 클래스
- InMemorySarver: 메모리 기반 구현
- 휘발성: 프로그램 종료 시 데이터 삭제
- 빠른 속도: 메모리 접근이므로 빠름
- 개발/테스트: 프로토타이핑과 테스트에 사용
- SQLiteSaver, PostgressSaver: DB기반 영구 저장소 구현
코드 예시
import json
import os
from dotenv import load_dotenv
from typing import Dict, Any
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver # 메모리 기반
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
load_dotenv()
class MemoryBotState(BaseModel):
user_message: str = Field(default="", description="사용자 입력 메시지")
user_name: str = Field(default="", description="사용자 이름")
user_preferences: Dict[str, Any] = Field(default_factory=dict, description="사용자 선호도")
response: str = Field(default="", description="최종 응답")
llm = ChatOpenAI(model="gpt-5-mini", api_key=os.getenv("OPENAI_API_KEY"))
# 메시지 처리 노드 - 메모리 로드/저장 로직 제거
def process_message(state: MemoryBotState) -> Dict[str, Any]:
message = state.user_message
user_name = state.user_name
preferences = state.user_preferences.copy()
# 시스템 프롬프트
system_prompt = f"""
당신은 사용자의 정보를 기억하는 메모리 봇입니다.
현재 기억하는 정보:
- 사용자 이름: {user_name if user_name else "모름"}
- 좋아하는 것: {preferences.get("likes", [])}
- 싫어하는 것: {preferences.get("dislikes", [])}
사용자 메시지를 분석하여 다음 JSON 형식으로 응답하세요:
{{
"response": "사용자에게 줄 응답 메시지",
"new_name": "새로 알게 된 이름 (없으면 null)",
"new_likes": ["새로 알게 된 좋아하는 것들"],
"new_dislikes": ["새로 알게 된 싫어하는 것들"]
}}
"""
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=message)
]
response = llm.invoke(messages)
result = json.loads(response.content)
# 새로운 정보 업데이트
if result.get("new_name"):
user_name = result["new_name"]
if result.get("new_likes"):
preferences.setdefault("likes", []).extend(result["new_likes"])
if result.get("new_dislikes"):
preferences.setdefault("dislikes", []).extend(result["new_dislikes"])
bot_response = result.get("response", "죄송해요, 이해하지 못했어요.")
return {
"response": bot_response,
"user_name": user_name,
"user_preferences": preferences
}
# 메모리 봇 그래프 생성 - InMemorySaver 통합
def create_memory_bot_graph():
# InMemorySaver로 자동 메모리 관리
checkpointer = InMemorySaver()
workflow = StateGraph(MemoryBotState)
# 단일 노드 구조
workflow.add_node("process_message", process_message)
# 단순한 워크플로: START -> process_message -> END
workflow.add_edge(START, "process_message")
workflow.add_edge("process_message", END)
return workflow.compile(checkpointer=checkpointer)
def main():
print("=== InMemorySaver 메모리 봇 테스트===\n")
app = create_memory_bot_graph()
thread_id = "mochabbang_123"
# 테스트 대화
conversations = [
"안녕하세요!",
"내 이름은 모카빵이야",
"김치짜글이를 좋아해",
"굴 싫어해",
"내 이름은 뭐였지?",
"내가 좋아하는 것과 싫어하는 것은?"
]
for i, message in enumerate(conversations, 1):
print(f"[{i}] 사용자: {message}")
# InMemorySaver 사용 시 config 설정
config = {"configurable": {"thread_id": thread_id}}
result = app.invoke({"user_message": message}, config)
print(f"[{i}] 챗봇: {result['response']}")
print(
f"메모리: 이름={result.get('user_name', '없음')}, "
f"좋아하는 것={result.get('user_preferences', {})}\n"
)
if __name__ == "__main__":
main()
루프 워크플로우
- 점수 맞추기 게임 예시를 진행
- 조건부 라우팅을 이용하여 사용자 입력 부분을 반복적으로 진행되도록 구성
- 실패할 경우 시도 횟 수 증가와 함께 최대 횟 수 초과 할 경우 게임 진 것으로 하여 종료 처리 진행
import random
import os
from dotenv import load_dotenv
from typing import Dict, Any, Literal
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
# 그래프 상태 정의
class GuessGameState(BaseModel):
target_number: int = Field(default=0, description="맞춰야 할 숫자")
user_guess: int = Field(default=0, description="사용자 추측")
attempts: int = Field(default=0, description="시도 횟수")
max_attempts: int = Field(default=5, description="최대 시도 횟수")
game_status: str = Field(default="playing", description="게임 상태")
response: str = Field(default="", description="응답 메시지")
# 게임 설정 노드
def game_setup(state: GuessGameState) -> Dict[str, Any]:
target = random.randint(1, 50)
print("게임 시작!")
# 상태 업데이트 - 딕셔너리로 변경할 필드만 반환
return {
"target_number": target,
"game_status": "playing",
"response": f"1~50 사이의 숫자를 맞춰보세요! (최대 {state.max_attempts}회)",
"attempts": 0,
}
# 사용자 추측 노드 (시뮬레이션)
def user_guess(state:GuessGameState) -> Dict[str, Any]:
guess = input("입력 : ")
print(f"[{state.attempts + 1}번째 시도] 추측: {guess}")
return {"user_guess": int(guess), "attempts": state.attempts + 1}
# 추측 확인 노드
def check_guess(state: GuessGameState) -> Dict[str, Any]:
target = state.target_number
guess = state.user_guess
attempts = state.attempts
print(f"[check_guess] {guess} (시도: {attempts}회)")
if guess == target:
print("정답")
return {
"game_status": "won",
"response": f"정답! {guess}를 {attempts}번 만에 맞췄습니다!"
}
elif attempts >= state.max_attempts:
print("시도 횟수 초과")
return {
"game_status": "lost",
"response": f"게임 종료! 정답은 {target}이었습니다.",
}
else:
hint = "더 큰 수" if guess < target else "더 작은 수"
remaining = state.max_attempts - attempts
print(f"계속 진행: {hint}")
return {
"game_status": "playing",
"response": f"{guess}는 틀렸습니다. {hint}를 시도해보세요! (남은 기회: {remaining}회)",
}
# 조건부 라우팅 함수 - 루프 구현의 핵심
def route_game(state: GuessGameState) -> Literal["continuite", "end"]:
print(f"라우팅 체크: 상태={state.game_status}, 시도={state.attempts}")
if state.game_status == "playing":
return "continue"
else:
return "end"
# 루프 워크플로 그래프 생성
def create_guess_game_graph():
# StataGraph 초기화 - 상태 스키마 지정
workflow = StateGraph(GuessGameState)
# 노드 추가
workflow.add_node("setup", game_setup)
workflow.add_node("guess", user_guess)
workflow.add_node("check", check_guess)
# 엣지 추가
workflow.add_edge(START, "setup")
workflow.add_edge("setup", "guess")
workflow.add_edge("guess", "check")
# 조건부 엣지
workflow.add_conditional_edges(
"check",
route_game,
{
"continue": "guess",
"end": END
}
)
# 그래프 컴파일
return workflow.compile()
def main():
print("=== 루프 워크플로우 예제 ===\n")
app = create_guess_game_graph()
# 그래프 실행 - 초기 상태로 시작
initial_state = GuessGameState(max_attempts=5)
result = app.invoke(initial_state)
print(f"\n최종 결과 : {result['response']}")
print(f"게임 상태: {result['game_status']}")
print(f"총 시도: {result['attempts']}회")
if __name__ == "__main__":
main()
병렬 처리 워크플로 구현
- LLM 실행, API 호출, DB접근 등 처리 시간 오래 걸리는 것들을 병렬로 실행하여 시간 단축
import time
import random
from typing import Dict, Any
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel, Field
# State 클래스 정의
class DashboardState(BaseModel):
user_location: str = "서울"
weather_data: Dict[str, Any] = {}
news_data: Dict[str, Any] = {}
stock_data: Dict[str, Any] = {}
dashboard_report: str = ""
start_time: float = 0.0
# 코디네이터 노드
def coordinator(state: DashboardState) -> Dict[str, Any]:
print(f"대시보드 생성 시작 - 위치: {state.user_location}")
return {"start_time": time.time()}
# 병렬 실행 노드 1 - 날씨 데이터 수집
def weather_checker(state: DashboardState) -> Dict[str, Any]:
print("날씨 확인 중...")
time.sleep(random.uniform(1.0, 2.0))
weather_info = {
"location": state.user_location,
"condition": "맑음",
"temperature": 22,
"humidity": 65
}
print(f"날씨: {weather_info['condition']}, {weather_info['temperature']}도")
return {"weather_data": weather_info}
# 병렬 실행 노드 2 - 뉴스 데이터 수집
def news_fetcher(state: DashboardState) -> Dict[str, Any]:
print("뉴스 수집 중...")
time.sleep(random.uniform(1.5, 2.5))
news_info = {
"articles": [
{"title": "AI 기술 발전 소식", "summary": "AI 분야 새로운 혁신"},
{"title": "경제 동향 분석", "summary": "글로벌 경제 전망"},
],
"count": 2
}
print(f"뉴스 {news_info['count']}개 수집완료")
return {"news_data": news_info}
# 병렬 실행 노드 3 - 주식 데이터 분석
def stock_analyzer(state: DashboardState) -> Dict[str, Any]:
print("주식 분석 중...")
time.sleep(random.uniform(2.0, 3.0))
stock_info = {
"KOSPI": {"price": 2650.5, "change": +1.2},
"NASDAQ": {"price": 15780.3, "change": -0.8}
}
print("주식 분석 완료")
return {"stock_data": stock_info}
# 집계 노드 - 모든 병렬 작업 완료 후 실행
def aggregator(state: DashboardState) -> Dict[str, Any]:
print("리포트 생성 중...")
parallel_time = time.time() - state.start_time
report = f"""
대시보드 리포트
날씨: {state.weather_data.get('condition', 'N/A')} {state.weather_data.get('temperature', 'N/A')}도
뉴스: {state.news_data.get('count', 0)}개 기사
주식: KOSPI {state.stock_data.get('KOSPI', {}).get('price', 'N/A')}
실행시간: {parallel_time:.1f}초
"""
print(f"대시보드 완료 ({parallel_time:.1f}초)")
return {"dashboard_report": report}
def create_graph():
workflow = StateGraph(DashboardState)
workflow.add_node("coordinator", coordinator)
workflow.add_node("weather", weather_checker)
workflow.add_node("news", news_fetcher)
workflow.add_node("stock", stock_analyzer)
workflow.add_node("aggregator", aggregator)
#병렬 실행 구조 정의
workflow.add_edge(START, "coordinator")
workflow.add_edge("coordinator", "weather")
workflow.add_edge("coordinator", "news")
workflow.add_edge("coordinator", "stock")
workflow.add_edge("weather", "aggregator")
workflow.add_edge("news", "aggregator")
workflow.add_edge("stock", "aggregator")
workflow.add_edge("aggregator", END)
return workflow.compile()
def main():
print("=== LangGraph 병렬 실행 예제 ===\n")
app = create_graph()
initial_state = DashboardState(user_location="부산")
print("병렬 실행 시작!")
result = app.invoke(initial_state)
print("\n최종 결과 : ")
print(result['dashboard_report'])
if __name__ == "__main__":
main()
ToolNode 사용한 도구 사용 기능 만들기
- 도구 함수 구성
- 질문 목록을 LLM 모델에게 전달하여 도구 호출 여부를 정하고,
도구 호출 해야한다는 결과가 나오면 tool 관련 함수를 재호출 하여 결과를 전달 받는 구조로 설정
import httpx
import os
import math
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, ToolMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain.chat_models import init_chat_model
from geopy.geocoders import Nominatim
load_dotenv()
# 계산기 도구
def calculator(expression: str) -> str:
"""수학 계산을 수행합니다."""
print(f"계산 요청: {expression}")
try:
expression = expression.replace("sqrt", "math.sqrt")
expression = expression.replace("sin", "math.sin")
expression = expression.replace("cos", "math.cos")
# 안전한 계산 실행
result = eval(expression, {"__builtins__": {}, "math": math}) # eval 함수는 파이썬 모두 실행할 수 있으므로 위험한 코드 예방하기 위해 math 패키지만 이용한다고 구성
return f"계산 결과 : {result}"
except Exception as e:
return f"계산 오류 : {str(e)}"
# 날씨 관련 도구 함수
def get_weather(city_name: str) -> dict:
"""도시 이름을 받아 해당 도시의 현재 날씨 정보를 반환합니다."""
if city_name:
latitude, longitude = get_coordinates(city_name)
else:
raise ValueError("City name must be provided to get weather information.")
url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}¤t_weather=true"
response = httpx.get(url)
response.raise_for_status()
return response.json()
def get_coordinates(city_name: str) -> tuple[float, float]:
"""도시 이름을 ㅏㄷ아 위도와 경도를 반환합니다."""
geolocator = Nominatim(user_agent="weather_app")
location = geolocator.geocode(city_name)
if location:
return location.latitude, location.longitude
else:
raise ValueError(f"Could not find coordinates for {city_name}")
def currency_converter(amount: float, from_currency: str, to_currency: str) -> str:
"""통화 간 환율을 계산합니다."""
print(f"{amount} {from_currency}를 {to_currency}로 변환합니다.")
rates = {("USD", "KRW"): 1320.50, ("KRW", "USD"): 0.00076}
rate_key = (from_currency.upper(), to_currency.upper())
if rate_key in rates:
rate = rates[rate_key]
converted = amount * rate
return f"{amount} {from_currency} = {converted:.2f} {to_currency}"
return f"{amount} {from_currency} = {amount} {to_currency} (동일 통화)"
def should_continue(state: MessagesState):
print("\n--- 분기 결정 ---")
last_message = state["messages"][-1]
if last_message.tool_calls:
print(f"결정: 도구 호출 필요 ({len(last_message.tool_calls)}개)")
return "tools"
else:
print("결정: 최종 응답으로 종료")
return END
# call_model 노드를 생성하는 함수
def create_call_model_function(model_with_tools):
"""model_with_tools를 클로저로 캡처하는 call_model 함수 생성"""
def call_model(state: MessagesState):
"""LLM을 호출하여 응답을 생성하는 노드 함수"""
last_message = state['messages'][-1]
# 도구 실행 결과를 받았는지, 아니면 사용자 질문을 받았는지에 따라 분기
if isinstance(last_message, ToolMessage):
print("\n--- 모델 호출 (도구 결과 기반) ---")
# 도구 실행 결과가 길 수 있으므로 일부만 출력
print(f"입력(도구 결과): {last_message.content[:300]}...")
else:
print("\n--- 모델 호출 (사용자 질문 기반) ---")
print(f"입력(사용자 메시지): {last_message.content}")
# 모델을 호출하여 다음 행동을 결정하게 함
response = model_with_tools.invoke(state["messages"])
# 모델의 결정에 따라 로그 출력
if response.tool_calls:
print(f"모델의 판단: 도구 호출 -> {response.tool_calls}")
else:
print(f"모델의 판단: 최종 답변 생성 -> {response.content}")
return {"messages": [response]}
return call_model
# 워크플로 생성 함수
def create_graph(model_with_tools, tool_node):
"""LLM 워크플로 그래프 생성"""
workflow = StateGraph(MessagesState)
call_model = create_call_model_function(model_with_tools)
workflow.add_node("call_model", call_model)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "call_model")
workflow.add_conditional_edges("call_model", should_continue, ["tools", END])
workflow.add_edge("tools", "call_model")
return workflow.compile()
# 워크플로 생성 및 실행
def llm_tool_call(query: str):
"""하나의 질문에 대해 전체 LLM 워크플로를 실행하고 로그를 출력"""
tools = [calculator, get_weather, currency_converter]
tool_node = ToolNode(tools)
model = init_chat_model(model="gpt-5-mini", model_provider="openai")
model_with_tools = model.bind_tools(tools)
print(f"질문: {query}")
print("-" * 50)
# LLM 기반 워크플로 생성
app = create_graph(model_with_tools, tool_node)
# 그래프 시각화
mermaid_png = app.get_graph().draw_mermaid_png()
with open("./06.png", "wb") as f:
f.write(mermaid_png)
# 워크플로 실행
app.invoke({"messages": [HumanMessage(content=query)]})
# 최종 응답은 call_model 노드의 로그에서 출력
print("-" * 50)
print("처리 완료")
print("=" * 50 + "\n")
def main():
print("=== LangGraph ToolNode 예제 (LLM 기반) ===\n")
test_queries = [
"2 + 3 * 4를 계산해줘",
"서울 날씨 어때?",
"100달러를 원화로 바꿔줘",
"sqrt(16)을 계산해줘",
"도쿄 날씨가 궁금해",
"1000원을 달러로 환전해줘",
]
print("\nLLM 기반 도구 호출 시작:")
for query in test_queries:
try:
llm_tool_call(query)
except Exception as e:
print(f"'{query}' 처리 중 오류 발생: {e}")
print("=" * 50 + "\n")
if __name__ == "__main__":
main()
휴먼 인 더 루프 반영하기
- LLM이 좋은 결과를 내기 위해서 최선을 다하지만 사람의 의도와 다른 결과를 나타낼 수 있음.
- LLM이 사람의 의도를 알기 위해서는 사람의 개입이 필요함. (LLM에게 힌트 제공)
- 휴먼 인 더 루프의 4단계
- 중단점(BreakPoint) : AI가 인간의 갱신이 필요한 지점을 인식
- 컨텍스트 제공 : 인간이 올바른 결정을 내릴 수 있도록 충분한 정보 제공
- 피드백 수집 : 구조화 된 방식으로 인간의 입력 수집
- 재개(Resume) : 인간의 결정에 따라 워크플로 계속 진행
from dotenv import load_dotenv
from typing import Literal
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from pydantic import BaseModel, Field
load_dotenv()
# State 정의
class AgentState(BaseModel):
user_message: str = Field(default="", description="사용자 입력 작업")
task_details: str = Field(default="", description="작업 상세 정보")
response: str = Field(default="", description="응답 결과")
# 노드 함수 정의
def get_llm_response_node(state: AgentState, llm):
"""LLM과 상호작용하여 응답을 생성하거나, 추가 정보를 요청하는 노드"""
details = state.task_details
if details:
print(f"\n상세 정보를 바탕으로 작업 실행: '{details}'")
prompt = f"다음 요청에 따라 보고서를 작성해주세요: {details}"
else:
task = state.user_message
print(f"\n작업 실행: '{task}' 작업을 수행합니다...")
# LLM에게 상세 정보 묻는 질문을 유도하고, 반드시 '?' 끝내도록 지시
prompt = f"'{task}' 작업을 수행하려고 합니다. 어떤 종류의 보고서가 필요한지, 구체적인 주제는 무엇인지 질문해주세요. 추가 정보가 필요하면, 반드시 응답의 마지막을 물음표('?')로 끝내주세요."
response = llm.invoke(prompt).content
print("--- LLM 응답 ---")
print(response)
print("----------------------")
return {"response": response, "task_details": ""}
# 사람의 입력을 받는 노드
def get_task_details_node(state: AgentState) -> AgentState:
"""LLM의 질문에 대한 사용자 답변을 입력 받는 노드"""
print("\nLLM의 질문에 답변해주세요.")
user_input = input("답변: ")
return {"task_details": user_input}
# 조건부 분기 노드
def check_llm_response(state: AgentState) -> Literal["get_details", "end"]:
"""LLM의 응답이 질문인지 확인하여 다음 단계를 결정합니다."""
print("LLM 응답 분석 중...")
if state.response.strip().endswith("?"):
print("LLM이 추가 정보를 요청했습니다. 사용자 입력을 받습니다.")
return "get_details"
print("최종 보고서가 생성되었습니다. 워크플로를 종료합니다.")
return "end"
# 그래프 생성
def create_graph():
"""Human-in-the-loop 워크플로 그래프를 생성합니다."""
# 그래프 전체에서 사용할 LLM 모델을 초기화 합니다.
llm = init_chat_model(model="gpt-5-mini", model_provider="openai")
def get_llm_response_with_llm(state):
return get_llm_response_node(state, llm)
workflow = StateGraph(AgentState)
workflow.add_node("get_llm_response", get_llm_response_with_llm)
workflow.add_node("get_details", get_task_details_node)
workflow.add_edge(START, "get_llm_response")
workflow.add_conditional_edges(
"get_llm_response",
check_llm_response,
{
"get_details": "get_details",
"end": END
}
)
workflow.add_edge("get_details", "get_llm_response")
return workflow.compile()
def main():
print("=== LangGraph Human-in-the-loop 간소화 예제 ===\n")
app = create_graph()
# 그래프 시각화
mermaid_png = app.get_graph().draw_mermaid_png()
with open("./chapter6/langgraph/07.png", "wb") as f:
f.write(mermaid_png)
final_state = app.invoke(AgentState(user_message="블로그 글 작성"))
print("\n--- 워크플로 종료 ---")
print("최종 응답: ")
print(final_state["response"])
if __name__ == "__main__":main()
하위 그래프 사용하기
- 하나의 노드가 그래프인 하위 그래프 사용
- 그래프 안에 그래프가 들어가는 형태가 되며, 매우 복잡한 워크플로가 구성됨
- 하위 그래프는 독립적으로 잘 동작하는 하나의 워크플로우로 구성,
상위에서 이를 재사용 하는 방식이 바람직함
모듈화 잘되면 테스트, 재사용성, 확장성 용이
(일반적인 개발에서의 ‘단일책임원칙’과 비슷한 맥락)
- 예시 코드와 이미지
import httpx
import json
from dotenv import load_dotenv
from geopy.geocoders import Nominatim
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain.chat_models import init_chat_model
from typing import Literal
load_dotenv()
# 날씨 관련 도구 함수
def get_weather(city_name: str) -> dict:
"""도시 이름을 받아 해당 도시의 현재 날씨 정보를 반환합니다."""
if city_name:
latitude, longitude = get_coordinates(city_name)
else:
raise ValueError("City name must be provided to get weather information.")
url = f"https://api.open-meteo.com/v1/forecast?latitude={latitude}&longitude={longitude}¤t_weather=true"
response = httpx.get(url)
response.raise_for_status()
print(f"\n\n도구 호출: {response.json()}")
return response.json()
def get_coordinates(city_name: str) -> tuple[float, float]:
"""도시 이름을 ㅏㄷ아 위도와 경도를 반환합니다."""
geolocator = Nominatim(user_agent="weather_app")
location = geolocator.geocode(city_name)
if location:
return location.latitude, location.longitude
else:
raise ValueError(f"Could not find coordinates for {city_name}")
# 기상 전문가 하위 그래프 생성
def create_weather_agent():
"""날씨 관련 질문을 처리하는 전문가 하위 그래프 생성"""
model = init_chat_model(model="gpt-5-mini", model_provider="openai").bind_tools([get_weather])
tool_node = ToolNode([get_weather])
def call_model(state: MessagesState):
result = {"messages": [model.invoke(state["messages"])]}
print(f"\n\n 기상전문가 result: {result}, toolcalls: {result["messages"][-1].tool_calls}")
return result
graph = StateGraph(MessagesState)
graph.add_node("call_model", call_model)
graph.add_node("tool_node", tool_node)
graph.add_edge(START, "call_model")
graph.add_conditional_edges(
"call_model",
lambda s: "tool_node" if s["messages"][-1].tool_calls else END,
{ "tool_node": "tool_node", END: END},
)
graph.add_edge("tool_node", "call_model")
return graph.compile()
def router(state: MessagesState) -> Literal["weather_expert", "general_agent"]:
query = state["messages"][-1].content.lower()
if "날씨" in query or "기온" in query:
print("라우팅 결정: 기상전문가에게 위임")
return "weather_expert"
print("라우팅 결정: 일반 에이전트가 처리")
return "general_agent"
# 메인 그래프 생성
def create_main_agent(weather_subgraph):
"""질문을 라우팅하고 처리하는 메인 에이전트 그래프를 생성"""
main_model = init_chat_model(model="gpt-5-mini", model_provider="openai")
workflow = StateGraph(MessagesState)
workflow.add_node(
"general_agent",
lambda s: {"messages": [main_model.invoke(s["messages"])]}
)
workflow.add_node("weather_expert", weather_subgraph)
workflow.add_conditional_edges(
START,
router,
{
"weather_expert": "weather_expert",
"general_agent": "general_agent"
},
)
workflow.add_edge("general_agent", END)
workflow.add_edge("weather_expert", END)
return workflow.compile()
def main():
print("=== LangGraph 하위 그래프 예제 (기상전문가) ===\n")
weather_agent = create_weather_agent()
main_agent = create_main_agent(weather_agent)
main_graph_image = main_agent.get_graph(xray=True).draw_mermaid_png()
with open("./chapter6/langgraph/10_main_agent_graph.png", "wb") as f:
f.write(main_graph_image)
queries = ["성남 날씨 어때?", "잠은 몇 시간 자는게 좋을까?"]
for query in queries:
print(f"\n--- 질문: {query} ---")
result = main_agent.invoke({"messages": [HumanMessage(content=query)]})
print(f"최종 답변: {result['messages'][-1].content}")
print("-" * 20)
if __name__ == "__main__":
main()