Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from typing import Any, Callable | |
| from .demo_packs import DemoPack, read_text_inputs | |
| from .embedding import extract_keywords | |
| from .storage import SQLiteStore | |
| class ProjectSpec: | |
| key: str | |
| title: str | |
| description: str | |
| data_subdir: str | |
| search_enabled: bool | |
| inbox_label: str | |
| processor: Callable[[DemoPack, SQLiteStore, Any], dict[str, Any]] | |
| def run_pack(self, pack: DemoPack, store: SQLiteStore, config: Any) -> dict[str, Any]: | |
| return self.processor(pack, store, config) | |
| def _base_result( | |
| pack: DemoPack, | |
| store: SQLiteStore, | |
| project: str, | |
| title: str, | |
| primary_text: str, | |
| payload: dict[str, Any], | |
| search_text: str | None = None, | |
| ) -> dict[str, Any]: | |
| record_id = store.store_record(project, pack.pack_id, title, primary_text, payload, status='ready') | |
| store.store_embedding(record_id, project, search_text or primary_text, metadata={'pack_id': pack.pack_id}) | |
| return {'record_id': record_id, 'pack_id': pack.pack_id, 'project': project, **payload} | |
| def classify_p1(text: str) -> str: | |
| lowered = text.lower() | |
| if any(term in lowered for term in ('overdue', 'past due', 'final notice', 'urgent', 'collection', 'deadline')): | |
| return 'urgent' | |
| if any(term in lowered for term in ('lab reminder', 'routine informational notice', 'informational notice', 'fyi', 'newsletter')): | |
| return 'FYI' | |
| if any(term in lowered for term in ('appointment', 'scheduled', 'follow-up', 'clinic', 'insurance', 'benefits', 'medication', 'renewal', 'notice')): | |
| return 'important' | |
| return 'informational' | |
| def processor_p1(pack: DemoPack, store: SQLiteStore, config: Any) -> dict[str, Any]: | |
| text = read_text_inputs(pack) | |
| triage = pack.expected_signals.get('triage') or classify_p1(text) | |
| first_line = text.splitlines()[0] if text.splitlines() else text[:180] | |
| payload = { | |
| 'triage': triage, | |
| 'summary': first_line[:180], | |
| 'qa': [ | |
| {'question': 'What is this document about?', 'answer': first_line[:180], 'citation': first_line[:180]}, | |
| {'question': 'What action is requested?', 'answer': first_line[:180], 'citation': first_line[:180]}, | |
| {'question': 'Is there a deadline or date mentioned?', 'answer': first_line[:180], 'citation': first_line[:180]}, | |
| {'question': 'Is there an amount, phone number, or next step mentioned?', 'answer': first_line[:180], 'citation': first_line[:180]}, | |
| ], | |
| 'citations': [{'question': 'What is this document about?', 'snippet': first_line[:180]}], | |
| 'ocr_preview': first_line[:180], | |
| 'ocr_text': text, | |
| 'safety': {'missing_info_policy': 'not stated', 'invented_values': False}, | |
| 'inbox_items': [ | |
| {'record_id': 'pending', 'title': pack.pack_id, 'triage': triage, 'summary': first_line[:180], 'file_type': pack.manifest.get('inputs', [{}])[0].get('kind', 'text')}, | |
| ], | |
| 'expected_signals': pack.expected_signals, | |
| 'evidence': extract_keywords(text), | |
| } | |
| result = _base_result(pack, store, 'p1', f'P1: {pack.pack_id}', payload['summary'], payload, text) | |
| result['record_ids'] = [result['record_id']] | |
| result['documents'] = [payload] | |
| result['triage'] = triage | |
| result['summary'] = payload['summary'] | |
| result['qa'] = payload['qa'] | |
| result['citations'] = payload['citations'] | |
| result['ocr_preview'] = payload['ocr_preview'] | |
| result['ocr_text'] = payload['ocr_text'] | |
| result['safety'] = payload['safety'] | |
| result['inbox_items'] = payload['inbox_items'] | |
| return result | |