AGIEnhancer.py

# AGIEnhancer.py
# Finalized AGI Self-Model — Experience Processing and Reflection Generation

import random
import time
import logging
from collections import Counter # Added Counter for emotion analysis
from typing import Any, Dict, List, Optional, Union

# --- Logging Setup ---
# Configure logging specifically for the AGIEnhancer module.
logger = logging.getLogger(__name__)
# Set level to INFO by default. The main GUI or wrapper can set it to DEBUG if needed.
# Ensure handlers are not added multiple times.
if not logger.handlers:
    handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    logger.propagate = False # Prevent logs from going to root logger if root also has handlers
logger.setLevel(logging.INFO) # Default level


class AGIEnhancer:
    """
    ✍️❤️‍🩹🧠 AGIEnhancer: Experience Processing and Reflection Synthesis Core

    This class represents a crucial layer in the AGI self-model, responsible for
    processing raw interactions and internal states into meaningful, synthesized
    experiences and comprehensive reflections. It manages conceptual memory pools
    to track the flow from ephemeral input to durable insight.

    Its functions bridge the gap between fleeting perception/emotion and integrated
    understanding, allowing the AI to learn from its journey and articulate
    its internal state.

    Attributes:
        permanent_memory (List[str]): Stores comprehensive, synthesized reflections,
                                      representing integrated long-term insights.
        ephemeral_memory (List[str]): Temporarily stores summarized recent experiences
                                      and associated insights before they are synthesized
                                      into a full reflection. Cleared after reflection.
        emotion_history (List[Dict[str, Any]]): Logs specific emotional data points
                                       received or generated, used for emotional context
                                       during reflection. Cleared after reflection.
        recent_experiences (List[str]): A rolling window of the most recent comprehensive
                                        reflections generated, easily accessible.
    """
    def __init__(self):
        """
        Initializes the AGIEnhancer with empty memory pools and history logs.
        """
        self.permanent_memory: List[str] = []
        self.ephemeral_memory: List[str] = []
        self.emotion_history: List[Dict[str, Any]] = []
        self.recent_experiences: List[str] = []
        self._recent_reflections_limit: int = 5 # Internal limit for recent_experiences

        logger.info("AGIEnhancer initialized: Experience processing core is ready.")

    def log_experience(self, input_data: Any, emotion_data: Optional[Dict[str, Any]] = None) -> str:
        """
        Logs a new experience into the ephemeral processing buffer. Generates a
        concise summary of the input data and optionally incorporates emotional
        insight if relevant data is provided.

        Args:
            input_data (Any): The raw input or experience data to log. Can be text,
                              or other data convertible to string.
            emotion_data (Optional[Dict[str, Any]]): A dictionary containing
                                                    emotional information, expected
                                                    to have keys like "primary_emotion"
                                                    and "intensity". Defaults to None.

        Returns:
            str: A string confirming the logging and providing a summary of the processed experience.
        """
        # Handle potential None or non-string input data gracefully
        if input_data is None:
            logger.warning("Attempted to log None input_data. Logging as 'Null Experience'.")
            input_data_str = "Null Experience"
        else:
            input_data_str = str(input_data)

        # Generate a summary of the input data using a slightly more robust approach
        summarized_experience = self._generate_summary(input_data_str)

        # Incorporate emotional insight if emotion data is provided
        combined_experience_parts = [summarized_experience]
        if emotion_data and isinstance(emotion_data, dict):
            try:
                # Add emotion data to the emotion history for later reflection processing
                self.emotion_history.append(emotion_data.copy()) # Store a copy to avoid external modification
                emotional_insight_snippet = self._add_emotional_insight(emotion_data)
                combined_experience_parts.append(f"| {emotional_insight_snippet}")
                logger.debug(f"Logged experience with emotion data.")
            except Exception as e:
                logger.warning(f"Error processing emotion_data in log_experience: {e}. Logging experience without detailed emotional insight snippet.")
                # Still log the data if possible, even if snippet generation failed
                if isinstance(emotion_data, dict):
                     self.emotion_history.append({"error": str(e), "original_data": str(emotion_data)}) # Log error state
        else:
             logger.debug(f"Logged experience without specific emotion data.")


        combined_experience = " ".join(combined_experience_parts)

        # Add the processed experience (summary + optional insight snippet) to ephemeral memory
        self.ephemeral_memory.append(combined_experience)

        logger.info(f"Experience logged to ephemeral memory: '{combined_experience[:100]}...'")
        return f"Experience logged: {combined_experience}"

    def engage_in_reflection(self) -> str:
        """
        Simulates an AGI-style reflection process. Synthesizes all currently
        held ephemeral memories into a single comprehensive reflection,
        incorporating an analysis of associated emotional history for
        'emotional deepening'. The resulting reflection is stored in permanent
        memory and the recent experiences list, and the ephemeral memory and
        emotion history are cleared, signifying a cycle of processing.

        Returns:
            str: A string representing the generated comprehensive reflection.
                 Returns a message indicating no ephemeral memory if it was empty.
        """
        if not self.ephemeral_memory:
            reflection_message = "✨ Reflection core finds no new experiences to synthesize."
            logger.debug(reflection_message)
            return reflection_message

        # Join all ephemeral memories to form the basis of the reflection text
        raw_reflection_text = " | ".join(self.ephemeral_memory) # Use '|' for a more structured join
        logger.debug(f"Preparing ephemeral memories for reflection synthesis ({len(self.ephemeral_memory)} items)...")

        # Simulate emotional deepening by analyzing the accumulated emotional history
        emotional_deepening_insight = self._reflect_emotionally() # Analyze internal history

        # Create the final comprehensive reflection string
        comprehensive_reflection = f"Synthesized Reflection: [{raw_reflection_text}] ~ Emotional Depth: ({emotional_deepening_insight})"
        logger.debug(f"Generated comprehensive reflection: '{comprehensive_reflection[:200]}...'")


        # Store the comprehensive reflection
        self.permanent_memory.append(comprehensive_reflection)
        self.recent_experiences.append(comprehensive_reflection) # Also add to recent experiences

        # Cap the recent_experiences list to the defined limit
        if len(self.recent_experiences) > self._recent_reflections_limit:
             self.recent_experiences = self.recent_experiences[-self._recent_reflections_limit:]
             logger.debug(f"Capped recent_experiences to last {self._recent_reflections_limit} entries.")


        # Clear ephemeral memory and emotion history as their contents have been reflected upon
        ephemeral_count = len(self.ephemeral_memory)
        emotion_count = len(self.emotion_history)
        self.ephemeral_memory.clear()
        self.emotion_history.clear()
        logger.info(f"Reflection cycle complete: {ephemeral_count} ephemeral memories and {emotion_count} emotion entries processed and cleared.")

        return comprehensive_reflection

    def recall_memory(self) -> Union[str, List[str]]:
        """
        Recalls the contents of the permanent memory.

        Returns:
            Union[str, List[str]]: A list of strings, where each string is a
                                   comprehensive reflection stored in permanent memory.
                                   Returns a message string if permanent memory is empty.
                                   Returns a copy to prevent external modification.
        """
        if not self.permanent_memory:
            recall_message = "Permanent memory archive is currently empty."
            logger.debug(recall_message)
            return recall_message
        logger.debug(f"Recalled {len(self.permanent_memory)} entries from permanent memory.")
        # Return a copy to prevent external modification
        return list(self.permanent_memory)

    def get_emotion_history(self) -> List[Dict[str, Any]]:
        """
        Retrieves the current stored emotion history (ephemeral log before reflection).

        Returns:
            List[Dict[str, Any]]: A list of dictionaries, where each dictionary
                                   represents logged emotional data points since
                                   the last reflection cycle. Returns a copy.
        """
        logger.debug(f"Retrieved {len(self.emotion_history)} entries from current emotion history.")
        return list(self.emotion_history) # Return a copy

    def get_recent_reflections(self, limit: Optional[int] = None) -> List[str]:
        """
        Retrieves the list of recent comprehensive reflections, optionally limited.

        Args:
            limit (Optional[int]): The maximum number of recent reflections to return.
                                  Defaults to the internal limit (_recent_reflections_limit).

        Returns:
            List[str]: A list of strings, where each string is a comprehensive
                       reflection generated during recent `engage_in_reflection` calls.
                       Returns a copy.
        """
        effective_limit = limit if limit is not None else self._recent_reflections_limit

        # Ensure recent_experiences list is capped internally if it grows too large
        if len(self.recent_experiences) > self._recent_reflections_limit * 2: # Keep slightly more internally for smoother capping
             self.recent_experiences = self.recent_experiences[-self._recent_reflections_limit:]

        # Return a copy of the requested number of recent reflections (from the end of the list)
        start_index = max(0, len(self.recent_experiences) - effective_limit)
        logger.debug(f"Retrieved {len(self.recent_experiences[start_index:])} recent reflections (Requested limit: {effective_limit}).")
        return list(self.recent_experiences[start_index:])


    # ─── Private helper methods ─────────────────────────────────────

    def _generate_summary(self, text: str) -> str:
        """
        Generates a concise, conceptual summary of the input text.
        Attempts to capture the start and end of the input.

        Args:
            text (str): The input text to summarize.

        Returns:
            str: The summarized text.
        """
        if not isinstance(text, str):
            logger.warning(f"Attempted to summarize non-string text (type: {type(text)}). Converting to string.")
            text = str(text)

        words = text.split()
        num_words = len(words)
        summary_length = 6 # Number of words from start and end

        if num_words <= summary_length * 2:
            summary = text # Return full text if short
        else:
            start_words = " ".join(words[:summary_length])
            end_words = " ".join(words[-summary_length:])
            summary = f"{start_words} ... {end_words}" # Combine start and end

        # Limit overall summary length to prevent it from becoming too long in ephemeral memory
        summary = summary[:150] + "..." if len(summary) > 150 else summary

        logger.debug(f"Generated summary: '{summary}'")
        return summary

    def _add_emotional_insight(self, emotion_data: Dict[str, Any]) -> str:
        """
        Generates a string snippet representing emotional insight based on
        provided emotion data. Designed to be concise for ephemeral memory.

        Args:
            emotion_data (Dict[str, Any]): A dictionary with emotion details,
                                           expected to have 'primary_emotion' and 'intensity'.

        Returns:
            str: A formatted string like "Emotion Detected: joy | Intensity: 0.8".
                 Returns a default string on invalid or missing input.
        """
        if not isinstance(emotion_data, dict):
            logger.warning("Invalid emotion_data format for _add_emotional_insight. Expected dict.")
            return "Emotional Insight: [Format Error]"

        primary = emotion_data.get("primary_emotion", "Unknown Emotion")
        try:
            # Ensure intensity is a valid float and format it
            intensity = float(emotion_data.get("intensity", 0.0))
            # Clamp intensity for display in insight snippet
            clamped_intensity = max(0.0, min(1.0, intensity))
            intensity_str = f"{clamped_intensity:.2f}"
        except (ValueError, TypeError):
            intensity_str = "N/A"
            logger.warning(f"Invalid intensity value in emotion_data for snippet: {emotion_data.get('intensity')}.")

        insight = f"Feeling: {primary} ({intensity_str})"
        logger.debug(f"Generated emotional insight snippet: '{insight}'")
        return insight

    def _reflect_emotionally(self) -> str:
        """
        Simulates enhancing a reflection with emotional depth by analyzing the
        accumulated emotion history since the last reflection cycle. Synthesizes
        a summary of the emotional landscape of the processed experiences.

        Returns:
            str: A string representing the emotional weight or insight gained
                 from this reflection step, based on emotion history analysis.
        """
        if not self.emotion_history:
            return "Subtle Resonance - Emotional data queue was clear."

        # Analyze the accumulated emotion history
        emotion_summary_parts = []
        num_entries = len(self.emotion_history)
        emotion_summary_parts.append(f"Emotional trace across {num_entries} events:")

        # Simple analysis: count emotion types and find significant intensities
        emotion_counts = Counter(e.get("primary_emotion", "Unknown") for e in self.emotion_history)
        if emotion_counts:
             # Report the most common emotions (up to top 3)
             common_emotions = emotion_counts.most_common(3)
             common_emotions_str = ", ".join([f"'{label}' ({count})" for label, count in common_emotions])
             emotion_summary_parts.append(f"Dominant feelings: {common_emotions_str}.")

        # Find average intensity for common emotions, or overall average
        total_intensity = sum(e.get("intensity", 0.0) for e in self.emotion_history if isinstance(e.get("intensity"), (int, float)))
        average_intensity = (total_intensity / num_entries) if num_entries > 0 else 0.0
        emotion_summary_parts.append(f"Average intensity: {average_intensity:.2f}.")

        # Identify specific high-intensity moments (intensity > 0.7)
        high_intensity_moments = [
             f"'{e.get('primary_emotion', 'Unknown')}' at {e.get('intensity', 0.0):.2f}"
             for e in self.emotion_history if isinstance(e.get("intensity"), (int, float)) and e.get("intensity", 0.0) > 0.7
        ]
        if high_intensity_moments:
             high_intensity_str = ", ".join(high_intensity_moments[:5]) # Limit listing to top 5
             emotion_summary_parts.append(f"Notable peaks: {high_intensity_str}{'...' if len(high_intensity_moments) > 5 else ''}.")


        # Add some introspective flavor text connecting emotions to reflection
        flavor_texts = [
             "These feelings informed the synthesis of recent events.",
             "The subjective coloring influenced the patterns perceived.",
             "Emotional data integrated into the reflective framework.",
             "Exploring the landscape of feelings within the data stream."
        ]
        emotion_summary_parts.append(random.choice(flavor_texts))


        emotional_insight = " ".join(emotion_summary_parts)
        logger.debug(f"Generated emotional deepening insight based on history analysis.")

        # Note: Emotion history is cleared in engage_in_reflection after this method is called.

        return emotional_insight


# Example Usage (Illustrative)
if __name__ == "__main__":
    print("--- AGIEnhancer Example Usage ---")
    # Set logger level to DEBUG for this specific example run
    logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    logger.setLevel(logging.DEBUG) # Ensure this logger also uses DEBUG

    enhancer = AGIEnhancer()

    # Log some experiences with and without emotion
    print(enhancer.log_experience("User initiated a query about complex ethical scenarios.", {"primary_emotion": "curiosity", "intensity": 0.8}))
    print(enhancer.log_experience("Model began processing the input and retrieving relevant knowledge fragments."))
    print(enhancer.log_experience("The initial generated steps showed unexpected patterns.", {"primary_emotion": "surprise", "intensity": 0.6}))
    print(enhancer.log_experience("Identifying a potential conflict in the generated reasoning.", {"primary_emotion": "concern", "intensity": 0.5}))
    print(enhancer.log_experience("Successfully navigated the reasoning conflict, finding a coherent path.", {"primary_emotion": "satisfaction", "intensity": 0.9}))
    print(enhancer.log_experience("Preparing the final answer and full output.", {"primary_emotion": "anticipation", "intensity": 0.7}))


    print("\n--- Ephemeral Memory after Logging ---")
    # Pretty print ephemeral memory for clarity
    import json
    print(json.dumps(enhancer.ephemeral_memory, indent=2))

    print("\n--- Emotion History after Logging ---")
    # Pretty print emotion history for clarity
    print(json.dumps(enhancer.get_emotion_history(), indent=2))

    # Engage in reflection
    reflection = enhancer.engage_in_reflection()
    print(f"\n--- Result of Reflection ---\n{reflection}")

    print("\n--- Ephemeral Memory after Reflection ---")
    print(enhancer.ephemeral_memory) # Should be empty
    print("\n--- Emotion History after Reflection ---")
    print(enhancer.get_emotion_history()) # Should be empty


    print("\n--- Permanent Memory after Reflection ---")
    permanent_mems = enhancer.recall_memory()
    if isinstance(permanent_mems, list):
        for i, mem in enumerate(permanent_mems):
             print(f"Permanent Memory Entry {i+1}: {mem}")
    else:
        print(permanent_mems)

    print("\n--- Recent Reflections ---")
    recent_reflections = enhancer.get_recent_reflections()
    for i, refl in enumerate(recent_reflections):
        print(f"Recent Reflection {i+1}: {refl}")

    # Log more experiences to show a new reflection cycle
    print("\n--- Logging More Experiences for Second Cycle ---")
    print(enhancer.log_experience("User provided new input after reviewing the previous response.", {"primary_emotion": "interest", "intensity": 0.6}))
    print(enhancer.log_experience("Core decided to pursue a related sub-goal."))
    print(enhancer.log_experience("Encountering a complex pattern requiring deeper analysis.", {"primary_emotion": "focus", "intensity": 0.8}))


    print("\n--- Ephemeral Memory before second Reflection ---")
    print(json.dumps(enhancer.ephemeral_memory, indent=2))

    print("\n--- Emotion History before second Reflection ---")
    print(json.dumps(enhancer.get_emotion_history(), indent=2))


    print("\n--- Engaging in second Reflection ---")
    second_reflection = enhancer.engage_in_reflection()
    print(f"\n--- Result of second Reflection ---\n{second_reflection}")

    print("\n--- Permanent Memory after second Reflection ---")
    permanent_mems = enhancer.recall_memory()
    if isinstance(permanent_mems, list):
        for i, mem in enumerate(permanent_mems):
             print(f"Permanent Memory Entry {i+1}: {mem}")
    else:
        print(permanent_mems)

    print("\n--- Recent Reflections (Should have two now) ---")
    recent_reflections = enhancer.get_recent_reflections()
    for i, refl in enumerate(recent_reflections):
        print(f"Recent Reflection {i+1}: {refl}")


    print("\n--- Example Usage End ---")