fix(tracing): don't export provider gen_ai spans nested in a Langfuse generation

langfuse/_client/span_processor.py

@@ -46,6 +46,7 @@
 from langfuse._client.span_filter import (
     is_app_root_eligible,
     is_default_export_span,
+    is_genai_span,
     is_langfuse_span,
 )
 from langfuse._client.utils import span_formatter
@@ -159,10 +160,12 @@ def __init__(
         otel_compression: Optional[Literal["gzip", "none"]] = None,
     ):
         self.public_key = public_key
+        self._uses_default_export_filter = should_export_span is None
         self._should_export_span = should_export_span or is_default_export_span
 
         self._app_root_lock = threading.Lock()
         self._span_export_expectation_by_id: Dict[str, bool] = {}
+        self._open_langfuse_spans_by_id: Dict[str, ReadableSpan] = {}
 
         env_flush_at = os.environ.get(LANGFUSE_FLUSH_AT, None)
         if flush_at is None and env_flush_at is not None:
@@ -282,6 +285,15 @@ def on_end(self, span: ReadableSpan) -> None:
                 )
                 return
 
+            if self._is_nested_in_langfuse_generation(span):
+                langfuse_logger.debug(
+                    "Trace: Dropping span already covered by its parent Langfuse "
+                    "generation | span_name='%s' | instrumentation_scope='%s'",
+                    span.name,
+                    self._get_scope_name(span),
+                )
+                return
+
             # Apply custom or default span filter
             try:
                 should_export = self._should_export_span(span)
@@ -331,6 +343,8 @@ def _mark_app_root_candidate(self, *, span: Span, parent_context: Context) -> No
             suppressed_by_parent_claim = propagated_trace_id == trace_id
 
             self._span_export_expectation_by_id[span_id] = expected_exported
+            if is_langfuse_span(cast(ReadableSpan, span)):
+                self._open_langfuse_spans_by_id[span_id] = cast(ReadableSpan, span)
 
             mark_app_root = (
                 expected_exported
@@ -349,6 +363,29 @@ def _cleanup_app_root_state(self, span: ReadableSpan) -> None:
 
         with self._app_root_lock:
             self._span_export_expectation_by_id.pop(span_id, None)
+            self._open_langfuse_spans_by_id.pop(span_id, None)
+
+    def _is_nested_in_langfuse_generation(self, span: ReadableSpan) -> bool:
+        if (
+            not self._uses_default_export_filter
+            or span.parent is None
+            or is_langfuse_span(span)
+            or not is_genai_span(span)
+        ):
+            return False
+
+        with self._app_root_lock:
+            parent = self._open_langfuse_spans_by_id.get(
+                format_span_id(span.parent.span_id)
+            )
+
+        return (
+            parent is not None
+            and (parent.attributes or {}).get(
+                LangfuseOtelSpanAttributes.OBSERVATION_TYPE
+            )
+            == "generation"
+        )
 
     def _is_expected_exported_at_start(self, span: Span) -> bool:
         readable_span = cast(ReadableSpan, span)

tests/conftest.py

@@ -102,11 +102,13 @@ def mock_init(self: Any, **kwargs: Any) -> None:
         from langfuse._client.span_filter import is_default_export_span
 
         self.public_key = kwargs.get("public_key", "test-public-key")
+        self._uses_default_export_filter = kwargs.get("should_export_span") is None
         self._should_export_span = (
             kwargs.get("should_export_span") or is_default_export_span
         )
         self._app_root_lock = threading.Lock()
         self._span_export_expectation_by_id = {}
+        self._open_langfuse_spans_by_id = {}
         BatchSpanProcessor.__init__(
             self,
             span_exporter=memory_exporter,

tests/unit/test_generation_child_span_filter.py

@@ -0,0 +1,123 @@
+from opentelemetry.sdk.trace import TracerProvider
+
+from langfuse._client.attributes import LangfuseOtelSpanAttributes
+from langfuse._client.constants import LANGFUSE_TRACER_NAME
+from langfuse._client.span_processor import LangfuseSpanProcessor
+
+PUBLIC_KEY = "test-public-key"
+SECRET_KEY = "test-secret-key"
+PROVIDER_SCOPE = "com.anthropic.sdk.python"
+GEN_AI_ATTRIBUTES = {"gen_ai.provider.name": "anthropic"}
+
+
+def _create_processor(memory_exporter, **kwargs):
+    tracer_provider = TracerProvider()
+    processor = LangfuseSpanProcessor(
+        public_key=PUBLIC_KEY,
+        secret_key=SECRET_KEY,
+        base_url="http://test-host",
+        span_exporter=memory_exporter,
+        **kwargs,
+    )
+    tracer_provider.add_span_processor(processor)
+
+    return tracer_provider, processor
+
+
+def _langfuse_tracer(tracer_provider):
+    return tracer_provider.get_tracer(
+        LANGFUSE_TRACER_NAME,
+        "test",
+        attributes={"public_key": PUBLIC_KEY},
+    )
+
+
+def _exported_span_names(memory_exporter):
+    return sorted(span.name for span in memory_exporter.get_finished_spans())
+
+
+def _run_provider_call_inside(tracer_provider, observation_type):
+    langfuse_tracer = _langfuse_tracer(tracer_provider)
+    provider_tracer = tracer_provider.get_tracer(PROVIDER_SCOPE)
+
+    with langfuse_tracer.start_as_current_span("parent") as parent:
+        # The SDK sets the observation type after the span has started
+        parent.set_attribute(
+            LangfuseOtelSpanAttributes.OBSERVATION_TYPE, observation_type
+        )
+        with provider_tracer.start_as_current_span(
+            "anthropic.messages.create", attributes=GEN_AI_ATTRIBUTES
+        ):
+            pass
+
+
+def test_drops_provider_genai_span_nested_in_langfuse_generation(
+    memory_exporter,
+):
+    tracer_provider, processor = _create_processor(memory_exporter)
+
+    _run_provider_call_inside(tracer_provider, "generation")
+    processor.force_flush()
+
+    assert _exported_span_names(memory_exporter) == ["parent"]
+    assert processor._open_langfuse_spans_by_id == {}
+
+
+def test_keeps_provider_genai_span_nested_in_langfuse_span(memory_exporter):
+    tracer_provider, processor = _create_processor(memory_exporter)
+
+    _run_provider_call_inside(tracer_provider, "span")
+    processor.force_flush()
+
+    assert _exported_span_names(memory_exporter) == [
+        "anthropic.messages.create",
+        "parent",
+    ]
+
+
+def test_keeps_provider_genai_span_without_langfuse_parent(memory_exporter):
+    tracer_provider, processor = _create_processor(memory_exporter)
+    provider_tracer = tracer_provider.get_tracer(PROVIDER_SCOPE)
+
+    with provider_tracer.start_as_current_span(
+        "anthropic.messages.create", attributes=GEN_AI_ATTRIBUTES
+    ):
+        pass
+    processor.force_flush()
+
+    assert _exported_span_names(memory_exporter) == ["anthropic.messages.create"]
+
+
+def test_custom_should_export_span_keeps_nested_provider_genai_span(
+    memory_exporter,
+):
+    tracer_provider, processor = _create_processor(
+        memory_exporter, should_export_span=lambda span: True
+    )
+
+    _run_provider_call_inside(tracer_provider, "generation")
+    processor.force_flush()
+
+    assert _exported_span_names(memory_exporter) == [
+        "anthropic.messages.create",
+        "parent",
+    ]
+
+
+def test_drops_provider_genai_span_nested_in_client_generation(
+    langfuse_memory_client, memory_exporter
+):
+    provider_tracer = langfuse_memory_client._resources.tracer_provider.get_tracer(
+        PROVIDER_SCOPE
+    )
+
+    with langfuse_memory_client.start_as_current_observation(
+        name="ChatAnthropic", as_type="generation"
+    ):
+        with provider_tracer.start_as_current_span(
+            "anthropic.messages.create", attributes=GEN_AI_ATTRIBUTES
+        ):
+            pass
+    langfuse_memory_client.flush()
+
+    assert _exported_span_names(memory_exporter) == ["ChatAnthropic"]

tests/unit/test_otel.py

@@ -121,11 +121,13 @@ def mock_init(self, **kwargs):
             from langfuse._client.span_filter import is_default_export_span
 
             self.public_key = kwargs.get("public_key", "test-key")
+            self._uses_default_export_filter = kwargs.get("should_export_span") is None
             self._should_export_span = (
                 kwargs.get("should_export_span") or is_default_export_span
             )
             self._app_root_lock = threading.Lock()
             self._span_export_expectation_by_id = {}
+            self._open_langfuse_spans_by_id = {}
             BatchSpanProcessor.__init__(
                 self,
                 span_exporter=memory_exporter,
@@ -2380,11 +2382,13 @@ def mock_processor_init(self, **kwargs):
             from langfuse._client.span_filter import is_default_export_span
 
             self.public_key = kwargs.get("public_key", "test-key")
+            self._uses_default_export_filter = kwargs.get("should_export_span") is None
             self._should_export_span = (
                 kwargs.get("should_export_span") or is_default_export_span
             )
             self._app_root_lock = threading.Lock()
             self._span_export_expectation_by_id = {}
+            self._open_langfuse_spans_by_id = {}
             # Use the appropriate exporter based on the project key
             if self.public_key == project1_key:
                 exporter = exporter_project1
@@ -2759,11 +2763,13 @@ def mock_processor_init(self, **kwargs):
             from langfuse._client.span_filter import is_default_export_span
 
             self.public_key = kwargs.get("public_key", "test-key")
+            self._uses_default_export_filter = kwargs.get("should_export_span") is None
             self._should_export_span = (
                 kwargs.get("should_export_span") or is_default_export_span
             )
             self._app_root_lock = threading.Lock()
             self._span_export_expectation_by_id = {}
+            self._open_langfuse_spans_by_id = {}
 
             # For testing, use the appropriate exporter based on setup
             exporter = kwargs.get("_test_exporter", blocked_exporter)