Python AI Agent ツール使用：2026年ReActパターンでLLMにツールを本当に使いこなさせる
AI与大数据
ツールが「使える」LLMと「使えない」LLM、どれくらいの差があるか？
GPT-4oはコードを書き、分析し、質問に答えられますが、リアルタイムデータベースのクエリ、API呼び出し、ファイルシステム操作はできません——ツールを与えるまでは。しかし、LLMにツールを接続しても、それを使いこなせるわけではありません。モデルは間違ったツールを選び、パラメータを間違え、ループに陥り、存在しないツールを幻覚することさえあります。
ReAct（Reasoning + Acting）パターンは、「思考-行動-観察」ループにより、LLMが推論とツール使用を真に協調できるようにします。2026年、これは本番級AIエージェントを構築するためのコアパラダイムです。
ReActパターンコア概念
| 概念 | 説明 | 比較 |
|---|---|---|
| Reasoning | モデルが現在の状態と次の行動を推論 | 直接ツール呼び出しと比較 |
| Acting | ツール呼び出しを実行し結果を取得 | 純テキスト推論と比較 |
| Observation | ツール結果を観察し推論を更新 | 結果を無視することと比較 |
| Tool Definition | JSON Schemaでツールインタフェースを定義 | 自然言語記述と比較 |
| Function Calling | モデルのネイティブツール呼び出し機能 | 正規表現抽出と比較 |
| Tool Orchestration | マルチツールの調整と依存管理 | 単一ツール呼び出しと比較 |
ReActループフロー：
ユーザー質問 → [Thought] 問題を分析 → [Action] ツールを選択 → [Observation] 結果を取得
→ [Thought] 推論を継続 → [Action] ツールを選択 → [Observation] 結果を取得
→ [Thought] 十分な情報 → [Answer] 最終回答
問題の深掘り：なぜツール呼び出しはこんなに難しいのか？
| 問題 | 原因 | ReActソリューション |
|---|---|---|
| 間違ったツール選択 | モデルがツールの境界を理解していない | Thoughtステップで推論後に選択 |
| パラメータエラー | Schema理解のズレ | 厳格なSchema検証+リトライ |
| ループ呼び出し | 終了条件がない | 最大ステップ数制限+観察判断 |
| 幻覚ツール | 存在しないツールを捏造 | ツールホワイトリスト+強制選択 |
| 結果の誤読 | ツール結果の重要情報を無視 | Observationステップで明示的処理 |
| コンテキスト消失 | 長い会話で早期情報を忘却 | 構造化メモリ管理 |
ステップバイステップ：ゼロからReActエージェントを構築
ステップ1：ツールSchemaの定義
from typing import Optional
from pydantic import BaseModel, Field
import json
import datetime
class ToolParameter(BaseModel):
name: str
type: str
description: str
required: bool = True
enum: Optional[list[str]] = None
class ToolDefinition(BaseModel):
name: str
description: str
parameters: list[ToolParameter]
def to_openai_format(self) -> dict:
properties = {}
required = []
for param in self.parameters:
prop = {"type": param.type, "description": param.description}
if param.enum:
prop["enum"] = param.enum
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {"type": "object", "properties": properties, "required": required},
},
}
class WeatherInput(BaseModel):
city: str = Field(description="都市名")
unit: str = Field(default="celsius", description="温度単位")
class DatabaseQueryInput(BaseModel):
sql: str = Field(description="SQLクエリ、SELECTのみ")
database: str = Field(default="production", description="データベース名")
class CalculatorInput(BaseModel):
expression: str = Field(description="数式")
precision: int = Field(default=2, description="小数精度")
ステップ2：ツールエグゼキュータの実装
import sqlite3
import re
from typing import Callable
class ToolRegistry:
def __init__(self):
self._tools: dict[str, dict] = {}
def register(self, name: str, description: str, parameters: list[ToolParameter], executor: Callable):
self._tools[name] = {
"definition": ToolDefinition(name=name, description=description, parameters=parameters),
"executor": executor,
}
def get_openai_tools(self) -> list[dict]:
return [tool["definition"].to_openai_format() for tool in self._tools.values()]
def execute(self, name: str, arguments: dict) -> str:
if name not in self._tools:
return json.dumps({"error": f"Tool '{name}' not found", "available": list(self._tools.keys())})
try:
result = self._tools[name]["executor"](**arguments)
return json.dumps(result, ensure_ascii=False, default=str)
except Exception as e:
return json.dumps({"error": str(e), "tool": name})
def execute_weather(city: str, unit: str = "celsius") -> dict:
mock_data = {
"東京": {"temp": 26, "condition": "晴れ", "humidity": 55},
"大阪": {"temp": 28, "condition": "曇り", "humidity": 65},
}
data = mock_data.get(city, {"temp": 25, "condition": "不明", "humidity": 50})
return {"city": city, **data, "unit": unit, "updated_at": datetime.datetime.now().isoformat()}
def execute_database_query(sql: str, database: str = "production") -> dict:
forbidden = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
if any(kw in sql.upper() for kw in forbidden):
return {"error": "SELECTクエリのみサポート", "sql": sql}
conn = sqlite3.connect(":memory:")
cursor = conn.cursor()
cursor.execute("CREATE TABLE IF NOT EXISTS orders (id INTEGER, product TEXT, amount REAL, status TEXT)")
cursor.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", [
(1, "キーボード", 599.0, "completed"),
(2, "モニター", 2999.0, "pending"),
(3, "マウス", 199.0, "completed"),
])
conn.commit()
try:
cursor.execute(sql)
columns = [desc[0] for desc in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
return {"results": rows, "row_count": len(rows), "database": database}
except Exception as e:
return {"error": str(e), "sql": sql}
finally:
conn.close()
def execute_calculator(expression: str, precision: int = 2) -> dict:
safe_chars = re.sub(r'[^0-9+\-*/().%\s]', '', expression)
if safe_chars != expression.strip():
return {"error": "安全でない文字が含まれています", "expression": expression}
try:
result = eval(safe_chars, {"__builtins__": {}}, {"abs": abs, "round": round, "pow": pow})
return {"result": round(result, precision), "expression": expression}
except Exception as e:
return {"error": str(e), "expression": expression}
registry = ToolRegistry()
registry.register("get_weather", "指定した都市の天気情報を取得", [
ToolParameter(name="city", type="string", description="都市名"),
ToolParameter(name="unit", type="string", description="温度単位", required=False),
], execute_weather)
registry.register("query_database", "SQLクエリを実行", [
ToolParameter(name="sql", type="string", description="SQLクエリ"),
ToolParameter(name="database", type="string", description="データベース名", required=False),
], execute_database_query)
registry.register("calculate", "数式を計算", [
ToolParameter(name="expression", type="string", description="数式"),
ToolParameter(name="precision", type="integer", description="小数精度", required=False),
], execute_calculator)
ステップ3：ReActエージェントコアの実装
from openai import OpenAI
import os
class ReActAgent:
def __init__(self, tool_registry: ToolRegistry, model: str = "gpt-4o", max_steps: int = 8):
self.registry = tool_registry
self.model = model
self.max_steps = max_steps
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.conversation_history: list[dict] = []
def _build_system_prompt(self) -> str:
return """あなたはツールを使用してユーザーの質問に答えるインテリジェントアシスタントです。
ReActパターンに従ってください：
1. Thought: 現在の状況を分析し、次の行動を決定
2. Action: 適切なツールを選択して呼び出し
3. Observation: ツールの結果を観察
4. 十分な情報が得られるまで繰り返し
5. Answer: 収集した情報に基づいて最終回答
重要ルール：
- 1ステップにつき1つのツールのみ呼び出し
- ツール結果を慎重に読んでから次のステップを決定
- エラーが返されたらパラメータを修正してリトライ
- 存在しないツールを捏造しない
- 情報が十分な場合は直接回答"""
def run(self, user_message: str) -> str:
self.conversation_history = [
{"role": "system", "content": self._build_system_prompt()},
{"role": "user", "content": user_message},
]
tools = self.registry.get_openai_tools()
for step in range(self.max_steps):
response = self.client.chat.completions.create(
model=self.model,
messages=self.conversation_history,
tools=tools,
tool_choice="auto",
temperature=0.1,
)
message = response.choices[0].message
self.conversation_history.append(message.to_dict())
if message.content:
print(f"Thought: {message.content}")
if not message.tool_calls:
return message.content or "回答を生成できません"
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f"Action: {func_name}({json.dumps(func_args, ensure_ascii=False)})")
result = self.registry.execute(func_name, func_args)
print(f"Observation: {result[:200]}...")
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result,
})
return "最大ステップ数に達しました。推論を完了できませんでした。"
if __name__ == "__main__":
agent = ReActAgent(registry)
answer = agent.run("東京の天気はどうですか？完了した注文の合計金額も調べてください。")
print(f"\n最終回答: {answer}")
完全コード：エラー回復付きマルチツールエージェント
from openai import OpenAI
from pydantic import BaseModel
from typing import Optional
import json
import os
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AgentConfig(BaseModel):
model: str = "gpt-4o"
max_steps: int = 10
retry_attempts: int = 2
retry_delay: float = 1.0
temperature: float = 0.1
class AgentState(BaseModel):
step_count: int = 0
tool_calls_count: int = 0
errors: list[str] = []
tool_history: list[dict] = []
class ProductionAgent:
def __init__(self, tool_registry: ToolRegistry, config: AgentConfig | None = None):
self.registry = tool_registry
self.config = config or AgentConfig()
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.state = AgentState()
def _execute_tool_with_retry(self, name: str, arguments: dict) -> str:
for attempt in range(self.config.retry_attempts):
try:
result = self.registry.execute(name, arguments)
parsed = json.loads(result)
if "error" in parsed and attempt < self.config.retry_attempts - 1:
logger.warning(f"Tool '{name}' error (attempt {attempt + 1})")
time.sleep(self.config.retry_delay)
continue
return result
except Exception as e:
if attempt < self.config.retry_attempts - 1:
time.sleep(self.config.retry_delay)
else:
return json.dumps({"error": f"Failed after {self.config.retry_attempts} attempts: {e}"})
return json.dumps({"error": "Unexpected retry exit"})
def _validate_tool_call(self, tool_call) -> Optional[str]:
if tool_call.function.name not in self.registry._tools:
return json.dumps({"error": f"ツール '{tool_call.function.name}' は存在しません", "available": list(self.registry._tools.keys())})
try:
json.loads(tool_call.function.arguments)
except json.JSONDecodeError as e:
return json.dumps({"error": f"引数JSONパース失敗: {e}"})
return None
def run(self, user_message: str, context: str = "") -> dict:
start_time = time.time()
self.state = AgentState()
messages = [
{"role": "system", "content": f"あなたはプロのAIアシスタントです。\n{context}\nReActパターンで作業: Thought → Action → Observation → Answer"},
{"role": "user", "content": user_message},
]
tools = self.registry.get_openai_tools()
for step in range(self.config.max_steps):
self.state.step_count = step + 1
try:
response = self.client.chat.completions.create(
model=self.config.model, messages=messages, tools=tools,
tool_choice="auto", temperature=self.config.temperature,
)
except Exception as e:
self.state.errors.append(f"API呼び出し失敗: {e}")
break
message = response.choices[0].message
messages.append(message.to_dict())
if not message.tool_calls:
break
for tool_call in message.tool_calls:
self.state.tool_calls_count += 1
validation_error = self._validate_tool_call(tool_call)
if validation_error:
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": validation_error})
continue
func_args = json.loads(tool_call.function.arguments)
self.state.tool_history.append({"tool": tool_call.function.name, "args": func_args})
result = self._execute_tool_with_retry(tool_call.function.name, func_args)
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
return {
"answer": messages[-1].get("content", ""),
"steps": self.state.step_count,
"tool_calls": self.state.tool_calls_count,
"errors": self.state.errors,
"tool_history": self.state.tool_history,
"elapsed_seconds": round(time.time() - start_time, 2),
}
よくある落とし穴ガイド
落とし穴1：モデルが存在しないツールを幻覚する
解決策：ツールホワイトリスト検証、利用可能ツールリストを返却、システムプロンプトにツール名を明記。
落とし穴2：パラメータ型の不一致
解決策：Schemaでenumを使用、Pydanticモデルでパラメータを検証。
落とし穴3：エージェントがループに陥る
解決策：max_stepsハードリミット（8-10推奨）、ツール履歴で重複呼び出しを検出。
落とし穴4：ツール結果が長すぎる
解決策：エグゼキュータで返却行数を制限（最大20行）、長い結果は要約して返却。
落とし穴5：同時ツール呼び出しの順序問題
解決策：1ステップ1ツールに制限、システムプロンプトで「段階的に実行」を明記。
エラートラブルシューティング
| # | エラーメッセージ | 原因 | 解決方法 |
|---|---|---|---|
| 1 | Invalid function_call: function name not found |
未定義ツールの呼び出し | ホワイトリスト検証を追加 |
| 2 | JSON decode error in function arguments |
無効なJSON引数 | JSONパースエラーハンドリング |
| 3 | Rate limit exceeded: 429 |
APIレート制限 | 指数バックオフリトライ |
| 4 | Context length exceeded |
履歴+結果が長すぎる | 履歴の切り詰めまたは圧縮 |
| 5 | Tool execution timeout |
ツール実行タイムアウト | タイムアウト制限を追加 |
| 6 | Function argument missing required parameter |
必須パラメータ不足 | Schemaでrequiredを設定 |
| 7 | Circular tool call detected |
ツールループ呼び出し | max_stepsと重複検出 |
| 8 | Model refused to use tools |
ツール使用を拒否 | system promptとtool_choiceを確認 |
| 9 | Tool returned unexpected format |
予期しない返却形式 | JSON形式を統一 |
| 10 | Token usage exceeded budget |
トークン予算超過 | トークンカウントと予算制御 |
高度な最適化
1. メモリ管理
class ConversationMemory:
def __init__(self, max_messages: int = 20, summary_threshold: int = 15):
self.messages: list[dict] = []
self.max_messages = max_messages
self.summary_threshold = summary_threshold
self.summary: str = ""
def add(self, message: dict):
self.messages.append(message)
if len(self.messages) > self.summary_threshold:
self._compress()
def _compress(self):
old = self.messages[:len(self.messages) - 5]
self.summary += f"\n[要約] {len(old)}メッセージを圧縮"
self.messages = self.messages[len(self.messages) - 5:]
def get_messages(self) -> list[dict]:
result = []
if self.summary:
result.append({"role": "system", "content": f"過去の会話要約: {self.summary}"})
result.extend(self.messages)
return result[-self.max_messages:]
2. ツールオーケストレーションDSL
class ToolPipeline:
def __init__(self, steps: list[dict]):
self.steps = steps
def execute(self, registry: ToolRegistry, initial_input: dict) -> dict:
result = initial_input
for step in self.steps:
args = {}
for key, value in step.get("args", {}).items():
if isinstance(value, str) and value.startswith("$"):
args[key] = result.get(value[1:])
else:
args[key] = value
output = json.loads(registry.execute(step["tool"], args))
result[step.get("output_as", step["tool"])] = output
return result
3. トークン予算制御
class TokenBudget:
def __init__(self, max_total: int = 100000, max_per_step: int = 8000):
self.max_total = max_total
self.max_per_step = max_per_step
self.used = 0
def can_proceed(self, estimated: int = 4000) -> bool:
return (self.used + estimated) < self.max_total
def consume(self, tokens: int):
self.used += tokens
def remaining(self) -> int:
return max(0, self.max_total - self.used)
比較分析
| 次元 | ReActエージェント | Plan-and-Execute | Reflexion | AutoGPT |
|---|---|---|---|---|
| 推論方式 | 段階的推論+行動 | 先に計画してから実行 | 自己反省修正 | 自律目標駆動 |
| ツール呼び出し | 1ステップ1つ | 計画に従いバッチ | 反省付きリトライ | 自由呼び出し |
| エラー回復 | 観察後に調整 | 再計画 | 自己批判修正 | 試行錯誤 |
| 適用シナリオ | マルチステップ推論 | 複雑タスク分解 | 高品質出力が必要 | オープンなタスク |
| トークン消費 | 中 | やや高 | 高 | 非常に高い |
| 制御性 | 高 | 中 | 中 | 低 |
| 実装複雑度 | 低 | 中 | 高 | 高 |
| 安定性 | 高 | 中 | 中 | 低 |
まとめ：ReActパターンは「思考-行動-観察」ループにより、LLMを「受動的回答」から「能動的推論+ツール使用」へと進化させます。2026年に本番級AIエージェントを構築する鍵は、厳格なツールSchema定義、信頼性の高いリトライとエラー回復メカニズム、適切なステップ数とトークン予算制御です。単一ツールエージェントから始め、段階的にマルチツールオーケストレーション、メモリ管理、自己反省を導入するのが、AIエージェントを本番投入する最良の道です。良いエージェントとは、モデルを自由に走らせることではなく、構造化された制約を通じてモデルが正しい意思決定を行うよう導くことです。
オンラインツールおすすめ
- JSONフォーマッター：/ja/json/format — Function CallingのSchemaとツールレスポンスのフォーマット
- Base64エンコード/デコード：/ja/encode/base64 — APIキーとエージェント設定のエンコード/デコード
- Curl to Code：/ja/dev/curl-to-code — AI APIデバッグcurlをPythonコードに変換
ブラウザローカルツールを無料で試す →
#Python#AI Agent#Tool Use#Function Calling#ReAct#工具编排#LangChain#智能体