Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions conformance.toml
Original file line number Diff line number Diff line change
Expand Up @@ -610,12 +610,11 @@ note = "Retrieval-provider embedding capability. The EmbeddingProvider protocol
# Spec v0.70.0 (proposal 0060). Retrieval-provider rerank protocol — the
# ``RerankProvider`` surface + ``RerankEvent`` / ``RerankFailedEvent`` typed
# variants + OTel ``openarmature.rerank.complete`` span / Langfuse Retriever
# observation / rerank metrics. Sibling to the embedding surface (0059);
# python has not shipped retrieval-provider, so rerank is not-yet. Crossed by
# the v0.70.1 pin (adopted for fixture 110 / 0075); the 11 rerank observability
# fixtures (099-109) defer with it.
# observation / rerank metrics. Sibling to the embedding surface (0059).
[proposals."0060"]
status = "not-yet"
status = "partial"
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."

# Spec v0.55.0 (proposal 0065; repo pins v0.55.1). Failure-isolation
# cause fidelity at non-node placements (pipeline-utilities §6.3 /
Expand Down
140 changes: 139 additions & 1 deletion src/openarmature/graph/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@
if TYPE_CHECKING:
from openarmature.llm.messages import ToolCall
from openarmature.llm.response import Usage
from openarmature.retrieval.response import EmbeddingUsage
from openarmature.retrieval.response import EmbeddingUsage, RerankUsage, ScoredDocument

# Sentinel empty metadata mapping for events constructed without a
# live caller-metadata snapshot (test helpers, synthetic events).
Expand Down Expand Up @@ -877,6 +877,142 @@ class EmbeddingFailedEvent:
caller_invocation_metadata: Mapping[str, AttributeValue] | None = None


# Spec: realizes graph-engine §6 -- the typed RerankEvent / RerankFailedEvent
# pair (proposal 0060, retrieval-provider rerank capability), the rerank
# sibling to the EmbeddingEvent / EmbeddingFailedEvent pair. Dispatched on the
# observer delivery queue per RerankProvider.rerank() call: the success variant
# after the response is parsed + validated (retrieval-provider §6), the failure
# variant alongside a raised §7 category exception (mutually exclusive per
# call). Scalar fan_out_index / branch_name only, matching the embedding pair
# (the lineage chains arrive uniformly across the provider events with proposal
# 0084). query / documents / request_extras / output_results are payload-
# bearing, populated unconditionally; observer-side privacy gates (OTel
# disable_provider_payload, Langfuse equivalents) apply at rendering, symmetric
# with EmbeddingEvent.
@dataclass(frozen=True)
class RerankEvent:
"""A typed rerank provider call event delivered to observers.

Carries identity, scoping, and outcome data for a successful
``RerankProvider.rerank()`` call. Observer code filters by type
discrimination (``isinstance(event, RerankEvent)``).

The identity / scoping / request-side fields mirror ``EmbeddingEvent``'s
convention; the outcome fields are rerank-specific:

- ``query``: the query string the call was made with; non-nullable,
populated unconditionally (privacy gating is observer-side at
rendering).
- ``documents``: the input documents list; non-nullable, populated
unconditionally. Same privacy posture as ``query``.
- ``document_count``: ``len(documents)``; a convenience field.
- ``top_k``: the caller-supplied ``top_k`` value; ``None`` when the
caller passed ``None``.
- ``result_count``: ``len(output_results)``; a convenience field.
- ``output_results``: the scored results the call returned; populated
unconditionally on the success event (privacy gating is observer-side
at rendering).
- ``response_model`` / ``response_id``: the provider-returned model and
response identifiers; ``None`` when the provider returned none.
- ``usage``: the rerank usage record; ``None`` when the call returned
no usage.
- ``request_params``: the rerank request parameters the caller supplied
(e.g. ``return_documents``). Absence-is-meaningful: only supplied keys
appear; an empty mapping when none.
- ``request_extras``: the runtime-config extras pass-through bag.
- ``active_prompt`` / ``active_prompt_group``: prompt-context snapshots
at rerank-call time; ``None`` outside a binding.
- ``call_id``: a per-call disambiguator, always present, freshly minted
per ``rerank()`` call.
"""

invocation_id: str
correlation_id: str | None
node_name: str
namespace: tuple[str, ...]
attempt_index: int
fan_out_index: int | None
branch_name: str | None
provider: str
model: str
response_id: str | None
response_model: str | None
# RerankUsage is a string-typed forward reference per the TYPE_CHECKING
# import -- keeps the runtime import direction graph -> retrieval off the
# module-load path.
usage: "RerankUsage | None"
latency_ms: float | None
query: str
documents: list[str]
document_count: int
top_k: int | None
result_count: int
# Spec graph-engine §6 (proposal 0089): the output scored results (sourced
# from RerankResponse.results at dispatch). Populated unconditionally on the
# success event -- payload-bearing like query / documents, gated observer-
# side at the rendering boundary. The §8.4.7 rerank output mapping reads
# this field. No output field on RerankFailedEvent (no response).
output_results: list["ScoredDocument"]
request_params: Mapping[str, Any]
request_extras: Mapping[str, Any]
active_prompt: Any
active_prompt_group: Any
call_id: str
caller_invocation_metadata: Mapping[str, AttributeValue] | None = None


# Spec: the failure sibling of RerankEvent (proposal 0060). Dispatched
# whenever RerankProvider.rerank() raises a §7 category exception -- covers
# both the provider-caught path and the pre-send validation raise
# (provider_invalid_request on an empty query / documents list / non-positive
# top_k). Dispatched ALONGSIDE the exception, not in place of it; mutually
# exclusive with RerankEvent on the same call. The response-side fields
# (usage / response_id / response_model / result_count / output_results) are
# absent (no response).
@dataclass(frozen=True)
class RerankFailedEvent:
"""A typed rerank provider call failure event delivered to observers.

Carries identity, scoping, and failure-context data for a ``rerank()``
call that raised a retrieval-provider category exception. Observer code
filters by type discrimination (``isinstance(event, RerankFailedEvent)``).

The identity / scoping / request-side field set mirrors ``RerankEvent``;
the response-side fields are absent. Failure-specific fields:

- ``error_category``: the error category the call raised (one of the
rerank-applicable provider categories). Always present.
- ``error_type``: an optional impl-level / vendor-specific type or code;
``None`` when unavailable.
- ``error_message``: a human-readable message; always present (the empty
string when the exception carried no message).
"""

invocation_id: str
correlation_id: str | None
node_name: str
namespace: tuple[str, ...]
attempt_index: int
fan_out_index: int | None
branch_name: str | None
provider: str
model: str
latency_ms: float | None
query: str
documents: list[str]
document_count: int
top_k: int | None
request_params: Mapping[str, Any]
request_extras: Mapping[str, Any]
active_prompt: Any
active_prompt_group: Any
call_id: str
error_category: str
error_message: str
error_type: str | None = None
caller_invocation_metadata: Mapping[str, AttributeValue] | None = None


# Spec: realizes pipeline-utilities §6.3 failure-isolation middleware
# (proposal 0050). Emitted by FailureIsolationMiddleware when it
# catches an exception escaping the inner chain and substitutes a
Expand Down Expand Up @@ -1045,6 +1181,8 @@ class ToolCallFailedEvent:
"MetadataAugmentationEvent",
"NodeEvent",
"ParallelBranchesEventConfig",
"RerankEvent",
"RerankFailedEvent",
"ToolCallEvent",
"ToolCallFailedEvent",
]
8 changes: 7 additions & 1 deletion src/openarmature/graph/observer.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@
LlmRetryAttemptEvent,
MetadataAugmentationEvent,
NodeEvent,
RerankEvent,
RerankFailedEvent,
ToolCallEvent,
ToolCallFailedEvent,
)
Expand All @@ -67,7 +69,9 @@
# dispatched by FailureIsolationMiddleware when it catches an exception
# escaping the inner chain and substitutes a degraded partial update);
# and EmbeddingEvent / EmbeddingFailedEvent (proposal 0059 typed embedding
# provider call events, dispatched on every EmbeddingProvider.embed()).
# provider call events, dispatched on every EmbeddingProvider.embed());
# and RerankEvent / RerankFailedEvent (proposal 0060 typed rerank provider
# call events, dispatched on every RerankProvider.rerank()).
ObserverEvent = (
NodeEvent
| MetadataAugmentationEvent
Expand All @@ -81,6 +85,8 @@
| ToolCallFailedEvent
| EmbeddingEvent
| EmbeddingFailedEvent
| RerankEvent
| RerankFailedEvent
)


Expand Down
9 changes: 9 additions & 0 deletions src/openarmature/observability/langfuse/observer.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@
LlmRetryAttemptEvent,
MetadataAugmentationEvent,
NodeEvent,
RerankEvent,
RerankFailedEvent,
ToolCallEvent,
ToolCallFailedEvent,
)
Expand Down Expand Up @@ -485,6 +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.
if isinstance(event, RerankEvent | RerankFailedEvent):
return
if event.phase == "started":
self._open_started_observation(event)
elif event.phase == "completed":
Expand Down
9 changes: 9 additions & 0 deletions src/openarmature/observability/otel/observer.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,8 @@
LlmRetryAttemptEvent,
MetadataAugmentationEvent,
NodeEvent,
RerankEvent,
RerankFailedEvent,
ToolCallEvent,
ToolCallFailedEvent,
)
Expand Down Expand Up @@ -786,6 +788,13 @@ 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.
if isinstance(event, RerankEvent | RerankFailedEvent):
return
# Proposal 0063 tool-execution observability: emit the
# openarmature.tool.call span from the typed tool events.
# Independent of disable_llm_spans (that flag is scoped to LLM
Expand Down
29 changes: 25 additions & 4 deletions src/openarmature/retrieval/__init__.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,47 @@
"""The retrieval-provider capability.

The embedding provider protocol, its response types, and the bundled
reference embedding provider. The rerank protocol is a sibling surface on
the same capability.
The embedding + rerank provider protocols, their response types, and the
bundled reference providers (an OpenAI-compatible embedding provider and a
Cohere-shape reranker). Embedding and rerank are sibling surfaces on the same
capability.
"""

from __future__ import annotations

from .provider import (
EmbeddingProvider,
RerankProvider,
validate_embedding_input,
validate_embedding_response,
validate_rerank_input,
validate_rerank_response,
)
from .providers.cohere import CohereRerankProvider
from .providers.openai import OpenAIEmbeddingProvider
from .response import EmbeddingResponse, EmbeddingRuntimeConfig, EmbeddingUsage
from .response import (
EmbeddingResponse,
EmbeddingRuntimeConfig,
EmbeddingUsage,
RerankResponse,
RerankRuntimeConfig,
RerankUsage,
ScoredDocument,
)

__all__ = [
"CohereRerankProvider",
"EmbeddingProvider",
"EmbeddingResponse",
"EmbeddingRuntimeConfig",
"EmbeddingUsage",
"OpenAIEmbeddingProvider",
"RerankProvider",
"RerankResponse",
"RerankRuntimeConfig",
"RerankUsage",
"ScoredDocument",
"validate_embedding_input",
"validate_embedding_response",
"validate_rerank_input",
"validate_rerank_response",
]
Loading