Datenverarbeitung und Validierung - Einführung in Python

In realen Projekten machen das Laden, Bereinigen und Validieren von Daten oft den Großteil des Codes aus.

!uv venv --quiet
!uv sync --quiet
!uv pip --quiet install "omegaconf" "jmespath" "pydantic[email]" "validators" "pydantic-ai" "huggingface_hub" "llama-cpp-python"

CSV¶

Für flache, tabellarische Daten sind CSV und Excel der Standard. Während csv in der Standardbibliothek ist, benötigt Excel externe Hilfe (openpyxl oder pandas).

import csv

# CSV schreiben
with open("users.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["id", "name", "email"])
    writer.writeheader()
    writer.writerow({"id": 1, "name": "Alice", "email": "alice@example.com"})

# CSV lesen
with open("users.csv", "r", encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"Geladen: {row['name']}")

# Hinweis zu Excel:
# import pandas as pd
# df = pd.read_excel('data.xlsx') # Benötigt openpyxl

Geladen: Alice

Konfigurationsmanagement¶

YAML ist ideal für hierarchische Konfigurationen. OmegaConf (sieh den Abschnitt zu Hydra im Testing-Kapitel) bietet zusätzlich einen objektorientierten Zugriff und Merging-Funktionen.

import yaml
from omegaconf import OmegaConf

yaml_data = """
server:
  port: 8080
  host: localhost
database:
  driver: postgres
"""

# Mit pyyaml laden
config_dict = yaml.safe_load(yaml_data)

# Mit OmegaConf für hierarchisches Management nutzen
conf = OmegaConf.create(config_dict)
print(f"Port: {conf.server.port}")

# Mergen von Defaults und Overrides
overrides = OmegaConf.create({"server": {"port": 9000}})
final_conf = OmegaConf.merge(conf, overrides)
print(f"Neuer Port: {final_conf.server.port}")

Port: 8080
Neuer Port: 9000

JSON Queries mit jmespath¶

Wenn REST-APIs tief verschachtelte JSON-Strukturen liefern, wird der Zugriff über Dictionaries (data['a']['b'][0]['c']) schnell unübersichtlich. jmespath erlaubt deklarative Abfragen.

import jmespath

complex_json = {
    "users": [
        {"id": 1, "info": {"email": "alice@example.com", "active": True}},
        {"id": 2, "info": {"email": "bob_at_example.com", "active": False}},
        {"id": 3, "info": {"email": "charlie@example.com", "active": True}},
    ]
}

# Extrahiere Emails aller aktiven Nutzer
query = "users[?info.active == `true`].info.email"
active_emails = jmespath.search(query, complex_json)
print(f"Aktive Emails: {active_emails}")

Aktive Emails: ['alice@example.com', 'charlie@example.com']

Datenvalidierung mit Pydantic¶

Pydantic ist das Standardwerkzeug für Datenvalidierung. Im Gegensatz zu dataclasses erzwingt Pydantic Typen zur Laufzeit und führt automatisches Casting (Coercion) durch.

Abgrenzung zu Dataclasses¶

Dataclasses: Primär zur Reduktion von Boilerplate (Record-Typen). Keine Laufzeitvalidierung.
Pydantic: Validierung, Parsing, Fehlerbehandlung.

Tutorial: Pydantic Get Started

from pydantic import BaseModel, EmailStr, Field, ValidationError


class User(BaseModel):
    id: int
    name: str = Field(min_length=2)
    email: EmailStr


try:
    # Beachte: '1' (String) wird automatisch zu 1 (int) konvertiert
    user = User(id="1", name="Alice", email="alice@example.com")
    print(user.model_dump())
except ValidationError as e:
    print(e.json())

{'id': 1, 'name': 'Alice', 'email': 'alice@example.com'}

Oxidization: Pydantic V2¶

Pydantic V2 wurde in Rust neu geschrieben (pydantic-core). Dieser Prozess (“Oxidization”) macht die Validierung um Faktoren schneller und zeigt den Trend im Python-Ökosystem: Performance-kritische Teile werden in Rust implementiert, während die API in Python bleibt.

Weitere Validierungstools¶

Validators: Für einfache Checks ohne Objektmodell.
Great Expectations (GX): Für die Validierung ganzer Daten-Pipelines (Data Quality).

import validators

# Einfacher Check
email_ok = validators.email("someone@example.com")
print(f"Email valid: {email_ok}")

Email valid: True

Serialisierung: Protobuf vs. Pickle¶

Pickle: In Python eingebaut, sehr mächtig (kann fast alles serialisieren), aber unsicher (Remote Code Execution möglich) und sprachspezifisch.
Protobuf: Sprachübergreifend, performant, Schema-basiert. Ideal für Microservices.

Ressourcen¶

Strukturierte KI-Datentransformation mit pydantic-ai¶

Ein grundlegendes Problem beim Einsatz von LLMs in produktiver Software ist die Unvorhersehbarkeit des Text-Outputs. Wenn eine API strukturierte Daten (wie JSON) erwartet, reicht ein normaler Prompt oft nicht aus, um Typ-Sicherheit zu garantieren.

Hier setzt pydantic-ai an. Es nutzt die Bibliothek pydantic, um Agenten zu zwingen, Antworten exakt nach einem vorgegebenen Schema zurückzugeben. Das Framework validiert den Output des LLMs automatisch; ist er fehlerhaft, bittet pydantic-ai das LLM um Korrektur.

Kernkonzepte:

Type-Safe: Die Rückgabe der KI ist kein einfacher String mehr, sondern ein instanziiertes Pydantic-Objekt.
Model Agnostic: Es lässt sich mit Modellen von Google, OpenAI oder lokalen Modellen nutzen.
Tools als Funktionen: Man kann dem Agenten reguläre Python-Funktionen als Werkzeuge mitgeben, die er bei Bedarf aufruft.

Beispiel mit Hilfe einer (bezahlten) API¶

%%writefile pydantic_ai_example.py
from pydantic import BaseModel
from pydantic_ai import Agent

class UserProfile(BaseModel):
    name: str
    age: int
    hobbies: list[str]

agent = Agent(
    'gemini-1.5-pro',
    result_type=UserProfile,
    system_prompt="Extrahiere die Benutzerdaten aus dem Text."
)

# 3. Ausführung (In einem echten Skript ggf. asynchron)
result = agent.run_sync("Das ist Thomas, er ist 35 Jahre alt und geht gerne wandern und liest Science-Fiction.")
print(result.data.hobbies) # Output: ['Wandern', 'Science-Fiction']

Writing pydantic_ai_example.py

Dieser Code führt zu einem Fehler, wenn kein Google-API-Key konfiguriert wurde:

!uv run python pydantic_ai_example.py

Testing mit Mockup¶

Wir können natürlich mit Mockups arbeiten:

%%writefile pydantic_ai_mockup.py
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic import BaseModel

class UserProfile(BaseModel):
    name: str
    age: int
    hobbies: list[str]

agent = Agent(
    TestModel(), 
    output_type=UserProfile,
    system_prompt="Extrahiere die Benutzerdaten aus dem Text."
)

result = agent.run_sync("Das ist Thomas, er ist 35 Jahre alt...")
print(result.output)

Writing pydantic_ai_mockup.py

!uv run python pydantic_ai_mockup.py

name='a' age=0 hobbies=['a']

Wenn man anstelle des Mockups ein richtiges LLM (oder auch ein relativ kleines, 20 MB könnten genügen) einsetzt, generiert das LLM einen JSON-Output, der von Pydantic in ein Python-Objekt umgewandelt wird. Das sähe dann so aus:

name='Thomas' age=35 hobbies=[]

Beispiel mit kleinem lokalen Modell¶

Wir nutzen llama.cpp über llama-cpp-python und laden ein Modell (bezogen über Huggingface) lokal:

%%writefile downloader.py
from huggingface_hub import hf_hub_download

model_path = hf_hub_download(
    repo_id="wattsonandme/survival-qwen-0.5b",
	filename="v3-survival-qwen-0.5b-q4_k_m.gguf",
)
print(model_path)

Writing downloader.py

!uv run python downloader.py

Warning: You are sending unauthenticated requests to the HF Hub. Please set a HF_TOKEN to enable higher rate limits and faster downloads.
/home/konrad/.cache/huggingface/hub/models--wattsonandme--survival-qwen-0.5b/snapshots/0ed76b2ca44add906844a758475f36920e127636/v3-survival-qwen-0.5b-q4_k_m.gguf

Das ist optional, man kann auch die Datei .gguf manuell herunterladen und auf huggingface_hub verzichten.

%%writefile llama_example.py

from llama_cpp import Llama

GGUF_FILE = "/home/konrad/.cache/huggingface/hub/models--wattsonandme--survival-qwen-0.5b/snapshots/0ed76b2ca44add906844a758475f36920e127636/v3-survival-qwen-0.5b-q4_k_m.gguf"

def run_local_chat(model_path: str, prompt: str) -> str:
    # 1. Modell laden
    llm = Llama(
        model_path=model_path,
        n_ctx=32768,
        verbose=False
    )

    # 2. Inferenz über die Chat-API (wendet das interne Template an)
    response = llm.create_chat_completion(
        messages=[
            {"role": "system", "content": "You are a precise and helpful assistant."},
            {"role": "user", "content": prompt}
        ],
        max_tokens=100,
        temperature=0.1 # Niedrige Temperatur für deterministischere Mathe-Antworten
    )

    # 3. Output extrahieren
    return response["choices"][0]["message"]["content"].strip()

if __name__ == "__main__":
    test_prompt = "What is 2 + 3 expressed as a single number? Answer a number only."
    print(run_local_chat(GGUF_FILE, test_prompt))

Writing llama_example.py

!uv run python llama_example.py

2 + 3 = 5. This calculation is not accurate because it is incorrect in the first half of the calculation. It is important to verify the results against the original calculations. Always verify with multiple methods to ensure results are accurate.

Hierbei sehen wir ganz schön die typischen Schwierigkeiten mit LLM-gestützter Datenextraktion: Obwohl im Prompt um ein bestimmtes Format gebettelt wurde mit “Answer a number only” enthält die Antwort einen Freitext, der zudem sachlich falsch ist.

Wir kombinieren jetzt Llama.cpp mit Pydantic:

%%writefile llama_pydantic.py
import json
from pydantic import BaseModel
from llama_cpp import Llama

GGUF_FILE = "/home/konrad/.cache/huggingface/hub/models--wattsonandme--survival-qwen-0.5b/snapshots/0ed76b2ca44add906844a758475f36920e127636/v3-survival-qwen-0.5b-q4_k_m.gguf"

class UserProfile(BaseModel):
    name: str
    age: int
    hobbies: list[str]

def extract_profile(model_path: str, text: str) -> UserProfile:
    # 1. Modell laden
    llm = Llama(model_path=model_path, n_ctx=1024, verbose=False)

    # 2. Inferenz mit Grammar/Schema Enforcement
    response = llm.create_chat_completion(
        messages=[
            {"role": "system", "content": "Extract the user profile data as JSON."},
            {"role": "user", "content": text}
        ],
        # Hier zwingen wir Llama, exakt das Pydantic-Schema auszuspucken
        response_format={
            "type": "json_object",
            "schema": UserProfile.model_json_schema()
        },
        temperature=0.1,
        max_tokens=200, # um Endlosigkeit zu vermeiden
        repeat_penalty=1.2
    )

    # 3. Rohen JSON-String in Pydantic-Objekt umwandeln
    json_string = response["choices"][0]["message"]["content"]
    return UserProfile.model_validate_json(json_string)

if __name__ == "__main__":
    text = "This is Thomas, Thomas is 35 years old and likes hiking and reading."
    
    result = extract_profile(GGUF_FILE, text)
    print(result.hobbies) # Gewünschter Output: ['hiking', 'reading']

Writing llama_pydantic.py

!uv run python llama_pydantic.py

llama_context: n_ctx_seq (1024) < n_ctx_train (32768) -- the full capacity of the model will not be utilized
['hike', 'work', 'watch', 'museum', 'jambalunka']

Das Ergebnis ist immer noch unterirdisch, kann aber durch ein Modell mit mehr Parametern (mehr als 400 MB) verbessert werden.

Komplexe Datentypen¶

Für bestimmte Datentypen/Datenstrukturen gibt es spezialisierte Lösungen. Hier sind ein paar Stichpunkte:

Graphen bzw. Netzwerke mit networkx (klassich) oder dem Drop-in Ersatz rustworkx (https://www.rustworkx.org/networkx.html).
Parsen von Webseiten mit selectolax oder dem älteren beautifulsoup oder parsel.
Klicken auf Webseiten mit playwright oder dem älteren selenium.
HTTP-Zugriffe mit httpx sowie Retries mit stamina (ein Wrapper um tenacity).
Bildverarbeitung mit pillow (Ersatz für PIL = Python Imaging Library) und opencv für komplexere Transformationen und einfaches Machine Learning.
Textverarbeitungstools (trrex, flashtext, hyperscan) und für einfache NLP-Aufgaben spaCy (eher nicht die älteren NLTK und GenSim).
Lese-, Schreib- und Extraktionsvorgänge bei PDFs mit pdfminer.six, pypdf und pymupdf (import fitz) sowie wrapper pdfplumber.
Datenformate wie Tabellen oder Datenbanken einlesen mit Polars, siehe Big-Data Kapitel.
GIS/GNSS-Daten mit geopandas bzw. geopolars und shapely (siehe auch pythongis.org )

!rm -f downloader.py llama_example.py llama_pydantic.py pydantic_ai_example.py pydantic_ai_mockup.py users.csv # Aufräumen