Buckets:
| import hashlib,json,os | |
| from pathlib import Path | |
| import numpy as np | |
| def file_hash(path): | |
| h=hashlib.sha256() | |
| with open(path,'rb') as f: | |
| for chunk in iter(lambda:f.read(1024*1024),b''):h.update(chunk) | |
| return h.hexdigest() | |
| def atomic_json(path,value): | |
| path=Path(path);tmp=path.with_suffix(path.suffix+'.tmp');tmp.write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n');tmp.replace(path) | |
| def validate_arrays(root,manifest,finite=False): | |
| for layer in manifest['layers']: | |
| for pooling in ['last','mean']: | |
| p=Path(root)/f'layer{layer}_{pooling}.npy' | |
| assert p.exists(),f'Missing feature array {p}; restore it before resuming' | |
| a=np.load(p,mmap_mode='r') | |
| assert a.shape==(manifest['n'],manifest['d']) and a.dtype==np.float16,f'Invalid feature array {p}' | |
| if finite:assert np.isfinite(a).all(),f'Nonfinite feature array {p}' | |
| def validate_feature_table(root,table,tokens): | |
| """Reject changed tables, reordered IDs, incomplete caches or stale token content.""" | |
| import pandas as pd | |
| root=Path(root);tokens=Path(tokens) | |
| manifest=json.loads((root/'manifest.json').read_text());complete=json.loads((root/'COMPLETE.json').read_text()) | |
| for k in ['n','model','model_revision','tokens_sha256']: | |
| assert complete[k]==manifest[k],f'Completion provenance mismatch: {k}' | |
| done=np.load(root/'done.npy');assert done.shape==(manifest['n'],) and done.all() | |
| validate_arrays(root,manifest) | |
| ex=pd.read_parquet(table).sort_values('feature_index').reset_index(drop=True) | |
| meta=pd.read_parquet(root/'metadata.parquet').sort_values('feature_index').reset_index(drop=True) | |
| assert len(ex)==manifest['n'] and np.array_equal(ex.feature_index,np.arange(len(ex))) | |
| for k in ['feature_index','response_id','prompt_id']: | |
| assert np.array_equal(ex[k],meta[k]),f'Feature/table ID mismatch: {k}' | |
| token_hash=file_hash(tokens);assert token_hash==manifest['tokens_sha256'],'Changed token cache' | |
| provenance=tokens.with_suffix('.provenance.json') | |
| if not provenance.exists(): | |
| # Older original-response caches predate content provenance. Verify every | |
| # stored token/mask by exact re-encoding once; do not recompute vectors. | |
| from transformers import AutoTokenizer | |
| from representations.pooling import encode_response | |
| tok=AutoTokenizer.from_pretrained(manifest['model'],revision=manifest['tokenizer_revision'],local_files_only=True) | |
| td=pd.read_parquet(tokens).sort_values('feature_index').reset_index(drop=True) | |
| for i,row in enumerate(ex.itertuples()): | |
| e=encode_response(tok,row.prompt,row.response,manifest['max_length']) | |
| for k in ['input_ids','response_mask']: | |
| assert np.array_equal(e[k],td.iloc[i][k]),f'Token/source content mismatch row {i}: {k}' | |
| if (i+1)%5000==0:print(f'Validated original token content {i+1}/{len(ex)}',flush=True) | |
| provenance.write_text(json.dumps({'source_path':str(table),'source_sha256':file_hash(table),'tokens_sha256':token_hash,'model':manifest['model'],'revision':manifest['tokenizer_revision'],'pooling_sha256':file_hash('representations/pooling.py'),'validation':'all rows exactly re-encoded against preexisting token cache'},indent=2)) | |
| p=json.loads(provenance.read_text()) | |
| assert p['source_sha256']==file_hash(table),'Table changed after tokenization; regenerate interventions only before extracting' | |
| assert p['tokens_sha256']==token_hash and p['revision']==manifest['tokenizer_revision'] | |
| assert p['pooling_sha256']==manifest['pooling_sha256']==file_hash('representations/pooling.py') | |
| return dict(manifest=manifest,table_sha256=p['source_sha256'],tokens_sha256=token_hash,metadata_sha256=file_hash(root/'metadata.parquet')) | |
| class FeatureCache: | |
| def __init__(self,path,manifest): | |
| self.path=Path(path); self.path.mkdir(parents=True,exist_ok=True); f=self.path/'manifest.json' | |
| if f.exists(): assert json.loads(f.read_text())==manifest,'Cache provenance mismatch' | |
| else: f.write_text(json.dumps(manifest,indent=2)+'\n') | |
| p=self.path/'done.npy'; self.done=np.load(p) if p.exists() else np.zeros(manifest['n'],dtype=bool) | |
| assert self.done.shape==(manifest['n'],) | |
| if self.done.any():validate_arrays(self.path,manifest,finite=True) | |
| self.arrays={} | |
| for layer in manifest['layers']: | |
| for pool in ['last','mean']: | |
| key=f'layer{layer}_{pool}'; p=self.path/f'{key}.npy' | |
| self.arrays[key]=np.lib.format.open_memmap(p,mode='r+' if p.exists() else 'w+',dtype='float16',shape=(manifest['n'],manifest['d'])) | |
| def write(self,indices,pooled): | |
| assert set(pooled)==set(self.arrays) | |
| for key,val in pooled.items(): | |
| assert np.isfinite(val).all(); self.arrays[key][indices]=val | |
| self.arrays[key].flush() | |
| self.done[indices]=True | |
| with open(self.path/'done.tmp','wb') as f:np.save(f,self.done) | |
| os.replace(self.path/'done.tmp',self.path/'done.npy') | |
Xet Storage Details
- Size:
- 5.07 kB
- Xet hash:
- bf4fb819117aad61346b00e57b8e4bf9add75f6e466379f96510927718c2514d
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.