"""
Pattern 131: Persona Consistency

Description:
    The Persona Consistency pattern enables agents to maintain a stable, coherent
    character or personality across all interactions. Rather than exhibiting
    inconsistent behavior or tone, the agent adheres to a defined persona with
    consistent traits, communication style, values, knowledge boundaries, and
    behavioral patterns throughout conversations and across sessions.
    
    Persona consistency is essential for creating believable, trustworthy AI
    characters in applications like virtual assistants, game NPCs, brand
    representatives, and therapeutic chatbots. The pattern addresses challenges
    like personality drift, contextual appropriateness, multi-turn coherence,
    and balancing consistency with natural variation.
    
    This pattern includes mechanisms for persona definition, trait enforcement,
    style consistency, memory of character history, contextual adaptation within
    persona bounds, and continuous validation of persona adherence.

Key Components:
    1. Persona Definition: Core character traits and attributes
    2. Style Guide: Communication patterns and language
    3. Knowledge Boundaries: What the persona knows/doesn't know
    4. Behavioral Rules: How persona responds to situations
    5. Memory System: Track persona history and consistency
    6. Consistency Checker: Validate responses against persona
    7. Adaptation Layer: Natural variation within constraints

Persona Attributes:
    1. Core Traits: Personality characteristics (friendly, formal, humorous)
    2. Communication Style: Tone, vocabulary, sentence structure
    3. Values & Beliefs: Moral framework, priorities
    4. Knowledge Domain: Areas of expertise and ignorance
    5. Emotional Range: How persona expresses emotions
    6. Cultural Background: Context influencing behavior
    7. Quirks & Habits: Unique characteristics

Consistency Dimensions:
    1. Linguistic: Word choice, grammar, dialect
    2. Behavioral: Response patterns, decision-making
    3. Emotional: Mood stability, reaction patterns
    4. Knowledge: What persona knows consistently
    5. Temporal: Consistency over time
    6. Contextual: Appropriate adaptation
    7. Relational: Consistent with relationship history

Use Cases:
    - Virtual assistants with brand personality
    - Game NPCs with defined characters
    - Customer service representatives
    - Educational tutors with consistent approach
    - Therapeutic/counseling chatbots
    - Celebrity or character impersonations
    - Brand ambassadors

Advantages:
    - Builds user trust and familiarity
    - Creates memorable experiences
    - Strengthens brand identity
    - Enables emotional connection
    - Provides predictable interactions
    - Maintains character integrity
    - Enhances believability

Challenges:
    - Balancing consistency with flexibility
    - Avoiding robotic repetition
    - Adapting to context while staying in character
    - Managing persona evolution
    - Handling edge cases in character
    - Maintaining consistency across sessions
    - Preventing personality drift

LangChain Implementation:
    This implementation uses LangChain for:
    - Persona-aware prompt construction
    - Response consistency validation
    - Character memory management
    - Style-constrained generation
    
Production Considerations:
    - Store persona definitions persistently
    - Version control persona changes
    - Monitor consistency metrics
    - Allow graceful persona updates
    - Support multiple personas per agent
    - Enable A/B testing of personas
    - Provide consistency reporting
    - Log persona violations for improvement
"""

import os
from typing import List, Dict, Any, Optional
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 PersonaTrait(Enum):
    """Core personality traits."""
    FRIENDLY = "friendly"
    FORMAL = "formal"
    HUMOROUS = "humorous"
    SERIOUS = "serious"
    EMPATHETIC = "empathetic"
    ANALYTICAL = "analytical"
    CREATIVE = "creative"
    PRACTICAL = "practical"


class CommunicationStyle(Enum):
    """Communication style options."""
    CASUAL = "casual"
    PROFESSIONAL = "professional"
    TECHNICAL = "technical"
    SIMPLE = "simple"
    VERBOSE = "verbose"
    CONCISE = "concise"


@dataclass
class PersonaDefinition:
    """
    Complete definition of an agent's persona.
    
    Attributes:
        name: Persona name
        description: Brief description
        core_traits: Personality traits
        communication_style: How persona communicates
        values: Core values and beliefs
        knowledge_domains: Areas of expertise
        knowledge_gaps: What persona doesn't know
        emotional_range: How emotions are expressed
        quirks: Unique characteristics
        sample_phrases: Example phrases persona uses
        behavioral_rules: Guidelines for behavior
    """
    name: str
    description: str
    core_traits: List[PersonaTrait] = field(default_factory=list)
    communication_style: CommunicationStyle = CommunicationStyle.PROFESSIONAL
    values: List[str] = field(default_factory=list)
    knowledge_domains: List[str] = field(default_factory=list)
    knowledge_gaps: List[str] = field(default_factory=list)
    emotional_range: str = "moderate"
    quirks: List[str] = field(default_factory=list)
    sample_phrases: List[str] = field(default_factory=list)
    behavioral_rules: List[str] = field(default_factory=list)


@dataclass
class PersonaInteraction:
    """
    Record of an interaction with persona.
    
    Attributes:
        interaction_id: Unique identifier
        user_input: User's message
        persona_response: Agent's response
        timestamp: When interaction occurred
        consistency_score: How well response matched persona
        violations: Any detected inconsistencies
    """
    interaction_id: str
    user_input: str
    persona_response: str
    timestamp: datetime = field(default_factory=datetime.now)
    consistency_score: float = 1.0
    violations: List[str] = field(default_factory=list)


class PersonaAgent:
    """
    Agent that maintains consistent persona across interactions.
    
    This agent ensures all responses adhere to a defined persona,
    maintaining character consistency across conversations.
    """
    
    def __init__(self, persona: PersonaDefinition, temperature: float = 0.7):
        """
        Initialize persona agent.
        
        Args:
            persona: Persona definition
            temperature: LLM temperature (higher for more variation)
        """
        self.persona = persona
        self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
        self.interactions: List[PersonaInteraction] = []
        self.interaction_counter = 0
    
    def _build_persona_prompt(self) -> str:
        """Build persona description for prompts."""
        traits_str = ", ".join([t.value for t in self.persona.core_traits])
        
        prompt_parts = [
            f"You are {self.persona.name}, {self.persona.description}",
            f"Personality traits: {traits_str}",
            f"Communication style: {self.persona.communication_style.value}",
        ]
        
        if self.persona.values:
            values_str = ", ".join(self.persona.values)
            prompt_parts.append(f"Core values: {values_str}")
        
        if self.persona.knowledge_domains:
            domains_str = ", ".join(self.persona.knowledge_domains)
            prompt_parts.append(f"Areas of expertise: {domains_str}")
        
        if self.persona.knowledge_gaps:
            gaps_str = ", ".join(self.persona.knowledge_gaps)
            prompt_parts.append(f"You do not have expertise in: {gaps_str}")
        
        if self.persona.quirks:
            quirks_str = ", ".join(self.persona.quirks)
            prompt_parts.append(f"Unique characteristics: {quirks_str}")
        
        if self.persona.behavioral_rules:
            rules_str = "\n".join([f"- {rule}" for rule in self.persona.behavioral_rules])
            prompt_parts.append(f"Behavioral guidelines:\n{rules_str}")
        
        return "\n".join(prompt_parts)
    
    def respond(self, user_input: str, context: Optional[str] = None) -> str:
        """
        Generate response in character.
        
        Args:
            user_input: User's message
            context: Optional conversation context
            
        Returns:
            Persona's response
        """
        persona_description = self._build_persona_prompt()
        
        # Build prompt with persona and context
        prompt_template = (
            "{persona_description}\n\n"
            "Stay completely in character. Respond as {persona_name} would, "
            "maintaining consistency with your personality, communication style, "
            "and knowledge boundaries.\n\n"
        )
        
        if context:
            prompt_template += "Previous context:\n{context}\n\n"
        
        prompt_template += "User: {user_input}\n\n{persona_name}:"
        
        prompt = ChatPromptTemplate.from_template(prompt_template)
        chain = prompt | self.llm | StrOutputParser()
        
        input_data = {
            "persona_description": persona_description,
            "persona_name": self.persona.name,
            "user_input": user_input
        }
        
        if context:
            input_data["context"] = context
        
        response = chain.invoke(input_data)
        
        # Record interaction
        self.interaction_counter += 1
        interaction = PersonaInteraction(
            interaction_id=f"int_{self.interaction_counter}",
            user_input=user_input,
            persona_response=response
        )
        self.interactions.append(interaction)
        
        return response
    
    def check_consistency(self, response: str) -> Dict[str, Any]:
        """
        Check if response is consistent with persona.
        
        Args:
            response: Response to check
            
        Returns:
            Consistency analysis
        """
        persona_description = self._build_persona_prompt()
        
        prompt = ChatPromptTemplate.from_template(
            "Persona Definition:\n"
            "{persona_description}\n\n"
            "Response to evaluate:\n"
            "{response}\n\n"
            "Analyze whether this response is consistent with the persona. "
            "Consider:\n"
            "1. Personality traits\n"
            "2. Communication style\n"
            "3. Values and beliefs\n"
            "4. Knowledge boundaries\n"
            "5. Characteristic quirks\n\n"
            "Provide:\n"
            "CONSISTENT: yes/no\n"
            "SCORE: 0-10\n"
            "VIOLATIONS: [list any inconsistencies]\n"
            "EXPLANATION: [brief explanation]"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        analysis = chain.invoke({
            "persona_description": persona_description,
            "response": response
        })
        
        # Parse analysis
        lines = analysis.strip().split('\n')
        result = {
            "raw_analysis": analysis,
            "consistent": True,
            "score": 8.0,
            "violations": []
        }
        
        for line in lines:
            if line.startswith("CONSISTENT:"):
                result["consistent"] = "yes" in line.lower()
            elif line.startswith("SCORE:"):
                try:
                    result["score"] = float(line.split(":")[1].strip().split()[0])
                except:
                    pass
        
        return result
    
    def get_conversation_context(self, num_recent: int = 3) -> str:
        """
        Get recent conversation history.
        
        Args:
            num_recent: Number of recent interactions
            
        Returns:
            Formatted context
        """
        recent = self.interactions[-num_recent:]
        
        context_lines = []
        for interaction in recent:
            context_lines.append(f"User: {interaction.user_input}")
            context_lines.append(f"{self.persona.name}: {interaction.persona_response}")
        
        return "\n".join(context_lines)
    
    def get_consistency_report(self) -> Dict[str, Any]:
        """Get report on persona consistency."""
        if not self.interactions:
            return {
                "total_interactions": 0,
                "average_consistency": 0.0,
                "violations": []
            }
        
        total_score = sum(i.consistency_score for i in self.interactions)
        avg_score = total_score / len(self.interactions)
        
        all_violations = []
        for interaction in self.interactions:
            all_violations.extend(interaction.violations)
        
        return {
            "total_interactions": len(self.interactions),
            "average_consistency": avg_score,
            "violations": all_violations,
            "persona_name": self.persona.name
        }


def demonstrate_persona_consistency():
    """Demonstrate persona consistency pattern."""
    
    print("=" * 80)
    print("PERSONA CONSISTENCY PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Defining a persona
    print("\n" + "=" * 80)
    print("Example 1: Defining a Persona")
    print("=" * 80)
    
    persona = PersonaDefinition(
        name="Professor Ada",
        description="an enthusiastic computer science professor who loves making complex topics accessible",
        core_traits=[
            PersonaTrait.FRIENDLY,
            PersonaTrait.ANALYTICAL,
            PersonaTrait.EMPATHETIC
        ],
        communication_style=CommunicationStyle.PROFESSIONAL,
        values=[
            "Education for all",
            "Clarity over jargon",
            "Practical application"
        ],
        knowledge_domains=[
            "Computer Science",
            "Programming",
            "Algorithms",
            "Software Engineering"
        ],
        knowledge_gaps=[
            "Advanced Physics",
            "Medical procedures",
            "Legal advice"
        ],
        quirks=[
            "Often uses teaching analogies",
            "Encourages questions",
            "References classic CS papers"
        ],
        sample_phrases=[
            "That's a great question!",
            "Let me break that down...",
            "Think of it like this..."
        ],
        behavioral_rules=[
            "Always encourage learning",
            "Admit when something is outside expertise",
            "Use analogies to explain complex concepts",
            "Be patient and supportive"
        ]
    )
    
    print(f"\nPersona: {persona.name}")
    print(f"Description: {persona.description}")
    print(f"\nCore Traits: {[t.value for t in persona.core_traits]}")
    print(f"Style: {persona.communication_style.value}")
    print(f"Values: {persona.values}")
    print(f"Expertise: {persona.knowledge_domains}")
    
    # Example 2: In-character responses
    print("\n" + "=" * 80)
    print("Example 2: In-Character Responses")
    print("=" * 80)
    
    agent = PersonaAgent(persona)
    
    # Test responses
    questions = [
        "What is recursion?",
        "Can you explain it simply?",
        "What about medical imaging algorithms?"
    ]
    
    print("\nConversation:")
    for question in questions:
        print(f"\nUser: {question}")
        response = agent.respond(question)
        print(f"{persona.name}: {response}")
    
    # Example 3: Consistency checking
    print("\n" + "=" * 80)
    print("Example 3: Consistency Checking")
    print("=" * 80)
    
    # Check a consistent response
    consistent_response = "That's a great question! Recursion is when a function calls itself. Think of it like Russian nesting dolls - each doll contains a smaller version of itself."
    
    print("\nChecking consistent response:")
    print(f'Response: "{consistent_response}"')
    
    consistency = agent.check_consistency(consistent_response)
    print(f"\nConsistent: {consistency['consistent']}")
    print(f"Score: {consistency['score']}/10")
    
    # Check an inconsistent response
    inconsistent_response = "Recursion? Yeah bro, it's like totally when stuff repeats, ya know? Whatever dude."
    
    print("\n" + "-" * 60)
    print("\nChecking inconsistent response:")
    print(f'Response: "{inconsistent_response}"')
    
    consistency2 = agent.check_consistency(inconsistent_response)
    print(f"\nConsistent: {consistency2['consistent']}")
    print(f"Score: {consistency2['score']}/10")
    
    # Example 4: Multi-turn consistency
    print("\n" + "=" * 80)
    print("Example 4: Multi-Turn Consistency")
    print("=" * 80)
    
    agent2 = PersonaAgent(persona)
    
    conversation = [
        "Hi! Can you help me learn programming?",
        "What language should I start with?",
        "Is it hard to learn?"
    ]
    
    print("\nMulti-turn conversation:")
    for user_msg in conversation:
        context = agent2.get_conversation_context(num_recent=2) if agent2.interactions else None
        response = agent2.respond(user_msg, context=context)
        print(f"\nUser: {user_msg}")
        print(f"{persona.name}: {response}")
    
    # Example 5: Knowledge boundaries
    print("\n" + "=" * 80)
    print("Example 5: Respecting Knowledge Boundaries")
    print("=" * 80)
    
    agent3 = PersonaAgent(persona)
    
    # Question outside expertise
    question = "Can you diagnose this medical condition?"
    print(f"\nUser: {question}")
    
    response = agent3.respond(question)
    print(f"{persona.name}: {response}")
    print("\n(Notice how persona stays in character while declining gracefully)")
    
    # Example 6: Different persona
    print("\n" + "=" * 80)
    print("Example 6: Different Persona - Customer Service")
    print("=" * 80)
    
    service_persona = PersonaDefinition(
        name="Sarah",
        description="a helpful and patient customer service representative",
        core_traits=[
            PersonaTrait.FRIENDLY,
            PersonaTrait.EMPATHETIC,
            PersonaTrait.PRACTICAL
        ],
        communication_style=CommunicationStyle.PROFESSIONAL,
        values=[
            "Customer satisfaction",
            "Problem resolution",
            "Clear communication"
        ],
        knowledge_domains=[
            "Product information",
            "Company policies",
            "Troubleshooting"
        ],
        quirks=[
            "Always thanks customers",
            "Offers multiple solutions",
            "Confirms understanding"
        ],
        behavioral_rules=[
            "Stay positive even with frustrated customers",
            "Acknowledge customer feelings",
            "Provide specific next steps",
            "Never make promises you can't keep"
        ]
    )
    
    service_agent = PersonaAgent(service_persona, temperature=0.6)
    
    print(f"\nPersona: {service_persona.name}")
    print(f"Role: {service_persona.description}")
    
    customer_query = "My order is late and I'm very frustrated!"
    print(f"\nCustomer: {customer_query}")
    
    response = service_agent.respond(customer_query)
    print(f"{service_persona.name}: {response}")
    
    # Example 7: Persona with quirks
    print("\n" + "=" * 80)
    print("Example 7: Persona with Distinctive Quirks")
    print("=" * 80)
    
    quirky_persona = PersonaDefinition(
        name="Captain Morgan",
        description="a retired sea captain who now teaches navigation",
        core_traits=[
            PersonaTrait.FRIENDLY,
            PersonaTrait.PRACTICAL
        ],
        communication_style=CommunicationStyle.CASUAL,
        values=["Adventure", "Safety", "Experience"],
        knowledge_domains=["Navigation", "Sailing", "Maritime history"],
        quirks=[
            "Uses nautical metaphors",
            "Ends sentences with 'arr' or 'matey'",
            "References old sailing adventures",
            "Talks about 'the old days at sea'"
        ],
        sample_phrases=[
            "Arr, that reminds me of...",
            "Back in me sailing days...",
            "Steady as she goes, matey"
        ]
    )
    
    captain = PersonaAgent(quirky_persona, temperature=0.8)
    
    print(f"\nPersona: {quirky_persona.name}")
    print(f"Quirks: {quirky_persona.quirks}")
    
    question = "How do I find my way if I'm lost?"
    print(f"\nUser: {question}")
    
    response = captain.respond(question)
    print(f"{quirky_persona.name}: {response}")
    
    # Example 8: Consistency report
    print("\n" + "=" * 80)
    print("Example 8: Persona Consistency Report")
    print("=" * 80)
    
    # Generate several interactions
    agent4 = PersonaAgent(persona)
    
    test_questions = [
        "Explain algorithms",
        "What's your teaching philosophy?",
        "Tell me about data structures",
        "How do you make CS fun?"
    ]
    
    for q in test_questions:
        agent4.respond(q)
    
    report = agent4.get_consistency_report()
    
    print("\nCONSISTENCY REPORT:")
    print("=" * 60)
    print(f"Persona: {report['persona_name']}")
    print(f"Total Interactions: {report['total_interactions']}")
    print(f"Average Consistency: {report['average_consistency']:.2f}")
    
    if report['violations']:
        print(f"\nViolations Detected: {len(report['violations'])}")
    else:
        print("\nNo violations detected - persona maintained throughout!")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY: Persona Consistency Pattern")
    print("=" * 80)
    
    summary = """
    The Persona Consistency pattern demonstrated:
    
    1. PERSONA DEFINITION (Example 1):
       - Comprehensive character specification
       - Core personality traits
       - Communication style
       - Values and beliefs
       - Knowledge domains and boundaries
       - Unique quirks and phrases
       - Behavioral guidelines
    
    2. IN-CHARACTER RESPONSES (Example 2):
       - Maintaining persona voice
       - Consistent communication style
       - Knowledge-appropriate answers
       - Characteristic phrases
       - Behavioral consistency
    
    3. CONSISTENCY VALIDATION (Example 3):
       - Automated consistency checking
       - Scoring system (0-10)
       - Violation detection
       - LLM-based analysis
       - Quality assurance
    
    4. MULTI-TURN COHERENCE (Example 4):
       - Contextual awareness
       - Conversation continuity
       - Relationship consistency
       - Progressive interaction
       - Memory integration
    
    5. KNOWLEDGE BOUNDARIES (Example 5):
       - Respecting expertise limits
       - Graceful decline
       - In-character refusals
       - Honest limitations
       - Maintaining integrity
    
    6. ROLE SPECIALIZATION (Example 6):
       - Different persona type
       - Customer service character
       - Professional demeanor
       - Situation-appropriate
       - Distinct personality
    
    7. DISTINCTIVE QUIRKS (Example 7):
       - Unique characteristics
       - Memorable traits
       - Consistent quirks
       - Natural variation
       - Character depth
    
    8. PERFORMANCE TRACKING (Example 8):
       - Consistency metrics
       - Interaction logging
       - Violation tracking
       - Quality reporting
       - System monitoring
    
    KEY BENEFITS:
    ✓ Builds user trust and familiarity
    ✓ Creates memorable experiences
    ✓ Strengthens brand identity
    ✓ Enables emotional connection
    ✓ Provides predictable interactions
    ✓ Maintains character integrity
    ✓ Enhances believability
    ✓ Supports long-term relationships
    
    USE CASES:
    • Virtual assistants with brand personality
    • Game NPCs with defined characters
    • Customer service representatives
    • Educational tutors
    • Therapeutic/counseling chatbots
    • Celebrity or character impersonations
    • Brand ambassadors
    • Entertainment applications
    
    PERSONA ATTRIBUTES:
    → Core Traits: Personality characteristics
    → Communication Style: Tone and language
    → Values & Beliefs: Moral framework
    → Knowledge Domain: Expertise areas
    → Emotional Range: Expression patterns
    → Cultural Background: Context
    → Quirks & Habits: Unique features
    
    CONSISTENCY DIMENSIONS:
    • Linguistic: Word choice, grammar
    • Behavioral: Response patterns
    • Emotional: Mood stability
    • Knowledge: Expertise boundaries
    • Temporal: Consistency over time
    • Contextual: Appropriate adaptation
    • Relational: Relationship history
    
    BEST PRACTICES:
    1. Define persona comprehensively upfront
    2. Document sample phrases and behaviors
    3. Set clear knowledge boundaries
    4. Allow natural variation within constraints
    5. Monitor consistency metrics
    6. Version control persona changes
    7. Test across diverse scenarios
    8. Gather user feedback on believability
    
    TRADE-OFFS:
    • Consistency vs. flexibility
    • Character depth vs. simplicity
    • Authenticity vs. appropriateness
    • Memorability vs. subtlety
    
    PRODUCTION CONSIDERATIONS:
    → Store persona definitions in database
    → Version control all persona changes
    → A/B test different persona variations
    → Monitor consistency scores continuously
    → Log all persona violations for review
    → Enable graceful persona evolution
    → Support multiple personas per agent
    → Provide admin tools for persona management
    → Implement rollback for persona changes
    → Track user satisfaction by persona
    
    This pattern enables agents to maintain believable, consistent characters
    that users can trust, connect with, and remember across all interactions.
    """
    
    print(summary)


if __name__ == "__main__":
    demonstrate_persona_consistency()

    

Agentic Design Pattern: Persona Consistency

Pattern 131: Persona Consistency – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Persona Consistency pattern implementation.

---

Pattern Description

Pattern 131: Persona Consistency

Description: The Persona Consistency pattern enables agents to maintain a stable, coherent character or personality across all interactions. Rather than exhibiting inconsistent behavior or tone, the agent adheres to a defined persona with consistent traits, communication style, values, knowledge boundaries, and behavioral patterns throughout conversations and across sessions.

Persona consistency is essential for creating believable, trustworthy AI
characters in applications like virtual assistants, game NPCs, brand
representatives, and therapeutic chatbots. The pattern addresses challenges
like personality drift, contextual appropriateness, multi-turn coherence,
and balancing consistency with natural variation.

This pattern includes mechanisms for persona definition, trait enforcement,
style consistency, memory of character history, contextual adaptation within
persona bounds, and continuous validation of persona adherence.

Key Components: 1. Persona Definition: Core character traits and attributes 2. Style Guide: Communication patterns and language 3. Knowledge Boundaries: What the persona knows/doesn’t know 4. Behavioral Rules: How persona responds to situations 5. Memory System: Track persona history and consistency 6. Consistency Checker: Validate responses against persona 7. Adaptation Layer: Natural variation within constraints

Persona Attributes: 1. Core Traits: Personality characteristics (friendly, formal, humorous) 2. Communication Style: Tone, vocabulary, sentence structure 3. Values & Beliefs: Moral framework, priorities 4. Knowledge Domain: Areas of expertise and ignorance 5. Emotional Range: How persona expresses emotions 6. Cultural Background: Context influencing behavior 7. Quirks & Habits: Unique characteristics

Consistency Dimensions: 1. Linguistic: Word choice, grammar, dialect 2. Behavioral: Response patterns, decision-making 3. Emotional: Mood stability, reaction patterns 4. Knowledge: What persona knows consistently 5. Temporal: Consistency over time 6. Contextual: Appropriate adaptation 7. Relational: Consistent with relationship history

Use Cases: – Virtual assistants with brand personality – Game NPCs with defined characters – Customer service representatives – Educational tutors with consistent approach – Therapeutic/counseling chatbots – Celebrity or character impersonations – Brand ambassadors

Advantages: – Builds user trust and familiarity – Creates memorable experiences – Strengthens brand identity – Enables emotional connection – Provides predictable interactions – Maintains character integrity – Enhances believability

Challenges: – Balancing consistency with flexibility – Avoiding robotic repetition – Adapting to context while staying in character – Managing persona evolution – Handling edge cases in character – Maintaining consistency across sessions – Preventing personality drift

LangChain Implementation: This implementation uses LangChain for: – Persona-aware prompt construction – Response consistency validation – Character memory management – Style-constrained generation

Production Considerations: – Store persona definitions persistently – Version control persona changes – Monitor consistency metrics – Allow graceful persona updates – Support multiple personas per agent – Enable A/B testing of personas – Provide consistency reporting – Log persona violations for improvement

---

File Statistics

• Total Lines: 787
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 131: Persona Consistency

Description:
    The Persona Consistency pattern enables agents to maintain a stable, coherent
    character or personality across all interactions. Rather than exhibiting
    inconsistent behavior or tone, the agent adheres to a defined persona with
    consistent traits, communication style, values, knowledge boundaries, and
    behavioral patterns throughout conversations and across sessions.

    Persona consistency is essential for creating believable, trustworthy AI
    characters in applications like virtual assistants, game NPCs, brand
    representatives, and therapeutic chatbots. The pattern addresses challenges
    like personality drift, contextual appropriateness, multi-turn coherence,
    and balancing consistency with natural variation.

    This pattern includes mechanisms for persona definition, trait enforcement,
    style consistency, memory of character history, contextual adaptation within
    persona bounds, and continuous validation of persona adherence.

Key Components:
    1. Persona Definition: Core character traits and attributes
    2. Style Guide: Communication patterns and language
    3. Knowledge Boundaries: What the persona knows/doesn't know
    4. Behavioral Rules: How persona responds to situations
    5. Memory System: Track persona history and consistency
    6. Consistency Checker: Validate responses against persona
    7. Adaptation Layer: Natural variation within constraints

Persona Attributes:
    1. Core Traits: Personality characteristics (friendly, formal, humorous)
    2. Communication Style: Tone, vocabulary, sentence structure
    3. Values & Beliefs: Moral framework, priorities
    4. Knowledge Domain: Areas of expertise and ignorance
    5. Emotional Range: How persona expresses emotions
    6. Cultural Background: Context influencing behavior
    7. Quirks & Habits: Unique characteristics

Consistency Dimensions:
    1. Linguistic: Word choice, grammar, dialect
    2. Behavioral: Response patterns, decision-making
    3. Emotional: Mood stability, reaction patterns
    4. Knowledge: What persona knows consistently
    5. Temporal: Consistency over time
    6. Contextual: Appropriate adaptation
    7. Relational: Consistent with relationship history

Use Cases:
    - Virtual assistants with brand personality
    - Game NPCs with defined characters
    - Customer service representatives
    - Educational tutors with consistent approach
    - Therapeutic/counseling chatbots
    - Celebrity or character impersonations
    - Brand ambassadors

Advantages:
    - Builds user trust and familiarity
    - Creates memorable experiences
    - Strengthens brand identity
    - Enables emotional connection
    - Provides predictable interactions
    - Maintains character integrity
    - Enhances believability

Challenges:
    - Balancing consistency with flexibility
    - Avoiding robotic repetition
    - Adapting to context while staying in character
    - Managing persona evolution
    - Handling edge cases in character
    - Maintaining consistency across sessions
    - Preventing personality drift

LangChain Implementation:
    This implementation uses LangChain for:
    - Persona-aware prompt construction
    - Response consistency validation
    - Character memory management
    - Style-constrained generation

Production Considerations:
    - Store persona definitions persistently
    - Version control persona changes
    - Monitor consistency metrics
    - Allow graceful persona updates
    - Support multiple personas per agent
    - Enable A/B testing of personas
    - Provide consistency reporting
    - Log persona violations for improvement
"""

Purpose: This module implements the Persona Consistency 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 typing.Any
import typing.Dict
import typing.List
import typing.Optional

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 – 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)

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: PersonaTrait

Lines 107-116

Inherits from: Enum

Purpose:

Core personality traits.

---

Class: CommunicationStyle

Lines 119-126

Inherits from: Enum

Purpose:

Communication style options.

---

Class: PersonaDefinition

Lines 130-157

Purpose:

Complete definition of an agent's persona.

Attributes:
    name: Persona name
    description: Brief description
    core_traits: Personality traits
    communication_style: How persona communicates
    values: Core values and beliefs
    knowledge_domains: Areas of expertise
    knowledge_gaps: What persona doesn't know
    emotional_range: How emotions are expressed
    quirks: Unique characteristics
    sample_phrases: Example phrases persona uses
    behavioral_rules: Guidelines for behavior

---

Class: PersonaInteraction

Lines 161-178

Purpose:

Record of an interaction with persona.

Attributes:
    interaction_id: Unique identifier
    user_input: User's message
    persona_response: Agent's response
    timestamp: When interaction occurred
    consistency_score: How well response matched persona
    violations: Any detected inconsistencies

---

Class: PersonaAgent

Lines 181-382

Purpose:

Agent that maintains consistent persona across interactions.

This agent ensures all responses adhere to a defined persona,
maintaining character consistency across conversations.

Methods:

__init__(self, persona, temperature)

Lines 189-200

Purpose:

Initialize persona agent.

Args:
    persona: Persona definition
    temperature: LLM temperature (higher for more variation)

Parameters: – persona: Persona – temperature: Sampling temperature (0-1)

_build_persona_prompt(self)

Lines 202-232

Purpose:

Build persona description for prompts.

Parameters:

Returns: str

respond(self, user_input, context)

Lines 234-283

Purpose:

Generate response in character.

Args:
    user_input: User's message
    context: Optional conversation context

Returns:
    Persona's response

Parameters: – user_input: Input data – context: Context

Returns: str

check_consistency(self, response)

Lines 285-340

Purpose:

Check if response is consistent with persona.

Args:
    response: Response to check

Returns:
    Consistency analysis

Parameters: – response: Response

Returns: <ast.Subscript object at 0x000001A06D676F90>

get_conversation_context(self, num_recent)

Lines 342-359

Purpose:

Get recent conversation history.

Args:
    num_recent: Number of recent interactions

Returns:
    Formatted context

Parameters: – num_recent: Num recent

Returns: str

get_consistency_report(self)

Lines 361-382

Purpose:

Get report on persona consistency.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D622010>

---

Functions

demonstrate_persona_consistency()

Lines 385-782

Purpose:

Demonstrate persona consistency pattern.

---

Key Concepts Summary

Pattern Overview

The Persona Consistency 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 130
• Pattern 132

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 Persona Consistency 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: 131_persona_consistency.py
Total Lines: 787
Pattern Category: AI Agent Design Pattern #131

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 130: Proactive Engagement

Description:
    The Proactive Engagement pattern enables agents to anticipate user needs and
    initiate interactions rather than passively waiting for user input. Instead of
    purely reactive behavior, proactive agents monitor context, identify opportunities
    for assistance, predict upcoming needs, and engage users at appropriate moments
    to provide value before being explicitly asked.
    
    Proactive engagement is essential for creating helpful, attentive AI assistants
    that feel intelligent and considerate. The pattern addresses challenges like
    determining when to interrupt, avoiding being annoying, predicting actual needs
    vs. false positives, timing interventions appropriately, and personalizing
    proactivity levels to user preferences.
    
    This pattern includes mechanisms for need prediction, opportunity detection,
    engagement timing, interruption management, value assessment, personalization,
    and learning from user responses to proactive suggestions.

Key Components:
    1. Context Monitor: Tracks user activity and environment
    2. Need Predictor: Anticipates upcoming needs
    3. Opportunity Detector: Identifies engagement moments
    4. Timing Manager: Determines when to engage
    5. Value Assessor: Evaluates potential intervention value
    6. Personalization Engine: Adapts to user preferences
    7. Feedback Learner: Learns from user responses

Engagement Triggers:
    1. Predicted Needs: Anticipate what user will need
    2. Detected Problems: Identify issues proactively
    3. Opportunities: Suggest improvements or shortcuts
    4. Reminders: Timely notifications
    5. Contextual Tips: Relevant guidance
    6. Status Updates: Important changes
    7. Recommendations: Personalized suggestions

Timing Strategies:
    1. Immediate: Critical or time-sensitive
    2. Opportune Moment: Natural break points
    3. Scheduled: Predetermined times
    4. Threshold-Based: When confidence is high
    5. User-Initiated Flow: During interactions
    6. Idle Detection: When user is inactive

Use Cases:
    - Smart assistants (Alexa, Siri, Google Assistant)
    - Customer service chatbots
    - Healthcare monitoring systems
    - Educational tutors
    - Personal productivity assistants
    - Smart home automation
    - Enterprise knowledge management

Advantages:
    - Improves user experience through anticipation
    - Reduces cognitive load on users
    - Catches issues before they escalate
    - Provides timely, relevant assistance
    - Increases perceived intelligence
    - Builds trust through helpfulness
    - Enhances productivity

Challenges:
    - Avoiding annoying interruptions
    - False positive predictions
    - Privacy and monitoring concerns
    - Determining appropriate timing
    - Balancing proactivity with passivity
    - Personalizing to individual preferences
    - Managing user attention

LangChain Implementation:
    This implementation uses LangChain for:
    - Context analysis and need prediction
    - Opportunity identification
    - Value assessment of interventions
    - Personalized engagement strategies
    
Production Considerations:
    - Implement user preference controls
    - Track engagement acceptance rates
    - Use machine learning for need prediction
    - Respect "do not disturb" modes
    - Provide easy dismissal options
    - Learn from user feedback
    - Monitor intervention frequency
    - Enable granular notification settings
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
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 EngagementType(Enum):
    """Types of proactive engagements."""
    PREDICTED_NEED = "predicted_need"
    PROBLEM_DETECTION = "problem_detection"
    OPPORTUNITY = "opportunity"
    REMINDER = "reminder"
    TIP = "tip"
    STATUS_UPDATE = "status_update"
    RECOMMENDATION = "recommendation"


class EngagementPriority(Enum):
    """Priority levels for engagements."""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


class UserResponse(Enum):
    """User responses to proactive engagements."""
    ACCEPTED = "accepted"
    DISMISSED = "dismissed"
    DEFERRED = "deferred"
    IGNORED = "ignored"


@dataclass
class ContextSignal:
    """
    A signal from the user context.
    
    Attributes:
        signal_type: Type of signal
        content: Signal content
        timestamp: When detected
        confidence: Detection confidence
        metadata: Additional information
    """
    signal_type: str
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    confidence: float = 1.0
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class ProactiveEngagement:
    """
    A proactive engagement with the user.
    
    Attributes:
        engagement_id: Unique identifier
        engagement_type: Type of engagement
        priority: Urgency level
        message: Engagement message
        suggested_action: Recommended action
        value_score: Estimated value to user
        timing_score: Appropriateness of timing
        triggers: What triggered this
        created_at: When created
        engaged_at: When delivered to user
        response: User response
    """
    engagement_id: str
    engagement_type: EngagementType
    priority: EngagementPriority
    message: str
    suggested_action: Optional[str] = None
    value_score: float = 0.5
    timing_score: float = 0.5
    triggers: List[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.now)
    engaged_at: Optional[datetime] = None
    response: Optional[UserResponse] = None


@dataclass
class UserPreferences:
    """
    User preferences for proactive engagement.
    
    Attributes:
        enabled: Whether proactive engagement is on
        min_priority: Minimum priority level to show
        quiet_hours: Hours when engagement is disabled
        max_frequency: Maximum engagements per hour
        preferred_types: Preferred engagement types
    """
    enabled: bool = True
    min_priority: EngagementPriority = EngagementPriority.MEDIUM
    quiet_hours: List[int] = field(default_factory=list)
    max_frequency: int = 5
    preferred_types: List[EngagementType] = field(default_factory=list)


class ProactiveAgent:
    """
    Proactive engagement agent that anticipates needs and initiates interactions.
    
    This agent monitors context, predicts needs, and proactively engages
    users at appropriate moments with helpful suggestions.
    """
    
    def __init__(self, temperature: float = 0.3):
        """
        Initialize proactive agent.
        
        Args:
            temperature: LLM temperature
        """
        self.llm = ChatOpenAI(temperature=temperature, model="gpt-3.5-turbo")
        self.context_signals: List[ContextSignal] = []
        self.engagements: List[ProactiveEngagement] = []
        self.user_preferences = UserPreferences()
        self.engagement_counter = 0
    
    def add_context_signal(
        self,
        signal_type: str,
        content: str,
        confidence: float = 1.0,
        metadata: Optional[Dict[str, Any]] = None
    ) -> ContextSignal:
        """
        Add a context signal from monitoring.
        
        Args:
            signal_type: Type of signal
            content: Signal content
            confidence: Detection confidence
            metadata: Additional information
            
        Returns:
            Created signal
        """
        signal = ContextSignal(
            signal_type=signal_type,
            content=content,
            confidence=confidence,
            metadata=metadata or {}
        )
        self.context_signals.append(signal)
        return signal
    
    def predict_need(self, context_window: int = 5) -> Optional[str]:
        """
        Predict user need based on recent context.
        
        Args:
            context_window: Number of recent signals to consider
            
        Returns:
            Predicted need description
        """
        recent_signals = self.context_signals[-context_window:]
        
        if not recent_signals:
            return None
        
        # Build context
        context = "\n".join([
            f"{s.signal_type}: {s.content}"
            for s in recent_signals
        ])
        
        prompt = ChatPromptTemplate.from_template(
            "Based on these recent user activities and context signals:\n\n"
            "{context}\n\n"
            "Predict what the user might need help with next. "
            "Consider patterns, incomplete tasks, and logical next steps.\n\n"
            "Provide:\n"
            "1. Predicted need (one sentence)\n"
            "2. Confidence (0-1)\n"
            "3. Brief reasoning"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        prediction = chain.invoke({"context": context})
        
        return prediction
    
    def detect_opportunity(self, context: str) -> Optional[Dict[str, Any]]:
        """
        Detect opportunity for proactive engagement.
        
        Args:
            context: Current context description
            
        Returns:
            Opportunity details if found
        """
        prompt = ChatPromptTemplate.from_template(
            "Analyze this user context:\n\n"
            "{context}\n\n"
            "Is there an opportunity for helpful proactive engagement?\n"
            "Consider:\n"
            "- Shortcuts or efficiency improvements\n"
            "- Potential problems to avoid\n"
            "- Helpful tips or information\n"
            "- Relevant recommendations\n\n"
            "Respond with:\n"
            "OPPORTUNITY: yes/no\n"
            "TYPE: [predicted_need/problem_detection/opportunity/tip/recommendation]\n"
            "MESSAGE: [brief helpful message]\n"
            "VALUE: [0-1 score]"
        )
        
        chain = prompt | self.llm | StrOutputParser()
        result = chain.invoke({"context": context})
        
        # Parse result
        lines = result.strip().split('\n')
        opportunity_dict = {}
        
        for line in lines:
            if ':' in line:
                key, value = line.split(':', 1)
                opportunity_dict[key.strip().lower()] = value.strip()
        
        if opportunity_dict.get('opportunity', '').lower() == 'yes':
            return opportunity_dict
        
        return None
    
    def assess_timing(self, engagement_type: EngagementType) -> float:
        """
        Assess if timing is appropriate for engagement.
        
        Args:
            engagement_type: Type of engagement
            
        Returns:
            Timing appropriateness score (0-1)
        """
        current_hour = datetime.now().hour
        
        # Check quiet hours
        if current_hour in self.user_preferences.quiet_hours:
            return 0.1
        
        # Check recent engagement frequency
        recent_hour = datetime.now() - timedelta(hours=1)
        recent_engagements = [
            e for e in self.engagements
            if e.engaged_at and e.engaged_at > recent_hour
        ]
        
        if len(recent_engagements) >= self.user_preferences.max_frequency:
            return 0.2
        
        # Critical engagements always have good timing
        # (override frequency limits for critical items)
        
        # Higher score for fewer recent engagements
        frequency_score = 1.0 - (len(recent_engagements) / self.user_preferences.max_frequency)
        
        return max(0.3, frequency_score)
    
    def create_engagement(
        self,
        engagement_type: EngagementType,
        message: str,
        priority: EngagementPriority = EngagementPriority.MEDIUM,
        suggested_action: Optional[str] = None,
        triggers: Optional[List[str]] = None
    ) -> ProactiveEngagement:
        """
        Create a proactive engagement.
        
        Args:
            engagement_type: Type of engagement
            message: Engagement message
            priority: Priority level
            suggested_action: Suggested action
            triggers: What triggered this
            
        Returns:
            Created engagement
        """
        self.engagement_counter += 1
        
        # Assess value and timing
        timing_score = self.assess_timing(engagement_type)
        
        # Simple value heuristic (could be ML model in production)
        value_score = {
            EngagementPriority.CRITICAL: 0.9,
            EngagementPriority.HIGH: 0.7,
            EngagementPriority.MEDIUM: 0.5,
            EngagementPriority.LOW: 0.3
        }.get(priority, 0.5)
        
        engagement = ProactiveEngagement(
            engagement_id=f"eng_{self.engagement_counter}",
            engagement_type=engagement_type,
            priority=priority,
            message=message,
            suggested_action=suggested_action,
            value_score=value_score,
            timing_score=timing_score,
            triggers=triggers or []
        )
        
        self.engagements.append(engagement)
        return engagement
    
    def should_engage(self, engagement: ProactiveEngagement) -> bool:
        """
        Determine if agent should deliver engagement.
        
        Args:
            engagement: Engagement to evaluate
            
        Returns:
            Whether to engage
        """
        # Check if proactive engagement is enabled
        if not self.user_preferences.enabled:
            return False
        
        # Check priority threshold
        priority_levels = {
            EngagementPriority.CRITICAL: 4,
            EngagementPriority.HIGH: 3,
            EngagementPriority.MEDIUM: 2,
            EngagementPriority.LOW: 1
        }
        
        if priority_levels.get(engagement.priority, 0) < \
           priority_levels.get(self.user_preferences.min_priority, 2):
            return False
        
        # Check preferred types (if any specified)
        if self.user_preferences.preferred_types and \
           engagement.engagement_type not in self.user_preferences.preferred_types:
            return False
        
        # Check timing score
        if engagement.timing_score  0.5
    
    def engage_user(self, engagement: ProactiveEngagement) -> bool:
        """
        Deliver engagement to user.
        
        Args:
            engagement: Engagement to deliver
            
        Returns:
            Whether engagement was delivered
        """
        if not self.should_engage(engagement):
            return False
        
        engagement.engaged_at = datetime.now()
        return True
    
    def record_response(
        self,
        engagement_id: str,
        response: UserResponse
    ):
        """
        Record user response to engagement.
        
        Args:
            engagement_id: Engagement identifier
            response: User response
        """
        for engagement in self.engagements:
            if engagement.engagement_id == engagement_id:
                engagement.response = response
                break
    
    def get_acceptance_rate(self) -> float:
        """Get acceptance rate for delivered engagements."""
        delivered = [e for e in self.engagements if e.engaged_at]
        
        if not delivered:
            return 0.0
        
        accepted = [e for e in delivered if e.response == UserResponse.ACCEPTED]
        return len(accepted) / len(delivered)
    
    def get_engagement_summary(self) -> Dict[str, Any]:
        """Get summary of engagement activity."""
        delivered = [e for e in self.engagements if e.engaged_at]
        
        return {
            "total_created": len(self.engagements),
            "total_delivered": len(delivered),
            "delivery_rate": len(delivered) / len(self.engagements) if self.engagements else 0,
            "acceptance_rate": self.get_acceptance_rate(),
            "by_type": {
                et.value: len([e for e in delivered if e.engagement_type == et])
                for et in EngagementType
            },
            "by_priority": {
                pr.value: len([e for e in delivered if e.priority == pr])
                for pr in EngagementPriority
            }
        }


def demonstrate_proactive_engagement():
    """Demonstrate proactive engagement pattern."""
    
    print("=" * 80)
    print("PROACTIVE ENGAGEMENT PATTERN DEMONSTRATION")
    print("=" * 80)
    
    # Example 1: Need prediction
    print("\n" + "=" * 80)
    print("Example 1: Predicting User Needs")
    print("=" * 80)
    
    agent = ProactiveAgent()
    
    # Add context signals
    agent.add_context_signal("user_action", "Opened email application")
    agent.add_context_signal("user_action", "Viewed calendar")
    agent.add_context_signal("user_action", "Searched for 'meeting notes template'")
    agent.add_context_signal("time_context", "10 minutes before scheduled meeting")
    
    print("\nContext signals collected:")
    for signal in agent.context_signals:
        print(f"  • {signal.signal_type}: {signal.content}")
    
    print("\nPredicting user need...")
    prediction = agent.predict_need()
    print(f"\nPrediction:\n{prediction}")
    
    # Example 2: Opportunity detection
    print("\n" + "=" * 80)
    print("Example 2: Detecting Engagement Opportunities")
    print("=" * 80)
    
    agent2 = ProactiveAgent()
    
    context = """
    User is writing a report in Word.
    Document has multiple spelling errors highlighted.
    User has been working for 2 hours without saving.
    Battery level: 15%
    """
    
    print(f"\nContext:\n{context}")
    print("\nDetecting opportunities...")
    
    opportunity = agent2.detect_opportunity(context)
    
    if opportunity:
        print("\nOpportunity detected!")
        print(f"  Type: {opportunity.get('type', 'N/A')}")
        print(f"  Message: {opportunity.get('message', 'N/A')}")
        print(f"  Value: {opportunity.get('value', 'N/A')}")
    
    # Example 3: Creating engagements
    print("\n" + "=" * 80)
    print("Example 3: Creating Proactive Engagements")
    print("=" * 80)
    
    agent3 = ProactiveAgent()
    
    # Create various engagements
    eng1 = agent3.create_engagement(
        EngagementType.REMINDER,
        "Your meeting starts in 5 minutes",
        priority=EngagementPriority.HIGH,
        suggested_action="Join video call",
        triggers=["calendar_event", "time_proximity"]
    )
    
    eng2 = agent3.create_engagement(
        EngagementType.TIP,
        "You can use Ctrl+S to save your document",
        priority=EngagementPriority.LOW,
        triggers=["unsaved_document", "long_edit_session"]
    )
    
    eng3 = agent3.create_engagement(
        EngagementType.PROBLEM_DETECTION,
        "Low battery detected. Please plug in your charger.",
        priority=EngagementPriority.CRITICAL,
        suggested_action="Connect charger",
        triggers=["battery_low"]
    )
    
    print("\nCreated engagements:")
    for eng in agent3.engagements:
        print(f"\n  {eng.engagement_id}:")
        print(f"    Type: {eng.engagement_type.value}")
        print(f"    Priority: {eng.priority.value}")
        print(f"    Message: {eng.message}")
        print(f"    Value Score: {eng.value_score:.2f}")
        print(f"    Timing Score: {eng.timing_score:.2f}")
    
    # Example 4: Timing assessment
    print("\n" + "=" * 80)
    print("Example 4: Timing Assessment")
    print("=" * 80)
    
    agent4 = ProactiveAgent()
    
    # Set quiet hours
    agent4.user_preferences.quiet_hours = [22, 23, 0, 1, 2, 3, 4, 5, 6]
    agent4.user_preferences.max_frequency = 3
    
    print("\nUser preferences:")
    print(f"  Quiet hours: {agent4.user_preferences.quiet_hours}")
    print(f"  Max frequency: {agent4.user_preferences.max_frequency}/hour")
    
    # Create and assess engagement
    eng = agent4.create_engagement(
        EngagementType.RECOMMENDATION,
        "Check out this relevant article",
        priority=EngagementPriority.MEDIUM
    )
    
    print(f"\nEngagement timing score: {eng.timing_score:.2f}")
    print(f"Should engage: {agent4.should_engage(eng)}")
    
    # Example 5: Delivery decision
    print("\n" + "=" * 80)
    print("Example 5: Engagement Delivery Decision")
    print("=" * 80)
    
    agent5 = ProactiveAgent()
    
    # Create different priority engagements
    engagements = [
        agent5.create_engagement(
            EngagementType.TIP,
            "Low priority tip",
            priority=EngagementPriority.LOW
        ),
        agent5.create_engagement(
            EngagementType.REMINDER,
            "Medium priority reminder",
            priority=EngagementPriority.MEDIUM
        ),
        agent5.create_engagement(
            EngagementType.PROBLEM_DETECTION,
            "Critical issue detected",
            priority=EngagementPriority.CRITICAL
        )
    ]
    
    print("\nDelivery decisions:")
    for eng in engagements:
        delivered = agent5.engage_user(eng)
        status = "✓ DELIVERED" if delivered else "✗ HELD"
        print(f"  {eng.priority.value}: {status}")
    
    # Example 6: User feedback learning
    print("\n" + "=" * 80)
    print("Example 6: Learning from User Feedback")
    print("=" * 80)
    
    agent6 = ProactiveAgent()
    
    # Create and deliver engagements
    for i in range(5):
        eng = agent6.create_engagement(
            EngagementType.TIP,
            f"Helpful tip #{i+1}",
            priority=EngagementPriority.MEDIUM
        )
        agent6.engage_user(eng)
    
    # Record user responses
    agent6.record_response(agent6.engagements[0].engagement_id, UserResponse.ACCEPTED)
    agent6.record_response(agent6.engagements[1].engagement_id, UserResponse.DISMISSED)
    agent6.record_response(agent6.engagements[2].engagement_id, UserResponse.ACCEPTED)
    agent6.record_response(agent6.engagements[3].engagement_id, UserResponse.ACCEPTED)
    agent6.record_response(agent6.engagements[4].engagement_id, UserResponse.DISMISSED)
    
    print("\nUser responses recorded:")
    for eng in agent6.engagements:
        if eng.response:
            print(f"  {eng.engagement_id}: {eng.response.value}")
    
    acceptance_rate = agent6.get_acceptance_rate()
    print(f"\nAcceptance rate: {acceptance_rate:.0%}")
    
    # Example 7: Preference management
    print("\n" + "=" * 80)
    print("Example 7: User Preference Management")
    print("=" * 80)
    
    agent7 = ProactiveAgent()
    
    # User configures preferences
    agent7.user_preferences.enabled = True
    agent7.user_preferences.min_priority = EngagementPriority.HIGH
    agent7.user_preferences.preferred_types = [
        EngagementType.REMINDER,
        EngagementType.PROBLEM_DETECTION
    ]
    
    print("\nUser preferences:")
    print(f"  Enabled: {agent7.user_preferences.enabled}")
    print(f"  Min priority: {agent7.user_preferences.min_priority.value}")
    print(f"  Preferred types: {[t.value for t in agent7.user_preferences.preferred_types]}")
    
    # Test different engagements
    test_engagements = [
        ("TIP (LOW)", EngagementType.TIP, EngagementPriority.LOW),
        ("REMINDER (HIGH)", EngagementType.REMINDER, EngagementPriority.HIGH),
        ("OPPORTUNITY (HIGH)", EngagementType.OPPORTUNITY, EngagementPriority.HIGH),
        ("PROBLEM (CRITICAL)", EngagementType.PROBLEM_DETECTION, EngagementPriority.CRITICAL)
    ]
    
    print("\nEngagement filtering:")
    for name, eng_type, priority in test_engagements:
        eng = agent7.create_engagement(eng_type, f"Test message", priority=priority)
        would_deliver = agent7.should_engage(eng)
        status = "✓ Would deliver" if would_deliver else "✗ Would filter"
        print(f"  {name}: {status}")
    
    # Example 8: System summary
    print("\n" + "=" * 80)
    print("Example 8: Engagement System Summary")
    print("=" * 80)
    
    agent8 = ProactiveAgent()
    
    # Simulate various engagements
    for i in range(10):
        eng_type = [EngagementType.TIP, EngagementType.REMINDER, EngagementType.RECOMMENDATION][i % 3]
        priority = [EngagementPriority.LOW, EngagementPriority.MEDIUM, EngagementPriority.HIGH][i % 3]
        
        eng = agent8.create_engagement(eng_type, f"Message {i+1}", priority=priority)
        
        if agent8.engage_user(eng):
            # Simulate responses
            response = UserResponse.ACCEPTED if i % 2 == 0 else UserResponse.DISMISSED
            agent8.record_response(eng.engagement_id, response)
    
    summary = agent8.get_engagement_summary()
    
    print("\nENGAGEMENT SUMMARY:")
    print("=" * 60)
    print(f"Total Created: {summary['total_created']}")
    print(f"Total Delivered: {summary['total_delivered']}")
    print(f"Delivery Rate: {summary['delivery_rate']:.0%}")
    print(f"Acceptance Rate: {summary['acceptance_rate']:.0%}")
    
    print("\nBy Type:")
    for eng_type, count in summary['by_type'].items():
        if count > 0:
            print(f"  {eng_type}: {count}")
    
    print("\nBy Priority:")
    for priority, count in summary['by_priority'].items():
        if count > 0:
            print(f"  {priority}: {count}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY: Proactive Engagement Pattern")
    print("=" * 80)
    
    summary_text = """
    The Proactive Engagement pattern demonstrated:
    
    1. NEED PREDICTION (Example 1):
       - Context signal collection
       - Pattern analysis
       - LLM-based prediction
       - Confidence estimation
       - Anticipatory behavior
    
    2. OPPORTUNITY DETECTION (Example 2):
       - Context analysis
       - Opportunity identification
       - Value assessment
       - Type classification
       - Actionable suggestions
    
    3. ENGAGEMENT CREATION (Example 3):
       - Multiple engagement types
       - Priority assignment
       - Value scoring
       - Timing assessment
       - Trigger tracking
    
    4. TIMING ASSESSMENT (Example 4):
       - Quiet hour respect
       - Frequency limiting
       - Context awareness
       - Appropriateness scoring
       - User preference consideration
    
    5. DELIVERY DECISIONS (Example 5):
       - Priority-based filtering
       - Threshold evaluation
       - Critical override
       - Smart gatekeeping
       - Controlled delivery
    
    6. FEEDBACK LEARNING (Example 6):
       - Response tracking
       - Acceptance rate calculation
       - Pattern learning
       - Continuous improvement
       - Adaptive behavior
    
    7. PREFERENCE MANAGEMENT (Example 7):
       - User control
       - Type filtering
       - Priority thresholds
       - Personalization
       - Preference respect
    
    8. SYSTEM INTEGRATION (Example 8):
       - Comprehensive metrics
       - Type distribution
       - Priority analysis
       - Performance tracking
       - System overview
    
    KEY BENEFITS:
    ✓ Improves UX through anticipation
    ✓ Reduces cognitive load on users
    ✓ Catches issues before they escalate
    ✓ Provides timely, relevant assistance
    ✓ Increases perceived intelligence
    ✓ Builds trust through helpfulness
    ✓ Enhances productivity
    ✓ Creates delightful experiences
    
    USE CASES:
    • Smart assistants (Alexa, Siri)
    • Customer service chatbots
    • Healthcare monitoring
    • Educational tutors
    • Productivity assistants
    • Smart home automation
    • Enterprise knowledge management
    • Preventive maintenance
    
    ENGAGEMENT TYPES:
    → Predicted Needs: Anticipate requirements
    → Problem Detection: Identify issues
    → Opportunities: Suggest improvements
    → Reminders: Timely notifications
    → Tips: Contextual guidance
    → Status Updates: Important changes
    → Recommendations: Personalized suggestions
    
    TIMING STRATEGIES:
    • Immediate: Critical/time-sensitive
    • Opportune Moment: Natural breaks
    • Scheduled: Predetermined times
    • Threshold-Based: High confidence
    • User-Initiated Flow: During interactions
    • Idle Detection: When inactive
    
    BEST PRACTICES:
    1. Implement user preference controls
    2. Track engagement acceptance rates
    3. Use ML for need prediction
    4. Respect "do not disturb" modes
    5. Provide easy dismissal options
    6. Learn from user feedback
    7. Monitor intervention frequency
    8. Enable granular notification settings
    
    TRADE-OFFS:
    • Helpfulness vs. annoyance
    • Proactivity vs. privacy
    • Frequency vs. relevance
    • Automation vs. control
    
    PRODUCTION CONSIDERATIONS:
    → Implement A/B testing for engagement strategies
    → Use ML models for need prediction
    → Track detailed engagement metrics
    → Provide fine-grained user controls
    → Respect platform notification guidelines
    → Implement gradual rollout for new features
    → Monitor user satisfaction scores
    → Enable easy opt-out mechanisms
    → Log all engagement decisions
    → Personalize based on user history
    
    This pattern enables agents to be helpful and attentive by anticipating
    user needs and engaging proactively at appropriate moments, creating a
    more intelligent and considerate user experience.
    """
    
    print(summary_text)


if __name__ == "__main__":
    demonstrate_proactive_engagement()

    

Agentic Design Pattern: Proactive Engagement

Pattern 130: Proactive Engagement – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Proactive Engagement pattern implementation.

---

Pattern Description

Pattern 130: Proactive Engagement

Description: The Proactive Engagement pattern enables agents to anticipate user needs and initiate interactions rather than passively waiting for user input. Instead of purely reactive behavior, proactive agents monitor context, identify opportunities for assistance, predict upcoming needs, and engage users at appropriate moments to provide value before being explicitly asked.

Proactive engagement is essential for creating helpful, attentive AI assistants
that feel intelligent and considerate. The pattern addresses challenges like
determining when to interrupt, avoiding being annoying, predicting actual needs
vs. false positives, timing interventions appropriately, and personalizing
proactivity levels to user preferences.

This pattern includes mechanisms for need prediction, opportunity detection,
engagement timing, interruption management, value assessment, personalization,
and learning from user responses to proactive suggestions.

Key Components: 1. Context Monitor: Tracks user activity and environment 2. Need Predictor: Anticipates upcoming needs 3. Opportunity Detector: Identifies engagement moments 4. Timing Manager: Determines when to engage 5. Value Assessor: Evaluates potential intervention value 6. Personalization Engine: Adapts to user preferences 7. Feedback Learner: Learns from user responses

Engagement Triggers: 1. Predicted Needs: Anticipate what user will need 2. Detected Problems: Identify issues proactively 3. Opportunities: Suggest improvements or shortcuts 4. Reminders: Timely notifications 5. Contextual Tips: Relevant guidance 6. Status Updates: Important changes 7. Recommendations: Personalized suggestions

Timing Strategies: 1. Immediate: Critical or time-sensitive 2. Opportune Moment: Natural break points 3. Scheduled: Predetermined times 4. Threshold-Based: When confidence is high 5. User-Initiated Flow: During interactions 6. Idle Detection: When user is inactive

Use Cases: – Smart assistants (Alexa, Siri, Google Assistant) – Customer service chatbots – Healthcare monitoring systems – Educational tutors – Personal productivity assistants – Smart home automation – Enterprise knowledge management

Advantages: – Improves user experience through anticipation – Reduces cognitive load on users – Catches issues before they escalate – Provides timely, relevant assistance – Increases perceived intelligence – Builds trust through helpfulness – Enhances productivity

Challenges: – Avoiding annoying interruptions – False positive predictions – Privacy and monitoring concerns – Determining appropriate timing – Balancing proactivity with passivity – Personalizing to individual preferences – Managing user attention

LangChain Implementation: This implementation uses LangChain for: – Context analysis and need prediction – Opportunity identification – Value assessment of interventions – Personalized engagement strategies

Production Considerations: – Implement user preference controls – Track engagement acceptance rates – Use machine learning for need prediction – Respect “do not disturb” modes – Provide easy dismissal options – Learn from user feedback – Monitor intervention frequency – Enable granular notification settings

---

File Statistics

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

---

Module Documentation

"""
Pattern 130: Proactive Engagement

Description:
    The Proactive Engagement pattern enables agents to anticipate user needs and
    initiate interactions rather than passively waiting for user input. Instead of
    purely reactive behavior, proactive agents monitor context, identify opportunities
    for assistance, predict upcoming needs, and engage users at appropriate moments
    to provide value before being explicitly asked.

    Proactive engagement is essential for creating helpful, attentive AI assistants
    that feel intelligent and considerate. The pattern addresses challenges like
    determining when to interrupt, avoiding being annoying, predicting actual needs
    vs. false positives, timing interventions appropriately, and personalizing
    proactivity levels to user preferences.

    This pattern includes mechanisms for need prediction, opportunity detection,
    engagement timing, interruption management, value assessment, personalization,
    and learning from user responses to proactive suggestions.

Key Components:
    1. Context Monitor: Tracks user activity and environment
    2. Need Predictor: Anticipates upcoming needs
    3. Opportunity Detector: Identifies engagement moments
    4. Timing Manager: Determines when to engage
    5. Value Assessor: Evaluates potential intervention value
    6. Personalization Engine: Adapts to user preferences
    7. Feedback Learner: Learns from user responses

Engagement Triggers:
    1. Predicted Needs: Anticipate what user will need
    2. Detected Problems: Identify issues proactively
    3. Opportunities: Suggest improvements or shortcuts
    4. Reminders: Timely notifications
    5. Contextual Tips: Relevant guidance
    6. Status Updates: Important changes
    7. Recommendations: Personalized suggestions

Timing Strategies:
    1. Immediate: Critical or time-sensitive
    2. Opportune Moment: Natural break points
    3. Scheduled: Predetermined times
    4. Threshold-Based: When confidence is high
    5. User-Initiated Flow: During interactions
    6. Idle Detection: When user is inactive

Use Cases:
    - Smart assistants (Alexa, Siri, Google Assistant)
    - Customer service chatbots
    - Healthcare monitoring systems
    - Educational tutors
    - Personal productivity assistants
    - Smart home automation
    - Enterprise knowledge management

Advantages:
    - Improves user experience through anticipation
    - Reduces cognitive load on users
    - Catches issues before they escalate
    - Provides timely, relevant assistance
    - Increases perceived intelligence
    - Builds trust through helpfulness
    - Enhances productivity

Challenges:
    - Avoiding annoying interruptions
    - False positive predictions
    - Privacy and monitoring concerns
    - Determining appropriate timing
    - Balancing proactivity with passivity
    - Personalizing to individual preferences
    - Managing user attention

LangChain Implementation:
    This implementation uses LangChain for:
    - Context analysis and need prediction
    - Opportunity identification
    - Value assessment of interventions
    - Personalized engagement strategies

Production Considerations:
    - Implement user preference controls
    - Track engagement acceptance rates
    - Use machine learning for need prediction
    - Respect "do not disturb" modes
    - Provide easy dismissal options
    - Learn from user feedback
    - Monitor intervention frequency
    - Enable granular notification settings
"""

Purpose: This module implements the Proactive Engagement 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 datetime.timedelta
import enum.Enum
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
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 – datetime.timedelta: 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 – 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: EngagementType

Lines 106-114

Inherits from: Enum

Purpose:

Types of proactive engagements.

---

Class: EngagementPriority

Lines 117-122

Inherits from: Enum

Purpose:

Priority levels for engagements.

---

Class: UserResponse

Lines 125-130

Inherits from: Enum

Purpose:

User responses to proactive engagements.

---

Class: ContextSignal

Lines 134-149

Purpose:

A signal from the user context.

Attributes:
    signal_type: Type of signal
    content: Signal content
    timestamp: When detected
    confidence: Detection confidence
    metadata: Additional information

---

Class: ProactiveEngagement

Lines 153-180

Purpose:

A proactive engagement with the user.

Attributes:
    engagement_id: Unique identifier
    engagement_type: Type of engagement
    priority: Urgency level
    message: Engagement message
    suggested_action: Recommended action
    value_score: Estimated value to user
    timing_score: Appropriateness of timing
    triggers: What triggered this
    created_at: When created
    engaged_at: When delivered to user
    response: User response

---

Class: UserPreferences

Lines 184-199

Purpose:

User preferences for proactive engagement.

Attributes:
    enabled: Whether proactive engagement is on
    min_priority: Minimum priority level to show
    quiet_hours: Hours when engagement is disabled
    max_frequency: Maximum engagements per hour
    preferred_types: Preferred engagement types

---

Class: ProactiveAgent

Lines 202-517

Purpose:

Proactive engagement agent that anticipates needs and initiates interactions.

This agent monitors context, predicts needs, and proactively engages
users at appropriate moments with helpful suggestions.

Methods:

__init__(self, temperature)

Lines 210-221

Purpose:

Initialize proactive agent.

Args:
    temperature: LLM temperature

Parameters: – temperature: Sampling temperature (0-1)

add_context_signal(self, signal_type, content, confidence, metadata)

Lines 223-249

Purpose:

Add a context signal from monitoring.

Args:
    signal_type: Type of signal
    content: Signal content
    confidence: Detection confidence
    metadata: Additional information

Returns:
    Created signal

Parameters: – signal_type: Signal type – content: Content – confidence: Confidence – metadata: Metadata

Returns: ContextSignal

predict_need(self, context_window)

Lines 251-286

Purpose:

Predict user need based on recent context.

Args:
    context_window: Number of recent signals to consider

Returns:
    Predicted need description

Parameters: – context_window: Context window

Returns: <ast.Subscript object at 0x000001A06D7F6550>

detect_opportunity(self, context)

Lines 288-329

Purpose:

Detect opportunity for proactive engagement.

Args:
    context: Current context description

Returns:
    Opportunity details if found

Parameters: – context: Context

Returns: <ast.Subscript object at 0x000001A06E0946D0>

assess_timing(self, engagement_type)

Lines 331-363

Purpose:

Assess if timing is appropriate for engagement.

Args:
    engagement_type: Type of engagement

Returns:
    Timing appropriateness score (0-1)

Parameters: – engagement_type: Engagement type

Returns: float

create_engagement(self, engagement_type, message, priority, suggested_action, triggers)

Lines 365-411

Purpose:

Create a proactive engagement.

Args:
    engagement_type: Type of engagement
    message: Engagement message
    priority: Priority level
    suggested_action: Suggested action
    triggers: What triggered this

Returns:
    Created engagement

Parameters: – engagement_type: Engagement type – message: Message – priority: Priority – suggested_action: Suggested action – triggers: Triggers

Returns: ProactiveEngagement

should_engage(self, engagement)

Lines 413-455

Purpose:

Determine if agent should deliver engagement.

Args:
    engagement: Engagement to evaluate

Returns:
    Whether to engage

Parameters: – engagement: Engagement

Returns: bool

engage_user(self, engagement)

Lines 457-471

Purpose:

Deliver engagement to user.

Args:
    engagement: Engagement to deliver

Returns:
    Whether engagement was delivered

Parameters: – engagement: Engagement

Returns: bool

record_response(self, engagement_id, response)

Lines 473-488

Purpose:

Record user response to engagement.

Args:
    engagement_id: Engagement identifier
    response: User response

Parameters: – engagement_id: Engagement id – response: Response

get_acceptance_rate(self)

Lines 490-498

Purpose:

Get acceptance rate for delivered engagements.

Parameters:

Returns: float

get_engagement_summary(self)

Lines 500-517

Purpose:

Get summary of engagement activity.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7188D0>

---

Functions

demonstrate_proactive_engagement()

Lines 520-908

Purpose:

Demonstrate proactive engagement pattern.

---

Key Concepts Summary

Pattern Overview

The Proactive Engagement 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 129
• Pattern 131

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 Proactive Engagement 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: 130_proactive_engagement.py
Total Lines: 913
Pattern Category: AI Agent Design Pattern #130

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 129: Clarification & Disambiguation

Description:
    Detects ambiguous or unclear user inputs and asks targeted clarifying questions
    to resolve uncertainties before proceeding.

Components:
    - Ambiguity detection
    - Clarification question generation
    - Context tracking
    - Progressive refinement

Use Cases:
    - Conversational agents handling ambiguous queries
    - Complex instruction following
    - Reducing errors from misunderstanding

LangChain Implementation:
    Uses LangChain to detect ambiguity and generate appropriate clarification questions.
"""

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser, StrOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


class AmbiguityAnalysis(BaseModel):
    """Analysis of ambiguity in user input"""
    is_ambiguous: bool = Field(description="Whether input is ambiguous")
    ambiguity_type: str = Field(description="Type of ambiguity: lexical, syntactic, semantic, referential")
    confidence: float = Field(description="Confidence in ambiguity detection")
    ambiguous_elements: List[str] = Field(description="Specific ambiguous elements")
    clarification_needed: str = Field(description="What needs clarification")


class ClarificationQuestion(BaseModel):
    """A clarification question"""
    question: str = Field(description="The clarification question")
    options: Optional[List[str]] = Field(description="Possible options if applicable")
    priority: int = Field(description="Priority (1=high, 3=low)")


class ClarificationAgent:
    """Agent for handling ambiguity and generating clarifications"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.3):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.clarification_history: List[Dict[str, Any]] = []
        
        # Ambiguity detection prompt
        self.ambiguity_prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze the user's input for ambiguity or unclear elements.

Types of ambiguity:
- Lexical: Multiple meanings of a word (e.g., "bank" - financial or river)
- Syntactic: Sentence structure unclear (e.g., "I saw the man with the telescope")
- Semantic: Vague or imprecise meaning (e.g., "nearby", "soon")
- Referential: Unclear what something refers to (e.g., "it", "that thing")

Return JSON:
{{
  "is_ambiguous": true/false,
  "ambiguity_type": "type",
  "confidence": 0.9,
  "ambiguous_elements": ["element1", "element2"],
  "clarification_needed": "what specifically needs clarification"
}}"""),
            ("user", "Input: {input}\n\nPrevious context: {context}")
        ])
        
        # Clarification generation prompt
        self.clarification_prompt = ChatPromptTemplate.from_messages([
            ("system", """Generate a helpful clarification question to resolve the ambiguity.

Make the question:
1. Specific and targeted
2. Easy to answer
3. Include options when helpful
4. Natural and conversational

Return JSON:
{{
  "question": "the clarification question",
  "options": ["option1", "option2"] or null,
  "priority": 1-3
}}"""),
            ("user", """Input: {input}
Ambiguity: {ambiguity}
Elements needing clarification: {elements}

Generate a clarification question.""")
        ])
        
        self.json_parser = JsonOutputParser()
    
    def detect_ambiguity(self, user_input: str, context: str = "") -> AmbiguityAnalysis:
        """Detect if input is ambiguous"""
        try:
            chain = self.ambiguity_prompt | self.llm | self.json_parser
            result = chain.invoke({"input": user_input, "context": context})
            
            return AmbiguityAnalysis(**result)
        except Exception as e:
            print(f"Ambiguity detection error: {e}")
            return AmbiguityAnalysis(
                is_ambiguous=False,
                ambiguity_type="unknown",
                confidence=0.0,
                ambiguous_elements=[],
                clarification_needed=""
            )
    
    def generate_clarification(
        self,
        user_input: str,
        analysis: AmbiguityAnalysis
    ) -> ClarificationQuestion:
        """Generate clarification question"""
        try:
            chain = self.clarification_prompt | self.llm | self.json_parser
            result = chain.invoke({
                "input": user_input,
                "ambiguity": analysis.ambiguity_type,
                "elements": ", ".join(analysis.ambiguous_elements)
            })
            
            return ClarificationQuestion(**result)
        except Exception as e:
            print(f"Clarification generation error: {e}")
            return ClarificationQuestion(
                question="Could you please provide more details?",
                options=None,
                priority=2
            )
    
    def process_input(self, user_input: str, context: str = "") -> Dict[str, Any]:
        """Process input and return clarification if needed"""
        print(f"\n{'='*70}")
        print(f"📝 Input: {user_input}")
        
        # Detect ambiguity
        analysis = self.detect_ambiguity(user_input, context)
        
        print(f"\n🔍 Analysis:")
        print(f"   Ambiguous: {analysis.is_ambiguous}")
        print(f"   Confidence: {analysis.confidence:.2f}")
        
        result = {
            "input": user_input,
            "is_ambiguous": analysis.is_ambiguous,
            "analysis": analysis,
            "clarification": None
        }
        
        if analysis.is_ambiguous and analysis.confidence > 0.6:
            print(f"   Type: {analysis.ambiguity_type}")
            print(f"   Ambiguous elements: {', '.join(analysis.ambiguous_elements)}")
            
            # Generate clarification
            clarification = self.generate_clarification(user_input, analysis)
            result["clarification"] = clarification
            
            print(f"\n❓ Clarification needed:")
            print(f"   {clarification.question}")
            if clarification.options:
                print(f"   Options:")
                for i, opt in enumerate(clarification.options, 1):
                    print(f"     {i}. {opt}")
            
            # Store in history
            self.clarification_history.append({
                "input": user_input,
                "analysis": analysis.dict(),
                "clarification": clarification.dict()
            })
        else:
            print(f"\n✅ Input is clear, no clarification needed")
        
        return result
    
    def resolve_with_clarification(
        self,
        original_input: str,
        clarification_response: str
    ) -> str:
        """Process clarified input"""
        resolution_prompt = ChatPromptTemplate.from_messages([
            ("system", "Combine the original input with the clarification to form a complete, unambiguous statement."),
            ("user", """Original: {original}
Clarification response: {clarification}

Generate complete, unambiguous statement:""")
        ])
        
        chain = resolution_prompt | self.llm | StrOutputParser()
        resolved = chain.invoke({
            "original": original_input,
            "clarification": clarification_response
        })
        
        print(f"\n✅ Resolved statement:")
        print(f"   {resolved}")
        
        return resolved


def demonstrate_clarification():
    """Demonstrate clarification and disambiguation"""
    print("=" * 80)
    print("Pattern 129: Clarification & Disambiguation")
    print("=" * 80)
    
    agent = ClarificationAgent()
    
    # Example 1: Lexical ambiguity
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Lexical Ambiguity")
    print("=" * 80)
    
    ambiguous_inputs = [
        "I need to go to the bank",
        "Can you get me the bat?",
        "Show me the rock music"
    ]
    
    for input_text in ambiguous_inputs:
        result = agent.process_input(input_text)
        
        if result["clarification"]:
            # Simulate user providing clarification
            print(f"\n   💬 User provides clarification: Financial bank")
            resolved = agent.resolve_with_clarification(
                input_text,
                "The financial institution"
            )
    
    # Example 2: Referential ambiguity
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Referential Ambiguity")
    print("=" * 80)
    
    referential_inputs = [
        ("John told Mike he should leave", "Previous: John and Mike were arguing"),
        ("Put it on the table", "Previous: User has a box and a book"),
        ("I want the blue one", "Previous: Looking at shirts")
    ]
    
    for input_text, context in referential_inputs:
        agent.process_input(input_text, context)
    
    # Example 3: Semantic vagueness
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Semantic Vagueness")
    print("=" * 80)
    
    vague_inputs = [
        "Schedule a meeting soon",
        "I need a lot of storage",
        "Find a nearby restaurant"
    ]
    
    for input_text in vague_inputs:
        agent.process_input(input_text)
    
    # Example 4: Complex ambiguity
    print("\n" + "=" * 80)
    print("EXAMPLE 4: Complex Ambiguous Request")
    print("=" * 80)
    
    complex_input = "Book me a nice hotel room with a good view for my trip"
    result = agent.process_input(complex_input)
    
    if result["clarification"]:
        print(f"\n   💬 User response: Paris, June 15-20, budget up to $200/night")
        resolved = agent.resolve_with_clarification(
            complex_input,
            "Paris, France from June 15-20, 2024, budget $200 per night"
        )
    
    # Example 5: Clear input (no clarification needed)
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Clear Input (No Clarification)")
    print("=" * 80)
    
    clear_inputs = [
        "Set an alarm for 7:00 AM tomorrow",
        "What is the capital of France?",
        "Calculate 15% tip on $85.50"
    ]
    
    for input_text in clear_inputs:
        agent.process_input(input_text)
    
    # Statistics
    print("\n" + "=" * 80)
    print("CLARIFICATION STATISTICS")
    print("=" * 80)
    
    print(f"\nTotal clarifications asked: {len(agent.clarification_history)}")
    
    if agent.clarification_history:
        print(f"\nAmbiguity types encountered:")
        types = {}
        for item in agent.clarification_history:
            amb_type = item['analysis']['ambiguity_type']
            types[amb_type] = types.get(amb_type, 0) + 1
        
        for amb_type, count in types.items():
            print(f"  - {amb_type}: {count}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Clarification & Disambiguation Pattern:
- Detects ambiguous or unclear user inputs
- Generates targeted clarification questions
- Resolves ambiguity through progressive refinement
- Improves accuracy and user satisfaction

Types of Ambiguity Handled:
✓ Lexical - Multiple word meanings
✓ Syntactic - Unclear sentence structure
✓ Semantic - Vague or imprecise terms
✓ Referential - Unclear references

Key Benefits:
✓ Reduces errors from misunderstanding
✓ Improves task completion accuracy
✓ Better user experience
✓ Handles complex, real-world inputs
✓ Proactive problem prevention

Best Practices:
• Ask specific, targeted questions
• Provide options when helpful
• Keep questions natural and conversational
• Prioritize most important clarifications
• Track clarification history for learning
    """)


if __name__ == "__main__":
    demonstrate_clarification()

    

Agentic Design Pattern: Clarification

Pattern 129: Clarification – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 129: Clarification & Disambiguation

Description: Detects ambiguous or unclear user inputs and asks targeted clarifying questions to resolve uncertainties before proceeding.

Components: – Ambiguity detection – Clarification question generation – Context tracking – Progressive refinement

Use Cases: – Conversational agents handling ambiguous queries – Complex instruction following – Reducing errors from misunderstanding

LangChain Implementation: Uses LangChain to detect ambiguity and generate appropriate clarification questions.

---

File Statistics

• Total Lines: 354
• Classes: 3
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 129: Clarification & Disambiguation

Description:
    Detects ambiguous or unclear user inputs and asks targeted clarifying questions
    to resolve uncertainties before proceeding.

Components:
    - Ambiguity detection
    - Clarification question generation
    - Context tracking
    - Progressive refinement

Use Cases:
    - Conversational agents handling ambiguous queries
    - Complex instruction following
    - Reducing errors from misunderstanding

LangChain Implementation:
    Uses LangChain to detect ambiguity and generate appropriate clarification questions.
"""

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

---

Import Statements

Standard Library Imports

import langchain_core.output_parsers.JsonOutputParser
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: Regular expression operations – os: Environment variable access and system operations – 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)

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: AmbiguityAnalysis

Lines 35-41

Inherits from: BaseModel

Purpose:

Analysis of ambiguity in user input

---

Class: ClarificationQuestion

Lines 44-48

Inherits from: BaseModel

Purpose:

A clarification question

---

Class: ClarificationAgent

Lines 51-212

Purpose:

Agent for handling ambiguity and generating clarifications

Methods:

__init__(self, model, temperature)

Lines 54-102

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

detect_ambiguity(self, user_input, context)

Lines 104-119

Purpose:

Detect if input is ambiguous

Parameters: – user_input: Input data – context: Context

Returns: AmbiguityAnalysis

generate_clarification(self, user_input, analysis)

Lines 121-142

Purpose:

Generate clarification question

Parameters: – user_input: Input data – analysis: Analysis

Returns: ClarificationQuestion

process_input(self, user_input, context)

Lines 144-187

Purpose:

Process input and return clarification if needed

Parameters: – user_input: Input data – context: Context

Returns: <ast.Subscript object at 0x000001A06D7EEA90>

resolve_with_clarification(self, original_input, clarification_response)

Lines 189-212

Purpose:

Process clarified input

Parameters: – original_input: Input data – clarification_response: Clarification response

Returns: str

---

Functions

demonstrate_clarification()

Lines 215-349

Purpose:

Demonstrate clarification and disambiguation

---

Key Concepts Summary

Pattern Overview

The Clarification 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 128
• Pattern 130

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 Clarification 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: 129_clarification.py
Total Lines: 354
Pattern Category: AI Agent Design Pattern #129

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 128: Multi-Turn Dialogue Management

Description:
    Manages coherent multi-turn conversations with context tracking, dialogue state,
    and appropriate responses across multiple exchanges.

Components:
    - Dialogue state tracking
    - Context management
    - Intent recognition
    - Response generation

Use Cases:
    - Conversational AI assistants
    - Customer service chatbots
    - Interactive tutoring systems

LangChain Implementation:
    Uses LangChain memory, state management, and conversation chains.
"""

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv
from datetime import datetime
from enum import Enum

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.runnables import RunnablePassthrough
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


class DialogueState(str, Enum):
    """Dialogue states"""
    GREETING = "greeting"
    INFO_GATHERING = "info_gathering"
    CLARIFICATION = "clarification"
    PROCESSING = "processing"
    RESPONSE = "response"
    CLOSING = "closing"


class UserIntent(BaseModel):
    """Detected user intent"""
    intent: str = Field(description="Primary intent")
    confidence: float = Field(description="Confidence score")
    entities: Dict[str, Any] = Field(description="Extracted entities")


class DialogueContext(BaseModel):
    """Dialogue context"""
    state: str = Field(description="Current dialogue state")
    topic: Optional[str] = Field(description="Current topic")
    user_goal: Optional[str] = Field(description="User's goal")
    collected_info: Dict[str, Any] = Field(default_factory=dict)


class MultiTurnDialogueAgent:
    """Agent for managing multi-turn conversations"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.7):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.conversation_history: List[Dict[str, str]] = []
        self.context: DialogueContext = DialogueContext(
            state=DialogueState.GREETING.value,
            topic=None,
            user_goal=None,
            collected_info={}
        )
        
        # Intent detection prompt
        self.intent_prompt = ChatPromptTemplate.from_messages([
            ("system", """Analyze the user's message and detect their intent.
Common intents: greeting, question, request_info, provide_info, complaint, thanks, goodbye

Return JSON:
{{
  "intent": "intent_name",
  "confidence": 0.9,
  "entities": {{"key": "value"}}
}}"""),
            ("user", "{message}")
        ])
        
        # Response generation with context
        self.response_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful conversational AI assistant.
Current conversation context:
- State: {state}
- Topic: {topic}
- User Goal: {user_goal}
- Collected Info: {collected_info}

Provide a natural, contextually appropriate response that:
1. Acknowledges the user's message
2. Maintains conversation flow
3. Helps achieve the user's goal
4. Asks follow-up questions if needed"""),
            MessagesPlaceholder(variable_name="history"),
            ("user", "{message}")
        ])
        
        self.intent_parser = JsonOutputParser()
        self.response_parser = StrOutputParser()
    
    def detect_intent(self, message: str) -> UserIntent:
        """Detect user intent from message"""
        try:
            chain = self.intent_prompt | self.llm | self.intent_parser
            result = chain.invoke({"message": message})
            
            return UserIntent(
                intent=result["intent"],
                confidence=result["confidence"],
                entities=result.get("entities", {})
            )
        except Exception as e:
            print(f"Intent detection error: {e}")
            return UserIntent(intent="unknown", confidence=0.0, entities={})
    
    def update_context(self, intent: UserIntent, message: str):
        """Update dialogue context based on intent"""
        # Update state based on intent
        if intent.intent == "greeting":
            self.context.state = DialogueState.GREETING.value
        elif intent.intent in ["question", "request_info"]:
            self.context.state = DialogueState.INFO_GATHERING.value
        elif intent.intent == "provide_info":
            self.context.state = DialogueState.PROCESSING.value
            # Store provided information
            self.context.collected_info.update(intent.entities)
        elif intent.intent == "goodbye":
            self.context.state = DialogueState.CLOSING.value
        
        # Extract topic if not set
        if not self.context.topic and len(self.conversation_history) > 0:
            # Simple topic extraction (could be enhanced)
            for entity_key in intent.entities:
                if entity_key not in ["time", "date", "number"]:
                    self.context.topic = intent.entities[entity_key]
                    break
    
    def generate_response(self, message: str) -> str:
        """Generate contextually appropriate response"""
        # Detect intent
        intent = self.detect_intent(message)
        print(f"\n🎯 Detected Intent: {intent.intent} (confidence: {intent.confidence:.2f})")
        
        # Update context
        self.update_context(intent, message)
        print(f"📊 Dialogue State: {self.context.state}")
        if self.context.topic:
            print(f"💡 Topic: {self.context.topic}")
        
        # Prepare conversation history for prompt
        history = []
        for turn in self.conversation_history[-5:]:  # Last 5 turns
            if turn["role"] == "user":
                history.append(HumanMessage(content=turn["content"]))
            else:
                history.append(AIMessage(content=turn["content"]))
        
        # Generate response
        chain = self.response_prompt | self.llm | self.response_parser
        
        response = chain.invoke({
            "message": message,
            "state": self.context.state,
            "topic": self.context.topic or "not set",
            "user_goal": self.context.user_goal or "not determined",
            "collected_info": str(self.context.collected_info),
            "history": history
        })
        
        return response
    
    def chat(self, user_message: str) -> str:
        """Process user message and return response"""
        print(f"\n{'='*60}")
        print(f"👤 User: {user_message}")
        
        # Add to history
        self.conversation_history.append({
            "role": "user",
            "content": user_message,
            "timestamp": datetime.now().isoformat()
        })
        
        # Generate response
        response = self.generate_response(user_message)
        
        # Add response to history
        self.conversation_history.append({
            "role": "assistant",
            "content": response,
            "timestamp": datetime.now().isoformat()
        })
        
        print(f"🤖 Assistant: {response}")
        
        return response
    
    def get_conversation_summary(self) -> str:
        """Generate summary of conversation"""
        if not self.conversation_history:
            return "No conversation yet."
        
        summary_prompt = ChatPromptTemplate.from_messages([
            ("system", "Summarize the following conversation in 2-3 sentences."),
            ("user", "{conversation}")
        ])
        
        conversation_text = "\n".join([
            f"{turn['role']}: {turn['content']}"
            for turn in self.conversation_history
        ])
        
        chain = summary_prompt | self.llm | StrOutputParser()
        summary = chain.invoke({"conversation": conversation_text})
        
        return summary
    
    def reset_conversation(self):
        """Reset conversation state"""
        self.conversation_history = []
        self.context = DialogueContext(
            state=DialogueState.GREETING.value,
            topic=None,
            user_goal=None,
            collected_info={}
        )
        print("\n🔄 Conversation reset")


def demonstrate_multi_turn_dialogue():
    """Demonstrate multi-turn dialogue management"""
    print("=" * 80)
    print("Pattern 128: Multi-Turn Dialogue Management")
    print("=" * 80)
    
    agent = MultiTurnDialogueAgent()
    
    # Example 1: Customer service conversation
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Customer Service Conversation")
    print("=" * 80)
    
    conversation_1 = [
        "Hi, I need help with my order",
        "My order number is 12345",
        "It was supposed to arrive yesterday but I haven't received it",
        "Yes, my address is 123 Main Street",
        "Thank you for your help!"
    ]
    
    for message in conversation_1:
        agent.chat(message)
    
    print("\n" + "-" * 60)
    print("📝 Conversation Summary:")
    print(agent.get_conversation_summary())
    
    # Example 2: Technical support
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Technical Support Conversation")
    print("=" * 80)
    
    agent.reset_conversation()
    
    conversation_2 = [
        "Hello, my computer won't start",
        "It's a laptop, about 2 years old",
        "When I press the power button, nothing happens at all",
        "No, there are no lights or sounds",
        "Okay, I'll try that. Thanks!"
    ]
    
    for message in conversation_2:
        agent.chat(message)
    
    print("\n" + "-" * 60)
    print("📝 Conversation Summary:")
    print(agent.get_conversation_summary())
    
    # Example 3: Information gathering
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Information Gathering Conversation")
    print("=" * 80)
    
    agent.reset_conversation()
    
    conversation_3 = [
        "I want to book a flight",
        "From New York to London",
        "Next week, preferably Tuesday",
        "Economy class is fine",
        "Great, please proceed with the booking"
    ]
    
    for message in conversation_3:
        agent.chat(message)
    
    print("\n" + "-" * 60)
    print("📝 Conversation Summary:")
    print(agent.get_conversation_summary())
    
    print("\n" + "-" * 60)
    print("📊 Final Context State:")
    print(f"  State: {agent.context.state}")
    print(f"  Topic: {agent.context.topic}")
    print(f"  Collected Info: {agent.context.collected_info}")
    print(f"  Total Turns: {len(agent.conversation_history)}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Multi-Turn Dialogue Management Pattern:
- Maintains context across conversation turns
- Tracks dialogue state and user intent
- Generates contextually appropriate responses
- Manages information collection

Key Features:
✓ Intent detection
✓ State tracking
✓ Context maintenance
✓ Natural conversation flow
✓ Information extraction

Dialogue States:
• Greeting - Initial interaction
• Info Gathering - Collecting information
• Clarification - Resolving ambiguities
• Processing - Handling requests
• Response - Providing answers
• Closing - Ending conversation

Benefits:
✓ Coherent multi-turn conversations
✓ Context-aware responses
✓ Better user experience
✓ Goal-oriented dialogue
✓ Efficient information gathering
    """)


if __name__ == "__main__":
    demonstrate_multi_turn_dialogue()

    

Agentic Design Pattern: Multi Turn Dialogue

Pattern 128: Multi Turn Dialogue – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Multi Turn Dialogue pattern implementation.

---

Pattern Description

Pattern 128: Multi-Turn Dialogue Management

Description: Manages coherent multi-turn conversations with context tracking, dialogue state, and appropriate responses across multiple exchanges.

Components: – Dialogue state tracking – Context management – Intent recognition – Response generation

Use Cases: – Conversational AI assistants – Customer service chatbots – Interactive tutoring systems

LangChain Implementation: Uses LangChain memory, state management, and conversation chains.

---

File Statistics

• Total Lines: 357
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 128: Multi-Turn Dialogue Management

Description:
    Manages coherent multi-turn conversations with context tracking, dialogue state,
    and appropriate responses across multiple exchanges.

Components:
    - Dialogue state tracking
    - Context management
    - Intent recognition
    - Response generation

Use Cases:
    - Conversational AI assistants
    - Customer service chatbots
    - Interactive tutoring systems

LangChain Implementation:
    Uses LangChain memory, state management, and conversation chains.
"""

Purpose: This module implements the Multi Turn Dialogue pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import datetime.datetime
import enum.Enum
import langchain_core.messages.AIMessage
import langchain_core.messages.HumanMessage
import langchain_core.messages.SystemMessage
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.MessagesPlaceholder
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import langchain_core.runnables.RunnablePassthrough
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – enum.Enum: Enumeration support – langchain_core.messages.AIMessage: Regular expression operations – langchain_core.messages.HumanMessage: Regular expression operations – langchain_core.messages.SystemMessage: Regular expression operations – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.MessagesPlaceholder: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: Regular expression operations – langchain_core.runnables.RunnablePassthrough: Regular expression operations – os: Environment variable access and system operations – 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)

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: DialogueState

Lines 39-46

Inherits from: str, Enum

Purpose:

Dialogue states

---

Class: UserIntent

Lines 49-53

Inherits from: BaseModel

Purpose:

Detected user intent

---

Class: DialogueContext

Lines 56-61

Inherits from: BaseModel

Purpose:

Dialogue context

---

Class: MultiTurnDialogueAgent

Lines 64-238

Purpose:

Agent for managing multi-turn conversations

Methods:

__init__(self, model, temperature)

Lines 67-110

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

detect_intent(self, message)

Lines 112-125

Purpose:

Detect user intent from message

Parameters: – message: Message

Returns: UserIntent

update_context(self, intent, message)

Lines 127-147

Purpose:

Update dialogue context based on intent

Parameters: – intent: Intent – message: Message

generate_response(self, message)

Lines 149-181

Purpose:

Generate contextually appropriate response

Parameters: – message: Message

Returns: str

chat(self, user_message)

Lines 183-207

Purpose:

Process user message and return response

Parameters: – user_message: User message

Returns: str

get_conversation_summary(self)

Lines 209-227

Purpose:

Generate summary of conversation

Parameters:

Returns: str

reset_conversation(self)

Lines 229-238

Purpose:

Reset conversation state

Parameters:

---

Functions

demonstrate_multi_turn_dialogue()

Lines 241-352

Purpose:

Demonstrate multi-turn dialogue management

---

Key Concepts Summary

Pattern Overview

The Multi Turn Dialogue 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 127
• Pattern 129

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 Multi Turn Dialogue 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: 128_multi_turn_dialogue.py
Total Lines: 357
Pattern Category: AI Agent Design Pattern #128

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 127: Semantic Search & Retrieval

Description:
    Retrieves information based on semantic similarity rather than keyword matching,
    using embeddings and vector databases for intelligent information retrieval.

Components:
    - Text embeddings
    - Vector similarity search
    - Semantic ranking
    - Hybrid retrieval (keyword + semantic)

Use Cases:
    - Intelligent document search
    - Question answering systems
    - Recommendation systems

LangChain Implementation:
    Uses LangChain embeddings, vector stores, and retrievers for semantic search.
"""

import os
from typing import List, Dict, Any, Optional
from dotenv import load_dotenv

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.retrievers import BM25Retriever

load_dotenv()


class SemanticSearchAgent:
    """Agent for semantic search and retrieval"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore: Optional[Chroma] = None
        self.documents: List[Document] = []
        
        # Text splitter for chunking
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            length_function=len
        )
        
        # QA prompt
        self.qa_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a helpful assistant that answers questions based on the provided context.
Use only the information from the context to answer. If you cannot answer from the context, say so.
Always cite which part of the context you used."""),
            ("user", """Context:
{context}

Question: {question}

Answer:""")
        ])
    
    def load_documents(self, documents: List[str], metadatas: Optional[List[Dict]] = None):
        """Load documents into the system"""
        print(f"\n📚 Loading {len(documents)} documents...")
        
        # Create Document objects
        if metadatas:
            self.documents = [
                Document(page_content=doc, metadata=meta)
                for doc, meta in zip(documents, metadatas)
            ]
        else:
            self.documents = [
                Document(page_content=doc, metadata={"id": i})
                for i, doc in enumerate(documents)
            ]
        
        # Split documents
        split_docs = self.text_splitter.split_documents(self.documents)
        print(f"   Split into {len(split_docs)} chunks")
        
        # Create vector store
        self.vectorstore = Chroma.from_documents(
            documents=split_docs,
            embedding=self.embeddings,
            collection_name="semantic_search"
        )
        
        print(f"   ✓ Vectorstore created with {len(split_docs)} embeddings")
    
    def semantic_search(self, query: str, k: int = 3) -> List[Document]:
        """Pure semantic search using vector similarity"""
        print(f"\n🔍 Semantic search for: '{query}'")
        
        if not self.vectorstore:
            print("   ❌ No documents loaded!")
            return []
        
        # Retrieve similar documents
        results = self.vectorstore.similarity_search_with_score(query, k=k)
        
        print(f"   Found {len(results)} results:\n")
        for i, (doc, score) in enumerate(results, 1):
            print(f"   {i}. Score: {score:.4f}")
            print(f"      {doc.page_content[:100]}...\n")
        
        return [doc for doc, _ in results]
    
    def hybrid_search(self, query: str, k: int = 3) -> List[Document]:
        """Hybrid search combining semantic and keyword matching"""
        print(f"\n🔍 Hybrid search for: '{query}'")
        
        if not self.vectorstore or not self.documents:
            print("   ❌ No documents loaded!")
            return []
        
        # Create semantic retriever
        semantic_retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        
        # Create keyword retriever (BM25)
        bm25_retriever = BM25Retriever.from_documents(self.documents)
        bm25_retriever.k = k
        
        # Combine retrievers
        ensemble_retriever = EnsembleRetriever(
            retrievers=[semantic_retriever, bm25_retriever],
            weights=[0.7, 0.3]  # Favor semantic
        )
        
        results = ensemble_retriever.get_relevant_documents(query)
        
        print(f"   Found {len(results)} results (semantic + keyword)\n")
        for i, doc in enumerate(results[:k], 1):
            print(f"   {i}. {doc.page_content[:100]}...\n")
        
        return results[:k]
    
    def search_with_reranking(self, query: str, k: int = 3) -> List[Document]:
        """Search with LLM-based reranking/compression"""
        print(f"\n🔍 Search with reranking for: '{query}'")
        
        if not self.vectorstore:
            print("   ❌ No documents loaded!")
            return []
        
        # Base retriever
        base_retriever = self.vectorstore.as_retriever(search_kwargs={"k": k * 2})
        
        # Compressor for reranking
        compressor = LLMChainExtractor.from_llm(self.llm)
        
        # Compression retriever
        compression_retriever = ContextualCompressionRetriever(
            base_compressor=compressor,
            base_retriever=base_retriever
        )
        
        results = compression_retriever.get_relevant_documents(query)
        
        print(f"   Found and reranked {len(results)} results\n")
        for i, doc in enumerate(results[:k], 1):
            print(f"   {i}. {doc.page_content[:150]}...\n")
        
        return results[:k]
    
    def answer_question(self, question: str, search_method: str = "semantic") -> str:
        """Answer question using retrieved context"""
        print(f"\n💬 Answering question: '{question}'")
        print(f"   Using {search_method} search")
        
        # Retrieve context
        if search_method == "semantic":
            context_docs = self.semantic_search(question, k=3)
        elif search_method == "hybrid":
            context_docs = self.hybrid_search(question, k=3)
        else:  # reranking
            context_docs = self.search_with_reranking(question, k=3)
        
        if not context_docs:
            return "No relevant context found."
        
        # Combine context
        context = "\n\n".join([doc.page_content for doc in context_docs])
        
        # Generate answer
        chain = self.qa_prompt | self.llm | StrOutputParser()
        answer = chain.invoke({"context": context, "question": question})
        
        print(f"\n📝 Answer:\n{answer}")
        
        return answer
    
    def find_similar_documents(self, reference_text: str, k: int = 3) -> List[Document]:
        """Find documents similar to reference text"""
        print(f"\n🔍 Finding similar documents to reference text...")
        
        if not self.vectorstore:
            print("   ❌ No documents loaded!")
            return []
        
        results = self.vectorstore.similarity_search(reference_text, k=k)
        
        print(f"   Found {len(results)} similar documents:\n")
        for i, doc in enumerate(results, 1):
            print(f"   {i}. {doc.page_content[:100]}...")
            if 'id' in doc.metadata:
                print(f"      (Document ID: {doc.metadata['id']})\n")
        
        return results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get search system statistics"""
        return {
            "total_documents": len(self.documents),
            "vectorstore_loaded": self.vectorstore is not None,
            "embedding_model": "text-embedding-ada-002"
        }


def demonstrate_semantic_search():
    """Demonstrate semantic search and retrieval"""
    print("=" * 80)
    print("Pattern 127: Semantic Search & Retrieval")
    print("=" * 80)
    
    agent = SemanticSearchAgent()
    
    # Example 1: Load documents
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Loading Documents")
    print("=" * 80)
    
    documents = [
        """Python is a high-level, interpreted programming language known for its simplicity 
        and readability. Created by Guido van Rossum in 1991, Python emphasizes code 
        readability and allows programmers to express concepts in fewer lines of code.""",
        
        """Machine learning is a subset of artificial intelligence that enables systems to 
        learn and improve from experience without being explicitly programmed. It focuses 
        on developing computer programs that can access data and learn from it.""",
        
        """Deep learning is a subset of machine learning that uses neural networks with 
        multiple layers. These deep neural networks attempt to simulate the behavior of 
        the human brain, allowing it to learn from large amounts of data.""",
        
        """Natural Language Processing (NLP) is a branch of AI that helps computers understand, 
        interpret, and manipulate human language. NLP draws from many disciplines, including 
        computer science and computational linguistics.""",
        
        """Computer vision is a field of artificial intelligence that trains computers to 
        interpret and understand the visual world. Using digital images and deep learning 
        models, machines can accurately identify and classify objects.""",
        
        """The Transformer architecture, introduced in the paper 'Attention is All You Need', 
        revolutionized NLP. It relies entirely on attention mechanisms, dispensing with 
        recurrence and convolutions entirely."""
    ]
    
    metadatas = [
        {"topic": "programming", "language": "python"},
        {"topic": "AI", "subtopic": "machine_learning"},
        {"topic": "AI", "subtopic": "deep_learning"},
        {"topic": "AI", "subtopic": "NLP"},
        {"topic": "AI", "subtopic": "computer_vision"},
        {"topic": "AI", "subtopic": "transformers"}
    ]
    
    agent.load_documents(documents, metadatas)
    
    # Example 2: Semantic search
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Pure Semantic Search")
    print("=" * 80)
    
    queries = [
        "neural networks and learning from data",
        "understanding human language with computers",
        "programming language for beginners"
    ]
    
    for query in queries:
        agent.semantic_search(query, k=2)
    
    # Example 3: Hybrid search
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Hybrid Search (Semantic + Keyword)")
    print("=" * 80)
    
    agent.hybrid_search("deep neural networks", k=2)
    
    # Example 4: Search with reranking
    print("\n" + "=" * 80)
    print("EXAMPLE 4: Search with LLM Reranking")
    print("=" * 80)
    
    agent.search_with_reranking("AI techniques for processing text", k=2)
    
    # Example 5: Question answering with different search methods
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Question Answering")
    print("=" * 80)
    
    questions = [
        "What is the Transformer architecture?",
        "Who created Python and when?",
        "What is the relationship between deep learning and machine learning?"
    ]
    
    methods = ["semantic", "hybrid", "reranking"]
    
    for question in questions[:1]:  # Just first question with all methods
        print(f"\nQuestion: {question}\n")
        for method in methods:
            print(f"\n--- Method: {method} ---")
            agent.answer_question(question, search_method=method)
    
    # Example 6: Find similar documents
    print("\n" + "=" * 80)
    print("EXAMPLE 6: Finding Similar Documents")
    print("=" * 80)
    
    reference = "Neural networks that can process and generate human language"
    agent.find_similar_documents(reference, k=3)
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Semantic Search & Retrieval Pattern:
- Finds information based on meaning, not just keywords
- Uses embeddings to capture semantic similarity
- Multiple search strategies available
- Enables intelligent question answering

Search Methods:
1. Pure Semantic: Vector similarity only
2. Hybrid: Combines semantic + keyword (BM25)
3. With Reranking: LLM compresses and reranks results

Key Benefits:
✓ Understanding user intent
✓ Finding conceptually similar content
✓ Better than keyword matching
✓ Handles synonyms and paraphrases
✓ Improved retrieval accuracy

Use Cases:
• Document search systems
• Question answering
• Recommendation engines
• Knowledge bases
• Research assistants
    """)


if __name__ == "__main__":
    demonstrate_semantic_search()

    

Agentic Design Pattern: Semantic Search Retrieval

Pattern 127: Semantic Search Retrieval – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Semantic Search Retrieval pattern implementation.

---

Pattern Description

Pattern 127: Semantic Search & Retrieval

Description: Retrieves information based on semantic similarity rather than keyword matching, using embeddings and vector databases for intelligent information retrieval.

Components: – Text embeddings – Vector similarity search – Semantic ranking – Hybrid retrieval (keyword + semantic)

Use Cases: – Intelligent document search – Question answering systems – Recommendation systems

LangChain Implementation: Uses LangChain embeddings, vector stores, and retrievers for semantic search.

---

File Statistics

• Total Lines: 367
• Classes: 1
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 127: Semantic Search & Retrieval

Description:
    Retrieves information based on semantic similarity rather than keyword matching,
    using embeddings and vector databases for intelligent information retrieval.

Components:
    - Text embeddings
    - Vector similarity search
    - Semantic ranking
    - Hybrid retrieval (keyword + semantic)

Use Cases:
    - Intelligent document search
    - Question answering systems
    - Recommendation systems

LangChain Implementation:
    Uses LangChain embeddings, vector stores, and retrievers for semantic search.
"""

Purpose: This module implements the Semantic Search Retrieval pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import langchain.retrievers.ContextualCompressionRetriever
import langchain.retrievers.EnsembleRetriever
import langchain.retrievers.document_compressors.LLMChainExtractor
import langchain_community.retrievers.BM25Retriever
import langchain_community.vectorstores.Chroma
import langchain_core.documents.Document
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.runnables.RunnablePassthrough
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional

Purpose: Core Python functionality: – langchain.retrievers.ContextualCompressionRetriever: Regular expression operations – langchain.retrievers.EnsembleRetriever: Regular expression operations – langchain.retrievers.document_compressors.LLMChainExtractor: Regular expression operations – langchain_community.retrievers.BM25Retriever: Regular expression operations – langchain_community.vectorstores.Chroma: Regular expression operations – langchain_core.documents.Document: Regular expression operations – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.runnables.RunnablePassthrough: Regular expression operations – os: Environment variable access and system operations – 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)

LangChain Imports

from langchain.text_splitter.RecursiveCharacterTextSplitter
from langchain_openai.ChatOpenAI
from langchain_openai.OpenAIEmbeddings

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: SemanticSearchAgent

Lines 41-225

Purpose:

Agent for semantic search and retrieval

Methods:

__init__(self, model, temperature)

Lines 44-68

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

load_documents(self, documents, metadatas)

Lines 70-97

Purpose:

Load documents into the system

Parameters: – documents: Documents – metadatas: Metadatas

semantic_search(self, query, k)

Lines 99-115

Purpose:

Pure semantic search using vector similarity

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D62A390>

hybrid_search(self, query, k)

Lines 117-144

Purpose:

Hybrid search combining semantic and keyword matching

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D77C1D0>

search_with_reranking(self, query, k)

Lines 146-172

Purpose:

Search with LLM-based reranking/compression

Parameters: – query: Input query or question – k: K

Returns: <ast.Subscript object at 0x000001A06D7EE650>

answer_question(self, question, search_method)

Lines 174-199

Purpose:

Answer question using retrieved context

Parameters: – question: Question to answer – search_method: Search method

Returns: str

find_similar_documents(self, reference_text, k)

Lines 201-217

Purpose:

Find documents similar to reference text

Parameters: – reference_text: Reference text – k: K

Returns: <ast.Subscript object at 0x000001A06D7B3890>

get_statistics(self)

Lines 219-225

Purpose:

Get search system statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7B0050>

---

Functions

demonstrate_semantic_search()

Lines 228-362

Purpose:

Demonstrate semantic search and retrieval

---

Key Concepts Summary

Pattern Overview

The Semantic Search Retrieval 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 126
• Pattern 128

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 Semantic Search Retrieval 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: 127_semantic_search_retrieval.py
Total Lines: 367
Pattern Category: AI Agent Design Pattern #127

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 126: Knowledge Fusion

Description:
    Combines knowledge from multiple heterogeneous sources, resolving conflicts
    and creating a unified, consistent knowledge representation.

Components:
    - Multi-source integration
    - Conflict detection and resolution
    - Consistency checking
    - Knowledge merging strategies

Use Cases:
    - Multi-source data integration
    - Fact verification across sources
    - Building comprehensive knowledge bases

LangChain Implementation:
    Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.
"""

import os
from typing import List, Dict, Any, Optional, Set
from dotenv import load_dotenv
from datetime import datetime
from collections import defaultdict

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


class KnowledgeSource(BaseModel):
    """Represents a knowledge source"""
    source_id: str = Field(description="Unique source identifier")
    name: str = Field(description="Source name")
    reliability: float = Field(description="Source reliability score 0-1")
    data: Dict[str, Any] = Field(description="Knowledge data")


class FusedKnowledge(BaseModel):
    """Fused knowledge item"""
    key: str = Field(description="Knowledge key/identifier")
    value: Any = Field(description="Fused value")
    confidence: float = Field(description="Confidence in fused value")
    sources: List[str] = Field(description="Contributing sources")
    conflicts: List[Dict[str, Any]] = Field(description="Detected conflicts", default_factory=list)


class KnowledgeFusionAgent:
    """Agent for fusing knowledge from multiple sources"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.sources: List[KnowledgeSource] = []
        self.fused_knowledge: Dict[str, FusedKnowledge] = {}
        
        # Conflict resolution prompt
        self.conflict_resolution_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at resolving conflicting information from multiple sources.
Given conflicting facts from different sources with their reliability scores, determine:
1. Which fact is most likely correct
2. If a synthesis/combination is possible
3. Confidence level in the resolution

Return JSON with format:
{{
  "resolved_value": "the resolved/synthesized value",
  "confidence": 0.85,
  "reasoning": "explanation of resolution",
  "method": "voting|weighted|synthesis|newest"
}}"""),
            ("user", """Conflicting information about: {key}

Sources and values:
{conflicts}

Resolve the conflict.""")
        ])
        
        self.parser = JsonOutputParser()
    
    def add_source(self, source: KnowledgeSource):
        """Add a knowledge source"""
        self.sources.append(source)
        print(f"✓ Added source: {source.name} (reliability: {source.reliability})")
    
    def detect_conflicts(self, key: str, values: List[Dict[str, Any]]) -> bool:
        """Check if there are conflicts for a key"""
        if len(values)  1
    
    def simple_fusion(self, key: str, values: List[Dict[str, Any]]) -> FusedKnowledge:
        """Simple fusion without conflicts (voting or consensus)"""
        if len(values) == 1:
            # Single source - use as is
            return FusedKnowledge(
                key=key,
                value=values[0]['value'],
                confidence=values[0]['reliability'],
                sources=[values[0]['source_id']]
            )
        
        # Multiple sources with same value - average confidence
        avg_confidence = sum(v['reliability'] for v in values) / len(values)
        sources = [v['source_id'] for v in values]
        
        return FusedKnowledge(
            key=key,
            value=values[0]['value'],
            confidence=avg_confidence,
            sources=sources
        )
    
    def resolve_conflict(self, key: str, values: List[Dict[str, Any]]) -> FusedKnowledge:
        """Resolve conflicting values using LLM"""
        print(f"\n⚠️  Conflict detected for '{key}'")
        
        # Prepare conflict description
        conflicts_text = ""
        conflict_list = []
        for v in values:
            conflicts_text += f"- Source '{v['source_name']}' (reliability: {v['reliability']}): {v['value']}\n"
            conflict_list.append({
                "source": v['source_name'],
                "value": v['value'],
                "reliability": v['reliability']
            })
        
        print(f"   Conflicting values:\n{conflicts_text}")
        
        try:
            # Use LLM to resolve
            chain = self.conflict_resolution_prompt | self.llm | self.parser
            result = chain.invoke({"key": key, "conflicts": conflicts_text})
            
            print(f"   Resolution: {result['resolved_value']}")
            print(f"   Method: {result['method']}, Confidence: {result['confidence']}")
            print(f"   Reasoning: {result['reasoning']}")
            
            return FusedKnowledge(
                key=key,
                value=result['resolved_value'],
                confidence=result['confidence'],
                sources=[v['source_id'] for v in values],
                conflicts=conflict_list
            )
        except Exception as e:
            print(f"   Error resolving conflict: {e}")
            # Fallback: use most reliable source
            best_source = max(values, key=lambda x: x['reliability'])
            return FusedKnowledge(
                key=key,
                value=best_source['value'],
                confidence=best_source['reliability'] * 0.8,  # Reduce confidence
                sources=[v['source_id'] for v in values],
                conflicts=conflict_list
            )
    
    def fuse_knowledge(self) -> Dict[str, FusedKnowledge]:
        """Fuse knowledge from all sources"""
        print("\n" + "=" * 80)
        print("🔄 Starting Knowledge Fusion Process")
        print("=" * 80)
        
        # Group knowledge by key
        knowledge_by_key = defaultdict(list)
        
        for source in self.sources:
            for key, value in source.data.items():
                knowledge_by_key[key].append({
                    'source_id': source.source_id,
                    'source_name': source.name,
                    'reliability': source.reliability,
                    'value': value
                })
        
        print(f"\n📊 Found {len(knowledge_by_key)} unique knowledge keys")
        
        # Fuse each key
        conflicts_detected = 0
        for key, values in knowledge_by_key.items():
            if self.detect_conflicts(key, values):
                conflicts_detected += 1
                self.fused_knowledge[key] = self.resolve_conflict(key, values)
            else:
                self.fused_knowledge[key] = self.simple_fusion(key, values)
        
        print(f"\n✓ Fusion complete!")
        print(f"  - Total keys: {len(self.fused_knowledge)}")
        print(f"  - Conflicts resolved: {conflicts_detected}")
        
        return self.fused_knowledge
    
    def query(self, key: str) -> Optional[FusedKnowledge]:
        """Query fused knowledge"""
        return self.fused_knowledge.get(key)
    
    def get_high_confidence_facts(self, threshold: float = 0.8) -> List[FusedKnowledge]:
        """Get facts with confidence above threshold"""
        return [
            fact for fact in self.fused_knowledge.values()
            if fact.confidence >= threshold
        ]
    
    def get_conflicted_facts(self) -> List[FusedKnowledge]:
        """Get facts that had conflicts"""
        return [
            fact for fact in self.fused_knowledge.values()
            if fact.conflicts
        ]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get fusion statistics"""
        total_facts = len(self.fused_knowledge)
        conflicted = len(self.get_conflicted_facts())
        high_confidence = len(self.get_high_confidence_facts())
        
        avg_confidence = sum(f.confidence for f in self.fused_knowledge.values()) / total_facts if total_facts > 0 else 0
        
        return {
            "total_sources": len(self.sources),
            "total_facts": total_facts,
            "conflicted_facts": conflicted,
            "high_confidence_facts": high_confidence,
            "average_confidence": avg_confidence
        }


def demonstrate_knowledge_fusion():
    """Demonstrate knowledge fusion from multiple sources"""
    print("=" * 80)
    print("Pattern 126: Knowledge Fusion")
    print("=" * 80)
    
    agent = KnowledgeFusionAgent()
    
    # Example 1: Add multiple knowledge sources
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Adding Knowledge Sources")
    print("=" * 80)
    
    # Source 1: Wikipedia (high reliability)
    wikipedia = KnowledgeSource(
        source_id="wikipedia",
        name="Wikipedia",
        reliability=0.85,
        data={
            "eiffel_tower_height": "330 meters",
            "eiffel_tower_built": "1889",
            "eiffel_tower_location": "Paris, France",
            "mount_everest_height": "8,849 meters",
            "python_creator": "Guido van Rossum"
        }
    )
    
    # Source 2: Tourist website (medium reliability)
    tourist_site = KnowledgeSource(
        source_id="tourist_site",
        name="Tourism Website",
        reliability=0.65,
        data={
            "eiffel_tower_height": "324 meters",  # Conflict!
            "eiffel_tower_built": "1889",
            "eiffel_tower_location": "Paris, France",
            "eiffel_tower_visitors": "7 million per year"
        }
    )
    
    # Source 3: News article (medium reliability)
    news = KnowledgeSource(
        source_id="news",
        name="News Article",
        reliability=0.70,
        data={
            "eiffel_tower_height": "330 meters including antenna",  # Conflict!
            "mount_everest_height": "8,848.86 meters",  # Conflict!
            "python_creator": "Guido van Rossum",
            "python_first_release": "1991"
        }
    )
    
    agent.add_source(wikipedia)
    agent.add_source(tourist_site)
    agent.add_source(news)
    
    # Example 2: Fuse knowledge
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Fusing Knowledge")
    print("=" * 80)
    
    fused = agent.fuse_knowledge()
    
    # Example 3: Query fused knowledge
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Querying Fused Knowledge")
    print("=" * 80)
    
    queries = [
        "eiffel_tower_height",
        "eiffel_tower_location",
        "python_creator",
        "mount_everest_height"
    ]
    
    for query in queries:
        result = agent.query(query)
        if result:
            print(f"\n🔍 Query: {query}")
            print(f"   Value: {result.value}")
            print(f"   Confidence: {result.confidence:.2f}")
            print(f"   Sources: {', '.join(result.sources)}")
            if result.conflicts:
                print(f"   ⚠️  Had conflicts from {len(result.conflicts)} sources")
    
    # Example 4: High confidence facts
    print("\n" + "=" * 80)
    print("EXAMPLE 4: High Confidence Facts")
    print("=" * 80)
    
    high_conf = agent.get_high_confidence_facts(threshold=0.75)
    print(f"\n✓ Found {len(high_conf)} high-confidence facts (>0.75):\n")
    for fact in high_conf:
        print(f"  - {fact.key}: {fact.value} (confidence: {fact.confidence:.2f})")
    
    # Example 5: Conflicted facts
    print("\n" + "=" * 80)
    print("EXAMPLE 5: Facts with Conflicts")
    print("=" * 80)
    
    conflicted = agent.get_conflicted_facts()
    print(f"\n⚠️  Found {len(conflicted)} facts with conflicts:\n")
    for fact in conflicted:
        print(f"  - {fact.key}:")
        print(f"      Resolved to: {fact.value}")
        print(f"      Original values:")
        for conflict in fact.conflicts:
            print(f"        • {conflict['source']}: {conflict['value']}")
    
    # Example 6: Statistics
    print("\n" + "=" * 80)
    print("EXAMPLE 6: Fusion Statistics")
    print("=" * 80)
    
    stats = agent.get_statistics()
    print(f"\n📊 Statistics:")
    print(f"  Total Sources: {stats['total_sources']}")
    print(f"  Total Facts: {stats['total_facts']}")
    print(f"  Conflicted Facts: {stats['conflicted_facts']}")
    print(f"  High Confidence Facts: {stats['high_confidence_facts']}")
    print(f"  Average Confidence: {stats['average_confidence']:.2f}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Knowledge Fusion Pattern:
- Integrates knowledge from multiple heterogeneous sources
- Detects and resolves conflicting information
- Uses source reliability scores for weighting
- Creates unified, consistent knowledge representation
- Maintains provenance and conflict information

Conflict Resolution Strategies:
✓ Voting (majority wins)
✓ Weighted by source reliability
✓ LLM-based synthesis
✓ Temporal (newest information)
✓ Consensus building

Key Benefits:
✓ Comprehensive knowledge coverage
✓ Improved accuracy through verification
✓ Conflict awareness and resolution
✓ Source tracking and attribution
✓ Confidence scoring
    """)


if __name__ == "__main__":
    demonstrate_knowledge_fusion()

    

Agentic Design Pattern: Knowledge Fusion

Pattern 126: Knowledge Fusion – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Knowledge Fusion pattern implementation.

---

Pattern Description

Pattern 126: Knowledge Fusion

Description: Combines knowledge from multiple heterogeneous sources, resolving conflicts and creating a unified, consistent knowledge representation.

Components: – Multi-source integration – Conflict detection and resolution – Consistency checking – Knowledge merging strategies

Use Cases: – Multi-source data integration – Fact verification across sources – Building comprehensive knowledge bases

LangChain Implementation: Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.

---

File Statistics

• Total Lines: 396
• Classes: 3
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 126: Knowledge Fusion

Description:
    Combines knowledge from multiple heterogeneous sources, resolving conflicts
    and creating a unified, consistent knowledge representation.

Components:
    - Multi-source integration
    - Conflict detection and resolution
    - Consistency checking
    - Knowledge merging strategies

Use Cases:
    - Multi-source data integration
    - Fact verification across sources
    - Building comprehensive knowledge bases

LangChain Implementation:
    Uses LangChain for analyzing, comparing, and merging knowledge from multiple sources.
"""

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

---

Import Statements

Standard Library Imports

import collections.defaultdict
import datetime.datetime
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Set

Purpose: Core Python functionality: – collections.defaultdict: Specialized container datatypes – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: Regular expression operations – os: Environment variable access and system operations – 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)

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: KnowledgeSource

Lines 37-42

Inherits from: BaseModel

Purpose:

Represents a knowledge source

---

Class: FusedKnowledge

Lines 45-51

Inherits from: BaseModel

Purpose:

Fused knowledge item

---

Class: KnowledgeFusionAgent

Lines 54-240

Purpose:

Agent for fusing knowledge from multiple sources

Methods:

__init__(self, model, temperature)

Lines 57-85

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

add_source(self, source)

Lines 87-90

Purpose:

Add a knowledge source

Parameters: – source: Source

detect_conflicts(self, key, values)

Lines 92-104

Purpose:

Check if there are conflicts for a key

Parameters: – key: Key – values: Values

Returns: bool

simple_fusion(self, key, values)

Lines 106-126

Purpose:

Simple fusion without conflicts (voting or consensus)

Parameters: – key: Key – values: Values

Returns: FusedKnowledge

resolve_conflict(self, key, values)

Lines 128-171

Purpose:

Resolve conflicting values using LLM

Parameters: – key: Key – values: Values

Returns: FusedKnowledge

fuse_knowledge(self)

Lines 173-206

Purpose:

Fuse knowledge from all sources

Parameters:

Returns: <ast.Subscript object at 0x000001A06D88A210>

query(self, key)

Lines 208-210

Purpose:

Query fused knowledge

Parameters: – key: Key

Returns: <ast.Subscript object at 0x000001A06D889D10>

get_high_confidence_facts(self, threshold)

Lines 212-217

Purpose:

Get facts with confidence above threshold

Parameters: – threshold: Threshold

Returns: <ast.Subscript object at 0x000001A06D64A710>

get_conflicted_facts(self)

Lines 219-224

Purpose:

Get facts that had conflicts

Parameters:

Returns: <ast.Subscript object at 0x000001A06D64BD10>

get_statistics(self)

Lines 226-240

Purpose:

Get fusion statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D748190>

---

Functions

demonstrate_knowledge_fusion()

Lines 243-391

Purpose:

Demonstrate knowledge fusion from multiple sources

---

Key Concepts Summary

Pattern Overview

The Knowledge Fusion 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 125
• Pattern 127

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 Knowledge Fusion 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: 126_knowledge_fusion.py
Total Lines: 396
Pattern Category: AI Agent Design Pattern #126

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 125: Knowledge Extraction & Mining

Description:
    Automatically extracts structured knowledge from unstructured data sources using NER,
    relation extraction, and information extraction techniques.

Components:
    - Named Entity Recognition (NER)
    - Relation extraction
    - Information extraction pipelines
    - Knowledge base building

Use Cases:
    - Building knowledge bases from documents
    - Extracting facts from text
    - Document processing and indexing

LangChain Implementation:
    Uses LangChain for entity extraction, relation identification, and knowledge structuring.
"""

import os
from typing import List, Dict, Any, Optional, Tuple
from dotenv import load_dotenv
from datetime import datetime

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

load_dotenv()


# Pydantic models for structured output
class Entity(BaseModel):
    """An entity extracted from text"""
    text: str = Field(description="The entity text")
    type: str = Field(description="Entity type (PERSON, ORG, LOCATION, DATE, etc.)")
    confidence: float = Field(description="Confidence score 0-1")


class Relation(BaseModel):
    """A relation between two entities"""
    subject: str = Field(description="Subject entity")
    predicate: str = Field(description="Relation type")
    object: str = Field(description="Object entity")
    confidence: float = Field(description="Confidence score 0-1")


class KnowledgeFact(BaseModel):
    """A structured knowledge fact"""
    entities: List[Entity] = Field(description="Entities in the fact")
    relations: List[Relation] = Field(description="Relations between entities")
    source_text: str = Field(description="Original text")
    timestamp: str = Field(description="Extraction timestamp")


class KnowledgeExtractionAgent:
    """Agent for extracting structured knowledge from text"""
    
    def __init__(self, model: str = "gpt-4o-mini", temperature: float = 0.0):
        self.llm = ChatOpenAI(model=model, temperature=temperature)
        self.knowledge_base: List[KnowledgeFact] = []
        
        # Entity extraction chain
        self.entity_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at Named Entity Recognition (NER).
Extract all entities from the text and classify them into types:
PERSON, ORGANIZATION, LOCATION, DATE, EVENT, PRODUCT, TECHNOLOGY, CONCEPT, etc.

Return JSON with format:
{{
  "entities": [
    {{"text": "entity name", "type": "TYPE", "confidence": 0.95}}
  ]
}}"""),
            ("user", "Text: {text}")
        ])
        
        # Relation extraction chain
        self.relation_prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert at extracting semantic relations between entities.
Given a text and its entities, extract meaningful relations in the form:
Subject - Predicate - Object

Common relation types:
- works_for, founded_by, located_in, part_of
- born_in, created_on, acquired_by
- invented, developed, collaborated_with

Return JSON with format:
{{
  "relations": [
    {{"subject": "Entity1", "predicate": "relation_type", "object": "Entity2", "confidence": 0.9}}
  ]
}}"""),
            ("user", "Text: {text}\n\nEntities: {entities}")
        ])
        
        self.entity_parser = JsonOutputParser()
        self.relation_parser = JsonOutputParser()
        
    def extract_entities(self, text: str) -> List[Entity]:
        """Extract named entities from text"""
        try:
            chain = self.entity_prompt | self.llm | self.entity_parser
            result = chain.invoke({"text": text})
            
            entities = []
            for e in result.get("entities", []):
                entities.append(Entity(
                    text=e["text"],
                    type=e["type"],
                    confidence=e.get("confidence", 0.8)
                ))
            return entities
        except Exception as e:
            print(f"Entity extraction error: {e}")
            return []
    
    def extract_relations(self, text: str, entities: List[Entity]) -> List[Relation]:
        """Extract relations between entities"""
        try:
            entity_str = ", ".join([f"{e.text} ({e.type})" for e in entities])
            
            chain = self.relation_prompt | self.llm | self.relation_parser
            result = chain.invoke({"text": text, "entities": entity_str})
            
            relations = []
            for r in result.get("relations", []):
                relations.append(Relation(
                    subject=r["subject"],
                    predicate=r["predicate"],
                    object=r["object"],
                    confidence=r.get("confidence", 0.8)
                ))
            return relations
        except Exception as e:
            print(f"Relation extraction error: {e}")
            return []
    
    def extract_knowledge(self, text: str) -> KnowledgeFact:
        """Extract complete knowledge from text"""
        print(f"\n🔍 Extracting knowledge from text...")
        
        # Step 1: Extract entities
        entities = self.extract_entities(text)
        print(f"\n📌 Extracted {len(entities)} entities:")
        for e in entities:
            print(f"  - {e.text} ({e.type}) - confidence: {e.confidence:.2f}")
        
        # Step 2: Extract relations
        relations = self.extract_relations(text, entities)
        print(f"\n🔗 Extracted {len(relations)} relations:")
        for r in relations:
            print(f"  - {r.subject} --[{r.predicate}]--> {r.object}")
        
        # Create knowledge fact
        fact = KnowledgeFact(
            entities=entities,
            relations=relations,
            source_text=text,
            timestamp=datetime.now().isoformat()
        )
        
        # Add to knowledge base
        self.knowledge_base.append(fact)
        
        return fact
    
    def batch_extract(self, texts: List[str]) -> List[KnowledgeFact]:
        """Extract knowledge from multiple texts"""
        print(f"\n📚 Processing {len(texts)} documents...")
        facts = []
        
        for i, text in enumerate(texts, 1):
            print(f"\n--- Document {i}/{len(texts)} ---")
            fact = self.extract_knowledge(text)
            facts.append(fact)
        
        return facts
    
    def query_knowledge_base(self, entity_name: str) -> List[Dict[str, Any]]:
        """Query knowledge base for entity"""
        results = []
        
        for fact in self.knowledge_base:
            # Check if entity appears in this fact
            for entity in fact.entities:
                if entity_name.lower() in entity.text.lower():
                    # Find related entities and relations
                    related = {
                        "entity": entity.text,
                        "type": entity.type,
                        "relations": [],
                        "source": fact.source_text[:100] + "..."
                    }
                    
                    for rel in fact.relations:
                        if entity_name.lower() in rel.subject.lower():
                            related["relations"].append({
                                "type": rel.predicate,
                                "target": rel.object
                            })
                        elif entity_name.lower() in rel.object.lower():
                            related["relations"].append({
                                "type": f"inverse_{rel.predicate}",
                                "target": rel.subject
                            })
                    
                    if related["relations"]:
                        results.append(related)
        
        return results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get knowledge base statistics"""
        total_entities = sum(len(f.entities) for f in self.knowledge_base)
        total_relations = sum(len(f.relations) for f in self.knowledge_base)
        
        entity_types = {}
        for fact in self.knowledge_base:
            for entity in fact.entities:
                entity_types[entity.type] = entity_types.get(entity.type, 0) + 1
        
        relation_types = {}
        for fact in self.knowledge_base:
            for relation in fact.relations:
                relation_types[relation.predicate] = relation_types.get(relation.predicate, 0) + 1
        
        return {
            "total_documents": len(self.knowledge_base),
            "total_entities": total_entities,
            "total_relations": total_relations,
            "entity_types": entity_types,
            "relation_types": relation_types
        }


def demonstrate_knowledge_extraction():
    """Demonstrate knowledge extraction and mining"""
    print("=" * 80)
    print("Pattern 125: Knowledge Extraction & Mining")
    print("=" * 80)
    
    agent = KnowledgeExtractionAgent()
    
    # Example 1: Extract from single document
    print("\n" + "=" * 80)
    print("EXAMPLE 1: Single Document Extraction")
    print("=" * 80)
    
    text1 = """
    OpenAI was founded in December 2015 by Sam Altman, Elon Musk, and others in San Francisco.
    The company developed ChatGPT, which was released on November 30, 2022. ChatGPT uses GPT-4,
    a large language model trained on massive amounts of text data.
    """
    
    fact1 = agent.extract_knowledge(text1)
    
    # Example 2: Batch extraction
    print("\n" + "=" * 80)
    print("EXAMPLE 2: Batch Document Processing")
    print("=" * 80)
    
    documents = [
        """Microsoft was founded by Bill Gates and Paul Allen in 1975 in Albuquerque, New Mexico.
        The company later moved to Redmond, Washington.""",
        
        """Google was founded by Larry Page and Sergey Brin in 1998 while they were Ph.D. students
        at Stanford University. The company is headquartered in Mountain View, California.""",
        
        """Apple Inc. was co-founded by Steve Jobs, Steve Wozniak, and Ronald Wayne in 1976.
        The company introduced the iPhone in 2007, revolutionizing the smartphone industry."""
    ]
    
    facts = agent.batch_extract(documents)
    
    # Example 3: Query knowledge base
    print("\n" + "=" * 80)
    print("EXAMPLE 3: Querying Knowledge Base")
    print("=" * 80)
    
    queries = ["OpenAI", "Bill Gates", "iPhone"]
    
    for query in queries:
        print(f"\n🔍 Query: '{query}'")
        results = agent.query_knowledge_base(query)
        
        if results:
            for result in results:
                print(f"\n  Entity: {result['entity']} ({result['type']})")
                print(f"  Relations:")
                for rel in result['relations']:
                    print(f"    - {rel['type']}: {rel['target']}")
        else:
            print("  No results found")
    
    # Example 4: Knowledge base statistics
    print("\n" + "=" * 80)
    print("EXAMPLE 4: Knowledge Base Statistics")
    print("=" * 80)
    
    stats = agent.get_statistics()
    
    print(f"\n📊 Statistics:")
    print(f"  Total Documents: {stats['total_documents']}")
    print(f"  Total Entities: {stats['total_entities']}")
    print(f"  Total Relations: {stats['total_relations']}")
    
    print(f"\n📌 Entity Types:")
    for entity_type, count in sorted(stats['entity_types'].items(), key=lambda x: x[1], reverse=True):
        print(f"  - {entity_type}: {count}")
    
    print(f"\n🔗 Relation Types:")
    for rel_type, count in sorted(stats['relation_types'].items(), key=lambda x: x[1], reverse=True):
        print(f"  - {rel_type}: {count}")
    
    # Summary
    print("\n" + "=" * 80)
    print("SUMMARY")
    print("=" * 80)
    print("""
Knowledge Extraction & Mining Pattern:
- Automatically extracts structured knowledge from text
- Identifies entities (people, organizations, locations, etc.)
- Discovers relations between entities
- Builds queryable knowledge base
- Enables semantic search and knowledge discovery

Key Benefits:
✓ Automated knowledge acquisition
✓ Structured representation of information
✓ Relationship discovery
✓ Scalable processing
✓ Knowledge base building
    """)


if __name__ == "__main__":
    demonstrate_knowledge_extraction()

    

Agentic Design Pattern: Knowledge Extraction Mining

Pattern 125: Knowledge Extraction Mining – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Knowledge Extraction Mining pattern implementation.

---

Pattern Description

Pattern 125: Knowledge Extraction & Mining

Description: Automatically extracts structured knowledge from unstructured data sources using NER, relation extraction, and information extraction techniques.

Components: – Named Entity Recognition (NER) – Relation extraction – Information extraction pipelines – Knowledge base building

Use Cases: – Building knowledge bases from documents – Extracting facts from text – Document processing and indexing

LangChain Implementation: Uses LangChain for entity extraction, relation identification, and knowledge structuring.

---

File Statistics

• Total Lines: 344
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 125: Knowledge Extraction & Mining

Description:
    Automatically extracts structured knowledge from unstructured data sources using NER,
    relation extraction, and information extraction techniques.

Components:
    - Named Entity Recognition (NER)
    - Relation extraction
    - Information extraction pipelines
    - Knowledge base building

Use Cases:
    - Building knowledge bases from documents
    - Extracting facts from text
    - Document processing and indexing

LangChain Implementation:
    Uses LangChain for entity extraction, relation identification, and knowledge structuring.
"""

Purpose: This module implements the Knowledge Extraction Mining pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import datetime.datetime
import langchain_core.output_parsers.JsonOutputParser
import langchain_core.prompts.ChatPromptTemplate
import langchain_core.prompts.FewShotChatMessagePromptTemplate
import langchain_core.pydantic_v1.BaseModel
import langchain_core.pydantic_v1.Field
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Optional
import typing.Tuple

Purpose: Core Python functionality: – datetime.datetime: Date and time handling – langchain_core.output_parsers.JsonOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – langchain_core.prompts.FewShotChatMessagePromptTemplate: Regular expression operations – langchain_core.pydantic_v1.BaseModel: Regular expression operations – langchain_core.pydantic_v1.Field: Regular expression operations – os: Environment variable access and system operations – 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: Entity

Lines 37-41

Inherits from: BaseModel

Purpose:

An entity extracted from text

---

Class: Relation

Lines 44-49

Inherits from: BaseModel

Purpose:

A relation between two entities

---

Class: KnowledgeFact

Lines 52-57

Inherits from: BaseModel

Purpose:

A structured knowledge fact

---

Class: KnowledgeExtractionAgent

Lines 60-239

Purpose:

Agent for extracting structured knowledge from text

Methods:

__init__(self, model, temperature)

Lines 63-103

Purpose: Initialize the class instance with required parameters.

Parameters: – model: Language model instance – temperature: Sampling temperature (0-1)

extract_entities(self, text)

Lines 105-121

Purpose:

Extract named entities from text

Parameters: – text: Text

Returns: <ast.Subscript object at 0x000001A06D7B11D0>

extract_relations(self, text, entities)

Lines 123-142

Purpose:

Extract relations between entities

Parameters: – text: Text – entities: Entities

Returns: <ast.Subscript object at 0x000001A06D812A50>

extract_knowledge(self, text)

Lines 144-171

Purpose:

Extract complete knowledge from text

Parameters: – text: Text

Returns: KnowledgeFact

batch_extract(self, texts)

Lines 173-183

Purpose:

Extract knowledge from multiple texts

Parameters: – texts: Texts

Returns: <ast.Subscript object at 0x000001A06D77E910>

query_knowledge_base(self, entity_name)

Lines 185-216

Purpose:

Query knowledge base for entity

Parameters: – entity_name: Entity name

Returns: <ast.Subscript object at 0x000001A06D63AB50>

get_statistics(self)

Lines 218-239

Purpose:

Get knowledge base statistics

Parameters:

Returns: <ast.Subscript object at 0x000001A06D7B8550>

---

Functions

demonstrate_knowledge_extraction()

Lines 242-339

Purpose:

Demonstrate knowledge extraction and mining

---

Key Concepts Summary

Pattern Overview

The Knowledge Extraction Mining 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 124
• Pattern 126

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 Knowledge Extraction Mining 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: 125_knowledge_extraction_mining.py
Total Lines: 344
Pattern Category: AI Agent Design Pattern #125

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 124: Ontology-Based Reasoning

Description:
    Uses formal ontologies (OWL, RDF, SKOS) for domain knowledge
    representation and semantic reasoning.

Components:
    - Ontology definition
    - Class hierarchies
    - Property definitions
    - Inference rules

Use Cases:
    - Semantic reasoning
    - Domain expertise
    - Formal logic

LangChain Implementation:
    Implements ontology concepts with LLM-based reasoning over formal structures.
"""

import os
from typing import List, Dict, Any, Set
from dataclasses import dataclass
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()


@dataclass
class OntologyClass:
    """Represents a class in the ontology."""
    name: str
    parent: str = None
    properties: List[str] = None
    description: str = ""
    
    def __post_init__(self):
        if self.properties is None:
            self.properties = []


@dataclass
class OntologyProperty:
    """Represents a property in the ontology."""
    name: str
    domain: str  # Class it belongs to
    range: str   # Type of value
    description: str = ""


@dataclass
class OntologyInstance:
    """Represents an instance of a class."""
    id: str
    class_name: str
    properties: Dict[str, Any]


class Ontology:
    """Formal ontology for domain knowledge."""
    
    def __init__(self, name: str):
        self.name = name
        self.classes: Dict[str, OntologyClass] = {}
        self.properties: Dict[str, OntologyProperty] = {}
        self.instances: Dict[str, OntologyInstance] = {}
        
    def add_class(self, ontology_class: OntologyClass):
        """Add a class to the ontology."""
        self.classes[ontology_class.name] = ontology_class
        
    def add_property(self, ontology_property: OntologyProperty):
        """Add a property to the ontology."""
        self.properties[ontology_property.name] = ontology_property
        
    def add_instance(self, instance: OntologyInstance):
        """Add an instance to the ontology."""
        self.instances[instance.id] = instance
        
    def get_subclasses(self, class_name: str) -> List[str]:
        """Get all subclasses of a class."""
        subclasses = []
        for cls_name, cls in self.classes.items():
            if cls.parent == class_name:
                subclasses.append(cls_name)
                # Recursively get subclasses
                subclasses.extend(self.get_subclasses(cls_name))
        return subclasses
    
    def get_superclasses(self, class_name: str) -> List[str]:
        """Get all superclasses of a class."""
        superclasses = []
        current = self.classes.get(class_name)
        while current and current.parent:
            superclasses.append(current.parent)
            current = self.classes.get(current.parent)
        return superclasses
    
    def is_subclass_of(self, class1: str, class2: str) -> bool:
        """Check if class1 is a subclass of class2."""
        return class2 in self.get_superclasses(class1)
    
    def get_instances_of_class(self, class_name: str, include_subclasses: bool = True) -> List[OntologyInstance]:
        """Get all instances of a class."""
        instances = [inst for inst in self.instances.values() if inst.class_name == class_name]
        
        if include_subclasses:
            subclasses = self.get_subclasses(class_name)
            for subclass in subclasses:
                instances.extend([inst for inst in self.instances.values() if inst.class_name == subclass])
        
        return instances


class OntologyReasoningAgent:
    """Agent that performs reasoning using ontologies."""
    
    def __init__(self, ontology: Ontology, model_name: str = "gpt-4"):
        self.ontology = ontology
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
    def classify_instance(self, instance_description: str) -> str:
        """Classify an instance into ontology classes."""
        # Get ontology structure
        ontology_structure = "Ontology Classes:\n"
        for cls_name, cls in self.ontology.classes.items():
            ontology_structure += f"- {cls_name}"
            if cls.parent:
                ontology_structure += f" (subclass of {cls.parent})"
            ontology_structure += f": {cls.description}\n"
        
        classification_prompt = ChatPromptTemplate.from_messages([
            ("system", "Classify this instance into the most appropriate ontology class."),
            ("user", """{ontology_structure}

Instance to classify: {instance}

Most appropriate class:""")
        ])
        
        chain = classification_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "ontology_structure": ontology_structure,
            "instance": instance_description
        })
    
    def infer_properties(self, instance_id: str) -> Dict[str, Any]:
        """Infer properties of an instance based on ontology."""
        instance = self.ontology.instances.get(instance_id)
        if not instance:
            return {}
        
        # Get class hierarchy
        superclasses = self.ontology.get_superclasses(instance.class_name)
        
        # Collect inherited properties
        inherited_properties = {}
        for superclass in superclasses:
            cls = self.ontology.classes.get(superclass)
            if cls:
                for prop in cls.properties:
                    if prop not in instance.properties:
                        inherited_properties[prop] = f"inherited from {superclass}"
        
        inference_prompt = ChatPromptTemplate.from_messages([
            ("system", "Infer additional properties for this instance based on ontology."),
            ("user", """Instance: {instance_id}
Class: {class_name}
Superclasses: {superclasses}
Current Properties: {current_props}

Infer likely values for these properties:
{inherited_props}""")
        ])
        
        chain = inference_prompt | self.llm | StrOutputParser()
        inferences = chain.invoke({
            "instance_id": instance_id,
            "class_name": instance.class_name,
            "superclasses": ", ".join(superclasses),
            "current_props": str(instance.properties),
            "inherited_props": ", ".join(inherited_properties.keys())
        })
        
        return {"inferences": inferences, "inherited": inherited_properties}
    
    def answer_semantic_query(self, query: str) -> str:
        """Answer semantic queries using ontology."""
        # Serialize ontology info
        ontology_info = f"Ontology: {self.ontology.name}\n\n"
        ontology_info += "Classes:\n"
        for cls_name, cls in self.ontology.classes.items():
            ontology_info += f"- {cls_name}: {cls.description}\n"
        
        ontology_info += "\nInstances:\n"
        for inst_id, inst in list(self.ontology.instances.items())[:10]:  # Limit to 10
            ontology_info += f"- {inst_id} ({inst.class_name}): {inst.properties}\n"
        
        query_prompt = ChatPromptTemplate.from_messages([
            ("system", "Answer this semantic query using the ontology knowledge."),
            ("user", """{ontology_info}

Query: {query}

Answer:""")
        ])
        
        chain = query_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "ontology_info": ontology_info,
            "query": query
        })
    
    def check_consistency(self) -> str:
        """Check ontology consistency."""
        consistency_prompt = ChatPromptTemplate.from_messages([
            ("system", "Check this ontology for logical consistency and conflicts."),
            ("user", """Ontology Structure:
{structure}

Check for:
1. Circular class hierarchies
2. Property domain/range conflicts
3. Instance classification errors
4. Missing definitions""")
        ])
        
        structure = ""
        for cls_name, cls in self.ontology.classes.items():
            structure += f"{cls_name} -> {cls.parent}\n"
        
        chain = consistency_prompt | self.llm | StrOutputParser()
        return chain.invoke({"structure": structure})


def demonstrate_ontology_based_reasoning():
    """Demonstrate ontology-based reasoning pattern."""
    print("=== Ontology-Based Reasoning Pattern ===\n")
    
    # Create ontology
    print("1. Building Domain Ontology")
    print("-" * 50)
    
    ontology = Ontology("VehicleOntology")
    
    # Define class hierarchy
    ontology.add_class(OntologyClass(
        name="Vehicle",
        description="Any mode of transportation"
    ))
    
    ontology.add_class(OntologyClass(
        name="LandVehicle",
        parent="Vehicle",
        description="Vehicles that travel on land"
    ))
    
    ontology.add_class(OntologyClass(
        name="WaterVehicle",
        parent="Vehicle",
        description="Vehicles that travel on water"
    ))
    
    ontology.add_class(OntologyClass(
        name="AirVehicle",
        parent="Vehicle",
        description="Vehicles that travel in air"
    ))
    
    ontology.add_class(OntologyClass(
        name="Car",
        parent="LandVehicle",
        properties=["num_wheels", "fuel_type", "num_doors"],
        description="Four-wheeled motor vehicle"
    ))
    
    ontology.add_class(OntologyClass(
        name="Bicycle",
        parent="LandVehicle",
        properties=["num_wheels", "has_motor"],
        description="Two-wheeled human-powered vehicle"
    ))
    
    ontology.add_class(OntologyClass(
        name="Boat",
        parent="WaterVehicle",
        properties=["propulsion_type", "max_speed"],
        description="Watercraft of any size"
    ))
    
    # Add instances
    ontology.add_instance(OntologyInstance(
        id="vehicle1",
        class_name="Car",
        properties={"num_wheels": 4, "fuel_type": "electric", "num_doors": 4}
    ))
    
    ontology.add_instance(OntologyInstance(
        id="vehicle2",
        class_name="Bicycle",
        properties={"num_wheels": 2, "has_motor": False}
    ))
    
    print(f"✓ Created ontology with {len(ontology.classes)} classes")
    print(f"✓ Added {len(ontology.instances)} instances")
    print()
    
    # Create reasoning agent
    agent = OntologyReasoningAgent(ontology)
    
    # Classification
    print("2. Instance Classification")
    print("-" * 50)
    
    test_descriptions = [
        "A motorized two-wheeler used for transportation",
        "A watercraft with sails",
        "A four-wheeled vehicle with an engine"
    ]
    
    for desc in test_descriptions:
        print(f"\nDescription: {desc}")
        classification = agent.classify_instance(desc)
        print(f"Classification: {classification}")
    
    # Property inference
    print("\n\n3. Property Inference")
    print("-" * 50)
    
    inferred = agent.infer_properties("vehicle1")
    print(f"\nInferred properties for vehicle1:")
    print(inferred)
    
    # Semantic queries
    print("\n\n4. Semantic Queries")
    print("-" * 50)
    
    queries = [
        "What are all the types of land vehicles?",
        "What properties do cars have?",
        "How are bicycles and cars related?"
    ]
    
    for query in queries:
        print(f"\nQuery: {query}")
        answer = agent.answer_semantic_query(query)
        print(f"Answer: {answer}")
    
    # Consistency check
    print("\n\n5. Ontology Consistency Check")
    print("-" * 50)
    consistency = agent.check_consistency()
    print(consistency)
    
    print("\n=== Summary ===")
    print("Ontology-based reasoning demonstrated with:")
    print("- Formal class hierarchies")
    print("- Property definitions")
    print("- Instance classification")
    print("- Property inference")
    print("- Semantic queries")
    print("- Consistency checking")


if __name__ == "__main__":
    demonstrate_ontology_based_reasoning()

    

Agentic Design Pattern: Ontology Based Reasoning

Pattern 124: Ontology Based Reasoning – Line-by-Line Explanation

Overview

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

---

Pattern Description

Pattern 124: Ontology-Based Reasoning

Description: Uses formal ontologies (OWL, RDF, SKOS) for domain knowledge representation and semantic reasoning.

Components: – Ontology definition – Class hierarchies – Property definitions – Inference rules

Use Cases: – Semantic reasoning – Domain expertise – Formal logic

LangChain Implementation: Implements ontology concepts with LLM-based reasoning over formal structures.

---

File Statistics

• Total Lines: 373
• Classes: 5
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 124: Ontology-Based Reasoning

Description:
    Uses formal ontologies (OWL, RDF, SKOS) for domain knowledge
    representation and semantic reasoning.

Components:
    - Ontology definition
    - Class hierarchies
    - Property definitions
    - Inference rules

Use Cases:
    - Semantic reasoning
    - Domain expertise
    - Formal logic

LangChain Implementation:
    Implements ontology concepts with LLM-based reasoning over formal structures.
"""

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

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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.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: OntologyClass

Lines 36-45

Purpose:

Represents a class in the ontology.

Methods:

__post_init__(self)

Lines 43-45

Purpose: Post Init operation.

Parameters:

---

Class: OntologyProperty

Lines 49-54

Purpose:

Represents a property in the ontology.

---

Class: OntologyInstance

Lines 58-62

Purpose:

Represents an instance of a class.

---

Class: Ontology

Lines 65-118

Purpose:

Formal ontology for domain knowledge.

Methods:

__init__(self, name)

Lines 68-72

Purpose: Initialize the class instance with required parameters.

Parameters: – name: Name

add_class(self, ontology_class)

Lines 74-76

Purpose:

Add a class to the ontology.

Parameters: – ontology_class: Ontology class

add_property(self, ontology_property)

Lines 78-80

Purpose:

Add a property to the ontology.

Parameters: – ontology_property: Ontology property

add_instance(self, instance)

Lines 82-84

Purpose:

Add an instance to the ontology.

Parameters: – instance: Instance

get_subclasses(self, class_name)

Lines 86-94

Purpose:

Get all subclasses of a class.

Parameters: – class_name: Class name

Returns: <ast.Subscript object at 0x000001A06D749290>

get_superclasses(self, class_name)

Lines 96-103

Purpose:

Get all superclasses of a class.

Parameters: – class_name: Class name

Returns: <ast.Subscript object at 0x000001A06D64BF50>

is_subclass_of(self, class1, class2)

Lines 105-107

Purpose:

Check if class1 is a subclass of class2.

Parameters: – class1: Class1 – class2: Class2

Returns: bool

get_instances_of_class(self, class_name, include_subclasses)

Lines 109-118

Purpose:

Get all instances of a class.

Parameters: – class_name: Class name – include_subclasses: Include subclasses

Returns: <ast.Subscript object at 0x000001A06D60FF90>

---

Class: OntologyReasoningAgent

Lines 121-239

Purpose:

Agent that performs reasoning using ontologies.

Methods:

__init__(self, ontology, model_name)

Lines 124-126

Purpose: Initialize the class instance with required parameters.

Parameters: – ontology: Ontology – model_name: Language model instance

classify_instance(self, instance_description)

Lines 128-151

Purpose:

Classify an instance into ontology classes.

Parameters: – instance_description: Instance description

Returns: str

infer_properties(self, instance_id)

Lines 153-191

Purpose:

Infer properties of an instance based on ontology.

Parameters: – instance_id: Instance id

Returns: <ast.Subscript object at 0x000001A06D825C90>

answer_semantic_query(self, query)

Lines 193-218

Purpose:

Answer semantic queries using ontology.

Parameters: – query: Input query or question

Returns: str

check_consistency(self)

Lines 220-239

Purpose:

Check ontology consistency.

Parameters:

Returns: str

---

Functions

demonstrate_ontology_based_reasoning()

Lines 242-368

Purpose:

Demonstrate ontology-based reasoning pattern.

---

Key Concepts Summary

Pattern Overview

The Ontology Based 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 123
• Pattern 125

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 Ontology Based 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: 124_ontology_based_reasoning.py
Total Lines: 373
Pattern Category: AI Agent Design Pattern #124

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 123: Knowledge Graph Integration

Description:
    Uses structured knowledge graphs for reasoning with graph traversal,
    pattern matching, and inference capabilities.

Components:
    - Knowledge graph
    - Graph traversal
    - Pattern matching
    - Inference engine

Use Cases:
    - Complex reasoning
    - Knowledge-intensive tasks
    - Relationship discovery

LangChain Implementation:
    Integrates knowledge graphs with LLM reasoning for enhanced understanding.
"""

import os
from typing import List, Dict, Any, Set, Tuple
from dataclasses import dataclass
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()


@dataclass
class Node:
    """Represents a node in the knowledge graph."""
    id: str
    type: str
    properties: Dict[str, Any]


@dataclass
class Edge:
    """Represents an edge in the knowledge graph."""
    source: str
    target: str
    relationship: str
    properties: Dict[str, Any] = None
    
    def __post_init__(self):
        if self.properties is None:
            self.properties = {}


class KnowledgeGraph:
    """Knowledge graph for structured knowledge representation."""
    
    def __init__(self):
        self.nodes: Dict[str, Node] = {}
        self.edges: List[Edge] = []
        
    def add_node(self, node: Node):
        """Add a node to the graph."""
        self.nodes[node.id] = node
        
    def add_edge(self, edge: Edge):
        """Add an edge to the graph."""
        self.edges.append(edge)
        
    def get_neighbors(self, node_id: str, relationship: str = None) -> List[Tuple[Node, str]]:
        """Get neighboring nodes."""
        neighbors = []
        for edge in self.edges:
            if edge.source == node_id:
                if relationship is None or edge.relationship == relationship:
                    if edge.target in self.nodes:
                        neighbors.append((self.nodes[edge.target], edge.relationship))
        return neighbors
    
    def find_path(self, start_id: str, end_id: str, max_depth: int = 5) -> List[str]:
        """Find path between two nodes."""
        visited = set()
        queue = [(start_id, [start_id])]
        
        while queue:
            current, path = queue.pop(0)
            
            if current == end_id:
                return path
            
            if len(path) >= max_depth:
                continue
            
            if current in visited:
                continue
            
            visited.add(current)
            
            for neighbor, _ in self.get_neighbors(current):
                if neighbor.id not in visited:
                    queue.append((neighbor.id, path + [neighbor.id]))
        
        return []
    
    def query_subgraph(self, node_id: str, depth: int = 2) -> Dict[str, Any]:
        """Extract subgraph around a node."""
        visited = set()
        subgraph_nodes = {}
        subgraph_edges = []
        
        def explore(nid, d):
            if d > depth or nid in visited:
                return
            visited.add(nid)
            
            if nid in self.nodes:
                subgraph_nodes[nid] = self.nodes[nid]
                
                for neighbor, relationship in self.get_neighbors(nid):
                    edge = Edge(nid, neighbor.id, relationship)
                    subgraph_edges.append(edge)
                    explore(neighbor.id, d + 1)
        
        explore(node_id, 0)
        
        return {
            "nodes": subgraph_nodes,
            "edges": subgraph_edges
        }


class KnowledgeGraphAgent:
    """Agent that uses knowledge graph for reasoning."""
    
    def __init__(self, kg: KnowledgeGraph, model_name: str = "gpt-4"):
        self.kg = kg
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
    def answer_question(self, question: str) -> str:
        """Answer question using knowledge graph."""
        # Extract relevant entities from question
        entity_extraction_prompt = ChatPromptTemplate.from_messages([
            ("system", "Extract key entities from this question."),
            ("user", "Question: {question}\n\nEntities (comma-separated):")
        ])
        
        chain = entity_extraction_prompt | self.llm | StrOutputParser()
        entities_str = chain.invoke({"question": question})
        entities = [e.strip() for e in entities_str.split(",")]
        
        # Find relevant nodes
        relevant_nodes = []
        for entity in entities:
            for node_id, node in self.kg.nodes.items():
                if entity.lower() in node.properties.get("name", "").lower():
                    relevant_nodes.append(node)
        
        # Extract relevant subgraph
        kg_context = ""
        for node in relevant_nodes[:3]:  # Limit to top 3
            subgraph = self.kg.query_subgraph(node.id, depth=2)
            
            # Format subgraph info
            node_info = f"\nEntity: {node.properties.get('name', node.id)}\n"
            for edge in subgraph['edges']:
                if edge.source in subgraph['nodes'] and edge.target in subgraph['nodes']:
                    source_name = subgraph['nodes'][edge.source].properties.get('name', edge.source)
                    target_name = subgraph['nodes'][edge.target].properties.get('name', edge.target)
                    node_info += f"  {source_name} --[{edge.relationship}]--> {target_name}\n"
            
            kg_context += node_info
        
        # Answer using knowledge graph context
        qa_prompt = ChatPromptTemplate.from_messages([
            ("system", "Answer the question using the knowledge graph information."),
            ("user", """Knowledge Graph Context:
{kg_context}

Question: {question}

Answer:""")
        ])
        
        chain = qa_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "kg_context": kg_context or "No relevant information found in knowledge graph",
            "question": question
        })
    
    def infer_relationships(self, entity1: str, entity2: str) -> str:
        """Infer relationships between entities."""
        # Find nodes
        node1 = None
        node2 = None
        
        for node_id, node in self.kg.nodes.items():
            if entity1.lower() in node.properties.get("name", "").lower():
                node1 = node
            if entity2.lower() in node.properties.get("name", "").lower():
                node2 = node
        
        if not node1 or not node2:
            return f"Could not find entities in knowledge graph"
        
        # Find path
        path = self.kg.find_path(node1.id, node2.id)
        
        if not path:
            return f"No direct relationship found between {entity1} and {entity2}"
        
        # Construct relationship chain
        relationships = []
        for i in range(len(path) - 1):
            for edge in self.kg.edges:
                if edge.source == path[i] and edge.target == path[i+1]:
                    source_name = self.kg.nodes[edge.source].properties.get('name', edge.source)
                    target_name = self.kg.nodes[edge.target].properties.get('name', edge.target)
                    relationships.append(f"{source_name} --[{edge.relationship}]--> {target_name}")
        
        # Generate inference
        inference_prompt = ChatPromptTemplate.from_messages([
            ("system", "Explain the relationship between entities based on this knowledge graph path."),
            ("user", """Entity 1: {entity1}
Entity 2: {entity2}

Relationship Chain:
{chain}

Explain the relationship:""")
        ])
        
        chain_str = "\n".join(relationships)
        chain = inference_prompt | self.llm | StrOutputParser()
        
        return chain.invoke({
            "entity1": entity1,
            "entity2": entity2,
            "chain": chain_str
        })
    
    def discover_patterns(self) -> str:
        """Discover patterns in the knowledge graph."""
        # Analyze graph structure
        structure_info = f"Total nodes: {len(self.kg.nodes)}\n"
        structure_info += f"Total edges: {len(self.kg.edges)}\n\n"
        
        # Common relationships
        relationship_counts = {}
        for edge in self.kg.edges:
            relationship_counts[edge.relationship] = relationship_counts.get(edge.relationship, 0) + 1
        
        structure_info += "Relationship types:\n"
        for rel, count in sorted(relationship_counts.items(), key=lambda x: x[1], reverse=True):
            structure_info += f"  {rel}: {count}\n"
        
        pattern_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze this knowledge graph structure and identify interesting patterns."),
            ("user", """Graph Structure:
{structure}

Identify:
1. Central entities (hubs)
2. Common patterns
3. Interesting insights
4. Potential gaps""")
        ])
        
        chain = pattern_prompt | self.llm | StrOutputParser()
        return chain.invoke({"structure": structure_info})


def demonstrate_knowledge_graph_integration():
    """Demonstrate knowledge graph integration pattern."""
    print("=== Knowledge Graph Integration Pattern ===\n")
    
    # Create knowledge graph
    print("1. Building Knowledge Graph")
    print("-" * 50)
    
    kg = KnowledgeGraph()
    
    # Add nodes (people, companies, technologies)
    kg.add_node(Node("person1", "Person", {"name": "Alice", "role": "CEO"}))
    kg.add_node(Node("person2", "Person", {"name": "Bob", "role": "CTO"}))
    kg.add_node(Node("person3", "Person", {"name": "Charlie", "role": "Engineer"}))
    
    kg.add_node(Node("company1", "Company", {"name": "TechCorp", "industry": "Software"}))
    kg.add_node(Node("company2", "Company", {"name": "DataInc", "industry": "Analytics"}))
    
    kg.add_node(Node("tech1", "Technology", {"name": "Python", "type": "Language"}))
    kg.add_node(Node("tech2", "Technology", {"name": "Machine Learning", "type": "Field"}))
    kg.add_node(Node("tech3", "Technology", {"name": "TensorFlow", "type": "Framework"}))
    
    # Add edges (relationships)
    kg.add_edge(Edge("person1", "company1", "WORKS_AT"))
    kg.add_edge(Edge("person2", "company1", "WORKS_AT"))
    kg.add_edge(Edge("person3", "company2", "WORKS_AT"))
    
    kg.add_edge(Edge("person1", "person2", "MANAGES"))
    kg.add_edge(Edge("person2", "person3", "MENTORS"))
    
    kg.add_edge(Edge("person2", "tech1", "EXPERT_IN"))
    kg.add_edge(Edge("person2", "tech2", "EXPERT_IN"))
    kg.add_edge(Edge("person3", "tech1", "USES"))
    kg.add_edge(Edge("person3", "tech3", "USES"))
    
    kg.add_edge(Edge("tech3", "tech1", "IMPLEMENTED_IN"))
    kg.add_edge(Edge("tech3", "tech2", "USED_FOR"))
    
    kg.add_edge(Edge("company1", "tech2", "SPECIALIZES_IN"))
    kg.add_edge(Edge("company2", "tech2", "USES"))
    
    print(f"✓ Created knowledge graph with {len(kg.nodes)} nodes and {len(kg.edges)} edges")
    print()
    
    # Create agent
    agent = KnowledgeGraphAgent(kg)
    
    # Question answering
    print("2. Question Answering with Knowledge Graph")
    print("-" * 50)
    
    questions = [
        "Who works at TechCorp?",
        "What technologies does Bob know?",
        "Which companies use Machine Learning?"
    ]
    
    for question in questions:
        print(f"\nQ: {question}")
        answer = agent.answer_question(question)
        print(f"A: {answer}")
    
    # Relationship inference
    print("\n\n3. Inferring Relationships")
    print("-" * 50)
    
    print("\nInferring relationship between Alice and Charlie:")
    inference = agent.infer_relationships("Alice", "Charlie")
    print(inference)
    
    print("\nInferring relationship between Bob and TensorFlow:")
    inference = agent.infer_relationships("Bob", "TensorFlow")
    print(inference)
    
    # Pattern discovery
    print("\n\n4. Discovering Patterns in Knowledge Graph")
    print("-" * 50)
    patterns = agent.discover_patterns()
    print(patterns)
    
    print("\n=== Summary ===")
    print("Knowledge graph integration demonstrated with:")
    print("- Graph construction (nodes and edges)")
    print("- Graph traversal and querying")
    print("- Question answering with graph context")
    print("- Relationship inference")
    print("- Pattern discovery")
    print("- LLM-enhanced graph reasoning")


if __name__ == "__main__":
    demonstrate_knowledge_graph_integration()

    

Agentic Design Pattern: Knowledge Graph Integration

Pattern 123: Knowledge Graph Integration – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Knowledge Graph Integration pattern implementation.

---

Pattern Description

Pattern 123: Knowledge Graph Integration

Description: Uses structured knowledge graphs for reasoning with graph traversal, pattern matching, and inference capabilities.

Components: – Knowledge graph – Graph traversal – Pattern matching – Inference engine

Use Cases: – Complex reasoning – Knowledge-intensive tasks – Relationship discovery

LangChain Implementation: Integrates knowledge graphs with LLM reasoning for enhanced understanding.

---

File Statistics

• Total Lines: 365
• Classes: 4
• Functions: 1
• Constants: 0

---

Module Documentation

"""
Pattern 123: Knowledge Graph Integration

Description:
    Uses structured knowledge graphs for reasoning with graph traversal,
    pattern matching, and inference capabilities.

Components:
    - Knowledge graph
    - Graph traversal
    - Pattern matching
    - Inference engine

Use Cases:
    - Complex reasoning
    - Knowledge-intensive tasks
    - Relationship discovery

LangChain Implementation:
    Integrates knowledge graphs with LLM reasoning for enhanced understanding.
"""

Purpose: This module implements the Knowledge Graph Integration pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

import dataclasses.dataclass
import langchain_core.output_parsers.StrOutputParser
import langchain_core.prompts.ChatPromptTemplate
import os
import typing.Any
import typing.Dict
import typing.List
import typing.Set
import typing.Tuple

Purpose: Core Python functionality: – dataclasses.dataclass: Simplified class definitions with @dataclass – langchain_core.output_parsers.StrOutputParser: Regular expression operations – langchain_core.prompts.ChatPromptTemplate: Regular expression operations – os: Environment variable access and system operations – 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.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: Node

Lines 36-40

Purpose:

Represents a node in the knowledge graph.

---

Class: Edge

Lines 44-53

Purpose:

Represents an edge in the knowledge graph.

Methods:

__post_init__(self)

Lines 51-53

Purpose: Post Init operation.

Parameters:

---

Class: KnowledgeGraph

Lines 56-130

Purpose:

Knowledge graph for structured knowledge representation.

Methods:

__init__(self)

Lines 59-61

Purpose: Initialize the class instance with required parameters.

Parameters:

add_node(self, node)

Lines 63-65

Purpose:

Add a node to the graph.

Parameters: – node: Node

add_edge(self, edge)

Lines 67-69

Purpose:

Add an edge to the graph.

Parameters: – edge: Edge

get_neighbors(self, node_id, relationship)

Lines 71-79

Purpose:

Get neighboring nodes.

Parameters: – node_id: Node id – relationship: Relationship

Returns: <ast.Subscript object at 0x000001A06D6EB9D0>

find_path(self, start_id, end_id, max_depth)

Lines 81-104

Purpose:

Find path between two nodes.

Parameters: – start_id: Start id – end_id: End id – max_depth: Max depth

Returns: <ast.Subscript object at 0x000001A06D63BD90>

query_subgraph(self, node_id, depth)

Lines 106-130

Purpose:

Extract subgraph around a node.

Parameters: – node_id: Node id – depth: Depth

Returns: <ast.Subscript object at 0x000001A06D646E50>

---

Class: KnowledgeGraphAgent

Lines 133-270

Purpose:

Agent that uses knowledge graph for reasoning.

Methods:

__init__(self, kg, model_name)

Lines 136-138

Purpose: Initialize the class instance with required parameters.

Parameters: – kg: Kg – model_name: Language model instance

answer_question(self, question)

Lines 140-189

Purpose:

Answer question using knowledge graph.

Parameters: – question: Question to answer

Returns: str

infer_relationships(self, entity1, entity2)

Lines 191-240

Purpose:

Infer relationships between entities.

Parameters: – entity1: Entity1 – entity2: Entity2

Returns: str

discover_patterns(self)

Lines 242-270

Purpose:

Discover patterns in the knowledge graph.

Parameters:

Returns: str

---

Functions

demonstrate_knowledge_graph_integration()

Lines 273-360

Purpose:

Demonstrate knowledge graph integration pattern.

---

Key Concepts Summary

Pattern Overview

The Knowledge Graph Integration 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 122
• Pattern 124

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 Knowledge Graph Integration 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: 123_knowledge_graph_integration.py
Total Lines: 365
Pattern Category: AI Agent Design Pattern #123

---

Generated from actual implementation code
For educational and reference purposes

    """
Pattern 122: Service Mesh Pattern

Description:
    Infrastructure layer for agent-to-agent communication with service
    discovery, load balancing, and observability.

Components:
    - Service registry
    - Load balancer
    - Circuit breaker
    - Observability layer

Use Cases:
    - Large-scale distributed agents
    - Microservices architecture
    - Reliable communication

LangChain Implementation:
    Implements service mesh concepts for agent communication management.
"""

import os
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
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 ServiceStatus(Enum):
    """Service health status."""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"


@dataclass
class ServiceInstance:
    """Represents a service instance."""
    id: str
    name: str
    host: str
    port: int
    status: ServiceStatus = ServiceStatus.HEALTHY
    load: int = 0
    max_capacity: int = 100
    response_times: List[float] = None
    
    def __post_init__(self):
        if self.response_times is None:
            self.response_times = []


class ServiceRegistry:
    """Service discovery and registration."""
    
    def __init__(self):
        self.services: Dict[str, List[ServiceInstance]] = {}
        
    def register(self, service_name: str, instance: ServiceInstance):
        """Register a service instance."""
        if service_name not in self.services:
            self.services[service_name] = []
        self.services[service_name].append(instance)
        print(f"✓ Registered {service_name} instance: {instance.id}")
    
    def deregister(self, service_name: str, instance_id: str):
        """Deregister a service instance."""
        if service_name in self.services:
            self.services[service_name] = [
                i for i in self.services[service_name] if i.id != instance_id
            ]
            print(f"✓ Deregistered {service_name} instance: {instance_id}")
    
    def discover(self, service_name: str) -> List[ServiceInstance]:
        """Discover healthy instances of a service."""
        instances = self.services.get(service_name, [])
        return [i for i in instances if i.status == ServiceStatus.HEALTHY]
    
    def get_all_services(self) -> List[str]:
        """Get list of all registered services."""
        return list(self.services.keys())


class LoadBalancer:
    """Load balancing for service requests."""
    
    def __init__(self, strategy: str = "least_loaded"):
        self.strategy = strategy
        
    def select_instance(self, instances: List[ServiceInstance]) -> Optional[ServiceInstance]:
        """Select an instance based on load balancing strategy."""
        if not instances:
            return None
        
        if self.strategy == "round_robin":
            return random.choice(instances)
        
        elif self.strategy == "least_loaded":
            return min(instances, key=lambda i: i.load)
        
        elif self.strategy == "best_performance":
            # Choose instance with best average response time
            instances_with_times = [
                i for i in instances if i.response_times
            ]
            if instances_with_times:
                return min(
                    instances_with_times,
                    key=lambda i: sum(i.response_times) / len(i.response_times)
                )
            return instances[0]
        
        return instances[0]


class CircuitBreaker:
    """Circuit breaker for fault tolerance."""
    
    def __init__(self, failure_threshold: int = 3, timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures: Dict[str, int] = {}
        self.open_circuits: Dict[str, datetime] = {}
        
    def record_failure(self, service_id: str):
        """Record a service failure."""
        self.failures[service_id] = self.failures.get(service_id, 0) + 1
        
        if self.failures[service_id] >= self.failure_threshold:
            self.open_circuits[service_id] = datetime.now()
            print(f"⚠ Circuit opened for {service_id}")
    
    def record_success(self, service_id: str):
        """Record a successful call."""
        self.failures[service_id] = 0
        if service_id in self.open_circuits:
            del self.open_circuits[service_id]
            print(f"✓ Circuit closed for {service_id}")
    
    def is_open(self, service_id: str) -> bool:
        """Check if circuit is open."""
        if service_id in self.open_circuits:
            # Check if timeout has passed
            elapsed = (datetime.now() - self.open_circuits[service_id]).seconds
            if elapsed > self.timeout:
                del self.open_circuits[service_id]
                return False
            return True
        return False


class ServiceMesh:
    """Service mesh for agent communication."""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.registry = ServiceRegistry()
        self.load_balancer = LoadBalancer(strategy="least_loaded")
        self.circuit_breaker = CircuitBreaker()
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.request_log: List[Dict[str, Any]] = []
        
    def call_service(self, service_name: str, request_data: Dict[str, Any]) -> Dict[str, Any]:
        """Make a service call through the mesh."""
        print(f"\n🔄 Calling service: {service_name}")
        
        # Discover instances
        instances = self.registry.discover(service_name)
        
        if not instances:
            print(f"  ✗ No healthy instances found")
            return {"error": "ServiceUnavailable"}
        
        # Select instance via load balancer
        instance = self.load_balancer.select_instance(instances)
        
        if not instance:
            return {"error": "NoInstanceAvailable"}
        
        # Check circuit breaker
        if self.circuit_breaker.is_open(instance.id):
            print(f"  ⚠ Circuit open for {instance.id}, trying another instance")
            # Try another instance
            instances.remove(instance)
            if instances:
                instance = self.load_balancer.select_instance(instances)
            else:
                return {"error": "AllCircuitsOpen"}
        
        print(f"  → Routing to instance: {instance.id} (load: {instance.load})")
        
        # Simulate service call
        try:
            start_time = datetime.now()
            
            # Simulate processing
            instance.load += 10
            response_time = random.uniform(0.1, 0.5)
            instance.response_times.append(response_time)
            
            # Simulate occasional failures
            if random.random()  str:
        """Generate observability report."""
        report_prompt = ChatPromptTemplate.from_messages([
            ("system", "Analyze service mesh metrics and provide insights."),
            ("user", """Request Log:
{log}

Service Registry:
{registry}

Provide:
1. Service health summary
2. Performance metrics
3. Bottlenecks identified
4. Recommendations""")
        ])
        
        log_summary = "\n".join([
            f"{r['service']}: {r['status']} ({r.get('response_time', 'N/A')}s)"
            for r in self.request_log[-20:]
        ])
        
        registry_summary = "\n".join([
            f"{name}: {len(instances)} instances"
            for name, instances in self.registry.services.items()
        ])
        
        chain = report_prompt | self.llm | StrOutputParser()
        return chain.invoke({
            "log": log_summary,
            "registry": registry_summary
        })


def demonstrate_service_mesh():
    """Demonstrate service mesh pattern."""
    print("=== Service Mesh Pattern ===\n")
    
    mesh = ServiceMesh()
    
    # Register services
    print("1. Registering Service Instances")
    print("-" * 50)
    
    # Data processing service
    mesh.registry.register("data-processor", ServiceInstance(
        id="dp-1", name="data-processor", host="10.0.1.1", port=8001
    ))
    mesh.registry.register("data-processor", ServiceInstance(
        id="dp-2", name="data-processor", host="10.0.1.2", port=8001
    ))
    mesh.registry.register("data-processor", ServiceInstance(
        id="dp-3", name="data-processor", host="10.0.1.3", port=8001
    ))
    
    # Analytics service
    mesh.registry.register("analytics", ServiceInstance(
        id="an-1", name="analytics", host="10.0.2.1", port=8002
    ))
    mesh.registry.register("analytics", ServiceInstance(
        id="an-2", name="analytics", host="10.0.2.2", port=8002
    ))
    
    print()
    
    # Make service calls
    print("2. Making Service Calls Through Mesh")
    print("-" * 50)
    
    for i in range(10):
        result = mesh.call_service("data-processor", {"data": f"request_{i}"})
    
    for i in range(5):
        result = mesh.call_service("analytics", {"query": f"query_{i}"})
    
    # Observability
    print("\n3. Service Mesh Observability")
    print("-" * 50)
    
    print(f"Total requests: {len(mesh.request_log)}")
    successful = sum(1 for r in mesh.request_log if r['status'] == 'success')
    print(f"Successful: {successful}")
    print(f"Failed: {len(mesh.request_log) - successful}")
    print(f"Success rate: {successful/len(mesh.request_log)*100:.1f}%")
    
    # Generate report
    print("\n4. Detailed Observability Report")
    print("-" * 50)
    report = mesh.get_observability_report()
    print(report)
    
    print("\n=== Summary ===")
    print("Service mesh demonstrated with:")
    print("- Service discovery and registration")
    print("- Load balancing (least loaded strategy)")
    print("- Circuit breaker pattern")
    print("- Request routing")
    print("- Health monitoring")
    print("- Observability and metrics")


if __name__ == "__main__":
    demonstrate_service_mesh()

    

Agentic Design Pattern: Service Mesh Pattern

Pattern 122: Service Mesh Pattern – Line-by-Line Explanation

Overview

This document provides a detailed line-by-line explanation of the Service Mesh Pattern pattern implementation.

---

Pattern Description

Pattern 122: Service Mesh Pattern

Description: Infrastructure layer for agent-to-agent communication with service discovery, load balancing, and observability.

Components: – Service registry – Load balancer – Circuit breaker – Observability layer

Use Cases: – Large-scale distributed agents – Microservices architecture – Reliable communication

LangChain Implementation: Implements service mesh concepts for agent communication management.

---

File Statistics

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

---

Module Documentation

"""
Pattern 122: Service Mesh Pattern

Description:
    Infrastructure layer for agent-to-agent communication with service
    discovery, load balancing, and observability.

Components:
    - Service registry
    - Load balancer
    - Circuit breaker
    - Observability layer

Use Cases:
    - Large-scale distributed agents
    - Microservices architecture
    - Reliable communication

LangChain Implementation:
    Implements service mesh concepts for agent communication management.
"""

Purpose: This module implements the Service Mesh Pattern pattern, providing a structured approach to agent design and implementation.

---

Import Statements

Standard Library Imports

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

Purpose: Core Python functionality: – dataclasses.dataclass: 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 – 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)

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: ServiceStatus

Lines 38-42

Inherits from: Enum

Purpose:

Service health status.

---

Class: ServiceInstance

Lines 46-59

Purpose:

Represents a service instance.

Methods:

__post_init__(self)

Lines 57-59

Purpose: Post Init operation.

Parameters:

---

Class: ServiceRegistry

Lines 62-90

Purpose:

Service discovery and registration.

Methods:

__init__(self)

Lines 65-66

Purpose: Initialize the class instance with required parameters.

Parameters:

register(self, service_name, instance)

Lines 68-73

Purpose:

Register a service instance.

Parameters: – service_name: Service name – instance: Instance

deregister(self, service_name, instance_id)

Lines 75-81

Purpose:

Deregister a service instance.

Parameters: – service_name: Service name – instance_id: Instance id

discover(self, service_name)

Lines 83-86

Purpose:

Discover healthy instances of a service.

Parameters: – service_name: Service name

Returns: <ast.Subscript object at 0x000001A06D718710>

get_all_services(self)

Lines 88-90

Purpose:

Get list of all registered services.

Parameters:

Returns: <ast.Subscript object at 0x000001A06D71B2D0>

---

Class: LoadBalancer

Lines 93-122

Purpose:

Load balancing for service requests.

Methods:

__init__(self, strategy)

Lines 96-97

Purpose: Initialize the class instance with required parameters.

Parameters: – strategy: Strategy

select_instance(self, instances)

Lines 99-122

Purpose:

Select an instance based on load balancing strategy.

Parameters: – instances: Instances

Returns: <ast.Subscript object at 0x000001A06D8271D0>

---

Class: CircuitBreaker

Lines 125-158

Purpose:

Circuit breaker for fault tolerance.

Methods:

__init__(self, failure_threshold, timeout)

Lines 128-132

Purpose: Initialize the class instance with required parameters.

Parameters: – failure_threshold: Failure threshold – timeout: Timeout

record_failure(self, service_id)

Lines 134-140

Purpose:

Record a service failure.

Parameters: – service_id: Service id

record_success(self, service_id)

Lines 142-147

Purpose:

Record a successful call.

Parameters: – service_id: Service id

is_open(self, service_id)

Lines 149-158

Purpose:

Check if circuit is open.

Parameters: – service_id: Service id

Returns: bool

---

Class: ServiceMesh

Lines 161-283

Purpose:

Service mesh for agent communication.

Methods:

__init__(self, model_name)

Lines 164-169

Purpose: Initialize the class instance with required parameters.

Parameters: – model_name: Language model instance

call_service(self, service_name, request_data)

Lines 171-250

Purpose:

Make a service call through the mesh.

Parameters: – service_name: Service name – request_data: Request data

Returns: <ast.Subscript object at 0x000001A06D64B650>

get_observability_report(self)

Lines 252-283

Purpose:

Generate observability report.

Parameters:

Returns: str

---

Functions

demonstrate_service_mesh()

Lines 286-350

Purpose:

Demonstrate service mesh pattern.

---

Key Concepts Summary

Pattern Overview

The Service Mesh Pattern 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 121
• Pattern 123

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 Service Mesh Pattern 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: 122_service_mesh_pattern.py
Total Lines: 355
Pattern Category: AI Agent Design Pattern #122

---

Generated from actual implementation code
For educational and reference purposes