diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index 438551fd6..16874c4d2 100644 --- a/langfuse/_client/span_processor.py +++ b/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) diff --git a/tests/conftest.py b/tests/conftest.py index 97a78d212..a4795b50e 100644 --- a/tests/conftest.py +++ b/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, diff --git a/tests/unit/test_generation_child_span_filter.py b/tests/unit/test_generation_child_span_filter.py new file mode 100644 index 000000000..9bd47cfdc --- /dev/null +++ b/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"] diff --git a/tests/unit/test_otel.py b/tests/unit/test_otel.py index dfe1336fc..4deda5efb 100644 --- a/tests/unit/test_otel.py +++ b/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)