[扫描] orders → 发现 4 个异常:
   pk_duplicates(30行)  empty_rate(300行)  date_format(100行)  reference_integrity(50行)

[第1轮] Fixer 生成修复方案(4条SQL) → Verifier 副本验证
   ❌ 去重SQL用了 PostgreSQL 的 ctid(DuckDB 不支持) → 返回问题清单

[第2轮] Fixer 靶向修正(ctid → rowid + ROW_NUMBER) → Verifier 副本重跑质检
   ✅ 目标异常全消失、无新异常 → Orchestrator 落库 → 归档

🎉 表 orders 质检完成(2轮收敛)

flowchart LR
    subgraph data["数据层"]
        W[("warehouse.duckdb<br/>脏数据数仓")] --> R[["质量规则引擎<br/>6类检测"]]
    end

    subgraph loop["质检循环(Loop Engineering)"]
        SCAN["Python for 循环 ≤3轮<br/>扫描 → 修复 → 验证 → 收敛"] --> FIXER["Fixer 数据修复员<br/>只读+生成方案"]
        FIXER --> VERIFIER["Verifier 数据核验员<br/>副本 dry-run 验证"]
        VERIFIER --> ORCH["Orchestrator<br/>通过才落库"]
    end

    R --> SCAN
    ORCH --> W
    A[("SQLite<br/>审计")] --> ORCH

    subgraph access["接入层(复用同一核心)"]
        CLI["CLI"]
        API["FastAPI<br/>Trigger/人工介入/报告"]
        MCP["MCP Server<br/>任意 Agent 可调用"]
    end
    SCAN --> CLI & API & MCP

# Python 3.12
pip install -r requirements.txt
cp .env.example .env   # 填入 DeepSeek API Key

python scripts/gen_dirty_data.py
python scripts/scan_check.py        # M1 验收:扫描能发现全部已知问题

# 一次性跑全部表
python -m eval.run_eval             # 收敛评测:修复率/残留/新问题

# 持续监控(每30秒扫描一次)
python -m data_quality_loop.data_quality_loop --once

# FastAPI(扫描/异常/触发修复/报告/人工介入)
PYTHONPATH=src python -m uvicorn data_quality_loop.api.app:app --port 8500

# MCP Server(任何 Agent 可调用质检)
PYTHONPATH=src python -m data_quality_loop.mcp

# 双服务容器化: API(宿主 8502 → 容器 8500) + Gradio 看板(8600)
docker compose up -d --build
docker compose run --rm api python scripts/gen_dirty_data.py   # 重建数仓(首次)
docker compose restart api                                     # duckdb catalog 重新读取
curl http://localhost:8502/health                              # {"status":"ok"}

# 本地运行（需 gradio）
DQL_API_URL=http://localhost:8500 API_TOKEN=... python -m data_quality_loop.ui.app
# 或 Docker: docker compose up -d ui (端口 8600)

Data_Quality_Loop/
├── configs/
│   ├── quality_rules.yaml      # 质量规则配置(声明式,加表不改代码)
│   └── settings.yaml
├── scripts/
│   ├── gen_dirty_data.py       # 造脏数据 + 已知问题清单(种子42可复现)
│   └── scan_check.py           # M1 验收
├── data/
│   ├── warehouse.duckdb        # 被质检数仓
│   └── quality_audit.db        # SQLite 审计
├── src/data_quality_loop/
│   ├── quality_rules.py        # 质量规则引擎(6类检测)
│   ├── data_quality_loop.py    # 核心循环(Fixer/Verifier/Orch + Python控制)
│   ├── audit.py                # SQLite 审计
│   ├── api/                    # FastAPI 接入
│   └── mcp/                    # MCP Server 接入
├── skills/data-quality-fixer/
│   └── SKILL.md                # 质量修复标准(渐进加载)
├── eval/
│   ├── known_issues.json       # 已知问题清单(评测基准)
│   ├── run_eval.py             # 收敛评测
│   └── test_mcp.py             # MCP 冒烟测试
└── tests/

python -m eval.run_eval

// Claude Desktop / Claude Code 配置
{
  "mcpServers": {
    "data-quality-loop": {
      "command": "python",
      "args": ["-m", "data_quality_loop.mcp"],
      "env": { "PYTHONPATH": "D:/Data_Quality_Loop/src" }
    }
  }
}