Class GraphIndexingModule

rag_modules/graph_indexing.py:38–374

图索引模块 核心功能： 1. 为实体创建键值对（名称作为唯一索引键） 2. 为关系创建键值对（多个索引键，包含全局主题） 3. 去重和优化图操作 4. 支持增量更新

    metadata: Dict[str, Any]

class GraphIndexingModule:
    """
    图索引模块
    核心功能：
    1. 为实体创建键值对（名称作为唯一索引键）
    2. 为关系创建键值对（多个索引键，包含全局主题）
    3. 去重和优化图操作
    4. 支持增量更新
    """
    
    def __init__(self, config, llm_client):
        self.config = config
        self.llm_client = llm_client
        
        # 键值对存储
        self.entity_kv_store: Dict[str, EntityKeyValue] = {}
        self.relation_kv_store: Dict[str, RelationKeyValue] = {}
        
        # 索引映射：key -> entity/relation IDs
        self.key_to_entities: Dict[str, List[str]] = defaultdict(list)
        self.key_to_relations: Dict[str, List[str]] = defaultdict(list)
        
    def create_entity_key_values(self, diseases: List[Any], symptoms: List[Any],
                                treatments: List[Any], medications: List[Any],
                                risk_factors: List[Any], care_tips: List[Any]) -> Dict[str, EntityKeyValue]:
        """
        为实体创建键值对结构
        每个实体使用其名称作为唯一索引键
        """
        logger.info("开始创建实体键值对...")
        
        label_display = {
            "Disease": "疾病",
            "Symptom": "症状",
            "Treatment": "治疗方案",
            "Medication": "药物",
            "RiskFactor": "风险因素",
            "CareTip": "护理建议"
        }

        def _store_entity(entity_list, entity_label, aliases=None):
            for entity in entity_list:
                entity_id = entity.node_id
                entity_name = entity.name or f"{entity_label}_{entity_id}"
                
                props = getattr(entity, "properties", {})
                display_label = label_display.get(entity_label, entity_label)
                content_parts = [f"{display_label}名称: {entity_name}"]
                if props.get("description"):
                    content_parts.append(f"描述: {props['description']}")
                if props.get("category"):
                    content_parts.append(f"分类: {props['category']}")
                if props.get("severity"):
                    content_parts.append(f"严重程度: {props['severity']}")
                if props.get("tags"):
                    content_parts.append(f"标签: {props['tags']}")
                if props.get("methods"):
                    content_parts.append(f"方法: {props['methods']}")