0Pricing
AI Engineering Academy · Lektion

Schema-Weiterentwicklung und Abwärtskompatibilität

Verwalten Sie inkompatible Schemaänderungen in langfristig laufenden Extraktionspipelines, indem Sie Schemas versionieren, historische Extraktionen migrieren und während der Übergänge eine parallele Validierung ausführen.

Schema-Weiterentwicklung und Abwärtskompatibilität ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Warum sich Schemas im Laufe der Zeit ändern

Extraktionsschemas sind nicht statisch. Geschäftsanforderungen entwickeln sich weiter, neue Dokumenttypen kommen hinzu, und Sie entdecken Felder, die Sie von Anfang an hätten erfassen sollen. Das Ändern eines Schemas in einer aktiven Pipeline führt zu einem Abwärtskompatibilitätsproblem: Bereits extrahierte Datensätze verwenden das alte Schema, während neue Datensätze das neue verwenden. Diese Umstellung sicher zu verwalten, ist der Kern der Schemaentwicklung.

Schemas versionieren

Weisen Sie jedem Schema eine Versionsnummer zu und speichern Sie diese zusammen mit jedem extrahierten Datensatz. Erhöhen Sie die Version, wenn Sie das Schema ändern. So können Sie Datensätze nach Schema-Version abfragen, Migrationen für alte Datensätze ausführen und für jede Version eine separate Validierungslogik beibehalten. Ein einfaches Zeichenkettenfeld schema_version in jedem Ausgabemodell ist ausreichend.

from pydantic import BaseModel
from typing import Literal

class InvoiceV1(BaseModel):
    schema_version: Literal['1.0'] = '1.0'
    vendor: str
    total_amount: float

class InvoiceV2(BaseModel):
    schema_version: Literal['2.0'] = '2.0'
    vendor: str
    vendor_tax_id: str | None = None  # new field
    total_amount: float
    currency: str = 'USD'  # new field with default

Additive und nicht abwärtskompatible Änderungen

Additive Änderungen sind sicher: Das Hinzufügen eines Optional-Felds oder eines Felds mit Standardwert beschädigt weder alten Extraktionscode noch alte Datensätze. Nicht abwärtskompatible Änderungen sind riskant: Das Umbenennen eines Felds, die Änderung eines Typs von string zu int oder das Entfernen eines Felds beschädigt nachgelagerte Verbraucher. Bevorzugen Sie immer additive Änderungen. Wenn eine nicht abwärtskompatible Änderung unvermeidbar ist, erstellen Sie eine neue Major-Schema-Version und führen Sie die Migration kontrolliert durch.

# Safe: additive change - add optional field
class ProductV2(BaseModel):
    name: str
    price: float
    sku: str | None = None   # NEW optional field - backward safe
    category: str = 'general'  # NEW with default - backward safe

# Risky: breaking change - rename or retype
# class ProductV2(BaseModel):
#     product_name: str  # RENAMED from name - breaks consumers
#     price_cents: int   # RETYPED from float - breaks data

Schema-Version in der Datenbank speichern

Nehmen Sie die Schema-Version in Ihre Tabelle mit den Extraktionsergebnissen auf, damit Sie immer wissen, welche Version den jeweiligen Datensatz erzeugt hat. Eine häufige Vorgehensweise ist eine jsonb-Spalte mit den vollständigen extrahierten Daten sowie eine Textspalte schema_version. So können Sie versionsabhängige Abfragen schreiben und ältere Datensätze gezielt in Zeitfenstern mit geringem Datenverkehr migrieren.

-- PostgreSQL table design
CREATE TABLE extractions (
    doc_id TEXT PRIMARY KEY,
    schema_version TEXT NOT NULL,
    extracted_data JSONB NOT NULL,
    extracted_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_schema_version ON extractions(schema_version);

-- Query old records needing migration
SELECT doc_id, extracted_data
FROM extractions
WHERE schema_version = '1.0'
LIMIT 1000;

Migrationsskripte schreiben

Schreiben Sie für jeden Übergang zwischen Schema-Versionen ein Migrationsskript, das alte Datensätze liest, in das neue Format umwandelt und mit der neuen Version zurückschreibt. Führen Sie Migrationen in kleinen Batches mit Transaktionen aus, damit ein Fehler die Datenbank nicht in einem teilweise migrierten Zustand hinterlässt. Halten Sie das alte Schema immer verfügbar, bis die Migration nachweislich abgeschlossen ist.

import asyncpg
import json

async def migrate_v1_to_v2(pool, batch_size=100):
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            'SELECT doc_id, extracted_data FROM extractions WHERE schema_version=$1 LIMIT $2',
            '1.0', batch_size
        )
        for row in rows:
            old = row['extracted_data']
            new_data = {
                'schema_version': '2.0',
                'vendor': old['vendor'],
                'vendor_tax_id': None,  # unknown for old records
                'total_amount': old['total_amount'],
                'currency': 'USD'  # assume USD for old records
            }
            await conn.execute(
                'UPDATE extractions SET extracted_data=$1, schema_version=$2 WHERE doc_id=$3',
                json.dumps(new_data), '2.0', row['doc_id']
            )

Parallele Validierung während der Umstellung

Führen Sie während einer Schema-Migration eine parallele Validierung durch: Extrahieren Sie für eine Stichprobe eingehender Dokumente gleichzeitig mit dem alten und dem neuen Schema. Vergleichen Sie die Ergebnisse, um zu überprüfen, dass das neue Schema alles erfasst, was das alte erfasst hat, und zusätzlich die neuen Felder. Deaktivieren Sie das alte Schema erst, wenn die parallele Validierung bei einer statistisch signifikanten Stichprobe eine stabile Übereinstimmung zeigt.

async def parallel_validate(text: str) -> dict:
    v1_result, v2_result = await asyncio.gather(
        extract_with_schema(text, InvoiceV1),
        extract_with_schema(text, InvoiceV2)
    )
    discrepancy = (
        v1_result.vendor != v2_result.vendor or
        abs(v1_result.total_amount - v2_result.total_amount) > 0.01
    )
    if discrepancy:
        log_discrepancy(text, v1_result, v2_result)
    return {'v1': v1_result, 'v2': v2_result, 'discrepancy': discrepancy}

Feature Flags für die Schema-Einführung

Verwenden Sie Feature Flags, um zu steuern, wann Ihre Pipeline vom alten auf das neue Schema umschaltet. So können Sie das neue Schema schrittweise für einen Prozentsatz des Datenverkehrs einführen, Fehlerraten überwachen und bei Problemen sofort zurücksetzen – ohne den Code erneut bereitzustellen. Sowohl Dienste wie LaunchDarkly als auch eine einfache Datenbankzeile eignen sich dafür.

import os

def get_active_schema():
    version = os.environ.get('EXTRACTION_SCHEMA_VERSION', '1.0')
    schemas = {
        '1.0': InvoiceV1,
        '2.0': InvoiceV2,
    }
    return schemas.get(version, InvoiceV1)

async def extract_document(text: str):
    SchemaClass = get_active_schema()
    return await extract_with_schema(text, SchemaClass)

Kompatibilität von Verbrauchern mit Union-Typen

Nachgelagerte Verbraucher, die extrahierte Daten lesen, müssen mehrere Schema-Versionen zuverlässig verarbeiten können. Verwenden Sie in Ihrem Verbrauchercode eine discriminated union, die anhand des Felds schema_version die passende Parsing-Logik auswählt. Das ist robuster als bedingte if-else-Ketten und lässt sich leichter erweitern, wenn Version 3 hinzukommt.

from pydantic import BaseModel
from typing import Union, Annotated
from typing import Literal

def parse_extraction(raw: dict) -> Union[InvoiceV1, InvoiceV2]:
    version = raw.get('schema_version', '1.0')
    if version == '1.0':
        return InvoiceV1(**raw)
    elif version == '2.0':
        return InvoiceV2(**raw)
    else:
        raise ValueError(f'Unknown schema version: {version}')

Schemaänderungen vor der Bereitstellung testen

Führen Sie ein neues Schema vor der Bereitstellung mit Ihrem gesamten Regressionstestsatz aus: einer kuratierten Sammlung repräsentativer Dokumente mit bekannten erwarteten Ausgaben. Vergleichen Sie die F1-Scores für jedes Feld zwischen dem alten und dem neuen Schema. Eine Verschlechterung des F1-Scores für ein beliebiges Feld bedeutet, dass die Beschreibung des neuen Schemas das Modell verwirrt hat – korrigieren Sie die Feldbeschreibung vor der Veröffentlichung.

def eval_schema_on_test_set(test_cases: list, SchemaClass) -> dict:
    field_f1 = {}
    for case in test_cases:
        result = extract_with_schema(case['text'], SchemaClass)
        for field in case['expected']:
            expected = case['expected'][field]
            actual = getattr(result, field, None)
            # Update precision/recall counters
            update_metrics(field_f1, field, expected, actual)
    return {k: compute_f1(v) for k, v in field_f1.items()}

Schema-Ablösung verwalten

Sobald eine Schema-Version nicht mehr für neue Extraktionen verwendet wird, können Sie sie als veraltet kennzeichnen. Das bedeutet: Keine neuen Datensätze mehr in dieser Version akzeptieren, alte Datensätze lesbar halten und einen Termin für die Ablösung festlegen, an dem alte Datensätze migriert oder archiviert werden. Dokumentieren Sie die Ablösung in einem Changelog, damit alle Verbraucher wissen, dass sie ihren Parsing-Code aktualisieren müssen.

DEPRECATED_VERSIONS = {'1.0'}
SUNSET_DATE = '2026-09-01'

def warn_if_deprecated(version: str):
    if version in DEPRECATED_VERSIONS:
        import warnings
        warnings.warn(
            f'Schema version {version} is deprecated. '
            f'It will be removed after {SUNSET_DATE}. '
            'Migrate consumers to version 2.0.',
            DeprecationWarning,
            stacklevel=2
        )

Changelog und Kommunikation

Jede Schemaänderung muss von einem Changelog-Eintrag begleitet werden, der beschreibt, was geändert wurde, warum die Änderung vorgenommen wurde, wie die Migration funktioniert und welche Auswirkungen zu erwarten sind. Teilen Sie Changelog-Einträge vor der Bereitstellung der Änderung mit allen Teams, die die extrahierten Daten verwenden. Viele Katastrophen bei Schema-Migrationen entstehen nicht durch technische Fehler, sondern durch Verbraucher, die nicht über eine bevorstehende Änderung informiert wurden.

# CHANGELOG.md entry format:
# ## Schema v2.0 (2026-07-01)
# ### Changes
# - ADDED: vendor_tax_id (Optional[str]) - VAT/EIN extracted from header
# - ADDED: currency (str, default='USD') - detected from symbol/code
# ### Migration
# Run: python scripts/migrate_v1_to_v2.py --batch-size=500
# ### Consumers
# - billing-service: update parse_extraction() to handle v2
# - audit-service: query now supports currency filter

Kurzer Test

Testen Sie Ihr Verständnis der Schemaentwicklung und Abwärtskompatibilität in Extraktionspipelines.

Zusammenfassung der Lektion

In dieser Lektion haben Sie Folgendes gelernt: Schema-Versionierung speichert neben jedem extrahierten Datensatz eine Versionskennung, sodass Sie gezielt migrieren können, additive Änderungen sind sicher, während das Umbenennen oder Ändern von Feldtypen eine sorgfältige Migration erfordert, und die parallele Validierung ermöglicht es Ihnen, das neue Schema zu überprüfen, bevor Sie das alte außer Betrieb nehmen. Als Nächstes messen wir die LLM-Latenz mit den Metriken TTFT und TPOT.

Häufig gestellte Fragen

Ist die Lektion „Schema-Weiterentwicklung und Abwärtskompatibilität“ kostenlos?

Ja — der vollständige Text von „Schema-Weiterentwicklung und Abwärtskompatibilität“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Schema-Weiterentwicklung und Abwärtskompatibilität“?

Verwalten Sie inkompatible Schemaänderungen in langfristig laufenden Extraktionspipelines, indem Sie Schemas versionieren, historische Extraktionen migrieren und während der Übergänge eine parallele… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Engineering Academy zu starten?

Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Schema-Weiterentwicklung und Abwärtskompatibilität“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?

Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Instructor: Typisierte Extraktion mit Pydantic
  2. Unvollständige und fehlende Daten verarbeiten
  3. Batch-Verarbeitung mit Async und Queues
  4. Schema-Weiterentwicklung und Abwärtskompatibilität
← Zurück zu AI Engineering Academy