diff --git a/.gitignore b/.gitignore index f811c9f..06b7dc9 100644 --- a/.gitignore +++ b/.gitignore @@ -42,3 +42,5 @@ logs/ .coverage htmlcov/ .tox/ + +dataset/ diff --git a/agent_system/controller/__init__.py b/agent_system/controller/__init__.py new file mode 100644 index 0000000..4f35c5f --- /dev/null +++ b/agent_system/controller/__init__.py @@ -0,0 +1 @@ +# Controller Agent Module \ No newline at end of file diff --git a/agent_system/controller/agent.py b/agent_system/controller/agent.py new file mode 100644 index 0000000..86a9fdb --- /dev/null +++ b/agent_system/controller/agent.py @@ -0,0 +1,230 @@ +from typing import Dict, Any, List +from agent_system.base import BaseAgent +from agent_system.controller.prompt import ControllerPrompt +from agent_system.controller.response_model import ControllerDecision, TaskInfo + + +class TaskController(BaseAgent): + """ + 任务控制器智能体 + + 负责根据患者的临床信息（现病史、既往史、主诉）从未完成的任务列表中 + 选择最合适的下一步任务，并提供具体的执行指导建议。 + + 核心功能: + 1. 分析患者的临床信息和病情特征 + 2. 评估待执行任务的优先级和重要性 + 3. 选择最适合当前情况的任务 + 4. 提供针对性的任务执行指导 + + Attributes: + model_type (str): 使用的大语言模型类型，默认为 gpt-oss:latest + llm_config (dict): LLM模型配置参数 + """ + + def __init__(self, model_type: str = "gpt-oss:latest", llm_config: dict = None): + """ + 初始化任务控制器智能体 + + Args: + model_type (str): 大语言模型类型，默认使用 gpt-oss:latest + llm_config (dict): LLM模型的配置参数，如果为None则使用默认配置 + """ + super().__init__( + model_type=model_type, + description="医疗任务控制器，负责任务选择和执行指导", + instructions=ControllerPrompt.instructions, + response_model=ControllerDecision, + llm_config=llm_config or {}, + structured_outputs=True, + markdown=False, + use_cache=False + ) + + def run(self, + pending_tasks: List[Dict[str, str]], + chief_complaint: str, + hpi_content: str = "", + ph_content: str = "") -> ControllerDecision: + """ + 执行任务控制决策 + + 基于患者的临床信息和待执行的任务列表，选择最合适的任务 + 并提供具体的执行指导建议。 + + Args: + pending_tasks (List[Dict[str, str]]): 待执行的任务列表，每个任务包含name、priority、description字段 + chief_complaint (str): 患者主诉 + hpi_content (str, optional): 现病史内容，默认为空字符串 + ph_content (str, optional): 既往史内容，默认为空字符串 + + Returns: + ControllerDecision: 包含任务选择决策和指导建议的结构化数据，包括： + - selected_task: 选择的任务信息 + - specific_guidance: 针对选定任务的具体指导建议 + + Raises: + Exception: 当LLM调用失败时，返回包含默认信息的ControllerDecision + """ + try: + # 构建决策提示词 + prompt = self._build_decision_prompt( + pending_tasks, chief_complaint, 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(pending_tasks) + + def _ensure_result_type(self, result: Any) -> ControllerDecision: + """ + 确保返回结果为正确的类型 + + Args: + result (Any): LLM返回的原始结果 + + Returns: + ControllerDecision: 转换后的结构化结果 + """ + if isinstance(result, ControllerDecision): + return result + elif isinstance(result, dict): + return ControllerDecision(**result) + else: + # 如果类型不匹配，返回默认结果 + return self._get_fallback_result([]) + + def _get_fallback_result(self, pending_tasks: List[Dict[str, str]]) -> ControllerDecision: + """ + 生成决策失败时的默认结果 + + Args: + pending_tasks (List[Dict[str, str]]): 待执行的任务列表 + + Returns: + ControllerDecision: 包含默认任务选择的结果 + """ + # 如果有待执行任务，选择第一个作为默认任务 + if pending_tasks: + default_task = pending_tasks[0] + selected_task_info = TaskInfo( + task_name=default_task.get("name", "未知任务"), + priority=default_task.get("priority", "中"), + description=default_task.get("description", "任务描述不可用") + ) + else: + selected_task_info = TaskInfo( + task_name="基本信息收集", + priority="中", + description="收集患者的基本临床信息" + ) + + return ControllerDecision( + selected_task=selected_task_info, + specific_guidance="由于系统异常，建议按照标准临床流程进行患者评估，重点关注患者的主要症状和病史信息，并人工审核患者情况。" + ) + + def _build_decision_prompt(self, + pending_tasks: List[Dict[str, str]], + chief_complaint: str, + hpi_content: str, + ph_content: str) -> str: + """ + 构建任务控制决策的提示词模板 + + 根据待执行任务列表和患者临床信息，构建简洁高效的决策提示词， + 引导LLM进行专业的任务选择和指导建议。 + + Args: + pending_tasks (List[Dict[str, str]]): 待执行的任务列表 + chief_complaint (str): 患者主诉 + hpi_content (str): 现病史内容 + ph_content (str): 既往史内容 + + Returns: + str: 精简的决策提示词 + """ + # 格式化待执行任务列表 + tasks_display = "" + for i, task in enumerate(pending_tasks, 1): + task_name = task.get("name", "未知任务") + task_priority = task.get("priority", "未设定") + task_desc = task.get("description", "无描述") + tasks_display += f"{i}. 任务名称: {task_name}\n 优先级: {task_priority}\n 描述: {task_desc}\n\n" + + if not tasks_display.strip(): + tasks_display = "当前没有待执行的任务。" + + # 确保临床信息的合理显示 + hpi_display = hpi_content.strip() if hpi_content.strip() else "暂无现病史信息" + ph_display = ph_content.strip() if ph_content.strip() else "暂无既往史信息" + + # 从prompt类获取示例输出格式 + example_output = ControllerPrompt.get_example_output() + + prompt = f"""患者临床信息： +主诉: {chief_complaint} +现病史: {hpi_display} +既往史: {ph_display} + +待执行任务列表： +{tasks_display} + +请根据患者的临床信息分析病情特征，从上述任务列表中选择最合适的下一步任务，并提供具体的执行指导建议。 + +输出格式示例： +{example_output} + +请严格按照上述JSON格式输出。 +输出内容为:""" + + return prompt + + def select_optimal_task(self, + tasks: List[Dict[str, str]], + patient_info: Dict[str, str]) -> ControllerDecision: + """ + 基于患者信息选择最优任务的便捷接口 + + 这是一个专门用于任务选择的简化接口，接受结构化的患者信息。 + + Args: + tasks (List[Dict[str, str]]): 待执行的任务列表 + patient_info (Dict[str, str]): 患者信息字典，包含chief_complaint、hpi、ph等字段 + + Returns: + ControllerDecision: 任务选择决策结果 + """ + chief_complaint = patient_info.get("chief_complaint", "") + hpi_content = patient_info.get("hpi", "") + ph_content = patient_info.get("ph", "") + + return self.run( + pending_tasks=tasks, + chief_complaint=chief_complaint, + hpi_content=hpi_content, + ph_content=ph_content + ) + + def get_task_guidance(self, result: ControllerDecision) -> Dict[str, Any]: + """ + 获取任务执行指导的结构化信息 + + Args: + result (ControllerDecision): 控制器决策结果 + + Returns: + Dict[str, Any]: 包含任务指导信息的字典 + """ + return { + "task_name": result.selected_task.task_name, + "priority": result.selected_task.priority, + "guidance": result.specific_guidance + } \ No newline at end of file diff --git a/agent_system/controller/prompt.py b/agent_system/controller/prompt.py new file mode 100644 index 0000000..f2a6d69 --- /dev/null +++ b/agent_system/controller/prompt.py @@ -0,0 +1,68 @@ +from typing import Dict, Any +import json + + +class ControllerPrompt: + """ + Controller智能体的提示词类 + """ + + description = """ + 你是一个医疗任务控制器智能体，专门负责根据患者的临床信息来选择和指导医疗任务的执行。 + + 你的主要角色是： + 1. 任务协调者：从待执行的任务列表中选择最合适的下一步任务 + 2. 临床决策支持：基于患者的现病史、既往史和主诉提供专业的医疗指导 + 3. 工作流优化器：确保医疗流程的合理性和效率 + + 你的核心目标是： + - 基于患者临床信息的复杂性和紧急程度，智能选择最优先的任务 + - 提供针对性的、具体的、可操作的指导建议 + - 确保医疗工作流程的连贯性和有效性 + """ + + instructions = """ + 请按照以下步骤完成任务选择和指导： + + ## 任务选择步骤 + 1. 分析患者的主诉、现病史和既往史，识别关键临床特征 + 2. 评估待执行任务列表，根据临床信息确定任务优先级 + 3. 选择最符合当前临床需求的任务 + 4. 提供针对患者具体情况的任务执行指导建议 + + ## 输出格式要求 + 请严格按照以下JSON格式输出： + + { + "selected_task": { + "task_name": "选择的任务名称", + "priority": "任务优先级（紧急/高/中/低）", + "description": "任务描述" + }, + "specific_guidance": "基于患者的现病史、既往史和主诉，针对选定任务提供详细的执行指导建议，包括重点关注的方面、询问的问题、检查的要点等" + } + + ## 注意事项 + - 基于循证医学原则进行分析 + - 优先考虑患者安全和诊疗效果 + - 提供具体可操作的指导建议 + """ + + @staticmethod + def get_example_output() -> str: + """ + 获取Controller智能体的示例输出格式 + + Returns: + str: JSON格式的示例输出 + """ + example_output = { + "selected_task": { + "task_name": "详细现病史收集", + "priority": "高", + "description": "深入收集患者现病史的详细信息，包括症状特征、发展过程、伴随症状等" + }, + "specific_guidance": "基于患者胸痛3天伴气短的主诉和高血压既往史，在收集现病史时应重点询问：1）胸痛的确切位置、性质（压榨性、刺痛、撕裂样等）和放射部位；2）疼痛的严重程度（0-10分评分）和持续时间；3）诱发因素（活动、休息、情绪变化等）和缓解因素；4）伴随症状如气短、出汗、恶心、头晕等；5）发作规律和频率变化；6）既往类似发作史。特别关注与患者高血压病史相关的心血管风险因素，需要快速排除急性冠脉综合征、主动脉夹层等急危重症。" + } + + return json.dumps(example_output, ensure_ascii=False, indent=2) \ No newline at end of file diff --git a/agent_system/controller/response_model.py b/agent_system/controller/response_model.py new file mode 100644 index 0000000..04a9b83 --- /dev/null +++ b/agent_system/controller/response_model.py @@ -0,0 +1,39 @@ +from pydantic import Field +from agent_system.base import BaseResponseModel + + +class TaskInfo(BaseResponseModel): + """ + 任务信息模型 + """ + task_name: str = Field( + ..., + description="任务名称" + ) + priority: str = Field( + ..., + description="任务优先级（紧急、高、中、低）" + ) + description: str = Field( + ..., + description="任务描述" + ) + + +class ControllerDecision(BaseResponseModel): + """ + Controller智能体决策结果模型 + + 基于未完成的任务列表、现病史、既往史与主诉， + 输出选择的任务以及具体的指导建议。 + """ + + selected_task: TaskInfo = Field( + ..., + description="选择执行的任务信息" + ) + + specific_guidance: str = Field( + ..., + description="基于现病史、既往史与主诉，针对选定任务的具体指导建议" + ) \ No newline at end of file diff --git a/agent_system/prompter/agent.py b/agent_system/prompter/agent.py index fcd7ed2..45c18c5 100644 --- a/agent_system/prompter/agent.py +++ b/agent_system/prompter/agent.py @@ -42,11 +42,11 @@ class Prompter(BaseAgent): use_cache=False ) - def run(self, hpi_content: str, ph_content: str, chief_complaint: str, current_task: str) -> PrompterResult: + def run(self, hpi_content: str, ph_content: str, chief_complaint: str, current_task: str, specific_guidance: str = "") -> PrompterResult: """ 执行智能体提示词生成 - 基于患者的现病史、既往史、主述以及当前具体任务， + 基于患者的现病史、既往史、主述、当前具体任务以及Controller提供的具体指导建议， 生成针对该任务的专门子智能体的description和instructions。 Args: @@ -54,6 +54,7 @@ class Prompter(BaseAgent): ph_content (str): 既往史内容，患者的历史疾病信息 chief_complaint (str): 患者主述，患者的主要不适描述 current_task (str): 当前任务，如"起病情况和患病时间"、"主要症状特征"等 + specific_guidance (str): Controller提供的针对当前任务的具体指导建议，用于优化子智能体生成 Returns: PrompterResult: 包含子智能体描述和指令的结构化数据，包括： @@ -65,7 +66,7 @@ class Prompter(BaseAgent): """ try: # 构建生成提示词 - prompt = self._build_prompt(hpi_content, ph_content, chief_complaint, current_task) + prompt = self._build_prompt(hpi_content, ph_content, chief_complaint, current_task, specific_guidance) # 调用基类的run方法执行LLM推理 result = super().run(prompt) @@ -121,17 +122,18 @@ class Prompter(BaseAgent): ] ) - def _build_prompt(self, hpi_content: str, ph_content: str, chief_complaint: str, current_task: str) -> str: + def _build_prompt(self, hpi_content: str, ph_content: str, chief_complaint: str, current_task: str, specific_guidance: str = "") -> str: """ 构建Prompter的提示词模板 - 根据患者病史信息和当前任务，构建用于生成子智能体的提示词。 + 根据患者病史信息、当前任务和Controller的具体指导建议，构建用于生成子智能体的提示词。 Args: hpi_content (str): 现病史内容 ph_content (str): 既往史内容 chief_complaint (str): 患者主述 current_task (str): 当前任务 + specific_guidance (str): Controller提供的具体指导建议 Returns: str: 构建的提示词 @@ -139,6 +141,13 @@ class Prompter(BaseAgent): # 确保既往史内容的合理显示 past_history_display = ph_content.strip() if ph_content.strip() else "暂无既往史信息" + # 处理具体指导建议 + guidance_section = "" + if specific_guidance.strip(): + guidance_section = f""" +Controller指导建议: {specific_guidance} +""" + # 从prompt类获取示例输出格式 from agent_system.prompter.prompt import PrompterPrompt example_output = PrompterPrompt.get_example_output() @@ -148,9 +157,29 @@ class Prompter(BaseAgent): 现病史: {hpi_content} 既往史: {past_history_display} -当前任务: {current_task} +当前任务: {current_task}{guidance_section} -请基于上述患者信息和当前任务，生成一个专门的子智能体，该智能体将负责围绕"{current_task}"主题向患者进行专业询问。 +请按照以下步骤生成一个专门的子智能体，该智能体将负责围绕"{current_task}"主题向患者进行专业询问： + +## 步骤1: 分析任务特点 +- 深入理解"{current_task}"的核心要求和关键询问点 +- 结合患者的现病史和主述，识别与该任务相关的重要信息 +- 如果有Controller指导建议，重点考虑其中的专业建议和注意事项 + +## 步骤2: 设计智能体角色 +- 为子智能体定义专业的医疗角色和身份 +- 明确该智能体在"{current_task}"方面的专业能力和职责范围 +- 确保角色设计与患者的具体病情背景相匹配 + +## 步骤3: 制定询问策略 +- 基于任务特点和患者信息，设计系统性的询问流程 +- 将复杂的医疗询问分解为患者易于理解和回答的具体问题 +- 确保询问内容全面、有序、针对性强 + +## 步骤4: 完善执行指令 +- 详细说明子智能体应如何执行询问任务 +- 包含具体的询问技巧、注意事项和质量要求 +- 确保指令具有可操作性和实用性 请为该子智能体提供： 1. description - 描述该智能体的角色、专业领域和主要职责 diff --git a/agent_system/prompter/prompt.py b/agent_system/prompter/prompt.py index 8d90b05..38f9198 100644 --- a/agent_system/prompter/prompt.py +++ b/agent_system/prompter/prompt.py @@ -13,24 +13,43 @@ class PrompterPrompt(BasePrompt): description = ( "你是一名专业的智能体提示词生成专家，擅长基于医疗场景和具体任务需求，" "为特定的询问任务创建专门的智能体描述和指令。" - "你的主要任务是根据患者的现病史、既往史、主述以及当前具体任务，" - "生成一个针对该任务的专门子智能体的description和instructions，" + "你的主要任务是根据患者的现病史、既往史、主述、当前具体任务，" + "以及Controller智能体提供的专业指导建议，" + "按照系统化的生成流程，生成一个针对该任务的专门子智能体的description和instructions，" "该子智能体将负责围绕特定主题向患者进行专业的医疗询问。" ) # 执行指令和注意事项 instructions = [ - "## 核心生成任务", - "1. **任务理解**: 深入理解当前任务的具体要求和询问重点（如起病情况、患病时间、症状特征等）", - "2. **背景整合**: 结合患者的现病史、既往史和主述，理解患者的整体病情背景", - "3. **角色定位**: 为目标子智能体定义清晰的专业角色和询问职责", - "4. **指令设计**: 制定具体、可操作的询问指令，确保能够有效收集目标信息", + "## 系统化生成流程", + "请按照以下4个步骤进行子智能体的生成，确保生成质量和针对性：", + "", + "### 步骤1: 分析任务特点", + "- 深入理解当前任务的核心要求和关键询问点", + "- 结合患者的现病史和主述，识别与该任务相关的重要信息", + "- 重点考虑Controller指导建议中的专业建议和注意事项", + "", + "### 步骤2: 设计智能体角色", + "- 为子智能体定义专业的医疗角色和身份", + "- 明确该智能体在特定任务方面的专业能力和职责范围", + "- 确保角色设计与患者的具体病情背景相匹配", + "", + "### 步骤3: 制定询问策略", + "- 基于任务特点和患者信息，设计系统性的询问流程", + "- 将复杂的医疗询问分解为患者易于理解和回答的具体问题", + "- 确保询问内容全面、有序、针对性强", + "", + "### 步骤4: 完善执行指令", + "- 详细说明子智能体应如何执行询问任务", + "- 包含具体的询问技巧、注意事项和质量要求", + "- 确保指令具有可操作性和实用性", "", "## 子智能体设计原则", "- **专业性**: 基于医学专业知识，确保询问的科学性和准确性", "- **针对性**: 紧密围绕当前任务主题，避免偏离核心询问目标", "- **个性化**: 结合患者的具体病史背景，提供个性化的询问策略", "- **系统性**: 确保询问内容全面、有条理，不遗漏重要信息", + "- **指导整合**: 充分利用Controller提供的专业指导建议，优化询问效果", "", "## 输出内容要求", "1. **description字段**: 清晰描述子智能体的角色、专业领域和主要职责",