手把手教你用 LLM 图转换器构建知识图谱：从文本到知识的智能转换

pip install transformers networkx pyvis spacypython -m spacy download en_core_web_sm

import jsonimport networkx as nxfrom transformers import pipelineimport spacy
class LLMGraphTransformer:    def __init__(self):        # 初始化NER和关系提取管道        self.ner_pipeline = pipeline(            "token-classification",             model="dslim/bert-base-NER"        )        self.relation_pipeline = pipeline(            "text2text-generation",             model="Babelscape/rebel-large"        )        self.nlp = spacy.load("en_core_web_sm")        self.graph = nx.DiGraph()            def extract_entities(self, text):        """使用LLM提取实体"""        entities = self.ner_pipeline(text)        # 处理并合并实体结果        consolidated_entities = []        current_entity = ""        current_label = ""                for entity in entities:            if entity['word'].startswith('##'):                current_entity += entity['word'][2:]            else:                if current_entity:                    consolidated_entities.append({                        'entity': current_entity,                        'label': current_label                    })                current_entity = entity['word']                current_label = entity['entity']                return consolidated_entities        def extract_relations(self, text, entities):        """使用LLM提取实体间关系"""        relation_prompt = f"""        提取以下文本中的关系：{text}        已知实体：{json.dumps(entities)}        返回JSON格式的关系列表，包含subject, relation, object        """                relations = self.relation_pipeline(relation_prompt)        return json.loads(relations[0]['generated_text'])        def build_knowledge_graph(self, text):        """构建知识图谱主方法"""        # 提取实体        entities = self.extract_entities(text)                # 提取关系        relations = self.extract_relations(text, entities)                # 构建图结构        for entity in entities:            self.graph.add_node(entity['entity'], label=entity['label'])                for relation in relations:            self.graph.add_edge(                relation['subject'],                relation['object'],                label=relation['relation']            )                return self.graph
# 使用示例transformer = LLMGraphTransformer()sample_text = "Apple Inc. was founded by Steve Jobs in California. Tim Cook is the current CEO."knowledge_graph = transformer.build_knowledge_graph(sample_text)

def entity_linking(self, entities):    """实体链接到知识库"""    linked_entities = []    for entity in entities:        # 使用Wikipedia API进行实体链接        wiki_url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{entity['entity']}"        response = requests.get(wiki_url)        if response.status_code == 200:            entity['wiki_id'] = response.json().get('pageid')            entity['description'] = response.json().get('description')        linked_entities.append(entity)    return linked_entities

def validate_relations(self, relations, text):    """验证提取的关系的可靠性"""    validated_relations = []    for relation in relations:        validation_prompt = f"""        验证以下关系是否在文本中正确：{text}        关系：{relation['subject']} - {relation['relation']} - {relation['object']}        返回JSON格式：{{"valid": boolean, "confidence": float}}        """                validation_result = self.relation_pipeline(validation_prompt)        if validation_result['valid']:            relation['confidence'] = validation_result['confidence']            validated_relations.append(relation)        return validated_relations

def visualize_graph(graph):    """可视化知识图谱"""    from pyvis.network import Network        net = Network(height="750px", width="100%", bgcolor="#222222", font_color="white")        for node in graph.nodes(data=True):        net.add_node(node[0], label=node[0], title=node[1].get('label', ''))        for edge in graph.edges(data=True):        net.add_edge(edge[0], edge[1], label=edge[2].get('label', ''))        net.show("knowledge_graph.html")

class MedicalGraphBuilder(LLMGraphTransformer):    def __init__(self):        super().__init__()        # 加载医疗领域特定模型        self.medical_ner = pipeline(            "token-classification",             model="emilyalsentzer/Bio_ClinicalBERT"        )        def extract_medical_relations(self, text):        """提取医疗领域特定关系"""        medical_template = """        从以下医疗文本中提取疾病、症状、治疗方法之间的关系：        {text}                返回JSON格式：[{            "subject": "实体1",            "relation": "关系类型",            "object": "实体2"        }]        关系类型包括：has_symptom, causes, treats, prevents        """                result = self.relation_pipeline(medical_template.format(text=text))        return json.loads(result[0]['generated_text'])
# 构建医疗知识图谱medical_builder = MedicalGraphBuilder()medical_text = "Diabetes causes increased thirst and frequent urination. Metformin treats diabetes."medical_graph = medical_builder.build_knowledge_graph(medical_text)

def incremental_building(self, new_text, existing_graph):    """增量更新知识图谱"""    new_entities = self.extract_entities(new_text)    new_relations = self.extract_relations(new_text, new_entities)        # 合并到现有图谱    for entity in new_entities:        if not existing_graph.has_node(entity['entity']):            existing_graph.add_node(entity['entity'], label=entity['label'])        for relation in new_relations:        if not existing_graph.has_edge(relation['subject'], relation['object']):            existing_graph.add_edge(                relation['subject'],                 relation['object'],                 label=relation['relation']            )        return existing_graph

def evaluate_graph_quality(self, graph, gold_standard):    """评估图谱质量"""    precision, recall, f1 = calculate_metrics(graph, gold_standard)    return {        "precision": precision,        "recall": recall,        "f1_score": f1,        "node_count": graph.number_of_nodes(),        "edge_count": graph.number_of_edges()    }

def process_large_corpus(self, corpus_path, batch_size=1000):    """处理大规模文本语料"""    graph = nx.DiGraph()        with open(corpus_path, 'r', encoding='utf-8') as f:        batch = []        for i, line in enumerate(f):            batch.append(line.strip())            if len(batch) >= batch_size:                self.process_batch(batch, graph)                batch = []        return graph

class MultilingualGraphBuilder(LLMGraphTransformer):    def __init__(self):        super().__init__()        self.multilingual_ner = pipeline(            "token-classification",            model="xlm-roberta-large"        )