夜雨聆风1. 基本介绍
近年来，大模型发展主要由大型科技公司主导，其领先的核心在于规模庞大且高质量的数据资源。然而，这些公司通常并不公开其原始数据及数据处理工具，使得学术界在大模型训练数据的构建与优化方面难以追赶，受制甚深。
尽管近年来开源了大量数据集，学术界在大模型数据准备方面仍面临诸多挑战。目前，大模型训练数据的清洗与构建仍主要依赖各个研究团队 “闭门造车”，缺乏系统化、高效的工具支持。现有的数据处理工具如 Hadoop 和 Spark 等，支持的操作算子大多偏向传统方法，尚未有效集成基于最新大语言模型（LLMs）的智能算子，对于构建先进大模型的训练数据支持有限。
为此，张文涛和鄂维南院士团队提出了以数据为中心的 AI 系统 DataFlow。它系统实现了 100 余个基于规则、本地大模型或大模型 API 的数据治理算子（Operators），并在此基础上构建 8 条预设数据处理流水线（Pipeline），包括：大规模嘈杂数据（如 PDF 文档、纯文本、低质量问答数据、爬虫数据等）的清洗、扩增与评估；带有思维链的强推理数据合成；RAG 数据提取与合成等等主流数据治理需求。该系统可供用户灵活组织现有算子，开发新算子，修改现有流水线，组装定制化流水线，以满足多样的数据治理任务。
目前 DataFlow 全面支持文本模态的数据治理，可供大语言模型（LLM）的预训练（Pre-training）、有监督微调（Supervised Finetuning）、强化学习微调（Reinforcement Finetuning ）使用。经过治理的数据可以有效提升大语言模型在通用领域的推理能力和检索能力，与医疗、金融、法律等特定领域的性能。此外，多模态版本的 DataFlow 正在如火如荼开发中，会在不久的将来与大家见面。
官方文档：https://opendcai.github.io/DataFlow-Doc
Github 官方仓库：https://github.com/OpenDCAI/DataFlow
2. 设计理念
DataFlow 框架的数据处理逻辑主要分为算子层 (operator) 和流水线层 (pipeline)。其次，DataFlow 通过数据管理类（storage）实现读写管理，大模型后端类（LLMServing）支持算子调用大模型，进而实现对数据的复杂过滤、扩增和评分。此外，我们还设计了 Agent for DataFlow 模块。使用大模型 Agent 实现多种智能数据治理功能：(1) 根据用户描述自动编排算子构成新 Pipeline (2) 根据用户描述自动编写新算子 (3) Agent 自动解决数据分析任务。
2.1 框架设计
数据管理：DataFlow 现阶段主要支持大模型文本数据处理，为提高易用性，DataFlow 内核使用 pandas 的 DataFrame 数据表作为载体实现读写数据。
大模型后端：目前 DataFlow 支持两种后端，包括：
（1）使用 vLLM 或 SGLang 作为本地推理后端，在本地 GPU 部署大模型作为推理服务。支持算子决策。
（2）使用 request 方式向大模型服务商的 API（如 ChatGPT，Deepseek）发起请求，并支持多进程并发请求。
2.2 功能设计
DataFlow 算子是基本数据处理单元，通常基于规则、深度学习模型或大语言模型（LLM）实现处理逻辑。
DataFlow Pipeline 是对多个 DataFlow 算子的有序编排，旨在完成一个完整的数据处理任务。DataFlow 目前提供了 8 条完整流水线以供参考，用户可以自定义修改。
DataFlow Agent 是一个基于多智能体协同的自动化任务处理系统，覆盖 “任务拆解 → 工具注册 → 调度执行 → 结果验证 → 报告生成” 完整流程，致力于复杂任务的智能化管理与执行。
2.3 DataFlow 具体示例：
评估算子示例：我们主要通过 “通用评估算子” 和 “专有评估算子” 进行评估。以通用算子为例，主要包含文本结构、多样性和复杂性、安全性、流畅性和可理解性、教育价值、内容准确性和有效性等维度。
详见官方文档
https://opendcai.github.io/DataFlow-Doc/zh/guide/f50mqhmb/
数据处理流水线示例：我们提供多条推荐使用的数据处理流水线，数据处理及评估结果如下。
（1）通用文本数据处理流水线
主要由去重、改写和过滤算子等组建，能够实现删除冗余字符、提取有效信息的处理效果。
详见官方文档
https://opendcai.github.io/DataFlow-Doc/zh/guide/textpipeline/
如下图实验结果显示，清洗后的数据在多个评估维度都有提升。
（2）强推理数据合成流水线
主要包括以下三方面算子，能够支持数据合成及多维度数据评估。
1. 问题处理：过滤非数学问题、合成新问题、验证问题正确性、进行难度评分和类别分类。
2. 答案生成与处理：根据问题的标准答案或模型生成的答案进行处理，包括格式过滤、长度过滤和正确性验证等。
3. 数据去重：对生成的问答数据进行去重，确保数据集的质量。
详见官方文档
https://opendcai.github.io/DataFlow-Doc/zh/guide/agenticrag_pipeline/
如下图实验结果显示，合成数据在多个评估维度都有提升。
（3）Text2SQL 数据合成流水线
主要由以下几个步骤组成：
1. 数据过滤：筛选无效 SQL 和问题描述不一致的数据；
2. 难度分类：基于 SQL 语法复杂度和执行通过率划分难度等级；
3. Schema 处理：提取和格式化数据库 Schema 信息；
4. 知识生成：构建自然语言问题到 SQL 查询所需的额外知识；
5. 问题优化：标准化问题表述；
6. 提示词生成：构建用于训练的高质量问答数据。
详见官方文档
https://opendcai.github.io/DataFlow-Doc/zh/guide/text2sqlpipeline/
如下图实验结果显示，处理后数据在多个评估维度上都有提升。
（4）Agentic RAG 数据处理流程
主要由以下算子构成：
1. 片段遴选：选取过滤有意义文本；
2. 构造问答对：基于现有片段中的信息，构建问答对。该问答对需要原文本支持；
3. 问题质量评估：对问答对和原文本的质量进行评估与分类，便于构建 RAG 知识库；
4. 问题横向合并：可以合并多个问答对的内容，提高问题复杂性与跨度；
5. 问题纵向合并：将多个问题合并为需要多次检索的问答对。
详见官方文档
https://opendcai.github.io/DataFlow-Doc/zh/guide/agenticrag_pipeline/
如下图实验结果显示，合成数据训练模型在多个评估维度都有提升。
（5）Agent 自动数据处理流程
Agent 主要由以下六个模块组成，可以添加算子并且编排已有算子，以实现自动的数据处理流水线。
1. Planning Agent：任务分解，定义任务链；
2. Task Dispatcher：分配任务；
3. Execution Agent：生成工具，处理任务；
4. Tool Register：注册工具；
5. Evaluation Agent：调试与验证；
6. Analysis Agent：总结并生成报告。
详见官方文档
https://opendcai.github.io/DataFlow-Doc/zh/guide/agenticrag_pipeline/
效果如下：
3. 使用方式
3.1 基于代码使用
代码仓库
目前 DataFlow 已经部署在 PyPi，可以通过 pip install open-DataFlow 轻松一键安装。
DataFlow 借鉴了 PyTorch 的风格的算子组织与调用方式，算子声明通过__init__函数实现，算子运行通过 run 函数实现。接口简明清晰，易于上手。
Github 官方仓库：https://github.com/OpenDCAI/DataFlow
使用文档
我们还提供了详细的教程文档，也欢迎开源社区提出有趣的见解，一起丰富完善文档内容，让 DataFlow 更加新手友好，利于上手。
官方文档：https://opendcai.github.io/DataFlow-Doc
3.2 基于前端使用
无代码拖拽式 Pipeline 搭建：满足需要针对业务场景个性化定制 / 微调 pipeline 的需求。
一键调用数据准备 Pipeline：提供已跑通的行业 / 特定场景的最佳实践模版。
Agent 自动化 Pipeline 设计：通过多轮对话理解并分析客户需求，给出合理的数据过滤、数据合成、pipeline 调整等建议，确认后可一键配置。
项目开源
官方文档：https://opendcai.github.io/DataFlow-Doc
Github 官方仓库：https://github.com/OpenDCAI/DataFlow
© THE END
转载请联系本公众号获得授权
投稿或寻求报道：liyazhou@jiqizhixin.com
