为Agent编译数据库知识包：从DDL到OKF的工程实践

第二次让 Agent 直接连数据库做查询它把订单表和客户表的关联字段猜错了当天晚上我收到一排告警。那一刻我才想明白把一堆CREATE TABLE丢给 Agent跟把一本没有目录、没有注释的字典丢给新来的实习生本质没有区别。后来我花了两周时间写了一个 Python 编译器把 DDL、业务术语、字段枚举、关系强弱这些信息统一编译成一份Agent 就绪的数据库 OKF 知识包。这套东西不解决“怎么写 SQL”它解决的是“怎么让 Agent 在写 SQL 之前真的读懂数据库”。这篇文章就把这个项目的完整思路拆开讲为什么要从 DDL 里编译知识包而不是直接让 Agent 问元数据接口、知识包内部结构怎么设计、编译器流水线怎么搭、Agent 侧怎么消费以及我踩过的那些低级的坑。适合正在搭 Agent 数据助手、或者准备把数据库能力开放给大模型工具链的人参考。1. 为什么 Agent 接数据库之前先要有一份 OKF 知识包先说一个最直接的体感。让 Agent 调用数据库工具时大多数实现就是把表名、列名、类型、主键外键这些元数据拼进 prompt或者让 Agent 自己SHOW TABLES之后去猜。我的项目前三天就是这么干的效果非常糟糕。1.1 表结构信息与业务语义之间的断层数据库表结构天然缺三块东西第一块是字段的业务含义。customer_id到底是下单客户还是收货联系人status字段是 0 表示正常还是 1 表示正常这类信息在 DDL 里往往只有一个字段名再看一眼类型没了。Agent 面对这种信息缺口它只会靠大模型预训练时积累的“通用常识”去脑补而通用常识和数据仓库的真实业务往往不是一回事。第二块是枚举值和取值逻辑。很多表用 tinyint 存状态DDL 里根本不写0 代表什么、1 代表什么。Agent 一旦猜错生成的 SQL 虽然语法完全正确查出来的业务口径却是全错的。这类问题比语法报错隐蔽得多因为SQL跑得通、结果也非空但就是不对。第三块是表与表之间的真实关系。外键约束写得很清楚的库是少数更多场景里两个表靠一个名叫code的字段软关联业务上却是一对多。Agent 看不到业务约束就很容易把关联方向搞反或者在一张宽表上本来能直接过滤它偏要去 join 一张外表。1.2 直接拼 DDL 为什么不经济有人会说那把 DDL 全文贴进 prompt 不就行了我也试过。一个稍微规范一点的库十来张表DDL 加起来几千行很正常。塞进上下文之后Agent 确实能“看到”字段但注意它看到的是纯物理结构不是认知结构。DDL 里有大量 Agent 不需要关心的内容存储配置、索引定义、字符集、分区策略。这些内容不仅浪费 token还会干扰它对核心语义的聚焦。还有一个问题物理结构有歧义。同一条 MySQL 的CREATE TABLE语句在不同版本里字段类型写法不一样注释可能写了一大段也可能完全没有。你无法保证 Agent 每次能从这些原始文本里稳定提取出同样的信息。而知识包提供的是经过清洗、归一化、补全之后的确定性产物同一份 DDL 输入永远编译出同一份知识包。这样你至少能控制 Agent 拿到手的数据库画像是一致的不会因为模型心情不同而对同一张表产生两种理解。所以这个项目我给自己定的目标是把数据库知识的构建从“让模型临场发挥”变成“预先编译、按需检索”。这也是 “Agent 就绪的数据库 OKF 知识包” 这个名字的由来——先有格式再有编译器最后才谈 Agent。2. OKF 知识包长什么样我给数据结构定了哪些规矩OKF 是我在项目内部给这套知识包格式起的代号全称是 Open Knowledge Format。它不是公开标准而是我根据“Agent 消费数据库知识”这个具体场景定制的一套 JSON 约定。为什么不用现成的 schema registry 或者说元数据模型因为那些模型面向的是工程师不是面向“需要理解语义的模型”。2.1 格式选型为什么是 JSON 渲染模板我一开始考虑过直接用 Markdown 文件每一张表写一段说明。优点是写起来快但缺点是没法程序化校验、没法按字段检索、没法稳定渲染成不同长度的上下文。后来又试了 YAMLYAML 写起来比 JSON 舒服但在代码里做 schema 校验、嵌套校验时Pydantic 和 JSON 的配合最顺。所以最终定的是JSON 作为知识包底层的“知识存储格式”再加上一组渲染模板把 JSON 渲染成 Agent 真正读到的文本片段。这里要区分两个概念知识包是结构化的底稿也就是 JSONAgent 消化的是渲染后的文本。底稿负责精确和无歧义渲染负责可读和节省 token。你在 prompt 里给 Agent 看的永远是不超过几百字的渲染结果而不是把整个 JSON 丢过去。2.2 一张表的知识包长什么样一个最小可用的知识包大致长这样{ db: shop, version: 20250601, checksum: 9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08, tables: [ { name: customers, schema: public, comment: 客户主数据一客户一行, fields: [ { name: id, type: bigint, nullable: false, comment: 客户唯一标识, aliases: [customer_id, user_id], enum_values: null, sensitive: false }, { name: status, type: tinyint, nullable: true, comment: 客户状态枚举, enum_values: { 0: 正常, 1: 已冻结, 2: 已注销 }, sensitive: false }, { name: email, type: varchar(128), nullable: true, comment: 登录邮箱个人敏感信息, aliases: [mail], enum_values: null, sensitive: true } ], pk: [id] } ], relationships: [ { from: {table: orders, field: customer_id}, to: {table: customers, field: id}, cardinality: many-to-one } ], glossary: [ { term: 活跃客户, definition: 近30天内至少下单1次的客户, tables: [customers, orders] } ], sample_queries: [ { scenario: 统计本月成交额, sql: select sum(amount) from orders where created_at date_trunc(month, current_date) } ], meta: { generated_by: okf_compiler, source_ddl: ddl/shop_20250601.sql } }这里每个字段项都尽量包含四个要素属性定义、类型约束、业务注释、候选别名。尤其aliases字段是给 Agent 用的——用户可能会说“客户ID”“用户ID”术语映射全部提前在知识包里做好Agent 就不用自己推断。字段里的sensitive标记也很重要。它不是为了阻止 Agent 使用字段而是让 Agent 在生成 SQL 时意识到这个字段涉及隐私输出结果时可以提示用户脱敏。后面讲 Agent 消费时会再展开。2.3 为什么关系必须显式声明只把每张表的信息做成 JSON 是不够的Agent 经常需要对多表 join而 join 恰恰是幻觉重灾区。我在知识包里单拎了一个relationships数组每条关系表达四类信息关联方向、关联字段、基数、业务约束。一开始我以为从外键约束里提取关系就行结果发现很多线上表根本没有外键两个表之间的关联纯粹是业务约定。后来我在编译器里加了一个补充输入文件让人维护这些说明性的关系声明。这是知识包区别于“自动抓取元数据”的关键自动抓取只能告诉你有什么知识包还要告诉你“为什么这两个表能关联”“关联后统计口径是什么”。这一层语义哪怕再强的解析器也猜不出来必须有人的输入或已有的设计文档参与。3. Python 编译器的三层流水线DDL 到知识包之间发生了什么知识包的格式定了之后紧接着的问题是怎么生成。手工维护 JSON 不是不行但数据库一变知识包就过时人很难每次都记得同步。所以我把整套流程做成了编译器输入是 DDL 和补充说明文件输出是知识包 JSON 和一份渲染好的上下文索引。3.1 流水线概览编译器内部是一个纯函数式的流水线一端进 SQL 文本另一端出知识包中间不访问数据库。这么做的好处是可复现、可测试、不依赖环境。你只要把同一批输入文件放进去任何时候跑出来的知识包都一模一样。我把流水线分成三层第一层是解析层负责把CREATE TABLE这种 DDL 拆成 AST抽取出表名、字段名、类型、可空性、默认值、注释、主键、外键。这一层不负责理解业务语义只负责把物理结构变成中间表示。第二层是丰富层把解析结果和人工补充的语义说明合并。比如在补充文件里写着customers.status 字段枚举0 正常 1 冻结 2 注销编译器就把这段文本翻译成结构化的enum_values挂到对应字段上。这层还负责做归一化字段别名统一小写、全角转半角、注释去重等。第三层是输出层把完整模型序列化成知识包 JSON同时用渲染模板生成面向 Agent 的文本片段最后计算整个知识包的 checksum 写入meta字段。3.2 项目模块结构编译器本身是一个 Python 包典型布局如下okf_compiler/ ├── cli.py # 命令行入口 ├── parser_sql.py # 解析 DDL 的 token/AST 逻辑 ├── models.py # Pydantic 数据模型对应知识包 schema ├── enrich.py # 合并人工语义说明、校验一致性 ├── renderer.py # 渲染 Agent 可读的上下文文本 ├── validator.py # 知识包合法性校验 └── artifact/ └── templates/ ├── table.j2 # 单表文本模板 └── relationship.j2 # 关系文本模板为什么用 Python 而不是 Node 或者 Go原因很务实Python 生态里sqlparse、pydantic、jinja2这三个库加起来几乎覆盖了解析、校验、渲染的全部需求。sqlparse 能粗粒度地把 SQL 拆成语句和 token 流虽然它不做完整的 AST 语义分析但对付建表语句已经够用。pydantic 能给知识包做严格的类型校验——字段类型写错、枚举值类型不匹配编译期就会报错而不是等到 Agent 用的时候才暴露。jinja2 负责把结构化数据渲染成文本模板模板里可以控制讲多少细节、用多大篇幅。这里还有一个容易忽略的设计点编译器必须是无状态的。我没有在项目里引入任何数据库连接没有在运行时去SHOW COLUMNS。为什么不呢第一很多数据库权限受限Agent 的账号可能根本没有读元数据的权限第二DDL 文件本身就是事实来源之一如果运行时再查一遍库两份事实不一致时你根本不知道以谁为准。凡是加入不确定性的环节都会让后续排查变得困难。4. 核心代码实战从建表语句里蒸出知识片段接下来进入真正能抄的部分。下面这几段代码就是我项目里最核心的编译逻辑简略了很多错误处理但骨架是完整的。4.1 解析一条 CREATE TABLE解析我首选sqlparse。它的 AST 不如商业级解析器那么深但好处是容错性好不会因为一两个语法怪癖直接崩掉。做一个 DDL 编译器稳定比完整更重要。import sqlparse from sqlparse.sql import IdentifierList, Identifier from sqlparse.tokens import Keyword, Name, Punctuation def extract_create_table(statements): for stmt in statements: if not stmt.get_type() CREATE: continue tokens [t for t in stmt.tokens if not t.is_whitespace] table_name None columns [] started_columns False for token in tokens: if token.match(Keyword, TABLE): # 下一个非关键字的 token 通常就是表名 for t in tokens: if isinstance(t, Identifier) and not table_name: table_name t.get_real_name() continue if token.match(Punctuation, (): started_columns True elif started_columns: if isinstance(token, IdentifierList): for item in token.get_identifiers(): cols _extract_column(item) if cols: columns.append(cols) elif isinstance(token, Identifier): cols _extract_column(token) if cols: columns.append(cols) if table_name: yield {table: table_name, columns: columns} def _extract_column(identifier): # 以 status tinyint 这类简单字段为主太复杂的语法暂时跳过 tokens [t for t in identifier.tokens if not t.is_whitespace] name tokens[0].value if tokens else None type_token for t in tokens[1:]: if t.is_keyword and t.value.upper() in (NOT, NULL, DEFAULT, COMMENT, PRIMARY, KEY, UNIQUE): break type_token t.value return {name: name, type: type_token.strip().lower()}这段代码不追求解析完美主打“常见建表语句都能被拆干净”。我建议你设计解析逻辑时只承诺四种能力取到表名、取到字段名、取到字段类型、取到字段注释。主键、外键、默认值这些信息能解析出来就解析解析不出来的宁可交给人工补充文件也不要在解析器里硬写一堆正则去猜。4.2 业务语义注入光有 DDL 远远不够解析完只是拿到了结构下一步要把业务术语挂载上去。我专门维护了一个enrich.yaml文件让会写 SQL 但不想动代码的同事也能往里加业务描述tables: customers: comment: 客户主数据一客户一行 fields: status: comment: 客户状态枚举 enums: 0: 正常 1: 已冻结 2: 已注销 email: aliases: [mail] sensitive: true relationships: - from: orders.customer_id to: customers.id cardinality: many-to-one note: 订单表通过 customer_id 指向客户主数据不允许存在孤儿订单 glossary: - term: 活跃客户 definition: 近30天内至少下单1次的客户 related_tables: [customers, orders]然后是丰富层合并代码from pydantic import BaseModel class FieldMeta(BaseModel): name: str type: str nullable: bool True comment: str | None None aliases: list[str] [] enum_values: dict[str, str] | None None sensitive: bool False class TableMeta(BaseModel): name: str schema: str | None None comment: str | None None fields: list[FieldMeta] pk: list[str] [] def merge_meta(parsed_table: dict, enrich_cfg: dict) - TableMeta: table_name parsed_table[table] cfg enrich_cfg[tables].get(table_name, {}) field_cfgs cfg.get(fields, {}) fields [] for raw_field in parsed_table[columns]: fcfg field_cfgs.get(raw_field[name], {}) fields.append( FieldMeta( nameraw_field[name], typeraw_field[type], commentfcfg.get(comment), aliasesfcfg.get(aliases, []), enum_valuesfcfg.get(enums), sensitivefcfg.get(sensitive, False), ) ) return TableMeta( nametable_name, schemacfg.get(schema), commentcfg.get(comment), fieldsfields, pkcfg.get(pk, []), )这一段干的事很朴素但我认为这是整个项目里价值密度最高的一段把人类脑子里的业务约定转成机器可读的数据。没这一步知识包和普通的information_schema导出的元数据就没有本质区别。4.3 渲染成 Agent 直接消费的文本知识包里面存储用 JSON但真正喂给 Agent 的是渲染后的文本。渲染模板我放在 jinja2 里保存成table.j2## Table: {{ table.name }}{{ table.schema or public }} {{ table.comment or }} 字段列表 {% for f in table.fields -%} - {{ f.name }}{{ f.comment or 暂无说明 }}{{ f.type }}{% if not f.nullable %}非空{% else %}可空{% endif %} {% if f.enum_values %} 枚举值 {% for k, v in f.enum_values.items() %} - {{ k }} {{ v }} {% endfor %} {% endif %} {% if f.aliases %} 别称{{ f.aliases | join(, ) }}{% endif %} {% if f.sensitive %} 敏感字段生成 SQL 与展示结果时须提示脱敏{% endif %} {% endfor %} 关联关系 {% for r in table.relations -%} - {{ r.from.table }}.{{ r.from.field }} - {{ r.to.table }}.{{ r.to.field }}{{ r.cardinality }} {% endfor %}渲染出来的效果就是 Agent 真正读到的一段文本比如## Table: customerspublic 客户主数据一客户一行 字段列表 - id客户唯一标识bigint非空 - status客户状态枚举tinyint可空 枚举值 - 0 正常 - 1 已冻结 - 2 已注销 - email登录邮箱个人敏感信息varchar(128)可空 别称mail 敏感字段生成 SQL 与展示结果时须提示脱敏 关联关系 - orders.customer_id - customers.idmany-to-one这段文本的核心价值在于它完全贴近“一个数据库 DBA 给新人讲解业务时说的话”而不是SHOW CREATE TABLE吐出来的冷冰冰的物理定义。5. 让 Agent 真正用起来知识包的检索与上下文装配知识包编译出来了不接进 Agent 的调用链路里就等于白做。我项目里接的方式不是全文塞 prompt而是按需检索。Agent 需要知道哪张表的信息才把哪张表的知识片段取出来。5.1 上下文太长按需检索假设你有 20 张表渲染后的全文可能有 5000 个 token。全部塞进系统提示Agent 会长篇大论地注意到无关表还浪费预算。正确的做法是把每张表的知识片段作为一个独立的“检索单元”用户提问时先做一次粗粒度检索只取最相关的三五张表。我自己用的是纯 Python 实现的轻量检索没有上向量数据库。为什么知识包本身是高度结构化的文本关键词重叠度已经能匹配得很好用户问“本月活跃客户”分词后命中的是“活跃客户”“客户”“customers”这个信号足够强。向量检索适合语义距离远但表达相似的内容而数据库知识包恰恰要避免这种模糊匹配。简单方案可控、无额外服务对于中小规模的库完全够用。检索的输入输出很像一个工具函数def retrieve_knowledge_package(query: str, index: dict, top_k: int 3) - list[str]: tokens tokenize(query) scored [] for table_name, block in index.items(): score sum(1 for t in tokens if t in block.lower()) scored.append((score, table_name, block)) scored.sort(reverseTrue, keylambda x: (x[0], len(x[1]))) return [block for _, _, block in scored[:top_k] if _ 0]这段代码没什么黑魔法。但它把关键的一件事做了让 Agent 在收到具体任务之前已经拿到它应该看哪几张表的提示。5.2 把知识包挂进工具函数与提示词仅仅检索还不够要让 Agent 在工具调用时真正“想到”去用。我注册给 Agent 的工具有两个def get_table_context(table_name: str) - str: 返回指定表的业务语义、字段枚举、关联关系等知识包片段 return render_table_block(table_name) def run_sql(sql: str) - list[dict]: 执行只读 SQL 查询禁止修改操作 ...在系统提示词里我会写清楚使用顺序先调用get_table_context获取相关表的上下文再基于上下文写 SQL最后调用run_sql。这是很典型的 ReAct 模式但关键不在于模式本身而在于get_table_context返回的内容质量。如果它返回的只是字段列表Agent 依然要猜如果返回的是带枚举、带别名、带关系提醒的知识片段Agent 写出错误 join 的概率就会显著下降。我再补一个经常被忽略的细节知识包里不应包含真实数据只能包含结构和语义。真实数据可能涉及隐私而且体积不可控。知识包只做“地图”Agent 运行 SQL 之后拿到的结果才是“现场”。地图和现场分离权限控制和数据安全都好做很多。6. 踩坑记录与边界控制什么情况下别硬上编译器最后这部分是最想分享的。项目整体跑通不难但中间有不少决策如果重新来一遍我会做得更果断。6.1 解析 SQL 的“80% 原则”第一个坑是过度追求解析器的完整度。我一开始想让解析器支持存储生成列、分区表、复杂默认表达式、索引定义结果一周时间全耗在这个上面真正的知识包结构反而没怎么动。后来我把解析目标砍到只剩四件事表名、字段名、字段类型、基础注释。凡是解析不了的直接跳过并打一条 warning在编译日志里标出来让人工补充文件去兜底。这里分享一个判断标准知识包的错误容忍策略应该和 Agent 的容错能力匹配。Agent 本身就很擅长从自由文本里抓重点你不需要给它一个 100% 精确的 AST你只需要给它 80% 的准确结构加上 20% 的人工兜底效果就会好过追求完美解析。把精力花在补全业务语义上回报比高得多。6.2 包失效与重建策略checksum 和 CI第二个坑是知识包不同步。数据库的 DDL 一改知识包还是旧版本Agent 拿到旧信息去查新表必然出错。我用两招解决。第一招是给知识包打 checksum。编译时把所有输入文件拼接后算一个哈希存在知识包meta.checksum里。每次 Agent 加载知识包时先核对发现不对就提示“知识包已过期需要重新编译”。这一步成本极低但能避免很多诡异的线上问题。第二招是把编译过程接进 CI。我现在的做法是数据库的 DDL 迁移脚本一提交流水线自动跑一次编译器。编译失败或者 checksum 变化都会在合并请求里直接标出来。这样知识包始终跟随数据库结构版本走而不是靠某个人想起来手动更新。变更类型知识包是否需要重建说明新增一张表需要新表可能被 Agent 需要新增/删除字段需要字段列表变化修改字段注释/枚举需要语义变化是重构核心只改索引或分区不需要Agent 不需要感知物理优化只有数据量变化不需要结构层知识包不存数据统计6.3 什么时候不要搞知识包编译器最后一个建议可能有点反直觉表数量很少、结构非常稳定的项目不要上编译器。如果是五六张表手动写 JSON 或 Markdown 可能只需要半天而编译器需要写解析逻辑、写合并逻辑、写渲染模板、配 CI整套下来怎么也要一两周。我判断是否值得搞知识包编译器的三个条件表数量超过两位数表结构在持续演进你确实要把数据库能力开放给 Agent 做自动化查询。三个条件至少满足两个才值得投入。如果只是给一个固定报表的数据库接个问答 Demo那直接把业务口径写成一小段提示词塞进系统提示里比做知识包高效得多。反过来如果目标是让 Agent 自主探索一个持续变化的数据仓库那么没有知识包的 Agent 就是一台没有地图的自动驾驶车迟早撞墙。我个人的体会是这个项目最有价值的部分不是那几千行 Python 代码而是它逼着我把数据库的“隐性知识”显式化了。过去 DBA 脑子里那点东西——哪个字段是敏感字段、哪张表和哪张表能用软关联、字段枚举到底什么含义——现在全部变成了一份可以版本管理、可以自动校验、可以随时渲染给 Agent 看的知识包。从此 Agent 学到的不是猜出来的表结构而是这个数据库真实运行的业务规则。如果你也在做类似的事情建议先别急着调大模型先把数据库知识管好后面所有环节都会轻松很多。