diff --git a/conformance.toml b/conformance.toml index 7db595a..d84fbf8 100644 --- a/conformance.toml +++ b/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 diff --git a/src/openarmature/graph/events.py b/src/openarmature/graph/events.py index c6eddcf..b092a61 100644 --- a/src/openarmature/graph/events.py +++ b/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 diff --git a/src/openarmature/observability/langfuse/adapter.py b/src/openarmature/observability/langfuse/adapter.py index 6f926f3..9861bf0 100644 --- a/src/openarmature/observability/langfuse/adapter.py +++ b/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], diff --git a/src/openarmature/observability/langfuse/client.py b/src/openarmature/observability/langfuse/client.py index ba73a03..523e135 100644 --- a/src/openarmature/observability/langfuse/client.py +++ b/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`` diff --git a/src/openarmature/observability/langfuse/observer.py b/src/openarmature/observability/langfuse/observer.py index 716b106..8f48245 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -476,11 +476,14 @@ async def __call__( if isinstance(event, MetadataAugmentationEvent): self._handle_metadata_augmentation(event) return - # Proposal 0059 embedding events: the bundled Langfuse Embedding - # observation is a follow-up. Until it lands the events are safely - # ignored here rather than falling through to the NodeEvent phase - # dispatch (which would AttributeError on the absent ``phase``). + # Proposal 0059 embedding observability (observability §8.4.5): render + # the dedicated Langfuse Embedding observation (asType "embedding") + # under the calling node's Span observation. NOT gated by + # disable_llm_spans (scoped to LLM completion per §5.5.8); the + # input / output payload is gated by disable_provider_payload inside + # the handler. if isinstance(event, EmbeddingEvent | EmbeddingFailedEvent): + self._handle_embedding(event) return if event.phase == "started": self._open_started_observation(event) @@ -1878,6 +1881,110 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: ) handle.end(end_time=end_time) + # Spec: observability §8.4.5 (proposal 0059) Embedding observation + # (Langfuse asType="embedding"). The failure path is the generic §4.2 / + # §8.4.2 ERROR mapping with the §7 error category, mirroring the tool + # failure. + def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> None: + """Open + close a dedicated Embedding observation (Langfuse + ``asType="embedding"``) under the calling node's Span observation. + + Success (``EmbeddingEvent``): DEFAULT level, ``model`` = + ``response_model`` (falling back to the requested ``model``), + ``usage`` from the embedding token record, the OA identity metadata + (``openarmature_input_count`` / ``openarmature_dimensions`` / + ``openarmature_response_id``), and the payload-gated ``input`` + strings + ``output`` vectors. + + Failure (``EmbeddingFailedEvent``): ERROR level with the + ``error_category`` as the status message and ``error_type`` / + ``error_message`` in metadata, mirroring the tool failure. The + request-side ``input`` strings are still payload-gated; there is NO + ``output`` (no response received). + """ + from openarmature.observability.correlation import ( + current_correlation_id, + current_invocation_id, + ) + + invocation_id = current_invocation_id() + if invocation_id is None: + return + correlation_id = current_correlation_id() + if invocation_id not in self._inv_states: + self._open_trace_for_typed_event(invocation_id, correlation_id, event) + inv_state = self._inv_states[invocation_id] + end_time = datetime.now(UTC) + if event.latency_ms is not None: + start_time = end_time - timedelta(milliseconds=event.latency_ms) + else: + start_time = end_time + parent_observation_id = self._resolve_llm_parent_observation_id( + inv_state, + calling_namespace_prefix=event.namespace, + calling_attempt_index=event.attempt_index, + calling_fan_out_index=event.fan_out_index, + calling_branch_name=event.branch_name, + ) + # §8.4.5 metadata: input_count always; dimensions / response_id are + # response-derived (success-only). correlation_id + caller metadata + # mirror the other observation handlers' scoping rows. + metadata: dict[str, Any] = {} + if correlation_id is not None: + metadata["correlation_id"] = correlation_id + if event.fan_out_index is not None: + metadata["fan_out_index"] = event.fan_out_index + if event.branch_name is not None: + metadata["branch_name"] = event.branch_name + if event.caller_invocation_metadata is not None: + _apply_caller_metadata(metadata, event.caller_invocation_metadata) + input_value: Any = None + if not self.disable_provider_payload and event.input_strings: + input_value = self._maybe_truncate_for_input(event.input_strings) + if isinstance(event, EmbeddingEvent): + metadata["openarmature_input_count"] = event.input_count + if event.dimensions is not None: + metadata["openarmature_dimensions"] = event.dimensions + if event.response_id is not None: + metadata["openarmature_response_id"] = event.response_id + output_value: Any = None + if not self.disable_provider_payload and event.output_vectors: + output_value = self._maybe_truncate_for_input(event.output_vectors) + usage = LangfuseUsage(input=event.usage.input_tokens) if event.usage is not None else None + target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) + handle = self.client.embedding( + trace_id=target_trace_id, + name="openarmature.embedding.complete", + model=event.response_model or event.model, + usage=usage, + input=input_value, + output=output_value, + metadata=metadata, + parent_observation_id=parent_observation_id, + start_time=start_time, + ) + handle.end(end_time=end_time) + return + # Failure path: request-side input_count survives; the response-derived + # rows do not. No output. ERROR level + category-as-statusMessage. + metadata["openarmature_input_count"] = len(event.input_strings) + if event.error_type is not None: + metadata["error_type"] = event.error_type + metadata["error_message"] = event.error_message + target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) + handle = self.client.embedding( + trace_id=target_trace_id, + name="openarmature.embedding.complete", + model=event.model, + input=input_value, + metadata=metadata, + parent_observation_id=parent_observation_id, + level="ERROR", + status_message=event.error_category, + start_time=start_time, + ) + handle.end(end_time=end_time) + def _resolve_llm_parent_observation_id( self, inv_state: _InvState, @@ -2004,10 +2111,16 @@ def _open_trace_for_typed_event( self, invocation_id: str, correlation_id: str | None, - event: LlmCompletionEvent | LlmFailedEvent | ToolCallEvent | ToolCallFailedEvent, + event: LlmCompletionEvent + | LlmFailedEvent + | ToolCallEvent + | ToolCallFailedEvent + | EmbeddingEvent + | EmbeddingFailedEvent, ) -> None: - """Trace open path for a typed event (LLM completion / failure or - tool execution) arriving before any node-started event reached + """Trace open path for a typed event (LLM completion / failure, + tool execution, or embedding) arriving before any node-started event + reached this observer. Synthesizes the minimal trace shape from the typed event's scoping fields (all read generically).""" if event.namespace: diff --git a/src/openarmature/observability/otel/observer.py b/src/openarmature/observability/otel/observer.py index a5c04df..9d0236f 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -599,6 +599,8 @@ class OTelObserver: | LlmRetryAttemptEvent | ToolCallEvent | ToolCallFailedEvent + | EmbeddingEvent + | EmbeddingFailedEvent | None, ], None, @@ -708,6 +710,8 @@ def _run_enrichers( | LlmRetryAttemptEvent | ToolCallEvent | ToolCallFailedEvent + | EmbeddingEvent + | EmbeddingFailedEvent | None, ) -> None: """Invoke configured enrichers against ``span`` before @@ -774,12 +778,13 @@ async def __call__( # consumers, so the OTel observer ignores them here. if isinstance(event, LlmCompletionEvent | LlmFailedEvent): return - # Proposal 0059 embedding events: the bundled OTel embedding span - # (openarmature.embedding.complete) is a follow-up. Until it lands - # the events are safely ignored here rather than falling through to - # the NodeEvent phase dispatch (which would AttributeError on the - # absent ``phase`` field). + # Proposal 0059 embedding observability (observability §5.5.8): emit + # the openarmature.embedding.complete span from the typed embedding + # events. NOT gated by disable_llm_spans (that flag is scoped to LLM + # completion spans per §5.5.8); only the payload / GenAI-semconv + # attribute subsets are gated, inside the handler. if isinstance(event, EmbeddingEvent | EmbeddingFailedEvent): + self._handle_embedding(event) return # Proposal 0063 tool-execution observability: emit the # openarmature.tool.call span from the typed tool events. @@ -1685,6 +1690,115 @@ def _handle_tool_call(self, event: ToolCallEvent | ToolCallFailedEvent) -> None: self._run_enrichers(span, event) span.end(end_time=end_time_ns) + # Spec: observability §5.5.8 (proposal 0059) embedding span. Payload + # gating is §5.5.4 + the §5.5.5 truncation contract; gen_ai.operation.name + # is deferred per the stable-only adoption policy. + def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> None: + """Emit an ``openarmature.embedding.complete`` span for an + ``EmbeddingProvider.embed()`` call, parented under the calling node. + + An ``EmbeddingEvent`` renders OK with the GenAI response subset + + the OA embedding identity attributes; an ``EmbeddingFailedEvent`` + renders ERROR with the standard OTel ``error.type`` attribute, an + exception event carrying the message, and the + ``openarmature.error.category`` attribute. ``input.strings`` / + ``request.extras`` are payload, gated by ``disable_provider_payload`` + (with the standard truncation cap); the GenAI semconv subset is gated + by ``disable_genai_semconv``. No ``gen_ai.operation.name`` is emitted; + operation discrimination is via the span name + provider. + ``disable_llm_spans`` does not gate this span. + """ + from openarmature.observability.correlation import ( + current_correlation_id, + current_invocation_id, + ) + + invocation_id = current_invocation_id() + if invocation_id is None: + return + inv_state = self._inv_state_for(invocation_id) + # Back-date start_time by latency_ms so the span duration reflects the + # adapter-boundary measurement, mirroring the LLM / tool spans. + end_time_ns = time.time_ns() + if event.latency_ms is not None: + start_time_ns = end_time_ns - int(event.latency_ms * 1_000_000) + else: + start_time_ns = end_time_ns + parent_ctx = self._resolve_llm_parent( + inv_state, + invocation_id, + calling_namespace_prefix=event.namespace, + calling_attempt_index=event.attempt_index, + calling_fan_out_index=event.fan_out_index, + calling_branch_name=event.branch_name, + ) + attrs: dict[str, Any] = {} + cid = current_correlation_id() + if cid is not None: + attrs["openarmature.correlation_id"] = cid + if event.caller_invocation_metadata is not None: + _apply_caller_metadata(attrs, event.caller_invocation_metadata) + # OA-namespace embedding identity attributes (ungated). input_count is + # available on both variants (the failure event carries input_strings); + # dimensions is response-derived, so success-only. + if isinstance(event, EmbeddingEvent): + attrs["openarmature.embedding.input_count"] = event.input_count + if event.dimensions is not None: + attrs["openarmature.embedding.dimensions"] = event.dimensions + else: + attrs["openarmature.embedding.input_count"] = len(event.input_strings) + # GenAI semconv subset, gated by disable_genai_semconv. The response + # subset (response.model / response.id / usage.input_tokens) is + # success-only; the request subset (system / request.model) emits on + # both variants. + if not self.disable_genai_semconv: + attrs["gen_ai.system"] = event.provider + attrs["gen_ai.request.model"] = event.model + if isinstance(event, EmbeddingEvent): + if event.response_model is not None: + attrs["gen_ai.response.model"] = event.response_model + if event.response_id is not None: + attrs["gen_ai.response.id"] = event.response_id + if event.usage is not None: + attrs["gen_ai.usage.input_tokens"] = event.usage.input_tokens + # Payload-gated (disable_provider_payload): input strings + extras. + if not self.disable_provider_payload: + if event.input_strings: + serialized = _serialize_for_attribute(event.input_strings) + attrs["openarmature.embedding.input.strings"] = _truncate_for_attribute( + serialized, self.payload_max_bytes + ) + if event.request_extras: + serialized_extras = _serialize_for_attribute(event.request_extras) + attrs["openarmature.embedding.request.extras"] = _truncate_for_attribute( + serialized_extras, self.payload_max_bytes + ) + span = self._tracer.start_span( + name="openarmature.embedding.complete", + context=cast("Any", parent_ctx), + kind=SpanKind.CLIENT, + attributes=attrs, + start_time=start_time_ns, + ) + if isinstance(event, EmbeddingFailedEvent): + # Generic §4.2 / §8.4.2 error mapping: ERROR status + error.type + + # an OTel exception event + the §7 category attribute. + if event.error_type is not None: + span.set_attribute("error.type", event.error_type) + span.add_event( + "exception", + attributes={ + "exception.type": event.error_type or "", + "exception.message": event.error_message, + }, + ) + span.set_attribute("openarmature.error.category", event.error_category) + span.set_status(Status(StatusCode.ERROR, description=event.error_category)) + else: + span.set_status(Status(StatusCode.OK)) + self._run_enrichers(span, event) + span.end(end_time=end_time_ns) + def _handle_failure_isolated(self, event: FailureIsolatedEvent) -> None: """Emit a zero-duration ``openarmature.failure_isolated`` span for a FailureIsolationMiddleware catch. diff --git a/src/openarmature/retrieval/providers/openai.py b/src/openarmature/retrieval/providers/openai.py index 55fb276..0cf4564 100644 --- a/src/openarmature/retrieval/providers/openai.py +++ b/src/openarmature/retrieval/providers/openai.py @@ -419,6 +419,11 @@ def _build_embedding_event( input_strings=input_strings, input_count=len(input_strings), dimensions=response.dimensions, + # Populated unconditionally on success per observability §5.5.9; + # privacy gating is observer-side at rendering (symmetric with + # input_strings). Sources EmbeddingEvent.output_vectors from the + # parsed response vectors for the §8.4.5 embedding.output mapping. + output_vectors=response.vectors, request_params=request_params, request_extras=request_extras, active_prompt=active_prompt, diff --git a/tests/conformance/test_fixture_parsing.py b/tests/conformance/test_fixture_parsing.py index 0c466f6..afa1f92 100644 --- a/tests/conformance/test_fixture_parsing.py +++ b/tests/conformance/test_fixture_parsing.py @@ -496,33 +496,37 @@ def _id(case: tuple[str, Path]) -> str: "graph-engine/033-drain-events-for-parallel-branches-coverage": ( "Proposal 0054 fixture-shape models pending; contract pinned by unit tests" ), - # Proposal 0059 (retrieval-provider / embedding, v0.54.0): the - # observability embedding-event fixtures (074-083) model the - # EmbeddingEvent / EmbeddingFailedEvent + embedding span / Langfuse - # observation surface, which python does not implement (0059 is - # not-yet in the manifest; only its cross-spec disable_provider_payload - # rename is adopted). The retrieval-provider fixtures are a new - # capability dir this runner does not parse. - "observability/074-embedding-event-dispatch": "Proposal 0059 embedding events; not implemented", + # Proposal 0059 (retrieval-provider / embedding) — the embedding + # typed-collector fixtures (074-081) share the ``expected.observers`` + # shape of 050-068 / 092-095 (which the harness schema models loosely, + # hence the same parser-deferral); they still RUN via test_observability. + # The OTel-span (082) and Langfuse (083, 137) embedding fixtures parse + # fine against the existing span_tree / langfuse_trace shapes and are NOT + # deferred (they run in test_observability too). + "observability/074-embedding-event-dispatch": ( + "Proposal 0059 typed-event-collector shape; runs in test_observability" + ), "observability/075-embedding-failure-event-dispatch-on-provider-unavailable": ( - "Proposal 0059 embedding events; not implemented" + "Proposal 0059 typed-event-collector shape; runs in test_observability" + ), + "observability/076-embedding-event-mutual-exclusion": ( + "Proposal 0059 typed-event-collector shape; runs in test_observability" + ), + "observability/077-embedding-event-call-id-distinct": ( + "Proposal 0059 typed-event-collector shape; runs in test_observability" ), - "observability/076-embedding-event-mutual-exclusion": "Proposal 0059 embedding events; not implemented", - "observability/077-embedding-event-call-id-distinct": "Proposal 0059 embedding events; not implemented", "observability/078-embedding-event-input-strings-populated": ( - "Proposal 0059 embedding events; not implemented" + "Proposal 0059 typed-event-collector shape; runs in test_observability" ), "observability/079-embedding-event-request-params-populated": ( - "Proposal 0059 embedding events; not implemented" + "Proposal 0059 typed-event-collector shape; runs in test_observability" ), "observability/080-embedding-event-input-count-and-dimensions-populated": ( - "Proposal 0059 embedding events; not implemented" + "Proposal 0059 typed-event-collector shape; runs in test_observability" ), "observability/081-embedding-event-active-prompt-populated": ( - "Proposal 0059 embedding events; not implemented" + "Proposal 0059 typed-event-collector shape; runs in test_observability" ), - "observability/082-otel-embedding-span-attributes": "Proposal 0059 embedding events; not implemented", - "observability/083-langfuse-embedding-observation": "Proposal 0059 embedding events; not implemented", # Proposal 0060 (retrieval-provider rerank, v0.70.0): the rerank # observability fixtures (099-109) model the RerankEvent / # RerankFailedEvent + rerank span / Langfuse Retriever / rerank-metrics diff --git a/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index bafbee0..7f24789 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -214,6 +214,25 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "096-tool-call-payload-gating", "097-otel-tool-span-attributes", "098-langfuse-tool-observation", + # v0.16.0 — proposal 0059 embedding observability (0059b). A + # calls_embed node awaits OpenAIEmbeddingProvider.embed() inside the + # node body; the typed EmbeddingEvent / EmbeddingFailedEvent drive the + # typed-event collector (074-081), the OTel embedding span + # (082, openarmature.embedding.complete), and the Langfuse Embedding + # observation (083 success, 137 failure -- the proposal 0089 + # output_vectors-sourced rendering). 089 (embedding metrics) stays + # deferred with the §11 embedding-metrics path. + "074-embedding-event-dispatch", + "075-embedding-failure-event-dispatch-on-provider-unavailable", + "076-embedding-event-mutual-exclusion", + "077-embedding-event-call-id-distinct", + "078-embedding-event-input-strings-populated", + "079-embedding-event-request-params-populated", + "080-embedding-event-input-count-and-dimensions-populated", + "081-embedding-event-active-prompt-populated", + "082-otel-embedding-span-attributes", + "083-langfuse-embedding-observation", + "137-langfuse-embedding-failure-observation", # v0.70.1 — proposal 0075 callable-branch span shape (observability # §5.7). The ORIGINAL fixture 110 (span shape + skip-emits-no-span); # the branch_count assertion arrives with the v0.73.1 pin (v0.16.0). @@ -222,9 +241,10 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: ) -_EMBEDDING_DEFER = ( - "embedding capability (proposal 0059) unimplemented until v0.16.0; " - "no embedding event/provider to record from" +_EMBEDDING_METRICS_DEFER = ( + "embedding observability (074-083 + 137) is wired (proposal 0059b); only the " + "§11 embedding-metrics path stays deferred -- the OTelObserver records metrics " + "from the LLM per-attempt event, not yet from EmbeddingEvent" ) _RERANK_DEFER = ( @@ -247,28 +267,12 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "039-nested-lineage-augmentation": ( "nested-case Langfuse harness wiring not yet implemented (proposal 0045 nested fan-out)" ), - # Embedding observability (proposals 0059 / 0067 §11). The embedding - # capability is unshipped until v0.16.0; the LLM-path equivalents run. - **{ - fixture_id: _EMBEDDING_DEFER - for fixture_id in ( - "074-embedding-event-dispatch", - "075-embedding-failure-event-dispatch-on-provider-unavailable", - "076-embedding-event-mutual-exclusion", - "077-embedding-event-call-id-distinct", - "078-embedding-event-input-strings-populated", - "079-embedding-event-request-params-populated", - "080-embedding-event-input-count-and-dimensions-populated", - "081-embedding-event-active-prompt-populated", - "082-otel-embedding-span-attributes", - "083-langfuse-embedding-observation", - "089-embedding-metrics-token-and-duration", - # proposal 0089 (v0.84.0) embedding failure observation -- the - # EmbeddingFailedEvent ERROR-level Langfuse rendering; blocked on - # the same unimplemented embedding capability. - "137-langfuse-embedding-failure-observation", - ) - }, + # Embedding observability (proposal 0067 §11). 074-083 + 137 are wired + # (proposal 0059b -- see _SUPPORTED_FIXTURES). Only the §11 embedding- + # metrics path (089) stays deferred: the OTelObserver records metrics from + # the LLM per-attempt event only; extending the metric instruments to the + # embedding path rides the §11 embedding-metrics work, not 0059b. + "089-embedding-metrics-token-and-duration": _EMBEDDING_METRICS_DEFER, # Rerank observability (proposal 0060, v0.70.0). The rerank protocol is # unshipped in python until v0.16.0; no rerank provider/event exists. **{ @@ -672,6 +676,20 @@ async def test_observability_fixture(fixture_path: Path) -> None: "098-langfuse-tool-observation", }: await _run_tool_fixture(spec) + elif fixture_id in { + "074-embedding-event-dispatch", + "075-embedding-failure-event-dispatch-on-provider-unavailable", + "076-embedding-event-mutual-exclusion", + "077-embedding-event-call-id-distinct", + "078-embedding-event-input-strings-populated", + "079-embedding-event-request-params-populated", + "080-embedding-event-input-count-and-dimensions-populated", + "081-embedding-event-active-prompt-populated", + "082-otel-embedding-span-attributes", + "083-langfuse-embedding-observation", + "137-langfuse-embedding-failure-observation", + }: + await _run_embedding_fixture(spec) elif fixture_id in { "043-get-invocation-metadata-roundtrip", "044-get-invocation-metadata-fan-out-scoping", @@ -3983,8 +4001,10 @@ def _assert_langfuse_observation_tree( ) -> None: """Recursively match expected observations against the trace's flat observation list (linked by parent_observation_id). type + name are - matched exactly; level / input / output exactly when present; - metadata is subset-matched.""" + matched exactly; level / statusMessage / model / input / output exactly + when present; usageDetails is subset-matched against the observation's + usage record; metadata is subset-matched (values honor the + ```` token via ``_value_matches``).""" # Mutable copy: each matched observation is consumed so two # same-shape expected siblings can't both bind to one actual. remaining = list(trace.children_of(parent_id)) @@ -4002,6 +4022,17 @@ def _assert_langfuse_observation_tree( remaining.remove(match) if "level" in exp: assert match.level == exp["level"], f"{exp_name!r}: level {match.level!r} != {exp['level']!r}" + if "statusMessage" in exp: + assert match.status_message == exp["statusMessage"], ( + f"{exp_name!r}: statusMessage {match.status_message!r} != {exp['statusMessage']!r}" + ) + if "model" in exp: + assert match.model == exp["model"], f"{exp_name!r}: model {match.model!r} != {exp['model']!r}" + if "usageDetails" in exp: + assert match.usage is not None, f"{exp_name!r}: expected usageDetails but usage is None" + for uk, uv in cast("dict[str, Any]", exp["usageDetails"]).items(): + actual_usage = getattr(match.usage, uk, None) + assert actual_usage == uv, f"{exp_name!r}: usageDetails.{uk} {actual_usage!r} != {uv!r}" if "input" in exp: assert match.input == exp["input"], f"{exp_name!r}: input {match.input!r} != {exp['input']!r}" if "output" in exp: @@ -4009,7 +4040,7 @@ def _assert_langfuse_observation_tree( f"{exp_name!r}: output {match.output!r} != {exp['output']!r}" ) for key, val in cast("dict[str, Any]", exp.get("metadata") or {}).items(): - assert match.metadata.get(key) == val, ( + assert _value_matches(match.metadata.get(key), val), ( f"{exp_name!r}: metadata.{key} {match.metadata.get(key)!r} != {val!r}" ) children = cast("list[dict[str, Any]] | None", exp.get("children")) @@ -4107,6 +4138,226 @@ async def _run_tool_case(case: Mapping[str, Any]) -> None: ) +# --------------------------------------------------------------------------- +# Proposal 0059 — embedding observability fixtures (074-083, 137) +# --------------------------------------------------------------------------- +# +# A calls_embed node constructs an OpenAIEmbeddingProvider on an +# httpx.MockTransport built from the fixture's mock_embedding block and awaits +# provider.embed(inputs) INSIDE the node body, so the calling-node ContextVars +# are set and the EmbeddingEvent / EmbeddingFailedEvent flow through the +# observer queue. One graph builder serves all three assertion shapes: +# typed-event-collector (074-081), OTel span_tree (082), and the Langfuse +# Embedding observation tree (083 / 137). Dispatch is by which expected.* key +# appears, mirroring the tool path (_run_tool_case). + + +def _embedding_model_from_first_response(case: Mapping[str, Any]) -> str | None: + """Return the ``model`` on the first ``mock_embedding`` response body, so + the provider binds to the model the fixture's expected event reports.""" + responses = cast("list[dict[str, Any]] | None", case.get("mock_embedding")) or [] + if not responses: + return None + body = cast("dict[str, Any] | None", responses[0].get("body")) or {} + model = body.get("model") + return model if isinstance(model, str) else None + + +def _build_embedding_runtime_config(config_spec: Mapping[str, Any] | None) -> Any: + """Build an EmbeddingRuntimeConfig from a ``calls_embed.config`` block + (fixture 079: ``dimensions``), or None when absent.""" + if not config_spec: + return None + from openarmature.retrieval import EmbeddingRuntimeConfig + + return EmbeddingRuntimeConfig(**dict(config_spec)) + + +def _make_embedding_node_body(node_spec: Mapping[str, Any], provider: Any, case: Mapping[str, Any]) -> Any: + calls_embed = cast("dict[str, Any]", node_spec["calls_embed"]) + stores_in = cast("str | None", calls_embed.get("stores_response_in")) + config = _build_embedding_runtime_config(cast("dict[str, Any] | None", calls_embed.get("config"))) + # A node may render a prompt first (081, RAG retrieval): the rendered + # PromptResult is stamped active for the embed() call and supplies the + # input text (one-element list) when input_from_rendered_prompt is set. + renders_prompt_name = cast("str | None", node_spec.get("renders_prompt")) + prompt_result = _render_prompt_result(case, renders_prompt_name) if renders_prompt_name else None + input_from_rendered = bool(calls_embed.get("input_from_rendered_prompt", False)) + explicit_input = cast("list[str] | None", calls_embed.get("input")) + + async def body(_state: Any) -> Mapping[str, Any]: + if input_from_rendered and prompt_result is not None: + inputs = [cast("str", prompt_result.messages[0].content)] + else: + inputs = list(explicit_input or []) + if prompt_result is not None: + from openarmature.prompts import with_active_prompt + + with with_active_prompt(prompt_result): + response = await provider.embed(inputs, config=config) + else: + response = await provider.embed(inputs, config=config) + return {stores_in: response.model_dump()} if stores_in else {} + + return body + + +def _build_embedding_graph( + case: Mapping[str, Any], + *, + populate_caller_metadata: bool, +) -> tuple[Any, type[Any], list[Any]]: + """Build a graph whose nodes are calls_embed. Every embed node gets its own + OpenAIEmbeddingProvider sharing one MockTransport that replays the case's + ``mock_embedding`` queue in order (077 has two embed nodes in series). + Returns (compiled_graph, state_cls, providers-to-close).""" + import json + + import httpx + + from openarmature.graph import END, GraphBuilder + from openarmature.retrieval import OpenAIEmbeddingProvider + + from .adapter import build_state_cls + + state_cls = build_state_cls( + "EmbeddingFixtureState", cast("dict[str, dict[str, Any]]", case["state"]["fields"]) + ) + builder = GraphBuilder(state_cls) + providers: list[Any] = [] + mock_responses = list(cast("list[dict[str, Any]]", case.get("mock_embedding") or [])) + + def _handler(_request: httpx.Request) -> httpx.Response: + if not mock_responses: + raise AssertionError("mock_embedding queue exhausted") + spec_resp = mock_responses.pop(0) + body = cast("dict[str, Any]", spec_resp.get("body") or {}) + return httpx.Response( + int(spec_resp.get("status", 200)), + content=json.dumps(body).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + + transport = httpx.MockTransport(_handler) + bound_model = _embedding_model_from_first_response(case) or "text-embedding-test" + + nodes = cast("dict[str, Any]", case["nodes"]) + for node_name, node_spec in nodes.items(): + nd = cast("dict[str, Any]", node_spec) + if "calls_embed" not in nd: + raise AssertionError(f"embedding fixture node {node_name!r} has no calls_embed directive") + provider = OpenAIEmbeddingProvider( + base_url="http://mock-embed.test", + model=bound_model, + api_key="test", + transport=transport, + populate_caller_metadata=populate_caller_metadata, + ) + providers.append(provider) + builder.add_node(node_name, _make_embedding_node_body(nd, provider, case)) + + for edge in cast("list[dict[str, str]]", case["edges"]): + target = END if edge["to"] == "END" else edge["to"] + builder.add_edge(edge["from"], target) + builder.set_entry(cast("str", case["entry"])) + return builder.compile(), state_cls, providers + + +async def _run_embedding_fixture(spec: Mapping[str, Any]) -> None: + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + try: + await _run_embedding_case(case) + except AssertionError as e: + raise AssertionError(f"case {case.get('name')!r}: {e}") from e + + +async def _run_embedding_case(case: Mapping[str, Any]) -> None: + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + + from openarmature.graph import NodeException + from openarmature.observability.langfuse import InMemoryLangfuseClient, LangfuseObserver + + from .harness.llm_attribute_assertions import ( + assert_attribute_does_not_contain, + assert_attribute_parses_as_messages, + assert_attribute_parses_as_object, + assert_attribute_truncation, + assert_attributes_absent, + ) + + expected = cast("dict[str, Any]", case["expected"]) + expected_error = cast("dict[str, Any] | None", case.get("expected_error")) + # The provider's caller-metadata snapshot is opt-in via the collector's + # include_caller_metadata flag (none of 074-083 set it, so the events carry + # caller_invocation_metadata=None, matching the fixtures' null assertion). + collectors, populate_caller_metadata = _parse_typed_observers(case) + graph, state_cls, providers = _build_embedding_graph( + case, populate_caller_metadata=populate_caller_metadata + ) + state = _make_state_instance(case, state_cls) + + exporter: Any = None + otel_observer: Any = None + langfuse_client: Any = None + if "observers" in expected: + for collector in collectors.values(): + graph.attach_observer(collector) + if "span_tree" in expected: + exporter = InMemorySpanExporter() + otel_kwargs: dict[str, Any] = {"span_processor": SimpleSpanProcessor(exporter)} + if "disable_provider_payload" in case: + otel_kwargs["disable_provider_payload"] = bool(case["disable_provider_payload"]) + otel_observer = OTelObserver(**otel_kwargs) + graph.attach_observer(otel_observer) + if "langfuse_trace" in expected: + langfuse_client = InMemoryLangfuseClient() + lf_kwargs: dict[str, Any] = {"client": langfuse_client} + lf_cfg = cast("dict[str, Any] | None", case.get("langfuse_observer")) or {} + if "disable_provider_payload" in lf_cfg: + lf_kwargs["disable_provider_payload"] = bool(lf_cfg["disable_provider_payload"]) + graph.attach_observer(LangfuseObserver(**lf_kwargs)) + + try: + if expected_error is not None: + with pytest.raises(NodeException): + await graph.invoke(state) + else: + await graph.invoke(state) + await graph.drain() + finally: + for provider in providers: + await provider.aclose() + if otel_observer is not None: + otel_observer.shutdown() + + if "observers" in expected: + for obs_name, obs_spec in cast("dict[str, Any]", expected["observers"]).items(): + _assert_observer_expectations(obs_name, collectors[obs_name], cast("dict[str, Any]", obs_spec)) + if "span_tree" in expected and exporter is not None: + _check_payload_span_tree( + exporter.get_finished_spans(), + cast("list[dict[str, Any]]", expected["span_tree"]), + full_input_serialization=None, + assert_attributes_absent=assert_attributes_absent, + assert_attribute_parses_as_messages=assert_attribute_parses_as_messages, + assert_attribute_parses_as_object=assert_attribute_parses_as_object, + assert_attribute_does_not_contain=assert_attribute_does_not_contain, + assert_attribute_truncation=assert_attribute_truncation, + ) + if "langfuse_trace" in expected and langfuse_client is not None: + assert len(langfuse_client.traces) == 1, ( + f"expected 1 Langfuse trace; got {sorted(langfuse_client.traces)}" + ) + trace = next(iter(langfuse_client.traces.values())) + _assert_langfuse_observation_tree( + trace, cast("list[dict[str, Any]]", expected["langfuse_trace"].get("observations") or []) + ) + + # --------------------------------------------------------------------------- # Proposal 0049 — typed LlmCompletionEvent fixtures (050-056) # --------------------------------------------------------------------------- diff --git a/tests/unit/test_retrieval_provider.py b/tests/unit/test_retrieval_provider.py index 3428899..d23020b 100644 --- a/tests/unit/test_retrieval_provider.py +++ b/tests/unit/test_retrieval_provider.py @@ -124,6 +124,7 @@ def _embedding_event() -> EmbeddingEvent: input_strings=["x"], input_count=1, dimensions=2, + output_vectors=[[0.1, 0.2]], request_params={}, request_extras={}, active_prompt=None, @@ -155,7 +156,7 @@ def _embedding_failed_event() -> EmbeddingFailedEvent: ) -async def test_otel_observer_safely_ignores_embedding_events() -> None: +async def test_otel_observer_embedding_no_op_without_invocation_context() -> None: from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter @@ -163,21 +164,24 @@ async def test_otel_observer_safely_ignores_embedding_events() -> None: exporter = InMemorySpanExporter() observer = OTelObserver(span_processor=SimpleSpanProcessor(exporter)) - # Without the skip branch these would fall through to ``event.phase`` - # and AttributeError (the events have no ``phase`` field). The bundled - # embedding span is a follow-up; here they must no-op. + # The observer renders openarmature.embedding.complete spans from these + # events (covered end-to-end by conformance fixture 082). Here they are + # delivered outside any invocation: the observer reads the invocation_id + # from the ContextVar (None here) and returns early, so no span is opened. await observer(_embedding_event()) await observer(_embedding_failed_event()) assert len(exporter.get_finished_spans()) == 0 -async def test_langfuse_observer_safely_ignores_embedding_events() -> None: +async def test_langfuse_observer_embedding_no_op_without_invocation_context() -> None: from openarmature.observability.langfuse import InMemoryLangfuseClient, LangfuseObserver client = InMemoryLangfuseClient() observer = LangfuseObserver(client=client) - # Must not raise (same skip-branch contract as the OTel observer), and no - # trace / observation is created for the skipped events. + # The observer renders dedicated Embedding observations from these events + # (covered end-to-end by conformance fixtures 083 / 137). Delivered outside + # any invocation, the observer returns early on the None invocation_id, so + # no trace / observation is created. await observer(_embedding_event()) await observer(_embedding_failed_event()) assert client.traces == {}