架構

智能體協同的資料管道:建構 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;由於智能體給出的第一反應是診斷而非告警,管道故障導致的值班時間也隨之下降。智能體要麼依據經過測試的資料進行報價,要麼拒絕報價——它絕不會依據一行未通過測試的資料進行報價。