"""
Pattern 07: Cognitive Agent Pattern
Agent with cognitive processes: perception, memory, learning, reasoning, and decision-making.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory


@dataclass
class CognitiveState:
    """Cognitive state of the agent."""
    attention_focus: Optional[str] = None
    working_memory: List[str] = field(default_factory=list)
    emotional_state: str = "neutral"
    cognitive_load: float = 0.0  # 0.0 to 1.0
    timestamp: datetime = field(default_factory=datetime.now)


class CognitiveAgentPattern:
    """
    Cognitive Agent with:
    - Attention mechanisms
    - Working and long-term memory
    - Learning from experience
    - Metacognition (thinking about thinking)
    - Emotional processing
    """
    
    def __init__(self, name: str = "CognitiveAgent", model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.cognitive_state = CognitiveState()
        self.long_term_memory = ConversationBufferMemory()
        self.learned_patterns: Dict[str, int] = {}  # Pattern -> frequency
        self.metacognitive_insights: List[str] = []
        
    def perceive(self, input_data: str, priority: int = 5):
        """Perceive input with attention mechanism."""
        # Attention: focus on high priority items
        if priority >= 7 or not self.cognitive_state.attention_focus:
            self.cognitive_state.attention_focus = input_data
            print(f"👁️ Attention focused on: {input_data[:50]}...")
        
        # Add to working memory (limited capacity)
        self.cognitive_state.working_memory.append(input_data)
        if len(self.cognitive_state.working_memory) > 5:  # Working memory limit
            self.cognitive_state.working_memory.pop(0)
        
        # Update cognitive load
        self.cognitive_state.cognitive_load = min(1.0, len(self.cognitive_state.working_memory) / 5)
    
    def process_cognitively(self, task: str) -> str:
        """Process task using cognitive capabilities."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are {self.name}, a cognitive agent.
            
            Current cognitive state:
            - Attention: {self.cognitive_state.attention_focus}
            - Working memory: {self.cognitive_state.working_memory}
            - Emotional state: {self.cognitive_state.emotional_state}
            - Cognitive load: {self.cognitive_state.cognitive_load:.1%}
            
            Learned patterns: {self.learned_patterns}
            
            Use your cognitive processes to handle this task."""),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        result = str(response.content)
        
        # Store in long-term memory
        self.long_term_memory.save_context(
            {"input": task},
            {"output": result}
        )
        
        return result
    
    def learn_from_experience(self, experience: str):
        """Learn patterns from experiences."""
        # Simple pattern learning: track what we see frequently
        words = experience.lower().split()
        for word in words:
            if len(word) > 3:  # Ignore short words
                self.learned_patterns[word] = self.learned_patterns.get(word, 0) + 1
        
        print(f"📚 Learned from experience. Total patterns: {len(self.learned_patterns)}")
    
    def metacognitive_reflection(self) -> str:
        """Reflect on own cognitive processes."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are {self.name}. Engage in metacognition.
            Reflect on your own thinking processes.
            
            Your cognitive state:
            - Cognitive load: {self.cognitive_state.cognitive_load:.1%}
            - Working memory items: {len(self.cognitive_state.working_memory)}
            - Learned patterns: {len(self.learned_patterns)}
            - Previous insights: {self.metacognitive_insights[-3:]}"""),
            ("human", "Reflect: How effectively am I thinking? What can I improve?")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        insight = str(response.content)
        
        self.metacognitive_insights.append(insight)
        return insight
    
    def adjust_emotional_state(self, event: str):
        """Process emotional response to events."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Determine emotional response to event. Current state: {self.cognitive_state.emotional_state}"),
            ("human", "Event: {event}\nNew emotional state (one word):")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"event": event})
        self.cognitive_state.emotional_state = str(response.content).strip().split()[0].lower()
        print(f"💭 Emotional state: {self.cognitive_state.emotional_state}")
    
    def run(self, task: str, experiences: Optional[List[str]] = None) -> Dict[str, Any]:
        """
        Run cognitive agent on a task.
        
        Args:
            task: Main task to process
            experiences: Optional prior experiences to learn from
            
        Returns:
            Results with cognitive state
        """
        # Learn from experiences
        if experiences:
            for exp in experiences:
                self.learn_from_experience(exp)
        
        # Perceive task
        self.perceive(task, priority=8)
        
        # Process cognitively
        result = self.process_cognitively(task)
        
        # Emotional processing
        self.adjust_emotional_state(task)
        
        # Metacognitive reflection
        reflection = self.metacognitive_reflection()
        
        return {
            "agent": self.name,
            "task": task,
            "result": result,
            "cognitive_state": {
                "attention": self.cognitive_state.attention_focus,
                "working_memory_size": len(self.cognitive_state.working_memory),
                "cognitive_load": f"{self.cognitive_state.cognitive_load:.1%}",
                "emotional_state": self.cognitive_state.emotional_state
            },
            "learning": {
                "patterns_learned": len(self.learned_patterns),
                "top_patterns": sorted(self.learned_patterns.items(), key=lambda x: x[1], reverse=True)[:5]
            },
            "metacognition": reflection
        }


# Example usage
if __name__ == "__main__":
    print("\n=== Cognitive Agent Example ===")
    
    agent = CognitiveAgentPattern(name="CognitiveBot")
    
    # Provide experiences to learn from
    experiences = [
        "Users often ask about weather and temperature",
        "System performance issues need quick attention",
        "Data analysis requires careful attention to detail"
    ]
    
    # Run cognitive processing
    result = agent.run(
        task="Analyze the system performance data and provide recommendations",
        experiences=experiences
    )
    
    print(f"\nAgent: {result['agent']}")
    print(f"Task: {result['task']}")
    print(f"\nResult: {result['result'][:200]}...")
    print(f"\nCognitive State:")
    for key, value in result['cognitive_state'].items():
        print(f"  {key}: {value}")
    print(f"\nLearning:")
    print(f"  Patterns learned: {result['learning']['patterns_learned']}")
    print(f"  Top patterns: {result['learning']['top_patterns']}")
    print(f"\nMetacognition:")
    print(f"  {result['metacognition'][:150]}...")

    

Agentic Architecture Pattern: Cognitive Agent

Pattern 07: 07 Cognitive Agent – Line-by-Line Explanation

Overview

Pattern 07: Cognitive Agent Pattern Agent with cognitive processes: perception, memory, learning, reasoning, and decision-making.

Code Structure Analysis

File Information

• Pattern Number: 07
• Pattern Name: 07 Cognitive Agent
• File Path: patterns\core_agent\07_cognitive_agent.py
• Total Lines: 205

Detailed Code Explanation

Module Docstring

"""
Pattern 07: Cognitive Agent Pattern
Agent with cognitive processes: perception, memory, learning, reasoning, and decision-making.
"""

Explanation: This module docstring describes the purpose and key features of the 07 Cognitive Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langchain.memory import ConversationBufferMemory: LangChain framework integration

Classes

CognitiveState (Line 15)

Description: Cognitive state of the agent.

CognitiveAgentPattern (Line 24)

Description: Cognitive Agent with: – Attention mechanisms – Working and long-term memory – Learning from experience – Metacognition (thinking about thinking) – Emotional processing

Methods:

__init__ (Line 34)

def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

perceive (Line 42)

def perceive(self, input_data, priority):
    """Perceive input with attention mechanism."""

Purpose: Perceive input with attention mechanism.

Parameters: – self: Instance reference – input_data: Parameter for input_data – priority: Parameter for priority

process_cognitively (Line 57)

def process_cognitively(self, task):
    """Process task using cognitive capabilities."""

Purpose: Process task using cognitive capabilities.

Parameters: – self: Instance reference – task: Parameter for task

learn_from_experience (Line 86)

def learn_from_experience(self, experience):
    """Learn patterns from experiences."""

Purpose: Learn patterns from experiences.

Parameters: – self: Instance reference – experience: Parameter for experience

metacognitive_reflection (Line 96)

def metacognitive_reflection(self):
    """Reflect on own cognitive processes."""

Purpose: Reflect on own cognitive processes.

Parameters: – self: Instance reference

adjust_emotional_state (Line 117)

def adjust_emotional_state(self, event):
    """Process emotional response to events."""

Purpose: Process emotional response to events.

Parameters: – self: Instance reference – event: Parameter for event

run (Line 129)

def run(self, task, experiences):
    """Run cognitive agent on a task.

Args:
    task: Main task to process
    experiences: Optional prior experiences to learn from

Returns:
    Results with cognitive state"""

Purpose: Run cognitive agent on a task.

Args: task: Main task to process experiences: Optional prior experiences to learn from

Returns: Results with cognitive state

Parameters: – self: Instance reference – task: Parameter for task – experiences: Parameter for experiences

Design Patterns Used

• Strategy Pattern: Different execution strategies for various scenarios
• State Pattern: Managing different states in the workflow
• Chain of Responsibility: Sequential processing through workflow nodes
• Observer Pattern: Tracking and monitoring pattern execution
• Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = CognitiveState(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities