aradotso HyperAgents Self-Improving AI Skill
Skill by ara.so — AI Agent Skills collection.
Overview
HyperAgents is a framework for building self-referential self-improving AI agents that can optimize for any computable task. The system uses a meta-agent to iteratively improve a task-agent by generating and evaluating code modifications. The framework supports multiple domains (code generation, reasoning, math, etc.) and uses foundation models to drive the self-improvement loop.
Key Capabilities:
- Self-referential meta-learning where agents modify their own code
- Multi-domain support (code, math, reasoning tasks)
- Iterative improvement through generation-evaluation loops
- Integration with OpenAI, Anthropic, and Google Gemini models
- Docker-based safe execution environment
Installation
Prerequisites
# Install system dependencies (Fedora/RHEL)
sudo dnf install -y python3.12-devel graphviz graphviz-devel cmake ninja-build bzip2-devel zlib-devel ncurses-devel libffi-devel
# For Ubuntu/Debian:
# sudo apt-get install -y python3.12-dev graphviz libgraphviz-dev cmake ninja-build libbz2-dev zlib1g-dev libncurses-dev libffi-dev
Setup
# Clone the repository
git clone https://github.com/facebookresearch/HyperAgents.git
cd HyperAgents
# Create virtual environment
python3.12 -m venv venv_nat
source venv_nat/bin/activate
# Install dependencies
pip install -r requirements.txt
pip install -r requirements_dev.txt
# Build Docker container for safe execution
docker build --network=host -t hyperagents .
Environment Configuration
Create a .env file with your API keys:
# .env file
OPENAI_API_KEY=your_openai_key_here
ANTHROPIC_API_KEY=your_anthropic_key_here
GEMINI_API_KEY=your_gemini_key_here
Initialize Agents
# Setup initial agent implementations
bash ./setup_initial.sh
Core Concepts
Architecture
- Task Agent: Solves domain-specific tasks (code generation, math, etc.)
- Meta Agent: Observes task agent performance and generates improvements
- Generation Loop: Iteratively evolves agents through self-improvement cycles
File Structure
HyperAgents/
├── agent/ # Foundation model interfaces
├── domains/ # Task-specific implementations
├── utils/ # Common utilities
├── meta_agent.py # Meta-agent implementation
├── task_agent.py # Task-agent implementation
├── generate_loop.py # Main entry point
└── run_meta_agent.py # Meta-agent execution script
Usage
Running the Self-Improvement Loop
# Basic usage with default settings
python generate_loop.py --domains code_generation
# Multiple domains
python generate_loop.py --domains math reasoning
# Custom configuration
python generate_loop.py \
--domains code_generation \
--max_iterations 10 \
--output_dir ./my_outputs \
--model_name gpt-4
Key Command-Line Arguments
# Common arguments for generate_loop.py
--domains # Domain(s) to optimize (code_generation, math, reasoning, etc.)
--max_iterations # Maximum improvement iterations
--output_dir # Directory for outputs (default: outputs/)
--model_name # Foundation model to use
--baseline # Baseline agent to compare against
--temperature # Sampling temperature for generation
--num_samples # Number of samples per iteration
Working with Task Agents
Creating a Custom Task Agent
# task_agent.py - Basic structure
from typing import Any, Dict, List
from agent.base_agent import BaseAgent
class MyTaskAgent(BaseAgent):
"""Custom task agent for specific domain."""
def __init__(self, config: Dict[str, Any]):
super().__init__(config)
self.domain = config.get('domain', 'custom')
def solve_task(self, task_input: str) -> str:
"""
Main method to solve a task.
Args:
task_input: Input task specification
Returns:
Solution to the task
"""
# Generate prompt for the model
prompt = self._create_prompt(task_input)
# Get model response
response = self.model.generate(
prompt=prompt,
temperature=self.config.get('temperature', 0.7),
max_tokens=self.config.get('max_tokens', 2048)
)
# Post-process response
solution = self._parse_solution(response)
return solution
def _create_prompt(self, task_input: str) -> str:
"""Create prompt for the model."""
return f"""Solve the following task:
Task: {task_input}
Solution:"""
def _parse_solution(self, response: str) -> str:
"""Extract solution from model response."""
# Custom parsing logic
return response.strip()
def evaluate(self, task_input: str, solution: str) -> float:
"""
Evaluate solution quality.
Returns:
Score between 0 and 1
"""
# Domain-specific evaluation
return self._compute_score(task_input, solution)
Using the Task Agent
from task_agent import MyTaskAgent
# Initialize agent
config = {
'domain': 'custom',
'model_name': 'gpt-4',
'temperature': 0.7,
'max_tokens': 2048
}
agent = MyTaskAgent(config)
# Solve a task
task = "Write a function to compute Fibonacci numbers"
solution = agent.solve_task(task)
score = agent.evaluate(task, solution)
print(f"Solution: {solution}")
print(f"Score: {score}")
Working with Meta Agents
Meta Agent Structure
# meta_agent.py - Core implementation
from typing import Dict, List, Any
import difflib
class MetaAgent:
"""Meta-agent that improves task agents."""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.model = self._initialize_model()
self.history = []
def generate_improvement(
self,
current_code: str,
performance_data: List[Dict[str, Any]]
) -> str:
"""
Generate improved version of task agent.
Args:
current_code: Current task agent implementation
performance_data: Performance metrics from recent runs
Returns:
Improved code implementation
"""
# Analyze performance
insights = self._analyze_performance(performance_data)
# Generate improvement prompt
prompt = self._create_meta_prompt(current_code, insights)
# Generate new code
improved_code = self.model.generate(
prompt=prompt,
temperature=self.config.get('meta_temperature', 0.8),
max_tokens=self.config.get('meta_max_tokens', 4096)
)
# Validate and extract code
validated_code = self._validate_code(improved_code)
# Store in history
self.history.append({
'original': current_code,
'improved': validated_code,
'insights': insights
})
return validated_code
def _analyze_performance(
self,
performance_data: List[Dict[str, Any]]
) -> Dict[str, Any]:
"""Analyze performance metrics to identify improvement areas."""
# Compute statistics
scores = [d['score'] for d in performance_data]
avg_score = sum(scores) / len(scores)
# Identify failure patterns
failures = [d for d in performance_data if d['score'] < 0.5]
return {
'average_score': avg_score,
'num_failures': len(failures),
'failure_patterns': self._extract_patterns(failures)
}
def _create_meta_prompt(
self,
current_code: str,
insights: Dict[str, Any]
) -> str:
"""Create prompt for meta-level improvement."""
return f"""You are a meta-agent tasked with improving an AI task agent.
Current Implementation:
```python
{current_code}
Performance Analysis:
- Average Score: {insights['average_score']:.2f}
- Failures: {insights['num_failures']}
- Common Issues: {insights.get('failure_patterns', 'None identified')}
Generate an improved version that addresses these issues. Output only the complete improved code.
Improved Implementation:
def _validate_code(self, code: str) -> str:
"""Validate and extract code from response."""
# Extract code block
if '```python' in code:
code = code.split('```python')[1].split('```')[0]
# Basic syntax validation
try:
compile(code, '<string>', 'exec')
except SyntaxError as e:
raise ValueError(f"Generated code has syntax error: {e}")
return code.strip()
def compute_diff(self, old_code: str, new_code: str) -> List[str]:
"""Compute diff between code versions."""
diff = difflib.unified_diff(
old_code.splitlines(keepends=True),
new_code.splitlines(keepends=True),
fromfile='old_agent.py',
tofile='new_agent.py'
)
return list(diff)
Running Meta Agent
# run_meta_agent.py - Example usage
from meta_agent import MetaAgent
from task_agent import MyTaskAgent
import json
def run_meta_improvement_cycle(
initial_agent_code: str,
test_tasks: List[str],
num_iterations: int = 5
):
"""Run multiple iterations of meta-improvement."""
# Initialize meta-agent
meta_config = {
'model_name': 'gpt-4',
'meta_temperature': 0.8,
'meta_max_tokens': 4096
}
meta_agent = MetaAgent(meta_config)
current_code = initial_agent_code
for iteration in range(num_iterations):
print(f"\n=== Iteration {iteration + 1} ===")
# Evaluate current agent
performance_data = evaluate_agent(current_code, test_tasks)
avg_score = sum(d['score'] for d in performance_data) / len(performance_data)
print(f"Current Performance: {avg_score:.3f}")
# Generate improvement
improved_code = meta_agent.generate_improvement(
current_code,
performance_data
)
# Show diff
diff = meta_agent.compute_diff(current_code, improved_code)
print("Changes:")
print(''.join(diff[:20])) # Show first 20 lines
# Update current code
current_code = improved_code
# Save checkpoint
with open(f'agent_iteration_{iteration}.py', 'w') as f:
f.write(current_code)
return current_code
def evaluate_agent(agent_code: str, test_tasks: List[str]) -> List[Dict[str, Any]]:
"""Evaluate agent on test tasks."""
# Create agent from code
namespace = {}
exec(agent_code, namespace)
AgentClass = namespace['MyTaskAgent']
agent = AgentClass({'model_name': 'gpt-4'})
results = []
for task in test_tasks:
solution = agent.solve_task(task)
score = agent.evaluate(task, solution)
results.append({
'task': task,
'solution': solution,
'score': score
})
return results
# Usage
if __name__ == '__main__':
# Read initial agent code
with open('initial_agent.py', 'r') as f:
initial_code = f.read()
# Define test tasks
test_tasks = [
"Implement binary search",
"Write a function to reverse a linked list",
"Create a trie data structure"
]
# Run improvement loop
final_code = run_meta_improvement_cycle(
initial_code,
test_tasks,
num_iterations=5
)
print("\nFinal agent saved!")
Domain-Specific Implementation
Code Generation Domain
# domains/code_generation/agent.py
from typing import Dict, Any, List
import ast
import subprocess
class CodeGenerationAgent:
"""Agent specialized for code generation tasks."""
def generate_code(self, specification: str) -> str:
"""Generate code from specification."""
prompt = f"""Generate Python code for the following specification:
{specification}
Requirements:
- Include proper error handling
- Add docstrings
- Follow PEP 8 style guide
Code:
```python"""
code = self.model.generate(prompt)
return self._extract_code(code)
def test_code(self, code: str, test_cases: List[Dict[str, Any]]) -> float:
"""Test generated code against test cases."""
try:
# Create temporary module
namespace = {}
exec(code, namespace)
passed = 0
for test in test_cases:
func_name = test['function']
inputs = test['inputs']
expected = test['expected']
func = namespace[func_name]
result = func(*inputs)
if result == expected:
passed += 1
return passed / len(test_cases)
except Exception as e:
print(f"Test error: {e}")
return 0.0
def _extract_code(self, response: str) -> str:
"""Extract code from model response."""
if '```python' in response:
code = response.split('```python')[1].split('```')[0]
else:
code = response
# Validate syntax
try:
ast.parse(code)
except SyntaxError:
raise ValueError("Generated code has syntax errors")
return code.strip()
Math Reasoning Domain
# domains/math/agent.py
import re
from typing import Optional
class MathReasoningAgent:
"""Agent for mathematical reasoning tasks."""
def solve_math_problem(self, problem: str) -> Dict[str, Any]:
"""Solve a math problem with step-by-step reasoning."""
prompt = f"""Solve the following math problem step by step:
Problem: {problem}
Show your work clearly. Format your final answer as: ANSWER: <value>
Solution:"""
response = self.model.generate(prompt)
return {
'reasoning': response,
'answer': self._extract_answer(response)
}
def _extract_answer(self, response: str) -> Optional[str]:
"""Extract final answer from reasoning."""
# Look for ANSWER: pattern
match = re.search(r'ANSWER:\s*([^\n]+)', response, re.IGNORECASE)
if match:
return match.group(1).strip()
# Look for boxed answer (LaTeX)
match = re.search(r'\\boxed\{([^}]+)\}', response)
if match:
return match.group(1).strip()
# Try to find number at end
numbers = re.findall(r'-?\d+\.?\d*', response)
if numbers:
return numbers[-1]
return None
def evaluate_answer(
self,
predicted: str,
ground_truth: str,
tolerance: float = 1e-5
) -> bool:
"""Evaluate if answer is correct."""
try:
pred_val = float(predicted)
true_val = float(ground_truth)
return abs(pred_val - true_val) < tolerance
except (ValueError, TypeError):
# Fallback to string comparison
return predicted.strip() == ground_truth.strip()
Configuration Patterns
Agent Configuration
# config.py - Common configuration patterns
from dataclasses import dataclass
from typing import Optional
@dataclass
class AgentConfig:
"""Configuration for task agents."""
model_name: str = 'gpt-4'
temperature: float = 0.7
max_tokens: int = 2048
top_p: float = 1.0
frequency_penalty: float = 0.0
presence_penalty: float = 0.0
timeout: int = 60
max_retries: int = 3
@dataclass
class MetaAgentConfig:
"""Configuration for meta-agents."""
model_name: str = 'gpt-4'
meta_temperature: float = 0.8
meta_max_tokens: int = 4096
improvement_iterations: int = 5
min_improvement_threshold: float = 0.05
use_reflection: bool = True
@dataclass
class ExperimentConfig:
"""Configuration for experiments."""
domain: str
num_iterations: int = 10
num_eval_samples: int = 100
output_dir: str = './outputs'
save_checkpoints: bool = True
checkpoint_interval: int = 1
seed: Optional[int] = None
# Usage
agent_config = AgentConfig(
model_name='gpt-4-turbo',
temperature=0.5,
max_tokens=4096
)
meta_config = MetaAgentConfig(
improvement_iterations=10,
min_improvement_threshold=0.02
)
Loading Models
# agent/base_agent.py - Model initialization
from typing import Dict, Any
import os
from dotenv import load_dotenv
class BaseAgent:
"""Base class for all agents."""
def __init__(self, config: Dict[str, Any]):
load_dotenv()
self.config = config
self.model = self._initialize_model()
def _initialize_model(self):
"""Initialize the foundation model."""
model_name = self.config.get('model_name', 'gpt-4')
if 'gpt' in model_name.lower():
from openai import OpenAI
api_key = os.getenv('OPENAI_API_KEY')
client = OpenAI(api_key=api_key)
return OpenAIModel(client, model_name)
elif 'claude' in model_name.lower():
from anthropic import Anthropic
api_key = os.getenv('ANTHROPIC_API_KEY')
client = Anthropic(api_key=api_key)
return AnthropicModel(client, model_name)
elif 'gemini' in model_name.lower():
import google.generativeai as genai
api_key = os.getenv('GEMINI_API_KEY')
genai.configure(api_key=api_key)
return GeminiModel(model_name)
else:
raise ValueError(f"Unsupported model: {model_name}")
class OpenAIModel:
"""Wrapper for OpenAI models."""
def __init__(self, client, model_name: str):
self.client = client
self.model_name = model_name
def generate(
self,
prompt: str,
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs
) -> str:
"""Generate response from OpenAI model."""
response = self.client.chat.completions.create(
model=self.model_name,
messages=[{"role": "user", "content": prompt}],
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
return response.choices[0].message.content
Advanced Patterns
Batched Evaluation
# utils/evaluation.py
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor, as_completed
import numpy as np
class BatchEvaluator:
"""Efficiently evaluate agents on multiple tasks."""
def __init__(self, max_workers: int = 10):
self.max_workers = max_workers
def evaluate_batch(
self,
agent,
tasks: List[str],
ground_truths: List[Any]
) -> Dict[str, Any]:
"""Evaluate agent on batch of tasks in parallel."""
results = []
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {
executor.submit(self._evaluate_single, agent, task, truth): idx
for idx, (task, truth) in enumerate(zip(tasks, ground_truths))
}
for future in as_completed(futures):
idx = futures[future]
try:
result = future.result()
results.append((idx, result))
except Exception as e:
print(f"Task {idx} failed: {e}")
results.append((idx, {'score': 0.0, 'error': str(e)}))
# Sort by original order
results.sort(key=lambda x: x[0])
results = [r[1] for r in results]
return {
'results': results,
'mean_score': np.mean([r['score'] for r in results]),
'std_score': np.std([r['score'] for r in results]),
'success_rate': sum(1 for r in results if r['score'] > 0.5) / len(results)
}
def _evaluate_single(self, agent, task: str, ground_truth: Any) -> Dict[str, Any]:
"""Evaluate single task."""
solution = agent.solve_task(task)
score = agent.evaluate(task, solution, ground_truth)
return {
'task': task,
'solution': solution,
'score': score,
'correct': score > 0.5
}
Checkpointing
# utils/checkpointing.py
import json
import pickle
from pathlib import Path
from typing import Any, Dict
class CheckpointManager:
"""Manage experiment checkpoints."""
def __init__(self, output_dir: str):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
def save_checkpoint(
self,
iteration: int,
agent_code: str,
performance_data: Dict[str, Any],
meta_data: Dict[str, Any]
):
"""Save checkpoint for an iteration."""
checkpoint_dir = self.output_dir / f'iteration_{iteration}'
checkpoint_dir.mkdir(exist_ok=True)
# Save agent code
with open(checkpoint_dir / 'agent.py', 'w') as f:
f.write(agent_code)
# Save performance data
with open(checkpoint_dir / 'performance.json', 'w') as f:
json.dump(performance_data, f, indent=2)
# Save metadata
with open(checkpoint_dir / 'metadata.pkl', 'wb') as f:
pickle.dump(meta_data, f)
print(f"Checkpoint saved to {checkpoint_dir}")
def load_checkpoint(self, iteration: int) -> Dict[str, Any]:
"""Load checkpoint from an iteration."""
checkpoint_dir = self.output_dir / f'iteration_{iteration}'
with open(checkpoint_dir / 'agent.py', 'r') as f:
agent_code = f.read()
with open(checkpoint_dir / 'performance.json', 'r') as f:
performance_data = json.load(f)
with open(checkpoint_dir / 'metadata.pkl', 'rb') as f:
meta_data = pickle.load(f)
return {
'agent_code': agent_code,
'performance_data': performance_data,
'meta_data': meta_data
}
def list_checkpoints(self) -> List[int]:
"""List available checkpoints."""
iterations = []
for path in self.output_dir.glob('iteration_*'):
if path.is_dir():
iteration = int(path.name.split('_')[1])
iterations.append(iteration)
return sorted(iterations)
Troubleshooting
Common Issues
1. API Key Errors
# Verify environment variables
import os
from dotenv import load_dotenv
load_dotenv()
required_keys = ['OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'GEMINI_API_KEY']
for key in required_keys:
value = os.getenv(key)
if value:
print(f"{key}: {'*' * 20} (set)")
else:
print(f"{key}: NOT SET")
2. Docker Execution Errors
# Verify Docker is running
docker ps
# Rebuild container if needed
docker build --no-cache --network=host -t hyperagents .
# Check container logs
docker logs <container_id>
3. Code Generation Syntax Errors
# Add validation wrapper
import ast
def validate_generated_code(code: str) -> bool:
"""Validate Python syntax before execution."""
try:
ast.parse(code)
return True
except SyntaxError as e:
print(f"Syntax error at line {e.lineno}: {e.msg}")
print(f"Text: {e.text}")
return False
# Use in meta-agent
if validate_generated_code(improved_code):
current_code = improved_code
else:
print("Generated code has errors, keeping current version")
4. Performance Degradation
# Track performance over iterations
def monitor_performance(history: List[Dict[str, float]]):
"""Monitor for performance degradation."""
if len(history) < 3:
return
recent_scores = [h['score'] for h in history[-3:]]
if all(recent_scores[i] < recent_scores[i-1] for i in range(1, len(recent_scores))):
print("WARNING: Performance degrading for 3 consecutive iterations")
print("Consider:")
print(" - Reducing temperature")
print(" - Changing meta-agent prompt")
print(" - Rolling back to earlier checkpoint")
5. Memory Issues with Large Contexts
# Implement context truncation
def truncate_context(
context: str,
max_tokens: int = 8000,
tokenizer=None
) -> str:
"""Truncate context to fit within token limit."""
if tokenizer is None:
# Rough approximation: 4 chars per token
max_chars = max_tokens * 4
if len(context) > max_chars:
return context[:max_chars] + "\n... (truncated)"
else:
tokens = tokenizer.encode(context)
if len(tokens) > max_tokens:
truncated = tokenizer.decode(tokens[:max_tokens])
return truncated + "\n... (truncated)"
return context
Debugging Tips
# Enable verbose logging
import logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('hyperagents.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger('hyperagents')
# Use in code
logger.debug(f"Generating improvement for iteration {iteration}")
logger.info(f"Current score: {current_score:.3f}")
logger.warning(f"Low performance detected: {score:.3f}")
logger.error(f"Failed to generate valid code: {error}")
Safety Checks
# Implement safety checks before executing generated code
import re
def safety_check(code: str) -> Dict[str, bool]:
"""Check for potentially dangerous operations."""
checks = {
'no_file_deletion': 'os.remove' not in code and 'shutil.rmtree' not in code,
'no_system_calls': 'os.system' not in code and 'subprocess.call' not in code,
'no_network': 'requests.' not in code and 'urllib' not in code,
'no_eval': 'eval(' not in code and 'exec(' not in code,
}
all_safe = all(checks.values())
return {
'safe': all_safe,
'checks': checks
}
# Use before execution
safety_result = safety_check(generated_code)
if not safety_result['safe']:
print("WARNING: Potentially unsafe code detected!")
print(f"Failed checks: {[k for k, v in safety_result['checks'].items() if not v]}")
# Decide whether to proceed
Best Practices
- Always use environment variables for API keys, never hardcode
- Checkpoint frequently to avoid losing progress
- Validate generated code before execution
- Monitor performance across iterations to detect degradation
- Use Docker containers for safe code execution
- Implement timeouts for long-running operations
- Log extensively for debugging and analysis
- Test on small batches before full-scale runs