⚡ 第8课：事件抽取

from dataclasses import dataclass, field from typing import List, Dict, Optional import re @dataclass class EventSchema: """事件类型定义""" event_type: str trigger_words: List[str] # 触发词列表 required_args: List[str] # 必需论元 optional_args: List[str] # 可选论元 description: str = "" @dataclass class ExtractedEvent: """抽取的事件实例""" event_type: str trigger: str # 实际出现的触发词 arguments: Dict[str, str] # {论元角色: 值} text: str = "" # 原文 confidence: float = 1.0 # 置信度 def to_triple(self): ">>>转为三元组列表（事件中心化表示）""" triples = [] event_id = f"event_{id(self)}" triples.append((event_id, "类型", self.event_type)) triples.append((event_id, "触发词", self.trigger)) for role, value in self.arguments.items(): triples.append((event_id, role, value)) return triples class EventExtractor: """基于规则的事件抽取器""" def __init__(self): self.schemas: Dict[str, EventSchema] = {} self.arg_patterns: Dict[str, List] = {} # {event_type: [(role, pattern)]} def register_event_type(self, schema): self.schemas[schema.event_type] = schema def add_arg_pattern(self, event_type, role, pattern_str): """添加论元抽取模式""" self.arg_patterns.setdefault(event_type, []).append((role, re.compile(pattern_str))) def extract(self, text): """从文本中抽取事件""" events = [] # 1. 触发词检测 for event_type, schema in self.schemas.items(): for trigger in schema.trigger_words: if trigger in text: # 2. 论元抽取 arguments = {} for role, pattern in self.arg_patterns.get(event_type, []): match = pattern.search(text) if match: arguments[role] = match.group(1) if match.groups() else match.group() # 3. 检查必需论元 missing = [a for a in schema.required_args if a not in arguments] confidence = 1.0 - 0.2 * len(missing) event = ExtractedEvent( event_type=event_type, trigger=trigger, arguments=arguments, text=text, confidence=max(confidence, 0.1) ) events.append(event) return events # ========== 构建事件抽取系统 ========== ee = EventExtractor() # 人事变动事件 ee.register_event_type(EventSchema( event_type="人事变动", trigger_words=["辞职", "离职", "就任", "任命", "升任"], required_args=["人物"], optional_args=["职位", "机构", "时间"], description="人事变动事件" )) ee.add_arg_pattern("人事变动", "人物", r'([一-鿿]{2,4})(辞职|离职|就任|任命|升任)') ee.add_arg_pattern("人事变动", "职位", r'([一-鿿]{2,6})职位?') ee.add_arg_pattern("人事变动", "机构", r'([一-鿿]{2,8})(公司|集团|大学|部门)') # 商业收购事件 ee.register_event_type(EventSchema( event_type="商业收购", trigger_words=["收购", "并购", "并购了", "收购了"], required_args=["收购方", "被收购方"], optional_args=["金额", "时间"], )) ee.add_arg_pattern("商业收购", "收购方", r'([一-鿿]{2,6})(收购|并购)') ee.add_arg_pattern("商业收购", "被收购方", r'(?:收购|并购)(?:了)?([一-鿿]{2,6})') ee.add_arg_pattern("商业收购", "金额", r'(\d+[.亿万千百]+元?)') # 测试 test_texts = [ "张三辞职了公司CEO职位", "阿里巴巴收购了优酷，金额为45亿美元", "李四被任命为技术部门经理", "腾讯并购了美团部分业务", ] print("=== 事件抽取结果 ===") for text in test_texts: events = ee.extract(text) for event in events: print(f" 事件类型: {event.event_type}") print(f" 触发词: {event.trigger}") print(f" 论元: {event.arguments}") print(f" 置信度: {event.confidence:.2f}") print(f" 三元组: {event.to_triple()}") print()

class EventGraph: """事件图谱：事件间的关系网络""" def __init__(self): self.events = {} # {event_id: ExtractedEvent} self.relations = [] # [(source_id, relation, target_id)] def add_event(self, event): eid = f"e{len(self.events)}" self.events[eid] = event return eid def add_relation(self, src, rel, tgt): self.relations.append((src, rel, tgt)) def find_causal_chain(self, event_id, max_depth=5): """查找事件的因果链""" chain = [event_id] current = event_id for _ in range(max_depth): for src, rel, tgt in self.relations: if src == current and rel == "导致": chain.append(tgt) current = tgt break else: break return chain # 构建事件图谱 eg = EventGraph() e1 = eg.add_event(ExtractedEvent("经济危机", "爆发", {"时间": "2008"})) e2 = eg.add_event(ExtractedEvent("失业潮", "出现", {"规模": "大规模"})) e3 = eg.add_event(ExtractedEvent("政策调整", "实施", {"类型": "刺激政策"})) eg.add_relation(e1, "导致", e2) eg.add_relation(e2, "引发", e3) print("=== 事件因果链 ===") chain = eg.find_causal_chain(e1) for eid in chain: ev = eg.events[eid] print(f" → {ev.event_type}({ev.trigger})")