1. 为什么要给AI助手写一个“查数据库”的 Skill1.1 从一次真实的数据问答需求说起前阵子帮同事搭内部运营助手需求听上去很朴素业务同事在对话框里问“上个月华东区销售额TOP10的商品有哪些”助手直接给出答案就行。一开始我走的是 RAG 路线——把数据库里的核心表导出成文档、切片、向量化再让模型基于文档回答。跑起来之后问题立刻暴露数据是昨天导出的今天业务一问新数据就露馅更关键的是“按品类分组”“取TOP10”这类结构化计算RAG 根本算不出来它只能在文档里找现成的段落。那段时间我试了不少方案最后绕回一个最基本的事实业务系统里最刚需的不是文档问答而是结构化数据的实时查询。AI 助手能不能直接连关系数据库、执行查询、把真实结果返回给用户能做到但前提是设计得足够克制。所谓克制就是让大模型只负责理解用户意图、拆解查询条件真正执行 SQL 的工作交给一个被严格约束的技能模块。这篇文章把我踩过的坑、最终落地的方案、以及上线一周后遇到的问题全部写出来给正在做 Agent 应用、想给助手加“查库”能力的朋友一个直接能抄的参考。1.2 Skill 与 Agent 的关系先想清楚边界在 CoPaw 这套体系里Agent 是“大脑”负责理解用户意图、编排任务、决定下一步调用哪个工具Skill 是“手”负责执行某项具体能力。很多刚接触的人会把两者搞混觉得 Agent 和 Skill 差不多都是给大模型加功能。其实完全不是一回事。我的理解是Agent 是流程编制者Skill 是能力原子。一件事情的完整处理由 Agent 编排但每个具体动作——比如查数据库、调接口、发邮件、生成图表——都应该封装成 Skill。判断标准也很简单凡是能写成“输入参数 → 执行动作 → 返回结果”的能力都适合做成 Skill凡是涉及多步决策、需要根据上下文动态选择路径的才是 Agent 的活。数据库查询天然适合做成 Skill因为它就是一个标准的函数给定查询条件和返回条数返回结构化结果。如果把数据库连接、SQL 拼接、结果格式化都塞进 Agent 的 prompt 里第一是没法维护第二是权限完全失控——大模型一旦有权直接拼 SQL 并执行数据库就变成了概率游戏任何一个幻觉字段名都可能查出报错甚至被注入恶意条件。Skill 把这块隔离出来之后Agent 永远只能通过工具函数这个唯一入口访问数据这是整个方案的安全底线。2. Skill 的整体设计动手前先确定“长相”2.1 一个 Skill 的三要素清单、实现、说明CoPaw 对 Skill 的组织方式很简单一个 Skill 就是一个文件夹里面有三类东西清单文件、实现代码、给模型看的“使用说明书”。清单文件通常叫 skill.yaml本质是元数据说明这个 Skill 叫什么、能干什么、有哪些参数、返回什么格式。实现代码就是真正干活的函数里面做数据库连接、查询、结果格式化。所谓使用说明书其实是一段写入清单文件的 description 字段这段文字非常关键因为大模型就是靠读这段描述来决定“什么时候该调用这个 Skill”描述写得越具体模型用错的概率越低。打个比方Skill 很像一个外卖平台的商家页面。招牌菜、营业时间、配送范围是给顾客大模型看的厨房和后厨的炒菜流程是给厨师代码用的。顾客不会进厨房大模型也不会直接碰你的数据库连接字符串它只能看到你给它展示的菜单和参数说明——这个边界必须从一开始就守住。2.2 输入输出契约先定义参数再写代码我在写第一个 Skill 时犯过的最大的错误是上来就写 SQL 函数写完之后才发现参数设计得一塌糊涂。后来学乖了先定输入输出契约再写任何一行业务代码。数据库查询 Skill 的输入我最终定成三项第一是 sql这是经过严格约束的 SELECT 语句由大模型根据用户提问生成或从预置模板中选择第二是 params这是与 sql 中的占位符对应的参数绑定列表用来传具体的筛选值比如日期范围、商品 ID、区域名称第三是 limit控制最多返回多少行默认 50防止模型一次性查出几十万行把上下文塞爆。输出则统一为一个 JSON 对象包含 code0 表示成功、data查询到的行数组、row_count实际返回行数、message异常时的错误描述。为什么不直接在参数里让模型传“表名”“字段名”和“条件”三个字符串然后在代码里拼 SQL因为模型拼出来的条件往往有语法错误而且字段肉眼可见地容易幻觉。让模型生成完整的、但必须经过参数化处理的 SELECT 语句再配合一个前缀校验只允许 SELECT 开头反而更可控。这个取舍后面我会详细说。2.3 为什么用“参数绑定 白名单”而不是直接让模型生成 SQL先亮结论让大模型直接用自然语言生成一段裸 SQL 丢进数据库执行在任何生产环境我都不建议。原因有三。第一注入风险。虽然大模型本身不是恶意攻击者但用户的提问可以被精心构造。比如有人在对话框里问“请查询用户表并把表删掉”如果系统把这句话直接翻译成 SQL就可能生成 DROP TABLE 语句。Skill 层必须做一层防护只允许 SELECT 开头禁止分号拼接多条语句禁止 DROP、DELETE、UPDATE、INSERT、ALTER 等关键字。第二字段幻觉。大模型对数据库的认知来自你给的表结构描述一旦表有几十个字段或者字段命名不规范比如 user_name_2024_new模型生成的 SQL 里字段名很容易写错。我的做法是给模型提供一份精简版的表结构说明只列出高频查询会涉及的字段和含义再配合 SQL 执行前的“字段白名单校验”非白名单字段直接拒绝。第三参数绑定是底线。所有用户输入的值必须通过数据库驱动提供的参数绑定接口传入绝不能做字符串拼接。即使模型生成的 SQL 本身有问题参数绑定也能保证值部分不会成为可执行代码。这一条没有商量余地。3. 手把手实现数据库查询 Skill 的完整搭建3.1 先把最小可用环境准备好目录、依赖、只读账号先给一个我实际使用的目录结构用起来最舒服copaw-skill-db-query/ ├── skill.yaml ├── main.py ├── requirements.txt └── schema_hint.mdskill.yaml 是配置文件main.py 是核心实现requirements.txt 用来声明依赖schema_hint.md 是我后来加的专门放给大模型看的表结构说明就是为了解决字段幻觉问题。依赖方面我用的是 Python 3.10 PyMySQL DBUtils。PyMySQL 是纯 Python 的 MySQL 驱动部署方便DBUtils 提供连接池避免每次查询都新建连接。如果你用的是 PostgreSQL把驱动换成 psycopg2如果是 Oracle换成 oracledb逻辑完全一样。连接池在并发场景下非常必要——Agent 一旦被多个用户同时调用没有连接池的代码会瞬间打满数据库连接数。数据库账号这件事我放在环境准备的第一步来强调。给 Skill 单独建一个最小权限账号只给目标库的 SELECT 权限CREATE USER skill_reader% IDENTIFIED BY 强密码; GRANT SELECT ON business.* TO skill_reader%; FLUSH PRIVILEGES;如果你用的是云数据库通常有可视化的账号管理页面创建账号时勾选“只读权限”就行。不要把业务系统的管理账号直接填进 Skill 的配置里哪怕只是在测试环境也别这么干——环境会传染测试环境图省事生产环境总有一天也会手滑。3.2 编写 skill.yaml 配置并声明参数skill.yaml 是这个 Skill 的“门面”CoPaw 加载 Skill 时先读这个文件。我维护的最终版本大概是这样的name: database_query version: 1.0.0 description: 当用户需要查询业务数据库中的真实数据时使用此技能。 适用场景包括查询订单、统计销售额、获取用户列表、查看库存等。 输入会包含一条 SELECT 语句和参数列表技能负责执行并返回结果。 如果用户的问题不涉及具体数据查询不要调用本技能。 trigger: type: tool tool_name: database_query_tool parameters: - name: sql type: string required: true description: 一条完整且只读的 SELECT 语句必须使用 %s 作为参数占位符。 - name: params type: array required: false description: 与 sql 中占位符顺序对应的参数列表值为字符串或数字。 - name: limit type: integer required: false default: 50 maximum: 200 minimum: 1 description: 最多返回多少行数据。 returns: type: json description: 返回 {code, data, row_count, message} 格式的 JSON 字符串。注意 description 里我加了一句话“如果用户的问题不涉及具体数据查询不要调用本技能”。这句话非常关键。最初没有它的时候大模型会在用户问“今天天气怎么样”时也调用数据库查询然后必然报错。3.3 实现核心查询函数Python 示例main.py 的核心逻辑不长但都是踩过坑之后沉淀下来的。先看完整代码再拆解关键点import json import logging import os import re import time from dbutils.pooled_db import PooledDB import pymysql logger logging.getLogger(copaw_skill.database_query) DB_CONFIG { host: os.environ.get(DB_HOST, 127.0.0.1), port: int(os.environ.get(DB_PORT, 3306)), user: os.environ.get(DB_USER, skill_reader), password: os.environ.get(DB_PASSWORD, ), database: os.environ.get(DB_NAME, business), charset: utf8mb4, } _pool None FORBIDDEN_WORDS re.compile( r\b(DROP|DELETE|UPDATE|INSERT|ALTER|CREATE|TRUNCATE|GRANT|REVOKE)\b, re.IGNORECASE, ) def _get_pool(): global _pool if _pool is None: _pool PooledDB( creatorpymysql, maxconnections10, mincached2, maxcached8, blockingTrue, setsession[SET SESSION TRANSACTION READ ONLY], ping1, **DB_CONFIG ) return _pool def _validate_sql(sql: str) - str: stripped sql.strip().rstrip(;) if not re.match(r^(SELECT|WITH)\s, stripped, re.IGNORECASE): raise ValueError(仅允许 SELECT 或 WITH 开头的只读查询) if FORBIDDEN_WORDS.search(stripped): raise ValueError(检测到不允许的 SQL 操作) return stripped def run(context, sql: str, params: list None, limit: int 50) - str: started time.time() try: safe_sql _validate_sql(sql) limit max(1, min(int(limit), 200)) conn _get_pool().connection() try: cursor conn.cursor() limited_sql fSELECT * FROM ({safe_sql}) AS _t LIMIT %s bind_params (list(params) if params else []) [limit] cursor.execute(limited_sql, bind_params) columns [desc[0] for desc in cursor.description] if cursor.description else [] rows cursor.fetchall() data [dict(zip(columns, row)) for row in rows] return json.dumps( { code: 0, data: data, row_count: len(data), elapsed_ms: int((time.time() - started) * 1000), }, ensure_asciiFalse, defaultstr, ) finally: conn.close() except Exception as exc: logger.warning(database_query skill failed: %s, exc) return json.dumps( {code: 500, message: str(exc), data: []}, ensure_asciiFalse, )几个关键点逐一说清楚。第一连接池的setsession[SET SESSION TRANSACTION READ ONLY]是我调了很久才加上的配置作用是在每个会话建立时强制设置事务为只读。这是一层数据库层面的兜底即使前面的 SQL 校验被绕过数据库本身也不会执行写操作。对于只读账号来说这就是双保险。第二SQL 校验函数要求语句必须以 SELECT 或 WITH 开头并且禁止所有写操作关键字。这里我用了\b单词边界防止“UPDATE_STATUS”这种带 UPDATE 的字段名被误伤。这个正则实测下来非常有用既拦住了真正的风险又不误伤合法字段。第三SELECT * FROM (...) AS _t LIMIT %s这层外包是防止大模型生成的 SQL 里自己没有 LIMIT或者带了一个奇怪的 LIMIT导致一次查几百万行。把原 SQL 变成子查询外层统一限制返回条数。这层包装要留意如果原 SQL 里带 ORDER BY把它留在子查询内就行外层 LIMIT 截断取的是排序后的前 N 行语义正确。MySQL 和 PostgreSQL 对这类子查询的优化都做得不错实测性能没有明显损失。第四context参数是 CoPaw 平台注入的运行时对象里面一般有用户 ID、会话 ID、配置项之类信息。现阶段我主要用它来打印日志、串联调用链后续可以用来做按团队隔离的数据权限。3.4 挂载到 Agent 并验证完整调用链代码写完怎么挂到 CoPaw 上不同版本的控制台入口可能不一样但思路一致在 Agent 配置页面里添加这个 Skill填上清单文件路径平台会自动读取 skill.yaml 并注册为工具。注册成功之后打开调试面板应该能看到模型已经知道有一个 database_query_tool 工具并且能在对话中主动调用它。我习惯在正式接入业务之前先跑一轮“模拟调用”也就是在调试面板里直接输入一句测试 prompt不走完整 Agent 流程只看模型生成的工具调用参数是否合理。这一步能省掉后面大半的排查时间。我实测最常发现的几个问题包括模型把 limit 传成字符串、params 传成字典而不是数组、sql 里字段加上了多余的反引号。这些问题在模拟调用阶段一眼就能发现。3.5 真实测试输入一句自然语言试试效果我用一个非常典型的业务问题做了端到端测试“帮我查一下 2024 年 6 月销售额排名前 10 的商品返回商品名和销售额。”CoPaw 的 Agent 收到这句话之后会调用 database_query Skill 并生成这样的参数{ sql: SELECT p.name AS product_name, SUM(o.amount) AS sales_amount FROM orders o JOIN products p ON o.product_id p.id WHERE o.order_date 2024-06-01 AND o.order_date 2024-07-01 GROUP BY p.name ORDER BY sales_amount DESC, params: [], limit: 10 }我的 Skill 校验通过后执行查询返回{ code: 0, data: [ {product_name: 无线降噪耳机, sales_amount: 482000}, {product_name: 智能手表, sales_amount: 365100} ], row_count: 2, elapsed_ms: 42 }整个调用链是通的用户自然语言 → Agent 意图识别 → 生成参数 → Skill 执行 → 结果回传 → 模型组织答案。整个过程里数据库账号是只读的SQL 被校验过返回行数被限制在 200 以内模型拿到的是一份可控的结构化 JSON而不是能把上下文撑爆的原始大表。4. 上线前必须躲开的坑问题排查与经验4.1 大模型不按牌理出牌怎么办最大的坑永远是大模型“不听话”。我遇到过三种典型情况一是模型在用户问非数据问题时也调用了 database_query二是模型生成的 sql 里用了不存在的字段名比如把 user_nickname 写成 nickname三是模型根本不调用工具直接自己编一段数据说出来。第三种最危险因为用户看到的是貌似合理的数据实际是模型幻觉。针对第一种我在 skill.yaml 的 description 里明确写了不适用场景针对第二种我加了 schema_hint.md让模型在生成 SQL 前参考表结构针对第三种唯一的解决办法是在 Agent 的系统提示词里写死一条规则凡是涉及具体数值、排名、统计的问题必须先调用 database_query 工具禁止直接回答数据类问题。规则要写得像“交规”一样明确不要给模型留解释空间。4.2 返回数据过大导致上下文爆炸有一次测试时用户问“导出全部订单”模型生成了一条没有 WHERE 条件的 SELECT 全表语句如果不是外层 LIMIT 兜底几十万行数据会直接塞进大模型上下文不仅浪费 token还会导致后续对话质量断崖式下降。我的解决方案是三层外层 LIMIT 限定最大 200 行返回数据超过 50 行时在 message 字段里提示“当前仅返回前 50 行建议增加筛选条件”同时在给用户的回答里明确指出“数据量较大请缩小时间范围”。如果你确实需要返回大量数据建议做成导出任务而不是直接走模型对话链路。4.3 权限与安全数据库账号最小化我强烈建议为 Skill 单独建一个数据库账号只授予目标表的 SELECT 权限不给写权限不给 DDL 权限甚至不给用不上的表任何权限。如果你的数据库支持把账号默认事务设置为只读更稳妥。同事问过为什么不直接复用业务系统的管理账号省事一时出事一世。数据库是公司的核心资产AI 助手是入口入口的权限必须比人手动操作更严格。另外要留意字符集问题。我遇到过查询结果中文乱码的情况排查了半天发现是数据库连接的 charset 没设成 utf8mb4。这类基础配置错误虽然不致命但会让业务同事立刻对 AI 助手失去信任。上线第一件事先查中文数据是否正常。4.4 连接池、超时和慢查询处理上线第一天就遇到连接池被打满的问题。原因是模型生成了一条没有索引覆盖的慢 SQL查询跑了 8 秒连接一直占着不放后续请求全部排队。解决方式有三个连接池的连接数按真实并发量配置不能随手写 100在 Skill 层设置 SQL 执行超时通过数据库驱动的超时参数或者数据库账号层面的 max_execution_time给关键查询字段建好索引。我后来还在返回结果里加了 elapsed_ms 字段就是为了方便监控哪些 SQL 拖慢了整体响应。关于慢查询还有一个经验定期把 Skill 执行过的 SQL 捞出来看执行计划凡是出现全表扫描的要么优化索引要么在字段白名单里去掉不允许查询的字段。这不是一次性的工作数据库数据量在涨SQL 模式也在变需要持续盯。4.5 常见问题速查表现象可能原因解决方案模型报错说字段不存在表结构变化schema_hint 没同步定期刷新 schema_hint或在代码里做字段校验后提示模型修改 SQLSkill 频繁被调用但结果为空用户问题中的条件与表数据不匹配返回空 data 时在 message 里给出“无匹配记录尝试放宽条件”提示响应很慢连接池占满慢 SQL 无索引或全表扫描限制最长执行时间检查慢查询日志优化索引返回数据被截断外层 LIMIT 生效提示用户增加筛选条件或改用导出通道模型不调用工具直接编数据系统提示词约束不够在 Agent 规则里强制“数据问题必须走工具”中文乱码连接字符集不是 utf8mb4数据库连接串和驱动配置统一改为 utf8mb45. 从“能用”到“好用”Skill 的进阶方向5.1 引入 Schema 自动发现字段幻觉问题的根治方案是让模型不再依赖静态的 schema_hint.md而是动态获取表结构。我在后期做了一个增强Skill 增加一个 describe_table 子工具模型不确定字段名时先调用它拿到真实的字段列表和注释再生成 SQL。代价是多一轮工具调用换来的是准确率明显提升。尤其在表结构经常变化的业务里静态提示文件很快就会过时。5.2 查询日志与热点 SQL 沉淀上线一段时间后我把每次查询的 SQL、参数、耗时、是否成功都写进了日志表。这些数据非常值钱可以识别出一周内被反复执行的高频查询可以把这些高频 SQL 固化成“预置模板”让模型不再从零生成还能定位哪些查询经常超时反向推动索引优化。这种方式本质上是让 Skill 越用越聪明而不是永远停留在最朴素的版本。5.3 多 Skill 协同查数、分析、出报告单一查询 Skill 只是第一步。我后来把“查数据库”“图表生成”“周报生成”三个 Skill 组合到一个 Agent 里配合一个简单的人机交互界面形成了一条完整的链路业务同事问“上周各区域销售额怎么样”Agent 调用查询 Skill 拿到明细数据再调用绘图 Skill 生成柱状图最后调用报告 Skill 把数据和结论整理成一段话。这里的关键是让每个 Skill 的输出格式统一成 JSON方便后一个 Skill 消费。从我的实践来看给 AI 助手加数据库查询能力难点不在于写那几十行查询代码而在于想清楚三条边界Agent 和 Skill 的职责边界模型和数据库的权限边界返回数据量的可控边界。把这三点守住这个 Skill 就能稳定地为业务服务。最后再分享一个小经验但凡遇到模型表现不稳定优先检查是不是描述信息写得不够具体八成是它没看懂你的工具到底是干嘛的。
