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

В первой статье главы мы разобрали концептуальный четырехшаговый цикл взаимодействия хоста и модели. В реальной инженерной практике перед нами встает задача интеграции: как именно провайдеры (OpenAI, Anthropic, Google Gemini) кодируют вызовы функций в своих API, как эффективно выполнять независимые вызовы параллельно, как безопасно собирать потоковый JSON и как построить отказоустойчивую маршрутизацию без жесткой привязки к единственному поставщику.

Каждый крупный провайдер пришел к собственному диалекту вызова инструментов. Различия затрагивают формат объявления схем, способ упаковки аргументов и механизм сопоставления результатов.

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

Несмотря на единую концептуальную модель (Describe -> Decide -> Execute -> Observe), каждый провайдер использует собственный формат сообщений и соглашения по передаче данных.

Ключевое различие, с которым сталкиваются разработчики: OpenAI возвращает аргументы вызова как сырую JSON-строку, требующую явного вызова json.loads на стороне хоста, тогда как Anthropic и Gemini отдают уже десериализованный объект.

Чтобы архитектура приложения не зависела от конкретного SDK, на уровне хоста обычно хранится единое каноническое описание инструмента, а шлюз или адаптер клиентского слоя транслирует его в формат целевого API непосредственно при формировании запроса.

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

По умолчанию языковая модель сама оценивает контекст и решает, отвечать текстом или запросить вызов функции. Однако во многих сценариях хосту требуется принудительно управлять поведением модели с помощью параметра tool_choice (или tool_config в терминологии Gemini).

// OpenAI: принудительный вызов конкретного инструмента
{
  "tool_choice": {
    "type": "function",
    "function": { "name": "extract_user_profile" }
  }
}
 
// Anthropic: принудительный вызов инструмента
{
  "tool_choice": {
    "type": "tool",
    "name": "extract_user_profile"
  }
}

Существует четыре основных режима выбора:

1. Автоматический (auto): режим по умолчанию. Модель анализирует запрос и выбирает между генерацией текста и вызовом одного или нескольких доступных инструментов.
2. Обязательный вызов (required / any): модель обязана вызвать хотя бы один инструмент из списка, но сама выбирает, какой именно. Этот режим используется, когда агент находится на специализированном шаге пайплайна (например, этап сбора данных или классификации намерений).
3. Запрет инструментов (none): запрещает модели вызывать любые функции, даже если их схемы переданы в запросе. Полезно на финальном шаге генерации ответа пользователю, чтобы модель синтезировала текст на основе уже собранного контекста.
4. Фиксированный инструмент (Forced Tool): модель принудительно вызывает строго указанный инструмент по имени. Это гарантирует извлечение данных нужной структуры без риска получить текст в свободной форме.

Параллельные вызовы инструментов (Parallel Tool Calls)

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

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

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

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

При включенном параллельном вызове (parallel_tool_calls: true) модель за один шаг рассуждения возвращает массив из нескольких tool_calls. Хост запускает все исполнители одновременно в пуле потоков или асинхронных корутинах:

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

Общее время ожидания сокращается на 60-70%, схлопываясь до длительности самого медленного единичного вызова.

Корреляция по call_id и порядок ответов

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

Модель связывает результаты со своими вызовами исключительно по идентификатору tool_call_id (tool_use_id). Хост отправляет пачку результатов обратно модели, при этом строгий порядок сообщений в массиве не имеет значения, пока идентификаторы совпадают.

Обработка частичных сбоев (Partial Failures)

В реальных распределенных системах один из параллельных вызовов может завершиться ошибкой (таймаут, сетевой сбой, HTTP 500), пока остальные отработали успешно.

Хост не должен аварийно прерывать всю операцию. Упавший вызов упаковывается в стандартное сообщение роли tool с описанием ошибки, а успешные вызовы возвращают свои данные. Получив такой комбинированный контекст, модель на следующем шаге может использовать успешные результаты и предпринять повторную попытку только для упавшего действия.

Параллельный запуск инструментов через пул потоков с изоляцией ошибок

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:
        # Сбой одного инструмента не роняет весь батч: ошибка возвращается в контекст модели
        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

Границы безопасности: когда параллелизм необходимо отключать

Параллельные вызовы не являются универсальным решением. В трех сценариях параллелизм создает серьезные риски и должен быть принудительно отключен:

1. Причинно-следственные зависимости (Ordering Dependencies): если результат первого шага требуется на вход второму (например, create_workspace и последующий create_document), запросы обязаны выполняться строго последовательно.
2. Необратимые деструктивные действия (Consequential Operations): операции списания средств, отправки писем клиентам или удаления разделов базы данных требуют явного подтверждения и строго детерминированного порядка.
3. Ограничения пропускной способности (Downstream Rate Limits): всплеск из 10-20 одновременных вызовов к внешнему API может мгновенно исчерпать квоту запросов и привести к блокировке клиента по HTTP 429.

Потоковая передача аргументов и ловушка преждевременного парсинга

При включении потоковой передачи ответов (stream: true) модель генерирует аргументы инструментов небольшими фрагментами токенов. Для пользователя это обеспечивает минимальную задержку до начала отображения активности агента.

Однако потоковая передача таит в себе распространенную архитектурную ошибку - ловушку преждевременного парсинга (The Parse-Early Trap).

Фрагменты аргументов приходят в виде неполных кусков JSON-строки:

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

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

Правильный алгоритм сборки потока

1. Инициализация аккумулятора: при получении первого чанка с идентификатором вызова хост создает строковый буфер для данного index или call_id.
2. Накопление: все последующие фрагменты аргументов конкатенируются в буфер.
3. Ожидание сигнального токена: хост не пытается парсить JSON до тех пор, пока от провайдера не придет явный сигнал окончания генерации блока (finish_reason: "tool_calls" в OpenAI, событие content_block_stop в Anthropic).
4. Финальная десериализация и запуск: только после закрытия потока аргументы десериализуются целиком и передаются исполнителю.

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

При проектировании платформы важно учитывать жесткие ограничения моделей и инфраструктуры:

Превышение лимита инструментов перегружает контекст и ухудшает внимание модели. Если в вашей системе сотни сервисов, передавать их напрямую в список tools нельзя - необходимо использовать паттерн Dynamic Tool Retrieval.

Маршрутизация моделей и цепочки переключения (Fallback Chains)

Привязка логики вызова функций к единственному провайдеру создает критическую точку отказа. Сбои в дата-центрах, исчерпание квот по TPM/RPM, внезапный рост задержек или разница в стоимости делают необходимым использование промежуточного уровня маршрутизации (LLM Routing Layer).

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

1. Единый OpenAI-совместимый интерфейс: хост отправляет запросы в каноническом формате, а шлюз прозрачно транслирует схемы и аргументы под целевого провайдера.
2. Логические псевдонимы (Model Aliases): вместо захардкоженных названий версий (gpt-5.6-2026-05 или claude-5-sonnet-latest) приложение обращается к псевдонимам general-reasoning или fast-triage. Переключение конкретной модели под капотом происходит в конфигурации шлюза без правки кода агента.
3. Цепочки переключения (Fallback Chains): при возникновении ошибки сервера (HTTP 5xx), превышении таймаута или исчерпании лимитов (HTTP 429) шлюз автоматически перенаправляет вызов следующему провайдеру по приоритету.

Пример декларативной конфигурации цепочки переключения:

{
  "routes": {
    "agent_tools_execution": {
      "strategy": "priority_fallback",
      "targets": [
        {
          "provider": "openai",
          "model": "gpt-5.6-terra",
          "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.7-flash",
          "timeout_ms": 6000
        }
      ]
    }
  }
}

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

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

Паттерн надежного восстановления строится на контролируемой обратной связи:

[Попытка 1: Модель]
tool_call -> get_weather({"city": "Moscow", "units": "kelvin"})

[Проверка: Хост]
Валидация JSON Schema не пройдена: 'kelvin' не входит в enum ['celsius', 'fahrenheit']

[Сообщение хоста модели]
role: "tool", tool_call_id: "call_123"
content: "Ошибка валидации параметра 'units': значение 'kelvin' недопустимо. Допустимые значения: ['celsius', 'fahrenheit']."

[Попытка 2: Модель]
tool_call -> get_weather({"city": "Moscow", "units": "celsius"})

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

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

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

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

• Формулировка задачи: спроектировать диалогового ассистента для поиска туров, который по запросу пользователя ("Найди отели в Риме и билеты на выходные") опрашивает независимые API авиакомпаний, гостиниц и проката авто.
• Пример ответа:
  1. Параллельные вызовы (parallel_tool_calls): модель за один шаг генерирует массив вызовов (search_flights, search_hotels, search_car_rentals). Хост запускает их конкурентно через пул потоков или асинхронные задачи.
  2. Сокращение задержки: суммарное время ожидания сокращается с суммы последовательных задержек (sum T_i) до времени самого медленного API (max T_i).
  3. Корреляция по call_id: хост возвращает результаты пачкой сообщений роли tool, связывая их с исходными вызовами модели по идентификаторам.

Торговый агент и изоляция частичных сбоев

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

Оформление заказов в E-Commerce и границы параллелизма

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