DeepSeek结构化输出：JSON Schema+Pydantic校验实战

在 AI 应用开发中，让模型生成一段 JSON 并不困难。真正影响系统稳定性的，是模型能否持续按照业务要求返回字段完整、类型正确、结构一致的数据。

例如，企业开发一个智能客服系统，需要从用户消息中提取订单编号、问题类型、商品名称和处理建议。如果模型有时返回字符串，有时返回数组，甚至遗漏关键字段，后续数据库写入和业务流程就可能出现异常。

这也是结构化输出（Structured Output）需要解决的问题。

本文以 DeepSeek API 为例，结合 Python、Pydantic 和 JSON Schema，搭建一个可运行的数据提取流程，涵盖接口调用、数据定义、格式校验、异常处理和批量处理。重点不是让模型生成看起来像 JSON 的文本，而是让模型输出能够被程序可靠消费的数据。

一、为什么普通 JSON 输出仍然可能出现数据错误？

普通文本生成的结果具有较强的灵活性。模型可能使用不同的字段名称、调整字段顺序，或者在 JSON 前后添加解释文字。

例如，业务系统希望获得以下数据：

{
  "order_id": "A20261102001",
  "issue_type": "物流问题",
  "product_name": "无线耳机",
  "priority": "high"
}

但模型可能返回：

{
  "order": "A20261102001",
  "type": "物流",
  "product": "无线耳机",
  "priority": "较高"
}

从自然语言理解角度看，两份内容表达的意思接近。但对于程序而言，字段名称和字段值都发生了变化。

假设后端代码直接读取 order_id，第二份结果就可能产生 KeyError。

结构化输出的目标，是尽量减少这种不确定性，让模型输出符合预先定义的数据格式。

需要区分三个概念：

• JSON 格式合法性： 数据是否符合 JSON 语法，能否被解析。
• Schema 结构约束： 字段名称、字段类型和必填属性是否符合预先定义的规则。
• 业务数据校验： 字段内容是否符合实际业务要求，例如订单编号是否为空、优先级是否属于允许范围。

这三个层次不能互相替代。即使 JSON 能够正常解析，也不意味着其中的数据可以直接写入业务数据库。

二、DeepSeek JSON Output 与 JSON Schema 有什么区别？

DeepSeek 官方提供 JSON Output 功能。在 Chat Completions API 中，可以通过 response_format 设置 JSON 输出模式。

典型配置如下：

response_format={
    "type": "json_object"
}

该模式用于让模型输出合法 JSON 字符串，但它本身并不意味着模型一定按照开发者定义的每一个字段和类型返回数据。

例如，开发者希望模型返回：

{
  "name": "张三",
  "age": 25,
  "email": "test@example.com"
}

JSON Output 可以帮助模型生成合法 JSON，但仍需要程序检查 name、age 和 email 是否存在，以及对应类型是否正确。

JSON Schema 则用于描述数据结构，包括：

• type：数据类型；
• properties：对象属性；
• required：必填字段；
• enum：允许出现的枚举值；
• items：数组元素结构。

DeepSeek 的 Responses API 文档另外提供了 json_schema 输出格式，可以在请求中提交 Schema。

因此，开发者需要根据实际接口选择实现方式：

表格

本文后续采用 JSON Output + Pydantic 的方式实现，便于展示模型调用与本地数据校验之间的关系。

需要注意，Pydantic 是本地验证层，不会自动改变模型生成过程，也不能替代服务端的 Schema 约束。

三、使用 Pydantic 定义业务数据结构

在实际项目中，不建议直接在业务代码里反复判断 JSON 字段。

更好的方法是提前定义数据模型，将字段名称、类型、必填条件和业务约束集中管理。

首先安装依赖：

pip install -U openai pydantic

然后创建一个数据结构：

from pydantic import BaseModel, Field
from typing import Literal

class CustomerIssue(BaseModel):
    order_id: str = Field(
        min_length=1,
        description="订单编号"
    )

    issue_type: Literal[
        "物流问题",
        "商品问题",
        "退款问题",
        "其他问题"
    ]

    product_name: str = Field(
        min_length=1,
        description="商品名称"
    )

    priority: Literal[
        "low",
        "medium",
        "high"
    ]

    summary: str = Field(
        min_length=1,
        description="问题摘要"
    )

这里使用了 Literal 限制字段取值。

例如，priority 只能是 low、medium 或 high。即使模型返回了一个语法正确的 JSON，只要优先级是 urgent，本地校验仍会失败。

接下来可以生成 JSON Schema：

schema = CustomerIssue.model_json_schema()

print(schema)

Pydantic 会根据模型定义生成相应的 JSON Schema 描述。

这一步的价值在于减少重复维护。业务字段发生变化时，开发者只需要修改数据模型，再重新生成 Schema，而不必手工维护多份字段说明。

四、调用 DeepSeek API 获取结构化数据

完成数据模型定义后，就可以接入 DeepSeek API。

先配置 API Key：
Linux 或 macOS：

export DEEPSEEK_API_KEY="your_api_key"

Windows PowerShell：

$env:DEEPSEEK_API_KEY="your_api_key"

初始化客户端：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com"
)

这里使用 DeepSeek 官方 API 地址。如果使用其他 API 网关，应根据相应平台的接口文档配置地址和鉴权信息。

接下来定义提示词：

system_prompt = """
你是一个电商客服数据提取助手。

请从用户提供的内容中提取订单信息，
并以 JSON 格式返回。

必须包含以下字段： order_id、issue_type、product_name、 priority、summary。
issue_type 只能是： 物流问题、商品问题、退款问题、其他问题。
priority 只能是： low、medium、high。
没有明确的信息时，不要自行编造。
请使用空字符串表示无法确认的文本字段。
"""

调用接口：

import json

user_message = """
我的订单编号是 A20261102001，
购买的商品是无线耳机。

物流显示已经三天没有更新，
我想咨询一下什么时候能够送到。
"""

response = client.chat.completions.create(
    model="deepseek-flash",
    messages=[
        {
            "role": "system",
            "content": system_prompt
        },
        {
            "role": "user",
            "content": user_message
        }
    ],
    response_format={
        "type": "json_object"
    },
    max_tokens=2048
)

content = response.choices[0].message.content

if not content or not content.strip():
    raise ValueError("模型返回内容为空")

data = json.loads(content)

print(json.dumps(
    data,
    ensure_ascii=False,
    indent=2
))

这里有几个容易忽略的细节。

第一，DeepSeek 官方 JSON Output 文档要求提示词中包含 JSON 指令，并提供目标格式示例。否则，模型可能无法按照预期组织内容。

第二，max_tokens 需要根据输出规模合理设置。如果输出被截断，可能出现 JSON 解析失败。

第三，官方文档特别提示，JSON Output 在部分情况下可能返回只有空格的 content，即使 finish_reason 为 stop。因此，程序需要主动检查空内容，不能直接调用 json.loads ()。

五、使用 Pydantic 验证模型返回结果

模型返回合法 JSON 后，还需要检查字段是否符合业务定义。

直接使用：

validated_data = CustomerIssue.model_validate(data)

print(validated_data)

如果返回的数据满足模型定义，Pydantic 会生成对应的数据模型对象。

如果字段缺失、类型错误或者枚举值不合法，就会抛出 ValidationError。

例如：

invalid_data = {
    "order_id": "A20261102001",
    "issue_type": "物流",
    "product_name": "无线耳机",
    "priority": "urgent",
    "summary": "物流长时间未更新"
}

validated_data = CustomerIssue.model_validate(
    invalid_data
)

这段代码会因为 issue_type 和 priority 不符合定义而校验失败。

开发者可以捕获异常：

from pydantic import ValidationError

try:
    validated_data = CustomerIssue.model_validate(data)

except ValidationError as e:
    print("数据校验失败：")
    print(e.errors())

e.errors () 可以返回具体的错误位置、错误类型和相关信息。

例如，开发者可以据此判断究竟是优先级不合法，还是订单编号缺失。

需要注意，Pydantic 的字段校验只能验证规则本身，不能证明模型提取的信息一定真实。订单编号是否存在，还需要查询业务数据库。

六、字段缺失和类型错误，如何实现自动修复？

在生产环境中，遇到校验失败就直接终止任务，通常不是理想的处理方式。

可以增加一个有限次数的修复流程：将原始数据、校验错误和字段要求重新交给模型，让它只修复不符合要求的部分。

下面给出一个简化实现：

from pydantic import ValidationError
import json

def validate_with_retry(data, max_retries=2):

    current_data = data
    for attempt in range(max_retries + 1):
        try:
            return CustomerIssue.model_validate(
                current_data
            )
        except ValidationError as e:
            if attempt >= max_retries:
                raise
            repair_prompt = f"""
请修复下面的 JSON 数据。

原始数据： {json.dumps(
    current_data,
    ensure_ascii=False
)}

校验错误： {json.dumps(
    e.errors(),
    ensure_ascii=False
)}

请严格按照原有字段定义修复，
不要编造无法确认的业务信息。

只返回 JSON，不要输出解释。
"""
            response = client.chat.completions.create(
                model="deepseek-flash",
                messages=[
                    {
                        "role": "system",
                        "content": "你是 JSON 数据修复助手。请输出合法 JSON。"
                    },
                    {
                        "role": "user",
                        "content": repair_prompt
                    }
                ],
                response_format={
                    "type": "json_object"
                },
                max_tokens=2048
            )

            content = response.choices[0].message.content

            if not content or not content.strip():
                raise ValueError("修复请求返回空内容")

            current_data = json.loads(content)

    raise ValueError("数据修复失败")

这个实现采用最多两次额外修复请求。

实际部署时，还需要增加请求超时、异常捕获和重试次数限制，避免修复流程无限循环。

更重要的是，不能把所有校验错误都交给模型修复。

例如：

• JSON 语法错误，可以尝试重新生成；
• 字段类型错误，可以要求模型重新组织；
• 数据库中不存在的订单编号，应该查询业务系统；
• 用户没有提供的关键信息，应该标记为待补充，而不是让模型猜测。

自动修复的目标是修正数据格式，而不是制造缺失的业务事实。

七、批量数据提取时，如何避免单条错误影响整个任务？

当结构化输出用于批量客服记录、商品信息提取或文档解析时，通常需要处理大量输入。

这时不应该因为某一条数据校验失败，就让整个批次停止。

可以为每条任务单独记录执行状态：

def process_one(text):

    try:
        response = client.chat.completions.create(
            model="deepseek-flash",
            messages=[
                {
                    "role": "system",
                    "content": system_prompt
                },
                {
                    "role": "user",
                    "content": text
                }
            ],
            response_format={
                "type": "json_object"
            },
            max_tokens=2048
        )

        content = response.choices[0].message.content

        if not content or not content.strip():
            return {
                "status": "failed",
                "reason": "empty_content"
            }

        data = json.loads(content)
        validated = CustomerIssue.model_validate(data)

        return {
            "status": "success",
            "data": validated.model_dump()
        }

    except json.JSONDecodeError:
        return {
            "status": "failed",
            "reason": "invalid_json"
        }
    except ValidationError as e:
        return {
            "status": "failed",
            "reason": "schema_validation_error",
            "details": e.errors()
        }
    except Exception as e:
        return {
            "status": "failed",
            "reason": "request_or_runtime_error",
            "details": str(e)
        }

批量处理时：

records = [
    "订单A001物流已经三天没有更新。",
    "订单A002商品存在破损，需要申请退款。",
    "订单A003客户咨询商品颜色。"
]

results = []

for record in records:
    results.append(process_one(record))

for index, result in enumerate(results):
    print(index, result)

这是一个串行处理示例，适合说明单条任务隔离的基本逻辑。

在高吞吐量场景下，可以进一步使用 asyncio 和信号量控制并发，但需要结合 API 限流规则、请求超时和上游服务能力确定并发规模。

建议为每条任务记录以下信息：

表格

这样可以区分模型输出问题、网络问题和业务校验问题，也便于后续分析成本与失败率。

八、JSON Schema 在生产环境中还需要注意什么？

结构化输出并不是只配置一个参数就能解决全部问题。

正式上线时，至少需要考虑以下几个问题。

1. 不要将字段缺失统一处理为空值。

不同字段有不同业务含义。订单编号为空，可能意味着无法定位订单；问题摘要为空，则可能只是需要补充用户信息。

建议根据字段的重要程度设置不同的处理策略。

1. 枚举字段需要与业务系统保持一致。

例如，业务系统使用 refund 表示退款，而模型输出中文 “退款问题”，即使语义一致，也可能无法直接写入数据库。

可以在数据模型中统一映射关系。

1. Schema 修改需要考虑版本兼容。

当业务新增字段时，旧版本消费者可能无法处理新的数据结构。

建议为数据结构增加版本标识，并在数据库和消息队列中保留必要的兼容逻辑。

1. 不要将模型输出直接作为数据库操作指令。

模型生成的字段即使通过 Pydantic 校验，也不代表它有权执行退款、删除记录或修改订单。

涉及真实业务操作时，还需要权限校验、业务规则判断和人工审核机制。

1. 监控结构化输出的实际质量。

建议持续统计 JSON 解析成功率、Schema 校验通过率、字段缺失率、修复成功率和平均 Token 消耗。

这些指标比单纯观察模型是否返回 JSON 更有实际意义。

九、多模型项目如何管理结构化输出接口？

在实际开发中，结构化输出只是 AI 应用中的一个环节。

一个项目可能使用 DeepSeek 处理文本抽取，使用 Kimi 处理长文档，再由其他国产模型承担不同的业务任务。

当模型来源增加时，开发团队还需要解决接口配置分散、数据结构不一致和模型切换成本增加等问题。

即使不同模型都能生成 JSON，也不代表它们使用相同的参数格式、响应字段和结构化输出能力。直接在业务代码中为每个模型编写一套解析逻辑，后续维护会越来越复杂。

更合理的做法，是在应用层增加统一的数据模型和模型适配层。

1. 使用统一的数据结构，隔离模型差异

例如，业务系统可以始终要求模型返回 CustomerIssue 对象，无论底层使用哪个模型，都先转换为统一的数据结构，再交给业务逻辑处理。

可以将调用过程拆分为三个部分：

• 模型适配层：负责处理不同模型的 API 请求和响应格式。
• 数据校验层：使用 Pydantic 检查字段、类型和业务规则。
• 业务处理层：负责数据库写入、工单创建和后续业务操作。

整体流程如下：

用户输入
    ↓
模型选择与 API 调用
    ↓
模型适配层
    ↓
JSON 数据解析
    ↓
Pydantic Schema 校验
    ├─ 校验失败 → 错误记录 / 有限修复
    ↓
统一业务数据对象
    ↓
数据库 / 业务系统

这种设计的关键是让业务层不依赖某个模型特有的响应格式。

例如，后续需要更换模型时，只需要调整对应的适配逻辑，而不必重新编写整个订单信息提取流程。

1. 统一模型调用入口，减少重复配置

当项目同时使用 DeepSeek、Kimi 等国产模型时，可以考虑通过统一 API 网关管理模型接入。

例如，星链 API定位于国产大模型统一接入与 API 中转网关，聚合了 DeepSeek、Kimi、Qwen、GLM 等模型，已上架 65 + 国产大模型。对于需要管理多个模型入口的开发团队，可以将其作为接入方案之一。

平台支持 20 + 主流开发工具，但具体模型的参数支持情况、JSON 输出格式和接口兼容程度，仍需要根据实时文档逐项验证。

需要特别说明：统一 API 入口并不意味着所有模型都支持相同的 JSON Schema 参数。开发者仍应在适配层中保留模型能力检查，不能因为统一了请求地址，就省略对底层模型能力的判断。

1. 生产环境如何选择结构化输出方案？

表格

对于需要处理订单、客户资料和企业内部数据的系统，建议将模型输出视为待验证的数据，而不是可信的业务指令。

即使结构化输出通过了 Schema 校验，也仍然需要业务规则、数据库状态和权限系统进行二次确认。

十、总结：结构化输出的关键是让模型结果能够被程序可靠使用

DeepSeek API 的 JSON Output 可以帮助开发者获得格式更规范的 JSON 数据，但在实际业务中，仅仅保证 JSON 语法正确还不够。

从模型输出到业务系统真正使用，中间至少需要经过三个环节：JSON 解析、Schema 校验和业务规则验证。

Python 开发者可以使用 Pydantic 定义统一的数据结构，将字段类型、必填条件和枚举规则集中管理。当模型返回的数据不符合要求时，再根据错误类型决定重新生成、有限修复或转交业务系统处理。

对于多模型应用，还应增加独立的模型适配层，避免将不同模型的参数和响应格式直接写进业务逻辑。

这套方案适合客服信息提取、文档解析、企业知识查询、订单分类和 Agent 数据处理等场景。真正投入生产前，建议先准备一组具有代表性的测试数据，持续观察 JSON 解析成功率、Schema 校验通过率、字段缺失率和异常修复成本。

结构化输出并不是让模型回答得更像程序，而是让模型生成的数据能够经过明确规则验证，再安全地进入后续业务流程。