docs(prompt): 为 PromptComponentManager 植入详尽的代码文档与注释

为了让提示词注入系统的核心 `PromptComponentManager` 变得更加透明和易于理解，本次提交对代码进行了全面的文档化。此举旨在降低新接手者的认知门槛，并为未来的维护工作铺平道路。

具体而言，我们为类和所有公开方法添加了详细的文档字符串，阐明了其设计哲学、职责边界和参数约定。同时，在关键代码块（如规则数据结构、内容提供者闭包、并发锁的使用）旁增加了内联注释，揭示了其背后的实现考量和工作机制。

src/chat/utils/prompt_component_manager.py

@@ -13,34 +13,51 @@ from src.plugin_system.core.component_registry import component_registry
 logger = get_logger("prompt_component_manager")
 
 
-
 class PromptComponentManager:
     """
-    统一的、动态的、可观测的提示词管理中心。
+    一个统一的、动态的、可观测的提示词组件管理中心。
 
-    该管理器负责：
-    1. 在启动时，将所有 `BasePrompt` 组件的静态 `injection_rules` 加载为默认的动态规则。
-    2. 提供 API 以在运行时动态地添加、更新、移除注入规则。
-    3. 提供查询 API 以观测系统当前的完整注入状态。
-    4. 在构建核心 Prompt 时，根据统一的规则集应用注入，修改模板。
+    该管理器是整个提示词动态注入系统的核心，它负责：
+    1.  **规则加载**: 在系统启动时，自动扫描所有已注册的 `BasePrompt` 组件，
+        并将其静态定义的 `injection_rules` 加载为默认的动态规则。
+    2.  **动态管理**: 提供线程安全的 API，允许在运行时动态地添加、更新或移除注入规则，
+        使得提示词的结构可以被实时调整。
+    3.  **状态观测**: 提供丰富的查询 API，用于观测系统当前完整的注入状态，
+        例如查询所有注入到特定目标的规则、或查询某个组件定义的所有规则。
+    4.  **注入应用**: 在构建核心 Prompt 时，根据统一的、按优先级排序的规则集，
+        动态地修改和装配提示词模板，实现灵活的提示词组合。
     """
 
     def __init__(self):
-        """初始化管理器。"""
-        # _dynamic_rules 存储统一的注入规则
-        # 结构: { "target_prompt_name": { "prompt_component_name": (InjectionRule, content_provider, source) } }
+        """初始化管理器实例。"""
+        # _dynamic_rules 是管理器的核心状态，存储所有注入规则。
+        # 结构: {
+        #   "target_prompt_name": {
+        #     "prompt_component_name": (InjectionRule, content_provider, source)
+        #   }
+        # }
+        # content_provider 是一个异步函数，用于在应用规则时动态生成注入内容。
+        # source 记录了规则的来源（例如 "static_default" 或 "runtime"）。
         self._dynamic_rules: dict[str, dict[str, tuple[InjectionRule, Callable[..., Awaitable[str]], str]]] = {}
-        self._lock = asyncio.Lock()
-        self._initialized = False
+        self._lock = asyncio.Lock()  # 使用异步锁确保对 _dynamic_rules 的并发访问安全。
+        self._initialized = False  # 标记静态规则是否已加载，防止重复加载。
+
+    # --- 核心生命周期与初始化 ---
 
     def load_static_rules(self):
         """
-        在系统启动时被调用，扫描所有已注册的 Prompt 组件，
-        将其静态的 `injection_rules` 转换为动态规则并加载到管理器中。
+        在系统启动时加载所有静态注入规则。
+
+        该方法会扫描所有已在 `component_registry` 中注册并启用的 Prompt 组件，
+        将其类变量 `injection_rules` 转换为管理器的动态规则。
+        这确保了所有插件定义的默认注入行为在系统启动时就能生效。
+        此操作是幂等的，一旦初始化完成就不会重复执行。
         """
         if self._initialized:
             return
         logger.info("正在加载静态 Prompt 注入规则...")
+
+        # 从组件注册表中获取所有已启用的 Prompt 组件
         enabled_prompts = component_registry.get_enabled_components_by_type(ComponentType.PROMPT)
 
         for prompt_name, prompt_info in enabled_prompts.items():
@@ -53,65 +70,95 @@ class PromptComponentManager:
                 continue
 
             def create_provider(cls: type[BasePrompt]) -> Callable[[PromptParameters], Awaitable[str]]:
-                """为静态组件创建一个内容提供者闭包。"""
+                """
+                为静态组件创建一个内容提供者闭包 (Content Provider Closure)。
+
+                这个闭包捕获了组件的类 `cls`，并返回一个标准的 `content_provider` 异步函数。
+                当 `apply_injections` 需要内容时，它会调用这个函数。
+                函数内部会实例化组件，并执行其 `execute` 方法来获取注入内容。
+
+                Args:
+                    cls (type[BasePrompt]): 需要为其创建提供者的 Prompt 组件类。
+
+                Returns:
+                    Callable[[PromptParameters], Awaitable[str]]: 一个符合管理器标准的异步内容提供者。
+                """
 
                 async def content_provider(params: PromptParameters) -> str:
+                    """实际执行内容生成的异步函数。"""
                     try:
+                        # 从注册表获取最新的组件信息，包括插件配置
                         p_info = component_registry.get_component_info(cls.prompt_name, ComponentType.PROMPT)
                         plugin_config = {}
                         if isinstance(p_info, PromptInfo):
                             plugin_config = component_registry.get_plugin_config(p_info.plugin_name)
 
+                        # 实例化组件并执行
                         instance = cls(params=params, plugin_config=plugin_config)
                         result = await instance.execute()
                         return str(result) if result is not None else ""
                     except Exception as e:
                         logger.error(f"执行静态规则提供者 '{cls.prompt_name}' 时出错: {e}", exc_info=True)
-                        return ""
+                        return ""  # 出错时返回空字符串，避免影响主流程
 
                 return content_provider
 
+            # 为该组件的每条静态注入规则创建并注册一个动态规则
             for rule in prompt_info.injection_rules:
                 provider = create_provider(component_class)
                 target_rules = self._dynamic_rules.setdefault(rule.target_prompt, {})
                 target_rules[prompt_name] = (rule, provider, "static_default")
-        self._initialized = True
-        logger.info("静态 Prompt 注入规则加载完成。")
 
+        self._initialized = True
+        logger.info(f"静态 Prompt 注入规则加载完成，共处理 {len(enabled_prompts)} 个组件。")
+
+    # --- 运行时规则管理 API ---
 
     async def add_injection_rule(
         self,
         prompt_name: str,
         rule: InjectionRule,
-        content_provider: Callable[..., Awaitable[str]] | None = None,
+        content_provider: Callable[..., Awaitable[str]],
         source: str = "runtime",
     ) -> bool:
         """
         动态添加或更新一条注入规则。
 
-        Args:
-            prompt_name (str): 动态注入组件名称。
-            rule (InjectionRule): 注入规则。
-            content_provider (Callable | None, optional): 动态内容提供者。
-                如果提供，apply_injections 时会调用此函数获取注入内容。
-                函数签名应为: async def provider(params: "PromptParameters") -> str
-            source (str, optional): 规则来源，默认为 "runtime"。
-        """
-        if not content_provider:
-            logger.error(f"为 '{prompt_name}' 添加动态注入规则失败：必须提供 content_provider。")
-            return False
+        此方法允许在系统运行时，由外部逻辑（如插件、命令）向管理器中添加新的注入行为。
+        如果已存在同名组件针对同一目标的规则，此方法会覆盖旧规则。
 
+        Args:
+            prompt_name (str): 动态注入组件的唯一名称。
+            rule (InjectionRule): 描述注入行为的规则对象。
+            content_provider (Callable[..., Awaitable[str]]):
+                一个异步函数，用于在应用注入时动态生成内容。
+                函数签名应为: `async def provider(params: "PromptParameters") -> str`
+            source (str, optional): 规则的来源标识，默认为 "runtime"。
+
+        Returns:
+            bool: 如果成功添加或更新，则返回 True。
+        """
         async with self._lock:
             target_rules = self._dynamic_rules.setdefault(rule.target_prompt, {})
             target_rules[prompt_name] = (rule, content_provider, source)
-        logger.info(f"成功添加/更新注入规则: '{prompt_name}' -> '{rule.target_prompt}'")
+        logger.info(f"成功添加/更新注入规则: '{prompt_name}' -> '{rule.target_prompt}' (来源: {source})")
         return True
 
     async def remove_injection_rule(self, prompt_name: str, target_prompt: str) -> bool:
-        """移除一条动态注入规则。"""
+        """
+        移除一条动态注入规则。
+
+        Args:
+            prompt_name (str): 要移除的注入组件的名称。
+            target_prompt (str): 该组件注入的目标核心提示词名称。
+
+        Returns:
+            bool: 如果成功移除，则返回 True；如果规则不存在，则返回 False。
+        """
         async with self._lock:
             if target_prompt in self._dynamic_rules and prompt_name in self._dynamic_rules[target_prompt]:
                 del self._dynamic_rules[target_prompt][prompt_name]
+                # 如果目标下已无任何规则，则清理掉这个键
                 if not self._dynamic_rules[target_prompt]:
                     del self._dynamic_rules[target_prompt]
                 logger.info(f"成功移除注入规则: '{prompt_name}' from '{target_prompt}'")
@@ -119,34 +166,54 @@ class PromptComponentManager:
         logger.warning(f"尝试移除注入规则失败: 未找到 '{prompt_name}' on '{target_prompt}'")
         return False
 
+    # --- 核心注入逻辑 ---
+
     async def apply_injections(
         self, target_prompt_name: str, original_template: str, params: PromptParameters
     ) -> str:
         """
         【核心方法】根据目标名称，应用所有匹配的注入规则，返回修改后的模板。
+
+        这是提示词构建流程中的关键步骤。它会执行以下操作：
+        1. 检查并确保静态规则已加载。
+        2. 获取所有注入到 `target_prompt_name` 的规则。
+        3. 按照规则的 `priority` 属性进行升序排序，优先级数字越小越先应用。
+        4. 依次执行每个规则的 `content_provider` 来异步获取注入内容。
+        5. 根据规则的 `injection_type` (如 PREPEND, APPEND, REPLACE 等) 将内容应用到模板上。
+
+        Args:
+            target_prompt_name (str): 目标核心提示词的名称。
+            original_template (str): 未经修改的原始提示词模板。
+            params (PromptParameters): 当前请求的参数，会传递给 `content_provider`。
+
+        Returns:
+            str: 应用了所有注入规则后，最终生成的提示词模板字符串。
         """
         if not self._initialized:
             self.load_static_rules()
 
-        # 1. 从 _dynamic_rules 中获取所有相关的规则
+        # 步骤 1: 获取所有指向当前目标的规则
+        # 使用 .values() 获取 (rule, provider, source) 元组列表
         rules_for_target = list(self._dynamic_rules.get(target_prompt_name, {}).values())
         if not rules_for_target:
             return original_template
 
-        # 2. 按优先级排序
+        # 步骤 2: 按优先级排序，数字越小越优先
         rules_for_target.sort(key=lambda x: x[0].priority)
 
-        # 3. 依次执行内容提供者 (content_provider) 并拼接模板
+        # 步骤 3: 依次执行内容提供者并根据注入类型修改模板
         modified_template = original_template
         for rule, provider, source in rules_for_target:
             content = ""
+            # 对于非 REMOVE 类型的注入，需要先获取内容
             if rule.injection_type != InjectionType.REMOVE:
                 try:
                     content = await provider(params)
                 except Exception as e:
-                    logger.error(f"执行规则 '{rule}' (来源: {source}) 的内容提供者时失败: {e}")
-                    continue  # 跳过失败的 provider
+                    logger.error(f"执行规则 '{rule}' (来源: {source}) 的内容提供者时失败: {e}", exc_info=True)
+                    continue  # 跳过失败的 provider，不中断整个流程
 
+            # 应用注入逻辑
             try:
                 if rule.injection_type == InjectionType.PREPEND:
                     if content:
@@ -155,10 +222,12 @@ class PromptComponentManager:
                     if content:
                         modified_template = f"{modified_template}\n{content}"
                 elif rule.injection_type == InjectionType.REPLACE:
+                    # 只有在 content 不为 None 且 target_content 有效时才执行替换
                     if content is not None 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:
+                        # 使用 `\g<0>` 在正则匹配的整个内容后添加新内容
                         replacement = f"\\g<0>\n{content}"
                         modified_template = re.sub(rule.target_content, replacement, modified_template)
                 elif rule.injection_type == InjectionType.REMOVE:
@@ -167,11 +236,41 @@ class PromptComponentManager:
             except re.error as e:
                 logger.error(f"应用规则时发生正则错误: {e} (pattern: '{rule.target_content}')")
             except Exception as e:
-                logger.error(f"应用注入规则 '{rule}' 失败: {e}")
+                logger.error(f"应用注入规则 '{rule}' (来源: {source}) 失败: {e}", exc_info=True)
 
         return modified_template
 
-    # --- 状态查询API ---
+    async def preview_prompt_injections(
+        self, target_prompt_name: str, params: PromptParameters
+    ) -> str:
+        """
+        【预览功能】模拟应用所有注入规则，返回最终生成的模板字符串，而不实际修改任何状态。
+
+        这个方法对于调试和测试非常有用，可以查看在特定参数下，
+        一个核心提示词经过所有注入规则处理后会变成什么样子。
+
+        Args:
+            target_prompt_name (str): 希望预览的目标核心提示词名称。
+            params (PromptParameters): 模拟的请求参数。
+
+        Returns:
+            str: 模拟生成的最终提示词模板字符串。如果找不到模板，则返回错误信息。
+        """
+        try:
+            # 从全局提示词管理器获取最原始的模板内容
+            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."
+
+        # 直接调用核心注入逻辑来模拟结果
+        return await self.apply_injections(target_prompt_name, original_template, params)
+
+    # --- 状态观测与查询 API ---
 
     def get_core_prompts(self) -> list[str]:
         """获取所有已注册的核心提示词模板名称列表（即所有可注入的目标）。"""
@@ -187,11 +286,22 @@ class PromptComponentManager:
         return [info for info in components if isinstance(info, PromptInfo)]
 
     async def get_full_injection_map(self) -> dict[str, list[dict]]:
-        """获取当前完整的注入映射图。"""
+        """
+        获取当前完整的注入映射图。
+
+        此方法提供了一个系统全局的注入视图，展示了每个核心提示词（target）
+        被哪些注入组件（source）以何种优先级注入。
+
+        Returns:
+            dict[str, list[dict]]: 一个字典，键是目标提示词名称，
+            值是按优先级排序的注入信息列表。
+            `[{"name": str, "priority": int, "source": str}]`
+        """
         injection_map = {}
         async with self._lock:
+            # 合并所有动态规则的目标和所有核心提示词，确保所有潜在目标都被包含
             all_targets = set(self._dynamic_rules.keys()) | set(self.get_core_prompts())
-            for target in all_targets:
+            for target in sorted(list(all_targets)):
                 rules = self._dynamic_rules.get(target, {})
                 if not rules:
                     injection_map[target] = []
@@ -199,15 +309,23 @@ class PromptComponentManager:
 
                 info_list = []
                 for prompt_name, (rule, _, source) in rules.items():
-                    info_list.append(
-                        {"name": prompt_name, "priority": rule.priority, "source": source}
-                    )
+                    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
 
     async def get_injections_for_prompt(self, target_prompt_name: str) -> list[dict]:
-        """获取指定核心提示词模板的所有注入信息（增强版，包含更多细节）。"""
+        """
+        获取指定核心提示词模板的所有注入信息（包含详细规则）。
+
+        Args:
+            target_prompt_name (str): 目标核心提示词的名称。
+
+        Returns:
+            list[dict]: 一个包含注入规则详细信息的列表，已按优先级排序。
+        """
         rules_for_target = self._dynamic_rules.get(target_prompt_name, {})
         if not rules_for_target:
             return []
@@ -227,13 +345,15 @@ class PromptComponentManager:
         return info_list
 
     def get_all_dynamic_rules(self) -> dict[str, dict[str, "InjectionRule"]]:
-        """获取所有当前的动态注入规则，以 InjectionRule 对象形式返回。"""
+        """
+        获取所有当前的动态注入规则，以 InjectionRule 对象形式返回。
+
+        此方法返回一个深拷贝的规则副本，隐藏了 `content_provider` 等内部实现细节。
+        适合用于展示或序列化当前的规则配置。
+        """
         rules_copy = {}
-        # 只返回规则对象，隐藏 provider 实现细节
         for target, rules in self._dynamic_rules.items():
-            target_copy = {}
-            for name, (rule, _, _) in rules.items():
-                target_copy[name] = rule
+            target_copy = {name: rule for name, (rule, _, _) in rules.items()}
             rules_copy[target] = target_copy
         return copy.deepcopy(rules_copy)
 
@@ -241,23 +361,12 @@ class PromptComponentManager:
         """
         获取所有注入到指定核心提示词的动态规则。
 
-        此函数允许你查询一个特定的核心Prompt（例如 'core_prompt'），
-        并获取一个包含所有以它为目标的注入规则的字典。
-
         Args:
             target_prompt (str): 目标核心提示词的名称。
 
         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} 注入")
+            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()}
@@ -266,23 +375,12 @@ class PromptComponentManager:
         """
         获取由指定的单个注入组件定义的所有动态规则。
 
-        此函数允许你查询一个特定的注入组件（例如 'persona_component'），
-        并找出它注入到了哪些核心Prompt中，以及具体是如何注入的。
-
         Args:
             component_name (str): 注入组件的名称。
 
         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}'")
+            dict[str, InjectionRule]: 一个字典，键是目标核心提示词的名称，值是 `InjectionRule` 对象。
+            如果该组件没有定义任何注入规则，则返回一个空字典。
         """
         found_rules = {}
         for target, rules in self._dynamic_rules.items():
@@ -291,26 +389,8 @@ class PromptComponentManager:
                 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:
-                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)
-
-# 创建全局单例
+# 创建全局单例 (Singleton)
+# 在整个应用程序中，应该只使用这一个 `prompt_component_manager` 实例，
+# 以确保所有部分都共享和操作同一份动态规则集。
 prompt_component_manager = PromptComponentManager()