multi source rag

---
title: Multi-Source RAG
category: concepts
created: 2025-12-19
updated: 2025-01-04
tags: [multi-source-rag, federated-search, embedding-models, filter-propagation, heterogeneous-corpora, source-coordination, rag-architecture, postgres-embeddings, interface-consistency, production-issues, sql-generation, ministry-filters, chunk-objects, shared-embeddings, dependency-management, critical-bugs, multi-source-filtering, embedding-space-compatibility, category-5-bugs, embedding-space-validation, filter-bypass-bugs, assistant-rh-bugs, runtime-crashes, autonomous-retrieval, cross-source-compatibility]
sources: [raw/conversations/2025-10-23-codex-assistant-rh-c34c0439.md]
confidence: high
---

# Multi-Source RAG

Retrieval-Augmented Generation architecture that queries multiple heterogeneous document collections simultaneously, combining results for comprehensive response generation. Requires careful coordination of embedding models, filtering logic, and result fusion strategies.

## Core Challenges

### Embedding Space Compatibility

Different document sources may be embedded using different models or at different times, creating incompatible vector spaces. Attempting to compare or rank results across these spaces produces meaningless similarity scores.

The assistant-rh system exemplifies this challenge with three document sources:
- **Service-Public.gouv.fr**: Government service documentation
- **DGAFP**: Civil service administration guidelines  
- **MATTE**: Ministry-specific regulations

Each source potentially uses different embedding models or model versions, making cross-source similarity comparison unreliable.

### Filter Propagation

Query filters (e.g., ministry restrictions, document types, date ranges) must be consistently applied across all sources. Failure to propagate filters creates several issues:

- **Inconsistent Results**: Some sources respect filters while others don't
- **Security Leaks**: Sensitive documents bypass access restrictions
- **Performance Degradation**: Unnecessary retrieval from irrelevant sources

### Interface Consistency

Multiple retrievers must provide uniform interfaces for:
- Query input formats
- Result data structures  
- Metadata schemas
- Error handling patterns

The [interface-inconsistency](/concepts/interface-inconsistency) anti-pattern is particularly problematic in multi-source contexts where downstream consumers must handle results from multiple retrievers polymorphically.

## Architecture Patterns

### Federated Search

Each source maintains its own retriever with source-specific optimizations:

```python
class MultiSourceRetriever:
    def __init__(self, sources: Dict[str, BaseRetriever]):
        self.sources = sources
    
    def search(self, query: str, filters: Dict) -> List[Chunk]:
        all_results = []
        for source_name, retriever in self.sources.items():
            source_filters = self._adapt_filters(filters, source_name)
            results = retriever.search(query, source_filters)
            all_results.extend(self._tag_source(results, source_name))
        return self._rank_and_merge(all_results)

Unified Index

All sources are embedded into a single vector database with source metadata:

# Single embedding space with source tags
chunks = [
    Chunk(content="...", source="service-public", ministry="interior"),
    Chunk(content="...", source="dgafp", ministry="finance"),
    # ...
]

This approach ensures embedding compatibility but may lose source-specific optimizations.

Critical Implementation Issues

Assistant-RH Multi-Source Bugs

The October 2025 Codex analysis identified critical bugs in Assistant-RH's multi-source implementation:

Filter Bypass

def search_multi_source(self, query: str, filters: Dict) -> List[Chunk]:
    # BUG: filters parameter ignored completely
    service_public_results = self.service_public_retriever.search(query)
    dgafp_results = self.dgafp_retriever.search(query) 
    matte_results = self.matte_retriever.search(query)
    # Ministry filters not applied to any source

Embedding Space Assumption

# BUG: Assumes all sources share embedding space
combined_scores = [r.score for r in all_results]  
# Meaningless if sources use different embedding models

Missing Error Handling

# BUG: No validation of source compatibility
def merge_results(self, results_by_source: Dict) -> List[Chunk]:
    # Assumes all sources return compatible Chunk objects
    # Crashes if sources return different data structures

Solutions and Best Practices

Embedding Model Standardization

Ensure all sources use the same embedding model and version:

class StandardizedEmbedder:
    def __init__(self, model_name: str, model_version: str):
        self.model = load_model(model_name, model_version)
        
    def embed_source(self, source_name: str, documents: List[str]):
        # Consistent embedding across all sources
        return self.model.encode(documents)

Filter Translation Layer

Abstract source-specific filter formats:

class FilterTranslator:
    def translate(self, filters: Dict, source: str) -> Dict:
        if source == "service-public":
            return {"category": filters.get("ministry")}
        elif source == "dgafp":  
            return {"department": filters.get("ministry")}
        # Source-specific filter mapping

Result Normalization

Ensure consistent result formats across sources:

def normalize_results(self, results: List, source: str) -> List[Chunk]:
    normalized = []
    for result in results:
        chunk = Chunk(
            content=self._extract_content(result, source),
            score=self._normalize_score(result.score, source),
            metadata=self._standardize_metadata(result, source)
        )
        normalized.append(chunk)
    return normalized

Cross-Source Validation

Implement runtime checks for source compatibility:

def validate_source_compatibility(self, sources: Dict[str, BaseRetriever]):
    embedding_models = {}
    for name, retriever in sources.items():
        model_info = retriever.get_embedding_model_info()
        embedding_models[name] = model_info
        
    if len(set(embedding_models.values())) > 1:
        raise IncompatibleEmbeddingError(
            f"Sources use different embedding models: {embedding_models}"
        )

Performance Considerations

Parallel Retrieval

Query sources concurrently to minimize latency:

import asyncio

async def search_async(self, query: str) -> List[Chunk]:
    tasks = []
    for source in self.sources.values():
        task = asyncio.create_task(source.search_async(query))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return self._merge_results(results)

Result Caching

Cache merged results to avoid repeated multi-source queries:

from functools import lru_cache

@lru_cache(maxsize=1000)
def search_cached(self, query: str, filters_hash: str) -> List[Chunk]:
    filters = self._deserialize_filters(filters_hash)
    return self.search(query, filters)

Load Balancing

Distribute load across source-specific infrastructure:

class LoadBalancedMultiSource:
    def __init__(self, source_weights: Dict[str, float]):
        self.weights = source_weights
        
    def search(self, query: str, max_results: int) -> List[Chunk]:
        results_per_source = self._distribute_load(max_results)
        # Query each source for proportional results