Hexagonal Architecture for Provider-Agnostic RAG Pipelines

Demo stack: LangChain, LlamaCpp, Mistral.ai, Qwen Embeddings, PostgreSQL, Telegraf, Prometheus on Podman/Docker.

---

I recently came across Machine Learning Mastery's guide on Building a RAG Pipeline with llama.cpp and decided to try it myself.

One thing quickly became apparent: some of the APIs and methods used in the example had already changed or become deprecated.
That led me to a bigger question:
How do you design a RAG system that can evolve as the underlying technologies change?
Instead of tightly coupling the application to a specific LLM, vector database, or document-processing framework, I experimented with a more modular architecture.
This way I get a more resilient architecture, for instance if I need a transition between local on-premise to fully on cloud or just hybrid; ability to switch to different database like PostgreSQL instead of ChromaDB, due to the data integration and ACID compliance!

A few principles became particularly important:

• 🧩 Decouples the technology dependency Provider implementations can be replaced without changing core logic, as the architecture relies on interfaces, abstractions & models, while Lazy Imports isolate technology-specific dependencies.

def importOnCall(..):
    from.. import .. # Lazy import
    return ResultModel(..) # tied to Abstracts & Models

Dependencies are wrapped as "optional-dependencies" based on the desired inferastructure flavors, for instance pip install yoga1290.rag[infrastructure-local-llama-postgres] is a shortcut to install all dependencies required for that infrastructure; see pyproject.toml

infrastructure-local-llama-postgres = [
    "yoga1290-rag[documents, postgres, llama-cpp]",
]

• 🔌️ On-premise, hybrid & Multi-platform support Decoupling from certain SDK or API or local implementations keep my options open to plug & play the document sources, different LLM models or vector stores/databases, regardless of being local, hybrid or on cloud services. 🖥️🔄☁️

vectorstore: VectorStore = PGVectorStore(...)
vectorstore: VectorStore = OtherVectorDBService(..)

• 🛠️ Separation of responsibilities
  Each layer has a focused responsibility:

  • 🧠 Domain — business concepts, models, and contracts
  • 🔌 Ports — define what the application needs
  • ⚙️ Infrastructure — provides concrete implementations
  • 🔄 Pipelines/Application — orchestrates the workflow
  • 🧩 Composition — selects and wires implementations together

• 📦️ Observability, Containerization & Resource/Network monitoring, Confiurations are pass in form of Environment Variables, see sample.env. Telegraf is used to monitor the resource consumsion and network traffic, and project metrics to Prometheus. See the 👁️ Observability section.

---

---

Overview

Startup this Jupiter notebook, PostgressSQL database & Promethus+Grafana servers via: docker compose up notebook.

• 📚️ Ingestion Pipeline
• 🔍️ Retrieval Pipeline
• 🏗️ Extraction Pipeline
• 👁️ Observability
• 🔧️ Turning Implementations

---

📚️ Ingestion Pipeline

All ENV variables required to run rag-ingestion.py are wrapped in rag-ingestion.env

# compose.yaml
rag-ingestion:
    build:
      context: ./ci/container-python
      # yoga1290.rag[infrastructure-local-llama-postgres] in a python slim image
    env_file:
      - rag-ingestion.env # SAMPLE ENV
    command: ['python', "rag-ingestion.py"]
    working_dir: /usr/app
    volumes:
      - "./:/usr/app"
      - "~/LLM:/LLM"
    depends_on:
      postgres:
        condition: service_healthy
    networks:
      - vectorstore # PRIVATE NETWORK

# rag-ingestion.py

########################### INTERFACES & MODELS ###########################
from yoga1290.rag.domain.models import (
                            Document,
                            ParsedDocument,
                            SearchDocument)
from yoga1290.rag.domain.ports import (
                            DocumentSource,
                            DocumentParser,
                            VectorStore,
                            SearchPreparer,
                            Embedder)
########################### IMPLEMENTATIONS ###########################
from yoga1290.rag.infrastructure.embeddings import LlamaCppEmbeddings
from yoga1290.rag.infrastructure.vectorstores import PGVectorStore
from yoga1290.rag.infrastructure.local.sources.csv_document_source import CsvDocumentSource
from yoga1290.rag.infrastructure.local.parsing.local_document_parser import LocalDocumentParser
from yoga1290.rag.infrastructure.local.search.local_search_preparer import LocalSearchPreparer

from yoga1290.rag.application.pipelines import IngestionPipeline

document_source: DocumentSource = (
                    CsvDocumentSource())

document_parser: DocumentParser = (
                    LocalDocumentParser())

document_search_preparer: SearchPreparer = (
                    LocalSearchPreparer())

embedder: Embedder = (
                    LlamaCppEmbeddings())

vectorstore: VectorStore = (
                    PGVectorStore(
                        # configration from Environment Variables; connection_string=f"postgresql+psycopg://{os.getenv("POSTGRES_USER")}.."
                        embedder=embedder))

IngestionPipeline(
    parser= document_parser,
    extractor= None,
    classifier= None,
    search_preparer= document_search_preparer,
    vectorstore= vectorstore
).process( documents= document_source )

🔍️ Retrieval Pipeline

All ENV variables required to run rag-retrieval.py are wrapped in rag-retrieval.env

# compose.yaml
rag-retrieval:
    build:
        context: ./ci/container-python
        # yoga1290.rag[infrastructure-local-llama-postgres] in a python slim image
    env_file:
    - rag-retrieval.env
    command: ['python', "rag-retrieval.py"]
    working_dir: /usr/app
    volumes:
    - "./:/usr/app"
    - "~/LLM:/LLM"
    depends_on:
    postgres:
        condition: service_healthy
    networks:
    - vectorstore # PRIVATE NETWORK

# rag-retrieval.py
########################### INTERFACES & MODELS ###########################
from yoga1290.rag.domain.models import (
                            Document,
                            ParsedDocument,
                            SearchDocument,
                            )
from yoga1290.rag.domain.ports import (
                            DocumentSource,
                            DocumentParser,
                            VectorStore,
                            SearchPreparer,
                            Embedder,
                            Retriever)
########################### IMPLEMENTATIONS ###########################
from yoga1290.rag.infrastructure.embeddings import LlamaCppEmbeddings
from yoga1290.rag.infrastructure.vectorstores import PGVectorStore, PGVectorRetriever
from yoga1290.rag.infrastructure.local.sources.csv_document_source import CsvDocumentSource
from yoga1290.rag.infrastructure.local.parsing.local_document_parser import LocalDocumentParser
from yoga1290.rag.infrastructure.local.search.local_search_preparer import LocalSearchPreparer

from yoga1290.rag.factories.llm_factory import LLMFactory
from yoga1290.rag.application.pipelines import RetrievalPipeline

embedder: Embedder = (
                    LlamaCppEmbeddings())

vectorstore: VectorStore = (
                    PGVectorStore(
                        #connection_string=f"postgresql+psycopg://{os.getenv("POSTGRES_USER")}.."
                        embedder=embedder))

pgvector_retriever: Retriever = (
                    PGVectorRetriever(vectorstore=vectorstore, ));

llm_llama= LLMFactory.createLocalLlamaCppLLM()

response =  RetrievalPipeline(
                llm= llm_llama,
                retriever= pgvector_retriever,
                top_k=2,
            ).run( question= "Make a good introduction about my backend skillset" )

print(f"Answer {response.answer}")

🏗️ Extraction Pipeline

Here's an example of asking the LLM (Mistral on llamaCpp) to extract fields from my receipt emails pulled using yoga1290.emailtable that outputs to CSV table.
It simply generates inner prompt per each requested field and collects the responses into a dict map.

# compose.yaml
pull-documents:
    image: ghcr.io/yoga1290/python-emailtable:26.10
    env_file:
      - emailtable.env
    command:
      - python
      - -c
      - |
        from yoga1290.emailtable import processIMAPToCSV
        processIMAPToCSV()
    volumes:
      - "./${IMAP_OUTPUT_DIR}:/usr/app/${IMAP_OUTPUT_DIR}"
    networks:
      - public

########################### INTERFACES & MODELS ###########################
from yoga1290.rag.domain.models import (
                            ParsedDocument,
                            ExtractedData,)
from yoga1290.rag.domain.ports import (
                            DocumentSource,
                            DocumentParser,
                            DocumentExtractor,)
############################################################################
from yoga1290.rag.infrastructure.local.extraction import LlamaCppMistralDocumentExtractor
from yoga1290.rag.infrastructure.local.sources.csv_document_source import CsvDocumentSource
from yoga1290.rag.infrastructure.local.parsing.local_document_parser import LocalDocumentParser

document_source: DocumentSource = (
                    CsvDocumentSource(
                        # configuration injected from Environment Variables
                        # ENV for /src/yoga1290/rag/infrastructure/local/sources/csv_document_source.py
                        LOCAL_CSV_DOCUMENTS_SOURCE_PATH= "documents/output.csv",
                        LOCAL_CSV_DOCUMENTS_SOURCE_FILEPATH_COLUMN="attachments"
                    ))

document_parser: DocumentParser = (
                    LocalDocumentParser())
document_extractor: DocumentExtractor = (
                    LlamaCppMistralDocumentExtractor())

import datetime
table_data = []
for document in document_source:
    try:
        parsed_document: ParsedDocument = (
                            document_parser.parse(document))
        start_time = datetime.datetime.now()
        response = document_extractor.extract(
                            parsed_document=parsed_document,
                            fields= ['Is there a payment receipt?',
                                    'Total Payment Amount',
                                    'Vendor' ,
                                    'Item name',
                                    'Date of purchase'])
        duration = datetime.datetime.now() - start_time
        print(f'response after {duration}: {response.data}')
    except: pass

import pandas as pd
df = pd.DataFrame(table_data)
df.to_csv(f'extractions.csv')
df

👁️ Container Observability

Monitoring the resource consumption, network traffic & isolation can ideicate how well different LLM models can perform under larger sets.
In my compose.yaml for Root-less Podman (recommended) or docker-compose.yml configuration for docker, there're the following 3 containers:

• monitored-job: a container with Python & Telegraf pre-installed, see my Containerfile.
• Promethus: collecting metric data from the Telegraf server in the monitored-job
• Grafana: for visualizing Promethus metrics into an intuitive dashboard; I used the Grafana's dashboard: System Metrics for the Linux Hosts, which is compatible with Telegraf projected metrics but it needs a tweak:
  • Make sure, Prometheus can see the Job container, try query the monitored-job
  • Make sure, the DS_PROMETHEUS dashboard variable matches the name of the Datasource variable in the Grafana's datasource.yml, which is DS_SERVERMONITOR in my case.

🔧️ Tuning Implementations

To add support for a new LLM, you will need to implement on the existing abstracts, interfaces & return the domain's data models, for example LlamaCppLLM:

# %load ./src/yoga1290/rag/domain/ports/llm.py
from abc import ABC, abstractmethod
class LLM(ABC):
    @abstractmethod
    def generate(self, prompt: str) -> str:
        """
        Generate a response from a prompt.
        """
        raise NotImplementedError

# %load ./src/yoga1290/rag/infrastructure/local/llm/llama_cpp_llm.py
from yoga1290.rag.domain.ports import LLM

class LlamaCppLLM(LLM):

    def __init__(
        self,
        model_path: str,
        temperature: float = 0.15,
        max_tokens: int = 450,
        context_size: int = 4096,
        batch_size: int = 384,
    ) -> None:
        from langchain_community.llms import LlamaCpp
        self._llm = LlamaCpp(
            model_path=model_path,
            temperature=temperature,
            max_tokens=max_tokens,
            n_ctx=context_size,
            n_batch=batch_size,
            verbose=False,
        )

    def generate(self, prompt: str) -> str:
        return self._llm.invoke(prompt)