Structured Outputs 与 Function Calling：怎么选、怎么验收

{
  "type": "object",
  "properties": {
    "state": { "type": "string", "enum": ["ready", "needs_input"] },
    "visit_date": { "type": ["string", "null"] },
    "headcount": { "type": ["integer", "null"] },
    "question": { "type": ["string", "null"] }
  },
  "required": ["state", "visit_date", "headcount", "question"],
  "additionalProperties": false
}

from datetime import date
from typing import Literal
from openai import OpenAI
from pydantic import BaseModel

class RoomRequest(BaseModel):
    state: Literal["ready", "needs_input"]
    visit_date: str | None
    headcount: int | None
    question: str | None

def validate_request(value):
    if value.visit_date is not None:
        date.fromisoformat(value.visit_date)
    if value.headcount is not None and value.headcount < 1:
        raise ValueError("人数必须为正整数")
    complete = value.visit_date is not None and value.headcount is not None
    if value.state == "ready":
        if not complete or value.question is not None:
            raise ValueError("ready 与字段内容不一致")
    elif complete or not value.question or not value.question.strip():
        raise ValueError("needs_input 必须对应缺失信息和明确追问")
    return value

client = OpenAI()
response = client.responses.parse(
    model="gpt-6-astra",
    input=[
        {"role": "system", "content":
         "提取教室需求。日期必须包含年、月、日；未给年份不要猜。"
         "缺日期或人数时使用 needs_input，缺失值设 null 并追问；"
         "都齐全时使用 ready，question 设 null。不要声称已查课表。"},
        {"role": "user", "content": "10 月 12 日需要容纳 18 人的教室。"}
    ],
    text_format=RoomRequest,
)
if response.status != "completed":
    raise RuntimeError(f"回答未完成：{response.incomplete_details}")
for item in response.output:
    if item.type == "message":
        for part in item.content:
            if part.type == "refusal":
                raise RuntimeError(f"请求被拒绝：{part.refusal}")
if response.output_parsed is None:
    raise RuntimeError("没有可消费的结构化回答")
value = validate_request(response.output_parsed)
print(value.model_dump())

import json
from datetime import date
from openai import OpenAI

TOOLS = [{
    "type": "function",
    "name": "lookup_room_day",
    "description": "查询当前用户可访问教室在指定日期的课表状态。",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {
            "room_code": {"type": "string"},
            "visit_date": {"type": "string"}
        },
        "required": ["room_code", "visit_date"],
        "additionalProperties": False
    }
}]
CARD_SCHEMA = {
    "type": "object",
    "properties": {
        "visit_date": {"type": "string"},
        "rooms": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "room_code": {"type": "string"},
                    "available": {"type": "boolean"},
                    "seats": {"type": "integer"},
                    "checked_at": {"type": "string"}
                },
                "required": ["room_code", "available", "seats", "checked_at"],
                "additionalProperties": False
            }
        }
    },
    "required": ["visit_date", "rooms"],
    "additionalProperties": False
}
DEMO_SCHEDULE = {
    ("A201", "2026-10-12"): {
        "available": True, "seats": 24,
        "checked_at": "2026-10-07T09:00:00+08:00"},
    ("A202", "2026-10-12"): {
        "available": False, "seats": 32,
        "checked_at": "2026-10-07T09:00:00+08:00"}
}

def require_completed(response):
    if response.status != "completed":
        raise RuntimeError(f"未完成：{response.incomplete_details}")
    for item in response.output:
        if item.type == "message":
            for part in item.content:
                if part.type == "refusal":
                    raise RuntimeError(f"拒绝：{part.refusal}")

def lookup_room_day(args, allowed_rooms, target_date):
    if not isinstance(args, dict) or set(args) != {"room_code", "visit_date"}:
        return {"ok": False, "error": "invalid_arguments"}
    room, day = args["room_code"], args["visit_date"]
    if not isinstance(room, str) or not isinstance(day, str):
        return {"ok": False, "error": "invalid_arguments"}
    try:
        date.fromisoformat(day)
    except ValueError:
        return {"ok": False, "error": "invalid_date"}
    if day != target_date:
        return {"ok": False, "error": "unexpected_date"}
    if room not in allowed_rooms:
        return {"ok": False, "error": "not_allowed"}
    row = DEMO_SCHEDULE.get((room, day))
    if row is None:
        return {"ok": False, "error": "not_found"}
    return {"ok": True, "room_code": room, "visit_date": day, **row}

def validate_card(card, evidence, target_rooms, target_date):
    if not isinstance(card, dict) or set(card) != {"visit_date", "rooms"}:
        raise ValueError("卡片字段错误")
    if card["visit_date"] != target_date or not isinstance(card["rooms"], list):
        raise ValueError("卡片日期或列表错误")
    by_room = {}
    for row in card["rooms"]:
        if not isinstance(row, dict) or set(row) != {
            "room_code", "available", "seats", "checked_at"
        }:
            raise ValueError("教室字段错误")
        room = row["room_code"]
        if not isinstance(room, str) or room in by_room or room not in evidence:
            raise ValueError("重复或没有查询依据的教室")
        if type(row["available"]) is not bool or type(row["seats"]) is not int:
            raise ValueError("教室字段类型错误")
        expected = {key: evidence[room][key] for key in row}
        if row != expected:
            raise ValueError("卡片与查询结果不一致")
        by_room[room] = row
    if set(by_room) != set(target_rooms):
        raise ValueError("查询卡片有遗漏或多余教室")
    return card

def query_then_card(client, target_rooms, target_date, allowed_rooms):
    date.fromisoformat(target_date)
    if not target_rooms or not set(target_rooms) <= set(allowed_rooms):
        raise ValueError("请求教室超出访问范围")
    history = [{"role": "user", "content":
                f"查询 {target_date} 的这些教室：{', '.join(target_rooms)}。"}]
    evidence, seen = {}, {}
    for _ in range(4):
        response = client.responses.create(
            model="gpt-6-astra", input=history, tools=TOOLS,
            instructions="使用工具逐一查询请求的教室，不猜课表。",
            tool_choice="auto"
        )
        require_completed(response)
        history.extend(response.output)
        calls = [item for item in response.output if item.type == "function_call"]
        if not calls:
            raise RuntimeError("尚无完整查询结果，模型已停止调用")
        for call in calls:
            fingerprint = (call.name, call.arguments)
            if call.call_id in seen:
                old_fingerprint, result = seen[call.call_id]
                if old_fingerprint != fingerprint:
                    raise RuntimeError("同一 call_id 对应了不同调用")
            else:
                if call.name != "lookup_room_day":
                    result = {"ok": False, "error": "unknown_tool"}
                else:
                    try:
                        args = json.loads(call.arguments)
                    except (TypeError, ValueError):
                        result = {"ok": False, "error": "invalid_json"}
                    else:
                        result = lookup_room_day(args, allowed_rooms, target_date)
                seen[call.call_id] = (fingerprint, result)
            if result.get("ok") is True:
                evidence[result["room_code"]] = result
            history.append({
                "type": "function_call_output", "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False)
            })
        if set(target_rooms) <= set(evidence):
            break
    else:
        raise RuntimeError("查询轮次耗尽，不能生成成功卡片")
    final = client.responses.create(
        model="gpt-6-astra", input=history,
        instructions="只根据成功的工具结果生成所请求教室的卡片，不添加教室。",
        text={"format": {
            "type": "json_schema", "name": "room_day_card",
            "strict": True, "schema": CARD_SCHEMA
        }}
    )
    require_completed(final)
    card = json.loads(final.output_text)
    return validate_card(card, evidence, target_rooms, target_date)

if __name__ == "__main__":
    client = OpenAI()
    card = query_then_card(
        client, ["A201", "A202"], "2026-10-12", {"A201", "A202"}
    )
    print(json.dumps(card, ensure_ascii=False, indent=2))

{
  "wrong_nullable_enum": {
    "type": ["string", "null"],
    "enum": ["morning", "afternoon"]
  },
  "nullable_enum": {
    "type": ["string", "null"],
    "enum": ["morning", "afternoon", null]
  }
}