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

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

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

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