Merge pull request #138 from Dooders/agent-farm-converter

Agent farm converter

Author: csmangum

TEST_CHANGELOG.md

@@ -0,0 +1,13 @@
+"""
+AgentFarm DB to Memory System Converter
+
+This module provides functionality to convert data from an AgentFarm SQLite database
+into a memory system, handling the import of agents and their associated memories.
+"""
+
+__version__ = "0.1.0"
+
+from .config import DEFAULT_CONFIG
+from .converter import from_agent_farm
+
+__all__ = ["from_agent_farm", "DEFAULT_CONFIG"] 

converter/__init__.py

@@ -0,0 +1,173 @@
+"""
+Agent import system for the AgentFarm DB to Memory System converter.
+"""
+
+import logging
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+from .config import ConverterConfig
+from .db import DatabaseManager
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class AgentMetadata:
+    """Metadata for an imported agent."""
+
+    agent_id: str
+    name: str
+    metadata: Dict[str, Any]
+    created_at: str
+    updated_at: str
+
+
+class AgentImporter:
+    """
+    Handles the import of agents from AgentFarm database to memory system.
+
+    This class manages the process of importing agents, including validation,
+    metadata preservation, and error handling.
+    """
+
+    def __init__(self, db_manager: DatabaseManager, config: ConverterConfig):
+        """
+        Initialize the agent importer.
+
+        Args:
+            db_manager: Database manager instance
+            config: Converter configuration
+        """
+        self.db_manager = db_manager
+        self.config = config
+
+    def import_agents(self) -> List[AgentMetadata]:
+        """
+        Import agents from the database.
+
+        Returns:
+            List of imported agent metadata
+
+        Raises:
+            ValueError: If agent validation fails and error_handling is 'fail'
+        """
+        agents = []
+        with self.db_manager.session() as session:
+            # Get agent query based on import mode
+            query = self._get_agent_query(session)
+
+            # Process agents in batches
+            for batch in self._batch_query(query):
+                for agent in batch:
+                    try:
+                        agent_metadata = self._import_agent(agent)
+                        agents.append(agent_metadata)
+                    except Exception as e:
+                        self._handle_import_error(e, agent)
+
+        return agents
+
+    def _get_agent_query(self, session):
+        """Get the appropriate agent query based on import mode."""
+        query = session.query(self.db_manager.AgentModel)
+
+        if self.config.import_mode == "incremental":
+            # Add incremental import conditions
+            pass
+
+        if self.config.selective_agents:
+            query = query.filter(
+                self.db_manager.AgentModel.agent_id.in_(self.config.selective_agents)
+            )
+
+        return query
+
+    def _batch_query(self, query):
+        """Process query in batches."""
+        offset = 0
+        while True:
+            batch = query.offset(offset).limit(self.config.batch_size).all()
+            if not batch:
+                break
+            yield batch
+            offset += self.config.batch_size
+
+    def _import_agent(self, agent) -> AgentMetadata:
+        """
+        Import a single agent.
+
+        Args:
+            agent: Agent model instance
+
+        Returns:
+            AgentMetadata instance
+
+        Raises:
+            ValueError: If agent validation fails
+        """
+        # Validate agent
+        if self.config.validate:
+            self._validate_agent(agent)
+
+        # Create agent metadata
+        metadata = AgentMetadata(
+            agent_id=agent.agent_id,
+            # Use agent_id as the name if agent doesn't have a name attribute
+            name=getattr(agent, "name", f"Agent-{agent.agent_id}"),
+            metadata=self._extract_agent_metadata(agent),
+            created_at=str(agent.birth_time),
+            updated_at=str(agent.death_time or agent.birth_time),
+        )
+
+        return metadata
+
+    def _validate_agent(self, agent):
+        """
+        Validate an agent.
+
+        Args:
+            agent: Agent model instance
+
+        Raises:
+            ValueError: If validation fails
+        """
+        if not agent.agent_id:
+            raise ValueError("Agent must have an ID")
+
+    def _extract_agent_metadata(self, agent) -> Dict[str, Any]:
+        """
+        Extract metadata from an agent.
+
+        Args:
+            agent: Agent model instance
+
+        Returns:
+            Dictionary of agent metadata
+        """
+        return {
+            "type": agent.agent_type,
+            "position": {"x": agent.position_x, "y": agent.position_y},
+            "initial_resources": agent.initial_resources,
+            "starting_health": agent.starting_health,
+            "starvation_threshold": agent.starvation_threshold,
+            "genome_id": agent.genome_id,
+            "generation": agent.generation,
+            "action_weights": agent.action_weights,
+        }
+
+    def _handle_import_error(self, error: Exception, agent: Any):
+        """
+        Handle agent import error based on configuration.
+
+        Args:
+            error: The error that occurred
+            agent: The agent that caused the error
+        """
+        error_msg = f"Error importing agent {agent.agent_id}: {str(error)}"
+
+        if self.config.error_handling == "fail":
+            raise ValueError(error_msg)
+        elif self.config.error_handling == "log":
+            logger.error(error_msg)
+        # Skip mode just continues without raising or logging

converter/agent_import.py

@@ -0,0 +1,106 @@
+"""
+Configuration system for the AgentFarm DB to Memory System converter.
+"""
+
+import logging
+from dataclasses import dataclass, field
+from typing import Dict, Optional, List, Union
+
+from .tiering import TieringStrategy, create_tiering_strategy
+from .mapping import MemoryTypeMapper
+
+logger = logging.getLogger(__name__)
+
+@dataclass
+class ConverterConfig:
+    """Configuration for the AgentFarm DB to Memory System converter."""
+    
+    # Redis configuration
+    use_mock_redis: bool = True
+    
+    # Validation settings
+    validate: bool = True
+    error_handling: str = "skip"  # One of: "skip", "fail", "log"
+    
+    # Processing settings
+    batch_size: int = 100
+    show_progress: bool = True
+    
+    # Memory type mapping
+    memory_type_mapping: Dict[str, str] = field(default_factory=lambda: {
+        'AgentStateModel': 'state',
+        'ActionModel': 'action',
+        'SocialInteractionModel': 'interaction'
+    })
+    memory_type_mapper: Optional[MemoryTypeMapper] = None
+    
+    # Tiering strategy
+    tiering_strategy_type: str = "step_based"  # One of: "step_based", "importance_aware"
+    tiering_strategy: Optional[TieringStrategy] = None
+    
+    # Import settings
+    import_mode: str = "full"  # One of: "full", "incremental"
+    selective_agents: Optional[List[int]] = None
+    total_steps: Optional[int] = None  # Total number of steps in the simulation
+    
+    def __post_init__(self):
+        """Validate configuration after initialization."""
+        self._validate_error_handling()
+        self._validate_import_mode()
+        self._validate_batch_size()
+        self._validate_tiering_strategy()
+        
+        # Initialize memory type mapper
+        if self.memory_type_mapper is None:
+            self.memory_type_mapper = MemoryTypeMapper(mapping=self.memory_type_mapping)
+        
+        # Initialize tiering strategy if not provided
+        if self.tiering_strategy is None:
+            self.tiering_strategy = create_tiering_strategy(self.tiering_strategy_type)
+        
+    def _validate_error_handling(self):
+        """Validate error handling mode."""
+        if self.error_handling not in ["skip", "fail", "log"]:
+            raise ValueError(
+                f"Invalid error_handling mode: {self.error_handling}. "
+                "Must be one of: skip, fail, log"
+            )
+            
+    def _validate_import_mode(self):
+        """Validate import mode."""
+        if self.import_mode not in ["full", "incremental"]:
+            raise ValueError(
+                f"Invalid import_mode: {self.import_mode}. "
+                "Must be one of: full, incremental"
+            )
+            
+    def _validate_batch_size(self):
+        """Validate batch size."""
+        if self.batch_size < 1:
+            raise ValueError("batch_size must be greater than 0")
+            
+    def _validate_tiering_strategy(self):
+        """Validate tiering strategy settings."""
+        valid_types = ["step_based", "importance_aware"]
+        if self.tiering_strategy_type not in valid_types:
+            raise ValueError(
+                f"Invalid tiering_strategy_type: {self.tiering_strategy_type}. "
+                f"Must be one of: {valid_types}"
+            )
+
+# Default configuration
+DEFAULT_CONFIG = {
+    'use_mock_redis': True,
+    'validate': True,
+    'error_handling': 'skip',
+    'batch_size': 100,
+    'show_progress': True,
+    'memory_type_mapping': {
+        'AgentStateModel': 'state',
+        'ActionModel': 'action',
+        'SocialInteractionModel': 'interaction'
+    },
+    'tiering_strategy_type': 'step_based',
+    'import_mode': 'full',
+    'selective_agents': None
+}