Как написать потокобезопасный TracingProcessor для Agents SDK

Готовность определяется не отсутствием исключения. Нужен проверяемый признак: конкретное поле trace, событие processor, отсутствие секретного маркера или завершённый экспорт. Сначала запускайте пример на staging-проекте. В metadata не помещайте prompt, email, телефон, токены и ответы tools. Для корреляции используйте внутренний непрямой ID.

До изменения сохраните контрольную trace со стандартной конфигурацией и запишите её структуру: trace_id, имена spans, parent_id и число событий. Это baseline. После изменения сравнивайте именно ожидаемую разницу. Если одновременно обновить SDK, exporter и конфигурацию, причину изменения определить будет трудно.

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

1. Создайте виртуальное окружение, установите openai-agents и зафиксируйте разрешённую версию в lock-файле.
2. Настройте ключи через переменные окружения или secret manager. Не вставляйте рабочий ключ в код, trace metadata и тестовый отчёт.
3. Скопируйте пример ниже в отдельный интеграционный тест. Имена workflow и атрибуты замените на значения вашей предметной области.
4. Выполните один контрольный запуск и сохраните только технические идентификаторы. Текст сообщения нужен лишь локальному assert и не должен попадать в общий лог.
5. Проверьте позитивный критерий из раздела примера. Для processor считайте callbacks; для редактирования сериализуйте событие в тестовый sink; для flush проверяйте порядок закрытия контекста.
6. Выполните негативный сценарий. Он должен отличать ошибочную конфигурацию от временной задержки backend.
7. Запустите два параллельных trace. Их trace_id должны различаться, parent_id не должны пересекать workflows, а общий group_id допускается только если это один диалог.
8. Добавьте cleanup: flush для критичного короткого job, shutdown для собственного processor и очистку тестовых данных по политике проекта.

Готовый пример

from queue import Queue, Empty
from agents import add_trace_processor, flush_traces, trace
from agents.tracing import TracingProcessor

class QueueProcessor(TracingProcessor):
    def __init__(self): self.q=Queue(); self.exported=[]
    def on_trace_start(self,t): self.q.put(("trace_start",t.trace_id))
    def on_trace_end(self,t): self.q.put(("trace_end",t.trace_id))
    def on_span_start(self,s): self.q.put(("span_start",s.span_id))
    def on_span_end(self,s): self.q.put(("span_end",s.span_id))
    def force_flush(self):
        while True:
            try: self.exported.append(self.q.get_nowait())
            except Empty: break
    def shutdown(self): self.force_flush()

p=QueueProcessor(); add_trace_processor(p)
with trace("queue_processor"): pass
flush_traces()
assert any(x[0]=="trace_end" for x in p.exported)
print(len(p.exported))

В примере нет универсального «попробуйте посмотреть dashboard»: состояние проверяется программно. Для реального exporter дополните тест контролируемым sink, потому что появление записи в веб-интерфейсе может быть отложено фоновой пакетной отправкой. Не делайте сетевой запрос внутри синхронного callback processor — складывайте событие в очередь.

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

Вход: Одна trace queue_processor; callbacks складывают события в потокобезопасную Queue.

Ожидаемый результат: После flush очередь пуста, exported содержит trace_start и trace_end, а callback не выполнял сетевой ввод-вывод.

Зафиксируйте expected как автоматический критерий. Если сравнивается trace_id, проверяйте формат и уникальность, но не жёстко заданное случайное значение. Если сравнивается набор spans, используйте тип и имя, а не время выполнения: длительность зависит от среды и сети. Если проверяется отсутствие данных, ищите уникальные маркеры во всём сериализованном событии, включая error и metadata.

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

Замените Queue обычным разделяемым списком и запустите параллельные traces как нагрузочный тест. Такое изменение не должно приниматься без отдельной синхронизации.

Негативный тест важен, потому что tracing не должен ломать основной ответ агента. Сбой дополнительного exporter следует журналировать технически и обрабатывать внутри processor. При этом тест конфигурации обязан падать, если ожидаемая защита данных отключена или события отправляются не в тот проект. Разделяйте доступность observability и корректность бизнес-операции.

Производственная интеграция

Инициализируйте глобальные processors один раз при старте процесса. Повторная регистрация после каждого HTTP-запроса создаёт дубликаты. Конфигурацию отдельного run передавайте через RunConfig, когда поведение действительно относится только к нему. Глобальные set-функции меняют процесс целиком и требуют явного владельца в bootstrap-коде.