Глава 9. Работа через API
Учебник «Большие языковые модели»
Для многих применений LLM — это внешний или собственный HTTP-сервис: приложение передаёт входные данные и параметры, получает события или JSON и учитывает токены. Но единого «LLM API» нет. Даже интерфейсы с одинаковыми названиями различаются по состоянию, инструментам, структурированному выводу, ошибкам и тарификации. В этой главе механика показана на официальном Python SDK OpenAI, а переносимость рассматривается как отдельное требование, которое надо проверять тестами.
9.1. Стили API и минимальный вызов
У OpenAI есть два основных интерфейса генерации: Responses API, рекомендуемый для новых проектов, и по-прежнему поддерживаемый Chat Completions API. Формат Chat Completions стал популярной целью совместимости для серверов открытых моделей; Anthropic использует отдельный Messages API. Совпадение базового маршрута ещё не означает поддержку одинаковых событий, схем, инструментов или параметров.
Минимальный вызов Responses API:
# pip install openai
import os
from openai import OpenAI
client = OpenAI() # читает OPENAI_API_KEY
MODEL = os.environ["OPENAI_MODEL"] # доступная вам модель
response = client.responses.create(
model=MODEL,
instructions="Ты редактор ИТ-новостей. Отвечай кратко.",
input=("Сформулируй заголовок: вышла новая версия PostgreSQL "
"с улучшенной репликацией."),
max_output_tokens=100,
)
if response.status != "completed":
raise RuntimeError(
f"ответ не завершён: {response.status}, {response.incomplete_details}")
print(response.output_text)
print(response.usage)
output_text — удобное агрегированное поле, но в рабочем приложении надо также обрабатывать отказ модели, ошибку запроса, пустой вывод и незавершённый ответ. Структура статуса и причин остановки зависит от API: универсального finish_reason для всех провайдеров нет.
Не менее важно разделить состояние модели и состояние сервиса. Каждый прямой проход видит только поданный или связанный контекст, но API может хранить ответы и продолжать цепочку по previous_response_id. Условия хранения и удаления конфигурируются отдельно, а токены предыдущего контекста могут снова учитываться в тарификации. Поле usage измеряет использование, но не всегда равно готовой сумме счёта: цены на обычные, кэшированные и reasoning-токены, инструменты и хранение различаются. Логируйте нормализованные счётчики вместе с моделью и актуальной таблицей тарифов, не записывая секреты и персональные данные.
9.2. Потоковая генерация
Для интерактивного интерфейса ответ удобно показывать по мере генерации. Responses API передаёт семантические события; текстовые дельты надо отличать от событий завершения, ошибок, инструментов и других типов вывода:
stream = client.responses.create(
model=MODEL,
input="Объясни, что такое SSI.",
stream=True,
)
parts = []
for event in stream:
if event.type == "response.output_text.delta":
parts.append(event.delta)
print(event.delta, end="", flush=True)
full_text = "".join(parts)
Код потока должен переживать разрыв после частичного ответа, иметь общий тайм-аут и не выдавать незавершённый текст за подтверждённый результат. Автоматически продолжать оборванный поток безопасно не всегда: второй вызов может повторить или изменить уже показанную часть. Для пакетной обработки обычный непотоковый ответ проще.
9.3. Структурированный вывод
Когда ответ читает программа, одной просьбы «верни JSON» мало. JSON mode обычно гарантирует только синтаксически корректный JSON; Structured Outputs связывает вывод со схемой. В SDK схему удобно задавать моделью Pydantic:
from typing import Literal
from pydantic import BaseModel, Field
class NewsItem(BaseModel):
title: str
category: Literal["БД", "ИИ", "безопасность", "прочее"]
importance: int = Field(ge=1, le=5)
news_text = "Вышла новая версия PostgreSQL с улучшенной репликацией."
response = client.responses.parse(
model=MODEL,
instructions="Размечай ИТ-новости по заданной схеме.",
input=news_text,
text_format=NewsItem,
)
if response.status != "completed" or response.output_parsed is None:
raise RuntimeError("нет завершённого структурированного результата")
item: NewsItem = response.output_parsed
print(item.model_dump())
Схема ограничивает структуру, но не доказывает истинность значений. Приложение всё равно проверяет предметные инварианты, права доступа и допустимость операции. Отказ, превышение лимита и ошибка API лежат вне успешного схемного ответа и требуют отдельной ветви обработки.
9.4. Вызов функций
Function calling — протокол, а не удалённое выполнение. Модель возвращает имя функции и JSON-аргументы; приложение проверяет запрос, исполняет разрешённую операцию и отправляет результат обратно. Responses API может вернуть несколько вызовов за один шаг, поэтому нельзя молча брать только первый:
import json
from pydantic import BaseModel
class ArticleArgs(BaseModel):
url: str
TOOLS = [{
"type": "function",
"name": "get_article_stats",
"description": "Возвращает статистику посещений статьи CITForum",
"parameters": {
"type": "object",
"properties": {"url": {"type": "string"}},
"required": ["url"],
"additionalProperties": False,
},
"strict": True,
}]
items = [{
"role": "user",
"content": "Сколько читали статью /news/2026/pg18.shtml?",
}]
response = client.responses.create(model=MODEL, input=items, tools=TOOLS)
items += response.output
called = False
for call in response.output:
if call.type != "function_call":
continue
called = True
if call.name != "get_article_stats":
raise ValueError(f"неразрешённый инструмент: {call.name}")
args = ArticleArgs.model_validate_json(call.arguments)
validate_and_authorize_article_path(args.url) # проверка приложения
result = get_article_stats(args.url) # исполняет ваш код
items.append({
"type": "function_call_output",
"call_id": call.call_id,
"output": json.dumps(result, ensure_ascii=False),
})
if called:
response = client.responses.create(
model=MODEL,
input=items,
tools=TOOLS,
)
print(response.output_text)
Аргументы и результат инструмента — недоверенные данные. Проверяйте JSON, имя функции, диапазоны, авторизацию пользователя и размер ответа. Операции с побочными эффектами требуют идемпотентного ключа и, для существенных действий, явного подтверждения человеком. Результат инструмента может содержать prompt injection; модель не должна получать из него новые права.
9.5. Устойчивость: ошибки, повторы и лимиты
Сетевые обрывы, ограничение частоты и временная перегрузка требуют ограниченных повторов с экспоненциальной выдержкой и jitter. Учитывайте Retry-After, если его прислал сервер, и повторяйте только ошибки, которые документация считает временными. Ошибки авторизации, схемы или размера контекста повтор сам не исправит.
Официальный SDK умеет выполнять базовые повторы и позволяет явно задать их число и тайм-аут:
from openai import OpenAI
client = OpenAI(
max_retries=5,
timeout=60.0,
)
Сверяйте поведение с документацией версии SDK. Повтор генерации может удвоить стоимость и вернуть другой текст; повтор инструмента способен дважды отправить письмо или провести платёж. Поэтому конвейер хранит идентификаторы заданий, отделяет генерацию от побочных эффектов и делает последние идемпотентными.
Запасная модель или провайдер повышают доступность лишь после интеграционных тестов. У них могут различаться роли, схемы, инструменты, политики хранения, фильтры безопасности и качество. Смена base_url не доказывает взаимозаменяемость.
9.6. Пакетная обработка
Для несрочных независимых заданий некоторые провайдеры предлагают batch API. Конкретный пример: OpenAI Batch API заявляет исполнение в пределах 24 часов и скидку 50% относительно синхронных вызовов. Это условия данного сервиса, а не отраслевой стандарт; лимиты, цены, поддерживаемые маршруты и формат результата надо проверять перед запуском.
Пакетный режим подходит для разметки архива, eval-прогонов и регулярной классификации. Входные строки должны иметь собственные идентификаторы: порядок результатов не следует считать гарантированным. Частичные ошибки обрабатывают поштучно, а повторный пакет не должен дублировать уже принятые результаты.
9.7. Итоги
API — контракт конкретного провайдера, а не единый стандарт. Проверяйте статус и неполный вывод, нормализуйте usage, обрабатывайте потоковые события, валидируйте структурированный результат и все вызовы функций. Повторы допустимы только для временных и безопасных операций; побочные эффекты требуют идемпотентности и авторизации. Пакетный режим, кэширование и выбор модели экономят деньги лишь по актуальным условиям сервиса. Совместимость и резервирование подтверждаются тестами.
Следующая глава — RAG: как подключить к модели большой корпус проверяемых внешних знаний.
Упражнения
1. Дополните первый пример нормализованным учётом входных, кэшированных, выходных и reasoning-токенов, которые действительно возвращает ваш провайдер. Посчитайте стоимость по актуальному тарифу и сохраните дату тарифа.
2. Проверьте потоковый вызов при принудительном сетевом обрыве. Убедитесь, что интерфейс помечает ответ как незавершённый, а повтор не приводит к двойному выполнению последующей операции.
3. Перенесите одну структурированную задачу на другой API. Составьте таблицу различий в ролях, схеме, событиях потока, usage, ошибках и хранении данных; закрепите её интеграционными тестами.