📖 第8章：信息抽取¶
学习时间：8小时 难度星级：⭐⭐⭐⭐ 前置知识：序列标注、文本分类、预训练模型 学习目标：掌握关系抽取、事件抽取和知识图谱构建基础
📋 目录¶
1. 信息抽取概述¶
1.1 什么是信息抽取¶
信息抽取（Information Extraction, IE）从非结构化文本中自动提取结构化信息。
Python
ie_tasks = {
"实体识别(NER)": {
"描述": "识别文本中的实体及其类型",
"例子": "[乔布斯/PER]在[加利福尼亚/LOC]创立了[苹果公司/ORG]",
"参考": "第5章已详细讲解",
},
"关系抽取(RE)": {
"描述": "识别实体之间的语义关系",
"例子": "(乔布斯, 创立, 苹果公司), (乔布斯, 出生于, 加利福尼亚)",
},
"事件抽取(EE)": {
"描述": "识别事件及其参与者、时间、地点等",
"例子": "事件:创立 | 主体:乔布斯 | 客体:苹果公司 | 时间:1976年 | 地点:车库",
},
"属性抽取": {
"描述": "提取实体的属性信息",
"例子": "(苹果公司, 成立时间, 1976年)",
},
}
ie_pipeline = """
信息抽取Pipeline:
原始文本 → NER → 关系抽取 → 事件抽取 → 知识图谱
│ │ │ │ │
│ 识别实体 实体间关系 事件结构化 结构化存储
│ │
└──────── 非结构化 ───────────→ 结构化 ──┘
"""
print(ie_pipeline)
2. 关系抽取¶
2.1 关系抽取的方法¶
Text Only
关系抽取方法分类:
├── Pipeline方法
│ ├── 先做NER，再对实体对做关系分类
│ └── 优点：模块化，缺点：误差传递
│
├── 联合抽取方法
│ ├── 同时抽取实体和关系
│ └── 优点：避免误差传递，缺点：更复杂
│
├── 远程监督方法
│ ├── 用知识库自动标注语料
│ └── 优点：无需人工标注，缺点：标签噪声
│
└── 大模型方法
├── 用Prompt让LLM直接抽取
└── 优点：零样本能力，缺点：效率低
2.2 基于分类的关系抽取（Pipeline）¶
Python
import torch
import torch.nn as nn
import numpy as np
class RelationClassifier(nn.Module): # 继承nn.Module定义网络层
"""基于BERT的关系分类器"""
def __init__(self, hidden_dim, num_relations, dropout=0.3):
super().__init__() # super()调用父类方法
# 实体标记嵌入（标记实体位置）
self.entity_fc = nn.Linear(hidden_dim * 2, hidden_dim)
self.classifier = nn.Sequential(
nn.Linear(hidden_dim, hidden_dim // 2),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_dim // 2, num_relations),
)
def forward(self, entity1_repr, entity2_repr):
"""
entity1_repr: 第一个实体的表示 (batch, hidden)
entity2_repr: 第二个实体的表示 (batch, hidden)
"""
# 拼接两个实体表示
combined = torch.cat([entity1_repr, entity2_repr], dim=-1) # torch.cat沿已有维度拼接张量
combined = torch.relu(self.entity_fc(combined))
logits = self.classifier(combined)
return logits
# 关系定义
relation_types = {
0: "无关系",
1: "创立",
2: "就职于",
3: "出生于",
4: "位于",
5: "毕业于",
6: "父/母亲",
7: "配偶",
}
# 简单演示
hidden_dim = 256
model = RelationClassifier(hidden_dim, len(relation_types))
# 模拟实体表示
e1 = torch.randn(1, hidden_dim)
e2 = torch.randn(1, hidden_dim)
logits = model(e1, e2)
pred = logits.argmax(dim=-1).item() # 将单元素张量转为Python数值
print(f"预测关系: {relation_types[pred]}")
2.3 基于标注的联合抽取¶
Python
class JointExtractionModel:
"""
联合实体关系抽取（标注方案）
使用特殊标注体系将关系抽取转化为序列标注：
对于三元组(subject, relation, object)：
- 对subject用 B-SUB_关系类型, I-SUB_关系类型 标注
- 对object用 B-OBJ_关系类型, I-OBJ_关系类型 标注
"""
def __init__(self):
self.relations = ["创立", "就职于", "出生于", "位于"]
def generate_tags(self, text, triples):
"""为文本生成标签"""
chars = list(text)
tags = ["O"] * len(chars)
for subj, rel, obj in triples:
# 标注subject
subj_start = text.find(subj)
if subj_start >= 0:
tags[subj_start] = f"B-SUB_{rel}"
for i in range(1, len(subj)):
tags[subj_start + i] = f"I-SUB_{rel}"
# 标注object
obj_start = text.find(obj)
if obj_start >= 0:
tags[obj_start] = f"B-OBJ_{rel}"
for i in range(1, len(obj)):
tags[obj_start + i] = f"I-OBJ_{rel}"
return list(zip(chars, tags)) # zip按位置配对
def decode_triples(self, chars, tags):
"""从标签序列解码三元组"""
subjects = {} # rel -> subject text
objects = {} # rel -> object text
current_text = ""
current_type = None
current_rel = None
for char, tag in zip(chars, tags):
if tag.startswith("B-SUB_") or tag.startswith("B-OBJ_"):
# 保存之前的
if current_text and current_type and current_rel:
if current_type == "SUB":
subjects[current_rel] = current_text
else:
objects[current_rel] = current_text
parts = tag.split("_", 1)
current_type = "SUB" if "SUB" in parts[0] else "OBJ"
current_rel = parts[1]
current_text = char
elif tag.startswith("I-"):
current_text += char
else:
if current_text and current_type and current_rel:
if current_type == "SUB":
subjects[current_rel] = current_text
else:
objects[current_rel] = current_text
current_text = ""
current_type = None
current_rel = None
# 最后一个
if current_text and current_type and current_rel:
if current_type == "SUB":
subjects[current_rel] = current_text
else:
objects[current_rel] = current_text
# 组合三元组
triples = []
for rel in set(subjects.keys()) & set(objects.keys()):
triples.append((subjects[rel], rel, objects[rel]))
return triples
# 测试
je = JointExtractionModel()
text = "乔布斯在加利福尼亚创立了苹果公司"
triples = [("乔布斯", "创立", "苹果公司")]
tagged = je.generate_tags(text, triples)
print("联合抽取标注:")
for char, tag in tagged:
if tag != "O":
print(f" {char} → {tag}")
# 解码
chars = [c for c, _ in tagged]
tags = [t for _, t in tagged]
decoded = je.decode_triples(chars, tags)
print(f"\n解码三元组: {decoded}")
3. 事件抽取¶
3.1 事件抽取任务¶
Python
"""
事件抽取层次:
1. 事件检测(Event Detection): 判断句子中是否存在事件，识别触发词
2. 论元抽取(Argument Extraction): 识别事件的参与者和元素
"""
event_example = {
"文本": "2024年3月15日，马斯克以440亿美元收购了Twitter公司",
"事件类型": "收购",
"触发词": "收购",
"论元": {
"时间": "2024年3月15日",
"收购方": "马斯克",
"金额": "440亿美元",
"被收购方": "Twitter公司",
},
}
# 事件类型体系
event_schema = {
"商业事件": ["收购", "合并", "上市", "融资", "破产"],
"人事事件": ["任职", "离职", "退休", "就职"],
"司法事件": ["起诉", "判决", "逮捕", "释放"],
"灾害事件": ["地震", "火灾", "洪水", "台风"],
"军事事件": ["冲突", "谈判", "协议"],
}
print("事件抽取示例:")
print(f" 文本: {event_example['文本']}")
print(f" 事件类型: {event_example['事件类型']}")
print(f" 触发词: {event_example['触发词']}")
print(f" 论元:")
for role, value in event_example['论元'].items():
print(f" {role}: {value}")
3.2 事件抽取实现¶
Python
import re
class SimpleEventExtractor:
"""基于规则和模式的事件抽取器"""
def __init__(self):
self.event_patterns = {
"收购": {
"trigger": re.compile(r"收购|并购|买下"),
"args": {
"收购方": re.compile(r"([\u4e00-\u9fa5·]+(?:公司|集团|科技)?)(?:以|用|花)"),
"被收购方": re.compile(r"收购(?:了)?([\u4e00-\u9fa5a-zA-Z]+(?:公司|集团)?)"),
"金额": re.compile(r"(\d+(?:\.\d+)?(?:万|亿)?(?:美元|元|人民币))"),
"时间": re.compile(r"(\d{4}年\d{1,2}月\d{1,2}日)"),
},
},
"任职": {
"trigger": re.compile(r"就任|担任|出任|上任"),
"args": {
"人物": re.compile(r"([\u4e00-\u9fa5]{2,4})(?:就任|担任|出任)"),
"职位": re.compile(r"(?:就任|担任|出任)([\u4e00-\u9fa5]+)"),
},
},
}
def extract(self, text):
"""从文本中抽取事件"""
events = []
for event_type, pattern_info in self.event_patterns.items():
trigger_match = pattern_info["trigger"].search(text)
if trigger_match:
event = {
"event_type": event_type,
"trigger": trigger_match.group(),
"trigger_span": (trigger_match.start(), trigger_match.end()),
"arguments": {},
}
for arg_role, arg_pattern in pattern_info["args"].items():
arg_match = arg_pattern.search(text)
if arg_match:
event["arguments"][arg_role] = arg_match.group(1)
events.append(event)
return events
# 测试
ee = SimpleEventExtractor()
texts = [
"2024年3月15日，马斯克以440亿美元收购了Twitter公司",
"张三就任华为公司首席技术官",
]
for text in texts:
events = ee.extract(text)
print(f"\n文本: {text}")
for event in events:
print(f" 事件类型: {event['event_type']}")
print(f" 触发词: {event['trigger']}")
for role, value in event['arguments'].items():
print(f" {role}: {value}")
4. 开放信息抽取¶
4.1 Open IE¶
Python
class OpenIE:
"""开放信息抽取：不限定关系类型"""
def __init__(self):
self.patterns = [
# 主语 + 谓语 + 宾语
re.compile(r'([\u4e00-\u9fa5]+(?:公司|大学|集团|人|先生|女士)?)'
r'(创立|发明|创建|开发|研究|发现|提出|设计|生产|制造)'
r'了?([\u4e00-\u9fa5a-zA-Z]+)'),
# 主语 + 是 + 宾语
re.compile(r'([\u4e00-\u9fa5]+)'
r'是'
r'([\u4e00-\u9fa5]+的[\u4e00-\u9fa5]+|[\u4e00-\u9fa5]+)'),
# 主语 + 位于/在 + 地点
re.compile(r'([\u4e00-\u9fa5]+)'
r'(?:位于|坐落于|在)'
r'([\u4e00-\u9fa5]+)'),
]
def extract(self, text):
"""抽取开放三元组"""
triples = []
for pattern in self.patterns:
matches = pattern.finditer(text)
for match in matches:
groups = match.groups()
if len(groups) == 3:
triples.append({
"subject": groups[0],
"predicate": groups[1],
"object": groups[2],
})
elif len(groups) == 2:
triples.append({
"subject": groups[0],
"predicate": "是/在",
"object": groups[1],
})
return triples
# 测试
oie = OpenIE()
texts = [
"乔布斯创立了苹果公司",
"北京大学位于北京市海淀区",
"华为公司研究5G技术",
]
for text in texts:
triples = oie.extract(text)
print(f"\n'{text}':")
for t in triples:
print(f" ({t['subject']}, {t['predicate']}, {t['object']})")
5. 知识图谱构建¶
5.1 知识图谱基础¶
Python
class KnowledgeGraph:
"""简易知识图谱"""
def __init__(self):
self.entities = {} # 实体: {类型, 属性}
self.relations = [] # (头实体, 关系, 尾实体)
def add_entity(self, name, entity_type, attributes=None):
self.entities[name] = {
"type": entity_type,
"attributes": attributes or {},
}
def add_relation(self, head, relation, tail):
self.relations.append((head, relation, tail))
def query(self, entity=None, relation=None):
"""简单查询"""
results = []
for h, r, t in self.relations:
if entity and (h == entity or t == entity):
results.append((h, r, t))
elif relation and r == relation:
results.append((h, r, t))
return results
def get_neighbors(self, entity, hops=1):
"""获取实体的邻居"""
visited = {entity}
current_level = {entity}
all_triples = []
for _ in range(hops):
next_level = set()
for h, r, t in self.relations:
if h in current_level and t not in visited:
next_level.add(t)
all_triples.append((h, r, t))
elif t in current_level and h not in visited:
next_level.add(h)
all_triples.append((h, r, t))
visited |= next_level
current_level = next_level
return all_triples
def stats(self):
return {
"entities": len(self.entities),
"relations": len(self.relations),
"relation_types": len(set(r for _, r, _ in self.relations)),
}
def to_triples(self):
"""导出为三元组列表"""
return [(h, r, t) for h, r, t in self.relations]
# 构建知识图谱
kg = KnowledgeGraph()
# 添加实体
kg.add_entity("苹果公司", "ORG", {"成立年份": "1976", "总部": "库比蒂诺"})
kg.add_entity("乔布斯", "PER", {"出生年份": "1955", "国籍": "美国"})
kg.add_entity("库克", "PER", {"出生年份": "1960", "国籍": "美国"})
kg.add_entity("加利福尼亚", "LOC", {"国家": "美国"})
kg.add_entity("iPhone", "PRODUCT")
kg.add_entity("斯坦福大学", "ORG")
# 添加关系
kg.add_relation("乔布斯", "创立", "苹果公司")
kg.add_relation("乔布斯", "出生于", "加利福尼亚")
kg.add_relation("库克", "就职于", "苹果公司")
kg.add_relation("库克", "担任CEO", "苹果公司")
kg.add_relation("苹果公司", "生产", "iPhone")
kg.add_relation("苹果公司", "位于", "加利福尼亚")
kg.add_relation("乔布斯", "辍学于", "斯坦福大学")
# 查询
print("知识图谱统计:", kg.stats())
print("\n查询'乔布斯'的关系:")
for h, r, t in kg.query(entity="乔布斯"):
print(f" ({h}, {r}, {t})")
print("\n查询'苹果公司'的1跳邻居:")
for h, r, t in kg.get_neighbors("苹果公司", hops=1):
print(f" ({h}, {r}, {t})")
print("\n查询'苹果公司'的2跳邻居:")
for h, r, t in kg.get_neighbors("苹果公司", hops=2):
print(f" ({h}, {r}, {t})")
5.2 知识图谱嵌入¶
Python
class TransE(nn.Module):
"""TransE知识图谱嵌入模型
核心思想：h + r ≈ t
"""
def __init__(self, num_entities, num_relations, embedding_dim=50, margin=1.0):
super().__init__()
self.entity_embeddings = nn.Embedding(num_entities, embedding_dim)
self.relation_embeddings = nn.Embedding(num_relations, embedding_dim)
self.margin = margin
# 初始化
nn.init.xavier_uniform_(self.entity_embeddings.weight)
nn.init.xavier_uniform_(self.relation_embeddings.weight)
def forward(self, h, r, t):
"""计算三元组得分: ||h + r - t||"""
h_emb = self.entity_embeddings(h)
r_emb = self.relation_embeddings(r)
t_emb = self.entity_embeddings(t)
# L2距离
score = torch.norm(h_emb + r_emb - t_emb, p=2, dim=-1)
return score
def loss(self, pos_h, pos_r, pos_t, neg_h, neg_r, neg_t):
"""Margin-based loss"""
pos_score = self.forward(pos_h, pos_r, pos_t)
neg_score = self.forward(neg_h, neg_r, neg_t)
loss = torch.relu(self.margin + pos_score - neg_score).mean()
return loss
# 创建TransE模型
transe = TransE(num_entities=100, num_relations=10, embedding_dim=50)
print(f"TransE参数量: {sum(p.numel() for p in transe.parameters()):,}")
6. 远程监督¶
Python
class DistantSupervision:
"""远程监督关系抽取
核心假设：如果知识库中存在关系(e1, r, e2)，
那么包含e1和e2的句子都表达了关系r
"""
def __init__(self):
# 知识库中的事实
self.kb_facts = {
("乔布斯", "创立", "苹果公司"),
("马化腾", "创立", "腾讯"),
("马云", "创立", "阿里巴巴"),
("任正非", "创立", "华为"),
}
def auto_label(self, sentences):
"""自动标注语料"""
labeled_data = []
for sent in sentences:
for e1, rel, e2 in self.kb_facts:
if e1 in sent and e2 in sent:
labeled_data.append({
"sentence": sent,
"entity1": e1,
"entity2": e2,
"relation": rel,
"confidence": "distant", # 远程监督标注
})
return labeled_data
# 测试
ds = DistantSupervision()
sentences = [
"乔布斯与好友沃兹尼亚克一起在车库创立了苹果公司",
"在乔布斯的带领下苹果公司推出了革命性的产品",
"马化腾在深圳创立了腾讯公司",
"马云退休后辞去了阿里巴巴董事长的职务",
]
labeled = ds.auto_label(sentences)
print("远程监督自动标注结果:")
for item in labeled:
print(f"\n 句子: {item['sentence']}")
print(f" 三元组: ({item['entity1']}, {item['relation']}, {item['entity2']})")
print(f" 标注来源: {item['confidence']}")
print("\n⚠️ 远程监督的问题:")
print(" - '马云退休后辞去了阿里巴巴董事长的职务'被标注为'创立'关系 → 标签噪声！")
print(" - 解决：多实例学习(MIL)、降噪方法、注意力机制选择可靠句子")
7. 实战：构建领域知识图谱¶
Python
"""
实战项目：从新闻文本构建科技领域知识图谱
"""
class TechKGBuilder:
"""科技领域知识图谱构建器"""
def __init__(self):
self.kg = KnowledgeGraph()
self.ner_extractor = self._build_ner()
self.re_extractor = self._build_re()
def _build_ner(self):
"""构建NER提取器（基于规则）"""
import re
patterns = {
"ORG": re.compile(r'([\u4e00-\u9fa5]+(?:公司|集团|科技|研究院|大学|实验室))'),
"PER": re.compile(r'([\u4e00-\u9fa5]{2,4})(?:表示|说|认为|指出|创立|担任|就任)'),
"PRODUCT": re.compile(r'([\u4e00-\u9fa5]*[a-zA-Z]+[\u4e00-\u9fa5a-zA-Z0-9]*)'),
}
return patterns
def _build_re(self):
"""构建关系提取器"""
import re
patterns = {
"创立": re.compile(r'([\u4e00-\u9fa5]+)创立了?([\u4e00-\u9fa5]+(?:公司|集团))'),
"发布": re.compile(r'([\u4e00-\u9fa5]+(?:公司|集团))发布了?([\u4e00-\u9fa5a-zA-Z0-9]+)'),
"收购": re.compile(r'([\u4e00-\u9fa5]+(?:公司|集团))收购了?([\u4e00-\u9fa5a-zA-Z]+(?:公司)?)'),
"投资": re.compile(r'([\u4e00-\u9fa5]+(?:公司|集团))投资了?([\u4e00-\u9fa5a-zA-Z]+)'),
}
return patterns
def process_text(self, text):
"""处理单条文本，提取知识"""
extracted = {"entities": [], "relations": []}
# NER
for etype, pattern in self.ner_extractor.items():
for match in pattern.finditer(text):
entity = match.group(1)
if len(entity) >= 2:
extracted["entities"].append({"name": entity, "type": etype})
self.kg.add_entity(entity, etype)
# 关系抽取
for rel_type, pattern in self.re_extractor.items():
for match in pattern.finditer(text):
head = match.group(1)
tail = match.group(2)
extracted["relations"].append((head, rel_type, tail))
self.kg.add_relation(head, rel_type, tail)
return extracted
def build_from_texts(self, texts):
"""从文本集合构建知识图谱"""
all_extracted = []
for text in texts:
result = self.process_text(text)
all_extracted.append(result)
return all_extracted
# 构建知识图谱
builder = TechKGBuilder()
news_texts = [
"马斯克创立了SpaceX公司和特斯拉公司",
"苹果公司发布了iPhone15和MacBook新品",
"微软公司收购了暴雪公司",
"腾讯公司投资了快手科技",
]
results = builder.build_from_texts(news_texts)
print("="*50)
print("科技领域知识图谱构建结果")
print("="*50)
for text, result in zip(news_texts, results):
print(f"\n📰 {text}")
if result["entities"]:
print(f" 实体: {[(e['name'], e['type']) for e in result['entities']]}")
if result["relations"]:
print(f" 关系: {result['relations']}")
print(f"\n📊 知识图谱统计: {builder.kg.stats()}")
print(f"\n所有三元组:")
for h, r, t in builder.kg.to_triples():
print(f" ({h}, {r}, {t})")
8. 面试要点¶
🔑 面试高频考点
考点1：Pipeline vs 联合抽取的优缺点？¶
Text Only
✅ 标准答案要点：
Pipeline (分步):
- 流程：先NER → 再对实体对做关系分类
- 优点：模块化，每个子任务可以独立优化
- 缺点：误差传递（NER错误会影响RE）
联合抽取 (Joint):
- 流程：同时抽取实体和关系
- 优点：避免误差传递，实体和关系互相增强
- 缺点：模型更复杂，训练更难
实际选择：
- 数据少/类型简单 → Pipeline
- 追求最佳效果 → 联合抽取
- 零样本/少样本 → 大模型Prompt
考点2：远程监督的噪声问题如何解决？¶
Text Only
✅ 标准答案要点：
1. 多实例学习(MIL)：将同一实体对的多个句子看作一个bag
2. 注意力降噪：AT-LEAST-ONE假设 + 句子级注意力选择可靠句子
3. 强化学习选择：用RL agent选择有用训练句子
4. 知识蒸馏：用大模型标注数据替代远程监督
考点3：知识图谱有哪些下游应用？¶
Text Only
✅ 标准答案要点：
1. 知识问答(KBQA)：基于知识图谱回答问题
2. 推荐系统：利用图谱中的关系增强推荐
3. 搜索引擎：Google Knowledge Panel
4. 辅助NLP任务：实体链接、指代消解
5. RAG增强：为大模型提供结构化知识
9. 练习题¶
📝 基础题¶
- 解释关系抽取中Pipeline方法和联合方法的区别。
答案：Pipeline方法：分步串行——先NER识别实体，再对实体对做关系分类。优点是各模块独立训练、灵活可替换；缺点是误差传播（NER错误传递到关系抽取）、无法利用任务间交互信息。联合方法：一个模型同时学习实体识别和关系抽取，共享底层表示。优点是能捕捉实体和关系的相互依赖（如知道有"出生于"关系有助于判断实体类型），减少误差传播；缺点是模型设计复杂、训练难度大。目前联合方法是主流趋势。
- 什么是远程监督？它的核心假设和局限性是什么？
答案：远程监督利用已有知识库（如Wikidata）对齐原始文本来自动标注关系训练数据。核心假设：若知识库中存在三元组(实体A, 关系r, 实体B)，则任何同时提到A和B的句子都表达关系r。局限性：①假设过强：同时提到两实体的句子不一定表达该关系，导致大量噪声标签；②知识库不完整：导致假阴例；③关系偏置：高频实体对容易被过度关联。缓解方法包括多实例学习、注意力去噪等。
💻 编程题¶
- 实现一个完整的中文关系抽取系统（使用规则或BERT）。
- 从给定的新闻语料中构建一个小型知识图谱。
- 实现TransE知识图谱嵌入模型的训练。
🔬 思考题¶
- 在大模型时代，传统的信息抽取方法还有必要吗？大模型在IE任务中的优劣势是什么？
答案：仍有必要。大模型优势：零样本/少样本能力强；Prompt灵活定义抽取schema；理解复杂语境。大模型劣势：①结构化输出不稳定，可能格式不一致或产生幻觉；②大规模文档抽取成本和延迟高；③垂直领域精度不及专用微调模型；④难以严格遵循预定义schema。最佳实践：大模型适合快速原型和少样本场景；生产系统用专用模型保证精度和效率，大模型辅助数据标注和难例处理。
✅ 自我检查清单¶
Text Only
□ 我理解信息抽取的主要任务
□ 我知道Pipeline和联合抽取方法的区别
□ 我能实现简单的关系抽取
□ 我理解事件抽取的框架
□ 我知道知识图谱的构建流程
□ 我理解远程监督的原理和问题
□ 我完成了至少3道练习题
📚 延伸阅读¶
- Distant Supervision for Relation Extraction
- TransE论文: Translating Embeddings for Modeling Multi-relational Data
- OpenIE综述
- 中文开放知识图谱OpenKG
下一篇：09-问答系统 — 从检索式到生成式问答