用 JSON Schema 让大模型稳定输出结构化数据

请严格输出 JSON，不要输出任何解释，不要输出 Markdown。

我已经等了三天了，订单状态一直没有更新。客服每次都说帮我催一下，但没有任何结果。
如果今天还不处理，我就要退款了。

{
  "sentiment": "negative",
  "reason": "物流太慢",
  "priority": "high"
}

{
  "type": "object",
  "properties": {
    "sentiment": {
      "type": "string",
      "enum": ["positive", "neutral", "negative"]
    },
    "summary": {
      "type": "string"
    },
    "priority": {
      "type": "string",
      "enum": ["low", "medium", "high"]
    },
    "action_items": {
      "type": "array",
      "items": {
        "type": "string"
      }
    }
  },
  "required": ["sentiment", "summary", "priority", "action_items"],
  "additionalProperties": false
}

pip install openai pydantic

from typing import Literal

from openai import OpenAI
from pydantic import BaseModel, Field

import json


client = OpenAI()


class FeedbackAnalysis(BaseModel):
    sentiment: Literal["positive", "neutral", "negative"] = Field(
        description="用户反馈的整体情绪"
    )
    summary: str = Field(
        description="用一句话概括用户反馈"
    )
    priority: Literal["low", "medium", "high"] = Field(
        description="处理优先级"
    )
    action_items: list[str] = Field(
        description="建议后续处理动作"
    )


raw_feedback = """
我已经等了三天了，订单状态一直没有更新。
客服每次都说帮我催一下，但没有任何结果。
如果今天还不处理，我就要退款了。
"""

response = client.responses.parse(
    model="gpt-4o-2024-08-06",
    input=[
        {
            "role": "system",
            "content": (
                "你是一个客服反馈分析助手。"
                "请把用户反馈转成结构化数据，不要扩展不存在的信息。"
            )
        },
        {
            "role": "user",
            "content": raw_feedback
        }
    ],
    text_format=FeedbackAnalysis,
)

result = response.output_parsed

print(json.dumps(result.model_dump(), ensure_ascii=False, indent=2))

{
  "sentiment": "negative",
  "summary": "用户因订单状态长期未更新且客服处理无结果而感到不满，并提出退款威胁。",
  "priority": "high",
  "action_items": [
    "查询订单当前状态",
    "升级给人工客服或物流专员处理",
    "主动向用户反馈明确处理时间",
    "必要时提供退款或补偿方案"
  ]
}

class FeedbackAnalysis(BaseModel):
    sentiment: Literal["positive", "neutral", "negative"]
    summary: str
    priority: Literal["low", "medium", "high"]
    action_items: list[str]

pip install jsonschema

import json

from jsonschema import ValidationError, validate


FEEDBACK_SCHEMA = {
    "type": "object",
    "properties": {
        "sentiment": {
            "type": "string",
            "enum": ["positive", "neutral", "negative"]
        },
        "summary": {
            "type": "string"
        },
        "priority": {
            "type": "string",
            "enum": ["low", "medium", "high"]
        },
        "action_items": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["sentiment", "summary", "priority", "action_items"],
    "additionalProperties": False
}


def parse_and_validate_json(raw_text: str) -> dict:
    """
    解析并校验模型返回的 JSON。
    不合格就抛错，避免脏数据进入后续系统。
    """
    try:
        data = json.loads(raw_text)
    except json.JSONDecodeError as e:
        raise ValueError(f"模型输出不是合法 JSON: {e}") from e

    try:
        validate(instance=data, schema=FEEDBACK_SCHEMA)
    except ValidationError as e:
        raise ValueError(f"模型输出不符合 Schema: {e.message}") from e

    return data


if __name__ == "__main__":
    model_output = """
    {
      "sentiment": "negative",
      "summary": "用户对订单处理延迟不满",
      "priority": "high",
      "action_items": ["查询订单", "联系用户", "升级处理"]
    }
    """

    result = parse_and_validate_json(model_output)
    print(result)

import json


def ask_model(prompt: str) -> str:
    """
    替换成实际的大模型调用。
    返回模型原始文本。
    """
    raise NotImplementedError


def generate_valid_json(base_prompt: str, max_retries: int = 3) -> dict:
    prompt = base_prompt

    for attempt in range(1, max_retries + 1):
        raw_output = ask_model(prompt)

        try:
            return parse_and_validate_json(raw_output)
        except ValueError as e:
            prompt = f"""
你上一次输出的 JSON 不合格。

错误信息：
{str(e)}

请只修复 JSON，不要解释，不要输出 Markdown。
必须符合这个 Schema：
{json.dumps(FEEDBACK_SCHEMA, ensure_ascii=False, indent=2)}

原始输出：
{raw_output}
"""

    raise RuntimeError("多次重试后，模型仍然没有输出合格 JSON")

你是一个结构化信息抽取助手。

任务：
从用户提供的文本中抽取信息，并填入指定结构。

规则：
1. 不要编造文本中没有的信息。
2. 如果信息缺失，请使用空字符串、空数组或 Schema 中允许的默认值。
3. 分类字段必须严格按照 Schema 中的枚举值选择。
4. summary 字段控制在 50 字以内。
5. action_items 只写可执行动作，不写泛泛建议。

{
  "data": "...",
  "result": "...",
  "info": "..."
}

{
  "user_intent": "...",
  "risk_level": "...",
  "next_actions": []
}

{
  "priority": "比较着急"
}

{
  "priority": "high"
}

{
  "additionalProperties": false
}

priority: Literal["low", "medium", "high"] = Field(
    description=(
        "处理优先级。high 表示用户明确投诉、退款、流失风险或订单阻塞；"
        "medium 表示需要人工跟进但没有紧急风险；"
        "low 表示普通咨询或低风险反馈。"
    )
)

## 分析结果

```json
{
  "summary": "用户对物流延迟不满"
}
```