hermes agent通过集成dbt实现数据质量全链路治理：可观测（docs）、可验证（schema/test）、可修复（snapshots）、可追溯（exposures）、可拦截（质量门禁）。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
如果您在使用Hermes Agent进行数据处理时发现下游分析结果异常、指标波动剧烈或模型训练效果下降，则很可能是由于原始数据存在重复、空值、类型错乱或业务逻辑不一致等质量问题。dbt（data build tool）作为现代数据栈中事实标准的数据转换与测试框架，可与Hermes Agent深度协同，实现数据质量的可观测、可验证、可修复。以下是将dbt集成至Hermes Agent工作流的具体实施步骤：
一、配置dbt项目与Hermes Agent环境联动
该步骤旨在建立Hermes Agent对dbt核心资产（models、tests、macros）的识别与调用能力，使Agent可在任务执行中动态触发dbt模型编译、测试运行与文档生成。需确保dbt CLI已安装且版本不低于1.8，并完成profile.yml配置以对接Hermes Agent所连接的目标数据库（如PostgreSQL、BigQuery或DuckDB）。
1、在Hermes Agent项目根目录下创建dbt_project/子目录，并将dbt项目完整复制至此路径。
2、编辑hermes_cli/config.py，添加dbt配置区块：
DBT_CONFIG = {
"project_dir": "dbt_project",
"profiles_dir": "dbt_project",
"target": "prod"
}
3、在tools/registry.py中注册dbt工具类：
from tools.dbt_tools import DbtRunner
register_tool("dbt-run", DbtRunner().run_model)
register_tool("dbt-test", DbtRunner().run_tests)
二、在Hermes Agent技能中嵌入dbt质量检查节点
此方法将dbt的schema测试（not_null、unique、relationships等）和自定义数据测试（via test/目录下的SQL文件）封装为可调度的技能节点，使Agent在ETL流程关键路径上自动执行质量门禁。所有测试失败将触发中断并返回具体错误字段与行样本，支持快速定位脏数据源头。
1、在skills/data_quality/目录下新建dbt_schema_guard.py，继承SkillBase类并重写execute方法。
2、在execute方法中调用DbtRunner().run_tests(select="test_type:schema")，捕获返回的TestResult对象。
3、若results.failures > 0，则构造结构化错误响应，包含failed_test_name、column_name及sample_rows（通过dbt ls --select加临时查询获取）。
4、将该技能注册至skills/skill_registry.yaml，设置触发条件为after: load_raw_conversations。
三、利用dbt docs生成可交互式数据字典并接入Hermes Agent UI层
该方式将dbt自动生成的数据文档（含模型血缘、列级描述、测试覆盖率、最近运行状态）嵌入Hermes Agent前端界面，供用户在执行分析任务前实时查阅字段语义与质量水位，避免因误解字段含义导致错误推理。文档服务以静态HTML形式托管，由Agent的Web工具模块提供内嵌访问入口。
1、在environments/hermes_swe_env/中新增docs_server.py，启动轻量HTTP服务托管dbt_project/target/docs/目录。
2、修改web_tools.py中的get_data_catalog函数，使其返回指向http://localhost:8081/#!/overview的iframe嵌入链接。
3、在前端ExecutorAvatar组件中，于“数据资产”侧边栏添加按钮，点击后加载该iframe，并设置sandbox="allow-scripts allow-same-origin"属性保障交互功能可用。
4、确保每次dbt docs generate执行后，target/docs/内容被同步至ECS实例对应路径，可通过hermes config watch dbt_docs命令监听dbt_project/models/变更并自动重建。
四、基于dbt exposures构建AI任务影响范围分析图谱
此方案利用dbt的exposures功能（定义在models/exposures.yml）显式声明Hermes Agent中各技能所依赖的数据资产及其业务用途，当某张源表结构变更或测试失败时，Agent可自动解析exposures依赖链，反向推导出受影响的技能ID、上次成功运行时间及关联告警接收人，实现数据问题的精准影响评估与闭环响应。
1、在dbt_project/models/exposures.yml中定义每个核心技能的数据契约：
exposures:
- name: skill_conversation_summarizer
type: analysis
owner:
email: ai-ops@hermes.local
depends_on:
- ref('stg_conversation_events')
- ref('dim_users')
2、在skills/mlops/dbt_exposure_analyzer.py中实现依赖图谱解析逻辑，调用dbt list --output json --select exposure:* --resource-type exposure提取结构化依赖关系。
3、当检测到stg_conversation_events模型测试失败时，调用该技能，输出受影响技能列表：skill_conversation_summarizer、skill_user_intent_classifier。
4、将输出结果注入CliAcpSessionPool的会话上下文，供后续自动降级策略或人工介入流程读取。
五、通过dbt snapshots实现关键业务实体变更历史追踪
该方法针对Hermes Agent中需强一致性保障的核心实体（如user_profiles、task_executions），启用dbt快照机制，在每次运行时捕获记录级变更（insert/update/delete），生成带valid_from与valid_to的时间切片表，供Agent在回溯分析、A/B实验归因或合规审计场景中精确查询任意历史时刻的状态快照。
1、在dbt_project/models/snapshots/下创建snapshot_user_profiles.sql，使用{% snapshot user_profiles_snapshot %}语法声明快照策略。
2、配置快照策略为timestamp模式，指定updated_at字段作为变更判断依据，并启用hard_delete标记以保留逻辑删除记录。
3、在skills/audit/tracking_snapshot.py中封装快照查询接口：
def get_state_as_of(entity: str, as_of: str) -> pd.DataFrame:
return run_dbt_sql(f"SELECT * FROM {entity}_snapshot WHERE '{as_of}' BETWEEN valid_from AND COALESCE(valid_to, '9999-12-31')")
4、当用户在前端输入自然语言查询“用户张三在2026年3月15日的权限等级”，Agent自动调用该接口，传入as_of="2026-03-15"，返回对应快照行。
