| name | hyperagents-self-improving-ai |
| description | Self-referential self-improving AI agents that optimize for any computable task using meta-learning and code generation |
| triggers | ["how do I use HyperAgents for self-improving AI","set up HyperAgents meta-agent system","run HyperAgents optimization loop","create custom domain for HyperAgents","configure HyperAgents task and meta agents","implement HyperAgents self-improvement","debug HyperAgents generated code","extend HyperAgents with new tasks"] |
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
sudo dnf install -y python3.12-devel graphviz graphviz-devel cmake ninja-build bzip2-devel zlib-devel ncurses-devel libffi-devel
Setup
git clone https://github.com/facebookresearch/HyperAgents.git
cd HyperAgents
python3.12 -m venv venv_nat
source venv_nat/bin/activate
pip install -r requirements.txt
pip install -r requirements_dev.txt
docker build --network=host -t hyperagents .
Environment Configuration
Create a .env file with your API keys:
OPENAI_API_KEY=your_openai_key_here
ANTHROPIC_API_KEY=your_anthropic_key_here
GEMINI_API_KEY=your_gemini_key_here
Initialize Agents
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
python generate_loop.py --domains code_generation
python generate_loop.py --domains math reasoning
python generate_loop.py \
--domains code_generation \
--max_iterations 10 \
--output_dir ./my_outputs \
--model_name gpt-4
Key Command-Line Arguments
--domains
--max_iterations
--output_dir
--model_name
--baseline
--temperature
--num_samples
Working with Task Agents
Creating a Custom Task Agent
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
"""
prompt = self._create_prompt(task_input)
response = self.model.generate(
prompt=prompt,
temperature=self.config.get('temperature', 0.7),
max_tokens=self.config.get('max_tokens', 2048)
)
solution = self._parse_solution(response)
return solution
def _create_prompt() -> :
() -> :
response.strip()
() -> :
._compute_score(task_input, solution)
Using the Task Agent
from task_agent import MyTaskAgent
config = {
'domain': 'custom',
'model_name': 'gpt-4',
'temperature': 0.7,
'max_tokens': 2048
}
agent = MyTaskAgent(config)
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
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
"""
insights = self._analyze_performance(performance_data)
prompt = self._create_meta_prompt(current_code, insights)
improved_code = self.model.generate(
prompt=prompt,
temperature=self.config.get('meta_temperature', 0.8),
max_tokens=self.config.get(, )
)
validated_code = ._validate_code(improved_code)
.history.append({
: current_code,
: validated_code,
: insights
})
validated_code
() -> [, ]:
scores = [d[] d performance_data]
avg_score = (scores) / (scores)
failures = [d d performance_data d[] < ]
{
: avg_score,
: (failures),
: ._extract_patterns(failures)
}
() -> :
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
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."""
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} ===")
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}")
improved_code = meta_agent.generate_improvement(
current_code,
performance_data
)
diff = meta_agent.compute_diff(current_code, improved_code)
print()
(.join(diff[:]))
current_code = improved_code
(, ) f:
f.write(current_code)
current_code
() -> [[, ]]:
namespace = {}
(agent_code, namespace)
AgentClass = namespace[]
agent = AgentClass({: })
results = []
task test_tasks:
solution = agent.solve_task(task)
score = agent.evaluate(task, solution)
results.append({
: task,
: solution,
: score
})
results
__name__ == :
(, ) f:
initial_code = f.read()
test_tasks = [
,
,
]
final_code = run_meta_improvement_cycle(
initial_code,
test_tasks,
num_iterations=
)
()
Domain-Specific Implementation
Code Generation Domain
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:
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 +=
passed / (test_cases)
Exception e:
()
() -> :
response:
code = response.split()[].split()[]
:
code = response
:
ast.parse(code)
SyntaxError:
ValueError()
code.strip()
Math Reasoning Domain
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."""
match = re.search(r'ANSWER:\s*([^\n]+)', response, re.IGNORECASE)
if match:
return match.group(1).strip()
match = re.search(r'\\boxed\{([^}]+)\}', response)
if match:
return .group().strip()
numbers = re.findall(, response)
numbers:
numbers[-]
() -> :
:
pred_val = (predicted)
true_val = (ground_truth)
(pred_val - true_val) < tolerance
(ValueError, TypeError):
predicted.strip() == ground_truth.strip()
Configuration Patterns
Agent Configuration
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: =
output_dir: =
save_checkpoints: =
checkpoint_interval: =
seed: [] =
agent_config = AgentConfig(
model_name=,
temperature=,
max_tokens=
)
meta_config = MetaAgentConfig(
improvement_iterations=,
min_improvement_threshold=
)
Loading Models
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():
google.generativeai genai
api_key = os.getenv()
genai.configure(api_key=api_key)
GeminiModel(model_name)
:
ValueError()
:
():
.client = client
.model_name = model_name
() -> :
response = .client.chat.completions.create(
model=.model_name,
messages=[{: , : prompt}],
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
response.choices[].message.content
Advanced Patterns
Batched Evaluation
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:
()
results.append((idx, {: , : (e)}))
results.sort(key= x: x[])
results = [r[] r results]
{
: results,
: np.mean([r[] r results]),
: np.std([r[] r results]),
: ( r results r[] > ) / (results)
}
() -> [, ]:
solution = agent.solve_task(task)
score = agent.evaluate(task, solution, ground_truth)
{
: task,
: solution,
: score,
: score >
}
Checkpointing
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)
with open(checkpoint_dir / 'agent.py', 'w') as f:
f.write(agent_code)
with open(checkpoint_dir / 'performance.json', 'w') f:
json.dump(performance_data, f, indent=)
(checkpoint_dir / , ) f:
pickle.dump(meta_data, f)
()
() -> [, ]:
checkpoint_dir = .output_dir /
(checkpoint_dir / , ) f:
agent_code = f.read()
(checkpoint_dir / , ) f:
performance_data = json.load(f)
(checkpoint_dir / , ) f:
meta_data = pickle.load(f)
{
: agent_code,
: performance_data,
: meta_data
}
() -> []:
iterations = []
path .output_dir.glob():
path.is_dir():
iteration = (path.name.split()[])
iterations.append(iteration)
(iterations)
Troubleshooting
Common Issues
1. API Key Errors
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
docker ps
docker build --no-cache --network=host -t hyperagents .
docker logs <container_id>
3. Code Generation Syntax Errors
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
if validate_generated_code(improved_code):
current_code = improved_code
else:
print("Generated code has errors, keeping current version")
4. Performance Degradation
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
def truncate_context(
context: str,
max_tokens: int = 8000,
tokenizer=None
) -> str:
"""Truncate context to fit within token limit."""
if tokenizer is None:
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
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')
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
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
}
safety_result = safety_check(generated_code)
if not safety_result['safe']:
print()
()
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
Resources