架构

智能体编排的数据管道:构建 Dagster + dbt + MCP 技术栈

作者：IdeaBosque最后更新：2026年8月28日

关键要点

Fivetran 与 dbt Labs 于 2026 年 6 月 1 日完成合并(合并后 ARR 约 6 亿美元,服务超 10 万个数据团队),并推出了 Agents Schema——一种开放标准,将数据仓库模式转化为面向智能体的受治理共享上下文层。
Databricks 报告称其 Neon 单元上超过 80% 的数据库是由 AI 智能体而非人类创建的——数据栈的主要使用者已经从分析师转向了智能体。
三种智能体模式定义了新的数据管道——智能体编写并搭建管道代码框架、管道通过提出并应用修复实现自我修复、智能体在聊天中对运行故障进行分诊——这三者都需要机器可读的血缘信息,而不是渲染好的仪表盘。
技术栈由三个受治理的层组成——Dagster 负责资产血缘,dbt 负责经过测试的转换与共享上下文,一个 MCP 模块负责有策略范围限制的访问——从而让智能体在与人类相同的管控下读取管道的结构。

数据技术栈原本是为人类分析师设计的:夜间运行管道,早上查看仪表盘,发现数字异常就提交工单。而 AI 智能体消费数据的方式不同。正如合并后的 Fivetran + dbt Labs 所说,智能体"持续地、并行地、以机器速度运行"——它们需要的是管道的结构(血缘、测试、定义),而不仅仅是输出结果。在短短一个季度内,数据移动和转换领域最大的两家厂商就围绕这一事实完成了重构:Fivetran 与 dbt 的合并推出了 Agents Schema,并将 dbt Fusion 引擎作为 dbt Core v2.0 开源;Databricks 收购了 Electric,让每个智能体都拥有自己的一次性 Postgres 实例。

本指南以 B2B 采购场景搭建这一模式:一条摄取供应商目录、价格和库存数据、并将其暴露给 RFQ 智能体的管道。文中涵盖三个层——Dagster 负责以资产为中心的编排,dbt 负责受治理的转换,一个 MCP 模块负责范围受限的智能体访问——以及让管道能够自我维护的三种智能体模式。读完之后,你将了解每一层各自贡献了什么、为什么资产血缘是这一架构中承重的关键选择,以及人类应保留在流程的哪个环节。

为何以资产为中心的编排是基础

大多数面向智能体的编排失败都源于错误的心智模型。以任务为中心的调度器(经典的 cron 加 DAG 设计)回答的是"这个任务运行了吗?"而当智能体问"为什么 Acme 的价格是过时的?"时,它需要的是不同的答案:"哪个数据资产已过期,它依赖什么,又是什么在为它提供数据?"这是一个资产问题,这也是为什么 Dagster 以资产为中心的模型在这里是基础,而不只是一种偏好。

在 Dagster 中,你声明的是你产出的事物——supplier_catalog、normalized_prices、availability_snapshot——以及它们之间的依赖关系。编排器由此掌握完整的血缘图。Dagster 的 Declarative Automation 允许资产在其上游发生变化时刷新,而不是按固定时钟刷新,于是"过时"就成了系统可以推理的一个属性。这张血缘图正是智能体在无需猜测的情况下,将错误数字追溯到源头所需要的。

import dagster as dg

@dg.asset(group_name="procurement")
def supplier_catalog(context: dg.AssetExecutionContext) -> dg.MaterializeResult:
    rows = fetch_supplier_feed()  # NetSuite, EDI, CSV drop, etc.
    write_bronze("supplier_catalog", rows)
    return dg.MaterializeResult(metadata={"row_count": len(rows)})

@dg.asset(deps=[supplier_catalog], group_name="procurement",
          automation_condition=dg.AutomationCondition.eager())
def normalized_prices() -> None:
    # dbt owns the transformation logic; Dagster owns the lineage + trigger
    run_dbt(select="normalized_prices")

deps 和 automation_condition 正是关键所在:智能体(以及下文的自我修复循环)可以将这张图作为数据来读取。Airflow 从另一个方向得出了同样的结论——Airflow 3.2 新增了 Common AI Provider 和资产感知调度——而行业整合也在真实发生:Prefect 于 2026 年 7 月收购了 Dagster。无论你最终标准化到哪个编排器,要求都是一样的:拥有声明式血缘的资产,而不是不透明的任务。

为何 dbt 掌管转换与受治理的上下文

Dagster 触发工作并追踪血缘,但不应包含你的业务逻辑。那属于 dbt 的范畴——在这里,每一次转换都是一个受版本控制、附带测试、文档和语义定义的 SQL 模型。对智能体而言,这不是锦上添花,而是信任边界本身。dbt 自身的立场是,转换层正是使智能体化管道值得信任的关键:一个针对未定义、未测试表编写 SQL 的智能体,只会更快地把混乱自动化。

合并之后最重要的新增功能是 Agents Schema:一个专门指定的数据仓库模式,以普通 SQL 表的形式存储指标定义、语义模型、dbt 血缘和业务文档。这样一来,不必让每个智能体各自重新推导"现有可用库存"这样的定义究竟意味着什么,该定义只存在于一个受治理、由客户拥有的地方,供智能体读取。这是受治理连接器模块在数据侧的对应物——一个有策略范围限制的共享上下文来源,而不是会随时间产生偏差的、按智能体各自维护的副本。

-- models/marts/availability_snapshot.sql
select
    sku,
    warehouse_id,
    on_hand - allocated as available_qty,   -- the governed definition
    updated_at
from {{ ref('normalized_inventory') }}

-- schema.yml: the test that gates the agent's trust
-- - name: available_qty
--   tests: [not_null, {dbt_utils.accepted_range: {min_value: 0}}]

一次测试失败就是在告诉智能体:不应引用这一行数据。正是这一个机器可读的通过/失败信号(附加在每个模型上),使接下来的两种模式得以在无需人类逐步盯守的情况下运行。

智能体在何处接入:一个 MCP 模块,而不是数据库登录

智能体永远不应持有原始的数据仓库凭据。它应当调用一个受治理的 MCP 模块,该模块暴露一小组带类型的工具——get_availability(sku, warehouse)、get_tier_price(sku, customer_tier)、list_substitutes(sku)——每个工具都映射到一个经过测试的 dbt 模型,并各自带有策略范围、速率限制和审计日志。这与用于 ERP 和商务连接器的模块模式相同,只是应用到了管道自身的输出上。它把爆炸半径控制得很小:智能体可以读取 availability_snapshot,但不能运行任意 SQL,而且每一次调用都会被记录。

这个边界也是每个智能体自身状态的归属之处。Databricks 收购 Electric——在智能体沙箱内运行基于 WASM 的 Postgres(PGlite),并与中心受治理状态同步——之所以存在,是因为智能体"需要成千上万个微小的、一次性的数据库"来承载工作上下文,并将其与持久的受治理表分离开来。经验法则是:持久、共享、受治理的数据位于 MCP 模块之后;快速变化、按次运行的临时上下文则位于智能体自己的沙箱中。

该技术栈支撑的三种智能体模式

有了血缘(Dagster)、经过测试的定义与共享上下文(dbt + Agents Schema),以及范围受限的访问(MCP),三种模式就变得可行:

智能体化开发。 智能体针对现有的血缘图,搭建新的资产和转换框架——起草 dbt 模型、提出模式测试、接入 Dagster 资产。Dagster 为 Claude Code 和 Codex 提供了 dagster-io/skills,以及一个名为 Compass 的 Slack 助手;Bruin 则为同一目的暴露了一个 MCP 服务器。人类审查的是一份 pull request,而不是一个空白文件。
自我修复的管道。 当某个模式测试失败,或某个上游资产出现故障时,智能体会读取血缘、定位出问题的模型、提出修复方案,并在金丝雀发布中应用该方案,或者提交一个 PR。由于 Dagster 掌握依赖图、dbt 知道具体哪个测试失败了,这样的修复是基于血缘信息做出的,而不是盲目重试。
智能体化故障排查。 出现故障时,智能体读取运行日志和元数据,并在 Slack 或 Teams 中回复可能的原因和一个建议的修复方案——这正是 Dagster Compass 和 Snowflake Cortex 所围绕构建的模式。值班排障的工作由查看仪表盘转变为审阅智能体给出的诊断。

以上这些都没有把人类从流程中排除。每一次实际应用的变更都要经过测试关卡、金丝雀发布或人工审查——这正是 dbt Summit 2026 主题演讲所强调的,让智能体触碰生产数据必须付出的代价。

一图看懂整个技术栈

一个代表性的构建案例

一家运营 NetSuite、拥有两个仓库和三个供应商目录的分销商,希望有一个 RFQ 智能体能够在无需人工手动查询库存的情况下完成报价。真正的瓶颈是管道,而不是模型。我们在 Dagster 中以显式血缘声明了目录、价格和库存资产;将定价和可用性逻辑迁移到经过测试的 dbt 模型中,并用一个 Agents Schema 固定了"可承诺库存"的定义;再通过一个只读范围的 MCP 模块暴露了三个带类型的工具。如今,自我修复循环会在早间报价运行之前捕获损坏的供应商数据源,并提交带有修复方案的 PR;由于智能体给出的第一反应是诊断而不是告警,管道故障导致的值班时间也随之下降。智能体要么依据经过测试的数据进行报价,要么拒绝报价——它绝不会依据一行未通过测试的数据进行报价。