LAB 14 · 图像问答：视觉条件与问题先验

# Detached readonly runs write only to an isolated ephemeral directory.
import os, tempfile
_LAB_WORKDIR = tempfile.mkdtemp(prefix="lecture03-lab14-")
os.chdir(_LAB_WORKDIR)
print("ephemeral workdir:", _LAB_WORKDIR)

cases=[{'cups':2,'left':'red','q':'count'},{'cups':2,'left':'red','q':'left-color'},{'cups':2,'left':'blue','q':'left-color'}]
for c in cases:
    prior='2' if c['q']=='count' else 'red'
    oracle=str(c['cups']) if c['q']=='count' else c['left']
    print(c,'question-only rule',prior,'known-attribute oracle',oracle)
# No image was read above. The learned visual path begins below.

from pathlib import Path
import json, hashlib, platform, sys, time
from datetime import datetime, timezone
import numpy as np
from PIL import Image, ImageDraw
import matplotlib
if 'get_ipython' not in globals(): matplotlib.use('Agg')
import matplotlib.pyplot as plt
import torch
from torch import nn
import torch.nn.functional as F
OUT=Path('outputs/lab14');OUT.mkdir(parents=True,exist_ok=True)
torch.set_num_threads(1);torch.manual_seed(1403);np.random.seed(1403);START=time.perf_counter()
COLORS={'red':(220,55,55),'blue':(45,95,225),'green':(35,170,75)}
QUESTIONS=['how many objects?','what color is the leftmost object?','what shape is the rightmost object?']
ANSWERS=['one','two','red','blue','green','square','circle']; A={x:i for i,x in enumerate(ANSWERS)}
print('CPU',torch.__version__,'seed 1403; answer vocabulary:',ANSWERS)

def draw_scene(count,left_color,right_shape,rng):
    img=Image.new('RGB',(40,32),(244,242,236));d=ImageDraw.Draw(img)
    slots=[10] if count==1 else [10,30]
    for j,cx in enumerate(slots):
        cy=16+int(rng.integers(-3,4));r=int(rng.integers(4,7));cx+=int(rng.integers(-1,2))
        color=left_color if j==0 else str(rng.choice(list(COLORS)))
        shape=right_shape if j==len(slots)-1 else str(rng.choice(['square','circle']))
        box=(cx-r,cy-r,cx+r,cy+r)
        if shape=='square':d.rectangle(box,fill=COLORS[color])
        else:d.ellipse(box,fill=COLORS[color])
    arr=np.array(img,dtype=np.float32)/255
    return np.clip(arr+rng.normal(0,.014,arr.shape),0,1).astype(np.float32)

def make_vqa(n,seed,biased):
    rng=np.random.default_rng(seed);images=[];answers=[];meta=[]
    # Training has a 75% red prior. Test covers all 12 condition combinations equally.
    combos=[(c,col,sh) for c in [1,2] for col in COLORS for sh in ['square','circle']]
    for i in range(n):
        if biased:
            count=int(rng.choice([1,2]));color=str(rng.choice(list(COLORS),p=[.75,.125,.125]));shape=str(rng.choice(['square','circle']))
        else:count,color,shape=combos[i%len(combos)]
        images.append(draw_scene(count,color,shape,rng))
        answers.append([A['one' if count==1 else 'two'],A[color],A[shape]])
        meta.append({'count':count,'left_color':color,'right_shape':shape})
    return torch.tensor(np.stack(images)).permute(0,3,1,2),torch.tensor(answers),meta
Xtr,Ytr,train_meta=make_vqa(720,14031,True)
Xte,Yte,test_meta=make_vqa(180,14032,False)
train_hashes={hashlib.sha256(x.numpy().tobytes()).hexdigest() for x in Xtr}
assert not train_hashes.intersection(hashlib.sha256(x.numpy().tobytes()).hexdigest() for x in Xte)
# Metadata remains only in this data/label construction and plot captions.
# model.forward receives tensors (pixels, question_id), never these dictionaries.
np.savez_compressed(OUT/'lab14-data.npz',train_images=Xtr.numpy(),train_answers=Ytr.numpy(),test_images=Xte.numpy(),test_answers=Yte.numpy())
print('train/test image counts:',len(Xtr),len(Xte),'pairs:',Ytr.numel(),Yte.numel())
print('train left-color prior:',{c:sum(m['left_color']==c for m in train_meta)/len(train_meta) for c in COLORS})
print('test left-color prior:',{c:sum(m['left_color']==c for m in test_meta)/len(test_meta) for c in COLORS})

class VQA(nn.Module):
    def __init__(self,visual):
        super().__init__();self.visual=visual
        self.question=nn.Embedding(len(QUESTIONS),12)
        if visual:
            self.cnn=nn.Sequential(nn.Conv2d(3,12,3,padding=1),nn.ReLU(),nn.MaxPool2d(2),nn.Conv2d(12,24,3,padding=1),nn.ReLU(),nn.MaxPool2d(2),nn.Flatten(),nn.Linear(24*8*10,64),nn.ReLU())
        self.answer=nn.Sequential(nn.Linear(64+12 if visual else 12,64),nn.ReLU(),nn.Linear(64,len(ANSWERS)))
    def forward(self,pixels,question_id):
        q=self.question(question_id)
        features=torch.cat([self.cnn(pixels),q],-1) if self.visual else q
        return self.answer(features)

full=VQA(True);question_only=VQA(False)
curve=[]
for label,net in [('image+question',full),('question-only',question_only)]:
    optimizer=torch.optim.Adam(net.parameters(),lr=.002)
    for epoch in range(28):
        order=torch.randperm(Ytr.numel());net.train();loss_sum=0.
        for pair_ids in order.split(72):
            image_ids=pair_ids//3;qids=pair_ids%3
            logits=net(Xtr[image_ids],qids);target=Ytr[image_ids,qids]
            loss=F.cross_entropy(logits,target)
            optimizer.zero_grad();loss.backward();optimizer.step();loss_sum+=float(loss.detach())*len(pair_ids)
        if epoch in [0,5,13,27]:print(label,'epoch',epoch+1,'CE',round(loss_sum/Ytr.numel(),4))
        curve.append({'model':label,'epoch':epoch+1,'train_ce':loss_sum/Ytr.numel()})
    torch.save(net.state_dict(),OUT/('lab14-'+('visual' if net.visual else 'question-only')+'-weights.pt'))

@torch.no_grad()
def answer_all(net,images):
    net.eval()
    return torch.stack([net(images,torch.full((len(images),),q)).argmax(-1) for q in range(3)],1)
normal=answer_all(full,Xte);qo=answer_all(question_only,Xte)
zero=answer_all(full,torch.zeros_like(Xte));shuffled=answer_all(full,Xte[torch.roll(torch.arange(len(Xte)),1)])
train_full=answer_all(full,Xtr);train_qo=answer_all(question_only,Xtr)
def score(p,y):return {'overall':float(p.eq(y).float().mean()),**{QUESTIONS[q]:float(p[:,q].eq(y[:,q]).float().mean()) for q in range(3)}}
metrics={'test_image_question':score(normal,Yte),'test_question_only':score(qo,Yte),'test_zero_pixels':score(zero,Yte),'test_shuffled_pixels':score(shuffled,Yte),'train_image_question':score(train_full,Ytr),'train_question_only':score(train_qo,Ytr)}
print(json.dumps(metrics,indent=2))
# Recreate two images with matched nuisance randomness, only left color differs.
control_specs=[(2,'red','square'),(2,'blue','square'),(1,'green','circle')]
controls_x=torch.tensor(np.stack([draw_scene(*s,np.random.default_rng(444)) for s in control_specs])).permute(0,3,1,2)
controls_pred=answer_all(full,controls_x);controls_qo=answer_all(question_only,controls_x)
controls=[{'image_condition':list(s),'question':q,'image_question':ANSWERS[int(controls_pred[i,j])],'question_only':ANSWERS[int(controls_qo[i,j])]} for i,s in enumerate(control_specs) for j,q in enumerate(QUESTIONS)]
print('same image / changed question and same question / changed image:')
for row in controls:print(row)

fig,axes=plt.subplots(1,3,figsize=(11,4))
for i,ax in enumerate(axes):
    ax.imshow(controls_x[i].permute(1,2,0));ax.axis('off')
    rows=['count: ','left color: ','right shape: ']
    text='\n'.join(rows[j]+ANSWERS[int(controls_pred[i,j])]+' [Q-only: '+ANSWERS[int(controls_qo[i,j])]+']' for j in range(3))
    ax.set_title(text,fontsize=10)
fig.suptitle('LAB14: same questions, changed image; three questions per image',fontsize=13)
fig.tight_layout();fig.savefig(OUT/'lab14-vqa-controls.png',dpi=160);plt.show();plt.close(fig)
fig,ax=plt.subplots(figsize=(10,3.6));labels=['image + Q','Q only','zero pixels','shuffled pixels'];keys=['test_image_question','test_question_only','test_zero_pixels','test_shuffled_pixels'];values=[metrics[k]['overall'] for k in keys]
ax.bar(labels,values,color=['#356b91','#ad8251','#9da6ad','#9da6ad']);ax.set(ylim=(0,1.1),ylabel='answer accuracy (540 held-out pairs)')
for i,v in enumerate(values):ax.text(i,v+.02,f'{v:.1%}',ha='center')
fig.tight_layout();fig.savefig(OUT/'lab14-vqa-ablation.png',dpi=160);plt.show();plt.close(fig)
fig,ax=plt.subplots(figsize=(8,3.2))
for label in ['image+question','question-only']:
    r=[v for v in curve if v['model']==label];ax.plot([v['epoch'] for v in r],[v['train_ce'] for v in r],label=label)
ax.set(xlabel='epoch',ylabel='answer cross entropy');ax.legend();fig.tight_layout();fig.savefig(OUT/'lab14-vqa-learning.png',dpi=160);plt.show();plt.close(fig)
result={'notebook_id':'11d9f74871544a77b42024a9fe06b81a','execution':'local CPU, fresh process; not a Deepnote cloud run','created_utc':datetime.now(timezone.utc).isoformat(),'elapsed_seconds':time.perf_counter()-START,'environment':{'python':sys.version,'torch':torch.__version__,'numpy':np.__version__,'platform':platform.platform()},'split':{'train_seed':14031,'test_seed':14032,'train_images':720,'test_images':180,'questions_per_image':3,'split_before_question_expansion':True,'train_red_prior':.75,'test_color_prior':'uniform by construction'},'metrics':metrics,'curve':curve,'controls':controls,'predictions':[{'truth':[ANSWERS[int(v)] for v in Yte[i]],'image_question':[ANSWERS[int(v)] for v in normal[i]],'question_only':[ANSWERS[int(v)] for v in qo[i]]} for i in range(len(Xte))],'files_sha256':{p.name:hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(OUT.glob('*.pt'))},'data_sha256':hashlib.sha256((OUT/'lab14-data.npz').read_bytes()).hexdigest()}
(OUT/'lab14-results.json').write_text(json.dumps(result,indent=2));print('saved',OUT,'seconds',round(result['elapsed_seconds'],2))