让模型输出自由文本，下游程序很难直接用：字段名可能变、类型可能错、偶尔还会多出一段解释。把模型当成一个「需要校验的接口」来对待，才是接进生产系统的正确姿势。本教程用 OpenAI 兼容接口的 JSON 结构化输出，再用 Pydantic 做本地校验，校验不过就带着错误把原话回灌给模型重试，直到拿到能直接入库的干净数据。
💡 演示仍然用 DeepSeek 的 OpenAI 兼容接口。结构化输出（response_format=json_schema）多数兼容接口都支持，但「strict 模式」各家实现有差异，请以你所用平台的文档为准。
Step 1：先定义「我要的数据长什么样」
1 用 Pydantic 把期望结构写成代码
与其事后用一堆 if 判断字段，不如先用 Pydantic 把结构钉死。模型返回后直接往里灌，类型不对立刻报错。
from pydantic import BaseModel, Field
from typing import Literal
class Feedback(BaseModel):
sentiment: Literal["positive", "neutral", "negative"] = Field(description="情绪倾向")
topics: list[str] = Field(description="提到的主题，最多 3 个")
urgency: Literal["low", "medium", "high"] = Field(description="紧急程度")
summary: str = Field(description="一句话摘要，不超过 30 字")Literal 是免费的「取值范围护栏」。把取值锁死成枚举，模型就不会乱编一个没定义的标签，下游也不用写一堆兼容分支。
Step 2：把 Pydantic 模型转成 JSON Schema 交给模型
2 模型只看 schema，不看你的 Python
Pydantic v2 自带导出 JSON Schema 的方法，直接喂给接口的 response_format。
import json
from openai import OpenAI
client = OpenAI(base_url=OPENAI_BASE_URL, api_key=OPENAI_API_KEY)
schema = Feedback.model_json_schema()
response_format = {
"type": "json_schema",
"json_schema": {
"name": "feedback",
"schema": schema,
"strict": True,
},
}Step 3：发请求，拿到结构化结果
3 把用户原话丢给模型抽取
user_text = "你们的服务太慢了，周三那单到现在还没发货，我很着急，希望尽快处理"
resp = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": user_text}],
response_format=response_format,
)
raw = resp.choices[0].message.content
print(raw) # 形如 {"sentiment": "negative", "topics": ["物流"], ...}💡 即便开了 json_schema，某些接口返回的 content 仍可能是带 markdown 代码围栏的字符串。解析前先 strip 掉 ```json 与 ``` 再 json.loads，更稳。
Step 4：本地校验，别信模型「一定对的承诺」
4 用 Pydantic 当最后一道闸
结构化输出会减少格式错误，但不保证业务正确：可能 topics 塞了 5 个、summary 写了一百字。这一步强制收口。
def validate(raw_text):
data = json.loads(raw_text)
obj = Feedback(**data) # 类型/枚举/必填全在这里卡
if len(obj.topics) > 3:
raise ValueError("topics 超过 3 个")
if len(obj.summary) > 30:
raise ValueError("summary 超过 30 字")
return obj不要因为「开了 strict」就跳过本地校验。strict 只保证 schema 形状，不保证你的业务规则（比如字数上限）。校验必须落在你自己的代码里。
Step 5：校验不过就重试，把错误回灌给模型
5 让模型自己修正，而不是你手写兜底
校验失败时，把报错信息作为一条新消息追加回去，让模型基于错误再生成一次。最多重试 3 次，避免无限循环。
def structured_extract(text, max_retry=3):
messages = [{"role": "user", "content": text}]
for attempt in range(max_retry):
resp = client.chat.completions.create(
model="deepseek-chat", messages=messages, response_format=response_format)
raw = resp.choices[0].message.content
try:
return validate(raw) # 成功就返回
except Exception as e:
messages.append({"role": "assistant", "content": raw})
messages.append({"role": "user", "content": f"校验失败：{e}，请严格按 schema 重新输出。"})
raise RuntimeError("重试次数用尽仍不合规")💡 把上一次的错误 raw 也回传，模型能「看到」自己哪里错了，第二次往往就对了。这正是 Agent 里「自我修正」的最小形态。
Step 6：封装成一个可复用函数
6 以后换模型只改一处
把「schema 定义 + 请求 + 校验 + 重试」收成一个通用函数，任何 Pydantic 模型都能用。
def complete_structured(model_cls, text, model="deepseek-chat", max_retry=3):
rf = {"type": "json_schema", "json_schema": {"name": model_cls.__name__,
"schema": model_cls.model_json_schema(), "strict": True}}
messages = [{"role": "user", "content": text}]
for _ in range(max_retry):
resp = client.chat.completions.create(model=model, messages=messages, response_format=rf)
try:
return model_cls(**json.loads(resp.choices[0].message.content))
except Exception as e:
messages.append({"role": "user", "content": f"校验失败：{e}，请重输。"})
raise RuntimeError("结构化输出重试失败")Step 7：端到端跑一遍，并记住边界
7 验证闭环真的能自修
故意给一段容易让 summary 超长的文本，观察它初次失败、随后被校验拉回正轨。
result = complete_structured(Feedback, "你们的服务太慢了，周三那单到现在还没发货，我很着急")
print(result.sentiment, result.urgency, result.topics, result.summary)三道边界要记牢：① strict 模式不是所有接口都支持，不支持时退而用「系统提示 + 普通 JSON 解析 + 校验」一样可行；② 校验失败要保留原始文本日志，别静默吞掉，否则出问题无法排查；③ 涉及金额、身份等高风险字段，结构化只是起点，关键结论仍建议人工复核。