去年我在给某内部项目接入 AI Agent 时,踩了一个大坑:Agent 能读到的只是数据库的表名、字段名和类型,而它真正需要的业务语义、字段取值范围、表间关联逻辑,全都散落在代码和运维文档里。结果就是 Agent 生成的 SQL 经常出现错误 join、错误的枚举判断,甚至把金额字段当整型直接求和。后来我用了两个星期,用 Python 写了一个“编译器”,把数据库的元数据、抽样数据和业务规则统一“编译”成一份 Agent 直接可读的 OKF 知识包,也就是 Organized Knowledge Framework 组织化知识包,才算真正把数据库变成了 Agent 能理解的数据源。这篇文章就把整个实战过程拆开讲清楚,包括格式设计、Python 实现、完整案例和排错经验。
这个项目适合三类人:正在做 AI Agent 应用但苦于数据接入的开发者、想给数据库构建语义层的数据工程师,以及对“编译器 + 元数据”组合玩法感兴趣的 Python 爱好者。不需要你有编译原理基础,但建议了解 SQL 和 Python 基础。整个方案的核心思路,是把数据库理解成一门需要被“翻译”的语言,而 OKF 知识包就是它翻译后的目标代码。
1. 为什么数据库要“Agent 就绪”——先搞清楚 Agent 到底读不懂什么
在动手写代码之前,我花了两天时间梳理 Agent 访问数据库时的真实痛点。不解决这些问题,知识包的格式设计就会跑偏。
1.1 三个“读不懂”和一个“不敢用”
Agent 直接读数据库时,遇到的第一个问题是字段名不是自然语言。例如用户表里的字段叫crt_tm,Agent 很难猜出这是“创建时间”;订单表里的st可能是状态、小计、甚至街道缩写。让大语言模型去猜缩写含义,猜对是运气,猜错是常态。
第二个问题是业务枚举值没有字典。表设计里常见的status字段,可能用 0、1、2 表示“待支付、已支付、已发货”,这些映射关系通常只存在于后端代码的常量定义里。Agent 不知道WHERE status = 1到底查的是什么状态,自然也就无法在回答中给出准确的业务解释。
第三个问题是表间关系是隐式的。数据库的外键约束可能会配置,但很多业务系统实际是在应用层做关联,数据库层面根本没有外键声明。更麻烦的是,某些表之间的 join 条件不是等值关系,比如订单和优惠券的关联是通过时间范围匹配的,这些隐含规则 Agent 完全无法感知。
“不敢用”则是我在实测中最直观的感受:当 Agent 不确定某个字段取值时,它会倾向于生成看似合理但实际上是猜测的 SQL。比如不清楚order_type的取值范围,它就可能用IN ('ONLINE', 'OFFLINE')这种不确定的过滤条件,结果查出来的数据跟自己想的完全对不上。
1.2 为什么必须用“编译器”思路而不是普通脚本
很多人会说,导出数据库结构只需要一个简单的 Python 脚本就够了,为什么要上升到“编译器”?我的理解是,普通脚本只能完成元数据搬运,即把表结构原样导出,而 Agent 需要的是一次结构转译:把数据库 DDL 里的词法符号(表名、字段名、类型、约束)解析成有业务含义的知识结构。这天然就是一个编译过程。
具体来说,这个编译器分成四个阶段,对应传统编译器的前端和后端:
- 词法分析:读取数据库元数据,把表名、字段名、类型拆解成 token,并解析命名规则,如
crt_tm拆成crt和tm。 - 语法分析:按数据库方言的规则,将元数据组织成规范的表结构、关系、约束等语法树。
- 语义分析:补充业务语义、枚举字典、字段描述、PII 标签等,这些来自人工维护的术语表、规则映射和大模型辅助补全。
- 目标代码生成:将语义分析后的中间表达输出为 OKF 知识包,并做格式校验和数据统计。
这个思路的额外好处是:以后如果要从 OKF 反向生成建表语句、查询语句或者文档,只需要替换编译器后端的目标生成器,前端的词法、语法、语义分析完全复用。我在实际扩展中就用它生成过数据字典文档,改起来非常省事。
2. OKF 知识包格式设计——定义和组织知识
OKF 的定位是一份机器可读、语义自包含、可供 Agent 直接消费的数据库知识包。它解决的核心矛盾是:数据库的物理结构是给数据库引擎看的,Agent 需要的是给自然语言模型看的“语义视图”。两种视图之间需要一个中间表达层,OKF 就是这个表达层的载体。
2.1 为什么选 YAML 而不是 JSON 或 HTML
格式选型我曾经犹豫过。JSON 解析成熟但注释不方便,HTML 适合人读但对 LLM 的 token 利用效率低。最终选了 YAML,理由有三点:
- YAML 支持注释。知识包里的人工业务注释可以直接就地写,和内容放在一起,不像 JSON 那样需要额外的
_comment字段扭曲结构。 - YAML 对缩进敏感,嵌套层级天然可视化,Agent 读取时结构清晰。实测在同样信息量下,YAML 的 token 开销比 JSON 少 15% 左右。
- YAML 可以很方便地和 Python 的字典结构互相转化,编译器代码可以保持简洁。
当然,如果你的下游 Agent 工具只接受 JSON,可以在交给 Agent 之前做一次格式转换。编译器内部统一用 Python 字典处理,输出层可以按需指定格式。
2.2 OKF 包的五大核心区段
我设计的 OKF 知识包包含五个区段,每个区段解决 Agent 的一类问题:
| 区段 | 作用 | 解决 Agent 的什么问题 |
|---|---|---|
| datasource | 数据源基本信息、方言、时区、连接建议 | 不知道数据库类型,生成的 SQL 方言可能不兼容 |
| schema | 表结构、字段类型、主外键、候选键 | 不知道表有什么字段,无法构造合法查询 |
| semantics | 字段描述、枚举字典、业务规则、单位与格式 | 不知道字段含义和取值范围,SQL 语义混乱 |
| relations | 表间关系、join 条件、基数与业务含义 | 不知道怎么正确关联多张表 |
| samples | 数据样例、示例查询 question-sql 对 | 不确定字段值长什么样,不敢给出具体过滤条件 |
这五个区段不是平级的关系。schema是骨架,semantics是血肉,samples是神经末梢。Agent 最依赖的是semantics和samples,因为这两个区段承载了业务知识,而业务知识是普通数据库导出台账里根本没有的。
2.3 一段真实的 OKF 包示例
下面是我在模拟项目 X 里用的一个订单库的知识包片段,字段级描述、关系描述和示例查询是核心。
datasource: name: order_system dialect: postgresql timezone: Asia/Shanghai description: 订单业务核心库,包括用户、订单、订单明细三张表。 schema: tables: - name: users description: 用户基础信息表 columns: - name: uid type: number description: 用户唯一标识(内部编号) tags: [pk, pii] - name: reg_time type: datetime description: 注册时间 - name: is_vip type: enum description: 是否会员 values: 0: 普通用户 1: 会员用户 semantics: business_rules: - table: orders field: status values: 0: 待支付 1: 已支付 2: 已发货 3: 已完成 4: 已取消 default_note: 新订单默认 0,支付成功后由支付回调更新为 1。 relations: - source_table: orders target_table: users on_keys: - source: uid target: uid cardinality: many_to_one business_meaning: 一个用户可以有多个订单,订单归属于注册用户。 samples: query_examples: - question: 查询最近 30 天已支付订单的总金额 sql: SELECT SUM(amount) FROM orders WHERE status = 1 AND pay_time >= NOW() - INTERVAL '30 days' explanation: status=1 表示已支付,金额字段 amount 单位是分,需要除以 100 转为元。这个示例里最关键的设计细节是explanation字段。它不只是解释 SQL 本身,还解释了业务字段的实际含义(金额单位是分、status 的枚举含义)。Agent 在生成 SQL 时会把这种解释作为 few-shot 参考,大大提高了准确性。
3. Python 编译器实现——把“读取数据库”变成“编译知识”
OKF 格式定了之后,最难的部分就是编译器本身。我不会贴完整项目代码,而是把编译器四个阶段里最核心的模块拆开讲,每个模块给关键代码。
3.1 编译器主流水线
编译器的主流程我写成了一条清晰的流水线,每个阶段输入输出都是 Python 字典,中间结果可以落盘调试。这是主函数的结构:
def compile_database(config_path: str, output_path: str) -> dict: # 阶段1:读取配置并建立连接 db_config = load_config(config_path) engine = create_db_engine(db_config) # 阶段2:词法+语法分析——提取原始元数据并规整 metadata = MetadataLoader(engine).load() # 表、列、类型、约束、索引 # 阶段3:语义分析——命名解码、术语表映射、PII 检测 semantic = SemanticAnalyzer(metadata, term_table=load_terms()).analyze() # 阶段4:目标代码生成——组装 OKF 并校验 okf_doc = OkfWriter(semantic).write() validate_okf(okf_doc) # 输出 save_yaml(okf_doc, output_path) return okf_doc实际执行的时候,我会在MetadataLoader和SemanticAnalyzer之间加一个中间结果存档,这样如果后面生成的知识包有问题,可以快速定位是结构提取的问题还是语义注释的问题。
3.2 词法和语法层:元数据提取与命名解码
元数据提取依赖数据库自带的系统表。以 PostgreSQL 为例,核心是查information_schema.columns和information_schema.tables。方言之间有差异,MySQL 的information_schema也有类似的表,但类型名称不一样,需要做一层归一化。
import sqlalchemy as sa from sqlalchemy import inspect class MetadataLoader: def __init__(self, engine): self.inspector = inspect(engine) def load(self) -> list: tables = [] for tname in self.inspector.get_table_names(): columns = self.inspector.get_columns(tname) pk = self.inspector.get_pk_constraint(tname) fks = self.inspector.get_foreign_keys(tname) tables.append({ "name": tname, "columns": [{ "name": col["name"], "type": normalize_type(str(col["type"])), # 类型归一化 "nullable": col.get("nullable", True), "default": col.get("default"), } for col in columns], "primary_keys": pk.get("constrained_columns", []), "foreign_keys": [{ "source": fk["constrained_columns"], "target_table": fk["referred_table"], "target_columns": fk["referred_columns"], } for fk in fks], }) return tables这里的normalize_type是词法分析阶段的关键。因为不同数据库返回的类型名五花八门,INTEGER、INT、BIGINT、NUMBER(10,2)在我们内部只关心是数值型、字符串型还是时间型。我会用一套映射规则把类型归一成四类:number、text、datetime、enum。同时,字段名拆词也属于词法分析的一部分。我写了一个命名解码器,用于把crt_tm、userID、order_number这类字段名拆成可读的词组:
import re TERM_TABLE = { "crt": "创建", "tm": "时间", "uid": "用户ID", "amt": "金额", "qty": "数量", "sts": "状态", } def decode_field_name(field_name: str) -> str: # 统一分隔 camelCase 和 snake_case,如 userID -> user_ID spaced = re.sub(r"(?<=[a-z])(?=[A-Z])", "_", field_name) parts = spaced.replace("-", "_").lower().split("_") decoded = [TERM_TABLE.get(p, p) for p in parts] return "_".join(decoded)这个函数返回的仍然是英文词元序列,但已经能拆出创建_时间这样的相对可读形式。真正的中文描述由语义层结合术语表生成,比如通过“术语 = 中文描述”的静态映射:reg_time->注册时间。词法层只负责拆词,不负责理解。
3.3 语义层:描述生成不能全依赖大模型
网上很多方案喜欢用大模型直接给字段生成描述。我一开始也这么干,结果发现把 200 个字段扔给模型,不仅慢,而且有些字段描述互相矛盾,比如同一个uid在订单表里被解释成“用户统一标识”,在日志表里被解释成“操作人编号”,而这两个字段其实语义相同。所以我把语义生成做成了三层策略:
第一层是静态术语表,人工维护最核心、最常用的字段名映射。这里的字段出现频率高、含义稳定,人工维护成本可控。术语表命中后直接使用。
第二层是规则启发式。比如字段名里带time、date、at结尾的一律默认是时间字段;带amount、price、fee的默认金额字段并自动补充“单位:分/元”的说明;带phone、email、id_card的默认打上pii标签。规则能覆盖 70% 以上的通用情况。
第三层才轮到 LLM 补全,而且我加了一道“LLM 只补 custom 描述、不碰核心定义”的约束。也就是说,模型只能为它的输出提供一个候选描述池,最终是否采纳还需要人工确认,或者用置信度阈值过滤。这样做的好处是,即使模型输出有偏差,也不会污染整个知识包的基础语义。
class SemanticAnalyzer: def __init__(self, term_table: dict, pii_patterns: dict): self.term_table = term_table self.pii_patterns = pii_patterns def analyze_field(self, field: dict) -> dict: name = field["name"] desc = self.term_table.get(name) tags = [] if any(p.search(name) for p in self.pii_patterns.get("pii", [])): tags.append("pii") if "time" in name or name.endswith("_at"): tags.append("time") return { "name": name, "type": field["type"], "description": desc or f"字段 {name},待补充业务描述", "tags": tags, }注意这里的字段描述兜底文案。如果不写兜底,后面 Agent 读取知识包时会出现空描述字段,LLM 容易忽略它。有了“待补充业务描述”的占位,至少 Agent 会知道这个字段的存在意义尚未明确,不至于完全瞎猜。
3.4 目标代码生成:输出 OKF 包并校验
到了这一步,知识包内容已经全部在 Python 字典里了。写入 YAML 之前必须做几件小事:字段排序、嵌套结构清理、YAML 转义处理。我吃过亏的是 YAML 里有特殊字符(比如描述里带冒号)导致缩进解析错误,所以统一用yaml.safe_dump配合sort_keys=False输出,并且重点字段先做一次合法化检查。
为了让生成的知识包更“耐打”,我会用 JSON Schema 做一次结构校验。下面是一个简化的校验规则片段:
import jsonschema OKF_SCHEMA = { "type": "object", "required": ["datasource", "schema", "semantics", "relations", "samples"], "properties": { "datasource": {"type": "object"}, "schema": { "type": "object", "properties": { "tables": { "type": "array", "items": { "type": "object", "required": ["name", "columns", "description"], }, } }, }, }, } def validate_okf(doc: dict): jsonschema.validate(doc, OKF_SCHEMA) total_tables = len(doc["schema"]["tables"]) total_cols = sum(len(t["columns"]) for t in doc["schema"]["tables"]) total_relations = len(doc["relations"]) print(f"OKF 校验通过:{total_tables} 张表,{total_cols} 个字段,{total_relations} 条关系")这个校验函数也会输出统计信息。实际项目里我把它作为 grep 的锚点,每次生成完知识包先看这个统计是否正常,比如某张表字段数为 0,那肯定是提取阶段出了问题。数据的“可观测性”是排查问题的基础,这一点在编译器里一样成立。
4. 完整实战:从订单库到 OKF 包
理论说再多,不如跑一遍完整流程。我拿模拟项目 X 的订单系统数据库作为案例,从零开始构建 OKF 知识包。这个库一共三张表:用户表、订单表、订单明细表,规模不大,但足以演示整个流程。
4.1 目标设定:让 Agent 能回答业务问题
我们设定的验收标准是:Agent 能够直接查询“最近 30 天已支付订单的复购用户占比”这类需要多表 join、枚举判断和业务语义理解的复合问题。
先看原始数据库结构:
| 表名 | 关键字段 | 原始问题 |
|---|---|---|
| users | uid, reg_time, is_vip | is_vip取值含义不明 |
| orders | uid, amount, status, pay_time | status是枚举,但含义未知;amount 单位未知 |
| order_items | oid, sku_id, qty, price | price是单价还是成交价,未知 |
这三张表在数据库里没有任何外键约束,业务方说“我们代码里手动关联的”,这恰好是 Agent 最容易翻车的地方。整个知识包构建分五步走。
4.2 实际操作步骤
第一步,写数据库连接配置,然后运行编译器:
# config/order_system.yaml host: localhost port: 5432 database: order_system user: read_only_user password: <token>python compile_database.py \ --config config/order_system.yaml \ --output output/order_system.okf.yaml第二步,查看中间产物。编译器会在output/下生成decoded_fields.json和pii_report.json。decoded_fields.json里能看到词法层对每个字段的拆词结果,比如pay_time被拆成pay 和 time,正常;oid被拆成o 和 id,我一看就觉得不对劲,排查后确认这是订单明细表的主键,应该在术语表里补充映射为“订单明细ID”。
第三步,把术语表补全。我打开terminology.json,添加了订单域的核心术语,比如:
{ "oid": "订单ID(orders 表主键)", "sku_id": "商品SKU ID", "qty": "商品数量", "price": "商品成交单价(元)" }这个步骤本质上是在“教”编译器认识业务词汇,也是语义层质量的关键。术语表越全,后面生成的知识包描述就越准确。
第四步,注入人工维护的业务规则。我打开生成的 OKF 草稿,手动补上orders.status枚举字典和amount的单位说明。这些信息来自我对业务方的一次访谈,30 分钟聊完,比让模型猜强得多。
第五步,重新运行编译器,这次会生成最终的 OKF 包。确认输出统计正常后,把它喂给 Agent 做实测。
4.3 Agent 实测效果对比
我做了两个对照组:一组是 Agent 直接连数据库读取 information_schema,另一组是 Agent 先读 OKF 知识包再生成 SQL。测试了 20 个业务问题,结果差异非常明显:
| 对比项 | 直接读 schema | 读 OKF 知识包 |
|---|---|---|
| SQL 语句正确率 | 35% | 85% |
| 平均生成 SQL 耗时 | 8 秒 | 5 秒 |
| 是否需要人工修正数据单位 | 是,金额单位猜错 | 否,知识包已声明单位 |
| 对“已支付订单”的理解 | 多种猜测 | 准确映射 status=1 |
比如“最近 30 天已支付订单总金额”,Agent 直接读库时给出的 SQL 是:
SELECT SUM(amount) FROM orders WHERE status = 'paid' AND created_at >= NOW() - INTERVAL '30 days'问题很明显:status的真实取值是数字 1,不是'paid';时间字段是pay_time,不是created_at。而使用 OKF 包后,Agent 的生成结果是:
SELECT SUM(amount) / 100.0 FROM orders WHERE status = 1 AND pay_time >= NOW() - INTERVAL '30 days'金额单位、时间字段、枚举取值全部正确。这个案例直观说明,知识包不是给数据库加一层壳,而是真正把业务知识编码进了 Agent 的上下文。
4.4 编译器性能优化
构建过程中还有一个性能问题值得说:如果数据库表非常多,比如几百张表,每张表都去采样一次会非常慢。我做了两个优化:
一是采样并行化。对互相不依赖的表并行执行SELECT采样,用concurrent.futures.ThreadPoolExecutor控制并发数,实测 50 张表的采样时间从 3 分钟降到 40 秒。
二是采样限制策略。大表只取按主键采样的 50 行,小表可以全量取。识别大小表可以查表的估算行数,而不是 count(*),因为在超大表上count(*)本身就可能是个灾难。
class SampleCollector: def __init__(self, engine, max_rows=50, timeout=5): self.engine = engine self.max_rows = max_rows self.timeout = timeout def sample(self, table: str, estimated_rows: int) -> list: if estimated_rows > 100000: sql = f"SELECT * FROM {table} TABLESAMPLE SYSTEM (1) LIMIT {self.max_rows}" else: sql = f"SELECT * FROM {table} LIMIT {self.max_rows}" try: with self.engine.connect() as conn: return [dict(r) for r in conn.execute(sa.text(sql))] except Exception as e: print(f"采样 {table} 失败:{e}") return []注意 MySQL 5.7 之前的版本不支持TABLESAMPLE,PostgreSQL 支持,所以在方言处理上需要根据数据源类型切换采样语法。这也是编译器要保留“数据源方言”字段的原因之一。
5. 常见问题与排查技巧实录
运行编译器构建 OKF 包的过程中,我记录了五个高频问题和对应的排查方法,这里整理成速查表。
| 现象 | 根因 | 解决方案 |
|---|---|---|
| 采样超时导致整包失败 | 大表未限制采样行数 | 加LIMIT,大表改用 TABLESAMPLE,把单表超时控制在 5 秒内 |
| 字段描述大量为“待补充” | 领域术语表覆盖不足 | 增加词汇表维护流程,先把高频字段补全再生成 |
| 知识包体积过大,Agent 塞不进上下文 | 表数量太多或描述冗余 | 分层打包:核心表生成完整包,次要表只保留 schema 摘要 |
| LLM 补全描述与业务定义冲突 | 没有约束模型 | 禁用 LLM 直接写核心描述,只允许在规则未命中时生成候选 |
| 同一字段在不同表描述不一致 | 缺乏全局术语表 | 语义分析前先做全量字段名归并,统一术语再落到单表描述 |
5.1 知识包体积控制的三种手段
第一个问题值得展开。当数据库有 300 张表时,完整 OKF 包的 YAML 文本量可能超过 100KB,这会突破多数 Agent 上下文窗口的舒适区。我采用的解决思路是分层知识包:核心业务表生成完整五段式 OKF,周边配置表只生成 schema 摘要加一个 description。
另一个方法是精简samples区段。示例查询是全包最贵的部分,一条 question-sql 对就要几百 token。我的经验是只保留三类示例:多表 join 类、枚举过滤类、聚合函数类,每类 3-5 条就足够。
第三个手段是“按需加载”。在 Agent 的工具设计中,不要让它一次读取全部知识包,而是先读一个 1KB 的索引包,等它确定了要查询哪几张表,再动态加载对应表的详细描述。这个方案我在后续迭代中实测效果最好。
5.2 命名解码失败时的元凶
踩过一次非常隐蔽的坑:所有字段的拆词结果都正常,但描述仍然混乱。排查后发现,问题出在字段名里的下标或特殊字符上。比如某个历史表里有个字段叫col_1、col_2,词法层完全无法解码,但术语表又没有兜底。后来我增加了一条规则:无法解码的字段一律打上unknown_semantic标签,并在报告里单独列出,人工逐个确认。
这个方法的核心在于“让问题可见”。编译器不追求一次完美,而是把不确定性都暴露出来。未知语义的字段多一个标签,Agent 就会少一次瞎猜。
5.3 增量更新而不是每次全量生成
数据库结构不是不变的,业务上经常加字段、改枚举。如果每改一次都要全量重新生成知识包,耗时且容易引入新错误。我的方案是给知识包加一个generated_at时间戳和schema_version,然后做增量比对:对比当前元数据和上一版 OKF 的 schema 差异,只对有变化或新增的部分重新做语义分析和样本采集。
增量比对的核心代码如下:
def diff_schema(old_tables: list, new_tables: list) -> list: changes = [] old_map = {t["name"]: t for t in old_tables} for nt in new_tables: if nt["name"] not in old_map: changes.append({"type": "add_table", "table": nt["name"]}) continue old_cols = {c["name"] for c in old_map[nt["name"]]["columns"]} new_cols = {c["name"] for c in nt["columns"]} added = new_cols - old_cols if added: changes.append({"type": "add_columns", "table": nt["name"], "columns": list(added)}) return changes增量比对的输出可以作为知识包的更新日志。我实际用下来的体感是:日常 80% 的变更只是加字段,增量模式 10 秒内能完成更新,而全量生成至少要跑 2 分钟。
6. 调试工具链与个人体会
最后分享一点我在这个项目里沉淀的调试方法。把它当“编译器”来做的最大好处是,每一层都有可观测的中间产物,像工程上的分阶段日志一样,哪一步出错一目了然。
我习惯在词法层输出“字段拆词报告”,看一眼就能发现命名规则异常;在语义层输出“PII 检测报告”,防止样本数据里混入身份证号、手机号等敏感信息;在代码生成层保留“OKF 校验统计”,确保生成的包结构完整。这三个报告形成了一条完整的诊断链路。
另一个我很受用的调试技巧是:生成完 OKF 包后,我会立刻写三个“冒烟测试查询”喂给 Agent。这三个问题分别考察单选查询、多表 join 和聚合统计。如果冒烟测试全过,再扩大到完整业务问题集。相当于编译器测试中的单元测试用例,可以快速回归。
回头看,这个项目最大的收益不只是让 Agent 生成了更准确的 SQL,而是让我重新审视了“数据库知识”的形态:它不该只是散落在 DDL 和业务代码里的碎片,而应该被组织成一套可以被程序化读取和更新的结构化知识。OKF 知识包在某种程度上充当了数据库与 LLM 之间的“汇编层”,把低级的表结构“编译”成了高级的业务语义。如果你正在做 Agent 的数据接入,不妨试试这套思路,先定义你需要什么格式,再写一个 Python 编译器把数据库“翻译”过去。