基于 Pydantic + LLM 的结构化数据提取

在非结构化数据处理（ETL）领域，正则表达式（Regex）长期占据统治地位，但其脆弱性在面对复杂的自然语言时暴露无遗。本文将深入探讨一种“Schema-First”的 AI 编程范式：利用结合的类型系统，构建具备思维链推理（CoT）自动纠错（Self-Healing）能力的结构化数据提取引擎。

且去填词

374人浏览 · 2026-01-15 08:58:12

且去填词 · 2026-01-15 08:58:12 发布

引言：在非结构化数据处理（ETL）领域，正则表达式（Regex）长期占据统治地位，但其脆弱性在面对复杂的自然语言时暴露无遗。本文将深入探讨一种“Schema-First”的 AI 编程范式：利用 LLM (DeepSeek/OpenAI) 结合 Pydantic 的类型系统，构建具备思维链推理（CoT）与自动纠错（Self-Healing）能力的结构化数据提取引擎。

1. 正则表达式的“鲁棒性陷阱”

在处理发票识别、简历解析、日志分析等任务时，传统工程往往陷入“正则地狱”：

维护成本高：r'^(\d{4})[-/](\d{2})[-/](\d{2})$' 这样的代码可读性极差。
语义理解缺失：正则无法理解“下周三”代表的具体日期，也无法识别“这点小事不急”隐含的低优先级。
容错率为零：OCR 识别产生的一个错别字，可能导致整个匹配链路失效。

大模型（LLM）虽然擅长语义理解，但其原生的非结构化输出（聊天模式）难以直接接入数据库。我们需要一个中间层，将模糊的语义坍缩为精确的类型。

2. 基于类型系统的“自愈循环”

我们采用的核心架构模式是 "Validation Loop"（校验闭环）。不同于传统的“提示词工程”，这种模式将 Python 的类型系统（Type System）直接映射为 LLM 的输出约束。

2.1 架构流程图

下图展示了 Instructor 库如何劫持 OpenAI SDK，在客户端与 LLM 之间构建了一个自动纠错的中间层：

sequenceDiagram
    participant User as 业务代码
    participant Lib as Instructor/Pydantic层
    participant LLM as DeepSeek/OpenAI API

    User->>Lib: 传入非结构化文本 + Pydantic Schema
    loop 自愈重试循环 (Max Retries)
        Lib->>LLM: 发送 Prompt + JSON Schema 定义
        LLM-->>Lib: 返回 JSON 字符串
        Lib->>Lib: Pydantic 校验 (Validate)
        alt 校验通过
            Lib-->>User: 返回 Python 对象
        else 校验失败 (ValueError)
            Lib->>LLM: 发送错误堆栈 + "请修正你的JSON"
            Note right of LLM: LLM 根据错误提示<br/>重新生成符合规范的数据
        end
    end

2.2 核心机制解析

Schema 映射：Pydantic 模型被自动转换为 OpenAI 的 tool_definitions 或 json_schema。
思维链注入：通过在 Schema 中定义 reasoning 字段，强制 LLM 在生成结果前输出思考过程，显著减少幻觉。
错误回传：当校验失败时，错误信息（如“日期格式错误”）被作为新的 Prompt 回传给 LLM，实现自我修复。

3. 构建智能会议纪要解析器

在生产环境中，我们不能只写脚本，需要将其封装为具备日志记录、重试机制和环境隔离的类。

3.1 环境准备

pip install openai pydantic instructor loguru

3.2 进阶 Schema 设计（含思维链与跨字段校验）

import os
import datetime
from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field, model_validator, ValidationError
from loguru import logger  # 使用 loguru 进行生产级日志记录

# 1. 定义枚举，限制 LLM 的输出范围
class Priority(str, Enum):
    HIGH = "高"
    MEDIUM = "中"
    LOW = "低"

# 2. 定义原子任务模型
class Task(BaseModel):
    # 【思维链核心】强制 LLM 先思考，提升后续字段准确率
    chain_of_thought: str = Field(
        ..., 
        description="分析任务的紧迫性、隐含时间以及上下文依赖关系。"
    )
    
    content: str = Field(..., description="任务核心内容，去除口语化词汇")
    assignee: str = Field(..., description="负责人姓名，若未提及填'待定'")
    due_date: Optional[str] = Field(
        None, 
        description="截止日期，格式 ISO 8601 (YYYY-MM-DD)"
    )
    priority: Priority = Field(..., description="任务优先级")

    # 【逻辑校验】确保高优先级任务必须有 Deadline
    @model_validator(mode='after')
    def validate_urgent_tasks(self):
        if self.priority == Priority.HIGH and not self.due_date:
            # 这个错误信息会回传给 LLM，指导它修正
            raise ValueError(
                "高优先级 (HIGH) 任务必须包含明确的截止日期 (due_date)。"
                "如果原文未提及，请根据语境推断一个合理日期（如次日），并在 chain_of_thought 中说明。"
            )
        return self

# 3. 定义整体输出结构
class MeetingAnalysis(BaseModel):
    summary: str = Field(..., description="会议内容的简短摘要")
    tasks: List[Task] = Field(..., description="提取的任务列表")

3.3 封装提取引擎

import instructor
from openai import OpenAI

class IntelligentExtractor:
    def __init__(self, api_key: str, base_url: str = "[https://api.deepseek.com](https://api.deepseek.com)"):
        self.client = instructor.from_openai(
            OpenAI(api_key=api_key, base_url=base_url),
            mode=instructor.Mode.JSON,
        )
    
    def extract(self, text: str, output_model: type[BaseModel], retries: int = 3):
        """
        通用提取方法，支持泛型返回
        """
        logger.info(f"开始提取任务，文本长度: {len(text)} 字符")
        
        try:
            return self.client.chat.completions.create(
                model="deepseek-chat",
                response_model=output_model,
                messages=[
                    {
                        "role": "system",
                        "content": (
                            f"你是一个严谨的数据分析师。当前日期：{datetime.date.today()}。"
                            "请严格遵循 JSON Schema 定义，任何逻辑错误都会导致任务失败。"
                        )
                    },
                    {"role": "user", "content": text},
                ],
                max_retries=retries, # 开启自愈循环
            )
        except Exception as e:
            logger.error(f"提取彻底失败，重试次数耗尽: {e}")
            raise

# 实例化引擎
extractor = IntelligentExtractor(api_key="your-api-key")

4. “隐含逻辑”与“自愈能力”测试

我们构造一个充满陷阱的测试用例：

模糊时间：“马上上线” —— 需要推断为高优且由于校验规则必须补全日期。
干扰信息：“点外卖” —— 需要被过滤。
多任务混合。

raw_text = """
会议记录 2026-01-14：
1. 老王，那个海报的事情，周五前必须搞定，客户催得很紧！
2. 小李你去问问运营流量券的事，这个不急，下个月再说。
3. 对了，中午大家一起点个麻辣香锅，别点微辣的。
4. 警告：支付接口有一个严重 Bug，张工你必须马上修复上线！
"""

try:
    result = extractor.extract(raw_text, MeetingAnalysis)
    
    print(f"\n=== 会议摘要: {result.summary} ===\n")
    for idx, task in enumerate(result.tasks, 1):
        print(f"任务 {idx}:")
        print(f"  [思考] {task.chain_of_thought}")
        print(f"  [内容] {task.content}")
        print(f"  [人员] {task.assignee}")
        print(f"  [状态] {task.priority.value} | 截止: {task.due_date}")
        print("-" * 40)
        
except Exception as e:
    print("处理失败，请检查日志")

4.1 预期输出与分析

=== 会议摘要: 讨论海报制作、运营沟通及支付接口紧急修复事项 ===

任务 1:
  [思考] 提到“周五前”，当前是14号（周三），故截止日为16号。语气“催得很紧”，定为高优先级。
  [内容] 制作宣传海报
  [人员] 老王
  [状态] 高 | 截止: 2026-01-16
----------------------------------------
任务 2:
  [思考] “下个月再说”表明时间充裕，优先级低。
  [内容] 咨询运营流量券事宜
  [人员] 小李
  [状态] 低 | 截止: 2026-02-01
----------------------------------------
任务 3:
  [思考] “马上修复”意味着最高紧急度。虽然未直接说日期，但根据高优先级校验规则，推断需在今日或明日完成，设定为明日。
  [内容] 修复支付接口严重Bug
  [人员] 张工
  [状态] 高 | 截止: 2026-01-15
----------------------------------------

技术亮点分析：

噪音过滤：“麻辣香锅”完全未出现在结果中，因为 Schema 约束了 AI 关注任务相关内容。
规则驱动的幻觉抑制：对于张工的任务，AI 被迫在 chain_of_thought 中解释为什么设置了 2026-01-15，这是代码中 validate_urgent_tasks 校验器强制要求的成果。

5. 工程化考量：技术选型对比

在实际落地时，我们不能盲目使用 LLM。下表对比了不同方案的 ROI（投资回报率）：

维度	正则表达式 (Regex)	传统 NLP (Spacy/Bert)	LLM + Pydantic (本文方案)
开发效率	低（需逐个编写规则）	中（需训练/微调模型）	极高（仅需定义 Schema）
运行成本	接近零	低（本地 CPU 推理）	中（Token 计费）
延迟	< 1ms	10-100ms	1s - 5s
鲁棒性	极差	中	极强
适用场景	固定格式 (身份证/手机号)	实体抽取 (NER)	复杂语义分析、异构数据清洗