Die Analyse langer Dokumente stellt hohe Anforderungen an KI-APIs. In diesem Tutorial vergleichen wir die besten Lösungen für die Verarbeitung umfangreicher Texte und zeigen, wie Sie mit HolySheep AI bis zu 85% Ihrer Kosten sparen können.
Vergleichstabelle: Die wichtigsten Anbieter im Überblick
| Kriterium | HolySheep AI | Offizielle APIs | Andere Relay-Dienste |
|---|---|---|---|
| Preis GPT-4.1 | $0.42/MTok | $8/MTok | $2-4/MTok |
| Preis Claude Sonnet 4.5 | $1.50/MTok | $15/MTok | $5-8/MTok |
| Preis Gemini 2.5 Flash | $0.25/MTok | $2.50/MTok | $0.80-1.20/MTok |
| Latenz | <50ms | 100-300ms | 80-200ms |
| WeChat/Alipay | ✅ Ja | ❌ Nein | Variiert |
| Kostenlose Credits | ✅ Ja | ❌ Nein | Selten |
| Wechselkurs | ¥1=$1 | $1=$1 | Variiert |
Warum HolySheep AI für Langdokumente?
In meiner mehrjährigen Praxis mit Dokumentenanalyse habe ich festgestellt: Die Wahl des richtigen API-Anbieters kann den Unterschied zwischen profitablen und defizitären Projekten ausmachen. HolySheep AI bietet nicht nur unschlagbare Preise, sondern auch eine stabile Infrastruktur mit garantierter Latenz unter 50 Millisekunden.
Mit dem Wechselkurs von ¥1=$1 und der Unterstützung für WeChat und Alipay ist HolySheep besonders attraktiv für den chinesischen Markt und Entwickler, die kosteneffiziente Lösungen suchen.
Implementation: Long Document Analysis mit HolySheep
Beispiel 1: Grundlegende Langdokument-Verarbeitung
#!/usr/bin/env python3
"""
Long Document Analysis mit HolySheep AI
Kostengünstige Alternative zu offiziellen APIs
"""
import requests
import json
import time
HolySheep API Configuration
WICHTIG: base_url MUSS https://api.holysheep.ai/v1 sein
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def analyze_long_document(document_text: str, model: str = "gpt-4.1") -> dict:
"""
Analysiert ein langes Dokument mit strukturiertem Output.
Args:
document_text: Der zu analysierende Text
model: Modell-Auswahl (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash)
Returns:
Dictionary mit Analyseergebnissen
"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
# Chunk-Handling für lange Dokumente
# HolySheep unterstützt bis zu 128K Token pro Anfrage
chunks = split_into_chunks(document_text, chunk_size=100000)
all_summaries = []
for i, chunk in enumerate(chunks):
payload = {
"model": model,
"messages": [
{
"role": "system",
"content": "Du bist ein professioneller Dokumentanalyst. "
"Analysiere das Dokument strukturiert und gebe "
"JSON-Format zurück."
},
{
"role": "user",
"content": f"Analyse diesen Dokumentabschnitt (Teil {i+1}/{len(chunks)}):\n\n{chunk}"
}
],
"temperature": 0.3,
"response_format": {"type": "json_object"}
}
start_time = time.time()
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=60
)
latency = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
all_summaries.append({
"chunk_index": i + 1,
"summary": result["choices"][0]["message"]["content"],
"latency_ms": round(latency, 2),
"usage": result.get("usage", {})
})
else:
print(f"Fehler bei Chunk {i+1}: {response.status_code}")
# Finale Zusammenfassung
return {
"total_chunks": len(chunks),
"chunk_analyses": all_summaries,
"estimated_cost": calculate_cost(all_summaries)
}
def split_into_chunks(text: str, chunk_size: int = 100000) -> list:
"""Teilt Text in verarbeitbare Chunks."""
sentences = text.replace('\n', ' ').split('. ')
chunks = []
current_chunk = ""
for sentence in sentences:
if len(current_chunk) + len(sentence) < chunk_size:
current_chunk += sentence + ". "
else:
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = sentence + ". "
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
def calculate_cost(analyses: list) -> dict:
"""Berechnet die Kosten basierend auf HolySheep-Preisen 2026."""
prices = {
"gpt-4.1": 0.42,
"claude-sonnet-4.5": 1.50,
"gemini-2.5-flash": 0.25,
"deepseek-v3.2": 0.042
}
total_input = sum(a["usage"].get("prompt_tokens", 0) for a in analyses)
total_output = sum(a["usage"].get("completion_tokens", 0) for a in analyses)
# Durchschnittspreis annehmen
avg_price = 0.42 / 1000 # GPT-4.1 Preis pro Token
return {
"input_tokens": total_input,
"output_tokens": total_output,
"estimated_cost_usd": round((total_input + total_output) * avg_price, 4)
}
Beispiel-Nutzung
if __name__ == "__main__":
# Test mit Beispieldokument
sample_doc = """
Die künstliche Intelligenz hat in den letzten Jahren massive Fortschritte gemacht.
Große Sprachmodelle können nun komplexe Dokumente analysieren, zusammenfassen und
kritisch bewerten. Dies eröffnet neue Möglichkeiten für Unternehmen und Forscher.
Die Kosten für solche Analysen waren bisher jedoch ein Hindernis für kleinere
Organisationen. Mit Diensten wie HolySheep AI wird KI nun für alle zugänglich.
"""
result = analyze_long_document(sample_doc)
print(json.dumps(result, indent=2, ensure_ascii=False))
Beispiel 2: Streaming-Analyse für große Dokumente
#!/usr/bin/env python3
"""
Streaming Document Analysis mit HolySheep AI
Optimiert für Echtzeit-Verarbeitung langer Dokumente
"""
import requests
import json
from typing import Generator, Iterator
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class DocumentStreamAnalyzer:
"""
Streaming-basierte Dokumentenanalyse für große Dateien.
Verarbeitet Dokumente chunk-weise mit Echtzeit-Feedback.
"""
def __init__(self, api_key: str, model: str = "gemini-2.5-flash"):
self.api_key = api_key
self.model = model
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def stream_analyze(self, document_path: str) -> Generator[dict, None, None]:
"""
Analysiert ein Dokument im Streaming-Modus.
Yields:
Dictionary mit Fortschritt und Ergebnissen
"""
with open(document_path, 'r', encoding='utf-8') as f:
full_text = f.read()
# Für Gemini 2.5 Flash mit höherem Context
chunks = self._smart_chunk(full_text, target_chunks=10)
total_chunks = len(chunks)
for idx, chunk in enumerate(chunks):
yield {
"status": "processing",
"chunk": idx + 1,
"total": total_chunks,
"progress": round((idx + 1) / total_chunks * 100, 1)
}
result = self._analyze_chunk(chunk)
yield {
"status": "complete",
"chunk": idx + 1,
"result": result
}
def _smart_chunk(self, text: str, target_chunks: int) -> list:
"""Intelligente Chunk-Aufteilung für optimale Kontextnutzung."""
chunk_size = len(text) // target_chunks
chunks = []
for i in range(0, len(text), chunk_size):
# Bei Chunk-Grenzen an Satzgrenzen orientieren
end = min(i + chunk_size, len(text))
if end < len(text):
# Zum letzten Satzende zurückgehen
last_period = text.rfind('.', i, end)
if last_period > i:
end = last_period + 1
chunks.append(text[i:end].strip())
return [c for c in chunks if c]
def _analyze_chunk(self, chunk: str) -> dict:
"""Analysiert einen einzelnen Chunk mit Streaming."""
payload = {
"model": self.model,
"messages": [
{
"role": "system",
"content": "Analysiere diesen Textausschnitt. "
"Identifiziere: Hauptthemen, Schlüsselargumente, "
"wichtige Daten und Fazit."
},
{
"role": "user",
"content": chunk
}
],
"stream": True,
"temperature": 0.2
}
full_response = ""
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=self.headers,
json=payload,
stream=True,
timeout=120
)
for line in response.iter_lines():
if line:
line_text = line.decode('utf-8')
if line_text.startswith("data: "):
data = line_text[6:]
if data != "[DONE]":
try:
chunk_data = json.loads(data)
if "choices" in chunk_data:
delta = chunk_data["choices"][0].get("delta", {})
if "content" in delta:
full_response += delta["content"]
except json.JSONDecodeError:
continue
return {"analysis": full_response, "chars": len(chunk)}
Optimierte Batch-Verarbeitung
def batch_analyze_documents(documents: list, model: str = "deepseek-v3.2") -> list:
"""
Parallele Batch-Verarbeitung mehrerer Dokumente.
DeepSeek V3.2 bietet hier das beste Preis-Leistungs-Verhältnis.
Kostenvergleich:
- GPT-4.1: $8/MTok
- Claude Sonnet 4.5: $15/MTok
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3.2: $0.42/MTok (heilig!)
"""
results = []
for doc in documents:
payload = {
"model": model,
"messages": [
{"role": "system", "content": "Fasse das Dokument prägnant zusammen."},
{"role": "user", "content": doc[:50000]} # Max 50K Zeichen
],
"temperature": 0.3
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
json=payload,
timeout=90
)
if response.status_code == 200:
data = response.json()
results.append({
"summary": data["choices"][0]["message"]["content"],
"usage": data.get("usage", {}),
"latency_ms": response.elapsed.total_seconds() * 1000
})
return results
Kostenoptimierte Zusammenfassung
def generate_executive_summary(document_text: str) -> dict:
"""
Generiert eine Executive Summary mit Kostenanalyse.
Nutzt Gemini 2.5 Flash für Geschwindigkeit und niedrige Kosten.
"""
payload = {
"model": "gemini-2.5-flash",
"messages": [
{
"role": "system",
"content": """Du bist ein erfahrener Analyst. Erstelle eine Executive Summary mit:
1. Kernthema (max 20 Wörter)
2. 3 wichtigste Erkenntnisse
3. Handlungsempfehlung
4. Risikoeinschätzung
Antworte im JSON-Format."""
},
{
"role": "user",
"content": document_text[:80000] # Gemini 2.5 Flash Context
}
],
"response_format": {"type": "json_object"},
"temperature": 0.2
}
response = requests.post(
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
json=payload,
timeout=60
)
if response.status_code == 200:
return {
"summary": response.json()["choices"][0]["message"]["content"],
"usage": response.json().get("usage", {}),
"cost_usd": response.json()["usage"]["total_tokens"] * 0.0025
}
return {"error": f"HTTP {response.status_code}"}
if __name__ == "__main__":
# Initialisierung
analyzer = DocumentStreamAnalyzer(API_KEY)
# Streaming-Analyse starten
print("Starte Streaming-Analyse...")
for update in analyzer.stream_analyze("grosses_dokument.txt"):
if update["status"] == "processing":
print(f"Verarbeite Chunk {update['chunk']}/{update['total']} "
f"({update['progress']}%)")
else:
print(f"Chunk {update['chunk']} abgeschlossen")
Beispiel 3: Fortgeschrittene Document Intelligence Pipeline
#!/usr/bin/env python3
"""
Fortschrittene Document Intelligence Pipeline mit HolySheep AI
Multi-Model Architektur für optimale Analyseergebnisse
"""
import requests
import hashlib
import time
from datetime import datetime
from typing import Optional, List
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class DocumentIntelligencePipeline:
"""
Multi-Stage Pipeline für umfassende Dokumentenanalyse.
Nutzt verschiedene Modelle für unterschiedliche Aufgaben.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.cache = {} # Einfacher Cache für wiederholte Anfragen
# Modell-Konfiguration mit Preisen 2026
self.models = {
"extraction": "deepseek-v3.2", # $0.42/MTok - Extraktion
"analysis": "gemini-2.5-flash", # $2.50/MTok - Analyse
"synthesis": "gpt-4.1", # $8/MTok - Synthese
"review": "claude-sonnet-4.5" # $15/MTok - Review
}
# Kosten-Tracking
self.total_cost = 0.0
self.total_tokens = 0
def process_document(self, document: str, mode: str = "full") -> dict:
"""
Verarbeitet ein Dokument durch die vollständige Pipeline.
Stages:
1. Extraction (DeepSeek) - Fakten extrahieren
2. Analysis (Gemini) - Themen und Muster identifizieren
3. Synthesis (GPT-4.1) - Zusammenfassung erstellen
4. Review (Claude) - Qualitätsprüfung
"""
start_time = time.time()
stages = {}
if mode in ["full", "extraction"]:
extraction_result = self._stage_extraction(document)
stages["extraction"] = extraction_result
if mode == "extraction":
return self._finalize(stages, time.time() - start_time)
if mode in ["full", "analysis"]:
analysis_result = self._stage_analysis(document, stages.get("extraction"))
stages["analysis"] = analysis_result
if mode == "analysis":
return self._finalize(stages, time.time() - start_time)
if mode in ["full", "synthesis"]:
synthesis_result = self._stage_synthesis(
document,
stages.get("extraction"),
stages.get("analysis")
)
stages["synthesis"] = synthesis_result
if mode == "synthesis":
return self._finalize(stages, time.time() - start_time)
if mode == "full":
review_result = self._stage_review(
stages.get("synthesis", {}).get("text", "")
)
stages["review"] = review_result
return self._finalize(stages, time.time() - start_time)
return {"error": "Invalid mode"}
def _stage_extraction(self, document: str) -> dict:
"""Stage 1: Faktenextraktion mit DeepSeek V3.2"""
# Cache-Check
doc_hash = hashlib.md5(document[:10000].encode()).hexdigest()
if doc_hash in self.cache:
return {"cached": True, "text": self.cache[doc_hash]}
payload = {
"model": self.models["extraction"],
"messages": [
{
"role": "system",
"content": """Extrahiere folgende Informationen aus dem Text:
- Schlüsselpersonen und Organisationen
- Wichtige Daten und Zahlen
- Chronologische Ereignisse
- Kernaussagen
Format: Strukturiertes JSON"""
},
{"role": "user", "content": document[:50000]}
],
"response_format": {"type": "json_object"},
"temperature": 0.1
}
response = self._make_request(payload)
if response:
self.cache[doc_hash] = response.get("content", "")
return {"cached": False, "text": response.get("content", "")}
return {"error": "Extraction failed"}
def _stage_analysis(self, document: str, extraction: Optional[dict]) -> dict:
"""Stage 2: Themenanalyse mit Gemini 2.5 Flash"""
context = ""
if extraction and "text" in extraction:
context = f"\nExtrahierte Fakten:\n{extraction['text']}\n"
payload = {
"model": self.models["analysis"],
"messages": [
{
"role": "system",
"content": "Analysiere das Dokument systematisch. "
"Identifiziere Hauptthemen, Zusammenhänge und Muster."
},
{
"role": "user",
"content": f"{context}\n\nAnalyse diesen Text:\n{document[:60000]}"
}
],
"temperature": 0.2
}
response = self._make_request(payload)
return {"text": response.get("content", "") if response else ""}
def _stage_synthesis(self, document: str, extraction: Optional[dict],
analysis: Optional[dict]) -> dict:
"""Stage 3: Synthese mit GPT-4.1"""
context = ""
if extraction and "text" in extraction:
context += f"\nExtrahierte Daten:\n{extraction['text'][:2000]}\n"
if analysis and "text" in analysis:
context += f"\nAnalyse:\n{analysis['text'][:2000]}\n"
payload = {
"model": self.models["synthesis"],
"messages": [
{
"role": "system",
"content": "Erstelle eine prägnante, gut strukturierte "
"Zusammenfassung auf Basis der extrahierten "
"Informationen und Analysen."
},
{
"role": "user",
"content": f"Originaldokument:\n{document[:30000]}\n\n"
f"Kontext:\n{context}"
}
],
"temperature": 0.3
}
response = self._make_request(payload)
return {"text": response.get("content", "") if response else ""}
def _stage_review(self, synthesis: str) -> dict:
"""Stage 4: Qualitätsreview mit Claude Sonnet 4.5"""
payload = {
"model": self.models["review"],
"messages": [
{
"role": "system",
"content": "Du bist ein kritischer Qualitätsprüfer. "
"Bewerte die Zusammenfassung auf Vollständigkeit, "
"Genauigkeit und Lesbarkeit. Markiere eventuelle "
"Fehler oder Lücken."
},
{"role": "user", "content": synthesis[:10000]}
],
"temperature": 0.1
}
response = self._make_request(payload)
return {"text": response.get("content", "") if response else ""}
def _make_request(self, payload: dict) -> Optional[dict]:
"""Führt einen API-Request durch mit Fehlerbehandlung."""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = requests.post(
f"{BASE_URL}/chat/completions",
headers=headers,
json=payload,
timeout=120
)
if response.status_code == 200:
data = response.json()
# Token-Nutzung tracken
usage = data.get("usage", {})
tokens = usage.get("total_tokens", 0)
self.total_tokens += tokens
# Modell-Preis ermitteln und Kosten berechnen
model = payload.get("model", "")
price = self._get_price(model)
self.total_cost += (tokens / 1_000_000) * price
return {
"content": data["choices"][0]["message"]["content"],
"usage": usage
}
elif response.status_code == 429:
print("Rate limit erreicht, warte...")
time.sleep(5)
return self._make_request(payload)
else:
print(f"API-Fehler: {response.status_code}")
return None
except requests.exceptions.Timeout:
print("Timeout bei Request")
return None
except requests.exceptions.RequestException as e:
print(f"Netzwerkfehler: {e}")
return None
def _get_price(self, model: str) -> float:
"""Gibt den Preis pro Million Token zurück."""
prices = {
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00
}
return prices.get(model, 1.0)
def _finalize(self, stages: dict, elapsed: float) -> dict:
"""Finalisiert das Ergebnis mit Metadaten."""
return {
"stages": stages,
"metrics": {
"processing_time_seconds": round(elapsed, 2),
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost, 4),
"timestamp": datetime.now().isoformat()
}
}
Kostenanalyse-Tool
def compare_costs(document_size_chars: int, num_documents: int = 1) -> dict:
"""
Vergleicht die Kosten verschiedener Modelle für Langdokumente.
Annahme: ~4 Zeichen pro Token
"""
tokens = (document_size_chars * num_documents) / 4
models = {
"GPT-4.1": 8.00,
"Claude Sonnet 4.5": 15.00,
"Gemini 2.5 Flash": 2.50,
"DeepSeek V3.2": 0.42
}
results = {}
for name, price_per_mtok in models.items():
cost = (tokens / 1_000_000) * price_per_mtok
results[name] = {
"tokens": int(tokens),
"cost_per_1m_tokens_usd": price_per_mtok,
"total_cost_usd": round(cost, 4)
}
# HolySheep Ersparnis berechnen
holysheep_cost = results["DeepSeek V3.2"]["total_cost_usd"]
official_cost = results["GPT-4.1"]["total_cost_usd"]
return {
"scenarios": results,
"savings": {
"vs_gpt4": round((1 - holysheep_cost/official_cost) * 100, 1),
"vs_claude": round((1 - holysheep_cost/results["Claude Sonnet 4.5"]["total_cost_usd"]) * 100, 1)
}
}
if __name__ == "__main__":
# Pipeline initialisieren
pipeline = DocumentIntelligencePipeline(API_KEY)
# Beispiel: 10.000 Zeichen Dokument
sample_doc = "A" * 10000
result = pipeline.process_document(sample_doc, mode="full")
print(f"Verarbeitungszeit: {result['metrics']['processing_time_seconds']}s")
print(f"Gesamtkosten: ${result['metrics']['total_cost_usd']}")
# Kostenvergleich
comparison = compare_costs(50000, num_documents=100)
print("\nKostenvergleich für 100 Dokumente à 50.000 Zeichen:")
for model, data in comparison["scenarios"].items():
print(f" {model}: ${data['total_cost_usd']}")
print(f"\nHolySheep Ersparnis vs GPT-4.1: {comparison['savings']['vs_gpt4']}%")
Häufige Fehler und Lösungen
Fehler 1: Rate Limit überschritten (HTTP 429)
# FEHLERHAFT - Keine Retry-Logik
response = requests.post(url, headers=headers, json=payload)
result = response.json() # crashed bei 429
LÖSUNG - Exponential Backoff mit Retry
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def robust_request(url: str, headers: dict, payload: dict, max_retries: int = 3) -> dict:
"""
Robuster API-Request mit automatischem Retry bei Rate Limits.
"""
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
for attempt in range(max_retries):
try:
response = session.post(
url,
headers=headers,
json=payload,
timeout=(30, 120)
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
wait_time = 2 ** attempt + random.uniform(0, 1)
print(f"Rate limit. Warte {wait_time:.1f}s...")
time.sleep(wait_time)
continue
elif response.status_code == 400:
error = response.json()
raise ValueError(f"Ungültige Anfrage: {error}")
else:
print(f"HTTP {response.status_code}: {response.text}")
except requests.exceptions.Timeout:
print(f"Timeout bei Versuch {attempt + 1}")
if attempt == max_retries - 1:
raise
raise RuntimeError("Max retries überschritten")
Nutzung mit HolySheep
result = robust_request(
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
payload={"model": "gpt-4.1", "messages": [{"role": "user", "content": "Hallo"}]}
)
Fehler 2: Token-Limit überschritten bei langen Dokumenten
# FEHLERHAFT - Sendet gesamtes Dokument ohne Chunking
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": large_document}] # 100K+ Zeichen!
}
Führt zu 400 Bad Request oder 413 Payload Too Large
LÖSUNG - Intelligentes Chunking mit Overlap
def smart_chunk_document(text: str, max_tokens: int = 6000, overlap: int = 500) -> list:
"""
Teilt ein Dokument in überlappende Chunks für maximale Kontextnutzung.
Args:
text: Eingabetext
max_tokens: Maximale Token pro Chunk (mit Safety Margin)
overlap: Überlappung zwischen Chunks für Kontextkontinuität
Returns:
Liste von Dictionaries mit Chunk-Text und Metadaten
"""
# Annahme: ~4 Zeichen pro Token
chars_per_token = 4
max_chars = max_tokens * chars_per_token
overlap_chars = overlap * chars_per_token
chunks = []
start = 0
while start < len(text):
end = min(start + max_chars, len(text))
# Bei Chunk-Ende an sinnvoller Stelle trennen
if end < len(text):
# Erstelle: Zum nächsten Satzende oder Absatz
for delimiter in ['\n\n', '\n', '. ', '; ']:
last_delimiter = text.rfind(delimiter, start + max_chars - 200, end)
if last_delimiter > start + max_chars // 2:
end = last_delimiter + len(delimiter)
break
chunk_text = text[start:end].strip()
if chunk_text:
chunks.append({
"text": chunk_text,
"start_char": start,
"end_char": end,
"index": len(chunks)
})
# Überlappend zum nächsten Chunk
start = end - overlap_chars if end < len(text) else end
return chunks
def process_long_document(document: str, task: str) -> list:
"""
Verarbeitet langes Dokument chunk-weise und sammelt Ergebnisse.
"""
chunks = smart_chunk_document(document, max_tokens=8000)
results = []
for chunk in chunks:
payload = {
"model": "gpt-4.1",
"messages": [
{
"role": "system",
"content": f"{task}\nWichtig: Antworte prägnant und strukturiert."
},
{"role": "user", "content": chunk["text"]}
],
"max_tokens": 500
}
try:
response = robust_request(
f"{BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json=payload
)
results.append({
"chunk_index": chunk["index"],
"result": response["choices"][0]["message"]["content"]
})
except Exception as e:
results.append({
"chunk_index": chunk["index"],
"error": str(e)
})
return results
Nutzung
long_doc = open("forschungspapier.txt").read()
results = process_long_document(long_doc, "Extrahiere die wichtigsten Forschungsfragen")
Fehler 3: Falsches Error-Handling bei API-Timeout
# FEHLERHAFT - Ignoriert Timeouts komplett
try:
response = requests.post(url, json=payload)
data = response.json()
except: # Zu breites Catch!
data = {"error": "unknown"}
LÖSUNG - Spezifisches Error-Handling mit Fallback
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Union
class APIError(Exception):
"""Spezifische Exception für API-Fehler."""
def __init__(self, code: int, message: str, retryable: bool = False):
self.code = code
self.message = message
self.retryable = retryable
super().__init__(f"[{code}] {message}")
@dataclass
class APIResponse:
"""Strukturierte API-Antwort mit Fehlerbehandlung."""
success: bool
data: Optional[dict] = None
error: Optional[str