diff --git a/conformance.toml b/conformance.toml index 226cb29..4758bb1 100644 --- a/conformance.toml +++ b/conformance.toml @@ -612,9 +612,9 @@ note = "Retrieval-provider embedding capability. The EmbeddingProvider protocol # variants + OTel ``openarmature.rerank.complete`` span / Langfuse Retriever # observation / rerank metrics. Sibling to the embedding surface (0059). [proposals."0060"] -status = "partial" +status = "implemented" since = "0.16.0" -note = "Retrieval-provider rerank protocol. The RerankProvider protocol (retrieval-provider §5) + the RerankResponse / ScoredDocument / RerankUsage / RerankRuntimeConfig response types (§6) + the RerankRuntimeConfig runtime config (§2) ship, alongside the CohereRerankProvider reference reranker against POST /v2/rerank (§8.4, Cohere-shape): the request body is {model, query, documents (string array), top_n from top_k}, return_documents is a silent no-op (the Cohere wire has no such field), and max_tokens_per_doc rides the extras pass-through bag. The provider parses {id, model?, results: [{index, relevance_score, document?}], meta.billed_units.{search_units, input_tokens}}, sorts results by relevance_score descending, reads the document echo only when present (never auto-filled), builds a RerankUsage record only when the provider surfaces >= 1 usage figure (else usage = null, never fabricated), and enforces the §6 invariants -- valid index into the input documents, no duplicate index, len(results) <= top_k when supplied -- raising provider_invalid_response otherwise. Pre-send validation raises provider_invalid_request on an empty query, an empty documents list, or top_k <= 0 (top_k MAY exceed len(documents)). The provider dispatches the typed RerankEvent / RerankFailedEvent (graph-engine §6) per rerank() call, mutually exclusive, alongside the §7-category exception on failure; RerankEvent.output_results is populated unconditionally on success (proposal 0089). The §7 categories reuse the shared llm-provider taxonomy. Protocol fixtures 006-012 pass. partial because the observability rendering (OTel §5.5.13 openarmature.rerank.complete span + Langfuse §8.4.7 Retriever observation + the rerank-metrics surface; fixtures 099-109 / 138 / 141 / 142) lands in the 0060b follow-on -- the bundled OTel + Langfuse observers safely skip the rerank events for now. Wire-format details (0090 Cohere rerank fixtures 028-031) are a separate proposal." +note = "Retrieval-provider rerank protocol. The RerankProvider protocol (retrieval-provider §5) + the RerankResponse / ScoredDocument / RerankUsage / RerankRuntimeConfig response types (§6) + the RerankRuntimeConfig runtime config (§2) ship, alongside the CohereRerankProvider reference reranker against POST /v2/rerank (§8.4, Cohere-shape): the request body is {model, query, documents (string array), top_n from top_k}, return_documents is a silent no-op (the Cohere wire has no such field), and max_tokens_per_doc rides the extras pass-through bag. The provider parses {id, model?, results: [{index, relevance_score, document?}], meta.billed_units.{search_units, input_tokens}}, sorts results by relevance_score descending, reads the document echo only when present (never auto-filled), builds a RerankUsage record only when the provider surfaces >= 1 usage figure (else usage = null, never fabricated), and enforces the §6 invariants -- valid index into the input documents, no duplicate index, len(results) <= top_k when supplied -- raising provider_invalid_response otherwise. Pre-send validation raises provider_invalid_request on an empty query, an empty documents list, or top_k <= 0 (top_k MAY exceed len(documents)). The provider dispatches the typed RerankEvent / RerankFailedEvent (graph-engine §6) per rerank() call, mutually exclusive, alongside the §7-category exception on failure; RerankEvent.output_results is populated unconditionally on success (proposal 0089). The §7 categories reuse the shared llm-provider taxonomy. Both bundled observers render the rerank events (0060b): the OTelObserver emits an openarmature.rerank.complete span (observability §5.5.13, SpanKind.CLIENT) parented lineage-aware under the calling node, carrying the GenAI semconv subset (gen_ai.system / request.model / response.model / response.id + the conditional usage.input_tokens, gated by disable_genai_semconv) + the OA openarmature.rerank.query_length (UTF-8 bytes) / .document_count / .top_k (omitted when null) / .result_count identity attrs + the conditional .search_units + the payload-gated (disable_provider_payload) .query / .documents / .results (the results sourced from RerankEvent.output_results per proposal 0089) -- no gen_ai.operation.name (deferred, no upstream rerank coverage), not gated by disable_llm_spans; failure renders ERROR status + error.type + an exception event + openarmature.error.category. The LangfuseObserver renders a dedicated Retriever observation (asType retriever, §8.4.7) with model = response_model-or-model, usageDetails.{input, searchUnits} (omitted when usage = null), the openarmature_query_length / _document_count / _top_k / _result_count / _response_id metadata, and the payload-gated input {query, documents} + output results; failure renders an ERROR-level Retriever observation with error_category as statusMessage, error_type / error_message in metadata, and no output. Protocol fixtures 006-012 + observability fixtures 099-108 / 138 / 141 / 142 pass. The §11 rerank-metrics path (fixture 109) is proposal 0067's metrics surface (one GenAI instrument across LLM + embedding + rerank calls), tracked and deferred under 0067 (cf. 089), not part of 0060's own observability surface. Wire-format details (0090 Cohere rerank fixtures 028-031) are a separate proposal." # Spec v0.55.0 (proposal 0065; repo pins v0.55.1). Failure-isolation # cause fidelity at non-node placements (pipeline-utilities §6.3 / @@ -856,9 +856,9 @@ status = "not-yet" # (graph-engine §6 EmbeddingEvent.output_vectors / RerankEvent. # output_results + observability output mappings). [proposals."0089"] -status = "partial" +status = "implemented" since = "0.16.0" -note = "The embedding half shipped in 0059b (#198): EmbeddingEvent.output_vectors is populated by OpenAIEmbeddingProvider and rendered as the Langfuse Embedding observation output (observability §8.4.5), sourced event-side per 0089; observability fixtures 083 (re-sourced) + 137 pass. partial because the rerank half is unshipped -- RerankEvent.output_results and the rerank observation output rendering ride the rerank capability (proposal 0060, not-yet); observability fixtures 108 + 138 defer with it." +note = "The embedding half shipped in 0059b (#198): EmbeddingEvent.output_vectors is populated by OpenAIEmbeddingProvider and rendered as the Langfuse Embedding observation output (observability §8.4.5), sourced event-side per 0089; observability fixtures 083 (re-sourced) + 137 pass. The rerank half now ships with the 0060b rerank observability: RerankEvent.output_results is populated unconditionally on success by CohereRerankProvider and rendered as both the OTel openarmature.rerank.results attribute (observability §5.5.13, payload-gated) and the Langfuse Retriever observation output (§8.4.7), sourced event-side per 0089; observability fixtures 108 + 138 pass. This completes 0089 across the embedding + rerank surfaces." # Spec v0.85.0 (proposal 0090). Cohere rerank wire mapping # (retrieval-provider §8.4, POST /v2/rerank; backs 0060's Cohere-shaped diff --git a/src/openarmature/observability/langfuse/adapter.py b/src/openarmature/observability/langfuse/adapter.py index 9861bf0..b100740 100644 --- a/src/openarmature/observability/langfuse/adapter.py +++ b/src/openarmature/observability/langfuse/adapter.py @@ -519,6 +519,65 @@ def embedding( ) return _SpanHandle(obs) + def retriever( + 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=); a Retriever + # observation routes through as_type="retriever" (proposal 0060), the + # SDK's native LangfuseRetriever span. It carries model + usage like an + # Embedding but is a distinct type; usage maps to v4's usage_details. + # The rerank usageDetails convention (§8.4.7) exposes input tokens under + # "input" and search-units under "searchUnits". Back-date via the + # private OTel tracer when start_time is supplied, exactly as + # generation()/tool()/embedding() 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.search_units is not None: + usage_details["searchUnits"] = usage.search_units + 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 LangfuseRetriever + + obs = self._start_back_dated_observation( + LangfuseRetriever, + 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="retriever", + 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 d05c43d..c041262 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", "embedding"] +ObservationType = Literal["span", "generation", "event", "tool", "embedding", "retriever"] # Langfuse-supported `level` values per spec §8.4.2 (statusMessage pair). ObservationLevel = Literal["DEFAULT", "DEBUG", "INFO", "WARNING", "ERROR"] @@ -46,6 +46,12 @@ class LangfuseUsage: input: int | None = None output: int | None = None total: int | None = None + # Rerank observability (proposal 0060, observability §8.4.7): Langfuse's + # usageDetails is an open-shape map, and the OA convention adds a + # `searchUnits` figure for the Retriever observation. Modeled here as a + # snake_case field; the SDK adapter maps it to the camelCase `searchUnits` + # usage_details key. None when the provider reported no search-units count. + search_units: int | None = None @dataclass @@ -297,6 +303,35 @@ def embedding( the caller ``.end()``s at outcome time.""" ... + # Proposal 0060 rerank observability: a dedicated Retriever observation + # (Langfuse asType="retriever", spec observability §8.4.7), NOT a + # Generation with a discriminator. Langfuse positions Retriever for data + # retrieval steps (broad enough to cover reranking); the `usage` record's + # `search_units` / `input` figures render into the open-shape usageDetails + # map (the OA `searchUnits` key convention is defined in §8.4.7). + def retriever( + 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 Retriever observation. Like :meth:`embedding` + for the model / usage surface but a distinct observation type: + carries ``model`` / ``usage`` (input-token + search-unit figures) / + ``input`` (the query + documents) / ``output`` (the scored results) / + 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. @@ -591,6 +626,39 @@ def embedding( trace.observations.append(observation) return _InMemorySpanHandle(observation=observation) + def retriever( + 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="retriever", + 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 cbfd063..ec34451 100644 --- a/src/openarmature/observability/langfuse/observer.py +++ b/src/openarmature/observability/langfuse/observer.py @@ -487,12 +487,13 @@ async def __call__( if isinstance(event, EmbeddingEvent | EmbeddingFailedEvent): self._handle_embedding(event) return - # Proposal 0060 rerank observability (observability §8.4.7): the - # dedicated Langfuse Retriever observation renders from the typed - # rerank events. Not implemented yet (the rerank rendering lands in the - # 0060b follow-on); the observer safely skips the events for now rather - # than falling through to the NodeEvent phase dispatch. + # Proposal 0060 rerank observability (observability §8.4.7): render the + # dedicated Langfuse Retriever observation (asType "retriever") under + # the calling node's Span observation. NOT gated by disable_llm_spans + # (scoped to LLM completion per §5.5.13); the input / output payload is + # gated by disable_provider_payload inside the handler. if isinstance(event, RerankEvent | RerankFailedEvent): + self._handle_rerank(event) return if event.phase == "started": self._open_started_observation(event) @@ -1995,6 +1996,127 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non ) handle.end(end_time=end_time) + # Spec: observability §8.4.7 (proposal 0060) Retriever observation + # (Langfuse asType="retriever"). The failure path is the generic §4.2 / + # §8.4.2 ERROR mapping with the §7 error category, mirroring the tool / + # embedding failure. The output results are sourced from + # RerankEvent.output_results per proposal 0089; the searchUnits usageDetails + # key follows the §8.4.7 OA convention. + def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: + """Open + close a dedicated Retriever observation (Langfuse + ``asType="retriever"``) under the calling node's Span observation. + + Success (``RerankEvent``): DEFAULT level, ``model`` = ``response_model`` + (falling back to the requested ``model``), ``usage`` carrying the input- + token + search-unit figures, the OA identity metadata + (``openarmature_query_length`` / ``openarmature_document_count`` / + ``openarmature_top_k`` / ``openarmature_result_count`` / + ``openarmature_response_id``), and the payload-gated ``input`` + (``{query, documents}``) + ``output`` (scored results). + + Failure (``RerankFailedEvent``): ERROR level with the ``error_category`` + as the status message and ``error_type`` / ``error_message`` in + metadata, mirroring the tool / embedding failure. The request-side + ``input`` is 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.7 request-side identity metadata (present on both variants): + # query_length (UTF-8 byte length), document_count, top_k (when + # supplied). 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) + metadata["openarmature_query_length"] = len(event.query.encode("utf-8")) + metadata["openarmature_document_count"] = event.document_count + if event.top_k is not None: + metadata["openarmature_top_k"] = event.top_k + input_value: Any = None + if not self.disable_provider_payload: + input_value = self._maybe_truncate_for_input( + {"query": event.query, "documents": list(event.documents)} + ) + if isinstance(event, RerankEvent): + # §8.4.7 response-derived metadata (success-only). + metadata["openarmature_result_count"] = event.result_count + 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_results: + output_value = self._maybe_truncate_for_input( + [result.model_dump(exclude_none=True) for result in event.output_results] + ) + # usageDetails omitted entirely when no usage record; when present, + # only the non-null figures render (input / searchUnits) per §8.4.7. + usage = None + if event.usage is not None: + usage = LangfuseUsage( + input=event.usage.input_tokens, + search_units=event.usage.search_units, + ) + target_trace_id = self._trace_id_for(inv_state, event.namespace, event.fan_out_index) + handle = self.client.retriever( + trace_id=target_trace_id, + name="openarmature.rerank.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: the request-side metadata survives; the response-derived + # rows do not. No output. ERROR level + category-as-statusMessage. + 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.retriever( + trace_id=target_trace_id, + name="openarmature.rerank.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, @@ -2126,11 +2248,13 @@ def _open_trace_for_typed_event( | ToolCallEvent | ToolCallFailedEvent | EmbeddingEvent - | EmbeddingFailedEvent, + | EmbeddingFailedEvent + | RerankEvent + | RerankFailedEvent, ) -> None: """Trace open path for a typed event (LLM completion / failure, - tool execution, or embedding) arriving before any node-started event - reached + tool execution, embedding, or rerank) 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 fb148a4..23e022f 100644 --- a/src/openarmature/observability/otel/observer.py +++ b/src/openarmature/observability/otel/observer.py @@ -603,6 +603,8 @@ class OTelObserver: | ToolCallFailedEvent | EmbeddingEvent | EmbeddingFailedEvent + | RerankEvent + | RerankFailedEvent | None, ], None, @@ -714,6 +716,8 @@ def _run_enrichers( | ToolCallFailedEvent | EmbeddingEvent | EmbeddingFailedEvent + | RerankEvent + | RerankFailedEvent | None, ) -> None: """Invoke configured enrichers against ``span`` before @@ -788,12 +792,14 @@ async def __call__( if isinstance(event, EmbeddingEvent | EmbeddingFailedEvent): self._handle_embedding(event) return - # Proposal 0060 rerank observability (observability §5.5.13): the - # openarmature.rerank.complete span + rerank metrics render from the - # typed rerank events. Not implemented yet (the rerank rendering lands - # in the 0060b follow-on); the observer safely skips the events for now - # rather than falling through to the NodeEvent phase dispatch. + # Proposal 0060 rerank observability (observability §5.5.13): emit the + # openarmature.rerank.complete span from the typed rerank events. NOT + # gated by disable_llm_spans (that flag is scoped to LLM completion + # spans per §5.5.13); only the payload / GenAI-semconv attribute subsets + # are gated, inside the handler. Rerank metrics ride proposal 0067 + # (§11), not wired here. if isinstance(event, RerankEvent | RerankFailedEvent): + self._handle_rerank(event) return # Proposal 0063 tool-execution observability: emit the # openarmature.tool.call span from the typed tool events. @@ -1812,6 +1818,130 @@ def _handle_embedding(self, event: EmbeddingEvent | EmbeddingFailedEvent) -> Non self._run_enrichers(span, event) span.end(end_time=end_time_ns) + # Spec: observability §5.5.13 (proposal 0060) rerank span. Parallels the + # §5.5.8 embedding span. Payload gating is §5.5.4 + the §5.5.5 truncation + # contract; gen_ai.operation.name is deferred (no upstream rerank coverage). + # The output results (openarmature.rerank.results) are sourced from + # RerankEvent.output_results per proposal 0089. + def _handle_rerank(self, event: RerankEvent | RerankFailedEvent) -> None: + """Emit an ``openarmature.rerank.complete`` span for a + ``RerankProvider.rerank()`` call, parented under the calling node. + + A ``RerankEvent`` renders OK with the GenAI response subset + the OA + rerank identity attributes; a ``RerankFailedEvent`` renders ERROR with + the standard OTel ``error.type`` attribute, an exception event carrying + the message, and the ``openarmature.error.category`` attribute. The + ``query`` / ``documents`` / ``results`` payload is 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 embedding / 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 rerank identity attributes (ungated). query_length / + # document_count / top_k are request-side (present on both variants); + # result_count is response-derived (success-only). query_length is the + # UTF-8 byte length of the query, not a token count. + attrs["openarmature.rerank.query_length"] = len(event.query.encode("utf-8")) + attrs["openarmature.rerank.document_count"] = event.document_count + if event.top_k is not None: + attrs["openarmature.rerank.top_k"] = event.top_k + if isinstance(event, RerankEvent): + attrs["openarmature.rerank.result_count"] = event.result_count + # 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. input_tokens is conditional: emitted only when a usage + # record is present AND its input_tokens is non-null. + if not self.disable_genai_semconv: + attrs["gen_ai.system"] = event.provider + attrs["gen_ai.request.model"] = event.model + if isinstance(event, RerankEvent): + 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 and event.usage.input_tokens is not None: + attrs["gen_ai.usage.input_tokens"] = event.usage.input_tokens + # openarmature.rerank.search_units is conditional (success-only, and + # only when a usage record reports a non-null search_units figure). + if ( + isinstance(event, RerankEvent) + and event.usage is not None + and event.usage.search_units is not None + ): + attrs["openarmature.rerank.search_units"] = event.usage.search_units + # Payload-gated (disable_provider_payload): the query string, the input + # documents (JSON list), and the scored results (JSON list of records). + if not self.disable_provider_payload: + attrs["openarmature.rerank.query"] = _truncate_for_attribute(event.query, self.payload_max_bytes) + if event.documents: + serialized_docs = _serialize_for_attribute(event.documents) + attrs["openarmature.rerank.documents"] = _truncate_for_attribute( + serialized_docs, self.payload_max_bytes + ) + if isinstance(event, RerankEvent) and event.output_results: + serialized_results = _serialize_for_attribute( + [result.model_dump(exclude_none=True) for result in event.output_results] + ) + attrs["openarmature.rerank.results"] = _truncate_for_attribute( + serialized_results, self.payload_max_bytes + ) + span = self._tracer.start_span( + name="openarmature.rerank.complete", + context=cast("Any", parent_ctx), + kind=SpanKind.CLIENT, + attributes=attrs, + start_time=start_time_ns, + ) + if isinstance(event, RerankFailedEvent): + # 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/tests/conformance/test_observability.py b/tests/conformance/test_observability.py index ca14172..1cc7a6d 100644 --- a/tests/conformance/test_observability.py +++ b/tests/conformance/test_observability.py @@ -233,6 +233,27 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "082-otel-embedding-span-attributes", "083-langfuse-embedding-observation", "137-langfuse-embedding-failure-observation", + # v0.16.0 — proposal 0060 rerank observability (0060b). A calls_rerank + # node awaits CohereRerankProvider.rerank() inside the node body; the + # typed RerankEvent / RerankFailedEvent drive the typed-event collector + # (099-106), the OTel rerank span (107 / 141, + # openarmature.rerank.complete), and the Langfuse Retriever observation + # (108 success, 138 failure, 142 no-usage -- the proposal 0089 + # output_results-sourced rendering + the proposal 0093 nullable-usage + # path). 109 (rerank metrics) stays deferred with the §11 path. + "099-rerank-event-dispatch", + "100-rerank-failure-event-dispatch-on-provider-unavailable", + "101-rerank-event-mutual-exclusion", + "102-rerank-event-call-id-distinct", + "103-rerank-event-query-and-documents-populated", + "104-rerank-event-request-params-populated", + "105-rerank-event-top-k-and-result-count-populated", + "106-rerank-event-active-prompt-populated", + "107-otel-rerank-span-attributes", + "108-langfuse-rerank-observation", + "138-langfuse-rerank-failure-observation", + "141-otel-rerank-no-usage-attributes-omitted", + "142-langfuse-rerank-no-usage-usagedetails-omitted", # 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). @@ -248,9 +269,9 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: ) _RERANK_DEFER = ( - "rerank observability rendering (OTel span / Langfuse Retriever observation / rerank metrics) " - "lands in 0060b; the RerankProvider + typed RerankEvent / RerankFailedEvent ship in 0060a, but " - "the bundled observers safe-skip the rerank events until the rendering is wired" + "rerank observability (099-108 + 138 + 141/142) is wired (proposal 0060b); only the " + "§11 rerank-metrics path stays deferred -- the OTelObserver records metrics from the " + "LLM per-attempt event, not yet from RerankEvent (rides proposal 0067, cf. 089)" ) @@ -275,28 +296,12 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: # 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 RerankProvider + typed - # rerank events ship in 0060a; only the observer rendering defers to 0060b. - **{ - fixture_id: _RERANK_DEFER - for fixture_id in ( - "099-rerank-event-dispatch", - "100-rerank-failure-event-dispatch-on-provider-unavailable", - "101-rerank-event-mutual-exclusion", - "102-rerank-event-call-id-distinct", - "103-rerank-event-query-and-documents-populated", - "104-rerank-event-request-params-populated", - "105-rerank-event-top-k-and-result-count-populated", - "106-rerank-event-active-prompt-populated", - "107-otel-rerank-span-attributes", - "108-langfuse-rerank-observation", - "109-rerank-metrics-token-and-duration", - # proposal 0089 (v0.84.0) rerank failure observation -- the - # RerankFailedEvent ERROR-level Langfuse rendering; lands with the - # 0060b rerank observability. - "138-langfuse-rerank-failure-observation", - ) - }, + # Rerank observability (proposal 0060). 099-108 + 138 + 141/142 are wired + # (proposal 0060b -- see _SUPPORTED_FIXTURES). Only the §11 rerank-metrics + # path (109) stays deferred with the embedding-metrics sibling (089): the + # OTelObserver records metrics from the LLM per-attempt event only; + # extending the instruments to the rerank path rides proposal 0067. + "109-rerank-metrics-token-and-duration": _RERANK_DEFER, # ---- v0.16.0 spec-pin bump (v0.70.1 -> v0.84.0): new fixtures for # proposals deferred to their own later PRs of this release. ---- # Proposal 0062 (LLM completion streaming, spec v0.71.0). The stream @@ -401,18 +406,9 @@ def _reset_otel_global_tracer_provider(restore_to: object) -> None: "140-langfuse-embedding-no-usage-usagedetails-omitted", ) }, - # 141/142 -- rerank no-usage; the rendering lands with the 0060b rerank - # observability (the rerank capability ships in 0060a). - **{ - fixture_id: ( - "rerank no-usage rendering (proposal 0093); lands with the 0060b " - "rerank observability (the rerank capability ships in 0060a)" - ) - for fixture_id in ( - "141-otel-rerank-no-usage-attributes-omitted", - "142-langfuse-rerank-no-usage-usagedetails-omitted", - ) - }, + # 141/142 -- rerank no-usage (OTel span + Langfuse observation) are wired + # with the 0060b rerank observability (see _SUPPORTED_FIXTURES); the + # conditional-usage rendering handles the record-null branch. # 143 -- embedding no-usage metric; rides the §11 embedding-metrics path # (proposal 0067), which is itself deferred (cf. 089). "143-embedding-metrics-no-usage-no-token-observation": ( @@ -723,6 +719,22 @@ async def test_observability_fixture(fixture_path: Path) -> None: "137-langfuse-embedding-failure-observation", }: await _run_embedding_fixture(spec) + elif fixture_id in { + "099-rerank-event-dispatch", + "100-rerank-failure-event-dispatch-on-provider-unavailable", + "101-rerank-event-mutual-exclusion", + "102-rerank-event-call-id-distinct", + "103-rerank-event-query-and-documents-populated", + "104-rerank-event-request-params-populated", + "105-rerank-event-top-k-and-result-count-populated", + "106-rerank-event-active-prompt-populated", + "107-otel-rerank-span-attributes", + "108-langfuse-rerank-observation", + "138-langfuse-rerank-failure-observation", + "141-otel-rerank-no-usage-attributes-omitted", + "142-langfuse-rerank-no-usage-usagedetails-omitted", + }: + await _run_rerank_fixture(spec) elif fixture_id in { "043-get-invocation-metadata-roundtrip", "044-get-invocation-metadata-fan-out-scoping", @@ -4029,6 +4041,12 @@ async def _update_body(_s: Any, _payload: dict[str, Any] = update_block) -> dict return builder.compile(), state_cls, providers +# Maps a usageDetails key (the SDK's camelCase wire form) to the matching +# LangfuseUsage dataclass attribute. Only keys whose casing diverges appear +# here; input / output / total already match their attribute names. +_USAGE_DETAIL_ATTR: dict[str, str] = {"searchUnits": "search_units"} + + def _assert_langfuse_observation_tree( trace: Any, expected: list[dict[str, Any]], parent_id: str | None = None ) -> None: @@ -4062,10 +4080,30 @@ def _assert_langfuse_observation_tree( 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}" + expected_usage = cast("dict[str, Any]", exp["usageDetails"]) + if expected_usage: + assert match.usage is not None, f"{exp_name!r}: expected usageDetails but usage is None" + for uk, uv in expected_usage.items(): + # usageDetails keys are the SDK's camelCase form; map to the + # LangfuseUsage snake_case attribute (searchUnits -> + # search_units). Keys already matching an attribute name + # (input / output / total) pass through unchanged. + attr = _USAGE_DETAIL_ATTR.get(uk, uk) + actual_usage = getattr(match.usage, attr, None) + assert actual_usage == uv, f"{exp_name!r}: usageDetails.{uk} {actual_usage!r} != {uv!r}" + else: + # An empty usageDetails ({}) asserts no usage figures were + # reported: the observation carries no usage record, or one whose + # figures are all None (the open-map "empty" convention). + assert match.usage is None or all( + v is None + for v in ( + match.usage.input, + match.usage.output, + match.usage.total, + match.usage.search_units, + ) + ), f"{exp_name!r}: expected empty usageDetails but usage carries figures: {match.usage!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: @@ -4391,6 +4429,220 @@ async def _run_embedding_case(case: Mapping[str, Any]) -> None: ) +# A calls_rerank node constructs a CohereRerankProvider on an httpx.MockTransport +# built from the fixture's mock_rerank block and awaits provider.rerank(query, +# documents) INSIDE the node body, so the calling-node ContextVars are set and +# the RerankEvent / RerankFailedEvent flow through the observer queue. One graph +# builder serves all three assertion shapes: typed-event-collector (099-106), +# OTel span_tree (107 / 141), and the Langfuse Retriever observation tree +# (108 / 138 / 142). Dispatch is by which expected.* key appears, mirroring the +# embedding path (_run_embedding_case). + + +def _rerank_model_from_first_response(case: Mapping[str, Any]) -> str | None: + """Return the ``model`` on the first ``mock_rerank`` response body, so the + provider binds to the model the fixture's expected event reports. Failure + fixtures return an error body with no model, so this falls back to None.""" + responses = cast("list[dict[str, Any]] | None", case.get("mock_rerank")) 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_rerank_runtime_config(config_spec: Mapping[str, Any] | None) -> Any: + """Build a RerankRuntimeConfig from a ``calls_rerank.config`` block + (fixture 104: ``return_documents``), or None when absent.""" + if not config_spec: + return None + from openarmature.retrieval import RerankRuntimeConfig + + return RerankRuntimeConfig(**dict(config_spec)) + + +def _make_rerank_node_body(node_spec: Mapping[str, Any], provider: Any, case: Mapping[str, Any]) -> Any: + calls_rerank = cast("dict[str, Any]", node_spec["calls_rerank"]) + stores_in = cast("str | None", calls_rerank.get("stores_response_in")) + config = _build_rerank_runtime_config(cast("dict[str, Any] | None", calls_rerank.get("config"))) + top_k = cast("int | None", calls_rerank.get("top_k")) + documents = list(cast("list[str]", calls_rerank.get("documents") or [])) + # A node may render a query prompt first (106, RAG retrieval): the rendered + # PromptResult is stamped active for the rerank() call and supplies the + # query when query_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 + query_from_rendered = bool(calls_rerank.get("query_from_rendered_prompt", False)) + explicit_query = cast("str | None", calls_rerank.get("query")) + + async def body(_state: Any) -> Mapping[str, Any]: + if query_from_rendered and prompt_result is not None: + query = cast("str", prompt_result.messages[0].content) + else: + query = explicit_query or "" + if prompt_result is not None: + from openarmature.prompts import with_active_prompt + + with with_active_prompt(prompt_result): + response = await provider.rerank(query, documents, top_k=top_k, config=config) + else: + response = await provider.rerank(query, documents, top_k=top_k, config=config) + return {stores_in: response.model_dump()} if stores_in else {} + + return body + + +def _build_rerank_graph( + case: Mapping[str, Any], + *, + populate_caller_metadata: bool, +) -> tuple[Any, type[Any], list[Any]]: + """Build a graph whose nodes are calls_rerank. Every rerank node gets its + own CohereRerankProvider sharing one MockTransport that replays the case's + ``mock_rerank`` queue in order (102 has two rerank 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 CohereRerankProvider + + from .adapter import build_state_cls + + state_cls = build_state_cls( + "RerankFixtureState", 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_rerank") or [])) + + def _handler(_request: httpx.Request) -> httpx.Response: + if not mock_responses: + raise AssertionError("mock_rerank 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 = _rerank_model_from_first_response(case) or "rerank-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_rerank" not in nd: + raise AssertionError(f"rerank fixture node {node_name!r} has no calls_rerank directive") + provider = CohereRerankProvider( + base_url="http://mock-rerank.test", + model=bound_model, + api_key="test", + transport=transport, + populate_caller_metadata=populate_caller_metadata, + ) + providers.append(provider) + builder.add_node(node_name, _make_rerank_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_rerank_fixture(spec: Mapping[str, Any]) -> None: + cases = cast("list[dict[str, Any]]", spec["cases"]) + for case in cases: + try: + await _run_rerank_case(case) + except AssertionError as e: + raise AssertionError(f"case {case.get('name')!r}: {e}") from e + + +async def _run_rerank_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")) + collectors, populate_caller_metadata = _parse_typed_observers(case) + graph, state_cls, providers = _build_rerank_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) # ---------------------------------------------------------------------------