instructor 结构化输出

import instructor

# 同步客户端（默认）
client = instructor.from_provider("openai/gpt-4o")
client = instructor.from_provider("anthropic/claude-3-5-sonnet-20240620")
client = instructor.from_provider("google/gemini-2.5-flash")
client = instructor.from_provider("ollama/llama3")       # 本地模型
client = instructor.from_provider("deepseek/deepseek-chat")

# 异步客户端
client = instructor.from_provider("openai/gpt-4o", async_client=True)

# 指定模式
client = instructor.from_provider(
    "openai/gpt-4o",
    mode=instructor.Mode.TOOLS,
    api_key="sk-...",  # 也可通过环境变量设置
)

import instructor
from openai import OpenAI, AsyncOpenAI

# 同步客户端
client = instructor.from_openai(OpenAI(api_key="sk-..."))

# 异步客户端
client = instructor.from_openai(AsyncOpenAI(api_key="sk-..."))

# 指定模式
client = instructor.from_openai(
    OpenAI(),
    mode=instructor.Mode.JSON,
)

import instructor
from anthropic import Anthropic, AsyncAnthropic

# 同步客户端
client = instructor.from_anthropic(Anthropic(api_key="sk-ant-..."))

# 异步客户端
client = instructor.from_anthropic(AsyncAnthropic(api_key="sk-ant-..."))

import instructor
import openai

# 补丁方式：修改 openai.Client 使其支持 response_model
client = instructor.patch(openai.OpenAI())

from instructor import Mode

# 常用模式
Mode.TOOLS              # 使用 Function Calling / Tool Calling API（推荐，最可靠）
Mode.JSON               # 使用 JSON Mode，强制返回 JSON
Mode.MD_JSON            # 普通 Chat Completion，从 Markdown 代码块中解析 JSON
Mode.TOOLS_STRICT       # 使用 OpenAI Structured Outputs（受限 JSON Schema 采样）
Mode.PARALLEL_TOOLS     # 并行工具调用，单次返回多个结构化对象
Mode.FUNCTIONS          # 旧版 Function Calling（已弃用）
Mode.RESPONSES_TOOLS    # 使用 OpenAI Responses API（新一代 API）
Mode.JSON_O1            # O1 模型专用（不支持 system message 和 tool calling）

# Anthropic 专用
Mode.ANTHROPIC_TOOLS    # Anthropic Tool Calling
Mode.ANTHROPIC_JSON     # Anthropic JSON Mode

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

client = instructor.from_provider("openai/gpt-4o")

# 基本使用
class User(BaseModel):
    name: str
    age: int

user = client.create(
    response_model=User,
    messages=[
        {"role": "user", "content": "提取信息：张三今年25岁"},
    ],
)
print(user)  # User(name='张三', age=25)

# 带字段描述和约束
class Movie(BaseModel):
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份", ge=1900, le=2030)
    genre: str = Field(description="电影类型")
    rating: Optional[float] = Field(None, description="评分（0-10）", ge=0, le=10)

movie = client.create(
    response_model=Movie,
    messages=[{"role": "user", "content": "电影《盗梦空间》是2010年上映的科幻片，评分8.8"}],
)
print(movie)  # Movie(title='盗梦空间', year=2010, genre='科幻', rating=8.8)

# 嵌套模型
class Address(BaseModel):
    street: str
    city: str
    country: str

class Person(BaseModel):
    name: str
    age: int
    addresses: List[Address]

person = client.create(
    response_model=Person,
    messages=[{"role": "user", "content": "李明30岁，住在北京朝阳区建国路88号，中国"}],
)

# 枚举类型
class Sentiment(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"

class SentimentResult(BaseModel):
    text: str
    sentiment: Sentiment
    confidence: float = Field(ge=0, le=1)

result = client.create(
    response_model=SentimentResult,
    messages=[{"role": "user", "content": "这个产品非常好用，我非常满意！"}],
)

class UserProfile(BaseModel):
    name: str
    age: int
    bio: str
    interests: List[str]

for partial_user in client.create_partial(
    response_model=UserProfile,
    messages=[{"role": "user", "content": "为一位28岁的软件工程师创建用户档案，名字叫小王"}],
):
    # 每次迭代返回一个更完整的 UserProfile 实例
    # 尚未填充的字段为 None
    print(partial_user)

# 输出示例：
# UserProfile(name='小', age=None, bio=None, interests=None)
# UserProfile(name='小王', age=28, bio='一位', interests=None)
# UserProfile(name='小王', age=28, bio='一位热爱编程的软件工程师', interests=['编程'])
# UserProfile(name='小王', age=28, bio='一位热爱编程的软件工程师', interests=['编程', '阅读', '游戏'])

class Tag(BaseModel):
    name: str
    confidence: float

for tag in client.create_iterable(
    response_model=Tag,
    messages=[{"role": "user", "content": "从以下文本中提取标签：Python是数据科学和机器学习领域最流行的编程语言，也被广泛用于Web开发"}],
):
    print(tag)

# 输出示例：
# Tag(name='Python', confidence=0.99)
# Tag(name='数据科学', confidence=0.95)
# Tag(name='机器学习', confidence=0.93)
# Tag(name='Web开发', confidence=0.85)

user, completion = client.create_with_completion(
    response_model=User,
    messages=[{"role": "user", "content": "提取：李四今年35岁"}],
)

print(user)         # User(name='李四', age=35)
print(completion)   # 原始 OpenAI ChatCompletion 对象
print(completion.usage)  # token 使用信息

from pydantic import BaseModel, Field, field_validator

class EmailContact(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    email: str

    @field_validator('email')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('邮箱格式不正确，必须包含@符号')
        return v

# 当LLM返回的email不合法时，Instructor会自动重试
# 重试时会把验证错误信息附加到消息中，帮助LLM修正
contact = client.create(
    response_model=EmailContact,
    messages=[{"role": "user", "content": "联系人是张三，邮箱是zhangsan-at-gmail.com"}],
    max_retries=3,  # 最多重试3次
)

from typing_extensions import Annotated
from pydantic import BeforeValidator
from instructor import llm_validator

class QuestionAnswer(BaseModel):
    question: str
    answer: Annotated[
        str,
        BeforeValidator(
            llm_validator("回答不能包含有害、违法或不道德的内容", client=client)
        ),
    ]

qa = client.create(
    response_model=QuestionAnswer,
    messages=[{"role": "user", "content": "问题：如何学习Python？"}],
)

from instructor import Maybe

# Maybe[User] 意味着：要么返回完整的User，要么返回部分数据+错误信息
result = client.create(
    response_model=Maybe[User],
    messages=[{"role": "user", "content": "提取用户信息：有一只猫叫小花"}],
)

if result.is_valid():
    print(result.value)  # 完整的 User 对象
else:
    print(result.error)  # 验证失败的原因
    print(result.value)  # 部分提取的数据

client = instructor.from_provider("openai/gpt-4o")

# 请求前日志
client.on("completion:kwargs", lambda **kw: print(f"请求参数: model={kw.get('model')}"))

# 错误处理
client.on("completion:error", lambda e: print(f"请求错误: {e}"))

# 成功回调
client.on("completion:success", lambda response: print(f"成功获取响应"))

class ExtractionResult(BaseModel):
    entities: List[str]
    relationships: List[str]

result = client.create(
    response_model=ExtractionResult,
    messages=[
        {
            "role": "user",
            "content": """从以下文本中提取实体和关系：
            
            
            请使用回答。""",
        },
    ],
    context={"text": "苹果公司由史蒂夫·乔布斯创立，总部位于加利福尼亚", "language": "中文"},
)

from instructor.processing.multimodal import Image, PDF, Audio

# 图片分析
class ImageDescription(BaseModel):
    objects: List[str]
    scene: str
    mood: str

response = client.create(
    response_model=ImageDescription,
    messages=[
        {
            "role": "user",
            "content": [
                "描述这张图片",
                Image.from_url("https://example.com/photo.jpg"),
            ],
        },
    ],
)

# 从 PDF 提取结构化数据
class InvoiceInfo(BaseModel):
    invoice_number: str
    total_amount: float
    date: str

response = client.create(
    response_model=InvoiceInfo,
    messages=[
        {
            "role": "user",
            "content": [
                "提取发票信息",
                PDF.from_path("./invoice.pdf"),
            ],
        },
    ],
)

# 音频转录
class TranscriptSummary(BaseModel):
    summary: str
    key_points: List[str]

response = client.create(
    response_model=TranscriptSummary,
    messages=[
        {
            "role": "user",
            "content": [
                "总结这段音频",
                Audio.from_path("./meeting.mp3"),
            ],
        },
    ],
)

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

client = instructor.from_provider("openai/gpt-4o")

class DocumentType(str, Enum):
    CONTRACT = "contract"
    INVOICE = "invoice"
    REPORT = "report"
    EMAIL = "email"
    OTHER = "other"

class Entity(BaseModel):
    name: str = Field(description="实体名称")
    entity_type: str = Field(description="实体类型：人名/公司/地点/日期/金额等")

class DocumentAnalysis(BaseModel):
    doc_type: DocumentType = Field(description="文档类型")
    title: str = Field(description="文档标题/主题")
    summary: str = Field(description="100字以内的摘要")
    entities: List[Entity] = Field(description="提取的命名实体")
    key_dates: List[str] = Field(description="关键日期列表")
    confidence: float = Field(description="分析置信度(0-1)", ge=0, le=1)

text = """
合同编号：HT-2024-0089
甲方：北京科技有限公司
乙方：上海创新软件有限公司
签署日期：2024年3月15日
合同金额：人民币580,000元
项目内容：AI智能客服系统开发
交付日期：2024年9月30日
"""

analysis = client.create(
    response_model=DocumentAnalysis,
    messages=[{"role": "user", "content": f"分析以下文档：\n{text}"}],
    max_retries=2,
)

print(f"类型: {analysis.doc_type}")
print(f"标题: {analysis.title}")
print(f"摘要: {analysis.summary}")
for entity in analysis.entities:
    print(f"  实体: {entity.name} ({entity.entity_type})")

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

client = instructor.from_provider("openai/gpt-4o")

# 第1步：提取关键词
class KeywordExtraction(BaseModel):
    keywords: List[str] = Field(description="关键词列表")
    main_topic: str = Field(description="主主题")

# 第2步：生成分类
class Classification(BaseModel):
    category: str = Field(description="分类类别")
    subcategory: str = Field(description="子类别")
    relevance_score: float = Field(description="相关度分数", ge=0, le=1)

# 第3步：生成摘要
class Summary(BaseModel):
    brief: str = Field(description="一句话摘要")
    detailed: str = Field(description="详细摘要（200字以内）")

# 链式调用
article = "量子计算是一种利用量子力学原理进行计算的技术..."

# Step 1
keywords = client.create(
    response_model=KeywordExtraction,
    messages=[{"role": "user", "content": f"提取关键词：{article}"}],
)
print(f"关键词: {keywords.keywords}")

# Step 2（使用 Step 1 的结果）
classification = client.create(
    response_model=Classification,
    messages=[{"role": "user", "content": f"对以下主题进行分类：{keywords.main_topic}"}],
)
print(f"分类: {classification.category} > {classification.subcategory}")

# Step 3
summary = client.create(
    response_model=Summary,
    messages=[{"role": "user", "content": f"总结：{article}"}],
)
print(f"摘要: {summary.brief}")

import asyncio
import instructor
from pydantic import BaseModel
from typing import List

async_client = instructor.from_provider("openai/gpt-4o-mini", async_client=True)

class SentimentResult(BaseModel):
    text: str
    sentiment: str
    score: float

async def analyze_sentiment(text: str) -> SentimentResult:
    return await async_client.create(
        response_model=SentimentResult,
        messages=[{"role": "user", "content": f"分析情感：{text}"}],
    )

async def batch_analyze(texts: List[str]) -> List[SentimentResult]:
    # 并发处理所有文本
    tasks = [analyze_sentiment(text) for text in texts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if not isinstance(r, Exception)]

# 运行
texts = [
    "这个产品太棒了！",
    "服务态度很差，再也不来了",
    "还行吧，中规中矩",
    "性价比超高，推荐购买",
]

results = asyncio.run(batch_analyze(texts))
for r in results:
    print(f"[{r.sentiment}] {r.text} → 得分: {r.score}")