Render embedding observability (0059b)

conformance.toml

@@ -601,14 +601,11 @@ since = "0.13.0"
 # Spec v0.54.0 (proposal 0059).  Retrieval-provider capability —
 # the ``EmbeddingProvider`` protocol + ``EmbeddingEvent`` /
 # ``EmbeddingFailedEvent`` typed variants + OTel/Langfuse embedding
-# mapping.  Python has not yet shipped the embedding surface, so the
-# capability is not-yet.  The one piece adopted at this pin is the
-# proposal's cross-spec consequence: the observer-level privacy flag
-# ``disable_llm_payload`` is renamed ``disable_provider_payload`` (the
-# §5.5.4 rename touches existing LLM-payload gating, so it lands with
-# the pin even though the embedding capability does not).
+# mapping.
 [proposals."0059"]
-status = "not-yet"
+status = "implemented"
+since = "0.16.0"
+note = "Retrieval-provider embedding capability. The EmbeddingProvider protocol (retrieval-provider §3) ships as OpenAIEmbeddingProvider against POST /v1/embeddings (protocol fixtures 001-005 pass). The provider dispatches the typed EmbeddingEvent / EmbeddingFailedEvent (graph-engine §6) per embed() call, mutually exclusive, alongside the §7-category exception on failure. Both bundled observers render them: the OTelObserver emits an openarmature.embedding.complete span (observability §5.5.8) parented lineage-aware under the calling node, carrying the GenAI semconv subset (gen_ai.system / request.model / response.model / response.id / usage.input_tokens, gated by disable_genai_semconv) + the OA openarmature.embedding.input_count / .dimensions identity attrs + the payload-gated (disable_provider_payload) .input.strings / .request.extras -- no gen_ai.operation.name (deferred per the stable-only adoption policy), not gated by disable_llm_spans; failure renders ERROR status + error.type + an exception event + openarmature.error.category. The LangfuseObserver renders a dedicated Embedding observation (asType embedding, §8.4.5) with model = response_model-or-model, usageDetails.input, the openarmature_input_count / _dimensions / _response_id metadata, and the payload-gated input strings + output vectors (the output sourced from EmbeddingEvent.output_vectors per proposal 0089); failure renders an ERROR-level Embedding observation with error_category as statusMessage, error_type / error_message in metadata, and no output. Conformance fixtures 074-083 + 137 pass. The §11 embedding-metrics path (fixture 089) is proposal 0067's metrics surface (one GenAI instrument across LLM + embedding calls), tracked and deferred under 0067, not part of 0059's own observability surface. Wire-format details (0079) + input_type (0077) + rerank (0060) are separate proposals."
 
 # Spec v0.70.0 (proposal 0060).  Retrieval-provider rerank protocol — the
 # ``RerankProvider`` surface + ``RerankEvent`` / ``RerankFailedEvent`` typed
@@ -655,7 +652,7 @@ since = "0.14.0"
 [proposals."0067"]
 status = "partial"
 since = "0.15.0"
-note = "OTel GenAI metrics (observability §11): an opt-in enable_metrics flag (default off, normative name) on the bundled OTelObserver, independent of span emission (§11.1). When on, two OA-namespaced histograms record per provider-call ATTEMPT from the python-internal LlmRetryAttemptEvent (the per-attempt LLM-span source since 0050): openarmature.gen_ai.client.token.usage ({token}; two observations -- input + output token counts from the response usage record, openarmature.gen_ai.token.type dim) and openarmature.gen_ai.client.operation.duration (s; once per attempt INCLUDING failed attempts, error.type dim on failure), both configured with the §11.2 explicit bucket advisories. Dimensions: openarmature.gen_ai.operation ('chat'), gen_ai.request.model + gen_ai.system (recognized-core, used directly), openarmature.gen_ai.token.type, error.type. The Meter comes from the configured MeterProvider (injectable; falls back to the OTel global, which is the no-op meter when none is set). PARTIAL: the embedding-call metrics (the §11 embedding path, fixture 089) are deferred -- the embedding capability (proposal 0059, observability §5.5.8 / §5.5.9) is unimplemented in python until v0.16.0, so there is no embedding event/provider to record from. The LLM path (fixtures 088 / 090 / 091) is implemented and wired via a private MeterProvider + InMemoryMetricReader (the §6.9 metric-capture primitive). No Langfuse change (metrics are OTel-only). Streaming / server / rerank metrics + the cutover to the upstream gen_ai.client.* instrument names are out of scope per the proposal."
+note = "OTel GenAI metrics (observability §11): an opt-in enable_metrics flag (default off, normative name) on the bundled OTelObserver, independent of span emission (§11.1). When on, two OA-namespaced histograms record per provider-call ATTEMPT from the python-internal LlmRetryAttemptEvent (the per-attempt LLM-span source since 0050): openarmature.gen_ai.client.token.usage ({token}; two observations -- input + output token counts from the response usage record, openarmature.gen_ai.token.type dim) and openarmature.gen_ai.client.operation.duration (s; once per attempt INCLUDING failed attempts, error.type dim on failure), both configured with the §11.2 explicit bucket advisories. Dimensions: openarmature.gen_ai.operation ('chat'), gen_ai.request.model + gen_ai.system (recognized-core, used directly), openarmature.gen_ai.token.type, error.type. The Meter comes from the configured MeterProvider (injectable; falls back to the OTel global, which is the no-op meter when none is set). PARTIAL: the embedding-call metrics (the §11 embedding path, fixture 089) are deferred -- the embedding capability (proposal 0059, observability §5.5.8 / §5.5.9) now ships (v0.16.0) and dispatches EmbeddingEvent, so the remaining gap is wiring the §11 metric instruments to the embedding event (they record from the LLM per-attempt event only today). The LLM path (fixtures 088 / 090 / 091) is implemented and wired via a private MeterProvider + InMemoryMetricReader (the §6.9 metric-capture primitive). No Langfuse change (metrics are OTel-only). Streaming / server / rerank metrics + the cutover to the upstream gen_ai.client.* instrument names are out of scope per the proposal."
 
 # Spec v0.57.0 (proposal 0068).  Failure-isolation event structured cause
 # chain (pipeline-utilities §6.3).  ``caught_exception`` gains a ``chain`` of

src/openarmature/graph/events.py

@@ -812,6 +812,14 @@ class EmbeddingEvent:
     input_strings: list[str]
     input_count: int
     dimensions: int | None
+    # Spec graph-engine §6 (proposal 0089, v0.84.0): the output embedding
+    # vectors (sourced from EmbeddingResponse.vectors at dispatch). Populated
+    # unconditionally on the success event -- payload-bearing like
+    # input_strings, gated observer-side at the rendering boundary (OTel
+    # disable_provider_payload, Langfuse equivalents) per observability
+    # §5.5.9. The §8.4.5 embedding.output mapping reads this field, not the
+    # response object. No output field on EmbeddingFailedEvent (no response).
+    output_vectors: list[list[float]]
     request_params: Mapping[str, Any]
     request_extras: Mapping[str, Any]
     active_prompt: Any

src/openarmature/observability/langfuse/adapter.py

@@ -460,6 +460,65 @@ def tool(
             )
         return _SpanHandle(obs)
 
+    def embedding(
+        self,
+        *,
+        trace_id: str,
+        name: str | None = None,
+        model: str | None = None,
+        usage: LangfuseUsage | None = None,
+        metadata: dict[str, Any] | None = None,
+        parent_observation_id: str | None = None,
+        level: ObservationLevel = "DEFAULT",
+        status_message: str | None = None,
+        input: Any = None,
+        output: Any = None,
+        start_time: datetime | None = None,
+    ) -> LangfuseSpanHandle:
+        # v4 unifies observations under start_observation(as_type=); an
+        # Embedding observation routes through as_type="embedding" (proposal
+        # 0059), the SDK's native LangfuseEmbedding span. It carries model +
+        # usage like a Generation but is a distinct type; usage maps to v4's
+        # usage_details. Back-date via the private OTel tracer when start_time
+        # is supplied, exactly as generation()/tool() do.
+        extra_kwargs: dict[str, Any] = {"model": model, "input": input, "output": output}
+        if usage is not None:
+            usage_details: dict[str, int] = {}
+            if usage.input is not None:
+                usage_details["input"] = usage.input
+            if usage.output is not None:
+                usage_details["output"] = usage.output
+            if usage.total is not None:
+                usage_details["total"] = usage.total
+            extra_kwargs["usage_details"] = usage_details
+        present_extra = {k: v for k, v in extra_kwargs.items() if v is not None}
+        if start_time is not None:
+            from langfuse._client.span import LangfuseEmbedding
+
+            obs = self._start_back_dated_observation(
+                LangfuseEmbedding,
+                trace_id=trace_id,
+                name=name,
+                metadata=metadata,
+                parent_observation_id=parent_observation_id,
+                level=level,
+                status_message=status_message,
+                start_time=start_time,
+                **present_extra,
+            )
+        else:
+            obs = self._start_observation(
+                as_type="embedding",
+                trace_id=trace_id,
+                name=name,
+                metadata=metadata,
+                parent_observation_id=parent_observation_id,
+                level=level,
+                status_message=status_message,
+                **present_extra,
+            )
+        return _SpanHandle(obs)
+
     def _start_back_dated_observation(
         self,
         observation_cls: type[Any],

src/openarmature/observability/langfuse/client.py

@@ -33,7 +33,7 @@
 from datetime import datetime
 from typing import Any, Literal, Protocol, runtime_checkable
 
-ObservationType = Literal["span", "generation", "event", "tool"]
+ObservationType = Literal["span", "generation", "event", "tool", "embedding"]
 
 # Langfuse-supported `level` values per spec §8.4.2 (statusMessage pair).
 ObservationLevel = Literal["DEFAULT", "DEBUG", "INFO", "WARNING", "ERROR"]
@@ -268,6 +268,31 @@ def tool(
         handle the caller ``.end()``s at outcome time."""
         ...
 
+    # Proposal 0059 embedding observability: a dedicated Embedding
+    # observation (Langfuse asType="embedding", spec observability §8.4.5),
+    # not a Generation with a discriminator.
+    def embedding(
+        self,
+        *,
+        trace_id: str,
+        name: str | None = None,
+        model: str | None = None,
+        usage: LangfuseUsage | None = None,
+        metadata: dict[str, Any] | None = None,
+        parent_observation_id: str | None = None,
+        level: ObservationLevel = "DEFAULT",
+        status_message: str | None = None,
+        input: Any = None,
+        output: Any = None,
+        start_time: datetime | None = None,
+    ) -> LangfuseSpanHandle:
+        """Open a dedicated Embedding observation. Like :meth:`generation`
+        for the model / usage surface but a distinct observation type:
+        carries ``model`` / ``usage`` (input tokens) / ``input`` (strings) /
+        ``output`` (vectors) / metadata / level. Returns a minimal handle
+        the caller ``.end()``s at outcome time."""
+        ...
+
     def force_flush(self, timeout_ms: int = 30_000) -> bool:
         """Flush any pending outbound buffer in the underlying sink.
 
@@ -529,6 +554,39 @@ def tool(
         trace.observations.append(observation)
         return _InMemorySpanHandle(observation=observation)
 
+    def embedding(
+        self,
+        *,
+        trace_id: str,
+        name: str | None = None,
+        model: str | None = None,
+        usage: LangfuseUsage | None = None,
+        metadata: dict[str, Any] | None = None,
+        parent_observation_id: str | None = None,
+        level: ObservationLevel = "DEFAULT",
+        status_message: str | None = None,
+        input: Any = None,
+        output: Any = None,
+        start_time: datetime | None = None,
+    ) -> LangfuseSpanHandle:
+        trace = self._get_trace(trace_id)
+        observation = LangfuseObservation(
+            id=self._mint_observation_id(),
+            type="embedding",
+            name=name,
+            metadata=dict(metadata) if metadata is not None else {},
+            parent_observation_id=parent_observation_id,
+            level=level,
+            status_message=status_message,
+            input=input,
+            output=output,
+            model=model,
+            usage=usage,
+            start_time=start_time,
+        )
+        trace.observations.append(observation)
+        return _InMemorySpanHandle(observation=observation)
+
     def force_flush(self, timeout_ms: int = 30_000) -> bool:
         # In-memory recorder has no outbound buffer; every observation
         # is captured synchronously on its create call. The ``timeout_ms``