| """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) |
|
|
