rewrite openai agents instrumentation to util genai

Author: lmolkova

instrumentation/opentelemetry-instrumentation-genai-openai-agents/examples/manual/main.py

@@ -6,28 +6,62 @@
 
 from __future__ import annotations
 
-from agents import Agent, Runner, function_tool
+from agents import Agent, RunConfig, Runner, function_tool
 from dotenv import load_dotenv
 
 from opentelemetry import trace
+from opentelemetry._logs import set_logger_provider
+from opentelemetry.exporter.otlp.proto.grpc._log_exporter import (
+    OTLPLogExporter,
+)
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
+    OTLPMetricExporter,
+)
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
     OTLPSpanExporter,
 )
 from opentelemetry.instrumentation.genai.openai_agents import (
     OpenAIAgentsInstrumentor,
 )
+from opentelemetry.metrics import set_meter_provider
+from opentelemetry.sdk._logs import LoggerProvider
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
 
-def configure_otel() -> None:
-    """Configure the OpenTelemetry SDK for exporting spans."""
+def configure_otel() -> (
+    tuple[TracerProvider, MeterProvider, LoggerProvider]
+):
+    """Configure OpenTelemetry providers and install the instrumentor."""
+
+    tracer_provider = TracerProvider()
+    tracer_provider.add_span_processor(
+        BatchSpanProcessor(OTLPSpanExporter())
+    )
+    trace.set_tracer_provider(tracer_provider)
+
+    meter_provider = MeterProvider(
+        metric_readers=[
+            PeriodicExportingMetricReader(OTLPMetricExporter()),
+        ],
+    )
+    set_meter_provider(meter_provider)
 
-    provider = TracerProvider()
-    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
-    trace.set_tracer_provider(provider)
+    logger_provider = LoggerProvider()
+    logger_provider.add_log_record_processor(
+        BatchLogRecordProcessor(OTLPLogExporter())
+    )
+    set_logger_provider(logger_provider)
 
-    OpenAIAgentsInstrumentor().instrument(tracer_provider=provider)
+    OpenAIAgentsInstrumentor().instrument(
+        tracer_provider=tracer_provider,
+        meter_provider=meter_provider,
+        logger_provider=logger_provider,
+    )
+    return tracer_provider, meter_provider, logger_provider
 
 
 @function_tool
@@ -39,7 +73,7 @@ def get_weather(city: str) -> str:
 
 def main() -> None:
     load_dotenv()
-    configure_otel()
+    tracer_provider, meter_provider, logger_provider = configure_otel()
     weather_specialist = Agent(
         name="weather_specialist",
         instructions=(
@@ -60,13 +94,22 @@ def main() -> None:
         model="gpt-4o-mini",
     )
 
-    result = Runner.run_sync(
-        triage_agent,
-        "I'm visiting Barcelona this weekend. How should I pack?",
-    )
-
-    print("Agent response:")
-    print(result.final_output)
+    try:
+        # ``RunConfig.workflow_name`` is the agents library's own knob for
+        # naming the workflow. The instrumentation reads it and emits the
+        # value as the ``gen_ai.workflow.name`` attribute on the workflow
+        # span — without it, the default "Agent workflow" is used.
+        result = Runner.run_sync(
+            triage_agent,
+            "I'm visiting Barcelona this weekend. How should I pack?",
+            run_config=RunConfig(workflow_name="weather-triage"),
+        )
+        print("Agent response:")
+        print(result.final_output)
+    finally:
+        tracer_provider.shutdown()
+        meter_provider.shutdown()
+        logger_provider.shutdown()
 
 
 if __name__ == "__main__":

instrumentation/opentelemetry-instrumentation-genai-openai-agents/src/opentelemetry/instrumentation/genai/openai_agents/__init__.py

@@ -1,213 +1,118 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
-"""OpenAI Agents instrumentation for OpenTelemetry."""
+"""OpenAI Agents instrumentation for OpenTelemetry.
 
-# pylint: disable=too-many-locals
+Registers a :class:`GenAITracingProcessor` with the agents library's
+public ``add_trace_processor`` extension API. The processor reacts
+synchronously to the agents library's own ``Trace`` / ``AgentSpan`` /
+``FunctionSpan`` / ``HandoffSpan`` start/end callbacks and turns them
+into ``invoke_workflow`` / ``invoke_agent`` / ``execute_tool`` spans via
+``opentelemetry-util-genai`` (plus a raw ``handoff`` span until handoff
+semconv lands).
+
+LLM-level spans (``chat`` / ``responses`` / ``embeddings``) are produced
+by ``opentelemetry-instrumentation-genai-openai`` when both packages are
+installed; this instrumentation does not emit them.
+
+Usage
+-----
+
+.. code:: python
+
+    from opentelemetry.instrumentation.genai.openai_agents import (
+        OpenAIAgentsInstrumentor,
+    )
+
+    # Default: keep the OpenAI native trace exporter; add our OTel emission.
+    OpenAIAgentsInstrumentor().instrument()
+
+    # Replace the default exporter so traces are only sent via OTel.
+    OpenAIAgentsInstrumentor().instrument(disable_openai_trace_export=True)
+"""
 
 from __future__ import annotations
 
-import importlib
 import logging
-import os
 from typing import Any, Collection
 
+from agents.tracing import (
+    add_trace_processor,
+    get_trace_provider,
+    set_trace_processors,
+)
+
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAI,
 )
-from opentelemetry.semconv.schemas import Schemas
-from opentelemetry.trace import get_tracer
+from opentelemetry.util.genai.handler import (
+    TelemetryHandler,
+    get_telemetry_handler,
+)
 
 from .package import _instruments
-from .span_processor import (
-    ContentCaptureMode,
-    GenAIEvaluationAttributes,
-    GenAIOperationName,
-    GenAIOutputType,
-    GenAIProvider,
-    GenAISemanticProcessor,
-    GenAIToolType,
-)
+from .processor import GenAITracingProcessor
 
-__all__ = [
-    "OpenAIAgentsInstrumentor",
-    "GenAIProvider",
-    "GenAIOperationName",
-    "GenAIToolType",
-    "GenAIOutputType",
-    "GenAIEvaluationAttributes",
-]
+__all__ = ["OpenAIAgentsInstrumentor"]
 
 logger = logging.getLogger(__name__)
 
-_CONTENT_CAPTURE_ENV = "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"
-_SYSTEM_OVERRIDE_ENV = "OTEL_INSTRUMENTATION_OPENAI_AGENTS_SYSTEM"
-_CAPTURE_CONTENT_ENV = "OTEL_INSTRUMENTATION_OPENAI_AGENTS_CAPTURE_CONTENT"
-_CAPTURE_METRICS_ENV = "OTEL_INSTRUMENTATION_OPENAI_AGENTS_CAPTURE_METRICS"
-
-
-def _load_tracing_module():  # pragma: no cover - exercised via tests
-    return importlib.import_module("agents.tracing")
-
-
-def _get_registered_processors(provider) -> list:
-    multi = getattr(provider, "_multi_processor", None)
-    processors = getattr(multi, "_processors", ())
-    return list(processors)
-
-
-def _resolve_system(value: str | None) -> str:
-    if not value:
-        return GenAI.GenAiSystemValues.OPENAI.value
 
-    normalized = value.strip().lower()
-    for member in GenAI.GenAiSystemValues:
-        if normalized == member.value:
-            return member.value
-        if normalized == member.name.lower():
-            return member.value
-    return value
-
-
-def _resolve_content_mode(value: Any) -> ContentCaptureMode:
-    if isinstance(value, ContentCaptureMode):
-        return value
-    if isinstance(value, bool):
-        return (
-            ContentCaptureMode.SPAN_AND_EVENT
-            if value
-            else ContentCaptureMode.NO_CONTENT
-        )
+class OpenAIAgentsInstrumentor(BaseInstrumentor):
+    """Instrument the openai-agents library.
 
-    if value is None:
-        return ContentCaptureMode.SPAN_AND_EVENT
-
-    text = str(value).strip().lower()
-    if not text:
-        return ContentCaptureMode.SPAN_AND_EVENT
-
-    mapping = {
-        "span_only": ContentCaptureMode.SPAN_ONLY,
-        "span-only": ContentCaptureMode.SPAN_ONLY,
-        "span": ContentCaptureMode.SPAN_ONLY,
-        "event_only": ContentCaptureMode.EVENT_ONLY,
-        "event-only": ContentCaptureMode.EVENT_ONLY,
-        "event": ContentCaptureMode.EVENT_ONLY,
-        "span_and_event": ContentCaptureMode.SPAN_AND_EVENT,
-        "span-and-event": ContentCaptureMode.SPAN_AND_EVENT,
-        "span_and_events": ContentCaptureMode.SPAN_AND_EVENT,
-        "all": ContentCaptureMode.SPAN_AND_EVENT,
-        "true": ContentCaptureMode.SPAN_AND_EVENT,
-        "1": ContentCaptureMode.SPAN_AND_EVENT,
-        "yes": ContentCaptureMode.SPAN_AND_EVENT,
-        "no_content": ContentCaptureMode.NO_CONTENT,
-        "false": ContentCaptureMode.NO_CONTENT,
-        "0": ContentCaptureMode.NO_CONTENT,
-        "no": ContentCaptureMode.NO_CONTENT,
-        "none": ContentCaptureMode.NO_CONTENT,
-    }
-
-    return mapping.get(text, ContentCaptureMode.SPAN_AND_EVENT)
-
-
-def _resolve_bool(value: Any, default: bool) -> bool:
-    if value is None:
-        return default
-    if isinstance(value, bool):
-        return value
-    text = str(value).strip().lower()
-    if text in {"true", "1", "yes", "on"}:
-        return True
-    if text in {"false", "0", "no", "off"}:
-        return False
-    return default
+    Constructor takes no arguments. Configure behavior via ``instrument()``:
 
+    ``disable_openai_trace_export`` (default ``False``)
+        When ``False`` (default), the agents library's built-in trace
+        exporter (which sends traces to OpenAI's hosted tracing backend
+        when ``OPENAI_API_KEY`` is set) remains active alongside our OTel
+        emission.
 
-class OpenAIAgentsInstrumentor(BaseInstrumentor):
-    """Instrumentation that bridges OpenAI Agents tracing to OpenTelemetry."""
+        When ``True``, the default exporter is removed via
+        ``agents.tracing.set_trace_processors`` so traces flow only through
+        OpenTelemetry. Any other processors previously registered by the
+        user are also removed; if you have a custom processor list, manage
+        it yourself instead of using this flag.
+    """
 
     def __init__(self) -> None:
         super().__init__()
-        self._processor: GenAISemanticProcessor | None = None
+        self._processor: GenAITracingProcessor | None = None
+
+    def instrumentation_dependencies(self) -> Collection[str]:
+        return _instruments
 
-    def _instrument(self, **kwargs) -> None:
+    def _instrument(self, **kwargs: Any) -> None:
         if self._processor is not None:
             return
 
-        tracer_provider = kwargs.get("tracer_provider")
-        tracer = get_tracer(
-            __name__,
-            "",
-            tracer_provider,
-            schema_url=Schemas.V1_28_0.value,
+        handler: TelemetryHandler = get_telemetry_handler(
+            tracer_provider=kwargs.get("tracer_provider"),
+            meter_provider=kwargs.get("meter_provider"),
+            logger_provider=kwargs.get("logger_provider"),
+            completion_hook=kwargs.get("completion_hook"),
         )
+        provider = GenAI.GenAiProviderNameValues.OPENAI.value
+        self._processor = GenAITracingProcessor(handler, provider)
 
-        system_override = kwargs.get("system") or os.getenv(
-            _SYSTEM_OVERRIDE_ENV
-        )
-        system = _resolve_system(system_override)
-
-        content_override = kwargs.get("capture_message_content")
-        if content_override is None:
-            content_override = os.getenv(_CONTENT_CAPTURE_ENV) or os.getenv(
-                _CAPTURE_CONTENT_ENV
-            )
-        content_mode = _resolve_content_mode(content_override)
-
-        metrics_override = kwargs.get("capture_metrics")
-        if metrics_override is None:
-            metrics_override = os.getenv(_CAPTURE_METRICS_ENV)
-        metrics_enabled = _resolve_bool(metrics_override, default=True)
-
-        agent_name = kwargs.get("agent_name")
-        agent_id = kwargs.get("agent_id")
-        agent_description = kwargs.get("agent_description")
-        base_url = kwargs.get("base_url")
-        server_address = kwargs.get("server_address")
-        server_port = kwargs.get("server_port")
-
-        processor = GenAISemanticProcessor(
-            tracer=tracer,
-            system_name=system,
-            include_sensitive_data=content_mode
-            != ContentCaptureMode.NO_CONTENT,
-            content_mode=content_mode,
-            metrics_enabled=metrics_enabled,
-            agent_name=agent_name,
-            agent_id=agent_id,
-            agent_description=agent_description,
-            base_url=base_url,
-            server_address=server_address,
-            server_port=server_port,
-            agent_name_default="OpenAI Agent",
-            agent_id_default="agent",
-            agent_description_default="OpenAI Agents instrumentation",
-            base_url_default="https://api.openai.com",
-            server_address_default="api.openai.com",
-            server_port_default=443,
-        )
+        if kwargs.get("disable_openai_trace_export"):
+            set_trace_processors([self._processor])
+        else:
+            add_trace_processor(self._processor)
 
-        tracing = _load_tracing_module()
-        provider = tracing.get_trace_provider()
-        existing = _get_registered_processors(provider)
-        provider.set_processors([*existing, processor])
-        self._processor = processor
-
-    def _uninstrument(self, **kwargs) -> None:
+    def _uninstrument(self, **kwargs: Any) -> None:
         if self._processor is None:
             return
 
-        tracing = _load_tracing_module()
-        provider = tracing.get_trace_provider()
-        current = _get_registered_processors(provider)
-        filtered = [proc for proc in current if proc is not self._processor]
-        provider.set_processors(filtered)
-
+        provider = get_trace_provider()
+        current = getattr(
+            getattr(provider, "_multi_processor", None), "_processors", ()
+        )
+        filtered = [p for p in current if p is not self._processor]
+        set_trace_processors(filtered)
         try:
             self._processor.shutdown()
         finally:
             self._processor = None
-
-    def instrumentation_dependencies(self) -> Collection[str]:
-        return _instruments