LCODER AI Agent问数实战（三）：搭建元数据知识库

时间：2026-09-19 11:34:01 编辑：袖梨来源：一聚教程网

问数智能体要根据自然语言生成可靠的 SQL，首先需要理解数据库中有哪些表和字段、业务术语对应哪些列，以及维度字段可能出现哪些取值。为此，项目需要把数据仓库信息加工成一套可配置、可校验并支持多种检索方式的元数据知识库。本节将从配置结构入手，逐步梳理元数据同步所需的模型与存储设计。

在上一章，我们已经连接好了服务，4个客户端管理类也都跑通了，这一章就来学习怎么往元数据库中写入数据，也就是怎么构建元数据知识库。目前来说，现在数据库中的状态是这样的：

数据库	当前状态	我们需要的状态
MySQL meta 库	4 张空表（有结构无数据）	填满表、字段、指标的元数据
Qdrant	空的（没有任何集合）	有字段和指标的向量索引
ES	空的（没有任何索引）	有字段取值的全文索引

这一章的目标，就是把数据仓库，也就是dw库中的表结构信息提取出来，加工成元数据，写入这三个数据库中。

假设现在AI Agent要帮你做事，你需要让他根据业务问题直接写出SQL查询语句，就需要提供给他三份资料：

资料	内容	放在哪里
资料 A	数据库有哪些表、哪些字段、字段是什么意思	MySQL meta 库
资料 B	字段和指标的"语义索引"（用户说"销售额"能找到"order_amount"）	Qdrant
资料 C	字段的具体取值（"华北""数码""会员"等）	ES

但是数据库中可能有几十张表，上百个字段，并不是所有的表和字段都需要让AI知道，有些表是内部用的，比如日志表，临时表。有些字段的取值不需要做全文检索。

所以需要一个配置文件来声明：哪些表需要同步、每张表的哪些字段需要同步、哪些字段的取值要放进ES、有哪些业务指标需要同步。

因此我们需要一个meta_config.yaml文件来声明这些内容。

一、配置文件

meta_config.yaml文件放在'data_agent/conf'目录中，根据业务需要分tables和metrics两大模块。

1.1 table模块：声明要同步哪些表和字段

tables:
  - name: dim_region  # 真实表名
    role: dim   #表角色 dim = 维度表 fact = 事实表
    description: 地区维度表，用于描述订单发生的地理区域信息。
    columns:
      - name: region_id
        role: primary_key
        description: 地区唯一标识。
        alias: [地区ID,区域ID]
        sync: false


      - name: province
        role: dimension
        description: 订单所属的省份名称。
        alias: [省份,省,所在省份]
        sync: true


      - name: region_name
        role: dimension
        description: 订单所属的大区名称，如华东、华南等。
        alias: [地区,区域,大区]
        sync: true

      - name: country
        role: dimension
        description: 地区所属国家名称。
        alias: [国家,国家名称]
        sync: true



  - name: dim_customer
    role: dim
    description: 客户维度表，描述下单客户的基本属性。
    columns:
      - name: customer_id
        role: primary_key
        description: 客户唯一标识。
        alias: [客户ID,用户ID]
        sync: false

      - name: customer_name
        role: dimension
        description: 客户名称。
        alias: [客户名称,用户名称]
        sync: true

      - name: gender
        role: dimension
        description: 客户性别。
        alias: [性别]
        sync: true

      - name: member_level
        role: dimension
        description: 客户会员等级。
        alias: [会员等级,用户等级]
        sync: true





  - name: dim_product
    role: dim
    description: 商品维度表，描述商品的基本属性信息。
    columns:
      - name: product_id
        role: primary_key
        description: 商品唯一标识。
        alias: [商品ID,产品ID]
        sync: false

      - name: product_name
        role: dimension
        description: 商品名称。
        alias: [商品名称,产品名称]
        sync: true

      - name: category
        role: dimension
        description: 商品所属品类。
        alias: [商品类别,品类,分类]
        sync: true

      - name: brand
        role: dimension
        description: 商品品牌名称。
        alias: [品牌,品牌名称]
        sync: true




  - name: dim_date
    role: dim
    description: 时间维度表，用于多时间粒度分析。
    columns:
      - name: date_id
        role: primary_key
        description: 日期唯一标识，格式 yyyyMMdd。
        alias: [日期ID,日期]
        sync: false

      - name: year
        role: dimension
        description: 年份。
        alias: [年,年份]
        sync: false

      - name: quarter
        role: dimension
        description: 季度。
        alias: [季度]
        sync: true

      - name: month
        role: dimension
        description: 月份。
        alias: [月,月份]
        sync: false

      - name: day
        role: dimension
        description: 日。
        alias: [日,天]
        sync: false

  - name: fact_order
    role: face
    description: 订单事实表，记录订单数量和金额等核心指标。
    columns:
      - name: order_id
        role: primary_key
        description: 订单唯一标识。
        alias: [订单ID]
        sync: false

      - name: customer_id
        role: foreign_key
        description: 关联客户维度的外键。
        alias: [客户ID,用户ID]
        sync: false

      - name: product_id
        role: foreign_key
        description: 关联商品维度的外键。
        alias: [ 商品ID, 产品ID ]
        sync: false

      - name: date_id
        role: foreign_key
        description: 关联时间维度的外键。
        alias: [ 日期, 下单日期 ]
        sync: false

      - name: region_id
        role: foreign_key
        description: 关联地区维度的外键。
        alias: [ 地区ID, 区域ID ]
        sync: false

      - name: order_quantity
        role: measure
        description: 订单中商品的购买数量。
        alias: [ 销量, 购买数量, 件数 ]
        sync: false

      - name: order_amount
        role: measure
        description: 订单金额。
        alias: [ 销售额, 订单金额, 收入 ]
        sync: false

字段角色（role）的含义：

值	含义	举例
`primary_key`	主键，唯一标识	region_id
`foreign_key`	外键，关联其他表	customer_id
`dimension`	维度字段，用来分组/筛选	province, brand
`measure`	度量字段，用来聚合计算	order_amount

sync 字段：这个字段的取值要不要同步到 ES。

sync: true：像"华北""数码"这种维度值，用户可能用自然语言提到，需要建全文索引
sync: false：像 ID（1,2,3）或金额数值，用户不会用自然语言提到，不用建索引

1.2 metrics 块——声明业务指标：

指标不是数据库里直接存的数据，而是用字段计算出来的业务概念。比如 GMV = SUM(order_amount)，用户会直接问"GMV 是多少"，但数据库里没有叫 GMV 的字段。

metrics:
  - name: GMV
    description: 全称Gross Merchandise Value，表示所有订单的成交金额总和。
    relevant_columns:
      - fact_order.order_amount
    alias: [ 成交总额, 订单总额 ]
  - name: AOV
    description: 全称Average Order Value，表示所有订单的成交金额平均值。
    relevant_columns:
      - fact_order.order_quantity
    alias: [ 平均单价, 平均订单金额 ]

本项目中配置了五张表 + 2个指标

表名	角色	说明
`dim_region`	dim	地区维度表
`dim_customer`	dim	客户维度表
`dim_product`	dim	商品维度表
`dim_date`	dim	时间维度表
`fact_order`	fact	订单事实表

指标名	全称	相关字段
GMV	Gross Merchandise Value（成交总额）	fact_order.order_amount
AOV	Average Order Value（客单价）	fact_order.order_quantity

1.3 meta_config.py 类型定义

光有YAML还不够。和app_config.py一样，需要用Python的@dataclass定义配置的结构，通过OmegaConf做类型校验，确保YAML格式正确。

from dataclasses import dataclass
from typing import Optional


@dataclass
class ColumnConfig:
    name: str
    role: str
    description: str
    alias: list[str]
    sync: bool


@dataclass
class TableConfig:
    name: str
    role: str
    description: str
    columns: list[ColumnConfig]


@dataclass
class MetricConfig:
    name: str
    description: str
    relevant_columns: list[str]
    alias: list[str]


@dataclass
class MetaConfig:
    tables: Optional[list[TableConfig]] = None
    metrics: Optional[list[MetricConfig]] = None

层级关系：

MetaConfig（顶层）
  ├── tables: list[TableConfig]      → 每张表的配置
  │     └── columns: list[ColumnConfig]  → 每个字段的配置
  └── metrics: list[MetricConfig]   → 每个指标的配置

用 Optional[...] = None 表示两个配置块都是可选的——你可以只同步表不同步指标，或反过来。

1.4 生成MetaConfig对象

创建MetaKnowledgeService类，这个类位于service层，用来管理整个创建元数据库的流程，创建元知识数据库的第一个流程就是合并meta_config.yaml文件和meta_config.py文件，生成MetaConfig对象。

# 1.加载配置文件
context = OmegaConf.load(config_path)
schema = OmegaConf.structured(MetaConfig)
meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

和前面生成APPConfig的过程一样，这几句代码执行完后，就生成了MetaConfig对象。MetaConfig对象内部的嵌套结构也全部转成了对应的dataclass实例：

meta_config 是 MetaConfig 对象
  ├── meta_config.tables 是 list[TableConfig] 对象
  │     └── 每个元素是 TableConfig 对象
  │           └── table.columns 是 list[ColumnConfig] 对象
  │                 └── 每个元素是 ColumnConfig 对象
  └── meta_config.metrics 是 list[MetricConfig] 对象
        └── 每个元素是 MetricConfig 对象

meta_config.tables[0].columns[0].name 拿到的就是 ColumnConfig 对象的 name 属性

二、创建ORM实体类

2.1 什么是ORM

ORM的全称是Object—Relational Mapping(对象关系映射）。它的核心思想是：用Python的类来表示数据库中的表，用类的属性来表示表中的字段。

不用ORM时，你要手写SQL：

# 不用 ORM：手写 SQL，容易出错
cursor.execute("SELECT id, name, role, description FROM table_info WHERE id = %s", ("dim_region",))
row = cursor.fetchone()
table_id = row[0]
table_name = row[1]

用ORM后，你操作的是Python对象：

# 用 ORM：操作对象，自动生成SQL
stmt = select(TableInfoMySQL).where(TableInfoMySQL.id == "dim_region")
result = await session.execute(stmt)
table = result.scalar_one()
print(table.name)   # 直接点号访问，不用记字段在第几列

ORM 帮你自动完成了 Python 对象 ↔ 数据库表 的翻译:

Python 类 TableInfoMySQL     数据库表 table_info
─────────────────────       ──────────────────
id: str               ←→    id VARCHAR(64)
name: str | None       ←→    name VARCHAR(128)
role: str | None       ←→    role VARCHAR(32)
description: str | None ←→   description TEXT

2.2 SQLAlchemy 2.0 的 ORM 语法

本项目使用的是 SQLAlchemy 2.0 的新语法，涉及三个核心概念：

2.2.1 DeclarativeBase — 所有实体类的基类

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

DeclarativeBase 是 SQLAlchemy 提供的基类，所有 ORM 实体类都要继承它。它告诉 SQLAlchemy："继承我的类都是数据库表的映射"。这个 Base 类是全局唯一的，项目里所有实体类都继承同一个 Base。

2.2.2 `tablename` — 指定表名

class TableInfoMySQL(Base):
    __tablename__ = "table_info"

__tablename__ 是一个类变量，告诉 SQLAlchemy 这个类对应数据库里的哪张表。

2.2.3 `Mapped` 和 `mapped_column` — 定义字段

id: Mapped[str] = mapped_column(
    String(64),        # 数据库中的类型
    primary_key=True,  # 是主键
    comment="表编号"    # 字段注释
)

这是 SQLAlchemy 2.0 的新语法，两个部分配合使用：

部分	作用
`id: Mapped[str]`	Python 层面的类型标注，告诉 IDE 这个属性是 `str` 类型，有代码补全
`mapped_column(...)`	数据库层面的定义，告诉 SQLAlchemy 这个字段在数据库里是什么类型、有什么约束

Mapped[str] 和 Mapped[str | None] 的区别：

id: Mapped[str]              # NOT NULL，数据库里这个字段不能为空
name: Mapped[str | None]     # 可以为 NULL，数据库里这个字段允许为空

Mapped[str | None] 中的 | None 就表示这个字段允许为空，等价于 SQL 中的 NULL。

2.2.4 `mapped_column` 的常用参数

参数	作用	例子
第一个位置参数	数据库列类型	`String(64)`、`Text`、`JSON`
`primary_key`	是否主键	`True` / `False`
`comment`	字段注释	`"表名称"`
`nullable`	是否允许为空	`True` / `False`（一般通过 `Mapped` 的类型标注控制）

2.2.5 常用的数据库列类型

类型	对应 SQL	用途
`String(64)`	`VARCHAR(64)`	定长字符串，括号里是最大长度
`Text`	`TEXT`	长文本，不限长度
`JSON`	`JSON`	JSON 格式数据，可以存列表和字典
`Integer`	`INT`	整数

2.3 创建ORM

2.3.1 base.py 基类

ORM类都位于app/models包中。

base.py是所有实体类的根基。后面 4 个实体类都继承这个 Base。

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

2.3.2 TableInfoMySql：表信息实体类

为了让大模型理解元数据库中的具体有哪些表，表的字段是什么，需要创建元数据实体类。这些实体类都存放在data-agent/app/models包中。

实体类	对应表	存什么	类比
`TableInfoMySQL`	`table_info`	表信息（表名、表类型、表描述）	书架上有哪些书
`ColumnInfoMySQL`	`column_info`	字段信息（字段名、类型、角色、示例值、别名、描述、所属表）	每本书有哪些章节
`MetricInfoMySQL`	`metric_info`	指标定义（指标名、描述、关联字段、别名）	书中定义的核心概念
`ColumnMetricMySQL`	`column_metric`	字段与指标的多对多关联关系	概念出现在哪些章节

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapper, mapped_column

from data_agent.app.models.Base import Base


class TableInfoMySQL(Base):
    __tablename__ = 'table_info'

    id: Mapper[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="表编号"
    )

    name : Mapper[str | None] = mapped_column(
        String(128),
        comment="表名称"
    )

    role: Mapper[str | None] = mapped_column(
        String(32),
        comment="表类型(fact/dim)"
    )

    description: Mapper[str | None] = mapped_column(
        Text,
        comment="表描述"
    )

作用：告诉大模型元数据库中有哪些表，每张表是干嘛的。后续创建的工作流中大模型会先从这里判断“这个问题查哪张表”。

 id: Mapper[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="表编号"
    )

这里的Mapper是SQLAlchemy的关键字，是定义ORM字段用的，在上面的2.2.3小节有介绍，作用是告诉SQLAlchemy这个属性对应数据库的一个字段，要注意的是，这里的Mapper和分层架构中的Mapper层没有任何关系。

和数据库表的对应关系：

Python 类 TableInfoMySQL          数据库表 table_info
────────────────────────        ─────────────────────
__tablename__ = "table_info"  →  表名：table_info

id: Mapped[str]               →  id VARCHAR(64) PRIMARY KEY
    String(64)                    -- 表编号
    primary_key=True
    comment="表编号"

name: Mapped[str | None]      →  name VARCHAR(128) NULL
    String(128)                   -- 表名称
    comment="表名称"

role: Mapped[str | None]       →  role VARCHAR(32) NULL
    String(32)                    -- 表类型(fact/dim)
    comment="表类型(fact/dim)"

description: Mapped[str | None] →  description TEXT NULL
    Text                          -- 表描述
    comment="表描述"

id 用 String(64) 而不是自增整数——因为 ID 直接用表名（如 "dim_region"），是业务 ID
description 用 Text 不用 String——表描述可能比较长，Text 不限长度
除 id 外都允许为 None——防止某些字段没填导致写入失败

2.3.3 column_info.py：字段信息实体

from sqlalchemy import String, Text
from sqlalchemy.types import JSON
from sqlalchemy.orm import Mapped, mapped_column

from data_agent.app.models.Base import Base


class ColumnInfoMySQL(Base):
    __tablename__ = "column_info"

    id: Mapped[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="列编号"
    )
    name: Mapped[str | None] = mapped_column(
        String(128),
        comment="列名称"
    )
    type: Mapped[str | None] = mapped_column(
        String(64),
        comment="数据类型"
    )
    role: Mapped[str | None] = mapped_column(
        String(32),
        comment="列类型(primary_key,foreign_key,measure,dimension)"
    )
    examples: Mapped[dict | list | None] = mapped_column(
        JSON,
        comment="数据示例"
    )
    description: Mapped[str | None] = mapped_column(
        Text,
        comment="列描述"
    )
    alias: Mapped[dict | list | None] = mapped_column(
        JSON,
        comment="列别名"
    )
    table_id: Mapped[str | None] = mapped_column(
        String(64),
        comment="所属表编号"
    )

作用：告诉大模型每个字段叫什么，是什么类型，有哪些别名，长什么样。这是SQL生成的核心知识，大模型要知道order_amount字段才能写出SUM(order_amount)。特别注意 examples（示例值）和 alias（别名）两个 JSON 字段：用户说"销售额"，LLM 需要通过别名知道对应字段是 order_amount。

examples: Mapped[dict | list | None] = mapped_column(JSON, comment="数据示例")
alias: Mapped[dict | list | None] = mapped_column(JSON, comment="列别名")

MySQL 8.0 原生支持 JSON 类型。这两个字段存的是结构化数据，不是简单字符串：

# examples 示例值（从dw库查出的10条数据）
["华北", "华东", "华南", "华中", "西南", "西北", "东北"]

# alias 别名
["销售额", "成交金额", "营收"]

用 JSON 类型的好处：存进去是列表，取出来还是列表，不用手动做字符串拆分。

和数据库表的对应关系：

Python 类 ColumnInfoMySQL          数据库表 column_info
────────────────────────         ──────────────────────────
id: Mapped[str]                 →  id VARCHAR(64) PRIMARY KEY
name: Mapped[str | None]        →  name VARCHAR(128)
type: Mapped[str | None]        →  type VARCHAR(64)
role: Mapped[str | None]        →  role VARCHAR(32)
examples: Mapped[...| None]     →  examples JSON
description: Mapped[str|None]   →  description TEXT
alias: Mapped[...| None]        →  alias JSON
table_id: Mapped[str | None]    →  table_id VARCHAR(64)

table_id 是外键关联到 table_info 表的 id，但这里没有显式声明 ForeignKey，因为本项目通过业务逻辑（代码层面）来维护关联，不用数据库的外键约束。

2.3.4 metric_info.py 指标信息实体

from sqlalchemy import String, Text
from sqlalchemy.types import JSON
from sqlalchemy.orm import Mapped, mapped_column

from data_agent.app.models.Base import Base


class MetricInfoMySQL(Base):
    __tablename__ = "metric_info"

    id: Mapped[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="指标编码"
    )
    name: Mapped[str | None] = mapped_column(
        String(128),
        comment="指标名称"
    )
    description: Mapped[str | None] = mapped_column(
        Text,
        comment="指标描述"
    )
    relevant_columns: Mapped[dict | list | None] = mapped_column(
        JSON,
        comment="关联字段"
    )
    alias: Mapped[dict | list | None] = mapped_column(
        JSON,
        comment="指标别名"
    )

作用：业务指标是比字段更高一层的抽象。用户问"GMV"或"客单价"，这些不是数据库里直接有的字段，而是需要通过计算（如 SUM(order_amount)）得到的指标。指标表告诉大模型每个指标对应哪些字段、怎么计算。和前两个结构类似，重点看 relevant_columns：

relevant_columns: Mapped[dict | list | None] = mapped_column(JSON, comment="关联字段")

存的是字段 ID 列表，如 ["fact_order.order_amount"]。用 JSON 存列表，取出后直接就是 Python list。

2.3.5 column_metric.py 字段-指标关联实体

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from data_agent.app.models.Base import Base


class ColumnMetricMySQL(Base):
    __tablename__ = "column_metric"

    column_id: Mapped[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="列编号"
    )
    metric_id: Mapped[str] = mapped_column(
        String(64),
        primary_key=True,
        comment="指标编号"
    )

联合主键：两个字段都标了 primary_key=True，组成联合主键。这意味着 (column_id, metric_id) 的组合必须唯一，但单独一个字段可以重复：

column_id              metric_id
──────────             ──────────
fact_order.order_amount    GMV       ← 合法
fact_order.order_amount    AOV       ← 合法（同字段不同指标）
fact_order.order_quantity  AOV       ← 合法（同指标不同字段）
fact_order.order_amount    GMV       ← 非法！和第一行完全重复

三、Entity层，业务实体

3.1 为什么需要Entity层？

前面讲过，项目中有两套数据：

models/ 下的 ORM 实体（如 TableInfoMySQL）——直接对应 MySQL 表结构，是 SQLAlchemy 用的
entities/ 下的 业务实体（如 TableInfo）——纯 Python 数据类，业务逻辑用的

为什么要分两套？因为职责分离：ORM实体绑定了数据库实现细节（字段类型、映射关系），业务实体则是纯粹的数据载体，不关心存在哪里、怎么存。将来换数据库不用改业务代码。

3.2 具体的业务实体

3.2.1 TableInfo 表信息实体

@dataclass
class TableInfo:
    id: str
    name: str
    role: str
    description: str

3.2.2 ColumnInfo 字段信息实体

@dataclass
class ColumnInfo:
    id: str
    name: str
    type: str
    role: str
    example: list[Any]
    description: str
    alias: list[str]
    table_id: str

3.2.3 MetricInfo 指标信息实体

@dataclass
class MetricInfo:
    id: str
    name: str
    description: str
    relevant_columns: list[str]
    alias: list[str]

3.2.4 ColumnMetric 字段指标关联实体

@dataclass
class ColumnMetric:
    column_id: str
    metric_id: str

3.2.5 ValueInfo 字段取值实体

@dataclass
class ValueInfo:
    id: str
    value: str
    column_id: str

3.3 业务实体和ORM实体的对比

以 ColumnInfo 为例：

对比维度	ORM 实体 (`ColumnInfoMySQL`)	业务实体 (`ColumnInfo`)
文件位置	`app/models/`	`app/entities/`
依赖	SQLAlchemy（`Mapped`、`mapped_column`）	无依赖（纯 dataclass）
字段类型	`Mapped[str]`、`mapped_column(JSON)`	`Optional[str]`、`Optional[Union[dict, list]]`
用途	数据库读写	业务逻辑传递、作为 Qdrant payload
能被序列化吗	不方便（有 SQLAlchemy 元数据）	方便（纯数据，可以直接转 JSON）

四、Mapper层 ORM实体和业务实体的转换器

4.1 Mapper做什么

Mapper是ORM实体和业务实体之间的翻译官：

写入数据库时：业务实体 → Mapper → ORM 实体 → SQLAlchemy 写入
从数据库读出时：SQLAlchemy 查出 ORM 实体 → Mapper → 业务实体 → 业务使用

4.1.1 TableInfoMapper

class TableInfoMapper:
    @staticmethod
    def to_entity(table_info_mysql: TableInfoMySQL) -> TableInfo:
        return TableInfo(
            id=table_info_mysql.id,
            name=table_info_mysql.name,
            role=table_info_mysql.role,
            description=table_info_mysql.description,
        )

    @staticmethod
    def to_model(table_info: TableInfo) -> TableInfoMySQL:
        return TableInfoMySQL(**asdict(table_info))

4.1.2 ColumnInfoMapper

class ColumnInfoMapper:
    @staticmethod
    def to_entity(column_info_mysql: ColumnInfoMySQL) -> ColumnInfo:
        return ColumnInfo(
            id=column_info_mysql.id,
            name=column_info_mysql.name,
            type=column_info_mysql.type,
            role=column_info_mysql.role,
            example=column_info_mysql.examples,
            description=column_info_mysql.description,
            alias=column_info_mysql.alias,
            table_id=column_info_mysql.table_id,
        )

    @staticmethod
    def to_model(column_info: ColumnInfo) -> ColumnInfoMySQL:
        return ColumnInfoMySQL(**asdict(column_info))

4.1.3 MetricInfoMapper

class MetricInfoMapper:
    @staticmethod
    def to_entity(metric_info_mysql:MetricInfoMySQL) -> MetricInfo:
        return MetricInfo(
            id=metric_info_mysql.id,
            name=metric_info_mysql.name,
            description=metric_info_mysql.description,
            relevant_columns=metric_info_mysql.relevant_columns,
            alias=metric_info_mysql.alias
        )

    @staticmethod
    def to_model(metric_info:MetricInfo) -> MetricInfoMySQL:
        return MetricInfoMySQL(**asdict(metric_info))

4.1.4 ColumnMetricMapper

class ColumnMetricMapper:
    @staticmethod
    def to_entity(column_metric_mysql: ColumnMetricMySQL) -> ColumnMetric:
        return ColumnMetric(
            column_id=column_metric_mysql.column_id,
            metric_id=column_metric_mysql.metric_id,
        )

    @staticmethod
    def to_model(column_metric: ColumnMetric) -> ColumnMetricMySQL:
        return ColumnMetricMySQL(**asdict(column_metric))

看到这里，我想大家肯定有疑问，这层不就是字段名一样转换一遍吗？有必要吗？当然是有必要的，我们以TableInfoMapper为例：

可以看到，上面TableInfoMySQL数据的两种形态，字段名一样，值也一样，为什么要使用第一种呢？这里的关键就在于：两种形态运行在不同的环境中，有不同的约束。ORM实体绑定着SQLAlchemy的内部状态。当它被session.add()后，SQLAlchemy会给它挂上一堆“追踪标记”（对象状态、变更记录、懒加载代理等）。这些标记只在session生命周期内有效，离开session就可能出问题。

# ORM 实体在 session 内可以正常工作
orm = TableInfoMySQL(id="dim_region", name="dim_region", ...)
session.add(orm)
await session.commit()   # 正常，SQLAlchemy 管着它

# 但拿出去用就有问题了
# 比如想存进 Qdrant 的 payload 里
await qdrant_client.upsert(
    collection_name="columns",
    points=[PointStruct(
        id="xxx",
        vector=[0.1, 0.2, ...],
        payload=orm   # ← 问题来了
    )]
)

把ORM实体直接塞进Qdrant的payload里，会出什么问题？

序列化失败：ORM 实体不是普通 dict，Qdrant 无法直接序列化它
SQLAlchemy 报错：离开 session 后访问某些属性，SQLAlchemy 会触发"detached instance"错误
冗余数据：ORM实体携带了大量 SQLAlchemy 内部元数据，全存进去浪费空间

业务实体就没有这些问题：

# 业务实体是纯 dataclass，到处都能用
entity = TableInfo(id="dim_region", name="dim_region", ...)

# 存进 Qdrant 的 payload，没问题
payload = entity.model_dump()   # 转成 dict，干净的数据

# 在 Service 里传递，没问题
service.do_something(entity)

# 序列化成 JSON，没问题
import json
json.dumps(entity.model_dump())

五、Respository层数据访问层

5.1 Repository 做什么

Repository 封装了对各个存储引擎的具体读写操作。上层（Service）只调 Repository 的方法，数据操作都交给Repository。

# ✅ Service 只管业务，数据操作交给 Repository
class MetaKnowledgeService:
    async def _save_tables_to_meta_db(self, meta_config):
        table_infos = [TableInfo(...), TableInfo(...)]
        # 告诉仓管员：把这些表存进去
        await self.meta_repository.save_table_infos(table_infos)
        # 不用管是 INSERT 还是 MERGE，不用管 SQL 怎么写

如果没有Repository，Service直接写SQL，会造成SQL语句散落在业务代码里面，又乱又难维护。

# ❌ Service 直接写 SQL，又乱又难维护
class MetaKnowledgeService:
    async def _save_tables_to_meta_db(self, meta_config):
        # Service 直接拼 SQL 字符串
        sql = "INSERT INTO table_info (id, name, role, description) VALUES "
        values = []
        for table in meta_config.tables:
            values.append(f"('{table.name}', '{table.name}', '{table.role}', '{table.description}')")
        sql += ", ".join(values)
        
        # 直接调 session.execute()
        await self.session.execute(text(sql))

5.2 Repository 层的整体布局

项目中有 5 个 Repository，对应 3 种存储引擎：

                    Repository 层
                         │
          ┌──────────────┼──────────────┐
          ↓              ↓              ↓
       MySQL          Qdrant          ES
     （关系库）      （向量库）     （全文检索）
          │              │              │
    ┌─────┴─────┐     2 个          1 个
    ↓           ↓
 MetaMySQL   DWMySQL
 Repository  Repository
 (meta库)    (dw库)

Repository	操作的存储	负责什么
`MetaMySQLRepository`	MySQL meta 库	读写表信息、字段信息、指标信息
`DWMySQLRepository`	MySQL dw 库	从数仓查字段类型、查字段取值
`ColumnQdrantRepository`	Qdrant	字段向量集合的建集合、写入、检索
`MetricQdrantRepository`	Qdrant	指标向量集合的建集合、写入、检索
`ValueESRepository`	ES	字段取值全文索引的建索引、写入、检索

每个 Repository 只管自己对应的那个存储引擎。

5.3 MetaMySQLRepository — meta 库读写

这个 Repository 负责读写 meta 库的 4 张表。

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.entities.column_info import ColumnInfo
from app.entities.column_metric import ColumnMetric
from app.entities.metric_info import MetricInfo
from app.entities.table_info import TableInfo
from app.models.column_info_mysql import ColumnInfoMySQL
from app.models.table_info_mysql import TableInfoMySQL
from app.repositories.mysql.meta.mappers.column_info_mapper import ColumnInfoMapper
from app.repositories.mysql.meta.mappers.column_metric_mapper import ColumnMetricMapper
from app.repositories.mysql.meta.mappers.metric_info_mapper import MetricInfoMapper
from app.repositories.mysql.meta.mappers.table_info_mapper import TableInfoMapper


class MetaMySQLRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def save_table_infos(self, table_infos: list[TableInfo]):
        models = [TableInfoMapper.to_model(table_info) for table_info in table_infos]
        self.session.add_all(models)

    async def save_column_infos(self, columns_info: list[ColumnInfo]):
        models = [ColumnInfoMapper.to_model(column_info) for column_info in columns_info]
        self.session.add_all(models)

    async def save_metric_infos(self, metric_infos: list[MetricInfo]):
        self.session.add_all([MetricInfoMapper.to_model(metric_info) for metric_info in metric_infos])

    async def save_column_metrics(self, column_metrics: list[ColumnMetric]):
        self.session.add_all([ColumnMetricMapper.to_model(column_metric) for column_metric in column_metrics])

    async def get_column_info_by_id(self, column_id: str) -> ColumnInfo | None:
        result: ColumnInfoMySQL | None = await self.session.get(ColumnInfoMySQL, column_id)
        if result:
            return ColumnInfoMapper.to_entity(result)
        return None

    async def get_table_info_by_id(self, table_id: str) -> TableInfo | None:
        result: TableInfoMySQL | None = await self.session.get(TableInfoMySQL, table_id)
        if result:
            return TableInfoMapper.to_entity(result)
        return None

    async def get_key_columns_by_table_id(self, table_id: str) -> list[ColumnInfo]:
        sql = """
            select * 
            from column_info 
            where table_id = :table_id 
            and role in ('primary_key', 'foreign_key')
        """
        result = await self.session.execute(text(sql), {"table_id": table_id})
        return [ColumnInfo(**row) for row in result.mappings().fetchall()]

构造函数

class MetaMySQLRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

接收一个 AsyncSession（数据库会话），存在 self.session 里。所有方法都用这个 session 来操作数据库。

为什么 session 从外面传进来？因为事务要在上层管理。比如"存表信息"和"存字段信息"要在同一个事务里（要么都成功，要么都失败），如果 Repository 自己创建 session，就没法和别人共用事务了。

方法 1：save_table_infos — 批量保存表信息

async def save_table_infos(self, table_infos: list[TableInfo]):
    """
    入参：table_infos — 业务实体列表（TableInfo对象）
    出参：无（成功就不报错，失败就抛异常）
    做什么：把表信息批量写入 table_info 表
    """
    # 1. 业务实体 → Mapper 转成 ORM 实体
    orm_list = [TableInfoMapper.to_orm(t) for t in table_infos]
    # 2. 把所有ORM实体加入session（准备写入）
    self.session.add_all(orm_list)

调用时机：构建知识库时，_save_tables_to_meta_db() 方法中调用。

入参示例：

[
    TableInfo(id="dim_region", name="dim_region", role="dim", description="地区维度表"),
    TableInfo(id="dim_customer", name="dim_customer", role="dim", description="客户维度表"),
    TableInfo(id="fact_order", name="fact_order", role="fact", description="订单事实表"),
    ...
]

执行过程：

遍历列表，每个 TableInfo 业务实体通过 TableInfoMapper.to_orm() 转成 TableInfoMySQL ORM 实体
session.add_all(orm_list) 把所有 ORM 实体加入 session（相当于排队等待写入）

执行完的结果：数据还没真正写到数据库里，只是在 session 的"待写入队列"中。等上层调用 session.commit() 时才会真正执行 INSERT 语句写入数据库。

为什么不在 Repository 里 commit？因为一次操作可能涉及多个表（如表和字段要同时写入），事务控制应该在上层（Service）统一管理，保证要么都成功要么都失败。

方法 2：save_column_infos — 批量保存字段信息

async def save_column_infos(self, column_infos: list[ColumnInfo]):
    """
    入参：column_infos — 字段业务实体列表
    出参：无
    做什么：把字段信息批量写入 column_info 表
    """
    orm_list = [ColumnInfoMapper.to_orm(c) for c in column_infos]
    self.session.add_all(orm_list)

和 save_table_infos 逻辑完全一样，只是操作的是 column_info 表，用的是 ColumnInfoMapper。

入参示例：

[
    ColumnInfo(id="dim_region.region_id", name="region_id", type="int", role="primary_key", ...),
    ColumnInfo(id="dim_region.region_name", name="region_name", type="varchar", role="dimension", ...),
    ...
]

方法 3：save_metric_infos — 批量保存指标信息

async def save_metric_infos(self, metric_infos: list[MetricInfo]):
    orm_list = [MetricInfoMapper.to_orm(m) for m in metric_infos]
    self.session.add_all(orm_list)

写入 metric_info 表。三个保存方法的结构完全一致，都是：业务实体 → Mapper 转 ORM → add_all 加入 session。

方法 4：save_column_metrics — 批量保存字段-指标关联

async def save_column_metrics(self, column_metrics: list[ColumnMetric]):
    orm_list = [ColumnMetricMapper.to_orm(cm) for cm in column_metrics]
    self.session.add_all(orm_list)

写入 column_metric 关联表。

方法 5：get_table_info_by_id — 根据 ID 查询表信息

async def get_table_info_by_id(self, table_id: str) -> TableInfo:
    """
    入参：table_id — 表编号，如 "dim_region"
    出参：TableInfo 业务实体（查不到返回 None）
    做什么：根据主键查询一条表信息
    """
    # 1. 构造 SQLAlchemy 查询语句
    stmt = select(TableInfoMySQL).where(TableInfoMySQL.id == table_id)
    # 2. 执行查询
    result = await self.session.execute(stmt)
    # 3. 取出第一条结果（ORM实体）
    orm = result.scalar_one_or_none()
    # 4. ORM实体 → Mapper 转成业务实体
    return TableInfoMapper.to_entity(orm) if orm else None

调用时机：查询阶段，根据表 ID 反查表的完整信息。

执行过程：

select(TableInfoMySQL).where(...) 构造查询（类似写 SQL：SELECT * FROM table_info WHERE id = 'dim_region'）
session.execute(stmt) 执行查询，拿到结果集
scalar_one_or_none() 取出第一条 ORM 实体，没有就返回 None
TableInfoMapper.to_entity(orm) 把 ORM 实体转成业务实体返回

返回结果：TableInfo(id="dim_region", name="dim_region", role="dim", description="地区维度表")

方法 6：get_column_infos_by_table — 根据表 ID 查所有字段

async def get_column_infos_by_table(self, table_id: str) -> list[ColumnInfo]:
    """
    入参：table_id — 表编号
    出参：ColumnInfo 列表（这张表的所有字段）
    做什么：查询某张表的所有字段信息
    """
    stmt = select(ColumnInfoMySQL).where(ColumnInfoMySQL.table_id == table_id)
    result = await self.session.execute(stmt)
    orms = result.scalars().all()
    return [ColumnInfoMapper.to_entity(orm) for orm in orms]

和上一个方法类似，区别是查的是 column_info 表，按 table_id 筛选，返回的是列表（一张表有多个字段）。

5.4 DWMySQLRepository — dw 库读取

文件位置：app/repositories/mysql/dw/dw_mysql_repository.py

这个 Repository 负责从 dw 数据仓库里读数据（只读不写），因为 dw 库的业务数据是现成的，构建知识库时只需要从中查一些信息。

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession


class DWMySQLRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_column_types(self, table_name: str) -> dict[str, str]:
        sql = f"show columns from {table_name}"
        result = await self.session.execute(text(sql))
        return {row.Field: row.Type for row in result.fetchall()}

    async def get_column_values(self, table_name: str, column_name: str, limit: int):
        sql = f"select distinct {column_name} from {table_name} limit {limit}"
        result = await self.session.execute(text(sql))
        return result.scalars().fetchall()

    async def get_db_info(self):
        result = await self.session.execute(text("select version()"))
        version = result.scalar()

        dialect = self.session.get_bind().dialect.name

        return {'version': version, 'dialect': dialect}

    async def validate_sql(self, sql):
        await self.session.execute(text(f"explain {sql}"))

    async def execute_sql(self, sql):
        result = await self.session.execute(text(sql))
        return [dict(row) for row in result.mappings().fetchall()]

方法 1：get_column_types — 查询某张表所有字段的数据类型

async def get_column_types(self, table_name: str) -> dict[str, str]:
    """
    入参：table_name — 表名，如 "dim_region"
    出参：字典 {字段名: 数据类型}，如 {"region_id": "int", "region_name": "varchar"}
    做什么：从系统表查字段类型
    """
    sql = text("""
        SELECT COLUMN_NAME, DATA_TYPE 
        FROM INFORMATION_SCHEMA.COLUMNS 
        WHERE TABLE_SCHEMA = 'dw' AND TABLE_NAME = :table_name
    """)
    result = await self.session.execute(sql, {"table_name": table_name})
    return {row[0]: row[1] for row in result.fetchall()}

调用时机：构建知识库时，Service 需要知道每个字段的数据类型（INT、VARCHAR、DECIMAL 等），这些信息配置文件里没写，直接从数据库查最准确。

查询的是 INFORMATION_SCHEMA.COLUMNS——这是 MySQL 自带的系统表，记录了所有库、所有表、所有字段的元信息。每个 MySQL 都有，不用自己建。

返回结果示例：

{
    "region_id": "int",
    "region_name": "varchar",
    "province": "varchar",
    "country": "varchar",
}

方法 2：get_column_values — 查询某个字段的不重复取值

async def get_column_values(self, table_name: str, column_name: str, limit: int = 10) -> list:
    """
    入参：
      - table_name: 表名
      - column_name: 字段名
      - limit: 最多返回多少条，默认10条
    出参：取值列表
    做什么：用 SELECT DISTINCT 查字段的不重复值
    """
    sql = text(f"SELECT DISTINCT `{column_name}` FROM `{table_name}` LIMIT {limit}")
    result = await self.session.execute(sql)
    return [row[0] for row in result.fetchall()]

调用时机：有两个地方用到：

构建字段示例值时：每个字段查 10 条，填入 ColumnInfo.examples，作为元数据的一部分
构建 ES 全文索引时：对 sync=true 的字段，查 10 万条（所有取值），写入 ES

返回结果示例（查 dim_region.region_name）：

["华北", "华东", "华南", "华中", "西南", "西北", "东北"]

DISTINCT 的意思是"去重"——如果表里有 1000 条记录，region_name 可能有很多重复的，加了 DISTINCT 后只返回不重复的值。

方法3：get_db_info() — 获取数据库信息

async def get_db_info(self):
    result = await self.session.execute(text("select version()"))
    version = result.scalar()
    
    dialect = self.session.get_bind().dialect.name
    
    return {'version': version, 'dialect': dialect}

它做了什么？

查询当前连接的数据库的版本号和方言类型，返回一个字典。

返回结果示例：

{'version': '8.0.35', 'dialect': 'mysql'}

为什么需要这个方法？

在add_extra_context 节点中，我们会把数据库信息喂给 LLM。原因是：

不同数据库的 SQL 语法有差异（MySQL 和 PostgreSQL 的函数名、字符串拼接方式等都不一样）
不同版本支持的语法也不一样（比如 MySQL 8.0 支持窗口函数，5.x 不支持）

LLM 拿到dialect: "mysql" 和version: "8.0.35" 后，就能生成对应语法的 SQL，而不是瞎猜用什么方言。

方法4 validate_sql(sql) 校验 SQL

async def validate_sql(self, sql):
    await self.session.execute(text(f"explain {sql}"))

用 MySQL 的EXPLAIN 命令检查一条 SQL 是否语法正确、表和字段是否存在。

如果 SQL 没问题，方法正常返回（不返回任何值）；如果 SQL 有问题（比如表名写错了、字段不存在），MySQL 会抛异常，这个异常会向上抛出。

方法5 execute_sql(sql) 执行 SQL

async def execute_sql(self, sql):
    result = await self.session.execute(text(sql))
    return [dict(row) for row in result.mappings().fetchall()]

执行一条 SQL 查询，把结果以字典列表的形式返回。

5.5 ColumnQdrantRepository 字段向量集合读写

文件位置：app/repositories/qdrant/column_qdrant_repository.py

负责操作 Qdrant 中名叫 "columns" 的向量集合。

from dataclasses import asdict

from qdrant_client import AsyncQdrantClient
from qdrant_client.http.models import VectorParams, Distance, PointStruct

from data_agent.app.conf.app_config import app_config
from data_agent.app.entities.column_info import ColumnInfo


class ColumnQdrantRepository:
    collection_name: str = 'data-agent-column'

    def __init__(self, client: AsyncQdrantClient):
        self.client = client

    async def ensure_collection(self):
        if not await self.client.collection_exists(self.collection_name):
            await self.client.create_collection(self.collection_name,
                                                vectors_config=VectorParams(
                                                    size=app_config.qdrant.embedding_size, distance=Distance.COSINE))

    async def upsert(self, ids: list[str], embeddings: list[list[float]], payloads: list[ColumnInfo],
                     batch_size: int = 20):
        zipped = list(zip(ids, embeddings, payloads))
        for i in range(0, len(zipped), batch_size):
            batch = zipped[i:i + batch_size]
            batch_points = [PointStruct(id=id, vector=embeddings, payload=asdict(payload)) for id, payload in batch]
            await self.client.upsert(collection_name=self.collection_name, points=batch_points)

    async def search(self, embeddings: list[float], score_threshold: float = 0.6, limit: int = 5) -> list[ColumnInfo]:
        result = await self.client.query_points(collection_name=self.collection_name,
                                                query=embeddings,
                                                score_threshold=score_threshold,
                                                limit=limit)
        return [ColumnInfo(**point.payload) for point in result.points]

构造函数

class ColumnQdrantRepository:
    def __init__(self, client: AsyncQdrantClient):
        self.client = client

接收 Qdrant 异步客户端，存在 self.client 里。

方法 1：ensure_collection — 确保集合存在

async def ensure_collection(self):
    """
    入参：无
    出参：无
    做什么：检查 columns 集合是否存在，不存在就创建
    """
    if not await self.client.collection_exists("columns"):
        await self.client.create_collection(
            collection_name="columns",
            vectors_config=models.VectorParams(
                size=1024,                  # 向量维度（bge-large-zh输出1024维）
                distance=models.Distance.COSINE,  # 余弦相似度
            ),
        )

调用时机：构建知识库之前，先调用这个方法保证集合存在。

执行过程：

先查 "columns" 集合是否存在
如果不存在，就创建一个：
- 向量维度：1024（和 Embedding 模型输出一致）
- 距离度量：余弦相似度（语义检索用余弦效果最好）

执行完的结果：Qdrant 里有了一个空的 columns 集合，可以往里面写向量了。

为什么叫 ensure？ 因为它是"确保存在"——有就不管，没有就建。不管调多少次，结果都一样：集合一定存在。这种写法可以避免重复创建报错。

方法 2：upsert — 批量写入向量

async def upsert(self, ids, vectors, payloads):
    """
    入参：
      - ids: ID列表（每条向量的唯一标识）
      - vectors: 向量列表（每条向量是1024个浮点数）
      - payloads: 附加信息列表（完整的ColumnInfo业务实体）
    出参：无
    做什么：把向量数据批量写入Qdrant
    """
    points = [
        models.PointStruct(
            id=str(id_),
            vector=vector,
            payload=payload.model_dump() if hasattr(payload, 'model_dump') else payload
        )
        for id_, vector, payload in zip(ids, vectors, payloads)
    ]
    await self.client.upsert(collection_name="columns", points=points)

调用时机：构建知识库时，字段描述转成向量后，调用这个方法写入 Qdrant。

三个入参的关系：三个列表是一一对应的，第 i 个 id 配第 i 个 vector 配第 i 个 payload。

ids:        [id1,          id2,          id3,         ...]
vectors:    [向量1(1024个), 向量2(1024个), 向量3(1024个), ...]
payloads:   [字段信息1,     字段信息2,     字段信息3,    ...]
              ↓              ↓              ↓
            组合成 PointStruct 写入 Qdrant

执行过程：

用 zip 把三个列表对应位置的元素拼在一起
每个三元组组装成一个 PointStruct（Qdrant 的数据点结构）
payload.model_dump() 把业务实体转成字典（dataclass 才能序列化）
调用 client.upsert() 批量写入

执行完的结果：Qdrant 的 columns 集合里多了一批向量点，每个点都有 ID、向量、附加信息。

upsert 是什么意思？ update + insert 的合成词——如果 ID 已存在就更新，不存在就插入。所以重复调用不会重复插入，而是覆盖更新。

方法 3：search — 相似度检索

async def search(self, query_vector, limit=5, score_threshold=0.6):
    """
    入参：
      - query_vector: 查询向量（用户问题转向量后的结果）
      - limit: 最多返回多少条，默认5条
      - score_threshold: 相似度阈值，低于0.6的不要
    出参：检索结果（包含匹配到的点的id、score、payload）
    做什么：用语义相似度找最相关的字段
    """
    results = await self.client.query_points(
        collection_name="columns",
        query=query_vector,
        limit=limit,
        score_threshold=score_threshold,
    )
    return results

调用时机：查询阶段（第 6 章），用户的关键词转成向量后，调用这个方法找最相关的字段。

返回结果：按相似度从高到低排列的字段列表，每条包含：

id：向量点的 ID
score：相似度分数（0 到 1，越接近 1 越相似）
payload：附加的完整字段信息（ColumnInfo）

比如用户问"各地区的销售额"，查询向量后可能返回：

第1名：fact_order.order_amount（score=0.92，最相关）
第2名：fact_order.order_quantity（score=0.78）
第3名：dim_region.region_name（score=0.71）
...

LLM 拿到这些字段信息后，就能判断要查哪张表、用哪些字段。

5.6 MetricQdrantRepository 指标向量集合读写

from dataclasses import asdict

from qdrant_client import AsyncQdrantClient
from qdrant_client.http.models import VectorParams, Distance, PointStruct

from data_agent.app.conf.app_config import app_config
from data_agent.app.entities.metric_info import MetricInfo


class MetricQdrantRepository:
    collection_name = 'data-agent-metric'

    def __init__(self, client: AsyncQdrantClient):
        self.client = client

    async def ensure_collection(self):
        if not await self.client.collection_exists(self.collection_name):
            await self.client.create_collection(self.collection_name,
                                                vectors_config=VectorParams(size=app_config.qdrant.embedding_size,
                                                                            distance=Distance.COSINE))

    async def upsert(self, ids: list[str], embeddings: list[list[float]], payloads: list[MetricInfo],
                     batch_size: int = 20):
        zipped = list(zip(ids, embeddings, payloads))
        for i in range(0, len(zipped), batch_size):
            batch = zipped[i: i + batch_size]
            batch_points = [PointStruct(id=id, vector=embedding, payload=asdict(payload)) for id, embedding, payload in
                            batch]
            await self.client.upsert(collection_name=self.collection_name, points=batch_points)

    async def search(self, embeddings: list[float], score_threshold: float = 0.6, limit: int = 5) -> list[MetricInfo]:
        result = await self.client.query_points(collection_name=self.collection_name,
                                                query=embeddings,
                                                score_threshold=score_threshold,
                                                limit=limit)
        return [MetricInfo(**point.payload) for point in result.points]

结构和 ColumnQdrantRepository 完全一样，只是集合名是 "metrics"，操作的是指标数据。三个方法：

方法	作用
`ensure_collection()`	确保 metrics 集合存在
`upsert(ids, vectors, payloads)`	批量写入指标向量
`search(query_vector, limit, score_threshold)`	语义检索相关指标

5.7 ValueESRepository — 字段取值全文索引读写

文件位置：app/repositories/es/value_es_repository.py

负责操作 ES 中名叫 "data_agent" 的全文索引。

from dataclasses import asdict

from elasticsearch import AsyncElasticsearch

from data_agent.app.entities.value_info import ValueInfo


class ValueEsRepository:
    index_name = 'data-agent-value'
    index_mappings = {
        "dynamic": False,
        "properties": {
            "id": {"type": "keyword"},
            "value": {"type": "text", "analyzer": "ik_max_word", "search_analyzer": "ik_max_word"},
            "column_id": {"type": "keyword"},
        }
    }

    def __init__(self, client: AsyncElasticsearch):
        self.client = client

    async def ensure_index(self):
        if not await self.client.indices.exists(self.index_name):
            await self.client.indices.create(index=self.index_name, mappings=self.index_mappings)

    async def index(self, value_infos: list[ValueInfo], batch_size=20):
        for i in range(0, len(value_infos), batch_size):
            batch = value_infos[i:i + batch_size]
            operations = []
            for value_info in batch:
                operations.append({"index": {"_index": self.index_name,
                                             "_id": value_info.id}})
                operations.append(asdict(value_info))
            await self.client.bulk(operations=operations)

    async def search(self, keyword: str, score_threshold: float = 0.6, limit: int = 5) -> list[ValueInfo]:
        result = await self.client.search(index=self.index_name,
                                          query={
                                              "match": {
                                                  "value": keyword
                                              }
                                          },
                                          min_score=score_threshold,
                                          size=limit)
        return [ValueInfo(**hit['_source']) for hit in result['hits']['hits']]

方法 1：ensure_index — 确保索引存在

async def ensure_index(self):
    """
    做什么：检查 data_agent 索引是否存在，不存在就创建
    """
    index_name = "data_agent"
    if not await self.client.indices.exists(index=index_name):
        await self.client.indices.create(
            index=index_name,
            mappings={
                "dynamic": False,          # 不允许动态加字段
                "properties": {
                    "value": {"type": "text", "analyzer": "ik_max_word"},  # 中文分词
                    "column_id": {"type": "keyword"},                    # 精确匹配
                },
            },
        )

两个字段的区别：

字段	类型	作用	举例
`value`	`text` + `ik_max_word`	全文检索字段，会被中文分词器切分	"华北地区" → 切成 "华北" "地区"
`column_id`	`keyword`	精确匹配字段，不分词	"dim_region.region_name" 整个当一个整体

dynamic: False 的意思是：只能存 value 和 column_id 两个字段，存其他字段会被忽略，保证结构可控。

方法 2：index — 批量写入

async def index(self, value_infos):
    """
    入参：value_infos — ValueInfo 业务实体列表
    出参：无
    做什么：把字段取值批量写入ES
    """
    operations = []
    for vi in value_infos:
        # 操作头：告诉ES这是一条写入操作
        operations.append({"index": {"_index": "data_agent"}})
        # 数据体：实际内容
        operations.append({"value": vi.value, "column_id": vi.column_id})
    await self.client.bulk(operations=operations)

ES bulk API 的格式很特殊：操作头 + 数据体，交替排列。

operations = [
    {"index": {"_index": "data_agent"}},   ← 第1条：操作头（写入）
    {"value": "华北", "column_id": "dim_region.region_name"},  ← 第1条：数据
    {"index": {"_index": "data_agent"}},   ← 第2条：操作头
    {"value": "华东", "column_id": "dim_region.region_name"},  ← 第2条：数据
    {"index": {"_index": "data_agent"}},   ← 第3条：操作头
    {"value": "数码", "column_id": "dim_product.category"},   ← 第3条：数据
    ...
]

为什么要这么设计？因为一次批量操作中可以混合不同类型（写入、更新、删除），所以每条数据前都要有一个"操作头"说明接下来这条是什么操作。

执行完的结果：ES 的 data_agent 索引里多了一批文档，每条文档包含取值内容和所属字段 ID。

方法 3：search — 全文检索

async def search(self, keyword):
    """
    入参：keyword — 搜索关键词（从用户问题中提取的）
    出参：检索结果（匹配到的文档列表）
    做什么：在value字段中做全文匹配
    """
    result = await self.client.search(
        index="data_agent",
        query={"match": {"value": keyword}},
    )
    return result

调用时机：查询阶段，用户问题中提取出关键词后，调用这个方法找相关的维度值。

比如关键词是"华北"，ES 会在所有 value 字段中搜索，返回匹配的文档：

第1名：value="华北地区"，column_id="dim_region.region_name"（score最高）
第2名：value="华北"，column_id="dim_region.region_name"
第3名：value="华北大区"，column_id="dim_region.region_name"

LLM 拿到这些结果后，就知道用户提到的"华北"对应 dim_region.region_name 字段，生成 SQL 时可以写出 WHERE region_name LIKE '%华北%'。

到这里，我们已经学了两层结构：

层级	职责
Client Manager	管理连接（连接MySql、Qdrant等）
Repository	执行单条读写操作（存一条数据，查一个字段）

Repository 层的每个方法只做一件事 —— 比如"存表信息"或"查字段类型"。但构建知识库是一个多步骤流程：

从 YAML 配置读出有哪些表
从数据仓库查出每张表的字段类型
从数据仓库查出每个字段的示例值
把表信息和字段信息写入元数据库
把字段信息做成向量存入 Qdrant
把字段取值存入 Elasticsearch
把指标信息写入元数据库
把指标信息做成向量存入 Qdrant

这 8 个步骤，每一步都要调用不同的 Repository 方法。

如果把这些编排逻辑散落在入口脚本里，脚本会变得极其臃肿、无法复用。 Service 层就是把这些编排逻辑集中到一个类里，对外只暴露一个build() 方法，调用者只需一句话就能完成整个知识库构建。

接下来我们就来构建Service层，也是构建元知识数据库的最后一步。

六、 Service层

6.1 入口脚本

在讲 Service 类之前，先看它的调用方 —— 入口脚本。这个脚本做三件事：解析参数 → 初始化所有客户端 → 组装 Service 并执行。

import asyncio
from argparse import ArgumentParser
from pathlib import Path

from data_agent.app.clients.embedding_client_manager import embedding_client_manager
from data_agent.app.clients.es_client_manager import es_client_manager
from data_agent.app.clients.mysql_client_manager import meta_mysql_client, dw_mysql_client
from data_agent.app.clients.qdrant_client_manager import qdrant_client_manager
from data_agent.app.repositories.dw.dw_mysql_repository import DWMySQLRepository
from data_agent.app.repositories.es.value_es_repository import ValueEsRepository
from data_agent.app.repositories.meta.meta_mysql_repository import MetaMySQLRepository
from data_agent.app.repositories.qdrant.column_qdrant_repository import ColumnQdrantRepository
from data_agent.app.repositories.qdrant.metric_qdrant_repository import MetricQdrantRepository
from data_agent.app.service.meta_knowledge_service import MetaKnowledgeService


async def bulid(config_path: Path):
    # 1. 初始化所有客户端管理器（建立连接池）
    meta_mysql_client.init()
    dw_mysql_client.init()
    qdrant_client_manager.init()
    embedding_client_manager.init()
    es_client_manager.init()

    # 2. 创建数据库会话（上下文管理器，退出时自动提交/回滚）
    async with(
        meta_mysql_client.session_factory() as meta_session,
        dw_mysql_client.session_factory() as dw_session,
    ):
        # 3. 用会话创建 Repository 实例
        meta_mysql_repository = MetaMySQLRepository(meta_session)
        dw_mysql_repository = DWMySQLRepository(dw_session)

        column_qdrant_repository = ColumnQdrantRepository(qdrant_client_manager.client)
        embedding_client = embedding_client_manager.client  # 获取Embedding客户端实例
        value_es_repository = ValueEsRepository(es_client_manager.client)
        metric_qdrant_repository = MetricQdrantRepository(qdrant_client_manager.client)

        # 4. 把所有 Repository 注入到 Service 中
        mete_knowledge_service = MetaKnowledgeService(
            meta_mysql_repository=meta_mysql_repository,
            dw_mysql_repository=dw_mysql_repository,
            column_qdrant_repository=column_qdrant_repository,
            embedding_client=embedding_client,
            value_es_repository=value_es_repository,
            metric_qdrant_repository=metric_qdrant_repository,
        )
        # 5. 调用 build 方法，执行整个知识库构建流程
        await mete_knowledge_service.build(config_path)  # 构建元知识库

    await meta_mysql_client.close()  # 关闭元数据MySQL客户端
    await dw_mysql_client.close()  # 关闭数据仓库MySQL客户端
    await qdrant_client_manager.close()  # 关闭Qdrant客户端
    await es_client_manager.close()  # 关闭Elasticsearch客户端


if __name__ == "__main__":
    parser = ArgumentParser()

    parser.add_argument("-c", "--conf")  # option that takes a value

    args = parser.parse_args()

    config_path = Path(args.conf)

    asyncio.run(bulid(config_path))

这个脚本的核心逻辑用一句话总结：准备好所有零件（客户端、会话、Repository），然后交给 Service 这个总指挥去组装。

运行方式：

python -m app.scripts.build_meta_knowledge -c .confmeta_config.yaml

6.2 核心类：MetaKnowledgeService

6.2.1 构造函数 — 依赖注入

class MetaKnowledgeService:
    def __init__(
        self,
        meta_mysql_repository: MetaMySQLRepository,
        dw_mysql_repository: DWMySQLRepository,
        column_qdrant_repository: ColumnQdrantRepository,
        embedding_client: HuggingFaceEndpointEmbeddings,
        value_es_repository: ValueESRepository,
        metric_qdrant_repository: MetricQdrantRepository,
    ):
        self.meta_mysql_repository = meta_mysql_repository
        self.dw_mysql_repository = dw_mysql_repository
        self.column_qdrant_repository = column_qdrant_repository
        self.embedding_client = embedding_client
        self.value_es_repository = value_es_repository
        self.metric_qdrant_repository = metric_qdrant_repository

这个类需要 6 个"零件"：

注入的组件	来源	作用
`meta_mysql_repository`	元数据库会话	往 meta 库写表信息、字段信息、指标信息
`dw_mysql_repository`	数据仓库会话	从 dw 库查字段类型、查字段取值
`column_qdrant_repository`	Qdrant 客户端	往 Qdrant 存字段向量
`embedding_client`	Embedding 服务客户端	把文本转成向量
`value_es_repository`	ES 客户端	往 ES 存字段取值（全文检索）
`metric_qdrant_repository`	Qdrant 客户端	往 Qdrant 存指标向量

为什么不自己在 Service 里 new 这些对象？ 因为"依赖注入"让 Service 只负责用这些零件，不负责造零件。这样换零件（比如换一个不同的 Embedding 模型）时，不需要改 Service 的代码。

6.2.2 `build()` 方法 — 总指挥的调度逻辑

async def build(self, config_path: Path):
    # 1. 加载配置文件
    context = OmegaConf.load(config_path)
    schema = OmegaConf.structured(MetaConfig)
    meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))
    logger.info("加载配置文件")

    # 2. 处理表信息
    if meta_config.tables:
        # 2.1 保存表信息到meta数据库（同时返回所有字段信息）
        column_infos = await self._save_tables_to_meta_db(meta_config)
        logger.info("保存表信息到meta数据库")

        # 2.2 为字段信息建立向量索引
        await self._save_column_info_to_qdrant(column_infos)
        logger.info("为字段信息建立向量索引")

        # 2.3 为字段取值建立全文索引
        await self._save_value_info_to_es(meta_config, column_infos)
        logger.info("为字段取值建立全文索引")

    # 3. 处理指标信息
    if meta_config.metrics:
        # 3.1 保存指标信息到meta数据库
        metric_infos = await self._save_metrics_to_meta_db(meta_config)
        logger.info("保存指标信息到meta数据库")

        # 3.2 为指标信息建立向量索引
        await self._save_metric_info_to_qdrant(metric_infos)
        logger.info("为指标信息建立向量索引")

    logger.info("元数据知识库构建完成")

build() 做了 5 件事，按顺序执行：

步骤	调用的方法	做什么	数据流向
1	OmegaConf	加载 meta_config.yaml	YAML → MetaConfig 对象
2.1	`_save_tables_to_meta_db()`	保存表和字段信息	YAML配置 + dw库 → meta库
2.2	`_save_column_info_to_qdrant()`	字段信息向量化	字段文本 → Embedding → Qdrant
2.3	`_save_value_info_to_es()`	字段取值存入ES	dw库取值 → ES全文索引
3.1	`_save_metrics_to_meta_db()`	保存指标信息	YAML配置 → meta库
3.2	`_save_metric_info_to_qdrant()`	指标信息向量化	指标文本 → Embedding → Qdrant

注意 _save_tables_to_meta_db 返回 column_infos，这个返回值会被后续两步使用 —— 这就是 Service 层编排的价值：协调各步骤之间的数据传递。

6.2.3 `_save_tables_to_meta_db()` — 从配置+数据仓库 → 元数据库

async def _save_tables_to_meta_db(self, meta_config: MetaConfig) -> list[ColumnInfo]:
    table_infos: list[TableInfo] = []
    column_infos: list[ColumnInfo] = []

    for table in meta_config.tables:
        # 1. 从配置构造 TableInfo 业务实体
        table_info = TableInfo(
            id=table.name,
            name=table.name,
            role=table.role,
            description=table.description,
        )
        table_infos.append(table_info)

        # 2. 从数据仓库查该表所有字段的类型
        column_types: dict[str, str] = await self.dw_mysql_repository.get_column_types(table.name)

        for column in table.columns:
            # 3. 从数据仓库查该字段的 10 个示例值
            column_values: list = await self.dw_mysql_repository.get_column_values(
                table.name, column.name, 10
            )
            # 4. 构造 ColumnInfo 业务实体
            column_info = ColumnInfo(
                id=f"{table.name}.{column.name}",
                name=column.name,
                type=column_types[column.name],
                role=column.role,
                examples=column_values,
                description=column.description,
                alias=column.alias,
                table_id=table.name,
            )
            column_infos.append(column_info)

    # 5. 在一个事务中保存所有表信息和字段信息
    async with self.meta_mysql_repository.session.begin():
        await self.meta_mysql_repository.save_table_infos(table_infos)
        await self.meta_mysql_repository.save_column_infos(column_infos)

    return column_infos

逐行解读：

遍历配置中的每张表（meta_config.tables 来自 YAML）
构造 TableInfo：把 YAML 中的 name、role、description 打包成一个业务实体对象。id 直接用表名
查字段类型：调用 dw_mysql_repository.get_column_types("ods_order_log")，底层执行 SHOW COLUMNS FROM ods_order_log，返回 {"order_id": "bigint", "order_status": "varchar", ...}
查字段示例值：调用 dw_mysql_repository.get_column_values("ods_order_log", "order_status", 10)，底层执行 SELECT DISTINCT order_status FROM ods_order_log LIMIT 10，返回 ["已支付", "已发货", "已签收", ...]
构造 ColumnInfo：把 YAML 中的描述 + dw 库查到的类型和示例值，组合成一个完整的字段信息对象。id 用 表名.字段名 格式（如 ods_order_log.order_status）
事务保存：session.begin() 开启一个事务，在里面同时保存所有 TableInfo 和 ColumnInfo。如果中间出错，整个事务回滚，不会留下半截数据
返回 column_infos：这个列表会被传给后面的 _save_column_info_to_qdrant 和 _save_value_info_to_es

6.2.4 `_save_column_info_to_qdrant()` — 字段信息 → 向量 → Qdrant

async def _save_column_info_to_qdrant(self, column_infos: list[ColumnInfo]):
    # 1. 确保 Qdrant 中的 collection 存在
    await self.column_qdrant_repository.ensure_collection()

    # 2. 构造待保存的数据点
    points: list[dict] = []
    for column_info in column_infos:
        # 字段名作为一条向量文本
        points.append({
            "id": uuid.uuid4(),
            "embedding_text": column_info.name,
            "payload": column_info,
        })
        # 字段描述作为一条向量文本
        points.append({
            "id": uuid.uuid4(),
            "embedding_text": column_info.description,
            "payload": column_info,
        })
        # 每个别名作为一条向量文本
        for alia in column_info.alias:
            points.append({
                "id": uuid.uuid4(),
                "embedding_text": alia,
                "payload": column_info,
            })

    # 3. 批量生成向量（每次最多 10 条）
    embedding_texts = [point["embedding_text"] for point in points]
    embedding_batch_size = 10
    embeddings = []
    for i in range(0, len(embedding_texts), embedding_batch_size):
        batch_embedding_texts = embedding_texts[i : i + embedding_batch_size]
        batch_embeddings = await self.embedding_client.aembed_documents(batch_embedding_texts)
        embeddings.extend(batch_embeddings)

    # 4. 拆出 id、payload 列表，批量写入 Qdrant
    ids = [point["id"] for point in points]
    payloads = [point["payload"] for point in points]
    await self.column_qdrant_repository.upsert(ids, embeddings, payloads)

为什么一个字段要生成多条向量？因为用户提问时可能用不同的表达方式：

用户说"订单状态" → 需要匹配字段名 order_status
用户说"订单的处理进度" → 需要匹配字段描述"订单的处理状态"
用户说"物流情况" → 需要匹配别名

所以每个字段用名字、描述、每个别名分别生成一条向量，全部指向同一个 ColumnInfo（存在 payload 里）。这样无论用户怎么表达，都能通过语义相似度找到正确的字段。

分批处理（embedding_batch_size = 10）是因为 Embedding 服务有并发和 token 限制，一次传太多会报错。

6.2.5 `_save_value_info_to_es()` — 字段取值 → Elasticsearch

async def _save_value_info_to_es(
    self, meta_config: MetaConfig, column_infos: list[ColumnInfo]
):
    # 1. 确保 ES 中的 index 存在
    await self.value_es_repository.ensure_index()

    # 2. 从配置中取出哪些字段需要同步取值（sync=true 的字段）
    column2sync: dict[str, bool] = {}
    for table in meta_config.tables:
        for column in table.columns:
            column2sync[f"{table.name}.{column.name}"] = column.sync

    # 3. 对需要同步取值的字段，从 dw 库查出全部取值
    value_infos: list[ValueInfo] = []
    for column_info in column_infos:
        sync = column2sync[column_info.id]
        if sync:
            table_name = column_info.table_id
            column_name = column_info.name
            values = await self.dw_mysql_repository.get_column_values(
                table_name, column_name, 100000  # 最多取 10 万条
            )
            # 4. 每个取值构造一个 ValueInfo
            current_value_infos = [
                ValueInfo(
                    id=f"{column_info.id}.{value}",
                    value=value,
                    column_id=column_info.id,
                )
                for value in values
            ]
            value_infos.extend(current_value_infos)

    # 5. 批量写入 Elasticsearch
    await self.value_es_repository.index(value_infos)

为什么需要 ES？因为有些字段的取值是有限枚举（如订单状态：已支付、已发货、已签收），用户提问时可能直接说取值名称。ES 做的是全文检索（精确匹配 + 分词匹配），和 Qdrant 的语义检索互补。

不是所有字段都需要同步取值 —— 配置中 sync: true 的字段才会查取值存入 ES。比如 order_id 这种唯一值字段不需要同步，但 order_status 这种枚举值字段需要。

6.2.6 `_save_metrics_to_meta_db()` — 指标信息 → 元数据库

async def _save_metrics_to_meta_db(self, meta_config):
    metric_infos: list[MetricInfo] = []
    column_metrics: list[ColumnMetric] = []

    for metric in meta_config.metrics:
        # 1. 构造 MetricInfo（指标信息）
        metric_info = MetricInfo(
            id=metric.name,
            name=metric.name,
            description=metric.description,
            relevant_columns=metric.relevant_columns,
            alias=metric.alias,
        )
        metric_infos.append(metric_info)

        # 2. 构造 ColumnMetric（字段-指标关联关系）
        for relevant_column in metric.relevant_columns:
            column_metric = ColumnMetric(
                column_id=relevant_column,
                metric_id=metric.name
            )
            column_metrics.append(column_metric)

    # 3. 在一个事务中保存
    async with self.meta_mysql_repository.session.begin():
        await self.meta_mysql_repository.save_metric_infos(metric_infos)
        await self.meta_mysql_repository.save_column_metrics(column_metrics)

    return metric_infos

这个方法的逻辑：

遍历配置中的每个指标（如"总销售额"）
把指标信息打包成 MetricInfo
对每个指标的每个关联字段，创建一条 ColumnMetric 关联记录（比如"总销售额"关联了 order_amount 字段，就生成一条 {column_id: "ods_order_log.order_amount", metric_id: "总销售额"} 的记录）
事务保存

ColumnMetric 表的作用是多对多关联：一个指标可能用到多个字段，一个字段也可能被多个指标引用。

6.2.8 `_save_metric_info_to_qdrant()` — 指标信息 → 向量 → Qdrant

async def _save_metric_info_to_qdrant(self, metric_infos: list[MetricInfo]):
    # 1. 确保 collection 存在
    await self.metric_qdrant_repository.ensure_collection()

    # 2. 构造数据点（和字段信息同理：名字、描述、别名各一条向量）
    points: list[dict] = []
    for metric_info in metric_infos:
        points.append({
            "id": uuid.uuid4(),
            "embedding_text": metric_info.name,
            "payload": metric_info,
        })
        points.append({
            "id": uuid.uuid4(),
            "embedding_text": metric_info.description,
            "payload": metric_info,
        })
        for alia in metric_info.alias:
            points.append({
                "id": uuid.uuid4(),
                "embedding_text": alia,
                "payload": metric_info,
            })

    # 3. 批量生成向量
    ids = [point["id"] for point in points]
    embeddings = []
    embedding_texts = [point["embedding_text"] for point in points]
    embedding_batch_size = 10
    for i in range(0, len(embedding_texts), embedding_batch_size):
        batch_embedding_texts = embedding_texts[i : i + embedding_batch_size]
        batch_embeddings = await self.embedding_client.aembed_documents(batch_embedding_texts)
        embeddings.extend(batch_embeddings)

    payloads = [point["payload"] for point in points]

    # 4. 写入 Qdrant
    await self.metric_qdrant_repository.upsert(ids, embeddings, payloads)

逻辑和 _save_column_info_to_qdrant 完全一样，只是数据源从 column_infos 换成了 metric_infos，目标 collection 从 data-agent-column 换成了 data-agent-metric。

6.3 完整流程图

执行 build() 后，数据流向如下：

meta_config.yaml (配置)
        |
        v
  MetaConfig 对象
        |
        +---> tables -----> _save_tables_to_meta_db()
        |                      |        |
        |                      |        +---> dw库查字段类型 + 示例值
        |                      |        +---> meta库写入 table_info + column_info
        |                      |
        |                      v
        |                  column_infos (返回值)
        |                      |
        |                      +---> _save_column_info_to_qdrant()
        |                      |        文本 -> Embedding -> Qdrant (data-agent-column)
        |                      |
        |                      +---> _save_value_info_to_es()
        |                               dw库查全部取值 -> ES (data-agent-value)
        |
        +---> metrics ----> _save_metrics_to_meta_db()
                               |        |
                               |        +---> meta库写入 metric_info + column_metric
                               |
                               v
                           metric_infos (返回值)
                               |
                               +---> _save_metric_info_to_qdrant()
                                        文本 -> Embedding -> Qdrant (data-agent-metric)

最终产生的结果：

存储系统	Collection / Table	内容
MySQL meta 库	`table_info`	每张表的名称、角色、描述
MySQL meta 库	`column_info`	每个字段的名称、类型、角色、描述、别名、示例值、所属表
MySQL meta 库	`metric_info`	每个指标的名称、描述、关联字段、别名
MySQL meta 库	`column_metric`	字段与指标的多对多关联
Qdrant	`data-agent-column`	字段名/描述/别名的向量 + payload
Qdrant	`data-agent-metric`	指标名/描述/别名的向量 + payload
Elasticsearch	`data-agent-value`	字段取值的全文索引

6.4 执行方式

python -m app.scripts.build_meta_knowledge -c .confmeta_config.yaml

执行完毕后，可以分别检查三个存储系统是否有数据写入：

MySQL：在 PyCharm Database 面板查看 meta 库的四张表
Qdrant：访问 http://localhost:6333/dashboard 查看 collections
Elasticsearch：访问 http://localhost:9200/data-agent-value/_count 查看文档数量

元数据库构建完了，接下来就要构建智能体了，下一章讲构建问数智能体。

推荐专题

最新下载

热门教程

LCODER AI Agent问数实战（三）：搭建元数据知识库

一、配置文件

1.1 table模块：声明要同步哪些表和字段

1.2 metrics 块——声明业务指标：

1.3 meta_config.py 类型定义

1.4 生成MetaConfig对象

二、创建ORM实体类

2.1 什么是ORM

2.2 SQLAlchemy 2.0 的 ORM 语法

2.2.1 DeclarativeBase — 所有实体类的基类

2.2.2 __tablename__ — 指定表名

2.2.3 Mapped 和 mapped_column — 定义字段

2.2.4 mapped_column 的常用参数

2.2.5 常用的数据库列类型

2.3 创建ORM

2.3.1 base.py 基类

2.3.2 TableInfoMySql：表信息实体类

2.3.3 column_info.py：字段信息实体

2.3.4 metric_info.py 指标信息实体

2.3.5 column_metric.py 字段-指标关联实体

三、Entity层，业务实体

3.1 为什么需要Entity层？

3.2 具体的业务实体

3.2.1 TableInfo 表信息实体

3.2.2 ColumnInfo 字段信息实体

3.2.3 MetricInfo 指标信息实体

3.2.4 ColumnMetric 字段指标关联实体

3.2.5 ValueInfo 字段取值实体

3.3 业务实体和ORM实体的对比

四、Mapper层 ORM实体和业务实体的转换器

4.1 Mapper做什么

4.1.1 TableInfoMapper

4.1.2 ColumnInfoMapper

4.1.3 MetricInfoMapper

4.1.4 ColumnMetricMapper

五、Respository层 数据访问层

5.1 Repository 做什么

5.2 Repository 层的整体布局

5.3 MetaMySQLRepository — meta 库读写

构造函数

方法 1：save_table_infos — 批量保存表信息

方法 2：save_column_infos — 批量保存字段信息

方法 3：save_metric_infos — 批量保存指标信息

方法 4：save_column_metrics — 批量保存字段-指标关联

方法 5：get_table_info_by_id — 根据 ID 查询表信息

方法 6：get_column_infos_by_table — 根据表 ID 查所有字段

5.4 DWMySQLRepository — dw 库读取

方法 1：get_column_types — 查询某张表所有字段的数据类型

方法 2：get_column_values — 查询某个字段的不重复取值

方法3：get_db_info() — 获取数据库信息

它做了什么？

返回结果示例：

为什么需要这个方法？

方法4 validate_sql(sql) 校验 SQL

方法5 execute_sql(sql) 执行 SQL

5.5 ColumnQdrantRepository 字段向量集合读写

构造函数

方法 1：ensure_collection — 确保集合存在

方法 2：upsert — 批量写入向量

方法 3：search — 相似度检索

5.6 MetricQdrantRepository 指标向量集合读写

5.7 ValueESRepository — 字段取值全文索引读写

方法 1：ensure_index — 确保索引存在

方法 2：index — 批量写入

方法 3：search — 全文检索

六 、 Service层

6.1 入口脚本

6.2 核心类：MetaKnowledgeService

6.2.1 构造函数 — 依赖注入

6.2.2 build() 方法 — 总指挥的调度逻辑

6.2.3 _save_tables_to_meta_db() — 从配置+数据仓库 → 元数据库

6.2.4 _save_column_info_to_qdrant() — 字段信息 → 向量 → Qdrant

6.2.5 _save_value_info_to_es() — 字段取值 → Elasticsearch

6.2.6 _save_metrics_to_meta_db() — 指标信息 → 元数据库

6.2.8 _save_metric_info_to_qdrant() — 指标信息 → 向量 → Qdrant

6.3 完整流程图

6.4 执行方式

2.2.2 `tablename` — 指定表名

2.2.3 `Mapped` 和 `mapped_column` — 定义字段

2.2.4 `mapped_column` 的常用参数

五、Respository层数据访问层

六、 Service层

6.2.2 `build()` 方法 — 总指挥的调度逻辑

6.2.3 `_save_tables_to_meta_db()` — 从配置+数据仓库 → 元数据库

6.2.4 `_save_column_info_to_qdrant()` — 字段信息 → 向量 → Qdrant

6.2.5 `_save_value_info_to_es()` — 字段取值 → Elasticsearch

6.2.6 `_save_metrics_to_meta_db()` — 指标信息 → 元数据库

6.2.8 `_save_metric_info_to_qdrant()` — 指标信息 → 向量 → Qdrant