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 関数がこの1行に置き換わります
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="カイロの気温を取得して、その数値を3倍してください。")],
    "llm_calls": 0,
})
 
print(result["messages"][-1].content)
print(f"\nTotal LLM calls: {result['llm_calls']}")

現在のカイロの気温: 31°C。3倍すると = 93。
 
Total LLM calls: 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="あなたは親切なアシスタントです。",
)
 
result = agent.invoke({
    "messages": [HumanMessage(content="カイロの気温を取得して、その数値を3倍してください。")],
})
print(result["messages"][-1].content)

現在のカイロの気温: 31°C。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="東京の天気はどうですか?")],
})
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] About to call LLM, current messages: {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="カイロの気温を取得して、その数値を3倍してください。")],
})
print(result["messages"][-1].content)

[before_model] About to call LLM, current messages: 1
[before_model] About to call LLM, current messages: 3
[before_model] About to call LLM, current messages: 5
現在のカイロの気温: 31°C。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: 営業時間やあいさつなどの一般的な質問。 "
            "complex: 紛争や返金など、慎重な推論が必要な相談。 "
            "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"顧客のリクエストを分類してください。\n\nリクエスト: {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": "配送中、明日到着予定", "67890": "配達済み"}
    return fake_data.get(order_id, f"注文 {order_id} が見つかりません。")
 
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 [
    "営業時間は何時ですか?",
    "注文品が破損して届きました。交換すべきですか、返金すべきですか?",
    "注文番号 12345 の配送状況は?",
]:
    result = agent.invoke({"messages": [HumanMessage(content=question)]})
    print(f"Q: {question}")
    print(f"[{result['intent']}] A: {result['messages'][-1].content}\n")

Q: 営業時間は何時ですか?
[simple] A: 決まった営業時間はありません — 24時間365日ご利用いただけます。
...
 
Q: 注文品が破損して届きました。交換すべきですか、返金すべきですか?
[complex] A: 注文品が破損して届いた場合、一般的に **交換または全額返金** を受ける権利があるはずです。
...
 
Q: 注文番号 12345 の配送状況は?
[order] A: 注文 12345 は **配送中** で、**明日到着予定** です。