企业官网建设流程全解析

日志系统里躺着一行触目惊心的记录：DELETE FROM users WHERE id = 1024，执行时间是凌晨 03:22:17，发起方是一段名为“数据分析助手”的自动化脚本。看到这行日志的那一刻，我的第一反应是：谁把生产库的连接串直接喂给大模型了？

这个场景不是假设，而是最近圈子里真实发生过的事情。很多团队拿到大模型的 API Key 之后，第一个想法就是让 AI 直接去查数据库、生成报表、分析业务，模型生成 SQL、程序负责执行，听起来很顺畅。但直接把数据库连接串丢给大模型，等于把一把万能钥匙交给了一个说话不过脑子的实习生，你不知道它下一秒会生成SELECT * FROM orders还是DROP TABLE orders。

这篇文章不打算劝退你，而是分享一个我和团队在实际落地中验证过的方案：用某家大模型平台提供的 Agents API，配合独立工具函数层和审计日志层，把大模型变成一个“可审计、可拦截、可追踪”的 AI 数据分析师。这个方案的核心思路是：永远不要给大模型直接访问数据库的能力，而是给它一组合法的工具，让它只能通过工具间接地、受控地触碰数据，同时把所有操作记录在案。

看完这篇文章，你应该能搞清楚三件事：为什么不能把数据库直接暴露给大模型；如何基于 Agents API 设计一个“只读 + 白名单 + 审计”的数据分析 Agent；我在实际开发过程中踩过哪些坑、怎么排查。整个内容会夹杂代码、配置和思维过程，建议你打开编辑器跟着敲一遍。

1. 先直面问题：为什么 “大模型 + 数据库连接串” 是个高危组合

很多人觉得，让大模型直接连数据库没什么大不了的，它有 SQL 能力，能写 JOIN、能写窗口函数，甚至能自动加索引建议。但你有没有想过一个关键问题：大模型不是按照“最小权限原则”来工作的，它是按照“最大概率生成原则”来工作的。

1.1 LLM 生成 SQL 的三宗罪

第一宗罪是路径不可控。你让模型写一个“统计各部门平均工资”的查询，它可能生成一个没有 WHERE 条件的SELECT * FROM employees JOIN departments ON...，这个 SQL 语法完全正确，但执行起来会导致数据库长时间锁表，甚至把内存撑爆。人工写 SQL 时会先想想这张表有多大、有哪些索引，但模型不会，它只会“生成最符合字面意思的语句”。

第二宗罪是权限越界。如果你把连接串交给了模型，模型生成的 SQL 理论上可以执行任意语句。就算你在数据库层只给了只读账号，它依然能写出SELECT * FROM sys.tables这种元数据探测语句，把表结构、字段名、数据库名全部摸一遍。更极端的情况是，如果模型真的生成了UPDATE或DELETE语句，而你的连接串恰好是管理权限，后果不堪设想。

第三宗罪是行为不可追溯。假设模型执行了一条有问题的 SQL，但你没有记录它完整的输入上下文、生成的 SQL 文本、执行时间、影响行数和结果物，事后复盘时你根本不知道它为什么这么做。黑盒+自动执行+权限过重，这三点叠在一起就是一场事故库。

我自己的切身体会是：凡是把数据库连接串直接塞给大模型的系统，上线之后都会经历一轮又一轮的“救火”。“它为什么会执行这条 SQL”“谁允许它删除数据的”“这个结果是不是它自己编的”，这些问题的本质是：你试图用“模型本身的能力”去解决数据可信问题，但模型根本没有稳定的约束能力，你必须用系统层面的机制去兜底。

1.2 “可审计”到底审的是什么

我说的“可审计”不仅仅是指系统日志里有多一行记录，而是指整个 Agent 的行为链路必须可还原。具体来说，你要能回答下面这几个问题：

• 用户问了什么（原始提问）
• 模型理解成了什么（经过系统提示词改写后的任务描述）
• 模型调用了什么工具（函数名和参数）
• 工具实际执行了什么 SQL（最终执行的规范化语句）
• 访问了哪些表、哪些列（数据血缘信息）
• 返回了什么结果（带脱敏的摘要）
• 是谁在什么时间触发的（用户、会话、请求ID）

只有把这些信息完整记录并按同一 ID 串联起来，你才敢说这个 AI 数据分析师是“可控的”。否则，它只是一个会说漂亮话的黑盒子。

2. 整体架构：把大模型“关进笼子”而不是“放虎归山”

当我决定做一个 AI 数据分析师时，第一版设计其实和大多数人一样：写一个 prompt，让大模型直接输出 SQL，然后丢给数据库执行。后来在测试阶段就发现了不少问题：模型生成的 SQL 不符合预期、偶尔会尝试访问无关的表、没有统一的上限控制。于是才逐步演进出下面这套架构。

2.1 理清 Agents API 的基本运行逻辑

传统的 API 调用模式是：你发一段 prompt，模型返回一段文本，仅此而已。但 Agents API 的用法不一样，它把“模型 + 工具 + 循环”做成了一个可编排的运行时。你可以定义一套工具（也就是函数），把这些工具的名字、描述、参数 schema 注册给 Agent，然后当用户提问涉及某个工具能力时，模型会主动发起一次“工具调用请求”，系统收到这个请求后执行真实的函数，再把函数结果返回给模型，模型继续推理直至完成。

这套机制天然适合“AI 数据分析”场景，原因很简单：模型的职责是理解问题、规划步骤、生成诉求，但真正执行 SQL 的职责必须交给工具函数。工具函数是普通代码，你可以加校验、加白名单、加审计逻辑、加超时控制，所有系统级约束都写在工具函数里。模型永远接触不到真正的数据库连接对象，它拿到的只是一份“函数清单”。

"""伪代码：Agents API 的基本形态（具体包名以你使用的平台为准）""" from agents_sdk import Agent, Runner agent = Agent( name="audit_analyst", instructions="你是数据分析师，只能使用提供的工具查询数据，禁止猜测数据。", tools=[describe_schema, run_select_query, audit_log], # 工具列表 ) result = Runner.run_sync(agent, "上个月订单量是多少？") print(result.final_output)

这段代码是核心形态。看到没有，tools列表里传的是三个函数，它们就是整个 Agent 的“手和脚”，模型不能自己打开数据库连接，它只能请求这些函数替它做事。

2.2 服务分层与目录结构

我建议把整个服务分成三层，各司其职，尽量不要写在一起：

• 交互层：负责接收用户输入、初始化 Agent、返回结果，可以是 FastAPI 接口。
• 工具层：注册给 Agent 的所有函数，包括表结构描述、SELECT 查询、审计写入，这一层是安全核心。
• 存储层：数据库连接池、审计日志表、结果缓存，所有和存储交互的代码都收敛在这一层。

目录结构大概长这样：

services/ ├── app.py # FastAPI 入口 ├── agents_setup.py # Agent 配置与注册工具 ├── tools/ │ ├── schema_tools.py # 表结构白名单工具 │ ├── query_tools.py # SELECT 只读查询工具 │ ├── validation.py # SQL 校验器 │ └── audit_tools.py # 审计日志工具 ├── storage/ │ ├── db_pool.py # 数据库连接池 │ └── audit_repo.py # 审计日志数据访问 └── config.py # 策略配置（表白名单、限额等）

这样的分层方式带来的直接好处是，后续你要加一个“数据脱敏”规则，只需要改query_tools.py和validation.py，完全不用碰 Agent 的 prompt；要加一张可访问的新表，只需要改配置和schema_tools.py。系统演进的空间大了很多。

2.3 工具清单与权限矩阵

我最终暴露给 Agent 的工具只有三个：

这里最重要的设计是：不要给模型提供一把“万能执行器”。你不需要让 Agent 去执行任意 SQL，它需要什么数据，就通过“描述表结构”和“执行单选查询”这两个动作去拿。模型没有能力去执行UPDATE、DELETE、DROP，因为这个能力压根就不在工具列表里。

3. 关键代码拆解：实现一个只读且留痕的查询链路

方案大家都会讲，真正有价值的是落地细节。下面我按工具函数的开发顺序，一五一十地把核心代码和配置讲透。

3.1 第零步：给模型一张“白名单地图”

AI 数据分析师不能什么都敢查。我和业务方一起盘了所有表之后，最终确定了一个表白名单，只有名单内的表才允许被 Agent 感知和查询。describe_schema工具是这样写的：

"""表结构白名单工具：模型只能看到被允许的表""" import sqlalchemy as sa ALLOWED_TABLES = { "public.orders", "public.users", "public.products", "public.order_items", "public.payments", } def describe_schema(table_name: str) -> dict: # 第一步：白名单校验，不属于允许范围直接拒答 if table_name not in ALLOWED_TABLES: return {"error": "table_not_allowed"} # 第二步：从 information_schema 拉取结构信息 query = text(""" SELECT column_name, data_type, is_nullable FROM information_schema.columns WHERE table_schema = :schema AND table_name = :name """) schema, name = table_name.split(".") # 第三步：把字段信息压缩成模型容易理解的描述 columns = [] with engine.connect() as conn: rows = conn.execute(query, {"schema": schema, "name": name}) for row in rows: columns.append({ "name": row.column_name, "type": row.data_type, "nullable": row.is_nullable }) return {"table": table_name, "columns": columns}

你会发现我特意把返回值压成了一个字典，而不是直接把原始查询结果抛给模型。模型读这种结构化描述比读大段文本要容易得多，token 消耗也少。另外要注意，绝对不要在工具返回里附带连接串、账号密码之类的东西，这种信息一旦被模型读入，就可能在对话中间接泄露出去。

# 示例：模型视角下 describe_schema 的返回结果 {"table": "public.orders", "columns": [ {"name": "id", "type": "integer", "nullable": false}, {"name": "created_at", "type": "timestamp", "nullable": false}, {"name": "amount", "type": "numeric", "nullable": true} ]}

这一步的本质是：先决定模型能“看见”什么，再决定它能“操作”什么。你看不见的东西，自然就不会去碰。

3.2 第一步：让 Agent 只能通过工具函数碰数据

接下来是核心的查询工具。run_select_query接收模型生成的 SQL，但它不是拿到就执行，而是先过一层校验器。只有校验通过，才允许进入数据库执行。

"""只读查询工具：SQL 必须经过五层校验才能执行""" import hashlib from sqlalchemy import text from sqlalchemy.exc import SQLAlchemyError def run_select_query(sql: str, max_rows: int = 200) -> dict: # 第一层：结构校验 v = SQLValidator(sql) if not v.is_select_only(): return {"error": "query_not_allowed"} # 第二层：命中的表必须在白名单内 tables = v.extract_tables() for t in tables: if t not in ALLOWED_TABLES: return {"error": f"table_not_in_whitelist: {t}"} # 第三层：强制追加 LIMIT sql_with_limit = v.force_limit(sql, max_rows) # 第四层：在只读事务中执行 try: with engine.connect() as conn: trans = conn.begin() # 默认不提交 result = conn.execute(text(sql_with_limit)) rows = result.fetchmany(max_rows + 1) # 多取一行做截断判断 row_count = len(rows) truncated = row_count > max_rows rows = rows[:max_rows] trans.rollback() # 确保只读，不产生任何写入 except SQLAlchemyError as e: return {"error": str(e), "sql": sql_with_limit} # 第五层：返回摘要和签名，原始结果不直接给模型 response = { "columns": list(result.keys()), "rows": rows, "row_count": min(row_count, max_rows), "truncated": truncated, "result_hash": hashlib.sha256( repr(rows).encode() ).hexdigest()[:16], } return response

这里有几个容易被忽略但非常关键的细节：

• 事务为什么要rollback()？因为我们要确保这条链路是真正只读的。即使校验器漏掉了某个语句，事务回滚也会把任何潜在写入撤销。双保险比单保险可靠得多。
• 为什么要多取一行fetchmany(max_rows + 1)？因为只有多取一行，你才知道结果有没有被截断。如果取回来刚好max_rows行，你没法判断后面还有没有更多数据。
• result_hash有什么用？它是审计日志里面的关键字段，用来验证模型返回给用户的结果和实际数据库结果是否一致，杜绝模型“编造数字”的可能。

3.3 SQL 校验器：逐层拆解模型生成的语句

我觉得这个项目里最值得花功夫的部分就是SQLValidator。一个直觉上“没什么大不了”的校验器，在实际使用中能挡下 90% 的意外。

先看核心逻辑：

"""SQL 校验器：一层层把模型生成的语句过筛子""" import re import sqlparse class SQLValidator: def __init__(self, sql: str): self.sql = sql.strip().rstrip(";") # 去掉结尾分号 def is_select_only(self) -> bool: # 不允许模型生成任何非 SELECT 语句 forbidden_tokens = [ "UPDATE", "DELETE", "INSERT", "DROP", "ALTER", "CREATE", "GRANT", "REVOKE", "TRUNCATE", "EXEC", "CALL" ] # 用词法分析而非简单字符串匹配 parsed = sqlparse.parse(self.sql) if not parsed: return False first_token = parsed[0].token_first().value.upper() if first_token != "SELECT": return False if any(token.upper() in forbidden_tokens for token in re.findall(r"\b\w+\b", self.sql)): return False return True def extract_tables(self) -> set: # 从 SQL 的 FROM/JOIN 子句中提取表名 tables = set() from_matches = re.findall(r"FROM\s+([a-zA-Z0-9_.]+)", self.sql, re.IGNORECASE) join_matches = re.findall(r"JOIN\s+([a-zA-Z0-9_.]+)", self.sql, re.IGNORECASE) tables.update(from_matches) tables.update(join_matches) return tables def force_limit(self, sql: str, limit: int) -> str: # 如果模型没写 LIMIT，系统强制追加 pattern = re.compile(r"\bLIMIT\b", re.IGNORECASE) if pattern.search(sql): return sql # 已包含 LIMIT，则不干预 return f"SELECT * FROM ({sql.rstrip(';')}) AS _inner_ LIMIT {limit}"

这里有一个很多人会踩的坑：判断 SELECT 语句不要用简单的字符串 contains 方法。模型可能会生成带注释的 SQL，比如/* hint */ SELECT ...，此时第一个 token 不是 SELECT，但语句确实是 SELECT。所以最好用专业的 SQL 词法分析库（比如sqlparse）去拿到真正的第一个 token。我见过不少团队第一版实现就是if "SELECT" in sql，结果注释里带一个 SELECT 就放行了，虽然最终影响不大，但不严谨。

force_limit这个函数有一个值得讨论的地方：为什么我用SELECT * FROM (子查询) LIMIT n而不是直接在原 SQL 后面加LIMIT n？因为有些 SQL 本身是一个完整的复合查询，直接追加可能造成语法错误；更稳妥的做法是把它包成一个子查询，再在外层做限制。需要注意的是，这种做法在 MySQL 8.0+ 上会导致派生表必须要有别名，所以我在代码里加了AS _inner_。这个别名是我在线上报错之后补上的，原因后面细说。

3.4 写审计日志的正确姿势

我不建议把审计日志写在 Agent 的after_tool_call钩子里顺手打一行，也不建议把它硬编码在run_select_query内部。更推荐的做法是：把“写审计”本身也实现为一个独立函数，然后在查询工具里去调用。这样审计逻辑可以单独演进、单独测试。

"""审计日志工具：记录一次查询的完整生命周期""" from datetime import datetime, timezone import uuid def write_audit_log(request_id: str, event_type: str, payload: dict) -> None: # 这里用独立的审计数据库连接池，避免与业务查询相互干扰 with audit_engine.begin() as conn: conn.execute(text(""" INSERT INTO query_audit_log ( log_id, request_id, event_type, user_input, model_generated_sql, normalized_sql, table_accessed, row_count, result_hash, created_at ) VALUES ( :log_id, :request_id, :event_type, :user_input, :model_generated_sql, :normalized_sql, :table_accessed, :row_count, :result_hash, :created_at ) """), payload | { "log_id": str(uuid.uuid4()), "request_id": request_id, "created_at": datetime.now(timezone.utc), })

审计模块最重要的一点是：审计库连接池必须和业务查询库分开。想一想，如果某个查询锁死了业务库的表，而审计日志恰好写在同一台库的同一张表里，那么审计操作也会跟着卡住。分开之后，即使业务库发生了严重问题，审计记录还是能正常写入，保证事后能排查清楚现场。

另外一个细节：不要在审计日志表里直接存完整的结果数据，那会爆炸。存result_hash和row_count就足够了。真到了需要追溯结果数据的阶段，你可以在结果缓存表里再去定位当时的快照。

4. 数据血缘：从“用户提问”到“返回报表”的完整留痕

市面上很多工具都在聊可观测性，但真正放到 AI Agent 场景里，“可观测”的含义会变得更具体。用户可能一次提问会触发多轮工具调用：先 describe 订单表，再 describe 用户表，再执行一次 JOIN 查询。每一轮调用都应该被记录，并且这些记录必须能关联回同一次用户请求。

4.1 审计日志的字段设计

一张审计表想要撑起整个追溯链路，字段至少要覆盖下面这些维度：

这里加一个turn_id很有必要，因为 Agent 在回答问题时可能要先看表结构再执行查询，同一个request_id下会有多条日志。没有turn_id，你很难还原“它到底先做了什么、后做了什么”。

4.2 全链路追踪 ID 怎么串起来

我的做法是：在请求入口生成一个request_id，然后把它作为上下文对象传入 Agent 和所有工具函数。这样从用户敲下问题开始，到每一次工具调用、每一个 SQL 执行，全部挂到同一个请求 ID 下。

"""在 FastAPI 接口中串联全链路 request_id""" from fastapi import FastAPI, Request, Depends from contextvars import ContextVar app = FastAPI() request_id_ctx: ContextVar[str] = ContextVar("request_id", default="") @app.post("/analyst/query") def handle_query(req: Request, body: QueryRequest): rid = str(uuid.uuid4()) token = request_id_ctx.set(rid) try: agent = build_agent_context(rid) result = run_agent(agent, body.question) return {"answer": result, "request_id": rid} finally: request_id_ctx.reset(token)

用ContextVar的好处是，即便后续把 Agent 改成异步并发调用，每个任务依然能正确拿到自己的request_id，不会串号。这个设计对上生产环境尤其重要，因为线上永远不可能只有一个用户同时提问。

4.3 结果脱敏与摘要回传

为了让 Agent 能基于数据做推理，我们必须把数据库查询结果回传给模型。但如果直接回传原始数据，会带来两个问题：第一，结果里的敏感字段可能被模型在后续对话中原样复述出来；第二，结果集太大会把上下文窗口塞爆，导致后续推理质量下降。

所以我的做法是：在工具函数返回结果之前，增加一个脱敏摘要层。

"""结果摘要化：把原始查询结果压缩成模型能用的统计信息""" SENSITIVE_COLUMNS = {"email", "phone", "address", "id_card"} def summarize_result(columns, rows, max_preview_rows=5): # 敏感字段直接丢弃，不回传模型 safe_idx = [i for i, col in enumerate(columns) if col.lower() not in SENSITIVE_COLUMNS] preview_columns = [columns[i] for i in safe_idx] # 数值型字段自动生成统计摘要 stats = {} numeric_idx = [] for i, col in enumerate(preview_columns): if col in {"amount", "price", "quantity", "count"} and rows: numeric_idx.append(i) values = [r[i] for r in rows if r[i] is not None] if values: stats[col] = { "sum": round(sum(values), 2), "avg": round(sum(values) / len(values), 2), "max": max(values), "min": min(values), } # 给模型只留预览行 + 统计量 preview_rows = rows[:max_preview_rows] preview_data = [ [r[i] for i in safe_idx] for r in preview_rows ] return { "preview_columns": preview_columns, "preview_rows": preview_data, "numeric_stats": stats, "total_row_count": len(rows), "truncated": len(rows) > max_preview_rows, }

这个函数的设计意图是：给模型足够的信息去理解数据，而不是给它完整的数据。模型只需要知道“订单总金额是 100 万、平均单价是 85 元、最大单笔是 5 万”就可以回答用户的问题了，完全不需要看到每一行原始记录。

5. 实战踩坑实录：这些坑我替你踩过了

代码写得再仔细，不上线跑一段时间，你永远发现不了真正的问题。这一节我把过程中遇到的比较典型的坑整理出来，希望对你有参考价值。

5.1 坑一：没有强制 LIMIT，一条 SELECT 打爆数据库内存

第一版校验器里，我以为模型一定会自己写LIMIT，就没做强制追加。结果测试的时候，用户问了一句“分析一下所有用户的下单规模分布”，模型生成的 SQL 直接是SELECT * FROM users JOIN orders ON...不带任何限制，一个 1000 万行的订单表直接被扫了个全表，数据库内存占用瞬间飙到 90%，整台机器的查询响应慢到像死机一样。

后来我吸取教训，强制在工具层追加外层 LIMIT，但第二次踩坑是在 MySQL 上执行SELECT * FROM (SELECT ...) LIMIT 200时数据库报了一个 “Every derived table must have its own alias” 的错误。原因就是派生表没有别名。这也是我在前面代码里写AS _inner_的背景。教训：光有 LIMIT 还不够，还要保证 LIMIT 包裹的语法在目标数据库上完全没有兼容性问题。

5.2 坑二：模型生成带注释的 SQL，词法校验被绕过了

有一次我查审计日志，发现模型生成了一条SELECT /*+ MAX_EXECUTION_TIME(1000) */ * FROM orders，因为当时校验器用的是简单的sql.startswith("SELECT")判断，这条带注释的 SQL 开头确实是 SELECT，顺利通过了校验。虽然这条查询本身没有危害，但它给我敲了警钟：模型是会“变通”的，你的校验规则必须建立在语法分析而不是字符串匹配上。

用sqlparse改造之后，我对所有关键字都做了 token 级别的判断。你可以想象一下，如果你只做简单的关键字过滤，模型生成一段含DELETE的存储过程调用，你的过滤可能根本发现不了。在大模型时代，字符串级别的安全校验已经不够了，必须用语法级别的解析。

5.3 坑三：审计表和业务表放在同一个事务里，互相锁死

开发初期，我把审计日志写入和业务查询放在同一个数据库连接里，想着能省一个连接。但有一次跑测试，我故意模拟了一个慢查询：模型生成的查询没有 WHERE，扫描了几百万行。结果这个查询锁住了业务表，审计日志写入也在同一条数据库连接里排队等待，导致审计模块也“死”了。最讽刺的是，当时我们想查一下审计记录看看到底发生了什么，结果审计表也卡住了。

后来我把两个数据库连接池彻底分离，一个是业务查询池，一个是审计写入池，互不干扰，这才解决。审计系统必须是系统的“事故记录仪”，如果事故记录仪也跟着事故一起坏了，那整个体系就失去意义了。

5.4 坑四：把 SQL 错误信息原样返回给用户，结果泄露了表结构

有一次模型生成了一条合法的查询，但查询中引用了一个已经删除的字段，数据库返回了详细的错误信息：“column xxx of relation public.users does not exist”。系统没有做任何包装，直接把错误信息拼到了模型上下文中，模型下一步直接把这个错误信息原封不动地复述给了用户。

从功能角度看，用户得到了一个“技术性错误”，虽然不算严重。但从信息安全角度看，这个模式很危险：如果模型生成的 SQL 触发了数据库的内部异常，数据库可能返回包括表结构、索引、甚至部分数据内容的详细信息。这些信息一旦出现在用户侧，就是泄露。

我后来在工具函数里做了一个统一错误包装器，只返回一个通用的“查询执行失败，请联系管理员”，把详细错误信息完整记录到审计日志里。对外界只展示结果，对内才展示全貌，这是审计类系统的基本素养。

5.5 问题排查速查表

这些坑并不是一次性踩完的，很多都是上线之后被业务人员“无意中触发”出来的。比如一个管理层用户问了一个“顺便看一下这个字段”，模型就开始尝试列一张不在白名单里的表，结果被校验器拦了下来。这个时候你会真实感受到，有一个独立的工具函数层挡在模型和数据库之间，是有多么重要。

最后想分享的一点个人体会

之前我一直觉得“让大模型写 SQL”是数据分析的终极形态，但做过这个项目之后我的认知完全转变了：真正有价值的不是让模型替我们写 SQL，而是建立一套机制，让模型在我们划定的安全边界内发挥它的推理和规划能力。一个可审计的 AI 数据分析师，重点并不在“AI”，而在于“可审计”三个字。

如果你现在正准备做类似的事情，我建议先从最小可用版本开始：三个工具、一张审计表、一个校验器。跑通之后再加数据脱敏、结果缓存、权限细化这些高级特性，不要一开始就想着做得很重。架构上记住一句话：模型永远是“提议者”，而不是“执行者”，所有落到数据库里的动作，必须经过代码，必须有记录。

这套方案后续还可以继续扩展：比如把表结构描述改成自动加载、把审计日志做报表可视化、把模型生成的 SQL 缓存下来做查询优化参考。但第一步永远是把“直接连数据库”这条危险路径封死，建立起属于你自己的安全边界。