learning_paths.py · 可读全文
这里保留原文的中文、公式标记、代码和链接。可直接交给 Agent 读取，也可选取下方原文复制。此入口返回 HTML；原始 .md / .py 文件仍保留，完整候选包同时包含 UTF-8 原件。
"""DN2: read-only checkpoint diagnostics and untrained mechanism controls. No training."""
from pathlib import Path
import hashlib, json, platform, re
import numpy as np
import torch
import seq2seq as seq
import mechanisms as mech
import legacy_numpy as old
DOCUMENT_RECORDS=[
('09:00','林岑把备用钥匙放在蓝柜；取件人是周禾。'),
('09:10','陈默说红柜空着；这句话只说明红柜，没有改动钥匙。'),
('09:20','林岑通知：把那把备用钥匙改放红柜。'),
('09:30','周禾回复：我还没取；“那把”指备用钥匙。'),
('09:40','林岑撤回09:20的改放通知：它仍在蓝柜，取件人不变。')]
QUESTION='现在周禾应去哪个柜子取备用钥匙？'
def document(order=None):
order=list(range(5)) if order is None else list(order)
assert sorted(order)==list(range(5)), '只允许移动完整记录；时间与正文一起移动'
return '\n'.join(t+' '+s for i in order for t,s in [DOCUMENT_RECORDS[i]])+'\n问题：'+QUESTION
@torch.no_grad()
def path_a(root):
root=Path(root);p=root/'results/cloud/seq2seq'
models,data,report=seq.load_pair(p)
rows=data['test'];model=models['dynamic'];pred,_,_=seq.greedy(model,rows)
# Rule fixed before inspecting the selected answer.
idx=next(i for i,r in enumerate(rows) if pred[i]!=r['target']);row=rows[idx]
src,l,di,y=seq.batch([row]);logits,_=model(src,l,di);lp=logits.log_softmax(-1)[0]
free=pred[idx];first=next(i for i,(a,b) in enumerate(zip(free,row['target'])) if a!=b)
z,mask=model.encode(src,l);state=model.initial(1);token=torch.tensor([seq.BOS]);trace=[];prefix=[]
for t in range(max(12,len(row['target'])+2)):
out,state,weights=model.step(token,state,z,mask);prob=out.softmax(-1)[0];chosen=int(prob.argmax())
trace.append({'step_one_based':t+1,'input_prefix':[seq.VOCAB[seq.BOS]]+[seq.VOCAB[x] for x in prefix],
'chosen':seq.VOCAB[chosen],'p_chosen':float(prob[chosen]),
'p_reference_here':float(prob[row['target'][t]]) if t<len(row['target']) else None,
'reference':seq.VOCAB[row['target'][t]] if t<len(row['target']) else None,
'source_weights':weights[0].tolist()})
prefix.append(chosen);token=torch.tensor([chosen])
if chosen==seq.EOS:break
tf=[{'step_one_based':t+1,'input_prefix':[seq.VOCAB[seq.BOS]]+[seq.VOCAB[x] for x in row['target'][:t]],
'target':seq.VOCAB[tok],'argmax':seq.VOCAB[int(lp[t].argmax())],
'p_target':float(lp[t,tok].exp()),'nll':float(-lp[t,tok])} for t,tok in enumerate(row['target'])]
metrics={}
for name,m in models.items():
nll=correct=count=0
for i in range(0,len(rows),32):
s,ll,dd,yy=seq.batch(rows[i:i+32]);oo,_=m(s,ll,dd);valid=yy!=seq.PAD
nll+=float(torch.nn.functional.cross_entropy(oo.flatten(0,1),yy.flatten(),ignore_index=seq.PAD,reduction='sum'))
correct+=int(((oo.argmax(-1)==yy)&valid).sum());count+=int(valid.sum())
pp,_,_=seq.greedy(m,rows)
metrics[name]={'teacher_forced_nll':nll/count,'teacher_forced_token_accuracy':correct/count,
'free_generation_exact_match':sum(a==b['target'] for a,b in zip(pp,rows))/len(rows),'valid_tokens':count}
return {'identity':'DN2 local CPU read-only reevaluation of DN1 cloud checkpoints; no retraining',
'selection_rule':'first dynamic greedy failure in stored test order, zero-based index',
'test_index':idx,'source':[seq.VOCAB[x] for x in row['source']],
'target':[seq.VOCAB[x] for x in row['target']], 'greedy':[seq.VOCAB[x] for x in free],
'first_divergence_one_based':first+1,'teacher_forced':tf,'free_trace':trace,
'beam':{str(w):seq.beam(model,row,w) for w in [1,2,4]},'metrics':metrics,
'config':model.config,'checkpoint_sha256':{n:hashlib.sha256((p/f'{n}-500.pt').read_bytes()).hexdigest() for n in models},
'data_sha256':hashlib.sha256((p/'data.json').read_bytes()).hexdigest()}
def path_b(root):
root=Path(root);p=mech.vision_parameters();names=['caption.webp','bank.webp']
runs=[mech.vision_read(root/'assets'/n,p,(0,1),enabled) for enabled in [True,False] for n in names]
assert np.array_equal(runs[2]['logits'],runs[3]['logits'])
return {'identity':'DN2 local CPU untrained mechanism; all weights random fixed seed 20261016',
'prefix':['a','scene'],'vocab':['a','scene','river','person','bank','EOS'],
'conditions':['A on','B on','A off','B off'],'probabilities':[r['probabilities'].tolist() for r in runs],
'max_Z_change':float(np.max(abs(runs[0]['Z']-runs[1]['Z']))),
'max_on_logit_change':float(np.max(abs(runs[0]['logits']-runs[1]['logits']))),
'max_off_logit_change':float(np.max(abs(runs[2]['logits']-runs[3]['logits']))),
'shapes':{k:list(runs[0][k].shape) for k in ['features','Z','Q','K','V','weights','read','branch','logits']},
'images':{n:hashlib.sha256((root/'assets'/n).read_bytes()).hexdigest() for n in names}}
def path_c():
text=document();moved=document([4,0,1,2,3]);vocab=sorted(set(text+'绿'));ids={c:i for i,c in enumerate(vocab)}
encode=lambda s:[ids[c] for c in s]
tokens=encode(text);model=old.TinyDecoder(seed=20261021,vocab=len(vocab));split=text.index('\n问题：')
full,_=model.full(tokens);_,cache=model.full(tokens[:split]);errors=[]
for t in range(split,len(tokens)):
actual,cache=model.incremental(tokens[t],cache,t);errors.append(float(np.max(abs(actual-full[t]))))
changed=text.replace('蓝柜','绿柜',1);changed_tokens=encode(changed)
_,oldcache=model.full(tokens[:-1]);stale,_=model.incremental(changed_tokens[-1],oldcache,len(tokens)-1)
fresh,_=model.full(changed_tokens);delta=float(np.max(abs(stale-fresh[-1])))
assert max(errors)<1e-10 and delta>1e-10
def spans(s):
return [{'event':t,'start_zero_based':re.search(r'(?m)^'+re.escape(t)+' ',s).start(),'end_exclusive':re.search(r'(?m)^'+re.escape(t)+' ',s).start()+len(t+' '+body)} for t,body in DOCUMENT_RECORDS]
return {'identity':'DN2 untrained TinyDecoder mechanism on explicit teaching character encoding, NOT document QA',
'document':text,'moved_document':moved,'question':QUESTION,'human_rule_answer':'蓝柜',
'encoding':'one Unicode code point -> one sorted-vocabulary ID; whitespace/newline/punctuation count; no BOS/template added',
'vocab':vocab,'token_ids':tokens,'token_count':len(tokens),'vocab_size':len(vocab),'prefill_tokens':split,
'spans':spans(text),'moved_spans':spans(moved),'moved_token_count':len(encode(moved)),
'full_incremental_max_error':max(errors),'changed_prefix_stale_cache_max_error':delta,
'cached_bytes':int(sum(c['K'].nbytes+c['V'].nbytes for c in cache)),
'model':{'layers':2,'heads':2,'width':8,'dtype':'float64','trained':False,'seed':20261021},
'task_prediction':'not evaluated; human answer is not a model prediction'}
def all_paths(root):
return {'environment':{'python':platform.python_version(),'torch':torch.__version__,'numpy':np.__version__},'A':path_a(root),'B':path_b(root),'C':path_c()}
if __name__=='__main__':
root=Path(__file__).resolve().parents[1];out=root/'results/dn2';out.mkdir(exist_ok=True)
result=all_paths(root);(out/'learning-paths.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n')
print(json.dumps({'A_index':result['A']['test_index'],'A_source':result['A']['source'],'A_target':result['A']['target'],
'A_greedy':result['A']['greedy'],'A_divergence':result['A']['first_divergence_one_based'],'metrics':result['A']['metrics'],
'beam':result['A']['beam'],'B':result['B'],'C':{k:v for k,v in result['C'].items() if k not in ['vocab','token_ids']}},ensure_ascii=False,indent=2))
def bpe_demo(merge_count=4, word='newer'):
"""Constructed word-boundary BPE, frequency weighted; not a production tokenizer."""
from collections import Counter
if type(merge_count) is not int or not 0<=merge_count<=8:raise ValueError('merge_count must be integer 0..8')
counts={'low':3,'lower':2,'new':1};symbols={w:list(w)+['</w>'] for w in counts};rules=[];trace=[]
def merge(parts,pair):
out=[];i=0
while i<len(parts):
if i+1<len(parts) and (parts[i],parts[i+1])==pair:out.append(parts[i]+parts[i+1]);i+=2
else:out.append(parts[i]);i+=1
return out
for step in range(merge_count):
pairs=Counter()
for w,parts in symbols.items():
for pair in zip(parts,parts[1:]):pairs[pair]+=counts[w]
if not pairs:break
best=sorted(pairs,key=lambda p:(-pairs[p],p))[0];rules.append(best)
symbols={w:merge(parts,best) for w,parts in symbols.items()}
trace.append({'round':step+1,'pair':list(best),'weighted_count':pairs[best],'symbols':symbols.copy()})
alphabet=set(''.join(counts))
if set(word)-alphabet:raise ValueError('教学字符词表未覆盖此输入；不静默假装有byte fallback')
parts=list(word)+['</w>']
for rule in rules:parts=merge(parts,rule)
return {'training_counts':counts,'merge_count':merge_count,'tie_rule':'lexicographic symbol pair',
'rules':rules,'trace':trace,'new_word':word,'encoded_new_word':parts}