Как перенести вызов функций с Chat Completions на Responses API: безопасный Python-контур

Переход завершён не после замены /v1/chat/completions на /v1/responses, а после восстановления всего контура: Responses возвращает типизированный function_call, приложение проверяет имя и аргументы, выполняет операцию ровно один раз, отправляет function_call_output с тем же call_id и получает финальный ответ.

Первое действие — сохранить очищенную baseline-трассу одного рабочего Chat Completions-сценария: имя функции, аргументы, бизнес-результат, число реальных операций и ответ пользователю. Новый путь принимается только при совпадении бизнес-смысла, наличии результата для каждого call_id и отсутствии повторного side effect после искусственного retry. Если выбранный provider не подтверждает /v1/responses, typed Items и возврат function_call_output, остановите перенос: похожие поля не доказывают совместимость контрактов.

Переносите шесть договоров, а не один URL

Официальный migration guide OpenAI подчёркивает, что response.output может содержать не только message, но и reasoning, function_call и другие Items. Поэтому старый parser, который сразу читает только текст, способен пропустить работу приложения.

Функция в Responses задаётся без вложенного объекта function. Для strict-схемы ниже все поля обязательны, дополнительные свойства запрещены, а приложение повторно валидирует JSON до исполнения:

python
service_visit_tool = {
    "type": "function",
    "name": "schedule_service_visit",
    "description": "Schedule one technician visit for an existing service case.",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {
            "case_id": {"type": "string", "minLength": 1},
            "slot": {
                "type": "string",
                "enum": ["2026-07-30T10:00:00+03:00", "2026-07-30T14:00:00+03:00"],
            },
        },
        "required": ["case_id", "slot"],
        "additionalProperties": False,
    },
}

Не описывайте поведение strict как вечный default. Задавайте значение явно и проверяйте текущую схему на выбранной модели по официальной документации. Если задача требует только получить JSON по схеме, а не выполнить функцию приложения, сначала разделите эти контракты по сравнению Structured Outputs и function calling.

Python-контур: запись выезда без двойного бронирования

Этот пример планирует выезд сервисного инженера. Операция создаёт реальную запись, поэтому демонстрационный ledger отделяет call_id от стабильного OPERATION_ID. Неизвестная функция, плохой JSON, ошибка schema или handler не запускают следующий произвольный код: приложение возвращает структурированный error output с исходным call_id, и модель может корректно объяснить отказ.

Установите официальный SDK и задайте доступную модель с поддержкой function calling:

bash
python -m pip install -U openai
export OPENAI_API_KEY="project-api-key"
export OPENAI_MODEL="your-tool-capable-model"
export OPERATION_ID="visit-request-20260728-0042"
python responses_service_visit.py

Сохраните код как responses_service_visit.py:

python
import hashlib
import json
import os
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone

from openai import OpenAI


client = OpenAI()
model = os.environ.get("OPENAI_MODEL")
operation_id = os.environ.get("OPERATION_ID")

if not model or not operation_id:
    raise RuntimeError("Set OPENAI_MODEL and a stable OPERATION_ID.")

TOOLS = [
    {
        "type": "function",
        "name": "schedule_service_visit",
        "description": "Schedule one technician visit for an existing service case.",
        "strict": True,
        "parameters": {
            "type": "object",
            "properties": {
                "case_id": {"type": "string", "minLength": 1},
                "slot": {
                    "type": "string",
                    "enum": [
                        "2026-07-30T10:00:00+03:00",
                        "2026-07-30T14:00:00+03:00",
                    ],
                },
            },
            "required": ["case_id", "slot"],
            "additionalProperties": False,
        },
    }
]

ALLOWED_SLOTS = {
    "2026-07-30T10:00:00+03:00",
    "2026-07-30T14:00:00+03:00",
}
VISITS = {}
EXECUTION_LEDGER = {}
LEDGER_LOCK = threading.Lock()


def validate_arguments(raw_arguments):
    try:
        value = json.loads(raw_arguments)
    except json.JSONDecodeError as exc:
        raise ValueError("arguments_not_json") from exc

    if not isinstance(value, dict) or set(value) != {"case_id", "slot"}:
        raise ValueError("arguments_schema_mismatch")
    if not isinstance(value["case_id"], str) or not value["case_id"].strip():
        raise ValueError("invalid_case_id")
    if value["slot"] not in ALLOWED_SLOTS:
        raise ValueError("unsupported_slot")
    return value


def business_key(function_name, arguments):
    canonical = json.dumps(
        {
            "operation_id": operation_id,
            "function": function_name,
            "case_id": arguments["case_id"],
            "slot": arguments["slot"],
        },
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def schedule_service_visit(arguments):
    visit_id = f"{operation_id}:{arguments['case_id']}"
    VISITS[visit_id] = {
        "visit_id": visit_id,
        "case_id": arguments["case_id"],
        "slot": arguments["slot"],
        "status": "scheduled",
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    return VISITS[visit_id]


HANDLERS = {"schedule_service_visit": schedule_service_visit}


def error_output(call_id, code):
    return {
        "type": "function_call_output",
        "call_id": call_id,
        "output": json.dumps({"ok": False, "error": code}, ensure_ascii=False),
    }


def execute_call(call):
    if not call.call_id:
        raise RuntimeError("Protocol error: function_call has no call_id.")

    handler = HANDLERS.get(call.name)
    if handler is None:
        return error_output(call.call_id, "unknown_function")

    try:
        arguments = validate_arguments(call.arguments)
    except ValueError as exc:
        return error_output(call.call_id, str(exc))

    key = business_key(call.name, arguments)
    try:
        # Демо делает check + side effect + ledger одной критической секцией.
        # В production это должна быть DB-транзакция или transactional outbox.
        with LEDGER_LOCK:
            if key in EXECUTION_LEDGER:
                payload = EXECUTION_LEDGER[key]
            else:
                payload = {"ok": True, "data": handler(arguments)}
                EXECUTION_LEDGER[key] = payload
    except Exception:
        return error_output(call.call_id, "tool_execution_failed")

    return {
        "type": "function_call_output",
        "call_id": call.call_id,
        "output": json.dumps(payload, ensure_ascii=False),
    }


instructions = (
    "Вы записываете выезд сервисного инженера. "
    "Перед подтверждением вызовите schedule_service_visit. "
    "Не выдумывайте номер заявки или время. "
    "Если tool output содержит ok=false, объясните отказ без повторного выполнения."
)

input_items = [
    {
        "role": "user",
        "content": (
            "Запишите инженера по заявке CASE-1842 "
            "на 30 июля 2026 года в 14:00 по Москве."
        ),
    }
]

final_text = None

for round_number in range(1, 7):
    response = client.responses.create(
        model=model,
        instructions=instructions,
        input=input_items,
        tools=TOOLS,
        parallel_tool_calls=True,
        store=False,
    )

    if response.status == "incomplete":
        details = getattr(response, "incomplete_details", None)
        raise RuntimeError(
            f"Trace {response.id}: incomplete response; "
            f"do not accept partial text. Details: {details!r}"
        )

    # Manual replay сохраняет все typed Items, включая reasoning Items.
    input_items.extend(response.output)
    calls = [item for item in response.output if item.type == "function_call"]

    if not calls:
        if not response.output_text:
            raise RuntimeError(
                f"Trace {response.id}: no function calls and no final text."
            )
        final_text = response.output_text
        break

    # Весь пакет текущего раунда завершается до следующего Responses-запроса.
    with ThreadPoolExecutor(max_workers=min(4, len(calls))) as pool:
        outputs = list(pool.map(execute_call, calls))
    input_items.extend(outputs)

if final_text is None:
    raise RuntimeError(
        f"Tool loop exceeded 6 rounds; inspect the last trace before retry."
    )

print(final_text)

Код возвращает ошибки модели, но это не разрешение бесконечно повторять функцию: цикл ограничен шестью раундами. Отсутствующий call_id — нарушение протокола, поэтому continuation безопасно прекращается. Для неизвестного имени, плохих аргументов и исключения handler формируется function_call_output с тем же ID и стабильным кодом ошибки.

ThreadPoolExecutor принимает весь пакет вызовов, но корреляция остаётся по call_id, а не по порядку завершения. Демонстрационный lock делает side effect и запись ledger атомарными внутри процесса. В production замените его уникальным индексом по business key и транзакцией, которая одновременно изменяет ресурс и фиксирует результат; для внешнего сервиса используйте transactional outbox и reconciliation. Новый запрос модели может получить новый call_id, поэтому только call_id не защищает от повторной бизнес-операции.

Function Calling guide OpenAI фиксирует границу ответственности: модель предлагает вызов, а allowlist, авторизация, проверка аргументов, выполнение и retry принадлежат приложению.

State выбирают по требованиям к данным

Conversation state guide остаётся источником текущего поведения. Responses по умолчанию хранит ответы, если не задано store: false. Цепочка с previous_response_id не делает прошлые input tokens бесплатными, а прежние top-level instructions автоматически не наследуются.

Streaming переносите отдельным изменением

Сначала добейтесь стабильного non-streaming контура. Затем замените Chat Completions parser на обработчик typed SSE events. Для функции он должен накапливать response.function_call_arguments.delta отдельно по Item, разбирать JSON только после response.function_call_arguments.done и различать response.completed, response.failed и response.incomplete.

Если соединение оборвалось до done, не выполняйте функцию. Если side effect завершился, а поток финального ответа оборвался, сначала прочитайте durable ledger и реальное состояние заявки; слепой retry недопустим. Текущие имена событий сверяйте с официальным streaming guide.

Provider-граница для русскоязычной команды

Пример выше описывает direct OpenAI SDK contract. Azure OpenAI использует собственные base URL, authentication и deployment names; это отдельно показано в руководстве Microsoft RU по миграции на Responses API. Yandex AI Studio и совместимые gateways также имеют собственные модели, billing, storage и набор поддерживаемых событий. Каждый маршрут должен пройти тот же staging harness; документация с похожим названием не заменяет реальный probe.

Российская Google SERP в исследовании была открыта с hl=ru&gl=RU, но физический регион захвата остался неизвестен. Поэтому статья не делает вывод о подтверждённом локальном спросе или доступности direct OpenAI. Регион и условия аккаунта проверяйте через русское руководство по API-ключу и официальной доступности, не обходя ограничения.

Приёмка и обратимый rollout

Сохраните обе очищенные трассы и требуйте выполнения каждого пункта:

• Responses получает плоскую schema с ожидаемым name, parameters и явным strict.
• Parser читает каждый output Item по type, а не только output_text.
• Неизвестная функция и плохие аргументы дают структурированный error output с исходным call_id.
• Для каждого принятого function_call есть ровно один результат с тем же call_id.
• Fixture с двумя calls возвращает оба результата до следующего Responses-раунда.
• Retry после имитации timeout создаёт одну сервисную запись благодаря стабильному business key.
• Tool exception не вызывает динамический handler и не запускает бесконечный retry.
• После tool outputs появляется непустой финальный ответ или явная trace-ошибка.
• Выбран ровно один state contract; при previous_response_id инструкции отправляются снова.
• Streaming fixture различает done, completed, failed, incomplete и разрыв соединения.
• Логи содержат response ID, call_id, operation ID и статус без API key и чувствительных аргументов.
• Feature flag возвращает старый Chat Completions-путь, не повторяя уже завершённый side effect.

Если во время rollout появляется 429, отделите исчерпанную квоту от rate limit по русскому разбору OpenAI API quota. Следующий практический шаг: выполнить Python-контур в staging с фиксированным OPERATION_ID, дважды повторить один запрос, проверить одну запись в ledger и только затем включить небольшую долю внутреннего трафика.