diff --git a/src/chat/utils/prompt_component_manager.py b/src/chat/utils/prompt_component_manager.py index 1aa3b1a10..74f10a345 100644 --- a/src/chat/utils/prompt_component_manager.py +++ b/src/chat/utils/prompt_component_manager.py @@ -206,111 +206,111 @@ class PromptComponentManager: injection_map[target] = info_list return injection_map -async def get_injections_for_prompt(self, target_prompt_name: str) -> list[dict]: - """获取指定核心提示词模板的所有注入信息（增强版，包含更多细节）。""" - rules_for_target = self._dynamic_rules.get(target_prompt_name, {}) - if not rules_for_target: - return [] + async def get_injections_for_prompt(self, target_prompt_name: str) -> 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 + 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 对象形式返回。""" - rules_copy = {} - # 只返回规则对象，隐藏 provider 实现细节 - for target, rules in self._dynamic_rules.items(): - target_copy = {} - for name, (rule, _, _) in rules.items(): - target_copy[name] = rule - rules_copy[target] = target_copy - return copy.deepcopy(rules_copy) + def get_all_dynamic_rules(self) -> dict[str, dict[str, "InjectionRule"]]: + """获取所有当前的动态注入规则，以 InjectionRule 对象形式返回。""" + rules_copy = {} + # 只返回规则对象，隐藏 provider 实现细节 + for target, rules in self._dynamic_rules.items(): + target_copy = {} + for name, (rule, _, _) in rules.items(): + target_copy[name] = rule + rules_copy[target] = target_copy + return copy.deepcopy(rules_copy) -def get_rules_for_target(self, target_prompt: str) -> dict[str, InjectionRule]: - """ - 获取所有注入到指定核心提示词的动态规则。 + def get_rules_for_target(self, target_prompt: str) -> dict[str, InjectionRule]: + """ + 获取所有注入到指定核心提示词的动态规则。 - 此函数允许你查询一个特定的核心Prompt（例如 'core_prompt'）， - 并获取一个包含所有以它为目标的注入规则的字典。 + 此函数允许你查询一个特定的核心Prompt（例如 'core_prompt'）， + 并获取一个包含所有以它为目标的注入规则的字典。 - Args: - target_prompt (str): 目标核心提示词的名称。 + Args: + target_prompt (str): 目标核心提示词的名称。 - Returns: - dict[str, InjectionRule]: - 一个字典，其中： - - 键 (str): 是注入组件的名称 (例如 'persona_component')。 - - 值 (InjectionRule): 是该组件对应的 `InjectionRule` 完整对象。 - 如果找不到任何注入到该目标的规则，则返回一个空字典。 + Returns: + dict[str, InjectionRule]: + 一个字典，其中： + - 键 (str): 是注入组件的名称 (例如 'persona_component')。 + - 值 (InjectionRule): 是该组件对应的 `InjectionRule` 完整对象。 + 如果找不到任何注入到该目标的规则，则返回一个空字典。 - 示例: - >>> rules = prompt_component_manager.get_rules_for_target("core_prompt") - >>> for component_name, rule in rules.items(): - ... print(f"组件 '{component_name}' 以优先级 {rule.priority} 注入") - """ - target_rules = self._dynamic_rules.get(target_prompt, {}) - return {name: copy.deepcopy(rule_info[0]) for name, rule_info in target_rules.items()} + 示例: + >>> rules = prompt_component_manager.get_rules_for_target("core_prompt") + >>> for component_name, rule in rules.items(): + ... print(f"组件 '{component_name}' 以优先级 {rule.priority} 注入") + """ + 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]: - """ - 获取由指定的单个注入组件定义的所有动态规则。 + def get_rules_by_component(self, component_name: str) -> dict[str, InjectionRule]: + """ + 获取由指定的单个注入组件定义的所有动态规则。 - 此函数允许你查询一个特定的注入组件（例如 'persona_component'）， - 并找出它注入到了哪些核心Prompt中，以及具体是如何注入的。 + 此函数允许你查询一个特定的注入组件（例如 'persona_component'）， + 并找出它注入到了哪些核心Prompt中，以及具体是如何注入的。 - Args: - component_name (str): 注入组件的名称。 + Args: + component_name (str): 注入组件的名称。 - Returns: - dict[str, InjectionRule]: - 一个字典，其中： - - 键 (str): 是该组件注入的目标核心提示词的名称 (例如 'core_prompt')。 - - 值 (InjectionRule): 是该组件对应的 `InjectionRule` 完整对象。 - 如果该组件没有定义任何注入规则，则返回一个空字典。 + Returns: + dict[str, InjectionRule]: + 一个字典，其中： + - 键 (str): 是该组件注入的目标核心提示词的名称 (例如 'core_prompt')。 + - 值 (InjectionRule): 是该组件对应的 `InjectionRule` 完整对象。 + 如果该组件没有定义任何注入规则，则返回一个空字典。 - 示例: - >>> rules = prompt_component_manager.get_rules_by_component("persona_component") - >>> for target_name, rule in rules.items(): - ... print(f"组件 'persona_component' 注入到了 '{target_name}'") - """ - 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 + 示例: + >>> rules = prompt_component_manager.get_rules_by_component("persona_component") + >>> for target_name, rule in rules.items(): + ... print(f"组件 'persona_component' 注入到了 '{target_name}'") + """ + 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 -async def preview_prompt_injections( - self, target_prompt_name: str, params: PromptParameters -) -> str: - """ - 【预览功能】模拟应用所有注入规则，返回最终生成的模板字符串，而不实际修改任何状态。 - """ - # 确保我们从最新的模板开始 - try: - # Note: 这里我们不关心注入，只想获取最原始的模板，所以直接访问内部属性 - original_prompt = global_prompt_manager._prompts.get(target_prompt_name) - if not original_prompt: + async def preview_prompt_injections( + self, target_prompt_name: str, params: PromptParameters + ) -> str: + """ + 【预览功能】模拟应用所有注入规则，返回最终生成的模板字符串，而不实际修改任何状态。 + """ + # 确保我们从最新的模板开始 + try: + # Note: 这里我们不关心注入，只想获取最原始的模板，所以直接访问内部属性 + original_prompt = global_prompt_manager._prompts.get(target_prompt_name) + if not original_prompt: + logger.warning(f"无法预览 '{target_prompt_name}'，因为找不到这个核心 Prompt。") + return f"Error: Prompt '{target_prompt_name}' not found." + original_template = original_prompt.template + except KeyError: logger.warning(f"无法预览 '{target_prompt_name}'，因为找不到这个核心 Prompt。") return f"Error: Prompt '{target_prompt_name}' not found." - original_template = original_prompt.template - except KeyError: - 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) + # 直接调用核心注入逻辑 + return await self.apply_injections(target_prompt_name, original_template, params) # 创建全局单例 prompt_component_manager = PromptComponentManager()