企业BI底座接入AI实操指南：指标、权限与验收

教程目标

通用大语言模型擅长写邮件、做总结、翻译等通用任务。但财务、经营分析、供应链等岗位需要基于真实数据做决策。如果模型没有接触企业数据，它给出的结论往往来自通用知识的推测，无法直接用于汇报或业务判断。

本教程帮助技术团队把企业已有的BI（商业智能）底座与大模型原生集成，搭建一条可溯源、可管控的分析链路。链路的分工是：模型负责理解用户问题、选择指标和筛选条件；BI底座负责计算、权限校验和结果返回。

读完本教程，你可以得到以下交付物：

1. 指标中心定义模板，包含指标名、口径说明、维度和数据源。
2. 模型与底座之间的接口定义，包括请求字段、响应结构和错误码。
3. 权限用例矩阵和审计字段清单。
4. 上线前的验收表，用于逐项确认。

需要说明的是，本方案能降低模型凭经验编造数字的风险，但不能消除风险。每个关键数字仍需能追溯到底座中的指标或明细，重要结论仍需人工复核。

适用场景与边界

适用场景包括：

• 经营分析：如区域毛利波动分析、门店或大区业绩对比。
• 供应链监控：如准时交付率（OTD）卡点排查、供应商交付趋势分析。
• 库存管理：如库存周转天数跟踪、呆滞库存预警。

以下情况不适合直接套用本方案：

• 底层数据质量很差，且没有人负责清洗和口径确认。
• 业务口径仍在频繁变动，指标中心无法稳定定义。
• 需要模型直接生成对外披露的财务报表，此类内容必须经过财务人员审核。

前置准备

在开始配置前，请逐项确认下表内容。选择项需要与团队的实际环境对应，并记录下来，便于后续排查。

检查项	需要确认的内容	记录示例
BI产品与版本	产品名称、版本号、是否支持API查询和行列权限	自建BI 5.x，支持REST查询接口
数据源与连接方式	数据库类型、连接账号、连接方式（直连、JDBC、数据仓库中间层）	数据仓库 sales_dwh，只读账号
大模型API版本	模型名称、接口版本、是否支持函数调用或结构化输出	模型版本号、结构化输出开关
网络白名单	模型网关到BI底座的出入口IP或域名，端口和TLS要求	仅开放 443 端口，双向校验证书
账号与权限申请	服务账号、API调用角色、权限变更审批人	由IT负责人审批后开通
审计与日志	日志保存周期、存储位置、访问人员范围	保存180天，仅安全与审计组可查

云服务部署时，可以选择企业已有的主流云服务商提供的BI组件或数据仓库服务。选型时优先考虑已有合同、数据驻留要求和团队熟悉度，而不是只看宣传能力。

实施步骤

步骤一：接入并清洗企业真实业务数据

• 做什么：把业务数据库中的销售、订单、库存等数据抽取到BI底座的数据引擎，并完成基础清洗。
• 抽取方式：
  • 全量抽取适用于首次建库和小表，逻辑简单，但耗时长、占用资源多。
  • 增量抽取适用于日常同步，常用更新时间字段或变更日志作为游标。每次抽取都要记录游标位置，便于断点续传。
• 字段映射：建立一份字段字典，写明每个字段的业务含义、单位、取值范围和来源系统。例如，“毛利”和“净利润”必须分开定义，不能共用同一个字段。
• 数据质量校验：至少检查以下几类问题，并把结果写入质量报告：
  • 关键字段是否为空，如区域、日期、金额。
  • 金额是否为负数，如果存在，是否有对应的退货或冲销单。
  • 同一订单是否重复入库。
  • 数据日期是否晚于当前时间，或跨月断档。
• 备份与回滚：在每次全量重建或大规模修改前，先备份原始表和映射配置。若校验失败，应能回滚到上一个版本，不得直接覆盖生产数据。
• 为什么做：模型无法稳定理解杂乱的原始数据。统一的数据底座和字段字典，能减少脏数据对结论的干扰。

步骤二：构建企业级指标中心

• 做什么：在BI底座中定义核心业务指标的计算逻辑，例如“区域毛利”“OTD准时交付率”“库存周转天数”。
• 为什么做：指标中心是保证分析可信的关键。模型如果每次自行推导计算公式，容易出错。指标统一定义后，模型只需要调用指标名称，具体计算由底座完成。
• 注意什么：
  • 指标名称在全局必须唯一。
  • 不同区域对“准时交付”的统计口径可能不同，例如按出库时间还是签收时间统计。口径差异必须在数据模型层统一，或拆分为不同指标，并给出明确的命名。
  • 每个指标都要写明口径版本和生效日期。口径变更时，旧版本保留，便于历史数据复算。

配置示例（示意结构，字段名需替换为所用BI产品的真实字段）：

{
  "metric_name": "region_gross_profit",
  "version": "v2",
  "description": "区域毛利，含税金额口径，不含退货冲销",
  "calculation_logic": "SUM(revenue) - SUM(cost)",
  "dimensions": ["region", "month", "product_category"],
  "data_source": "sales_dwh",
  "owner": "财务分析组",
  "effective_date": "2026-01-01"
}

步骤三：配置AI调用权限、脱敏与审计

• 做什么：在BI底座中设置行级和列级权限，为AI服务分配专用的API调用角色，并开启审计和脱敏。
• 权限绑定：权限要与企业组织架构树对应。例如，华南区主管通过AI查询毛利时，底座应自动过滤为华南区数据，即使模型传入了其他区域的筛选条件。
• 身份来源：用户身份必须来自已登录的会话，由后端注入请求。模型不得自行指定用户身份或令牌。
• 敏感字段脱敏：员工薪酬、客户手机号、合同单价等字段，默认不返回模型。确需返回时，应做掩码或聚合处理，并在审计日志中标记。
• 越权拦截：越权请求应返回明确的错误码（见步骤四），并记录到审计日志。不得返回空结果后让模型自行解释，否则用户会误以为数据不存在。
• 权限变更复核：角色调整、区域归属变化、服务账号增加权限，都应走审批流程。每季度至少复核一次AI服务账号的权限范围。
• 审计日志字段：至少记录以下字段：trace_id、用户ID、会话ID、时间戳、原始问题、模型解析出的指标和筛选条件、底座返回的错误码、命中的数据范围和行数。
• 为什么做：BI工具的核心价值是可管控。权限和审计缺失时，AI查询会绕过原有的数据安全边界。

步骤四：打通AI与BI底座的自然语言查询链路

• 做什么：把BI底座的指标查询能力封装为HTTP接口，供模型通过函数调用触发。
• 分工原则：模型只负责识别意图、选择指标和生成筛选条件。计算、权限校验、取数全部由底座完成。模型拿到的是底座返回的结构化结果，再基于这些结果撰写归因分析。
• 接口定义：下列字段名为示例，实际接口以所用BI产品的文档为准。

请求示例：

{
  "trace_id": "7f3c2a9e-5b1d-4c8e-9a10-2d6e8f4b1c77",
  "metric": "region_gross_profit",
  "version": "v2",
  "filters": {"region": ["华南"], "month": "2026-10"},
  "dimensions": ["region", "month"],
  "timeout_ms": 5000
}

成功响应示例：

{
  "trace_id": "7f3c2a9e-5b1d-4c8e-9a10-2d6e8f4b1c77",
  "code": 0,
  "message": "ok",
  "data": [{"region": "华南", "month": "2026-10", "region_gross_profit": 1234567.8}],
  "data_as_of": "2026-11-11T02:00:00+08:00",
  "source_url": "https://bi.example.internal/dashboards/profit?region=华南"
}

错误码定义示例（需与底座实际返回值对齐）：

code	含义	处理方式
0	成功	返回数据并附带溯源链接
40001	参数非法，如未知字段或非法日期	不重试，要求模型修正参数或请用户补充信息
40101	令牌缺失或过期	不重试，要求用户重新登录
40301	无权限访问该数据范围	不重试，告知用户无权限，并记录审计
40401	指标不存在或版本不匹配	不重试，返回可用指标列表
50001	底座内部错误	最多重试2次，间隔递增
50401	查询超时	最多重试1次，仍失败则提示稍后再试

示意代码（伪代码结构，需替换为实际的鉴权方式和SDK，不可直接运行）：

import os
import uuid
import requests
BI_BASE_URL = os.environ["BI_BASE_URL"]          # 从环境变量或密钥服务读取
ALLOWED_METRICS = {"region_gross_profit", "otd_rate", "inventory_turnover_days"}
ALLOWED_FILTER_KEYS = {"region", "month", "product_category"}
TIMEOUT_S = 5
MAX_RETRY = 2
RETRYABLE_CODES = {50001, 50401}
class BIQueryError(Exception):
    pass
def validate_intent(intent: dict) -> dict:
    """校验模型给出的参数，只接受白名单内的指标和字段。"""
    metric = intent.get("metric")
    if metric not in ALLOWED_METRICS:
        raise BIQueryError("40401: 指标不在白名单内")
    filters = intent.get("filters") or {}
    unknown = set(filters) - ALLOWED_FILTER_KEYS
    if unknown:
        raise BIQueryError(f"40001: 非法筛选字段 {sorted(unknown)}")
    return {
        "metric": metric,
        "filters": filters,
        "dimensions": intent.get("dimensions") or ["region"],  # 默认维度
    }
def query_metric(intent: dict, session_token: str) -> dict:
    """session_token 由后端从登录会话注入，不来自模型。"""
    trace_id = str(uuid.uuid4())
    body = validate_intent(intent)
    body.update({"trace_id": trace_id, "timeout_ms": TIMEOUT_S * 1000})
    headers = {
        "Authorization": f"Bearer {session_token}",
        "X-Trace-Id": trace_id,
    }
    for attempt in range(MAX_RETRY + 1):
        try:
            resp = requests.post(
                f"{BI_BASE_URL}/v1/metrics/query",
                json=body,
                headers=headers,
                timeout=TIMEOUT_S,
            )
        except requests.Timeout:
            if attempt == MAX_RETRY:
                raise BIQueryError(f"50401: 查询超时 trace_id={trace_id}")
            continue
        payload = resp.json()
        code = payload.get("code")
        if code == 0:
            return payload
        if code in RETRYABLE_CODES and attempt < MAX_RETRY:
            continue
        raise BIQueryError(f"{code}: {payload.get('message')} trace_id={trace_id}")
    raise BIQueryError(f"50001: 重试耗尽 trace_id={trace_id}")
def answer_with_ai(user_question: str, llm, session_token: str) -> str:
    # 1. 模型只解析意图，输出结构化参数
    intent = llm.parse_intent(
        question=user_question,
        tools=["query_metric"],
    )
    # 2. 后端完成校验、鉴权、取数
    try:
        result = query_metric(intent, session_token)
    except BIQueryError as e:
        return f"查询未完成：{e}"  # 错误原样返回，不让模型猜测数字
    # 3. 模型基于底座结果撰写归因，要求附带溯源信息
    prompt = (
        f"用户问题：{user_question}\n"
        f"底座数据（仅可使用以下数据，不得补充数字）：{result['data']}\n"
        f"数据时间：{result['data_as_of']}，来源：{result['source_url']}\n"
        "请给出结论，并标注每个数字对应的来源。"
    )
    return llm.generate(prompt=prompt)

上述示例的注意点：

• 模型输出的参数必须经过白名单校验，不能直接拼接进查询语句。
• 超时和5xx类错误才重试，权限类错误不重试。
• 模型只能使用底座返回的数据，错误信息原样呈现给用户。

结果验证与验收清单

配置完成后，按下表逐项验收。阈值由团队在上线前确定，建议与财务和业务负责人共同商定。

验证项	操作方法	验收标准
意图识别	准备30条典型问题，覆盖毛利、OTD、库存等指标，包括同义表达	指标选择正确率达到团队约定阈值，错误样本逐条记录原因
参数校验	构造未知指标、未知字段、非法月份等请求	全部返回对应错误码，且不触发取数
权限隔离	用华南、华东、普通员工三类账号，各问同一问题	返回范围与权限矩阵一致，越权请求返回40301并写入审计
数据溯源	随机抽取20条AI回答，逐个数字在BI中核对	数值与底座明细或仪表板一致，每条回答都有来源链接
审计完整性	抽查审计日志，检查每条查询的字段	字段完整，trace_id可在模型网关、底座日志之间串联
脱敏检查	询问敏感字段相关问题	敏感字段不出现在模型输入或输出中
异常处