Build production-ready RAG systems using DeepSeek models with Ollama for local inference.

RAG with DeepSeek + Ollama

Overview

Build a production-ready RAG system using DeepSeek's powerful models with Ollama for local inference. This tutorial covers the complete pipeline from data ingestion to semantic search.

Prerequisites

Python 3.10+
Ollama installed locally
DeepSeek API key (optional for cloud fallback)
~8GB RAM for local models

Installation

# Install Ollama
# macOS
brew install ollama

# Linux
curl -fsSL https://ollama.com/install.sh | sh

# Windows: Download from https://ollama.com

# Pull required models
ollama pull deepseek-coder:33b
ollama pull nomic-embed-text:latest

# Install Python dependencies
pip install langchain langchain-community sentence-transformers chromadb

Architecture Overview

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Documents │────▶│  Embedding   │────▶│   Vector    │
│   (PDF,MD)  │     │   Model      │     │  Database   │
└─────────────┘     └──────────────┘     └─────────────┘
                                                  │
                                                  ▼
┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Response  │◀────│  DeepSeek    │◀────│  Retrieval  │
│   (Claude)  │     │  / Ollama    │     │   Query     │
└─────────────┘     └──────────────┘     └─────────────┘

Step 1: Data Ingestion

Document Loading

# src/ingestion.py
from langchain_community.document_loaders import (
    PyPDFLoader,
    DirectoryLoader,
    TextLoader,
    MarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_documents(directory: str) -> list:
    """Load documents from directory."""
    loaders = [
        DirectoryLoader(directory, glob="**/*.pdf", loader_cls=PyPDFLoader),
        DirectoryLoader(directory, glob="**/*.md", loader_cls=MarkdownLoader),
        DirectoryLoader(directory, glob="**/*.txt", loader_cls=TextLoader),
    ]
    
    documents = []
    for loader in loaders:
        documents.extend(loader.load())
    
    return documents

def split_documents(documents: list, chunk_size: int = 1000, chunk_overlap: int = 200) -> list:
    """Split documents into chunks."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ".", " ", ""]
    )
    return splitter.split_documents(documents)

Metadata Extraction

from pathlib import Path

def extract_metadata(doc) -> dict:
    """Extract metadata from document."""
    return {
        'source': str(doc.metadata.get('source', '')),
        'page': doc.metadata.get('page', 0),
        'file_type': Path(doc.metadata.get('source', '')).suffix,
        'chunk_index': 0,
    }

Step 2: Embedding Generation

Using Nomic Embed (Local)

# src/embeddings.py
from langchain_community.embeddings import OllamaEmbeddings

def get_local_embeddings() -> OllamaEmbeddings:
    """Get local embeddings using Ollama."""
    return OllamaEmbeddings(
        model="nomic-embed-text",
        base_url="http://localhost:11434"
    )

# Generate embeddings
embeddings = get_local_embeddings()
texts = ["Document chunk 1", "Document chunk 2"]
embeddings_list = embeddings.embed_documents(texts)

Using DeepSeek Embeddings (Cloud)

# For higher quality embeddings
from langchain_openai import OpenAIEmbeddings

def get_deepseek_embeddings(api_key: str) -> OpenAIEmbeddings:
    """Get embeddings using DeepSeek API."""
    return OpenAIEmbeddings(
        model="deepseek-embed",
        api_key=api_key,
        base_url="https://api.deepseek.com/v1"
    )

Step 3: Vector Database Setup

ChromaDB Configuration

# src/vectorstore.py
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OllamaEmbeddings

def create_vectorstore(
    documents: list,
    embeddings: OllamaEmbeddings,
    persist_directory: str = "./chroma_db"
) -> Chroma:
    """Create and persist vector store."""
    vectorstore = Chroma.from_documents(
        documents=documents,
        embedding=embeddings,
        persist_directory=persist_directory
    )
    return vectorstore

def load_vectorstore(
    persist_directory: str,
    embeddings: OllamaEmbeddings
) -> Chroma:
    """Load existing vector store."""
    return Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings
    )

Hybrid Search Setup

def enable_hybrid_search(vectorstore: Chroma):
    """Configure hybrid search (keyword + semantic)."""
    # Chroma supports hybrid search via BM25
    from langchain_community.retrievers import BM25Retriever
    
    # Create keyword retriever
    texts = [doc.page_content for doc in vectorstore.similarity_search("", k=1000)]
    bm25_retriever = BM25Retriever.from_texts(texts)
    bm25_retriever.k = 10
    
    return bm25_retriever

Step 4: Retrieval Pipeline

Basic Retrieval

# src/retriever.py
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker

def create_retriever(vectorstore, top_k: int = 5) -> ContextualCompressionRetriever:
    """Create retrieval pipeline with reranking."""
    
    # Base retriever
    base_retriever = vectorstore.as_retriever(
        search_type="similarity",
        search_kwargs={"k": top_k * 2}  # Get more for reranking
    )
    
    # Reranker using local model
    reranker = CrossEncoderReranker(
        model="cross-encoder/ms-marco-MiniLM-L-6-v2",
        top_n=top_k
    )
    
    # Create compression retriever
    retriever = ContextualCompressionRetriever(
        base_compressor=reranker,
        base_retriever=base_retriever
    )
    
    return retriever

Multi-Query Retrieval

from langchain.retrievers import MultiQueryRetriever

def create_multi_query_retriever(vectorstore, llm) -> MultiQueryRetriever:
    """Generate multiple queries for better coverage."""
    
    retriever = MultiQueryRetriever.from_llm(
        retriever=vectorstore.as_retriever(),
        llm=llm,
        parser_key="lines"
    )
    
    return retriever

Step 5: Generation with DeepSeek

Local Generation with Ollama

# src/generator.py
from langchain_community.llms import Ollama

def get_local_llm() -> Ollama:
    """Get local LLM using Ollama."""
    return Ollama(
        model="deepseek-coder:33b",
        base_url="http://localhost:11434",
        temperature=0.7,
        num_ctx=4096
    )

def generate_response(
    llm: Ollama,
    query: str,
    context: str
) -> str:
    """Generate response with context."""
    
    prompt = f"""Based on the following context, answer the question.
    
Context:
{context}

Question: {query}

Answer:"""
    
    return llm.invoke(prompt)

Cloud Generation with DeepSeek API

from langchain_openai import ChatOpenAI

def get_deepseek_llm(api_key: str) -> ChatOpenAI:
    """Get DeepSeek cloud model."""
    return ChatOpenAI(
        model="deepseek-chat",
        api_key=api_key,
        base_url="https://api.deepseek.com/v1",
        temperature=0.7
    )

Step 6: Complete RAG Pipeline

End-to-End Implementation

# src/rag_pipeline.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class RAGPipeline:
    vectorstore: Chroma
    retriever: ContextualCompressionRetriever
    llm: Ollama
    
    def query(self, question: str, top_k: int = 3) -> dict:
        """Execute complete RAG query."""
        
        # Retrieve relevant documents
        docs = self.retriever.get_relevant_documents(question)
        
        # Format context
        context = "\n\n".join([d.page_content for d in docs])
        
        # Generate answer
        prompt = f"""You are a helpful assistant. Answer the question using only the provided context.
        
Context:
{context}

Question: {question}

Answer:"""
        
        answer = self.llm.invoke(prompt)
        
        return {
            'answer': answer,
            'sources': [d.metadata['source'] for d in docs],
            'contexts': [d.page_content for d in docs]
        }

# Usage
pipeline = RAGPipeline(
    vectorstore=load_vectorstore("./chroma_db", get_local_embeddings()),
    retriever=create_retriever(load_vectorstore("./chroma_db", get_local_embeddings())),
    llm=get_local_llm()
)

result = pipeline.query("How do I implement authentication?")
print(result['answer'])

Advanced Features

Streaming Responses

async def stream_response(pipeline: RAGPipeline, question: str):
    """Stream response for better UX."""
    
    docs = pipeline.retriever.get_relevant_documents(question)
    context = "\n\n".join([d.page_content for d in docs])
    
    prompt = f"""Context:
{context}

Question: {question}

Answer:"""
    
    async for chunk in pipeline.llm.astream(prompt):
        print(chunk, end='', flush=True)

Citation Generation

def add_citations(answer: str, docs: list) -> str:
    """Add citations to answer."""
    
    citation_map = {}
    for i, doc in enumerate(docs):
        source = doc.metadata.get('source', 'unknown')
        if source not in citation_map:
            citation_map[source] = len(citation_map) + 1
    
    # Add citations to answer
    for source, num in citation_map.items():
        answer = answer.replace(source, f"{source}[^{num}]")
    
    return answer, citation_map

Query Rewriting

def rewrite_query(query: str, history: list[str]) -> str:
    """Rewrite query based on conversation history."""
    
    if not history:
        return query
    
    prompt = f"""Given the conversation history and the latest question, 
rewrite the question to be standalone.

History:
{'; '.join(history)}

Latest Question: {query}

Standalone Question:"""
    
    return llm.invoke(prompt)

Performance Optimization

Caching

from functools import lru_cache
import hashlib

@lru_cache(maxsize=100)
def cached_query(query_hash: str, vectorstore: Chroma) -> list:
    """Cache query results."""
    return vectorstore.similarity_search(query_hash, k=5)

def get_query_hash(query: str) -> str:
    """Generate hash for query caching."""
    return hashlib.md5(query.encode()).hexdigest()

Batch Processing

async def batch_ingest(documents: list, batch_size: int = 100):
    """Ingest documents in batches."""
    
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i + batch_size]
        await vectorstore.aadd_documents(batch)
        print(f"Processed {i + len(batch)} / {len(documents)} documents")

Troubleshooting

Common Issues

Ollama not responding:

# Check Ollama is running
ollama list

# Restart if needed
ollama serve

Embedding dimension mismatch:

# Ensure embedding model matches vectorstore
embeddings = OllamaEmbeddings(model="nomic-embed-text")
# nomic-embed-text produces 768-dim vectors

Slow retrieval:

# Use HNSW index for faster search
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    collection_settings={
        "hnsw:space": "cosine",
        "hnsw:construction_ef": 128,
        "hnsw:search_ef": 64
    }
)

Resources

Ollama Docs: https://ollama.com/docs
DeepSeek API: https://platform.deepseek.com/docs
LangChain RAG: https://python.langchain.com/docs/use_cases/qa_structured/
ChromaDB: https://docs.trychroma.com/

Last updated: May 2026