Agent接入数据库的正确姿势：工具封装、连接池与安全架构全解析

做Agent接入数据库这件事，我前后折腾了快两年。最早抱着“给大模型一个MySQL连接串，让它自己查”的想法，结果被现实狠狠教育：幻觉SQL、连接池被打爆、权限裸奔、事务悬挂，每个坑都踩了个遍。后来我逐渐总结出一套相对稳的接入方式，今天把这套“正确姿势”完整写出来，包含工具封装、连接管理、安全审计、Schema感知、记忆与向量检索，以及一套可以直接抄走的Demo。不管你是用LangChain、自研框架，还是直接调Function Calling，这篇文章都能帮你少走很多弯路。

1. 先搞清楚Agent和数据库之间到底该隔几层

1.1 直接给连接串等于开门揖盗

很多初学者做Agent接数据库，最自然的想法就是：把主机、端口、用户名、密码写进Prompt或环境变量，然后让Agent自己拼SQL去执行。这个方式在Demo阶段能跑通，但一上生产就出事。

我见过一个项目，Agent拿到连接串后，在对话里生成了一条DELETE FROM orders WHERE status='pending'，因为没有WHERE限制条件，它直接把整张表清空了。更离谱的是，这个Agent的数据库账号用的还是root。还有个项目，Agent每次回答一条查询就新建一个连接，高峰期几百个并发请求直接把MySQL的max_connections打满，整个业务系统跟着瘫痪。

这里面的核心问题不是Agent笨，而是你给了它一把万能钥匙，却没有给它使用边界。大模型本质上是概率生成器，它生成的SQL再流畅，也只是一串文本，没有经过任何安全校验和资源控制。数据库连接意味着直接访问存储引擎、系统表、日志文件，如果让Agent裸奔在连接串上，等于把仓库钥匙交给了一个不懂规则的实习生，好处是它能帮你干活，坏处是它可能把仓库烧了。

1.2 正确心智：Agent是终端用户，不是DBA

后来我调整了思路：Agent不该直接访问数据库，它应该面向一组合法工具编程。你希望Agent做什么，就给它封装什么工具，比如search_orders、create_order、get_schema_info，工具内部再去做SQL拼接、参数校验、权限控制、超时管理。Agent只需要决定调用哪个工具、传什么参数，而不是决定SQL长什么样。

这个心智模型很重要，它类似现实世界里的前后端分工：前端不会直接连数据库，而是通过后端的API拿数据。Agent面对数据库时，它的身份就是一个终端用户，你的工具层就是后端API，数据库只在API后面工作。这样一来，Agent的权限边界、错误处理、审计日志都可以收敛在工具层，数据库连接串也不再需要暴露给模型。

另外，我建议在工具层加入“用途说明”。每个工具的description里写清楚“这个工具负责什么，应该在什么时候调用，参数的含义是什么”，甚至可以直接把对应SQL的模板写进去。这样模型在Function Calling时会更容易选对工具，比让它自由发挥写SQL可靠得多。

2. 用工具函数把SQL包起来

2.1 设计最小工具集

很多人误以为封装工具就是把SQL原封不动放进函数里，没有本质区别。其实关键在于“最小命令集”和“输入约束”。

我常用的一组最小工具集包含四个：

• query_database(sql, params, limit)：执行只读查询，强制带上LIMIT上限，默认50条。
• execute_update(sql, params)：执行写操作，仅限INSERT、UPDATE、DELETE，且必须经过字段白名单校验。
• get_table_schema(table_name)：按需返回某张表的字段、类型、注释、索引和外键关系。
• get_query_example(intent)：从记忆缓存里检索与该意图最接近的历史SQL示例。

每个工具都要定义严格的JSON Schema。例如query_database的输入可以这样定义：

{ "type": "object", "properties": { "sql": {"type": "string", "description": "只读SELECT语句，不允许包含分号或UNION"}, "params": {"type": "array", "items": {"type": "string"}}, "limit": {"type": "integer", "default": 50, "minimum": 1, "maximum": 200} }, "required": ["sql"] }

这里有几个细节值得注意：工具描述里明确禁止分号和UNION，是为了减少被注入的可能；limit有硬上限，防止Agent一次拉回全表；params用数组形式，可以在底层强制使用参数化查询。这样即便模型生成了一条带' OR '1'='1的SQL，参数化之后也只是当字符串处理，不会真正改变查询逻辑。

2.2 自然语言转SQL的进阶取舍

工具层有了，接下来最难的是让Agent生成质量足够的SQL。完全依赖模型的Text-to-SQL能力是不靠谱的，需要在工程上做几层配套。

首先是Schema感知。不要给Agent全量DDL，而是给它一张表里最关键的字段和注释。我习惯在工具get_table_schema内部做一次缓存，把表名、字段名、字段类型、注释、枚举值、常用过滤条件拼成一段结构化文本。举例，订单表order_info，我会在一开始把“orders.id, orders.user_id, orders.amount, orders.status(枚举: pending/paid/shipped/cancelled), orders.created_at”这串信息放进工具的description或system prompt。这样Agent生成SQL时，用的字段名基本不会幻觉。

其次是示例增强。给模型一两个“相似需求-正确SQL”的few-shot例子，比单纯定义工具效果明显好。比如用户问“昨天付款的订单有多少”，你可以提前在工具描述里写“示例：查询某时间范围内的订单数量 -> SELECT COUNT(*) FROM orders WHERE status='paid' AND created_at >= '2023-01-01 00:00:00'”。模型看到这种映射之后，生成SQL的成功率会提高很多。

最后是分读写。只读需求走只读工具，写需求走写工具，而且写工具在逻辑里要强制限制影响行数。我踩过一次坑：Agent需要更新一万条订单，但它没带WHERE条件，直接跑了全表更新。后来我在写工具里加了一条铁律：execute_update必须解析出WHERE条件，否则直接拒绝执行。这个规则帮助我拦截了至少十几次危险操作。

3. 连接管理是Agent并发下的生死线

3.1 连接池参数和踩坑记录

Agent不像传统API那样每个请求按固定节奏访问数据库。用户和Agent对话过程中，Agent可能在几秒内连续调用七八次工具，而且对话可以同时开很多个。这种情况下，数据库连接池的设计直接决定系统稳不稳。

我一开始直接用SQLAlchemy默认连接池，结果在并发测试时发现连接数大量增加，原因是SQLAlchemy的pool_size默认5、max_overflow默认10，当Agent同时发起多个tool call时，每个协程都可能从池里拿连接，超过上限就排队，排队超时就报QueuePool limit overflow。而且如果某个SQL查询慢，连接会一直被占用，其他请求全部卡死。

后来我把参数调成：pool_size=20, max_overflow=10, pool_timeout=30, pool_recycle=600。这个配置的意思是：基础连接20个，峰值最多30个，获取连接超时30秒报错，连接在600秒后回收重连。同时，我在工具层加了并发闸门，同一个Agent实例同时只允许“一个查询在执行”，类似信号量限制，避免单个Agent把所有连接抢走。

顺带说一个很多教程不会提的参数：pool_pre_ping=True。这个参数会在每次从池里拿连接前执行一次轻量的SELECT 1，如果底层数据库连接已经被释放或断掉，它会自动重连。在长连接场景下，这几乎是必须开的选项，否则你会遇到“偶发Connection reset”这种玄学问题。

如果你用的是Django、Go的database/sql或者其他语言的连接池，核心思路完全一样：控制连接上限、设置闲置回收、开启连接保活。数据库连接是稀缺资源，Agent的每一步操作都要尽可能复用连接，而不是每次都新建。

3.2 事务边界怎么划才不悬挂

事务是Agent接入数据库时最容易被忽略的坑。普通API请求一般会在一个函数里完成“开启事务-执行-提交或回滚”，生命周期很清晰。但Agent是多轮决策的，它可能第一步查数据，第二步计算，第三步再更新，中间还可能调用其他工具。如果Agent在每一步都自动提交，那么两步之间的状态就断了，如果它中途放弃，第一步的写入就会残留下来。

我目前的策略是：除非业务明确要求多步事务，否则默认让工具层自动提交。也就是说，execute_update内部直接conn.commit()，不给Agent“保存点”和“回滚”的能力。这样做的好处是简单，坏处是Agent不能跨工具做原子操作。

如果你的业务确实需要Agent执行“先扣库存，再创建订单”这种原子性操作，不要指望Agent自己控制事务，而是应该把整个流程封装成一个大工具，例如place_order(product_id, quantity, user_id)，工具内部负责事务和异常回滚。Agent只需要调用这个“服务型”工具，而不是拆成多个SQL步骤。

还有一点：一定要给SQL设置max_execution_time。MySQL可以在会话里设置SET SESSION max_execution_time = 5000，Postgres可以通过statement_timeout控制。这样即使Agent生成了一条全表扫描的慢SQL，也不会拖垮数据库。我在工具层对所有查询统一设置了5秒超时，超时就直接抛错提示模型“查询超时，请尝试增加WHERE条件或缩小范围”。

4. 权限、审计和防注入一条龙

4.1 最小权限原理和落地姿势

很多人的Agent服务用的是数据库管理员账号，这在生产环境就是定时炸弹。最小权限原则在Agent场景下尤其实用，因为模型生成的SQL不可预测，你的权限越小，风险越大。

落地姿势很简单：创建两个数据库账号，一个给只读查询，一个给写入操作。只读账号只授予SELECT权限，并且只能访问业务相关的表和视图；写账号只授予INSERT、UPDATE、DELETE权限，而且最好不要给它DROP、ALTER、CREATE权限。MySQL里可以这样建：

CREATE USER 'agent_read'@'%' IDENTIFIED BY 'strong_password'; GRANT SELECT ON appdb.orders TO 'agent_read'@'%'; GRANT SELECT ON appdb.products TO 'agent_read'@'%'; CREATE USER 'agent_write'@'%' IDENTIFIED BY 'another_password'; GRANT INSERT, UPDATE, DELETE ON appdb.orders TO 'agent_write'@'%';

如果数据库里有敏感字段，比如用户手机号、身份证号，最好的方式是不要授权给Agent，或者在数据库层做脱敏视图。我见过一个金融项目，Agent能查询客户身份证号，这其实完全没必要。正确做法是建一个v_agent_customer视图，里面只保留需要公开的字段，Agent读写都走视图，底层表的安全由DBA控制。

对于像国产的达梦、金仓这类数据库，它们同样支持标准SQL权限管理。我在适配达梦时发现它的权限粒度比MySQL更细，还可以控制用户是否允许访问某些系统视图。如果你的生产环境用的是这些数据库，记得先去查一下语法差异，好在授权逻辑几乎一致。

4.2 审计日志与防注入实战

做审计不是为了应付检查，而是你排查问题的唯一依据。Agent执行了什么SQL，用了什么参数，耗时多久，成功还是失败，这些都需要记录。

在工具层的装饰器里，我会统一记录三条信息：原始SQL（脱敏后的）、实际参数、执行结果摘要。如果出错，还要记录错误类型。这样当Agent哪天“发疯”生成了一条奇怪SQL时，你能快速定位是模型的问题还是工具的问题。

防注入方面，参数化查询是底线，也就是用cursor.execute(sql, params)这种形式，而不是把参数拼进SQL字符串。除此之外，我建议在工具层做几道校验：

• 检查SQL是否以白名单关键字开头（SELECT、INSERT、UPDATE、DELETE）。
• 检查SQL里是否包含--注释符、/* */、;多语句分隔符。
• 检查写操作是否有WHERE条件。
• 检查SQL里是否出现information_schema、pg_catalog、mysql等系统库名。

不要觉得这多余，大模型在生成复杂SQL时经常会在后面多加一个分号，或者顺手拼上什么莫名其妙的注释。校验层的代码量不大，但能挡住80%的低级事故。我自己在代码里是这样写的：

def validate_sql(sql: str, operation: str): sql_stripped = sql.strip().rstrip(';') if sql_stripped.lower().split()[0] not in ALLOWED_OPERATIONS.get(operation, []): raise ValueError(f"操作类型不允许: {sql}") if '--' in sql or '/*' in sql or ';' in sql: raise ValueError("SQL包含非法注释或多条语句") if operation == 'write' and 'where' not in sql.lower(): raise ValueError("写操作必须包含WHERE条件") if any(word in sql.lower() for word in ['information_schema', 'pg_catalog', 'mysql.']): raise ValueError("不允许访问系统表") return sql_stripped

这里有个小技巧：先rstrip(';')再去掉末尾的分号，是为了避免误判合法SQL。因为有些模型会在最后多加一个分号，你直接判断;存在会误杀很多正常语句。

5. 让Agent真正“看得懂”库

5.1 Schema摘要与元数据查询

一个常见的失败案例是：Agent生成了一条SQL用了不存在的字段名，数据库报错后它又瞎猜了一个字段名。这种问题的根源是给模型的Schema信息不足。

我建议维护一份“Agent专用元数据文档”，它不是完整的建表DDL，而是精简过的摘要。包括：表名及注释、核心字段名及类型、字段注释、枚举值、常用查询条件、外键关系。把这些摘要直接放到System Prompt里，或者让Agent在开始工作前先调用get_table_schema获取相关表的信息。

对于表特别多的系统，一次性把所有表结构丢给模型会超过上下文窗口。正确处理是按需加载：先让list_tables工具返回所有表名和注释，Agent根据用户需求挑出候选表，再调用get_table_schema拿具体结构。这样既节省Token，又不会让模型迷失在冗余信息里。

还有一个更进阶的做法，叫“元数据表”。我在库里建了一张agent_schema_map表，专门存放每张表对应的业务描述、字段说明和常用查询模板。Agent遇到不熟悉的表时，可以先查这张表。因为它是你自己的数据，所以可控性更强，更新DB表结构时只需要同步维护这张表，而不需要改Prompt。

5.2 Agent记忆与向量数据库怎么配合

Agent接数据库，不只是“查询”这一个动作。很多场景下，Agent需要记住“用户之前查过什么”“某类请求通常转换为什么SQL”，这就涉及到记忆系统。

我目前的做法是：把用户请求的自然语言、经过清洗后的SQL、执行结果摘要一起存入向量数据库（比如pgvector、Milvus或Qdrant）。每次Agent需要写SQL之前，先在向量库里做一次相似度检索，找到最接近的历史样例，作为few-shot输入。这比每次从零生成SQL要稳得多，查询成功率能提升20%到30%。

这里容易踩的坑是数据同步。如果你的业务库是MySQL，向量库是单独的Milvus，两边数据一致性问题就很麻烦。轻量级的做法是不做实时同步，只在Agent执行成功后将最新样本插入向量库，或者定时用ETL工具抽取。如果你的向量库支持外部数据源，也可以直接用pgvector把向量和业务数据放在一个库，省去同步的烦恼。

关于Agent记忆推荐去看一下吴恩达的“Agent for Beginner”课程，里面提到记忆对Agent决策质量的影响。我自己的体会是，记忆不只是给Agent添加历史对话，更重要的是把“任务类型、数据结构、成功模式”这种偏“经验”的信息存进去，数据库接入的质量提升会非常明显。

6. 实操：一个最小可用的Agent查库Demo

6.1 技术选型和代码骨架

这里我给出一个可以直接跑起来的骨架，技术栈是Python + FastAPI + SQLAlchemy + OpenAI Function Calling。你不用照搬，核心是把工具层和连接管理看懂。

首先定义一个数据库连接引擎，注意连接池参数：

from sqlalchemy import create_engine engine = create_engine( "mysql+pymysql://agent_read:password@localhost/appdb", pool_size=20, max_overflow=10, pool_timeout=30, pool_recycle=600, pool_pre_ping=True, echo=False, )

然后定义工具函数：

def query_database(sql: str, params: list = None, limit: int = 50): sql = validate_sql(sql, "read") sql = f"SELECT * FROM ({sql.rstrip(';')}) AS _t LIMIT {min(int(limit), 200)}" with engine.connect() as conn: result = conn.execute(text(sql), params or {}) rows = result.fetchall() columns = result.keys() return [dict(zip(columns, row)) for row in rows] def get_table_schema(table_name: str): sql = f""" SELECT COLUMN_NAME, COLUMN_TYPE, COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t """ with engine.connect() as conn: result = conn.execute(text(sql), {"t": table_name}) return [dict(row._mapping) for row in result]

这里有个关键操作：query_database会把Agent生成的SQL包一层子查询再做LIMIT。这样即使Agent没有写LIMIT，底层也会强制限制返回行数。代价是如果原始SQL里有ORDER BY，子查询包裹后会报顺序错误，所以更好的做法是从AST层面改写，但作为Demo，这个方案简单有效。

接下来把工具注册到Function Calling循环里：

tools = [ { "type": "function", "function": { "name": "query_database", "description": "执行只读查询，返回结构化结果，必须使用SELECT语句", "parameters": { "type": "object", "properties": { "sql": {"type": "string", "description": "只读SELECT语句"}, "limit": {"type": "integer", "description": "返回行数，默认50，最大200"} }, "required": ["sql"] } } } ]

循环里收到模型返回的工具调用后，执行对应函数，把结果附加到消息里继续对话。这样Agent就能做到“问一句、查一库、答一句”，而且每一步都经过工具层控制。

6.2 从SQLite迁移到生产库的细节

很多人在本地用SQLite做Demo，觉得一切都好，然后上线切到MySQL/Postgres就各种问题。SQLite是单文件数据库，没有真正的连接池，也没有行级写锁，并发一高就直接database is locked。所以如果你的Agent要上生产，数据库选型尽早换。

切换到MySQL/Postgres时，有几个坑是高频的：时间字段格式不同，SQLite的TEXT时间戳到MySQL要改DATETIME或TIMESTAMP；JSON字段处理方式不同，SQLite没有原生JSON类型，要用TEXT，MySQL和Postgres都有原生JSON，查询语法也不一样；事务隔离级别不同，SQLite默认串行化，MySQL常用REPEATABLE READ，Postgres是READ COMMITTED，这会影响Agent在多步查询中看到的数据一致性。

如果你不想自己维护数据库基础设施，直接买托管数据库服务是性价比最高的选择。托管数据库一般自带连接池、自动备份、监控告警，能省掉很多运维精力。我之前在云上试过托管Postgres，它的连接池（PgBouncer）和Agent场景配合得非常好，因为Agent的短小查询特别多，连接复用率很高。

7. 常见问题排查速查表

7.1 高频故障和解决思路

下面这张表是我在Agent接入数据库项目中遇到最多的问题，直接按“现象-原因-解法”来写，方便你抄作业。

印象最深的是有一次线上Agent突然报“sqlalchemy.exc.TimeoutError: QueuePool limit of size 20 overflow 10 reached”。查了很久才发现，是某个工具函数在异常分支里开着连接没关，导致连接泄漏。后来我强制给所有数据库操作包上with engine.connect()上下文管理器，这个问题就再没出现过。

7.2 独家避坑经验

最后分享几条我自己的体感，不一定写在文档里，但真能救命。

第一，让Agent先跑EXPLAIN再执行。你可以在查询工具里加入一个“验证模式”，当Agent生成SQL后，先执行EXPLAIN看扫描行数和类型，如果扫描行数超过阈值就拦截并提示Agent改写。这个策略能挡住绝大多数慢查询，我用了之后数据库负载至少降了一半。

第二，给写操作加“二次确认”。当Agent的意图识别为删除或大批量更新时，工具层可以返回一个“需要用户确认”的信号，让Agent在对话里向用户确认“你确定要执行这个操作吗？”这个看起来多了一步，但对生产环境非常友好。尤其是企业内部使用的Agent，用户对“AI直接删数据”这件事天然不信任，二次确认反而能提升使用意愿。

第三，只读副本和主库分离。如果你的Agent既要查又要写，务必让查询走只读副本，写入走主库。这样即使Agent生成了非常耗时的分析查询，也不会拖累主库写入性能。MySQL可以用主从架构，Postgres可以搭配只读节点。如果你的数据平台有专门的OLAP通道，比如ClickHouse或StarRocks，也可以把分析类查询路由过去，让Agent面对的是统一的数据服务层。

第四，把数据库错误信息“翻译”给模型吗？这里我的建议是不要给原始报错。原始报错会包含一些底层信息，但模型不一定理解。我在工具层会把常见数据库错误映射为友好提示，例如“字段不存在”映射为“请参考表结构中的字段名称”，“数据超长”映射为“参数超过字段长度限制，请缩小范围”。这样模型更容易根据提示自我修正。这个映射表是Agent查库成功率提升最明显的优化之一。

最后说点实在的

我最早做Agent接数据库时，也是直接抛一个连接串给模型，然后被花式教育。现在回头看，正确姿势的核心其实不是“让Agent学会写SQL”，而是“在Agent和数据库之间建立一道可控的闸门”，把权限、并发、事务、审计、上下文都管理起来。Agent本身只是一个会决策的大脑，你给它什么骨架，它就长成什么样。

如果你手里正在做Agent项目，我的建议是先别急着堆功能，花一个下午把数据库工具层设计好：最小化工具集、参数化查询、连接池参数、危险SQL校验、Schema摘要。这些东西写起来不复杂，但在生产环境里，它们的价值比任何花哨的Prompt技巧都大。

再分享一个小技巧：在本地测试Agent查库时，可以故意把连接池调成1，并发调成2，跑一轮压力测试。如果在这种极限条件下你的Agent还能稳定工作，那么生产环境基本不会出大问题。这个测试我每次上线前都会做，它曾经帮我提前发现过三次连接泄漏的问题，希望你也能用上。