1. 为什么数据库要"Agent就绪":从裸连接到知识包
先说个我近期踩过的坑。某个内部项目需要让智能体(Agent)替业务同学查询"上个月华东区订单量TOP10的商品",一开始我直接把数据库连接信息、表结构文档丢给Agent,让它自己"看着办"。结果惨不忍睹:Agent要么把时间字段当成字符串拼接,要么把"华东区"映射到根本不存在的城市字典,要么生成了一堆自认为正确但逻辑完全跑不通的SQL。问题不在于Agent笨,而在于我根本没有把数据库"怎么用"这件事说清楚。
这让我重新思考了一个问题:当我们说"让Agent访问数据库"时,到底应该给Agent什么?
最粗浅的做法是给连接串加一句"你去查吧",这相当于让一个刚入职的新人直接面对几万张表的线下库,既没告诉他业务口径,也没告诉他表之间的血缘关系,更没告诉他哪些字段是废弃的。稍微好一点的做法是写一份静态的Markdown文档,把核心表、关键字段列一列。但文档是给"人"看的,Agent读起来效率极低——它要在几千行文档里自己找线索,而且文档很难与运行时数据、权限体系、SQL风格对齐。
我最终采用的做法,是把数据库的"知识"编译成一种结构化的、可检索的、带语义标注的知识包。我管它叫OKF知识包(Open Knowledge Format for Agent),然后用Python写了一个编译器,把数据库元数据、业务术语、查询样例、权限约束等内容,自动打包成Agent可以直接消费的JSON格式。这样做的好处是:知识包既是给Agent看的"数据库使用说明书",也是给上层检索系统用的"语义索引",还能在生成SQL时充当few-shot示例库。
这个方案解决的核心问题,就是把"如何正确使用这个数据库"从隐性的、分散的、人的经验里,转变成显性的、结构化的、机器可读的知识资产。它适合谁?如果你也在做Agent+数据库方向的项目,或者你想让自己的RAG系统能真正回答"带条件的业务查询",而不是只能做文档摘要,这篇实战记录会很有参考价值。下面我把整个设计思路和编译器的实现过程拆开讲。
2. OKF知识包的Schema设计:结构决定Agent好不好用
知识包不是简单地把数据库建表语句抄一遍。我前两版做得太糙,直接把CREATE TABLE语句和字段注释倒进JSON里,结果Agent虽然能看到字段名,却不知道"这个字段在业务上到底意味着什么""它和别的表怎么关联""什么样的查询条件是合法的"。后来我重新设计了Schema,围绕"Agent生成的每一个SQL都需要回答哪些问题"来倒推结构。
2.1 元数据区:表、字段、约束、索引的规范化描述
第一个区块是元数据区,负责描述数据库的物理结构,但比information_schema更挑重点。我不建议把每张表的每个字段都塞进去,那样知识包会膨胀到几MB,检索时反而把关键信息淹没。我采用的策略是按热度分层:核心业务表全量描述,维表、日志表只保留表级说明和主外键关系。
字段级的描述需要包含以下内容:
field_name:物理字段名field_type:物理类型,如VARCHAR(64)comment:原始注释(如果有)semantic_alias:语义别名列表,比如user_id的别名可以是用户ID、用户编号、uidenum_values:如果字段是枚举值,列出合法取值和对应中文含义unit:单位,比如金额类的分、元,避免Agent把单位搞混is_sensitive:是否敏感字段,比如手机号、身份证号,标记后Agent应避免直接返回原始值
索引和约束信息我会单独抽一个join_hints小节,专门描述表间常用的关联路径。比如"订单表可以通过buyer_id关联用户表,也可以通过seller_id关联商家表"。这一节看着简单,但它能显著减少Agent在生成多表JOIN时"发明"不存在的关联条件。
2.2 语义区:业务词汇、同义词与问法模板
这是OKF知识包和普通数据字典最大的区别。数据库里的字段名往往是英文缩写,业务同学习惯说"客单""动销""毛利",Agent如果没有语义映射,根本不知道这些词对应哪些字段。
我在语义区设计了三层结构:
第一层是业务词表(glossary),每个词条包含三部分:term(业务词)、canonical_field(规范字段路径)、definition(一句话定义)。比如:
{ "term": "客单价", "canonical_field": "order_summary.avg_amount", "definition": "单笔订单的平均成交金额,等于总成交金额除以订单总数" }第二层是同义词网络(synonyms),把不同说法映射到同一个标准概念上。这里要注意的是,同义词膨胀会带来检索污染,所以我会给同义词加confidence权重,高权重的词才能在向量检索时被重点召回。比如"客户"和"用户"权重高,"主顾""买主"权重低,避免让Agent形成错误结论。
第三层是问法模板(query_patterns),基于历史真实查询整理,描述"用户大概会怎么问、对应的SQL骨架是什么"。比如:
question: 本月销售额前五的品类 sql_pattern: SELECT category, SUM(amount) AS total_sales FROM orders WHERE order_time >= DATE_TRUNC('month', CURRENT_DATE) GROUP BY category ORDER BY total_sales DESC LIMIT 5这一层其实就是给Agent喂了几十到几百条"标准答案",让它在面对类似问题时有据可依,而不是每次都从零推断。
2.3 样例区:代表性SQL与返回结果模板
样例区是给Agent做few-shot学习用的。我通常会挑三类样例:
- 简单查询:单表、带WHERE过滤、带GROUP BY统计
- 复杂查询:多表JOIN、子查询、窗口函数
- 边界样例:需要特殊处理的口径,比如"不含退款订单""只看成功支付订单"
每个样例除了SQL本身,我还会附上返回结果的字段模板。这很重要,因为Agent拿到查询结果后需要自己转述成自然语言,如果它不知道结果集长什么样,就会乱编字段名。举个例子:
{ "query_id": "sample_021", "question": "查询最近7天各渠道的注册转化率", "sql": "SELECT channel, COUNT(DISTINCT user_id) AS new_users, SUM(CASE WHEN register_status = 1 THEN 1 ELSE 0 END) AS valid_users FROM user_register_log WHERE register_date >= CURRENT_DATE - INTERVAL '7 days' GROUP BY channel", "result_template": [ {"column": "channel", "description": "渠道名称"}, {"column": "new_users", "description": "新增注册用户数"}, {"column": "valid_users", "description": "有效注册用户数"}, {"column": "conversion_rate", "description": "有效注册用户数/新增注册用户数", "computed": true} ] }有个容易忽略的细节:result_template里如果包含计算字段,一定要标注computed: true并写明计算逻辑,否则Agent会把转化率误会成数据库里真实存在的字段。
2.4 版本与权限信息
最后是版本区和权限区。版本区用一个schema_version字段记录当前知识包对应的数据库版本,配合compiled_at时间戳,方便后续增量更新和对齐。权限区则是把数据库里不同的访问权限沉淀成"角色-可见数据范围-允许操作"的描述,比如"运营角色的Agent只能查询脱敏后的用户数据,不能执行DELETE/UPDATE"。这一块在单机演示时体现不出价值,但一旦接入生产库,没有权限描述的知识包会让Agent踩大坑——它可能生成删除语句,哪怕底层账号没权限,也会白白浪费一次调用。
3. Python编译器实战:从数据库元数据到知识包的完整流水线
Schema设计得再好,没有工具把它自动化生成出来,都是废纸。接下来是重头戏:我用Python写了一个知识包编译器,输入是一个数据库连接,输出是一个校验过的OKF JSON文件。整个过程可以拆成五个阶段:抽取、清洗、映射、打包、校验。
3.1 环境与工具选型:为什么我选Python做编译器
选Python做编译器有几个现实原因。一是生态里有非常成熟的数据库元数据读取库,不需要自己解析information_schema;二是Agent这边天然是Python生态,知识包后续要接向量化和函数调用都方便;三是我需要用pydantic做Schema校验,用jinja2做模板渲染,这两个库都足够稳定。
我第一版用的是原生sqlite3模块,后来切到了SQLAlchemy来统一元数据抽取接口。原因很直接:SQLAlchemy的inspect()接口能拿到统一的表、列、主外键、索引结构,不用为每一种数据库写一套解析逻辑。虽然标题里是"编译器实战",但本质上它做的是"把数据库元数据翻译成知识包格式"的编译工作,SQLAlchemy恰好提供了理想的中间表示。
3.2 抽取阶段:把数据库结构变成中间表示
抽取阶段的目标是拿到原始元数据。我这里直接给一段核心代码,大家在本地跑的时候只要改一下连接串就行:
from sqlalchemy import create_engine, inspect engine = create_engine("sqlite:///demo.db") insp = inspect(engine) tables_raw = {} for table_name in insp.get_table_names(): columns = insp.get_columns(table_name) pk = insp.get_pk_constraint(table_name) fks = insp.get_foreign_keys(table_name) indexes = insp.get_indexes(table_name) tables_raw[table_name] = { "columns": columns, "primary_key": pk, "foreign_keys": fks, "indexes": indexes, }到这里我只做了第一步。真正的工作量在后面的清洗和映射,因为数据库里很多信息是缺失的,比如字段注释为空、外键命名不规范、枚举值散落在代码里。这些都不能直接进知识包。
3.3 清洗阶段:补注释、归一化类型、过滤废弃表
清洗阶段我做了三件事。
第一,字段注释缺失补全。如果原始注释为空,我会去业务词表里反向匹配:如果字段名和某个term的命名规范一致,就把这个词的定义作为候选注释;如果匹配不上,就保留空,等映射阶段手动补。注意这里不要硬编AI猜测,宁缺毋滥。
第二,类型归一化。把不同数据库的类型名归一成统一形态,比如VARCHAR(64)、NVARCHAR2(64)统一成string(64),TIMESTAMP WITH TIME ZONE统一成datetime_tz。这样做不是要丢掉原始信息,而是让Agent在生成SQL时不至于因为类型名差异产生困惑。
第三,过滤废弃表。开发中经常有_tmp、_bak结尾的表,或者最近三个月内last_accessed都为空的历史表。这些表如果进知识包,只会增加Agent的检索噪音。我维护了一个table_filter_rules配置,支持按前缀、后缀、访问热度过滤。
3.4 映射阶段:让物理结构"说人话"
映射是整个流水线里最依赖业务经验的部分。我会在编译器之外维护一个mapping_config.yaml,里面存四类映射规则:
field_aliases: user_id: - 用户ID - 用户编号 - 顾客ID order_status: - 订单状态 enum_values: order_status: "0": 待支付 "1": 已支付 "2": 已发货 "3": 已完成 business_terms: 销售额: canonical: orders.total_amount definition: 订单支付成功后的实付金额总和 table_joins: orders-user: left: orders.buyer_id right: users.user_id comment: 用户维表关联编译器在打包时会读取这个配置,自动把语义信息注入到抽取结果里,生成带语义标注的中间结构。这块我会强烈建议大家把配置文件和代码分离,因为业务词表是持续更新的,如果硬编码在Python里,每改一次都要重新发布代码。
3.5 打包与校验:生成OKF JSON并通过Schema检查
映射完成后,进入打包阶段。我用pydantic定义了一个OKFKnowledgePackage模型,它把前文说的元数据区、语义区、样例区、版本区全部串起来。打包时用model_dump()转成dict,再json.dump()写出。校验是关键一步,除了pydantic本身的结构校验,我还写了几条自定义业务校验规则:
- 所有
canonical_field必须真实存在于元数据区,否则说明语义区有"脏数据" - 所有
sql_pattern中的表名和字段名必须与元数据区一致 - 所有枚举值映射必须字段类型可覆盖
- 样例区SQL应该用连接账号做一次
EXPLAIN验证,语法不通的直接剔除
这一步的好处是:知识包如果编译错了,在发布前就会被拦住,不会等到Agent用的时候才暴露。我建议每个CI流程里都加上这一步检查,宁可慢两秒,别让坏包上线。
下面是一段简化后的打包核心逻辑:
from pydantic import BaseModel, Field, ValidationError class FieldMeta(BaseModel): field_name: str field_type: str comment: str = "" aliases: list[str] = Field(default_factory=list) enum_values: dict[str, str] = Field(default_factory=dict) is_sensitive: bool = False class TableMeta(BaseModel): table_name: str description: str = "" fields: list[FieldMeta] join_hints: list[str] = Field(default_factory=list) class OKFKnowledgePackage(BaseModel): schema_version: str compiled_at: str database_alias: str tables: list[TableMeta] glossary: list[dict] query_patterns: list[dict] sample_queries: list[dict] access_control: dict = Field(default_factory=dict) def validate_references(self): table_names = {t.table_name for t in self.tables} field_pairs = { (t.table_name, f.field_name) for t in self.tables for f in t.fields } for pattern in self.sample_queries: for ref_table in pattern.get("referenced_tables", []): if ref_table not in table_names: raise ValueError(f"unknown table: {ref_table}") for ref_field in pattern.get("referenced_fields", []): if tuple(ref_field) not in field_pairs: raise ValueError(f"unknown field: {ref_field}")3.6 增量编译与缓存策略
最后是工程化层面。生产环境的库可能每天都有DDL变更,全量编译一次耗时太长。我采用了两级增量策略:结构级增量和语义级增量。
结构级增量:每天定时对比information_schema里表的数量、字段数量、字段类型的指纹(比如取MD5哈希表),如果指纹没变,说明物理结构没变,跳过抽取阶段。语义级增量:业务词表和样例区单独维护版本号,每次更新只重新生成语义区和样例区,然后拼接缓存的元数据区,最后做一次全量校验。
实测下来,一个包含800多张表、上万字段的中型库,全量编译需要40秒左右,增量编译只要2~3秒。对于Agent应用来说,这个编译频率完全够用,甚至可以做到每次发布前自动编译一次。
4. Agent消费侧对接:RAG检索与Function Calling的实测
知识包编译好只是第一步,真正关键是Agent怎么用起来。我在实测中走了两条路线:一条是把知识包切片后向量化,走RAG检索;另一条是把知识包生成成OpenAI Function Calling风格的工具描述,让Agent在需要时调用。两条路线各有利弊,下面说下实测数据。
4.1 路线一:知识包切片 + 向量检索
这条路线适合Agent本身不直接生成SQL,而是先查知识库、再借助检索结果生成答案的场景。我把OKF JSON里的每个表描述、每个字段描述、每条样例SQL切分成独立的检索单元,然后分别向量化。
切片粒度是我调试了很久才定下来的。最初我按"表"切片,结果一个表的所有字段塞进一个向量里,检索"华东区"时经常召回整个订单表,噪音很大。后来改成**"字段+表"双级切片**:字段级的切片负责精确召回,表级的切片负责兜底,防止Agent完全找不到入口。检索时用两级得分加权,字段级得分占比0.7,表级占比0.3。
实测效果很直观。同样的问题"最近一个月分城市的退款率排名",在不加知识包的情况下,裸提示词版的Agent给出的SQL是SELECT city, COUNT(*) FROM refund WHERE date >= ...,既没算退款率,也没和订单表关联;加了知识包之后,Agent能准确找到退款事实表和订单维表的JOIN路径,还会顺手带上refund_amount IS NOT NULL这个过滤条件。
4.2 路线二:知识包生成Function Calling描述
这条路线的做法是把"查询数据库"这件事封装成一个函数,函数参数用JSON Schema描述,描述里的枚举值、字段说明、示例全部取自知识包。Agent在对话过程中自己决定要不要调用这个函数、参数填什么。
实测下来有个细节特别关键:函数描述里的枚举值必须来源于知识包语义区,而不是数据库物理字段。比如order_status字段物理存储是"0","1","2","3",如果直接把这个枚举值亮给Agent,它可能填写"0",但用户问的是"待发货订单",Agent没做映射就会错。正确的做法是让函数参数接收业务值,函数内部再通过知识包的enum_values映射转换成物理值。
{ "name": "query_order_metrics", "description": "根据业务条件查询订单指标,适合回答销量、销售额、订单量等问题", "parameters": { "type": "object", "properties": { "status": { "type": "string", "enum": ["待支付", "已支付", "已发货", "已完成"], "description": "订单状态,不传表示全部" }, "city": { "type": "string", "description": "城市名称,如华东区城市列表" }, "metric": { "type": "string", "enum": ["订单量", "销售额", "退款率", "客单价"], "description": "要计算的指标" } } } }这条路线的优势是稳定,Agent不会生成乱七八糟的SQL,因为它只能在函数参数里选;劣势是灵活度有限,遇到"用户自己发明的新指标"就无能为力了。所以我现在的推荐方案是混合模式:冷门问题走RAG检索知识包,热门问题走函数调用。
4.3 实测效果对比:一个简单的评测表
我搭了一个20条业务问题的小评测集,问题覆盖单表统计、多表JOIN、带时间范围、带同义词映射、带排除条件等类型。对比三种模式,结论很清晰:
| 模式 | 回答正确率(人工评估) | 平均推理耗时 | 问题类型覆盖度 |
|---|---|---|---|
| 裸提示词(只给连接串和一句话) | 30% | 8秒 | 低,只覆盖简单单表查询 |
| 静态Markdown文档注入 | 55% | 12秒 | 中,多表JOIN易错 |
| OKF知识包 + 混合消费 | 85% | 9秒 | 高,覆盖大部分业务口径 |
说句公道话,那15%的错误里有一半不是知识包的问题,而是测试问题本身口径模糊,比如"销售额"到底是含税还是不含税,业务方自己都说不清。知识包能解决的,是把那些已经明确过口径的业务常识稳定地带给Agent,而不是替业务去发明新口径。
5. 实战中的坑与优化心得
最后这部分是我最想分享的,因为每个坑都是真金白银试出来的。
5.1 字段注释缺失:别让AI去"猜"字段含义
我在清洗阶段遇到过一个问题:一张老表的字段注释几乎全空,字段名也是缩写,比如prd_tp、gmv_amt。第一版我图省事,直接调大语言模型让AI根据字段名猜含义,结果AI把prd_tp猜成"产品类型"(合理解释是product_type),查了代码才知道是"产品投放渠道"。这种错误一旦进入知识包,Agent后续生成的SQL会一直错下去,而且很难追溯。
后来我定了一条硬规矩:原始注释为空、且无法从映射配置里匹配到业务词的字段,一律在知识包标记为needs_human_review,不进自动检索区,只保留在附录区供人工确认。宁可少一点知识,也不能喂"看起来很对但其实是错的"知识。
5.2 同义词膨胀:检索"用户"时把"买主"也召回
同义词是双刃剑。早期我的同步词表里塞了十几个说法,结果Agent检索任何带"用户"的问题时,会同时召回"主顾""买主""会员""注册账号"这些词的向量,导致检索结果里混入大量不相关片段。后来我做了两个改动:
一是同义词分级。只有高频、标准的同义词才参与向量检索;低频、口语化的同义词只保存在语义区,用于Agent生成SQL后的结果解释环节,比如用户说"客官"时,Agent知道对应user表,但不把它作为主要检索词。
二是同义词和字段绑定校验。编译器在打包时会做交叉检查:如果一个同义词在glossary里没有对应的canonical_field,就丢弃。这避免了"有词没法用"的情况。
5.3 版本管理与Schema变更:知识包过期比没有更危险
数据库Schema一旦变更,老知识包里的字段名和JOIN路径就会失效。我遇到过最严重的一次是:某张表把order_amount改成了order_amt,但知识包样例区里还有几条SQL引用旧字段,Agent生成SQL时报错,后来又因为重试机制把同样的错误SQL反复执行了好几遍,白白浪费了资源。
我的解法是把数据库指纹作为知识包版本号的输入。每次编译时都计算一次所有表结构的MD5指纹,如果指纹变了,就把版本号递增,并强制重新编译样例区;同时把样例区SQL统一加一层EXPLAIN校验,语法不过的直接剔除。这个机制运行一个月后,基本没有因为Schema过期导致的Agent执行报错了。
5.4 编译性能:从40秒到2秒的取舍
前面提到增量编译能到2秒,但这不是从天上掉下来的。我第一次用纯information_schema全量扫描时,一个800多张表的中型库花了将近3分钟。后来优化了三处:
- 抽取阶段用SQLAlchemy的
inspect()时,只加载核心表的列级信息,维表和日志表只加载表级信息,启动耗时缩了一半 - 样例区SQL的语法校验从"每条都
EXPLAIN"改成"只校验变更过的SQL",因为历史样例SQL在Schema没变时一定会通过 - 把知识包从单文件拆成"元数据区+语义区"两个文件,检索Agent侧只加载语义区和索引,元数据区按需读取
这里我想提醒一句:不要把编译时间和Agent的响应时间混为一谈。编译是异步的、离线的,可以跑在夜里;Agent消费时读的是已经编译好的知识包文件,这个过程应该控制在毫秒级。如果你发现Agent等知识包生成才回答问题,那一定是架构搞错了。
6. 最后想补充的一点实操体会
做完全部搭建和调试后,我的体会是:OKF知识包这套思路的价值不在于"格式"本身有多先进,而在于它逼着我把数据库知识从"人脑经验"变成了"机器可读的资产"。以前业务同学问"为什么Agent又查错了",我往往只能凭记忆说"它可能不知道这个口径";现在我可以直接打开知识包,查看某个字段的语义映射是否缺失、某条样例SQL是否过期,问题定位速度快了一个数量级。
如果你也想在自己的项目里落地,我建议从最小可行版本开始:先挑两三个最核心的业务表,手工写一份简化的OKF JSON,让Agent先用起来,再逐步把编译器自动化。不要一上来就想覆盖全库,否则你会被无穷无尽的字段注释和同义词问题淹没。
一个小技巧留给你:知识包里的query_patterns样例,最好每隔一个季度回看一次,把那些"过时的写法""不再使用的字段"清理掉。知识包和数据库一样,需要定期维护,否则它也会长出"死代码"。我现在每次发布新版本前,都会顺手跑一遍样例区SQL的EXPLAIN校验,这个习惯帮我拦住了至少三次因为底层表重构而导致的样例失效问题。