Coverage for agentos/swarm/eval_feedback_loop.py: 31%
131 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:19 +0800
1"""
2v1.9.4: Eval-Feedback Loop — closes the gap between CompositeScorer and AutoPilot.
4Wires evaluation scores back into the execution layer, creating a true
5execute → evaluate → feedback → retry 闭环 (closed loop).
6"""
8from __future__ import annotations
10import asyncio
11import time
12from collections.abc import Callable
13from dataclasses import dataclass, field
14from typing import Any
17@dataclass
18class FeedbackSignal:
19 """A signal derived from evaluation that triggers self-healing."""
21 source: str # Which scorer produced this
22 metric: str # metric name (e.g. "rouge_l", "bleu", "judge")
23 score: float # raw score
24 threshold: float # expected threshold
25 passed: bool # did it meet threshold?
26 detail: str = "" # human-readable detail
27 suggestion: str = "" # what to improve
30@dataclass
31class RetryConfig:
32 """Configuration for the retry loop."""
34 max_retries: int = 3
35 backoff_base: float = 1.0 # seconds
36 backoff_multiplier: float = 2.0
37 score_improvement_min: float = 0.05 # min score gain to consider improvement
38 timeout: float = 60.0 # total loop timeout (seconds)
41@dataclass
42class LoopResult:
43 """Result of a feedback loop execution."""
45 task: str
46 final_output: Any = None
47 scores: dict[str, float] = field(default_factory=dict)
48 attempts: int = 0
49 best_score: float = 0.0
50 converged: bool = False
51 duration: float = 0.0
52 history: list[dict] = field(default_factory=list) # each attempt trace
55class EvalFeedbackLoop:
56 """Connects evaluation scores to AutoPilot-style retry with incremental
57 prompt refinement.
59 Usage:
60 loop = EvalFeedbackLoop(scorer, RetryConfig(max_retries=3))
61 result = await loop.run(task, executor_fn, expected_output)
62 """
64 def __init__(
65 self,
66 scorer: Any = None, # CompositeScorer / CompositeScorerV2
67 config: RetryConfig | None = None,
68 reflection_prompt: str | None = None,
69 ):
70 self._scorer = scorer
71 self._config = config or RetryConfig()
72 self._reflection_prompt = reflection_prompt or (
73 "The previous attempt scored {score:.3f} (threshold: {threshold:.3f}). "
74 "Weak metrics: {weak_metrics}. "
75 "Please improve the output focusing on these weaknesses."
76 )
78 async def run(
79 self,
80 task: str,
81 executor: Callable[[str], Any],
82 expected: str = "",
83 strategy: str = "general",
84 ) -> LoopResult:
85 """Execute task with evaluation-driven retry loop.
87 Args:
88 task: original task description
89 executor: async/sync callable (task_str) → output
90 expected: expected/reference output for scoring
91 strategy: scoring strategy (qa/code/summary/translation)
93 Returns:
94 LoopResult with final output and convergence info
95 """
96 start = time.time()
97 result = LoopResult(task=task)
98 previous_score = 0.0
100 for attempt in range(1, self._config.max_retries + 1):
101 # Execute
102 output = executor(task)
103 if asyncio.iscoroutine(output):
104 output = await output
106 attempt_trace = {"attempt": attempt, "output": str(output)[:500]}
108 # Score
109 scores = self._score(output, expected, strategy)
110 attempt_trace["scores"] = scores
111 score = scores.get("weighted", 0.0)
112 passed = scores.get("passed", False)
113 attempt_trace["passed"] = passed
115 result.history.append(attempt_trace)
117 # Track best
118 if score > result.best_score:
119 result.best_score = score
120 result.final_output = output
122 # Emit feedback signal
123 signals = self._signals_from_scores(scores, strategy)
124 attempt_trace["signals"] = [
125 {"metric": s.metric, "score": s.score, "passed": s.passed} for s in signals
126 ]
128 # Check convergence
129 if passed:
130 result.converged = True
131 result.attempts = attempt
132 result.scores = scores
133 result.duration = time.time() - start
134 return result
136 # Improvement check
137 if attempt > 1 and (score - previous_score) < self._config.score_improvement_min:
138 result.converged = False
139 result.attempts = attempt
140 result.scores = scores
141 result.final_output = output
142 result.duration = time.time() - start
143 return result
145 previous_score = score
147 # Refine task prompt for next attempt
148 task = self._refine_task(task, signals, score, attempt)
149 attempt_trace["refined_task"] = task
151 # Backoff
152 wait = self._config.backoff_base * (self._config.backoff_multiplier ** (attempt - 1))
153 if (time.time() - start + wait) > self._config.timeout:
154 break
155 await asyncio.sleep(wait)
157 result.attempts = self._config.max_retries
158 result.scores = scores
159 result.duration = time.time() - start
160 return result
162 def _score(self, output: str, expected: str, strategy: str) -> dict[str, Any]:
163 """Score output against expected using CompositeScorer."""
164 if not self._scorer or not expected:
165 # Heuristic scoring when no scorer/reference available
166 return self._heuristic_score(output, expected)
168 try:
169 result = self._scorer.score(
170 reference=expected,
171 candidate=str(output),
172 task=strategy,
173 )
174 return {
175 "weighted": result.weighted_score,
176 "passed": result.passed,
177 "details": result.details,
178 "raw_scores": result.scores,
179 }
180 except Exception:
181 return self._heuristic_score(output, expected)
183 def _heuristic_score(self, output: str, expected: str) -> dict:
184 """Fallback scoring when no scorer is available."""
185 if not expected:
186 # No reference — score based on output quality heuristics
187 text = str(output) if output else ""
188 quality = 0.5
189 if len(text) > 50:
190 quality += 0.1
191 if len(text) > 200:
192 quality += 0.1
193 if any(kw in text.lower() for kw in ("conclusion", "result", "answer")):
194 quality += 0.1
195 return {"weighted": min(quality, 1.0), "passed": quality >= 0.5, "details": "heuristic"}
197 # Check contains match
198 contains = 1.0 if expected.lower() in str(output).lower() else 0.0
199 return {"weighted": contains * 0.5, "passed": contains > 0, "details": "contains_heuristic"}
201 def _signals_from_scores(self, scores: dict, strategy: str) -> list[FeedbackSignal]:
202 """Convert score dict to FeedbackSignal list."""
203 raw = scores.get("raw_scores", {})
204 thresholds = {
205 "qa": {"rouge_l": 0.3, "contains": 0.5, "judge": 0.55},
206 "code": {"rouge_l": 0.1, "exact": 0.3, "contains": 0.5, "judge": 0.55},
207 "summary": {"rouge_l": 0.5, "semantic": 0.4, "judge": 0.55},
208 "translation": {"bleu": 0.3, "rouge_l": 0.3, "judge": 0.55},
209 }
210 strat_thresholds = thresholds.get(strategy, {"rouge_l": 0.3, "contains": 0.5})
212 signals = []
213 for metric, score in raw.items():
214 threshold = strat_thresholds.get(metric, 0.5)
215 passed = score >= threshold
216 detail = f"{metric}={score:.3f} vs {threshold:.3f}"
217 suggestion = ""
218 if not passed:
219 suggestion = self._suggestion_for_metric(metric, score, threshold)
220 signals.append(
221 FeedbackSignal(
222 source=strategy,
223 metric=metric,
224 score=score,
225 threshold=threshold,
226 passed=passed,
227 detail=detail,
228 suggestion=suggestion,
229 )
230 )
231 return signals
233 def _suggestion_for_metric(self, metric: str, score: float, threshold: float) -> str:
234 """Generate improvement suggestion based on weak metric."""
235 gap = threshold - score
236 suggestions = {
237 "rouge_l": "Make output more comprehensive; include key phrases from expected answer.",
238 "bleu": "Improve translation accuracy; check terminology and phrasing.",
239 "exact": "Output format doesn't match expected; check structure and delimiters.",
240 "contains": f"Missing key concepts; include: (gap: {gap:.2f})",
241 "semantic": "Semantic meaning differs; rephrase to be closer to expected intent.",
242 "judge": "Output quality below LLM-judge threshold; improve clarity and completeness.",
243 }
244 return suggestions.get(metric, f"Improve {metric} by at least {gap:.2f}")
246 def _refine_task(
247 self,
248 task: str,
249 signals: list[FeedbackSignal],
250 current_score: float,
251 attempt: int,
252 ) -> str:
253 """Enrich task prompt with feedback for next attempt."""
254 weak = [s for s in signals if not s.passed]
255 if not weak:
256 return task
258 weak_metrics = ", ".join(f"{s.metric}({s.score:.2f} < {s.threshold:.2f})" for s in weak)
259 suggestions = "; ".join(s.suggestion for s in weak)
261 reflection = (
262 f"[Retry #{attempt} feedback — score {current_score:.3f}] "
263 f"Weak: {weak_metrics}. {suggestions}"
264 )
266 # Append reflection to task
267 if "---" in task:
268 base, _ = task.split("---", 1)
269 return f"{base.strip()}\n---\n{reflection}"
270 return f"{task}\n---\n{reflection}"