Building an AI-Powered Local Newsroom

                    ┌──────────────────────────────────────────────────────┐
                    │                  DATA SOURCES                         │
                    │  75+ News Outlets    Gmail Inbox    Reddit/Web        │
                    └────────────┬─────────────┬────────────┬──────────────┘
                                 │             │            │
                    ┌────────────▼─────────────▼────────────▼──────────────┐
                    │              INGESTION LAYER                          │
                    │  Scrapy Spider    Gmail API    Web Scraper            │
                    │  URL Cache        Deduplication  Rate Limiting        │
                    └────────────┬─────────────┬────────────┬──────────────┘
                                 │             │            │
                    ┌────────────▼─────────────▼────────────▼──────────────┐
                    │           PROCESSING LAYER                            │
                    │  Article Extraction    Email Classification           │
                    │  Entity Extraction     Participant Intelligence       │
                    │  Chunking              Embedding Generation           │
                    └────────────┬─────────────┬────────────┬──────────────┘
                                 │             │            │
                    ┌────────────▼─────────────▼────────────▼──────────────┐
                    │              DATA WAREHOUSE                           │
                    │         PostgreSQL 17 + pgvector                      │
                    │  1.79M Articles   218K Emails   8M+ Chunks            │
                    │  453K Participants   10K Sender Profiles              │
                    │            HNSW Vector Indexes                        │
                    └────────────┬─────────────┬────────────┬──────────────┘
                                 │             │            │
          ┌──────────────────────┼─────────────┼────────────┼──────────────────────┐
          │                      │             │            │                      │
          ▼                      ▼             ▼            ▼                      ▼
   ┌─────────────┐     ┌─────────────┐   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
   │   RAG       │     │ NYCNews     │   │   Email     │   │  Memory     │   │  Publish    │
   │  Chatbot    │     │  Scanner    │   │  Assistant  │   │  System     │   │  Pipeline   │
   │             │     │             │   │             │   │             │   │             │
   │ Semantic    │     │ Multi-Agent │   │ Triage      │   │ Adaptive    │   │ WordPress   │
   │ Search      │     │ Research    │   │ Drafting    │   │ Learning    │   │ Integration │
   │ Q&A         │     │ Story       │   │ Task        │   │ Feedback    │   │ SEO Gen     │
   │             │     │ Extraction  │   │ Orchestr.   │   │ Loop        │   │ Image AI    │
   └─────────────┘     └─────────────┘   └─────────────┘   └─────────────┘   └─────────────┘
          │                      │             │            │                      │
          └──────────────────────┼─────────────┼────────────┼──────────────────────┘
                                 │             │            │
                    ┌────────────▼─────────────▼────────────▼──────────────┐
                    │               HUMAN INTERFACE                         │
                    │  CLI Commands    Web Dashboard    API Endpoints       │
                    │              Human-in-the-Loop Review                 │
                    └──────────────────────────────────────────────────────┘

-- Article Storage
outlets (75 active)
  └── articles (1.79M records)
        └── article_content (full HTML/text with tsvector search)
              └── article_chunks (4.56M chunks with 384-dim embeddings)

-- Email Storage
classified_emails (218K records, 60 columns)
  ├── email_chunks (3.58M chunks with embeddings)
  ├── email_participants (453K relationship records)
  │     └── sender_profiles (10K with communication patterns)
  ├── email_attachments (Google Drive links, SHA-256 dedup)
  └── email_events (ICS calendar extraction)

-- Entity Intelligence
entities_v2 (2,580+ entities)
  └── entity_mentions_v2 (cross-corpus tracking)
        └── entity_aliases_v2 (disambiguation)

CREATE INDEX idx_email_chunks_embedding_hnsw
ON email_chunks USING hnsw (embedding vector_cosine_ops);

-- Index size: 5.8 GB for 3.58M email chunks
-- Query time: ~27ms average (verified December 2024)

def triage_page(self, response):
    """Multi-stage article detection."""

    # Stage 1: OpenGraph metadata
    if response.css('meta[property="og:type"][content="article"]'):
        yield from self.parse_article(response)
        return

    # Stage 2: Semantic container analysis
    for selector in ['article', 'main', 'div[role="main"]']:
        container = response.css(selector)
        if container:
            text_content = "".join(container.css("::text").getall())
            html_content = container.get()

            # Text-to-markup ratio filter
            if len(text_content) / len(html_content) > 0.6:
                if len(text_content) > 250:
                    yield from self.parse_article(response)
                    return

class UrlCacheMiddleware:
    def spider_opened(self, spider):
        with engine.connect() as conn:
            urls = conn.execute(
                text("SELECT url FROM articles WHERE outlet_id = :id"),
                {"id": outlet_id}
            ).fetchall()
            self.scraped_urls = {self.normalize_url(row[0]) for row in urls}

    def process_request(self, request, spider):
        if self.normalize_url(request.url) in self.scraped_urls:
            raise IgnoreRequest(f"Already scraped: {request.url}")
        return None

hellgatenyc.com:
  article_url_pattern: '/[a-z0-9-]+/$'
  body_fallback_chain:
    - "article.post-content"
    - "div.entry-content"
    - "div.article-body"

Step 1:   Gmail API extraction with service account delegation
Step 1.1: URL extraction from email content
Step 1.25: Deterministic triage (email/notify/no/spam)
Step 1.5: Attachment extraction → Google Drive with SHA-256 dedup
Step 2:   Semantic chunking with 500-char segments
Step 3:   Gemini Flash classification (16 categories)
Step 3.5: Structured event extraction (ICS parsing)
Step 4:   Participant intelligence (role extraction, fingerprinting)
Step 5:   Story extraction for journalism workflow
Step 6:   LangChain agent processing
Step 7:   Entity extraction (SpaCy NER)

class DeterministicFilter:
    # Spam TLDs with >95% spam rate
    SPAM_TLDS = {'.xyz', '.top', '.gdn', '.click', '.loan'}

    # Known newsletter domains
    NEWSLETTER_DOMAINS = {'mailchimp.com', 'substack.com', 'constantcontact.com'}

    # Financial transaction patterns
    FINANCIAL_PATTERNS = [r'paypal.*receipt', r'venmo.*paid']

    def classify(self, email) -> Optional[str]:
        # Check spam indicators
        if any(email.sender.endswith(tld) for tld in self.SPAM_TLDS):
            return 'spam'

        # Check newsletter patterns
        sender_domain = email.sender.split('@')[1]
        if sender_domain in self.NEWSLETTER_DOMAINS:
            return 'marketing_newsletter'

        # ... additional rules
        return None  # Fall through to LLM classification

def generate_fingerprint(self, email):
    # Normalize content
    content = self.normalize_tracking_params(email.body_text)
    content = self.normalize_whitespace(content)

    # Generate hash
    return hashlib.sha256(content.encode('utf-8')).hexdigest()

ALTER TABLE article_chunks
ADD CONSTRAINT fk_article
FOREIGN KEY (article_id) REFERENCES articles(id) ON DELETE CASCADE;

clean_text = text.encode('utf-8', 'ignore').decode('utf-8')
clean_text = clean_text.replace('\x00', '')  # Remove null bytes

                         ┌─────────────────────┐
                         │  News Coordinator   │
                         │  (Orchestrator)     │
                         └──────────┬──────────┘
            ┌───────────────────────┼───────────────────────┐
            ▼                       ▼                       ▼
 ┌──────────────────┐    ┌──────────────────┐    ┌──────────────────┐
 │  DB Researcher   │    │  Email Researcher │    │  Web Researcher  │
 │  (7 outlets)     │    │  (Press/News)     │    │  (Reddit/Web)    │
 └────────┬─────────┘    └────────┬─────────┘    └────────┬─────────┘
          │                       │                       │
          └───────────────────────┴───────────────────────┘
                                  ▼
                       ┌──────────────────────┐
                       │  Story Extractor     │
                       │  (5W Analysis)       │
                       └──────────┬───────────┘
                                  ▼
                       ┌──────────────────────┐
                       │  Memory Integration  │←─── Quality Gate
                       │  (4-Level Scope)     │←─── Feedback Parser
                       └──────────┬───────────┘
                                  ▼
                       ┌──────────────────────┐
                       │  Pitch Generation    │
                       │  + Human Review      │
                       └──────────────────────┘

def calculate_newsworthiness(self, story, entities, email_context):
    score = 0.0

    # Geographic proximity to Bushwick
    bushwick_entities = ['Bushwick', 'Community Board 4', 'BK90']
    for entity in entities:
        if entity in bushwick_entities:
            score += 0.3

    # Known community figures
    if self.entity_db.is_known_figure(story.who):
        score += 0.2

    # Temporal urgency
    if story.when and story.when < datetime.now() + timedelta(days=7):
        score += 0.2

    # Source credibility
    if email_context.sender_profile.is_government:
        score += 0.15
    elif email_context.sender_profile.is_journalist:
        score += 0.1

    return min(score, 1.0)

CREATE TABLE am_memories (
    id SERIAL PRIMARY KEY,
    scope_type VARCHAR CHECK (scope_type IN ('global', 'org', 'project', 'sender')),
    scope_id VARCHAR,  -- NULL for global, email for sender, org_id for org
    content TEXT NOT NULL,
    embedding VECTOR(384),
    memory_type VARCHAR,  -- 'preference', 'fact', 'procedure', 'relationship'
    confidence FLOAT DEFAULT 1.0,
    times_applied INTEGER DEFAULT 0,
    times_led_to_acceptance INTEGER DEFAULT 0,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    last_accessed TIMESTAMPTZ
);

CREATE FUNCTION search_memories(query_text TEXT, query_embedding VECTOR, scope_filter JSONB)
RETURNS TABLE (memory_id INT, content TEXT, score FLOAT) AS $$
    -- Vector similarity component
    WITH vector_results AS (
        SELECT id, content, 1 - (embedding <=> query_embedding) as vscore,
               ROW_NUMBER() OVER (ORDER BY embedding <=> query_embedding) as vrank
        FROM am_memories
        WHERE scope_type = ANY(scope_filter->'scopes')
    ),
    -- Full-text component
    text_results AS (
        SELECT id, content, ts_rank(search_vector, plainto_tsquery(query_text)) as tscore,
               ROW_NUMBER() OVER (ORDER BY ts_rank DESC) as trank
        FROM am_memories
        WHERE search_vector @@ plainto_tsquery(query_text)
    )
    -- RRF combination
    SELECT COALESCE(v.id, t.id), COALESCE(v.content, t.content),
           (1.0/(60 + COALESCE(vrank, 1000)) + 1.0/(60 + COALESCE(trank, 1000))) as rrf_score
    FROM vector_results v FULL OUTER JOIN text_results t ON v.id = t.id
    ORDER BY rrf_score DESC;
$$ LANGUAGE sql;

class QualityGate:
    def should_process(self, signal: FeedbackSignal) -> bool:
        # Minimum edit magnitude
        if signal.edit_ratio < 0.10:
            return False

        # Confidence threshold
        if signal.confidence < 0.60:
            return False

        # Debounce rapid edits
        if signal.seconds_since_last < 300:
            return False

        return True

def retrieve_memories(self, context):
    base_memories = self.search_memories(context.query)

    # Apply scope boosting
    for memory in base_memories:
        if memory.scope_type == 'sender' and memory.scope_id == context.sender:
            memory.score *= 1.3  # Sender memories most relevant
        elif memory.scope_type == 'org' and memory.scope_id == context.org:
            memory.score *= 1.2
        elif memory.scope_type == 'project' and memory.scope_id in context.projects:
            memory.score *= 1.1
        # Global memories use base score

    return sorted(base_memories, key=lambda m: m.score, reverse=True)[:10]

def compute_effectiveness(self, memory):
    if memory.times_applied == 0:
        return 0.5  # Neutral default

    acceptance_rate = memory.times_led_to_acceptance / memory.times_applied
    edit_ratio = memory.total_edit_distance / memory.total_output_length

    # Weighted combination
    return 0.6 * acceptance_rate + 0.4 * (1 - edit_ratio)

# DON'T: Send raw edits directly to LLM for reflection
def learn_from_edit(original, edited):
    prompt = f"What preference does this edit reveal?\nOriginal: {original}\nEdited: {edited}"
    return llm.complete(prompt)  # Degrades over time

# DO: Deterministic parsing, quality gate, then targeted LLM extraction
def learn_from_edit(original, edited):
    # Step 1: Deterministic classification
    diff = difflib.SequenceMatcher(a=original, b=edited)
    edit_type = classify_edit_type(diff)  # tone, factual, intent, etc.

    # Step 2: Quality gate
    if not quality_gate.should_process(edit_type, diff.ratio()):
        return None

    # Step 3: Targeted LLM extraction (only for qualified signals)
    return extract_learnable_preference(original, edited, edit_type)

@dataclass
class EditorPitch:
    pitch_id: str
    headline: str
    key_facts: List[str]
    source_attribution: str
    source_email_id: int
    newsworthiness_score: float
    suggested_angle: str
    related_articles: List[int]  # IDs of existing coverage

┌─────────────────────────────────────────────────────────────┐
│ PITCH: City Announces New Affordable Housing Lottery       │
├─────────────────────────────────────────────────────────────┤
│ Source: HPD Press Release (maria.torres@nyc.gov)           │
│ Score: 0.87 (HIGH)                                          │
│                                                             │
│ KEY FACTS:                                                  │
│ • 150 units at 1234 Bushwick Ave                           │
│ • Income bands: 40%, 60%, 80% AMI                          │
│ • Applications open March 20                                │
│                                                             │
│ RELATED COVERAGE: 3 prior articles on this development     │
│                                                             │
│ [a]pprove  [r]eject  [e]dit  [n]ext  [q]uit                │
└─────────────────────────────────────────────────────────────┘

Gmail API → Triage → HITL Review → Orchestrator → Skills → Draft Review → Send
                ↓
        Deterministic    Pre-orchestrator     Task           Skill         Human
        Filter           checkpoint           Extraction     Dispatch      Approval

$ ea hitl review --limit 50

Thread: Re: Interview Request - Community Garden
From: reporter@brooklyneagle.com
Triage: EMAIL (needs response)
Preview: Hi Alec, I'm working on a story about...

Actions: [x]spam [m]mute [o]rchestrator [n]ext [q]uit

profile = sender_profiles.get(sender_email)
# Returns: organization, role, is_journalist, is_government,
#          communication_history, preferred_tone, last_interaction

thread = email_threads.get_full_thread(thread_id)
# Returns: all messages in thread, participants, subject evolution

memories = memory_store.retrieve_for_context(
    sender=sender_email,
    org=profile.organization,
    projects=active_projects,
    query=email_subject
)
# Returns: ranked preferences, facts, procedures relevant to this email

examples = few_shot_retriever.get_similar_responses(
    email_type=classification,
    sender_type=profile.classification,
    limit=3
)
# Returns: successful past responses to similar emails

prompt = f"""
You are drafting a response for Alec Meeker, publisher of Bushwick Daily.

SENDER CONTEXT:
{sender_profile_summary}

THREAD CONTEXT:
{thread_history}

RELEVANT MEMORIES:
{format_memories(memories)}

SIMILAR PAST RESPONSES:
{format_examples(examples)}

WRITING GUIDELINES:
- Warm but direct tone
- Business development orientation
- NO em-dashes, NO "I hope this finds you well"
- Be specific about next steps

EMAIL TO RESPOND TO:
{current_email}

Draft a response:
"""

┌─────────────────────────────────────────────────────────────┐
│ DRAFT RESPONSE                                              │
├─────────────────────────────────────────────────────────────┤
│ To: maria.torres@nyc.gov                                    │
│ Subject: Re: Press Credential Application                   │
│                                                             │
│ Hi Maria,                                                   │
│                                                             │
│ Thanks for following up on the credential application.      │
│ I've attached our updated circulation numbers and added     │
│ the publication schedule you requested.                     │
│                                                             │
│ Let me know if you need anything else for the review.       │
│                                                             │
│ Best,                                                       │
│ Alec                                                        │
│                                                             │
│ [s]end [e]dit [r]evise [a]nswer [d]iscard [n]ext [q]uit    │
└─────────────────────────────────────────────────────────────┘

Phase 1: Analyze form → Show proposed values → Await approval
Phase 2: Fill form → Capture screenshot → Confirm submission

# All components use consistent email representation
@dataclass
class EmailContext:
    email_id: int
    thread_id: str
    sender_email: str
    sender_profile: Optional[SenderProfile]
    subject: str
    body_text: str
    body_html: str
    received_at: datetime
    classification: str
    participant_roles: Dict[str, str]
    attachments: List[Attachment]

CREATE TABLE content_lineage_events (
    id SERIAL PRIMARY KEY,
    content_type VARCHAR,  -- email, story, pitch, article, queue_item
    content_id INTEGER,
    event_type VARCHAR,    -- extracted, approved, rejected, published
    actor_type VARCHAR,    -- system, human
    actor_id VARCHAR,
    parent_content_type VARCHAR,
    parent_content_id INTEGER,
    metadata JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

class SessionStore:
    def save_session(self, session_id, state):
        # Write-through: every update persists immediately
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO nycnews_agent_sessions (session_id, state, updated_at)
                VALUES (%s, %s, NOW())
                ON CONFLICT (session_id) DO UPDATE SET state = %s, updated_at = NOW()
            """, (session_id, json.dumps(state), json.dumps(state)))
            self.conn.commit()

    def load_session(self, session_id):
        # Read-through: database fallback when not in memory
        with self.conn.cursor() as cur:
            cur.execute(
                "SELECT state FROM nycnews_agent_sessions WHERE session_id = %s",
                (session_id,)
            )
            result = cur.fetchone()
            return json.loads(result[0]) if result else None

-- Draft knows its generating task
ALTER TABLE ea_email_drafts ADD COLUMN source_task_id INTEGER REFERENCES ea_tasks(id);

-- Task knows its output draft
ALTER TABLE ea_tasks ADD COLUMN generated_draft_id INTEGER REFERENCES ea_email_drafts(id);

def sync_task_draft_status(task_id, draft_id, action):
    """Atomic status update for linked task and draft."""
    with conn.cursor() as cur:
        if action == 'send':
            cur.execute("""
                UPDATE ea_tasks SET status = 'completed', completed_at = NOW()
                WHERE id = %s;
                UPDATE ea_email_drafts SET status = 'sent', sent_at = NOW()
                WHERE id = %s;
            """, (task_id, draft_id))
        elif action == 'reject':
            cur.execute("""
                UPDATE ea_tasks SET status = 'cancelled'
                WHERE id = %s;
                UPDATE ea_email_drafts SET status = 'rejected'
                WHERE id = %s;
            """, (task_id, draft_id))
        conn.commit()

def process_email_with_fallbacks(email):
    try:
        # Primary path: full context enrichment
        profile = get_sender_profile(email.sender)
        memories = get_relevant_memories(email, profile)
        examples = get_few_shot_examples(email)
    except ProfileServiceError:
        # Fallback: basic context only
        profile = None
        memories = []
        examples = get_generic_examples(email.classification)

    # Always proceed with available context
    return generate_draft(email, profile, memories, examples)

@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def generate_embeddings(self, article_id):
    try:
        # Embedding generation
        chunks = chunk_article(article_id)
        embeddings = model.encode(chunks)
        store_embeddings(article_id, embeddings)
    except TransientError as e:
        # Retry for transient failures
        raise self.retry(exc=e, countdown=60 * (self.request.retries + 1))
    except PermanentError as e:
        # Log and skip for permanent failures
        log_problematic_article(article_id, str(e))
        return False

def recover_missing_embeddings():
    """Find and regenerate embeddings that failed to persist."""
    with conn.cursor() as cur:
        cur.execute("""
            SELECT ce.id FROM classified_emails ce
            LEFT JOIN email_chunks ec ON ce.id = ec.email_id
            WHERE ec.email_id IS NULL
            AND ce.processed_at IS NOT NULL
        """)
        missing = cur.fetchall()

    for email_id in missing:
        regenerate_embeddings.delay(email_id)

Master-Scrape-Rag-Pipe/
├── news_scraper_project/           # Article ingestion
│   ├── spiders/article_spider.py   # Intelligent heuristic spider
│   ├── pipelines.py                # PostgreSQL atomic writes
│   ├── tasks.py                    # Celery embedding tasks
│   └── middlewares.py              # URL cache middleware
│
├── email_assistant/                # Email processing CLI
│   ├── cli.py                      # Main entry point
│   ├── orchestration/              # Task extraction & routing
│   │   ├── orchestrator.py         # Central dispatcher
│   │   ├── task_extractor.py       # LLM-based extraction
│   │   └── task_classifier.py      # Capability-aware matching
│   ├── drafting/                   # Response generation
│   │   ├── draft_response.py       # Context-enriched drafting
│   │   └── context_enrichment.py   # RAG context injection
│   ├── memory/                     # Adaptive learning
│   │   ├── store.py                # Memory persistence
│   │   ├── feedback_parser.py      # Deterministic parsing
│   │   └── quality_gate.py         # Signal validation
│   ├── skills/                     # Automation capabilities
│   │   ├── rsvp_skill.py           # Form filling
│   │   ├── calendar_skill.py       # Event management
│   │   └── smart_form_skill.py     # Intelligent form analysis
│   └── rsvp/                       # Browser automation
│       └── playwright_controller.py
│
├── rag_chatbot/                    # RAG interface
│   ├── app.py                      # FastAPI application
│   ├── ai_generator.py             # Claude API integration
│   ├── vector_store_pg.py          # pgvector search
│   └── NYCNewsAgent/               # Research agent system
│       ├── research_agent/         # Multi-agent research
│       │   ├── agent.py            # Claude Agent SDK
│       │   └── email_story_extractor.py
│       ├── publish_pipeline/       # WordPress integration
│       │   ├── metadata_extractor.py
│       │   ├── seo_generator.py
│       │   └── image_analyzer.py
│       └── api/                    # Workflow APIs
│           ├── workflow_api.py
│           └── lineage/            # Content traceability
│
├── entity_extraction/              # Entity intelligence
│   ├── spacy_entity_pipeline.py    # NER pipeline
│   └── entity_disambiguation.py    # Alias management
│
├── attachment_extraction/          # Email attachments
│   ├── drive_uploader.py           # Google Drive streaming
│   └── calendar_parser.py          # ICS parsing
│
├── event_extraction/               # Calendar events
│   └── extract_events_from_emails.py
│
├── migrations/                     # Database schemas
│   └── 001-068_*.sql
│
├── update_emails_v2.sh             # Email pipeline orchestrator
├── run_crawls.py                   # Article scraping orchestrator
└── CLAUDE.md                       # AI assistant instructions