from __future__ import annotations from dataclasses import asdict, dataclass from typing import Any from app.config import config from app.llm import DefaultModelSelectionPolicy, LLMTask from app.services.context_budgeting import deduplicate_texts, limit_items from app.services.soat_processing import ( _build_validation_structured_prompt, _coerce_validation_evidence_items, ) from modules.processing.resumen_google import _prepare_historia_prompt_text @dataclass(frozen=True, slots=True) class Phase8BenchmarkSnapshot: task: str provider: str model: str baseline_chars: int optimized_chars: int reduced_chars: int reduction_pct: float note: str def _build_snapshot(*, task: LLMTask, baseline: str, optimized: str, note: str) -> Phase8BenchmarkSnapshot: route = DefaultModelSelectionPolicy().resolve(task) baseline_chars = len(str(baseline or "")) optimized_chars = len(str(optimized or "")) reduced_chars = max(0, baseline_chars - optimized_chars) reduction_pct = round((reduced_chars / baseline_chars) * 100, 2) if baseline_chars else 0.0 return Phase8BenchmarkSnapshot( task=task.value, provider=route.provider, model=route.model, baseline_chars=baseline_chars, optimized_chars=optimized_chars, reduced_chars=reduced_chars, reduction_pct=reduction_pct, note=note, ) def _historia_sample_text() -> str: lines = [ "Nombre del paciente: ANA MARIA LOPEZ", "Fecha de ingreso: 2026-05-20", "Motivo de consulta: trauma en miembro inferior izquierdo", "Procedimientos realizados", "123456 Lavado quirúrgico", "Medicamentos administrados", "Dipirona 1 g IV cada 8 horas", ] repeated = ["Evolución estable sin cambios clínicos relevantes." for _ in range(30)] return "\n".join(lines + repeated) def _build_historia_benchmark() -> list[Phase8BenchmarkSnapshot]: raw_text = _historia_sample_text() prompt_text = _prepare_historia_prompt_text( raw_text, summary_threshold=10**9, summary_chunk=config.HISTORIA_SUMMARY_CHUNK, max_chars=config.HISTORIA_MAX_CHARS, llm_router=None, cache_repository=None, username="", ) note = "Historia compactada sin resumen LLM; se mide reducción por deduplicación y budgeting." return [ _build_snapshot( task=LLMTask.HISTORIA_CHUNK_SUMMARY, baseline=raw_text, optimized=prompt_text, note=note, ), _build_snapshot( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, baseline=raw_text, optimized=prompt_text, note="Extracción clínica usa el texto ya compactado cuando la historia entra por gating.", ), ] def _build_soat_benchmark() -> Phase8BenchmarkSnapshot: base_extract = ( "Procedimiento SOAT relevante con referencia normativa y detalle clínico suficiente para auditoría " "y correlación con soporte quirúrgico. " ) * 12 raw_evidence = [ {"page": 12 + (idx % 3), "extracto": f"{base_extract}{idx}", "score": 0.95 - (idx * 0.01)} for idx in range(12) ] optimized_evidence = _coerce_validation_evidence_items(raw_evidence) raw_cups = [ {"codigo_cups": f"{120000 + idx}", "procedimiento": f"Procedimiento candidato {idx}", "score": 0.99 - idx * 0.01} for idx in range(12) ] optimized_cups, _ = limit_items(raw_cups, max_items=config.SOAT_PROMPT_MAX_CUPS) raw_cie10 = [ {"codigo": f"S{820 + idx}.0", "descripcion": f"Diagnóstico candidato {idx}"} for idx in range(12) ] optimized_cie10, _ = limit_items(raw_cie10, max_items=config.SOAT_PROMPT_MAX_CIE10) baseline_prompt = _build_validation_structured_prompt( codigo_soat="S1", descripcion_soat="Lavado quirúrgico por trauma abierto", diagnosticos_prequx="Fractura expuesta de tibia derecha", procedimientos="Lavado quirúrgico y osteosíntesis", hallazgos="Compromiso de tejidos blandos y contaminación moderada", descripcion_proc="Se realiza lavado, desbridamiento y fijación interna", justificacion="Control de daño y prevención de infección", diagnosticos_postqx="Fractura estabilizada", evidencia_soat=raw_evidence, propuestas_cups=raw_cups, propuestas_cie10=raw_cie10, contexto_cups="\n".join(f"CÓDIGO {item['codigo_cups']}: {item['procedimiento']}" for item in raw_cups), contexto_cie10="\n".join(f"CÓDIGO {item['codigo']}: {item['descripcion']}" for item in raw_cie10), ) optimized_prompt = _build_validation_structured_prompt( codigo_soat="S1", descripcion_soat="Lavado quirúrgico por trauma abierto", diagnosticos_prequx="Fractura expuesta de tibia derecha", procedimientos="Lavado quirúrgico y osteosíntesis", hallazgos="Compromiso de tejidos blandos y contaminación moderada", descripcion_proc="Se realiza lavado, desbridamiento y fijación interna", justificacion="Control de daño y prevención de infección", diagnosticos_postqx="Fractura estabilizada", evidencia_soat=optimized_evidence, propuestas_cups=optimized_cups, propuestas_cie10=optimized_cie10, contexto_cups="\n".join( f"CÓDIGO {item['codigo_cups']}: {item['procedimiento']}" for item in optimized_cups ), contexto_cie10="\n".join( f"CÓDIGO {item['codigo']}: {item['descripcion']}" for item in optimized_cie10 ), ) return _build_snapshot( task=LLMTask.SOAT_REASONING, baseline=baseline_prompt, optimized=optimized_prompt, note="SOAT reasoning limita evidencia, extractos y candidatos CIE-10/CUPS antes del prompt.", ) def _build_resolution_benchmark() -> list[Phase8BenchmarkSnapshot]: diagnosticos_raw = [ "S82.2 - Fractura de tibia", "S82.2 - Fractura de tibia", "T14.9 - Trauma no especificado", "Trauma no especificado", ] + [f"Diagnóstico clínico adicional {idx}" for idx in range(20)] procedimientos_raw = [ "123456 Lavado quirúrgico", "123456 Lavado quirúrgico", "654321 Curación avanzada", ] + [f"Procedimiento candidato {idx}" for idx in range(20)] diagnosticos_dedup, _ = deduplicate_texts(diagnosticos_raw) procedimientos_dedup, _ = deduplicate_texts(procedimientos_raw) diagnosticos_budgeted, _ = limit_items(diagnosticos_dedup, max_items=config.CIE10_RESOLUTION_K) procedimientos_budgeted, _ = limit_items(procedimientos_dedup, max_items=config.CIE10_RESOLUTION_K) baseline_dx = "\n".join(diagnosticos_raw) optimized_dx = "\n".join(diagnosticos_budgeted) baseline_px = "\n".join(procedimientos_raw) optimized_px = "\n".join(procedimientos_budgeted) return [ _build_snapshot( task=LLMTask.CIE10_RESOLUTION, baseline=baseline_dx, optimized=optimized_dx, note="Resolución CIE-10 deduplica diagnósticos y aplica límite configurable de entradas.", ), _build_snapshot( task=LLMTask.CUPS_RESOLUTION, baseline=baseline_px, optimized=optimized_px, note="Resolución CUPS evita procedimientos duplicados y reduce el lote de entrada.", ), ] def build_phase8_benchmark_report() -> dict[str, Any]: snapshots = [ *_build_historia_benchmark(), _build_soat_benchmark(), *_build_resolution_benchmark(), ] return { "schema_version": "v1", "total_tasks": len(snapshots), "snapshots": [asdict(snapshot) for snapshot in snapshots], } def render_phase8_benchmark_markdown(report: dict[str, Any]) -> str: lines = [ "# Phase 8 LLM Benchmark", "", "| Task | Provider | Model | Baseline chars | Optimized chars | Reduced chars | Reduction % |", "| --- | --- | --- | ---: | ---: | ---: | ---: |", ] for item in report.get("snapshots", []): lines.append( "| {task} | {provider} | {model} | {baseline_chars} | {optimized_chars} | {reduced_chars} | " "{reduction_pct}% |".format(**item) ) lines.extend(["", "## Notes", ""]) for item in report.get("snapshots", []): lines.append(f"- `{item['task']}`: {item['note']}") return "\n".join(lines)