"""Small deterministic task, data builder and evaluation helper. No training on import.""" import argparse import hashlib import json import random import re import time import urllib.request from pathlib import Path SYSTEM = ( '你是到手价计算助手。所有金额单位都是分。小计=单价乘数量。' '小计达到满减门槛时减优惠金额，否则不减。' '包邮门槛也按优惠前小计判断：小计达到包邮门槛时运费为0，否则收基础运费。' '应付=max(小计-实际优惠,0)+实际运费。' '仅输出一个JSON对象，且只有payable_cents这个键，值为非负十进制整数。' '不要输出解释、代码围栏或任何其他文字。' ) FIELDS = ( "unit_price_cents", "quantity", "coupon_threshold_cents", "coupon_cents", "shipping_cents", "free_shipping_threshold_cents", ) PATTERN = re.compile(r'\s*\{\s*"payable_cents"\s*:\s*(0|[1-9][0-9]*)\s*\}\s*') def oracle(case): """Teaching rules, deliberately independent of any real platform.""" if set(case) != set(FIELDS): raise ValueError("Case fields do not match the task contract") if any(type(case[k]) is not int or case[k] < 0 for k in FIELDS): raise ValueError("All inputs must be non-negative integers") if case["quantity"] < 1: raise ValueError("Quantity must be positive") subtotal = case["unit_price_cents"] * case["quantity"] discount = case["coupon_cents"] if subtotal >= case["coupon_threshold_cents"] else 0 shipping = 0 if subtotal >= case["free_shipping_threshold_cents"] else case["shipping_cents"] return max(subtotal - discount, 0) + shipping def parse_completion(text): # The lexical gate also rejects duplicate keys, booleans, floats and exponent notation. if not isinstance(text, str) or not PATTERN.fullmatch(text): return None value = json.loads(text)["payable_cents"] return value if value <= 2**53 - 1 else None def grade(text, expected): value = parse_completion(text) valid = value is not None correct = valid and value == expected return {"format_ok": valid, "amount_ok": correct, "reward": 1.0 if correct else (0.2 if valid else 0.0)} def prompt_for(case, template=0): prefixes = ["请计算这笔订单的应付金额：", "请按约定规则处理以下订单：", "给出这笔订单最终需要支付的分数金额："] return prefixes[template] + json.dumps(case, ensure_ascii=False, separators=(",", ":")) def self_check(): case = dict(zip(FIELDS, [9900, 3, 20000, 3000, 600, 30000])) assert oracle(case) == 27300 for field in ["coupon_threshold_cents", "free_shipping_threshold_cents"]: for threshold in [29699, 29700, 29701]: probe = {**case, field: threshold} discount = 3000 if probe["coupon_threshold_cents"] <= 29700 else 0 freight = 0 if probe["free_shipping_threshold_cents"] <= 29700 else 600 assert oracle(probe) == 29700 - discount + freight assert oracle({**case, "coupon_cents": 50000}) == 600 assert grade('{"payable_cents":27300}', 27300)["reward"] == 1.0 assert grade('{"payable_cents":27000}', 27300)["reward"] == 0.2 invalid = ['27300', '{"payable_cents":true}', '{"payable_cents":27300.0}', '{"payable_cents":2.73e4}', '{"payable_cents":-1}', '{"payable_cents":27300,"extra":1}', '{"payable_cents":1,"payable_cents":27300}', '```json\n{"payable_cents":27300}\n```', '答案是27300。', '{"payable_cents":9007199254740992}'] for text in invalid: assert grade(text, 27300)["reward"] == 0.0, text for value in [True, 9.9, -1]: try: oracle({**case, "unit_price_cents": value}) except ValueError: pass else: raise AssertionError("Invalid input accepted") print("Oracle boundaries and reward counterexamples passed; no model was loaded.") def write_jsonl(path, rows): path.parent.mkdir(parents=True, exist_ok=True) with path.open("x", encoding="utf-8") as handle: for row in rows: handle.write(json.dumps(row, ensure_ascii=False) + "\n") def make_data(out, seed): if out.exists(): raise FileExistsError("Use a new output directory to preserve the frozen data") rng, seen = random.Random(seed), set() partitions = {} sizes = {"sft_train": 600, "sft_valid": 120, "rl_train": 300, "rl_valid": 80, "test": 160} for name, count in sizes.items(): rows = [] while len(rows) < count: i = len(rows) price, qty = rng.randint(100, 20000), rng.randint(1, 5) subtotal = price * qty boundary = i % 2 == 0 threshold = max(0, subtotal + [-1, 0, 1][i % 3]) if boundary else rng.randint(0, 90000) free = max(0, subtotal + [-1, 0, 1][(i // 3) % 3]) if boundary else rng.randint(0, 90000) case = dict(zip(FIELDS, [price, qty, threshold, rng.randint(0, 15000), rng.choice([0, 600, 800, 1200]), free])) signature = json.dumps(case, sort_keys=True) if signature in seen: continue seen.add(signature) template = (2 if i % 2 else 0) if name == "test" else i % 2 rows.append({"case_id": hashlib.sha256(signature.encode()).hexdigest()[:16], "case": case, "prompt": prompt_for(case, template), "system": SYSTEM, "expected": oracle(case), "bucket": "boundary" if boundary else "ordinary", "template": template}) partitions[name] = rows for name in ["train", "valid"]: sft = [{"messages": [{"role": "system", "content": SYSTEM}, {"role": "user", "content": r["prompt"]}, {"role": "assistant", "content": json.dumps({"payable_cents": r["expected"]})}]} for r in partitions["sft_" + name]] rl = [{"system": SYSTEM, "prompt": r["prompt"], "answer": str(r["expected"]), "type": "price"} for r in partitions["rl_" + name]] write_jsonl(out / "sft" / (name + ".jsonl"), sft) write_jsonl(out / "rl" / (name + ".jsonl"), rl) for name, rows in partitions.items(): write_jsonl(out / "cases" / (name + ".jsonl"), rows) hashes = {str(p.relative_to(out)): hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(out.rglob("*.jsonl"))} manifest = {"seed": seed, "counts": sizes, "sha256": hashes, "distinct_case_count": len(seen), "system": SYSTEM, "limitations": "Synthetic, partially held-out wording; not real commerce generalization"} (out / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2)) print(json.dumps(manifest, ensure_ascii=False, indent=2)) def evaluate(model_path, cases_path, out): # Third-party dependencies are needed only for real model evaluation. import mlx.core as mx from mlx_lm import generate, load from mlx_lm.sample_utils import make_sampler if out.exists(): raise FileExistsError("Evaluation output already exists") model, tokenizer = load(model_path) rows = [json.loads(line) for line in cases_path.read_text().splitlines() if line.strip()] results = [] for row in rows: messages = [{"role": "system", "content": row["system"]}, {"role": "user", "content": row["prompt"]}] prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) started = time.perf_counter() text = generate(model, tokenizer, prompt=prompt, max_tokens=64, sampler=make_sampler(temp=0.0), verbose=False) results.append({**row, "completion": text, **grade(text, row["expected"]), "seconds": time.perf_counter() - started}) write_jsonl(out, results) if not rows: raise ValueError("Empty evaluation set") summary = {"model_path": model_path, "n": len(results), "format_rate": sum(r["format_ok"] for r in results) / len(results), "joint_accuracy": sum(r["amount_ok"] for r in results) / len(results), "mean_seconds": sum(r["seconds"] for r in results) / len(results), "mlx_peak_bytes": mx.get_peak_memory(), "cases_sha256": hashlib.sha256(cases_path.read_bytes()).hexdigest(), "decode": {"temperature": 0.0, "max_tokens": 64}} summary["amount_accuracy_given_format"] = ( sum(r["amount_ok"] for r in results) / sum(r["format_ok"] for r in results) if any(r["format_ok"] for r in results) else None) for bucket in ["boundary", "ordinary"]: group = [r for r in results if r["bucket"] == bucket] summary[bucket + "_accuracy"] = sum(r["amount_ok"] for r in group) / len(group) if group else None out.with_suffix(".summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2)) print(json.dumps(summary, ensure_ascii=False, indent=2)) def api_check(base_url, model_name, out): case = dict(zip(FIELDS, [9900, 3, 20000, 3000, 600, 30000])) payload = {"model": model_name, "messages": [{"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt_for(case)}], "temperature": 0, "max_tokens": 64, "stream": False} request = urllib.request.Request(base_url.rstrip("/") + "/v1/chat/completions", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(request, timeout=120) as response: body = json.load(response) content = body["choices"][0]["message"]["content"] receipt = {"request": payload, "response": body, "expected": oracle(case), "grade": grade(content, oracle(case)), "url": request.full_url} write_jsonl(out, [receipt]) print(json.dumps(receipt, ensure_ascii=False, indent=2)) def main(): parser = argparse.ArgumentParser(description=__doc__) sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("self-check") data = sub.add_parser("data") data.add_argument("--out", type=Path, required=True) data.add_argument("--seed", type=int, default=42) ev = sub.add_parser("evaluate") ev.add_argument("--model", required=True) ev.add_argument("--cases", type=Path, required=True) ev.add_argument("--out", type=Path, required=True) api = sub.add_parser("api-check") api.add_argument("--base-url", default="http://127.0.0.1:8088") api.add_argument("--model", required=True) api.add_argument("--out", type=Path, required=True) args = parser.parse_args() if args.command == "self-check": self_check() elif args.command == "data": make_data(args.out, args.seed) elif args.command == "evaluate": evaluate(args.model, args.cases, args.out) else: api_check(args.base_url, args.model, args.out) if __name__ == "__main__": main()