MohanChippala123/lineage-sre

# Lineage SRE 🚨 **一个由 DataHub 驱动的、用于数据 pipeline 的自主事件响应 agent。** 当 pipeline 在凌晨 3 点发生故障时，人类数据工程师会打开 DataHub，沿血缘图谱向上游追溯，对比 schema，在 Slack 上通知 owner，修补 dbt 模型，并且——如果团队运气好的话——写下一篇六个月后没人能找到的复盘报告。 Lineage SRE 是一个能在约一分钟内完成所有这些工作的 AI agent，**并且会将其学到的知识写回 DataHub**，这样下一位工程师（或下一个 agent）就能直接继承这些知识，而不是从零开始重新诊断。 ## 闭环 ``` detect ──> diagnose ──> fix ──> verify ──> write back │ │ │ │ │ │ walks DataHub │ re-runs the raises a DataHub health lineage upstream,│ health check incident + appends check diffs model SQL │ a postmortem to the vs live schemas │ asset docs in DataHub │ generates a contract-preserving SQL fix, validates it, applies it ``` 具体来说，给定一个失败的模型，该 agent 会： 1. **复现**针对数据仓库（demo 中为 DuckDB）的故障。 2. **在 DataHub 中向上游追溯 lineage**，从失败的资产开始，对于每一跳，将模型 SQL *期望*的结构与 DataHub 中的*当前* schema 进行比较。一旦发现不匹配，就能定位根本原因（在 demo 中：供应商在原始数据流中悄无声息地将 `amount_usd` 重命名为 `amount` —— 这是典型的静默 schema 破坏）。 3. **通过向下游追溯 lineage 来衡量影响范围** —— 包括 ML 资产：demo 中的 pipeline 向一个流失模型（churn-model）的特征表及其评分输出供数据，因此 agent 会在其造成经济损失*之前*报告“你的流失模型正在基于损坏的特征进行评分”。 4. **从 DataHub 的 ownership 元数据中确定需要通知的人员**。 5. **生成一个最小修复方案**，该方案能保留下游契约（为重命名的列设置别名，而不是传播重命名），使用真实的 SQL 对其进行验证，将其作为可审查的 artifact 写入 `fixes/`，并且在带有 `--apply` 参数的情况下，应用该修复并重新验证整个 pipeline。 6. **将知识写回 DataHub**：在根本原因资产上发起一个 `DATA_SCHEMA` 事件，将复盘报告追加到资产文档中，并且只有在修复经过验证后才解决该事件。之后在 DataHub UI 中打开该数据集 —— 事件和复盘报告就会赫然在目。 ## 架构 ``` flowchart LR subgraph Agent["Lineage SRE agent (Claude tool-use loop)"] A[Diagnosis loop] end subgraph DataHub["DataHub"] L[Lineage graph] S[Schemas & properties] O[Ownership] I[Incidents] D[Docs] end subgraph Stack["Data stack"] W[(DuckDB warehouse)] M[SQL model repo] end A -- "reads: MCP Server (--mcp)\nor GraphQL (default)" --> L & S & O A -- "writes back: incidents, postmortem docs" --> I & D A -- "run_sql / apply_fix" --> W A -- "get_model_sql / propose_fix" --> M ``` 两种读取模式，一条写入路径： | 模式 | DataHub 读取方式 | 备注 | |---|---|---| | default | 内置 GraphQL 工具 | 零额外依赖，随时可用 | | `--mcp` | **官方 [DataHub MCP Server](https://docs.datahub.com/docs/features/feature-guides/mcp)** (`mcp-server-datahub`) | 通过官方服务器进行搜索、lineage、schema 和查询分析 | 回写操作（通过 `raiseIncident` 发起事件，通过 `updateDescription` 写入复盘报告）始终通过 DataHub 的 GraphQL API 进行 —— MCP 服务器专注于读取，而向图谱中贡献数据才是核心目的。 ## 快速开始 前置条件：[uv](https://docs.astral.sh/uv/) 和一个 DataHub 实例 —— 以下任意一种即可： - **DataHub Cloud 免费试用**（无需 Docker）：在 [datahub.com/free-trial](https://datahub.com/free-trial/) 注册， 生成一个个人访问令牌（Settings → Access Tokens），并在 `.env` 中设置 `DATAHUB_GMS_URL=https://.acryl.io/gms` 和 `DATAHUB_TOKEN`。 - **本地快速启动**（需 Docker）：`uv tool install acryl-datahub && datahub docker quickstart`， 然后默认的 `DATAHUB_GMS_URL=http://localhost:8080` 即可直接使用。 该 agent 可以在三种后端上运行（`--engine gemini|api|sdk`，从 `.env` 中自动检测）： | 引擎 | 需求 | 成本 | |---|---|---| | `gemini` | `GEMINI_API_KEY`（[免费，免信用卡](https://aistudio.google.com/apikey)） | 免费额度 | | `api` | `ANTHROPIC_API_KEY` | 按 token 付费 | | `sdk` | 登录 [Claude Code](https://claude.com/claude-code)，无需密钥 | 包含在您的 Claude 订阅中 | ``` # 安装 git clone && cd lineage-sre uv sync # add: --extra mcp for MCP mode cp .env.example .env # point it at your DataHub instance # 端到端运行整个 story uv run lineage-sre demo ``` 或者分步进行： ``` uv run lineage-sre seed # build the demo warehouse + ingest metadata into DataHub uv run lineage-sre break # the vendor ships feed v2: amount_usd -> amount 💥 uv run lineage-sre check # health check: 3 of 4 models FAILING uv run lineage-sre diagnose --apply # the agent takes it from here uv run lineage-sre diagnose --apply --mcp # same, reading DataHub via the official MCP server ``` 然后打开 http://localhost:9002，找到 `demo.raw_payments`，查看 **Incidents** 标签页和 **Documentation** 标签页 —— agent 的工作成果已经记录在图谱中。 ## Demo pipeline ``` raw_payments ──> stg_payments ──> fct_daily_revenue (exec dashboard) raw_customers ──> stg_customers ──> churn_features ──> churn_model_predictions (production ML) stg_payments ──┘ ``` 数据集、schema、列级别类型、lineage、两个 owner（数据平台 vs. 数据分析）以及供应商数据流元数据（`feed_version` 自定义属性）都通过 `seed` 命令摄取到 DataHub 中。`break` 命令模拟了夜间数据摄取时获取到供应商静默重命名列的情况 —— DataHub 中关于 `raw_payments` 的 schema 更新了，而下游的 SQL 并未更新，导致从执行仪表板到流失模型的一切都发生了故障。 ## 示例输出 该 agent 生成的 artifact 位于 [`examples/`](examples/)： - [`rca_report_example.md`](examples/rca_report_example.md) — agent 生成的完整根本原因分析（RCA）报告 - [`fix_stg_payments.sql`](examples/fix_stg_payments.sql) — 生成的保留契约的修复脚本 每次实际运行还会生成 `reports/rca_.md` 和 `fixes/.sql` 文件。 ## 它是如何利用 DataHub 的（致评委） **读取** — lineage 遍历（`searchAcrossLineage` / MCP lineage 工具）、schema 检查、ownership 查找、自定义属性（供应商数据流版本是诊断线索）、数据集搜索。 **回写** — `raiseIncident`（在根因资产上标记为 `DATA_SCHEMA` 类型）、`updateIncidentStatus`（仅在验证修复后解决）、`updateDescription`（将复盘报告追加到资产文档中）。经过 agent 处理后，图谱的状态比之前更好了：事件历史和复盘报告被永久保存在对应的资产上。 **超越开箱即用** — DataHub 为您展示 lineage；而 Lineage SRE 将其* operationalizes（运营化）*：自主实现 lineage → 根本原因 → 保留契约的修复 → 验证修复 → 形成组织记忆。 ## 仓库结构 ``` src/lineage_sre/ agent.py # Claude tool-use loop + RCA report tools.py # tool definitions & dispatch (reads + write-back actions) datahub_client.py # GraphQL reads, incidents, docs write-back, REST emitter mcp_bridge.py # official DataHub MCP Server integration (--mcp) demo_stack.py # demo pipeline metadata + break scenario warehouse.py # DuckDB warehouse, health checks, guarded SQL cli.py # seed / break / check / diagnose / demo demo/models/ # the pipeline's SQL models (dbt-style) examples/ # sample agent outputs for judges docs/ # demo video script ``` ## 许可证 [Apache 2.0](LICENSE)

标签：AIOps, DataHub, 数据工程, 数据血缘, 数据运维, 模块化设计, 自动化修复, 逆向工具