Eine RAG-Pipeline (Retrieval Augmented Generation) nimmt deine eigenen Dokumente, findet relevante Teile und lässt ein LLM darauf antworten. Das funktioniert auch vollständig lokal — ohne Cloud-Abhängigkeit.
Wir bauen hier ein Produktionssystem mit echten Working Examples.
Was brauchst du?
- Python 3.10+
- 8 GB RAM (min.), 16 GB empfohlen
- GPU optional aber empfohlen (NVIDIA mit CUDA)
- ~2 GB Disk für Modelle
Teil 1: Setup & Installation
Virtuelle Umgebung erstellen
python -m venv rag-env
source rag-env/bin/activate # Windows: rag-env\Scripts\activate
pip install --upgrade pip
Abhängigkeiten installieren
pip install \
langchain==0.2.6 \
langchain-community==0.0.28 \
chromadb==0.4.28 \
ollama==0.1.12 \
pypdf==4.1.0 \
python-dotenv==1.0.0
Warum diese Versionen? Wir verwenden stabile, bewährte Versionen. Neuere Versionen können Breaking Changes haben.
Teil 2: Ollama einrichten
Ollama lädt Modelle lokal — ohne API-Kosten.
Installation
Geh zu https://ollama.ai und installiere Ollama für dein OS.
Modell laden
ollama pull llama2:7b-chat-q4_0
Das q4_0 bedeutet 4-Bit Quantisierung = 5 GB Download, 4 GB RAM im Speicher.
Test: Ollama läuft?
curl http://localhost:11434/api/generate -d '{
"model": "llama2:7b-chat-q4_0",
"prompt": "Wer war Albert Einstein?",
"stream": false
}' | python -m json.tool
Wenn JSON mit response zurückkommt → OK.
Teil 3: ChromaDB initialisieren
ChromaDB speichert Embeddings mit Metadaten.
Persistente Datenbank erstellen
# setup_chromadb.py
import chromadb
# Lokale persistente Kollektion
client = chromadb.PersistentClient(path="./chroma_data")
# Kollektion erstellen (oder laden wenn schon vorhanden)
collection = client.get_or_create_collection(
name="dokumente",
metadata={"hnsw:space": "cosine"}
)
print(f"Collection '{collection.name}' erstellt/geladen")
print(f"Dokumente im Index: {collection.count()}")
Laufen:
python setup_chromadb.py
Output: Dokumente im Index: 0 (noch leer).
Teil 4: Dokumente laden & chunken
RAG braucht gute Chunks: nicht zu klein (verliert Kontext), nicht zu gross (verliert Relevanz).
PDF-Datei laden
# load_documents.py
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import chromadb
# PDF laden
loader = PyPDFLoader("./sample.pdf")
documents = loader.load()
print(f"PDF geladen: {len(documents)} Seiten")
# In Chunks aufteilen
splitter = RecursiveCharacterTextSplitter(
chunk_size=800, # Zeichen pro Chunk
chunk_overlap=100, # Überlappung (für Kontext)
separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(documents)
print(f"Chunks erstellt: {len(chunks)}")
# Sample-Chunk anschauen
print("\n--- Chunk 0 ---")
print(chunks[0].page_content[:200])
Chunk-Tuning:
chunk_size=800: Gut für deutsche Dokumentationchunk_overlap=100: Überlappung hilft bei Grenzbereichenseparators: Zuerst auf Absätze, dann Sätze splitten
Test-PDF erstellen
Keine PDF? Hier ein Test:
# create_test_pdf.py
from reportlab.lib.pagesizes import letter
from reportlab.pdfgen import canvas
def create_sample_pdf():
c = canvas.Canvas("./sample.pdf", pagesize=letter)
texts = [
"Künstliche Intelligenz und ihre Auswirkungen",
"Machine Learning Modelle benötigen große Datenmengen.",
"RAG-Systeme kombinieren Retrieval mit Generierung.",
"Die Zukunft liegt in hybriden Ansätzen.",
]
y = 750
for text in texts:
c.drawString(50, y, text)
y -= 50
c.save()
print("sample.pdf erstellt")
create_sample_pdf()
Laufen:
pip install reportlab
python create_test_pdf.py
python load_documents.py
Teil 5: Embeddings erzeugen
Ein Embedding ist ein numerischer "Fingerabdruck" eines Textes. Ähnliche Texte haben ähnliche Embeddings.
Mit Ollama-Embedding
# embed_documents.py
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import OllamaEmbeddings
import chromadb
# Modell laden (falls noch nicht getan)
import subprocess
subprocess.run(["ollama", "pull", "nomic-embed-text"], check=True)
# Loader & Splitter
loader = PyPDFLoader("./sample.pdf")
documents = loader.load()
splitter = RecursiveCharacterTextSplitter(
chunk_size=800,
chunk_overlap=100,
)
chunks = splitter.split_documents(documents)
# Embeddings erzeugen (lokal via Ollama)
embeddings = OllamaEmbeddings(model="nomic-embed-text")
# ChromaDB client
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection(
name="dokumente",
metadata={"hnsw:space": "cosine"}
)
# Chunks in ChromaDB einträgen
for i, chunk in enumerate(chunks):
embedding_vector = embeddings.embed_query(chunk.page_content)
collection.add(
ids=[f"chunk_{i}"],
documents=[chunk.page_content],
metadatas=[{
"source": chunk.metadata.get("source", "unknown"),
"page": chunk.metadata.get("page", 0)
}],
embeddings=[embedding_vector] # Optional: manuell speichern
)
print(f"Chunk {i} embedded")
print(f"\nGesamt {collection.count()} Chunks in ChromaDB")
Embedding-Modelle:
nomic-embed-text: 768-dim, gut für Deutschmxbai-embed-large: 1024-dim, noch besserall-minilm-l6-v2: Klein & schnell (22 MB)
Laufen:
python embed_documents.py
Teil 6: Retrieval — ähnliche Dokumente finden
# retrieve.py
from langchain_community.embeddings import OllamaEmbeddings
import chromadb
embeddings = OllamaEmbeddings(model="nomic-embed-text")
client = chromadb.PersistentClient(path="./chroma_data")
collection = client.get_or_create_collection(name="dokumente")
# Abfrage
query = "Was ist Machine Learning?"
query_embedding = embeddings.embed_query(query)
# Top 3 ähnliche Chunks finden
results = collection.query(
query_embeddings=[query_embedding],
n_results=3,
include=["documents", "metadatas", "distances"]
)
print("Top Retrieval Results:")
for i, (doc, meta, distance) in enumerate(zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
)):
print(f"\n--- Result {i+1} (Ähnlichkeit: {1 - distance:.2%}) ---")
print(f"Source: {meta['source']}, Page: {meta['page']}")
print(f"Text: {doc[:200]}...")
Laufen:
python retrieve.py
Teil 7: Generation — RAG vollständig
Jetzt kombinieren wir Retrieval + LLM:
# rag_pipeline.py
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.llms import Ollama
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain_community.vectorstores import Chroma
import chromadb
# ChromaDB als LangChain VectorStore
client = chromadb.PersistentClient(path="./chroma_data")
embeddings = OllamaEmbeddings(model="nomic-embed-text")
vectorstore = Chroma(
client=client,
collection_name="dokumente",
embedding_function=embeddings
)
# LLM initialisieren
llm = Ollama(model="llama2:7b-chat-q4_0", temperature=0.3)
# Custom Prompt
template = """Du bist ein hilfreicher Assistent.
Beantworte die Frage basierend auf dem folgenden Kontext.
Wenn du die Antwort nicht im Kontext findest, sag "Ich weiß es nicht".
Kontext:
{context}
Frage: {question}
Antwort:"""
prompt = PromptTemplate(
template=template,
input_variables=["context", "question"]
)
# RAG Chain
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff", # Einfach: alle Chunks zusammenfügen
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
chain_type_kwargs={"prompt": prompt},
return_source_documents=True
)
# Abfrage
result = qa_chain({"query": "Was ist RAG?"})
print("Antwort:")
print(result["result"])
print("\nQuellen:")
for doc in result["source_documents"]:
print(f"- {doc.metadata['source']} (S.{doc.metadata['page']})")
Laufen:
python rag_pipeline.py
Output:
Antwort:
RAG ist eine Methode, die Dokumentenretrieval mit Sprachmodellen kombiniert...
Quellen:
- ./sample.pdf (S.0)
- ./sample.pdf (S.1)
Teil 8: Production-Grade RAG mit Caching
Produktion braucht Speed. Wir cachen häufig gestellte Fragen:
# rag_production.py
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.llms import Ollama
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain_community.vectorstores import Chroma
import chromadb
import hashlib
import json
from pathlib import Path
# Cache-Datei
CACHE_FILE = "rag_cache.json"
def load_cache():
if Path(CACHE_FILE).exists():
with open(CACHE_FILE) as f:
return json.load(f)
return {}
def save_cache(cache):
with open(CACHE_FILE, "w") as f:
json.dump(cache, f)
def get_query_hash(query):
return hashlib.md5(query.lower().encode()).hexdigest()
cache = load_cache()
# ChromaDB setup
client = chromadb.PersistentClient(path="./chroma_data")
embeddings = OllamaEmbeddings(model="nomic-embed-text")
vectorstore = Chroma(
client=client,
collection_name="dokumente",
embedding_function=embeddings
)
llm = Ollama(model="llama2:7b-chat-q4_0", temperature=0.3)
template = """Du bist ein hilfreicher Assistent.
Beantworte die Frage basierend auf dem folgenden Kontext.
Kontext:
{context}
Frage: {question}
Antwort:"""
prompt = PromptTemplate(
template=template,
input_variables=["context", "question"]
)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
chain_type_kwargs={"prompt": prompt},
return_source_documents=True
)
# Abfrage mit Cache
def query_with_cache(question):
query_hash = get_query_hash(question)
# Im Cache?
if query_hash in cache:
print(f"✓ Cache hit: {question}")
return cache[query_hash]
# Nicht im Cache → generieren
print(f"✗ Cache miss, generiere Antwort...")
result = qa_chain({"query": question})
# In Cache speichern
cache[query_hash] = {
"answer": result["result"],
"sources": [
{
"source": doc.metadata["source"],
"page": doc.metadata["page"]
}
for doc in result["source_documents"]
]
}
save_cache(cache)
return cache[query_hash]
# Test
answer = query_with_cache("Was ist RAG?")
print("Antwort:", answer["answer"][:200])
# Zweimal fragen → Cache hit
answer2 = query_with_cache("Was ist RAG?")
Teil 9: Error Handling & Monitoring
Real-World Systeme brauchen Robustheit:
# rag_robust.py
import logging
from typing import Optional, Dict
from langchain_community.embeddings import OllamaEmbeddings
from langchain_community.llms import Ollama
from langchain.chains import RetrievalQA
from langchain_community.vectorstores import Chroma
import chromadb
# Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RAGPipeline:
def __init__(self, model_name="llama2:7b-chat-q4_0", embedding_model="nomic-embed-text"):
try:
self.embeddings = OllamaEmbeddings(model=embedding_model)
self.llm = Ollama(model=model_name, temperature=0.3)
client = chromadb.PersistentClient(path="./chroma_data")
self.vectorstore = Chroma(
client=client,
collection_name="dokumente",
embedding_function=self.embeddings
)
logger.info(f"RAG Pipeline initialisiert: {self.vectorstore._collection.count()} Chunks")
except Exception as e:
logger.error(f"Fehler bei RAG-Initialisierung: {e}")
raise
def query(self, question: str, max_retries: int = 2) -> Optional[Dict]:
"""Sichere Abfrage mit Fehlerbehandlung"""
# Validierung
if not question.strip():
logger.warning("Leere Frage")
return None
if len(question) > 1000:
logger.warning("Frage zu lang (>1000 Zeichen)")
return None
# Retry-Loop
for attempt in range(max_retries):
try:
logger.info(f"Versuch {attempt + 1}: '{question}'")
# Retrieval
retriever = self.vectorstore.as_retriever(search_kwargs={"k": 3})
relevant_docs = retriever.get_relevant_documents(question)
if not relevant_docs:
logger.warning(f"Keine relevanten Dokumente gefunden")
return {
"answer": "Keine relevanten Dokumente gefunden.",
"sources": [],
"confidence": 0.0
}
# Generation
context = "\n\n".join([doc.page_content for doc in relevant_docs])
prompt = f"Frage: {question}\n\nKontext:\n{context}\n\nAntwort:"
answer = self.llm(prompt)
return {
"answer": answer,
"sources": [doc.metadata.get("source") for doc in relevant_docs],
"confidence": 0.85 # Vereinfacht
}
except Exception as e:
logger.error(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
if attempt == max_retries - 1:
return {
"answer": "Fehler bei der Verarbeitung. Versuchen Sie es später.",
"sources": [],
"confidence": 0.0
}
# Test
rag = RAGPipeline()
result = rag.query("Was ist Künstliche Intelligenz?")
print(result)
Teil 10: Performance-Tuning
Embedding-Caching
Embeddings sind teuer. Zwischenspeichern:
# caching_embeddings.py
from langchain_community.embeddings import OllamaEmbeddings
import json
from pathlib import Path
class CachedOllamaEmbeddings(OllamaEmbeddings):
def __init__(self, *args, cache_file="embedding_cache.json", **kwargs):
super().__init__(*args, **kwargs)
self.cache_file = cache_file
self.cache = self._load_cache()
def _load_cache(self):
if Path(self.cache_file).exists():
with open(self.cache_file) as f:
return json.load(f)
return {}
def _save_cache(self):
with open(self.cache_file, "w") as f:
json.dump(self.cache, f)
def embed_query(self, text: str):
if text in self.cache:
return self.cache[text]
embedding = super().embed_query(text)
self.cache[text] = embedding
self._save_cache()
return embedding
# Nutzen
embeddings = CachedOllamaEmbeddings(model="nomic-embed-text")
Retriever k-Tuning
# Zu viele Chunks → Noise
# Zu wenige → Fehlende Info
# Test verschiedene k-Werte
for k in [1, 3, 5, 10]:
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
docs = retriever.get_relevant_documents("Meine Abfrage")
print(f"k={k}: {len(docs)} Dokumente, total {sum(len(d.page_content) for d in docs)} Zeichen")
# Zu viel Text → LLM wird langsam
# Zu wenig Text → LLM hat nicht genug Kontext
Fehlerbehebung
Problem: "Connection refused" bei Ollama
# Ollama läuft nicht
ollama serve
# Oder im anderen Terminal testen
curl http://localhost:11434/api/tags
Problem: ChromaDB ist leer
import chromadb
client = chromadb.PersistentClient(path="./chroma_data")
collections = client.list_collections()
print(collections)
# Falls leer: Documents nochmal laden & embedden
Problem: LLM-Antwort ist Unsinn
Ursachen:
- Falsche Chunks retrieved → prüfe
chunk_size/chunk_overlap - Prompt zu komplex → vereinfachen
- Modell zu klein →
llama2:7b→llama2:13bprobieren - Temperature zu hoch → auf 0.1–0.3 setzen
Zusammenfassung
Eine RAG-Pipeline hat 5 Teile:
- Dokumente laden → PyPDFLoader, RecursiveCharacterTextSplitter
- Embeddings erzeugen → OllamaEmbeddings lokal
- Index speichern → ChromaDB persistent
- Retrieval → Top-k ähnliche Chunks
- Generation → LLM antwortet mit Kontext
Das funktioniert vollständig offline. Kein API-Limit, keine Kosten pro Abfrage.
Nächste Schritte:
- RAG mit mehreren Datenquellen (CSV, Webseiten, etc.)
- Reranking für bessere Relevanz
- Zusammenfassungs-Chains für lange Antworten