模式演进与向后兼容
通过为模式进行版本管理、迁移历史提取结果并在过渡期间运行并行验证，管理长期运行的提取流程中的破坏性模式变更。
模式演进与向后兼容 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
Why Schemas Change Over Time
Extraction schemas are not static. Business requirements evolve, new document types appear, and you discover fields you should have captured from the start. Changing a schema in a live pipeline creates a backward compatibility problem: existing extracted records use the old schema, while new records use the new one. Managing this transition safely is what schema evolution is about.
Versioning Your Schemas
Assign a version number to each schema and store it alongside every extracted record. When you change the schema, increment the version. This lets you query records by schema version, run migrations on old records, and maintain separate validation logic for each version. A simple string field schema_version in every output model is sufficient.
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 defaultAdditive vs Breaking Changes
Additive changes are safe: adding an Optional field or a field with a default does not break old extraction code or old records. Breaking changes are risky: renaming a field, changing a type from string to int, or removing a field will break downstream consumers. Always prefer additive changes. When a breaking change is unavoidable, create a new major schema version and migrate in a controlled way.
# 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 dataStoring Schema Version in the Database
Include the schema version in your extraction results table so you always know which version produced each record. A jsonb column storing the full extracted data plus a schema_version text column is a common pattern. This lets you write version-aware queries and selectively migrate older records during low-traffic windows.
-- 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;Writing Migration Scripts
Write a migration script for each schema version transition that reads old records, transforms them to the new format, and writes them back with the new version. Run migrations in small batches with transactions so a failure does not leave the database in a half-migrated state. Always keep the old schema available until migration is verified complete.
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']
)Parallel Validation During Transitions
During a schema migration, run parallel validation: extract with both the old and new schema simultaneously for a sample of incoming documents. Compare the results to verify the new schema captures everything the old one did plus the new fields. Only retire the old schema after parallel validation shows stable parity on a statistically significant sample.
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 for Schema Rollout
Use feature flags to control when your pipeline switches from the old schema to the new one. This lets you gradually roll out the new schema to a percentage of traffic, monitor error rates, and instantly roll back if something goes wrong — without redeploying code. Feature flag services like LaunchDarkly or a simple database row both work.
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)Consumer Compatibility with Union Types
Downstream consumers that read extracted data need to handle multiple schema versions gracefully. Use a discriminated union in your consumer code that selects the correct parsing logic based on the schema_version field. This is more robust than writing conditional if-else chains and easier to extend when version 3 arrives.
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}')Testing Schema Changes Before Deployment
Before deploying a new schema, run it against your entire regression test set: a curated collection of representative documents with known expected outputs. Compare F1 scores for each field between the old and new schema. A regression in F1 for any field means the new schema description confused the model — fix the field description before shipping.
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()}Handling Schema Deprecation
Once a schema version is no longer used for new extractions, you can deprecate it. Deprecation means: stop accepting new records in that version, keep old records readable, and schedule a sunset date when old records will be migrated or archived. Document the deprecation in a changelog so all consumers know to upgrade their parsing code.
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 and Communication
Every schema change must be accompanied by a changelog entry that describes what changed, why, migration instructions, and the expected impact. Share changelog entries with all teams that consume extracted data before deploying the change. Many schema migration disasters happen not from technical failures but from consumers that were not informed a change was coming.
# 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 filterQuick Check
Test your understanding of schema evolution and backward compatibility in extraction pipelines.
Lesson Recap
In this lesson you learned: schema versioning stores a version identifier alongside every extracted record so you can migrate selectively, additive changes are safe while renaming or retyping fields requires careful migration, and parallel validation lets you verify the new schema before retiring the old one. Next up we measure LLM latency with TTFT and TPOT metrics.
常见问题解答
「模式演进与向后兼容」课时是免费的吗？
是的 — 「模式演进与向后兼容」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「模式演进与向后兼容」这节课中我会学到什么？
通过为模式进行版本管理、迁移历史提取结果并在过渡期间运行并行验证，管理长期运行的提取流程中的破坏性模式变更。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「模式演进与向后兼容」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。