knowledge/cdss/capbility.py

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