数据团队每天收到最多的需求是什么不是建模不是拖拽一个新图表而是那句听到就头疼的“帮我导个数急着要今天下班前”。过去半年我一直在做一个实验把这种高频、重复、长期占用数据工程师人力的取数请求交给一个基于 OpenAI Agents-API 体系搭建的数据分析师 Agent 来消化。业务同事在对话框里用自然语言提问这个 Agent 负责理解诉求、发现表结构、编写 SQL、调用安全校验、执行查询最后把结果用普通人看得懂的话解释回来。整个过程不是模型一次性“猜”出 SQL而是通过工具调用与多轮反馈完成的关键是每一次尝试都留有审计记录每一次执行都会经过权限校验。这个方案的定位很明确它不是让 AI 替数据分析师写复杂报表而是把“取数”这条最基础、最重复、最容易出问题的链路自动化让业务方自己就能拿到数据同时数据团队从 SQL 翻译官变成规则制定者和监护人。适合谁参考如果你和我一样是正在做 AI 应用落地的大模型开发工程师或者你在数据团队里每天被取数需求淹没想找一条既能提效又不失控的路线这篇整理就是给你准备的。1. 需求分析与整体设计思路1.1 被取数需求淹没的团队真正缺的是什么数据团队内部都有这种体会业务部门提“我要一个数”实际背后是一连串隐性知识。数据表分散在几十张表里哪些表是权威的、日期口径到底是自然日还是工作日、订单金额含不含退款这些背景知识不在系统里而在做了三年的老同事脑子里。传统自助 BI 工具确实能降低门槛但它有两个硬伤一是业务人员还是得学会拖字段、做过滤很多人学不会也不想学二是指标口径一旦做成固定维度就失去灵活性业务人员想按自己的方式切片还是要提需求。所以真正缺的不是又一个看板工具而是一个“能听懂人话、知道去哪儿取数、又不会乱来”的中间层。自然语言取数方案听起来很顺理成章难点落在工程上模型如何知道数据库里有什么表如何把用户模糊描述映射到精确字段如何阻止模型生成越权或危险 SQL如何审计和追责方案选型上我做过一轮比较大致三类方案优点缺点直接 Prompt 一次生成 SQL实现最快Demo 五分钟拿不到真实 schema、SQL 错了不会改、不可审计不可控语义层 LLM指标口径质量高额外基建重指标体系维护成本高Agent 工具调用本文方案多步推理、失败可修正、步骤可审计对工程与安全设计要求高选型结论我比较坚持取数在本质上不是一个“单次生成”问题而是一个“多步操作”问题。它天然适合 Agent 的循环结构先看库、再锁表、再确认条件、再生成 SQL、遇到报错修正参数、最后把结果整理成业务语言。工具调用让模型行为变成可见、可限、可回退这一点在涉及企业数据时非常重要。1.2 一个案例看清 Agent 与一次性 Prompt 的差异举个例子。业务运营想要“华东区三月销售额”。传统方式是她提工单你写 SQL她等一天。换成 Agent 方式链路是这样的第一轮模型调用 get_table_list发现有几张可能的销售表第二轮对候选表调用 get_table_schema看字段名和注释发现 region、sale_date、amount 这些字段符合第三轮模型生成 SQL这条 SQL 穿过 validator被强制加上 LIMIT且只允许 SELECT第四轮执行结果返回模型把它转成简要结论“华东区三月销售额是 xxx数据来自订单表已过滤退款订单”。这个过程中用户只做了一件事——提问而模型做了完整的信息收集与 SQL 生成工作。一次 Prompt 无法在不知道表名的情况下写出可用 SQLAgent 可以“先看再写”。企业里的取数需求天然就是这样数据库几十张表模型不可能全部预先知道只有让它自己发现、自己决策才能应对业务方千奇百怪的问法。1.3 企业级“安全可控”不是口号而是三层防线很多人看到 Text-to-SQL 的 Demo 觉得“这有什么难的”真到企业落地难的不是模型语言能力而是可控。我把“安全可控”拆成了三个层面这也是整个项目里最重要的设计决策。权限层用户能查哪几张表、哪些列一开始就确定。不是靠提示词约定而是在执行工具里做白名单校验。执行层数据库连接使用只读账号SQL 只能是 SELECT强制 LIMIT敏感列自动脱敏。审计层每一次工具调用、每条 SQL、每条返回结果都要落库。出了数据问题能倒查是哪个用户、哪个会话、哪一条指令产生。这三层缺一层都不算企业级。我早期在提示词里写过“不要访问某张表”结果用户通过诱导提问直接绕过了。提示词只是建议执行层校验才是底线。2. 核心架构与关键设计2.1 从下到上看系统架构整体架构我把它分成三层五类组件。底层是数据源包括数仓和 OLTP 库统一通过只读账号连接中间层是工具库包括 Schema 发现工具、SQL 执行工具、指标口径查询工具上层是 Agent Runtime负责对话循环、工具调度、会话记忆和上下文摘要。外部接入层可以是 Web 页面、企业微信机器人或者一个简单的 OpenAPI本文示例用 HTTP 接口演示。这里面有个设计原则值得单独强调上层 Agent 不直接持有数据库账号信息所有数据访问必须通过工具函数工具函数内部完成连接池管理、权限校验和审计日志。这个强制约定让后续所有安全策略有唯一落脚点不会出现某个角落绕过校验的情况。我见过不少团队把 SQL 执行逻辑写在 Agent 代码里模型可以直接拼字符串调用那基本等于给业务方开了一个裸 SQL 查询接口跟“安全可控”沾不上边。2.2 数据源访问层与 Schema 懒加载数据源连接用的是 psycopg2 连接池。Host、密码放在环境变量或者密钥管理服务里不写进代码。一开始我图省事每个工具调用都新建连接结果并发一高连接就被打满数据库层报了太多连接错误。后来改成连接池每个 Worker 持有一个 PoolAgent 循环内的多次工具调用复用同一连接。有一个坑提醒一下不能用全局单例连接。Agent 循环和多个会话并发时全局连接会交错上一次查询的游标状态可能污染下一次结果出现“A 用户问华东区结果返回了华南区数据”这种严重串台问题。连接池为每个会话分配独立连接从根上避免了这类故障。Schema 发现的策略也值得说道说道。不要每次提问都全量扫描成本高且结果太大。我的做法是启动时做一次全量表清单扫描把表名、注释、行数估算放到内存缓存里当 Agent 认为某张表相关时再通过工具获取该表的完整字段清单与注释。这套“懒加载”机制既保证模型能拿到最新 schema又不会把上下文撑爆。2.3 会话记忆与上下文管理策略取数 Agent 的对话往往是多轮的用户经常会说“上一条不对把华东区改成华南区再看看”。这要求 Agent 具备短期记忆但 LLM 上下文窗口是有限资源企业场景下不能无限堆积。我采用分层记忆方案最近 N 轮对话原样保留供模型参考当前上下文超过 N 轮后用摘要机制把早期对话压缩成一段指令摘要维持上下文一致性工具调用结果里的大表格只取摘要不直接丢给模型。特别提醒一个我在实测中反复遇到的问题不要让前一问答出的数据结果长期留在上下文里。模型会在下一轮对话中误把“结果里的数字”当年份字段或者把结果表格当数据库表来引用。工具结果默认只保留当轮后面再提到时模型应重新调用工具获取而不是依赖历史内容。3. 实操过程与核心环节实现3.1 基于 OpenAI Agents SDK 定义分析师 Agent先交代技术栈Python 3.11openai-agents 官方 Agents SDK底层走 Responses APISQL 解析用 sqlglot。这是目前官方维护比较活跃的 Agent 框架定义 Agent、绑定工具、注册指令都比较直白。pip install openai-agents sqlglot sqlparse psycopg2-binary初始化 Agent 的代码长这样import os from agents import Agent, Runner, function_tool, RunConfig, set_default_openai_key set_default_openai_key(os.environ[OPENAI_API_KEY]) agent Agent( nameData Analyst Agent, instructions( 你是一名企业数据分析师用户会提出取数需求。 你必须按以下流程工作先调用 get_table_list 发现可用表 再调用 get_table_schema 获取相关表结构 最后编写合法的 SQL 查询并通过 run_sql 执行。 执行结果返回后用通俗语言解释结果不要只丢一张表。 ), tools[get_table_list, get_table_schema, run_sql], )Instructions 不需要写太长关键是让模型“按流程走”。我发现把“禁止做什么”写太多反而让模型畏手畏脚更重要的是告诉它“先怎么做、再怎么做”。安全限制放在工具层强制执行而不是靠提示词约束。3.2 核心工具函数取数链路的三板斧第一个工具是 get_table_list返回数据库里当前用户已授权的表列表以及注释。注意这里返回的不是全库表清单而是权限过滤后的表名。模型不知道无权表的存在这是最干净的权限隔离。function_tool def get_table_list() - str: 返回当前用户可访问的表清单包含表名、表注释、估算行数。 当用户提问涉及某个业务指标但你不确定数据在哪张表时调用该工具。 tables list_auth_tables(current_user()) return \n.join(f{t.name} | {t.comment} | ~{t.rows} rows for t in tables)第二个工具是 get_table_schema按表名返回该表的字段、类型、注释。这里做了压缩字段清单以文本形式返回模型读起来效率更高。第三个工具 run_sql 是重点。它在内部做四件事调用 SQL 静态检查器校验权限白名单执行查询把结果转成限制行数的 Markdown 表格。返回给模型的内容既有列头又有前 50 行足够它做解读又不会把上下文撑爆。关于工具库的 docstring有个细节容易被忽略docstring 不是写给人看的是写给模型看的。避免在 docstring 里写“如果……此类 SQL 无法执行”这样的负面描述模型对负面指令的遵循能力不稳定不如正面引导“在编写 SQL 前先调用 get_table_schema”。3.3 SQL 安全校验器与权限矩阵的实现安全方案的核心是校验函数。这里给出简化模式import sqlglot from sqlglot import exp def validate_sql(user_identity: str, query: str) - str: 校验SQL并返回带LIMIT的等价SQL任何违规直接抛异常。 parsed sqlglot.parse_one(query, readpostgres) if parsed is None or not isinstance(parsed, exp.Select): raise PermissionError(仅允许 SELECT 查询) tables set() for node in parsed.find_all(exp.Table): tables.add(node.name) allowed get_allowed_tables(user_identity) forbidden tables - allowed if forbidden: raise PermissionError(f无权访问表: {, .join(sorted(forbidden))}) if parsed.args.get(limit) is None: parsed parsed.limit(100) return parsed.sql(dialectpostgres)强调为什么用 AST 解析而不是正则模型完全可能写出带子查询、JOIN、WITH 的表引用。正则匹配外层 FROM 字段容易漏掉子查询里的越权表而 AST 能稳定枚举所有表引用。我在内部测试时遇到过一次正则方案导致的“绕过”模型把无权限表写在子查询里正则没匹配到SQL 直接被放行。从那以后我坚持所有 SQL 都走 AST 解析。敏感列脱敏也可以在 AST 层处理。在 SELECT 列表里检查列是否属于敏感列集合如果是就替换成脱敏表达式。比如手机号统一替换成concat(left(phone, 3), ****, right(phone, 4))这种形式。业务需要查列表可以但每个人拿到的都是遮蔽版本。3.4 端到端验证从自然语言到安全取数我用一个完整测试案例来说明整条链路。用户提问“请对比华东区三月和四月的销售额按产品线拆分”。Agent 运行路径大概是这样的调用 get_table_list返回业务表清单锁定 sales_order 表调用 get_table_schema拿到 product_line、region、sale_date、amount 等字段生成 SQLSELECT product_line, SUM(amount) AS total_amount FROM sales_order WHERE region 华东区 AND sale_date 2025-03-01 AND sale_date 2025-05-01 GROUP BY product_line ORDER BY product_linerun_sql 内部 validator 检查表权限、强制 SELECT、附加 LIMIT执行查询返回带列名的结果表格模型解读“华东区三月与四月销售额对比如下数据来自 sales_order 主表已剔除测试数据”并附表格。这个流程中模型没有直接触碰数据库账号也没有任何跳过校验的途径因为校验发生在工具内部模型无法绕过——除非它不调用工具但不调用工具就拿不到数据。这就是把安全底线设计在数据路径而不是模型行为上的价值。4. 常见问题与排查技巧实录4.1 模型编造不存在的表名或列名怎么办这是 Text-to-SQL 最常见的翻车点。第一次试跑时用户问“退款率”模型不看 schema 直接写了一个 refund_rate 列数据库里根本没有。根因是模型觉得“这个字段听起来就该存在”。解决办法是强制工具流程instructions 里写清楚“不允许在未调用 get_table_schema 的前提下写 SQL”同时 get_table_schema 返回的信息要足够详尽列注释直接告诉模型这是“退款占比范围 0 到 1”。如果遇到列名和业务词不一致我还加了一个“指标字典”工具让模型先查指标描述再写 SQL效果立竿见影。4.2 多轮对话后的上下文污染典型现象是第一轮查询返回了一个结果表第二轮用户说“北方区同样看一下”模型直接参考第一轮的表结构甚至把结果里的日期当字段名生成“WHERE 2024-03-01 ...”这种废 SQL。排查后发现是工具返回结果被保留在上下文中模型误当成可查询对象。我调整了记忆策略工具的大结果只保留当轮不写入历史上下文模型后续需要时重新触发 schema 发现。这个调整直接减少了大半错误 SQL。4.3 Agent 陷入死循环或反复执行同一个坏 SQLAgent 循环里最让人头疼的是同一个 SQL 报错三次还继续重试。openai-agents Runner 支持 max_turns 参数建议设成 8 到 12。但更关键的是校验器返回的报错信息要“像人一样指出问题在哪”。表不存在时返回“表 employee_salary 不存在可用表有订单表、客户表、退款表”模型看到提示后往往会修正而不是死循环。如果返回统一的“权限不足”模型无从猜起只能一遍遍尝试。4.4 并发、性能与成本问题企业内部署一个 Agent 实例扛不住几十个用户同时问。主要瓶颈有三个LLM 调用频率、数据库连接和上下文 token。连接用连接池解决LLM 调用频率一方面靠队列限流另一方面对简单查询做模型路由——识别用户问的是“昨天订单量多少”这种低复杂度请求用便宜快模型复杂多表关联才喂给强模型。成本上我在每个会话里维护了一个 Token 消耗计数器为审计与预算控制留数据。常用问题整理成速查表现象根因解决方案模型写不存在的列没看 schema 就写 SQL强制先调 get_table_schema 指标字典上下文被结果污染历史工具结果长期保留工具结果只保留当轮摘要化SQL 越权访问权限校验缺失执行层 AST 白名单校验并发连接打满连接不复用连接池 每会话独立 Runner无限重试坏 SQL报错信息不明确返回具体报错 max_turns 限制5. 实测总结与演进方向记录5.1 项目上线后的真实经验项目上线跑了两个月我自己的心态从一开始追求“回答准确率”逐渐转到“失败可审计”。我统计过样本前两周成功率大约 72%大部分失败集中在多轮混淆和字段识别补上指标字典和强制 schema 流程后成功率升到 90% 左右。但即使失败样本也能在审计日志里找出是哪一步逻辑错了这比之前黑盒式的“AI 不好用”体验强太多。真正难啃的还是业务口径。比如“销售额”这个词销售部门要含税已发货金额财务部门要按发票开票日不含税金额两个口径对应不同表或不同过滤条件。Agent 再聪明如果表注释里不写清楚口径它也会随机猜测。后来我在 Schema 工具返回里增加“口径说明”字段让数据团队维护表级和字段级的口径标签模型的准确率才明显提升。5.2 从取数入口到数据服务的演进这个方案目前的定位是“取数入口”后续有几个自然的扩展方向。第一把 Agent 接进企业 IM 机器人业务方在群里直接提问结果推送到群里比单独开一个平台更轻量。第二让它定期执行定时巡检发现指标异常自动汇总成日报或告警这就从一个被动应答工具变成主动监控助手。第三引入多数据源联邦让 Agent 在数仓、业务库、外部 Excel 之间做统一查询。第四增加“导出自助审批”流程当业务方需要超过 100 行的原始明细时Agent 生成导出任务转人工审批而不是直接放行。最后再分享一个小技巧测试 Agent 时不要把注意力全放在成功路径上多设计“诱导越权”和“误导性问题”这类用例。我就是在测试用户问“你看得到员工薪资表吗”这个问题时才发现提示词层的权限约束根本挡不住模型的联想能力。安全性的验收标准不是常规场景不出错而是攻击性场景不失控。