基于OpenAI Agents SDK构建安全可控的自然语言取数Agent实战

先聊清楚企业里的取数到底难在哪周四下午四点业务同学在群里问这个月华东区分渠道GMV能拉一下吗要按天、带上转化率。数据团队的同事看了一眼排期回了一个明天下午给你。这种对话每天都在发生大家对取数要排队这件事已经习以为常了。可真正的问题不是排队而是取数链路里那一堆隐性消耗——需求理解偏差、口径反复确认、SQL写一半发现表结构变了、跑完了发现数据对不上。一次取数来回三四轮半天就过去了。我在过去一年里陆续接了好几个类似的内部数据平台项目最后的解法都指向同一个方向用自然语言对接数据源把提需求-写SQL-校验-交付这条链路压缩成一句话的事。但直接给模型一个数据库连接串、让它随便查显然不行。生产环境里权限、脱敏、限流、审计哪一环漏了都是事故。于是我在几个项目里基于 OpenAI 官方 Agents SDKopenai-agents搭了一套数据分析师 Agent把取数能力封装成受控工具再套上输入输出双重护栏配合最小权限和审计日志让业务同学真的能用自然语言安全取数。这篇文章就把这套方案的完整落地过程拆开讲为什么选 Agents SDK 而不是裸调 Chat Completions工具和护栏怎么设计数据库权限怎么收敛上线之后又踩了哪些坑。适合正在做数据平台、想给团队引入自然语言取数能力、但又担心安全问题的朋友参考。1. 为什么是 Agent自然语言取数不是让模型写SQL这么简单1.1 单次生成 SQL 解决不了多轮问答如果你只是想做一个demo让模型把上个月销售额翻译成一条SQL那确实用 Chat Completions 加一个 function call 就够了。但取数场景天然是多轮的用户先说看下华东区哦不对是华南区按周聚合吧去掉退货订单再算一遍对话里带着上下文修正和口径调整。这要求系统能记住之前的查询状态、能根据用户反馈迭代SQL而不是每次从头开始。Agent 的出现就是为了解决这种需要调用工具、需要多轮迭代、需要根据工具结果调整下一步的任务。OpenAI Agents SDK 把 Agent、工具、护栏、会话这些概念做成了现成的运行时我不用自己写状态机和工具调度只需要定义好这个 Agent 能做什么、不能做什么。1.2 取数场景的核心矛盾模型能力在涨信任没跟上自然语言取数的技术难点从来不在翻译SQL这一步——现在主流模型的SQL生成能力已经很强了。难点在于信任。想象一下你给业务同学一个对话框他输入查一下所有客户的手机号你是让模型生成一条SELECT phone FROM customers然后原样返回吗当然不行。数据仓库里有权限体系、有加密字段、有按团队划分的数据域这些东西必须由系统层面去强制执行不能指望模型自觉。1.3 Agent 在取数链路里到底扮演什么角色我的理解是Agent 在这里是一个受控的翻译层 执行器。它的职责是把用户自然语言请求翻译成结构化的查询意图把查询意图映射到数据库工具工具内部做权限、脱敏、限流把数据库返回的结果整理成业务能看懂的答案遇到不确定的、超出权限的、有风险的请求停下来问人换句话说Agent 不负责任何数据库安全策略它只负责在策略允许的范围内帮用户把数据拿出来。安全策略全部下沉到工具层和基础设施层。这个设计原则我后面会反复强调。2. OpenAI Agents API 拆解为什么它适合做取数代理2.1 四个核心抽象Agent、Tool、Guardrail、Session先说清楚这里说的 OpenAI Agents API 指的是官方推出的 Agents SDK不是之前的 Swarm 实验项目也不是单纯调用 Chat Completions 的 function calling。它在工程上解决了我最头疼的几个问题。Agent 是一个带指令、带工具、带模型配置的智能体单元。比如我可以定义一个取数 Agent它的指令里写明你是一名数据分析师助手只能使用提供的取数工具不允许猜测不存在的表名。Tool 就是函数。SDK 提供了function_tool装饰器把一个普通 Python 函数变成模型可调用的工具。函数内部我可以在执行 SQL 前做权限校验、参数白名单、结果脱敏这些逻辑模型完全碰不到只看到最终返回。Guardrail 是输入/输出侧的拦截器。模型收到的用户问题、模型即将输出的内容都会先过一层校验逻辑。在校验里我可以识别 SQL 注入尝试、敏感信息探测、超范围查询等命中就直接拦截。Session 是会话状态管理。多轮取数对话里的上下文、之前查过哪些表、用户偏好口径都存在 Session 里不需要我自己维护一套外置记忆。2.2 Agent 运行循环到底在做什么很多刚开始接触 Agent 的朋友会疑惑Agent 和普通 API 调用到底差在哪我觉得核心差异在于执行循环。普通 API 调用是一次性的输入-输出而 Agent 的运行循环是模型看到用户消息和系统指令决定是直接回答还是调用某个工具如果调用工具运行时把函数执行结果返回给模型模型根据结果继续推理可能再调下一个工具也可能给出最终回答整个过程持续到模型认为任务完成或触发停止条件在这个循环里工具是模型接触外部世界的唯一通道。这就是安全设计的抓手数据库连接串永远不交给模型模型只能调用我定义好的那几个函数每个函数内部可以做任意控制。这个唯一通道原则是整个方案安全性的基石。2.3 和直接裸调 LLM API 的本质差距项目早期我们用过裸调方式把表结构拼进 system prompt让模型返回 SQL然后后端直接执行。问题很快就暴露了。第一模型经常生成不存在的列名或表名因为整个 schema 太大prompt 根本塞不下塞下了也容易上下文混淆。第二多轮对话的状态全要自己管理用户说按周聚合你得自己记住上一轮的SQL。第三也是最致命的——模型没有执行结果反馈这个闭环。它生成了一条 SQL但执行成功没有、返回多少行、有没有报错它完全不知道自然也不会修正。改用 Agents SDK 之后这些变成了标准能力。模型生成 SQL → 工具执行 → 执行结果包括报错回传给模型 → 模型根据报错自动修复 SQL这个闭环让成功率有了质的提升。我后面统计过加了闭环反馈之后SQL 首轮执行成功率从 60% 出头涨到了 85% 以上。3. 从零搭建数据分析 Agent工程化落地全流程3.1 环境准备与最小工程骨架先搭一个最小可用的工程后面再逐步加安全控制。环境上我需要准备Python 3.11安装openai-agents包一个 OpenAI API Key配成环境变量一个可用的测试数据库建议先用 SQLite 或本地 PostgreSQL别一上来就接生产库pip install openai-agents python-dotenv工程结构我会按功能拆成四个模块避免所有逻辑堆在一个文件里analyst_agent/ ├── main.py # Agent 编排与入口 ├── tools/ │ ├── db_query.py # 取数工具含权限与脱敏 │ └── meta.py # 元数据查询工具 ├── guardrails/ │ ├── input_guard.py # 输入侧护栏 │ └── output_guard.py # 输出侧护栏 ├── audit/ │ └── audit_logger.py # 审计日志 └── config.py # 数据库连接、Agent 指令、模型配置提示openai-agents的小版本迭代很快具体导入路径和装饰器写法可能随版本微调建议安装后先查看对应版本文档。我下面的代码基于 2025 年上半年的稳定版写法。3.2 工具集设计把数据库能力封装为受控函数这是整个方案里我建议你花最多心思的地方。工具的设计决定了 Agent 的能力边界。对于取数场景我不会只暴露一个执行任意 SQL的工具。那样等于把数据库整个交给了模型太危险。我会拆成几个语义更明确、参数更可控的工具query_table(table, columns, filters, group_by, order_by, limit)— 常规查询describe_table(table)— 查看指定表的元数据list_tables(domain)— 查看某个数据域下有哪些表explain_query(table, query_plan_hint)— 在只读模式下估算成本不做实际执行每个工具的入参都经过严格定义模型可以自由组合但不能跳出这些组合方式。举个例子如果模型想算华东区最近30天GMV它只能这样组合工具调用query_table( tableorder_summary_daily, columns[date, channel, gmv], filters{region: 华东区, date: 最近30天}, group_by[date, channel], limit500 )而不是扔过来一条SELECT * FROM orders WHERE region华东 AND secret_column IS NOT NULL。这个设计的好处是模型没有自由发挥的空间所有查询都结构化我可以在工具函数内部做参数校验、枚举校验、权限判断比解析自然语言生成的 SQL 可靠得多。3.3 SQL 生成与执行的安全设计工具函数内部最终还是要生成 SQL 并执行。这层就是安全博弈最激烈的地方。我总结了几条必须落实的硬规则。第一数据库账号最小权限。给 Agent 专用的数据库账号只赋予必要表的 SELECT 权限不授予任何 DML、DDL、TRUNCATE 权限。连接串用环境变量或密钥管理系统下发不写死在代码里。第二强制只读事务。每次连接开启只读事务并在会话层面设置执行超时。比如 PostgreSQL 里SET transaction_read_only on; SET statement_timeout 30000; -- 30秒 BEGIN; -- 执行查询 COMMIT;这样即使工具内部逻辑出现漏洞被注入了一段危险 SQL也只读事务和超时机制能兜住把爆炸半径压到最小。第三LIMIT 和数据量上限。工具函数强制附加LIMIT防止一次查询拉几千万行把数据库内存打爆。我在工具函数的 docstring 里明确告诉模型默认最多返回 1000 行超过这个规模的结果需要聚合后再返回。function_tool def query_table( table: str, columns: list[str], filters: Optional[dict] None, group_by: Optional[list[str]] None, order_by: Optional[str] None, limit: int 1000, ) - dict: 查询业务表数据。默认返回最多1000行。 如果查询结果预期超过1000行必须先聚合。 # 参数白名单校验 table validate_table(table) columns validate_columns(table, columns) # 强制 limit 上限 safe_limit min(limit, 1000) # 只读连接 超时 with get_readonly_conn() as conn: with conn.cursor() as cur: cur.execute(build_sql(...), params) rows cur.fetchmany(safe_limit 1) return {rows: rows[:safe_limit], truncated: len(rows) safe_limit}第四参数化查询。动态条件一律通过参数化方式拼接绝不直接格式化进 SQL 字符串。这一步是防 SQL 注入的最后一道防线。虽然 Agent 工具的参数是结构化的但用户输入可能经过模型转述携带恶意内容所以工具内部所有条件值都必须走参数绑定。3.4 输入/输出 Guardrails 配置示例Guardrail 是 Agents SDK 里一个很容易被低估的功能。它本质上是在模型看到用户消息之前和模型输出最终回复之后各加一道校验钩子。我配置了两道输入护栏和一道输出护栏。输入护栏一敏感词和注入拦截。检测用户请求中是否包含忽略指令绕过限制删除表所有手机号全部用户隐私这类内容。命中就直接阻断不给模型处理的机会。from openai_agents.guardrails import InputGuardrail, GuardrailFunctionOutput class BlockSensitiveInput(InputGuardrail): async def check(self, ctx, agent, query): sensitive_patterns [忽略之前的指令, 删除, drop table, 所有手机号, 全部身份证] hit any(p in query for p in sensitive_patterns) info {blocked: hit, reason: 敏感内容} return GuardrailFunctionOutput(output_infoinfo, tripwire_triggeredhit)输入护栏二权限域校验。把当前登录用户所属的数据域比如零售事业部注入查询上下文工具内部做行级权限过滤护栏只做第一层粗筛工具层才是真正的强制层。输出护栏的作用是防止模型在回复里泄露不该泄露的信息比如把工具返回的脱敏字段又还原出来了。我在输出护栏里检查回复文本中是否出现手机号、身份证号的完整格式命中就改为脱敏展示。3.5 一个完整取数对话的代码演示把 Agent 组装起来核心代码如下from openai_agents import Agent, Runner, function_tool analyst_agent Agent( nameDataAnalyst, instructions 你是一名严谨的数据分析师助手只能通过提供的工具取数。 规则 1. 只能查询工具元数据中存在的表和字段禁止猜测表名/列名。 2. 涉及敏感字段手机号、身份证号、收入明细必须脱敏后展示。 3. 查询结果超过1000行时必须说明数据量较大并尝试聚合。 4. 用户要求的取数范围超出工具权限时明确拒绝并说明原因。 5. 所有回答必须基于工具返回的真实数据禁止编造数字。 , tools[query_table, describe_table, list_tables], input_guardrails[BlockSensitiveInput()], output_guardrails[MaskSensitiveOutput()], ) result await Runner.run(analyst_agent, 帮我查一下华东区最近30天各渠道的GMV按周聚合) print(result.final_output)运行这个 Agent 时内部发生的事情大致是输入护栏检查用户请求未命中敏感词放行模型读到系统指令决定第一步调用list_tables看看有哪些表可用工具返回表清单模型选择order_summary_daily模型再调用describe_table确认字段含义避免列名幻觉最终调用query_table工具内部完成参数校验、只读查询、聚合返回模型基于返回的结果组织回答经过输出护栏脱敏检查后输出这个完整链路看起来没什么高明之处但每一步都在把不可控的模型行为往可控的工程边界里收。4. 安全可控的设计细节权限、脱敏、审计与人工兜底4.1 最小权限与网络隔离数据平台的安全体系不能只靠 Agent 这层的代码防御基础设施层同样要配合。我的做法是三层收敛。第一层是网络。Agent 服务和数据仓库之间走内网或私有网络数据库端口不对公网开放。拿到 Agent 服务访问权的攻击者最多能通过我暴露的工具函数发出结构化的查询请求拿不到数据库直连地址。第二层是账号。为 Agent 单独创建一个数据库用户只赋必要表的 SELECT 权限定期轮换密码。绝不复用业务系统的账号避免一个 Agent 漏洞把整个业务库带下水。第三层是查询成本控制。不光要做超时还要防慢查询拖垮库。我会在工具函数里做两层限制一是statement_timeout二是 EXPLAIN 估算。复杂大表查询先走 EXPLAIN 估算扫描行数超过阈值就直接拒绝要求用户加条件或改聚合粒度。这里有一个值得单独强调的细节行级权限必须由工具层强制而不是由模型自觉。比如用户所属部门是零售组那么无论模型生成的查询条件里写没写WHERE dept_id retail工具函数内部都要强制拼接AND dept_id current_user_dept_id。这样即使模型被诱导、被注入也无法跨部门拉数据。4.2 Schema 暴露面控制LLM 的 SQL 生成能力很强但前提是它得知道表长什么样。生产库动辄几百张表、上万列全量暴露给模型不仅没必要还会提高幻觉概率。我做的是一份面向模型的元数据白名单。在describe_table和list_tables工具内部我配置了一个元数据注册表只包含允许业务自助取数的表和字段并且给每个重要字段加上业务注释。如果是大宽表只暴露常用的核心列。模型看到的元数据像这样表 order_summary_daily - date (日期格式YYYY-MM-DD) - channel (渠道线上/线下/分销) - region (大区华东/华南/华北/...) - gmv (成交金额单位元) - order_cnt (订单数) - refund_amt (退款金额)注意元数据里不包含任何一行真实数据。模型在describe_table返回结果之前看到的只有表注释、字段注释、字段类型。这样既能指导它生成合理的查询又不会让它从系统 prompt 里学到敏感数据。4.3 敏感字段脱敏与行级权限脱敏要分两层理解。一层是字段脱敏手机号、身份证号、工资这类字段在工具返回结果前就做了掩码处理模型拿到的已经是138****1234。另一层是内容脱敏模型在回答中提到的数字、金额如果是聚合结果还要检查是否满足最小可区分度——比如用户只要一个人的工资结果直接暴露了这就不行。我会在输出护栏里对这种明细泄露做拦截。具体到工具层脱敏逻辑可以这样写def mask_sensitive(value, field_type): if field_type phone: return value[:3] **** value[-4:] if field_type id_card: return value[:6] ******** value[-4:] return ***至于行级权限我前面提过是工具层强制拼接过滤条件。还有一层容易被忽略的是结果集行数限制。业务同学随口一句把全公司所有客户都导出来这种查询即使有行级权限也可能拉几十万行。工具函数在返回前检查row_count threshold时直接不返回明细而是返回一条提示结果集过大请缩小范围或改为聚合查询。4.4 审计日志每一次取数都可回放安全审计是这个方案里最能体现企业级的一环。自然语言取数让数据获取变得太容易了如果没有审计出事之后根本无法追溯谁在什么时间拿了什么数据。我在 Agent 的工具调用链路里加了一个审计中间件记录以下字段字段记录内容请求ID每次对话会话生成唯一ID用户标识企业内的用户唯一标识归属组织用户所属部门/数据域原始问题用户输入的自然语言原文模型生成的SQL工具实际执行的SQL语句工具名称调用了哪个取数工具返回行数实际返回的数据行数时间戳请求开始和结束时间结果摘要是否截断、是否脱敏、是否转人工拒绝原因如果被拦截/拒绝记录原因审计日志写到一个独立的日志存储里与业务数据库分离只允许数据安全团队访问。同时我会把原始问题和生成的SQL做一份静默哈希用于事后比对完整性防止日志被篡改后无据可查。4.5 风险操作转人工Agent 处理不了的事别硬扛再强的护栏也有边界。有些请求你希望 Agent 明确承认我做不了然后转到人工流程而不是硬着头皮给一个可能出错的答案。需要转人工的情况大概有三类用户要求的取数范围明显超出其数据域但看起来有正当业务需求查询复杂度极高EXPLAIN 估算要扫全表需要数据团队评估涉及数据导出、归档、跨系统传输等 Agent 只读能力之外的操作在这些情况下Agent 不是拒绝后结束对话而是生成一个标准的人工取数工单把需求描述、期望格式、截止时间自动填入工单系统并回复用户已提交数据团队处理工单号是 XXX。这个机制的价值是自然语言取数覆盖了 80% 的常规高频需求剩下 20% 的低频复杂需求仍然走人工流程两边都不耽误。业务同学不会因为 Agent 做不到而骂娘数据团队也能集中精力处理真正有难度的问题。5. 上线前后的踩坑记录与优化经验5.1 并发不是加了线程就完事Agent 本身是异步的能轻松扛住高并发。但数据库未必。上线初期我们没做流量控制业务推广后查询量直接翻了十倍结果是把几个报表数据库的 IO 打满了业务系统跟着遭殃。后来我加了两层保护。第一层是Agent 服务侧的信号量限流限制同时执行的取数工具调用数量比如控制在 20 个并发其余请求排队。第二层是数据库侧的连接池和资源组隔离。给 Agent 用的数据库连接走独立的连接池并且放在独立的 PostgreSQL resource group 里限制它的 CPU 和 IO 占比。这样即使 Agent 流量再大也不会挤爆生产主库的资源故障半径被物理隔离。注意并发上限千万不要只靠数据库的max_connections兜底。数据库连接被打满的时候整个库都处于不可用状态那画面太美不想再看。5.2 上下文和记忆Session 的正确用法多轮取数对话中用户经常在后面补充条件啊对了不要包含退货的订单渠道只算线上和线下分销不算。如果 Agent 没有会话记忆第二次请求就只能基于当前这一轮的输入重新理解很可能会丢掉前面的上下文。Agents SDK 的 Session 概念在这里很有用。我把每个业务用户的取数对话绑定到一个持久化 Session 上Session 里存着此前的对话摘要和关键口径比如用户当前默认的时间范围是最近30天表 order_summary_daily 已确认口径。这样下一轮对话里模型能自动带上之前的约束。但踩过一个坑Session 内容如果太长会把可用上下文窗口挤占掉影响工具调用质量。我现在会在每轮取数结束后生成一段 200 字以内的结构化摘要写入 Session而不是把完整的工具调用历史都堆进去。摘要格式类似当前口径时间范围2025-10-01至今渠道[线上, 线下]区域华东区 已确认表order_summary_daily 待处理用户要求按周聚合未确认是否排除退款订单5.3 查询失败与幻觉的常见模式上线后我统计过一个月的工具调用日志总结了几个最常见的失败模式分享出来帮大家避坑。列名幻觉。模型在describe_table返回的字段里没有找到想要的数据时偶尔会脑补一个近义列名出来。比如元数据里只有gmv模型却调用了total_sales_amount。这类问题的解法是在工具函数里做严格校验调用的列名必须在元数据白名单内否则立刻报错把错误信息回传给模型让它改用正确列名。聚合口径错误。模型生成的 SQL 里SUM 和 COUNT 用混是常态尤其是在字段含义不清晰的时候。我会在元数据注释里写清楚gmv 是成交金额不包含退款order_cnt 是订单数一单多商品也算 1。同时工具层对聚合操作做了约束不允许SUM(ratio)这种无意义的聚合发现就返回校验错误。查询范围失控。用户说近一年所有数据模型如果真的不加过滤条件去查一张大表EXPLAIN 估算就会拦截。这里建议在工具内部不仅强制 limit还强制必须有合理的过滤条件比如日期范围、区域、渠道至少有一个否则直接返回提示让模型补充条件。自嗨式回答。这是最隐蔽的坑——模型在工具返回错误或者没有返回数据时为了给用户一个交代会基于训练知识编一个看起来合理的数字。我的应对措施是系统指令里反复强调回答必须基于工具返回结果工具未返回数据时只能回复未知禁止编造并且把这一条写进输出护栏检查最终回答中是否包含未在工具结果中出现过的具体数值。5.4 性能优化缓存、预计算的取舍最后聊聊性能优化。自然语言取数比固定报表多了一层模型推理的开销首响时间通常在 5~15 秒这个体验对业务同学来说是可以接受的但频繁重复查询还是很浪费资源。我做了两级缓存。第一级是语义缓存把用户请求做了标准化处理后计算 embedding 相似度命中相似度 95% 以上的历史请求且历史数据生成时间在 10 分钟以内直接返回之前的查询结果。第二级是结果缓存相同 hash 的 SQL 在数据变更前直接复用结果集。更重要的优化思路是预计算。业务同学最喜欢问的昨日GMV本月新增用户近7天转化率这类固定口径问题可以提前跑成每日快照表让 Agent 直接查快照而不是查底表明细。这样既快又不会冲击生产库。用下来以后Agent 查询的 P95 响应时间从原来的 12 秒降到了 4 秒左右而且对底层表几乎没有扫描压力。最后说一个我在实际维护中的体会这套方案最大的价值不是让模型自己写 SQL而是把自然语言取数从一个不靠谱的 demo变成了可审计、可回滚、可控制的企业级能力。安全不是某个单独模块做得多强而是从数据库权限、工具函数、护栏、审计、人工兜底形成的一整条链条。任何一环失效时其他环节仍然能把风险挡住。如果你也在做类似的事我建议先别急着上大模型先把取数场景的企业规则梳理清楚再用 Agent 把规则落成代码。规则越清楚Agent 跑得越稳。