刘金宝
183 lines
8.2 KiB
Python
183 lines
8.2 KiB
Python
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.exceptions import AppError
|
|
from app.models.source_case import CaseBase
|
|
from app.repositories.case_repository import CaseRepository
|
|
from app.schemas.training_config import (
|
|
ConfigOption,
|
|
PatientConfig,
|
|
TrainingConfigOptionsResponse,
|
|
TrainingConfigRecommendedResponse,
|
|
)
|
|
|
|
|
|
class TrainingConfigService:
|
|
"""训练配置服务：提供训练页病人初始化配置，不写数据库。"""
|
|
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.case_repo = CaseRepository(db)
|
|
|
|
def get_recommended(self, case_id: int) -> TrainingConfigRecommendedResponse:
|
|
"""推荐配置：根据病例内容返回训练页默认病人初始化配置。"""
|
|
case = self._get_case(case_id)
|
|
recommended = self.default_patient_config(case)
|
|
return TrainingConfigRecommendedResponse(
|
|
case_id=case_id,
|
|
recommended=recommended,
|
|
recommended_labels=self.config_labels(recommended),
|
|
options=self.config_options(),
|
|
)
|
|
|
|
def get_options(self, case_id: int) -> TrainingConfigOptionsResponse:
|
|
"""配置选项：返回训练页自定义病人初始化配置的全部可选项和病例推荐值。"""
|
|
case = self._get_case(case_id)
|
|
recommended = self.default_patient_config(case)
|
|
return TrainingConfigOptionsResponse(
|
|
case_id=case_id,
|
|
recommended=recommended,
|
|
recommended_labels=self.config_labels(recommended),
|
|
options=self.config_options(),
|
|
)
|
|
|
|
def default_patient_config(self, case: CaseBase | None = None) -> PatientConfig:
|
|
"""默认配置：按病例年龄、科室语义和病情关键词初始化 AI 病人沟通风格。"""
|
|
if case is None:
|
|
return PatientConfig(
|
|
visit_environment="outpatient",
|
|
age_group="youth",
|
|
education_level="higher",
|
|
personality="calm",
|
|
)
|
|
|
|
text = self._case_text(case)
|
|
patient_age = case.patient_age
|
|
|
|
visit_environment = "outpatient"
|
|
personality = "calm"
|
|
if self._contains_any(
|
|
text,
|
|
["急诊", "危重", "休克", "昏迷", "意识障碍", "呼吸困难", "低氧", "抢救", "SpO2 90", "血氧明显下降"],
|
|
):
|
|
visit_environment = "emergency"
|
|
personality = "anxious"
|
|
elif self._contains_any(text, ["住院", "病房", "入院", "住院治疗"]):
|
|
visit_environment = "ward"
|
|
|
|
age_group = "youth"
|
|
if (patient_age is not None and patient_age < 14) or self._contains_any(
|
|
text,
|
|
["儿童", "患儿", "儿科", "小儿", "幼儿", "婴儿"],
|
|
):
|
|
age_group = "child"
|
|
elif patient_age is not None and patient_age >= 65:
|
|
age_group = "elderly"
|
|
elif patient_age is not None and patient_age >= 45:
|
|
age_group = "middle_aged"
|
|
|
|
return PatientConfig(
|
|
visit_environment=visit_environment,
|
|
age_group=age_group,
|
|
education_level="higher",
|
|
personality=personality,
|
|
)
|
|
|
|
def normalize_patient_config(self, config: PatientConfig | None, case: CaseBase | None = None) -> dict:
|
|
"""配置归一：校验并补齐前端传入的病人初始化配置。"""
|
|
selected = config or self.default_patient_config(case)
|
|
values = selected.model_dump()
|
|
allowed = {key: {item.value for item in items} for key, items in self.config_options().items()}
|
|
for key, value in values.items():
|
|
if value not in allowed.get(key, set()):
|
|
raise AppError("TRAINING_CONFIG_INVALID", f"invalid patient config field: {key}", 400)
|
|
return {
|
|
"values": values,
|
|
"labels": self.config_labels(selected),
|
|
}
|
|
|
|
def config_labels(self, config: PatientConfig) -> dict[str, str]:
|
|
"""配置标签：把配置值转换为前端和提示词可读的中文标签。"""
|
|
option_map = {
|
|
key: {item.value: item.label for item in items}
|
|
for key, items in self.config_options().items()
|
|
}
|
|
values = config.model_dump()
|
|
return {key: option_map.get(key, {}).get(value, value) for key, value in values.items()}
|
|
|
|
def config_options(self) -> dict[str, list[ConfigOption]]:
|
|
"""配置选项：训练页可选病人初始化配置。"""
|
|
return {
|
|
"visit_environment": [
|
|
ConfigOption(value="outpatient", label="门诊", description="适合常规问诊训练"),
|
|
ConfigOption(value="emergency", label="急诊", description="病情紧急，沟通节奏更快"),
|
|
ConfigOption(value="ward", label="病房", description="适合住院病情追踪和处置沟通"),
|
|
],
|
|
"age_group": [
|
|
ConfigOption(value="child", label="儿童", description="由家属代述为主"),
|
|
ConfigOption(value="youth", label="青年", description="表达清楚，能配合问诊"),
|
|
ConfigOption(value="middle_aged", label="中年", description="关注工作、家庭和慢病背景"),
|
|
ConfigOption(value="elderly", label="老年", description="表达较慢，需关注基础病和用药史"),
|
|
],
|
|
"education_level": [
|
|
ConfigOption(value="primary_or_below", label="小学及以下", description="医学术语理解弱"),
|
|
ConfigOption(value="secondary", label="中等教育", description="能理解常见健康解释"),
|
|
ConfigOption(value="higher", label="高等教育", description="理解能力强，能描述细节"),
|
|
],
|
|
"personality": [
|
|
ConfigOption(value="calm", label="平和", description="情绪稳定，按问题回答"),
|
|
ConfigOption(value="anxious", label="焦虑", description="更担心病情和治疗风险"),
|
|
ConfigOption(value="impatient", label="急躁", description="希望快速获得结论"),
|
|
ConfigOption(value="cooperative", label="配合", description="愿意补充细节"),
|
|
ConfigOption(value="suspicious", label="多疑", description="会追问检查和用药依据"),
|
|
],
|
|
}
|
|
|
|
def _get_case(self, case_id: int) -> CaseBase:
|
|
"""病例校验：读取配置请求对应的已发布病例。"""
|
|
case = self.case_repo.get_active_case(case_id)
|
|
if not case:
|
|
raise AppError("CASE_NOT_FOUND", "case not found or inactive", 404)
|
|
return case
|
|
|
|
def _case_text(self, case: CaseBase) -> str:
|
|
"""病例文本：拼接用于推荐配置推断的病例字段。"""
|
|
parts: list[str] = [
|
|
case.title or "",
|
|
case.chief_complaint or "",
|
|
case.description or "",
|
|
case.tags or "",
|
|
case.patient_gender or "",
|
|
]
|
|
for field_name in (
|
|
"symptom_tags",
|
|
"disease_tags",
|
|
"competency_tags",
|
|
"guideline_tags",
|
|
"knowledge_points",
|
|
"icd_codes",
|
|
):
|
|
parts.extend(self._stringify(getattr(case, field_name, None)))
|
|
if case.traditional_case:
|
|
parts.extend(self._stringify(case.traditional_case.guideline_reference))
|
|
if case.teaching_case:
|
|
parts.extend(self._stringify(case.teaching_case.teaching_goal))
|
|
return " ".join(item for item in parts if item)
|
|
|
|
def _stringify(self, value: Any) -> list[str]:
|
|
"""字段展开：把 JSON、列表和普通文本统一转换为可检索字符串。"""
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, str):
|
|
return [value]
|
|
if isinstance(value, dict):
|
|
return [str(item) for pair in value.items() for item in pair if item is not None]
|
|
if isinstance(value, list | tuple | set):
|
|
return [str(item) for item in value if item is not None]
|
|
return [str(value)]
|
|
|
|
def _contains_any(self, text: str, keywords: list[str]) -> bool:
|
|
"""关键词判断：判断病例文本是否命中任一推荐配置关键词。"""
|
|
return any(keyword in text for keyword in keywords)
|