最近一直在折腾一件事:怎么让 Agent 真正读懂数据库。很多团队把 Agent 接入了数据库,结果发现效果并不理想——模型不是不会写 SQL,而是它根本不知道该查哪张表、哪个字段代表什么意思、哪些表之间有关联。归根结底,Agent 缺的不是接口,而是知识。于是我在做一个面向 Agent 的数据库 OKF 知识包,并且用 Python 写了一个编译器,把散落在 DDL、接口文档、历史查询里的数据库知识,编译成 Agent 能直接消费的结构化知识包。
这篇文章适合两类人:一类是正在做 Agent 应用,但被数据库工具调用折磨得够呛的开发者;另一类是把数据库元数据管理当长期工程,想提前给 AI 生态铺路的数据平台同学。我要讲的不只是概念,而是整套可复现的思路、格式设计和编译器的核心实现。你不需要有编译器背景,只要写过 Python、用过数据库,就能跟着做出来。
1. 整体设计:为什么 Agent 需要一份“就绪”的数据库知识包
1.1 Agent 消费数据库知识的三种经典姿势
先看一眼大家现在都在怎么让 Agent 操作数据库。
第一种是塞提示词。把建表语句、字段注释、业务规则直接写进系统提示词里。小库没问题,几十张表还能塞得下。一旦表超过一百张,提示词就会被稀释,模型开始丢三落四,甚至把两张业务含义相近的表搞混。
第二种是走 RAG。把文档切片、向量化,然后让 Agent 根据问题先检索相关表结构。这个方法解决了“知识太大塞不进上下文”的问题,但它有个天然的坑:检索到的片段往往是碎片化的,比如只找到了某张表的注释,却丢掉了与它关联的外键。对数据库这种强结构的知识,纯向量检索的召回是有损的。
第三种是让 Agent 直接查询数据库元数据。也就是在 Agent 的工具列表里放一个“查询 information_schema”的函数,让模型自己去发现表结构。这个思路在实验环境里很好看,一上生产就出问题:频繁的系统表查询拖慢业务库、权限模型要把元数据暴露给模型、生成的 SQL 里出现幻觉字段……维护成本一下子变得很高。
这三种姿势我都踩过坑。最后得出的结论是:Agent 真正需要的,是一份在进入推理循环之前就被加工好的、结构清晰的、经过裁剪和冗余消除的数据库知识。这份知识不再散落在各处,而是打包成一个自包含的知识包,Agent 每次只需要加载与自己相关的部分。这就是 OKF 知识包的出发点。
1.2 为什么是“编译器”而不是“运行时生成”
很快会遇到一个问题:知识包是每次调用时临时拼装,还是提前编译好?
我一开始做的是运行时拼装:Agent 发起请求,程序去数据库拉元数据,再拼成 JSON 塞给模型。实测下来问题很多。第一,延迟不可控。一次数据库元数据查询可能只要几十毫秒,但几十张表的外键、索引、视图查下来,耗时往往超过一秒,甚至拖垮业务库。第二,内容不稳定。同一个问题,每次拿到的元数据顺序、格式都可能有细微差异,模型的推理结果也随机波动。第三,没法做质量审查。运行时生成的包你根本来不及检查,字段错漏就直接暴露给模型了。
后来我切换到编译模式:预先写一份“源知识”——包括建表语句、字段说明、业务规则、常用查询模板——再用 Python 写一个编译器,把它编译成一份 OKF 知识包,存成静态文件。Agent 运行时直接读取编译产物,不再碰数据库元数据。
这个转变的关键,就是把“每次现查”变成“一次编译、多次使用”。编译器的好处是可以反复跑,知识更新后重新编译即可。编译过程可以加各种校验,比如外键是否指向真实存在的表、字段注释是否齐全、是否有表完全没被引用,这些都能在编译期暴露出来。这跟写代码是一样的:静态检查总比运行时崩溃好。
1.3 OKF 知识包的设计目标
OKF 是我为这个方案定的一套格式,全称叫 Open Knowledge Format。它不是什么国际标准,而是一套适合 Agent 消费的、自描述的、纯 JSON 的知识组织规范。核心设计目标有三个:
第一,自包含。一个知识包就是一个 JSON 文件,里面既有表结构,也有业务解释,还附带示例查询。Agent 不需要再去查别的系统就能理解数据库的语义。
第二,分层裁剪。整个库的结构可能很大,但一个 Agent 通常只关心某个子集。知识包按业务域拆分成多个小包,每个包控制 token 数量,这样 Agent 按需加载,不会把上下文撑爆。
第三,可验证。包里的每一个表名、字段名、路径引用,都应该能通过编译器校验。只要编译通过,就说明这份知识在结构上是完整的。模型基于这样的知识去生成 SQL,幻觉概率明显下降。
我把 OKF 知识包想象成一个“数据库说明书”。说明书不是把数据库所有文档堆给你,而是告诉你:这个系统有哪些实体,它们之间什么关系,什么场景下该用哪个表。Agent 拿到说明书,再去操作数据库,自然就从容很多。
2. 核心格式拆解:OKF 知识包里到底装了什么
2.1 包的目录结构与 Manifest
一个完整的 OKF 知识包不是只有单个 JSON,而是一个目录,里面包含一个 manifest 文件和多个领域包。编译器的输出大致长这样:
okf-output/ ├── manifest.json ├── domain_order/ │ ├── schema.json │ ├── business_rules.json │ └── examples.json └── domain_customer/ ├── schema.json ├── business_rules.json └── examples.jsonmanifest.json 是包的入口,记录版本号、编译时间、包含哪些领域包、每个包的 token 估算值。Agent 加载知识包时,先读 manifest,再按需加载对应领域包。
领域包里的三份文件分别解决不同问题:schema.json 描述表和字段结构;business_rules.json 描述业务约束和口径;examples.json 存放经典查询样例。分文件而不是合并成一个大 JSON,是为了让编译器和 Agent 都能按需读取,不用一次性解析全部内容。
Manifest 还会记录每个包的血缘信息:它由哪些源文件编译而来。这样当上游 DDL 发生变化时,可以准确判断哪些下游包需要重新编译。
2.2 领域包的数据结构设计
以 domain_order 为例,schema.json 的核心结构我是这样定义的:
{ "domain": "order", "description": "订单域:包含订单主表、明细表与支付信息", "tables": [ { "name": "orders", "alias": "订单主表", "description": "每笔用户订单的一条主记录", "columns": [ {"name": "id", "type": "bigint", "description": "订单唯一ID", "primary_key": true}, {"name": "user_id", "type": "bigint", "description": "下单用户ID,关联 users.id", "foreign_key": "users.id"}, {"name": "status", "type": "varchar", "description": "订单状态:pending/paid/shipped/cancelled"}, {"name": "total_amount", "type": "numeric", "description": "订单总金额,单位分"} ], "indexes": [ {"name": "idx_orders_user_id", "columns": ["user_id"], "description": "按用户查询订单"} ] } ], "relations": [ {"from": "orders.user_id", "to": "users.id", "type": "many-to-one", "description": "一个用户有多笔订单"} ] }设计时特别注意了字段注释。很多数据库的注释非常简洁,比如字段叫 status,注释只有“状态”两个字。对 Agent 来说,这种注释是不够的,它需要知道 status 到底有哪几个枚举值、每个枚举值代表什么业务含义。所以我要求字段的 description 里必须包含可枚举的值说明,如果可能,还要给出口径定义。编译器在解析源文档时,会把 Markdown 表格里的枚举说明自动合并进字段描述里。
2.3 为什么用 JSON 而不是 Markdown 或 YAML
我调研过用 Markdown 直接做知识包的可能性。Markdown 是人类友好没错,但机器解析容易出歧义,尤其是表格和列表混排时。Agent 虽然能读 Markdown,但 JSON 更能保证“每个键都确定”。YAML 虽然简洁,但缩进和类型转换的坑太多,在编译链路上容易出错。
JSON 有 schema 约束,可以用编程方式做强校验。我定义了一个 JSON Schema 来描述 OKF 包的结构,编译器在输出前先做本地校验,确保字段没少、类型正确、引用有效。这一步对质量保障特别重要,后面会专门说。
为了兼顾人类阅读,编译器还会额外生成一份 Markdown 版本的知识文档,供数据平台的同事 review。Markdown 是给开发者和业务方看的,JSON 是给 Agent 吃的。两者来自同一份中间表示,保证不会出现文档和知识包不一致的情况。
3. Python 编译器实战:从源知识到 OKF 包
3.1 编译器整体流程
编译器用 Python 写,整体分四个阶段:
- 读取源知识。
- 解析数据库结构。
- 合并业务规则与文档。
- 输出多个领域包并校验。
项目结构保持简单:
db_okf_compiler/ ├── compiler/ │ ├── __init__.py │ ├── cli.py │ ├── sources.py │ ├── parser.py │ ├── builder.py │ └── validator.py ├── sources/ │ ├── schema.sql │ ├── rules/ │ │ ├── order.md │ │ └── customer.md │ └── queries/ │ ├── order_queries.sql │ └── customer_queries.sql └── okf_output/ ├── manifest.json └── ...这一步的设计原则是“源知识可读、可维护”。schema.sql 就是日常维护的建表脚本;rules 目录放业务口径文档;queries 目录放高质量的查询样例。编译器只是把这些既有资产加工成 Agent 要的形态,而不是让大家从零开始维护一套新知识。
3.2 源知识读取与解析
源知识里最核心的是 schema.sql。我在实测中使用的是 SQLite 数据库,直接用标准库读取结构,省去自己写 DDL 解析器的麻烦:
import sqlite3 import json from pathlib import Path def load_schema_from_sqlite(db_path): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cursor = conn.cursor() tables = [] table_rows = cursor.execute( "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" ).fetchall() for table_row in table_rows: table_name = table_row["name"] if table_name.startswith("sqlite_"): continue columns = [] col_rows = cursor.execute(f"PRAGMA table_info('{table_name}')").fetchall() for col in col_rows: columns.append({ "name": col["name"], "type": col["type"], "not_null": bool(col["notnull"]), "primary_key": bool(col["pk"]), }) fk_rows = cursor.execute(f"PRAGMA foreign_key_list('{table_name}')").fetchall() foreign_keys = [] for fk in fk_rows: foreign_keys.append({ "column": fk["from"], "references_table": fk["table"], "references_column": fk["to"], }) tables.append({ "name": table_name, "columns": columns, "foreign_keys": foreign_keys, }) return tables这段代码把 SQLite 的系统表信息变成了 Python 字典。注意我没有去写复杂的 DDL 正则解析,而是借助数据库自己的元数据接口。这个思路可以平移:PostgreSQL 用 information_schema,MySQL 用 information_schema,SQL Server 用系统视图。编译器只面向元数据接口,不关心具体 DDL 方言。
如果你手头只有 DDL 文件而没有真实数据库,建议用一个嵌入式关系库临时建库,把 DDL 灌进去再读元数据。这样比手写解析器稳得多,也顺便验证了 DDL 本身是否能执行。
3.3 业务规则与查询样例的合并
光有表结构还不够。我在 sources/rules/order.md 里会写这样的业务规则:
# 订单域业务规则 - 订单金额 total_amount 一律以“分”为单位存储。 - 状态为 cancelled 的订单不计入 GMV。 - 订单创建超过 30 天且状态为 pending,视为超时未支付。编译器解析这份 Markdown 很简单,按行分割,过滤空行和标题,把列表项提取出来就好。实际中可以用现成的 Markdown 解析库,但量不大时我自己用简单逻辑处理:
import re from pathlib import Path def parse_business_rules(md_path): rules = [] for line in Path(md_path).read_text(encoding="utf-8").splitlines(): line = line.strip() if line.startswith("- "): rules.append(line[2:].strip()) return rules查询样例文件则是纯 SQL:
-- 查询用户最近 30 天的有效订单 SELECT id, user_id, total_amount FROM orders WHERE user_id = :user_id AND status != 'cancelled' AND created_at >= datetime('now', '-30 days');编译器的解析逻辑是:按分号拆分 SQL,把注释行提取成描述,SQL 本身保留为模板。这样 Agent 以后想查“用户最近 30 天的有效订单”,可以直接复用模板,而不需要从头构思 SQL。
为了让合并后的知识不丢失上下文,builder 阶段会把表、规则、样例按业务域归拢到一起。比如订单域对应订单相关的表、规则和查询样例,客户域对应客户相关的表、规则和查询样例。
3.4 编译输出与自动校验
输出阶段我用 dataclass 先把整个知识包构建成内存对象,再转成 JSON。定义几个简单的数据结构:
from dataclasses import dataclass, field, asdict from typing import List, Dict @dataclass class ColumnInfo: name: str type: str description: str = "" primary_key: bool = False foreign_key: str = "" @dataclass class TableInfo: name: str description: str = "" alias: str = "" columns: List[ColumnInfo] = field(default_factory=list) indexes: List[Dict] = field(default_factory=list) @dataclass class DomainPack: domain: str tables: List[TableInfo] = field(default_factory=list) relations: List[Dict] = field(default_factory=list) rules: List[str] = field(default_factory=list) examples: List[Dict] = field(default_factory=list)转 JSON 的时候,把 dataclass 序列化标准库的 asdict 就行。输出文件后,我会跑一轮校验,检查这几件事:
- 所有外键引用的表名和字段名是否真实存在;
- 每个表是否至少有一个主键;
- 每个字段是否都有 description;
- 每个领域包估算的 token 数量是否超过阈值。
这些校验看起来基础,但实际帮了大忙。曾经有个表的外键引用了另一个领域已下线但还没删除的字段,编译器直接报错,避免了 Agent 在真实查询时产出无效 SQL。
3.5 与 Agent 调用链集成
编译产物最终怎么给 Agent 用?我采用的是“工具描述 + 知识包加载”的组合方式。
Agent 接入数据库的方式不是直接把库接进去,而是暴露一个经过封装的 SQL 执行工具。工具描述里会写清楚:这是一个只读查询接口,执行前需先加载 OKF 知识包。业务系统在 Agent 启动时把对应领域包注入到上下文中,或者放到共享文件里让 Agent 按需读取。
和 Agent 的集成代码大致如下:
import json def load_okf_domain_pack(pack_dir, domain): pack_file = Path(pack_dir) / domain / "schema.json" rules_file = Path(pack_dir) / domain / "business_rules.json" examples_file = Path(pack_dir) / domain / "examples.json" return { "schema": json.loads(pack_file.read_text(encoding="utf-8")), "rules": json.loads(rules_file.read_text(encoding="utf-8")), "examples": json.loads(examples_file.read_text(encoding="utf-8")), }然后把这些内容拼进发给模型的 user 消息里,接一句“根据上述数据库知识,使用工具查询满足条件的数据”。模型看到的是干净的、已经裁剪好的领域知识,而不是一堆它自己都分不清的建表语句。实测下来,生成的 SQL 正确率明显提升,尤其是在列名和表名选择上,基本不再出现幻觉。
4. 踩坑记录:常见问题与排查技巧
4.1 元数据解析失败与字段丢失
最常翻车的是字段类型解析。SQLite 的 PRAGMA 返回的类型是字符串,比如 “VARCHAR(64)”,但实际业务代码里表结构可能经过了视图、虚拟表、甚至 CTE 的映射,直接从 sqlite_master 拿到的信息会和真实可用字段对不上。
排查技巧:不要只信 PRAGMA table_info,要同时查索引和外键。对每个外键,确认引用的列确实存在,并且类型兼容。编译器里加一个“空引用检查”,把出现“引用了不存在字段”的情况直接在编译期报错,而不是等到 Agent 执行 SQL 才暴露。另外,注意 SQLite 里大小写不敏感,但 PostgreSQL 里字段名是区分大小写的,不同数据库的元数据大小写规则要单独适配。
还有一个隐蔽问题:字段注释缺失。很多历史表建表时根本没写 COMMENT,导致编译出来的知识包大量字段 description 为空。后来我定了一条规矩:凡是 description 为空的字段,编译器默认把字段名校验提示放进去,同时发警告给维护人。宁缺毋滥,不硬编虚假语义。
4.2 知识包体积膨胀与 Token 预算失控
知识包做得太全,Agent 一次根本读不完。我第一次编译出来的订单包有 15 万 token,放给 Agent 用,效果反而很差,因为上下文里塞满了大量历史存量表和废弃字段。
解决方案是给每个领域包设置 token 上限。比如设定 8000 token 的硬阈值,编译器在输出前估算 token 量,超过就报错。再用三种手段压缩:砍掉从未被引用过的表和字段;把过长的字段说明压缩成枚举列表;把重复的前缀信息提取到包顶部。
代码里做了一个简单的估算函数:
def estimate_tokens(obj): text = json.dumps(obj, ensure_ascii=False) # 中文按一个字一个 token 估算,英文按4字符一个 token 估算 cn_chars = sum(1 for ch in text if ord(ch) > 127) en_chars = len(text) - cn_chars return int(cn_chars + en_chars / 4)估算不是精确值,但足够做预算检查。真正调用大模型时,还要把系统提示词、工具描述、Agent 自身逻辑的 token 都算进去,给知识包留出余量。
另外要提醒一句:别把所有领域的知识包装进一个 Docker 镜像就完事。后期包里数据会漂移,表结构一变,旧包就失效了。一定要给 manifest 加版本号和上游 source 的哈希值,这样 Agent 或上层系统可以检测到包是否过期。
4.3 Agent 查询断言错误与模板污染
我在测试 Agent 生成 SQL 时发现一个很有意思的现象:知识包里如果有一些过时的示例查询,模型会把过时写法当成金科玉律。比如旧表还叫 order_info,新表已经改名成 orders,示例里残留了旧表名,Agent 就照着报错。
这个问题的根治办法是编译器每次编译前,对示例 SQL 做一次“表名校验”:解析每条 SQL 里出现的表名,并检查它们是否都在当前表清单里。如果某个示例 SQL 引用了不存在的表,编译器直接将该 SQL 标记为失效,缺省情况下不放进知识包。
另一个更隐蔽的问题是外键关联方向。一张表可能同时有多条外键,编译器提取 relations 时如果只写“表 A 关联表 B”,不写关联字段和方向,Agent 就可能在多表 JOIN 时把关联条件写反。所以 relation 里我会带上 from 字段和 to 字段,并要求编译时做对称性检查:from 和 to 必须能关联到 schema.json 里真实存在的字段。
4.4 知识更新与增量编译
数据库结构不是一劳永逸的。最常见的是加字段、加索引、改注释,偶尔会有表重命名或迁移。全量重编容易,但耗时也会随源知识增长而变长。
我后来给编译器加了增量模式。它会对比上次编译产物的哈希值,只重新解析有变化的源文件。因为编译器每个领域包对应独立源文件目录,所以可以只编译受影响的领域,不动其他包。manifest 里记录了每个包依赖哪些源文件,编译时先做依赖检查,没有变化的包直接复用。
增量编译带来的副作用是可能产出新老混合的内容。所以 manifest 中必须记录编译时间,并且 Agent 侧做一个简单的过期检查:超过 24 小时的包不允许直接用于生产推理,需要重新编译。这个严格程度可以按团队节奏调整,但一定要有阈值。
4.5 敏感信息泄漏与权限收敛
知识包里有时候会包含非常具体的业务口径和统计指标,比如“毛利率低于 10% 的订单视为异常”,这些内容从安全角度看并不都是可以扔给任意 Agent 的。
我的处理办法是:在编译前把所有源知识都过一遍脱敏脚本。把涉及具体手机号、地址、身份证号等真实数据的示例统一替换成模板变量。权限更大的业务规则独立放在一个单独的领域包中,只有具备高权限的 Agent 才能加载。编译器在 manifest 里给每个包增加“访问级别”字段,Agent 加载前校验自己的身份。
这可能是整个工程里最容易被忽略的部分。很多人觉得知识包无非是给模型多一点信息,但实际上它已经把数据库的语义结构大规模暴露给了外部系统。权限模型和脱敏策略必须跟知识包的版本管理一起做。
写在最后的一些体会
我真正做完这套 OKF 编译链路之后,最大的感触是:Agent 的数据库能力,并不取决于模型有多聪明,而是取决于你给了多少高质量的知识。模型再强,面对一张没有注释、没有关联提示、没有示例查询的表,也只能瞎猜。反过来,当你把数据库知识整理成一份带结构、带约束、带上下文的 OKF 包,模型的表现会有非常直观的提升。
如果你也想在自己的场景里试,我建议不要一上来就追求完美格式。先用 Python 脚本把自己手头最熟的几张表做成一个 JSON 包,丢给 Agent 跑几个查询,感受一下差距,再慢慢补业务规则和查询样例。编译器这个东西,最大的价值不是自动化,而是逼着你想清楚:Agent 到底需要哪些数据库知识,哪些知识是噪音,哪些知识会误导它。
最后分享一个小技巧:把编译输出的 OKF 包当成普通代码一样提交到版本仓库里,用 diff 来审查每次知识变更。你会发现,数据库结构的演化、文档的修订、业务口径的调整,都会变成一行行清晰可见的 diff,这比任何管理后台都直观。