16. 사전 구축 컴포넌트와 다중 분기 라우팅

from langgraph.prebuilt import ToolNode
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """도시의 현재 날씨를 가져옵니다."""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """간단한 산술 표현식을 계산합니다. 예: '3 * 21'."""
    return str(eval(expression))  # 주의: eval()은 보안에 취약합니다. 프로덕션에서는 사용하지 마세요.
 
tools = [get_weather, calculate]
 
# 15장의 tool_map + tool_node 함수가 이 한 줄로 대체됩니다
tool_node = ToolNode(tools)

def should_continue(state: AgentState) -> Literal["tool_node", "__end__"]:
    """도구를 실행할지 그래프를 종료할지 결정합니다."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"
    return END

from langgraph.prebuilt import ToolNode, tools_condition
 
builder.add_node("tools", ToolNode(tools))                  # 노드 이름을 "tools"로 등록
builder.add_conditional_edges("llm_call", tools_condition)  # "tools" 또는 END로 라우팅

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain.tools import tool
 
@tool
def get_weather(city: str) -> str:
    """도시의 현재 날씨를 가져옵니다."""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """간단한 산술 표현식을 계산합니다. 예: '3 * 21'."""
    return str(eval(expression))  # 주의: eval()은 보안에 취약합니다. 프로덕션에서는 사용하지 마세요.
 
tools = [get_weather, calculate]
 
class AgentState(MessagesState):
    llm_calls: int
 
llm = ChatOpenAI(model="gpt-5-mini")
model_with_tools = llm.bind_tools(tools)
 
def llm_call(state: AgentState):
    """LLM을 호출하고 응답을 리턴합니다."""
    response = model_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "llm_calls": state.get("llm_calls", 0) + 1,
    }
 
builder = StateGraph(AgentState)
 
builder.add_node("llm_call", llm_call)
builder.add_node("tools", ToolNode(tools))  # 15장의 tool_node 함수 대신 ToolNode 사용
 
builder.add_edge(START, "llm_call")
builder.add_conditional_edges("llm_call", tools_condition)  # 15장의 should_continue 대신 tools_condition 사용
builder.add_edge("tools", "llm_call")
 
agent = builder.compile()

result = agent.invoke({
    "messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\n총 LLM 호출 횟수: {result['llm_calls']}")

Current temperature in Cairo: 31°C. Multiplied by 3 = 93.
 
총 LLM 호출 횟수: 3

builder.add_node("run_tools", ToolNode(tools))
builder.add_conditional_edges(
    "llm_call",
    tools_condition,
    {"tools": "run_tools", END: END}   # "tools" 반환 → run_tools 노드로, END 반환 → 종료
)

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_core.messages import HumanMessage
 
@tool
def get_weather(city: str) -> str:
    """도시의 현재 날씨를 가져옵니다."""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """간단한 산술 표현식을 계산합니다. 예: '3 * 21'."""
    return str(eval(expression))  # 주의: eval()은 보안에 취약합니다. 프로덕션에서는 사용하지 마세요.
 
agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[get_weather, calculate],
    system_prompt="You are a helpful assistant.",
)
 
result = agent.invoke({
    "messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
})
print(result["messages"][-1].content)

Current temperature in Cairo: 31°C. Multiplied by 3 = 93.

from pydantic import BaseModel
from langchain.agents import create_agent
 
class WeatherReport(BaseModel):
    city: str
    temperature: str
    condition: str
 
agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[get_weather, calculate],
    response_format=WeatherReport,
)
 
result = agent.invoke({
    "messages": [HumanMessage(content="What's the weather in Tokyo?")],
})
print(result["structured_response"])

city='Tokyo' temperature='18°C' condition='cloudy'

from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
 
@before_model
def log_llm_call(state: AgentState, runtime) -> None:
    """LLM 호출 직전마다 메시지 개수를 출력합니다."""
    print(f"[before_model] LLM 호출 직전, 현재 메시지 {len(state['messages'])}개")
 
agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[get_weather, calculate],
    middleware=[log_llm_call],
)
 
result = agent.invoke({
    "messages": [HumanMessage(content="Get the temperature in Cairo, then multiply the number by 3.")],
})
print(result["messages"][-1].content)

[before_model] LLM 호출 직전, 현재 메시지 1개
[before_model] LLM 호출 직전, 현재 메시지 3개
[before_model] LLM 호출 직전, 현재 메시지 5개
Current temperature in Cairo: 31°C. Multiplied by 3 = 93.

from langgraph.graph import MessagesState
 
class State(MessagesState):
    intent: str    # 분류 결과: "simple", "complex", "order"

from typing import Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
 
class IntentRoute(BaseModel):
    """사용자 문의의 분류 결과."""
    intent: Literal["simple", "complex", "order"] = Field(
        description=(
            "simple: general questions like business hours or greetings. "
            "complex: consultations that need careful reasoning, like disputes or refunds. "
            "order: requests to look up a specific order."
        )
    )
 
classifier_llm = ChatOpenAI(model="gpt-5.4-nano").with_structured_output(IntentRoute)
 
def classify(state: State):
    """사용자 문의의 유형을 분류합니다."""
    question = state["messages"][-1].content
    result = classifier_llm.invoke(
        f"Classify the customer's request.\n\nRequest: {question}"
    )
    return {"intent": result.intent}

def route_by_intent(state: State) -> Literal["simple", "complex", "order"]:
    """분류 결과에 따라 다음 노드를 결정합니다."""
    return state["intent"]

simple_llm = ChatOpenAI(model="gpt-5.4-mini")
 
def simple_handler(state: State):
    """단순 문의에 소형 모델로 답변합니다."""
    response = simple_llm.invoke(state["messages"])
    return {"messages": [response]}

complex_llm = ChatOpenAI(model="gpt-5.4")
 
def complex_handler(state: State):
    """복잡한 상담에 고성능 모델로 답변합니다."""
    response = complex_llm.invoke(state["messages"])
    return {"messages": [response]}

from langchain.tools import tool
from langchain.agents import create_agent
 
@tool
def get_order_status(order_id: str) -> str:
    """주문 번호로 배송 상태를 조회합니다."""
    fake_data = {"12345": "In transit, expected tomorrow", "67890": "Delivered"}
    return fake_data.get(order_id, f"Order {order_id} not found.")
 
order_agent = create_agent(
    model="openai:gpt-5.4-mini",
    tools=[get_order_status],
)

from langgraph.graph import StateGraph, START, END
 
builder = StateGraph(State)
 
builder.add_node("classify", classify)
builder.add_node("simple", simple_handler)
builder.add_node("complex", complex_handler)
builder.add_node("order", order_agent)      # create_agent가 만든 그래프를 노드로 등록
 
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_by_intent)
builder.add_edge("simple", END)
builder.add_edge("complex", END)
builder.add_edge("order", END)
 
agent = builder.compile()

from langchain_core.messages import HumanMessage
 
for question in [
    "What are your business hours?",
    "My order arrived damaged. Should I get an exchange or a refund?",
    "What's the delivery status of order 12345?",
]:
    result = agent.invoke({"messages": [HumanMessage(content=question)]})
    print(f"Q: {question}")
    print(f"[{result['intent']}] A: {result['messages'][-1].content}\n")

Q: What are your business hours?
[simple] A: I don’t have fixed business hours—I’m available 24/7.
...
 
Q: My order arrived damaged. Should I get an exchange or a refund?
[complex] A: If your order arrived damaged, you should generally be entitled to a **replacement/exchange or a full refund**.
...
 
Q: What's the delivery status of order 12345?
[order] A: Order 12345 is **in transit** and is **expected tomorrow**.