DataFlow 开发助手 (dataflow-dev)

April 3, 2026 · View on GitHub

激活时必须执行的步骤

加载知识库：读取 ${SKILL_DIR}/context/knowledge_base.md（架构 + API 参考）
加载开发规范：读取 ${SKILL_DIR}/context/dev_notes.md（规范 + 最佳实践）
加载已知问题：读取 ${SKILL_DIR}/diagnostics/known_issues.md（诊断快速匹配表）

探测仓库状态（在 DataFlow 仓库根目录下执行）：

git branch --show-current          # 当前分支
git log --oneline -3               # 最近提交
git diff --name-only HEAD~1 HEAD   # 最近一次变更文件列表

向用户报告当前上下文摘要（1-3行，不要冗长）

子命令路由

根据用户意图，路由到对应工作流：

用户意图关键词	执行流程
新建算子 / new operator / create operator	→ 算子创建流程
新建 Pipeline / new pipeline	→ Pipeline 创建流程
新建 Prompt / new prompt	→ Prompt 创建流程
报错 / error / KeyError / AttributeError / Warning	→ 诊断流程
审查代码 / check / review / 规范检查	→ 规范审查流程
更新知识库 / sync / check updates / 仓库有新算子	→ 知识库更新感知流程

算子创建流程

Step 1: 防重复检查（必须）

在生成代码前，先检查是否已有功能相近算子：

# 查看各模块已注册算子
grep -r "^from \." dataflow/operators/general_text/__init__.py | grep TYPE_CHECKING -A 200 | grep "^    from"
grep -r "^    from" dataflow/operators/text_sft/__init__.py
grep -r "^    from" dataflow/operators/reasoning/__init__.py

对照 context/knowledge_base.md §八的算子列表，确认无重复后再继续。

Step 2: 向用户确认规格

使用 AskUserQuestion 一次性询问以下信息（合并为一轮）：

算子类型（filter / generate / refine / eval）
所属模块（general_text / text_sft / reasoning / code / 其他）
算子功能描述（一句话）
是否依赖 LLM（是/否）
输入列名（input_key）、输出列名（output_key）

Step 3: 生成代码

使用 templates/operator_template.py 作为骨架，填入用户规格。

硬性规范 checklist（每次生成必须逐项检查）：

Step 4: 提示注册

提醒用户在对应模块的 __init__.py 的 TYPE_CHECKING 块中添加：

if TYPE_CHECKING:
    from .filter.my_new_filter import MyNewFilter  # 按实际路径

Pipeline 创建流程

Step 1: 向用户确认规格

数据来源（本地 jsonl / HuggingFace / ModelScope）
需要哪些算子（按功能描述，skill 来决定使用哪些算子）
是否需要 LLM Serving，并发量要求
是否需要断点续传（BatchedPipelineABC）

Step 2: 算子选择策略

优先使用已有算子，参考 context/knowledge_base.md §八。若为 core_text 算子，参考 generating-dataflow-pipeline skill 的算子选择规则。

Step 3: 生成代码

使用 templates/pipeline_template.py 作为骨架。

硬性规范 checklist：

storage 在 __init__ 中声明，而非在 forward() 里临时创建
每个算子调用传 storage=self.storage.step()（每次调用自动递增）
算子成员变量命名带 _stepN 后缀（N 为执行顺序）
LLM Serving 在 __init__ 中统一声明
max_workers 根据 API 能力设置（自建/代理 API 推荐 200+）
包含 if __name__ == "__main__": 入口
API key 通过环境变量注入，不硬编码

Prompt 创建流程

继承 PromptABC（标准）或 DIYPromptABC（绕过白名单限制）
加上 @PROMPT_REGISTRY.register() 装饰器
实现 build_prompt(self, ...) -> str 方法
若算子需限制 Prompt 类型，在算子类上用 @prompt_restrict(MyPrompt)
注意 @prompt_restrict 必须紧贴类定义上方（参见 Issue #007）

诊断流程

读取用户的报错信息
在 diagnostics/known_issues.md 中匹配已知 Issue
若命中已知 Issue，直接给出根因 + 解决方案
若未命中，结合 context/knowledge_base.md 中的架构知识进行分析
给出修复代码示例

快速匹配表（无需读文件时的速查）：

报错关键词	对应 Issue
`Unexpected key 'xxx' in operator`	Issue #001（配置参数命名，仅警告非错误）
`No object named 'Xxx' found in 'operators' registry`	Issue #002（init.py 未注册）
`Key Matching Error` / `does not match any output keys`	Issue #003（Pipeline key 不一致）
`You must call storage.step() before`	Issue #004（缺少 storage.step()）
`DummyStorage` + `AttributeError` / `TypeError`	Issue #005（DummyStorage 不支持 get_keys_from_dataframe）
`ModuleNotFoundError` + `dataflow.operators.reasoning.refine`	Issue（LazyLoader 路径，应从父模块 import）
`AttributeError: 'NoneType' object has no attribute 'strip'` + `re.split`	Issue #006（re.split 捕获组 None 问题）

规范审查流程

对用户提供的代码文件，逐项检查以下规范：

算子审查 checklist

□ 继承链：OperatorABC → super().__init__() → self.logger 可用
□ 注册：@OPERATOR_REGISTRY.register() 在类定义上方（不是函数上方）
□ run() 参数命名：input_* / output_* / storage
□ run() 返回值：list of output key names
□ storage.read() + storage.write() 都存在
□ LLM 驱动算子：self.llm_serving 命名正确
□ 容错：LLM 响应逐条 try/except，失败有默认值
□ CoT 模型输出：非 CoT 训练目标字段已剥离 <think> 标签
□ get_desc() 存在且支持 zh/en
□ __init__.py TYPE_CHECKING 块已注册
□ @prompt_restrict（如有）紧贴类定义

Pipeline 审查 checklist

□ storage 在 __init__ 中声明
□ 所有算子为成员变量（compile() 依赖此机制）
□ forward() 中每次 op.run() 传 storage.step()
□ max_workers 值合理（不要用默认的 10）
□ API key 不硬编码
□ 有 if __name__ == "__main__": 入口

知识库更新感知流程

使用 GitHub CLI 感知变更

# 前提：需要 gh CLI 已认证（gh auth login）
# 查看最近 Issues（可能包含新算子/新功能讨论）
gh issue list --repo OpenDCAI/DataFlow --limit 20 --state open

# 查看最近合并的 PR（新算子通常以 PR 形式合入）
gh pr list --repo OpenDCAI/DataFlow --state merged --limit 20

# 查看某 PR 的具体文件变更
gh pr view <PR_NUMBER> --repo OpenDCAI/DataFlow

本地变更检测

# 查看 operators 目录下最近修改的文件
git log --oneline --diff-filter=A -- 'dataflow/operators/**/*.py' | head -20

# 列出所有算子模块 __init__.py 中注册的算子名（用于对比知识库）
python -c "
import sys; sys.path.insert(0, '.')
from dataflow.utils.registry import OPERATOR_REGISTRY
OPERATOR_REGISTRY._get_all()
names = sorted(OPERATOR_REGISTRY.get_obj_map().keys())
for n in names: print(n)
"

# 快速扫描：找出所有已注册但未在 knowledge_base.md 中出现的算子

判断是否需要更新知识库

当以下任一情况发生时，建议更新知识库：

gh pr list 发现有新 PR 涉及 dataflow/operators/ 路径
git log 发现有新的算子文件（_filter.py / _generator.py / _refiner.py / _evaluator.py）
用户反馈某算子不存在于知识库中但实际存在于代码中
算子签名（__init__ / run() 参数）发生了变更

更新知识库的步骤

读取新算子文件，提取：类名、__init__ 参数、run() 参数、get_desc() 说明
在 context/knowledge_base.md §八对应模块下补充算子条目
在 context/dev_notes.md 七、版本变更记录中追加条目
提交说明：docs: sync knowledge_base with new operators from <PR/commit>

重要架构提醒（每次生成代码前隐式检查）

LazyLoader import 路径：必须从父模块 import，不能直接用子包路径

# ✅ 正确
from dataflow.operators.reasoning import ReasoningAnswerGenerator
# ❌ 错误
from dataflow.operators.reasoning.generate.reasoning_answer_generator import ReasoningAnswerGenerator

storage.step() 用法：
- Pipeline forward() 中：op.run(storage=self.storage.step(), ...)（每次调用传递）
- 独立测试脚本中：先 storage.step()，再 op.run(storage=storage, ...)
re.split() 捕获组：凡在 re.split() pattern 中用 (...)，一律改为 (?:...)
Serving 生命周期：不要在算子内手动调用 serving.cleanup()，由 Pipeline 管理

文件索引

文件	用途
`context/knowledge_base.md`	DataFlow 架构、API、目录结构、算子列表（只读参考）
`context/dev_notes.md`	开发规范、最佳实践（可追加更新）；已知问题指针指向 known_issues.md
`diagnostics/known_issues.md`	结构化 Issue 数据库，供诊断快速匹配
`templates/operator_template.py`	算子骨架模板
`templates/pipeline_template.py`	Pipeline 骨架模板
`templates/prompt_template.py`	Prompt 骨架模板
`scripts/check_updates.sh`	检测仓库变更、感知是否需要更新知识库的脚本