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

问数智能体要根据自然语言生成可靠的 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="订单事实表"),
    ...
]

执行过程：

1. 遍历列表，每个 TableInfo 业务实体通过 TableInfoMapper.to_orm() 转成 TableInfoMySQL ORM 实体
2. 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 反查表的完整信息。

执行过程：

1. select(TableInfoMySQL).where(...) 构造查询（类似写 SQL：SELECT * FROM table_info WHERE id = 'dim_region'）
2. session.execute(stmt) 执行查询，拿到结果集
3. scalar_one_or_none() 取出第一条 ORM 实体，没有就返回 None
4. 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()]

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

1. 构建字段示例值时：每个字段查 10 条，填入 ColumnInfo.examples，作为元数据的一部分
2. 构建 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,  # 余弦相似度
            ),
        )

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

执行过程：

1. 先查 "columns" 集合是否存在
2. 如果不存在，就创建一个：
   • 向量维度：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

执行过程：

1. 用 zip 把三个列表对应位置的元素拼在一起
2. 每个三元组组装成一个 PointStruct（Qdrant 的数据点结构）
3. payload.model_dump() 把业务实体转成字典（dataclass 才能序列化）
4. 调用 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

逐行解读：

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

这个方法的逻辑：

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

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 查看文档数量

---

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