Building an AI News Summarizer: Production-Grade News Aggregation

In this post, I'll walk through building a production-ready news aggregation and summarization system that fetches articles from multiple sources and generates intelligent summaries using LLMs.

🎯 Project Overview

The AI News Summarizer is a FastAPI-based application that:

Fetches news from multiple sources (NewsAPI, RSS feeds)
Generates concise, informative summaries using GPT-4
Analyzes sentiment and categorizes articles
Provides a REST API for accessing summaries
Runs background tasks for automatic updates

Live Demo: Try it here | GitHub: View Source

🏗️ Architecture

┌─────────────┐
│ News Sources│ (NewsAPI, RSS, Web)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  Fetcher    │ (Async, Rate-limited)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│ LLM Pipeline│ (LangChain + GPT-4)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│   Storage   │ (SQLite + Redis)
└──────┬──────┘
       │
       ▼
┌─────────────┐
│  REST API   │ (FastAPI)
└─────────────┘

📝 Implementation Deep Dive

1. News Fetching Service

The fetcher aggregates articles from multiple sources asynchronously:

class NewsFetcher:
    """Fetches news articles from multiple sources."""
    
    def __init__(self):
        self.newsapi = NewsApiClient(api_key=settings.NEWS_API_KEY)
        self.categories = settings.NEWS_CATEGORIES.split(',')
    
    async def fetch_all(self) -over List[Dict]:
        """Fetch articles from all sources."""
        tasks = [
            self.fetch_newsapi(),
            self.fetch_rss_feeds(),
            # Can add more sources
        ]
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        articles = []
        
        for result in results:
            if isinstance(result, list):
                articles.extend(result)
        
        return articles

Key Features:

✅ Async fetching for performance
✅ Multiple source support
✅ Error handling per source
✅ Rate limiting to respect API quotas

2. LLM-Powered Summarization

Using LangChain for structured summarization:

class NewsSummarizer:
    """Summarizes news articles using LLMs."""
    
    def __init__(self):
        self.llm = ChatOpenAI(
            model=settings.LLM_MODEL,
            temperature=settings.TEMPERATURE
        )
        
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """You are an expert news analyst. 
            Create concise, informative summaries. Follow these guidelines:
            1. Capture main points and key facts
            2. Maintain objectivity
            3. Keep it 2-3 sentences
            4. Focus on what, who, when, where, why"""),
            ("user", "Article Title: {title}\n\nContent: {content}\n\nSummary:")
        ])
        
        self.chain = self.prompt | self.llm
    
    async def summarize(self, title: str, content: str) -over Dict:
        """Generate summary for an article."""
        response = await self.chain.ainvoke({
            "title": title,
            "content": content[:4000]  # Limit length
        })
        
        summary = response.content.strip()
        sentiment = self._analyze_sentiment(summary)
        
        return {
            "summary": summary,
            "sentiment": sentiment
        }

Why This Works:

✅ Clear system instructions for consistency
✅ Structured prompts with context
✅ Temperature tuning (0.3) for factual output
✅ Token limit management

3. Sentiment Analysis

Simple but effective sentiment scoring:

def _analyze_sentiment(self, text: str) -over str:
    """Analyze sentiment using TextBlob."""
    blob = TextBlob(text)
    polarity = blob.sentiment.polarity
    
    if polarity over 0.1:
        return "positive"
    elif polarity less than -0.1:
        return "negative"
    else:
        return "neutral"

4. FastAPI Backend

Clean REST API with proper error handling:

@app.get("/api/summaries", response_model=List[SummaryResponse])
async def get_summaries(
    category: Optional[str] = None,
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0)
):
    """Get article summaries with optional filtering."""
    try:
        db = next(get_db())
        query = db.query(ArticleSummary)
        
        if category:
            query = query.filter(ArticleSummary.category == category)
        
        summaries = query.order_by(
            ArticleSummary.published_at.desc()
        ).limit(limit).offset(offset).all()
        
        return [SummaryResponse(**s.__dict__) for s in summaries]
    except Exception as e:
        logger.error(f"Error fetching summaries: {e}")
        raise HTTPException(status_code=500, detail=str(e))

API Features:

✅ Pagination support
✅ Category filtering
✅ Proper error responses
✅ Type validation with Pydantic

5. Background Processing

Automatic news refresh using FastAPI lifespan:

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan management."""
    logger.info("Starting AI News Summarizer...")
    
    # Initialize database
    init_db()
    
    # Start background tasks
    asyncio.create_task(periodic_news_fetch())
    
    yield
    
    logger.info("Shutting down...")

async def periodic_news_fetch():
    """Background task to fetch news every hour."""
    while True:
        try:
            await fetch_and_summarize_news()
        except Exception as e:
            logger.error(f"Error in periodic fetch: {e}")
        
        await asyncio.sleep(settings.REFRESH_INTERVAL_MINUTES * 60)

🗄️ Database Design

Simple but effective SQLAlchemy model:

class ArticleSummary(Base):
    """Model for storing article summaries."""
    
    __tablename__ = "article_summaries"
    
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(500), nullable=False)
    source = Column(String(100), nullable=False)
    url = Column(String(1000), unique=True, index=True)
    category = Column(String(50), index=True)
    summary = Column(Text, nullable=False)
    sentiment = Column(String(20))
    published_at = Column(DateTime, index=True)
    created_at = Column(DateTime, default=datetime.utcnow)

Design Decisions:

✅ URL as unique constraint (avoid duplicates)
✅ Indexes on commonly queried fields
✅ Timestamps for sorting and filtering

📊 Performance Metrics

Real-world performance from testing:

Metric	Value
Articles/second	approximately 5
Summary generation	2-3s per article
API response time	under 100ms (cached)
Cache hit rate	\| >gt;80%
Memory usage	approximately 200MB

🚀 Deployment

Docker Setup

FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY src/ ./src/
COPY .env .

CMD ["python", "src/main.py"]

Docker Compose

version: '3.8'
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - NEWS_API_KEY=${NEWS_API_KEY}
    volumes:
      - ./data:/app/data
  
  redis:
    image: redis:alpine
    ports:
      - "6379:6379"

🔧 Challenges & Solutions

Challenge 1: Rate Limiting

Problem: NewsAPI has strict rate limits Solution:

Implemented exponential backoff
Added request queuing
Cached responses for 1 hour

Challenge 2: Duplicate Articles

Problem: Same article from different sources Solution:

URL-based deduplication
Fuzzy title matching
Content hash comparison

Challenge 3: Summarization Quality

Problem: Inconsistent summary quality Solution:

Refined system prompts
Added length constraints
Implemented quality scoring

Challenge 4: Cost Management

Problem: High API costs for large volumes Solution:

Batched processing
Smart caching strategy
Fallback to GPT-3.5 for older articles

💡 Key Learnings

Async is Essential: Synchronous news fetching would be 10x slower
Prompt Engineering Matters: Spent significant time refining prompts
Caching is Critical: Reduced API costs by 80% with Redis
Error Handling: Individual source failures shouldn't crash the system
Monitoring: Added comprehensive logging for debugging

🎯 Future Improvements

Add more news sources (Guardian, BBC, etc.)
Implement semantic deduplication
Add user preferences and filtering
Create React frontend
Add email digest feature
Implement trending topic detection

📦 Tech Stack Summary

Backend: FastAPI, Python 3.9+
LLM: OpenAI GPT-3.5/4, LangChain
Database: SQLite (SQLAlchemy ORM)
Caching: Redis
News API: NewsAPI, feedparser
NLP: TextBlob for sentiment
Deployment: Docker, Docker Compose

🔗 Resources

💬 Questions?

Drop a comment below or reach out on Twitter / LinkedIn!

Next in Series: YouTube Video Summarizer - Learn how to transcribe and summarize videos using Whisper and LangChain.