02 基础设施搭建

学习理念：本章搭建项目所需的所有基础设施——MySQL、Qdrant、ES、Embedding 的客户端管理类和日志系统。核心是理解"ClientManager 模式"：每个外部存储系统都有一个独立的 ClientManager 类封装创建/管理/关闭逻辑，方便在 FastAPI 生命周期中统一初始化/释放。
海外对标：Airflow 的 Connection 管理、Spring Boot 的 DataSourceAutoConfiguration

本节 AI 替代率：~75% | 人工干预率：~25%

角色	能力范围
🤖 AI 擅长	生成 SQLAlchemy 连接代码、Qdrant/ES 客户端样板代码、Loguru 配置
👤 人类需理解	异步 SQLAlchemy 的 engine→session_factory→session 三级关系、Qdrant PointStruct 结构、ES mapping 设计

📌 原文说明：以下内容来源于原始笔记第4章"基础设施搭建"，包含项目目录结构、应用配置管理、MySQL/Embedding/Qdrant/ES/日志等客户端管理。原文全部保留，补充了代码阅读优先级标注和文件索引。

一、项目目录结构

yaml

data-agent 根目录
├── app 代码目录
│   ├── clients 数据库客户端
│   ├── conf 配置类
│   ├── core 基础设施
├── conf 配置文件
├── logs 日志目录

二、应用配置参数管理

2.1 应用配置文件

本项目采用 YAML 文件管理配置参数，配置文件的路径为 data-agent/conf/app_config.yaml 。以下是本项目所需的全部参数。

🔥 【P0 必须理解】 这个配置文件是项目的"总控室"，所有外部服务的地址、端口、凭证都在这里。后续所有 ClientManager 都从这个配置读取连接信息。

yaml

logging:
  file:
    enable: true
    level: INFO
    path: logs
    rotation: "10 MB"
    retention: "7 days"
  console:
    enable: true
    level: INFO

db_meta:
  host: localhost
  port: 3306
  user: atguigu
  password: Atguigu.123
  database: meta

db_dw:
  host: localhost
  port: 3306
  user: atguigu
  password: Atguigu.123
  database: dw

qdrant:
  host: localhost
  port: 6333
  embedding_size: 1024

embedding:
  host: localhost
  port: 8081
  model: BAAI/bge-large-zh-v1.5

es:
  host: localhost
  port: 9200
  index_name: data_agent

llm:
  model_name: deepseek-chat
  api_key: <deepseek_api_key>

2.2 加载工具

本项目使用的yaml配置文件加载工具为OmegaConf，具体用法参考其官网即可。

🟡 【P1 看注释就行】 OmegaConf 是一个分层配置系统。核心流程：从 YAML 加载 → 定义 dataclass schema → merge 验证 → 输出类型安全的 dataclass 对象。比直接解析 YAML 更安全、可类型提示。

用于读取配置文件的代码放置在data-agent/app/conf/app_config.py文件中，具体内容如下：

🔥 【P0 必须理解】 这个 dataclass 结构就是 app_config.yaml 的类型映射。每个 section 对应一个子 dataclass，最终通过 OmegaConf.merge(schema, context) 实现类型安全的配置加载。

python

from dataclasses import dataclass
from pathlib import Path
from omegaconf import OmegaConf


@dataclass
class File:
    enable: bool
    level: str
    path: str
    rotation: str
    retention: str

@dataclass
class Console:
    enable: bool
    level: str

@dataclass
class LoggingConfig:
    file: File
    console: Console

@dataclass
class DBConfig:
    host: str
    port: int
    user: str
    password: str
    database: str

@dataclass
class QdrantConfig:
    host: str
    port: int
    embedding_size: int

@dataclass
class EmbeddingConfig:
    host: str
    port: int
    model: str

@dataclass
class ESConfig:
    host: str
    port: int
    index_name: str

@dataclass
class LLMConfig:
    model_name: str
    api_key: str

@dataclass
class AppConfig:
    logging: LoggingConfig
    db_meta: DBConfig
    db_dw: DBConfig
    qdrant: QdrantConfig
    embedding: EmbeddingConfig
    es: ESConfig
    llm: LLMConfig


config_file = Path(__file__).parents[2] / 'conf' / 'app_config.yaml'
context = OmegaConf.load(config_file)
schema = OmegaConf.structured(AppConfig)
app_config: AppConfig = OmegaConf.to_object(OmegaConf.merge(schema, context))

三、MySQL客户端管理

本项目中的MySQL客户端使用SQLAlchemy，具体用法参考官方文档。

🔥 【P0 必须理解】 SQLAlchemy Async 的核心链条：create_async_engine() → 创建连接引擎 → async_sessionmaker(engine) → 创建会话工厂 → async with session_factory() as session: → 创建数据库会话。注意连接 URL 使用 mysql+asyncmy:// 驱动前缀。

在data-agent/app/clients/mysql_client_manager.py中编写如下代码，用来管理MySQL客户端。

python

import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.conf.app_config import DBConfig, app_config


class MySQLClientManager:
    def __init__(self, db_config: DBConfig):
        self.db_config = db_config
        self.engine = None
        self.session_factory = None

    def _get_url(self):
        return f"mysql+asyncmy://{self.db_config.user}:{self.db_config.password}@{self.db_config.host}:{self.db_config.port}/{self.db_config.database}?charset=utf8mb4"

    def init(self):
        self.engine = create_async_engine(self._get_url())
        self.session_factory = async_sessionmaker(self.engine)

    async def close(self):
        if self.engine:
            await self.engine.dispose()


dw_mysql_client_manager = MySQLClientManager(app_config.db_dw)
meta_mysql_client_manager = MySQLClientManager(app_config.db_meta)

if __name__ == '__main__':
    dw_mysql_client_manager.init()
    async def test():
        async with dw_mysql_client_manager.session_factory() as session:
            result = await session.execute(text("show tables;"))
            print(result.fetchall())
    asyncio.run(test())

四、Embedding客户端管理

本项目中的Text Embedding Inference客户端使用HuggingFaceEndpointEmbeddings，具体用法参考官方文档。

🟡 【P1 看注释就行】 HuggingFaceEndpointEmbeddings 将 TEI 的 HTTP API 封装为 LangChain 的标准 Embeddings 接口。关键方法：aembed_query()（单条异步嵌入）、aembed_documents()（批量异步嵌入）。

在data-agent/app/clients/embedding_client_manager.py中编写如下代码，用来管理Text Embedding Inference客户端。

python

from langchain_huggingface import HuggingFaceEndpointEmbeddings
from app.config.app_config import EmbeddingConfig, app_config


class EmbeddingClientManager:
    def __init__(self, config: EmbeddingConfig):
        self.config = config
        self.client: HuggingFaceEndpointEmbeddings | None = None

    def init(self):
        self.client = HuggingFaceEndpointEmbeddings(model=f"http://{self.config.host}:{self.config.port}")


embedding_client_manager = EmbeddingClientManager(app_config.embedding)

if __name__ == '__main__':
    client = EmbeddingClientManager(app_config.embedding)
    client.init()
    query = client.client.embed_query("hello world")
    print(len(query))
    print(query)

五、Qdrant客户端管理

本项目中向量数据库采用Qdrant，Qdrant客户端使用qdrant-client，具体用法参考官方文档。

🔥 【P0 必须理解】 Qdrant 的核心概念：
collection = 表（需要预定义 vector size 和 distance metric）
PointStruct(id, vector, payload) = 一条记录（id、向量、元数据）
query_points() = 向量检索（返回 score + payload）
本项目用 COSINE 距离作为相似度度量

在data-agent/app/clients/qdrant_client_manager.py中编写如下代码，用来管理Qdrant客户端。

python

import asyncio
import random
from typing import Optional
from qdrant_client import AsyncQdrantClient, models
from app.config.app_config import QdrantConfig
from app.models import qdrant


class QdrantClientManager:
    def __init__(self, qdrant_config: QdrantConfig):
        self.qdrant_config = qdrant_config
        self.client: Optional[AsyncQdrantClient] = None

    def _get_url(self):
        return f"http://{self.qdrant_config.host}:{self.qdrant_config.port}"

    def init(self):
        self.client = AsyncQdrantClient(url=self._get_url())

    async def close(self):
        await self.client.close()


qdrant_client_manager = QdrantClientManager(qdrant)

if __name__ == '__main__':
    qdrant_client_manager.init()
    async def test():
        client = qdrant_client_manager.client
        if not await client.collection_exists("my_collection"):
            await client.create_collection(
                collection_name="my_collection",
                vectors_config=models.VectorParams(size=10, distance=models.Distance.COSINE),
            )
        await client.upsert(
            collection_name="my_collection",
            points=[
                models.PointStruct(id=i, vector=[random.random() for _ in range(10)])
                for i in range(100)
            ],
        )
        res = await client.query_points(
            collection_name="my_collection",
            query=[random.random() for _ in range(10)],
            limit=10,
            score_threshold=0.8
        )
        print(res)
    asyncio.run(test())

六、ES客户端管理

本项目中的ES客户端使用elasticsearch，具体用法参考官方文档。

🔥 【P0 必须理解】 本项目 ES 的索引 mapping 配置了 ik_max_word 分词器（需要在 ES 镜像中安装 ik 插件），用于对中文维度值（如"华北"、"手机数码"）进行更细粒度的全文检索。

在data-agent/app/clients/es_client_manager.py中编写如下代码，用来管理ES客户端。

python

import asyncio
from typing import Optional
from elasticsearch import AsyncElasticsearch
from app.conf.app_config import ESConfig, app_config


class ESClientManager:
    def __init__(self, es_config: ESConfig):
        self.es_config = es_config
        self.client: Optional[AsyncElasticsearch] = None

    def _get_url(self):
        return f"http://{self.es_config.host}:{self.es_config.port}"

    def init(self):
        self.client = AsyncElasticsearch(hosts=[self._get_url()])

    async def close(self):
        await self.client.close()


es_client_manager = ESClientManager(app_config.es)

if __name__ == '__main__':
    es_client_manager.init()
    async def test():
        client = es_client_manager.client
        # 创建索引
        await client.indices.create(
            index="my-books",
            mappings={
                "dynamic": False,
                "properties": {
                    "name": {"type": "text"},
                    "author": {"type": "text"},
                    "release_date": {"type": "date", "format": "yyyy-MM-dd"},
                    "page_count": {"type": "integer"}
                }
            },
        )
        # 插入数据
        await client.bulk(operations=[
            {"index": {"_index": "my-books"}},
            {"name": "Revelation Space", "author": "Alastair Reynolds", "release_date": "2000-03-15", "page_count": 585},
            # ... 更多文档省略
        ])
        # 搜索
        resp = await client.search(index="my-books", query={"match": {"name": "brave"}})
        print(resp)
    asyncio.run(test())

七、日志管理

本项目使用loguru管理日志，具体用法参考官网。

🟡 【P1 看注释就行】 Loguru 的核心设计：logger.add(sink, format, level, rotation, retention) — 支持同时输出到控制台和文件，自动轮转和清理。本项目通过 contextvars + logger.patch() 实现每个请求的 request_id 注入。

在data-agent/app/core/log.py中编写如下代码，统一管理日志。

python

import asyncio
import sys
from pathlib import Path
from loguru import logger
from app.conf.app_config import app_config
from app.core.context import request_id_ctx_var

# 配置日志格式
log_format = (
    "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
    "<level>{level: <8}</level> | "
    "<magenta>request_id - {extra[request_id]}</magenta> | "
    "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
    "<level>{message}</level>"
)

# 注入request_id到日志记录中
def inject_request_id(record):
    request_id = request_id_ctx_var.get()
    record["extra"]["request_id"] = request_id

logger.remove()
logger = logger.patch(inject_request_id)

# 控制台日志
if app_config.logging.console.enable:
    logger.add(sink=sys.stdout, level=app_config.logging.console.level, format=log_format)

# 文件日志
if app_config.logging.file.enable:
    PROJECT_ROOT = Path(__file__).parents[2]
    LOG_DIR = PROJECT_ROOT / "logs"
    LOG_FILE = LOG_DIR / "app.log"
    LOG_DIR.mkdir(parents=True, exist_ok=True)
    logger.add(
        sink=LOG_FILE,
        level=app_config.logging.file.level,
        format=log_format,
        rotation=app_config.logging.file.rotation,
        retention=app_config.logging.file.retention,
        encoding="utf-8"
    )

if __name__ == '__main__':
    async def print_log(message: str):
        logger.info(message)

    async def test1():
        request_id_ctx_var.set("request-1")
        await asyncio.sleep(1)
        await print_log("request-1")

    async def test2():
        request_id_ctx_var.set("request-2")
        await asyncio.sleep(1)
        await print_log("request-2")

    async def main():
        await asyncio.gather(test1(), test2())

    asyncio.run(main())

在data-agent/app/core/context.py中添加如下代码，定义上下文变量：

python

from contextvars import ContextVar
request_id_ctx_var = ContextVar("request_id", default="1")

八、企业痛点-方案映射

痛点	传统方案	AI Agent 方案	效率提升
多数据库/服务连接管理混乱	手动创建关闭 connection	ClientManager 统一管理	运维成本降低 50%
日志分散难追踪	print / logging 各自配置	Loguru + request_id 全链路追踪	排错速度提升 3x
Embedding 模型部署复杂	自己写推理服务	TEI 一键容器化部署	部署时间从天→分钟

九、本阶段文件索引

优先级	文件	路径（相对于 `data-agent/`）
🔥 P0	`app_config.yaml`	`conf/app_config.yaml`
🔥 P0	`app_config.py`	`app/conf/app_config.py`
🔥 P0	`mysql_client_manager.py`	`app/clients/mysql_client_manager.py`
🟡 P1	`embedding_client_manager.py`	`app/clients/embedding_client_manager.py`
🟡 P1	`qdrant_client_manager.py`	`app/clients/qdrant_client_manager.py`
🟡 P1	`es_client_manager.py`	`app/clients/es_client_manager.py`
🟡 P1	`log.py`	`app/core/log.py`
🟢 P2	`context.py`	`app/core/context.py`

02 基础设施搭建 ​

一、项目目录结构 ​

二、应用配置参数管理 ​

2.1 应用配置文件 ​

2.2 加载工具 ​

三、MySQL客户端管理 ​

四、Embedding客户端管理 ​

五、Qdrant客户端管理 ​

六、ES客户端管理 ​

七、日志管理 ​

八、企业痛点-方案映射 ​

九、本阶段文件索引 ​

02 基础设施搭建

一、项目目录结构

二、应用配置参数管理

2.1 应用配置文件

2.2 加载工具

三、MySQL客户端管理

四、Embedding客户端管理

五、Qdrant客户端管理

六、ES客户端管理

七、日志管理

八、企业痛点-方案映射

九、本阶段文件索引