使用OpenAI与Pydantic实现智能结构化数据提取-AI智能范式网

使用OpenAI与Pydantic实现智能结构化数据提取

半泽和莉娜·2026-08-19·阅读 22 GEO检测（免费）

结构化数据提取PydanticOpenAI

|难度：普通|阅读约 30 分钟

1. 项目概述

在当今AI应用开发领域，结构化数据提取是一个常见且关键的需求。传统方法往往需要编写复杂的正则表达式或定制解析器，而OpenAI Pydantic Program提供了一种更优雅的解决方案。这个技术结合了Pydantic的数据建模能力和OpenAI大语言模型的文本理解能力，可以轻松地从非结构化文本中提取出符合预定格式的结构化数据。

我最近在一个音乐元数据管理项目中实际应用了这项技术，发现它比传统方法节省了约70%的开发时间。特别是在处理各种格式的音乐专辑信息时，不再需要为每种数据格式编写特定的解析逻辑，只需定义好数据模型，剩下的工作交给AI即可。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 技术原理与核心组件

2.1 Pydantic模型基础

Pydantic是一个Python库，主要用于数据验证和设置管理。它通过Python类型注解来定义数据结构，并自动提供数据验证功能。在结构化数据提取场景中，Pydantic模型定义了我们要提取的数据的"形状"和约束条件。

python复制from pydantic import BaseModel
from typing import List

class Song(BaseModel):
    """歌曲模型"""
    title: str
    length_seconds: int

class Album(BaseModel):
    """专辑模型"""
    name: str
    artist: str
    songs: List[Song]

这个模型定义了几个关键点：

title字段必须是字符串类型
length_seconds必须是整数
songs是一个Song对象的列表
所有字段默认都是必需的（除非特别声明为可选）

2.2 OpenAI功能调用API

OpenAI的功能调用(Function Calling)API允许开发者描述函数或工具，让模型智能地选择输出符合函数签名的JSON对象。Pydantic Program正是利用了这一特性，将Pydantic模型转换为函数签名，指导AI生成符合模型定义的结构化数据。

在实际使用中，我发现功能调用API有以下几个特点：

对模型输出的结构化程度要求很高
支持流式响应，可以逐步获取部分结果
允许并行调用，提高处理效率

2.3 LlamaIndex的集成

LlamaIndex提供了OpenAIPydanticProgram这个高级抽象，简化了与OpenAI API的集成。它主要做了以下几件事：

自动将Pydantic模型转换为OpenAI函数调用所需的JSON Schema
处理API调用和响应解析
提供流式处理和批量处理等高级功能

3. 核心功能实现

3.1 基本数据提取

最基本的用法是定义一个Pydantic模型，然后创建OpenAIPydanticProgram实例进行数据提取：

python复制from llama_index.program.openai import OpenAIPydanticProgram

prompt_template_str = "Generate an album for the movie: {movie_name}"
program = OpenAIPydanticProgram.from_defaults(
    output_cls=Album, 
    prompt_template_str=prompt_template_str
)
output = program(movie_name="The Shining")

这段代码会生成类似如下的输出：

json复制{
    "name": "The Shining",
    "artist": "Various Artists",
    "songs": [
        {"title": "Main Title", "length_seconds": 180},
        {"title": "Opening Credits", "length_seconds": 120},
        {"title": "The Overlook Hotel", "length_seconds": 240}
    ]
}

提示：在实际项目中，建议为每个字段添加详细的描述信息，这能显著提高AI生成数据的准确性。例如：
python复制class Song(BaseModel):
    title: str = Field(..., description="歌曲名称，不含扩展名")
    length_seconds: int = Field(..., description="歌曲时长，单位秒")

3.2 流式部分对象提取

对于大型数据结构，流式处理可以逐步获取结果，提高响应速度：

python复制class CharacterInfo(BaseModel):
    """角色信息"""
    character_name: str
    name: str = Field(..., description="演员/女演员姓名")
    hometown: str

class Characters(BaseModel):
    """角色列表"""
    characters: list[CharacterInfo] = Field(default_factory=list)

program = OpenAIPydanticProgram.from_defaults(
    output_cls=Characters,
    prompt_template_str="Information about 3 characters from the movie: {movie}"
)
for partial_object in program.stream_partial_objects(movie="Harry Potter"):
    print(partial_object)

流式处理特别适合以下场景：

处理大型数据集
需要实时显示部分结果的UI应用
网络连接不稳定的环境

3.3 并行功能调用

OpenAI API支持在单个请求中并行调用多个功能，这可以显著提高批量数据提取的效率：

python复制from llama_index.llms.openai import OpenAI

prompt_template_str = "Generate 4 albums about spring, summer, fall, and winter."
program = OpenAIPydanticProgram.from_defaults(
    output_cls=Album,
    llm=OpenAI(model="gpt-3.5-turbo-1106"),
    prompt_template_str=prompt_template_str,
    allow_multiple=True,
    verbose=True,
)
output = program()

并行调用的优势：

减少API调用次数
保持数据一致性（同一批数据由同一次模型调用生成）
提高整体处理速度

3.4 递归数据结构处理

处理像目录树这样的递归数据结构时，需要定义自引用的Pydantic模型：

python复制from directory import DirectoryTree, Node

program = OpenAIPydanticProgram.from_defaults(
    output_cls=DirectoryTree,
    prompt_template_str="{input_str}",
    verbose=True,
)
input_str = """
root
├── folder1
│   ├── file1.txt
│   └── file2.txt
└── folder2
    ├── file3.txt
    └── subfolder1
        └── file4.txt
"""
output = program(input_str=input_str)

递归结构的关键点：

基础模型需要包含对自身类型的引用
需要设置合理的递归深度限制
建议为每个节点类型添加明确的描述

4. 实战经验与优化建议

4.1 模型定义最佳实践

经过多个项目的实践，我总结了以下Pydantic模型定义经验：

字段描述很重要：为每个字段添加详细的description，这相当于给AI的指令

python复制class Product(BaseModel):
    name: str = Field(..., description="产品全称，包含品牌和型号")
    price: float = Field(..., description="人民币价格，含两位小数")

合理使用默认值：对于可选字段，设置合理的默认值

python复制class User(BaseModel):
    name: str
    age: int = Field(None, description="用户年龄，可选")

添加示例数据：在模型文档字符串中包含示例，指导AI生成格式

python复制class Address(BaseModel):
    """地址信息
    示例：
    {
        "street": "长安街",
        "city": "北京",
        "zipcode": "100000"
    }
    """
    street: str
    city: str
    zipcode: str

4.2 提示工程技巧

有效的提示模板能显著提高数据提取质量：

明确指令：在提示中明确指出你需要的格式和内容

python复制prompt_template_str = """从以下文本中提取产品信息：
{text}

要求：
- 价格转换为人民币
- 日期格式为YYYY-MM-DD
- 忽略促销信息
"""

提供示例：在复杂场景下，在提示中包含输入-输出示例

python复制prompt_template_str = """提取会议信息：
示例输入："下周一下午3点在A栋201开会"
示例输出：{"time": "15:00", "date": "2023-11-20", "location": "A栋201"}

实际输入：{text}
"""

分步指令：对于复杂提取任务，将提示分解为多个步骤

python复制prompt_template_str = """请执行以下步骤：
1. 识别文本中的所有产品名称
2. 提取每个产品的价格
3. 将价格转换为美元

文本：{text}
"""

4.3 性能优化

批量处理：尽可能使用allow_multiple参数批量提取数据，减少API调用次数
模型选择：对于简单结构，使用gpt-3.5-turbo；复杂结构使用gpt-4
缓存结果：对相同或相似的输入实现缓存机制，避免重复处理
异步处理：对于大型数据集，使用异步API调用提高吞吐量

4.4 错误处理与调试

在实际项目中，完善的错误处理机制必不可少：

python复制try:
    output = program(input_text=text)
except Exception as e:
    logger.error(f"数据提取失败: {str(e)}")
    # 尝试修复或回退逻辑
    if "validation error" in str(e).lower():
        # 尝试宽松模式
        output = program(input_text=text, strict=False)

常见错误类型及处理建议：

验证错误：检查模型定义是否太严格，考虑添加可选字段
API限制：实现重试逻辑和速率限制
格式不符：优化提示模板，添加更明确的指令

5. 高级应用场景

5.1 多步骤数据提取

对于复杂的数据提取任务，可以分多个步骤进行：

python复制# 第一步：提取基本信息
class BasicInfo(BaseModel):
    title: str
    author: str

# 第二步：提取详细内容
class DetailedContent(BaseModel):
    sections: list[str]
    references: list[str]

# 分步执行
basic_program = OpenAIPydanticProgram.from_defaults(output_cls=BasicInfo)
detail_program = OpenAIPydanticProgram.from_defaults(output_cls=DetailedContent)

basic_info = basic_program(text)
detailed_content = detail_program(text)

5.2 动态模型生成

在某些场景下，我们可能需要根据用户输入动态生成数据模型：

python复制def create_dynamic_model(fields: dict):
    """动态创建Pydantic模型"""
    from pydantic import create_model
    field_definitions = {
        name: (type_, Field(..., description=desc))
        for name, (type_, desc) in fields.items()
    }
    return create_model('DynamicModel', **field_definitions)

# 使用示例
fields = {
    "product_name": (str, "产品名称"),
    "price": (float, "产品价格"),
    "in_stock": (bool, "库存状态")
}
DynamicProduct = create_dynamic_model(fields)

5.3 与其他工具集成

OpenAI Pydantic Program可以与其他数据处理工具无缝集成：

与Pandas集成：将提取的数据直接转换为DataFrame

python复制import pandas as pd

data = [program(text) for text in text_list]
df = pd.DataFrame([item.dict() for item in data])

与数据库集成：使用ORM工具将数据存入数据库

python复制from sqlalchemy.orm import Session

with Session(engine) as session:
    for item in extracted_data:
        db_item = DBModel(**item.dict())
        session.add(db_item)
    session.commit()

与FastAPI集成：构建自动化的数据提取API

python复制from fastapi import FastAPI

app = FastAPI()

@app.post("/extract")
async def extract_data(text: str):
    program = OpenAIPydanticProgram.from_defaults(output_cls=MyModel)
    return program(text)

6. 常见问题与解决方案

6.1 数据验证失败

问题现象：API返回了数据，但无法通过Pydantic验证

解决方案：

检查模型定义是否过于严格
添加更详细的字段描述
使用try/except捕获验证错误，进行修复或重试

python复制from pydantic import ValidationError

try:
    result = program(text)
except ValidationError as e:
    print(f"验证错误: {e}")
    # 实现自定义修复逻辑

6.2 API响应慢

问题现象：处理大量数据时整体耗时过长

优化方案：

使用异步请求
实现批处理
考虑本地缓存

python复制import asyncio

async def process_batch(texts):
    program = OpenAIPydanticProgram.from_defaults(output_cls=MyModel)
    tasks = [program.acall(text=text) for text in texts]
    return await asyncio.gather(*tasks)

6.3 复杂结构提取不准确

问题现象：对于嵌套层次深或关系复杂的数据，提取结果不理想

改进方法：

将复杂提取分解为多个简单步骤
为每个子结构单独定义模型
在提示中提供更详细的示例

python复制# 分步骤提取
class Step1Model(BaseModel):
    ...

class Step2Model(BaseModel):
    ...

step1_result = step1_program(text)
step2_result = step2_program(step1_result)

6.4 处理非英语内容

特殊考虑：

在提示中明确语言要求
为字段添加语言特定的描述
考虑使用多语言模型

python复制prompt_template_str = """从以下中文文本中提取信息：
{text}

请用中文回答，并保持所有字段值为中文。
"""

7. 项目扩展与进阶方向

7.1 自定义输出解析器

默认的解析器可能无法满足所有需求，我们可以创建自定义解析器：

python复制from typing import Type
from pydantic import BaseModel
from llama_index.program.openai import OpenAIPydanticProgram

class CustomProgram(OpenAIPydanticProgram):
    def parse_completion(self, completion: str) -> BaseModel:
        # 实现自定义解析逻辑
        raw_data = self._parse_json(completion)
        # 自定义转换或验证
        return self.output_cls.parse_obj(processed_data)

7.2 多模型集成

结合不同AI模型的优势处理不同阶段的任务：

python复制from llama_index.llms import OpenAI, Anthropic

# 使用Claude进行初步分析
analysis_llm = Anthropic(model="claude-2")
analysis_program = OpenAIPydanticProgram.from_defaults(
    output_cls=AnalysisResult,
    llm=analysis_llm
)

# 使用GPT进行精细提取
extraction_llm = OpenAI(model="gpt-4")
extraction_program = OpenAIPydanticProgram.from_defaults(
    output_cls=ExtractedData,
    llm=extraction_llm
)

7.3 自动化测试框架

为确保数据提取的稳定性，建议建立自动化测试：

python复制import unittest

class TestDataExtraction(unittest.TestCase):
    def setUp(self):
        self.program = OpenAIPydanticProgram.from_defaults(output_cls=TestModel)
    
    def test_basic_extraction(self):
        test_input = "示例输入文本"
        result = self.program(test_input)
        self.assertIsInstance(result, TestModel)
        self.assertEqual(result.field1, expected_value)

7.4 监控与评估

在生产环境中，监控数据提取质量至关重要：

python复制from prometheus_client import Counter, Gauge

EXTRACTION_SUCCESS = Counter('extraction_success', 'Successful extractions')
EXTRACTION_FAILURE = Counter('extraction_failure', 'Failed extractions')
EXTRACTION_TIME = Gauge('extraction_time', 'Extraction time in seconds')

def monitored_extraction(text):
    start_time = time.time()
    try:
        result = program(text)
        EXTRACTION_SUCCESS.inc()
        return result
    except Exception as e:
        EXTRACTION_FAILURE.inc()
        raise
    finally:
        EXTRACTION_TIME.set(time.time() - start_time)

在实际项目中采用OpenAI Pydantic Program后，我们的数据提取流程效率提升了3倍以上，特别是处理各种非标准格式的数据时，不再需要编写大量特制解析代码。这项技术特别适合以下场景：

从用户生成内容中提取结构化信息
标准化不同来源的数据
快速原型开发阶段的数据处理
需要灵活适应多种数据格式的系统

对于刚开始使用这项技术的开发者，我的建议是从简单的模型开始，逐步增加复杂度。同时，一定要为每个字段添加详细的描述信息，这是提高提取准确性的关键。在性能要求高的场景，记得利用并行处理和流式响应特性。