OpenAI 内部 Data Agent 拆解：从 LLM 到 SQL 的 Codex 配置骨架

1. 为什么 Data Agent 的瓶颈从来不是写 SQL我见过太多团队做 Text-to-SQL第一反应是“把 schema 塞进 prompt 让模型写 SQL”。真跑起来才发现模型写 SQL 的语法几乎不会错错的是它根本不知道该用哪张表。OpenAI 数据平台那位工程主管说得很直白分析师在写第一行 SQL 之前可能已经花了好几个小时搞清楚该用哪些表、怎么关联。这个场景放到 9 万张表、1.5 EB 数据的规模下问题会被放大到无法用“多给点 schema”来解决。所以 Data Agent 的本质不是 SQL 生成器而是一个“上下文组装 工具调度 结果校验”的闭环。LLM 只负责推理真正决定成败的是它拿到什么上下文、能调用哪些工具、以及运行时怎么把自然语言问题一步步收敛成可验证的答案。这篇就按这个思路用 Codex 侧的config.toml和settings.json搭一个最小可跑的骨架把“自然语言 → SQL → 执行 → 校验”这条链路走通。适合已经会写 Python、想复刻同类链路但不想一上来就堆路由器和微调的开发者。2. TaoToken 前置把模型调用和 Key 准备好在动手写配置之前先把模型入口固定下来。Data Agent 的运行时框架需要一个稳定的 LLM 调用端点否则后面调试 SQL 生成时你会分不清是 prompt 问题还是网络问题。我习惯用 TaoToken 作为统一入口它的 API 地址是https://taotoken.net/api兼容 OpenAI 风格的调用方式Codex 和自定义 harness 都能直接对接。你需要先拿到一个 API Key。打开控制台页面https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content在 API Keys 页面创建一个新 Key复制出来存到环境变量里。注意不要把它硬编码进config.toml提交到仓库后面配置里我会用环境变量引用。export TAOTOKEN_API_KEYsk-你的key如果你还没决定用哪个模型可以先到模型对话页面https://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content手动问几个数据相关的问题感受一下模型对表结构描述的理解能力。这一步不是必须的但能帮你后面写上下文组装逻辑时心里有数。接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content参数细节以文档为准。3. Codex 侧 config.toml 与 settings.json 可复制骨架Codex 的配置分两层config.toml管模型和运行时行为settings.json管工具集和上下文来源。下面这份骨架是我按“vanilla agent, rich foundation”的思路精简过的工具只保留 6 个核心的避免模型在重叠工具之间选错。3.1 config.toml模型与运行时# ~/.codex/config.toml model gpt-5.5 model_provider taotoken [model_providers.taotoken] name TaoToken base_url https://taotoken.net/api env_key TAOTOKEN_API_KEY wire_api chat [agent] max_turns 12 tool_choice auto temperature 0.2 [agent.runtime] # 每轮把工具返回结果回灌给模型形成 推理→行动→观察 循环 loop reason_act_observe stop_on_sql_error false max_sql_retries 3max_turns设 12 是实测下来比较稳的值太少会在复杂问题上提前放弃太多会让模型在错误方向上反复试。temperature压到 0.2因为 SQL 生成不需要创造性需要的是稳定复现。stop_on_sql_error false很关键SQL 报错本身就是有价值的观察信号让模型看到错误再修正比直接中断好得多。3.2 settings.json工具集与上下文层{ tools: [ { name: list_tables, description: 根据关键词返回候选表名及一句话描述, type: builtin }, { name: get_table_schema, description: 返回指定表的列定义、类型、主键, type: builtin }, { name: get_table_usage, description: 返回该表的历史查询权重和血缘关系, type: builtin }, { name: run_sql, description: 在只读副本上执行 SQL 并返回前 100 行, type: builtin }, { name: search_docs, description: 检索内部知识库中与问题相关的制度文档, type: builtin }, { name: get_memory, description: 读取全局和个人级别的历史修正记录, type: builtin } ], context_layers: [ table_usage_metadata, human_annotations, code_enrichment, institutional_knowledge, memory, runtime_probe ], retrieval: { top_k_tables: 8, top_k_docs: 3, hybrid_search: true } }这里top_k_tables设 8 而不是 20是因为上下文窗口里塞太多相似表描述反而会干扰模型判断。hybrid_search打开语义加精确文本匹配因为有些表名是缩写纯向量检索会漏。3.3 上下文组装的最小实现工具定义好了还需要一个组装层把六层上下文拼成 prompt。下面这段 Python 是骨架重点看它怎么把“表描述 记忆修正 制度知识”叠在一起。import os import json import requests API_URL https://taotoken.net/api/chat/completions HEADERS { Authorization: fBearer {os.environ[TAOTOKEN_API_KEY]}, Content-Type: application/json, } def assemble_context(question: str) - str: # 1. 向量化问题检索最匹配的表描述 tables retrieve_tables(question, top_k8) # 2. 叠加记忆层的修正 memory get_memory(question) # 3. 检索制度知识 docs search_docs(question, top_k3) blocks [] for t in tables: blocks.append(f表 {t[name]}: {t[description]}\n f粒度: {t[grain]}\n血缘: {t[lineage]}) if memory: blocks.append(历史修正:\n \n.join(memory)) if docs: blocks.append(制度知识:\n \n.join(docs)) return \n\n.join(blocks) def ask_data_agent(question: str) - dict: context assemble_context(question) messages [ {role: system, content: 你是数据代理。先用 list_tables 和 get_table_schema 确认表 再写 SQL用 run_sql 执行。如果报错读错误信息后修正重试。}, {role: user, content: f上下文:\n{context}\n\n问题: {question}}, ] resp requests.post(API_URL, headersHEADERS, json{ model: gpt-5.5, messages: messages, temperature: 0.2, }) return resp.json()注意 system prompt 里没有写“第一步做什么、第二步做什么”这种逐步指令。OpenAI 团队的经验是详细指令反而损害结果高层指导加正确上下文就够了。我试过把步骤写死模型遇到稍微不同的表结构就卡住改成目标导向后反而更灵活。4. 验证请求一次自然语言到 SQL 的完整动作配置搭好后跑一个真实问题来验证闭环。假设问题是“上个月每个渠道的活跃用户数是多少”。下面看代理循环里发生了什么。4.1 第一轮模型先探表模型拿到问题后不会直接写 SQL而是先调list_tables{tool: list_tables, arguments: {keyword: 活跃 用户 渠道}}返回候选表[ {name: dws_user_active_daily, desc: 日粒度活跃用户含渠道维度}, {name: dws_user_login_daily, desc: 日粒度登录用户不含登出}, {name: ads_channel_summary, desc: 渠道汇总T1 更新} ]这里就是关键点dws_user_active_daily和dws_user_login_daily都有user_idschema 看起来几乎一样但一张含已登出用户一张不含。模型靠描述里的“不含登出”这个人工注释才能选对。4.2 第二轮确认 schema 和血缘{tool: get_table_schema, arguments: {table: dws_user_active_daily}}返回列定义后模型再调get_table_usage确认这张表是高频仪表盘在用权重高可以信任。4.3 第三轮生成并执行 SQLSELECT channel, COUNT(DISTINCT user_id) AS active_users FROM dws_user_active_daily WHERE dt BETWEEN 2026-01-01 AND 2026-01-31 GROUP BY channel ORDER BY active_users DESC;run_sql在只读副本上执行返回前 100 行。如果 SQL 报错比如列名拼错错误信息会回灌给模型它看到column channel_id does not exist后会改成channel重试。max_sql_retries 3就是给这个修正循环留的空间。4.4 结果校验拿到结果后代理不会直接返回数字而是做一次自检行数是否合理、有没有 null 渠道、总数是否和已知的日活量级对得上。这一步可以再调一次模型把结果和问题一起发过去问“这个结果是否回答了问题有没有异常”。校验通过才把答案、SQL、来源表一起返回。result ask_data_agent(上个月每个渠道的活跃用户数是多少) print(json.dumps(result, ensure_asciiFalse, indent2))成功时你会看到类似输出{ answer: 上个月各渠道活跃用户数自然流量 128400投放 96300社交 54100, sql: SELECT channel, COUNT(DISTINCT user_id) ..., tables: [dws_user_active_daily], verified: true }5. 本篇常见错排查跑不通的时候按下面几个方向查基本能覆盖 90% 的问题。模型一直选错表。先看top_k_tables是不是设太大8 是上限超过 12 相似表会互相干扰。再看表描述里有没有写清粒度和“何时用这张而不是那张”裸 schema 不够必须有人工注释或代码富化补上语义。SQL 报错后模型不修正直接放弃。检查stop_on_sql_error是不是被设成了 true以及max_sql_retries是不是 0。错误信息必须完整回灌截断错误信息会让模型猜不到问题在哪。工具调用循环停不下来。max_turns设太大加上工具描述重叠模型会在两个做类似事的工具之间反复横跳。把重叠工具删掉工具数控制在 10 个以内。OpenAI 团队从 40 个砍到 13 个才稳定你从 6 个起步更稳。上下文里塞了太多一次性查询记录。历史查询不是越多越好一次性探索查询权重最低高频仪表盘查询权重最高。如果你的get_table_usage返回的是全部历史模型会被噪声带偏按可信度排序后再取 top。API 调用返回 401 或 403。检查TAOTOKEN_API_KEY环境变量有没有正确导出config.toml里的env_key名字要和环境变量一致。Key 相关操作在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content页面管理接入细节看https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content。结果校验总是失败。校验 prompt 别写太严量级对得上、没有明显 null 就算过。校验的目的是拦住明显错误不是追求完美。6. 把闭环跑通之后往哪走最小闭环跑通后你会发现真正花时间的不是模型调用而是上下文层的数据准备。表使用元数据、人工注释、代码富化这三层决定了模型能不能在 9 万张表里找到对的那张。OpenAI 用 Codex 每晚爬管道代码做富化每张表 5 到 10 分钟这个投入换来的是一张表描述里包含“实际包含什么数据、怎么派生、数据新鲜度、何时用这张而不是相似表”。如果你只有几百张表可以先手工写注释规模上来后再考虑自动化富化。另一个值得提前想的是记忆层。全局记忆存团队公认的修正个人记忆存单个用户的偏好检索时叠加在表描述之上。这样代理不用每次从零开始上次纠正过的表选择下次直接生效。长期跑编码和 Agent 任务的话可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content把模型调用和工具链的额度统一管理省得每次调试都担心额度。最后提醒一句工具贵精不贵多指令越少越好时间线重新定义。如果这个项目听起来该花一年正确的问题是——有了代理它能不能在一个季度内跑通。