在 RoPE 中，一对坐标随位置旋转。用高中三角函数的语言说，一组旋转的角度可以写成 $p\omega$：$p$ 是位置，$\omega$ 是这一组的角频率。不同坐标组使用不同频率，像几根转得快慢不同的指针。

假设原来希望覆盖长度 $L$，现在想扩展到 $L'=sL$，其中 $s>1$。一种简单的位置插值思路是：

$$p'=p/s$$

例如放大四倍时，把位置 12000 映到 3000。范围压回来了，但相邻位置的间隔也从 1 变成了 $1/4$。所以，问题不仅是“远处能不能表示”，还包括“附近的两个位置会不会变得难以区分”。

YaRN 并非只改一个最大长度数字。Peng、Quesnelle、Fan 与 Shippole 的方法按频率所对应的波长，区别处理不同坐标组，并结合注意力缩放。本页核对的是 2026-02-06 的 v3：第 3 节说明机制，第 4.1 节使用 Llama 2 7B/13B、PG19 长片段进行扩展训练；128K 设置还接续了 64K 设置的检查点。这里没有重跑它的训练。上面的 $p/s$ 只解释最简单的插值直觉，不是完整 YaRN 实现。方法与训练设置

评价时仍要同时检查长输入与短输入。位置规则改变后，某个长文任务改善，不能直接证明原来的短请求表现也保持不变。

"""长上下文与记忆：标准库算例、固定探测方案与保存结果的重新计分。"""

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()

"""显式 --run 才运行长上下文与记忆专题的小模型探测；CPU，无静默截断。"""

import argparse
from datetime import datetime, timezone
from importlib.metadata import version
import json
from pathlib import Path
import platform
import time

import context_memory as lab


def experiment(directory):
    directory.mkdir(parents=True, exist_ok=False)
    plan = lab.make_plan()
    lab.save_new(directory / "plan.json", plan)  # 在下载、加载、生成之前固定。
    import torch
    from huggingface_hub import snapshot_download
    from transformers import AutoModelForCausalLM, AutoTokenizer
    settings = plan["generation"]
    torch.set_num_threads(settings["threads"])
    torch.manual_seed(settings["seed"])
    snapshot = Path(snapshot_download(
        plan["model"], revision=plan["revision"],
        allow_patterns=["*.json", "*.safetensors", "merges.txt", "README.md", "LICENSE"]))
    tokenizer = AutoTokenizer.from_pretrained(snapshot, trust_remote_code=False)
    encoded = []
    for case in plan["cases"]:
        conversation = lab.messages(case)
        rendered = tokenizer.apply_chat_template(
            conversation, tokenize=False, add_generation_prompt=True)
        inputs = tokenizer.apply_chat_template(
            conversation, tokenize=True, add_generation_prompt=True,
            return_dict=True, return_tensors="pt")
        length = inputs["input_ids"].shape[1]
        if length + settings["max_new_tokens"] > settings["input_plus_output_cap"]:
            raise ValueError(f"{case['id']} 超出预先固定的预算，停止而不截断")
        encoded.append((case, conversation, rendered, inputs))
    model = AutoModelForCausalLM.from_pretrained(
        snapshot, dtype=torch.float32, attn_implementation=settings["attention"],
        trust_remote_code=False).eval()
    report = dict(
        complete=False, started_at=datetime.now(timezone.utc).isoformat(),
        model=plan["model"], revision=plan["revision"], generation=settings,
        plan_sha256=lab.fingerprint(plan),
        source_sha256={path.name: lab.digest(path) for path in
                       (Path(lab.__file__), Path(__file__))},
        model_files_sha256={path.name: lab.digest(path) for path in snapshot.iterdir()
                            if path.is_file() and (path.suffix in {".json", ".safetensors"}
                                                   or path.name == "merges.txt")},
        environment=dict(python=platform.python_version(), platform=platform.platform(),
                         packages={name: version(name) for name in
                                   ("torch", "transformers", "huggingface-hub",
                                    "tokenizers", "safetensors")}),
        rows=[],
    )
    target = directory / "model.json"

    def checkpoint():
        temporary = directory / "model.pending.json"
        temporary.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
        temporary.replace(target)

    checkpoint()
    eos = model.generation_config.eos_token_id
    eos = [eos] if isinstance(eos, int) else eos
    for case, conversation, rendered, inputs in encoded:
        length = inputs["input_ids"].shape[1]
        started = time.perf_counter()
        with torch.inference_mode():
            output = model.generate(
                **inputs, max_new_tokens=settings["max_new_tokens"],
                do_sample=False, temperature=None, top_p=None, top_k=None,
                pad_token_id=tokenizer.eos_token_id)
        seconds = time.perf_counter() - started
        ids = output[0, length:].tolist()
        text = tokenizer.decode(ids, skip_special_tokens=True)
        row = dict(id=case["id"], messages=conversation, rendered_prompt=rendered,
                   input_ids=inputs["input_ids"][0].tolist(), input_tokens=length,
                   output_ids=ids, output_tokens=len(ids), text=text,
                   stopped_on_eos=bool(ids and ids[-1] in eos), seconds=seconds,
                   **lab.grade(case, text))
        report["rows"].append(row)
        checkpoint()
        print(f"{case['id']}: {text!r}, 输入 {length}, "
              f"约定正确 {row['contract_correct']}", flush=True)
    report["complete"] = True
    report["finished_at"] = datetime.now(timezone.utc).isoformat()
    checkpoint()
    lab.save_new(directory / "memory.json", lab.memory_report())
    print(f"已完成全部 {len(report['rows'])} 次探测：{target}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--run", action="store_true")
    parser.add_argument("--out-dir", type=Path)
    args = parser.parse_args()
    if not args.run:
        print("显式传入 --run --out-dir 新目录，才下载并运行固定的 27 次模型探测。")
        return
    if args.out_dir is None:
        parser.error("请用 --out-dir 指定尚不存在的目录")
    experiment(args.out_dir)


if __name__ == "__main__":
    main()

1. 3700+512=4212，超过 4096；保持输出上限不变，至少减少 116 个输入词元。减少后仍要重新套模板并计数。
2. 384 MiB，只有该假设下的 KV。权重、临时结果和框架等内存另算。
3. 代码已从摘要中删除，回答 UNKNOWN 符合约定，却没有满足查回原始代码的需求。这是两个不同指标。
4. 不能。这里只有三个固定目标和一种模板；记录数量变化也改变了整体输入。需要更充分的配对实验与任务覆盖。
5. e07 在 10/14 才被收到，却描述 10/5 起的状态。回看时可以使用它；站在 10/7 的知识范围内则还不知道它。
6. 时间规则下，past 从星期四变成星期二，其他问题不变。到达顺序基线的 now、past、future 会从 e07 的星期四变成 e06 的星期三，仍有错误。说明只删一个迟到事件不能解决引用识别问题。
7. 不能保证。若原句还保留，可以重新核查；若只剩错误的摘要，读取阶段可能再也看不到丢掉的限定条件。
8. 例如一处记录设备代号与负责人，另一处记录该负责人的取件安排。分别统计两条证据的保留和检索情况，以及证据齐全时的回答正确率；不要把最终错误一律归为检索失败。
9. 前者表示规则不允许访问，后者表示允许查询但没有可用事实。例子假定身份已经由可信调用方给定，没有实现登录、身份认证或完整授权系统。