181 lines
7.2 KiB
Python
181 lines
7.2 KiB
Python
|
from typing import Dict, Any, List
|
|||
|
|
from agent_system.base import BaseAgent
|
|||
|
|
from agent_system.disease_analyzer.prompt import DiseaseAnalyzerPrompt
|
|||
|
|
from agent_system.disease_analyzer.response_model import DiseaseAnalysisResult
|
|||
|
|
|
|||
|
|
|
|||
|
|
class DiseaseContextAnalyzer(BaseAgent):
|
|||
|
|
"""
|
|||
|
|
疾病上下文分析智能体
|
|||
|
|
|
|||
|
|
基于患者的现病史和既往史内容，分析疾病类型特点，
|
|||
|
|
生成初步诊断判断，并确定各子任务的评估重点。
|
|||
|
|
|
|||
|
|
核心功能:
|
|||
|
|
1. 识别疾病所属系统（神经、心血管、呼吸、消化等）
|
|||
|
|
2. 分析起病模式（急性、亚急性、慢性）和严重程度
|
|||
|
|
3. 推断可能的疾病诊断
|
|||
|
|
4. 为后续子任务确定针对性的评估重点
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
model_type (str): 使用的大语言模型类型，默认为 deepseek-v3
|
|||
|
|
llm_config (dict): LLM模型配置参数
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
def __init__(self, model_type: str = "deepseek-v3", llm_config: dict = None):
|
|||
|
|
"""
|
|||
|
|
初始化疾病上下文分析智能体
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
model_type (str): 大语言模型类型，默认使用 deepseek-v3
|
|||
|
|
llm_config (dict): LLM模型的配置参数，如果为None则使用默认配置
|
|||
|
|
"""
|
|||
|
|
super().__init__(
|
|||
|
|
model_type=model_type,
|
|||
|
|
description="基于患者主述分析疾病上下文并确定评估重点",
|
|||
|
|
instructions=DiseaseAnalyzerPrompt.instructions,
|
|||
|
|
response_model=DiseaseAnalysisResult,
|
|||
|
|
llm_config=llm_config or {},
|
|||
|
|
structured_outputs=True,
|
|||
|
|
markdown=False,
|
|||
|
|
use_cache=False
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def run(self, hpi_content: str, ph_content: str = "") -> DiseaseAnalysisResult:
|
|||
|
|
"""
|
|||
|
|
执行疾病上下文分析
|
|||
|
|
|
|||
|
|
基于现病史和既往史内容，分析疾病特点，生成初步判断，
|
|||
|
|
并为后续的现病史收集、既往史收集等子任务确定评估重点。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
hpi_content (str): 现病史内容，患者的主要症状描述
|
|||
|
|
ph_content (str, optional): 既往史内容，患者的历史疾病信息，默认为空字符串
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
DiseaseAnalysisResult: 包含疾病分析结果的结构化数据，包括：
|
|||
|
|
- disease_category: 疾病类别（如神经系统疾病等）
|
|||
|
|
- suspected_conditions: 可能的诊断列表
|
|||
|
|
- onset_pattern: 起病模式（急性/亚急性/慢性）
|
|||
|
|
- severity_level: 疾病严重程度（轻度/中度/重度）
|
|||
|
|
- evaluation_priorities: 各子任务的评估重点
|
|||
|
|
- medical_reasoning: 医学分析推理过程
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
Exception: 当LLM调用失败时，返回包含默认信息的DiseaseAnalysisResult
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
# 构建分析提示词
|
|||
|
|
prompt = self._build_analysis_prompt(hpi_content, ph_content)
|
|||
|
|
|
|||
|
|
# 调用基类的run方法执行LLM推理
|
|||
|
|
result = super().run(prompt)
|
|||
|
|
|
|||
|
|
# 确保返回正确的类型并进行类型转换
|
|||
|
|
return self._ensure_result_type(result)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
# 当分析失败时记录错误并返回默认结果
|
|||
|
|
print(f"疾病上下文分析失败: {str(e)}")
|
|||
|
|
return self._get_fallback_result()
|
|||
|
|
|
|||
|
|
def _ensure_result_type(self, result: Any) -> DiseaseAnalysisResult:
|
|||
|
|
"""
|
|||
|
|
确保返回结果为正确的类型
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
result (Any): LLM返回的原始结果
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
DiseaseAnalysisResult: 转换后的结构化结果
|
|||
|
|
"""
|
|||
|
|
if isinstance(result, DiseaseAnalysisResult):
|
|||
|
|
return result
|
|||
|
|
elif isinstance(result, dict):
|
|||
|
|
return DiseaseAnalysisResult(**result)
|
|||
|
|
else:
|
|||
|
|
# 如果类型不匹配，返回默认结果
|
|||
|
|
return self._get_fallback_result()
|
|||
|
|
|
|||
|
|
def _get_fallback_result(self) -> DiseaseAnalysisResult:
|
|||
|
|
"""
|
|||
|
|
生成分析失败时的默认结果
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
DiseaseAnalysisResult: 包含默认评估重点的结果
|
|||
|
|
"""
|
|||
|
|
return DiseaseAnalysisResult(
|
|||
|
|
disease_category="未知疾病类型",
|
|||
|
|
suspected_conditions=["需进一步分析"],
|
|||
|
|
onset_pattern="未明确",
|
|||
|
|
severity_level="未评估",
|
|||
|
|
evaluation_priorities={
|
|||
|
|
"诊疗经过": ["既往就诊经历", "诊断检查结果", "治疗方案及效果"],
|
|||
|
|
"主要症状特征": ["症状的具体表现", "症状的严重程度", "症状的持续时间"],
|
|||
|
|
"伴随症状": ["相关系统症状", "全身性症状", "功能性症状"],
|
|||
|
|
"病情发展与演变": ["症状变化趋势", "诱发或缓解因素", "病程发展规律"]
|
|||
|
|
},
|
|||
|
|
medical_reasoning="由于分析过程中出现异常，系统提供了通用的评估重点，建议人工进一步分析患者病情。"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
def _build_analysis_prompt(self, hpi_content: str, ph_content: str) -> str:
|
|||
|
|
"""
|
|||
|
|
构建疾病分析的提示词模板
|
|||
|
|
|
|||
|
|
根据现病史和既往史内容，构建简洁高效的分析提示词，
|
|||
|
|
引导LLM进行专业的医学分析和判断。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
hpi_content (str): 现病史内容
|
|||
|
|
ph_content (str): 既往史内容
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
str: 精简的分析提示词
|
|||
|
|
"""
|
|||
|
|
# 确保既往史内容的合理显示
|
|||
|
|
past_history_display = ph_content.strip() if ph_content.strip() else "暂无既往史信息"
|
|||
|
|
|
|||
|
|
# 从prompt类获取示例输出格式
|
|||
|
|
from agent_system.disease_analyzer.prompt import DiseaseAnalyzerPrompt
|
|||
|
|
example_output = DiseaseAnalyzerPrompt.get_example_output()
|
|||
|
|
|
|||
|
|
prompt = f"""患者病史信息：
|
|||
|
|
现病史: {hpi_content}
|
|||
|
|
既往史: {past_history_display}
|
|||
|
|
|
|||
|
|
请分析疾病系统、起病模式、初步诊断，并为关键子任务确定评估重点。
|
|||
|
|
|
|||
|
|
输出格式示例：
|
|||
|
|
{example_output}
|
|||
|
|
|
|||
|
|
请严格按照上述JSON格式输出。"""
|
|||
|
|
|
|||
|
|
return prompt
|
|||
|
|
|
|||
|
|
def analyze_patient_chief_complaint(self, chief_complaint: str) -> DiseaseAnalysisResult:
|
|||
|
|
"""
|
|||
|
|
基于患者主述进行初步疾病分析的便捷接口
|
|||
|
|
|
|||
|
|
这是一个专门针对患者主述（chief complaint）的分析方法，
|
|||
|
|
适用于初诊时仅有患者主述信息的情况。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
chief_complaint (str): 患者的主要症状主述
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
DiseaseAnalysisResult: 基于主述的初步分析结果
|
|||
|
|
"""
|
|||
|
|
return self.run(hpi_content=chief_complaint, ph_content="")
|
|||
|
|
|
|||
|
|
def get_evaluation_priorities_for_task(self, result: DiseaseAnalysisResult, task_name: str) -> List[str]:
|
|||
|
|
"""
|
|||
|
|
获取特定子任务的评估重点
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
result (DiseaseAnalysisResult): 疾病分析结果
|
|||
|
|
task_name (str): 子任务名称
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
List[str]: 该任务的评估重点列表，如果任务不存在则返回空列表
|
|||
|
|
"""
|
|||
|
|
return result.evaluation_priorities.get(task_name, [])
|