Coverage for agentos/evolution/autopilot.py: 30%
274 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +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 Enum
32from pathlib import Path
33from typing import Optional, Any
35from agentos.evolution.engine import EvolutionEngine, EvolutionProposal, EvolutionStatus
36from agentos.evolution.learner import Learner
39# ── Types ───────────────────────────────────────────────────────────
41class AutoPilotMode(str, Enum):
42 """AutoPilot operating mode."""
43 SUGGEST_ONLY = "suggest_only" # Only generate proposals, don't apply
44 ASK_BEFORE = "ask_before" # Generate + ask user before applying
45 CONFIDENCE_GATED = "confidence" # Auto-apply if confidence > threshold
46 FULL_AUTO = "full_auto" # Auto-apply everything (⚠️ use with guardrails)
49class ChangeResult(str, Enum):
50 """Result of an auto-applied change."""
51 SUCCESS = "success"
52 FAILED = "failed"
53 REGRESSION = "regression"
54 ROLLED_BACK = "rolled_back"
55 SKIPPED = "skipped"
58@dataclass
59class CodeChange:
60 """A code change generated from an evolution proposal."""
61 proposal_id: str
62 file_path: str
63 description: str
64 diff: str # Unified diff
65 old_content: str = "" # Pre-change content (for rollback)
66 new_content: str = "" # Post-change content
67 language: str = "python"
68 risk_level: str = "medium" # low / medium / high
69 test_results: dict[str, Any] = field(default_factory=dict)
71@dataclass
72class EvolutionRun:
73 """Record of a single evolution execution."""
74 run_id: str
75 proposal: EvolutionProposal
76 changes: list[CodeChange] = field(default_factory=list)
77 result: ChangeResult = ChangeResult.SKIPPED
78 started_at: float = 0.0
79 finished_at: float = 0.0
80 rollback_info: dict[str, Any] = field(default_factory=dict)
81 metrics_before: dict[str, Any] = field(default_factory=dict)
82 metrics_after: dict[str, Any] = field(default_factory=dict)
85# ── Code Generator ──────────────────────────────────────────────────
87class CodeGenerator:
88 """Generate code changes from evolution proposals using LLM.
90 Takes a high-level proposal (e.g., 'add retry logic to API calls')
91 and generates concrete unified diffs.
92 """
94 SYSTEM_PROMPT = """You are an expert Python code generator for an agent framework.
95Given an evolution proposal, generate precise, minimal code changes.
96Output ONLY a unified diff format. No explanations, no markdown code blocks.
97Focus on: correctness, backward compatibility, performance, readability."""
99 def __init__(self, llm_client=None):
100 self._llm = llm_client
102 async def generate(self, proposal: EvolutionProposal, codebase: dict[str, str]) -> list[CodeChange]:
103 """Generate code changes for a proposal.
105 Args:
106 proposal: The evolution proposal to implement
107 codebase: Dict of {file_path: file_content} for context
109 Returns:
110 List of CodeChange objects with unified diffs.
111 """
112 changes: list[CodeChange] = []
114 if not self._llm:
115 # Fallback: generate skeleton changes based on proposal type
116 return self._skeleton_generate(proposal)
118 for target_file in proposal.target_files:
119 content = codebase.get(target_file, "")
120 prompt = self._build_prompt(proposal, target_file, content)
122 response = await self._llm.complete(prompt, system=self.SYSTEM_PROMPT)
123 diff = self._extract_diff(response)
125 if diff:
126 new_content = self._apply_diff(content, diff)
127 changes.append(CodeChange(
128 proposal_id=proposal.id,
129 file_path=target_file,
130 description=proposal.description,
131 diff=diff,
132 old_content=content,
133 new_content=new_content,
134 risk_level=proposal.risk_level,
135 ))
137 return changes
139 def _skeleton_generate(self, proposal: EvolutionProposal) -> list[CodeChange]:
140 """Skeleton code generation for proposals (no LLM available)."""
141 changes = []
142 for target_file in proposal.target_files:
143 changes.append(CodeChange(
144 proposal_id=proposal.id,
145 file_path=target_file,
146 description=proposal.description,
147 diff=f"# SKELETON: {proposal.description}\n# File: {target_file}",
148 risk_level=proposal.risk_level,
149 ))
150 return changes
152 def _build_prompt(self, proposal: EvolutionProposal, target_file: str, content: str) -> str:
153 return f"""Proposal: {proposal.description}
154Category: {proposal.category}
155File: {target_file}
156Priority: {proposal.priority}
158Current file content:
159```python
160{content[:3000]}
161```
163Generate a unified diff to implement this change. Focus only on {target_file}."""
165 def _extract_diff(self, response: str) -> str:
166 """Extract unified diff from LLM response."""
167 if response.startswith("---") or response.startswith("diff "):
168 return response
169 if "```diff" in response:
170 start = response.index("```diff") + 7
171 end = response.index("```", start) if "```" in response[start:] else len(response)
172 return response[start:end].strip()
173 return response.strip()
175 def _apply_diff(self, content: str, diff: str) -> str:
176 """Simple diff application for well-known patterns."""
177 if diff.startswith("# SKELETON"):
178 return content
179 try:
180 result = subprocess.run(
181 ["patch", "-o", "-", "-"],
182 input=f"--- a/file\n+++ b/file\n{diff}".encode(),
183 capture_output=True,
184 timeout=10,
185 )
186 if result.returncode == 0:
187 return result.stdout.decode()
188 except Exception:
189 pass
190 return content
193# ── Auto Tester ─────────────────────────────────────────────────────
195class AutoTester:
196 """Run test suite to validate changes."""
198 def __init__(self, test_dir: str = "", pytest_args: str = ""):
199 self._test_dir = Path(test_dir) if test_dir else Path("tests")
200 self._pytest_args = pytest_args or "-x --tb=short -q"
202 async def run_tests(self) -> dict[str, Any]:
203 """Run the test suite.
205 Returns:
206 Dict with passed/failed/total counts and error details.
207 """
208 try:
209 result = subprocess.run(
210 ["python3", "-m", "pytest", str(self._test_dir)] + self._pytest_args.split(),
211 capture_output=True, text=True, timeout=120,
212 cwd=str(self._test_dir.parent) if self._test_dir.parent else None,
213 )
214 passed = "passed" in result.stdout.lower() or result.returncode == 0
215 return {
216 "passed": passed,
217 "total": self._parse_test_count(result.stdout),
218 "failures": result.returncode if not passed else 0,
219 "output": result.stdout[-1000:],
220 "duration": 0,
221 }
222 except FileNotFoundError:
223 return {"passed": True, "total": 0, "failures": 0, "output": "pytest not installed", "duration": 0}
224 except Exception as e:
225 return {"passed": False, "total": 0, "failures": 1, "output": str(e), "duration": 0}
227 def _parse_test_count(self, output: str) -> int:
228 """Parse test count from pytest output."""
229 for line in output.split("\n"):
230 if "passed" in line.lower():
231 try:
232 return int(line.strip().split()[0])
233 except (ValueError, IndexError):
234 pass
235 return 0
238# ── Rollback Manager ─────────────────────────────────────────────────
240class RollbackManager:
241 """Instant undo of any auto-applied change."""
243 def __init__(self, backup_dir: str = ""):
244 self._backup_dir = Path(backup_dir) if backup_dir else Path.home() / ".agentos" / "evolution" / "backups"
245 self._backup_dir.mkdir(parents=True, exist_ok=True)
246 self._history: list[dict[str, Any]] = []
248 def snapshot(self, file_path: str, content: str) -> str:
249 """Create a backup snapshot of a file before modification."""
250 snapshot_id = hashlib.sha256(f"{file_path}:{time.time()}".encode()).hexdigest()[:12]
251 snapshot_path = self._backup_dir / f"{snapshot_id}.bak"
252 snapshot_path.write_text(content, encoding="utf-8")
253 self._history.append({
254 "snapshot_id": snapshot_id,
255 "file_path": file_path,
256 "timestamp": time.time(),
257 "size": len(content),
258 })
259 return snapshot_id
261 def rollback(self, snapshot_id: str) -> bool:
262 """Restore file from snapshot."""
263 snapshot_path = self._backup_dir / f"{snapshot_id}.bak"
264 if not snapshot_path.exists():
265 return False
267 for entry in self._history:
268 if entry["snapshot_id"] == snapshot_id:
269 target = Path(entry["file_path"])
270 target.write_text(snapshot_path.read_text(encoding="utf-8"), encoding="utf-8")
271 return True
273 return False
275 def get_history(self, limit: int = 20) -> list[dict[str, Any]]:
276 """Get recent evolution history."""
277 return sorted(self._history, key=lambda x: x["timestamp"], reverse=True)[:limit]
280# ── A/B Evaluator ───────────────────────────────────────────────────
282class ABEvaluator:
283 """Compare agent performance before and after evolution changes."""
285 def __init__(self, test_cases: list[dict[str, str]] | None = None):
286 self._test_cases = test_cases or []
287 self._results_before: list[dict] = []
288 self._results_after: list[dict] = []
290 async def evaluate_before(self, agent) -> list[dict]:
291 """Run evaluation before changes."""
292 self._results_before = await self._run_eval_loop(agent)
293 return self._results_before
295 async def evaluate_after(self, agent) -> list[dict]:
296 """Run evaluation after changes."""
297 self._results_after = await self._run_eval_loop(agent)
298 return self._results_after
300 def compare(self) -> dict[str, Any]:
301 """Compare before/after results."""
302 if not self._results_before or not self._results_after:
303 return {"status": "no_data"}
305 before_success = sum(1 for r in self._results_before if r.get("passed", False))
306 after_success = sum(1 for r in self._results_after if r.get("passed", False))
307 total = max(len(self._results_before), len(self._results_after))
309 return {
310 "before_pass_rate": before_success / total if total else 0,
311 "after_pass_rate": after_success / total if total else 0,
312 "improvement": (after_success - before_success) / total if total else 0,
313 "regressions": after_success < before_success,
314 "total_cases": total,
315 }
317 async def _run_eval_loop(self, agent) -> list[dict]:
318 """Run evaluation loop."""
319 results = []
320 for case in self._test_cases:
321 try:
322 result = await agent.run(case.get("input", ""))
323 passed = case.get("expected", "") in str(result)
324 results.append({"case": case.get("id", ""), "passed": passed, "output": str(result)[:500]})
325 except Exception as e:
326 results.append({"case": case.get("id", ""), "passed": False, "error": str(e)})
327 return results
330# ── Evolution Journal ───────────────────────────────────────────────
332class EvolutionJournal:
333 """Complete audit trail of every evolution step."""
335 def __init__(self, journal_path: str = ""):
336 self._path = Path(journal_path) if journal_path else Path.home() / ".agentos" / "evolution" / "journal.jsonl"
337 self._path.parent.mkdir(parents=True, exist_ok=True)
339 def log(self, entry: dict[str, Any]):
340 """Append an entry to the journal."""
341 entry["_timestamp"] = datetime.now().isoformat()
342 with open(self._path, "a", encoding="utf-8") as f:
343 f.write(json.dumps(entry, ensure_ascii=False) + "\n")
345 def read(self, limit: int = 50) -> list[dict]:
346 """Read recent journal entries."""
347 if not self._path.exists():
348 return []
349 entries = []
350 with open(self._path, "r", encoding="utf-8") as f:
351 for line in f:
352 entries.append(json.loads(line))
353 return entries[-limit:]
355 def stats(self) -> dict[str, Any]:
356 """Compute evolution statistics from journal."""
357 entries = self.read(limit=10000)
358 if not entries:
359 return {}
361 by_type: dict[str, int] = {}
362 by_result: dict[str, int] = {}
363 for entry in entries:
364 by_type[entry.get("type", "unknown")] = by_type.get(entry.get("type", "unknown"), 0) + 1
365 by_result[entry.get("result", "unknown")] = by_result.get(entry.get("result", "unknown"), 0) + 1
367 return {
368 "total_entries": len(entries),
369 "by_type": by_type,
370 "by_result": by_result,
371 "first_entry": entries[0].get("_timestamp", ""),
372 "last_entry": entries[-1].get("_timestamp", ""),
373 }
376# ── AutoPilot ───────────────────────────────────────────────────────
378class AutoPilot:
379 """Closed-loop self-evolution engine.
381 The AutoPilot orchestrates the entire evolution pipeline:
382 signals → insights → proposals → code changes → tests → apply/rollback.
384 Usage:
385 from agentos.evolution import SignalCollector, EvolutionEngine, Learner, AutoPilot
387 collector = SignalCollector()
388 engine = EvolutionEngine()
389 learner = Learner(collector, engine)
390 autopilot = AutoPilot(engine, learner, mode=AutoPilotMode.CONFIDENCE_GATED)
392 # After accumulating signals...
393 run = await autopilot.evolve()
394 print(f"Evolved: {run.result} — {len(run.changes)} changes applied")
395 """
397 def __init__(
398 self,
399 engine: EvolutionEngine,
400 learner: Learner,
401 mode: AutoPilotMode = AutoPilotMode.CONFIDENCE_GATED,
402 confidence_threshold: float = 0.7,
403 code_generator: Optional[CodeGenerator] = None,
404 tester: Optional[AutoTester] = None,
405 rollback: Optional[RollbackManager] = None,
406 evaluator: Optional[ABEvaluator] = None,
407 journal: Optional[EvolutionJournal] = None,
408 ):
409 self.engine = engine
410 self.learner = learner
411 self.mode = mode
412 self.confidence_threshold = confidence_threshold
414 self.codegen = code_generator or CodeGenerator()
415 self.tester = tester or AutoTester()
416 self.rollback = rollback or RollbackManager()
417 self.evaluator = evaluator or ABEvaluator()
418 self.journal = journal or EvolutionJournal()
420 self._run_history: list[EvolutionRun] = []
422 async def evolve(self, agent=None) -> list[EvolutionRun]:
423 """Execute one complete evolution cycle.
425 1. Analyze signals → generate insights
426 2. Convert insights → proposals
427 3. For each proposal: generate code → test → apply/rollback
428 4. Journal everything
430 Args:
431 agent: Optional agent instance for A/B evaluation.
433 Returns:
434 List of EvolutionRun records for this cycle.
435 """
436 # Step 1: Analyze signals
437 insights = self.learner.analyze()
439 # Step 2: Generate proposals
440 proposals = []
441 for insight in insights:
442 proposal = self.learner.propose_from_insight(insight)
443 proposals.append(proposal)
445 # Step 3: Gate by confidence and mode
446 runs: list[EvolutionRun] = []
447 for proposal in proposals:
448 run = await self._process_proposal(proposal, agent)
449 runs.append(run)
450 self._run_history.append(run)
452 # Step 4: Journal results
453 self.journal.log({
454 "type": "evolution_cycle",
455 "insights": len(insights),
456 "proposals": len(proposals),
457 "applied": sum(1 for r in runs if r.result == ChangeResult.SUCCESS),
458 "failed": sum(1 for r in runs if r.result == ChangeResult.FAILED),
459 "rolled_back": sum(1 for r in runs if r.result == ChangeResult.ROLLED_BACK),
460 })
462 return runs
464 async def _process_proposal(self, proposal: EvolutionProposal, agent=None) -> EvolutionRun:
465 """Process a single proposal through the pipeline."""
466 run = EvolutionRun(
467 run_id=f"ev_{proposal.id}",
468 proposal=proposal,
469 started_at=time.time(),
470 )
472 # Check if we should proceed based on mode
473 if not self._should_proceed(proposal):
474 run.result = ChangeResult.SKIPPED
475 run.finished_at = time.time()
476 return run
478 # Snapshot before
479 for target in proposal.target_files:
480 if os.path.exists(target):
481 content = Path(target).read_text(encoding="utf-8")
482 self.rollback.snapshot(target, content)
484 # Evaluate before (if agent provided)
485 if agent:
486 run.metrics_before = await self.evaluator.compare()
488 # Generate code changes
489 codebase = {}
490 for target in proposal.target_files:
491 if os.path.exists(target):
492 codebase[target] = Path(target).read_text(encoding="utf-8")
494 changes = await self.codegen.generate(proposal, codebase)
496 if not changes:
497 run.result = ChangeResult.FAILED
498 run.finished_at = time.time()
499 return run
501 run.changes = changes
503 # Apply changes
504 for change in changes:
505 try:
506 if change.new_content:
507 Path(change.file_path).write_text(change.new_content, encoding="utf-8")
508 except Exception:
509 # Rollback and fail
510 for ch in run.changes:
511 self.rollback.rollback(ch.proposal_id)
512 run.result = ChangeResult.ROLLED_BACK
513 run.finished_at = time.time()
514 return run
516 # Run tests
517 test_results = await self.tester.run_tests()
519 if not test_results.get("passed", False):
520 # Regression detected — rollback
521 for change in changes:
522 if change.old_content:
523 Path(change.file_path).write_text(change.old_content, encoding="utf-8")
524 run.result = ChangeResult.REGRESSION
525 run.rollback_info = {"test_results": test_results}
526 run.finished_at = time.time()
527 return run
529 # Success!
530 run.result = ChangeResult.SUCCESS
531 run.test_results = test_results
533 # Evaluate after (if agent provided)
534 if agent:
535 run.metrics_after = await self.evaluator.compare()
537 # Update proposal status
538 proposal.status = EvolutionStatus.APPLIED
540 run.finished_at = time.time()
541 return run
543 def _should_proceed(self, proposal: EvolutionProposal) -> bool:
544 """Determine if we should proceed with this proposal based on mode."""
545 if self.mode == AutoPilotMode.SUGGEST_ONLY:
546 return False
548 if self.mode == AutoPilotMode.ASK_BEFORE:
549 # User interaction required — return False, caller must handle
550 return False
552 if self.mode == AutoPilotMode.CONFIDENCE_GATED:
553 return proposal.confidence >= self.confidence_threshold
555 if self.mode == AutoPilotMode.FULL_AUTO:
556 # Only safe changes in FULL_AUTO
557 return proposal.risk_level in ("low", "medium")
559 return False
561 def get_run_history(self, limit: int = 20) -> list[EvolutionRun]:
562 """Get recent evolution runs."""
563 return self._run_history[-limit:]
565 def get_stats(self) -> dict[str, Any]:
566 """Get AutoPilot statistics."""
567 runs = self._run_history
568 return {
569 "total_runs": len(runs),
570 "successful": sum(1 for r in runs if r.result == ChangeResult.SUCCESS),
571 "failed": sum(1 for r in runs if r.result == ChangeResult.FAILED),
572 "regressions": sum(1 for r in runs if r.result == ChangeResult.REGRESSION),
573 "rolled_back": sum(1 for r in runs if r.result == ChangeResult.ROLLED_BACK),
574 "total_changes": sum(len(r.changes) for r in runs),
575 "journal_stats": self.journal.stats(),
576 "rollback_history": len(self.rollback.get_history()),
577 }