做Agent开发有一段时间的朋友应该都听过“技能Skill”这个词。Agent本身再聪明没有工具也只能“纸上谈兵”而能让AI直接操作数据库、把大白话变成查询结果的SQL Executor恰恰是打通“对话”和“数据”之间最短路径的一个关键技能。这一篇我会围绕“自然语言转SQL的智能查询”这个核心把SQL Executor技能从设计思路、环境准备、代码实现到问题排查完整拆一遍记录一下我在实际项目中把它跑通、用稳的过程。适合正在做AI Agent开发、想做智能问数工具或者单纯对LangChain、MCP这类技术栈感兴趣的朋友这篇内容可以直接参考落地。先说结论SQL Executor不是一个简单的“调用数据库的工具”它更像是一个“能看懂用户意图、还能自己纠错”的数据库助手。用户说一句“帮我查一下上个月各区域的销售额排名”它不是把这句话原封不动丢给数据库而是先把这句话翻译成一条正确的SQL然后执行再把结果用通俗的话讲给用户听。整个过程涉及大模型的推理能力、SQL语法的约束、数据库Schema的理解以及结果格式化的表达任何一个环节偷懒最后出来的体验都会很拉胯。我这次实现的SQL Executor技能核心流程分四步意图识别与Schema获取、自然语言转SQL、SQL安全校验与执行、结果格式化返回。下面我按这个顺序把每一步的细节和踩过的坑都写出来。1. 技能设计与整体思路1.1 为什么需要SQL Executor技能在很多AI Agent场景里我们最常遇到的需求就是“用户想从系统里拿数据”。以前的做法是让用户自己去看报表、自己去后台翻或者让技术人员手写SQL查询。但现在有了大模型我们可以让用户用最自然的话提问比如“这个月退单最多的是哪些商品”这种问题放在传统模式里需要一个会SQL的人去写代码而SQL Executor技能做的是把这一步自动化。你可能会问“直接用大模型生成SQL然后执行不就行了”实际远没那么简单。生产环境里的SQL不是随便生成的你既要保证表名、字段名不对又要防止用户有意无意地输入破坏性语句还要考虑执行超时、结果集过大等问题。SQL Executor技能的本质就是把“自然语言转SQL”和“安全执行”这两个能力封装成一个标准工具供Agent编排调用。它扮演的是AI和数据库之间“翻译官安全员”的双重角色。我在实际设计中把它做成了Agent的一个独立工具函数而不是直接把数据库连接暴露给大模型。这样做的原因是工具函数可以附加校验逻辑、日志记录、错误处理而直接暴露连接则会把所有风险都摊在桌面上。1.2 技术选型背后的考量在做技术选型时我对比过几种主流方案方案优点缺点适用场景纯提示词SQLAlchemy实现简单依赖少大模型可能生成不存在的字段名快速验证原型LangChain SQL Agent官方封装自带执行链可定制性差出错时排查麻烦标准业务查询自研工具LangGraph全流程可控安全校验灵活需要写更多代码生产级复杂业务MCP SQL Server标准化协议未来趋势生态还不够成熟跨平台Agent互操作考虑到项目要做成“生产可用级别”我最终选了“自研工具函数LangGraph编排”。这个选择基于几个主要的考虑第一SQL执行场景千差万别有的要分页有的要聚合有的要联表查询完全依赖框架默认行为往往满足不了业务需求第二安全校验这个环节极为重要任何通用的框架都不会给你定制化到“只允许SELECT且必须强制LIMIT”必须自己在代码层实现第三LangGraph可以让后续扩展多Agent协作更灵活比如让“SQL Executor”和“数据可视化Agent”串联起来。不过我也得坦白说一句如果只是做个Demo或者内部小工具用LangChain自带的SQLDatabaseToolkit会更省事。它内置了查表、查Schema、执行SQL等一整套工具十几行代码就能跑起来。反而是像我这样完全自研前期开发量会大一些但后期可控性极好一旦跑顺了后维护成本反而更低。2. 环境准备与Schema管理2.1 依赖安装与数据库连接配置先说一下我使用的环境Python 3.10数据库用的MySQL 8.0ORM和连接层用SQLAlchemy 2.0Agent框架用LangChain LangGraph。基础的依赖包如下需要版本对齐避免后续踩到API变更的坑langchain0.2.* langchain-openai0.1.* langgraph0.1.* sqlalchemy2.0.* pymysql1.1.* pydantic2.* tabulate0.9.*数据库连接配置这里有一个很关键的点——建议单独建一个低权限账号给Agent用。我在项目里专门建了agent_query账号只授予SELECT权限连INSERT、UPDATE都不给。这样即便提示词被恶意注入最坏的后果也只是读数据不会把数据搞坏。这一点非常重要后面安全部分还会细讲。-- MySQL创建低权限账号示例 CREATE USER agent_query% IDENTIFIED BY your_password; GRANT SELECT ON your_database.* TO agent_query%; FLUSH PRIVILEGES;连接串我放在了环境变量里import os from sqlalchemy import create_engine DB_USER os.getenv(DB_USER, agent_query) DB_PASSWORD os.getenv(DB_PASSWORD, ) DB_HOST os.getenv(DB_HOST, localhost) DB_PORT os.getenv(DB_PORT, 3306) DB_NAME os.getenv(DB_NAME, your_database) engine create_engine( fmysqlpymysql://{DB_USER}:{DB_PASSWORD}{DB_HOST}:{DB_PORT}/{DB_NAME}?charsetutf8mb4, pool_size5, max_overflow10, pool_timeout30, pool_recycle1800, echoFalse )2.2 让大模型“看懂”数据库结构大模型天生不知道你的数据库长什么样它必须知道有哪些表、每个表有哪些字段、字段含义是什么才能生成正确的SQL。所以在真正执行自然语言转SQL之前需要先把数据库的Schema信息“喂”给模型。我实现了一个get_schema_info()函数通过查询information_schema把表和字段信息拉出来并格式化成大模型能理解的文本结构。核心代码如下def get_schema_info(engine) - str: 获取数据库Schema信息格式化为文本返回给LLM。 这一步是SQL Executor技能的“眼睛”决定了LLM对数据结构的理解程度。 schema_lines [] with engine.connect() as conn: # 查询所有表 tables conn.execute( text(SELECT TABLE_NAME, TABLE_COMMENT FROM information_schema.TABLES WHERE TABLE_SCHEMA :db), {db: DB_NAME} ).fetchall() for table_name, table_comment in tables: schema_lines.append(f表名: {table_name} (注释: {table_comment})) # 查询每个表的字段信息 columns conn.execute( text(SELECT COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT, IS_NULLABLE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA :db AND TABLE_NAME :tbl), {db: DB_NAME, tbl: table_name} ).fetchall() for col in columns: schema_lines.append( f - {col.COLUMN_NAME} ({col.COLUMN_TYPE}, f{可空 if col.IS_NULLABLE YES else 非空}), 注释: {col.COLUMN_COMMENT or 无} ) return \n.join(schema_lines)这样返回的Schema文本会拼接进提示词里让模型知道有哪些表可以用。注意一点生产环境的表可能有几十上百个全量塞入提示词会导致上下文过长且模型注意力分散。建议做一次“Schema裁剪”根据用户问题的关键词先做一个粗粒度检索只把相关的几张表信息拼进去。比如用户问“销售额”你只需要把包含“销售”“订单”“金额”的表信息放进去其余表不用让它看到。这种检索式Schema增强对查询准确率提升非常明显。3. 核心实现自然语言转SQL的完整链路3.1 提示词模板的精细化设计自然语言转SQL本质上是“让模型在受限的Schema范围内做文本生成”。提示词的质量直接决定了SQL的正确率。我设计提示词时重点做了三件事角色设定、规则约束、示例强化。SQL_GENERATION_PROMPT 你是一个专业的SQL查询助手。请根据用户的问题结合给定的数据库Schema信息生成一条符合业务逻辑的SQL查询语句。 数据库Schema信息 {schema_text} 查询要求 1. 只允许生成SELECT查询语句禁止生成INSERT、UPDATE、DELETE、DROP、ALTER等任何会修改数据或结构的语句。 2. 如果用户的问题涉及时间范围默认优先使用最近30天除非用户明确指定了时间。 3. 所有字段名和表名必须严格使用Schema信息中提供的名称不允许臆造字段。 4. 如果用户想查询的是排名或TopN请在SQL中使用LIMIT子句。 5. 永远不要在SELECT中使用 SELECT *必须显式列出需要的字段。 6. 如果用户问题表述模糊无法确定业务含义时请优先给出最合理的SQL并附加备注说明你的假设。 7. 只输出SQL语句本身不要输出Markdown代码块标记不要附加任何解释文字。 参考示例 问题: 查询最近一周每天的订单数 SQL: SELECT DATE(create_time) AS order_date, COUNT(*) AS order_count FROM orders WHERE create_time DATE_SUB(CURDATE(), INTERVAL 7 DAY) GROUP BY DATE(create_time) ORDER BY order_date; 问题: 统计各商品类目的总销售额按销售额降序 SQL: SELECT c.category_name, SUM(o.amount) AS total_sales FROM orders o INNER JOIN category c ON o.category_id c.id GROUP BY c.category_name ORDER BY total_sales DESC; 用户问题{user_query} SQL: 这个提示词能稳定工作的关键是“输出格式极简”和“规则明确”这两件事。我见过不少朋友把提示词写得特别长、特别复杂反而模型容易跑偏。SQL生成的提示词应当像“高考作文要求”那样简洁明确把红线列出来剩下空间让模型自己发挥。我做完之后测过很多次按要求走的SQL生成错误率低于10%但一旦提示词里少了“明确字段来自Schema”这一句错误率马上翻倍。3.2 安全校验与SQL执行封装模型生成SQL以后绝对不能直接扔进数据库跑。我分了三道防线来拦截问题SQL第一道防线正则快速拦截关键词第二道防线SQL语法解析校验检查是否为纯SELECT第三道防线执行环境层面的防护兜底下面是我封装的安全校验加执行函数import re import time from sqlalchemy import text from sqlalchemy.exc import SQLAlchemyError # 危险关键字黑名单用于第一道防线 BLOCK_KEYWORDS [ INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, TRUNCATE, REPLACE, GRANT, REVOKE, RENAME, INTO OUTFILE, INTO DUMPFILE, LOAD_FILE, SLEEP, BENCHMARK, INFORMATION_SCHEMA, ] def validate_sql(sql: str) - bool: 校验SQL是否为安全的SELECT查询。 返回True表示通过False表示不通过并记录日志。 if not sql or not sql.strip(): return False sql_upper sql.upper().strip() # 1. 必须以SELECT开头允许WITH子句 if not (sql_upper.startswith(SELECT) or sql_upper.startswith(WITH)): return False # 2. 黑名单关键词检测 for kw in BLOCK_KEYWORDS: if kw in sql_upper: return False # 3. 分号检查防止多语句注入 if sql_upper.count(;) 1: return False return True def execute_sql(engine, sql: str, max_rows: int 200): 安全执行SQL控制返回行数和执行超时。 返回结构化结果或错误信息。 if not validate_sql(sql): return { success: False, message: SQL验签未通过检测到非SELECT语句或危险关键字已拒绝执行。 } try: # 检查关键强制加LIMIT防止用户/模型忘记限制行数 sql_upper sql.upper().rstrip().rstrip(;) if LIMIT not in sql_upper: # 如果SQL本身是子查询包裹的复杂语句这里直接拼LIMIT会有风险 # 所以在提示词里强制要求模型生成LIMIT这里仅兜底 if sql_upper.startswith(SELECT) and LIMIT not in sql_upper.split(ORDER)[-1]: sql LIMIT str(max_rows) # 记录执行开始时间 start_time time.time() # 设置语句级超时超过10秒直接终止 with engine.connect() as conn: conn.execute(text(SET SESSION MAX_EXECUTION_TIME10000)) result conn.execute(text(sql)) # 只取前max_rows1条多出的那条用于提示用户结果被截断 rows result.fetchmany(max_rows 1) columns list(result.keys()) elapsed round(time.time() - start_time, 2) # 判断是否有截断 truncated len(rows) max_rows if truncated: rows rows[:max_rows] return { success: True, columns: columns, rows: rows, row_count: len(rows), truncated: truncated, elapsed: elapsed } except SQLAlchemyError as e: return { success: False, message: fSQL执行失败: {str(e)} }这里有个很值得说的细节在执行前我用MAX_EXECUTION_TIME把语句级执行时间卡死在了10秒。原因很简单在真实业务中如果模型生成了一条漏了索引的慢查询可能会把数据库拖到卡死。有了这个参数即使SQL有问题最多10秒就会被数据库自己终止不至于影响其他业务。另外一个兜底是fetchmany(max_rows 1)多取一条用来判断是否需要提示用户“结果过多已截断”这样用户体验会好很多。3.3 执行结果自动格式化SQL执行出来的原始结果是列表套元组的结构直接丢给用户看既不直观也不优雅。我在技能里加了一个“结果格式化”层把原始结果转换成Markdown表格这样Agent拿到了之后可以直接把表格拼进回复内容里。from tabulate import tabulate def format_results(result: dict) - str: 将SQL执行结果格式化为Markdown表格文本。 if not result.get(success): return result.get(message, 查询失败) columns result[columns] rows result[rows] if not rows: return 查询执行成功但没有找到符合条件的数据。 # 用tabulate生成markdown格式表格 table_str tabulate(rows, headerscolumns, tablefmtgithub, maxcolwidths50) # 拼接总结信息 summary f\n\n共返回 {result[row_count]} 条记录耗时 {result[elapsed]} 秒。 if result.get(truncated): summary \n\n 提示结果数据量超过限制仅展示前200条如需完整数据请补充更精确的筛选条件。 return table_str summary做成表格还有一个额外的好处如果后续接了“图表生成Agent”它可以直接解析这个表格结构去绘图而不用重新理解原始数据。我在实际项目中就是把SQL Executor的输出作为“可视化技能”的输入用户问完数据之后甚至可以接一句“画个柱状图看看”Agent会通过编排把表格数据传给画图工具整个链路非常顺滑。4. 错误修正机制与多轮对话4.1 让SQL Executor学会自我纠错这是我觉得最值得分享的一个设计点。大模型生成的SQL第一次就能跑通的概率我实测下来大概在70%到85%之间剩下的情况多半是表名字段名写错、语法差一点点、或者业务理解有偏差。以前的做法是报错就结束让用户换个问法重试体验很差。后来我参照RAG里的“Self-Correction”思路在技能里加了一个“错误反馈循环”SQL执行失败时把数据库返回的报错信息、出错的SQL、以及用户原始问题一起打包回传给模型让模型根据错误信息修正SQL后再试一次。最多重试两次如果还不行就放弃并给用户一个友好的提示。def run_sql_with_retry(engine, sql: str, user_query: str, retries: int 2): 带自我纠错机制的SQL执行流程 第一次执行失败后将错误信息回传大模型进行修正最多重试retries次。 current_sql sql last_error for attempt in range(retries 1): # 先执行 result execute_sql(engine, current_sql) if result.get(success): return result, current_sql last_error result.get(message, 未知错误) if attempt retries: # 让语言模型根据报错信息修正SQL correction_prompt f 用户的问题是{user_query} 你之前生成的SQL存在执行错误错误信息如下 {last_error} 请分析错误原因修正这条SQL只输出修正后的SQL语句本身不要输出任何解释。 错误SQL参考不要直接使用{current_sql} current_sql llm.predict(correction_prompt).strip() # 去掉可能生成的Markdown代码块标记 if current_sql.startswith(): current_sql current_sql.strip() if current_sql.startswith(sql): current_sql current_sql[3:] # 重试次数已用完返回最后一次的错误信息 return {success: False, message: last_error}, current_sql有一个细节需要注意修正过程中一定要把“用户原始问题”和“错误SQL”都传给模型而不是只传错误信息。因为只有原始问题在模型才能理解原本的查询意图如果只给它报错SQL它可能只是机械地修语法修完了还是不符合业务要求。我见过不少项目在这个地方偷懒最后模型改来改去还是在同一个坑里绕圈。这个机制上线后我这边生成SQL的一次执行成功率从原来的78%左右提升到了91%效果非常明显。4.2 上下文记忆与多轮查询串联单个查询好做多轮对话就麻烦一些。用户可能会先问“销售情况怎么样”然后紧接着问“那华东区呢”。如果没有上下文记忆第二个问题就会被模型理解成完整的句子模型会因为缺少“对比的主体”而生成错误的SQL。所以我在Agent状态里加了一个“最近查询上下文”的存储每一次成功执行的SQL和结果摘要都会存下来后续的问题可以引用。具体实现时我是在执行新SQL之前把历史对话的最后两轮关键信息拼接进提示词里。比如context_prompt f 历史查询记录 用户上次问题{last_query.user_question} 上次SQL{last_query.sql} 上次查询意图{last_query.intent_summary} 当前用户问题{user_query} 请基于历史查询上下文理解用户意图如果当前问题依赖历史结果请务必基于上次查询的结果做进一步分析。 这个设计在做“数据追问”场景时特别好用。用户连续问“各品类销售额排名”“那前三个品类最近一个月的趋势呢”Agent能准确理解“前三个品类”指的是上一次查询结果中的前三名而不是随机选三个品类。这种多轮联动的体验是SQL Executor从“玩具级”走向“可用级”的关键一步。5. 常见问题与排查技巧实录5.1 高发问题速查表我这套SQL Executor技能经历了从Demo到生产的迭代过程中碰到过很多问题。我整理了一份高频问题速查表基本覆盖了我遇到的大部分坑问题现象根本原因解决方案生成的SQL字段名不存在Schema信息缺失或模型臆造字段在提示词中强制说明“字段名必须来自Schema文本”并按问题关键词裁剪Schema查询结果为空但数据其实存在时间范围默认逻辑错误检查提示词中的时间默认值设定建议明确“如果问题没有指定时间默认查最近30天”用户问题涉及同义词表名模型分不清业务实体对应的表在Schema文本中补充“业务实体说明”给表加别名注释执行报错“Unknown column”联表查询时字段归属不清晰修正提示词要求模型生成SQL时始终用“表别名.字段名”的格式结果集过大卡死缺少LIMIT限制强制提示词必须加LIMIT并在代码里做二次兜底连续追问时意图漂移缺少上下文记忆在Agent状态中记录历史SQL和查询意图多轮对话时拼接上下文模型生成的SQL带Markdown代码块提示词输出格式约束不够在解析结果时统一去除反引号和sql标记双保险并发查询量大导致连接池耗尽连接池参数不合理调大max_overflow并设置合理的pool_timeout增加只读副本分流5.2 我踩过的三个典型坑第一个坑是表名大小写问题。我一开始在Linux环境上跑MySQL默认表名大小写敏感但模型生成的SQL清一色是小写表名结果频繁报错“Table doesnt exist”。后来我把Schema信息里原始表名原样展示出来并在提示词里加了“表名必须和Schema信息中的大小写完全一致”问题才解决。如果不想这么麻烦也可以给MySQL设置lower_case_table_names1但要注意这会影响后续建表命名规范最好在项目初期就定好。第二个坑是模型把用户问题里的中文条件直接塞进了SQL。比如用户问“查一下状态为已完成的订单”模型可能生成WHERE status 已完成但实际上数据库里存的是状态码1。这种业务语义映射问题光靠提示词很难完全避免。我的方案是加一个“业务前缀词表”在Schema信息里附加说明“状态字段的取值含义1已完成2进行中3已取消”让模型在生成SQL时做一步翻译。这个方法对枚举类字段特别有效。第三个坑是数据库连接长期空闲被MySQL服务端断开。开发测试时没什么感觉但Agent服务跑几天后第一次查询总是报Lost connection to MySQL server during query。这是因为wait_timeout默认8小时连接池里的连接过期了。解决也简单连接串里加上pool_recycle1800让连接每30分钟回收一次问题彻底消失。5.3 性能优化与慢查询治理SQL Executor的上线很自然地会带来一批新的数据库负载压力。因为用户问的问题千奇百怪很可能触发很多没有索引的全表扫描。我的建议是双管齐下在应用层除了前面说的MAX_EXECUTION_TIME兜底还应该给Agent加一层“查询频率限制”同一个用户短期内不要让它反复执行结构相同的SQL。在数据库层开启慢查询日志定期把Agent产生的慢SQL捞出来分析针对高频慢查询场景提前在相关字段上建好索引。# 查看MySQL慢查询是否开启 SHOW VARIABLES LIKE slow_query_log; # 开启慢查询日志记录执行时间超过2秒的SQL SET GLOBAL slow_query_log ON; SET GLOBAL long_query_time 2;这个机制跑一段时间后AI查询对核心业务库的负载影响基本能控制在可接受范围内。如果数据量上了千万级强烈建议为Agent单独准备一个只读从库避免临时性的统计查询打满主库IOPS影响线上交易系统。6. 进阶优化方向6.1 注入防御与权限边界再收紧虽然SQL Executor本质上只是个读查询工具但我还是建议把安全级别拉满。除了前面提到的低权限账号、黑名单关键字、LIMIT强制之外我后来又给系统加了一层“查询白名单”机制某些核心业务表不允许Agent直接查询只能通过预定义的视图去查。比如用户想查用户的手机号Agent被引导去查v_user_basic_info视图这个视图已经过滤掉了敏感字段。这比在应用层做过滤要彻底得多因为权限是数据库层面强制管控的应用层被绕过也没有用。6.2 从单技能走向多技能Agent协作把SQL Executor做成一个独立技能之后最大的好处是它可以被任意Agent编排调用。我现在这套系统里SQL Executor已经不是一个孤立的工具而是和另一个“意图路由Agent”配合工作意图路由Agent先判断用户问的是“数据查询类”还是“闲聊类”还是“操作类”如果是数据查询类就调用SQL Executor查询结果出来后再根据用户后续指令决定是否交给“图表生成Agent”进一步加工。这种多技能协同让Agent系统的结构变得灵活得多每个技能各司其职互不干扰后续每新增一个技能也只需要在路由层多配一个入口。6.3 向量检索增强的Schema选择上面提到过Schema裁剪我用的是基于关键词匹配的简单方式。如果数据库表特别多、表结构特别复杂我更推荐用向量检索来做Schema召回把每张表的字段注释、示例数据转成向量用户提问时对问题进行向量化然后从Schema向量库里找出最相关的表结构再拼入提示词。这样哪怕用户问题的表达方式和Schema注释完全不同也能通过语义相似度检索到正确表结构。比如用户问“上个月赚了多少钱”它照样能把orders.amount这张表的字段捞出来。这个优化方向我个人认为是自然语言转SQL从“能查”走向“好查”的很关键一步。写在最后做SQL Executor这段时间我最大的感受是它看着只是个“生成SQL再执行”的小工具但实际上牵涉到提示词工程、数据库运维、安全防护、Agent编排每一块都要考虑得很细。我自己踩过几个坑后有个很深的体会永远不要信任模型生成的SQL也不要信任用户输入的每一句话。把能力给模型的同时边界和底线一定要自己在代码层焊死低权限账号、关键字拦截、执行超时、LIMIT兜底这几个动作一个都不能少。另外一个感受是自然语言转SQL要想做好光靠模型本身不够Schema的信息密度和表达方式才是“隐藏的胜负手”。你给模型喂的Schema越清晰、越规范它生成SQL的准确率提升就越明显。这套SQL Executor的技能实现后续我还会继续扩展目前正在打算把“查询失败原因分析”做成一个独立的小Agent把数据库慢查询日志和模型生成过程都记录下来逐步形成一套专属的“问题样本库”用来做微调或者few-shot示例。这样Agent会越用越聪明查询准确率也会逐步回升。如果你也在做类似方向欢迎多交流实际经验和踩坑记录。