Spaces:
Sleeping
Sleeping
File size: 4,966 Bytes
c2b7494 | 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 | from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
import yaml
@dataclass
class PathsConfig:
edf_dir: str = "data/edfs/yamls"
codes_parquet_dir: str = "data/codes/parquet/snomed_definitive_20261030"
embeddings_dir: str = "data/codes/embeddings/snomed_definitive_20261030"
results_dir: str = "results"
aesi_datasets: dict[str, str] = field(default_factory=dict)
code_systems: dict[str, str] = field(
default_factory=lambda: {
"ICD10CM": "data/codes/csv/ICD10CM@2026-codes.csv",
"ICD9CM": "data/codes/csv/ICD9CM@2014-codes.csv",
"ICPC": "data/codes/csv/ICPC@1993-codes.csv",
"MDR": "data/codes/csv/MDR@28.0-codes.csv",
"RCD2": "data/codes/csv/RCD2@20200401-codes.csv",
"SNOMEDCT_US": "data/codes/csv/SNOMEDCT_US@2025_09_01-codes definitive.csv",
}
)
@dataclass
class RetrievalConfig:
lexical_weight: float = 0.1
embedding_weight: float = 0.9
embedding_provider: str = "openai"
embedding_model: str = "text-embedding-3-large"
embedding_dimensions: int | None = 3072
embedding_batch_size: int = 512
top_k: int = 50
use_top_k_limit: bool = False
per_vocabulary_top_k: bool = True
vocabularies: list[str] = field(
default_factory=lambda: ["ICD10CM", "ICD9CM", "ICPC", "MDR", "RCD2", "SNOMEDCT_US"]
)
@dataclass
class LLMConfig:
provider: str = "jev" # Stage 2 defaults to Jev; other providers remain available.
anthropic_model: str = "claude-opus-4-8"
openai_model: str = "gpt-6-luna"
google_model: str = "gemini-3-flash-preview"
jev_model: str = "jev-latest"
openai_decisions_model: str = "gpt-6-luna"
laya_model: str = "multilingual"
jpt_model: str = "kirp/jpt-4b"
mlx_model: str = "mlx-community/Qwen3.5-9B-8bit"
rank_effort: str = "low"
classify_effort: str = "medium"
max_retries: int = 3
use_possible_category: bool = False
adaptive_stopping_enabled: bool = True
sparse_narrow_threshold: int = 0
consecutive_sparse_batches: int = 10
minimum_batches: int = 3
@dataclass
class UncertaintyConfig:
review_provider: str = "openai"
review_model: str | None = None
confidence_threshold: float = 0.7
gpt_review_max_candidates: int = 200
possible_requires_review: bool = False
gpt_review_enabled: bool = True
gpt_review_batch_size: int = 10
gpt_review_reasoning_enabled: bool = True
@dataclass
class EvaluationConfig:
narrow_only_as_positive: bool = True
compare_vocabularies: list[str] = field(
default_factory=lambda: ["ICD10CM", "ICD9CM", "ICPC", "MDR", "RCD2", "SNOMEDCT_US"]
)
@dataclass
class CodeMapperConfig:
enabled: bool = False
base_url: str = "https://app.vac4eu.org/codemapper-testing/api"
cache_dir: str = "results/codemapper/cache"
batch_size: int = 100
requests_per_minute: float = 6
timeout_seconds: float = 30
max_retries: int = 3
vocabulary_versions: dict[str, str] = field(default_factory=lambda: {
"ICD10CM": "2026", "ICD9CM": "2014", "ICPC": "1993",
"MDR": "28.0", "RCD2": "20200401", "SNOMEDCT_US": "2025_09_01",
})
def __post_init__(self):
if self.batch_size < 1 or self.requests_per_minute <= 0:
raise ValueError("CodeMapper batch size and request rate must be positive")
if self.timeout_seconds <= 0 or self.max_retries < 1:
raise ValueError("CodeMapper timeout and retry count must be positive")
if not self.base_url.startswith("https://"):
raise ValueError("CodeMapper requires an HTTPS base URL")
@dataclass
class PipelineConfig:
phenotype: str = "myocarditis"
verbose: bool = True
paths: PathsConfig = field(default_factory=PathsConfig)
retrieval: RetrievalConfig = field(default_factory=RetrievalConfig)
llm: LLMConfig = field(default_factory=LLMConfig)
uncertainty: UncertaintyConfig = field(default_factory=UncertaintyConfig)
evaluation: EvaluationConfig = field(default_factory=EvaluationConfig)
codemapper: CodeMapperConfig = field(default_factory=CodeMapperConfig)
def load_config(path: str | Path) -> PipelineConfig:
"""Load a YAML config file into a PipelineConfig, falling back to defaults
for any keys not present in the file."""
path = Path(path)
raw = yaml.safe_load(path.read_text()) or {}
return PipelineConfig(
phenotype=raw.get("phenotype", PipelineConfig.phenotype),
verbose=raw.get("verbose", PipelineConfig.verbose),
paths=PathsConfig(**raw.get("paths", {})),
retrieval=RetrievalConfig(**raw.get("retrieval", {})),
llm=LLMConfig(**raw.get("llm", {})),
uncertainty=UncertaintyConfig(**raw.get("uncertainty", {})),
evaluation=EvaluationConfig(**raw.get("evaluation", {})),
codemapper=CodeMapperConfig(**raw.get("codemapper", {})),
)
|