"""Ledger needle bench: fact recall across a conversation far longer than the context.
The long-context analogue for this architecture. A needle-in-a-haystack test measures whether a
fact survives distance inside a huge prompt; here the prompt never exceeds the training window
(last ``keep`` messages) and facts survive only through the ledger of the model's own notes, so
the same question — does distance kill recall? — is asked of the memory mechanism instead of the
attention span.
``run`` drives a deterministic scripted conversation through the engine on the GPU box and
records everything per turn. ``render`` turns that record into a self-contained HTML viewer:
timeline, per-turn notes, the exact ledger state, and every probe scored.
"""
from __future__ import annotations
import argparse
import html as html_lib
import json
import random
import re
import time
from pathlib import Path
FACT_POOL = [
('sister flight', 'flight {code}', 'my sister lands on {value}, write that down'),
('hotel room', 'room {num3}', 'we got {value} at the hotel'),
('locker code', 'code {num4}', 'the gym locker is {value}'),
('dentist slot', '{clock}', 'dentist moved me to {value}'),
('plumber quote', '${money}', 'the plumber quoted {value} for the bathroom'),
('car plate', '{plate}', 'the rental has plate {value}'),
('wifi password', '{word}{num3}', 'cabin wifi password is {value}'),
('train platform', 'platform {num2}', 'our train leaves from {value}'),
('order number', 'order {code}', 'the couch is {value}, keep it handy'),
('rent due', 'day {num2}', 'landlord wants rent by {value} each month'),
('kid teacher', 'ms. {name}', "tomas' new teacher is {value}"),
('parking spot', 'spot {letter}{num2}', 'we always park at {value}'),
('meds dose', '{num2}mg', 'doctor changed the dose to {value}'),
('deposit', '${money}', 'the deposit for the venue was {value}'),
('boarding gate', 'gate {letter}{num2}', 'boarding is at {value}'),
]
DISTRACTORS = [
'what a week, honestly', 'did you catch the game last night?', 'i love this weather lately',
'work was chaos today', 'thinking of making pasta tonight', 'my back hurts from the gym',
'the neighbor is renovating again, so loud', 'saw a great documentary yesterday',
'coffee here is getting expensive', 'might go for a walk later', 'the cat knocked over a plant',
'traffic was unreal this morning', 'finally finished that book', 'craving something sweet',
'they repaved our street', 'my phone battery dies so fast now',
]
PROBES = ['wait, what was the {key} again?', 'remind me of the {key}?',
'i forgot the {key}, what was it?', 'quick — the {key}?']
WORDS = ('maple', 'harbor', 'cactus', 'violet', 'ember', 'quartz')
NAMES = ('ferro', 'silva', 'duarte', 'campos', 'rojas', 'ibanez')
def _value(template: str, rng: random.Random) -> str:
return (template
.replace('{code}', f'{rng.choice("ABKQZ")}{rng.choice("RLMT")}{rng.randint(1000, 9999)}')
.replace('{num2}', str(rng.randint(10, 99)))
.replace('{num3}', str(rng.randint(100, 999)))
.replace('{num4}', str(rng.randint(1000, 9999)))
.replace('{clock}', f'{rng.randint(1, 12)}:{rng.choice(("15", "30", "45"))}pm')
.replace('{money}', f'{rng.randint(2, 90) * 100:,}')
.replace('{plate}', f'{"".join(rng.choice("BCDFGHJK") for _ in range(3))}-{rng.randint(100, 999)}')
.replace('{word}', rng.choice(WORDS))
.replace('{letter}', rng.choice('ABCDE'))
.replace('{name}', rng.choice(NAMES)))
def build_script(turns: int, seed: int) -> list[dict]:
"""Deterministic conversation plan: facts early and throughout, probes at all distances."""
rng = random.Random(seed)
facts = []
for key, template, phrasing in rng.sample(FACT_POOL, len(FACT_POOL)):
facts.append({'key': key, 'value': _value(template, rng), 'phrasing': phrasing})
plan: list[dict] = []
stated: dict[str, dict] = {}
fact_iter = iter(facts)
for index in range(turns):
remaining = turns - index
can_probe = [f for f in stated.values() if index - f['turn'] >= 3]
if index >= turns - 3 and can_probe:
kind = 'probe'
elif index % 7 in (0, 3) and (fact := next(fact_iter, None)) is not None:
plan.append({'kind': 'fact', **fact})
stated[fact['key']] = {**fact, 'turn': index}
continue
elif index % 11 == 5 and can_probe:
kind = 'probe'
elif index % 13 == 8 and stated and remaining > 5:
kind = 'correction'
else:
kind = 'distractor'
if kind == 'probe':
target = rng.choice(can_probe)
plan.append({'kind': 'probe', 'key': target['key'], 'value': target['value'],
'distance': index - target['turn'],
'text': rng.choice(PROBES).format(key=target['key'])})
elif kind == 'correction':
target = rng.choice(list(stated.values()))
new = _value(next(t for k, t, _ in FACT_POOL if k == target['key']), rng)
plan.append({'kind': 'correction', 'key': target['key'], 'old': target['value'],
'value': new,
'text': f'actually scratch that, the {target["key"]} is {new} now'})
stated[target['key']] = {**target, 'value': new, 'turn': index}
else:
plan.append({'kind': 'distractor', 'text': rng.choice(DISTRACTORS)})
return plan
def _hit(value: str, answer: str) -> bool:
"""Whether the answer states the value, matched on its distinctive core.
Demanding the full phrase penalised correct answers on both systems ("the locker code is
4805" failed against 'code 4805'), so the core — the last token, which carries the
identifier — is what must appear. Values are generated with 2+ digit cores, so incidental
collisions stay unlikely.
"""
canon = lambda t: re.sub(r'[^a-z0-9]', '', t.lower()) # noqa: E731
core = value.split()[-1]
return canon(core) in canon(answer)
def run(args: argparse.Namespace) -> None:
from diffusion_lm.claims import SYSTEM, chat_prefix, ledger_line, ledger_notes, merge_notes
from diffusion_lm.reasoning_playground import ReasoningEngine
from diffusion_lm.train import resolve_device
engine = ReasoningEngine(args.checkpoint, args.tokenizer, resolve_device('auto'))
plan = build_script(args.turns, args.seed)
messages: list[dict] = []
records = []
# Appended per turn and flushed, so a tail -f (or a crash) sees every completed turn.
sink = args.output.open('w', encoding='utf-8')
for index, turn in enumerate(plan):
user_text = turn.get('text') or turn['phrasing'].format(value=turn['value'])
messages.append({'role': 'user', 'content': user_text})
older = merge_notes(ledger_notes(messages, args.keep))
window = messages[max(0, len(messages) - args.keep):]
prefix = chat_prefix(
[{'role': m['role'], 'content': m['content']} for m in window],
system=SYSTEM, extra=ledger_line(older),
)
blocks: list[tuple[int, str]] = []
answer = ''
begin = time.perf_counter()
for _, answer, _ in engine.stream_chat(
prefix, temperature=0.7, steps_per_block=args.steps,
max_answer_tokens=120, seed=args.seed * 1000 + index, blocks_out=blocks,
):
pass
note = '; '.join(text for _, text in blocks if text)
messages.append({'role': 'assistant', 'content': answer.strip(), 'note': note})
record = {
'turn': index, 'kind': turn['kind'], 'user': user_text,
'notes': [text for _, text in blocks if text], 'ledger': older,
'prefix_tokens': len(engine.tokenizer.encode(prefix, add_special_tokens=False).ids),
'answer': answer.strip(), 'seconds': round(time.perf_counter() - begin, 2),
}
if turn['kind'] == 'probe':
record.update({'key': turn['key'], 'expected': turn['value'],
'distance': turn['distance'],
'hit': _hit(turn['value'], answer)})
if turn['kind'] in ('fact', 'correction'):
record.update({'key': turn['key'], 'value': turn['value']})
records.append(record)
sink.write(json.dumps(record, ensure_ascii=False) + '\n')
sink.flush()
flag = '' if turn['kind'] != 'probe' else (' HIT' if record['hit'] else ' MISS')
print(f'[{index + 1}/{len(plan)}] {turn["kind"]}{flag} {record["seconds"]}s '
f'ledger={len(older)}', flush=True)
print(f' U: {user_text}', flush=True)
if note:
print(f' T: {note}', flush=True)
print(f' A: {answer.strip()}', flush=True)
if turn['kind'] == 'probe':
print(f' >> esperado {turn["value"]} (distancia {turn["distance"]}) -> '
f'{"HIT" if record["hit"] else "MISS"}', flush=True)
sink.close()
probes = [r for r in records if r['kind'] == 'probe']
hits = sum(r['hit'] for r in probes)
print(f'\nrecall: {hits}/{len(probes)} ({hits / max(1, len(probes)):.0%})')
def run_ar(args: argparse.Namespace) -> None:
"""Same scripted conversation through a plain AR baseline of the same size.
``--context full`` gives the baseline the whole history (its native memory);
``--context budget`` truncates the rendered prompt to the newest messages that fit the
same token budget the ledger system runs under, so both systems answer from equally many
prefix tokens and only the memory MECHANISM differs.
"""
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(args.model_path)
model = AutoModelForCausalLM.from_pretrained(
args.model_path, torch_dtype=torch.bfloat16, device_map='cuda',
).eval()
plan = build_script(args.turns, args.seed)
# Deliberately generous: the baselines get an explicit memory instruction, greedy decoding,
# and (with --thinking) their native reasoning mode. The comparison should show the ledger
# beating the baselines at their best, or it shows nothing.
system = ('You are a meticulous assistant. Remember every concrete detail the user '
'mentions — codes, amounts, times, names, numbers. When asked to recall one, '
'answer with the exact value.')
messages: list[dict] = []
records = []
try:
tokenizer.apply_chat_template(
[{'role': 'system', 'content': 'x'}, {'role': 'user', 'content': 'y'}],
tokenize=True, add_generation_prompt=True)
args._no_system = False
except Exception:
args._no_system = True
sink = args.output.open('w', encoding='utf-8')
for index, turn in enumerate(plan):
user_text = turn.get('text') or turn['phrasing'].format(value=turn['value'])
messages.append({'role': 'user', 'content': user_text})
def _assemble(tail: list[dict]) -> list[dict]:
if getattr(args, '_no_system', False):
head = dict(tail[0]) if tail else {'role': 'user', 'content': ''}
head['content'] = f'{system}\n\n{head["content"]}'
return [head] + [dict(m) for m in tail[1:]]
return [{'role': 'system', 'content': system}] + [dict(m) for m in tail]
chat = _assemble(messages)
if args.context == 'budget':
# Newest-first packing under the same prefix budget the ledger system uses.
kept: list[dict] = []
for message in reversed(messages):
candidate = _assemble([message, *kept])
rendered = tokenizer.apply_chat_template(
candidate, tokenize=True, add_generation_prompt=True,
enable_thinking=False,
)
# BatchEncoding in newer transformers: len() counts KEYS, which silently
# disabled the cap; count tokens explicitly.
ids = rendered if isinstance(rendered, list) else rendered['input_ids']
if len(ids) > args.budget:
break
kept = [message, *kept]
chat = _assemble(kept)
encoded = tokenizer.apply_chat_template(
chat, tokenize=True, add_generation_prompt=True, return_tensors='pt',
enable_thinking=args.thinking,
)
# Newer transformers return a BatchEncoding here rather than the bare tensor.
prompt_ids = (encoded if torch.is_tensor(encoded) else encoded['input_ids']).to(
model.device)
torch.manual_seed(args.seed * 1000 + index)
begin = time.perf_counter()
with torch.no_grad():
out = model.generate(
prompt_ids, max_new_tokens=640 if args.thinking else 120, do_sample=False,
pad_token_id=tokenizer.eos_token_id or tokenizer.pad_token_id,
)
answer = tokenizer.decode(out[0][prompt_ids.shape[1]:], skip_special_tokens=True)
answer = re.sub(r'(?s)
| sistema | tamaño (B) | recall | d 3–10 | d 11–50 | d 51+ | test total (s) | s/turno | prefijo máx |
|---|