内容
# dagsmith
**数据工程的智能脚手架。** 将新的数据源的普通英文描述转换为完整的、经过测试的、生产就绪的管道：dbt 模型（分阶段 → 市场）、Airflow DAG、Great Expectations 套件和新鲜度监控器 — 由一个 LLM 代理生成、执行和自我纠正，该代理通过 MCP 读取仓库的实时元数据。
```
┌───────────────────────────────────────────────────────────────────────┐
│ │
│ 分析师写道： │
│ "从 Postgres 摄取每日注册事件。粒度 = 每个 registry_id + item_id 的一个事件。加入客户进行 LTV汇总。" │
│ │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ dagsmith 代理 │ │
│ │ (Claude Sonnet 5) │ │
│ └──┬──────────────┬───┘ │
│ │ │ │
│ MCP 工具 ───────▶│ │──── Jinja 模板 │
│ (Snowflake 元数据) │ │ + 验证循环 │
│ ▼ ▼ │
│ ┌─────────────────┐ ┌─────────────────┐ │
│ │ dbt 模型 │ │ Airflow DAG │ │
│ │ stg_*.sql │ │ registry_ev.py │ │
│ │ fct_*.sql │ │ │ │
│ │ schema.yml │ │ GE 检查点 │ │
│ └─────────────────┘ └─────────────────┘ │
│ │ │ │
│ └──────┬───────┘ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ 验证器 │ │
│ │ (真实 Snowflake + │ │
│ │ 沙箱 dbt 运行) │ │
│ └──────┬──────────────┘ │
│ │ │
│ 通过 ──┴── 失败 → 循环返回代理 │
│ │
└───────────────────────────────────────────────────────────────────────┘
```
## 为什么存在
每个数据工程师为每个新数据源构建相同的八个文件：一个分阶段模型、一个市场模型、一个 schema.yml、一个快照配置、一个 DAG、一个 GE 套件、一个监视器、一个文档条目。 90% 的工作是机械的 — 但它不够同质化，无法用纯代码模板化。 这个差距是代理很好地覆盖的。
`dagsmith` 将管道生成视为一个**编译步骤**，而不是一个检查清单。您描述意图；代理生成工件；验证器在沙箱模式下针对真实的 Snowflake 运行它们；失败循环返回到代理的上下文，直到它们通过。 当您合并时，您会得到真正的经过测试的代码 — 而不是演示。
## 设计
代理是一个 Claude Sonnet 5 工具使用循环，包含三个工具类别：**发现**（MCP 调用 Snowflake 的 `INFORMATION_SCHEMA` 和 `ACCOUNT_USAGE`，以便代理可以检查已存在的内容）、**生成**（从 Jinja 模板写入文件）和**验证**（在隔离的沙箱模式下运行 `dbt build --target ci` 和 Great Expectations 检查点）。验证器的 stdout+stderr 反馈到代理的上下文中，它迭代直到绿色或达到重试上限。
## 包含内容
| 路径 | 内容 |
|---|---|
| `src/dagsmith/agent.py` | 工具使用循环。Sonnet 5，约 200 行。 |
| `src/dagsmith/mcp_server.py` | MCP 服务器，暴露 12 个只读 Snowflake 元数据工具。独立运行；可从 Claude Desktop、Cursor 或任何 MCP 客户端使用。 |
| `src/dagsmith/verifier.py` | 沙箱执行器 — 复制模板到临时 dbt 项目，运行 `dbt build`，流式传输结果。 |
| `src/dagsmith/templates/*.j2` | Jinja 模板，代理填充。故意保持最小，以便代理执行实际设计工作。 |
| `src/dagsmith/prompts/system.md` | 具有建模约定的系统提示（Kimball 用于市场，插入专用于事件，`stg_<source>__<table>` 命名）。 |
| `src/dagsmith/cli.py` | `dagsmith generate` 和 `dagsmith mcp`。 |
| `examples/babylist_registry_events/` | 完整的端到端示例 — 自然语言输入 + 代理生成的所有内容。 |
| `docs/architecture.md` | 详细组件设计 + 故障模式。 |
| `docs/mcp_tools.md` | 代理拥有的每个 MCP 工具的参考。 |
## 快速入门
```bash
git clone https://github.com/<you>/dagsmith && cd dagsmith
pip install -e ".[dev]"
# 1. 指向您的 Snowflake 沙箱（参见文档/setup.md）
export SNOWFLAKE_ACCOUNT=abc12345.us-east-1.aws
export SNOWFLAKE_USER=you@example.com
export SNOWFLAKE_ROLE=DATA_ENG
export SNOWFLAKE_WAREHOUSE=DEV_WH
export ANTHROPIC_API_KEY=sk-ant-...
# 2. 从规范生成管道
dagsmith generate examples/babylist_registry_events/input/spec.md \
--out examples/babylist_registry_events/generated \
--sandbox-schema DAGSMITH_CI_TAYLOR
# 3. 或为 Claude Desktop / Cursor / Zed 运行 MCP 服务器
dagsmith mcp
```
## MCP 服务器（可单独使用）
即使您不运行代理，MCP 服务器也很有用。 将任何 MCP 客户端指向它，Claude/Cursor/Zed 获得 12 个只读工具：
- `list_databases`、`list_schemas`、`list_tables`
- `describe_table`（列、类型、集群键、注释）
- `sample_rows`（限制上限预览）
- `query_history`（按成本/持续时间排名前 N 的查询）
- `access_history`（谁读取什么）
- `column_lineage`（通过 `ACCESS_HISTORY` 获取目标列的上游列）
- `dead_tables`（N 天内未读取的表）
- `storage_by_schema` 和 `credits_by_warehouse`
- `check_freshness`（每个表的最后一次成功写入）
适用于 Claude Desktop 的 MCP 配置：
```json
{
"mcpServers": {
"dagsmith": {
"command": "dagsmith",
"args": ["mcp"],
"env": {
"SNOWFLAKE_ACCOUNT": "abc12345.us-east-1.aws",
"SNOWFLAKE_USER": "you@example.com",
"SNOWFLAKE_ROLE": "READONLY_MC"
}
}
}
}
```
## 代理将执行和不执行的内容
**将执行：** 生成具有测试的分阶段 + 市场 dbt 模型、使用 Cosmos 的 Airflow DAG、具有合理默认值的 Great Expectations 检查点（粒度唯一性、行数范围、FK 非空）、新鲜度监视器查询和 docs.md 条目。
**不会执行：** 对生产环境运行破坏性 DDL、推测无法看到的业务逻辑或在验证通过之前生成代码。 每个生成的文件在代理声称完成之前都是 `dbt parse` 清洁的。
## 路线图
- [ ] `dagsmith audit` — 指向现有 dbt 项目并生成覆盖率缺口报告（缺少测试、过时模型、过度仓库查询）
- [ ] `dagsmith migrate` — 将旧 Fivetran → SQL 管道移植到代理生成的 dbt+Airflow 布局
- [ ] 评估框架 — 20 多个源规范的黄金输出测试，以便捕获回归
- [ ] 除了 Snowflake 之外的后端 — BigQuery + Redshift 适配器共享 ~80% 的模板
## 为什么 JD 的语言适合
该项目直接映射到 Babylist 高级数据工程师角色中的“利用生成、测试和维护它们的工具和智能脚手架”这一短语：
- **元层，而不仅仅是管道** — 代理是一个管道工厂，而不是一个管道。
- **MCP 服务器依赖于真实用户** — 元数据服务器可单独使用，而不仅仅作为代理输入。
- **具有工具集成的智能工作流** — 12+ 工具，真实验证反馈循环。
- **Airflow + dbt + Snowflake + Python** — JD 命名的整个堆栈，端到端连接。
- **跨多系统旅程的数据监控** — 新鲜度 + GE 套件是生成的包的一部分。
## 许可证
MIT。参见 [LICENSE](LICENSE)。
