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

Миграция вызова функций с Chat Completions на Responses API завершена, когда приложение получает function_call, проверяет аргументы, выполняет разрешённую операцию, возвращает function_call_output с исходным call_id и получает завершённый ответ. Замены URL недостаточно: меняются описание функции, формат истории и разбор результата. Пользовательская функция по-прежнему выполняется вашим приложением, а модель предлагает её вызов.

Ниже — полный Python-контур с конкретной операцией: запись сервисного инженера в локальную SQLite-базу. Он обрабатывает весь пакет вызовов, возвращает понятные ошибки и не создаёт вторую запись при повторе одного бизнес-запроса. Это демонстрационная база, не подключение к реальной диспетчерской. Синтаксис и поведение проверены локально на искусственных ответах; запросы к OpenAI, совместимость установленного SDK и доступ к модели здесь не проверялись.

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

Chat Completions остаётся поддерживаемым API. Для новых проектов OpenAI рекомендует Responses; при переносе существующего приложения важно сохранить его результат и ограничения. Официальное руководство по миграции.

Перебирайте все элементы output: среди сообщений и элементов рассуждения может быть ноль, один или несколько вызовов. item.id и call_id имеют разные назначения; связывать результат с вызовом по позиции массива нельзя. После возврата результатов модель может дать текст или запросить следующую функцию. Руководство по Function Calling.

В схеме примера явно задано strict: True: все свойства перечислены в required, дополнительные поля запрещены. Если строгость не указана, Responses пытается привести схему к строгому режиму, а при невозможности может использовать best effort с strict: false. Поэтому отсутствие strict не означает ни безусловную строгость, ни обязательную ошибку 400. Проверка схемы не заменяет проверку прав и бизнес-условий в обработчике. Текущие правила strict mode.

Если нужен только JSON-ответ заданной формы, без работы приложения, это другая задача: см. когда выбирать Structured Outputs, а когда Function Calling.

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

Сохраните следующий файл как responses_visit.py. Для настоящего запроса понадобится официальный пакет openai, уже настроенный серверный ключ и доступная модель с поддержкой функций через Responses. Модель указывается явно: перенос API сам по себе не требует смены модели. Если старый запрос задавал уровень рассуждения, передайте совместимое значение через --effort; при отсутствии этой настройки пример использует выбор API. Инструкция по получению API-ключа и доступности официального сервиса относится к подготовке подключения.

Пример делает все шаги цикла, включая диспетчеризацию и запись в базе. account_id, разрешённые заявки и request_id поступают из приложения; в рабочей системе их нужно получать из аутентифицированного запроса и проверки прав. Аргументы модели не могут расширить этот набор. Указанный в коде demo-account — только демонстрационный контекст.

import argparse
import json
import sqlite3
import time

SLOTS = ("2026-10-12T10:00:00+03:00", "2026-10-12T14:00:00+03:00")
TOOLS = [{
    "type": "function", "name": "book_visit",
    "description": "Записать инженера по разрешённой сервисной заявке.",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {
            "case_id": {"type": "string"},
            "slot": {"type": "string", "enum": list(SLOTS)},
        },
        "required": ["case_id", "slot"], "additionalProperties": False,
    },
}]
INSTRUCTIONS = (
    "Вы помогаете записать инженера. Подтверждайте запись только после "
    "book_visit с ok=true. При ok=false объясните отказ. "
    "Не придумывайте номер записи и не меняйте время без запроса пользователя."
)


def reject_constant(value):
    raise ValueError("invalid_json_constant")


def unique_keys(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise ValueError("duplicate_json_key")
        value[key] = item
    return value


def arguments(raw):
    try:
        value = json.loads(raw, object_pairs_hook=unique_keys,
                           parse_constant=reject_constant)
    except (ValueError, TypeError):
        raise ValueError("invalid_arguments_json") from None
    if not isinstance(value, dict) or set(value) != {"case_id", "slot"}:
        raise ValueError("invalid_arguments_shape")
    if not isinstance(value["case_id"], str) or not value["case_id"].strip():
        raise ValueError("invalid_case_id")
    if not isinstance(value["slot"], str) or value["slot"] not in SLOTS:
        raise ValueError("unsupported_slot")
    return value


class VisitApp:
    def __init__(self, db, account_id, request_id, allowed_cases):
        if not account_id or not request_id:
            raise ValueError("missing_trusted_request_context")
        self.db = db
        self.account_id = account_id
        self.request_id = request_id
        self.allowed_cases = frozenset(allowed_cases)
        self.db.execute("""CREATE TABLE IF NOT EXISTS visits (
            account_id TEXT NOT NULL, request_id TEXT NOT NULL,
            case_id TEXT NOT NULL, slot TEXT NOT NULL, result TEXT NOT NULL,
            PRIMARY KEY (account_id, request_id))""")
        self.db.commit()

    def book_visit(self, value):
        if value["case_id"] not in self.allowed_cases:
            raise ValueError("case_not_authorized")
        try:
            self.db.execute("BEGIN IMMEDIATE")
            row = self.db.execute(
                "SELECT case_id, slot, result FROM visits "
                "WHERE account_id=? AND request_id=?",
                (self.account_id, self.request_id),
            ).fetchone()
            if row is not None:
                if row[:2] != (value["case_id"], value["slot"]):
                    raise ValueError("request_id_conflict")
                result = json.loads(row[2])
            else:
                result = {"ok": True, "visit": {
                    "id": "visit:" + self.request_id,
                    "case_id": value["case_id"], "slot": value["slot"],
                    "status": "scheduled",
                }}
                self.db.execute(
                    "INSERT INTO visits VALUES (?, ?, ?, ?, ?)",
                    (self.account_id, self.request_id, value["case_id"],
                     value["slot"], json.dumps(result, ensure_ascii=False)),
                )
            self.db.commit()
            return result
        except Exception:
            self.db.rollback()
            raise

    def recorded_result(self):
        row = self.db.execute(
            "SELECT result FROM visits WHERE account_id=? AND request_id=?",
            (self.account_id, self.request_id),
        ).fetchone()
        return json.loads(row[0]) if row else None


def dispatch(call, app):
    # Имя выбирает только явно разрешённый обработчик.
    handlers = {"book_visit": app.book_visit}
    handler = handlers.get(call.name)
    if handler is None:
        payload = {"ok": False, "error": "unknown_function"}
    else:
        try:
            payload = handler(arguments(call.arguments))
        except ValueError as exc:
            payload = {"ok": False, "error": str(exc)}
        except Exception:
            payload = {"ok": False, "error": "local_tool_failed"}
    return {"type": "function_call_output", "call_id": call.call_id,
            "output": json.dumps(payload, ensure_ascii=False)}


def complete_text(items):
    texts = []
    for item in items:
        if item.type != "message":
            continue
        for part in item.content:
            if part.type == "refusal":
                raise RuntimeError("model_refusal")
            if part.type == "output_text" and getattr(item, "phase", None) != "commentary":
                texts.append(part.text)
    text = "\n".join(texts).strip()
    if not text:
        raise RuntimeError("no_final_text")
    return text


def run_loop(create_response, model, app, user_text, max_rounds=6, effort=None):
    history = [{"role": "user", "content": user_text}]
    completed_calls = {}
    deadline = time.monotonic() + 120
    for _ in range(max_rounds):
        if time.monotonic() >= deadline:
            raise RuntimeError("application_deadline")
        request = dict(
            model=model, instructions=INSTRUCTIONS, input=history,
            tools=TOOLS, parallel_tool_calls=True, store=False,
            max_output_tokens=1600,
        )
        if effort is not None:
            request["reasoning"] = {"effort": effort}
        response = create_response(**request)
        if response.status != "completed":
            details = getattr(response, "incomplete_details", None)
            error = getattr(response, "error", None)
            raise RuntimeError(
                f"response_not_completed:{response.id}:{response.status}:"
                f"{details!r}:{error!r}"
            )
        calls = [item for item in response.output if item.type == "function_call"]
        # Проверка всего пакета до первой записи в БД.
        batch = {}
        for call in calls:
            if (not isinstance(call.call_id, str) or not call.call_id
                    or not isinstance(call.name, str)
                    or not isinstance(call.arguments, str)):
                raise RuntimeError("invalid_function_call")
            if (getattr(call, "status", None) not in (None, "completed")
                    or getattr(call, "async", False)
                    or getattr(call, "namespace", None)
                    or getattr(call, "caller", None)):
                raise RuntimeError("unsupported_function_variant")
            signature = (call.name, call.arguments)
            prior = completed_calls.get(call.call_id)
            if (prior and prior[0] != signature) or (
                call.call_id in batch and batch[call.call_id][0] != signature
            ):
                raise RuntimeError("call_id_conflict")
            batch[call.call_id] = (signature, call)
        # SDK-объекты сохраняются целиком, вместе с reasoning и phase.
        history.extend(response.output)
        if not calls:
            return {"text": complete_text(response.output),
                    "operation": app.recorded_result()}
        for call_id, (signature, call) in batch.items():
            if call_id not in completed_calls:
                completed_calls[call_id] = (signature, dispatch(call, app))
            history.append(completed_calls[call_id][1])
    raise RuntimeError("tool_round_limit")


def main():
    from openai import OpenAI
    parser = argparse.ArgumentParser()
    parser.add_argument("--model", required=True)
    parser.add_argument("--effort")
    parser.add_argument("--request-id", required=True)
    parser.add_argument("--db", default="visits.sqlite3")
    options = parser.parse_args()
    # Здесь демонстрационный серверный контекст, а не аргументы модели.
    db = sqlite3.connect(options.db)
    try:
        app = VisitApp(db, "demo-account", options.request_id, {"CASE-1842"})
        client = OpenAI(timeout=20.0, max_retries=0)
        result = run_loop(
            client.responses.create, options.model, app,
            "Запишите инженера по CASE-1842 на 12 октября 2026 года "
            "в 14:00 по Москве.", effort=options.effort,
        )
        print(json.dumps(result, ensure_ascii=False, indent=2))
    finally:
        db.close()


if __name__ == "__main__":
    main()

Пример запуска после настройки вашего подключения:

python responses_visit.py --model YOUR_RESPONSES_TOOL_MODEL \
  --request-id service-request-1842 --db visits.sqlite3

Перед повтором того же пользовательского действия сохраняйте request_id и путь к той же базе; для нового действия нужен новый номер. Это значение не должно генерироваться заново при каждом сетевом повторе. Даты и лимиты в примере выбраны для демонстрации, не задают условия OpenAI или вашей службы.

Результат программы содержит два поля: text — завершённый текст модели, operation — сохранённый результат приложения. Успешная демонстрационная запись выглядит так:

{
  "ok": true,
  "visit": {
    "id": "visit:service-request-1842",
    "case_id": "CASE-1842",
    "slot": "2026-10-12T14:00:00+03:00",
    "status": "scheduled"
  }
}

Этот JSON — ожидаемая форма локальной записи, не результат живого вызова модели. Если operation равен null, приложение не записало выезд: текст нельзя считать подтверждением. Если запись есть, а формулировка модели ей противоречит, показывайте пользователю проверенный результат приложения и регистрируйте расхождение; новая операция для «исправления текста» не нужна.

Почему здесь два разных идентификатора

call_id связывает предложение модели с function_call_output. Словарь completed_calls хранит результат в пределах одного запуска цикла. Повтор того же ID с теми же аргументами получает прежний результат; тот же ID с другими аргументами останавливает цикл до исполнения всего нового пакета. При перезапуске Python этот словарь исчезает.

request_id относится к пользовательскому действию. Пара (account_id, request_id) уникальна в базе, поэтому новый call_id не создаёт повторный выезд. Совпали заявка и время — возвращается сохранённый результат. Время изменилось под прежним номером запроса — возвращается request_id_conflict. Обновление записи и сохранение результата происходят в одной SQLite-транзакции.

Это защита конкретной локальной записи, а не универсальное «ровно один раз». Если обработчик отправляет запрос внешней службе, локальная транзакция не охватывает её действие. Понадобятся ключ идемпотентности принимающего сервиса либо журнал операций с устойчивыми состояниями, отправкой после фиксации намерения и сверкой неизвестных исходов. При таймауте внешняя запись могла уже появиться; слепой повтор не решает неопределённость.

Пакет вызовов не требует параллельного исполнения

parallel_tool_calls=True разрешает модели предложить несколько функций. Пример исполняет принятый пакет последовательно и возвращает результаты всех его уникальных call_id перед следующим запросом. Так SQLite-соединение не делится между потоками, а конфликтующие записи не гоняются наперегонки.

Если указать parallel_tool_calls=False, модель может вернуть ноль или один вызов; это не обещание обязательного вызова. tool_choice="auto" также допускает ответ без функции. Поэтому проверка operation нужна независимо от настройки параллельности. Правила выбора и параллельных вызовов.

Незавершённый ответ, отказ и ошибка функции — разные исходы

Проверяйте статус до запуска обработчиков. completed означает завершение генерации ответа, а не выполнение вашего бизнес-действия. При incomplete, failed, cancelled, queued или in_progress этот синхронный пример останавливается: незавершённый вывод не принимается за готовое поручение. Причину смотрите в incomplete_details или error; например, лимит max_output_tokens может закончиться до финального ответа. Этот лимит включает видимый текст и токены рассуждения. Параметры создания Response.

У примера нет автоматических сетевых повторов SDK; задан таймаут одного запроса. Шесть раундов и проверяемый между запросами срок ограничивают цикл приложения, но не гарантируют завершение любой внешней операции за 120 секунд. Для реального сервиса нужны собственные сроки и отмена обработчиков. max_tool_calls в API ограничивает встроенные инструменты OpenAI, не число ваших диспетчеризаций и не бюджет в деньгах. Справочник параметров Responses.

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

В примере история передаётся вручную: исходный пользовательский ввод, все полученные объекты response.output, затем результаты функций. Не пересобирайте её из output_text: это удобное свойство SDK для текста, а не полный контекст и не универсальное поле сырого REST-ответа. Сохраняйте элементы рассуждения и phase у сообщений. Текущая документация указывает, что зашифрованные элементы рассуждения возвращаются по умолчанию. Управление состоянием диалога.

previous_response_id работает и в HTTP, а не только в WebSocket. В одном запросе его нельзя сочетать с conversation. Ссылка на прошлый ответ уменьшает объём передаваемого JSON, но предыдущий контекст цепочки продолжает учитываться как входные токены. Состояние диалога, параметры запроса.

store=False отключает обычное хранение Response, но само по себе не даёт Zero Data Retention и не удаляет логи приложения или поставщика. Для стандартного хранимого Response документация описывает 30-дневное хранение состояния с оговорёнными исключениями; элементы Conversation не имеют такого 30-дневного срока. Политики мониторинга злоупотреблений и одобренные режимы MAM/ZDR рассматриваются отдельно. Выбирайте способ продолжения вместе с требованиями к данным, а не по одному флагу. Документация OpenAI о данных.

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

Сначала проверьте обычный цикл, затем замените обработчик Chat-потока на типизированные SSE-события Responses. Для вызова функции важны три разные готовности: завершились аргументы, завершился элемент вызова, завершился весь ответ. В этой статье предлагается консервативная политика: выпускать вызовы на исполнение только после response.completed и проверки его полного результата. Она увеличивает ожидание по сравнению с ранним запуском, зато незавершённый поток не успевает изменить данные.

Точные поля arguments.done приведены в схеме события OpenAI; полный элемент — в схеме output_item.done. Не подменяйте call_id значением item_id или сгенерированным номером. Если собственный буфер и итоговый Response расходятся по ID, имени или аргументам, остановите обработку как ошибку протокола.

При EOF до response.completed отбросьте неисполненный пакет. При разрыве после записи в приложении сохраните её результат и продолжайте формирование ответа с ним; не повторяйте запись ради потерянного текста. Для внешней операции с неизвестным исходом сначала выполните сверку. Правила SSE и терминальных событий задаёт руководство по потоковым ответам; решение, когда выпускать работу на исполнение, принадлежит приложению.

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

Официальный контракт OpenAI и совместимость стороннего сервиса проверяются отдельно. Например, русскоязычное руководство AGTCloud по переносу Chat → Responses описывает собственный маршрут, состояние и поток. Оно не подтверждает, что его модели, хранение или ограничения совпадают с OpenAI. Azure также использует свои пути, аутентификацию и имена развёртываний; требования одного поставщика нельзя переносить на другой только из-за похожего SDK.

Перед включением нового пути зафиксируйте реальный URL, тип ключа, идентификатор модели и поддерживаемые события. Если запрос проходит, но результаты функций, продолжение или ошибки имеют другую форму, нужна адаптация и отдельная приёмка этого маршрута. Русский язык интерфейса не устанавливает доступность сервиса в конкретной стране; условия официального подключения разобраны в руководстве по API-ключу.

Приёмка и обратимое включение трафика

Сначала сохраните очищенную трассу работающего Chat-сценария: исходный запрос, разрешённое действие, результат приложения и ответ пользователю. Не включайте в неё ключи, персональные данные и лишние аргументы. Сравнивайте новый путь по тому же бизнес-смыслу, а не по дословному тексту.

Для опубликованного Python-примера локально проверены разбор аргументов, запрет чужой заявки, результаты с исходными ID, конфликт ID до записи, повтор с новым call_id, конфликт времени под прежним request_id, статусы, отказ, пустой ответ, лимит раундов и сохранение всей истории. Использовались искусственные ответы, SQLite в памяти и временный файл базы для проверки перезапуска; это проверка логики приложения, не приёмка OpenAI-сервера. Для потоковой политики отдельно проверены синтетические последовательности завершения и обрыва; они не являются записью реального SSE-потока.

Перед рабочим включением нужен ваш стенд с реальным разрешённым подключением:

• Плоская схема принимается выбранным маршрутом и моделью.
• Каждый разрешённый вызов исполняется приложением и получает результат с тем же call_id.
• Пакет из нескольких вызовов полностью возвращён перед следующим раундом.
• Плохой JSON, неизвестное имя и отсутствие прав не меняют данные.
• Повтор того же бизнес-действия, в том числе после перезапуска, не создаёт вторую запись.
• Потеря ответа после записи восстанавливается из журнала; неизвестный исход внешней операции сначала сверяется.
• Применяется ровно один способ продолжения истории, а ошибки и отказы не принимаются за успех.
• Финальный ответ согласован с сохранённым результатом приложения.
• Переключение на прежний путь сохраняет модель, совместимые параметры, контекст и уже выполненные операции.

Возвращать любую новую модель обратно в Chat нельзя. По текущей таблице совместимости OpenAI, GPT-6 Astra и GPT-6.1 Sol требуют Responses для вызова инструментов; GPT-6 Sol и GPT-6 Luna поддерживают функции в Chat только при reasoning_effort: "none". В Responses используется reasoning.effort; Astra и GPT-6.1 Sol не поддерживают значение none. Сохраняйте прежний эффективный уровень, если новая комбинация его поддерживает, а изменение проверяйте отдельно. При ненулевом рассуждении также проверьте неподдерживаемые параметры семплирования. Рабочий откат — заранее проверенная комбинация маршрута, модели, настроек и восстановления операции, а не единственный переключатель URL.

Короткие ответы на частые вопросы

Обязательно ли уходить с Chat Completions?

Нет: OpenAI продолжает поддерживать Chat Completions. Но совместимость зависит от выбранной модели и функций; у некоторых актуальных моделей вызов инструментов доступен только через Responses. Миграция API, совместимость моделей.

Почему call_id совпал, а операция выполнилась дважды?

call_id связывает вызов и результат внутри протокола. Он не заменяет устойчивый ключ пользовательского действия, запись результата и атомарное изменение данных. Новый раунд или повторный запрос могут дать другой call_id; защищайте саму операцию в приложении.

Что означает retry timeout exceeded и можно ли повторить весь цикл?

Буквально — «исчерпан лимит времени повторных попыток». Точную причину по этой строке установить нельзя: значение зависит от SDK, шлюза или вашего кода. Найдите последний подтверждённый этап. Если функция уже изменила данные, восстановите её результат. Если исход неизвестен, сначала сверяйте запись с принимающей системой. Повторять весь цикл без этой проверки опасно для операций записи.

Можно ли считать arguments.done разрешением выполнить функцию?

Только если приложение сознательно выбрало ранний запуск и умеет обрабатывать последующий отказ или обрыв. В предложенной здесь политике нужно дождаться response.completed, проверить полный вызов, аргументы и права. arguments.done завершает строку аргументов, а не бизнес-действие. Точная схема события.