Проверка ответа Chat Completions в Python до обработки текста

Сначала ограничьте поддерживаемый контракт

Пример рассчитан на нестриминговый Chat Completions, n=1, object=chat.completion и единственный choice с index=0. Он не обрабатывает SSE chunks, Responses output, аудио или произвольные расширения OpenAI-совместимых провайдеров. Отказ от неизвестного формата здесь намеренный: лучше явная ошибка, чем тихая подстановка пустой строки. Если ваше приложение запрашивает несколько choices или другой протокол, измените контракт и fixtures одновременно. Не переименовывайте поле output в choices и не принимайте незнакомый объект только потому, что внутри нашлась строка content.

Текст, отказ и вызов инструмента — разные результаты

В схеме Chat Completions message может содержать отказ или tool_calls, а finish_reason сообщает причину окончания генерации. При length результат мог быть обрезан, при content_filter — ограничен фильтром. Такие результаты не надо показывать как гарантированно завершённый ответ. Классификатор возвращает text, refusal, tool_calls или incomplete. Ошибки HTTP и структуры JSON представлены ResponseContractError с коротким кодом без тела ответа. Для tool_calls проверяется только форма вызова функции. Строка arguments ещё не прошла JSON Schema, проверку прав и бизнес-валидацию. Даже корректный JSON не является разрешением исполнять действие.

Классификатор с ограничением размера тела

Сохраните код в chat_response.py. Передайте ему HTTP-статус и bytes тела ответа. Ограничение в один мегабайт защищает разбор уже полученного буфера; реальный HTTP-адаптер должен дополнительно ограничивать чтение до загрузки всего тела. Классификатор не записывает текст ответа или промпт в лог. Вызовы функций возвращаются как данные. Поддержка ограничена type=function; неизвестный тип инструмента отклоняется. Устаревшее function_call не преобразуется молча в современную структуру. Для отображения текста и refusal всё равно требуется штатное экранирование интерфейса.

"""A deliberately narrow non-streaming Chat Completions contract."""
import json


class ResponseContractError(ValueError):
    pass


def classify_response(status, raw):
    if type(status) is not int or not 200 <= status < 300:
        raise ResponseContractError("http_status")
    if not isinstance(raw, bytes) or len(raw) > 1024 * 1024:
        raise ResponseContractError("body_limit")
    try:
        data = json.loads(raw)
    except (ValueError, UnicodeError, RecursionError):
        raise ResponseContractError("invalid_json") from None
    if not isinstance(data, dict) or data.get("object") != "chat.completion":
        raise ResponseContractError("wrong_protocol")
    choices = data.get("choices")
    if not isinstance(choices, list) or len(choices) != 1:
        raise ResponseContractError("expected_one_choice")
    choice = choices[0]
    if (not isinstance(choice, dict) or type(choice.get("index")) is not int
            or choice["index"] != 0):
        raise ResponseContractError("invalid_choice")
    message = choice.get("message")
    if not isinstance(message, dict) or message.get("role") != "assistant":
        raise ResponseContractError("invalid_message")
    reason = choice.get("finish_reason")
    if not isinstance(reason, str):
        raise ResponseContractError("invalid_finish_reason")
    if reason in {"length", "content_filter"}:
        return {"kind": "incomplete", "reason": reason}
    refusal, calls = message.get("refusal"), message.get("tool_calls")
    if refusal is not None:
        if not isinstance(refusal, str) or not refusal.strip() or calls:
            raise ResponseContractError("invalid_refusal")
        if reason != "stop":
            raise ResponseContractError("invalid_finish_reason")
        return {"kind": "refusal", "value": refusal}
    if reason == "tool_calls":
        if not isinstance(calls, list) or not calls:
            raise ResponseContractError("invalid_tool_calls")
        ids = set()
        for call in calls:
            if not isinstance(call, dict) or call.get("type") != "function":
                raise ResponseContractError("unsupported_tool")
            function = call.get("function")
            if (not isinstance(call.get("id"), str) or not call["id"]
                    or call["id"] in ids or not isinstance(function, dict)
                    or not isinstance(function.get("name"), str)
                    or not function["name"]
                    or not isinstance(function.get("arguments"), str)):
                raise ResponseContractError("invalid_function_call")
            ids.add(call["id"])
        # Shape only. Never execute the model's function name/arguments here.
        return {"kind": "tool_calls", "value": calls}
    if reason != "stop" or calls:
        raise ResponseContractError("unsupported_finish_reason")
    content = message.get("content")
    if not isinstance(content, str) or not content.strip():
        raise ResponseContractError("missing_text")
    return {"kind": "text", "value": content}

Fixtures: текст, отказ, tools и HTTP-ошибка

Ниже минимальный воспроизводимый тест. Вызов функции специально содержит arguments, которые ещё не являются корректным JSON: классификация формы и разрешение исполнения — разные этапы. HTTP 429 проверяется до JSON-разбора, поэтому повреждённое тело не скрывает статус ошибки. Полный набор тестов проекта дополнительно проверяет пустые choices, неизвестный object, неверные типы и ограничения размера.

import json
from chat_response import classify_response, ResponseContractError

def fixture(message, reason="stop"):
    return json.dumps({"object": "chat.completion", "choices": [{
        "index": 0, "finish_reason": reason, "message": message
    }]}).encode()

assert classify_response(200, fixture({"role": "assistant", "content": "Готово"}))["kind"] == "text"
assert classify_response(200, fixture({"role": "assistant", "content": None, "refusal": "Не могу помочь"}))["kind"] == "refusal"
tool = {"id": "call_demo", "type": "function", "function": {"name": "weather", "arguments": "needs validation"}}
assert classify_response(200, fixture({"role": "assistant", "tool_calls": [tool]}, "tool_calls"))["kind"] == "tool_calls"
assert classify_response(200, fixture({"role": "assistant", "content": "часть"}, "length"))["kind"] == "incomplete"
try:
    classify_response(429, b"not JSON")
except ResponseContractError as error:
    assert str(error) == "http_status"
else:
    raise AssertionError("HTTP error was accepted")
print("PASS: distinct response outcomes")

Что должно произойти после классификации

Для text приложение выбирает безопасный способ отображения; для refusal показывает отказ без повторного запуска того же действия. Для incomplete можно предложить отдельное продолжение, но нельзя выдавать частичный JSON за завершённую структуру. Для tool_calls сначала проверяются allowlist функций, идентификаторы, схема аргументов и полномочия пользователя. Побочные эффекты требуют собственной защиты от дублей. События наблюдаемости могут содержать HTTP-статус, короткий код нарушения контракта и внутренний correlation id. Не добавляйте полный response body ради удобства диагностики: в нём может оказаться пользовательский контент. Обработка ошибки структуры не должна автоматически запускать второй платный запрос.

Что эти тесты подтверждают, а что нет

Fixtures доказывают поведение конкретной функции на заданных входах. Они не подтверждают доступность модели, качество ответа, стоимость, поддержку tools каждой моделью или полную совместимость любого API-шлюза. Зафиксируйте поддерживаемую версию клиентского контракта и добавляйте новые ветки только с отдельными тестами. Первую проверку настоящего HTTP-адаптера проводите отдельно от этих офлайн-тестов: с явно выбранной моделью, ограниченным бюджетом и безопасной диагностикой. Здесь live-проверка не выполнялась. Сравнение протоколов и настройка SDK остаются в связанных статьях блога, а не дублируются в этом гайде.