CLIP(Contrastive Language–Image Pre-training):把文本和图片映射到同一个向量空间里然后比较它们的语义是否接近。CLIP 特别适合做这些事情图文匹配一张图和一句话是否描述同一内容图片检索输入“穿红衣服的人”找相关图片零样本分类不用专门训练分类器也可以判断图片属于“猫 / 狗 / 汽车”多模态聚类把语义相似的文字和图片聚到一起Milvus是一个向量数据库向量相似度查询例如“找最像这个语义的内容”AlignmentPipeline它把已经预处理好的资源进一步转成“主题语义表示”再把资源组织成事件并通过即时 全局校验尽量修正错误事件绑定。局部即时校验每绑定一条马上检查全局校验所有资源绑定完成后再处理wash.json表示的是“当前这一轮数据预处理产生的完整数据”而不是永久积累所有历史数据pipeline_data_preprocessing.py:配置给你、数据库连接给你然后我把真正干活的SourceIngestionPipeline叫过来让它处理当前这一轮数据。它处理完后把结果告诉我我再把结果往上层返回。T_Source:pipline_source_ingestion.py:可以把这个文件夹理解成原始数据入库前的总负责人。它做的事情不是某一个单独动作而是把下面这些动作串起来原始数据先被外部采集程序放进resource_receive。这个文件会去扫描里面有哪些 Twitter、Wiki、Website、Facebook、News 或 PDF 数据。找到以后它不会马上处理而是先判断这一整个批次是不是已经“写完了”。为什么要等稳定因为外部程序可能还在持续往目录里写文件。如果现在就去读可能只能读到半份 BCP 数据。所以程序会连续比较目录的状态包括文件数量、总大小、最新修改时间连续若干轮都没变化才认为这一批数据已经稳定可以正式处理。默认是连续 2 轮稳定。批次稳定以后程序先把整个批次原样复制一份到resource_target。这相当于先做一个备份和存档防止以后原始数据丢失也方便后续追溯。接下来开始真正处理数据。如果是 Twitter、Wiki、Website、Facebook、News 这种信源它会找到该目录里的唯一.bcp文件然后逐行读取里面的 JSON 数据。每一条原始 JSON 数据都会交给SourceRecordMapper把原始字段转换成T_Source所需要的统一字段。映射完成以后如果当前不是dry_run模式就调用repository.upsert_source(source_data)把数据正式写进 MySQL 的T_Source。同时这些映射完成的数据也会暂时保存到wash_items里面准备后面生成wash.json。PDF 的处理稍微不一样。PDF 不需要找 BCP而是直接读取 PDF 文件一个 PDF 最终对应一条T_Source记录。如果一个文件夹里有多个 PDF就逐个处理每个 PDF 产生一条记录。等当前整个批次里的所有信源都处理完以后程序会把这些标准化后的数据交给BatchWashJsonBuilder统一生成“大 wash.json”。如果这一轮同时有多个稳定批次第一个批次会覆盖旧的wash.json后面的批次继续追加最后这一轮所有稳定批次的数据会合并到同一个大wash.json里面。最后这个批次会被标记成ingested以后再次扫描到它时就不会重复处理从而避免重复入库。实现的功能 6 个核心功能发现数据从resource_receive里找到需要处理的批次和信源目录。判断数据是否完整通过目录快照判断批次是否已经稳定避免读取还没写完的数据。备份原始数据稳定以后复制到resource_target保存原始版本。把不同来源的数据转成统一格式BCP、PDF 最终都转换成T_Source的标准字段。写数据库调用MySQLRepository.upsert_source()写入T_Source。解决了什么问题1.解决“数据还没写完就被读取”的问题外部程序还在写 BCP你的程序已经开始读读到一半数据数据库数据不完整。通过目录快照连续两轮不变才开始处理。2.解决“不同数据来源格式不统一”的问题数据来源很多每种原始结构可能都不一样需要把他们统一变成T_Source 标准字段后面的模型就不用再管了3.解决“重复处理同一个批次”的问题程序会记录folder_status.json如果某个批次已经是ingested下次再运行的时候会直接跳过。一句话总结它负责把resource_receive里的原始批次等到数据稳定后先备份再解析、映射成统一的T_Source数据写进 MySQL同时生成给后续模型使用的大wash.json。source_record_mapper.py:这个文件不负责读取 BCP也不负责真正操作数据库。它负责把“读出来的一条原始数据”加工成“一条可以直接写入 T_Source 的标准数据”。一句话总结把 Twitter、Wiki、Website、Facebook、News、PDF 等不同来源的数据统一转换成 T_Source 标准格式。词语解释tweet_topic 不同信源最终统一整理出来的“核心文本内容”。所有有意义的文本都尽量进入tweet_topic供后续主题抽取和事件匹配使用。SourceRecordMapper信源记录映射器原始数据格式转换器把不同来源的一条原始记录转换成统一格式。解决了什么问题如果没有这个文件后面的数据库可能面对Twitter一种格式、Wiki一种格式后面代码可能就得写Twitter怎么处理、Wiki怎么处理每一种格式都要分别处理非常乱。它解决的核心问题是把多源异构数据统一成一种标准格式让后面的数据库、topic 提取、事件对齐模块不用关心数据最开始是从哪里来的。我提出的问题bcp信源交给 source_record_mapper和pdf信源交给 pdf_source_processor都是为了进行字段映射吗 字段映射之后再统一字段吗准确描述BCP 数据本身已经是结构化记录可以直接交给 source_record_mapper做字段映射PDF是非结构化文件需要先交给pdf_source_processor 提取正文和图片然后再调用source_record_mapper 映射成统一的 T_source 字段。“解析”是把文件内容读取出来“清洗”是处理空值、时间、列表和异常格式“字段映射”是把不同名称、不同结构转换成统一字段映射结束后得到的已经是统一的T_Source数据。说明原始数据可能使用不同名称表示正文text、context、body、article_content经过source_record_mapper后都会整理到统一字段中例如 tweet_topic,时间、URL和id也是如此。说明pdf_source_processor不只是做字段映射它主要负责打开和解析 PDF提取正文提取 Markdown整理PDF中的图片复制或记录相关资源文件MinerU失败时进行降级解析最后调用SourceRecordMapper完成标准字段映射。source_topic_builder.py它的任务不是写数据库而是把各个信源里的标题、正文、转推内容、Wiki 内容等整理成一个干净的tweet_topic供后面的主题抽取、事件匹配、事件对齐继续使用。普通文本处理 纯图片数据处理一句话总结先把各种原始文本转成字符串 → 再把多个文本拼起来 → 再统一清洗 → 最后得到干净的tweet_topic。如果只有图片没有文本就尝试通过多模态模型生成一段文本再走同样的清洗流程。专业术语解释resource_receive:资源接受目录可理解为“系统收货区”这是外部采集程序投放原始数据的入口。Twitter、新闻、Wiki、Facebook、PDF等采集结果先放到这里。watching正在观察可以理解为货已经到了但还没卸完先不要处理。表示系统已经发现某个采集批次但还不能确认文件是否传输完成。系统会连续检查文件数量、文件总大小、最后修改时间如果这些数据还在变化就保持watching。resource_target资源目标目录可以理解为验收后的正式存档区。采集批次确认稳定后会从resource_receive复制到resource_target用于保存原始数据副本、后续处理 、出错后重新运行。iterator_status迭代器状态文件可以理解为批次处理进度表或书签。迭代器状态文件通常记录当前处理到第几条总共有多少条当前资源的source_id批量wash.json的文件指纹当前状态上一条成功完成的资源。item_in_progress表示当前有一条资源正在处理中可以理解为这条任务已经发出但还没收到“完成”确认。此时游标不会前进。如果程序中断下次仍会重新输出这条资源。mark_done表示明确告诉迭代器“当前这条处理成功了。可以理解为“消息队列中的“确认收货”。它会 记录当前资源已经完成游标加一清除当前wash.json准备处理下一条。batch_done表示当前批量wash.json中允许处理的资源已经全部完成。MinerUBCP原始数据文件就是装原始信源数据的文件容器后续程序从这里把数据读出来再做统一处理一行 一条json记录。source_id:一条资源在整个系统中的唯一编号tweet_topic:资源的主要文本内容T_Source:原始资源表,可以理解为标准化后的原材料仓库。T_Derived:派生信息表wash_itemswash.json:wash 在这里表示“清洗并标准化后的数据”不是一种特殊文件格式本质上还是JSON。项目中有两种wash.json.大 wash.json 一批任务 当前 wash.json 现在正在处理的一条任务event_id:一个事件的唯一编号core_id:当前事件的代表核心资源IDcore_ids:当前事件包含的所有核心资源IDassociation_ids:与事件相关、但不作为事件事实核心的资源IDcore_ids 用来证明或定义这件事的材料 association_ids 围绕这件事产生的相关讨论材料source_id:一条资源source_ids:多条资源ID组成的列表T_Entity_Bind实体绑定表通常一个QID对应一行继承父推文绑定推文之间可能存在回复、转发、引用其中当前推文是“子推文”被回复、转发或引用的推文是“父推文”。如果父推文已经绑定到某个事件子推文可以直接继承父推文的事件绑定。口语化理解父推文已经确定在讨论某个事件那么它的回复、转发或引用大概率也在讨论这个事件所以先把子推文跟过去再由语义校验进行复核。MinerU一个文档解析工具主要用于处理PDF。项目中主要使用MinerU提取PDF文本和图片。如果MinerU不能运行还会尝试普通PDF文本提取作为降级方案。Milvus:一个向量数据库Milvus擅长查询哪段文字或哪张图片在语义上最相似Markdown一种轻量文本格式PDF或网页解析结果中可能包含Markdown格式。进入模型前Align1会根据配置删除部分Markdown符号减少格式噪声。HTML网页的结构标记语言网页采集内容可能混有HTML标签。预处理或文本融合时通常会去除标签只保留可读文字。URL:网页或网络资源地址,URL可以用于保存原始资源位置但如果直接混入关键词和主题文本可能产生无意义的词元所以Align1通常会在融合文本中移除URL。学习顺序pipline_data_preprocessing.py数据预处理总开T_Source先判断外部有没有传配置没有就自动创建config_manager。从config.yaml中读取resource_receive在哪里resource_target在哪里批次稳定需要检查几轮wash.json写到哪里MySQL连接参数等。再判断外部有没有传数据库操作对象如果没有就根据配置创建一个用于写入T_Source。1.扫描接受目录将这些信源按照resource_receive下的一级目录归并成批次。整个batch_001被当成一个批次。resource_receive/└── batch_001/├── twitter-xxx/├── news-xxx/└── documents/2.判断批次是否传输完成系统不会发现问题立马处理因为采集程序可能还在向目录写数据。它会记录目标快照文件数量、文件总大小、最后修改时间和上一轮比较。如果快照发生变化说明文件还在传输状态保持watching暂时不处理如果连续多轮快照没有变化说明文件基本传输完毕并且批次状态变为stable开始处理。3备份原始批次批次稳定后将整个目录复制到resource_target目的是为了保留一份原始数据后续人工检查出错后重新处理防止接受目录中的文件丢失。4.解析不同类型的数据程序根据资源类型不同选择不同的解析的方式。BCP信源里的Twitter、Wiki通常读取对应目录中的.bcp文件项目里的BCP按照JSONL处理一行 一条JSON资源。然后逐条交给source_record_mapper进行字段映射。PDF信源交给pdf_source_processor.对于pdf先尝试MinerU解析提取正文和图片MinerU失败了再使用普通pdf文本提取将pdf整理成标准资源。5.统一字段不同信源的字段不同因此要转换成统一的T_Source结构。6.写入T_source,每条资源映射完成后会执行repository.upsert_source(source_data)upsert表示数据不存在就新增、数据已经存在就更新、重跑时不会简单的重复插入。如果使用了dry_run会跳过数据库写入但仍会完成扫描和字段转换。7.生成批量wash.jsonwash_items表示多条已经整理好资源wash.json是把wash_items保存到磁盘形成的json文件所有资源处理完成后把映射结果汇总成批量wash.json大致结构如下。同一轮如果有多个稳定批次第一个稳定批次 → 覆盖原来的大 wash.json后面的稳定批次 → 继续追加到这份 wash.json。所以一轮结束后当前稳定批次会被合并成一份大wash.json。生成新文件后还会重置washjson_iterator的游标状态让后续从新批次的第一条开始处理。[{source_id: source_001,source_type: news,tweet_topic: ...},{source_id: source_002,source_type: twitter,tweet_topic: ...}]8.保存批次状态并返回结果最后更新folder_status.json,记录哪些批次还在观察、哪些已经稳定、哪些已经复制、哪些已经入库、哪些处理失败、每个批次生成了多少条资源。为什么预处理只运行一轮预处理只运行一轮它把当前所有稳定批次处理完之后立即返回让外出总流水线继续执行派生处理、对齐等阶段任务。否则预处理如果一直监听就永远不会把程序控制权交给后面的模块。我们设置了“最多扫描三轮、每轮等待31秒”在pipline_source_preprocessing.py中有写。这个文件不负责什么它不负责生成单条当前now_washjson/wash.json遍历大wash.json文本、图片、视频派生处理实体消歧Align1主题建模Align2事件对齐资源绑定。原始采集批次 → 判断稳定 → 备份 → 解析和统一字段 → 写入T_Source → 生成批量wash.json总结它先看收货区有没有新货再看看货是不是已经送完。货没送完就继续等货送完了就复制一份留档然后让不同工作人员拆解BCP和PDF把各种格式统一贴上系统标签放进T_Source仓库。最后再列一张大清单wash.json交给后面的流水线逐条处理。它做完这一轮就下班不会一直霸占程序。写入当前单条wash.json”的意思是从包含很多条数据的批量wash.json中取出当前要处理的一条数据单独保存到另一个wash.json文件中供后续模块读取。Align2事件指纹时间、地点、实体预处理之后的完整流程数据预处理流程Align1流程Align2流程事件对齐语义校验资源绑定