LAB 12 · CLIP延伸：缓存与检索（近似索引选读）

# 每次整册运行使用临时工作目录，保留项目旧文件与历史结果。
import os, tempfile
L03_RUN_DIR=tempfile.mkdtemp(prefix="lecture03-lab12-")
os.chdir(L03_RUN_DIR)
import numpy as np
I=np.array([[1,0],[0,1],[.8,.2]],float); T=np.array([[1,0],[0,1]],float); I/=np.linalg.norm(I,axis=1,keepdims=True); T/=np.linalg.norm(T,axis=1,keepdims=True)
S=T@I.T; print('similarity\n',S.round(3)); ranks=np.argsort(-S,axis=1); truth=[0,1]
for k in [1,2]: print('Recall@'+str(k),np.mean([truth[i] in ranks[i,:k] for i in range(2)]))

# 独立材料，不依赖另一册 notebook 的内存；重复运行复用本项目已下载内容。
from pathlib import Path
import io, json, zipfile, requests, sys, hashlib
_candidates=[Path.cwd()/'lecture03-clip-retrieval', Path.cwd()]
RETRIEVAL_ROOT=next((p for p in _candidates if (p/'cache/clip-features.npz').exists()),None)
if RETRIEVAL_ROOT is None:
    _response=requests.get('https://codingai-lec03.pages.dev/course/materials/lecture03-clip-retrieval.zip',timeout=60)
    _response.raise_for_status()
    _data=_response.content
    print('下载材料 ZIP:',len(_data),'bytes; SHA256',hashlib.sha256(_data).hexdigest())
    with zipfile.ZipFile(io.BytesIO(_data)) as z:
        for info in z.infolist():
            target=(Path.cwd()/info.filename).resolve()
            assert target.is_relative_to(Path.cwd().resolve()), '拒绝越界路径'
            if info.is_dir():target.mkdir(parents=True,exist_ok=True)
            elif not target.exists():
                target.parent.mkdir(parents=True,exist_ok=True);target.write_bytes(z.read(info))
    RETRIEVAL_ROOT=Path.cwd()/'lecture03-clip-retrieval'
assert (RETRIEVAL_ROOT/'cache/provenance.json').is_file()
sys.path.insert(0,str(RETRIEVAL_ROOT))
import run_clip_retrieval as cr
import numpy as np
p=json.loads((RETRIEVAL_ROOT/'cache/provenance.json').read_text())
assert cr.sha(RETRIEVAL_ROOT/'cache/clip-features.npz')==p['cache_sha256']
assert cr.sha(RETRIEVAL_ROOT/'assets/manifest.json')==p['manifest_sha256']
manifest=json.loads((RETRIEVAL_ROOT/'assets/manifest.json').read_text())
for r in manifest['gallery']+manifest['relation']+manifest['holder']:
    assert cr.sha(RETRIEVAL_ROOT/r['path'])==r['sha256']
a=np.load(RETRIEVAL_ROOT/'cache/clip-features.npz')
print('已核验真实 CLIP 缓存；生成时间',p['generated_at_utc'])
print('本格只载入缓存，没有在当前 notebook 运行 CLIP 编码器。')
print('gallery, queries:',a['gallery'].shape,a['queries'].shape)

X,Q=a['gallery'],a['queries']
assert np.allclose(np.linalg.norm(X,axis=1),1,atol=2e-6)
labels=np.array([r['label'] for r in manifest['gallery']])
truth=[r['label'] for r in manifest['queries']]
S=Q@X.T
exact=np.stack([cr.stable_rank(row,k=3) for row in S])
for i,row in enumerate(exact):
    print(manifest['queries'][i]['text'],'->',[(manifest['gallery'][j]['id'],round(float(S[i,j]),4)) for j in row])
hit3=np.mean([any(labels[j]==truth[i] for j in row) for i,row in enumerate(exact)])
print('exact semantic Hit@3:',hit3)
# 反方向：每张图对 12 条文字排序。每类有两条同义候选。
i2t=np.stack([cr.stable_rank(row,k=1) for row in S.T])
print('image-to-text Hit@1:',np.mean([truth[row[0]]==labels[i] for i,row in enumerate(i2t)]))

unit=cr.unit
def spherical_kmeans(X,nlist=8,seed=17,iterations=40):
    rng=np.random.default_rng(seed)
    centers=X[rng.choice(len(X),nlist,replace=False)].copy()
    for _ in range(iterations):
        ids=(X@centers.T).argmax(1)
        new=np.stack([unit(X[ids==i].mean(0,keepdims=True))[0] if np.any(ids==i) else centers[i] for i in range(nlist)])
        if np.allclose(new,centers,atol=1e-7):break
        centers=new
    ids=(X@centers.T).argmax(1)
    return centers,[np.flatnonzero(ids==i) for i in range(nlist)]

centers,lists=spherical_kmeans(X,nlist=8,seed=17)
def ivf_search(q,nprobe,k=3):
    buckets=cr.stable_rank(q@centers.T,k=nprobe)
    candidates=np.concatenate([lists[int(b)] for b in buckets])
    ranked=cr.stable_rank(q@X[candidates].T,ids=candidates,k=k)
    return ranked,candidates
for nprobe in [1,2,4,8]:
    answer=[ivf_search(q,nprobe) for q in Q]
    overlap=np.mean([len(set(e)&set(r))/3 for e,(r,c) in zip(exact,answer)])
    hit=np.mean([any(labels[j]==truth[i] for j in r) for i,(r,c) in enumerate(answer)])
    print('nprobe',nprobe,'mean scanned',np.mean([len(c) for r,c in answer]),'overlap',overlap,'Hit@3',hit)
    if nprobe==8:assert all(np.array_equal(e,r) for e,(r,c) in zip(exact,answer))
# 逐项找出 nprobe=1 的遗漏，不只看一个平均分。
for i,q in enumerate(Q):
    ranked,candidates=ivf_search(q,1)
    missing=sorted(set(exact[i])-set(ranked))
    if missing:
        print('query:',manifest['queries'][i]['text'])
        print('missed exact neighbors:',[manifest['gallery'][j]['id'] for j in missing])

N,d=X.shape;Qn=len(Q)
print('解析样本编码次数：每对重算',N*Qn,N*Qn,'；缓存',N,Qn)
print('矩阵字节：',X.nbytes,'；中心字节：',centers.nbytes,'；ID字节：',sum(l.nbytes for l in lists))
print('真实缓存生成记录：',p['image_count'],p['text_count'],'样本；',p['image_batch_calls'],p['text_batch_calls'],'批次')
print('另两对逐对重算与缓存的最大差：',p['pairwise_validation_max_abs_delta'])

R=a['relation_regions'][:,1:]  # 两图 × 两区域 × 512，丢弃full区域
# 同一图片换查询，观察权重确实变化。
atomic=R[0]@Q[[0,2]].T
weights=np.exp((atomic-atomic.max(0,keepdims=True))/.05)
weights/=weights.sum(0,keepdims=True)
print('cat / astronaut 查询的 [左区,右区] 权重：',weights.T)
# 同样两个对象交换左右；无坐标的对称聚合不可能分辨顺序。
local=np.einsum('ijd,kd->ikj',R,a['relation_text'])
w=np.exp((local-local.max(-1,keepdims=True))/.05);w/=w.sum(-1,keepdims=True)
score=(w*local).sum(-1)
print('两张交换图 × 两条关系句：',score)
assert np.allclose(score[0],score[1],atol=1e-6)

import subprocess
completed=subprocess.run([sys.executable,str(RETRIEVAL_ROOT/'run_clip_retrieval.py')],check=True,text=True,capture_output=True)
print(completed.stdout)
print('当前运行只重做检索；完整图文编码的生成身份见cache/provenance.json。')