Spaces:
Running
Running
File size: 7,003 Bytes
5f7bbf3 bc5dc6b 5f7bbf3 a681817 5f7bbf3 a681817 bc5dc6b 5f7bbf3 d074d4b bc5dc6b c4d5c60 bc5dc6b 5f7bbf3 d074d4b 5f7bbf3 d074d4b 5f7bbf3 d074d4b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """Public SystemOne transport compatibility; native model probabilities are unchanged."""
from copy import deepcopy
from datetime import date
import json
from pathlib import Path
from fastapi import HTTPException
from model_registry import MODEL_ORDER, PROFILES
# Descending published parameter size, with general Kai before specialist Lex at 0.6B.
ORDER = MODEL_ORDER
MODEL_RELEASES_PATH = Path(__file__).with_name('MODEL_RELEASES.json')
CONFIDENCE_DEFINITION = "max(p); native peak probability, not calibrated correctness or TypeSafe formula equivalence"
def arrange_models(registry):
return {key: registry[key] for key in ORDER if key in registry}
def resolve_model(value, registry):
if not isinstance(value, str) or not value.strip():
raise HTTPException(422, "Specify an available Decision model")
value = value.casefold()
for key, item in registry.items():
aliases = [key, item['label'], item.get('repo_id', '')]
if item.get('repo_id'):
aliases.append(item['repo_id'].rsplit('/', 1)[-1])
if value in (alias.casefold() for alias in aliases if alias):
return key
raise HTTPException(422, "This model is not available in this Studio")
def resolve_canonical_model(value, registry):
"""Resolve the public SystemOne identifier without accepting wire aliases."""
if not isinstance(value, str) or not value.strip():
raise HTTPException(422, "Specify an available canonical Decision model")
for key, item in registry.items():
if value == item['repo_id']:
return key
raise HTTPException(
422,
"Use the exact Hugging Face repository ID for an available Decision model.",
)
def _release_date(value):
if not isinstance(value, str):
raise ValueError('Release date must be an ISO calendar date')
try:
parsed = date.fromisoformat(value)
except ValueError as exc:
raise ValueError('Release date must be an ISO calendar date') from exc
if parsed.isoformat() != value:
raise ValueError('Release date must be an ISO calendar date')
return value
def load_model_releases(path=None):
"""Validate presentation metadata; never use it to admit a serving artifact."""
source = MODEL_RELEASES_PATH if path is None else Path(path)
document = json.loads(source.read_text(encoding='utf-8'))
if not isinstance(document, dict) or set(document) != {'release', 'models'}:
raise ValueError('Invalid model release metadata')
default_date = _release_date(document['release'])
entries = document['models']
if not isinstance(entries, list) or not 1 <= len(entries) <= len(ORDER):
raise ValueError('Invalid model release entries')
releases = {}
for entry in entries:
if not isinstance(entry, dict) or set(entry) - {
'label', 'repo_id', 'revision', 'manifest_sha256',
'complete_input_tokens', 'release_date',
} or {'repo_id', 'revision', 'manifest_sha256'} - set(entry):
raise ValueError('Invalid model release entry')
repo_id = entry['repo_id']
revision = entry['revision']
manifest = entry['manifest_sha256']
if (not isinstance(repo_id, str) or repo_id not in {
profile['repo_id'] for profile in PROFILES.values()
} or repo_id in releases
or not isinstance(revision, str) or len(revision) != 40
or any(c not in '0123456789abcdef' for c in revision)
or not isinstance(manifest, str) or len(manifest) != 64
or any(c not in '0123456789abcdef' for c in manifest)):
raise ValueError('Invalid model release identity')
if 'label' in entry and (not isinstance(entry['label'], str) or not entry['label']):
raise ValueError('Invalid model release label')
if 'complete_input_tokens' in entry and (
type(entry['complete_input_tokens']) is not int
or entry['complete_input_tokens'] < 1
):
raise ValueError('Invalid model release input limit')
releases[repo_id] = {
'revision': revision,
'manifest_sha256': manifest,
'release_date': _release_date(entry.get('release_date', default_date)),
}
return releases
def public_models(registry, *, releases=None):
if releases is None:
try:
releases = load_model_releases()
except (OSError, ValueError, UnicodeError, RecursionError):
# Release dates are presentation metadata, not model admission.
# Explicit load_model_releases() callers still get validation errors.
releases = {}
result = []
for key, item in registry.items():
public = dict(
item, id=item['repo_id'], wire_id=key, name=item['label'],
description=item.get('description', 'Decision foundation model'),
)
release = releases.get(item['repo_id'])
if (release is not None and item.get('revision') == release['revision']
and item['manifest_sha256'] == release['manifest_sha256']):
public['release_date'] = release['release_date']
result.append(public)
return result
def sdk_response(result, *, public_model=None):
"""Add the SDK's required distribution statistic without altering native outputs."""
result = deepcopy(result)
if public_model is not None:
result['model'] = public_model
groups = [result['answers']] if 'answers' in result else [r['answers'] for r in result['results']]
for answers in groups:
for answer in answers.values():
if answer['type'] in ('choice', 'score'):
probabilities = list(answer['probabilities'].values())
answer['confidence'] = max(probabilities)
result['profile'] = dict(result.get('profile', {}), confidence_definition=CONFIDENCE_DEFINITION)
return result
def public_response(result, *, public_model, batch):
"""Project native diagnostics onto the strict public Decision envelope."""
prepared = sdk_response(result, public_model=public_model)
def answers_only(answers):
fields = {
'noul': ('type', 'noul'),
'choice': ('type', 'choice', 'confidence', 'probabilities'),
'score': ('type', 'score', 'confidence', 'legend', 'probabilities'),
}
return {
key: {field: answer[field] for field in fields[answer['type']]}
for key, answer in answers.items()
}
if batch:
return {
'model': public_model,
'results': [
{'id': row['id'], 'answers': answers_only(row['answers']), 'usage': row['usage']}
for row in prepared['results']
],
'usage': prepared['usage'],
}
return {
'model': public_model,
'answers': answers_only(prepared['answers']),
'usage': prepared['usage'],
}
|