Environment API Reference¶

class Environment:
    """
    Base environment for cognitive simulations.

    Manages agent interactions, environmental dynamics, resources,
    and provides the context for cognitive agent behavior.
    """

    def __init__(
        self,
        env_id: str,
        environment_type: str = "basic",
        capacity: int = 100,
        resources: Optional[List[str]] = None,
        dynamics: Optional[Dict[str, Any]] = None
    ):
        """
        Initialize environment.

        Args:
            env_id: Unique environment identifier
            environment_type: Type of environment
            capacity: Maximum number of agents
            resources: Available resources
            dynamics: Environmental dynamics configuration
        """

def add_agent(self, agent: CognitiveAgent) -> bool:
    """
    Add an agent to the environment.

    Args:
        agent: Cognitive agent to add

    Returns:
        bool: True if agent added successfully

    Raises:
        EnvironmentError: If agent cannot be added
    """

def remove_agent(self, agent_id: str) -> bool:
    """
    Remove an agent from the environment.

    Args:
        agent_id: ID of agent to remove

    Returns:
        bool: True if agent removed successfully
    """

def step(self) -> EnvironmentState:
    """
    Execute one environment time step.

    Returns:
        EnvironmentState: New state after step execution

    Raises:
        EnvironmentError: If step execution fails
    """

def get_percepts_for_agent(self, agent_id: str) -> List[Percept]:
    """
    Get perceptual information for specific agent.

    Args:
        agent_id: ID of agent requesting percepts

    Returns:
        List[Percept]: Available perceptual information
    """

from cognito_sim_engine import Environment, CognitiveAgent

# Create basic environment
env = Environment(
    env_id="research_lab",
    environment_type="collaborative",
    capacity=50,
    resources=["computing_cluster", "research_papers", "collaboration_tools"],
    dynamics={
        "knowledge_sharing_rate": 0.1,
        "resource_availability": 0.8,
        "collaboration_encouragement": 0.7
    }
)

# Add agents to environment
researcher_1 = CognitiveAgent("researcher_001")
researcher_2 = CognitiveAgent("researcher_002")

env.add_agent(researcher_1)
env.add_agent(researcher_2)

print(f"Environment has {len(env.get_agents())} agents")

# Environment simulation loop
for step in range(100):
    # Environment step
    env_state = env.step()

    # Get percepts for each agent
    for agent in env.get_agents():
        percepts = env.get_percepts_for_agent(agent.agent_id)
        # Agent processes percepts and acts
        agent.process_percepts(percepts)

    print(f"Step {step}: Environment state updated")

def get_state(self) -> EnvironmentState:
    """
    Get current environment state.

    Returns:
        EnvironmentState: Complete environment state
    """

def set_state(self, state: EnvironmentState) -> None:
    """
    Set environment state.

    Args:
        state: New environment state

    Raises:
        EnvironmentError: If state cannot be set
    """

def get_agents(self) -> List[CognitiveAgent]:
    """
    Get all agents in environment.

    Returns:
        List[CognitiveAgent]: All agents in environment
    """

def get_resources(self) -> Dict[str, Resource]:
    """
    Get available resources.

    Returns:
        Dict[str, Resource]: Available resources mapped by name
    """

def execute_action(self, agent_id: str, action: Action) -> ActionResult:
    """
    Execute agent action in environment.

    Args:
        agent_id: ID of acting agent
        action: Action to execute

    Returns:
        ActionResult: Result of action execution

    Raises:
        EnvironmentError: If action cannot be executed
    """

def facilitate_interaction(
    self,
    initiator_id: str,
    target_id: str,
    interaction_type: str,
    interaction_data: Dict[str, Any]
) -> InteractionResult:
    """
    Facilitate interaction between agents.

    Args:
        initiator_id: ID of initiating agent
        target_id: ID of target agent  
        interaction_type: Type of interaction
        interaction_data: Interaction parameters

    Returns:
        InteractionResult: Result of interaction
    """

class CollaborativeEnvironment(Environment):
    """
    Environment optimized for collaborative agent interactions.

    Supports knowledge sharing, joint problem-solving, and
    collaborative learning with sophisticated communication channels.
    """

    def __init__(
        self,
        env_id: str,
        collaboration_mechanisms: List[str] = None,
        knowledge_sharing_enabled: bool = True,
        communication_channels: List[str] = None,
        **kwargs
    ):
        """
        Initialize collaborative environment.

        Args:
            env_id: Environment identifier
            collaboration_mechanisms: Available collaboration mechanisms
            knowledge_sharing_enabled: Enable knowledge sharing
            communication_channels: Available communication channels
        """

def enable_knowledge_sharing(
    self,
    sharing_rate: float = 0.1,
    knowledge_types: List[str] = None,
    sharing_mechanisms: List[str] = None
) -> None:
    """
    Enable knowledge sharing between agents.

    Args:
        sharing_rate: Rate of knowledge sharing
        knowledge_types: Types of knowledge to share
        sharing_mechanisms: Mechanisms for sharing
    """

def create_collaboration_group(
    self,
    group_id: str,
    member_agents: List[str],
    collaboration_goal: str,
    coordination_strategy: str = "democratic"
) -> CollaborationGroup:
    """
    Create a collaboration group.

    Args:
        group_id: Group identifier
        member_agents: IDs of member agents
        collaboration_goal: Goal of collaboration
        coordination_strategy: Strategy for coordination

    Returns:
        CollaborationGroup: Created collaboration group
    """

def facilitate_peer_learning(
    self,
    learning_topic: str,
    participant_agents: List[str],
    learning_structure: str = "discussion"
) -> PeerLearningResult:
    """
    Facilitate peer learning session.

    Args:
        learning_topic: Topic for peer learning
        participant_agents: Participating agents
        learning_structure: Structure of learning session

    Returns:
        PeerLearningResult: Results of peer learning
    """

from cognito_sim_engine import CollaborativeEnvironment, CollaborativeAgent

# Create collaborative environment
collab_env = CollaborativeEnvironment(
    env_id="research_collaboration_lab",
    collaboration_mechanisms=["knowledge_sharing", "joint_problem_solving", "peer_review"],
    knowledge_sharing_enabled=True,
    communication_channels=["direct_messaging", "group_discussion", "presentation"]
)

# Configure knowledge sharing
collab_env.enable_knowledge_sharing(
    sharing_rate=0.15,
    knowledge_types=["research_findings", "methodologies", "insights"],
    sharing_mechanisms=["automatic", "on_request", "periodic"]
)

# Add collaborative agents
collaborators = []
for i in range(5):
    agent = CollaborativeAgent(
        agent_id=f"researcher_{i:03d}",
        collaboration_style="cooperative",
        communication_enabled=True,
        theory_of_mind=True
    )
    collab_env.add_agent(agent)
    collaborators.append(agent)

# Create research collaboration group
research_group = collab_env.create_collaboration_group(
    group_id="ai_research_team",
    member_agents=[agent.agent_id for agent in collaborators],
    collaboration_goal="develop_novel_ai_architecture",
    coordination_strategy="expertise_based"
)

# Facilitate peer learning
peer_learning_result = collab_env.facilitate_peer_learning(
    learning_topic="transformer_architectures",
    participant_agents=[agent.agent_id for agent in collaborators[:3]],
    learning_structure="structured_discussion"
)

print(f"Peer learning effectiveness: {peer_learning_result.learning_effectiveness:.2f}")
print(f"Knowledge gained: {peer_learning_result.total_knowledge_gained:.2f}")

class LearningEnvironment(Environment):
    """
    Environment specialized for learning and education.

    Provides curriculum management, assessment systems, and
    adaptive learning support for educational simulations.
    """

    def __init__(
        self,
        env_id: str,
        curriculum: Optional[Curriculum] = None,
        assessment_system: str = "adaptive",
        learning_analytics: bool = True,
        personalization_enabled: bool = True,
        **kwargs
    ):
        """
        Initialize learning environment.

        Args:
            env_id: Environment identifier
            curriculum: Learning curriculum
            assessment_system: Assessment system type
            learning_analytics: Enable learning analytics
            personalization_enabled: Enable personalized learning
        """

def deliver_instruction(
    self,
    learner_id: str,
    instruction: Instruction,
    delivery_method: str = "adaptive"
) -> InstructionResult:
    """
    Deliver instruction to learner.

    Args:
        learner_id: ID of learning agent
        instruction: Instruction content
        delivery_method: Method of delivery

    Returns:
        InstructionResult: Result of instruction delivery
    """

def assess_learning(
    self,
    learner_id: str,
    assessment_type: str = "formative",
    topics: List[str] = None
) -> AssessmentResult:
    """
    Assess learner's knowledge and skills.

    Args:
        learner_id: ID of learning agent
        assessment_type: Type of assessment
        topics: Topics to assess

    Returns:
        AssessmentResult: Assessment results
    """

def adapt_curriculum(
    self,
    learner_id: str,
    performance_data: PerformanceData,
    adaptation_strategy: str = "difficulty_adjustment"
) -> CurriculumAdaptation:
    """
    Adapt curriculum based on learner performance.

    Args:
        learner_id: ID of learning agent
        performance_data: Learner's performance data
        adaptation_strategy: Strategy for adaptation

    Returns:
        CurriculumAdaptation: Curriculum adaptations made
    """

from cognito_sim_engine import LearningEnvironment, LearningAgent, Curriculum

# Create machine learning curriculum
ml_curriculum = Curriculum(
    curriculum_id="ml_fundamentals",
    topics=[
        "linear_regression",
        "logistic_regression", 
        "neural_networks",
        "deep_learning",
        "evaluation_metrics"
    ],
    prerequisites={
        "logistic_regression": ["linear_regression"],
        "neural_networks": ["linear_regression", "logistic_regression"],
        "deep_learning": ["neural_networks"]
    },
    difficulty_progression="adaptive"
)

# Create learning environment
learning_env = LearningEnvironment(
    env_id="ml_classroom",
    curriculum=ml_curriculum,
    assessment_system="continuous",
    learning_analytics=True,
    personalization_enabled=True
)

# Add learning agents (students)
students = []
for i in range(20):
    student = LearningAgent(
        agent_id=f"student_{i:03d}",
        learning_rate=random.uniform(0.01, 0.05),
        learning_style=random.choice(["visual", "auditory", "kinesthetic"]),
        prior_knowledge=random.uniform(0.1, 0.3)
    )
    learning_env.add_agent(student)
    students.append(student)

# Simulation of learning process
for week in range(12):  # 12-week course
    for student in students:
        # Deliver personalized instruction
        instruction = learning_env.create_personalized_instruction(
            learner_id=student.agent_id,
            current_topic=ml_curriculum.get_current_topic(student.agent_id)
        )

        instruction_result = learning_env.deliver_instruction(
            learner_id=student.agent_id,
            instruction=instruction,
            delivery_method="adaptive"
        )

        # Student processes instruction
        student.learn_from_instruction(instruction_result)

        # Assess learning progress
        if week % 2 == 0:  # Bi-weekly assessments
            assessment_result = learning_env.assess_learning(
                learner_id=student.agent_id,
                assessment_type="formative"
            )

            # Adapt curriculum based on performance
            if assessment_result.needs_adaptation:
                adaptation = learning_env.adapt_curriculum(
                    learner_id=student.agent_id,
                    performance_data=assessment_result.performance_data,
                    adaptation_strategy="difficulty_adjustment"
                )
                print(f"Adapted curriculum for {student.agent_id}: {adaptation.adaptations}")

# Generate learning analytics
analytics = learning_env.generate_learning_analytics()
print(f"Average class performance: {analytics.average_performance:.2f}")
print(f"Completion rate: {analytics.completion_rate:.1f}%")

class CompetitiveEnvironment(Environment):
    """
    Environment for competitive agent interactions.

    Supports tournaments, competitions, and competitive learning
    scenarios with ranking systems and performance metrics.
    """

    def __init__(
        self,
        env_id: str,
        competition_type: str = "tournament",
        scoring_system: str = "elo",
        rankings_enabled: bool = True,
        **kwargs
    ):
        """
        Initialize competitive environment.

        Args:
            env_id: Environment identifier
            competition_type: Type of competition
            scoring_system: System for scoring performance
            rankings_enabled: Enable agent rankings
        """

def create_competition(
    self,
    competition_id: str,
    participants: List[str],
    competition_rules: CompetitionRules,
    duration: int
) -> Competition:
    """
    Create a new competition.

    Args:
        competition_id: Competition identifier
        participants: Participating agent IDs
        competition_rules: Rules governing competition
        duration: Competition duration

    Returns:
        Competition: Created competition object
    """

def run_tournament(
    self,
    tournament_config: TournamentConfig
) -> TournamentResult:
    """
    Run a tournament between agents.

    Args:
        tournament_config: Tournament configuration

    Returns:
        TournamentResult: Tournament results and rankings
    """

def update_rankings(
    self,
    competition_results: List[CompetitionResult]
) -> RankingUpdate:
    """
    Update agent rankings based on competition results.

    Args:
        competition_results: Results from competitions

    Returns:
        RankingUpdate: Updated rankings information
    """

from cognito_sim_engine import CompetitiveEnvironment, CompetitionRules

# Create competitive environment for ML model development
comp_env = CompetitiveEnvironment(
    env_id="ml_competition_arena",
    competition_type="kaggle_style",
    scoring_system="performance_based",
    rankings_enabled=True
)

# Define competition rules
competition_rules = CompetitionRules(
    objective="maximize_accuracy",
    dataset="image_classification",
    evaluation_metric="f1_score",
    time_limit=7200,  # 2 hours
    resource_constraints={"memory": "4GB", "compute": "single_gpu"},
    submission_limit=5
)

# Add competitive agents
competitors = []
for i in range(10):
    agent = CognitiveAgent(
        agent_id=f"competitor_{i:03d}",
        agent_type=AgentType.COGNITIVE,
        personality_traits={
            "openness": random.uniform(0.6, 1.0),
            "conscientiousness": random.uniform(0.7, 1.0)
        }
    )
    comp_env.add_agent(agent)
    competitors.append(agent)

# Create and run competition
competition = comp_env.create_competition(
    competition_id="image_classification_challenge",
    participants=[agent.agent_id for agent in competitors],
    competition_rules=competition_rules,
    duration=7200
)

# Run the competition
tournament_config = TournamentConfig(
    format="single_elimination",
    seeding="random",
    matches_per_round=1
)

tournament_result = comp_env.run_tournament(tournament_config)

print(f"Tournament winner: {tournament_result.winner}")
print(f"Final rankings:")
for rank, agent_id in enumerate(tournament_result.final_rankings, 1):
    print(f"  {rank}. {agent_id}: {tournament_result.scores[agent_id]:.3f}")

@dataclass
class EnvironmentConfig:
    # Basic settings
    environment_type: str = "basic"
    capacity: int = 100
    time_step_duration: float = 1.0

    # Resource management
    resources: List[str] = None
    resource_renewal_rate: float = 0.1
    resource_scarcity: float = 0.0

    # Agent interaction
    interaction_enabled: bool = True
    communication_enabled: bool = False
    collaboration_mechanisms: List[str] = None

    # Environmental dynamics
    dynamic_properties: bool = False
    weather_simulation: bool = False
    day_night_cycle: bool = False
    seasonal_changes: bool = False

    # Monitoring and analytics
    performance_tracking: bool = True
    interaction_logging: bool = True
    state_history_enabled: bool = True

    # Advanced features
    physics_simulation: bool = False
    spatial_modeling: bool = False
    network_topology: str = "fully_connected"

class EnvironmentalDynamics:
    """Manages dynamic changes in environment properties."""

    def __init__(self, environment: Environment):
        self.environment = environment
        self.dynamics_rules = []
        self.state_history = []

    def add_dynamics_rule(self, rule: DynamicsRule) -> None:
        """Add a rule governing environmental changes."""
        self.dynamics_rules.append(rule)

    def update_dynamics(self) -> None:
        """Update environmental properties based on dynamics rules."""

        current_state = self.environment.get_state()

        for rule in self.dynamics_rules:
            if rule.condition(current_state):
                changes = rule.apply_changes(current_state)
                self.environment.apply_changes(changes)

        self.state_history.append(current_state)

    def simulate_resource_dynamics(self) -> None:
        """Simulate resource availability changes."""

        resources = self.environment.get_resources()

        for resource_name, resource in resources.items():
            # Resource consumption by agents
            consumption = self.calculate_resource_consumption(resource_name)

            # Resource renewal
            renewal = resource.renewal_rate * resource.max_capacity

            # Update resource availability
            new_availability = max(0, resource.current_amount - consumption + renewal)
            resource.current_amount = min(new_availability, resource.max_capacity)

# Example environmental dynamics
def create_dynamic_research_environment():
    """Create research environment with dynamic properties."""

    env = Environment(
        env_id="dynamic_research_lab",
        environment_type="research"
    )

    dynamics = EnvironmentalDynamics(env)

    # Add funding availability cycles
    funding_rule = DynamicsRule(
        name="funding_cycles",
        condition=lambda state: state.time_step % 720 == 0,  # Monthly cycles
        changes=lambda state: {"funding_availability": random.uniform(0.5, 1.0)}
    )
    dynamics.add_dynamics_rule(funding_rule)

    # Add collaborative opportunity dynamics
    collaboration_rule = DynamicsRule(
        name="collaboration_opportunities",
        condition=lambda state: len(state.active_agents) > 5,
        changes=lambda state: {"collaboration_bonus": 0.2}
    )
    dynamics.add_dynamics_rule(collaboration_rule)

    return env, dynamics

class EnvironmentMonitor:
    """Monitor and analyze environment state and agent interactions."""

    def __init__(self, environment: Environment):
        self.environment = environment
        self.metrics_history = []
        self.interaction_network = InteractionNetwork()

    def collect_metrics(self) -> EnvironmentMetrics:
        """Collect comprehensive environment metrics."""

        agents = self.environment.get_agents()
        current_state = self.environment.get_state()

        metrics = EnvironmentMetrics(
            timestamp=time.time(),
            num_agents=len(agents),
            agent_activity_level=self.calculate_activity_level(agents),
            resource_utilization=self.calculate_resource_utilization(),
            interaction_frequency=self.calculate_interaction_frequency(),
            collaboration_index=self.calculate_collaboration_index(),
            learning_progress=self.calculate_aggregate_learning_progress(agents),
            environmental_complexity=self.calculate_complexity(current_state)
        )

        self.metrics_history.append(metrics)
        return metrics

    def analyze_agent_interactions(self) -> InteractionAnalysis:
        """Analyze patterns in agent interactions."""

        interactions = self.environment.get_interaction_history()

        # Build interaction network
        for interaction in interactions:
            self.interaction_network.add_interaction(
                interaction.initiator,
                interaction.target,
                interaction.type,
                interaction.strength
            )

        # Analyze network properties
        analysis = InteractionAnalysis(
            network_density=self.interaction_network.calculate_density(),
            clustering_coefficient=self.interaction_network.calculate_clustering(),
            centrality_measures=self.interaction_network.calculate_centrality(),
            community_structure=self.interaction_network.detect_communities(),
            interaction_patterns=self.identify_interaction_patterns()
        )

        return analysis

    def generate_environment_report(self) -> EnvironmentReport:
        """Generate comprehensive environment analysis report."""

        current_metrics = self.collect_metrics()
        interaction_analysis = self.analyze_agent_interactions()

        report = EnvironmentReport(
            environment_id=self.environment.env_id,
            reporting_period=(
                self.metrics_history[0].timestamp if self.metrics_history else time.time(),
                time.time()
            ),
            summary_metrics=current_metrics,
            interaction_analysis=interaction_analysis,
            agent_performance_summary=self.summarize_agent_performance(),
            resource_usage_analysis=self.analyze_resource_usage(),
            recommendations=self.generate_optimization_recommendations()
        )

        return report

# Usage example
monitor = EnvironmentMonitor(collaborative_environment)

# Regular monitoring during simulation
for step in range(1000):
    # Environment and agents step
    env_state = collaborative_environment.step()

    # Collect metrics every 50 steps
    if step % 50 == 0:
        metrics = monitor.collect_metrics()
        print(f"Step {step}: Activity level {metrics.agent_activity_level:.2f}")

    # Generate comprehensive report every 500 steps
    if step % 500 == 0 and step > 0:
        report = monitor.generate_environment_report()
        print(f"Environment Report: Collaboration index {report.summary_metrics.collaboration_index:.2f}")

class CustomResearchEnvironment(CollaborativeEnvironment):
    """Custom environment for specific research scenarios."""

    def __init__(self, research_domain: str, **kwargs):
        super().__init__(**kwargs)
        self.research_domain = research_domain
        self.knowledge_base = DomainKnowledgeBase(research_domain)
        self.peer_review_system = PeerReviewSystem()
        self.publication_tracker = PublicationTracker()

    def facilitate_research_collaboration(
        self,
        research_question: str,
        participant_agents: List[str]
    ) -> ResearchCollaborationResult:
        """Facilitate collaborative research on specific question."""

        # Form research group
        research_group = self.create_collaboration_group(
            group_id=f"research_{hash(research_question)}",
            member_agents=participant_agents,
            collaboration_goal=research_question,
            coordination_strategy="expertise_complementary"
        )

        # Provide domain knowledge access
        for agent_id in participant_agents:
            agent = self.get_agent(agent_id)
            relevant_knowledge = self.knowledge_base.get_relevant_knowledge(
                research_question,
                agent.get_expertise_areas()
            )
            agent.incorporate_knowledge(relevant_knowledge)

        # Facilitate research process
        research_phases = [
            "literature_review",
            "hypothesis_generation", 
            "methodology_design",
            "experimentation",
            "analysis",
            "writing"
        ]

        results = {}
        for phase in research_phases:
            phase_result = self.execute_research_phase(
                research_group,
                phase,
                research_question
            )
            results[phase] = phase_result

            # Peer review for critical phases
            if phase in ["methodology_design", "analysis"]:
                review_result = self.peer_review_system.conduct_review(
                    content=phase_result,
                    reviewers=self.select_peer_reviewers(research_group)
                )
                results[f"{phase}_review"] = review_result

        return ResearchCollaborationResult(
            research_question=research_question,
            participants=participant_agents,
            phase_results=results,
            final_output=self.synthesize_research_output(results)
        )

    def create_conference_simulation(
        self,
        conference_theme: str,
        duration: int
    ) -> ConferenceSimulation:
        """Create and run academic conference simulation."""

        conference = ConferenceSimulation(
            theme=conference_theme,
            environment=self,
            duration=duration
        )

        # Schedule presentations
        presentations = self.generate_presentation_schedule()

        # Facilitate networking
        networking_sessions = self.schedule_networking_sessions()

        # Run conference
        conference_result = conference.run(
            presentations=presentations,
            networking_sessions=networking_sessions
        )

        return conference_result

# Create and use custom environment
research_env = CustomResearchEnvironment(
    env_id="ai_research_institute",
    research_domain="artificial_intelligence",
    collaboration_mechanisms=["joint_research", "peer_review", "conferences"],
    knowledge_sharing_enabled=True
)

# Facilitate research collaboration
collaboration_result = research_env.facilitate_research_collaboration(
    research_question="How can we improve few-shot learning in neural networks?",
    participant_agents=["researcher_001", "researcher_002", "researcher_003"]
)

print(f"Research collaboration completed with {len(collaboration_result.phase_results)} phases")