[Bug]: Dataflow pipeline persists component outputs (incl. embedding vectors) in pipeline_operation_log DSL — oversized INSERT marks documen

问题场景

用户将数据集绑定到自定义 ingestion pipeline（dataflow），典型链路为 Parser → Title Chunker → Indexer（检索方式为 Embedding + Full-text）。在上传并解析一个较大的 Markdown 文件（约 1.2 MB，切成 3641 个 chunk，使用 1024 维嵌入模型）后，日志显示 Indexing done、分块已可见、chunk 计数已更新、进度置为 1.0，但随后任务与文档被标记为 FAIL。dataset 列表中该文档显示失败，而实际上所有 chunk 都已存在于 Elasticsearch 中。

报错原文

07:17:14 Indexing done (981.17s). Task done (1087.78s)
07:17:30 [ERROR][Exception]: (0, '')

File ".../pymysql/connections.py", line 813, in _write_bytes
    self._sock.sendall(data)
BrokenPipeError: [Errno 32] Broken pipe
...
peewee.OperationalError: (2006, "MySQL server has gone away (BrokenPipeError(32, 'Broken pipe'))")

File ".../peewee.py", line 3259, in connect
    raise OperationalError('Connection already opened.')
...
File ".../pymysql/connections.py", line 844, in _execute_command
    raise err.InterfaceError(0, "")
pymysql.err.InterfaceError: (0, '')

原因分析

最可能的原因是 pipeline 操作日志的序列化负载过大，超出 MySQL 单次写入上限，具体链路如下（基于 Issue 中的分析）：

1. run_dataflow 在最后一步把 pipeline 以 PipelineOperationLogService.create(..., dsl=str(pipeline)) 持久化，而 Graph.__str__ 会序列化每个组件的 _param.outputs。
2. 若 pipeline 末尾是执行嵌入的 Tokenizer / Indexer 节点，outputs 中会包含所有 chunk 及其嵌入向量（q_1024_vec，每个 1024 个 float）。该测试文档序列化后的 DSL 约 70+ MB。
3. 这条 INSERT 超过 MySQL 默认 max_allowed_packet（64 MB），客户端仍在发送数据时服务端已关闭连接，触发 BrokenPipeError 和 MySQL server has gone away。
4. 重试机制未能恢复（对应 #15198）：RetryingPooledMySQLDatabase._handle_connection_loss 调用 self.close() 时在已死 socket 上执行 ROLLBACK 再次抛错，连接未被真正关闭，5 次重试全部以 (0, '') 失败。
5. 异常传播到 handle_task，进度被置为 -1，于是文档在摄取完全成功之后被标记为 FAIL。该问题可稳定复现（同一文档 2/2 次），而小文档（几百个 chunk 以内）不受影响。

此外，Issue 确认 task_executor.py 中有 4 处、重构版 task_executor_refactor/dataflow_service.py 中有 3 处调用点都使用 dsl=str(pipeline) 序列化完整 pipeline，且 dsl 列为 JSONField，没有应用层大小保护，唯一约束就是 MySQL 的 max_allowed_packet。

环境排查

• RAGFlow 版本：确认为 v0.26.4 或 main（受影响代码路径未变化）。
• MySQL：确认 max_allowed_packet 取值（默认 64 MB）。
• 数据集 pipeline 配置：是否为自定义 dataflow，链路中是否包含执行嵌入的 Indexer / Tokenizer 节点。
• 嵌入维度与文档规模：确认嵌入向量维度（如 1024 维）及分块数量，估算序列化 DSL 体积。
• 数据库客户端依赖：确认 PyMySQL / peewee 版本，便于比对报错堆栈行号。
• Issue 未提供操作系统、Python、CUDA、显卡信息，无需按这些维度排查。

解决步骤

1. 先确认现象特征：日志中 Indexing done 之后才出现 [ERROR][Exception]: (0, '')，且文档 chunk 实际已写入 doc engine。若符合，基本可定位到日志写入阶段而非摄取阶段。
2. 核对 MySQL 的 max_allowed_packet，并估算本次序列化后的 DSL 大小（分块数 × 嵌入维度 × 浮点字节数），确认是否超过该限制。
3. 可优先尝试临时方案：提高 MySQL max_allowed_packet，让当前文档能够写入完成。注意这只是缓解，文档更大时会再次触发。
4. 可优先尝试代码侧规避：在调用 PipelineOperationLogService.create 前，对 pipeline 做深拷贝并清空各组件的 _param.outputs（或至少剔除嵌入向量等大字段），再执行 str(pipeline)。Issue 分析认为这是能从根本上消除该类问题的方向。
5. 可优先尝试防御性处理：将 PipelineOperationLogService.create 调用包在 try/except 中，使日志写入失败不再把进度已是 1.0、计数已提交的任务翻转为 FAIL。
6. 若需要彻底解决，关注上游是否引入专门的序列化方法（如 Graph.to_dsl(include_outputs=False)）以及 #15198 的连接丢失重试修复；当前 Issue 中未确认有已合并的官方修复，需自行验证。

验证方法

用同一份大文档（例如约 1.2 MB、数千个 chunk、1024 维嵌入）重新跑一遍 pipeline：若日志在 Indexing done 之后不再出现 (0, '') 与 MySQL server has gone away，任务与文档状态保持成功而非 FAIL，即说明触发点已被规避。同时可检查 pipeline_operation_log 中 dsl 字段的大小是否回落到合理范围（不再包含嵌入向量），作为剥离 outputs 是否生效的直接证据。

参考来源

infiniflow/ragflow #17466

AI 工具推荐

想把多个 AI 模型放在一个入口？

GamsGo AI 集成 ChatGPT、DeepSeek、Gemini、Claude、Midjourney、Veo 等常用模型，适合写作、绘图、视频和日常 AI 工作流。

了解 GamsGo AI

推广链接：通过此链接购买，我可能获得佣金，不影响你的价格。