Haystack pipelines with self-hosted DeepSeek-V3

Self-hosting DeepSeek-V3 gives you full control over data privacy, latency, and cost — but wiring it into a retrieval-augmented generation pipeline requires more than swapping a model name. This tutorial walks through a complete Haystack pipeline that indexes your documents, retrieves relevant context, and generates answers with a locally served DeepSeek-V3. You’ll end up with runnable code, a clear deployment topology, and the observability hooks you need to run this in production.

Prerequisites

Before starting, ensure you have:

• A machine with at least 48 GB VRAM (2× A100 80 GB or 4× A100 40 GB recommended for FP8) or 96 GB+ system RAM for quantized inference
• Docker and Docker Compose installed
• Python 3.10+ with uv or pip
• Familiarity with Haystack 2.x concepts: components, pipelines, and document stores

We’ll use vLLM for high-throughput serving and Ollama as a lighter alternative. Pick one path — both work with the same Haystack code.

Serving DeepSeek-V3 locally

Option A: vLLM (recommended for throughput)

vLLM’s PagedAttention and continuous batching make it the default choice for self-hosted LLM serving. DeepSeek-V3’s MoE architecture benefits significantly from vLLM’s optimized kernels.

Create docker-compose.vllm.yml:

version: "3.10"
services:
  vllm:
    image: vllm/vllm-openai:v0.6.3
    runtime: nvidia
    environment:
      - HF_TOKEN=${HF_TOKEN}
    ports:
      - "8000:8000"
    ipc: host
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 2
              capabilities: [gpu]
    command: >
      --model deepseek-ai/DeepSeek-V3
      --tensor-parallel-size 2
      --dtype auto
      --max-model-len 32768
      --gpu-memory-utilization 0.9
      --enable-prefix-caching
      --api-key ${VLLM_API_KEY:-changeme}

Start it:

export HF_TOKEN=your_huggingface_token
export VLLM_API_KEY=sk-local-123
docker compose -f docker-compose.vllm.yml up -d

Verify the OpenAI-compatible endpoint:

curl -H "Authorization: Bearer sk-local-123" \
  http://localhost:8000/v1/models

Expected output:

{
  "object": "list",
  "data": [
    {
      "id": "deepseek-ai/DeepSeek-V3",
      "object": "model",
      "owned_by": "vllm"
    }
  ]
}

Option B: Ollama (simpler, lower throughput)

Ollama packages model weights and a server in one binary. Use it when you need zero-config deployment or run on CPU/MPS.

# On the host (Linux/macOS)
curl -fsSL https://ollama.com/install.sh | sh
ollama pull deepseek-v3:671b-fp8  # or deepseek-v3:671b-q4_k_m for 4-bit
ollama serve

The API listens on http://localhost:11434/v1 — also OpenAI-compatible.

Haystack pipeline components

Install dependencies:

uv pip install haystack-ai==2.7.0 \
  sentence-transformers==3.0.1 \
  weaviate-client==4.8.0 \
  python-dotenv==1.0.1

We’ll use Weaviate as the vector store — it runs locally, supports hybrid search, and scales. Create docker-compose.weaviate.yml:

version: "3.10"
services:
  weaviate:
    image: cr.weaviate.io/semitechnologies/weaviate:1.26.0
    ports:
      - "8080:8080"
      - "50051:50051"
    environment:
      QUERY_DEFAULTS_LIMIT: 25
      AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED: "true"
      PERSISTENCE_DATA_PATH: /var/lib/weaviate
      DEFAULT_VECTORIZER_MODULE: none
      ENABLE_MODULES: ""
    volumes:
      - weaviate_data:/var/lib/weaviate

volumes:
  weaviate_data:

Start it:

docker compose -f docker-compose.weaviate.yml up -d

Wait for readiness:

curl -s http://localhost:8080/v1/.well-known/ready
# Returns 200 when ready

Building the indexing pipeline

Create indexing_pipeline.py:

import os
from pathlib import Path
from haystack import Pipeline, Document
from haystack.components.writers import DocumentWriter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.components.preprocessors import DocumentSplitter
from haystack.components.converters import PyPDFToDocument, TextFileToDocument
from haystack.components.routers import FileTypeRouter
from haystack.document_stores.types import DuplicatePolicy
from haystack_integrations.document_stores.weaviate import WeaviateDocumentStore

# Configuration
WEAVIATE_URL = os.getenv("WEAVIATE_URL", "http://localhost:8080")
EMBED_MODEL = "sentence-transformers/all-mpnet-base-v2"
DOCS_DIR = Path("./data/docs")
INDEX_NAME = "DeepSeekRAG"

def build_indexing_pipeline() -> Pipeline:
    document_store = WeaviateDocumentStore(
        url=WEAVIATE_URL,
        index=INDEX_NAME,
        embedding_dim=768,  # all-mpnet-base-v2 dimension
        similarity="cosine",
    )

    pipeline = Pipeline()
    pipeline.add_component("file_type_router", FileTypeRouter(mime_types=["text/plain", "application/pdf"]))
    pipeline.add_component("text_converter", TextFileToDocument())
    pipeline.add_component("pdf_converter", PyPDFToDocument())
    pipeline.add_component("splitter", DocumentSplitter(
        split_by="sentence",
        split_length=10,
        split_overlap=2,
        split_threshold=5,
    ))
    pipeline.add_component("embedder", SentenceTransformersDocumentEmbedder(model=EMBED_MODEL))
    pipeline.add_component("writer", DocumentWriter(
        document_store=document_store,
        policy=DuplicatePolicy.OVERWRITE,
    ))

    pipeline.connect("file_type_router.text/plain", "text_converter.sources")
    pipeline.connect("file_type_router.application/pdf", "pdf_converter.sources")
    pipeline.connect("text_converter.documents", "splitter.documents")
    pipeline.connect("pdf_converter.documents", "splitter.documents")
    pipeline.connect("splitter.documents", "embedder.documents")
    pipeline.connect("embedder.documents", "writer.documents")

    return pipeline

def run_indexing(pipeline: Pipeline, docs_dir: Path):
    files = list(docs_dir.rglob("*.txt")) + list(docs_dir.rglob("*.pdf"))
    if not files:
        print(f"No .txt or .pdf files found in {docs_dir}")
        return

    print(f"Indexing {len(files)} files...")
    result = pipeline.run({"file_type_router": {"sources": files}})
    written = result["writer"]["documents_written"]
    print(f"Indexed {written} document chunks")

if __name__ == "__main__":
    pipe = build_indexing_pipeline()
    run_indexing(pipe, DOCS_DIR)

Create a test document and run:

mkdir -p data/docs
cat > data/docs/company_policy.txt << 'EOF'
Our remote work policy allows employees to work from home up to 3 days per week.
Team leads must approve schedules in advance. Core hours are 10am-3pm UTC.
VPN access requires MFA enrollment. Equipment stipend: $500/year.
EOF

python indexing_pipeline.py

Expected output:

Indexing 1 files...
Indexed 12 document chunks

Verify in Weaviate:

curl -s "http://localhost:8080/v1/objects?class=DeepSeekRAG&limit=3" | jq '.objects[].properties'

Building the query pipeline

Now the RAG pipeline: retrieve, prompt, generate. Create rag_pipeline.py:

import os
from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.generators import OpenAIGenerator
from haystack.components.retrievers import WeaviateEmbeddingRetriever
from haystack_integrations.document_stores.weaviate import WeaviateDocumentStore

# Configuration
WEAVIATE_URL = os.getenv("WEAVIATE_URL", "http://localhost:8080")
INDEX_NAME = "DeepSeekRAG"
VLLM_BASE_URL = os.getenv("VLLM_BASE_URL", "http://localhost:8000/v1")
VLLM_API_KEY = os.getenv("VLLM_API_KEY", "sk-local-123")
EMBED_MODEL = "sentence-transformers/all-mpnet-base-v2"
GENERATION_MODEL = "deepseek-ai/DeepSeek-V3"

PROMPT_TEMPLATE = """
You are a helpful assistant answering questions using only the provided context.
If the context doesn't contain the answer, say you don't know.

Context:
{% for doc in documents %}
  {{ doc.content }}
{% endfor %}

Question: {{ question }}
Answer:
"""

def build_rag_pipeline() -> Pipeline:
    document_store = WeaviateDocumentStore(
        url=WEAVIATE_URL,
        index=INDEX_NAME,
        embedding_dim=768,
        similarity="cosine",
    )

    pipeline = Pipeline()
    pipeline.add_component("text_embedder", SentenceTransformersTextEmbedder(model=EMBED_MODEL))
    pipeline.add_component("retriever", WeaviateEmbeddingRetriever(
        document_store=document_store,
        top_k=5,
    ))
    pipeline.add_component("prompt_builder", PromptBuilder(template=PROMPT_TEMPLATE))
    pipeline.add_component("llm", OpenAIGenerator(
        api_key=VLLM_API_KEY,
        api_base_url=VLLM_BASE_URL,
        model=GENERATION_MODEL,
        generation_kwargs={
            "max_tokens": 512,
            "temperature": 0.1,
            "top_p": 0.95,
        },
    ))

    pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
    pipeline.connect("retriever.documents", "prompt_builder.documents")
    pipeline.connect("prompt_builder.prompt", "llm.prompt")

    return pipeline

def ask(question: str) -> str:
    pipe = build_rag_pipeline()
    result = pipe.run({
        "text_embedder": {"text": question},
        "prompt_builder": {"question": question},
    })
    return result["llm"]["replies"][0]

if __name__ == "__main__":
    import sys
    question = " ".join(sys.argv[1:]) or "What is the remote work policy?"
    print(f"Q: {question}\n")
    answer = ask(question)
    print(f"A: {answer}")

Run a query:

python rag_pipeline.py "How many remote days are allowed?"

Expected output:

Q: How many remote days are allowed?

A: Employees are allowed to work from home up to 3 days per week.

Adding hybrid search for better retrieval

Pure vector search misses exact matches (acronyms, IDs, proper nouns). Weaviate’s hybrid search combines BM25 and vector scores. Update the retriever in rag_pipeline.py:

from haystack.components.retrievers import WeaviateHybridRetriever

# Replace WeaviateEmbeddingRetriever with:
pipeline.add_component("retriever", WeaviateHybridRetriever(
    document_store=document_store,
    top_k=5,
    alpha=0.5,  # 0 = pure BM25, 1 = pure vector
))

The pipeline connections stay identical. Hybrid search is especially valuable for technical documentation where model numbers and error codes matter.

Streaming responses

For chat interfaces, stream tokens instead of waiting for the full response. Modify the generator:

pipeline.add_component("llm", OpenAIGenerator(
    api_key=VLLM_API_KEY,
    api_base_url=VLLM_BASE_URL,
    model=GENERATION_MODEL,
    generation_kwargs={
        "max_tokens": 512,
        "temperature": 0.1,
        "top_p": 0.95,
        "stream": True,
    },
    streaming_callback=lambda chunk: print(chunk, end="", flush=True),
))

The streaming_callback receives each token as it arrives. In a FastAPI endpoint, you’d yield SSE events instead of printing.

Production hardening

Health checks and timeouts

Wrap the generator with a component that enforces deadlines:

from haystack import component
from haystack.dataclasses import StreamingChunk
from typing import Any, Callable, Optional
import httpx

@component
class TimedOpenAIGenerator(OpenAIGenerator):
    def __init__(self, timeout: float = 30.0, **kwargs):
        super().__init__(**kwargs)
        self.timeout = timeout

    @component.output_types(replies=list[str], meta=list[dict])
    def run(self, prompt: str, generation_kwargs: Optional[dict] = None):
        import asyncio
        async def _run():
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                # ... delegate to parent with custom client
                pass
        return asyncio.run(_run())

Better: put a reverse proxy (nginx, Traefik) in front of vLLM with proxy_read_timeout 60s and let the infrastructure handle it.

Structured logging

Haystack pipelines emit spans via OpenTelemetry. Configure once at startup:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
)

Now every component execution appears in your tracing backend (Jaeger, Tempo, Datadog) with input/output sizes, latency, and errors.

Fallback routing

If your self-hosted model goes down, route to a cloud provider automatically. This is where a gateway like n4n.ai fits — one OpenAI-compatible endpoint that addresses 240+ models with automatic fallback when a provider is rate-limited or degraded, per-token usage metering, and it honors client routing directives while forwarding provider cache-control hints. Configure your OpenAIGenerator to point at the gateway URL instead of vLLM directly, and the fallback logic lives in infrastructure, not application code.

Evaluation harness

Don’t ship without measuring. Create evaluate.py:

from haystack import Pipeline
from haystack.components.evaluators import FaithfulnessEvaluator, SASEvaluator
from haystack_integrations.document_stores.weaviate import WeaviateDocumentStore

EVAL_QUESTIONS = [
    ("What is the remote work policy?", "Employees can work from home up to 3 days per week."),
    ("What are core hours?", "Core hours are 10am-3pm UTC."),
    ("What is the equipment stipend?", "$500 per year."),
]

def run_evaluation():
    pipe = build_rag_pipeline()
    faithfulness = FaithfulnessEvaluator()
    sas = SASEvaluator()

    for question, ground_truth in EVAL_QUESTIONS:
        result = pipe.run({
            "text_embedder": {"text": question},
            "prompt_builder": {"question": question},
        })
        prediction = result["llm"]["replies"][0]
        retrieved_docs = [d.content for d in result["retriever"]["documents"]]

        faith_result = faithfulness.run(
            questions=[question],
            contexts=[retrieved_docs],
            predicted_answers=[prediction],
        )
        sas_result = sas.run(
            predicted_answers=[prediction],
            ground_truth_answers=[ground_truth],
        )

        print(f"Q: {question}")
        print(f"  Faithfulness: {faith_result['individual_scores'][0]:.2f}")
        print(f"  SAS: {sas_result['individual_scores'][0]:.2f}")
        print(f"  Answer: {prediction[:100]}...")
        print()

if __name__ == "__main__":
    run_evaluation()

Run it:

python evaluate.py

Expected output:

Q: What is the remote work policy?
  Faithfulness: 1.00
  SAS: 0.92
  Answer: Employees are allowed to work from home up to 3 days per week.

Q: What are core hours?
  Faithfulness: 1.00
  SAS: 0.88
  Answer: Core hours are 10am-3pm UTC.

Q: What is the equipment stipend?
  Faithfulness: 1.00
  SAS: 0.95
  Answer: The equipment stipend is $500 per year.

Faithfulness near 1.0 means the model only uses retrieved context. SAS (Semantic Answer Similarity) near 1.0 means the answer matches the ground truth semantically.

Scaling considerations

• Multiple GPUs: Increase --tensor-parallel-size in vLLM to match GPU count. For 4× A100 40 GB, use --tensor-parallel-size 4 --pipeline-parallel-size 1.
• Quantization: FP8 (default for DeepSeek-V3 on vLLM) cuts VRAM in half with minimal quality loss. For tighter budgets, --quantization awq --dtype int4 works but degrades reasoning.
• KV cache: --enable-prefix-caching reuses KV cache across requests with shared prefixes — huge win for RAG where the system prompt and retrieved context often overlap.
• Concurrent requests: vLLM handles 100+ concurrent requests on 2× A100. Tune --max-num-batched-tokens and --max-num-seqs for your latency budget.

Common failure modes

What’s next

You now have a working self-hosted RAG pipeline. From here:

1. Add reranking: Insert a SentenceTransformersRanker (cross-encoder) between retriever and prompt builder for precision gains.
2. Persist conversations: Store chat history in PostgreSQL, feed last N turns into the prompt as few-shot examples.
3. Guardrails: Add a PIIFilter component before the LLM, and a FaithfulnessChecker after — reject or flag low-faithfulness responses.
4. Multi-tenancy: Use Weaviate’s tenant isolation or separate indexes per customer with shared embedding model.

The code in this tutorial is deliberately minimal. Each component swaps independently — change the embedder, the retriever, the generator, or the document store without rewriting the pipeline topology. That’s the point of Haystack.