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, timedeltaclient = 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, timedeltaclient = 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元, 餐饮, 买了午餐"
}

流程说明：

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

---

优化建议

1. 动态日期处理：

   • 在系统提示中明确日期推断规则，如“‘昨天’应转换为当前日期减一天”。
   • 示例："将相对日期（如‘昨天’）转换为 YYYY-MM-DD 格式，基于当前日期 2025-03-16。"

2. 错误处理增强：

   • 添加对无效金额或类别的验证。
   • 示例：若用户输入“花了abc元”，返回 {"status": "error", "message": "金额无效"}。

3. 多记录支持：

   • 修改 Schema 支持数组，如：

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

4. 流式输出：

   • 对于长响应，使用 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 datetimeclient = 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 uuidclient = 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：优化与调试

调试支持

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

1. 提示不明确：

   • 问题：模型未识别“收集数据”和“撰写初稿”为子任务。
   • 解决：调整系统提示，添加“将‘包括’后的内容拆分为子任务”。
   • 示例："将‘包括’后的内容拆分为独立的子任务，每个子任务需有唯一 task_id 和默认状态‘待办’。"

2. 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}")

---

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

1. 问题：模型未填充所有字段

   • 原因：输入信息不足或提示未明确要求填充。
   • 解决：在系统提示中添加默认值规则，如“若持续时间未知，返回 null”。

2. 问题：输出不符合 Schema

   • 原因：Schema 定义错误（如漏写 required）。
   • 解决：检查 Schema，确保 additionalProperties: false 和所有字段在 required 中。

3. 问题：复杂嵌套导致性能下降

   • 原因：递归结构过深或属性过多。
   • 解决：简化 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))