Coverage for agentos/evolution/autopilot.py: 30%
274 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-09 10:19 +0800
1"""
2Closed-Loop Self-Evolution v2 (v1.9.0)
4AutoPilot — from behavior signals to code changes, fully automated.
6Pipeline:
7 1. SignalCollector gathers user behavior (corrections, ratings, tool usage)
8 2. Learner detects patterns → generates EvolutionProposal
9 3. AutoPilot validates → generates code change → auto-tests → applies
10 4. Regression tests verify no breakage
11 5. Proposal archived with before/after metrics
13v2 New Features:
14 - CodeGenerator: LLM-based code diff generation from proposals
15 - AutoTester: Run regression suite before/after each change
16 - RollbackManager: Instant undo if regression detected
17 - Confidence Gating: Only auto-apply proposals above confidence threshold
18 - A/B Evaluator: Side-by-side before/after comparison
19 - EvolutionJournal: Full audit trail of every evolution step
20"""
22from __future__ import annotations
24import hashlib
25import json
26import os
27import subprocess
28import time
29from dataclasses import dataclass, field
30from datetime import datetime
31from enum import StrEnum
32from pathlib import Path
33from typing import Any
35from agentos.evolution.engine import EvolutionEngine, EvolutionProposal, EvolutionStatus
36from agentos.evolution.learner import Learner
38# ── Types ───────────────────────────────────────────────────────────
41class AutoPilotMode(StrEnum):
42 """AutoPilot operating mode."""
44 SUGGEST_ONLY = "suggest_only" # Only generate proposals, don't apply
45 ASK_BEFORE = "ask_before" # Generate + ask user before applying
46 CONFIDENCE_GATED = "confidence" # Auto-apply if confidence > threshold
47 FULL_AUTO = "full_auto" # Auto-apply everything (⚠️ use with guardrails)
50class ChangeResult(StrEnum):
51 """Result of an auto-applied change."""
53 SUCCESS = "success"
54 FAILED = "failed"
55 REGRESSION = "regression"
56 ROLLED_BACK = "rolled_back"
57 SKIPPED = "skipped"
60@dataclass
61class CodeChange:
62 """A code change generated from an evolution proposal."""
64 proposal_id: str
65 file_path: str
66 description: str
67 diff: str # Unified diff
68 old_content: str = "" # Pre-change content (for rollback)
69 new_content: str = "" # Post-change content
70 language: str = "python"
71 risk_level: str = "medium" # low / medium / high
72 test_results: dict[str, Any] = field(default_factory=dict)
75@dataclass
76class EvolutionRun:
77 """Record of a single evolution execution."""
79 run_id: str
80 proposal: EvolutionProposal
81 changes: list[CodeChange] = field(default_factory=list)
82 result: ChangeResult = ChangeResult.SKIPPED
83 started_at: float = 0.0
84 finished_at: float = 0.0
85 rollback_info: dict[str, Any] = field(default_factory=dict)
86 metrics_before: dict[str, Any] = field(default_factory=dict)
87 metrics_after: dict[str, Any] = field(default_factory=dict)
90# ── Code Generator ──────────────────────────────────────────────────
93class CodeGenerator:
94 """Generate code changes from evolution proposals using LLM.
96 Takes a high-level proposal (e.g., 'add retry logic to API calls')
97 and generates concrete unified diffs.
98 """
100 SYSTEM_PROMPT = """You are an expert Python code generator for an agent framework.
101Given an evolution proposal, generate precise, minimal code changes.
102Output ONLY a unified diff format. No explanations, no markdown code blocks.
103Focus on: correctness, backward compatibility, performance, readability."""
105 def __init__(self, llm_client=None):
106 self._llm = llm_client
108 async def generate(
109 self, proposal: EvolutionProposal, codebase: dict[str, str]
110 ) -> list[CodeChange]:
111 """Generate code changes for a proposal.
113 Args:
114 proposal: The evolution proposal to implement
115 codebase: Dict of {file_path: file_content} for context
117 Returns:
118 List of CodeChange objects with unified diffs.
119 """
120 changes: list[CodeChange] = []
122 if not self._llm:
123 # Fallback: generate skeleton changes based on proposal type
124 return self._skeleton_generate(proposal)
126 for target_file in proposal.target_files:
127 content = codebase.get(target_file, "")
128 prompt = self._build_prompt(proposal, target_file, content)
130 response = await self._llm.complete(prompt, system=self.SYSTEM_PROMPT)
131 diff = self._extract_diff(response)
133 if diff:
134 new_content = self._apply_diff(content, diff)
135 changes.append(
136 CodeChange(
137 proposal_id=proposal.id,
138 file_path=target_file,
139 description=proposal.description,
140 diff=diff,
141 old_content=content,
142 new_content=new_content,
143 risk_level=proposal.risk_level,
144 )
145 )
147 return changes
149 def _skeleton_generate(self, proposal: EvolutionProposal) -> list[CodeChange]:
150 """Skeleton code generation for proposals (no LLM available)."""
151 changes = []
152 for target_file in proposal.target_files:
153 changes.append(
154 CodeChange(
155 proposal_id=proposal.id,
156 file_path=target_file,
157 description=proposal.description,
158 diff=f"# SKELETON: {proposal.description}\n# File: {target_file}",
159 risk_level=proposal.risk_level,
160 )
161 )
162 return changes
164 def _build_prompt(self, proposal: EvolutionProposal, target_file: str, content: str) -> str:
165 return f"""Proposal: {proposal.description}
166Category: {proposal.category}
167File: {target_file}
168Priority: {proposal.priority}
170Current file content:
171```python
172{content[:3000]}
173```
175Generate a unified diff to implement this change. Focus only on {target_file}."""
177 def _extract_diff(self, response: str) -> str:
178 """Extract unified diff from LLM response."""
179 if response.startswith("---") or response.startswith("diff "):
180 return response
181 if "```diff" in response:
182 start = response.index("```diff") + 7
183 end = response.index("```", start) if "```" in response[start:] else len(response)
184 return response[start:end].strip()
185 return response.strip()
187 def _apply_diff(self, content: str, diff: str) -> str:
188 """Simple diff application for well-known patterns."""
189 if diff.startswith("# SKELETON"):
190 return content
191 try:
192 result = subprocess.run(
193 ["patch", "-o", "-", "-"],
194 input=f"--- a/file\n+++ b/file\n{diff}".encode(),
195 capture_output=True,
196 timeout=10,
197 )
198 if result.returncode == 0:
199 return result.stdout.decode()
200 except Exception:
201 pass
202 return content
205# ── Auto Tester ─────────────────────────────────────────────────────
208class AutoTester:
209 """Run test suite to validate changes."""
211 def __init__(self, test_dir: str = "", pytest_args: str = ""):
212 self._test_dir = Path(test_dir) if test_dir else Path("tests")
213 self._pytest_args = pytest_args or "-x --tb=short -q"
215 async def run_tests(self) -> dict[str, Any]:
216 """Run the test suite.
218 Returns:
219 Dict with passed/failed/total counts and error details.
220 """
221 try:
222 result = subprocess.run(
223 ["python3", "-m", "pytest", str(self._test_dir)] + self._pytest_args.split(),
224 capture_output=True,
225 text=True,
226 timeout=120,
227 cwd=str(self._test_dir.parent) if self._test_dir.parent else None,
228 )
229 passed = "passed" in result.stdout.lower() or result.returncode == 0
230 return {
231 "passed": passed,
232 "total": self._parse_test_count(result.stdout),
233 "failures": result.returncode if not passed else 0,
234 "output": result.stdout[-1000:],
235 "duration": 0,
236 }
237 except FileNotFoundError:
238 return {
239 "passed": True,
240 "total": 0,
241 "failures": 0,
242 "output": "pytest not installed",
243 "duration": 0,
244 }
245 except Exception as e:
246 return {"passed": False, "total": 0, "failures": 1, "output": str(e), "duration": 0}
248 def _parse_test_count(self, output: str) -> int:
249 """Parse test count from pytest output."""
250 for line in output.split("\n"):
251 if "passed" in line.lower():
252 try:
253 return int(line.strip().split()[0])
254 except (ValueError, IndexError):
255 pass
256 return 0
259# ── Rollback Manager ─────────────────────────────────────────────────
262class RollbackManager:
263 """Instant undo of any auto-applied change."""
265 def __init__(self, backup_dir: str = ""):
266 self._backup_dir = (
267 Path(backup_dir) if backup_dir else Path.home() / ".agentos" / "evolution" / "backups"
268 )
269 self._backup_dir.mkdir(parents=True, exist_ok=True)
270 self._history: list[dict[str, Any]] = []
272 def snapshot(self, file_path: str, content: str) -> str:
273 """Create a backup snapshot of a file before modification."""
274 snapshot_id = hashlib.sha256(f"{file_path}:{time.time()}".encode()).hexdigest()[:12]
275 snapshot_path = self._backup_dir / f"{snapshot_id}.bak"
276 snapshot_path.write_text(content, encoding="utf-8")
277 self._history.append(
278 {
279 "snapshot_id": snapshot_id,
280 "file_path": file_path,
281 "timestamp": time.time(),
282 "size": len(content),
283 }
284 )
285 return snapshot_id
287 def rollback(self, snapshot_id: str) -> bool:
288 """Restore file from snapshot."""
289 snapshot_path = self._backup_dir / f"{snapshot_id}.bak"
290 if not snapshot_path.exists():
291 return False
293 for entry in self._history:
294 if entry["snapshot_id"] == snapshot_id:
295 target = Path(entry["file_path"])
296 target.write_text(snapshot_path.read_text(encoding="utf-8"), encoding="utf-8")
297 return True
299 return False
301 def get_history(self, limit: int = 20) -> list[dict[str, Any]]:
302 """Get recent evolution history."""
303 return sorted(self._history, key=lambda x: x["timestamp"], reverse=True)[:limit]
306# ── A/B Evaluator ───────────────────────────────────────────────────
309class ABEvaluator:
310 """Compare agent performance before and after evolution changes."""
312 def __init__(self, test_cases: list[dict[str, str]] | None = None):
313 self._test_cases = test_cases or []
314 self._results_before: list[dict] = []
315 self._results_after: list[dict] = []
317 async def evaluate_before(self, agent) -> list[dict]:
318 """Run evaluation before changes."""
319 self._results_before = await self._run_eval_loop(agent)
320 return self._results_before
322 async def evaluate_after(self, agent) -> list[dict]:
323 """Run evaluation after changes."""
324 self._results_after = await self._run_eval_loop(agent)
325 return self._results_after
327 def compare(self) -> dict[str, Any]:
328 """Compare before/after results."""
329 if not self._results_before or not self._results_after:
330 return {"status": "no_data"}
332 before_success = sum(1 for r in self._results_before if r.get("passed", False))
333 after_success = sum(1 for r in self._results_after if r.get("passed", False))
334 total = max(len(self._results_before), len(self._results_after))
336 return {
337 "before_pass_rate": before_success / total if total else 0,
338 "after_pass_rate": after_success / total if total else 0,
339 "improvement": (after_success - before_success) / total if total else 0,
340 "regressions": after_success < before_success,
341 "total_cases": total,
342 }
344 async def _run_eval_loop(self, agent) -> list[dict]:
345 """Run evaluation loop."""
346 results = []
347 for case in self._test_cases:
348 try:
349 result = await agent.run(case.get("input", ""))
350 passed = case.get("expected", "") in str(result)
351 results.append(
352 {"case": case.get("id", ""), "passed": passed, "output": str(result)[:500]}
353 )
354 except Exception as e:
355 results.append({"case": case.get("id", ""), "passed": False, "error": str(e)})
356 return results
359# ── Evolution Journal ───────────────────────────────────────────────
362class EvolutionJournal:
363 """Complete audit trail of every evolution step."""
365 def __init__(self, journal_path: str = ""):
366 self._path = (
367 Path(journal_path)
368 if journal_path
369 else Path.home() / ".agentos" / "evolution" / "journal.jsonl"
370 )
371 self._path.parent.mkdir(parents=True, exist_ok=True)
373 def log(self, entry: dict[str, Any]):
374 """Append an entry to the journal."""
375 entry["_timestamp"] = datetime.now().isoformat()
376 with open(self._path, "a", encoding="utf-8") as f:
377 f.write(json.dumps(entry, ensure_ascii=False) + "\n")
379 def read(self, limit: int = 50) -> list[dict]:
380 """Read recent journal entries."""
381 if not self._path.exists():
382 return []
383 entries = []
384 with open(self._path, encoding="utf-8") as f:
385 for line in f:
386 entries.append(json.loads(line))
387 return entries[-limit:]
389 def stats(self) -> dict[str, Any]:
390 """Compute evolution statistics from journal."""
391 entries = self.read(limit=10000)
392 if not entries:
393 return {}
395 by_type: dict[str, int] = {}
396 by_result: dict[str, int] = {}
397 for entry in entries:
398 by_type[entry.get("type", "unknown")] = by_type.get(entry.get("type", "unknown"), 0) + 1
399 by_result[entry.get("result", "unknown")] = (
400 by_result.get(entry.get("result", "unknown"), 0) + 1
401 )
403 return {
404 "total_entries": len(entries),
405 "by_type": by_type,
406 "by_result": by_result,
407 "first_entry": entries[0].get("_timestamp", ""),
408 "last_entry": entries[-1].get("_timestamp", ""),
409 }
412# ── AutoPilot ───────────────────────────────────────────────────────
415class AutoPilot:
416 """Closed-loop self-evolution engine.
418 The AutoPilot orchestrates the entire evolution pipeline:
419 signals → insights → proposals → code changes → tests → apply/rollback.
421 Usage:
422 from agentos.evolution import SignalCollector, EvolutionEngine, Learner, AutoPilot
424 collector = SignalCollector()
425 engine = EvolutionEngine()
426 learner = Learner(collector, engine)
427 autopilot = AutoPilot(engine, learner, mode=AutoPilotMode.CONFIDENCE_GATED)
429 # After accumulating signals...
430 run = await autopilot.evolve()
431 print(f"Evolved: {run.result} — {len(run.changes)} changes applied")
432 """
434 def __init__(
435 self,
436 engine: EvolutionEngine,
437 learner: Learner,
438 mode: AutoPilotMode = AutoPilotMode.CONFIDENCE_GATED,
439 confidence_threshold: float = 0.7,
440 code_generator: CodeGenerator | None = None,
441 tester: AutoTester | None = None,
442 rollback: RollbackManager | None = None,
443 evaluator: ABEvaluator | None = None,
444 journal: EvolutionJournal | None = None,
445 ):
446 self.engine = engine
447 self.learner = learner
448 self.mode = mode
449 self.confidence_threshold = confidence_threshold
451 self.codegen = code_generator or CodeGenerator()
452 self.tester = tester or AutoTester()
453 self.rollback = rollback or RollbackManager()
454 self.evaluator = evaluator or ABEvaluator()
455 self.journal = journal or EvolutionJournal()
457 self._run_history: list[EvolutionRun] = []
459 async def evolve(self, agent=None) -> list[EvolutionRun]:
460 """Execute one complete evolution cycle.
462 1. Analyze signals → generate insights
463 2. Convert insights → proposals
464 3. For each proposal: generate code → test → apply/rollback
465 4. Journal everything
467 Args:
468 agent: Optional agent instance for A/B evaluation.
470 Returns:
471 List of EvolutionRun records for this cycle.
472 """
473 # Step 1: Analyze signals
474 insights = self.learner.analyze()
476 # Step 2: Generate proposals
477 proposals = []
478 for insight in insights:
479 proposal = self.learner.propose_from_insight(insight)
480 proposals.append(proposal)
482 # Step 3: Gate by confidence and mode
483 runs: list[EvolutionRun] = []
484 for proposal in proposals:
485 run = await self._process_proposal(proposal, agent)
486 runs.append(run)
487 self._run_history.append(run)
489 # Step 4: Journal results
490 self.journal.log(
491 {
492 "type": "evolution_cycle",
493 "insights": len(insights),
494 "proposals": len(proposals),
495 "applied": sum(1 for r in runs if r.result == ChangeResult.SUCCESS),
496 "failed": sum(1 for r in runs if r.result == ChangeResult.FAILED),
497 "rolled_back": sum(1 for r in runs if r.result == ChangeResult.ROLLED_BACK),
498 }
499 )
501 return runs
503 async def _process_proposal(self, proposal: EvolutionProposal, agent=None) -> EvolutionRun:
504 """Process a single proposal through the pipeline."""
505 run = EvolutionRun(
506 run_id=f"ev_{proposal.id}",
507 proposal=proposal,
508 started_at=time.time(),
509 )
511 # Check if we should proceed based on mode
512 if not self._should_proceed(proposal):
513 run.result = ChangeResult.SKIPPED
514 run.finished_at = time.time()
515 return run
517 # Snapshot before
518 for target in proposal.target_files:
519 if os.path.exists(target):
520 content = Path(target).read_text(encoding="utf-8")
521 self.rollback.snapshot(target, content)
523 # Evaluate before (if agent provided)
524 if agent:
525 run.metrics_before = await self.evaluator.compare()
527 # Generate code changes
528 codebase = {}
529 for target in proposal.target_files:
530 if os.path.exists(target):
531 codebase[target] = Path(target).read_text(encoding="utf-8")
533 changes = await self.codegen.generate(proposal, codebase)
535 if not changes:
536 run.result = ChangeResult.FAILED
537 run.finished_at = time.time()
538 return run
540 run.changes = changes
542 # Apply changes
543 for change in changes:
544 try:
545 if change.new_content:
546 Path(change.file_path).write_text(change.new_content, encoding="utf-8")
547 except Exception:
548 # Rollback and fail
549 for ch in run.changes:
550 self.rollback.rollback(ch.proposal_id)
551 run.result = ChangeResult.ROLLED_BACK
552 run.finished_at = time.time()
553 return run
555 # Run tests
556 test_results = await self.tester.run_tests()
558 if not test_results.get("passed", False):
559 # Regression detected — rollback
560 for change in changes:
561 if change.old_content:
562 Path(change.file_path).write_text(change.old_content, encoding="utf-8")
563 run.result = ChangeResult.REGRESSION
564 run.rollback_info = {"test_results": test_results}
565 run.finished_at = time.time()
566 return run
568 # Success!
569 run.result = ChangeResult.SUCCESS
570 run.test_results = test_results
572 # Evaluate after (if agent provided)
573 if agent:
574 run.metrics_after = await self.evaluator.compare()
576 # Update proposal status
577 proposal.status = EvolutionStatus.APPLIED
579 run.finished_at = time.time()
580 return run
582 def _should_proceed(self, proposal: EvolutionProposal) -> bool:
583 """Determine if we should proceed with this proposal based on mode."""
584 if self.mode == AutoPilotMode.SUGGEST_ONLY:
585 return False
587 if self.mode == AutoPilotMode.ASK_BEFORE:
588 # User interaction required — return False, caller must handle
589 return False
591 if self.mode == AutoPilotMode.CONFIDENCE_GATED:
592 return proposal.confidence >= self.confidence_threshold
594 if self.mode == AutoPilotMode.FULL_AUTO:
595 # Only safe changes in FULL_AUTO
596 return proposal.risk_level in ("low", "medium")
598 return False
600 def get_run_history(self, limit: int = 20) -> list[EvolutionRun]:
601 """Get recent evolution runs."""
602 return self._run_history[-limit:]
604 def get_stats(self) -> dict[str, Any]:
605 """Get AutoPilot statistics."""
606 runs = self._run_history
607 return {
608 "total_runs": len(runs),
609 "successful": sum(1 for r in runs if r.result == ChangeResult.SUCCESS),
610 "failed": sum(1 for r in runs if r.result == ChangeResult.FAILED),
611 "regressions": sum(1 for r in runs if r.result == ChangeResult.REGRESSION),
612 "rolled_back": sum(1 for r in runs if r.result == ChangeResult.ROLLED_BACK),
613 "total_changes": sum(len(r.changes) for r in runs),
614 "journal_stats": self.journal.stats(),
615 "rollback_history": len(self.rollback.get_history()),
616 }