我接到这个标题的时候，第一反应是：又一个想用大模型做取数的项目。但真正动手做过企业级数据平台的都知道，卡住你的从来不是“大模型能不能写对SQL”这件事，而是“写完SQL之后谁敢直接让它执行”。这篇文章我结合最近几个月在真实业务环境里落地 OpenAI Agents-API 数据分析师 Agent 的经验，把从方案设计、核心代码、权限隔离到观测运营的整套链路拆开讲清楚。内容会偏实操，涉及不少我踩过的坑（尤其是安全边界和权限控制的部分），适合正在做 AI Agent 开发、想要把 “自然语言取数” 从 Demo 推向生产环境的大模型开发工程师、数据平台负责人和数据工程师参考。
1. 为什么“LLM 直接写 SQL”做不了企业级取数——先想清楚 Agent 化拆分的本质
很多团队拿到 OpenAI API 之后，第一版“自然语言取数”基本都是同一个套路：把用户问题、表结构拼进 Prompt，让模型输出 SQL，然后丢给数据库执行。Demo 跑得很欢，一旦面对真实业务，问题全冒出来了。我先说三个最痛的点。
第一个是不可信。模型生成的 SQL 本身没有经过校验，表名写错、字段拼错是常态，更麻烦的是它在写复杂 JOIN 和聚合的时候“自信满满地搞错业务口径”。比如一张订单表里含“已取消”状态，模型可能忘记加过滤条件，导致业绩数字虚高。这种错误在取数场景里不是小瑕疵，而是直接让决策链条崩掉。
第二个是不可控。企业数据库不是给你随便玩的。一个没加 WHERE 条件的 SELECT 全表扫描已经算温柔了，万一模型被诱导生成 DELETE 或者 UPDATE，又或者通过 UNION 读到权限外的数据，那就不是事故而是灾难了。很多团队不敢放开这个能力，就是因为执行链路上没有任何安全关卡。
第三个是不可运营。业务方问“上季度华东区销售额前三的品类是什么”，你很难以“我调了大模型”作为交付物。谁问的、什么时候问的、用了哪张表、跑了什么 SQL、花了多少 token，这些东西如果都不可追溯，数据团队就没法对结果负责，也没法做成本核算。
所以要真正解决企业级自然语言取数问题，关键不是让模型“更聪明”，而是把它从“一个生成 SQL 的接口”重新设计成“一个完整的工作流”。“取数”这件事天然适合 Agent 化拆分，原因有三点：
- 它可以被拆解为多个明确子任务：解析意图、生成 SQL、校验语义、执行查询、解释结果；
- 每个子任务都有独立的验证点，而不是“一步到位赌模型输出”；
- 每个子任务都能接入权限控制、审计日志和人工审批，形成安全闭环。
这也是为什么我最终选择了 OpenAI Agents SDK（Agents-API，官方叫法 Agents SDK，PyPI 上是openai-agents包），而不是直接裸调 Chat Completions。这么说吧：裸调 API 是“让一个实习生活泼大胆地干活”， Agents-API 则是“给它一套明确的办事流程、专用工具和纪律手册”。
当然，Agents-API 不是唯一选择，LangGraph、AutoGen 也能做，核心思想相通。但对我这个场景来说，Agents SDK 最务实：它够轻、对 Python 原生支持够好、内置 Agent/Tool/Run 抽象，基本项目管理成本很低。下面我会先把这个框架的关键抽象讲清楚，然后再进入整套取数方案的设计和实现。
2. Agents-API 的核心抽象：Agent、Tool、Run 与循环执行机制
如果之前没接触过 Agents SDK，我先用一个比喻帮大家建立直观印象。你是一个项目经理，手下有一个“数据分析师实习生”（Agent）。它手里有一套专用工具（Tool），比如“读表结构”“查指标口径”“跑SQL”。你给它交代任务（输入 prompt），它自己决定先调哪个工具、按什么顺序调、拿到结果之后下一步干什么，整个过程由调度器（Run）管理，直到它认为任务完成了。
在 Agents-API 里，这三个抽象是整个体系的地基。
Agent是最核心的实体。它本质上是一个由大模型驱动的“可编程执行体”，有自己的名字、指令（instructions）、可用工具（tools）和模型配置。指令就是系统提示词，但它的价值不只是“描述角色”，而是定义了一套工作边界和工作流，比如“你只负责数据查询”“查询前必须先校验表名在白名单内”“任何非 SELECT 语句直接拒绝”。
Tool是 Agent 可以调用的外部能力。在取数场景里，Tool 可以是：
get_allowed_tables()：返回当前用户有权访问的表清单（白名单）get_schema(table_name)：获取指定表的字段、类型和注释validate_sql(sql)：静态检查 SQL 是否合规execute_select(sql)：真正执行查询并返回结构化结果
Run则是一次完整的执行过程。它处理 Agent 与 LLM 之间的多轮循环（LLM 返回意图 → Agent 调工具 → 结果回填给 LLM → LLM 继续决策），还负责把中间过程（Tool 调用、输出结果、运行状态）暴露给开发者做观测。这个机制非常关键，因为 Agent 的一切行为都变成了可追踪的“步骤”，而不是“给个模型一个 prompt 然后祈祷输出对”。
SDK 还提供了生命周期钩子（lifecycle hooks），比如RunStart、RunEnd、FunctionCall等事件，可以在每个阶段插入日志、鉴权、限流逻辑。我强烈建议读一遍官方文档里关于“lifecycle”的部分——企业落地 Agent，绝大多数合规细节都依赖在这些事件里拦截。
从取数方案的视角看，这些抽象帮我解决了一个很重要的问题：以往“自然语言转 SQL”是 AI 平台和人打交道；现在它变成了 Agent 通过工具跟“受控的数据环境”打交道。模型无法直接拼凑数据库连接串，因为它没有这个 Tool；模型不能访问权限外的表，因为 Tool 只返回白名单内的 schema。安全边界从“提示词约束”变成了“工具层强制”。
3. 企业级取数方案的整体架构：两段式执行与四层安全边界
先说整体架构。我没有选择让 Agent 直接连生产数据库读数据，而是分成两个阶段：
- 阶段一：语义理解和 SQL 生成（AI 层，不碰数据库）
- 阶段二：SQL 执行与结果下发（执行层，强校验后执行）
对应到实现上，就是一个主控 Agent 配上多个工具，但其中唯一能碰数据库的execute_select工具被严格保护。架构图用文字描述大概是这个样子的：
用户请求 │ ▼ API 入口（鉴权 + 身份识别） │ ▼ 主控 Agent（Agents-API） ├── 工具：白名单查询（按身份返回可访问表） ├── 工具：Schema 查询（只暴露当前用户可看字段） ├── 工具：口径查询（Meta 表，业务指标定义） ├── 工具：SQL 语法校验 + 静态分析 └── 工具：execute_select（唯一数据库入口，受控执行） │ ▼ 执行层（四层安全检查 → 只读事务 → 行级/列级权限 → 脱敏 → 返回）在代码层面我做了四层安全边界，每一层解决一类问题：
第一层：身份与资源隔离。用户的 API 入口需要携带身份令牌。Agent 启动时根据身份初始化一个AccessContext对象，内含该用户的角色、可访问的表清单、字段级别的可见性和行级过滤条件。越权访问在源头就被掐断了，因为 Agent 的“白名单工具”返回的数据本身就不含无权资源。
第二层：静态 SQL 校验。这是我对抗“模型乱写 SQL”的第一道防线。校验器做的事情包括：
- 只允许
SELECT开头的语句，一律拒绝INSERT/UPDATE/DELETE/ALTER/CREATE/DROP/TRUNCATE/GRANT； - 解析 SQL 中的表名和列名，必须全部来自当前用户的白名单和可见字段集合；
- 阻止多语句拼接（分号截断）、注释混淆、内联脚本等注入模式；
- 限制扫描上限，比如模型生成的 SQL 如果明显缺少 WHERE 条件或包含无 LIMIT 的全属性查询，会直接打回。
第三层：数据库账号沙箱化。这是最关键的一层。千万不要用高权限账号去执行模型生成的 SQL。我在数据库里单独建了一套只读账号，默认Only SELECT、SET SESSION TRANSACTION READ ONLY，并且把账号能看到的表限制到指定 Schema，用数据库层面的权限做好兜底。即使前面两层被绕过，这一层也能挡住非查询操作。
第四层：结果脱敏和行级限制。返回给模型和用户的数据不是“查出什么就给什么”，而是要经过一个脱敏通道。手机号、身份证、邮箱等敏感字段会根据用户角色自动掩码；同时每个查询都会在 SQL 里强制注入行级过滤条件（比如“只看本门店”“只看数据权限范围内的法人实体”），避免越权行数据流出。
从执行链路的角度，四层不是替代关系，而是叠加关系。静态校验挡住大部分模型低级错误；数据库账号沙箱挡住注入和恶意操作；行级过滤挡住水平和垂直越权；脱敏保证即使查到了敏感字段也不能直接落出平台。这套设计跑下来，我最大的体会是：在 Agent 架构里，“安全”不能靠某一个点做得多强，而是要靠各层之间互相兜底。
4. 落地实战：一个带安全边界的数据分析师 Agent 核心代码
代码部分我直接给出一个可运行的最小版本，然后用注释和后面的小节把关键设计讲透。这里的技术栈是 Python 3.11 +openai-agents（0.0.x 系列，具体小版本建议装最新稳定版）。
import json from typing import Any, Optional from pydantic import BaseModel, Field from agents import Agent, Runner, function_tool, RunStart, RunEnd from agents.lifecycle import LifecycleHooks # ---------- 数据模型 ---------- class AccessContext(BaseModel): user_id: str role: str allowed_tables: list[str] = Field(default_factory=list) row_filter: str = "" # 例如 "dim_org_id = 'D10086'" sensitive_fields: list[str] = Field(default_factory=list) class SqlCheckResult(BaseModel): passed: bool reason: str normalized_sql: Optional[str] = None # ---------- 安全校验层 ---------- FORBIDDEN_KEYWORDS = ["INSERT", "UPDATE", "DELETE", "ALTER", "CREATE", "DROP", "TRUNCATE", "GRANT", "REVOKE", "MERGE", "CALL"] def static_sql_check(sql: str, ctx: AccessContext) -> SqlCheckResult: sql_stripped = sql.strip().rstrip(";").strip() if not sql_stripped.upper().startswith("SELECT"): return SqlCheckResult(passed=False, reason="仅允许 SELECT 查询") for kw in FORBIDDEN_KEYWORDS: if re.search(rf"\b{kw}\b", sql_stripped.upper()): return SqlCheckResult(passed=False, reason=f"语句包含禁止关键字 {kw}") # 拒绝分号多语句 if ";" in sql_stripped: return SqlCheckResult(passed=False, reason="不允许多语句拼接") # 表名白名单校验（简化版，生产环境建议用 sqlglot 做 AST 级解析） for table in ctx.allowed_tables: if table.lower() in sql_stripped.lower(): break else: return SqlCheckResult(passed=False, reason="SQL 未引用任何白名单表") return SqlCheckResult(passed=True, reason="OK", normalized_sql=sql_stripped)接下来是核心工具函数。重点说下execute_select：它接受一个context参数，框架里的function_tool会自动把RunContextWrapper里的外部上下文注入进来，这就让我们能够拿到“当前请求属于哪个用户、拥有什么权限”等信息，而不是让模型自己声明身份。这一条对 Agent 开发来说特别重要。
import sqlite3 import re # ---------- 模拟数据层 ---------- # 实际生产环境接入 PostgreSQL/MySQL，这里用 sqlite 演示整体流程 _DB = sqlite3.connect("analytics_demo.db", check_same_thread=False) @function_tool def get_allowed_tables(ctx: RunContextWrapper[AccessContext]) -> str: """返回当前用户可访问的表的清单，用于决定 FROM 子句可以引用的表。""" ctx.context.user_id # 通过 ctx 读取真实身份 return json.dumps(ctx.context.allowed_tables, ensure_ascii=False) @function_tool def get_schema(ctx: RunContextWrapper[AccessContext], table_name: str) -> str: """返回指定表的字段信息。非白名单表返回拒绝信息。""" if table_name not in ctx.context.allowed_tables: return json.dumps({"error": f"无权访问表 {table_name}"}) cur = _DB.execute("PRAGMA table_info({})".format(table_name.replace("`", ""))) cols = [{"name": row[1], "type": row[2]} for row in cur.fetchall()] # 对敏感字段做列级标记 for col in cols: if "phone" in col["name"] or "idcard" in col["name"]: col["sensitive"] = True return json.dumps(cols, ensure_ascii=False) @function_tool def validate_sql(ctx: RunContextWrapper[AccessContext], sql: str) -> str: """对模型生成的 SQL 进行静态安全检查。""" result = static_sql_check(sql, ctx.context) return json.dumps(result.model_dump(), ensure_ascii=False) @function_tool def execute_select(ctx: RunContextWrapper[AccessContext], sql: str) -> str: """受控执行 SELECT 查询。执行前强制校验并注入行级过滤条件。""" # 二次校验：即使模型跳过 validate_sql，这里也要完整检查 check = static_sql_check(sql, ctx.context) if not check.passed: raise Exception(f"SQL 被安全策略拒绝：{check.reason}") # 强制注入行级权限（简化写法：拼装 WHERE 子句） enforced_sql = sql.rstrip().rstrip(";") if ctx.context.row_filter: if "where" in enforced_sql.lower(): enforced_sql += f" AND {ctx.context.row_filter}" else: enforced_sql += f" WHERE {ctx.context.row_filter}" enforced_sql += " LIMIT 200" try: cur = _DB.execute(enforced_sql) rows = cur.fetchall() col_names = [desc[0] for desc in cur.description] # 脱敏处理 result_rows = [] for row in rows: new_row = [] for idx, col in enumerate(col_names): if col in ctx.context.sensitive_fields: v = row[idx] if v and len(str(v)) > 6: new_row.append(str(v)[:3] + "****" + str(v)[-2:]) else: new_row.append("***") else: new_row.append(row[idx]) result_rows.append(new_row) return json.dumps({"columns": col_names, "rows": result_rows}, ensure_ascii=False, default=str) except Exception as e: return json.dumps({"error": str(e)}, ensure_ascii=False)主控 Agent 定义如下。指令部分不只是“你是数据分析师”，而是明确地写清了任务流程、安全纪律和输出要求。在 Agent 提示词里写清楚“使用工具的先后顺序”，对减少无效调用、降低 token 消耗很有帮助。
INSTRUCTIONS = """你是一名企业数据分析师。你的任务是根据用户的自然语言问题，生成 SQL 查询并返回结果。 工作流程： 1. 先用 get_allowed_tables 查看当前用户可访问的表。 2. 根据问题涉及的维度，用 get_schema 查看相关表结构。 3. 若存在业务口径疑问，可先查询指标口径（此处简化为 schema 信息）。 4. 生成 SQL 后用 validate_sql 自检，确认无安全风险。 5. 通过 execute_select 执行查询，将结果整理成用户能读懂的结论。 安全纪律： - 只使用 get_allowed_tables 返回的表，禁止猜测表名。 - 只执行 SELECT 查询，禁止生成任何非 SELECT 语句。 - 禁止将手机号、身份证等敏感字段明文展示，除非用户明确要求且脱敏处理。 - 如果用户的问题涉及不在权限范围内的表，直接拒绝并说明。 回答风格： - 先给出结论，再附上简短的取数逻辑说明。 - 用 markdown 表格展示结构化数据。 """ data_agent = Agent( name="DataAnalystAgent", instructions=INSTRUCTIONS, tools=[get_allowed_tables, get_schema, validate_sql, execute_select], )跑起来也很直观，传入AccessContext作为上下文即可。这里的设计原则是“身份决定上下文，上下文决定工具视野”，模型本身不接触真实用户身份之外的信息。
async def ask(question: str, user_id: str): ctx = AccessContext( user_id=user_id, role="analyst", allowed_tables=["fact_orders", "dim_product", "dim_store", "dim_org"], row_filter="dim_org_id = 'D10086'", sensitive_fields=["customer_phone", "customer_idcard"], ) result = await Runner.run(starting_agent=data_agent, input=question, context=ctx) return result.final_output在真实项目里，我强烈建议别把 AccessContext 放在客户端传参——一定要由后端根据 token 解析服务端生成，否则几行代码就能被恶意伪造。
5. 让 Agent 更可靠的关键细节：从系统提示词到查询重写
如果开发 Agent 只写提示词然后交付，那跟裸调 API 的区别就不大。到了企业场景，可靠性取决于在代码里塞进多少“纠错机制”。下面分享几个我实测很有用的手段。
规范提示词里的工具调用顺序。上面代码里，我将“先白名单、再 schema、再自检、再执行”写得很明确。实际跑下来，“自动对齐流程”可以明显减少模型跳跃式调用。尤其是validate_sql这个工具的存在本身，会促使模型在生成 SQL 之后多一道自我反思，比单纯在 System Prompt 里写“请仔细检查你的 SQL”有效得多。
对模型的“事实性幻觉”做收敛。模型经常会自由发挥，比如发现用户要查“大客户订单”，就自作主张创造一张big_customer表。这种幻觉的根治方案是把工具返回的 schema 信息做得尽量结构化，让模型“只使用提供的表结构拼 SQL”。如果get_allowed_tables返回的信息足够完整，且明确规定“没有出现在白名单里的表视为不存在”，幻觉率会大幅下降。
做一个查询重写（Query Rewriting）函数。有时候模型生成的 SQL 逻辑正确但性能很差，比如对事实表做全表扫描、或者 JOIN 时没有用索引字段。查询重写层干的事很朴素：分析 SQL 是否包含对分区表的高成本扫描；如果有超过成本的路径，直接拒绝并引导模型重写。这一层也负责把“SELECT *”展开为白名单内可见字段——避免模型把敏感列全部带出来，同时提升查询性能。
处理“Agent 自说自话”的循环。Agents-API 的 Run 会自动终止——设置了 max_turns 或检测到最终输出条件。但在取数场景中我推荐额外限制每个会话最多 3 轮工具调用。取数任务很少需要长时间多轮推理，一旦超过轮次说明问题不够清晰，直接转人工比让 Agent 瞎折腾更靠谱。这种“快速失败”策略可以显著降低 token 成本，也避免 Agent 在异常状态下反复重试。
把业务口径做成显式工具。我在实际项目里增加了一个get_metric_definition(metric_name)工具，里面存着一张口径表：什么是 GMV、什么是净营收、退款订单怎么算。模型在查询前应先查看口径定义再写 SQL。这个设计帮我解决了一个很头疼的问题——模型对业务术语的解释往往“语义上通顺、实际上错误”。用工具接入口径而不是让模型自己理解，口径维护变成了业务团队能直接操作的事情。
这部分我特别想强调一点：Agent 的可靠性提升主要不是靠“更大更强的模型”，而是靠给模型加约束和反馈回路。你把校验结果返回给模型（比如“你的 SQL 中引用了无权访问表 xxx，请仅基于白名单内的表重写”），模型的下一次输出通常会更守规矩。这种“约束-反馈-重试”的循环，正是 Agent 相比传统一次生成接口的核心优势。
6. 权限隔离的工程细节：多租户、字段级脱敏与行级强制过滤
企业取数场景里，权限设计是“面子工程”的重灾区。很多团队在 Agent 演示时用统一的高权限账号，PPT 上写着“支持权限控制”，实际上根本没有。这里我分享一套分层权限落地细节，确保切换身份后真的有效果。
白名单不等于库表授权，要“按用户动态生成”。不要把数据字典完整丢给模型。每次用户发起请求，工具先根据身份获取该用户可见表集合。不要返回全库 schema，否则模型总会指染不该看的数据。我给出的代码里get_allowed_tables返回的就是当前用户的表名列表，模型没有机会看到“文言文表”。这一点对防止越权查询数据库至关重要。
行级过滤：不是在 SQL 后拼接，而是解析后注入。不少人做行级权限时简单地在 SQL 末尾追加AND org_id = 'xxx'，遇到带子查询、GROUP BY 或 EXISTS 的复杂 SQL 就会出错。正确做法是用 SQL AST 解析器（我用的是 sqlglot）找到主查询的 WHERE 位置，将过滤表达式正确地 AST 合并进去。代码示例里简化成字符串拼接，但生产中强烈建议升级为sqlglot这类工具。它的优势是能从语义层面识别真正的查询片段，避免文本注入导致的语法破坏。
同一套 Agent 服务多租户时，Context 的生命周期要盯紧。Agents SDK 的context是传给 Run 的参数，本身是进程内的。但如果有异步并发请求，稍不小心就可能把 A 用户的 context 串到 B 用户的请求里。我的做法是在入口处生成不可变的 AccessContext ID，并绑定一个asyncio.TaskLocal或在 runner 的每次调用时显式传入，确保隔离。这类问题在并发量上来后特别容易出现，引发的后果不只是数据泄露，还可能是跨租户的数据串隐形事故。
敏感字段不是查出来再想脱敏，而是查询时就拦截。我见过不少方案把脱敏放在 AI 平台返回结果前做，但 SQL 执行完数据已流出内存。更好的做法是在 schema 层就把敏感字段标记出来，然后让鉴权层决定查询者是否有权看明文。如果没权限，直接不允许该列进入查询结果；如果有权限，在脱敏通道中按策略输出。只有需要明文的角色（财务、合规）才能透视明文列，其他人一律只能拿到掩码结果。
数据库账号设计得越细，越能兜底 Agent 的“不可预测性”。我的生产配置是：AI 平台只持有只读账号，而且每个核心业务域一组账号。跨域的 JOIN 若需要涉敏，则由另外的服务账号处理。这样即使模型生成一个绕过了我在应用层校验的恶意 SQL（虽然概率低），数据库自身的权限系统还是能挡一层。
一句话总结这层的建设原则：应用层校验是防“模型犯傻”，数据库层权限是防“系统失守”。两层缺一不可。
7. 观测与审计：从“模型跑通”到“老板敢用”的最后一公里
安全报告、可控性和权限做好之后，还有一道关卡容易被忽略：可观测性。一个取数 Agent 如果没有观测系统，任何一次查询出问题都只能靠人工翻日志，效率极低。我在这块踩了不少坑，给大家梳理几个最实用的点。
会话级追踪。在 API 入口生成request_id，并伴随整个 Run 生命周期。Agents SDK 的RunState会输出每步运行信息，务必把它持久化为 JSON 审计日志，至少包括：提问人、问题、上下文身份（AccessContext 摘要）、每一步工具调用的入参出参、最终 SQL、执行时长、token 消耗。有了这份数据，再配合差异分析就能回答“谁在什么时候基于什么逻辑得到了什么数据”。
给 Agent 装上“成本计量器”。用 lifecycle hook 记录每一轮 LLM 调用的输入 token 和输出 token，按用户或部门聚合。取数 Agent 的 token 消耗跟问题复杂度高度相关，有的查询一次需要好几轮工具调用，成本差距可能到 10 倍甚至更高。我的经验是把 token 成本透明化给平台运营团队，并设定部门配额和预警阈值，避免月底账单“爆表”。
定期做“影子评测”。我会从历史真实取数请求里抽一批（已经有人工答案的），形成一条评测集，每隔一段时间用 Agent 跑一遍。关注指标有三个：
- 安全拦截率：请求里有多少可疑 SQL 被校验器拦下来（理想不是 0，拦截说明系统在工作）；
- 最终正确率：Agent 最终交付的结果与人工答案是否一致；
- 无效轮次率：有多少次工具调用是重复试错。
这三个指标能直接反映 Agent 的整体健康度，比单纯看“有没有报错”靠谱得多。
人工审批模式的取舍。不是所有取数请求都需要自动执行。我的实践是分三级：只读敏感数据的轻查询自动放行；涉及多表 JOIN 或第一次访问新表的进入“审批队列”；DML 类请求永远拒绝。审批模式会牺牲一些体验，但对于数据团队来说，关键政策的落实比“分钟级响应”更重要。
关于 Agent 的“可解释性”问题。做过企业交付的都知道，业务方不会因为你说是“大模型生成的”就接受结果。所以我要求 Agent 在最终输出里包含“取数链路说明”——比如“我基于 fact_orders 和 dim_product 做了聚合，过滤条件包含 org_id 和 status=‘paid’，未计算退款订单”。这种说明从审计角度来看就是一朵“可复现的数据血缘记录”，业务方也能据此判断结果是否符合预期。
8. 从“能跑”到“真好用”：进一步优化方向
基础框架搭建完，我开始在这套系统上做体验和性能优化，下面几个方向很值得继续深入。
异步查询和长任务分离。如果查询的表很大，同步等待数据库响应会耗尽 Agent 的耐心和成本。我的做法是把execute_select改造为提交一个查询任务并立即返回“查询已进入队列，结果稍后查看”的状态，然后通过回调或轮询接口去拿结果。这样既避免模型在长时间等待时产生幻觉式补全，也让整体的用户体验更像成熟 BI 工具。
给 Agent 装一个“结果解释器”子 Agent。主 Agent 只负责生成和执行 SQL，拿到查询结果之后，可以由一个小参数的解释子 Agent 负责把枯燥的表数据转化成业务洞察。两个 Agent 解耦之后还有个额外好处：主 Agent 的上下文长度限制不会因为塞进大量结果行而告急。通常我会让主 Agent 只传“聚合统计结果”给解释子 Agent，原始明细不外泄。
支持对话式追问。取数很少一步到位。用户说“看下华东区销售情况”，Agent 给了数据后，用户可能追问“那对比一下华南区”“只看线上渠道”。如果 Agent 没有记忆，每次都要重置上下文，既浪费 token 体验也差。Agents-API 支持在 Run 中维护历史消息，我将历史消息摘要化（只保留最近的 k 条对话），这样既能延续上下文又不至于撑爆窗口。
提升自然语言到语义映射的“贴合度”。数据团队反映比较多的一个问题是模型偏好在嵌套子查询里写复杂业务逻辑，导致结果难以解释。我的处理方式是在口径工具里预留“最常用查询模式模板”，模型被引导优先使用模板，再在模板基础上做参数替换。训练出“逼模型用公司定义的查询模式”的习惯之后，SQL 的可复用性大大提高，也直接减少了很多低级语法错误。
多 Agent 协作的路子也可以走。在数据域特别多的大型企业，单 Agent 把全库表都背在身上不现实。我目前正在实验的方向是“一个主控路由 Agent + 多个领域查询 Agent”，路由 Agent 负责理解用户意图、判断该找哪个域的数据，然后交给对应子 Agent 取数。这能让提示词更短、工具集合更精简，也匹配企业里按业务域管理数据权限的现状。每个域自己维护自己的口径工具和数据字典，质量提高非常明显。
用一句话来总结这些优化方向：在 Agent 落地过程中，真正拉开差距的，不是模型能力的差异，而是系统工程化水平的高低。把工具边界做干净、把校验逻辑做准确、把上下文管理做成常规手段，产品能力就能往前走一大步。
9. 踩坑实录：生产环境中那些你没预期到的问题
最后这部分，我把过去一段时间里踩过的坑集中复盘一下，基本是按照“现场现象 → 根因 → 解决方案”的格式来写。这些坑绝大多数在官方文档里不会详细写，但对做生产级 Agent 开发的人来说价值很大。
坑一：Agent 上下文里塞了太多 schema，把上下文窗口撑爆了。起初我把整个库的 schema 一次性放到 prompt 里，一旦库存几百张表，Prompt 直接超长。后来改成按需加载 schema：用户提出问题时先通过一个“意图识别/表定位”步骤，提取候选表清单，再向 Agent 提供这几张表的 schema。这样既不牺牲准确性，又守住了窗口成本。
坑二：模型对工具返回的 JSON 里的“空值”理解得很差。例如execute_select返回的“columns”为空列表，模型会以为查询没结果，其实是没有构造好输出；或者返回 Numeric 类型的空值为 null，模型可能说“查询不到数据”。解决方式是对工具返回结果做标准化封装：永远输出“是否有数据、列名、行数、数据摘要、当前得分”等固定结构，模型就不至于误解工具结果。
坑三：在 SELECT 查询里隐含“子查询绕过”。静态校验如果只做文本匹配，很容易被SELECT * FROM (DELETE ... RETURNING *)这类嵌套写法绕过去。我把校验从文本正则升级为 AST 级别检查之后，这个问题才真正封死。生产环境务必用sqlglot或sqlparse的语法树，先解析再遍历验证每个节点操作类型，而不是靠关键词黑名单——黑名单只是辅助。
坑四：Agent 执行失败时，模型常常“编造执行结果”。一旦某个工具抛异常，模型为了完成输出会自己脑补一个看似合理的结果。我后来对所有工具返回都加了二维约束：每个返回值都必须包含is_error字段，而且失败的话必须携带结构化错误信息。在系统提示里写清楚“工具返回 is_error=true 时必须向用户如实汇报失败原因，禁止猜测结果”，情况才明显好转。
坑五：concurrency 引发的 AccessContext 交错。前面提过，高并发下跨用户 context 串线是我最紧张的一个问题。虽然 Agents-API 的input/context参数本身是显式的，但在Runner.run外面如果自定义了异步缓存或任务并发复用，就很容易把上个用户的上下文带到下个用户。最终我选择“每次请求都新建一个不可变 Context，并强制审计日志关联 request_id”，这种做法牺牲了一点性能，但换来了权限上的绝对安心。看到这会点头的读者，应该都体会过“多租户并发隔离之痛”。
坑六：日志打得过于详细，反而给安全和合规带来风险。一开始为了排查方便，把每一步 prompt 和工具输出都打印到日志。后来被合规团队提醒，这些日志包含了敏感数据。最后改为“日志脱敏存储”：工具入参中涉及敏感字段值的部分默认截断，完整的“检查单”类数据只保存在加密审计区。这个平衡很重要——观测和隐私不能二选一，必须共存。
踩坑到最后，最核心的体会是什么呢？做企业级 Agent，专业知识当然重要，但更考验的是“把不确定性关进笼子里”的系统设计能力。大模型天然存在不可预测性，我们要做的不是消除它，而是通过架构、工具和流程让不可预测性只能在一个受控范围内影响结果。数据安全、权限隔离、观测审计、快速失败——这些比单点思考“怎么让模型更聪明”重要得多。
回到这个项目本身，我的方案目前在公司内部已经稳定运行了两个多月，日均处理数百个取数请求，安全拦截和权限隔离都经受住了真实业务的考验。如果你正准备在这个方向落地，建议先照猫画虎跑通最小闭环，再逐步叠加权限细节和观测体系。等这些土木工程都到位了，你才会真正感受到 Agent 在取数场景的威力：它不只是让你从写 SQL 的重复劳动里解放出来，更重要的是让数据取用变得更“可管、可控、可审计”。继续往下做，后面你会发现可扩展的方向多得很——多域子 Agent 协作、指标口径自动维护、甚至面向业务方的自助取数门户，都会在你搭好的地基上自然生长出来。
