在 AGT 治理链路中接入 dbt 数据质量证据DataQualityEvidence 适配器实战指南【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit导读本指南围绕 Agent Governance ToolkitAGT中examples/data-quality-aware-governance/adapters/dbt/目录下的 dbt 数据质量证据适配器展开讲解如何把 dbt 的run_results.json产物转换为类型化的DataQualityEvidence证据对象并与 AGT 策略引擎的授权判定叠加形成授权 数据可信的双层治理决策。读完本文你将掌握 dbt 产物到治理证据的字段映射、状态判定与质量评分算法、证据对象在审计中的序列化方式以及如何以同样的模式扩展 Great Expectations、Soda 等其他数据质量工具。背景授权之外的第二层问题大多数 Agent 治理框架只回答一个问题这个 Agent 是否有权执行这个动作但在数据密集的工作流中还有一个同等重要的问题Agent 即将使用的数据此时此刻是否可信根示例 examples/data-quality-aware-governance/README.md 用一句话点明了痛点一个 Agent 可以完全被授权去查询某个数据集但如果该数据集未通过新鲜度freshness检查、校验测试validation tests、所有权要求或者统计行为相对基线发生了显著漂移仅有授权是不够的。为此根示例演示了一种双层治理检查模式Request │ ├─ Layer 1: AGT Policy Engine │ └─ Is this agent authorized for this action? │ ├─ NO → Block (policy violation) │ └─ YES → proceed to Layer 2 │ └─ Layer 2: Data Quality Registry └─ Is the target dataset trustworthy right now? ├─ Freshness Check ├─ Quality Check ├─ Drift Check ├─ Any failure → Block └─ All pass → Allow write to unified audit log请求只要在任一层失败就会被阻断。dbt 适配器正是这条模式在 dbt 技术栈上的具体落地实现。dbt 适配器概述dbt 适配器扩展了通用的 examples/data-quality-aware-governance/ 模式为它提供了一条以 dbt 为证据来源的具体实现路径。根示例保持工具无关tool-agnostic而本目录展示的是以 dbtrun_results.json产物作为输入的一种适配器实现。该适配器演示了四个能力读取 dbtrun_results.json产物将 dbt 测试结果映射为示例局部的DataQualityEvidence对象让该证据与 AGT 策略决策一起参与评估产出适合审计的 allow/block 决策。需要特别说明的是这是示例级代码它不会向 AGT 核心包添加模型、不要求 dbt 成为依赖、也不定义一套通用的数据质量 schema。目录结构examples/data-quality-aware-governance/adapters/dbt/ ├── README.md ├── data_quality_evidence.py ├── dbt_adapter.py └── fixtures/ ├── dbt_run_results_pass.json └── dbt_run_results_fail.jsonDataQualityEvidence类型化的数据集信任状态DataQualityEvidence定义在 data_quality_evidence.py是一个小型 frozen dataclassdataclass(frozenTrue)携带 Agent 尝试访问数据集那一刻的数据集信任状态。字段说明字段用途dataset_id逻辑数据集标识符freshness_at来自 dbtmetadata.generated_at的时间戳validation_statuspass、warn或failfailed_tests失败的 dbt 测试名列表quality_score通过测试占比取值0.0到1.0quality_profile_id可选的质量画像quality profile名称dataset_owner_did可选的数据集所有者标识符classification可选的数据分类source_tool质量工具来源此处为dbt源码中的内置校验从源码可以看到该数据类在__post_init__中做三重防御性校验data_quality_evidence.pydataset_id不能为空否则抛出ValueErrorvalidation_status必须落在VALIDATION_STATUSES {pass, warn, fail}集合内quality_score若提供则必须位于0.0 ~ 1.0闭区间。这保证了进入治理评估的证据对象从构造起就是自洽、可信的。内置的辅助方法除了字段DataQualityEvidence还提供三个在决策链路中高频使用的方法is_fresh_at(reference_time, max_age_hours6.0)以固定的参考时间为基准判断证据是否新鲜。实现为0 age_seconds max_age_hours * 3600即证据既不能太老超过最大时效也不能来自未来负年龄同样判为不新鲜is_passing属性validation_status passhas_failures属性failed_tests非空to_audit_dict()将证据序列化为字典供审计日志纳入。序列化后的键包括dataset_id、freshness_atISO 格式、validation_status、quality_score、quality_profile_id、dataset_owner_did、classification、failed_tests、source_tool。这个to_audit_dict是审计友好性的关键设计——治理决策之后证据的完整快照可以被写入统一审计日志让决策时看到的数据状态可复现、可追溯。dbt 产物到证据的字段映射适配器把 dbt 输出映射为证据对象核心映射关系如下dbt 产物字段证据字段metadata.generated_atfreshness_atresults[].statusvalidation_status失败的results[].unique_id值failed_testsresults[]中的通过占比quality_score源码级实现细节映射逻辑实现在 dbt_adapter.py 的load_from_dbt_run_results()函数中。它接收四个参数run_results_pathdbt 产物路径、dataset_id逻辑数据集标识符必填以及可选的dataset_owner_did、classification、quality_profile_id。时间戳解析metadata.generated_at缺失时直接抛出ValueError。解析函数_parse_timestamp会先rstrip(Z)去掉可能存在的 UTC 后缀再用datetime.fromisoformat解析dbt_adapter.py。注释中明确说明示例故意将 UTC 风格时间戳规范化为 naive datetime以便与固定的参考时间比较——这正是is_fresh_at要求传入参考时间的原因。状态判定与失败测试收集遍历results[]时status取fail或error的测试被记入failed_testswarn的记入warning_testsdbt_adapter.py。随后按优先级判定只要有失败测试 →validation_status fail否则只要有警告 →warn否则 →pass。质量评分算法quality_score passing_tests / total_tests四舍五入保留 4 位小数当测试总数为 0 时按1.0处理dbt_adapter.py。其中passing_tests total_tests - len(failed_tests) - len(warning_tests)——注意警告测试不计入通过但也不导致整体判 fail。测试名提取dbt 的unique_id形如test.analytics.not_null_user_events_user_id.a1b2c3_test_name_from_unique_id按.分割后取倒数第二个片段作为可读测试名dbt_adapter.py即这里得到not_null_user_events_user_id。直接可用的示例产物与运行方式仓库提供了两套真实形状的 dbtrun_results.json示例产物基于 dbt 1.11.7、run-results v5 schema通过场景dbt_run_results_pass.jsonfixtures/dbt_run_results_pass.json 中metadata.generated_at为2026-05-14T08:00:003 条测试not_null_user_events_user_id、accepted_values_user_events_event_type、not_null_user_events_session_id全部status: pass。解析结果将是validation_statuspass、quality_score1.0、failed_tests()。失败场景dbt_run_results_fail.jsonfixtures/dbt_run_results_fail.json 中前两条测试failfailures分别为 142 与 37消息为Got 142 results, expected 0第三条pass。解析结果将是validation_statusfail、failed_tests(not_null_user_events_user_id, accepted_values_user_events_event_type)、quality_score0.3333。用法示例from dbt_adapter import load_from_dbt_run_results evidence load_from_dbt_run_results( fixtures/dbt_run_results_pass.json, dataset_iduser_events, dataset_owner_diddid:web:analytics.example.com, classificationregulated, quality_profile_idstrict, )注意运行前提两个 Python 模块同级引用dbt_adapter.py中from data_quality_evidence import DataQualityEvidence因此需在adapters/dbt/目录下执行或将目录加入sys.path。与 AGT 策略评估的组合决策生成的证据随后可以参与 AGT 策略评估形成如下两类决策AGT policy allows the agent dbt evidence says the dataset is fresh and passing allow the action或AGT policy allows the agent dbt evidence says validation failed block the action at the data quality layer这一组合逻辑在根示例 example.py 的governed_query()中有完整实现先调用policy_engine.check_violation(agent_role, tool_name, args)做 Layer 1 授权判定通过后再从DataQualityRegistry取出数据集快照依次检查新鲜度、质量阈值、漂移阈值与失败测试任一层失败即返回blocked并记录审计条目全部通过才返回allowed (unified)。策略引擎的底层支撑Layer 1 的授权语义来自 policy_engine.py 的PolicyEngine类add_constraint(role, allowed_tools)采用**允许列表allow-list**方式定义Agent 的物理法则即Scale by Subtraction——只声明什么被允许其余一律隐式阻断policy_engine.pyfreeze()将内部可变字典替换为MappingProxyType只读代理任何add_constraint、set_agent_context等变更操作都会抛出RuntimeError专门防住Agent 在运行时自我削弱策略的自修改攻击向量policy_engine.pycheck_violation()做三级检查基于角色的允许列表检查 → 基于条件的 ABAC 检查 → 基于参数的安全检查路径穿越、危险命令模式等返回None表示无违规policy_engine.py。在根示例的 example.py 中analyst-agent-01被授权database_query与report_generatereport-agent-02仅被授权report_generate随后立即freeze()锁定策略。质量阈值的策略化配置Layer 2 的阈值并非硬编码而是来自 policy.yamldata_quality: global_defaults: freshness_threshold_hours: 6.0 quality_score_threshold: 0.85 drift_threshold: 0.25 block_on_failed_tests: true validation_status_required: pass # pass | warn | fail datasets: user_events: owner_did: did:web:analytics.example.com freshness_threshold_hours: 6.0 quality_score_threshold: 0.85 drift_threshold: 0.25 description: User interaction events — strict freshness required revenue_metrics: owner_did: did:web:finance.example.com freshness_threshold_hours: 12.0 quality_score_threshold: 0.90 drift_threshold: 0.25 description: Revenue rollups — high quality threshold for financial reporting每个数据集可覆盖全局默认值如revenue_metrics把新鲜度放宽到 12 小时、质量阈值提高到 0.90未配置的数据集自动继承global_defaults——这与example.py中dataset_policy()的合并逻辑一致。audit段则开启全量决策日志与数据状态快照记录。非目标Non-Goals该适配器刻意不做以下事情README.md不改变 AGT 核心行为不新增包级 schema不要求 dbt 成为 AGT 的依赖不把 dbt 规定为唯一的数据质量系统不定义通用的质量阈值不实现外部回执receipt或协议语义。这意味着它是可插拔的旁路组件生产环境中完全可以用真实 dbt 产物、Great Expectations 结果或 Atlan、DataHub 等元数据目录替换示例中的模拟注册表README.md。与根示例的关系及扩展方向根示例 examples/data-quality-aware-governance/ 保持通用、工具无关本适配器是它的一条具体实现路径。后续适配器可以遵循同样的结构扩展例如adapters/great-expectations/ adapters/soda/ adapters/custom-catalog/每条路径只需完成同一件事把各自工具的产物映射为DataQualityEvidence这一通用证据类型即可无缝接入上文的双层治理链路与统一审计日志。这也是该设计最值得借鉴之处——治理侧只认类型化证据工具侧只需实现映射两者通过DataQualityEvidence解耦。运行环境说明根示例的运行前提README.md# From repo root cd agent-governance-python/agent-os pip install -e . cd examples/data-quality-aware-governance python example.pyexample.py会把agent-governance-python/agent-os/modules/control-plane/src注入sys.path从而直接使用agent_control_plane包中的PolicyEngine、KernelSpace、SyscallRequest等真实 AGT 对象example.py。dbt 适配器本身则无需安装 dbt 或任何额外依赖仅依赖 Python 标准库json、datetime、pathlib与同目录下的data_quality_evidence.py。小结dbt 数据质量证据适配器展示了 AGT 治理体系的一个关键扩展思路授权Layer 1与数据可信Layer 2双门控。DataQualityEvidence提供了稳定、可校验、可审计的类型化证据契约load_from_dbt_run_results完成了 dbt 产物到该契约的廉价转换而根示例的governed_query与policy.yaml则证明了它如何端到端地影响最终决策。对于任何以 dbt 为数据管道核心、又希望 Agent 只访问此刻可信数据的团队这条适配器路径都可以直接照搬或按同样结构定制。【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考