09 | 重构项目结构

09 | 重构项目结构

本文目标

前面三篇已经完成了生成阶段的元数据同步工作：

1. 把表、字段、指标和它们之间的关联关系写入 MySQL meta 库
2. 把字段和指标的向量写入 Qdrant
3. 把低基数的维度值写入 Elasticsearch

等功能逐渐完善了，原先的目录结构就开始暴露出一些问题。这些问题其实挺典型的，比如说：

• scripts/rebuild_metadata.py 这个脚本，同时包揽了配置校验、数据读取、实体构造、向量生成、数据库写入和客户端关闭——几乎一个人干了所有活。
• 多个 Service 层，里面只有一行 Repository 的转发调用，基本没什么真正的业务逻辑。
• Elasticsearch 的 Mapping 和 Bulk 细节，直接写在 Service 里，应用层知道了太多底层的存储实现。
• FastAPI 的路由、请求解析和应用创建，全都挤在根目录的 main.py 里。
• LangGraph 的 State、节点函数、Graph 声明和演示代码，全在一个文件里混着。
• 应用配置用的是多层 dict，配置字段拼错了，只能等到运行时才能发现。
• 更关键的是，没有自动化测试，每次重构以后，只能手动跑一次破坏性的全量同步来验证。

所以，这次的核心任务不是增加新功能，而是完成一次彻底的结构重构。重构完成后，已有功能完全不变，启动方式也保持兼容。

---

1. 为什么现在需要重构

项目刚起步的时候，代码量少，把逻辑直接往脚本里堆是最快的路子：

 复制代码读取 YAML
→ 查询 DW
→ 写 MySQL
→ 生成向量
→ 写 Qdrant
→ 写 Elasticsearch

这时候如果过早地设计一堆目录，反而会拖慢开发速度。

但当同步目标增加到三个之后，一个脚本需要知道的事情就太多了：

• MySQL 的 Engine、Session 和事务管理
• SQLAlchemy 的 Model、Mapper、Repository
• TEI 的地址、超时和批次大小
• Qdrant 的 Collection 名称和向量维度
• Elasticsearch 的 Index、Mapping、Bulk 和 Refresh
• 三种存储的写入顺序和失败边界

继续往这个脚本里塞功能，比如加入召回、日志、监控，它就会变成一个越来越难维护的“大泥球”。

所以，判断要不要重构，不是看文件行数有没有超过某个固定的数字，而是看它是否出现了下面这些信号：

1. 一个文件需要知道太多不同技术的细节
2. 修改一种存储，会影响到与它无关的代码
3. 同一段初始化或资源关闭的逻辑，在多个入口重复出现
4. 业务流程只能通过阅读大量底层实现才能理解
5. 很难在不连接真实数据库的情况下进行测试

当前项目已经出现了这些信号，所以，是时候动手了。

2. 重构前后的目录对比

先看看重构前的核心目录长什么样：

 复制代码app/
├── agent/
│   └── graph.py
├── entities/
├── infrastructure/
├── mappers/
├── models/
├── repositories/
└── services/
    ├── table_info_service.py
    ├── column_info_service.py
    ├── metric_info_service.py
    ├── column_metric_service.py
    ├── dw_db_service.py
    ├── vector_sync_service.py
    └── dim_value_sync_service.py
conf/
└── sync_db.py
main.py

重构之后，变成这样：

 复制代码app/
├── main.py
├── api/
│   ├── routes/
│   │   └── query.py
│   └── schemas/
│       └── query.py
├── agent/
│   ├── state.py
│   ├── nodes.py
│   └── graph.py
├── application/
│   ├── metadata_rebuild_service.py
│   ├── vector_sync_service.py
│   └── dim_value_sync_service.py
├── config/
│   └── settings.py
├── entities/
├── infrastructure/
├── mappers/
├── models/
└── repositories/
    ├── dim_value_repository.py
    ├── dw_db_repository.py
    ├── qdrant_repository.py
    └── ...
conf/
├── app_config.yaml
├── get_config.py
└── meta_config.yaml
scripts/
└── rebuild_metadata.py
tests/
└── unit/

每层的主要职责整理了一下：

层	目录	主要职责
API	app/api/	HTTP 请求解析、响应格式、路由
Agent	app/agent/	LangGraph 状态、节点和工作流
Application	app/application/	编排一个完整业务用例
Entity	app/entities/	与具体存储无关的数据对象
Repository	app/repositories/	表达数据读取和写入语义
Mapper	app/mappers/	Entity 与 ORM Model 转换
Model	app/models/	SQLAlchemy 表映射
Infrastructure	app/infrastructure/	外部 SDK、连接池和 HTTP Client
Config	app/config/、conf/	配置结构和 YAML 配置内容
Script	scripts/	组装依赖并启动应用用例
Test	tests/	自动验证行为

这里采用的是轻量分层架构，并没有追求把每个概念都拆成独立的目录。

---

3. 第一步：增加 Application 层

旧的 scripts/rebuild_metadata.py 直接编排了整个流程。重构之后，把真正的业务用例放到了这里：

 复制代码app/application/metadata_rebuild_service.py

核心类长这样：

 复制代码class MetadataRebuildService:
    """准备并重建 MySQL、Qdrant 和 Elasticsearch 元数据。"""
    def __init__(
        self,
        dw_database: MySQLDatabase,
        meta_database: MySQLDatabase,
        vector_sync_service: VectorSyncService,
        dim_value_sync_service: DimValueSyncService,
        column_collection: str,
        metric_collection: str,
    ):
        self.dw_database = dw_database
        self.meta_database = meta_database
        self.vector_sync_service = vector_sync_service
        self.dim_value_sync_service = dim_value_sync_service
        self.column_collection = column_collection
        self.metric_collection = metric_collection

注意，它不自己创建 MySQL、TEI、Qdrant 和 ES 客户端，而是通过构造参数接收这些能力。

这样有两个好处：

1. Application Service 只关心流程，不关心客户端怎么初始化
2. 单元测试可以传入 Fake 对象，不需要真的启动 Docker 服务

完整的重建入口变成了这样：

 复制代码async def rebuild(self, config: dict[str, Any]) -> MetadataRebuildResult:
    prepared = await self.prepare(config)
    await self._write_meta_database(prepared)
    column_vector_count = await self.vector_sync_service.replace_collection(
        self.column_collection,
        prepared.column_points,
    )
    metric_vector_count = await self.vector_sync_service.replace_collection(
        self.metric_collection,
        prepared.metric_points,
    )
    dim_value_count = await self.dim_value_sync_service.replace_index(
        prepared.dim_value_documents
    )
    return MetadataRebuildResult(...)

光看这个方法，就能清楚理解整个业务流程：

 复制代码准备全部数据
→ 写 MySQL
→ 写 Qdrant 字段向量
→ 写 Qdrant 指标向量
→ 写 Elasticsearch 维度值
→ 返回每部分数量

这就是 Application Service 的价值所在。

4. 第二步：同步过程拆成准备和写入两个阶段

全量重建会删除旧表、旧 Collection 或旧 Index。如果读取 DW 或者生成向量的时候中途失败了，可不能先把已有的数据清空。

所以，重建过程拆成了两个阶段。

4.1 准备阶段

 复制代码async def prepare(self, config: dict[str, Any]) -> PreparedMetadata:
    self.validate_config(config)
    table_infos = self._build_table_infos(config)
    metric_infos = self._build_metric_infos(config)
    column_metrics = self._build_column_metrics(config)
    column_infos, dim_value_documents = await self._read_dw_metadata(config)
    column_points = await self.vector_sync_service.prepare_columns(
        self.column_collection,
        column_infos,
    )
    metric_points = await self.vector_sync_service.prepare_metrics(
        self.metric_collection,
        metric_infos,
    )
    return PreparedMetadata(...)

准备阶段只做这些事：

• 校验 meta_config
• 构造 Entity
• 读取 DW 字段类型、样例和维度值
• 调用 TEI 生成向量
• 校验向量维度

这个阶段不会修改 MySQL meta、Qdrant 和 Elasticsearch 里的数据。

4.2 写入阶段

等所有准备操作都成功了，才开始重建目标存储：

 复制代码PreparedMetadata
       │
       ├─ 写 MySQL meta
       ├─ 替换 Qdrant Collections
       └─ 替换 Elasticsearch Index

这样虽然不能实现三个数据库之间的真正原子事务，但可以避免大量“前置计算失败了，旧数据却已经提前丢失”的尴尬情况。

如果生产环境要求切换过程完全无感，那还需要：

• 临时 MySQL 表
• 临时 Qdrant Collection
• 临时 Elasticsearch Index
• 校验完成后，通过重命名或 Alias 来切换

当前项目还处于生成和学习阶段，先用“准备成功后再重建”的方案，已经够用了。

5. 第三步：合并只有透传作用的 Service

旧代码里存在这样的 Service：

 复制代码class ColumnInfoService:
    def __init__(self, repository: ColumnInfoRepository):
        self.repository = repository
    async def add_all(self, column_infos):
        return await self.repository.add_all(column_infos)

它没有校验、组合、事务或业务判断，只是把调用原样转交给 Repository。

调用链反而变得更长了：

 复制代码Script
→ ColumnInfoService
→ ColumnInfoRepository
→ Mapper
→ Model

删除这些纯透传的 Service 之后，调用链变成了：

 复制代码async with self.meta_database.session() as session, session.begin():
    await TableInfoRepository(session).add_all(prepared.table_infos)
    await ColumnInfoRepository(session).add_all(prepared.column_infos)
    await MetricInfoRepository(session).add_all(prepared.metric_infos)
    await ColumnMetricRepository(session).add_all(prepared.column_metrics)

现在调用链是：

 复制代码MetadataRebuildService
→ Repository
→ Mapper
→ Model

Application Service 负责业务流程，Repository 负责持久化，这两层职责已经足够了。

这里要记住一个原则：只有当一层确实包含独立职责的时候，它才值得存在。

6. 第四步：统一 Repository 的职责

重构之前，情况是这样的：

• Qdrant 的 Collection 和 Point 操作，在 QdrantRepository 里
• Elasticsearch 的 Mapping、Bulk、Refresh，却都在 DimValueSyncService 里

抽象标准不一致，这显然不好。

重构后，新增了：

 复制代码app/repositories/dim_value_repository.py

现在 Repository 负责 ES 的存储细节：

 复制代码class DimValueRepository:
    async def reset_index(self) -> None:
        ...
    async def index_values(self, documents) -> int:
        ...
    async def refresh(self) -> None:
        ...

Application Service 只负责调用顺序：

 复制代码class DimValueSyncService:
    async def replace_index(self, documents) -> int:
        await self.repository.reset_index()
        total = await self.repository.index_values(documents)
        await self.repository.refresh()
        return total

现在职责划分就更加一致了：

 复制代码DimValueSyncService
└─ 编排：reset → index → refresh
DimValueRepository
└─ 实现：Index Mapping、Bulk 格式、稳定 UUID、错误解析
ESClient
└─ 连接：官方异步 SDK、地址、超时

7. 第五步：让 Script 只负责启动

重构前的同步脚本超过 400 行，包含了大量业务实现。

重构后的 scripts/rebuild_metadata.py，只负责三件事：

1. 根据配置创建客户端和 Database
2. 把依赖组装成 MetadataRebuildService
3. 执行、打印结果并关闭资源

核心入口：

 复制代码async def main() -> None:
    dw_database = MySQLDatabase(...)
    meta_database = MySQLDatabase(...)
    embedding_client = EmbeddingClient(...)
    qdrant_client = QdrantClient(...)
    es_client = ESClient(...)
    rebuild_service = MetadataRebuildService(...)
    try:
        result = await rebuild_service.rebuild(meta_config)
        print_result(result)
    finally:
        await asyncio.gather(
            dw_database.close(),
            meta_database.close(),
            embedding_client.close(),
            qdrant_client.close(),
            es_client.close(),
        )

Script 仍然需要知道如何创建具体依赖，因为它是这个命令的组合根。

但它不再负责：

• 怎样校验元数据
• 怎样查询 DW
• 怎样生成 Qdrant Point
• 怎样组织 ES Bulk 请求
• 怎样控制业务写入顺序

运行方式：

 复制代码uv run python scripts/rebuild_metadata.py

8. 第六步：拆分 FastAPI API 层

原来的根目录 main.py 同时包含了：

• FastAPI 创建
• CORS 配置
• 健康检查
• SSE 生成器
• 查询路由
• 请求 dict 解析

重构后：

 复制代码app/
├── main.py
└── api/
    ├── routes/query.py
    └── schemas/query.py

请求模型：

 复制代码class QueryRequest(BaseModel):
    query: str = Field(
        min_length=1,
        description="需要转换为 SQL 的自然语言问题",
    )

这样空字符串会自动返回 HTTP 422，而不需要在路由里手动判断。

查询路由：

 复制代码router = APIRouter(prefix="/api", tags=["query"])
@router.post("/query", response_class=StreamingResponse)
async def query(payload: QueryRequest) -> StreamingResponse:
    return StreamingResponse(
        sse_stream(payload.query),
        media_type="text/event-stream",
    )

应用工厂：

 复制代码def create_app() -> FastAPI:
    application = FastAPI(title="n2sql-agent")
    application.add_middleware(...)
    application.include_router(query_router)
    return application
app = create_app()

推荐启动方式：

 复制代码uv run fastapi dev app/main.py

根目录的 main.py 仍然保留了兼容导入，所以旧的命令也能继续使用：

 复制代码uv run fastapi dev main.py

9. 第七步：拆分 Agent

原来的 app/agent/graph.py 同时包含了 State、节点函数、Graph 声明和演示代码。

重构后：

 复制代码app/agent/
├── state.py   # State、RuntimeContext、进度事件
├── nodes.py   # 工作流节点
└── graph.py   # Graph 拓扑和路由

9.1 state.py

 复制代码class State(TypedDict, total=False):
    query: str
    keywords: list[str]
    error: str | None

total=False 表示每个节点不需要返回全部字段，只返回自己修改的那部分就可以了。

比如关键词节点只返回：

 复制代码return {"keywords": keywords}

9.2 nodes.py

这里放节点的具体行为：

 复制代码extract_keywords
recall_column
recall_metric
recall_value
generate_sql
validate_sql
run_sql
...

当前未实现的节点，共用一个占位辅助函数，避免重复写进度推送的代码。

9.3 graph.py

这里只声明节点之间怎么连接：

 复制代码def route_after_validation(state: State) -> str:
    return "run_sql" if state.get("error") is None else "correct_sql"
def create_graph():
    return (
        StateGraph(...)
        .add_node(extract_keywords)
        .add_node(recall_column)
        ...
        .add_conditional_edges(...)
        .compile()
    )

以后查看工作流结构时，不需要再穿过每个节点的实现细节。

10. 第八步：把应用配置改成强类型

旧代码用的是多层字典：

 复制代码app_config["qdrant"]["embedding_size"]
app_config["embedding"]["batch_size"]

如果写成：

 复制代码app_config["qdrant"]["embeding_size"]

只能运行到这一行的时候，才发现 KeyError。

重构后，在 app/config/settings.py 定义了 Pydantic Model：

 复制代码class QdrantSettings(StrictSettings):
    host: str
    port: int
    embedding_size: int = Field(gt=0)
    column_collection: str
    metric_collection: str
    timeout: float = Field(default=60, gt=0)

总配置：

 复制代码class AppSettings(StrictSettings):
    logging: LoggingSettings
    meta_db: DatabaseSettings
    dw_db: DatabaseSettings
    qdrant: QdrantSettings
    embedding: EmbeddingSettings
    es: ElasticsearchSettings
    llm: LLMSettings

配置加载后，可以用属性访问：

 复制代码app_config.qdrant.embedding_size
app_config.embedding.batch_size
app_config.es.index_name

extra="forbid" 会拒绝没有声明的字段，所以 YAML 的 key 拼错了，程序会在启动阶段直接报错，不会等到运行时才发现。

需要注意：

• conf/app_config.yaml 仍然保存配置值
• app/config/settings.py 描述配置结构和约束
• .env 仍然保存密码、API Key 等敏感值

三者职责不同，互不冲突。

11. 第九步：补充自动化测试

重构最危险的地方在于：目录看起来更整齐了，但业务行为可能已经变了。

因此，新增了：

 复制代码tests/unit/
├── test_api.py
├── test_dim_value_repository.py
├── test_dim_value_sync_service.py
├── test_metadata_rebuild_service.py
└── test_vector_sync_service.py

重点验证几个方面：

11.1 向量准备阶段不写 Qdrant

 复制代码points = await service.prepare_columns("columns", columns)
self.assertEqual(repository.calls, [])

只有执行了：

 复制代码await service.replace_collection("columns", points)

才允许调用 reset_collection 和 upsert。

11.2 ES 写入顺序

 复制代码reset_index
→ index_values
→ refresh

11.3 配置引用校验

指标不能引用不存在的字段，只有 dimension 字段才能配置 sync: true。

11.4 API 参数校验

空查询应返回 HTTP 422，健康检查应返回 200。

运行测试：

 复制代码uv run python -m unittest discover -s tests -v

当前结果：

 复制代码Ran 8 tests
OK

12. 重构后的依赖方向

生成阶段的完整依赖关系：

 复制代码scripts/rebuild_metadata.py
          │
          ▼
MetadataRebuildService             Application
          │
          ├─ VectorSyncService      Application
          ├─ DimValueSyncService    Application
          │
          ▼
Repository                         Data access
          │
          ├─ MySQL Repository
          ├─ QdrantRepository
          └─ DimValueRepository
          │
          ▼
Infrastructure Client              External technology
          │
          ├─ MySQLDatabase
          ├─ EmbeddingClient
          ├─ QdrantClient
          └─ ESClient

依赖方向应该从业务流程指向外部实现：

 复制代码入口 → 应用用例 → 数据访问 → 外部 SDK

而不是让 Repository 反过来 import API，或者让 Entity import Elasticsearch Client。

13. 哪些目录没有继续拆

这次没有把项目改造成特别重的 DDD 结构。

保留了：

 复制代码app/entities/
app/models/
app/mappers/
app/repositories/

没有继续增加：

 复制代码domain/ports/
domain/repositories/
infrastructure/adapters/
application/commands/
application/handlers/

原因是当前项目规模还不需要这么多抽象。

举个例子，Repository 目前只有一种实现，没有必要先定义一个完全相同的抽象接口：

 复制代码class AbstractColumnInfoRepository(Protocol):
    async def add_all(...): ...

等到真的出现以下需求的时候，再增加接口也不迟：

• MySQL 与 PostgreSQL 两种实现需要共存
• 生产实现与内存实现需要互换
• 多个业务用例依赖同一套稳定抽象

架构设计应该解决已经出现或很快会出现的真实问题，而不是试图预测所有可能性。

14. 验证重构结果

14.1 编译检查

 复制代码uv run python -m compileall -q app conf scripts tests main.py

14.2 单元测试

 复制代码uv run python -m unittest discover -s tests -v

14.3 启动 API

 复制代码uv run fastapi dev app/main.py

14.4 完整重建

确认 MySQL、TEI、Qdrant 和 Elasticsearch 已经启动后，执行：

 复制代码uv run python scripts/rebuild_metadata.py

注意，这个命令会重建开发环境中的：

• MySQL meta 表
• Qdrant 字段和指标 Collection
• Elasticsearch 维度值 Index

不要直接把当前“删除后重建”的实现用于生产环境。

---

科普项目架构

1. 什么是项目架构

项目架构不是目录树本身，也不是文件夹名字听起来是否“高级”。

架构描述的是：

• 系统由哪些部分组成
• 每部分负责什么
• 各部分怎样通信
• 谁可以依赖谁
• 改动一个部分时，会影响多少其他部分

目录只是架构的一种可见表达。

举个例子，下面两个项目都叫 services，但内部的职责可能完全不同：

 复制代码项目 A：Service 负责完整业务用例
项目 B：Service 只转发 Repository

不能只通过目录名来判断架构是否合理，要深入看到实际的依赖和行为。

2. 什么是分层架构

分层架构就是把不同职责放在不同层：

 复制代码API 层
  ↓
Application 层
  ↓
Repository 层
  ↓
Infrastructure 层

一种常见的理解方式：

• API：外界怎样调用系统
• Application：系统要完成什么用例
• Repository：数据怎样读取和保存
• Infrastructure：具体使用哪个数据库、SDK 或 HTTP 服务

分层的核心不是“所有请求必须经过固定数量的文件”，而是：

3. 什么是关注点分离

关注点分离，英文是 Separation of Concerns。

它表示不同类型的问题，应该由不同的代码来处理。

例如：

 复制代码QueryRequest
→ 负责 HTTP 输入校验
MetadataRebuildService
→ 负责同步流程
DimValueRepository
→ 负责 ES Bulk 格式
ESClient
→ 负责连接 Elasticsearch

如果一个类同时处理 HTTP、业务规则、SQL 和日志文件轮转，那就混合了太多不同的关注点。

4. 什么是依赖注入

依赖注入不是某个框架的专属功能。

最简单的依赖注入，就是把对象从构造参数传进去：

 复制代码service = VectorSyncService(
    embedding_client=embedding_client,
    qdrant_repository=qdrant_repository,
    vector_size=1024,
    model_name="BAAI/bge-large-zh-v1.5",
)

而不是在 Service 内部写死：

 复制代码class VectorSyncService:
    def __init__(self):
        self.client = EmbeddingClient("http://localhost:8081")

通过外部传入依赖后：

• 生产环境可以传真实的 Client
• 测试环境可以传 Fake Client
• Service 不需要知道地址从哪里读取
• 客户端的生命周期可以由入口统一管理

5. 什么是组合根

组合根，英文是 Composition Root。

它是集中创建并连接各个对象的地方：

 复制代码创建 Database
创建 Client
创建 Repository
创建 Application Service
调用用例

本项目的命令行组合根是：

 复制代码scripts/rebuild_metadata.py

组合根可以依赖很多具体类，因为它的职责就是“把系统组装起来”。

业务类内部则不应该到处重复创建这些对象。

6. 什么是 Repository

Repository 把数据存取表达成业务可以理解的操作：

 复制代码get_all_column_types()
get_distinct_column_values()
reset_collection()
index_values()

它隐藏了具体实现：

• SQL 怎么写
• Qdrant PointStruct 怎么构造
• Elasticsearch Bulk 的 operations 怎么排列
• 稳定 UUID 怎么生成

Repository 不应该决定整个业务用例的执行顺序，也不应该随意提交上层的数据库事务。

本次重构删除了 TableInfoRepository.add() 中自行 commit() 的做法，统一由 Application 层来控制事务边界。

7. 什么是 Application Service

Application Service 表达一个完整用例，比如：

 复制代码重建全部元数据
同步字段向量
替换维度值索引
执行一次自然语言查询

它通常负责：

• 调用多个 Repository
• 控制步骤顺序
• 控制事务边界
• 组织输入和输出结果
• 处理跨组件的失败场景

它通常不负责：

• HTTP JSON 格式
• SQLAlchemy 字段声明
• Elasticsearch Mapping 细节
• 官方 SDK 初始化参数

8. 什么是重构

重构，是在尽量不改变外部行为的前提下，改善内部结构。

本次重构后，用户仍然可以这样启动：

 复制代码uv run fastapi dev main.py
uv run python scripts/rebuild_metadata.py

同步的数据结构、Collection 名称和 Index 名称也没有改变。

改变的是内部的职责划分和代码位置。

一次安全的重构通常包含以下步骤：

1. 先确认当前行为
2. 补测试或准备验证命令
3. 小步迁移
4. 更新所有 import
5. 执行测试和集成验证
6. 最后删除旧代码

9. 什么是过度设计

过度设计，就是为尚未出现的复杂度，提前增加大量不必要的结构。

常见的表现：

• 每张表都有 Interface、Base、Impl、Factory、Manager、Service
• 只有一种实现，却提前设计了很多可插拔的接口
• 一个十行功能，需要跳转七八个文件才能看懂
• 文件夹很多，但每层都只是参数透传

避免过度设计的方法：

1. 先看是否存在真实的变化点
2. 同一种模式重复出现两三次之后，再考虑抽象
3. 抽象之后，必须减少调用方需要知道的细节
4. 删除没有独立职责的层

“代码少”不一定简单，“目录多”也不一定专业。

10. 单元测试与集成测试有什么区别

单元测试只验证一个较小的单元，通常不连接真实的外部服务：

 复制代码FakeEmbeddingClient
FakeQdrantRepository
FakeDimValueRepository

优点：

• 速度快
• 结果稳定
• 失败原因明确
• 不会清空真实数据

集成测试则验证多个真实组件能否一起工作：

 复制代码SQLAlchemy → MySQL
Qdrant Client → Qdrant
ES Client → Elasticsearch
EmbeddingClient → TEI

它更接近真实环境，但速度慢，也需要准备和清理数据。

一个成熟的项目，通常两种测试都需要：

 复制代码大量快速的单元测试
+
少量关键的集成测试

11. 架构需要一直重构吗

不需要。

当前的结构已经能清楚地表达：

 复制代码API
Agent
Application
Repository
Infrastructure

下一步的重点，应该是继续实现真正的字段召回、指标召回、维度值召回和 SQL 生成，而不是继续搬动目录。

等到出现新的真实问题时，再考虑调整，比如：

• API 节点需要统一共享数据库客户端
• Repository 出现了第二种存储实现
• MetadataRebuildService 再次膨胀
• 需要生产级的 Alias 原子切换
• 单元测试之外，需要 Docker 集成测试

架构的目标，是帮助业务持续演进，而不是让项目永远处于“重构中”的状态。

本文小结

本文完成了项目的第一次结构性重构：

1. 增加 application 层，承载完整业务用例
2. 将元数据同步拆成准备与写入两个阶段
3. 删除没有独立职责的透传 Service
4. 将 Elasticsearch 存取细节迁入 Repository
5. 把重建脚本缩小为组合根
6. 拆分 FastAPI 路由和 Pydantic Schema
7. 拆分 LangGraph 的 State、Nodes 和 Graph
8. 使用 Pydantic 建立强类型应用配置
9. 增加单元测试，保护重构行为
10. 保留轻量结构，避免过度设计

重构完成后，核心原则归结起来其实就一句话：

 复制代码入口负责组装
Application 负责用例
Repository 负责存取
Infrastructure 负责连接
Entity 保持独立
测试保护行为

完成这一步后，项目就可以在更稳定的结构上，继续实现 NL2SQL 的运行阶段了。