Discovery → Architecture → Production

langsmith>=0.1.0
anthropic>=0.25.0
psycopg[binary]>=3.1.0
pgvector>=0.2.0

import json
import time
from datetime import datetime
from typing import Any
import sqlite3

class CostTracker:
    """Track cost, latency, and quality metrics for each search."""
    
    def __init__(self, db_path: str = "search_metrics.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS search_metrics (
                    id INTEGER PRIMARY KEY,
                    timestamp TEXT NOT NULL,
                    query TEXT,
                    sql_latency_ms REAL,
                    vector_latency_ms REAL,
                    rerank_latency_ms REAL,
                    total_latency_ms REAL,
                    input_tokens INTEGER,
                    output_tokens INTEGER,
                    estimated_cost_cents REAL,
                    result_count INTEGER,
                    user_feedback_relevant BOOLEAN,
                    user_feedback_rating INTEGER
                )
            """)
    
    def log_search(self, **kwargs):
        """Log a single search operation."""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT INTO search_metrics 
                (timestamp, query, sql_latency_ms, vector_latency_ms, 
                 rerank_latency_ms, total_latency_ms, input_tokens, 
                 output_tokens, estimated_cost_cents, result_count)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                datetime.utcnow().isoformat(),
                kwargs.get('query'),
                kwargs.get('sql_latency_ms'),
                kwargs.get('vector_latency_ms'),
                kwargs.get('rerank_latency_ms'),
                kwargs.get('total_latency_ms'),
                kwargs.get('input_tokens', 0),
                kwargs.get('output_tokens', 0),
                self._estimate_cost(kwargs.get('input_tokens', 0), 
                                    kwargs.get('output_tokens', 0)),
                kwargs.get('result_count', 0)
            ))
    
    @staticmethod
    def _estimate_cost(input_tokens: int, output_tokens: int) -> float:
        """Estimate cost in cents for Claude 3.5 Sonnet."""
        # Input: $3/1M tokens  → 0.0003 cents per token
        # Output: $15/1M tokens → 0.0015 cents per token
        input_cost  = input_tokens  * 0.0003
        output_cost = output_tokens * 0.0015
        return input_cost + output_cost
    
    def get_daily_summary(self, days: int = 7) -> dict:
        """Get cost, latency, and quality metrics for the last N days."""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute("""
                SELECT 
                    DATE(timestamp) as date,
                    COUNT(*) as search_count,
                    ROUND(AVG(total_latency_ms), 1) as avg_latency_ms,
                    ROUND(MAX(total_latency_ms), 1) as p99_latency_ms,
                    ROUND(SUM(estimated_cost_cents) / 100, 2) as cost_usd,
                    ROUND(AVG(CAST(user_feedback_relevant AS FLOAT)), 2) as quality_relevant_ratio
                FROM search_metrics
                WHERE timestamp > datetime('now', '-' || ? || ' days')
                GROUP BY DATE(timestamp)
                ORDER BY date DESC
            """, (days,))
            return [dict(row) for row in rows]

from anthropic import Anthropic
from observability import CostTracker
import time
import json

class HybridSearchService:
    def __init__(self, db_conn, tracker: CostTracker):
        self.db = db_conn
        self.tracker = tracker
        self.client = Anthropic()
    
    def search(self, query: str, customer_id: str, limit: int = 10) -> list[dict]:
        """Execute hybrid search with full observability."""
        
        total_start = time.time()
        
        # 1. SQL full-text search
        sql_start = time.time()
        sql_results = self._sql_search(query, customer_id, limit * 2)
        sql_latency = (time.time() - sql_start) * 1000
        
        # 2. Vector semantic search
        vector_start = time.time()
        vector_results = self._vector_search(query, customer_id, limit * 2)
        vector_latency = (time.time() - vector_start) * 1000
        
        # 3. Combine + rerank using Claude
        combined = self._deduplicate_results(sql_results, vector_results)
        
        rerank_start = time.time()
        ranked = self._rerank_with_claude(query, combined, limit)
        rerank_latency = (time.time() - rerank_start) * 1000
        
        total_latency = (time.time() - total_start) * 1000
        
        # 4. Log everything
        self.tracker.log_search(
            query=query,
            sql_latency_ms=sql_latency,
            vector_latency_ms=vector_latency,
            rerank_latency_ms=rerank_latency,
            total_latency_ms=total_latency,
            input_tokens=150,   # approximate; replace with response.usage.input_tokens
            output_tokens=50,
            result_count=len(ranked)
        )
        
        return ranked
    
    def _rerank_with_claude(self, query: str, candidates: list, limit: int) -> list:
        """Use Claude to rerank candidates."""
        prompt = f"""Given this support ticket search query:
"{query}"

And these candidate tickets, rank them by relevance (most relevant first):
{json.dumps(candidates, indent=2)}

Return a JSON array of the top {limit} tickets, ordered by relevance."""
        
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1000,
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return candidates[:limit]
    
    def _sql_search(self, query: str, customer_id: str, limit: int) -> list:
        cur = self.db.cursor()
        cur.execute("""
            SELECT id, customer_id, content, created_at
            FROM tickets
            WHERE customer_id = %s
            AND to_tsvector('english', content) @@ plainto_tsquery('english', %s)
            LIMIT %s
        """, (customer_id, query, limit))
        return [dict(row) for row in cur.fetchall()]
    
    def _vector_search(self, query: str, customer_id: str, limit: int) -> list:
        cur = self.db.cursor()
        cur.execute("""
            SELECT id, customer_id, content, created_at,
                   1 - (embedding <=> %s::vector) as similarity
            FROM tickets
            WHERE customer_id = %s
            ORDER BY embedding <=> %s::vector
            LIMIT %s
        """, (query, customer_id, query, limit))
        return [dict(row) for row in cur.fetchall()]
    
    def _deduplicate_results(self, sql_results, vector_results) -> list:
        seen, combined = set(), []
        for r in sql_results + vector_results:
            if r['id'] not in seen:
                seen.add(r['id'])
                combined.append(r)
        return combined

API Dashboard: Support Ticket Search
Last 7 days:

📊 PERFORMANCE
  Avg Latency:     1.65s (target: <2s) ✅
  P99 Latency:     1.98s
  Search Count:    412 queries

💰 COST
  Day 1: $0.34
  Day 2: $0.41
  Day 3: $0.38
  Week Total: $2.65 (annualized: ~$138)

📈 QUALITY
  Relevant (user rated):  78% ✅
  Not Relevant:           14%
  Skipped Rating:          8%

🔴 ALERTS
  None this week!

import pytest
import sqlite3
import time
from unittest.mock import patch, MagicMock
from search_service import HybridSearchService
from observability import CostTracker

class TestHybridSearchEdgeCases:
    
    @pytest.fixture
    def db_conn(self):
        """Create in-memory DB with test data."""
        conn = sqlite3.connect(":memory:")
        conn.row_factory = sqlite3.Row
        conn.execute("""
            CREATE TABLE tickets (
                id INTEGER PRIMARY KEY,
                customer_id TEXT,
                content TEXT,
                created_at TEXT,
                embedding VECTOR(1536)
            )
        """)
        conn.execute("""
            INSERT INTO tickets (customer_id, content, created_at)
            VALUES ('cust_123', 'User cannot login to account', '2024-04-10')
        """)
        conn.commit()
        return conn
    
    def test_empty_result_set_no_tickets_for_customer(self, db_conn):
        """EDGE CASE 1: Customer has no tickets.
           Should return empty list, NOT crash."""
        tracker = CostTracker()
        service = HybridSearchService(db_conn, tracker)
        
        results = service.search(
            query="login issue",
            customer_id="cust_999"   # No tickets for this customer
        )
        
        assert results == []
        assert isinstance(results, list)
    
    def test_timeout_on_rerank_exceeds_latency_sla(self, db_conn):
        """EDGE CASE 2: Rerank step times out (>5s).
           Should degrade gracefully, NOT exceed 2s SLA."""
        tracker = CostTracker()
        service = HybridSearchService(db_conn, tracker)
        
        with patch.object(service.client.messages, 'create') as mock_claude:
            def slow_api_call(*args, **kwargs):
                time.sleep(0.1)
                raise TimeoutError("API took too long")
            mock_claude.side_effect = slow_api_call
            
            start = time.time()
            results = service.search(query="login", customer_id="cust_123")
            elapsed = (time.time() - start) * 1000
            
            assert len(results) > 0    # Still return something
            assert elapsed < 2000      # Stayed under SLA
    
    def test_duplicate_tickets_in_combined_results(self, db_conn):
        """EDGE CASE 3: SQL and vector search both return same ticket.
           Should deduplicate, NOT show ticket twice."""
        tracker = CostTracker()
        service = HybridSearchService(db_conn, tracker)
        
        ticket = {
            'id': 1,
            'customer_id': 'cust_123',
            'content': 'User cannot login',
            'created_at': '2024-04-10'
        }
        
        with patch.object(service, '_sql_search', return_value=[ticket]):
            with patch.object(service, '_vector_search', return_value=[ticket]):
                with patch.object(service, '_rerank_with_claude', return_value=[ticket]):
                    results = service.search(
                        query="login", customer_id="cust_123", limit=10
                    )
                    ids = [r['id'] for r in results]
                    assert len(ids) == len(set(ids)), "Duplicate IDs found"
                    assert ids.count(1) == 1
    
    def test_malformed_rerank_response_json(self, db_conn):
        """EDGE CASE 4 (bonus): Claude returns invalid JSON.
           Should fallback to original results."""
        tracker = CostTracker()
        service = HybridSearchService(db_conn, tracker)
        
        with patch.object(service.client.messages, 'create') as mock_claude:
            mock_response = MagicMock()
            mock_response.content = [MagicMock(text="Not valid JSON {{{")]
            mock_claude.return_value = mock_response
            
            candidates = [
                {'id': 1, 'content': 'ticket 1'},
                {'id': 2, 'content': 'ticket 2'}
            ]
            results = service._rerank_with_claude(
                query="test", candidates=candidates, limit=2
            )
            assert len(results) <= 2
            assert results[0]['id'] in [1, 2]
    
    def test_cost_tracking_under_high_load(self, db_conn):
        """EDGE CASE 5 (bonus): 100 searches in a row.
           Cost tracking should not slow down searches."""
        tracker = CostTracker()
        service = HybridSearchService(db_conn, tracker)
        
        times = []
        for i in range(100):
            start = time.time()
            service.search(query=f"query {i}", customer_id="cust_123")
            times.append((time.time() - start) * 1000)
        
        avg_latency = sum(times) / len(times)
        assert avg_latency < 300, f"Avg latency {avg_latency}ms is too high"

$ pytest tests/test_search_edge_cases.py -v

tests/...::test_empty_result_set_no_tickets_for_customer PASSED
tests/...::test_timeout_on_rerank_exceeds_latency_sla PASSED
tests/...::test_duplicate_tickets_in_combined_results PASSED
tests/...::test_malformed_rerank_response_json PASSED
tests/...::test_cost_tracking_under_high_load PASSED

====== 5 passed in 2.34s ======

# Architecture Decisions

## Decision 1: PostgreSQL + pgvector, NOT Neo4j or Pinecone

### Option A: Neo4j (Knowledge Graph)
Pros: Excellent for "related tickets" queries, strong for cross-customer insights
Cons: Overkill for single-tenant search; requires Cypher (client doesn't know it); 5–10s latency on large datasets
Why Not: "The client's problem is 'find the right past ticket for this one customer, fast.'
Graph queries are powerful, but we'd pay in complexity and latency for features they don't need."

### Option B: Pinecone Vector-Only
Pros: Managed service, fastest single-query latency (~500ms), battle-tested
Cons: $1,500–2,000/month at their scale; proprietary; junior engineer can't debug failures
Why Not: "Cost explodes budget 20x. Also junior engineer can't debug why results are bad —
Pinecone is a black box. We need transparency for maintenance."

### Option C: Elasticsearch
Pros: Open source, excellent hybrid search, widely known
Cons: Requires separate infrastructure; more complex than PostgreSQL; overkill for 50k tickets
Why Not: "PostgreSQL + pgvector gets 95% of value with 50% less complexity.
Client has existing Postgres. No new infra to manage."

### ✅ Final Choice: PostgreSQL + pgvector
1. Data already in Postgres → no migration
2. Full-text search is fast for exact matches
3. Vector search is fast for semantic matches
4. Junior engineer understands SQL
5. Cost = hosting only (~$50/month)
6. Transparent failure modes
7. Upgrade path: Qdrant if they outgrow it (same pgvector code)

---

## Decision 2: Claude 3.5 Sonnet, NOT GPT-4

### Option A: GPT-4
Pros: Slightly better semantic understanding
Cons: 10x cost; for a reranking task the extra quality buys almost nothing; would triple monthly budget
Why Not: "Reranking is binary ('is this relevant?'). GPT-4 and Sonnet are both 99% accurate.
GPT-4 costs 10x more. Spend that on better retrieval instead."

### Option B: Open Source (Llama 2, Mixtral)
Pros: Free (no per-token cost); runs on their server; complete control
Cons: Requires GPU infrastructure ($500–1000/month); slower inference (3–5s); hard to maintain
Why Not: "Infrastructure + maintenance + latency hit cancels the 'free' savings.
Managed API is cheaper and faster."

### ✅ Final Choice: Claude 3.5 Sonnet
1. Cost-optimal for this task (reranking, not generation)
2. Fast (<1s)
3. Client pays per-use (no waste if searches drop)
4. Anthropic API is reliable

---

## Decision 3: SQLite for Metrics, NOT Datadog or CloudWatch

### Option A: Datadog
Pros: Enterprise monitoring, dashboards, alerting
Cons: Minimum $25/month + $0.25/GB; overkill for <100 searches/day
Why Not: "Client doesn't need enterprise monitoring for an MVP. SQLite is free and queryable."

### Option B: CloudWatch (AWS)
Pros: Integrated if on AWS, scales easily
Cons: Client is on Azure — adds vendor complexity
Why Not: "Client's stack is Azure. Keep it simple: one vendor."

### ✅ Final Choice: SQLite + simple Python dashboard
1. Zero cost
2. Client's engineer can query it directly: `SELECT * FROM metrics`
3. Easy to migrate: `INSERT INTO datadog_metrics SELECT * FROM sqlite_metrics`
4. Metrics live with the code (version controlled together)

---

## When to Revisit These Decisions
- If search volume exceeds 1,000 queries/day → Migrate pgvector to Qdrant or Weaviate
- If monthly cost exceeds $150 → Evaluate open-source LLM (self-hosted)
- If latency hits 2.5s → Add caching, prioritize recent tickets, or switch embedding model
- If quality drops below 70% → Collect more training data, or use GPT-4 for rerank