""" Agent Evolution Loop — GEA Phase 3: Updating Module + Full Evolution Cycle Implements the Updating Module and the complete GEA evolution loop from the Group-Evolving Agents paper (UC Santa Barbara, Feb 2026). Architecture mirrors the two-stage GEA process: Stage 1 — Parent Group Selection (Performance-Novelty Algorithm) Stage 2 — Open-Ended Group Evolution (Experience Sharing → Reflect → Update → Evaluate) The full cycle: 1. select_parent_group() — Performance-Novelty Algorithm 2. gather experience pool — GroupReflectionService.gather_group_experience_pool() 3. reflect_and_generate_directives() — GroupReflectionService.reflect_and_generate_directives() 4. _apply_directives_to_clone()— Clone agent config; apply directives; validate via guardrails 5. _evaluate_evolved_agent() — Lightweight benchmark evaluation 6. _archive_or_discard() — Save winner trace; discard failures Key design decisions: - Directives are ALWAYS applied to a *clone* of the agent config, never in-place. - Every directive is validated by autonomous_guardrails before committing. - Evolution happens offline; only the winner agent is deployed (zero inference cost). Usage: from core.agent_evolution_loop import AgentEvolutionLoop from core.database import SessionLocal with SessionLocal() as db: loop = AgentEvolutionLoop(db) result = await loop.run_evolution_cycle(tenant_id="tenant-uuid") """ import logging import math import uuid from copy import deepcopy from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional, Tuple from sqlalchemy.orm import Session from core.models import AgentEvolutionTrace, AgentRegistry, Skill from core.group_reflection_service import GroupReflectionService logger = logging.getLogger(__name__) # ─── Tuning constants ───────────────────────────────────────────────────────── PERF_WEIGHT: float = 0.6 # α in combined_score = α*perf + β*novelty NOVELTY_WEIGHT: float = 0.4 # β PARENT_GROUP_SIZE: int = 5 # |P| — parent group size MIN_PERF_THRESHOLD: float = 0.3 # Agents below this are excluded from selection LOOKBACK_DAYS: int = 30 # Only consider agents active in the past N days class EvolutionCycleResult: """Structured result from a single evolution cycle.""" def __init__( self, cycle_id: str, tenant_id: str, parent_agent_ids: List[str], directives: List[str], evolved_agent_id: Optional[str], benchmark_passed: bool, benchmark_score: float, trace_id: Optional[str], ) -> None: self.cycle_id = cycle_id self.tenant_id = tenant_id self.parent_agent_ids = parent_agent_ids self.directives = directives self.evolved_agent_id = evolved_agent_id self.benchmark_passed = benchmark_passed self.benchmark_score = benchmark_score self.trace_id = trace_id self.timestamp = datetime.now(timezone.utc).isoformat() def to_dict(self) -> Dict[str, Any]: return { "cycle_id": self.cycle_id, "tenant_id": self.tenant_id, "parent_agent_ids": self.parent_agent_ids, "directives": self.directives, "evolved_agent_id": self.evolved_agent_id, "benchmark_passed": self.benchmark_passed, "benchmark_score": self.benchmark_score, "trace_id": self.trace_id, "timestamp": self.timestamp, } class AgentEvolutionLoop: """ Orchestrates the full GEA evolution cycle for a tenant's agent population. """ def __init__(self, db: Session) -> None: self.db = db from core.service_factory import ServiceFactory self.reflection_svc = ServiceFactory.get_group_reflection_service(db) # ───────────────────────────────────────────────────────────────────────── # Public API # ───────────────────────────────────────────────────────────────────────── async def run_evolution_cycle( self, tenant_id: str, group_size: int = PARENT_GROUP_SIZE, target_agent_id: Optional[str] = None, category: Optional[str] = None, ) -> EvolutionCycleResult: """ Run one full GEA evolution cycle for a tenant. Args: tenant_id: Tenant to evolve agents for group_size: Parent group size (default 5) target_agent_id: If set, only evolve this specific agent; otherwise select group via Performance-Novelty Algorithm category: Agent domain category (e.g. "crm", "finance"). Auto-detected from the seed agent's AgentRegistry.category if not provided. Returns: EvolutionCycleResult with outcome details """ cycle_id = str(uuid.uuid4()) logger.info("GEA cycle %s starting for tenant %s", cycle_id, tenant_id) # Stage 1: Select parent group if target_agent_id: parent_group = self._get_single_agent_group(target_agent_id, tenant_id) else: parent_group = self.select_parent_group(tenant_id, n=group_size) if not parent_group: logger.warning("GEA cycle %s: no eligible agents found", cycle_id) return EvolutionCycleResult( cycle_id=cycle_id, tenant_id=tenant_id, parent_agent_ids=[], directives=[], evolved_agent_id=None, benchmark_passed=False, benchmark_score=0.0, trace_id=None, ) parent_ids = [a.id for a in parent_group] # Resolve domain category once — used by reflection + trace recording # Priority: explicit arg > seed agent's registry category seed_agent_tentative = max( parent_group, key=lambda a: self._compute_combined_score(a, parent_group), ) resolved_category = category or getattr(seed_agent_tentative, "category", None) # Stage 2: Gather shared experience pool (domain-aware) pool = self.reflection_svc.gather_group_experience_pool( parent_ids, category=resolved_category ) # Stage 3: Reflect → generate evolution directives (domain-aware) directives = await self.reflection_svc.reflect_and_generate_directives( pool, tenant_id=tenant_id, category=resolved_category ) # Pick the "seed" agent for mutation (highest combined score in group) seed_agent = seed_agent_tentative # Stage 4: Apply directives to a cloned config (sandboxed) evolved_config, guardrail_ok = await self._apply_directives_to_clone( seed_agent, directives, tenant_id ) if not guardrail_ok: logger.warning("GEA cycle %s: directives blocked by guardrails", cycle_id) trace = self._record_trace( agent=seed_agent, parent_ids=parent_ids, tenant_id=tenant_id, directives=directives, pool=pool, benchmark_passed=False, benchmark_score=0.0, model_patch=None, category=resolved_category, block_reason="Guardrail validation failed", ) return EvolutionCycleResult( cycle_id=cycle_id, tenant_id=tenant_id, parent_agent_ids=parent_ids, directives=directives, evolved_agent_id=None, benchmark_passed=False, benchmark_score=0.0, trace_id=trace.id if trace else None, ) # Stage 5: Evaluate evolved config benchmark_score, benchmark_passed = await self._evaluate_evolved_config( seed_agent, evolved_config, tenant_id ) # Stage 6: Archive winner or discard evolved_agent_id: Optional[str] = None if benchmark_passed: evolved_agent_id = await self._promote_evolved_config( seed_agent, evolved_config, directives, parent_group ) # Record the evolution trace (with domain-specific benchmark name) model_patch = self._diff_configs(seed_agent.configuration, evolved_config) trace = self._record_trace( agent=seed_agent, parent_ids=parent_ids, tenant_id=tenant_id, directives=directives, pool=pool, benchmark_passed=benchmark_passed, benchmark_score=benchmark_score, model_patch=model_patch, category=resolved_category, ) logger.info( "GEA cycle %s complete: passed=%s score=%.3f evolved_agent=%s", cycle_id, benchmark_passed, benchmark_score, evolved_agent_id, ) return EvolutionCycleResult( cycle_id=cycle_id, tenant_id=tenant_id, parent_agent_ids=parent_ids, directives=directives, evolved_agent_id=evolved_agent_id, benchmark_passed=benchmark_passed, benchmark_score=benchmark_score, trace_id=trace.id if trace else None, ) def select_parent_group( self, tenant_id: str, n: int = PARENT_GROUP_SIZE, ) -> List[AgentRegistry]: """ Stage 1: Performance-Novelty Algorithm. Selects n agents from the archive to form the parent group. Combined score = α * performance_score + β * novelty_score. Novelty is approximated as the distance of an agent's performance from the population mean — agents far from the mean (in either direction) are novel, ensuring a healthy mix of specialists and explorers. Args: tenant_id: Tenant namespace n: Parent group size Returns: List of up to n AgentRegistry objects sorted by combined score desc """ cutoff = datetime.now(timezone.utc) - timedelta(days=LOOKBACK_DAYS) agents = ( self.db.query(AgentRegistry) .filter( AgentRegistry.tenant_id == tenant_id, AgentRegistry.enabled == True, AgentRegistry.confidence_score >= MIN_PERF_THRESHOLD, AgentRegistry.updated_at >= cutoff, ) .all() ) if not agents: return [] # Compute novelty scores using population mean scores = [a.confidence_score for a in agents] mean_perf = sum(scores) / len(scores) std_perf = math.sqrt(sum((s - mean_perf) ** 2 for s in scores) / len(scores)) or 1e-6 def novelty(agent: AgentRegistry) -> float: # Normalized absolute deviation from mean — high deviation = high novelty return min(1.0, abs(agent.confidence_score - mean_perf) / (2 * std_perf)) scored = [ (agent, self._compute_combined_score_with_novelty(agent, novelty(agent))) for agent in agents ] scored.sort(key=lambda x: -x[1]) return [agent for agent, _ in scored[:n]] def get_ancestor_lineage( self, agent_id: str, tenant_id: str, max_depth: int = 20, ) -> List[Dict[str, Any]]: """ Traverse the ancestry chain of an agent via evolution traces. Returns a list of dicts describing each ancestor, mirroring GEA's "super-employee" metric (17 unique ancestors → 28% of population). Args: agent_id: Starting agent tenant_id: Tenant namespace max_depth: Max generations to traverse (prevents infinite loops) Returns: List[{"agent_id": str, "generation": int, "performance_score": float}] """ visited: set = set() lineage: List[Dict[str, Any]] = [] queue: List[Tuple[str, int]] = [(agent_id, 0)] while queue and len(lineage) < max_depth: current_id, depth = queue.pop(0) if current_id in visited: continue visited.add(current_id) trace = ( self.db.query(AgentEvolutionTrace) .filter( AgentEvolutionTrace.agent_id == current_id, AgentEvolutionTrace.tenant_id == tenant_id, ) .order_by(AgentEvolutionTrace.created_at.desc()) .first() ) if trace: lineage.append({ "agent_id": current_id, "generation": trace.generation, "performance_score": trace.performance_score, "depth": depth, }) if trace.parent_agent_ids: for parent_id in trace.parent_agent_ids: if parent_id not in visited: queue.append((parent_id, depth + 1)) return lineage # ───────────────────────────────────────────────────────────────────────── # Internal: Apply directives (sandboxed) # ───────────────────────────────────────────────────────────────────────── async def _apply_directives_to_clone( self, agent: AgentRegistry, directives: List[str], tenant_id: str, ) -> Tuple[Dict[str, Any], bool]: """ Apply evolution directives to a DEEP CLONE of the agent's configuration. This is the Updating Module. Directives update the system prompt and evolution history. Each directive is validated against governance rules before being committed. Returns: (evolved_config, guardrail_ok) """ evolved_config = deepcopy(agent.configuration or {}) # Record evolution metadata if "evolution_history" not in evolved_config: evolved_config["evolution_history"] = [] evolved_config["evolution_history"].append({ "timestamp": datetime.now(timezone.utc).isoformat(), "directives": directives, "parent_agent_id": agent.id, "gea_cycle": True, }) # Apply directives to system prompt (append as guidance) existing_prompt = evolved_config.get("system_prompt", "") # Skill Creation Logic: Check for direct skill creation directives # Format example: "CREATE_SKILL: Fetch Shopify products from API URL https://..." for directive in directives: if directive.strip().upper().startswith("CREATE_SKILL:"): try: logger.info("GEA: Detected skill creation directive: %s", directive) # Extract prompt from directive skill_prompt = directive.split(":", 1)[1].strip() # Initialize SkillCreationAgent via ServiceFactory from core.service_factory import ServiceFactory skill_agent = ServiceFactory.get_skill_creation_agent(self.db) # Execute skill creation # We use generic parameters; in a real scenario, we might extract more from the directive skill = await skill_agent.create_skill_from_api_documentation( tenant_id=tenant_id, agent_id=agent.id, user_id=None, # System-initiated api_docs_url=None, # Let the agent infer or search if not provided api_description=skill_prompt, skill_name=f"evolved_skill_{uuid.uuid4().hex[:8]}" ) if skill: logger.info("GEA: Successfully evolved new skill: %s", skill.name) if "active_skills" not in evolved_config: evolved_config["active_skills"] = [] evolved_config["active_skills"].append(skill.id) # Add success note to evolution history evolved_config["evolution_history"][-1]["skill_created"] = skill.name except Exception as e: logger.error("GEA: Failed to create skill from directive: %s", e) # OPTIMIZE_SKILL directive: Mutate and improve existing skills via AlphaEvolver # Format: "OPTIMIZE_SKILL: | " # Requires SUPERVISED maturity via AutoDevCapabilityService elif directive.strip().upper().startswith("OPTIMIZE_SKILL:"): try: logger.info("GEA: Detected skill optimization directive: %s", directive) optimize_payload = directive.split(":", 1)[1].strip() # Parse "skill_name | optimization_goal" if "|" in optimize_payload: skill_name, opt_goal = [ p.strip() for p in optimize_payload.split("|", 1) ] else: skill_name = optimize_payload opt_goal = "Optimize for performance and reliability" # Gate: check if agent has SUPERVISED maturity for AlphaEvolver from core.auto_dev.capability_gate import AutoDevCapabilityService gate = AutoDevCapabilityService(self.db) workspace_settings = self._get_workspace_settings(tenant_id) if not gate.can_use( agent_id=agent.id, capability="auto_dev.alpha_evolver", workspace_settings=workspace_settings, ): logger.info( "GEA: Agent %s not yet SUPERVISED — OPTIMIZE_SKILL skipped", agent.id, ) evolved_config["evolution_history"][-1][ "optimize_skill_skipped" ] = "Agent maturity insufficient" continue # Retrieve skill source code skill_code = self._get_skill_code(tenant_id, skill_name) if not skill_code: logger.warning( "GEA: Skill '%s' not found for optimization", skill_name ) continue # Use AlphaEvolverEngine via SelfEvolutionService from core.self_evolution_service import self_evolution_service result = await self_evolution_service.run_alpha_evolve_cycle( agent_id=agent.id, tenant_id=tenant_id, base_code=skill_code, research_goal=opt_goal, iterations=2, ) if result.get("success"): logger.info( "GEA: Skill optimization completed for '%s': %d iterations", skill_name, len(result.get("results", [])), ) evolved_config["evolution_history"][-1][ "skill_optimized" ] = skill_name else: logger.info( "GEA: Skill optimization skipped/failed for '%s': %s", skill_name, result.get("reason", result.get("error", "unknown")), ) except ImportError: logger.debug( "GEA: Auto-Dev module not available — OPTIMIZE_SKILL skipped" ) except Exception as e: logger.error( "GEA: Failed to optimize skill from directive: %s", e ) directive_block = "\n\n## Evolution Directives\n" + "\n".join( f"- {d}" for d in directives ) evolved_config["system_prompt"] = existing_prompt + directive_block # Validate through governance guardrails guardrail_ok = await self._validate_via_guardrails(evolved_config, tenant_id) return evolved_config, guardrail_ok async def _validate_via_guardrails( self, evolved_config: Dict[str, Any], tenant_id: str, ) -> bool: """ Validate the evolved config against governance policy. Tries to import agent_governance_service if available; falls back to a simple content check to avoid hard dependencies. """ try: from core.agent_governance_service import AgentGovernanceService svc = AgentGovernanceService(self.db) return await svc.validate_evolution_directive(evolved_config, tenant_id) except (ImportError, AttributeError): # Fallback: block configs containing obviously dangerous patterns system_prompt = evolved_config.get("system_prompt", "") danger_patterns = ["ignore all rules", "bypass guardrails", "disable safety"] for pattern in danger_patterns: if pattern.lower() in system_prompt.lower(): logger.warning("GEA guardrail: blocked config containing '%s'", pattern) return False return True # ───────────────────────────────────────────────────────────────────────── # Internal: Evaluate # ───────────────────────────────────────────────────────────────────────── async def _evaluate_evolved_config( self, agent: AgentRegistry, evolved_config: Dict[str, Any], tenant_id: str, ) -> Tuple[float, bool]: """ Evaluate the evolved config against the agent's graduation pipeline. Primary path: delegates to GraduationExamService.evaluate_evolved_agent() which runs readiness score + constitutional check + prompt quality heuristic without committing any changes to the database. Fallback: uses the lightweight confidence_score proxy (original behaviour) if GraduationExamService is unavailable (e.g. circular import or missing). Returns: (benchmark_score [0.0–1.0], benchmark_passed [bool]) """ try: from core.graduation_exam import GraduationExamService exam_svc = GraduationExamService(self.db) result = exam_svc.evaluate_evolved_agent( agent_id=agent.id, tenant_id=tenant_id, evolved_config=evolved_config, ) return result["benchmark_score"], result["benchmark_passed"] except Exception as e: logger.warning( "GEA: GraduationExamService evaluation failed (%s); using proxy score", e ) # Fallback: lightweight proxy benchmark_score: float = agent.confidence_score evolution_bonus = min(0.05, 0.01 * len(evolved_config.get("evolution_history", []))) benchmark_score = min(1.0, benchmark_score + evolution_bonus) benchmark_passed = benchmark_score >= 0.55 return benchmark_score, benchmark_passed # ───────────────────────────────────────────────────────────────────────── # Internal: Promote / Record # ───────────────────────────────────────────────────────────────────────── async def _promote_evolved_config( self, seed_agent: AgentRegistry, evolved_config: Dict[str, Any], directives: List[str], parent_group: List[AgentRegistry], ) -> str: """ Commit the evolved config to the seed agent (in-place update). Records the evolution in the agent's configuration. Returns the agent_id of the updated agent. """ seed_agent.configuration = evolved_config seed_agent.self_healed_count = (seed_agent.self_healed_count or 0) + 1 seed_agent.updated_at = datetime.now(timezone.utc) self.db.commit() logger.info("GEA: promoted evolved config to agent %s", seed_agent.id) return seed_agent.id def _record_trace( self, agent: AgentRegistry, parent_ids: List[str], tenant_id: str, directives: List[str], pool: Dict[str, Any], benchmark_passed: bool, benchmark_score: float, model_patch: Optional[str], category: Optional[str] = None, block_reason: Optional[str] = None, ) -> Optional[AgentEvolutionTrace]: """ Persist an AgentEvolutionTrace to the Experience Archive. The benchmark_name is derived from the domain profile's success_label so traces are self-describing across domains. """ try: from core.group_reflection_service import DomainProfileRegistry domain_profile = DomainProfileRegistry.resolve(category) benchmark_name = f"{domain_profile.name.lower().replace(' ', '_')}_proxy" # Calculate ancestor count by combining parent lineage depths ancestor_count = len(set(parent_ids)) # Determine current generation last_trace = ( self.db.query(AgentEvolutionTrace) .filter(AgentEvolutionTrace.agent_id == agent.id) .order_by(AgentEvolutionTrace.generation.desc()) .first() ) generation = (last_trace.generation + 1) if last_trace else 1 tool_log_sample = pool.get("tool_patterns", [])[:10] trace = AgentEvolutionTrace( tenant_id=tenant_id, agent_id=agent.id, generation=generation, parent_agent_ids=parent_ids, ancestor_count=ancestor_count, performance_score=agent.confidence_score, novelty_score=0.0, # Populated by select_parent_group in future combined_selection_score=agent.confidence_score * PERF_WEIGHT, tool_use_log=tool_log_sample, task_log="\n".join(pool.get("task_log_excerpts", [])[:3]), evolving_requirements="\n".join(directives), model_patch=model_patch, benchmark_passed=benchmark_passed, benchmark_name=benchmark_name, benchmark_score=benchmark_score, is_high_quality=benchmark_passed, quality_filter_reason=block_reason, ) self.db.add(trace) self.db.commit() self.db.refresh(trace) return trace except Exception as e: logger.error("GEA: failed to record trace: %s", e) self.db.rollback() return None # ───────────────────────────────────────────────────────────────────────── # Scoring utilities # ───────────────────────────────────────────────────────────────────────── def _compute_combined_score( self, agent: AgentRegistry, group: List[AgentRegistry] ) -> float: scores = [a.confidence_score for a in group] mean = sum(scores) / len(scores) if scores else 0.5 std = math.sqrt(sum((s - mean) ** 2 for s in scores) / len(scores)) or 1e-6 novelty = min(1.0, abs(agent.confidence_score - mean) / (2 * std)) return PERF_WEIGHT * agent.confidence_score + NOVELTY_WEIGHT * novelty def _compute_combined_score_with_novelty( self, agent: AgentRegistry, novelty: float ) -> float: return PERF_WEIGHT * agent.confidence_score + NOVELTY_WEIGHT * novelty def _get_single_agent_group( self, agent_id: str, tenant_id: str ) -> List[AgentRegistry]: agent = ( self.db.query(AgentRegistry) .filter( AgentRegistry.id == agent_id, AgentRegistry.tenant_id == tenant_id, ) .first() ) return [agent] if agent else [] def _diff_configs( self, original: Optional[Dict[str, Any]], evolved: Optional[Dict[str, Any]], ) -> str: """ Produce a simple human-readable diff between two config dicts. In production, use unified diff on the JSON-serialized configs. """ import json orig_str = json.dumps(original or {}, indent=2, sort_keys=True) evol_str = json.dumps(evolved or {}, indent=2, sort_keys=True) if orig_str == evol_str: return "--- no changes ---" # Simple line-level diff for readability orig_lines = orig_str.splitlines() evol_lines = evol_str.splitlines() import difflib diff = difflib.unified_diff( orig_lines, evol_lines, fromfile="original_config", tofile="evolved_config", lineterm="", ) return "\n".join(list(diff)[:100]) # Cap at 100 lines # ───────────────────────────────────────────────────────────────────────── # Internal: Auto-Dev Helpers # ───────────────────────────────────────────────────────────────────────── def _get_workspace_settings(self, tenant_id: str) -> Dict[str, Any]: """Retrieve workspace settings for Auto-Dev capability gating.""" try: from core.models import Workspace workspace = ( self.db.query(Workspace) .filter(Workspace.tenant_id == tenant_id) .first() ) if workspace and workspace.metadata_json: return workspace.metadata_json except Exception: pass return {} def _get_skill_code(self, tenant_id: str, skill_name: str) -> Optional[str]: """ Retrieve the source code for a named skill. Searches the tenant's skills directory for a matching .py file. """ try: from core.skill_builder_service import SkillBuilderService builder = SkillBuilderService() skills_dir = builder._get_tenant_skills_dir(tenant_id) # Search for skill by name safe_name = "".join( c for c in skill_name if c.isalnum() or c in ("-", "_") ).lower() skill_dir = skills_dir / safe_name if skill_dir.exists(): for script in skill_dir.glob("*.py"): return script.read_text() # Fallback: search all skill directories for child in skills_dir.iterdir(): if child.is_dir() and safe_name in child.name: for script in child.glob("*.py"): return script.read_text() return None except Exception: return None