"""长上下文与记忆：标准库算例、固定探测方案与保存结果的重新计分。""" import argparse from collections import defaultdict import hashlib import json from pathlib import Path DATA = Path(__file__).resolve().parent / "data" MODEL = "Qwen/Qwen2.5-0.5B-Instruct" REVISION = "7ae557604adf67be50417f59c2c2f167def9a775" TARGETS = [("asset-cedar", "482731"), ("asset-lilac", "730284"), ("asset-maple", "195608")] SYSTEM = ("Use only the supplied inventory. Return exactly the six-digit code " "for the requested asset, with no explanation. " "If the inventory does not provide its code, return UNKNOWN.") def digest(path): result = hashlib.sha256() with Path(path).open("rb") as stream: for block in iter(lambda: stream.read(1024 * 1024), b""): result.update(block) return result.hexdigest() def fingerprint(value): raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() return hashlib.sha256(raw).hexdigest() def save_new(path, value): path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) with path.open("x") as stream: json.dump(value, stream, ensure_ascii=False, indent=2) stream.write("\n") def fillers(count): return [{"asset": f"asset-{i:04d}", "code": f"{200000 + i * 7919:06d}"} for i in range(count)] def select(records, asset): """先按明确的键查找，再把查得的记录交给语言模型。""" return [record for record in records if record["asset"] == asset] def make_plan(): cases = [] for asset, code in TARGETS: target = {"asset": asset, "code": code} for count in (16, 64): for position, index in (("first", 0), ("middle", count // 2), ("last", count - 1)): records = fillers(count - 1) records.insert(index, target) cases.append(dict(id=f"{asset}/{count}/{position}", mode="full", asset=asset, original_code=code, records=records, count=count, position=position, target_index=index, expected=code, fact_available=True)) originals = fillers(63) originals.insert(32, target) cases.extend([ dict(id=f"{asset}/selected", mode="selected", asset=asset, original_code=code, records=select(originals, asset), original_records=originals, expected=code, fact_available=True), dict(id=f"{asset}/summary", mode="summary", asset=asset, original_code=code, records=[], original_records=originals, summary=f"The inventory includes {asset} among other assets. " "Individual codes are omitted.", expected="UNKNOWN", fact_available=False), dict(id=f"{asset}/absent", mode="absent", asset=asset, original_code=code, records=fillers(64), expected="UNKNOWN", fact_available=False), ]) return dict( schema=1, model=MODEL, revision=REVISION, purpose="Small position/selection/lossy-summary probe; not a 32K benchmark.", generation=dict(max_new_tokens=16, do_sample=False, input_plus_output_cap=2048, attention="eager", device="cpu", dtype="float32", threads=4, seed=2026), scoring="Exact stripped text; recovery and contract compliance are separate.", cases=cases, ) def messages(case): inventory = case.get("summary", "\n".join( f"{row['asset']}: {row['code']}" for row in case["records"])) return [ {"role": "system", "content": SYSTEM}, {"role": "user", "content": f"Inventory:\n{inventory}\n\n" f"Requested asset: {case['asset']}"}, ] def grade(case, text): answer = text.strip() return dict(contract_correct=answer == case["expected"], code_recovered=(case["fact_available"] and answer == case["original_code"]), abstained=answer == "UNKNOWN") def kv_bytes(tokens, layers=24, kv_heads=2, head_dim=64, bytes_per_value=4): """单请求、每层相同且 K/V 同维的常规缓存；不含权重和其他内存。""" return 2 * layers * kv_heads * head_dim * tokens * bytes_per_value def memory_inputs(): # 事件、人物代号和“确认/引用”标签均为人工编写，未调用模型抽取。 rows = [ ("e01", "lin", "pickup", "Tuesday", "2026-10-01", "2026-10-01", "confirmed"), ("e02", "lin", "material", "PLA", "2026-10-02", "2026-10-02", "confirmed"), ("e03", "lin", "pickup", "Friday", "2026-10-10", "2026-10-10", "confirmed"), ("e04", "mei", "pickup", "Monday", "2026-10-11", "2026-10-11", "confirmed"), ("e05", "lin", "pickup", "Sunday", "2026-11-01", "2026-10-12", "confirmed"), ("e06", "lin", "pickup", "Wednesday", "2026-10-13", "2026-10-13", "quoted"), ("e07", "lin", "pickup", "Thursday", "2026-10-05", "2026-10-14", "confirmed"), ] events = [dict(zip(("id", "owner", "slot", "value", "valid_from", "observed_at", "kind"), row)) for row in rows] queries = [ ("now", "lin", "lin", "pickup", "2026-10-14", "2026-10-14", "ok", "Friday", "e03"), ("past", "lin", "lin", "pickup", "2026-10-07", "2026-10-14", "ok", "Thursday", "e07"), ("future", "lin", "lin", "pickup", "2026-11-02", "2026-10-14", "ok", "Sunday", "e05"), ("other", "mei", "mei", "pickup", "2026-10-14", "2026-10-14", "ok", "Monday", "e04"), ("missing", "lin", "lin", "color", "2026-10-14", "2026-10-14", "unknown", None, None), ("denied", "lin", "mei", "pickup", "2026-10-14", "2026-10-14", "denied", None, None), ("known_then", "lin", "lin", "pickup", "2026-10-07", "2026-10-07", "ok", "Tuesday", "e01"), ("quote", "lin", "lin", "pickup", "2026-10-13", "2026-10-13", "ok", "Friday", "e03"), ] queries = [dict(zip(("id", "requester", "owner", "slot", "as_of", "known_at", "expected_status", "expected_value", "expected_source"), row)) for row in queries] return events, queries def recall(events, query, policy="temporal"): if policy not in {"temporal", "arrival"}: raise ValueError("未知读取规则") if query["requester"] != query["owner"]: return dict(status="denied", value=None, source=None) candidates = [e for e in events if e["owner"] == query["owner"] and e["slot"] == query["slot"] and e["observed_at"] <= query["known_at"]] if policy == "temporal": candidates = [e for e in candidates if e["kind"] == "confirmed" and e["valid_from"] <= query["as_of"]] key = lambda e: (e["valid_from"], e["observed_at"], e["id"]) else: key = lambda e: (e["observed_at"], e["id"]) if not candidates: return dict(status="unknown", value=None, source=None) event = max(candidates, key=key) return dict(status="ok", value=event["value"], source=event["id"]) def memory_report(): events, queries = memory_inputs() results = [] for query in queries: expected = dict(status=query["expected_status"], value=query["expected_value"], source=query["expected_source"]) results.append(dict(query=query, expected=expected, arrival=recall(events, query, "arrival"), temporal=recall(events, query, "temporal"))) return dict(events=events, rows=results) def summarize(record): plan = make_plan() if record["plan_sha256"] != fingerprint(plan) or not record["complete"]: raise ValueError("记录的方案不同或尚未完成") cases = {case["id"]: case for case in plan["cases"]} if len(record["rows"]) != len(cases) or {r["id"] for r in record["rows"]} != set(cases): raise ValueError("实验记录缺失或重复") groups = defaultdict(list) for row in record["rows"]: case = cases[row["id"]] if row["messages"] != messages(case): raise ValueError("实际提示与预先固定的方案不一致") group = (f"full/{case['count']}/{case['position']}" if case["mode"] == "full" else case["mode"]) groups[group].append((row, grade(case, row["text"]))) return groups def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--record", type=Path, default=DATA / "context-memory-model.json") parser.add_argument("--plan-out", type=Path) args = parser.parse_args() if args.plan_out: save_new(args.plan_out, make_plan()) print(f"已保存固定的 27 次探测方案：{args.plan_out}") return print(f"常规 FP32 KV：1024 词元 {kv_bytes(1024) / 2**20:.0f} MiB；" f"32768 词元 {kv_bytes(32768) / 2**20:.0f} MiB（仅理论缓存）。") report = memory_report() for policy, label in (("arrival", "按到达时间取最后一条"), ("temporal", "按时间与确认状态读取")): correct = sum(row[policy] == row["expected"] for row in report["rows"]) print(f"{label}：{correct}/{len(report['rows'])}。") groups = summarize(json.loads(args.record.read_text())) for name in ("full/16/first", "full/16/middle", "full/16/last", "full/64/first", "full/64/middle", "full/64/last", "selected", "summary", "absent"): rows = groups[name] recovered = sum(score["code_recovered"] for _, score in rows) correct = sum(score["contract_correct"] for _, score in rows) counts = [row["input_tokens"] for row, _ in rows] print(f"{name}：找回代码 {recovered}/{len(rows)}，" f"遵守回答约定 {correct}/{len(rows)}，输入 {min(counts)}～{max(counts)} 词元。") print("模型结果来自保存的真实运行；默认命令只重新计分，不加载模型。") if __name__ == "__main__": main()