From f6302e89b5be7e7a51f535fe10dd5f38c0b5cd10 Mon Sep 17 00:00:00 2001 From: Marcus <1922576605@qq.com> Date: Wed, 30 Sep 2026 09:00:29 -0700 Subject: [PATCH] =?UTF-8?q?feat=EF=BC=9A=E7=89=A9=E6=96=99=E5=AD=98?= =?UTF-8?q?=E5=85=A5=E6=95=B0=E6=8D=AE=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/materials.py | 19 ++++++++++ app/main.py | 11 ++++-- app/resources/component.py | 59 +++++++++++++++++++------------ app/services/material_mongo.py | 63 ++++++++++++++++++++++++++++++++++ app/services/persistence.py | 5 ++- 5 files changed, 131 insertions(+), 26 deletions(-) create mode 100644 app/api/materials.py create mode 100644 app/services/material_mongo.py diff --git a/app/api/materials.py b/app/api/materials.py new file mode 100644 index 0000000..84e3f97 --- /dev/null +++ b/app/api/materials.py @@ -0,0 +1,19 @@ +"""前端物料目录接口""" + +from fastapi import APIRouter + +from app.common.response import ApiResponse, success_response +from app.core.exceptions import BizException +from app.resources.component import get_catalog_cache +from app.schemas.component_resource import ComponentCatalog + +router = APIRouter(prefix="/materials", tags=["Material"]) + + +@router.get("") +async def list_materials() -> ApiResponse[ComponentCatalog]: + # 直接返回启动时载入的内存缓存，不查库 + catalog = get_catalog_cache() + if catalog is None: + raise BizException(503, "物料目录尚未加载") + return success_response(catalog) diff --git a/app/main.py b/app/main.py index abe679a..7b6c98f 100644 --- a/app/main.py +++ b/app/main.py @@ -8,6 +8,7 @@ from loguru import logger from app.api.agent import router from app.api.conversations import router as conversation_router +from app.api.materials import router as material_router from app.core.exceptions import register_exception_handlers from app.core.logging import setup_logging @@ -19,15 +20,18 @@ async def lifespan(application: FastAPI): from app.observability.tracing import flush_tracing from app.processor.graph import build_graph from app.services.agent import AgentService + from app.services.material_mongo import warm_material_cache from app.services.persistence import open_persistence # 按配置选择 MongoDB 或进程内存，会话记录与图检查点使用同一种存储 persistence = await open_persistence() application.state.conversation_store = persistence.store - application.state.agent_service = AgentService( - build_graph(persistence.checkpointer), persistence.store - ) try: + # 物料目录启动时载入内存，agent 规划与前端接口共用，避免重复查询数据库 + await warm_material_cache(persistence.db) + application.state.agent_service = AgentService( + build_graph(persistence.checkpointer), persistence.store + ) yield finally: # 共享客户端只在应用退出时关闭，追踪刷新放到线程避免阻塞事件循环 @@ -50,6 +54,7 @@ def create_app() -> FastAPI: # 统一 /api 前缀，方便前端代理和反向代理只配置一条规则 application.include_router(router, prefix="/api") application.include_router(conversation_router, prefix="/api") + application.include_router(material_router, prefix="/api") return application diff --git a/app/resources/component.py b/app/resources/component.py index a8c00a4..da2b7c5 100644 --- a/app/resources/component.py +++ b/app/resources/component.py @@ -1,4 +1,4 @@ -"""读取前端组件目录""" +"""读取前端组件目录：启动时载入进程内存缓存，未载入时回退读取本地 JSON 文件""" from pathlib import Path @@ -8,38 +8,53 @@ from app.schemas.component_resource import ( ComponentResult, ) +DEFAULT_CATALOG_PATH = Path(__file__).resolve().parent / "catalog" / "components.json" + +# 进程内共享的组件目录，由启动流程载入；agent 与前端接口读取同一份 +_cached_catalog: ComponentCatalog | None = None + + +def set_catalog_cache(catalog: ComponentCatalog | None) -> None: + global _cached_catalog + _cached_catalog = catalog + + +def get_catalog_cache() -> ComponentCatalog | None: + return _cached_catalog + + +def load_catalog_file(path: str | Path | None = None) -> ComponentCatalog | None: + # 如果传入的是相对路径，以项目根目录为基准，避免受启动时工作目录影响 + target = Path(path) if path is not None else DEFAULT_CATALOG_PATH + if not target.is_absolute(): + target = Path(__file__).resolve().parents[2] / target + try: + # utf-8-sig 同时兼容带 BOM 和不带 BOM 的 UTF-8 文件 + content = target.read_text(encoding="utf-8-sig") + except FileNotFoundError: + return None + # 文件损坏或定义不合法直接报错，不能伪装成查无结果 + return ComponentCatalog.model_validate_json(content) + class ComponentProvider: def __init__(self, catalog_path: str | Path | None = None): - # 保存可选的目录文件路径，未指定时由查询方法使用项目内的默认目录 + # 指定路径时始终读取该文件；未指定时优先使用启动缓存，缓存为空再读默认文件 self.catalog_path = catalog_path def query(self, query: ComponentQuery) -> ComponentResult: - # TODO 以后迁移数据库，保持组件查询接口不变 - # 优先使用调用方指定的路径，否则读取当前模块旁的 catalog/components.json - path = ( - Path(self.catalog_path) - if self.catalog_path is not None - else Path(__file__).resolve().parent / "catalog" / "components.json" + catalog = ( + _cached_catalog + if self.catalog_path is None and _cached_catalog is not None + else load_catalog_file(self.catalog_path) ) - - # 如果传入的是相对路径，以项目根目录为基准，避免受启动时工作目录影响 - if not path.is_absolute(): - path = Path(__file__).resolve().parents[2] / path - - try: - # 每次读取当前文件，避免更新目录后仍使用旧缓存 - # utf-8-sig 同时兼容带 BOM 和不带 BOM 的 UTF-8 文件 - content = path.read_text(encoding="utf-8-sig") - except FileNotFoundError: + if catalog is None: # 如果目录文件不存在，返回未支持状态和原因，由调用方决定失败处理方式 return ComponentResult( status="unsupported", message="尚未配置前端组件目录，请先导出真实组件定义", ) - # 文件损坏或定义不合法直接报错，不能伪装成查无结果 - catalog = ComponentCatalog.model_validate_json(content) # 如果提供关键词，去除首尾空白并统一大小写，未提供时不启用关键词筛选 keyword = ( query.keyword.strip().casefold() if query.keyword is not None else None @@ -67,8 +82,8 @@ class ComponentProvider: # 如果设置了关键词，但检索文本中不包含该关键词，则跳过当前组件 if keyword is not None and keyword not in searchable: continue - # 所有筛选条件通过，保留完整组件定义供后续规划和动作生成使用 - matches.append(component) + # 深拷贝后返回，调用方修改结果不会污染共享缓存 + matches.append(component.model_copy(deep=True)) # 根据是否有匹配项区分查询成功和未找到，同时保留目录版本、来源及结果说明 return ComponentResult( diff --git a/app/services/material_mongo.py b/app/services/material_mongo.py new file mode 100644 index 0000000..790b49b --- /dev/null +++ b/app/services/material_mongo.py @@ -0,0 +1,63 @@ +"""物料目录的 MongoDB 存储：每个组件一条记录，目录版本和来源放在元数据记录中""" + +from loguru import logger +from pymongo.asynchronous.database import AsyncDatabase + +from app.resources.component import load_catalog_file, set_catalog_cache +from app.schemas.component_resource import ComponentCatalog + +# 元数据记录的 _id，组件记录的 _id 为组件 key +META_ID = "__catalog__" + + +class MongoMaterialRepository: + def __init__(self, db: AsyncDatabase, prefix: str = ""): + # 集合名前缀用于测试隔离 + self._materials = db.get_collection(f"{prefix}materials") + + async def load_all(self) -> ComponentCatalog | None: + # 没有元数据记录视为尚未导入，由调用方决定回退方式 + meta = await self._materials.find_one({"_id": META_ID}) + if meta is None: + return None + documents = await self._materials.find({"_id": {"$ne": META_ID}}).to_list() + components = [ + {key: value for key, value in item.items() if key not in ("_id", "order")} + for item in sorted(documents, key=lambda item: item.get("order", 0)) + ] + return ComponentCatalog.model_validate( + { + "version": meta["version"], + "source": meta["source"], + "components": components, + } + ) + + async def replace_all(self, catalog: ComponentCatalog) -> None: + # 整体替换目录，供导入脚本使用；order 保持导出文件里的组件顺序 + documents = [ + { + **component.model_dump(mode="json", by_alias=True), + "_id": component.key, + "order": index, + } + for index, component in enumerate(catalog.components) + ] + await self._materials.delete_many({}) + if documents: + await self._materials.insert_many(documents) + await self._materials.insert_one( + {"_id": META_ID, "version": catalog.version, "source": catalog.source} + ) + + +async def warm_material_cache(db: AsyncDatabase | None) -> ComponentCatalog | None: + """启动时把物料目录载入进程缓存；Mongo 未启用或尚未导入时回退读本地 JSON""" + catalog = await MongoMaterialRepository(db).load_all() if db is not None else None + if catalog is None: + catalog = load_catalog_file() + logger.warning("物料未从 MongoDB 加载，回退使用本地目录文件") + else: + logger.info("已从 MongoDB 加载物料 {} 个", len(catalog.components)) + set_catalog_cache(catalog) + return catalog diff --git a/app/services/persistence.py b/app/services/persistence.py index 04b411b..77902bf 100644 --- a/app/services/persistence.py +++ b/app/services/persistence.py @@ -2,6 +2,7 @@ from collections.abc import Awaitable, Callable from dataclasses import dataclass +from typing import Any from langgraph.checkpoint.base import BaseCheckpointSaver from loguru import logger @@ -16,6 +17,8 @@ class Persistence: # 为空时由图使用默认的内存检查点 checkpointer: BaseCheckpointSaver | None close: Callable[[], Awaitable[None]] + # 启用 MongoDB 时的数据库，供物料等其他集合复用同一连接 + db: Any = None async def _close_nothing() -> None: @@ -67,4 +70,4 @@ async def open_persistence() -> Persistence: raise logger.info("已启用 MongoDB 持久化，库名 {}", settings.mongodb_db) - return Persistence(store, checkpointer, close) + return Persistence(store, checkpointer, close, async_client[settings.mongodb_db])