« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Feedback processor registry for extensible feedback handling."""
3from __future__ import annotations
5from typing import Any, Protocol
7from lexigram.ai.feedback.services.collector import FeedbackCollector
10class FeedbackProcessor(Protocol):
11 """Protocol for feedback processors."""
13 async def process(
14 self,
15 value: Any,
16 context: dict[str, Any],
17 collector: FeedbackCollector,
18 *,
19 owner_id: str,
20 ) -> str:
21 """Process feedback and return feedback ID.
23 Args:
24 value: Feedback value.
25 context: Context dict for the feedback.
26 collector: Collector to store the feedback.
27 owner_id: Owner scope; the item is recorded under this owner.
28 """
29 ...
32class RatingFeedbackProcessor:
33 """Processor for rating feedback."""
35 async def process(
36 self,
37 value: Any,
38 context: dict[str, Any],
39 collector: FeedbackCollector,
40 *,
41 owner_id: str,
42 ) -> str:
43 return await collector.collect_rating(
44 rating=value, owner_id=owner_id, context=context
45 )
48class TextFeedbackProcessor:
49 """Processor for text feedback."""
51 async def process(
52 self,
53 value: Any,
54 context: dict[str, Any],
55 collector: FeedbackCollector,
56 *,
57 owner_id: str,
58 ) -> str:
59 return await collector.collect_text(
60 text=value, owner_id=owner_id, context=context
61 )
64class CorrectionFeedbackProcessor:
65 """Processor for correction feedback."""
67 async def process(
68 self,
69 value: Any,
70 context: dict[str, Any],
71 collector: FeedbackCollector,
72 *,
73 owner_id: str,
74 ) -> str:
75 return await collector.collect_correction(
76 original=value["original"],
77 corrected=value["corrected"],
78 owner_id=owner_id,
79 context=context,
80 )
83class LabelFeedbackProcessor:
84 """Processor for label feedback."""
86 async def process(
87 self,
88 value: Any,
89 context: dict[str, Any],
90 collector: FeedbackCollector,
91 *,
92 owner_id: str,
93 ) -> str:
94 return await collector.collect_label(
95 label=value["label"],
96 input_data=value["input"],
97 owner_id=owner_id,
98 context=context,
99 )
102class FeedbackProcessorRegistry:
103 """Central registry for feedback processors."""
105 def __init__(self) -> None:
106 self._processors: dict[str, FeedbackProcessor] = {}
108 @classmethod
109 def with_defaults(cls) -> FeedbackProcessorRegistry:
110 """Create a registry pre-populated with the built-in feedback processors."""
111 from lexigram.ai.feedback.types import FeedbackType
113 registry = cls()
114 registry._processors[FeedbackType.RATING] = RatingFeedbackProcessor()
115 registry._processors[FeedbackType.TEXT] = TextFeedbackProcessor()
116 registry._processors[FeedbackType.CORRECTION] = CorrectionFeedbackProcessor()
117 registry._processors[FeedbackType.LABEL] = LabelFeedbackProcessor()
118 return registry
120 def register(self, fb_type: str, processor: FeedbackProcessor) -> None:
121 """Register a new feedback processor."""
122 self._processors[fb_type] = processor
124 async def process(
125 self,
126 fb_type: str,
127 value: Any,
128 context: dict[str, Any],
129 collector: FeedbackCollector,
130 *,
131 owner_id: str,
132 ) -> str:
133 """Process feedback using the appropriate processor.
135 Args:
136 fb_type: Feedback type key.
137 value: Feedback value.
138 context: Context dict for the feedback.
139 collector: Collector to store the feedback.
140 owner_id: Owner scope; the item is recorded under this owner.
141 """
142 processor = self._processors.get(fb_type)
143 if not processor:
144 raise ValueError(f"No processor for feedback type: {fb_type}")
145 return await processor.process(value, context, collector, owner_id=owner_id)