Spaces:
Sleeping
Sleeping
| """Canonical observation and evaluation contracts for RAG telemetry. | |
| Every later observability task (OTel boundary, RAG instrumentation, benchmark | |
| harness, Control Room wiring, SigNoz dashboards, the final write-up) imports | |
| these shapes instead of redefining its own. The goal is the one stated in the | |
| plan preamble: "Every value has one name and meaning across local | |
| observations, spans, metrics, Control Room, SigNoz, and the final report." | |
| This module defines shapes only -> it emits nothing. No span, metric, log, or | |
| file write happens here. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Literal | |
| from pydantic import BaseModel, ConfigDict, Field | |
| # --- Telemetry mode ------------------------------------------------- | |
| # "disabled": no OTel emission at all. | |
| # "local": emit locally (local_store JSONL / Control Room) without exporting | |
| # to a remote collector. | |
| # "full": emit locally AND export via OTLP to SigNoz. | |
| TelemetryMode = Literal["disabled", "local", "full"] | |
| # --- Retrieval outcome ------------------------------------------------- | |
| # Deliberately distinct from a plain empty list: "success_empty" means the | |
| # retrieval pipeline ran correctly and genuinely found nothing, while | |
| # "error_fallback" means the pipeline itself failed and the caller fell back | |
| # to an empty/degraded result. Collapsing these into one signal was the | |
| # whole reason retrieval failures used to be invisible before instrumentation. | |
| RetrievalStatus = Literal["success", "success_empty", "error_fallback"] | |
| class RetrievalOutcome(BaseModel): | |
| """The result of one retrieval attempt, kept distinct from a plain list. | |
| Product code and instrumentation both consume this: product code gets a | |
| real status even in fallback paths, and observability code gets a | |
| ready-made `retrieval_status` / `empty_result` / `retrieval_error` triple | |
| per the canonical "Retrieval-behavior measures" contract. | |
| """ | |
| model_config = ConfigDict(extra="forbid") | |
| status: RetrievalStatus | |
| candidates: list[dict[str, Any]] = Field(default_factory=list) | |
| empty_result: bool = False | |
| retrieval_error: bool = False | |
| error_type: str | None = None | |
| def success(cls, candidates: list[dict[str, Any]]) -> "RetrievalOutcome": | |
| return cls( | |
| status="success" if candidates else "success_empty", | |
| candidates=candidates, | |
| empty_result=not candidates, | |
| ) | |
| def failure(cls, error_type: str) -> "RetrievalOutcome": | |
| return cls(status="error_fallback", retrieval_error=True, error_type=error_type) | |
| # --- Metric-safe attribute filtering ------------------------------------------------- | |
| # Exactly the "Metric-safe dimensions" list from the plan preamble. Never add | |
| # run/query/trace/project/document/operation IDs here -> those stay trace and | |
| # log attributes only, per the global constraint that plain project IDs are | |
| # allowed as trace/log attributes but must never become metric dimensions. | |
| METRIC_ATTRIBUTE_KEYS: frozenset[str] = frozenset( | |
| { | |
| "pipeline.version", | |
| "query.category", | |
| "cache.state", | |
| "consumer", | |
| "operation", | |
| "subsystem", | |
| "status", | |
| "service.name", | |
| "evaluation_run", | |
| # Single bounded operator-controlled boolean. This lets canonical | |
| # product cohorts exclude controlled diagnostic signals without using | |
| # a high-cardinality run or request identifier. | |
| "diagnostic.smoke", | |
| } | |
| ) | |
| def metric_attributes(attributes: dict[str, Any]) -> dict[str, Any]: | |
| """Narrow an arbitrary attribute dict down to metric-safe dimensions. | |
| Keeps only keys in `METRIC_ATTRIBUTE_KEYS` whose value is a plain scalar | |
| (str/int/float/bool). Non-scalar values (lists, dicts, None) are dropped | |
| even for an otherwise-safe key, since a metric dimension must be a single | |
| bounded value, not a structure that can blow up cardinality on its own. | |
| Does not mutate the input. | |
| """ | |
| return { | |
| key: value | |
| for key, value in attributes.items() | |
| if key in METRIC_ATTRIBUTE_KEYS and isinstance(value, (str, int, float, bool)) | |
| } | |
| # --- Experiment identity ------------------------------------------------- | |
| class ExperimentIdentity(BaseModel): | |
| """The canonical experiment-identity fields from the plan preamble. | |
| Ordinary product traffic leaves these unset; labelled evaluation runs | |
| (the benchmark harness in a later task) populate them so every trace, | |
| metric, log, and local record for one run can be joined back together. | |
| Field names are snake_case; the dotted preamble names they mirror are | |
| noted alongside each field. | |
| """ | |
| model_config = ConfigDict(extra="forbid") | |
| experiment_id: str | None = None # experiment.id | |
| run_id: str | None = None # experiment.run_id | |
| pipeline_version: str | None = None # pipeline.version | |
| query_set_version: str | None = None # query_set.version | |
| corpus_version: str | None = None # corpus.version | |
| query_id: str | None = None # query.id | |
| query_category: str | None = None # query.category | |
| repetition: int | None = None # repetition | |
| cache_state: str | None = None # cache.state | |
| telemetry_mode: TelemetryMode | None = None # telemetry.mode | |
| cold_start: bool | None = None | |
| git_commit: str | None = None # git.commit | |
| embedding_model: str | None = None # embedding.model | |
| cerebras_model: str | None = None # cerebras.model | |
| chroma_version: str | None = None # chroma.version | |
| cognee_version: str | None = None # cognee.version | |
| # --- Operation observation ------------------------------------------------- | |
| class OperationObservation(BaseModel): | |
| """Compact, per-operation local observation. | |
| This is the shape the (not-yet-built) single operation helper produces | |
| for Control Room and experiment evidence: "latest stage durations, | |
| counts, statuses, and trace ID" per the preamble's signal-responsibility | |
| table. It carries no prompts, full responses, PDF text, chunks, | |
| annotation contents, filesystem paths, or raw document IDs -> only | |
| durations, counts, statuses, and identity. | |
| `status` is intentionally a plain string rather than a narrow Literal: | |
| it must accommodate both ordinary per-operation statuses (e.g. "success", | |
| "error") and the canonical missing-data sentinels (`not_instrumented`, | |
| `not_applicable`, `not_observed`, `degraded`) -> a numeric duration must | |
| stay `None` under those sentinels, never silently become 0. | |
| """ | |
| model_config = ConfigDict(extra="forbid") | |
| operation: str | |
| status: str | |
| subsystem: str | None = None | |
| consumer: str | None = None | |
| duration_ms: float | None = None | |
| stage_durations_ms: dict[str, float] = Field(default_factory=dict) | |
| counts: dict[str, int | float] = Field(default_factory=dict) | |
| trace_id: str | None = None | |
| experiment: ExperimentIdentity | None = None | |
| retrieval: RetrievalOutcome | None = None | |
| evaluation_run: bool = False | |
| attributes: dict[str, Any] = Field(default_factory=dict) | |