ツール呼び出し（Function Calling）

Function Calling（関数呼び出し）は、LLM が外部ツールを使用できるようにするための中核メカニズムです。

Function Calling（関数呼び出し）により、モデルはいつツールを呼び出すか、どのツールを呼び出すか、どのようなパラメータを渡すかを決定できます。

なぜ Function Calling が必要なのか？

想像してみてください。LLM は賢いけれども武器を持たないコンサルタントのようなものです。多くの知識を持っていますが、次のことはできません：

• リアルタイム情報の取得（例：最新の天気、株価）
• 計算の実行（例：複雑な数学的演算）
• 外部システムの操作（例：メール送信、ファイルの読み書き）

Function Calling は LLM に両手を与え、自身の制限を突破して実際のタスクを実行できるようにします。

Function Calling の動作原理

Function Calling の基本的なフローは次のとおりです：

中核概念

• ツール定義：ツールの機能、パラメータ、戻り値を記述します
• ツール選択：LLM がユーザーの質問に基づいて適切なツールを選択します
• パラメータ抽出：LLM が質問からツールに必要なパラメータを抽出します
• 結果処理：ツールの実行結果を最終回答に統合します

具体例による説明

天気を照会するツールがあると仮定して、ユーザーが次のように尋ねたとき：北京の今日の天気は？のとき：

1. LLM は天気照会ツールを呼び出す必要があると認識します
2. 質問からパラメータを抽出します：city="北京", date="今天"
3. 天気 API を呼び出してデータを取得します
4. 天気データを分かりやすい回答にまとめてユーザーに返します

ツール定義と説明文の作成

LLM にツールを正しく使用させるには、まずツールを明確に定義する必要があります。優れたツール定義は分かりやすい説明書のようなものであり、LLM に次のことを理解させます：

• このツールは何をするものか？
• いつ使用するのか？
• どのようなパラメータが必要か？
• パラメータの形式は何か？

ツール定義の構造

完全なツール定義には通常、次の部分が含まれます：

高品質なツール説明文の作成

1. 説明は明確かつ具体的に

• 良いとは言えない説明：天気を調べる
• 良い説明：指定した都市の特定日付の天気情報を取得します。温度、湿度、風速、天気状況（晴れ、雨、曇りなど）を含みます

2. パラメータの説明は詳細に

良くないパラメータ説明：

"city": {"type": "string"}

良いパラメータ説明：

"city": {
    "type": "string",
    "description": "完整的城市名称，如'北京市'、'上海市'。不要使用简称或拼音。"
}

3. 列挙型（enum）を使用して選択肢を制限する

限られた選択肢については、列挙型（enum）を使用して LLM の理解を助けます：

"unit": {
    "type": "string",
    "description": "温度单位",
    "enum": ["celsius", "fahrenheit"],
    "default": "celsius"
}

4. 例となる値を提供する

説明の中で例を提供し、LLM が形式を理解できるようにします：

"date": {
    "type": "string",
    "description": "日期，格式应为'YYYY-MM-DD'，例如'2024-06-15'"
}

実用的なツール定義の例

1. 電卓ツール

calculator_tool = {
    "name": "calculate",
    "description": "执行数学计算，支持加减乘除、幂运算等基本运算",
    "parameters": {
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "数学表达式，如'2 + 3 * 4'、'sqrt(16)'、'sin(30)'"
            }
        },
        "required": ["expression"]
    }
}

2. 検索ツール

search_tool = {
    "name": "search_web",
    "description": "在互联网上搜索最新信息",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词，尽量具体明确"
            },
            "num_results": {
                "type": "integer",
                "description": "返回结果数量，默认为5",
                "default": 5,
                "minimum": 1,
                "maximum": 10
            }
        },
        "required": ["query"]
    }
}

3. ファイル操作ツール

file_tool = {
    "name": "read_file",
    "description": "读取指定文件的内容",
    "parameters": {
        "type": "object",
        "properties": {
            "file_path": {
                "type": "string",
                "description": "文件的完整路径，如'/home/user/document.txt'"
            },
            "encoding": {
                "type": "string",
                "description": "文件编码，默认为'utf-8'",
                "default": "utf-8"
            }
        },
        "required": ["file_path"]
    }
}

ツール定義のベストプラクティス

• 名前は簡潔かつ明確に：動詞で始める。例：get_、calculate_、search_
• 説明は完全かつ詳細に：ツールの用途、適用シナリオ、制限を説明します
• パラメータ検証を十分に行う：パラメータの型、範囲、形式の要件を定義します
• デフォルト値を提供する：オプションパラメータに合理的なデフォルト値を提供します
• エラー状況を考慮する：説明の中で発生し得るエラーと制限を述べます

ツール定義の検証

ツールを定義したら、検証テストを実施すべきです：

def validate_tool_definition(tool_def):
    """验证工具定义是否完整"""
    required_fields = ["name", "description", "parameters"]

    for field in required_fields:
        if field not in tool_def:
            return False, f"缺少必要字段: {field}"

    # 检查参数结构
    if "properties" not in tool_def["parameters"]:
        return False, "参数定义缺少properties字段"

    return True, "工具定义完整"

# 测试验证
is_valid, message = validate_tool_definition(weather_tool)
print(f"验证结果: {is_valid}, 消息: {message}")

パラメータ抽出と検証

LLM がユーザーの質問からパラメータを抽出した後、これらのパラメータを検証・処理して、ツールが正しく実行できるようにする必要があります。

パラメータ抽出のプロセス

LLM がツールの呼び出しを決定すると、ユーザー入力を分析し、ツールに必要なパラメータを抽出します：

用户输入: "查询北京明天的天气温度"
工具: get_weather
提取参数: {"city": "北京", "date": "明天"}

パラメータ抽出の課題

パラメータ抽出では以下の問題が発生する可能性があります：

• 情報の欠落：ユーザーが必要な情報をすべて提供していない
• 形式の不一致：ユーザーが提供した形式がツールの要件と一致しない
• 曖昧性の解釈：同じ情報でも複数の解釈がある可能性がある
• 文脈への依存：パラメータは会話履歴と組み合わせて理解する必要がある

パラメータ検証の方法

1. 型検証

パラメータの型が要件を満たしていることを確認します：

def validate_parameters(params, tool_def):
    """验证参数类型"""
    errors = []

    for param_name, param_def in tool_def["parameters"]["properties"].items():
        if param_name in params:
            param_value = params[param_name]
            expected_type = param_def.get("type")

            # 类型检查
            if expected_type == "string" and not isinstance(param_value, str):
                errors.append(f"参数'{param_name}'应为字符串类型")
            elif expected_type == "integer" and not isinstance(param_value, int):
                errors.append(f"参数'{param_name}'应为整数类型")
            elif expected_type == "number" and not isinstance(param_value, (int, float)):
                errors.append(f"参数'{param_name}'应为数字类型")
            elif expected_type == "boolean" and not isinstance(param_value, bool):
                errors.append(f"参数'{param_name}'应为布尔类型")

    return errors

2. 範囲検証

パラメータの値が許容範囲内かどうかを確認します：

def validate_range(params, tool_def):
    """验证参数范围"""
    errors = []

    for param_name, param_def in tool_def["parameters"]["properties"].items():
        if param_name in params:
            param_value = params[param_name]

            # 检查最小值
            if "minimum" in param_def and param_value < param_def["minimum"]:
                errors.append(f"参数'{param_name}'不能小于{param_def['minimum']}")

            # 检查最大值
            if "maximum" in param_def and param_value > param_def["maximum"]:
                errors.append(f"参数'{param_name}'不能大于{param_def['maximum']}")

            # 检查枚举值
            if "enum" in param_def and param_value not in param_def["enum"]:
                errors.append(f"参数'{param_name}'必须是{param_def['enum']}中的一个")

    return errors

3. 必須パラメータの検証

すべての必須パラメータが提供されていることを確認します：

def validate_required(params, tool_def):
    """验证必填参数"""
    errors = []
    required_params = tool_def["parameters"].get("required", [])

    for param_name in required_params:
        if param_name not in params:
            errors.append(f"缺少必填参数: {param_name}")

    return errors

完全なパラメータ検証システム

パラメータ抽出と検証の完全なフロー

パラメータ修正戦略

パラメータ検証が失敗した場合は、次の戦略を取ることができます：

• ユーザーに問い合わせる：欠落または誤ったパラメータをユーザーに直接問い合わせる
• デフォルト値を使用する：オプションのパラメータには、事前定義されたデフォルト値を使用する
• スマート推論：コンテキストに基づいて妥当なパラメータ値を推論する
• フォーマット変換：ユーザーが提供した形式をツールが要求する形式に変換する

def fix_parameters(params, errors, tool_def):
    """尝试修正参数错误"""
    fixed_params = params.copy()

    for error in errors:
        if "缺少必填参数" in error:
            param_name = error.split(": ")[1]
            # 尝试从上下文推断或使用默认值
            if param_name == "date":
                fixed_params[param_name] = "今天"  # 使用当天作为默认值

        elif "应为字符串类型" in error:
            param_name = error.split("'")[1]
            # 尝试转换为字符串
            fixed_params[param_name] = str(params[param_name])

    return fixed_params

エラー処理とリトライ（再試行）メカニズム

実際のアプリケーションでは、ツール呼び出しが失敗する可能性があります。堅牢なエラーハンドリングとリトライメカニズムは、信頼性の高いAI Agentを構築するための鍵です。

<よくあるツール呼び出しエラー

• ネットワークエラー：API呼び出しのタイムアウト、接続失敗
• パラメータエラー：パラメータ検証の失敗、形式が正しくない
• 権限エラー：APIキーが無効、権限が不十分
• リソースエラー：サービスが利用不可、呼び出し制限に達した
• ロジックエラー：ツール内部のロジックエラー

エラー処理戦略

1. 段階的エラーハンドリング

エラーの種類に応じて異なる処理戦略を採用する：

class ErrorHandler:
    """错误处理器"""

    def handle(self, error, context):
        """处理错误"""
        error_type = self.classify_error(error)

        if error_type == "network":
            return self.handle_network_error(error, context)
        elif error_type == "parameter":
            return self.handle_parameter_error(error, context)
        elif error_type == "authentication":
            return self.handle_auth_error(error, context)
        elif error_type == "resource":
            return self.handle_resource_error(error, context)
        else:
            return self.handle_unknown_error(error, context)

    def classify_error(self, error):
        """分类错误类型"""
        error_str = str(error).lower()

        if any(word in error_str for word in ["timeout", "connection", "network"]):
            return "network"
        elif any(word in error_str for word in ["parameter", "invalid", "missing"]):
            return "parameter"
        elif any(word in error_str for word in ["auth", "key", "permission"]):
            return "authentication"
        elif any(word in error_str for word in ["limit", "quota", "unavailable"]):
            return "resource"
        else:
            return "unknown"

    def handle_network_error(self, error, context):
        """处理网络错误"""
        return {
            "success": False,
            "error": "网络连接失败，请检查网络后重试",
            "retryable": True,
            "retry_after": 5  # 5秒后重试
        }

    def handle_parameter_error(self, error, context):
        """处理参数错误"""
        return {
            "success": False,
            "error": f"参数错误: {str(error)}",
            "retryable": False,
            "suggestion": "请检查输入参数是否正确"
        }

    # 其他错误处理方法类似...

2. ユーザーフレンドリーなエラーメッセージ

技術的なエラーをユーザーが理解できるメッセージに変換する：

def user_friendly_error(error):
    """生成用户友好的错误消息"""
    error_map = {
        "timeout": "请求超时，请稍后重试",
        "connection_error": "网络连接失败，请检查网络",
        "invalid_api_key": "API 密钥无效，请检查配置",
        "rate_limit_exceeded": "调用频率过高，请稍后再试",
        "invalid_parameters": "输入参数不正确，请检查后重试"
    }

    error_str = str(error).lower()
    for key, message in error_map.items():
        if key in error_str:
            return message

    return "系统繁忙，请稍后重试"

リトライ（再試行）メカニズム

リトライ可能なエラーについては、スマートなリトライ戦略を実装する必要がある：

1. 指数バックオフリトライ

import time
import random

class ExponentialBackoffRetry:
    """指数退避重试"""

    def __init__(self, max_retries=3, base_delay=1, max_delay=30):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay

    def retry(self, func, *args, **kwargs):
        """执行带重试的函数调用"""
        last_error = None

        for attempt in range(self.max_retries):
            try:
                return func(*args, **kwargs)

            except Exception as e:
                last_error = e

                # 检查是否可重试
                if not self.is_retryable(e):
                    raise

                # 最后一次尝试，不再重试
                if attempt == self.max_retries - 1:
                    break

                # 计算等待时间（指数退避 + 随机抖动）
                delay = min(
                    self.base_delay * (2 ** attempt) + random.uniform(0, 1),
                    self.max_delay
                )

                print(f"第{attempt + 1}次尝试失败，{delay:.1f}秒后重试...")
                time.sleep(delay)

        # 所有重试都失败
        raise last_error

    def is_retryable(self, error):
        """判断错误是否可重试"""
        error_str = str(error).lower()
        retryable_errors = ["timeout", "connection", "busy", "temporarily"]

        for retryable_error in retryable_errors:
            if retryable_error in error_str:
                return True

        return False

# 使用示例
retry = ExponentialBackoffRetry(max_retries=3)

def call_weather_api(city, date):
    """调用天气API（模拟可能失败）"""
    # 模拟API调用
    if random.random() < 0.3:  # 30%概率失败
        raise ConnectionError("API连接超时")
    return {"temperature": 25, "condition": "晴"}

try:
    result = retry.retry(call_weather_api, "北京", "今天")
    print(f"成功获取天气: {result}")
except Exception as e:
    print(f"获取天气失败: {e}")

2. サーキットブレーカーパターン

連続した失敗によるシステムの雪崩を防ぐ：

class CircuitBreaker:
    """熔断器"""

    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open

    def execute(self, func, *args, **kwargs):
        """通过熔断器执行函数"""
        if self.state == "open":
            # 检查是否应该尝试恢复
            if self.should_try_reset():
                self.state = "half-open"
            else:
                raise Exception("熔断器开启，服务暂时不可用")

        try:
            result = func(*args, **kwargs)

            # 成功调用，重置状态
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0

            return result

        except Exception as e:
            self.record_failure()
            raise e

    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.failure_count >= self.failure_threshold:
            self.state = "open"

    def should_try_reset(self):
        """检查是否应该尝试重置"""
        if self.last_failure_time is None:
            return True

        elapsed = time.time() - self.last_failure_time
        return elapsed > self.reset_timeout

# 使用示例
breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30)

def unreliable_service():
    """模拟不可靠的服务"""
    if random.random() < 0.5:
        raise Exception("服务异常")
    return "服务正常"

for i in range(10):
    try:
        result = breaker.execute(unreliable_service)
        print(f"调用{i+1}: {result}")
    except Exception as e:
        print(f"调用{i+1}: {e}")
        time.sleep(1)

完全なツール呼び出しフレームワーク

上記のすべてのコンポーネントを組み合わせることで、完全なツール呼び出しフレームワークを構築できます：