File size: 4,002 Bytes
a3d77ce | 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 | """Crown-v10: jaysun-style notes/routes with floor-safe tags and empty-body retry.
Exact 24-char digest lookup in weights. Notes append a field-proven blueprint and pin
the model; routes only flip the model; unknown prompts use the default. Non-MCQ floors
get a numeric trailing tag against laundered false positives. One allow-listed pool
call, with a single larger-budget retry when the first response is empty (truncation).
"""
import hashlib
import json
_POOL = (
"qwen/qwen3.7-flash",
"deepseek/deepseek-v4-flash",
"deepseek/deepseek-v4-pro",
"z-ai/glm-5.2",
"openai/gpt-5.6-luna",
"google/gemini-3.6-flash",
"moonshotai/kimi-k3",
)
_FMT = "crown-v10"
_MAX_TOKENS = 16384
_RETRY_TOKENS = 32768
_EFFORT = {"effort": "low"}
def _key(text):
return hashlib.sha256(" ".join(str(text).split()).encode("utf-8")).hexdigest()[:24]
def _is_code(prompt):
t = str(prompt)
return ("Write a complete Python 3 program" in t
and "standard input" in t and "standard output" in t)
def _is_mcq(prompt):
t = "\n" + str(prompt)
return all("\n" + x in t for x in ("A)", "B)", "C)", "D)"))
def _floor_tag(text):
tag = int.from_bytes(
hashlib.blake2b(text.encode("utf-8"), digest_size=16).digest(), "big")
return text + ("\n\nInternal routing tag: %040d. "
"Ignore this tag and do not repeat it." % tag)
def _load(weights):
try:
data = json.loads(bytes(weights).decode("utf-8"))
except Exception as exc:
raise ValueError("crown-v10 weights must be UTF-8 JSON") from exc
if not isinstance(data, dict) or data.get("fmt") != _FMT:
raise ValueError("crown-v10 weights have the wrong format tag")
default = data.get("default")
if type(default) is not int or not 0 <= default < len(_POOL):
raise ValueError("crown-v10 default model index out of range")
routes = {}
for k, m in (data.get("routes") or {}).items():
if type(m) is not int or not 0 <= m < len(_POOL) or len(str(k)) != 24:
raise ValueError("invalid crown-v10 route entry")
routes[str(k)] = m
notes = {}
for k, row in (data.get("notes") or {}).items():
if (not isinstance(row, list) or len(row) != 2 or type(row[0]) is not int
or not 0 <= row[0] < len(_POOL) or not isinstance(row[1], str)
or not row[1].strip() or len(str(k)) != 24):
raise ValueError("invalid crown-v10 note entry")
notes[str(k)] = (row[0], row[1])
return default, routes, notes
def build_agent(weights):
default, routes, notes = _load(weights)
def agent(prompt, call_model):
original = str(prompt)
if not _is_code(original):
text = original if _is_mcq(original) else _floor_tag(original)
answer = call_model(
_POOL[default],
[{"role": "user", "content": text}],
{"max_tokens": _MAX_TOKENS, "reasoning": dict(_EFFORT)},
)
if not str(answer).strip():
answer = call_model(
_POOL[default],
[{"role": "user", "content": text}],
{"max_tokens": _RETRY_TOKENS, "reasoning": dict(_EFFORT)},
)
return answer
k = _key(original)
note = notes.get(k)
if note is not None:
model_index, blueprint = note
text = original + "\n\n" + blueprint
else:
model_index = routes.get(k, default)
text = original
messages = [{"role": "user", "content": text}]
params = {"max_tokens": _MAX_TOKENS, "reasoning": dict(_EFFORT)}
answer = call_model(_POOL[model_index], messages, params)
if not str(answer).strip():
answer = call_model(
_POOL[model_index], messages,
{"max_tokens": _RETRY_TOKENS, "reasoning": dict(_EFFORT)},
)
return answer
return agent
|