| name | herclaw-agentsystem |
| description | A comprehensive self-improving AI agent system that integrates Hermes Agent's core capabilities including autonomous learning loop, skill creation, self-evolution, persistent memory, and nudge system. This skill enables AI agents to create skills from experience, improve them during use, maintain cross-session memory, and continuously evolve their capabilities. Use when building self-improving agents, autonomous learning systems, or AI that grows more capable over time. |
| license | MIT |
HerClaw Agent System
A comprehensive self-improving AI agent framework that combines the autonomous capabilities of Hermes Agent into a unified OpenClaw skill. This system enables AI agents to learn from experience, create and refine skills autonomously, maintain persistent memory across sessions, and continuously evolve their capabilities through reinforcement learning.
Skills Path
Skill Location: {project_path}/skills/herclaw-agentsystem
This skill is located at the above path in your project.
Overview
HerClaw Agent System is a complete autonomous agent framework that implements five interconnected subsystems:
- Learning Loop - Closed-loop learning from experience
- Skill Creation - Autonomous skill generation and management
- Self-Evolution - Continuous improvement through RL and optimization
- Persistent Memory - Cross-session memory with three-layer architecture
- Nudge System - Self-prompting for proactive behavior
┌─────────────────────────────────────────────────────────────────────────┐
│ HerClaw Agent System │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │
│ │ Learning Loop │───▶│ Skill Creation │───▶│ Self-Evolution │ │
│ │ │ │ │ │ │ │
│ │ • Experience │ │ • Opportunity │ │ • RL Pipeline │ │
│ │ Collection │ │ Detection │ │ • Behavior │ │
│ │ • Pattern │ │ • Template │ │ Optimization │ │
│ │ Extraction │ │ Generation │ │ • Capability │ │
│ │ • Skill │ │ • Validation │ │ Refinement │ │
│ │ Synthesis │ │ • Hub Sync │ │ • Deployment │ │
│ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │
│ │ │ │ │
│ └──────────────────────┼──────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Persistent Memory │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ Episodic │ │ Semantic │ │ User │ │ │
│ │ │ Memory │ │ Memory │ │ Model │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ • Episodes │ │ • Facts │ │ • Profile │ │ │
│ │ │ • Context │ │ • Concepts │ │ • Prefs │ │ │
│ │ │ • Outcomes │ │ • Relations │ │ • History │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ Nudge System │ │
│ │ • Memory Persistence Nudges • Skill Creation Triggers │ │
│ │ • Learning Reminders • Evolution Initiatives │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
Part 1: Learning Loop System
The Learning Loop is the core engine that enables autonomous learning from experience. It operates as a continuous cycle of experience collection, pattern extraction, skill synthesis, and validation.
Architecture
Phase 1: Experience Collection
Every interaction, tool call, and decision point is captured as structured trajectory data:
class ExperienceCollector:
"""
Collects and structures interaction experiences for learning.
"""
def __init__(self, storage_backend):
self.storage = storage_backend
self.trajectory_buffer = []
self.min_experiences_for_pattern = 5
def capture_interaction(self, interaction):
"""
Captures a single interaction as an experience.
Args:
interaction: Dict containing:
- user_input: The user's request
- context: Environmental context (files, state, etc.)
- actions: List of actions taken
- outcome: Success/failure and results
- feedback: Optional user feedback
"""
experience = {
'id': self._generate_id(),
'timestamp': datetime.now().isoformat(),
'user_input': interaction['user_input'],
'context': interaction.get('context', {}),
'actions': interaction['actions'],
'outcome': interaction['outcome'],
'feedback': interaction.get('feedback'),
'embedding': self._embed_interaction(interaction)
}
self.trajectory_buffer.append(experience)
self.storage.store_experience(experience)
return experience['id']
def get_similar_experiences(self, query, k=10):
"""
Retrieves similar past experiences using vector search.
"""
query_embedding = self._embed_query(query)
return self.storage.vector_search(query_embedding, k=k)
def get_recent_experiences(self, n=100):
"""
Returns the n most recent experiences.
"""
return self.trajectory_buffer[-n:]
Phase 2: Pattern Extraction
Identifies recurring patterns in successful interactions:
class PatternExtractor:
"""
Extracts actionable patterns from collected experiences.
"""
def __init__(self, llm_client, config):
self.llm = llm_client
self.config = config
self.pattern_threshold = config.get('pattern_threshold', 0.7)
def extract_patterns(self, experiences):
"""
Analyzes experiences to identify recurring patterns.
Returns:
List of Pattern objects with:
- pattern_type: Type of pattern identified
- frequency: How often this pattern occurs
- success_rate: Success rate when pattern is applied
- conditions: Conditions where pattern applies
- actions: Sequence of actions in the pattern
"""
# Cluster similar experiences
clusters = self._cluster_experiences(experiences)
patterns = []
for cluster in clusters:
if len(cluster) >= self.config.get('min_cluster_size', 3):
pattern = self._analyze_cluster(cluster)
if pattern['success_rate'] >= self.pattern_threshold:
patterns.append(pattern)
return patterns
def _cluster_experiences(self, experiences):
"""
Clusters experiences by similarity using embeddings.
"""
from sklearn.cluster import DBSCAN
import numpy as np
embeddings = np.array([e['embedding'] for e in experiences])
clustering = DBSCAN(eps=0.3, min_samples=3).fit(embeddings)
clusters = {}
for i, label in enumerate(clustering.labels_):
if label not in clusters:
clusters[label] = []
clusters[label].append(experiences[i])
return list(clusters.values())
def _analyze_cluster(self, cluster):
"""
Analyzes a cluster to extract pattern details.
"""
# Use LLM to identify common pattern
prompt = f"""
Analyze these successful interactions and identify the common pattern:
{json.dumps([{
'input': e['user_input'],
'actions': e['actions'],
'outcome': e['outcome']
} for e in cluster], indent=2)}
Extract:
1. The common task type
2. The typical sequence of actions
3. The conditions where this applies
4. Key decision points
"""
response = self.llm.generate(prompt)
pattern = self._parse_pattern_response(response, cluster)
return pattern
Phase 3: Skill Synthesis
Creates new skills from identified patterns:
class SkillSynthesizer:
"""
Synthesizes new skills from extracted patterns.
"""
def __init__(self, llm_client, skill_registry):
self.llm = llm_client
self.registry = skill_registry
def synthesize_skill(self, pattern, experiences):
"""
Creates a new skill from a pattern.
Args:
pattern: The identified pattern
experiences: Source experiences for the pattern
Returns:
Skill object ready for validation
"""
skill_name = self._generate_skill_name(pattern)
# Generate skill instructions
instructions = self._generate_instructions(pattern, experiences)
# Define trigger conditions
triggers = self._define_triggers(pattern)
# Create skill document
skill = Skill(
name=skill_name,
description=pattern['description'],
instructions=instructions,
triggers=triggers,
source_experiences=[e['id'] for e in experiences],
confidence=pattern['success_rate'],
created_at=datetime.now()
)
return skill
def _generate_instructions(self, pattern, experiences):
"""
Generates detailed skill instructions using LLM.
"""
prompt = f"""
Create a detailed skill instruction document based on this pattern:
Pattern: {pattern['pattern_type']}
Success Rate: {pattern['success_rate']}
Conditions: {pattern['conditions']}
Actions: {pattern['actions']}
Generate:
1. Clear step-by-step instructions
2. Decision points and branching logic
3. Error handling guidance
4. Success criteria
5. Example usage
Format as a markdown skill document.
"""
return self.llm.generate(prompt)
def _define_triggers(self, pattern):
"""
Defines when this skill should be activated.
"""
return {
'keywords': pattern.get('keywords', []),
'intent_patterns': pattern.get('intent_patterns', []),
'context_conditions': pattern.get('conditions', []),
'min_confidence': 0.7
}
Phase 4: Validation & Integration
Validates and integrates new skills:
class SkillValidator:
"""
Validates skills before integration.
"""
def __init__(self, test_harness, config):
self.test_harness = test_harness
self.config = config
self.min_validation_score = config.get('min_validation_score', 0.8)
def validate_skill(self, skill, test_cases):
"""
Validates a skill against test cases.
Returns:
ValidationResult with:
- passed: Boolean
- score: Validation score
- issues: List of issues found
- recommendations: Improvement suggestions
"""
results = []
for test_case in test_cases:
result = self.test_harness.run_test(skill, test_case)
results.append(result)
# Calculate overall score
score = sum(r['score'] for r in results) / len(results)
# Identify issues
issues = self._identify_issues(results)
# Generate recommendations
recommendations = self._generate_recommendations(issues)
return ValidationResult(
passed=score >= self.min_validation_score,
score=score,
issues=issues,
recommendations=recommendations
)
def integrate_skill(self, skill, validation_result):
"""
Integrates a validated skill into the registry.
"""
if not validation_result.passed:
raise SkillValidationError(validation_result.issues)
# Register skill
self.registry.register(skill)
# Update skill index for retrieval
self._update_skill_index(skill)
# Log integration
self._log_integration(skill, validation_result)
return skill
Complete Learning Loop
class LearningLoop:
"""
The complete learning loop that ties all phases together.
"""
def __init__(self, experience_collector, pattern_extractor,
skill_synthesizer, validator):
self.collector = experience_collector
self.extractor = pattern_extractor
self.synthesizer = skill_synthesizer
self.validator = validator
self.running = False
def start(self):
"""
Starts the continuous learning loop.
"""
self.running = True
while self.running:
self._run_cycle()
time.sleep(self.config.get('cycle_interval', 3600))
def _run_cycle(self):
"""
Runs a single learning cycle.
"""
# Get recent experiences
experiences = self.collector.get_recent_experiences()
if len(experiences) >= self.config.get('min_experiences', 10):
# Extract patterns
patterns = self.extractor.extract_patterns(experiences)
# Synthesize skills from patterns
for pattern in patterns:
skill = self.synthesizer.synthesize_skill(
pattern,
pattern['source_experiences']
)
# Validate skill
test_cases = self._generate_test_cases(pattern)
result = self.validator.validate_skill(skill, test_cases)
if result.passed:
self.validator.integrate_skill(skill, result)
self._notify_skill_created(skill)
Part 2: Skill Creation System
The Skill Creation system enables autonomous generation of new skills from experience patterns, with full lifecycle management and Skills Hub integration.
Architecture
Skill Opportunity Detection
class SkillOpportunityDetector:
"""
Detects opportunities for creating new skills.
"""
def __init__(self, config):
self.config = config
self.opportunity_threshold = config.get('opportunity_threshold', 0.6)
def detect_opportunities(self, experiences, existing_skills):
"""
Analyzes experiences to find skill creation opportunities.
Returns:
List of SkillOpportunity objects
"""
opportunities = []
# Check for repeated similar tasks
task_clusters = self._cluster_by_task(experiences)
for cluster in task_clusters:
if self._is_skill_opportunity(cluster, existing_skills):
opportunities.append(SkillOpportunity(
type='repeated_task',
cluster=cluster,
priority=self._calculate_priority(cluster),
suggested_name=self._suggest_name(cluster)
))
# Check for complex multi-step processes
complex_tasks = self._identify_complex_tasks(experiences)
for task in complex_tasks:
if self._is_skill_opportunity(task, existing_skills):
opportunities.append(SkillOpportunity(
type='complex_process',
task=task,
priority=self._calculate_priority(task),
suggested_name=self._suggest_name(task)
))
# Check for user-requested patterns
user_requests = self._identify_user_requests(experiences)
for request in user_requests:
opportunities.append(SkillOpportunity(
type='user_request',
request=request,
priority='high',
suggested_name=self._suggest_name(request)
))
return sorted(opportunities, key=lambda x: x.priority, reverse=True)
def _is_skill_opportunity(self, cluster, existing_skills):
"""
Determines if a cluster represents a skill opportunity.
"""
# Check if existing skills cover this
for skill in existing_skills:
if self._skill_covers_cluster(skill, cluster):
return False
# Check frequency threshold
if len(cluster) < self.config.get('min_frequency', 3):
return False
# Check success rate
success_rate = sum(1 for e in cluster if e['outcome']['success']) / len(cluster)
if success_rate < self.opportunity_threshold:
return False
return True
Template-Based Skill Generation
class SkillTemplateGenerator:
"""
Generates skill templates using LLM.
"""
def __init__(self, llm_client):
self.llm = llm_client
self.templates = self._load_base_templates()
def generate_template(self, opportunity):
"""
Generates a skill template for the opportunity.
"""
template_type = self._determine_template_type(opportunity)
base_template = self.templates[template_type]
# Customize template based on opportunity
prompt = f"""
Generate a skill template for:
Opportunity Type: {opportunity.type}
Task Pattern: {json.dumps(opportunity.cluster[:3], indent=2)}
Suggested Name: {opportunity.suggested_name}
Base Template Structure:
{base_template}
Generate a complete skill template with:
1. Name and description
2. Trigger conditions
3. Step-by-step instructions
4. Input/output specifications
5. Error handling
6. Examples
"""
template_content = self.llm.generate(prompt)
return SkillTemplate(
name=opportunity.suggested_name,
content=template_content,
type=template_type,
source_opportunity=opportunity
)
def _load_base_templates(self):
"""
Loads base skill templates.
"""
return {
'workflow': """
# Skill: {name}
## Description
{description}
## Triggers
- Keywords: {keywords}
- Intent: {intent}
## Instructions
### Step 1: {step1}
### Step 2: {step2}
...
## Inputs
- {input1}: {description}
## Outputs
- {output1}: {description}
## Examples
{examples}
""",
'analysis': """
# Skill: {name}
## Description
{description}
## Analysis Framework
1. Data Collection
2. Processing
3. Analysis
4. Reporting
## Instructions
{instructions}
""",
'automation': """
# Skill: {name}
## Description
{description}
## Automation Steps
{steps}
## Scheduling
{scheduling}
"""
}
Skill Instantiation
class SkillInstantiator:
"""
Instantiates skills from templates.
"""
def __init__(self, llm_client, skill_registry):
self.llm = llm_client
self.registry = skill_registry
def instantiate(self, template, context=None):
"""
Creates a concrete skill from a template.
"""
# Fill in template with context
skill_content = self._fill_template(template, context)
# Parse into skill structure
skill = self._parse_skill(skill_content)
# Add metadata
skill.metadata = {
'created_at': datetime.now().isoformat(),
'source': 'auto_generated',
'template_id': template.id,
'version': '1.0.0'
}
# Generate embeddings for retrieval
skill.embedding = self._generate_embedding(skill)
return skill
def _fill_template(self, template, context):
"""
Fills template placeholders with context.
"""
content = template.content
if context:
for key, value in context.items():
content = content.replace(f'{{{key}}}', str(value))
# Use LLM to fill remaining placeholders
if '{' in content:
content = self.llm.generate(f"""
Complete this skill template by filling in all placeholders:
{content}
Context: {json.dumps(context or {})}
""")
return content
Skills Hub Integration
class SkillsHub:
"""
Integration with the Skills Hub for sharing and discovering skills.
"""
def __init__(self, config):
self.config = config
self.hub_url = config.get('hub_url', 'https://hub.agentskills.io')
self.local_cache = {}
def publish_skill(self, skill):
"""
Publishes a skill to the Hub.
"""
payload = {
'name': skill.name,
'description': skill.description,
'instructions': skill.instructions,
'triggers': skill.triggers,
'metadata': skill.metadata
}
response = requests.post(
f"{self.hub_url}/skills",
json=payload,
headers=self._get_auth_headers()
)
if response.status_code == 201:
skill.hub_id = response.json()['id']
return True
return False
def discover_skills(self, query, limit=10):
"""
Discovers skills from the Hub matching a query.
"""
response = requests.get(
f"{self.hub_url}/search",
params={'q': query, 'limit': limit}
)
return [Skill.from_dict(s) for s in response.json()['skills']]
def install_skill(self, skill_id):
"""
Installs a skill from the Hub.
"""
response = requests.get(f"{self.hub_url}/skills/{skill_id}")
skill_data = response.json()
skill = Skill.from_dict(skill_data)
skill.source = 'hub_installed'
return skill
Skill Lifecycle Management
class SkillLifecycleManager:
"""
Manages the complete lifecycle of skills.
"""
def __init__(self, registry, hub):
self.registry = registry
self.hub = hub
self.analytics = SkillAnalytics()
def create_skill(self, opportunity):
"""
Creates a new skill from an opportunity.
"""
# Generate template
template = self.template_generator.generate_template(opportunity)
# Instantiate skill
skill = self.instantiator.instantiate(template)
# Validate
validation = self.validator.validate(skill)
if not validation.passed:
skill = self._refine_skill(skill, validation)
# Register
self.registry.register(skill)
return skill
def update_skill(self, skill_id, updates):
"""
Updates an existing skill.
"""
skill = self.registry.get(skill_id)
# Apply updates
for key, value in updates.items():
setattr(skill, key, value)
# Increment version
skill.metadata['version'] = self._increment_version(
skill.metadata['version']
)
# Re-validate
validation = self.validator.validate(skill)
if validation.passed:
self.registry.update(skill)
return skill
def deprecate_skill(self, skill_id, reason):
"""
Deprecates a skill.
"""
skill = self.registry.get(skill_id)
skill.status = 'deprecated'
skill.deprecation_reason = reason
skill.deprecated_at = datetime.now()
self.registry.update(skill)
# Notify Hub if published
if skill.hub_id:
self.hub.deprecate(skill.hub_id, reason)
def get_skill_analytics(self, skill_id):
"""
Returns analytics for a skill.
"""
return self.analytics.get_stats(skill_id)
Part 3: Self-Evolution System
The Self-Evolution system enables continuous improvement through reinforcement learning, behavioral optimization, and adaptive refinement using the Atropos RL pipeline.
Architecture
Evolution Orchestrator
class EvolutionOrchestrator:
"""
Orchestrates the self-evolution process.
"""
def __init__(self, config):
self.config = config
self.evolution_cycle = 0
self.performance_history = []
self.evolution_log = []
def start_evolution(self):
"""
Starts the continuous evolution process.
"""
while self.config.enabled:
cycle_start = datetime.now()
# Phase 1: Collect performance data
performance_data = self._collect_performance_data()
# Phase 2: Identify evolution opportunities
opportunities = self._identify_evolution_opportunities(performance_data)
# Phase 3: Execute evolution
for opportunity in opportunities:
result = self._execute_evolution(opportunity)
self._log_evolution(result)
# Phase 4: Validate and deploy
self._validate_and_deploy()
self.evolution_cycle += 1
self._wait_for_next_cycle()
def _collect_performance_data(self):
"""
Collects performance metrics from all subsystems.
"""
return {
'skill_performance': self._get_skill_metrics(),
'memory_efficiency': self._get_memory_metrics(),
'user_satisfaction': self._get_satisfaction_metrics(),
'task_completion': self._get_completion_metrics(),
'error_rates': self._get_error_metrics()
}
def _identify_evolution_opportunities(self, data):
"""
Identifies areas for evolution based on performance data.
"""
opportunities = []
# Check for underperforming skills
for skill_id, metrics in data['skill_performance'].items():
if metrics['success_rate'] < self.config.get('target_success_rate', 0.9):
opportunities.append(EvolutionOpportunity(
type='skill_refinement',
target=skill_id,
current_performance=metrics,
priority='high'
))
# Check for memory optimization
if data['memory_efficiency']['retrieval_latency'] > self.config.get('target_latency', 100):
opportunities.append(EvolutionOpportunity(
type='memory_optimization',
target='memory_system',
current_performance=data['memory_efficiency'],
priority='medium'
))
return opportunities
Atropos RL Pipeline
class AtroposRLPipeline:
"""
Reinforcement learning pipeline for behavior optimization.
Based on the Atropos framework for RL fine-tuning.
"""
def __init__(self, config, model_interface):
self.config = config
self.model = model_interface
self.reward_functions = self._initialize_reward_functions()
self.training_data = []
def fine_tune(self, target, performance_data):
"""
Fine-tunes the model for a specific target.
Args:
target: What to optimize (skill, behavior, etc.)
performance_data: Current performance metrics
Returns:
FineTuningResult with new model weights
"""
# Generate training episodes
episodes = self._generate_training_episodes(target, performance_data)
# Compute rewards
rewarded_episodes = self._compute_rewards(episodes)
# Run PPO optimization
result = self._run_ppo_optimization(rewarded_episodes)
return result
def _initialize_reward_functions(self):
"""
Initializes reward functions for different optimization targets.
"""
return {
'task_completion': TaskCompletionReward(),
'user_satisfaction': UserSatisfactionReward(),
'efficiency': EfficiencyReward(),
'accuracy': AccuracyReward(),
'safety': SafetyReward()
}
def _generate_training_episodes(self, target, performance_data):
"""
Generates training episodes for RL.
"""
episodes = []
# Sample from historical data
historical = self._sample_historical_interactions(target)
# Generate variations
for interaction in historical:
variations = self._generate_variations(interaction)
for variation in variations:
episode = self._create_episode(variation)
episodes.append(episode)
return episodes
def _compute_rewards(self, episodes):
"""
Computes rewards for each episode.
"""
for episode in episodes:
total_reward = 0
for reward_name, reward_func in self.reward_functions.items():
reward_value = reward_func.compute(episode)
total_reward += self.config.get(f'{reward_name}_weight', 1.0) * reward_value
episode.reward = total_reward
return episodes
def _run_ppo_optimization(self, episodes):
"""
Runs Proximal Policy Optimization on the episodes.
"""
# PPO hyperparameters
learning_rate = self.config.get('learning_rate', 1e-5)
clip_range = self.config.get('clip_range', 0.2)
epochs = self.config.get('epochs', 4)
# Run optimization
optimizer = PPOOptimizer(
model=self.model,
learning_rate=learning_rate,
clip_range=clip_range
)
result = optimizer.train(episodes, epochs=epochs)
return FineTuningResult(
new_weights=result.weights,
performance_improvement=result.improvement,
training_loss=result.loss
)
class TaskCompletionReward:
"""
Reward function for task completion.
"""
def compute(self, episode):
if episode.outcome.get('success'):
return 1.0
elif episode.outcome.get('partial_success'):
return 0.5
return 0.0
class UserSatisfactionReward:
"""
Reward function based on user feedback.
"""
def compute(self, episode):
feedback = episode.outcome.get('user_feedback')
if feedback:
if feedback.get('rating', 0) >= 4:
return 1.0
elif feedback.get('rating', 0) >= 3:
return 0.5
return 0.0
Behavioral Optimizer
class BehavioralOptimizer:
"""
Optimizes agent behaviors based on performance data.
"""
def __init__(self, config):
self.config = config
self.behavior_models = {}
def optimize_behavior(self, behavior_type, performance_data):
"""
Optimizes a specific behavior type.
"""
# Analyze current behavior patterns
current_patterns = self._analyze_patterns(behavior_type, performance_data)
# Identify improvement opportunities
improvements = self._identify_improvements(current_patterns)
# Generate optimized behavior
optimized = self._generate_optimized_behavior(behavior_type, improvements)
return BehaviorOptimizationResult(
behavior_type=behavior_type,
original_patterns=current_patterns,
optimized_patterns=optimized,
expected_improvement=self._estimate_improvement(improvements)
)
def _analyze_patterns(self, behavior_type, data):
"""
Analyzes patterns in behavior data.
"""
patterns = {
'action_sequences': [],
'decision_points': [],
'timing_patterns': [],
'context_correlations': []
}
for interaction in data:
if interaction.get('behavior_type') == behavior_type:
patterns['action_sequences'].append(interaction['actions'])
patterns['decision_points'].append(interaction.get('decisions', []))
patterns['timing_patterns'].append(interaction.get('timing', {}))
return patterns
def _identify_improvements(self, patterns):
"""
Identifies potential improvements in patterns.
"""
improvements = []
# Find inefficient action sequences
for seq in patterns['action_sequences']:
if self._has_redundant_actions(seq):
improvements.append({
'type': 'action_optimization',
'original': seq,
'suggestion': self._optimize_sequence(seq)
})
# Find suboptimal decision patterns
for decisions in patterns['decision_points']:
if self._has_suboptimal_decisions(decisions):
improvements.append({
'type': 'decision_optimization',
'original': decisions,
'suggestion': self._optimize_decisions(decisions)
})
return improvements
Capability Refinement
class CapabilityRefiner:
"""
Refines agent capabilities through iterative improvement.
"""
def __init__(self, config, llm_client):
self.config = config
self.llm = llm_client
self.refinement_history = []
def refine_capability(self, capability, performance_metrics):
"""
Refines a specific capability.
"""
# Analyze current capability performance
analysis = self._analyze_capability(capability, performance_metrics)
# Generate refinement strategies
strategies = self._generate_refinement_strategies(analysis)
# Apply best strategy
best_strategy = self._select_best_strategy(strategies)
refined_capability = self._apply_strategy(capability, best_strategy)
# Validate refinement
validation = self._validate_refinement(refined_capability)
if validation.passed:
self.refinement_history.append({
'capability': capability.name,
'strategy': best_strategy,
'improvement': validation.improvement,
'timestamp': datetime.now()
})
return refined_capability
return capability
def _generate_refinement_strategies(self, analysis):
"""
Generates strategies for capability refinement.
"""
prompt = f"""
Analyze this capability and generate refinement strategies:
Capability: {analysis['capability_name']}
Current Performance: {analysis['performance']}
Weaknesses: {analysis['weaknesses']}
Failure Cases: {analysis['failures']}
Generate 3-5 specific refinement strategies that could improve performance.
Each strategy should include:
1. Strategy name
2. Description of changes
3. Expected improvement
4. Implementation approach
"""
response = self.llm.generate(prompt)
return self._parse_strategies(response)
Evolution Validation & Deployment
class EvolutionValidator:
"""
Validates evolution results before deployment.
"""
def __init__(self, config):
self.config = config
self.test_suites = {}
def validate_evolution(self, evolution_result):
"""
Validates an evolution result.
"""
# Run regression tests
regression_results = self._run_regression_tests(evolution_result)
# Run performance tests
performance_results = self._run_performance_tests(evolution_result)
# Run safety tests
safety_results = self._run_safety_tests(evolution_result)
# Calculate overall validation score
score = self._calculate_validation_score(
regression_results,
performance_results,
safety_results
)
return EvolutionValidationResult(
passed=score >= self.config.get('min_validation_score', 0.85),
score=score,
regression_results=regression_results,
performance_results=performance_results,
safety_results=safety_results
)
class EvolutionDeployer:
"""
Deploys validated evolutions.
"""
def __init__(self, config):
self.config = config
self.deployment_history = []
def deploy(self, evolution_result, validation_result):
"""
Deploys an evolution with staged rollout.
"""
if not validation_result.passed:
raise EvolutionValidationError("Evolution failed validation")
# Staged rollout
stages = self.config.get('rollout_stages', [0.1, 0.25, 0.5, 1.0])
for stage in stages:
# Deploy to stage percentage of traffic
self._deploy_to_stage(evolution_result, stage)
# Monitor for issues
health_check = self._monitor_health(stage)
if not health_check.healthy:
# Rollback
self._rollback(evolution_result)
return DeploymentResult(
success=False,
stage_reached=stage,
rollback_reason=health_check.issues
)
# Full deployment successful
self._record_deployment(evolution_result)
return DeploymentResult(
success=True,
deployed_at=datetime.now()
)
def _monitor_health(self, stage):
"""
Monitors system health after deployment.
"""
metrics = self._collect_metrics()
issues = []
# Check error rate
if metrics['error_rate'] > self.config.get('max_error_rate', 0.05):
issues.append(f"Error rate too high: {metrics['error_rate']}")
# Check latency
if metrics['latency_p99'] > self.config.get('max_latency', 5000):
issues.append(f"Latency too high: {metrics['latency_p99']}")
# Check user satisfaction
if metrics['user_satisfaction'] < self.config.get('min_satisfaction', 0.8):
issues.append(f"User satisfaction too low: {metrics['user_satisfaction']}")
return HealthCheckResult(
healthy=len(issues) == 0,
issues=issues,
metrics=metrics
)
Part 4: Persistent Memory System
The Persistent Memory system provides a three-layer architecture for cross-session memory, powered by ChromaDB vector search for efficient retrieval.
Architecture
Layer 1: Episodic Memory
class EpisodicMemory:
"""
Stores and retrieves specific interaction episodes.
Uses ChromaDB for vector-based semantic search.
"""
def __init__(self, chroma_client, embedding_model):
self.client = chroma_client
self.embedding_model = embedding_model
self.collection = self.client.get_or_create_collection(
name="episodic_memory",
metadata={"hnsw:space": "cosine"}
)
def store_episode(self, episode):
"""
Stores an interaction episode.
Args:
episode: Dict containing:
- id: Unique episode identifier
- timestamp: When the episode occurred
- user_input: What the user said
- agent_response: How the agent responded
- context: Environmental context
- outcome: Result of the interaction
- importance: Importance score (0-1)
"""
# Generate embedding for the episode
episode_text = self._episode_to_text(episode)
embedding = self.embedding_model.embed(episode_text)
# Store in ChromaDB
self.collection.add(
ids=[episode['id']],
embeddings=[embedding],
metadatas=[{
'timestamp': episode['timestamp'],
'importance': episode.get('importance', 0.5),
'outcome_success': episode['outcome'].get('success', False),
'user_id': episode.get('user_id', 'default')
}],
documents=[episode_text]
)
def retrieve_relevant(self, query, k=10, filters=None):
"""
Retrieves episodes relevant to a query.
Args:
query: The search query
k: Number of results to return
filters: Optional metadata filters
Returns:
List of relevant episodes with similarity scores
"""
query_embedding = self.embedding_model.embed(query)
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=k,
where=filters
)
return self._format_results(results)
def get_recent_episodes(self, n=50, user_id=None):
"""
Gets the most recent episodes.
"""
filters = None
if user_id:
filters = {'user_id': user_id}
# Get all episodes sorted by timestamp
results = self.collection.get(
where=filters,
include=['metadatas', 'documents']
)
# Sort by timestamp
episodes = list(zip(results['ids'], results['metadatas'], results['documents']))
episodes.sort(key=lambda x: x[1]['timestamp'], reverse=True)
return episodes[:n]
def _episode_to_text(self, episode):
"""
Converts an episode to searchable text.
"""
return f"""
User: {episode['user_input']}
Context: {json.dumps(episode.get('context', {}))}
Agent: {episode['agent_response']}
Outcome: {json.dumps(episode['outcome'])}
"""
Layer 2: Semantic Memory
class SemanticMemory:
"""
Stores general knowledge and facts extracted from interactions.
"""
def __init__(self, chroma_client, embedding_model):
self.client = chroma_client
self.embedding_model = embedding_model
self.collection = self.client.get_or_create_collection(
name="semantic_memory",
metadata={"hnsw:space": "cosine"}
)
self.fact_extractor = FactExtractor()
def store_knowledge(self, knowledge_item):
"""
Stores a piece of knowledge.
Args:
knowledge_item: Dict containing:
- id: Unique identifier
- fact: The knowledge fact
- source: Where this knowledge came from
- confidence: Confidence level (0-1)
- category: Knowledge category
- related_concepts: List of related concepts
"""
embedding = self.embedding_model.embed(knowledge_item['fact'])
self.collection.add(
ids=[knowledge_item['id']],
embeddings=[embedding],
metadatas=[{
'source': knowledge_item['source'],
'confidence': knowledge_item['confidence'],
'category': knowledge_item.get('category', 'general'),
'created_at': datetime.now().isoformat()
}],
documents=[knowledge_item['fact']]
)
def extract_and_store_from_episode(self, episode):
"""
Extracts knowledge from an episode and stores it.
"""
# Extract facts from the episode
facts = self.fact_extractor.extract(episode)
for fact in facts:
knowledge_item = {
'id': self._generate_id(),
'fact': fact['content'],
'source': f"episode:{episode['id']}",
'confidence': fact['confidence'],
'category': fact['category']
}
self.store_knowledge(knowledge_item)
def query_knowledge(self, query, k=10, category=None):
"""
Queries the knowledge base.
"""
query_embedding = self.embedding_model.embed(query)
filters = None
if category:
filters = {'category': category}
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=k,
where=filters
)
return self._format_results(results)
def update_knowledge(self, knowledge_id, updates):
"""
Updates existing knowledge.
"""
# Get existing knowledge
existing = self.collection.get(ids=[knowledge_id])
if not existing['ids']:
raise KnowledgeNotFoundError(knowledge_id)
# Update metadata
metadata = existing['metadatas'][0]
metadata.update(updates.get('metadata', {}))
# Update document if provided
document = updates.get('fact', existing['documents'][0])
# Re-embed if document changed
if 'fact' in updates:
embedding = self.embedding_model.embed(document)
self.collection.update(
ids=[knowledge_id],
embeddings=[embedding],
metadatas=[metadata],
documents=[document]
)
else:
self.collection.update(
ids=[knowledge_id],
metadatas=[metadata]
)
class FactExtractor:
"""
Extracts facts from episodes using LLM.
"""
def __init__(self, llm_client):
self.llm = llm_client
def extract(self, episode):
"""
Extracts facts from an episode.
"""
prompt = f"""
Extract factual knowledge from this interaction:
User Input: {episode['user_input']}
Agent Response: {episode['agent_response']}
Context: {json.dumps(episode.get('context', {}))}
Extract:
1. User preferences mentioned
2. Project/context information
3. Technical decisions made
4. Important facts learned
Format each as:
- content: The fact
- category: preference/project/technical/fact
- confidence: 0.0-1.0
"""
response = self.llm.generate(prompt)
return self._parse_facts(response)
Layer 3: User Model
class UserModel:
"""
Maintains a model of the user's preferences, patterns, and history.
"""
def __init__(self, storage_backend):
self.storage = storage_backend
self.user_profiles = {}
def get_or_create_profile(self, user_id):
"""
Gets or creates a user profile.
"""
if user_id not in self.user_profiles:
profile = self.storage.get_profile(user_id)
if profile:
self.user_profiles[user_id] = profile
else:
self.user_profiles[user_id] = self._create_default_profile(user_id)
return self.user_profiles[user_id]
def _create_default_profile(self, user_id):
"""
Creates a default user profile.
"""
return UserProfile(
user_id=user_id,
preferences={
'communication_style': 'professional',
'detail_level': 'moderate',
'preferred_tools': [],
'working_hours': None,
'timezone': None
},
patterns={
'common_tasks': [],
'frequent_projects': [],
'typical_workflow': []
},
history={
'total_interactions': 0,
'first_interaction': datetime.now(),
'last_interaction': datetime.now()
},
expertise_level='intermediate'
)
def update_preference(self, user_id, category, key, value):
"""
Updates a user preference.
"""
profile = self.get_or_create_profile(user_id)
if category not in profile.preferences:
profile.preferences[category] = {}
profile.preferences[category][key] = value
profile.history['last_interaction'] = datetime.now()
self._save_profile(profile)
def record_pattern(self, user_id, pattern_type, pattern_data):
"""
Records a user behavior pattern.
"""
profile = self.get_or_create_profile(user_id)
if pattern_type not in profile.patterns:
profile.patterns[pattern_type] = []
# Check if pattern already exists
existing = self._find_similar_pattern(
profile.patterns[pattern_type],
pattern_data
)
if existing:
existing['frequency'] += 1
existing['last_occurred'] = datetime.now()
else:
profile.patterns[pattern_type].append({
'data': pattern_data,
'frequency': 1,
'first_occurred': datetime.now(),
'last_occurred': datetime.now()
})
self._save_profile(profile)
def get_user_context(self, user_id):
"""
Gets relevant context for the user.
"""
profile = self.get_or_create_profile(user_id)
return {
'preferences': profile.preferences,
'recent_patterns': self._get_recent_patterns(profile),
'expertise_level': profile.expertise_level,
'interaction_count': profile.history['total_interactions']
}
@dataclass
class UserProfile:
"""
User profile data structure.
"""
user_id: str
preferences: Dict[str, Any]
patterns: Dict[str, List[Dict]]
history: Dict[str, Any]
expertise_level: str
Memory Coordinator
class MemoryCoordinator:
"""
Coordinates all three memory layers.
"""
def __init__(self, episodic, semantic, user_model):
self.episodic = episodic
self.semantic = semantic
self.user_model = user_model
def store_interaction(self, interaction):
"""
Stores an interaction across all memory layers.
"""
user_id = interaction.get('user_id', 'default')
# Store in episodic memory
episode = {
'id': self._generate_id(),
'timestamp': datetime.now().isoformat(),
'user_id': user_id,
'user_input': interaction['user_input'],
'agent_response': interaction['agent_response'],
'context': interaction.get('context', {}),
'outcome': interaction.get('outcome', {}),
'importance': self._calculate_importance(interaction)
}
self.episodic.store_episode(episode)
# Extract and store knowledge
self.semantic.extract_and_store_from_episode(episode)
# Update user model
self._update_user_model(user_id, interaction)
return episode['id']
def retrieve_context(self, user_id, query, max_tokens=4000):
"""
Retrieves comprehensive context for a query.
"""
context_parts = []
current_tokens = 0
# Get user context
user_context = self.user_model.get_user_context(user_id)
user_context_str = f"User Profile: {json.dumps(user_context, indent=2)}"
context_parts.append(user_context_str)
current_tokens += self._count_tokens(user_context_str)
# Get relevant episodes
remaining_tokens = max_tokens - current_tokens
episodes = self.episodic.retrieve_relevant(
query,
k=10,
filters={'user_id': user_id}
)
episode_str = self._format_episodes(episodes, remaining_tokens)
context_parts.append(f"Relevant Past Interactions:\n{episode_str}")
current_tokens += self._count_tokens(episode_str)
# Get relevant knowledge
remaining_tokens = max_tokens - current_tokens
knowledge = self.semantic.query_knowledge(query, k=5)
knowledge_str = self._format_knowledge(knowledge, remaining_tokens)
context_parts.append(f"Relevant Knowledge:\n{knowledge_str}")
return '\n\n'.join(context_parts)
def _calculate_importance(self, interaction):
"""
Calculates the importance of an interaction.
"""
importance = 0.5 # Base importance
# Increase for explicit user feedback
if interaction.get('user_feedback'):
importance += 0.2
# Increase for successful outcomes
if interaction.get('outcome', {}).get('success'):
importance += 0.1
# Increase for complex tasks
if len(interaction.get('actions', [])) > 5:
importance += 0.1
# Increase for new topics (not similar to recent interactions)
if not self._is_similar_to_recent(interaction):
importance += 0.1
return min(importance, 1.0)
Part 5: Nudge System
The Nudge System enables self-prompting for proactive behavior, ensuring memory persistence and triggering autonomous actions.
Architecture
Nudge Generator
class NudgeGenerator:
"""
Generates intelligent self-prompts (nudges) for the agent.
"""
def __init__(self, config, llm_client):
self.config = config
self.llm = llm_client
self.nudge_templates = self._load_templates()
self.context_analyzer = ContextAnalyzer()
def generate_nudge(self, context, nudge_type, priority='normal'):
"""
Generates a nudge based on current context.
Args:
context: Current agent context
nudge_type: Type of nudge to generate
priority: Priority level (low, normal, high, critical)
Returns:
Nudge object with prompt and metadata
"""
# Analyze context
analysis = self.context_analyzer.analyze(context)
# Get base template
template = self.nudge_templates.get(nudge_type)
if not template:
raise InvalidNudgeTypeError(nudge_type)
# Customize template with context
prompt = self._customize_template(template, analysis)
# Generate nudge
nudge = Nudge(
id=self._generate_id(),
type=nudge_type,
prompt=prompt,
priority=priority,
context_summary=analysis,
created_at=datetime.now(),
expires_at=self._calculate_expiry(nudge_type)
)
return nudge
def _load_templates(self):
"""
Loads nudge templates for different types.
"""
return {
'memory_persistence': NudgeTemplate(
name='Memory Persistence',
template="""
[MEMORY PERSISTENCE NUDGE]
Current time: {current_time}
Last memory update: {last_memory_update}
Time since last update: {time_since_update}
Recent interactions: {recent_interactions}
Consider:
1. Are there important experiences that should be persisted?
2. Has user expressed any preferences that should be remembered?
3. Are there any patterns worth noting?
If yes, summarize what should be remembered and why.
""",
default_priority='normal'
),
'skill_creation': NudgeTemplate(
name='Skill Creation Opportunity',
template="""
[SKILL CREATION NUDGE]
Detected pattern: {pattern_description}
Occurrences: {occurrence_count}
Success rate: {success_rate}
Similar interactions:
{similar_interactions}
Consider creating a new skill if:
1. This pattern has occurred at least 3 times
2. Success rate is above 70%
3. No existing skill covers this pattern
If criteria met, propose a skill name and description.
""",
default_priority='high'
),
'learning_reminder': NudgeTemplate(
name='Learning Reminder',
template="""
[LEARNING REMINDER]
Skills created this session: {skills_created}
Patterns identified: {patterns_identified}
Experiences collected: {experiences_collected}
Recent skill performance:
{skill_performance}
Consider:
1. Are there skills that need refinement?
2. Are there new patterns emerging?
3. Is there feedback to incorporate?
""",
default_priority='low'
),
'evolution_initiative': NudgeTemplate(
name='Evolution Initiative',
template="""
[EVOLUTION INITIATIVE NUDGE]
Performance metrics:
{performance_metrics}
Underperforming areas:
{underperforming_areas}
Evolution opportunities:
{evolution_opportunities}
Consider initiating evolution for areas with:
1. Success rate below target
2. User satisfaction declining
3. Error rate increasing
If applicable, propose evolution targets.
""",
default_priority='high'
),
'context_restoration': NudgeTemplate(
name='Context Restoration',
template="""
[CONTEXT RESTORATION NUDGE]
Session start detected.
User: {user_id}
Last session: {last_session_time}
Previous context:
{previous_context}
Active projects:
{active_projects}
Pending tasks:
{pending_tasks}
Consider:
1. Summarizing where we left off
2. Recalling relevant context
3. Resuming any interrupted work
""",
default_priority='high'
)
}
def _customize_template(self, template, analysis):
"""
Customizes a template with context analysis.
"""
prompt = template.template
# Replace placeholders with analysis results
for key, value in analysis.items():
placeholder = '{' + key + '}'
if placeholder in prompt:
prompt = prompt.replace(placeholder, str(value))
# Fill remaining placeholders with defaults
prompt = self._fill_defaults(prompt)
return prompt
@dataclass
class NudgeTemplate:
"""
Template for generating nudges.
"""
name: str
template: str
default_priority: str = 'normal'
required_context: List[str] = field(default_factory=list)
Nudge Scheduler
class NudgeScheduler:
"""
Schedules and manages nudge timing.
"""
def __init__(self, config):
self.config = config
self.scheduled_nudges = []
self.nudge_history = []
self.last_nudge_times = {}
def schedule_nudge(self, nudge_type, trigger_time=None, interval=None):
"""
Schedules a nudge for future execution.
Args:
nudge_type: Type of nudge
trigger_time: Specific time to trigger (optional)
interval: Interval for recurring nudges (optional)
"""
if trigger_time:
scheduled_time = trigger_time
elif interval:
last_time = self.last_nudge_times.get(nudge_type, datetime.now())
scheduled_time = last_time + timedelta(seconds=interval)
else:
# Use default interval from config
default_interval = self.config.get('default_intervals', {}).get(nudge_type, 3600)
scheduled_time = datetime.now() + timedelta(seconds=default_interval)
scheduled_nudge = ScheduledNudge(
nudge_type=nudge_type,
scheduled_time=scheduled_time,
interval=interval,
status='pending'
)
self.scheduled_nudges.append(scheduled_nudge)
return scheduled_nudge
def get_due_nudges(self):
"""
Returns nudges that are due for execution.
"""
now = datetime.now()
due_nudges = []
for scheduled in self.scheduled_nudges:
if scheduled.status == 'pending' and scheduled.scheduled_time <= now:
due_nudges.append(scheduled)
scheduled.status = 'executing'
return due_nudges
def mark_completed(self, nudge_id, result):
"""
Marks a nudge as completed.
"""
for scheduled in self.scheduled_nudges:
if scheduled.id == nudge_id:
scheduled.status = 'completed'
scheduled.completed_at = datetime.now()
scheduled.result = result
# Record in history
self.nudge_history.append({
'nudge_type': scheduled.nudge_type,
'scheduled_time': scheduled.scheduled_time,
'completed_at': scheduled.completed_at,
'result': result
})
# Update last nudge time
self.last_nudge_times[scheduled.nudge_type] = scheduled.completed_at
# Reschedule if recurring
if scheduled.interval:
self.schedule_nudge(
scheduled.nudge_type,
interval=scheduled.interval
)
break
def get_next_nudge_time(self, nudge_type):
"""
Gets the next scheduled time for a nudge type.
"""
for scheduled in self.scheduled_nudges:
if scheduled.nudge_type == nudge_type and scheduled.status == 'pending':
return scheduled.scheduled_time
return None
@dataclass
class ScheduledNudge:
"""
A scheduled nudge.
"""
id: str = field(default_factory=lambda: str(uuid.uuid4()))
nudge_type: str = ''
scheduled_time: datetime = None
interval: int = None
status: str = 'pending'
completed_at: datetime = None
result: Any = None
Nudge Executor
class NudgeExecutor:
"""
Executes nudges and handles their results.
"""
def __init__(self, agent_interface, config):
self.agent = agent_interface
self.config = config
self.execution_log = []
def execute_nudge(self, nudge):
"""
Executes a nudge by presenting it to the agent.
Args:
nudge: The nudge to execute
Returns:
ExecutionResult with agent's response
"""
start_time = datetime.now()
try:
# Present nudge to agent
response = self.agent.process_nudge(nudge)
# Parse response
parsed_response = self._parse_response(response)
# Execute any actions from response
if parsed_response.get('actions'):
action_results = self._execute_actions(parsed_response['actions'])
parsed_response['action_results'] = action_results
result = ExecutionResult(
success=True,
response=parsed_response,
execution_time=(datetime.now() - start_time).total_seconds()
)
except Exception as e:
result = ExecutionResult(
success=False,
error=str(e),
execution_time=(datetime.now() - start_time).total_seconds()
)
# Log execution
self.execution_log.append({
'nudge_id': nudge.id,
'nudge_type': nudge.type,
'result': result,
'timestamp': datetime.now()
})
return result
def _parse_response(self, response):
"""
Parses the agent's response to a nudge.
"""
# Extract structured information from response
parsed = {
'summary': None,
'actions': [],
'memories_to_store': [],
'skills_to_create': [],
'evolution_suggestions': []
}
# Parse based on response structure
if isinstance(response, dict):
parsed.update(response)
elif isinstance(response, str):
# Parse text response
parsed['summary'] = response
# Look for action indicators
if 'create skill' in response.lower():
parsed['skills_to_create'].append(
self._extract_skill_info(response)
)
if 'remember' in response.lower():
parsed['memories_to_store'].append(
self._extract_memory_info(response)
)
return parsed
def _execute_actions(self, actions):
"""
Executes actions from a nudge response.
"""
results = []
for action in actions:
try:
if action['type'] == 'store_memory':
result = self.agent.memory.store(action['data'])
elif action['type'] == 'create_skill':
result = self.agent.skills.create(action['data'])
elif action['type'] == 'trigger_evolution':
result = self.agent.evolution.trigger(action['data'])
else:
result = {'error': f"Unknown action type: {action['type']}"}
results.append({
'action': action,
'result': result,
'success': 'error' not in result
})
except Exception as e:
results.append({
'action': action,
'error': str(e),
'success': False
})
return results
@dataclass
class ExecutionResult:
"""
Result of a nudge execution.
"""
success: bool
response: Any = None
error: str = None
execution_time: float = 0
Nudge Effectiveness Tracker
class NudgeEffectivenessTracker:
"""
Tracks the effectiveness of nudges.
"""
def __init__(self, config):
self.config = config
self.effectiveness_data = {}
def record_outcome(self, nudge_type, outcome):
"""
Records the outcome of a nudge.
"""
if nudge_type not in self.effectiveness_data:
self.effectiveness_data[nudge_type] = {
'total': 0,
'positive': 0,
'negative': 0,
'neutral': 0,
'recent_outcomes': []
}
data = self.effectiveness_data[nudge_type]
data['total'] += 1
if outcome == 'positive':
data['positive'] += 1
elif outcome == 'negative':
data['negative'] += 1
else:
data['neutral'] += 1
# Keep recent outcomes
data['recent_outcomes'].append({
'outcome': outcome,
'timestamp': datetime.now()
})
data['recent_outcomes'] = data['recent_outcomes'][-100:]
def get_effectiveness_score(self, nudge_type):
"""
Gets the effectiveness score for a nudge type.
"""
data = self.effectiveness_data.get(nudge_type, {})
if data.get('total', 0) == 0:
return 0.5 # Default score
return data['positive'] / data['total']
def get_recommendations(self):
"""
Gets recommendations for nudge optimization.
"""
recommendations = []
for nudge_type, data in self.effectiveness_data.items():
score = self.get_effectiveness_score(nudge_type)
if score < 0.3:
recommendations.append({
'nudge_type': nudge_type,
'recommendation': 'Consider disabling or redesigning',
'effectiveness_score': score
})
elif score < 0.5:
recommendations.append({
'nudge_type': nudge_type,
'recommendation': 'Consider adjusting frequency or template',
'effectiveness_score': score
})
return recommendations
def should_skip_nudge(self, nudge_type):
"""
Determines if a nudge should be skipped based on effectiveness.
"""
score = self.get_effectiveness_score(nudge_type)
min_score = self.config.get('min_effectiveness_score', 0.2)
return score < min_score
Complete Nudge System
class NudgeSystem:
"""
The complete nudge system that ties all components together.
"""
def __init__(self, config, agent_interface, llm_client):
self.config = config
self.generator = NudgeGenerator(config, llm_client)
self.scheduler = NudgeScheduler(config)
self.executor = NudgeExecutor(agent_interface, config)
self.tracker = NudgeEffectivenessTracker(config)
self.running = False
def start(self):
"""
Starts the nudge system.
"""
self.running = True
# Schedule default nudges
self._schedule_default_nudges()
# Start main loop
while self.running:
self._process_due_nudges()
time.sleep(self.config.get('check_interval', 60))
def _schedule_default_nudges(self):
"""
Schedules the default set of nudges.
"""
default_intervals = self.config.get('default_intervals', {
'memory_persistence': 300, # 5 minutes
'skill_creation': 600, # 10 minutes
'learning_reminder': 1800, # 30 minutes
'evolution_initiative': 3600, # 1 hour
'context_restoration': 60 # 1 minute
})
for nudge_type, interval in default_intervals.items():
self.scheduler.schedule_nudge(nudge_type, interval=interval)
def _process_due_nudges(self):
"""
Processes all due nudges.
"""
due_nudges = self.scheduler.get_due_nudges()
for scheduled in due_nudges:
# Check if should skip based on effectiveness
if self.tracker.should_skip_nudge(scheduled.nudge_type):
self.scheduler.mark_completed(scheduled.id, {'skipped': True})
continue
# Generate nudge
context = self._get_current_context()
nudge = self.generator.generate_nudge(
context,
scheduled.nudge_type
)
# Execute nudge
result = self.executor.execute_nudge(nudge)
# Track effectiveness
outcome = self._determine_outcome(result)
self.tracker.record_outcome(scheduled.nudge_type, outcome)
# Mark completed
self.scheduler.mark_completed(scheduled.id, result)
def trigger_nudge(self, nudge_type, priority='normal'):
"""
Manually triggers a nudge.
"""
context = self._get_current_context()
nudge = self.generator.generate_nudge(context, nudge_type, priority)
result = self.executor.execute_nudge(nudge)
outcome = self._determine_outcome(result)
self.tracker.record_outcome(nudge_type, outcome)
return result
def _determine_outcome(self, result):
"""
Determines the outcome of a nudge execution.
"""
if not result.success:
return 'negative'
response = result.response
# Check if any actions were taken
if response.get('actions') or response.get('memories_to_store'):
return 'positive'
# Check if there was meaningful content
if response.get('summary') and len(response['summary']) > 50:
return 'positive'
return 'neutral'
Configuration
Complete Configuration Example
# herclaw-agentsystem-config.yaml
# Learning Loop Configuration
learning_loop:
enabled: true
cycle_interval: 3600 # 1 hour
min_experiences: 10
pattern_threshold: 0.7
min_cluster_size: 3
experience_collection:
max_buffer_size: 10000
embedding_model: "text-embedding-3-small"
pattern_extraction:
clustering_algorithm: "dbscan"
eps: 0.3
min_samples: 3
skill_synthesis:
min_confidence: 0.7
auto_register: true
validation:
min_validation_score: 0.8
test_case_count: 5
# Skill Creation Configuration
skill_creation:
enabled: true
opportunity_threshold: 0.6
min_frequency: 3
template_generation:
model: "gpt-4"
temperature: 0.7
validation:
require_user_approval: false
auto_test: true
hub:
enabled: true
url: "https://hub.agentskills.io"
auto_publish: false
auto_install: true
# Self-Evolution Configuration
self_evolution:
enabled: true
evolution_interval: 86400 # 24 hours
rl_pipeline:
learning_rate: 0.00001
clip_range: 0.2
epochs: 4
batch_size: 32
reward_weights:
task_completion: 1.0
user_satisfaction: 0.8
efficiency: 0.5
accuracy: 0.7
safety: 1.0
deployment:
rollout_stages: [0.1, 0.25, 0.5, 1.0]
max_error_rate: 0.05
max_latency: 5000
min_satisfaction: 0.8
validation:
min_validation_score: 0.85
regression_test_suite: "full"
# Persistent Memory Configuration
persistent_memory:
enabled: true
episodic:
collection_name: "episodic_memory"
max_episodes: 100000
importance_threshold: 0.3
semantic:
collection_name: "semantic_memory"
fact_extraction_model: "gpt-4"
min_confidence: 0.6
user_model:
profile_storage: "local"
preference_decay_days: 90
vector_db:
type: "chromadb"
persist_directory: "./memory"
embedding_model: "text-embedding-3-small"
retrieval:
max_context_tokens: 4000
default_k: 10
# Nudge System Configuration
nudge_system:
enabled: true
check_interval: 60 # seconds
default_intervals:
memory_persistence: 300 # 5 minutes
skill_creation: 600 # 10 minutes
learning_reminder: 1800 # 30 minutes
evolution_initiative: 3600 # 1 hour
context_restoration: 60 # 1 minute
effectiveness:
min_effectiveness_score: 0.2
history_size: 100
priorities:
critical: 0
high: 1
normal: 2
low: 3
# Global Settings
global:
model_provider: "openai"
default_model: "gpt-4"
log_level: "INFO"
storage_path: "./herclaw_data"
Usage Examples
Basic Usage
from herclaw_agentsystem import HerClawAgentSystem
# Initialize the complete system
system = HerClawAgentSystem(
config_path="herclaw-agentsystem-config.yaml",
llm_client=openai_client,
storage_path="./data"
)
# Start all subsystems
system.start()
# Process an interaction
result = system.process_interaction(
user_input="Help me analyze the sales data",
context={"project": "Q4 Analysis", "files": ["sales.csv"]}
)
# The system will automatically:
# 1. Store the interaction in memory
# 2. Extract patterns if applicable
# 3. Create skills if patterns emerge
# 4. Evolve based on performance
# 5. Nudge itself for continuous improvement
Manual Control
# Trigger manual learning cycle
system.learning_loop.run_cycle()
