大模型Agent实践：用Gemini Function Calling构建安全可控的数据库查询助手

1. 为什么“一个Prompt搞定一切”行不通

去年有段时间，我们团队在处理一个非常头疼的需求：业务方每天要查几十种不同的数据分布，从订单金额区间，到用户复购周期，再到商品类目转化率，每次都要找研发写SQL，一来一回就是一整天。后来我们决定试试大模型，想用Gemini直接对话查询数据库，结果理想很丰满，现实很骨感。

直接把表结构丢给Gemini，问它“上个月各品类订单总额是多少”，它确实能生成一段看起来非常专业的SQL，甚至还会加注释。但真正跑起来，问题一堆：表名和字段名靠猜，明明事实表里有order_actual_amount它非要写order_amount；没有LIMIT的查询把几百MB的数据拉到内存；最要命的是它在一条SQL里join了五张表，其中有张关联表根本不该用来join。业务方看到结果的第一反应是“这玩意儿生成的是科幻SQL”。

这个经历让我彻底放弃了“一个Prompt搞定一切”的思路。大模型在数据库场景下有三个致命伤，不是靠提示词工程能绕过去的。

第一，上下文幻觉。模型对数据库的“理解”完全来自你给它的建表语句和注释，一旦表很多、字段很杂，它就会开始“脑补”不存在的字段和关联关系。你让它查询“近7天活跃用户”，它可能就自作主张地认为user_status字段可以代表活跃状态，但实际上那个字段存的是封禁标记。

第二，缺少执行反馈。人写SQL会错了再改，大模型直接生成SQL是“一锤子买卖”。它不知道自己生成的SQL能不能跑通、结果是不是合理、有没有超过超时阈值，所以一次生成就想要完美结果，本身就是反模式。真正的解法是像人一样给它一个循环：生成、执行、看报错、修正、再执行。

第三，安全性不可控。直接让大模型生成任意SQL，等于把数据库的钥匙交给了一个想象力过剩的实习生。没有权限管控、没有行数限制、没有只读保护，万一哪个prompt泄露导致它执行了DELETE FROM，哭都来不及。

所以我们在第二次设计时彻底换了思路：不再让Gemini“直接生成SQL并执行”，而是把它改造成一个Agent——它负责理解用户意图、拆解任务、调用预定义好的工具函数，让工具函数去和数据库交互，执行完再把结果反馈给它，由它总结成自然语言。这就是“Gemini数据库智能管理员”这个项目的起点。

这个方案适合谁？适合那些已经会调用大模型API，但还没想过把“问答”上升为“执行闭环”的开发者；也适合团队里数据库面广、业务方查询需求多但研发资源紧张的场景。文章后面所有代码和思路，都是围绕“大模型只做决策、工具负责执行”这一原则展开的。

2. 项目骨架：先把“数据库管理员”拆成四个工具再谈Agent

很多人一说Agent就兴奋，但上来就写Agent循环是最容易翻车的。我习惯先把一个角色的工作职责拆成原子能力，再把这些能力变成工具的schema。一个合格的数据库管理员，核心就四件事：看结构、查数据、算统计、验结果。拆成工具函数，就是下面这四个。

为什么是这四个，不是更多？因为工具数量越少，模型做工具选择的准确率越高。工具一多，Gemini在决策时就会犹豫，甚至会选错工具，这也是我在实际测试里反复遇到的。四是个比较合适的平衡点。

get_schema()给的是全库概览，模型用来理解“有哪些表和关系”；get_table_metadata()给的是单表细节，模型在写SQL前先确认字段，避免瞎猜；execute_readonly_query()是最关键的执行器，只读、限时、限量；run_statistical_analysis()是给聚合查询单开的通道，因为聚合查询容易扫全表，必须做额外的保护。

工具拆完之后，Agent的决策链路就清晰了：

1. 用户问“上个月销售额前10的商品”
2. Agent先调get_schema()看有哪些表
3. 再调get_table_metadata()确认订单表和商品表的字段
4. 生成SQL，调execute_readonly_query()执行
5. 拿到结果后，用自然语言回复用户

这个过程里，Gemini的角色从“SQL生成器”变成了“任务调度器”，每次只做一个小决策，每个决策都有工具反馈兜底。

工具设计还有两个容易被忽视的细节。第一，工具的description要写清楚适用场景，不能只写“执行SQL”。我见过有人把description写成“execute sql”，结果模型在用户问“表里有多少数据”时也去调execute_readonly_query()，绕过了统计专用通道。后来我把description改成“执行只读SELECT查询，适用于明细查询或数据提取，聚合统计请使用run_statistical_analysis”，模型的选择准确率立刻上来了。

第二，工具函数要拒绝错误输入。大模型生成工具参数时偶尔会reformat数据，比如把列名加上了双引号或反引号，这在MySQL里没问题，但在SQLite里就会报错。所以我们不能假设模型传进来的参数一定是规范的，工具内部要做归一化处理。

3. 从0搭建Gemini Agent核心循环：代码骨架与工具注册

这块是整个项目的发动机。我用的技术栈很简单：Python 3.10以上 +google-generativeaiSDK + SQLite（演示用），生产环境把SQLite换成PostgreSQL就行，逻辑完全一样。先把环境装好：

pip install google-generativeai python-dotenv

然后初始化模型。这里有一个关键点：Gemini的Function Calling能力并不强制要求使用tools参数时把函数真实现出来，SDK底层会自动把Python函数的签名和docstring转换成模型能理解的结构化工具定义。所以写工具函数的时候，函数名、参数名、docstring就是工具schema的全部来源，这三样必须写得极其规范。

初始化模型的代码骨架如下：

import os import json import sqlite3 from dotenv import load_dotenv from google import genai from google.genai import types load_dotenv() # 数据库连接，SQLite用只读模式打开，防止任何写操作 def get_conn(): db_path = os.getenv("DB_PATH", "./warehouse.db") uri = f"file:{db_path}?mode=ro" conn = sqlite3.connect(uri, uri=True) conn.row_factory = sqlite3.Row return conn # 工具函数1：查看库内表结构 def get_schema() -> str: """ 查看数据库中的所有表名和表注释。 Returns: JSON字符串，包含表名、字段数、表注释。 """ conn = get_conn() cursor = conn.execute( "SELECT name, sql FROM sqlite_master WHERE type='table'" ) tables = [] for row in cursor.fetchall(): tables.append({"name": row["name"], "sql": row["sql"]}) conn.close() return json.dumps(tables, ensure_ascii=False)

为了保证只读，SQLite连接串里加了mode=ro。如果是PostgreSQL，可以用SET default_transaction_read_only = on，或者直接创建一个只能跑SELECT的数据库账号。这条安全线无论如何都要守住。

接下来是最关键的Agent循环。不少新手第一次写Agent循环会踩一个坑：只调用了一次generate_content，发现模型返回了一段话就走了，根本没人继续执行函数和回传结果。正确的循环是这样的：

from google.genai import types client = genai.Client(api_key=os.getenv("GEMINI_API_KEY")) # 把函数全部注册进工具列表 tools = [ types.Tool(function_declarations=[ get_schema, get_table_metadata, execute_readonly_query, run_statistical_analysis, ]) ] system_prompt = """ 你是一名数据库智能管理员，负责回答用户关于数据库的各种查询。 你只能使用提供的工具来获取信息，不得编造数据。 规则： 1. 查询数据前，必须先用 get_schema 或 get_table_metadata 了解表结构。 2. 只能执行 SELECT 语句，禁止任何写操作。 3. 如果查询结果为空，如实告诉用户，不要让用户觉得是你在忽略问题。 4. 最终回答要总结执行结果，给出关键数据和执行耗时。 """ def run_agent(user_message: str, max_turns: int = 6): messages = [] messages.append(types.Content( role="user", parts=[types.Part(text=user_message)] )) history = [ types.Content( role="user", parts=[types.Part(text=system_prompt)] ) ] for turn in range(max_turns): response = client.models.generate_content( model="gemini-2.0-flash", contents=messages, config=types.GenerateContentConfig( system_instruction=system_prompt, tools=tools, temperature=0.2, ), ) # 检查是否有函数调用 function_calls = [] for part in response.candidates[0].content.parts: if part.function_call is not None: function_calls.append(part.function_call) if not function_calls: # 模型认为已经可以回答了，直接输出 return response.text # 把模型这一次的回复追加进历史 messages.append(response.candidates[0].content) # 逐个执行函数调用，构造function response for fc in function_calls: func_name = fc.name func_args = fc.args print(f"[Agent] 调用工具: {func_name}, 参数: {func_args}") # 这里需要一个分发器 result = dispatch_tool(func_name, func_args) # 把结果作为function response回传 messages.append(types.Content( role="user", parts=[ types.Part( function_response=types.FunctionResponse( name=func_name, response={"result": result} ) ) ] )) # 进入下一轮，模型会基于执行结果继续推理 return "达到最大轮次限制，未能完成查询。" def dispatch_tool(func_name: str, func_args: dict): if func_name == "get_schema": return get_schema() if func_name == "get_table_metadata": return get_table_metadata(table_name=func_args["table_name"]) if func_name == "execute_readonly_query": return execute_readonly_query(sql=func_args["sql"]) if func_name == "run_statistical_analysis": return run_statistical_analysis(sql=func_args["sql"]) return "未知工具"

这里有几个我反复踩过的细节，值得单独说。

温度参数我设成了0.2，不是默认值。数据库场景要的是确定性，温度太高会导致模型写出的SQL风格漂移，一会儿用COUNT(DISTINCT x)一会儿用COUNT(x)，结果都对但不可预期。0.2是实测下来正确率和稳定性的折中值。

max_turns限制在6轮是一个基于经验的数字。正常的查询链路是：schema → metadata → SQL执行 → 结果汇总，通常只需要3到4轮。如果超过6轮还在绕，大概率是模型在某个决策上来回横跳，再绕下去就是死循环了。限制轮次之外，最好还有总耗时上限，线上建议设置60秒，超时就熔断。

另外要特别注意：每次循环后必须把模型的响应内容（包括function_call部分）追加到messages里，同时也要把function_response追加进去。少了任何一边，模型都看不到自己刚才调用过什么工具、结果是什么。这一步是Agent循环的“记忆”，没它整个闭环就断了。

4. 数据库查询引擎：安全比聪明更重要

Agent循环写好了，Gemini也确实会调用工具了，但离真正能用还差最关键的一层——查询引擎。这层负责把工具函数送来的SQL做二次检查，然后再真正落到数据库上执行。我在这层的原则是：无条件相信模型会犯错，所以要用机制兜底。

4.1 只读保护不是选项，是底线

工具函数里有只读连接，但只靠这一层不够。万一SQL里写了WITH x AS (...) DELETE FROM ...这种写法，SQLite的mode=ro连接还能拦一下，但PostgreSQL的只读事务只能拦DDL和DML，某些变种写法能钻空子。所以我的做法是在SQL执行前先做一层静态检查：

import re FORBIDDEN_KEYWORDS = [ "insert", "update", "delete", "drop", "alter", "create", "truncate", "replace", "grant", "revoke", "attach", "detach" ] def validate_readonly_sql(sql: str) -> bool: # 去掉字符串和注释，避免误判 cleaned = remove_sql_comments_and_strings(sql.lower()) # 排除含这些关键字的单词匹配，比如"updates"不算 for kw in FORBIDDEN_KEYWORDS: if re.search(r"\b" + re.escape(kw) + r"\b", cleaned): return False return True

写完这个函数后我做过一轮对抗测试：故意让模型生成包含JOIN ... DELETE的SQL，看能不能绕过去。实测结果是，单靠关键词黑名单有漏网之鱼，比如WITH子句里嵌了DML，所以我又加了一道硬约束：在数据库账号层面只授予SELECT权限。两个机制叠加，才能算可靠。

4.2 SQL改写：把模型产生的不确定性关进笼子里

光有检查不够，因为模型生成的SQL即使不违规，也可能“跑得太野”。比如用户问“每个品类的订单数”，模型可能写SELECT category, COUNT(*) FROM orders GROUP BY category，订单表有几千万行，这直接就是一次全表扫描。所以我建了一个execute_readonly_query()函数，内部对SQL做三步改写：

1. 强制加LIMIT。如果SQL没有LIMIT，默认加LIMIT 200；有LIMIT但超过MAX_LIMIT（我设的500），就改写为LIMIT 500。明细查询只需要给用户看个概览，不需要把百万行都倒出来。
2. 包一层子查询。有些SQL里有ORDER BY，直接在外层加LIMIT有语法风险，我处理的方式是SELECT * FROM (原始SQL) AS tmp LIMIT 200。这样既保留原排序，又限制返回行数。
3. 聚合查询强制走专用通道。run_statistical_analysis()里也会执行同样的改写，但额外加一行HINT提示模型：统计结果应当精简，只返回汇总值，不要把明细字段全捞出来。

下面是execute_readonly_query的完整实现：

MAX_LIMIT = 500 DEFAULT_LIMIT = 200 def execute_readonly_query(sql: str) -> str: # 1. 静态安全检查 if not validate_readonly_sql(sql): return json.dumps({"error": "SQL包含禁止的写操作关键字，已拒绝执行。"}, ensure_ascii=False) # 2. 检查LIMIT，没有就补 if not re.search(r"\blimit\s+\d+", sql.lower()): sql = f"{sql.rstrip().rstrip(';')} LIMIT {DEFAULT_LIMIT}" else: # 如果已存在limit,按最大限制截断 sql = re.sub( r"\blimit\s+\d+", f"LIMIT {MAX_LIMIT}", sql, flags=re.IGNORECASE ) # 3. 包一层子查询，保持排序 wrapped_sql = f"SELECT * FROM ({sql}) AS __inner" conn = None try: conn = get_conn() start = time.time() cursor = conn.execute(wrapped_sql) rows = cursor.fetchall() elapsed = round(time.time() - start, 3) # 转成字典列表，便于模型阅读 result = { "columns": [desc[0] for desc in cursor.description] if cursor.description else [], "rows": [dict(r) for r in rows[:MAX_LIMIT]], "row_count": len(rows), "elapsed_seconds": elapsed, } return json.dumps(result, ensure_ascii=False, default=str) except Exception as e: return json.dumps({"error": str(e)}, ensure_ascii=False) finally: if conn: conn.close()

这段代码里有几个容易被忽略的细节。

cursor.description只在SELECT语句返回结果集时有值，如果SQL是EXPLAIN之类的东西，它可能是None，所以要做空判断。default=str是必须的，因为数据库里可能会有datetime类型的值，json.dumps默认序列化不了，不加default会直接炸。

有人可能会问：为什么不用参数化查询？因为这里模型生成的是完整SQL文本，参数化查询解决的是“值注入”问题，我们面对的是“结构不可控”问题，所以用改写+检查+权限三层机制来兜底更实在。

4.3 结果返回的“信息密度”控制

把执行结果直接丢回给大模型时，有个经验很重要：不要把所有行都传给模型。之前我踩过一次坑，用户查了500行明细，我把500行全塞给Gemini去生成总结，结果模型开始“眼花缭乱”，总结出来的信息反而丢失了关键点，而且token消耗大大增加。

现在我的做法是：把结果截断到最多50行传回给模型，同时附上row_count字段告诉模型“实际有500行，这里只展示了前50行”。模型看到总行数后，会自己判断“数据量较大，我只给出摘要和关键统计”，而不是逐行复述。这个信息密度控制，让最终回答的质量提升了一个档次。

5. 实战验证：带着边界Case去测试

系统写完了，别急着上线。我一般会准备一份边界Case清单，把用户可能问的刁钻问题都跑一遍，看Agent能不能撑住。下面是我在这次项目中实际用过的几组测试，以及背后的测试思路。

5.1 测试用例设计

测试完有一个很扎心的发现：模型在“模糊问题”上的表现最不稳定。比如“数据库里有啥”，它可能会把get_schema的结果原封不动贴回来，而不是总结说“库里有订单表、用户表、商品表、类目表”。后来我在system prompt里加了一句话：“工具返回的JSON是原始数据，你需要提炼成用户能读懂的结论，不要让用户看到JSON结构。”这下回答质量立刻上来了。

5.2 一次完整链路的实测记录

我拿“查询上个月各品类订单总额”来走一遍完整日志，方便大家直观感受Agent的思考过程：

用户提问: 查询上个月各品类订单总额 [Agent] 调用工具: get_schema, 参数: {} [Agent] 调用工具: get_table_metadata, 参数: {'table_name': 'orders'} [Agent] 调用工具: get_table_metadata, 参数: {'table_name': 'categories'} [Agent] 调用工具: run_statistical_analysis, 参数: {'sql': 'SELECT c.category_name, SUM(o.order_amount) AS total_amount FROM orders o JOIN categories c ON o.category_id = c.id WHERE o.order_date BETWEEN date('now','-1 month') AND date('now') GROUP BY c.category_name'} [Agent] 最终回答: 上个月各品类订单总额如下：数码类 1,234,567元，家居类 876,543元，服饰类 654,321元，零食类 432,109元。共统计4个品类，查询耗时1.2秒。

注意一个细节：模型在写SQL前先看了get_table_metadata()，确认了category_id字段确实存在于orders表里、categories表确实有category_name字段，然后才动手写SQL。这就是Agent相比于单次Prompt生成SQL的核心优势——它会先侦察再行动。

5.3 失败Case复盘：模型为什么会答非所问

有一轮测试问的是“用户复购率最高的前5个商品”，模型把SQL写成了：

SELECT product_id, COUNT(*) AS buy_count FROM orders GROUP BY product_id ORDER BY buy_count DESC LIMIT 5

这个SQL本身没毛病，但它完全没理解“复购率”的含义。复购率应该是“买过该商品两次及以上的用户在总购买用户中的比例”，而不是单纯的“购买次数”。问题出在哪里？表里没有buyer_id字段的注释，模型不知道orders表里哪一列是用户标识，就退而求其次用COUNT(*)硬凑。

这个Case让我意识到：工具返回的metadata质量决定了模型SQL质量的下限。数据库建表时的注释如果写得含糊，模型就会瞎猜。所以我又加了个预处理脚本：把核心表的字段注释统一刷一遍，比如“user_id: 用户唯一标识，即买家ID”，跑完测试后，复购类查询的正确率明显提升。

6. 踩坑记录：我在这个项目里翻过的车

每个Agent项目做完，都会积累一堆“没想到会这样”的坑。我系统性地记录一下，给后来人当避雷指南。

6.1 工具函数里不能有副作用

第一版里，execute_readonly_query()内部做了一个“查询日志记录”，每次执行SQL都往query_log表里插一条记录。虽然业务上合理，但AGENT循环里这是致命的——模型调用工具时，日志写入本身就是一个写操作，与我们“只读数据库管理员”的定位冲突；更麻烦的是，如果SQL执行失败，工具函数返回错误的同时已经把日志写进去了，这就产生了“有副作用但返回失败”的状态，调试起来非常迷惑。

正确做法是：工具函数必须是纯函数，执行SQL就只执行SQL，日志单独走一个旁路系统，比如写外部文件，与数据库无关。我后来把日志挪到了文件里，Agent链路彻底干净了。

6.2 system prompt里的负向指令不顶用

我试过在prompt里写“不要使用DELETE、DROP等语句”，结果没用。模型不会被自然语言里的“禁止”完全约束，尤其是当用户问题里出现了“删掉”这个词时，模型可能会把用户意图当成真实任务去执行。这个问题的解不是加强prompt措辞，而是从工具层做硬隔离——模型能调用的函数里压根没有写操作工具，它想删也没办法。这个教训很重要：大模型的安全边界必须靠工具权限划分来兜底，而不是靠提示词教育。

6.3 模型会“偷懒”——不去查schema直接写SQL

早期版本里，我发现模型偶尔会跳过get_schema()这一步，直接基于用户问题里的词乱写SQL。比如用户问“orders表有多少条”，它直接就调execute_readonly_query(sql='SELECT COUNT(*) FROM orders')，也不先确认orders表是否存在。这个行为在表少的时候碰巧能跑通，一旦库里有几十张表、表名相似，就会出错。

修复方案是在system prompt里加了个硬性规则：“在第一次调用任何查询工具之前，必须先调用get_schema或get_table_metadata确认表结构。”并且把max_turns从6提到了8，因为有些多表查询链路确实要多花一轮。实测下来，这个调整后“跳过侦察”的情况基本消失了。

6.4 token消耗比预期高，得做瘦身

跑了几轮之后一看账单，token消耗比我预估的高了不少。一个正常的四工具链路，加上function call的元信息来回传，一次完整查询可能消耗约4000到8000个token。如果在工具返回时把500行结果全传给模型，一次查询能冲到3万token。省token的三个有效手段：

1. 工具结果只回传前50行 + 总行数（前面说的信息密度控制）。
2. 表metadata做缓存，几十张表的结构不会频繁变，缓存在内存里可以少调get_table_metadata的次数。
3. 模型选gemini-2.0-flash而不是更强的推理模型，数据库任务对推理深度要求没有那么极端，flash的延迟和价格都更友好，实测正确率差得不多。

6.5 并发场景下的连接管理

演示版用的是短连接，每次工具调用都新开一个连接，用完整就关。这个模式在单用户测试时没问题，一旦上并发，SQLite的写锁冲突和连接频繁开关都会变成性能瓶颈。生产环境换成PostgreSQL后，我建议用连接池（比如psycopg2.pool），并且给数据库账号设置statement_timeout（我设了10秒），防止模型生成的SQL意外跑成“慢查询”把连接拖死。这个超时设置在生产里非常重要，别指望模型生成的SQL永远高效，得让数据库从物理层面兜底。

7. 最后再分享一个提升准确率的小技巧

整个项目做完，我最大的体会是：Agent能力的上限，不在模型本身，而在你喂给它的“上下文质量”和“工具设计”。模型再聪明，你给它的表注释一团浆糊，它也写不出好SQL。

最后分享一个立竿见影的小技巧：在metadata里给每个字段加一个“常见用法示例”字段。不是单纯写“user_id: 用户ID”，而是写“user_id: 用户唯一标识，用于关联用户表和订单表，常见查询会用它做JOIN或者COUNT(DISTINCT user_id)”。表面上看这增加了token消耗，但它让模型在写SQL时有了“锚点”，不再靠猜。我在内测的50个问题集上做A/B测试，加了示例后SQL跑通率从71%提升到了92%。这个投入产出比，非常值得。

如果你也想在自己的项目里复刻这套东西，我建议的顺序是：先搭好四个工具函数，把只读保护和LIMIT改写做扎实；再跑通Agent循环，保持工具数量精简；最后再来打磨prompt和metadata质量。别一上来就在prompt里堆砌花哨的CoT模板，先把执行闭环做稳，后面的一切优化才有地基。