14. エージェントループの構築

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
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]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}

def run_agent(user_input: str) -> str:
    """LLM が最終的な回答を返すまで Think-Act-Observe ループを実行します。"""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    while True:
        # THINK: LLM に次のアクションを判断させます
        print("THINK: Asking LLM to decide")
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        # 終了条件: tool_calls がなければ、これが最終的な回答です
        if not ai_message.tool_calls:
            return ai_message.content
 
        # ACT + OBSERVE: リクエストされたツールを実行し、結果を会話に追加します
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            print(f"ACT: calling '{tool_call['name']}', args={tool_call['args']}")
 
            tool_message = selected_tool.invoke(tool_call)
            print(f"OBSERVE: {tool_message.content}")
 
            messages.append(tool_message)

answer = run_agent("What's the weather in Tokyo, and is it warm enough for a walk?")
print(f'Final answer: {answer}')

THINK: Asking LLM to decide
ACT: calling 'get_weather', args={'city': 'Tokyo'}
OBSERVE: 18°C, cloudy
THINK: Asking LLM to decide
Final answer: 現在の東京は18°C(約64°F)で曇りです。
この気温は一般的に穏やかで、ほとんどの人にとって散歩に快適です。

answer = run_agent("Get the temperature in Cairo, then multiply the number by 3.")
print(f'Final answer: {answer}')

THINK: Asking LLM to decide
ACT: calling 'get_weather', args={'city': 'Cairo'}
OBSERVE: 31°C, sunny
THINK: Asking LLM to decide
ACT: calling 'calculate', args={'expression': '31 * 3'}
OBSERVE: 93
THINK: Asking LLM to decide
Final answer: カイロの現在の気温: 31°C。3倍すると = 93。

def run_agent(user_input: str, max_steps: int = 10) -> str:
    """max_steps 回のイテレーション以内で Think-Act-Observe ループを実行します。"""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        # THINK
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        # 終了条件: tool_calls がなければ、これが最終的な回答です
        if not ai_message.tool_calls:
            return ai_message.content
 
        # ACT + OBSERVE
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            tool_message = selected_tool.invoke(tool_call)
            messages.append(tool_message)
 
    # max_steps に到達: 最終的な回答なしでループが終了しました
    return f"[Stopped after reaching max iterations ({max_steps})]"

@tool
def unhelpful_search(query: str) -> str:
    """情報を検索します。"""
    return "No results found. Try rephrasing your query."
 
llm_with_bad_tool = llm.bind_tools([unhelpful_search])
tool_map_bad = {unhelpful_search.name: unhelpful_search}
 
def run_agent_bad(user_input: str, max_steps: int = 5) -> str:
    messages = [
        SystemMessage(content=(
            "You must ALWAYS use the unhelpful_search tool to find information. "
            "You are NOT allowed to answer from your own knowledge. "
            "If the tool returns no results, you MUST rephrase and search again. "
            "Keep searching until you find the answer."
        )),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        print(f"--- Step {step + 1} ---")
        ai_message = llm_with_bad_tool.invoke(messages)
        messages.append(ai_message)
 
        if not ai_message.tool_calls:
            return ai_message.content
 
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map_bad[tool_call["name"]]
            tool_message = selected_tool.invoke(tool_call)
            print(f"ACT: '{tool_call['name']}' → {tool_message.content}")
            messages.append(tool_message)
 
    return f"[Stopped after reaching max iterations ({max_steps})]"
 
answer = run_agent_bad("What is the population of France?")
print(f"\nFinal answer: {answer}")

--- Step 1 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 2 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 3 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 4 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 5 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
 
Final answer: [Stopped after reaching max iterations (5)]

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain.tools import tool
 
# ツールを定義します
@tool
def calculate(expression: str) -> str:
    """'3 * 21' のような単純な算術式を評価します。"""
    return str(eval(expression))  # 警告: eval() はセキュリティ上のリスクがあります。本番環境では使用しないでください。
 
# ツールをバインドします
tools = [calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}
 
def run_agent(user_input: str, max_steps: int = 10) -> str:
    """ツールのエラーを LLM に渡し、自己修正を可能にするエージェントループ。"""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        # THINK
        print("THINK: Asking LLM to decide")
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        if not ai_message.tool_calls:
            return ai_message.content
 
        # ACT + OBSERVE
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            try:
                print(f"ACT: calling '{tool_call['name']}', args={tool_call['args']}")
                tool_message = selected_tool.invoke(tool_call)
                print(f"OBSERVE: {tool_message.content}")
            except Exception as e:
                print(f"OBSERVE: Error - {e}")
                tool_message = ToolMessage(
                    content=f"Error: {e}",
                    tool_call_id=tool_call["id"],
                )
            messages.append(tool_message)
 
    return f"[Stopped after reaching max iterations ({max_steps})]"

answer = run_agent("Use the calculator tool to compute 10 / 0")
print(f"Final answer: {answer}")

THINK: Asking LLM to decide
ACT: calling 'calculate', args={'expression': '10 / 0'}
OBSERVE: Error - division by zero
THINK: Asking LLM to decide
Final answer: 電卓ツールを使用したところ、エラーが返されました: "division by zero(ゼロ除算)"。
 
説明: 10 / 0 は通常の算術では定義されていないため、有限の数を生成できません。
 
以下のいずれかをご希望でしょうか:
- 片側極限を計算する
- IEEE-754 の浮動小数点結果を表示する
- または別の式を評価する