Skip to main content

E-06 RAG Pipelines

Brick E: Data, RAG, context, memory. French version: E-06-rag-pipelines

1. Role and responsibilities​

Haystack pipelines for ingestion (chunking, DLP, embedding) and query (vector or hybrid retrieval, ACL filtering, optional reranking).

  • Embeddings through the Model Gateway (POST /v1/gateway/embeddings) with the embedding model of the knowledge base
  • Retrieval filtered by ACL and clearance, mode vector or hybrid
  • Optional reranking via POST /v1/gateway/rerank
  • Reindex to another embedding model, without mixing vectors
  • Citations with provenance

Each knowledge base has an embedding model of the registry (embedding_model, capability embeddings) and a dimension, chosen at creation (default EAI_RAG_EMBEDDING_MODEL). Every embedding call (document chunks, query, reindex) goes through the Model Gateway controls: kill switch, classification ceiling and retention of the model, region, DLP, policy decision model.invoke, egress guard, quota reservation and a Usage Event measured on the provider response. Chunks are sent with the classification of their document: a document above the model ceiling is refused before any outbound call.

Shipped models:

  • bge-m3-vllm: BAAI/bge-m3 served by vLLM in embedding mode (1024 dimensions, French and English, ceiling secret, no retention), the only shipped model for restricted and secret data;
  • text-embedding-3-small: OpenAI (1536 dimensions, or 256, 512 and 1024 with dimensions), ceiling confidential;
  • local-hash-256: hashing of words and trigrams, no model and no outbound call, accepted in dev and test only.

2. Position in the architecture and interactions​

  • Callers: Agents (rag tool), SDK, Context Manager.
  • Execution path: Synchronous (ingestion and query), asynchronous (reindex by the worker).
  • Every interaction goes through the Control Plane API v1 (/v1) behind the API Gateway (D-14).

3. Open source component and developed part​

  • Open source: Haystack
  • Developed: ACL, provenance components, gateway integration, embedding spaces and reindex.
  • External integration through an adapter eai/integrations/<system>.py with an in-process fallback when the URL is not configured.

4. Data model​

Tables (schema eai, tenant_id RLS):

  • eai.knowledge_bases (embedding_model, dimensions, vector_generation, vector_collection)
  • eai.kb_documents
  • eai.kb_chunks (generation, dimensions checked against the vector size)
  • eai.kb_reindex_jobs

Common columns: id uuid, tenant_id uuid NOT NULL, created_at, updated_at; RLS policy current_setting('app.tenant_id', true)::uuid.

5. Contract​

  • Applied contract: context-contract
  • JSON Schema: contracts/v1/context-contract.schema.json
  • Key fields: segments[].source, segments[].provenance
  • Validation: POST /v1/contracts/{name}/validate; manifest apply: POST /v1/manifests/apply.

6. API endpoints​

  • GET /v1/knowledge-bases
  • POST /v1/knowledge-bases
  • GET /v1/knowledge-bases/{id}
  • DELETE /v1/knowledge-bases/{id}
  • POST /v1/knowledge-bases/{id}/documents
  • GET /v1/knowledge-bases/{id}/documents
  • DELETE /v1/knowledge-bases/{id}/documents/{doc_id}
  • POST /v1/knowledge-bases/{id}/query
  • PUT /v1/knowledge-bases/{id}/acl
  • POST /v1/knowledge-bases/{id}/reindex
  • GET /v1/knowledge-bases/{id}/reindex-jobs

Request POST /v1/knowledge-bases:

{
  "name": "support",
  "classification": "confidential",
  "embedding_model": "bge-m3-vllm"
}

Request POST /v1/knowledge-bases/{id}/query:

{
  "query": "refund delay",
  "top_k": 5,
  "mode": "hybrid",
  "rerank": true,
  "rerank_model": "rerank-v3-5"
}

Response:

{
  "knowledge_base_id": "0b6f3c1e-9a52-4d7e-8c1a-2f4e6d8b0a13",
  "query": "refund delay",
  "results": [
    {
      "chunk_id": "5d2a7e90-3c14-4b8f-a6d1-7e9b2c4f6a80",
      "document_id": "a3c5e7f9-1b2d-4e6a-8c0e-2f4a6c8e0b1d",
      "title": "Refund terms",
      "ordinal": 3,
      "content": "The refund is issued within 14 days...",
      "score": 0.82,
      "vector_score": 0.79,
      "keyword_score": 0.85,
      "rerank_score": 0.91,
      "classification": "internal",
      "provenance": {
        "source_type": "rag",
        "source_id": "a3c5e7f9-1b2d-4e6a-8c0e-2f4a6c8e0b1d",
        "version": "2",
        "classification": "internal"
      }
    }
  ],
  "count": 1,
  "mode": "hybrid",
  "reranked": true,
  "warnings": [],
  "embedding_model": "bge-m3-vllm"
}

Request POST /v1/knowledge-bases/{id}/reindex (202 response with the job job_id, status, from, to, documents_total, run by the worker job embeddings_reindex):

{
  "embedding_model": "text-embedding-3-small",
  "dimensions": 512
}

7. SDK examples​

Python:

from enterprise_ai import EnterpriseAI

client = EnterpriseAI(base_url="<control-plane-url>", api_key="eai_xxx_secret")
kb = client.rag.create(name="support", chunk_size=800, embedding_model="bge-m3-vllm")
res = client.rag.query(kb.id, query="refund delay", top_k=5, mode="hybrid")

TypeScript:

import { EnterpriseAI } from "@enterprise-ai/sdk";

const client = new EnterpriseAI({ baseUrl: "<control-plane-url>", apiKey: process.env.EAI_API_KEY });
const res = await client.knowledgeBases.query("kb_support", { query: "refund delay", top_k: 5 });

Java:

EnterpriseAIClient client = EnterpriseAIClient.builder()
    .baseUrl("<control-plane-url>")
    .apiKey(System.getenv("EAI_API_KEY"))
    .build();
JsonNode res = client.knowledge().query("kb_support", Map.of("query", "refund delay", "top_k", 5));

Go:

c, _ := eai.NewClient(eai.WithBaseURL("<control-plane-url>"), eai.WithAPIKey(os.Getenv("EAI_API_KEY")))
res, _ := c.Knowledge.Query(ctx, "kb_support", map[string]any{"query": "refund delay", "top_k": 5})

8. Configuration​

  • Environment variables: EAI_RAG_EMBEDDING_MODEL, EAI_MEMORY_EMBEDDING_MODEL, EAI_RAG_RERANK_MODEL, EAI_EMBEDDINGS_BATCH_SIZE, EAI_WORKER_EMBEDDINGS_REINDEX_INTERVAL_SECONDS, EAI_QDRANT_URL, EAI_DATABASE_URL
  • Helm values keys: qdrant.enabled
  • Local model: entry bge-m3 of vllm.models (option --task embedding) and entry bge-m3-vllm of litellm.config.model_list (mode: embedding) in values-onprem-gpu.yaml

9. Security controls​

  • Scope rag:query; ACL applied before return, never after generation
  • Classification > clearance -> chunk excluded
  • Chunks sent to the embedding model with the classification of their document: above the model ceiling, 403 classification_exceeded before any outbound call
  • Provider key injected server side by the gateway, never read by the brick
  • Every request carries the tenant resolved from the credential; PostgreSQL RLS
  • Every mutation and policy decision writes a hash-chained audit event

10. Observability​

  • Prometheus metrics: eai_rag_queries_total{outcome}, eai_rag_retrieval_latency_seconds, eai_rag_embeddings_total{operation}, eai_vector_search_latency_seconds{store}
  • Traces: OpenTelemetry spans with GenAI semantic conventions (gen_ai.operation.name, gen_ai.agent.id, gen_ai.tool.name), attributes eai.tenant_id, eai.run_id, traceparent propagation.
  • Audit actions: knowledge.query, knowledge.document.ingest, knowledge_base.reindex.request, knowledge_base.reindex.complete, model.invoke (every embedding call)

11. FinOps units​

  • Embeddings: one Usage Event embeddings per gateway call, with the provider, the region and the measured tokens of the knowledge base model (labels.operation: ingest, query, reindex).
  • Queries: Usage Event rag (retrieval_queries); reranking: Usage Event reranking.
  • Emitted as Usage Event (POST /v1/finops/usage) with tenant, project, agent and run.

12. Failure modes and resilience​

  • Reranker down, refused or not configured: results in retrieval order, reranked: false and warning reranker_unavailable, rerank_refused or reranker_not_configured
  • Provider vectors of another size: 502 embedding_dimension_mismatch; vector store down: 503 vector_store_unavailable
  • Failed reindex: the previous generation stays the one searched; transient error (429, 502, 503, 504) retried up to 3 times
  • Normalized errors (error envelope, message_i18n en/fr)

13. Standalone adoption (client of a client)​

  • enterprise_ai.rag + Control Plane + Qdrant.
  • SDK: enterprise_ai.privacy / dlp / knowledge_bases / memory / retention / lineage
  • Minimal deployment: Helm chart with controlPlane.enabled=true and PostgreSQL; disabled external components use their in-process fallback.

14. Acceptance tests​

  1. User without ACL receives no chunk of the document
  2. Document deletion removes its vectors
  3. Cross-tenant access refused
  4. Error returned in fr with Accept-Language: fr
  5. Secret document refused by an embedding model capped at confidential, without outbound call
  6. Two embedding models of one tenant never mix their vectors, reindex included

Governed RAG (kit L4)​

Governed RAG: Haystack splitting (haystack-ai, extra rag) when the library is installed, document and chunk ACL (chunk_readers) filtered in the vector store then again before the context, source connector check (source_connector) at ingestion and query, quarantine of suspect documents (injection, hidden text, abnormal instructions, outlier embeddings) reviewed by the data steward (GET /v1/knowledge-bases/{id}/quarantine, POST .../documents/{doc_id}/review), citations with document version, source URI and chunk position, tenant scoped encrypted sensitive cache. Tests: tests/test_rag_governance.py.