Train en toets een eigen diffusiegenerator
Verbind synthetische data, echte Adamupdates, generatie en verdelingsdiagnostiek in vijf notebookcellen.
De opdracht, leerdoelen en voorkennis
- Je traint een eigen kleine ruisvoorspeller en vergelijkt begin- en eindparameters eerlijk.
- Je beoordeelt voorspelfout, modusbalans en spreiding met drie echte rapportfiguren.
Lees eerst: Pas de stapgrootte per parameter aan · Genereer stap voor stap vanuit nieuwe ruis.
Je kunt nu ruis toevoegen, ruisschattingen berekenen en achterwaarts samplen. Het ontbrekende onderdeel is een netwerk dat de ruisschatting werkelijk heeft geleerd. In dit notebook verbind je de functies tot een klein getraind diffusiesysteem en controleer je wat de nieuwe punten doen.
Onze doelverdeling bestaat uit vier even waarschijnlijke wolkjes rond [−2;−2], [−2;2], [2;−2] en [2;2]. Iedere coördinaat krijgt normale ruis met standaardafwijking 0,15. De data zijn volledig synthetisch; het project leert geen foto's of taalbeschrijvingen genereren.
De schema-, netwerk- en gradiënthulpfuncties uit de conceptlessen zijn zichtbaar gegeven. Jij bouwt data, training, generatie, evaluatie en figuren. Iedere cel bewaart haar resultaat voor de volgende; bij projectcontrole worden alle cellen opnieuw in een schone omgeving uitgevoerd.
1. Maak een reproduceerbare taak
Schrijf synthetic_points met de aangeleverde NumPy-generator. Trek eerst de centrumindices als batch en daarna de normale ruis als matrix. Zo ligt ook vast hoe een seed een concrete trekproef bepaalt; deze functie mag de RNG bewust voortzetten.
points = synthetic_points(8,
np.random.default_rng(5))
print(points.shape)
# (8, 2)
Willekeurige centrumkeuze betekent geen verplichte gelijke aantallen in iedere minibatch. Een batch van acht punten hoeft niet precies twee punten per wolkje te bevatten. De gelijke kans is een eigenschap van het trekproces, niet een quota per kleine batch.
012. Bewaar wat training echt doet
Schrijf train_denoiser met de gegeven initialisatie en gradiënten. Trek voor iedere update nieuwe schone punten, tijdindices en doelruis, en verstoor de punten op hun eigen tijdstappen. Pas daarna de bekende Adamregel toe op alle vier parameterarrays.
betas = np.linspace(.001,.15,64)
trained = train_denoiser(betas)
print(trained['steps'])
print(len(trained['curve']))
# 2000
# 41
Bewaar de beginparameters als onafhankelijke kopie. Registreer het berekende minibatchverlies bij stap één, iedere vijftig updates en de laatste update. De curve mag schommelen: iedere minibatch bevat nieuwe punten en ruis, dus een daling bij iedere afzonderlijke meetstap is geen eis.
023. Genereer zonder het antwoord mee te geven
Schrijf sample_diffusion met één eigen generator voor start- en innovatieruis. Loop terug door het schema en gebruik alleen de getrainde parameters, de huidige punten en de tijdindices. Geen schoon doelpunt of bekende trainingsruis meesturen.
samples = sample_diffusion(
trained['parameters'],betas,
count=256,seed=97)
print(samples['points'].shape)
# (256, 2)
De snapshots helpen om te zien hoe een wolk verandert. Ze zijn zelfstandige kopieën: de laatste update mag een eerder beeld niet achteraf overschrijven. Een plausibele puntenwolk blijft slechts één onderdeel van de beoordeling.
034. Meet meer dan afstand tot centra
Schrijf diffusion_diagnostics en evaluate_diffusion. Vergelijk begin- en eindparameters op exact dezelfde onafhankelijke validatiepunten, tijdindices en ruis. Gebruik ook dezelfde samplingseed voor beide generaties, zodat toeval het verschil minder vertroebelt.
report = evaluate_diffusion(
trained,betas)
print(round(
report['validation']['initial'],3))
print(round(
report['validation']['trained'],3))
# 0.956
# 0.303
Deze getallen horen bij onze vaste configuratie en seed. Ze zijn geen algemene kwaliteitsgarantie. Het rapport bevat ook modusbezetting, centrumafstand en populatievariantie binnen iedere toegewezen modus; lege groepen krijgen geen verzonnen variantie.
Voor balans vergelijken we de geobserveerde fracties met een kwart. N is het totale aantal punten en n_k het aantal bij centrum k. De totale variatieafstand TV is nul bij perfect gelijke aantallen.
Exact de vier centra uitsturen geeft perfecte balans en centrumafstand nul, maar spreiding nul. De echte wolkjes hebben coördinaatvariantie 0,0225. Daarom tonen we de gewogen binnen-modusvariantie erbij; ook deze combinatie is nog geen volledige toets van de verdeling.
045. Teken de resultaten zonder opnieuw te leren
Schrijf plot_diffusion met uitsluitend report als invoer. Geef drie geopende figuren terug: de werkelijke curve, de doel-/gegenereerde punten en de balans-/spreidingspanelen. De artists moeten precies de rapportdata weergeven.
figures = plot_diffusion(report)
print(len(figures))
# 3
Opnieuw trainen tijdens plotten is een misvatting: dan kunnen figuur en rapport bij verschillende modelparameters horen. De nakijker controleert bovendien of de figuren zichtbaar blijven. Hierna verken je hoe geluid in meetbare representaties wordt omgezet.
05Wat neem je mee?
- Ruisdoelen komen uit training; de generator krijgt die antwoorden niet.
- Bewaar werkelijke updates, parameters, zaden en meetdata bij elkaar.
- Beoordeel voorspelling, balans en spreiding samen; één lage foutmaat volstaat niet.
Bronnen: DDPM 2020 v2 — ruisvoorspelling en omgekeerde diffusie ↗ · Gecontroleerd 2026-10-29.
1. Maak controleerbare trainingspunten
Vul synthetic_points in voor vier even waarschijnlijke centra met onafhankelijke normale ruis van standaardafwijking 0,15. De gegeven generator levert eerst centrumindices en daarna ruis; alleen die RNG-toestand wordt bewust voortgezet.
Daaronder staan de gewone validators, het schema, featureopbouw, netwerkberekening en gradiënten uit de conceptlessen. Ze zijn gegeven en zichtbaar; gebruik hun vaste grenzen.
De demo maakt acht punten en bewaart het schema van 64 stappen. Pas wanneer jouw functie een puntenmatrix teruggeeft, gaat de volgende cel verder.
Bekijk voorbeeld
import numpy as np
def synthetic_points(count,rng):
count=_dd_int(count,1,1024)
centers=np.array([[-2.,-2.],[-2.,2.],[2.,-2.],[2.,2.]])
return centers[rng.integers(0,4,size=count)]+rng.normal(0,.15,size=(count,2))
# Gegeven helpers
def _dd_array(value,ndim):
raw=np.asarray(value)
if raw.ndim!=ndim or raw.dtype.kind not in 'iuf' or not raw.size or raw.size>65536:
raise ValueError('Niet-lege reële array met de juiste vorm tot65536cellen vereist')
result=raw.astype(float,copy=True)
if not np.isfinite(result).all() or np.max(np.abs(result))>1e6:raise ValueError('Eindige begrensde waarden vereist')
return result
def _dd_int(value,minimum,maximum):
if type(value) is not int or not minimum<=value<=maximum:raise ValueError('Geheel getal buiten oefengrens')
return value
def diffusion_schedule(betas):
beta=_dd_array(betas,1)
if not 2<=len(beta)<=128 or np.any(beta<.0001) or np.any(beta>.2):raise ValueError('2–128ruisstappen met beta0.0001–0.2 vereist')
alpha=1-beta;bar=np.cumprod(alpha);previous=np.concatenate([[1.],bar[:-1]])
variance=beta*(1-previous)/(1-bar)
return {'beta':beta,'alpha':alpha,'bar':bar,'previous':previous,'variance':variance}
def _dd_indices(indices,count,steps):
indices=np.asarray(indices)
if indices.dtype.kind not in 'iu' or indices.shape!=(count,) or np.any(indices<0) or np.any(indices>=steps):
raise ValueError('Eén gehele nulgebaseerde stapindex per punt vereist')
return indices.astype(int,copy=True)
def forward_diffusion(clean,noise,indices,betas):
clean=_dd_array(clean,2);noise=_dd_array(noise,2);schedule=diffusion_schedule(betas)
if clean.shape!=noise.shape or clean.shape[1]!=2:raise ValueError('Passende tweedimensionale punten vereist')
indices=_dd_indices(indices,len(clean),len(schedule['beta']));bar=schedule['bar'][indices,None]
return np.sqrt(bar)*clean+np.sqrt(1-bar)*noise
def diffusion_features(noisy,indices,steps):
noisy=_dd_array(noisy,2);steps=_dd_int(steps,2,128)
if noisy.shape[1]!=2 or len(noisy)>1024:raise ValueError('Maximaal1024tweedimensionale punten vereist')
indices=_dd_indices(indices,len(noisy),steps);phase=np.pi*(indices+1)/steps
return np.column_stack([noisy,np.sin(phase),np.cos(phase)])
def initialize_denoiser(hidden=32,seed=83):
hidden=_dd_int(hidden,1,64);seed=_dd_int(seed,0,2**32-1);rng=np.random.default_rng(seed)
return {'w1':rng.normal(0,.3,size=(4,hidden)),'b1':np.zeros(hidden),
'w2':rng.normal(0,.1,size=(hidden,2)),'b2':np.zeros(2)}
def _dd_parameters(parameters):
if type(parameters) is not dict or set(parameters)!={'w1','b1','w2','b2'}:raise ValueError('Vier expliciete parameterarrays vereist')
result={name:_dd_array(parameters[name],2 if name.startswith('w') else 1) for name in parameters}
hidden=len(result['b1'])
if not 1<=hidden<=64 or result['w1'].shape!=(4,hidden) or result['w2'].shape!=(hidden,2) or result['b2'].shape!=(2,):
raise ValueError('Aansluitende kleine netwerkvormen vereist')
return result
def predict_noise(parameters,noisy,indices,steps):
parameters=_dd_parameters(parameters);features=diffusion_features(noisy,indices,steps)
if len(features)*len(parameters['b1'])>65536:raise ValueError('Verborgen laag te groot')
hidden=np.tanh(features@parameters['w1']+parameters['b1'])
return hidden@parameters['w2']+parameters['b2']
def noise_loss_and_grad(parameters,noisy,indices,target_noise,steps):
parameters=_dd_parameters(parameters);features=diffusion_features(noisy,indices,steps);target=_dd_array(target_noise,2)
if target.shape!=(len(features),2) or len(features)*len(parameters['b1'])>65536:raise ValueError('Passende ruisdoelen en begrensde activaties vereist')
hidden=np.tanh(features@parameters['w1']+parameters['b1']);prediction=hidden@parameters['w2']+parameters['b2']
residual=prediction-target;loss=float(np.mean(residual**2));dout=2*residual/residual.size
dhidden=(dout@parameters['w2'].T)*(1-hidden**2)
gradients={'w1':features.T@dhidden,'b1':dhidden.sum(axis=0),'w2':hidden.T@dout,'b2':dout.sum(axis=0)}
return loss,gradients
def reverse_diffusion_step(noisy,predicted_noise,index,betas,noise):
noisy=_dd_array(noisy,2);predicted=_dd_array(predicted_noise,2);noise=_dd_array(noise,2);schedule=diffusion_schedule(betas)
index=_dd_int(index,0,len(schedule['beta'])-1)
if noisy.shape!=predicted.shape or noise.shape!=noisy.shape or noisy.shape[1]!=2:raise ValueError('Passende punt-/ruisvormen vereist')
mean=(noisy-schedule['beta'][index]*predicted/np.sqrt(1-schedule['bar'][index]))/np.sqrt(schedule['alpha'][index])
# The first forward step is the last reverse step: posterior variance is zero.
return mean+np.sqrt(schedule['variance'][index])*noise
betas = np.linspace(.001,.15,64)
points = synthetic_points(8,
np.random.default_rng(5))
print(points.shape
if isinstance(points,np.ndarray)
else None)
2. Train de ruisvoorspeller
Vul train_denoiser in. Trek voor iedere update nieuwe schone punten, tijdindices en doelruis; geef de voorspeller alleen verstoorde punten en tijdkenmerken.
Gebruik de gegeven gradiënten en de bekende Adamregel met biascorrecties en epsilon buiten de wortel. Bewaar beginparameters apart, en registreer het echte verlies bij update 1, iedere 50 updates en de laatste update.
De vaste demo traint 2000 updates met batchgrootte 128. Begin-, eindparameters en curve komen in trained; geen andere cel mag deze training ongemerkt opnieuw uitvoeren.
Bekijk voorbeeld
def train_denoiser(betas,steps=2000,batch=128,hidden=32,rate=.005,seed=83):
schedule=diffusion_schedule(betas);steps=_dd_int(steps,1,3000);batch=_dd_int(batch,1,256)
if type(rate) not in (int,float) or not .00001<=rate<=.02:raise ValueError('Leersnelheid buiten oefengrens')
parameters=initialize_denoiser(hidden,seed);initial={key:value.copy() for key,value in parameters.items()}
rng=np.random.default_rng(seed+1);first={key:np.zeros_like(value) for key,value in parameters.items()};second={key:np.zeros_like(value) for key,value in parameters.items()}
curve=[]
for step in range(1,steps+1):
clean=synthetic_points(batch,rng);indices=rng.integers(0,len(schedule['beta']),size=batch);target=rng.normal(size=clean.shape)
noisy=forward_diffusion(clean,target,indices,betas);loss,gradients=noise_loss_and_grad(parameters,noisy,indices,target,len(betas))
for key in parameters:
first[key]=.9*first[key]+.1*gradients[key];second[key]=.999*second[key]+.001*gradients[key]**2
parameters[key]-=rate*(first[key]/(1-.9**step))/(np.sqrt(second[key]/(1-.999**step))+1e-8)
if step==1 or step%50==0 or step==steps:curve.append({'step':step,'loss':loss})
return {'initial':initial,'parameters':parameters,'curve':curve,'steps':steps,'batch':batch}
trained = None
if isinstance(points,np.ndarray):
trained = train_denoiser(betas)
print(len(trained['curve'])
if isinstance(trained,dict)
else None)
3. Genereer vanuit nieuwe ruis
Vul sample_diffusion in met een eigen generator voor startruis en innovaties. Gebruik dezelfde getrainde parameters op alle achterwaartse stappen.
De sampler krijgt geen schone punten of bekende trainingsruis. Bewaar zelfstandige snapshots van start, na de middelste index en einde.
De demo genereert 256 nieuwe punten. Dit losse experiment bereidt de vergelijkbare begin/eindevaluatie van de volgende cel voor.
Bekijk voorbeeld
def sample_diffusion(parameters,betas,count=256,seed=97):
parameters=_dd_parameters(parameters);schedule=diffusion_schedule(betas);count=_dd_int(count,1,1024);seed=_dd_int(seed,0,2**32-1)
rng=np.random.default_rng(seed);points=rng.normal(size=(count,2));snapshots=[points.copy()]
for index in range(len(betas)-1,-1,-1):
predicted=predict_noise(parameters,points,np.full(count,index),len(betas))
points=reverse_diffusion_step(points,predicted,index,betas,rng.normal(size=points.shape))
if index in [len(betas)//2,0]:snapshots.append(points.copy())
return {'points':points,'snapshots':snapshots}
samples = None
if isinstance(trained,dict):
samples = sample_diffusion(
trained['parameters'],betas,
count=256,seed=97)
print(samples['points'].shape
if isinstance(samples,dict)
else None)
4. Meet voorspelling en verdeling
Vul diffusion_diagnostics en evaluate_diffusion in. Meet centrumafstand, modusbezetting, totale variatieafstand tot uniforme bezetting en gewogen populatievariantie binnen modi.
Beoordeel begin- en eindparameters op exact dezelfde nieuwe validatiepunten, tijdindices en ruis. Gebruik voor beide generaties dezelfde samplingseed; neem ook een onafhankelijke doeltrekking op.
Geef samples, validation, diagnostics, een gekopieerde curve en de meetmetadata terug. Het rapport bevat echte berekeningen, geen conclusies uit een dalende trainingscurve alleen.
Bekijk voorbeeld
def diffusion_diagnostics(points):
points=_dd_array(points,2)
if points.shape[1]!=2 or not 1<=len(points)<=1024:raise ValueError('Maximaal1024tweedimensionale punten vereist')
centers=np.array([[-2.,-2.],[-2.,2.],[2.,-2.],[2.,2.]])
distances=np.sum((points[:,None,:]-centers[None,:,:])**2,axis=2);chosen=np.argmin(distances,axis=1)
counts=np.bincount(chosen,minlength=4);within=np.zeros(2);groups=[]
for index in range(4):
group=points[chosen==index]
if len(group):
residual=group-centers[index];variance=np.mean((residual-residual.mean(axis=0))**2,axis=0)
within+=len(group)*variance
groups.append({'count':len(group),'offset':residual.mean(axis=0).tolist(),'variance':variance.tolist()})
else:groups.append({'count':0,'offset':None,'variance':None})
return {'mean_squared_center_distance':float(np.min(distances,axis=1).mean()),'counts':counts.tolist(),
'balance_tv':float(.5*np.abs(counts/len(points)-.25).sum()),'within_variance':(within/len(points)).tolist(),'groups':groups}
def evaluate_diffusion(trained,betas,count=512,seed=9901,sample_seed=97):
if type(trained) is not dict or not {'initial','parameters','curve','steps','batch'}<=set(trained):
raise ValueError('Een trainingsresultaat met begin/eindparameters en curve vereist')
initial=_dd_parameters(trained['initial']);parameters=_dd_parameters(trained['parameters'])
schedule=diffusion_schedule(betas);count=_dd_int(count,1,1024);seed=_dd_int(seed,0,2**32-1);sample_seed=_dd_int(sample_seed,0,2**32-1)
steps=_dd_int(trained['steps'],1,3000);batch=_dd_int(trained['batch'],1,256)
expected=sorted(set([1,steps,*range(50,steps+1,50)]))
if type(trained['curve']) is not list or len(trained['curve'])!=len(expected):raise ValueError('Een volledig geregistreerde trainingscurve vereist')
curve=[]
for row,index in zip(trained['curve'],expected):
if type(row) is not dict or set(row)!={'step','loss'} or type(row['step']) is not int or row['step']!=index:
raise ValueError('Bewaar de werkelijke geregistreerde updatestappen')
if type(row['loss']) not in (int,float) or not 0<=row['loss']<=1e100:raise ValueError('Eindig niet-negatief verlies vereist')
curve.append({'step':row['step'],'loss':float(row['loss'])})
rng=np.random.default_rng(seed);clean=synthetic_points(count,rng);indices=rng.integers(0,len(schedule['beta']),size=count);noise=rng.normal(size=clean.shape)
noisy=forward_diffusion(clean,noise,indices,betas)
losses={key:noise_loss_and_grad(value,noisy,indices,noise,len(schedule['beta']))[0] for key,value in [('initial',initial),('trained',parameters)]}
samples={key:sample_diffusion(value,betas,count,sample_seed)['points'] for key,value in [('initial',initial),('trained',parameters)]}
samples['target']=clean
return {'validation':losses,'samples':samples,'diagnostics':{key:diffusion_diagnostics(value) for key,value in samples.items()},
'curve':curve,'steps':steps,'batch':batch,'count':count,'terminal_alpha_bar':float(schedule['bar'][-1])}
report = None
if isinstance(trained,dict) and \
isinstance(samples,dict):
probe = diffusion_diagnostics(
samples['points'])
if isinstance(probe,dict):
report = evaluate_diffusion(
trained,betas)
print(report['count']
if isinstance(report,dict)
else None)
5. Teken de echte uitkomsten
Vul plot_diffusion in en geef drie werkelijk geopende Matplotlib-figuren terug. De curve moet de geregistreerde stappen/verliezen tonen en de puntenwolk de echte doel- en gegenereerde punten.
De derde figuur toont modusbalans en binnen-modusvariantie. Zet de vaste variantie 0,0225 en het geval met alleen centra op nul erbij om de beperkingen van centrumafstand zichtbaar te maken.
Gebruik uitsluitend report. Plotten mag niet opnieuw trainen of samplen; de nakijker vergelijkt artists met de aangeleverde data en controleert dat de figuren zichtbaar blijven.
Bekijk voorbeeld
def plot_diffusion(report):
# Reports are produced by evaluate_diffusion; plots never retrain or resample.
figures=[]
figure,axis=plt.subplots(figsize=(4.2,3.5),layout='constrained')
axis.plot([row['step'] for row in report['curve']],[row['loss'] for row in report['curve']],color='#236b4f')
axis.set(title='Werkelijk minibatchverlies',xlabel='Update',ylabel='Ruis-MSE');figures.append(figure)
figure,axis=plt.subplots(figsize=(4.2,4.2),layout='constrained')
for key,label,color in [('target','Doelverdeling','#3868a0'),('trained','Nieuwe generatie','#236b4f')]:
points=report['samples'][key];axis.scatter(points[:,0],points[:,1],s=7,alpha=.35,label=label,color=color)
axis.set(title='Nieuwe punten uit normale ruis',xlabel='Coördinaat 1',ylabel='Coördinaat 2');axis.legend(fontsize=11.5);figures.append(figure)
figure,axes=plt.subplots(2,1,figsize=(4.2,6.5),layout='constrained')
order=['initial','trained','target'];labels=['Begin','Getraind','Doel']
axes[0].bar(labels,[report['diagnostics'][key]['balance_tv'] for key in order],color=['#a66a43','#236b4f','#3868a0'])
axes[0].set(title='Modusbalans: afwijking van ¼',ylabel='TV-afstand',ylim=(0,1))
axes[1].bar(['Getraind','Doel','Centra'],[float(np.mean(report['diagnostics'][key]['within_variance'])) for key in ['trained','target']]+[0],color=['#236b4f','#3868a0','#777'])
axes[1].axhline(.0225,color='#555',linestyle='--',linewidth=1,label='Werkelijke variantie 0,0225')
axes[1].set(title='Spreiding binnen de modi',ylabel='Gem. coördinaatvariantie');axes[1].legend(fontsize=11.5);figures.append(figure)
for figure in figures:
for axis in figure.axes:
axis.tick_params(labelsize=11);axis.title.set_fontsize(12);axis.xaxis.label.set_size(11);axis.yaxis.label.set_size(11)
return figures
import matplotlib.pyplot as plt
figures = None
if isinstance(report,dict):
figures = plot_diffusion(report)
print(len(figures)
if isinstance(figures,list)
else None)