fix: Duplicate LLM calls for the Agents SDK

agentops/instrumentation/openai/instrumentor.py

@@ -24,13 +24,124 @@
 """
 
 from typing import List
-from opentelemetry.trace import get_tracer
+from opentelemetry.trace import get_tracer, SpanKind, Status, StatusCode
 from opentelemetry.instrumentation.openai.v1 import OpenAIV1Instrumentor as ThirdPartyOpenAIV1Instrumentor
 
 from agentops.logging import logger
 from agentops.instrumentation.common.wrappers import WrapConfig, wrap, unwrap
 from agentops.instrumentation.openai import LIBRARY_NAME, LIBRARY_VERSION
 from agentops.instrumentation.openai.attributes.common import get_response_attributes
+from opentelemetry import context as context_api
+
+
+def responses_wrapper(tracer, wrapped, instance, args, kwargs):
+    """Custom wrapper for OpenAI Responses API that checks for context from OpenAI Agents SDK"""
+    # Skip instrumentation if it's suppressed in the current context
+    if context_api.get_value("suppress_instrumentation"):
+        return wrapped(*args, **kwargs)
+
+    return_value = None
+
+    # Check if we have trace context from OpenAI Agents SDK
+    trace_id = context_api.get_value("openai_agents.trace_id", None)
+    parent_id = context_api.get_value("openai_agents.parent_id", None)
+    workflow_input = context_api.get_value("openai_agents.workflow_input", None)
+
+    if trace_id:
+        logger.debug(
+            f"[OpenAI Instrumentor] Found OpenAI Agents trace context: trace_id={trace_id}, parent_id={parent_id}"
+        )
+
+    with tracer.start_as_current_span(
+        "openai.responses.create",
+        kind=SpanKind.CLIENT,
+    ) as span:
+        try:
+            attributes = get_response_attributes(args=args, kwargs=kwargs)
+            for key, value in attributes.items():
+                span.set_attribute(key, value)
+
+            # If we have trace context from OpenAI Agents SDK, add it as attributes
+            if trace_id:
+                span.set_attribute("openai_agents.trace_id", trace_id)
+                if parent_id:
+                    span.set_attribute("openai_agents.parent_id", parent_id)
+                if workflow_input:
+                    span.set_attribute("workflow.input", workflow_input)
+
+            return_value = wrapped(*args, **kwargs)
+
+            attributes = get_response_attributes(return_value=return_value)
+            for key, value in attributes.items():
+                span.set_attribute(key, value)
+
+            span.set_status(Status(StatusCode.OK))
+        except Exception as e:
+            attributes = get_response_attributes(args=args, kwargs=kwargs, return_value=return_value)
+            for key, value in attributes.items():
+                span.set_attribute(key, value)
+
+            span.record_exception(e)
+            span.set_status(Status(StatusCode.ERROR, str(e)))
+            raise
+
+    return return_value
+
+
+async def async_responses_wrapper(tracer, wrapped, instance, args, kwargs):
+    """Custom async wrapper for OpenAI Responses API that checks for context from OpenAI Agents SDK"""
+    # Skip instrumentation if it's suppressed in the current context
+    if context_api.get_value("suppress_instrumentation"):
+        return await wrapped(*args, **kwargs)
+
+    return_value = None
+
+    # Check if we have trace context from OpenAI Agents SDK
+    trace_id = context_api.get_value("openai_agents.trace_id", None)
+    parent_id = context_api.get_value("openai_agents.parent_id", None)
+    workflow_input = context_api.get_value("openai_agents.workflow_input", None)
+
+    if trace_id:
+        logger.debug(
+            f"[OpenAI Instrumentor] Found OpenAI Agents trace context in async wrapper: trace_id={trace_id}, parent_id={parent_id}"
+        )
+
+    with tracer.start_as_current_span(
+        "openai.responses.create",
+        kind=SpanKind.CLIENT,
+    ) as span:
+        try:
+            # Add the input attributes to the span before execution
+            attributes = get_response_attributes(args=args, kwargs=kwargs)
+            for key, value in attributes.items():
+                span.set_attribute(key, value)
+
+            # If we have trace context from OpenAI Agents SDK, add it as attributes
+            if trace_id:
+                span.set_attribute("openai_agents.trace_id", trace_id)
+                if parent_id:
+                    span.set_attribute("openai_agents.parent_id", parent_id)
+                if workflow_input:
+                    span.set_attribute("workflow.input", workflow_input)
+
+            return_value = await wrapped(*args, **kwargs)
+
+            attributes = get_response_attributes(return_value=return_value)
+            for key, value in attributes.items():
+                span.set_attribute(key, value)
+
+            span.set_status(Status(StatusCode.OK))
+        except Exception as e:
+            # Add everything we have in the case of an error
+            attributes = get_response_attributes(args=args, kwargs=kwargs, return_value=return_value)
+            for key, value in attributes.items():
+                span.set_attribute(key, value)
+
+            span.record_exception(e)
+            span.set_status(Status(StatusCode.ERROR, str(e)))
+            raise
+
+    return return_value
 
 
 # Methods to wrap beyond what the third-party instrumentation handles
@@ -74,24 +185,60 @@
         tracer_provider = kwargs.get("tracer_provider")
         tracer = get_tracer(LIBRARY_NAME, LIBRARY_VERSION, tracer_provider)
 
-        for wrap_config in WRAPPED_METHODS:
-            try:
-                wrap(wrap_config, tracer)
-                logger.debug(f"Successfully wrapped {wrap_config}")
-            except (AttributeError, ModuleNotFoundError) as e:
-                logger.debug(f"Failed to wrap {wrap_config}: {e}")
+        # Only use custom wrappers to avoid duplicate spans
+        from wrapt import wrap_function_wrapper
+
+        try:
+            wrap_function_wrapper(
+                "openai.resources.responses",
+                "Responses.create",
+                lambda wrapped, instance, args, kwargs: responses_wrapper(tracer, wrapped, instance, args, kwargs),
+            )
+            logger.debug("Successfully wrapped Responses.create with custom wrapper")
+
+            wrap_function_wrapper(
+                "openai.resources.responses",
+                "AsyncResponses.create",
+                lambda wrapped, instance, args, kwargs: async_responses_wrapper(
+                    tracer, wrapped, instance, args, kwargs
+                ),
+            )
+            logger.debug("Successfully wrapped AsyncResponses.create with custom wrapper")
+        except (AttributeError, ModuleNotFoundError) as e:
+            logger.debug(f"Failed to wrap Responses API with custom wrapper: {e}")
+
+            # Fall back to standard wrappers only if custom wrappers fail
+            for wrap_config in WRAPPED_METHODS:
+                try:
+                    wrap(wrap_config, tracer)
+                    logger.debug(f"Falling back to standard wrapper for {wrap_config}")
+                except (AttributeError, ModuleNotFoundError) as e:
+                    logger.debug(f"Failed to wrap {wrap_config}: {e}")
 
         logger.debug("Successfully instrumented OpenAI API with Response extensions")
 
     def _uninstrument(self, **kwargs):
         """Remove instrumentation from OpenAI API."""
         super()._uninstrument(**kwargs)
 
-        for wrap_config in WRAPPED_METHODS:
-            try:
-                unwrap(wrap_config)
-                logger.debug(f"Successfully unwrapped {wrap_config}")
-            except Exception as e:
-                logger.debug(f"Failed to unwrap {wrap_config}: {e}")
+        # First try to unwrap custom wrappers
+        from opentelemetry.instrumentation.utils import unwrap as _unwrap
+
+        try:
+            _unwrap("openai.resources.responses.Responses", "create")
+            logger.debug("Successfully unwrapped Responses.create custom wrapper")
+
+            _unwrap("openai.resources.responses.AsyncResponses", "create")
+            logger.debug("Successfully unwrapped AsyncResponses.create custom wrapper")
+        except Exception as e:
+            logger.debug(f"Failed to unwrap Responses API custom wrapper: {e}")
+
+            # Fall back to standard unwrapping only if custom unwrapping fails
+            for wrap_config in WRAPPED_METHODS:
+                try:
+                    unwrap(wrap_config)
+                    logger.debug(f"Falling back to standard unwrapper for {wrap_config}")
+                except Exception as e:
+                    logger.debug(f"Failed to unwrap {wrap_config}: {e}")
 
         logger.debug("Successfully removed OpenAI API instrumentation with Response extensions")

agentops/instrumentation/openai_agents/attributes/common.py

@@ -19,6 +19,7 @@
     get_model_config_attributes,
 )
 from agentops.instrumentation.openai_agents.attributes.completion import get_generation_output_attributes
+from agentops.semconv import ToolAttributes
 
 
 # Attribute mapping for AgentSpanData
@@ -33,7 +34,7 @@
 
 # Attribute mapping for FunctionSpanData
 FUNCTION_SPAN_ATTRIBUTES: AttributeMap = {
-    AgentAttributes.AGENT_NAME: "name",
+    ToolAttributes.TOOL_NAME: "name",
     WorkflowAttributes.WORKFLOW_INPUT: "input",
     WorkflowAttributes.FINAL_OUTPUT: "output",
     AgentAttributes.FROM_AGENT: "from_agent",

agentops/instrumentation/openai_agents/exporter.py

@@ -87,10 +87,13 @@
     span_data = span.span_data
     span_type = span_data.__class__.__name__
 
+    span_name = ""
     if hasattr(span_data, "name") and span_data.name:
-        return span_data.name
+        span_name = span_data.name
     else:
-        return span_type.replace("SpanData", "").lower()  # fallback
+        span_name = span_type.replace("SpanData", "").lower()
+
+    return span_name
 
 
 def _get_span_lookup_key(trace_id: str, span_id: str) -> str:
@@ -303,6 +306,31 @@
         trace_id = getattr(span, "trace_id", "unknown")
         parent_id = getattr(span, "parent_id", None)
 
+        # Special handling for ResponseSpanData to avoid duplicate spans
+        # and ensure proper trace hierarchy
+        if span_type == "ResponseSpanData":
+            logger.debug(
+                "[agentops.instrumentation.openai_agents] Processing ResponseSpanData for trace context propagation"
+            )
+
+            # Store the trace context information in a global context that can be accessed
+            # by the OpenAI instrumentation when it creates spans
+            ctx = context_api.get_current()
+
+            # Store the OpenAI Agents trace context in the current context
+            ctx = context_api.set_value("openai_agents.trace_id", trace_id, ctx)
+            ctx = context_api.set_value("openai_agents.span_id", span_id, ctx)
+            ctx = context_api.set_value("openai_agents.parent_id", parent_id, ctx)
+
+            if hasattr(span_data, "input") and span_data.input:
+                ctx = context_api.set_value("openai_agents.workflow_input", str(span_data.input), ctx)
+            context_api.attach(ctx)
+
+            logger.debug(
+                f"[agentops.instrumentation.openai_agents] Propagated trace context: trace_id={trace_id}, parent_id={parent_id}"
+            )
+            return
+
         # Check if this is a span end event
         is_end_event = hasattr(span, "status") and span.status == StatusCode.OK.name
 

tests/unit/instrumentation/openai_agents/test_openai_agents_attributes.py

@@ -43,6 +43,7 @@
 )
 
 from agentops.semconv import (
+    ToolAttributes,
     SpanAttributes,
     MessageAttributes,
     AgentAttributes,
@@ -190,7 +191,7 @@ def test_function_span_attributes(self):
         attrs = get_function_span_attributes(mock_function_span)
 
         # Verify extracted attributes - note that complex objects should be serialized to strings
-        assert attrs[AgentAttributes.AGENT_NAME] == "test_function"
+        assert attrs[ToolAttributes.TOOL_NAME] == "test_function"
         assert attrs[WorkflowAttributes.WORKFLOW_INPUT] == '{"arg1": "value1"}'  # Serialized string
         assert attrs[WorkflowAttributes.FINAL_OUTPUT] == '{"result": "success"}'  # Serialized string
         assert attrs[AgentAttributes.FROM_AGENT] == "caller_agent"
@@ -456,7 +457,7 @@ def __init__(self):
             assert AgentAttributes.AGENT_NAME in agent_attrs
 
             function_attrs = get_span_attributes(function_span)
-            assert AgentAttributes.AGENT_NAME in function_attrs
+            assert ToolAttributes.TOOL_NAME in function_attrs
 
             # Unknown span type should return empty dict
             unknown_attrs = get_span_attributes(unknown_span)