From 49229977e7ac5992d2d79b54994562fb20747996 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 3 Aug 2026 16:55:32 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20=E6=8F=92=E4=BB=B6=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E7=94=B1=E6=8F=92=E4=BB=B6=E8=87=AA=E5=B7=B1=E7=AE=A1=E7=90=86?= =?UTF-8?q?=EF=BC=8C=E5=8F=96=E6=B6=88=E5=85=A8=E5=B1=80=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/chongming/app/core/lifespan_setup.py | 90 +++++++++++++++++++++--- test_aicc.py | 2 +- 2 files changed, 81 insertions(+), 11 deletions(-) diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 53934c5..ef1746f 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -1,6 +1,8 @@ import os import inspect import time +import tomllib +import glob import multiprocessing import asyncio import socket @@ -215,31 +217,100 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager: app.state.refresh_task = refresh_task return manager +def _discover_plugins() -> list[dict]: + """ + 发现所有插件配置。 + + 优先从各插件的独立配置文件 config/workers//config.toml 的 [workers] 节读取， + 兼容旧的 config/workers/config.toml 全局配置。 + + Returns: + [{"type": "doubao", "module": "plugins.doubao", "factory": "create_workers", + "cleanup": "cleanup", "count": 3, "config": {...}}, ...] + """ + plugins = [] + + # 1. 自动扫描 config/workers/*/config.toml + for path in sorted(glob.glob("config/workers/*/config.toml")): + plugin_type = os.path.basename(os.path.dirname(path)) + try: + with open(path, "rb") as f: + plugin_conf = tomllib.load(f) + except Exception as e: + logger.warning("无法读取插件配置 %s: %s", path, e) + continue + + workers_section = plugin_conf.get("workers", {}) + if not workers_section: + continue + + # 合并运行时 config: 整个 TOML 顶层 + [workers.config] 展平 + runtime_config = { + **plugin_conf, + **workers_section.get("config", {}), + } + + plugins.append({ + "type": workers_section.get("type", plugin_type), + "module": workers_section.get("module", f"plugins.{plugin_type}"), + "factory": workers_section.get("factory", "create_workers"), + "cleanup": workers_section.get("cleanup", "cleanup"), + "count": workers_section.get("count", 1), + "config": runtime_config, + }) + + # 2. 兼容旧写法: config/workers/config.toml（仅加载尚未发现的插件） + old_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") + if os.path.isfile(old_config_path): + existing_types = {p["type"] for p in plugins} + old_conf = get_workers_config(old_config_path).get("workers", []) + for wc in old_conf: + wtype = wc.get("type", "") + if wtype and wtype not in existing_types: + plugins.append({ + "type": wtype, + "module": wc.get("module", ""), + "factory": wc.get("factory", ""), + "cleanup": wc.get("cleanup", ""), + "count": wc.get("count", 1), + "config": wc.get("config", {}), + }) + logger.info("从旧配置加载插件: %s (config/workers/config.toml)", wtype) + + return plugins + + async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerManager): """ 加载配置中的 worker 插件，启动对应的 worker 任务。 + 自动从 config/workers//config.toml 发现并加载插件。 + 每个插件的 [workers] 节定义加载参数，[workers.config] 节定义运行时配置。 + 同时自动发现插件模块中暴露的 APIRouter（约定：模块顶层名为 `router` 的 APIRouter 实例）， 收集后供 register_custom_routes() 统一注册，实现网关与插件的解耦。 返回: (worker_tasks, custom_routers) """ - workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") - workers_conf = get_workers_config(workers_config_path).get("workers", []) - logger.info(f"读取到 {len(workers_conf)} 个 worker 配置") + workers_conf = _discover_plugins() + logger.info(f"发现 {len(workers_conf)} 个 worker 插件") worker_tasks = [] custom_routers = [] for wc in workers_conf: - module_name = wc.get("module", "") - factory_name = wc.get("factory", "") - count = wc.get("count", 1) - worker_config = wc.get("config", {}) + module_name = wc["module"] + factory_name = wc["factory"] + count = wc["count"] + worker_type = wc["type"] + worker_config = wc["config"] if not module_name or not factory_name: logger.warning(f"Worker 配置不完整: {wc}") continue + logger.debug("加载插件 %s: config/workers/%s/config.toml (count=%d)", + worker_type, worker_type, count) + try: module = __import__(module_name, fromlist=[factory_name]) @@ -262,7 +333,7 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana import traceback logger.error( f"加载 Worker 插件失败 {module_name}: {e}\n" - f" 配置: type={wc.get('type')}, count={count}, config={worker_config}\n" + f" 配置: type={worker_type}, count={count}\n" f" 堆栈:\n{traceback.format_exc()}" ) @@ -361,8 +432,7 @@ async def shutdown_mqtt(app: FastAPI): async def cleanup_workers(app: FastAPI): """调用每个 worker 配置中定义的 cleanup 函数。""" - workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") - workers_conf = get_workers_config(workers_config_path).get("workers", []) + workers_conf = _discover_plugins() for wc in workers_conf: module_name = wc.get("module", "") diff --git a/test_aicc.py b/test_aicc.py index 0398a7c..57954f6 100644 --- a/test_aicc.py +++ b/test_aicc.py @@ -4,7 +4,7 @@ from openai import AsyncOpenAI from bytedance.volcengine_aicc_sdk.secure_http_client import AICCConfig, AsyncSecureHttpClient ep_name = "ep-20260721154333-fre3bey" -api_key = "db04d42e-71e3-4119-abee-12ecc7ef5c34" +api_key = "1363bf65-da9e-407b-b074-73fb9748ed67" base_url= "https://sd77a0f4ummt3i4vnems0.apigateway-cn-beijing.volceapi.com/api/v3" async def main():