feat(plugin_system): 引入高级Prompt注入规则系统以取代旧注入点机制
引入了一套全新的、基于规则的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
committed by
Windpicker-owo
Windpicker-owo
parent
c96434299f
commit
92f5f9bbd9
@@ -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]
|
||||
# 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_rules.append((rule, component_class))
|
||||
|
||||
# 检查当前注入点是否匹配
|
||||
if injection_point in injection_points:
|
||||
# 获取组件类
|
||||
component_class = component_registry.get_component_class(prompt_name, ComponentType.PROMPT)
|
||||
if component_class and issubclass(component_class, BasePrompt):
|
||||
matching_components.append(component_class)
|
||||
# 根据规则的优先级进行排序，数字越小，优先级越高，越先应用
|
||||
matching_rules.sort(key=lambda x: x[0].priority)
|
||||
return matching_rules
|
||||
|
||||
return matching_components
|
||||
|
||||
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:
|
||||
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_map = {} # 存储组件实例，虽然目前没直接用，但留着总没错
|
||||
tasks = [] # 存放所有需要并行执行的 execute 异步任务
|
||||
components_to_execute = [] # 存放需要执行的组件类，用于后续结果映射
|
||||
|
||||
instance = component_class(params=params, plugin_config=plugin_config)
|
||||
tasks.append(instance.execute())
|
||||
except Exception as e:
|
||||
logger.error(f"实例化 Prompt 组件 '{component_class.prompt_name}' 失败: {e}")
|
||||
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):
|
||||
plugin_config = {}
|
||||
else:
|
||||
# 从注册表获取该组件所属插件的配置
|
||||
plugin_config = component_registry.get_plugin_config(prompt_info.plugin_name)
|
||||
|
||||
if not tasks:
|
||||
return ""
|
||||
# 实例化组件，并传入参数和插件配置
|
||||
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
|
||||
|
||||
# 并行执行所有组件
|
||||
# --- 第二步: 并行执行所有组件的 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
|
||||
|
||||
|
||||
# 创建全局单例
|
||||
|
||||
Reference in New Issue
Block a user