仓储层与其他层的交互

通过基于 MySQL、Qdrant、ElasticSearch 构建元数据的流程，了解仓储层与其他层的交互。

项目来源自电商问数。

目录

一、入口脚本全流程

1.1 命令行

如果想使用命令行把同步脚本运行起来，应该在项目根目录下的 powershell 运行：

uv run python -m app.scripts.build_meta_knowledge -c conf/meta_config.yaml

• uv run ...：uv 是包管理器，run 是其子命令，决定运行当前项目的虚拟环境下的哪些依赖
• python -m ...：python 是依赖，-m 表示按模块运行后面的脚本，解决“把app/scripts/当做根目录，导致找不到其他依赖文件”的问题
• app.scripts.build_meta_knowledge：是要运行的 Python 模块路径，由于前面加了 -m，这里 . 对应层级目录，实际上是 app/scripts/build_meta_knowledge.py
• -c：-c 是 —config 的缩写，表示指定配置文件
• conf/meta_config.yaml：是传入的 YAML 配置文件路径

该脚本如何解析命令行呢？

argparse 就是来解决这个问题的，

脚本中用以下代码来解析命令行和传递配置信息：

parser = argparse.ArgumentParser()
parser.add_argument("-c", "--conf", required=True)
args = parser.parse_args()
asyncio.run(build(Path(args.conf)))

• argparse 是 Python 的标准库，详细详细可参考argparse 官方文档

1.2 初始化与对象创建

初始化客户端：

meta_mysql_client_manager.init()
dw_mysql_client_manager.init()
qdrant_client_manager.init()
embedding_client_manager.init()
es_client_manager.init()

然后是打开两个异步 Session，创建 repository 对象：

async with (
    meta_mysql_client_manager.session_factory() as meta_session,
    dw_mysql_client_manager.session_factory() as dw_session,
):
    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
    value_es_repository = ValueESRepository(es_client_manager.client)
    metric_qdrant_repository = MetricQdrantRepository(qdrant_client_manager.client)

这里唯独 embedding_client 没有包成 repository 对象是因为其只需要调用模型接口，不需要像其他的那些数据库/搜索引擎那样负责数据的持久化或搜素

1.3 服务层构建

服务层构建也被包含在 async with 中，

repository 对象被注入到创建的 service 对象中，其 build 函数被异步调用：

    meta_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,
    )
    await meta_knowledge_service.build(config_path)

meta_knowledge_service 的 build 函数读取了配置文件和配置结构定义：

context = OmegaConf.load(config_path)
schema = OmegaConf.structured(MetaConfig)

在2.1中 context 和 schema 会被合并

1.4 关闭客户端

async with 结束后，所有有持久连接的客户端都被断开了：

await meta_mysql_client_manager.close()
await dw_mysql_client_manager.close()
await qdrant_client_manager.close()
await es_client_manager.close()

二、Service 调用 Repository

视角来到 meta_knowledge_service 的 build 函数：

async def build(self, config_path: Path):
    """读取配置并依次构建 Meta MySQL Qdrant 和 ES 中的元数据索引"""
    context = OmegaConf.load(config_path)
    schema = OmegaConf.structured(MetaConfig)
    meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))
    if meta_config.tables:
        column_infos = await self._save_tables_to_meta_db(meta_config)
        logger.info("保存表信息和字段信息到 Meta MySQL")
        await self._save_column_info_to_qdrant(column_infos)
        logger.info("为字段信息建立向量索引")
        await self._save_value_info_to_es(meta_config, column_infos)
        logger.info("为字段取值建立全文索引")
    if meta_config.metrics:
        metric_infos = await self._save_metrics_to_meta_db(meta_config)
        logger.info("保存指标信息到数据库成功")
        await self._save_metrics_to_qdrant(metric_infos)
        logger.info("为指标信息建立向量索引成功")

2.1 合成 meta_config

context = OmegaConf.load(config_path)
schema = OmegaConf.structured(MetaConfig)
meta_config: MetaConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

• context 加载的是 meta_config.yaml，是通过手动设定的内容
• schema 加载的是 meta_config.py，是配置规范的模版
• OmegaConf.merge 校验并合并了两者，审核并把内容覆盖到了模板上，但返回的还不是 MetaConfig 类实例
• OmegaConf.to_object 其转换为 MetaConfig 类实例

OmegaConf 具体用法可参考 OmegaConf 官网

2.2 table：表与字段信息与字段真实值

if meta_config.tables:
    column_infos = await self._save_tables_to_meta_db(meta_config)
    logger.info("保存表信息和字段信息到 Meta MySQL")
    await self._save_column_info_to_qdrant(column_infos)
    logger.info("为字段信息建立向量索引")
    await self._save_value_info_to_es(meta_config, column_infos)
    logger.info("为字段取值建立全文索引")

如果 meta_config.ymal 中填写了 tables 相关的内容：

1. 将表信息和字段信息保存到 Meta MySQL
2. 对字段信息建立向量索引
3. 对指定的维度字段取值建立全文索引

2.3 metrics：指标信息与字段关系

if meta_config.metrics:
    metric_infos = await self._save_metrics_to_meta_db(meta_config)
    logger.info("保存指标信息到数据库成功")
    await self._save_metrics_to_qdrant(metric_infos)
    logger.info("为指标信息建立向量索引成功")

如果 meta_config.ymal 中填写了 metrics 相关的内容：

1. 将指标信息和字段依赖关系保存到 Meta MySQL
2. 对指标信息建立向量索引

三、Repository 从数仓读取数据

meta_knowledge_service 中的以下函数已经被异步调用：

• _save_tables_to_meta_db(meta_config)
• _save_column_info_to_qdrant(column_infos)
• _save_value_info_to_es(meta_config, column_infos)
• _save_metrics_to_meta_db(meta_config)
• _save_metrics_to_qdrant(metric_infos)

3.1 MySQL 读取和映射实体

table_infos: list[TableInfo] = []
column_infos: list[ColumnInfo] = []
for table in meta_config.tables:
    table_info = TableInfo(
        id=table.name,
        name=table.name,
        role=table.role,
        description=table.description,
    )
    table_infos.append(table_info)
    column_types = await self.dw_mysql_repository.get_column_types(table.name)
    for column in table.columns:
        column_values = await self.dw_mysql_repository.get_column_values(
            table.name, column.name
        )
        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)

_save_tables_to_meta_db(meta_config) 里对部分数据的读取与映射的定义：

• 遍历 tables
  • 把表映射整理成业务实体
  • 到 DW 查询 column_types
  • 遍历 columns
    • 到 DW 查询 column_values，就是少量示例值
    • 把列映射整理成业务实体，字段 id 使用 table.column 形式

metric_infos: list[MetricInfo] = []
column_metrics: list[ColumnMetric] = []
for metric in meta_config.metrics:
    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)
    for column in metric.relevant_columns:
        column_metric = ColumnMetric(column_id=column, metric_id=metric.name)
        column_metrics.append(column_metric)

• 遍历 metrics
  • 把指标映射整理成业务实体，id 直接用指标名
  • 遍历 relevant_columns
    • 获取 relevant_columns，表示某个指标依赖某个字段

3.2 Qdrant 映射实体

await self.column_qdrant_repository.ensure_collection()
points: list[dict] = []
for column_info in column_infos:
    points.append(
        {
            "id": uuid.uuid4(),
            "embedding_text": column_info.name,
            "payload": asdict(column_info),
        }
    )
    points.append(
        {
            "id": uuid.uuid4(),
            "embedding_text": column_info.description,
            "payload": asdict(column_info),
        }
    )
    for alia in column_info.alias:
        points.append(
            {
                "id": uuid.uuid4(),
                "embedding_text": alia,
                "payload": asdict(column_info),
            }
        )

• 一个字段不会只生成一个向量点，而是把名字 描述 别名都拆开建立语义入口

await self.metric_qdrant_repository.ensure_collection()
points: list[dict] = []
for metric_info in metric_infos:
    points.append(
        {
            "id": uuid.uuid4(),
            "embedding_text": metric_info.name,
            "payload": asdict(metric_info),
        }
    )
    points.append(
        {
            "id": uuid.uuid4(),
            "embedding_text": metric_info.description,
            "payload": asdict(metric_info),
        }
    )
    for alia in metric_info.alias:
        points.append(
            {
                "id": uuid.uuid4(),
                "embedding_text": alia,
                "payload": asdict(metric_info),
            }
        )

• 和字段一样，一个指标也会拆成名字 描述 别名这几类语义入口

3.3 ElasticSearch 读取和映射实体

await self.value_es_repository.ensure_index()
column2sync: dict[str, bool] = {}
for table in meta_config.tables:
    for column in table.columns:
        column2sync[f"{table.name}.{column.name}"] = column.sync
value_infos: list[ValueInfo] = []
for column_info in column_infos:
    sync = column2sync[column_info.id]
    if sync:
        current_column_values = (
            await self.dw_mysql_repository.get_column_values(
                column_info.table_id, column_info.name, 100000
            )
        )
        current_values_infos = [
            ValueInfo(
                id=f"{column_info.id}.{current_column_value}",
                value=current_column_value,
                column_id=column_info.id,
            )
            for current_column_value in current_column_values
        ]
        value_infos.extend(current_values_infos)

1. 收集所有 id 和 sync 的 true/false 形成字典，决定同步真实值。
2. 遍历 column_infos，通过 字典中的 sync 决定是否同步真实值，向 DW 获取真实值全集后记入 value_infos。

四、Repository 数据写入与事务提交

4.1 MySQL 写入

.session.begin():

1. 如果上下文操作都成功，事务自动提交。
2. 如果中途抛出异常，事务自动回滚。

async with self.meta_mysql_repository.session.begin():
    self.meta_mysql_repository.save_table_infos(table_infos)
    self.meta_mysql_repository.save_column_infos(column_infos)
return column_infos

• 原子性地提交两个数据
• 返回 column_infos，便于交给下一个函数使用

async with self.meta_mysql_repository.session.begin():
    self.meta_mysql_repository.save_metric_infos(metric_infos)
    self.meta_mysql_repository.save_column_metrics(column_metrics)
return metric_infos

• 原子性地提交两个数据
• 返回 metric_infos，便于交给下一个函数使用

4.2 Qdrant 写入与 Embedding 服务

embeddings: list[list[float]] = []
embedding_texts = [point["embedding_text"] for point in points]
embedding_batch_size = 20
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)
ids = [point["id"] for point in points]
payloads = [point["payload"] for point in points]
await self.column_qdrant_repository.upsert(ids, embeddings, payloads)

1. 把向量点的 embedding_text 部分抽出，分批调用 Embedding 服务，把向量点的 id 和 payload 记入 ids 和 payloads。
2. 批量 upsert (插入)向量点。

注意：向量点的 id 用 uuid.uuid4() 是因为同一个字段会有多个向量点，直接用 column_info.id 会重复。

embeddings: list[list[float]] = []
embedding_texts = [point["embedding_text"] for point in points]
embedding_batch_size = 20
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)
ids = [point["id"] for point in points]
payloads = [point["payload"] for point in points]
await self.metric_qdrant_repository.upsert(ids, embeddings, payloads)

1. 把向量点的 embedding_text 部分抽出，分批调用 Embedding 服务，把向量点的 id 和 payload 记入 ids 和 payloads。
2. 批量 upsert (插入)向量点。

注意：这里向量点的 id 直接用 metric.name，和前者不一样，指标配置里本身就已经要求 name 具有业务唯一性。

4.3 ElasticSearch 写入

await self.value_es_repository.index(value_infos)

• 批量 index (记入) 真实值全集

五、Entity、ORM Model 和 Mapper

5.1 entity：业务实体

• app/entities/column_info.py
• app/entities/column_metric.py
• app/entities/metric_info.py
• app/entities/table_info.py
• app/entities/value_info.py

都是业务实体，其中 column_info.py 内是这样的：

from dataclasses import dataclass
from typing import Any
@dataclass
class ColumnInfo:
    """系统内部统一使用的字段元数据表达"""
    id: str
    name: str
    type: str
    role: str
    examples: list[Any]
    description: str
    alias: list[str]
    table_id: str

前面的配置文件和配置结构定义通过 dataclass + OmegaConf 把配置结构和配置内容合并成统一对象。

这里的 dataclass 也是同理，也是对配置结构的定义，是服务于 Service 层的。

5.2 model：ORM 模型

• app/models/base.py
• app/models/column_info.py
• app/models/column_metric.py
• app/models/metric_info.py
• app/models/table_info.py

都是ORM 模型，其中 column_info.py 内是这样的：

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy.types import JSON
from app.models.base import Base
class ColumnInfoMySQL(Base):
    """字段元数据表对应的 ORM 模型"""
    __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="所属表编号")

1. id 采用 表名.字段名 的组合形式，这样在整个数仓范围内更容易保证唯一性。
2. examples 和 alias 都使用 JSON 存储，方便保存列表结构而不是再拆独立子表。

服务于 Repository 层的数据库。

5.3 mapper：转换器

• app/repositories/mysql/meta/mappers/column_info_mapper.py
• app/repositories/mysql/meta/mappers/column_metric_mapper.py
• app/repositories/mysql/meta/mappers/metric_info_mapper.py
• app/repositories/mysql/meta/mappers/table_info_mapper.py

都是转换器，其中 column_info_mapper.py 内是这样的：

from dataclasses import asdict
from app.entities.column_info import ColumnInfo
from app.models.column_info import ColumnInfoMySQL
class ColumnInfoMapper:
    """负责 `ColumnInfo` 与 `ColumnInfoMySQL` 之间的双向转换"""
    @staticmethod
    def to_entity(column_info_mysql: ColumnInfoMySQL) -> ColumnInfo:
        """把 ORM 模型转换回字段业务实体"""
        return ColumnInfo(
            id=column_info_mysql.id,
            name=column_info_mysql.name,
            type=column_info_mysql.type,
            role=column_info_mysql.role,
            examples=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:
        """把字段业务实体转换成 ORM 模型用于持久化"""
        return ColumnInfoMySQL(**asdict(column_info))

• ColumnInfo 是业务实体
• ColumnInfoMySQL 是 ORM 模型
• Mapper 负责把前者转换成后者

5.4 数据持久化

• app/repositories/mysql/meta/meta_mysql_repository.py

负责 meta MySQL 数据库的读写接口：

class MetaMySQLRepository:
    """负责把元数据业务实体持久化到 Meta MySQL"""
    def __init__(self, session: AsyncSession):
        self.session = session
    def save_table_infos(self, table_infos: list[TableInfo]):
        """批量保存表元数据。输入仍然是业务实体，而不是 ORM 模型"""
        self.session.add_all(
            [TableInfoMapper.to_model(table_info) for table_info in table_infos]
        )
    def save_column_infos(self, column_infos: list[ColumnInfo]):
        """批量保存字段元数据。实体到模型的转换统一通过 Mapper 完成"""
        self.session.add_all(
            [ColumnInfoMapper.to_model(column_info) for column_info in column_infos]
        )
    def save_metric_infos(self, metric_infos: list[MetricInfo]):
        """批量保存指标元数据。指标本身和字段关联关系分开写入"""
        self.session.add_all(
            [MetricInfoMapper.to_model(metric_info) for metric_info in metric_infos]
        )
    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, id: str) -> ColumnInfo | None:
        """按字段 id 查询字段元数据，供召回信息合并阶段补齐字段上下文"""
        column_info: ColumnInfoMySQL | None = await self.session.get(
            ColumnInfoMySQL, id
        )
        if column_info:
            return ColumnInfoMapper.to_entity(column_info)
        else:
            return None
    async def get_table_info_by_id(self, id: str) -> TableInfo | None:
        """按表 id 查询表元数据，最终组装成提示词里的表结构信息"""
        table_info: TableInfoMySQL | None = await self.session.get(TableInfoMySQL, id)
        if table_info:
            return TableInfoMapper.to_entity(table_info)
        else:
            return None
    async def get_key_columns_by_table_id(self, table_id: str) -> list[ColumnInfo]:
        """查询指定表的主外键字段，避免 Join 关键字段被向量召回漏掉"""
        stmt = select(ColumnInfoMySQL).where(
            ColumnInfoMySQL.table_id == table_id,
            ColumnInfoMySQL.role.in_(("primary_key", "foreign_key")),
        )
        result = await self.session.scalars(stmt)
        return [ColumnInfoMapper.to_entity(model) for model in result.all()]

• 创建 session (会话)
  • init(session)：初始化 session (会话)
• 批量数据加入 session (会话)，暂不提交
  • save_table_infos(table_infos)：使用 to_model
  • save_column_infos(column_infos)：使用 to_model
  • save_metric_infos(metric_infos)：使用 to_model
  • save_column_metrics(column_metrics)：使用 to_model
• 按 id 获取 table (表) / columns (列)
  • get_column_info_by_id(id)：异步，使用 to_entity
  • get_table_info_by_id(id)：异步，使用 to_entity
• 按表获取 columns (列) 的 key (键)
  • get_key_columns_by_table_id(table_id)：异步，获取所有是 primary_key (主键)，foreign_key (外键) 的字段，使用 to_entity

结语

看教程，记笔记，写文章是我这几周在重复的事；

从一开始的不会总结，乱起标题，只能靠生成式语言模型的回复填充文章空缺内容；

到现在只靠代码、教程、笔记就能完成文章，甚至发现笔记中的错误；

我能感觉到自己的进步；

从全靠 AI 到 AI 辅助，虽然离独立完成项目还有一段距离；

但经验在积累。