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 里固定写了几条强约束，实测下来很稳：

1. 只允许执行 SELECT 查询，禁止任何 INSERT、UPDATE、DELETE、DROP、ALTER 语句。
2. 查询必须使用 LIMIT，默认不超过 200 行。
3. 如果用户的问法语义不明确，需要区分“最值”和“明细”，比如“课程号是 CS101 的选课人数”是明细聚合，“哪门课选的人最多”是分组排序。
4. 不要把表名字段名翻译成中文，保持原样；别名可以用简单字母。
5. 不确定字段含义时，使用 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 sqlalchemy

SQLite 的好处是单文件、零配置，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(schema=schema_text, question=question) resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "你是SQL专家。"}, {"role": "user", "content": prompt} ], temperature=0 ) 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 = f"SELECT * 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?mode=ro这样的 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_ascii=False ) 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": f"SQL执行失败: {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(schema=schema_text, question=question) prompt = base_prompt + FEEDBACK_TEMPLATE.format(error=error) resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "你是SQL专家。"}, {"role": "user", "content": prompt} ], temperature=0 ) 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的平均/最大/最小”这种句式对应什么写法，模型学得很快。

我整理了一个速查表，方便你对照排查：

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 就算真正补上了数据库查询这条腿。后续想往深做，可以继续研究多轮查询的上下文衔接、表格结果的流式展示、以及对复杂嵌套查询的约束生成——这些都是建立在这一篇基础链路之上的事了。