「考えながら調べる」AIエージェントの作り方 — Claude API 拡張思考 × Tool Use 本番実装ガイド

import anthropic
import json
 
client = anthropic.Anthropic()
 
# シンプルなツール定義（計算ツール）
tools = [
    {
        "name": "calculate",
        "description": "数値計算を実行します。複雑な計算式を渡すと結果を返します。",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "評価する数式（例: '2 ** 32 + 100'）"
                }
            },
            "required": ["expression"]
        }
    }
]
 
def handle_tool_call(tool_name: str, tool_input: dict) -> str:
    if tool_name == "calculate":
        try:
            result = eval(tool_input["expression"], {"__builtins__": {}}, {})
            return str(result)
        except Exception as e:
            return f"計算エラー: {e}"
    return "不明なツール"
 
def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]
    max_iterations = 5
 
    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=16000,
            thinking={
                "type": "enabled",
                "budget_tokens": 8000
            },
            tools=tools,
            tool_choice={"type": "auto"},
            messages=messages
        )
 
        print(f"[イテレーション {iteration + 1}] stop_reason: {response.stop_reason}")
 
        # 思考ブロックを表示（デバッグ用）
        for block in response.content:
            if block.type == "thinking":
                print(f"  [思考] {block.thinking[:200]}...")
 
        # ツール呼び出しが要求されていない場合 → 最終回答
        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    return block.text
            return ""
 
        # ツール呼び出しが要求された場合 → 実行して結果を返す
        if response.stop_reason == "tool_use":
            # アシスタントの応答（thinking + tool_useブロックを含む）を履歴に追加
            messages.append({"role": "assistant", "content": response.content})
 
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"  [ツール呼び出し] {block.name}({block.input})")
                    result = handle_tool_call(block.name, block.input)
                    print(f"  [ツール結果] {result}")
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
 
            messages.append({"role": "user", "content": tool_results})
 
    return "最大イテレーション数に達しました"
 
# 動作確認
result = run_agent(
    "2の32乗に100を足した値と、その値の平方根を計算してください。計算の過程も説明してください。"
)
print("\n=== 最終回答 ===")
print(result)

import anthropic
import json
from pathlib import Path
 
client = anthropic.Anthropic()
 
# ツール定義
TOOLS = [
    {
        "name": "web_search",
        "description": (
            "指定したクエリでWeb検索を実行し、関連する情報を返します。"
            "最新情報の調査や、知識に不確かな点がある場合に使用します。"
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "検索クエリ文字列"
                },
                "max_results": {
                    "type": "integer",
                    "description": "取得する最大件数（デフォルト5）",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "read_file",
        "description": "ローカルファイルの内容を読み込みます。テキストファイルのみ対応。",
        "input_schema": {
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "読み込むファイルのパス"
                }
            },
            "required": ["file_path"]
        }
    },
    {
        "name": "calculate",
        "description": "数値計算を実行します。Pythonの数式をそのまま渡せます。",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "評価する数式（例: '(100 * 1.1) ** 2'）"
                }
            },
            "required": ["expression"]
        }
    }
]
 
def execute_tool(tool_name: str, tool_input: dict) -> str:
    """ツールを実行し、結果を文字列で返す。エラーも文字列として返す（例外は投げない）"""
    try:
        if tool_name == "web_search":
            # 実際の実装では SerpAPI / Tavily API を使用する
            # import requests
            # resp = requests.get(
            #     "https://serpapi.com/search",
            #     params={"q": tool_input["query"], "api_key": os.getenv("SERPAPI_KEY"),
            #             "num": tool_input.get("max_results", 5)}
            # )
            # results = resp.json().get("organic_results", [])
            # return "\n".join(f"- {r['title']}: {r['snippet']}" for r in results)
            return f"[モック] クエリ: {tool_input['query']} の検索結果（実装時はSerpAPI等に置き換えてください）"
 
        elif tool_name == "read_file":
            path = Path(tool_input["file_path"])
            if not path.exists():
                return f"エラー: ファイルが見つかりません: {path}"
            if path.stat().st_size > 100_000:
                return f"エラー: ファイルが大きすぎます（100KB超）"
            return path.read_text(encoding="utf-8")
 
        elif tool_name == "calculate":
            import math
            safe_builtins = {"abs": abs, "round": round, "min": min, "max": max,
                             "sum": sum, "len": len, "int": int, "float": float}
            safe_env = {**safe_builtins, **vars(math)}
            result = eval(tool_input["expression"], {"__builtins__": {}}, safe_env)
            return str(result)
 
        else:
            return f"エラー: 不明なツール: {tool_name}"
 
    except Exception as e:
        return f"ツール実行エラー ({tool_name}): {type(e).__name__}: {e}"
 
class ResearchAgent:
    def __init__(
        self,
        model: str = "claude-sonnet-4-6",
        budget_tokens: int = 10000,
        max_tokens: int = 16000,
        max_iterations: int = 8,
        verbose: bool = True
    ):
        self.model = model
        self.budget_tokens = budget_tokens
        self.max_tokens = max_tokens
        self.max_iterations = max_iterations
        self.verbose = verbose
 
    def run(self, task: str, context: str = "") -> dict:
        """
        タスクを実行し、最終回答と使用状況を返す。
 
        Returns:
            answer (str): 最終回答
            thinking_steps (list): 各イテレーションの思考
            tool_calls (list): 実行したツール呼び出しの記録
            total_usage (dict): トークン使用量
        """
        system_prompt = (
            "あなたは優秀なリサーチエージェントです。\n"
            "ユーザーのタスクを達成するために、利用可能なツールを効果的に使ってください。\n"
            "情報が不確かな場合は必ず web_search で確認し、計算が必要な場合は calculate ツールを使ってください。"
        )
        if context:
            system_prompt += f"\n\n参考情報:\n{context}"
 
        messages = [{"role": "user", "content": task}]
        thinking_steps = []
        tool_calls = []
        total_input_tokens = 0
        total_output_tokens = 0
 
        for iteration in range(self.max_iterations):
            if self.verbose:
                print(f"\n{'='*50}")
                print(f"イテレーション {iteration + 1}/{self.max_iterations}")
 
            response = client.messages.create(
                model=self.model,
                max_tokens=self.max_tokens,
                thinking={"type": "enabled", "budget_tokens": self.budget_tokens},
                tools=TOOLS,
                tool_choice={"type": "auto"},
                system=system_prompt,
                messages=messages
            )
 
            total_input_tokens += response.usage.input_tokens
            total_output_tokens += response.usage.output_tokens
 
            for block in response.content:
                if block.type == "thinking":
                    thinking_steps.append({
                        "iteration": iteration + 1,
                        "thinking": block.thinking
                    })
                    if self.verbose:
                        print(f"[思考] {block.thinking[:300].replace(chr(10), ' ')}...")
 
            if response.stop_reason == "end_turn":
                final_answer = "".join(
                    block.text for block in response.content if block.type == "text"
                )
                if self.verbose:
                    print(f"\n[完了] {iteration + 1}回のイテレーションで終了")
                    print(f"[使用量] 入力: {total_input_tokens} / 出力: {total_output_tokens} トークン")
                return {
                    "answer": final_answer,
                    "thinking_steps": thinking_steps,
                    "tool_calls": tool_calls,
                    "total_usage": {
                        "input_tokens": total_input_tokens,
                        "output_tokens": total_output_tokens
                    }
                }
 
            if response.stop_reason == "tool_use":
                # thinkingブロックを含むresponse.contentをそのまま追加（重要）
                messages.append({"role": "assistant", "content": response.content})
 
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        if self.verbose:
                            print(f"[ツール] {block.name}({json.dumps(block.input, ensure_ascii=False)})")
 
                        result = execute_tool(block.name, block.input)
 
                        # 長すぎる結果はトリム（コンテキスト超過防止）
                        if len(result) > 5000:
                            result = result[:5000] + "\n\n[省略: 先頭5000文字のみ]"
 
                        if self.verbose:
                            print(f"  → {result[:150].replace(chr(10), ' ')}...")
 
                        tool_calls.append({
                            "iteration": iteration + 1,
                            "tool": block.name,
                            "input": block.input,
                            "result_preview": result[:200]
                        })
 
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result
                        })
 
                messages.append({"role": "user", "content": tool_results})
 
        return {
            "answer": "最大イテレーション数に達しました。タスクを分割するか budget_tokens を増やしてください。",
            "thinking_steps": thinking_steps,
            "tool_calls": tool_calls,
            "total_usage": {"input_tokens": total_input_tokens, "output_tokens": total_output_tokens}
        }
 
# 実行例
if __name__ == "__main__":
    agent = ResearchAgent(
        model="claude-sonnet-4-6",
        budget_tokens=10000,
        max_tokens=16000,
        max_iterations=8,
        verbose=True
    )
 
    result = agent.run(
        task=(
            "Anthropic が2026年に発表した主要なモデルアップデートについて調査し、"
            "claude-sonnet-4-6 と claude-haiku-4-5 の価格を比較してください。"
            "1000万トークン入力・100万トークン出力という条件でコスト差を計算してください。"
        )
    )
 
    print("\n" + "="*50)
    print("最終回答:")
    print(result["answer"])

def compress_messages(messages: list) -> list:
    """最新ターン以外の thinking ブロックを除去してコンテキストを圧縮する"""
    if len(messages) < 4:
        return messages
 
    compressed = []
    for i, msg in enumerate(messages):
        if msg["role"] == "assistant" and i < len(messages) - 2:
            # 最新以外のアシスタントターンから thinking を除去
            content_without_thinking = [
                block for block in msg["content"]
                if not (hasattr(block, "type") and block.type == "thinking")
            ]
            compressed.append({"role": "assistant", "content": content_without_thinking})
        else:
            compressed.append(msg)
    return compressed

# ストリーミング時の正しい処理パターン
with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=16000,
    thinking={"type": "enabled", "budget_tokens": 8000},
    tools=tools,
    messages=messages
) as stream:
    current_block_type = None
 
    for event in stream:
        if hasattr(event, "type"):
            if event.type == "content_block_start":
                if hasattr(event.content_block, "type"):
                    current_block_type = event.content_block.type
 
            elif event.type == "content_block_delta":
                if current_block_type == "text":
                    if hasattr(event.delta, "text"):
                        print(event.delta.text, end="", flush=True)
                # thinking_delta は無視（または別途収集）

def code_review_agent(file_path: str, review_criteria: list) -> dict:
    """
    指定ファイルを読み込み、Extended Thinking で深くレビューするエージェント。
    review_criteria の例:
        ["セキュリティ脆弱性", "パフォーマンス問題", "可読性・保守性"]
    """
    agent = ResearchAgent(
        model="claude-sonnet-4-6",
        budget_tokens=12000,  # コードレビューは思考量を多めに
        max_tokens=18000,
        max_iterations=3,     # ファイル読み込み→レビュー→まとめの3ステップ
        verbose=False
    )
 
    criteria_text = "\n".join(f"- {c}" for c in review_criteria)
 
    task = (
        f"以下のファイルをコードレビューしてください:\n"
        f"ファイルパス: {file_path}\n\n"
        f"レビュー観点:\n{criteria_text}\n\n"
        f"まず read_file ツールでファイルを読み込み、"
        f"それぞれの観点について具体的な問題箇所（行番号とコード）を示しながら"
        f"詳細なフィードバックを提供してください。"
    )
 
    return agent.run(task)
 
# 使用例
result = code_review_agent(
    file_path="./src/api/checkout.py",
    review_criteria=[
        "SQLインジェクション・認証バイパスなどのセキュリティ脆弱性",
        "N+1クエリ問題・不必要なループなどのパフォーマンス問題",
        "エラーハンドリングの漏れ・例外の握りつぶし"
    ]
)
print(result["answer"])