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

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

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

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