File size: 13,181 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 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 | """Evidence packages (v3) in the answer service: per-unit, source-grounded, page-referenced.
Text sources: each question unit (entity / label / boolean / choice) gets an evidence package
(solomon.evidence_packages.build): its selected spans plus automatically attached governing context,
all exact source ranges with code-point offsets, roles and reasons, and -- when a renderer page
map is registered for the source -- rendered page IDs. `sufficiency` re-asks each unit with ONLY
its package as a fresh document; `removal` re-asks the whole question with every selected
evidence span removed. Budgets are checked before any model call. Evidence never promotes answer
qualification and never establishes causal faithfulness.
Image sources are delegated unchanged to solomon.service_evidence.EvidenceService (page selector path).
"""
import copy
from solomon import evidence_packages as v3
from solomon.evidence import digest, validate_spans
from solomon.retrieval import lexical_select, remove
from solomon.service_evidence import EvidenceService, question_units, prediction
CONTRACT = 'solomon-evidence-packages-v3'
SIZE_CAP = v3.SIZE_CAP
def unit_requests(task, request):
"""(question text, subject, restricted decision request) per unit, in answer-vector order."""
base = {k: request[k] for k in ('question', 'options', 'labels', 'template', 'entities') if k in request}
qs = question_units(task, request)
if task == 'entity':
return [(q, e, {**base, 'entities': [e]}) for q, e in zip(qs, request['entities'])]
if task == 'multilabel':
return [(q, None, {**base, 'labels': [label]}) for q, label in zip(qs, request['labels'])]
return [(qs[0], None, base)]
class PackageEvidenceService(EvidenceService):
"""page_maps: optional {source_sha256: rendered page map} from source-bound renderer manifests."""
def __init__(self, answers, decide, selector=None, page_selector=None, *, page_maps=None, size_cap=SIZE_CAP, package_format='per_unit'):
super().__init__(answers, decide, selector, page_selector)
if package_format not in ('per_unit', 'union'):
raise ValueError('package_format must be per_unit or union')
self.package_format = package_format
self.page_maps = dict(page_maps or {})
if not 0 < size_cap <= 1:
raise ValueError('invalid package size cap')
self.size_cap = size_cap
def create(self, document):
result = self.answers.create(document);result['contract'] = CONTRACT;return result
def _unit_evidence(self, text, units, selection):
"""Per-unit selected spans. Selectors reporting `units` keep per-unit attribution;
otherwise every unit receives the selector's (validated) evidence."""
union = validate_spans(text, selection['evidence'])
if 'units' not in selection:
return [union for _ in units]
by = {u['unit']: u['evidence'] for u in selection['units']}
if set(by) - set(range(len(units))):
raise ValueError('selection reports unknown question units')
out = []
for i in range(len(units)):
spans = validate_spans(text, by.get(i, []))
if any(not any(u['start'] <= s['start'] and s['end'] <= u['end'] for u in union) for s in spans):
raise ValueError('unit evidence outside the reported selection')
out.append(spans)
return out
def ask(self, key, task, *, evidence='none', evidence_max_calls=2, **request):
if evidence not in ('none', 'support', 'sufficiency', 'removal'):
raise ValueError('invalid evidence verification level')
if type(evidence_max_calls) is not int or not 0 <= evidence_max_calls <= 512:
raise ValueError('invalid evidence call budget')
with self.answers.lock:
state = self.answers._warm(key)
if any('image' in p for p in state['parts']):
result = super().ask(key, task, evidence=evidence, evidence_max_calls=evidence_max_calls, **request)
result['contract'] = CONTRACT
if evidence != 'none' and isinstance(result.get('evidence'), dict):
result['evidence']['package_format'] = 'pages_only'
return result
result = self.answers.ask(key, task, **request);result['contract'] = CONTRACT
if evidence == 'none':
return result
parts = copy.deepcopy(state['parts']);text = ''.join(p['text'] for p in parts if 'text' in p)
units = unit_requests(task, request);qs = [u[0] for u in units]
only = len(units) if self.package_format == 'per_unit' else 1
needed = {'support': 0, 'sufficiency': only, 'removal': only + 1}[evidence]
selector = self.selector or lexical_select;plan = None;planned = 0
if hasattr(selector, 'plan'):
plan = (selector.plan(text, qs, task=task) if getattr(selector, 'task_aware', False) else selector.plan(text, qs))
if type(plan.get('calls')) is not int or plan['calls'] < 0:
raise ValueError('invalid selection call estimate')
planned = plan['calls']
if needed + planned > evidence_max_calls:
result['evidence'] = {'status': 'budget_exhausted', 'calls': 0, 'required_calls': needed + planned, 'verification': 'none'}
return result
selection = selector.execute(plan) if plan is not None else selector(text, qs)
cost = selection.get('cost', {'calls': 0, 'input_tokens': 0, 'branches': 0})
if any(type(cost.get(k)) is not int or cost[k] < 0 for k in ('calls', 'input_tokens', 'branches')):
raise ValueError('invalid selection static cost')
if cost['calls'] != planned:
raise ValueError('selection exceeded declared inference plan')
per_unit = self._unit_evidence(text, units, selection)
st = v3.Structure(text);page_map = self.page_maps.get(digest(text))
packages = []
for i, ((q, subject, _), spans) in enumerate(zip(units, per_unit)):
p = v3.build(text, q, spans, subject=subject, structure=st, cap=self.size_cap)
if page_map is not None:
p = v3.with_pages(p, text, page_map)
packages.append({'unit': i, 'question': q, 'subject': subject, 'no_support_found': p['no_support_found'],
'spans': p['spans'], 'evidence_ranges': p['evidence_ranges'], 'context_reasons': p['context_reasons'],
'rendered_characters': p['rendered_characters'], 'covered_characters': p['covered_characters'],
'source_fraction': p['source_fraction'], 'within_size_cap': p['within_size_cap'], 'trimmed_for_cap': p['trimmed_for_cap'],
**({'pages': p['pages']} if page_map is not None else {}), '_text': p['text']})
union_refs = []
for s in sorted({(s['start'], s['end']) for spans in per_unit for s in spans}):
if union_refs and s[0] < union_refs[-1][1]:
union_refs[-1] = (union_refs[-1][0], max(s[1], union_refs[-1][1]))
else:
union_refs.append(s)
refs = [{'start': a, 'end': b, 'text': text[a:b], 'source_sha256': digest(text)} for a, b in union_refs]
body = {'status': 'no_support_found' if not refs else 'found', 'references': refs,
'packages': [{k: v for k, v in p.items() if k != '_text'} for p in packages],
'package_format': self.package_format, 'package_schema': v3.SCHEMA, 'size_cap_source_fraction': self.size_cap,
'method': selection.get('method', 'external_selector'), 'verification': selection.get('verification', 'retrieval_only'),
'faithfulness_established': False, 'calls': 0, 'input_tokens': 0, 'branches': 0,
'selection_calls': cost['calls'], 'source_text_sha256': digest(text), 'answer_qualification_unchanged': True,
**({'page_map_registered': True} if page_map is not None else {'page_map_registered': False})}
body['calls'] += cost['calls'];body['input_tokens'] += cost['input_tokens'];body['branches'] += cost['branches']
for field in ('relevance_weights_sha256', 'dependency_rule'):
if field in selection:
body[field] = selection[field]
baseline = prediction(result.get('answer') or result.get('candidate'))
def call(document, req):
output = self.decide(copy.deepcopy(document), task, copy.deepcopy(req))
if not isinstance(output, dict) or not {'answer', 'cost'} <= set(output):
raise ValueError('invalid evidence decision callback')
c = output['cost']
if any(type(c.get(k)) is not int or c[k] < 0 for k in ('input_tokens', 'branches')):
raise ValueError('invalid evidence static cost')
body['calls'] += 1;body['input_tokens'] += c['input_tokens'];body['branches'] += c['branches']
return prediction(output['answer'])
if evidence in ('sufficiency', 'removal') and self.package_format == 'union':
un = v3.union(text, [{'spans': p['spans']} for p in packages])
decision = {k: request[k] for k in ('question', 'options', 'labels', 'template', 'entities') if k in request}
assembled = call([{'text': un['text']}], decision)
body['union_package'] = {'spans': [{k: v for k, v in s.items() if k != 'pages'} for s in un['spans']],
'source_fraction': un['source_fraction'], 'rendered_characters': un['rendered_characters']}
body['evidence_only'] = {'prediction': assembled, 'agrees_with_full': assembled == baseline,
'input': 'union of the question\'s unit packages, one fresh prefill'}
elif evidence in ('sufficiency', 'removal'):
predictions = [call([{'text': p['_text']}], req) for p, (_, _, req) in zip(packages, units)]
if task in ('entity', 'multilabel'):
assembled = [x for pr in predictions for x in (pr if isinstance(pr, list) else [pr])]
else:
assembled = predictions[0]
body['evidence_only'] = {'prediction': assembled, 'agrees_with_full': assembled == baseline,
'per_unit': [{'unit': i, 'prediction': pr} for i, pr in enumerate(predictions)],
'input': 'per-unit evidence package (selected spans + governing context), fresh prefill per unit'}
if evidence == 'removal':
decision = {k: request[k] for k in ('question', 'options', 'labels', 'template', 'entities') if k in request}
removed = call([{'text': remove(text, refs) if refs else text}], decision)
body['evidence_removed'] = {'prediction': removed, 'agrees_with_full': removed == baseline,
'input': 'source with every selected evidence span removed (context kept)'}
if body['calls'] > body['selection_calls']:
body['verification'] = 'fresh_source_reencoding'
result['evidence'] = body
return result
def handler(service):
from solomon.service_evidence import handler as base_handler
base = base_handler(service)
class Handler(base):
def do_GET(self):
if self.path != '/health':
return self.reply(404, {'error': 'not found'})
return self.reply(200, {'contract': CONTRACT, 'answer_types': ['boolean', 'single', 'ordered', 'multilabel', 'entity'],
'evidence_levels': ['none', 'support', 'sufficiency', 'removal'], 'evidence_package': v3.SCHEMA,
'package_format': self.package_format, 'strict_reliability_default': True,
'runtime_identity': self.server.service.runtime_identity if hasattr(self.server, 'service') else service.runtime_identity})
return Handler
def service(store, engine, design, *, artifact=None, views_artifact=None, views_tasks=(), selector=None, page_selector=None,
page_maps=None, size_cap=SIZE_CAP, package_format='per_unit'):
"""Serving constructor mirroring solomon.serving.service, with v3 evidence packages."""
from pathlib import Path
from solomon.serving import ConfidenceHeadsService, FreshDecider
answers = ConfidenceHeadsService(store, engine, design, artifact=artifact, views_artifact=views_artifact, views_tasks=views_tasks)
return PackageEvidenceService(answers, FreshDecider(engine, design, Path(store)/'.evidence-readout'), selector, page_selector,
page_maps=page_maps, size_cap=size_cap, package_format=package_format)
|