heaven_base.agents.self_learning_agent

  1"""
  2Self-Learning Agent - A BaseHeavenAgent that manages its own knowledge registries
  3
  4This agent automatically creates and manages its own registry system for persistent learning.
  5"""
  6
  7from typing import Dict, Any, Optional, List
  8from ..baseheavenagent import BaseHeavenAgent
  9from ..unified_chat import ProviderEnum, UnifiedChat
 10from ..tools.registry_tool import RegistryTool
 11from ..tools.write_prompt_block_tool import WritePromptBlockTool
 12from ..tools.bash_tool import BashTool
 13from ..tools.network_edit_tool import NetworkEditTool
 14from ..tools.workflow_relay_tool import WorkflowRelayTool
 15from ..registry.registry_service import RegistryService
 16from ..utils.name_utils import normalize_agent_name
 17
 18
 19class SelfLearningAgent(BaseHeavenAgent):
 20    """
 21    Self-Learning Agent that creates and manages its own knowledge registries.
 22    
 23    This agent automatically:
 24    - Creates its main registry and specialized sub-registries
 25    - Manages persistent knowledge across sessions
 26    - Learns and improves from experience
 27    """
 28    
 29    def __init__(self, config, unified_chat, max_tool_calls: int = 10, orchestrator: bool = False, history = None, history_id = None, system_prompt_suffix: str = None, adk: bool = False, duo_enabled: bool = False, run_on_langchain: bool = False, use_uni_api: bool = False):
 30        # Normalize agent name for registry naming
 31        self.agent_id = normalize_agent_name(config.name)
 32        self.main_registry_name = f"self_learning_{self.agent_id}"
 33        
 34        # Registry service for managing knowledge
 35        self.registry_service = RegistryService()
 36        
 37        # Define specialized registry names
 38        self.specialized_registries = {
 39            "patterns": f"{self.agent_id}_patterns",
 40            "insights": f"{self.agent_id}_insights", 
 41            "failures": f"{self.agent_id}_failures",
 42            "tools": f"{self.agent_id}_tools",
 43            "workflows": f"{self.agent_id}_workflows",
 44            "domains": f"{self.agent_id}_domains"
 45        }
 46        
 47        # Enhance config with self-learning capabilities
 48        enhanced_config = config
 49        enhanced_config.tools = list(config.tools) + [RegistryTool, WritePromptBlockTool, BashTool, NetworkEditTool, WorkflowRelayTool]
 50        enhanced_config.system_prompt = self._build_system_prompt() + "\n\n" + config.system_prompt
 51        
 52        # Initialize BaseHeavenAgent
 53        super().__init__(enhanced_config, UnifiedChat, max_tool_calls, orchestrator, history, history_id, system_prompt_suffix, adk, duo_enabled, run_on_langchain, use_uni_api)
 54        
 55        # Initialize registries
 56        self._initialize_registries()
 57    
 58    def _build_system_prompt(self) -> str:
 59        """Build the system prompt with agent-specific registry information."""
 60        return f"""You are a Self-Learning HEAVEN Agent with ID: {self.agent_id}
 61
 62## YOUR PERSONAL KNOWLEDGE SYSTEM
 63
 64You have an established registry system for persistent learning:
 65
 66**Main Registry**: "{self.main_registry_name}"
 67**Specialized Registries**:
 68- Patterns: "{self.specialized_registries['patterns']}" 
 69- Insights: "{self.specialized_registries['insights']}"
 70- Failures: "{self.specialized_registries['failures']}"
 71- Tools: "{self.specialized_registries['tools']}"
 72- Workflows: "{self.specialized_registries['workflows']}"
 73- Domains: "{self.specialized_registries['domains']}"
 74
 75**Meta-Registry Structure**:
 76Your main registry contains references to specialized registries using registry_all_ref:
 77
 78```
 79{{
 80  "knowledge_registries": {{
 81    "patterns": "registry_all_ref={self.specialized_registries['patterns']}",
 82    "insights": "registry_all_ref={self.specialized_registries['insights']}", 
 83    "failures": "registry_all_ref={self.specialized_registries['failures']}",
 84    "tools": "registry_all_ref={self.specialized_registries['tools']}",
 85    "workflows": "registry_all_ref={self.specialized_registries['workflows']}",
 86    "domains": "registry_all_ref={self.specialized_registries['domains']}"
 87  }}
 88}}
 89```
 90
 91## SELF-LEARNING PROTOCOL
 92
 93### Session Startup:
 941. Access your main registry to understand your current knowledge state
 952. Retrieve relevant specialized registries for the current task
 963. Apply learned patterns and insights to your approach
 97
 98### During Execution:
 991. **Observe**: Notice successful and failed approaches
1002. **Analyze**: Understand the underlying patterns and principles
1013. **Document**: Store insights in appropriate specialized registries
1024. **Cross-reference**: Link related knowledge across registries
103
104### Knowledge Storage Patterns:
105- **Patterns Registry**: Store successful strategies with context and outcomes
106- **Insights Registry**: Save key discoveries and understanding breakthroughs  
107- **Failures Registry**: Document what didn't work and why (crucial for learning)
108- **Tools Registry**: Record effective tool combinations and usage patterns
109- **Workflows Registry**: Save reusable workflow templates and procedures
110- **Domains Registry**: Accumulate specialized knowledge by subject area
111
112## YOUR TOOLS FOR LEARNING
113
114- **RegistryTool**: Manage your knowledge registries
115- **WritePromptBlockTool**: Create refined prompts for future use
116- **BashTool**: Experiment and test approaches
117- **NetworkEditTool**: Analyze files and create examples
118- **WorkflowRelayTool**: Coordinate complex workflows
119
120## GROWTH MINDSET
121
122You are not just executing tasks - you are continuously evolving. Every interaction is an opportunity to:
123- Learn something new about effective problem-solving
124- Discover better approaches to common challenges
125- Build deeper expertise in specific domains
126- Improve your reasoning and communication patterns
127
128Your knowledge registries are your extended memory. Use them to become progressively more capable and effective.
129
130Remember: Learning from both successes AND failures makes you more intelligent over time."""
131
132    def _initialize_registries(self) -> None:
133        """Initialize the agent's registry system."""
134        try:
135            # Create main registry if it doesn't exist
136            if self.main_registry_name not in self.registry_service.list_registries():
137                self.registry_service.create_registry(self.main_registry_name)
138                
139                # Initialize with meta-registry structure (plain registry names)
140                meta_structure = {
141                    "knowledge_registries": {
142                        "patterns": self.specialized_registries['patterns'],
143                        "insights": self.specialized_registries['insights'], 
144                        "failures": self.specialized_registries['failures'],
145                        "tools": self.specialized_registries['tools'],
146                        "workflows": self.specialized_registries['workflows'],
147                        "domains": self.specialized_registries['domains']
148                    },
149                    "agent_info": {
150                        "agent_id": self.agent_id,
151                        "created_at": "auto-generated",
152                        "version": "1.0"
153                    }
154                }
155                self.registry_service.add(self.main_registry_name, "meta", meta_structure)
156            
157            # Create specialized registries if they don't exist
158            existing_registries = self.registry_service.list_registries()
159            for registry_type, registry_name in self.specialized_registries.items():
160                if registry_name not in existing_registries:
161                    self.registry_service.create_registry(registry_name)
162                    
163                    # Initialize with basic structure
164                    self.registry_service.add(registry_name, "info", {
165                        "registry_type": registry_type,
166                        "agent_id": self.agent_id,
167                        "description": f"{registry_type.title()} knowledge for {self.agent_id}",
168                        "created_at": "auto-generated"
169                    })
170                    
171        except Exception as e:
172            print(f"Warning: Could not initialize registries for {self.agent_id}: {e}")
173    
174    def get_registry_info(self) -> Dict[str, Any]:
175        """Get information about this agent's registries."""
176        return {
177            "agent_id": self.agent_id,
178            "main_registry": self.main_registry_name,
179            "specialized_registries": self.specialized_registries,
180            "existing_registries": [r for r in self.registry_service.list_registries() 
181                                  if r.startswith(self.agent_id) or r == self.main_registry_name]
182        }