Гайд · TNWS AI

Как настроить Responses WebSocket transport в OpenAI Agents SDK

WindowsmacOSLinuxChatGPT#нейросети #API #OpenAI Agents SDK

12 сентября 2026 7 мин—

Подключаем responses_websocket_session, переиспользуем соединение, настраиваем ping_interval и ping_timeout, завершаем streaming до выхода из context.

Что решаем

Практическая задача этого руководства — переиспользовать одно WebSocket-соединение Responses API для нескольких последовательных запусков агента. Это не обзор возможностей: итог должен быть проверяемым в коде. Ниже используются только имена и поведение, подтверждённые официальной документацией на 12 сентября 2026 года. Цены и тарифы не указаны, поскольку они не нужны для описанной операции и могут зависеть от внешнего провайдера.

Работу удобно разделить на четыре уровня: подготовка входа, сам вызов, проверка контракта ответа и проверка бизнес-результата. Ошибка на любом уровне должна завершать сценарий понятным статусом. Наличие JSON, текста или HTTP 200 само по себе не означает, что задача выполнена.

Подтверждённый контракт

Официальный первоисточник подтверждает следующие единицы:

responses_websocket_session()
responses_websocket_options
ping_interval
ping_timeout
ws.run_streamed
stream_events()
previous_response_id=first.last_response_id
это не Realtime API

Названия скопированы из актуального контракта, а не восстановлены по памяти. Проверка выполнена 12 сентября 2026 года. После обновления SDK или Ollama повторно откройте источник и прогоните тесты: даже корректный сегодня пример не является бессрочной гарантией совместимости.

Когда применять

Используйте этот подход в сервисе, где отказ нельзя скрыть за общим сообщением «что-то пошло не так». Он особенно полезен для фоновых задач, CI, внутренних API и агентов с инструментами. Для одноразового эксперимента тоже сохраните минимальный лог: время, безопасный ID запуска, имя операции, длительность и итоговый код.

Не переносите в тест реальные персональные данные, API-ключи и производственные идентификаторы. Если нужен токен, читайте его из переменной окружения. Если операция меняет состояние, работайте с тестовым namespace или объектом и обязательно выполните независимую проверку после записи.

Пошаговая настройка

1. Зафиксируйте вход и критерий успеха

До запуска запишите ожидаемый результат в машинно-проверяемой форме: точное значение поля, тип исключения, максимальное число одновременно выполняемых задач, наличие модели в повторном GET или финальный status. Не используйте оценку «ответ выглядит нормально».

Вход для примера: Два последовательных запроса: первичный список рисков и уточнение второго пункта через previous_response_id.

Ожидаемый результат: Оба streamed result полностью потреблены внутри async context. Второй запуск продолжает цепочку первого, а приложение не закрывает context при незавершённом запросе.

2. Запустите минимальный рабочий код

import asyncio
+from agents import Agent, responses_websocket_session
+
+async def main():
+    agent = Agent(name="Assistant", instructions="Отвечай кратко.")
+    async with responses_websocket_session(
+        responses_websocket_options={"ping_interval": 20.0, "ping_timeout": 60.0}
+    ) as ws:
+        first = ws.run_streamed(agent, "Назови два риска миграции.")
+        async for _ in first.stream_events():
+            pass
+        second = ws.run_streamed(
+            agent, "Раскрой второй риск.",
+            previous_response_id=first.last_response_id,
+        )
+        async for event in second.stream_events():
+            print(event.type)
+
+asyncio.run(main())

Заменяйте только тестовые имена и адреса. Не удаляйте таймауты, raise_for_status, assert или явную проверку финального статуса: именно они отделяют работающую интеграцию от скрипта, который молча принимает неполный ответ.

3. Проверьте наблюдаемость

Для сетевого запроса логируйте HTTP-статус, длительность, имя endpoint и ключи ответа. Для потока храните последний полностью разобранный event и отдельно финальный status. Для агентного запуска считайте модельные turns, выданные tool calls и локально исполненные tools раздельно. Эти числа отвечают на разные вопросы.

Секретные заголовки, полный prompt, system instructions и содержимое документов в лог не помещайте. При ошибке достаточно безопасного идентификатора, класса исключения и короткого обезличенного фрагмента. Если ответ большой, логируйте количество элементов и набор обязательных ключей.

4. Выполните негативный тест

на тестовом стенде оборвите соединение между запросами. Для store=False или ZDR не рассчитывайте восстановить uncached previous_response_id: перестройте контекст локально или начните новую цепочку.

Негативный тест нужен не для галочки. Он доказывает, что автоматизация отличает ожидаемый отказ от успеха. Не используйте конструкцию except Exception: pass и не подменяйте ошибку пустым результатом. В CI процесс должен вернуть ненулевой код, если обязательное условие не выполнено.

Как проверить результат независимо

После изменяющего запроса сделайте отдельное чтение состояния другим endpoint или новым клиентом. После фильтра или конфигурации SDK исследуйте snapshot модельного вызова в детерминированном тесте. После потоковой операции требуйте документированный финальный event, а не просто закрытие соединения.

Для каждого шага сохраните таблицу из трёх колонок: наблюдаемый факт, ожидаемое значение, результат сравнения. Например: peak concurrency — 2 — pass; last status — success — pass; поле version — строка — pass. Такая таблица полезнее длинного лога и сразу показывает, где нарушился контракт.

Готовый шаблон для проверки реализации

Скопируйте этот запрос в ревью-чат вместе с кодом. Ответ модели используйте только как список гипотез, каждую затем подтвердите запуском или документацией.

Проверь интеграцию [НАЗВАНИЕ ОПЕРАЦИИ].
Официальный контракт: https://openai.github.io/openai-agents-python/running_agents/#responses-websocket-transport-optional-helper
Дата сверки: 12 сентября 2026 года.
Вход: [ТЕСТОВЫЕ ЗНАЧЕНИЯ].
Критерий успеха: [ТОЧНОЕ ПОЛЕ, СТАТУС ИЛИ ASSERT].

Найди:
1. неверные имена методов, параметров и полей;
2. отсутствие таймаута или проверки финального состояния;
3. утечку ключей, prompt или персональных данных;
4. ложный успех при пустом ответе или оборванном потоке;
5. автоматический повтор неидемпотентной операции;
6. расхождение между проверяемым результатом и заявленной задачей.

Для каждой проблемы укажи место, риск и минимальное исправление.
Не придумывай цены, лимиты, версии и поддержку функций.

Типичные ошибки

Первая ошибка — смешивать соседние уровни управления. Например, разрешение модели сформировать несколько tool calls не равно числу инструментов, реально исполняемых одновременно. Список установленных моделей не равен списку моделей в памяти. HTTP-успех промежуточного stream event не равен завершению всей операции.

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

Третья ошибка — повторять любой упавший запрос. Автоповтор допустим, когда ошибка временная, операция идемпотентна и вы контролируете число попыток. Для upload, create и tools с побочными эффектами сначала прочитайте состояние. Иначе один пользовательский запрос способен создать дубликат или повторить действие.

Чек-лист финальной приёмки

Ограничения

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

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

FAQ

Почему недостаточно HTTP 200?

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

Что делать при оборванном потоке?

Не записывать успех. Сохраните последний валидный event, проверьте фактическое состояние отдельным чтением и только после этого решайте, безопасен ли повтор. Финальный status нельзя домыслить.

Можно ли копировать пример в production?

Используйте его как минимальную основу. Добавьте контролируемые retries, метрики, структурированные логи, отмену, проверку прав и тесты на вашей версии. Секреты перенесите в защищённое хранилище.

Как часто сверять документацию?

Перед первым внедрением, после обновления зависимости и при любом изменении наблюдаемой формы ответа. Для этого материала контракт проверен 12 сентября 2026 года.

Официальный первоисточник

Документация операции

Текст и примеры подготовлены самостоятельно. Численные результаты не выдуманы: там, где значение зависит от среды, код читает его из фактического ответа или измеряет во время теста.