Structured OutputsとFunction Callingの違い：最終JSONとツール実行の使い分け

{
  "type": "object",
  "properties": {
    "equipment_id": {"type": ["string", "null"]},
    "availability": {
      "type": "string",
      "enum": ["available", "unavailable", "unknown"]
    },
    "remaining": {"type": ["integer", "null"]},
    "checked_at": {"type": ["string", "null"]}
  },
  "required": ["equipment_id", "availability", "remaining", "checked_at"],
  "additionalProperties": false
}

import json

CARD_SCHEMA = {
    "type": "object",
    "properties": {
        "equipment_id": {"type": ["string", "null"]},
        "availability": {
            "type": "string",
            "enum": ["available", "unavailable", "unknown"],
        },
        "remaining": {"type": ["integer", "null"]},
        "checked_at": {"type": ["string", "null"]},
    },
    "required": ["equipment_id", "availability", "remaining", "checked_at"],
    "additionalProperties": False,
}

def ensure_completed(response):
    if response.status != "completed":
        details = getattr(response, "incomplete_details", None)
        reason = getattr(details, "reason", response.status)
        raise RuntimeError(f"応答を使用できません: {reason}")
    for item in response.output:
        if item.type == "message":
            for part in item.content:
                if part.type == "refusal":
                    raise RuntimeError(f"拒否: {part.refusal}")

def expected_card(record):
    if not isinstance(record, dict) or type(record.get("ok")) is not bool:
        raise ValueError("照会結果の成否が不正です")
    equipment_id = record.get("equipment_id")
    if equipment_id is not None and not isinstance(equipment_id, str):
        raise ValueError("照会結果の機材番号が不正です")
    if record["ok"]:
        if not equipment_id:
            raise ValueError("成功した照会の機材番号がありません")
        count = record["remaining"]
        if type(count) is not int or count < 0:
            raise ValueError("照会元の台数が不正です")
        stamp = record["checked_at"]
        if not isinstance(stamp, str) or not stamp.strip():
            raise ValueError("照会元の確認日時がありません")
        return {
            "equipment_id": record["equipment_id"],
            "availability": "available" if count > 0 else "unavailable",
            "remaining": count,
            "checked_at": stamp,
        }
    return {
        "equipment_id": record["equipment_id"],
        "availability": "unknown",
        "remaining": None,
        "checked_at": None,
    }

def read_card(response, record):
    ensure_completed(response)
    if any(item.type == "function_call" for item in response.output):
        raise RuntimeError("最終出力に未処理の関数呼び出しがあります")
    if not response.output_text:
        raise RuntimeError("最終JSONがありません")
    card = json.loads(response.output_text)
    if not isinstance(card, dict) or set(card) != set(CARD_SCHEMA["properties"]):
        raise ValueError("画面データの項目が違います")
    if card["equipment_id"] is not None and not isinstance(card["equipment_id"], str):
        raise ValueError("機材番号の型が違います")
    if card["availability"] not in {"available", "unavailable", "unknown"}:
        raise ValueError("貸出状態が不正です")
    if card["remaining"] is not None and type(card["remaining"]) is not int:
        raise ValueError("台数は整数またはnullで指定してください")
    if card["checked_at"] is not None and not isinstance(card["checked_at"], str):
        raise ValueError("確認日時の型が違います")
    if card != expected_card(record):
        raise ValueError("最終JSONが照会結果と一致しません")
    return card

def final_card(client, model, record):
    expected_card(record)
    response = client.responses.create(
        model=model,
        input=[{
            "role": "user",
            "content": (
                "この照会結果を画面用JSONにしてください。ok=falseなら"
                "availability=unknown、remainingとchecked_atはnullです。"
                "機材番号は維持し、台数や日時を推測しないでください。"
                + json.dumps(record, ensure_ascii=False)
            ),
        }],
        text={"format": {
            "type": "json_schema", "name": "equipment_card",
            "strict": True, "schema": CARD_SCHEMA,
        }},
    )
    return read_card(response, record)

{
  "equipment_id": "CAM-17",
  "availability": "available",
  "remaining": 2,
  "checked_at": "2026-10-07T03:00:00Z"
}

READ_TOOL = {
    "type": "function",
    "name": "read_equipment",
    "description": "指定した機材の貸出可能台数と確認日時を取得する。",
    "strict": True,
    "parameters": {
        "type": "object",
        "properties": {"equipment_id": {"type": "string"}},
        "required": ["equipment_id"],
        "additionalProperties": False,
    },
}

FIXTURES = {
    "CAM-17": {"remaining": 2, "checked_at": "2026-10-07T03:00:00Z"},
    "MIC-08": {"remaining": 0, "checked_at": "2026-10-07T03:00:00Z"},
}

def read_equipment(equipment_id):
    row = FIXTURES.get(equipment_id)
    if row is None:
        return {"ok": False, "equipment_id": equipment_id, "error": "not_found"}
    return {"ok": True, "equipment_id": equipment_id, **row}

def execute_call(call, selected_id, allowed_ids, handler):
    if call.name != "read_equipment":
        raise ValueError("許可していない関数です")
    args = json.loads(call.arguments)
    if not isinstance(args, dict) or set(args) != {"equipment_id"}:
        raise ValueError("照会引数が違います")
    equipment_id = args["equipment_id"]
    if not isinstance(equipment_id, str) or equipment_id != selected_id:
        raise ValueError("依頼した機材と照会対象が違います")
    if equipment_id not in allowed_ids:
        raise PermissionError("この機材は照会できません")
    try:
        result = handler(equipment_id)
    except TimeoutError:
        result = {"ok": False, "equipment_id": equipment_id, "error": "timeout"}
    if not isinstance(result, dict) or type(result.get("ok")) is not bool:
        raise ValueError("照会処理の戻り値が不正です")
    if result.get("equipment_id") != equipment_id:
        raise ValueError("戻り値の機材番号が違います")
    expected_card(result)
    return result

def lookup(client, model, selected_id, allowed_ids, structured=False,
           handler=read_equipment):
    if not isinstance(selected_id, str) or not selected_id.strip():
        raise ValueError("機材番号を先に確認してください")
    if selected_id not in allowed_ids:
        raise PermissionError("照会対象が許可されていません")
    inputs = [{"role": "user", "content": f"機材{selected_id}の貸出状況を照会して"}]
    seen = {}
    last_result = None
    for turn in range(4):
        response = client.responses.create(
            model=model, input=inputs, tools=[READ_TOOL],
            tool_choice="required" if turn == 0 else "auto",
            parallel_tool_calls=False,
        )
        ensure_completed(response)
        inputs.extend(response.output)
        calls = [item for item in response.output if item.type == "function_call"]
        if not calls:
            if last_result is None:
                raise RuntimeError("必要な照会結果がありません")
            if structured:
                return final_card(client, model, last_result)
            if not response.output_text:
                raise RuntimeError("最終説明がありません")
            return {"record": last_result, "explanation": response.output_text}
        for call in calls:
            if not isinstance(call.call_id, str) or not call.call_id:
                raise ValueError("call_idがありません")
            signature = (call.name, call.arguments)
            if call.call_id in seen:
                old_signature, result = seen[call.call_id]
                if old_signature != signature:
                    raise ValueError("同じcall_idで引数が変わりました")
            else:
                result = execute_call(call, selected_id, allowed_ids, handler)
                seen[call.call_id] = (signature, result)
            last_result = result
            inputs.append({
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result, ensure_ascii=False),
            })
    raise RuntimeError("照会の往復回数が上限に達しました")