Skip to content
Published on

Tool Calling実践ガイド：AIが外部世界を操る方法と落とし穴

🇰🇷KO🇺🇸EN🇯🇵JA
Split View
Authors

LLM単体ではテキストを変換するだけです。ツールが加わると、実際に何かができるようになります。Tool Callingは受動的なテキスト生成器を能動的なエージェントに変えるメカニズムです。

この記事ではTool Callingが実際にどう動作するか、正しい実装方法、そして本番環境で必ずぶつかる具体的な間違いを解説します。

Tool CallingがAgentの核心である理由

エージェントの能力は使えるツールの範囲と同じです：

  • 検索ツール → リアルタイム情報へのアクセス
  • 計算機ツール → 数学的な正確さの保証
  • コード実行ツール → 実際のコードの作成と実行
  • APIツール → 外部サービスとの統合
  • データベースツール → データの読み書き

ツールなしでは、LLMは学習データの中からしか答えられません。ツールがあれば、今日の株価を調べ、メールを送り、コードを実行できます。

OpenAI Function Calling完全実装

OpenAIのfunction callingは事実上の標準APIフォーマットになっています。

ツールの定義

import openai
import json

client = openai.OpenAI(api_key="your-api-key")

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": (
                "特定の都市の現在の天気を取得します。"
                "ユーザーが現在の天気状況を尋ねるときに使用してください。"
                "天気予報や過去の天気データには使用しないでください。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "英語の都市名。例：'Tokyo'、'Seoul'、'New York'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度単位。デフォルト：celsius"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": (
                "最新情報を取得するためにウェブを検索します。"
                "最近のニュース、イベント、モデルの学習カットオフ以降に"
                "変わった可能性のある情報に使用してください。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "検索クエリ"
                    },
                    "num_results": {
                        "type": "integer",
                        "description": "返す結果の数（1-10）",
                        "default": 3
                    }
                },
                "required": ["query"]
            }
        }
    }
]

実際の関数の実装

def get_weather(city: str, unit: str = "celsius") -> dict:
    response = weather_api.get(city=city, unit=unit)
    return {
        "city": city,
        "temperature": response.temp,
        "unit": unit,
        "condition": response.condition,
        "humidity": response.humidity
    }

def search_web(query: str, num_results: int = 3) -> list:
    results = search_api.search(query, count=num_results)
    return [
        {"title": r.title, "snippet": r.snippet, "url": r.url}
        for r in results
    ]

available_tools = {
    "get_weather": get_weather,
    "search_web": search_web
}

完全なTool Callループ

def run_agent(user_message: str, max_iterations: int = 10) -> str:
    messages = [{"role": "user", "content": user_message}]

    for iteration in range(max_iterations):
        response = client.chat.completions.create(
            model="gpt-4",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        assistant_message = response.choices[0].message
        messages.append(assistant_message)

        # ツール呼び出しなし → 最終回答
        if not assistant_message.tool_calls:
            return assistant_message.content

        # 各ツール呼び出しを実行
        for tool_call in assistant_message.tool_calls:
            function_name = tool_call.function.name
            function_args = json.loads(tool_call.function.arguments)

            print(f"呼び出し中: {function_name}({function_args})")

            if function_name in available_tools:
                try:
                    result = available_tools[function_name](**function_args)
                    tool_result = json.dumps(result, ensure_ascii=False)
                except Exception as e:
                    tool_result = f"エラー: {str(e)}。別のアプローチを試してください。"
            else:
                tool_result = f"不明なツール: {function_name}"

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": tool_result
            })

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

result = run_agent("東京の天気は？今日のAIニュースも調べて")
print(result)

Anthropic Claude Tool Use

ClaudeはOpenAIと異なるフォーマットを使いますが、コンセプトは同じです：

import anthropic
import json

client = anthropic.Anthropic(api_key="your-api-key")

tools = [
    {
        "name": "get_weather",
        "description": "都市の現在の天気を取得します",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "都市名"}
            },
            "required": ["city"]
        }
    }
]

def run_claude_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            return " ".join(
                block.text for block in response.content
                if hasattr(block, "text")
            )

        # tool_useブロックの処理
        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                try:
                    result = available_tools[block.name](**block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })
                except Exception as e:
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": f"Error: {str(e)}",
                        "is_error": True
                    })

        messages.append({"role": "user", "content": tool_results})

よくある5つの間違いと解決法（現場経験より）

このセクションがこの記事で最も価値があります。本番環境で実際に直面する問題です。

間違い1：ツールの説明が曖昧

LLMはdescriptionを使っていつどうやってツールを呼ぶかを決めます。曖昧な説明は誤った呼び出しを招きます。

# 悪い例：曖昧すぎる
{
    "name": "query",
    "description": "データをクエリする"
}

# 良い例：いつ使うか、何を返すか、何に使わないかを明示
{
    "name": "query_customer_orders",
    "description": (
        "特定の顧客の注文履歴を取得します。customer_idが必要です。"
        "返り値：注文ID、日付、金額、ステータスのリスト。"
        "顧客プロファイル情報（名前、メール）にはget_customer_infoを使用してください。"
    )
}

間違い2：エラーハンドリングなし

未処理の例外はエージェントループ全体を殺します。

# 悪い例：エラーでエージェントがクラッシュ
result = database.query(sql)

# 良い例：エラーをツール結果として返す → LLMが別のアプローチを試みられる
try:
    result = database.query(sql)
    return json.dumps(result)
except DatabaseError as e:
    return f"データベースエラー: {str(e)}。クエリに構文エラーがある可能性があります。"
except TimeoutError:
    return "クエリがタイムアウトしました。より具体的なフィルターを追加するか、クエリを簡略化してください。"
except Exception as e:
    return f"予期しないエラー: {str(e)}。別のアプローチを試してください。"

間違い3：ループ検出なし

エージェントが同じ引数で同じツールを繰り返し呼び続けることがあります。

MAX_ITERATIONS = 15
call_count = {}

for iteration in range(MAX_ITERATIONS):
    # ...
    for tool_call in tool_calls:
        name = tool_call.function.name
        args_str = tool_call.function.arguments

        call_key = f"{name}:{args_str}"
        call_count[call_key] = call_count.get(call_key, 0) + 1

        if call_count[call_key] > 3:
            return "エラー：同じツール呼び出しが繰り返されすぎです。ループを中断します。"

間違い4：一度に定義するツールが多すぎる

ツールが多すぎるとLLMの精度が下がります。10個を超える場合はダイナミックなツール選択を検討してください。

def select_relevant_tools(user_query: str, all_tools: list, max_tools: int = 8) -> list:
    """ユーザーのクエリに関連するツールのみを選択"""
    if len(all_tools) <= max_tools:
        return all_tools

    query_words = set(user_query.lower().split())
    scored_tools = []

    for tool in all_tools:
        desc_words = set(tool["function"]["description"].lower().split())
        overlap = len(query_words & desc_words)
        scored_tools.append((overlap, tool))

    scored_tools.sort(key=lambda x: x[0], reverse=True)
    return [t for _, t in scored_tools[:max_tools]]

間違い5：不可逆な操作に確認なし

削除、支払い、メール送信などには必ずHuman-in-the-loopを追加してください。

HIGH_RISK_TOOLS = {"delete_record", "send_email", "process_payment", "deploy_code"}

def execute_with_approval(tool_name: str, args: dict) -> str:
    if tool_name in HIGH_RISK_TOOLS:
        # 実際のアプリではUIモーダルやSlack通知などで承認を取得
        print(f"高リスク操作: {tool_name}({args})")
        approval = input("この操作を承認しますか？（yes/no）: ")
        if approval.lower() != "yes":
            return "ユーザーによってキャンセルされました。"

    return available_tools[tool_name](**args)

Parallel Tool Calls（並列実行）

最新のLLMは1回のレスポンスで複数のツール呼び出しをリクエストできます。活用しないと損です — エージェントの速度が大幅に向上します。

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

async def execute_tool_calls_parallel(tool_calls: list) -> list:
    """複数のツール呼び出しを並列実行"""

    async def execute_single(tool_call):
        name = tool_call.function.name
        args = json.loads(tool_call.function.arguments)

        try:
            loop = asyncio.get_event_loop()
            with ThreadPoolExecutor() as pool:
                result = await loop.run_in_executor(
                    pool,
                    lambda: available_tools[name](**args)
                )
            return {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            }
        except Exception as e:
            return {
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": f"エラー: {str(e)}"
            }

    tasks = [execute_single(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks)
    return results

# 例：「東京、ソウル、ニューヨークの天気を教えて」
# LLMが3つのget_weather呼び出しを同時に返す
# 順次実行：3回のAPI呼び出し × 1秒 = 3秒
# 並列実行：3回のAPI呼び出し同時 = 約1秒

ツール設計の原則

良く設計されたツールと悪く設計されたツールの差は、エージェントのパフォーマンスに直結します。

原則1：単一責任

1つのツールは1つのことだけを行います。

# 悪い例：全てを1つの関数でこなす
def manage_customer(action: str, customer_id: int, **kwargs): ...

# 良い例：各機能を分離
def get_customer(customer_id: int) -> dict: ...
def update_customer(customer_id: int, name: str = None, email: str = None) -> dict: ...
def delete_customer(customer_id: int) -> dict: ...

原則2：読み取りと書き込みを分離

読み取りツールは副作用なし。書き込みツールは状態を変更します。明確に分離してください。

# 読み取りツール：安全、冪等、何度呼び出しても問題なし
def get_user_balance(user_id: int) -> float: ...
def search_products(query: str) -> list: ...

# 書き込みツール：不可逆、注意が必要
def transfer_money(from_id: int, to_id: int, amount: float) -> dict: ...
def delete_order(order_id: int) -> dict: ...

原則3：構造化データを返す

自然言語の文章ではなくJSONを返してください。LLMは構造化データをはるかに効率よく推論できます。

# 悪い例：LLMがパースしなければならない
def get_user_info(user_id: int) -> str:
    return f"John Doe、30歳、メール：john@example.com、2022年参加"

# 良い例：LLMが正確に推論できる構造化データ
def get_user_info(user_id: int) -> dict:
    return {
        "id": user_id,
        "name": "John Doe",
        "age": 30,
        "email": "john@example.com",
        "joined_at": "2022-01-15"
    }

原則4：エラー情報を結果に含める

def safe_tool_wrapper(func):
    """一貫したエラーハンドリングでツール関数をラップ"""
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            return {"success": True, "data": result}
        except ValueError as e:
            return {"success": False, "error": "invalid_input", "message": str(e)}
        except PermissionError as e:
            return {"success": False, "error": "permission_denied", "message": str(e)}
        except Exception as e:
            return {"success": False, "error": "unexpected", "message": str(e)}
    return wrapper

@safe_tool_wrapper
def create_order(product_id: int, quantity: int, user_id: int) -> dict:
    ...

本番環境ミドルウェア

本番環境で必要なセーフガードを全てまとめたミドルウェアクラス：

import time
import asyncio

class ToolCallMiddleware:
    def __init__(self, tools: dict, max_iterations: int = 10):
        self.tools = tools
        self.max_iterations = max_iterations
        self.call_log = []

    async def execute(self, tool_name: str, args: dict, timeout: int = 30) -> dict:
        start_time = time.time()

        # 1. ツール存在確認
        if tool_name not in self.tools:
            return {"error": f"不明なツール: {tool_name}"}

        # 2. 入力検証
        try:
            validated_args = self._validate_args(tool_name, args)
        except ValueError as e:
            return {"error": f"無効な引数: {str(e)}"}

        # 3. タイムアウト付きで実行
        try:
            result = await asyncio.wait_for(
                asyncio.to_thread(self.tools[tool_name], **validated_args),
                timeout=timeout
            )
        except asyncio.TimeoutError:
            result = {"error": f"{timeout}秒後にタイムアウト"}
        except Exception as e:
            result = {"error": str(e)}

        # 4. 全呼び出しをログ記録
        elapsed = time.time() - start_time
        self.call_log.append({
            "tool": tool_name,
            "args": args,
            "elapsed_ms": round(elapsed * 1000),
            "success": "error" not in result,
            "timestamp": time.time()
        })

        return result

まとめ

Tool CallingはLLMをエージェントとして本当に使えるものにする技術です。基本的な実装は難しくありませんが、本番環境で耐えられるTool Callingを作ることはディテールの戦いです。

最も重要な3つのこと：明確なツールの説明、適切なエラーハンドリング、ループ防止。この3つをきちんとやるだけで、ツール対応エージェントを沈める問題の大半を避けられます。

このシリーズではエージェント設計パターン（Post 5）、MCP（Post 6）、マルチエージェントシステム（Post 7）、そしてTool Calling（Post 8）を解説しました。理論より実装から学ぶ方が速い — 今すぐ何か作ってみてください。

Discuss on Twitter • View on GitHub