modify: 插件配置由插件自己管理,取消全局配置

This commit is contained in:
2026-08-03 16:55:32 +08:00
parent 83a9824686
commit 49229977e7
2 changed files with 81 additions and 11 deletions
+80 -10
View File
@@ -1,6 +1,8 @@
import os import os
import inspect import inspect
import time import time
import tomllib
import glob
import multiprocessing import multiprocessing
import asyncio import asyncio
import socket import socket
@@ -215,31 +217,100 @@ def setup_worker_manager(app: FastAPI, session: zenoh.Session) -> WorkerManager:
app.state.refresh_task = refresh_task app.state.refresh_task = refresh_task
return manager return manager
def _discover_plugins() -> list[dict]:
"""
发现所有插件配置。
优先从各插件的独立配置文件 config/workers/<type>/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): async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerManager):
""" """
加载配置中的 worker 插件,启动对应的 worker 任务。 加载配置中的 worker 插件,启动对应的 worker 任务。
自动从 config/workers/<type>/config.toml 发现并加载插件。
每个插件的 [workers] 节定义加载参数,[workers.config] 节定义运行时配置。
同时自动发现插件模块中暴露的 APIRouter(约定:模块顶层名为 `router` 的 APIRouter 实例), 同时自动发现插件模块中暴露的 APIRouter(约定:模块顶层名为 `router` 的 APIRouter 实例),
收集后供 register_custom_routes() 统一注册,实现网关与插件的解耦。 收集后供 register_custom_routes() 统一注册,实现网关与插件的解耦。
返回: (worker_tasks, custom_routers) 返回: (worker_tasks, custom_routers)
""" """
workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") workers_conf = _discover_plugins()
workers_conf = get_workers_config(workers_config_path).get("workers", []) logger.info(f"发现 {len(workers_conf)} 个 worker 插件")
logger.info(f"读取到 {len(workers_conf)} 个 worker 配置")
worker_tasks = [] worker_tasks = []
custom_routers = [] custom_routers = []
for wc in workers_conf: for wc in workers_conf:
module_name = wc.get("module", "") module_name = wc["module"]
factory_name = wc.get("factory", "") factory_name = wc["factory"]
count = wc.get("count", 1) count = wc["count"]
worker_config = wc.get("config", {}) worker_type = wc["type"]
worker_config = wc["config"]
if not module_name or not factory_name: if not module_name or not factory_name:
logger.warning(f"Worker 配置不完整: {wc}") logger.warning(f"Worker 配置不完整: {wc}")
continue continue
logger.debug("加载插件 %s: config/workers/%s/config.toml (count=%d)",
worker_type, worker_type, count)
try: try:
module = __import__(module_name, fromlist=[factory_name]) module = __import__(module_name, fromlist=[factory_name])
@@ -262,7 +333,7 @@ async def load_workers(app: FastAPI, session: zenoh.Session, manager: WorkerMana
import traceback import traceback
logger.error( logger.error(
f"加载 Worker 插件失败 {module_name}: {e}\n" 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()}" f" 堆栈:\n{traceback.format_exc()}"
) )
@@ -361,8 +432,7 @@ async def shutdown_mqtt(app: FastAPI):
async def cleanup_workers(app: FastAPI): async def cleanup_workers(app: FastAPI):
"""调用每个 worker 配置中定义的 cleanup 函数。""" """调用每个 worker 配置中定义的 cleanup 函数。"""
workers_config_path = config.get("plugins_worker", {}).get("config_path", "config/workers/config.toml") workers_conf = _discover_plugins()
workers_conf = get_workers_config(workers_config_path).get("workers", [])
for wc in workers_conf: for wc in workers_conf:
module_name = wc.get("module", "") module_name = wc.get("module", "")
+1 -1
View File
@@ -4,7 +4,7 @@ from openai import AsyncOpenAI
from bytedance.volcengine_aicc_sdk.secure_http_client import AICCConfig, AsyncSecureHttpClient from bytedance.volcengine_aicc_sdk.secure_http_client import AICCConfig, AsyncSecureHttpClient
ep_name = "ep-20260721154333-fre3bey" 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" base_url= "https://sd77a0f4ummt3i4vnems0.apigateway-cn-beijing.volceapi.com/api/v3"
async def main(): async def main():