diff --git a/src/everos/component/embedding/openai_provider.py b/src/everos/component/embedding/openai_provider.py index 1e0f436..b26b89e 100644 --- a/src/everos/component/embedding/openai_provider.py +++ b/src/everos/component/embedding/openai_provider.py @@ -90,8 +90,11 @@ class OpenAIEmbeddingProvider: """One ``/embeddings`` call, semaphore-guarded.""" # Wrap in an EMBEDDING-typed span so token usage lands on an embedding # observation (which Langfuse can price) rather than the enclosing - # retriever span. No-op when tracing is off. - with memory_span("everos.embedding", observation_type="embedding"): + # retriever span. nested_only: skip when there is no active trace (e.g. + # cascade-time indexing) so we don't spawn one root trace per chunk. + with memory_span( + "everos.embedding", observation_type="embedding", nested_only=True + ): async with self._semaphore: try: response = await self._client.embeddings.create( diff --git a/src/everos/core/observability/tracing/attributes.py b/src/everos/core/observability/tracing/attributes.py index d12d167..d184cf7 100644 --- a/src/everos/core/observability/tracing/attributes.py +++ b/src/everos/core/observability/tracing/attributes.py @@ -85,6 +85,13 @@ except ImportError: # pragma: no cover - only without the [otel] extra _OTEL_AVAILABLE = False +def _has_active_span() -> bool: + """True when a valid span context is currently active (a parent exists).""" + if not _OTEL_AVAILABLE: + return False + return _otel_trace.get_current_span().get_span_context().is_valid + + @contextmanager def memory_span( name: str, @@ -94,6 +101,7 @@ def memory_span( user_id: str | None = None, metadata: Mapping[str, Any] | None = None, tags: Sequence[str] = DEFAULT_TAGS, + nested_only: bool = False, ) -> Iterator[Any]: """Open a span named ``name`` and stamp the langfuse.* attributes. @@ -105,7 +113,14 @@ def memory_span( metadata: Flat mapping → ``langfuse.trace.metadata.``; None values are dropped rather than emitted as the string "None". tags: ``langfuse.trace.tags`` list. + nested_only: When True, only open a span if one is already active. + Calls that run outside any request trace (e.g. cascade-time + embedding during indexing) would otherwise each start a NEW root + trace — a per-chunk trace explosion — so they no-op instead. """ + if nested_only and not _has_active_span(): + yield _otel_trace.get_current_span() if _OTEL_AVAILABLE else None + return tracer = get_tracer("everos") with tracer.start_as_current_span(name) as span: span.set_attribute(LF_OBSERVATION_TYPE, observation_type) diff --git a/tests/unit/test_core/test_observability/test_span_attributes.py b/tests/unit/test_core/test_observability/test_span_attributes.py index dafe74d..fe8970c 100644 --- a/tests/unit/test_core/test_observability/test_span_attributes.py +++ b/tests/unit/test_core/test_observability/test_span_attributes.py @@ -97,6 +97,34 @@ def test_set_generation_usage_accumulates_across_calls( assert attrs["gen_ai.usage.output_tokens"] == 12 +def test_nested_only_span_skips_when_no_active_parent( + captured: InMemorySpanExporter, +) -> None: + # A nested_only span with no active parent must NOT start a new root trace + # (this is what prevented cascade-time embeddings from exploding into one + # orphan trace per chunk). + with memory_span( + "everos.embedding", observation_type="embedding", nested_only=True + ): + pass + force_flush() + assert captured.get_finished_spans() == () + + +def test_nested_only_span_opens_under_active_parent( + captured: InMemorySpanExporter, +) -> None: + with ( + memory_span("everos.memory.search", observation_type="retriever"), + memory_span("everos.embedding", observation_type="embedding", nested_only=True), + ): + pass + force_flush() + names = {s.name for s in captured.get_finished_spans()} + assert "everos.embedding" in names + assert "everos.memory.search" in names + + def test_set_generation_usage_outside_span_is_noop() -> None: # No active span → must not raise (and nothing to record). set_generation_usage(model="x", input_tokens=1, output_tokens=2)