Coverage for src / lexigram / contracts / ai / feedback.py: 95%
43 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-19 05:41 +0800
1"""AI feedback contracts and types."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from datetime import UTC, datetime
7from enum import StrEnum
8from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
9from uuid import uuid4
11if TYPE_CHECKING:
12 from lexigram.contracts.core.result import Result
15class FeedbackType(StrEnum):
16 """Type of feedback collected."""
18 RATING = "rating"
19 TEXT = "text"
20 CORRECTION = "correction"
21 LABEL = "label"
24@dataclass(frozen=True)
25class FeedbackItem:
26 """A single feedback item.
28 Attributes:
29 feedback_type: Type of feedback (rating, text, correction, or label).
30 value: The feedback value (e.g. rating score, text comment).
31 owner_id: Owner scope for the item (user, tenant, or composite).
32 context: Context about what was being evaluated (session_id, model, etc.).
33 metadata: Additional metadata dictionary.
34 id: Unique feedback identifier.
35 created_at: Timestamp when feedback was created.
36 """
38 feedback_type: FeedbackType
39 value: Any
40 owner_id: str
41 context: dict[str, Any] = field(default_factory=dict)
42 metadata: dict[str, Any] = field(default_factory=dict)
43 id: str = field(default_factory=lambda: str(uuid4()))
44 created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
46 @property
47 def type(self) -> FeedbackType:
48 """Alias for feedback_type for backward compatibility.
50 Returns:
51 The feedback type.
52 """
53 return self.feedback_type
55 def to_dict(self) -> dict[str, Any]:
56 """Convert to dictionary.
58 Returns:
59 Dictionary representation with ISO-formatted timestamp.
60 """
61 return {
62 "id": self.id,
63 "type": self.feedback_type.value,
64 "value": self.value,
65 "owner_id": self.owner_id,
66 "context": self.context,
67 "metadata": self.metadata,
68 "created_at": self.created_at.isoformat(),
69 }
71 def __repr__(self) -> str:
72 """String representation."""
73 return (
74 f"FeedbackItem(id={self.id}, "
75 f"type={self.feedback_type.value}, value={self.value})"
76 )
79@dataclass(frozen=True)
80class FeedbackSummary:
81 """Aggregated statistics for collected feedback items.
83 Attributes:
84 total_count: Total number of feedback items in the window.
85 average_rating: Mean rating across all RATING-type items,
86 or None if no ratings were collected.
87 count_by_type: Item count keyed by FeedbackType enum value
88 (e.g. "rating", "text", "correction", "label").
89 """
91 total_count: int = 0
92 average_rating: float | None = None
93 count_by_type: dict[str, int] = field(default_factory=dict)
96@runtime_checkable
97class FeedbackStoreProtocol(Protocol):
98 """Persist and query collected feedback items.
100 All implementations must be async and treat save() as safe to
101 call from request-handling hot-paths (synchronous returns acceptable
102 for cache misses).
103 """
105 async def save(self, feedback: FeedbackItem) -> Result[str, Exception]:
106 """Persist a single feedback item.
108 Args:
109 feedback: The feedback item to store.
111 Returns:
112 Ok(feedback.id) on success, Err(exception) on failure.
113 """
114 ...
116 async def find_by_session(
117 self, session_id: str, *, owner_id: str
118 ) -> list[FeedbackItem]:
119 """Retrieve all feedback items for a session, scoped to an owner.
121 Args:
122 session_id: Session identifier from feedback context.
123 owner_id: Owner scope; only this owner's items are returned.
125 Returns:
126 All collected items in that session, newest first.
127 """
128 ...
130 async def find_by_type(
131 self,
132 feedback_type: FeedbackType,
133 *,
134 owner_id: str,
135 limit: int = 100,
136 ) -> list[FeedbackItem]:
137 """Retrieve feedback items of a given type, scoped to an owner.
139 Args:
140 feedback_type: The type to filter by.
141 owner_id: Owner scope; only this owner's items are returned.
142 limit: Maximum number of results (default 100).
144 Returns:
145 Matching items ordered by creation time descending.
146 """
147 ...
149 async def aggregate(
150 self, *, owner_id: str, window_hours: int = 24
151 ) -> FeedbackSummary:
152 """Compute summary statistics for an owner's items in a time window.
154 Args:
155 owner_id: Owner scope; only this owner's items are aggregated.
156 window_hours: Look-back window in hours (default 24).
158 Returns:
159 Aggregated FeedbackSummary statistics.
160 """
161 ...
164@runtime_checkable
165class FeedbackProtocol(Protocol):
166 """Protocol for submitting and querying AI feedback."""
168 async def submit_feedback(
169 self,
170 trace_id: str,
171 score: float,
172 *,
173 owner_id: str,
174 comment: str | None = None,
175 metadata: dict[str, Any] | None = None,
176 ) -> None:
177 """Submit feedback for an AI generation, scoped to an owner.
179 Args:
180 trace_id: Identifier of the AI generation trace.
181 score: Numeric feedback score (e.g. 0.0-1.0 or 1-5).
182 owner_id: Owner scope; the item is recorded under this owner.
183 comment: Optional free-text comment stored in metadata.
184 metadata: Optional additional key-value metadata.
185 """
186 ...
188 async def get_feedback_stats(
189 self,
190 *,
191 owner_id: str,
192 model: str | None = None,
193 provider: str | None = None,
194 ) -> dict[str, Any]:
195 """Query aggregate feedback statistics for an owner.
197 Args:
198 owner_id: Owner scope; only this owner's items are aggregated.
199 model: Optional model name for context (currently informational).
200 provider: Optional provider name for context (currently informational).
201 """
202 ...
205__all__ = [
206 "FeedbackItem",
207 "FeedbackProtocol",
208 "FeedbackStoreProtocol",
209 "FeedbackSummary",
210 "FeedbackType",
211]