Function Calling 2.0: Параллельное выполнение и вложенные вызовы

Введение: Эволюция вызова функций в Gemini

Приветствую вас в третьем модуле. До этого момента мы рассматривали Function Calling как линейный процесс: модель получает запрос, решает вызвать одну функцию, мы возвращаем результат, и модель генерирует ответ. Это работает отлично для простых задач, вроде «Какая погода в Лондоне?».

Но реальный мир сложнее. Представьте, что вы разрабатываете ассистента для путешествий. Пользователь пишет: «Забронируй билеты в Париж на 5 мая и подбери отель в центре на 3 ночи». В классической схеме (Function Calling 1.0) модели пришлось бы действовать последовательно: сначала вызвать поиск билетов, дождаться ответа, и только потом искать отель. Это создает лишнюю задержку (latency) и тратит время пользователя.

Function Calling 2.0 в Gemini (доступный в моделях 1.5 Pro и Flash) меняет правила игры, вводя концепцию параллельного выполнения (Parallel Function Calling). Теперь модель может проанализировать запрос и выдать намерение вызвать несколько инструментов одновременно. Ваша задача как разработчика — правильно обработать этот «пакет» вызовов и вернуть результаты скопом.

Архитектура параллельного выполнения

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

• Анализ промпта: Модель распознает, что для ответа на вопрос требуются независимые друг от друга данные (например, погода в Токио и курс иены к доллару).
• Генерация вызовов: Вместо одного объекта function_call, модель возвращает список вызовов. В структуре ответа API это выглядит как массив объектов в поле parts (или tool_calls в зависимости от используемой SDK-обертки).
• Сторона клиента: Ваше приложение получает этот список. Критически важно здесь то, что вы можете (и должны) запустить эти функции параллельно — используя асинхронность (asyncio) или потоки (threads), чтобы минимизировать общее время ожидания.
• Агрегация: Вы собираете результаты всех функций и отправляете их обратно модели в одном сообщении function_response.

Важное отличие: Если функции зависимы (например, результат первой функции нужен как аргумент для второй), модель не будет вызывать их параллельно. Она достаточно умна, чтобы понять цепочку зависимостей и выполнит их последовательно (multi-turn).

import os
import google.generativeai as genai
from google.generativeai.types import FunctionDeclaration, Tool

# Настройка API
genai.configure(api_key=os.environ["GEMINI_API_KEY"])

# Определение функций-заглушек для демонстрации
def get_weather(location: str):
    """Получает текущую погоду в указанном городе."""
    print(f"[API CALL] Запрос погоды для: {location}")
    # Имитация ответа
    return {"temperature": 22, "condition": "sunny", "location": location}

def get_stock_price(ticker: str):
    """Получает текущую цену акции."""
    print(f"[API CALL] Запрос цены акции: {ticker}")
    # Имитация ответа
    return {"price": 150.50, "currency": "USD", "ticker": ticker}

# Создание инструментов
tools_list = [get_weather, get_stock_price]

# Инициализация модели с инструментами
model = genai.GenerativeModel(
    model_name='gemini-1.5-flash',
    tools=tools_list
)

# Запуск чат-сессии с автоматическим вызовом функций (auto-execution)
# В реальных сценариях часто используется ручной контроль, но здесь мы покажем магию
chat = model.start_chat(enable_automatic_function_execution=True)

# Запрос, требующий параллельного выполнения
response = chat.send_message(
    "Какая сейчас погода в Нью-Йорке и сколько стоят акции Apple?"
)

print(f"\nОтвет модели: {response.text}")

Разбор механики ручного управления (Manual Control)

Хотя автоматическое выполнение (как в примере выше) удобно для прототипов, в продакшене (Production) вам часто потребуется ручной контроль. Это нужно для обработки ошибок, логирования, проверки прав доступа или выполнения запросов к БД.

При ручной обработке структура ответа модели меняется. Вместо текста вы получите объект, содержащий список function_call. Ваша задача — пройтись по этому списку циклом.

Обратите внимание на структуру данных parts. Когда модель хочет вызвать несколько функций, она присылает несколько частей (Parts), каждая из которых содержит function_call. Вы должны сформировать ответный список function_response, соблюдая тот же порядок или сопоставляя ID вызовов (в зависимости от версии API, Gemini опирается на последовательность контекста).

Ниже приведен пример, как реализовать цикл обработки параллельных вызовов вручную.

# Инициализация чата БЕЗ автоматического выполнения
chat_manual = model.start_chat(enable_automatic_function_execution=False)

query = "Сравни погоду в Лондоне и Париже."
response = chat_manual.send_message(query)

# Проверяем, есть ли вызовы функций в ответе
if response.candidates[0].content.parts[0].function_call:
    print("Обнаружены запросы к функциям...")
    
    # Список для сбора ответов от функций
    function_responses = []
    
    # Словарь доступных функций для вызова по имени
    available_functions = {
        'get_weather': get_weather,
        'get_stock_price': get_stock_price
    }

    # Итерация по всем частям ответа (Model может вернуть несколько function_call)
    for part in response.candidates[0].content.parts:
        if fn := part.function_call:
            func_name = fn.name
            func_args = dict(fn.args)
            
            print(f" -> Вызов: {func_name} с аргументами {func_args}")
            
            if func_name in available_functions:
                # Выполнение функции
                api_response = available_functions[func_name](**func_args)
                
                # Формирование части ответа
                # ВАЖНО: Структура ответа должна соответствовать ожиданиям API
                function_responses.append(
                    genai.protos.Part(
                        function_response=genai.protos.FunctionResponse(
                            name=func_name,
                            response={'result': api_response}
                        )
                    )
                )

    # Отправка результатов обратно модели
    if function_responses:
        print("Отправка результатов обратно в модель...")
        final_response = chat_manual.send_message(function_responses)
        print(f"Итоговый ответ: {final_response.text}")

Вложенные вызовы и сложные сценарии

Термин «вложенные вызовы» (Nested Calls) в контексте LLM часто понимают двояко. Давайте проясним:

1. Композиция функций (Technical Nesting): Это когда результат одной функции передается сразу в другую, например send_email(translate(get_text())). Gemini напрямую не строит такие синтаксические конструкции в JSON. Вместо этого модель использует итеративный подход (Multi-turn). Она вызовет get_text, получит результат, затем в следующем шаге вызовет translate с полученным текстом. Это не истинная параллельность, а последовательная зависимость.
2. Логическая вложенность (Complex Schema): Вы можете определить функцию, которая принимает сложный объект (JSON) в качестве аргумента. Например, функция create_calendar_event может принимать аргумент attendees, который является списком объектов с полями name и email. Это позволяет передавать структурированные данные за один проход.

Совет эксперта: Если ваша бизнес-логика требует строгой последовательности (сначала проверить баланс, потом списать деньги), не полагайтесь на то, что модель «случайно» вызовет их в правильном порядке в параллельном режиме. Лучше спроектируйте инструменты так, чтобы модель вынуждена была делать два шага (two-step reasoning), либо объедините эти действия в одну атомарную функцию на бэкенде.

python
import google.generativeai as genai

# 1. Определение функций
def turn_light(room: str, state: str):
    return f"Light in {room} turned {state}"

def set_thermostat(room: str, temperature: int):
    return f"Thermostat in {room} set to {temperature}"

tools = {'turn_light': turn_light, 'set_thermostat': set_thermostat}

# 2. Инициализация (псевдокод конфигурации)
model = genai.GenerativeModel('gemini-1.5-flash', tools=tools.values())
chat = model.start_chat()

# 3. Запрос
prompt = "Выключи свет в гостиной и на кухне, установи температуру в спальне на 24 градуса."
response = chat.send_message(prompt)

# 4. Обработка (Решение)
response_parts = []

# Проверяем части ответа
if response.candidates[0].content.parts:
    for part in response.candidates[0].content.parts:
        if part.function_call:
            fname = part.function_call.name
            fargs = dict(part.function_call.args)
            
            # Выполнение (в реальности здесь стоит добавить try/except)
            if fname in tools:
                result = tools[fname](**fargs)
                
                # Сборка ответа
                response_parts.append(
                    genai.protos.Part(
                        function_response=genai.protos.FunctionResponse(
                            name=fname,
                            response={'result': result}
                        )
                    )
                )

# Отправка результатов
if response_parts:
    final_res = chat.send_message(response_parts)
    print(final_res.text)
# Ожидаемый результат: Модель подтвердит выполнение всех трех действий.

Лучшие практики и обработка ошибок

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

• Изоляция ошибок: Если модель вызывает 5 функций, и одна падает с ошибкой, это не должно ломать весь процесс. Оборачивайте выполнение каждой функции в try/except. Возвращайте модели сообщение об ошибке в структурированном виде (например, {"error": "Timeout connecting to weather service"}). Модель часто способна понять ошибку и извиниться перед пользователем или попытаться снова.
• Идемпотентность: Убедитесь, что ваши функции безопасны для повторного вызова. В редких случаях сетевых сбоев модель может попытаться повторить вызов.
• Контекстное окно: Помните, что каждый вызов функции и ее результат добавляются в историю чата. Параллельный вызов 10 функций с объемными JSON-ответами может быстро съесть лимит токенов. Оптимизируйте возвращаемые данные — отдавайте только то, что нужно модели для ответа (например, только статус «OK», а не весь дамп базы данных).