日志系统里躺着一行触目惊心的记录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( nameaudit_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 的工具只有三个工具函数作用输入参数权限边界describe_schema向模型提供表结构和字段说明表名白名单内只能返回已授权的表信息字段仅限已授权列run_select_query执行只读查询并返回结果摘要SQL 文本、返回行数上限强制 SELECT、强制 LIMIT、只读事务write_audit_log记录审计事件请求ID、SQL、结果哈希只能写审计库无法被模型读取回传这里最重要的设计是不要给模型提供一把“万能执行器”。你不需要让 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: ftable_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(rFROM\s([a-zA-Z0-9_.]), self.sql, re.IGNORECASE) join_matches re.findall(rJOIN\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 fSELECT * 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 审计日志的字段设计一张审计表想要撑起整个追溯链路字段至少要覆盖下面这些维度字段类型用途log_idUUID单条日志主键request_idUUID同一次用户请求的唯一标识turn_id整型多轮对话中的轮次序号event_typeVARCHAR枚举schema_describe / query_execute / audit_writeuser_inputTEXT原始提问model_generated_sqlTEXT模型生成的 SQL未规范化normalized_sqlTEXT经过校验器改写后的 SQLtable_accessedVARCHAR实际访问的表名row_count整型返回行数result_hashVARCHAR结果摘要哈希error_messageTEXT如果有错误记录原始错误信息created_atTIMESTAMP操作时间这里加一个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_rows5): # 敏感字段直接丢弃不回传模型 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表结构描述太复杂、字段名不直观在describe_schema里补充字段的注释说明比如“status1 表示已完成”查询执行极慢缺少索引在审计日志里找到高频查询交给 DBA 在对应列上加索引模型返回结果和库里对不上查询结果被截断检查truncated字段提示模型使用聚合查询而不是明细查询用户输入了非数据库问题没有触发任何工具在系统提示词里面加一条规则非数据问题直接拒绝回答SQL 校验器误杀合法查询校验规则过严查看被拦截的 SQL 日志逐条分析后调整白名单或放宽规则审计日志里有大量重复查询模型多次重试为相同 request_id 下相同的 SQL 增加去重逻辑这些坑并不是一次性踩完的很多都是上线之后被业务人员“无意中触发”出来的。比如一个管理层用户问了一个“顺便看一下这个字段”模型就开始尝试列一张不在白名单里的表结果被校验器拦了下来。这个时候你会真实感受到有一个独立的工具函数层挡在模型和数据库之间是有多么重要。最后想分享的一点个人体会之前我一直觉得“让大模型写 SQL”是数据分析的终极形态但做过这个项目之后我的认知完全转变了真正有价值的不是让模型替我们写 SQL而是建立一套机制让模型在我们划定的安全边界内发挥它的推理和规划能力。一个可审计的 AI 数据分析师重点并不在“AI”而在于“可审计”三个字。如果你现在正准备做类似的事情我建议先从最小可用版本开始三个工具、一张审计表、一个校验器。跑通之后再加数据脱敏、结果缓存、权限细化这些高级特性不要一开始就想着做得很重。架构上记住一句话模型永远是“提议者”而不是“执行者”所有落到数据库里的动作必须经过代码必须有记录。这套方案后续还可以继续扩展比如把表结构描述改成自动加载、把审计日志做报表可视化、把模型生成的 SQL 缓存下来做查询优化参考。但第一步永远是把“直接连数据库”这条危险路径封死建立起属于你自己的安全边界。