Structured Output 与 JSON Mode：大模型结构化输出技术全景

from pydantic import BaseModel
from openai import OpenAI

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

client = OpenAI()
completion = client.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "Extract the event information."},
        {"role": "user", "content": "Alice and Bob are going to a science fair on Friday."},
    ],
    response_format=CalendarEvent,
)

event = completion.choices[0].message.parsed
print(event.name, event.date, event.participants)

import OpenAI from "openai";
import { zodResponseFormat } from "openai/helpers/zod";
import { z } from "zod";

const CalendarEvent = z.object({
  name: z.string(),
  date: z.string(),
  participants: z.array(z.string()),
});

const openai = new OpenAI();
const completion = await openai.chat.completions.parse({
  model: "gpt-4o-2024-08-06",
  messages: [
    { role: "system", content: "Extract the event information." },
    { role: "user", content: "Alice and Bob are going to a science fair on Friday." },
  ],
  response_format: zodResponseFormat(CalendarEvent, "event"),
});

const event = completion.choices[0].message.parsed;

from anthropic import Anthropic

client = Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Extract key info from: John Smith (john@example.com) wants Enterprise plan demo next Tuesday."}
    ],
    output_config={
        "format": {
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "email": {"type": "string"},
                    "plan_interest": {"type": "string"},
                    "demo_requested": {"type": "boolean"},
                },
                "required": ["name", "email", "plan_interest", "demo_requested"],
                "additionalProperties": False,
            },
        }
    },
)

print(response.content[0].text)

from pydantic import BaseModel
from anthropic import Anthropic

class ContactInfo(BaseModel):
    name: str
    email: str
    plan_interest: str
    demo_requested: bool

client = Anthropic()
response = client.messages.parse(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Extract contact info from the email above."}],
    output_format=ContactInfo,
)

print(response.parsed_output)

import google.generativeai as genai

def get_structured_data():
    model = genai.GenerativeModel("gemini-2.0-flash")

    response = model.generate_content(
        "Extract: Alice loves hiking and photography",
        generation_config=genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema={
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "hobbies": {"type": "array", "items": {"type": "string"}},
                },
                "required": ["name", "hobbies"],
            },
        ),
    )
    print(response.text)

get_structured_data()

┌──────────────────────────────────────────────────────┐
│              约束解码核心流程                           │
│                                                      │
│  1. Schema 编译                                       │
│     JSON Schema ──→ Context-Free Grammar (CFG)        │
│                                                      │
│  2. 逐 Token 约束                                     │
│  ┌────────────┐    ┌─────────────┐    ┌──────────┐  │
│  │ LLM 生成   │    │ 计算合法     │    │ 掩码过滤 │  │
│  │ 概率分布    │──→│ Token 集合   │──→│ 采样     │  │
│  │ (vocab)    │    │ (mask)      │    │ 输出     │  │
│  └────────────┘    └─────────────┘    └──────────┘  │
│                                                      │
│  3. 拼接已生成的 token，更新 CFG 状态                   │
│  4. 重复 2-3 直到生成结束标记                           │
└──────────────────────────────────────────────────────┘

from openai import OpenAI
from pydantic import BaseModel

class SentimentResult(BaseModel):
    sentiment: str
    confidence: float
    keywords: list[str]

client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

# 方式一：直接使用 JSON Schema
completion = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[{"role": "user", "content": "Analyze sentiment: vLLM is wonderful!"}],
    extra_body={"response_format": SentimentResult},
)

# 方式二：使用枚举约束
completion = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[{"role": "user", "content": "Classify sentiment: This movie is great!"}],
    extra_body={"structured_outputs": {"choice": ["positive", "negative", "neutral"]}},
)

# 方式三：使用正则表达式约束
completion = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[{"role": "user", "content": "Generate an email for Alan Turing"}],
    extra_body={"structured_outputs": {"regex": r"\w+@\w+\.com"}},
)

# 方式四：使用 EBNF 语法约束（如生成 SQL）
sql_grammar = """
root ::= select_statement
select_statement ::= "SELECT " column " FROM " table " WHERE " condition
column ::= "name" | "email" | "age"
table ::= "users" | "orders"
condition ::= column "= " number
number ::= "1" | "2" | "3"
"""
completion = client.chat.completions.create(
    model="Qwen/Qwen2.5-7B-Instruct",
    messages=[{"role": "user", "content": "Generate SQL query"}],
    extra_body={"structured_outputs": {"grammar": sql_grammar}},
)

from vllm import LLM, SamplingParams
from vllm.sampling_params import StructuredOutputsParams

llm = LLM(model="Qwen/Qwen2.5-7B-Instruct")

params = SamplingParams(
    structured_outputs=StructuredOutputsParams(
        json={
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer"},
            },
            "required": ["name", "age"],
        }
    )
)

outputs = llm.generate(
    prompts="Extract: My name is Alice and I am 30 years old.",
    sampling_params=params,
)
print(outputs[0].outputs[0].text)

{
  "type": "object",
  "properties": {
    "priority": {
      "type": "string",
      "enum": ["low", "medium", "high", "critical"],
      "description": "Issue priority level for routing"
    },
    "confidence": {
      "type": "number",
      "minimum": 0.0,
      "maximum": 1.0,
      "description": "Model confidence score"
    }
  },
  "required": ["priority", "confidence"],
  "additionalProperties": false
}

from pydantic import BaseModel, Field
from enum import Enum

class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class IssueAnalysis(BaseModel):
    title: str = Field(description="Brief issue title")
    priority: Priority
    confidence: float = Field(ge=0.0, le=1.0)
    summary: str = Field(max_length=500)
    requires_human: bool

# 直接传给 OpenAI
from openai import OpenAI
client = OpenAI()

completion = client.chat.completions.parse(
    model="gpt-4o",
    messages=[...],
    response_format=IssueAnalysis,
)

result = completion.choices[0].message.parsed
# result 是 IssueAnalysis 类型，字段保证存在且类型正确

try:
    completion = client.chat.completions.parse(
        model="gpt-4o",
        messages=[...],
        response_format=IssueAnalysis,
    )

    result = completion.choices[0].message.parsed
    if result is None:
        print("Model refused or output was truncated")
    else:
        # 业务逻辑校验
        assert result.confidence > 0.5, "Low confidence, needs review"

except Exception as e:
    print(f"Structured output failed: {e}")