Agent接入Text-to-SQL实操：从Schema注入到安全执行

先说一下这次的背景。我一直在做一套手搓 Agent 的系列前面已经把 Agent 的基础循环、记忆、工具调用这几块讲完了。到了第 2.3 关主题是给 Agent 补上数据库查询能力核心技术点就是Text-to-SQL——让 Agent 听懂用户的自然语言问题自己把问题翻译成 SQL去数据库里把结果查出来再组织成回答。这一篇是“中”篇聚焦怎么把 Text-to-SQL 真正接入 Agent 的实操链路不聊论文公式直接讲能跑通的东西。先说这能力到底解决了什么问题。Agent 光会聊天、光会算数是没法和业务数据打交道的。用户问“我们数据库里哪些课程的选课人数最多”如果 Agent 没有数据库查询能力它只能瞎编一个答案有了 Text-to-SQL它能自己读表结构、写查询、拿结果、再总结。适合正在做 Agent 应用开发、想让自己的 Agent 告别“一问数据就胡说八道”的读者。这篇我会把一个可运行的实现拆开讲包括表结构设计、Schema 注入、SQL 生成、安全执行这几个环节中间穿插我踩过的坑和一些工程上的取舍。1. 内容整体设计与思路拆解1.1 Text-to-SQL 在 Agent 里的定位先说清楚一件事Text-to-SQL 不是一个独立的产品它是 Agent 的一个工具。就好比人长了一只手这只手专门负责“查数据库”。在 Agent 的主循环里用户提问进来Agent 判断要不要查数据库如果决定查那就调用一个查询工具这个工具内部就是 Text-to-SQL 的完整流程。我在网上看到不少初学者把 Text-to-SQL 理解得太重又是搭训练管道又是搞模型微调其实在 Agent 场景下大多数时候用不着这么复杂。现代大语言模型本身就有很强的 SQL 生成能力关键在于你怎么把数据库结构、字段含义、约束规则这些“上下文”喂给它以及怎么处理它生成的 SQL 跑不通的情况。所以我的整体设计思路是三个词结构注入、生成执行、失败重试。把这三个词展开就是整篇博文的骨架。结构注入就是让模型看到有哪些表、哪些字段、什么类型、什么含义生成执行就是让模型产出一条 SQL我们在受控环境里去执行失败重试就是当 SQL 报错或者结果不对时把报错信息返回给模型让它自己改。这一步其实是决定体验好不好的关键因为模型写 SQL 不可能一次全对能自我纠错和不能自我纠错的 Agent用起来完全两个感受。1.2 为什么我选择“工具模块”而不是“专用接口”第二个取舍是Text-to-SQL 在 Agent 里应该做成一个通用 Query 工具还是给每个查询需求单独写死一个接口比如用户经常问选课人数那就写一个 get_course_count() 接口是不是更简单我的观点是如果只有三五个固定查询需求写死接口没问题但只要查询需求的组合是开放的那就必须做通用 Text-to-SQL 工具。原因很好理解用户的自然语言问题组合是无限的今天问“哪门课选的人最多”明天问“每个学院的平均学分绩点”后天问“哪个老师的课被退课最多”你不可能把所有问题都预先写成接口。通用 Text-to-SQL 工具等于把“数据库查询”这整件事抽象成了一个能力Agent 拿到任何带数据的问题都能自己拆解。代价就是要处理 SQL 生成的不确定性这也是后面几节要重点讲的部分。这个定位想清楚了整个模块的边界就清晰了输入是自然语言问题输出是查询结果摘要内部夹着 schema 读取、Prompt 组装、SQL 执行、异常反馈这四件事。2. 核心细节解析与实操要点2.1 Schema 注入把数据库结构翻译给模型Text-to-SQL 最容易被忽视的环节其实是 schema 注入。不少人直接写一句“把用户的自然语言问题转成 SQL”然后期望模型自己猜数据库结构这几乎一定会翻车。模型再强也不会猜到一个应用的表名、字段名、枚举值所以必须把结构信息明确喂进去。我用的做法是程序启动时读取数据库的表结构自动拼成一段结构描述文本。比如一张 courses 表会生成这样的描述表 courses: - id: INTEGER, 主键 - name: VARCHAR(100), 课程名称 - teacher_id: INTEGER, 外键关联 teachers.id - credit: INTEGER, 学分 - max_student: INTEGER, 课程容量这段话拼到 system prompt 里模型写 SQL 的时候就有一个明确的“世界模型”。需要注意的是字段最好加上业务含义注释不要只写类型。比如 max_student 如果不注释“课程容量”模型可能把它当成实际选课人数注释清楚之后写“选课人数最多”这种查询时模型就会去关联选课表做 COUNT而不是拿 max_student 字段出来糊弄事。还有一点是表的数量问题。如果你有几十张甚至上百张表一次全塞进 prompt一是浪费 token二是会让模型发懵。实战里我的处理是两层第一层先给模型一个表清单只有表名和一句话说明让模型先选表第二层把选中的表的详细字段结构再注入一轮接下来才生成 SQL。这就是把“选表”和“写 SQL”拆成两步。对于大部分中小型应用表数量在十几张以内直接全部注入也是可以的我下面的例子就采用了这种更简单的单次注入方案。2.2 Prompt 里的约束条件比示例更重要关于 Text-to-SQL 的 prompt网上能搜到很多花哨的 few-shot 示例我实际对比下来的感受是示例要有但规则约束更重要。示例只能覆盖有限的写法而规则能框住模型的边界行为。我在 system prompt 里固定写了几条强约束实测下来很稳只允许执行 SELECT 查询禁止任何 INSERT、UPDATE、DELETE、DROP、ALTER 语句。查询必须使用 LIMIT默认不超过 200 行。如果用户的问法语义不明确需要区分“最值”和“明细”比如“课程号是 CS101 的选课人数”是明细聚合“哪门课选的人最多”是分组排序。不要把表名字段名翻译成中文保持原样别名可以用简单字母。不确定字段含义时使用 schema 定义里的说明判断不要自己臆造字段。这几条写进去之后SQL 的可用率提升非常明显尤其是第 2 条和第 5 条。第 2 条防止用户一个“把所有数据都查出来”就把表整个拉爆第 5 条防止模型自己发明不存在的字段。你可以在测试里故意让模型写一条复杂的 JOIN 查询加不加第 5 条效果差距很大。另外我还做了一个小技巧把建表语句本身附到 prompt 里而不是只放摘要。因为 DDL 里包含字段类型、默认值、索引、外键关系这些信息比我自己写摘要准确得多也不用维护双层文档。缺点是最开始的几张表 DDL 比较长但换来的是模型对字段类型的理解精准很多比如 DATE 类型的字段模型不会默认当成字符串去 LIKE。2.3 框架选型从裸调 API 到 Agent 工具注册第 2.3 关的中篇毕竟是在 Agent 语境下讲所以实现上要和 Agent 主循环结合起来。我看很多项目会用现成的 Agent 框架比如 LangChain 的 create_sql_agent、LlamaIndex 的自然语言查询包这些都能跑但封装得太厚出了问题不好调试。我的建议是这阶段一定要手搓一遍。不是说框架不好而是你要理解每个环节的数据流自然语言进到工具函数工具函数组装 prompt调用 LLM 拿 SQL执行 SQL 拿结果结果返回给 Agent 主循环。用框架你只调用一个黑盒出错了不知道在哪一环断的。手搓完之后再上框架你会对框架的内部机制心里有数。这里我用轻量注册函数来模拟 Agent 的工具调用机制。一个查询工具本质上就是定义好名字、描述、参数格式以及一个 callable。下面的例子会直接实现这个 callable 的内部逻辑而不是依赖任何具体框架。3. 实操过程与核心环节实现3.1 环境准备与演示表结构我用 Python 3.10 以上版本数据库用 SQLite方便演示不用额外起服务。先把环境准备好pip install openai sqlalchemySQLite 的好处是单文件、零配置sqlite3是 Python 内置的只要再用 SQLAlchemy 来做连接和反射读 schema 会方便很多。如果你后续要换 MySQL 或 PostgreSQL只需改连接串。我先建一组和“选课系统”相关的演示表这也是网上搜 Text-to-SQL 时高频出现的场景。三张表CREATE TABLE students ( id INTEGER PRIMARY KEY, name VARCHAR(50) NOT NULL, major VARCHAR(50), enrolled_year INTEGER ); CREATE TABLE courses ( id INTEGER PRIMARY KEY, name VARCHAR(100) NOT NULL, teacher VARCHAR(50), credit INTEGER, max_student INTEGER ); CREATE TABLE course_enrollment ( id INTEGER PRIMARY KEY, student_id INTEGER NOT NULL, course_id INTEGER NOT NULL, score REAL, FOREIGN KEY (student_id) REFERENCES students(id), FOREIGN KEY (course_id) REFERENCES courses(id) );你要是自己复现往里面塞一点模拟数据就行。这里的关键是表的命名和注释要清晰因为后面 schema 反射会自动把它们喂给模型。3.2 核心代码Schema 提取与 Prompt 构建现在写核心模块。我用 SQLAlchemy 的 inspector 来反射数据库里的表结构然后拼成 schema 描述。这样比直接解析 DDL 更省事也能统一处理多种数据库方言。from sqlalchemy import create_engine, inspect, text import json def build_schema_description(db_url: str) - str: 从数据库连接串反射表结构生成schema文本。 engine create_engine(db_url) inspector inspect(engine) lines [] for table_name in inspector.get_table_names(): lines.append(f表 {table_name}:) for col in inspector.get_columns(table_name): col_desc f- {col[name]}: {col[type]} if col.get(primary_key): col_desc , 主键 if col.get(comment): col_desc f, {col[comment]} lines.append(col_desc) # 外键信息可选延伸 fks inspector.get_foreign_keys(table_name) for fk in fks: constrained fk[constrained_columns] referred fk[referred_columns] lines.append(f- 外键: {constrained} - {fk[referred_table]}.{referred}) return \n.join(lines)这段代码拿到表名、字段名、类型、主键、外键关系。SQLite 对字段注释支持一般但 MySQL 等数据库会返回 comment放到描述里能让模型对业务含义理解得更准这个特性建议保留。接下来是组装 prompt 的核心函数以及被 Agent 调用的查询工具函数。先看 prompt 部分SYSTEM_PROMPT_TEMPLATE 你是一个数据库查询助手。根据用户的自然语言问题生成一条SQL查询语句。 数据库结构如下 {schema} 规则 1. 只允许生成 SELECT 查询。 2. 查询必须带 LIMIT默认限制 200 行。 3. 不要臆造表名或字段名。 4. 如果问题语义不明确优先做聚合查询。 5. 只输出SQL代码不要多余解释。 用户问题{question} def generate_sql(question: str, schema_text: str, client, model: str) - str: prompt SYSTEM_PROMPT_TEMPLATE.format(schemaschema_text, questionquestion) resp client.chat.completions.create( modelmodel, messages[ {role: system, content: 你是SQL专家。}, {role: user, content: prompt} ], temperature0 ) return resp.choices[0].message.content.strip()这里有几个细节我解释一下。第一temperature 必须设 0。SQL 生成是确定性问题不需要创造性temperature 不为 0 会导致同样的问题每次生成不一样的 SQL效率极差。第二我在 user prompt 里再次强调“只输出 SQL”因为有些模型会把解释和代码混在一起后面解析会很麻烦。第三system 消息和 user 消息里的规则有点像重复了但实测对模型遵守格式有用你可以理解为双保险。如果表特别多prompt 太长可以先用一个表清单让模型选表再针对选中的表生成详细 schema。我上面的 build_schema_description 是一次性输出全部结构适合表在 20 张以内的系统超过 20 张建议拆成“选表 详细结构”两步避免 prompt 过长把模型的注意力稀释掉。3.3 安全执行只读连接与结果集限制SQL 生成出来之后怎么执行是个大坑。你肯定不能拿带写权限的连接去跑模型的输出万一模型哪天生成了 DELETE FROM courses那整个演示库就没了。我在工程里强制使用只读连接从根上杜绝写操作。SQLite 的只读连接可以用 URI 参数实现from sqlalchemy import create_engine, text from sqlalchemy.engine import make_url import pandas as pd def execute_query_readonly(db_url: str, sql: str, max_rows: int 200): 在只读连接上执行SQL限制返回行数。 # 强制把SQL包一层限制最大返回行数 wrapped_sql fSELECT * FROM ({sql.rstrip(;)}) AS _t LIMIT {max_rows} engine create_engine(db_url, connect_args{uri: True}) with engine.connect() as conn: # SQLAlchemy 2.0写法 result conn.execute(text(wrapped_sql)) columns list(result.keys()) rows [dict(zip(columns, r)) for r in result.fetchall()] return columns, rows关键点是connect_args{uri: True}配合连接串使用sqlite:///file:demo.db?modero这样的 URI 格式数据库就以只读模式打开。如果模型生成的 SQL 里有写操作SQLite 只读模式会直接抛错不会真的改数据。这个“包一层 LIMIT”的做法背后有另一个考虑LLM 生成的 SQL 可能自带 LIMIT 也可能不自带若它自己写了很大的值或者在子查询里用了窗口函数即使外层限了行数内部计算还是可能消耗比较大。所以更严谨的做法是在数据库账号层面就设置资源限制比如 MySQL 的 max_execution_time 或 PostgreSQL 的 statement_timeout。SQLite 场景下只读 外层 LIMIT 已经够用。执行完后结果要转成 Agent 能读的格式。我一般转成 JSON 字符串每行是一个 dict这样模型能直接依据结果做总结。如果你希望结果更紧凑也可以转成 Markdown 表格看你要下发给 Agent 主循环的内容形态。3.4 与 Agent 主循环对接工具注册与错误重试现在到了关键收口把上面的逻辑包成一个 Agent 可调用的工具函数。以下是带错误重试的完整版本import json def sql_query_tool(question: str, client, model: str, schema_text: str, db_url: str): Text-to-SQL查询工具供Agent主循环调用。 max_retries 2 last_error for attempt in range(max_retries 1): if attempt 0: sql generate_sql(question, schema_text, client, model) else: # 把上一次的报错信息反馈给模型让它自我修正 sql generate_sql_with_feedback( question, schema_text, last_error, client, model ) print(f[SQL] {sql}) try: columns, rows execute_query_readonly(db_url, sql) # 结果摘要限制token summary json.dumps( {columns: columns, rows: rows[:50]}, ensure_asciiFalse ) return { status: success, sql: sql, result: summary, row_count: len(rows) } except Exception as e: last_error str(e) print(f[SQL_ERROR] {last_error}) continue return { status: error, sql: last_error, result: fSQL执行失败: {last_error} }错误重试的思路是这样的第一遍生成的 SQL 很可能有语法错误或者字段名写错报错信息里包含了具体原因第二遍把last_error拼到 prompt 里让模型看一眼报错再改一版成功率能提升一大截。实测下来加了这一轮重试查询工具的整体成功率能从 70% 提到 90% 以上。generate_sql_with_feedback和generate_sql的区别只是在 prompt 里多了一段话我贴一下关键改动FEEDBACK_TEMPLATE 你上一轮生成的SQL执行报错错误信息如下 {error} 请根据错误信息修正SQL仍然只输出SQL代码。 def generate_sql_with_feedback(question, schema_text, error, client, model): base_prompt SYSTEM_PROMPT_TEMPLATE.format(schemaschema_text, questionquestion) prompt base_prompt FEEDBACK_TEMPLATE.format(errorerror) resp client.chat.completions.create( modelmodel, messages[ {role: system, content: 你是SQL专家。}, {role: user, content: prompt} ], temperature0 ) return resp.choices[0].message.content.strip()在这个设计里工具函数的入参是 question出参是 status、sql、result 三段。Agent 主循环拿到返回后如果 status 是 success就把 result 里的 JSON 拼到自己的上下文里再组织自然语言回答如果 status 是 error就让 Agent 如实告诉用户“查询失败了原因是什么”而不是强行编一个结果。这样整个链路的边界非常清楚。工具注册这块不同框架的写法不同但原理一致声明一个工具的名字、描述、参数 JSON Schema再把上面的函数作为执行体。比如在 OpenAI Function Calling 风格中工具的 description 可以这样写{ type: function, function: { name: sql_query_tool, description: 根据自然语言问题查询选课系统数据库返回JSON格式查询结果。, parameters: { type: object, properties: { question: {type: string, description: 用户的自然语言查询问题} }, required: [question] } } }工具描述写得越清楚Agent 在判断“要不要调用这个工具”的时候就越准。有些 Agent 会在不该查库的时候硬查比如用户只是闲聊“今天天气不错”它也触发一次数据库查询这多半就是工具描述里没有写清适用边界。所以我喜欢在 description 里加一句“仅当问题涉及数据库中的课程、学生、选课人数、成绩等数据时使用”。4. 常见问题与排查技巧实录4.1 LLM 生成 SQL 跑不通时的三种典型报错我在调这个模块时最常遇到的模型输出错误整理一下方便你排查。第一种是字段名幻觉。模型看到一个问题里有“人数”就自己编一个 student_count 字段但表里根本没有。这类报错通常是“no such column: student_count”。解决办法有两个层面一是 schema 描述要把字段含义写清二是重试机制把报错喂回去模型看到错误后大概率会改成 COUNT(*) 这类写法。第二种是 SQL 语法错误比如 SQLite 不支持模型生成的某些高级语法。比如模型生成了SELECT TOP 5 ...这是 SQL Server 语法SQLite 里要写LIMIT 5。这类问题的根源是模型对“方言”的感知不敏感我的处理是在 system prompt 里明确写“目标数据库是 SQLite请使用 SQLite 支持的语法”。如果你是 MySQL就写 MySQL 8.0 语法。第三种是聚合和 GROUP BY 的语义不对。举一个实际例子用户问“每门课程的平均成绩”模型可能写SELECT course_id, AVG(score) FROM course_enrollment GROUP BY course_id但最后又 select 了一个不在 group by 里的字段。这类问题靠规则约束效果有限我一般是把常见的“聚合查询模板”直接在 few-shot 里给出比如“求每个X的平均/最大/最小”这种句式对应什么写法模型学得很快。我整理了一个速查表方便你对照排查报错类型典型报错信息主要原因首选解法字段不存在no such column: xxx模型臆造字段补全schema说明 重试反馈方言错误near TOP: syntax error模型生成了其他方言的SQLprompt里写清数据库方言表名错误no such table: xxx模型没选对表检查schema是否包含表清单类型不匹配datatype mismatch字段类型判断错误在schema中强调字段类型权限不足attempt to write a readonly database模型生成了写操作确保使用只读连接串4.2 查询结果太多或太散怎么办就算 SQL 语法全对也可能遇到结果集大、信息密度低的问题。比如用户问“帮我看看所有学生的选课情况”模型可能真的返回 500 条明细Agent 拿到那么长的 JSON 根本没法组织回答还容易超出上下文窗口。我的处理是把结果集做两层压缩。第一层是代码层面SQL 外层加 LIMIT最多取 200 行第二层是在传给 Agent 主循环之前只保留前 50 行的 JSON 摘要并单独记录 row_count。这样 Agent 既能基于部分数据给用户一个概要回答又能知道总行数如果需要明细可以再追问。还有一个实战技巧对高频出现的“大数据量”查询场景与其让模型自由发挥不如在 few-shot 里给一条聚合示例。比如用户问“所有学生的选课情况”正确做法是返回每个学生选了几门课而不是逐行列出选课记录。你在 prompt 里给一个示例模型会自动朝聚合方向理解。4.3 避免 Agent 被数据库错误带偏最后一个经常被忽略的问题是Agent 拿到 SQL 报错后可能会把技术细节原封不动丢给用户或者更糟自己脑补一个答案。比如 SQL 执行报错Agent 直接跟用户说“数据库出现了一个 no such column 错误”——这体验很糟糕用户只想知道数据是什么不想看底层报错。工程上我的做法是工具返回里把 sql 和 result 分开。sql 字段可能包含敏感信息只用于调试和日志result 字段是格式化好的摘要包含 status。LLM 主循环只消费 result如果 status 是 error我会在 result 里写“查询失败请提示用户稍后重试或调整提问方式”而不是把原始异常直接塞进去。这样 Agent 就能做一层“翻译”把技术错误转化为用户友好的表达。至于原始 SQL我统一写进本地日志文件方便事后排查。这个细节看上去很小但决定了整个 Agent 的“专业感”。一个成熟的 Agent 不是永远不出错而是出错了知道怎么兜底。数据库查询能力越强越要在上层把异常处理做细。5. 再多做一步给结果加一层“自然语言总结”这个环节算是我自己加的收尾方案。Text-to-SQL 工具如果只把 JSON 结果抛给 Agent用户问“哪个老师教的课最多”时Agent 需要自己从 JSON 里去数、去比较、去组织语言这一步在大模型看来并不难但容易出错尤其当结果里有多个课程、排序逻辑复杂时。我的做法是让查询工具内部再做一个 summarize 步骤拿到 JSON 结果之后调用一次 LLM把“用户问题 SQL 查询结果”压缩成一段简明结论再把这个结论返回给 Agent 主循环。第一次调用是 Text-to-SQL第二次调用是 Text-to-Summary。多一次调用多几秒延迟和一点 token 成本但用户体验提升是质的。用户问“选修人数最多的课是什么”Agent 第一次得到的是 SQL 结果第二次直接得到“选修人数最多的课是《数据结构》共 128 人选修”这个词就是给用户看的。你可能会觉得 Agent 主循环自己也能做总结为什么要放在工具层我的经验是工具层做完总结之后返回主循环的内容更短更准主循环不再需要“理解数据”只需要“转述结论”这会大幅降低 Agent 在长上下文里的出错率。而且工具层的总结逻辑可以单独调试不影响主循环的设计。当然这个总结步骤需要额外传一次模型调用如果你的场景对延迟特别敏感也可以跳过。但如果你在做的是面向真实用户的产品我还是强烈建议加上它的收益远超成本。最后再分享一个我在调这个模块时的个人习惯每次给数据库 schema 加字段、改表结构之后我都会跑几个固定的测试问题比如“每门课的选课人数”“成绩最高的学生是谁”“哪个学院的学分平均分最高”确保改动没有破坏 prompt 里对结构的描述。久而久之这套东西就成了一个“回归测试集”后来我把它写成了自动化脚本每次改动后自动跑一遍非常省心。Text-to-SQL 看起来是一个小能力但它把 Agent 从“能说”推到了“能做事”的关键一步。只要把 Schema 注入、SQL 生成、安全执行、失败重试这四个环节理顺你的 Agent 就算真正补上了数据库查询这条腿。后续想往深做可以继续研究多轮查询的上下文衔接、表格结果的流式展示、以及对复杂嵌套查询的约束生成——这些都是建立在这一篇基础链路之上的事了。