基础设施层封装

对 Qdrant / Elasticsearch / Embedding / MySQL / Loguru 这几个基础设施的封装过程。

在一个完整的 Agent 项目中，除了核心业务逻辑，还需要依赖各种外部基础设施来完成具体任务。

例如：

• Qdrant：负责字段和指标的语义向量检索。
• Elasticsearch：负责字段实际取值的检索。
• Embedding：负责将文本转换为向量。
• MySQL：负责保存业务数据和元数据。
• Loguru：负责统一管理项目日志。

这些服务各自承担不同的职责，但对于业务代码来说，它们都有一个共同点：需要通过对应的 Python 客户端或 SDK 进行调用。

如果每个业务模块都自行创建客户端、读取配置、管理连接，不仅会产生重复代码，还会使资源管理和异常处理变得分散。

因此，项目将这些基础设施统一放在 clients 模块中进行封装，负责客户端初始化、配置管理和资源释放，为上层业务代码提供统一的访问入口。

本文将结合项目实际代码，介绍 Qdrant、Elasticsearch、Embedding、MySQL 和 Loguru 的封装过程。

本机通过 docker 挂载端口
6333 和 6334 访问 Qdrant 的 HTTP 和 gRPC 端口
9200 访问 Elasticsearch
5601 访问 Kibana (Elasticsearch的可视化前端)
8081 访问 Embedding 服务
3306 访问 MySQL

目录

Qdrant 客户端

Qdrant 简介

Qdrant：在这里一个专门用于保存向量并执行相似度检索的数据库。

1. 保存向量化的字段和指标信息(不负责向量化)
2. 向量相似度检索
3. 召回语义相关的字段和指标信息

Qdrant 概念

封装 Qdrant 客户端

项目对应文件路径：shopkeeper-agent/app/clients/qdrant_client_manager.py

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

1. 创建客户端管理器

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()

1. 使用 AsyncQdrantClient 而不是同步客户端，因为当前 Agent 项目采用异步编程模式。
2. 客户端初始化没有直接放在 __init__() 中，而是单独提供 init() 方法这样做可以将对象创建和资源初始化分离。

• __init__()：先保存配置对象，后把 client 声明出来，但不把资源初始化
• _get_url()；根据配置文件拼出 Qdrant 服务地址
• init()：创建异步客户端，不在 __init__ 中直接初始化，是为了和项目的生命周期管理保持一致
• close()：项目关闭时统一关闭客户端连接，async 是因为 await 只能写在异步函数内，这里需要等待 TCP 断开

2. 创建全局管理器对象

qdrant_client_manager = QdrantClientManager(app_config.qdrant)

• 创建一个全局唯一的管理器对象
• 后续项目中的其他模块都通过它来获取同一套 Qdrant 客户端

测试 Qdrant 客户端

完整 main 代码：

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())

• 后续内容都将在 main 中执行。
• 步骤1 以后的内容在 async def test(): 内

1. 初始化和释放客户端

qdrant_client_manager.init()
asyncio.run(test())

• 初始化与项目生命周期一致的客户端
• 用异步的方式运行 test()

2. 创建 Collection

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,
        ),
    )

在没有同名集合的情况下创建了一个名为 my_collection 的集合。 其中：

• size=10：表示向量维度为 10。
• distance=models.Distance.COSINE：表示使用余弦距离进行向量相似度计算。

实际项目中，向量维度需要与 Embedding 模型输出的维度保持一致。

3. 写入 Point

await client.upsert(
    collection_name="my_collection",
    points=[
        models.PointStruct(
            id=i,
            vector=[random.random() for _ in range(10)],
        )
        for i in range(100)
    ],
)

这里使用 upsert() 向集合中写入 100 条 Point，每条 Point 包含：

• 一个唯一 ID。
• 一个 10 维随机向量。

upsert()：如果指定 ID 的 Point 不存在，则创建；如果已经存在，则更新。

这里使用随机向量只是为了验证 Qdrant 的基本写入功能，实际业务中应该使用 Embedding 模型生成的向量。

4. 执行向量检索

res = await client.query_points(
    collection_name="my_collection",
    query=[random.random() for _ in range(10)],  # type: ignore
    limit=10,
    score_threshold=0.8,
)

这里使用随机生成的 10 维向量作为查询向量。参数说明：

• collection_name：指定需要检索的集合。
• query：待检索的向量。
• limit=10：最多返回 10 条结果。
• score_threshold=0.8：设置最低相似度分数阈值。

5. 测试

执行文件：

python -m app.clients.qdrant_client_manager

得到检索结果：

points=[ScoredPoint(id=33, version=4, score=0.9287777, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=45, version=4, score=0.9019778, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=58, version=4, score=0.90187067, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=75, version=4, score=0.8799222, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=66, version=4, score=0.87921834, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=87, version=4, score=0.8753782, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=73, version=4, score=0.86765623, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=27, version=4, score=0.86216307, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=16, version=4, score=0.8618801, payload={}, vector=None, shard_key=None, order_value=None), ScoredPoint(id=69, version=4, score=0.8546801, payload={}, vector=None, shard_key=None, order_value=None)]

Elasticsearch 客户端

Elasticsearch 简介

Elasticsearch：在这里是一个负责文本关键词 / 全文检索的搜索引擎。

1. 为字段取值建立全文索引
2. 自然语言表达映射到真实存在的值域

Elasticsearch 概念

mapping 是 Elasticsearch 里的说法，而 schema 是很多数据库或数据系统里更常见的说法。

• dynamic mapping：自动推断字段类型，动态映射
• explicit mapping：显式定义字段结构和字段类型

封装 Elasticsearch 客户端

项目对应文件路径：shopkeeper-agent/app/clients/es_client_manager.py

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

1. 创建客户端管理器

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()

几乎和 qdrant 的客户端管理器一致，函数功能：

• __init__()：保存 ES 配置对象，声明 client
• _get_url()：ES 服务地址
• init()：创建异步 ES 客户端，这里和 qdrant 不同，hosts 之所以是列表，是为了兼容 ES 常见的集群连接方式
• close()：关闭客户端连接

2. 创建全局管理器对象

es_client_manager = ESClientManager(app_config.es)

创建唯一的 ES 客户端管理器对象。

测试 Embedding 客户端

完整 main 代码：

if __name__ == "__main__":
    # 先初始化客户端，后面的测试逻辑才能真正访问 ES
    es_client_manager.init()
    async def test():
        # 取出真正的 AsyncElasticsearch 客户端
        client = es_client_manager.client
        # 创建索引
        # 这里同时显式定义了字段结构
        # dynamic=False 表示关闭动态映射，要求写入数据必须符合当前定义
        await client.indices.create(
            index="my-books",
            mappings={
                "dynamic": False,
                "properties": {
                    # 书名和作者适合做全文检索，所以定义为 text
                    "name": {"type": "text"},
                    "author": {"type": "text"},
                    # 日期字段按日期类型处理
                    "release_date": {"type": "date", "format": "yyyy-MM-dd"},
                    # 页数字段按整数处理
                    "page_count": {"type": "integer"},
                },
            },
        )
        # 插入数据
        # bulk 采用“操作说明 + 数据本体”交替出现的格式
        # 适合一次性写入多条文档
        await client.bulk(
            operations=[
                {"index": {"_index": "my-books"}},
                {
                    "name": "Revelation Space",
                    "author": "Alastair Reynolds",
                    "release_date": "2000-03-15",
                    "page_count": 585,
                },
                {"index": {"_index": "my-books"}},
                {
                    "name": "1984",
                    "author": "George Orwell",
                    "release_date": "1985-06-01",
                    "page_count": 328,
                },
                {"index": {"_index": "my-books"}},
                {
                    "name": "Fahrenheit 451",
                    "author": "Ray Bradbury",
                    "release_date": "1953-10-15",
                    "page_count": 227,
                },
                {"index": {"_index": "my-books"}},
                {
                    "name": "Brave New World",
                    "author": "Aldous Huxley",
                    "release_date": "1932-06-01",
                    "page_count": 268,
                },
                {"index": {"_index": "my-books"}},
                {
                    "name": "The Handmaids Tale",
                    "author": "Margaret Atwood",
                    "release_date": "1985-06-01",
                    "page_count": 311,
                },
            ],
        )
        # 搜索
        # 在 name 字段上执行 match 查询
        # 这里演示的是最基础的全文检索能力
        resp = await client.search(
            index="my-books",
            query={"match": {"name": "brave"}},
        )
        # 打印查询结果，便于观察 hits 和返回结构
        print(resp)
        # 测试结束后关闭客户端连接
        await es_client_manager.close()
    # 运行异步测试函数
    asyncio.run(test())

1. 初始化和释放客户端

if __name__ == "__main__":
    es_client_manager.init()
    asyncio.run(test())

初始化后运行。

2. 创建索引

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"},
        },
    },
)

client.indices.create 创建索引，定义了字段结构。索引名称及其属性：

• index：单个索引名称
• mapping：字段结构
• dynamic：动态映射，False 时为关闭
• name、author：type 为 text 时全文检索，文本可被拆开、搜索、匹配；type 为 keyword 时精确匹配，文本不可拆开
• release_date：日期字段按日期类型处理，设定相应格式。
• page_count：页数字段按整数处理

3. 插入数据

await client.bulk(
    operations=[
        {"index": {"_index": "my-books"}},
        {
            "name": "Revelation Space",
            "author": "Alastair Reynolds",
            "release_date": "2000-03-15",
            "page_count": 585,
        },
        {"index": {"_index": "my-books"}},
        {
            "name": "1984",
            "author": "George Orwell",
            "release_date": "1985-06-01",
            "page_count": 328,
        },
        {"index": {"_index": "my-books"}},
        {
            "name": "Fahrenheit 451",
            "author": "Ray Bradbury",
            "release_date": "1953-10-15",
            "page_count": 227,
        },
        {"index": {"_index": "my-books"}},
        {
            "name": "Brave New World",
            "author": "Aldous Huxley",
            "release_date": "1932-06-01",
            "page_count": 268,
        },
        {"index": {"_index": "my-books"}},
        {
            "name": "The Handmaids Tale",
            "author": "Margaret Atwood",
            "release_date": "1985-06-01",
            "page_count": 311,
        },
    ],
)

client.bulk 插入数据，bulk 采用“操作说明 + 数据本体”交替出现的格式，适合一次性写入多条文档。

4. 搜索与打印

resp = await client.search(
    index="my-books",
    query={"match": {"name": "brave"}},
)
print(resp)

client.search 在 name 字段里搜索 “brave” 的文档。参数说明：

• index：指定搜索的索引
• query：使用 match 做全文检索，如果是 term 则是精确检索

最后打印查询结果，便于观察 hits 和返回结构。

5. 测试

执行文件：

python -m app.clients.es_client_manager

得到结果：

{'took': 5, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 1, 'relation': 'eq'}, 'max_score': 1.2067741, 'hits': [{'_index': 'my-books', '_id': 'SDB_np0BrzSqcXAMSMVk', '_score': 1.2067741, '_source': {'name': 'Brave New World', 'author': 'Aldous Huxley', 'release_date': '1932-06-01', 'page_count': 268}}]}}

Embedding 客户端

Embedding 简介

Embedding：是一个负责把自然语言、字段或指标转换成可以参与向量计算的转换器。

• 用户问题里的关键词，需要先做向量化，才能去 Qdrant 检索相似字段和指标
• 元数据库中的表描述、字段描述、指标描述，也需要先做向量化，才能提前写入向量库

TEI 简介

Text Embeddings Inference，通常简称 TEI。它本质上就是一个专门用来部署 Embedding 模型、并通过 HTTP 接口对外提供向量推理能力的服务框架。

• BAAI/bge-large-zh-v1.5 是真正负责把文本转成向量的模型
• TEI 负责把这个模型跑起来，并暴露成一个可调用的服务
• EmbeddingClientManager 负责在项目里把这层服务接进来

TEI 的访问方式：这里使用 LangChain 提供的 HuggingFaceEndpointEmbeddings。

封装 Embedding 客户端

项目对应文件路径：shopkeeper-agent/app/clients/embedding_client_manager.py

import asyncio
from typing import Optional
from huggingface_hub import AsyncInferenceClient, InferenceClient
from langchain_huggingface import HuggingFaceEndpointEmbeddings
from app.conf.app_config import EmbeddingConfig, app_config

1. 创建客户端管理器

class EmbeddingClientManager:
    def __init__(self, config: EmbeddingConfig):
        self.client: Optional[HuggingFaceEndpointEmbeddings] = None
        self.config = config
    def _get_url(self):
        return f"http://{self.config.host}:{self.config.port}"
    def init(self):
        self.client = HuggingFaceEndpointEmbeddings(model=self.config.model)
        endpoint_url = f"{self._get_url()}/embed"
        self.client.client = InferenceClient(model=endpoint_url)
        self.client.async_client = AsyncInferenceClient(model=endpoint_url)

前面的 Qdrant、Elasticsearch、MySQL 客户端，很多都需要显式初始化和关闭，是因为它们更接近：长连接、连接池、持久化数据库连接。

而当前这个 Embedding 客户端，本质上更接近一个无状态的 HTTP 调用封装。它不像数据库连接那样需要长期持有连接资源，所以这里没有专门去写一个 close() 方法。这也是为什么它的实现会明显更轻。函数说明：

• __init__()：客户端在模块导入阶段先不立即创建，避免启动时就发起外部依赖连接，然后保存 Embedding 服务配置，供 init() 时组装服务访问地址使用
• _get_url()：通过 host + port 访问外部已启动的 Embedding 推理服务
• init()：应用启动阶段时初始化，LangChain 的 model 参数使用模型 ID，本地 TEI 地址交给 InferenceClient 和 AsyncInferenceClient，实现通过接入 TEI 服务来获取 Embedding 模型的输出结果

2. 创建全局管理器对象

embedding_client_manager = EmbeddingClientManager(app_config.embedding)

模块级单例，供其他模块按需复用同一个客户端管理器。

测试 Embedding 客户端

完整 main 文件：

if __name__ == "__main__":
    # 本地调试入口：初始化客户端后执行一次最小化向量化调用
    embedding_client_manager.init()
    client = embedding_client_manager.client
    async def test():
        # 使用示例文本验证 Embedding 服务是否可正常响应
        text = "What is deep learning?"
        query_result = await client.aembed_query(text)
        # 只打印前 3 个维度，便于快速确认返回结果结构正确
        print(query_result[:3])
    # 运行调试测试
    asyncio.run(test())

1. 初始化和释放客户端

if __name__ == "__main__":
    embedding_client_manager.init()
    client = embedding_client_manager.client
    asyncio.run(test())

初始化，获取客户端，运行测试。

2. 创建测试函数

async def test():
    text = "What is deep learning?"
    query_result = await client.aembed_query(text)
    print(query_result[:3])

这个函数测试 Embedding 服务是否可正常响应：

• client.aembed_query：aembed_查询 What is deep learning?
• query_result[:3]：只打印前 3 个维度

3. 测试

文件测试：

python -m app.clients.embedding_client_manager

获得结果：

[-0.005919582676142454, 0.005813124123960733, 0.018300632014870644]

MySQL 客户端

MySQL 简介

MySQL：在这里是两个负责不同职责的数据库.

MySQL 概念

• Base：是所有 ORM 模型共同继承的基类
• User、Address：是程序里的类，同时也分别对应数据库里的两张表
• __tablename__：指定这个类映射到数据库中的哪张表
• mapped_column(...)：定义字段类型、主键等信息
• relationship(...)：用来描述表和表之间的关联关系
• Engine：不是“某次查询”，更像数据库连接层的核心对象，底层会维护连接池
• Session：往数据库写数据时，真正和数据库交互

封装 MySQL 客户端

项目对应文件路径：shopkeeper-agent/app/clients/mysql_client_manager.py

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

1. 创建客户端管理器

class MySQLClientManager:
    def __init__(self, config: DBConfig):
        self.engine: AsyncEngine | None = None
        self.session_factory = None
        self.config = config
    def _get_url(self):
        return f"mysql+asyncmy://{self.config.user}:{self.config.password}@{self.config.host}:{self.config.port}/{self.config.database}?charset=utf8mb4"
    def init(self):
        self.engine = create_async_engine(
            self._get_url(), pool_size=10, pool_pre_ping=True
        )
        self.session_factory = async_sessionmaker(
            self.engine, autoflush=True, expire_on_commit=False
        )
    async def close(self):
        await self.engine.dispose()

• __init__()：准备创建 engine 维护连接池，然后准备创建 session_factory 用来按需创建新的 AsyncSession，保存数据库配置
• _get_url()：mysql+asyncmy 表示：连接 MySQL，并使用 asyncmy 作为异步驱动
• init()：创建异步 Engine，然后基于 Engine 创建 Session 工厂，后面真正查库时再拿 session
• close()：程序结束时释放连接池资源

2. 创建全局管理器对象

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

一套连元数据库，一套连数仓模拟库。

测试 Embedding 客户端

完整 main 代码：

if __name__ == "__main__":
    # 这里演示的是数仓库查询，所以先初始化 dw 这一套客户端
    dw_mysql_client_manager.init()
    async def test():
        # 通过 session_factory 创建一次数据库会话
        async with dw_mysql_client_manager.session_factory() as session:
            sql = "select * from fact_order limit 10"
            # text(sql) 表示把原生 SQL 语句交给 SQLAlchemy 执行
            result = await session.execute(text(sql))
            # mappings().fetchall() 会把结果转成“按列名访问”的行对象列表
            rows = result.mappings().fetchall()
            # 下面三行只是为了帮助观察返回结果的结构
            print(type(rows))
            print(type(rows[0]))
            print(rows[0]["order_id"])
    asyncio.run(test())

1. 初始化和释放客户端

if __name__ == "__main__":
    dw_mysql_client_manager.init()
    asyncio.run(test())

这里演示的是数仓库查询，所以先初始化 dw 这一套客户端。

2. 创建测试函数

async def test():
    async with dw_mysql_client_manager.session_factory() as session:
        sql = "select * from fact_order limit 10"
        result = await session.execute(text(sql))
        rows = result.mappings().fetchall()
        print(type(rows))
        print(type(rows[0]))
        print(rows[0]["order_id"])

分四步进行一次数据库对话：

• async with：打开 session_factory
• text(sql)：把原生 SQL 语句交给 SQLAlchemy 执行
• mappings().fetchall()：把结果转成“按列名访问”的行对象列表
• print()：返回结果

3. 测试

执行文件：

python -m app.clients.mysql_client_manager

输出结果：

<class 'list'>
<class 'sqlalchemy.engine.row.RowMapping'>
ORD20250101001

Loguru 日志配置

Loguru 简介

Loguru：这里是一套相对于 print() 更稳定、更可配置的日志方案。 不用 print 原因主要有两类：

• 输出信息不够规范：如果只用 print，你通常还得自己拼接时间、模块名、函数名、行号、日志级别等信息。
• 输出目的地不够灵活：项目上线后，日志往往不只是输出到控制台，还需要同时写入日志文件，方便排查和归档。

配置文件

项目对应文件路径：shopkeeper-agent/conf/app_config.yaml

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

输出目的地：

• console：输出到控制
• file：输出到日志文件

属性：

• enable：决定这个输出通道是否启用。
• level：设定这个输出通道的最低日志级别
• path：指定日志文件目录
• rotation：指定多大体积后切分新文件
• retention：指定日志保留多久
• rotation: "10 MB"：表示日志文件写到大约 10MB 时，自动切分成新文件
• retention: "7 days"：表示只保留最近 7 天的日志

两个参数方便清理数据，解决了日志文件长期运行时的可维护性问题。

封装日志配置

项目对应文件路径：shopkeeper-agent/app/core/log.py

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

1. 定义统一日志格式

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>"
)

统一日志展示格式，包含时间、级别、请求 ID 和调用位置等关键信息。

2. 给每条日志自动补充 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)

• inject_request_id(record)：通过 Loguru patch 钩子，把上下文中的 request_id 写入每条日志的 extra 字段
• logger.remove()：移除 Loguru 默认的输出目标，避免和项目自定义配置重复打印
• logger.patch(inject_request_id)：生成带 request_id 注入能力的 logger，后续业务代码统一使用这个实例

3. 决定日志输出到控制台还是文件

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:
    path = Path(app_config.logging.file.path)
    path.mkdir(parents=True, exist_ok=True)
    logger.add(
        sink=path / "app.log",
        level=app_config.logging.file.level,
        format=log_format,
        rotation=app_config.logging.file.rotation,
        retention=app_config.logging.file.retention,
        encoding="utf-8",
    )

根据配置文件启用或关闭不同输出通道：

• 根据配置决定是否输出控制台日志，适合本地开发和容器标准输出采集
• 根据配置决定是否写入文件日志，并在启动时确保日志目录存在

结语

我正处在对某些 Agent 项目的学习阶段，

会做笔记，会写总结，会反复实践，

但还是时常反思自己的学习方式，

最近是不是有点急于求成？

这篇文章有点头重脚轻的。