YandexGPT Function Calling: полный гайд на Python
Как интегрировать YandexGPT tool calling в Python приложения. Разбор подводных камней и best practices из реального production-проекта.
Представьте: оператор пишет в чат «проанализируй системный лог». Модель может ответить общими фразами про «проверьте ошибки», а может реально вызвать функцию analyze_system_logs(), получить Markdown-отчёт и вернуть его пользователю. Второе — это function calling. И сегодня разберём как это работает в YandexGPT 5.1, на примере нашего production-проекта SCADA.AI.
Что вообще такое function calling
Function calling (или tool calling) — это механизм, который позволяет LLM не просто генерировать текст, а вызывать внешние функции, которые вы описали. Модель сама решает, когда позвать функцию, с какими аргументами, а вы обрабатываете результат и возвращаете его обратно в LLM.
Схема выглядит так:
Пользователь: "проанализируй лог"
↓
LLM: "нужна функция analyze_system_logs"
↓
LLM возвращает JSON с именем функции и аргументами
↓
Ваш код выполняет функцию → получает результат
↓
Результат отправляется обратно в LLM
↓
LLM: "## Анализ лога: обнаружено 3 предупреждения..."Это не просто «красивая фича» — это то, что превращает чат-бота в реального ассистента, который умеет делать, а не только говорить.
Шаг 1. Описываем инструменты
YandexGPT принимает описание инструментов в формате JSON Schema:
tools = [{
"function": {
"name": "analyze_system_logs",
"description": (
"Анализирует системный лог SCADA.AI. "
"Вызывается когда пользователь просит 'проанализируй лог', "
"'что происходит в системе', 'анализ логов'."
),
"parameters": {
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Количество последних записей (по умолчанию 500)"
}
}
}
}
}]Несколько важных моментов из практики:
description— это инструкция для LLM. Модель буквально читает это описание, чтобы понять, когда вызывать функцию. Если написать сухо «анализирует лог», модель может проигнорировать tool. Пишите триггеры — конкретные фразы пользователя, на которые функция должна реагировать.Имена параметров должны быть осмысленными.
limitлучше чемn,sensor_idлучше чемid. Модель их читает.Не делайте слишком много tools за раз. 5–7 инструментов — потолок. При большем количестве модель начинает путаться.
Шаг 2. Отправляем запрос в YandexGPT
Используем httpx (асинхронный HTTP-клиент):
import httpx
async def call_yandexgpt(messages: list[dict], tools: list[dict] | None = None):
payload = {
"modelUri": f"gpt://{FOLDER_ID}/yandexgpt/latest",
"completionOptions": {
"stream": False,
"temperature": 0.6,
"maxTokens": 8000
},
"messages": messages
}
if tools:
payload["tools"] = tools
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
"https://llm.api.cloud.yandex.net/foundationModels/v1/completion",
headers={
"Authorization": f"Api-Key {API_KEY}",
"Content-Type": "application/json"
},
json=payload
)
response.raise_for_status()
return response.json()Шаг 3. Обрабатываем ответ (тут начинаются грабли)
YandexGPT 5.1 использует новый формат — toolCallList. Это важно, потому что в старых версиях был functionCall напрямую в message, и многие примеры в интернете показывают устаревший синтаксис.
data = await call_yandexgpt(messages, tools)
alternatives = data["result"]["alternatives"]
message = alternatives[0]["message"]
status = alternatives[0]["status"]
if status == "ALTERNATIVE_STATUS_TOOL_CALLS":
# Модель хочет вызвать функцию
tool_calls = message.get("toolCallList", {}).get("toolCalls", [])
if tool_calls:
function_call = tool_calls[0]["functionCall"]
name = function_call["name"] # "analyze_system_logs"
args = function_call["arguments"] # {} или {"limit": 500}
print(f"Модель просит вызвать: {name}({args})")
else:
# Финальный текстовый ответ
return message.get("text", "")Первый подводный камень: если вы используете пример из старого туториала и проверяете message.get("functionCall"), вы никогда не увидите вызов tool. В новой версии это message.toolCallList.toolCalls[].functionCall.
Шаг 4. Выполняем функцию и возвращаем результат
После того как модель запросила вызов, мы:
- Выполняем функцию с переданными аргументами
- Формируем сообщение с результатом
- Отправляем второй запрос — уже с полной историей
import json
# 1. Выполняем функцию
result = await execute_tool(name, args)
# 2. Формируем историю для второго запроса
messages = [
{"role": "system", "text": system_prompt},
{"role": "user", "text": user_message},
# Ответ модели с запросом tool
{"role": "assistant", "toolCallList": message["toolCallList"]},
# Наш ответ с результатом
{"role": "user", "toolResultList": {
"toolResults": [{
"functionResult": {
"name": name,
"content": json.dumps(result, ensure_ascii=False)
}
}]
}}
]
# 3. Второй запрос → финальный ответ
final_data = await call_yandexgpt(messages, tools)
final_text = final_data["result"]["alternatives"][0]["message"]["text"]
return final_textВторой подводный камень, самый болезненный: обратите внимание на структуру ответа tool. Должно быть:
role: "user"— не"assistant"и не"function"toolResultList.toolResults[].functionResult— неfunctionResponsecontent— просто строка, а не объект сcontentType
Если сделаете любую из трёх ошибок — получите tool result type not specified или invalid message role. Мы потратили на это полдня, пока не нашли правильную структуру в официальной документации.
Полный pipeline: цикл обработки
В production система может вызывать несколько инструментов подряд. Например, пользователь попросил «сравни вчера и сегодня по температуре» — модель сначала вызовет get_data("yesterday"), потом get_data("today"), потом compare(). Поэтому нужен цикл с ограничением итераций:
async def process_chat(user_message: str, system_prompt: str, tools: list[dict]):
messages = [
{"role": "system", "text": system_prompt},
{"role": "user", "text": user_message}
]
for iteration in range(5): # защита от бесконечного цикла
data = await call_yandexgpt(messages, tools)
alt = data["result"]["alternatives"][0]
status = alt["status"]
message = alt["message"]
# Финальный ответ — выходим
if status == "ALTERNATIVE_STATUS_FINAL":
return message.get("text", "")
# Модель просит tool
if status == "ALTERNATIVE_STATUS_TOOL_CALLS":
tool_call_list = message["toolCallList"]
tool_calls = tool_call_list.get("toolCalls", [])
# Добавляем сообщение ассистента в историю
messages.append({"role": "assistant", "toolCallList": tool_call_list})
# Выполняем все запрошенные tools
tool_results = []
for tool_call in tool_calls:
func = tool_call["functionCall"]
name = func["name"]
args = func["arguments"]
try:
result = await execute_tool(name, args)
content = json.dumps(result, ensure_ascii=False, default=str)
except Exception as e:
content = json.dumps({"error": str(e)})
tool_results.append({
"functionResult": {
"name": name,
"content": content
}
})
# Добавляем результаты в историю
messages.append({
"role": "user",
"toolResultList": {"toolResults": tool_results}
})
continue
return "Превышено максимальное количество итераций"Паттерн Tool Executor
Удобно вынести регистрацию и вызов функций в отдельный класс. Это даёт централизованное логирование, обработку ошибок и чистую архитектуру:
class ToolExecutor:
def __init__(self):
self._tools: dict[str, Callable] = {}
def register(self, name: str, func: Callable):
"""Регистрирует функцию как доступный инструмент"""
self._tools[name] = func
async def execute(self, name: str, args: dict):
"""Выполняет функцию по имени"""
if name not in self._tools:
raise ValueError(f"Неизвестный инструмент: {name}")
log.info("Tool executing", tool=name, args=args)
result = await self._tools[name](**args)
log.info("Tool completed", tool=name)
return result
# Использование
executor = ToolExecutor()
from modules.logs.tools import analyze_system_logs
from modules.weather.tools import get_weather
executor.register("analyze_system_logs", analyze_system_logs)
executor.register("get_weather", get_weather)
# В цикле обработки
result = await executor.execute(name, args)Такой подход позволяет каждому модулю системы регистрировать свои инструменты независимо. Модуль logs регистрирует функции для работы с логами, модуль weather — для погоды, и так далее.
Подводные камни из реального опыта
За время разработки SCADA.AI мы собрали коллекцию граблей. Делюсь самыми болезненными:
1. Пустой ответ от модели
Симптом: {"text": "", "functionCall": null} при ALTERNATIVE_STATUS_FINAL.
Причина: обычно это либо отсутствие system message в истории, либо слишком расплывчатое описание tool. Модель не понимает, что ей делать.
Решение: всегда передавайте осмысленный system prompt, где явно сказано: «когда пользователь просит X — вызывай tool Y».
2. Модель не вызывает tool, а отвечает сама
Симптом: пользователь пишет «проанализируй лог», а модель выдаёт общий текст без вызова функции.
Причина: плохое описание функции. Если description слишком короткое, модель решает что может ответить сама.
Решение: добавьте в описание конкретные триггеры:
"Вызывается когда пользователь просит 'проанализируй лог',
'анализ логов', 'что происходит в системе'."3. Бесконечный цикл tool calls
Симптом: модель вызывает одну и ту же функцию снова и снова.
Причина: функция возвращает результат, который модель не может интерпретировать. Например, пустой список без объяснения, почему он пуст.
Решение: возвращайте из tool не просто данные, а структурированный ответ с пояснениями:
return {
"status": "ok",
"count": len(logs),
"analysis": "...",
"note": "Если count == 0, лог пуст — сообщи об этом пользователю"
}4. Ошибки в tool убивают весь pipeline
Симптом: исключение в функции → весь чат падает.
Решение: ловите все исключения внутри tool и возвращайте их как часть ответа:
async def analyze_system_logs(limit: int = 500):
try:
logs = await read_logs(limit)
analysis = await llm_analyze(logs)
return {"status": "ok", "analysis": analysis}
except Exception as e:
log.error("Tool failed", error=str(e))
return {"status": "error", "error": str(e)}LLM увидит ошибку и сообщит пользователю о проблеме, а не зависнет.
Когда использовать function calling
Используйте когда:
- Нужно получить реальные данные из БД, API, файлов
- Требуется выполнить действие — отправить уведомление, создать тикет, перезапустить сервис
- Пользователь хочет анализ конкретных данных, а не общие рассуждения
- Нужна интеграция с внешними сервисами
Не используйте когда:
- Вопрос общий («что такое SCADA?», «расскажи про IoT»)
- Нужен креативный ответ — написать стих, сочинить историю
- Модель уверенно отвечает на основе обучения
- Объём данных слишком большой для контекстного окна (тут нужен RAG)
Best practices на каждый день
Логируйте всё. Когда что-то пойдёт не так (а оно пойдёт), логи спасут часы отладки:
log.info("LLM calls tool", tool=name, args=args)
log.info("Tool executed", tool=name, status="ok")
log.error("Tool failed", tool=name, error=str(e))Не передавайте сырые данные в LLM. Если у вас 50 000 строк лога, сначала агрегируйте — посчитайте статистику, выделите аномалии, сгруппируйте. LLM работает с осмысленными summary, а не с сырыми точками.
Тестируйте tools отдельно. Прежде чем интегрировать в чат, убедитесь что функция работает сама по себе с разными входными данными.
Начинайте с 1-2 tools. Не пытайтесь сразу построить «супер-ассистента с 20 функциями». Добавьте одну, отладьте, потом следующую.
Итог
YandexGPT Function Calling — мощный инструмент для построения AI-ассистентов, которые умеют делать, а не только говорить. Да, API имеет свои особенности — новый формат toolCallList, специфичные роли сообщений, строковый content — но разобравшись один раз, вы получаете production-ready решение.
В SCADA.AI мы используем function calling для анализа системных логов, получения данных с датчиков, сравнения исторических периодов и генерации отчётов. Результат: операторы получают ответы на вопросы типа «что происходит в системе?» за секунды, вместо ручного чтения сотен строк лога.
Главный совет: не пытайтесь сделать идеально с первого раза. Начните с одного tool, пройдите все грабли на нём, потом масштабируйте.
Код из статьи доступен в открытом репозитории SCADA.AI.