feat(plugin_system): 扩展插件管理API，增加多种查询与辅助函数
对 `plugin_manage_api.py` 进行了大规模的重构和功能增强，以提高其可维护性和可用性。 主要变更包括： - **结构重构**: 将API按功能划分为四个逻辑部分：生命周期管理、状态管理、信息查询和工具函数，使代码结构更清晰。 - **功能扩展**: 新增了多个API端点用于查询插件和组件状态，例如 `get_plugin_details`, `list_plugins`, `search_components_by_name` 等。 - **文档完善**: 为所有公共API添加了详细的中文文档字符串和内联注释，解释了其用途、参数和返回值。 这些变更旨在为上层应用（如UI、CLI工具）提供一个更强大、更易于集成的插件系统管理接口。
This commit is contained in:
minecraft1024a
@@ -1,29 +1,47 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Plugin Manage API
|
||||
=================
|
||||
|
||||
该模块提供了用于管理插件和组件生命周期、状态和信息查询的核心API。
|
||||
功能包括插件的加载、重载、注册、扫描，组件的启用/禁用，以及系统状态报告的生成。
|
||||
所有函数都设计为异步或同步，以适应不同的调用上下文。
|
||||
"""
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
from typing import Any, Literal
|
||||
|
||||
from src.common.logger import get_logger
|
||||
from src.plugin_system.base.component_types import ComponentType
|
||||
from src.plugin_system.core.component_registry import ComponentInfo, component_registry
|
||||
from src.plugin_system.core.plugin_manager import plugin_manager
|
||||
|
||||
# 初始化日志记录器
|
||||
logger = get_logger("plugin_manage_api")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------------
|
||||
# Section 1: 插件生命周期管理 (Plugin Lifecycle Management)
|
||||
# --------------------------------------------------------------------------------
|
||||
# 该部分包含控制插件加载、重载、注册和发现的核心功能。
|
||||
|
||||
|
||||
async def reload_all_plugins() -> bool:
|
||||
"""
|
||||
重新加载所有当前已成功加载的插件。
|
||||
|
||||
此操作会先卸载所有插件，然后重新加载它们。
|
||||
此操作会遍历所有已加载的插件，逐一进行卸载和重新加载。
|
||||
如果任何一个插件重载失败，整个过程会继续，但最终返回 False。
|
||||
|
||||
Returns:
|
||||
bool: 如果所有插件都成功重载，则为 True，否则为 False。
|
||||
"""
|
||||
logger.info("开始重新加载所有插件...")
|
||||
# 使用 list() 复制一份列表，防止在迭代时修改原始列表
|
||||
# 使用 list() 创建一个当前已加载插件列表的副本，以避免在迭代过程中修改原始列表
|
||||
loaded_plugins = list(plugin_manager.list_loaded_plugins())
|
||||
all_success = True
|
||||
|
||||
# 遍历副本列表中的每个插件进行重载
|
||||
for plugin_name in loaded_plugins:
|
||||
try:
|
||||
success = await reload_plugin(plugin_name)
|
||||
@@ -32,7 +50,7 @@ async def reload_all_plugins() -> bool:
|
||||
logger.error(f"重载插件 {plugin_name} 失败。")
|
||||
except Exception as e:
|
||||
all_success = False
|
||||
logger.error(f"重载插件 {plugin_name} 时发生异常: {e}", exc_info=True)
|
||||
logger.error(f"重载插件 {plugin_name} 时发生未知异常: {e}", exc_info=True)
|
||||
|
||||
logger.info("所有插件重载完毕。")
|
||||
return all_success
|
||||
@@ -42,46 +60,162 @@ async def reload_plugin(name: str) -> bool:
|
||||
"""
|
||||
重新加载指定的单个插件。
|
||||
|
||||
该函数首先检查插件是否已注册，然后调用插件管理器执行重载操作。
|
||||
|
||||
Args:
|
||||
name (str): 要重载的插件的名称。
|
||||
|
||||
Returns:
|
||||
bool: 成功则为 True。
|
||||
bool: 如果插件成功重载，则为 True。
|
||||
|
||||
Raises:
|
||||
ValueError: 如果插件未找到。
|
||||
ValueError: 如果插件未在插件管理器中注册。
|
||||
"""
|
||||
# 验证插件是否存在于注册列表中
|
||||
if name not in plugin_manager.list_registered_plugins():
|
||||
raise ValueError(f"插件 '{name}' 未注册。")
|
||||
raise ValueError(f"插件 '{name}' 未注册，无法重载。")
|
||||
# 调用插件管理器的核心重载方法
|
||||
return await plugin_manager.reload_registered_plugin(name)
|
||||
|
||||
|
||||
def rescan_and_register_plugins(load_after_register: bool = True) -> tuple[int, int]:
|
||||
"""
|
||||
重新扫描所有插件目录，以发现并注册新插件。
|
||||
|
||||
此函数会触发插件管理器扫描其配置的所有插件目录。
|
||||
可以选择在注册新发现的插件后立即加载它们。
|
||||
|
||||
Args:
|
||||
load_after_register (bool): 如果为 True，新发现的插件将在注册后立即被加载。默认为 True。
|
||||
|
||||
Returns:
|
||||
tuple[int, int]: 一个元组，包含 (成功加载的插件数量, 加载失败的插件数量)。
|
||||
"""
|
||||
# 扫描插件目录，获取新注册成功和失败的数量
|
||||
success_count, fail_count = plugin_manager.rescan_plugin_directory()
|
||||
|
||||
# 如果不需要在注册后加载，则直接返回扫描结果
|
||||
if not load_after_register:
|
||||
return success_count, fail_count
|
||||
|
||||
# 找出新注册但尚未加载的插件
|
||||
newly_registered = [
|
||||
p for p in plugin_manager.list_registered_plugins() if p not in plugin_manager.list_loaded_plugins()
|
||||
]
|
||||
|
||||
loaded_success_count = 0
|
||||
# 尝试加载所有新注册的插件
|
||||
for plugin_name in newly_registered:
|
||||
status, _ = plugin_manager.load_registered_plugin_classes(plugin_name)
|
||||
if status:
|
||||
loaded_success_count += 1
|
||||
|
||||
# 计算总的成功和失败数量
|
||||
total_failed = fail_count + (len(newly_registered) - loaded_success_count)
|
||||
return loaded_success_count, total_failed
|
||||
|
||||
|
||||
def register_plugin_from_file(plugin_name: str, load_after_register: bool = True) -> bool:
|
||||
"""
|
||||
从插件目录中查找、注册并选择性地加载一个指定的插件。
|
||||
|
||||
如果插件已经加载，此函数将直接返回 True。
|
||||
如果插件未注册，它会遍历所有插件目录以查找匹配的插件文件夹。
|
||||
|
||||
Args:
|
||||
plugin_name (str): 插件的名称（通常是其目录名）。
|
||||
load_after_register (bool): 注册成功后是否立即加载该插件。默认为 True。
|
||||
|
||||
Returns:
|
||||
bool: 如果插件成功注册（并且根据参数成功加载），则为 True。
|
||||
"""
|
||||
# 如果插件已经加载，无需执行任何操作
|
||||
if plugin_name in plugin_manager.list_loaded_plugins():
|
||||
logger.warning(f"插件 '{plugin_name}' 已经加载，无需重复注册。")
|
||||
return True
|
||||
|
||||
# 如果插件尚未注册，则开始搜索流程
|
||||
if plugin_name not in plugin_manager.list_registered_plugins():
|
||||
logger.info(f"插件 '{plugin_name}' 未注册，开始在插件目录中搜索...")
|
||||
found_path = None
|
||||
|
||||
# 遍历所有配置的插件目录
|
||||
for directory in plugin_manager.plugin_directories:
|
||||
potential_path = os.path.join(directory, plugin_name)
|
||||
# 检查是否存在与插件同名的目录
|
||||
if os.path.isdir(potential_path):
|
||||
found_path = potential_path
|
||||
break
|
||||
|
||||
# 如果未找到插件目录，则报告错误
|
||||
if not found_path:
|
||||
logger.error(f"在所有插件目录中都未找到名为 '{plugin_name}' 的插件。")
|
||||
return False
|
||||
|
||||
# 检查插件的核心 'plugin.py' 文件是否存在
|
||||
plugin_file = os.path.join(found_path, "plugin.py")
|
||||
if not os.path.exists(plugin_file):
|
||||
logger.error(f"在插件目录 '{found_path}' 中未找到核心的 plugin.py 文件。")
|
||||
return False
|
||||
|
||||
# 尝试从文件加载插件模块
|
||||
module = plugin_manager._load_plugin_module_file(plugin_file)
|
||||
if not module:
|
||||
logger.error(f"从文件 '{plugin_file}' 加载插件模块失败。")
|
||||
return False
|
||||
|
||||
# 验证模块加载后，插件是否已成功注册
|
||||
if plugin_name not in plugin_manager.list_registered_plugins():
|
||||
logger.error(f"插件 '{plugin_name}' 在加载模块后依然未能成功注册。请检查插件定义。")
|
||||
return False
|
||||
|
||||
logger.info(f"插件 '{plugin_name}' 已成功发现并注册。")
|
||||
|
||||
# 根据参数决定是否在注册后立即加载插件
|
||||
if load_after_register:
|
||||
logger.info(f"正在加载插件 '{plugin_name}'...")
|
||||
status, _ = plugin_manager.load_registered_plugin_classes(plugin_name)
|
||||
return status
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------------
|
||||
# Section 2: 组件状态管理 (Component State Management)
|
||||
# --------------------------------------------------------------------------------
|
||||
# 这部分 API 负责控制单个组件的启用和禁用状态，支持全局和局部（临时）范围。
|
||||
|
||||
|
||||
async def set_component_enabled(name: str, component_type: ComponentType, enabled: bool) -> bool:
|
||||
"""
|
||||
全局范围内启用或禁用一个组件。
|
||||
在全局范围内启用或禁用一个组件。
|
||||
|
||||
此更改会更新组件注册表中的状态，但不会持久化到文件。
|
||||
此更改会直接修改组件在注册表中的状态，但此状态是临时的，不会持久化到配置文件中。
|
||||
包含一个保护机制，防止禁用最后一个已启用的 Chatter 组件。
|
||||
|
||||
Args:
|
||||
name (str): 组件名称。
|
||||
component_type (ComponentType): 组件类型。
|
||||
enabled (bool): True 为启用, False 为禁用。
|
||||
name (str): 要操作的组件的名称。
|
||||
component_type (ComponentType): 组件的类型。
|
||||
enabled (bool): True 表示启用, False 表示禁用。
|
||||
|
||||
Returns:
|
||||
bool: 操作成功则为 True。
|
||||
bool: 如果操作成功，则为 True。
|
||||
"""
|
||||
# Chatter 唯一性保护
|
||||
# 特殊保护：确保系统中至少有一个 Chatter 组件处于启用状态
|
||||
if component_type == ComponentType.CHATTER and not enabled:
|
||||
enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER)
|
||||
# 如果当前启用的 Chatter 少于等于1个，并且要禁用的就是它，则阻止操作
|
||||
if len(enabled_chatters) <= 1 and name in enabled_chatters:
|
||||
logger.warning(f"操作被阻止：不能禁用最后一个启用的 Chatter 组件 ('{name}')。")
|
||||
return False
|
||||
|
||||
# 注意：这里我们直接修改 ComponentInfo 中的状态
|
||||
# 获取组件信息
|
||||
component_info = component_registry.get_component_info(name, component_type)
|
||||
if not component_info:
|
||||
logger.error(f"未找到组件 {name} ({component_type.value})，无法更改其状态。")
|
||||
return False
|
||||
|
||||
# 直接修改组件实例的 enabled 状态
|
||||
component_info.enabled = enabled
|
||||
logger.info(f"组件 {name} ({component_type.value}) 的全局状态已设置为: {enabled}")
|
||||
return True
|
||||
@@ -91,179 +225,82 @@ def set_component_enabled_local(stream_id: str, name: str, component_type: Compo
|
||||
"""
|
||||
在一个特定的 stream_id 上下文中临时启用或禁用组件。
|
||||
|
||||
此状态仅存于内存，不影响全局状态。
|
||||
此状态仅存在于内存中，并且只对指定的 stream_id 有效，不影响全局组件状态。
|
||||
同样包含对 Chatter 组件的保护机制。
|
||||
|
||||
Args:
|
||||
stream_id (str): 上下文标识符。
|
||||
stream_id (str): 唯一的上下文标识符，例如一个会话ID。
|
||||
name (str): 组件名称。
|
||||
component_type (ComponentType): 组件类型。
|
||||
enabled (bool): True 为启用, False 为禁用。
|
||||
|
||||
Returns:
|
||||
bool: 操作成功则为 True。
|
||||
bool: 如果操作成功，则为 True。
|
||||
"""
|
||||
# 首先，检查组件是否存在
|
||||
# 首先，验证组件是否存在
|
||||
component_info = component_registry.get_component_info(name, component_type)
|
||||
if not component_info:
|
||||
logger.error(f"尝试设置局部状态失败：未找到组件 {name} ({component_type.value})。")
|
||||
return False
|
||||
|
||||
# Chatter 唯一性保护
|
||||
# Chatter 唯一性保护（在 stream_id 上下文中）
|
||||
if component_type == ComponentType.CHATTER and not enabled:
|
||||
# 检查当前 stream_id 上下文中的启用状态
|
||||
enabled_chatters = component_registry.get_enabled_components_by_type(
|
||||
ComponentType.CHATTER, stream_id=stream_id
|
||||
)
|
||||
# 检查当前 stream_id 上下文中启用的 Chatter
|
||||
enabled_chatters = component_registry.get_enabled_components_by_type(ComponentType.CHATTER, stream_id=stream_id)
|
||||
if len(enabled_chatters) <= 1 and name in enabled_chatters:
|
||||
logger.warning(
|
||||
f"操作被阻止：在 stream '{stream_id}' 中，不能禁用最后一个启用的 Chatter 组件 ('{name}')。"
|
||||
)
|
||||
logger.warning(f"操作被阻止：在 stream '{stream_id}' 中，不能禁用最后一个启用的 Chatter 组件 ('{name}')。")
|
||||
return False
|
||||
|
||||
# 设置局部状态
|
||||
component_registry.set_local_component_state(stream_id, name, component_type, enabled)
|
||||
logger.info(f"在 stream '{stream_id}' 中，组件 {name} ({component_type.value}) 的局部状态已设置为: {enabled}")
|
||||
return True
|
||||
|
||||
|
||||
def rescan_and_register_plugins(load_after_register: bool = True) -> tuple[int, int]:
|
||||
"""
|
||||
重新扫描所有插件目录，发现新插件并注册。
|
||||
|
||||
Args:
|
||||
load_after_register (bool): 如果为 True，新发现的插件将在注册后立即被加载。
|
||||
|
||||
Returns:
|
||||
Tuple[int, int]: (成功数量, 失败数量)
|
||||
"""
|
||||
success_count, fail_count = plugin_manager.rescan_plugin_directory()
|
||||
if not load_after_register:
|
||||
return success_count, fail_count
|
||||
|
||||
newly_registered = [
|
||||
p for p in plugin_manager.list_registered_plugins() if p not in plugin_manager.list_loaded_plugins()
|
||||
]
|
||||
loaded_success = 0
|
||||
for plugin_name in newly_registered:
|
||||
status, _ = plugin_manager.load_registered_plugin_classes(plugin_name)
|
||||
if status:
|
||||
loaded_success += 1
|
||||
|
||||
return loaded_success, fail_count + (len(newly_registered) - loaded_success)
|
||||
|
||||
|
||||
def register_plugin_from_file(plugin_name: str, load_after_register: bool = True) -> bool:
|
||||
"""
|
||||
从默认插件目录中查找、注册并加载一个插件。
|
||||
|
||||
Args:
|
||||
plugin_name (str): 插件的名称（即其目录名）。
|
||||
load_after_register (bool): 注册后是否立即加载。
|
||||
|
||||
Returns:
|
||||
bool: 成功则为 True。
|
||||
"""
|
||||
if plugin_name in plugin_manager.list_loaded_plugins():
|
||||
logger.warning(f"插件 '{plugin_name}' 已经加载。")
|
||||
return True
|
||||
|
||||
# 如果插件未注册，则遍历插件目录去查找
|
||||
if plugin_name not in plugin_manager.list_registered_plugins():
|
||||
logger.info(f"插件 '{plugin_name}' 未注册，开始在插件目录中搜索...")
|
||||
found_path = None
|
||||
for directory in plugin_manager.plugin_directories:
|
||||
potential_path = os.path.join(directory, plugin_name)
|
||||
if os.path.isdir(potential_path):
|
||||
found_path = potential_path
|
||||
break
|
||||
|
||||
if not found_path:
|
||||
logger.error(f"在所有插件目录中都未找到名为 '{plugin_name}' 的插件。")
|
||||
return False
|
||||
|
||||
plugin_file = os.path.join(found_path, "plugin.py")
|
||||
if not os.path.exists(plugin_file):
|
||||
logger.error(f"在 '{found_path}' 中未找到 plugin.py 文件。")
|
||||
return False
|
||||
|
||||
module = plugin_manager._load_plugin_module_file(plugin_file)
|
||||
if not module:
|
||||
logger.error(f"从 '{plugin_file}' 加载插件模块失败。")
|
||||
return False
|
||||
|
||||
if plugin_name not in plugin_manager.list_registered_plugins():
|
||||
logger.error(f"插件 '{plugin_name}' 在加载模块后依然未注册成功。")
|
||||
return False
|
||||
|
||||
logger.info(f"插件 '{plugin_name}' 已成功发现并注册。")
|
||||
|
||||
if load_after_register:
|
||||
status, _ = plugin_manager.load_registered_plugin_classes(plugin_name)
|
||||
return status
|
||||
return True
|
||||
|
||||
|
||||
def get_component_count(component_type: ComponentType, stream_id: str | None = None) -> int:
|
||||
"""
|
||||
获取指定类型的已加载并启用的组件的总数。
|
||||
|
||||
可以根据 stream_id 考虑局部状态。
|
||||
|
||||
Args:
|
||||
component_type (ComponentType): 要查询的组件类型。
|
||||
stream_id (str | None): 可选的上下文ID。
|
||||
|
||||
Returns:
|
||||
int: 该类型组件的数量。
|
||||
"""
|
||||
return len(component_registry.get_enabled_components_by_type(component_type, stream_id=stream_id))
|
||||
|
||||
|
||||
def get_component_info(name: str, component_type: ComponentType) -> ComponentInfo | None:
|
||||
"""
|
||||
获取任何一个已注册组件的详细信息。
|
||||
|
||||
Args:
|
||||
name (str): 组件的唯一名称。
|
||||
component_type (ComponentType): 组件的类型。
|
||||
|
||||
Returns:
|
||||
ComponentInfo: 包含组件信息的对象，如果找不到则返回 None。
|
||||
"""
|
||||
return component_registry.get_component_info(name, component_type)
|
||||
# --------------------------------------------------------------------------------
|
||||
# Section 3: 信息查询与报告 (Information Querying & Reporting)
|
||||
# --------------------------------------------------------------------------------
|
||||
# 这部分 API 用于获取关于插件和组件的详细信息、列表和统计数据。
|
||||
|
||||
|
||||
def get_system_report() -> dict[str, Any]:
|
||||
"""
|
||||
生成一份详细的系统状态报告。
|
||||
|
||||
报告包含已加载插件、失败插件和组件的全面信息，是调试和监控系统状态的核心工具。
|
||||
|
||||
Returns:
|
||||
dict: 包含系统、插件和组件状态的详细报告。
|
||||
dict[str, Any]: 包含系统、插件和组件状态的详细报告字典。
|
||||
"""
|
||||
loaded_plugins_info = {}
|
||||
# 遍历所有已加载的插件实例
|
||||
for name, instance in plugin_manager.loaded_plugins.items():
|
||||
plugin_info = component_registry.get_plugin_info(name)
|
||||
if not plugin_info:
|
||||
continue
|
||||
|
||||
components_details = []
|
||||
for comp_info in plugin_info.components:
|
||||
components_details.append(
|
||||
{
|
||||
"name": comp_info.name,
|
||||
"component_type": comp_info.component_type.value,
|
||||
"description": comp_info.description,
|
||||
"enabled": comp_info.enabled,
|
||||
}
|
||||
)
|
||||
# 收集该插件下所有组件的详细信息
|
||||
components_details = [
|
||||
{
|
||||
"name": comp_info.name,
|
||||
"component_type": comp_info.component_type.value,
|
||||
"description": comp_info.description,
|
||||
"enabled": comp_info.enabled,
|
||||
}
|
||||
for comp_info in plugin_info.components
|
||||
]
|
||||
|
||||
# 从 plugin_info (PluginInfo) 而不是 instance (PluginBase) 获取元数据
|
||||
# 构建单个插件的信息字典
|
||||
# 元数据从 PluginInfo 获取，而启用状态(enable_plugin)从插件实例获取
|
||||
loaded_plugins_info[name] = {
|
||||
"display_name": plugin_info.display_name or name,
|
||||
"version": plugin_info.version,
|
||||
"author": plugin_info.author,
|
||||
"enabled": instance.enable_plugin, # enable_plugin 状态还是需要从实例获取
|
||||
"enabled": instance.enable_plugin,
|
||||
"components": components_details,
|
||||
}
|
||||
|
||||
# 构建最终的完整报告
|
||||
report = {
|
||||
"system_info": {
|
||||
"loaded_plugins_count": len(plugin_manager.loaded_plugins),
|
||||
@@ -273,3 +310,228 @@ def get_system_report() -> dict[str, Any]:
|
||||
"failed_plugins": plugin_manager.failed_plugins,
|
||||
}
|
||||
return report
|
||||
|
||||
|
||||
def get_plugin_details(plugin_name: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
获取单个插件的详细报告。
|
||||
|
||||
报告内容包括插件的元数据、所有组件的详细信息及其当前状态。
|
||||
这是 `get_system_report` 的单插件聚焦版本。
|
||||
|
||||
Args:
|
||||
plugin_name (str): 要查询的插件名称。
|
||||
|
||||
Returns:
|
||||
dict | None: 包含插件详细信息的字典，如果插件未注册则返回 None。
|
||||
"""
|
||||
plugin_info = component_registry.get_plugin_info(plugin_name)
|
||||
if not plugin_info:
|
||||
logger.warning(f"尝试获取插件详情失败：未找到名为 '{plugin_name}' 的插件。")
|
||||
return None
|
||||
|
||||
# 收集该插件下所有组件的信息
|
||||
components_details = [
|
||||
{
|
||||
"name": comp_info.name,
|
||||
"component_type": comp_info.component_type.value,
|
||||
"description": comp_info.description,
|
||||
"enabled": comp_info.enabled,
|
||||
}
|
||||
for comp_info in plugin_info.components
|
||||
]
|
||||
|
||||
# 获取插件实例以检查其启用状态
|
||||
plugin_instance = plugin_manager.get_plugin_instance(plugin_name)
|
||||
is_enabled = plugin_instance.enable_plugin if plugin_instance else False
|
||||
|
||||
# 组装详细信息字典
|
||||
return {
|
||||
"name": plugin_info.name,
|
||||
"display_name": plugin_info.display_name or plugin_info.name,
|
||||
"version": plugin_info.version,
|
||||
"author": plugin_info.author,
|
||||
"license": plugin_info.license,
|
||||
"description": plugin_info.description,
|
||||
"enabled": is_enabled,
|
||||
"status": "loaded" if is_plugin_loaded(plugin_name) else "registered",
|
||||
"components": components_details,
|
||||
}
|
||||
|
||||
|
||||
def list_plugins(status: Literal["loaded", "registered", "failed"]) -> list[str]:
|
||||
"""
|
||||
根据指定的状态列出插件名称列表。
|
||||
|
||||
提供了一种快速、便捷的方式来监控和调试插件系统，而无需解析完整的系统报告。
|
||||
|
||||
Args:
|
||||
status (str): 插件状态，可选值为 'loaded', 'registered', 'failed'。
|
||||
|
||||
Returns:
|
||||
list[str]: 对应状态的插件名称列表。
|
||||
|
||||
Raises:
|
||||
ValueError: 如果传入了无效的状态字符串。
|
||||
"""
|
||||
if status == "loaded":
|
||||
# 返回所有当前已成功加载的插件
|
||||
return plugin_manager.list_loaded_plugins()
|
||||
if status == "registered":
|
||||
# 返回所有已注册（但不一定已加载）的插件
|
||||
return plugin_manager.list_registered_plugins()
|
||||
if status == "failed":
|
||||
# 返回所有加载失败的插件的名称
|
||||
return list(plugin_manager.failed_plugins.keys())
|
||||
# 如果状态无效，则引发错误
|
||||
raise ValueError(f"无效的插件状态: '{status}'。有效选项为 'loaded', 'registered', 'failed'。")
|
||||
|
||||
|
||||
def list_components(component_type: ComponentType, enabled_only: bool = True) -> list[dict[str, Any]]:
|
||||
"""
|
||||
列出指定类型的所有组件的详细信息。
|
||||
|
||||
这是查找和管理组件的核心功能，例如，获取所有可用的工具或所有注册的聊天器。
|
||||
|
||||
Args:
|
||||
component_type (ComponentType): 要查询的组件类型。
|
||||
enabled_only (bool, optional): 是否只返回已启用的组件。默认为 True。
|
||||
|
||||
Returns:
|
||||
list[dict[str, Any]]: 一个包含组件信息字典的列表。
|
||||
"""
|
||||
# 根据 enabled_only 参数决定是获取所有组件还是仅获取已启用的组件
|
||||
if enabled_only:
|
||||
components = component_registry.get_enabled_components_by_type(component_type)
|
||||
else:
|
||||
components = component_registry.get_components_by_type(component_type)
|
||||
|
||||
# 将组件信息格式化为字典列表
|
||||
return [
|
||||
{
|
||||
"name": info.name,
|
||||
"plugin_name": info.plugin_name,
|
||||
"description": info.description,
|
||||
"enabled": info.enabled,
|
||||
}
|
||||
for info in components.values()
|
||||
]
|
||||
|
||||
|
||||
def search_components_by_name(
|
||||
name_keyword: str,
|
||||
component_type: ComponentType | None = None,
|
||||
case_sensitive: bool = False,
|
||||
exact_match: bool = False,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
根据名称关键字搜索组件，支持模糊匹配和精确匹配。
|
||||
|
||||
极大地增强了组件的可发现性，用户无需知道完整名称即可找到所需组件。
|
||||
|
||||
Args:
|
||||
name_keyword (str): 用于搜索的名称关键字。
|
||||
component_type (ComponentType | None, optional): 如果提供，则只在该类型中搜索。默认为 None (搜索所有类型)。
|
||||
case_sensitive (bool, optional): 是否进行大小写敏感的搜索。默认为 False。
|
||||
exact_match (bool, optional): 是否进行精确匹配。默认为 False (模糊匹配)。
|
||||
|
||||
Returns:
|
||||
list[dict[str, Any]]: 匹配的组件信息字典的列表。
|
||||
"""
|
||||
results = []
|
||||
# 如果未指定组件类型，则搜索所有类型
|
||||
types_to_search = [component_type] if component_type else list(ComponentType)
|
||||
|
||||
# 根据是否大小写敏感，预处理搜索关键字
|
||||
compare_str = name_keyword if case_sensitive else name_keyword.lower()
|
||||
|
||||
# 遍历要搜索的组件类型
|
||||
for comp_type in types_to_search:
|
||||
all_components = component_registry.get_components_by_type(comp_type)
|
||||
for name, info in all_components.items():
|
||||
# 同样地，预处理组件名称
|
||||
target_name = name if case_sensitive else name.lower()
|
||||
|
||||
# 根据 exact_match 参数决定使用精确比较还是模糊包含检查
|
||||
is_match = (compare_str == target_name) if exact_match else (compare_str in target_name)
|
||||
|
||||
# 如果匹配，则将组件信息添加到结果列表
|
||||
if is_match:
|
||||
results.append(
|
||||
{
|
||||
"name": info.name,
|
||||
"component_type": info.component_type.value,
|
||||
"plugin_name": info.plugin_name,
|
||||
"description": info.description,
|
||||
"enabled": info.enabled,
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
def get_component_info(name: str, component_type: ComponentType) -> ComponentInfo | None:
|
||||
"""
|
||||
获取任何一个已注册组件的详细信息对象。
|
||||
|
||||
Args:
|
||||
name (str): 组件的唯一名称。
|
||||
component_type (ComponentType): 组件的类型。
|
||||
|
||||
Returns:
|
||||
ComponentInfo | None: 包含组件完整信息的 ComponentInfo 对象，如果找不到则返回 None。
|
||||
"""
|
||||
return component_registry.get_component_info(name, component_type)
|
||||
|
||||
|
||||
def get_component_count(component_type: ComponentType, stream_id: str | None = None) -> int:
|
||||
"""
|
||||
获取指定类型的已加载并启用的组件的总数。
|
||||
|
||||
可以根据 `stream_id` 考虑局部状态，从而获得特定上下文中的组件数量。
|
||||
|
||||
Args:
|
||||
component_type (ComponentType): 要查询的组件类型。
|
||||
stream_id (str | None): 可选的上下文ID。如果提供，将计入局部状态。
|
||||
|
||||
Returns:
|
||||
int: 该类型下已启用的组件的数量。
|
||||
"""
|
||||
return len(component_registry.get_enabled_components_by_type(component_type, stream_id=stream_id))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------------
|
||||
# Section 4: 工具函数 (Utility Helpers)
|
||||
# --------------------------------------------------------------------------------
|
||||
# 这部分提供了一些轻量级的辅助函数，用于快速检查状态。
|
||||
|
||||
|
||||
def is_plugin_loaded(plugin_name: str) -> bool:
|
||||
"""
|
||||
快速检查一个插件当前是否已成功加载。
|
||||
|
||||
这是一个比 `get_plugin_details` 更轻量级的检查方法，适用于需要快速布尔值判断的场景。
|
||||
|
||||
Args:
|
||||
plugin_name (str): 要检查的插件名称。
|
||||
|
||||
Returns:
|
||||
bool: 如果插件已加载，则为 True，否则为 False。
|
||||
"""
|
||||
return plugin_name in plugin_manager.list_loaded_plugins()
|
||||
|
||||
|
||||
def get_component_plugin(component_name: str, component_type: ComponentType) -> str | None:
|
||||
"""
|
||||
查找一个特定组件属于哪个插件。
|
||||
|
||||
在调试或管理组件时，此函数能够方便地追溯其定义的源头。
|
||||
|
||||
Args:
|
||||
component_name (str): 组件的名称。
|
||||
component_type (ComponentType): 组件的类型。
|
||||
|
||||
Returns:
|
||||
str | None: 组件所属的插件名称，如果找不到组件则返回 None。
|
||||
"""
|
||||
component_info = component_registry.get_component_info(component_name, component_type)
|
||||
return component_info.plugin_name if component_info else None
|
||||
|
||||
Reference in New Issue
Block a user