大模型结构化输出实战:用 Instructor + Pydantic 让 LLM 返回 100% 合法的 JSON

阿里云上云采购季 2核2G云服务器99元/年 续费同价

还在用正则解析 LLM 返回的 JSON?这篇文章带你彻底告别 json.loads() 报错的日子。


一个让后端抓狂的经典场景

上周我在接微信消息通知系统时,遇到了一个经典问题:需要让大模型从用户输入中提取结构化信息(姓名、手机号、地址、意向产品),然后存入 MySQL。

我写了这样一段 Prompt:

prompt = """
从用户输入中提取以下字段,以 JSON 格式返回:
{
  "name": "姓名",
  "phone": "手机号", 
  "address": "地址",
  "product": "意向产品"
}
请务必返回合法的 JSON。
"""

然后发生了这些:

# 第一次返回
当然,这是提取的 JSON:\n```json\n{...}\n```

# 第二次返回  
{"name": "张三", "phone": "138xxx", "address": "河南省郑州市", "product": "轻量应用服务器",}

# 第三次返回
{name: "张三", phone: "138xxx"}

3 次调用,3 种格式。正则写了 4 行,try-except 套了 3 层。我一度怀疑自己是不是在写爬虫。

直到我发现了 Instructor 这个库 —— 它用 Pydantic 的类型系统强制 LLM 返回合法 JSON,从此 json.loads() 再也没报过错。


Instructor 是什么?

Instructor 是一个轻量级的 LLM 结构化输出库。核心思想非常简单:

“你定义 Pydantic Model,我保证 LLM 返回的数据 100% 匹配这个 Model。做不到我就重试。”

3 个核心能力

能力 说明
类型强制 用 Pydantic Field 定义字段类型,LLM 输出自动校验
自动重试 校验失败时自动重新请求 LLM,直到通过
多 Provider 支持 OpenAI / Claude / DeepSeek / 通义千问 等 10+ 平台

对比传统方式:

环节 传统方式 Instructor
定义输出格式 手写 JSON Schema / Prompt 描述 Pydantic Model(带类型提示)
解析返回 json.loads() + try-except 自动解析,失败自动重试
字段校验 手写 if-else Pydantic validators
类型安全 无(全是 dict) 完全类型安全(IDE 补全)
错误处理 自己写兜底逻辑 内置重试 + fallback

5 分钟快速上手

1. 安装

pip install instructor openai

2. 定义数据结构

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class ProductType(str, Enum):
    """产品类型"""
    CV = "云服务器"
    LIGHTHOUSE = "轻量应用服务器"
    CDN = "内容分发网络"
    COS = "对象存储"
    UNKNOWN = "未知"

class CustomerInfo(BaseModel):
    """客户意向信息"""
    name: str = Field(description="客户姓名")
    phone: str = Field(description="手机号,11位数字")
    company: Optional[str] = Field(default=None, description="公司名称,如有")
    product: ProductType = Field(description="意向产品类型")
    budget: Optional[str] = Field(default=None, description="预算范围,如 1000-5000元/月")
    urgency: str = Field(
        default="一般",
        description="紧急程度:紧急/一般/不急"
    )

class SalesLead(BaseModel):
    """销售线索"""
    customer: CustomerInfo = Field(description="客户信息")
    summary: str = Field(description="客户需求一句话总结(50字内)")
    recommended_package: str = Field(description="推荐的产品套餐")
    follow_up_priority: int = Field(
        ge=1, le=10,
        description="跟进优先级,1最低,10最高"
    )

3. 调用 LLM 获取结构化数据

import instructor
from openai import OpenAI

# 注意:用 instructor.patch 包装 OpenAI 客户端
client = instructor.from_openai(
    OpenAI(
        api_key="your-api-key",
        base_url="https://api.deepseek.com/v1"  # 也可以用 DeepSeek
    )
)

# 模拟用户输入
user_input = """
我叫王伟,电话是13812345678,在郑州一家电商公司做技术总监。
最近公司在搞大促,想在腾讯云部署一套高可用的电商系统,
需要负载均衡 + 云服务器 + MySQL,预算大概 5000-8000 一个月。
比较急,最好这周能搞定。
"""

# 提取结构化信息
lead: SalesLead = client.chat.completions.create(
    model="gpt-4o",  # 或 deepseek-chat
    response_model=SalesLead,
    messages=[
        {"role": "system", "content": "你是腾讯云销售助手,从客户对话中提取结构化销售线索。"},
        {"role": "user", "content": f"请从以下客户输入中提取销售线索:\n{user_input}"}
    ],
    max_retries=3,  # 最多重试 3 次
)

# 直接访问类型化字段
print(f"客户:{lead.customer.name}")
print(f"电话:{lead.customer.phone}")
print(f"产品:{lead.customer.product.value}")
print(f"优先级:{lead.follow_up_priority}/10")
print(f"推荐:{lead.recommended_package}")

输出结果:

客户:王伟
电话:13812345678
产品:云服务器
优先级:9/10
推荐:腾讯云高可用电商方案(CLB + 2x CVM + TencentDB MySQL)

注意代码里没有任何 json.loads() 或 try-except。lead 是类型安全的 SalesLead 对象,IDE 自动补全所有字段。


进阶技巧:让结构化输出更可靠

技巧1:添加字段级校验器

from pydantic import field_validator
import re

class CustomerInfo(BaseModel):
    name: str = Field(description="客户姓名")
    phone: str = Field(description="手机号,11位数字")

    @field_validator("phone")
    @classmethod
    def validate_phone(cls, v: str) -> str:
        # 去掉非数字字符
        cleaned = re.sub(r'\D', '', v)
        if len(cleaned) != 11:
            raise ValueError(f"手机号必须为11位数字,当前为 {len(cleaned)} 位")
        return cleaned

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str) -> str:
        if len(v) < 2:
            raise ValueError(f"姓名至少2个字符,当前为 '{v}'")
        if len(v) > 10:
            raise ValueError(f"姓名不应超过10个字符,当前为 {len(v)} 个字符")
        return v.strip()

当 LLM 返回的字段不符合校验规则时,Instructor 会自动重新请求,并在 Prompt 中附带前一次失败的原因。这比手动写校验循环可靠得多。

技巧2:从非结构化文本批量提取

from typing import List

class MeetingNote(BaseModel):
    company: str = Field(description="公司名称")
    contact: str = Field(description="联系人")
    demand: str = Field(description="需求描述")
    estimated_amount: Optional[int] = Field(default=None, description="预估金额(元/月)")

class MeetingNotes(BaseModel):
    """会议记录批量提取"""
    notes: List[MeetingNote] = Field(description="各公司需求列表")

meeting_text = """
今天拜访了3家公司:
1. 郑州中科纺织,张总,想上私有云做面料AI质检,预算2万/月
2. 盈云信息技术,李经理,想迁移现有的阿里云服务器到腾讯云,大概15台CVM
3. 广西银壹数字,王总监,做直播电商需要CDN和对象存储,预算没明确说
"""

result: MeetingNotes = client.chat.completions.create(
    model="gpt-4o",
    response_model=MeetingNotes,
    messages=[{"role": "user", "content": f"从以下会议记录中提取各公司的需求:\n{meeting_text}"}],
)

技巧3:使用 Maybe 处理 “可能为空” 的提取

from instructor import Maybe

# 当输入可能不包含目标信息时
extraction = client.chat.completions.create(
    model="gpt-4o",
    response_model=Maybe[CustomerInfo],
    messages=[
        {"role": "system", "content": "谨慎判断:只提取明确存在的客户信息。"},
        {"role": "user", "content": "今天天气不错,适合出去跑客户。"}
    ],
)

if extraction.result is None:
    print("未检测到有效客户信息,跳过")  # ← 这种情况会走这里
else:
    print(f"检测到客户:{extraction.result.name}")

技巧4:带思维链的结构化输出

class AnalysisResult(BaseModel):
    chain_of_thought: str = Field(description="分析推理过程")
    sentiment: str = Field(description="客户情绪:积极/中性/消极")
    key_concerns: List[str] = Field(description="客户关心的要点列表")
    next_action: str = Field(description="下一步行动建议")

让 LLM 在输出结构化数据前先输出推理链,能显著提高结果的准确性 —— 这是从 CoT(Chain-of-Thought)Prompting 中借鉴的思路。


Instructor vs LangChain:结构化输出怎么选?

很多朋友会问:LangChain 也有 with_structured_output(),为什么还要用 Instructor?

维度 Instructor LangChain with_structured_output
安装大小 轻量(~2MB) 重型(~50MB+)
学习曲线 5分钟上手 需要理解 LangChain 概念体系
Pydantic 集成 原生深度集成 基本支持
重试机制 内置智能重试 需自己实现
Provider 支持 10+ 平台 依赖 LangChain 生态
与现有代码集成 非侵入式 需要围绕 LangChain 构建
适用场景 快速项目、已有代码库 全新项目、复杂 Agent

我的选择标准:

  • 已有 FastAPI/Django 项目,只想加个 LLM 结构化输出 → Instructor
  • 从零开始构建复杂 AI Agent 系统 → LangChain(可以叠加 Instructor)
  • 极简需求:只做文本分类/实体提取 → Instructor(3 行代码搞定)

生产环境踩坑记录

坑1:Token 超限

Instructor 重试时会携带历史消息,导致 Token 用量累积。解决:

client.chat.completions.create(
    model="gpt-4o",
    response_model=SalesLead,
    messages=[...],
    max_retries=2,            # 限制重试次数
    max_tokens=2000,          # 限制每次响应 Token
)

坑2:DeepSeek 兼容性

DeepSeek API 部分兼容 OpenAI SDK,但可能不支持某些高级参数:

client = instructor.from_openai(
    OpenAI(
        api_key="sk-xxx",
        base_url="https://api.deepseek.com/v1"
    ),
    mode=instructor.Mode.JSON,  # 强制使用 JSON Mode
)

坑3:数值精度丢失

float 类型字段可能因 LLM 四舍五入导致偏差:

# 不好
price: float = Field(description="产品价格")

# 更好:用 str 接收,再手动转换
price_raw: str = Field(description="产品价格(保留原始格式)")

@property
def price(self) -> float:
    return float(self.price_raw.replace("元", "").strip())

总结

Instructor 解决了一个看似简单但实际很烦人的问题:让 LLM 输出合法的结构化数据。它不是”解决所有问题的银弹”,但它精准地解决了 LLM 应用开发中 80% 的解析痛苦。

三个核心收益:
1. 告别 json.loads() try-except 地狱
2. Pydantic 类型安全 + IDE 自动补全
3. 自动重试机制,即使 LLM 偶尔抽风也不影响系统稳定性

如果你也在做 AI 应用开发,强烈建议花 10 分钟把 Instructor 集成到项目里。相信我,你会后悔没有早点用。

👤 作者简介

一枚在大中原腹地(河南)卖公有云的从业者,主营腾讯云/阿里云/火山云,曾踩坑无数,现专注AI大模型应用落地。关注公众号「公有云cloud」,围观AI前沿动态~

博客:y1cloud.cn

发表评论