企业官网建设流程全解析

做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 技术选型背后的考量

在做技术选型时，我对比过几种主流方案：

考虑到项目要做成“生产可用级别”，我最终选了“自研工具函数+LangGraph编排”。这个选择基于几个主要的考虑：

• 第一，SQL执行场景千差万别，有的要分页，有的要聚合，有的要联表查询，完全依赖框架默认行为往往满足不了业务需求；
• 第二，安全校验这个环节极为重要，任何通用的框架都不会给你定制化到“只允许SELECT且必须强制LIMIT”，必须自己在代码层实现；
• 第三，LangGraph可以让后续扩展多Agent协作更灵活，比如让“SQL Executor”和“数据可视化Agent”串联起来。

不过我也得坦白说一句，如果只是做个Demo或者内部小工具，用LangChain自带的SQLDatabaseToolkit会更省事。它内置了查表、查Schema、执行SQL等一整套工具，十几行代码就能跑起来。反而是像我这样完全自研，前期开发量会大一些，但后期可控性极好，一旦跑顺了后维护成本反而更低。

2. 环境准备与Schema管理

2.1 依赖安装与数据库连接配置

先说一下我使用的环境：Python 3.10+，数据库用的MySQL 8.0，ORM和连接层用SQLAlchemy 2.0，Agent框架用LangChain + LangGraph。基础的依赖包如下，需要版本对齐，避免后续踩到API变更的坑：

langchain==0.2.* langchain-openai==0.1.* langgraph==0.1.* sqlalchemy==2.0.* pymysql==1.1.* pydantic==2.* tabulate==0.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( f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4", pool_size=5, max_overflow=10, pool_timeout=30, pool_recycle=1800, echo=False )

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_TIME=10000")) result = conn.execute(text(sql)) # 只取前max_rows+1条，多出的那条用于提示用户结果被截断 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": f"SQL执行失败: {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, headers=columns, tablefmt="github", maxcolwidths=50) # 拼接总结信息 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到生产的迭代，过程中碰到过很多问题。我整理了一份高频问题速查表，基本覆盖了我遇到的大部分坑：

5.2 我踩过的三个典型坑

第一个坑，是表名大小写问题。我一开始在Linux环境上跑，MySQL默认表名大小写敏感，但模型生成的SQL清一色是小写表名，结果频繁报错“Table doesn't exist”。后来我把Schema信息里原始表名原样展示出来，并在提示词里加了“表名必须和Schema信息中的大小写完全一致”，问题才解决。如果不想这么麻烦，也可以给MySQL设置lower_case_table_names=1，但要注意这会影响后续建表命名规范，最好在项目初期就定好。

第二个坑，是模型把用户问题里的中文条件直接塞进了SQL。比如用户问“查一下状态为已完成的订单”，模型可能生成WHERE status = '已完成'，但实际上数据库里存的是状态码1。这种业务语义映射问题，光靠提示词很难完全避免。我的方案是加一个“业务前缀词表”，在Schema信息里附加说明“状态字段的取值含义：1=已完成，2=进行中，3=已取消”，让模型在生成SQL时做一步翻译。这个方法对枚举类字段特别有效。

第三个坑，是数据库连接长期空闲被MySQL服务端断开。开发测试时没什么感觉，但Agent服务跑几天后，第一次查询总是报Lost connection to MySQL server during query。这是因为wait_timeout默认8小时，连接池里的连接过期了。解决也简单，连接串里加上pool_recycle=1800，让连接每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会越用越聪明，查询准确率也会逐步回升。如果你也在做类似方向，欢迎多交流实际经验和踩坑记录。