0Pricing
AI Agents · 课时

知识图谱的实体提取

命名实体识别、关系提取和图谱填充

知识图谱的实体提取 是 CoddyKit 上的免费 AI Agents 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

什么是实体提取

实体提取（命名实体识别，NER）用于识别文本中的命名对象，例如人员、组织、地点和日期等。这是从非结构化文本构建知识图谱的第一步。

spaCy NER 基础

spaCy 的 en_core_web_sm 模型可以识别标准实体类型：PERSON、ORG、GPE（地缘政治实体）、DATE、MONEY 等。

import spacy

# Load English model (install: python -m spacy download en_core_web_sm)
nlp = spacy.load('en_core_web_sm')

text = 'Elon Musk founded SpaceX in 2002 in Hawthorne, California. Tesla is headquartered in Austin, Texas.'

doc = nlp(text)

for ent in doc.ents:
    print(f'{ent.text:30} {ent.label_:15} {spacy.explain(ent.label_)}')

# Output:
# Elon Musk                      PERSON          People, including fictional
# SpaceX                         ORG             Companies, agencies...
# 2002                           DATE            Absolute or relative dates
# Hawthorne, California          GPE             Countries, cities, states

将实体提取为结构化数据

将 spaCy 的实体结果转换为适合存储在知识图谱中的结构化格式。按类型对实体进行分组，并在文档内部去重。

import spacy
from collections import defaultdict

nlp = spacy.load('en_core_web_sm')

def extract_entities(text: str) -> dict:
    doc = nlp(text)
    entities = defaultdict(set)
    
    for ent in doc.ents:
        entities[ent.label_].add(ent.text.strip())
    
    # Convert sets to lists for JSON serialization
    return {k: list(v) for k, v in entities.items()}

text = 'Apple CEO Tim Cook announced new products. The event was held in Cupertino on September 12, 2023.'
result = extract_entities(text)

import json
print(json.dumps(result, indent=2))
# {
#   "ORG": ["Apple"],
#   "PERSON": ["Tim Cook"],
#   "GPE": ["Cupertino"],
#   "DATE": ["September 12, 2023"]
# }

使用 LLM 进行关系提取

spaCy 可以识别实体，但无法识别实体之间的关系。请向 LLM 提问：X 与 Y 之间是什么关系？，以提取知识图谱中的边。

import openai
import json

client = openai.OpenAI(api_key='sk-...')

def extract_relations(text: str, entities: list) -> list:
    if len(entities) < 2:
        return []
    
    entity_list = ', '.join(entities)
    prompt = (
        f'Given this text and these entities: {entity_list}\n\n'
        f'Text: {text}\n\n'
        'Extract relationships between the entities. '
        'Return a JSON array of objects with fields: '
        'subject (string), relation (string), object (string). '
        'Use concise relation labels like FOUNDED_BY, WORKS_AT, LOCATED_IN, ACQUIRED_BY.'
    )
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    result = json.loads(response.choices[0].message.content)
    return result.get('relations', [])

text = 'Elon Musk founded SpaceX in Hawthorne, California.'
entities = ['Elon Musk', 'SpaceX', 'Hawthorne']
relations = extract_relations(text, entities)
print(relations)
# [{'subject': 'Elon Musk', 'relation': 'FOUNDED', 'object': 'SpaceX'},
#  {'subject': 'SpaceX', 'relation': 'LOCATED_IN', 'object': 'Hawthorne'}]

实体规范化

同一实体可能以不同的表面形式出现：Elon Musk、Musk、E. Musk。在将实体添加到图谱之前，规范化会将这些形式映射到规范形式。

import openai
import json

client = openai.OpenAI(api_key='sk-...')

def normalize_entities(entity_mentions: list, entity_type: str) -> dict:
    '''
    Groups variations of the same entity together.
    Returns {canonical_name: [mention1, mention2, ...]}
    '''
    if len(entity_mentions) <= 1:
        return {m: [m] for m in entity_mentions}
    
    mentions_str = json.dumps(entity_mentions)
    prompt = (
        f'These are {entity_type} entity mentions from text: {mentions_str}\n'
        'Group mentions that refer to the same entity. '
        'Return JSON: {"groups": [["canonical", "alias1", "alias2"], ...]}'
    )
    response = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    result = json.loads(response.choices[0].message.content)
    
    normalized = {}
    for group in result.get('groups', []):
        if group:
            canonical = group[0]
            for mention in group:
                normalized[mention] = canonical
    return normalized

mentions = ['Elon Musk', 'Musk', 'E. Musk', 'Tim Cook']
result = normalize_entities(mentions, 'PERSON')
print(result)

实体去重

插入知识图谱之前，请检查实体是否已经存在。使用模糊匹配或规范 ID，将来自多个文档来源的重复实体合并。

from difflib import SequenceMatcher

class EntityRegistry:
    def __init__(self, similarity_threshold=0.85):
        self.entities = {}  # {canonical_name: entity_data}
        self.threshold = similarity_threshold
    
    def similarity(self, a: str, b: str) -> float:
        return SequenceMatcher(None, a.lower(), b.lower()).ratio()
    
    def find_existing(self, name: str) -> str:
        for canonical in self.entities:
            if self.similarity(name, canonical) >= self.threshold:
                return canonical
        return None
    
    def add_entity(self, name: str, entity_type: str, metadata: dict = None) -> str:
        existing = self.find_existing(name)
        if existing:
            print(f'Merged "{name}" -> "{existing}"')
            return existing
        self.entities[name] = {'type': entity_type, 'metadata': metadata or {}}
        print(f'New entity: "{name}"')
        return name

registry = EntityRegistry()
registry.add_entity('OpenAI', 'ORG', {})
registry.add_entity('Open AI', 'ORG', {})   # Merged
registry.add_entity('OpenAI Inc', 'ORG', {})  # Merged
registry.add_entity('Microsoft', 'ORG', {})

处理多个文档

从语料库构建知识图谱时，请按顺序处理文档，并累积实体和关系。记录每条事实来自哪个文档（溯源信息）。

import spacy
from typing import List, Dict

nlp = spacy.load('en_core_web_sm')

@dataclass
class KGFact:
    subject: str
    relation: str
    obj: str
    source_doc: str

def process_corpus(documents: List[Dict]) -> List:
    facts = []
    entity_registry = EntityRegistry()
    
    for doc in documents:
        doc_id = doc['id']
        text = doc['text']
        
        # Extract entities
        spacy_doc = nlp(text)
        persons = [ent.text for ent in spacy_doc.ents if ent.label_ == 'PERSON']
        orgs = [ent.text for ent in spacy_doc.ents if ent.label_ == 'ORG']
        
        # Register entities (deduplication)
        for p in persons:
            entity_registry.add_entity(p, 'PERSON')
        for o in orgs:
            entity_registry.add_entity(o, 'ORG')
        
        print(f'Processed doc {doc_id}: {len(persons)} persons, {len(orgs)} orgs')
    
    return entity_registry.entities

docs = [
    {'id': 'doc1', 'text': 'Sundar Pichai leads Google.'},
    {'id': 'doc2', 'text': 'Google CEO Sundar Pichai spoke at the conference.'}
]
entities = process_corpus(docs)

知识图谱的实体类型

在构建图谱之前，请设计实体分类体系。业务知识图谱中的常见类型包括：Person、Organization、Product、Location、Event、Technology。自定义类型取决于您的业务领域。

ENTITY_TYPES = {
    'PERSON': {
        'description': 'Individual human being',
        'properties': ['name', 'title', 'email'],
        'spacy_labels': ['PERSON']
    },
    'ORGANIZATION': {
        'description': 'Company, institution, or group',
        'properties': ['name', 'industry', 'founded'],
        'spacy_labels': ['ORG']
    },
    'LOCATION': {
        'description': 'Geographic place',
        'properties': ['name', 'country', 'coordinates'],
        'spacy_labels': ['GPE', 'LOC', 'FAC']
    },
    'PRODUCT': {
        'description': 'Product or service (custom, not in spaCy)',
        'properties': ['name', 'category', 'version'],
        'spacy_labels': ['PRODUCT']
    }
}

# Map spaCy labels to our types
def map_spacy_to_entity_type(spacy_label: str) -> str:
    for entity_type, config in ENTITY_TYPES.items():
        if spacy_label in config['spacy_labels']:
            return entity_type
    return 'UNKNOWN'

print(map_spacy_to_entity_type('ORG'))     # ORGANIZATION
print(map_spacy_to_entity_type('GPE'))     # LOCATION
print(map_spacy_to_entity_type('PERSON'))  # PERSON

结合使用 spaCy 和 LLM

使用 spaCy 快速、低成本地检测实体，仅在更困难的任务中使用 LLM，例如关系提取、实体消歧，以及处理 spaCy 遗漏的领域特定实体。

import spacy
import openai

nlp = spacy.load('en_core_web_sm')
client = openai.OpenAI(api_key='sk-...')

def full_extraction_pipeline(text: str) -> dict:
    # Step 1: Fast spaCy extraction
    doc = nlp(text)
    entities = [
        {'text': ent.text, 'type': ent.label_}
        for ent in doc.ents
        if ent.label_ in ['PERSON', 'ORG', 'GPE', 'PRODUCT']
    ]
    
    if len(entities) < 2:
        return {'entities': entities, 'relations': []}
    
    # Step 2: LLM relation extraction (only when we have multiple entities)
    entity_names = [e['text'] for e in entities]
    relations = extract_relations(text, entity_names)
    
    return {
        'entities': entities,
        'relations': relations
    }

text = 'Satya Nadella of Microsoft acquired Activision Blizzard for $69 billion.'
result = full_extraction_pipeline(text)
print('Entities:', result['entities'])
print('Relations:', result['relations'])

验证提取的数据

在将提取出的实体存储到知识图谱之前，请先进行验证。检查主语和宾语实体是否已知，关系标签是否来自批准的词汇表，以及数据是否完整。

VALID_RELATIONS = {
    'FOUNDED_BY', 'WORKS_AT', 'LOCATED_IN', 'ACQUIRED_BY',
    'PARTNERED_WITH', 'INVESTED_IN', 'CEO_OF', 'PRODUCT_OF'
}

def validate_relation(relation: dict, known_entities: set) -> tuple:
    errors = []
    subject = relation.get('subject', '')
    relation_label = relation.get('relation', '')
    obj = relation.get('object', '')
    
    if not subject:
        errors.append('Missing subject')
    if not obj:
        errors.append('Missing object')
    if relation_label not in VALID_RELATIONS:
        errors.append(f'Unknown relation: {relation_label}')
    if subject and subject not in known_entities:
        errors.append(f'Unknown entity: {subject}')
    if obj and obj not in known_entities:
        errors.append(f'Unknown entity: {obj}')
    
    return len(errors) == 0, errors

known = {'Elon Musk', 'SpaceX', 'California'}
relation = {'subject': 'Elon Musk', 'relation': 'FOUNDED_BY', 'object': 'SpaceX'}
valid, errors = validate_relation(relation, known)
print('Valid:', valid, 'Errors:', errors)

增量构建图谱

随着新文档到达，逐步构建知识图谱。处理每个文档，提取实体和关系，进行去重与验证，然后将其写入图数据库。

def build_knowledge_graph_incremental(new_documents: list, graph_db, entity_registry):
    for doc in new_documents:
        print(f'Processing document: {doc["id"]}')
        
        # Extract
        extraction = full_extraction_pipeline(doc['text'])
        
        # Register and deduplicate entities
        canonical_entities = {}
        for entity in extraction['entities']:
            canonical = entity_registry.add_entity(
                entity['text'],
                entity['type']
            )
            canonical_entities[entity['text']] = canonical
            # Upsert node in graph
            graph_db.upsert_node(canonical, entity['type'])
        
        # Validate and insert relations
        for relation in extraction['relations']:
            # Map to canonical names
            subj = canonical_entities.get(relation['subject'], relation['subject'])
            obj = canonical_entities.get(relation['object'], relation['object'])
            valid, errors = validate_relation(
                {'subject': subj, 'relation': relation['relation'], 'object': obj},
                set(canonical_entities.values())
            )
            if valid:
                graph_db.upsert_edge(subj, relation['relation'], obj, doc['id'])
            else:
                print(f'Skipping invalid relation: {errors}')

print('Graph build pipeline defined')

知识检查：实体提取

请测试您对知识图谱实体提取的理解。

实体提取总结

完整的知识图谱实体提取流水线应结合以下部分：用于快速检测实体的 spaCy NER、基于 LLM 的关系提取、用于将表面形式映射到规范名称的实体规范化、用于避免冗余节点的去重、插入前的验证，以及用于追踪每条事实来源文档的溯源信息。

常见问题解答

「知识图谱的实体提取」课时是免费的吗？

是的 — 「知识图谱的实体提取」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「知识图谱的实体提取」这节课中我会学到什么？

命名实体识别、关系提取和图谱填充 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。

「知识图谱的实体提取」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 知识图谱的实体提取
  2. 通过智能体工具查询 Neo4j
  3. 结合向量检索与图谱检索
  4. 构建知识增强型智能体
← 返回 AI Agents