Вызов функций

В статье про интерфейс инструментов мы разобрали базовый четырехшаговый цикл взаимодействия хоста и модели: describe -> decide -> execute -> observe.

Здесь мы рассмотрим инженерную реализацию этого цикла: как разные API кодируют вызовы функций, как безопасно запускать независимые действия параллельно, как собирать потоковый JSON и как построить слой маршрутизации без привязки к единственному поставщику моделей.

Диалекты провайдеров: объявление, вызов и результат

Хотя общий цикл у всех провайдеров одинаков, между ними есть три ключевых архитектурных различия:

Остальные различия сводятся к именам полей (tool_calls против tool_use), структуре контейнеров и префиксам идентификаторов (call_ против toolu_). Хост изолирует эти детали в слое адаптеров: агент работает с единым внутренним представлением инструментов, а адаптер преобразует вызовы под контракт конкретной модели.

Управление выбором инструментов через tool_choice

По умолчанию модель сама решает, сформировать текстовый ответ или вызвать функцию. Хост может переопределить это поведение: у OpenAI и Anthropic параметр называется tool_choice, у Gemini - tool_config.function_calling_config.

// OpenAI: принудительный вызов конкретного инструмента
{
  "tool_choice": {
    "type": "function",
    "function": { "name": "extract_user_profile" }
  }
}
 
// Anthropic: принудительный вызов конкретного инструмента
{
  "tool_choice": {
    "type": "tool",
    "name": "extract_user_profile"
  }
}
 
// Gemini: режим ANY и белый список разрешенных имен
{
  "tool_config": {
    "function_calling_config": {
      "mode": "ANY",
      "allowed_function_names": ["extract_user_profile"]
    }
  }
}

Четыре основных режима управления инструментами:

• Автоматический (auto / AUTO): режим по умолчанию. Модель сама решает, сформировать текстовый ответ или вызвать один либо несколько инструментов.
• Обязательный вызов (required / any / ANY): модель обязана вызвать хотя бы один инструмент, но сама выбирает какой. Режим используют для жестких шагов конвейера: сбор первичных данных или классификация намерения.
• Запрет инструментов (none / NONE): вызовы функций заблокированы, даже если схемы переданы в запросе. Режим включают на финальном шаге генерации ответа пользователю.
• Фиксированный инструмент: модель обязана вызвать конкретный указанный инструмент. Параметр гарантирует выбор инструмента, но не корректность аргументов: соответствие типам обеспечивает strict: true, а бизнес-правила проверяет валидатор на хосте.

Параллельные вызовы инструментов

Агенту часто требуется запросить несколько независимых источников данных: проверить баланс на трех счетах или получить погоду в трех городах.

При последовательном выполнении общее время складывается:

Latency (последовательно) = (LLM_RTT_1 + Tool_1) + (LLM_RTT_2 + Tool_2) + (LLM_RTT_3 + Tool_3)

При задержке инференса в 1 секунду и времени ответа каждого инструмента в 500 мс три последовательных шага займут 4.5 секунды.

Если модель возвращает несколько вызовов за один шаг, хост выполняет их одновременно:

Latency (параллельно) = LLM_RTT_1 + max(Tool_1, Tool_2, Tool_3) + LLM_RTT_2

В таком сценарии общее время ожидания сокращается с 4.5 до 2.5 секунд. Чем больше независимых вызовов, тем выше выигрыш: время работы инструментов определяется самым долгим запросом, а количество обращений к модели сокращается до двух.

Корреляция по call_id и формат Observe

Параллельные запросы завершаются асинхронно и в произвольном порядке: чтение из локального кэша занимает 10 мс, внешний HTTP-запрос - 800 мс. Модель сопоставляет результат с исходным вызовом исключительно по идентификатору.

Правила возврата результатов зависят от API:

• OpenAI: отдельные сообщения с ролью tool и полем tool_call_id. Порядок сообщений в массиве не имеет значения.
• Anthropic: все блоки tool_result передаются внутри одного сообщения с ролью user. Блоки результатов обязаны располагаться в начале сообщения, до любого пользовательского текста, иначе API возвращает ошибку 400. Модель связывает результат с вызовом по tool_use_id.
• Gemini: каждый блок functionResponse содержит идентификатор соответствующего functionCall. В современных версиях модели это единственный надежный способ различить несколько параллельных вызовов одной и той же функции.

Изоляция частичных сбоев

Один из параллельных вызовов может завершиться ошибкой по таймауту, из-за сетевого сбоя или статуса HTTP 500, пока остальные успешно возвращают данные.

Хост не прерывает работу всей системы. Неудачный вызов передается на шаг Observe с сообщением об ошибке, а успешные вызовы - как обычные результаты. Формат соответствует требованиям провайдера: сообщения с ролью tool у OpenAI, массив блоков tool_result у Anthropic или объекты functionResponse у Gemini.

На следующем шаге модель анализирует полученные данные и повторяет только неудачный запрос либо принимает решение на основе доступных ответов.

Параллельный запуск в OpenAI: пул потоков, изоляция ошибок и корреляция по call_id

Copy code

from concurrent.futures import ThreadPoolExecutor, as_completed
import json
from typing import Any, Callable


def execute_single_call(
    call: dict[str, Any],
    executors: dict[str, Callable[..., Any]],
) -> dict[str, Any]:
    """Исполняет один вызов инструмента с изоляцией ошибок."""
    call_id = call.get("id") or call.get("tool_call_id", "call_unknown")

    # Поддержка плоского формата и формата OpenAI с объектом 'function'
    func_payload = call.get("function", {})
    name = call.get("name") or func_payload.get("name")
    raw_args = call.get("arguments") or func_payload.get("arguments", {})

    if not name or name not in executors:
        return {
            "role": "tool",
            "tool_call_id": call_id,
            "content": f"Ошибка: инструмент '{name}' не найден в реестре.",
        }

    try:
        # Парсим JSON-строку, если аргументы переданы строкой (OpenAI формат)
        args = json.loads(raw_args) if isinstance(raw_args, str) else raw_args
        if not isinstance(args, dict):
            raise ValueError("Аргументы инструмента должны быть JSON-объектом.")

        result = executors[name](**args)
        content = json.dumps(result, ensure_ascii=False) if not isinstance(result, str) else result
        return {
            "role": "tool",
            "tool_call_id": call_id,
            "content": content,
        }
    except Exception as err:
        # Ошибка одного инструмента не прерывает выполнение остальных:
        # сообщение об ошибке возвращается в контекст модели на шаге Observe
        return {
            "role": "tool",
            "tool_call_id": call_id,
            "content": f"Сбой выполнения '{name}': {str(err)}",
        }


def dispatch_parallel_calls(
    tool_calls: list[dict[str, Any]],
    executors: dict[str, Callable[..., Any]],
    max_workers: int = 5,
) -> list[dict[str, Any]]:
    """Параллельный запуск независимых инструментов с сохранением correlation id."""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = [
            pool.submit(execute_single_call, call, executors)
            for call in tool_calls
        ]
        for future in as_completed(futures):
            results.append(future.result())
    return results

Ограничения параллельного выполнения

Параллельный запуск подходит только для независимых операций чтения. Есть случаи, когда хост обязан выполнять вызовы строго последовательно. Для этого отключают параллелизм параметром parallel_tool_calls: false у OpenAI или disable_parallel_tool_use: true у Anthropic.

• Зависимости по данным: когда результат первого вызова необходим для выполнения второго (например, create_workspace, затем create_document), действия выполняются строго по очереди.
• Необратимые действия: списание средств, отправка писем или удаление записей требуют подтверждения пользователя и строгого порядка выполнения.
• Ограничения внешних сервисов (Rate Limits): одновременная отправка 10-20 запросов к стороннему API приводит к исчерпанию лимитов и ошибкам HTTP 429.

Потоковая передача аргументов и сборка JSON

В режиме stream: true модель передает аргументы фрагментами токенов. Это позволяет интерфейсу отображать действия агента до завершения полного ответа.

Типичная ошибка реализации - попытка преждевременного парсинга (Parse-Early Trap). Фрагменты потока образуют незавершенный JSON:

Chunk 1: {"city": "Mos
Chunk 2: cow", "days
Chunk 3: ": 3}

Вызов json.loads на промежуточном фрагменте завершается ошибкой JSONDecodeError. Подсчет открывающих и закрывающих скобок также ненадежен: фигурная скобка может находиться внутри строкового литерала.

Кроме того, при параллельных вызовах фрагменты разных функций в едином потоке перемешиваются. Хост поддерживает отдельный строковый буфер для каждого индекса index (OpenAI) или блока tool_use (Anthropic). В API Gemini потоковую передачу аргументов активируют флагом streamFunctionCallArguments.

Как собирать поток

1. Открываем строковый буфер под каждый вызов по его index или call_id.
2. Дописываем входящие токены в буфер этого вызова.
3. Ждем, пока провайдер закроет этот вызов:
   • Anthropic: событие content_block_stop.
   • Gemini: признак willContinue: false.
   • OpenAI: в стриме пошел следующий index или пришел финальный finish_reason: "tool_calls".
4. Парсим JSON через json.loads и сразу запускаем инструмент.

Потоковый парсинг неполного JSON полезен для индикации прогресса в интерфейсе, но шаг Execute всегда требует валидного и полного объекта. Подробнее этот механизм рассмотрен в статье про структурированный вывод.

Системные лимиты провайдеров

Даже если API допускает десятки схем, передача большого каталога перегружает контекст и снижает точность выбора. Для объемных реестров применяют динамический поиск инструментов (Dynamic Tool Retrieval).

Маршрутизация моделей и резервные цепочки

Прямая привязка к одному провайдеру создает единую точку отказа: сбой дата-центра, исчерпание лимитов TPM/RPM или резкий рост задержки парализуют работу агента. Для отказоустойчивости между агентом и моделями размещают шлюз маршрутизации (LLM Gateway).

Принципы устойчивой маршрутизации

• Единый формат запросов: хост отправляет запросы в унифицированном формате (часто на базе спецификации OpenAI), а шлюз преобразует схемы и аргументы в формат целевого API.
• Логические псевдонимы (Model Aliases): код агента обращается к абстрактным именам, таким как general-reasoning или fast-triage. Переключение конкретных моделей и версий происходит в конфигурации шлюза без правок кода агента.
• Резервные цепочки (Fallback Chains): при получении ошибок HTTP 5xx, таймаутов или статуса HTTP 429 шлюз автоматически перенаправляет запрос к следующему поставщику по списку приоритетов.

Пример декларативной конфигурации маршрутизации: агент обращается к логическому имени agent_tools_execution, а шлюз управляет переключением между провайдерами:

{
  "routes": {
    "agent_tools_execution": {
      "strategy": "priority_fallback",
      "targets": [
        {
          "provider": "openai",
          "model": "gpt-6-astra",
          "timeout_ms": 3000,
          "retry_on": [429, 500, 502, 503]
        },
        {
          "provider": "anthropic",
          "model": "claude-sonnet-5",
          "timeout_ms": 5000,
          "retry_on": [429, 500, 502, 503]
        },
        {
          "provider": "google",
          "model": "gemini-3.8-flash",
          "timeout_ms": 6000
        }
      ]
    }
  }
}

Самовосстановление при ошибках валидации

Режим strict: true гарантирует соответствие базовой структуре JSON: типы полей, обязательные ключи required, запрет лишних свойств additionalProperties: false и допустимые значения enum. Хост выполняет роль второго уровня защиты: проверяет числовые диапазоны, регулярные выражения pattern, значения по умолчанию и бизнес-правила. Если строгий режим не поддерживается, хост также проверяет наличие обязательных полей и корректность синтаксиса JSON.

Механизм восстановления использует стандартный шаг Observe: сообщение об ошибке валидации возвращается модели как результат выполнения инструмента, без перезапуска всей сессии.

[Попытка 1: Модель]
tool_call -> get_weather({"city": "Moscow", "days": 30})
 
[Проверка: Хост]
Валидация не пройдена: days=30 больше maximum=7
 
[Сообщение хоста модели]
role: "tool", tool_call_id: "call_123"
content: "Ошибка валидации параметра 'days': значение 30 недопустимо. Допустимый диапазон: 1-7."
 
[Попытка 2: Модель]
tool_call -> get_weather({"city": "Moscow", "days": 7})

В этом сценарии хост проверяет допустимый диапазон самостоятельно: некоторые провайдеры игнорируют или отклоняют ключевые слова minimum и maximum в схеме. Точное сообщение валидатора с указанием границ позволяет модели скорректировать аргументы на следующем шаге.

Повторные попытки подходят для исправления ошибок схемы и сбоев выполнения. Отказ модели по соображениям безопасности в этот цикл не включают: структуру таких ответов (refusal, stop_reason: "refusal") подробно разбираем в статье про структурированный вывод.

Как применять на интервью

На собеседованиях по проектированию ИИ-систем тему вызова функций оценивают спрашивая про оптимизацию задержки, изоляцию сбоев и управление параллелизмом.

Оптимизация задержки в поиске путешествий

• Сценарий: спроектировать ассистента для поиска туров. По запросу "Найди билеты в Рим и отели на выходные" система опрашивает независимые API авиакомпаний, отелей и аренды автомобилей.
• Пример ответа: за один шаг модель генерирует вызовы search_flights, search_hotels и search_car_rentals. Хост выполняет запросы параллельно. Задержка составляет LLM_RTT_1 + max(T_i) + LLM_RTT_2 вместо суммы всех последовательных вызовов. Результаты передаются модели на шаге Observe с привязкой по идентификаторам: сообщениями с ролью tool для OpenAI или единым сообщением user с блоками tool_result для Anthropic.

Изоляция сбоев при агрегации биржевых котировок

• Сценарий: торговый агент запрашивает котировки актива на трех биржах. Один из сервисов недоступен по таймауту или возвращает HTTP 504. Как сохранить устойчивость системы?
• Пример ответа: хост не прерывает работу с общей ошибкой 500. Успешно полученные данные и сообщение об ошибке недоступного сервиса возвращаются модели одновременно. Модель либо повторяет запрос к источнику с ошибкой, либо строит аналитику на основе оставшихся двух котировок.

Ограничения параллелизма при оформлении заказа

• Сценарий: можно ли одновременно запустить поиск товара, списание средств, резервирование на складе и отправку уведомления?
• Пример ответа: финансовые и изменяющие состояние операции нельзя запускать параллельно без подтверждения предыдущих шагов: списание средств выполняется строго после успешного резервирования на складе. Параллельно выполняют только независимые операции чтения (проверка остатков бонусов, расчет стоимости доставки разными курьерскими службами). Для предотвращения гонок хост принудительно отключает параллельный вызов функций в параметрах запроса к модели, а не полагается на ее поведение.