统一语义层：如何让AI Agent与BI报表共享同一份业务口径

如果你把一个“上月销售额是多少”的问题，同时抛给业务分析师和某个大模型Agent，大概率会拿到两个不一样的数据。问题不只在模型能力，更在于企业根本没有把“销售额”这个口径沉淀成可编程、可复用、可审计的资产。业务人员看的报表是一套SQL，AI问答接口接的是另一套表和文档，两边各说各话。

真正值得做的，不是继续往Agent里堆提示词，而是回到数据建模层：一次建模，同时服务于人类和AI。这个思路用英文讲就是 “Model your business once – for humans and AI alike”。本文会用一个可运行的电商零售最小项目，讲清楚什么是统一语义层，为什么它比“把表结构丢给LLM”更可靠，以及怎样用一套YAML业务模型同时支撑BI报表、API查询和大模型Agent调用。

读完这篇文章，你可以理解并落地一套最小可行的语义层架构：业务口径只定义一次，人类通过SQL和BI工具消费，AI通过语义API消费，两边拿到的是同一份业务事实。

1. 这篇文章真正要解决的问题

目前企业里做数据分析，普遍存在三种割裂：

第一种是口径割裂。业务周报里的“销售额”可能是支付成功订单的实付金额，财务系统的“销售额”可能是未扣除退款的下单金额，数据中台里的“销售额”又可能是订单完成后的金额。同一个指标，三个系统三个数。

第二种是接口割裂。BI报表通过写死的SQL取数，API服务通过独立实现的数据服务取数，AI Agent则直接把数据库表结构塞给大模型，让它“自己看着办”。结果是每次新场景都要重新解释口径，AI生成的SQL一段时间后就没人敢信。

第三种是信任割裂。当AI说“根据数据，本月客单价环比下降了5%”时，业务方第一反应不是看结论，而是问“你这个数是从哪来的？口径是什么？有没有权限看这些数据？”

这些问题都指向同一个根因：企业缺少一个单一的业务语义层。所谓语义层，就是把自己定义好的维度、度量、指标、口径、关系和权限，集中管理在一个逻辑层中，在底层数据和上层消费者之间建立统一的翻译器。

这篇文章要解决的问题，不是教你把大模型接入数据库，而是教你如何构建这一层“翻译器”，让人类和AI共同消费同一份业务模型。适合的读者包括数据平台工程师、后端开发、AI应用开发者，以及正在做AI Agent落地但被“取数不可信”卡住的人。

2. 基础概念与核心原理

在进入代码之前，先把几个概念统一一下。

业务模型：描述业务世界的结构化定义，包括有哪些实体、哪些关系、哪些可分析维度、哪些可度量指标。

维度：观察业务的角度，比如时间、渠道、商品类目、地区。维度通常用于分组和筛选。

度量：底层数据表中的原始数值字段，比如订单金额、退款金额、用户ID。度量本身不具备业务含义，只有经过聚合规则定义后才变成指标。

指标：有明确业务口径的度量计算，比如“销售额=支付成功订单的实付金额-退款金额”“订单数=支付成功订单的去重订单数”。指标是业务模型的灵魂。

口径：一条指标到底怎么算。它决定了数字可信还是不可信。

语义层：建立在数据仓库或数据库之上的一层逻辑模型。它把物理表的字段、表关系和底层SQL细节封装起来，对上层暴露的是“客单价”“销售额”“新增用户”这类业务语义。

对比一下当前主流做法和语义层做法：

这里的核心原理可以概括为一句话：让AI不要直接猜业务口径，而是调用业务口径。

传统方案里，大模型像是一个新入职但不太靠谱的分析师，你给它一堆表名和字段名，它自己翻表、猜字段含义、拼SQL。统一语义层方案里，大模型更像是调用企业内部指标平台的“消费者”，它只负责理解用户问题、选择合适的工具，最终计算一律走语义层API。

这个思路的价值在于，业务口径成为可编程资产。它不依赖某个人的记忆，不依赖文档更新，也不依赖大模型当时的“状态”。一次建模，双端消费，本质上是把“知识”和“计算”解耦。

3. 环境准备与前置条件

本文采用Python技术栈实现一个最小可运行的语义层项目。运行环境如下：

• 操作系统：Windows / macOS / Linux均可
• Python版本：3.9及以上
• 数据库：SQLite（零部署，适合演示）
• Web框架：FastAPI
• 依赖库：fastapi、uvicorn、pyyaml、requests

请先创建项目目录，后面所有文件都放在这个目录下。

mkdir demo-semantic-layer cd demo-semantic-layer

然后创建一个虚拟环境并安装依赖：

python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate pip install fastapi uvicorn pyyaml requests

项目结构如下：

demo-semantic-layer/ ├── init_db.py ├── semantic_model.yaml ├── app.py └── ai_agent_demo.py

需要说明的是，这里的实现是为了讲清原理，生产环境还要考虑参数化查询、连接池、缓存、鉴权和审计，后面会单独讲。

4. 定义一次业务模型——YAML语义模型示例

现在用一个电商零售场景来演示。先明确业务口径：

• 销售额：支付成功订单的实付金额，扣除退款金额。
• 订单数：支付成功订单的去重订单ID数量。
• 客单价：销售额除以订单数。
• 主维度：订单支付日期、渠道、商品类目。

这些口径作为唯一的业务事实来源，写入semantic_model.yaml。

# 文件路径：semantic_model.yaml model: name: retail description: 零售业务统一语义模型，BI和AI共用 tables: - name: orders alias: o dimensions: pay_date: column: pay_time type: datetime granularities: [day, month] description: 订单支付时间 channel: column: channel type: string description: 推广渠道 category: column: product_category type: string description: 商品类目 measures: order_amount: aggregation: sum sql: "CASE WHEN o.order_status = 'paid' THEN o.pay_amount - COALESCE(o.refund_amount, 0) ELSE 0 END" description: 销售额，支付成功订单实付金额扣除退款 format: currency order_count: aggregation: count_distinct sql: "CASE WHEN o.order_status = 'paid' THEN o.order_id END" description: 订单数，支付成功订单去重计数 metrics: avg_order_value: formula: "{order_amount} * 1.0 / NULLIF({order_count}, 0)" description: 客单价，销售额除以订单数

这个YAML文件最关键的地方是：指标口径只定义一次，而且以机器可读的方式存在。

先看维度：pay_date、channel、category分别映射到底层表字段。其中pay_date属于时间维度，后续可以按天或按月聚合。

再看度量：order_amount是销售额，SQL表达式里加了一个CASE WHEN过滤，只统计支付成功的订单，并且扣除了退款。order_count同样只统计支付成功的订单ID，避免把已退款订单也算进去。

最后看派生指标：avg_order_value没有直接写SQL，而是通过公式引用上面的两个度量。这样做的好处是，将来如果修改了“销售额”的口径，客单价会自动响应，不需要再改一处。

这里需要强调一点：这个YAML文件本质上是企业业务口径的“源码”。它应该走代码审查、版本管理，而不是躺在某位业务同学或开发同学的脑海和聊天记录里。

5. 让AI消费统一模型——实现语义查询API

人类可以使用SQL和BI工具消费语义模型，AI则需要一个结构化的工具接口。

在实践中，最常见的做法是暴露一个语义查询API。AI Agent通过这个API传入指标、维度、过滤条件，API内部根据YAML模型编译成SQL，查数后返回结果。

下面是用 FastAPI 实现的最小版本。

# 文件路径：app.py import sqlite3 import re from typing import Dict, List, Optional import yaml from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI(title="Semantic Layer API", version="1.0.0") MODEL_PATH = "semantic_model.yaml" with open(MODEL_PATH, "r", encoding="utf-8") as f: SEMANTIC_MODEL = yaml.safe_load(f)["model"] class SemanticRequest(BaseModel): metrics: List[str] dimensions: Optional[List[str]] = [] filters: Optional[Dict[str, dict]] = {} limit: Optional[int] = 100 def resolve_formula(formula: str) -> str: """把指标公式中的 {measure_name} 替换为底层聚合表达式。""" model = SEMANTIC_MODEL alias = model["tables"][0]["alias"] def replace_measure(match): name = match.group(1) measure = model["measures"].get(name) if measure is None: raise HTTPException(status_code=400, detail=f"公式引用了未知度量: {name}") agg = measure["aggregation"] if agg == "sum": return f"SUM({measure['sql']})" if agg == "count_distinct": return f"COUNT(DISTINCT {measure['sql']})" return f"AVG({measure['sql']})" return re.sub(r"\{(\w+)\}", replace_measure, formula) def compile_sql(req: SemanticRequest) -> str: """根据语义模型和请求参数编译SQL。""" model = SEMANTIC_MODEL table = model["tables"][0] alias = table["alias"] select_parts = [] group_by_parts = [] for dim in req.dimensions: dim_def = model["dimensions"].get(dim) if dim_def is None: raise HTTPException(status_code=400, detail=f"未知维度: {dim}") column = f"{alias}.{dim_def['column']}" if dim_def["type"] == "datetime": expr = f"date({column})" select_parts.append(f"{expr} AS {dim}") group_by_parts.append(expr) else: select_parts.append(f"{column} AS {dim}") group_by_parts.append(column) for metric in req.metrics: if metric in model["metrics"]: formula_expr = resolve_formula(model["metrics"][metric]["formula"]) select_parts.append(f"{formula_expr} AS {metric}") elif metric in model["measures"]: measure = model["measures"][metric] agg = measure["aggregation"] if agg == "sum": expr = f"SUM({measure['sql']})" elif agg == "count_distinct": expr = f"COUNT(DISTINCT {measure['sql']})" else: expr = f"AVG({measure['sql']})" select_parts.append(f"{expr} AS {metric}") else: raise HTTPException(status_code=400, detail=f"未知指标: {metric}") sql = f"SELECT {', '.join(select_parts)} FROM {table['name']} {alias}" where_parts = [] for dim, cond in req.filters.items(): dim_def = model["dimensions"].get(dim) if dim_def is None: raise HTTPException(status_code=400, detail=f"未知过滤维度: {dim}") column = f"{alias}.{dim_def['column']}" if "eq" in cond: where_parts.append(f"{column} = '{cond['eq']}'") if "gte" in cond: where_parts.append(f"{column} >= '{cond['gte']}'") if "lte" in cond: where_parts.append(f"{column} <= '{cond['lte']}'") if where_parts: sql += " WHERE " + " AND ".join(where_parts) if group_by_parts: sql += " GROUP BY " + ", ".join(group_by_parts) if req.limit: sql += f" LIMIT {req.limit}" return sql @app.get("/model") def get_model(): """返回语义模型元数据，人类和AI都可以通过这个接口理解业务口径。""" return SEMANTIC_MODEL @app.post("/query") def query(req: SemanticRequest): """根据语义模型执行查询。""" try: sql = compile_sql(req) except HTTPException: raise except Exception as e: raise HTTPException(status_code=500, detail=f"SQL编译失败: {e}") conn = sqlite3.connect("retail.db") conn.row_factory = sqlite3.Row try: rows = conn.execute(sql).fetchall() return {"sql": sql, "rows": [dict(row) for row in rows]} except Exception as e: raise HTTPException(status_code=500, detail=f"查询失败: {e}") finally: conn.close()

这个API有两个关键端点。

GET /model返回整个语义模型，里面包含维度、度量、指标和描述。AI Agent在不知道有哪些指标可用时，可以先调用这个接口获取“能力清单”。

POST /query接收请求体，要求调用方指定metrics，可选指定dimensions和filters。服务端根据YAML模型编译SQL，执行查询后返回SQL和结果。

需要再次提醒：为了让示例简洁，上面的代码把eq、gte、lte的值直接拼进了SQL。生产环境必须改成参数化查询，并且对表达式做白名单校验，否则会引入SQL注入风险。

6. 人类侧消费——用同一模型生成SQL与BI报表

“人类消费”不等于不用技术。如果业务分析师要核对数据，或者要在BI工具里做报表，理想路径是：从语义模型导出的SQL，和AI查到的SQL完全一致。

现在手动写一条符合业务口径的查询SQL，演示人类侧消费。

SELECT date(o.pay_time) AS pay_date, SUM( CASE WHEN o.order_status = 'paid' THEN o.pay_amount - COALESCE(o.refund_amount, 0) ELSE 0 END ) AS order_amount, COUNT( DISTINCT CASE WHEN o.order_status = 'paid' THEN o.order_id END ) AS order_count, SUM( CASE WHEN o.order_status = 'paid' THEN o.pay_amount - COALESCE(o.refund_amount, 0) ELSE 0 END ) * 1.0 / NULLIF( COUNT( DISTINCT CASE WHEN o.order_status = 'paid' THEN o.order_id END ), 0 ) AS avg_order_value FROM orders o WHERE date(o.pay_time) >= '2024-11-01' AND date(o.pay_time) <= '2024-11-04' GROUP BY date(o.pay_time) ORDER BY pay_date;

这条SQL可以放到Superset、Metabase、帆软等BI工具里，也可以直接在数据库客户端里人工核验。

细看你会发现，这段SQL里的CASE WHEN逻辑和前面semantic_model.yaml中定义的销售额表达式完全一致。这不是偶然，而是语义层要达到的目标：人类用的SQL和AI经过API得到的SQL，都来自同一个业务模型。

在实际项目里，更成熟的方案是让语义层直接生成并下发SQL给BI引擎，而不是让人手工维护一份。这样即使将来销售额增加了新的抵扣规则，也只需要修改YAML模型一处，所有下游自动更新。

7. Agent接入示例——大模型函数调用

接下来看AI侧怎么接入。

现在主流的大模型应用是通过Function Calling或Tool Calling让Agent调用外部工具。核心在两步：

1. 把语义查询API描述成大模型可以理解的工具。
2. 当用户问题涉及业务指标时，让模型决定调用这个工具，并把参数填好。

先看工具描述，通常是JSON Schema：

{ "name": "query_semantic_layer", "description": "查询企业统一业务语义模型。销售额、订单数、客单价等指标必须通过该接口获取，禁止自行猜测口径。", "parameters": { "type": "object", "properties": { "metrics": { "type": "array", "items": {"type": "string"}, "description": "指标列表，可选 order_amount, order_count, avg_order_value" }, "dimensions": { "type": "array", "items": {"type": "string"}, "description": "维度列表，可选 pay_date, channel, category" }, "filters": { "type": "object", "description": "过滤条件，例如 {\"pay_date\": {\"gte\": \"2024-11-01\", \"lte\": \"2024-11-30\"}}" } }, "required": ["metrics"] } }

配合这条Prompt：

你是公司数据分析助手。回答业务指标问题时，必须调用 query_semantic_layer 工具。 不允许自行拼接SQL，不允许猜测指标含义。如果用户问的指标不在工具支持列表中， 请明确告知无法回答，并建议用户在指标平台上申请新指标。

下面写一个模拟Agent调用过程的Python脚本。这个脚本里不实际调用大模型，而是演示“Agent做出了工具调用”后，请求语义API的过程。

# 文件路径：ai_agent_demo.py import json import requests def ask_assistant(user_question: str): # 真实项目中，这里会先经过LLM，由LLM判断并输出工具调用参数。 # 这里模拟LLM决定调用语义层接口的结果。 print(f"用户问题: {user_question}") print("Agent 决策: 调用 query_semantic_layer 工具") tool_call_arguments = { "metrics": ["order_amount", "order_count", "avg_order_value"], "dimensions": ["pay_date"], "filters": { "pay_date": { "gte": "2024-11-01", "lte": "2024-11-04" } } } resp = requests.post( "http://127.0.0.1:8000/query", json=tool_call_arguments, timeout=10 ) resp.raise_for_status() return json.dumps(resp.json(), ensure_ascii=False, indent=2) if __name__ == "__main__": print(ask_assistant("11月1日到4日每天的销售额、订单数和客单价是多少？"))

Agent与传统程序的差别在于，用户问题不是固定参数，而是自然语言。模型要先理解用户意图，再把问题转成一个工具调用。这个过程中，模型的作用是“编排”，不是“计算”。真正的数据计算必须回到语义层。

这样做有一个直接好处：AI永远无法“自由发挥”指标口径。它不能自己决定“销售额”是SUM(pay_amount)还是COUNT(order_id)，因为可选指标是固定的，计算逻辑由服务端决定。

8. 运行结果与效果验证

先把示例数据初始化到SQLite。这里直接创建一个表和少量测试订单。

# 文件路径：init_db.py import sqlite3 conn = sqlite3.connect("retail.db") cur = conn.cursor() cur.execute("DROP TABLE IF EXISTS orders") cur.execute(""" CREATE TABLE orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id TEXT NOT NULL, pay_time TEXT NOT NULL, pay_amount REAL NOT NULL, refund_amount REAL DEFAULT 0, order_status TEXT NOT NULL, channel TEXT NOT NULL, product_category TEXT NOT NULL, user_id TEXT NOT NULL, register_time TEXT NOT NULL ) """) orders = [ ("A001", "2024-11-01 10:23:00", 299.00, 0, "paid", "organic", "手机数码", "u1001", "2024-10-20 09:00:00"), ("A002", "2024-11-01 15:01:00", 89.00, 0, "paid", "paid_social", "家居生活", "u1002", "2024-10-21 10:00:00"), ("A003", "2024-11-02 09:11:00", 599.00, 50.00, "paid", "email", "家电", "u1003", "2024-10-22 11:00:00"), ("A004", "2024-11-02 20:44:00", 129.00, 0, "paid", "organic", "服饰鞋包", "u1004", "2024-10-23 12:00:00"), ("A005", "2024-11-03 12:02:00", 399.00, 0, "paid", "paid_social", "手机数码", "u1005", "2024-10-25 13:00:00"), ("A006", "2024-11-03 18:32:00", 159.00, 0, "paid", "email", "家居生活", "u1006", "2024-10-28 14:00:00"), ("A007", "2024-11-04 11:21:00", 219.00, 219.00, "refunded", "organic", "服饰鞋包", "u1007", "2024-11-01 15:00:00"), ("A008", "2024-11-04 16:45:00", 888.00, 0, "paid", "paid_social", "家电", "u1008", "2024-11-02 16:00:00"), ] cur.executemany(""" INSERT INTO orders (order_id, pay_time, pay_amount, refund_amount, order_status, channel, product_category, user_id, register_time) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, orders) conn.commit() conn.close() print("初始化完成，已创建 retail.db")

按以下顺序验证：

python init_db.py uvicorn app:app --reload --port 8000

另开一个终端执行：

python ai_agent_demo.py

预期返回类似下面的结果：

{ "sql": "SELECT date(o.pay_time) AS pay_date, SUM(CASE WHEN o.order_status = 'paid' THEN o.pay_amount - COALESCE(o.refund_amount, 0) ELSE 0 END) AS order_amount, COUNT(DISTINCT CASE WHEN o.order_status = 'paid' THEN o.order_id END) AS order_count, SUM(CASE WHEN o.order_status = 'paid' THEN o.pay_amount - COALESCE(o.refund_amount, 0) ELSE 0 END) * 1.0 / NULLIF(COUNT(DISTINCT CASE WHEN o.order_status = 'paid' THEN o.order_id END), 0) AS avg_order_value FROM orders o WHERE o.pay_time >= '2024-11-01' AND o.pay_time <= '2024-11-04' GROUP BY date(o.pay_time) LIMIT 100", "rows": [ {"pay_date": "2024-11-01", "order_amount": 388.0, "order_count": 2, "avg_order_value": 194.0}, {"pay_date": "2024-11-02", "order_amount": 678.0, "order_count": 2, "avg_order_value": 339.0}, {"pay_date": "2024-11-03", "order_amount": 558.0, "order_count": 2, "avg_order_value": 279.0}, {"pay_date": "2024-11-04", "order_amount": 888.0, "order_count": 1, "avg_order_value": 888.0} ] }

注意2024-11-04这条数据里有一个A007订单，状态是refunded，所以它既没有被算进销售额，也没有被算进订单数。这正是语义层存在的意义：CASE WHEN拦截掉了不符合口径的数据，AI不需要理解“退款订单要不要剔除”这种业务问题。

如果返回结果中数据缺失，或者数字对不上，第一件事不是查代码，而是先打开semantic_model.yaml确认口径定义，再检查底层orders表数据是否被正确初始化。

9. 常见问题与排查思路

这里最容易被忽视的是“口径变更”问题。语义层一旦投入使用，它就不是一个普通配置文件，而是和数据库表结构一样需要管理的基础设施。任何指标的变更都要走评审、测试和发布流程，否则AI和BI会同时得到错误结果。

10. 最佳实践与工程建议

从“能跑”到“能生产”，还需要关注下面这些工程化问题。

10.1 指标命名必须全局唯一

不要出现“销售额”“销售金额”“GMV”混杂使用的情况。每个指标在语义模型中只能有一个全局唯一ID，例如order_amount、gmv。命名最好带上业务域前缀，比如trade.order_amount，避免将来扩展时撞名。

10.2 语义模型必须版本管理

YAML文件要进入Git仓库，和代码一起走MR评审。模型变更时，要记录变更原因和对下游的影响。建议在模型里增加version字段，并在API响应中带上版本号。

10.3 权限要下沉到指标和维度

不是所有AI Agent都有权限看所有数据。生产环境中，语义层要支持指标级、维度级、行级权限控制。比如普通客服Agent可以看“订单数”，但不能看“退款金额”；区域经理只能看自己负责区域的数据。这一层能力不能留给数据库白名单去处理，因为AI的一个工具调用可能覆盖多个指标。

10.4 性能优先靠缓存和物化

语义层API可能同时服务BI和多个AI Agent，实时计算压力会很大。常用的做法是：

• 高频指标查询使用结果缓存。
• 日级别指标预聚合到物化表。
• 底层引擎从SQLite换到ClickHouse、Doris或StarRocks等OLAP数据库。

本文的代码是逻辑演示，不建议直接在大型业务中让语义层直连业务库。

10.5 做好血缘和可观测性

每次语义查询API被调用，都应该记录：请求参数、生成的SQL、返回行数、耗时、调用方身份。这样才能回答“AI刚刚告诉我客单价下降5%，这个结论是怎么算出来的”。血缘追踪越完整，业务方对AI的信任度越高。

10.6 不要让大模型直接拼SQL

这条可以算是铁律。无论Prompt写得多好，都不要让大模型直接访问底层表结构并生成SQL。原因有三个：第一，口径不稳定；第二，权限难控制；第三，一旦产生错误结果，责任很难界定。最好的边界是：大模型负责意图识别和工具编排，语义层负责可靠计算。

10.7 用MCP等标准协议对接AI应用

如果Agent生态比较丰富，可以考虑把语义查询能力封装成MCP（Model Context Protocol）工具或OpenAPI工具。这样不同的AI应用（对话助手、自动化分析、Copilot）可以通过统一协议接入，不需要每个应用都重新写一遍调用逻辑。

11. 总结与后续学习方向

一个可靠的数据驱动AI应用，不能建立在“让模型自己翻表”的基础上。 “Model your business once – for humans and AI alike”不是一句口号，而是一种可行的架构路径：先把业务口径沉淀成统一语义模型，再通过API同时服务人类消费者和AI消费者。

本文用一个电商零售的最小项目，带你走完了从YAML语义模型定义、FastAPI语义查询接口、BI SQL核验到Agent工具调用的完整流程。你会发现，人类和AI第一次不再需要各自维护一份“数字解释”，而是共享同一份可复用、可审计、可版本化的业务模型。

接下来如果要继续深入，可以按这三个方向拓展：

• 语义层计算引擎：把只有YAML配置的最小实现升级为真正的指标平台，比如调研dbt Semantic Layer、Cube、Lightdash等开源方案。
• AI协议接入：把语义查询API封装成MCP工具，让支持MCP的AI客户端直接调用。
• 指标治理：联合业务方、数据团队共同制定指标口径评审机制，让每个数字都有明确责任人和审计记录。

建议先把本文的示例跑通，再把它映射到你自己的业务表里。花半天时间搭一个最小语义层，比给AI写二十条取数Prompt更值得。