Architecture

This comprehensive guide outlines the internal architecture of @DevilsDev/rag-pipeline-utils, emphasizing enterprise-grade modularity, plugin design patterns, and SOLID-compliant structure. Understanding this architecture is crucial for extending the system, creating custom plugins, and optimizing performance.

---

Core Design Philosophy

The architecture is built on proven software engineering principles that ensure scalability, maintainability, and extensibility:

SOLID Principles Implementation

• Single Responsibility Principle (SRP): Each component handles one specific domain concern

  • Loaders only handle document ingestion
  • Embedders only convert text to vectors
  • Retrievers only manage vector storage and search
  • LLMs only generate responses

• Open/Closed Principle (OCP): System is open for extension but closed for modification

  • New plugins can be added without changing core code
  • Plugin interfaces define stable contracts
  • Middleware can be injected without core changes

• Liskov Substitution Principle (LSP): Any plugin can be replaced with another of the same type

  • All PDF loaders implement the same interface
  • OpenAI embedder can be swapped with Cohere embedder
  • Vector stores are interchangeable

• Interface Segregation Principle (ISP): Plugins only depend on interfaces they use

  • Loaders don't depend on LLM interfaces
  • Embedders don't need retriever methods
  • Clean separation of concerns

• Dependency Inversion Principle (DIP): High-level modules don't depend on low-level modules

  • Pipeline depends on plugin abstractions, not implementations
  • Configuration drives dependency injection
  • Easy testing with mock implementations

Additional Architectural Principles

• Plugin-Based Architecture: Complete modularity with hot-swappable components
• Streaming-First Design: Async flows with real-time, token-by-token output
• Configuration-Driven: Environment-safe config via .ragrc.json files
• Event-Driven Processing: Comprehensive event system for monitoring and hooks
• Fail-Fast Validation: Early error detection with detailed error messages
• Observability Built-In: Comprehensive logging, tracing, and metrics collection

---

System Architecture Overview

High-Level Component Diagram

graph TB
    subgraph "Application Layer"
        CLI[CLI Interface]
        API[Programmatic API]
        Dashboard[Evaluation Dashboard]
    end

    subgraph "Core Pipeline"
        Factory[Pipeline Factory]
        Registry[Plugin Registry]
        Pipeline[RAG Pipeline]
        Middleware[Middleware Stack]
    end

    subgraph "Plugin Ecosystem"
        Loaders[Document Loaders]
        Embedders[Embedding Models]
        Retrievers[Vector Stores]
        LLMs[Language Models]
        Rerankers[Context Rerankers]
    end

    subgraph "Infrastructure"
        Config[Configuration Manager]
        Logger[Event Logger]
        Metrics[Metrics Collector]
        Cache[Caching Layer]
    end

    CLI --> Factory
    API --> Factory
    Dashboard --> API

    Factory --> Registry
    Factory --> Pipeline
    Pipeline --> Middleware

    Registry --> Loaders
    Registry --> Embedders
    Registry --> Retrievers
    Registry --> LLMs
    Registry --> Rerankers

    Pipeline --> Config
    Pipeline --> Logger
    Pipeline --> Metrics
    Pipeline --> Cache

Data Flow Architecture

sequenceDiagram
    participant U as User
    participant P as Pipeline
    participant L as Loader
    participant E as Embedder
    participant R as Retriever
    participant LLM as Language Model
    participant Re as Reranker

    Note over U,Re: Document Ingestion Phase
    U->>P: ingest(documents)
    P->>L: load(document)
    L->>P: chunks[]
    P->>E: embed(chunks)
    E->>P: embeddings[]
    P->>R: store(embeddings)
    R->>P: success

    Note over U,Re: Query Processing Phase
    U->>P: query(prompt)
    P->>E: embed(prompt)
    E->>P: queryVector
    P->>R: search(queryVector)
    R->>P: candidates[]
    P->>Re: rerank(candidates, prompt)
    Re->>P: rankedContext[]
    P->>LLM: generate(prompt, context)
    LLM->>P: response
    P->>U: answer + metadata

---

Plugin System Architecture

Plugin Registry Implementation

The PluginRegistry is the heart of the plugin system, managing plugin lifecycle and dependency injection:

// Core plugin registry structure
class PluginRegistry {
  private plugins: Map<string, Map<string, Plugin>> = new Map();
  private contracts: Map<string, PluginContract> = new Map();

  register<T extends Plugin>(
    type: PluginType,
    name: string,
    plugin: T,
    metadata?: PluginMetadata,
  ): void {
    // Validate plugin against contract
    this.validateContract(type, plugin);

    // Register plugin with metadata
    if (!this.plugins.has(type)) {
      this.plugins.set(type, new Map());
    }
    this.plugins.get(type)!.set(name, plugin);

    // Emit registration event
    this.eventEmitter.emit("plugin:registered", {
      type,
      name,
      plugin,
      metadata,
    });
  }

  get<T extends Plugin>(type: PluginType, name: string): T {
    const typePlugins = this.plugins.get(type);
    if (!typePlugins?.has(name)) {
      throw new PluginNotFoundError(`Plugin ${type}:${name} not found`);
    }
    return typePlugins.get(name) as T;
  }

  list(type?: PluginType): PluginInfo[] {
    // Return available plugins with metadata
  }
}

Plugin Contract System

Each plugin type implements a strict contract interface:

// Loader Plugin Contract
interface LoaderPlugin {
  readonly metadata: PluginMetadata;

  load(source: string | Buffer, options?: LoaderOptions): Promise<Document[]>;
  supports(mimeType: string): boolean;
  validate(source: string | Buffer): Promise<ValidationResult>;
}

// Embedder Plugin Contract
interface EmbedderPlugin {
  readonly metadata: PluginMetadata;

  embed(texts: string[], options?: EmbedderOptions): Promise<Embedding[]>;
  embedSingle(text: string, options?: EmbedderOptions): Promise<Embedding>;
  getDimensions(): number;
  getMaxTokens(): number;
}

// Retriever Plugin Contract
interface RetrieverPlugin {
  readonly metadata: PluginMetadata;

  store(embeddings: EmbeddingDocument[]): Promise<void>;
  search(query: Embedding, options?: SearchOptions): Promise<SearchResult[]>;
  delete(ids: string[]): Promise<void>;
  getStats(): Promise<IndexStats>;
}

// LLM Plugin Contract
interface LLMPlugin {
  readonly metadata: PluginMetadata;

  generate(
    prompt: string,
    context?: string[],
    options?: LLMOptions,
  ): AsyncIterable<string>;
  generateSync(
    prompt: string,
    context?: string[],
    options?: LLMOptions,
  ): Promise<string>;
  getTokenCount(text: string): number;
  getMaxContextLength(): number;
}

Plugin Metadata System

interface PluginMetadata {
  name: string;
  version: string;
  description: string;
  author: string;
  license: string;
  homepage?: string;
  repository?: string;
  keywords: string[];

  // Plugin-specific metadata
  capabilities: string[];
  requirements: {
    node: string;
    dependencies?: Record<string, string>;
    environment?: string[];
  };

  // Performance characteristics
  performance: {
    throughput?: string;
    latency?: string;
    memoryUsage?: string;
  };

  // Configuration schema
  configSchema?: JSONSchema;
}

---

Pipeline Factory Pattern

The createRagPipeline function implements the Factory pattern to create configured pipeline instances:

export function createRagPipeline(config: PipelineConfig): RagPipeline {
  // Initialize plugin registry
  const registry = new PluginRegistry();

  // Register built-in plugins
  registerBuiltinPlugins(registry);

  // Load custom plugins from config
  if (config.plugins) {
    await loadCustomPlugins(registry, config.plugins);
  }

  // Resolve plugin instances
  const loader = registry.get("loader", config.loader);
  const embedder = registry.get("embedder", config.embedder);
  const retriever = registry.get("retriever", config.retriever);
  const llm = registry.get("llm", config.llm);
  const reranker = config.useReranker
    ? registry.get("reranker", config.reranker || "default")
    : null;

  // Create middleware stack
  const middleware = createMiddlewareStack(config.middleware || []);

  // Initialize pipeline with dependencies
  return new RagPipeline({
    loader,
    embedder,
    retriever,
    llm,
    reranker,
    middleware,
    config: config.pipelineConfig || {},
    eventEmitter: new EventEmitter(),
    logger: createLogger(config.logging || {}),
    metrics: new MetricsCollector(config.metrics || {}),
  });
}

---

Middleware Architecture

The middleware system allows for cross-cutting concerns like retry logic, caching, and monitoring:

// Middleware interface
interface Middleware {
  name: string;
  priority: number;

  beforeLoad?(context: LoadContext): Promise<LoadContext>;
  afterLoad?(context: LoadContext, result: Document[]): Promise<Document[]>;

  beforeEmbed?(context: EmbedContext): Promise<EmbedContext>;
  afterEmbed?(context: EmbedContext, result: Embedding[]): Promise<Embedding[]>;

  beforeRetrieve?(context: RetrieveContext): Promise<RetrieveContext>;
  afterRetrieve?(
    context: RetrieveContext,
    result: SearchResult[],
  ): Promise<SearchResult[]>;

  beforeGenerate?(context: GenerateContext): Promise<GenerateContext>;
  afterGenerate?(context: GenerateContext, result: string): Promise<string>;

  onError?(error: Error, context: any): Promise<void>;
}

// Built-in middleware examples
class RetryMiddleware implements Middleware {
  constructor(private options: RetryOptions) {}

  async beforeGenerate(context: GenerateContext): Promise<GenerateContext> {
    context.retryCount = 0;
    context.maxRetries = this.options.maxRetries;
    return context;
  }

  async onError(error: Error, context: any): Promise<void> {
    if (context.retryCount < context.maxRetries) {
      context.retryCount++;
      // Implement exponential backoff
      await this.delay(Math.pow(2, context.retryCount) * 1000);
      throw new RetryableError(error);
    }
    throw error;
  }
}

class CachingMiddleware implements Middleware {
  constructor(private cache: Cache) {}

  async beforeEmbed(context: EmbedContext): Promise<EmbedContext> {
    const cacheKey = this.generateCacheKey(context.texts);
    const cached = await this.cache.get(cacheKey);
    if (cached) {
      context.cachedResult = cached;
    }
    return context;
  }

  async afterEmbed(
    context: EmbedContext,
    result: Embedding[],
  ): Promise<Embedding[]> {
    if (!context.cachedResult) {
      const cacheKey = this.generateCacheKey(context.texts);
      await this.cache.set(cacheKey, result, { ttl: 3600 });
    }
    return context.cachedResult || result;
  }
}

---

Event System & Observability

The architecture includes a comprehensive event system for monitoring, debugging, and analytics:

// Event types emitted by the system
interface PipelineEvents {
  "pipeline:created": { config: PipelineConfig };
  "pipeline:destroyed": { pipelineId: string };

  "document:loading": { source: string; loader: string };
  "document:loaded": { source: string; chunks: number; duration: number };
  "document:error": { source: string; error: Error };

  "embedding:started": { texts: number; embedder: string };
  "embedding:completed": { embeddings: number; duration: number };
  "embedding:cached": { texts: number; cacheHit: boolean };

  "retrieval:query": { query: string; retriever: string };
  "retrieval:results": { results: number; duration: number };

  "generation:started": { prompt: string; llm: string };
  "generation:token": { token: string; position: number };
  "generation:completed": {
    response: string;
    duration: number;
    tokens: number;
  };

  "error:handled": { error: Error; context: string; recovered: boolean };
}

// Event-driven metrics collection
class MetricsCollector {
  constructor(private config: MetricsConfig) {
    this.setupEventListeners();
  }

  private setupEventListeners(): void {
    // Track performance metrics
    this.on("embedding:completed", (event) => {
      this.histogram("embedding.duration", event.duration);
      this.counter("embedding.requests").inc();
    });

    this.on("generation:completed", (event) => {
      this.histogram("generation.duration", event.duration);
      this.histogram("generation.tokens", event.tokens);
      this.counter("generation.requests").inc();
    });

    // Track error rates
    this.on("error:handled", (event) => {
      this.counter("errors.total", {
        context: event.context,
        recovered: event.recovered.toString(),
      }).inc();
    });
  }
}

---

Security Architecture

Security is built into the architecture at multiple levels:

Plugin Sandboxing

• Plugins run in isolated contexts with limited system access
• API key management through secure configuration
• Input validation and sanitization at plugin boundaries

Configuration Security

• Encrypted storage of sensitive configuration data
• Environment variable injection for secrets
• Configuration schema validation

Network Security

• TLS/SSL enforcement for all external API calls
• Request rate limiting and timeout handling
• Secure credential rotation support

---

Enterprise Architecture Components

Dependency Injection Container

The enterprise-grade dependency injection system provides IoC (Inversion of Control) for modular, testable architecture:

// src/core/dependency-injection.js
class DependencyContainer {
  private services: Map<string, ServiceDefinition> = new Map();
  private instances: Map<string, any> = new Map();

  register<T>(
    name: string,
    factory: ServiceFactory<T>,
    options?: ServiceOptions,
  ): void {
    this.services.set(name, {
      factory,
      lifecycle: options?.lifecycle || "singleton",
      dependencies: options?.dependencies || [],
    });
  }

  resolve<T>(name: string): T {
    if (this.instances.has(name)) {
      return this.instances.get(name);
    }

    const service = this.services.get(name);
    if (!service) {
      throw new ServiceNotFoundError(`Service ${name} not registered`);
    }

    // Resolve dependencies
    const dependencies = service.dependencies.map((dep) => this.resolve(dep));

    // Create instance
    const instance = service.factory(...dependencies);

    if (service.lifecycle === "singleton") {
      this.instances.set(name, instance);
    }

    return instance;
  }
}

SLO Monitoring System

Built-in Service Level Objectives tracking with error budgets and alerting:

// src/observability/slo-monitor.js
class SLOMonitor {
  private slos: Map<string, SLODefinition> = new Map();
  private measurements: TimeSeriesDB = new TimeSeriesDB();

  defineSLO(name: string, definition: SLODefinition): void {
    this.slos.set(name, {
      ...definition,
      errorBudget: this.calculateErrorBudget(definition),
    });
  }

  recordMeasurement(sloName: string, success: boolean, latency?: number): void {
    const measurement = {
      timestamp: Date.now(),
      success,
      latency,
      slo: sloName,
    };

    this.measurements.record(measurement);
    this.checkSLOViolation(sloName, measurement);
  }

  getErrorBudgetStatus(sloName: string): ErrorBudgetStatus {
    const slo = this.slos.get(sloName);
    const recentMeasurements = this.measurements.getRecent(sloName, slo.window);

    return {
      remaining: this.calculateRemainingBudget(slo, recentMeasurements),
      burnRate: this.calculateBurnRate(recentMeasurements),
      alerting: this.shouldAlert(slo, recentMeasurements),
    };
  }
}

External API Mocking Infrastructure

Deterministic test infrastructure with network simulation for reliable CI/CD:

// __tests__/mocks/external-apis.js
class ExternalAPIMocker {
  private mocks: Map<string, MockDefinition> = new Map();
  private networkSimulator: NetworkSimulator;

  mockAPI(service: string, config: MockConfig): void {
    this.mocks.set(service, {
      responses: config.responses,
      latency: config.latency || { min: 100, max: 500 },
      errorRate: config.errorRate || 0.05,
      rateLimiting: config.rateLimiting,
    });
  }

  async simulateRequest(
    service: string,
    request: APIRequest,
  ): Promise<APIResponse> {
    const mock = this.mocks.get(service);

    // Simulate network conditions
    await this.networkSimulator.delay(mock.latency);

    // Simulate errors
    if (Math.random() < mock.errorRate) {
      throw new MockNetworkError("Simulated network failure");
    }

    // Return mock response
    return this.generateResponse(mock, request);
  }
}

---

Advanced AI Architecture

Multi-Modal Processing Engine

Handle text, images, and structured data in unified pipelines:

// src/ai/multimodal/multi-modal-processor.js
class MultiModalProcessor {
  private processors: Map<string, ModalityProcessor> = new Map();

  async process(input: MultiModalInput): Promise<ProcessedContent> {
    const results = await Promise.all(
      input.modalities.map(async (modality) => {
        const processor = this.processors.get(modality.type);
        return processor.process(modality.content, modality.metadata);
      }),
    );

    return this.fuseResults(results, input.fusionStrategy);
  }
}

Federated Learning Coordinator

Distributed model training with privacy-preserving aggregation:

// src/ai/federation/federated-learning-coordinator.js
class FederatedLearningCoordinator {
  async coordinateTraining(
    participants: FederatedNode[],
  ): Promise<GlobalModel> {
    // Distribute training tasks
    const localUpdates = await this.distributeTraining(participants);

    // Aggregate updates with differential privacy
    const aggregatedUpdate = await this.aggregateWithPrivacy(localUpdates);

    // Update global model
    return this.updateGlobalModel(aggregatedUpdate);
  }
}

Adaptive Retrieval Engine

Dynamic retrieval strategies with performance optimization:

// src/ai/retrieval/adaptive-retrieval-engine.js
class AdaptiveRetrievalEngine {
  async adaptiveRetrieve(
    query: Query,
    context: RetrievalContext,
  ): Promise<RetrievalResult> {
    // Analyze query complexity
    const strategy = await this.selectStrategy(query, context);

    // Execute retrieval with chosen strategy
    const results = await this.executeStrategy(strategy, query);

    // Learn from results for future optimization
    await this.updateStrategyPerformance(strategy, results);

    return results;
  }
}

---

Enhanced Developer Tools

CLI Doctor Diagnostics

Comprehensive system health checking and troubleshooting:

// src/cli/doctor-command.js
class DoctorCommand {
  async runDiagnostics(): Promise<DiagnosticReport> {
    const checks = [
      this.checkNodeVersion(),
      this.checkDependencies(),
      this.checkConfiguration(),
      this.checkPluginHealth(),
      this.checkExternalServices(),
      this.checkPerformanceBottlenecks(),
    ];

    const results = await Promise.allSettled(checks);
    return this.generateReport(results);
  }
}

Plugin Marketplace

Certified plugin ecosystem with discovery and installation workflows:

// src/core/plugin-marketplace/marketplace.js
class PluginMarketplace {
  async discoverPlugins(criteria: SearchCriteria): Promise<PluginListing[]> {
    const plugins = await this.searchRegistry(criteria);
    return plugins.filter((plugin) => this.meetsCertificationStandards(plugin));
  }

  async installPlugin(pluginId: string): Promise<InstallationResult> {
    // Verify plugin certification
    await this.verifyCertification(pluginId);

    // Install with security scanning
    return this.secureInstall(pluginId);
  }
}

---

This enterprise-grade architecture enables @DevilsDev/rag-pipeline-utils to scale from simple prototypes to mission-critical production systems while maintaining flexibility, security, and observability. Continue to Usage for practical implementation examples, or explore Plugins to learn about creating custom components.