« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Feedback collection system for continuous learning.
3Collects user feedback on AI/ML predictions for model improvement.
4"""
6from __future__ import annotations
8from typing import Any
10from lexigram.ai.feedback.constants import MAX_CONTEXT_SIZE, MAX_FEEDBACK_TEXT_LENGTH
11from lexigram.ai.feedback.exceptions import FeedbackTooLargeError
12from lexigram.ai.feedback.storage.protocols import FeedbackStoreProtocol
13from lexigram.ai.feedback.types import FeedbackItem, FeedbackType
14from lexigram.serialization import dumps_str
17class FeedbackCollector:
18 """Collects and stores user feedback.
20 Provides methods to capture different types of feedback and
21 integrate with ML retraining pipelines.
23 Example:
24 >>> collector = FeedbackCollector()
25 >>> # Collect rating
26 >>> await collector.collect_rating(
27 ... rating=5,
28 ... context={"model": "gpt-4", "input": "Hello"}
29 ... )
30 >>> # Collect correction
31 >>> await collector.collect_correction(
32 ... original="incorrect output",
33 ... corrected="correct output",
34 ... context={"model_id": "123"}
35 ... )
36 >>> # Get all feedback
37 >>> items = collector.get_feedback()
38 """
40 def __init__(self, storage: FeedbackStoreProtocol | None = None):
41 """Initialize feedback collector.
43 Args:
44 storage: Optional storage backend (e.g., database, file)
45 If None, uses in-memory storage
46 """
47 self.storage = storage
48 self._feedback: list[FeedbackItem] = []
50 async def collect_rating(
51 self,
52 rating: float,
53 *,
54 owner_id: str,
55 context: dict[str, Any] | None = None,
56 metadata: dict[str, Any] | None = None,
57 ) -> str:
58 """Collect a rating feedback.
60 Args:
61 rating: Numeric rating value.
62 owner_id: Owner scope; the item is recorded under this owner.
63 context: Context about what was rated.
64 metadata: Additional metadata.
66 Returns:
67 Feedback ID.
68 """
69 item = FeedbackItem(
70 feedback_type=FeedbackType.RATING,
71 value=rating,
72 owner_id=owner_id,
73 context=context or {},
74 metadata=metadata or {},
75 )
77 await self._store(item)
78 return item.id
80 async def collect_text(
81 self,
82 text: str,
83 *,
84 owner_id: str,
85 context: dict[str, Any] | None = None,
86 metadata: dict[str, Any] | None = None,
87 ) -> str:
88 """Collect text feedback.
90 Args:
91 text: Feedback text.
92 owner_id: Owner scope; the item is recorded under this owner.
93 context: Context about what feedback is for.
94 metadata: Additional metadata.
96 Returns:
97 Feedback ID.
98 """
99 item = FeedbackItem(
100 feedback_type=FeedbackType.TEXT,
101 value=text,
102 owner_id=owner_id,
103 context=context or {},
104 metadata=metadata or {},
105 )
107 await self._store(item)
108 return item.id
110 async def collect_correction(
111 self,
112 original: str,
113 corrected: str,
114 *,
115 owner_id: str,
116 context: dict[str, Any] | None = None,
117 metadata: dict[str, Any] | None = None,
118 ) -> str:
119 """Collect a correction feedback.
121 Args:
122 original: Original (incorrect) output.
123 corrected: Corrected output.
124 owner_id: Owner scope; the item is recorded under this owner.
125 context: Context.
126 metadata: Additional metadata.
128 Returns:
129 Feedback ID.
130 """
131 item = FeedbackItem(
132 feedback_type=FeedbackType.CORRECTION,
133 value={"original": original, "corrected": corrected},
134 owner_id=owner_id,
135 context=context or {},
136 metadata=metadata or {},
137 )
139 await self._store(item)
140 return item.id
142 async def collect_label(
143 self,
144 label: Any,
145 input_data: Any,
146 *,
147 owner_id: str,
148 context: dict[str, Any] | None = None,
149 metadata: dict[str, Any] | None = None,
150 ) -> str:
151 """Collect ground truth label.
153 Useful for supervised learning and model retraining.
155 Args:
156 label: Ground truth label.
157 input_data: Input that this label corresponds to.
158 owner_id: Owner scope; the item is recorded under this owner.
159 context: Context.
160 metadata: Additional metadata.
162 Returns:
163 Feedback ID.
164 """
165 item = FeedbackItem(
166 feedback_type=FeedbackType.LABEL,
167 value={"label": label, "input": input_data},
168 owner_id=owner_id,
169 context=context or {},
170 metadata=metadata or {},
171 )
173 await self._store(item)
174 return item.id
176 async def _store(self, item: FeedbackItem) -> None:
177 """Store feedback item, enforcing the declared size limits.
179 Rejects payloads exceeding MAX_FEEDBACK_TEXT_LENGTH (TEXT values)
180 or MAX_CONTEXT_SIZE (serialized context/metadata) with
181 FeedbackTooLargeError before any mutation.
183 Args:
184 item: Feedback item to store
186 Raises:
187 FeedbackTooLargeError: If the item exceeds the size limits.
188 """
189 if (
190 item.feedback_type == FeedbackType.TEXT
191 and isinstance(item.value, str)
192 and len(item.value) > MAX_FEEDBACK_TEXT_LENGTH
193 ):
194 raise FeedbackTooLargeError(
195 f"feedback text exceeds the {MAX_FEEDBACK_TEXT_LENGTH}-character limit"
196 )
197 if len(dumps_str(item.context, default=str)) > MAX_CONTEXT_SIZE:
198 raise FeedbackTooLargeError(
199 f"serialized context exceeds the {MAX_CONTEXT_SIZE}-character limit"
200 )
201 if len(dumps_str(item.metadata, default=str)) > MAX_CONTEXT_SIZE:
202 raise FeedbackTooLargeError(
203 f"serialized metadata exceeds the {MAX_CONTEXT_SIZE}-character limit"
204 )
205 self._feedback.append(item)
206 if self.storage:
207 await self.storage.save(item)
209 async def get_feedback(
210 self,
211 *,
212 owner_id: str,
213 feedback_type: FeedbackType | None = None,
214 limit: int | None = None,
215 ) -> list[FeedbackItem]:
216 """Get feedback items for an owner.
218 Args:
219 owner_id: Owner scope; only this owner's items are returned.
220 feedback_type: Optional filter by type.
221 limit: Maximum number of items to return.
223 Returns:
224 List of feedback items from memory or the storage backend.
225 """
226 if self.storage:
227 return await self._get_feedback_from_storage(
228 owner_id=owner_id,
229 feedback_type=feedback_type,
230 limit=limit,
231 )
232 return self._filter_memory_feedback(
233 owner_id=owner_id, feedback_type=feedback_type, limit=limit
234 )
236 async def _get_feedback_from_storage(
237 self,
238 *,
239 owner_id: str,
240 feedback_type: FeedbackType | None,
241 limit: int | None,
242 ) -> list[FeedbackItem]:
243 if limit == 0:
244 return []
246 if self.storage is None:
247 return []
249 if feedback_type:
250 per_type_limit = limit if limit is not None else 100
251 return await self.storage.find_by_type(
252 feedback_type,
253 owner_id=owner_id,
254 limit=per_type_limit,
255 )
257 per_type_limit = limit if limit is not None else 100
258 results: list[FeedbackItem] = []
259 for ft in FeedbackType:
260 batch = await self.storage.find_by_type(
261 ft, owner_id=owner_id, limit=per_type_limit
262 )
263 results.extend(batch)
265 results.sort(key=lambda item: item.created_at, reverse=True)
266 if limit:
267 return results[:limit]
268 return results
270 def _filter_memory_feedback(
271 self,
272 *,
273 owner_id: str,
274 feedback_type: FeedbackType | None,
275 limit: int | None,
276 ) -> list[FeedbackItem]:
277 items = [item for item in self._feedback if item.owner_id == owner_id]
278 if feedback_type:
279 items = [item for item in items if item.feedback_type == feedback_type]
280 if limit is not None:
281 items = items[:limit]
282 return items
284 async def get_feedback_dict(
285 self,
286 *,
287 owner_id: str,
288 feedback_type: FeedbackType | None = None,
289 limit: int | None = None,
290 ) -> list[dict[str, Any]]:
291 """Get feedback as dictionaries for an owner.
293 Args:
294 owner_id: Owner scope; only this owner's items are returned.
295 feedback_type: Optional filter by type.
296 limit: Maximum number of items.
298 Returns:
299 List of feedback dictionaries.
300 """
301 items = await self.get_feedback(
302 owner_id=owner_id, feedback_type=feedback_type, limit=limit
303 )
304 return [item.to_dict() for item in items]
306 def clear(self) -> None:
307 """Clear all feedback from the in-memory buffer."""
308 self._feedback.clear()
310 def __len__(self) -> int:
311 """Number of feedback items."""
312 return len(self._feedback)
314 def __repr__(self) -> str:
315 """String representation."""
316 return f"FeedbackCollector(items={len(self._feedback)})"