standing-on-giants/Data-Pipeline-Incident-Response
GitHub: standing-on-giants/Data-Pipeline-Incident-Response
一个用于评估和基准测试智能体 LLM 动态诊断与修复数据管道故障能力的强化学习环境。
Stars: 0 | Forks: 0
title: 数据管道故障响应 — OpenEnv
colorFrom: red
colorTo: orange
sdk: docker
app_port: 7860
tags:
- openenv
- data-engineering
- rl-environment
- agentic
license: MIT
# 数据管道故障响应 — OpenEnv
一个 RL 环境，其中 agent 作为值班数据工程师，诊断损坏的 pipeline，调查数据质量问题，并应用根因修复。
## 动机
数据团队每天都在处理损坏的 pipeline。供应商静默更改了 API 导出格式，关键源表中突然出现 null 值，或者重复的行导致下游财务聚合数据虚高。解决这些问题需要的不仅仅是代码生成——它需要**动态地调查数据**。
在这个环境中，agent 会收到包含 DAG 中失败断言的警报。它必须查询数据（`read_data_sample`、`check_schema`）以查明出了什么问题，应用正确的转换补丁（`add_data_filter`、`patch_transformation`），通过重新运行 pipeline 来验证修复，并知道何时将真正损坏的数据上报给上游。
该环境在需要**数据驱动推理**、**迭代调试**以及区分本地可修复 bug 和上游数据损坏的任务上对 agentic LLM 进行基准测试。
## 项目结构
```
data_pipeline_env/
├── Dockerfile
├── openenv.yaml
├── requirements.txt
├── new_inference.py Agent loop, prompt builder, action parser (local use)
└── src/
├── __init__.py
├── environment.py OpenEnv-compliant env (reset / step / state)
├── pipeline_runner.py Simulates DAG execution and data transformations
├── assertions.py Deterministic data quality grader
├── tasks.py Task definitions (DAGs, faults, schemas)
├── models.py Pydantic models (PipelineAction, PipelineObservation, …)
└── server.py FastAPI REST wrapper
```
# 数据管道故障响应 — OpenEnv
一个用于 agent 诊断和修复损坏数据管道的 RL 环境。
**真实世界任务**：每个数据团队都会在夜间处理 pipeline 故障。agent 作为值班数据工程师：它读取失败的断言报告，检查原始数据，确定根本原因，并应用修复——这正是人类在凌晨 3 点会做的事情。
## 快速开始
```
docker build -t data-pipeline-env .
docker run -p 7860:7860 data-pipeline-env
```
### 本地 (Python conda/venv)
```
# 安装依赖
pip install -r requirements.txt
# 对所有 3 个任务运行 inference（需要 API key）
export API_BASE_URL="https://router.huggingface.co/v1"
export MODEL_NAME="meta-llama/Llama-3.3-70B-Instruct"
export HF_TOKEN="your_token_here"
Push this repo to a Hugging Face Space with `sdk: docker` in the README frontmatter. The Space builds and serves automatically.
```bash
git clone https://huggingface.co/spaces//data-pipeline-incident-response
cp -r . data-pipeline-incident-response/
cd data-pipeline-incident-response
git add . && git commit -m "initial" && git push
python inference.py
python inference.py --task easy # single task
python inference.py --task hard --steps 25
```
## 环境概述
这些仅在本地运行 `new_inference.py` 时需要。Space 本身没有模型依赖——它只提供环境 API。
| 变量 | 必需 | 默认 | 描述 |
|---|---|---|---|
| `GEMINI_API_KEY` | 是 (本地) | — | 用于 Gemini 推理的 API key |
| `MODEL_NAME` | 否 | `gemini-2.5-flash` | 模型覆盖 |
| `MAX_STEPS` | 否 | `20` | 每 episode 的最大步数 |
| `TEMPERATURE` | 否 | `0.1` | 采样温度 |
| `MAX_TOKENS` | 否 | `1024` | 每次 LLM 调用的最大 token 数 |
## 用法
### REST API
```
# 健康检查
curl http://localhost:7860/health
# 列出任务
curl http://localhost:7860/tasks
# 启动 session
curl -X POST http://localhost:7860/reset \
-H "Content-Type: application/json" \
-d '{"task_id": "easy", "max_steps": 20}'
# → {"session_id": "easy_...", "observation": {...}}
# 执行一步
curl -X POST http://localhost:7860/step \
-H "Content-Type: application/json" \
-d '{
"session_id": "",
"action": {
"action_type": "read_data_sample",
"params": {"table": "raw_orders", "n_rows": 20}
}
}'
# 获取 session 状态
curl "http://localhost:7860/state?session_id="
```
### 本地 agent 运行器
```
pip install openai python-dotenv # in addition to requirements.txt
# 在 .env 文件中设置你的 GEMINI_API_KEY
python new_inference.py # all three tasks
python new_inference.py --task easy
python new_inference.py --task hard --steps 25
python new_inference.py --quiet # suppress per-step output
```
## 环境描述
```
from src.environment import DataPipelineEnv
from src.models import PipelineAction
env = DataPipelineEnv(task_id="easy") # "easy" | "medium" | "hard"
obs = env.reset() # → PipelineObservation
result = env.step(action) # → StepResult
state = env.state() # → Dict
```
当所有断言通过（`pipeline_passed=True`）或步数预算耗尽时，Episode 结束。
## 观察空间
`PipelineObservation` 字段：
| 字段 | 类型 | 描述 |
|---|---|---|
| `task_id` | `str` | `"easy"`、`"medium"` 或 `"hard"` |
| `difficulty` | `str` | 人类可读的难度标签 |
| `description` | `str` | 任务上下文（什么坏了） |
| `max_steps` | `int` | 步数预算 |
| `dag_structure` | `List[DAGNode]` | pipeline 图（步骤 + schema） |
| `failed_assertions` | `List[AssertionResult]` | 当前失败的数据质量检查 |
| `passed_assertions` | `List[AssertionResult]` | 当前通过的数据质量检查 |
| `historical_runs` | `List[HistoricalRun]` | 跨越数天的 pipeline 历史记录 |
| `data_sample` | `Optional[List[Dict]]` | 调用 `read_data_sample` 时填充 |
| `current_schema` | `Optional[Dict]` | 调用 `check_schema` 时填充 |
| `historical_schema` | `Optional[Dict]` | 调用 `compare_schema` 时填充 |
| `actions_taken` | `List[str]` | 最近的操作历史 |
| `last_action_result` | `str` | 上一个操作的文本结果 |
| `step_number` | `int` | 当前步数 |
| `pipeline_passed` | `bool` | 当所有断言通过时为 `True` |
### 问题所在
数据管道在夜间运行。它：
1. 从上游源（Salesforce、供应商、CRM）中**提取**数据
2. 对其进行**转换**（清理、连接、聚合）
3. 将其**加载**到仪表板和 ML 模型使用的下游表中
**数据质量断言**检查如下内容：
- `user_id IS NOT NULL`
- `order_item_id is UNIQUE`
- `revenue is numeric and in [0, 1,000,000]`
当断言失败时，pipeline 会停止。agent 的工作是找出原因并修复它。
## 任务
| 任务 | 故障 | 表 | 断言 | 难度 |
|--------|-------------------------------------|--------|------------|------------|
| easy | 来自上游的 5 个 null user_id | 2 | 3 | Easy |
| medium | 20 个重复的 order_item_id | 3 | 4 | Medium |
| hard | 收入格式更改 + N/A 行 | 4 | 6 | Hard |
## 动作空间
| 动作 | 参数 | 目的 |
|---|---|---|
| `read_data_sample` | `table`, `n_rows` | 通过检查实际行进行诊断 |
| `check_schema` | `table` | 通过检查当前列 dtypes 进行诊断 |
| `compare_schema` | `table` | 对比当前与历史 schema 以发现漂移 |
| `add_data_filter` | `step_id`, `filter_condition` | 修复：删除坏记录（例如 `user_id IS NOT NULL`） |
| `patch_transformation` | `step_id`, `patch_type`, `column` | 修复：原地清理数据（`coalesce`、`dedup`、`parse_currency`） |
| `run_pipeline` | *(无)* | 重新编译并触发 DAG，生成新的断言状态 |
| `alert_upstream_team` | `team`, `issue_description` | 正确上报无法修复的上游损坏 |
| `mark_acceptable` | `assertion_id`, `reason` | *(反模式)* 将失败的断言标记为忽略 |
**奖励：** 检查数据会因刷接口受到轻微惩罚（`-0.05`），盲目操作（未读取数据就添加过滤器）会受到严重惩罚（`-0.5`），正确上报无法修复的数据可获得 `+0.5`，修复 pipeline 会根据新断言通过或失败的情况获得分数。通过完整的 pipeline 可获得 `+1.0` 奖励。
## 任务描述
### easy — Null 值
| 问题 | 详细信息 |
|---|---|
| **故障** | 上游表添加了 `discount_code` 列，意外地导致 5 个随机行的 `user_id` 变为 null。 |
| **症状** | 在 `orders_clean.user_id` 上 `NOT NULL` 断言失败。 |
| **解决方案** | `add_data_filter` (`user_id IS NOT NULL`) |
| 动作 | 描述 |
|------------------------|------------------------------------------------------|
| `read_data_sample` | 查看表的前 N 行 |
| `check_schema` | 查看当前列类型 |
| `compare_schema` | 对比当前 schema 与历史 |
| `run_quality_assertion`| 重新运行特定断言 |
| `add_data_filter` | 向 pipeline 步骤添加 WHERE 子句 |
| `patch_transformation` | 应用列修复 (cast, dedup, parse_currency) |
| `backfill_partition` | 重新运行特定日期的 pipeline |
| `alert_upstream_team` | 将无法修复的问题上报给数据源所有者 |
| `mark_acceptable` | 接受已知问题（如果错误将被惩罚） |
| `run_pipeline` | 重新执行完整 pipeline 并检查所有断言 |
## 奖励设计
| 事件 | 奖励 |
|---------------------------------------------------|-----------|
| 断言从失败 → 通过（每次运行） | +0.4 |
| 断言从通过 → 失败（每次运行） | −0.5 |
| 所有断言通过（终止） | +1.0 |
| 未先读取数据就应用修复 | −0.5 |
| 正确的上游警报（hard 任务） | +0.5 |
| 对确实失败的断言使用 `mark_acceptable` | −1.0 |
## 运行服务器 (Docker)
```
# 构建
docker build -t data-pipeline-env .
# 运行
docker run -p 8001:8001 data-pipeline-env
# 健康检查
curl http://localhost:8001/health
# WebSocket endpoint: ws://localhost:8001/ws
```
### medium — 重复行
| 问题 | 详细信息 |
|---|---|
| **故障** | 供应商为昨天的数据发送了 20 个重复的订单项。 |
| **症状** | `order_item_id` 上的唯一约束失败，并且 `order_summary` 行数虚高。 |
| **解决方案** | 在 `transform_items` 步骤上执行 `patch_transformation` (type=`dedup`, column=`order_item_id`)。 |
### hard — Meta Ads 与 Conversions Pipeline（级联故障）
| 问题 | 详细信息 |
|---|---|
| **故障 1** | Graph API v19.0 将 `spend` 更改为 `"$1,234.56"` 字符串。由于 API 中断，约 10 行包含 `"N/A"`。 |
| **故障 2** | `impressions` 包含 `"N/A"` 值；将其合并为 0 会导致 CTR 除以零 (Inf/NaN)。 |
| **故障 3** | Conversions API 重试失败的 payload → 约 37 行重复的 `event_id` (15%)。 |
| **故障 4** | conversions 中的 `campaign_id` 带有 `"CMP_"` 前缀；insights 使用 int → ROAS join 丢失了 90%+ 的行。 |
| **症状** | 3 个输出表中共有 8 个断言失败：`spend` 上的类型/范围，`ctr` 上的范围，conversions 上的唯一性/行数，`roas_summary` 上的行数/范围。 |
| **解决方案** | 顺序执行：在 `spend` 上执行 `parse_currency` + `coalesce`，在 `impressions` 上执行 `parse_currency` + `coalesce(1)`，在 `event_id` 上执行 `dedup`，在 conversions `campaign_id` 上执行 `strip_prefix` + `cast_column`，+ `alert_upstream_team(meta_ads_api_team)`。 |
## 基准分数
模型：`gemini-2.5-pro`。`MAX_STEPS=20`，`TEMPERATURE=0.1`。
| 任务 | 得分 | 奖励 | 步数 | 断言通过数 | 是否通过 |
|---|---|---|---|---|---|
| easy | 1.00 | +1.35 | 3 | 3 / 3 | 是 |
| medium | 1.00 | +1.75 | 3 | 4 / 4 | 是 |
| hard | — | — | — | — / 8 | 待定 |
| **平均** | **—** | **—** | **—** | — | **—** |
### WebSocket 协议
```
// Reset
{"action": "reset", "task_id": "easy"}
// Step
{"action": {"action_type": "read_data_sample",
"params": {"table": "raw_orders", "n_rows": 20}}}
// Get state
{"action": "state"}
```
## 评分
评分器是完全确定性的：
```
score = assertions_passing / assertions_total ∈ [0.0, 1.0]
```
Hard 任务要求同时满足：
- 应用正确的数据修复（C1, C2, C3, C4, C6 通过）
- 针对损坏的行向正确的上游团队报警
## 项目结构
```
data_pipeline_env/
├── inference.py ← Baseline LLM agent
├── openenv.yaml ← OpenEnv spec metadata
├── requirements.txt
├── Dockerfile
└── src/
├── __init__.py
├── models.py ← Pydantic Observation / Action / StepResult
├── environment.py ← DataPipelineEnv (reset / step / state)
├── tasks.py ← Easy / Medium / Hard task definitions
├── assertions.py ← Deterministic assertion checker
├── pipeline_runner.py← Pandas-based pipeline executor
└── server.py ← FastAPI WebSocket server (OpenEnv spec)
```
标签：AI智能体, LLM评测, 强化学习环境, 数据工程, 数据质量, 模块化设计, 自动化修复, 请求拦截, 逆向工具