| """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=[] |
| |
| 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'}) |
| |
| 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: |
| |
| 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)} |
|
|
