openai 标准化协议 Structured Outputs 具体示例教程

Structured Outputs 具体示例教程

场景：个人财务管理助手

假设我们要构建一个 AI 助手，帮助用户记录和管理个人财务支出。用户可以输入自然语言描述（如“昨天我花了50元买了午餐”），助手将提取关键信息并以结构化 JSON 格式返回，包括日期、金额、类别和备注。

示例 1：使用 Structured Outputs 提取财务记录

步骤 1：定义 JSON Schema

我们需要一个清晰的 Schema 来描述财务记录：

{
    "type": "object",
    "properties": {
        "date": {
            "type": "string",
            "description": "支出日期，格式为 YYYY-MM-DD"
        },
        "amount": {
            "type": "number",
            "description": "支出金额，单位为人民币（元）"
        },
        "category": {
            "type": "string",
            "enum": ["餐饮", "交通", "娱乐", "购物", "其他"],
            "description": "支出类别"
        },
        "note": {
            "type": ["string", "null"],
            "description": "可选备注，若无则为 null"
        }
    },
    "required": ["date", "amount", "category", "note"],
    "additionalProperties": false
}

设计要点：

date 使用标准日期格式。
amount 为数字类型，确保精确。
category 使用枚举限制可选值。
note 可选，通过 "type": ["string", "null"] 实现。

步骤 2：实现 API 调用

使用 Python 实现，提取用户输入中的财务信息：

from openai import OpenAI
import json
from datetime import datetime, timedelta

client = OpenAI()

# 定义 Schema
schema = {
    "type": "object",
    "properties": {
        "date": {"type": "string", "description": "支出日期，格式为 YYYY-MM-DD"},
        "amount": {"type": "number", "description": "支出金额，单位为人民币（元）"},
        "category": {
            "type": "string",
            "enum": ["餐饮", "交通", "娱乐", "购物", "其他"],
            "description": "支出类别"
        },
        "note": {"type": ["string", "null"], "description": "可选备注，若无则为 null"}
    },
    "required": ["date", "amount", "category", "note"],
    "additionalProperties": False
}

# 计算昨天的日期（假设当前日期为 2025-03-16）
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")

response = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=[
        {"role": "system", "content": "你是一个财务管理助手，从用户输入中提取结构化支出信息。如果信息不完整，返回合理默认值。"},
        {"role": "user", "content": "昨天我花了50元买了午餐"}
    ],
    text={
        "format": {
            "type": "json_schema",
            "name": "expense_record",
            "schema": schema,
            "strict": True
        }
    }
)

# 解析结果
expense = json.loads(response.output_text)
print(json.dumps(expense, indent=2, ensure_ascii=False))

输出

{
  "date": "2025-03-15",
  "amount": 50,
  "category": "餐饮",
  "note": "买了午餐"
}

解析说明：

date：模型根据“昨天”推断为 2025-03-15（假设当前为 2025-03-16）。
amount：从“50元”提取为数字 50。
category：根据“午餐”推断为“餐饮”。
note：提取“买了午餐”作为备注。

步骤 3：处理边缘情况

添加错误处理，应对拒绝或不完整响应：

try:
    response = client.responses.create(
        model="gpt-4o-2024-08-06",
        input=[
            {"role": "system", "content": "你是一个财务管理助手，从用户输入中提取结构化支出信息。如果信息不完整，返回合理默认值。"},
            {"role": "user", "content": "昨天我花了50元买了午餐"}
        ],
        max_output_tokens=20,  # 模拟令牌限制
        text={
            "format": {"type": "json_schema", "name": "expense_record", "schema": schema, "strict": True}
        }
    )

    if response.status == "incomplete" and response.incomplete_details.reason == "max_output_tokens":
        print("错误：输出令牌数不足，无法生成完整响应")
    elif response.output[0].content[0].type == "refusal":
        print(f"模型拒绝：{response.output[0].content[0].refusal}")
    else:
        expense = json.loads(response.output_text)
        print(json.dumps(expense, indent=2, ensure_ascii=False))

except Exception as e:
    print(f"API 调用失败：{e}")

可能输出（令牌限制情况）：

错误：输出令牌数不足，无法生成完整响应

示例 2：结合 Function Calling 和 Structured Outputs

场景：保存财务记录到数据库

现在我们扩展功能，让模型不仅提取支出信息，还调用函数将其保存到数据库。

步骤 1：定义 Function Calling 和 Structured Outputs

Function Schema

{
    "type": "function",
    "name": "save_expense",
    "description": "将支出记录保存到数据库",
    "parameters": {
        "type": "object",
        "properties": {
            "date": {"type": "string", "description": "支出日期，YYYY-MM-DD"},
            "amount": {"type": "number", "description": "支出金额（元）"},
            "category": {"type": "string", "enum": ["餐饮", "交通", "娱乐", "购物", "其他"]},
            "note": {"type": ["string", "null"]}
        },
        "required": ["date", "amount", "category", "note"],
        "additionalProperties": False
    }
}

Structured Output Schema（用于最终响应）

{
    "type": "object",
    "properties": {
        "status": {"type": "string", "enum": ["success", "error"]},
        "message": {"type": "string"}
    },
    "required": ["status", "message"],
    "additionalProperties": False
}

步骤 2：实现代码

from openai import OpenAI
import json
from datetime import datetime, timedelta

client = OpenAI()

# 函数定义
tools = [{
    "type": "function",
    "name": "save_expense",
    "description": "将支出记录保存到数据库",
    "parameters": {
        "type": "object",
        "properties": {
            "date": {"type": "string", "description": "支出日期，YYYY-MM-DD"},
            "amount": {"type": "number", "description": "支出金额（元）"},
            "category": {"type": "string", "enum": ["餐饮", "交通", "娱乐", "购物", "其他"]},
            "note": {"type": ["string", "null"]}
        },
        "required": ["date", "amount", "category", "note"],
        "additionalProperties": False
    }
}]

# Structured Output Schema
response_schema = {
    "type": "object",
    "properties": {
        "status": {"type": "string", "enum": ["success", "error"]},
        "message": {"type": "string"}
    },
    "required": ["status", "message"],
    "additionalProperties": False
}

# 用户输入
input_messages = [
    {"role": "system", "content": "你是一个财务管理助手，提取支出信息并保存到数据库。"},
    {"role": "user", "content": "昨天我花了50元买了午餐"}
]

# 第一次调用：提取并调用函数
response = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=input_messages,
    tools=tools
)

# 处理函数调用
tool_call = response.output[0]
if tool_call.type == "function_call":
    args = json.loads(tool_call.arguments)

    def save_expense(date, amount, category, note):
        # 模拟数据库保存
        return f"记录保存成功：{date}, {amount}元, {category}, {note}"

    result = save_expense(**args)

    # 将函数调用和结果追加到消息中
    input_messages.append(tool_call)
    input_messages.append({
        "type": "function_call_output",
        "call_id": tool_call.call_id,
        "output": result
    })

# 第二次调用：生成结构化响应
response_2 = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=input_messages,
    text={
        "format": {
            "type": "json_schema",
            "name": "save_response",
            "schema": response_schema,
            "strict": True
        }
    }
)

final_response = json.loads(response_2.output_text)
print(json.dumps(final_response, indent=2, ensure_ascii=False))

输出

{
  "status": "success",
  "message": "记录保存成功：2025-03-15, 50元, 餐饮, 买了午餐"
}

流程说明：

第一次调用识别并调用 save_expense 函数。
执行函数，模拟保存到数据库。
第二次调用使用 Structured Outputs 返回最终状态。

优化建议

动态日期处理：
- 在系统提示中明确日期推断规则，如“‘昨天’应转换为当前日期减一天”。
- 示例："将相对日期（如‘昨天’）转换为 YYYY-MM-DD 格式，基于当前日期 2025-03-16。"
错误处理增强：
- 添加对无效金额或类别的验证。
- 示例：若用户输入“花了abc元”，返回 {"status": "error", "message": "金额无效"}。

多记录支持：

修改 Schema 支持数组，如：

{
    "type": "array",
    "items": {"$ref": "#/definitions/expense"},
    "definitions": {"expense": {...}}
}

流式输出：
- 对于长响应，使用 stream=True 实时显示结果。

示例 1：健康记录场景实现

场景描述

我们要构建一个健康管理助手，用户可以输入自然语言（如“今天早上我跑了5公里，心率达到120次/分钟”），助手将提取健康数据并以结构化 JSON 格式返回，包括日期、活动类型、持续时间、心率等信息。

步骤 1：定义 JSON Schema

{
    "type": "object",
    "properties": {
        "date": {
            "type": "string",
            "description": "活动日期，格式为 YYYY-MM-DD"
        },
        "activity": {
            "type": "string",
            "enum": ["跑步", "游泳", "骑行", "瑜伽", "其他"],
            "description": "活动类型"
        },
        "duration": {
            "type": ["number", "null"],
            "description": "活动持续时间（分钟），若未知则为 null"
        },
        "heart_rate": {
            "type": ["number", "null"],
            "description": "平均心率（次/分钟），若未知则为 null"
        },
        "notes": {
            "type": ["string", "null"],
            "description": "附加备注，若无则为 null"
        }
    },
    "required": ["date", "activity", "duration", "heart_rate", "notes"],
    "additionalProperties": false
}

设计要点：

duration 和 heart_rate 可选，使用 "type": ["number", "null"]。
activity 使用枚举限制常见类型。
date 要求标准格式。

步骤 2：实现代码

from openai import OpenAI
import json
from datetime import datetime

client = OpenAI()

# 定义 Schema
health_schema = {
    "type": "object",
    "properties": {
        "date": {"type": "string", "description": "活动日期，格式为 YYYY-MM-DD"},
        "activity": {"type": "string", "enum": ["跑步", "游泳", "骑行", "瑜伽", "其他"]},
        "duration": {"type": ["number", "null"], "description": "活动持续时间（分钟）"},
        "heart_rate": {"type": ["number", "null"], "description": "平均心率（次/分钟）"},
        "notes": {"type": ["string", "null"], "description": "附加备注"}
    },
    "required": ["date", "activity", "duration", "heart_rate", "notes"],
    "additionalProperties": False
}

# 当前日期
today = datetime.now().strftime("%Y-%m-%d")

response = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=[
        {
            "role": "system",
            "content": "你是一个健康管理助手，从用户输入中提取结构化健康数据。‘今天’指 {today}，若信息缺失则返回 null。"
        },
        {"role": "user", "content": "今天早上我跑了5公里，心率达到120次/分钟"}
    ],
    text={
        "format": {
            "type": "json_schema",
            "name": "health_record",
            "schema": health_schema,
            "strict": True
        }
    }
)

health_record = json.loads(response.output_text)
print(json.dumps(health_record, indent=2, ensure_ascii=False))

输出

{
  "date": "2025-03-16",
  "activity": "跑步",
  "duration": null,
  "heart_rate": 120,
  "notes": "跑了5公里"
}

解析说明：

date：从“今天”推断为 2025-03-16。
activity：识别为“跑步”。
duration：未提供分钟数，返回 null。
heart_rate：提取为 120。
notes：记录“跑了5公里”。

步骤 3：优化与错误处理

try:
    response = client.responses.create(
        model="gpt-4o-2024-08-06",
        input=[
            {
                "role": "system",
                "content": f"你是一个健康管理助手，从用户输入中提取结构化健康数据。‘今天’指 {today}，若信息缺失则返回 null。"
            },
            {"role": "user", "content": "今天早上我跑了5公里，心率达到120次/分钟"}
        ],
        text={
            "format": {"type": "json_schema", "name": "health_record", "schema": health_schema, "strict": True}
        }
    )

    if response.status == "incomplete":
        print(f"响应不完整：{response.incomplete_details.reason}")
    elif response.output[0].content[0].type == "refusal":
        print(f"模型拒绝：{response.output[0].content[0].refusal}")
    else:
        health_record = json.loads(response.output_text)
        print(json.dumps(health_record, indent=2, ensure_ascii=False))

except Exception as e:
    print(f"错误：{e}")

示例 2：任务管理（复杂 Schema 设计）

场景描述

构建一个任务管理助手，支持嵌套子任务和递归结构，用户输入（如“明天完成项目报告，包括收集数据和撰写初稿”），返回任务及其子任务的结构化数据。

步骤 1：定义复杂 JSON Schema

使用递归结构表示任务和子任务：

{
    "type": "object",
    "properties": {
        "task_id": {
            "type": "string",
            "description": "唯一任务ID"
        },
        "title": {
            "type": "string",
            "description": "任务标题"
        },
        "due_date": {
            "type": "string",
            "description": "截止日期，格式 YYYY-MM-DD"
        },
        "subtasks": {
            "type": "array",
            "description": "子任务列表",
            "items": {"$ref": "#"}
        },
        "status": {
            "type": "string",
            "enum": ["待办", "进行中", "已完成"],
            "description": "任务状态"
        }
    },
    "required": ["task_id", "title", "due_date", "subtasks", "status"],
    "additionalProperties": false
}

设计要点：

subtasks 使用 "$ref": "#" 表示递归引用。
task_id 确保唯一性。
status 使用枚举限制状态。

步骤 2：实现代码

from openai import OpenAI
import json
from datetime import datetime, timedelta
import uuid

client = OpenAI()

# 定义 Schema
task_schema = {
    "type": "object",
    "properties": {
        "task_id": {"type": "string", "description": "唯一任务ID"},
        "title": {"type": "string", "description": "任务标题"},
        "due_date": {"type": "string", "description": "截止日期，格式 YYYY-MM-DD"},
        "subtasks": {"type": "array", "description": "子任务列表", "items": {"$ref": "#"}},
        "status": {"type": "string", "enum": ["待办", "进行中", "已完成"]}
    },
    "required": ["task_id", "title", "due_date", "subtasks", "status"],
    "additionalProperties": False
}

# 计算明天日期
tomorrow = (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%d")

response = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=[
        {
            "role": "system",
            "content": f"你是一个任务管理助手，生成结构化任务数据。‘明天’指 {tomorrow}，为每个任务生成唯一 task_id（如 UUID）。"
        },
        {"role": "user", "content": "明天完成项目报告，包括收集数据和撰写初稿"}
    ],
    text={
        "format": {
            "type": "json_schema",
            "name": "task",
            "schema": task_schema,
            "strict": True
        }
    }
)

task = json.loads(response.output_text)
print(json.dumps(task, indent=2, ensure_ascii=False))

输出

{
  "task_id": "a1b2c3d4-5678-90ef-ghij-klmn",
  "title": "完成项目报告",
  "due_date": "2025-03-17",
  "subtasks": [
    {
      "task_id": "e5f6g7h8-9012-34ij-klmn-opqr",
      "title": "收集数据",
      "due_date": "2025-03-17",
      "subtasks": [],
      "status": "待办"
    },
    {
      "task_id": "i9j0k1l2-3456-78mn-opqr-stuv",
      "title": "撰写初稿",
      "due_date": "2025-03-17",
      "subtasks": [],
      "status": "待办"
    }
  ],
  "status": "待办"
}

解析说明：

主任务“完成项目报告”包含两个子任务。
每个任务都有唯一 task_id（UUID）。
due_date 推断为明天（2025-03-17）。

步骤 3：优化与调试

调试支持

若输出不符合预期（如子任务缺失），可能原因及解决方法：

提示不明确：
- 问题：模型未识别“收集数据”和“撰写初稿”为子任务。
- 解决：调整系统提示，添加“将‘包括’后的内容拆分为子任务”。
- 示例："将‘包括’后的内容拆分为独立的子任务，每个子任务需有唯一 task_id 和默认状态‘待办’。"
Schema 限制：
- 问题：嵌套层级超过 5 层（Structured Outputs 限制）。
- 解决：检查输出，确保不超过限制，或简化结构。

优化代码

try:
    response = client.responses.create(
        model="gpt-4o-2024-08-06",
        input=[
            {
                "role": "system",
                "content": f"你是一个任务管理助手，生成结构化任务数据。‘明天’指 {tomorrow}，为每个任务生成唯一 task_id（如 UUID）。将‘包括’后的内容拆分为子任务。"
            },
            {"role": "user", "content": "明天完成项目报告，包括收集数据和撰写初稿"}
        ],
        text={
            "format": {"type": "json_schema", "name": "task", "schema": task_schema, "strict": True}
        }
    )

    if response.status == "incomplete":
        print(f"不完整：{response.incomplete_details.reason}")
    elif response.output[0].content[0].type == "refusal":
        print(f"拒绝：{response.output[0].content[0].refusal}")
    else:
        task = json.loads(response.output_text)
        print(json.dumps(task, indent=2, ensure_ascii=False))

except Exception as e:
    print(f"错误：{e}")

调试支持：常见问题及优化建议

问题：模型未填充所有字段
- 原因：输入信息不足或提示未明确要求填充。
- 解决：在系统提示中添加默认值规则，如“若持续时间未知，返回 null”。
问题：输出不符合 Schema
- 原因：Schema 定义错误（如漏写 required）。
- 解决：检查 Schema，确保 additionalProperties: false 和所有字段在 required 中。
问题：复杂嵌套导致性能下降
- 原因：递归结构过深或属性过多。
- 解决：简化 Schema，或使用 Function Calling 分担复杂逻辑。

示例调试代码

假设健康记录示例中 heart_rate 未正确提取：

# 修改提示以明确要求
response = client.responses.create(
    model="gpt-4o-2024-08-06",
    input=[
        {
            "role": "system",
            "content": f"你是一个健康管理助手，从用户输入中提取结构化健康数据。‘今天’指 {today}，若信息缺失则返回 null。明确提取‘心率’并以数字表示。"
        },
        {"role": "user", "content": "今天早上我跑了5公里，心率达到120次/分钟"}
    ],
    text={
        "format": {"type": "json_schema", "name": "health_record", "schema": health_schema, "strict": True}
    }
)

health_record = json.loads(response.output_text)
print(json.dumps(health_record, indent=2, ensure_ascii=False))