Production RAG Architecture

import asyncio
import logging
from collections import defaultdict

from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever

log = logging.getLogger(__name__)


async def _one(name: str, retriever: BaseRetriever, query: str, timeout_s: float) -> list[Document]:
    try:
        docs = await asyncio.wait_for(retriever.ainvoke(query), timeout=timeout_s)
    except TimeoutError:
        return []                         # a slow backend degrades the answer, it does not block it
    except Exception:
        log.exception("retriever %s failed", name)
        return []                         # and so does a failing one, instead of failing the whole gather
    for doc in docs:
        doc.metadata["retriever"] = name
    return docs


async def fan_out(query: str, retrievers: dict[str, BaseRetriever], timeout_s: float = 0.3,
                  rrf_k: int = 60, top_n: int = 20) -> list[Document]:
    """Run every retriever concurrently, then fuse the rankings with reciprocal rank fusion."""
    ranked_lists = await asyncio.gather(*(_one(n, r, query, timeout_s) for n, r in retrievers.items()))
    scores: dict[str, float] = defaultdict(float)
    by_id: dict[str, Document] = {}
    for docs in ranked_lists:
        for rank, doc in enumerate(docs, start=1):
            chunk_id = doc.metadata["chunk_id"]
            scores[chunk_id] += 1.0 / (rrf_k + rank)
            by_id.setdefault(chunk_id, doc)
    best = sorted(scores, key=scores.get, reverse=True)[:top_n]
    return [by_id[chunk_id] for chunk_id in best]

package com.example.docsassistant.rag.architecture;

import org.springframework.ai.chat.client.ChatClient;
import org.springframework.ai.rag.advisor.RetrievalAugmentationAdvisor;
import org.springframework.ai.rag.generation.augmentation.ContextualQueryAugmenter;
import org.springframework.ai.rag.preretrieval.query.expansion.MultiQueryExpander;
import org.springframework.ai.rag.retrieval.join.ConcatenationDocumentJoiner;
import org.springframework.ai.rag.retrieval.search.VectorStoreDocumentRetriever;
import org.springframework.ai.vectorstore.VectorStore;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;

@Configuration
public class RagPipelineConfig {

    @Bean
    RetrievalAugmentationAdvisor ragAdvisor(ChatClient.Builder builder, VectorStore vectorStore,
                                            TaskExecutor applicationTaskExecutor) {
        return RetrievalAugmentationAdvisor.builder()
                // three paraphrases plus the original question, retrieved concurrently
                .queryExpander(MultiQueryExpander.builder()
                        .chatClientBuilder(builder.clone())
                        .numberOfQueries(3)
                        .includeOriginal(true)
                        .build())
                .documentRetriever(VectorStoreDocumentRetriever.builder()
                        .vectorStore(vectorStore)
                        .similarityThreshold(0.5)
                        .topK(8)
                        .build())
                .documentJoiner(new ConcatenationDocumentJoiner())      // deduplicates by document ID
                .queryAugmenter(ContextualQueryAugmenter.builder().allowEmptyContext(false).build())
                .taskExecutor(applicationTaskExecutor)
                .build();
    }

    @Bean
    ChatClient docsChatClient(ChatClient.Builder builder, RetrievalAugmentationAdvisor ragAdvisor) {
        return builder.defaultAdvisors(ragAdvisor).build();
    }
}

package com.example.docsassistant.l4j.architecture;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import dev.langchain4j.model.scoring.ScoringModel;
import dev.langchain4j.rag.DefaultRetrievalAugmentor;
import dev.langchain4j.rag.RetrievalAugmentor;
import dev.langchain4j.rag.content.aggregator.ReRankingContentAggregator;
import dev.langchain4j.rag.content.retriever.ContentRetriever;
import dev.langchain4j.rag.query.router.DefaultQueryRouter;

public final class FanOutAugmentor {

    private FanOutAugmentor() {
    }

    public static RetrievalAugmentor create(ContentRetriever vectorRetriever, ContentRetriever keywordRetriever,
                                            ScoringModel crossEncoder) {
        ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
        return DefaultRetrievalAugmentor.builder()
                .queryRouter(new DefaultQueryRouter(vectorRetriever, keywordRetriever))
                .contentAggregator(ReRankingContentAggregator.builder()
                        .scoringModel(crossEncoder)
                        .minScore(0.3)
                        .maxResults(6)
                        .build())
                .executor(executor)            // both retrievers run concurrently
                .build();
    }
}