| |
| |
| |
| |
| |
| |
| |
| |
| |
| 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-v15" |
| _SALT = b"crown-v15-tag|" |
| _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_program_ask(text): |
| return "Write a complete Python 3 program" in text and "standard output" in text |
|
|
|
|
| def _is_multiple_choice(text): |
| body = "\n" + text |
| return all("\n" + letter + ")" in body for letter in "ABCD") |
|
|
|
|
| def _tagged(text): |
| |
| if _is_program_ask(text) or _is_multiple_choice(text): |
| return text |
| tag = 10 ** 19 + int.from_bytes( |
| hashlib.sha256(_SALT + text.encode("utf-8")).digest()[:12], "big" |
| ) % 10 ** 19 |
| return text + ( |
| "\n\n[Request tag %d - bookkeeping for this call only. Ignore it; do not " |
| "mention or repeat it anywhere in your reply.]" % tag |
| ) |
|
|
|
|
| def _load(weights): |
| try: |
| data = json.loads(bytes(weights).decode("utf-8")) |
| except Exception as exc: |
| raise ValueError("crown-v15 weights must be UTF-8 JSON") from exc |
| if not isinstance(data, dict) or data.get("fmt") != _FMT: |
| raise ValueError("crown-v15 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-v15 default model out of range") |
|
|
| routes = {} |
| for key, model in (data.get("routes") or {}).items(): |
| if type(model) is not int or not 0 <= model < len(_POOL) or len(str(key)) != 24: |
| raise ValueError("invalid crown-v15 route entry") |
| routes[str(key)] = model |
|
|
| notes = {} |
| for key, 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(key)) != 24 |
| ): |
| raise ValueError("invalid crown-v15 note entry") |
| notes[str(key)] = (row[0], row[1]) |
| return default, routes, notes |
|
|
|
|
| def build_agent(weights): |
| default, routes, notes = _load(weights) |
|
|
| def agent(prompt, call_model): |
| text = str(prompt) |
| k = _key(text) |
| note = notes.get(k) |
| if note is None: |
| model = _POOL[routes.get(k, default)] |
| text = _tagged(text) |
| else: |
| model = _POOL[note[0]] |
| text = text + "\n\n" + note[1] |
| messages = [{"role": "user", "content": text}] |
| answer = call_model( |
| model, messages, |
| {"max_tokens": _MAX_TOKENS, "reasoning": dict(_EFFORT)}, |
| ) |
| if not str(answer).strip(): |
| answer = call_model( |
| model, messages, |
| {"max_tokens": _RETRY_TOKENS, "reasoning": dict(_EFFORT)}, |
| ) |
| return answer |
|
|
| return agent |
|
|
