← All the evidencesentinel_interrupt_learning_experiment.py
SHA-256 d7aae1b004caa4430e6d514f2408387aad96088d82b989f83f7bd490319cdd1d · download the original
#!/usr/bin/env python3
"""Sentinel synthetic study: learning whether/when a candidate thought merits an interruption.
No user data, no LLM. All latent preferences/outcomes are generated solely by a
simulation and kept out of every non-oracle policy. Run with NumPy only.
"""
from __future__ import annotations
import argparse, csv, json, math, hashlib, statistics
from collections import defaultdict, deque
from pathlib import Path
import numpy as np
ROOT=Path(__file__).resolve().parent
BLOCK=100
CAP=8 # Max alerts per block of 100 candidates
AUDIT_PROB=.035 # Independent, user-consented retrospective reviews
AUDIT_COST=.18 # Utility cost per retrospective review
N_FEATURES=16
METHODS=("silent", "fixed", "naive_silence_negative", "bandit_explore", "audit_only", "prior_paced", "feedback_paced", "audit_paced", "audit_stratified", "audit_plus_probe", "oracle")
def features(p, goal, deadline, busy, cat):
t1=float(cat==1);t2=float(cat==2);t3=float(cat==3)
return np.array([1, p, p*goal, p*deadline, busy,
t1,t2,t3, p*t1,p*t2,p*t3,
p*goal*t1,p*goal*t2,p*goal*t3,
p*deadline*t3, deadline*busy], dtype=np.float64)
def make_world(seed, n, condition):
r=np.random.default_rng(seed)
catprob=np.array([.50,.29,.18,.03])
if condition=='rare':catprob=np.array([.51,.30,.178,.012])
cat=r.choice(4,size=n,p=catprob)
confidence=np.clip(r.beta(2.2,2.0,n),.04,.96)
goal=r.binomial(1,.42,n)
deadline=r.binomial(1,.27,n)
busy=r.beta(2.,3.,n)
busy += .18*(r.random(n)<.17)
busy=np.minimum(busy,1.)
# Permission is a hard constraint and is visible to the policy.
permitted=(r.random(n)>.08)&(busy<.91)
period=np.arange(n)>=n//2
pref=np.array([.25, 1.25, .5, 2.55])[cat].astype(float)
if condition=='drift':
post=np.array([.25, -.15, 1.95, 2.6])[cat]
pref=np.where(period,post,pref)
if condition in ('personal','personal_drift'):
offsets=r.normal(0,1.65,4)
pref=pref+offsets[cat]
if condition=='personal_drift':
next_offsets=r.normal(0,1.65,4)
pref=np.where(period,np.array([.25,1.25,.5,2.55])[cat]+next_offsets[cat],pref)
p_actual=confidence.copy()
if condition=='miscalibrated':
p_actual=np.where((cat==1)&period,1-confidence,confidence)
truth=(r.random(n)<p_actual)
benefit=(.85+1.55*goal+1.55*deadline)*(.72+pref)
penalty=.62 + 1.3*busy
if condition=='drift': penalty=penalty+period*.65*busy
true_value=truth*benefit-(~truth)*2.0-penalty + r.normal(0,.22,n)
x=np.stack([features(confidence[i],goal[i],deadline[i],busy[i],int(cat[i])) for i in range(n)])
# Responses are selectively missing and are delayed. Silence yields no response.
response_probability=np.clip(.34+.36*(true_value>0)-.28*busy,.05,.93)
if condition=='delayed':response_probability*=.52
response=r.random(n)<response_probability
feedback_lag=r.integers(0,28,size=n)
audit_lag=r.integers(4,35,size=n)
if condition=='delayed':
feedback_lag+=r.integers(40,115,size=n)
audit_lag+=r.integers(45,130,size=n)
audit_random=r.random(n)<AUDIT_PROB
if condition=='biased_audit':
# Outcome audit is NOT missing at random: rarely samples unpopular or low-confidence suggestions.
audit_random=(r.random(n)<(AUDIT_PROB*(.15+.85*(confidence>.65))))
audit_label=true_value+r.normal(0,.95,size=n)
# This unrealistically strong label is an explicit assumption. It is not raw implicit engagement.
return {'audit_opt_in':True,'x':x,'cat':cat,'deadline':deadline,'busy':busy,'confidence':confidence,
'permitted':permitted,'true_value':true_value,'response':response,
'feedback_lag':feedback_lag,'audit_lag':audit_lag,'audit_random':audit_random,
'audit_label':audit_label,'truth':truth,'period':period,'condition':condition}
def initial():
w=np.zeros(N_FEATURES)
w[0]=-1.7;w[1]=3.0;w[2]=1.4;w[3]=1.15;w[4]=-1.0
return w
def update(w, prior, x, score, rate):
prediction=np.clip(np.dot(w,x),-6,14)
# Stable normalized online LMS. Do not train on silent events unless audit supplied an outcome.
w += rate * np.clip(score-prediction,-12,12)*x/(.75+np.dot(x,x))
w += .0005*(prior-w)
def simulate(world, method, seed):
random=np.random.default_rng(seed)
n=len(world['cat']); prior=initial();w=prior.copy()
due=defaultdict(list)
alerts=np.zeros(n,dtype=bool)
audits=np.zeros(n,dtype=bool)
total_feedback=0;total_audits=0;missing_feedback=0
score_history=deque(maxlen=500); paced_cutoff=.55; cat_counts=np.zeros(4,dtype=np.int64)
for i in range(n):
for x,score,kind in due.pop(i,[]):
update(w,prior,x,score,rate=(.29 if kind=='audit' else .19))
if kind=='audit':total_audits+=1
else:total_feedback+=1
allowed=bool(world['permitted'][i]) and (int(alerts[max(i//BLOCK*BLOCK,0):i].sum())<CAP)
x=world['x'][i]
if method=='silent': decision=False
elif method=='oracle':decision=world['true_value'][i]>0
elif method=='fixed':decision=(world['confidence'][i]>.74 and world['busy'][i]<.55)
else:
score=float(w@x)
# No trained model may access the simulated real condition or unobserved actual reward.
if method in ('audit_paced','feedback_paced','prior_paced') and i%30==0 and len(score_history)>=100:
recent=np.fromiter(score_history,dtype=np.float64)
k=int(.90*(len(recent)-1))
paced_cutoff=max(.55,float(np.partition(recent,k)[k]))
decision=score>(paced_cutoff if method in ('audit_paced','feedback_paced','prior_paced') else .55)
score_history.append(score)
if method=='bandit_explore':decision=decision or random.random()<.025
elif method=='audit_plus_probe':decision=decision or random.random()<.008
decision=bool(decision and allowed)
alerts[i]=decision
if decision:
if world['response'][i] and method!='prior_paced':
due[i+int(world['feedback_lag'][i])+1].append((x,world['true_value'][i], 'feedback'))
else: missing_feedback+=1
should_audit=(method in ('audit_only','audit_paced','audit_plus_probe') and world['audit_random'][i])
if method=='audit_stratified':
frequency=(cat_counts[int(world['cat'][i])]+1)/(i+4)
audit_prob=min(.25,AUDIT_PROB*math.sqrt(.25/max(frequency,.005)))
should_audit=(random.random()<audit_prob)
cat_counts[int(world['cat'][i])]+=1
if should_audit and world['audit_opt_in']:
audits[i]=True
due[i+int(world['audit_lag'][i])+1].append((x,world['audit_label'][i],'audit'))
if method=='naive_silence_negative' and not decision:
# Biased baseline: incorrectly encodes absence of a notification as a negative outcome.
update(w,prior,x,-1.0,rate=.08)
reward=world['true_value']; val=np.where(world['permitted'],reward,0.)
positives=(val>0); allowed=world['permitted']
tp=int((alerts&positives).sum());fp=int((alerts&~positives).sum())
eligible_positive=int(positives.sum()); al=int(alerts.sum())
rare=(world['cat']==3)&positives
last=np.arange(n)>=3*n//4
return {'method':method,'world':seed//100,'condition':world['condition'],'n':n,
'alerts':al,'true_alerts':tp,'false_alerts':fp,'positive':eligible_positive,
'rare_positive':int(rare.sum()),'rare_true_alerts':int((alerts&rare).sum()),
'net_utility':float(reward[alerts].sum()-AUDIT_COST*audits.sum()),
'audit_count':int(audits.sum()),'feedback_count':total_feedback,
'feedback_missing':missing_feedback,
'late_utility':float(reward[alerts&last].sum()-AUDIT_COST*audits[last].sum()),
'late_alerts':int((alerts&last).sum()),
'late_true_alerts':int((alerts&last&positives).sum()),
'late_positive':int((last&positives).sum()),
'late_false_alerts':int((alerts&last&~positives).sum()),
'permission_violations':int((alerts&~allowed).sum()),
'peak_block_alerts':int(max(alerts[i:i+BLOCK].sum() for i in range(0,n,BLOCK)))}
def aggregate(rows):
ans={}
for cond in sorted(set(r['condition'] for r in rows)):
ans[cond]={}
for method in METHODS:
rr=[r for r in rows if r['method']==method and r['condition']==cond]
if not rr:continue
s=lambda k:sum(r[k] for r in rr)
ans[cond][method]={
'worlds':len(rr),'situations':s('n'),
'recall_pct':round(100*s('true_alerts')/max(s('positive'),1),2),
'precision_pct':round(100*s('true_alerts')/max(s('alerts'),1),2) if s('alerts') else None,
'false_per_1000':round(1000*s('false_alerts')/s('n'),2),
'alerts_per_1000':round(1000*s('alerts')/s('n'),2),
'rare_recall_pct':round(100*s('rare_true_alerts')/max(1,s('rare_positive')),2),
'net_per_1000':round(1000*s('net_utility')/s('n'),2),
'late_recall_pct':round(100*s('late_true_alerts')/max(s('late_positive'),1),2),
'late_net_per_1000':round(1000*s('late_utility')/(len(rr)*rr[0]['n']/4),2),
'audits_per_1000':round(1000*s('audit_count')/s('n'),2),
'feedback_count':s('feedback_count'),
'permission_violations':s('permission_violations'),
'max_alerts_in_100':max(r['peak_block_alerts'] for r in rr)
}
return ans
def identifiability_check():
# Observed logs: learner's always-silent policy sees the same context, action, feedback
# in two possible worlds, while the counterfactual reward for interrupting differs.
log_a=[('x',0,None)]*100; log_b=[('x',0,None)]*100
counterfactual_a=[7.]*100;counterfactual_b=[-7.]*100
assert log_a==log_b and sum(counterfactual_a)!=sum(counterfactual_b)
return {'identical_observed_logs':True,'unobserved_counterfactual_a':7,
'unobserved_counterfactual_b':-7,
'consequence':'No method using only these logs can identify the unobserved utility of interrupting.'}
def run(stage, worlds, n, output):
records=[]
conditions=('ordinary','drift','delayed','rare','miscalibrated','biased_audit','personal','personal_drift')
for condition_idx,cond in enumerate(conditions):
for wi in range(worlds):
worldseed=(100000 if stage=='development' else 900000)+condition_idx*10000+wi*100
world=make_world(worldseed,n,cond)
for method_idx, method in enumerate(METHODS):
records.append(simulate(world,method,worldseed+method_idx+1))
print(f'completed {stage} {cond}', flush=True)
agg=aggregate(records)
output.mkdir(parents=True,exist_ok=True)
with (output/f'sentinel-interruption-{stage}-worlds.csv').open('w',newline='') as f:
writer=csv.DictWriter(f,fieldnames=records[0].keys());writer.writeheader();writer.writerows(records)
with (output/f'sentinel-interruption-{stage}-summary.json').open('w') as f:
json.dump({'stage':stage,'worlds_per_condition':worlds,'events_per_world':n,
'audit_assumption':'3.5% independent opt-in retrospective outcome score, noisy and delayed; unavailable by default in real world',
'identifiability':identifiability_check(),'metrics':agg},f,indent=2)
for cond,values in agg.items():
print('\n'+cond)
for method,m in values.items():
print(f" {method:24} recall {m['recall_pct']:6.2f} prec {str(m['precision_pct']):>6} false/1000 {m['false_per_1000']:6.2f} util/1000 {m['net_per_1000']:7.2f} late {m['late_net_per_1000']:7.2f} rare {m['rare_recall_pct']:6.2f}")
return agg
if __name__=='__main__':
parser=argparse.ArgumentParser();parser.add_argument('--stage',choices=['development','heldout'],default='development')
parser.add_argument('--worlds',type=int,default=8);parser.add_argument('--events',type=int,default=2500)
a=parser.parse_args();run(a.stage,a.worlds,a.events,ROOT)