OpenDCAI/DataFlow

# DataFlow ## 📰 0. 最新动态 * **[2026-07-13] 🚀 DataFlow-WebUI 迎来重大升级！** [DataFlow-WebUI](https://github.com/OpenDCAI/DataFlow-webui) 支持 AI Agent 辅助的 pipeline 构建、管理和执行。 * **[2026-05-22] DataFlow-Skills 现已上线！** 一系列关于使用 DataFlow 的技巧与教程。👉 [DataFlow-Skills](https://github.com/OpenDCAI/DataFlow-Skills) * **[2026-02-02] 🖥️ DataFlow WebUI 现已上线！** 仅需一条命令即可启动可视化 pipeline 构建器：`dataflow webui`。通过直观的 Web 界面构建并运行 DataFlow pipeline。👉 [WebUI 文档](#dfwebui) * **[2026-01-20] 🌟 基于 DataFlow 的优秀作品集正式上线！** 这是一个展示基于 DataFlow 构建的开源项目与研究的全新板块。欢迎踊跃贡献！👉 [优秀作品集](#awesome-dataflow) * **[2025-12-19] 🎉 我们的 DataFlow 技术报告现已发布！** 在 arXiv 上阅读并引用我们的工作：[https://arxiv.org/abs/2512.16676](https://arxiv.org/abs/2512.16676) * **[2025-11-20] 🤖 推出 DataFlow 全新 Data Agents！** 快来体验，并在 Bilibili 上观看教程：[https://space.bilibili.com/3546929239689711/lists/6761342?type=season](https://space.bilibili.com/3546929239689711/lists/6761342?type=season) * **[2025-06-28] 🎉 DataFlow 正式发布！** 我们的以数据为中心的 AI 系统现已公开。敬请关注未来的更新。 ## 🔍 1. 什么是 DataFlow？ DataFlow 是一个数据准备与训练系统，旨在从嘈杂的来源（PDF、纯文本、低质量 QA）中**生成、精炼、评估和过滤**高质量的 AI 数据，从而通过针对性的训练（Pre-training、Supervised Fine-tuning、RL training）或 RAG 系统，提升大型语言模型（LLM）在医疗保健、金融、法律和学术研究等特定领域的性能。 通过基于 `operator`（算子）的设计，DataFlow 将整个数据清洗工作流转变为可重现、可复用且可共享的 `pipeline`，为 Data-Centric AI 社区提供核心基础设施。此外，我们还开发了智能 `DataFlow-agent`，能够根据需求重新组合现有的 `operator` 或创建新的 `operator`，动态组装出新的 `pipeline`。  ## 🔍 2. 核心特性 ### ✅2.1 开箱即用的数据合成与清洗 Pipeline - 高质量训练数据生成 - 文本、数学和代码数据生成（结果详见 DataFlow-Instruct-10K） - 通过 AgenticRAG 和 Text2SQL 等工具进行数据生成 - 结构化数据提取 - 大规模 PDF 转 QA - 大规模书籍 PDF 转 Visual-QA - 科学数据工作流管理 - Text2SQL 工作流管理（被 ICDE 2026 接收） - 数学数据工作流（被 KDD 2026 接收） ### ⚙️2.2 灵活的自定义 Pipeline 编排 - 10 多个核心 operator 定义了交互模式与设计原则 - 100 多个特定 pipeline 的 operator 可供复用或参考 - 全面支持创建自定义 operator —— 即插即用，可通过 GitHub 或 PyPI 轻松打包和分发 ### 🧠2.3 可重现、可复用且可共享的 Data-Centric AI 系统 - 数据治理算法被封装为 operator pipeline，实现了不同数据治理策略的可重现性与公平比较（❤️对研究友好） - 轻松复用并替换底层大模型，以快速分析模型性能与数据质量之间的关系 - 基于 Python 和 Git 生态系统构建，便于分发、管理和追溯高质量、**用户自定义**的数据治理 operator 与 pipeline（❤️对企业友好） ## 🛠️ 3. DataFlow 套件 DataFlow 套件提供了必要的基础设施，以配合 DataFlow 主仓库实现 LLM 数据准备的自动化与规模化。它包含四个紧密集成的层级： - [DataFlow-Skills](https://github.com/OpenDCAI/DataFlow-Skills) – 一系列关于使用 DataFlow 的技巧和教程，涵盖 operator 开发、pipeline 构建以及面向 Data-Centric AI 的最佳实践。 - [DataFlow-WebUI](https://github.com/OpenDCAI/DataFlow-webui) – 一个可视化的、有 AI Agent 辅助的工作空间，用于构建、管理和运行 DataFlow pipeline。它结合了拖拽式 pipeline 编辑与自然语言交互，使用户能够更高效地构建和执行数据工作流。 - [DataFlow-Ecosystem](#awesome-dataflow) – 一个模块化的分发层，用于标准化 operator 注册。它使特定领域的模块（例如 [DataFlow-MM](https://github.com/OpenDCAI/DataFlow-MM)、DataFlow-AI4S）能够在统一抽象下贡献可扩展的库。 - [RayOrch](https://github.com/OpenDCAI/RayOrch) – 基于 Ray 构建的高性能编排层，为海量数据任务提供分布式计算调度和资源管理。 这些组件共同构成了一个统一、可扩展的环境，将原始数据转化为模型可用的智能。 ## ✅ 4. 为什么使用 DataFlow？ 数据生成和清洗对于构建高质量模型至关重要，但对于企业和个人而言，这些任务通常耗时、费力且成本高昂。**DataFlow 提供了一站式解决方案，以高效应对这些挑战。** 与 Nemo-Curator 和 Data-Juicer 等系统相比，DataFlow 提供了： - **对数据合成模块的强化支持** – 无缝集成文本、代码和数学数据生成 pipeline，以获取高质量训练数据集。 - **类 PyTorch 的编程管理** – 清晰的 **Pipeline → Operator → Prompt** 层次结构，便于工作流控制。 - **原则性且多类别的 Operator 分类** – Operator 被系统地组织成多个功能类别，如**生成、评估、过滤和精炼**，形成了一个科学严谨的多维分类体系，反映了数据准备的不同阶段，并实现了精准的 operator 选择与组合。 - **对调试和新手友好的设计** – 简化的工作流模式，降低了学习门槛并加速了实验进程。 ## 🔧 5. Operator 是如何工作的？ DataFlow operator 的设计宗旨是**简单清晰**。 operator 接收结构化输入（JSON、JSONL、CSV），并在经过智能处理后产生高质量输出。 每个 operator 都封装了特定的数据处理任务，提供简洁一致的 API，易于理解和集成。类 PyTorch 的设计使其直观且开箱即用，让您能够快速构建、组合和自定义 pipeline，而无需处理复杂的样板代码。 有关更多详细信息，请参阅 [Operator 文档](https://opendcai.github.io/DataFlow-Doc/zh/api/home/)。以下是一个演示如何调用 `PromptedGenerator` operator 的最简示例：  示例输入数据（json/jsonl 风格）： ``` // input.json [ {"problem": "What is 17 + 25?"}, {"problem": "If x = 3, compute 2x^2 + 1."} ] ``` Operator 调用代码： ``` from dataflow.operators.core_text import PromptedGenerator from dataflow.utils.storage import FileStorage from dataflow.serving import APILLMServing_request # 将 input file 设置为 global storage class storage = FileStorage(first_entry_file_name="./input.json",) # 配置 LLM serving（例如：OpenAI API） # 需要通过 `export DF_API_KEY=sk-xxx` 设置 api key llm_serving = APILLMServing_request( api_url="https://api.openai.com/v1/chat/completions", ) prompted_generator = PromptedGenerator( llm_serving=llm_serving, # pre-configured LLM backend system_prompt="Please solve this math problem." ) prompted_generator.run( storage=self.storage.step(), # data management (details omitted) input_key="problem", # read from this column output_key="solution" # write to this column ) ``` 运行后，operator 会将生成的结果追加到 output_key 中。例如，输出数据（json/jsonl 风格）将变为： ``` // dataflow_step1.json [ {"problem":"What is 17 + 25?","solution":"42"}, {"problem":"If x = 3, compute 2x^2 + 1.","solution":"19"} ] ```

🛠️ 6. Pipelines（点击展开）

### 🔧 6.1 开箱即用的 Pipeline Dataflow 中目前的 Pipeline 如下： - [📝 ** Pipeline**](https://opendcai.github.io/DataFlow-Doc/en/guide/textpipeline)：从大规模纯文本数据（多从互联网抓取）中挖掘问答对，用于 SFT 和 RL training。 -  - [[HuggingFace🤗 **文本 Pipeline** 的演示输入 & 输出]](https://huggingface.co/datasets/Open-Dataflow/dataflow-demo-Text) - [🧠 **推理 Pipeline**](https://opendcai.github.io/DataFlow-Doc/en/guide/reasoningpipeline/#_2-question-handling)：通过 (1) 扩展的思维链、(2) 类别分类和 (3) 难度评估来增强现有的问答对。 -  - [[HuggingFace🤗 **推理 Pipeline** 的演示输入 & 输出]](https://huggingface.co/datasets/Open-Dataflow/dataflow-demo-Reasonning) - [🗃️ **Text2SQL Pipeline**](https://opendcai.github.io/DataFlow-Doc/en/guide/text2sqlpipeline/)：将自然语言问题转换为 SQL 查询，并辅以解释、思维链推理和上下文 schema 信息。 -  - [[HuggingFace🤗 **Text2SQL Pipeline** 的演示输入 & 输出]](https://huggingface.co/datasets/Open-Dataflow/dataflow-demo-Text2SQL) - [📚 **知识库清洗 Pipeline**](https://opendcai.github.io/DataFlow-Doc/en/guide/r51ooua8/)：从表格、PDF 和 Word 文档等无序来源中提取并结构化知识，形成可用于下游 RAG 或生成 QA 对的数据条目。 -  - [🤖 **Agentic RAG Pipeline**](https://opendcai.github.io/DataFlow-Doc/en/guide/agenticrag_pipeline/)：从现有的 QA 数据集或知识库中识别并提取需要外部知识才能回答的 QA 对，用于下游 Agentic RAG 任务的训练。 -  ### ⚙️ 6.2 灵活的 Operator Pipeline 在该框架中，operator 被分为基础 Operator（Fundamental Operators）、通用 Operator（Generic Operators）、特定领域 Operator（Domain-Specific Operators）和评估 Operator（Evaluation Operators）等，支持数据处理与评估功能。详情请参阅[文档](https://OpenDCAI.github.io/DataFlow-Doc/)。 ### 🤖 6.3 Agent 引导的 Pipeline - **DataFlow Agent**：一个智能助手，能够根据特定任务目标进行数据分析，编写自定义 `operator`，并自动将它们编排到 `pipeline` 中。 -  - [[HuggingFace🤗 **DataFlow Agent** 的演示输入 & 输出]](https://huggingface.co/datasets/Open-Dataflow/dataflow-demo-Agent)