简介本资源面向计算机、人工智能及相关专业学生与开发者提供一套基于触发词的燃气事件抽取完整Python实现可用于课程设计、毕业设计或NLP入门进阶。项目围绕爆炸、起火、泄漏、中毒等触发词识别燃气火灾、爆炸、泄漏、CO中毒、闪爆等事故类型并抽取发生时间、事故地点、救援组织、事故损伤与事故原因等实体信息形成结构化事件数据。压缩包共23个文件约98KB包含12个Python源码文件、2个Excel与2个CSV数据表、2个Word说明文档以及测试结果、新闻样例、JSON配置和README等覆盖数据处理、命名实体识别、事件抽取、图数据库导入等模块。已有124人学习下载。代码均经测试运行成功附文档说明适合小白学习进阶也可在现有基础上修改扩展用于毕设、课设或项目立项演示。1. 燃气事件抽取到底在抽什么从一句工单里挖出时间、地点、原因、后果和组织燃气行业的工单、抢修记录、巡检日志里藏着大量非结构化文本比如「3月12日晚8点朝阳区某小区因第三方施工挖断中压管道导致周边300户停气约4小时燃气集团抢修二队到场处置」。人一眼能看懂但要让系统自动统计「哪个区事故最多」「第三方施工占多少比例」「平均停气时长」就得先把这些信息结构化。事件抽取要做的就是围绕一个触发词这里是「挖断」把时间、地点、原因、后果、组织等实体信息一次性抽出来填进固定字段。触发词是整条流水线的锚点——没有它你不知道哪句话在描述一个事件也不知道该抽哪一类实体。这套方案适合燃气公司信息化岗、做安全生产数据分析的工程师以及想拿真实场景练手 NLP 的开发者。Python 源代码加文档说明的组合意味着你拿到的不只是思路而是能直接跑起来、改字段就能用的东西。2. 触发词驱动的抽取框架为什么先定触发词再谈模型2.1 触发词是事件抽取的「开关」不是可选项很多人一上来就想上大模型端到端抽结果发现模型把「管道」当组织、把「停气」当原因字段全乱。问题出在缺少约束。触发词的作用是给抽取划定边界只有句子里出现了「泄漏」「爆炸」「挖断」「起火」「超压」这类词才认为这是一个需要抽取的事件句触发词本身的类别还决定了后续该重点抽哪些实体。比如「泄漏」类事件原因字段大概率是「腐蚀」「老化」「接口松动」「挖断」类事件原因字段大概率是「第三方施工」「未交底」。先定触发词表等于先给数据分了类后面无论用规则还是模型准确率都会明显不一样。常见做法是维护一张触发词表按事件类型分组每组给一个类型标签。这张表不需要很大燃气场景下 50 到 80 个词就能覆盖绝大多数工单。关键是每个词都要有明确的归属不能一个词同时属于两类事件否则后续消歧会很痛苦。2.2 实体字段的定义要先于代码否则返工成本极高在写任何抽取代码之前必须把字段定义固定下来。燃气事件抽取通常包含这几类字段含义示例抽取难点时间事件发生时间3月12日晚8点相对时间需归一化地点事发位置朝阳区某小区粒度不统一原因直接诱因第三方施工挖断常跨句后果影响结果300户停气4小时数值和单位分离组织处置或责任单位燃气集团抢修二队简称与全称混用字段定义不清后面标注、训练、评估全部要推倒重来。我一般会先拿 200 条真实工单人工标一遍看看哪些字段实际能抽到、哪些字段大量缺失再决定最终字段集。缺失率超过 40% 的字段要么合并要么放弃不要硬抽。2.3 最小可运行流程从原始文本到结构化记录整个流程分四步分句、触发词匹配、实体抽取、字段组装。下面是一个可以直接跑的最小实现依赖 jieba 和 re不需要 GPU。import jieba import re # 触发词表事件类型 - 触发词列表 TRIGGER_WORDS { 泄漏: [泄漏, 漏气, 渗漏], 挖断: [挖断, 挖破, 破坏], 爆炸: [爆炸, 爆燃, 起火], 超压: [超压, 压力异常, 憋压], } # 实体正则时间、数量、组织 TIME_PATTERN re.compile(r(\d{1,2}月\d{1,2}日|\d{1,2}点|\d{1,2}时|晚|凌晨|上午|下午)) NUM_PATTERN re.compile(r(\d)\s*(户|栋|米|小时|分钟|公里)) ORG_PATTERN re.compile(r([\u4e00-\u9fa5]{2,10}(?:燃气|抢修|巡检|集团|公司|队|站))) def split_sentences(text): # 按中文标点分句保留分隔符 parts re.split(r(?[。]), text) return [p.strip() for p in parts if p.strip()] def match_trigger(sentence): # 返回命中的事件类型和触发词 for etype, words in TRIGGER_WORDS.items(): for w in words: if w in sentence: return etype, w return None, None def extract_entities(sentence): # 抽取时间、数量后果、组织 times TIME_PATTERN.findall(sentence) nums NUM_PATTERN.findall(sentence) orgs ORG_PATTERN.findall(sentence) return { time: times, consequence: [.join(n) for n in nums], org: orgs, } def extract_event(text): results [] for sent in split_sentences(text): etype, trigger match_trigger(sent) if not etype: continue ents extract_entities(sent) results.append({ event_type: etype, trigger: trigger, sentence: sent, **ents, }) return results if __name__ __main__: raw 3月12日晚8点朝阳区某小区因第三方施工挖断中压管道导致周边300户停气约4小时燃气集团抢修二队到场处置。 for ev in extract_event(raw): print(ev)逻辑说明split_sentences先按句号、感叹号、问号、分号切分保证触发词匹配在句子级别进行避免跨句误匹配。match_trigger遍历触发词表命中即返回事件类型这里用「先到先得」策略实际项目中如果一句多触发词需要改成多标签。extract_entities用正则抽时间、数量和后果、组织正则只做粗抽后面可以接模型精抽。extract_event把三者组装成结构化记录。参数说明TRIGGER_WORDS是核心配置新增事件类型只需加一组词TIME_PATTERN里的「晚」「凌晨」等词是为了兜住没有具体数字的时间表达NUM_PATTERN的单位列表要根据实际工单调整比如有些工单用「立方」而不是「户」。跑通之后你会发现纯规则能覆盖 60% 到 70% 的简单句剩下的复杂句才是模型要解决的问题。3. 把触发词匹配做稳分词、同义词和边界处理3.1 分词粒度直接决定触发词能不能命中jieba 默认分词会把「挖断」切成「挖」和「断」吗不一定取决于词典。如果触发词不在 jieba 词典里很可能被切碎导致w in sentence这种字符串匹配还能命中但如果你改成按词匹配就会漏。稳妥做法是把触发词表加载进 jieba 自定义词典保证触发词作为一个整体被切出来。import jieba def load_trigger_dict(trigger_words): # 把触发词加入 jieba 词典保证不被切碎 for words in trigger_words.values(): for w in words: jieba.add_word(w, freq1000) load_trigger_dict(TRIGGER_WORDS) seg jieba.lcut(第三方施工挖断中压管道) print(seg) # [第三方, 施工, 挖断, 中压, 管道]freq1000是经验值给高词频让 jieba 优先保留这个词。如果你的触发词里有「漏气」这种本身就可能被切开的词这一步是必须的。不做这一步后面按词匹配的规则会大量漏召回。3.2 同义词扩展别让「漏气」和「泄漏」分成两套逻辑燃气工单里同一个意思有多种写法「泄漏」「漏气」「渗漏」「跑气」。如果触发词表只写「泄漏」另外三种写法就全漏了。常见做法是维护同义词映射把同义写法归一到同一个标准触发词再统一走后续逻辑。SYNONYM_MAP { 漏气: 泄漏, 渗漏: 泄漏, 跑气: 泄漏, 挖破: 挖断, 破坏: 挖断, } def normalize_trigger(word): return SYNONYM_MAP.get(word, word)这样触发词表只需要维护标准词同义词通过映射表扩展。新增写法时只改映射表不动主逻辑。注意「破坏」这种词在别的语境下可能不是燃气事件所以映射表要配合上下文窗口使用不能全局替换。3.3 否定和假设句的过滤不是所有触发词都代表真实事件「如果发生泄漏应立即关闭阀门」这句话里有触发词「泄漏」但它不是事件是预案描述。不做过滤抽取结果里会混入大量非事件句。简单有效的做法是检查触发词前后 5 个字符内是否出现「如果」「若」「一旦」「防止」「避免」「演练」等词命中则跳过。NEGATIVE_CUES [如果, 若, 一旦, 防止, 避免, 演练, 假设] def is_real_event(sentence, trigger): idx sentence.find(trigger) window sentence[max(0, idx-5): idxlen(trigger)5] for cue in NEGATIVE_CUES: if cue in window: return False return True这个窗口大小 5 是调出来的太小会漏过滤太大会误杀。实际项目中我一般会把这个逻辑做成可配置不同数据源窗口不同。过滤之后事件句的准确率能提升 10 到 15 个百分点代价是少量真实事件被误杀需要根据业务容忍度权衡。4. 实体抽取的落地细节时间归一化、后果拆分和组织识别4.1 时间归一化把「3月12日晚8点」变成可计算的值抽取到时间字符串只是第一步要统计「哪个时段事故高发」必须把时间转成标准格式。燃气工单里的时间表达非常杂「3月12日晚8点」「12日20时」「昨晚八点左右」「凌晨3点」。常见做法是分两层先用正则抽候选时间串再用规则映射到标准时间。import re from datetime import datetime def normalize_time(text, base_dateNone): # base_date 用于补全年份默认当前年 if base_date is None: base_date datetime.now() year base_date.year # 匹配 月日 时分 m re.search(r(\d{1,2})月(\d{1,2})日.*?(\d{1,2})[点时], text) if m: month, day, hour int(m.group(1)), int(m.group(2)), int(m.group(3)) # 处理「晚8点」- 20点 if 晚 in text and hour 12: hour 12 return datetime(year, month, day, hour) # 匹配 仅时分 相对日 m re.search(r(\d{1,2})[点时], text) if m: hour int(m.group(1)) if 晚 in text and hour 12: hour 12 day_offset 0 if 昨 in text: day_offset -1 elif 前 in text: day_offset -2 base base_date.replace(hourhour, minute0, second0, microsecond0) return base.replace(daybase.day day_offset) return None逻辑说明先尝试匹配「月日时分」的完整表达命中后处理「晚」字带来的 12 小时偏移。再尝试只匹配时分结合「昨」「前」等相对词做日期偏移。base_date参数是为了让函数可测试实际运行时传工单的创建日期而不是当前日期否则历史工单的时间会全错。参数说明hour 12这个判断是为了避免「晚20点」被再加 12 变成 32 点。如果你的数据里有「晚12点」这种表达需要单独处理成 0 点。时间归一化的准确率直接决定后续时段统计的可信度建议单独写测试用例覆盖至少 20 种表达。4.2 后果字段拆分数值和单位必须分开存「300户停气4小时」抽成一个字符串后面没法算「平均停气时长」。正确做法是把数值和单位拆成两个字段或者拆成多条记录。常见做法是用正则捕获组分别拿数值和单位再按单位类型归到不同字段。CONSEQUENCE_PATTERN re.compile(r(\d)\s*(户|栋|米|小时|分钟|公里|立方)) def extract_consequence(sentence): results [] for num, unit in CONSEQUENCE_PATTERN.findall(sentence): if unit in (户, 栋): results.append({type: affected_count, value: int(num), unit: unit}) elif unit in (小时, 分钟): results.append({type: duration, value: int(num), unit: unit}) elif unit in (米, 公里): results.append({type: distance, value: int(num), unit: unit}) elif unit 立方: results.append({type: volume, value: int(num), unit: unit}) return results这样「300户」和「4小时」分别落到affected_count和duration后续做聚合统计时直接按 type 过滤即可。注意单位统一如果有的工单写「4个小时」正则要能兼容「个」字可以在单位前加(?:个)?。后果字段的完整性往往比原因字段高因为工单里影响范围是必填项原因反而经常缺失。4.3 组织识别简称、全称和角色区分组织字段的难点不在识别而在归一。「燃气集团抢修二队」「抢修二队」「燃气集团」可能指同一个或不同层级。常见做法是维护组织别名词典把识别到的组织映射到标准名同时标注角色责任方/处置方/受影响方。ORG_ALIAS { 抢修二队: 燃气集团抢修二队, 抢修一队: 燃气集团抢修一队, 燃气集团: 燃气集团, } ORG_ROLE_CUES { 责任: [因, 由于, 系], 处置: [到场, 处置, 抢修, 派], } def classify_org_role(sentence, org): idx sentence.find(org) window sentence[max(0, idx-10): idx] for role, cues in ORG_ROLE_CUES.items(): for cue in cues: if cue in window: return role return unknownORG_ALIAS把简称归一到全称classify_org_role根据组织前面的线索词判断角色。这个逻辑不完美但比不区分角色强很多。实际项目中组织字段的召回率通常低于时间和后果因为很多工单只写「已派单」不写具体队伍这种情况下宁可留空也不要瞎填。5. 避坑与排查触发词抽取最容易翻车的五个地方5.1 触发词表一改历史结果全变现象新增几个触发词后重新跑历史数据发现之前统计的事故数量突然涨了一截报表对不上。原因触发词表是全局配置改了之后所有数据重新匹配新增词命中了以前没被识别的事件句。解决触发词表加版本号每次变更记录变更内容和生效日期历史统计按版本隔离。我一般会在输出结果里带一个trigger_version字段方便回溯。5.2 正则抽时间把「3月12日」抽成「12日」现象时间字段出现大量只有日没有月的记录归一化时全部落到当前月。原因正则(\d{1,2})月(\d{1,2})日在某些文本里因为「月」字被分词或空格隔开而匹配失败回退到只匹配日。解决在归一化前先做一次文本清洗把「3 月 12 日」这种带空格的表达压成「3月12日」再跑正则。同时给只匹配到日的记录打标记不要静默补当前月。5.3 后果字段把「4小时」和「4户」混在一起现象统计平均停气时长时结果明显偏小排查发现部分「4户」被算进了时长。原因后果正则的单位列表里「户」和「小时」都在但后续聚合时没有按 type 过滤把所有数值平均了。解决后果抽取必须带 type 字段聚合时严格按 type 分组。这个坑很隐蔽因为抽取阶段看起来是对的错在消费阶段。5.4 否定句过滤把真实事件也滤掉了现象某次真实泄漏事件没有被抽出来排查发现句子里有「防止泄漏扩大」这样的表述被否定词过滤误杀。原因否定窗口设得太大或者否定词表里包含了「防止」这种在真实事件描述里也会出现的词。解决否定过滤只对触发词紧邻的前后 3 个字符生效并且把「防止泄漏扩大」这类「防止触发词扩大」的模式单独放行。否定过滤是召回和准确率的权衡没有完美解只能按业务调。5.5 组织识别把「燃气集团」和「燃气公司」当成两个组织现象统计各组织处置量时同一个单位被拆成好几条。原因没有做组织别名归一正则抽到什么就是什么。解决维护组织别名词典抽取后先归一再统计。别名词典的维护成本不低但比统计错误强。常见做法是从历史数据里把所有组织串抽出来人工过一遍建一次词典能用很久。6. 从规则到模型触发词抽取的进阶用法和验证方法规则跑通之后你会遇到天花板复杂句、跨句原因、隐式触发词。这时候可以考虑用模型做精抽但触发词匹配这一步不要丢它仍然是最高效的候选句筛选器。常见做法是规则先召回候选事件句再用序列标注模型比如 BERTCRF在候选句上抽实体这样模型只需要处理真正有事件的句子训练数据也更容易构造。验证方法上不要只看整体准确率。按事件类型分开看泄漏类事件的召回率、挖断类事件的准确率、超压类事件的字段完整率。燃气场景下漏报一个真实泄漏事件的代价远高于误报一个非事件句所以召回率优先。我一般会设一个硬指标泄漏类事件召回率不低于 90%达不到就继续加触发词和同义词而不是急着上模型。一个具体技巧是给触发词加权重。不是所有触发词都同等重要「爆炸」的优先级高于「超压」「挖断」的优先级高于「破坏」。当一句话命中多个触发词时按权重选最高的作为事件类型可以减少类型误判。权重可以按历史数据里各触发词对应事件的真实比例来定也可以人工拍拍完在验证集上看效果。TRIGGER_WEIGHT { 爆炸: 10, 爆燃: 10, 起火: 9, 泄漏: 8, 漏气: 8, 渗漏: 7, 挖断: 7, 挖破: 6, 破坏: 5, 超压: 6, 压力异常: 5, 憋压: 5, } def match_trigger_weighted(sentence): hits [] for etype, words in TRIGGER_WORDS.items(): for w in words: if w in sentence: hits.append((etype, w, TRIGGER_WEIGHT.get(w, 1))) if not hits: return None, None hits.sort(keylambda x: x[2], reverseTrue) return hits[0][0], hits[0][1]这个权重表不需要很精确它的作用是给多触发词冲突一个确定的裁决顺序。跑一段时间后把模型判错的案例拿出来看如果发现某类冲突频繁出现再调权重。我自己的习惯是每季度过一遍触发词表和权重表把新出现的写法补进去把从来不命中的词删掉。这套东西不复杂但贵在持续维护。希望帮到你。本文还有配套的精品资源点击获取
