feat:物料存入数据库

This commit is contained in:
2026-09-30 09:00:44 -07:00
parent 3e88b339a0
commit f6302e89b5
5 changed files with 131 additions and 26 deletions
+19
View File
@@ -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)
+6 -1
View File
@@ -8,6 +8,7 @@ from loguru import logger
from app.api.agent import router from app.api.agent import router
from app.api.conversations import router as conversation_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.exceptions import register_exception_handlers
from app.core.logging import setup_logging 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.observability.tracing import flush_tracing
from app.processor.graph import build_graph from app.processor.graph import build_graph
from app.services.agent import AgentService from app.services.agent import AgentService
from app.services.material_mongo import warm_material_cache
from app.services.persistence import open_persistence from app.services.persistence import open_persistence
# 按配置选择 MongoDB 或进程内存,会话记录与图检查点使用同一种存储 # 按配置选择 MongoDB 或进程内存,会话记录与图检查点使用同一种存储
persistence = await open_persistence() persistence = await open_persistence()
application.state.conversation_store = persistence.store application.state.conversation_store = persistence.store
try:
# 物料目录启动时载入内存,agent 规划与前端接口共用,避免重复查询数据库
await warm_material_cache(persistence.db)
application.state.agent_service = AgentService( application.state.agent_service = AgentService(
build_graph(persistence.checkpointer), persistence.store build_graph(persistence.checkpointer), persistence.store
) )
try:
yield yield
finally: finally:
# 共享客户端只在应用退出时关闭,追踪刷新放到线程避免阻塞事件循环 # 共享客户端只在应用退出时关闭,追踪刷新放到线程避免阻塞事件循环
@@ -50,6 +54,7 @@ def create_app() -> FastAPI:
# 统一 /api 前缀,方便前端代理和反向代理只配置一条规则 # 统一 /api 前缀,方便前端代理和反向代理只配置一条规则
application.include_router(router, prefix="/api") application.include_router(router, prefix="/api")
application.include_router(conversation_router, prefix="/api") application.include_router(conversation_router, prefix="/api")
application.include_router(material_router, prefix="/api")
return application return application
+37 -22
View File
@@ -1,4 +1,4 @@
"""读取前端组件目录""" """读取前端组件目录:启动时载入进程内存缓存,未载入时回退读取本地 JSON 文件"""
from pathlib import Path from pathlib import Path
@@ -8,38 +8,53 @@ from app.schemas.component_resource import (
ComponentResult, 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: class ComponentProvider:
def __init__(self, catalog_path: str | Path | None = None): def __init__(self, catalog_path: str | Path | None = None):
# 保存可选的目录文件路径,未指定时由查询方法使用项目内的默认目录 # 指定路径时始终读取该文件;未指定时优先使用启动缓存,缓存为空再读默认文件
self.catalog_path = catalog_path self.catalog_path = catalog_path
def query(self, query: ComponentQuery) -> ComponentResult: def query(self, query: ComponentQuery) -> ComponentResult:
# TODO 以后迁移数据库,保持组件查询接口不变 catalog = (
# 优先使用调用方指定的路径,否则读取当前模块旁的 catalog/components.json _cached_catalog
path = ( if self.catalog_path is None and _cached_catalog is not None
Path(self.catalog_path) else load_catalog_file(self.catalog_path)
if self.catalog_path is not None
else Path(__file__).resolve().parent / "catalog" / "components.json"
) )
if catalog is None:
# 如果传入的是相对路径,以项目根目录为基准,避免受启动时工作目录影响
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:
# 如果目录文件不存在,返回未支持状态和原因,由调用方决定失败处理方式 # 如果目录文件不存在,返回未支持状态和原因,由调用方决定失败处理方式
return ComponentResult( return ComponentResult(
status="unsupported", status="unsupported",
message="尚未配置前端组件目录,请先导出真实组件定义", message="尚未配置前端组件目录,请先导出真实组件定义",
) )
# 文件损坏或定义不合法直接报错,不能伪装成查无结果
catalog = ComponentCatalog.model_validate_json(content)
# 如果提供关键词,去除首尾空白并统一大小写,未提供时不启用关键词筛选 # 如果提供关键词,去除首尾空白并统一大小写,未提供时不启用关键词筛选
keyword = ( keyword = (
query.keyword.strip().casefold() if query.keyword is not None else None 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: if keyword is not None and keyword not in searchable:
continue continue
# 所有筛选条件通过,保留完整组件定义供后续规划和动作生成使用 # 深拷贝后返回,调用方修改结果不会污染共享缓存
matches.append(component) matches.append(component.model_copy(deep=True))
# 根据是否有匹配项区分查询成功和未找到,同时保留目录版本、来源及结果说明 # 根据是否有匹配项区分查询成功和未找到,同时保留目录版本、来源及结果说明
return ComponentResult( return ComponentResult(
+63
View File
@@ -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
+4 -1
View File
@@ -2,6 +2,7 @@
from collections.abc import Awaitable, Callable from collections.abc import Awaitable, Callable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any
from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.checkpoint.base import BaseCheckpointSaver
from loguru import logger from loguru import logger
@@ -16,6 +17,8 @@ class Persistence:
# 为空时由图使用默认的内存检查点 # 为空时由图使用默认的内存检查点
checkpointer: BaseCheckpointSaver | None checkpointer: BaseCheckpointSaver | None
close: Callable[[], Awaitable[None]] close: Callable[[], Awaitable[None]]
# 启用 MongoDB 时的数据库,供物料等其他集合复用同一连接
db: Any = None
async def _close_nothing() -> None: async def _close_nothing() -> None:
@@ -67,4 +70,4 @@ async def open_persistence() -> Persistence:
raise raise
logger.info("已启用 MongoDB 持久化,库名 {}", settings.mongodb_db) logger.info("已启用 MongoDB 持久化,库名 {}", settings.mongodb_db)
return Persistence(store, checkpointer, close) return Persistence(store, checkpointer, close, async_client[settings.mongodb_db])