验证并重试错误输出
实现验证层,根据业务规则检查提取的数据;验证失败时自动结合纠正反馈进行重试,并记录失败模式。
验证并重试错误输出 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 4 节课,共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分,你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
为什么 LLM 输出需要验证
即使使用结构化输出和 Pydantic 架构,LLM 抽取仍可能产生语法有效但语义错误的输出。例如,置信度分数为 1.5(超出 0 到 1 的范围)、价格为 -99.99、无法解析的日期字符串,或包含字母的电话号码——这些内容都能通过 JSON 解析,却无法通过您的业务规则。
验证是独立于抽取的另一项职责。抽取要回答的是:“我们是否获得了结构化数据?”验证要回答的是:“这些结构化数据是否正确且可用?”对于生产级流水线,这两个层次都是必需的。您可以将其理解为两阶段过滤器:LLM 负责抽取,验证器负责接受或拒绝。
验证层次
健壮的输出验证系统会在多个层次上运行:
- 架构验证(Pydantic):字段类型正确、必填字段存在、枚举值符合允许范围——由结构化输出自动处理
- 格式验证:电话号码符合正则表达式,电子邮件地址有效,日期可解析,金额处于合理范围内
- 业务逻辑验证:发票总额等于行项目之和,结束日期晚于开始日期,数量为正整数
- 跨字段验证:一个字段的值取决于另一个字段的值(例如,折扣百分比不能超过 100)
- 语义验证:抽取出的公司名称与数据库中的已知公司匹配
用于格式检查的 Pydantic 验证器
Pydantic 的 field_validator 装饰器允许您添加自定义验证逻辑,该逻辑会在模型实例化时运行。您可以使用它执行格式级检查,例如对电话号码和电子邮件进行正则表达式验证、解析日期,以及检查数值字段是否处于合理范围内。
from pydantic import BaseModel, Field, field_validator
from typing import Optional
import re
from datetime import datetime
class ExtractedInvoice(BaseModel):
vendor: str
invoice_number: Optional[str]
amount: float = Field(gt=0, description='Must be positive')
currency: str = Field(min_length=3, max_length=3)
invoice_date: str
@field_validator('currency')
@classmethod
def currency_must_be_uppercase(cls, v):
return v.upper()
@field_validator('invoice_date')
@classmethod
def parse_date(cls, v):
# Try to parse common date formats
for fmt in ('%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y', '%B %d, %Y'):
try:
datetime.strptime(v, fmt)
return v
except ValueError:
continue
raise ValueError(f'Cannot parse date: {v}')
@field_validator('amount')
@classmethod
def reasonable_amount(cls, v):
if v > 10_000_000:
raise ValueError(f'Amount {v} seems unreasonably large. Flag for review.')
return round(v, 2)带纠正反馈的重试模式
验证失败时,最有效的恢复策略是带纠正反馈重试:将验证错误消息作为上下文发回模型,解释哪里出了问题,并要求模型只修复失败的字段。这样,模型就能获得纠正输出所需的信息,而不是在毫无依据的情况下盲目重试。
import openai
from pydantic import BaseModel, ValidationError, Field
client = openai.OpenAI()
class PriceExtraction(BaseModel):
product: str
price_usd: float = Field(gt=0, lt=100000)
quantity: int = Field(ge=1)
def extract_with_retry(text: str, max_retries: int = 3) -> PriceExtraction:
messages = [
{'role': 'system', 'content': 'Extract product pricing information.'},
{'role': 'user', 'content': text}
]
for attempt in range(max_retries):
result = client.beta.chat.completions.parse(
model='gpt-4o-mini',
messages=messages,
response_format=PriceExtraction
)
msg = result.choices[0].message
if msg.refusal:
raise ValueError(f'Model refused: {msg.refusal}')
try:
return msg.parsed # Pydantic validates on parse
except ValidationError as e:
if attempt == max_retries - 1:
raise
# Add corrective feedback for the next attempt
messages.append({'role': 'assistant', 'content': msg.content})
messages.append({'role': 'user', 'content': f'The previous extraction failed validation: {e}\nPlease correct and try again.'})
print(f'Attempt {attempt+1} failed. Retrying with feedback...')业务逻辑验证
业务逻辑验证会检查跨越多个字段的属性,或依赖外部数据源的属性。Pydantic 的 model_validator 会在所有字段级验证器运行后执行,并且可以访问已完整填充的模型,因此适合用于跨字段检查。
from pydantic import BaseModel, Field, model_validator
from typing import List
class LineItem(BaseModel):
description: str
quantity: int = Field(ge=1)
unit_price: float = Field(ge=0)
line_total: float
@model_validator(mode='after')
def check_line_total(self):
expected = round(self.quantity * self.unit_price, 2)
actual = round(self.line_total, 2)
if abs(expected - actual) > 0.02: # Allow 2-cent rounding tolerance
raise ValueError(
f'Line total {actual} does not match quantity*price={expected}'
)
return self
class Invoice(BaseModel):
line_items: List[LineItem]
subtotal: float
tax: float
total: float
@model_validator(mode='after')
def check_invoice_total(self):
expected_total = round(self.subtotal + self.tax, 2)
if abs(expected_total - round(self.total, 2)) > 0.02:
raise ValueError(
f'Invoice total {self.total} != subtotal+tax ({expected_total})'
)
return self记录验证失败
每次验证失败都表明您的流水线在某处出现了问题。请记录每次失败的以下信息:输入文本(出于隐私考虑,也可以记录其哈希值)、抽取出的输出、具体的验证错误以及尝试次数。汇总这些日志,以识别系统性模式——模型是否总是将某个字段弄错?是否有某类文档会导致失败?这些数据可以推动有针对性的提示词改进。
import logging
from pydantic import ValidationError
logger = logging.getLogger(__name__)
def extract_with_logging(text: str, doc_id: str) -> dict:
result = None
for attempt in range(3):
try:
result = run_extraction(text) # Your extraction function
logger.info('Extraction success', extra={
'doc_id': doc_id,
'attempt': attempt + 1
})
return result
except ValidationError as e:
logger.warning('Validation failure', extra={
'doc_id': doc_id,
'attempt': attempt + 1,
'errors': e.errors(),
'error_count': len(e.errors())
})
# All retries failed
logger.error('Extraction failed after max retries', extra={'doc_id': doc_id})
return {'error': 'extraction_failed', 'doc_id': doc_id}
def run_extraction(text):
pass # Placeholder for actual extraction logic置信度分数和阈值
在抽取架构中添加 confidence 字段,并指示模型为每项抽取结果评定 0 到 1 之间的置信度。然后根据置信度应用业务规则:高置信度的抽取结果直接写入数据库,中等置信度的抽取结果标记出来进行抽查,低置信度的抽取结果则进入人工审核队列。
这种概率式方法比要求 LLM 达到 100% 的准确率实用得多——您应设计流水线,以从容应对不确定性,而不是假装不确定性不存在。
from pydantic import BaseModel, Field
from typing import Optional
class ExtractedWithConfidence(BaseModel):
value: Optional[str]
confidence: float = Field(ge=0.0, le=1.0)
reason: Optional[str] = None # Why confidence is low, if below threshold
class DocumentExtraction(BaseModel):
vendor_name: ExtractedWithConfidence
invoice_amount: ExtractedWithConfidence
due_date: ExtractedWithConfidence
def route_by_confidence(extraction: DocumentExtraction, threshold=0.85):
low_confidence_fields = []
for field_name, field_val in extraction.model_dump().items():
if isinstance(field_val, dict) and field_val.get('confidence', 1.0) < threshold:
low_confidence_fields.append(field_name)
if not low_confidence_fields:
return 'auto_approve'
elif len(low_confidence_fields) > 2:
return 'human_review'
else:
return f'spot_check: {low_confidence_fields}'重试失败时的后备策略
当所有重试次数都已用尽而验证仍然失败时,您需要采用后备策略。按优先级排序的选项如下:
- 部分结果:返回已通过验证的字段,并将失败字段标记为 null
- 人工审核队列:将文档加入队列进行人工审核,尤其适用于高价值文档
- 低保真抽取:改用要求较少字段的简单架构,以较少的结构化程度换取更高的健壮性
- 原始文本存储:保存带有元数据的原始文本,以便改进抽取流水线后重新处理
绝不要悄悄丢弃文档。请始终记录失败信息,并确保之后能够重新处理该文档。
基于外部数据的语义验证
有些验证规则需要查询外部数据,而这些查询无法在 Pydantic 验证器内部完成。例如,检查提取出的公司名称是否出现在您的 CRM 中,或检查提取出的产品 SKU 是否存在于您的库存中。这些检查应放在提取后验证步骤中,并在 Pydantic 验证通过后执行。
from typing import Optional
# Simulated external data source
KNOWN_VENDORS = {'acme corp', 'techsupplies inc', 'globex corporation'}
def validate_against_crm(extraction: dict) -> dict:
vendor = extraction.get('vendor', '').lower()
warnings = []
if vendor and vendor not in KNOWN_VENDORS:
warnings.append({
'field': 'vendor',
'issue': f'Vendor "{vendor}" not found in CRM',
'severity': 'warning'
})
# Optionally suggest closest match
# from difflib import get_close_matches
# matches = get_close_matches(vendor, KNOWN_VENDORS, n=1, cutoff=0.8)
# if matches: warnings[-1]['suggestion'] = matches[0]
return {
'extraction': extraction,
'warnings': warnings,
'requires_review': len(warnings) > 0
}
print('Semantic validation pattern defined')测试您的验证流程
您的验证逻辑需要单独的测试套件,与提取测试分开。请编写单元测试,将已知有问题的输出传入验证器,并确认验证器抛出了正确的错误。请测试各种边界情况:处于边界值的金额、格式异常的日期、包含意外空白的字段,以及以数字字符串而非数字表示的字段。
这套验证测试无需调用任何 API,因此运行速度快、成本低,可以在每次代码变更时执行。它也是验证规则的最佳文档——测试用例明确展示了您的流程所强制执行的每一种格式和业务约束。
from pydantic import ValidationError
def test_extraction_validation():
# Test cases: (input_data, should_pass)
test_cases = [
({'vendor': 'ACME', 'amount': 150.00, 'currency': 'USD', 'invoice_date': '2025-01-15'}, True),
({'vendor': 'ACME', 'amount': -50.00, 'currency': 'USD', 'invoice_date': '2025-01-15'}, False), # negative
({'vendor': 'ACME', 'amount': 150.00, 'currency': 'EURO', 'invoice_date': '2025-01-15'}, False), # 4-char
({'vendor': 'ACME', 'amount': 150.00, 'currency': 'USD', 'invoice_date': 'yesterday'}, False), # bad date
]
passed = failed = 0
for data, should_pass in test_cases:
try:
# ExtractedInvoice(**data) # Your Pydantic model
if should_pass:
passed += 1
else:
print(f'MISSED: Should have failed for {data}')
failed += 1
except (ValidationError, ValueError):
if not should_pass:
passed += 1
else:
print(f'UNEXPECTED FAIL for {data}')
failed += 1
print(f'Tests: {passed} passed, {failed} failed')
test_extraction_validation()模式失败分析与提示调优
在样本文档上运行提取流程并检查验证失败情况后,您很可能会发现,80% 的失败都源于少数几个根本原因。常见原因包括:模型始终错误地格式化某个特定地区的日期,混淆税额和总额,或在验证器要求连字符时生成不带连字符的电话号码。
对于每种反复出现的失败模式,请在提取提示中加入具体示例和约束,以防止该错误发生。更新后,请重新运行完整的测试集,确认修复提高了准确率,同时没有导致其他情况退化。这种“提示—评估”的迭代循环,正是生产环境中的提取流程在真实数据上达到 95% 以上准确率的方式。
快速检查
测试您对本课 AI 工程概念的理解。
课程回顾
在本课中,您学到了:验证在多个层级上运行——架构、格式、业务逻辑和语义层,每个层级都需要不同的实现方式;带有纠正反馈的重试会向模型提供具体的错误上下文,从而高效修正其输出;以及置信度分数可以根据提取的确定性,将结果概率性地路由到自动批准、抽查或人工审核队列。您现在已经完成了结构化输出和 JSON 模式课程。接下来,我们将探索向量嵌入——所有 RAG 系统的基础。
常见问题解答
「验证并重试错误输出」课时是免费的吗?
是的 — 「验证并重试错误输出」的完整文本可在网页上免费阅读。要进行交互式练习(内置代码编辑器和全天候 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 反馈 — 无需本地设置。