""" 提示词版本管理器 读写/版本化 config/prompt/ 下的 YAML 文件（实际路径为 core/construction_review/component/reviewers/prompt/）。 管理提示词的版本历史、激活状态和差异对比，兼容 prompt_loader.py 的加载格式。 """ import os import re import difflib import logging from datetime import datetime from typing import Optional, List import yaml logger = logging.getLogger(__name__) # ─── 路径常量 ─────────────────────────────────────────────── # prompt_manager.py 位于 core/debug/prompt_manager.py # 项目根目录 = 当前文件上 2 级 _PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')) # 与 prompt_loader.py 使用相同的 prompt 目录 PROMPT_DIR = os.path.join( _PROJECT_ROOT, 'core', 'construction_review', 'component', 'reviewers', 'prompt', ) VERSIONS_DIR = os.path.join(PROMPT_DIR, 'prompt_versions') # ─── 映射表 ───────────────────────────────────────────────── # 主文件映射：file_key -> 文件名 FILE_MAP = { 'basic': 'basic_reviewers.yaml', 'technical': 'technical_reviewers.yaml', 'rag': 'rag_reviewers.yaml', 'ai': 'ai_suggestion.yaml', 'outline': 'outline_reviewers.yaml', 'query_extract': 'query_extract.yaml', } # 提示词名称 -> file_key 映射（共 19 个提示词） PROMPT_FILE_MAP = { # basic_reviewers.yaml 'completeness_check': 'basic', 'grammar_check': 'basic', 'semantic_logic_check': 'basic', 'sensitive_word_check': 'basic', 'timeliness_check': 'basic', 'reference_check': 'basic', # technical_reviewers.yaml 'non_parameter_compliance_check': 'technical', 'parameter_compliance_check': 'technical', # rag_reviewers.yaml 'rag_enhanced_review': 'rag', 'vector_search_review': 'rag', 'hybrid_search_review': 'rag', # ai_suggestion.yaml 'professional_suggestion': 'ai', 'standardization_suggestion': 'ai', 'completeness_suggestion': 'ai', 'readability_suggestion': 'ai', # outline_reviewers.yaml 'outline_completeness_classifier': 'outline', 'outline_completeness_review': 'outline', 'overall_outline_completeness_review': 'outline', # query_extract.yaml 'query_extract': 'query_extract', } # 提示词名称 -> 链路中文名称映射 PROMPT_CHAIN_MAP = { 'completeness_check': '完整性', 'grammar_check': '语法', 'semantic_logic_check': '语义逻辑', 'sensitive_word_check': '敏感词', 'timeliness_check': '时效性', 'reference_check': '规范性', 'non_parameter_compliance_check': '专业性', 'parameter_compliance_check': '专业性', 'rag_enhanced_review': '专业性', 'vector_search_review': '专业性', 'hybrid_search_review': '专业性', 'professional_suggestion': '专业性', 'standardization_suggestion': '规范性', 'completeness_suggestion': '完整性', 'readability_suggestion': '完整性', 'outline_completeness_classifier': '完整性', 'outline_completeness_review': '完整性', 'overall_outline_completeness_review': '完整性', 'query_extract': '专业性', } # 可筛选的链路列表（7 条） CHAINS = ['完整性', '时效性', '规范性', '敏感词', '语义逻辑', '语法', '专业性'] # ─── 工具函数 ─────────────────────────────────────────────── def _extract_variables(text: str) -> List[str]: """从文本中提取 {var} 占位符变量""" return re.findall(r'\{(\w+)\}', text) def _parse_version(ver: str) -> tuple: """解析版本号字符串为可比较的元组，如 'v1.2' -> (1, 2)""" try: parts = ver.lstrip('v').split('.') return tuple(int(p) for p in parts) except (ValueError, IndexError): return (0, 0) def _make_diff_lines(a_text: str, b_text: str) -> list: """生成行级差异列表，使用 difflib.unified_diff""" lines = [] for line in difflib.unified_diff( a_text.splitlines(keepends=False), b_text.splitlines(keepends=False), lineterm='', n=999, # 显示所有行（不折叠） ): if line.startswith('---') or line.startswith('+++') or line.startswith('@@'): continue # 跳过 unified diff 头部 if line.startswith('+'): lines.append({'type': 'add', 'text': line[1:]}) elif line.startswith('-'): lines.append({'type': 'del', 'text': line[1:]}) else: # 上下文行以空格开头 text = line[1:] if line.startswith(' ') else line lines.append({'type': 'ctx', 'text': text}) return lines # ─── PromptManager ────────────────────────────────────────── class PromptManager: """提示词版本管理器 提供提示词的查询、版本管理、对比、激活和回滚功能。 版本历史存储在 PROMPT_DIR/prompt_versions/{prompt_name}/{version}.yaml。 """ def __init__(self): os.makedirs(VERSIONS_DIR, exist_ok=True) self._init_versions() # ==================== 首次运行初始化 ==================== def _init_versions(self): """首次运行时，从当前主文件为所有提示词创建 v1.0 初始版本""" for file_key, filename in FILE_MAP.items(): filepath = os.path.join(PROMPT_DIR, filename) if not os.path.exists(filepath): continue with open(filepath, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) or {} for prompt_name, entry in config.items(): prompt_dir = os.path.join(VERSIONS_DIR, prompt_name) if os.path.exists(prompt_dir): continue # 已有版本目录，跳过 os.makedirs(prompt_dir, exist_ok=True) # 兼容 query_extract 的特殊字段名 if 'system' in entry and 'user_template' in entry: system_prompt = entry.get('system', '') user_prompt = entry.get('user_template', '') else: system_prompt = entry.get('system_prompt', '') user_prompt = entry.get('user_prompt_template', '') version_data = { 'version': 'v1.0', 'created_at': datetime.now().isoformat(), 'based_on': None, 'is_current': True, 'note': '初始版本', 'system_prompt': system_prompt, 'user_prompt_template': user_prompt, } with open(os.path.join(prompt_dir, 'v1.0.yaml'), 'w', encoding='utf-8') as f: yaml.dump(version_data, f, allow_unicode=True, default_flow_style=False) # ==================== 内部辅助方法 ==================== def _get_main_file_path(self, prompt_name: str) -> Optional[str]: """获取提示词对应的主 YAML 文件路径""" file_key = PROMPT_FILE_MAP.get(prompt_name) if not file_key: return None filename = FILE_MAP.get(file_key) if not filename: return None return os.path.join(PROMPT_DIR, filename) def _read_current_from_main(self, prompt_name: str) -> Optional[dict]: """从主文件读取当前激活的提示词内容""" filepath = self._get_main_file_path(prompt_name) if not filepath or not os.path.exists(filepath): return None with open(filepath, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) or {} entry = config.get(prompt_name) if not entry: return None # 兼容 query_extract 的特殊字段名 if 'system' in entry and 'user_template' in entry: return { 'system_prompt': entry.get('system', ''), 'user_prompt': entry.get('user_template', ''), } return { 'system_prompt': entry.get('system_prompt', ''), 'user_prompt': entry.get('user_prompt_template', ''), } def _get_version_dir(self, prompt_name: str) -> str: return os.path.join(VERSIONS_DIR, prompt_name) def _get_version_file(self, prompt_name: str, version: str) -> str: return os.path.join(self._get_version_dir(prompt_name), f'{version}.yaml') def _load_version_file(self, prompt_name: str, version: str) -> Optional[dict]: """加载指定版本文件，失败时返回 None""" filepath = self._get_version_file(prompt_name, version) if not os.path.exists(filepath): return None try: with open(filepath, 'r', encoding='utf-8') as f: return yaml.safe_load(f) except Exception as e: logger.error('读取版本文件失败: %s, 错误: %s', filepath, e) return None def _list_version_files(self, prompt_name: str) -> List[str]: """列出某提示词的所有版本文件名，按版本号降序""" version_dir = self._get_version_dir(prompt_name) if not os.path.exists(version_dir): return [] files = [f for f in os.listdir(version_dir) if f.endswith('.yaml')] files.sort(key=lambda x: _parse_version(x.replace('.yaml', '')), reverse=True) return files def _get_current_main_version_info(self, prompt_name: str) -> dict: """获取当前激活版本的元信息""" version_dir = self._get_version_dir(prompt_name) if os.path.exists(version_dir): for fname in sorted(os.listdir(version_dir), reverse=True): if not fname.endswith('.yaml'): continue data = self._load_version_file(prompt_name, fname.replace('.yaml', '')) if data and data.get('is_current'): return data # 降级：从主文件读取 current = self._read_current_from_main(prompt_name) if current: return { 'version': 'current', 'is_current': True, 'system_prompt': current['system_prompt'], 'user_prompt_template': current['user_prompt'], 'created_at': '', 'note': '', } return {} def _update_is_current_flag(self, prompt_name: str, active_version: str): """更新某提示词的 is_current 标记，清除其他版本的标记""" version_dir = self._get_version_dir(prompt_name) if not os.path.exists(version_dir): return for fname in os.listdir(version_dir): if not fname.endswith('.yaml'): continue ver = fname.replace('.yaml', '') filepath = os.path.join(version_dir, fname) data = self._load_version_file(prompt_name, ver) if data is None: continue data['is_current'] = (ver == active_version) with open(filepath, 'w', encoding='utf-8') as f: yaml.dump(data, f, allow_unicode=True, default_flow_style=False) def _sync_main_to_versions(self, prompt_name: str): """确保主文件当前内容在版本目录中有对应版本（兜底逻辑）""" version_files = self._list_version_files(prompt_name) if version_files: return # 已有版本记录，无需同步 current = self._read_current_from_main(prompt_name) if not current: return prompt_dir = self._get_version_dir(prompt_name) os.makedirs(prompt_dir, exist_ok=True) version_data = { 'version': 'v1.0', 'created_at': datetime.now().isoformat(), 'based_on': None, 'is_current': True, 'note': '初始版本（同步自主文件）', 'system_prompt': current['system_prompt'], 'user_prompt_template': current['user_prompt'], } with open(os.path.join(prompt_dir, 'v1.0.yaml'), 'w', encoding='utf-8') as f: yaml.dump(version_data, f, allow_unicode=True, default_flow_style=False) # ==================== 公开方法 ==================== def get_all_prompts(self, chain_filter: str = None, search: str = None) -> list: """获取所有提示词及其版本信息列表 Args: chain_filter: 链路名称筛选 search: 提示词名称搜索关键字 Returns: list[dict]: 每项包含 name, version, time, chain, is_current, note """ results = [] for prompt_name in PROMPT_FILE_MAP: if search and search.lower() not in prompt_name.lower(): continue chain = PROMPT_CHAIN_MAP.get(prompt_name, '') if chain_filter and chain != chain_filter: continue self._sync_main_to_versions(prompt_name) version_files = self._list_version_files(prompt_name) if not version_files: # 没有版本记录，直接从主文件读取 current = self._read_current_from_main(prompt_name) if current: results.append({ 'name': prompt_name, 'version': 'current', 'time': '', 'chain': chain, 'is_current': True, 'note': '', }) else: for fname in version_files: ver = fname.replace('.yaml', '') data = self._load_version_file(prompt_name, ver) if data is None: continue results.append({ 'name': prompt_name, 'version': data.get('version', ver), 'time': data.get('created_at', ''), 'chain': chain, 'is_current': data.get('is_current', False), 'note': data.get('note', ''), }) return results def get_prompt_detail(self, name: str, version: str = None) -> Optional[dict]: """获取提示词详情 Args: name: 提示词名称 version: 版本号，不指定则返回当前激活版本 Returns: dict | None: 包含完整提示词详情，不存在时返回 None """ filepath = self._get_main_file_path(name) if not filepath: return None if name not in PROMPT_FILE_MAP: return None chain = PROMPT_CHAIN_MAP.get(name, '') if version: # 从指定版本文件读取 data = self._load_version_file(name, version) if not data: return None system_prompt = data.get('system_prompt', '') user_prompt = data.get('user_prompt_template', '') return { 'name': name, 'version': data.get('version', version), 'time': data.get('created_at', ''), 'chain': chain, 'is_current': data.get('is_current', False), 'system_prompt': system_prompt, 'user_prompt': user_prompt, 'note': data.get('note', ''), 'variables': _extract_variables(user_prompt), 'based_on': data.get('based_on'), 'file_path': filepath, } else: # 读取当前激活版本 version_info = self._get_current_main_version_info(name) if not version_info: return None system_prompt = version_info.get('system_prompt', '') user_prompt = version_info.get('user_prompt_template', '') return { 'name': name, 'version': version_info.get('version', 'current'), 'time': version_info.get('created_at', ''), 'chain': chain, 'is_current': True, 'system_prompt': system_prompt, 'user_prompt': user_prompt, 'note': version_info.get('note', ''), 'variables': _extract_variables(user_prompt), 'based_on': version_info.get('based_on'), 'file_path': filepath, } def save_new_version(self, name: str, system_prompt: str, user_prompt: str, note: str = '', set_current: bool = True) -> dict: """保存新版本 Args: name: 提示词名称 system_prompt: 系统提示词内容 user_prompt: 用户提示词模板 note: 版本说明 set_current: 是否设为当前激活版本 Returns: dict: {version, name, time} Raises: ValueError: 提示词名称不存在 """ if name not in PROMPT_FILE_MAP: raise ValueError(f'未知的提示词: {name}') # 同步主文件到版本目录（兜底） self._sync_main_to_versions(name) # 计算版本号：递增 major 版本 version_files = self._list_version_files(name) if version_files: latest_ver = version_files[0].replace('.yaml', '') latest_num = _parse_version(latest_ver) new_ver = f'v{latest_num[0] + 1}.0' based_on = latest_ver else: new_ver = 'v1.0' based_on = None # 写入版本文件 prompt_dir = self._get_version_dir(name) os.makedirs(prompt_dir, exist_ok=True) version_data = { 'version': new_ver, 'created_at': datetime.now().isoformat(), 'based_on': based_on, 'is_current': False, 'note': note, 'system_prompt': system_prompt, 'user_prompt_template': user_prompt, } with open(self._get_version_file(name, new_ver), 'w', encoding='utf-8') as f: yaml.dump(version_data, f, allow_unicode=True, default_flow_style=False) # 更新主文件 if set_current: self.activate_version(name, new_ver) return { 'version': new_ver, 'name': name, 'time': version_data['created_at'], } def get_versions(self, name: str) -> List[dict]: """获取某提示词的所有历史版本列表 Args: name: 提示词名称 Returns: list[dict]: 每项包含 version, time, is_current, note, system_prompt_preview, user_prompt_preview """ self._sync_main_to_versions(name) version_files = self._list_version_files(name) versions = [] for fname in version_files: ver = fname.replace('.yaml', '') data = self._load_version_file(name, ver) if data is None: continue sys_prompt = data.get('system_prompt', '') user_prompt = data.get('user_prompt_template', '') versions.append({ 'version': data.get('version', ver), 'time': data.get('created_at', ''), 'is_current': data.get('is_current', False), 'note': data.get('note', ''), 'system_prompt_preview': (sys_prompt[:50] + '...' if len(sys_prompt) > 50 else sys_prompt), 'user_prompt_preview': (user_prompt[:50] + '...' if len(user_prompt) > 50 else user_prompt), }) return versions def compare_versions(self, name: str, base_version: str, target_version: str) -> dict: """对比两个版本的差异（行级 Diff） Args: name: 提示词名称 base_version: 基准版本号 target_version: 目标版本号 Returns: dict: {name, base_version, target_version, diffs} Raises: FileNotFoundError: 版本文件不存在 """ base_data = self._load_version_file(name, base_version) target_data = self._load_version_file(name, target_version) if not base_data or not target_data: missing = [] if not base_data: missing.append(base_version) if not target_data: missing.append(target_version) raise FileNotFoundError(f'版本文件不存在: {name} [{", ".join(missing)}]') diffs = [] for section_key, display_section in [ ('system_prompt', 'system_prompt'), ('user_prompt_template', 'user_prompt'), ]: base_text = base_data.get(section_key, '') target_text = target_data.get(section_key, '') diff_lines = _make_diff_lines(base_text, target_text) diffs.append({ 'section': display_section, 'type': 'text_diff', 'lines': diff_lines, }) return { 'name': name, 'base_version': base_version, 'target_version': target_version, 'diffs': diffs, } def activate_version(self, name: str, version: str) -> dict: """激活指定版本（将版本内容覆盖写入主 YAML 文件） Args: name: 提示词名称 version: 版本号 Returns: dict: {success, name, version} Raises: ValueError: 提示词名称不存在 FileNotFoundError: 版本文件或主文件不存在 """ if name not in PROMPT_FILE_MAP: raise ValueError(f'未知的提示词: {name}') version_data = self._load_version_file(name, version) if not version_data: raise FileNotFoundError(f'版本文件不存在: {name}/{version}.yaml') # 读取主文件 filepath = self._get_main_file_path(name) if not filepath or not os.path.exists(filepath): raise FileNotFoundError(f'主文件不存在: {filepath}') with open(filepath, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) or {} # 保留主文件原有字段名（兼容 query_extract 等特殊格式） original_entry = config.get(name, {}) new_system_prompt = version_data.get('system_prompt', '') new_user_prompt = version_data.get('user_prompt_template', '') if name not in config: config[name] = {} # 根据原字段名决定写入方式 if 'system' in original_entry and 'user_template' in original_entry: config[name]['system'] = new_system_prompt config[name]['user_template'] = new_user_prompt else: config[name]['system_prompt'] = new_system_prompt config[name]['user_prompt_template'] = new_user_prompt # 写回主文件 with open(filepath, 'w', encoding='utf-8') as f: yaml.dump(config, f, allow_unicode=True, default_flow_style=False) # 更新版本文件的 is_current 标记 self._update_is_current_flag(name, version) # 清除 prompt_loader 缓存 self._clear_prompt_loader_cache() return { 'success': True, 'name': name, 'version': version, } def rollback_version(self, name: str, version: str) -> dict: """回滚到指定历史版本（等同于激活该版本） Args: name: 提示词名称 version: 版本号 Returns: dict: {success, name, version} """ return self.activate_version(name, version) @staticmethod def _clear_prompt_loader_cache(): """清除 prompt_loader 缓存，使下一次读取使用最新激活的内容""" try: from core.construction_review.component.reviewers.utils.prompt_loader import ( prompt_loader, ) prompt_loader.manage_cache(action='clear') except ImportError: logger.warning('无法导入 prompt_loader，跳过缓存清理') except Exception as e: logger.error('清理 prompt_loader 缓存失败: %s', e)