Files
kontextual-engine/src/kontextual_engine/core/retrieval_feedback.py

60 lines
1.9 KiB
Python

"""Retrieval feedback and quality signal primitives."""
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any
from .primitives import compact_dict, new_id, utc_now
class RetrievalFeedbackLabel(str, Enum):
USEFUL = "useful"
IRRELEVANT = "irrelevant"
MISSING = "missing"
UNSAFE = "unsafe"
LOW_CONFIDENCE = "low_confidence"
@dataclass(frozen=True)
class RetrievalFeedbackRecord:
label: RetrievalFeedbackLabel
query: dict[str, Any]
result_ref: dict[str, Any]
actor_id: str
correlation_id: str
notes: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
feedback_id: str = field(default_factory=lambda: new_id("feedback"))
created_at: str = field(default_factory=lambda: utc_now().isoformat())
def to_dict(self) -> dict[str, Any]:
return compact_dict(
{
"feedback_id": self.feedback_id,
"label": self.label.value,
"query": dict(self.query),
"result_ref": dict(self.result_ref),
"actor_id": self.actor_id,
"correlation_id": self.correlation_id,
"notes": self.notes,
"metadata": dict(self.metadata),
"created_at": self.created_at,
}
)
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "RetrievalFeedbackRecord":
return cls(
feedback_id=data["feedback_id"],
label=RetrievalFeedbackLabel(data["label"]),
query=dict(data.get("query", {})),
result_ref=dict(data.get("result_ref", {})),
actor_id=data["actor_id"],
correlation_id=data["correlation_id"],
notes=data.get("notes"),
metadata=dict(data.get("metadata", {})),
created_at=data["created_at"],
)