Building Reliable AI Pipelines: Lessons from 10M Daily Requests

interface CircuitBreakerConfig {
  failureThreshold: number;    // Failures before opening
  successThreshold: number;    // Successes to close from half-open
  timeout: number;             // Time before trying half-open
  monitorWindow: number;       // Window for counting failures
}

class CircuitBreaker {
  private state: 'closed' | 'open' | 'half-open' = 'closed';
  private failures: number[] = [];
  private successes: number = 0;
  private lastFailure: number = 0;
  
  constructor(
    private name: string,
    private config: CircuitBreakerConfig
  ) {}
  
  async execute<T>(fn: () => Promise<T>): Promise<T> {
    // Check if we should try half-open
    if (this.state === 'open') {
      if (Date.now() - this.lastFailure > this.config.timeout) {
        this.state = 'half-open';
        this.successes = 0;
        metrics.circuitStateChange(this.name, 'half-open');
      } else {
        throw new CircuitOpenError(this.name);
      }
    }
    
    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }
  
  private onSuccess(): void {
    if (this.state === 'half-open') {
      this.successes++;
      if (this.successes >= this.config.successThreshold) {
        this.state = 'closed';
        this.failures = [];
        metrics.circuitStateChange(this.name, 'closed');
      }
    }
  }
  
  private onFailure(): void {
    const now = Date.now();
    this.lastFailure = now;
    
    // Remove old failures outside the monitor window
    this.failures = this.failures.filter(
      t => now - t < this.config.monitorWindow
    );
    this.failures.push(now);
    
    if (this.state === 'half-open') {
      // Single failure in half-open reopens the circuit
      this.state = 'open';
      metrics.circuitStateChange(this.name, 'open');
    } else if (this.failures.length >= this.config.failureThreshold) {
      this.state = 'open';
      metrics.circuitStateChange(this.name, 'open');
    }
  }
  
  getState(): string {
    return this.state;
  }
}

interface FallbackConfig {
  primary: () => Promise<AIResponse>;
  secondary?: () => Promise<AIResponse>;
  cached?: () => Promise<AIResponse | null>;
  default?: () => AIResponse;
}

async function executeWithFallbacks(
  config: FallbackConfig,
  context: RequestContext
): Promise<AIResponse> {
  const attempts: FallbackAttempt[] = [];
  
  // Try primary
  try {
    const result = await config.primary();
    metrics.fallbackUsed(context.requestId, 'primary');
    return result;
  } catch (error) {
    attempts.push({ level: 'primary', error });
    logger.warn('Primary failed, trying secondary', { error, context });
  }
  
  // Try secondary
  if (config.secondary) {
    try {
      const result = await config.secondary();
      metrics.fallbackUsed(context.requestId, 'secondary');
      return { ...result, degraded: true };
    } catch (error) {
      attempts.push({ level: 'secondary', error });
      logger.warn('Secondary failed, trying cache', { error, context });
    }
  }
  
  // Try cache
  if (config.cached) {
    try {
      const result = await config.cached();
      if (result) {
        metrics.fallbackUsed(context.requestId, 'cached');
        return { ...result, degraded: true, fromCache: true };
      }
    } catch (error) {
      attempts.push({ level: 'cached', error });
    }
  }
  
  // Return default
  if (config.default) {
    metrics.fallbackUsed(context.requestId, 'default');
    return { ...config.default(), degraded: true, isDefault: true };
  }
  
  // All fallbacks failed
  throw new AllFallbacksFailedError(attempts);
}

async function summarizeDocument(doc: Document): Promise<Summary> {
  return executeWithFallbacks({
    // Primary: GPT-4 for best quality
    primary: async () => {
      return circuitBreakers.gpt4.execute(() =>
        openai.chat.completions.create({
          model: 'gpt-4-turbo',
          messages: [
            { role: 'system', content: 'Summarize this document concisely.' },
            { role: 'user', content: doc.content }
          ],
          max_tokens: 500
        })
      );
    },
    
    // Secondary: GPT-3.5 (faster, cheaper, slightly lower quality)
    secondary: async () => {
      return circuitBreakers.gpt35.execute(() =>
        openai.chat.completions.create({
          model: 'gpt-3.5-turbo',
          messages: [
            { role: 'system', content: 'Summarize this document concisely.' },
            { role: 'user', content: doc.content }
          ],
          max_tokens: 500
        })
      );
    },
    
    // Cached: Return previous summary if available
    cached: async () => {
      const cached = await cache.getSummary(doc.id);
      if (cached && cached.age < 24 * 60 * 60 * 1000) {
        return cached.summary;
      }
      return null;
    },
    
    // Default: Extractive summary (no AI needed)
    default: () => ({
      text: extractFirstParagraphs(doc.content, 3),
      type: 'extractive',
      confidence: 0.3
    })
  }, { requestId: generateId(), docId: doc.id });
}

interface QueueConfig {
  maxSize: number;
  maxWait: number;
  priorityLevels: number;
}

class PriorityRequestQueue {
  private queues: Map<number, RequestItem[]> = new Map();
  private processing: number = 0;
  private maxConcurrent: number;
  
  constructor(
    private config: QueueConfig,
    maxConcurrent: number
  ) {
    this.maxConcurrent = maxConcurrent;
    for (let i = 0; i < config.priorityLevels; i++) {
      this.queues.set(i, []);
    }
  }
  
  async enqueue<T>(
    priority: number,
    fn: () => Promise<T>,
    timeout?: number
  ): Promise<T> {
    const totalSize = this.getTotalSize();
    
    // Backpressure: reject if queue is full
    if (totalSize >= this.config.maxSize) {
      metrics.queueRejection(priority);
      throw new QueueFullError(totalSize, this.config.maxSize);
    }
    
    return new Promise((resolve, reject) => {
      const item: RequestItem = {
        fn,
        resolve,
        reject,
        priority,
        enqueuedAt: Date.now(),
        timeout: timeout || this.config.maxWait
      };
      
      this.queues.get(priority)!.push(item);
      metrics.queueSize(this.getTotalSize());
      
      this.processNext();
    });
  }
  
  private async processNext(): Promise<void> {
    if (this.processing >= this.maxConcurrent) {
      return;
    }
    
    const item = this.getHighestPriorityItem();
    if (!item) {
      return;
    }
    
    // Check if item has timed out while waiting
    const waitTime = Date.now() - item.enqueuedAt;
    if (waitTime > item.timeout) {
      metrics.queueTimeout(item.priority, waitTime);
      item.reject(new QueueTimeoutError(waitTime));
      this.processNext();
      return;
    }
    
    this.processing++;
    metrics.queueProcessing(this.processing);
    
    try {
      const result = await item.fn();
      item.resolve(result);
    } catch (error) {
      item.reject(error);
    } finally {
      this.processing--;
      metrics.queueProcessing(this.processing);
      this.processNext();
    }
  }
  
  private getHighestPriorityItem(): RequestItem | null {
    for (let i = 0; i < this.config.priorityLevels; i++) {
      const queue = this.queues.get(i)!;
      if (queue.length > 0) {
        return queue.shift()!;
      }
    }
    return null;
  }
  
  private getTotalSize(): number {
    let total = 0;
    this.queues.forEach(q => total += q.length);
    return total;
  }
}

const aiMetrics = {
  // Latency
  requestDuration: new Histogram({
    name: 'ai_request_duration_seconds',
    help: 'AI request duration',
    labelNames: ['model', 'operation', 'status'],
    buckets: [0.1, 0.5, 1, 2, 5, 10, 30]
  }),
  
  // Tokens
  tokensUsed: new Counter({
    name: 'ai_tokens_total',
    help: 'Total tokens used',
    labelNames: ['model', 'type'] // type: input/output
  }),
  
  // Errors
  errors: new Counter({
    name: 'ai_errors_total',
    help: 'AI errors by type',
    labelNames: ['model', 'error_type', 'retryable']
  }),
  
  // Circuit breaker
  circuitState: new Gauge({
    name: 'ai_circuit_state',
    help: 'Circuit breaker state (0=closed, 1=half-open, 2=open)',
    labelNames: ['service']
  }),
  
  // Fallbacks
  fallbackUsage: new Counter({
    name: 'ai_fallback_total',
    help: 'Fallback usage by level',
    labelNames: ['operation', 'level']
  }),
  
  // Queue
  queueSize: new Gauge({
    name: 'ai_queue_size',
    help: 'Current queue size',
    labelNames: ['priority']
  }),
  
  // Cost
  estimatedCost: new Counter({
    name: 'ai_estimated_cost_dollars',
    help: 'Estimated cost in dollars',
    labelNames: ['model', 'operation']
  })
};

# High error rate
- alert: AIHighErrorRate
  expr: rate(ai_errors_total[5m]) / rate(ai_request_duration_seconds_count[5m]) > 0.05
  for: 2m
  labels:
    severity: warning
  annotations:
    summary: "AI error rate above 5%"

# Circuit breaker opened
- alert: AICircuitOpen
  expr: ai_circuit_state > 1
  for: 1m
  labels:
    severity: critical
  annotations:
    summary: "Circuit breaker opened for {{ $labels.service }}"

# Queue backing up
- alert: AIQueueBacklog
  expr: ai_queue_size > 1000
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "AI request queue has {{ $value }} items"

# Cost spike
- alert: AICostSpike
  expr: rate(ai_estimated_cost_dollars[1h]) > 100
  for: 15m
  labels:
    severity: warning
  annotations:
    summary: "AI costs exceeding $100/hour"

async function processAIRequest(
  request: AIRequest
): Promise<AIResponse> {
  const context = createRequestContext(request);
  const span = tracer.startSpan('ai.process', { attributes: context });
  
  try {
    // 1. Enqueue with priority
    const priority = getPriority(request);
    
    return await requestQueue.enqueue(priority, async () => {
      // 2. Execute with fallbacks
      return await executeWithFallbacks({
        primary: () => circuitBreakers.gpt4.execute(() => 
          callGPT4(request)
        ),
        secondary: () => circuitBreakers.gpt35.execute(() => 
          callGPT35(request)
        ),
        cached: () => getFromCache(request),
        default: () => getDefaultResponse(request)
      }, context);
    }, getTimeout(priority));
    
  } catch (error) {
    // 3. Handle errors appropriately
    if (error instanceof CircuitOpenError) {
      metrics.errors.inc({ error_type: 'circuit_open', retryable: 'true' });
      throw new ServiceUnavailableError('AI service temporarily unavailable');
    }
    
    if (error instanceof QueueFullError) {
      metrics.errors.inc({ error_type: 'queue_full', retryable: 'true' });
      throw new TooManyRequestsError('System at capacity, please retry');
    }
    
    if (error instanceof QueueTimeoutError) {
      metrics.errors.inc({ error_type: 'timeout', retryable: 'true' });
      throw new TimeoutError('Request timed out in queue');
    }
    
    throw error;
  } finally {
    span.end();
  }
}