
    """
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

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

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