Spaces:
Sleeping
Sleeping
File size: 5,592 Bytes
73c6377 4fd13fd 73c6377 4fd13fd 73c6377 2a9dafb 73c6377 2a9dafb 73c6377 2a9dafb 73c6377 4fd13fd 2a9dafb 73c6377 22770b7 73c6377 4fd13fd 73c6377 |
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 |
"""
HBV Patient Assessment Router
API endpoint for HBV treatment eligibility assessment
"""
from fastapi import APIRouter, HTTPException
from api.models import HBVPatientInput, HBVAssessmentResponse, TextAssessmentInput
import logging
from contextlib import contextmanager
from typing import List
from core.assessment_chain import (
run_assessment_chain,
run_assessment_chain_from_prompt,
build_prompt_from_raw_text,
)
logger = logging.getLogger(__name__)
VERBOSE_LOGGER_NAMES = [
"",
"api",
"api.app",
"api.middleware",
"api.routers",
"api.routers.hbv_assessment",
"core",
"core.assessment_chain",
"uvicorn",
"uvicorn.error",
"uvicorn.access",
]
TEXT_ASSESS_VERBOSE = True
@contextmanager
def temporary_verbose_logging(enabled: bool):
"""
Temporarily raise logging levels to DEBUG so the request pipeline is visible.
"""
if not enabled:
yield
return
target_loggers: List[logging.Logger] = [logging.getLogger(name) for name in VERBOSE_LOGGER_NAMES]
original_levels = [logger.level for logger in target_loggers]
for log_obj in target_loggers:
log_obj.setLevel(logging.DEBUG)
logger.info("Verbose logging enabled for this request")
try:
yield
finally:
for log_obj, level in zip(target_loggers, original_levels):
log_obj.setLevel(level)
router = APIRouter(
prefix="",
tags=["HBV Assessment"]
)
@router.post("/assess", response_model=HBVAssessmentResponse)
async def assess_patient(patient: HBVPatientInput) -> HBVAssessmentResponse:
"""
Assess HBV patient eligibility for treatment according to SASLT 2021 guidelines
This endpoint uses a LangChain Chain with hybrid approach:
1. Phase 1 (Deterministic): Validates patient data, computes eligibility deterministically
2. Phase 2 (LLM): Generates narrative recommendations with citations
3. Returns structured assessment with eligibility and comprehensive recommendations
Returns:
HBVAssessmentResponse containing:
- eligible: Whether patient is eligible for treatment
- recommendations: Comprehensive narrative including eligibility determination,
specific criteria met, treatment options (ETV, TDF, TAF), and special considerations,
with inline citations in format [SASLT 2021, Page X]
"""
try:
logger.info(f"Assessing HBV patient: Age {patient.age}, Sex {patient.sex}, HBV DNA {patient.hbv_dna_level}")
# Convert Pydantic model to dict for chain
patient_data = patient.dict()
# Run assessment chain (hybrid: deterministic + LLM)
result = run_assessment_chain(patient_data)
# Convert dict result back to Pydantic response model
response = HBVAssessmentResponse(**result)
logger.info(f"Assessment complete: Eligible={response.eligible}")
return response
except Exception as e:
logger.error(f"Error assessing patient: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error assessing patient: {str(e)}")
@router.post("/assess/text", response_model=HBVAssessmentResponse)
async def assess_patient_from_text(text_input: TextAssessmentInput) -> HBVAssessmentResponse:
"""
Assess HBV patient eligibility from free-form text input
This endpoint:
1. Uses the free-form text directly as the LLM prompt (no intermediate structuring)
2. Starts the assessment chain from the LLM invocation step
3. Runs the same Phase 2 post-processing as the structured endpoint
4. Returns structured assessment with eligibility and recommendations
Example text input:
"45-year-old male patient
HBsAg: Positive for 12 months
HBV DNA: 5000 IU/mL
HBeAg: Positive
ALT: 80 U/L
Fibrosis stage: F2
Necroinflammatory activity: A2
No extrahepatic manifestations
No immunosuppression
No coinfections (HIV, HCV, HDV)
No family history of cirrhosis or HCC"
Returns:
HBVAssessmentResponse containing:
- eligible: Whether patient is eligible for treatment
- recommendations: Comprehensive narrative with inline citations
"""
try:
raw_text = text_input.text_input
logger.info(f"Received text input for assessment (length: {len(raw_text)} characters)")
logger.info(f"Text input preview: {raw_text[:200]}...")
# Enable verbose logging for observability when configured.
with temporary_verbose_logging(TEXT_ASSESS_VERBOSE):
prompt = build_prompt_from_raw_text(raw_text)
# Run assessment chain starting from the LLM invocation step,
# using the synthesized prompt that forces JSON output.
logger.info("Performing HBV eligibility assessment from raw text prompt using LangChain chain...")
result = run_assessment_chain_from_prompt(prompt)
# Convert dict result to Pydantic response model
response = HBVAssessmentResponse(**result)
logger.info(f"Text-based assessment complete: Eligible={response.eligible}")
return response
except ValueError as e:
logger.error(f"Validation error in text assessment: {str(e)}")
raise HTTPException(status_code=400, detail=f"Invalid patient data: {str(e)}")
except Exception as e:
logger.error(f"Error in text-based assessment: {str(e)}")
raise HTTPException(status_code=500, detail=f"Error processing text input: {str(e)}")
|