做了这么久 AI 应用我越来越觉得真正能落地的东西往往不是那些花哨的提示词工程而是把底层数据通道打通。手头这个 Easy Data 项目说白了就是给 Agent 和数据库之间搭一座桥——让 AI Agent 安全、可控、高效地把数据库用起来而不是让它凭感觉编 SQL、瞎猜表名、甚至把生产库搞出不可逆的问题。这篇文章我会从方案选型到工具设计再到完整接入过程和排障经验全部分享出来适合正在做 AI Agent 应用、或者想把大模型能力接进现有业务系统的后端工程师和数据平台同学。1. 对接之前先把问题想清楚1.1 为什么不能让 Agent 直接摸数据库有朋友问过我为什么不能直接把数据库连接信息告诉 Agent让它自己写 SQL 去查听起来很美但实操过的人都知道这基本等于给一个刚入行的实习生直接开放生产库的读写权限早晚要出事。第一个致命伤是幻觉。大模型生成 SQL 的时候对不存在的表名和字段名有一种天然的“自信”。你让它查用户订单它可能张口就来SELECT * FROM orders JOIN users ON ...结果你的表实际叫t_order字段叫customer_id根本对不上。它会编一个看起来很合理的表结构然后理直气壮地执行留下一堆报错日志。第二个问题是权限失控。模型本身没有善恶观念它只会朝着“完成用户指令”的方向走。我见过最夸张的一次是模型为了让一个查询跑通反复尝试创建临时表、修改字段类型甚至想把一张表里的数据迁移到它自己新建的“更合理”的表结构里去。如果这时候给的是管理员账号整张表的元数据都可能被改得面目全非。第三个问题更难发现——不可审计。当模型直接连库时每个执行过的 SQL 是否合理、谁触发的、为什么触发这些都无从追溯。一旦出了问题连复盘的材料都拿不出来。商业场景里这已经不是技术问题而是合规问题了。所以“让 Agent 直接用数据库”这条路方向对姿势不对。需要中间加一层把 Agent 的“意图”翻译成可控、可审计、可限制的数据库操作这就是 Easy Data 存在的意义。1.2 正确的边界Agent 是操作员不是数据库管理员一个合格的 Agent 数据访问方案本质上是在做一个权限收缩。我会让 Agent 觉得自己是一个“很有经验但权限受限的数据分析师”它可以查询、可以分析、可以按条件筛选但不可以改表结构、不可以无限制地全表扫描、不可以执行危险语句。这里的关键转变是不要向 Agent 暴露数据库的表结构语义而是要暴露业务语义。举个反例如果你把t_customer_order_2024这张表直接丢给 Agent它能猜到这是一张订单表但大概率猜不到flag字段代表的是“是否已支付”更不知道status3和status4之间的业务差别。结果就是它生成的 SQL 在语法上完全正确在业务上完全是错的。正解是给 Agent 一套“工具”。比如query_orders(customer_name, date_range, status)按业务条件查订单get_order_detail(order_id)查单笔订单的完整信息get_customer_profile(customer_id)查客户画像list_recent_refunds(days)查最近 N 天的退款记录每个工具的名字和参数都带着业务含义模型只需要理解“用户想要什么业务结果”而不需要知道底层是几张表、怎么 join、字段叫什么。这样既降低了对模型 SQL 能力的依赖又把操作范围限制在了预设的几种动作里。这个设计思路其实和我做很多 API 网关的思路类似对外暴露的是服务能力不是底层数据。Easy Data 就是这套思路在大模型场景下的落地。1.3 项目目标和适用范围Easy Data 的核心目标可以概括成一句话让 Agent 在“能查到数据、能找对数据、不破坏数据”这三个前提下自然流畅地完成数据库相关任务。适用场景其实很宽。我做过的落地场景包括客服机器人查订单物流、内部知识库 Agent 统计项目进度、运营助手拉取渠道日报数据、财务场景查询发票和回款状态。这些场景都有一个共同点用户是在用自然语言问“业务问题”而不是在写 SQL。真正的 SQL 生成和优化应该由中间层来完成而不是让模型去猜。不适合的场景也要说清楚如果是高频、高并发的固定查询直接用写好的 API 就行没必要套一层 Agent如果业务需要复杂的多表联合分析、或者需要极其精确的行级权限矩阵Easy Data 也得做不少定制化改造不是开箱即用。2. 工具层设计给 Agent 配一套安全的数据操作工具2.1 先别急着写代码先把工具清单列出来很多人一上来就写execute_sql(sql: str)这种万能工具这等于没做设计。我在第一个版本就吃过这个亏——模型拿到这个工具后依然在自由发挥依然写错表名依然尝试各种危险操作。因为工具本身没有约束力它本质上还是让模型直接写 SQL只是中间包了一层皮。后来我把工具拆成了按业务域划分的独立函数每个函数只做一件事。设计工具清单时我会遵循几个原则参数尽量用业务词汇比如customer_name、start_date、order_status而不是where_condition、params_json返回结果要结构化让模型能直接读懂避免二次猜测每个工具都声明明确的边界比如“只返回最近 90 天数据”“每次最多返回 100 行”对写操作保持极度克制能不开就不开下面是我在订单查询场景里常用的一份工具清单供参考工具名参数返回内容边界条件query_orderscustomer_name, date_range, status订单列表订单号、金额、状态、时间只查最近 90 天最多 100 条get_order_detailorder_id订单详情加商品明细单笔订单无条数限制get_customer_profilecustomer_id客户基本信息、累计消费、等级只读query_refund_listdate_range, status退款单列表只查最近 180 天get_sales_summarydimension, date_range按维度聚合的销售额统计聚合结果不返回明细你可能会发现这本质上就是在设计和业务强相关的 API。没错就是 API只不过消费方从“前端页面”换成了“大模型”。想清楚这一点设计难度立刻降了一半。2.2 工具描述怎么写模型才不跑偏工具函数的名字和描述决定了模型在什么场景下会调用它。描述写得不清楚模型就会乱点工具或者在两个工具之间反复横跳。这块是真有技巧的我踩过不少坑。最核心的一点是描述里要写清楚“这个工具什么时候该用、什么时候不该用”。举个例子我最初给query_orders写的描述是“查询订单”结果模型在用户问“帮我看看上周的退款情况”时也去调query_orders然后对着返回结果里没有退款状态的记录一脸茫然。后来我把描述改成查询客户订单列表。当用户想了解订单状态、订单金额、下单时间等订单基本信息时使用。退款相关请使用 query_refund_list。支持按客户姓名、日期范围、订单状态筛选。只返回最近 90 天内的数据最多返回 100 条。这样改了之后工具选择的准确率明显提高。模型对描述的依赖程度超出很多人的直觉它不会“理解”你的系统它只会在描述文本中找到和用户问题匹配的关键词。描述里写了“退款”它才会在退款场景下调用正确的工具。另外参数类型和枚举值也要写得非常具体。比如status字段我会在描述里写清楚“可选值pending待支付、paid已支付、shipped已发货、completed已完成、refunded已退款”。模型拿到这个枚举就不会传一个done或者finish进来。2.3 为什么 DDL 必须拦在外面我见过不少团队在做类似项目时为了“让模型更灵活”开放了CREATE TABLE、ALTER TABLE、DROP INDEX这类 DDL 语句理由是“模型有时候需要临时表来跑复杂分析”。我的态度非常明确生产环境永远不应该开放 DDL测试环境也尽量别开。原因很简单。DDL 是不可逆的造成的破坏远超单行数据的误删。而且 DDL 语句一旦执行会触发数据库的元数据锁可能导致线上业务瞬间卡死。我就见过一次模型想让一个查询跑通加了一个索引结果这个索引建得极其不合理反而拖慢了所有写入操作最后运维同学半夜爬起来删索引。在 Easy Data 的实现里我做了两层拦截。第一层在工具描述层面压根不提供 DDL 相关工具第二层在中间层代码里对任何传入的 SQL 做静态检查只要语句开头不是 SELECT直接拒绝执行。双保险。有的读者会问那模型需要做复杂分析怎么办答案是让它把数据查出来在应用层做聚合和计算而不是指望数据库动态改结构。数据库是用来存数据的不是用来给模型写代码的。3. 中间的连接层怎么搭Easy Data 的核心实现3.1 连接管理从连接池到会话级隔离连接层是整个中间件的心脏。Agent 场景和普通 Web 后端最大的区别是一次对话可能会触发多次数据库查询而且这些查询之间还有上下文依赖。用户先问“上个月销售额”模型查了一次接着用户说“再按地区分一下”模型又要再查一次。这两次查询之间如果不做会话级别的数据隔离很容易把不同用户的数据搅在一起。我用的是 SQLite 做早期原型因为文件型数据库天然支持会话级别的简单隔离——每个 Agent 会话可以单独开一个连接不需要处理复杂的线程调度。但如果你用 PostgreSQL 或者 MySQL就需要在连接池之外维护一个“会话连接映射表”。核心逻辑是这样的import sqlite3 import threading from contextvars import ContextVar class EasyData: def __init__(self, db_path: str): self.db_path db_path self._local threading.local() def get_connection(self): # 每个线程/会话持有一个独立连接 if not hasattr(self._local, conn): self._local.conn sqlite3.connect(self.db_path) self._local.conn.row_factory sqlite3.Row return self._local.conn def close(self): if hasattr(self._local, conn): self._local.conn.close() del self._local.connPython 的threading.local()可以让每个线程拿到自己那份数据。再加上contextvars.ContextVar就能把“当前会话 ID”“当前用户角色”这些信息透传到每一个数据库调用里从而实现会话级隔离。这里有个很容易被忽视的坑连接必须在使用完毕后显式关闭。我在早期版本里偷懒没做清理结果跑了一晚上文件句柄全部耗尽整个服务直接挂掉。后来加了一个finally块统一处理才彻底解决。3.2 把查询结果压成 Agent 能用的格式数据库返回的结果集长什么样直接决定了模型能不能继续往下分析。很多人忽略了这一步结果模型拿到一堆没头没尾的数据开始胡编乱造。我的做法是把结果集转成“紧凑但完整”的文本或 JSON同时做三层限制行数限制默认 50 行最多 200 行字段限制每行最多返回 10 个字段长度截断单个字段值超过 200 字符就截断并加省略号标记代码大概是这样的def format_result(rows, max_rows50, max_fields10, max_field_len200): if len(rows) max_rows: rows rows[:max_rows] truncated True else: truncated False formatted [] for row in rows: record {} for idx, key in enumerate(row.keys()): if idx max_fields: break value str(row[key]) if len(value) max_field_len: value value[:max_field_len] ... record[key] value formatted.append(record) return {data: formatted, count: len(formatted), truncated: truncated}为什么要截断两个原因。第一模型有上下文窗口限制一次对话塞进去几千行数据后面的信息就进不来了。第二模型在信息过载时更容易产生幻觉它分不清哪些是真正的数据、哪些是它脑补的所以给它的信息越精炼它越不容易跑偏。同时我会在返回结果里明确标注truncated: true让模型知道“数据没给你看全不要武断下结论”。这个细节很重要它能触发模型执行“缩小查询范围”的动作而不是拿着部分数据直接给用户一个不准确的答案。3.3 审计日志每一个 Agent 动作都留痕数据库操作最怕的是不可追溯。在 Easy Data 里我在中间层写了一个简单的审计模块把每一次 Agent 触发的数据访问都记录到单独的日志表里。审计字段包括会话 ID、用户 ID、触发时间、工具名称、原始参数、实际执行的 SQL、返回行数、耗时。这样一旦出现数据异常可以完整还原“谁在什么时候通过什么方式查了什么数据”。我一般用单独的 SQLite 文件存审计日志和生产数据分开避免日志写入影响主业务。审计写入时机要放在 SQL 执行之后防止把失败的调用也漏掉。def audit(session_id, user_id, tool_name, params, sql, rows_returned, duration_ms): conn sqlite3.connect(audit.db) conn.execute( INSERT INTO agent_audit (session_id, user_id, tool_name, params, sql, rows_returned, duration_ms) VALUES (?,?,?,?,?,?,?), (session_id, user_id, tool_name, json.dumps(params, ensure_asciiFalse), sql, rows_returned, duration_ms) ) conn.commit() conn.close()这个小模块在开发阶段帮了我大忙。模型行为不对的时候翻一翻审计日志立刻就能定位它是不是在同一个工具上反复调用是不是在尝试传入奇怪的参数这些信息比看模型的原始输出有价值得多。4. 实操记录从零接入到跑通一次完整查询4.1 环境准备和数据准备理论说了不少直接进入实操。这里我用一个非常常见的“员工订单查询”场景来演示底层用 SQLite对外提供一个 FastAPI 接口模型通过 OpenAI 兼容的 function calling 调用工具。建表脚本很简单CREATE TABLE employees ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, department TEXT NOT NULL, hire_date TEXT NOT NULL ); CREATE TABLE orders ( id INTEGER PRIMARY KEY, employee_id INTEGER NOT NULL, customer_name TEXT NOT NULL, amount REAL NOT NULL, status TEXT NOT NULL CHECK(status IN (pending,paid,shipped,completed,refunded)), created_at TEXT NOT NULL );两张表员工表和订单表。业务场景是用户用自然语言问“销售部最近 30 天有哪些已支付的订单”这个查询在传统后端里就是一个普通的 JOIN 查询但我们要让 Agent 通过工具自动完成。关键点在于模型需要先知道“销售部”对应哪些员工再查这些员工的订单。这里有两种方案写一个query_orders_by_department(department, date_range)的综合工具或者拆成两个小工具让模型自己组合。实践下来综合工具在准确率上更高因为模型不需要做“先查员工再查订单”的推理链路少一步就少一次犯错机会。我最终工具列表长这样{ type: function, function: { name: query_orders_by_department, description: 按部门名称查询该部门员工的订单列表。只支持查询部门维度订单若需查询单个员工订单请用 query_orders_by_employee。日期范围不超过 90 天。, parameters: { type: object, properties: { department: { type: string, description: 部门名称例如销售部 }, start_date: { type: string, description: 开始日期格式 YYYY-MM-DD }, end_date: { type: string, description: 结束日期格式 YYYY-MM-DD }, limit: { type: integer, description: 返回条数默认 20最大 100 } }, required: [department, start_date, end_date] } } }这里我把limit也暴露给模型是因为模型在统计场景下确实需要控制返回规模。但我会在中间层强制校验超过 100 就裁掉。4.2 接入流程逐步走中间层工具函数实现如下。这个函数会把部门名转换成员工 ID 列表再查订单并且在最后附加一句“本结果最多显示 20 条”让模型知道数据范围def query_orders_by_department(department: str, start_date: str, end_date: str, limit: int 20): if limit 100: limit 100 conn get_connection() # 第一步查部门下的员工 emp_ids conn.execute( SELECT id FROM employees WHERE department ?, (department,) ).fetchall() if not emp_ids: return {data: [], message: f未找到部门{department}} emp_id_list [row[id] for row in emp_ids] placeholders ,.join(? * len(emp_id_list)) # 第二步查订单 rows conn.execute( fSELECT id, customer_name, amount, status, created_at FROM orders WHERE employee_id IN ({placeholders}) AND created_at BETWEEN ? AND ? ORDER BY created_at DESC LIMIT ?, (*emp_id_list, start_date, end_date, limit), ).fetchall() return format_result(rows, max_rowslimit)注意我在这里没有把employees表暴露给模型而是让中间层直接处理表关联。模型只需要提供“部门名”其他脏活累活都由中间层完成。这既保证了业务语义正确又避免了模型去猜表结构。接入 Agent 的流程也很直接读取工具定义拼进 messages把用户的自然语言问题发给模型模型如果决定调用工具就返回一个 function_call 请求应用层执行中间层函数再把结果作为新的 message 回传给模型最终得到一个完整的自然语言回答。messages [ {role: system, content: 你是一个数据分析助手可以直接调用工具获取数据。不要编造数据。}, {role: user, content: 销售部最近30天有多少已支付订单总金额是多少} ] # 调用模型 API携带 tools 参数 response client.chat.completions.create( modelyour-model, messagesmessages, tools[orders_tool_definition], )4.3 一次真实查询的完整过程复盘我用一次实际运行的日志来复盘整个链路。用户输入是“销售部最近30天的已支付订单有多少总金额多少”第一步模型分析用户意图判断需要调用query_orders_by_department参数为department销售部、start_date2025-02-15、end_date2025-03-16。第二步中间层执行 SQL返回一个 JSON 数组里面是 18 行订单数据。format_result输出了count: 18并且没有截断。第三步模型拿到这些数据后自己数了一下行数又用工具返回里的amount字段做了累加最终回答“销售部最近30天共有18笔已支付订单总金额为 126,540 元。”这个链路看着简单但每一个环节都是有讲究的。第一个关键是参数解析——模型准确地把“最近30天”转换成了具体的起止日期这得益于工具描述里明确写了日期格式。第二个关键是返回结构——模型能直接靠count字段统计行数不需要自己倒腾数组长度。第三个关键是模型“自己数数”这一点——虽然中间层也可以做聚合统计但在这个场景里让模型基于返回数据计算能保证结果的自然语言表达更贴合用户问题。我实际测试过如果不给模型明确的limit参数它往往会用默认值 20而不会主动要求更多数据。这说明模型对参数的利用非常保守所以工具设计时一定要给合理的默认值并在返回结果里明确告知数据完整性。5. 常见问题与排障实录5.1 模型编造了不存在的表名症状模型生成的 SQL 引用了一张完全不存在的数据表比如SELECT * FROM user_orders而你实际的表叫orders。原因很直接——你让模型直接写 SQL但你没给它表结构。排查方法看审计日志。如果日志里 SQL 里的表名和实际表名对不上第一反应应该是工具设计不合理模型在猜你的数据模型。解法分两层。短期把“获取表结构”也做成一个工具也就是先说清楚有哪些表、哪些字段再让模型写 SQL。长期更推荐我前面讲的做法不要把原始表结构暴露给模型而是把业务查询封装成高层工具。模型不接触表名就永远不会编造表名。5.2 Agent 在同一个查询上反复空转这是我个人觉得最经典的一个问题。模型明明已经拿到了数据却还在不停地调用同一个工具输出一会儿说“查询中”一会儿说“让我再确认一下”看起来就像陷入了死循环。产生这个问题的原因通常是返回结果里缺少“终止信号”。模型不知道这次查询已经成功了或者说它无法把当前结果和用户问题对应起来。我在返回结果里加了一个显式的completion_note比如“本次查询已返回全部匹配记录可以基于此结果回答用户”就能明显减少空转。还有一种情况是模型把工具结果当成了“需要继续加工的数据”而不是“最终答案的原材料”。这时候我一般会在系统提示词里加一句“当你拿到工具返回数据后应该立刻基于数据组织回答除非数据明确提示错误否则不要重复调用。”实测这句话比什么温度参数调优都管用。5.3 并发一多连接池就炸SQLite 在并发场景下的表现确实不理想写入和读取之间会有锁竞争。我的第一个版本在压测时50 个并发请求一进来服务直接 50% 的超时。排查后发现每个请求都在创建新的连接文件锁互相等待。解决思路是两层。第一加一个简单的连接池复用连接减少文件打开和关闭的开销from queue import Queue class ConnectionPool: def __init__(self, db_path, size10): self._pool Queue(maxsizesize) for _ in range(size): conn sqlite3.connect(db_path) conn.row_factory sqlite3.Row self._pool.put(conn) def acquire(self): return self._pool.get() def release(self, conn): self._pool.put(conn)第二Agent 场景本身就是低频查询没必要追求超高并发。真正的瓶颈通常在模型 API 的延迟上而不是数据库。所以我后来给中间层加了一个简单的信号量限制同时执行的 SQL 查询数量超出部分排队等待。这个操作把服务稳定性拉高了一个档次。如果你用的是 PostgreSQL 或 MySQL连接池直接用现成的组件就行SQLAlchemy 的连接池配合pymysql/psycopg2很成熟不需要自己造轮子。核心思路是一样的连接复用 并发限量。5.4 返回结果太大把上下文塞满了模型上下文窗口有限如果你让 Agent 查一个没有数量限制的明细数据比如“把去年所有订单都列出来”中间层可能一次性返回几千行。模型根本处理不了这么长的输入要么直接报错要么开始胡言乱语。我的策略是三层截断层层兜底。第一层在 SQL 层强制LIMIT。即使模型传入的参数没有带 limit中间层也会自动补一个默认值。宁可结果不完整也不能让上下文爆炸。第二层在格式化时做字段截断。长文本、JSON 字段只保留前 200 个字符。这个策略对那种“订单备注写了 1000 字”的场景特别有效。第三层在返回结果末尾加提示“数据量过大仅返回前 N 条。若需要完整数据请增加筛选条件或缩小范围。”这个提示是在引导模型如果用户确实需要更多数据可以让它继续追问用户具体条件而不是一口吃成胖子。这里我想多说一句很多 Agent 项目跑着跑着就崩其实百分之八十是因为数据量没控制住。模型不是被难题难死的是被数据撑死的。把输出控制好整个系统就稳了一半。6. 落地后的一点体会与后续扩展最后讲一点我自己的体会。Easy Data 这个项目做下来最大的收获不是代码而是对“Agent 能做什么、不能做什么”有了更清醒的认知。Agent 的强大之处在于意图理解和语言组织它能把用户的模糊需求拆解成清晰的动作序列。但它的弱点也同样明显不精确、爱猜测、缺乏边界感。中间层的意义就是把 Agent 的优点留下来把它的弱点用工程手段约束住。我目前在这套方案上做的扩展方向有三个。第一个是缓存层——把模型生成的 SQL 做 hash相同的查询直接走缓存减少数据库压力。第二个是权限矩阵——不同用户角色看到的工具列表不一样比如普通员工只能查自己的订单管理员才能查全量。第三个是 Schema 变更通知——数据库结构变动时自动触发工具定义的重新生成避免模型拿着过时的工具定义瞎猜。如果这个项目继续往下走我还考虑把多轮对话里的“记忆”接进来。现在的工具调用是纯靠上下文传递状态一旦对话轮数变多模型容易忘记前面查过什么。后面会尝试把每次查询结果的关键摘要写进一个小的状态表让模型需要时主动去取而不是全都堆在上下文里。不管怎么说让 Agent 把数据库用起来这件事本身非常有价值。数据是一座金矿但金矿不能直接吞得先挖出来、淘洗过、炼成金条才能拿到应用层去花。Agent 是那个能挖矿的工人而 Easy Data 就是那条淘洗和冶炼的流水线。把这条流水线搭稳了剩下的就是持续迭代和积累场景了。