你可能早就遇到过这种情况业务同学想拉一份数据又不好直接给他数据库权限于是需求单排到你手里从“帮我查一下最近30天每个渠道的订单量和GMV”到“分析一下这季度复购用户的消费习惯”来回沟通、写SQL、跑数、返工一上午就没了。SQL Agent要解决的就是让这些高频但重复的取数需求通过自然语言直接变成SQL并跑出结果。我基于大型语言模型和Python生态里的主流LLM框架做了一版SQL代理主流程已经能覆盖日常90%以上的查询场景这篇文章把我从架构选型到落地踩坑的完整过程拆开讲。这篇内容适合正在做Text-to-SQL、内部数据分析工具或者打算把LLM接进公司数据链路的人看。我会把“为什么用Agent而不是单纯Prompt生成SQL”“工具层怎么设计”“系统提示词怎么写”“执行安全怎么兜底”这些关键决策讲清楚并附上可复现的代码片段和问题排查表。1. 先想清楚你要做的是“查询助手”还是“决策Agent”1.1 SQL Agent和“一次性生成SQL”根本不是一回事最初我图省事直接设计一个Prompt让LLM根据用户问题返回SQL然后丢给数据库执行。这种方案在简单场景下没问题一旦涉及多表关联、子查询、需要确认字段口径或者用户表达有歧义模型很容易生成一张“看起来对但实际上跑错”的SQL。后来我把架构切换成Agent模式模型不再是一步到位的“翻译器”而是被包装成一个可以多步推理、自主调用工具的循环体。这个循环体每一步都会经历“思考 → 选择工具 → 拿到结果 → 再思考”的过程直到它认为自己有足够信息回答用户的问题。这是个本质区别。前者是“闭卷考试”后者是“开卷考试允许查资料、试错、修正”。对于数据库操作这种对准确性要求极高的场景开卷远比闭卷可靠。1.2 整体架构拆成三块脑、手、数据我的最终架构可以抽象成三层调度层脑负责理解用户问题、拆解任务、维护对话状态调用下一步动作。工具层手暴露给模型的一组可调用函数包括查看表清单、查看表结构、执行只读SQL、格式化结果。数据层数据源存放真实业务数据的数据库我这次用的是MySQL理论上PostgreSQL、ClickHouse也能用同样模式接入。调度层是核心但很多人花大量时间调模型却忽略了工具层和数据层的设计。实际上Agent表现崩不崩70%的锅在工具定义和系统提示词只有30%在模型本身。1.3 为什么选这个方案三个绕不开的现实原因选Agent架构不是赶时髦是三个现实约束逼出来的。第一业务库的Schema很复杂。我们有几十张表字段命名习惯不统一靠一次Prompt让模型“猜”字段含义命中率很低。Agent可以在推理过程中主动拉取某个表的建表语句再结合问题里的关键词判断是否用对了表。第二用户的自然语言大多是有歧义的。“最近30天”是按下单时间还是支付时间“用户数”是去重口径还是累计值这些问题模型自己无法判断Agent可以针对缺省信息向用户追问或者从few-shot示例里找到相似表达。第三SQL是允许试错的语言。Agent拿到了查询结果如果发现是空集或报错可以把错误信息回传给模型进行自纠正。这就像人写SQL时先跑一遍报错了看一眼提示再改。这个自纠能力是普通“生成SQL再执行”方案给不了的。2. 技术选型与依赖准备工具不要多顺手最重要2.1 底座Python 主流LLM应用编排框架我个人用的是Python这是Agent生态最成熟的语言没有之一。LLM框架这块我不建议自己从头写循环逻辑虽然难度不大但Debug成本高直接用现成的Agent编排框架能省很多事。框架选择上我对比过几类列在下面供参考方案优点缺点适合场景通用Agent编排框架如LangChain社区活跃、Agent工具抽象成熟、文档多抽象层级多黑盒感强快速验证、需要多工具组合自己写ReAct循环可控性强、依赖少、好Debug要自己处理模型调用、记忆管理深入学习、需要定制复杂交互偏轻量的Function Calling封装结构清晰、少依赖多步推理能力薄弱简单查询场景、不需要Agent我最后选的是通用框架但把它当成一个“带工具的函数调用器”来用没有用太多高级抽象。你如果只是想快速出活也可以直接起步但一定要花时间理解它的Agent执行流程不然后面出问题很难定位。2.2 模型选择与调用配置模型方面有条件的话直接上支持Function Calling的大模型接口效果比“把工具描述塞进Prompt”要稳得多。我实测下来函数调用机制把工具名、参数Schema结构化传给模型模型返回的调用请求也严格遵循JSON结构比纯文本解析靠谱好几个档次。配置上有几个容易被忽略的点温度参数尽量设为0或接近0SQL生成不需要创造力温度高了容易产生结构飘忽的SQL。max_tokens不能太小复杂的多表Join SQL可能超过500个token截断会导致SQL语法不完整。建议至少给到1000以上。调用频率限制要注意Agent一次复杂查询可能调用5到8次模型接口如果内部系统有频控需要预留缓冲。2.3 数据库连接与安全边界连接数据库这块我用的是SQLAlchemy统一管理连接底层引擎选用PyMySQL。选SQLAlchemy的原因一是连接池好管理二是在测试阶段可以方便地切换成SQLite做本地验证不污染真实数据。这里我要特别强调一个设计原则代理能连接的数据库账号必须是只读权限。所有DDL、DML操作从根上切断这不是技术问题是安全底线。我建了一个专用账号只授予SELECT权限连临时表的创建权限都没给。不要图省事用管理员账号跑Agent鬼知道模型会不会“灵机一动”生成一条DELETE语句。3. 核心实现从零搭一个能跑通的SQL代理3.1 先给Agent立规矩系统提示词设计系统提示词决定了Agent的“行为底色”。我改了无数版最后沉淀下来一份核心提示词你可以直接参考system_prompt 你是一个数据库SQL代理负责将用户的自然语言问题转化为SQL并执行查询。 你需要遵循以下规则 1. 先理解问题再选择工具。不要在没有查看表结构的情况下凭空生成SQL。 2. 查询流程固定为 - 调用list_tables获取所有表名 - 调用get_schema获取相关表的建表语句和采样数据 - 确认字段含义和关联键后调用execute_sql执行只读查询 3. 如果用户问题存在歧义比如时间范围、去重口径、维度归属先向用户追问不要擅自假设。 4. 如果SQL执行报错把报错信息作为观察结果结合表结构修正SQL最多重试3次。 5. 所有回答必须以自然语言总结查询结果不允许直接输出大段SQL代码除非用户明确要求。 6. 严禁执行任何非SELECT开头的SQL语句。 第3条和第4条是真正的点睛之笔。很多Agent“看起来很聪明但结果不靠谱”就是因为遇到歧义时默默做了假设遇到报错时直接放弃或乱编。这两条规则能把很多失败模式提前拦下来。3.2 给Agent配上“手”工具函数定义工具层是整个Agent的“手感”所在工具定义得好不好直接影响模型能不能正确调用。我设计了三个核心工具import json import sqlalchemy as sa # 工具1列出所有表 def list_tables() - list[str]: 返回数据库中所有业务表的名称列表 inspector sa.inspect(engine) return inspector.get_table_names() # 工具2获取表结构 def get_schema(table_name: str) - str: 获取指定表的建表语句、字段注释和3条采样数据 inspector sa.inspect(engine) columns inspector.get_columns(table_name) col_desc [] for col in columns: comment col.get(comment, ) col_desc.append(f{col[name]} {col[type]} COMMENT {comment}) sample_sql sa.text(fSELECT * FROM {table_name} LIMIT 3) with engine.connect() as conn: samples conn.execute(sample_sql).fetchall() schema_text CREATE TABLE table_name (\n ,\n .join(col_desc) \n);\n schema_text 样例数据\n json.dumps([dict(row._mapping) for row in samples], ensure_asciiFalse, defaultstr) return schema_text一个小细节get_schema返回的不是纯建表语句而是额外带了几条采样数据。模型看到真实的数据样例对字段内容的理解会直观很多。比如有一个字段叫status建表注释写的是“状态”模型不一定能猜到0、1、2都代表什么但看到样例数据里一列的取值分布就能结合上下文推断出口径含义。这个技巧是纯Prompt工程给不了的属于“用小成本换大收益”。# 工具3执行只读SQL def execute_sql(sql: str) - dict: 执行只读SQL查询返回结果行数和前50行数据 if not sql.strip().upper().startswith(SELECT): return {error: 只允许执行SELECT语句} try: with engine.connect() as conn: result conn.execute(sa.text(sql)) rows result.fetchmany(50) columns list(result.keys()) data [dict(zip(columns, row)) for row in rows] total result.rowcount if result.rowcount ! -1 else len(rows) return {columns: columns, rows: data, total: total, truncated: total 50} except Exception as e: return {error: str(e)}注意我在工具说明里明确写了“返回前50行”这个限制很重要。如果不限制返回行数Agent在调试阶段可能一次性拉回几十万行既拖垮数据库也浪费上下文窗口。3.3 把工具注册给模型参数描述决定调用质量工具函数的Python代码写好了接下来要把它们注册为模型可调用的工具描述。这一步非常容易被低估但模型能不能正确传参很大程度取决于描述写得好不好。以下是标准写法tools [ { type: function, function: { name: list_tables, description: 获取数据库中所有业务表的名称列表。当你不确定有哪些表或需要了解整体表结构时调用。, parameters: {type: object, properties: {}} } }, { type: function, function: { name: get_schema, description: 获取指定表的建表语句、字段注释以及3条采样数据。每次开始生成SQL前必须调用该工具了解目标表结构。, parameters: { type: object, properties: { table_name: { type: string, description: 要查看的目标表名例如 orders } }, required: [table_name] } } }, { type: function, function: { name: execute_sql, description: 执行只读SQL查询返回结果集。仅支持SELECT其他语句一律拒绝。执行结果会被用于回答用户问题。, parameters: { type: object, properties: { sql: { type: string, description: 完整的SELECT SQL语句必须是只读查询 } }, required: [sql] } } } ]描述里我反复强调“必须调用”“一律拒绝”“用于回答用户问题”这都是在引导模型形成正确的行为链路。LLM本身对文本高度敏感工具描述里的措辞会影响行为概率。你不逼它查表结构它就敢跳过这步直接编SQL。3.4 主循环执行流程把整个Agent串起来框架层面的Agent循环核心逻辑其实不长。我剥掉框架封装后本质就是一个while循环messages [{role: system, content: system_prompt}, {role: user, content: user_question}] max_steps 8 for step in range(max_steps): response llm.chat(messagesmessages, toolstools, temperature0, max_tokens1200) msg response.choices[0].message if msg.tool_calls: for tool_call in msg.tool_calls: fn_name tool_call.function.name fn_args json.loads(tool_call.function.arguments) if fn_name list_tables: tool_result list_tables() elif fn_name get_schema: tool_result get_schema(fn_args[table_name]) elif fn_name execute_sql: tool_result execute_sql(fn_args[sql]) else: tool_result {error: fUnknown tool: {fn_name}} messages.append({ role: assistant, content: None, tool_calls: [tool_call] }) messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(tool_result, ensure_asciiFalse, defaultstr) }) else: # 没有调用工具说明模型已经给出最终回答 final_answer msg.content breakmax_steps 8这个数字我试过多次调整太小容易在复杂查询时被迫中断太大会让无意义的循环白白消耗token。8步在多数场景下足够覆盖“查表 → 看结构 → 执行SQL → 修正 → 再执行”的完整链路。3.5 结果后处理别让原始执行结果直接糊到用户脸上execute_sql返回的是结构化的JSON数组如果直接把Python字典列表拼装成字符串丢回给模型上下文里会有大量无用信息还可能超出窗口长度。我会做一层轻量化整理def format_result_for_llm(result: dict) - str: if error in result: return fSQL执行失败{result[error]} if not result[rows]: return 查询成功但结果集为空。 names result[columns] lines [] for row in result[rows][:10]: line | .join(f{name}{row[name]} for name in names) lines.append(line) text 查询成功共{}行展示前{}行\n{}.format( result[total], len(lines), \n.join(lines) ) if result[truncated]: text \n注意结果已截断只展示前50行的统计信息。 return text这一步看似不起眼却直接影响Agent在拿到结果后的总结质量。如果给模型的是一坨没格式化的JSON它可能读着读着“忘了”用户原始问题开始复述数据如果给的是按列名对齐的文本它更容易把数值对应到正确字段上最终回答也更自然。4. 落地过程中的坑与排查记录复盘这轮实操最值钱的不是代码而是踩过的坑。我把几个高频问题整理成一份速查表再挑几个典型场景展开说。问题现象根本原因解决方案Agent生成了不存在的表名/字段名模型对业务Schema不熟悉产生幻觉强制调用get_schema、在提示词中规定必须先看表结构后再写SQLSQL语法正确但查询结果为空Agent直接放弃二义性问题模型假设了错误的时间口径或状态值追加few-shot示例、遇到空结果主动返回样例数据帮助推断复杂聚合查询老出错JOIN条件缺失、分组字段不对多表关联体关系没有被模型理解在工具返回中添加外键关系说明、给每个表写业务口径说明Agent突然生成一条耗时全表的SQL没走索引或SELECT了不必要的字段限制返回行数、设置SQL超时时间、提示词要求先WHERE过滤多轮对话里上下文太长历史消息和工具结果不断累积只保留最近2轮消息把用户问题重写为独立问题的query改写模块4.1 幻觉字段名三个解法组合起来才有效这是我遇到频率最高的问题具体表现是用户问“本月新客数”Agent直接写SELECT COUNT(*) FROM users WHERE new_customer 1结果users表里根本没有new_customer字段执行报错Agent再换个类似的名字继续猜。单靠提示词约束解决不彻底我最终上了三招组合第一招强制前置检查。系统提示词里规定“没有查看表结构禁止生成SQL”并从流程上保证get_schema的调用否则不进入下一步。第二招在工具返回中加强业务口径。给关键表维护一份字段说明文档注入到get_schema的返回内容中。比如“new_customer_flag1表示新客0表示老客”。模型看到的字段越多猜错的概率越低。第三招错误反馈循环。当SQL执行报错把数据库的报错信息原样返回给模型并附一句“请基于字段列表修正”。实测这个自纠流程能将首次失败后的修正成功率从40%拉高到75%。4.2 二义性问题宁可追问不要硬猜有次业务方问“最近一个月的订单客单价多少”Agent直接按下单时间算了过去30天。但业务方的真实口径是“按支付成功时间、排除退款订单”。两者结果能差30%。这个问题的根治方案是建立一份“业务口径字典”内置在系统提示词或工具说明里常用口径约定 - “最近一个月”优先按支付成功时间计算范围是[当前日期-30天, 当前日期) - “客单价”指支付成功订单金额 / 支付成功订单数排除退款 - “用户数”一律按用户ID去重 - “GMV”指支付成功状态的订单金额总和模型看到这些约定后遇到二义性表达不再是“自由发挥”而会优先套用内部约定。如果内部约定也覆盖不了它才会主动向用户追问。这一步让SQL准确性提升非常明显。4.3 性能与超时控制别让一条SQL拖垮库Agent在探索阶段可能会生成一条没带WHERE条件的全表查询这在测试环境还好生产环境就麻烦了。我给工具层加了两道保险第一道是SQL静态检查。在execute_sql里先做简单文本检查如果发现SELECT语句没有WHERE条件且不是聚合查询没有GROUP BY就拦截并提示“该查询可能扫描全表请补充过滤条件”。第二道是数据库语句超时。SQLAlchemy连接参数里设置connect_timeout和statement_timeout超时直接终止查询防止慢SQL把连接池占满。这两道保险看起来粗暴但确实能兜住大部分低级错误。如果你是给生产库做Agent强烈建议在这两层基础上再加一层“限制扫描行数”的数据库级配置。4.4 多轮对话一长就乱Query改写比硬记历史更有效早期做多轮对话我把所有历史消息全部塞进上下文结果token很快打满而且模型容易把当前问题和历史错误尝试混淆越问越乱。后来我调整成每轮提问前先用大模型做一次query改写把多轮上下文压缩成一个独立的问题。比如用户先问“上个月的订单总量”再问“那环比上上个月呢”改写后的独立问题就是“对比上上个月的订单总量上个月的订单总量环比变化是多少”。然后用改写后的问题去触发Agent流程上下文只保留最近两轮。这样改造之后系统既保留了多轮对话的连续感又避免了长上下文干扰token消耗还降低了不少。5. 一点实战心得项目跑通后我又花了很多精力去建评估集、做回归测试这是整个项目里最耗时但也最值得的一部分。我整理了约120条典型业务问题覆盖单表查询、多表Join、聚合统计、趋势对比、异常排查等场景每次迭代模型或修改提示词就用这套测试集跑一遍把生成SQL与预期SQL做对比计算执行结果一致率。这个过程中我发现一个有趣的现象Agent偶尔生成的SQL和人工写的SQL完全不同但执行结果完全一致。比如人工用子查询实现的逻辑Agent可能用窗口函数完成。这说明评价一个SQL Agent的好坏SQL文本是否和预期一致不重要执行结果正确才重要。所以后来我的评估指标改成了“结果一致性”而不是“文本相似度”。如果你也想做类似的SQL Agent我的建议是从一个垂直的、表数量有限的业务域开始先把工具层和提示词打磨稳定再慢慢扩展表范围。一次性接入几十张表模型的选表准确率一定会下降需要引入更复杂的路由策略。项目后续我打算往两个方向扩展一是把Agent接入定时报表任务让它可以自动生成日报并发给对应负责人二是增加数据权限隔离让不同角色只能查询自己范围内的数据。这些都属于“第二层围墙”等到第一层跑稳定了再说也不迟。