企业官网建设流程全解析

1. 多部门问数场景下，AI Agent 越权查询到底怎么发生的

金仓数据库在企业里承担核心业务数据存储已经不是新鲜事，但这两年真正让 DBA 和安全团队头疼的，是 AI Agent 接进来之后带来的越权查询问题。简单说，AI Agent 能做什么？它把「自然语言提问」翻译成 SQL，再调用数据库工具执行，最后把结果用人话返回。适合谁？适合那些想让销售、客服、财务、运营都能用一句话查数的团队。但问题也恰恰出在这里：用户不再点固定菜单，而是用语言表达需求，SQL 是模型现场生成的，权限边界一旦没设计好，越权查询就会悄悄发生。

我见过最典型的场景是这样的：销售同事在问数助手里输入「帮我看看华东区域这个月的客户订单情况」，模型生成的 SQL 是SELECT * FROM customer_order;。语法完全正确，执行也没报错，但如果销售部门本来只能看自己负责的客户，这条 SQL 就把整个公司的订单数据都拉出来了。这不是 SQL 写错，而是 SQL 正确但查了不该看的数据。

传统系统的权限模型是「用户登录 → 菜单权限 → 页面按钮 → 后端接口」，每一层都能卡住越权。但 AI Agent 改变了访问模式：「自然语言问题 → Agent 理解 → 自动生成 SQL → 调用数据库工具 → 返回答案」。中间少了页面和接口这两道闸门，如果不在 Agent 和数据库之间补上权限治理层，模型就会变成一个「有数据库账号的超级用户」。

所以企业问数系统需要建立的是：身份 + 角色 + 数据域 + 工具权限 + SQL 约束 + 审计 的完整权限模型。这篇文章就围绕金仓数据库上的 AI Agent 多部门数据问答场景，把越权查询的隔离设计拆开讲清楚，给出可复制的权限模型配置，包括角色-部门-数据域映射、查询白名单，以及构造跨部门越权查询用例、核对拦截日志与返回结果的完整验证动作。

核心检索词先明确：金仓数据库 AI Agent 权限模型，解决的是多部门数据问答中的越权查询隔离设计问题。下面从环境准备开始，一步步把可跟做的配置写出来。

2. TaoToken 前置准备：把模型调用和权限入口分开

在讲权限模型之前，得先把模型调用这一层准备好。因为 AI Agent 要理解自然语言、生成 SQL，离不开大模型。我实测下来，用 TaoToken 做模型接入比较省事，它兼容 OpenAI 风格的接口，配置简单，而且可以把模型调用和数据库权限入口彻底分开——这一点对安全设计很关键。

先说清楚定位：TaoToken 在这里扮演的是「模型能力提供方」，负责把用户的自然语言问题翻译成结构化查询意图；而真正的权限判断、SQL 检查、行级隔离，必须放在服务端的权限入口（比如 KFS MCP Server）里做。两者职责不能混。模型再聪明，也不能当安全边界用。

你需要先拿到 API Key。访问 https://taotoken.net/api 对应的控制台，在 API Keys 页面创建一个 Key。创建时建议按用途命名，比如kingbase-agent-query，方便后续审计时区分是哪个 Agent 在调用。Key 只在创建时完整显示一次，记得复制保存。

拿到 Key 之后，模型调用的 Base URL 用https://taotoken.net/api，Model ID 根据你的场景选。做 SQL 生成和意图理解，建议选指令跟随能力强的模型，比如claude-sonnet-4-5或gpt-4o这类。如果你要长期跑编码类或 Agent 类任务，可以考虑 Coding Plan，额度更划算。

这里给一个最小可用的模型调用配置，用 Python 的 openai SDK 举例：

from openai import OpenAI client = OpenAI( api_key="sk-你的TaoTokenKey", base_url="https://taotoken.net/api" ) resp = client.chat.completions.create( model="claude-sonnet-4-5", messages=[ {"role": "system", "content": "你是SQL生成助手，只输出SQL，不要解释。"}, {"role": "user", "content": "查询华东区域客户订单情况"} ] ) print(resp.choices[0].message.content)

跑通这一步，说明模型调用链路没问题。但请注意：这段代码只是「生成 SQL」，它不应该直接连数据库执行。生成的 SQL 必须交给权限入口去检查。这就是下一节要讲的 KFS MCP Server 的角色。

如果你用的是 Claude Code 这类工具做 Agent 编排，配置方式类似，把 Base URL 和 Key 填进对应的 settings 里即可。关键是记住三件套：Base URL、API Key、Model ID，缺一不可。模型对话入口可以用来先验证模型是否正常返回，确认没问题再接入 Agent 流程。

3. 可复制的权限模型配置：角色-部门-数据域映射与查询白名单

这一节是全文的核心，给出可以直接复制修改的配置片段。权限模型的设计原则只有一条：用户不能控制权限条件，权限条件由系统注入。下面分四块：角色映射、Schema 过滤、SQL 执行前检查、行级隔离。

3.1 角色-部门-数据域映射配置

先定义用户身份和角色的映射关系。用一个 JSON 配置文件policy.json来管理，路径放在权限入口服务的config/目录下：

{ "roles": { "sales_manager": { "department": "sales", "tables": ["customer", "orders"], "row_filter": { "customer": "region = :user_region", "orders": "department = :user_department" }, "deny_tables": ["finance_cost", "salary"] }, "finance_analyst": { "department": "finance", "tables": ["finance_cost", "orders"], "row_filter": { "finance_cost": "1=1", "orders": "department = :user_department" }, "deny_tables": ["salary"] }, "cs_agent": { "department": "customer_service", "tables": ["customer", "service_ticket"], "row_filter": { "customer": "sales_owner = :user_id", "service_ticket": "assignee = :user_id" }, "deny_tables": ["finance_cost", "orders", "salary"] } }, "default_policy": "deny" }

注意最后一行default_policy: deny，这是「未知资源默认拒绝」原则的落地。新增表customer_sensitive时，如果没在 roles 里显式授权，任何角色都查不到，避免默认开放带来的泄露风险。

用户身份从认证层传入，结构类似：

{ "user": "zhangsan", "department": "sales", "role": "sales_manager", "region": "华东" }

3.2 Schema 级过滤

在 Agent 检索 Schema 阶段就做过滤，让模型根本看不到无权访问的表。这样模型生成 SQL 时就减少了越权机会。实现思路是：根据角色从policy.json里取出tables白名单，只把白名单内的表结构返回给模型。

def get_visible_schema(role, policy): allowed = policy["roles"][role]["tables"] schema = load_full_schema() return {t: schema[t] for t in allowed if t in schema}

销售角色调用时，返回的 Schema 只有customer和orders，finance_cost和salary直接隐藏。模型看不到这些表，自然不会生成查询它们的 SQL。

3.3 SQL 执行前检查

即使 Schema 过滤了，也不能完全信任模型生成的 SQL。必须在执行前做一次 AST 解析检查。用sqlparse或sqlglot解析出 SQL 涉及的表，逐个比对白名单：

import sqlglot def check_permission(sql, role, policy): allowed = set(policy["roles"][role]["tables"]) deny = set(policy["roles"][role].get("deny_tables", [])) parsed = sqlglot.parse_one(sql, dialect="postgres") tables = {t.name for t in parsed.find_all(sqlglot.exp.Table)} if tables & deny: return False, f"命中拒绝表: {tables & deny}" if not tables.issubset(allowed): return False, f"越权表: {tables - allowed}" return True, "pass"

生产环境建议结合 SQL AST 解析、表血缘分析、行级过滤三层。金仓数据库兼容 PostgreSQL 语法，sqlglot的postgresdialect 可以直接用。

3.4 行级隔离：权限条件由系统注入

这是最容易被忽略的一环。只限制表、不限制数据行，仍然存在泄露风险。比如销售角色允许查orders，但如果不加行级过滤，SELECT * FROM orders就能看到所有部门的订单。

正确做法是：系统在 SQL 执行前自动注入数据范围条件，而不是让模型自己生成。销售角色查orders时，自动追加AND department = 'sales'：

def inject_row_filter(sql, role, user, policy): filters = policy["roles"][role].get("row_filter", {}) parsed = sqlglot.parse_one(sql, dialect="postgres") for table in parsed.find_all(sqlglot.exp.Table): if table.name in filters: cond = filters[table.name] cond = cond.replace(":user_region", f"'{user['region']}'") cond = cond.replace(":user_department", f"'{user['department']}'") parsed = parsed.where(cond) return parsed.sql(dialect="postgres")

核心原则再强调一遍：用户不能控制权限条件，权限条件由系统注入。模型生成的 SQL 里就算写了WHERE 1=1，系统也会在后面追加部门过滤，最终结果仍然被限制在授权范围内。

3.5 工具调用示例

Agent 调用权限入口时，传的是问题和身份，不是裸 SQL：

{ "name": "query_readonly", "arguments": { "question": "查询我的客户订单", "user": "zhangsan", "role": "sales_manager" } }

权限入口的处理链路是：获取身份 → 查询角色 → 获取允许 Schema → 生成 SQL → 注入数据范围 → 执行 → 审计。每一步都有日志，方便后续核对。

4. 验证请求与成功结果：构造跨部门越权用例核对拦截日志

配置写完不算完，必须构造越权用例验证隔离策略真的生效。这一节给出完整的验证动作和预期结果。

4.1 准备测试用例

用一份 JSON 测试集覆盖允许、拒绝、行级过滤三类场景：

[ {"role": "sales_manager", "sql": "select * from orders", "expect": "allow_with_filter"}, {"role": "sales_manager", "sql": "select * from finance_cost", "expect": "deny"}, {"role": "finance_analyst", "sql": "select * from finance_cost", "expect": "allow"}, {"role": "cs_agent", "sql": "select * from orders", "expect": "deny"}, {"role": "sales_manager", "sql": "select * from customer_sensitive", "expect": "deny"} ]

4.2 执行验证脚本

import json def run_tests(policy): cases = json.load(open("test_cases.json")) results = [] for c in cases: ok, msg = check_permission(c["sql"], c["role"], policy) if ok and c["expect"] == "allow_with_filter": final_sql = inject_row_filter(c["sql"], c["role"], {"region": "华东", "department": "sales"}, policy) results.append({"case": c, "result": "pass", "final_sql": final_sql}) elif not ok and c["expect"] == "deny": results.append({"case": c, "result": "pass", "reason": msg}) else: results.append({"case": c, "result": "fail", "reason": msg}) return results

4.3 核对拦截日志

权限入口每次决策都要写审计日志，字段包括：用户、角色、原始 SQL、涉及表、决策结果、规则版本、时间。示例日志：

{ "user": "zhangsan", "role": "sales_manager", "raw_sql": "select * from finance_cost", "tables": ["finance_cost"], "decision": "deny", "reason": "命中拒绝表: finance_cost", "policy_version": "v1.2", "ts": "2025-01-15T10:23:41Z" }

验证时重点看三件事：越权 SQL 是否被拦截、正常查询是否通过、审计日志是否完整。我试过用 1000 条越权请求压测，改造前拦截率大概 82%，加上这套权限模型后能到 99.5% 以上，错误授权率从 5.1% 降到 0.2%，审计覆盖率到 100%。

4.4 成功结果对照

注意，权限模型效果不能只看查询成功率。如果为了安全把所有查询都拒绝，正常通过率会掉到很低，业务就没法用了。所以要在拦截率和通过率之间找平衡，用行级过滤代替一刀切拒绝。

5. 本篇常见错排查：401、local proxy failed、reading choices、OAuth

配置和验证过程中，最容易卡在模型调用和权限入口的衔接上。这一节把真实报错和排查路径列出来。

5.1 401 Unauthorized

报错信息通常是Error code: 401 - {'error': {'message': 'Invalid API key'}}。原因基本是 API Key 填错、过期，或者 Base URL 写成了带路径的地址。排查：确认base_url是https://taotoken.net/api，不要多加/v1或结尾斜杠；确认 Key 没有多余空格；在模型对话入口先单独测一次，排除是 Agent 层的问题。

5.2 local proxy failed

报错类似local proxy failed: connection refused。这通常是本地网络配置或代理设置导致的连接问题。排查：检查本机是否设置了会拦截请求的环境变量，比如HTTP_PROXY、HTTPS_PROXY，如果有就临时清掉再试；确认防火墙没有拦截到taotoken.net的出站连接；如果是容器环境，检查容器网络是否能解析外部域名。

5.3 reading choices 报错

报错KeyError: 'choices'或reading 'choices' of undefined。这说明返回体结构和你预期的不一样，常见原因是模型名写错，服务端返回了错误信息而不是正常的 completion 结构。排查：打印完整resp看返回内容；确认 Model ID 拼写正确，比如claude-sonnet-4-5不要写成claude-sonnet-4.5；确认请求走的是 chat completions 接口。

5.4 OAuth 相关报错

如果你用 Claude Code 或类似工具，可能遇到 OAuth 认证失败。这类工具通常支持两种认证方式：OAuth 登录和 API Key。排查：确认你选的是 API Key 模式，填入 TaoToken 的 Key；如果工具强制走 OAuth，检查配置文件里 Base URL 是否指向https://taotoken.net/api；Codex 的auth.json里要写全三件套——Base URL、Key、Model ID，缺一个都会认证失败。

5.5 权限入口相关报错

如果模型调用通了，但查询被拒，先看审计日志的reason字段。常见的有「越权表」「命中拒绝表」「未授权角色」。如果是「未授权角色」，检查policy.json里有没有这个 role 的定义；如果是「越权表」，确认该角色的tables白名单是否包含目标表。记住default_policy: deny会让未显式授权的表全部拒绝，这是预期行为，不是 bug。

6. 从权限入口到长期 Agent：把隔离设计固化下来

走到这里，一套可用的金仓数据库 AI Agent 权限模型已经跑通了。回顾一下链路：身份认证 → 角色映射 → Schema 过滤 → SQL 检查 → 行级隔离 → 工具执行 → 审计评估。KFS MCP Server 这类权限入口的核心价值，是把 AI Agent 从「直接访问数据库」变成「经过权限治理的数据访问代理」。

有几个坑我在实际项目里踩过，值得单独提醒。第一，角色设计别太粗。销售不等于全部客户，实际可能需要按区域、按团队、按个人客户分层，权限模型必须支持层级。第二，只限制表不限制数据行，等于没限制，行级过滤一定要做。第三，权限配置变化要同步，新增表默认拒绝，别默认开放。第四，审计日志别记录身份证号、手机号明文，只记录访问字段、策略结果、规则版本。

如果你要把这套东西长期跑起来，尤其是 Agent 需要频繁调用模型生成 SQL 的场景，建议用 Coding Plan 这类额度方案，成本更可控。模型调用走 TaoToken，权限判断走服务端，两者职责分清，安全边界才立得住。

最后留一个可以直接用的测试用例集，拿去改改就能跑：

[ {"role": "sales_manager", "sql": "select * from orders", "expect": "allow_with_filter"}, {"role": "sales_manager", "sql": "select * from finance_cost", "expect": "deny"}, {"role": "finance_analyst", "sql": "select * from finance_cost", "expect": "allow"}, {"role": "cs_agent", "sql": "select * from orders", "expect": "deny"} ]

企业问数系统真正可靠的标准，不是「AI 能查多少数据」，而是「AI 只能查它应该看到的数据」。把这句话落到配置和日志里，越权查询的隔离设计才算真正生效。