| """Crown-hybrid-8: 24-char prompt digests, blueprints in weights, floor provenance tag. |
| |
| One allow-listed pool call. Known code digests append a weights-side blueprint; other |
| code digests pick a routed model; unrecognized prompts use the default. Non-MCQ floors |
| get a numeric trailing tag so extract_number cannot false-DQ as laundered. |
| """ |
|
|
| 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-hybrid-8" |
| _GEN = {"max_tokens": 16384, "reasoning": {"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-hybrid-8 weights must be UTF-8 JSON") from exc |
| if not isinstance(data, dict) or data.get("fmt") != _FMT: |
| raise ValueError("crown-hybrid-8 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-hybrid-8 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-hybrid-8 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-hybrid-8 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) |
| text = original |
| if not _is_code(original): |
| if not _is_mcq(original): |
| text = _floor_tag(original) |
| return call_model( |
| _POOL[default], |
| [{"role": "user", "content": text}], |
| {"max_tokens": _GEN["max_tokens"], |
| "reasoning": dict(_GEN["reasoning"])}, |
| ) |
| k = _key(original) |
| hit = notes.get(k) |
| if hit is not None: |
| model_index, blueprint = hit |
| text = original + "\n\n" + blueprint |
| else: |
| model_index = routes.get(k, default) |
| return call_model( |
| _POOL[model_index], |
| [{"role": "user", "content": text}], |
| {"max_tokens": _GEN["max_tokens"], |
| "reasoning": dict(_GEN["reasoning"])}, |
| ) |
|
|
| return agent |
|
|
