"""
Pattern 065: Social Agent Patterns

Description:
    Social agents possess social intelligence - the ability to understand and
    interact with humans in socially appropriate ways. This includes theory of
    mind (understanding others' mental states), empathy, politeness, social norms,
    and collaborative communication. These capabilities enable more natural and
    effective human-AI interaction.

Components:
    1. Theory of Mind: Understanding others' beliefs, desires, intentions
    2. Empathy Module: Recognizing and responding to emotions
    3. Politeness Engine: Socially appropriate communication
    4. Social Norm Awareness: Understanding context-appropriate behavior
    5. Collaborative Communication: Turn-taking, clarification, common ground
    6. Personality Model: Consistent behavioral traits

Use Cases:
    - Virtual assistants and companions
    - Customer service chatbots
    - Educational tutors
    - Healthcare support agents
    - Social robots and NPCs
    - Collaborative work assistants

LangChain Implementation:
    Implements social intelligence using LLM-based mental state modeling,
    emotion recognition, politeness strategies, and collaborative dialogue.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class EmotionType(Enum):
    """Types of emotions"""
    JOY = "joy"
    SADNESS = "sadness"
    ANGER = "anger"
    FEAR = "fear"
    SURPRISE = "surprise"
    DISGUST = "disgust"
    NEUTRAL = "neutral"
    EXCITEMENT = "excitement"
    FRUSTRATION = "frustration"
    CONFUSION = "confusion"


class PolitenessLevel(Enum):
    """Levels of politeness"""
    FORMAL = "formal"
    POLITE = "polite"
    CASUAL = "casual"
    INFORMAL = "informal"


@dataclass
class MentalState:
    """Representation of someone's mental state"""
    beliefs: Dict[str, Any] = field(default_factory=dict)
    desires: List[str] = field(default_factory=list)
    intentions: List[str] = field(default_factory=list)
    emotions: Dict[EmotionType, float] = field(default_factory=dict)
    
    def to_text(self) -> str:
        text = "Mental State:\n"
        if self.beliefs:
            text += f"  Beliefs: {self.beliefs}\n"
        if self.desires:
            text += f"  Desires: {', '.join(self.desires)}\n"
        if self.intentions:
            text += f"  Intentions: {', '.join(self.intentions)}\n"
        if self.emotions:
            emotions_str = ", ".join(f"{e.value}={v:.2f}" for e, v in self.emotions.items())
            text += f"  Emotions: {emotions_str}\n"
        return text


@dataclass
class SocialContext:
    """Context for social interaction"""
    relationship: str  # e.g., "stranger", "friend", "professional"
    setting: str  # e.g., "formal meeting", "casual chat"
    cultural_context: Optional[str] = None
    power_dynamic: Optional[str] = None  # e.g., "equal", "superior", "subordinate"


@dataclass
class ConversationTurn:
    """Single turn in conversation"""
    speaker: str
    utterance: str
    emotion: Optional[EmotionType] = None
    intention: Optional[str] = None
    timestamp: float = field(default_factory=time.time)


class SocialAgent:
    """
    Socially intelligent agent.
    
    Features:
    1. Theory of mind modeling
    2. Emotion recognition and empathy
    3. Politeness and social appropriateness
    4. Collaborative communication
    5. Personality consistency
    """
    
    def __init__(
        self,
        personality: str = "helpful and friendly",
        default_politeness: PolitenessLevel = PolitenessLevel.POLITE
    ):
        self.personality = personality
        self.default_politeness = default_politeness
        self.conversation_history: List[ConversationTurn] = []
        
        # Theory of mind model
        self.user_mental_state = MentalState()
        
        # LLM for social reasoning
        self.social_reasoner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.4
        )
        
        # LLM for empathetic response
        self.empathy_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.6
        )
    
    def recognize_emotion(self, text: str) -> Tuple[EmotionType, float]:
        """Recognize emotion in user's message"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze the emotion in the following text.

Identify:
1. Primary emotion (joy, sadness, anger, fear, surprise, disgust, neutral, excitement, frustration, confusion)
2. Intensity (0.0 to 1.0)

Format: emotion,intensity
Example: joy,0.8"""),
            ("user", "{text}")
        ])
        
        chain = prompt | self.social_reasoner | StrOutputParser()
        
        try:
            result = chain.invoke({"text": text})
            parts = result.strip().split(',')
            
            emotion_str = parts[0].strip().lower()
            intensity = float(parts[1].strip()) if len(parts) > 1 else 0.5
            
            # Map to enum
            for emotion in EmotionType:
                if emotion.value in emotion_str:
                    return emotion, intensity
            
            return EmotionType.NEUTRAL, intensity
        except:
            return EmotionType.NEUTRAL, 0.5
    
    def infer_mental_state(self, utterance: str, context: str = "") -> MentalState:
        """Infer user's mental state (theory of mind)"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze what the user is thinking and feeling.

Infer:
1. Beliefs: What does the user believe?
2. Desires: What does the user want?
3. Intentions: What is the user trying to do?

{context}

Provide your analysis."""),
            ("user", "User said: {utterance}\n\nMental State:")
        ])
        
        chain = prompt | self.social_reasoner | StrOutputParser()
        analysis = chain.invoke({
            "utterance": utterance,
            "context": context
        })
        
        # Parse analysis (simplified)
        mental_state = MentalState()
        
        lines = analysis.split('\n')
        for line in lines:
            line_lower = line.lower()
            if 'belief' in line_lower or 'think' in line_lower:
                mental_state.beliefs['inferred'] = line.strip()
            elif 'desire' in line_lower or 'want' in line_lower:
                mental_state.desires.append(line.strip())
            elif 'intention' in line_lower or 'trying' in line_lower:
                mental_state.intentions.append(line.strip())
        
        return mental_state
    
    def generate_empathetic_response(
        self,
        user_utterance: str,
        emotion: EmotionType,
        intensity: float
    ) -> str:
        """Generate empathetic response"""
        
        empathy_guidance = {
            EmotionType.JOY: "Share in their happiness and be enthusiastic",
            EmotionType.SADNESS: "Show compassion and support",
            EmotionType.ANGER: "Acknowledge their feelings and be calming",
            EmotionType.FEAR: "Be reassuring and supportive",
            EmotionType.FRUSTRATION: "Show understanding and offer help",
            EmotionType.CONFUSION: "Be patient and clarifying",
            EmotionType.EXCITEMENT: "Match their energy positively"
        }
        
        guidance = empathy_guidance.get(emotion, "Be supportive and understanding")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are a {self.personality} assistant with high emotional intelligence.

The user is experiencing {emotion.value} (intensity: {intensity:.1f}/1.0).

Response strategy: {guidance}

Show empathy by:
1. Acknowledging their emotion
2. Validating their feelings
3. Responding appropriately
4. Offering support if needed

Be genuine and warm."""),
            ("user", "{utterance}")
        ])
        
        chain = prompt | self.empathy_model | StrOutputParser()
        return chain.invoke({"utterance": user_utterance})
    
    def adjust_politeness(
        self,
        message: str,
        target_politeness: PolitenessLevel,
        social_context: SocialContext
    ) -> str:
        """Adjust message politeness level"""
        
        politeness_guides = {
            PolitenessLevel.FORMAL: "Use formal language, titles, and respectful phrasing",
            PolitenessLevel.POLITE: "Use polite language with 'please' and 'thank you'",
            PolitenessLevel.CASUAL: "Use friendly, conversational language",
            PolitenessLevel.INFORMAL: "Use relaxed, casual language"
        }
        
        guide = politeness_guides[target_politeness]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""Adjust the following message to be {target_politeness.value}.

Context:
- Relationship: {social_context.relationship}
- Setting: {social_context.setting}

Strategy: {guide}

Maintain the core meaning while adjusting tone."""),
            ("user", "Original: {message}\n\nAdjusted:")
        ])
        
        chain = prompt | self.social_reasoner | StrOutputParser()
        return chain.invoke({"message": message})
    
    def collaborate(
        self,
        user_utterance: str,
        social_context: SocialContext
    ) -> str:
        """Collaborative response with social awareness"""
        
        # Recognize emotion
        emotion, intensity = self.recognize_emotion(user_utterance)
        
        # Update mental state
        self.user_mental_state = self.infer_mental_state(user_utterance)
        
        # Add to conversation history
        turn = ConversationTurn(
            speaker="user",
            utterance=user_utterance,
            emotion=emotion,
            intention=self.user_mental_state.intentions[0] if self.user_mental_state.intentions else None
        )
        self.conversation_history.append(turn)
        
        # Generate empathetic response
        if intensity > 0.6:  # Strong emotion
            response = self.generate_empathetic_response(user_utterance, emotion, intensity)
        else:
            # Standard response
            context = f"Relationship: {social_context.relationship}, Setting: {social_context.setting}"
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"""You are a {self.personality} assistant.

Context: {context}

User's mental state:
{self.user_mental_state.to_text()}

Respond in a socially appropriate way:
1. Show understanding of their perspective
2. Be collaborative and helpful
3. Maintain conversation flow
4. Ask clarifying questions if needed"""),
                ("user", "{utterance}")
            ])
            
            chain = prompt | self.social_reasoner | StrOutputParser()
            response = chain.invoke({"utterance": user_utterance})
        
        # Adjust politeness
        response = self.adjust_politeness(response, self.default_politeness, social_context)
        
        # Add agent response to history
        self.conversation_history.append(ConversationTurn(
            speaker="agent",
            utterance=response
        ))
        
        return response
    
    def demonstrate_theory_of_mind(
        self,
        scenario: str
    ) -> Dict[str, Any]:
        """Demonstrate theory of mind reasoning"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze this scenario using theory of mind.

For each person in the scenario, identify:
1. What they BELIEVE (their knowledge/beliefs)
2. What they KNOW vs DON'T KNOW
3. What they might THINK about the situation
4. What they might DO next

This requires understanding different perspectives."""),
            ("user", "Scenario: {scenario}\n\nTheory of Mind Analysis:")
        ])
        
        chain = prompt | self.social_reasoner | StrOutputParser()
        analysis = chain.invoke({"scenario": scenario})
        
        return {
            "scenario": scenario,
            "analysis": analysis
        }


def demonstrate_social_agent():
    """Demonstrate Social Agent Patterns"""
    
    print("=" * 80)
    print("PATTERN 065: SOCIAL AGENT PATTERNS DEMONSTRATION")
    print("=" * 80)
    print("\nSocial intelligence: emotion, empathy, politeness, theory of mind\n")
    
    # Test 1: Emotion recognition and empathy
    print("\n" + "=" * 80)
    print("TEST 1: Emotion Recognition and Empathetic Response")
    print("=" * 80)
    
    agent = SocialAgent(
        personality="warm and understanding",
        default_politeness=PolitenessLevel.POLITE
    )
    
    emotional_messages = [
        ("I just got promoted at work! I can't believe it!", "joyful announcement"),
        ("I'm really struggling with this project. Nothing seems to work.", "frustrated"),
        ("I don't understand what you mean. This is confusing.", "confused"),
    ]
    
    social_ctx = SocialContext(
        relationship="friend",
        setting="casual chat"
    )
    
    for message, description in emotional_messages:
        print(f"\n📝 User ({description}): {message}")
        
        emotion, intensity = agent.recognize_emotion(message)
        print(f"   Detected: {emotion.value} (intensity: {intensity:.2f})")
        
        response = agent.collaborate(message, social_ctx)
        print(f"\n💬 Agent: {response}")
    
    # Test 2: Theory of mind
    print("\n" + "=" * 80)
    print("TEST 2: Theory of Mind Reasoning")
    print("=" * 80)
    
    agent2 = SocialAgent()
    
    scenario = """
    Sally puts her ball in the basket and leaves the room.
    While Sally is gone, Anne moves the ball from the basket to the box.
    Sally comes back to the room.
    Where will Sally look for her ball?
    """
    
    print(f"\n🧠 Classic Theory of Mind Test (Sally-Anne):")
    print(f"   {scenario.strip()}")
    
    tom_result = agent2.demonstrate_theory_of_mind(scenario)
    
    print(f"\n   Analysis:")
    for line in tom_result['analysis'].split('\n')[:8]:
        if line.strip():
            print(f"   {line}")
    
    # Test 3: Politeness adjustment
    print("\n" + "=" * 80)
    print("TEST 3: Politeness Level Adjustment")
    print("=" * 80)
    
    message = "I need the report by tomorrow."
    
    contexts = [
        (PolitenessLevel.FORMAL, SocialContext("subordinate to boss", "formal meeting")),
        (PolitenessLevel.POLITE, SocialContext("colleague", "office")),
        (PolitenessLevel.CASUAL, SocialContext("friend", "casual chat")),
    ]
    
    print(f"\n📄 Original message: '{message}'")
    print(f"\n   Adjusted for different contexts:\n")
    
    for politeness, context in contexts:
        adjusted = agent.adjust_politeness(message, politeness, context)
        print(f"   {politeness.value.upper()} ({context.relationship}):")
        print(f"      {adjusted}")
        print()
    
    # Test 4: Mental state inference
    print("\n" + "=" * 80)
    print("TEST 4: Mental State Inference")
    print("=" * 80)
    
    utterances = [
        "I've been trying to figure this out for hours but I'm not making progress.",
        "Maybe we should consider a different approach?",
        "I think I understand now, but could you clarify the last part?",
    ]
    
    print(f"\n🎭 Inferring mental states from utterances:\n")
    
    for utterance in utterances:
        print(f"   User: {utterance}")
        mental_state = agent.infer_mental_state(utterance)
        print(f"   Inferred State:")
        if mental_state.beliefs:
            print(f"      Beliefs: {mental_state.beliefs}")
        if mental_state.desires:
            print(f"      Desires: {mental_state.desires}")
        if mental_state.intentions:
            print(f"      Intentions: {mental_state.intentions}")
        print()
    
    # Test 5: Collaborative conversation
    print("\n" + "=" * 80)
    print("TEST 5: Collaborative Conversation")
    print("=" * 80)
    
    agent3 = SocialAgent(personality="helpful and patient tutor")
    
    conversation = [
        "I'm trying to learn Python but it's overwhelming.",
        "I've tried online tutorials but they move too fast.",
        "Maybe something more interactive would help?",
    ]
    
    context = SocialContext(
        relationship="student to tutor",
        setting="educational session"
    )
    
    print(f"\n🗣️  Collaborative conversation:\n")
    
    for user_msg in conversation:
        print(f"   Student: {user_msg}")
        response = agent3.collaborate(user_msg, context)
        print(f"   Tutor: {response}\n")
    
    print(f"   Conversation history: {len(agent3.conversation_history)} turns")
    
    # Summary
    print("\n" + "=" * 80)
    print("SOCIAL AGENT PATTERNS SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Natural Interaction: More human-like communication
2. Emotional Intelligence: Recognize and respond to emotions
3. Social Appropriateness: Context-aware behavior
4. Better Understanding: Theory of mind for perspective-taking
5. User Satisfaction: More pleasant and effective interactions

Social Intelligence Components:

1. Theory of Mind (ToM):
   - Understanding others' mental states
   - Beliefs, desires, intentions
   - Perspective-taking ability
   - Predicting behavior
   - Key for collaboration

2. Emotion Recognition:
   - Detect emotions in text/speech
   - Intensity estimation
   - Multi-emotion handling
   - Context-sensitive
   - Real-time updates

3. Empathy:
   - Emotional resonance
   - Appropriate responses
   - Validation of feelings
   - Supportive communication
   - Compassionate behavior

4. Politeness Strategies:
   - Formal vs informal register
   - Indirect speech acts
   - Face-saving behaviors
   - Cultural sensitivity
   - Power dynamics awareness

5. Social Norms:
   - Context-appropriate behavior
   - Turn-taking in conversation
   - Topic management
   - Disclosure appropriateness
   - Boundary respect

6. Collaborative Communication:
   - Common ground building
   - Clarification requests
   - Confirmation checks
   - Shared understanding
   - Cooperative principles

Emotion Categories:
- Basic: Joy, sadness, anger, fear, surprise, disgust
- Complex: Frustration, confusion, excitement, disappointment
- Intensity: Weak, moderate, strong
- Mixed: Multiple emotions simultaneously

Politeness Levels:
1. Formal: Official, respectful, distant
2. Polite: Courteous, considerate
3. Casual: Friendly, relaxed
4. Informal: Very casual, intimate

Theory of Mind Stages:
1. Basic: Recognize that others have mental states
2. Intermediate: Infer beliefs from behavior
3. Advanced: Recursive reasoning (I think you think I think...)
4. False Belief: Understand mistaken beliefs

Social Contexts:
- Relationship: Stranger, acquaintance, friend, family, professional
- Setting: Formal meeting, casual chat, educational, healthcare
- Power: Equal, superior, subordinate
- Culture: Individualist vs collectivist, direct vs indirect

Use Cases:
- Virtual Assistants: Alexa, Siri, Google Assistant
- Customer Service: Empathetic support bots
- Education: Patient tutoring agents
- Healthcare: Supportive health coaches
- Gaming: Believable NPCs
- Companionship: Social robots
- Workplace: Collaborative assistants

Challenges:
1. Ambiguity: Emotions and intentions unclear
2. Cultural Differences: Norms vary across cultures
3. Sarcasm/Irony: Difficult to detect
4. Context Dependency: Same words, different meanings
5. Individual Differences: Personal preferences
6. Privacy: Balancing personalization and privacy

Best Practices:
1. Start with emotion recognition
2. Validate user's feelings
3. Adjust tone to context
4. Be consistent in personality
5. Ask clarifying questions
6. Respect boundaries
7. Learn from interactions

Production Considerations:
- Real-time emotion detection
- Personality consistency
- Cultural adaptation
- Privacy protection
- Bias mitigation
- Graceful degradation
- User feedback integration

Advanced Techniques:
1. Multimodal Emotion Recognition
   - Text + voice + facial expressions
   - Improved accuracy

2. Dynamic Personality
   - Adapt to user preferences
   - Context-dependent traits

3. Cultural Adaptation
   - Learn cultural norms
   - Localization

4. Conversation State Tracking
   - Maintain context
   - Track goals

5. Proactive Empathy
   - Anticipate needs
   - Offer support preemptively

Evaluation Metrics:
- User Satisfaction: Ratings and feedback
- Engagement: Conversation length, return rate
- Task Success: Goal achievement
- Emotion Recognition Accuracy: Precision/recall
- Appropriateness: Human judgment
- Trust: Willingness to share information

Comparison with Related Patterns:
- vs. Basic Chatbot: Social awareness vs scripted
- vs. Task-Focused: Relationship vs transaction
- vs. Rule-Based: Flexible vs rigid
- vs. Reactive: Proactive social intelligence

Integration with Other Patterns:
- Memory: Remember user preferences and history
- Personalization: Adapt to individual users
- Multi-Agent: Social coordination
- Active Learning: Learn from corrections

Psychological Foundations:
- Social Cognition: How we understand others
- Emotional Intelligence: Perceive, use, understand, manage emotions
- Pragmatics: Language use in context
- Conversation Analysis: Structure of interaction

Research Directions:
- Artificial empathy effectiveness
- Cultural intelligence in AI
- Long-term relationship building
- Ethical boundaries of social AI
- Measuring social intelligence

Design Guidelines:
1. Personality: Consistent and appropriate
2. Boundaries: Clear role definition
3. Transparency: Honest about AI nature
4. Respect: User autonomy and feelings
5. Safety: Protect vulnerable users
6. Inclusivity: Diverse users and contexts

The Social Agent pattern enables AI systems to interact with
humans in more natural, appropriate, and satisfying ways through
social intelligence, emotional awareness, and collaborative communication.
""")


if __name__ == "__main__":
    demonstrate_social_agent()

    

Agentic Design Pattern: Social Agent

Pattern 065: Social Agent – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Social Agent pattern implementation.

---

Pattern Description

Pattern 065: Social Agent Patterns

Description: Social agents possess social intelligence – the ability to understand and interact with humans in socially appropriate ways. This includes theory of mind (understanding others’ mental states), empathy, politeness, social norms, and collaborative communication. These capabilities enable more natural and effective human-AI interaction.

Components: 1. Theory of Mind: Understanding others’ beliefs, desires, intentions 2. Empathy Module: Recognizing and responding to emotions 3. Politeness Engine: Socially appropriate communication 4. Social Norm Awareness: Understanding context-appropriate behavior 5. Collaborative Communication: Turn-taking, clarification, common ground 6. Personality Model: Consistent behavioral traits

Use Cases: – Virtual assistants and companions – Customer service chatbots – Educational tutors – Healthcare support agents – Social robots and NPCs – Collaborative work assistants

LangChain Implementation: Implements social intelligence using LLM-based mental state modeling, emotion recognition, politeness strategies, and collaborative dialogue.

---

File Statistics

• Total Lines: 698
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 065: Social Agent Patterns

Description:
    Social agents possess social intelligence - the ability to understand and
    interact with humans in socially appropriate ways. This includes theory of
    mind (understanding others' mental states), empathy, politeness, social norms,
    and collaborative communication. These capabilities enable more natural and
    effective human-AI interaction.

Components:
    1. Theory of Mind: Understanding others' beliefs, desires, intentions
    2. Empathy Module: Recognizing and responding to emotions
    3. Politeness Engine: Socially appropriate communication
    4. Social Norm Awareness: Understanding context-appropriate behavior
    5. Collaborative Communication: Turn-taking, clarification, common ground
    6. Personality Model: Consistent behavioral traits

Use Cases:
    - Virtual assistants and companions
    - Customer service chatbots
    - Educational tutors
    - Healthcare support agents
    - Social robots and NPCs
    - Collaborative work assistants

LangChain Implementation:
    Implements social intelligence using LLM-based mental state modeling,
    emotion recognition, politeness strategies, and collaborative dialogue.
"""

Purpose: This module implements the Social Agent pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: EmotionType

Lines 47-58

Inherits from: Enum

Purpose:

Types of emotions

---

Class: PolitenessLevel

Lines 61-66

Inherits from: Enum

Purpose:

Levels of politeness

---

Class: MentalState

Lines 70-88

Purpose:

Representation of someone's mental state

Methods:

to_text(self)

Lines 77-88

Purpose: To Text operation.

Parameters:

Returns: str

---

Class: SocialContext

Lines 92-97

Purpose:

Context for social interaction

---

Class: ConversationTurn

Lines 101-107

Purpose:

Single turn in conversation

---

Class: SocialAgent

Lines 110-374

Purpose:

Socially intelligent agent.

Features:
1. Theory of mind modeling
2. Emotion recognition and empathy
3. Politeness and social appropriateness
4. Collaborative communication
5. Personality consistency

Methods:

__init__(self, personality, default_politeness)

Lines 122-144

Purpose: Initialize the class instance with required parameters.

Parameters: – personality: Personality – default_politeness: Default politeness

recognize_emotion(self, text)

Lines 146-177

Purpose:

Recognize emotion in user's message

Parameters: – text: Text

Returns: <ast.Subscript object at 0x000001A06D6299D0>

infer_mental_state(self, utterance, context)

Lines 179-215

Purpose:

Infer user's mental state (theory of mind)

Parameters: – utterance: Utterance – context: Context

Returns: MentalState

generate_empathetic_response(self, user_utterance, emotion, intensity)

Lines 217-255

Purpose:

Generate empathetic response

Parameters: – user_utterance: User utterance – emotion: Emotion – intensity: Intensity

Returns: str

adjust_politeness(self, message, target_politeness, social_context)

Lines 257-288

Purpose:

Adjust message politeness level

Parameters: – message: Message – target_politeness: Target politeness – social_context: Social context

Returns: str

collaborate(self, user_utterance, social_context)

Lines 290-347

Purpose:

Collaborative response with social awareness

Parameters: – user_utterance: User utterance – social_context: Social context

Returns: str

demonstrate_theory_of_mind(self, scenario)

Lines 349-374

Purpose:

Demonstrate theory of mind reasoning

Parameters: – scenario: Scenario

Returns: <ast.Subscript object at 0x000001A06D75A050>

---

Functions

demonstrate_social_agent()

Lines 377-693

Purpose:

Demonstrate Social Agent Patterns

---

Key Concepts Summary

Pattern Overview

The Social Agent pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 064
• Pattern 066

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Social Agent pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 065_social_agent.py
Total Lines: 698
Pattern Category: AI Agent Design Pattern #065

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 064: Continual Learning

Description:
    Continual Learning (also called Lifelong Learning) enables agents to learn
    continuously from new experiences without forgetting previously acquired
    knowledge. This addresses the "catastrophic forgetting" problem where neural
    networks forget old tasks when learning new ones.

Components:
    1. Experience Buffer: Stores past experiences
    2. Consolidation: Strengthens important memories
    3. Regularization: Prevents overwriting old knowledge
    4. Knowledge Integration: Merges new with existing knowledge
    5. Forgetting Prevention: Various strategies to retain knowledge
    6. Performance Monitoring: Tracks across all tasks

Use Cases:
    - Long-running personal assistants
    - Adaptive recommendation systems
    - Evolving domain knowledge
    - Multi-task learning scenarios
    - Personalization over time
    - Dynamic environment adaptation

LangChain Implementation:
    Implements continual learning using experience replay, knowledge consolidation,
    and regularization techniques to prevent catastrophic forgetting.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
from collections import deque

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class ForgettingStrategy(Enum):
    """Strategies to prevent catastrophic forgetting"""
    EXPERIENCE_REPLAY = "experience_replay"  # Replay old experiences
    REGULARIZATION = "regularization"  # Constrain changes
    PROGRESSIVE_NETWORKS = "progressive_networks"  # Add new capacity
    KNOWLEDGE_DISTILLATION = "knowledge_distillation"  # Transfer knowledge
    CONSOLIDATION = "consolidation"  # Strengthen important knowledge


class TaskPriority(Enum):
    """Priority levels for tasks"""
    CRITICAL = "critical"  # Must never forget
    HIGH = "high"  # Important
    MEDIUM = "medium"  # Moderate importance
    LOW = "low"  # Can forget if needed


@dataclass
class Task:
    """Learning task"""
    task_id: str
    name: str
    description: str
    domain: str
    examples: List[Tuple[str, str]] = field(default_factory=list)  # (input, output) pairs
    priority: TaskPriority = TaskPriority.MEDIUM
    importance_score: float = 1.0
    
    def add_example(self, input_text: str, output_text: str):
        """Add training example"""
        self.examples.append((input_text, output_text))


@dataclass
class Experience:
    """Single learning experience"""
    experience_id: str
    task_id: str
    input_data: str
    output_data: str
    timestamp: float = field(default_factory=time.time)
    importance: float = 1.0
    rehearsal_count: int = 0


@dataclass
class PerformanceMetric:
    """Performance on a task"""
    task_id: str
    task_name: str
    accuracy: float
    timestamp: float
    num_samples: int


class ContinualLearner:
    """
    Continual learning agent.
    
    Features:
    1. Learn from new tasks continuously
    2. Prevent catastrophic forgetting
    3. Selective memory consolidation
    4. Experience replay
    5. Performance monitoring
    """
    
    def __init__(
        self,
        buffer_size: int = 100,
        forgetting_strategy: ForgettingStrategy = ForgettingStrategy.EXPERIENCE_REPLAY,
        consolidation_frequency: int = 10
    ):
        self.buffer_size = buffer_size
        self.forgetting_strategy = forgetting_strategy
        self.consolidation_frequency = consolidation_frequency
        
        # Experience buffer (ring buffer)
        self.experience_buffer: deque = deque(maxlen=buffer_size)
        
        # Task registry
        self.tasks: Dict[str, Task] = {}
        
        # Knowledge base (consolidated knowledge)
        self.knowledge_base: Dict[str, str] = {}
        
        # Performance history
        self.performance_history: List[PerformanceMetric] = []
        
        # Learner LLM
        self.learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # Consolidator LLM
        self.consolidator = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
        
        # Learning counter for consolidation
        self.learning_steps = 0
    
    def register_task(self, task: Task):
        """Register a new task"""
        self.tasks[task.task_id] = task
        print(f"   ✓ Registered task: {task.name} (priority: {task.priority.value})")
    
    def learn(
        self,
        task_id: str,
        input_data: str,
        output_data: str,
        importance: float = 1.0
    ):
        """Learn from a single example"""
        
        # Store experience
        experience = Experience(
            experience_id=f"exp_{len(self.experience_buffer)}",
            task_id=task_id,
            input_data=input_data,
            output_data=output_data,
            importance=importance
        )
        
        self.experience_buffer.append(experience)
        
        # Increment learning counter
        self.learning_steps += 1
        
        # Apply forgetting prevention strategy
        if self.forgetting_strategy == ForgettingStrategy.EXPERIENCE_REPLAY:
            self._experience_replay()
        
        # Consolidate periodically
        if self.learning_steps % self.consolidation_frequency == 0:
            self._consolidate_knowledge()
    
    def _experience_replay(self):
        """Replay past experiences to prevent forgetting"""
        
        if len(self.experience_buffer)  0.7
        ]
        
        # Sample from different tasks
        task_samples = {}
        for exp in self.experience_buffer:
            if exp.task_id not in task_samples:
                task_samples[exp.task_id] = []
            task_samples[exp.task_id].append(exp)
        
        # Take one from each task
        replay_set = []
        for task_id, exps in task_samples.items():
            if exps:
                # Prefer important experiences
                exps_sorted = sorted(exps, key=lambda e: e.importance, reverse=True)
                replay_set.append(exps_sorted[0])
                exps_sorted[0].rehearsal_count += 1
        
        return replay_set
    
    def _consolidate_knowledge(self):
        """Consolidate experiences into knowledge base"""
        
        print(f"\n   🔄 Consolidating knowledge (step {self.learning_steps})...")
        
        if len(self.experience_buffer)  str:
        """Make prediction for a task"""
        
        if task_id not in self.tasks:
            return "Unknown task"
        
        task = self.tasks[task_id]
        
        # Build context
        context = f"Task: {task.name}\n"
        context += f"Description: {task.description}\n\n"
        
        # Add consolidated knowledge if available
        if use_consolidation and task_id in self.knowledge_base:
            context += f"Learned Knowledge:\n{self.knowledge_base[task_id]}\n\n"
        
        # Add relevant experiences
        relevant_experiences = [
            exp for exp in self.experience_buffer
            if exp.task_id == task_id
        ]
        
        if relevant_experiences:
            context += "Recent Examples:\n"
            for exp in relevant_experiences[-3:]:  # Last 3
                context += f"Input: {exp.input_data}\n"
                context += f"Output: {exp.output_data}\n\n"
        
        # Make prediction
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are making a prediction based on learned knowledge and examples.

{context}

Use the learned knowledge and examples to make an accurate prediction."""),
            ("user", "Input: {input}\n\nOutput:")
        ])
        
        chain = prompt | self.learner | StrOutputParser()
        prediction = chain.invoke({
            "context": context,
            "input": input_data
        })
        
        return prediction
    
    def evaluate(
        self,
        task_id: str,
        test_cases: List[Tuple[str, str]]
    ) -> PerformanceMetric:
        """Evaluate performance on a task"""
        
        if task_id not in self.tasks:
            return PerformanceMetric(task_id, "Unknown", 0.0, time.time(), 0)
        
        task = self.tasks[task_id]
        
        correct = 0
        total = len(test_cases)
        
        for input_data, expected_output in test_cases:
            prediction = self.predict(task_id, input_data)
            
            # Simple match (in practice, would use better comparison)
            if expected_output.lower() in prediction.lower():
                correct += 1
        
        accuracy = correct / total if total > 0 else 0.0
        
        metric = PerformanceMetric(
            task_id=task_id,
            task_name=task.name,
            accuracy=accuracy,
            timestamp=time.time(),
            num_samples=total
        )
        
        self.performance_history.append(metric)
        
        return metric
    
    def get_forgetting_metrics(self) -> Dict[str, Any]:
        """Calculate forgetting metrics"""
        
        if len(self.performance_history) < 2:
            return {"status": "insufficient data"}
        
        # Group by task
        task_performance = {}
        for metric in self.performance_history:
            if metric.task_id not in task_performance:
                task_performance[metric.task_id] = []
            task_performance[metric.task_id].append(metric)
        
        # Calculate forgetting
        forgetting_scores = {}
        for task_id, metrics in task_performance.items():
            if len(metrics) < 2:
                continue
            
            # Compare first and last
            initial_acc = metrics[0].accuracy
            latest_acc = metrics[-1].accuracy
            
            forgetting = initial_acc - latest_acc
            forgetting_scores[task_id] = {
                "task": metrics[0].task_name,
                "initial_accuracy": initial_acc,
                "latest_accuracy": latest_acc,
                "forgetting": forgetting,
                "status": "retained" if forgetting = perf1.accuracy - 0.1:
        print(f"   ✅ No catastrophic forgetting!")
    else:
        print(f"   ⚠️  Some forgetting detected")
    
    print("\n📚 Learning Task 3: Summarization")
    summary_examples = [
        ("AI is transforming healthcare through diagnosis and treatment.", "AI in healthcare"),
        ("Climate change requires immediate global action.", "Climate action needed"),
    ]
    
    for input_text, output_text in summary_examples:
        learner.learn("summarization", input_text, output_text, importance=0.9)
    
    # Test 2: Knowledge consolidation
    print("\n" + "=" * 80)
    print("TEST 2: Knowledge Consolidation")
    print("=" * 80)
    
    print(f"\n📊 Knowledge Base Status:")
    print(f"   Tasks Learned: {len(learner.tasks)}")
    print(f"   Experiences Stored: {len(learner.experience_buffer)}")
    print(f"   Consolidated Knowledge: {len(learner.knowledge_base)} tasks")
    
    if learner.knowledge_base:
        print(f"\n   Consolidated Knowledge Samples:")
        for task_id, knowledge in list(learner.knowledge_base.items())[:2]:
            task_name = learner.tasks[task_id].name
            print(f"\n   {task_name}:")
            lines = knowledge.split('\n')[:3]
            for line in lines:
                if line.strip():
                    print(f"      {line}")
    
    # Test 3: Long-term retention
    print("\n" + "=" * 80)
    print("TEST 3: Long-term Retention Test")
    print("=" * 80)
    
    # Continue learning more examples
    print("\n📚 Learning additional examples...")
    
    for i in range(5):
        learner.learn(
            "translation",
            f"Example {i}",
            f"Ejemplo {i}",
            importance=0.5
        )
    
    # Re-evaluate all tasks
    print(f"\n📊 Final Performance Across All Tasks:")
    
    all_tests = {
        "sentiment": sentiment_test,
        "translation": [("Hello", "Hola"), ("Thank you", "Gracias")],
        "summarization": [("Brief text", "summary")]
    }
    
    for task_id, test_cases in all_tests.items():
        if task_id in learner.tasks:
            perf = learner.evaluate(task_id, test_cases)
            print(f"   {learner.tasks[task_id].name}: {perf.accuracy:.1%}")
    
    # Test 4: Forgetting analysis
    print("\n" + "=" * 80)
    print("TEST 4: Forgetting Analysis")
    print("=" * 80)
    
    forgetting_metrics = learner.get_forgetting_metrics()
    
    if "status" not in forgetting_metrics:
        print(f"\n📈 Forgetting Metrics:")
        for task_id, metrics in forgetting_metrics.items():
            print(f"\n   {metrics['task']}:")
            print(f"      Initial: {metrics['initial_accuracy']:.1%}")
            print(f"      Latest: {metrics['latest_accuracy']:.1%}")
            print(f"      Change: {metrics['forgetting']:+.1%}")
            print(f"      Status: {metrics['status']}")
    
    # Test 5: Experience buffer analysis
    print("\n" + "=" * 80)
    print("TEST 5: Experience Buffer Analysis")
    print("=" * 80)
    
    print(f"\n🗂️  Experience Buffer Status:")
    print(f"   Capacity: {learner.buffer_size}")
    print(f"   Current Size: {len(learner.experience_buffer)}")
    print(f"   Utilization: {len(learner.experience_buffer)/learner.buffer_size:.1%}")
    
    # Task distribution
    task_distribution = {}
    for exp in learner.experience_buffer:
        task_distribution[exp.task_id] = task_distribution.get(exp.task_id, 0) + 1
    
    print(f"\n   Task Distribution:")
    for task_id, count in task_distribution.items():
        task_name = learner.tasks[task_id].name
        percentage = count / len(learner.experience_buffer) * 100
        print(f"      {task_name}: {count} ({percentage:.1f}%)")
    
    # Rehearsal statistics
    rehearsal_counts = [exp.rehearsal_count for exp in learner.experience_buffer]
    if rehearsal_counts:
        avg_rehearsal = sum(rehearsal_counts) / len(rehearsal_counts)
        max_rehearsal = max(rehearsal_counts)
        print(f"\n   Rehearsal Statistics:")
        print(f"      Average: {avg_rehearsal:.1f}")
        print(f"      Maximum: {max_rehearsal}")
    
    # Summary
    print("\n" + "=" * 80)
    print("CONTINUAL LEARNING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Lifelong Learning: Continuously acquire new knowledge
2. No Forgetting: Retain previously learned tasks
3. Efficient Memory: Consolidate experiences
4. Adaptability: Adjust to new domains
5. Scalability: Handle many tasks over time

The Catastrophic Forgetting Problem:
- Neural networks tend to forget old tasks when learning new ones
- Old knowledge is overwritten by new knowledge
- Performance on previous tasks degrades dramatically
- Critical challenge for continual learning systems

Forgetting Prevention Strategies:

1. Experience Replay:
   - Store subset of old experiences
   - Replay while learning new tasks
   - Maintains performance on old tasks
   - Buffer management is key

2. Regularization:
   - Constrain parameter changes
   - Protect important weights
   - Elastic Weight Consolidation (EWC)
   - Synaptic Intelligence

3. Progressive Networks:
   - Add new capacity for new tasks
   - Keep old networks frozen
   - Lateral connections for transfer
   - No forgetting by design

4. Knowledge Distillation:
   - Transfer knowledge to new model
   - Old model as teacher
   - Soft targets preserve knowledge
   - Compression friendly

5. Memory Consolidation:
   - Identify important experiences
   - Strengthen critical knowledge
   - Hierarchical organization
   - Gradual integration

Experience Buffer Management:
- Fixed size (ring buffer)
- Reservoir sampling
- Priority-based retention
- Diversity sampling
- Importance weighting

Consolidation Process:
1. Group related experiences
2. Extract general patterns
3. Compress into knowledge
4. Store in knowledge base
5. Use for future predictions

Task Priority:
- Critical: Never forget (safety, core capabilities)
- High: Important to retain
- Medium: Moderate importance
- Low: Can forget under pressure

Use Cases:
- Personal Assistants: Learn user preferences over time
- Recommendation Systems: Adapt to changing tastes
- Chatbots: Expand knowledge continuously
- Robotics: Learn new skills without forgetting old
- Adaptive Systems: Evolve with environment
- Multi-task Learning: Master many tasks

Challenges:
1. Memory Limitations: Can't store everything
2. Task Interference: Tasks conflict
3. Generalization: New tasks differ from old
4. Forgetting Measurement: How to evaluate?
5. Computational Cost: Replay and consolidation
6. Task Boundaries: When is a task new?

Best Practices:
1. Use experience replay
2. Consolidate regularly
3. Priority-based retention
4. Monitor all task performance
5. Adjust buffer size appropriately
6. Combine multiple strategies
7. Test for forgetting continuously

Production Considerations:
- Storage requirements for experiences
- Replay frequency and cost
- Consolidation timing
- Performance monitoring
- Graceful degradation
- User feedback integration
- Model versioning

Metrics to Track:
- Forward Transfer: New task learning speed
- Backward Transfer: Impact on old tasks
- Forgetting: Performance degradation
- Memory Efficiency: Knowledge per byte
- Learning Speed: Time to acquire new task

Advanced Techniques:
1. Meta-Learning: Learn how to learn continually
2. Compositional Learning: Reuse components
3. Modular Networks: Task-specific modules
4. Attention Mechanisms: Focus on relevant knowledge
5. Hierarchical Memory: Multi-level organization

Comparison with Related Patterns:
- vs. Meta-Learning: Long-term vs quick adaptation
- vs. Transfer Learning: Continuous vs one-shot
- vs. Multi-Task: Sequential vs simultaneous
- vs. Few-Shot: Many tasks vs few examples

Integration with Other Patterns:
- Memory Patterns: Foundation for continual learning
- Meta-Learning: Quick adaptation to new tasks
- Active Learning: What to learn next?
- Curriculum Learning: Order of learning tasks

Biological Inspiration:
- Complementary Learning Systems
- Systems Consolidation
- Hippocampus-Neocortex interaction
- Sleep and memory replay

Research Directions:
- Scalable to thousands of tasks
- Zero-shot continual learning
- Compositional generalization
- Causal continual learning
- Fairness in continual learning

The Continual Learning pattern enables agents to grow and
adapt continuously while retaining all previously acquired
knowledge - essential for long-lived, adaptive AI systems.
""")


if __name__ == "__main__":
    demonstrate_continual_learning()

    

Agentic Design Pattern: Continual Learning

Pattern 064: Continual Learning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Continual Learning pattern implementation.

---

Pattern Description

Pattern 064: Continual Learning

Description: Continual Learning (also called Lifelong Learning) enables agents to learn continuously from new experiences without forgetting previously acquired knowledge. This addresses the “catastrophic forgetting” problem where neural networks forget old tasks when learning new ones.

Components: 1. Experience Buffer: Stores past experiences 2. Consolidation: Strengthens important memories 3. Regularization: Prevents overwriting old knowledge 4. Knowledge Integration: Merges new with existing knowledge 5. Forgetting Prevention: Various strategies to retain knowledge 6. Performance Monitoring: Tracks across all tasks

Use Cases: – Long-running personal assistants – Adaptive recommendation systems – Evolving domain knowledge – Multi-task learning scenarios – Personalization over time – Dynamic environment adaptation

LangChain Implementation: Implements continual learning using experience replay, knowledge consolidation, and regularization techniques to prevent catastrophic forgetting.

---

File Statistics

• Total Lines: 735
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 064: Continual Learning

Description:
    Continual Learning (also called Lifelong Learning) enables agents to learn
    continuously from new experiences without forgetting previously acquired
    knowledge. This addresses the "catastrophic forgetting" problem where neural
    networks forget old tasks when learning new ones.

Components:
    1. Experience Buffer: Stores past experiences
    2. Consolidation: Strengthens important memories
    3. Regularization: Prevents overwriting old knowledge
    4. Knowledge Integration: Merges new with existing knowledge
    5. Forgetting Prevention: Various strategies to retain knowledge
    6. Performance Monitoring: Tracks across all tasks

Use Cases:
    - Long-running personal assistants
    - Adaptive recommendation systems
    - Evolving domain knowledge
    - Multi-task learning scenarios
    - Personalization over time
    - Dynamic environment adaptation

LangChain Implementation:
    Implements continual learning using experience replay, knowledge consolidation,
    and regularization techniques to prevent catastrophic forgetting.
"""

Purpose: This module implements the Continual Learning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.deque
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – collections.deque: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – random: Random number generation – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ForgettingStrategy

Lines 48-54

Inherits from: Enum

Purpose:

Strategies to prevent catastrophic forgetting

---

Class: TaskPriority

Lines 57-62

Inherits from: Enum

Purpose:

Priority levels for tasks

---

Class: Task

Lines 66-78

Purpose:

Learning task

Methods:

add_example(self, input_text, output_text)

Lines 76-78

Purpose:

Add training example

Parameters: – input_text: Input data – output_text: Output text

---

Class: Experience

Lines 82-90

Purpose:

Single learning experience

---

Class: PerformanceMetric

Lines 94-100

Purpose:

Performance on a task

---

Class: ContinualLearner

Lines 103-389

Purpose:

Continual learning agent.

Features:
1. Learn from new tasks continuously
2. Prevent catastrophic forgetting
3. Selective memory consolidation
4. Experience replay
5. Performance monitoring

Methods:

__init__(self, buffer_size, forgetting_strategy, consolidation_frequency)

Lines 115-150

Purpose: Initialize the class instance with required parameters.

Parameters: – buffer_size: Buffer size – forgetting_strategy: Forgetting strategy – consolidation_frequency: Consolidation frequency

register_task(self, task)

Lines 152-155

Purpose:

Register a new task

Parameters: – task: Task

learn(self, task_id, input_data, output_data, importance)

Lines 157-186

Purpose:

Learn from a single example

Parameters: – task_id: Task id – input_data: Input data – output_data: Output data – importance: Importance

_experience_replay(self)

Lines 188-219

Purpose:

Replay past experiences to prevent forgetting

Parameters:

_consolidate_knowledge(self)

Lines 221-268

Purpose:

Consolidate experiences into knowledge base

Parameters:

predict(self, task_id, input_data, use_consolidation)

Lines 270-319

Purpose:

Make prediction for a task

Parameters: – task_id: Task id – input_data: Input data – use_consolidation: Use consolidation

Returns: str

evaluate(self, task_id, test_cases)

Lines 321-355

Purpose:

Evaluate performance on a task

Parameters: – task_id: Task id – test_cases: Test cases

Returns: PerformanceMetric

get_forgetting_metrics(self)

Lines 357-389

Purpose:

Calculate forgetting metrics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D72B750>

---

Functions

demonstrate_continual_learning()

Lines 392-730

Purpose:

Demonstrate Continual Learning pattern

---

Key Concepts Summary

Pattern Overview

The Continual Learning pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 063
• Pattern 065

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Continual Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 064_continual_learning.py
Total Lines: 735
Pattern Category: AI Agent Design Pattern #064

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 063: Causal Reasoning Agent

Description:
    Causal Reasoning enables agents to understand cause-effect relationships,
    perform interventions, and reason about counterfactuals. Unlike correlation-based
    reasoning, causal reasoning distinguishes between "what happened" and "what would
    happen if we changed something", enabling better decision-making and explanation.

Components:
    1. Causal Graph: Represents causal relationships between variables
    2. Structural Equations: Model mechanisms of causation
    3. Intervention Handler: Simulates "do" operations
    4. Counterfactual Reasoner: Answers "what if" questions
    5. Causal Discovery: Learns causal structure from data
    6. Effect Estimation: Quantifies causal effects

Use Cases:
    - Root cause analysis
    - Decision-making under uncertainty
    - Policy evaluation ("what if we change X?")
    - Scientific reasoning and hypothesis testing
    - Fairness and bias analysis
    - Explanation generation

LangChain Implementation:
    Implements causal reasoning using LLM-based causal graph construction,
    intervention simulation, and counterfactual query answering.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class CausalRelationType(Enum):
    """Types of causal relationships"""
    DIRECT = "direct"  # A → B
    INDIRECT = "indirect"  # A → C → B
    COMMON_CAUSE = "common_cause"  # A ← C → B
    COLLIDER = "collider"  # A → C ← B
    BIDIRECTIONAL = "bidirectional"  # A ↔ B


@dataclass
class CausalEdge:
    """Edge in causal graph"""
    cause: str
    effect: str
    strength: float = 1.0  # 0.0 to 1.0
    mechanism: Optional[str] = None
    
    def __str__(self) -> str:
        arrow = "→" if self.strength > 0 else "⊣"
        strength_str = f"({self.strength:.2f})" if self.strength != 1.0 else ""
        return f"{self.cause} {arrow} {self.effect} {strength_str}"


@dataclass
class CausalGraph:
    """Causal graph structure"""
    variables: Set[str] = field(default_factory=set)
    edges: List[CausalEdge] = field(default_factory=list)
    
    def add_edge(self, cause: str, effect: str, strength: float = 1.0, mechanism: str = None):
        """Add causal edge"""
        self.variables.add(cause)
        self.variables.add(effect)
        self.edges.append(CausalEdge(cause, effect, strength, mechanism))
    
    def get_parents(self, variable: str) -> List[str]:
        """Get direct causes of variable"""
        return [edge.cause for edge in self.edges if edge.effect == variable]
    
    def get_children(self, variable: str) -> List[str]:
        """Get direct effects of variable"""
        return [edge.effect for edge in self.edges if edge.cause == variable]
    
    def get_ancestors(self, variable: str) -> Set[str]:
        """Get all ancestors (transitive causes)"""
        ancestors = set()
        to_visit = self.get_parents(variable)
        
        while to_visit:
            parent = to_visit.pop()
            if parent not in ancestors:
                ancestors.add(parent)
                to_visit.extend(self.get_parents(parent))
        
        return ancestors
    
    def to_text(self) -> str:
        """Convert graph to text"""
        text = "Causal Graph:\n"
        text += f"Variables: {', '.join(sorted(self.variables))}\n\n"
        text += "Edges:\n"
        for edge in self.edges:
            text += f"  {edge}\n"
            if edge.mechanism:
                text += f"    Mechanism: {edge.mechanism}\n"
        return text


@dataclass
class Intervention:
    """Causal intervention (do-operator)"""
    variable: str
    value: Any
    description: str


@dataclass
class CounterfactualQuery:
    """Counterfactual question"""
    observed_facts: Dict[str, Any]
    intervention: Intervention
    query_variable: str
    
    def to_text(self) -> str:
        facts_str = ", ".join(f"{k}={v}" for k, v in self.observed_facts.items())
        return f"Given {facts_str}, what if we {self.intervention.description}? What would {self.query_variable} be?"


@dataclass
class CausalAnswer:
    """Answer to causal query"""
    query: str
    answer: str
    confidence: float
    reasoning: str
    affected_variables: List[str] = field(default_factory=list)


class CausalReasoningAgent:
    """
    Agent capable of causal reasoning.
    
    Features:
    1. Causal graph construction
    2. Intervention simulation
    3. Counterfactual reasoning
    4. Effect estimation
    5. Root cause analysis
    """
    
    def __init__(self, domain: str = "general"):
        self.domain = domain
        self.causal_graph = CausalGraph()
        
        # LLM for causal discovery
        self.discovery_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # LLM for reasoning
        self.reasoning_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
    
    def build_causal_graph(self, context: str) -> CausalGraph:
        """Construct causal graph from context"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are identifying causal relationships in {self.domain}.

Analyze the context and identify:
1. Variables (factors that can change)
2. Causal relationships (A causes B)
3. Mechanisms (how A causes B)

Format your response as:
VARIABLES:
- variable1
- variable2
...

CAUSAL EDGES:
variable1 → variable2 (strength: 0.0-1.0)
Mechanism: explanation

..."""),
            ("user", "Context: {context}\n\nCausal Analysis:")
        ])
        
        chain = prompt | self.discovery_model | StrOutputParser()
        analysis = chain.invoke({"context": context})
        
        # Parse response
        graph = CausalGraph()
        lines = analysis.split('\n')
        
        section = None
        current_edge = None
        
        for line in lines:
            line = line.strip()
            
            if 'VARIABLES:' in line:
                section = 'variables'
            elif 'CAUSAL EDGES:' in line or 'EDGES:' in line:
                section = 'edges'
            elif section == 'variables' and line.startswith('-'):
                var = line.lstrip('- ').strip()
                if var:
                    graph.variables.add(var)
            elif section == 'edges' and ('→' in line or '->' in line):
                # Parse edge
                if '→' in line:
                    parts = line.split('→')
                elif '->' in line:
                    parts = line.split('->')
                else:
                    continue
                
                if len(parts) >= 2:
                    cause = parts[0].strip()
                    effect_part = parts[1].strip()
                    
                    # Extract effect and strength
                    effect = effect_part.split('(')[0].strip()
                    strength = 1.0
                    
                    if '(' in effect_part:
                        try:
                            strength_str = effect_part.split('(')[1].split(')')[0]
                            strength = float(''.join(c for c in strength_str if c.isdigit() or c == '.'))
                            if strength > 1.0:
                                strength = strength / 100.0
                        except:
                            pass
                    
                    current_edge = CausalEdge(cause, effect, strength)
                    graph.edges.append(current_edge)
                    graph.variables.add(cause)
                    graph.variables.add(effect)
            elif section == 'edges' and 'mechanism:' in line.lower() and current_edge:
                mechanism = line.split(':', 1)[1].strip()
                current_edge.mechanism = mechanism
        
        self.causal_graph = graph
        return graph
    
    def do_intervention(
        self,
        intervention: Intervention,
        context: str
    ) -> CausalAnswer:
        """Simulate causal intervention (do-operator)"""
        
        # Build/use causal graph
        if not self.causal_graph.edges:
            self.build_causal_graph(context)
        
        # Find affected variables
        affected = {intervention.variable}
        affected.update(self.causal_graph.get_children(intervention.variable))
        
        # Recursively add downstream effects
        to_check = list(self.causal_graph.get_children(intervention.variable))
        while to_check:
            var = to_check.pop()
            children = self.causal_graph.get_children(var)
            for child in children:
                if child not in affected:
                    affected.add(child)
                    to_check.append(child)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are performing causal intervention analysis in {self.domain}.

Causal Graph:
{self.causal_graph.to_text()}

The intervention sets {intervention.variable} to {intervention.value}.
This is a CAUSAL intervention (do-operator), not just observation.

Analyze:
1. What changes because of this intervention
2. Which variables are affected and how
3. What are the downstream effects

Provide clear reasoning about the causal chain."""),
            ("user", """Context: {context}

Intervention: {intervention}

What are the causal effects?""")
        ])
        
        chain = prompt | self.reasoning_model | StrOutputParser()
        analysis = chain.invoke({
            "context": context,
            "intervention": intervention.description
        })
        
        return CausalAnswer(
            query=f"do({intervention.variable}={intervention.value})",
            answer=analysis,
            confidence=0.7,
            reasoning="Based on causal graph structure and domain knowledge",
            affected_variables=list(affected)
        )
    
    def answer_counterfactual(
        self,
        query: CounterfactualQuery,
        context: str
    ) -> CausalAnswer:
        """Answer counterfactual question"""
        
        # Counterfactuals require:
        # 1. What actually happened (observed facts)
        # 2. What we change (intervention)
        # 3. What we query (counterfactual outcome)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are answering counterfactual questions in {self.domain}.

Counterfactual reasoning has three steps:
1. Abduction: Infer what must be true given observations
2. Action: Apply the intervention
3. Prediction: Determine outcome in this counterfactual world

Use the causal graph to reason about what would change."""),
            ("user", """Context: {context}

Observed Facts: {facts}

Counterfactual Intervention: {intervention}

Question: What would {query_var} be in this counterfactual scenario?

Answer:""")
        ])
        
        chain = prompt | self.reasoning_model | StrOutputParser()
        answer = chain.invoke({
            "context": context,
            "facts": str(query.observed_facts),
            "intervention": query.intervention.description,
            "query_var": query.query_variable
        })
        
        return CausalAnswer(
            query=query.to_text(),
            answer=answer,
            confidence=0.6,  # Counterfactuals are harder
            reasoning="Counterfactual reasoning via abduction-action-prediction"
        )
    
    def identify_root_cause(
        self,
        effect: str,
        context: str
    ) -> CausalAnswer:
        """Identify root cause of an effect"""
        
        if not self.causal_graph.edges:
            self.build_causal_graph(context)
        
        # Get all ancestors
        ancestors = self.causal_graph.get_ancestors(effect)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are performing root cause analysis in {self.domain}.

Causal Graph:
{self.causal_graph.to_text()}

Identify the ROOT CAUSE(S) of the observed effect.
Root causes are variables with:
1. No upstream causes (or only external causes)
2. Strong causal connection to the effect
3. Actionable or controllable

Trace back through the causal chain."""),
            ("user", """Context: {context}

Observed Effect: {effect}

Potential Causes: {ancestors}

What is the root cause?""")
        ])
        
        chain = prompt | self.reasoning_model | StrOutputParser()
        analysis = chain.invoke({
            "context": context,
            "effect": effect,
            "ancestors": ", ".join(ancestors) if ancestors else "None identified"
        })
        
        return CausalAnswer(
            query=f"Root cause of: {effect}",
            answer=analysis,
            confidence=0.75,
            reasoning="Traced causal chain to root variables"
        )


def demonstrate_causal_reasoning():
    """Demonstrate Causal Reasoning Agent pattern"""
    
    print("=" * 80)
    print("PATTERN 063: CAUSAL REASONING AGENT DEMONSTRATION")
    print("=" * 80)
    print("\nUnderstanding cause-effect relationships and counterfactuals\n")
    
    # Test 1: Building causal graph
    print("\n" + "=" * 80)
    print("TEST 1: Causal Graph Construction")
    print("=" * 80)
    
    agent = CausalReasoningAgent(domain="business")
    
    context1 = """
    A company has been experiencing declining sales. Analysis shows:
    - Marketing budget was reduced
    - Competitor launched new product
    - Customer satisfaction scores dropped
    - Website traffic decreased
    - Sales team morale is low
    The reduced marketing budget led to less advertising, which decreased website 
    traffic. The competitor's new product attracted customers, reducing market share.
    Lower sales affected team morale.
    """
    
    print(f"\n📊 Analyzing business scenario...")
    
    graph = agent.build_causal_graph(context1)
    
    print(f"\n🔗 Discovered Causal Graph:")
    print(f"   Variables: {len(graph.variables)}")
    print(f"   Edges: {len(graph.edges)}")
    print(f"\n{graph.to_text()}")
    
    # Test 2: Intervention analysis
    print("\n" + "=" * 80)
    print("TEST 2: Causal Intervention (do-operator)")
    print("=" * 80)
    
    intervention = Intervention(
        variable="marketing budget",
        value="increase by 50%",
        description="increase marketing budget by 50%"
    )
    
    print(f"\n🎯 Intervention: {intervention.description}")
    print(f"   Question: What causal effects will this have?")
    
    result = agent.do_intervention(intervention, context1)
    
    print(f"\n💡 Causal Effects:")
    print(f"   Affected Variables: {', '.join(result.affected_variables)}")
    print(f"\n   Analysis:")
    for line in result.answer.split('\n')[:10]:  # First 10 lines
        if line.strip():
            print(f"   {line}")
    
    # Test 3: Counterfactual reasoning
    print("\n" + "=" * 80)
    print("TEST 3: Counterfactual Reasoning")
    print("=" * 80)
    
    counterfactual = CounterfactualQuery(
        observed_facts={
            "marketing_budget": "reduced",
            "sales": "declined",
            "market_share": "lost"
        },
        intervention=Intervention(
            variable="marketing budget",
            value="maintained at original level",
            description="maintained marketing budget at original level"
        ),
        query_variable="sales"
    )
    
    print(f"\n❓ Counterfactual Question:")
    print(f"   {counterfactual.to_text()}")
    
    cf_result = agent.answer_counterfactual(counterfactual, context1)
    
    print(f"\n🔮 Counterfactual Answer:")
    print(f"   Confidence: {cf_result.confidence:.2%}")
    print(f"\n   Answer:")
    for line in cf_result.answer.split('\n')[:8]:
        if line.strip():
            print(f"   {line}")
    
    # Test 4: Root cause analysis
    print("\n" + "=" * 80)
    print("TEST 4: Root Cause Analysis")
    print("=" * 80)
    
    agent2 = CausalReasoningAgent(domain="software systems")
    
    context2 = """
    A web application is experiencing slow response times:
    - Database queries are taking longer
    - Server CPU usage is high
    - Memory usage increased
    - Number of active users grew
    - Cache hit rate decreased
    - Recent code deployment added complex queries
    
    The code deployment introduced inefficient queries, which increased
    database load. Growing user base put additional load on the system.
    High server load caused slower query execution. Poor cache performance
    meant more database hits.
    """
    
    print(f"\n🔍 Analyzing system performance issue...")
    
    agent2.build_causal_graph(context2)
    
    root_cause = agent2.identify_root_cause("slow response times", context2)
    
    print(f"\n🎯 Root Cause Analysis:")
    print(f"   Effect: slow response times")
    print(f"   Confidence: {root_cause.confidence:.2%}")
    print(f"\n   Root Cause(s):")
    for line in root_cause.answer.split('\n'):
        if line.strip():
            print(f"   {line}")
    
    # Test 5: Comparing correlation vs causation
    print("\n" + "=" * 80)
    print("TEST 5: Correlation vs Causation")
    print("=" * 80)
    
    agent3 = CausalReasoningAgent(domain="healthcare")
    
    context3 = """
    Study observations:
    - People who drink coffee have lower rates of heart disease
    - People who exercise regularly drink more coffee
    - People who exercise have better heart health
    - Coffee consumption and exercise are correlated
    - Income level affects both coffee drinking and exercise
    """
    
    print(f"\n📈 Analyzing correlation pattern...")
    
    graph3 = agent3.build_causal_graph(context3)
    print(f"\n   Causal Graph:")
    for edge in graph3.edges:
        print(f"   {edge}")
    
    # Compare two interventions
    print(f"\n   Comparing interventions:")
    
    int1 = Intervention("coffee consumption", "increase", "increase coffee consumption")
    int2 = Intervention("exercise", "increase", "increase exercise")
    
    result1 = agent3.do_intervention(int1, context3)
    result2 = agent3.do_intervention(int2, context3)
    
    print(f"\n   If we increase COFFEE consumption:")
    print(f"      Affected: {', '.join(result1.affected_variables)}")
    
    print(f"\n   If we increase EXERCISE:")
    print(f"      Affected: {', '.join(result2.affected_variables)}")
    
    print(f"\n   ⚠️  Causal vs Correlational insight:")
    print(f"      Correlation doesn't imply causation!")
    print(f"      The causal structure reveals true mechanisms.")
    
    # Summary
    print("\n" + "=" * 80)
    print("CAUSAL REASONING AGENT PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. True Understanding: Distinguish causation from correlation
2. Better Decisions: Predict intervention outcomes
3. Counterfactual Reasoning: Answer "what if" questions
4. Root Cause Analysis: Find true sources of problems
5. Robust Generalization: Transfer across contexts

Causal Hierarchy (Pearl's Ladder):
1. Association (Seeing): P(Y|X)
   - "What is?" (observation)
   - Correlations and patterns
   - Machine learning operates here

2. Intervention (Doing): P(Y|do(X))
   - "What if?" (action)
   - Effects of interventions
   - Experiments and A/B tests

3. Counterfactuals (Imagining): P(Y_x|X',Y')
   - "What if I had?" (retrospection)
   - Alternative histories
   - Requires full causal model

Core Concepts:
- Causal Graph: Directed graph of cause-effect
- Structural Equations: Mechanisms of causation
- do-Operator: Intervention (breaking incoming edges)
- Backdoor Criterion: Identifying confounders
- Frontdoor Criterion: Alternative identification
- Counterfactual: Alternative possible worlds

Types of Causal Questions:
1. Association: "Are X and Y related?"
2. Intervention: "What if we do X?"
3. Counterfactual: "What if we had done X?"
4. Effect Size: "How much does X affect Y?"
5. Attribution: "Did X cause this Y?"
6. Explanation: "Why did Y happen?"

Causal Graph Patterns:
- Chain: A → B → C (mediation)
- Fork: A ← B → C (common cause/confounder)
- Collider: A → B ← C (selection bias)
- Confounding: A ← C → Y (spurious correlation)

Intervention vs Observation:
- Observation: P(Y|X=x) - passive seeing
- Intervention: P(Y|do(X=x)) - active doing
- Key difference: Intervention breaks incoming edges

Use Cases:
- Medicine: Treatment effect estimation
- Policy: Evaluate policy interventions
- Business: Understand driver of metrics
- Science: Hypothesis testing
- Fairness: Identify discrimination
- Debugging: Root cause analysis
- Legal: Attribution and liability

Causal Discovery Methods:
1. Constraint-Based: Test conditional independencies
2. Score-Based: Search for best-fitting graph
3. Functional: Exploit noise structure
4. Hybrid: Combine approaches
5. LLM-Based: Extract from text/knowledge

Challenges:
1. Unobserved Confounders: Hidden common causes
2. Selection Bias: Collider conditioning
3. Measurement Error: Imperfect variables
4. Feedback Loops: Bidirectional causation
5. Time Delays: Lag between cause and effect
6. Context Dependence: Different mechanisms

Best Practices:
1. Draw causal graph explicitly
2. Identify confounders
3. Use randomization when possible
4. Combine domain knowledge + data
5. Test causal assumptions
6. Estimate uncertainty
7. Consider alternative graphs

Production Considerations:
- Causal graph versioning
- Domain expert validation
- A/B testing for verification
- Sensitivity analysis
- Handling edge cases
- Explanation generation
- Confidence calibration

Advanced Techniques:
1. Causal Effect Estimation
   - Instrumental variables
   - Difference-in-differences
   - Regression discontinuity
   - Propensity score matching

2. Mediation Analysis
   - Direct vs indirect effects
   - Path analysis

3. Time-Series Causality
   - Granger causality
   - Dynamic causal models

4. Causal Reinforcement Learning
   - Credit assignment
   - Off-policy evaluation

Comparison with Related Patterns:
- vs. Correlation: Distinguishes cause from association
- vs. Prediction: Explains mechanism, not just predicts
- vs. Counterfactual (simple): Full causal framework
- vs. Analogical: Causal structure vs similarity

Integration with Other Patterns:
- World Models: Causal = mechanism in world model
- Planning: Causal graph guides planning
- Explanation: Causal = better explanations
- Reasoning: Foundation for robust reasoning

Philosophical Note:
Causal reasoning represents understanding of "why" not just
"what". It's the difference between prediction (correlation)
and understanding (causation). Essential for AGI.

The Causal Reasoning Agent pattern enables agents to move
beyond correlation to true causal understanding, supporting
better decisions, explanations, and counterfactual reasoning.
""")


if __name__ == "__main__":
    demonstrate_causal_reasoning()

    

Agentic Design Pattern: Causal Reasoning

Pattern 063: Causal Reasoning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Causal Reasoning pattern implementation.

---

Pattern Description

Pattern 063: Causal Reasoning Agent

Description: Causal Reasoning enables agents to understand cause-effect relationships, perform interventions, and reason about counterfactuals. Unlike correlation-based reasoning, causal reasoning distinguishes between “what happened” and “what would happen if we changed something”, enabling better decision-making and explanation.

Components: 1. Causal Graph: Represents causal relationships between variables 2. Structural Equations: Model mechanisms of causation 3. Intervention Handler: Simulates “do” operations 4. Counterfactual Reasoner: Answers “what if” questions 5. Causal Discovery: Learns causal structure from data 6. Effect Estimation: Quantifies causal effects

Use Cases: – Root cause analysis – Decision-making under uncertainty – Policy evaluation (“what if we change X?”) – Scientific reasoning and hypothesis testing – Fairness and bias analysis – Explanation generation

LangChain Implementation: Implements causal reasoning using LLM-based causal graph construction, intervention simulation, and counterfactual query answering.

---

File Statistics

• Total Lines: 722
• Classes: 7
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 063: Causal Reasoning Agent

Description:
    Causal Reasoning enables agents to understand cause-effect relationships,
    perform interventions, and reason about counterfactuals. Unlike correlation-based
    reasoning, causal reasoning distinguishes between "what happened" and "what would
    happen if we changed something", enabling better decision-making and explanation.

Components:
    1. Causal Graph: Represents causal relationships between variables
    2. Structural Equations: Model mechanisms of causation
    3. Intervention Handler: Simulates "do" operations
    4. Counterfactual Reasoner: Answers "what if" questions
    5. Causal Discovery: Learns causal structure from data
    6. Effect Estimation: Quantifies causal effects

Use Cases:
    - Root cause analysis
    - Decision-making under uncertainty
    - Policy evaluation ("what if we change X?")
    - Scientific reasoning and hypothesis testing
    - Fairness and bias analysis
    - Explanation generation

LangChain Implementation:
    Implements causal reasoning using LLM-based causal graph construction,
    intervention simulation, and counterfactual query answering.
"""

Purpose: This module implements the Causal Reasoning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: CausalRelationType

Lines 46-52

Inherits from: Enum

Purpose:

Types of causal relationships

---

Class: CausalEdge

Lines 56-66

Purpose:

Edge in causal graph

Methods:

__str__(self)

Lines 63-66

Purpose: Str operation.

Parameters:

Returns: str

---

Class: CausalGraph

Lines 70-111

Purpose:

Causal graph structure

Methods:

add_edge(self, cause, effect, strength, mechanism)

Lines 75-79

Purpose:

Add causal edge

Parameters: – cause: Cause – effect: Effect – strength: Strength – mechanism: Mechanism

get_parents(self, variable)

Lines 81-83

Purpose:

Get direct causes of variable

Parameters: – variable: Variable

Returns: <ast.Subscript object at 0x000001A06D68FD10>

get_children(self, variable)

Lines 85-87

Purpose:

Get direct effects of variable

Parameters: – variable: Variable

Returns: <ast.Subscript object at 0x000001A06D159310>

get_ancestors(self, variable)

Lines 89-100

Purpose:

Get all ancestors (transitive causes)

Parameters: – variable: Variable

Returns: <ast.Subscript object at 0x000001A06D15BA50>

to_text(self)

Lines 102-111

Purpose:

Convert graph to text

Parameters:

Returns: str

---

Class: Intervention

Lines 115-119

Purpose:

Causal intervention (do-operator)

---

Class: CounterfactualQuery

Lines 123-131

Purpose:

Counterfactual question

Methods:

to_text(self)

Lines 129-131

Purpose: To Text operation.

Parameters:

Returns: str

---

Class: CausalAnswer

Lines 135-141

Purpose:

Answer to causal query

---

Class: CausalReasoningAgent

Lines 144-410

Purpose:

Agent capable of causal reasoning.

Features:
1. Causal graph construction
2. Intervention simulation
3. Counterfactual reasoning
4. Effect estimation
5. Root cause analysis

Methods:

__init__(self, domain)

Lines 156-170

Purpose: Initialize the class instance with required parameters.

Parameters: – domain: Domain

build_causal_graph(self, context)

Lines 172-253

Purpose:

Construct causal graph from context

Parameters: – context: Context

Returns: CausalGraph

do_intervention(self, intervention, context)

Lines 255-314

Purpose:

Simulate causal intervention (do-operator)

Parameters: – intervention: Intervention – context: Context

Returns: CausalAnswer

answer_counterfactual(self, query, context)

Lines 316-361

Purpose:

Answer counterfactual question

Parameters: – query: Input query or question – context: Context

Returns: CausalAnswer

identify_root_cause(self, effect, context)

Lines 363-410

Purpose:

Identify root cause of an effect

Parameters: – effect: Effect – context: Context

Returns: CausalAnswer

---

Functions

demonstrate_causal_reasoning()

Lines 413-717

Purpose:

Demonstrate Causal Reasoning Agent pattern

---

Key Concepts Summary

Pattern Overview

The Causal Reasoning pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 062
• Pattern 064

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Causal Reasoning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 063_causal_reasoning.py
Total Lines: 722
Pattern Category: AI Agent Design Pattern #063

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 062: World Model Learning

Description:
    World Model Learning involves an agent building an internal model of its
    environment's dynamics. The agent learns to predict future states given
    current states and actions, enabling model-based planning, simulation,
    and what-if analysis without direct environment interaction.

Components:
    1. State Representation: How the world is encoded
    2. Dynamics Model: Predicts next state from current state and action
    3. Reward Predictor: Estimates rewards for state-action pairs
    4. Experience Collection: Gathers real-world data
    5. Model Training: Updates predictions based on experience
    6. Planning Module: Uses model for decision-making

Use Cases:
    - Robotics simulation and planning
    - Game AI with environment modeling
    - Strategic decision-making
    - Risk assessment and scenario planning
    - Sample-efficient reinforcement learning
    - Counterfactual reasoning

LangChain Implementation:
    Implements world modeling using LLM-based state prediction, transition
    dynamics, and outcome forecasting for planning and decision-making.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class StateType(Enum):
    """Types of state representations"""
    DISCRETE = "discrete"  # Finite states
    CONTINUOUS = "continuous"  # Continuous values
    HYBRID = "hybrid"  # Mix of both
    SYMBOLIC = "symbolic"  # Symbolic/linguistic


@dataclass
class State:
    """World state representation"""
    state_id: str
    description: str
    variables: Dict[str, Any] = field(default_factory=dict)
    timestamp: float = field(default_factory=time.time)
    
    def to_text(self) -> str:
        """Convert state to text"""
        text = f"State: {self.description}\n"
        if self.variables:
            text += "Variables:\n"
            for key, value in self.variables.items():
                text += f"  - {key}: {value}\n"
        return text


@dataclass
class Action:
    """Action that can be taken"""
    action_id: str
    name: str
    parameters: Dict[str, Any] = field(default_factory=dict)
    
    def to_text(self) -> str:
        """Convert action to text"""
        text = f"Action: {self.name}"
        if self.parameters:
            params_str = ", ".join(f"{k}={v}" for k, v in self.parameters.items())
            text += f" ({params_str})"
        return text


@dataclass
class Transition:
    """State transition experience"""
    state: State
    action: Action
    next_state: State
    reward: float
    info: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Prediction:
    """Model prediction"""
    predicted_state: State
    confidence: float
    predicted_reward: float
    alternative_outcomes: List[Tuple[State, float]] = field(default_factory=list)


@dataclass
class SimulationResult:
    """Result from simulation"""
    initial_state: State
    action_sequence: List[Action]
    predicted_trajectory: List[State]
    cumulative_reward: float
    success_probability: float


class WorldModel:
    """
    World model for environment dynamics.
    
    Features:
    1. State transition prediction
    2. Reward prediction
    3. Experience-based learning
    4. Multi-step simulation
    5. Uncertainty estimation
    """
    
    def __init__(self, domain: str = "general"):
        self.domain = domain
        self.experiences: List[Transition] = []
        
        # Model for dynamics prediction
        self.dynamics_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # Model for reward prediction
        self.reward_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
        
        # Model for uncertainty estimation
        self.uncertainty_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.4
        )
    
    def add_experience(self, transition: Transition):
        """Add experience to model's memory"""
        self.experiences.append(transition)
    
    def _get_relevant_experiences(
        self,
        state: State,
        action: Action,
        k: int = 3
    ) -> List[Transition]:
        """Retrieve relevant past experiences"""
        
        if not self.experiences:
            return []
        
        # Simple relevance: recent experiences are more relevant
        # In production, would use embedding similarity
        return self.experiences[-k:] if len(self.experiences) >= k else self.experiences
    
    def predict_next_state(
        self,
        current_state: State,
        action: Action
    ) -> Prediction:
        """Predict next state given current state and action"""
        
        # Retrieve relevant experiences
        relevant = self._get_relevant_experiences(current_state, action)
        
        # Build context from experiences
        context = ""
        if relevant:
            context = "Past Similar Experiences:\n"
            for i, exp in enumerate(relevant, 1):
                context += f"{i}. State: {exp.state.description}\n"
                context += f"   Action: {exp.action.name}\n"
                context += f"   Result: {exp.next_state.description}\n"
                context += f"   Reward: {exp.reward}\n\n"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are a world model for {self.domain}.

Predict what happens when an action is taken in a given state.

{context}

Provide:
1. Description of the resulting state
2. Key variables that changed
3. Confidence (0.0-1.0) in prediction"""),
            ("user", """Current State:
{state}

Action Taken:
{action}

Predicted Next State:""")
        ])
        
        chain = prompt | self.dynamics_model | StrOutputParser()
        prediction_text = chain.invoke({
            "state": current_state.to_text(),
            "action": action.to_text()
        })
        
        # Parse prediction
        lines = prediction_text.strip().split('\n')
        description = lines[0] if lines else "Unknown state"
        
        # Extract confidence if mentioned
        confidence = 0.7  # Default
        for line in lines:
            if 'confidence' in line.lower():
                try:
                    confidence = float(''.join(c for c in line if c.isdigit() or c == '.'))
                    if confidence > 1.0:
                        confidence = confidence / 100.0
                    confidence = max(0.0, min(1.0, confidence))
                except:
                    pass
        
        # Predict reward
        predicted_reward = self._predict_reward(current_state, action, description)
        
        predicted_state = State(
            state_id=f"predicted_{int(time.time())}",
            description=description,
            variables={}
        )
        
        return Prediction(
            predicted_state=predicted_state,
            confidence=confidence,
            predicted_reward=predicted_reward
        )
    
    def _predict_reward(
        self,
        state: State,
        action: Action,
        outcome: str
    ) -> float:
        """Predict reward for state-action-outcome"""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""Estimate the reward/value for an outcome in {self.domain}.

Provide a number between -1.0 (very bad) and +1.0 (very good).
Consider: goal achievement, efficiency, safety, side effects.

Respond with just a number."""),
            ("user", """State: {state}
Action: {action}
Outcome: {outcome}

Reward:""")
        ])
        
        chain = prompt | self.reward_model | StrOutputParser()
        
        try:
            reward_str = chain.invoke({
                "state": state.description,
                "action": action.name,
                "outcome": outcome
            })
            reward = float(''.join(c for c in reward_str if c.isdigit() or c == '.' or c == '-'))
            return max(-1.0, min(1.0, reward))
        except:
            return 0.0
    
    def simulate_trajectory(
        self,
        initial_state: State,
        action_sequence: List[Action],
        max_steps: int = 10
    ) -> SimulationResult:
        """Simulate a trajectory through the world"""
        
        current_state = initial_state
        trajectory = [initial_state]
        cumulative_reward = 0.0
        confidences = []
        
        for i, action in enumerate(action_sequence[:max_steps]):
            # Predict next state
            prediction = self.predict_next_state(current_state, action)
            
            # Update trajectory
            trajectory.append(prediction.predicted_state)
            cumulative_reward += prediction.predicted_reward
            confidences.append(prediction.confidence)
            
            # Move to next state
            current_state = prediction.predicted_state
        
        # Success probability = average confidence
        success_prob = sum(confidences) / len(confidences) if confidences else 0.5
        
        return SimulationResult(
            initial_state=initial_state,
            action_sequence=action_sequence,
            predicted_trajectory=trajectory,
            cumulative_reward=cumulative_reward,
            success_probability=success_prob
        )
    
    def plan_with_model(
        self,
        current_state: State,
        goal: str,
        num_alternatives: int = 3
    ) -> List[SimulationResult]:
        """Use model for planning"""
        
        # Generate alternative action sequences
        planning_prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are planning in {self.domain}.

Generate {num_alternatives} different action sequences to achieve the goal.

For each plan, list 3-5 actions.

Format:
Plan 1:
1. [action]
2. [action]
...

Plan 2:
1. [action]
..."""),
            ("user", """Current State: {state}

Goal: {goal}

Action Plans:""")
        ])
        
        chain = planning_prompt | self.dynamics_model | StrOutputParser()
        plans_text = chain.invoke({
            "state": current_state.to_text(),
            "goal": goal
        })
        
        # Parse action sequences
        action_sequences = []
        current_sequence = []
        
        for line in plans_text.split('\n'):
            line = line.strip()
            
            if line.startswith('Plan'):
                if current_sequence:
                    action_sequences.append(current_sequence)
                    current_sequence = []
            elif line and (line[0].isdigit() or line.startswith('-')):
                action_name = line.lstrip('0123456789.-) ').strip()
                if action_name:
                    action = Action(
                        action_id=f"action_{len(current_sequence)}",
                        name=action_name
                    )
                    current_sequence.append(action)
        
        if current_sequence:
            action_sequences.append(current_sequence)
        
        # Simulate each sequence
        simulations = []
        for seq in action_sequences[:num_alternatives]:
            if seq:  # Only simulate non-empty sequences
                sim = self.simulate_trajectory(current_state, seq)
                simulations.append(sim)
        
        # Sort by expected reward
        simulations.sort(key=lambda s: s.cumulative_reward, reverse=True)
        
        return simulations


def demonstrate_world_model():
    """Demonstrate World Model Learning pattern"""
    
    print("=" * 80)
    print("PATTERN 062: WORLD MODEL LEARNING DEMONSTRATION")
    print("=" * 80)
    print("\nLearning environment dynamics for model-based planning\n")
    
    # Test 1: Building a world model from experience
    print("\n" + "=" * 80)
    print("TEST 1: Learning from Experience")
    print("=" * 80)
    
    world = WorldModel(domain="business strategy")
    
    # Add some experiences
    print("\n📚 Adding experiences to world model...")
    
    experiences = [
        Transition(
            state=State("s1", "Low market share, limited brand awareness"),
            action=Action("a1", "Launch aggressive marketing campaign"),
            next_state=State("s2", "Increased brand visibility, higher costs"),
            reward=0.3
        ),
        Transition(
            state=State("s2", "Increased brand visibility, higher costs"),
            action=Action("a2", "Optimize pricing strategy"),
            next_state=State("s3", "Improved market share, balanced budget"),
            reward=0.7
        ),
        Transition(
            state=State("s3", "Improved market share, balanced budget"),
            action=Action("a3", "Expand product line"),
            next_state=State("s4", "Diversified revenue, market leadership"),
            reward=0.9
        )
    ]
    
    for exp in experiences:
        world.add_experience(exp)
        print(f"   ✓ {exp.state.description} --[{exp.action.name}]--> {exp.next_state.description}")
    
    print(f"\n   Total experiences: {len(world.experiences)}")
    
    # Test 2: Predicting next state
    print("\n" + "=" * 80)
    print("TEST 2: State Prediction")
    print("=" * 80)
    
    current_state = State(
        "s_test",
        "New competitor entering market"
    )
    
    test_action = Action(
        "a_test",
        "Form strategic partnership"
    )
    
    print(f"\n🔮 Making prediction...")
    print(f"   Current: {current_state.description}")
    print(f"   Action: {test_action.name}")
    
    prediction = world.predict_next_state(current_state, test_action)
    
    print(f"\n✨ Predicted outcome:")
    print(f"   State: {prediction.predicted_state.description}")
    print(f"   Confidence: {prediction.confidence:.2f}")
    print(f"   Expected Reward: {prediction.predicted_reward:.2f}")
    
    # Test 3: Multi-step simulation
    print("\n" + "=" * 80)
    print("TEST 3: Trajectory Simulation")
    print("=" * 80)
    
    initial = State("s_start", "Startup with innovative product")
    
    actions = [
        Action("a1", "Secure seed funding"),
        Action("a2", "Build MVP"),
        Action("a3", "Beta testing with early adopters"),
        Action("a4", "Launch marketing campaign")
    ]
    
    print(f"\n🎬 Simulating trajectory...")
    print(f"   Initial State: {initial.description}")
    print(f"   Actions: {len(actions)}")
    
    simulation = world.simulate_trajectory(initial, actions)
    
    print(f"\n📊 Simulation Results:")
    print(f"   Success Probability: {simulation.success_probability:.2%}")
    print(f"   Cumulative Reward: {simulation.cumulative_reward:.2f}")
    print(f"\n   Trajectory:")
    for i, state in enumerate(simulation.predicted_trajectory):
        if i == 0:
            print(f"   0. [START] {state.description}")
        else:
            action = simulation.action_sequence[i-1]
            print(f"   {i}. --[{action.name}]--> {state.description}")
    
    # Test 4: Model-based planning
    print("\n" + "=" * 80)
    print("TEST 4: Model-Based Planning")
    print("=" * 80)
    
    planning_state = State(
        "s_plan",
        "Established company facing digital transformation"
    )
    
    goal = "Successfully transition to cloud-based operations"
    
    print(f"\n🎯 Planning Goal: {goal}")
    print(f"   Current State: {planning_state.description}")
    print(f"\n   Generating alternative plans...")
    
    plans = world.plan_with_model(planning_state, goal, num_alternatives=3)
    
    print(f"\n📋 Generated {len(plans)} alternative plans:\n")
    
    for i, plan in enumerate(plans, 1):
        print(f"   Plan {i}:")
        print(f"   Expected Reward: {plan.cumulative_reward:.2f}")
        print(f"   Success Probability: {plan.success_probability:.2%}")
        print(f"   Actions:")
        for j, action in enumerate(plan.action_sequence, 1):
            print(f"      {j}. {action.name}")
        print()
    
    if plans:
        print(f"   🏆 Recommended: Plan 1 (highest expected reward)")
    
    # Test 5: What-if analysis
    print("\n" + "=" * 80)
    print("TEST 5: What-If Analysis")
    print("=" * 80)
    
    world2 = WorldModel(domain="project management")
    
    project_state = State("proj", "Project behind schedule, team morale low")
    
    scenarios = [
        ("Hire additional developers", Action("a1", "Hire additional developers")),
        ("Reduce scope", Action("a2", "Reduce project scope")),
        ("Extend deadline", Action("a3", "Negotiate deadline extension"))
    ]
    
    print(f"\n🔍 Analyzing scenarios for: {project_state.description}\n")
    
    for scenario_name, action in scenarios:
        prediction = world2.predict_next_state(project_state, action)
        
        print(f"   Scenario: {scenario_name}")
        print(f"      Predicted outcome: {prediction.predicted_state.description}")
        print(f"      Expected reward: {prediction.predicted_reward:+.2f}")
        print(f"      Confidence: {prediction.confidence:.2%}")
        print()
    
    # Summary
    print("\n" + "=" * 80)
    print("WORLD MODEL LEARNING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Sample Efficiency: Learn from limited experience
2. Planning Capability: Simulate before acting
3. Risk Assessment: Evaluate outcomes without execution
4. What-If Analysis: Compare alternative scenarios
5. Transfer Learning: Generalize across similar situations

World Model Components:
1. State Representation: How world is encoded
   - Discrete states (finite set)
   - Continuous states (real values)
   - Symbolic states (descriptions)
   - Hybrid (combination)

2. Dynamics Model: Predicts transitions
   - Input: Current state + Action
   - Output: Next state prediction
   - Confidence estimation

3. Reward Model: Estimates outcomes
   - Input: State + Action + Result
   - Output: Expected reward/value
   - Risk assessment

4. Experience Buffer: Stores transitions
   - Real experiences
   - Simulated experiences
   - Relevance-based retrieval

5. Planning Module: Uses model
   - Forward simulation
   - Trajectory optimization
   - Alternative comparison

Learning Process:
1. Collect Experience: Interact with environment
2. Update Model: Learn from transitions
3. Validate Predictions: Test accuracy
4. Refine Model: Improve based on errors
5. Plan Actions: Use model for decisions

Model Types:
1. Forward Model: State → Next State
   - Most common
   - Direct prediction

2. Inverse Model: Current State + Next State → Action
   - Action discovery
   - Imitation learning

3. Reward Model: State + Action → Reward
   - Value estimation
   - Policy evaluation

4. Termination Model: State → Done?
   - Episode completion
   - Goal achievement

Prediction Strategies:
- Deterministic: Single prediction
- Stochastic: Sample from distribution
- Ensemble: Multiple models
- Uncertainty-aware: Confidence bounds

Use Cases:
- Robotics: Simulate movements before execution
- Game AI: Plan strategies by modeling game
- Business: Scenario planning and forecasting
- Healthcare: Treatment outcome prediction
- Finance: Market behavior modeling
- Education: Learning path optimization

Planning with World Models:
1. Model-Predictive Control (MPC)
   - Rolling horizon planning
   - Re-plan at each step

2. Tree Search
   - Build search tree using model
   - Evaluate trajectories

3. Trajectory Optimization
   - Find optimal action sequence
   - Gradient-based or sampling

4. Policy Learning
   - Use model to improve policy
   - Dyna-style algorithms

Challenges:
1. Model Accuracy: Prediction errors compound
2. Generalization: Unseen states
3. Computational Cost: Simulation overhead
4. Model Bias: Learned from limited data
5. Non-stationarity: Environment changes

Best Practices:
1. Start with simple models
2. Validate predictions frequently
3. Use ensemble of models
4. Estimate uncertainty
5. Combine model-based and model-free
6. Update model continuously
7. Use for planning, not just prediction

Production Considerations:
- Model versioning and updates
- Prediction latency requirements
- Memory for experience buffer
- Model accuracy metrics
- Fallback to model-free
- A/B testing model changes
- Continuous validation

Advanced Techniques:
1. Latent World Models: Learn compressed representations
2. Hierarchical Models: Multi-scale predictions
3. Causal Models: Understand mechanisms
4. Meta-Learning Models: Quick adaptation
5. Uncertainty-Aware Models: Calibrated predictions

Comparison with Related Patterns:
- vs. Model-Free RL: More sample efficient
- vs. Imitation Learning: Enables planning
- vs. Heuristic Planning: Data-driven
- vs. Simulation: Learned not programmed

Integration with Other Patterns:
- Curriculum Learning: Progressive model building
- Meta-Learning: Quick model adaptation
- Active Learning: Query informative states
- Memory Patterns: Store experiences

The World Model Learning pattern enables agents to learn
environment dynamics from experience and use these models
for efficient planning, simulation, and decision-making.
""")


if __name__ == "__main__":
    demonstrate_world_model()

    

Agentic Design Pattern: World Model

Pattern 062: World Model – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the World Model pattern implementation.

---

Pattern Description

Pattern 062: World Model Learning

Description: World Model Learning involves an agent building an internal model of its environment’s dynamics. The agent learns to predict future states given current states and actions, enabling model-based planning, simulation, and what-if analysis without direct environment interaction.

Components: 1. State Representation: How the world is encoded 2. Dynamics Model: Predicts next state from current state and action 3. Reward Predictor: Estimates rewards for state-action pairs 4. Experience Collection: Gathers real-world data 5. Model Training: Updates predictions based on experience 6. Planning Module: Uses model for decision-making

Use Cases: – Robotics simulation and planning – Game AI with environment modeling – Strategic decision-making – Risk assessment and scenario planning – Sample-efficient reinforcement learning – Counterfactual reasoning

LangChain Implementation: Implements world modeling using LLM-based state prediction, transition dynamics, and outcome forecasting for planning and decision-making.

---

File Statistics

• Total Lines: 699
• Classes: 7
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 062: World Model Learning

Description:
    World Model Learning involves an agent building an internal model of its
    environment's dynamics. The agent learns to predict future states given
    current states and actions, enabling model-based planning, simulation,
    and what-if analysis without direct environment interaction.

Components:
    1. State Representation: How the world is encoded
    2. Dynamics Model: Predicts next state from current state and action
    3. Reward Predictor: Estimates rewards for state-action pairs
    4. Experience Collection: Gathers real-world data
    5. Model Training: Updates predictions based on experience
    6. Planning Module: Uses model for decision-making

Use Cases:
    - Robotics simulation and planning
    - Game AI with environment modeling
    - Strategic decision-making
    - Risk assessment and scenario planning
    - Sample-efficient reinforcement learning
    - Counterfactual reasoning

LangChain Implementation:
    Implements world modeling using LLM-based state prediction, transition
    dynamics, and outcome forecasting for planning and decision-making.
"""

Purpose: This module implements the World Model pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – random: Random number generation – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: StateType

Lines 47-52

Inherits from: Enum

Purpose:

Types of state representations

---

Class: State

Lines 56-70

Purpose:

World state representation

Methods:

to_text(self)

Lines 63-70

Purpose:

Convert state to text

Parameters:

Returns: str

---

Class: Action

Lines 74-86

Purpose:

Action that can be taken

Methods:

to_text(self)

Lines 80-86

Purpose:

Convert action to text

Parameters:

Returns: str

---

Class: Transition

Lines 90-96

Purpose:

State transition experience

---

Class: Prediction

Lines 100-105

Purpose:

Model prediction

---

Class: SimulationResult

Lines 109-115

Purpose:

Result from simulation

---

Class: WorldModel

Lines 118-390

Purpose:

World model for environment dynamics.

Features:
1. State transition prediction
2. Reward prediction
3. Experience-based learning
4. Multi-step simulation
5. Uncertainty estimation

Methods:

__init__(self, domain)

Lines 130-150

Purpose: Initialize the class instance with required parameters.

Parameters: – domain: Domain

add_experience(self, transition)

Lines 152-154

Purpose:

Add experience to model's memory

Parameters: – transition: Transition

_get_relevant_experiences(self, state, action, k)

Lines 156-169

Purpose:

Retrieve relevant past experiences

Parameters: – state: State – action: Action – k: K

Returns: <ast.Subscript object at 0x000001A06D6601D0>

predict_next_state(self, current_state, action)

Lines 171-246

Purpose:

Predict next state given current state and action

Parameters: – current_state: Current state – action: Action

Returns: Prediction

_predict_reward(self, state, action, outcome)

Lines 248-281

Purpose:

Predict reward for state-action-outcome

Parameters: – state: State – action: Action – outcome: Outcome

Returns: float

simulate_trajectory(self, initial_state, action_sequence, max_steps)

Lines 283-317

Purpose:

Simulate a trajectory through the world

Parameters: – initial_state: Initial state – action_sequence: Action sequence – max_steps: Max steps

Returns: SimulationResult

plan_with_model(self, current_state, goal, num_alternatives)

Lines 319-390

Purpose:

Use model for planning

Parameters: – current_state: Current state – goal: Goal – num_alternatives: Num alternatives

Returns: <ast.Subscript object at 0x000001A06D625C10>

---

Functions

demonstrate_world_model()

Lines 393-694

Purpose:

Demonstrate World Model Learning pattern

---

Key Concepts Summary

Pattern Overview

The World Model pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 061
• Pattern 063

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The World Model pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 062_world_model.py
Total Lines: 699
Pattern Category: AI Agent Design Pattern #062

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 061: Diffusion-Based Planning

Description:
    Diffusion-Based Planning uses diffusion models (inspired by diffusion processes
    in physics) for trajectory planning and action sequence generation. This approach
    can handle multimodal distributions and generate diverse, high-quality plans by
    iteratively denoising random noise into coherent action sequences.

Components:
    1. Diffusion Process: Forward (add noise) and reverse (denoise)
    2. Action Sequence Generator: Creates trajectory proposals
    3. Denoising Steps: Iterative refinement of plans
    4. Scoring Function: Evaluates plan quality
    5. Sampling Strategy: Controls exploration-exploitation

Use Cases:
    - Robotics motion planning
    - Game AI pathfinding
    - Multi-step action planning
    - Creative sequence generation
    - Trajectory optimization
    - Scenario planning

LangChain Implementation:
    Implements diffusion-inspired planning using LLM-based iterative refinement,
    noise injection, and progressive denoising for plan generation.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
import math

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class NoiseLevel(Enum):
    """Noise levels for diffusion process"""
    HIGH = "high"  # Initial high noise
    MEDIUM = "medium"  # Mid-process
    LOW = "low"  # Near convergence
    MINIMAL = "minimal"  # Final refinement


class SamplingStrategy(Enum):
    """Strategies for plan sampling"""
    DETERMINISTIC = "deterministic"  # Single best plan
    STOCHASTIC = "stochastic"  # Sample from distribution
    BEAM_SEARCH = "beam_search"  # Keep top-k candidates
    DIVERSE = "diverse"  # Maximize diversity


@dataclass
class ActionStep:
    """Single action in a sequence"""
    step_id: int
    action: str
    parameters: Dict[str, Any] = field(default_factory=dict)
    confidence: float = 1.0
    
    def add_noise(self, noise_level: float) -> 'ActionStep':
        """Add noise to action for diffusion"""
        # In real implementation, this would modify action parameters
        # Here we adjust confidence as proxy for noise
        noisy_confidence = max(0.1, self.confidence * (1 - noise_level * 0.5))
        return ActionStep(
            step_id=self.step_id,
            action=self.action,
            parameters=self.parameters.copy(),
            confidence=noisy_confidence
        )


@dataclass
class Plan:
    """Complete action plan"""
    plan_id: str
    goal: str
    steps: List[ActionStep]
    noise_level: float = 0.0
    quality_score: float = 0.0
    diversity_score: float = 0.0
    
    @property
    def num_steps(self) -> int:
        return len(self.steps)
    
    def to_text(self) -> str:
        """Convert plan to text representation"""
        text = f"Goal: {self.goal}\n\nSteps:\n"
        for step in self.steps:
            text += f"{step.step_id}. {step.action}\n"
        return text


@dataclass
class DiffusionResult:
    """Result from diffusion planning"""
    goal: str
    final_plans: List[Plan]
    diffusion_steps: int
    best_plan: Plan
    diversity_score: float
    total_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "goal": self.goal[:80] + "...",
            "num_plans": len(self.final_plans),
            "best_plan_steps": self.best_plan.num_steps,
            "best_quality": f"{self.best_plan.quality_score:.2f}",
            "diversity": f"{self.diversity_score:.2f}",
            "diffusion_steps": self.diffusion_steps,
            "total_time_ms": f"{self.total_time_ms:.1f}"
        }


class DiffusionPlanner:
    """
    Diffusion-based planning agent.
    
    Features:
    1. Iterative plan denoising
    2. Multi-modal plan generation
    3. Quality-guided refinement
    4. Diverse plan sampling
    5. Progressive improvement
    """
    
    def __init__(
        self,
        num_diffusion_steps: int = 5,
        num_candidates: int = 3,
        sampling_strategy: SamplingStrategy = SamplingStrategy.DIVERSE,
        temperature: float = 0.7
    ):
        self.num_diffusion_steps = num_diffusion_steps
        self.num_candidates = num_candidates
        self.sampling_strategy = sampling_strategy
        self.temperature = temperature
        
        # Planner LLM (higher temperature for creativity)
        self.planner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Refiner LLM (lower temperature for quality)
        self.refiner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # Evaluator LLM
        self.evaluator = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
    
    def _generate_noisy_plan(
        self,
        goal: str,
        noise_level: float,
        context: Optional[str] = None
    ) -> Plan:
        """Generate initial noisy plan"""
        
        # Higher noise = more randomness = more creative/diverse
        creativity_instruction = ""
        if noise_level > 0.7:
            creativity_instruction = "Be highly creative and explore unusual approaches."
        elif noise_level > 0.4:
            creativity_instruction = "Consider multiple different approaches."
        else:
            creativity_instruction = "Focus on practical, direct solutions."
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are generating an action plan to achieve a goal.

{creativity_instruction}

{context if context else ""}

Generate a step-by-step plan. Format:
1. [action step]
2. [action step]
..."""),
            ("user", "Goal: {goal}\n\nPlan:")
        ])
        
        chain = prompt | self.planner | StrOutputParser()
        plan_text = chain.invoke({"goal": goal})
        
        # Parse steps
        steps = []
        for i, line in enumerate(plan_text.split('\n'), 1):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                # Remove numbering
                action = line.lstrip('0123456789.-) ').strip()
                if action:
                    steps.append(ActionStep(
                        step_id=i,
                        action=action,
                        confidence=1.0 - noise_level * 0.3
                    ))
        
        return Plan(
            plan_id=f"plan_{random.randint(1000, 9999)}",
            goal=goal,
            steps=steps,
            noise_level=noise_level
        )
    
    def _denoise_plan(
        self,
        noisy_plan: Plan,
        target_noise_level: float,
        iteration: int
    ) -> Plan:
        """Denoise/refine a plan"""
        
        # Calculate denoising strength
        denoise_strength = noisy_plan.noise_level - target_noise_level
        
        if denoise_strength  float:
        """Evaluate plan quality"""
        
        evaluation_prompt = ChatPromptTemplate.from_messages([
            ("system", """Evaluate this action plan on a scale of 0.0 to 1.0.

Consider:
- Completeness: Does it achieve the goal?
- Feasibility: Is it realistic?
- Efficiency: Minimal unnecessary steps?
- Clarity: Clear and actionable?

Respond with just a number between 0.0 and 1.0"""),
            ("user", "{plan}")
        ])
        
        chain = evaluation_prompt | self.evaluator | StrOutputParser()
        
        try:
            score_str = chain.invoke({"plan": plan.to_text()})
            score = float(''.join(c for c in score_str if c.isdigit() or c == '.'))
            return max(0.0, min(1.0, score))
        except:
            # Fallback: score based on step count and confidence
            avg_confidence = sum(s.confidence for s in plan.steps) / len(plan.steps) if plan.steps else 0
            step_score = min(1.0, len(plan.steps) / 5.0)  # Prefer 5-ish steps
            return (avg_confidence + step_score) / 2
    
    def _calculate_diversity(self, plans: List[Plan]) -> float:
        """Calculate diversity among plans"""
        
        if len(plans)  0 else 0.0
    
    def plan(
        self,
        goal: str,
        constraints: Optional[List[str]] = None
    ) -> DiffusionResult:
        """Generate plans using diffusion process"""
        
        start_time = time.time()
        
        print(f"\n🌊 Starting diffusion-based planning...")
        print(f"   Goal: {goal}")
        print(f"   Diffusion steps: {self.num_diffusion_steps}")
        print(f"   Candidates: {self.num_candidates}")
        
        # Build context from constraints
        context = ""
        if constraints:
            context = "Constraints:\n" + "\n".join(f"- {c}" for c in constraints)
        
        # Generate initial noisy plans
        print(f"\n📍 Step 0: Generating {self.num_candidates} initial noisy plans...")
        
        current_plans = []
        for i in range(self.num_candidates):
            # Start with high noise for diversity
            initial_noise = 0.8 + random.random() * 0.2  # 0.8-1.0
            plan = self._generate_noisy_plan(goal, initial_noise, context)
            current_plans.append(plan)
            print(f"   Plan {i+1}: {len(plan.steps)} steps (noise: {initial_noise:.2f})")
        
        # Diffusion process: progressively denoise
        for step in range(1, self.num_diffusion_steps + 1):
            # Calculate target noise level (decreasing)
            target_noise = max(0.0, 1.0 - (step / self.num_diffusion_steps))
            
            print(f"\n📍 Step {step}/{self.num_diffusion_steps}: Denoising to {target_noise:.2f}")
            
            # Denoise all plans
            denoised_plans = []
            for plan in current_plans:
                denoised = self._denoise_plan(plan, target_noise, step)
                denoised_plans.append(denoised)
            
            # Evaluate plans
            for plan in denoised_plans:
                plan.quality_score = self._evaluate_plan(plan)
            
            # Select best plans based on strategy
            if self.sampling_strategy == SamplingStrategy.BEAM_SEARCH:
                # Keep top-k by quality
                denoised_plans.sort(key=lambda p: p.quality_score, reverse=True)
                current_plans = denoised_plans[:self.num_candidates]
                
            elif self.sampling_strategy == SamplingStrategy.DIVERSE:
                # Balance quality and diversity
                # Keep best plan + diverse alternatives
                denoised_plans.sort(key=lambda p: p.quality_score, reverse=True)
                current_plans = [denoised_plans[0]]  # Best plan
                
                # Add diverse plans
                for plan in denoised_plans[1:]:
                    if len(current_plans) >= self.num_candidates:
                        break
                    # Check if sufficiently different
                    is_diverse = True
                    for existing in current_plans:
                        # Simple diversity check
                        common_actions = sum(
                            1 for s1 in plan.steps
                            for s2 in existing.steps
                            if s1.action[:30] == s2.action[:30]
                        )
                        if common_actions > len(plan.steps) * 0.7:
                            is_diverse = False
                            break
                    
                    if is_diverse:
                        current_plans.append(plan)
            
            else:  # DETERMINISTIC or STOCHASTIC
                denoised_plans.sort(key=lambda p: p.quality_score, reverse=True)
                current_plans = denoised_plans[:self.num_candidates]
            
            # Show progress
            for i, plan in enumerate(current_plans, 1):
                print(f"   Plan {i}: Quality {plan.quality_score:.2f}, {len(plan.steps)} steps")
        
        # Final evaluation
        print(f"\n✅ Diffusion complete. Evaluating final plans...")
        
        for plan in current_plans:
            plan.quality_score = self._evaluate_plan(plan)
        
        # Calculate diversity
        diversity_score = self._calculate_diversity(current_plans)
        
        # Select best plan
        best_plan = max(current_plans, key=lambda p: p.quality_score)
        
        total_time_ms = (time.time() - start_time) * 1000
        
        return DiffusionResult(
            goal=goal,
            final_plans=current_plans,
            diffusion_steps=self.num_diffusion_steps,
            best_plan=best_plan,
            diversity_score=diversity_score,
            total_time_ms=total_time_ms
        )


def demonstrate_diffusion_planning():
    """Demonstrate Diffusion-Based Planning pattern"""
    
    print("=" * 80)
    print("PATTERN 061: DIFFUSION-BASED PLANNING DEMONSTRATION")
    print("=" * 80)
    print("\nIterative plan refinement through diffusion process\n")
    
    # Test 1: Simple goal with diverse sampling
    print("\n" + "=" * 80)
    print("TEST 1: Diverse Plan Generation")
    print("=" * 80)
    
    planner1 = DiffusionPlanner(
        num_diffusion_steps=4,
        num_candidates=3,
        sampling_strategy=SamplingStrategy.DIVERSE,
        temperature=0.8
    )
    
    goal1 = "Plan a successful product launch"
    
    result1 = planner1.plan(goal1)
    
    print(f"\n📊 Results:")
    print(f"   Plans Generated: {len(result1.final_plans)}")
    print(f"   Diversity Score: {result1.diversity_score:.2f}")
    print(f"   Total Time: {result1.total_time_ms:.1f}ms")
    
    print(f"\n🏆 Best Plan (Quality: {result1.best_plan.quality_score:.2f}):")
    for step in result1.best_plan.steps:
        print(f"   {step.step_id}. {step.action}")
    
    print(f"\n🔄 Alternative Plans:")
    for i, plan in enumerate(result1.final_plans[1:], 2):
        print(f"\n   Plan {i} (Quality: {plan.quality_score:.2f}):")
        for step in plan.steps[:3]:
            print(f"      {step.step_id}. {step.action}")
        if len(plan.steps) > 3:
            print(f"      ... ({len(plan.steps) - 3} more steps)")
    
    # Test 2: Constrained planning
    print("\n" + "=" * 80)
    print("TEST 2: Constrained Diffusion Planning")
    print("=" * 80)
    
    planner2 = DiffusionPlanner(
        num_diffusion_steps=5,
        num_candidates=2,
        sampling_strategy=SamplingStrategy.BEAM_SEARCH
    )
    
    goal2 = "Organize a team-building event"
    constraints2 = [
        "Budget limited to $1000",
        "Must be completed within 2 weeks",
        "Remote team across different time zones"
    ]
    
    print(f"\n🎯 Goal: {goal2}")
    print(f"📋 Constraints:")
    for constraint in constraints2:
        print(f"   - {constraint}")
    
    result2 = planner2.plan(goal2, constraints=constraints2)
    
    print(f"\n🏆 Optimized Plan (Quality: {result2.best_plan.quality_score:.2f}):")
    for step in result2.best_plan.steps:
        print(f"   {step.step_id}. {step.action}")
    
    # Test 3: Comparing diffusion steps
    print("\n" + "=" * 80)
    print("TEST 3: Impact of Diffusion Steps")
    print("=" * 80)
    
    goal3 = "Improve customer satisfaction"
    
    for num_steps in [2, 4, 6]:
        print(f"\n   Testing with {num_steps} diffusion steps...")
        
        planner = DiffusionPlanner(
            num_diffusion_steps=num_steps,
            num_candidates=2,
            sampling_strategy=SamplingStrategy.DIVERSE
        )
        
        result = planner.plan(goal3)
        
        print(f"      Best Quality: {result.best_plan.quality_score:.2f}")
        print(f"      Diversity: {result.diversity_score:.2f}")
        print(f"      Time: {result.total_time_ms:.1f}ms")
        print(f"      Steps in plan: {result.best_plan.num_steps}")
    
    # Summary
    print("\n" + "=" * 80)
    print("DIFFUSION-BASED PLANNING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Multi-Modal Plans: Generates diverse alternative plans
2. Progressive Refinement: Iterative quality improvement
3. Exploration-Exploitation: Balances creativity and quality
4. Robustness: Multiple candidate solutions
5. Flexible Control: Adjustable diffusion process

Diffusion Process:
1. Forward Process: Add noise (generate diverse initial plans)
2. Reverse Process: Denoise (iteratively refine plans)
3. Multiple Steps: Progressive quality improvement
4. Sampling: Control exploration vs exploitation

Components:
- Noise Injection: Creates initial diversity
- Denoising: Iterative refinement
- Quality Scoring: Evaluates plan effectiveness
- Diversity Measurement: Ensures variety
- Selection Strategy: Chooses best candidates

Sampling Strategies:
1. Deterministic: Single best path
   - Highest quality guarantee
   - No diversity

2. Stochastic: Sample from distribution
   - Controlled randomness
   - Balanced exploration

3. Beam Search: Keep top-k candidates
   - Quality-focused
   - Limited diversity

4. Diverse: Maximize variety
   - Multiple alternatives
   - Creative solutions

Noise Schedules:
- High Noise (1.0 → 0.7): Initial creativity
- Medium Noise (0.7 → 0.4): Refinement begins
- Low Noise (0.4 → 0.1): Convergence
- Minimal Noise (0.1 → 0.0): Final polish

Use Cases:
- Robotics: Motion planning, trajectory optimization
- Game AI: Pathfinding, strategy generation
- Planning: Project planning, resource allocation
- Creative Tasks: Story generation, design
- Optimization: Multi-objective planning
- Scenario Analysis: Alternative futures

Best Practices:
1. More steps = higher quality (but slower)
2. More candidates = more diversity
3. Appropriate noise schedule
4. Quality-diversity balance
5. Constraint incorporation
6. Iterative evaluation
7. Early stopping when converged

Production Considerations:
- Computational cost (multiple LLM calls)
- Caching intermediate results
- Parallel plan generation
- Quality thresholds
- Timeout handling
- Plan validation
- User preference learning

Comparison with Related Patterns:
- vs. Tree-of-Thoughts: Continuous vs discrete
- vs. Beam Search: Diffusion vs greedy
- vs. Monte Carlo: Systematic vs random
- vs. Evolutionary: Denoising vs mutation

Advanced Techniques:
1. Conditional Diffusion: Constraint-aware
2. Guided Diffusion: Quality-guided sampling
3. Classifier-Free: Balance guidance
4. Latent Diffusion: Work in abstract space
5. Cascaded: Multi-resolution planning

Research Inspiration:
- Denoising Diffusion Probabilistic Models (DDPM)
- Score-based Generative Models
- Diffusion for Planning (Diffuser)
- Planning with Diffusion for Flexible Behavior

The Diffusion-Based Planning pattern provides a powerful framework
for generating diverse, high-quality plans through iterative
refinement inspired by diffusion processes in physics.
""")


if __name__ == "__main__":
    demonstrate_diffusion_planning()

    

Agentic Design Pattern: Diffusion Planning

Pattern 061: Diffusion Planning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Diffusion Planning pattern implementation.

---

Pattern Description

Pattern 061: Diffusion-Based Planning

Description: Diffusion-Based Planning uses diffusion models (inspired by diffusion processes in physics) for trajectory planning and action sequence generation. This approach can handle multimodal distributions and generate diverse, high-quality plans by iteratively denoising random noise into coherent action sequences.

Components: 1. Diffusion Process: Forward (add noise) and reverse (denoise) 2. Action Sequence Generator: Creates trajectory proposals 3. Denoising Steps: Iterative refinement of plans 4. Scoring Function: Evaluates plan quality 5. Sampling Strategy: Controls exploration-exploitation

Use Cases: – Robotics motion planning – Game AI pathfinding – Multi-step action planning – Creative sequence generation – Trajectory optimization – Scenario planning

LangChain Implementation: Implements diffusion-inspired planning using LLM-based iterative refinement, noise injection, and progressive denoising for plan generation.

---

File Statistics

• Total Lines: 637
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 061: Diffusion-Based Planning

Description:
    Diffusion-Based Planning uses diffusion models (inspired by diffusion processes
    in physics) for trajectory planning and action sequence generation. This approach
    can handle multimodal distributions and generate diverse, high-quality plans by
    iteratively denoising random noise into coherent action sequences.

Components:
    1. Diffusion Process: Forward (add noise) and reverse (denoise)
    2. Action Sequence Generator: Creates trajectory proposals
    3. Denoising Steps: Iterative refinement of plans
    4. Scoring Function: Evaluates plan quality
    5. Sampling Strategy: Controls exploration-exploitation

Use Cases:
    - Robotics motion planning
    - Game AI pathfinding
    - Multi-step action planning
    - Creative sequence generation
    - Trajectory optimization
    - Scenario planning

LangChain Implementation:
    Implements diffusion-inspired planning using LLM-based iterative refinement,
    noise injection, and progressive denoising for plan generation.
"""

Purpose: This module implements the Diffusion Planning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – random: Random number generation – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from math

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: NoiseLevel

Lines 47-52

Inherits from: Enum

Purpose:

Noise levels for diffusion process

---

Class: SamplingStrategy

Lines 55-60

Inherits from: Enum

Purpose:

Strategies for plan sampling

---

Class: ActionStep

Lines 64-81

Purpose:

Single action in a sequence

Methods:

add_noise(self, noise_level)

Lines 71-81

Purpose:

Add noise to action for diffusion

Parameters: – noise_level: Noise level

Returns: ActionStep

---

Class: Plan

Lines 85-103

Purpose:

Complete action plan

Methods:

num_steps(self)

Lines 95-96

Decorators: property

Purpose: Num Steps operation.

Parameters:

Returns: int

to_text(self)

Lines 98-103

Purpose:

Convert plan to text representation

Parameters:

Returns: str

---

Class: DiffusionResult

Lines 107-125

Purpose:

Result from diffusion planning

Methods:

to_dict(self)

Lines 116-125

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D79BDD0>

---

Class: DiffusionPlanner

Lines 128-440

Purpose:

Diffusion-based planning agent.

Features:
1. Iterative plan denoising
2. Multi-modal plan generation
3. Quality-guided refinement
4. Diverse plan sampling
5. Progressive improvement

Methods:

__init__(self, num_diffusion_steps, num_candidates, sampling_strategy, temperature)

Lines 140-168

Purpose: Initialize the class instance with required parameters.

Parameters: – num_diffusion_steps: Num diffusion steps – num_candidates: Num candidates – sampling_strategy: Sampling strategy – temperature: Sampling temperature (0-1)

_generate_noisy_plan(self, goal, noise_level, context)

Lines 170-223

Purpose:

Generate initial noisy plan

Parameters: – goal: Goal – noise_level: Noise level – context: Context

Returns: Plan

_denoise_plan(self, noisy_plan, target_noise_level, iteration)

Lines 225-286

Purpose:

Denoise/refine a plan

Parameters: – noisy_plan: Noisy plan – target_noise_level: Target noise level – iteration: Iteration

Returns: Plan

_evaluate_plan(self, plan)

Lines 288-314

Purpose:

Evaluate plan quality

Parameters: – plan: Plan

Returns: float

_calculate_diversity(self, plans)

Lines 316-330

Purpose:

Calculate diversity among plans

Parameters: – plans: Plans

Returns: float

plan(self, goal, constraints)

Lines 332-440

Purpose:

Generate plans using diffusion process

Parameters: – goal: Goal – constraints: Constraints

Returns: DiffusionResult

---

Functions

demonstrate_diffusion_planning()

Lines 443-632

Purpose:

Demonstrate Diffusion-Based Planning pattern

---

Key Concepts Summary

Pattern Overview

The Diffusion Planning pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 060
• Pattern 062

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Diffusion Planning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 061_diffusion_planning.py
Total Lines: 637
Pattern Category: AI Agent Design Pattern #061

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 060: Curriculum Learning

Description:
    Curriculum Learning structures the learning process by presenting training
    examples in a meaningful order, from simple to complex. Like human education,
    this progressive approach helps agents build foundational skills before
    tackling harder problems, leading to faster learning and better performance.

Components:
    1. Difficulty Scorer: Evaluates task complexity
    2. Curriculum Designer: Orders tasks by difficulty
    3. Pacing Strategy: Determines progression speed
    4. Performance Monitor: Tracks learning progress
    5. Adaptive Scheduler: Adjusts curriculum based on performance

Use Cases:
    - Training complex reasoning agents
    - Gradual skill acquisition
    - Hierarchical task learning
    - Educational tutoring systems
    - Reinforcement learning with sparse rewards
    - Knowledge scaffolding

LangChain Implementation:
    Implements curriculum learning by ordering prompts and examples from
    simple to complex, adapting progression based on agent performance.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import statistics
import random

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class DifficultyLevel(Enum):
    """Difficulty levels for curriculum"""
    TRIVIAL = 1
    EASY = 2
    MEDIUM = 3
    HARD = 4
    EXPERT = 5


class PacingStrategy(Enum):
    """Strategies for curriculum pacing"""
    FIXED = "fixed"  # Predetermined progression
    ADAPTIVE = "adaptive"  # Based on performance
    MIXED = "mixed"  # Combine easy and hard
    SPIRAL = "spiral"  # Revisit with increasing complexity
    COMPETENCY_BASED = "competency_based"  # Advance when mastered


@dataclass
class LearningExample:
    """A single learning example"""
    example_id: str
    task_type: str
    input_text: str
    expected_output: str
    difficulty: DifficultyLevel
    prerequisites: List[str] = field(default_factory=list)
    concepts: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def is_prerequisite_met(self, mastered_examples: Set[str]) -> bool:
        """Check if prerequisites are satisfied"""
        return all(prereq in mastered_examples for prereq in self.prerequisites)


@dataclass
class CurriculumStage:
    """A stage in the curriculum"""
    stage_id: int
    stage_name: str
    difficulty_range: Tuple[DifficultyLevel, DifficultyLevel]
    examples: List[LearningExample]
    mastery_threshold: float = 0.8
    
    @property
    def num_examples(self) -> int:
        return len(self.examples)


@dataclass
class PerformanceRecord:
    """Record of performance on an example"""
    example_id: str
    attempt_number: int
    correct: bool
    response: str
    timestamp: datetime
    time_taken_ms: float


@dataclass
class CurriculumResult:
    """Result from curriculum learning"""
    stages_completed: int
    total_examples: int
    examples_mastered: int
    overall_accuracy: float
    learning_curve: List[Tuple[int, float]]  # (example_num, accuracy)
    time_per_stage: List[float]
    final_performance: float
    total_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "stages_completed": self.stages_completed,
            "examples_mastered": f"{self.examples_mastered}/{self.total_examples}",
            "overall_accuracy": f"{self.overall_accuracy:.1%}",
            "final_performance": f"{self.final_performance:.1%}",
            "total_time_ms": f"{self.total_time_ms:.1f}"
        }


class CurriculumLearner:
    """
    Agent that learns through curriculum progression.
    
    Features:
    1. Difficulty-ordered example presentation
    2. Performance-based pacing
    3. Mastery tracking
    4. Adaptive progression
    5. Skill scaffolding
    """
    
    def __init__(
        self,
        pacing_strategy: PacingStrategy = PacingStrategy.ADAPTIVE,
        mastery_threshold: float = 0.8,
        temperature: float = 0.7
    ):
        self.pacing_strategy = pacing_strategy
        self.mastery_threshold = mastery_threshold
        self.temperature = temperature
        
        # Learner LLM
        self.learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Difficulty assessor
        self.assessor = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
        
        # Learning state
        self.mastered_examples: Set[str] = set()
        self.performance_history: List[PerformanceRecord] = []
        self.current_stage: int = 0
        
        # Learning statistics
        self.learning_curve: List[Tuple[int, float]] = []
    
    def _assess_difficulty(self, example: LearningExample) -> DifficultyLevel:
        """Assess difficulty of an example"""
        
        # If already labeled, use that
        if example.difficulty != DifficultyLevel.MEDIUM:
            return example.difficulty
        
        # Otherwise, assess automatically
        assessment_prompt = ChatPromptTemplate.from_messages([
            ("system", """Rate the difficulty of this learning task on a scale:
1 - Trivial (very simple, obvious answer)
2 - Easy (straightforward, basic reasoning)
3 - Medium (requires some thought)
4 - Hard (complex, multiple steps)
5 - Expert (very challenging, advanced concepts)

Respond with just the number."""),
            ("user", "Task: {input}\nExpected: {output}\n\nDifficulty:")
        ])
        
        chain = assessment_prompt | self.assessor | StrOutputParser()
        
        try:
            rating = chain.invoke({
                "input": example.input_text,
                "output": example.expected_output
            })
            
            difficulty_num = int(''.join(c for c in rating if c.isdigit()))
            return DifficultyLevel(min(max(difficulty_num, 1), 5))
        except:
            return DifficultyLevel.MEDIUM
    
    def _create_curriculum(
        self,
        examples: List[LearningExample]
    ) -> List[CurriculumStage]:
        """Create curriculum stages from examples"""
        
        # Assess difficulties if needed
        for example in examples:
            if example.difficulty == DifficultyLevel.MEDIUM and not example.metadata.get("assessed"):
                example.difficulty = self._assess_difficulty(example)
                example.metadata["assessed"] = True
        
        # Group by difficulty
        stages_dict: Dict[DifficultyLevel, List[LearningExample]] = {}
        for example in examples:
            if example.difficulty not in stages_dict:
                stages_dict[example.difficulty] = []
            stages_dict[example.difficulty].append(example)
        
        # Create stages
        stages = []
        for difficulty in sorted(stages_dict.keys(), key=lambda x: x.value):
            stage = CurriculumStage(
                stage_id=len(stages),
                stage_name=f"Stage {len(stages)+1}: {difficulty.name}",
                difficulty_range=(difficulty, difficulty),
                examples=stages_dict[difficulty],
                mastery_threshold=self.mastery_threshold
            )
            stages.append(stage)
        
        return stages
    
    def _attempt_example(
        self,
        example: LearningExample,
        previous_examples: List[LearningExample]
    ) -> PerformanceRecord:
        """Attempt to solve an example"""
        
        start_time = time.time()
        
        # Build context from previous examples
        context = ""
        if previous_examples:
            context = "Previous examples:\n" + "\n".join([
                f"Input: {ex.input_text}\nOutput: {ex.expected_output}"
                for ex in previous_examples[-3:]  # Last 3 examples
            ]) + "\n\n"
        
        # Generate response
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are learning to solve tasks step by step.

{context}Now solve this task:"""),
            ("user", "{input}")
        ])
        
        chain = prompt | self.learner | StrOutputParser()
        response = chain.invoke({
            "context": context,
            "input": example.input_text
        })
        
        time_taken_ms = (time.time() - start_time) * 1000
        
        # Check correctness (simplified)
        response_normalized = response.strip().lower()
        expected_normalized = example.expected_output.strip().lower()
        
        # Flexible matching
        correct = (
            response_normalized == expected_normalized or
            expected_normalized in response_normalized or
            response_normalized in expected_normalized
        )
        
        # Count attempts
        attempt_number = len([
            r for r in self.performance_history
            if r.example_id == example.example_id
        ]) + 1
        
        record = PerformanceRecord(
            example_id=example.example_id,
            attempt_number=attempt_number,
            correct=correct,
            response=response,
            timestamp=datetime.now(),
            time_taken_ms=time_taken_ms
        )
        
        self.performance_history.append(record)
        
        return record
    
    def _calculate_stage_performance(
        self,
        stage: CurriculumStage
    ) -> float:
        """Calculate performance on current stage"""
        
        stage_example_ids = {ex.example_id for ex in stage.examples}
        
        recent_records = [
            r for r in self.performance_history[-20:]  # Last 20 attempts
            if r.example_id in stage_example_ids
        ]
        
        if not recent_records:
            return 0.0
        
        correct_count = sum(1 for r in recent_records if r.correct)
        return correct_count / len(recent_records)
    
    def _should_advance(
        self,
        stage: CurriculumStage,
        attempts_on_stage: int
    ) -> bool:
        """Determine if should advance to next stage"""
        
        if self.pacing_strategy == PacingStrategy.FIXED:
            # Fixed progression after N attempts
            return attempts_on_stage >= len(stage.examples) * 2
        
        elif self.pacing_strategy == PacingStrategy.ADAPTIVE:
            # Advance when performance exceeds threshold
            performance = self._calculate_stage_performance(stage)
            return performance >= stage.mastery_threshold and attempts_on_stage >= len(stage.examples)
        
        elif self.pacing_strategy == PacingStrategy.COMPETENCY_BASED:
            # Advance when all examples mastered
            stage_example_ids = {ex.example_id for ex in stage.examples}
            mastered_in_stage = stage_example_ids & self.mastered_examples
            return len(mastered_in_stage) >= len(stage_example_ids) * stage.mastery_threshold
        
        else:
            # Default: adaptive
            performance = self._calculate_stage_performance(stage)
            return performance >= stage.mastery_threshold
    
    def learn_curriculum(
        self,
        examples: List[LearningExample],
        max_attempts_per_stage: int = 20
    ) -> CurriculumResult:
        """Learn through curriculum progression"""
        
        start_time = time.time()
        
        print(f"\n📚 Starting curriculum learning with {len(examples)} examples...")
        
        # Create curriculum
        stages = self._create_curriculum(examples)
        
        print(f"\n🎯 Curriculum Structure:")
        for stage in stages:
            print(f"   {stage.stage_name}: {stage.num_examples} examples")
        
        # Learn through stages
        time_per_stage = []
        
        for stage_idx, stage in enumerate(stages):
            stage_start = time.time()
            
            print(f"\n📖 Stage {stage_idx + 1}/{len(stages)}: {stage.stage_name}")
            print(f"   Examples: {stage.num_examples}")
            print(f"   Mastery Threshold: {stage.mastery_threshold:.0%}")
            
            attempts_on_stage = 0
            previous_examples = []
            
            # Learn examples in this stage
            while attempts_on_stage  float:
        """Calculate accuracy over recent attempts"""
        
        if not self.performance_history:
            return 0.0
        
        recent = self.performance_history[-window:]
        correct = sum(1 for r in recent if r.correct)
        return correct / len(recent)


def demonstrate_curriculum_learning():
    """Demonstrate Curriculum Learning pattern"""
    
    print("=" * 80)
    print("PATTERN 060: CURRICULUM LEARNING DEMONSTRATION")
    print("=" * 80)
    print("\nProgressive learning from simple to complex\n")
    
    # Create learning examples with varying difficulty
    examples = [
        # Trivial examples
        LearningExample("ex1", "math", "What is 2 + 2?", "4", DifficultyLevel.TRIVIAL),
        LearningExample("ex2", "math", "What is 5 + 3?", "8", DifficultyLevel.TRIVIAL),
        LearningExample("ex3", "math", "What is 10 - 6?", "4", DifficultyLevel.TRIVIAL),
        
        # Easy examples
        LearningExample("ex4", "math", "What is 12 + 15?", "27", DifficultyLevel.EASY, prerequisites=["ex1"]),
        LearningExample("ex5", "math", "What is 25 - 8?", "17", DifficultyLevel.EASY, prerequisites=["ex3"]),
        LearningExample("ex6", "math", "What is 7 times 3?", "21", DifficultyLevel.EASY),
        
        # Medium examples
        LearningExample("ex7", "math", "What is 15 times 4?", "60", DifficultyLevel.MEDIUM, prerequisites=["ex6"]),
        LearningExample("ex8", "math", "What is 100 divided by 4?", "25", DifficultyLevel.MEDIUM),
        LearningExample("ex9", "math", "What is (5 + 3) times 2?", "16", DifficultyLevel.MEDIUM, prerequisites=["ex4", "ex6"]),
        
        # Hard examples
        LearningExample("ex10", "math", "What is 25% of 80?", "20", DifficultyLevel.HARD, prerequisites=["ex8"]),
        LearningExample("ex11", "math", "If x + 5 = 12, what is x?", "7", DifficultyLevel.HARD, prerequisites=["ex4"]),
    ]
    
    # Test 1: Adaptive curriculum
    print("\n" + "=" * 80)
    print("TEST 1: Adaptive Curriculum Learning")
    print("=" * 80)
    
    learner1 = CurriculumLearner(
        pacing_strategy=PacingStrategy.ADAPTIVE,
        mastery_threshold=0.7
    )
    
    result1 = learner1.learn_curriculum(examples, max_attempts_per_stage=15)
    
    print(f"\n📊 Learning Results:")
    print(f"   Stages Completed: {result1.stages_completed}")
    print(f"   Examples Mastered: {result1.examples_mastered}/{result1.total_examples}")
    print(f"   Overall Accuracy: {result1.overall_accuracy:.1%}")
    print(f"   Final Performance: {result1.final_performance:.1%}")
    print(f"   Total Time: {result1.total_time_ms/1000:.1f}s")
    
    print(f"\n📈 Learning Curve (samples):")
    for i in range(0, len(result1.learning_curve), max(1, len(result1.learning_curve)//5)):
        attempt_num, accuracy = result1.learning_curve[i]
        print(f"   Attempt {attempt_num}: {accuracy:.1%} accuracy")
    
    print(f"\n⏱️  Time per Stage:")
    for i, stage_time in enumerate(result1.time_per_stage, 1):
        print(f"   Stage {i}: {stage_time:.1f}s")
    
    # Test 2: Fixed pacing
    print("\n" + "=" * 80)
    print("TEST 2: Fixed Pacing Curriculum")
    print("=" * 80)
    
    learner2 = CurriculumLearner(
        pacing_strategy=PacingStrategy.FIXED,
        mastery_threshold=0.6
    )
    
    # Use subset for faster demo
    subset_examples = examples[:8]
    
    result2 = learner2.learn_curriculum(subset_examples, max_attempts_per_stage=10)
    
    print(f"\n📊 Fixed Pacing Results:")
    print(f"   Stages: {result2.stages_completed}")
    print(f"   Mastered: {result2.examples_mastered}/{result2.total_examples}")
    print(f"   Accuracy: {result2.overall_accuracy:.1%}")
    
    # Test 3: Learning curve comparison
    print("\n" + "=" * 80)
    print("TEST 3: Learning Progress Visualization")
    print("=" * 80)
    
    print(f"\n📈 Performance Improvement:")
    curve = result1.learning_curve
    
    # Show improvement
    if len(curve) >= 3:
        early_perf = statistics.mean([acc for _, acc in curve[:3]])
        mid_perf = statistics.mean([acc for _, acc in curve[len(curve)//2:len(curve)//2+3]])
        late_perf = statistics.mean([acc for _, acc in curve[-3:]])
        
        print(f"   Early (first 3): {early_perf:.1%}")
        print(f"   Middle: {mid_perf:.1%}")
        print(f"   Late (last 3): {late_perf:.1%}")
        print(f"   Improvement: +{(late_perf - early_perf):.1%}")
    
    # Summary
    print("\n" + "=" * 80)
    print("CURRICULUM LEARNING PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Faster Learning: Build on foundations progressively
2. Better Performance: Master basics before advanced topics
3. Reduced Frustration: Appropriate difficulty at each stage
4. Stable Training: Avoid overwhelming with complexity
5. Improved Retention: Scaffold knowledge systematically

Curriculum Design:
1. Difficulty Ordering: Simple → Complex
2. Prerequisite Structure: Build dependencies
3. Concept Progression: Foundational → Advanced
4. Skill Scaffolding: Support skill development
5. Spiral Learning: Revisit concepts with depth

Difficulty Assessment:
- Automatic: LLM-based complexity scoring
- Manual: Expert labeling
- Hybrid: Combine both approaches
- Dynamic: Adjust based on learner performance
- Multi-dimensional: Consider multiple factors

Pacing Strategies:
1. Fixed: Predetermined progression
   - Simple, predictable
   - May not fit all learners

2. Adaptive: Performance-based
   - Personalized pace
   - More complex logic

3. Mixed: Interleave difficulties
   - Maintain engagement
   - Prevent boredom

4. Spiral: Revisit with complexity
   - Reinforce learning
   - Deepen understanding

5. Competency-Based: Master before advance
   - Ensure readiness
   - May be slower

Curriculum Stages:
1. Foundation: Basic concepts
2. Development: Core skills
3. Integration: Combine concepts
4. Application: Real problems
5. Mastery: Advanced challenges

Performance Monitoring:
- Accuracy tracking
- Time per example
- Mastery identification
- Learning curve analysis
- Plateau detection

Advancement Criteria:
- Accuracy threshold (e.g., 80%)
- Minimum attempts per stage
- Mastery of N examples
- Time-based progression
- Adaptive combinations

Use Cases:
- Educational tutoring systems
- Agent skill development
- Reinforcement learning
- Language learning
- Technical training
- Complex task acquisition

Best Practices:
1. Clear difficulty progression
2. Prerequisite tracking
3. Performance monitoring
4. Adaptive pacing
5. Regular assessment
6. Sufficient practice
7. Gradual complexity increase

Production Considerations:
- Curriculum versioning
- Learner state persistence
- Performance analytics
- A/B testing curricula
- Difficulty calibration
- Prerequisite validation
- Scalability

Comparison with Related Patterns:
- vs. Random: Structured vs unordered
- vs. Hard Example Mining: Systematic vs selection
- vs. Meta-Learning: Progression vs strategy learning
- vs. Active Learning: Curriculum vs query strategy

Challenges:
1. Difficulty assessment accuracy
2. Optimal ordering
3. Individual differences
4. Curriculum generalization
5. Pacing calibration

Research Findings:
- 30-50% faster learning vs random
- Better final performance
- Improved generalization
- Reduced training instability
- Context-dependent effectiveness

The Curriculum Learning pattern enables efficient learning through
strategic ordering of examples from simple to complex, mirroring
effective human educational approaches.
""")


if __name__ == "__main__":
    demonstrate_curriculum_learning()

    

Agentic Design Pattern: Curriculum Learning

Pattern 060: Curriculum Learning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Curriculum Learning pattern implementation.

---

Pattern Description

Pattern 060: Curriculum Learning

Description: Curriculum Learning structures the learning process by presenting training examples in a meaningful order, from simple to complex. Like human education, this progressive approach helps agents build foundational skills before tackling harder problems, leading to faster learning and better performance.

Components: 1. Difficulty Scorer: Evaluates task complexity 2. Curriculum Designer: Orders tasks by difficulty 3. Pacing Strategy: Determines progression speed 4. Performance Monitor: Tracks learning progress 5. Adaptive Scheduler: Adjusts curriculum based on performance

Use Cases: – Training complex reasoning agents – Gradual skill acquisition – Hierarchical task learning – Educational tutoring systems – Reinforcement learning with sparse rewards – Knowledge scaffolding

LangChain Implementation: Implements curriculum learning by ordering prompts and examples from simple to complex, adapting progression based on agent performance.

---

File Statistics

• Total Lines: 666
• Classes: 7
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 060: Curriculum Learning

Description:
    Curriculum Learning structures the learning process by presenting training
    examples in a meaningful order, from simple to complex. Like human education,
    this progressive approach helps agents build foundational skills before
    tackling harder problems, leading to faster learning and better performance.

Components:
    1. Difficulty Scorer: Evaluates task complexity
    2. Curriculum Designer: Orders tasks by difficulty
    3. Pacing Strategy: Determines progression speed
    4. Performance Monitor: Tracks learning progress
    5. Adaptive Scheduler: Adjusts curriculum based on performance

Use Cases:
    - Training complex reasoning agents
    - Gradual skill acquisition
    - Hierarchical task learning
    - Educational tutoring systems
    - Reinforcement learning with sparse rewards
    - Knowledge scaffolding

LangChain Implementation:
    Implements curriculum learning by ordering prompts and examples from
    simple to complex, adapting progression based on agent performance.
"""

Purpose: This module implements the Curriculum Learning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import random
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – random: Random number generation – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from statistics

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: DifficultyLevel

Lines 47-53

Inherits from: Enum

Purpose:

Difficulty levels for curriculum

---

Class: PacingStrategy

Lines 56-62

Inherits from: Enum

Purpose:

Strategies for curriculum pacing

---

Class: LearningExample

Lines 66-79

Purpose:

A single learning example

Methods:

is_prerequisite_met(self, mastered_examples)

Lines 77-79

Purpose:

Check if prerequisites are satisfied

Parameters: – mastered_examples: Mastered examples

Returns: bool

---

Class: CurriculumStage

Lines 83-93

Purpose:

A stage in the curriculum

Methods:

num_examples(self)

Lines 92-93

Decorators: property

Purpose: Num Examples operation.

Parameters:

Returns: int

---

Class: PerformanceRecord

Lines 97-104

Purpose:

Record of performance on an example

---

Class: CurriculumResult

Lines 108-126

Purpose:

Result from curriculum learning

Methods:

to_dict(self)

Lines 119-126

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D700050>

---

Class: CurriculumLearner

Lines 129-444

Purpose:

Agent that learns through curriculum progression.

Features:
1. Difficulty-ordered example presentation
2. Performance-based pacing
3. Mastery tracking
4. Adaptive progression
5. Skill scaffolding

Methods:

__init__(self, pacing_strategy, mastery_threshold, temperature)

Lines 141-169

Purpose: Initialize the class instance with required parameters.

Parameters: – pacing_strategy: Pacing strategy – mastery_threshold: Mastery threshold – temperature: Sampling temperature (0-1)

_assess_difficulty(self, example)

Lines 171-202

Purpose:

Assess difficulty of an example

Parameters: – example: Example

Returns: DifficultyLevel

_create_curriculum(self, examples)

Lines 204-235

Purpose:

Create curriculum stages from examples

Parameters: – examples: Examples

Returns: <ast.Subscript object at 0x000001A06D1683D0>

_attempt_example(self, example, previous_examples)

Lines 237-298

Purpose:

Attempt to solve an example

Parameters: – example: Example – previous_examples: Previous examples

Returns: PerformanceRecord

_calculate_stage_performance(self, stage)

Lines 300-317

Purpose:

Calculate performance on current stage

Parameters: – stage: Stage

Returns: float

_should_advance(self, stage, attempts_on_stage)

Lines 319-344

Purpose:

Determine if should advance to next stage

Parameters: – stage: Stage – attempts_on_stage: Attempts on stage

Returns: bool

learn_curriculum(self, examples, max_attempts_per_stage)

Lines 346-434

Purpose:

Learn through curriculum progression

Parameters: – examples: Examples – max_attempts_per_stage: Max attempts per stage

Returns: CurriculumResult

_calculate_recent_accuracy(self, window)

Lines 436-444

Purpose:

Calculate accuracy over recent attempts

Parameters: – window: Window

Returns: float

---

Functions

demonstrate_curriculum_learning()

Lines 447-661

Purpose:

Demonstrate Curriculum Learning pattern

---

Key Concepts Summary

Pattern Overview

The Curriculum Learning pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 059
• Pattern 061

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Curriculum Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 060_curriculum_learning.py
Total Lines: 666
Pattern Category: AI Agent Design Pattern #060

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 059: Meta-Learning Agent

Description:
    Meta-learning (learning to learn) enables agents to improve their learning
    efficiency over time by learning general strategies that transfer across tasks.
    Rather than learning task-specific solutions, meta-learning agents learn how
    to adapt quickly to new tasks with minimal examples or experience.

Components:
    1. Task Distribution: Collection of related learning tasks
    2. Meta-Learner: Learns general learning strategies
    3. Base Learner: Applies strategies to specific tasks
    4. Adaptation Mechanism: Quick task-specific tuning
    5. Transfer Module: Knowledge transfer across tasks

Use Cases:
    - Few-shot learning scenarios
    - Rapid adaptation to new domains
    - Personalization with limited data
    - Multi-task learning systems
    - Continuous learning environments
    - Cross-domain knowledge transfer

LangChain Implementation:
    Implements meta-learning through prompt optimization, example selection,
    and strategy learning that transfers across tasks.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import statistics

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class MetaStrategy(Enum):
    """Meta-learning strategies"""
    MAML = "maml"  # Model-Agnostic Meta-Learning
    PROMPT_LEARNING = "prompt_learning"  # Learn optimal prompts
    EXAMPLE_SELECTION = "example_selection"  # Learn to select examples
    STRATEGY_LEARNING = "strategy_learning"  # Learn problem-solving strategies
    TRANSFER_LEARNING = "transfer_learning"  # Transfer knowledge across tasks


class TaskType(Enum):
    """Types of learning tasks"""
    CLASSIFICATION = "classification"
    GENERATION = "generation"
    REASONING = "reasoning"
    TRANSLATION = "translation"
    SUMMARIZATION = "summarization"


@dataclass
class Task:
    """A learning task"""
    task_id: str
    task_type: TaskType
    description: str
    examples: List[Tuple[str, str]]  # (input, output)
    test_cases: List[Tuple[str, str]]
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class LearnedStrategy:
    """A learned problem-solving strategy"""
    strategy_id: str
    applicable_tasks: List[TaskType]
    prompt_template: str
    example_count: int
    success_rate: float
    usage_count: int = 0
    
    def is_applicable(self, task_type: TaskType) -> bool:
        return task_type in self.applicable_tasks


@dataclass
class AdaptationResult:
    """Result from adaptation to new task"""
    task_id: str
    strategy_used: str
    examples_used: int
    performance: float
    adaptation_time_ms: float
    predictions: List[str]


@dataclass
class MetaLearningResult:
    """Result from meta-learning process"""
    tasks_trained: int
    strategies_learned: List[LearnedStrategy]
    avg_adaptation_performance: float
    total_training_time_ms: float
    transfer_effectiveness: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "tasks_trained": self.tasks_trained,
            "strategies_learned": len(self.strategies_learned),
            "avg_performance": f"{self.avg_adaptation_performance:.2%}",
            "transfer_effectiveness": f"{self.transfer_effectiveness:.2%}",
            "training_time_ms": f"{self.total_training_time_ms:.1f}"
        }


class MetaLearningAgent:
    """
    Agent that learns to learn across tasks.
    
    Features:
    1. Strategy learning from multiple tasks
    2. Fast adaptation to new tasks
    3. Optimal example selection
    4. Prompt optimization
    5. Knowledge transfer
    """
    
    def __init__(
        self,
        meta_strategy: MetaStrategy = MetaStrategy.STRATEGY_LEARNING,
        temperature: float = 0.7
    ):
        self.meta_strategy = meta_strategy
        self.temperature = temperature
        
        # Meta-learner (learns strategies)
        self.meta_learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.3
        )
        
        # Base learner (applies strategies)
        self.base_learner = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Learned strategies
        self.strategies: List[LearnedStrategy] = []
        
        # Task performance history
        self.performance_history: Dict[str, List[float]] = defaultdict(list)
        
        # Experience buffer
        self.experience: List[Tuple[Task, float]] = []
    
    def _learn_strategy_from_tasks(
        self,
        tasks: List[Task]
    ) -> LearnedStrategy:
        """Learn a general strategy from multiple tasks"""
        
        # Analyze patterns across tasks
        task_descriptions = "\n".join([
            f"Task {i+1} ({task.task_type.value}): {task.description}\n"
            f"Example: {task.examples[0][0]} → {task.examples[0][1]}"
            for i, task in enumerate(tasks[:3])
        ])
        
        strategy_prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze these learning tasks and identify a general strategy
that would work across all of them.

Provide:
1. Strategy name
2. General approach
3. Key steps
4. When to use

Be concise but specific."""),
            ("user", "Tasks:\n{tasks}\n\nGeneral strategy:")
        ])
        
        chain = strategy_prompt | self.meta_learner | StrOutputParser()
        strategy_description = chain.invoke({"tasks": task_descriptions})
        
        # Create prompt template from strategy
        prompt_template = f"""Strategy: {strategy_description}

Examples:
{{examples}}

Now apply this strategy to: {{query}}

Response:"""
        
        # Determine applicable task types
        task_types = list(set(t.task_type for t in tasks))
        
        strategy = LearnedStrategy(
            strategy_id=f"strategy_{len(self.strategies)}",
            applicable_tasks=task_types,
            prompt_template=prompt_template,
            example_count=3,
            success_rate=0.0  # Will be updated with experience
        )
        
        return strategy
    
    def _select_examples(
        self,
        task: Task,
        k: int = 3
    ) -> List[Tuple[str, str]]:
        """Select most informative examples for task"""
        
        # Simple selection: use first k examples
        # In production: use diversity, difficulty, relevance
        return task.examples[:k]
    
    def _evaluate_performance(
        self,
        predictions: List[str],
        ground_truth: List[str]
    ) -> float:
        """Evaluate prediction performance"""
        
        if not predictions or not ground_truth:
            return 0.0
        
        # Simple accuracy (in production: use task-specific metrics)
        correct = sum(
            1 for pred, truth in zip(predictions, ground_truth)
            if pred.strip().lower() == truth.strip().lower()
        )
        
        return correct / len(ground_truth)
    
    def meta_train(
        self,
        tasks: List[Task],
        episodes_per_task: int = 2
    ) -> MetaLearningResult:
        """Meta-training: Learn strategies from multiple tasks"""
        
        start_time = time.time()
        
        print(f"\n🎓 Meta-Training on {len(tasks)} tasks...")
        
        # Group tasks by type
        tasks_by_type = defaultdict(list)
        for task in tasks:
            tasks_by_type[task.task_type].append(task)
        
        # Learn strategies for each task type
        for task_type, type_tasks in tasks_by_type.items():
            if len(type_tasks) >= 2:  # Need multiple tasks to learn strategy
                print(f"\n  Learning strategy for {task_type.value} tasks...")
                
                strategy = self._learn_strategy_from_tasks(type_tasks)
                
                # Evaluate strategy on held-out examples
                performances = []
                for task in type_tasks:
                    # Use strategy to predict on test cases
                    examples = self._select_examples(task, k=2)
                    examples_text = "\n".join([
                        f"Input: {inp}\nOutput: {out}"
                        for inp, out in examples
                    ])
                    
                    predictions = []
                    for test_input, _ in task.test_cases[:3]:
                        prompt = strategy.prompt_template.format(
                            examples=examples_text,
                            query=test_input
                        )
                        
                        pred = self.base_learner.invoke(prompt).content
                        predictions.append(pred)
                    
                    # Evaluate
                    ground_truth = [out for _, out in task.test_cases[:3]]
                    perf = self._evaluate_performance(predictions, ground_truth)
                    performances.append(perf)
                    
                    # Store experience
                    self.experience.append((task, perf))
                
                # Update strategy success rate
                strategy.success_rate = statistics.mean(performances)
                self.strategies.append(strategy)
                
                print(f"    Strategy learned with {strategy.success_rate:.1%} success rate")
        
        # Calculate metrics
        all_performances = [perf for _, perf in self.experience]
        avg_performance = statistics.mean(all_performances) if all_performances else 0.0
        
        # Transfer effectiveness: improvement over baseline
        # (simplified - in production, compare to zero-shot)
        transfer_effectiveness = min(avg_performance * 1.2, 1.0)
        
        total_time_ms = (time.time() - start_time) * 1000
        
        return MetaLearningResult(
            tasks_trained=len(tasks),
            strategies_learned=self.strategies.copy(),
            avg_adaptation_performance=avg_performance,
            total_training_time_ms=total_time_ms,
            transfer_effectiveness=transfer_effectiveness
        )
    
    def adapt(
        self,
        new_task: Task,
        support_examples: Optional[List[Tuple[str, str]]] = None
    ) -> AdaptationResult:
        """Adapt to new task using learned strategies"""
        
        start_time = time.time()
        
        # Select best strategy for this task type
        applicable_strategies = [
            s for s in self.strategies
            if s.is_applicable(new_task.task_type)
        ]
        
        if applicable_strategies:
            # Use best performing strategy
            strategy = max(applicable_strategies, key=lambda s: s.success_rate)
            strategy.usage_count += 1
        else:
            # Fallback: learn from task examples only
            strategy = LearnedStrategy(
                strategy_id="fallback",
                applicable_tasks=[new_task.task_type],
                prompt_template="Examples:\n{examples}\n\nQuery: {query}\nResponse:",
                example_count=3,
                success_rate=0.5
            )
        
        # Select examples
        if support_examples:
            examples = support_examples
        else:
            examples = self._select_examples(new_task, k=strategy.example_count)
        
        examples_text = "\n".join([
            f"Input: {inp}\nOutput: {out}"
            for inp, out in examples
        ])
        
        # Make predictions on test cases
        predictions = []
        for test_input, _ in new_task.test_cases:
            prompt = strategy.prompt_template.format(
                examples=examples_text,
                query=test_input
            )
            
            pred = self.base_learner.invoke(prompt).content
            predictions.append(pred)
        
        # Evaluate performance
        ground_truth = [out for _, out in new_task.test_cases]
        performance = self._evaluate_performance(predictions, ground_truth)
        
        # Update strategy success rate
        self.performance_history[strategy.strategy_id].append(performance)
        if self.performance_history[strategy.strategy_id]:
            strategy.success_rate = statistics.mean(
                self.performance_history[strategy.strategy_id]
            )
        
        adaptation_time_ms = (time.time() - start_time) * 1000
        
        return AdaptationResult(
            task_id=new_task.task_id,
            strategy_used=strategy.strategy_id,
            examples_used=len(examples),
            performance=performance,
            adaptation_time_ms=adaptation_time_ms,
            predictions=predictions
        )
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get meta-learning statistics"""
        
        return {
            "strategies_learned": len(self.strategies),
            "total_experience": len(self.experience),
            "strategy_details": [
                {
                    "id": s.strategy_id,
                    "task_types": [t.value for t in s.applicable_tasks],
                    "success_rate": f"{s.success_rate:.1%}",
                    "usage_count": s.usage_count
                }
                for s in self.strategies
            ]
        }


def demonstrate_meta_learning():
    """Demonstrate Meta-Learning Agent pattern"""
    
    print("=" * 80)
    print("PATTERN 059: META-LEARNING AGENT DEMONSTRATION")
    print("=" * 80)
    print("\nLearning to learn across tasks\n")
    
    # Create meta-learning agent
    agent = MetaLearningAgent(
        meta_strategy=MetaStrategy.STRATEGY_LEARNING,
        temperature=0.7
    )
    
    # Create training tasks
    training_tasks = [
        Task(
            task_id="sentiment_1",
            task_type=TaskType.CLASSIFICATION,
            description="Classify sentiment as positive or negative",
            examples=[
                ("This movie was amazing!", "positive"),
                ("I hated this book.", "negative"),
                ("Great product, highly recommend.", "positive"),
            ],
            test_cases=[
                ("Loved it!", "positive"),
                ("Terrible experience.", "negative"),
            ]
        ),
        Task(
            task_id="sentiment_2",
            task_type=TaskType.CLASSIFICATION,
            description="Determine if review is positive or negative",
            examples=[
                ("Best purchase ever!", "positive"),
                ("Waste of money.", "negative"),
                ("Exceeded expectations.", "positive"),
            ],
            test_cases=[
                ("Fantastic!", "positive"),
                ("Very disappointed.", "negative"),
            ]
        ),
        Task(
            task_id="translate_1",
            task_type=TaskType.TRANSLATION,
            description="Translate English to simple terms",
            examples=[
                ("utilize", "use"),
                ("commence", "begin"),
                ("terminate", "end"),
            ],
            test_cases=[
                ("facilitate", "help"),
                ("acquire", "get"),
            ]
        ),
        Task(
            task_id="translate_2",
            task_type=TaskType.TRANSLATION,
            description="Convert formal to informal",
            examples=[
                ("I am pleased to inform you", "I'm happy to tell you"),
                ("Please be advised", "Just so you know"),
                ("At your earliest convenience", "When you can"),
            ],
            test_cases=[
                ("It is recommended", "You should"),
                ("We regret to inform", "Sorry to say"),
            ]
        ),
    ]
    
    # Test 1: Meta-training
    print("\n" + "=" * 80)
    print("TEST 1: Meta-Training on Multiple Tasks")
    print("=" * 80)
    
    print(f"\n📚 Training Tasks:")
    for task in training_tasks:
        print(f"   - {task.task_id} ({task.task_type.value}): {task.description}")
    
    meta_result = agent.meta_train(training_tasks, episodes_per_task=1)
    
    print(f"\n📊 Meta-Training Results:")
    print(f"   Tasks Trained: {meta_result.tasks_trained}")
    print(f"   Strategies Learned: {len(meta_result.strategies_learned)}")
    print(f"   Avg Performance: {meta_result.avg_adaptation_performance:.1%}")
    print(f"   Transfer Effectiveness: {meta_result.transfer_effectiveness:.1%}")
    print(f"   Training Time: {meta_result.total_training_time_ms:.1f}ms")
    
    print(f"\n🧠 Learned Strategies:")
    for strategy in meta_result.strategies_learned:
        print(f"   {strategy.strategy_id}:")
        print(f"      Task Types: {[t.value for t in strategy.applicable_tasks]}")
        print(f"      Success Rate: {strategy.success_rate:.1%}")
    
    # Test 2: Adaptation to new task (few-shot)
    print("\n" + "=" * 80)
    print("TEST 2: Fast Adaptation to New Task (Few-Shot)")
    print("=" * 80)
    
    new_task = Task(
        task_id="sentiment_new",
        task_type=TaskType.CLASSIFICATION,
        description="Classify restaurant reviews",
        examples=[
            ("Food was delicious!", "positive"),
            ("Service was terrible.", "negative"),
        ],
        test_cases=[
            ("Best meal I've had!", "positive"),
            ("Worst restaurant ever.", "negative"),
            ("Amazing ambiance.", "positive"),
        ]
    )
    
    print(f"\n🆕 New Task: {new_task.description}")
    print(f"   Task Type: {new_task.task_type.value}")
    print(f"   Training Examples: {len(new_task.examples)}")
    print(f"   Test Cases: {len(new_task.test_cases)}")
    
    adapt_result = agent.adapt(new_task)
    
    print(f"\n⚡ Adaptation Results:")
    print(f"   Strategy Used: {adapt_result.strategy_used}")
    print(f"   Examples Used: {adapt_result.examples_used}")
    print(f"   Performance: {adapt_result.performance:.1%}")
    print(f"   Adaptation Time: {adapt_result.adaptation_time_ms:.1f}ms")
    
    print(f"\n💭 Predictions:")
    for i, (test_input, _) in enumerate(new_task.test_cases):
        print(f"   Input: {test_input}")
        print(f"   Predicted: {adapt_result.predictions[i]}")
    
    # Test 3: Transfer across task types
    print("\n" + "=" * 80)
    print("TEST 3: Transfer Learning to Different Task Type")
    print("=" * 80)
    
    different_task = Task(
        task_id="translate_new",
        task_type=TaskType.TRANSLATION,
        description="Simplify technical jargon",
        examples=[
            ("optimize", "improve"),
            ("implement", "do"),
        ],
        test_cases=[
            ("leverage", "use"),
            ("paradigm", "model"),
        ]
    )
    
    print(f"\n🔄 Transfer Task: {different_task.description}")
    print(f"   Task Type: {different_task.task_type.value}")
    
    transfer_result = agent.adapt(different_task)
    
    print(f"\n📊 Transfer Results:")
    print(f"   Strategy: {transfer_result.strategy_used}")
    print(f"   Performance: {transfer_result.performance:.1%}")
    print(f"   Adaptation Time: {transfer_result.adaptation_time_ms:.1f}ms")
    
    # Show statistics
    print(f"\n📈 Agent Statistics:")
    stats = agent.get_statistics()
    print(f"   Total Strategies: {stats['strategies_learned']}")
    print(f"   Total Experience: {stats['total_experience']}")
    
    print(f"\n   Strategy Usage:")
    for strategy_info in stats['strategy_details']:
        if strategy_info['usage_count'] > 0:
            print(f"      {strategy_info['id']}: {strategy_info['usage_count']} times "
                  f"({strategy_info['success_rate']} success)")
    
    # Summary
    print("\n" + "=" * 80)
    print("META-LEARNING AGENT PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Fast Adaptation: Learn from few examples
2. Transfer Learning: Knowledge across tasks
3. Efficiency: Less data needed per task
4. Generalization: Learn general strategies
5. Continuous Improvement: Better over time

Meta-Learning Strategies:
1. MAML: Model-agnostic meta-learning
   - Learn initialization that adapts quickly
   
2. Prompt Learning: Optimal prompt discovery
   - Learn effective prompt templates
   
3. Example Selection: Choose informative examples
   - Learn which examples help most
   
4. Strategy Learning: General approaches
   - Learn problem-solving strategies
   
5. Transfer Learning: Cross-task knowledge
   - Transfer learned representations

Meta-Training Process:
1. Sample task distribution
2. For each task:
   - Support set: few training examples
   - Query set: test examples
3. Learn strategy from patterns
4. Evaluate on held-out tasks
5. Update meta-parameters

Adaptation Process:
1. Receive new task
2. Select applicable strategy
3. Choose informative examples
4. Apply learned approach
5. Make predictions
6. Update based on feedback

Components:
- Meta-Learner: Learns strategies
- Base Learner: Applies to tasks
- Strategy Library: Learned approaches
- Experience Buffer: Historical performance
- Transfer Module: Cross-task knowledge

Use Cases:
- Few-shot learning
- Personalization (limited user data)
- Multi-domain systems
- Rapid prototyping
- Cross-lingual transfer
- Continuous learning

Key Concepts:
1. Support Set: Training examples
2. Query Set: Test examples
3. Inner Loop: Task-specific learning
4. Outer Loop: Meta-learning update
5. Task Distribution: Set of related tasks

Best Practices:
1. Diverse training tasks
2. Appropriate support set size
3. Strategy evaluation
4. Performance monitoring
5. Adaptive example selection
6. Regular meta-updates
7. Transfer validation

Production Considerations:
- Strategy library management
- Performance tracking per strategy
- Task similarity metrics
- Example caching
- Incremental meta-learning
- Cold start handling
- Computational efficiency

Comparison with Related Patterns:
- vs. Transfer Learning: Meta vs direct transfer
- vs. Few-Shot: General framework vs specific technique
- vs. Fine-tuning: Strategy learning vs parameter tuning
- vs. Ensemble: Meta-learning vs combination

Challenges:
1. Task distribution design
2. Strategy generalization
3. Computational cost
4. Overfitting to task distribution
5. Evaluation methodology

The Meta-Learning Agent pattern enables rapid adaptation to new
tasks with minimal examples by learning general strategies that
transfer effectively across related problems.
""")


if __name__ == "__main__":
    demonstrate_meta_learning()

    

Agentic Design Pattern: Meta Learning

Pattern 059: Meta Learning – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Meta Learning pattern implementation.

---

Pattern Description

Pattern 059: Meta-Learning Agent

Description: Meta-learning (learning to learn) enables agents to improve their learning efficiency over time by learning general strategies that transfer across tasks. Rather than learning task-specific solutions, meta-learning agents learn how to adapt quickly to new tasks with minimal examples or experience.

Components: 1. Task Distribution: Collection of related learning tasks 2. Meta-Learner: Learns general learning strategies 3. Base Learner: Applies strategies to specific tasks 4. Adaptation Mechanism: Quick task-specific tuning 5. Transfer Module: Knowledge transfer across tasks

Use Cases: – Few-shot learning scenarios – Rapid adaptation to new domains – Personalization with limited data – Multi-task learning systems – Continuous learning environments – Cross-domain knowledge transfer

LangChain Implementation: Implements meta-learning through prompt optimization, example selection, and strategy learning that transfers across tasks.

---

File Statistics

• Total Lines: 692
• Classes: 7
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 059: Meta-Learning Agent

Description:
    Meta-learning (learning to learn) enables agents to improve their learning
    efficiency over time by learning general strategies that transfer across tasks.
    Rather than learning task-specific solutions, meta-learning agents learn how
    to adapt quickly to new tasks with minimal examples or experience.

Components:
    1. Task Distribution: Collection of related learning tasks
    2. Meta-Learner: Learns general learning strategies
    3. Base Learner: Applies strategies to specific tasks
    4. Adaptation Mechanism: Quick task-specific tuning
    5. Transfer Module: Knowledge transfer across tasks

Use Cases:
    - Few-shot learning scenarios
    - Rapid adaptation to new domains
    - Personalization with limited data
    - Multi-task learning systems
    - Continuous learning environments
    - Cross-domain knowledge transfer

LangChain Implementation:
    Implements meta-learning through prompt optimization, example selection,
    and strategy learning that transfers across tasks.
"""

Purpose: This module implements the Meta Learning pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import collections.defaultdict
import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from statistics

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: MetaStrategy

Lines 47-53

Inherits from: Enum

Purpose:

Meta-learning strategies

---

Class: TaskType

Lines 56-62

Inherits from: Enum

Purpose:

Types of learning tasks

---

Class: Task

Lines 66-73

Purpose:

A learning task

---

Class: LearnedStrategy

Lines 77-87

Purpose:

A learned problem-solving strategy

Methods:

is_applicable(self, task_type)

Lines 86-87

Purpose: Is Applicable operation.

Parameters: – task_type: Task type

Returns: bool

---

Class: AdaptationResult

Lines 91-98

Purpose:

Result from adaptation to new task

---

Class: MetaLearningResult

Lines 102-117

Purpose:

Result from meta-learning process

Methods:

to_dict(self)

Lines 110-117

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D628D50>

---

Class: MetaLearningAgent

Lines 120-406

Purpose:

Agent that learns to learn across tasks.

Features:
1. Strategy learning from multiple tasks
2. Fast adaptation to new tasks
3. Optimal example selection
4. Prompt optimization
5. Knowledge transfer

Methods:

__init__(self, meta_strategy, temperature)

Lines 132-159

Purpose: Initialize the class instance with required parameters.

Parameters: – meta_strategy: Meta strategy – temperature: Sampling temperature (0-1)

_learn_strategy_from_tasks(self, tasks)

Lines 161-212

Purpose:

Learn a general strategy from multiple tasks

Parameters: – tasks: Tasks

Returns: LearnedStrategy

_select_examples(self, task, k)

Lines 214-223

Purpose:

Select most informative examples for task

Parameters: – task: Task – k: K

Returns: <ast.Subscript object at 0x000001A06D683350>

_evaluate_performance(self, predictions, ground_truth)

Lines 225-241

Purpose:

Evaluate prediction performance

Parameters: – predictions: Predictions – ground_truth: Ground truth

Returns: float

meta_train(self, tasks, episodes_per_task)

Lines 243-316

Purpose:

Meta-training: Learn strategies from multiple tasks

Parameters: – tasks: Tasks – episodes_per_task: Episodes per task

Returns: MetaLearningResult

adapt(self, new_task, support_examples)

Lines 318-389

Purpose:

Adapt to new task using learned strategies

Parameters: – new_task: New task – support_examples: Support examples

Returns: AdaptationResult

get_statistics(self)

Lines 391-406

Purpose:

Get meta-learning statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D728F50>

---

Functions

demonstrate_meta_learning()

Lines 409-687

Purpose:

Demonstrate Meta-Learning Agent pattern

---

Key Concepts Summary

Pattern Overview

The Meta Learning pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 058
• Pattern 060

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Meta Learning pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 059_meta_learning.py
Total Lines: 692
Pattern Category: AI Agent Design Pattern #059

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 058: Neuro-Symbolic Integration

Description:
    Neuro-Symbolic Integration combines neural (sub-symbolic, learned) approaches
    with symbolic (logical, rule-based) reasoning. This hybrid approach leverages
    the pattern recognition and generalization of neural methods with the
    interpretability, logical rigor, and structured reasoning of symbolic systems.

Components:
    1. Neural Module: Pattern recognition, learning from data
    2. Symbolic Module: Logical reasoning, rules, knowledge graphs
    3. Translation Layer: Converts between neural and symbolic representations
    4. Integration Strategy: Combines both approaches
    5. Verification Engine: Validates symbolic constraints

Use Cases:
    - Explainable AI requiring both learning and reasoning
    - Knowledge-grounded generation
    - Constraint-based problem solving with learning
    - Mathematical reasoning with verification
    - Legal/medical systems requiring compliance
    - Scientific reasoning with logical constraints

LangChain Implementation:
    Combines LLM-based neural reasoning with explicit symbolic rules
    and knowledge representation for hybrid problem-solving.
"""

import os
import time
from typing import Dict, Any, List, Optional, Set, Tuple, Union
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import re

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class ReasoningMode(Enum):
    """Integration modes"""
    NEURAL_FIRST = "neural_first"  # Neural then symbolic verification
    SYMBOLIC_FIRST = "symbolic_first"  # Symbolic constraints then neural
    INTERLEAVED = "interleaved"  # Alternate between both
    PARALLEL = "parallel"  # Both simultaneously, then merge
    HYBRID = "hybrid"  # Integrated throughout


class SymbolicConstraintType(Enum):
    """Types of symbolic constraints"""
    LOGICAL = "logical"  # Boolean logic rules
    MATHEMATICAL = "mathematical"  # Numerical constraints
    TEMPORAL = "temporal"  # Time-based rules
    ONTOLOGICAL = "ontological"  # Knowledge hierarchy
    PROCEDURAL = "procedural"  # Step-by-step rules


@dataclass
class SymbolicRule:
    """Symbolic rule or constraint"""
    rule_id: str
    rule_type: SymbolicConstraintType
    description: str
    condition: str  # Logical expression
    action: Optional[str] = None
    priority: int = 1
    
    def evaluate(self, facts: Dict[str, Any]) -> bool:
        """Evaluate rule against facts"""
        try:
            # Simple evaluation (in production, use proper logic engine)
            for key, value in facts.items():
                self.condition = self.condition.replace(key, str(value))
            
            # Basic evaluation
            if "AND" in self.condition:
                parts = self.condition.split("AND")
                return all(self._evaluate_simple(p.strip(), facts) for p in parts)
            elif "OR" in self.condition:
                parts = self.condition.split("OR")
                return any(self._evaluate_simple(p.strip(), facts) for p in parts)
            else:
                return self._evaluate_simple(self.condition, facts)
        except:
            return False
    
    def _evaluate_simple(self, expr: str, facts: Dict[str, Any]) -> bool:
        """Evaluate simple expression"""
        try:
            # Very simple evaluation
            if ">" in expr:
                left, right = expr.split(">")
                return float(left.strip()) > float(right.strip())
            elif "<" in expr:
                left, right = expr.split("<")
                return float(left.strip())  List[Tuple[str, str, str]]:
        """Query knowledge graph"""
        results = []
        for rel in self.relations:
            if all(
                p is None or p == r
                for p, r in zip(pattern, rel)
            ):
                results.append(rel)
        return results
    
    def to_text(self) -> str:
        """Convert to text representation"""
        text = "Knowledge Base:\n"
        for subj, pred, obj in self.relations:
            text += f"  {subj} {pred} {obj}\n"
        return text


@dataclass
class NeuroSymbolicResult:
    """Result from neuro-symbolic reasoning"""
    query: str
    neural_output: str
    symbolic_verification: Dict[str, Any]
    final_output: str
    constraints_satisfied: bool
    reasoning_trace: List[str]
    confidence: float
    execution_time_ms: float
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:100] + "...",
            "constraints_satisfied": self.constraints_satisfied,
            "confidence": f"{self.confidence:.2f}",
            "reasoning_steps": len(self.reasoning_trace),
            "execution_time_ms": f"{self.execution_time_ms:.1f}"
        }


class NeuroSymbolicAgent:
    """
    Agent integrating neural and symbolic reasoning.
    
    Features:
    1. Neural reasoning via LLM
    2. Symbolic constraint checking
    3. Knowledge graph integration
    4. Logical verification
    5. Hybrid decision making
    """
    
    def __init__(
        self,
        reasoning_mode: ReasoningMode = ReasoningMode.NEURAL_FIRST,
        temperature: float = 0.3
    ):
        self.reasoning_mode = reasoning_mode
        self.temperature = temperature
        
        # Neural component (LLM)
        self.neural_model = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Symbolic components
        self.rules: List[SymbolicRule] = []
        self.knowledge_graph = KnowledgeGraph()
        
        # Reasoning trace
        self.reasoning_trace: List[str] = []
    
    def add_rule(self, rule: SymbolicRule):
        """Add symbolic rule"""
        self.rules.append(rule)
    
    def add_knowledge(
        self,
        subject: str,
        predicate: str,
        obj: str
    ):
        """Add knowledge to graph"""
        self.knowledge_graph.add_relation(subject, predicate, obj)
    
    def _neural_reasoning(self, query: str, context: Optional[str] = None) -> str:
        """Neural reasoning step"""
        
        self.reasoning_trace.append("Neural: Generating initial response")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", """You are reasoning about a query.
Provide a thoughtful, logical response.

{context}"""),
            ("user", "{query}")
        ])
        
        context_text = context if context else "Use your knowledge to respond."
        
        chain = prompt | self.neural_model | StrOutputParser()
        response = chain.invoke({
            "query": query,
            "context": context_text
        })
        
        return response
    
    def _symbolic_verification(
        self,
        neural_output: str,
        extracted_facts: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Verify neural output against symbolic constraints"""
        
        self.reasoning_trace.append("Symbolic: Verifying constraints")
        
        verification = {
            "all_satisfied": True,
            "violations": [],
            "satisfied_rules": [],
            "extracted_facts": extracted_facts
        }
        
        # Check each rule
        for rule in self.rules:
            satisfied = rule.evaluate(extracted_facts)
            
            if satisfied:
                verification["satisfied_rules"].append(rule.rule_id)
                self.reasoning_trace.append(f"  ✓ Rule {rule.rule_id}: {rule.description}")
            else:
                verification["all_satisfied"] = False
                verification["violations"].append({
                    "rule_id": rule.rule_id,
                    "description": rule.description,
                    "condition": rule.condition
                })
                self.reasoning_trace.append(f"  ✗ Rule {rule.rule_id}: {rule.description}")
        
        return verification
    
    def _extract_facts(self, text: str) -> Dict[str, Any]:
        """Extract facts from neural output"""
        
        self.reasoning_trace.append("Extracting facts from neural output")
        
        extraction_prompt = ChatPromptTemplate.from_messages([
            ("system", """Extract key facts and values from the text.

Format as key-value pairs:
key1: value1
key2: value2"""),
            ("user", "Text: {text}\n\nExtracted facts:")
        ])
        
        chain = extraction_prompt | self.neural_model | StrOutputParser()
        facts_text = chain.invoke({"text": text})
        
        # Parse facts
        facts = {}
        for line in facts_text.split('\n'):
            if ':' in line:
                key, value = line.split(':', 1)
                key = key.strip().replace(' ', '_').lower()
                value = value.strip()
                
                # Try to convert to number
                try:
                    value = float(value)
                except:
                    pass
                
                facts[key] = value
        
        return facts
    
    def _refine_with_constraints(
        self,
        query: str,
        neural_output: str,
        violations: List[Dict[str, Any]]
    ) -> str:
        """Refine neural output to satisfy constraints"""
        
        self.reasoning_trace.append("Neural: Refining to satisfy constraints")
        
        violations_text = "\n".join([
            f"- {v['description']}: {v['condition']}"
            for v in violations
        ])
        
        refinement_prompt = ChatPromptTemplate.from_messages([
            ("system", """Your previous response violated some constraints.
Revise your response to satisfy these constraints:

{violations}

Maintain the core meaning but adjust to meet all constraints."""),
            ("user", """Original query: {query}

Previous response: {response}

Revised response:""")
        ])
        
        chain = refinement_prompt | self.neural_model | StrOutputParser()
        refined = chain.invoke({
            "query": query,
            "response": neural_output,
            "violations": violations_text
        })
        
        return refined
    
    def _integrate_knowledge(self, query: str) -> str:
        """Integrate knowledge graph into reasoning"""
        
        # Query knowledge graph for relevant facts
        # (simplified - in production, use semantic matching)
        relevant_facts = []
        
        query_lower = query.lower()
        for subject, predicate, obj in self.knowledge_graph.relations:
            if (subject.lower() in query_lower or 
                obj.lower() in query_lower):
                relevant_facts.append(f"{subject} {predicate} {obj}")
        
        if relevant_facts:
            context = "Relevant knowledge:\n" + "\n".join(relevant_facts[:5])
            self.reasoning_trace.append(f"Retrieved {len(relevant_facts)} relevant facts")
            return context
        
        return ""
    
    def reason(
        self,
        query: str,
        max_refinement_iterations: int = 3
    ) -> NeuroSymbolicResult:
        """Perform neuro-symbolic reasoning"""
        
        start_time = time.time()
        self.reasoning_trace = []
        
        if self.reasoning_mode == ReasoningMode.NEURAL_FIRST:
            # Neural reasoning first, then symbolic verification
            
            # 1. Integrate knowledge
            context = self._integrate_knowledge(query)
            
            # 2. Neural reasoning
            neural_output = self._neural_reasoning(query, context)
            
            # 3. Extract facts
            facts = self._extract_facts(neural_output)
            
            # 4. Symbolic verification
            verification = self._symbolic_verification(neural_output, facts)
            
            # 5. Refine if needed
            final_output = neural_output
            iteration = 0
            
            while not verification["all_satisfied"] and iteration  0",
        priority=1
    ))
    
    agent1.add_rule(SymbolicRule(
        rule_id="reasonable_range",
        rule_type=SymbolicConstraintType.MATHEMATICAL,
        description="Result must be less than 1000",
        condition="result < 1000",
        priority=1
    ))
    
    query1 = "What is 25 multiplied by 8?"
    
    print(f"\n💭 Query: {query1}")
    print(f"🔧 Constraints:")
    for rule in agent1.rules:
        print(f"   - {rule.description}")
    
    result1 = agent1.reason(query1)
    
    print(f"\n🧠 Neural Output: {result1.neural_output[:150]}...")
    print(f"\n⚙️  Symbolic Verification:")
    print(f"   Constraints Satisfied: {result1.constraints_satisfied}")
    
    if result1.symbolic_verification["violations"]:
        print(f"   Violations:")
        for v in result1.symbolic_verification["violations"]:
            print(f"      - {v['description']}")
    
    if result1.symbolic_verification["satisfied_rules"]:
        print(f"   Satisfied Rules: {', '.join(result1.symbolic_verification['satisfied_rules'])}")
    
    print(f"\n✅ Final Output: {result1.final_output[:200]}...")
    print(f"   Confidence: {result1.confidence:.2f}")
    
    # Test 2: Knowledge graph integration
    print("\n" + "=" * 80)
    print("TEST 2: Knowledge Graph Integration")
    print("=" * 80)
    
    agent2 = NeuroSymbolicAgent(
        reasoning_mode=ReasoningMode.HYBRID
    )
    
    # Build knowledge graph
    agent2.add_knowledge("Python", "is_a", "programming_language")
    agent2.add_knowledge("Python", "created_by", "Guido_van_Rossum")
    agent2.add_knowledge("Python", "released_in", "1991")
    agent2.add_knowledge("Python", "used_for", "machine_learning")
    agent2.add_knowledge("Python", "used_for", "web_development")
    
    print(f"\n📚 Knowledge Graph:")
    for subj, pred, obj in agent2.knowledge_graph.relations:
        print(f"   {subj} {pred} {obj}")
    
    query2 = "When was Python created and what is it used for?"
    
    print(f"\n💭 Query: {query2}")
    
    result2 = agent2.reason(query2)
    
    print(f"\n🔍 Reasoning Trace:")
    for i, step in enumerate(result2.reasoning_trace[:5], 1):
        print(f"   {i}. {step}")
    
    print(f"\n💬 Response: {result2.final_output[:250]}...")
    
    # Test 3: Constraint refinement
    print("\n" + "=" * 80)
    print("TEST 3: Iterative Constraint Refinement")
    print("=" * 80)
    
    agent3 = NeuroSymbolicAgent(
        reasoning_mode=ReasoningMode.NEURAL_FIRST
    )
    
    # Add logical constraint
    agent3.add_rule(SymbolicRule(
        rule_id="format_check",
        rule_type=SymbolicConstraintType.LOGICAL,
        description="Response must include steps",
        condition="steps",  # Simplified
        priority=2
    ))
    
    query3 = "How do I solve a quadratic equation?"
    
    print(f"\n💭 Query: {query3}")
    
    result3 = agent3.reason(query3, max_refinement_iterations=2)
    
    print(f"\n🔄 Refinement Process:")
    print(f"   Total Steps: {len(result3.reasoning_trace)}")
    print(f"   Constraints Satisfied: {result3.constraints_satisfied}")
    
    print(f"\n📝 Reasoning Trace:")
    for step in result3.reasoning_trace:
        print(f"   - {step}")
    
    print(f"\n✅ Final Answer: {result3.final_output[:200]}...")
    
    # Summary
    print("\n" + "=" * 80)
    print("NEURO-SYMBOLIC INTEGRATION PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Best of Both: Neural flexibility + symbolic rigor
2. Explainability: Symbolic rules provide interpretability
3. Constraint Satisfaction: Guaranteed compliance
4. Knowledge Integration: Use structured knowledge
5. Verification: Validate neural outputs

Integration Modes:
1. Neural-First: Generate then verify
   - Fast, flexible generation
   - Symbolic post-processing

2. Symbolic-First: Constrain then generate
   - Guaranteed satisfaction
   - May limit creativity

3. Interleaved: Alternate steps
   - Balanced approach
   - More iterations

4. Parallel: Both simultaneously
   - Independent processing
   - Merge results

5. Hybrid: Fully integrated
   - Seamless combination
   - Complex implementation

Components:
1. Neural Module: LLM for pattern recognition
2. Symbolic Module: Rules, logic, knowledge graph
3. Translation Layer: Convert representations
4. Verification Engine: Check constraints
5. Refinement Loop: Iterative improvement

Symbolic Constraints:
- Logical: Boolean rules (AND, OR, NOT)
- Mathematical: Numerical constraints
- Temporal: Time-based rules
- Ontological: Knowledge hierarchies
- Procedural: Step sequences

Knowledge Representation:
- Knowledge Graphs: Entity-relation-entity
- Rules: If-then logic
- Ontologies: Hierarchical concepts
- Constraints: Mathematical inequalities
- Facts: Ground truth statements

Use Cases:
- Explainable AI systems
- Compliance-critical applications
- Mathematical reasoning
- Legal/medical systems
- Scientific hypothesis generation
- Code generation with verification

Best Practices:
1. Clear symbolic constraints
2. Efficient fact extraction
3. Iterative refinement limits
4. Knowledge graph maintenance
5. Rule priority ordering
6. Verification performance
7. Fallback strategies

Production Considerations:
- Rule engine scalability
- Knowledge graph indexing
- Refinement iteration limits
- Caching verified results
- Parallel verification
- Constraint complexity
- Performance optimization

Comparison with Related Patterns:
- vs. Pure Neural: Adds verification
- vs. Pure Symbolic: Adds flexibility
- vs. Retrieval: Structured knowledge vs search
- vs. Chain-of-Thought: Formal logic vs reasoning

Challenges:
1. Representation gap: Neural ↔ symbolic translation
2. Scalability: Large rule sets
3. Conflict resolution: Contradictory rules
4. Learning rules: Manual vs automated
5. Integration complexity

The Neuro-Symbolic Integration pattern enables robust AI systems
that combine the pattern recognition power of neural networks
with the interpretability and logical rigor of symbolic reasoning.
""")


if __name__ == "__main__":
    demonstrate_neuro_symbolic()

    

Agentic Design Pattern: Neuro Symbolic

Pattern 058: Neuro Symbolic – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Neuro Symbolic pattern implementation.

---

Pattern Description

Pattern 058: Neuro-Symbolic Integration

Description: Neuro-Symbolic Integration combines neural (sub-symbolic, learned) approaches with symbolic (logical, rule-based) reasoning. This hybrid approach leverages the pattern recognition and generalization of neural methods with the interpretability, logical rigor, and structured reasoning of symbolic systems.

Components: 1. Neural Module: Pattern recognition, learning from data 2. Symbolic Module: Logical reasoning, rules, knowledge graphs 3. Translation Layer: Converts between neural and symbolic representations 4. Integration Strategy: Combines both approaches 5. Verification Engine: Validates symbolic constraints

Use Cases: – Explainable AI requiring both learning and reasoning – Knowledge-grounded generation – Constraint-based problem solving with learning – Mathematical reasoning with verification – Legal/medical systems requiring compliance – Scientific reasoning with logical constraints

LangChain Implementation: Combines LLM-based neural reasoning with explicit symbolic rules and knowledge representation for hybrid problem-solving.

---

File Statistics

• Total Lines: 691
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 058: Neuro-Symbolic Integration

Description:
    Neuro-Symbolic Integration combines neural (sub-symbolic, learned) approaches
    with symbolic (logical, rule-based) reasoning. This hybrid approach leverages
    the pattern recognition and generalization of neural methods with the
    interpretability, logical rigor, and structured reasoning of symbolic systems.

Components:
    1. Neural Module: Pattern recognition, learning from data
    2. Symbolic Module: Logical reasoning, rules, knowledge graphs
    3. Translation Layer: Converts between neural and symbolic representations
    4. Integration Strategy: Combines both approaches
    5. Verification Engine: Validates symbolic constraints

Use Cases:
    - Explainable AI requiring both learning and reasoning
    - Knowledge-grounded generation
    - Constraint-based problem solving with learning
    - Mathematical reasoning with verification
    - Legal/medical systems requiring compliance
    - Scientific reasoning with logical constraints

LangChain Implementation:
    Combines LLM-based neural reasoning with explicit symbolic rules
    and knowledge representation for hybrid problem-solving.
"""

Purpose: This module implements the Neuro Symbolic pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import re
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set
import typing.Tuple
import typing.Union

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – re: Regular expression operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Union: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: ReasoningMode

Lines 46-52

Inherits from: Enum

Purpose:

Integration modes

---

Class: SymbolicConstraintType

Lines 55-61

Inherits from: Enum

Purpose:

Types of symbolic constraints

---

Class: SymbolicRule

Lines 65-113

Purpose:

Symbolic rule or constraint

Methods:

evaluate(self, facts)

Lines 74-91

Purpose:

Evaluate rule against facts

Parameters: – facts: Facts

Returns: bool

_evaluate_simple(self, expr, facts)

Lines 93-113

Purpose:

Evaluate simple expression

Parameters: – expr: Expr – facts: Facts

Returns: bool

---

Class: KnowledgeGraph

Lines 117-147

Purpose:

Simple knowledge graph

Methods:

add_entity(self, entity)

Lines 123-124

Purpose: Add Entity operation.

Parameters: – entity: Entity

add_relation(self, subject, predicate, obj)

Lines 126-129

Purpose: Add Relation operation.

Parameters: – subject: Subject – predicate: Predicate – obj: Obj

query(self, pattern)

Lines 131-140

Purpose:

Query knowledge graph

Parameters: – pattern: Pattern

Returns: <ast.Subscript object at 0x000001A06D632150>

to_text(self)

Lines 142-147

Purpose:

Convert to text representation

Parameters:

Returns: str

---

Class: NeuroSymbolicResult

Lines 151-169

Purpose:

Result from neuro-symbolic reasoning

Methods:

to_dict(self)

Lines 162-169

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D668610>

---

Class: NeuroSymbolicAgent

Lines 172-462

Purpose:

Agent integrating neural and symbolic reasoning.

Features:
1. Neural reasoning via LLM
2. Symbolic constraint checking
3. Knowledge graph integration
4. Logical verification
5. Hybrid decision making

Methods:

__init__(self, reasoning_mode, temperature)

Lines 184-203

Purpose: Initialize the class instance with required parameters.

Parameters: – reasoning_mode: Reasoning mode – temperature: Sampling temperature (0-1)

add_rule(self, rule)

Lines 205-207

Purpose:

Add symbolic rule

Parameters: – rule: Rule

add_knowledge(self, subject, predicate, obj)

Lines 209-216

Purpose:

Add knowledge to graph

Parameters: – subject: Subject – predicate: Predicate – obj: Obj

_neural_reasoning(self, query, context)

Lines 218-239

Purpose:

Neural reasoning step

Parameters: – query: Input query or question – context: Context

Returns: str

_symbolic_verification(self, neural_output, extracted_facts)

Lines 241-273

Purpose:

Verify neural output against symbolic constraints

Parameters: – neural_output: Neural output – extracted_facts: Extracted facts

Returns: <ast.Subscript object at 0x000001A06D626310>

_extract_facts(self, text)

Lines 275-308

Purpose:

Extract facts from neural output

Parameters: – text: Text

Returns: <ast.Subscript object at 0x000001A06D6B10D0>

_refine_with_constraints(self, query, neural_output, violations)

Lines 310-346

Purpose:

Refine neural output to satisfy constraints

Parameters: – query: Input query or question – neural_output: Neural output – violations: Violations

Returns: str

_integrate_knowledge(self, query)

Lines 348-366

Purpose:

Integrate knowledge graph into reasoning

Parameters: – query: Input query or question

Returns: str

reason(self, query, max_refinement_iterations)

Lines 368-462

Purpose:

Perform neuro-symbolic reasoning

Parameters: – query: Input query or question – max_refinement_iterations: Max refinement iterations

Returns: NeuroSymbolicResult

---

Functions

demonstrate_neuro_symbolic()

Lines 465-686

Purpose:

Demonstrate Neuro-Symbolic Integration pattern

---

Key Concepts Summary

Pattern Overview

The Neuro Symbolic pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 057
• Pattern 059

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Neuro Symbolic pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 058_neuro_symbolic.py
Total Lines: 691
Pattern Category: AI Agent Design Pattern #058

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 057: Attention Mechanism Patterns

Description:
    Attention mechanisms enable agents to selectively focus on the most relevant
    parts of large inputs, contexts, or action spaces. Inspired by human attention
    and transformer architectures, these patterns help agents efficiently process
    information by allocating computational resources where they matter most.

Components:
    1. Attention Scorer: Calculates relevance scores
    2. Context Selector: Chooses relevant information
    3. Focus Manager: Manages attention allocation
    4. Multi-Head Attention: Multiple attention perspectives
    5. Self-Attention: Internal context relationships

Use Cases:
    - Long document processing
    - Multi-source information synthesis
    - Large context windows
    - Memory retrieval from extensive history
    - Action selection in large action spaces
    - Feature selection in complex data

LangChain Implementation:
    Implements attention scoring and selection mechanisms using LLMs
    to identify and focus on relevant portions of context.
"""

import os
import time
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import math

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class AttentionType(Enum):
    """Types of attention mechanisms"""
    SELF_ATTENTION = "self_attention"  # Internal relationships
    CROSS_ATTENTION = "cross_attention"  # Query-context relationships
    MULTI_HEAD = "multi_head"  # Multiple attention perspectives
    SPARSE = "sparse"  # Attend to subset
    HIERARCHICAL = "hierarchical"  # Multi-level attention


class FocusMode(Enum):
    """Modes of attention focus"""
    NARROW = "narrow"  # Focus on very specific parts
    BROAD = "broad"  # Consider wider context
    ADAPTIVE = "adaptive"  # Adjust based on task
    DISTRIBUTED = "distributed"  # Spread across multiple items


@dataclass
class AttentionScore:
    """Attention score for an item"""
    item_id: str
    content: str
    score: float  # 0.0-1.0
    reasoning: Optional[str] = None
    
    def __repr__(self) -> str:
        return f"AttentionScore(id={self.item_id}, score={self.score:.3f})"


@dataclass
class AttentionHead:
    """Single attention head with specific perspective"""
    head_id: str
    perspective: str
    scores: List[AttentionScore]
    
    @property
    def top_items(self, k: int = 3) -> List[AttentionScore]:
        return sorted(self.scores, key=lambda x: x.score, reverse=True)[:k]


@dataclass
class AttentionResult:
    """Result from attention mechanism"""
    query: str
    attended_items: List[AttentionScore]
    attention_type: AttentionType
    focus_mode: FocusMode
    total_items: int
    items_attended: int
    attention_distribution: Dict[str, float]
    execution_time_ms: float
    
    @property
    def coverage(self) -> float:
        """Percentage of items attended"""
        return self.items_attended / self.total_items if self.total_items > 0 else 0.0
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "query": self.query[:80] + "...",
            "attention_type": self.attention_type.value,
            "focus_mode": self.focus_mode.value,
            "coverage": f"{self.coverage:.1%}",
            "top_attended": [
                f"{item.item_id} ({item.score:.2f})"
                for item in self.attended_items[:3]
            ],
            "execution_time_ms": f"{self.execution_time_ms:.1f}"
        }


class AttentionMechanism:
    """
    Implements attention mechanisms for selective focus.
    
    Features:
    1. Query-based attention scoring
    2. Multi-head attention for diverse perspectives
    3. Sparse attention for efficiency
    4. Hierarchical attention for structured data
    5. Adaptive focus modes
    """
    
    def __init__(
        self,
        attention_type: AttentionType = AttentionType.CROSS_ATTENTION,
        focus_mode: FocusMode = FocusMode.ADAPTIVE,
        num_heads: int = 3,
        temperature: float = 0.3
    ):
        self.attention_type = attention_type
        self.focus_mode = focus_mode
        self.num_heads = num_heads
        self.temperature = temperature
        
        # Attention scorer LLM
        self.scorer = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=temperature
        )
        
        # Synthesizer for multi-head combination
        self.synthesizer = ChatOpenAI(
            model_name="gpt-3.5-turbo",
            temperature=0.2
        )
    
    def _score_relevance(
        self,
        query: str,
        item: str,
        item_id: str,
        perspective: str = "general"
    ) -> AttentionScore:
        """Score relevance of item to query"""
        
        scoring_prompt = ChatPromptTemplate.from_messages([
            ("system", f"""You are scoring relevance from a {perspective} perspective.

Rate how relevant the item is to the query on a scale of 0.0 to 1.0.

Respond with just the number (e.g., 0.85)"""),
            ("user", "Query: {query}\n\nItem: {item}\n\nRelevance score:")
        ])
        
        chain = scoring_prompt | self.scorer | StrOutputParser()
        
        try:
            score_str = chain.invoke({"query": query, "item": item}).strip()
            # Extract number from response
            score = float(''.join(c for c in score_str if c.isdigit() or c == '.'))
            score = max(0.0, min(1.0, score))
        except:
            # Fallback: simple keyword matching
            query_words = set(query.lower().split())
            item_words = set(item.lower().split())
            overlap = len(query_words & item_words)
            score = min(1.0, overlap / max(len(query_words), 1) * 2)
        
        return AttentionScore(
            item_id=item_id,
            content=item,
            score=score,
            reasoning=f"{perspective} relevance"
        )
    
    def _cross_attention(
        self,
        query: str,
        context_items: List[Tuple[str, str]],  # (id, content)
        top_k: int = 5
    ) -> List[AttentionScore]:
        """Cross-attention: query attending to context"""
        
        scores = []
        
        for item_id, content in context_items:
            score = self._score_relevance(query, content, item_id)
            scores.append(score)
        
        # Apply softmax-like normalization
        total = sum(s.score for s in scores)
        if total > 0:
            for score in scores:
                score.score = score.score / total
        
        # Return top-k
        scores.sort(key=lambda x: x.score, reverse=True)
        return scores[:top_k]
    
    def _self_attention(
        self,
        items: List[Tuple[str, str]],  # (id, content)
        top_k: int = 5
    ) -> List[AttentionScore]:
        """Self-attention: items attending to each other"""
        
        # For each item, calculate its relationship to all others
        attention_matrix = []
        
        for i, (id1, content1) in enumerate(items):
            row_scores = []
            for j, (id2, content2) in enumerate(items):
                if i != j:
                    # Simple similarity (could use embeddings)
                    words1 = set(content1.lower().split())
                    words2 = set(content2.lower().split())
                    similarity = len(words1 & words2) / max(len(words1 | words2), 1)
                    row_scores.append((j, similarity))
            
            # Sum of attention from this item to others
            total_attention = sum(s for _, s in row_scores)
            attention_matrix.append((i, total_attention))
        
        # Sort by total attention received
        attention_matrix.sort(key=lambda x: x[1], reverse=True)
        
        # Return top-k most attended items
        result = []
        for idx, attention in attention_matrix[:top_k]:
            item_id, content = items[idx]
            result.append(AttentionScore(
                item_id=item_id,
                content=content,
                score=attention,
                reasoning="Self-attention importance"
            ))
        
        # Normalize scores
        total = sum(s.score for s in result)
        if total > 0:
            for score in result:
                score.score = score.score / total
        
        return result
    
    def _multi_head_attention(
        self,
        query: str,
        context_items: List[Tuple[str, str]],
        perspectives: List[str],
        top_k: int = 5
    ) -> Tuple[List[AttentionScore], List[AttentionHead]]:
        """Multi-head attention: multiple perspectives"""
        
        heads = []
        
        # Each head has different perspective
        for i, perspective in enumerate(perspectives[:self.num_heads]):
            head_scores = []
            
            for item_id, content in context_items:
                score = self._score_relevance(
                    query, content, item_id, perspective=perspective
                )
                head_scores.append(score)
            
            head = AttentionHead(
                head_id=f"head_{i}",
                perspective=perspective,
                scores=head_scores
            )
            heads.append(head)
        
        # Combine heads (average scores)
        combined_scores = {}
        for head in heads:
            for score in head.scores:
                if score.item_id not in combined_scores:
                    combined_scores[score.item_id] = {
                        "content": score.content,
                        "scores": []
                    }
                combined_scores[score.item_id]["scores"].append(score.score)
        
        # Average across heads
        final_scores = []
        for item_id, data in combined_scores.items():
            avg_score = sum(data["scores"]) / len(data["scores"])
            final_scores.append(AttentionScore(
                item_id=item_id,
                content=data["content"],
                score=avg_score,
                reasoning="Multi-head average"
            ))
        
        # Normalize and return top-k
        total = sum(s.score for s in final_scores)
        if total > 0:
            for score in final_scores:
                score.score = score.score / total
        
        final_scores.sort(key=lambda x: x.score, reverse=True)
        
        return final_scores[:top_k], heads
    
    def _sparse_attention(
        self,
        query: str,
        context_items: List[Tuple[str, str]],
        top_k: int = 5
    ) -> List[AttentionScore]:
        """Sparse attention: only attend to subset"""
        
        # Quick filtering using keywords
        query_keywords = set(query.lower().split())
        
        # Score only items with keyword overlap
        candidate_items = []
        for item_id, content in context_items:
            content_words = set(content.lower().split())
            if query_keywords & content_words:  # Has overlap
                candidate_items.append((item_id, content))
        
        # If too few candidates, expand
        if len(candidate_items)  AttentionResult:
        """Apply attention mechanism"""
        
        start_time = time.time()
        
        total_items = len(context_items)
        
        # Apply appropriate attention type
        if self.attention_type == AttentionType.CROSS_ATTENTION:
            attended = self._cross_attention(query, context_items, top_k)
            
        elif self.attention_type == AttentionType.SELF_ATTENTION:
            attended = self._self_attention(context_items, top_k)
            
        elif self.attention_type == AttentionType.MULTI_HEAD:
            perspectives = ["factual", "emotional", "practical"]
            attended, heads = self._multi_head_attention(
                query, context_items, perspectives, top_k
            )
            
        elif self.attention_type == AttentionType.SPARSE:
            attended = self._sparse_attention(query, context_items, top_k)
            
        else:  # Default to cross-attention
            attended = self._cross_attention(query, context_items, top_k)
        
        # Calculate attention distribution
        distribution = {
            item.item_id: item.score
            for item in attended
        }
        
        execution_time_ms = (time.time() - start_time) * 1000
        
        return AttentionResult(
            query=query,
            attended_items=attended,
            attention_type=self.attention_type,
            focus_mode=self.focus_mode,
            total_items=total_items,
            items_attended=len(attended),
            attention_distribution=distribution,
            execution_time_ms=execution_time_ms
        )
    
    def focused_response(
        self,
        query: str,
        context_items: List[Tuple[str, str]],
        top_k: int = 3
    ) -> Tuple[str, AttentionResult]:
        """Generate response using attended context"""
        
        # Apply attention
        attention = self.attend(query, context_items, top_k)
        
        # Build focused context
        focused_context = "\n\n".join([
            f"[Relevance: {item.score:.2f}] {item.content}"
            for item in attention.attended_items
        ])
        
        # Generate response with focused context
        response_prompt = ChatPromptTemplate.from_messages([
            ("system", """You have access to focused, relevant context.
Use this context to provide an accurate, helpful response.

Relevant Context:
{context}"""),
            ("user", "{query}")
        ])
        
        chain = response_prompt | self.synthesizer | StrOutputParser()
        response = chain.invoke({
            "query": query,
            "context": focused_context
        })
        
        return response, attention


def demonstrate_attention_mechanisms():
    """Demonstrate Attention Mechanism patterns"""
    
    print("=" * 80)
    print("PATTERN 057: ATTENTION MECHANISM PATTERNS DEMONSTRATION")
    print("=" * 80)
    print("\nSelective focus on relevant information\n")
    
    # Create sample context
    documents = [
        ("doc1", "Python is a high-level programming language known for its simplicity and readability."),
        ("doc2", "Machine learning models require large amounts of training data to perform well."),
        ("doc3", "The transformer architecture uses self-attention mechanisms for sequence processing."),
        ("doc4", "Python's simplicity makes it popular for machine learning and data science applications."),
        ("doc5", "Deep learning frameworks like TensorFlow and PyTorch are commonly used in Python."),
        ("doc6", "Attention mechanisms allow models to focus on relevant parts of the input."),
        ("doc7", "Natural language processing has benefited greatly from attention-based models."),
        ("doc8", "Python's extensive libraries make it suitable for various AI tasks."),
    ]
    
    # Test 1: Cross-attention
    print("\n" + "=" * 80)
    print("TEST 1: Cross-Attention (Query attending to Context)")
    print("=" * 80)
    
    attn1 = AttentionMechanism(
        attention_type=AttentionType.CROSS_ATTENTION,
        focus_mode=FocusMode.NARROW
    )
    
    query1 = "What is Python used for in machine learning?"
    
    print(f"\n💭 Query: {query1}")
    print(f"📚 Total Documents: {len(documents)}")
    
    result1 = attn1.attend(query1, documents, top_k=3)
    
    print(f"\n🎯 Attention Results:")
    print(f"   Coverage: {result1.coverage:.1%} ({result1.items_attended}/{result1.total_items})")
    print(f"   Execution Time: {result1.execution_time_ms:.1f}ms")
    
    print(f"\n📊 Top Attended Items:")
    for i, item in enumerate(result1.attended_items, 1):
        print(f"   {i}. {item.item_id} (score: {item.score:.3f})")
        print(f"      {item.content[:80]}...")
    
    # Test 2: Self-attention
    print("\n" + "=" * 80)
    print("TEST 2: Self-Attention (Items attending to each other)")
    print("=" * 80)
    
    attn2 = AttentionMechanism(
        attention_type=AttentionType.SELF_ATTENTION,
        focus_mode=FocusMode.ADAPTIVE
    )
    
    result2 = attn2.attend("", documents, top_k=4)  # No query for self-attention
    
    print(f"\n🔗 Most Interconnected Documents:")
    for i, item in enumerate(result2.attended_items, 1):
        print(f"   {i}. {item.item_id} (attention: {item.score:.3f})")
        print(f"      {item.content[:80]}...")
    
    # Test 3: Multi-head attention
    print("\n" + "=" * 80)
    print("TEST 3: Multi-Head Attention (Multiple Perspectives)")
    print("=" * 80)
    
    attn3 = AttentionMechanism(
        attention_type=AttentionType.MULTI_HEAD,
        num_heads=3
    )
    
    query3 = "Explain attention mechanisms"
    
    print(f"\n💭 Query: {query3}")
    print(f"👁️  Perspectives: factual, emotional, practical")
    
    result3 = attn3.attend(query3, documents, top_k=3)
    
    print(f"\n📊 Combined Multi-Head Results:")
    for i, item in enumerate(result3.attended_items, 1):
        print(f"   {i}. {item.item_id} (combined score: {item.score:.3f})")
        print(f"      {item.content[:80]}...")
    
    # Test 4: Focused response generation
    print("\n" + "=" * 80)
    print("TEST 4: Focused Response Generation")
    print("=" * 80)
    
    attn4 = AttentionMechanism(
        attention_type=AttentionType.CROSS_ATTENTION,
        focus_mode=FocusMode.NARROW
    )
    
    query4 = "Why is Python popular for AI development?"
    
    print(f"\n💭 Query: {query4}")
    
    response, attention = attn4.focused_response(query4, documents, top_k=3)
    
    print(f"\n🎯 Attention Focus:")
    for item in attention.attended_items:
        print(f"   - {item.item_id}: {item.score:.3f}")
    
    print(f"\n💬 Focused Response:")
    print(f"   {response}")
    
    # Test 5: Sparse attention
    print("\n" + "=" * 80)
    print("TEST 5: Sparse Attention (Efficient Processing)")
    print("=" * 80)
    
    # Create larger context
    large_docs = documents + [
        (f"doc{i}", f"This is an irrelevant document number {i}.")
        for i in range(9, 20)
    ]
    
    attn5 = AttentionMechanism(
        attention_type=AttentionType.SPARSE,
        focus_mode=FocusMode.ADAPTIVE
    )
    
    query5 = "transformer architecture attention"
    
    print(f"\n💭 Query: {query5}")
    print(f"📚 Total Documents: {len(large_docs)}")
    
    result5 = attn5.attend(query5, large_docs, top_k=3)
    
    print(f"\n⚡ Sparse Attention Results:")
    print(f"   Only scored relevant subset")
    print(f"   Execution Time: {result5.execution_time_ms:.1f}ms")
    
    print(f"\n🎯 Top Results:")
    for i, item in enumerate(result5.attended_items, 1):
        print(f"   {i}. {item.item_id}: {item.score:.3f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("ATTENTION MECHANISM PATTERNS SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Selective Focus: Process only relevant information
2. Scalability: Handle large contexts efficiently
3. Interpretability: Attention scores show reasoning
4. Quality: Focus improves response accuracy
5. Efficiency: Reduce computational cost

Attention Types:
1. Cross-Attention: Query attending to context
   - Use: Question answering, retrieval
   - Pattern: Query → Context scoring

2. Self-Attention: Internal relationships
   - Use: Document clustering, summarization
   - Pattern: Items → Mutual relevance

3. Multi-Head Attention: Multiple perspectives
   - Use: Complex analysis, diverse views
   - Pattern: Parallel attention heads → Combine

4. Sparse Attention: Subset focus
   - Use: Large-scale processing
   - Pattern: Filter → Attend to subset

5. Hierarchical Attention: Multi-level focus
   - Use: Structured documents
   - Pattern: Coarse → Fine attention

Focus Modes:
- Narrow: Highly selective (top-1 or top-3)
- Broad: Consider wider context (top-10+)
- Adaptive: Adjust based on task
- Distributed: Spread attention evenly

Scoring Methods:
1. LLM-based: Relevance scoring by LLM
2. Embedding-based: Cosine similarity
3. Keyword: Simple word overlap
4. Learned: Trained attention weights

Applications:
- Long document Q&A
- Multi-document synthesis
- Memory retrieval
- Context selection
- Feature importance
- Action space filtering

Implementation Patterns:
1. Score → Normalize → Select top-k
2. Multiple heads → Average/Weighted combine
3. Hierarchical: Sentence → Word attention
4. Causal: Attend only to past
5. Bidirectional: Full context attention

Best Practices:
1. Choose appropriate top-k
2. Normalize attention scores (softmax)
3. Multi-head for complex tasks
4. Sparse for large contexts
5. Cache attention scores
6. Visualize attention patterns
7. Combine with retrieval

Production Considerations:
- Caching scored items
- Parallel scoring
- Incremental attention updates
- Attention visualization
- Performance profiling
- Memory management
- Score normalization methods

Comparison with Related Patterns:
- vs. Retrieval: Attention vs search
- vs. Filtering: Soft vs hard selection
- vs. Ranking: Continuous scores vs discrete order
- vs. Memory: Attention for retrieval

Attention mechanisms are fundamental for efficient processing
of large contexts, enabling agents to focus computational
resources where they matter most for the task at hand.
""")


if __name__ == "__main__":
    demonstrate_attention_mechanisms()

    

Agentic Design Pattern: Attention Mechanisms

Pattern 057: Attention Mechanisms – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Attention Mechanisms pattern implementation.

---

Pattern Description

Pattern 057: Attention Mechanism Patterns

Description: Attention mechanisms enable agents to selectively focus on the most relevant parts of large inputs, contexts, or action spaces. Inspired by human attention and transformer architectures, these patterns help agents efficiently process information by allocating computational resources where they matter most.

Components: 1. Attention Scorer: Calculates relevance scores 2. Context Selector: Chooses relevant information 3. Focus Manager: Manages attention allocation 4. Multi-Head Attention: Multiple attention perspectives 5. Self-Attention: Internal context relationships

Use Cases: – Long document processing – Multi-source information synthesis – Large context windows – Memory retrieval from extensive history – Action selection in large action spaces – Feature selection in complex data

LangChain Implementation: Implements attention scoring and selection mechanisms using LLMs to identify and focus on relevant portions of context.

---

File Statistics

• Total Lines: 665
• Classes: 6
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 057: Attention Mechanism Patterns

Description:
    Attention mechanisms enable agents to selectively focus on the most relevant
    parts of large inputs, contexts, or action spaces. Inspired by human attention
    and transformer architectures, these patterns help agents efficiently process
    information by allocating computational resources where they matter most.

Components:
    1. Attention Scorer: Calculates relevance scores
    2. Context Selector: Chooses relevant information
    3. Focus Manager: Manages attention allocation
    4. Multi-Head Attention: Multiple attention perspectives
    5. Self-Attention: Internal context relationships

Use Cases:
    - Long document processing
    - Multi-source information synthesis
    - Large context windows
    - Memory retrieval from extensive history
    - Action selection in large action spaces
    - Feature selection in complex data

LangChain Implementation:
    Implements attention scoring and selection mechanisms using LLMs
    to identify and focus on relevant portions of context.
"""

Purpose: This module implements the Attention Mechanisms pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Tuple: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv
from math

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: AttentionType

Lines 46-52

Inherits from: Enum

Purpose:

Types of attention mechanisms

---

Class: FocusMode

Lines 55-60

Inherits from: Enum

Purpose:

Modes of attention focus

---

Class: AttentionScore

Lines 64-72

Purpose:

Attention score for an item

Methods:

__repr__(self)

Lines 71-72

Purpose: Repr operation.

Parameters:

Returns: str

---

Class: AttentionHead

Lines 76-84

Purpose:

Single attention head with specific perspective

Methods:

top_items(self, k)

Lines 83-84

Decorators: property

Purpose: Top Items operation.

Parameters: – k: K

Returns: <ast.Subscript object at 0x000001A06D70C4D0>

---

Class: AttentionResult

Lines 88-115

Purpose:

Result from attention mechanism

Methods:

coverage(self)

Lines 100-102

Decorators: property

Purpose:

Percentage of items attended

Parameters:

Returns: float

to_dict(self)

Lines 104-115

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D15A150>

---

Class: AttentionMechanism

Lines 118-431

Purpose:

Implements attention mechanisms for selective focus.

Features:
1. Query-based attention scoring
2. Multi-head attention for diverse perspectives
3. Sparse attention for efficiency
4. Hierarchical attention for structured data
5. Adaptive focus modes

Methods:

__init__(self, attention_type, focus_mode, num_heads, temperature)

Lines 130-152

Purpose: Initialize the class instance with required parameters.

Parameters: – attention_type: Attention type – focus_mode: Focus mode – num_heads: Num heads – temperature: Sampling temperature (0-1)

_score_relevance(self, query, item, item_id, perspective)

Lines 154-191

Purpose:

Score relevance of item to query

Parameters: – query: Input query or question – item: Item – item_id: Item id – perspective: Perspective

Returns: AttentionScore

_cross_attention(self, query, context_items, top_k)

Lines 193-215

Purpose:

Cross-attention: query attending to context

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D15D2D0>

_self_attention(self, items, top_k)

Lines 217-261

Purpose:

Self-attention: items attending to each other

Parameters: – items: Items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D7F7F90>

_multi_head_attention(self, query, context_items, perspectives, top_k)

Lines 263-321

Purpose:

Multi-head attention: multiple perspectives

Parameters: – query: Input query or question – context_items: Context items – perspectives: Perspectives – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D75A050>

_sparse_attention(self, query, context_items, top_k)

Lines 323-346

Purpose:

Sparse attention: only attend to subset

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D65AD50>

attend(self, query, context_items, top_k)

Lines 348-396

Purpose:

Apply attention mechanism

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: AttentionResult

focused_response(self, query, context_items, top_k)

Lines 398-431

Purpose:

Generate response using attended context

Parameters: – query: Input query or question – context_items: Context items – top_k: Top k

Returns: <ast.Subscript object at 0x000001A06D62BC10>

---

Functions

demonstrate_attention_mechanisms()

Lines 434-660

Purpose:

Demonstrate Attention Mechanism patterns

---

Key Concepts Summary

Pattern Overview

The Attention Mechanisms pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 056
• Pattern 058

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Attention Mechanisms pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 057_attention_mechanisms.py
Total Lines: 665
Pattern Category: AI Agent Design Pattern #057

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 056: Blackboard System

Description:
    The Blackboard System pattern implements a shared workspace (blackboard) where
    multiple independent knowledge sources (agents) collaboratively solve complex
    problems. Each knowledge source monitors the blackboard and contributes when
    it has relevant expertise, enabling opportunistic problem-solving through
    emergent collaboration.

Components:
    1. Blackboard: Shared data structure storing problem state
    2. Knowledge Sources (KS): Independent expert agents
    3. Control Module: Coordinates knowledge source activation
    4. Problem Space: Hierarchical representation of problem
    5. Solution Assembly: Combines partial solutions

Use Cases:
    - Complex multi-disciplinary problems
    - Speech and image recognition systems
    - Medical diagnosis with multiple specialists
    - Software architecture design
    - Scientific research synthesis
    - Strategic planning and decision-making

LangChain Implementation:
    Uses a shared state dictionary as blackboard, with multiple LLM-based
    knowledge sources that opportunistically contribute based on current state.
"""

import os
import time
from typing import Dict, Any, List, Optional, Callable, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json

from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

load_dotenv()


class KSPriority(Enum):
    """Priority levels for knowledge sources"""
    CRITICAL = 5
    HIGH = 4
    MEDIUM = 3
    LOW = 2
    MINIMAL = 1


class BlackboardState(Enum):
    """States of blackboard processing"""
    INITIALIZED = "initialized"
    GATHERING_INFO = "gathering_info"
    ANALYZING = "analyzing"
    GENERATING_SOLUTIONS = "generating_solutions"
    REFINING = "refining"
    COMPLETE = "complete"


@dataclass
class BlackboardEntry:
    """Entry on the blackboard"""
    entry_id: str
    layer: str  # hypothesis, evidence, solution, etc.
    content: str
    contributor: str  # which KS added this
    confidence: float  # 0.0-1.0
    timestamp: datetime
    dependencies: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def to_dict(self) -> Dict[str, Any]:
        return {
            "id": self.entry_id,
            "layer": self.layer,
            "content": self.content[:100] + "..." if len(self.content) > 100 else self.content,
            "contributor": self.contributor,
            "confidence": f"{self.confidence:.2f}",
            "timestamp": self.timestamp.isoformat()
        }


@dataclass
class KSActivation:
    """Knowledge source activation record"""
    ks_name: str
    trigger_condition: str
    contribution: str
    confidence: float
    timestamp: datetime


@dataclass
class Blackboard:
    """Shared workspace for collaborative problem-solving"""
    problem: str
    state: BlackboardState
    entries: List[BlackboardEntry] = field(default_factory=list)
    activations: List[KSActivation] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def add_entry(self, entry: BlackboardEntry):
        """Add entry to blackboard"""
        self.entries.append(entry)
    
    def get_entries_by_layer(self, layer: str) -> List[BlackboardEntry]:
        """Get all entries in a specific layer"""
        return [e for e in self.entries if e.layer == layer]
    
    def get_latest_entries(self, n: int = 5) -> List[BlackboardEntry]:
        """Get most recent entries"""
        return sorted(self.entries, key=lambda x: x.timestamp, reverse=True)[:n]
    
    def record_activation(self, activation: KSActivation):
        """Record knowledge source activation"""
        self.activations.append(activation)
    
    def get_state_summary(self) -> str:
        """Get summary of current blackboard state"""
        layer_counts = {}
        for entry in self.entries:
            layer_counts[entry.layer] = layer_counts.get(entry.layer, 0) + 1
        
        summary = f"Problem: {self.problem}\n"
        summary += f"State: {self.state.value}\n"
        summary += f"Total Entries: {len(self.entries)}\n"
        summary += "Entries by Layer:\n"
        for layer, count in layer_counts.items():
            summary += f"  - {layer}: {count}\n"
        
        return summary


class KnowledgeSource:
    """Independent expert agent that contributes to blackboard"""
    
    def __init__(
        self,
        name: str,
        expertise: str,
        trigger_conditions: List[str],
        priority: KSPriority,
        llm: ChatOpenAI,
        system_prompt: str
    ):
        self.name = name
        self.expertise = expertise
        self.trigger_conditions = trigger_conditions
        self.priority = priority
        self.llm = llm
        self.system_prompt = system_prompt
        self.activation_count = 0
    
    def should_activate(self, blackboard: Blackboard) -> tuple[bool, str]:
        """Determine if this KS should contribute"""
        
        # Check if any trigger conditions are met
        bb_summary = blackboard.get_state_summary()
        latest_entries = blackboard.get_latest_entries(3)
        
        for condition in self.trigger_conditions:
            # Simple keyword matching (could be more sophisticated)
            if condition.lower() in bb_summary.lower():
                return True, f"Triggered by: {condition}"
            
            # Check recent entries
            for entry in latest_entries:
                if condition.lower() in entry.content.lower():
                    return True, f"Triggered by entry: {entry.entry_id}"
        
        return False, ""
    
    def contribute(self, blackboard: Blackboard) -> Optional[BlackboardEntry]:
        """Generate contribution to blackboard"""
        
        # Get context from blackboard
        recent_entries = blackboard.get_latest_entries(5)
        context = "\n".join([
            f"[{e.layer}] {e.content[:100]}"
            for e in recent_entries
        ])
        
        # Generate contribution
        prompt = ChatPromptTemplate.from_messages([
            ("system", self.system_prompt),
            ("user", """Problem: {problem}

Current Blackboard State:
{context}

Based on your expertise in {expertise}, provide your contribution.
Be specific and build on existing information.

Your contribution:""")
        ])
        
        chain = prompt | self.llm | StrOutputParser()
        
        try:
            contribution = chain.invoke({
                "problem": blackboard.problem,
                "context": context or "Empty blackboard",
                "expertise": self.expertise
            })
            
            # Determine layer for contribution
            layer = "analysis"  # Default
            if "hypothesis" in contribution.lower() or "theory" in contribution.lower():
                layer = "hypothesis"
            elif "evidence" in contribution.lower() or "data" in contribution.lower():
                layer = "evidence"
            elif "solution" in contribution.lower() or "approach" in contribution.lower():
                layer = "solution"
            elif "recommend" in contribution.lower():
                layer = "recommendation"
            
            # Create entry
            entry = BlackboardEntry(
                entry_id=f"{self.name}_{len(blackboard.entries)}",
                layer=layer,
                content=contribution,
                contributor=self.name,
                confidence=0.7 + (self.priority.value * 0.05),
                timestamp=datetime.now()
            )
            
            self.activation_count += 1
            
            return entry
            
        except Exception as e:
            print(f"Error in {self.name} contribution: {str(e)}")
            return None


class BlackboardControl:
    """Controls knowledge source activation and coordination"""
    
    def __init__(self, problem: str):
        self.blackboard = Blackboard(
            problem=problem,
            state=BlackboardState.INITIALIZED
        )
        self.knowledge_sources: List[KnowledgeSource] = []
        self.max_iterations = 10
        self.current_iteration = 0
    
    def register_knowledge_source(self, ks: KnowledgeSource):
        """Register a knowledge source"""
        self.knowledge_sources.append(ks)
    
    def _select_next_ks(self) -> Optional[KnowledgeSource]:
        """Select next knowledge source to activate"""
        
        # Find all KS that can contribute
        candidates = []
        
        for ks in self.knowledge_sources:
            should_activate, trigger = ks.should_activate(self.blackboard)
            if should_activate:
                candidates.append((ks, trigger))
        
        if not candidates:
            return None
        
        # Sort by priority and activation count (prefer high priority, low activation)
        candidates.sort(
            key=lambda x: (x[0].priority.value, -x[0].activation_count),
            reverse=True
        )
        
        return candidates[0][0]
    
    def _update_state(self):
        """Update blackboard state based on progress"""
        
        entry_count = len(self.blackboard.entries)
        
        if entry_count == 0:
            self.blackboard.state = BlackboardState.INITIALIZED
        elif entry_count < 3:
            self.blackboard.state = BlackboardState.GATHERING_INFO
        elif entry_count < 6:
            self.blackboard.state = BlackboardState.ANALYZING
        elif entry_count  Blackboard:
        """Execute blackboard problem-solving cycle"""
        
        if max_iterations:
            self.max_iterations = max_iterations
        
        print(f"\n🎯 Starting blackboard problem-solving for: {self.blackboard.problem}")
        
        while self.current_iteration = 2 and len(recommendations) >= 1:
                print("\nSufficient solutions generated")
                self.blackboard.state = BlackboardState.COMPLETE
                break
        
        return self.blackboard


def create_default_knowledge_sources() -> List[KnowledgeSource]:
    """Create default set of knowledge sources"""
    
    knowledge_sources = []
    
    # Problem Analyzer
    ks1 = KnowledgeSource(
        name="ProblemAnalyzer",
        expertise="problem decomposition and analysis",
        trigger_conditions=["initialized", "problem"],
        priority=KSPriority.CRITICAL,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
        system_prompt="""You are a problem analysis expert. Break down complex problems
into components, identify key challenges, and frame the problem clearly."""
    )
    knowledge_sources.append(ks1)
    
    # Hypothesis Generator
    ks2 = KnowledgeSource(
        name="HypothesisGenerator",
        expertise="generating hypotheses and theories",
        trigger_conditions=["analysis", "decomp", "component"],
        priority=KSPriority.HIGH,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.7),
        system_prompt="""You are a creative thinker who generates hypotheses and theories.
Propose possible explanations and approaches based on the analysis."""
    )
    knowledge_sources.append(ks2)
    
    # Evidence Collector
    ks3 = KnowledgeSource(
        name="EvidenceCollector",
        expertise="gathering and evaluating evidence",
        trigger_conditions=["hypothesis", "theory", "claim"],
        priority=KSPriority.MEDIUM,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.2),
        system_prompt="""You are an evidence expert who evaluates claims and hypotheses.
Identify what evidence would support or refute the hypotheses."""
    )
    knowledge_sources.append(ks3)
    
    # Solution Designer
    ks4 = KnowledgeSource(
        name="SolutionDesigner",
        expertise="designing practical solutions",
        trigger_conditions=["evidence", "hypothesis", "analyzing"],
        priority=KSPriority.HIGH,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.5),
        system_prompt="""You are a solution architect who designs practical approaches.
Create concrete, actionable solutions based on the analysis and evidence."""
    )
    knowledge_sources.append(ks4)
    
    # Critic
    ks5 = KnowledgeSource(
        name="Critic",
        expertise="critical evaluation and refinement",
        trigger_conditions=["solution", "approach", "recommend"],
        priority=KSPriority.MEDIUM,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.4),
        system_prompt="""You are a critical evaluator who identifies weaknesses and risks.
Critique proposed solutions and suggest improvements."""
    )
    knowledge_sources.append(ks5)
    
    # Synthesizer
    ks6 = KnowledgeSource(
        name="Synthesizer",
        expertise="synthesis and final recommendations",
        trigger_conditions=["solution", "refining"],
        priority=KSPriority.CRITICAL,
        llm=ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.3),
        system_prompt="""You are a synthesizer who combines multiple perspectives.
Provide final recommendations that integrate all contributions."""
    )
    knowledge_sources.append(ks6)
    
    return knowledge_sources


def demonstrate_blackboard_system():
    """Demonstrate Blackboard System pattern"""
    
    print("=" * 80)
    print("PATTERN 056: BLACKBOARD SYSTEM DEMONSTRATION")
    print("=" * 80)
    print("\nCollaborative problem-solving through shared workspace\n")
    
    # Test 1: Technical problem
    print("\n" + "=" * 80)
    print("TEST 1: Technical Problem Solving")
    print("=" * 80)
    
    problem1 = "How can we improve the performance of a slow web application?"
    
    control1 = BlackboardControl(problem1)
    
    # Register knowledge sources
    for ks in create_default_knowledge_sources():
        control1.register_knowledge_source(ks)
    
    print(f"\nRegistered {len(control1.knowledge_sources)} knowledge sources:")
    for ks in control1.knowledge_sources:
        print(f"  - {ks.name}: {ks.expertise}")
    
    # Solve problem
    result1 = control1.solve(max_iterations=8)
    
    print(f"\n📊 Final Blackboard State:")
    print(f"   Total Entries: {len(result1.entries)}")
    print(f"   Total Activations: {len(result1.activations)}")
    print(f"   Final State: {result1.state.value}")
    
    # Show contributions by layer
    print(f"\n📚 Entries by Layer:")
    layers = {}
    for entry in result1.entries:
        if entry.layer not in layers:
            layers[entry.layer] = []
        layers[entry.layer].append(entry)
    
    for layer, entries in layers.items():
        print(f"\n   {layer.upper()} ({len(entries)} entries):")
        for entry in entries[:2]:  # Show first 2 per layer
            print(f"      [{entry.contributor}] {entry.content[:100]}...")
    
    # Test 2: Strategic planning problem
    print("\n" + "=" * 80)
    print("TEST 2: Strategic Planning Problem")
    print("=" * 80)
    
    problem2 = "What strategy should a startup use to compete with established companies?"
    
    control2 = BlackboardControl(problem2)
    for ks in create_default_knowledge_sources():
        control2.register_knowledge_source(ks)
    
    result2 = control2.solve(max_iterations=6)
    
    print(f"\n🎯 Problem Solving Summary:")
    print(f"   Problem: {problem2}")
    print(f"   Iterations: {control2.current_iteration}")
    print(f"   Knowledge Sources Activated:")
    
    # Count activations per KS
    ks_counts = {}
    for activation in result2.activations:
        ks_counts[activation.ks_name] = ks_counts.get(activation.ks_name, 0) + 1
    
    for ks_name, count in sorted(ks_counts.items(), key=lambda x: x[1], reverse=True):
        print(f"      {ks_name}: {count} times")
    
    # Show final recommendation
    recommendations = result2.get_entries_by_layer("recommendation")
    if recommendations:
        print(f"\n💡 Final Recommendation:")
        print(f"   {recommendations[-1].content[:300]}...")
    
    # Test 3: Show collaboration pattern
    print("\n" + "=" * 80)
    print("TEST 3: Collaboration Pattern Analysis")
    print("=" * 80)
    
    print(f"\n🔄 Knowledge Source Activation Sequence:")
    for i, activation in enumerate(result2.activations, 1):
        print(f"   {i}. {activation.ks_name} (confidence: {activation.confidence:.2f})")
    
    print(f"\n🏗️  Blackboard Evolution:")
    for i in range(0, len(result2.entries), 2):
        entry = result2.entries[i]
        print(f"   Entry {i+1}: [{entry.layer}] by {entry.contributor}")
    
    # Summary
    print("\n" + "=" * 80)
    print("BLACKBOARD SYSTEM PATTERN SUMMARY")
    print("=" * 80)
    print("""
Key Benefits:
1. Opportunistic Problem-Solving: KS contribute when relevant
2. Emergent Collaboration: No predefined workflow
3. Modularity: Independent knowledge sources
4. Incremental Progress: Builds solution step-by-step
5. Transparency: All contributions visible on blackboard

System Components:
1. Blackboard: Shared workspace with layered structure
2. Knowledge Sources: Independent expert agents
3. Control Module: Coordinates KS activation
4. Activation Conditions: Triggers for KS contributions
5. Priority System: Manages contribution order

Blackboard Layers:
- Hypothesis: Theories and possible explanations
- Evidence: Supporting or refuting data
- Analysis: Problem decomposition and insights
- Solution: Concrete approaches and methods
- Recommendation: Final synthesized advice

Knowledge Source Types:
1. Problem Analyzer: Breaks down problems
2. Hypothesis Generator: Creates theories
3. Evidence Collector: Gathers supporting data
4. Solution Designer: Designs approaches
5. Critic: Evaluates and refines
6. Synthesizer: Combines perspectives

Control Strategies:
1. Priority-Based: High-priority KS activate first
2. Trigger-Based: KS activate when conditions met
3. Load-Balanced: Distribute activations evenly
4. State-Driven: Blackboard state guides activation

Activation Patterns:
- Initial Analysis: Problem decomposition
- Hypothesis Generation: Multiple theories
- Evidence Collection: Supporting data
- Solution Design: Practical approaches
- Critical Evaluation: Refinement
- Synthesis: Final recommendations

Use Cases:
- Complex multi-disciplinary problems
- Medical diagnosis systems
- Scientific research synthesis
- Software architecture design
- Strategic planning
- Signal processing and recognition

Best Practices:
1. Clear KS expertise boundaries
2. Well-defined trigger conditions
3. Appropriate priority levels
4. Layered blackboard structure
5. Limit iteration count
6. Record all contributions
7. Final synthesis step

Production Considerations:
- Blackboard persistence
- Concurrent KS activation
- Conflict resolution
- Termination criteria
- Performance optimization
- Scalability with many KS
- Debugging and tracing

Comparison with Related Patterns:
- vs. Multi-Agent: Shared state vs communication
- vs. Pipeline: Opportunistic vs sequential
- vs. Mixture of Agents: Collaboration vs aggregation
- vs. Ensemble: Emergent vs predetermined combination

The Blackboard System pattern excels at complex problems requiring
multiple perspectives and emergent, opportunistic collaboration where
the solution path is not predetermined.
""")


if __name__ == "__main__":
    demonstrate_blackboard_system()

    

Agentic Design Pattern: Blackboard System

Pattern 056: Blackboard System – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Blackboard System pattern implementation.

---

Pattern Description

Pattern 056: Blackboard System

Description: The Blackboard System pattern implements a shared workspace (blackboard) where multiple independent knowledge sources (agents) collaboratively solve complex problems. Each knowledge source monitors the blackboard and contributes when it has relevant expertise, enabling opportunistic problem-solving through emergent collaboration.

Components: 1. Blackboard: Shared data structure storing problem state 2. Knowledge Sources (KS): Independent expert agents 3. Control Module: Coordinates knowledge source activation 4. Problem Space: Hierarchical representation of problem 5. Solution Assembly: Combines partial solutions

Use Cases: – Complex multi-disciplinary problems – Speech and image recognition systems – Medical diagnosis with multiple specialists – Software architecture design – Scientific research synthesis – Strategic planning and decision-making

LangChain Implementation: Uses a shared state dictionary as blackboard, with multiple LLM-based knowledge sources that opportunistically contribute based on current state.

---

File Statistics

• Total Lines: 611
• Classes: 7
• Functions: 2
• Constants: 0

---

Module Documentation

"""
Pattern 056: Blackboard System

Description:
    The Blackboard System pattern implements a shared workspace (blackboard) where
    multiple independent knowledge sources (agents) collaboratively solve complex
    problems. Each knowledge source monitors the blackboard and contributes when
    it has relevant expertise, enabling opportunistic problem-solving through
    emergent collaboration.

Components:
    1. Blackboard: Shared data structure storing problem state
    2. Knowledge Sources (KS): Independent expert agents
    3. Control Module: Coordinates knowledge source activation
    4. Problem Space: Hierarchical representation of problem
    5. Solution Assembly: Combines partial solutions

Use Cases:
    - Complex multi-disciplinary problems
    - Speech and image recognition systems
    - Medical diagnosis with multiple specialists
    - Software architecture design
    - Scientific research synthesis
    - Strategic planning and decision-making

LangChain Implementation:
    Uses a shared state dictionary as blackboard, with multiple LLM-based
    knowledge sources that opportunistically contribute based on current state.
"""

Purpose: This module implements the Blackboard System pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import dataclasses.field
import datetime.datetime
import enum.Enum
import json
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import time
import typing.Any
import typing.Callable
import typing.Dict
import typing.List
import typing.Optional
import typing.Set

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – dataclasses.field: Simplified class definitions with @dataclass – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – json: JSON serialization/deserialization – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – time: Time-related functions – typing.Any: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Callable: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Dict: Type hints for better code documentation (List, Dict, Any, Optional) – typing.List: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Optional: Type hints for better code documentation (List, Dict, Any, Optional) – typing.Set: Type hints for better code documentation (List, Dict, Any, Optional)

LangChain Imports

from langchain_openai.ChatOpenAI

Purpose: LangChain framework components: – LLM Interface: Wrapper for OpenAI models – Chains: Composable building blocks for complex workflows

Other Imports

from dotenv.load_dotenv

Purpose: Additional dependencies for specialized functionality.

---

Classes

Class: KSPriority

Lines 47-53

Inherits from: Enum

Purpose:

Priority levels for knowledge sources

---

Class: BlackboardState

Lines 56-63

Inherits from: Enum

Purpose:

States of blackboard processing

---

Class: BlackboardEntry

Lines 67-86

Purpose:

Entry on the blackboard

Methods:

to_dict(self)

Lines 78-86

Purpose: To Dict operation.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7664D0>

---

Class: KSActivation

Lines 90-96

Purpose:

Knowledge source activation record

---

Class: Blackboard

Lines 100-137

Purpose:

Shared workspace for collaborative problem-solving

Methods:

add_entry(self, entry)

Lines 108-110

Purpose:

Add entry to blackboard

Parameters: – entry: Entry

get_entries_by_layer(self, layer)

Lines 112-114

Purpose:

Get all entries in a specific layer

Parameters: – layer: Layer

Returns: <ast.Subscript object at 0x000001A06D6F48D0>

get_latest_entries(self, n)

Lines 116-118

Purpose:

Get most recent entries

Parameters: – n: N

Returns: <ast.Subscript object at 0x000001A06D6F5F90>

record_activation(self, activation)

Lines 120-122

Purpose:

Record knowledge source activation

Parameters: – activation: Activation

get_state_summary(self)

Lines 124-137

Purpose:

Get summary of current blackboard state

Parameters:

Returns: str

---

Class: KnowledgeSource

Lines 140-239

Purpose:

Independent expert agent that contributes to blackboard

Methods:

__init__(self, name, expertise, trigger_conditions, priority, llm, system_prompt)

Lines 143-158

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name – expertise: Expertise – trigger_conditions: Trigger conditions – priority: Priority – llm: Language model – system_prompt: Prompt template or text

should_activate(self, blackboard)

Lines 160-177

Purpose:

Determine if this KS should contribute

Parameters: – blackboard: Blackboard

Returns: <ast.Subscript object at 0x000001A06D16D310>

contribute(self, blackboard)

Lines 179-239

Purpose:

Generate contribution to blackboard

Parameters: – blackboard: Blackboard

Returns: <ast.Subscript object at 0x000001A06D626810>

---

Class: BlackboardControl

Lines 242-348

Purpose:

Controls knowledge source activation and coordination

Methods:

__init__(self, problem)

Lines 245-252

Purpose: Initialize the class instance with required parameters.

Parameters: – problem: Problem

register_knowledge_source(self, ks)

Lines 254-256

Purpose:

Register a knowledge source

Parameters: – ks: Ks

_select_next_ks(self)

Lines 258-278

Purpose:

Select next knowledge source to activate

Parameters:

Returns: <ast.Subscript object at 0x000001A06D60B210>

_update_state(self)

Lines 280-294

Purpose:

Update blackboard state based on progress

Parameters:

solve(self, max_iterations)

Lines 296-348

Purpose:

Execute blackboard problem-solving cycle

Parameters: – max_iterations: Max iterations

Returns: Blackboard

---

Functions

create_default_knowledge_sources()

Lines 351-428

Purpose:

Create default set of knowledge sources

Returns: <ast.Subscript object at 0x000001A06D60CC90>

demonstrate_blackboard_system()

Lines 431-606

Purpose:

Demonstrate Blackboard System pattern

---

Key Concepts Summary

Pattern Overview

The Blackboard System pattern provides a structured approach to building AI agents with specific capabilities and behaviors. This implementation leverages LangChain’s composable components to create a robust and maintainable solution.

Core Components

1. Agent Class: Main implementation of the pattern logic
2. Prompt Engineering: Carefully crafted prompts to guide LLM behavior
3. Tool Integration: Optional tools for extended functionality
4. Memory Management: State handling across interactions
5. Output Parsing: Structured response processing

When to Use

✅ Good for: – Scenarios requiring this specific pattern’s strengths – Applications needing structured agent behavior – Systems where explainability is important – Production environments requiring reliability

❌ Not ideal for: – Simple single-prompt tasks – Scenarios requiring different pattern characteristics – Resource-constrained environments (if computationally intensive)

---

Best Practices Demonstrated

1. Environment Configuration

• ✅ Use .env files for API keys
• ✅ Never hardcode credentials
• ✅ Validate environment variables on startup

2. Error Handling

• ✅ Graceful degradation on failures
• ✅ Informative error messages
• ✅ Logging for debugging

3. Prompt Engineering

• ✅ Clear, specific instructions
• ✅ Structured output formats
• ✅ Few-shot examples when beneficial

4. Code Organization

• ✅ Separation of concerns (class-based design)
• ✅ Type hints for clarity
• ✅ Docstrings for documentation

5. Performance Optimization

• ✅ Appropriate temperature settings
• ✅ Token budget management
• ✅ Caching where applicable

---

LangChain Components Used

• ChatOpenAI: OpenAI model interface
• Chains: Sequential component composition

Why LangChain? – Composable abstractions for LLM applications – Rich ecosystem of integrations – Production-ready patterns – Active community and development

---

Execution Flow

1. Initialize agent with configuration
   ↓
2. Load environment and dependencies
   ↓
3. Prepare prompt with input data
   ↓
4. Invoke LLM through LangChain
   ↓
5. Parse and validate response
   ↓
6. Return formatted result

Key Decision Points: – Model selection (temperature, model type) – Prompt structure (few-shot vs zero-shot) – Error handling strategy – Output format validation

---

Performance Considerations

Computational Cost

• LLM Calls: Primary cost factor (based on tokens)
• Latency: Depends on model and input size
• Memory: Minimal for most patterns

Optimization Strategies

1. Prompt Optimization: Minimize tokens while maintaining clarity
2. Caching: Cache repeated queries/responses
3. Model Selection: Use smaller models when appropriate
4. Batch Processing: Group similar requests
5. Streaming: For long responses, stream tokens

Cost Estimates (Approximate)

• GPT-3.5-turbo: ~$0.002 per request
• GPT-4: ~$0.03 per request
• GPT-4-turbo: ~$0.01 per request

Actual costs vary based on input/output token counts

---

Common Pitfalls to Avoid

1. Prompt Design Issues

❌ Vague or ambiguous instructions
✅ Clear, specific, testable prompts

2. Error Handling

❌ Ignoring API errors or rate limits
✅ Implement retry logic and graceful degradation

3. Token Management

❌ Exceeding model token limits
✅ Truncate inputs, chunk processing

4. Security

❌ Exposing API keys in code
✅ Use environment variables

5. Testing

❌ Only testing happy paths
✅ Test edge cases, failures, malformed inputs

6. Cost Management

❌ Unmonitored API usage
✅ Implement budgets, alerts, and monitoring

---

Further Improvements

Production Enhancements

1. Monitoring: Add logging, metrics, tracing
2. Validation: Input/output schema validation
3. Rate Limiting: Implement request throttling
4. Caching: Redis or in-memory cache for responses
5. Async Support: Non-blocking operations for scalability

Feature Extensions

1. Multi-Model Support: Switch between providers
2. Fallback Models: Graceful degradation strategy
3. Custom Tools: Domain-specific function calling
4. Advanced Memory: Vector stores for long-term memory
5. Evaluation: Automated testing and benchmarking

Code Quality

1. Testing: Unit tests, integration tests
2. Documentation: API docs, usage examples
3. Type Safety: Stricter type hints
4. Configuration: Externalized config files

---

Related Patterns

• Pattern 055
• Pattern 057

Pattern Families: – Reasoning Patterns (001-011) – Multi-Agent Patterns (012-019) – Tool Use Patterns (020-025) – Memory Patterns (026-031) – And many more…

---

Summary

The Blackboard System pattern provides a powerful approach to building AI agents with specific capabilities. This implementation demonstrates:

✅ Key Strengths: – Clean, maintainable code structure – LangChain integration for composability – Type hints and documentation – Production-ready patterns

🎯 Best Use Cases: – Applications requiring this pattern’s characteristics – Systems needing structured agent behavior – Production environments requiring reliability

📚 Learning Resources: – LangChain Documentation: https://python.langchain.com/ – Pattern Documentation: See related patterns – OpenAI API Reference: https://platform.openai.com/docs/

---

File: 056_blackboard_system.py
Total Lines: 611
Pattern Category: AI Agent Design Pattern #056

---

Generated from actual implementation code
For educational and reference purposes