Category: aiml-architecture

Agentic Architecture Pattern: Decision Support Agent


    """
Pattern 240: Decision Support Agent

Purpose: Multi-criteria decision analysis with risk assessment and recommendations.
Helps evaluate alternatives, assess risks, and generate actionable decisions.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re


class DecisionType(Enum):
    """Types of decisions."""
    STRATEGIC = "strategic"  # Long-term strategic decisions
    OPERATIONAL = "operational"  # Day-to-day operations
    INVESTMENT = "investment"  # Financial investments
    RISK_ASSESSMENT = "risk_assessment"  # Risk evaluation
    PRIORITIZATION = "prioritization"  # Priority ranking
    SCENARIO_ANALYSIS = "scenario_analysis"  # What-if scenarios


class RiskLevel(Enum):
    """Risk level assessment."""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class DecisionCriteria:
    """Decision criteria with weight."""
    name: str
    weight: float  # 0.0 to 1.0
    description: str = ""
    is_benefit: bool = True  # True if higher is better


@dataclass
class Alternative:
    """Decision alternative."""
    name: str
    description: str
    scores: Dict[str, float] = field(default_factory=dict)  # Criteria -> score
    cost: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class DecisionResult:
    """Decision analysis result."""
    decision_type: DecisionType
    alternatives: List[Alternative]
    criteria: List[DecisionCriteria]
    ranked_alternatives: List[tuple[str, float]] = field(default_factory=list)
    recommendation: str = ""
    risks: List[str] = field(default_factory=list)
    risk_level: RiskLevel = RiskLevel.MEDIUM
    confidence: float = 0.0
    reasoning: str = ""


class DecisionSupportAgent:
    """
    Decision Support Agent - Multi-criteria decision analysis.
    
    Features:
    - Multi-criteria evaluation
    - Alternative ranking
    - Risk assessment
    - Scenario analysis
    - AI-powered recommendations
    - Trade-off analysis
    - Sensitivity analysis
    - Decision justification
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_decisions": 0,
            "decision_types": {t.value: 0 for t in DecisionType},
            "recommendations_made": 0,
            "avg_alternatives": 0.0
        }
    
    def evaluate_decision(
        self,
        alternatives: List[Alternative],
        criteria: List[DecisionCriteria],
        decision_type: DecisionType = DecisionType.STRATEGIC
    ) -> DecisionResult:
        """Evaluate alternatives based on criteria."""
        
        # Rank alternatives
        ranked = self._rank_alternatives(alternatives, criteria)
        
        # Assess risks
        risks, risk_level = self._assess_risks(alternatives, ranked, decision_type)
        
        # Generate recommendation
        recommendation = self._generate_recommendation(ranked, risks, decision_type)
        
        # Generate reasoning
        reasoning = self._generate_reasoning(alternatives, criteria, ranked)
        
        # Calculate confidence
        confidence = self._calculate_confidence(alternatives, criteria, ranked)
        
        # Update statistics
        self.statistics["total_decisions"] += 1
        self.statistics["decision_types"][decision_type.value] += 1
        self.statistics["recommendations_made"] += 1
        
        # Update average alternatives
        prev_avg = self.statistics["avg_alternatives"]
        n = self.statistics["total_decisions"]
        self.statistics["avg_alternatives"] = (prev_avg * (n - 1) + len(alternatives)) / n
        
        return DecisionResult(
            decision_type=decision_type,
            alternatives=alternatives,
            criteria=criteria,
            ranked_alternatives=ranked,
            recommendation=recommendation,
            risks=risks,
            risk_level=risk_level,
            confidence=confidence,
            reasoning=reasoning
        )
    
    def _rank_alternatives(
        self,
        alternatives: List[Alternative],
        criteria: List[DecisionCriteria]
    ) -> List[tuple[str, float]]:
        """Rank alternatives using weighted scoring."""
        
        # Normalize weights
        total_weight = sum(c.weight for c in criteria)
        if total_weight == 0:
            total_weight = 1.0
        
        scores = {}
        
        for alt in alternatives:
            score = 0.0
            
            for criterion in criteria:
                if criterion.name in alt.scores:
                    value = alt.scores[criterion.name]
                    weight = criterion.weight / total_weight
                    
                    # Adjust for benefit/cost criteria
                    if criterion.is_benefit:
                        score += value * weight
                    else:
                        # For cost criteria, lower is better
                        score += (1.0 - value) * weight
            
            scores[alt.name] = score
        
        # Sort by score (descending)
        ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
        
        return ranked
    
    def _assess_risks(
        self,
        alternatives: List[Alternative],
        ranked: List[tuple[str, float]],
        decision_type: DecisionType
    ) -> tuple[List[str], RiskLevel]:
        """Assess risks for decision."""
        
        risks = []
        
        # Check for close alternatives (uncertainty)
        if len(ranked) >= 2:
            diff = ranked[0][1] - ranked[1][1]
            if diff  50000:
            risks.append(f"Recommended option has high cost: ${top_alt.cost:,.2f}")
        
        # Decision type specific risks
        if decision_type == DecisionType.INVESTMENT:
            risks.append("Investment decisions carry inherent market risks")
        elif decision_type == DecisionType.STRATEGIC:
            risks.append("Strategic decisions have long-term implications")
        
        # Determine risk level
        if len(risks) >= 3 or (top_alt and top_alt.cost > 100000):
            risk_level = RiskLevel.HIGH
        elif len(risks) >= 2:
            risk_level = RiskLevel.MEDIUM
        else:
            risk_level = RiskLevel.LOW
        
        return risks, risk_level
    
    def _generate_recommendation(
        self,
        ranked: List[tuple[str, float]],
        risks: List[str],
        decision_type: DecisionType
    ) -> str:
        """Generate recommendation using LLM."""
        
        if not ranked:
            return "Unable to make recommendation - no alternatives provided"
        
        top_alt = ranked[0]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are a decision support expert. Provide a clear, actionable "
                      f"recommendation for a {decision_type.value} decision."),
            ("human", f"Top Alternative: {top_alt[0]} (Score: {top_alt[1]:.3f})\n"
                     f"Identified Risks: {', '.join(risks) if risks else 'None'}\n\n"
                     f"Provide a concise recommendation (2-3 sentences):")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _generate_reasoning(
        self,
        alternatives: List[Alternative],
        criteria: List[DecisionCriteria],
        ranked: List[tuple[str, float]]
    ) -> str:
        """Generate detailed reasoning."""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Explain the decision reasoning clearly and objectively."),
            ("human", f"Decision Analysis:\n"
                     f"- Alternatives: {', '.join(a.name for a in alternatives)}\n"
                     f"- Criteria: {', '.join(c.name for c in criteria)}\n"
                     f"- Top Choice: {ranked[0][0]} ({ranked[0][1]:.3f})\n\n"
                     f"Explain why this is the best choice (3-4 sentences):")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _calculate_confidence(
        self,
        alternatives: List[Alternative],
        criteria: List[DecisionCriteria],
        ranked: List[tuple[str, float]]
    ) -> float:
        """Calculate confidence in recommendation."""
        
        if not ranked:
            return 0.0
        
        # Base confidence
        confidence = 0.7
        
        # Boost if clear leader
        if len(ranked) >= 2:
            diff = ranked[0][1] - ranked[1][1]
            if diff > 0.2:
                confidence += 0.2
            elif diff = 4:
            confidence += 0.1
        
        # Clamp to [0, 1]
        return max(0.0, min(1.0, confidence))
    
    def compare_scenarios(
        self,
        scenarios: Dict[str, List[Alternative]],
        criteria: List[DecisionCriteria]
    ) -> Dict[str, Any]:
        """Compare different scenarios."""
        
        scenario_results = {}
        
        for scenario_name, alts in scenarios.items():
            result = self.evaluate_decision(
                alts,
                criteria,
                DecisionType.SCENARIO_ANALYSIS
            )
            scenario_results[scenario_name] = {
                "top_choice": result.ranked_alternatives[0] if result.ranked_alternatives else ("None", 0.0),
                "risk_level": result.risk_level.value,
                "confidence": result.confidence
            }
        
        return scenario_results
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get decision support statistics."""
        return {
            "name": self.name,
            **self.statistics
        }


class DecisionSupportPattern:
    """
    Decision Support Pattern - Manage decision support agents.
    
    Coordinates multiple decision support agents for different contexts.
    """
    
    def __init__(self):
        self.agents: Dict[str, DecisionSupportAgent] = {}
    
    def create_agent(self, name: str) -> DecisionSupportAgent:
        """Create a new decision support agent."""
        agent = DecisionSupportAgent(name)
        self.agents[name] = agent
        print(f"   Created decision support agent: {name}")
        return agent
    
    def run(
        self,
        agent_name: str,
        alternatives: List[Alternative],
        criteria: List[DecisionCriteria],
        decision_type: DecisionType = DecisionType.STRATEGIC
    ) -> Dict[str, Any]:
        """Perform decision analysis using specified agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        result = agent.evaluate_decision(alternatives, criteria, decision_type)
        
        print(f"\n=== Decision Analysis by {agent_name} ===")
        print(f"Decision Type: {result.decision_type.value}")
        print(f"Alternatives Evaluated: {len(result.alternatives)}")
        print(f"Criteria Used: {len(result.criteria)}\n")
        
        print("Rankings:")
        for i, (name, score) in enumerate(result.ranked_alternatives, 1):
            print(f"  {i}. {name}: {score:.3f}")
        
        print(f"\nRecommendation: {result.recommendation}")
        print(f"\nReasoning: {result.reasoning}")
        
        if result.risks:
            print(f"\nRisks ({result.risk_level.value} level):")
            for risk in result.risks:
                print(f"  - {risk}")
        
        print(f"\nConfidence: {result.confidence:.1%}")
        
        return {
            "agent": agent_name,
            "decision_type": result.decision_type.value,
            "top_choice": result.ranked_alternatives[0] if result.ranked_alternatives else ("None", 0.0),
            "risk_level": result.risk_level.value,
            "confidence": result.confidence,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Decision Support Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = DecisionSupportPattern()
    
    # Create agents
    print("1. Setting up decision support agents...")
    
    advisor = pattern.create_agent("StrategicAdvisor")
    
    # Define criteria
    print("\n2. Defining decision criteria...")
    
    criteria = [
        DecisionCriteria(name="ROI", weight=0.3, description="Return on Investment", is_benefit=True),
        DecisionCriteria(name="Risk", weight=0.2, description="Risk Level", is_benefit=False),
        DecisionCriteria(name="Timeline", weight=0.2, description="Implementation Time", is_benefit=False),
        DecisionCriteria(name="Impact", weight=0.3, description="Business Impact", is_benefit=True),
    ]
    
    # Define alternatives
    print("   Defining alternatives...")
    
    alternatives = [
        Alternative(
            name="Cloud Migration",
            description="Migrate infrastructure to cloud",
            scores={"ROI": 0.85, "Risk": 0.6, "Timeline": 0.7, "Impact": 0.9},
            cost=150000
        ),
        Alternative(
            name="In-House Development",
            description="Build custom solution internally",
            scores={"ROI": 0.7, "Risk": 0.4, "Timeline": 0.3, "Impact": 0.75},
            cost=200000
        ),
        Alternative(
            name="SaaS Solution",
            description="Adopt third-party SaaS",
            scores={"ROI": 0.8, "Risk": 0.2, "Timeline": 0.9, "Impact": 0.7},
            cost=50000
        ),
    ]
    
    # Perform analysis
    print("\n3. Performing decision analysis...")
    
    result = pattern.run("StrategicAdvisor", alternatives, criteria, DecisionType.STRATEGIC)
    
    print("\n" + "="*60 + "\n")
    
    # Investment decision
    print("4. Evaluating investment decision...")
    
    investment_alternatives = [
        Alternative(
            name="Stock Portfolio",
            description="Diversified stock investments",
            scores={"ROI": 0.75, "Risk": 0.7, "Timeline": 0.5, "Impact": 0.8},
            cost=100000
        ),
        Alternative(
            name="Real Estate",
            description="Commercial property",
            scores={"ROI": 0.65, "Risk": 0.4, "Timeline": 0.2, "Impact": 0.85},
            cost=500000
        ),
        Alternative(
            name="Bonds",
            description="Government bonds",
            scores={"ROI": 0.4, "Risk": 0.1, "Timeline": 0.8, "Impact": 0.5},
            cost=75000
        ),
    ]
    
    result2 = pattern.run("StrategicAdvisor", investment_alternatives, criteria, DecisionType.INVESTMENT)
    
    # Print statistics
    print("\n" + "="*60)
    print("\n5. Decision Support Statistics:")
    stats = result2["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Multi-criteria decision analysis with risk assessment complete!")

Agentic Architecture Pattern: Decision Support Agent

Pattern 240: 240 Decision Support Agent – Line-by-Line Explanation

Overview

Pattern 240: Decision Support Agent

Purpose: Multi-criteria decision analysis with risk assessment and recommendations. Helps evaluate alternatives, assess risks, and generate actionable decisions.

Code Structure Analysis

File Information

Pattern Number: 240
Pattern Name: 240 Decision Support Agent
File Path: patterns\domain_specific\240_decision_support_agent.py
Total Lines: 469

Detailed Code Explanation

Module Docstring

"""
Pattern 240: Decision Support Agent

Purpose: Multi-criteria decision analysis with risk assessment and recommendations.
Helps evaluate alternatives, assess risks, and generate actionable decisions.
"""

Explanation: This module docstring describes the purpose and key features of the 240 Decision Support Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re  # First 20 imports shown

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

Enumerations

DecisionType (Line 18)

class DecisionType(Enum):
    STRATEGIC
    OPERATIONAL
    INVESTMENT
    RISK_ASSESSMENT
    PRIORITIZATION
    SCENARIO_ANALYSIS

Purpose: Types of decisions.

Values: – STRATEGIC: Represents strategic – OPERATIONAL: Represents operational – INVESTMENT: Represents investment – RISK_ASSESSMENT: Represents risk assessment – PRIORITIZATION: Represents prioritization – SCENARIO_ANALYSIS: Represents scenario analysis

RiskLevel (Line 28)

class RiskLevel(Enum):
    LOW
    MEDIUM
    HIGH
    CRITICAL

Purpose: Risk level assessment.

Values: – LOW: Represents low – MEDIUM: Represents medium – HIGH: Represents high – CRITICAL: Represents critical

Classes

DecisionType (Line 18)

Description: Types of decisions.

RiskLevel (Line 28)

Description: Risk level assessment.

DecisionCriteria (Line 37)

Description: Decision criteria with weight.

Alternative (Line 46)

Description: Decision alternative.

DecisionResult (Line 56)

Description: Decision analysis result.

DecisionSupportAgent (Line 69)

Description: Decision Support Agent – Multi-criteria decision analysis.

Features: – Multi-criteria evaluation – Alternative ranking – Risk assessment – Scenario analysis – AI-powered recommendations – Trade-off analysis – Sensitivity analysis – Decision justification

Methods:

`init` (Line 84)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

`evaluate_decision` (Line 96)

def evaluate_decision(self, alternatives, criteria, decision_type):
    """Evaluate alternatives based on criteria."""

Purpose: Evaluate alternatives based on criteria.

Parameters: – self: Instance reference – alternatives: Parameter for alternatives – criteria: Parameter for criteria – decision_type: Parameter for decision_type

`_rank_alternatives` (Line 141)

def _rank_alternatives(self, alternatives, criteria):
    """Rank alternatives using weighted scoring."""

Purpose: Rank alternatives using weighted scoring.

Parameters: – self: Instance reference – alternatives: Parameter for alternatives – criteria: Parameter for criteria

`_assess_risks` (Line 177)

def _assess_risks(self, alternatives, ranked, decision_type):
    """Assess risks for decision."""

Purpose: Assess risks for decision.

Parameters: – self: Instance reference – alternatives: Parameter for alternatives – ranked: Parameter for ranked – decision_type: Parameter for decision_type

`_generate_recommendation` (Line 214)

def _generate_recommendation(self, ranked, risks, decision_type):
    """Generate recommendation using LLM."""

Purpose: Generate recommendation using LLM.

Parameters: – self: Instance reference – ranked: Parameter for ranked – risks: Parameter for risks – decision_type: Parameter for decision_type

`_generate_reasoning` (Line 240)

def _generate_reasoning(self, alternatives, criteria, ranked):
    """Generate detailed reasoning."""

Purpose: Generate detailed reasoning.

Parameters: – self: Instance reference – alternatives: Parameter for alternatives – criteria: Parameter for criteria – ranked: Parameter for ranked

`_calculate_confidence` (Line 262)

def _calculate_confidence(self, alternatives, criteria, ranked):
    """Calculate confidence in recommendation."""

Purpose: Calculate confidence in recommendation.

Parameters: – self: Instance reference – alternatives: Parameter for alternatives – criteria: Parameter for criteria – ranked: Parameter for ranked

`compare_scenarios` (Line 291)

def compare_scenarios(self, scenarios, criteria):
    """Compare different scenarios."""

Purpose: Compare different scenarios.

Parameters: – self: Instance reference – scenarios: Parameter for scenarios – criteria: Parameter for criteria

`get_statistics` (Line 314)

def get_statistics(self):
    """Get decision support statistics."""

Purpose: Get decision support statistics.

Parameters: – self: Instance reference

DecisionSupportPattern (Line 322)

Description: Decision Support Pattern – Manage decision support agents.

Coordinates multiple decision support agents for different contexts.

Methods:

`init` (Line 329)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 332)

def create_agent(self, name):
    """Create a new decision support agent."""

Purpose: Create a new decision support agent.

Parameters: – self: Instance reference – name: Identifier or label

`run` (Line 339)

def run(self, agent_name, alternatives, criteria, decision_type):
    """Perform decision analysis using specified agent."""

Purpose: Perform decision analysis using specified agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – alternatives: Parameter for alternatives – criteria: Parameter for criteria – decision_type: Parameter for decision_type

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

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

November 4, 2025

Agentic Architecture Pattern: Data Analysis Agent


    """
Pattern 239: Data Analysis Agent

Purpose: Statistical data analysis with insights, visualizations, and reporting.
Performs descriptive statistics, correlation analysis, trend detection, and generates insights.
"""

from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import statistics
import re


class AnalysisType(Enum):
    """Types of data analysis."""
    DESCRIPTIVE = "descriptive"  # Mean, median, std dev
    CORRELATION = "correlation"  # Relationships between variables
    TREND = "trend"  # Time series trends
    DISTRIBUTION = "distribution"  # Data distribution
    ANOMALY = "anomaly"  # Outlier detection
    COMPARATIVE = "comparative"  # Compare groups


@dataclass
class Dataset:
    """Dataset for analysis."""
    name: str
    data: List[Dict[str, Any]]
    columns: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def __post_init__(self):
        if not self.columns and self.data:
            self.columns = list(self.data[0].keys())


@dataclass
class AnalysisResult:
    """Analysis result."""
    dataset_name: str
    analysis_type: AnalysisType
    statistics: Dict[str, Any]
    insights: List[str]
    visualizations: List[str] = field(default_factory=list)
    recommendations: List[str] = field(default_factory=list)
    confidence: float = 0.0


class DataAnalysisAgent:
    """
    Data Analysis Agent - Statistical analysis and insights.
    
    Features:
    - Descriptive statistics
    - Correlation analysis
    - Trend detection
    - Anomaly detection
    - Distribution analysis
    - AI-powered insights
    - Recommendations
    - Visualization suggestions
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_analyses": 0,
            "analysis_types": {t.value: 0 for t in AnalysisType},
            "datasets_analyzed": 0,
            "insights_generated": 0
        }
    
    def analyze(self, dataset: Dataset, analysis_type: AnalysisType = AnalysisType.DESCRIPTIVE) -> AnalysisResult:
        """Perform analysis on dataset."""
        
        # Perform statistical analysis
        if analysis_type == AnalysisType.DESCRIPTIVE:
            stats = self._descriptive_statistics(dataset)
        elif analysis_type == AnalysisType.CORRELATION:
            stats = self._correlation_analysis(dataset)
        elif analysis_type == AnalysisType.TREND:
            stats = self._trend_analysis(dataset)
        elif analysis_type == AnalysisType.DISTRIBUTION:
            stats = self._distribution_analysis(dataset)
        elif analysis_type == AnalysisType.ANOMALY:
            stats = self._anomaly_detection(dataset)
        else:  # COMPARATIVE
            stats = self._comparative_analysis(dataset)
        
        # Generate insights using LLM
        insights = self._generate_insights(dataset, stats, analysis_type)
        
        # Generate recommendations
        recommendations = self._generate_recommendations(dataset, stats, insights)
        
        # Suggest visualizations
        visualizations = self._suggest_visualizations(dataset, analysis_type)
        
        # Calculate confidence
        confidence = self._calculate_confidence(dataset, stats)
        
        # Update statistics
        self.statistics["total_analyses"] += 1
        self.statistics["analysis_types"][analysis_type.value] += 1
        self.statistics["datasets_analyzed"] = self.statistics.get("datasets_analyzed", 0) + 1
        self.statistics["insights_generated"] += len(insights)
        
        return AnalysisResult(
            dataset_name=dataset.name,
            analysis_type=analysis_type,
            statistics=stats,
            insights=insights,
            visualizations=visualizations,
            recommendations=recommendations,
            confidence=confidence
        )
    
    def _descriptive_statistics(self, dataset: Dataset) -> Dict[str, Any]:
        """Calculate descriptive statistics."""
        stats = {}
        
        for column in dataset.columns:
            # Extract numeric values
            values = []
            for row in dataset.data:
                val = row.get(column)
                if isinstance(val, (int, float)):
                    values.append(val)
            
            if values:
                stats[column] = {
                    "count": len(values),
                    "mean": statistics.mean(values),
                    "median": statistics.median(values),
                    "std_dev": statistics.stdev(values) if len(values) > 1 else 0,
                    "min": min(values),
                    "max": max(values),
                    "range": max(values) - min(values)
                }
        
        return stats
    
    def _correlation_analysis(self, dataset: Dataset) -> Dict[str, Any]:
        """Analyze correlations between variables."""
        correlations = {}
        
        # Get numeric columns
        numeric_cols = []
        for col in dataset.columns:
            values = [row.get(col) for row in dataset.data if isinstance(row.get(col), (int, float))]
            if values:
                numeric_cols.append((col, values))
        
        # Calculate pairwise correlations
        for i in range(len(numeric_cols)):
            for j in range(i + 1, len(numeric_cols)):
                col1, vals1 = numeric_cols[i]
                col2, vals2 = numeric_cols[j]
                
                if len(vals1) == len(vals2) and len(vals1) > 1:
                    corr = self._pearson_correlation(vals1, vals2)
                    correlations[f"{col1}_vs_{col2}"] = corr
        
        return {"correlations": correlations}
    
    def _trend_analysis(self, dataset: Dataset) -> Dict[str, Any]:
        """Detect trends in data."""
        trends = {}
        
        for column in dataset.columns:
            values = [row.get(column) for row in dataset.data if isinstance(row.get(column), (int, float))]
            
            if len(values) >= 3:
                # Simple trend detection
                if values[-1] > values[0]:
                    direction = "increasing"
                    change_pct = ((values[-1] - values[0]) / values[0]) * 100 if values[0] != 0 else 0
                elif values[-1]  Dict[str, Any]:
        """Analyze data distribution."""
        distributions = {}
        
        for column in dataset.columns:
            values = [row.get(column) for row in dataset.data]
            
            # Frequency distribution
            freq = defaultdict(int)
            for val in values:
                freq[str(val)] += 1
            
            # Find mode
            mode = max(freq.items(), key=lambda x: x[1]) if freq else (None, 0)
            
            distributions[column] = {
                "unique_values": len(freq),
                "mode": mode[0],
                "mode_frequency": mode[1],
                "distribution": dict(freq)
            }
        
        return {"distributions": distributions}
    
    def _anomaly_detection(self, dataset: Dataset) -> Dict[str, Any]:
        """Detect anomalies/outliers."""
        anomalies = {}
        
        for column in dataset.columns:
            values = [row.get(column) for row in dataset.data if isinstance(row.get(column), (int, float))]
            
            if len(values) > 3:
                mean = statistics.mean(values)
                std_dev = statistics.stdev(values)
                
                # Values beyond 2 standard deviations
                outliers = [v for v in values if abs(v - mean) > 2 * std_dev]
                
                anomalies[column] = {
                    "outlier_count": len(outliers),
                    "outliers": outliers[:10],  # Limit to 10
                    "threshold": 2 * std_dev
                }
        
        return {"anomalies": anomalies}
    
    def _comparative_analysis(self, dataset: Dataset) -> Dict[str, Any]:
        """Compare groups or categories."""
        comparisons = {}
        
        # Group by categorical columns and compare numeric columns
        categorical_cols = []
        numeric_cols = []
        
        for col in dataset.columns:
            sample_vals = [row.get(col) for row in dataset.data[:10]]
            if all(isinstance(v, (int, float)) for v in sample_vals if v is not None):
                numeric_cols.append(col)
            else:
                categorical_cols.append(col)
        
        if categorical_cols and numeric_cols:
            cat_col = categorical_cols[0]
            num_col = numeric_cols[0]
            
            # Group by category
            groups = defaultdict(list)
            for row in dataset.data:
                groups[str(row.get(cat_col))].append(row.get(num_col))
            
            # Calculate stats for each group
            group_stats = {}
            for group, values in groups.items():
                numeric_values = [v for v in values if isinstance(v, (int, float))]
                if numeric_values:
                    group_stats[group] = {
                        "mean": statistics.mean(numeric_values),
                        "count": len(numeric_values)
                    }
            
            comparisons[f"{cat_col}_by_{num_col}"] = group_stats
        
        return {"comparisons": comparisons}
    
    def _pearson_correlation(self, x: List[float], y: List[float]) -> float:
        """Calculate Pearson correlation coefficient."""
        if len(x) != len(y) or len(x)  List[str]:
        """Generate insights using LLM."""
        stats_summary = str(stats)[:500]  # Limit length
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are a data analysis expert. Generate 3-5 clear, actionable insights "
                      f"from the {analysis_type.value} analysis results. "
                      f"Focus on patterns, trends, and important findings."),
            ("human", f"Dataset: {dataset.name}\n\nAnalysis Results:\n{stats_summary}\n\nInsights:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse insights
        content = str(response.content)
        insights = [line.strip() for line in content.split('\n') if line.strip()]
        insights = [re.sub(r'^\d+[\.\)]\s*', '', ins) for ins in insights]
        
        return insights[:5]
    
    def _generate_recommendations(self, dataset: Dataset, stats: Dict[str, Any], insights: List[str]) -> List[str]:
        """Generate recommendations."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Based on the analysis insights, provide 2-3 actionable recommendations."),
            ("human", f"Insights:\n" + "\n".join(f"- {ins}" for ins in insights) + "\n\nRecommendations:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        content = str(response.content)
        recommendations = [line.strip() for line in content.split('\n') if line.strip()]
        recommendations = [re.sub(r'^\d+[\.\)]\s*', '', rec) for rec in recommendations]
        
        return recommendations[:3]
    
    def _suggest_visualizations(self, dataset: Dataset, analysis_type: AnalysisType) -> List[str]:
        """Suggest appropriate visualizations."""
        viz_map = {
            AnalysisType.DESCRIPTIVE: ["Box plot", "Histogram", "Summary table"],
            AnalysisType.CORRELATION: ["Scatter plot", "Correlation matrix", "Heatmap"],
            AnalysisType.TREND: ["Line chart", "Area chart", "Trend lines"],
            AnalysisType.DISTRIBUTION: ["Histogram", "Bar chart", "Pie chart"],
            AnalysisType.ANOMALY: ["Scatter plot with outliers", "Box plot", "Time series"],
            AnalysisType.COMPARATIVE: ["Bar chart", "Grouped bar chart", "Box plot comparison"]
        }
        
        return viz_map.get(analysis_type, ["Bar chart", "Line chart"])
    
    def _calculate_confidence(self, dataset: Dataset, stats: Dict[str, Any]) -> float:
        """Calculate confidence in analysis."""
        sample_size = len(dataset.data)
        
        # Base confidence on sample size
        if sample_size >= 100:
            confidence = 0.9
        elif sample_size >= 30:
            confidence = 0.8
        elif sample_size >= 10:
            confidence = 0.7
        else:
            confidence = 0.5
        
        return confidence
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get analysis statistics."""
        return {
            "name": self.name,
            **self.statistics
        }


class DataAnalysisPattern:
    """
    Data Analysis Pattern - Manage data analysis agents.
    
    Coordinates multiple analysis agents for different purposes.
    """
    
    def __init__(self):
        self.agents: Dict[str, DataAnalysisAgent] = {}
    
    def create_agent(self, name: str) -> DataAnalysisAgent:
        """Create a new data analysis agent."""
        agent = DataAnalysisAgent(name)
        self.agents[name] = agent
        print(f"   Created data analysis agent: {name}")
        return agent
    
    def run(self, agent_name: str, dataset: Dataset, analysis_type: AnalysisType = AnalysisType.DESCRIPTIVE) -> Dict[str, Any]:
        """Perform analysis using specified agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        result = agent.analyze(dataset, analysis_type)
        
        print(f"\n=== Analysis by {agent_name} ===")
        print(f"Dataset: {result.dataset_name} ({len(dataset.data)} rows)")
        print(f"Analysis Type: {result.analysis_type.value}")
        print(f"Confidence: {result.confidence:.1%}\n")
        
        print("Key Statistics:")
        for key, value in list(result.statistics.items())[:3]:
            print(f"  {key}: {str(value)[:100]}")
        
        print(f"\nInsights ({len(result.insights)}):")
        for i, insight in enumerate(result.insights, 1):
            print(f"  {i}. {insight}")
        
        if result.recommendations:
            print(f"\nRecommendations:")
            for i, rec in enumerate(result.recommendations, 1):
                print(f"  {i}. {rec}")
        
        if result.visualizations:
            print(f"\nSuggested Visualizations: {', '.join(result.visualizations)}")
        
        return {
            "agent": agent_name,
            "analysis_type": result.analysis_type.value,
            "insights_count": len(result.insights),
            "confidence": result.confidence,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Data Analysis Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = DataAnalysisPattern()
    
    # Create agents
    print("1. Setting up data analysis agents...")
    
    analyst = pattern.create_agent("DataAnalyst")
    
    # Create sample datasets
    print("\n2. Creating sample datasets...")
    
    sales_data = Dataset(
        name="Monthly Sales",
        data=[
            {"month": "Jan", "sales": 45000, "costs": 30000, "customers": 120},
            {"month": "Feb", "sales": 52000, "costs": 32000, "customers": 145},
            {"month": "Mar", "sales": 48000, "costs": 31000, "customers": 135},
            {"month": "Apr", "sales": 61000, "costs": 35000, "customers": 170},
            {"month": "May", "sales": 58000, "costs": 34000, "customers": 160},
            {"month": "Jun", "sales": 65000, "costs": 36000, "customers": 180},
        ]
    )
    
    # Perform analyses
    print("\n3. Performing analyses...")
    
    # Descriptive statistics
    result1 = pattern.run("DataAnalyst", sales_data, AnalysisType.DESCRIPTIVE)
    
    print("\n" + "="*60 + "\n")
    
    # Trend analysis
    result2 = pattern.run("DataAnalyst", sales_data, AnalysisType.TREND)
    
    print("\n" + "="*60 + "\n")
    
    # Correlation analysis
    result3 = pattern.run("DataAnalyst", sales_data, AnalysisType.CORRELATION)
    
    # Print statistics
    print("\n" + "="*60)
    print("\n4. Analysis Statistics:")
    stats = result3["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Comprehensive data analysis with AI-powered insights complete!")

Agentic Architecture Pattern: Data Analysis Agent

Pattern 239: 239 Data Analysis Agent – Line-by-Line Explanation

Overview

Pattern 239: Data Analysis Agent

Purpose: Statistical data analysis with insights, visualizations, and reporting. Performs descriptive statistics, correlation analysis, trend detection, and generates insights.

Code Structure Analysis

File Information

Pattern Number: 239
Pattern Name: 239 Data Analysis Agent
File Path: patterns\domain_specific\239_data_analysis_agent.py
Total Lines: 483

Detailed Code Explanation

Module Docstring

"""
Pattern 239: Data Analysis Agent

Purpose: Statistical data analysis with insights, visualizations, and reporting.
Performs descriptive statistics, correlation analysis, trend detection, and generates insights.
"""

Explanation: This module docstring describes the purpose and key features of the 239 Data Analysis Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import statistics
import re  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Union: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

AnalysisType (Line 19)

class AnalysisType(Enum):
    DESCRIPTIVE
    CORRELATION
    TREND
    DISTRIBUTION
    ANOMALY
    COMPARATIVE

Purpose: Types of data analysis.

Values: – DESCRIPTIVE: Represents descriptive – CORRELATION: Represents correlation – TREND: Represents trend – DISTRIBUTION: Represents distribution – ANOMALY: Represents anomaly – COMPARATIVE: Represents comparative

Classes

AnalysisType (Line 19)

Description: Types of data analysis.

Dataset (Line 30)

Description: Dataset for analysis.

Methods:

`__post_init__` (Line 37)

def __post_init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

AnalysisResult (Line 43)

Description: Analysis result.

DataAnalysisAgent (Line 54)

Description: Data Analysis Agent – Statistical analysis and insights.

Features: – Descriptive statistics – Correlation analysis – Trend detection – Anomaly detection – Distribution analysis – AI-powered insights – Recommendations – Visualization suggestions

Methods:

`init` (Line 69)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

`analyze` (Line 81)

def analyze(self, dataset, analysis_type):
    """Perform analysis on dataset."""

Purpose: Perform analysis on dataset.

Parameters: – self: Instance reference – dataset: Parameter for dataset – analysis_type: Parameter for analysis_type

`_descriptive_statistics` (Line 126)

def _descriptive_statistics(self, dataset):
    """Calculate descriptive statistics."""

Purpose: Calculate descriptive statistics.

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

`_correlation_analysis` (Line 151)

def _correlation_analysis(self, dataset):
    """Analyze correlations between variables."""

Purpose: Analyze correlations between variables.

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

`_trend_analysis` (Line 174)

def _trend_analysis(self, dataset):
    """Detect trends in data."""

Purpose: Detect trends in data.

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

`_distribution_analysis` (Line 202)

def _distribution_analysis(self, dataset):
    """Analyze data distribution."""

Purpose: Analyze data distribution.

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

`_anomaly_detection` (Line 226)

def _anomaly_detection(self, dataset):
    """Detect anomalies/outliers."""

Purpose: Detect anomalies/outliers.

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

`_comparative_analysis` (Line 248)

def _comparative_analysis(self, dataset):
    """Compare groups or categories."""

Purpose: Compare groups or categories.

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

`_pearson_correlation` (Line 286)

def _pearson_correlation(self, x, y):
    """Calculate Pearson correlation coefficient."""

Purpose: Calculate Pearson correlation coefficient.

Parameters: – self: Instance reference – x: Parameter for x – y: Parameter for y

`_generate_insights` (Line 303)

def _generate_insights(self, dataset, stats, analysis_type):
    """Generate insights using LLM."""

Purpose: Generate insights using LLM.

Parameters: – self: Instance reference – dataset: Parameter for dataset – stats: Parameter for stats – analysis_type: Parameter for analysis_type

`_generate_recommendations` (Line 324)

def _generate_recommendations(self, dataset, stats, insights):
    """Generate recommendations."""

Purpose: Generate recommendations.

Parameters: – self: Instance reference – dataset: Parameter for dataset – stats: Parameter for stats – insights: Parameter for insights

`_suggest_visualizations` (Line 340)

def _suggest_visualizations(self, dataset, analysis_type):
    """Suggest appropriate visualizations."""

Purpose: Suggest appropriate visualizations.

Parameters: – self: Instance reference – dataset: Parameter for dataset – analysis_type: Parameter for analysis_type

`_calculate_confidence` (Line 353)

def _calculate_confidence(self, dataset, stats):
    """Calculate confidence in analysis."""

Purpose: Calculate confidence in analysis.

Parameters: – self: Instance reference – dataset: Parameter for dataset – stats: Parameter for stats

`get_statistics` (Line 369)

def get_statistics(self):
    """Get analysis statistics."""

Purpose: Get analysis statistics.

Parameters: – self: Instance reference

DataAnalysisPattern (Line 377)

Description: Data Analysis Pattern – Manage data analysis agents.

Coordinates multiple analysis agents for different purposes.

Methods:

`init` (Line 384)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 387)

def create_agent(self, name):
    """Create a new data analysis agent."""

Purpose: Create a new data analysis agent.

Parameters: – self: Instance reference – name: Identifier or label

`run` (Line 394)

def run(self, agent_name, dataset, analysis_type):
    """Perform analysis using specified agent."""

Purpose: Perform analysis using specified agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – dataset: Parameter for dataset – analysis_type: Parameter for analysis_type

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Code Generation Agent


    """
Pattern 238: Code Generation Agent

Purpose: AI-powered code generation with language detection, testing, and documentation.
Generates code from natural language descriptions with quality checks and best practices.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re


class ProgrammingLanguage(Enum):
    """Supported programming languages."""
    PYTHON = "python"
    JAVASCRIPT = "javascript"
    TYPESCRIPT = "typescript"
    JAVA = "java"
    CSHARP = "csharp"
    CPP = "cpp"
    GO = "go"
    RUST = "rust"
    RUBY = "ruby"
    PHP = "php"


class CodeQuality(Enum):
    """Code quality levels."""
    QUICK = "quick"  # Fast generation, basic functionality
    STANDARD = "standard"  # Good quality with comments
    PRODUCTION = "production"  # High quality, well-documented
    ENTERPRISE = "enterprise"  # Best practices, fully tested


@dataclass
class CodeRequest:
    """Code generation request."""
    description: str
    language: ProgrammingLanguage
    quality: CodeQuality = CodeQuality.STANDARD
    include_tests: bool = False
    include_docs: bool = True
    requirements: List[str] = field(default_factory=list)


@dataclass
class GeneratedCode:
    """Generated code result."""
    request: CodeRequest
    code: str
    tests: Optional[str] = None
    documentation: Optional[str] = None
    language: ProgrammingLanguage = ProgrammingLanguage.PYTHON
    complexity_score: float = 0.0
    line_count: int = 0
    has_error_handling: bool = False
    best_practices_applied: List[str] = field(default_factory=list)


class CodeGenerationAgent:
    """
    Code Generation Agent - AI-powered code synthesis.
    
    Features:
    - Natural language to code translation
    - Multi-language support
    - Test generation
    - Documentation generation
    - Code quality levels
    - Best practices enforcement
    - Error handling
    - Complexity analysis
    """
    
    def __init__(self, name: str, default_language: ProgrammingLanguage = ProgrammingLanguage.PYTHON):
        self.name = name
        self.default_language = default_language
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.2)
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_generations": 0,
            "languages_used": {lang.value: 0 for lang in ProgrammingLanguage},
            "quality_distribution": {q.value: 0 for q in CodeQuality},
            "avg_line_count": 0,
            "tests_generated": 0
        }
    
    def generate(self, request: CodeRequest) -> GeneratedCode:
        """Generate code from request."""
        
        # Generate main code
        code = self._generate_code(request)
        
        # Generate tests if requested
        tests = None
        if request.include_tests:
            tests = self._generate_tests(code, request)
            self.statistics["tests_generated"] += 1
        
        # Generate documentation if requested
        documentation = None
        if request.include_docs:
            documentation = self._generate_documentation(code, request)
        
        # Analyze code
        line_count = len(code.split('\n'))
        complexity_score = self._calculate_complexity(code)
        has_error_handling = self._check_error_handling(code, request.language)
        best_practices = self._identify_best_practices(code, request.language)
        
        # Update statistics
        self.statistics["total_generations"] += 1
        self.statistics["languages_used"][request.language.value] += 1
        self.statistics["quality_distribution"][request.quality.value] += 1
        
        old_avg = self.statistics["avg_line_count"]
        total = self.statistics["total_generations"]
        self.statistics["avg_line_count"] = (old_avg * (total - 1) + line_count) / total
        
        return GeneratedCode(
            request=request,
            code=code,
            tests=tests,
            documentation=documentation,
            language=request.language,
            complexity_score=complexity_score,
            line_count=line_count,
            has_error_handling=has_error_handling,
            best_practices_applied=best_practices
        )
    
    def _generate_code(self, request: CodeRequest) -> str:
        """Generate code from description."""
        
        quality_instructions = {
            CodeQuality.QUICK: "Generate working code quickly. Basic functionality is sufficient.",
            CodeQuality.STANDARD: "Generate clean, readable code with inline comments.",
            CodeQuality.PRODUCTION: "Generate production-quality code with comprehensive error handling, "
                                   "logging, and documentation.",
            CodeQuality.ENTERPRISE: "Generate enterprise-grade code following all best practices, "
                                   "with complete error handling, logging, type hints, and comprehensive documentation."
        }
        
        requirements_text = ""
        if request.requirements:
            requirements_text = f"\n\nSpecific requirements:\n" + "\n".join(f"- {req}" for req in request.requirements)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an expert {request.language.value} developer. "
                      f"Generate high-quality, idiomatic code. "
                      f"{quality_instructions.get(request.quality, '')} "
                      f"Include appropriate error handling and follow best practices. "
                      f"Return ONLY the code, no explanations."),
            ("human", f"Generate {request.language.value} code for:\n\n{request.description}{requirements_text}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Extract code from response
        code = str(response.content).strip()
        code = self._extract_code_block(code)
        
        return code
    
    def _generate_tests(self, code: str, request: CodeRequest) -> str:
        """Generate unit tests for code."""
        
        test_frameworks = {
            ProgrammingLanguage.PYTHON: "pytest",
            ProgrammingLanguage.JAVASCRIPT: "Jest",
            ProgrammingLanguage.TYPESCRIPT: "Jest",
            ProgrammingLanguage.JAVA: "JUnit",
            ProgrammingLanguage.CSHARP: "xUnit",
            ProgrammingLanguage.GO: "testing package",
            ProgrammingLanguage.RUST: "Rust testing framework"
        }
        
        framework = test_frameworks.get(request.language, "standard testing framework")
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an expert in {request.language.value} testing. "
                      f"Generate comprehensive unit tests using {framework}. "
                      f"Include edge cases and error scenarios. "
                      f"Return ONLY the test code, no explanations."),
            ("human", f"Generate tests for this code:\n\n{code}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        tests = str(response.content).strip()
        tests = self._extract_code_block(tests)
        
        return tests
    
    def _generate_documentation(self, code: str, request: CodeRequest) -> str:
        """Generate documentation for code."""
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Generate clear, comprehensive documentation for {request.language.value} code. "
                      f"Include: purpose, parameters, return values, usage examples, and edge cases."),
            ("human", f"Document this code:\n\n{code}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _extract_code_block(self, text: str) -> str:
        """Extract code from markdown code blocks."""
        # Try to find code block with language specifier
        pattern = r'```(?:\w+)?\n(.*?)```'
        matches = re.findall(pattern, text, re.DOTALL)
        
        if matches:
            return matches[0].strip()
        
        # If no code block found, return as is
        return text
    
    def _calculate_complexity(self, code: str) -> float:
        """Calculate code complexity score (simplified)."""
        lines = [line for line in code.split('\n') if line.strip() and not line.strip().startswith('#')]
        
        # Count complexity indicators
        complexity = 0
        complexity += len(re.findall(r'\bif\b|\bfor\b|\bwhile\b', code)) * 2
        complexity += len(re.findall(r'\bdef\b|\bclass\b|\bfunction\b', code)) * 3
        complexity += len(re.findall(r'\btry\b|\bcatch\b|\bexcept\b', code)) * 1
        complexity += len(re.findall(r'\band\b|\bor\b|\b&&\b|\b\|\|\b', code)) * 1
        
        # Normalize by line count
        if lines:
            return min(10.0, complexity / len(lines) * 10)
        return 0.0
    
    def _check_error_handling(self, code: str, language: ProgrammingLanguage) -> bool:
        """Check if code includes error handling."""
        error_keywords = {
            ProgrammingLanguage.PYTHON: ['try', 'except', 'raise', 'finally'],
            ProgrammingLanguage.JAVASCRIPT: ['try', 'catch', 'throw', 'finally'],
            ProgrammingLanguage.JAVA: ['try', 'catch', 'throw', 'throws'],
            ProgrammingLanguage.CSHARP: ['try', 'catch', 'throw', 'finally'],
            ProgrammingLanguage.GO: ['defer', 'panic', 'recover', 'error'],
            ProgrammingLanguage.RUST: ['Result', 'Option', 'unwrap', 'expect']
        }
        
        keywords = error_keywords.get(language, ['try', 'catch', 'error'])
        return any(keyword in code for keyword in keywords)
    
    def _identify_best_practices(self, code: str, language: ProgrammingLanguage) -> List[str]:
        """Identify applied best practices."""
        practices = []
        
        # Common practices
        if re.search(r'""".*?"""', code, re.DOTALL) or re.search(r'/\*\*.*?\*/', code, re.DOTALL):
            practices.append("Documentation strings")
        
        if self._check_error_handling(code, language):
            practices.append("Error handling")
        
        if re.search(r'#.*|//.*', code):
            practices.append("Inline comments")
        
        if language == ProgrammingLanguage.PYTHON:
            if 'def ' in code and '->' in code:
                practices.append("Type hints")
            if 'logging' in code:
                practices.append("Logging")
        
        if 'test' in code.lower() or 'assert' in code:
            practices.append("Testing utilities")
        
        return practices
    
    def generate_from_description(self, description: str, language: Optional[ProgrammingLanguage] = None) -> GeneratedCode:
        """Quick generation from description."""
        request = CodeRequest(
            description=description,
            language=language or self.default_language,
            quality=CodeQuality.STANDARD,
            include_tests=False,
            include_docs=True
        )
        return self.generate(request)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get generation statistics."""
        return {
            "name": self.name,
            "default_language": self.default_language.value,
            **self.statistics
        }


class CodeGenerationPattern:
    """
    Code Generation Pattern - Manage code generation agents.
    
    Coordinates multiple agents for different languages and purposes.
    """
    
    def __init__(self):
        self.agents: Dict[str, CodeGenerationAgent] = {}
    
    def create_agent(self, name: str, default_language: ProgrammingLanguage = ProgrammingLanguage.PYTHON) -> CodeGenerationAgent:
        """Create a new code generation agent."""
        agent = CodeGenerationAgent(name, default_language)
        self.agents[name] = agent
        print(f"   Created code generation agent: {name} ({default_language.value})")
        return agent
    
    def run(self, agent_name: str, request: CodeRequest) -> Dict[str, Any]:
        """Generate code using specified agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        result = agent.generate(request)
        
        print(f"\n=== Code Generated by {agent_name} ===")
        print(f"Language: {result.language.value}")
        print(f"Quality: {result.request.quality.value}")
        print(f"Lines: {result.line_count} | Complexity: {result.complexity_score:.1f}/10")
        print(f"Error Handling: {'✓' if result.has_error_handling else '✗'}")
        
        if result.best_practices_applied:
            print(f"Best Practices: {', '.join(result.best_practices_applied)}")
        
        print(f"\nGenerated Code:\n{result.code[:300]}...")
        
        if result.tests:
            print(f"\nTests Generated: ✓ ({len(result.tests.split(chr(10)))} lines)")
        
        if result.documentation:
            print(f"\nDocumentation: {result.documentation[:150]}...")
        
        return {
            "agent": agent_name,
            "language": result.language.value,
            "line_count": result.line_count,
            "complexity": result.complexity_score,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Code Generation Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = CodeGenerationPattern()
    
    # Create agents
    print("1. Setting up code generation agents...")
    
    python_agent = pattern.create_agent("PythonCodeGen", ProgrammingLanguage.PYTHON)
    js_agent = pattern.create_agent("JSCodeGen", ProgrammingLanguage.JAVASCRIPT)
    
    # Generate code examples
    print("\n2. Generating code...")
    
    # Python function
    request1 = CodeRequest(
        description="Create a function to calculate the Fibonacci sequence up to n terms using memoization",
        language=ProgrammingLanguage.PYTHON,
        quality=CodeQuality.PRODUCTION,
        include_tests=True,
        include_docs=True
    )
    result1 = pattern.run("PythonCodeGen", request1)
    
    print("\n" + "="*60 + "\n")
    
    # JavaScript class
    request2 = CodeRequest(
        description="Create a Task class with methods to add, complete, and list tasks",
        language=ProgrammingLanguage.JAVASCRIPT,
        quality=CodeQuality.STANDARD,
        include_tests=False,
        include_docs=True
    )
    result2 = pattern.run("JSCodeGen", request2)
    
    print("\n" + "="*60 + "\n")
    
    # Python data processing
    request3 = CodeRequest(
        description="Create a function to parse CSV data and calculate statistics (mean, median, std dev)",
        language=ProgrammingLanguage.PYTHON,
        quality=CodeQuality.PRODUCTION,
        include_tests=True,
        requirements=["Handle missing values", "Return dict with statistics"]
    )
    result3 = pattern.run("PythonCodeGen", request3)
    
    # Print statistics
    print("\n" + "="*60)
    print("\n3. Code Generation Statistics:")
    stats = result3["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ AI-powered code generation with quality checks complete!")

Agentic Architecture Pattern: Code Generation Agent

Pattern 238: 238 Code Generation Agent – Line-by-Line Explanation

Overview

Pattern 238: Code Generation Agent

Purpose: AI-powered code generation with language detection, testing, and documentation. Generates code from natural language descriptions with quality checks and best practices.

Code Structure Analysis

File Information

Pattern Number: 238
Pattern Name: 238 Code Generation Agent
File Path: patterns\domain_specific\238_code_generation_agent.py
Total Lines: 412

Detailed Code Explanation

Module Docstring

"""
Pattern 238: Code Generation Agent

Purpose: AI-powered code generation with language detection, testing, and documentation.
Generates code from natural language descriptions with quality checks and best practices.
"""

Explanation: This module docstring describes the purpose and key features of the 238 Code Generation Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re  # First 20 imports shown

Enumerations

ProgrammingLanguage (Line 17)

class ProgrammingLanguage(Enum):
    PYTHON
    JAVASCRIPT
    TYPESCRIPT
    JAVA
    CSHARP
    CPP
    GO
    RUST
    RUBY
    PHP

Purpose: Supported programming languages.

Values: – PYTHON: Represents python – JAVASCRIPT: Represents javascript – TYPESCRIPT: Represents typescript – JAVA: Represents java – CSHARP: Represents csharp – CPP: Represents cpp – GO: Represents go – RUST: Represents rust – RUBY: Represents ruby – PHP: Represents php

CodeQuality (Line 31)

class CodeQuality(Enum):
    QUICK
    STANDARD
    PRODUCTION
    ENTERPRISE

Purpose: Code quality levels.

Values: – QUICK: Represents quick – STANDARD: Represents standard – PRODUCTION: Represents production – ENTERPRISE: Represents enterprise

Classes

ProgrammingLanguage (Line 17)

Description: Supported programming languages.

CodeQuality (Line 31)

Description: Code quality levels.

CodeRequest (Line 40)

Description: Code generation request.

GeneratedCode (Line 51)

Description: Generated code result.

CodeGenerationAgent (Line 64)

Description: Code Generation Agent – AI-powered code synthesis.

Features: – Natural language to code translation – Multi-language support – Test generation – Documentation generation – Code quality levels – Best practices enforcement – Error handling – Complexity analysis

Methods:

`init` (Line 79)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – default_language: Parameter for default_language

`generate` (Line 93)

def generate(self, request):
    """Generate code from request."""

Purpose: Generate code from request.

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

`_generate_code` (Line 137)

def _generate_code(self, request):
    """Generate code from description."""

Purpose: Generate code from description.

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

`_generate_tests` (Line 171)

def _generate_tests(self, code, request):
    """Generate unit tests for code."""

Purpose: Generate unit tests for code.

Parameters: – self: Instance reference – code: Parameter for code – request: Parameter for request

`_generate_documentation` (Line 202)

def _generate_documentation(self, code, request):
    """Generate documentation for code."""

Purpose: Generate documentation for code.

Parameters: – self: Instance reference – code: Parameter for code – request: Parameter for request

`_extract_code_block` (Line 216)

def _extract_code_block(self, text):
    """Extract code from markdown code blocks."""

Purpose: Extract code from markdown code blocks.

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

`_calculate_complexity` (Line 228)

def _calculate_complexity(self, code):
    """Calculate code complexity score (simplified)."""

Purpose: Calculate code complexity score (simplified).

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

`_check_error_handling` (Line 244)

def _check_error_handling(self, code, language):
    """Check if code includes error handling."""

Purpose: Check if code includes error handling.

Parameters: – self: Instance reference – code: Parameter for code – language: Parameter for language

`_identify_best_practices` (Line 258)

def _identify_best_practices(self, code, language):
    """Identify applied best practices."""

Purpose: Identify applied best practices.

Parameters: – self: Instance reference – code: Parameter for code – language: Parameter for language

`generate_from_description` (Line 283)

def generate_from_description(self, description, language):
    """Quick generation from description."""

Purpose: Quick generation from description.

Parameters: – self: Instance reference – description: Parameter for description – language: Parameter for language

`get_statistics` (Line 294)

def get_statistics(self):
    """Get generation statistics."""

Purpose: Get generation statistics.

Parameters: – self: Instance reference

CodeGenerationPattern (Line 303)

Description: Code Generation Pattern – Manage code generation agents.

Coordinates multiple agents for different languages and purposes.

Methods:

`init` (Line 310)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 313)

def create_agent(self, name, default_language):
    """Create a new code generation agent."""

Purpose: Create a new code generation agent.

Parameters: – self: Instance reference – name: Identifier or label – default_language: Parameter for default_language

`run` (Line 320)

def run(self, agent_name, request):
    """Generate code using specified agent."""

Purpose: Generate code using specified agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – request: Parameter for request

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Translation Agent


    """
Pattern 237: Translation Agent

Purpose: Multi-language translation with context preservation, quality checking, and dialect support.
Implements automatic language detection, translation memory, and terminology management.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class TranslationQuality(Enum):
    """Translation quality levels."""
    DRAFT = "draft"  # Quick, basic translation
    STANDARD = "standard"  # Good quality
    PROFESSIONAL = "professional"  # High quality
    CERTIFIED = "certified"  # Legal/official level


class TranslationStyle(Enum):
    """Translation style preferences."""
    FORMAL = "formal"
    INFORMAL = "informal"
    TECHNICAL = "technical"
    CREATIVE = "creative"
    LITERAL = "literal"


@dataclass
class TranslationMemoryEntry:
    """Entry in translation memory."""
    source_text: str
    target_text: str
    source_lang: str
    target_lang: str
    quality: TranslationQuality
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class Translation:
    """A translation result."""
    source_text: str
    translated_text: str
    source_lang: str
    target_lang: str
    quality: TranslationQuality
    style: TranslationStyle
    confidence: float = 0.0
    detected_lang: Optional[str] = None
    alternatives: List[str] = field(default_factory=list)
    preserved_terms: Dict[str, str] = field(default_factory=dict)


class TranslationAgent:
    """
    Translation Agent - Multi-language translation with quality control.
    
    Features:
    - Automatic language detection
    - Multi-language support
    - Translation memory for consistency
    - Terminology management
    - Quality levels
    - Style preferences
    - Context preservation
    - Alternative translations
    """
    
    def __init__(self, name: str):
        self.name = name
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Translation resources
        self.translation_memory: List[TranslationMemoryEntry] = []
        self.terminology: Dict[str, Dict[str, str]] = defaultdict(dict)  # term -> {lang: translation}
        self.supported_languages: Set[str] = {
            "English", "Spanish", "French", "German", "Italian", "Portuguese",
            "Chinese", "Japanese", "Korean", "Arabic", "Russian", "Hindi"
        }
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_translations": 0,
            "languages_used": defaultdict(int),
            "quality_distribution": {q.value: 0 for q in TranslationQuality},
            "memory_hits": 0
        }
    
    def translate(self,
                  text: str,
                  target_lang: str,
                  source_lang: Optional[str] = None,
                  quality: TranslationQuality = TranslationQuality.STANDARD,
                  style: TranslationStyle = TranslationStyle.FORMAL) -> Translation:
        """Translate text to target language."""
        
        # Detect source language if not provided
        if source_lang is None:
            source_lang = self._detect_language(text)
        
        # Check translation memory
        memory_match = self._check_translation_memory(text, source_lang, target_lang)
        if memory_match:
            self.statistics["memory_hits"] += 1
            return memory_match
        
        # Apply terminology
        terms_to_preserve = self._extract_terms(text, source_lang, target_lang)
        
        # Perform translation
        translated_text = self._perform_translation(
            text, source_lang, target_lang, quality, style, terms_to_preserve
        )
        
        # Generate alternatives
        alternatives = self._generate_alternatives(text, source_lang, target_lang, 2)
        
        # Calculate confidence
        confidence = self._calculate_confidence(text, translated_text, quality)
        
        # Create translation object
        translation = Translation(
            source_text=text,
            translated_text=translated_text,
            source_lang=source_lang,
            target_lang=target_lang,
            quality=quality,
            style=style,
            confidence=confidence,
            detected_lang=source_lang,
            alternatives=alternatives,
            preserved_terms=terms_to_preserve
        )
        
        # Add to translation memory
        self._add_to_memory(translation)
        
        # Update statistics
        self.statistics["total_translations"] += 1
        self.statistics["languages_used"][f"{source_lang}->{target_lang}"] += 1
        self.statistics["quality_distribution"][quality.value] += 1
        
        return translation
    
    def _detect_language(self, text: str) -> str:
        """Detect source language of text."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Detect the language of the given text. "
                      "Return only the language name (e.g., 'English', 'Spanish', 'French')."),
            ("human", f"Text: {text[:200]}\n\nLanguage:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        detected = str(response.content).strip()
        return detected if detected in self.supported_languages else "English"
    
    def _perform_translation(self,
                            text: str,
                            source_lang: str,
                            target_lang: str,
                            quality: TranslationQuality,
                            style: TranslationStyle,
                            preserve_terms: Dict[str, str]) -> str:
        """Perform the actual translation."""
        
        quality_instructions = {
            TranslationQuality.DRAFT: "Provide a quick, basic translation focusing on meaning.",
            TranslationQuality.STANDARD: "Provide an accurate translation with natural flow.",
            TranslationQuality.PROFESSIONAL: "Provide a high-quality, polished translation with attention to nuance.",
            TranslationQuality.CERTIFIED: "Provide a precise, formal translation suitable for official documents."
        }
        
        style_instructions = {
            TranslationStyle.FORMAL: "Use formal language and professional tone.",
            TranslationStyle.INFORMAL: "Use casual, conversational language.",
            TranslationStyle.TECHNICAL: "Maintain technical terminology and precision.",
            TranslationStyle.CREATIVE: "Adapt idioms and expressions creatively.",
            TranslationStyle.LITERAL: "Translate as literally as possible while maintaining sense."
        }
        
        terms_note = ""
        if preserve_terms:
            terms_list = ", ".join([f"'{k}' as '{v}'" for k, v in preserve_terms.items()])
            terms_note = f"\n\nPreserve these terms: {terms_list}"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an expert translator from {source_lang} to {target_lang}. "
                      f"{quality_instructions.get(quality, '')} "
                      f"{style_instructions.get(style, '')} "
                      f"Maintain the original meaning, tone, and context.{terms_note}"),
            ("human", f"Translate to {target_lang}:\n\n{text}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _generate_alternatives(self,
                               text: str,
                               source_lang: str,
                               target_lang: str,
                               count: int) -> List[str]:
        """Generate alternative translations."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Provide {count} alternative translations from {source_lang} to {target_lang}. "
                      f"Each alternative should capture different nuances or phrasings. "
                      f"Return only the translations, one per line."),
            ("human", f"Text: {text}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        alternatives = [line.strip() for line in str(response.content).split('\n') if line.strip()]
        return alternatives[:count]
    
    def _extract_terms(self, text: str, source_lang: str, target_lang: str) -> Dict[str, str]:
        """Extract and translate terminology."""
        preserved = {}
        
        # Check terminology database
        for term, translations in self.terminology.items():
            if term.lower() in text.lower() and target_lang in translations:
                preserved[term] = translations[target_lang]
        
        return preserved
    
    def _check_translation_memory(self,
                                  text: str,
                                  source_lang: str,
                                  target_lang: str) -> Optional[Translation]:
        """Check if translation exists in memory."""
        for entry in self.translation_memory:
            if (entry.source_text == text and
                entry.source_lang == source_lang and
                entry.target_lang == target_lang):
                
                return Translation(
                    source_text=entry.source_text,
                    translated_text=entry.target_text,
                    source_lang=entry.source_lang,
                    target_lang=entry.target_lang,
                    quality=entry.quality,
                    style=TranslationStyle.FORMAL,
                    confidence=1.0,
                    detected_lang=source_lang
                )
        
        return None
    
    def _add_to_memory(self, translation: Translation):
        """Add translation to memory."""
        entry = TranslationMemoryEntry(
            source_text=translation.source_text,
            target_text=translation.translated_text,
            source_lang=translation.source_lang,
            target_lang=translation.target_lang,
            quality=translation.quality
        )
        
        self.translation_memory.append(entry)
        
        # Limit memory size
        if len(self.translation_memory) > 1000:
            self.translation_memory = self.translation_memory[-1000:]
    
    def _calculate_confidence(self, source: str, translation: str, quality: TranslationQuality) -> float:
        """Calculate translation confidence score."""
        # Simple heuristic (in production, use back-translation or BLEU score)
        base_confidence = {
            TranslationQuality.DRAFT: 0.6,
            TranslationQuality.STANDARD: 0.8,
            TranslationQuality.PROFESSIONAL: 0.9,
            TranslationQuality.CERTIFIED: 0.95
        }
        
        confidence = base_confidence.get(quality, 0.8)
        
        # Adjust based on length similarity
        len_ratio = len(translation) / len(source) if len(source) > 0 else 1.0
        if 0.5 <= len_ratio  List[Translation]:
        """Translate multiple texts."""
        return [self.translate(text, target_lang, source_lang) for text in texts]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get translation statistics."""
        return {
            "name": self.name,
            "supported_languages": len(self.supported_languages),
            "memory_size": len(self.translation_memory),
            "terminology_entries": len(self.terminology),
            **self.statistics
        }


class TranslationPattern:
    """
    Translation Pattern - Manage translation agents.
    
    Coordinates multiple translation agents for different purposes.
    """
    
    def __init__(self):
        self.agents: Dict[str, TranslationAgent] = {}
    
    def create_agent(self, name: str) -> TranslationAgent:
        """Create a new translation agent."""
        agent = TranslationAgent(name)
        self.agents[name] = agent
        print(f"   Created translation agent: {name}")
        return agent
    
    def run(self,
            agent_name: str,
            text: str,
            target_lang: str,
            source_lang: Optional[str] = None,
            quality: TranslationQuality = TranslationQuality.STANDARD) -> Dict[str, Any]:
        """Perform translation using specified agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        translation = agent.translate(text, target_lang, source_lang, quality)
        
        print(f"\n=== Translation by {agent_name} ===")
        print(f"Source ({translation.source_lang}): {translation.source_text[:100]}...")
        print(f"Target ({translation.target_lang}): {translation.translated_text}")
        print(f"Quality: {translation.quality.value} | Confidence: {translation.confidence:.1%}")
        
        if translation.alternatives:
            print("\nAlternatives:")
            for i, alt in enumerate(translation.alternatives, 1):
                print(f"  {i}. {alt}")
        
        if translation.preserved_terms:
            print(f"\nPreserved Terms: {', '.join(translation.preserved_terms.keys())}")
        
        return {
            "agent": agent_name,
            "source_lang": translation.source_lang,
            "target_lang": translation.target_lang,
            "confidence": translation.confidence,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Translation Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = TranslationPattern()
    
    # Create agents
    print("1. Setting up translation agents...")
    
    universal_translator = pattern.create_agent("UniversalTranslator")
    technical_translator = pattern.create_agent("TechnicalTranslator")
    
    # Add terminology
    print("\n2. Adding terminology...")
    universal_translator.add_terminology("machine learning", {
        "Spanish": "aprendizaje automático",
        "French": "apprentissage automatique",
        "German": "maschinelles Lernen"
    })
    
    # Perform translations
    print("\n3. Performing translations...")
    
    text1 = "Artificial intelligence is transforming the way we live and work."
    result1 = pattern.run("UniversalTranslator", text1, "Spanish", quality=TranslationQuality.PROFESSIONAL)
    
    print("\n" + "="*60 + "\n")
    
    text2 = "Machine learning algorithms analyze data to make predictions."
    result2 = pattern.run("UniversalTranslator", text2, "French", source_lang="English")
    
    print("\n" + "="*60 + "\n")
    
    text3 = "The quick brown fox jumps over the lazy dog."
    result3 = pattern.run("UniversalTranslator", text3, "German", quality=TranslationQuality.STANDARD)
    
    # Print statistics
    print("\n" + "="*60)
    print("\n4. Translation Statistics:")
    stats = result3["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Multi-language translation with quality control complete!")

Agentic Architecture Pattern: Translation Agent

Pattern 237: 237 Translation Agent – Line-by-Line Explanation

Overview

Pattern 237: Translation Agent

Purpose: Multi-language translation with context preservation, quality checking, and dialect support. Implements automatic language detection, translation memory, and terminology management.

Code Structure Analysis

File Information

Pattern Number: 237
Pattern Name: 237 Translation Agent
File Path: patterns\domain_specific\237_translation_agent.py
Total Lines: 412

Detailed Code Explanation

Module Docstring

"""
Pattern 237: Translation Agent

Purpose: Multi-language translation with context preservation, quality checking, and dialect support.
Implements automatic language detection, translation memory, and terminology management.
"""

Explanation: This module docstring describes the purpose and key features of the 237 Translation Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

TranslationQuality (Line 17)

class TranslationQuality(Enum):
    DRAFT
    STANDARD
    PROFESSIONAL
    CERTIFIED

Purpose: Translation quality levels.

Values: – DRAFT: Represents draft – STANDARD: Represents standard – PROFESSIONAL: Represents professional – CERTIFIED: Represents certified

TranslationStyle (Line 25)

class TranslationStyle(Enum):
    FORMAL
    INFORMAL
    TECHNICAL
    CREATIVE
    LITERAL

Purpose: Translation style preferences.

Values: – FORMAL: Represents formal – INFORMAL: Represents informal – TECHNICAL: Represents technical – CREATIVE: Represents creative – LITERAL: Represents literal

Classes

TranslationQuality (Line 17)

Description: Translation quality levels.

TranslationStyle (Line 25)

Description: Translation style preferences.

TranslationMemoryEntry (Line 35)

Description: Entry in translation memory.

Translation (Line 46)

Description: A translation result.

TranslationAgent (Line 60)

Description: Translation Agent – Multi-language translation with quality control.

Features: – Automatic language detection – Multi-language support – Translation memory for consistency – Terminology management – Quality levels – Style preferences – Context preservation – Alternative translations

Methods:

`init` (Line 75)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label

`translate` (Line 95)

def translate(self, text, target_lang, source_lang, quality, style):
    """Translate text to target language."""

Purpose: Translate text to target language.

Parameters: – self: Instance reference – text: Parameter for text – target_lang: Parameter for target_lang – source_lang: Parameter for source_lang – quality: Parameter for quality – style: Parameter for style

`_detect_language` (Line 151)

def _detect_language(self, text):
    """Detect source language of text."""

Purpose: Detect source language of text.

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

`_perform_translation` (Line 165)

def _perform_translation(self, text, source_lang, target_lang, quality, style, preserve_terms):
    """Perform the actual translation."""

Purpose: Perform the actual translation.

Parameters: – self: Instance reference – text: Parameter for text – source_lang: Parameter for source_lang – target_lang: Parameter for target_lang – quality: Parameter for quality – style: Parameter for style – preserve_terms: Parameter for preserve_terms

`_generate_alternatives` (Line 207)

def _generate_alternatives(self, text, source_lang, target_lang, count):
    """Generate alternative translations."""

Purpose: Generate alternative translations.

Parameters: – self: Instance reference – text: Parameter for text – source_lang: Parameter for source_lang – target_lang: Parameter for target_lang – count: Parameter for count

`_extract_terms` (Line 226)

def _extract_terms(self, text, source_lang, target_lang):
    """Extract and translate terminology."""

Purpose: Extract and translate terminology.

Parameters: – self: Instance reference – text: Parameter for text – source_lang: Parameter for source_lang – target_lang: Parameter for target_lang

`_check_translation_memory` (Line 237)

def _check_translation_memory(self, text, source_lang, target_lang):
    """Check if translation exists in memory."""

Purpose: Check if translation exists in memory.

Parameters: – self: Instance reference – text: Parameter for text – source_lang: Parameter for source_lang – target_lang: Parameter for target_lang

`_add_to_memory` (Line 260)

def _add_to_memory(self, translation):
    """Add translation to memory."""

Purpose: Add translation to memory.

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

`_calculate_confidence` (Line 276)

def _calculate_confidence(self, source, translation, quality):
    """Calculate translation confidence score."""

Purpose: Calculate translation confidence score.

Parameters: – self: Instance reference – source: Parameter for source – translation: Parameter for translation – quality: Parameter for quality

`add_terminology` (Line 295)

def add_terminology(self, term, translations):
    """Add term to terminology database."""

Purpose: Add term to terminology database.

Parameters: – self: Instance reference – term: Parameter for term – translations: Parameter for translations

`batch_translate` (Line 299)

def batch_translate(self, texts, target_lang, source_lang):
    """Translate multiple texts."""

Purpose: Translate multiple texts.

Parameters: – self: Instance reference – texts: Parameter for texts – target_lang: Parameter for target_lang – source_lang: Parameter for source_lang

`get_statistics` (Line 303)

def get_statistics(self):
    """Get translation statistics."""

Purpose: Get translation statistics.

Parameters: – self: Instance reference

TranslationPattern (Line 314)

Description: Translation Pattern – Manage translation agents.

Coordinates multiple translation agents for different purposes.

Methods:

`init` (Line 321)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 324)

def create_agent(self, name):
    """Create a new translation agent."""

Purpose: Create a new translation agent.

Parameters: – self: Instance reference – name: Identifier or label

`run` (Line 331)

def run(self, agent_name, text, target_lang, source_lang, quality):
    """Perform translation using specified agent."""

Purpose: Perform translation using specified agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – text: Parameter for text – target_lang: Parameter for target_lang – source_lang: Parameter for source_lang – quality: Parameter for quality

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Summarization Agent


    """
Pattern 236: Summarization Agent

Purpose: Intelligent text summarization with multiple strategies and customizable outputs.
Implements extractive, abstractive, and hybrid summarization approaches.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re


class SummarizationType(Enum):
    """Types of summarization strategies."""
    EXTRACTIVE = "extractive"  # Select key sentences
    ABSTRACTIVE = "abstractive"  # Generate new text
    HYBRID = "hybrid"  # Combination
    BULLET_POINTS = "bullet_points"  # Key points as bullets
    HEADLINE = "headline"  # Short headline/title


class SummaryLength(Enum):
    """Target summary lengths."""
    BRIEF = "brief"  # 1-2 sentences
    SHORT = "short"  # 3-5 sentences
    MEDIUM = "medium"  # 1-2 paragraphs
    DETAILED = "detailed"  # 3+ paragraphs


@dataclass
class Document:
    """Document to summarize."""
    id: str
    title: str
    content: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    word_count: int = 0
    
    def __post_init__(self):
        if self.word_count == 0:
            self.word_count = len(self.content.split())


@dataclass
class Summary:
    """Generated summary."""
    original_doc_id: str
    summary_text: str
    summary_type: SummarizationType
    length: SummaryLength
    key_points: List[str] = field(default_factory=list)
    compression_ratio: float = 0.0
    important_entities: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)


class SummarizationAgent:
    """
    Summarization Agent - Intelligent text summarization.
    
    Features:
    - Multiple summarization strategies (extractive, abstractive, hybrid)
    - Customizable summary lengths
    - Key point extraction
    - Bullet point generation
    - Headline creation
    - Entity extraction
    - Compression ratio tracking
    """
    
    def __init__(self,
                 name: str,
                 summary_type: SummarizationType = SummarizationType.HYBRID,
                 default_length: SummaryLength = SummaryLength.MEDIUM):
        self.name = name
        self.summary_type = summary_type
        self.default_length = default_length
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_summaries": 0,
            "avg_compression_ratio": 0.0,
            "summary_type_usage": {t.value: 0 for t in SummarizationType}
        }
    
    def summarize(self,
                  document: Document,
                  summary_type: Optional[SummarizationType] = None,
                  length: Optional[SummaryLength] = None) -> Summary:
        """Generate summary for document."""
        summary_type = summary_type or self.summary_type
        length = length or self.default_length
        
        # Generate summary based on type
        if summary_type == SummarizationType.EXTRACTIVE:
            summary_text = self._extractive_summary(document, length)
        elif summary_type == SummarizationType.ABSTRACTIVE:
            summary_text = self._abstractive_summary(document, length)
        elif summary_type == SummarizationType.BULLET_POINTS:
            summary_text = self._bullet_point_summary(document, length)
        elif summary_type == SummarizationType.HEADLINE:
            summary_text = self._headline_summary(document)
        else:  # HYBRID
            summary_text = self._hybrid_summary(document, length)
        
        # Extract key points
        key_points = self._extract_key_points(document, 5)
        
        # Extract important entities
        entities = self._extract_entities(document)
        
        # Calculate compression ratio
        summary_words = len(summary_text.split())
        compression_ratio = summary_words / document.word_count if document.word_count > 0 else 0
        
        # Update statistics
        self.statistics["total_summaries"] += 1
        self.statistics["summary_type_usage"][summary_type.value] += 1
        
        # Update average compression ratio
        old_avg = self.statistics["avg_compression_ratio"]
        total = self.statistics["total_summaries"]
        self.statistics["avg_compression_ratio"] = (old_avg * (total - 1) + compression_ratio) / total
        
        return Summary(
            original_doc_id=document.id,
            summary_text=summary_text,
            summary_type=summary_type,
            length=length,
            key_points=key_points,
            compression_ratio=compression_ratio,
            important_entities=entities,
            metadata={
                "original_word_count": document.word_count,
                "summary_word_count": summary_words,
                "title": document.title
            }
        )
    
    def _extractive_summary(self, document: Document, length: SummaryLength) -> str:
        """Extractive summarization - select important sentences."""
        # Split into sentences
        sentences = self._split_sentences(document.content)
        
        if not sentences:
            return document.content
        
        # Score sentences
        sentence_scores = self._score_sentences(sentences, document)
        
        # Determine number of sentences based on length
        num_sentences = self._get_sentence_count(length, len(sentences))
        
        # Select top sentences
        top_sentences = sorted(sentence_scores.items(), key=lambda x: x[1], reverse=True)[:num_sentences]
        
        # Sort by original order
        selected_indices = sorted([sentences.index(sent) for sent, _ in top_sentences])
        selected_sentences = [sentences[i] for i in selected_indices]
        
        return " ".join(selected_sentences)
    
    def _abstractive_summary(self, document: Document, length: SummaryLength) -> str:
        """Abstractive summarization - generate new text."""
        length_instructions = {
            SummaryLength.BRIEF: "in 1-2 sentences",
            SummaryLength.SHORT: "in 3-5 sentences",
            SummaryLength.MEDIUM: "in 1-2 concise paragraphs",
            SummaryLength.DETAILED: "in 3-4 detailed paragraphs"
        }
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an expert summarization agent. Create a clear, accurate summary "
                      f"that captures the main ideas and key information. "
                      f"Summarize {length_instructions.get(length, 'concisely')}."),
            ("human", f"Title: {document.title}\n\nContent:\n{document.content}\n\n"
                     f"Provide a {length.value} summary:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _bullet_point_summary(self, document: Document, length: SummaryLength) -> str:
        """Generate bullet point summary."""
        num_points = {
            SummaryLength.BRIEF: 3,
            SummaryLength.SHORT: 5,
            SummaryLength.MEDIUM: 7,
            SummaryLength.DETAILED: 10
        }
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an expert at creating concise bullet point summaries. "
                      f"Extract the most important information as clear, actionable bullet points."),
            ("human", f"Title: {document.title}\n\nContent:\n{document.content}\n\n"
                     f"Create {num_points.get(length, 5)} key bullet points:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _headline_summary(self, document: Document) -> str:
        """Generate headline-style summary."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert at creating compelling, accurate headlines. "
                      "Create a concise headline (5-12 words) that captures the essence of the content."),
            ("human", f"Content:\n{document.content}\n\nGenerate headline:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _hybrid_summary(self, document: Document, length: SummaryLength) -> str:
        """Hybrid summarization - combine extractive and abstractive."""
        # First, get key sentences (extractive)
        extractive_summary = self._extractive_summary(document, length)
        
        # Then, refine with abstractive approach
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are refining a summary. Improve the flow and clarity while "
                      "maintaining all key information. Keep it concise and well-structured."),
            ("human", f"Original Title: {document.title}\n\n"
                     f"Key sentences extracted:\n{extractive_summary}\n\n"
                     f"Refine into a {length.value} summary:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content).strip()
    
    def _extract_key_points(self, document: Document, max_points: int = 5) -> List[str]:
        """Extract key points from document."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"Extract the top {max_points} key points from the content. "
                      f"Each point should be a clear, concise statement (one sentence)."),
            ("human", f"Content:\n{document.content}\n\nKey points:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse response into list
        content = str(response.content)
        points = [line.strip() for line in content.split('\n') if line.strip()]
        
        # Remove numbering if present
        points = [re.sub(r'^\d+[\.\)]\s*', '', point) for point in points]
        
        return points[:max_points]
    
    def _extract_entities(self, document: Document) -> List[str]:
        """Extract important entities (names, places, concepts)."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Extract the most important entities from the content: "
                      "people, places, organizations, key concepts. "
                      "Return as a comma-separated list."),
            ("human", f"Content:\n{document.content[:500]}\n\nEntities:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse entities
        content = str(response.content)
        entities = [e.strip() for e in content.split(',') if e.strip()]
        
        return entities[:10]
    
    def _split_sentences(self, text: str) -> List[str]:
        """Split text into sentences."""
        # Simple sentence splitting (in production, use proper NLP)
        sentences = re.split(r'[.!?]+', text)
        sentences = [s.strip() for s in sentences if s.strip() and len(s.strip()) > 10]
        return sentences
    
    def _score_sentences(self, sentences: List[str], document: Document) -> Dict[str, float]:
        """Score sentences for importance."""
        scores = {}
        
        # Get word frequencies
        words = re.findall(r'\w+', document.content.lower())
        word_freq = {}
        for word in words:
            if len(word) > 3:  # Ignore short words
                word_freq[word] = word_freq.get(word, 0) + 1
        
        # Score each sentence
        for sentence in sentences:
            score = 0.0
            sentence_words = re.findall(r'\w+', sentence.lower())
            
            # Word frequency score
            for word in sentence_words:
                score += word_freq.get(word, 0)
            
            # Position bonus (earlier sentences often more important)
            position = sentences.index(sentence)
            position_bonus = 1.0 / (position + 1)
            score += position_bonus * 10
            
            # Length penalty (very short or very long sentences)
            word_count = len(sentence_words)
            if 10 <= word_count  int:
        """Determine number of sentences based on target length."""
        ratios = {
            SummaryLength.BRIEF: 0.1,
            SummaryLength.SHORT: 0.2,
            SummaryLength.MEDIUM: 0.35,
            SummaryLength.DETAILED: 0.5
        }
        
        count = int(total_sentences * ratios.get(length, 0.2))
        return max(1, min(count, total_sentences))
    
    def batch_summarize(self, documents: List[Document]) -> List[Summary]:
        """Summarize multiple documents."""
        return [self.summarize(doc) for doc in documents]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get summarization statistics."""
        return {
            "name": self.name,
            "default_strategy": self.summary_type.value,
            "default_length": self.default_length.value,
            **self.statistics
        }


class SummarizationPattern:
    """
    Summarization Pattern - Manage summarization agents.
    
    Coordinates multiple summarization agents for different needs.
    """
    
    def __init__(self):
        self.agents: Dict[str, SummarizationAgent] = {}
    
    def create_agent(self,
                     name: str,
                     summary_type: SummarizationType = SummarizationType.HYBRID,
                     default_length: SummaryLength = SummaryLength.MEDIUM) -> SummarizationAgent:
        """Create a new summarization agent."""
        agent = SummarizationAgent(name, summary_type, default_length)
        self.agents[name] = agent
        print(f"   Created summarization agent: {name} ({summary_type.value}, {default_length.value})")
        return agent
    
    def run(self,
            agent_name: str,
            document: Document,
            summary_type: Optional[SummarizationType] = None) -> Dict[str, Any]:
        """Generate summary using specified agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        summary = agent.summarize(document, summary_type=summary_type)
        
        print(f"\n=== Summary by {agent_name} ===")
        print(f"Document: {document.title}")
        print(f"Original: {document.word_count} words")
        print(f"Summary Type: {summary.summary_type.value}")
        print(f"Compression: {summary.compression_ratio:.1%}\n")
        print(f"Summary:\n{summary.summary_text}\n")
        
        if summary.key_points:
            print("Key Points:")
            for i, point in enumerate(summary.key_points, 1):
                print(f"  {i}. {point}")
        
        if summary.important_entities:
            print(f"\nImportant Entities: {', '.join(summary.important_entities[:5])}")
        
        return {
            "agent": agent_name,
            "document_id": document.id,
            "summary_length": len(summary.summary_text.split()),
            "compression_ratio": summary.compression_ratio,
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Summarization Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = SummarizationPattern()
    
    # Create agents
    print("1. Setting up summarization agents...")
    
    news_agent = pattern.create_agent("NewsDigest", SummarizationType.ABSTRACTIVE, SummaryLength.SHORT)
    research_agent = pattern.create_agent("ResearchSummarizer", SummarizationType.HYBRID, SummaryLength.MEDIUM)
    quick_agent = pattern.create_agent("QuickScan", SummarizationType.BULLET_POINTS, SummaryLength.BRIEF)
    
    # Create sample documents
    print("\n2. Creating sample documents...")
    
    doc1 = Document(
        id="doc1",
        title="Advances in Artificial Intelligence",
        content="""
        Artificial intelligence has made remarkable progress in recent years, transforming industries 
        and daily life. Machine learning algorithms now power everything from recommendation systems 
        to autonomous vehicles. Deep learning, a subset of machine learning, has achieved breakthrough 
        results in image recognition, natural language processing, and game playing. Neural networks 
        with millions of parameters can now understand context, generate human-like text, and even 
        create original artwork. However, challenges remain in areas such as explainability, bias, 
        and ethical considerations. Researchers are working on making AI systems more transparent 
        and fair. The future of AI promises even more sophisticated applications, including advanced 
        medical diagnosis, climate modeling, and personalized education. As AI continues to evolve, 
        collaboration between technologists, ethicists, and policymakers becomes increasingly important 
        to ensure responsible development and deployment.
        """
    )
    
    doc2 = Document(
        id="doc2",
        title="Climate Change and Renewable Energy",
        content="""
        Climate change poses one of the greatest challenges of our time. Rising global temperatures 
        are causing more frequent extreme weather events, sea level rise, and ecosystem disruption. 
        The transition to renewable energy sources is critical for reducing greenhouse gas emissions. 
        Solar and wind power have become increasingly cost-competitive with fossil fuels. Many countries 
        are setting ambitious targets for carbon neutrality by 2050. Battery technology improvements 
        are making energy storage more viable. Electric vehicles are gaining market share rapidly. 
        However, the transition requires significant infrastructure investment and policy support. 
        Individual actions also matter, from reducing consumption to supporting sustainable businesses.
        """
    )
    
    # Generate summaries
    print("\n3. Generating summaries...")
    
    # Abstractive summary
    result1 = pattern.run("NewsDigest", doc1)
    
    print("\n" + "="*60 + "\n")
    
    # Hybrid summary
    result2 = pattern.run("ResearchSummarizer", doc2)
    
    print("\n" + "="*60 + "\n")
    
    # Bullet point summary
    result3 = pattern.run("QuickScan", doc1, summary_type=SummarizationType.BULLET_POINTS)
    
    # Print statistics
    print("\n" + "="*60)
    print("\n4. Summarization Statistics:")
    stats = result3["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Intelligent summarization with multiple strategies complete!")

Agentic Architecture Pattern: Summarization Agent

Pattern 236: 236 Summarization Agent – Line-by-Line Explanation

Overview

Pattern 236: Summarization Agent

Purpose: Intelligent text summarization with multiple strategies and customizable outputs. Implements extractive, abstractive, and hybrid summarization approaches.

Code Structure Analysis

File Information

Pattern Number: 236
Pattern Name: 236 Summarization Agent
File Path: patterns\domain_specific\236_summarization_agent.py
Total Lines: 478

Detailed Code Explanation

Module Docstring

"""
Pattern 236: Summarization Agent

Purpose: Intelligent text summarization with multiple strategies and customizable outputs.
Implements extractive, abstractive, and hybrid summarization approaches.
"""

Explanation: This module docstring describes the purpose and key features of the 236 Summarization Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re  # First 20 imports shown

Enumerations

SummarizationType (Line 17)

class SummarizationType(Enum):
    EXTRACTIVE
    ABSTRACTIVE
    HYBRID
    BULLET_POINTS
    HEADLINE

Purpose: Types of summarization strategies.

Values: – EXTRACTIVE: Represents extractive – ABSTRACTIVE: Represents abstractive – HYBRID: Represents hybrid – BULLET_POINTS: Represents bullet points – HEADLINE: Represents headline

SummaryLength (Line 26)

class SummaryLength(Enum):
    BRIEF
    SHORT
    MEDIUM
    DETAILED

Purpose: Target summary lengths.

Values: – BRIEF: Represents brief – SHORT: Represents short – MEDIUM: Represents medium – DETAILED: Represents detailed

Classes

SummarizationType (Line 17)

Description: Types of summarization strategies.

SummaryLength (Line 26)

Description: Target summary lengths.

Document (Line 35)

Description: Document to summarize.

Methods:

`__post_init__` (Line 43)

def __post_init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

Summary (Line 49)

Description: Generated summary.

SummarizationAgent (Line 61)

Description: Summarization Agent – Intelligent text summarization.

Features: – Multiple summarization strategies (extractive, abstractive, hybrid) – Customizable summary lengths – Key point extraction – Bullet point generation – Headline creation – Entity extraction – Compression ratio tracking

Methods:

`init` (Line 75)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – summary_type: Parameter for summary_type – default_length: Parameter for default_length

`summarize` (Line 91)

def summarize(self, document, summary_type, length):
    """Generate summary for document."""

Purpose: Generate summary for document.

Parameters: – self: Instance reference – document: Parameter for document – summary_type: Parameter for summary_type – length: Parameter for length

`_extractive_summary` (Line 145)

def _extractive_summary(self, document, length):
    """Extractive summarization - select important sentences."""

Purpose: Extractive summarization – select important sentences.

Parameters: – self: Instance reference – document: Parameter for document – length: Parameter for length

`_abstractive_summary` (Line 168)

def _abstractive_summary(self, document, length):
    """Abstractive summarization - generate new text."""

Purpose: Abstractive summarization – generate new text.

Parameters: – self: Instance reference – document: Parameter for document – length: Parameter for length

`_bullet_point_summary` (Line 190)

def _bullet_point_summary(self, document, length):
    """Generate bullet point summary."""

Purpose: Generate bullet point summary.

Parameters: – self: Instance reference – document: Parameter for document – length: Parameter for length

`_headline_summary` (Line 211)

def _headline_summary(self, document):
    """Generate headline-style summary."""

Purpose: Generate headline-style summary.

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

`_hybrid_summary` (Line 224)

def _hybrid_summary(self, document, length):
    """Hybrid summarization - combine extractive and abstractive."""

Purpose: Hybrid summarization – combine extractive and abstractive.

Parameters: – self: Instance reference – document: Parameter for document – length: Parameter for length

`_extract_key_points` (Line 243)

def _extract_key_points(self, document, max_points):
    """Extract key points from document."""

Purpose: Extract key points from document.

Parameters: – self: Instance reference – document: Parameter for document – max_points: Parameter for max_points

`_extract_entities` (Line 263)

def _extract_entities(self, document):
    """Extract important entities (names, places, concepts)."""

Purpose: Extract important entities (names, places, concepts).

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

`_split_sentences` (Line 281)

def _split_sentences(self, text):
    """Split text into sentences."""

Purpose: Split text into sentences.

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

`_score_sentences` (Line 288)

def _score_sentences(self, sentences, document):
    """Score sentences for importance."""

Purpose: Score sentences for importance.

Parameters: – self: Instance reference – sentences: Parameter for sentences – document: Parameter for document

`_get_sentence_count` (Line 322)

def _get_sentence_count(self, length, total_sentences):
    """Determine number of sentences based on target length."""

Purpose: Determine number of sentences based on target length.

Parameters: – self: Instance reference – length: Parameter for length – total_sentences: Parameter for total_sentences

`batch_summarize` (Line 334)

def batch_summarize(self, documents):
    """Summarize multiple documents."""

Purpose: Summarize multiple documents.

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

`get_statistics` (Line 338)

def get_statistics(self):
    """Get summarization statistics."""

Purpose: Get summarization statistics.

Parameters: – self: Instance reference

SummarizationPattern (Line 348)

Description: Summarization Pattern – Manage summarization agents.

Coordinates multiple summarization agents for different needs.

Methods:

`init` (Line 355)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 358)

def create_agent(self, name, summary_type, default_length):
    """Create a new summarization agent."""

Purpose: Create a new summarization agent.

Parameters: – self: Instance reference – name: Identifier or label – summary_type: Parameter for summary_type – default_length: Parameter for default_length

`run` (Line 368)

def run(self, agent_name, document, summary_type):
    """Generate summary using specified agent."""

Purpose: Generate summary using specified agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – document: Parameter for document – summary_type: Parameter for summary_type

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Search Agent


    """
Pattern 235: Search Agent

Purpose: Intelligent search agent with query processing, ranking, and retrieval strategies.
Implements semantic search, keyword search, and hybrid approaches with relevance scoring.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re


class SearchType(Enum):
    """Types of search strategies."""
    KEYWORD = "keyword"  # Keyword matching
    SEMANTIC = "semantic"  # Semantic/meaning-based
    FUZZY = "fuzzy"  # Fuzzy matching
    BOOLEAN = "boolean"  # Boolean operators (AND, OR, NOT)
    HYBRID = "hybrid"  # Combination of strategies


class SearchScope(Enum):
    """Scope of search."""
    TITLE = "title"
    CONTENT = "content"
    TAGS = "tags"
    METADATA = "metadata"
    ALL = "all"


@dataclass
class Document:
    """Document to be searched."""
    id: str
    title: str
    content: str
    tags: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    views: int = 0
    popularity: float = 0.0


@dataclass
class SearchQuery:
    """Processed search query."""
    original_query: str
    processed_terms: List[str]
    operators: Dict[str, List[str]] = field(default_factory=dict)  # AND, OR, NOT
    filters: Dict[str, Any] = field(default_factory=dict)
    scope: SearchScope = SearchScope.ALL


@dataclass
class SearchResult:
    """A single search result."""
    document: Document
    relevance_score: float
    matched_fields: List[str]
    highlights: List[str] = field(default_factory=list)
    rank: int = 0


class SearchAgent:
    """
    Search Agent - Intelligent search with multiple strategies.
    
    Features:
    - Query processing and understanding
    - Multiple search strategies (keyword, semantic, fuzzy, boolean)
    - Relevance scoring and ranking
    - Result highlighting
    - Search filters
    - Query expansion
    - Search analytics
    """
    
    def __init__(self,
                 name: str,
                 search_type: SearchType = SearchType.HYBRID):
        self.name = name
        self.search_type = search_type
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
        
        # Document store
        self.documents: Dict[str, Document] = {}
        self.inverted_index: Dict[str, Set[str]] = defaultdict(set)  # term -> doc_ids
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_searches": 0,
            "documents_indexed": 0,
            "avg_results_per_query": 0.0,
            "search_type_usage": defaultdict(int)
        }
    
    def add_document(self, document: Document):
        """Add document to search index."""
        self.documents[document.id] = document
        self._index_document(document)
        self.statistics["documents_indexed"] += 1
    
    def _index_document(self, document: Document):
        """Build inverted index for document."""
        # Index title
        for term in self._tokenize(document.title):
            self.inverted_index[term].add(document.id)
        
        # Index content
        for term in self._tokenize(document.content):
            self.inverted_index[term].add(document.id)
        
        # Index tags
        for tag in document.tags:
            self.inverted_index[tag.lower()].add(document.id)
    
    def search(self,
               query: str,
               max_results: int = 10,
               scope: SearchScope = SearchScope.ALL,
               filters: Optional[Dict[str, Any]] = None) -> List[SearchResult]:
        """Execute search query."""
        self.statistics["total_searches"] += 1
        self.statistics["search_type_usage"][self.search_type.value] += 1
        
        # Process query
        processed_query = self._process_query(query, scope, filters)
        
        # Execute search strategy
        if self.search_type == SearchType.KEYWORD:
            results = self._keyword_search(processed_query)
        elif self.search_type == SearchType.SEMANTIC:
            results = self._semantic_search(processed_query)
        elif self.search_type == SearchType.FUZZY:
            results = self._fuzzy_search(processed_query)
        elif self.search_type == SearchType.BOOLEAN:
            results = self._boolean_search(processed_query)
        else:  # HYBRID
            results = self._hybrid_search(processed_query)
        
        # Apply filters
        if filters:
            results = self._apply_filters(results, filters)
        
        # Rank results
        results = self._rank_results(results, processed_query)
        
        # Limit results
        results = results[:max_results]
        
        # Add highlights
        for result in results:
            result.highlights = self._generate_highlights(result.document, processed_query)
        
        # Update statistics
        if results:
            avg = self.statistics["avg_results_per_query"]
            total = self.statistics["total_searches"]
            self.statistics["avg_results_per_query"] = (avg * (total - 1) + len(results)) / total
        
        return results
    
    def _process_query(self,
                      query: str,
                      scope: SearchScope,
                      filters: Optional[Dict[str, Any]]) -> SearchQuery:
        """Process and understand query."""
        # Extract boolean operators
        operators = {
            "AND": [],
            "OR": [],
            "NOT": []
        }
        
        # Simple boolean parsing
        if " AND " in query.upper():
            operators["AND"] = re.split(r'\s+AND\s+', query, flags=re.IGNORECASE)
            query = " ".join(operators["AND"])
        elif " OR " in query.upper():
            operators["OR"] = re.split(r'\s+OR\s+', query, flags=re.IGNORECASE)
        elif " NOT " in query.upper():
            parts = re.split(r'\s+NOT\s+', query, flags=re.IGNORECASE)
            query = parts[0]
            operators["NOT"] = parts[1:] if len(parts) > 1 else []
        
        # Tokenize
        processed_terms = self._tokenize(query)
        
        return SearchQuery(
            original_query=query,
            processed_terms=processed_terms,
            operators=operators,
            filters=filters or {},
            scope=scope
        )
    
    def _tokenize(self, text: str) -> List[str]:
        """Tokenize text into terms."""
        # Simple tokenization (in production, use proper NLP)
        text = text.lower()
        text = re.sub(r'[^\w\s]', ' ', text)
        return [term for term in text.split() if len(term) > 2]
    
    def _keyword_search(self, query: SearchQuery) -> List[SearchResult]:
        """Keyword-based search."""
        doc_scores: Dict[str, float] = defaultdict(float)
        doc_matches: Dict[str, Set[str]] = defaultdict(set)
        
        # Find matching documents
        for term in query.processed_terms:
            if term in self.inverted_index:
                for doc_id in self.inverted_index[term]:
                    doc_scores[doc_id] += 1.0
                    doc_matches[doc_id].add(term)
        
        # Create results
        results = []
        for doc_id, score in doc_scores.items():
            document = self.documents[doc_id]
            
            # TF-IDF-like scoring
            term_frequency = score / len(query.processed_terms)
            doc_frequency = len(doc_matches[doc_id]) / len(self.inverted_index.get(list(doc_matches[doc_id])[0], {doc_id}))
            relevance_score = term_frequency * doc_frequency
            
            results.append(SearchResult(
                document=document,
                relevance_score=relevance_score,
                matched_fields=list(doc_matches[doc_id])
            ))
        
        return results
    
    def _semantic_search(self, query: SearchQuery) -> List[SearchResult]:
        """Semantic/meaning-based search using LLM."""
        # Use LLM to understand query intent and find relevant documents
        doc_summaries = "\n".join([
            f"{doc_id}: {doc.title} - {doc.content[:100]}"
            for doc_id, doc in list(self.documents.items())[:50]
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a search expert. Analyze the query and rank documents by relevance. "
                      "Return document IDs in order of relevance, separated by commas."),
            ("human", f"Query: {query.original_query}\n\nDocuments:\n{doc_summaries}\n\n"
                     f"Return top relevant document IDs:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse LLM response
        relevant_doc_ids = [doc_id.strip() for doc_id in str(response.content).split(",")[:20]]
        
        # Create results with semantic scores
        results = []
        for i, doc_id in enumerate(relevant_doc_ids):
            if doc_id in self.documents:
                results.append(SearchResult(
                    document=self.documents[doc_id],
                    relevance_score=1.0 - (i * 0.05),  # Decreasing score
                    matched_fields=["semantic"]
                ))
        
        return results
    
    def _fuzzy_search(self, query: SearchQuery) -> List[SearchResult]:
        """Fuzzy matching search."""
        results = []
        
        for doc_id, document in self.documents.items():
            # Calculate fuzzy match score
            score = 0.0
            matched_fields = []
            
            for term in query.processed_terms:
                # Title fuzzy match
                if self._fuzzy_match(term, document.title):
                    score += 2.0
                    matched_fields.append("title")
                
                # Content fuzzy match
                if self._fuzzy_match(term, document.content):
                    score += 1.0
                    matched_fields.append("content")
                
                # Tag match
                for tag in document.tags:
                    if self._fuzzy_match(term, tag):
                        score += 1.5
                        matched_fields.append("tags")
            
            if score > 0:
                results.append(SearchResult(
                    document=document,
                    relevance_score=score / len(query.processed_terms),
                    matched_fields=list(set(matched_fields))
                ))
        
        return results
    
    def _fuzzy_match(self, term: str, text: str) -> bool:
        """Simple fuzzy matching."""
        term = term.lower()
        text = text.lower()
        
        # Exact match
        if term in text:
            return True
        
        # Partial match (at least 70% characters)
        match_threshold = int(len(term) * 0.7)
        matched_chars = sum(1 for c in term if c in text)
        return matched_chars >= match_threshold
    
    def _boolean_search(self, query: SearchQuery) -> List[SearchResult]:
        """Boolean operator search (AND, OR, NOT)."""
        if query.operators["AND"]:
            # All terms must match
            doc_ids: Optional[Set[str]] = None
            for term in query.operators["AND"]:
                term_docs = set()
                for t in self._tokenize(term):
                    term_docs.update(self.inverted_index.get(t, set()))
                
                doc_ids = term_docs if doc_ids is None else doc_ids & term_docs
            
            doc_ids = doc_ids or set()
            
        elif query.operators["OR"]:
            # Any term can match
            doc_ids = set()
            for term in query.operators["OR"]:
                for t in self._tokenize(term):
                    doc_ids.update(self.inverted_index.get(t, set()))
        
        else:
            # Default keyword search
            return self._keyword_search(query)
        
        # Exclude NOT terms
        if query.operators["NOT"]:
            exclude_ids = set()
            for term in query.operators["NOT"]:
                for t in self._tokenize(term):
                    exclude_ids.update(self.inverted_index.get(t, set()))
            doc_ids -= exclude_ids
        
        # Create results
        results = []
        for doc_id in doc_ids:
            if doc_id in self.documents:
                results.append(SearchResult(
                    document=self.documents[doc_id],
                    relevance_score=1.0,
                    matched_fields=["boolean"]
                ))
        
        return results
    
    def _hybrid_search(self, query: SearchQuery) -> List[SearchResult]:
        """Hybrid search combining multiple strategies."""
        # Combine keyword and semantic search
        keyword_results = self._keyword_search(query)
        fuzzy_results = self._fuzzy_search(query)
        
        # Merge results with weighted scores
        combined: Dict[str, SearchResult] = {}
        
        # Weight: keyword 60%, fuzzy 40%
        for result in keyword_results:
            combined[result.document.id] = SearchResult(
                document=result.document,
                relevance_score=result.relevance_score * 0.6,
                matched_fields=result.matched_fields
            )
        
        for result in fuzzy_results:
            if result.document.id in combined:
                combined[result.document.id].relevance_score += result.relevance_score * 0.4
                combined[result.document.id].matched_fields.extend(result.matched_fields)
            else:
                combined[result.document.id] = SearchResult(
                    document=result.document,
                    relevance_score=result.relevance_score * 0.4,
                    matched_fields=result.matched_fields
                )
        
        return list(combined.values())
    
    def _apply_filters(self,
                       results: List[SearchResult],
                       filters: Dict[str, Any]) -> List[SearchResult]:
        """Apply filters to search results."""
        filtered = []
        
        for result in results:
            doc = result.document
            include = True
            
            # Apply each filter
            if "tags" in filters:
                if not any(tag in doc.tags for tag in filters["tags"]):
                    include = False
            
            if "min_views" in filters:
                if doc.views < filters["min_views"]:
                    include = False
            
            if "created_after" in filters:
                if doc.created_at  List[SearchResult]:
        """Rank search results."""
        # Sort by relevance score (descending)
        sorted_results = sorted(results, key=lambda r: r.relevance_score, reverse=True)
        
        # Assign ranks
        for i, result in enumerate(sorted_results, 1):
            result.rank = i
        
        return sorted_results
    
    def _generate_highlights(self,
                           document: Document,
                           query: SearchQuery) -> List[str]:
        """Generate highlighted snippets."""
        highlights = []
        content_lower = document.content.lower()
        
        for term in query.processed_terms[:3]:  # Top 3 terms
            # Find term in content
            idx = content_lower.find(term.lower())
            if idx != -1:
                # Extract context (50 chars before and after)
                start = max(0, idx - 50)
                end = min(len(document.content), idx + len(term) + 50)
                snippet = document.content[start:end]
                
                # Add ellipsis
                if start > 0:
                    snippet = "..." + snippet
                if end  Dict[str, Any]:
        """Get search statistics."""
        return {
            "name": self.name,
            "search_strategy": self.search_type.value,
            **self.statistics
        }


class SearchPattern:
    """
    Search Pattern - Manage search agents for different domains.
    
    Coordinates multiple search agents with different strategies.
    """
    
    def __init__(self):
        self.agents: Dict[str, SearchAgent] = {}
    
    def create_agent(self,
                     name: str,
                     search_type: SearchType = SearchType.HYBRID) -> SearchAgent:
        """Create a new search agent."""
        agent = SearchAgent(name, search_type)
        self.agents[name] = agent
        print(f"   Created search agent: {name} ({search_type.value})")
        return agent
    
    def run(self,
            agent_name: str,
            query: str,
            max_results: int = 5) -> Dict[str, Any]:
        """Execute search query."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        results = agent.search(query, max_results)
        
        print(f"\n=== Search Results from {agent_name} ===")
        print(f"Query: \"{query}\"")
        print(f"Found {len(results)} results\n")
        
        for result in results:
            print(f"#{result.rank}. {result.document.title} (Score: {result.relevance_score:.2f})")
            print(f"   Tags: {', '.join(result.document.tags)}")
            if result.highlights:
                print(f"   Preview: {result.highlights[0]}")
            print()
        
        return {
            "agent": agent_name,
            "query": query,
            "results_count": len(results),
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Search Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = SearchPattern()
    
    # Create search agents
    print("1. Setting up search agents...")
    
    web_search = pattern.create_agent("WebSearch", SearchType.HYBRID)
    doc_search = pattern.create_agent("DocSearch", SearchType.KEYWORD)
    
    # Add sample documents
    print("\n2. Indexing documents...")
    
    documents = [
        Document("d1", "Introduction to Machine Learning",
                "Machine learning is a subset of artificial intelligence that focuses on data and algorithms.",
                ["AI", "ML", "tutorial"]),
        Document("d2", "Deep Learning Fundamentals",
                "Deep learning uses neural networks with multiple layers to learn from data.",
                ["AI", "deep-learning", "neural-networks"]),
        Document("d3", "Natural Language Processing Guide",
                "NLP enables computers to understand, interpret and generate human language.",
                ["NLP", "AI", "language"]),
        Document("d4", "Computer Vision Applications",
                "Computer vision allows machines to interpret and understand visual information.",
                ["CV", "AI", "image-processing"]),
        Document("d5", "Reinforcement Learning Basics",
                "Reinforcement learning trains agents through trial and error using rewards.",
                ["RL", "AI", "agents"]),
        Document("d6", "Data Science with Python",
                "Python is the leading language for data science and machine learning.",
                ["python", "data-science", "tutorial"]),
    ]
    
    for doc in documents:
        web_search.add_document(doc)
    
    # Execute searches
    print("\n3. Executing searches...")
    
    queries = [
        "machine learning neural networks",
        "python data science",
        "artificial intelligence applications"
    ]
    
    for query in queries:
        result = pattern.run("WebSearch", query, max_results=3)
    
    # Print statistics
    print("\n4. Search Statistics:")
    stats = result["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Intelligent search with ranking and relevance scoring complete!")

Agentic Architecture Pattern: Search Agent

Pattern 235: 235 Search Agent – Line-by-Line Explanation

Overview

Pattern 235: Search Agent

Purpose: Intelligent search agent with query processing, ranking, and retrieval strategies. Implements semantic search, keyword search, and hybrid approaches with relevance scoring.

Code Structure Analysis

File Information

Pattern Number: 235
Pattern Name: 235 Search Agent
File Path: patterns\domain_specific\235_search_agent.py
Total Lines: 581

Detailed Code Explanation

Module Docstring

"""
Pattern 235: Search Agent

Purpose: Intelligent search agent with query processing, ranking, and retrieval strategies.
Implements semantic search, keyword search, and hybrid approaches with relevance scoring.
"""

Explanation: This module docstring describes the purpose and key features of the 235 Search Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import re  # First 20 imports shown

Enumerations

SearchType (Line 18)

class SearchType(Enum):
    KEYWORD
    SEMANTIC
    FUZZY
    BOOLEAN
    HYBRID

Purpose: Types of search strategies.

Values: – KEYWORD: Represents keyword – SEMANTIC: Represents semantic – FUZZY: Represents fuzzy – BOOLEAN: Represents boolean – HYBRID: Represents hybrid

SearchScope (Line 27)

class SearchScope(Enum):
    TITLE
    CONTENT
    TAGS
    METADATA
    ALL

Purpose: Scope of search.

Values: – TITLE: Represents title – CONTENT: Represents content – TAGS: Represents tags – METADATA: Represents metadata – ALL: Represents all

Classes

SearchType (Line 18)

Description: Types of search strategies.

SearchScope (Line 27)

Description: Scope of search.

Document (Line 37)

Description: Document to be searched.

SearchQuery (Line 50)

Description: Processed search query.

SearchResult (Line 60)

Description: A single search result.

SearchAgent (Line 69)

Description: Search Agent – Intelligent search with multiple strategies.

Features: – Query processing and understanding – Multiple search strategies (keyword, semantic, fuzzy, boolean) – Relevance scoring and ranking – Result highlighting – Search filters – Query expansion – Search analytics

Methods:

`init` (Line 83)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – search_type: Parameter for search_type

`add_document` (Line 102)

def add_document(self, document):
    """Add document to search index."""

Purpose: Add document to search index.

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

`_index_document` (Line 108)

def _index_document(self, document):
    """Build inverted index for document."""

Purpose: Build inverted index for document.

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

`search` (Line 122)

def search(self, query, max_results, scope, filters):
    """Execute search query."""

Purpose: Execute search query.

Parameters: – self: Instance reference – query: Parameter for query – max_results: Parameter for max_results – scope: Parameter for scope – filters: Parameter for filters

`_process_query` (Line 168)

def _process_query(self, query, scope, filters):
    """Process and understand query."""

Purpose: Process and understand query.

Parameters: – self: Instance reference – query: Parameter for query – scope: Parameter for scope – filters: Parameter for filters

`_tokenize` (Line 202)

def _tokenize(self, text):
    """Tokenize text into terms."""

Purpose: Tokenize text into terms.

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

`_keyword_search` (Line 209)

def _keyword_search(self, query):
    """Keyword-based search."""

Purpose: Keyword-based search.

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

`_semantic_search` (Line 239)

def _semantic_search(self, query):
    """Semantic/meaning-based search using LLM."""

Purpose: Semantic/meaning-based search using LLM.

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

`_fuzzy_search` (Line 272)

def _fuzzy_search(self, query):
    """Fuzzy matching search."""

Purpose: Fuzzy matching search.

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

`_fuzzy_match` (Line 307)

def _fuzzy_match(self, term, text):
    """Simple fuzzy matching."""

Purpose: Simple fuzzy matching.

Parameters: – self: Instance reference – term: Parameter for term – text: Parameter for text

`_boolean_search` (Line 321)

def _boolean_search(self, query):
    """Boolean operator search (AND, OR, NOT)."""

Purpose: Boolean operator search (AND, OR, NOT).

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

`_hybrid_search` (Line 366)

def _hybrid_search(self, query):
    """Hybrid search combining multiple strategies."""

Purpose: Hybrid search combining multiple strategies.

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

`_apply_filters` (Line 396)

def _apply_filters(self, results, filters):
    """Apply filters to search results."""

Purpose: Apply filters to search results.

Parameters: – self: Instance reference – results: Parameter for results – filters: Parameter for filters

`_rank_results` (Line 424)

def _rank_results(self, results, query):
    """Rank search results."""

Purpose: Rank search results.

Parameters: – self: Instance reference – results: Parameter for results – query: Parameter for query

`_generate_highlights` (Line 437)

def _generate_highlights(self, document, query):
    """Generate highlighted snippets."""

Purpose: Generate highlighted snippets.

Parameters: – self: Instance reference – document: Parameter for document – query: Parameter for query

…and 1 more methods

SearchPattern (Line 472)

Description: Search Pattern – Manage search agents for different domains.

Coordinates multiple search agents with different strategies.

Methods:

`init` (Line 479)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 482)

def create_agent(self, name, search_type):
    """Create a new search agent."""

Purpose: Create a new search agent.

Parameters: – self: Instance reference – name: Identifier or label – search_type: Parameter for search_type

`run` (Line 491)

def run(self, agent_name, query, max_results):
    """Execute search query."""

Purpose: Execute search query.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – query: Parameter for query – max_results: Parameter for max_results

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Recommendation Agent


    """
Pattern 234: Recommendation Agent

Purpose: Personalized recommendation system that learns user preferences and suggests relevant items.
Implements collaborative filtering, content-based filtering, and hybrid approaches.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import math


class RecommendationType(Enum):
    """Types of recommendation strategies."""
    COLLABORATIVE = "collaborative"  # Based on similar users
    CONTENT_BASED = "content_based"  # Based on item features
    HYBRID = "hybrid"  # Combination of both
    POPULAR = "popular"  # Based on popularity
    PERSONALIZED = "personalized"  # LLM-based personalization


class ItemCategory(Enum):
    """Categories of items."""
    MOVIE = "movie"
    BOOK = "book"
    PRODUCT = "product"
    RESTAURANT = "restaurant"
    MUSIC = "music"
    ARTICLE = "article"
    COURSE = "course"


@dataclass
class Item:
    """Item to recommend."""
    id: str
    title: str
    category: ItemCategory
    tags: List[str] = field(default_factory=list)
    features: Dict[str, Any] = field(default_factory=dict)
    popularity_score: float = 0.0
    average_rating: float = 0.0
    description: str = ""


@dataclass
class UserProfile:
    """User profile with preferences."""
    user_id: str
    rated_items: Dict[str, float] = field(default_factory=dict)  # item_id -> rating
    viewed_items: Set[str] = field(default_factory=set)
    purchased_items: Set[str] = field(default_factory=set)
    preferred_tags: List[str] = field(default_factory=list)
    preferred_categories: List[ItemCategory] = field(default_factory=list)
    interaction_history: List[Dict[str, Any]] = field(default_factory=list)


@dataclass
class Recommendation:
    """A single recommendation."""
    item: Item
    score: float
    reason: str
    recommendation_type: RecommendationType


class RecommendationAgent:
    """
    Recommendation Agent - Personalized recommendations using multiple strategies.
    
    Features:
    - Collaborative filtering (user-based similarity)
    - Content-based filtering (item features)
    - Hybrid recommendations
    - Popularity-based recommendations
    - LLM-enhanced personalization
    - User profile learning
    """
    
    def __init__(self,
                 name: str,
                 recommendation_type: RecommendationType = RecommendationType.HYBRID):
        self.name = name
        self.recommendation_type = recommendation_type
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        
        # Data stores
        self.items: Dict[str, Item] = {}
        self.user_profiles: Dict[str, UserProfile] = {}
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "recommendations_made": 0,
            "users_served": 0,
            "items_added": 0,
            "strategy_usage": defaultdict(int)
        }
    
    def add_item(self, item: Item):
        """Add item to recommendation system."""
        self.items[item.id] = item
        self.statistics["items_added"] += 1
    
    def add_user(self, user_id: str) -> UserProfile:
        """Add or get user profile."""
        if user_id not in self.user_profiles:
            self.user_profiles[user_id] = UserProfile(user_id=user_id)
        return self.user_profiles[user_id]
    
    def record_rating(self, user_id: str, item_id: str, rating: float):
        """Record user rating for an item."""
        profile = self.add_user(user_id)
        profile.rated_items[item_id] = rating
        profile.interaction_history.append({
            "action": "rating",
            "item_id": item_id,
            "value": rating,
            "timestamp": datetime.now()
        })
        
        # Update item average rating
        if item_id in self.items:
            item = self.items[item_id]
            ratings = [p.rated_items.get(item_id, 0) for p in self.user_profiles.values() 
                      if item_id in p.rated_items]
            if ratings:
                item.average_rating = sum(ratings) / len(ratings)
    
    def record_view(self, user_id: str, item_id: str):
        """Record user viewing an item."""
        profile = self.add_user(user_id)
        profile.viewed_items.add(item_id)
        profile.interaction_history.append({
            "action": "view",
            "item_id": item_id,
            "timestamp": datetime.now()
        })
    
    def recommend(self,
                  user_id: str,
                  count: int = 5,
                  exclude_seen: bool = True) -> List[Recommendation]:
        """Generate recommendations for user."""
        profile = self.add_user(user_id)
        self.statistics["users_served"] = len(self.user_profiles)
        
        # Choose recommendation strategy
        if self.recommendation_type == RecommendationType.COLLABORATIVE:
            recommendations = self._collaborative_filtering(profile, count, exclude_seen)
        elif self.recommendation_type == RecommendationType.CONTENT_BASED:
            recommendations = self._content_based_filtering(profile, count, exclude_seen)
        elif self.recommendation_type == RecommendationType.POPULAR:
            recommendations = self._popular_recommendations(profile, count, exclude_seen)
        elif self.recommendation_type == RecommendationType.PERSONALIZED:
            recommendations = self._llm_personalized_recommendations(profile, count, exclude_seen)
        else:  # HYBRID
            recommendations = self._hybrid_recommendations(profile, count, exclude_seen)
        
        self.statistics["recommendations_made"] += len(recommendations)
        self.statistics["strategy_usage"][self.recommendation_type.value] += 1
        
        return recommendations
    
    def _collaborative_filtering(self,
                                 profile: UserProfile,
                                 count: int,
                                 exclude_seen: bool) -> List[Recommendation]:
        """Collaborative filtering based on similar users."""
        if not profile.rated_items:
            return self._popular_recommendations(profile, count, exclude_seen)
        
        # Find similar users
        similar_users = self._find_similar_users(profile)
        
        # Aggregate recommendations from similar users
        item_scores: Dict[str, float] = defaultdict(float)
        
        for similar_user_id, similarity in similar_users[:5]:  # Top 5 similar users
            similar_profile = self.user_profiles[similar_user_id]
            
            for item_id, rating in similar_profile.rated_items.items():
                if exclude_seen and (item_id in profile.rated_items or 
                                    item_id in profile.viewed_items):
                    continue
                
                # Weight by similarity and rating
                item_scores[item_id] += similarity * rating
        
        # Create recommendations
        recommendations = []
        for item_id, score in sorted(item_scores.items(), key=lambda x: x[1], reverse=True)[:count]:
            if item_id in self.items:
                recommendations.append(Recommendation(
                    item=self.items[item_id],
                    score=score,
                    reason="Based on users with similar tastes",
                    recommendation_type=RecommendationType.COLLABORATIVE
                ))
        
        return recommendations
    
    def _content_based_filtering(self,
                                 profile: UserProfile,
                                 count: int,
                                 exclude_seen: bool) -> List[Recommendation]:
        """Content-based filtering using item features."""
        if not profile.rated_items:
            return self._popular_recommendations(profile, count, exclude_seen)
        
        # Build user preference profile from rated items
        preferred_tags = self._extract_preferred_tags(profile)
        
        # Score items based on feature similarity
        item_scores: Dict[str, float] = {}
        
        for item_id, item in self.items.items():
            if exclude_seen and (item_id in profile.rated_items or 
                                item_id in profile.viewed_items):
                continue
            
            # Calculate similarity score based on tags
            tag_overlap = len(set(item.tags) & set(preferred_tags))
            category_match = 1.0 if item.category in profile.preferred_categories else 0.5
            
            score = (tag_overlap * 2.0 + category_match) * item.average_rating
            item_scores[item_id] = score
        
        # Create recommendations
        recommendations = []
        for item_id, score in sorted(item_scores.items(), key=lambda x: x[1], reverse=True)[:count]:
            matching_tags = list(set(self.items[item_id].tags) & set(preferred_tags))
            reason = f"Matches your interests: {', '.join(matching_tags[:3])}"
            
            recommendations.append(Recommendation(
                item=self.items[item_id],
                score=score,
                reason=reason,
                recommendation_type=RecommendationType.CONTENT_BASED
            ))
        
        return recommendations
    
    def _popular_recommendations(self,
                                profile: UserProfile,
                                count: int,
                                exclude_seen: bool) -> List[Recommendation]:
        """Popularity-based recommendations."""
        # Score by popularity and ratings
        item_scores = []
        
        for item_id, item in self.items.items():
            if exclude_seen and (item_id in profile.rated_items or 
                                item_id in profile.viewed_items):
                continue
            
            # Combine popularity and average rating
            score = item.popularity_score * 0.5 + item.average_rating * 0.5
            item_scores.append((item_id, score))
        
        # Create recommendations
        recommendations = []
        for item_id, score in sorted(item_scores, key=lambda x: x[1], reverse=True)[:count]:
            recommendations.append(Recommendation(
                item=self.items[item_id],
                score=score,
                reason="Popular among all users",
                recommendation_type=RecommendationType.POPULAR
            ))
        
        return recommendations
    
    def _hybrid_recommendations(self,
                               profile: UserProfile,
                               count: int,
                               exclude_seen: bool) -> List[Recommendation]:
        """Hybrid approach combining multiple strategies."""
        # Get recommendations from different strategies
        collab_recs = self._collaborative_filtering(profile, count, exclude_seen)
        content_recs = self._content_based_filtering(profile, count, exclude_seen)
        popular_recs = self._popular_recommendations(profile, count, exclude_seen)
        
        # Combine and deduplicate
        all_recs: Dict[str, Recommendation] = {}
        
        # Weight: collaborative 40%, content 40%, popular 20%
        for rec in collab_recs:
            all_recs[rec.item.id] = Recommendation(
                item=rec.item,
                score=rec.score * 0.4,
                reason="Hybrid: Similar users + Your preferences",
                recommendation_type=RecommendationType.HYBRID
            )
        
        for rec in content_recs:
            if rec.item.id in all_recs:
                all_recs[rec.item.id].score += rec.score * 0.4
            else:
                all_recs[rec.item.id] = Recommendation(
                    item=rec.item,
                    score=rec.score * 0.4,
                    reason="Hybrid: Matches your interests",
                    recommendation_type=RecommendationType.HYBRID
                )
        
        for rec in popular_recs:
            if rec.item.id in all_recs:
                all_recs[rec.item.id].score += rec.score * 0.2
            else:
                all_recs[rec.item.id] = Recommendation(
                    item=rec.item,
                    score=rec.score * 0.2,
                    reason="Hybrid: Popular choice",
                    recommendation_type=RecommendationType.HYBRID
                )
        
        # Sort and return top recommendations
        sorted_recs = sorted(all_recs.values(), key=lambda x: x.score, reverse=True)
        return sorted_recs[:count]
    
    def _llm_personalized_recommendations(self,
                                         profile: UserProfile,
                                         count: int,
                                         exclude_seen: bool) -> List[Recommendation]:
        """LLM-enhanced personalized recommendations."""
        # Get baseline recommendations
        baseline_recs = self._hybrid_recommendations(profile, count * 2, exclude_seen)
        
        if not baseline_recs:
            return []
        
        # Use LLM to refine and personalize
        items_info = "\n".join([
            f"- {rec.item.title} (Category: {rec.item.category.value}, "
            f"Tags: {', '.join(rec.item.tags[:3])}, Rating: {rec.item.average_rating:.1f})"
            for rec in baseline_recs[:10]
        ])
        
        user_history = ", ".join([
            self.items[item_id].title 
            for item_id in list(profile.rated_items.keys())[:5]
            if item_id in self.items
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a recommendation expert. Analyze user preferences and "
                      "select the most relevant items to recommend. Explain why each fits the user."),
            ("human", f"User has liked: {user_history}\n\n"
                     f"Available items:\n{items_info}\n\n"
                     f"Select top {count} items and explain why they match user preferences.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Return top recommendations with LLM explanation
        return baseline_recs[:count]
    
    # Helper methods
    def _find_similar_users(self, profile: UserProfile) -> List[tuple]:
        """Find users with similar rating patterns."""
        similarities = []
        
        for other_id, other_profile in self.user_profiles.items():
            if other_id == profile.user_id:
                continue
            
            # Calculate cosine similarity based on ratings
            similarity = self._calculate_user_similarity(profile, other_profile)
            if similarity > 0:
                similarities.append((other_id, similarity))
        
        return sorted(similarities, key=lambda x: x[1], reverse=True)
    
    def _calculate_user_similarity(self,
                                   profile1: UserProfile,
                                   profile2: UserProfile) -> float:
        """Calculate similarity between two users."""
        # Find common rated items
        common_items = set(profile1.rated_items.keys()) & set(profile2.rated_items.keys())
        
        if not common_items:
            return 0.0
        
        # Cosine similarity
        dot_product = sum(
            profile1.rated_items[item] * profile2.rated_items[item]
            for item in common_items
        )
        
        magnitude1 = math.sqrt(sum(r**2 for r in profile1.rated_items.values()))
        magnitude2 = math.sqrt(sum(r**2 for r in profile2.rated_items.values()))
        
        if magnitude1 == 0 or magnitude2 == 0:
            return 0.0
        
        return dot_product / (magnitude1 * magnitude2)
    
    def _extract_preferred_tags(self, profile: UserProfile) -> List[str]:
        """Extract preferred tags from user's rated items."""
        tag_scores: Dict[str, float] = defaultdict(float)
        
        for item_id, rating in profile.rated_items.items():
            if item_id in self.items:
                item = self.items[item_id]
                for tag in item.tags:
                    tag_scores[tag] += rating
        
        # Return tags sorted by preference
        return [tag for tag, _ in sorted(tag_scores.items(), key=lambda x: x[1], reverse=True)]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get recommendation system statistics."""
        return {
            "name": self.name,
            "total_items": len(self.items),
            "total_users": len(self.user_profiles),
            "recommendation_strategy": self.recommendation_type.value,
            **self.statistics
        }


class RecommendationPattern:
    """
    Recommendation Pattern - Manage recommendation systems.
    
    Coordinates multiple recommendation agents for different domains.
    """
    
    def __init__(self):
        self.agents: Dict[str, RecommendationAgent] = {}
    
    def create_agent(self,
                     name: str,
                     recommendation_type: RecommendationType = RecommendationType.HYBRID) -> RecommendationAgent:
        """Create a new recommendation agent."""
        agent = RecommendationAgent(name, recommendation_type)
        self.agents[name] = agent
        print(f"   Created recommendation agent: {name} ({recommendation_type.value})")
        return agent
    
    def run(self,
            agent_name: str,
            user_id: str,
            count: int = 5) -> Dict[str, Any]:
        """Generate recommendations for user."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        recommendations = agent.recommend(user_id, count)
        
        print(f"\n=== Recommendations from {agent_name} for User: {user_id} ===\n")
        
        for i, rec in enumerate(recommendations, 1):
            print(f"{i}. {rec.item.title} (Score: {rec.score:.2f})")
            print(f"   Category: {rec.item.category.value} | Rating: {rec.item.average_rating:.1f}/5.0")
            print(f"   Reason: {rec.reason}\n")
        
        return {
            "agent": agent_name,
            "user": user_id,
            "recommendations": len(recommendations),
            "statistics": agent.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Recommendation Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = RecommendationPattern()
    
    # Create agents
    print("1. Setting up recommendation agents...")
    
    movie_agent = pattern.create_agent("MovieRecommender", RecommendationType.HYBRID)
    book_agent = pattern.create_agent("BookRecommender", RecommendationType.CONTENT_BASED)
    
    # Add sample movies
    print("\n2. Adding sample items...")
    
    movies = [
        Item("m1", "The Matrix", ItemCategory.MOVIE, 
             ["sci-fi", "action", "philosophical"], popularity_score=0.95),
        Item("m2", "Inception", ItemCategory.MOVIE,
             ["sci-fi", "thriller", "mind-bending"], popularity_score=0.92),
        Item("m3", "Interstellar", ItemCategory.MOVIE,
             ["sci-fi", "drama", "space"], popularity_score=0.90),
        Item("m4", "The Shawshank Redemption", ItemCategory.MOVIE,
             ["drama", "inspiring", "classic"], popularity_score=0.98),
        Item("m5", "Pulp Fiction", ItemCategory.MOVIE,
             ["crime", "comedy", "nonlinear"], popularity_score=0.88),
        Item("m6", "The Dark Knight", ItemCategory.MOVIE,
             ["action", "superhero", "crime"], popularity_score=0.94),
        Item("m7", "Blade Runner 2049", ItemCategory.MOVIE,
             ["sci-fi", "philosophical", "visual"], popularity_score=0.82),
        Item("m8", "Arrival", ItemCategory.MOVIE,
             ["sci-fi", "drama", "linguistics"], popularity_score=0.85)
    ]
    
    for movie in movies:
        movie_agent.add_item(movie)
    
    # Simulate user interactions
    print("   Adding user ratings...")
    
    # User 1 likes sci-fi
    movie_agent.record_rating("user1", "m1", 5.0)
    movie_agent.record_rating("user1", "m2", 4.5)
    movie_agent.record_rating("user1", "m3", 4.0)
    
    # User 2 likes drama
    movie_agent.record_rating("user2", "m4", 5.0)
    movie_agent.record_rating("user2", "m3", 3.5)
    
    # User 3 similar to user1
    movie_agent.record_rating("user3", "m1", 4.8)
    movie_agent.record_rating("user3", "m2", 4.7)
    movie_agent.record_rating("user3", "m7", 4.5)
    
    # Get recommendations
    print("\n3. Generating recommendations...")
    
    result = pattern.run("MovieRecommender", "user1", count=3)
    
    # Print statistics
    print("\n4. System Statistics:")
    stats = result["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Personalized recommendations generated using multiple strategies!")

Agentic Architecture Pattern: Recommendation Agent

Pattern 234: 234 Recommendation Agent – Line-by-Line Explanation

Overview

Pattern 234: Recommendation Agent

Purpose: Personalized recommendation system that learns user preferences and suggests relevant items. Implements collaborative filtering, content-based filtering, and hybrid approaches.

Code Structure Analysis

File Information

Pattern Number: 234
Pattern Name: 234 Recommendation Agent
File Path: patterns\domain_specific\234_recommendation_agent.py
Total Lines: 539

Detailed Code Explanation

Module Docstring

"""
Pattern 234: Recommendation Agent

Purpose: Personalized recommendation system that learns user preferences and suggests relevant items.
Implements collaborative filtering, content-based filtering, and hybrid approaches.
"""

Explanation: This module docstring describes the purpose and key features of the 234 Recommendation Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
import math  # First 20 imports shown

Enumerations

RecommendationType (Line 18)

class RecommendationType(Enum):
    COLLABORATIVE
    CONTENT_BASED
    HYBRID
    POPULAR
    PERSONALIZED

Purpose: Types of recommendation strategies.

Values: – COLLABORATIVE: Represents collaborative – CONTENT_BASED: Represents content based – HYBRID: Represents hybrid – POPULAR: Represents popular – PERSONALIZED: Represents personalized

ItemCategory (Line 27)

class ItemCategory(Enum):
    MOVIE
    BOOK
    PRODUCT
    RESTAURANT
    MUSIC
    ARTICLE
    COURSE

Purpose: Categories of items.

Values: – MOVIE: Represents movie – BOOK: Represents book – PRODUCT: Represents product – RESTAURANT: Represents restaurant – MUSIC: Represents music – ARTICLE: Represents article – COURSE: Represents course

Classes

RecommendationType (Line 18)

Description: Types of recommendation strategies.

ItemCategory (Line 27)

Description: Categories of items.

Item (Line 39)

Description: Item to recommend.

UserProfile (Line 52)

Description: User profile with preferences.

Recommendation (Line 64)

Description: A single recommendation.

RecommendationAgent (Line 72)

Description: Recommendation Agent – Personalized recommendations using multiple strategies.

Features: – Collaborative filtering (user-based similarity) – Content-based filtering (item features) – Hybrid recommendations – Popularity-based recommendations – LLM-enhanced personalization – User profile learning

Methods:

`init` (Line 85)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – recommendation_type: Parameter for recommendation_type

`add_item` (Line 104)

def add_item(self, item):
    """Add item to recommendation system."""

Purpose: Add item to recommendation system.

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

`add_user` (Line 109)

def add_user(self, user_id):
    """Add or get user profile."""

Purpose: Add or get user profile.

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

`record_rating` (Line 115)

def record_rating(self, user_id, item_id, rating):
    """Record user rating for an item."""

Purpose: Record user rating for an item.

Parameters: – self: Instance reference – user_id: Parameter for user_id – item_id: Parameter for item_id – rating: Parameter for rating

`record_view` (Line 134)

def record_view(self, user_id, item_id):
    """Record user viewing an item."""

Purpose: Record user viewing an item.

Parameters: – self: Instance reference – user_id: Parameter for user_id – item_id: Parameter for item_id

`recommend` (Line 144)

def recommend(self, user_id, count, exclude_seen):
    """Generate recommendations for user."""

Purpose: Generate recommendations for user.

Parameters: – self: Instance reference – user_id: Parameter for user_id – count: Parameter for count – exclude_seen: Parameter for exclude_seen

`_collaborative_filtering` (Line 169)

def _collaborative_filtering(self, profile, count, exclude_seen):
    """Collaborative filtering based on similar users."""

Purpose: Collaborative filtering based on similar users.

Parameters: – self: Instance reference – profile: Parameter for profile – count: Parameter for count – exclude_seen: Parameter for exclude_seen

`_content_based_filtering` (Line 207)

def _content_based_filtering(self, profile, count, exclude_seen):
    """Content-based filtering using item features."""

Purpose: Content-based filtering using item features.

Parameters: – self: Instance reference – profile: Parameter for profile – count: Parameter for count – exclude_seen: Parameter for exclude_seen

`_popular_recommendations` (Line 248)

def _popular_recommendations(self, profile, count, exclude_seen):
    """Popularity-based recommendations."""

Purpose: Popularity-based recommendations.

Parameters: – self: Instance reference – profile: Parameter for profile – count: Parameter for count – exclude_seen: Parameter for exclude_seen

`_hybrid_recommendations` (Line 277)

def _hybrid_recommendations(self, profile, count, exclude_seen):
    """Hybrid approach combining multiple strategies."""

Purpose: Hybrid approach combining multiple strategies.

Parameters: – self: Instance reference – profile: Parameter for profile – count: Parameter for count – exclude_seen: Parameter for exclude_seen

`_llm_personalized_recommendations` (Line 325)

def _llm_personalized_recommendations(self, profile, count, exclude_seen):
    """LLM-enhanced personalized recommendations."""

Purpose: LLM-enhanced personalized recommendations.

Parameters: – self: Instance reference – profile: Parameter for profile – count: Parameter for count – exclude_seen: Parameter for exclude_seen

`_find_similar_users` (Line 364)

def _find_similar_users(self, profile):
    """Find users with similar rating patterns."""

Purpose: Find users with similar rating patterns.

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

`_calculate_user_similarity` (Line 379)

def _calculate_user_similarity(self, profile1, profile2):
    """Calculate similarity between two users."""

Purpose: Calculate similarity between two users.

Parameters: – self: Instance reference – profile1: Parameter for profile1 – profile2: Parameter for profile2

`_extract_preferred_tags` (Line 403)

def _extract_preferred_tags(self, profile):
    """Extract preferred tags from user's rated items."""

Purpose: Extract preferred tags from user’s rated items.

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

`get_statistics` (Line 416)

def get_statistics(self):
    """Get recommendation system statistics."""

Purpose: Get recommendation system statistics.

Parameters: – self: Instance reference

RecommendationPattern (Line 427)

Description: Recommendation Pattern – Manage recommendation systems.

Coordinates multiple recommendation agents for different domains.

Methods:

`init` (Line 434)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`create_agent` (Line 437)

def create_agent(self, name, recommendation_type):
    """Create a new recommendation agent."""

Purpose: Create a new recommendation agent.

Parameters: – self: Instance reference – name: Identifier or label – recommendation_type: Parameter for recommendation_type

`run` (Line 446)

def run(self, agent_name, user_id, count):
    """Generate recommendations for user."""

Purpose: Generate recommendations for user.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – user_id: Parameter for user_id – count: Parameter for count

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Virtual Assistant


    """
Pattern 233: Virtual Assistant

Purpose: Personalized AI assistant that executes tasks, manages schedules, and provides proactive help.
Handles calendar, reminders, information retrieval, and task management with user preferences.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class TaskType(Enum):
    """Types of tasks the assistant can handle."""
    SCHEDULE = "schedule"  # Calendar events
    REMINDER = "reminder"  # Set reminders
    SEARCH = "search"  # Information retrieval
    EMAIL = "email"  # Email management
    TODO = "todo"  # Task management
    WEATHER = "weather"  # Weather info
    NEWS = "news"  # News updates
    CALCULATION = "calculation"  # Math calculations
    TRANSLATION = "translation"  # Language translation
    GENERAL = "general"  # General assistance


class Priority(Enum):
    """Task priority levels."""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"


@dataclass
class CalendarEvent:
    """Calendar event."""
    title: str
    start_time: datetime
    end_time: datetime
    description: str = ""
    location: str = ""
    attendees: List[str] = field(default_factory=list)
    reminder_minutes: int = 15


@dataclass
class Reminder:
    """Reminder for user."""
    message: str
    remind_at: datetime
    priority: Priority = Priority.MEDIUM
    recurring: bool = False
    completed: bool = False


@dataclass
class TodoItem:
    """Todo list item."""
    task: str
    priority: Priority = Priority.MEDIUM
    due_date: Optional[datetime] = None
    completed: bool = False
    tags: List[str] = field(default_factory=list)


@dataclass
class UserPreferences:
    """User preferences and profile."""
    name: str
    timezone: str = "UTC"
    preferred_language: str = "English"
    work_hours: tuple = (9, 17)  # 9 AM to 5 PM
    interests: List[str] = field(default_factory=list)
    notification_style: str = "normal"  # normal, minimal, detailed


class VirtualAssistant:
    """
    Virtual Assistant - Personalized AI assistant for task execution and scheduling.
    
    Features:
    - Calendar management with scheduling
    - Reminder system with priorities
    - Todo list management
    - Information retrieval and search
    - Proactive suggestions
    - Personalized responses based on user preferences
    """
    
    def __init__(self,
                 name: str,
                 user_preferences: Optional[UserPreferences] = None):
        self.name = name
        self.user_preferences = user_preferences or UserPreferences(name="User")
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        
        # State management
        self.calendar: List[CalendarEvent] = []
        self.reminders: List[Reminder] = []
        self.todo_list: List[TodoItem] = []
        self.conversation_context: List[str] = []
        
        # Task handlers
        self.task_handlers: Dict[TaskType, Callable] = {
            TaskType.SCHEDULE: self._handle_schedule,
            TaskType.REMINDER: self._handle_reminder,
            TaskType.SEARCH: self._handle_search,
            TaskType.TODO: self._handle_todo,
            TaskType.WEATHER: self._handle_weather,
            TaskType.CALCULATION: self._handle_calculation,
            TaskType.GENERAL: self._handle_general
        }
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "tasks_completed": 0,
            "events_scheduled": 0,
            "reminders_set": 0,
            "proactive_suggestions": 0
        }
    
    def assist(self, user_request: str) -> str:
        """Process user request and provide assistance."""
        # Add to conversation context
        self.conversation_context.append(user_request)
        
        # Determine task type
        task_type = self._classify_task(user_request)
        
        # Execute appropriate handler
        handler = self.task_handlers.get(task_type, self._handle_general)
        response = handler(user_request)
        
        # Check for proactive suggestions
        suggestions = self._generate_proactive_suggestions()
        if suggestions:
            response += f"\n\n💡 Proactive suggestion: {suggestions}"
            self.statistics["proactive_suggestions"] += 1
        
        self.statistics["tasks_completed"] += 1
        
        return response
    
    def _classify_task(self, user_request: str) -> TaskType:
        """Classify the type of task from user request."""
        request_lower = user_request.lower()
        
        # Pattern matching for task classification
        if any(word in request_lower for word in ["schedule", "meeting", "appointment", "calendar", "book"]):
            return TaskType.SCHEDULE
        elif any(word in request_lower for word in ["remind", "reminder", "alert", "notify"]):
            return TaskType.REMINDER
        elif any(word in request_lower for word in ["search", "find", "look up", "google", "what is"]):
            return TaskType.SEARCH
        elif any(word in request_lower for word in ["todo", "task", "add to list", "need to do"]):
            return TaskType.TODO
        elif any(word in request_lower for word in ["weather", "temperature", "forecast", "rain"]):
            return TaskType.WEATHER
        elif any(word in request_lower for word in ["calculate", "compute", "math", "+", "-", "*", "/"]):
            return TaskType.CALCULATION
        else:
            return TaskType.GENERAL
    
    # Task handlers
    def _handle_schedule(self, request: str) -> str:
        """Handle calendar scheduling."""
        # Extract event details using LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Extract event details from the user's scheduling request. "
                      "Provide: title, date, time, and any other relevant details."),
            ("human", "{request}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"request": request})
        
        # Create event (simplified - in production would parse LLM response)
        event = CalendarEvent(
            title="Meeting",
            start_time=datetime.now() + timedelta(days=1, hours=2),
            end_time=datetime.now() + timedelta(days=1, hours=3),
            description=request
        )
        
        self.calendar.append(event)
        self.statistics["events_scheduled"] += 1
        
        return f"✓ Scheduled: {event.title} for {event.start_time.strftime('%Y-%m-%d at %H:%M')}. " \
               f"I'll remind you 15 minutes before."
    
    def _handle_reminder(self, request: str) -> str:
        """Handle reminder setting."""
        # Determine reminder time and message
        reminder = Reminder(
            message=request,
            remind_at=datetime.now() + timedelta(hours=1),  # Default 1 hour
            priority=Priority.MEDIUM
        )
        
        self.reminders.append(reminder)
        self.statistics["reminders_set"] += 1
        
        return f"✓ Reminder set: '{reminder.message}' for {reminder.remind_at.strftime('%Y-%m-%d at %H:%M')}"
    
    def _handle_search(self, request: str) -> str:
        """Handle information search."""
        # Use LLM for information retrieval
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a helpful virtual assistant. "
                      f"Provide concise, accurate information in response to user queries. "
                      f"User's interests: {', '.join(self.user_preferences.interests or ['general topics'])}"),
            ("human", "{query}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"query": request})
        
        return f"🔍 Search result: {response.content}"
    
    def _handle_todo(self, request: str) -> str:
        """Handle todo list management."""
        # Extract task from request
        task_text = request.replace("add", "").replace("to do", "").replace("todo", "").strip()
        
        todo = TodoItem(
            task=task_text,
            priority=Priority.MEDIUM,
            due_date=None
        )
        
        self.todo_list.append(todo)
        
        incomplete_count = sum(1 for t in self.todo_list if not t.completed)
        return f"✓ Added to todo list: '{todo.task}'. You now have {incomplete_count} pending tasks."
    
    def _handle_weather(self, request: str) -> str:
        """Handle weather queries."""
        # Simulate weather response (in production, would call weather API)
        return f"☀️ Weather forecast: Currently 72°F and sunny. " \
               f"Perfect weather for outdoor activities!"
    
    def _handle_calculation(self, request: str) -> str:
        """Handle mathematical calculations."""
        import re
        
        # Extract numbers and operators
        numbers = re.findall(r'\d+\.?\d*', request)
        
        if len(numbers) >= 2:
            try:
                # Simple calculation (in production, use safer evaluation)
                if '+' in request:
                    result = sum(float(n) for n in numbers)
                    return f"🔢 Calculation: {' + '.join(numbers)} = {result}"
                elif '-' in request:
                    result = float(numbers[0]) - float(numbers[1])
                    return f"🔢 Calculation: {numbers[0]} - {numbers[1]} = {result}"
                elif '*' in request or 'times' in request:
                    result = float(numbers[0]) * float(numbers[1])
                    return f"🔢 Calculation: {numbers[0]} × {numbers[1]} = {result}"
                elif '/' in request or 'divided' in request:
                    result = float(numbers[0]) / float(numbers[1])
                    return f"🔢 Calculation: {numbers[0]} ÷ {numbers[1]} = {result:.2f}"
            except:
                pass
        
        return "I can help with calculations. Please provide a clear mathematical expression."
    
    def _handle_general(self, request: str) -> str:
        """Handle general assistance requests."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a personalized virtual assistant for {self.user_preferences.name}. "
                      f"Be helpful, friendly, and proactive. Provide clear, actionable responses. "
                      f"User interests: {', '.join(self.user_preferences.interests or ['general topics'])}. "
                      f"Preferred language: {self.user_preferences.preferred_language}."),
            ("human", "{request}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"request": request})
        
        return response.content
    
    def _generate_proactive_suggestions(self) -> Optional[str]:
        """Generate proactive suggestions based on context."""
        current_hour = datetime.now().hour
        
        # Morning suggestion
        if 6 <= current_hour  0:
            next_event = min(self.calendar, key=lambda e: e.start_time)
            if next_event.start_time.date() == datetime.now().date():
                return f"You have '{next_event.title}' at {next_event.start_time.strftime('%H:%M')} today."
        
        # Pending tasks suggestion
        if len(self.todo_list) > 5:
            incomplete = [t for t in self.todo_list if not t.completed]
            if len(incomplete) > 3:
                return f"You have {len(incomplete)} pending tasks. Would you like me to prioritize them?"
        
        # Reminder check
        upcoming_reminders = [r for r in self.reminders 
                             if not r.completed and r.remind_at  List[CalendarEvent]:
        """Get schedule for a specific date."""
        target_date = date or datetime.now()
        return [e for e in self.calendar if e.start_time.date() == target_date.date()]
    
    def get_pending_tasks(self) -> List[TodoItem]:
        """Get all pending todo items."""
        return [t for t in self.todo_list if not t.completed]
    
    def complete_task(self, task_index: int) -> str:
        """Mark a todo task as completed."""
        if 0 <= task_index  str:
        """Get daily summary."""
        today_events = self.get_schedule()
        pending_tasks = self.get_pending_tasks()
        active_reminders = [r for r in self.reminders if not r.completed]
        
        summary = f"📊 Daily Summary for {self.user_preferences.name}:\n"
        summary += f"   • Events today: {len(today_events)}\n"
        summary += f"   • Pending tasks: {len(pending_tasks)}\n"
        summary += f"   • Active reminders: {len(active_reminders)}\n"
        
        return summary
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get assistant statistics."""
        return {
            "name": self.name,
            "user": self.user_preferences.name,
            "total_events": len(self.calendar),
            "total_reminders": len(self.reminders),
            "pending_tasks": len(self.get_pending_tasks()),
            **self.statistics
        }


class VirtualAssistantPattern:
    """
    Virtual Assistant Pattern - Manage personalized AI assistants.
    
    Coordinates multiple virtual assistants for different users.
    """
    
    def __init__(self):
        self.assistants: Dict[str, VirtualAssistant] = {}
    
    def create_assistant(self, name: str, user_preferences: UserPreferences) -> VirtualAssistant:
        """Create a new virtual assistant."""
        assistant = VirtualAssistant(name, user_preferences)
        self.assistants[name] = assistant
        print(f"   Created assistant: {name} for {user_preferences.name}")
        return assistant
    
    def run(self, assistant_name: str, requests: List[str]) -> Dict[str, Any]:
        """Run series of requests with assistant."""
        if assistant_name not in self.assistants:
            return {"error": f"Assistant {assistant_name} not found"}
        
        assistant = self.assistants[assistant_name]
        responses = []
        
        print(f"\n=== {assistant_name} Assistance Session ===\n")
        
        for request in requests:
            print(f"User: {request}")
            response = assistant.assist(request)
            print(f"{assistant_name}: {response}\n")
            responses.append(response)
        
        # Show summary
        print(assistant.get_summary())
        
        return {
            "assistant": assistant_name,
            "requests_handled": len(requests),
            "responses": responses,
            "statistics": assistant.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Virtual Assistant Pattern Demo ===\n")
    
    # Create pattern
    pattern = VirtualAssistantPattern()
    
    # Create user preferences
    print("1. Setting up virtual assistants...")
    
    user_prefs = UserPreferences(
        name="Alice",
        timezone="America/New_York",
        preferred_language="English",
        interests=["technology", "productivity", "health"]
    )
    
    assistant = pattern.create_assistant("Jarvis", user_prefs)
    
    # Add some initial data
    assistant.todo_list.append(TodoItem(task="Review project proposal", priority=Priority.HIGH))
    assistant.todo_list.append(TodoItem(task="Call dentist", priority=Priority.MEDIUM))
    
    # Run assistance session
    print("\n2. Running assistance session...")
    
    requests = [
        "What's on my schedule today?",
        "Add a reminder to buy groceries",
        "Schedule a meeting for tomorrow at 2 PM",
        "What's 25 times 4?",
        "Add 'Prepare presentation' to my todo list",
        "Search for information about artificial intelligence",
        "What's the weather like?",
        "Show me my pending tasks"
    ]
    
    result = pattern.run("Jarvis", requests)
    
    # Print statistics
    print("\n3. Assistant Statistics:")
    stats = result["statistics"]
    for key, value in stats.items():
        print(f"   {key}: {value}")
    
    print("\n✓ Virtual assistant provided personalized help and task management!")
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Virtual Assistant. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class VirtualAssistantPattern:
    """
    Virtual Assistant - Enhanced implementation
    
    This pattern provides comprehensive functionality for virtual assistant
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: VirtualAssistantConfig = None):
        self.config = config or VirtualAssistantConfig()
        self.agents: Dict[str, VirtualAssistantAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Virtual Assistant. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: VirtualAssistantAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Virtual Assistant",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Virtual Assistant",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Virtual Assistant Demo ===\n")
    
    # Create configuration
    config = VirtualAssistantConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = VirtualAssistantPattern(config)
    
    # Create and register agents
    agent1 = VirtualAssistantAgent("Agent1", config)
    agent2 = VirtualAssistantAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate virtual assistant functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Virtual Assistant

Pattern 233: 233 Virtual Assistant – Line-by-Line Explanation

Overview

Pattern 233: Virtual Assistant

Purpose: Personalized AI assistant that executes tasks, manages schedules, and provides proactive help. Handles calendar, reminders, information retrieval, and task management with user preferences.

Code Structure Analysis

File Information

Pattern Number: 233
Pattern Name: 233 Virtual Assistant
File Path: patterns\domain_specific\233_virtual_assistant.py
Total Lines: 640

Detailed Code Explanation

Module Docstring

"""
Pattern 233: Virtual Assistant

Purpose: Personalized AI assistant that executes tasks, manages schedules, and provides proactive help.
Handles calendar, reminders, information retrieval, and task management with user preferences.
"""

Explanation: This module docstring describes the purpose and key features of the 233 Virtual Assistant pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
        import re  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Design Patterns Used

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

Usage Example

# Usage example not available

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Chatbot


    """
Pattern 232: Chatbot

Purpose: Goal-oriented chatbot with intent recognition, dialogue flow, and response generation.
Handles predefined intents, manages conversation state, and provides structured responses.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import re
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class IntentType(Enum):
    """Common chatbot intents."""
    GREETING = "greeting"
    FAREWELL = "farewell"
    HELP = "help"
    FAQ = "faq"
    BOOKING = "booking"
    INQUIRY = "inquiry"
    COMPLAINT = "complaint"
    FEEDBACK = "feedback"
    CHITCHAT = "chitchat"
    UNKNOWN = "unknown"


class FlowState(Enum):
    """States in conversation flow."""
    INITIAL = "initial"
    COLLECTING_INFO = "collecting_info"
    CONFIRMING = "confirming"
    PROCESSING = "processing"
    COMPLETED = "completed"
    ERROR = "error"


@dataclass
class Intent:
    """Recognized user intent."""
    type: IntentType
    confidence: float
    entities: Dict[str, Any] = field(default_factory=dict)
    raw_input: str = ""


@dataclass
class Response:
    """Chatbot response."""
    text: str
    intent: IntentType
    follow_up_questions: List[str] = field(default_factory=list)
    actions: List[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ConversationFlow:
    """Manages conversation flow state."""
    current_state: FlowState = FlowState.INITIAL
    required_fields: List[str] = field(default_factory=list)
    collected_fields: Dict[str, Any] = field(default_factory=dict)
    flow_type: Optional[IntentType] = None


class Chatbot:
    """
    Chatbot - Goal-oriented dialogue with intent recognition and flow management.
    
    Features:
    - Intent recognition with pattern matching
    - Entity extraction
    - Dialogue flow management
    - FAQ handling
    - Context-aware responses
    """
    
    def __init__(self,
                 name: str,
                 domain: str = "general",
                 supported_intents: Optional[List[IntentType]] = None):
        self.name = name
        self.domain = domain
        self.supported_intents = supported_intents or [intent for intent in IntentType]
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
        
        # Intent patterns
        self.intent_patterns: Dict[IntentType, List[str]] = self._init_intent_patterns()
        
        # FAQ knowledge base
        self.faq_db: Dict[str, str] = {}
        
        # Conversation state
        self.current_flow: Optional[ConversationFlow] = None
        self.conversation_history: List[Dict[str, Any]] = []
        
        # Response handlers
        self.response_handlers: Dict[IntentType, Callable] = self._init_response_handlers()
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_messages": 0,
            "intent_counts": defaultdict(int),
            "avg_confidence": 0.0,
            "flow_completions": 0
        }
    
    def _init_intent_patterns(self) -> Dict[IntentType, List[str]]:
        """Initialize intent recognition patterns."""
        return {
            IntentType.GREETING: [
                r'\b(hi|hello|hey|greetings|good morning|good afternoon)\b',
                r'^(hi|hello|hey)[\s!.]*$'
            ],
            IntentType.FAREWELL: [
                r'\b(bye|goodbye|see you|farewell|take care)\b',
                r'^(bye|goodbye)[\s!.]*$'
            ],
            IntentType.HELP: [
                r'\b(help|assist|support|how do i|how can|what can)\b',
                r'^(help|assist me)\b'
            ],
            IntentType.FAQ: [
                r'\b(what is|tell me about|explain|how does|when is|where is)\b',
                r'\?$'
            ],
            IntentType.BOOKING: [
                r'\b(book|reserve|appointment|schedule|availability)\b',
                r'\b(i want to book|make a reservation)\b'
            ],
            IntentType.INQUIRY: [
                r'\b(information|details|more about|interested in)\b',
                r'\b(can you tell|would like to know)\b'
            ],
            IntentType.COMPLAINT: [
                r'\b(problem|issue|complaint|not working|broken|unhappy)\b',
                r'\b(dissatisfied|disappointed|frustrated)\b'
            ],
            IntentType.FEEDBACK: [
                r'\b(feedback|suggestion|recommend|improve|better)\b',
                r'\b(would like to suggest|my opinion)\b'
            ],
            IntentType.CHITCHAT: [
                r'\b(weather|today|how are you|nice day|weekend)\b',
                r'\b(thanks|thank you|appreciate)\b'
            ]
        }
    
    def _init_response_handlers(self) -> Dict[IntentType, Callable]:
        """Initialize intent-specific response handlers."""
        return {
            IntentType.GREETING: self._handle_greeting,
            IntentType.FAREWELL: self._handle_farewell,
            IntentType.HELP: self._handle_help,
            IntentType.FAQ: self._handle_faq,
            IntentType.BOOKING: self._handle_booking,
            IntentType.INQUIRY: self._handle_inquiry,
            IntentType.COMPLAINT: self._handle_complaint,
            IntentType.FEEDBACK: self._handle_feedback,
            IntentType.CHITCHAT: self._handle_chitchat,
            IntentType.UNKNOWN: self._handle_unknown
        }
    
    def chat(self, user_input: str) -> Response:
        """Process user input and generate response."""
        self.statistics["total_messages"] += 1
        
        # Recognize intent
        intent = self._recognize_intent(user_input)
        self.statistics["intent_counts"][intent.type.value] += 1
        
        # Extract entities
        entities = self._extract_entities(user_input, intent.type)
        intent.entities = entities
        
        # Generate response using appropriate handler
        handler = self.response_handlers.get(intent.type, self._handle_unknown)
        response = handler(intent)
        
        # Record in history
        self.conversation_history.append({
            "user_input": user_input,
            "intent": intent.type.value,
            "confidence": intent.confidence,
            "response": response.text,
            "timestamp": datetime.now()
        })
        
        # Update statistics
        self._update_statistics()
        
        return response
    
    def _recognize_intent(self, user_input: str) -> Intent:
        """Recognize user intent from input."""
        user_lower = user_input.lower()
        best_intent = IntentType.UNKNOWN
        max_confidence = 0.0
        
        # Pattern matching
        for intent_type, patterns in self.intent_patterns.items():
            if intent_type not in self.supported_intents:
                continue
            
            for pattern in patterns:
                if re.search(pattern, user_lower):
                    # Simple confidence based on pattern specificity
                    confidence = 0.7 + (0.3 * len(pattern) / 100)
                    if confidence > max_confidence:
                        max_confidence = confidence
                        best_intent = intent_type
        
        # If no pattern matched, use LLM for classification
        if max_confidence  max_confidence:
                best_intent = llm_intent
                max_confidence = llm_confidence
        
        return Intent(
            type=best_intent,
            confidence=min(1.0, max_confidence),
            raw_input=user_input
        )
    
    def _llm_classify_intent(self, user_input: str) -> tuple:
        """Use LLM to classify intent when patterns don't match."""
        intent_options = ", ".join([i.value for i in self.supported_intents])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are an intent classifier for a {self.domain} chatbot. "
                      f"Classify the user input into one of these intents: {intent_options}. "
                      f"Respond with ONLY the intent name."),
            ("human", "{input}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input": user_input})
        
        # Parse response
        intent_str = response.content.strip().lower()
        
        for intent_type in self.supported_intents:
            if intent_type.value in intent_str:
                return intent_type, 0.6
        
        return IntentType.UNKNOWN, 0.3
    
    def _extract_entities(self, user_input: str, intent_type: IntentType) -> Dict[str, Any]:
        """Extract entities from user input."""
        entities = {}
        
        # Simple entity extraction patterns
        # Date/time
        date_pattern = r'\b(\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{4}-\d{2}-\d{2})\b'
        dates = re.findall(date_pattern, user_input)
        if dates:
            entities["date"] = dates[0]
        
        # Time
        time_pattern = r'\b(\d{1,2}:\d{2}\s*(?:am|pm)?)\b'
        times = re.findall(time_pattern, user_input.lower())
        if times:
            entities["time"] = times[0]
        
        # Email
        email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
        emails = re.findall(email_pattern, user_input)
        if emails:
            entities["email"] = emails[0]
        
        # Phone
        phone_pattern = r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b'
        phones = re.findall(phone_pattern, user_input)
        if phones:
            entities["phone"] = phones[0]
        
        # Numbers
        number_pattern = r'\b\d+\b'
        numbers = re.findall(number_pattern, user_input)
        if numbers and intent_type == IntentType.BOOKING:
            entities["quantity"] = numbers[0]
        
        return entities
    
    # Intent handlers
    def _handle_greeting(self, intent: Intent) -> Response:
        """Handle greeting intent."""
        responses = [
            f"Hello! I'm {self.name}, your {self.domain} assistant. How can I help you today?",
            f"Hi there! Welcome to {self.name}. What can I do for you?",
            f"Greetings! I'm here to assist you with {self.domain}. What do you need?"
        ]
        
        import random
        return Response(
            text=random.choice(responses),
            intent=IntentType.GREETING,
            follow_up_questions=["What can I help you with today?"]
        )
    
    def _handle_farewell(self, intent: Intent) -> Response:
        """Handle farewell intent."""
        responses = [
            "Goodbye! Feel free to come back if you need anything!",
            "Take care! Have a great day!",
            "Bye! Thanks for chatting with me!"
        ]
        
        import random
        return Response(
            text=random.choice(responses),
            intent=IntentType.FAREWELL
        )
    
    def _handle_help(self, intent: Intent) -> Response:
        """Handle help request."""
        capabilities = [
            f"- Answer questions about {self.domain}",
            "- Help with bookings and appointments",
            "- Provide information and support",
            "- Handle complaints and feedback"
        ]
        
        help_text = f"I can help you with:\n" + "\n".join(capabilities)
        
        return Response(
            text=help_text,
            intent=IntentType.HELP,
            follow_up_questions=["What would you like to do?"]
        )
    
    def _handle_faq(self, intent: Intent) -> Response:
        """Handle FAQ query."""
        # Check if question matches FAQ database
        for question, answer in self.faq_db.items():
            if any(word in intent.raw_input.lower() for word in question.lower().split()):
                return Response(
                    text=answer,
                    intent=IntentType.FAQ,
                    follow_up_questions=["Is there anything else you'd like to know?"]
                )
        
        # Use LLM for dynamic FAQ
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are a helpful {self.domain} assistant. "
                      f"Provide clear, concise answers to user questions."),
            ("human", "{question}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"question": intent.raw_input})
        
        return Response(
            text=response.content,
            intent=IntentType.FAQ,
            follow_up_questions=["Do you have any other questions?"]
        )
    
    def _handle_booking(self, intent: Intent) -> Response:
        """Handle booking request."""
        # Initialize or update booking flow
        if not self.current_flow or self.current_flow.flow_type != IntentType.BOOKING:
            self.current_flow = ConversationFlow(
                current_state=FlowState.COLLECTING_INFO,
                required_fields=["date", "time", "email"],
                flow_type=IntentType.BOOKING
            )
        
        # Collect entities
        for field, value in intent.entities.items():
            if field in self.current_flow.required_fields:
                self.current_flow.collected_fields[field] = value
        
        # Check what's still missing
        missing_fields = [f for f in self.current_flow.required_fields 
                         if f not in self.current_flow.collected_fields]
        
        if missing_fields:
            field = missing_fields[0]
            return Response(
                text=f"I'd be happy to help you book! What {field} would you prefer?",
                intent=IntentType.BOOKING,
                follow_up_questions=[f"Please provide your {field}"]
            )
        else:
            # All info collected
            self.current_flow.current_state = FlowState.CONFIRMING
            booking_details = ", ".join([f"{k}: {v}" for k, v in 
                                        self.current_flow.collected_fields.items()])
            return Response(
                text=f"Great! I've got your booking for {booking_details}. Shall I confirm?",
                intent=IntentType.BOOKING,
                follow_up_questions=["Reply 'yes' to confirm or 'no' to cancel"],
                actions=["prepare_booking"]
            )
    
    def _handle_inquiry(self, intent: Intent) -> Response:
        """Handle general inquiry."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are a knowledgeable {self.domain} assistant. "
                      f"Provide helpful information about user inquiries."),
            ("human", "{inquiry}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"inquiry": intent.raw_input})
        
        return Response(
            text=response.content,
            intent=IntentType.INQUIRY,
            follow_up_questions=["Would you like more details about anything?"]
        )
    
    def _handle_complaint(self, intent: Intent) -> Response:
        """Handle complaint."""
        return Response(
            text="I'm sorry to hear you're experiencing issues. Let me help resolve this. "
                 "Can you provide more details about the problem?",
            intent=IntentType.COMPLAINT,
            follow_up_questions=["What specific issue are you facing?"],
            actions=["escalate_to_support"]
        )
    
    def _handle_feedback(self, intent: Intent) -> Response:
        """Handle feedback."""
        return Response(
            text="Thank you for your feedback! We really appreciate your input. "
                 "Your suggestions help us improve our service.",
            intent=IntentType.FEEDBACK,
            actions=["record_feedback"]
        )
    
    def _handle_chitchat(self, intent: Intent) -> Response:
        """Handle casual conversation."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a friendly chatbot. Engage in brief, pleasant conversation."),
            ("human", "{message}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"message": intent.raw_input})
        
        return Response(
            text=response.content,
            intent=IntentType.CHITCHAT
        )
    
    def _handle_unknown(self, intent: Intent) -> Response:
        """Handle unknown intent."""
        return Response(
            text="I'm not sure I understood that. Could you rephrase, or ask me for help?",
            intent=IntentType.UNKNOWN,
            follow_up_questions=["Try asking in a different way", "Type 'help' to see what I can do"]
        )
    
    def add_faq(self, question: str, answer: str):
        """Add FAQ to knowledge base."""
        self.faq_db[question.lower()] = answer
        print(f"   Added FAQ: {question}")
    
    def _update_statistics(self):
        """Update chatbot statistics."""
        if self.conversation_history:
            confidences = [h["confidence"] for h in self.conversation_history]
            self.statistics["avg_confidence"] = sum(confidences) / len(confidences)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get chatbot statistics."""
        return {
            "name": self.name,
            "domain": self.domain,
            "total_messages": self.statistics["total_messages"],
            "intent_distribution": dict(self.statistics["intent_counts"]),
            "avg_confidence": self.statistics["avg_confidence"],
            "faq_count": len(self.faq_db)
        }


class ChatbotPattern:
    """
    Chatbot Pattern - Manage goal-oriented chatbots.
    
    Coordinates multiple chatbots for different domains.
    """
    
    def __init__(self):
        self.chatbots: Dict[str, Chatbot] = {}
    
    def register_chatbot(self, chatbot: Chatbot):
        """Register a chatbot."""
        self.chatbots[chatbot.name] = chatbot
        print(f"   Registered chatbot: {chatbot.name} ({chatbot.domain})")
    
    def run(self, chatbot_name: str, conversation: List[str]) -> Dict[str, Any]:
        """Run conversation with chatbot."""
        if chatbot_name not in self.chatbots:
            return {"error": f"Chatbot {chatbot_name} not found"}
        
        chatbot = self.chatbots[chatbot_name]
        responses = []
        
        print(f"\n=== Conversation with {chatbot_name} ===\n")
        
        for user_input in conversation:
            print(f"User: {user_input}")
            response = chatbot.chat(user_input)
            print(f"{chatbot_name}: {response.text}")
            if response.follow_up_questions:
                print(f"   Follow-up: {response.follow_up_questions[0]}")
            print()
            responses.append(response)
        
        return {
            "chatbot": chatbot_name,
            "turns": len(conversation),
            "responses": responses,
            "statistics": chatbot.get_statistics()
        }


# Example usage
if __name__ == "__main__":
    print("=== Chatbot Pattern Demo ===\n")
    
    # Create pattern
    pattern = ChatbotPattern()
    
    # Create chatbots for different domains
    print("1. Creating chatbots...")
    
    customer_service = Chatbot(
        name="CustomerServiceBot",
        domain="customer service",
        supported_intents=[IntentType.GREETING, IntentType.HELP, IntentType.FAQ, 
                          IntentType.COMPLAINT, IntentType.FAREWELL]
    )
    
    # Add FAQs
    customer_service.add_faq("What are your hours", "We're open Monday-Friday, 9 AM to 5 PM.")
    customer_service.add_faq("How do I return", "Returns are accepted within 30 days with receipt.")
    customer_service.add_faq("Shipping cost", "Shipping is free for orders over $50.")
    
    pattern.register_chatbot(customer_service)
    
    booking_bot = Chatbot(
        name="BookingBot",
        domain="restaurant reservations",
        supported_intents=[IntentType.GREETING, IntentType.BOOKING, IntentType.INQUIRY, IntentType.FAREWELL]
    )
    pattern.register_chatbot(booking_bot)
    
    # Run conversations
    print("\n2. Running conversations...")
    
    # Customer service conversation
    cs_conversation = [
        "Hello!",
        "I have a problem with my order",
        "It arrived damaged",
        "Thank you for your help"
    ]
    
    result1 = pattern.run("CustomerServiceBot", cs_conversation)
    
    # Booking conversation
    booking_conversation = [
        "Hi, I'd like to make a reservation",
        "For tomorrow at 7 PM",
        "My email is john@example.com",
        "Yes, please confirm"
    ]
    
    result2 = pattern.run("BookingBot", booking_conversation)
    
    # Print statistics
    print("\n3. Chatbot Statistics:")
    for name, chatbot in pattern.chatbots.items():
        stats = chatbot.get_statistics()
        print(f"\n   {name}:")
        for key, value in stats.items():
            if isinstance(value, float):
                print(f"      {key}: {value:.3f}")
            elif isinstance(value, dict):
                print(f"      {key}:")
                for k, v in value.items():
                    print(f"         {k}: {v}")
            else:
                print(f"      {key}: {value}")
    
    print("\n✓ Chatbots handled conversations with intent recognition and flow!")

Agentic Architecture Pattern: Chatbot

Pattern 232: 232 Chatbot – Line-by-Line Explanation

Overview

Pattern 232: Chatbot

Purpose: Goal-oriented chatbot with intent recognition, dialogue flow, and response generation. Handles predefined intents, manages conversation state, and provides structured responses.

Code Structure Analysis

File Information

Pattern Number: 232
Pattern Name: 232 Chatbot
File Path: patterns\domain_specific\232_chatbot.py
Total Lines: 596

Detailed Code Explanation

Module Docstring

"""
Pattern 232: Chatbot

Purpose: Goal-oriented chatbot with intent recognition, dialogue flow, and response generation.
Handles predefined intents, manages conversation state, and provides structured responses.
"""

Explanation: This module docstring describes the purpose and key features of the 232 Chatbot pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import re
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
        import random
        import random  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

IntentType (Line 18)

class IntentType(Enum):
    GREETING
    FAREWELL
    HELP
    FAQ
    BOOKING
    INQUIRY
    COMPLAINT
    FEEDBACK
    CHITCHAT
    UNKNOWN

Purpose: Common chatbot intents.

Values: – GREETING: Represents greeting – FAREWELL: Represents farewell – HELP: Represents help – FAQ: Represents faq – BOOKING: Represents booking – INQUIRY: Represents inquiry – COMPLAINT: Represents complaint – FEEDBACK: Represents feedback – CHITCHAT: Represents chitchat – UNKNOWN: Represents unknown

FlowState (Line 32)

class FlowState(Enum):
    INITIAL
    COLLECTING_INFO
    CONFIRMING
    PROCESSING
    COMPLETED
    ERROR

Purpose: States in conversation flow.

Values: – INITIAL: Represents initial – COLLECTING_INFO: Represents collecting info – CONFIRMING: Represents confirming – PROCESSING: Represents processing – COMPLETED: Successfully finished – ERROR: Error severity level

Classes

IntentType (Line 18)

Description: Common chatbot intents.

FlowState (Line 32)

Description: States in conversation flow.

Intent (Line 43)

Description: Recognized user intent.

Response (Line 52)

Description: Chatbot response.

ConversationFlow (Line 62)

Description: Manages conversation flow state.

Chatbot (Line 70)

Description: Chatbot – Goal-oriented dialogue with intent recognition and flow management.

Features: – Intent recognition with pattern matching – Entity extraction – Dialogue flow management – FAQ handling – Context-aware responses

Methods:

`init` (Line 82)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – domain: Parameter for domain – supported_intents: Parameter for supported_intents

`_init_intent_patterns` (Line 112)

def _init_intent_patterns(self):
    """Initialize intent recognition patterns."""

Purpose: Initialize intent recognition patterns.

Parameters: – self: Instance reference

`_init_response_handlers` (Line 153)

def _init_response_handlers(self):
    """Initialize intent-specific response handlers."""

Purpose: Initialize intent-specific response handlers.

Parameters: – self: Instance reference

`chat` (Line 168)

def chat(self, user_input):
    """Process user input and generate response."""

Purpose: Process user input and generate response.

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

`_recognize_intent` (Line 198)

def _recognize_intent(self, user_input):
    """Recognize user intent from input."""

Purpose: Recognize user intent from input.

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

`_llm_classify_intent` (Line 230)

def _llm_classify_intent(self, user_input):
    """Use LLM to classify intent when patterns don't match."""

Purpose: Use LLM to classify intent when patterns don’t match.

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

`_extract_entities` (Line 253)

def _extract_entities(self, user_input, intent_type):
    """Extract entities from user input."""

Purpose: Extract entities from user input.

Parameters: – self: Instance reference – user_input: Parameter for user_input – intent_type: Parameter for intent_type

`_handle_greeting` (Line 291)

def _handle_greeting(self, intent):
    """Handle greeting intent."""

Purpose: Handle greeting intent.

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

`_handle_farewell` (Line 306)

def _handle_farewell(self, intent):
    """Handle farewell intent."""

Purpose: Handle farewell intent.

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

`_handle_help` (Line 320)

def _handle_help(self, intent):
    """Handle help request."""

Purpose: Handle help request.

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

`_handle_faq` (Line 337)

def _handle_faq(self, intent):
    """Handle FAQ query."""

Purpose: Handle FAQ query.

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

`_handle_booking` (Line 364)

def _handle_booking(self, intent):
    """Handle booking request."""

Purpose: Handle booking request.

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

`_handle_inquiry` (Line 402)

def _handle_inquiry(self, intent):
    """Handle general inquiry."""

Purpose: Handle general inquiry.

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

`_handle_complaint` (Line 419)

def _handle_complaint(self, intent):
    """Handle complaint."""

Purpose: Handle complaint.

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

`_handle_feedback` (Line 429)

def _handle_feedback(self, intent):
    """Handle feedback."""

Purpose: Handle feedback.

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

…and 5 more methods

ChatbotPattern (Line 484)

Description: Chatbot Pattern – Manage goal-oriented chatbots.

Coordinates multiple chatbots for different domains.

Methods:

`init` (Line 491)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_chatbot` (Line 494)

def register_chatbot(self, chatbot):
    """Register a chatbot."""

Purpose: Register a chatbot.

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

`run` (Line 499)

def run(self, chatbot_name, conversation):
    """Run conversation with chatbot."""

Purpose: Run conversation with chatbot.

Parameters: – self: Instance reference – chatbot_name: Parameter for chatbot_name – conversation: Parameter for conversation

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Agentic Architecture Pattern: Conversational Agent


    """
Pattern 231: Conversational Agent

Purpose: Engage in natural multi-turn dialogue with context awareness and personality.
Maintains conversation history, understands context, and provides coherent responses.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage


class ConversationMode(Enum):
    """Modes of conversation."""
    CASUAL = "casual"  # Friendly, informal chat
    PROFESSIONAL = "professional"  # Formal, business-like
    SUPPORTIVE = "supportive"  # Empathetic, helpful
    EDUCATIONAL = "educational"  # Explanatory, teaching


@dataclass
class Message:
    """A single message in conversation."""
    role: str  # "user" or "assistant"
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class ConversationTurn:
    """A complete turn in the conversation."""
    turn_number: int
    user_message: str
    agent_response: str
    context_used: List[str]
    intent: Optional[str] = None
    sentiment: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ConversationContext:
    """Context information for conversation."""
    user_name: Optional[str] = None
    user_preferences: Dict[str, Any] = field(default_factory=dict)
    topics_discussed: List[str] = field(default_factory=list)
    current_topic: Optional[str] = None
    conversation_goal: Optional[str] = None


class ConversationalAgent:
    """
    Conversational Agent - Natural multi-turn dialogue with context awareness.
    
    Engages in coherent conversations by maintaining history, understanding
    context, adapting tone, and providing relevant responses.
    """
    
    def __init__(self,
                 name: str,
                 personality: str = "helpful and friendly",
                 mode: ConversationMode = ConversationMode.CASUAL,
                 max_history: int = 20):
        self.name = name
        self.personality = personality
        self.mode = mode
        self.max_history = max_history
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.8)
        
        # Conversation state
        self.conversation_history: deque = deque(maxlen=max_history)
        self.turns: List[ConversationTurn] = []
        self.context = ConversationContext()
        self.current_turn = 0
        
        # Statistics
        self.statistics: Dict[str, Any] = {
            "total_turns": 0,
            "total_messages": 0,
            "topics_covered": 0,
            "avg_response_length": 0.0
        }
    
    def converse(self, user_input: str) -> str:
        """Process user input and generate response."""
        self.current_turn += 1
        
        # Add user message to history
        user_msg = Message(role="user", content=user_input)
        self.conversation_history.append(user_msg)
        
        # Analyze input
        intent = self._detect_intent(user_input)
        sentiment = self._detect_sentiment(user_input)
        
        # Extract context
        context_items = self._extract_context(user_input)
        
        # Generate response with context
        response = self._generate_response(user_input, context_items)
        
        # Add assistant message to history
        agent_msg = Message(role="assistant", content=response)
        self.conversation_history.append(agent_msg)
        
        # Record turn
        turn = ConversationTurn(
            turn_number=self.current_turn,
            user_message=user_input,
            agent_response=response,
            context_used=context_items,
            intent=intent,
            sentiment=sentiment
        )
        self.turns.append(turn)
        
        # Update statistics
        self.statistics["total_turns"] = self.current_turn
        self.statistics["total_messages"] = len(self.conversation_history)
        self._update_statistics()
        
        return response
    
    def _detect_intent(self, user_input: str) -> str:
        """Detect user intent from input."""
        user_lower = user_input.lower()
        
        if any(word in user_lower for word in ["?", "what", "how", "why", "when", "where"]):
            return "question"
        elif any(word in user_lower for word in ["please", "can you", "could you", "would you"]):
            return "request"
        elif any(word in user_lower for word in ["thank", "thanks", "appreciate"]):
            return "gratitude"
        elif any(word in user_lower for word in ["hello", "hi", "hey", "greetings"]):
            return "greeting"
        elif any(word in user_lower for word in ["bye", "goodbye", "see you"]):
            return "farewell"
        else:
            return "statement"
    
    def _detect_sentiment(self, user_input: str) -> str:
        """Detect sentiment of user input."""
        user_lower = user_input.lower()
        
        positive_words = ["good", "great", "excellent", "happy", "love", "wonderful"]
        negative_words = ["bad", "terrible", "hate", "sad", "angry", "frustrated"]
        
        positive_count = sum(1 for word in positive_words if word in user_lower)
        negative_count = sum(1 for word in negative_words if word in user_lower)
        
        if positive_count > negative_count:
            return "positive"
        elif negative_count > positive_count:
            return "negative"
        else:
            return "neutral"
    
    def _extract_context(self, user_input: str) -> List[str]:
        """Extract relevant context from conversation history."""
        context_items = []
        
        # Add recent conversation context
        recent_messages = list(self.conversation_history)[-4:]  # Last 2 turns
        if len(recent_messages) > 1:
            context_items.append(f"Recent context: {recent_messages[-2].content[:50]}...")
        
        # Add current topic if set
        if self.context.current_topic:
            context_items.append(f"Current topic: {self.context.current_topic}")
        
        # Add user preferences if relevant
        if self.context.user_preferences:
            context_items.append(f"User preferences: {list(self.context.user_preferences.keys())[:2]}")
        
        return context_items
    
    def _generate_response(self, user_input: str, context_items: List[str]) -> str:
        """Generate contextually appropriate response."""
        # Build conversation history for LLM
        messages = []
        
        # System message with personality and mode
        system_prompt = f"""You are {self.name}, a {self.personality} conversational agent.
Conversation mode: {self.mode.value}.
Maintain context awareness and provide coherent, relevant responses.
Be natural, engaging, and adapt to the user's tone."""
        messages.append(SystemMessage(content=system_prompt))
        
        # Add recent conversation history
        for msg in list(self.conversation_history)[-6:]:  # Last 3 turns
            if msg.role == "user":
                messages.append(HumanMessage(content=msg.content))
            else:
                messages.append(AIMessage(content=msg.content))
        
        # Add current user message if not already in history
        if not self.conversation_history or self.conversation_history[-1].content != user_input:
            messages.append(HumanMessage(content=user_input))
        
        # Generate response
        response = self.llm.invoke(messages)
        
        return response.content if hasattr(response, 'content') else str(response)
    
    def set_context(self, key: str, value: Any):
        """Set context information."""
        if key == "user_name":
            self.context.user_name = value
        elif key == "topic":
            self.context.current_topic = value
            if value not in self.context.topics_discussed:
                self.context.topics_discussed.append(value)
        elif key == "goal":
            self.context.conversation_goal = value
        else:
            self.context.user_preferences[key] = value
        
        print(f"   Context updated: {key} = {value}")
    
    def switch_mode(self, mode: ConversationMode):
        """Switch conversation mode."""
        old_mode = self.mode
        self.mode = mode
        print(f"   Switched from {old_mode.value} to {mode.value} mode")
    
    def get_conversation_summary(self) -> Dict[str, Any]:
        """Get summary of conversation."""
        if not self.turns:
            return {"total_turns": 0}
        
        intents = {}
        sentiments = {}
        for turn in self.turns:
            intents[turn.intent] = intents.get(turn.intent, 0) + 1
            sentiments[turn.sentiment] = sentiments.get(turn.sentiment, 0) + 1
        
        return {
            "total_turns": len(self.turns),
            "topics_discussed": self.context.topics_discussed,
            "intent_distribution": intents,
            "sentiment_distribution": sentiments,
            "current_topic": self.context.current_topic,
            "conversation_mode": self.mode.value
        }
    
    def _update_statistics(self):
        """Update conversation statistics."""
        if self.turns:
            total_length = sum(len(turn.agent_response) for turn in self.turns)
            self.statistics["avg_response_length"] = total_length / len(self.turns)
            self.statistics["topics_covered"] = len(self.context.topics_discussed)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive statistics."""
        return {
            "name": self.name,
            "personality": self.personality,
            "mode": self.mode.value,
            **self.statistics
        }
    
    def reset_conversation(self):
        """Reset conversation state."""
        self.conversation_history.clear()
        self.turns.clear()
        self.context = ConversationContext()
        self.current_turn = 0
        print(f"   Conversation reset for {self.name}")


class ConversationalPattern:
    """
    Conversational Pattern - Manage natural dialogue agents.
    
    Coordinates conversational agents for multi-turn interactions.
    """
    
    def __init__(self):
        self.agents: Dict[str, ConversationalAgent] = {}
    
    def register_agent(self, agent: ConversationalAgent):
        """Register conversational agent."""
        self.agents[agent.name] = agent
        print(f"   Registered agent: {agent.name}")
    
    def run(self, agent_name: str, conversation: List[str]) -> Dict[str, Any]:
        """Run conversation with agent."""
        if agent_name not in self.agents:
            return {"error": f"Agent {agent_name} not found"}
        
        agent = self.agents[agent_name]
        responses = []
        
        print(f"\n=== Conversation with {agent_name} ===\n")
        
        for user_input in conversation:
            print(f"User: {user_input}")
            response = agent.converse(user_input)
            print(f"{agent_name}: {response}\n")
            responses.append(response)
        
        summary = agent.get_conversation_summary()
        
        return {
            "agent": agent_name,
            "turns": len(conversation),
            "responses": responses,
            "summary": summary
        }


# Example usage
if __name__ == "__main__":
    print("=== Conversational Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = ConversationalPattern()
    
    # Create conversational agents with different personalities
    print("1. Creating conversational agents...")
    
    friendly_agent = ConversationalAgent(
        name="FriendlyBot",
        personality="warm, empathetic, and encouraging",
        mode=ConversationMode.SUPPORTIVE
    )
    pattern.register_agent(friendly_agent)
    
    professional_agent = ConversationalAgent(
        name="ProfessionalBot",
        personality="professional, concise, and efficient",
        mode=ConversationMode.PROFESSIONAL
    )
    pattern.register_agent(professional_agent)
    
    # Set context
    print("\n2. Setting context...")
    friendly_agent.set_context("user_name", "Alex")
    friendly_agent.set_context("topic", "productivity")
    
    # Run conversations
    print("\n3. Running conversations...")
    
    friendly_conversation = [
        "Hi! I'm feeling a bit overwhelmed with work.",
        "I have so many tasks and don't know where to start.",
        "That's helpful! What about prioritizing them?",
        "Thanks for the advice!"
    ]
    
    result1 = pattern.run("FriendlyBot", friendly_conversation)
    
    professional_conversation = [
        "I need help organizing my project timeline.",
        "The deadline is in 3 weeks.",
        "What's the best approach for risk management?"
    ]
    
    result2 = pattern.run("ProfessionalBot", professional_conversation)
    
    # Print summaries
    print("\n4. Conversation Summaries:")
    for agent_name, agent in pattern.agents.items():
        summary = agent.get_conversation_summary()
        print(f"\n   {agent_name}:")
        print(f"      Total turns: {summary['total_turns']}")
        print(f"      Topics: {summary.get('topics_discussed', [])}")
        print(f"      Intents: {summary.get('intent_distribution', {})}")
        print(f"      Sentiments: {summary.get('sentiment_distribution', {})}")
    
    # Statistics
    print("\n5. Agent Statistics:")
    for agent_name, agent in pattern.agents.items():
        stats = agent.get_statistics()
        print(f"\n   {agent_name}:")
        for key, value in stats.items():
            if isinstance(value, float):
                print(f"      {key}: {value:.2f}")
            else:
                print(f"      {key}: {value}")
    
    print("\n✓ Conversational agents engaged in natural dialogue!")

Agentic Architecture Pattern: Conversational Agent

Pattern 231: 231 Conversational Agent – Line-by-Line Explanation

Overview

Pattern 231: Conversational Agent

Purpose: Engage in natural multi-turn dialogue with context awareness and personality. Maintains conversation history, understands context, and provides coherent responses.

Code Structure Analysis

File Information

Pattern Number: 231
Pattern Name: 231 Conversational Agent
File Path: patterns\domain_specific\231_conversational_agent.py
Total Lines: 389

Detailed Code Explanation

Module Docstring

"""
Pattern 231: Conversational Agent

Purpose: Engage in natural multi-turn dialogue with context awareness and personality.
Maintains conversation history, understands context, and provides coherent responses.
"""

Explanation: This module docstring describes the purpose and key features of the 231 Conversational Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import deque
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage  # First 20 imports shown

Enumerations

ConversationMode (Line 18)

class ConversationMode(Enum):
    CASUAL
    PROFESSIONAL
    SUPPORTIVE
    EDUCATIONAL

Purpose: Modes of conversation.

Values: – CASUAL: Represents casual – PROFESSIONAL: Represents professional – SUPPORTIVE: Represents supportive – EDUCATIONAL: Represents educational

Classes

ConversationMode (Line 18)

Description: Modes of conversation.

Message (Line 27)

Description: A single message in conversation.

ConversationTurn (Line 36)

Description: A complete turn in the conversation.

ConversationContext (Line 48)

Description: Context information for conversation.

ConversationalAgent (Line 57)

Description: Conversational Agent – Natural multi-turn dialogue with context awareness.

Engages in coherent conversations by maintaining history, understanding context, adapting tone, and providing relevant responses.

Methods:

`init` (Line 65)

def __init__(self, name, personality, mode, max_history):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – personality: Parameter for personality – mode: Parameter for mode – max_history: Parameter for max_history

`converse` (Line 90)

def converse(self, user_input):
    """Process user input and generate response."""

Purpose: Process user input and generate response.

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

`_detect_intent` (Line 130)

def _detect_intent(self, user_input):
    """Detect user intent from input."""

Purpose: Detect user intent from input.

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

`_detect_sentiment` (Line 147)

def _detect_sentiment(self, user_input):
    """Detect sentiment of user input."""

Purpose: Detect sentiment of user input.

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

`_extract_context` (Line 164)

def _extract_context(self, user_input):
    """Extract relevant context from conversation history."""

Purpose: Extract relevant context from conversation history.

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

`_generate_response` (Line 183)

def _generate_response(self, user_input, context_items):
    """Generate contextually appropriate response."""

Purpose: Generate contextually appropriate response.

Parameters: – self: Instance reference – user_input: Parameter for user_input – context_items: Parameter for context_items

`set_context` (Line 211)

def set_context(self, key, value):
    """Set context information."""

Purpose: Set context information.

Parameters: – self: Instance reference – key: Parameter for key – value: Parameter for value

`switch_mode` (Line 226)

def switch_mode(self, mode):
    """Switch conversation mode."""

Purpose: Switch conversation mode.

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

`get_conversation_summary` (Line 232)

def get_conversation_summary(self):
    """Get summary of conversation."""

Purpose: Get summary of conversation.

Parameters: – self: Instance reference

`_update_statistics` (Line 252)

def _update_statistics(self):
    """Update conversation statistics."""

Purpose: Update conversation statistics.

Parameters: – self: Instance reference

`get_statistics` (Line 259)

def get_statistics(self):
    """Get comprehensive statistics."""

Purpose: Get comprehensive statistics.

Parameters: – self: Instance reference

`reset_conversation` (Line 268)

def reset_conversation(self):
    """Reset conversation state."""

Purpose: Reset conversation state.

Parameters: – self: Instance reference

ConversationalPattern (Line 277)

Description: Conversational Pattern – Manage natural dialogue agents.

Coordinates conversational agents for multi-turn interactions.

Methods:

`init` (Line 284)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 287)

def register_agent(self, agent):
    """Register conversational agent."""

Purpose: Register conversational agent.

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

`run` (Line 292)

def run(self, agent_name, conversation):
    """Run conversation with agent."""

Purpose: Run conversation with agent.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – conversation: Parameter for conversation

Design Patterns Used

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

Usage Example

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

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

# Check results
print(result)

Key Concepts

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

Extension Points

November 4, 2025

Category: aiml-architecture

Agentic Architecture Pattern: Decision Support Agent

Agentic Architecture Pattern: Decision Support Agent

Pattern 240: 240 Decision Support Agent – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Enumerations

DecisionType (Line 18)

RiskLevel (Line 28)

Classes

DecisionType (Line 18)

RiskLevel (Line 28)

DecisionCriteria (Line 37)

Alternative (Line 46)

DecisionResult (Line 56)

DecisionSupportAgent (Line 69)

__init__ (Line 84)

evaluate_decision (Line 96)

_rank_alternatives (Line 141)

_assess_risks (Line 177)

_generate_recommendation (Line 214)

_generate_reasoning (Line 240)

_calculate_confidence (Line 262)

compare_scenarios (Line 291)

get_statistics (Line 314)

DecisionSupportPattern (Line 322)

__init__ (Line 329)

create_agent (Line 332)

run (Line 339)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Data Analysis Agent

Agentic Architecture Pattern: Data Analysis Agent

Pattern 239: 239 Data Analysis Agent – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Enumerations

AnalysisType (Line 19)

Classes

AnalysisType (Line 19)

Dataset (Line 30)

__post_init__ (Line 37)

AnalysisResult (Line 43)

DataAnalysisAgent (Line 54)

__init__ (Line 69)

analyze (Line 81)

_descriptive_statistics (Line 126)

_correlation_analysis (Line 151)

_trend_analysis (Line 174)

_distribution_analysis (Line 202)

_anomaly_detection (Line 226)

_comparative_analysis (Line 248)

_pearson_correlation (Line 286)

_generate_insights (Line 303)

_generate_recommendations (Line 324)

_suggest_visualizations (Line 340)

_calculate_confidence (Line 353)

get_statistics (Line 369)

DataAnalysisPattern (Line 377)

__init__ (Line 384)

create_agent (Line 387)

run (Line 394)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Code Generation Agent

Agentic Architecture Pattern: Code Generation Agent

Pattern 238: 238 Code Generation Agent – Line-by-Line Explanation

Overview

`init` (Line 84)

`evaluate_decision` (Line 96)

`_rank_alternatives` (Line 141)

`_assess_risks` (Line 177)

`_generate_recommendation` (Line 214)

`_generate_reasoning` (Line 240)

`_calculate_confidence` (Line 262)

`compare_scenarios` (Line 291)

`get_statistics` (Line 314)

`init` (Line 329)

`create_agent` (Line 332)

`run` (Line 339)

`__post_init__` (Line 37)

`init` (Line 69)

`analyze` (Line 81)

`_descriptive_statistics` (Line 126)

`_correlation_analysis` (Line 151)

`_trend_analysis` (Line 174)

`_distribution_analysis` (Line 202)

`_anomaly_detection` (Line 226)

`_comparative_analysis` (Line 248)

`_pearson_correlation` (Line 286)

`_generate_insights` (Line 303)

`_generate_recommendations` (Line 324)

`_suggest_visualizations` (Line 340)

`_calculate_confidence` (Line 353)

`get_statistics` (Line 369)

`init` (Line 384)

`create_agent` (Line 387)

`run` (Line 394)

`init` (Line 79)

`generate` (Line 93)

`_generate_code` (Line 137)

`_generate_tests` (Line 171)

`_generate_documentation` (Line 202)

`_extract_code_block` (Line 216)

`_calculate_complexity` (Line 228)

`_check_error_handling` (Line 244)

`_identify_best_practices` (Line 258)

`generate_from_description` (Line 283)

`get_statistics` (Line 294)

`init` (Line 310)

`create_agent` (Line 313)

`run` (Line 320)

`init` (Line 75)

`translate` (Line 95)

`_detect_language` (Line 151)

`_perform_translation` (Line 165)

`_generate_alternatives` (Line 207)

`_extract_terms` (Line 226)

`_check_translation_memory` (Line 237)

`_add_to_memory` (Line 260)

`_calculate_confidence` (Line 276)

`add_terminology` (Line 295)

`batch_translate` (Line 299)

`get_statistics` (Line 303)

`init` (Line 321)

`create_agent` (Line 324)

`run` (Line 331)