LAB 04 · CLIP：候选、提示与关系检验

# 每次整册运行使用临时工作目录，保留项目旧文件与历史结果。
import os, tempfile
L03_RUN_DIR=tempfile.mkdtemp(prefix="lecture03-lab04-")
os.chdir(L03_RUN_DIR)
# 可独立运行；无需前面任何模型、文件或隐藏变量。
# 改 holder 为 left/right，先预测，再看两条规则是否仍然一致。
conditions = [
    {'condition': 1, 'left': 'red', 'right': 'blue', 'holder': 'left'},
    {'condition': 2, 'left': 'blue', 'right': 'red', 'holder': 'left'},
    {'condition': 3, 'left': 'red', 'right': 'blue', 'holder': 'right'},
    {'condition': 4, 'left': 'blue', 'right': 'red', 'holder': 'right'},
]
def choice_for_color(color):
    return {'red': 'A', 'blue': 'B'}[color]
def compare_rules(case):
    relation = choice_for_color(case[case['holder']])
    position = choice_for_color(case['left'])
    return relation, position
assert compare_rules({'left':'red','right':'blue','holder':'right'}) == ('B','A')
print('条件 | 持球关系规则 | 左侧颜色规则 | 两者不同')
for case in conditions:
    relation, position = compare_rules(case)
    print(case['condition'], relation, position, relation != position)

# 当前项目中的独立材料：已存在则复用，缺失才从本讲原站准备。
from pathlib import Path
import io, zipfile, requests
_base = Path.cwd()
_folder = 'lab10-clip'
_sentinel = 'assets/chelsea-cat.png'
_choices = [_base/_folder, _base/'lecture03-labs'/_folder, _base]
_ready = next((p for p in _choices if (p/_sentinel).is_file()), None)
if _ready is None:
    _url = 'https://codingai-lec03.pages.dev/course/materials/' + _folder + '.zip'
    _response = requests.get(_url, timeout=60)
    _response.raise_for_status()
    _payload = _response.content
    with zipfile.ZipFile(io.BytesIO(_payload)) as _archive:
        _names = [i.filename for i in _archive.infolist() if not i.is_dir()]
        _prefix = _folder + '/'
        _prefixed = all(n.startswith(_prefix) for n in _names)
        _destination = _base if _prefixed else _base/_folder
        for _item in _archive.infolist():
            _target = (_destination/_item.filename).resolve()
            if not _target.is_relative_to(_destination.resolve()):
                raise ValueError('材料包中有越界路径，停止解压。')
            if _item.is_dir():
                _target.mkdir(parents=True, exist_ok=True)
            elif not _target.exists():
                _target.parent.mkdir(parents=True, exist_ok=True)
                _target.write_bytes(_archive.read(_item))
    _ready = next((p for p in _choices if (p/_sentinel).is_file()), None)
assert _ready is not None, '请将本讲完整材料包解压到当前项目的 ' + _folder + '/；保留文件结构。'
LAB_ROOT = _ready
print('材料目录已就绪：', LAB_ROOT)

import json, hashlib, platform, sys
from pathlib import Path
import numpy as np
import matplotlib.pyplot as plt
ROOT=LAB_ROOT
if not (ROOT/'assets').is_dir() and (ROOT/'lab10-clip'/'assets').is_dir():ROOT=ROOT/'lab10-clip'
assert (ROOT/'assets').is_dir(), '请解压本实验完整包，并在该目录或其父目录运行。'
OUT=ROOT/'runs/current'; OUT.mkdir(parents=True,exist_ok=True)
FIG=OUT/'figures'; FIG.mkdir(exist_ok=True)
plt.rcParams.update({'figure.dpi':120,'font.size':10,'axes.spines.top':False,'axes.spines.right':False})
BLUE,GOLD,GREEN,PURPLE='#002676','#E6A500','#00856A','#69458D'
def sha_file(path):return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def savefig(name):
    if not plt.gcf().get_constrained_layout():plt.tight_layout()
    plt.savefig(FIG/name,dpi=140,bbox_inches='tight');plt.show()

import torch
from PIL import Image
# 若宿主缺少固定预处理依赖，仅装入本次临时目录；不修改项目环境或下载模型权重。
import importlib.util, importlib.metadata, subprocess
try:
    _transformers_version=importlib.metadata.version('transformers')
except importlib.metadata.PackageNotFoundError:
    _transformers_version=None
if _transformers_version!='4.57.1':
    _deps=ROOT/'_temporary_preprocessing_dependencies'
    subprocess.check_call([sys.executable,'-m','pip','install','--quiet','--disable-pip-version-check',
                           '--no-cache-dir','--target',str(_deps),'transformers==4.57.1',
                           'numpy=='+np.__version__,'filelock>=3.20.3,<4'])
    sys.path.insert(0,str(_deps))
    importlib.invalidate_caches()
from transformers import CLIPProcessor,CLIPImageProcessor,AutoTokenizer
import transformers
assert transformers.__version__=='4.57.1', '请使用包内固定 requirements.txt。'
torch.set_num_threads(2)
provenance=json.loads((ROOT/'assets/provenance.json').read_text())
assert sha_file(ROOT/'assets/chelsea-cat.png')==provenance['image_sha256']
REVISION=provenance['revision']
local=ROOT/'.model-cache'/REVISION
config_local=ROOT/'model-config'
# small verified config files ship in the package; weights are downloaded only on request.
manifest=json.loads((config_local/'manifest.json').read_text())
for f,expected in manifest['sha256'].items():assert sha_file(config_local/f)==expected
clip_processor=CLIPProcessor(image_processor=CLIPImageProcessor.from_pretrained(config_local,local_files_only=True),tokenizer=AutoTokenizer.from_pretrained(config_local,local_files_only=True,use_fast=True))
photo=Image.open(ROOT/'assets/chelsea-cat.png').convert('RGB')
arms={'A_base':(['cat','dog','horse'],'a photo of a {}.'),'B_expand_only':(['cat','dog','horse','tiger','fox'],'a photo of a {}.'),'C_prompt_only':(['cat','dog','horse'],'a cropped photo of a {}.'),'D_remove_cat':(['dog','horse','tiger','fox'],'a photo of a {}.')}
image_inputs=clip_processor(images=photo,return_tensors='pt')
all_prompts=list(dict.fromkeys(template.format(name) for names,template in arms.values() for name in names))
text_inputs=clip_processor(text=all_prompts,padding=True,return_tensors='pt')
assert tuple(image_inputs['pixel_values'].shape)==(1,3,224,224)
assert clip_processor.image_processor.__class__.__name__=='CLIPImageProcessor'
assert clip_processor.tokenizer.__class__.__name__=='CLIPTokenizerFast'
preprocess_verification={'scope':'preprocessing only, no weights loaded','pixel_shape':list(image_inputs['pixel_values'].shape),'pixel_sha256':hashlib.sha256(image_inputs['pixel_values'].numpy().tobytes()).hexdigest(),'prompt_count':len(all_prompts),'tokenizer':clip_processor.tokenizer.__class__.__name__,'image_processor':clip_processor.image_processor.__class__.__name__,'transformers':transformers.__version__,'image_sha256':sha_file(ROOT/'assets/chelsea-cat.png'),'revision':REVISION}
historical_preprocess=json.loads((ROOT/'historical-v16/results/preprocessing-audit-20260918.json').read_text())
assert preprocess_verification['pixel_sha256']==historical_preprocess['pixel_values_sha256']
legacy_processor=CLIPProcessor.from_pretrained(config_local,local_files_only=True)
legacy_image=legacy_processor(images=photo,return_tensors='pt')
legacy_text=legacy_processor(text=all_prompts,padding=True,return_tensors='pt')
assert torch.equal(legacy_image['pixel_values'],image_inputs['pixel_values'])
assert all(torch.equal(legacy_text[k],text_inputs[k]) for k in text_inputs)
preprocess_verification['legacy_default_pixel_max_abs_difference']=float((legacy_image['pixel_values']-image_inputs['pixel_values']).abs().max())
preprocess_verification['all_prompt_token_tensors_equal']=True
preprocess_verification['token_ids_sha256']=hashlib.sha256(text_inputs['input_ids'].numpy().tobytes()).hexdigest()
(OUT/'preprocessing-verification.json').write_text(json.dumps(preprocess_verification,indent=2));print(json.dumps(preprocess_verification,indent=2))

RUN_CLIP=False  # 已下载固定权重后改 True；默认只核对准备与预处理。
if RUN_CLIP:
    from transformers import CLIPModel, CLIPProcessor, CLIPImageProcessor, AutoTokenizer
    from PIL import Image
    REVISION='3d74acf9a28c67741b2f4f2ea7635f0aaf6f0268'
    local=ROOT/'.model-cache'/REVISION
    assert (local/'pytorch_model.bin').exists(), '先运行 python fetch_clip.py；本单元不伪造相似度。'
    assert sha_file(local/'pytorch_model.bin')==provenance['model_sha256'], '权重 SHA-256 不匹配，停止。'
    clip_model=CLIPModel.from_pretrained(local,local_files_only=True).eval()
    # Make the teacher run's slow image processing + fast tokenizer explicit.
    # A shared use_fast=False would also change the tokenizer; keep each choice separate.
    clip_processor=CLIPProcessor(
        image_processor=CLIPImageProcessor.from_pretrained(local,local_files_only=True),
        tokenizer=AutoTokenizer.from_pretrained(local,local_files_only=True,use_fast=True))
    for parameter in clip_model.parameters():parameter.requires_grad_(False)
    photo=Image.open(ROOT/'assets/chelsea-cat.png').convert('RGB')
    image_inputs=clip_processor(images=photo,return_tensors='pt')
    arms={
      'A_base':(['cat','dog','horse'],'a photo of a {}.'),
      'B_expand_only':(['cat','dog','horse','tiger','fox'],'a photo of a {}.'),
      'C_prompt_only':(['cat','dog','horse'],'a cropped photo of a {}.'),
      'D_remove_cat':(['dog','horse','tiger','fox'],'a photo of a {}.')}
    clip_records={}
    with torch.no_grad():
        image_feature=clip_model.get_image_features(**image_inputs)
        image_feature=image_feature/image_feature.norm(dim=-1,keepdim=True)
        scale=float(clip_model.logit_scale.exp())
        for arm,(names,template) in arms.items():
            prompts=[template.format(name) for name in names]
            text_inputs=clip_processor(text=prompts,return_tensors='pt',padding=True)
            text_features=clip_model.get_text_features(**text_inputs)
            text_features=text_features/text_features.norm(dim=-1,keepdim=True)
            cosine=(image_feature@text_features.T).squeeze(0)
            probabilities=(cosine*scale).softmax(-1)
            clip_records[arm]={'candidate_names':names,'prompts':prompts,'cosine':cosine.tolist(),
              'softmax':probabilities.tolist(),'top1':names[int(cosine.argmax())]}
    assert np.allclose(clip_records['A_base']['cosine'],clip_records['B_expand_only']['cosine'][:3],atol=2e-6)
    clip_result={'status':'actually run locally on CPU','model_repo':'openai/clip-vit-base-patch32','revision':REVISION,
      'model_sha256':sha_file(local/'pytorch_model.bin'),'image_sha256':sha_file(ROOT/'assets/chelsea-cat.png'),
      'image_source':'scikit-image v0.25.2 chelsea.png; Stefan van der Walt; CC0','logit_scale':scale,
      'arms':clip_records,'limitations':['one image; no dataset-level accuracy estimate','relative candidate scores are not open-set certainty']}
    (OUT/'clip-results.json').write_text(json.dumps(clip_result,indent=2))
    fig,axes=plt.subplots(1,3,figsize=(12,3.8));axes[0].imshow(photo);axes[0].axis('off');axes[0].set_title('Same real photograph in all arms')
    for arm,color in [('A_base',BLUE),('B_expand_only',GOLD)]:
        rr=clip_records[arm]; loc=np.arange(len(rr['candidate_names']))
        axes[1].plot(loc,rr['cosine'],marker='o',label=arm,color=color)
        axes[2].plot(loc,rr['softmax'],marker='o',label=arm,color=color)
    for ax,title in zip(axes[1:],['Shared cosine remains fixed','Relative softmax changes']):
        ax.set(xticks=range(5),xticklabels=arms['B_expand_only'][0],title=title);ax.legend(fontsize=8);ax.grid(alpha=.18)
    axes[1].set_ylabel('Cosine similarity');axes[2].set_ylabel('Softmax over chosen candidates')
    savefig('05-clip-candidates.png')
    print('Actual checkpoint SHA-256:',clip_result['model_sha256'])
    print('arm | top1 | candidate : cosine / softmax')
    for arm,r in clip_records.items():
        print(arm,r['top1'],[(name,round(co,4),round(pr,4)) for name,co,pr in zip(r['candidate_names'],r['cosine'],r['softmax'])])
else:
    print('CLIP 未运行：请不要将空结果写成真实相似度。')

# 当前单元无法确认宿主平台或是否使用全新进程；unknown 不等于 False。
verification={'identity':'v17 independent CLIP candidates','fresh_process':'unknown','preprocessing_actually_run':True,'clip_model_inference_actually_run':RUN_CLIP,'mnist_loaded':False,'cnn_loaded':False,'cloud_run':'unknown','historical_inference':'v16 saved CPU result; not this execution','revision':REVISION,'environment':{'python':platform.python_version(),'torch':torch.__version__,'transformers':transformers.__version__}}
(OUT/'verification.json').write_text(json.dumps(verification,indent=2));print(json.dumps(verification,indent=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)

# 可读的核心路径。默认不再重算；缓存已经由完全相同路径生成。
# 将 RUN_FRESH_CLIP 改为 True 前，按说明准备依赖和固定权重。
RUN_FRESH_CLIP=False
if RUN_FRESH_CLIP:
    import subprocess
    subprocess.run([sys.executable,str(RETRIEVAL_ROOT/'fetch_clip.py')],check=True)
    subprocess.run([sys.executable,str(RETRIEVAL_ROOT/'run_clip_retrieval.py'),'--encode'],check=True)
else:
    saved=json.loads((RETRIEVAL_ROOT/'cache/clip-results.json').read_text())
    for arm,r in saved['arms'].items():
        print(arm,list(zip(r['candidate_names'],np.round(r['cosine'],4),np.round(r['softmax'],4))))
    print('保存结果的来源:',p['execution'],p['generated_at_utc'])

# 重做缓存后的关系排序；可改排序/显示，不能改真值去迎合模型。
cos=a['relation_images']@a['relation_text'].T
print('关系 cosine:',cos)
print('预测:',cos.argmax(1),'真值:',[0,1])
assert cos.argmax(1).tolist()==[1,1]  # 当前已核验缓存的实际反例
for r in manifest['relation']:
    from IPython.display import display
    from PIL import Image
    display(Image.open(RETRIEVAL_ROOT/r['path']))