LLMは膨大な知識を持っていますが、リアルタイム情報の取得や計算、外部システムとの連携はできません。Tool Use（Function Calling）により、LLMが外部ツールを呼び出せるようになり、実用的なアプリケーションの幅が大きく広がります。
本記事では、LLMがツールを使用する仕組みと実装方法を解説します。
本記事の内容
- Tool Use / Function Callingの基本概念
- ツール定義とパラメータスキーマ
- LLMによるツール選択の仕組み
- Pythonでの実装例
前提知識
この記事を読む前に、以下の概念を理解しておくと役立ちます。
- LLMの基本的な仕組み
- JSON Schema
- REST APIの基礎
LLMの限界とTool Use
LLMができないこと
LLMには以下の限界があります。
- リアルタイム情報: 今日の天気、最新ニュースなど
- 正確な計算: 複雑な数値計算、統計処理
- 外部システム連携: データベース操作、API呼び出し
- ファイル操作: 読み書き、作成、削除
Tool Useによる解決
Tool Useにより、LLMが適切なタイミングで外部ツールを呼び出し、その結果を使って回答を生成できます。
$$ \text{Response} = \text{LLM}(\text{Query}, \text{Tool Results}) $$
Tool Useの基本アーキテクチャ
処理フロー
User Query
↓
LLM (ツール呼び出しが必要か判断)
↓
[ツールが必要な場合]
↓
ツール呼び出し (関数名, パラメータ)
↓
ツール実行 → 結果
↓
LLM (結果を使って回答生成)
↓
Final Response
主要コンポーネント
- ツール定義: ツールの名前、説明、パラメータスキーマ
- ツール選択: LLMが適切なツールを選ぶ
- パラメータ抽出: ユーザークエリからパラメータを抽出
- ツール実行: 実際にツールを実行
- 結果統合: ツール結果を使って最終回答を生成
ツール定義
JSON Schemaによる定義
ツールは以下の形式で定義されます。
{
"name": "get_weather",
"description": "指定された都市の現在の天気を取得します",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "天気を取得する都市名（例: 東京、大阪）"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度の単位"
}
},
"required": ["city"]
}
}
パラメータスキーマの要素
| 要素 | 説明 |
|---|---|
| type | データ型（string, number, boolean, array, object） |
| description | パラメータの説明（LLMが参照） |
| enum | 許容される値のリスト |
| required | 必須パラメータのリスト |
Pythonでの実装
ツールの定義と登録
import json
from typing import Callable, Dict, Any, List
from dataclasses import dataclass
@dataclass
class Tool:
"""ツールの定義"""
name: str
description: str
parameters: Dict[str, Any]
function: Callable
class ToolRegistry:
"""ツールの登録と管理"""
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register(self, name: str, description: str,
parameters: Dict[str, Any], function: Callable):
"""ツールを登録"""
self.tools[name] = Tool(
name=name,
description=description,
parameters=parameters,
function=function
)
def get_tool_definitions(self) -> List[Dict]:
"""LLMに渡すツール定義を取得"""
return [
{
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
for tool in self.tools.values()
]
def execute(self, name: str, arguments: Dict[str, Any]) -> Any:
"""ツールを実行"""
if name not in self.tools:
raise ValueError(f"Unknown tool: {name}")
return self.tools[name].function(**arguments)
# ツール関数の実装
def get_weather(city: str, unit: str = "celsius") -> Dict:
"""天気を取得（モック実装）"""
# 実際にはAPIを呼び出す
weather_data = {
"東京": {"temp": 22, "condition": "晴れ"},
"大阪": {"temp": 24, "condition": "曇り"},
"札幌": {"temp": 15, "condition": "雨"},
}
data = weather_data.get(city, {"temp": 20, "condition": "不明"})
if unit == "fahrenheit":
data["temp"] = data["temp"] * 9/5 + 32
return {
"city": city,
"temperature": data["temp"],
"unit": unit,
"condition": data["condition"]
}
def calculate(expression: str) -> Dict:
"""数式を計算"""
try:
# 安全のため eval は使わない（本番環境）
# ここでは簡易実装
result = eval(expression)
return {"expression": expression, "result": result}
except Exception as e:
return {"expression": expression, "error": str(e)}
def search_web(query: str, num_results: int = 3) -> Dict:
"""Web検索（モック実装）"""
# 実際には検索APIを呼び出す
return {
"query": query,
"results": [
{"title": f"検索結果{i+1}: {query}について", "url": f"https://example.com/{i}"}
for i in range(num_results)
]
}
# ツールの登録
registry = ToolRegistry()
registry.register(
name="get_weather",
description="指定された都市の現在の天気を取得します",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "天気を取得する都市名"},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度の単位",
"default": "celsius"
}
},
"required": ["city"]
},
function=get_weather
)
registry.register(
name="calculate",
description="数式を計算して結果を返します",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "計算する数式（例: 2+3*4）"
}
},
"required": ["expression"]
},
function=calculate
)
registry.register(
name="search_web",
description="指定されたクエリでWeb検索を実行します",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "検索クエリ"},
"num_results": {
"type": "integer",
"description": "取得する結果の数",
"default": 3
}
},
"required": ["query"]
},
function=search_web
)
print("登録されたツール:")
for tool_def in registry.get_tool_definitions():
print(f" - {tool_def['name']}: {tool_def['description']}")
LLMとのインタラクション
class ToolUseAgent:
"""Tool Useを行うエージェント"""
def __init__(self, llm_client, tool_registry: ToolRegistry):
self.llm_client = llm_client
self.tool_registry = tool_registry
def _create_system_prompt(self) -> str:
"""システムプロンプトを生成"""
tools_description = json.dumps(
self.tool_registry.get_tool_definitions(),
ensure_ascii=False,
indent=2
)
return f"""あなたは便利なアシスタントです。以下のツールを使用できます。
ツール一覧:
{tools_description}
ツールを使用する場合は、以下のJSON形式で出力してください:
```json
{{"tool": "ツール名", "arguments": {{"パラメータ名": "値"}}}}
ツールが不要な場合は、直接回答してください。”””
def _parse_tool_call(self, response: str) -> tuple:
"""レスポンスからツール呼び出しを解析"""
try:
# JSONブロックを抽出
if "```json" in response:
json_str = response.split("```json")[1].split("```")[0]
elif "{" in response and "}" in response:
start = response.index("{")
end = response.rindex("}") + 1
json_str = response[start:end]
else:
return None, None
data = json.loads(json_str)
return data.get("tool"), data.get("arguments", {})
except:
return None, None
def run(self, user_message: str, max_iterations: int = 5) -> str:
"""ユーザーメッセージを処理"""
system_prompt = self._create_system_prompt()
conversation = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
]
for iteration in range(max_iterations):
# LLMを呼び出し
response = self.llm_client.chat(conversation)
# ツール呼び出しを解析
tool_name, arguments = self._parse_tool_call(response)
if tool_name is None:
# ツール呼び出しなし → 最終回答
return response
# ツールを実行
try:
result = self.tool_registry.execute(tool_name, arguments)
tool_result = json.dumps(result, ensure_ascii=False)
except Exception as e:
tool_result = f"エラー: {str(e)}"
# 結果を会話に追加
conversation.append({"role": "assistant", "content": response})
conversation.append({
"role": "user",
"content": f"ツール実行結果:\n{tool_result}\n\nこの結果を使って質問に答えてください。"
})
return "最大反復回数に達しました。"
使用例（疑似コード）
agent = ToolUseAgent(llm_client, registry)
response = agent.run(“東京の天気を教えてください”)
print(response)
### 並列ツール呼び出し
```python
import asyncio
from concurrent.futures import ThreadPoolExecutor
class ParallelToolExecutor:
"""複数のツールを並列実行"""
def __init__(self, tool_registry: ToolRegistry, max_workers: int = 4):
self.tool_registry = tool_registry
self.executor = ThreadPoolExecutor(max_workers=max_workers)
async def execute_parallel(self, tool_calls: List[Dict]) -> List[Dict]:
"""複数のツール呼び出しを並列実行"""
loop = asyncio.get_event_loop()
async def execute_one(call):
name = call["name"]
arguments = call["arguments"]
try:
result = await loop.run_in_executor(
self.executor,
lambda: self.tool_registry.execute(name, arguments)
)
return {"name": name, "result": result, "success": True}
except Exception as e:
return {"name": name, "error": str(e), "success": False}
tasks = [execute_one(call) for call in tool_calls]
results = await asyncio.gather(*tasks)
return results
# 使用例
async def example_parallel_execution():
executor = ParallelToolExecutor(registry)
tool_calls = [
{"name": "get_weather", "arguments": {"city": "東京"}},
{"name": "get_weather", "arguments": {"city": "大阪"}},
{"name": "calculate", "arguments": {"expression": "123 * 456"}},
]
results = await executor.execute_parallel(tool_calls)
for r in results:
print(f"{r['name']}: {r.get('result', r.get('error'))}")
# asyncio.run(example_parallel_execution())
OpenAI Function Calling形式
API仕様
OpenAI APIでは以下の形式でツールを定義します。
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "指定された都市の天気を取得",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "都市名"
}
},
"required": ["city"]
}
}
}
]
レスポンス形式
ツール呼び出しが必要な場合、モデルは以下を返します。
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_abc123",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "東京"}'
}
}
]
}
OpenAI API対応の実装
import openai
import json
class OpenAIToolAgent:
"""OpenAI API対応のTool Useエージェント"""
def __init__(self, api_key: str, model: str = "gpt-4"):
self.client = openai.OpenAI(api_key=api_key)
self.model = model
self.tools = []
self.functions = {}
def register_tool(self, name: str, description: str,
parameters: Dict, function: Callable):
"""ツールを登録"""
self.tools.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
})
self.functions[name] = function
def run(self, messages: List[Dict]) -> str:
"""メッセージを処理"""
while True:
# API呼び出し
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools if self.tools else None,
tool_choice="auto"
)
assistant_message = response.choices[0].message
# ツール呼び出しがない場合は終了
if not assistant_message.tool_calls:
return assistant_message.content
# ツール呼び出しを処理
messages.append({
"role": "assistant",
"content": assistant_message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in assistant_message.tool_calls
]
})
# 各ツールを実行
for tool_call in assistant_message.tool_calls:
name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
try:
result = self.functions[name](**arguments)
result_str = json.dumps(result, ensure_ascii=False)
except Exception as e:
result_str = json.dumps({"error": str(e)})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result_str
})
# 使用例
# agent = OpenAIToolAgent(api_key="your-api-key")
# agent.register_tool("get_weather", "天気を取得", {...}, get_weather)
# result = agent.run([{"role": "user", "content": "東京の天気は？"}])
ツール選択の最適化
ツール説明の重要性
LLMがツールを正しく選択するには、明確な説明が重要です。
良い例:
"description": "指定された都市の現在の天気（気温、天候、湿度など）を取得します。
リアルタイムの気象情報が必要な場合に使用してください。"
悪い例:
"description": "天気を取得"
パラメータ説明のベストプラクティス
{
"properties": {
"date": {
"type": "string",
"description": "日付（YYYY-MM-DD形式、例: 2024-01-15）。省略時は今日。"
},
"category": {
"type": "string",
"enum": ["electronics", "books", "clothing"],
"description": "商品カテゴリ。electronics: 電子機器、books: 書籍、clothing: 衣類"
}
}
}
エラーハンドリング
ツール実行エラーの処理
class RobustToolExecutor:
"""エラーハンドリング付きツール実行"""
def __init__(self, tool_registry: ToolRegistry, max_retries: int = 3):
self.tool_registry = tool_registry
self.max_retries = max_retries
def execute_with_retry(self, name: str, arguments: Dict) -> Dict:
"""リトライ付きでツールを実行"""
last_error = None
for attempt in range(self.max_retries):
try:
result = self.tool_registry.execute(name, arguments)
return {
"success": True,
"result": result,
"attempts": attempt + 1
}
except Exception as e:
last_error = str(e)
# リトライ前に待機（指数バックオフ）
import time
time.sleep(2 ** attempt)
return {
"success": False,
"error": last_error,
"attempts": self.max_retries
}
def execute_with_fallback(self, name: str, arguments: Dict,
fallback_value: Any = None) -> Any:
"""フォールバック付きでツールを実行"""
result = self.execute_with_retry(name, arguments)
if result["success"]:
return result["result"]
else:
return fallback_value
まとめ
本記事では、Tool Use / Function Callingの仕組みと実装を解説しました。
- 基本概念: LLMが外部ツールを呼び出して能力を拡張
- ツール定義: JSON Schemaでパラメータを明確に定義
- 実行フロー: ツール選択 → パラメータ抽出 → 実行 → 結果統合
- ベストプラクティス: 明確な説明、適切なエラーハンドリング
次のステップとして、以下の記事も参考にしてください。