121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
|
# contracts/verify_possibility.py
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from dataclasses import dataclass, field
|
|||
|
|
from typing import Literal, Protocol, Annotated, Union, Any
|
|||
|
|
|
|||
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|||
|
|
|
|||
|
|
from contracts.base import _StrictModel
|
|||
|
|
from contracts.coders.GatherOutput import GatherOutput
|
|||
|
|
|
|||
|
|
@dataclass(frozen=True)
|
|||
|
|
class VerifyPossibilityInput:
|
|||
|
|
gather: GatherOutput
|
|||
|
|
|
|||
|
|
|
|||
|
|
def sanitize_for_openai(schema: dict[str, Any]) -> dict[str, Any]:
|
|||
|
|
"""
|
|||
|
|
Recursively rewrite a Pydantic-generated JSON schema into the subset
|
|||
|
|
OpenAI strict mode accepts.
|
|||
|
|
|
|||
|
|
- oneOf -> anyOf (OpenAI rejects oneOf)
|
|||
|
|
- discriminator -> removed (OpenAI rejects it; the const fields
|
|||
|
|
already make branches disjoint)
|
|||
|
|
- adds additionalProperties: false to every object missing it
|
|||
|
|
- strips None defaults
|
|||
|
|
"""
|
|||
|
|
return _walk(schema)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _walk(node: Any) -> Any:
|
|||
|
|
if isinstance(node, list):
|
|||
|
|
return [_walk(x) for x in node]
|
|||
|
|
if not isinstance(node, dict):
|
|||
|
|
return node
|
|||
|
|
|
|||
|
|
# Recurse into every nested schema container first.
|
|||
|
|
for key in ("properties", "$defs", "definitions"):
|
|||
|
|
if key in node and isinstance(node[key], dict):
|
|||
|
|
node[key] = {k: _walk(v) for k, v in node[key].items()}
|
|||
|
|
|
|||
|
|
for key in ("items", "additionalProperties"):
|
|||
|
|
if key in node and isinstance(node[key], (dict, list)):
|
|||
|
|
node[key] = _walk(node[key])
|
|||
|
|
|
|||
|
|
for key in ("anyOf", "allOf"):
|
|||
|
|
if key in node and isinstance(node[key], list):
|
|||
|
|
node[key] = [_walk(v) for v in node[key]]
|
|||
|
|
|
|||
|
|
# ── the two rewrites that matter ──────────────────────────
|
|||
|
|
if "oneOf" in node and isinstance(node["oneOf"], list):
|
|||
|
|
existing = node.get("anyOf", [])
|
|||
|
|
if not isinstance(existing, list):
|
|||
|
|
existing = []
|
|||
|
|
node["anyOf"] = existing + [_walk(v) for v in node["oneOf"]]
|
|||
|
|
node.pop("oneOf")
|
|||
|
|
|
|||
|
|
node.pop("discriminator", None) # OpenAI doesn't accept it
|
|||
|
|
|
|||
|
|
# ── strict-mode hygiene ───────────────────────────────────
|
|||
|
|
if node.get("type") == "object":
|
|||
|
|
node.setdefault("additionalProperties", False)
|
|||
|
|
props = node.get("properties")
|
|||
|
|
if isinstance(props, dict):
|
|||
|
|
# Strict mode: every property must be in required.
|
|||
|
|
node["required"] = list(props.keys())
|
|||
|
|
|
|||
|
|
if node.get("default", object()) is None:
|
|||
|
|
node.pop("default", None)
|
|||
|
|
|
|||
|
|
return node
|
|||
|
|
|
|||
|
|
class Option(_StrictModel):
|
|||
|
|
label: str = Field(min_length=1, max_length=80)
|
|||
|
|
description: str = Field(min_length=1)
|
|||
|
|
tradeoff: str | None = Field(
|
|||
|
|
description="Optional trade-off; null if none.",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
class _BaseVerdict(_StrictModel):
|
|||
|
|
reasoning: str = Field(
|
|||
|
|
min_length=1,
|
|||
|
|
description="2–5 sentences explaining the verdict.",
|
|||
|
|
)
|
|||
|
|
confidence: float = Field(
|
|||
|
|
ge=0.0, le=1.0,
|
|||
|
|
description="Your honest probability that the verdict is correct.",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class FeasibleVerdict(_BaseVerdict):
|
|||
|
|
verdict: Literal["feasible"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
class AmbiguousVerdict(_BaseVerdict):
|
|||
|
|
verdict: Literal["ambiguous"]
|
|||
|
|
question: str = Field(
|
|||
|
|
min_length=1,
|
|||
|
|
description="Exactly one clarifying question that unblocks planning.",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class InfeasibleVerdict(_BaseVerdict):
|
|||
|
|
verdict: Literal["infeasible"]
|
|||
|
|
options: list[Option] = Field(
|
|||
|
|
min_length=1, max_length=3,
|
|||
|
|
description="1–3 concrete alternatives.",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
Verdict = Annotated[
|
|||
|
|
Union[FeasibleVerdict, AmbiguousVerdict, InfeasibleVerdict],
|
|||
|
|
Field(discriminator="verdict"),
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
class VerifyPossibilityOutput(_StrictModel):
|
|||
|
|
"""Top-level object required by OpenAI strict mode."""
|
|||
|
|
result: Verdict = Field(
|
|||
|
|
description="The feasibility verdict.",
|
|||
|
|
)
|