from cdss.models.schemas import CDSSDict, CDSSInput,CDSSInt,CDSSOutput,CDSSText from cdss.libs.cdss_helper import CDSSHelper # CDSS能力核心类,负责处理临床决策支持系统的核心逻辑 class CDSSCapability: # CDSS帮助类实例 cdss_helper: CDSSHelper = None def __init__(self): # 初始化CDSS帮助类 self.cdss_helper = CDSSHelper() # 核心处理方法,接收输入数据并返回决策支持结果 # @param input: CDSS输入数据 # @param embeding_search: 是否使用embedding搜索,默认为True # @return: CDSS输出结果 def process(self, input: CDSSInput, embeding_search:bool = True) -> CDSSOutput: start_nodes = [] # 获取主诉信息 chief_complaint = input.get_value("chief_complaint") # 初始化输出对象 output = CDSSOutput() # 如果存在主诉信息,则进行处理 if chief_complaint: for keyword in chief_complaint: # 使用帮助类进行节点搜索,查找与关键词相关的节点 results = self.cdss_helper.node_search( keyword, limit=10, node_type="word" ) for item in results: if item['score']>1.9: start_nodes.append(item['id']) print("cdss start from :", start_nodes) # 使用帮助类进行CDSS路径遍历,最大跳数为2 result = self.cdss_helper.cdss_travel(input, start_nodes,max_hops=2) for item in result["details"]: name, data = item output.diagnosis.value[name] = data print(f"{name} {data['score'] *100:.2f} %") for disease in data["diseases"]: print(f"\t疾病:{disease[0]} ") for check in data["checks"]: print(f"\t检查:{check[0]} ") for drug in data["drugs"]: print(f"\t药品:{drug[0]} ") # 输出最终推荐的检查和药品信息 print("最终推荐的检查和药品如下:") for item in result["checks"][:5]: item[1]['score'] = item[1]['count'] / result["total_checks"] output.checks.value[item[0]] = item[1] # print(f"\t检查:{item[0]} {item[1]['score'] * 100:.2f} %") # print(f"\t检查:{item[0]} {item[1]['count'] / result["total_checks"] * 100:.2f} %") for item in result["drugs"][:5]: item[1]['score'] = item[1]['count'] / result["total_checks"] output.drugs.value[item[0]] = item[1] #print(f"\t药品:{item[0]} {item[1]['count'] / result["total_drugs"] * 100:.2f} %") return output