代码提交

src/knowledge/main.py

@@ -102,7 +102,7 @@ class DiseaseInfoRequest(BaseModel):
     type: Optional[str] = None
 
 
-@app.post("/symptom_diseases", response_model=StandardResponse, operation_id="getSymptomDiseases",
+@app.post("/symptom_diseases", operation_id="getSymptomDiseases",
           summary="根据症状获取相关疾病列表",
           description="""根据输入的症状列表，查询可能相关的疾病列表。
 
@@ -119,16 +119,8 @@ class DiseaseInfoRequest(BaseModel):
 输入要求：
 - 症状名称应为标准医学术语
 
-输出格式：
-- 返回标准JSON响应
-- 包含疾病列表""",
-          response_description="""返回标准响应格式，包含疾病名称列表。
-
 成功响应示例：
-{
-    "success": true,
-    "data": ["流感", "肺炎"]
-}
+["流感", "肺炎"]
 
 错误响应示例：
 {
@@ -156,7 +148,7 @@ async def get_symptom_diseases(
         results = search.get_symptom_diseases(symptoms)
         # 直接返回疾病名称列表
         disease_names = [disease["disease_name"] for disease in results] if results else []
-        return StandardResponse(success=True, data=disease_names)
+        return disease_names
     except Exception as e:
         logger.exception(f"获取症状相关疾病失败: {str(e)}")
         raise HTTPException(500, detail=StandardResponse.error(str(e)))
@@ -333,12 +325,7 @@ async def get_auxiliary_examinations(
         ]
     }
 }
-
-错误响应示例：
-{
-    "success": false,
-    "error": "Invalid disease name"
-}"""
+"""
           )
 async def getDiseaseSymptoms(
         disease_name: str = Query(...,
@@ -360,8 +347,8 @@ async def getDiseaseSymptoms(
             
         # 调用业务逻辑获取症状
         search = SearchBusiness()
-        results = search.get_disease_symptoms(disease_name)
-        return StandardResponse(success=True, data=results)
+        results = search.get_disease_symptoms(disease_name)     
+        return StandardResponse(success=True, data = results)
     except ValueError as e:
         logger.warning(f"无效的疾病名称: {disease_name}")
         raise HTTPException(400, detail=StandardResponse.error(str(e)))
@@ -447,6 +434,7 @@ async def get_drug_indications(
         # 实现获取药物适应症的逻辑
         search = SearchBusiness()
         results = search.get_drug_indications(drug_name)
+
         return StandardResponse(success=True, data=results)
     except Exception as e:
         logger.error(f"获取药物适应症失败: {str(e)}")
@@ -535,69 +523,135 @@ async def search_concept(
         # 实现搜索医学概念的逻辑
         search = SearchBusiness()
         results = search.search_concept(name=concept_name, type=concept_type)
+
         return StandardResponse(success=True, data=results)
     except Exception as e:
         logger.error(f"搜索医学概念失败: {str(e)}")
         raise HTTPException(500, detail=StandardResponse.error(str(e)))
 
+# @app.post("/get_relations", 
+#           response_model=StandardResponse,
+#           operation_id="getConceptRelations",
+#           summary="根据概念名称查询相关关系",
+#           description="""根据概念名称查询该概念的相关关系列表。
 
-@app.post("/get_relations", 
-          response_model=StandardResponse,
-          operation_id="getConceptRelations",
-          summary="根据概念名称查询相关关系",
-          description="""根据概念名称查询该概念的相关关系列表。
+# 该接口主要用于医疗知识图谱中的关系查询场景，例如：
+# - 通过输入疾病概念名称查询相关症状
+# - 通过输入症状概念名称查询相关疾病
+# - 通过输入药物概念名称查询适应症
+
+# 典型应用场景：
+# 1. 疾病症状关系查询：输入疾病名称获取相关症状
+# 2. 症状疾病关系查询：输入症状名称获取相关疾病
+# 3. 药物适应症查询：输入药物名称获取适应症
+
+# 输入要求：
+# - 概念名称应为标准医学术语
 
-该接口主要用于医疗知识图谱中的关系查询场景，例如：
-- 通过输入疾病概念名称查询相关症状
-- 通过输入症状概念名称查询相关疾病
-- 通过输入药物概念名称查询适应症
+# 输出格式：
+# - 返回标准JSON响应
+# - 包含相关关系列表""",
+#           response_description="""返回标准响应格式，包含相关关系列表。
+
+# 成功响应示例：
+# {
+#     "success": true,
+#     "data": {
+#         "relations": [
+#             {"relation_type": "has_symptom", "target_id": "S001", "target_name": "发热"},
+#             {"relation_type": "has_symptom", "target_id": "S002", "target_name": "咳嗽"}
+#         ]
+#     }
+# }
+
+# 错误响应示例：
+# {
+#     "success": false,
+#     "error": "Invalid concept ID"
+# }"""
+#           )
+
+# 接口定义
+@app.post("/kg_relations", 
+          operation_id="query_kg_relations",
+          summary="查询知识图谱节点关系",
+          description="""根据起始节点和关系类型，查询医疗知识图谱中的关联节点。
+
+该接口用于医疗知识图谱的图关系查询场景：
+- 通过指定节点查找直接关联的节点
+- 探索医疗实体间的关系网络
+- 支持临床决策支持系统的知识检索
 
 典型应用场景：
-1. 疾病症状关系查询：输入疾病名称获取相关症状
-2. 症状疾病关系查询：输入症状名称获取相关疾病
-3. 药物适应症查询：输入药物名称获取适应症
+1. 药物查询：查找某种药物的适应症（起始节点类型=药物, 关系=相关适应症）
+2. 疾病症状：查询疾病相关症状（起始节点类型=疾病, 关系=相关症状）
+3. 治疗方案：获取疾病的推荐治疗方式（起始节点类型=治疗方案, 关系=相关治疗方式）
 
 输入要求：
-- 概念名称应为标准医学术语
+- 起始节点应为知识图谱中存在的标准实体
+- 关系类型需符合预定义的关系名称
 
-输出格式：
-- 返回标准JSON响应
-- 包含相关关系列表""",
-          response_description="""返回标准响应格式，包含相关关系列表。
+知识图谱示例结构：
+[疾病:高血压] --(相关症状)--> [症状:头痛]
+[疾病:高血压] --(相关治疗药品)--> [药物:硝苯地平]
 
-成功响应示例：
-{
-    "success": true,
-    "data": {
-        "relations": [
-            {"relation_type": "has_symptom", "target_id": "S001", "target_name": "发热"},
-            {"relation_type": "has_symptom", "target_id": "S002", "target_name": "咳嗽"}
-        ]
-    }
-}
-
-错误响应示例：
-{
-    "success": false,
-    "error": "Invalid concept ID"
-}"""
-          )
-async def get_relations(
-        concept_name: str = Query(...,
-                  description="概念ID，应为标准医学术语ID。该参数用于查询该概念的相关关系。",
-                  examples=["糖尿病"],
-                  min_length=1),
-        concept_type: str = Query(None,
-                  description="概念类型，可选参数。用于限定搜索的概念类型。",
-                  examples=["疾病", "症状", "药物"])          
+输出格式：
+[
+    "右上腹或上腹疼痛 - 症状相关诱因 - 吃了油腻食物后",
+    "右上腹或上腹疼痛 - 症状相关诱因 - 过度劳累后"
+]
+
+
+""")
+async def query_knowledge_graph_relations(
+    start_node_name: str = Query(...,
+        alias="startNode",
+        description="""起始节点名称。
+        
+示例值：
+- "高血压" (疾病)
+- "阿司匹林" (药物)
+- "头痛" (症状)""",
+        examples=["高血压", "阿司匹林"]),
+    
+    start_node_type: str = Query(...,
+        alias="startType",
+        description="""起始节点类型。
+        
+常见类型：
+- 疾病
+- 症状
+- 药物
+- 治疗方式
+- 科室""",
+        examples=["疾病", "药物"]),
+    
+    relation_name: str = Query(...,
+        alias="relation",
+        description="""关系名称（需符合预定义关系）。
+        
+常用关系：
+- 相关诱因
+- 相关症状
+- 相关治疗药物
+- 推荐治疗
+- 属于（科室）
+- 不良反应""",
+        examples=["相关诱因", "相关症状"])
 ):
     try:
-        # 实现获取概念关系的逻辑
+        # 调用业务逻辑查询知识图谱关系
         search = SearchBusiness()
-        results = search.get_relations(name=concept_name, type=concept_type)
-        return StandardResponse(success=True, data=results)
+        results = search.get_relations(name=start_node_name, type=start_node_type, relation_name=relation_name)
+        # 将结果转换为三元组格式
+        triples = []
+        for node_name, node_data in results.items():
+            for relation in node_data["relations"]:
+                triple = f"{node_name} - {relation['relation_name']} - {relation['target_name']}"
+                triples.append(triple)
+        return triples
     except Exception as e:
-        logger.error(f"获取概念关系失败: {str(e)}")
+        logger.error(f"查询知识图谱关系失败: {str(e)}")
         raise HTTPException(500, detail=StandardResponse.error(str(e)))
 
 # @app.post("/parse_medical_record",
@@ -863,18 +917,12 @@ async def suggest_appropriate_department(
         # 实现科室推荐的逻辑
         search = SearchBusiness()
         results = search.suggest_appropriate_department(disease_name=disease_name)
+       
         return StandardResponse(success=True, data=results)
     except Exception as e:
         logger.error(f"科室推荐失败: {str(e)}")
         raise HTTPException(500, detail=StandardResponse.error(str(e)))
-#
-#
-# # 5. 病历质控接口
-# class MedicalRecordRequest(BaseModel):
-#     medical_record: Dict[str, Any]
-#     validate_type: str
-#
-#
+
 
 
 # @app.post("/medical/validate_record", response_model=StandardResponse, operation_id="validateMedicalRecord",

src/knowledge/service/search_service.py

@@ -17,7 +17,7 @@ ELASTICSEARCH_PWD = config.get_config("ELASTICSEARCH_PWD", "/tmp")
 ELASTICSEARCH_HOST = config.get_config("ELASTICSEARCH_HOST", "/tmp")
 
 class SearchBusiness:
-    MIN_SCORE = 5
+    MIN_SCORE = 4
 
     def __init__(self):
         self.es = Elasticsearch(hosts=[ELASTICSEARCH_HOST], verify_certs=False, http_auth=(ELASTICSEARCH_USER, ELASTICSEARCH_PWD))
@@ -91,7 +91,7 @@ class SearchBusiness:
                 "sort": [{"_score": {"order": "desc"}}]
             }
             if name:
-                query["query"]["bool"]["must"].append({"match": {"public_kg_edges_category": name}})
+                query["query"]["bool"]["must"].append({"match": {"public_kg_edges_name": name}})
                 query["min_score"] = self.MIN_SCORE
             if src_id:
                 query["query"]["bool"]["must"].append({"term": {"public_kg_edges_src_id": src_id}})
@@ -273,7 +273,7 @@ class SearchBusiness:
         ] if nodes else []
         return {"concepts": concepts}
 
-    def get_relations(self, name, type):
+    def get_relations(self, name, type=None,relation_name=None):
         # 首先根据name和type查询nodes表数据
         nodes = self.search_nodes(name=name, type=type)
         if not nodes:
@@ -286,7 +286,7 @@ class SearchBusiness:
                 continue
 
             # 使用node id作为src_id查询edges关系表
-            edges = self.search_edges(src_id=node["public_kg_nodes_id"])
+            edges = self.search_edges(src_id=node["public_kg_nodes_id"],name=relation_name)
             if not edges:
                 continue
 
@@ -302,7 +302,7 @@ class SearchBusiness:
                 dest_id = edge["public_kg_edges_dest_id"]
                 if dest_id in node_map:
                     relations.append({
-                        "relation_type": edge["public_kg_edges_category"],
+                        "relation_name": edge["public_kg_edges_name"],
                         "target_name": node_map[dest_id].get("public_kg_nodes_name", "")
                     })