gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' of https://github.com/MoFox-Studio/MoFox-Core into dev

Browse Source
This commit is contained in:
tt-P607
2025-11-23 02:24:35 +08:00
parent f464befe66 dad86bcbc0
commit 19fefeb284
13 changed files with 1397 additions and 1965 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docker-compose.yml
View File

@@ -2,9 +2,9 @@ services:
core:
container_name: MoFox-Bot
#### prod ####
image: hunuon/mofox:latest
image: ericterminal/mofox:latest
#### dev ####
# image: hunuon/mofox:dev
# image: ericterminal/mofox:dev
environment:
- TZ=Asia/Shanghai
volumes:

1
pyproject.toml
View File

@@ -79,6 +79,7 @@ dependencies = [
"inkfox>=0.1.1",
"rjieba>=0.1.13",
"fastmcp>=2.13.0",
"jinja2>=3.1.0"
]
[[tool.uv.index]]

84
src/chat/message_receive/bot.py
View File

@@ -197,80 +197,6 @@ class ChatBot:
logger.error(f"处理PlusCommand时出错: {e}")
return False, None, True # 出错时继续处理消息
async def _process_commands_with_new_system(self, message: DatabaseMessages, chat: ChatStream):
# sourcery skip: use-named-expression
"""使用新插件系统处理命令"""
try:
text = message.processed_plain_text or ""
# 使用新的组件注册中心查找命令
command_result = component_registry.find_command_by_text(text)
if command_result:
command_class, matched_groups, command_info = command_result
plugin_name = command_info.plugin_name
command_name = command_info.name
if (
chat
and chat.stream_id
and command_name
in global_announcement_manager.get_disabled_chat_commands(chat.stream_id)
):
logger.info("用户禁用的命令，跳过处理")
return False, None, True
message.is_command = True
# 获取插件配置
plugin_config = component_registry.get_plugin_config(plugin_name)
# 创建命令实例
command_instance: BaseCommand = command_class(message, plugin_config)
command_instance.set_matched_groups(matched_groups)
# 为插件实例设置 chat_stream 运行时属性
setattr(command_instance, "chat_stream", chat)
try:
# 检查聊天类型限制
if not command_instance.is_chat_type_allowed():
is_group = chat.group_info is not None
logger.info(
f"命令 {command_class.__name__} 不支持当前聊天类型: {'群聊' if is_group else '私聊'}"
)
return False, None, True # 跳过此命令，继续处理其他消息
# 执行命令
success, response, intercept_message = await command_instance.execute()
# 记录命令执行结果
if success:
logger.info(f"命令执行成功: {command_class.__name__} (拦截: {intercept_message})")
else:
logger.warning(f"命令执行失败: {command_class.__name__} - {response}")
# 根据命令的拦截设置决定是否继续处理消息
return True, response, not intercept_message # 找到命令，根据intercept_message决定是否继续
except Exception as e:
logger.error(f"执行命令时出错: {command_class.__name__} - {e}")
logger.error(traceback.format_exc())
try:
await command_instance.send_text(f"命令执行出错: {e!s}")
except Exception as send_error:
logger.error(f"发送错误消息失败: {send_error}")
# 命令出错时，根据命令的拦截设置决定是否继续处理消息
return True, str(e), False # 出错时继续处理消息
# 没有找到命令，继续处理消息
return False, None, True
except Exception as e:
logger.error(f"处理命令时出错: {e}")
return False, None, True # 出错时继续处理消息
async def _handle_adapter_response_from_dict(self, seg_data: dict | None):
"""处理适配器命令响应（从字典数据）"""
try:
@@ -412,16 +338,6 @@ class ChatBot:
logger.info(f"PlusCommand处理完成，跳过后续消息处理: {plus_cmd_result}")
return
# 如果不是PlusCommand，尝试传统的BaseCommand处理
if not is_plus_command:
is_command, cmd_result, continue_process = await self._process_commands_with_new_system(message, chat)
# 如果是命令且不需要继续处理，则直接返回
if is_command and not continue_process:
await MessageStorage.store_message(message, chat)
logger.info(f"命令处理完成，跳过后续消息处理: {cmd_result}")
return
result = await event_manager.trigger_event(EventType.ON_MESSAGE, permission_group="SYSTEM", message=message)
if result and not result.all_continue_process():
raise UserWarning(f"插件{result.get_summary().get('stopped_handlers', '')}于消息到达时取消了消息处理")

100
src/chat/utils/prompt_component_manager.py
View File

@@ -29,14 +29,16 @@ class PromptComponentManager:
def __init__(self):
"""初始化管理器实例。"""
# _dynamic_rules 仅用于存储通过 API 动态添加的、非静态组件的规则。
# _dynamic_rules 存储通过 API 在运行时动态添加/修改的规则。
# 这是实现提示词动态性的核心数据结构。
# 结构: {
# "target_prompt_name": {
# "prompt_component_name": (InjectionRule, content_provider, source)
# "target_prompt_name": { // 目标 Prompt 的名称
# "prompt_component_name": (InjectionRule, content_provider, source) // 注入组件的规则详情
# }
# }
self._dynamic_rules: dict[str, dict[str, tuple[InjectionRule, Callable[..., Awaitable[str]], str]]] = {}
self._lock = asyncio.Lock() # 锁现在保护 _dynamic_rules
# 使用 asyncio.Lock 来确保对 _dynamic_rules 的所有写操作都是线程安全的。
self._lock = asyncio.Lock()
# --- 运行时规则管理 API ---
@@ -64,9 +66,13 @@ class PromptComponentManager:
Returns:
bool: 如果成功添加或更新，则返回 True。
"""
# 加锁以保证多协程环境下的数据一致性
async with self._lock:
# 遍历所有待添加的规则
for rule in rules:
# 使用 setdefault 确保目标 prompt 的规则字典存在
target_rules = self._dynamic_rules.setdefault(rule.target_prompt, {})
# 添加或覆盖指定组件的规则、内容提供者和来源
target_rules[prompt_name] = (rule, content_provider, source)
logger.info(f"成功添加/更新注入规则: '{prompt_name}' -> '{rule.target_prompt}' (来源: {source})")
return True
@@ -88,15 +94,16 @@ class PromptComponentManager:
如果未找到该组件的任何现有规则（无法复用），则返回 False。
"""
async with self._lock:
# 步骤 1: 查找现有的 content_provider 和 source
# 步骤 1: 遍历所有动态规则，查找指定组件已存在的 provider 和 source
found_provider: Callable[..., Awaitable[str]] | None = None
found_source: str | None = None
for target_rules in self._dynamic_rules.values():
if prompt_name in target_rules:
# 如果找到，记录其 provider 和 source 并跳出循环
_, found_provider, found_source = target_rules[prompt_name]
break
# 步骤 2: 如果找不到 provider，则操作失败
# 步骤 2: 如果遍历完仍未找到 provider，说明该组件无任何规则，无法复用
if not found_provider:
logger.warning(
f"尝试为组件 '{prompt_name}' 添加规则失败: "
@@ -105,7 +112,7 @@ class PromptComponentManager:
return False
# 步骤 3: 使用找到的 provider 和 source 添加新规则
source_to_use = found_source or "runtime" # 提供一个默认值以防万一
source_to_use = found_source or "runtime" # 如果 source 为 None，提供默认值
target_rules = self._dynamic_rules.setdefault(rule.target_prompt, {})
target_rules[prompt_name] = (rule, found_provider, source_to_use)
logger.info(
@@ -126,13 +133,16 @@ class PromptComponentManager:
bool: 如果成功移除，则返回 True；如果规则不存在，则返回 False。
"""
async with self._lock:
# 检查目标和组件规则是否存在
if target_prompt in self._dynamic_rules and prompt_name in self._dynamic_rules[target_prompt]:
# 存在则删除
del self._dynamic_rules[target_prompt][prompt_name]
# 如果目标下已无任何规则，则清理掉这个键
# 如果删除后，该目标下已无任何规则，则清理掉这个目标键，保持数据结构整洁
if not self._dynamic_rules[target_prompt]:
del self._dynamic_rules[target_prompt]
logger.info(f"成功移除注入规则: '{prompt_name}' from '{target_prompt}'")
return True
# 如果规则不存在，记录警告并返回 False
logger.warning(f"尝试移除注入规则失败: 未找到 '{prompt_name}' on '{target_prompt}'")
return False
@@ -153,7 +163,9 @@ class PromptComponentManager:
async with self._lock:
# 创建一个目标列表的副本进行迭代，因为我们可能会在循环中修改字典
for target_prompt in list(self._dynamic_rules.keys()):
# 检查当前目标下是否存在该组件的规则
if prompt_name in self._dynamic_rules[target_prompt]:
# 存在则删除
del self._dynamic_rules[target_prompt][prompt_name]
removed = True
logger.info(f"成功移除注入规则: '{prompt_name}' from '{target_prompt}'")
@@ -176,17 +188,23 @@ class PromptComponentManager:
async def content_provider(params: PromptParameters, target_prompt_name: str) -> str:
"""实际执行内容生成的异步函数。"""
try:
# 从注册表获取组件信息，用于后续获取插件配置
p_info = component_registry.get_component_info(component_name, ComponentType.PROMPT)
plugin_config = {}
if isinstance(p_info, PromptInfo):
# 获取该组件所属插件的配置
plugin_config = component_registry.get_plugin_config(p_info.plugin_name)
# 实例化组件，并传入所需参数
instance = component_class(
params=params, plugin_config=plugin_config, target_prompt_name=target_prompt_name
)
# 执行组件的 execute 方法以生成内容
result = await instance.execute()
# 确保返回的是字符串
return str(result) if result is not None else ""
except Exception as e:
# 捕获并记录执行过程中的任何异常，返回空字符串以避免注入失败
logger.error(f"执行规则提供者 '{component_name}' 时出错: {e}", exc_info=True)
return ""
@@ -202,16 +220,20 @@ class PromptComponentManager:
if not isinstance(info, PromptInfo):
continue
# 实时检查组件是否启用
# 实时检查组件是否已启用，跳过禁用的组件
if not component_registry.is_component_available(name, ComponentType.PROMPT):
continue
# 获取组件的类定义
component_class = component_registry.get_component_class(name, ComponentType.PROMPT)
if not (component_class and issubclass(component_class, BasePrompt)):
continue
# 为该组件创建一个内容提供者
provider = self._create_content_provider(name, component_class)
# 遍历组件定义的所有注入规则
for rule in info.injection_rules:
# 如果规则的目标与当前目标匹配，则添加到列表中
if rule.target_prompt == target_prompt_name:
all_rules.append((rule, provider, "static"))
@@ -219,11 +241,13 @@ class PromptComponentManager:
async with self._lock:
runtime_rules = self._dynamic_rules.get(target_prompt_name, {})
for name, (rule, provider, source) in runtime_rules.items():
# 确保运行时组件不会与禁用的静态组件冲突
# 检查该运行时规则是否关联到一个已注册的静态组件
static_info = component_registry.get_component_info(name, ComponentType.PROMPT)
# 如果关联的静态组件存在且被禁用，则跳过此运行时规则
if static_info and not component_registry.is_component_available(name, ComponentType.PROMPT):
logger.debug(f"跳过运行时规则 '{name}'，因为它关联的静态组件当前已禁用。")
continue
# 将有效的运行时规则添加到列表
all_rules.append((rule, provider, source))
return all_rules
@@ -252,17 +276,19 @@ class PromptComponentManager:
Returns:
str: 应用了所有注入规则后，最终生成的提示词模板字符串。
"""
# 构建适用于当前目标的所有规则
rules_for_target = await self._build_rules_for_target(target_prompt_name)
if not rules_for_target:
# 如果没有规则，直接返回原始模板
return original_template
# --- 占位符保护机制 ---
# 1. 保护: 找到所有 {placeholder} 并用临时标记替换
placeholders = re.findall(r"({[^{}]+})", original_template)
placeholder_map: dict[str, str] = {
f"__PROMPT_PLACEHOLDER_{i}__": p for i, p in enumerate(placeholders)
}
# 1. 保护: 将占位符替换为临时标记
protected_template = original_template
for marker, placeholder in placeholder_map.items():
protected_template = protected_template.replace(placeholder, marker)
@@ -271,6 +297,7 @@ class PromptComponentManager:
for rule, _, source in rules_for_target:
if rule.injection_type in (InjectionType.REMOVE, InjectionType.REPLACE) and rule.target_content:
try:
# 检查规则的 target_content (正则) 是否可能匹配到任何一个占位符
for p in placeholders:
if re.search(rule.target_content, p):
logger.warning(
@@ -278,10 +305,10 @@ class PromptComponentManager:
f"规则 `target_content` ('{rule.target_content}') "
f"可能会影响核心占位符 '{p}'。为保证系统稳定，该占位符已被保护，不会被此规则修改。"
)
# 只对每个规则警告一次
# 每个规则只警告一次
break
except re.error:
# 正则表达式本身有误，后面执行时会再次捕获，这里可忽略
# 如果正则表达式本身有误，后续执行时会捕获，此处可忽略
pass
# 3. 安全执行: 按优先级排序并应用规则
@@ -290,13 +317,16 @@ class PromptComponentManager:
modified_template = protected_template
for rule, provider, source in rules_for_target:
content = ""
# REMOVE 类型不需要生成内容
if rule.injection_type != InjectionType.REMOVE:
try:
# 调用内容提供者生成要注入的文本
content = await provider(params, target_prompt_name)
except Exception as e:
logger.error(f"执行规则 '{rule}' (来源: {source}) 的内容提供者时失败: {e}", exc_info=True)
continue
continue # 执行失败则跳过此规则
# 应用注入规则
try:
if rule.injection_type == InjectionType.PREPEND:
if content:
@@ -309,6 +339,7 @@ class PromptComponentManager:
modified_template = re.sub(rule.target_content, str(content), modified_template)
elif rule.injection_type == InjectionType.INSERT_AFTER:
if content and rule.target_content:
# 使用 \\g<0> 在匹配项后插入内容
replacement = f"\\g<0>\n{content}"
modified_template = re.sub(rule.target_content, replacement, modified_template)
elif rule.injection_type == InjectionType.REMOVE:
@@ -319,7 +350,7 @@ class PromptComponentManager:
except Exception as e:
logger.error(f"应用注入规则 '{rule}' (来源: {source}) 失败: {e}", exc_info=True)
# 4. 占位符恢复
# 4. 占位符恢复: 将临时标记替换回原始的占位符
final_template = modified_template
for marker, placeholder in placeholder_map.items():
final_template = final_template.replace(marker, placeholder)
@@ -343,8 +374,9 @@ class PromptComponentManager:
str: 模拟生成的最终提示词模板字符串。如果找不到模板，则返回错误信息。
"""
try:
# 从全局提示词管理器获取最原始的模板内容
# 动态导入以避免循环依赖
from src.chat.utils.prompt import global_prompt_manager
# 从全局管理器获取原始的、未经修改的提示词对象
original_prompt = global_prompt_manager._prompts.get(target_prompt_name)
if not original_prompt:
logger.warning(f"无法预览 '{target_prompt_name}'，因为找不到这个核心 Prompt。")
@@ -354,14 +386,16 @@ class PromptComponentManager:
logger.warning(f"无法预览 '{target_prompt_name}'，因为找不到这个核心 Prompt。")
return f"Error: Prompt '{target_prompt_name}' not found."
# 直接调用核心注入逻辑来模拟结果
# 直接调用核心注入逻辑来模拟并返回结果
return await self.apply_injections(target_prompt_name, original_template, params)
# --- 状态观测与查询 API ---
def get_core_prompts(self) -> list[str]:
"""获取所有已注册的核心提示词模板名称列表（即所有可注入的目标）。"""
# 动态导入以避免循环依赖
from src.chat.utils.prompt import global_prompt_manager
# 返回所有核心 prompt 的名称列表
return list(global_prompt_manager._prompts.keys())
def get_core_prompt_contents(self, prompt_name: str | None = None) -> list[list[str]]:
@@ -381,9 +415,11 @@ class PromptComponentManager:
from src.chat.utils.prompt import global_prompt_manager
if prompt_name:
# 如果指定了名称，则查找并返回单个模板
prompt = global_prompt_manager._prompts.get(prompt_name)
return [[prompt_name, prompt.template]] if prompt else []
# 如果未指定名称，则返回所有模板的列表
return [[name, prompt.template] for name, prompt in global_prompt_manager._prompts.items()]
async def get_registered_prompt_component_info(self) -> list[PromptInfo]:
@@ -391,21 +427,23 @@ class PromptComponentManager:
获取所有已注册和动态添加的Prompt组件信息，并反映当前的注入规则状态。
此方法现在直接从 component_registry 获取静态组件信息，并合并纯运行时的组件信息。
"""
# 该方法现在直接从 component_registry 获取信息，因为它总是有最新的数据
# 从注册表获取所有已注册的静态 Prompt 组件信息
all_components = component_registry.get_components_by_type(ComponentType.PROMPT)
info_list = [info for info in all_components.values() if isinstance(info, PromptInfo)]
# 检查是否有纯动态组件需要添加
# 检查并合并仅在运行时通过 API 添加的“纯动态”组件
async with self._lock:
runtime_component_names = set()
# 收集所有动态规则中涉及的组件名称
for rules in self._dynamic_rules.values():
runtime_component_names.update(rules.keys())
static_component_names = {info.name for info in info_list}
# 找出那些只存在于动态规则中，但未在静态组件中注册的名称
pure_dynamic_names = runtime_component_names - static_component_names
for name in pure_dynamic_names:
# 为纯动态组件创建临时的 PromptInfo
# 为这些“纯动态”组件创建一个临时的信息对象
dynamic_info = PromptInfo(
name=name,
component_type=ComponentType.PROMPT,
@@ -413,7 +451,7 @@ class PromptComponentManager:
plugin_name="runtime",
is_built_in=False,
)
# 从 _dynamic_rules 中收集其所有规则
# 从动态规则中收集并关联其所有注入规则
for target, rules_in_target in self._dynamic_rules.items():
if name in rules_in_target:
rule, _, _ = rules_in_target[name]
@@ -433,10 +471,11 @@ class PromptComponentManager:
"""
info_map = {}
all_core_prompts = self.get_core_prompts()
# 确定要处理的目标：如果指定了有效的目标，则只处理它；否则处理所有核心 prompt
targets_to_process = [target_prompt] if target_prompt and target_prompt in all_core_prompts else all_core_prompts
for target in targets_to_process:
# 动态构建规则列表
# 动态构建该目标的所有有效规则
rules_for_target = await self._build_rules_for_target(target)
if not rules_for_target:
info_map[target] = []
@@ -444,9 +483,10 @@ class PromptComponentManager:
info_list = []
for rule, _, source in rules_for_target:
# 从规则本身获取组件名
# 从规则对象中获取其所属组件的名称
prompt_name = rule.owner_component
if detailed:
# 如果需要详细信息，则添加更多字段
info_list.append(
{
"name": prompt_name,
@@ -457,8 +497,10 @@ class PromptComponentManager:
}
)
else:
# 否则只添加基本信息
info_list.append({"name": prompt_name, "priority": rule.priority, "source": source})
# 按优先级对结果进行排序
info_list.sort(key=lambda x: x["priority"])
info_map[target] = info_list
return info_map
@@ -491,31 +533,33 @@ class PromptComponentManager:
for name, info in static_components.items():
if not isinstance(info, PromptInfo):
continue
# 应用 component_name 筛选
# 如果指定了 component_name 且不匹配，则跳过此组件
if component_name and name != component_name:
continue
for rule in info.injection_rules:
# 应用 target_prompt 筛选
# 如果指定了 target_prompt 且不匹配，则跳过此规则
if target_prompt and rule.target_prompt != target_prompt:
continue
target_dict = all_rules.setdefault(rule.target_prompt, {})
target_dict[name] = rule
# 2. 收集并合并所有纯运行时规则
# 2. 收集并合并所有运行时规则
async with self._lock:
for target, rules_in_target in self._dynamic_rules.items():
# 应用 target_prompt 筛选
# 如果指定了 target_prompt 且不匹配，则跳过此目标下的所有规则
if target_prompt and target != target_prompt:
continue
for name, (rule, _, _) in rules_in_target.items():
# 应用 component_name 筛选
# 如果指定了 component_name 且不匹配，则跳过此规则
if component_name and name != component_name:
continue
target_dict = all_rules.setdefault(target, {})
# 运行时规则会覆盖同名的静态规则
target_dict[name] = rule
# 返回深拷贝以防止外部修改影响内部状态
return copy.deepcopy(all_rules)

127
src/individuality/not_using/offline_llm.py
View File

@@ -1,127 +0,0 @@
import asyncio
import os
import time
import aiohttp
import requests
from rich.traceback import install
from src.common.logger import get_logger
from src.common.tcp_connector import get_tcp_connector
install(extra_lines=3)
logger = get_logger("offline_llm")
class LLMRequestOff:
def __init__(self, model_name="Pro/deepseek-ai/DeepSeek-V3", **kwargs):
self.model_name = model_name
self.params = kwargs
self.api_key = os.getenv("SILICONFLOW_KEY")
self.base_url = os.getenv("SILICONFLOW_BASE_URL")
if not self.api_key or not self.base_url:
raise ValueError("环境变量未正确加载：SILICONFLOW_KEY 或 SILICONFLOW_BASE_URL 未设置")
# logger.info(f"API URL: {self.base_url}") # 使用 logger 记录 base_url
def generate_response(self, prompt: str) -> str | tuple[str, str]:
"""根据输入的提示生成模型的响应"""
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
# 构建请求体
data = {
"model": self.model_name,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.4,
**self.params,
}
# 发送请求到完整的 chat/completions 端点
api_url = f"{self.base_url.rstrip('/')}/chat/completions" # type: ignore
logger.info(f"Request URL: {api_url}") # 记录请求的 URL
max_retries = 3
base_wait_time = 15 # 基础等待时间（秒）
for retry in range(max_retries):
try:
response = requests.post(api_url, headers=headers, json=data)
if response.status_code == 429:
wait_time = base_wait_time * (2**retry) # 指数退避
logger.warning(f"遇到请求限制(429)，等待{wait_time}秒后重试...")
time.sleep(wait_time)
continue
response.raise_for_status() # 检查其他响应状态
result = response.json()
if "choices" in result and len(result["choices"]) > 0:
content = result["choices"][0]["message"]["content"]
reasoning_content = result["choices"][0]["message"].get("reasoning_content", "")
return content, reasoning_content
return "没有返回结果", ""
except Exception as e:
if retry < max_retries - 1: # 如果还有重试机会
wait_time = base_wait_time * (2**retry)
logger.error(f"[回复]请求失败，等待{wait_time}秒后重试... 错误: {e!s}")
time.sleep(wait_time)
else:
logger.error(f"请求失败: {e!s}")
return f"请求失败: {e!s}", ""
logger.error("达到最大重试次数，请求仍然失败")
return "达到最大重试次数，请求仍然失败", ""
async def generate_response_async(self, prompt: str) -> str | tuple[str, str]:
"""异步方式根据输入的提示生成模型的响应"""
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
# 构建请求体
data = {
"model": self.model_name,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.5,
**self.params,
}
# 发送请求到完整的 chat/completions 端点
api_url = f"{self.base_url.rstrip('/')}/chat/completions" # type: ignore
logger.info(f"Request URL: {api_url}") # 记录请求的 URL
max_retries = 3
base_wait_time = 15
async with aiohttp.ClientSession(connector=await get_tcp_connector()) as session:
for retry in range(max_retries):
try:
async with session.post(api_url, headers=headers, json=data) as response:
if response.status == 429:
wait_time = base_wait_time * (2**retry) # 指数退避
logger.warning(f"遇到请求限制(429)，等待{wait_time}秒后重试...")
await asyncio.sleep(wait_time)
continue
response.raise_for_status() # 检查其他响应状态
result = await response.json()
if "choices" in result and len(result["choices"]) > 0:
content = result["choices"][0]["message"]["content"]
reasoning_content = result["choices"][0]["message"].get("reasoning_content", "")
return content, reasoning_content
return "没有返回结果", ""
except Exception as e:
if retry < max_retries - 1: # 如果还有重试机会
wait_time = base_wait_time * (2**retry)
logger.error(f"[回复]请求失败，等待{wait_time}秒后重试... 错误: {e!s}")
await asyncio.sleep(wait_time)
else:
logger.error(f"请求失败: {e!s}")
return f"请求失败: {e!s}", ""
logger.error("达到最大重试次数，请求仍然失败")
return "达到最大重试次数，请求仍然失败", ""

307
src/individuality/not_using/per_bf_gen.py
View File

@@ -1,307 +0,0 @@
import os
import random
import sys
import orjson
import toml
from dotenv import load_dotenv
from tqdm import tqdm
# 添加项目根目录到 Python 路径
root_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
sys.path.append(root_path)
# 加载配置文件
config_path = os.path.join(root_path, "config", "bot_config.toml")
with open(config_path, encoding="utf-8") as f:
config = toml.load(f)
# 现在可以导入src模块
from individuality.not_using.scene import get_scene_by_factor, PERSONALITY_SCENES # noqa E402
from individuality.not_using.questionnaire import FACTOR_DESCRIPTIONS
from individuality.not_using.offline_llm import LLMRequestOff
# 加载环境变量
env_path = os.path.join(root_path, ".env")
if os.path.exists(env_path):
print(f"从 {env_path} 加载环境变量")
load_dotenv(env_path)
else:
print(f"未找到环境变量文件: {env_path}")
print("将使用默认配置")
def adapt_scene(scene: str) -> str:
personality_core = config["personality"]["personality_core"]
personality_side = config["personality"]["personality_side"]
personality_side = random.choice(personality_side)
identitys = config["identity"]["identity"]
identity = random.choice(identitys)
"""
根据config中的属性，改编场景使其更适合当前角色
Args:
scene: 原始场景描述
Returns:
str: 改编后的场景描述
"""
try:
prompt = f"""
这是一个参与人格测评的角色形象:
- 昵称: {config["bot"]["nickname"]}
- 性别: {config["identity"]["gender"]}
- 年龄: {config["identity"]["age"]}岁
- 外貌: {config["identity"]["appearance"]}
- 性格核心: {personality_core}
- 性格侧面: {personality_side}
- 身份细节: {identity}
请根据上述形象，改编以下场景，在测评中，用户将根据该场景给出上述角色形象的反应:
{scene}
保持场景的本质不变，但最好贴近生活且具体，并且让它更适合这个角色。
改编后的场景应该自然、连贯，并考虑角色的年龄、身份和性格特点。只返回改编后的场景描述，不要包含其他说明。注意{config["bot"]["nickname"]}是面对这个场景的人，而不是场景的其他人。场景中不会有其描述，
现在，请你给出改编后的场景描述
"""
llm = LLMRequestOff(model_name=config["model"]["llm_normal"]["name"])
adapted_scene, _ = llm.generate_response(prompt)
# 检查返回的场景是否为空或错误信息
if not adapted_scene or "错误" in adapted_scene or "失败" in adapted_scene:
print("场景改编失败，将使用原始场景")
return scene
return adapted_scene
except Exception as e:
print(f"场景改编过程出错：{e!s}，将使用原始场景")
return scene
class PersonalityEvaluatorDirect:
def __init__(self):
self.personality_traits = {"开放性": 0, "严谨性": 0, "外向性": 0, "宜人性": 0, "神经质": 0}
self.scenarios = []
self.final_scores: dict[str, float] = {"开放性": 0, "严谨性": 0, "外向性": 0, "宜人性": 0, "神经质": 0}
self.dimension_counts = dict.fromkeys(self.final_scores, 0)
# 为每个人格特质获取对应的场景
for trait in PERSONALITY_SCENES:
scenes = get_scene_by_factor(trait)
if not scenes:
continue
# 从每个维度选择3个场景
import random
scene_keys = list(scenes.keys())
selected_scenes = random.sample(scene_keys, min(3, len(scene_keys)))
for scene_key in selected_scenes:
scene = scenes[scene_key]
# 为每个场景添加评估维度
# 主维度是当前特质，次维度随机选择一个其他特质
other_traits = [t for t in PERSONALITY_SCENES if t != trait]
secondary_trait = random.choice(other_traits)
self.scenarios.append(
{"场景": scene["scenario"], "评估维度": [trait, secondary_trait], "场景编号": scene_key}
)
self.llm = LLMRequestOff()
def evaluate_response(self, scenario: str, response: str, dimensions: list[str]) -> dict[str, float]:
"""
使用 DeepSeek AI 评估用户对特定场景的反应
"""
# 构建维度描述
dimension_descriptions = [f"- {dim}：{desc}" for dim in dimensions if (desc := FACTOR_DESCRIPTIONS.get(dim, ""))]
dimensions_text = "\n".join(dimension_descriptions)
prompt = f"""请根据以下场景和用户描述，评估用户在大五人格模型中的相关维度得分（1-6分）。
场景描述：
{scenario}
用户回应：
{response}
需要评估的维度说明：
{dimensions_text}
请按照以下格式输出评估结果（仅输出JSON格式）：
{{
"{dimensions[0]}": 分数,
"{dimensions[1]}": 分数
}}
评分标准：
1 = 非常不符合该维度特征
2 = 比较不符合该维度特征
3 = 有点不符合该维度特征
4 = 有点符合该维度特征
5 = 比较符合该维度特征
6 = 非常符合该维度特征
请根据用户的回应，结合场景和维度说明进行评分。确保分数在1-6之间，并给出合理的评估。"""
try:
ai_response, _ = self.llm.generate_response(prompt)
# 尝试从AI响应中提取JSON部分
start_idx = ai_response.find("{")
end_idx = ai_response.rfind("}") + 1
if start_idx != -1 and end_idx != 0:
json_str = ai_response[start_idx:end_idx]
scores = orjson.loads(json_str)
# 确保所有分数在1-6之间
return {k: max(1, min(6, float(v))) for k, v in scores.items()}
else:
print("AI响应格式不正确，使用默认评分")
return dict.fromkeys(dimensions, 3.5)
except Exception as e:
print(f"评估过程出错：{e!s}")
return dict.fromkeys(dimensions, 3.5)
def run_evaluation(self):
"""
运行整个评估过程
"""
print(f"欢迎使用{config['bot']['nickname']}形象创建程序！")
print("接下来，将给您呈现一系列有关您bot的场景（共15个）。")
print("请想象您的bot在以下场景下会做什么，并描述您的bot的反应。")
print("每个场景都会进行不同方面的评估。")
print("\n角色基本信息：")
print(f"- 昵称：{config['bot']['nickname']}")
print(f"- 性格核心：{config['personality']['personality_core']}")
print(f"- 性格侧面：{config['personality']['personality_side']}")
print(f"- 身份细节：{config['identity']['identity']}")
print("\n准备好了吗？按回车键开始...")
input()
total_scenarios = len(self.scenarios)
progress_bar = tqdm(
total=total_scenarios,
desc="场景进度",
ncols=100,
bar_format="{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}]",
)
for _i, scenario_data in enumerate(self.scenarios, 1):
# print(f"\n{'-' * 20} 场景 {i}/{total_scenarios} - {scenario_data['场景编号']} {'-' * 20}")
# 改编场景，使其更适合当前角色
print(f"{config['bot']['nickname']}祈祷中...")
adapted_scene = adapt_scene(scenario_data["场景"])
scenario_data["改编场景"] = adapted_scene
print(adapted_scene)
print(f"\n请描述{config['bot']['nickname']}在这种情况下会如何反应：")
response = input().strip()
if not response:
print("反应描述不能为空！")
continue
print("\n正在评估您的描述...")
scores = self.evaluate_response(adapted_scene, response, scenario_data["评估维度"])
# 更新最终分数
for dimension, score in scores.items():
self.final_scores[dimension] += score
self.dimension_counts[dimension] += 1
print("\n当前评估结果：")
print("-" * 30)
for dimension, score in scores.items():
print(f"{dimension}: {score}/6")
# 更新进度条
progress_bar.update(1)
# if i < total_scenarios:
# print("\n按回车键继续下一个场景...")
# input()
progress_bar.close()
# 计算平均分
for dimension in self.final_scores:
if self.dimension_counts[dimension] > 0:
self.final_scores[dimension] = round(self.final_scores[dimension] / self.dimension_counts[dimension], 2)
print("\n" + "=" * 50)
print(f" {config['bot']['nickname']}的人格特征评估结果 ".center(50))
print("=" * 50)
for trait, score in self.final_scores.items():
print(f"{trait}: {score}/6".ljust(20) + f"测试场景数：{self.dimension_counts[trait]}".rjust(30))
print("=" * 50)
# 返回评估结果
return self.get_result()
def get_result(self):
"""
获取评估结果
"""
return {
"final_scores": self.final_scores,
"dimension_counts": self.dimension_counts,
"scenarios": self.scenarios,
"bot_info": {
"nickname": config["bot"]["nickname"],
"gender": config["identity"]["gender"],
"age": config["identity"]["age"],
"height": config["identity"]["height"],
"weight": config["identity"]["weight"],
"appearance": config["identity"]["appearance"],
"personality_core": config["personality"]["personality_core"],
"personality_side": config["personality"]["personality_side"],
"identity": config["identity"]["identity"],
},
}
def main():
evaluator = PersonalityEvaluatorDirect()
result = evaluator.run_evaluation()
# 准备简化的结果数据
simplified_result = {
"openness": round(result["final_scores"]["开放性"] / 6, 1), # 转换为0-1范围
"conscientiousness": round(result["final_scores"]["严谨性"] / 6, 1),
"extraversion": round(result["final_scores"]["外向性"] / 6, 1),
"agreeableness": round(result["final_scores"]["宜人性"] / 6, 1),
"neuroticism": round(result["final_scores"]["神经质"] / 6, 1),
"bot_nickname": config["bot"]["nickname"],
}
# 确保目录存在
save_dir = os.path.join(root_path, "data", "personality")
os.makedirs(save_dir, exist_ok=True)
# 创建文件名，替换可能的非法字符
bot_name = config["bot"]["nickname"]
# 替换Windows文件名中不允许的字符
for char in ["\\", "/", ":", "*", "?", '"', "<", ">", "|"]:
bot_name = bot_name.replace(char, "_")
file_name = f"{bot_name}_personality.per"
save_path = os.path.join(save_dir, file_name)
# 保存简化的结果
with open(save_path, "w", encoding="utf-8") as f:
f.write(orjson.dumps(simplified_result, option=orjson.OPT_INDENT_2).decode("utf-8"))
print(f"\n结果已保存到 {save_path}")
# 同时保存完整结果到results目录
os.makedirs("results", exist_ok=True)
with open("results/personality_result.json", "w", encoding="utf-8") as f:
f.write(orjson.dumps(result, option=orjson.OPT_INDENT_2).decode("utf-8"))
if __name__ == "__main__":
main()

142
src/individuality/not_using/questionnaire.py
View File

@@ -1,142 +0,0 @@
# 人格测试问卷题目
# 王孟成, 戴晓阳, & 姚树桥. (2011).
# 中国大五人格问卷的初步编制Ⅲ:简式版的制定及信效度检验. 中国临床心理学杂志, 19(04), Article 04.
# 王孟成, 戴晓阳, & 姚树桥. (2010).
# 中国大五人格问卷的初步编制Ⅰ:理论框架与信度分析. 中国临床心理学杂志, 18(05), Article 05.
PERSONALITY_QUESTIONS = [
# 神经质维度 (F1)
{"id": 1, "content": "我常担心有什么不好的事情要发生", "factor": "神经质", "reverse_scoring": False},
{"id": 2, "content": "我常感到害怕", "factor": "神经质", "reverse_scoring": False},
{"id": 3, "content": "有时我觉得自己一无是处", "factor": "神经质", "reverse_scoring": False},
{"id": 4, "content": "我很少感到忧郁或沮丧", "factor": "神经质", "reverse_scoring": True},
{"id": 5, "content": "别人一句漫不经心的话，我常会联系在自己身上", "factor": "神经质", "reverse_scoring": False},
{"id": 6, "content": "在面对压力时，我有种快要崩溃的感觉", "factor": "神经质", "reverse_scoring": False},
{"id": 7, "content": "我常担忧一些无关紧要的事情", "factor": "神经质", "reverse_scoring": False},
{"id": 8, "content": "我常常感到内心不踏实", "factor": "神经质", "reverse_scoring": False},
# 严谨性维度 (F2)
{"id": 9, "content": "在工作上，我常只求能应付过去便可", "factor": "严谨性", "reverse_scoring": True},
{"id": 10, "content": "一旦确定了目标，我会坚持努力地实现它", "factor": "严谨性", "reverse_scoring": False},
{"id": 11, "content": "我常常是仔细考虑之后才做出决定", "factor": "严谨性", "reverse_scoring": False},
{"id": 12, "content": "别人认为我是个慎重的人", "factor": "严谨性", "reverse_scoring": False},
{"id": 13, "content": "做事讲究逻辑和条理是我的一个特点", "factor": "严谨性", "reverse_scoring": False},
{"id": 14, "content": "我喜欢一开头就把事情计划好", "factor": "严谨性", "reverse_scoring": False},
{"id": 15, "content": "我工作或学习很勤奋", "factor": "严谨性", "reverse_scoring": False},
{"id": 16, "content": "我是个倾尽全力做事的人", "factor": "严谨性", "reverse_scoring": False},
# 宜人性维度 (F3)
{
"id": 17,
"content": "尽管人类社会存在着一些阴暗的东西（如战争、罪恶、欺诈），我仍然相信人性总的来说是善良的",
"factor": "宜人性",
"reverse_scoring": False,
},
{"id": 18, "content": "我觉得大部分人基本上是心怀善意的", "factor": "宜人性", "reverse_scoring": False},
{"id": 19, "content": "虽然社会上有骗子，但我觉得大部分人还是可信的", "factor": "宜人性", "reverse_scoring": False},
{"id": 20, "content": "我不太关心别人是否受到不公正的待遇", "factor": "宜人性", "reverse_scoring": True},
{"id": 21, "content": "我时常觉得别人的痛苦与我无关", "factor": "宜人性", "reverse_scoring": True},
{"id": 22, "content": "我常为那些遭遇不幸的人感到难过", "factor": "宜人性", "reverse_scoring": False},
{"id": 23, "content": "我是那种只照顾好自己，不替别人担忧的人", "factor": "宜人性", "reverse_scoring": True},
{"id": 24, "content": "当别人向我诉说不幸时，我常感到难过", "factor": "宜人性", "reverse_scoring": False},
# 开放性维度 (F4)
{"id": 25, "content": "我的想象力相当丰富", "factor": "开放性", "reverse_scoring": False},
{"id": 26, "content": "我头脑中经常充满生动的画面", "factor": "开放性", "reverse_scoring": False},
{"id": 27, "content": "我对许多事情有着很强的好奇心", "factor": "开放性", "reverse_scoring": False},
{"id": 28, "content": "我喜欢冒险", "factor": "开放性", "reverse_scoring": False},
{"id": 29, "content": "我是个勇于冒险，突破常规的人", "factor": "开放性", "reverse_scoring": False},
{"id": 30, "content": "我身上具有别人没有的冒险精神", "factor": "开放性", "reverse_scoring": False},
{
"id": 31,
"content": "我渴望学习一些新东西，即使它们与我的日常生活无关",
"factor": "开放性",
"reverse_scoring": False,
},
{
"id": 32,
"content": "我很愿意也很容易接受那些新事物、新观点、新想法",
"factor": "开放性",
"reverse_scoring": False,
},
# 外向性维度 (F5)
{"id": 33, "content": "我喜欢参加社交与娱乐聚会", "factor": "外向性", "reverse_scoring": False},
{"id": 34, "content": "我对人多的聚会感到乏味", "factor": "外向性", "reverse_scoring": True},
{"id": 35, "content": "我尽量避免参加人多的聚会和嘈杂的环境", "factor": "外向性", "reverse_scoring": True},
{"id": 36, "content": "在热闹的聚会上，我常常表现主动并尽情玩耍", "factor": "外向性", "reverse_scoring": False},
{"id": 37, "content": "有我在的场合一般不会冷场", "factor": "外向性", "reverse_scoring": False},
{"id": 38, "content": "我希望成为领导者而不是被领导者", "factor": "外向性", "reverse_scoring": False},
{"id": 39, "content": "在一个团体中，我希望处于领导地位", "factor": "外向性", "reverse_scoring": False},
{"id": 40, "content": "别人多认为我是一个热情和友好的人", "factor": "外向性", "reverse_scoring": False},
]
# 因子维度说明
FACTOR_DESCRIPTIONS = {
"外向性": {
"description": "反映个体神经系统的强弱和动力特征。外向性主要表现为个体在人际交往和社交活动中的倾向性，"
"包括对社交活动的兴趣、"
"对人群的态度、社交互动中的主动程度以及在群体中的影响力。高分者倾向于积极参与社交活动，乐于与人交往，善于表达自我，"
"并往往在群体中发挥领导作用；低分者则倾向于独处，不喜欢热闹的社交场合，表现出内向、安静的特征。",
"trait_words": ["热情", "活力", "社交", "主动"],
"subfactors": {
"合群性": "个体愿意与他人聚在一起，即接近人群的倾向；高分表现乐群、好交际，低分表现封闭、独处",
"热情": "个体对待别人时所表现出的态度；高分表现热情好客，低分表现冷淡",
"支配性": "个体喜欢指使、操纵他人，倾向于领导别人的特点；高分表现好强、发号施令，低分表现顺从、低调",
"活跃": "个体精力充沛，活跃、主动性等特点；高分表现活跃，低分表现安静",
},
},
"神经质": {
"description": "反映个体情绪的状态和体验内心苦恼的倾向性。这个维度主要关注个体在面对压力、"
"挫折和日常生活挑战时的情绪稳定性和适应能力。它包含了对焦虑、抑郁、愤怒等负面情绪的敏感程度，"
"以及个体对这些情绪的调节和控制能力。高分者容易体验负面情绪，对压力较为敏感，情绪波动较大；"
"低分者则表现出较强的情绪稳定性，能够较好地应对压力和挫折。",
"trait_words": ["稳定", "沉着", "从容", "坚韧"],
"subfactors": {
"焦虑": "个体体验焦虑感的个体差异；高分表现坐立不安，低分表现平静",
"抑郁": "个体体验抑郁情感的个体差异；高分表现郁郁寡欢，低分表现平静",
"敏感多疑": "个体常常关注自己的内心活动，行为和过于意识人对自己的看法、评价；高分表现敏感多疑，"
"低分表现淡定、自信",
"脆弱性": "个体在危机或困难面前无力、脆弱的特点；高分表现无能、易受伤、逃避，低分表现坚强",
"愤怒-敌意": "个体准备体验愤怒，及相关情绪的状态；高分表现暴躁易怒，低分表现平静",
},
},
"严谨性": {
"description": "反映个体在目标导向行为上的组织、坚持和动机特征。这个维度体现了个体在工作、"
"学习等目标性活动中的自我约束和行为管理能力。它涉及到个体的责任感、自律性、计划性、条理性以及完成任务的态度。"
"高分者往往表现出强烈的责任心、良好的组织能力、谨慎的决策风格和持续的努力精神；低分者则可能表现出随意性强、"
"缺乏规划、做事马虎或易放弃的特点。",
"trait_words": ["负责", "自律", "条理", "勤奋"],
"subfactors": {
"责任心": "个体对待任务和他人认真负责，以及对自己承诺的信守；高分表现有责任心、负责任，"
"低分表现推卸责任、逃避处罚",
"自我控制": "个体约束自己的能力，及自始至终的坚持性；高分表现自制、有毅力，低分表现冲动、无毅力",
"审慎性": "个体在采取具体行动前的心理状态；高分表现谨慎、小心，低分表现鲁莽、草率",
"条理性": "个体处理事务和工作的秩序，条理和逻辑性；高分表现整洁、有秩序，低分表现混乱、遗漏",
"勤奋": "个体工作和学习的努力程度及为达到目标而表现出的进取精神；高分表现勤奋、刻苦，低分表现懒散",
},
},
"开放性": {
"description": "反映个体对新异事物、新观念和新经验的接受程度，以及在思维和行为方面的创新倾向。"
"这个维度体现了个体在认知和体验方面的广度、深度和灵活性。它包括对艺术的欣赏能力、对知识的求知欲、想象力的丰富程度，"
"以及对冒险和创新的态度。高分者往往具有丰富的想象力、广泛的兴趣、开放的思维方式和创新的倾向；低分者则倾向于保守、"
"传统，喜欢熟悉和常规的事物。",
"trait_words": ["创新", "好奇", "艺术", "冒险"],
"subfactors": {
"幻想": "个体富于幻想和想象的水平；高分表现想象力丰富，低分表现想象力匮乏",
"审美": "个体对于艺术和美的敏感与热爱程度；高分表现富有艺术气息，低分表现一般对艺术不敏感",
"好奇心": "个体对未知事物的态度；高分表现兴趣广泛、好奇心浓，低分表现兴趣少、无好奇心",
"冒险精神": "个体愿意尝试有风险活动的个体差异；高分表现好冒险，低分表现保守",
"价值观念": "个体对新事物、新观念、怪异想法的态度；高分表现开放、坦然接受新事物，低分则相反",
},
},
"宜人性": {
"description": "反映个体在人际关系中的亲和倾向，体现了对他人的关心、同情和合作意愿。"
"这个维度主要关注个体与他人互动时的态度和行为特征，包括对他人的信任程度、同理心水平、"
"助人意愿以及在人际冲突中的处理方式。高分者通常表现出友善、富有同情心、乐于助人的特质，善于与他人建立和谐关系；"
"低分者则可能表现出较少的人际关注，在社交互动中更注重自身利益，较少考虑他人感受。",
"trait_words": ["友善", "同理", "信任", "合作"],
"subfactors": {
"信任": "个体对他人和/或他人言论的相信程度；高分表现信任他人，低分表现怀疑",
"体贴": "个体对别人的兴趣和需要的关注程度；高分表现体贴、温存，低分表现冷漠、不在乎",
"同情": "个体对处于不利地位的人或物的态度；高分表现富有同情心，低分表现冷漠",
},
},
}

44
src/individuality/not_using/scene.py
View File

@@ -1,44 +0,0 @@
import os
from typing import Any
import orjson
def load_scenes() -> dict[str, Any]:
"""
从JSON文件加载场景数据
Returns:
Dict: 包含所有场景的字典
"""
current_dir = os.path.dirname(os.path.abspath(__file__))
json_path = os.path.join(current_dir, "template_scene.json")
with open(json_path, encoding="utf-8") as f:
return orjson.loads(f.read())
PERSONALITY_SCENES = load_scenes()
def get_scene_by_factor(factor: str) -> dict | None:
"""
根据人格因子获取对应的情景测试
Args:
factor (str): 人格因子名称
Returns:
dict: 包含情景描述的字典
"""
return PERSONALITY_SCENES.get(factor, None)
def get_all_scenes() -> dict:
"""
获取所有情景测试
Returns:
Dict: 所有情景测试的字典
"""
return PERSONALITY_SCENES

112
src/individuality/not_using/template_scene.json
View File

@@ -1,112 +0,0 @@
{
"外向性": {
"场景1": {
"scenario": "你刚刚搬到一个新的城市工作。今天是你入职的第一天，在公司的电梯里，一位同事微笑着和你打招呼：\n\n同事：「嗨！你是新来的同事吧？我是市场部的小林。」\n\n同事看起来很友善，还主动介绍说：「待会午饭时间，我们部门有几个人准备一起去楼下新开的餐厅，你要一起来吗？可以认识一下其他同事。」",
"explanation": "这个场景通过职场社交情境，观察个体对于新环境、新社交圈的态度和反应倾向。"
},
"场景2": {
"scenario": "在大学班级群里，班长发起了一个组织班级联谊活动的投票：\n\n班长：「大家好！下周末我们准备举办一次班级联谊活动，地点在学校附近的KTV。想请大家报名参加，也欢迎大家邀请其他班级的同学！」\n\n已经有几个同学在群里积极响应，有人@你问你要不要一起参加。",
"explanation": "通过班级活动场景，观察个体对群体社交活动的参与意愿。"
},
"场景3": {
"scenario": "你在社交平台上发布了一条动态，收到了很多陌生网友的评论和私信：\n\n网友A：「你说的这个观点很有意思！想和你多交流一下。」\n\n网友B：「我也对这个话题很感兴趣，要不要建个群一起讨论？」",
"explanation": "通过网络社交场景，观察个体对线上社交的态度。"
},
"场景4": {
"scenario": "你暗恋的对象今天主动来找你：\n\n对方：「那个...我最近在准备一个演讲比赛，听说你口才很好。能不能请你帮我看看演讲稿，顺便给我一些建议？如果你有时间的话，可以一起吃个饭聊聊。」",
"explanation": "通过恋爱情境，观察个体在面对心仪对象时的社交表现。"
},
"场景5": {
"scenario": "在一次线下读书会上，主持人突然点名让你分享读后感：\n\n主持人：「听说你对这本书很有见解，能不能和大家分享一下你的想法？」\n\n现场有二十多个陌生的读书爱好者，都期待地看着你。",
"explanation": "通过即兴发言场景，观察个体的社交表现欲和公众表达能力。"
}
},
"神经质": {
"场景1": {
"scenario": "你正在准备一个重要的项目演示，这关系到你的晋升机会。就在演示前30分钟，你收到了主管发来的消息：\n\n主管：「临时有个变动，CEO也会来听你的演示。他对这个项目特别感兴趣。」\n\n正当你准备回复时，主管又发来一条：「对了，能不能把演示时间压缩到15分钟？CEO下午还有其他安排。你之前准备的是30分钟的版本对吧？」",
"explanation": "这个场景通过突发的压力情境，观察个体在面对计划外变化时的情绪反应和调节能力。"
},
"场景2": {
"scenario": "期末考试前一天晚上，你收到了好朋友发来的消息：\n\n好朋友：「不好意思这么晚打扰你...我看你平时成绩很好，能不能帮我解答几个问题？我真的很担心明天的考试。」\n\n你看了看时间，已经是晚上11点，而你原本计划的复习还没完成。",
"explanation": "通过考试压力场景，观察个体在时间紧张时的情绪管理。"
},
"场景3": {
"scenario": "你在社交媒体上发表的一个观点引发了争议，有不少人开始批评你：\n\n网友A：「这种观点也好意思说出来，真是无知。」\n\n网友B：「建议楼主先去补补课再来发言。」\n\n评论区里的负面评论越来越多，还有人开始人身攻击。",
"explanation": "通过网络争议场景，观察个体面对批评时的心理承受能力。"
},
"场景4": {
"scenario": "你和恋人约好今天一起看电影，但在约定时间前半小时，对方发来消息：\n\n恋人：「对不起，我临时有点事，可能要迟到一会儿。」\n\n二十分钟后，对方又发来消息：「可能要再等等，抱歉！」\n\n电影快要开始了，但对方还是没有出现。",
"explanation": "通过恋爱情境，观察个体对不确定性的忍耐程度。"
},
"场景5": {
"scenario": "在一次重要的小组展示中，你的组员在演示途中突然卡壳了：\n\n组员小声对你说：「我忘词了，接下来的部分是什么来着...」\n\n台下的老师和同学都在等待，气氛有些尴尬。",
"explanation": "通过公开场合的突发状况，观察个体的应急反应和压力处理能力。"
}
},
"严谨性": {
"场景1": {
"scenario": "你是团队的项目负责人，刚刚接手了一个为期两个月的重要项目。在第一次团队会议上：\n\n小王：「老大，我觉得两个月时间很充裕，我们先做着看吧，遇到问题再解决。」\n\n小张：「要不要先列个时间表？不过感觉太详细的计划也没必要，点到为止就行。」\n\n小李：「客户那边说如果能提前完成有奖励，我觉得我们可以先做快一点的部分。」",
"explanation": "这个场景通过项目管理情境，体现个体在工作方法、计划性和责任心方面的特征。"
},
"场景2": {
"scenario": "期末小组作业，组长让大家分工完成一份研究报告。在截止日期前三天：\n\n组员A：「我的部分大概写完了，感觉还行。」\n\n组员B：「我这边可能还要一天才能完成，最近太忙了。」\n\n组员C发来一份没有任何引用出处、可能存在抄袭的内容：「我写完了，你们看看怎么样？」",
"explanation": "通过学习场景，观察个体对学术规范和质量要求的重视程度。"
},
"场景3": {
"scenario": "你在一个兴趣小组的群聊中，大家正在讨论举办一次线下活动：\n\n成员A：「到时候见面就知道具体怎么玩了！」\n\n成员B：「对啊，随意一点挺好的。」\n\n成员C：「人来了自然就热闹了。」",
"explanation": "通过活动组织场景，观察个体对活动计划的态度。"
},
"场景4": {
"scenario": "你的好友小明邀请你一起参加一个重要的演出活动，他说：\n\n小明：「到时候我们就即兴发挥吧！不用排练了，我相信我们的默契。」\n\n距离演出还有三天，但节目内容、配乐和服装都还没有确定。",
"explanation": "通过演出准备场景，观察个体的计划性和对不确定性的接受程度。"
},
"场景5": {
"scenario": "在一个重要的团队项目中，你发现一个同事的工作存在明显错误：\n\n同事：「差不多就行了，反正领导也看不出来。」\n\n这个错误可能不会立即造成问题，但长期来看可能会影响项目质量。",
"explanation": "通过工作质量场景，观察个体对细节和标准的坚持程度。"
}
},
"开放性": {
"场景1": {
"scenario": "周末下午，你的好友小美兴致勃勃地给你打电话：\n\n小美：「我刚发现一个特别有意思的沉浸式艺术展！不是传统那种挂画的展览，而是把整个空间都变成了艺术品。观众要穿特制的服装，还要带上VR眼镜，好像还有AI实时互动！」\n\n小美继续说：「虽然票价不便宜，但听说体验很独特。网上评价两极分化，有人说是前所未有的艺术革新，也有人说是哗众取宠。要不要周末一起去体验一下？」",
"explanation": "这个场景通过新型艺术体验，反映个体对创新事物的接受程度和尝试意愿。"
},
"场景2": {
"scenario": "在一节创意写作课上，老师提出了一个特别的作业：\n\n老师：「下周的作业是用AI写作工具协助创作一篇小说。你们可以自由探索如何与AI合作，打破传统写作方式。」\n\n班上随即展开了激烈讨论，有人认为这是对创作的亵渎，也有人对这种新形式感到兴奋。",
"explanation": "通过新技术应用场景，观察个体对创新学习方式的态度。"
},
"场景3": {
"scenario": "在社交媒体上，你看到一个朋友分享了一种新的学习方式：\n\n「最近我在尝试'沉浸式学习'，就是完全投入到一个全新的领域。比如学习一门陌生的语言，或者尝试完全不同的职业技能。虽然过程会很辛苦，但这种打破舒适圈的感觉真的很棒！」\n\n评论区里争论不断，有人认为这种学习方式效率高，也有人觉得太激进。",
"explanation": "通过新型学习方式，观察个体对创新和挑战的态度。"
},
"场景4": {
"scenario": "你的朋友向你推荐了一种新的饮食方式：\n\n朋友：「我最近在尝试'未来食品'，比如人造肉、3D打印食物、昆虫蛋白等。这不仅对环境友好，营养也很均衡。要不要一起来尝试看看？」\n\n这个提议让你感到好奇又犹豫，你之前从未尝试过这些新型食物。",
"explanation": "通过饮食创新场景，观察个体对新事物的接受度和尝试精神。"
},
"场景5": {
"scenario": "在一次朋友聚会上，大家正在讨论未来职业规划：\n\n朋友A：「我准备辞职去做自媒体，专门介绍一些小众的文化和艺术。」\n\n朋友B：「我想去学习生物科技，准备转行做人造肉研发。」\n\n朋友C：「我在考虑加入一个区块链创业项目，虽然风险很大。」",
"explanation": "通过职业选择场景，观察个体对新兴领域的探索意愿。"
}
},
"宜人性": {
"场景1": {
"scenario": "在回家的公交车上，你遇到这样一幕：\n\n一位老奶奶颤颤巍巍地上了车，车上座位已经坐满了。她站在你旁边，看起来很疲惫。这时你听到前排两个年轻人的对话：\n\n年轻人A：「那个老太太好像站不稳，看起来挺累的。」\n\n年轻人B：「现在的老年人真是...我看她包里还有菜，肯定是去菜市场买完菜回来的，这么多人都不知道叫子女开车接送。」\n\n就在这时，老奶奶一个趔趄，差点摔倒。她扶住了扶手，但包里的东西洒了一些出来。",
"explanation": "这个场景通过公共场合的助人情境，体现个体的同理心和对他人需求的关注程度。"
},
"场景2": {
"scenario": "在班级群里，有同学发起为生病住院的同学捐款：\n\n同学A：「大家好，小林最近得了重病住院，医药费很贵，家里负担很重。我们要不要一起帮帮他？」\n\n同学B：「我觉得这是他家里的事，我们不方便参与吧。」\n\n同学C：「但是都是同学一场，帮帮忙也是应该的。」",
"explanation": "通过同学互助场景，观察个体的助人意愿和同理心。"
},
"场景3": {
"scenario": "在一个网络讨论组里，有人发布了求助信息：\n\n求助者：「最近心情很低落，感觉生活很压抑，不知道该怎么办...」\n\n评论区里已经有一些回复：\n「生活本来就是这样，想开点！」\n「你这样子太消极了，要积极面对。」\n「谁还没点烦心事啊，过段时间就好了。」",
"explanation": "通过网络互助场景，观察个体的共情能力和安慰方式。"
},
"场景4": {
"scenario": "你的朋友向你倾诉工作压力：\n\n朋友：「最近工作真的好累，感觉快坚持不下去了...」\n\n但今天你也遇到了很多烦心事，心情也不太好。",
"explanation": "通过感情关系场景，观察个体在自身状态不佳时的关怀能力。"
},
"场景5": {
"scenario": "在一次团队项目中，新来的同事小王因为经验不足，造成了一个严重的错误。在部门会议上：\n\n主管：「这个错误造成了很大的损失，是谁负责的这部分？」\n\n小王看起来很紧张，欲言又止。你知道是他造成的错误，同时你也是这个项目的共同负责人。",
"explanation": "通过职场情境，观察个体在面对他人过错时的态度和处理方式。"
}
}
}

561
src/plugin_system/apis/plugin_manage_api.py
View File

@@ -1,29 +1,46 @@
# -*- 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 +49,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 +59,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,173 +224,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。
"""
# Chatter 唯一性保护
# 首先，验证组件是否存在
component_info = component_registry.get_component_info(name, component_type)
if not component_info:
logger.error(f"尝试设置局部状态失败：未找到组件 {name} ({component_type.value})。")
return False
# 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,
}
)
# 从 plugin_info (PluginInfo) 而不是 instance (PluginBase) 获取元数据
# 收集该插件下所有组件的详细信息
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
]
# 构建单个插件的信息字典
# 元数据从 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),
@@ -267,3 +309,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

1770
src/plugin_system/core/component_registry.py
View File

File diff suppressed because it is too large Load Diff

12
src/plugin_system/core/plugin_manager.py
View File

@@ -174,13 +174,11 @@ class PluginManager:
if plugin_name not in self.loaded_plugins:
logger.warning(f"插件 {plugin_name} 未加载")
return False
plugin_instance = self.loaded_plugins[plugin_name]
plugin_info = plugin_instance.plugin_info
success = True
for component in plugin_info.components:
success &= await component_registry.remove_component(component.name, component.component_type, plugin_name)
success &= component_registry.remove_plugin_registry(plugin_name)
del self.loaded_plugins[plugin_name]
# 调用 component_registry 中统一的卸载方法
success = await component_registry.unregister_plugin(plugin_name)
if success:
# 从已加载插件中移除
del self.loaded_plugins[plugin_name]
return success
async def reload_registered_plugin(self, plugin_name: str) -> bool:

98
src/plugins/built_in/system_management/plugin.py
View File

@@ -11,6 +11,7 @@ from typing import ClassVar
from src.chat.utils.prompt_component_manager import prompt_component_manager
from src.chat.utils.prompt_params import PromptParameters
from src.plugin_system.apis import (
chat_api,
plugin_manage_api,
)
from src.plugin_system.apis.logging_api import get_logger
@@ -21,6 +22,7 @@ from src.plugin_system.base.base_plugin import BasePlugin
from src.plugin_system.base.command_args import CommandArgs
from src.plugin_system.base.component_types import (
ChatType,
ComponentType,
PermissionNodeField,
PlusCommandInfo,
)
@@ -103,6 +105,9 @@ class SystemCommand(PlusCommand):
• `/system plugin load <插件名>` - 加载指定插件
• `/system plugin reload <插件名>` - 重新加载指定插件
• `/system plugin reload_all` - 重新加载所有插件
🎯 局部控制 (需要 `system.plugin.manage.local` 权限):
• `/system plugin enable_local <名称> [group <群号> | private <QQ号>]` - 在指定会话局部启用组件
• `/system plugin disable_local <名称> [group <群号> | private <QQ号>]` - 在指定会话局部禁用组件
"""
elif target == "permission":
help_text = """📋 权限管理命令帮助
@@ -157,6 +162,10 @@ class SystemCommand(PlusCommand):
await self._reload_plugin(remaining_args[0])
elif action in ["reload_all", "重载全部"]:
await self._reload_all_plugins()
elif action in ["enable_local", "局部启用"] and len(remaining_args) >= 1:
await self._set_local_component_state(remaining_args, enabled=True)
elif action in ["disable_local", "局部禁用"] and len(remaining_args) >= 1:
await self._set_local_component_state(remaining_args, enabled=False)
else:
await self.send_text("❌ 插件管理命令不合法\n使用 /system plugin help 查看帮助")
@@ -309,7 +318,7 @@ class SystemCommand(PlusCommand):
@require_permission("prompt.view", deny_message="❌ 你没有查看提示词注入信息的权限")
async def _list_prompt_components(self):
"""列出所有已注册的提示词组件"""
components = prompt_component_manager.get_registered_prompt_component_info()
components = await prompt_component_manager.get_registered_prompt_component_info()
if not components:
await self.send_text("🧩 当前没有已注册的提示词组件")
return
@@ -391,7 +400,7 @@ class SystemCommand(PlusCommand):
@require_permission("prompt.view", deny_message="❌ 你没有查看提示词组件信息的权限")
async def _show_prompt_component_info(self, component_name: str):
"""显示特定提示词组件的详细信息"""
all_components = prompt_component_manager.get_registered_prompt_component_info()
all_components = await prompt_component_manager.get_registered_prompt_component_info()
target_component = next((comp for comp in all_components if comp.name == component_name), None)
@@ -486,6 +495,87 @@ class SystemCommand(PlusCommand):
else:
await self.send_text("⚠️ 部分插件重载失败，请检查日志。")
@require_permission("plugin.manage.local", deny_message="❌ 你没有局部管理插件组件的权限")
async def _set_local_component_state(self, args: list[str], enabled: bool):
"""在局部范围内启用或禁用一个组件"""
# 命令格式: <component_name> [group <group_id> | private <user_id>]
if not args:
action = "enable_local" if enabled else "disable_local"
await self.send_text(f"❌ 用法: /system plugin {action} <名称> [group <群号> | private <QQ号>]")
return
comp_name = args[0]
context_args = args[1:]
stream_id = self.message.chat_info.stream_id # 默认作用于当前会话
# 1. 搜索组件
found_components = plugin_manage_api.search_components_by_name(comp_name, exact_match=True)
if not found_components:
await self.send_text(f"❌ 未找到名为 '{comp_name}' 的组件。")
return
if len(found_components) > 1:
suggestions = "\n".join([f"- `{c['name']}` (类型: {c['component_type']})" for c in found_components])
await self.send_text(f"❌ 发现多个名为 '{comp_name}' 的组件，操作已取消。\n找到的组件:\n{suggestions}")
return
component_info = found_components[0]
comp_type_str = component_info["component_type"]
component_type = ComponentType(comp_type_str)
# 2. 增加禁用保护
if not enabled: # 如果是禁用操作
# 定义不可禁用的核心组件类型
protected_types = [
ComponentType.INTEREST_CALCULATOR,
ComponentType.PROMPT,
ComponentType.ROUTER,
]
if component_type in protected_types:
await self.send_text(f"❌ 无法局部禁用核心组件 '{comp_name}' ({comp_type_str})。")
return
# 3. 解析上下文
if len(context_args) >= 2:
context_type = context_args[0].lower()
context_id = context_args[1]
target_stream = None
if context_type == "group":
target_stream = chat_api.get_stream_by_group_id(
group_id=context_id,
platform=self.message.chat_info.platform
)
elif context_type == "private":
target_stream = chat_api.get_stream_by_user_id(
user_id=context_id,
platform=self.message.chat_info.platform
)
else:
await self.send_text("❌ 无效的作用域类型，请使用 'group' 或 'private'。")
return
if not target_stream:
await self.send_text(f"❌ 在当前平台找不到指定的 {context_type}: `{context_id}`。")
return
stream_id = target_stream.stream_id
# 4. 执行操作
success = plugin_manage_api.set_component_enabled_local(
stream_id=stream_id,
name=comp_name,
component_type=component_type,
enabled=enabled
)
action_text = "启用" if enabled else "禁用"
if success:
await self.send_text(f"✅ 在会话 `{stream_id}` 中，已成功将组件 `{comp_name}` ({comp_type_str}) 设置为 {action_text} 状态。")
else:
await self.send_text(f"❌ 操作失败。可能无法禁用最后一个启用的 Chatter，或组件不存在。请检查日志。")
# =================================================================
# Permission Management Section
@@ -731,4 +821,8 @@ class SystemManagementPlugin(BasePlugin):
node_name="schedule.manage",
description="定时任务管理：暂停和恢复定时任务",
),
PermissionNodeField(
node_name="plugin.manage.local",
description="局部插件管理：在指定会话中启用或禁用组件",
),
]