File size: 7,617 Bytes
1d2de8a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """Deterministic source references and explicit evidence interventions.
Offsets count Python Unicode code points, never bytes. Candidate ranking is lexical
and is labelled as such; only an independent model callback supplies support scores.
Interventions re-encode edited source; an existing KV state cannot prove removal.
"""
import hashlib
import re
from pathlib import Path
def digest(text):
return hashlib.sha256(text.encode()).hexdigest()
def passages(text, max_chars=1200):
if not isinstance(text,str) or max_chars < 1:
raise ValueError('text and positive passage size required')
result=[]
# Cover every character, including whitespace; splitting does not normalize text.
start=0
while start < len(text):
limit=min(len(text),start+max_chars)
end=limit
if limit < len(text):
candidates=[m.end() for m in re.finditer(r'\n\s*\n|(?<=[.!?])\s+',text[start:limit])]
if candidates and candidates[-1] >= max_chars//2:end=start+candidates[-1]
result.append({'id':f'text:{start}:{end}','kind':'text','start':start,'end':end,
'text':text[start:end],'source_sha256':digest(text)})
start=end
return result
def image_pages(paths):
return [{'id':f'page:{i+1}','kind':'image','page':i+1,'path':str(p),
'source_sha256':hashlib.sha256(Path(p).read_bytes()).hexdigest()} for i,p in enumerate(paths)]
def validate_spans(text, spans):
ordered=sorted(spans,key=lambda s:(s['start'],s['end']))
last=0
for span in ordered:
start,end=span['start'],span['end']
if type(start) is not int or type(end) is not int or not 0<=start<end<=len(text) or start<last:
raise ValueError('invalid or overlapping evidence range')
if span.get('source_sha256',digest(text))!=digest(text) or span.get('text',text[start:end])!=text[start:end]:
raise ValueError('evidence no longer matches source')
last=end
return ordered
def rank_candidates(question, candidates, limit=8):
if limit<1:raise ValueError('positive candidate limit required')
tokens=set(re.findall(r'\w+',question.casefold()))
ranked=[]
for c in candidates:
words=set(re.findall(r'\w+',c.get('text','').casefold()))
ranked.append({**c,'candidate_score':len(words&tokens)/max(1,len(tokens)),
'candidate_method':'lexical_overlap' if c['kind']=='text' else 'page_order'})
# Stable ties retain source order (page 2 must precede page 10).
return sorted(ranked,key=lambda c:-c['candidate_score'])[:limit]
def select(question, candidates, support_score, limit=8, minimum_support=.5):
"""Callback receives only question+candidate, never benchmark gold."""
if not 0<=minimum_support<=1:raise ValueError('invalid support threshold')
results=[]
for candidate in rank_candidates(question,candidates,limit):
score=float(support_score(question,candidate))
if not 0<=score<=1:raise ValueError('finite support probability required')
results.append({**candidate,'support_score':score})
chosen=[r for r in results if r['support_score']>=minimum_support]
return {'evidence':chosen,'candidates':results,'verification':'model_support_only',
'faithfulness_established':False,'no_support_found':not chosen}
def intervene(text, spans, question, decide):
"""decide(document,question) must prefill each supplied document afresh.
All three actual calls are returned; agreement/disagreement is evidence, not a
guarantee that a passage is the unique cause of an answer.
"""
spans=validate_spans(text,spans)
evidence='\n\n'.join(text[s['start']:s['end']] for s in spans)
pieces=[];start=0
for span in spans:
pieces.append(text[start:span['start']]);start=span['end']
pieces.append(text[start:]);removed=''.join(pieces)
full=decide(text,question)
only=decide(evidence,question)
removal=decide(removed,question)
return {'full':full,'evidence_only':only,'evidence_removed':removal,
'evidence_only_agrees':only==full,'removal_changes_answer':removal!=full,
'verification':'fresh_source_reencoding','source_sha256':digest(text),
'evidence_sha256':digest(evidence),'removed_sha256':digest(removed),
'calls':3,'input_characters':len(text)+len(evidence)+len(removed)}
def validate_pages(pages, selected):
"""Validate ordered page manifests and selected immutable references."""
indexed = {}
for expected, page in enumerate(pages, 1):
if page.get('kind') != 'image' or type(page.get('page')) is not int or page['page'] != expected:
raise ValueError('image manifest must use consecutive source page IDs')
if page.get('id') != f'page:{expected}':
raise ValueError('image page ID does not match source position')
actual = hashlib.sha256(Path(page['path']).read_bytes()).hexdigest()
if page.get('source_sha256') != actual:
raise ValueError('image no longer matches source')
indexed[expected] = page
ids = []
for reference in selected:
number = reference.get('page')
if type(number) is not int or number not in indexed or number in ids:
raise ValueError('invalid or duplicate evidence page')
source = indexed[number]
if any(reference.get(key) != source[key] for key in ('id', 'kind', 'path', 'source_sha256')):
raise ValueError('evidence page no longer matches source manifest')
ids.append(number)
return [indexed[number] for number in sorted(ids)]
def intervene_pages(pages, selected, question, decide, text=''):
"""Re-encode full, evidence-only and page-removed multimodal documents.
decide(text, image_paths, question) must create a fresh state each time and
accept an empty image list. Evidence-only has no accompanying source text;
removal retains all source text and unselected pages. This isolates page
evidence and makes text-only sufficiency a visible competing explanation.
Original page IDs are recorded because subset images are renumbered on input.
"""
if not isinstance(text, str):
raise ValueError('source text must be a string')
selected = validate_pages(pages, selected)
selected_ids = {p['page'] for p in selected}
removed = [p for p in pages if p['page'] not in selected_ids]
calls = [(text, pages), ('', selected), (text, removed)]
answers = []
for source_text, source_pages in calls:
# Recheck all originals before each call: never silently mix revisions.
validate_pages(pages, selected)
answers.append(decide(source_text, [p['path'] for p in source_pages], question))
validate_pages(pages, selected)
full, only, removal = answers
return {'full': full, 'evidence_only': only, 'evidence_removed': removal,
'evidence_only_agrees': only == full, 'removal_changes_answer': removal != full,
'verification': 'fresh_source_reencoding', 'faithfulness_established': False,
'source_text_sha256': digest(text),
'source_pages': [{k: p[k] for k in ('id', 'page', 'source_sha256')} for p in pages],
'evidence_page_ids': [p['page'] for p in selected],
'removed_input_page_ids': [p['page'] for p in removed],
'calls': 3, 'input_images': sum(len(p) for _, p in calls),
'input_image_bytes': sum(Path(p['path']).stat().st_size for _, ps in calls for p in ps),
'input_characters': 2 * len(text)}
|