chwellofficial/nt360Slides
0
1from copy import deepcopy2from typing import Any, List3 4from openai import NOT_GIVEN5 6from utils.dict_utils import (7 get_dict_paths_with_key,8 get_dict_at_path,9 has_more_than_n_keys,10)11 12supported_string_formats = [13 "date-time",14 "time",15 "date",16 "duration",17 "email",18 "hostname",19 "ipv4",20 "ipv6",21 "uuid",22]23 24 25def remove_fields_from_schema(schema: dict, fields_to_remove: List[str]):26 schema = deepcopy(schema)27 properties_paths = get_dict_paths_with_key(schema, "properties")28 for path in properties_paths:29 parent_obj = get_dict_at_path(schema, path)30 if "properties" in parent_obj and isinstance(parent_obj["properties"], dict):31 for field in fields_to_remove:32 if field in parent_obj["properties"]:33 del parent_obj["properties"][field]34 35 required_paths = get_dict_paths_with_key(schema, "required")36 for path in required_paths:37 parent_obj = get_dict_at_path(schema, path)38 if "required" in parent_obj and isinstance(parent_obj["required"], list):39 parent_obj["required"] = [40 field41 for field in parent_obj["required"]42 if field not in fields_to_remove43 ]44 45 return schema46 47 48def add_field_in_schema(schema: dict, field: dict, required: bool = False) -> dict:49 50 if not isinstance(field, dict) or len(field) != 1:51 raise ValueError(52 "`field` must be a dict with exactly one entry: {name: schema_dict}"53 )54 55 field_name, field_schema = next(iter(field.items()))56 if not isinstance(field_name, str):57 raise TypeError("Field name must be a string")58 if not isinstance(field_schema, dict):59 raise TypeError("Field schema must be a dictionary")60 61 updated_schema: dict = deepcopy(schema)62 63 root_properties = updated_schema.get("properties")64 if not isinstance(root_properties, dict):65 updated_schema["properties"] = {}66 root_properties = updated_schema["properties"]67 68 root_properties[field_name] = field_schema69 70 # Update root-level required based on the flag71 existing_required = updated_schema.get("required")72 if not isinstance(existing_required, list):73 existing_required = []74 75 if required:76 if field_name not in existing_required:77 existing_required.append(field_name)78 else:79 if field_name in existing_required:80 existing_required = [name for name in existing_required if name != field_name]81 82 if existing_required:83 updated_schema["required"] = existing_required84 else:85 updated_schema.pop("required", None)86 87 return updated_schema88 89 90# From OpenAI91def ensure_strict_json_schema(92 json_schema: object,93 *,94 path: tuple[str, ...],95 root: dict[str, object],96) -> dict[str, Any]:97 """Mutates the given JSON schema to ensure it conforms to the `strict` standard98 that the API expects.99 """100 if not isinstance(json_schema, dict):101 raise TypeError(f"Expected {json_schema} to be a dictionary; path={path}")102 103 defs = json_schema.get("$defs")104 if isinstance(defs, dict):105 for def_name, def_schema in defs.items():106 ensure_strict_json_schema(107 def_schema, path=(*path, "$defs", def_name), root=root108 )109 110 definitions = json_schema.get("definitions")111 if isinstance(definitions, dict):112 for definition_name, definition_schema in definitions.items():113 ensure_strict_json_schema(114 definition_schema,115 path=(*path, "definitions", definition_name),116 root=root,117 )118 119 typ = json_schema.get("type")120 if typ == "object" and "additionalProperties" not in json_schema:121 json_schema["additionalProperties"] = False122 123 # object types124 # { 'type': 'object', 'properties': { 'a': {...} } }125 properties = json_schema.get("properties")126 if isinstance(properties, dict):127 json_schema["required"] = [prop for prop in properties.keys()]128 json_schema["properties"] = {129 key: ensure_strict_json_schema(130 prop_schema, path=(*path, "properties", key), root=root131 )132 for key, prop_schema in properties.items()133 }134 135 # arrays136 # { 'type': 'array', 'items': {...} }137 # OpenAI requires array schemas to have "items". Zod tuples may emit prefixItems only.138 items = json_schema.get("items")139 if isinstance(items, dict):140 json_schema["items"] = ensure_strict_json_schema(141 items, path=(*path, "items"), root=root142 )143 elif typ == "array":144 prefix_items = json_schema.get("prefixItems")145 if (146 isinstance(prefix_items, list)147 and len(prefix_items) > 0148 and isinstance(prefix_items[0], dict)149 ):150 json_schema["items"] = ensure_strict_json_schema(151 prefix_items[0], path=(*path, "items"), root=root152 )153 json_schema.pop("prefixItems", None)154 else:155 json_schema["items"] = {"type": "string"}156 157 # unions158 any_of = json_schema.get("anyOf")159 if isinstance(any_of, list):160 json_schema["anyOf"] = [161 ensure_strict_json_schema(variant, path=(*path, "anyOf", str(i)), root=root)162 for i, variant in enumerate(any_of)163 ]164 165 # intersections166 all_of = json_schema.get("allOf")167 if isinstance(all_of, list):168 if len(all_of) == 1:169 json_schema.update(170 ensure_strict_json_schema(171 all_of[0], path=(*path, "allOf", "0"), root=root172 )173 )174 json_schema.pop("allOf")175 else:176 json_schema["allOf"] = [177 ensure_strict_json_schema(178 entry, path=(*path, "allOf", str(i)), root=root179 )180 for i, entry in enumerate(all_of)181 ]182 183 # string184 if typ == "string":185 if "format" in json_schema:186 if json_schema["format"] not in supported_string_formats:187 del json_schema["format"]188 189 # strip `None` defaults as there's no meaningful distinction here190 # the schema will still be `nullable` and the model will default191 # to using `None` anyway192 if json_schema.get("default", NOT_GIVEN) is None:193 json_schema.pop("default")194 195 # we can't use `$ref`s if there are also other properties defined, e.g.196 # `{"$ref": "...", "description": "my description"}`197 #198 # so we unravel the ref199 # `{"type": "string", "description": "my description"}`200 ref = json_schema.get("$ref")201 if ref and has_more_than_n_keys(json_schema, 1):202 assert isinstance(ref, str), f"Received non-string $ref - {ref}"203 204 resolved = resolve_ref(root=root, ref=ref)205 if not isinstance(resolved, dict):206 raise ValueError(207 f"Expected `$ref: {ref}` to resolved to a dictionary but got {resolved}"208 )209 210 # properties from the json schema take priority over the ones on the `$ref`211 json_schema.update({**resolved, **json_schema})212 json_schema.pop("$ref")213 # Since the schema expanded from `$ref` might not have `additionalProperties: false` applied,214 # we call `_ensure_strict_json_schema` again to fix the inlined schema and ensure it's valid.215 return ensure_strict_json_schema(json_schema, path=path, root=root)216 217 return json_schema218 219 220def resolve_ref(*, root: dict[str, object], ref: str) -> object:221 if not ref.startswith("#/"):222 raise ValueError(f"Unexpected $ref format {ref!r}; Does not start with #/")223 224 path = ref[2:].split("/")225 resolved = root226 for key in path:227 value = resolved[key]228 assert isinstance(229 value, dict230 ), f"encountered non-dictionary entry while resolving {ref} - {resolved}"231 resolved = value232 233 return resolved234 235 236# Flattens a JSON schema by inlining all $ref references and removing $defs/definitions237def flatten_json_schema(schema: dict) -> dict:238 root_schema = deepcopy(schema)239 240 def _flatten(node: Any) -> Any:241 if isinstance(node, dict):242 # If node is a pure $ref (or combined with extra fields), inline it243 if "$ref" in node:244 ref_value = node["$ref"]245 assert isinstance(246 ref_value, str247 ), f"Received non-string $ref - {ref_value}"248 resolved = resolve_ref(root=root_schema, ref=ref_value)249 assert isinstance(250 resolved, dict251 ), f"Expected `$ref: {ref_value}` to resolve to a dictionary but got {type(resolved)}"252 # Merge: referenced first, then overlay current (excluding $ref)253 merged: dict[str, Any] = deepcopy(resolved)254 for key, value in node.items():255 if key == "$ref":256 continue257 merged[key] = value258 return _flatten(merged)259 260 flattened: dict[str, Any] = {}261 for key, value in node.items():262 # Drop defs/definitions in output263 if key in ("$defs", "definitions"):264 continue265 if key == "properties" and isinstance(value, dict):266 flattened[key] = {267 prop_key: _flatten(prop_val)268 for prop_key, prop_val in value.items()269 }270 elif key in ("items", "contains", "additionalProperties", "not"):271 if isinstance(value, dict):272 flattened[key] = _flatten(value)273 elif isinstance(value, list):274 flattened[key] = [_flatten(v) for v in value]275 else:276 flattened[key] = value277 elif key in ("allOf", "anyOf", "oneOf", "prefixItems") and isinstance(278 value, list279 ):280 flattened[key] = [_flatten(v) for v in value]281 else:282 flattened[key] = (283 _flatten(value) if isinstance(value, (dict, list)) else value284 )285 return flattened286 if isinstance(node, list):287 return [_flatten(v) for v in node]288 return node289 290 result = _flatten(schema)291 # Ensure top-level cleanup just in case292 if isinstance(result, dict):293 result.pop("$defs", None)294 result.pop("definitions", None)295 return result296 297 298def ensure_array_schemas_have_items(schema: dict) -> dict[str, Any]:299 """300 Recursively ensure every JSON schema node with type="array" has an "items" key.301 Codex Responses API requires array schemas to specify items. Mutates a deep copy.302 """303 result = deepcopy(schema)304 305 def _is_array_schema_type(type_value: Any) -> bool:306 if type_value == "array":307 return True308 if isinstance(type_value, list):309 return "array" in type_value310 return False311 312 def _ensure(node: Any) -> Any:313 if isinstance(node, dict):314 if _is_array_schema_type(node.get("type")) and "items" not in node:315 node["items"] = {"type": "string"}316 for key, value in list(node.items()):317 node[key] = _ensure(value)318 elif isinstance(node, list):319 for idx, value in enumerate(node):320 node[idx] = _ensure(value)321 return node322 323 return _ensure(result)324 325 326def remove_titles_from_schema(schema: dict) -> dict[str, Any]:327 328 def _strip_titles(node: Any) -> Any:329 if isinstance(node, dict):330 rebuilt: dict[str, Any] = {}331 for key, value in node.items():332 # Preserve properties named "title" under the JSON Schema "properties" mapping333 if key == "properties" and isinstance(value, dict):334 rebuilt[key] = {335 prop_name: _strip_titles(prop_schema)336 for prop_name, prop_schema in value.items()337 }338 continue339 340 # Remove schema metadata field "title" elsewhere341 if key == "title":342 continue343 344 rebuilt[key] = _strip_titles(value)345 return rebuilt346 if isinstance(node, list):347 return [_strip_titles(item) for item in node]348 return node349 350 return _strip_titles(deepcopy(schema))351 352 353# ? Not used354def generate_constraint_sentences(schema: dict) -> str:355 """356 Generate human-readable constraint sentences from a JSON schema.357 358 Args:359 schema: JSON schema dictionary360 361 Returns:362 String containing constraint sentences separated by newlines363 """364 constraints = []365 366 def extract_constraints_recursive(obj, prefix=""):367 if isinstance(obj, dict):368 if "properties" in obj:369 properties = obj["properties"]370 for prop_name, prop_def in properties.items():371 current_path = f"{prefix}.{prop_name}" if prefix else prop_name372 373 if isinstance(prop_def, dict):374 prop_type = prop_def.get("type")375 376 # Handle string constraints377 if prop_type == "string":378 min_length = prop_def.get("minLength")379 max_length = prop_def.get("maxLength")380 381 if min_length is not None and max_length is not None:382 constraints.append(383 f" - {current_path} should be less than {max_length} characters and greater than {min_length} characters"384 )385 elif max_length is not None:386 constraints.append(387 f" - {current_path} should be less than {max_length} characters"388 )389 elif min_length is not None:390 constraints.append(391 f" - {current_path} should be greater than {min_length} characters"392 )393 394 # Handle array constraints395 elif prop_type == "array":396 min_items = prop_def.get("minItems")397 max_items = prop_def.get("maxItems")398 399 if min_items is not None and max_items is not None:400 constraints.append(401 f" - {current_path} should have more than {min_items} items and less than {max_items} items"402 )403 elif max_items is not None:404 constraints.append(405 f" - {current_path} should have less than {max_items} items"406 )407 elif min_items is not None:408 constraints.append(409 f" - {current_path} should have more than {min_items} items"410 )411 412 # Recurse into nested objects413 if prop_type == "object" or "properties" in prop_def:414 extract_constraints_recursive(prop_def, current_path)415 416 # Handle array items if they have properties417 if prop_type == "array" and "items" in prop_def:418 items_def = prop_def["items"]419 if isinstance(items_def, dict) and (420 "properties" in items_def421 or items_def.get("type") == "object"422 ):423 extract_constraints_recursive(424 items_def, f"{current_path}[*]"425 )426 427 # Also recurse into other nested structures428 for key, value in obj.items():429 if key not in [430 "properties",431 "type",432 "minLength",433 "maxLength",434 "minItems",435 "maxItems",436 ] and isinstance(value, dict):437 extract_constraints_recursive(value, prefix)438 439 # Start extraction from the root schema440 extract_constraints_recursive(schema)441 442 return "\n".join(constraints)443