内容
# 数据质量代理
[](https://github.com/ARAVINDHRAJA123/data-quality-agent/actions/workflows/ci.yml)
一个自动化的AI系统，能够自动调查dbt测试失败，追踪根本原因，并生成一份简洁的事件报告 — 将调查时间从数小时缩短至数分钟。
📐 [架构图](docs/architecture.html) · 📄 [示例事件报告](docs/sample_report.md)
---
## 架构
---
## 功能
当dbt测试失败时，通常会收到一个神秘的错误消息。该代理：
1. **获取失败的行** 从BigQuery — 查看实际的坏数据
2. **读取dbt清单** — 了解完整的血缘图
3. **追踪上游** — 分析父模型和源表的列
4. **识别根本原因** — 找到坏数据进入管道的位置
5. **编写事件报告** — 简洁的根本原因、血缘追踪、推荐的修复、严重程度
```
not_null_fct_transactions_merchant 失败（23行）
│
▼
代理获取失败的行 → 读取 fct 血缘 → 追踪到 int_ → 追踪到 stg_ → 检查原始源
│
▼
根本原因：raw.bank_transactions 中的 23 行有 NULL 说明。
商家提取返回 NULL 时，说明为 NULL。
修复：在 stg_bank__transactions 中添加 COALESCE(narration, '')。
严重程度：高
```
---
## 三种触发模式
### 1 — 命令行界面（CLI）
```bash
python agent.py \
--test not_null_fct_transactions_merchant \
--model fct_transactions \
--column merchant \
--verbose
```
### 2 — Webhook（Airflow 或任何 HTTP 调用者）
```bash
python server.py # 在端口 5051 上启动
curl -X POST http://localhost:5051/investigate \
-H "Content-Type: application/json" \
-d '{"test_name": "not_null_fct_transactions_merchant", "model": "fct_transactions", "column": "merchant"}'
```
将 Airflow DAG 的 `on_failure_callback` 指向此端点。
### 3 — MCP（任何 AI 客户端）
MCP 服务器向任何 MCP 兼容的客户端（如 Claude Code、OpenClaw（ChatGPT / Gemini / 任何客户端）、Cursor、Zed）公开三个工具：
| 工具 | 功能 |
|---|---|
| `investigate_failure` | 完整的代理调查 → 事件报告 |
| `list_failures` | 从 run_results.json 列出失败的测试 |
| `get_report` | 读取保存的事件报告 |
**Claude Code：**
```bash
claude mcp add -s user \
-e GCP_PROJECT=your-project \
-e BQ_LOCATION=asia-south1 \
-e DBT_MANIFEST_PATH=/path/to/dbt_bank/target/manifest.json \
-e DBT_RUN_RESULTS_PATH=/path/to/dbt_bank/target/run_results.json \
-e GEMINI_API_KEY=your-key \
dbt-investigator \
-- /path/to/venv/bin/python /path/to/mcp_server.py
```
**OpenClaw（ChatGPT、Gemini 或任何其他客户端）：**
```bash
openclaw mcp set dbt-investigator '{
"command": "/path/to/venv/bin/python",
"args": ["/path/to/mcp_server.py"],
"cwd": "/path/to/data-quality-agent",
"env": {
"GCP_PROJECT": "your-project",
"GEMINI_API_KEY": "your-key",
"DBT_MANIFEST_PATH": "/path/to/manifest.json",
"DBT_RUN_RESULTS_PATH": "/path/to/run_results.json"
}
}'
openclaw mcp probe # → dbt-investigator: 3 个工具 ✔
```
---
## 代理工具
| 工具 | 代理调用 |
|---|---|
| `get_failing_rows` | 查询 BigQuery 获取实际的坏行 |
| `get_model_lineage` | 读取 manifest.json 获取上游/下游 |
| `get_model_sql` | 获取任何模型的编译 SQL |
| `get_column_profile` | 空值计数、不同计数、最小值、最大值 |
| `run_query` | 自定义只读 BigQuery 调查 |
| `get_source_freshness` | 检查源表的过时情况 |
| `write_report` | 编写最终的事件报告 |
**安全墙：** 所有 BigQuery 查询均为只读（仅限 SELECT/WITH）。DML/DDL 在执行前被拒绝。
---
## 设置
```bash
git clone https://github.com/ARAVINDHRAJA123/data-quality-agent.git
cd data-quality-agent
python3 -m venv venv && source venv/bin/activate
pip install -r requirements.txt
# 身份验证
gcloud auth application-default login
# 设置环境
export GCP_PROJECT=your-project
export BQ_LOCATION=asia-south1
export DBT_MANIFEST_PATH=/path/to/dbt_bank/target/manifest.json
export DBT_RUN_RESULTS_PATH=/path/to/dbt_bank/target/run_results.json
# LLM（选择一个）
export GEMINI_API_KEY=your-key # 免费
export ANTHROPIC_API_KEY=your-key # 付费
```
首先生成清单（从您的 dbt 项目）：
```bash
cd /path/to/dbt_project && dbt compile
# manifest.json 现在位于 target/manifest.json
```
---
## 技术栈
- **Claude / Gemini** — LLM 提供商（自动检测，免费 Gemini 支持）
- **BigQuery** — 数据仓库（GCP）
- **dbt manifest.json** — 血缘图和编译 SQL
- **FastMCP** — MCP 服务器（任何 AI 客户端）
- **Flask** — Webhook 服务器（Airflow 集成）
- **pytest** — 测试套件
---
## 项目结构
```
data-quality-agent/
├── agent.py ← 代理调查循环（Claude + Gemini）
├── server.py ← Flask Webhook 服务器
├── mcp_server.py ← FastMCP 服务器（任何 MCP 客户端）
├── report.py ← 事件报告格式化器
├── tools/
│ ├── bq_tools.py ← BigQuery：失败的行、查询、过时情况
│ └── dbt_tools.py ← 清单：血缘、SQL、测试结果
├── tests/
│ └── test_tools.py ← 11 个单元测试（无需 BQ/LLM）
├── reports/ ← 保存的事件报告（Markdown）
└── requirements.txt
```
