« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""High-level feedback submission and aggregation service."""
3from __future__ import annotations
5from typing import Any
7from lexigram.ai.feedback.constants import MAX_CONTEXT_SIZE, MAX_FEEDBACK_TEXT_LENGTH
8from lexigram.ai.feedback.exceptions import FeedbackTooLargeError
9from lexigram.contracts.ai.feedback import (
10 FeedbackItem,
11 FeedbackStoreProtocol,
12 FeedbackType,
13)
14from lexigram.logging import (
15 get_logger,
16)
17from lexigram.serialization import dumps_str
19__all__ = ["FeedbackService"]
21logger = get_logger(__name__)
24class FeedbackService:
25 """High-level feedback submission and querying service.
27 Implements :class:`~lexigram.contracts.ai.feedback.FeedbackProtocol`.
28 Wraps a :class:`~lexigram.contracts.ai.feedback.FeedbackStoreProtocol`
29 with trace-id correlation and aggregation logic.
31 Args:
32 store: Optional storage backend; when ``None`` the service operates
33 in a degraded no-op mode and logs warnings instead of persisting.
34 """
36 def __init__(self, store: FeedbackStoreProtocol | None = None) -> None:
37 self._store = store
39 async def submit_feedback(
40 self,
41 trace_id: str,
42 score: float,
43 *,
44 owner_id: str,
45 comment: str | None = None,
46 metadata: dict[str, Any] | None = None,
47 ) -> None:
48 """Submit feedback for a traced AI generation.
50 Args:
51 trace_id: Identifier of the AI generation trace.
52 score: Numeric feedback score (e.g. 0.0–1.0 or 1–5).
53 owner_id: Owner scope; the item is recorded under this owner.
54 comment: Optional free-text comment stored in metadata.
55 metadata: Optional additional key-value metadata.
57 Raises:
58 FeedbackTooLargeError: If comment exceeds
59 MAX_FEEDBACK_TEXT_LENGTH characters or the serialized
60 metadata (including the folded comment) exceeds
61 MAX_CONTEXT_SIZE characters.
62 """
63 meta: dict[str, Any] = dict(metadata or {})
64 if comment is not None:
65 if len(comment) > MAX_FEEDBACK_TEXT_LENGTH:
66 raise FeedbackTooLargeError(
67 f"feedback text exceeds the {MAX_FEEDBACK_TEXT_LENGTH}-character limit"
68 )
69 meta["comment"] = comment
70 if len(dumps_str(meta, default=str)) > MAX_CONTEXT_SIZE:
71 raise FeedbackTooLargeError(
72 f"serialized metadata exceeds the {MAX_CONTEXT_SIZE}-character limit"
73 )
75 if self._store is None:
76 logger.debug("feedback_store_unavailable", trace_id=trace_id)
77 return
79 item = FeedbackItem(
80 feedback_type=FeedbackType.RATING,
81 value=score,
82 owner_id=owner_id,
83 context={"trace_id": trace_id},
84 metadata=meta,
85 )
87 result = await self._store.save(item)
88 if result.is_ok():
89 logger.info(
90 "feedback_submitted",
91 trace_id=trace_id,
92 score=score,
93 owner_id=owner_id,
94 )
95 else:
96 logger.error(
97 "feedback_submit_failed",
98 trace_id=trace_id,
99 error=str(result.unwrap_err()),
100 )
102 async def get_feedback_stats(
103 self,
104 *,
105 owner_id: str,
106 model: str | None = None,
107 provider: str | None = None,
108 ) -> dict[str, Any]:
109 """Get aggregate feedback statistics for an owner.
111 Args:
112 owner_id: Owner scope; only this owner's items are aggregated.
113 model: Optional model name for context (currently informational).
114 provider: Optional provider name for context (currently informational).
116 Returns:
117 Dictionary with ``total_count``, ``average_rating``, and
118 ``by_type`` breakdown, plus optional ``model``/``provider`` keys
119 when supplied.
120 """
121 if self._store is None:
122 logger.debug("feedback_store_unavailable")
123 return {"total_count": 0, "average_rating": None, "by_type": {}}
125 summary = await self._store.aggregate(owner_id=owner_id, window_hours=24)
126 stats: dict[str, Any] = {
127 "total_count": summary.total_count,
128 "average_rating": summary.average_rating,
129 "by_type": summary.count_by_type,
130 }
131 if model is not None:
132 stats["model"] = model
133 if provider is not None:
134 stats["provider"] = provider
135 return stats