From ee7a37ce7033b68e2064f6906cfec200cd88e97b Mon Sep 17 00:00:00 2001 From: minecraft1024a Date: Fri, 24 Oct 2025 19:51:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(plugin=5Fsystem):=20=E5=BC=95=E5=85=A5?= =?UTF-8?q?=E9=AB=98=E7=BA=A7Prompt=E6=B3=A8=E5=85=A5=E8=A7=84=E5=88=99?= =?UTF-8?q?=E7=B3=BB=E7=BB=9F=E4=BB=A5=E5=8F=96=E4=BB=A3=E6=97=A7=E6=B3=A8?= =?UTF-8?q?=E5=85=A5=E7=82=B9=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 引入了一套全新的、基于规则的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` 以获得更强的控制力和未来的兼容性。 --- plugins/hello_world_plugin/plugin.py | 3 +- src/chat/utils/prompt.py | 33 ++-- src/chat/utils/prompt_component_manager.py | 170 ++++++++++++------ src/plugin_system/base/base_prompt.py | 20 ++- src/plugin_system/base/component_types.py | 53 +++++- .../affinity_flow_chatter/plan_filter.py | 2 +- 6 files changed, 196 insertions(+), 85 deletions(-) diff --git a/plugins/hello_world_plugin/plugin.py b/plugins/hello_world_plugin/plugin.py index 44c3ceb39..01e819463 100644 --- a/plugins/hello_world_plugin/plugin.py +++ b/plugins/hello_world_plugin/plugin.py @@ -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 diff --git a/src/chat/utils/prompt.py b/src/chat/utils/prompt.py index c4d23d5f4..a3b8455e2 100644 --- a/src/chat/utils/prompt.py +++ b/src/chat/utils/prompt.py @@ -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) diff --git a/src/chat/utils/prompt_component_manager.py b/src/chat/utils/prompt_component_manager.py index 765a81416..c1fb92e13 100644 --- a/src/chat/utils/prompt_component_manager.py +++ b/src/chat/utils/prompt_component_manager.py @@ -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 # 创建全局单例 diff --git a/src/plugin_system/base/base_prompt.py b/src/plugin_system/base/base_prompt.py index ff6556d59..ca6d56040 100644 --- a/src/plugin_system/base/base_prompt.py +++ b/src/plugin_system/base/base_prompt.py @@ -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, ) diff --git a/src/plugin_system/base/component_types.py b/src/plugin_system/base/component_types.py index 5db5fdeb3..d92383788 100644 --- a/src/plugin_system/base/component_types.py +++ b/src/plugin_system/base/component_types.py @@ -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: diff --git a/src/plugins/built_in/affinity_flow_chatter/plan_filter.py b/src/plugins/built_in/affinity_flow_chatter/plan_filter.py index 8f2d219a0..b51e13145 100644 --- a/src/plugins/built_in/affinity_flow_chatter/plan_filter.py +++ b/src/plugins/built_in/affinity_flow_chatter/plan_filter.py @@ -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)