GLM 5.2 结构化输出：JSON 模式、Schema 校验与可靠数据提取

Sure! Here is the product information you asked for:

**Name:** Widget Pro  
**Price:** $29.99  
**In Stock:** Yes

from openai import OpenAI

client = OpenAI(
    api_key="YOUR_ZHIPU_API_KEY",
    base_url="https://open.bigmodel.cn/api/paas/v4/"
)

response = client.chat.completions.create(
    model="glm-4-plus",
    response_format={"type": "json_object"},
    messages=[
        {
            "role": "system",
            "content": "Extract product information as JSON. Return only valid JSON."
        },
        {
            "role": "user",
            "content": "Widget Pro is priced at $29.99 and is currently in stock."
        }
    ]
)

import json
data = json.loads(response.choices[0].message.content)
print(data)
# {"name": "Widget Pro", "price": 29.99, "in_stock": true}

SYSTEM_PROMPT = """
You are a data extraction assistant. Extract the following fields from the user's input and return them as a JSON object with exactly these keys:

{
  "product_name": "string — the full product name",
  "price_usd": "number — price in USD, numeric only (no currency symbols)",
  "in_stock": "boolean — true if available, false if not",
  "category": "string — product category if mentioned, otherwise null",
  "description": "string — brief product description if present, otherwise null"
}

Return only the JSON object. Do not include explanation, markdown, or any text outside the JSON.
"""

response = client.chat.completions.create(
    model="glm-4-plus",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_input}
    ]
)

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

class ProductExtraction(BaseModel):
    product_name: str
    price_usd: float = Field(ge=0)
    in_stock: bool
    category: Optional[str] = None
    description: Optional[str] = None

def extract_product(text: str) -> ProductExtraction:
    response = client.chat.completions.create(
        model="glm-4-plus",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": text}
        ]
    )
    raw = response.choices[0].message.content
    data = json.loads(raw)
    return ProductExtraction(**data)  # raises ValidationError on schema mismatch

product = extract_product("Widget Pro — $29.99, electronics, in stock")
print(product.product_name)   # Widget Pro
print(product.price_usd)      # 29.99
print(product.in_stock)       # True

import json
from pydantic import ValidationError
import time

def extract_with_retry(text: str, max_attempts: int = 3) -> ProductExtraction:
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": text}
    ]

    for attempt in range(max_attempts):
        try:
            response = client.chat.completions.create(
                model="glm-4-plus",
                response_format={"type": "json_object"},
                messages=messages
            )
            raw = response.choices[0].message.content
            data = json.loads(raw)
            return ProductExtraction(**data)

        except json.JSONDecodeError as e:
            # JSON mode should prevent this, but handle defensively
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": f"Your response was not valid JSON: {e}. Please return only a JSON object."
            })

        except ValidationError as e:
            # Model returned JSON but with wrong structure
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": f"The JSON structure was incorrect: {e}. Please match the schema exactly."
            })

        if attempt < max_attempts - 1:
            time.sleep(1)

    raise RuntimeError(f"Failed to extract structured data after {max_attempts} attempts")

from concurrent.futures import ThreadPoolExecutor, as_completed

def batch_extract(documents: list[str], max_workers: int = 5) -> list[dict]:
    results = []
    failed = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_doc = {
            executor.submit(extract_with_retry, doc): i
            for i, doc in enumerate(documents)
        }

        for future in as_completed(future_to_doc):
            idx = future_to_doc[future]
            try:
                result = future.result()
                results.append({"index": idx, "data": result.model_dump()})
            except RuntimeError as e:
                failed.append({"index": idx, "error": str(e)})

    results.sort(key=lambda x: x["index"])
    return results, failed

documents = [
    "Widget Pro — $29.99, electronics, in stock",
    "Budget Desk — $149.00, furniture, limited availability",
    "Pro Camera Lens — $899, photography, backordered"
]

results, errors = batch_extract(documents)
for r in results:
    print(r["data"])