Миграция с OpenAI Assistants API на Responses API: история, функции и откат

Чтобы перенести приложение с Assistants API на Responses API сейчас, восстановите инструкции и доступную историю из собственных сохранённых данных, выберите способ хранения нового диалога и замените обработку Run явным циклом вызовов функций. Затем отдельно проверьте File Search и переключите трафик на проверенную версию Responses.

Assistants API отключён 26 августа 2026 года. Получить старые сообщения через threads.messages.list уже нельзя: текущая инструкция OpenAI по миграции прямо предлагает использовать сообщения, сохранённые вашим приложением. Если остался только thread_id, это указатель на прежний объект, а не восстановленная история. Начать новый диалог можно; обещать пользователю продолжение утраченного — нельзя.

Вторая дата касается другого объекта: отключение reusable Prompt objects и /v1/prompts запланировано на 30 ноября 2026 года. Поэтому переносить Assistant в управляемый Prompt как постоянную конфигурацию уже неразумно. OpenAI рекомендует вынести содержимое Prompt в код приложения: инструкции, модель, объявления инструментов и формат ответа должны иметь собственную версию.

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

С чего начать, если старый API уже недоступен

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

Для каждого старого диалога заведите запись в своей БД: владелец, прежний thread_id, место хранения исходного архива, версия преобразования и состояние восстановления. Например: ready_text, read_only или missing. Новый conversation_id либо локальный ID истории добавляется после успешного создания нового состояния; подмена префикса thread_ на conv_ ничего не мигрирует.

Сначала определите ожидаемый результат задачи. Для счёта это идентификатор документа, разрешённый получатель и подтверждённый срок оплаты. Словесное совпадение старого и нового ответа не требуется. Требуется, чтобы новый ответ сохранил факты, права доступа и статус действия.

Как преобразовать сохранённые сообщения и не потерять вложения

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

Функция ниже принимает локальную запись вашего экспорта, а не делает запрос к Assistants. Её входной формат задан в коде: thread_id, messages_order: "asc" и полный список messages. Пагинированный экспорт надо собрать заранее, порядок проверить по сохранённым отметкам времени, а строки без содержимого не считать сообщениями. Исходный файл храните отдельно без изменений.

from copy import deepcopy


class ArchiveNeedsReview(ValueError):
    pass


def archive_to_input(archive):
    # Local export envelope; no request to the retired API.
    envelope_fields = {"thread_id", "messages_order", "messages",
                       "run_steps", "pending_actions"}
    if not isinstance(archive, dict) or set(archive) - envelope_fields:
        raise ArchiveNeedsReview("Unknown archive representation")
    if archive.get("messages_order") != "asc":
        raise ArchiveNeedsReview("Verify chronological order first")
    if not isinstance(archive.get("thread_id"), str):
        raise ArchiveNeedsReview("Missing source thread identifier")
    messages = archive.get("messages")
    if not isinstance(messages, list) or not messages:
        raise ArchiveNeedsReview("No recoverable message list")
    if archive.get("run_steps") or archive.get("pending_actions"):
        raise ArchiveNeedsReview("Execution history needs a separate adapter")

    items = []
    for message in messages:
        message_fields = {"id", "object", "created_at", "thread_id",
                          "assistant_id", "run_id", "metadata",
                          "role", "attachments", "content"}
        if not isinstance(message, dict) or set(message) - message_fields:
            raise ArchiveNeedsReview("Unknown message fields")
        if message.get("role") not in {"user", "assistant"}:
            raise ArchiveNeedsReview("Unsupported role")
        if message.get("attachments"):
            raise ArchiveNeedsReview("Attachments must remain in the archive")
        content = message.get("content")
        if not isinstance(content, list) or not content:
            raise ArchiveNeedsReview("Missing content blocks")
        parts = []
        for block in content:
            if not isinstance(block, dict) or set(block) - {"type", "text"}:
                raise ArchiveNeedsReview("Unknown content representation")
            if block.get("type") != "text":
                raise ArchiveNeedsReview("Non-text block needs a separate adapter")
            text = block.get("text")
            if not isinstance(text, dict):
                raise ArchiveNeedsReview("Unexpected text representation")
            if set(text) - {"value", "annotations"}:
                raise ArchiveNeedsReview("Unknown text fields")
            if text.get("annotations"):
                raise ArchiveNeedsReview("Preserve citations before conversion")
            if not isinstance(text.get("value"), str):
                raise ArchiveNeedsReview("Missing text value")
            parts.append(text["value"])
        # Easy-input messages for this deliberately narrow text-only route.
        items.append({"role": message["role"], "content": "\n\n".join(parts)})
    return deepcopy(items)


saved_archive = {
    "thread_id": "thread_archived_demo",
    "messages_order": "asc",
    "messages": [
        {"role": "user", "content": [
            {"type": "text", "text": {"value": "Меня интересует счёт С-42.",
                                      "annotations": []}}
        ]},
        {"role": "assistant", "content": [
            {"type": "text", "text": {"value": "Уточните, что нужно проверить.",
                                      "annotations": []}}
        ]},
    ],
}
history = archive_to_input(saved_archive)

Здесь отказ означает «не продолжать автоматически», а не «удалить сообщение». Перехватите ArchiveNeedsReview в обработчике приложения, покажите сохранённую переписку в режиме чтения и предложите начать новый диалог. Не подменяйте ошибку пустым history: тогда пользователь будет думать, что система помнит данные, которых у неё нет.

Этот преобразователь рассчитан только на указанное представление текстового экспорта. Неизвестные поля вашего формата сначала классифицируйте: технические ID и время остаются в сопроводительной записи, а сведения, влияющие на смысл или исполнение, требуют расширения преобразователя либо отказа. Для изображения нужен доступный и разрешённый источник; старый file_id сам по себе не подтверждает такую доступность. Сохранённый результат завершённой операции — исторический факт, а не новый вызов функции. Утраченное reasoning-состояние из старого текста восстанавливать нельзя. Основание для переноса из собственных сохранённых сообщений — раздел Migrating your integration.

Где хранить новый диалог: БД, цепочка Response или Conversation

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

conversation и previous_response_id нельзя указывать вместе. В цепочке previous_response_id прежние instructions не наследуются: отправляйте применимые инструкции заново. Состояние диалога не заменяет конфигурацию и права пользователя. Эти условия указаны в справочнике создания Response.

При ручном продолжении храните полный массив output, включая вызовы, reasoning Items и phase сообщений, если они присутствуют. output_text годится для отображения ответа, но не заменяет историю исполнения. Документация по состоянию диалога рекомендует сохранять все выходные Items; руководство по reasoning отдельно требует не нарушать контекст между запросом пользователя и результатами функций. Зашифрованное reasoning храните как непрозрачные данные; не пытайтесь восстановить его из текста.

Для длительной истории заранее определите предел контекста и какие завершённые участки можно сокращать. Резюме полезно, но не является точной копией исходника: права доступа, невыполненные обязательства и результаты действий надо сохранять отдельно. В цепочке через previous_response_id предыдущие входные токены всё равно оплачиваются; небольшой JSON запроса не означает бесплатный контекст. Это прямо оговорено в руководстве OpenAI.

store=False не означает отсутствие любых данных у провайдера

Response обычно хранится 30 дней; документация по данным для стандартного режима уточняет срок не менее 30 дней при обычном сохранении. У Conversation и его Items нет этого 30-дневного TTL: они хранятся до удаления и не относятся к ZDR-совместимым объектам. Поэтому store=False вместе с Conversation не даёт «диалог без хранения». Кроме того, обычные журналы контроля злоупотреблений могут содержать пользовательские данные и храниться до 30 дней с оговорёнными исключениями. Проверяйте отдельно историю приложения, Response, Conversation, файлы и настройки организации по правилам хранения OpenAI.

Перенос функций: рабочий цикл вместо requires_action

Assistant раньше объединял конфигурацию и исполнение Run. Теперь приложение отправляет конфигурацию с запросом, получает function_call, проверяет его и возвращает строковый function_call_output. Связь сохраняется через исходный call_id, а не через ID Item вида fc_.... После результата нужен следующий запрос модели: она может вернуть окончательный ответ или запросить ещё одну функцию. Так устроен официальный протокол function calling.

Для проверки возьмём вопрос: «Когда надо оплатить счёт С-42?» Разрешённая функция invoice_due читает запись только текущей организации. У пользователя нет возможности назначить tenant_id аргументом функции: его передаёт авторизованный backend.

Следующие функции предназначены для Python-backend с уже настроенным SDK-клиентом. Параметр client — объект вашего существующего подключения, model — доступная вам модель, а history — список из преобразователя выше или сохранённые Items текущего диалога. Демо-БД заменяется вашим запросом к хранилищу с тем же ограничением владельца. Пример не создаёт подключение, не устанавливает пакет и не обращается к старому API.

import json
from copy import deepcopy

INSTRUCTIONS = (
    "Отвечай по-русски. Для срока оплаты используй invoice_due. "
    "Не выдумывай реквизиты и даты. При ошибке доступа не раскрывай данные."
)
TOOLS = [{
    "type": "function",
    "name": "invoice_due",
    "description": "Read an invoice due date for the authenticated tenant.",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {"invoice_id": {"type": "string"}},
        "required": ["invoice_id"],
        "additionalProperties": False,
    },
}]
DEMO_INVOICES = {
    ("company_demo", "С-42"): {"invoice_id": "С-42", "due_date": "2026-10-20"}
}


def execute_read(call, tenant_id):
    if call.name != "invoice_due":
        raise ValueError("Unknown function")
    args = json.loads(call.arguments)
    if (not isinstance(args, dict) or set(args) != {"invoice_id"}
            or not isinstance(args["invoice_id"], str)
            or not args["invoice_id"].strip()):
        raise ValueError("Invalid invoice arguments")
    record = DEMO_INVOICES.get((tenant_id, args["invoice_id"]))
    # Same answer for absent and inaccessible records.
    return record if record is not None else {"error": "not_available"}


def invoice_turn(client, model, history, user_text, tenant_id, max_requests=5):
    if not tenant_id or not user_text.strip():
        raise ValueError("Authenticated tenant and user text are required")
    items = deepcopy(history)
    items.append({"role": "user", "content": user_text})
    seen = {}
    verified_records = []

    for _ in range(max_requests):
        response = client.responses.create(
            model=model,
            instructions=INSTRUCTIONS,
            tools=TOOLS,
            input=items,
            store=False,
            include=["reasoning.encrypted_content"],
        )
        if response.status != "completed":
            raise RuntimeError(f"Response not completed: {response.status}")
        # Preserve the entire typed output, not just displayed text.
        items.extend(response.output)
        calls = [x for x in response.output if x.type == "function_call"]

        if not calls:
            answer = response.output_text
            if not answer or not answer.strip():
                raise RuntimeError("No final text; inspect output/refusal")
            if not verified_records:
                raise RuntimeError("Invoice answer lacks a verified record")
            return {"text": answer, "items": items,
                    "records": verified_records, "response_id": response.id}

        for call in calls:
            if not isinstance(call.call_id, str) or not call.call_id:
                raise ValueError("Missing call_id")
            signature = (call.name, call.arguments)
            if call.call_id in seen:
                old_signature, encoded = seen[call.call_id]
                if old_signature != signature:
                    raise RuntimeError("Reused call_id with different arguments")
            else:
                result = execute_read(call, tenant_id)
                if "error" in result:
                    raise PermissionError("Invoice is not available to this tenant")
                verified_records.append(result)
                encoded = json.dumps(result, ensure_ascii=False)
                seen[call.call_id] = (signature, encoded)
            items.append({"type": "function_call_output",
                          "call_id": call.call_id, "output": encoded})
    raise RuntimeError("Request limit reached without a final answer")

Вызов обработчика из вашего backend выглядит так: invoice_turn(client, model, history, "Когда надо оплатить счёт С-42?", authenticated_tenant_id). Успешный результат содержит текст, полные Items для сохранения и подтверждённую запись. До выдачи пользователю сверяйте ключевую дату и номер счёта с records; в платёжном интерфейсе сами реквизиты лучше отображать непосредственно из записи БД. Наличие records исключает ответ без чтения, но само по себе не гарантирует точность свободного текста модели.

Этот цикл обрабатывает ноль, один и несколько вызовов, отказывает при неизвестной функции, неверных аргументах, недоступной записи, незавершённом Response и отсутствии финального текста. Ошибки передайте в контролируемый обработчик: без автоматического повторения операции и без показа пользователю чужих данных. Сохраняйте итоговые Items в своей БД атомарно, блокируйте одновременные ходы одного диалога и назначьте таймауты существующему SDK-клиенту. Лимит запросов не ограничивает продолжительность отдельного сетевого вызова.

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

Почему схему функции надо переносить явно

В Responses поля name, parameters и strict находятся непосредственно в объявлении функции. Не оставляйте старую вложенность function: {...}. При strict: true каждому объекту нужен additionalProperties: false, все свойства должны входить в required; необязательное значение описывается как допускающее null.

Если strict опущен, Responses пытается нормализовать схему. Когда она несовместима со строгим режимом, возможен переход к best-effort с strict: false; это не безусловная гарантия строгой проверки. parallel_tool_calls: false ограничивает ответ нулём или одним вызовом, но не заставляет функцию выполниться. Эти различия и исключения для отдельных моделей описаны в разделах strict mode и parallel function calling. Проверка схемы не заменяет проверку владельца счёта.

File Search: восстановите корпус и проверьте источники

Перенос списка функций не переносит поиск документов. file_search выполняет OpenAI по указанным vector_store_ids; для него не нужен ваш function_call_output. Он использует семантический поиск и поиск по ключевым словам. Файлы должны быть доступны, а их обработка — завершена со статусом completed. См. руководство File Search.

Если существующий vector store доступен вашему проекту, сначала сверяйте его файлы с перечнем нужных документов. Если доступ утрачен, новый индекс можно построить из собственных разрешённых оригиналов. Старые связи Assistant или Thread с файлами не служат настройкой нового запроса и не доказывают наличие документов.

Проверочный запрос в уже подключённом Python-клиенте выглядит так. authorized_store_id backend выбирает по организации пользователя; пример предполагает изолированный корпус одной организации.

response = client.responses.create(
    model=model,
    instructions="Ответь по документам и укажи источник срока оплаты.",
    input="Какой срок оплаты установлен для стандартного договора?",
    tools=[{"type": "file_search", "vector_store_ids": [authorized_store_id]}],
    include=["file_search_call.results"],
    store=False,
)
search_calls = [item for item in response.output if item.type == "file_search_call"]

Без include=["file_search_call.results"] результаты поиска по умолчанию не возвращаются, хотя текст может содержать аннотации с цитатами. Поэтому для диагностики нужны и найденные результаты, и ссылки на файлы в ответе. Проверяйте, что найденный пункт действительно задаёт срок, документ актуален и принадлежит нужному владельцу. Для общего корпуса используйте проверенные атрибуты файлов и metadata-фильтры, которые backend формирует сам; фильтр из пользовательского текста не является авторизацией. Параметр store=False не удаляет отдельно сохранённые файлы или vector store. Параметры результатов, цитат и фильтров приведены в документации поиска.

Контрольные вопросы должны различать старую и новую редакции договора, отсутствие ответа в корпусе и документ другой организации. Нормальный ответ на пустой или недоступный результат — отсутствие подтверждения, а не догадка. Успешный вызов поиска с неверной цитатой не проходит проверку переноса.

Streaming, JSON-ответы и долгие задачи

Потребитель событий Run надо заменить обработчиком типизированных событий Responses. Текстовые response.output_text.delta можно выводить постепенно. Части response.function_call_arguments.delta собирайте отдельно по Item, а исполнять функцию разрешайте только после получения полного вызова и проверки аргументов. response.completed относится к завершению Response; следующее действие приложения и окончание пользовательской задачи определяются вашим циклом. Ошибку или неполный ответ нельзя превращать в успешный финал. События и различия форматов описаны в руководстве перехода на Responses.

Если приложение ожидает JSON по схеме, перенесите определение из response_format в text.format. Например, туда входят type: "json_schema", name, schema и strict. Это отдельная настройка выходного ответа, а не JSON Schema аргументов функции. Точный формат показан в разделе Structured Outputs.

Для задачи, которая должна продолжиться после отключения браузера, используйте background: true, сохраняйте response.id и опрашивайте статус, пока он queued или in_progress. Выход из этих состояний означает завершение ожидания, не обязательно успех. Для остановки фоновой генерации используется responses.cancel; обычная синхронная генерация отменяется прекращением соединения. Возобновление фонового потока требует, чтобы исходный запрос имел stream: true: сохраняйте sequence_number и используйте starting_after согласно поддержке своего клиента. Все эти условия приведены в руководстве background mode.

Есть существенное условие хранения: текущая документация разрешает фоновый режим в ZDR-проектах с store=false, но данные временно сохраняются на диск примерно на 10 минут для исполнения и опроса. Это нельзя описывать как отсутствие хранения. Для MAM фоновые данные сохраняются после этого периода только при явно указанном store=true; при пропущенном параметре или store=false они удаляются примерно через 10 минут. Если ваша политика запрещает даже временное сохранение, такой маршрут ей не соответствует. Проверяйте применимые условия по background mode и правилам /v1/responses.

Отмена генерации не отменяет платёж или запись, которые ваш backend уже успел зафиксировать. Именно поэтому запись действия и его результат должны жить отдельно от статуса Response.

Как проверить перенос и подготовить настоящий откат

Вместо параллельного обращения к закрытому Assistants сравнивайте новую реализацию с сохранёнными результатами завершённых задач либо с уже работающей версией Responses. Эталон для нашего примера: организация company_demo, счёт С-42, дата из БД, чтение разрешено; для другого владельца — отсутствие доступа. Для поиска эталоном служит пункт конкретного документа, а не старый текст модели.

Локально для примеров выше проверены синтаксис Python и синтетические случаи: сохранение ролей и порядка, отказ на вложение и аннотацию, отсутствие доступной истории, несколько вызовов с исходными call_id, неверные аргументы, другой владелец, неполный ответ и исчерпание лимита. Это проверка логики преобразования и обработчика на имитации ответов. Приём запроса реальным API, совместимость установленного SDK, доступность модели и паритет production здесь не проверялись.

В вашем тестовом окружении следующий этап — настоящий запрос Responses с разрешёнными данными и действующей конфигурацией. Зафиксируйте версию модели, инструкций и схем; сохраните полный типизированный ответ, результаты чтения и окончательный результат задачи. Проверяйте следующие условия до расширения трафика:

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

Сначала направьте на новую версию внутренние сессии, затем небольшую контролируемую долю новых диалогов. Закрепляйте выбранную версию конфигурации и оркестратора за сессией. Старую историю подключайте после успешного преобразования, а неподдерживаемую показывайте из архива. Сравнение двух реализаций с реальными изменениями не должно исполнять действие дважды.

Откат после 26 августа — это переключение на проверенную версию Responses или подготовленный резервный процесс приложения. Возможны фиксированный ответ из БД, очередь на ручную обработку либо временная остановка функции. Нужны конкретный переключатель, сохранённая совместимая конфигурация, формат состояния и проверенный результат такого переключения. Если старая версия Responses не понимает новые Items, используйте резервный процесс; не отправляйте ей неизвестный формат. Закрытый Assistants не может быть резервным endpoint.

Короткие ответы на вопросы при миграции

Можно ли восстановить Thread, если известен только его ID?

Сам ID не содержит переписку. После отключения Assistants официальный путь чтения сообщений не работает; OpenAI предлагает брать историю из данных приложения. Ищите собственные экспорты, БД и резервные копии. Если содержимого нет, обозначьте отсутствие истории и начните новый диалог.

Нужно ли обязательно создавать Conversation?

Нет. Можно передавать историю вручную либо использовать previous_response_id. Conversation удобен для долговременного серверного диалога, но имеет отдельные правила хранения. В цепочке Response инструкции отправляются заново; параметры conversation и previous_response_id несовместимы. См. справочник Responses.

Почему функция выполнилась, а ответа пользователю нет?

Результат функции — ещё не окончательный ответ. Отправьте function_call_output с исходным call_id и продолжите цикл. Следующий Response может содержать новые вызовы, отказ или ошибку. Завершайте задачу после подходящего итогового ответа либо переводите её в явно обозначенный резервный процесс. Это следует из протокола function calling.

Можно ли оставить новый код на Prompt objects?

Для долгосрочного переноса вынесите содержимое в код: отдельное отключение Prompt objects запланировано на 30 ноября 2026 года. Дата не относится к самому Responses API. Проверяйте план отключения reusable prompts.

Достаточно ли заменить base_url, чтобы использовать всё это через посредника?

Нет. Совместимость формата запроса не подтверждает поддержку Conversations, File Search, фоновой генерации, событий и правил хранения. Настройка адреса SDK разобрана отдельно в руководстве по OpenAI base_url; возможности другого провайдера проверяются по его собственному контракту и реальному запросу. Перенос официального OpenAI API не служит проверкой стороннего маршрута.