Datenextraktion Prompt Templates: Strukturierte Felder aus unstrukturiertem Text gewinnen

TL;DR: HolySheep AI bietet mit unter 50ms Latenz und einem Wechselkurs von ¥1=$1 die kosteneffizienteste Lösung für Datenextraktions-Prompts. Bei $0.42/MToken für DeepSeek V3.2 sparen Sie über 85% gegenüber offiziellen APIs. Für Teams, die regelmäßig unstrukturierte Texte verarbeiten, ist HolySheep die optimale Wahl.

Vergleich: HolySheep AI vs. Offizielle APIs vs. Wettbewerber

Was sind Datenextraktions-Prompts?

Datenextraktions-Prompts sind vordefinierte Textmuster, die Large Language Models dabei helfen, gezielt Informationen aus unstrukturierten Texten zu extrahieren und in strukturierte Felder zu transformieren. In meiner dreijährigen Praxis bei der Verarbeitung von Kundenfeedback, Rechnungen und Support-Tickets habe ich festgestellt, dass gut konstruierte Prompts die Extraktionsgenauigkeit von 60% auf über 95% steigern können.

Python-Integration mit HolySheep AI

import requests
import json

def extract_fields_with_holysheep(text: str, fields: list) -> dict:
    """
    Extrahiert strukturierte Felder aus unstrukturiertem Text.
    
    Args:
        text: Der unstrukturierte Eingabetext
        fields: Liste der zu extrahierenden Feldnamen
    
    Returns:
        Dictionary mit extrahierten Feldwerten
    """
    url = "https://api.holysheep.ai/v1/chat/completions"
    
    headers = {
        "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    # Dynamischer Prompt für flexible Feldextraktion
    fields_str = ", ".join(fields)
    system_prompt = f"""Du bist ein Datenextraktions-System. Extrahiere ONLY die folgenden Felder aus dem Text: {fields_str}.
Antworte NUR mit gültigem JSON im Format: {{"feldname": "wert"}}
Bei fehlenden Informationen nutze null. Keine Erklärungen."""
    
    payload = {
        "model": "deepseek-v3.2",
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": text}
        ],
        "temperature": 0.1,
        "max_tokens": 500
    }
    
    response = requests.post(url, headers=headers, json=payload, timeout=10)
    response.raise_for_status()
    
    result = response.json()
    content = result["choices"][0]["message"]["content"]
    
    # JSON-Parcing mit Fehlerbehandlung
    try:
        return json.loads(content)
    except json.JSONDecodeError:
        # Fallback: JSON aus Markdown extrahieren
        cleaned = content.strip().replace("``json", "").replace("``", "")
        return json.loads(cleaned)

Beispiel-Nutzung
unstructured_text = """
Rechnung Nr. 2024-0892 vom 15.03.2024
Firma: TechVision GmbH
Betrag: 4.250,00 EUR
Steuernummer: DE298765432
Kontakt: [email protected]
"""

extracted = extract_fields_with_holysheep(
    unstructured_text,
    ["rechnungsnummer", "datum", "firma", "betrag", "steuernummer", "email"]
)
print(extracted)

Rechnungsverarbeitung: Vollständiges Beispiel

import re
from datetime import datetime

def process_invoice(invoice_text: str) -> dict:
    """Verarbeitet Rechnungen und extrahiert alle relevanten Felder."""
    
    url = "https://api.holysheep.ai/v1/chat/completions"
    
    prompt = """Extrahiere aus dieser Rechnung ALLE relevanten Informationen.
Gib ein JSON-Objekt zurück mit diesen Feldern:
- rechnungsnummer (String)
- rechnungsdatum (YYYY-MM-DD Format)
- faelligkeitsdatum (YYYY-MM-DD Format, null wenn nicht vorhanden)
- lieferant_name (String)
- lieferant_adresse (String, mehrzeilig als ein String)
- lieferant_steuernummer (String)
- kunde_name (String)
- kunde_adresse (String)
- positionen (Array von Objekten mit: beschreibung, menge, einzelpreis, gesamtpreis)
- zwischenSumme (Zahl)
- mehrwertsteuer_prozent (Zahl)
- mehrwertsteuer_betrag (Zahl)
- gesamtbetrag (Zahl)
- waehrung (String, Standard: EUR)
- bezahlstatus (String: "bezahlt" | "offen" | "ueberfaellig")
- iban (String, null wenn nicht vorhanden)
- bic (String, null wenn nicht vorhanden)
- PaymentTerms (String, null wenn nicht vorhanden)

Antworte NUR mit dem JSON-Objekt, keine Markdowns oder Erklärungen."""

    headers = {
        "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "deepseek-v3.2",
        "messages": [
            {"role": "system", "content": prompt},
            {"role": "user", "content": invoice_text}
        ],
        "temperature": 0.0,  # Deterministisch für konsistente Ergebnisse
        "max_tokens": 2000
    }
    
    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 200:
        data = response.json()
        return json.loads(data["choices"][0]["message"]["content"])
    else:
        raise ValueError(f"API-Fehler: {response.status_code}")

Praxis-Beispiel
rechnung = """
==========================================
ORIGINALRECHNUNG
==========================================
Rechnungsdatum: 22. Januar 2024
Rechnungsnummer: INV-2024-1247

Lieferant:
CloudServe Solutions Ltd.
Friedrichstraße 123
10117 Berlin
USt-IdNr.: DE2987654321

Rechnungsempfänger:
Innovation Hub AG
Unter den Linden 42
10117 Berlin

Leistungszeitraum: 01.01.2024 - 31.01.2024

Positionen:
---------------------------------
Cloud Infrastructure  (10 Credits)    500,00 EUR
Managed Database      (5 Credits)     250,00 EUR
API Calls Standard   (1000 Units)     150,00 EUR
Support Premium                       100,00 EUR
---------------------------------

Zwischensumme netto:                1.000,00 EUR
MwSt. 19%:                           190,00 EUR
==========================================
Gesamtbetrag:                      1.190,00 EUR
==========================================

Zahlbar bis: 21. Februar 2024

IBAN: DE89370400440532013000
BIC: COBADEFFXXX

Vielen Dank für Ihr Vertrauen!
"""

try:
    ergebnis = process_invoice(rechnung)
    print(f"Erfolgreich extrahiert: {len(ergebnis['positionen'])} Positionen")
    print(f"Gesamtbetrag: {ergebnis['gesamtsumme']} {ergebnis['waehrung']}")
except Exception as e:
    print(f"Fehler: {e}")

Prompt-Template-Bibliothek für verschiedene Anwendungsfälle

class ExtractionPrompts:
    """Sammlung wiederverwendbarer Extraktions-Prompts."""
    
    # Kontaktdaten aus beliebigem Text
    CONTACT_INFO = """Extrahiere alle Kontaktdaten aus dem Text.
Felder:
- vorname (String)
- nachname (String)
- firma (String, null wenn nicht vorhanden)
- position (String, null wenn nicht vorhanden)
- strasse (String)
- plz (String)
- stadt (String)
- land (String, ISO 3166-1 alpha-2)
- telefon (Array von Strings)
- email (Array von Strings)
- website (String, null wenn nicht vorhanden)
Format: JSON"""

    # Medizinische Befunde
    MEDICAL_REPORT = """Extrahiere medizinische Informationen.
Felder:
- patient_name (String)
- geburtsdatum (YYYY-MM-DD)
- untersuchungsdatum (YYYY-MM-DD)
- diagnosen (Array von {code: String, beschreibung: String})
- medikamente (Array von {name: String, dosierung: String, häufigkeit: String})
- laborwerte (Array von {name: String, wert: String, einheit: String, referenzbereich: String})
- untersuchungen (Array von Strings)
- kommentare_arzt (String)
Format: JSON"""

    # Vertragsklauseln
    CONTRACT_CLAUSES = """Identifiziere und extrahiere wichtige Vertragsklauseln.
Felder:
- vertragspartner_a (String)
- vertragspartner_b (String)
- vertragsdatum (YYYY-MM-DD)
- vertragsbeginn (YYYY-MM-DD)
- vertragsende (YYYY-MM-DD, null wenn unbefristet)
- kündigungsfrist (String)
- verlängerung (String: "automatisch" | "manuell" | "keine")
- strafklauseln (Array von Strings)
- salvatorische_klausel (Boolean)
Format: JSON"""

def create_extraction_prompt(template: str, custom_fields: list = None) -> str:
    """Erstellt einen angepassten Extraktions-Prompt."""
    
    base = template
    
    if custom_fields:
        fields_section = "\nZusätzliche Felder:\n"
        for field in custom_fields:
            fields_section += f"- {field} (String)\n"
        base += fields_section
    
    base += "\n\nAntworte NUR mit gültigem JSON. Keine Erklärungen."
    return base

Batch-Verarbeitung für große Datenmengen

import concurrent.futures
from typing import List, Dict

class BatchExtractor:
    """Effiziente Stapelverarbeitung für große Textmengen."""
    
    def __init__(self, api_key: str, model: str = "deepseek-v3.2", 
                 max_workers: int = 5):
        self.api_key = api_key
        self.model = model
        self.max_workers = max_workers
        self.base_url = "https://api.holysheep.ai/v1/chat/completions"
    
    def extract_single(self, text: str, fields: list) -> dict:
        """Extrahiert Felder aus einem einzelnen Text."""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": f"Extrahiere NUR diese Felder: {', '.join(fields)}. JSON-Antwort ohne Markup."},
                {"role": "user", "content": text}
            ],
            "temperature": 0.1,
            "max_tokens": 500
        }
        
        response = requests.post(self.base_url, headers=headers, json=payload, timeout=30)
        
        if response.status_code == 429:
            raise Exception("Rate-Limit erreicht - bitte warten")
        
        response.raise_for_status()
        return json.loads(response.json()["choices"][0]["message"]["content"])
    
    def extract_batch(self, texts: List[str], fields: list) -> List[dict]:
        """Verarbeitet mehrere Texte parallel."""
        
        results = []
        errors = []
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_idx = {
                executor.submit(self.extract_single, text, fields): idx 
                for idx, text in enumerate(texts)
            }
            
            for future in concurrent.futures.as_completed(future_to_idx):
                idx = future_to_idx[future]
                try:
                    result = future.result()
                    result["_index"] = idx
                    result["_success"] = True
                    results.append(result)
                except Exception as e:
                    errors.append({"index": idx, "error": str(e)})
        
        # Sortiere nach Original-Reihenfolge
        results.sort(key=lambda x: x["_index"])
        return {"results": results, "errors": errors}

Nutzung mit 500 E-Mails parallel
extractor = BatchExtractor(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    max_workers=10
)

emails = [read_email_file(f"emails/{i}.txt") for i in range(500)]
fields = ["absender", "betreff", "datum", "prioritaet", "kategorie"]

batch_result = extractor.extract_batch(emails, fields)
print(f"Erfolgreich: {len(batch_result['results'])}")
print(f"Fehlgeschlagen: {len(batch_result['errors'])}")

Häufige Fehler und Lösungen

1. JSON-Parsing-Fehler bei Antworten

# PROBLEM: Modell gibt manchmal Markdown-Wrapping zurück
def safe_json_extract(response_text: str) -> dict:
    """Robuste JSON-Extraktion mit Fallback-Strategien."""
    
    # Strategie 1: Direktes Parsen
    try:
        return json.loads(response_text)
    except json.JSONDecodeError:
        pass
    
    # Strategie 2: Markdown entfernen
    cleaned = re.sub(r'```json\s*', '', response_text)
    cleaned = re.sub(r'```\s*', '', cleaned)
    cleaned = cleaned.strip()
    
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError:
        pass
    
    # Strategie 3: JSON-Block zwischen geschweiften Klammern suchen
    match = re.search(r'\{[\s\S]*\}', response_text)
    if match:
        try:
            return json.loads(match.group(0))
        except json.JSONDecodeError:
            pass
    
    # Strategie 4: Text mit洋溢-Regeln bereinigen
    cleaned = response_text.replace("'", '"').replace("None", "null")
    cleaned = re.sub(r'(\w+):', r'"\1":', cleaned)  # Keys quotieren
    
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError as e:
        raise ValueError(f"Konnte JSON nicht parsen: {response_text[:100]}")

2. Rate-Limit-Überschreitung bei Batch-Verarbeitung

import time
from ratelimit import limits, sleep_and_retry

class RateLimitedExtractor:
    """Extractor mit automatischer Rate-Limit-Behandlung."""
    
    def __init__(self, api_key: str, requests_per_minute: int = 60):
        self.api_key = api_key
        self.requests_per_minute = requests_per_minute
        self.base_url = "https://api.holysheep.ai/v1/chat/completions"
        self.delay = 60 / requests_per_minute
    
    @sleep_and_retry
    @limits(calls=1, period=1)  # Max 1 Request pro Sekunde
    def extract_with_backoff(self, text: str, fields: list, max_retries: int = 3) -> dict:
        """Extrahiert mit exponentieller Backoff-Wiederholung."""
        
        for attempt in range(max_retries):
            try:
                headers = {
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                }
                
                payload = {
                    "model": "deepseek-v3.2",
                    "messages": [
                        {"role": "system", "content": f"JSON mit Feldern: {', '.join(fields)}"},
                        {"role": "user", "content": text}
                    ],
                    "max_tokens": 500
                }
                
                response = requests.post(
                    self.base_url, 
                    headers=headers, 
                    json=payload,
                    timeout=30
                )
                
                if response.status_code == 429:
                    wait_time = 2 ** attempt * 10  # 10, 20, 40 Sekunden
                    print(f"Rate-Limit, warte {wait_time}s...")
                    time.sleep(wait_time)
                    continue
                
                response.raise_for_status()
                return json.loads(response.json()["choices"][0]["message"]["content"])
                
            except requests.exceptions.RequestException as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
        
        raise Exception("Max retries exceeded")

3. Inkonsistente Feldnamen bei unterschiedlichen Texten

# PROBLEM: Unterschiedliche Benennung desselben Feldes
("E-Mail", "EMAIL", "email", "mail", "Mail-Adresse")

STANDARDIZED_FIELDS = {
    # E-Mail-Felder
    "email": ["email", "e-mail", "mail", "mail_address", "email_address", "kontakt_email"],
    # Namens-Felder  
    "name": ["name", "full_name", "fullname", "bezeichnung", "personenname"],
    # Datums-Felder
    "datum": ["datum", "date", "erstellungsdatum", "created_at", "timestamp"],
    # Betrag-Felder
    "betrag": ["betrag", "amount", "summe", "total", "gesamtsumme", "preis"],
    # Adress-Felder
    "adresse": ["adresse", "address", "standort", "location", "anschrift"]
}

def normalize_fields(extracted: dict) -> dict:
    """Normalisiert Feldnamen auf einheitliche Standards."""
    
    normalized = {}
    
    for field_name, value in extracted.items():
        # Finde das Standardfeld
        standard_key = None
        for std, variants in STANDARDIZED_FIELDS.items():
            if field_name.lower() in variants:
                standard_key = std
                break
        
        if standard_key is None:
            # Kein Mapping gefunden - behalte Original
Verwandte Ressourcen
📚 KI API Tutorials
💰 Preise ansehen
📖 Entwickler-Dokumentation
🚀 Kostenlos registrieren
Verwandte Artikel
Sora API Video-Generierung: Komplette Anleitung für Einsteig
MCP 协议性能基准测试：延迟、吞吐量与并发极限 — Praxisleitfaden 2026
Agentic RAG 2026: Agent Dynamic Decision Retrieval Paths