Instructor：让LLM 100%输出有效JSON的Python库 —— 2026指南

# 安装Instructor
pip install instructor

# 安装你首选的LLM客户端（以OpenAI为例）
pip install openai

import instructor
from openai import OpenAI
from pydantic import BaseModel

# 用Instructor修补OpenAI客户端
client = instructor.from_openai(OpenAI())

# 将你的输出模式定义为Pydantic模型
class UserProfile(BaseModel):
    name: str
    age: int
    email: str
    interests: list[str]

# 从自然语言中提取结构化数据
def extract_profile(user_description: str) -> UserProfile:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=UserProfile,
        messages=[
            {
                "role": "user",
                "content": f"从这段描述中提取用户画像: {user_description}"
            }
        ]
    )

# 使用
profile = extract_profile(
    "Sarah是一名28岁的软件工程师，来自西雅图。"
    "她喜欢徒步旅行、摄影和阅读科幻小说。"
    "她的邮箱是sarah.chen@example.com"
)

print(profile)
# UserProfile(name='Sarah', age=28, email='sarah.chen@example.com', 
#             interests=['hiking', 'photography', 'reading sci-fi novels'])

# 直接访问类型化字段
print(f"姓名: {profile.name}, 年龄: {profile.age}")
print(f"邮箱有效: {'@' in profile.email}")

from pydantic import BaseModel, Field, field_validator

class ValidatedProduct(BaseModel):
    name: str = Field(description="产品名称，最多50个字符")
    price: float = Field(description="美元价格，必须为正数")
    category: str = Field(description="以下之一: electronics, clothing, food, books")
    
    @field_validator('category')
    @classmethod
    def validate_category(cls, v):
        allowed = {'electronics', 'clothing', 'food', 'books'}
        if v.lower() not in allowed:
            raise ValueError(f"类别必须是以下之一: {allowed}")
        return v.lower()
    
    @field_validator('price')
    @classmethod
    def validate_price(cls, v):
        if v <= 0:
            raise ValueError("价格必须为正数")
        return round(v, 2)

# 验证失败时Instructor自动重试
def parse_product(description: str) -> ValidatedProduct:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=ValidatedProduct,
        max_retries=3,  # 最多重试3次并附带反馈
        messages=[
            {"role": "user", "content": f"解析这个产品: {description}"}
        ]
    )

# 即使第一次尝试有问题，也能正常工作
product = parse_product(
    "无线蓝牙降噪耳机，售价$79.99。"
    "电子产品类别。"
)
print(product)
# ValidatedProduct(name='无线蓝牙降噪耳机', 
#                  price=79.99, category='electronics')

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

class Address(BaseModel):
    street: str
    city: str
    state: str = Field(description="2字母州代码")
    zip_code: str
    country: str = "US"

class OrderItem(BaseModel):
    product_name: str
    quantity: int = Field(ge=1, description="必须至少为1")
    unit_price: float = Field(gt=0)
    
    @property
    def total(self) -> float:
        return self.quantity * self.unit_price

class CustomerOrder(BaseModel):
    customer_name: str
    customer_email: str
    shipping_address: Address
    billing_address: Optional[Address] = None
    items: List[OrderItem]
    order_notes: Optional[str] = None
    
    @property
    def grand_total(self) -> float:
        return sum(item.total for item in self.items)

def extract_order(email_text: str) -> CustomerOrder:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=CustomerOrder,
        messages=[
            {"role": "user", "content": f"从邮件中提取订单:\n\n{email_text}"}
        ]
    )

order = extract_order("""
您好，我想下一个订单。

客户：John Smith (john.smith@email.com)
发货地址：123 Oak Street, San Francisco, CA 94102

商品：
- MacBook Pro M3，数量1，$1999
- USB-C集线器，数量2，每个$49

请为笔记本电脑提供礼品包装。
""")

print(f"客户: {order.customer_name}")
print(f"发货城市: {order.shipping_address.city}")
print(f"订单总额: ${order.grand_total:.2f}")

# --- Anthropic Claude ---
import anthropic
import instructor

anthropic_client = instructor.from_anthropic(anthropic.Anthropic())

result = anthropic_client.messages.create(
    model="claude-sonnet-4-20250514",
    response_model=UserProfile,
    max_tokens=1024,
    messages=[{"role": "user", "content": "提取：Mike 35岁，喜欢网球"}]
)
print(result)

# --- Google Gemini ---
import google.generativeai as genai
import instructor

gemini_client = instructor.from_gemini(
    genai.GenerativeModel("gemini-2.5-pro")
)

result = gemini_client.generate_content(
    response_model=UserProfile,
    contents=["提取：Lisa 29岁，喜欢烹饪和瑜伽"]
)
print(result)

# --- Cohere ---
import cohere
import instructor

cohere_client = instructor.from_cohere(cohere.Client())

result = cohere_client.chat(
    response_model=UserProfile,
    message="提取：David 42岁，喜欢高尔夫和钓鱼"
)
print(result)

import asyncio
import instructor
from openai import AsyncOpenAI
from pydantic import BaseModel

# 使用异步客户端进行批处理
async_client = instructor.from_openai(AsyncOpenAI())

class SentimentResult(BaseModel):
    text: str
    sentiment: str  # "positive", "negative", "neutral"
    confidence: float
    key_phrases: list[str]

async def analyze_single(text: str) -> SentimentResult:
    return await async_client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=SentimentResult,
        messages=[
            {"role": "user", "content": f"分析情感: {text}"}
        ]
    )

async def analyze_batch(texts: list[str]) -> list[SentimentResult]:
    """并发处理多个文本。"""
    tasks = [analyze_single(text) for text in texts]
    results = await asyncio.gather(*tasks)
    return results

# 并发处理100条评论
texts = [
    "这个产品超出了我的预期！",
    "质量太差，一天就坏了。",
    "还可以，没什么特别的但能用。",
    # ... 再添加97个
]

results = asyncio.run(analyze_batch(texts))
positive = sum(1 for r in results if r.sentiment == "positive")
print(f"正面: {positive}/{len(results)}")

from typing import Iterable
from pydantic import BaseModel

class PartialArticle(BaseModel):
    title: str
    sections: list[str]
    key_points: list[str]

# 在生成过程中流式传输结构化数据
def stream_article(topic: str) -> Iterable[PartialArticle]:
    return client.chat.completions.create_partial(
        model="gpt-4o",
        response_model=PartialArticle,
        stream=True,
        messages=[
            {"role": "user", "content": f"撰写关于以下主题的文章大纲: {topic}"}
        ]
    )

# 在部分结果到达时消费它们
for partial in stream_article("2026年可再生能源趋势"):
    print(f"标题: {partial.title}")
    print(f"已有章节数: {len(partial.sections)}")
    print("---")

from pydantic import BaseModel, field_validator

class StrictDateRange(BaseModel):
    start_date: str = Field(description="YYYY-MM-DD格式")
    end_date: str = Field(description="YYYY-MM-DD格式，必须在开始日期之后")
    
    @field_validator('start_date', 'end_date')
    @classmethod
    def validate_date_format(cls, v):
        from datetime import datetime
        datetime.strptime(v, "%Y-%m-%d")
        return v
    
    @field_validator('end_date')
    @classmethod
    def validate_order(cls, end, info):
        start = info.data.get('start_date')
        if start and end <= start:
            raise ValueError("end_date必须在start_date之后")
        return end

# Instructor将使用特定验证错误反馈进行重试
def extract_date_range(text: str) -> StrictDateRange:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=StrictDateRange,
        max_retries=3,
        messages=[
            {"role": "user", "content": f"提取日期范围: {text}"}
        ]
    )

# 即使模型最初交换日期或使用错误格式，
# Instructor也会使用具体的错误消息重新询问
try:
    result = extract_date_range(
        "项目从2026年3月15日持续到2026年1月10日"
    )
    print(result)
except Exception as e:
    print(f"达到最大重试次数后失败: {e}")

from typing import Literal

class SupportTicket(BaseModel):
    customer_query: str
    category: Literal[
        "billing", 
        "technical_support", 
        "account_access",
        "feature_request",
        "refund",
        "general_inquiry"
    ]
    priority: Literal["low", "medium", "high", "urgent"]
    suggested_response: str

def classify_ticket(ticket_text: str) -> SupportTicket:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=SupportTicket,
        messages=[
            {
                "role": "system",
                "content": "你是一名客户支持分类专家。"
            },
            {"role": "user", "content": f"分类这张工单:\n\n{ticket_text}"}
        ]
    )

# 分类结果保证是允许值之一
ticket = classify_ticket(
    "这个月我的订阅被重复扣费了。"
    "请立即退还重复扣款。"
)
print(f"类别: {ticket.category}")  # 永远是"billing"
print(f"优先级: {ticket.priority}")  # 永远是4个值之一

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import instructor
from openai import OpenAI

app = FastAPI(title="结构化LLM API")
client = instructor.from_openai(OpenAI())

# 请求模式
class ExtractionRequest(BaseModel):
    text: str
    extract_fields: list[str]

# 响应模式
class ExtractedData(BaseModel):
    entities: list[dict]
    relationships: list[dict]
    summary: str

@app.post("/extract", response_model=ExtractedData)
async def extract_entities(request: ExtractionRequest):
    """从非结构化文本中提取结构化实体。"""
    try:
        result = client.chat.completions.create(
            model="gpt-4o",
            response_model=ExtractedData,
            messages=[
                {
                    "role": "user",
                    "content": (
                        f"从这段文本中提取实体。"
                        f"关注: {', '.join(request.extract_fields)}\n\n"
                        f"文本: {request.text}"
                    )
                }
            ]
        )
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# 使用 uvicorn main:app --reload 运行

from typing import Type

class SearchQuery(BaseModel):
    """带有参数的生成搜索查询"""
    keywords: list[str]
    filters: dict[str, str]
    sort_by: Literal["relevance", "date", "price_asc", "price_desc"]
    
def generate_search(user_request: str) -> SearchQuery:
    return client.chat.completions.create(
        model="gpt-4o",
        response_model=SearchQuery,
        messages=[
            {
                "role": "user",
                "content": f"为以下请求生成优化搜索查询: {user_request}"
            }
        ]
    )

query = generate_search(
    "找100美元以下的无线耳机，电池续航好，最新的优先"
)
print(query.keywords)  # ['wireless earbuds', 'bluetooth']
print(query.filters)   # {'max_price': '100'}
print(query.sort_by)   # 'date'

import logging
import instructor

# 启用详细日志记录
instructor.enable_logging()
logging.basicConfig(level=logging.DEBUG)

# 或配置特定的记录器
logger = logging.getLogger("instructor")
logger.setLevel(logging.INFO)

# 为生产环境添加文件处理程序
handler = logging.FileHandler("instructor.log")
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)

# 现在所有重试尝试和验证错误都会被记录
result = client.chat.completions.create(
    model="gpt-4o",
    response_model=UserProfile,
    max_retries=3,
    messages=[{"role": "user", "content": "提取：Jane，25岁，喜欢艺术"}]
)