Coverage for agentos/swarm/agent_monitor.py: 28%
218 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 19:15 +0800
1"""
2v1.9.6: Agent Self-Monitoring & Quality Gates.
4Each agent execution passes through configurable quality checks before
5results are accepted. Failed checks trigger automatic fallback or retry.
6"""
8from __future__ import annotations
10import time
11import uuid
12from collections.abc import Callable
13from dataclasses import dataclass, field
14from enum import StrEnum
15from typing import Any
18class GateStatus(StrEnum):
19 PASS = "pass"
20 FAIL = "fail"
21 WARN = "warn"
22 SKIP = "skip"
25class GateAction(StrEnum):
26 ACCEPT = "accept" # Accept result as-is
27 RETRY = "retry" # Retry the task
28 FALLBACK = "fallback" # Use fallback result
29 ABORT = "abort" # Abort the task
30 WARN = "warn" # Accept but flag warning
33@dataclass
34class GateResult:
35 """Result of a single quality gate check."""
37 name: str
38 status: GateStatus = GateStatus.PASS
39 action: GateAction = GateAction.ACCEPT
40 score: float = 1.0
41 threshold: float = 0.7
42 detail: str = ""
43 timestamp: float = field(default_factory=time.time)
45 def to_dict(self) -> dict:
46 return {
47 "name": self.name,
48 "status": self.status.value,
49 "action": self.action.value,
50 "score": self.score,
51 "threshold": self.threshold,
52 "detail": self.detail,
53 }
56@dataclass
57class MonitorReport:
58 """Complete self-monitoring report for a task execution."""
60 task_id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])
61 task_name: str = ""
62 gates: list[GateResult] = field(default_factory=list)
63 overall_status: GateStatus = GateStatus.PASS
64 overall_action: GateAction = GateAction.ACCEPT
65 total_checks: int = 0
66 passed: int = 0
67 failed: int = 0
68 warned: int = 0
69 duration_ms: float = 0.0
70 retries_used: int = 0
71 max_retries: int = 3
72 fallback_used: bool = False
73 output_summary: str = ""
75 def to_dict(self) -> dict:
76 return {
77 "task_id": self.task_id,
78 "task_name": self.task_name,
79 "gates": [g.to_dict() for g in self.gates],
80 "overall_status": self.overall_status.value,
81 "overall_action": self.overall_action.value,
82 "total_checks": self.total_checks,
83 "passed": self.passed,
84 "failed": self.failed,
85 "warned": self.warned,
86 "duration_ms": f"{self.duration_ms:.1f}",
87 "retries_used": self.retries_used,
88 "fallback_used": self.fallback_used,
89 "output_summary": self.output_summary[:200],
90 }
93class QualityGate:
94 """A single quality check that validates agent output.
96 Built-in gate types: output_not_empty, output_length, confidence_min,
97 schema_valid, no_error, latency_max.
98 """
100 def __init__(
101 self,
102 name: str,
103 check_fn: Callable[[Any, dict], tuple[bool, str, float]],
104 threshold: float = 0.7,
105 on_fail: GateAction = GateAction.RETRY,
106 on_warn: GateAction = GateAction.ACCEPT,
107 max_retries: int = 1,
108 ):
109 """
110 Args:
111 name: Gate name for reporting
112 check_fn: (output, context) → (passed, detail, score)
113 threshold: Score threshold for pass (0-1)
114 on_fail: Action when gate fails
115 on_warn: Action when gate warns
116 max_retries: Max retries for this specific gate
117 """
118 self.name = name
119 self._check = check_fn
120 self.threshold = threshold
121 self.on_fail = on_fail
122 self.on_warn = on_warn
123 self.max_retries = max_retries
125 def evaluate(self, output: Any, context: dict | None = None) -> GateResult:
126 """Run the quality check."""
127 ctx = context or {}
128 try:
129 passed, detail, score = self._check(output, ctx)
130 except Exception as e:
131 return GateResult(
132 name=self.name,
133 status=GateStatus.FAIL,
134 action=self.on_fail,
135 score=0.0,
136 threshold=self.threshold,
137 detail=f"Gate check error: {e}",
138 )
140 if score >= self.threshold and passed:
141 return GateResult(
142 name=self.name,
143 status=GateStatus.PASS,
144 action=GateAction.ACCEPT,
145 score=score,
146 threshold=self.threshold,
147 detail=detail,
148 )
149 elif score >= self.threshold * 0.6:
150 return GateResult(
151 name=self.name,
152 status=GateStatus.WARN,
153 action=self.on_warn,
154 score=score,
155 threshold=self.threshold,
156 detail=detail,
157 )
158 else:
159 return GateResult(
160 name=self.name,
161 status=GateStatus.FAIL,
162 action=self.on_fail,
163 score=score,
164 threshold=self.threshold,
165 detail=detail,
166 )
169class AgentMonitor:
170 """
171 Self-monitoring pipeline for agent task execution.
173 Runs each task output through a chain of quality gates. Based on gate results,
174 decides whether to accept, retry, fallback, or abort.
176 Usage:
177 monitor = AgentMonitor()
178 monitor.add_gate(output_not_empty_gate)
179 monitor.add_gate(confidence_min_gate)
181 result = await monitor.monitor_execution(
182 task_fn=lambda: agent.run(task),
183 task_name="research_query",
184 )
185 if result.overall_action == GateAction.ACCEPT:
186 ...
187 """
189 def __init__(self, max_retries: int = 3, default_fallback: Any = None):
190 self._gates: list[QualityGate] = []
191 self.max_retries = max_retries
192 self.default_fallback = default_fallback
194 def add_gate(self, gate: QualityGate) -> AgentMonitor:
195 """Add a quality gate to the pipeline."""
196 self._gates.append(gate)
197 return self
199 def add_gates(self, gates: list[QualityGate]) -> AgentMonitor:
200 """Add multiple quality gates."""
201 self._gates.extend(gates)
202 return self
204 async def monitor_execution(
205 self,
206 task_fn: Callable[[], Any],
207 task_name: str = "",
208 context: dict | None = None,
209 fallback_fn: Callable[[], Any] | None = None,
210 ) -> tuple[Any, MonitorReport]:
211 """Execute a task with full monitoring and quality gating.
213 Args:
214 task_fn: Async/sync function that executes the task
215 context: Additional context for gate evaluation
216 fallback_fn: Fallback function to call if gates fail with FALLBACK action
218 Returns:
219 Tuple of (final_output, monitor_report)
220 """
221 import asyncio
223 report = MonitorReport(task_name=task_name)
224 ctx = context or {}
225 start = time.time()
227 output = None
228 retries = 0
230 while retries <= self.max_retries:
231 # Execute task
232 try:
233 result = task_fn()
234 if asyncio.iscoroutine(result):
235 output = await result
236 else:
237 output = result
238 except Exception as e:
239 report.gates.append(
240 GateResult(
241 name="execution_error",
242 status=GateStatus.FAIL,
243 action=GateAction.RETRY,
244 score=0.0,
245 detail=str(e),
246 )
247 )
248 retries += 1
249 if retries > self.max_retries:
250 report.overall_status = GateStatus.FAIL
251 report.overall_action = GateAction.FALLBACK
252 break
253 continue
255 # Run gates
256 report.gates = []
257 any_fail = False
258 worst_action = GateAction.ACCEPT
259 action_prio = {
260 GateAction.ACCEPT: 0,
261 GateAction.WARN: 1,
262 GateAction.RETRY: 2,
263 GateAction.FALLBACK: 3,
264 GateAction.ABORT: 4,
265 }
267 for gate in self._gates:
268 gr = gate.evaluate(output, ctx)
269 report.gates.append(gr)
271 if gr.status == GateStatus.FAIL:
272 any_fail = True
273 if action_prio.get(gr.action, 0) > action_prio.get(worst_action, 0):
274 worst_action = gr.action
276 # Tally
277 report.total_checks = len(report.gates)
278 report.passed = sum(1 for g in report.gates if g.status == GateStatus.PASS)
279 report.failed = sum(1 for g in report.gates if g.status == GateStatus.FAIL)
280 report.warned = sum(1 for g in report.gates if g.status == GateStatus.WARN)
282 if not any_fail:
283 report.overall_status = GateStatus.PASS
284 report.overall_action = worst_action
285 report.retries_used = retries
286 report.output_summary = self._summarize(output)
287 report.duration_ms = (time.time() - start) * 1000
288 return output, report
290 # Handle failure
291 if worst_action == GateAction.ABORT:
292 report.overall_status = GateStatus.FAIL
293 report.overall_action = GateAction.ABORT
294 report.retries_used = retries
295 report.duration_ms = (time.time() - start) * 1000
296 return output, report
298 if worst_action == GateAction.FALLBACK:
299 report.overall_status = GateStatus.FAIL
300 report.overall_action = GateAction.FALLBACK
301 report.retries_used = retries
302 report.fallback_used = True
303 report.duration_ms = (time.time() - start) * 1000
305 if fallback_fn:
306 fb_result = fallback_fn()
307 if asyncio.iscoroutine(fb_result):
308 output = await fb_result
309 else:
310 output = fb_result
311 elif self.default_fallback is not None:
312 output = self.default_fallback
314 report.output_summary = self._summarize(output)
315 return output, report
317 # RETRY or WARN — continue loop
318 retries += 1
320 # Exhausted retries
321 report.overall_status = GateStatus.FAIL if report.failed > 0 else GateStatus.WARN
322 report.overall_action = GateAction.FALLBACK
323 report.retries_used = retries
324 report.duration_ms = (time.time() - start) * 1000
326 if fallback_fn:
327 fb_result = fallback_fn()
328 if asyncio.iscoroutine(fb_result):
329 output = await fb_result
330 else:
331 output = fb_result
332 elif self.default_fallback is not None:
333 output = self.default_fallback
335 report.output_summary = self._summarize(output)
336 return output, report
338 def _summarize(self, output: Any) -> str:
339 """Create a brief summary of output for reporting."""
340 if output is None:
341 return "None"
342 s = str(output)
343 if len(s) > 200:
344 return s[:197] + "..."
345 return s
348# ── Built-in Quality Gates ────────────────────────────────────────
351def output_not_empty(
352 min_length: int = 1,
353 threshold: float = 0.9,
354) -> QualityGate:
355 """Gate: output must not be empty."""
357 def check(output: Any, ctx: dict) -> tuple[bool, str, float]:
358 s = str(output).strip() if output else ""
359 score = min(1.0, len(s) / max(min_length, 1))
360 if not s:
361 return False, "Output is empty", 0.0
362 if len(s) < min_length:
363 return False, f"Output too short ({len(s)} < {min_length})", score
364 return True, f"Output length {len(s)} OK", 1.0
366 return QualityGate("output_not_empty", check, threshold, on_fail=GateAction.RETRY)
369def output_length_range(
370 min_len: int = 10,
371 max_len: int = 10000,
372 threshold: float = 0.8,
373) -> QualityGate:
374 """Gate: output length must be in range."""
376 def check(output: Any, ctx: dict) -> tuple[bool, str, float]:
377 s = str(output).strip() if output else ""
378 length = len(s)
379 if length < min_len:
380 return False, f"Output too short: {length} < {min_len}", length / max(min_len, 1)
381 if length > max_len:
382 return False, f"Output too long: {length} > {max_len}", max_len / length
383 return True, f"Output length {length} OK", 1.0
385 return QualityGate("output_length", check, threshold, on_fail=GateAction.WARN)
388def no_error_output(threshold: float = 0.95) -> QualityGate:
389 """Gate: output must not contain error/exception patterns."""
390 ERROR_PATTERNS = [ # noqa: N806
391 "Traceback (most recent call last)",
392 "Error:",
393 "Exception:",
394 "failed to",
395 "cannot be",
396 "invalid",
397 "permission denied",
398 ]
400 def check(output: Any, ctx: dict) -> tuple[bool, str, float]:
401 s = str(output).lower()
402 hits = [p for p in ERROR_PATTERNS if p.lower() in s]
403 if hits:
404 score = 1.0 - (len(hits) / len(ERROR_PATTERNS))
405 return False, f"Error patterns found: {hits[:3]}", max(0, score)
406 return True, "No error patterns", 1.0
408 return QualityGate("no_error", check, threshold, on_fail=GateAction.RETRY)
411def contains_keywords(
412 keywords: list[str],
413 min_hits: int = 1,
414 threshold: float = 0.7,
415) -> QualityGate:
416 """Gate: output must contain at least N keywords."""
418 def check(output: Any, ctx: dict) -> tuple[bool, str, float]:
419 s = str(output).lower()
420 hits = [kw for kw in keywords if kw.lower() in s]
421 score = min(1.0, len(hits) / max(min_hits, 1))
422 if len(hits) < min_hits:
423 missing = [kw for kw in keywords if kw.lower() not in s]
424 return False, f"Missing keywords: {missing[:5]}", score
425 return True, f"Found {len(hits)}/{len(keywords)} keywords", 1.0
427 return QualityGate("keywords", check, threshold, on_fail=GateAction.WARN)
430def latency_max(max_ms: float, threshold: float = 0.9) -> QualityGate:
431 """Gate: execution must complete within time limit (ms)."""
433 def check(output: Any, ctx: dict) -> tuple[bool, str, float]:
434 elapsed = ctx.get("_latency_ms", 0)
435 score = max(0, 1.0 - (elapsed / max_ms))
436 if elapsed > max_ms:
437 return False, f"Latency {elapsed:.0f}ms > {max_ms}ms", score
438 return True, f"Latency {elapsed:.0f}ms OK", 1.0
440 return QualityGate("latency", check, threshold, on_fail=GateAction.WARN)
443def confidence_min(min_confidence: float = 0.5, threshold: float = 0.8) -> QualityGate:
444 """Gate: fused confidence must meet minimum."""
446 def check(output: Any, ctx: dict) -> tuple[bool, str, float]:
447 confidence = ctx.get("_confidence", 0.0)
448 score = min(1.0, confidence / max(min_confidence, 0.01))
449 if confidence < min_confidence:
450 return False, f"Confidence {confidence:.2f} < {min_confidence}", score
451 return True, f"Confidence {confidence:.2f} OK", 1.0
453 return QualityGate("confidence", check, threshold, on_fail=GateAction.RETRY)