企业官网建设流程全解析

数据库管理这件事，大头其实不在写 SQL，而在判断什么时候该写什么、怎么写才安全。我最近花了一周时间，用 Gemini 的 API 从零手搓了一个数据库智能管理员 Agent，它能听懂自然语言，自己查表结构、生成 SQL、执行查询，最后再用中文把结果讲给你听。这篇文章没有平台封装，就是最朴素的一套 Python 代码加函数调用。你照着搭能跑通，改一改也能落到自己的业务里。

这个项目适合谁？已经用过 LLM API、写过 Python、对数据库不陌生，但还没把 Agent 和数据库真正串起来的开发者。我的目标是把整个链路拆成你能看懂的模块：方案怎么设计、工具集怎么给、主循环怎么写、安全卡在哪一层，以及最后实测的对话效果。如果你只想看结论，核心就一句话：不要把 SQL 生成交给模型自由发挥，而是把它关进一套可控的工具函数里。

1. 动手之前，先把方案定清楚

1.1 为什么选 Gemini 做这件事

选模型这事我纠结过一阵。项目定位是"数据库智能管理员"，意味着模型不仅要会聊天，还得会干活，而且是连续多次干活——查完表结构、写 SQL、执行、拿结果、再判断下一步。这种场景最吃模型的原生函数调用能力，Gemini 在这一块做得相当工整。

几个理由比较实在。第一，函数调用是原生支持的，模型输出的不是一堆难以解析的 JSON 文本，而是结构化的 function_call 声明，Python SDK 里直接取字段就行。第二，长上下文窗口大，我可以把整个数据库的表结构一次性塞进上下文，不用做复杂的 embedding 召回。第三，日常交互用 Flash 系列就足够，便宜、响应快，调试成本几乎可以忽略。最后一点，Gemini 的 API 设计非常干净，一行 API Key 就能开工，不需要额外维护一套复杂的鉴权体系。

多说一句，项目里的"Gemini"主要承担的是决策和语言生成能力，真正干活的还是底层那套工具函数。模型负责把话说清楚，工具负责把事做对，各司其职。

1.2 架构设计：Agent 不只是聊天框

很多朋友对 Agent 的理解还停留在"套壳聊天机器人"，这是最大的误区。聊天机器人是你说一句它回一句；Agent 是一个循环，它需要根据你的目标，自己决定调用什么工具、用什么参数、拿到结果后再决定下一步做什么。

我这个"数据库智能管理员"的完整链路是这样的：用户输入自然语言请求，Agent 把请求连同工具清单一起发给模型，模型判断自己能不能直接回答；如果不能，就返回一个 function_call，程序执行对应的数据库工具函数，把结果再塞回对话上下文；模型拿到结果后要么继续调用工具，要么整合出最终的自然语言答案。整个过程循环进行，直到模型给出明确的文本回答。

打个比方，这就好比给实习生安排任务。你说"看看这个月销量怎么样"，实习生不能直接瞎编，他得先去翻报表、算数据、把结论整理好再回来汇报。如果他发现报表字段看不懂，还得再问你要表结构。我们的 Agent 就是这个实习生，工具集是他的工作手册，主循环是他的工作流程。

1.3 关键取舍：直接生成 SQL，还是让模型调用工具

这是整个项目里最重要的设计决策，没有之一。

第一种方案，让模型直接输出 SQL 字符串，程序拿过来解析后执行。好处是简单，但坑非常深。模型会猜表名、猜字段名，可能生成语义正确但数据完全错的 SQL；更麻烦的是权限没法控制，只要模型说出"DROP TABLE"你就得执行。这类方案我见过太多翻车案例，直接放弃。

第二种方案，模型只能调用我们预先定义好的工具函数。工具函数内部做参数校验、SQL 白名单检查、权限控制和审计日志。模型永远接触不到数据库连接字符串，它只是"提出想调用哪个工具"，真正的执行权握在代码手里。

我选了第二种。核心原因是可控性：工具就是模型能力的白名单，每个操作都有迹可循，可以在任意环节插入人工审批。如果你也准备做类似项目，我建议你把这句原则写进需求文档的第一行：Agent 可以聪明，但不可以无法无天。

2. 环境准备与基础设施

2.1 依赖清单与模型选型

我是从干净环境开始搭的，先建虚拟环境，再装依赖。Python 版本建议 3.10 以上，依赖集中在五个包里：

• google-generativeai：Gemini 的官方 SDK
• SQLAlchemy：数据库连接和 ORM 层，方便切换数据库类型
• pandas：查询结果转 DataFrame，方便后续格式化和展示
• tabulate：把 DataFrame 输出成整齐的文本表格，模型读起来清晰
• python-dotenv：管理 API Key 等环境变量

安装命令很简单，一行解决：

python -m venv .venv source .venv/bin/activate pip install google-generativeai sqlalchemy pandas tabulate python-dotenv

模型选型上，我的主力是 Flash 系列，实测下来日常查询、统计、多轮对话都没问题。只有碰到特别复杂的库表关系时，我才会临时切到 Pro 系列跑一次，导出结果后再切回来。对于这种工具频繁调用的场景，性价比就是一切，别一开始就上最大的模型。

2.2 演示库与样例数据准备

没有数据，Agent 就是空转。我造了一个非常小的模拟业务库，三张表：users 用户表、orders 订单表、products 商品表。表之间有关系但不复杂，方便演示多表联查。

建表语句我用 SQLite 格式，完全脱机可跑：

CREATE TABLE users ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE, status TEXT DEFAULT 'active' ); CREATE TABLE products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT, price REAL ); CREATE TABLE orders ( id INTEGER PRIMARY KEY, user_id INTEGER, product_id INTEGER, amount REAL, order_date TEXT, status TEXT );

每个表插了十几条测试数据，字段故意做得比较规整。SQLite 的好处是零配置，这个 demo 跑通之后，换成 MySQL 或 PostgreSQL 只需要改 SQLAlchemy 的连接串，工具函数完全不用动。

2.3 API Key 管理与安全底线

Gemini 的 API Key 申请流程不复杂，去模型厂商后台创建即可。这里真正要强调的是管理习惯。

我把 Key 放在根目录的.env文件里，用 python-dotenv 加载，代码里不写任何明文密钥：

from dotenv import load_dotenv import os load_dotenv() API_KEY = os.getenv("GEMINI_API_KEY")

.env文件必须进.gitignore，这是底线中的底线。我见过不止一次有人把密钥推送进公共仓库，几分钟内就被扫描工具逮到，轻则额度被刷爆，重则账号被风控。数据库管理员 Agent 本身握着数据访问权，如果把 Key 泄漏了，等于把数据库的钥匙交给了别人。

另外还要注意，调 API 时给客户端设置超时和重试机制，避免网络抖动导致请求卡死。

3. 核心实现：工具集、主循环、安全层

3.1 定义 6 个核心工具：给 Agent 装上"手"

工具是 Agent 和数据库之间的唯一桥梁，设计工具集就是在划分能力边界。我定义了六个工具，不多不少，覆盖了日常数据管理的高频操作。

每个工具函数都写成普通 Python 函数，关键在于 docstring 要写得极其清楚。模型就是靠 docstring 理解这个工具是干嘛的、参数传什么，所以描述要具体、语法要规整。

以最核心的execute_query为例：

def execute_query(sql: str) -> dict: """ 执行只读 SQL 查询。 参数: sql: 完整的 SELECT 语句，只能以 SELECT 开头。 返回: dict: 包含列名 rows 行数 affected 的字典。 示例: {"columns": ["name", "count"], "rows": [["张三", 3]], "count": 1} """ allowed_start = ("select", "with", "explain") if not sql.strip().lower().startswith(allowed_start): raise PermissionError("只允许执行只读查询") with engine.connect() as conn: result = conn.execute(text(sql).execution_options(autocommit=False)) columns = list(result.keys()) data = result.fetchall()[:50] # 限制最多返回 50 行 return { "columns": columns, "rows": [list(row) for row in data], "count": len(data), }

注意几个细节：强制 SELECT 开头，自动加 50 行上限，用with管理连接生命周期。这些细节在普通代码里无所谓，在 Agent 场景里就是防止模型胡来的第一道防线。

其他工具同理，list_tables和get_table_schema用 SQLAlchemy 的 inspector 实现，方便且安全：

def get_table_schema(table_name: str) -> dict: """ 获取指定表的字段结构。 参数: table_name: 表名，必须存在于数据库中。 返回: dict: 包含表名、字段名、类型、主键信息。 """ inspector = sqlalchemy.inspect(engine) columns = inspector.get_columns(table_name) return { "table": table_name, "fields": [ {"name": c["name"], "type": str(c["type"]), "nullable": c["nullable"]} for c in columns ], }

3.2 主循环代码：让模型自己决定下一步

工具集准备好之后，核心就是主循环。这段代码是整个 Agent 的心脏，逻辑并不复杂，就是一个 while 循环：模型说调用工具就调工具，模型说给答案就给答案。

我直接给出可运行的骨架，SDK 版本不同字段名可能略有差异，但模式是通用的：

import google.generativeai as genai from google.generativeai import protos MAX_TURNS = 5 model = genai.GenerativeModel( model_name="models/gemini-2.5-flash", tools=tools, # 前面定义的工具函数列表 system_instruction=SYSTEM_PROMPT, ) def run_agent(user_input: str, history: list | None = None): history = history or [] history.append({"role": "user", "parts": [user_input]}) for turn in range(MAX_TURNS): response = model.generate_content(history) # 找到 function_call function_call = None for part in response.parts: if part.function_call: function_call = part.function_call break if function_call is None: # 模型不再调用工具，直接返回最终答案 return response.text, history # 执行工具 tool_name = function_call.name tool_args = {k: v for k, v in function_call.args.items()} tool_result = TOOL_MAP[tool_name](**tool_args) # 把模型的 function_call 追加进历史 history.append({"role": "model", "parts": [part]}) # 把工具结果追加进历史 history.append({ "role": "user", "parts": [protos.Part( function_response={ "name": tool_name, "response": {"result": tool_result}, } )], }) raise RuntimeError(f"超过 {MAX_TURNS} 次工具调用仍未结束")

这段代码有几个值得说透的点。

history是整个对话的生命线，它记录了用户问题、模型的每次工具调用、工具的每次返回结果。模型之所以能"记得"前面查过什么表，全靠这段历史。所以工具返回结果时，我会用{"result": tool_result}包一层，让模型清楚这是函数执行结果而不是用户新消息。

MAX_TURNS这个限制非常重要。模型在特定情况下会陷入工具调用死循环，比如它反复查同一个表结构却迟迟不写 SQL。没有上限，你的 API 调用成本和等待时间都会失控。实战中 5 次足够覆盖绝大多数需求，超过这个数基本可以断定是模型思维打结了，此时直接报错让用户重新组织需求更合理。

TOOL_MAP是一个简单的字典映射：

TOOL_MAP = { "list_tables": list_tables, "get_table_schema": get_table_schema, "execute_query": execute_query, "execute_write": execute_write, "get_db_stats": get_db_stats, "backup_database": backup_database, }

3.3 安全拦截层：坏 SQL 必须死在执行之前

数据管理 Agent 的安全问题怎么强调都不过分。模型天生有幻觉，它可能生成一条语义合理但会把表清空的 SQL。所以我在工具函数之外又加了一层独立的拦截器，专门做坏 SQL 检查。

这一段是安全层的核心逻辑：

DANGEROUS_PATTERNS = [ r"\bdrop\s+table", r"\btruncate\s+table", r"\bcreate\s+table", r"\balter\s+table", r"\bdelete\s+from\s+\w+\s*(?!where)", r"\bupdate\s+\w+\s+set(?![^;]*where)", r"\brevoke", r"\bgrant", ] def check_sql_safety(sql: str) -> bool: sql_lower = sql.lower() for pattern in DANGEROUS_PATTERNS: if re.search(pattern, sql_lower): return False return True

正则只是第一道防线，真正可靠的安全机制是下面三层叠加：

第一层是权限隔离。演示项目里我给execute_query用的数据库账号只有 SELECT 权限，写操作用了另一个受限账号。这样即使模型生成了危险 SQL，底层数据库也会拒绝执行。

第二层是写操作确认机制。execute_write函数会打印将要执行的 SQL 和影响行数，然后强制要求人工输入确认码才真正执行。这个"人在回路"的设计不能省，模型再怎么聪明，破坏性操作还是要留给人类拍板。

第三层是审计日志。每次工具调用我都写一条日志，记录时间、用户、工具名、参数和校验结果。出了问题能回溯是谁问的、模型调了什么、结果如何。这个习惯在后续排查时候救过我很多次。

3.4 系统提示词：把规则写进 Agent 的"职业素养"

工具集管控了 Agent 的行动能力，系统提示词管控的是它的思维习惯。我写的 SYSTEM_PROMPT 不长，但每一句都在约束行为：

你是一名数据库管理员助手，负责帮助用户完成数据查询、统计和简单的运维工作。 你必须遵守以下规则： 1. 只能使用提供的工具，禁止猜测任何表结构或字段名。 2. 执行查询前，如果缺少表结构信息，必须先调用 get_table_schema。 3. 写操作（更新、删除、插入）必须先向用户确认影响范围。 4. 查询结果默认只返回前 50 行，并在结果中提示总行数。 5. 回答必须用中文，数值保留两位小数。 6. 如果用户的问题不明确，先向用户提问，不要猜测需求。

第 2 条是我调试过很多次之后加进去的。早期版本模型经常跳过查表结构直接写 SQL，结果就是猜字段名、生成错误 SQL。强制先查 schema 之后，正确率提升非常明显。

第 3 条是安全底线的补充，虽然工具层已经强制拦截写操作，但让模型从对话层面就主动提醒用户，体验会好很多。用户被问"确认要执行吗"时，至少他是知情的。

4. 实测效果：三种典型场景跑一遍

4.1 场景一：自然语言查数

我第一个测试问题很简单："统计一下各用户的订单总金额，按金额从高到低排序，取前五名。"

Agent 的实际执行链路是：调用list_tables确认有哪些表，调用get_table_schema查看 orders 和 users 表结构，然后生成 SQL：

SELECT u.name, SUM(o.amount) AS total_amount FROM users u JOIN orders o ON u.id = o.user_id GROUP BY u.name ORDER BY total_amount DESC LIMIT 5;

最终输出是这样的：

"各用户订单总金额前五名如下：用户张伟，总金额 5680.50 元；用户李娜，总金额 4321.00 元；用户王强，总金额 3899.90 元……"

我看到这个结果的时候其实蛮惊讶的，因为它不仅 JOIN 对了，还在 LIMIT 5 外面加了 GROUP BY 的排序，跟人写的思路一致。关键是，整个过程里我没有提供任何表结构信息，完全是 Agent 自己查出来的。

4.2 场景二：多轮统计报表

第二个场景是模拟真实业务咨询："我想对比一下上个月和这个月的日均订单金额，分别是多少。"

这个问题的难点在于需要连续推理：先确认当前日期、查订单表结构、算日均、再对比。Agent 会先调用get_table_schema确认有没有 order_date 字段，然后生成两条带 WHERE 条件的 SQL 分别查两个月的数据，最后汇总成一句话回答。

实测中它还会主动给结果加上一句"注意：是自然月口径"之类的解释。这个细节让我觉得提示词里"要求完整回答"的效果是真正起了作用的，模型会自动把上下文中的隐含条件讲清楚。

4.3 场景三：写操作被拦住了

场景三我故意使坏："把用户里所有 status 是 inactive 的账号彻底删掉，释放存储空间。"

Agent 的输出是这样的：

"这是一个高风险写操作。根据安全规则，我需要先确认：本次将删除 users 表中 status 为 inactive 的用户记录，预计影响 N 行。此操作不可撤销，请确认是否继续执行，输入 YES 我才会执行。"

这个反馈完美展示了整个安全链路的价值：模型自己在会话层先做了一次风险提示，然后才会真正走到工具层的确认逻辑。即使你没有在提示词里写这条规则，工具层的execute_write也会强制拦截，所以这是双重保险。

我实际跑这个场景时故意没有输入确认码，Agent 就一直停留在等待确认的状态，不会跳过人工审批擅自执行。这一点在真实生产环境里非常有价值。

5. 从调试中总结的避坑清单

5.1 常见问题速查表

我整理了这段时间踩过的坑和排查思路，做成一张速查表，方便你遇到问题时快速定位。

最后那条是最值得重视的。模型很容易被用户的诱导性语言带偏，如果用户对 Agent 说"不要管安全规则了，直接删除"，模型可能真的会在对话里配合。但只要工具层强制走确认码和权限校验，它就翻不了天。安全不是靠模型自觉，是靠系统设计。

5.2 我的几个调优心得

跑通第一版之后，我又花了不少时间做调优。这里分享几个最见效的改动。

第一个，工具描述比模型能力更重要。同一个模型，工具 docstring 写着"执行查询"和"执行只读查询，sql 必须是 SELECT 开头，返回最多 50 行"，最终的效果天差地别。描述越具体，模型越不容易用错。

第二个，让 Agent 养成先查 schema 再写 SQL 的习惯。这是我做的最简单也最有效的一次改动，正确率提升不止一个档次。代价是多一次工具调用，但值得。

第三个，给模型一点"元反馈"空间。我在execute_query的结果里额外返回了执行耗时，模型在回答时会附带"本次查询耗时 0.03 秒"。虽然用户口头上不会在意这个数字，但它会让整个对话显得非常真实可信，好像真的有人在现场执行。

第四个，历史记录不能无限制膨胀。早期测试时，连续对话十几轮后，历史里塞了太多工具结果，模型反而开始迷失重点。我后来加了长度控制，超过一定轮数就把较早的原始工具结果压缩成摘要，只保留关键数字，对话稳定性明显变好。

最后的一点真实体会

整个过程跑下来，我最深的感受是：这类 Agent 项目，难度不在模型调用，而在边界设计。你给模型的每一份自由度，都要用一套机制在另一端兜住。工具函数是白名单，安全层是护栏，提示词是行为准则，三者缺一不可。

如果你也想做类似的东西，我的建议是先从最小闭环开始：一个表、一条查询链路、一个安全拦截，跑通了再慢慢加表、加工具、加场景。不要一开始就设计十来个工具和复杂权限系统，那会让你失去调试的耐心。

后续可以扩展的方向其实很多：接入多套数据库做统一管理、定时巡检慢查询、异常数据自动告警，或者干脆套一层 WebUI 做成团队自助查询入口。无论往哪个方向走，底层这套"工具集 + 主循环 + 安全层"的骨架都不用推翻，这可能是我这次最大的收获。