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

feat(plugin_system): 引入高级Prompt注入规则系统以取代旧注入点机制

Browse Source 
引入了一套全新的、基于规则的Prompt注入系统，以取代原有的 `injection_point` 机制。这套新系统提供了更强大、更灵活的Prompt内容注入能力。

主要变更包括：
- **引入 `InjectionRule` 和 `InjectionType`**：定义了注入规则的数据结构和注入类型（如 `PREPEND`, `APPEND`, `REPLACE`, `REMOVE`, `INSERT_AFTER`），允许插件开发者精确控制注入行为。
- **重构 `PromptComponentManager`**：核心逻辑从简单地拼接字符串 (`execute_components_for`) 重构为按优先级应用注入规则 (`apply_injections`)，支持正则表达式匹配和更复杂的注入操作。
- **向后兼容**：`PromptInfo` 中增加了兼容逻辑，能自动将旧的 `injection_point` 定义转换为新的 `injection_rules`，确保现有插件无需立即修改即可正常工作。
- **更新 `BasePrompt`**：废弃了 `injection_point` 属性，并推荐使用新的 `injection_rules` 属性。
- **更新示例插件**：`hello_world_plugin` 已更新，展示了新注入规则的使用方法。

BREAKING CHANGE: `BasePrompt` 中的 `injection_point` 属性已被废弃。虽然目前存在向后兼容逻辑，但未来版本将移除该属性。所有Prompt组件都应迁移至使用 `injection_rules` 以获得更强的控制力和未来的兼容性。
This commit is contained in:
minecraft1024a
2025-10-24 19:51:35 +08:00
parent 0737f84fd4
commit ee7a37ce70
6 changed files with 196 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
plugins/hello_world_plugin/plugin.py
View File

@@ -19,6 +19,7 @@ from src.plugin_system import (
ToolParamType,
register_plugin,
)
from src.plugin_system.base.component_types import InjectionRule,InjectionType
from src.plugin_system.base.base_event import HandlerResult
logger = get_logger("hello_world_plugin")
@@ -188,7 +189,7 @@ class WeatherPrompt(BasePrompt):
prompt_name = "weather_info_prompt"
prompt_description = "向Planner注入当前天气信息，以丰富对话上下文。"
injection_point = "planner_prompt"
injection_rules = [InjectionRule(target_prompt="planner_prompt", injection_type=InjectionType.REPLACE, target_content="## 可用动作列表")]
async def execute(self) -> str:
# 在实际应用中，这里可以调用天气API

33
src/chat/utils/prompt.py
View File

@@ -196,19 +196,19 @@ class PromptManager:
# 确保我们有有效的parameters实例用于注入逻辑
params_for_injection = parameters or original_prompt.parameters
# 从组件管理器获取需要注入的内容
components_prefix = await prompt_component_manager.execute_components_for(
injection_point=original_prompt.name, params=params_for_injection
# 应用所有匹配的注入规则，获取修改后的模板
modified_template = await prompt_component_manager.apply_injections(
target_prompt_name=original_prompt.name,
original_template=original_prompt.template,
params=params_for_injection,
)
# 如果有内容需要注入
if components_prefix:
logger.info(f"为'{name}'注入插件内容: \n{components_prefix}")
# 将注入内容与原始模板拼接
new_template = f"{components_prefix}\n\n{original_prompt.template}"
# 创建一个新的、临时的Prompt实例。`should_register=False`是关键，
# 它防止了这个临时版本污染全局或上下文注册表。
# 如果模板被修改了，就创建一个新的临时Prompt实例
if modified_template != original_prompt.template:
logger.info(f"为'{name}'应用了Prompt注入规则")
# 创建一个新的临时Prompt实例，不进行注册
temp_prompt = Prompt(
template=new_template,
template=modified_template,
name=original_prompt.name,
parameters=original_prompt.parameters,
should_register=False, # 确保不重新注册
@@ -1238,13 +1238,12 @@ async def create_prompt_async(
# 如果提供了名称，就尝试为它注入插件内容
if name:
components_prefix = await prompt_component_manager.execute_components_for(
injection_point=name, params=final_params
modified_template = await prompt_component_manager.apply_injections(
target_prompt_name=name, original_template=template, params=final_params
)
if components_prefix:
logger.debug(f"为'{name}'注入插件内容: \n{components_prefix}")
# 将注入内容拼接到原始模板的前面
template = f"{components_prefix}\n\n{template}"
if modified_template != template:
logger.debug(f"为'{name}'应用了Prompt注入规则")
template = modified_template
# 使用可能已被修改的模板来创建最终的Prompt实例
prompt = create_prompt(template, name, final_params)

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

@@ -1,9 +1,11 @@
import asyncio
import re
from typing import Type
from src.chat.utils.prompt_params import PromptParameters
from src.common.logger import get_logger
from src.plugin_system.base.base_prompt import BasePrompt
from src.plugin_system.base.component_types import ComponentType, PromptInfo
from src.plugin_system.base.component_types import ComponentType, InjectionRule, InjectionType, PromptInfo
from src.plugin_system.core.component_registry import component_registry
logger = get_logger("prompt_component_manager")
@@ -19,89 +21,143 @@ class PromptComponentManager:
3. 提供一个接口，以便在构建核心Prompt时，能够获取并执行所有相关的组件。
"""
def get_components_for(self, injection_point: str) -> list[type[BasePrompt]]:
def _get_rules_for(self, target_prompt_name: str) -> list[tuple[InjectionRule, Type[BasePrompt]]]:
"""
获取指定注入点的所有已注册组件类。
获取指定目标Prompt的所有注入规则及其关联的组件类。
Args:
injection_point: 目标Prompt的名称。
target_prompt_name (str): 目标 Prompt 的名称。
Returns:
list[Type[BasePrompt]]: 与该注入点关联的组件类列表。
list[tuple[InjectionRule, Type[BasePrompt]]]: 一个元组列表，
每个元组包含一个注入规则和其对应的 Prompt 组件类，并已根据优先级排序。
"""
# 从组件注册中心获取所有启用的Prompt组件
# 从注册表中获取所有已启用的 PROMPT 类型的组件
enabled_prompts = component_registry.get_enabled_components_by_type(ComponentType.PROMPT)
matching_rules = []
matching_components: list[type[BasePrompt]] = []
# 遍历所有启用的 Prompt 组件，查找与目标 Prompt 相关的注入规则
for prompt_name, prompt_info in enabled_prompts.items():
# 确保 prompt_info 是 PromptInfo 类型
if not isinstance(prompt_info, PromptInfo):
continue
# 获取注入点信息
injection_points = prompt_info.injection_point
if isinstance(injection_points, str):
injection_points = [injection_points]
# 检查当前注入点是否匹配
if injection_point in injection_points:
# 获取组件类
# prompt_info.injection_rules 已经经过了后向兼容处理，确保总是列表
for rule in prompt_info.injection_rules:
# 如果规则的目标是当前指定的 Prompt
if rule.target_prompt == target_prompt_name:
# 获取该规则对应的组件类
component_class = component_registry.get_component_class(prompt_name, ComponentType.PROMPT)
# 确保获取到的确实是一个 BasePrompt 的子类
if component_class and issubclass(component_class, BasePrompt):
matching_components.append(component_class)
matching_rules.append((rule, component_class))
return matching_components
# 根据规则的优先级进行排序，数字越小，优先级越高，越先应用
matching_rules.sort(key=lambda x: x[0].priority)
return matching_rules
async def execute_components_for(self, injection_point: str, params: PromptParameters) -> str:
async def apply_injections(
self, target_prompt_name: str, original_template: str, params: PromptParameters
) -> str:
"""
实例化并执行指定注入点的所有组件，然后将它们的输出拼接成一个字符串。
获取、实例化并执行所有相关组件，然后根据注入规则修改原始模板。
这是一个三步走的过程：
1. 实例化所有需要执行的组件。
2. 并行执行它们的 `execute` 方法以获取注入内容。
3. 按照优先级顺序，将内容注入到原始模板中。
Args:
injection_point: 目标Prompt的名称。
params: 用于初始化组件的 PromptParameters 对象。
target_prompt_name (str): 目标 Prompt 的名称。
original_template (str): 原始的、未经修改的 Prompt 模板字符串。
params (PromptParameters): 传递给 Prompt 组件实例的参数。
Returns:
str: 所有相关组件生成的、用换行符连接的文本内容。
str: 应用了所有注入规则后，修改过的 Prompt 模板字符串。
"""
component_classes = self.get_components_for(injection_point)
if not component_classes:
return ""
rules_with_classes = self._get_rules_for(target_prompt_name)
# 如果没有找到任何匹配的规则，就直接返回原始模板，啥也不干
if not rules_with_classes:
return original_template
tasks = []
for component_class in component_classes:
# --- 第一步: 实例化所有需要执行的组件 ---
instance_map = {} # 存储组件实例，虽然目前没直接用，但留着总没错
tasks = [] # 存放所有需要并行执行的 execute 异步任务
components_to_execute = [] # 存放需要执行的组件类，用于后续结果映射
for rule, component_class in rules_with_classes:
# 如果注入类型是 REMOVE，那就不需要执行组件了，因为它不产生内容
if rule.injection_type != InjectionType.REMOVE:
try:
# 从注册中心获取组件信息
# 获取组件的元信息，主要是为了拿到插件名称来读取插件配置
prompt_info = component_registry.get_component_info(
component_class.prompt_name, ComponentType.PROMPT
)
if not isinstance(prompt_info, PromptInfo):
logger.warning(f"找不到 Prompt 组件 '{component_class.prompt_name}' 的信息，无法获取插件配置")
plugin_config = {}
else:
# 从注册表获取该组件所属插件的配置
plugin_config = component_registry.get_plugin_config(prompt_info.plugin_name)
# 实例化组件，并传入参数和插件配置
instance = component_class(params=params, plugin_config=plugin_config)
instance_map[component_class.prompt_name] = instance
# 将组件的 execute 方法作为一个任务添加到列表中
tasks.append(instance.execute())
components_to_execute.append(component_class)
except Exception as e:
logger.error(f"实例化 Prompt 组件 '{component_class.prompt_name}' 失败: {e}")
# 即使失败，也添加一个立即完成的空任务，以保持与其他任务的索引同步
tasks.append(asyncio.create_task(asyncio.sleep(0, result=e))) # type: ignore
if not tasks:
return ""
# 并行执行所有组件
# --- 第二步: 并行执行所有组件的 execute 方法 ---
# 使用 asyncio.gather 来同时运行所有任务，提高效率
results = await asyncio.gather(*tasks, return_exceptions=True)
# 创建一个从组件名到执行结果的映射，方便后续查找
result_map = {
components_to_execute[i].prompt_name: res
for i, res in enumerate(results)
if not isinstance(res, Exception) # 只包含成功的结果
}
# 单独处理并记录执行失败的组件
for i, res in enumerate(results):
if isinstance(res, Exception):
logger.error(f"执行 Prompt 组件 '{components_to_execute[i].prompt_name}' 失败: {res}")
# 过滤掉执行失败的结果和空字符串
valid_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
logger.error(f"执行 Prompt 组件 '{component_classes[i].prompt_name}' 失败: {result}")
elif result and isinstance(result, str) and result.strip():
valid_results.append(result.strip())
# --- 第三步: 按优先级顺序应用注入规则 ---
modified_template = original_template
for rule, component_class in rules_with_classes:
# 从结果映射中获取该组件生成的内容
content = result_map.get(component_class.prompt_name)
# 使用换行符拼接所有有效结果
return "\n".join(valid_results)
try:
if rule.injection_type == InjectionType.PREPEND:
if content:
modified_template = f"{content}\n{modified_template}"
elif rule.injection_type == InjectionType.APPEND:
if content:
modified_template = f"{modified_template}\n{content}"
elif rule.injection_type == InjectionType.REPLACE:
# 使用正则表达式替换目标内容
if content and rule.target_content:
modified_template = re.sub(rule.target_content, str(content), modified_template)
elif rule.injection_type == InjectionType.INSERT_AFTER:
# 在匹配到的内容后面插入
if content and rule.target_content:
# re.sub a little trick: \g<0> represents the entire matched string
replacement = f"\\g<0>\n{content}"
modified_template = re.sub(rule.target_content, replacement, modified_template)
elif rule.injection_type == InjectionType.REMOVE:
# 使用正则表达式移除目标内容
if rule.target_content:
modified_template = re.sub(rule.target_content, "", modified_template)
except re.error as e:
logger.error(
f"在为 '{component_class.prompt_name}' 应用规则时发生正则错误: {e} (pattern: '{rule.target_content}')"
)
except Exception as e:
logger.error(f"应用 Prompt 注入规则 '{rule}' 失败: {e}")
return modified_template
# 创建全局单例

20
src/plugin_system/base/base_prompt.py
View File

@@ -3,7 +3,7 @@ from typing import Any
from src.chat.utils.prompt_params import PromptParameters
from src.common.logger import get_logger
from src.plugin_system.base.component_types import ComponentType, PromptInfo
from src.plugin_system.base.component_types import ComponentType, InjectionRule, PromptInfo
logger = get_logger("base_prompt")
@@ -16,7 +16,7 @@ class BasePrompt(ABC):
子类可以通过类属性定义其行为：
- prompt_name: Prompt组件的唯一名称。
- injection_point: 指定要注入的目标Prompt名称（或名称列表）。
- injection_rules: 定义注入规则的列表。
"""
prompt_name: str = ""
@@ -24,11 +24,15 @@ class BasePrompt(ABC):
prompt_description: str = ""
"""Prompt组件的描述"""
# 定义此组件希望注入到哪个或哪些核心Prompt中
# 可以是一个字符串（单个目标）或字符串列表（多个目标）
# 例如: "planner_prompt" 或 ["s4u_style_prompt", "normal_style_prompt"]
injection_point: str | list[str] = ""
"""要注入的目标Prompt名称或列表"""
# 定义此组件希望如何注入到核心Prompt中
# 这是一个 InjectionRule 对象的列表，可以实现复杂的注入逻辑
# 例如: [InjectionRule(target_prompt="planner_prompt", injection_type=InjectionType.APPEND, priority=50)]
injection_rules: list[InjectionRule] = []
"""定义注入规则的列表"""
# 旧的注入点定义，用于向后兼容。如果定义了这个，它将被自动转换为 injection_rules。
injection_point: str | list[str] | None = None
"""[已废弃] 要注入的目标Prompt名称或列表，请使用 injection_rules"""
def __init__(self, params: PromptParameters, plugin_config: dict | None = None):
"""初始化Prompt组件
@@ -87,9 +91,11 @@ class BasePrompt(ABC):
if not cls.prompt_name:
raise ValueError("Prompt组件必须定义 'prompt_name' 类属性。")
# 同时传递新旧两种定义，PromptInfo的__post_init__将处理兼容性问题
return PromptInfo(
name=cls.prompt_name,
component_type=ComponentType.PROMPT,
description=cls.prompt_description,
injection_rules=cls.injection_rules,
injection_point=cls.injection_point,
)

53
src/plugin_system/base/component_types.py
View File

@@ -2,6 +2,38 @@ from dataclasses import dataclass, field
from enum import Enum
from typing import Any
class InjectionType(Enum):
"""Prompt注入类型枚举"""
PREPEND = "prepend" # 在开头添加
APPEND = "append" # 在末尾添加
REPLACE = "replace" # 替换指定内容
REMOVE = "remove" # 删除指定内容
INSERT_AFTER = "insert_after" # 在指定内容之后插入
def __str__(self) -> str:
return self.value
@dataclass
class InjectionRule:
"""Prompt注入规则"""
target_prompt: str # 目标Prompt的名称
injection_type: InjectionType = InjectionType.PREPEND # 注入类型
priority: int = 100 # 优先级，数字越小越先执行
target_content: str | None = None # 用于REPLACE、REMOVE和INSERT_AFTER操作的目标内容（支持正则表达式）
def __post_init__(self):
if self.injection_type in [
InjectionType.REPLACE,
InjectionType.REMOVE,
InjectionType.INSERT_AFTER,
] and self.target_content is None:
raise ValueError(f"'{self.injection_type.value}'类型的注入规则必须提供 'target_content'。")
from maim_message import Seg
from src.llm_models.payload_content.tool_option import ToolCall as ToolCall
@@ -271,13 +303,30 @@ class EventInfo(ComponentInfo):
class PromptInfo(ComponentInfo):
"""Prompt组件信息"""
injection_point: str | list[str] = ""
"""要注入的目标Prompt名称或列表"""
injection_rules: list[InjectionRule] = field(default_factory=list)
"""定义此组件如何注入到其他Prompt中"""
# 旧的injection_point，用于向后兼容
injection_point: str | list[str] | None = None
def __post_init__(self):
super().__post_init__()
self.component_type = ComponentType.PROMPT
# 向后兼容逻辑：如果定义了旧的 injection_point，则自动转换为新的 injection_rules
if self.injection_point:
if not self.injection_rules: # 仅当rules为空时转换
points = []
if isinstance(self.injection_point, str):
points.append(self.injection_point)
elif isinstance(self.injection_point, list):
points = self.injection_point
for point in points:
self.injection_rules.append(InjectionRule(target_prompt=point))
# 转换后可以清空旧字段，避免混淆
self.injection_point = None
@dataclass
class PluginInfo:

2
src/plugins/built_in/affinity_flow_chatter/plan_filter.py
View File

@@ -60,7 +60,7 @@ class ChatterPlanFilter:
prompt, used_message_id_list = await self._build_prompt(plan)
plan.llm_prompt = prompt
if global_config.debug.show_prompt:
logger.debug(f"规划器原始提示词:{prompt}")
logger.info(f"规划器原始提示词:{prompt}") #叫你不要改你耳朵聋吗😡😡😡😡😡
llm_content, _ = await self.planner_llm.generate_response_async(prompt=prompt)