资讯详情

资讯详情

用Python写数据库编译器:构建Agent可读的OKF语义知识包

去年我在给某内部项目接入 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_tmAgent 很难猜出这是“创建时间”订单表里的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: status1 表示已支付金额字段 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_tableload_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_keysFalse输出并且重点字段先做一次合法化检查。为了让生成的知识包更“耐打”我会用 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(fOKF 校验通过{total_tables} 张表{total_cols} 个字段{total_relations} 条关系)这个校验函数也会输出统计信息。实际项目里我把它作为 grep 的锚点每次生成完知识包先看这个统计是否正常比如某张表字段数为 0那肯定是提取阶段出了问题。数据的“可观测性”是排查问题的基础这一点在编译器里一样成立。4. 完整实战从订单库到 OKF 包理论说再多不如跑一遍完整流程。我拿模拟项目 X 的订单系统数据库作为案例从零开始构建 OKF 知识包。这个库一共三张表用户表、订单表、订单明细表规模不大但足以演示整个流程。4.1 目标设定让 Agent 能回答业务问题我们设定的验收标准是Agent 能够直接查询“最近 30 天已支付订单的复购用户占比”这类需要多表 join、枚举判断和业务语义理解的复合问题。先看原始数据库结构表名关键字段原始问题usersuid, reg_time, is_vipis_vip取值含义不明ordersuid, amount, status, pay_timestatus是枚举但含义未知amount 单位未知order_itemsoid, sku_id, qty, priceprice是单价还是成交价未知这三张表在数据库里没有任何外键约束业务方说“我们代码里手动关联的”这恰好是 Agent 最容易翻车的地方。整个知识包构建分五步走。4.2 实际操作步骤第一步写数据库连接配置然后运行编译器# config/order_system.yaml host: localhost port: 5432 database: order_system user: read_only_user password: tokenpython 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: 订单IDorders 表主键, 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 秒是否需要人工修正数据单位是金额单位猜错否知识包已声明单位对“已支付订单”的理解多种猜测准确映射 status1比如“最近 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_rows50, timeout5): 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 fSELECT * FROM {table} TABLESAMPLE SYSTEM (1) LIMIT {self.max_rows} else: sql fSELECT * 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 之前的版本不支持TABLESAMPLEPostgreSQL 支持所以在方言处理上需要根据数据源类型切换采样语法。这也是编译器要保留“数据源方言”字段的原因之一。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 编译器把数据库“翻译”过去。
觉得有用,分享给同行:

为您的企业打造数字门面

稳重轻奢商务风格,端正雅致视觉,长效耐看不易过时。

立即咨询 →