DataFlow：OpenDCAI开源的LLM数据中心化自动化处理框架

DataFlow是什么

DataFlow是由OpenDCAI团队开源的一款面向数据中心化AI（Data-Centric AI） 的自动化数据处理与数据集构建框架，DataFlow支持处理文本、PDF、表格、低质量问答对、非结构化文档等多种数据源，输出可直接用于大模型预训练、监督微调（SFT）、强化学习（RLHF）、RAG检索增强生成等场景的标准化数据，同时面向医疗、金融、法律、代码、Text2SQL等垂直领域提供专项优化，是一款兼顾学术研究与工业级部署的轻量化、高扩展数据处理基础设施。

在当前大模型开发流程中，模型结构优化日趋同质化，数据质量与数据处理流程逐渐成为决定模型效果的核心因素。但传统数据处理普遍存在流程碎片化、代码复用性差、质量不可控、垂直领域适配不足、依赖大量手工脚本等问题。DataFlow正是为解决这些痛点而生，它将数据清洗、转换、增强、生成、评估等能力封装为标准化算子，通过预置流水线与智能Agent编排，实现从原始噪声数据到高质量训练数据的一站式自动化生产。

与普通数据处理库不同，DataFlow并非简单提供数据加载与格式转换能力，而是构建了一套可组合、可调试、可复现、可自动编排的数据工程体系，让开发者无需重复编写复杂脚本，即可快速搭建稳定、高效的数据生产链路。

功能特色

模块化算子设计，高度可复用

DataFlow将所有数据处理逻辑抽象为独立、标准化的算子（Operator），每个算子只负责单一原子任务，例如去重、过滤、文本清洗、格式转换、语义增强、答案校验等。

• 所有算子遵循统一接口规范，可自由组合、插拔、替换与串联

• 支持自定义算子扩展，方便接入业务专属处理逻辑

• 支持流水线保存与复用，一次构建多次使用，大幅降低重复开发成本

五大预置垂直领域流水线

框架内置五大开箱即用的专业流水线，覆盖当前主流大模型数据需求：

1. 通用文本流水线：从网页、书籍、语料中自动挖掘高质量问答对，用于SFT与对话模型训练

2. 推理增强流水线：为问答数据生成思维链、难度分级、类型标注，提升模型推理能力

3. Text2SQL专用流水线：自动生成自然语言转SQL样本，附带语法校验、执行验证、表结构适配

4. 知识库清洗流水线：从PDF、DOCX、表格中提取结构化知识，用于企业知识库与RAG系统

5. Agentic RAG流水线：生成依赖外部知识的复杂问答样本，专门用于智能体与检索增强训练

LLM智能代理自动编排（DataFlow-Agent）

DataFlow-Agent是框架的核心亮点，它支持使用自然语言描述需求，自动完成数据流程规划、算子匹配、参数配置与执行调度：

• 无需编写代码，通过文本指令即可生成完整数据处理流水线

• 自动识别数据类型、任务目标与质量要求，智能优化执行路径

• 支持迭代校验与自动修复，对不合格样本进行二次处理

• 极大降低使用门槛，无编程基础也可完成专业级数据处理

全链路数据质量评估与控制

DataFlow内置多维度质量评估机制，从源头保证训练数据可靠性：

• 重复数据检测、噪声过滤、低质量样本剔除

• 事实一致性校验、逻辑正确性检查、答案合理性评估

• 难度均衡分布、长度规整、格式标准化

• 支持批量质量打分，自动筛选高分数据用于训练

轻量易用，工程化友好

• 基于Python开发，API风格简洁，接近PyTorch使用习惯

• 支持CPU/GPU运行，可搭配vllm实现加速处理

• 支持批量处理、增量处理、断点续跑、异常重试

• 兼容Hugging Face生态，可直接对接Transformers、Datasets等库

• 支持亿级数据规模处理，可扩展至分布式环境

多领域适配能力

除通用场景外，DataFlow针对专业领域提供深度优化：

• 医疗：病历结构化、医学术语标准化、医学问答生成

• 金融：研报解析、风险问答、合规文本过滤

• 法律：法条抽取、案例结构化、合同要素提取

• 代码：代码生成、语法校验、测试用例生成

• 多语言：完善支持中英文数据处理，适配国内使用场景

技术细节

整体架构设计

DataFlow采用四层解耦架构，保证灵活性与可扩展性：

1. 算子层：最基础执行单元，包含规则算子、模型算子、LLM算子三类

2. 流水线层：由多个算子按业务逻辑串联，形成端到端数据处理链路

3. 代理层（DataFlow-Agent）：基于LLM实现意图理解、流程规划、算子调度

4. 应用层：提供Python API、CLI命令行、配置文件调用等多种使用方式

核心技术原理

标准化算子抽象

所有算子继承统一基类，实现process()、batch_process()、validate()等统一接口，支持串行、并行、条件分支、循环等复杂执行逻辑，确保任意算子可自由组合。

LLM在环数据生成与增强

DataFlow允许在流水线中嵌入LLM接口，形成数据生成→质量评估→修正优化的闭环系统。通过自动校验机制，剔除事实错误、逻辑矛盾、格式混乱的样本，显著提升数据可用性。

DAG执行引擎

DataFlow-Agent将用户需求转化为有向无环图（DAG），自动优化执行顺序、消除冗余操作、合并重复计算，支持批量模式、流式模式、增量模式，在处理速度与内存占用之间实现高效平衡。

垂直领域规则引擎

针对医疗、金融、法律等强专业性领域，框架内置领域知识规则与约束条件，在数据生成与校验阶段强制保证专业性、合规性与准确性，避免通用LLM产生领域错误。

技术栈与环境依赖

• 开发语言：Python 3.10及以上

• 核心依赖：PyTorch、Transformers、Datasets、pandas

• 加速依赖：vllm（可选，用于本地LLM加速）

• 数据格式：支持TXT、JSON、JSONL、CSV、PDF、DOCX

• 操作系统：Linux、macOS、Windows

应用场景

大模型全阶段数据生产

• 预训练数据清洗：过滤低质网页、冗余文本、噪声内容，提升基座模型知识密度

• 监督微调（SFT）数据构建：生成指令、对话、推理、代码、SQL等高质量样本

• 强化学习数据生产：构建偏好数据、安全对齐数据、拒绝样本，提升模型可控性

垂直领域大模型研发

适用于医疗大模型、金融大模型、法律大模型、教育大模型等领域的数据生产，解决领域数据稀缺、质量低、标准化程度差的问题，快速构建专业数据集。

RAG与企业知识库构建

从PDF、手册、公告、研报、合同等非结构化文档中自动抽取知识，生成结构化知识库与问答对，用于智能客服、内部问答系统、企业搜索平台。

智能体与工具使用数据构建

生成需要调用工具、查询数据库、检索外部知识的复杂样本，用于训练具备Agent能力的大模型，提升工具使用、多步推理、复杂任务规划效果。

学术研究与快速实验

AI研究者可快速构建对比数据集，复现论文基线，验证不同数据策略对模型效果的影响，显著降低实验成本与数据处理时间。

使用方法

环境安装

首先配置Python环境，推荐使用Conda：

conda create -n dataflow python=3.10
conda activate dataflow

安装DataFlow核心库：

pip install open-dataflow

如需GPU加速，可安装扩展版本：

pip install open-dataflow[vllm]

查看版本验证安装：

dataflow --version

基础使用流程

1. 准备原始数据，支持本地文件、Hugging Face数据集等

2. 选择预置流水线或通过DataFlow-Agent自动构建

3. 运行流水线处理数据

4. 导出JSONL/CSV等标准格式用于模型训练

代码调用示例

from dataflow import Pipeline
from dataflow.operators import *

# 构建文本清洗+去重+改写流水线
pipe = Pipeline([
  TextClean(),
  Deduplicate(),
  Paraphrase(lang="zh")
])

# 加载原始数据
data = [
  "AI正在改变各行各业。",
  "人工智能正在改变各个行业。",
  "大模型技术发展迅速，应用越来越广泛。"
]

# 执行处理
result = pipe(data)

# 输出结果
for item in result:
  print(item)

DataFlow-Agent 自然语言调用示例

from dataflow import DataFlowAgent

agent = DataFlowAgent()

prompt = """
处理一批中文金融文档：
1. 清洗文本，去除特殊符号
2. 自动去重
3. 生成投资分析类问答对
4. 过滤低质量样本
5. 保存为jsonl格式
"""

pipeline = agent.build(prompt)
output = pipeline.run("./finance_docs")

大规模数据处理

DataFlow支持多进程、批处理、断点续跑，可处理百万至亿级数据量，同时支持对接Spark、Dask等分布式框架，满足企业级生产需求。

竞品对比

常见问题解答

DataFlow必须使用GPU才能运行吗？

不是必须。CPU环境可以正常运行基础清洗、转换、去重等算子；只有在使用本地LLM进行数据生成、增强时才建议使用GPU，以提升处理速度。

DataFlow是否支持中文数据处理？

完全支持。框架内置完善的中文分词、文本清洗、去重、改写、质量评估算子，对中文语料、文档、问答数据均有专项优化，是国内开发者友好型框架。

没有Python编程基础可以使用DataFlow吗？

可以。通过DataFlow-Agent模块，只需要使用自然语言描述数据处理需求，即可自动生成并执行流水线，无需编写任何代码。

处理后的数据可以直接用于Hugging Face模型训练吗？

可以。DataFlow支持导出JSONL、CSV、Hugging Face Dataset等标准格式，可直接接入Transformers库进行SFT、预训练或推理。

如何添加自己的业务逻辑算子？

非常简单。只需要继承框架提供的基础Operator类，实现process方法，即可注册为自定义算子，与内置算子无缝组合使用。

DataFlow可以处理PDF、Word等文档吗？

可以。框架内置文档解析算子，支持从PDF、DOCX、表格中提取文本，并自动进行结构化处理，适合知识库构建场景。

DataFlow与自己写Python数据脚本有什么区别？

手写脚本耦合度高、难以复用、无法调试、没有质量保障；DataFlow模块化、可插拔、支持自动编排与质量校验，处理流程可复现、可共享、可稳定用于生产环境。

相关链接

• GitHub 项目主页：https://github.com/OpenDCAI/DataFlow

• 配套学术论文：https://arxiv.org/abs/2512.16676

• 官方文档地址：https://opendcai.github.io/DataFlow-Doc

总结

DataFlow作为OpenDCAI开源的面向数据中心化AI的自动化数据处理框架，通过模块化算子、预置垂直领域流水线、DataFlow-Agent智能编排等核心能力，高效解决了大模型开发中数据准备碎片化、复用性差、质量不可控、开发成本高的问题，支持从原始噪声数据到高质量训练数据的一站式自动化生产，广泛适用于预训练、SFT、RLHF、RAG、智能体构建等场景，并在医疗、金融、法律等垂直领域具备显著优势，其简洁API、低代码使用方式、完善的质量评估机制与强大的工程化能力，使其成为学术研究与工业级大模型数据生产的高效工具，为开发者提供了标准化、自动化、可扩展的数据处理解决方案。