Вызов инструментов

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. Используйте перечисления для ограничения вариантов

Для ограниченного набора вариантов используйте перечисления, чтобы помочь 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-агента.

Распространённые ошибки при вызове инструментов

• Сетевые ошибки: Тайм-аут вызова 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)

Полный фреймворк вызова инструментов

Объединив все вышеперечисленные компоненты, мы можем построить полный фреймворк вызова инструментов: