How to Build an AI Image Generation API Pipeline: Complete Developer Guide 2026

ai-image-pipeline/
├── src/
│   ├── providers/
│   │   ├── base.ts           # Abstract provider interface
│   │   ├── dalle.ts          # OpenAI DALL-E adapter
│   │   ├── flux.ts           # Black Forest Labs Flux adapter
│   │   ├── stability.ts     # Stability AI (SD3.5) adapter
│   │   └── midjourney.ts    # Midjourney API adapter
│   ├── router/
│   │   ├── strategy.ts      # Routing strategies
│   │   ├── cost-optimizer.ts # Budget-aware routing
│   │   └── fallback.ts      # Fallback chain logic
│   ├── pipeline/
│   │   ├── ingestion.ts     # Request validation
│   │   ├── orchestrator.ts  # Main pipeline controller
│   │   └── postprocess.ts   # Result handling
│   ├── monitoring/
│   │   ├── metrics.ts       # Prometheus metrics
│   │   ├── circuit-breaker.ts
│   │   └── budget.ts        # Spend tracking
│   └── index.ts
├── config/
│   ├── providers.yaml       # Provider configurations
│   └── routing-rules.yaml   # Routing logic config
├── tests/
├── docker-compose.yaml
└── package.json

mkdir ai-image-pipeline && cd ai-image-pipeline
pnpm init
pnpm add openai replicate axios sharp bull ioredis prom-client zod yaml
pnpm add -D typescript @types/node vitest

// src/providers/base.ts
import { z } from 'zod';
 
export const GenerationRequestSchema = z.object({
  prompt: z.string().min(1).max(4000),
  negativePrompt: z.string().optional(),
  width: z.number().int().min(256).max(4096).default(1024),
  height: z.number().int().min(256).max(4096).default(1024),
  style: z.enum(['photorealistic', 'artistic', 'anime', 'illustration', 'abstract']).optional(),
  quality: z.enum(['draft', 'standard', 'high', 'ultra']).default('standard'),
  seed: z.number().int().optional(),
  metadata: z.record(z.string()).optional(),
});
 
export type GenerationRequest = z.infer<typeof GenerationRequestSchema>;
 
export interface GenerationResult {
  imageUrl: string;
  provider: string;
  model: string;
  latencyMs: number;
  cost: number;
  seed?: number;
  metadata?: Record<string, string>;
}
 
export interface ProviderHealth {
  available: boolean;
  latencyP50: number;
  latencyP99: number;
  errorRate: number;
  remainingQuota: number;
}
 
export abstract class BaseProvider {
  abstract readonly name: string;
  abstract readonly models: string[];
  abstract readonly maxConcurrent: number;
  abstract readonly costPerImage: Record<string, number>;
 
  abstract generate(request: GenerationRequest): Promise<GenerationResult>;
  abstract checkHealth(): Promise<ProviderHealth>;
  abstract estimateCost(request: GenerationRequest): number;
  abstract supportsFeature(feature: string): boolean;
}

// src/providers/dalle.ts
import OpenAI from 'openai';
import { BaseProvider, GenerationRequest, GenerationResult, ProviderHealth } from './base';
 
export class DalleProvider extends BaseProvider {
  readonly name = 'dalle';
  readonly models = ['dall-e-3'];
  readonly maxConcurrent = 5;
  readonly costPerImage = {
    '1024x1024': 0.04,
    '1024x1792': 0.08,
    '1792x1024': 0.08,
  };
 
  private client: OpenAI;
  private requestCount = 0;
  private errorCount = 0;
  private latencies: number[] = [];
 
  constructor(apiKey: string) {
    super();
    this.client = new OpenAI({ apiKey });
  }
 
  async generate(request: GenerationRequest): Promise<GenerationResult> {
    const startTime = Date.now();
    const size = this.mapSize(request.width, request.height);
 
    try {
      const response = await this.client.images.generate({
        model: 'dall-e-3',
        prompt: request.prompt,
        n: 1,
        size,
        quality: request.quality === 'ultra' ? 'hd' : 'standard',
        style: request.style === 'photorealistic' ? 'natural' : 'vivid',
      });
 
      const latencyMs = Date.now() - startTime;
      this.recordLatency(latencyMs);
      this.requestCount++;
 
      return {
        imageUrl: response.data[0].url!,
        provider: this.name,
        model: 'dall-e-3',
        latencyMs,
        cost: this.estimateCost(request),
        metadata: { revisedPrompt: response.data[0].revised_prompt || '' },
      };
    } catch (error) {
      this.errorCount++;
      throw error;
    }
  }
 
  async checkHealth(): Promise<ProviderHealth> {
    const sorted = [...this.latencies].sort((a, b) => a - b);
    return {
      available: this.errorCount / Math.max(this.requestCount, 1) < 0.3,
      latencyP50: sorted[Math.floor(sorted.length * 0.5)] || 0,
      latencyP99: sorted[Math.floor(sorted.length * 0.99)] || 0,
      errorRate: this.errorCount / Math.max(this.requestCount, 1),
      remainingQuota: Infinity,
    };
  }
 
  estimateCost(request: GenerationRequest): number {
    const size = this.mapSize(request.width, request.height);
    return this.costPerImage[size] || 0.04;
  }
 
  supportsFeature(feature: string): boolean {
    const supported = ['text-rendering', 'composition', 'prompt-enhancement'];
    return supported.includes(feature);
  }
 
  private mapSize(width: number, height: number): string {
    if (width > height) return '1792x1024';
    if (height > width) return '1024x1792';
    return '1024x1024';
  }
 
  private recordLatency(ms: number): void {
    this.latencies.push(ms);
    if (this.latencies.length > 1000) this.latencies.shift();
  }
}

// src/providers/flux.ts
import Replicate from 'replicate';
import { BaseProvider, GenerationRequest, GenerationResult, ProviderHealth } from './base';
 
export class FluxProvider extends BaseProvider {
  readonly name = 'flux';
  readonly models = ['flux-1.1-pro', 'flux-1.1-pro-ultra'];
  readonly maxConcurrent = 10;
  readonly costPerImage = {
    'flux-1.1-pro': 0.04,
    'flux-1.1-pro-ultra': 0.06,
  };
 
  private client: Replicate;
  private metrics = { requests: 0, errors: 0, latencies: [] as number[] };
 
  constructor(apiToken: string) {
    super();
    this.client = new Replicate({ auth: apiToken });
  }
 
  async generate(request: GenerationRequest): Promise<GenerationResult> {
    const startTime = Date.now();
    const model = request.quality === 'ultra' ? 'flux-1.1-pro-ultra' : 'flux-1.1-pro';
 
    try {
      const output = await this.client.run(
        `black-forest-labs/${model}` as any,
        {
          input: {
            prompt: request.prompt,
            width: this.snapToGrid(request.width),
            height: this.snapToGrid(request.height),
            num_inference_steps: request.quality === 'draft' ? 20 : 50,
            seed: request.seed,
          },
        }
      );
 
      const latencyMs = Date.now() - startTime;
      this.metrics.latencies.push(latencyMs);
      this.metrics.requests++;
 
      return {
        imageUrl: Array.isArray(output) ? output[0] : output as string,
        provider: this.name,
        model,
        latencyMs,
        cost: this.costPerImage[model],
        seed: request.seed,
      };
    } catch (error) {
      this.metrics.errors++;
      throw error;
    }
  }
 
  async checkHealth(): Promise<ProviderHealth> {
    const sorted = [...this.metrics.latencies].sort((a, b) => a - b);
    return {
      available: this.metrics.errors / Math.max(this.metrics.requests, 1) < 0.25,
      latencyP50: sorted[Math.floor(sorted.length * 0.5)] || 0,
      latencyP99: sorted[Math.floor(sorted.length * 0.99)] || 0,
      errorRate: this.metrics.errors / Math.max(this.metrics.requests, 1),
      remainingQuota: Infinity,
    };
  }
 
  estimateCost(request: GenerationRequest): number {
    const model = request.quality === 'ultra' ? 'flux-1.1-pro-ultra' : 'flux-1.1-pro';
    return this.costPerImage[model];
  }
 
  supportsFeature(feature: string): boolean {
    const supported = ['photorealistic', 'high-resolution', 'fast-generation', 'seed-reproducibility'];
    return supported.includes(feature);
  }
 
  private snapToGrid(dimension: number): number {
    return Math.round(dimension / 64) * 64;
  }
}

// src/router/strategy.ts
import { GenerationRequest, BaseProvider, ProviderHealth } from '../providers/base';
 
interface RoutingDecision {
  primary: BaseProvider;
  fallbacks: BaseProvider[];
  reason: string;
}
 
interface RoutingContext {
  request: GenerationRequest;
  providers: Map<string, BaseProvider>;
  healthMap: Map<string, ProviderHealth>;
  budgetRemaining: number;
  prioritizeSpeed: boolean;
}
 
export class RoutingStrategy {
  route(context: RoutingContext): RoutingDecision {
    const { request, providers, healthMap, budgetRemaining } = context;
    const candidates = this.filterAvailable(providers, healthMap);
 
    if (candidates.length === 0) {
      throw new Error('No healthy providers available');
    }
 
    // Feature-based routing
    if (this.needsTextRendering(request)) {
      return this.routeForText(candidates, context);
    }
 
    if (request.style === 'photorealistic') {
      return this.routeForPhotorealism(candidates, context);
    }
 
    if (context.prioritizeSpeed) {
      return this.routeForSpeed(candidates, context);
    }
 
    // Cost-optimized default routing
    return this.routeForCost(candidates, context);
  }
 
  private routeForText(candidates: BaseProvider[], context: RoutingContext): RoutingDecision {
    const textCapable = candidates.filter(p => p.supportsFeature('text-rendering'));
    const primary = textCapable[0] || candidates[0];
    return {
      primary,
      fallbacks: candidates.filter(p => p !== primary),
      reason: 'text-rendering-required',
    };
  }
 
  private routeForPhotorealism(candidates: BaseProvider[], context: RoutingContext): RoutingDecision {
    const photoCapable = candidates.filter(p => p.supportsFeature('photorealistic'));
    const sorted = photoCapable.sort((a, b) => {
      const healthA = context.healthMap.get(a.name);
      const healthB = context.healthMap.get(b.name);
      return (healthA?.latencyP50 || 9999) - (healthB?.latencyP50 || 9999);
    });
    return {
      primary: sorted[0] || candidates[0],
      fallbacks: sorted.slice(1),
      reason: 'photorealism-optimized',
    };
  }
 
  private routeForSpeed(candidates: BaseProvider[], context: RoutingContext): RoutingDecision {
    const sorted = candidates.sort((a, b) => {
      const healthA = context.healthMap.get(a.name);
      const healthB = context.healthMap.get(b.name);
      return (healthA?.latencyP50 || 9999) - (healthB?.latencyP50 || 9999);
    });
    return { primary: sorted[0], fallbacks: sorted.slice(1), reason: 'speed-optimized' };
  }
 
  private routeForCost(candidates: BaseProvider[], context: RoutingContext): RoutingDecision {
    const sorted = candidates.sort((a, b) => {
      return a.estimateCost(context.request) - b.estimateCost(context.request);
    });
    return { primary: sorted[0], fallbacks: sorted.slice(1), reason: 'cost-optimized' };
  }
 
  private filterAvailable(
    providers: Map<string, BaseProvider>,
    healthMap: Map<string, ProviderHealth>
  ): BaseProvider[] {
    return Array.from(providers.values()).filter(p => {
      const health = healthMap.get(p.name);
      return health?.available !== false;
    });
  }
 
  private needsTextRendering(request: GenerationRequest): boolean {
    const textPatterns = /\b(text|word|letter|sign|banner|label|caption|title)\b/i;
    return textPatterns.test(request.prompt);
  }
}

// src/monitoring/circuit-breaker.ts
enum CircuitState {
  CLOSED = 'closed',      // Normal operation
  OPEN = 'open',          // Blocking requests
  HALF_OPEN = 'half-open' // Testing recovery
}
 
export class CircuitBreaker {
  private state: CircuitState = CircuitState.CLOSED;
  private failureCount = 0;
  private lastFailureTime = 0;
  private successCount = 0;
 
  constructor(
    private readonly threshold: number = 5,
    private readonly resetTimeoutMs: number = 30000,
    private readonly halfOpenMax: number = 3
  ) {}
 
  async execute<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === CircuitState.OPEN) {
      if (Date.now() - this.lastFailureTime > this.resetTimeoutMs) {
        this.state = CircuitState.HALF_OPEN;
        this.successCount = 0;
      } else {
        throw new Error('Circuit breaker is OPEN');
      }
    }
 
    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }
 
  private onSuccess(): void {
    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= this.halfOpenMax) {
        this.state = CircuitState.CLOSED;
        this.failureCount = 0;
      }
    } else {
      this.failureCount = 0;
    }
  }
 
  private onFailure(): void {
    this.failureCount++;
    this.lastFailureTime = Date.now();
    if (this.failureCount >= this.threshold) {
      this.state = CircuitState.OPEN;
    }
  }
 
  getState(): CircuitState {
    return this.state;
  }
}

// src/pipeline/orchestrator.ts
import { GenerationRequest, GenerationResult, BaseProvider } from '../providers/base';
import { RoutingStrategy } from '../router/strategy';
import { CircuitBreaker } from '../monitoring/circuit-breaker';
import { BudgetTracker } from '../monitoring/budget';
import { MetricsCollector } from '../monitoring/metrics';
 
export class PipelineOrchestrator {
  private providers: Map<string, BaseProvider> = new Map();
  private breakers: Map<string, CircuitBreaker> = new Map();
  private router: RoutingStrategy;
  private budget: BudgetTracker;
  private metrics: MetricsCollector;
 
  constructor(config: PipelineConfig) {
    this.router = new RoutingStrategy();
    this.budget = new BudgetTracker(config.dailyBudget);
    this.metrics = new MetricsCollector();
  }
 
  registerProvider(provider: BaseProvider): void {
    this.providers.set(provider.name, provider);
    this.breakers.set(provider.name, new CircuitBreaker());
  }
 
  async generate(request: GenerationRequest): Promise<GenerationResult> {
    const startTime = Date.now();
 
    // Validate budget
    if (!this.budget.canSpend(this.estimateMaxCost(request))) {
      throw new Error('Daily budget exceeded');
    }
 
    // Get health status for all providers
    const healthMap = await this.getHealthMap();
 
    // Route the request
    const decision = this.router.route({
      request,
      providers: this.providers,
      healthMap,
      budgetRemaining: this.budget.remaining(),
      prioritizeSpeed: false,
    });
 
    // Execute with fallback chain
    const result = await this.executeWithFallbacks(
      request,
      [decision.primary, ...decision.fallbacks]
    );
 
    // Record metrics
    this.metrics.recordGeneration({
      provider: result.provider,
      latencyMs: result.latencyMs,
      cost: result.cost,
      routingReason: decision.reason,
    });
 
    this.budget.recordSpend(result.cost);
    return result;
  }
 
  private async executeWithFallbacks(
    request: GenerationRequest,
    providers: BaseProvider[]
  ): Promise<GenerationResult> {
    let lastError: Error | null = null;
 
    for (const provider of providers) {
      const breaker = this.breakers.get(provider.name)!;
 
      try {
        return await breaker.execute(() => provider.generate(request));
      } catch (error) {
        lastError = error as Error;
        this.metrics.recordFailure(provider.name, error as Error);
        continue;
      }
    }
 
    throw new Error(
      `All providers failed. Last error: ${lastError?.message}`
    );
  }
 
  private async getHealthMap() {
    const entries = await Promise.all(
      Array.from(this.providers.entries()).map(async ([name, provider]) => {
        const health = await provider.checkHealth();
        return [name, health] as const;
      })
    );
    return new Map(entries);
  }
 
  private estimateMaxCost(request: GenerationRequest): number {
    return Math.max(
      ...Array.from(this.providers.values()).map(p => p.estimateCost(request))
    );
  }
}
 
interface PipelineConfig {
  dailyBudget: number;
  providers: Record<string, any>;
}

// src/monitoring/budget.ts
export class BudgetTracker {
  private spent = 0;
  private readonly dailyLimit: number;
  private resetTime: number;
 
  constructor(dailyLimit: number) {
    this.dailyLimit = dailyLimit;
    this.resetTime = this.getNextMidnight();
  }
 
  canSpend(amount: number): boolean {
    this.checkReset();
    return this.spent + amount <= this.dailyLimit;
  }
 
  recordSpend(amount: number): void {
    this.spent += amount;
  }
 
  remaining(): number {
    this.checkReset();
    return this.dailyLimit - this.spent;
  }
 
  utilizationPercent(): number {
    return (this.spent / this.dailyLimit) * 100;
  }
 
  /**
   * Returns a cost multiplier based on remaining budget.
   * When budget is nearly exhausted, strongly prefer cheaper providers.
   */
  getCostSensitivity(): number {
    const utilization = this.utilizationPercent();
    if (utilization < 50) return 1.0;  // Normal routing
    if (utilization < 75) return 1.5;  // Slightly prefer cheaper
    if (utilization < 90) return 3.0;  // Strongly prefer cheaper
    return 10.0;                        // Only use cheapest available
  }
 
  private checkReset(): void {
    if (Date.now() > this.resetTime) {
      this.spent = 0;
      this.resetTime = this.getNextMidnight();
    }
  }
 
  private getNextMidnight(): number {
    const tomorrow = new Date();
    tomorrow.setDate(tomorrow.getDate() + 1);
    tomorrow.setHours(0, 0, 0, 0);
    return tomorrow.getTime();
  }
}

// src/pipeline/batch.ts
import { GenerationRequest, GenerationResult } from '../providers/base';
import { PipelineOrchestrator } from './orchestrator';
 
export class BatchProcessor {
  constructor(
    private pipeline: PipelineOrchestrator,
    private concurrency: number = 5
  ) {}
 
  async processBatch(
    requests: GenerationRequest[],
    onProgress?: (completed: number, total: number) => void
  ): Promise<GenerationResult[]> {
    const results: GenerationResult[] = [];
    const queue = [...requests];
    let completed = 0;
 
    const workers = Array.from({ length: this.concurrency }, async () => {
      while (queue.length > 0) {
        const request = queue.shift()!;
        try {
          const result = await this.pipeline.generate(request);
          results.push(result);
        } catch (error) {
          results.push(this.createErrorResult(request, error as Error));
        }
        completed++;
        onProgress?.(completed, requests.length);
      }
    });
 
    await Promise.all(workers);
    return results;
  }
 
  private createErrorResult(request: GenerationRequest, error: Error): GenerationResult {
    return {
      imageUrl: '',
      provider: 'error',
      model: 'none',
      latencyMs: 0,
      cost: 0,
      metadata: { error: error.message, prompt: request.prompt },
    };
  }
}

// src/pipeline/rate-limiter.ts
import { Queue, Worker } from 'bull';
import Redis from 'ioredis';
 
interface RateLimitConfig {
  requestsPerMinute: number;
  requestsPerDay: number;
  concurrentRequests: number;
}
 
export class ProviderRateLimiter {
  private redis: Redis;
  private configs: Map<string, RateLimitConfig> = new Map();
 
  constructor(redisUrl: string) {
    this.redis = new Redis(redisUrl);
    this.initDefaults();
  }
 
  private initDefaults(): void {
    this.configs.set('dalle', {
      requestsPerMinute: 50,
      requestsPerDay: 10000,
      concurrentRequests: 5,
    });
    this.configs.set('flux', {
      requestsPerMinute: 100,
      requestsPerDay: 50000,
      concurrentRequests: 10,
    });
    this.configs.set('stability', {
      requestsPerMinute: 150,
      requestsPerDay: 100000,
      concurrentRequests: 20,
    });
  }
 
  async canProceed(provider: string): Promise<boolean> {
    const config = this.configs.get(provider);
    if (!config) return true;
 
    const minuteKey = `ratelimit:${provider}:minute:${Math.floor(Date.now() / 60000)}`;
    const dayKey = `ratelimit:${provider}:day:${new Date().toISOString().slice(0, 10)}`;
    const concurrentKey = `ratelimit:${provider}:concurrent`;
 
    const [minuteCount, dayCount, concurrent] = await Promise.all([
      this.redis.get(minuteKey).then(v => parseInt(v || '0')),
      this.redis.get(dayKey).then(v => parseInt(v || '0')),
      this.redis.get(concurrentKey).then(v => parseInt(v || '0')),
    ]);
 
    return (
      minuteCount < config.requestsPerMinute &&
      dayCount < config.requestsPerDay &&
      concurrent < config.concurrentRequests
    );
  }
 
  async acquire(provider: string): Promise<() => Promise<void>> {
    const minuteKey = `ratelimit:${provider}:minute:${Math.floor(Date.now() / 60000)}`;
    const dayKey = `ratelimit:${provider}:day:${new Date().toISOString().slice(0, 10)}`;
    const concurrentKey = `ratelimit:${provider}:concurrent`;
 
    await Promise.all([
      this.redis.incr(minuteKey).then(() => this.redis.expire(minuteKey, 60)),
      this.redis.incr(dayKey).then(() => this.redis.expire(dayKey, 86400)),
      this.redis.incr(concurrentKey),
    ]);
 
    // Return release function
    return async () => {
      await this.redis.decr(concurrentKey);
    };
  }
}

// src/pipeline/postprocess.ts
import sharp from 'sharp';
import { GenerationResult } from '../providers/base';
 
export class PostProcessor {
  async process(result: GenerationResult, options: PostProcessOptions): Promise<ProcessedResult> {
    const imageBuffer = await this.downloadImage(result.imageUrl);
 
    // Validate the image is not corrupted
    const metadata = await sharp(imageBuffer).metadata();
    if (!metadata.width || !metadata.height) {
      throw new Error('Generated image is corrupted or empty');
    }
 
    // Apply optimizations
    let processed = sharp(imageBuffer);
 
    if (options.maxWidth && metadata.width > options.maxWidth) {
      processed = processed.resize(options.maxWidth, null, { fit: 'inside' });
    }
 
    if (options.format) {
      processed = processed.toFormat(options.format, {
        quality: options.quality || 85,
        effort: 6,
      });
    }
 
    // NSFW detection placeholder - integrate your preferred service
    if (options.safetyCheck) {
      await this.performSafetyCheck(imageBuffer);
    }
 
    const outputBuffer = await processed.toBuffer();
    const outputMetadata = await sharp(outputBuffer).metadata();
 
    return {
      buffer: outputBuffer,
      width: outputMetadata.width!,
      height: outputMetadata.height!,
      format: options.format || metadata.format || 'png',
      sizeBytes: outputBuffer.length,
      originalResult: result,
    };
  }
 
  private async downloadImage(url: string): Promise<Buffer> {
    const response = await fetch(url);
    if (!response.ok) throw new Error(`Failed to download: ${response.status}`);
    return Buffer.from(await response.arrayBuffer());
  }
 
  private async performSafetyCheck(buffer: Buffer): Promise<void> {
    // Integrate with your NSFW detection service
    // Throw if content violates policies
  }
}
 
interface PostProcessOptions {
  maxWidth?: number;
  format?: 'webp' | 'avif' | 'png' | 'jpeg';
  quality?: number;
  safetyCheck?: boolean;
}
 
interface ProcessedResult {
  buffer: Buffer;
  width: number;
  height: number;
  format: string;
  sizeBytes: number;
  originalResult: GenerationResult;
}

// src/monitoring/metrics.ts
import { Counter, Histogram, Gauge, Registry } from 'prom-client';
 
export class MetricsCollector {
  private registry: Registry;
  private generationCounter: Counter;
  private latencyHistogram: Histogram;
  private costGauge: Gauge;
  private activeRequests: Gauge;
  private failureCounter: Counter;
 
  constructor() {
    this.registry = new Registry();
 
    this.generationCounter = new Counter({
      name: 'image_generation_total',
      help: 'Total image generations',
      labelNames: ['provider', 'model', 'routing_reason', 'status'],
      registers: [this.registry],
    });
 
    this.latencyHistogram = new Histogram({
      name: 'image_generation_duration_seconds',
      help: 'Image generation latency',
      labelNames: ['provider', 'model'],
      buckets: [1, 2, 5, 10, 20, 30, 60, 120],
      registers: [this.registry],
    });
 
    this.costGauge = new Gauge({
      name: 'image_generation_cost_dollars',
      help: 'Cumulative cost in dollars',
      labelNames: ['provider'],
      registers: [this.registry],
    });
 
    this.activeRequests = new Gauge({
      name: 'image_generation_active_requests',
      help: 'Currently active generation requests',
      labelNames: ['provider'],
      registers: [this.registry],
    });
 
    this.failureCounter = new Counter({
      name: 'image_generation_failures_total',
      help: 'Total generation failures',
      labelNames: ['provider', 'error_type'],
      registers: [this.registry],
    });
  }
 
  recordGeneration(data: {
    provider: string;
    latencyMs: number;
    cost: number;
    routingReason: string;
  }): void {
    this.generationCounter.inc({
      provider: data.provider,
      routing_reason: data.routingReason,
      status: 'success',
    });
    this.latencyHistogram.observe(
      { provider: data.provider },
      data.latencyMs / 1000
    );
    this.costGauge.inc({ provider: data.provider }, data.cost);
  }
 
  recordFailure(provider: string, error: Error): void {
    this.failureCounter.inc({
      provider,
      error_type: error.constructor.name,
    });
  }
 
  async getMetrics(): Promise<string> {
    return this.registry.metrics();
  }
}

# docker-compose.yaml
version: '3.8'
services:
  pipeline:
    build: .
    ports:
      - '3000:3000'
    environment:
      - REDIS_URL=redis://redis:6379
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - REPLICATE_API_TOKEN=${REPLICATE_API_TOKEN}
      - STABILITY_API_KEY=${STABILITY_API_KEY}
    depends_on:
      - redis
 
  redis:
    image: redis:7-alpine
    ports:
      - '6379:6379'
 
  prometheus:
    image: prom/prometheus
    volumes:
      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml
    ports:
      - '9090:9090'
 
  grafana:
    image: grafana/grafana
    ports:
      - '3001:3000'
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: image-pipeline
spec:
  replicas: 3
  selector:
    matchLabels:
      app: image-pipeline
  template:
    metadata:
      labels:
        app: image-pipeline
    spec:
      containers:
        - name: pipeline
          image: your-registry/image-pipeline:latest
          resources:
            requests:
              memory: '512Mi'
              cpu: '500m'
            limits:
              memory: '1Gi'
              cpu: '1000m'
          ports:
            - containerPort: 3000
          env:
            - name: REDIS_URL
              valueFrom:
                secretKeyRef:
                  name: pipeline-secrets
                  key: redis-url
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: image-pipeline-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: image-pipeline
  minReplicas: 3
  maxReplicas: 20
  metrics:
    - type: Pods
      pods:
        metric:
          name: image_generation_active_requests
        target:
          type: AverageValue
          averageValue: '8'

// tests/pipeline.test.ts
import { describe, it, expect, beforeAll } from 'vitest';
import { PipelineOrchestrator } from '../src/pipeline/orchestrator';
import { DalleProvider } from '../src/providers/dalle';
import { FluxProvider } from '../src/providers/flux';
 
describe('Pipeline Orchestrator', () => {
  let pipeline: PipelineOrchestrator;
 
  beforeAll(() => {
    pipeline = new PipelineOrchestrator({ dailyBudget: 100 });
    pipeline.registerProvider(new DalleProvider(process.env.OPENAI_API_KEY!));
    pipeline.registerProvider(new FluxProvider(process.env.REPLICATE_API_TOKEN!));
  });
 
  it('should generate an image with automatic provider selection', async () => {
    const result = await pipeline.generate({
      prompt: 'A serene mountain landscape at sunset with golden light',
      width: 1024,
      height: 1024,
      quality: 'standard',
    });
 
    expect(result.imageUrl).toBeTruthy();
    expect(result.provider).toBeDefined();
    expect(result.latencyMs).toBeGreaterThan(0);
    expect(result.cost).toBeGreaterThan(0);
  });
 
  it('should route text-heavy prompts to DALL-E', async () => {
    const result = await pipeline.generate({
      prompt: 'A neon sign that reads "OPEN 24 HOURS" in a rainy city street',
      width: 1024,
      height: 1024,
      quality: 'standard',
    });
 
    expect(result.provider).toBe('dalle');
  });
 
  it('should fallback when primary provider fails', async () => {
    // Simulate failure by using invalid credentials for primary
    const result = await pipeline.generate({
      prompt: 'Abstract geometric patterns in blue and gold',
      width: 1024,
      height: 1024,
      quality: 'standard',
    });
 
    expect(result.imageUrl).toBeTruthy();
  });
});

// src/config/secrets.ts
import { SecretsManagerClient, GetSecretValueCommand } from '@aws-sdk/client-secrets-manager';
 
export async function loadProviderKeys(): Promise<Record<string, string>> {
  const client = new SecretsManagerClient({ region: 'us-east-1' });
  const command = new GetSecretValueCommand({ SecretId: 'image-pipeline/api-keys' });
  const response = await client.send(command);
  return JSON.parse(response.SecretString || '{}');
}

export function sanitizePrompt(prompt: string): string {
  // Remove potential injection attempts
  const sanitized = prompt
    .replace(/\[system\]/gi, '')
    .replace(/\[assistant\]/gi, '')
    .replace(/ignore previous/gi, '')
    .trim();
 
  // Enforce length limits
  if (sanitized.length > 4000) {
    return sanitized.slice(0, 4000);
  }
 
  return sanitized;
}