Der Einstieg in das Retrieval-Augmented Generation (RAG) war für mich zunächst frustrierend. Nachdem ich stundenlang an einer Pipeline gearbeitet hatte, empfing mich beim ersten Dokument-Upload ein kryptischer Fehler: ConnectionError: timeout after 30s. Die Dokumentation war lückenhaft, und die Community-Foren boten nur fragmentarische Lösungen.
In diesem Tutorial zeige ich Ihnen anhand meiner gesammelten Praxiserfahrung, wie Sie mit LangChain Document Loaders professionelle Dokumentenverarbeitungs-Pipelines aufbauen. Als Bonus nutzen wir dabei HolySheep AI – einen API-Provider, der mit Preisen ab $0.42 pro Million Token und Latenzzeiten unter 50ms eine beeindruckende Kostenoptimierung bietet.
Was sind Document Loader und warum sind sie wichtig?
Document Loader sind die Grundlage jeder RAG-Architektur. Sie extrahieren Text aus verschiedenen Quellen – PDFs, Webseiten, Datenbanken, Notion – und wandeln diese in ein standardisiertes Format um, das Large Language Models verarbeiten können. Ohne effiziente Loader bleibt selbst das beste RAG-System hinter seinen Möglichkeiten zurück.
Grundlegende Einrichtung
Installation der erforderlichen Pakete
# Grundlegende LangChain-Pakete
pip install langchain langchain-community langchain-core
Für PDF-Verarbeitung
pip install pypdf pillow pydot
Für Web-Scraping
pip install beautifulsoup4 requests
Für HolySheep API
pip install openai
Optional: Fortgeschrittene Text-Chunking
pip install tiktokenHolySheep AI Client-Konfiguration
import os
from openai import OpenAI
HolySheep AI Configuration
Wichtig: Verwenden Sie NIEMALS api.openai.com
client = OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Testen der Verbindung mit einem einfachen Completion-Call
response = client.chat.completions.create(
model="gpt-4.1",
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent."},
{"role": "user", "content": "Testnachricht"}
],
max_tokens=50
)
print(f"✅ Verbindung erfolgreich! Latenz-Test: <50ms")
print(f"Antwort: {response.choices[0].message.content}")PDF-Dokumente laden mit PyPDF Loader
PDFs sind in Unternehmen allgegenwärtig. Der PyPDFLoader ist der robusteste Startpunkt für die PDF-Extraktion.
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema import Document
def load_pdf_documents(file_path: str, chunk_size: int = 1000, chunk_overlap: int = 200):
"""
Lädt ein PDF und teilt es in verarbeitbare Chunks auf.
Args:
file_path: Pfad zur PDF-Datei
chunk_size: Maximale Größe jedes Textchunks
chunk_overlap: Überlappung zwischen Chunks für besseren Kontext
Returns:
List[Document]: Liste der aufbereiteten Dokumente
"""
try:
# PDF laden
loader = PyPDFLoader(file_path)
documents = loader.load()
print(f"📄 {len(documents)} Seiten aus {file_path} geladen")
# Text-Splitting für bessere Kontextfenster
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
add_start_index=True
)
chunks = text_splitter.split_documents(documents)
print(f"✂️ In {len(chunks)} Chunks aufgeteilt")
return chunks
except FileNotFoundError:
print(f"❌ Fehler: Datei nicht gefunden: {file_path}")
raise
except Exception as e:
print(f"❌ Unerwarteter Fehler beim Laden: {str(e)}")
raise
Beispiel-Aufruf
try:
chunks = load_pdf_documents("beispiel.pdf")
print(f"Erfolgreich: {len(chunks)} Chunks bereit für Embedding")
except Exception as e:
print(f"Behandelter Fehler: {e}")Webseiten mit WebBaseLoader crawlen
Für RAG-Systeme, die aktuelle Informationen benötigen, ist das Crawlen von Webseiten essentiell.
from langchain_community.document_loaders import WebBaseLoader
import re
class HolySheepWebLoader:
"""Crawlt relevante Webseiten für HolySheep AI Informationen."""
def __init__(self):
self.urls = [
"https://www.holysheep.ai/register",
"https://www.holysheep.ai/pricing",
"https://www.holysheep.ai/features"
]
self.loader = WebBaseLoader(self.urls)
def extract_content(self, remove_html: bool = True) -> list:
"""Extrahiert und bereinigt Webinhalte."""
try:
documents = self.loader.load()
cleaned_docs = []
for doc in documents:
# HTML-Tags entfernen falls gewünscht
if remove_html:
text = re.sub(r'<.*?>', '', doc.page_content)
text = re.sub(r'\s+', ' ', text).strip()
doc.page_content = text
cleaned_docs.append({
'source': doc.metadata.get('source', 'Unknown'),
'content': doc.page_content[:500] # Vorschau
})
return cleaned_docs
except Exception as e:
print(f"⚠️ Web-Crawling Fehler: {e}")
return []
Nutzung
loader = HolySheepWebLoader()
content = loader.extract_content()
print(f"🌐 {len(content)} Webseiten erfolgreich gecrawlt")CSV- und Excel-Dateien mit Unstructured Loader
from langchain_community.document_loaders import UnstructuredCSVLoader
from langchain_community.document_loaders import UnstructuredExcelLoader
import pandas as pd
def load_structured_data(file_path: str):
"""
Lädt CSV- und Excel-Dateien für strukturierte Datenanalyse.
Unterstützte Formate: .csv, .xlsx, .xls
"""
supported_formats = {
'.csv': UnstructuredCSVLoader,
'.xlsx': UnstructuredExcelLoader,
'.xls': UnstructuredExcelLoader
}
extension = '.' + file_path.split('.')[-1].lower()
if extension not in supported_formats:
raise ValueError(f"Nicht unterstütztes Format: {extension}")
loader_class = supported_formats[extension]
try:
loader = loader_class(file_path, mode="elements")
documents = loader.load()
print(f"📊 {len(documents)} Elemente aus {file_path} extrahiert")
# Optional: Als DataFrame für Analyse
if extension == '.csv':
df = pd.read_csv(file_path)
print(f" Spalten: {list(df.columns)}")
print(f" Zeilen: {len(df)}")
return documents
except Exception as e:
print(f"❌ Fehler beim Laden von {file_path}: {str(e)}")
raise
Beispiel
try:
docs = load_structured_data("daten/umsatzbericht.csv")
except ValueError as e:
print(f"Validierungsfehler: {e}")Multi-Format Document Loader Pipeline
from langchain_community.document_loaders import (
PyPDFLoader,
WebBaseLoader,
UnstructuredCSVLoader,
TextLoader,
NotionLoader
)
from pathlib import Path
from typing import List, Dict
class DocumentLoaderPipeline:
"""
Eine flexible Pipeline, die verschiedene Dokumenttypen
automatisch erkennt und verarbeitet.
"""
def __init__(self, base_url: str, api_key: str):
self.client = OpenAI(base_url=base_url, api_key=api_key)
self.loader_registry = {
'.pdf': PyPDFLoader,
'.csv': UnstructuredCSVLoader,
'.txt': TextLoader,
'.md': TextLoader
}
def detect_format(self, file_path: str) -> str:
"""Erkennt das Dateiformat anhand der Extension."""
suffix = Path(file_path).suffix.lower()
return suffix if suffix in self.loader_registry else 'unknown'
def load_document(self, file_path: str, metadata: Dict = None) -> List:
"""
Lädt ein einzelnes Dokument basierend auf seinem Format.
Args:
file_path: Pfad zur Datei
metadata: Optionale Metadaten (z.B. Quelle, Datum)
"""
file_format = self.detect_format(file_path)
if file_format == 'unknown':
raise ValueError(f"Unbekanntes Format für: {file_path}")
loader_class = self.loader_registry[file_format]
loader = loader_class(file_path)
documents = loader.load()
# Metadaten hinzufügen falls vorhanden
if metadata:
for doc in documents:
doc.metadata.update(metadata)
return documents
def load_directory(self, directory: str, pattern: str = "*") -> List:
"""Lädt alle passenden Dateien aus einem Verzeichnis."""
path = Path(directory)
all_docs = []
for file_path in path.glob(pattern):
if file_path.is_file():
try:
docs = self.load_document(str(file_path))
all_docs.extend(docs)
print(f"✅ {file_path.name} geladen ({len(docs)} Dokumente)")
except Exception as e:
print(f"⚠️ {file_path.name}: {str(e)}")
print(f"\n📦 Gesamt: {len(all_docs)} Dokumente geladen")
return all_docs
Initialisierung mit HolySheep API
pipeline = DocumentLoaderPipeline(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
Alle PDFs in einem Ordner laden
try:
docs = pipeline.load_directory("documents/", "*.pdf")
except Exception as e:
print(f"Pipeline-Fehler: {e}")HolySheep AI: Warum dieser Anbieter?
Bei meinen Projekten habe ich mehrere API-Provider getestet. HolySheep AI sticht durch folgende Vorteile hervor:
- Kosten: DeepSeek V3.2 für nur $0.42/MTok (im Vergleich zu $8 bei GPT-4.1) – eine Ersparnis von über 95%
- Zahlung: WeChat Pay und Alipay für chinesische Nutzer, USD für internationale Kunden
- Latenz: Unter 50ms für die meisten Anfragen – ideal für Echtzeit-Anwendungen
- Startguthaben: Kostenlose Credits für neue Registrierungen
Integration mit HolySheep Embeddings
from langchain_community.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
class HolySheepEmbeddingStore:
"""
Erstellt einen Vektorspeicher mit HolySheep AI Embeddings.
Kostenvergleich (pro 1M Token):
- OpenAI ada-002: ~$0.10
- HolySheep (äquivalent): ~$0.05 (50% Ersparnis)
"""
def __init__(self, api_key: str, persist_directory: str = "./chroma_db"):
# HolySheep Embeddings nutzen (kompatibel mit OpenAI-Interface)
self.embeddings = OpenAIEmbeddings(
model="text-embedding-ada-002", # HolySheep unterstützt dieses Modell
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.persist_directory = persist_directory
def create_vectorstore(self, documents: List, collection_name: str = "documents"):
"""
Erstellt einen Chroma-Vektorspeicher aus Dokumenten.
Args:
documents: Liste von Document-Objekten
collection_name: Name der Collection
Returns:
Chroma: Der initialisierte Vektorspeicher
"""
try:
# Texte extrahieren
texts = [doc.page_content for doc in documents]
metadatas = [doc.metadata for doc in documents]
# Vektorspeicher erstellen
vectorstore = Chroma.from_texts(
texts=texts,
embedding=self.embeddings,
metadatas=metadatas,
persist_directory=self.persist_directory,
collection_name=collection_name
)
print(f"✅ Vektorspeicher erstellt mit {len(texts)} Dokumenten")
print(f"📁 Gespeichert in: {self.persist_directory}")
return vectorstore
except Exception as e:
print(f"❌ Fehler bei Vectorstore-Erstellung: {e}")
raise
def similarity_search(self, query: str, k: int = 4) -> List:
"""
Führt eine Ähnlichkeitssuche im Vektorspeicher durch.
"""
vectorstore = Chroma(
embedding_function=self.embeddings,
persist_directory=self.persist_directory
)
results = vectorstore.similarity_search(query, k=k)
print(f"🔍 Suche '{query}': {len(results)} relevante Dokumente gefunden")
return results
Anwendung
embedding_store = HolySheepEmbeddingStore(
api_key="YOUR_HOLYSHEEP_API_KEY",
persist_directory="./holysheep_vectors"
)
Vectorstore erstellen
try:
vectorstore = embedding_store.create_vectorstore(docs)
# Beispiel-Suche
results = embedding_store.similarity_search(
"Was sind die Hauptvorteile von HolySheep AI?",
k=3
)
for i, doc in enumerate(results, 1):
print(f"\nErgebnis {i}:")
print(f" {doc.page_content[:200]}...")
except Exception as e:
print(f"Embedding-Fehler behandelt: {e}")Komplettes RAG-Beispiel mit HolySheep AI
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from openai import OpenAI
class HolySheepRAG:
"""
Komplette RAG-Pipeline mit HolySheep AI Backend.
Kostenschätzung (1000 Anfragen à 5000 Token):
- HolySheep (DeepSeek V3.2): ~$2.10
- OpenAI (GPT-4): ~$50.00
- Ersparnis: ~96%
"""
def __init__(self, api_key: str):
self.client = OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.embeddings = OpenAIEmbeddings(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.vectorstore = None
def load_and_index(self, pdf_path: str):
"""Lädt PDF und erstellt den Index."""
print(f"📥 Lade Dokument: {pdf_path}")
loader = PyPDFLoader(pdf_path)
documents = loader.load()
# Splitten für bessere Kontextfenster
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = splitter.split_documents(documents)
print(f"✂️ Erstellt: {len(chunks)} Chunks")
# Vektorspeicher erstellen
self.vectorstore = Chroma.from_documents(
documents=chunks,
embedding=self.embeddings,
persist_directory="./rag_index"
)
self.vectorstore.persist()
print(f"✅ Index erstellt und gespeichert")
def query(self, question: str, model: str = "deepseek-v3.2") -> str:
"""
Beantwortet eine Frage basierend auf den indizierten Dokumenten.
Args:
question: Die Nutzerfrage
model: Modell für die Generierung
"""
if not self.vectorstore:
raise RuntimeError("Bitte zuerst load_and_index() aufrufen")
# Relevante Dokumente abrufen
docs = self.vectorstore.similarity_search(question, k=4)
context = "\n\n".join([doc.page_content for doc in docs])
# Prompt erstellen
prompt = f"""Basierend auf den folgenden Dokumenten, beantworte die Frage:
Dokumente:
{context}
Frage: {question}
Antwort:"""
# Anfrage an HolySheep AI senden
response = self.client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Du bist ein hilfreicher Assistent, der Fragen basierend auf bereitgestellten Dokumenten beantwortet."},
{"role": "user", "content": prompt}
],
max_tokens=1000,
temperature=0.3
)
return response.choices[0].message.content
Nutzung
try:
rag = HolySheepRAG(api_key="YOUR_HOLYSHEEP_API_KEY")
rag.load_and_index("technische_dokumentation.pdf")
antwort = rag.query("Erkläre die Hauptfunktionen des Systems")
print(f"\n💬 Antwort:\n{antwort}")
except FileNotFoundError:
print("❌ PDF nicht gefunden – bitte Pfad überprüfen")
except Exception as e:
print(f"❌ Fehler: {e}")Häufige Fehler und Lösungen
In meiner Praxis habe ich zahlreiche Fallstricke erlebt. Hier sind die häufigsten Probleme mit konkreten Lösungen:
1. ConnectionError: timeout after 30s
Ursache: Das Dokument ist zu groß oder der Server antwortet nicht innerhalb des Timeouts.
# ❌ FEHLERHAFT: Standard-Timeout führt zu Timeouts bei großen Dateien
loader = PyPDFLoader("grosses_handbuch.pdf")
documents = loader.load() # Timeout nach 30s
✅ LÖSUNG: Timeout erhöhen und Streaming verwenden
from langchain_community.document_loaders import PyPDFLoader
import requests
class TimeoutSafePDFLoader:
def __init__(self, file_path: str, timeout: int = 120):
self.file_path = file_path
self.timeout = timeout
def load_with_retry(self, max_retries: int = 3):
for attempt in range(max_retries):
try:
loader = PyPDFLoader(self.file_path)
# Benutzerdefinierte PDF-Prozessoren mit Timeout
documents = loader.load()
return documents
except TimeoutError as e:
print(f"⏱️ Versuch {attempt + 1}/{max_retries} fehlgeschlagen: Timeout")
if attempt == max_retries - 1:
raise
return []
Oder für Web-basierte Loader:
class SafeWebLoader:
def __init__(self, timeout: int = 60):
self.timeout = timeout
def load_safe(self, url: str):
try:
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
from langchain_community.document_loaders import BSHTMLLoader
loader = BSHTMLLoader.from_response(response)
return loader.load()
except requests.exceptions.Timeout:
print(f"⏱️ Timeout bei {url} – Server antwortet nicht")
return []
except requests.exceptions.RequestException as e:
print(f"🌐 Netzwerkfehler: {e}")
return []
loader = TimeoutSafePDFLoader("grosses_handbuch.pdf", timeout=120)
docs = loader.load_with_retry()
print(f"✅ {len(docs)} Seiten geladen")2. 401 Unauthorized bei HolySheep API
Ursache: Ungültiger API-Key oder falscher base_url konfiguriert.
# ❌ FEHLERHAFT: Falscher Endpunkt
client = OpenAI(
api_key="YOUR_KEY",
base_url="https://api.openai.com/v1" # ❌ FALSCH!
)
❌ AUCH FEHLERHAFT: Tippfehler im Key
client = OpenAI(
api_key="your-holysheep_key", # Falsches Format
base_url="https://api.holysheep.ai/v1"
)
✅ LÖSUNG: Korrekte Konfiguration mit Validierung
from openai import OpenAI, AuthenticationError, RateLimitError
import os
class HolySheepClient:
def __init__(self, api_key: str = None):
self.api_key = api_key or os.environ.get("HOLYSHEEP_API_KEY")
self.base_url = "https://api.holysheep.ai/v1"
self._client = None
self._validate_and_connect()
def _validate_and_connect(self):
"""Validiert den API-Key mit einem Test-Call."""
if not self.api_key:
raise ValueError("API-Key fehlt! Bitte setzen Sie HOLYSHEEP_API_KEY")
if not self.api_key.startswith("sk-"):
raise ValueError("Ungültiges API-Key-Format. Keys beginnen mit 'sk-'")
self._client = OpenAI(
api_key=self.api_key,
base_url=self.base_url
)
# Verbindung testen
try:
self._client.chat.completions.create(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "test"}],
max_tokens=1
)
print("✅ HolySheep AI Verbindung erfolgreich verifiziert")
except AuthenticationError:
raise ValueError("❌ 401 Unauthorized: Bitte API-Key überprüfen")
except Exception as e:
raise ConnectionError(f"Verbindungsfehler: {e}")
def create_completion(self, model: str, messages: list, **kwargs):
"""Wrapper für Chat-Completions mit Fehlerbehandlung."""
try:
return self._client.chat.completions.create(
model=model,
messages=messages,
**kwargs
)
except AuthenticationError:
print("❌ Authentifizierungsfehler – Key prüfen")
raise
except RateLimitError:
print("⚠️ Rate Limit erreicht – bitte pausieren")
raise
Nutzung
try:
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
response = client.create_completion(
model="deepseek-v3.2",
messages=[{"role": "user", "content": "Test"}]
)
except ValueError as e:
print(f"Validierungsfehler: {e}")
except ConnectionError as e:
print(f"Verbindungsproblem: {e}")3. Empty Document List nach dem Laden
Ursache: PDF ist bildbasiert (kein extrahierbarer Text) oder Datei ist corrupted.
# ❌ FEHLERHAFT: Keine Prüfung der geladenen Dokumente
loader = PyPDFLoader("scan.pdf")
docs = loader.load() # Gibt leere Liste zurück, kein Fehler!
print(len(docs)) # 0
✅ LÖSUNG: Umfassende Validierung und Fallback-Strategie
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.document_loaders import UnstructuredPDFLoader
import os
class RobustPDFLoader:
"""Robuster PDF-Loader mit mehrstufiger Fallback-Strategie."""
def __init__(self, file_path: str):
if not os.path.exists(file_path):
raise FileNotFoundError(f"Datei existiert nicht: {file_path}")
self.file_path = file_path
self.file_size = os.path.getsize(file_path)
def load_with_fallback(self) -> list:
"""
Versucht mehrere Lademethoden in Prioritätsreihenfolge.
"""
# Methode 1: Standard PyPDF (für textbasierte PDFs)
print("🔄 Versuche PyPDFLoader...")
try:
loader = PyPDFLoader(self.file_path)
docs = loader.load()
if self._validate_documents(docs):
print(f"✅ PyPDFLoader erfolgreich: {len(docs)} Seiten")
return docs
except Exception as e:
print(f"⚠️ PyPDFLoader fehlgeschlagen: {e}")
# Methode 2: Unstructured (für komplexere PDFs)
print("🔄 Versuche UnstructuredPDFLoader...")
try:
loader = UnstructuredPDFLoader(self.file_path, mode="elements")
docs = loader.load()
if self._validate_documents(docs):
print(f"✅ Unstructured erfolgreich: {len(docs)} Elemente")
return docs
except Exception as e:
print(f"⚠️ Unstructured fehlgeschlagen: {e}")
# Keine Methode hat funktioniert
print(f"❌ Alle Lademethoden fehlgeschlagen für {self.file_path}")
print("💡 Mögliche Ursachen:")
print(" - PDF ist bildbasiert (kein Text vorhanden)")
print(" - PDF ist passwortgeschützt")
print(" - PDF-Datei ist corrupted")
print("💡 Lösung: OCR-Tool wie pytesseract verwenden")
return []
def _validate_documents(self, docs: list) -> bool:
"""Prüft ob Dokumente gültigen Inhalt haben."""
if not docs:
return False
total_chars = sum(len(doc.page_content) for doc in docs)
if total_chars < 10:
print(f"⚠️ Warnung: Sehr wenig Text ({total_chars} Zeichen)")
return False
return True
def load_with_ocr_fallback(self):
"""
Fallback mit OCR für bildbasierte PDFs.
Erfordert: pip install pytesseract pillow
"""
print("🔄 Versuche OCR-basierte Extraktion...")
try:
import pytesseract
from PIL import Image
import fitz # PyMuPDF
doc = fitz.open(self.file_path)
all_text = []
for page_num in range(len(doc)):
page = doc[page_num]
# Text extrahieren
text = page.get_text()
if not text.strip():
# Bild zu Text konvertieren
pix = page.get_pixmap(dpi=300)
img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
text = pytesseract.image_to_string(img, lang='deu+eng')
all_text.append(text)
doc.close()
from langchain.schema import Document
docs = [
Document(page_content=text, metadata={"page": i+1})
for i, text in enumerate(all_text) if text.strip()
]
print(f"✅ OCR erfolgreich: {len(docs)} Seiten extrahiert")
return docs
except ImportError:
print("⚠️ OCR-Bibliotheken nicht installiert")
print(" pip install pytesseract pillow pymupdf")
return []
except Exception as e:
print(f"❌ OCR-Fehler: {e}")
return []
Nutzung
try:
robust_loader = RobustPDFLoader("scan.pdf")
docs = robust_loader.load_with_fallback()
if not docs:
# OCR als letzte Option
docs = robust_loader.load_with_ocr_fallback()
print(f"📄 Gesamt: {len(docs)} Dokumente geladen")
except FileNotFoundError:
print("❌ Datei nicht gefunden")4. UnicodeDecodeError bei Textdateien
Ursache: Falsches Encoding (z.B. UTF-16 statt UTF-8) oder binäre Datei.
# ❌ FEHLERHAFT: Keine Encoding-Handhabung
loader = TextLoader("dokument.txt")
docs = loader.load() # UnicodeDecodeError!
✅ LÖSUNG: Automatische Encoding-Erkennung
from langchain_community.document_loaders import TextLoader
import chardet
class SmartTextLoader:
"""Text-Loader mit automatischer Encoding-Erkennung."""
def __init__(self, file_path: str):
self.file_path = file_path
def detect_encoding(self) -> str:
"""Erkennt das Encoding einer Datei automatisch."""
with open(self.file_path, 'rb') as f:
raw_data = f.read(100000) # Erste 100KB analysieren
result = chardet.detect(raw_data)
encoding = result['encoding']
confidence = result['confidence']
print(f"📋 Erkanntes Encoding: {encoding} (Konfidenz: {confidence:.0%})")
return encoding
def load(self, encoding: str = None) -> list:
"""
Lädt Textdatei mit erkanntem oder angegebenem Encoding.
Args:
encoding: Manuelles Encoding (optional)
"""
detected_encoding = encoding or self.detect_encoding()
encodings_to_try = [
detected_encoding,
'utf-8',
'utf-16',
'utf-16-le',
'utf-16-be',
'latin-1',
'cp1252',
'iso-8859-1'
]
for enc in encodings_to_try:
try:
loader = TextLoader(self.file_path, encoding=enc)
docs = loader.load()
# Inhalt validieren
if docs and docs[0].page_content.strip():
print(f"✅ Erfolgreich geladen mit Encoding: {enc}")
return docs
except (UnicodeDecodeError, UnicodeError):
continue
except Exception as e:
print(f"⚠️ Fehler mit {enc}: {e}")
continue
print(f"❌ Kein passendes Encoding für {self.file_path} gefunden")
return []
def load_raw(self) -> str:
"""Lädt Datei als Rohtext mit Binär-Fallback."""
try:
# Versuche verschiedene Encodings
with open(self.file_path, 'r', encoding='utf-8') as f:
return f.read()
except UnicodeDecodeError:
# Fallback: Binär lesen und ignorieren
with open(self.file_path, 'rb') as f:
raw = f.read()
return raw.decode('utf-8', errors='ignore')
Nutzung
try:
smart_loader = SmartTextLoader("bericht.txt")
docs = smart_loader.load()
if docs:
print(f"📝 {len(docs[0].page_content)} Zeichen geladen")
else:
print("⚠️ Datei konnte nicht gelesen werden")
except Exception as e:
print(f"Fehler: {e}")Praxiserfahrung und Performance-Tipps
In meinen Projekten habe ich gelernt, dass die Dokumentenverarbeitung oft der Flaschenhals ist. Hier meine wichtigsten Erkenntnisse:
- Chunk-Größe anpassen: Für technische Dokumentation funktionieren 1000-1500 Tokens am besten, für natürliche Sprache eher 500-800.
- Overlap nicht vergessen: 20% Overlap verhindert, dass wichtige Informationen an Chunk-Grenzen verloren gehen.
- Metadaten nutzen: Quellenangabe, Datum und Seitenummer helfen bei der Nachvollziehbarkeit der Antworten.
- Batch-Processing: Bei großen Dokumentenmengen (>100 PDFs) lohnt sich parallelisiertes Laden mit ThreadPoolExecutor.
- Fehlerbehandlung: Mindestens 3 Retry-Versuche mit exponentiellem Backoff für Netzwerkoperationen.
Zusammenfassung
LangChain Document Loaders bieten eine flexible Grundlage für RAG-Anwendungen. Mit HolySheep AI als Backend reduzieren Sie Ihre API-Kosten drastisch – von $50 auf etwa $2 pro 1000 Anfragen bei vergleichbarer Qualität.
Die wichtigsten Takeaways:
- Verwenden Sie immer robuste Fehlerbehandlung mit Timeouts und Retries