Auto-Dev API Reference
**Version:** 1.0
**Last Updated:** 2026-04-10
Complete API reference for Auto-Dev self-evolving agent system.
---
Table of Contents
---
Overview
Auto-Dev provides a modular, extensible framework for building self-evolving agents. The API is organized around several core components:
- **BaseLearningEngine**: Abstract base for all learning engines
- **MementoEngine**: Generates new skill candidates from experiences
- **AlphaEvolverEngine**: Optimizes and refines existing skills
- **ReflectionEngine**: Detects patterns and suggests improvements
- **FitnessService**: Evaluates skill performance and quality
- **CapabilityGateService**: Validates improvements before deployment
- **EventBus**: Event system for agent lifecycle events
---
Core Classes
BaseLearningEngine
**Location:** core.auto_dev.base_engine
**Purpose:** Abstract base class for all learning engines. Provides common functionality for episode analysis, skill generation, and deployment.
**Class Definition:**
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from datetime import datetime
class BaseLearningEngine(ABC):
"""Abstract base for learning engines."""
def __init__(
self,
db: Session,
tenant_id: str,
agent_id: str,
config: Optional[Dict[str, Any]] = None
):
"""
Initialize learning engine.
Args:
db: Database session
tenant_id: Tenant identifier
agent_id: Agent identifier
config: Optional configuration dictionary
"""
self.db = db
self.tenant_id = tenant_id
self.agent_id = agent_id
self.config = config or {}
@abstractmethod
async def analyze(
self,
episodes: List[Episode]
) -> LearningInsights:
"""
Analyze episodes and extract insights.
Args:
episodes: List of episodes to analyze
Returns:
LearningInsights with patterns and discoveries
"""
pass
@abstractmethod
async def generate_candidates(
self,
insights: LearningInsights
) -> List[SkillCandidate]:
"""
Generate skill candidates from insights.
Args:
insights: Learning insights from analysis
Returns:
List of skill candidates
"""
pass
@abstractmethod
async def evaluate_fitness(
self,
candidate: SkillCandidate,
test_episodes: List[Episode]
) -> FitnessScore:
"""
Evaluate fitness of skill candidate.
Args:
candidate: Skill candidate to evaluate
test_episodes: Episodes for testing
Returns:
Fitness score (0.0 to 1.0)
"""
pass
async def evolve(
self,
episode_window: int = 100,
min_fitness: float = 0.8
) -> EvolutionResult:
"""
Run evolution cycle.
Args:
episode_window: Number of episodes to analyze
min_fitness: Minimum fitness for deployment
Returns:
EvolutionResult with deployed skills and metrics
"""
# Fetch recent episodes
episodes = await self._fetch_episodes(episode_window)
# Analyze episodes
insights = await self.analyze(episodes)
# Generate candidates
candidates = await self.generate_candidates(insights)
# Evaluate and deploy
deployed = []
for candidate in candidates:
fitness = await self.evaluate_fitness(candidate, episodes)
if fitness.score >= min_fitness:
deployed.append(await self._deploy(candidate, fitness))
return EvolutionResult(
engine_name=self.__class__.__name__,
episodes_analyzed=len(episodes),
candidates_generated=len(candidates),
skills_deployed=len(deployed),
deployed_skills=deployed
)**Methods:**
| Method | Purpose | Returns |
|---|---|---|
analyze() | Extract patterns from episodes | LearningInsights |
generate_candidates() | Create skill candidates | List[SkillCandidate] |
evaluate_fitness() | Test candidate performance | FitnessScore |
evolve() | Run full evolution cycle | EvolutionResult |
---
MementoEngine
**Location:** core.auto_dev.memento_engine
**Purpose:** Generates new skill candidates from successful experiences. "Mementos" are memorable patterns that can be reused.
**Class Definition:**
class MementoEngine(BaseLearningEngine):
"""Generates skills from successful patterns."""
def __init__(
self,
db: Session,
tenant_id: str,
agent_id: str,
config: Optional[Dict[str, Any]] = None
):
super().__init__(db, tenant_id, agent_id, config)
self.pattern_miner = PatternMiner()
self.skill_generator = SkillGenerator()
async def analyze(
self,
episodes: List[Episode]
) -> LearningInsights:
"""
Analyze episodes for successful patterns.
Identifies:
- Repeated action sequences
- High-success patterns
- Parameter correlations
- Context patterns
"""
# Filter successful episodes
successful = [e for e in episodes if e.success]
# Mine patterns
patterns = await self.pattern_miner.mine(successful)
# Rank patterns
ranked = self._rank_patterns(patterns, episodes)
return LearningInsights(
engine_type="memento",
patterns_discovered=len(patterns),
insights=ranked,
confidence=self._calculate_confidence(ranked)
)
async def generate_candidates(
self,
insights: LearningInsights
) -> List[SkillCandidate]:
"""
Generate skill candidates from patterns.
Creates:
- New skills from patterns
- Parameterized templates
- Context-aware skills
"""
candidates = []
for pattern in insights.insights:
# Generate skill code
skill_code = await self.skill_generator.generate(
pattern=pattern,
agent_id=self.agent_id
)
# Create candidate
candidate = SkillCandidate(
engine_type="memento",
pattern_id=pattern.id,
code=skill_code.code,
parameters=skill_code.parameters,
description=pattern.description,
confidence=pattern.confidence
)
candidates.append(candidate)
return candidates
async def evaluate_fitness(
self,
candidate: SkillCandidate,
test_episodes: List[Episode]
) -> FitnessScore:
"""
Test candidate on historical episodes.
Evaluates:
- Success rate on relevant episodes
- Execution efficiency
- Output quality
- Consistency
"""
# Find relevant test episodes
relevant = self._find_relevant_episodes(
candidate,
test_episodes
)
# Test on episodes
results = []
for episode in relevant:
result = await self._test_on_episode(
candidate,
episode
)
results.append(result)
# Calculate fitness
success_rate = sum(r.success for r in results) / len(results)
efficiency = self._calculate_efficiency(results)
quality = self._calculate_quality(results)
consistency = self._calculate_consistency(results)
return FitnessScore(
score=self._calculate_fitness(
success_rate, efficiency, quality, consistency
),
success_rate=success_rate,
efficiency=efficiency,
quality=quality,
consistency=consistency,
test_count=len(results)
)**Configuration:**
config = {
"min_pattern_frequency": 5, # Min occurrences for pattern
"min_success_rate": 0.7, # Min success rate for pattern
"pattern_similarity_threshold": 0.8, # Pattern clustering threshold
"max_candidates_per_cycle": 10, # Max skills to generate
"parameter_learning": True # Learn parameters from patterns
}---
AlphaEvolverEngine
**Location:** core.auto_dev.alpha_evolver_engine
**Purpose:** Optimizes and refines existing skills through iterative improvement.
**Class Definition:**
class AlphaEvolverEngine(BaseLearningEngine):
"""Evolves and optimizes existing skills."""
def __init__(
self,
db: Session,
tenant_id: str,
agent_id: str,
config: Optional[Dict[str, Any]] = None
):
super().__init__(db, tenant_id, agent_id, config)
self.mutator = SkillMutator()
self.selector = SkillSelector()
async def analyze(
self,
episodes: List[Episode]
) -> LearningInsights:
"""
Analyze skill performance and identify improvement opportunities.
Identifies:
- Underperforming skills
- Parameter optimization opportunities
- Skill combination opportunities
- Failure patterns
"""
# Get current skills
current_skills = await self._get_current_skills()
# Analyze skill performance
skill_metrics = {}
for skill in current_skills:
metrics = await self._analyze_skill_performance(
skill,
episodes
)
skill_metrics[skill.id] = metrics
# Identify improvement opportunities
opportunities = await self._identify_opportunities(
current_skills,
skill_metrics,
episodes
)
return LearningInsights(
engine_type="alpha_evolver",
skills_analyzed=len(current_skills),
insights=opportunities,
confidence=self._calculate_confidence(opportunities)
)
async def generate_candidates(
self,
insights: LearningInsights
) -> List[SkillCandidate]:
"""
Generate optimized skill variants.
Creates:
- Parameter-optimized versions
- Logic-refined versions
- Combined skill variants
"""
candidates = []
for opportunity in insights.insights:
if opportunity.type == "parameter_optimization":
# Optimize parameters
optimized = await self.mutator.optimize_parameters(
skill=opportunity.skill,
episodes=opportunity.episodes
)
candidate = SkillCandidate(
engine_type="alpha_evolver",
parent_skill_id=opportunity.skill.id,
code=optimized.code,
parameters=optimized.parameters,
description=f"Optimized {opportunity.skill.name}",
confidence=optimized.confidence
)
elif opportunity.type == "logic_refinement":
# Refine logic
refined = await self.mutator.refine_logic(
skill=opportunity.skill,
patterns=opportunity.patterns
)
candidate = SkillCandidate(
engine_type="alpha_evolver",
parent_skill_id=opportunity.skill.id,
code=refined.code,
parameters=refined.parameters,
description=f"Refined {opportunity.skill.name}",
confidence=refined.confidence
)
candidates.append(candidate)
return candidates**Configuration:**
config = {
"optimization_iterations": 100, # Max optimization iterations
"mutation_rate": 0.1, # Probability of mutation
"crossover_rate": 0.7, # Probability of crossover
"elitism_count": 2, # Number of elite candidates
"population_size": 20, # Population size for evolution
"convergence_threshold": 0.001 # Convergence detection
}---
ReflectionEngine
**Location:** core.auto_dev.reflection_engine
**Purpose:** Detects patterns and suggests improvements through meta-cognitive analysis.
**Class Definition:**
class ReflectionEngine(BaseLearningEngine):
"""Reflective analysis for pattern detection."""
def __init__(
self,
db: Session,
tenant_id: str,
agent_id: str,
config: Optional[Dict[str, Any]] = None
):
super().__init__(db, tenant_id, agent_id, config)
self.pattern_detector = PatternDetector()
self.advisor = AdvisorService()
async def analyze(
self,
episodes: List[Episode]
) -> LearningInsights:
"""
Perform reflective analysis on episodes.
Detects:
- Recurring failure patterns
- Context dependencies
- Skill interaction patterns
- Performance trends
"""
# Detect patterns
patterns = await self.pattern_detector.detect(episodes)
# Classify patterns
classifications = await self._classify_patterns(patterns)
# Generate recommendations
recommendations = await self.advisor.generate_recommendations(
patterns,
classifications
)
return LearningInsights(
engine_type="reflection",
patterns_discovered=len(patterns),
insights=recommendations,
confidence=self._calculate_confidence(patterns)
)
async def generate_candidates(
self,
insights: LearningInsights
) -> List[SkillCandidate]:
"""
Generate skill candidates from recommendations.
Creates:
- New skills for detected patterns
- Mitigation skills for failure patterns
- Optimization skills for performance
"""
candidates = []
for recommendation in insights.insights:
if recommendation.type == "new_skill":
# Generate new skill
skill_code = await self._generate_skill_from_recommendation(
recommendation
)
candidate = SkillCandidate(
engine_type="reflection",
pattern_id=recommendation.pattern_id,
code=skill_code.code,
parameters=skill_code.parameters,
description=recommendation.description,
confidence=recommendation.confidence
)
elif recommendation.type == "mitigation":
# Generate mitigation skill
mitigation = await self._generate_mitigation_skill(
recommendation
)
candidate = SkillCandidate(
engine_type="reflection",
pattern_id=recommendation.pattern_id,
code=mitigation.code,
parameters=mitigation.parameters,
description=f"Mitigation: {recommendation.description}",
confidence=recommendation.confidence
)
candidates.append(candidate)
return candidates**Configuration:**
config = {
"min_pattern_occurrences": 3, # Min occurrences for pattern
"pattern_window": 50, # Episode window for pattern detection
"trend_analysis_window": 100, # Window for trend analysis
"confidence_threshold": 0.7, # Min confidence for recommendations
"max_recommendations": 15 # Max recommendations per cycle
}---
FitnessService
**Location:** core.auto_dev.fitness_service
**Purpose:** Evaluates skill performance and quality across multiple dimensions.
**Class Definition:**
class FitnessService:
"""Evaluates skill fitness and performance."""
def __init__(
self,
db: Session,
tenant_id: str,
config: Optional[Dict[str, Any]] = None
):
self.db = db
self.tenant_id = tenant_id
self.config = config or {}
async def evaluate_fitness(
self,
candidate: SkillCandidate,
test_episodes: List[Episode]
) -> FitnessScore:
"""
Comprehensive fitness evaluation.
Dimensions:
- Success Rate: 40% weight
- Efficiency: 30% weight
- Quality: 20% weight
- Consistency: 10% weight
"""
# Test on episodes
results = await self._test_candidate(candidate, test_episodes)
# Calculate dimensions
success_rate = self._calculate_success_rate(results)
efficiency = self._calculate_efficiency(results)
quality = self._calculate_quality(results)
consistency = self._calculate_consistency(results)
# Calculate overall fitness
fitness = (
success_rate * 0.4 +
efficiency * 0.3 +
quality * 0.2 +
consistency * 0.1
)
return FitnessScore(
score=fitness,
success_rate=success_rate,
efficiency=efficiency,
quality=quality,
consistency=consistency,
test_count=len(results),
details=self._generate_details(results)
)
async def compare_fitness(
self,
candidate_a: SkillCandidate,
candidate_b: SkillCandidate,
test_episodes: List[Episode]
) -> FitnessComparison:
"""
Compare two skill candidates.
Returns:
FitnessComparison with relative performance
"""
fitness_a = await self.evaluate_fitness(candidate_a, test_episodes)
fitness_b = await self.evaluate_fitness(candidate_b, test_episodes)
return FitnessComparison(
candidate_a_id=candidate_a.id,
candidate_b_id=candidate_b.id,
winner=candidate_a.id if fitness_a.score > fitness_b.score else candidate_b.id,
score_difference=abs(fitness_a.score - fitness_b.score),
comparison={
"success_rate": fitness_a.success_rate - fitness_b.success_rate,
"efficiency": fitness_a.efficiency - fitness_b.efficiency,
"quality": fitness_a.quality - fitness_b.quality,
"consistency": fitness_a.consistency - fitness_b.consistency
}
)---
CapabilityGateService
**Location:** core.auto_dev.capability_gate
**Purpose:** Validates skill improvements before deployment.
**Class Definition:**
class CapabilityGateService:
"""Validates and gates skill deployments."""
def __init__(
self,
db: Session,
tenant_id: str,
agent_id: str,
gate_level: str = "standard"
):
self.db = db
self.tenant_id = tenant_id
self.agent_id = agent_id
self.gate_level = gate_level
self.thresholds = self._get_thresholds(gate_level)
async def validate_deployment(
self,
candidate: SkillCandidate,
fitness: FitnessScore,
test_episodes: List[Episode]
) -> GateResult:
"""
Validate if candidate can be deployed.
Checks:
1. Performance threshold
2. Safety validation
3. Testing requirements
4. Gradual rollout
"""
checks = []
# Check 1: Performance threshold
performance_check = await self._check_performance_threshold(
candidate, fitness
)
checks.append(performance_check)
# Check 2: Safety validation
safety_check = await self._check_safety(candidate)
checks.append(safety_check)
# Check 3: Testing requirements
testing_check = await self._check_testing_requirements(
candidate, test_episodes
)
checks.append(testing_check)
# Check 4: Gradual rollout
rollout_check = await self._check_gradual_rollout(
candidate, fitness
)
checks.append(rollout_check)
# Overall decision
all_passed = all(check.passed for check in checks)
return GateResult(
passed=all_passed,
gate_level=self.gate_level,
checks=checks,
reason=self._generate_reason(checks)
)
async def deploy_with_gate(
self,
candidate: SkillCandidate,
fitness: FitnessScore,
test_episodes: List[Episode]
) -> DeploymentResult:
"""
Deploy candidate with gate validation.
Process:
1. Validate deployment
2. If passed, gradual rollout
3. Monitor during rollout
4. Full deployment or rollback
"""
# Validate
gate_result = await self.validate_deployment(
candidate, fitness, test_episodes
)
if not gate_result.passed:
return DeploymentResult(
deployed=False,
reason="Gate validation failed",
gate_result=gate_result
)
# Gradual rollout
rollout_result = await self._gradual_rollout(
candidate, fitness
)
if rollout_result.success:
# Full deployment
await self._full_deployment(candidate)
return DeploymentResult(
deployed=True,
rollout_percentage=100,
gate_result=gate_result
)
else:
# Rollback
await self._rollback(candidate)
return DeploymentResult(
deployed=False,
reason="Rollout failed, rolled back",
gate_result=gate_result,
rollout_result=rollout_result
)**Gate Levels:**
| Level | Fitness Threshold | Testing | Rollout |
|---|---|---|---|
| **Conservative** | 90% | 200 episodes | Manual |
| **Standard** | 80% | 100 episodes | Gradual |
| **Aggressive** | 70% | 50 episodes | Fast |
| **Disabled** | N/A | None | Immediate |
---
Event System
EventBus
**Location:** core.auto_dev.event_hooks
**Purpose:** Event system for agent lifecycle events.
**Event Types:**
class EventType(Enum):
"""Auto-Dev event types."""
EPISODE_COMPLETED = "episode.completed"
SKILL_EXECUTED = "skill.executed"
AGENT_ACTION = "agent.action"
EVOLUTION_CYCLE_STARTED = "evolution.cycle_started"
EVOLUTION_CYCLE_COMPLETED = "evolution.cycle_completed"
SKILL_DEPLOYED = "skill.deployed"
SKILL_ROLLEDBACK = "skill.rollback"
GATE_VIOLATION = "gate.violation"**Subscribing to Events:**
from core.auto_dev.event_hooks import EventBus
# Initialize event bus
event_bus = EventBus(db, tenant_id)
# Subscribe to events
async def on_episode_completed(event: Event):
"""Handle episode completion."""
print(f"Episode {event.episode_id} completed")
print(f"Success: {event.success}")
print(f"Duration: {event.duration}s")
# Subscribe
subscription_id = await event_bus.subscribe(
event_type=EventType.EPISODE_COMPLETED,
handler=on_episode_completed,
filter_params={"agent_id": "my-agent"}
)**Event Payloads:**
# Episode completed event
{
"event_type": "episode.completed",
"tenant_id": "tenant-uuid",
"agent_id": "agent-uuid",
"episode_id": "episode-uuid",
"success": true,
"duration": 15.3,
"timestamp": "2026-04-10T12:00:00Z"
}
# Skill deployed event
{
"event_type": "skill.deployed",
"tenant_id": "tenant-uuid",
"agent_id": "agent-uuid",
"skill_id": "skill-uuid",
"candidate_id": "candidate-uuid",
"fitness_score": 0.85,
"rollout_percentage": 10,
"timestamp": "2026-04-10T12:00:00Z"
}---
Database Models
SkillCandidate
**Purpose:** Represents a candidate skill for deployment.
**Fields:**
class SkillCandidate(Base):
"""Skill candidate for evaluation."""
__tablename__ = "skill_candidates"
id = Column(UUID, primary_key=True)
tenant_id = Column(UUID, nullable=False)
agent_id = Column(UUID, nullable=False)
engine_type = Column(String, nullable=False)
# Skill definition
code = Column(Text, nullable=False)
parameters = Column(JSONB)
description = Column(Text)
# Metadata
parent_skill_id = Column(UUID, nullable=True)
pattern_id = Column(String, nullable=True)
confidence = Column(Float, default=0.0)
# Evaluation results
fitness_score = Column(Float)
success_rate = Column(Float)
efficiency = Column(Float)
quality = Column(Float)
consistency = Column(Float)
# Deployment status
status = Column(String) # pending, testing, deployed, rejected
deployed_at = Column(DateTime)
rollback_at = Column(DateTime)
# Timestamps
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, onupdate=datetime.utcnow)EvolutionRecord
**Purpose:** Tracks evolution history and deployed skills.
**Fields:**
class EvolutionRecord(Base):
"""Record of evolution cycle."""
__tablename__ = "evolution_records"
id = Column(UUID, primary_key=True)
tenant_id = Column(UUID, nullable=False)
agent_id = Column(UUID, nullable=False)
# Cycle info
cycle_number = Column(Integer, nullable=False)
engine_type = Column(String, nullable=False)
# Episode analysis
episodes_analyzed = Column(Integer, default=0)
patterns_discovered = Column(Integer, default=0)
# Candidate generation
candidates_generated = Column(Integer, default=0)
candidates_tested = Column(Integer, default=0)
# Deployment
skills_deployed = Column(Integer, default=0)
skills_rolled_back = Column(Integer, default=0)
# Performance
avg_fitness_before = Column(Float)
avg_fitness_after = Column(Float)
improvement_score = Column(Float)
# Timestamps
started_at = Column(DateTime, default=datetime.utcnow)
completed_at = Column(DateTime)
created_at = Column(DateTime, default=datetime.utcnow)---
Configuration
Engine Configuration
**MementoEngine:**
{
"min_pattern_frequency": 5,
"min_success_rate": 0.7,
"pattern_similarity_threshold": 0.8,
"max_candidates_per_cycle": 10,
"parameter_learning": True,
"context_awareness": True
}**AlphaEvolverEngine:**
{
"optimization_iterations": 100,
"mutation_rate": 0.1,
"crossover_rate": 0.7,
"elitism_count": 2,
"population_size": 20,
"convergence_threshold": 0.001
}**ReflectionEngine:**
{
"min_pattern_occurrences": 3,
"pattern_window": 50,
"trend_analysis_window": 100,
"confidence_threshold": 0.7,
"max_recommendations": 15
}**FitnessService:**
{
"success_rate_weight": 0.4,
"efficiency_weight": 0.3,
"quality_weight": 0.2,
"consistency_weight": 0.1,
"min_test_episodes": 20,
"max_test_episodes": 100
}**CapabilityGate:**
{
"gate_level": "standard",
"fitness_threshold": 0.8,
"require_safety_check": True,
"require_testing": True,
"min_test_episodes": 50,
"rollout_strategy": "gradual",
"rollout_percentages": [10, 50, 100]
}---
Examples
Example 1: Enable Auto-Dev for Agent
from core.auto_dev.memento_engine import MementoEngine
from core.models import Agent
# Get agent
agent = db.query(Agent).filter(Agent.id == agent_id).first()
# Initialize Memento engine
engine = MementoEngine(
db=db,
tenant_id=tenant_id,
agent_id=agent.id,
config={
"min_pattern_frequency": 5,
"max_candidates_per_cycle": 10
}
)
# Run evolution cycle
result = await engine.evolve(
episode_window=100,
min_fitness=0.8
)
print(f"Episodes analyzed: {result.episodes_analyzed}")
print(f"Candidates generated: {result.candidates_generated}")
print(f"Skills deployed: {result.skills_deployed}")Example 2: Subscribe to Evolution Events
from core.auto_dev.event_hooks import EventBus, EventType
# Initialize event bus
event_bus = EventBus(db, tenant_id)
# Subscribe to skill deployment events
async def on_skill_deployed(event: Event):
print(f"New skill deployed: {event.skill_id}")
print(f"Fitness score: {event.fitness_score}")
print(f"Rollout: {event.rollout_percentage}%")
subscription_id = await event_bus.subscribe(
event_type=EventType.SKILL_DEPLOYED,
handler=on_skill_deployed,
filter_params={"agent_id": agent_id}
)Example 3: Custom Fitness Evaluation
from core.auto_dev.fitness_service import FitnessService
# Initialize fitness service
fitness_service = FitnessService(
db=db,
tenant_id=tenant_id,
config={
"success_rate_weight": 0.5, # Custom weight
"efficiency_weight": 0.3,
"quality_weight": 0.1,
"consistency_weight": 0.1
}
)
# Evaluate candidate
fitness = await fitness_service.evaluate_fitness(
candidate=skill_candidate,
test_episodes=test_episodes
)
print(f"Overall fitness: {fitness.score}")
print(f"Success rate: {fitness.success_rate}")
print(f"Efficiency: {fitness.efficiency}")---
Type Definitions
FitnessScore
class FitnessScore(BaseModel):
"""Fitness evaluation result."""
score: float # Overall fitness (0-1)
success_rate: float # Success rate (0-1)
efficiency: float # Efficiency score (0-1)
quality: float # Quality score (0-1)
consistency: float # Consistency score (0-1)
test_count: int # Number of test episodes
details: Dict[str, Any] # Additional detailsSkillCandidate
class SkillCandidate(BaseModel):
"""Skill candidate for deployment."""
id: UUID
tenant_id: UUID
agent_id: UUID
engine_type: str # memento, alpha_evolver, reflection
code: str # Skill code
parameters: Dict[str, Any] # Parameters
description: str # Description
parent_skill_id: Optional[UUID] # Parent skill (if evolution)
pattern_id: Optional[str] # Pattern (if memento)
confidence: float # Confidence (0-1)
fitness_score: Optional[float] # Fitness score
status: str # pending, testing, deployed, rejectedEvolutionResult
class EvolutionResult(BaseModel):
"""Result of evolution cycle."""
engine_name: str # Engine that ran
episodes_analyzed: int # Episodes processed
candidates_generated: int # Candidates created
skills_deployed: int # Skills deployed
deployed_skills: List[Dict] # Deployed skill details---
Error Handling
Common Errors
**Error: Insufficient Episodes**
try:
result = await engine.evolve(episode_window=100)
except InsufficientEpisodesError as e:
print(f"Need at least {e.min_episodes} episodes, have {e.current_episodes}")
print(f"Agent needs more experience before evolution")**Error: Gate Violation**
try:
await gate_service.deploy_with_gate(candidate, fitness, episodes)
except GateViolationError as e:
print(f"Gate validation failed: {e.reason}")
print(f"Violated checks: {e.violated_checks}")**Error: Fitness Too Low**
try:
await engine.evolve(min_fitness=0.8)
except LowFitnessError as e:
print(f"No candidates met fitness threshold {e.threshold}")
print(f"Best fitness was: {e.best_fitness}")
print(f"Consider lowering threshold or gaining more experience")---
Performance Considerations
Optimization Tips
- **Batch Episode Processing**
- Process episodes in batches of 50-100
- Use pagination for large datasets
- Cache episode queries
- **Parallel Fitness Evaluation**
- Evaluate candidates in parallel
- Use async/await for I/O operations
- Limit concurrent evaluations
- **Incremental Learning**
- Don't re-analyze all episodes each cycle
- Track analyzed episodes
- Only process new episodes
- **Resource Management**
- Limit episode window size
- Clean up old skill versions
- Monitor memory usage
---