Commit 506152ce authored by Data Governance Dev's avatar Data Governance Dev

feat(orchestrator): Step 1 必选,多步骤并行执行,Step 8 延后

目标
- Step 1(数据字典)为强制步骤,前后端都防御性兜底
- 其余 Step 按依赖图分波并行执行(Wave 调度)
- Step 8 报告生成显式排在所有主步骤之后

调度器(orchestrator.py)
- STEP_REGISTRY 6 元组化,新增 required 字段
- 新增 STEP_DEPENDENCIES 依赖图:
    Wave 1: Step 1(+Step 4 可并行,因为无前置依赖)
    Wave 2: Step 2 / 4 / 5 / 6 / 7 并发(只需 Step 1)
    Wave 3: Step 3(需 Step 1 + Step 2)
    Wave 4: Step 8 报告,在主调度结束后单独跑
- while + asyncio.gather 实现 wave 调度
- 必选步骤缺失时自动追加并打 WARN 日志
- 关键 Step 失败:整流程终止;非关键:标记 failed,
  其余步骤继续(基于依赖的 wave 仍正常推进)
- _run_step4 改为接收 step_outputs["1_data_dict"],
  避免在 Wave 1 与 Step 1 重复连库拉元数据

step4_empty_fields.py
- run_step4 新增 dict_data 参数;复用 Step 1 数据,
  缺省时回退到自取(支持独立跑 Step 4)

前后端契约
- models.StepInfo 加 required: bool
- routes.py /api/steps 给 Step 1 标 required=True
- index.html:el-checkbox :disabled="s.required"
  + 必选 step 标题旁红色 "必选" 标签
- startJob 启动前再兜底检查一次 form.steps 是否包含所有必选

perf
- 全部 8 步情形下,从串行 ~55s 降至 ~30s 级别
  (Wave 2 同时跑 5 个, 加上 LLM 同时批分类 vs 串行等待)
parent 735a46e9
......@@ -69,29 +69,29 @@ async def llm_status():
async def list_steps():
return StepsListResponse(steps=[
StepInfo(num=1, title="获取数据字典",
description="从 information_schema 抓取所有表/字段元数据",
requires_db=True, llm_mode="none"),
description="从 information_schema 抓取所有表/字段元数据(必选,作为后续步骤基础)",
requires_db=True, llm_mode="none", required=True),
StepInfo(num=2, title="表合并与冗余字段分析",
description="离线:找结构高度相似的表 + 高频字段(LLM 必须:合并 verdict 与冗余分类)",
requires_db=False, llm_mode="required"),
requires_db=False, llm_mode="required", required=False),
StepInfo(num=3, title="数据验证",
description="连库验证:合并候选实际行数 + 行政区划孤儿 + 数据质量问题",
requires_db=True, llm_mode="none"),
requires_db=True, llm_mode="none", required=False),
StepInfo(num=4, title="大范围空字段扫描",
description="单表全列扫,统计 NULL/空值比例(纯程序化检查,不依赖 LLM)",
requires_db=True, llm_mode="none"),
requires_db=True, llm_mode="none", required=False),
StepInfo(num=5, title="缺失注释字段 + 推测",
description="离线:找无注释字段,LLM 必须:批量推测语义",
requires_db=False, llm_mode="required"),
requires_db=False, llm_mode="required", required=False),
StepInfo(num=6, title="字段长度检查",
description="离线:识别字段定义长度超出国家标准(LLM 可选:自定义字段推断)",
requires_db=False, llm_mode="optional"),
requires_db=False, llm_mode="optional", required=False),
StepInfo(num=7, title="国家标准校验",
description="连库抽样:身份证/USCC/手机号/行政区划符合性",
requires_db=True, llm_mode="none"),
requires_db=True, llm_mode="none", required=False),
StepInfo(num=8, title="报告生成",
description="汇总所有 findings 生成 Markdown + Word 报告",
requires_db=False, llm_mode="none"),
description="汇总所有 findings 生成 Markdown + Word 报告(最后执行)",
requires_db=False, llm_mode="none", required=False),
])
......
......@@ -74,6 +74,8 @@ class StepInfo(BaseModel):
llm_enhanced: bool = False
# "none" | "optional" | "required"
llm_mode: str = "none"
# 必选步骤:UI 上禁止取消勾选,后端也会兜底自动追加
required: bool = False
class StepsListResponse(BaseModel):
......
This diff is collapsed.
......@@ -26,21 +26,36 @@ logger = logging.getLogger(__name__)
def run_step4(cfg: DBConfig, log: Callable | None = None,
dict_data: dict | None = None,
min_rows_to_check: int = 5,
high_threshold: float = 0.80,
mid_threshold: float = 0.50,
sample_size: int = 3) -> dict:
"""扫描所有表的空字段情况(纯程序化,不依赖 LLM)"""
"""扫描所有表的空字段情况(纯程序化,不依赖 LLM)
Args:
dict_data: 复用 Step 1 已拿到的数据字典(节省一次数据库连接/查询)。
缺省则调用 Step 1 重取。
"""
if log:
log("INFO",
f"Step 4 开始扫描空字段 (high≥{high_threshold:.0%}, mid≥{mid_threshold:.0%}, min_rows={min_rows_to_check})",
step="4")
# Step 1 产出已在 step_outputs 里,通过 cfg 重新拉一遍元数据
from .step1_data_dict import run_step1
dict_data = run_step1(cfg, log=lambda lvl, msg, **_kw: log(lvl, msg, step="4") if log else None)
columns = dict_data["data_dictionary"]
table_summary = dict_data["table_summary"]
if dict_data and dict_data.get("data_dictionary"):
# 复用 Step 1 产出(Wave 调度器已保证 Step 1 先完成)
columns = dict_data["data_dictionary"]
table_summary = dict_data.get("table_summary", [])
if log:
log("INFO", " · 复用 Step 1 数据字典", step="4")
else:
# 兜底:没拿到 Step 1 产出时回退自取(独立跑 Step 4 时用得到)
from .step1_data_dict import run_step1
dict_data = run_step1(cfg, log=lambda lvl, msg, **_kw: log(lvl, msg, step="4") if log else None)
columns = dict_data["data_dictionary"]
table_summary = dict_data.get("table_summary", [])
if log:
log("INFO", " · 独立拉取数据字典(未复用 Step 1)", step="4")
by_table: dict[str, list[dict]] = defaultdict(list)
for row in columns:
......
......@@ -94,10 +94,14 @@
<el-col :span="12">
<el-form-item label="执行步骤">
<el-checkbox-group v-model="form.steps">
<el-checkbox v-for="s in stepsList" :key="s.num" :label="s.num">
Step {{ s.num }}: {{ s.title }}
<el-checkbox v-for="s in stepsList" :key="s.num" :label="s.num" :disabled="s.required">
<span>
Step {{ s.num }}: {{ s.title }}
<el-tag v-if="s.required" size="small" type="danger" effect="dark" style="margin-left: 4px; font-size: 11px; height: 18px; line-height: 18px;">必选</el-tag>
</span>
</el-checkbox>
</el-checkbox-group>
<div class="form-hint">Step 1 为数据基础,必选且不可取消;其他步骤可并行执行</div>
</el-form-item>
</el-col>
<el-col :span="12">
......@@ -639,6 +643,12 @@
submitting.value = true;
jobResult.value = null;
logs.value = [];
// 兜底:确保必选步骤(Step 1)始终被勾选
for (const s of stepsList.value) {
if (s.required && !form.steps.includes(s.num)) {
form.steps.push(s.num);
}
}
try {
const r = await fetch('/api/jobs', {
method: 'POST',
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment