ReActエージェント実装ガイド

import os
import re
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    base_url="https://openrouter.ai/api/v1",
    api_key=os.getenv("OPENROUTER_API_KEY"),
)

# ================================
# ツール定義
# ================================
class Tools:
    @staticmethod
    def search(query: str) -> str:
        """Web検索をシミュレート（実際はTavily APIなどを使用）"""
        return f"検索結果: '{query}' に関する情報が見つかりました。"

    @staticmethod
    def calculate(expression: str) -> str:
        """数式を計算"""
        try:
            result = eval(expression)
            return f"計算結果: {expression} = {result}"
        except Exception as e:
            return f"計算エラー: {e}"

    @staticmethod
    def get_weather(city: str) -> str:
        """天気情報を取得（シミュレート）"""
        return f"{city}の天気: 晴れ、気温22度、湿度45%"

AVAILABLE_TOOLS = {
    "search": Tools.search,
    "calculate": Tools.calculate,
    "get_weather": Tools.get_weather,
}

# ================================
# ReActエージェント
# ================================
class ReActAgent:
    def __init__(self, model: str = "anthropic/claude-sonnet-4", max_iterations: int = 10):
        self.model = model
        self.max_iterations = max_iterations

    def parse_action(self, text: str) -> tuple[str, str] | None:
        """Action: tool_name(argument) を解析"""
        pattern = r"Action:\s*(\w+)\((.+?)\)"
        match = re.search(pattern, text)
        if match:
            return match.group(1), match.group(2).strip('"\'')
        return None

    def execute_action(self, tool_name: str, argument: str) -> str:
        """ツールを実行"""
        if tool_name in AVAILABLE_TOOLS:
            try:
                return AVAILABLE_TOOLS[tool_name](argument)
            except Exception as e:
                return f"ツール実行エラー: {e}"
        return f"未知のツール: {tool_name}"

    def run(self, user_query: str) -> str:
        """エージェントを実行"""
        messages = [
            {"role": "system", "content": REACT_SYSTEM_PROMPT},
            {"role": "user", "content": user_query}
        ]

        for iteration in range(self.max_iterations):
            response = client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0
            )

            assistant_message = response.choices[0].message.content
            messages.append({"role": "assistant", "content": assistant_message})

            if "Final Answer:" in assistant_message:
                return assistant_message.split("Final Answer:")[-1].strip()

            action = self.parse_action(assistant_message)
            if action:
                tool_name, argument = action
                observation = self.execute_action(tool_name, argument)
                messages.append({"role": "user", "content": f"Observation: {observation}"})

        return "最大イテレーション数に達しました。"

# 使用例
agent = ReActAgent()
result = agent.run("東京の今日の天気を教えてください")
print(result)

--- イテレーション 1 ---
エージェント:
Thought: ユーザーは一人当たりGDPを計算したいようです。
GDPが4.2兆ドル（4,200,000,000,000ドル）で、人口が1.24億人（124,000,000人）なので、
一人当たりGDP = GDP / 人口 で計算できます。
Action: calculate(4200000000000 / 124000000)

観察: 計算結果: 4200000000000 / 124000000 = 33870.96774193548

--- イテレーション 2 ---
エージェント:
Thought: 計算結果が出ました。一人当たりGDPは約33,871ドルです。
これをわかりやすく整形して回答します。
Final Answer: 2024年の日本の一人当たりGDPは約33,871ドル（約3.4万ドル）です。

# 追加ツールの例
class AdvancedTools:
    @staticmethod
    def read_file(path: str) -> str:
        """ファイルを読み込む"""
        with open(path, 'r') as f:
            return f.read()

    @staticmethod
    def write_file(path: str, content: str) -> str:
        """ファイルに書き込む"""
        with open(path, 'w') as f:
            f.write(content)
        return f"ファイル {path} に書き込みました"

    @staticmethod
    def run_code(code: str) -> str:
        """Pythonコードを実行"""
        import subprocess
        result = subprocess.run(
            ["python", "-c", code],
            capture_output=True,
            text=True,
            timeout=10
        )
        return result.stdout or result.stderr

    @staticmethod
    def call_api(url: str) -> str:
        """外部APIを呼び出す"""
        import requests
        response = requests.get(url)
        return response.text[:1000]

# 実際の検索ツール (Tavily API使用例)
def tavily_search(query: str) -> str:
    import requests
    response = requests.post(
        "https://api.tavily.com/search",
        json={
            "api_key": os.getenv("TAVILY_API_KEY"),
            "query": query,
            "max_results": 3
        }
    )
    results = response.json()
    return "\n".join([r["content"][:200] for r in results.get("results", [])])

class ReflexionAgent(ReActAgent):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.memory = []  # 過去の反省を保存

    def reflect(self, task: str, result: str, success: bool) -> str:
        """結果を振り返り、改善点を抽出"""
        reflection_prompt = f"""
タスク: {task}
結果: {result}
成功: {"はい" if success else "いいえ"}

この試行から学べることは何ですか？次回の改善点を3つ挙げてください。
"""
        response = client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": reflection_prompt}]
        )
        reflection = response.choices[0].message.content
        self.memory.append(reflection)
        return reflection

    def run_with_reflection(self, user_query: str, max_retries: int = 3) -> str:
        """反省を活用してリトライ"""
        for attempt in range(max_retries):
            context = ""
            if self.memory:
                context = "過去の反省:\n" + "\n".join(self.memory[-3:])

            result = self.run(f"{context}\n\n{user_query}")
            success = "エラー" not in result and "わかりません" not in result

            if success:
                return result

            self.reflect(user_query, result, success)

        return result