Гайд · TNWS AI

Как отладить tracing через ConsoleSpanExporter Agents SDK

6 мин

Подключаем ConsoleSpanExporter к BatchTraceProcessor, создаём тестовую trace и проверяем локальный вывод без backend.

Практическая задача и применимость

Этот гайд решает конкретную задачу: увидеть сформированные trace и span локально до подключения удалённого ingest. Он предназначен для разработчиков Python-сервисов, которые уже используют OpenAI Agents SDK и хотят проверять наблюдаемость кодом, а не только визуально в панели. Основной интерфейс материала — ConsoleSpanExporter.export(items).

Функция нужна на границе приложения и tracing-инфраструктуры: в bootstrap-модуле, worker, собственном runner-адаптере или интеграционном тесте. Не добавляйте такую настройку в каждый обработчик HTTP-запроса. Глобальные provider и processors должны иметь одного владельца жизненного цикла.

Что подтверждено официальной документацией

Сигнатуры и поведение проверены 13 сентября 2026 года по руководству OpenAI Agents SDK, общему API reference tracing и профильному reference. ConsoleSpanExporter реализует TracingExporter и печатает trace_id и имя trace либо экспортированный словарь span. Цены, тарифы и неподтверждённые лимиты в материале не используются. Версию зависимости фиксируйте в lock-файле и повторяйте контрактные тесты после обновления.

Важно разделять бизнес-успех и успех telemetry. Ошибка exporter не должна превращать уже выполненную бизнес-операцию в повторную, иначе возможен дубль платежа или задания. Одновременно нельзя молча объявлять tracing надёжным: измеряйте созданные, завершённые, отправленные и отброшенные события отдельно.

Архитектура решения

В примере есть четыре слоя. Первый создаёт или получает provider. Второй формирует trace/span data только из разрешённых технических полей. Третий управляет start, finish, flush и shutdown. Четвёртый проверяет результат через свойства объекта или export. Такая схема отделяет конфигурацию от предметного кода и делает ошибки воспроизводимыми.

Не записывайте в name динамические идентификаторы пользователей: используйте стабильное имя операции, а корреляцию храните в безопасном group_id либо ограниченной metadata. Input и output могут содержать чувствительные данные, поэтому примеры используют исключительно искусственные значения. Для production создайте allowlist полей и тест с секретным маркером.

Пошаговая инструкция

  1. Создайте отдельное виртуальное окружение и установите пакет openai-agents; точную версию зафиксируйте тем же инструментом зависимостей, который использует проект.
  2. Вынесите tracing bootstrap в один модуль. Зафиксируйте, кто вызывает shutdown при завершении процесса.
  3. Импортируйте только официальные классы и helpers, указанные в reference. Не обращайтесь к приватным полям с подчёркиванием в production-коде.
  4. Подготовьте искусственный вход без токенов, email, текста реального клиента и содержимого внутренних документов.
  5. Вызовите ConsoleSpanExporter.export(items) и сохраните возвращённый объект, если его свойства участвуют в контрактной проверке.
  6. Для ручного lifecycle используйте try/finally. Если объект становится текущим, при завершении обязательно восстановите contextvar.
  7. Проверьте конкретные поля: формат ID, parent_id, span_data, started_at/ended_at, export либо identity provider.
  8. Запустите негативный сценарий. Успех без такого теста не доказывает, что конфигурация ловит неправильную связь или утечку контекста.
  9. В отдельном concurrency-тесте запустите две asyncio-задачи и убедитесь, что их current trace/span не смешиваются.
  10. Перед выпуском вызовите flush там, где процесс короткоживущий, затем shutdown владельца processor/provider.

Готовый код

from contextlib import redirect_stdout
from io import StringIO
from agents import custom_span, trace
from agents import set_trace_processors
from agents.tracing.processors import BatchTraceProcessor, ConsoleSpanExporter

buffer = StringIO()
processor = BatchTraceProcessor(ConsoleSpanExporter(), schedule_delay=0.1)
set_trace_processors([processor])
try:
    with redirect_stdout(buffer):
        with trace("console_debug"):
            with custom_span("validate_payload", {"kind":"demo"}):
                pass
        processor.force_flush()
finally:
    processor.shutdown()
text = buffer.getvalue()
assert "console_debug" in text
assert "validate_payload" in text
print("console_export_verified")

Код можно скопировать в отдельный тестовый файл. Assertions являются частью решения: они превращают предположение о tracing в проверяемый контракт. Если пример использует тестовый ключ или домен example.test, не заменяйте их реальными значениями в репозитории. Секрет передавайте через менеджер секретов или переменную окружения только во время выполнения.

Реалистичный пример входа и результата

Вход: Trace console_debug содержит custom span validate_payload и только искусственное поле kind=demo.

Ожидаемый результат: Перехваченный stdout содержит имя workflow и имя span; после force_flush processor корректно завершается.

Проверяйте структуру, а не случайное значение сгенерированного ID или микросекунды времени. Для идентификатора важны префикс, формат, уникальность и сохранение связи. Для lifecycle важны наличие отметок и очистка contextvar. Для exporter важны полученная пачка, отсутствие секретов и корректное закрытие клиента.

Негативный тест

Добавьте маркер SECRET_TEST в custom data. Отдельный security-тест должен найти его в stdout и доказать, почему exporter нельзя оставлять в production.

Негативный сценарий должен быть автоматическим. Если он лишь описан в документации команды, регрессия останется незаметной. Поместите его рядом с позитивным тестом и запускайте в CI без реального сетевого экспорта. Для сетевой части используйте управляемый mock endpoint, который записывает количество запросов, заголовки без вывода секрета и последовательность кодов ответа.

Диагностика типичных ошибок

Если parent_id пуст, сначала проверьте, был ли родитель текущим в момент создания span или был ли он передан явно. Если current object остаётся после исключения, ищите ручной start без finish(reset_current=True). Если export равен None, проверьте disabled/no-op режим до анализа processor. Если события дублируются, убедитесь, что processor не зарегистрирован и глобально, и повторно внутри запроса.

Если процесс завершается раньше фонового экспорта, добавьте force_flush перед shutdown, но не вызывайте его после каждого запроса. Если тест зависит от порядка глобальных provider, восстанавливайте прежнее значение в finally. Тесты, которые меняют глобальную конфигурацию, нельзя бездумно запускать параллельно в одном процессе.

Копируемый шаблон внедрения

Операция: ConsoleSpanExporter.export(items)
Владелец lifecycle: <bootstrap/worker/test fixture>
Причина настройки: <одна конкретная задача>
Стабильное имя workflow/span: <значение>
Разрешённые metadata: <список>
Запрещённые данные: токены, PII, полный пользовательский ввод
Проверяемые свойства: <точные поля>
Ожидаемый parent: <trace/span/None>
Позитивный assert: <условие>
Негативный сценарий: <условие>
Поведение при исключении: finish/reset/restore
Поведение при shutdown: flush затем закрытие
Источник: https://openai.github.io/openai-agents-python/ref/tracing/processors/
Дата проверки: 13 сентября 2026 года

Контроль качества перед production

Сначала выполните unit-тест без сети. Затем интеграционный тест с тестовым processor или mock exporter. После этого создайте одну staging trace и сопоставьте её trace_id с техническим журналом. Проверьте, что ни сериализованный payload, ни stdout, ни сообщение об ошибке не содержат тестовый секретный маркер.

Отдельно смоделируйте исключение между start и finish, повторную инициализацию bootstrap и завершение короткой job. После каждого сценария current trace/span должен соответствовать ожидаемому объекту либо None. Зафиксируйте эти проверки в CI; ручной просмотр панели оставьте дополнительной, а не единственной проверкой.

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

  • Сигнатура сверена с официальным reference 13 сентября 2026 года.
  • Title операции и имена spans имеют низкую кардинальность.
  • Пример использует искусственные данные.
  • В metadata и payload нет токенов и PII.
  • Глобальная конфигурация создаётся один раз.
  • Ручной start всегда парен с finish в finally.
  • Contextvar восстанавливается после выхода и исключения.
  • Проверяются конкретные свойства или export.
  • Негативный тест действительно падает при неверной настройке.
  • Параллельные контексты изолированы.
  • Короткоживущий процесс выполняет flush.
  • Владелец ресурсов вызывает shutdown/close.

FAQ

Нужно ли тестировать tracing, если бизнес-функция работает?

Да. Бизнес-тест не обнаружит неверный parent_id, утечку contextvar, дубли processor или потерю буфера при завершении процесса. Проверяйте эти свойства отдельно.

Можно ли использовать ConsoleSpanExporter в production?

Для постоянной production-настройки это рискованно: stdout может получить содержимое span. Он удобен локально с искусственными данными и явной проверкой редактирования.

Почему нельзя регистрировать processor на каждый запрос?

Глобальный provider отправляет события всем зарегистрированным processors. Повторная регистрация создаёт дубли, расходует ресурсы и усложняет корректный shutdown.

Что делать, если tracing отключён?

Код должен корректно работать с no-op объектами и export=None. Не считайте отсутствие исключения доказательством того, что событие было отправлено.

Когда нужен force_flush?

Перед завершением короткоживущего worker/job или в контролируемом тесте немедленной доставки. В обычном серверном запросе полагайтесь на фоновый processor и общий lifecycle приложения.

Читайте также

Комментарии

Пока тихо. Скажите первое слово