RAVANA AGI

Action Registry

Action Registry

Update Summary

Changes Made

  • Added documentation for the newly integrated CollaborativeTaskAction
  • Updated the list of built-in actions to include CollaborativeTaskAction
  • Added section sources for newly referenced files
  • Updated the Action Registry Overview section to reflect the new action registration
  • Enhanced troubleshooting section with additional error scenarios

Table of Contents

  1. Introduction
  2. Action Registry Overview
  3. Internal Mapping Structure
  4. Registration Mechanism and Validation
  5. Automatic Action Discovery
  6. Integration with ActionManager
  7. Registering Custom Actions
  8. Troubleshooting Common Issues

Introduction

The Action Registry is a central component of the Ravana AGI system responsible for managing all executable actions. It provides a dynamic registry that allows the system to discover, register, and retrieve actions at runtime. This document provides a comprehensive analysis of the ActionRegistry implementation, its integration with the broader system, and best practices for extending its functionality with custom actions.

Action Registry Overview

The ActionRegistry serves as a centralized repository for all actions that the AGI can perform. It enables the system to dynamically discover and register action classes, making it easy to extend the AGI's capabilities without modifying core system components.

Diagram sources

Section sources

Internal Mapping Structure

The ActionRegistry maintains an internal dictionary that maps action names to their corresponding callable class instances. This structure enables efficient lookup and retrieval of actions by name.

The primary data structure is defined as:

self.actions: Dict[str, Action] = {}

This dictionary serves as the core registry, where:

  • Keys: String identifiers representing the unique name of each action
  • Values: Instantiated Action objects that can be executed

The mapping structure supports O(1) average-case time complexity for both registration and retrieval operations, making it highly efficient for the AGI system's needs.

Diagram sources

Section sources

Registration Mechanism and Validation

The ActionRegistry implements a robust registration mechanism that handles both explicit and automatic registration of action classes. The system includes validation processes to ensure action integrity and handle edge cases.

Registration Methods

The registry provides two primary methods for registering actions:

  1. Direct Registration: Using the register_action() method to register individual action instances
  2. Bulk Registration: Using the discover_actions() method to automatically discover and register all actions in the core.actions package

The registration process includes validation for duplicate entries:

def _register_action(self, action: Action) -> None:
    if action.name in self.actions:
        logger.warning(f"Action '{action.name}' is already registered. Overwriting.")
    self.actions[action.name] = action

When a duplicate action name is detected, the system logs a warning and overwrites the existing entry, ensuring that the most recent action definition takes precedence.

Error Handling for Invalid Entries

The registry handles invalid entries through exception handling during the discovery process:

try:
    instance = obj()
    if instance.name in self.actions:
        logger.warning(f"Action '{instance.name}' is already registered. Overwriting.")
    self.actions[instance.name] = instance
except Exception as e:
    logger.error(f"Failed to instantiate action {obj.__name__}: {e}", exc_info=True)

This try-except block ensures that failures in instantiating individual actions do not prevent the registration of other valid actions, maintaining system resilience.

Diagram sources

Section sources

Automatic Action Discovery

The ActionRegistry implements an automatic discovery mechanism that scans the core.actions package to find and register all action classes without requiring explicit registration calls.

Discovery Process

The discover_actions() method uses Python's introspection capabilities to automatically detect and register action classes:

def discover_actions(self):
    """Discovers and registers all actions in the 'core.actions' package."""
    actions_package = core.actions
    for _, name, is_pkg in pkgutil.walk_packages(actions_package.__path__, actions_package.__name__ + '.'):
        if not is_pkg:
            module = importlib.import_module(name)
            for _, obj in inspect.getmembers(module):
                if inspect.isclass(obj) and issubclass(obj, Action) and obj is not Action:
                    try:
                        instance = obj()
                        if instance.name in self.actions:
                            logger.warning(f"Action '{instance.name}' is already registered. Overwriting.")
                        self.actions[instance.name] = instance
                    except Exception as e:
                        logger.error(f"Failed to instantiate action {obj.__name__}: {e}", exc_info=True)

The discovery process follows these steps:

  1. Traverse all modules in the core.actions package using pkgutil.walk_packages
  2. Import each non-package module
  3. Inspect the module to find all classes that inherit from the Action base class
  4. Instantiate each action class and register it in the internal dictionary

Integration with System Initialization

The automatic discovery is integrated into the system's initialization process through the ActionManager:

class ActionManager:
    def __init__(self, system: 'AGISystem', data_service: 'DataService'):
        self.system = system
        self.data_service = data_service
        self.action_registry = ActionRegistry(system, data_service)
        logger.info(f"ActionManager initialized with {len(self.action_registry.actions)} actions.")
        self.log_available_actions()

During initialization, the ActionManager creates an ActionRegistry instance, which automatically registers several built-in actions in its constructor, and then the system can call discover_actions() to find additional actions.

Diagram sources

Section sources

Integration with ActionManager

The ActionRegistry is tightly integrated with the ActionManager, which serves as the interface between the decision-making components and the execution of actions.

Execution Flow

When the DecisionEngine determines an action to perform, the ActionManager uses the ActionRegistry to retrieve and execute the appropriate action:

async def execute_action(self, decision: Dict[str, Any]) -> Dict[str, Any]:
    # ... parsing logic ...
    
    try:
        action_name = action_data.get("action")
        action_params = action_data.get("params", {})

        if not action_name:
            logger.warning("No 'action' key found in the parsed JSON.")
            return {"error": "No action taken: 'action' key missing."}

        action = self.action_registry.get_action(action_name)
        if not action:
            raise ActionException(f"Action '{action_name}' not found.")

        logger.info(f"Executing action '{action_name}' with params: {action_params}")
        result = await action.execute(**action_params)
        # ... logging and return logic ...

Enhanced Action Manager Integration

The EnhancedActionManager extends the base ActionManager to register additional multi-modal actions:

class EnhancedActionManager(ActionManager):
    def __init__(self, agi_system, data_service):
        super().__init__(agi_system, data_service)
        self.multi_modal_service = MultiModalService()
        self.action_cache = {}
        self.parallel_limit = 3
        self.register_enhanced_actions()
        
    def register_enhanced_actions(self):
        """Register new multi-modal actions as Action instances."""
        self.action_registry.register_action(ProcessImageAction(self.system, self.data_service))
        self.action_registry.register_action(ProcessAudioAction(self.system, self.data_service))
        self.action_registry.register_action(AnalyzeDirectoryAction(self.system, self.data_service))
        self.action_registry.register_action(CrossModalAnalysisAction(self.system, self.data_service))

This demonstrates how the ActionRegistry can be extended with specialized actions while maintaining the same interface.

Diagram sources

Section sources

Registering Custom Actions

Developers can extend the AGI's capabilities by creating custom actions that inherit from the Action base class.

Built-in Action Examples

The system includes several built-in actions registered during initialization:

  • ProposeAndTestInventionAction: For proposing and testing new inventions
  • LogMessageAction: For logging messages to the system
  • WritePythonCodeAction: For writing Python code files
  • ExecutePythonFileAction: For executing Python files
  • BlogPublishAction: For publishing blog posts
  • CollaborativeTaskAction: For managing collaborative tasks between RAVANA and users with feedback mechanisms

These are registered in the ActionRegistry constructor:

def __init__(self,
             system: 'AGISystem',
             data_service: 'DataService'
             ) -> None:
    self.actions: Dict[str, Action] = {}
    self._register_action(ProposeAndTestInventionAction(system, data_service))
    self._register_action(LogMessageAction(system, data_service))
    self._register_action(WritePythonCodeAction(system, data_service))
    self._register_action(ExecutePythonFileAction(system, data_service))
    self._register_action(BlogPublishAction(system, data_service))
    self._register_action(CollaborativeTaskAction(system, data_service))

Creating Custom Actions

To create a custom action, follow these steps as documented in the DEVELOPER_GUIDE.md:

  1. Create a new Python file in the core/actions/ directory
  2. Define a class that inherits from core.actions.action.Action
  3. Implement the required properties and methods

Example of a custom "hello_world" action:

from core.actions.action import Action
from typing import Any, Dict, List

class HelloWorldAction(Action):
    @property
    def name(self) -> str:
        return "hello_world"

    @property
    def description(self) -> str:
        return "A simple action that prints a greeting."

    @property
    def parameters(self) -> List[Dict[str, Any]]:
        return [
            {
                "name": "name",
                "type": "string",
                "description": "The name to include in the greeting.",
                "required": True,
            }
        ]

    async def execute(self, **kwargs: Any) -> Any:
        name = kwargs.get("name")
        return f"Hello, {name}!"

Once created, the action will be automatically discovered and registered when the discover_actions() method is called, as it scans all modules in the core.actions package for classes that inherit from the Action base class.

Diagram sources

Section sources

Troubleshooting Common Issues

This section addresses common issues developers may encounter when working with the ActionRegistry and provides guidance for resolution.

Import-Time Registration Failures

Issue: Actions are not being discovered during the import process.

Causes and Solutions:

  • Circular imports: Ensure that action modules do not create circular import dependencies. Use lazy imports or refactor code structure.
  • Missing init.py: Verify that all directories in the core/actions/ hierarchy contain an __init__.py file to be recognized as packages.
  • Syntax errors: Check for syntax errors in action modules, as these will prevent successful import and discovery.

Namespace Conflicts

Issue: Multiple actions with the same name cause overwriting.

Behavior: When duplicate action names are detected, the system logs a warning and overwrites the existing entry:

if action.name in self.actions:
    logger.warning(f"Action '{action.name}' is already registered. Overwriting.")

Best Practices:

  • Use unique, descriptive names for custom actions
  • Follow a consistent naming convention (e.g., verb_noun pattern)
  • Check existing action names before creating new ones using get_all_actions()

Dynamic Reloading Limitations

Issue: The current implementation does not support hot-reloading of actions during runtime.

Current Behavior: Actions are registered during system initialization and discovery. Changes to action code require a system restart to take effect.

Workarounds:

  • Implement a custom reload mechanism that calls discover_actions() again
  • Use the register_action() method to add new actions at runtime
  • Design actions to be configurable rather than requiring code changes

Common Error Scenarios

Error Scenario Symptoms Resolution
Invalid action parameters InvalidActionParams exception during execution Validate parameters using the action's validate_params() method before execution
Action not found ValueError when calling get_action() Verify the action name is correct and the action is properly registered
Action instantiation failure Error logged during discovery process Check that the action class can be instantiated with the required parameters
Module import failure Action not discovered, import error in logs Verify module path and dependencies
Collaborative task action failure Error in conversational AI integration Verify system references and user ID parameters

Section sources

Referenced Files in This Document