"""
Farnsworth LoRA Evolver - Adapter Evolution and Merging
Novel Approaches:
1. Interaction-Based Training - Learn from user interactions
2. Adapter Breeding - Combine successful adapters
3. A/B Testing - Compare adapter variants
4. TIES-Merging - Advanced adapter merging
"""
import asyncio
import json
import random
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Optional, Any
from loguru import logger
@dataclass
class LoRAConfig:
"""LoRA adapter configuration."""
rank: int = 8
alpha: int = 16
dropout: float = 0.1
target_modules: list[str] = field(default_factory=lambda: ["q_proj", "v_proj"])
@dataclass
class LoRAAdapter:
"""A LoRA adapter with metadata."""
id: str
config: LoRAConfig
base_model: str
created_at: datetime = field(default_factory=datetime.now)
parent_ids: list[str] = field(default_factory=list)
# Performance metrics
uses: int = 0
successes: int = 0
avg_score: float = 0.0
# Training data
training_examples: int = 0
# File path (when saved)
path: Optional[str] = None
def fitness(self) -> float:
"""Calculate fitness score."""
if self.uses == 0:
return 0.5
return (self.successes / self.uses) * 0.7 + self.avg_score * 0.3
@dataclass
class TrainingExample:
"""An example for LoRA training."""
prompt: str
response: str
quality_score: float
timestamp: datetime = field(default_factory=datetime.now)
class LoRAEvolver:
"""
Evolves LoRA adapters through breeding and training.
Features:
- Collect training data from interactions
- Train adapters when enough data accumulates
- Breed successful adapters
- A/B test adapter variants
"""
def __init__(
self,
data_dir: str = "./data/lora",
min_training_examples: int = 100,
default_config: Optional[LoRAConfig] = None,
):
self.data_dir = Path(data_dir)
self.data_dir.mkdir(parents=True, exist_ok=True)
self.min_training_examples = min_training_examples
self.default_config = default_config or LoRAConfig()
# Adapters
self.adapters: dict[str, LoRAAdapter] = {}
self.active_adapter: Optional[str] = None
# Training data
self.training_data: list[TrainingExample] = []
# A/B testing
self.ab_tests: dict[str, dict] = {}
# Statistics
self.stats = {
"adapters_created": 0,
"adapters_trained": 0,
"merges_performed": 0,
"ab_tests_completed": 0,
}
def add_training_example(
self,
prompt: str,
response: str,
quality_score: float,
):
"""Add a training example from user interaction."""
if quality_score < 0.5:
return # Only learn from good examples
example = TrainingExample(
prompt=prompt,
response=response,
quality_score=quality_score,
)
self.training_data.append(example)
# Check if we should train
if len(self.training_data) >= self.min_training_examples:
logger.info(f"Training data threshold reached: {len(self.training_data)} examples")
async def create_adapter(
self,
base_model: str,
config: Optional[LoRAConfig] = None,
) -> LoRAAdapter:
"""Create a new adapter (without training)."""
config = config or self.default_config
adapter = LoRAAdapter(
id=f"lora_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{random.randint(0, 999):03d}",
config=config,
base_model=base_model,
)
self.adapters[adapter.id] = adapter
self.stats["adapters_created"] += 1
return adapter
async def train_adapter(
self,
adapter_id: str,
examples: Optional[list[TrainingExample]] = None,
) -> bool:
"""
Train an adapter with collected examples.
Note: Actual training requires PEFT library and GPU.
This is a placeholder for the training logic.
"""
if adapter_id not in self.adapters:
return False
adapter = self.adapters[adapter_id]
examples = examples or self.training_data
if len(examples) < self.min_training_examples:
logger.warning(f"Not enough training examples: {len(examples)}")
return False
logger.info(f"Training adapter {adapter_id} with {len(examples)} examples")
# Placeholder for actual training
# In production, this would use PEFT/transformers
try:
# Simulate training
adapter.training_examples = len(examples)
adapter.path = str(self.data_dir / f"{adapter_id}")
# Save training data for later
training_file = self.data_dir / f"{adapter_id}_data.jsonl"
with training_file.open('w', encoding='utf-8') as f:
for ex in examples:
f.write(json.dumps({
"prompt": ex.prompt,
"response": ex.response,
"score": ex.quality_score,
}) + "\n")
self.stats["adapters_trained"] += 1
logger.info(f"Adapter {adapter_id} training complete")
return True
except Exception as e:
logger.error(f"Training failed: {e}")
return False
async def merge_adapters(
self,
adapter_ids: list[str],
method: str = "ties",
weights: Optional[list[float]] = None,
) -> Optional[LoRAAdapter]:
"""
Merge multiple adapters into one.
Methods:
- "average": Simple weight averaging
- "ties": TIES-Merging (trim, elect, sign)
- "dare": Drop and rescale
"""
if len(adapter_ids) < 2:
return None
adapters = [self.adapters[aid] for aid in adapter_ids if aid in self.adapters]
if len(adapters) < 2:
return None
logger.info(f"Merging adapters: {adapter_ids} using {method}")
# Create merged adapter
merged = LoRAAdapter(
id=f"merged_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
config=adapters[0].config,
base_model=adapters[0].base_model,
parent_ids=adapter_ids,
)
# Placeholder for actual merging
# In production, this would use MergeKit or similar
merged.path = str(self.data_dir / merged.id)
self.adapters[merged.id] = merged
self.stats["merges_performed"] += 1
return merged
async def breed_adapters(
self,
parent1_id: str,
parent2_id: str,
) -> Optional[LoRAAdapter]:
"""
Breed two adapters to create offspring.
Combines configs and merges weights.
"""
if parent1_id not in self.adapters or parent2_id not in self.adapters:
return None
p1 = self.adapters[parent1_id]
p2 = self.adapters[parent2_id]
# Create child config (mix of parents)
child_config = LoRAConfig(
rank=random.choice([p1.config.rank, p2.config.rank]),
alpha=random.choice([p1.config.alpha, p2.config.alpha]),
dropout=(p1.config.dropout + p2.config.dropout) / 2,
target_modules=list(set(p1.config.target_modules + p2.config.target_modules)),
)
child = await self.create_adapter(p1.base_model, child_config)
child.parent_ids = [parent1_id, parent2_id]
# Merge parent weights
await self.merge_adapters([parent1_id, parent2_id])
return child
def start_ab_test(
self,
adapter_a_id: str,
adapter_b_id: str,
test_id: Optional[str] = None,
) -> str:
"""Start an A/B test between two adapters."""
test_id = test_id or f"ab_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
self.ab_tests[test_id] = {
"adapter_a": adapter_a_id,
"adapter_b": adapter_b_id,
"results_a": [],
"results_b": [],
"started_at": datetime.now().isoformat(),
"status": "running",
}
logger.info(f"Started A/B test {test_id}: {adapter_a_id} vs {adapter_b_id}")
return test_id
def record_ab_result(
self,
test_id: str,
adapter_id: str,
success: bool,
score: float,
):
"""Record a result for an A/B test."""
if test_id not in self.ab_tests:
return
test = self.ab_tests[test_id]
if adapter_id == test["adapter_a"]:
test["results_a"].append({"success": success, "score": score})
elif adapter_id == test["adapter_b"]:
test["results_b"].append({"success": success, "score": score})
def get_ab_winner(self, test_id: str, min_samples: int = 30) -> Optional[str]:
"""
Get the winner of an A/B test.
Returns None if not enough samples or no significant difference.
"""
if test_id not in self.ab_tests:
return None
test = self.ab_tests[test_id]
results_a = test["results_a"]
results_b = test["results_b"]
if len(results_a) < min_samples or len(results_b) < min_samples:
return None
# Calculate success rates
rate_a = sum(1 for r in results_a if r["success"]) / len(results_a)
rate_b = sum(1 for r in results_b if r["success"]) / len(results_b)
# Simple significance check (could use proper stats)
if abs(rate_a - rate_b) < 0.1:
return None # No significant difference
winner = test["adapter_a"] if rate_a > rate_b else test["adapter_b"]
test["status"] = "completed"
test["winner"] = winner
self.stats["ab_tests_completed"] += 1
return winner
def select_adapter(self) -> Optional[str]:
"""Select adapter for use (for A/B testing)."""
# If A/B test is running, randomly select
for test_id, test in self.ab_tests.items():
if test["status"] == "running":
return random.choice([test["adapter_a"], test["adapter_b"]])
# Otherwise, use best adapter
if not self.adapters:
return None
return max(self.adapters.values(), key=lambda a: a.fitness()).id
def record_adapter_result(self, adapter_id: str, success: bool, score: float):
"""Record result for an adapter."""
if adapter_id not in self.adapters:
return
adapter = self.adapters[adapter_id]
adapter.uses += 1
adapter.successes += int(success)
adapter.avg_score = adapter.avg_score * 0.9 + score * 0.1
def get_stats(self) -> dict:
"""Get evolver statistics."""
return {
**self.stats,
"adapter_count": len(self.adapters),
"training_examples": len(self.training_data),
"active_ab_tests": sum(
1 for t in self.ab_tests.values() if t["status"] == "running"
),
"top_adapters": [
{"id": a.id, "fitness": a.fitness()}
for a in sorted(self.adapters.values(), key=lambda x: x.fitness(), reverse=True)[:5]
],
}
