Гайд · TNWS AI
Как обрабатывать streaming-события в OpenAI Agents SDK
Практический гайд по Runner.run_streamed и stream_events: raw_response_event, run_item_stream_event, agent_updated_stream_event, завершение, отмена и сбор финального ответа.
Задача и применимость
Задача: показывать ответ по мере генерации и одновременно корректно обрабатывать смену агента, tool items, ошибки и завершение. Это самостоятельный production-ориентированный разбор: ниже есть точные интерфейсы, копируемый каркас, реалистичный пример, постусловия и негативные тесты. Сведения сверены с официальной документацией 12 сентября 2026 года. Неподтверждённые цены, лимиты и региональная доступность не используются.
Подтверждённые единицы API
Ключевые интерфейсы: Runner.run_streamed,stream_events(),RawResponsesStreamEvent,RunItemStreamEvent,AgentUpdatedStreamEvent. Их назначение и форма проверены по официальной документации. Название модели вынесено в переменную окружения: доступные модели и провайдеры меняются, поэтому старое имя из примера нельзя считать вечным.
Важно разделять четыре результата: транспортный ответ, соответствие схеме, смысловую корректность и разрешение на действие. Успешный HTTP не подтверждает три последних пункта. Score модели не заменяет проверку на вашей размеченной выборке, а schema-valid аргумент не доказывает право пользователя на ресурс.
Что подготовить
Создайте отдельное Python-окружение, установите официальный SDK и зафиксируйте версию в lock-файле. Секрет положите в переменную окружения; не вставляйте его в браузерный JavaScript, мобильное приложение, notebook или git. Подготовьте три fixture: обычный вход, пограничный и заведомо неподходящий. Для каждого запишите обязательные поля и допустимый отказ.
Если обрабатываются документы, изображения или пользовательские идентификаторы, заранее определите срок хранения и состав логов. В журнал достаточно писать operation, model, duration, status, размеры и внутренний request ID. Полный контент и токены доступа туда не помещают.
Пошаговая настройка
- Получите RunResultStreaming
- итерируйте async for event in result.stream_events()
- текст добавляйте только из документированного delta
- отдельно обрабатывайте run item и agent update
- после конца читайте final_output
- при отключении клиента отмените run и закройте поток.
После каждого шага оставляйте проверяемый артефакт: запись теста, список полей, сохранённый безопасный sample или screenshot графа. Это делает инструкцию воспроизводимой после обновления модели.
Копируемый шаблон
import os
# Установите официальный SDK, указанный в документации темы
MODEL = os.environ.get("AI_MODEL")
assert MODEL, "Задайте AI_MODEL из актуального списка вашего аккаунта"
TASK = "показывать ответ по мере генерации и одновременно корректно обрабатывать смену агента, tool items, ошибки и завершение"
CONTROL_INPUT = "Агент сначала вызывает get_order, затем пишет статус по частям"
EXPECTED = "UI показывает текст без JSON tool-вызова, журнал получает tool item, final_output доступен только после завершения потока."
# Вставьте вызов документированного интерфейса: Runner.run_streamed,stream_events(),RawResponsesStreamEvent,RunItemStreamEvent,AgentUpdatedStreamEvent
# Секрет берите только из окружения.
print({"task": TASK, "input": CONTROL_INPUT, "expected": EXPECTED})
Этот каркас задаёт контракт теста. Конкретный вызов собирайте по официальному примеру страницы, указанной выше, и не заменяйте документированные имена универсальными обёртками до первого успешного contract test.
Реалистичный пример
Вход: Агент сначала вызывает get_order, затем пишет статус по частям.
Ожидаемый результат: UI показывает текст без JSON tool-вызова, журнал получает tool item, final_output доступен только после завершения потока..
Сохраните пример как regression fixture. Текстовый ответ не сравнивайте целиком: проверяйте обязательные сущности, типы, диапазоны, имена агента, координаты, размеры или разрешённые действия — в зависимости от задачи. Любое нарушение инварианта считается ошибкой, даже если ответ выглядит убедительно.
Проверка результата по уровням
Транспорт
Проверьте HTTP-статус, таймаут и request ID. 401/403 и ошибки параметров не повторяйте автоматически. Для временных 429/5xx используйте ограниченный exponential backoff с jitter. Если операция имеет побочный эффект, повтор разрешён только с идемпотентным ключом и проверкой фактического состояния.
Контракт
Валидируйте обязательные поля и типы. Нельзя превращать отсутствующее поле в пустую строку и считать запрос успешным. При обновлении SDK прогоните сохранённый mock с поломанным типом: тест обязан упасть до бизнес-логики.
Смысл
Проверяйте числа, даты, отрицания, подписи, диапазоны и соответствие исходнику. Для классификации тестируйте неизвестный класс; для извлечения — точный срез; для графа — allowlist рёбер; для streaming — порядок событий и факт завершения.
Безопасность
Модель не авторизует пользователя. Проверка владельца заказа, разрешения на файл и допустимости операции выполняется серверным кодом после разбора аргументов и до побочного эффекта. Данные локального context не становятся видимыми модели автоматически, но могут утечь, если tool вернёт их в тексте.
Негативные тесты
- Уберите секрет: процесс должен остановиться до передачи пользовательских данных.
- Укажите недоступную модель: получите явную ошибку, а не пустой успех.
- Удалите обязательное поле из mock-ответа: контрактная проверка должна сработать.
- Передайте чужой идентификатор или неподходящий файл: действие блокируется.
- Имитируйте таймаут после отправки: повтор не создаёт второй эффект.
- Проверьте пустой и предельно длинный вход; обрезание не должно быть тихим.
Практическая приёмка
Для контрольной выборки заведите таблицу: fixture ID, версия модели, версия SDK, входной hash, ожидаемые инварианты, фактические значения, verdict и reviewer. Не выдумывайте универсальный порог качества: получите baseline на собственных примерах, отдельно для нормальных, граничных и отрицательных случаев.
Перед расширением трафика измеряйте долю ошибок транспорта, контрактных ошибок, пустых результатов, ручных отклонений и повторов. Резкое изменение после смены модели — сигнал остановить rollout. Производственный runbook должен описывать откат, отзыв ключа и восстановление очереди.
Чек-лист финальной проверки
- Интерфейсы сверены с официальной документацией 12 сентября 2026 года.
- Title, description и slug прошли ограничения CMS.
- Ключ хранится вне кода и логов.
- Модель доступна в рабочем аккаунте и подходит языку/домену.
- Контрольный пример проходит содержательные постусловия.
- Негативный пример не превращается в ложный успех.
- Числа, позиции, координаты или связи проверены кодом.
- Повторы ограничены и безопасны.
- Версии SDK и модели зафиксированы в журнале теста.
- Есть ручной маршрут для неуверенных результатов.
Ограничения
Результат зависит от модели, её обучающих классов, языка и текущего провайдера. Демонстрационный запрос не доказывает качество на всем потоке. Схема ответа не гарантирует истинность. Для финансовых, медицинских, юридических и иных существенных решений нужен отдельный доменный контроль.
Не полагайтесь на названия вроде “base”, “large” или высокий score без model card и тестовой выборки. Не публикуйте метрики, которых сами не измеряли. При недоступности функции лучше вернуть понятный отказ, чем незаметно переключиться на модель с другим контрактом.
FAQ
Что считать успешной проверкой?
Для задачи «показывать ответ по мере генерации и одновременно корректно обрабатывать смену агента, tool items, ошибки и завершение» HTTP 200 недостаточно: должны совпасть документированные поля, контрольные значения и негативный сценарий.
Можно ли использовать пример в production?
Только после добавления авторизации, таймаутов, ограниченных повторов, безопасного хранения ключей и регрессионных тестов на данных вашего домена.
Почему нет цены и фиксированного лимита?
Эти параметры меняются и зависят от аккаунта, модели и провайдера. Их нужно проверять в официальном кабинете в день запуска.
Что делать при изменении ответа SDK?
Зафиксировать сырой ответ без секретов, версию зависимости и сверить типы с официальной документацией. Пустой fallback не должен маскировать несовместимость.
Официальный первоисточник
- https://openai.github.io/openai-agents-python/streaming/ — проверено 12 сентября 2026 года.
Частые вопросы
Что считать успешной проверкой?
Для задачи «показывать ответ по мере генерации и одновременно корректно обрабатывать смену агента, tool items, ошибки и завершение» HTTP 200 недостаточно: должны совпасть документированные поля, контрольные значения и негативный сценарий.
Можно ли использовать пример в production?
Только после добавления авторизации, таймаутов, ограниченных повторов, безопасного хранения ключей и регрессионных тестов на данных вашего домена.
Почему нет цены и фиксированного лимита?
Эти параметры меняются и зависят от аккаунта, модели и провайдера. Их нужно проверять в официальном кабинете в день запуска.
Что делать при изменении ответа SDK?
Зафиксировать сырой ответ без секретов, версию зависимости и сверить типы с официальной документацией. Пустой fallback не должен маскировать несовместимость.
Читайте также
Как передавать Local Context в OpenAI Agents SDK без утечки данных
Практический гайд по RunContextWrapper: dataclass-контекст, context в Runner.run, доступ из tools, разделение локальных данных и LLM-контекста, проверка авторизации.
Как разбирать RunResult в OpenAI Agents SDK и продолжать диалог
Гайд по RunResult и RunResultStreaming: final_output, new_items, last_agent, raw_responses, to_input_list и проверка результата после handoff или инструмента.
Как добавить Cohere Rerank после BM25 или векторного поиска
Практическое руководство по запросу «как использовать Cohere Rerank»: реальные параметры Cohere API, рабочий код, ограничения и измеримая проверка результата.
Комментарии
Пока тихо. Скажите первое слово