Spaces:
Runtime error
Runtime error
File size: 4,415 Bytes
8018595 cc034ee 8018595 |
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 |
"""FastAPI application exposing cancer risk assessment endpoints."""
from pathlib import Path
from fastapi import FastAPI, HTTPException
from sentinel.config import AppConfig, ModelConfig, ResourcePaths
from sentinel.factory import SentinelFactory
from sentinel.models import InitialAssessment
from sentinel.user_input import UserInput
app = FastAPI(
title="Cancer Risk Assessment Assistant",
description="API for assessing cancer risks using LLMs.",
)
# Define base paths relative to the project root
BASE_DIR = Path(__file__).resolve().parents[2] # Go up to project root
CONFIGS_DIR = BASE_DIR / "configs"
PROMPTS_DIR = BASE_DIR / "prompts"
def create_knowledge_base_paths() -> ResourcePaths:
"""Build resource path configuration resolved from the repository root.
Returns:
ResourcePaths: Paths pointing to persona, prompt, and configuration
assets required by the API routes.
"""
return ResourcePaths(
persona=PROMPTS_DIR / "persona" / "default.md",
instruction_assessment=PROMPTS_DIR / "instruction" / "assessment.md",
instruction_conversation=PROMPTS_DIR / "instruction" / "conversation.md",
output_format_assessment=CONFIGS_DIR / "output_format" / "assessment.yaml",
output_format_conversation=CONFIGS_DIR / "output_format" / "conversation.yaml",
cancer_modules_dir=CONFIGS_DIR / "knowledge_base" / "cancer_modules",
dx_protocols_dir=CONFIGS_DIR / "knowledge_base" / "dx_protocols",
)
@app.get("/")
async def read_root() -> dict:
"""Return a simple greeting message.
Returns:
dict: A dictionary containing a greeting message.
"""
return {"message": "Hello, world!"}
@app.post("/assess/{provider}", response_model=InitialAssessment)
async def assess(
provider: str,
user_input: UserInput,
model: str | None = None,
cancer_modules: list[str] | None = None,
dx_protocols: list[str] | None = None,
) -> InitialAssessment:
"""Assess cancer risk for a user.
Args:
provider (str): LLM provider identifier (for example ``"openai"`` or
``"anthropic"``).
user_input (UserInput): Structured demographics and clinical
information supplied by the client.
model (str | None): Optional model name overriding the provider
default.
cancer_modules (list[str] | None): Optional list of cancer module slugs
to include in the knowledge base.
dx_protocols (list[str] | None): Optional list of diagnostic protocol
slugs to include.
Returns:
InitialAssessment: Parsed model output describing the initial
assessment.
Raises:
HTTPException: 400 for invalid input, 500 for unexpected errors.
"""
try:
# Create knowledge base paths
knowledge_base_paths = create_knowledge_base_paths()
# Set default model name if not provided
if model is None:
model_defaults = {
"openai": "gpt-4o-mini",
"anthropic": "claude-3-5-sonnet-20241022",
"google": "gemini-1.5-pro",
}
model = model_defaults.get(provider, "gpt-4o-mini")
# Set default modules if not provided
if cancer_modules is None:
cancer_modules_dir = knowledge_base_paths.cancer_modules_dir
cancer_modules = [p.stem for p in cancer_modules_dir.glob("*.yaml")]
if dx_protocols is None:
dx_protocols_dir = knowledge_base_paths.dx_protocols_dir
dx_protocols = [p.stem for p in dx_protocols_dir.glob("*.yaml")]
# Create AppConfig
app_config = AppConfig(
model=ModelConfig(provider=provider, model_name=model),
knowledge_base_paths=knowledge_base_paths,
selected_cancer_modules=cancer_modules,
selected_dx_protocols=dx_protocols,
)
# Create factory and conversation manager
factory = SentinelFactory(app_config)
conversation_manager = factory.create_conversation_manager()
# Run assessment
response = conversation_manager.initial_assessment(user_input)
return response
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal Server Error: {e!s}")
|