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

refactor(prompt): 简化和统一提示词注入查询方法

Browse Source 
将多个功能重叠的查询方法整合为两个核心方法，以提供更清晰、更灵活的 API，方便获取注入信息和规则。

- `get_injection_info` 方法取代了原有的 `get_full_injection_map` 和 `get_injections_for_prompt`。它现在支持按目标提示词进行筛选，并能通过 `detailed` 参数控制返回信息的详细程度。
- `get_injection_rules` 方法整合了 `get_all_dynamic_rules`、`get_rules_for_target` 和 `get_rules_by_component` 的功能。现在可以通过一个方法，灵活地按目标、按组件或按两者的组合来筛选注入规则。
- 已更新 `system_management` 插件中的相关命令以适配新的 API。

BREAKING CHANGE: `PromptComponentManager` 的公共 API 已更改。
移除了 `get_full_injection_map`, `get_injections_for_prompt`, `get_all_dynamic_rules`, `get_rules_for_target`, `get_rules_by_component` 方法。
请分别使用新的 `get_injection_info` 和 `get_injection_rules` 方法进行替代。
This commit is contained in:
minecraft1024a
2025-11-13 17:17:48 +08:00
parent 1e8042de00
commit ce2d1acc7c
2 changed files with 90 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -287,110 +287,117 @@ class PromptComponentManager:
components = component_registry.get_components_by_type(ComponentType.PROMPT).values()
return [info for info in components if isinstance(info, PromptInfo)]
async def get_full_injection_map(self) -> dict[str, list[dict]]:
async def get_injection_info(
self,
target_prompt: str | None = None,
detailed: bool = False,
) -> dict[str, list[dict]]:
"""
获取当前完整的注入映射图。
获取注入信息的映射图，可按目标筛选，并可控制信息的详细程度。
此方法提供了一个系统全局的注入视图，展示了每个核心提示词（target）
被哪些注入组件（source）以何种优先级注入。
- `get_injection_info()` 返回所有目标的摘要注入信息。
- `get_injection_info(target_prompt="...")` 返回指定目标的摘要注入信息。
- `get_injection_info(detailed=True)` 返回所有目标的详细注入信息。
- `get_injection_info(target_prompt="...", detailed=True)` 返回指定目标的详细注入信息。
Args:
target_prompt (str, optional): 如果指定，仅返回该目标的注入信息。
detailed (bool, optional): 如果为 True，则返回包含注入类型和内容的详细信息。
默认为 False，返回摘要信息。
Returns:
dict[str, list[dict]]: 一个字典，键是目标提示词名称，
值是按优先级排序的注入信息列表。
`[{"name": str, "priority": int, "source": str}]`
值是按优先级排序的注入信息列表。
"""
injection_map = {}
info_map = {}
async with self._lock:
# 合并所有动态规则的目标和所有核心提示词，确保所有潜在目标都被包含
all_targets = set(self._dynamic_rules.keys()) | set(self.get_core_prompts())
for target in sorted(all_targets):
# 如果指定了目标，则只处理该目标
targets_to_process = [target_prompt] if target_prompt and target_prompt in all_targets else sorted(all_targets)
for target in targets_to_process:
rules = self._dynamic_rules.get(target, {})
if not rules:
injection_map[target] = []
info_map[target] = []
continue
info_list = []
for prompt_name, (rule, _, source) in rules.items():
info_list.append({"name": prompt_name, "priority": rule.priority, "source": source})
if detailed:
info_list.append(
{
"name": prompt_name,
"priority": rule.priority,
"source": source,
"injection_type": rule.injection_type.value,
"target_content": rule.target_content,
}
)
else:
info_list.append({"name": prompt_name, "priority": rule.priority, "source": source})
# 按优先级排序后存入 map
info_list.sort(key=lambda x: x["priority"])
injection_map[target] = info_list
return injection_map
info_map[target] = info_list
return info_map
async def get_injections_for_prompt(self, target_prompt_name: str) -> list[dict]:
def get_injection_rules(
self,
target_prompt: str | None = None,
component_name: str | None = None,
) -> dict[str, dict[str, "InjectionRule"]]:
"""
获取指定核心提示词模板的所有注入信息（包含详细规则）。
获取动态注入规则，可通过目标或组件名称进行筛选。
- 不提供任何参数时，返回所有规则。
- 提供 `target_prompt` 时，仅返回注入到该目标的规则。
- 提供 `component_name` 时，仅返回由该组件定义的所有规则。
- 同时提供 `target_prompt` 和 `component_name` 时，返回满足两个条件的规则。
Args:
target_prompt_name (str): 目标核心提示词的名称。
target_prompt (str, optional): 按目标核心提示词名称筛选。
component_name (str, optional): 按注入组件名称筛选。
Returns:
list[dict]: 一个包含注入规则详细信息的列表，已按优先级排序。
"""
rules_for_target = self._dynamic_rules.get(target_prompt_name, {})
if not rules_for_target:
return []
info_list = []
for prompt_name, (rule, _, source) in rules_for_target.items():
info_list.append(
{
"name": prompt_name,
"priority": rule.priority,
"source": source,
"injection_type": rule.injection_type.value,
"target_content": rule.target_content,
}
)
info_list.sort(key=lambda x: x["priority"])
return info_list
def get_all_dynamic_rules(self) -> dict[str, dict[str, "InjectionRule"]]:
"""
获取所有当前的动态注入规则，以 InjectionRule 对象形式返回。
此方法返回一个深拷贝的规则副本，隐藏了 `content_provider` 等内部实现细节。
适合用于展示或序列化当前的规则配置。
dict[str, dict[str, InjectionRule]]: 一个深拷贝的规则字典。
结构: { "target_prompt": { "component_name": InjectionRule } }
"""
rules_copy = {}
for target, rules in self._dynamic_rules.items():
target_copy = {name: rule for name, (rule, _, _) in rules.items()}
rules_copy[target] = target_copy
# 筛选目标
targets_to_check = [target_prompt] if target_prompt else self._dynamic_rules.keys()
for target in targets_to_check:
if target not in self._dynamic_rules:
continue
rules_for_target = self._dynamic_rules[target]
target_copy = {}
# 筛选组件
if component_name:
if component_name in rules_for_target:
rule, _, _ = rules_for_target[component_name]
target_copy[component_name] = rule
else:
for name, (rule, _, _) in rules_for_target.items():
target_copy[name] = rule
if target_copy:
rules_copy[target] = target_copy
# 如果是按组件筛选且未指定目标，则需遍历所有目标
if component_name and not target_prompt:
found_rules = {}
for target, rules in self._dynamic_rules.items():
if component_name in rules:
rule, _, _ = rules[component_name]
if target not in found_rules:
found_rules[target] = {}
found_rules[target][component_name] = rule
return copy.deepcopy(found_rules)
return copy.deepcopy(rules_copy)
def get_rules_for_target(self, target_prompt: str) -> dict[str, InjectionRule]:
"""
获取所有注入到指定核心提示词的动态规则。
Args:
target_prompt (str): 目标核心提示词的名称。
Returns:
dict[str, InjectionRule]: 一个字典，键是注入组件的名称，值是 `InjectionRule` 对象。
如果找不到任何注入到该目标的规则，则返回一个空字典。
"""
target_rules = self._dynamic_rules.get(target_prompt, {})
return {name: copy.deepcopy(rule_info[0]) for name, rule_info in target_rules.items()}
def get_rules_by_component(self, component_name: str) -> dict[str, InjectionRule]:
"""
获取由指定的单个注入组件定义的所有动态规则。
Args:
component_name (str): 注入组件的名称。
Returns:
dict[str, InjectionRule]: 一个字典，键是目标核心提示词的名称，值是 `InjectionRule` 对象。
如果该组件没有定义任何注入规则，则返回一个空字典。
"""
found_rules = {}
for target, rules in self._dynamic_rules.items():
if component_name in rules:
rule_info = rules[component_name]
found_rules[target] = copy.deepcopy(rule_info[0])
return found_rules
# 创建全局单例 (Singleton)
# 在整个应用程序中，应该只使用这一个 `prompt_component_manager` 实例，

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

@@ -6,6 +6,7 @@
import re
from typing import ClassVar
from src.chat.utils.prompt_component_manager import prompt_component_manager
from src.plugin_system.apis import (
plugin_manage_api,
@@ -13,6 +14,7 @@ from src.plugin_system.apis import (
from src.plugin_system.apis.logging_api import get_logger
from src.plugin_system.apis.permission_api import permission_api
from src.plugin_system.apis.plugin_register_api import register_plugin
from src.plugin_system.apis.unified_scheduler import TriggerType, unified_scheduler
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 (
@@ -23,7 +25,6 @@ from src.plugin_system.base.component_types import (
from src.plugin_system.base.config_types import ConfigField
from src.plugin_system.base.plus_command import PlusCommand
from src.plugin_system.utils.permission_decorators import require_permission
from src.plugin_system.apis.unified_scheduler import TriggerType, unified_scheduler
logger = get_logger("SystemManagement")
@@ -266,7 +267,7 @@ class SystemCommand(PlusCommand):
@require_permission("prompt.view", deny_message="❌ 你没有查看提示词注入信息的权限")
async def _show_injection_map(self):
"""显示全局注入关系图"""
injection_map = await prompt_component_manager.get_full_injection_map()
injection_map = await prompt_component_manager.get_injection_info()
if not injection_map:
await self.send_text("📊 当前没有任何提示词注入关系")
return
@@ -312,7 +313,8 @@ class SystemCommand(PlusCommand):
@require_permission("prompt.view", deny_message="❌ 你没有查看提示词注入信息的权限")
async def _get_prompt_injection_info(self, target_name: str):
"""获取特定核心提示词的注入详情"""
injections = await prompt_component_manager.get_injections_for_prompt(target_name)
injection_info = await prompt_component_manager.get_injection_info(target_prompt=target_name, detailed=True)
injections = injection_info.get(target_name, [])
core_prompts = prompt_component_manager.get_core_prompts()
if target_name not in core_prompts: