刘金宝
589 lines
31 KiB
Python
589 lines
31 KiB
Python
import json
|
||
import logging
|
||
import re
|
||
import time
|
||
from typing import Any
|
||
|
||
from app.agents.llm_adapter import DeepSeekClient
|
||
from app.core.config import settings
|
||
from app.core.exceptions import AppError
|
||
from app.models.source_case import CaseBase
|
||
from app.models.training import SessionOrder, SessionSubmission, TrainingSession
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ScoringAgent:
|
||
"""评分 Agent：结合病例、问诊过程、检查结果、提交内容和 scoring_rule 生成结构化评价。"""
|
||
|
||
def __init__(self, llm: DeepSeekClient | None = None) -> None:
|
||
self.llm = llm or DeepSeekClient()
|
||
|
||
async def score(
|
||
self,
|
||
session: TrainingSession,
|
||
case: CaseBase,
|
||
memory_messages: list[dict],
|
||
orders: list[SessionOrder],
|
||
submission: SessionSubmission,
|
||
rubric: object | None,
|
||
guideline_refs: list[dict],
|
||
scoring_rules: list | None = None,
|
||
) -> dict:
|
||
"""评价生成：优先调用快速模型输出 JSON，失败时返回稳定兜底评分。"""
|
||
start = time.perf_counter()
|
||
messages = self._build_messages(session, case, memory_messages, orders, submission, guideline_refs, scoring_rules or [])
|
||
try:
|
||
response = await self.llm.chat(
|
||
messages,
|
||
settings.llm_fast_model,
|
||
thinking_enabled=settings.llm_fast_thinking_enabled,
|
||
reasoning_effort=None,
|
||
response_format={"type": "json_object"} if settings.llm_scoring_json_response else None,
|
||
max_tokens=min(settings.llm_scoring_max_tokens, 1800),
|
||
)
|
||
data = json.loads(response.content)
|
||
data = self._normalize_score_payload(data, session.score_type, guideline_refs)
|
||
data["_llm_model"] = response.model
|
||
data["_latency_metrics"] = {"scoring_latency_ms": response.latency_ms, "fallback_used": False}
|
||
return data
|
||
except (AppError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
|
||
logger.warning("scoring_agent.fallback session_id=%s error=%s", session.id, exc.__class__.__name__)
|
||
data = self._fallback_score(session.score_type, guideline_refs)
|
||
data["_llm_model"] = f"local-fallback-{settings.llm_fast_model}"
|
||
data["_latency_metrics"] = {
|
||
"scoring_latency_ms": int((time.perf_counter() - start) * 1000),
|
||
"fallback_used": True,
|
||
"fallback_reason": exc.__class__.__name__,
|
||
}
|
||
return data
|
||
|
||
async def score_teaching(
|
||
self,
|
||
*,
|
||
case: CaseBase,
|
||
teaching_payload: dict,
|
||
scoring_rules: list,
|
||
guideline_refs: list[dict],
|
||
score_type: str,
|
||
) -> dict:
|
||
"""教学互动评价：根据题目、标准答案、学生作答和评分规则生成结构化评价。"""
|
||
start = time.perf_counter()
|
||
messages = self._build_teaching_messages(case, teaching_payload, scoring_rules, guideline_refs, score_type)
|
||
try:
|
||
response = await self.llm.chat(
|
||
messages,
|
||
settings.llm_fast_model,
|
||
thinking_enabled=settings.llm_fast_thinking_enabled,
|
||
reasoning_effort=None,
|
||
response_format={"type": "json_object"} if settings.llm_scoring_json_response else None,
|
||
max_tokens=min(settings.llm_scoring_max_tokens, 1600),
|
||
)
|
||
data = json.loads(response.content)
|
||
data = self._normalize_score_payload(data, score_type, guideline_refs)
|
||
data["_llm_model"] = response.model
|
||
data["_latency_metrics"] = {"scoring_latency_ms": response.latency_ms, "fallback_used": False}
|
||
return data
|
||
except (AppError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
|
||
logger.warning("teaching_scoring_agent.fallback case_id=%s error=%s", case.id, exc.__class__.__name__)
|
||
data = self._fallback_teaching_score(score_type, guideline_refs, teaching_payload)
|
||
data["_llm_model"] = f"local-fallback-{settings.llm_fast_model}"
|
||
data["_latency_metrics"] = {
|
||
"scoring_latency_ms": int((time.perf_counter() - start) * 1000),
|
||
"fallback_used": True,
|
||
"fallback_reason": exc.__class__.__name__,
|
||
}
|
||
return data
|
||
|
||
def _build_teaching_messages(
|
||
self,
|
||
case: CaseBase,
|
||
teaching_payload: dict,
|
||
scoring_rules: list,
|
||
guideline_refs: list[dict],
|
||
score_type: str,
|
||
) -> list[dict]:
|
||
"""教学评分提示词：只传教学互动评价需要的病例、题目、答案和评分规则。"""
|
||
payload = {
|
||
"score_type": score_type,
|
||
"case": {
|
||
"case_id": case.id,
|
||
"title": case.title,
|
||
"chief_complaint": case.chief_complaint,
|
||
"description": self._truncate(case.description, 320),
|
||
"knowledge_points": case.knowledge_points or [],
|
||
"key_points": case.key_points or [],
|
||
},
|
||
"teaching": teaching_payload,
|
||
"scoring_rules": self._compact_scoring_rules(scoring_rules),
|
||
"guidelines": self._compact_guidelines(guideline_refs),
|
||
}
|
||
system = (
|
||
"你是医学教学互动评价专家，只输出合法 JSON。"
|
||
"请根据病例、教学目标、选择题、标准答案、解析文本、学生作答和评分规则生成教学评价。"
|
||
"输出字段固定为 score_type,total_score,dimension_scores,errors,improvement_plan,"
|
||
"evidence_summary,guideline_refs,overall_comment,score_details。"
|
||
"dimension_scores 包含 知识掌握、临床推理、检查理解、治疗决策、人文沟通 维度，"
|
||
"每项包含 dimension,score,max_score,comment,evidence,deductions,improvement。"
|
||
"score_details 对应 scoring_rules 或题目维度，每项包含 rule_id,dimension,score,"
|
||
"deducted_reason,evidence_message_ids,ai_confidence,comment。"
|
||
"必须指出答对题目、答错题目、错误原因、下一步学习重点。"
|
||
"本评价仅用于医学教学训练，不替代真实临床诊疗。"
|
||
)
|
||
return [{"role": "system", "content": system}, {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}]
|
||
|
||
def _fallback_teaching_score(self, score_type: str, guideline_refs: list[dict], teaching_payload: dict) -> dict:
|
||
"""教学评分兜底：LLM 不可用时按选择题正确率生成稳定评价结构。"""
|
||
results = teaching_payload.get("answer_results") or []
|
||
total = len(results)
|
||
correct = sum(1 for item in results if item.get("is_correct"))
|
||
accuracy = correct / total if total else 0
|
||
total_score = round(accuracy * 100, 1) if total else 0
|
||
incorrect = [item for item in results if not item.get("is_correct")]
|
||
incorrect_titles = [f"{item.get('question_id')}: {item.get('stem', '')}" for item in incorrect[:5]]
|
||
data = {
|
||
"score_type": "percentage",
|
||
"total_score": total_score,
|
||
"dimension_scores": [
|
||
{
|
||
"dimension": "知识掌握",
|
||
"score": round(total_score * 0.35, 1),
|
||
"max_score": 35,
|
||
"comment": f"共 {total} 题，答对 {correct} 题。",
|
||
"evidence": [f"正确率 {round(accuracy * 100, 1)}%"],
|
||
"deductions": incorrect_titles,
|
||
"improvement": "复习错题对应知识点和病例解析。",
|
||
},
|
||
{
|
||
"dimension": "临床推理",
|
||
"score": round(total_score * 0.25, 1),
|
||
"max_score": 25,
|
||
"comment": "根据选择题表现评估临床判断链路。",
|
||
"evidence": [item.get("stem", "") for item in results[:3]],
|
||
"deductions": incorrect_titles,
|
||
"improvement": "把题目选项与病例主诉、体征和检查结果逐项对应。",
|
||
},
|
||
{
|
||
"dimension": "检查理解",
|
||
"score": round(total_score * 0.15, 1),
|
||
"max_score": 15,
|
||
"comment": "重点关注检查项目与病情严重程度判断。",
|
||
"evidence": teaching_payload.get("scoring_focus", "").split("、")[:3],
|
||
"deductions": [],
|
||
"improvement": "理解血氧、胸片和炎症指标在肺炎评估中的作用。",
|
||
},
|
||
{
|
||
"dimension": "治疗决策",
|
||
"score": round(total_score * 0.15, 1),
|
||
"max_score": 15,
|
||
"comment": "根据题目表现评估治疗原则掌握情况。",
|
||
"evidence": teaching_payload.get("teaching_goal", "").split("、")[:3],
|
||
"deductions": [],
|
||
"improvement": "复习抗感染、平喘、氧合监测和风险预案。",
|
||
},
|
||
{
|
||
"dimension": "人文沟通",
|
||
"score": round(total_score * 0.10, 1),
|
||
"max_score": 10,
|
||
"comment": "教学互动中需继续强化家属沟通和健康教育。",
|
||
"evidence": ["教学互动题包含沟通与健康教育相关内容。"],
|
||
"deductions": [],
|
||
"improvement": "向家属说明病情、观察指标、复诊指征和用药注意事项。",
|
||
},
|
||
],
|
||
"score_details": [],
|
||
"errors": [
|
||
{
|
||
"title": "教学题目答题错误",
|
||
"description": "；".join(incorrect_titles) if incorrect_titles else "暂无明显错题。",
|
||
"severity": "medium" if incorrect_titles else "low",
|
||
"related_dimension": "知识掌握",
|
||
}
|
||
],
|
||
"improvement_plan": [
|
||
"复盘错题解析，明确每个选项与病例证据的对应关系。",
|
||
"把病例中的主诉、体征、检查和治疗原则整理成一条临床推理链。",
|
||
"针对血氧、胸片、炎症指标和医患沟通进行专项复习。",
|
||
],
|
||
"evidence_summary": [
|
||
f"教学互动共提交 {total} 题，答对 {correct} 题。",
|
||
"评分依据包括题目标准答案、解析文本、教学目标和评分规则。",
|
||
],
|
||
"guideline_refs": guideline_refs,
|
||
"overall_comment": f"本次教学互动正确率为 {round(accuracy * 100, 1)}%，请结合错题解析继续巩固病例关键知识点。",
|
||
}
|
||
return self._convert_to_five_point(data) if score_type == "five_point" else data
|
||
|
||
def _build_messages(
|
||
self,
|
||
session: TrainingSession,
|
||
case: CaseBase,
|
||
memory_messages: list[dict],
|
||
orders: list[SessionOrder],
|
||
submission: SessionSubmission,
|
||
guideline_refs: list[dict],
|
||
scoring_rules: list,
|
||
) -> list[dict]:
|
||
"""评分提示词：只传评价必需字段，避免旧表冗余字段和长病历全文拖慢生成。"""
|
||
transcript_summary = [
|
||
{"role": item.get("role"), "content": self._truncate(item.get("content", ""), 220)}
|
||
for item in memory_messages[-14:]
|
||
if item.get("content")
|
||
]
|
||
ordered_payload = [
|
||
{
|
||
"item_code": order.item_code,
|
||
"item_name": order.item_name,
|
||
"result_text": self._truncate(order.result_text, 180),
|
||
"is_key": order.is_key,
|
||
"is_abnormal": order.is_abnormal,
|
||
}
|
||
for order in orders
|
||
]
|
||
payload = {
|
||
"score_type": session.score_type,
|
||
"case": {
|
||
"title": case.title,
|
||
"chief_complaint": case.chief_complaint,
|
||
"standard_diagnosis": case.diagnosis_primary,
|
||
"diagnosis_basis": self._truncate(case.diagnosis_basis, 260),
|
||
"key_symptoms": case.key_symptoms or [],
|
||
"key_exams": case.key_exams or [],
|
||
"key_points": case.key_points or [],
|
||
},
|
||
"conversation": transcript_summary,
|
||
"orders": ordered_payload,
|
||
"submission": {
|
||
"primary_diagnosis": submission.primary_diagnosis,
|
||
"differential_diagnoses": submission.differential_diagnoses or [],
|
||
"diagnosis_basis": self._truncate(submission.diagnosis_basis, 320),
|
||
"treatment_principle": self._truncate(submission.treatment_principle, 220),
|
||
"treatment_measures": self._truncate(submission.treatment_measures, 320),
|
||
"risk_plan": self._truncate(submission.risk_plan, 220),
|
||
"communication": self._truncate(submission.communication, 220),
|
||
"follow_up": self._truncate(submission.follow_up, 180),
|
||
},
|
||
"scoring_rules": self._compact_scoring_rules(scoring_rules),
|
||
"guidelines": self._compact_guidelines(guideline_refs),
|
||
}
|
||
system = (
|
||
"你是医学教学问诊评分专家，只输出合法 JSON。"
|
||
"请结合病例、问诊过程、检查申请、诊断和治疗提交进行教学评价。"
|
||
"输出字段固定为 score_type,total_score,dimension_scores,errors,improvement_plan,"
|
||
"evidence_summary,guideline_refs,overall_comment,score_details。"
|
||
"dimension_scores 为 5-6 项，每项包含 dimension,score,max_score,comment,evidence,deductions,improvement。"
|
||
"score_details 对应 scoring_rules，每项包含 rule_id,dimension,score,deducted_reason,evidence_message_ids,ai_confidence,comment。"
|
||
"evidence、deductions、improvement_plan、evidence_summary 必须是数组，每个元素一句话。"
|
||
"errors 每项包含 title,description,severity,related_dimension。"
|
||
"评价必须具体指出用户问了什么、申请了什么检查、诊断治疗哪里充分或不足。"
|
||
"报告仅用于教学训练，不替代真实临床诊疗。"
|
||
)
|
||
return [{"role": "system", "content": system}, {"role": "user", "content": json.dumps(payload, ensure_ascii=False)}]
|
||
|
||
def _compact_scoring_rules(self, scoring_rules: list) -> list[dict]:
|
||
"""源库评分规则压缩：把 scoring_rule 转为评分 Agent 可直接使用的细则。"""
|
||
compact = []
|
||
for item in scoring_rules[:12]:
|
||
compact.append(
|
||
{
|
||
"rule_id": getattr(item, "id", None),
|
||
"dimension": getattr(item, "dimension", ""),
|
||
"competency_dimension": getattr(item, "competency_dimension", ""),
|
||
"score_weight": float(getattr(item, "score_weight", 0) or 0),
|
||
"ai_auto_score": bool(getattr(item, "ai_auto_score", True)),
|
||
"osce_dimension": bool(getattr(item, "osce_dimension", False)),
|
||
"scoring_standard": self._truncate(getattr(item, "scoring_standard", ""), 240),
|
||
"rubric_json": getattr(item, "rubric_json", {}) or {},
|
||
}
|
||
)
|
||
return compact
|
||
|
||
def _compact_guidelines(self, guideline_refs: list[dict]) -> list[dict]:
|
||
"""参考指南压缩：保留少量命中来源，供模型引用。"""
|
||
compact = []
|
||
for item in (guideline_refs or [])[:3]:
|
||
if isinstance(item, dict):
|
||
compact.append(
|
||
{
|
||
"title": item.get("title") or item.get("source"),
|
||
"content": self._truncate(item.get("content") or item.get("chunk"), 180),
|
||
}
|
||
)
|
||
return compact
|
||
|
||
def _normalize_score_payload(self, data: dict, score_type: str, guideline_refs: list[dict]) -> dict:
|
||
"""评分结构校验：补齐缺失字段，并拆分模型返回的长编号列表。"""
|
||
if not isinstance(data, dict):
|
||
data = self._fallback_score(score_type, guideline_refs)
|
||
|
||
data.setdefault("score_type", "percentage")
|
||
data.setdefault("total_score", 0)
|
||
data.setdefault("dimension_scores", [])
|
||
data.setdefault("score_details", [])
|
||
data.setdefault("errors", [])
|
||
data.setdefault("improvement_plan", [])
|
||
data.setdefault("evidence_summary", [])
|
||
data.setdefault("guideline_refs", guideline_refs)
|
||
data.setdefault("overall_comment", "")
|
||
|
||
normalized_dimensions = []
|
||
for item in data.get("dimension_scores") or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
normalized_dimensions.append(
|
||
{
|
||
"dimension": str(item.get("dimension") or "未命名维度"),
|
||
"score": float(item.get("score") or 0),
|
||
"max_score": float(item.get("max_score") or (100 if data.get("score_type") == "percentage" else 5)),
|
||
"comment": self._truncate(item.get("comment") or "", 180),
|
||
"evidence": self._ensure_list(item.get("evidence")),
|
||
"deductions": self._ensure_list(item.get("deductions")),
|
||
"improvement": self._truncate(item.get("improvement") or "", 180),
|
||
}
|
||
)
|
||
data["dimension_scores"] = normalized_dimensions or self._fallback_score("percentage", guideline_refs)["dimension_scores"]
|
||
data["score_details"] = self._normalize_score_details(data.get("score_details"), data["dimension_scores"])
|
||
data["errors"] = self._normalize_errors(data.get("errors"))
|
||
data["improvement_plan"] = self._ensure_list(data.get("improvement_plan"))
|
||
data["evidence_summary"] = self._ensure_list(data.get("evidence_summary"))
|
||
data["guideline_refs"] = self._normalize_guideline_refs(data.get("guideline_refs"), guideline_refs)
|
||
data["overall_comment"] = self._truncate(data.get("overall_comment") or "", 260)
|
||
|
||
if score_type == "five_point" and data.get("score_type") != "five_point":
|
||
return self._convert_to_five_point(data)
|
||
if score_type == "percentage" and data.get("score_type") != "percentage":
|
||
data["score_type"] = "percentage"
|
||
return data
|
||
|
||
def _normalize_score_details(self, raw_details: object, dimension_scores: list[dict]) -> list[dict]:
|
||
"""评分明细归一化：生成可落库的 training_score_detail 数据。"""
|
||
source = raw_details if isinstance(raw_details, list) and raw_details else dimension_scores
|
||
details = []
|
||
for item in source:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
deducted_reason = item.get("deducted_reason")
|
||
if not deducted_reason:
|
||
deducted_reason = "；".join(str(value) for value in item.get("deductions", []) if value)
|
||
details.append(
|
||
{
|
||
"rule_id": item.get("rule_id"),
|
||
"dimension": str(item.get("dimension") or "综合表现"),
|
||
"score": float(item.get("score") or 0),
|
||
"deducted_reason": self._truncate(deducted_reason or "", 260),
|
||
"evidence_message_ids": self._ensure_list(item.get("evidence_message_ids") or item.get("evidence")),
|
||
"ai_confidence": float(item.get("ai_confidence") or 0.85),
|
||
"comment": self._truncate(item.get("comment") or item.get("improvement") or "", 220),
|
||
}
|
||
)
|
||
return details
|
||
|
||
def _normalize_errors(self, errors: object) -> list[dict]:
|
||
"""错误项归一化：转为报告可渲染的扣分项。"""
|
||
normalized = []
|
||
for index, item in enumerate(self._ensure_list(errors), start=1):
|
||
if isinstance(item, dict):
|
||
normalized.append(
|
||
{
|
||
"title": self._truncate(item.get("title") or f"问题 {index}", 60),
|
||
"description": self._truncate(item.get("description") or item.get("comment") or "", 180),
|
||
"severity": item.get("severity") or "medium",
|
||
"related_dimension": item.get("related_dimension") or item.get("dimension") or "综合表现",
|
||
}
|
||
)
|
||
else:
|
||
normalized.append(
|
||
{
|
||
"title": f"问题 {index}",
|
||
"description": self._truncate(str(item), 180),
|
||
"severity": "medium",
|
||
"related_dimension": "综合表现",
|
||
}
|
||
)
|
||
return normalized[:6]
|
||
|
||
def _normalize_guideline_refs(self, value: object, fallback_refs: list[dict]) -> list[dict]:
|
||
"""指南引用归一化：保证字符串、字典或空值都能转成字典数组。"""
|
||
raw_items = value if isinstance(value, list) else ([value] if value else fallback_refs)
|
||
normalized = []
|
||
for index, item in enumerate(raw_items or [], start=1):
|
||
if isinstance(item, dict):
|
||
normalized.append(
|
||
{
|
||
"title": self._truncate(item.get("title") or item.get("source") or f"参考依据 {index}", 80),
|
||
"content": self._truncate(
|
||
item.get("content") or item.get("text") or item.get("chunk") or item.get("summary") or "",
|
||
180,
|
||
),
|
||
"source": self._truncate(item.get("source") or item.get("source_type") or "knowledge_base", 80),
|
||
}
|
||
)
|
||
elif item:
|
||
normalized.append({"title": f"参考依据 {index}", "content": self._truncate(str(item), 180), "source": "llm_output"})
|
||
return normalized[:6]
|
||
|
||
def _ensure_list(self, value: object) -> list:
|
||
"""列表规整：拆分模型常见的 1/2/3 编号长文本。"""
|
||
if value is None:
|
||
return []
|
||
raw_items = value if isinstance(value, list) else [value]
|
||
items: list = []
|
||
for raw in raw_items:
|
||
if raw is None:
|
||
continue
|
||
if isinstance(raw, dict):
|
||
items.append(raw)
|
||
continue
|
||
text = str(raw).strip()
|
||
if not text:
|
||
continue
|
||
parts = re.split(r"(?:^|\s)(?:\d+[\.\、)]|[；;]\d+[；;])\s*", text)
|
||
parts = [part.strip("；;、\n\t ") for part in parts if part and part.strip("；;、\n\t ")]
|
||
items.extend(parts if len(parts) > 1 else [text])
|
||
return [self._truncate(item, 180) if not isinstance(item, dict) else item for item in items[:8]]
|
||
|
||
def _fallback_score(self, score_type: str, guideline_refs: list[dict]) -> dict:
|
||
"""评分兜底：LLM 失败时生成稳定结构化评分，保证报告流程可展示。"""
|
||
data = {
|
||
"score_type": "percentage",
|
||
"total_score": 80,
|
||
"dimension_scores": [
|
||
{
|
||
"dimension": "信息获取",
|
||
"score": 20,
|
||
"max_score": 25,
|
||
"comment": "完成主要症状追问，但儿科专科病史仍需补充。",
|
||
"evidence": ["围绕发热、咳嗽、喘息等核心症状展开问诊。"],
|
||
"deductions": ["既往喘息史、过敏史、疫苗接种史、家属照护能力等信息不够完整。"],
|
||
"improvement": "下一次按主诉、现病史、既往史、个人史、家族史和家属顾虑逐项补全。",
|
||
},
|
||
{
|
||
"dimension": "分析推理",
|
||
"score": 16,
|
||
"max_score": 20,
|
||
"comment": "诊断方向基本正确，但严重程度分层需要更清晰。",
|
||
"evidence": ["主要诊断指向支气管肺炎。"],
|
||
"deductions": ["鉴别诊断和严重程度判断未充分引用血氧、胸片和炎症指标。"],
|
||
"improvement": "把症状、体征、影像、血氧和炎症指标逐条对应到诊断依据和病情分层。",
|
||
},
|
||
{
|
||
"dimension": "检查利用",
|
||
"score": 12,
|
||
"max_score": 15,
|
||
"comment": "关键检查申请较完整，但检查结果解释仍可细化。",
|
||
"evidence": ["胸片、血氧或炎症指标可支持肺炎诊断和严重程度判断。"],
|
||
"deductions": ["对 SpO2、胸片异常和炎症指标的临床意义解释不够具体。"],
|
||
"improvement": "在诊断依据中写明每项异常检查如何支持诊断、鉴别诊断和治疗决策。",
|
||
},
|
||
{
|
||
"dimension": "处置决策",
|
||
"score": 16,
|
||
"max_score": 20,
|
||
"comment": "治疗原则基本完整，仍需补充监测和升级处理条件。",
|
||
"evidence": ["治疗原则覆盖抗感染、止咳平喘、改善氧合和观察病情。"],
|
||
"deductions": ["药物选择、门诊/住院判断、风险预案和随访节点不够具体。"],
|
||
"improvement": "补充治疗适应证、监测指标、病情加重预案和复诊/住院指征。",
|
||
},
|
||
{
|
||
"dimension": "沟通人文",
|
||
"score": 8,
|
||
"max_score": 10,
|
||
"comment": "有基本沟通意识，但家属顾虑回应不够结构化。",
|
||
"evidence": ["治疗方案中包含向家属说明病情和复诊指征。"],
|
||
"deductions": ["知情同意、家属担心、用药注意事项和家庭护理教育仍需补充。"],
|
||
"improvement": "增加对家属顾虑的回应、危险信号说明和家庭护理指导。",
|
||
},
|
||
{
|
||
"dimension": "临床整合",
|
||
"score": 8,
|
||
"max_score": 10,
|
||
"comment": "完成训练闭环，证据衔接仍需强化。",
|
||
"evidence": ["完成问诊、检查、诊断、治疗和评价流程。"],
|
||
"deductions": ["各阶段证据之间的逻辑衔接和 SOAP 化表达不够清晰。"],
|
||
"improvement": "用 SOAP 结构归纳病情，把证据、判断和计划串联起来。",
|
||
},
|
||
],
|
||
"score_details": [
|
||
{
|
||
"rule_id": None,
|
||
"dimension": "信息获取",
|
||
"score": 20,
|
||
"deducted_reason": "既往喘息史、过敏史、疫苗接种史、家属照护能力等信息不够完整。",
|
||
"evidence_message_ids": ["围绕发热、咳嗽、喘息等核心症状展开问诊。"],
|
||
"ai_confidence": 0.85,
|
||
"comment": "完成主要症状追问，但儿科专科病史仍需补充。",
|
||
},
|
||
{
|
||
"rule_id": None,
|
||
"dimension": "分析推理",
|
||
"score": 16,
|
||
"deducted_reason": "鉴别诊断和严重程度判断未充分引用血氧、胸片和炎症指标。",
|
||
"evidence_message_ids": ["主要诊断指向支气管肺炎。"],
|
||
"ai_confidence": 0.84,
|
||
"comment": "诊断方向基本正确，但严重程度分层需要更清晰。",
|
||
},
|
||
{
|
||
"rule_id": None,
|
||
"dimension": "检查利用",
|
||
"score": 12,
|
||
"deducted_reason": "对 SpO2、胸片异常和炎症指标的临床意义解释不够具体。",
|
||
"evidence_message_ids": ["胸片、血氧或炎症指标可支持肺炎诊断和严重程度判断。"],
|
||
"ai_confidence": 0.84,
|
||
"comment": "关键检查申请较完整，但检查结果解释仍可细化。",
|
||
},
|
||
],
|
||
"errors": [
|
||
{
|
||
"title": "信息采集不够系统",
|
||
"description": "儿科特异性病史追问不足，影响对喘息诱因、感染风险和肺炎严重程度的判断。",
|
||
"severity": "medium",
|
||
"related_dimension": "信息获取",
|
||
}
|
||
],
|
||
"improvement_plan": [
|
||
"补充疫苗接种史、过敏史、既往喘息史、近期接触史和家庭照护能力评估。",
|
||
"诊断依据中逐条引用发热、咳嗽、喘息、肺部体征、胸片、血氧和炎症指标。",
|
||
"治疗方案中补充监测指标、病情加重预案、复诊/住院指征和家属健康教育。",
|
||
"沟通环节增加家属担心回应、知情说明和家庭护理要点。",
|
||
],
|
||
"evidence_summary": [
|
||
"问诊：已完成核心症状追问，但儿科特异性病史仍需补全。",
|
||
"检查：需重点评价是否申请并利用胸片、血氧和炎症指标。",
|
||
"诊断：主要诊断方向基本正确，鉴别诊断和严重程度分层需要更完整。",
|
||
"治疗：治疗原则基本覆盖抗感染、平喘、氧合和观察。",
|
||
"沟通：已有基本告知意识，仍需加强家属顾虑回应和健康教育。",
|
||
],
|
||
"guideline_refs": guideline_refs,
|
||
"overall_comment": "本次训练完成核心闭环，诊断和处理方向基本正确；后续需要强化问诊结构化、检查结果利用和治疗细化。",
|
||
}
|
||
return self._convert_to_five_point(data) if score_type == "five_point" else data
|
||
|
||
def _convert_to_five_point(self, data: dict) -> dict:
|
||
"""分数转换：将百分制评价转换为五分制，同时保留证据、扣分和改进细则。"""
|
||
converted = dict(data)
|
||
converted["score_type"] = "five_point"
|
||
converted["total_score"] = round(float(data.get("total_score", 0)) / 20, 1)
|
||
converted["dimension_scores"] = [
|
||
{
|
||
**item,
|
||
"score": round(float(item.get("score", 0)) / float(item.get("max_score") or 100) * 5, 1),
|
||
"max_score": 5,
|
||
}
|
||
for item in data.get("dimension_scores", [])
|
||
]
|
||
converted["score_details"] = [
|
||
{
|
||
**item,
|
||
"score": round(float(item.get("score", 0)) / 20, 1),
|
||
}
|
||
for item in data.get("score_details", [])
|
||
]
|
||
return converted
|
||
|
||
def _truncate(self, value: Any, limit: int) -> str:
|
||
"""文本截断：限制模型输入和报告字段长度，降低评分耗时并稳定排版。"""
|
||
if value is None:
|
||
return ""
|
||
text = str(value).strip()
|
||
return text if len(text) <= limit else f"{text[:limit]}..."
|