""" 提示词管理 API 端点 提供提示词的查询、版本管理、对比、激活和回滚功能。 基于 core/debug/prompt_manager.py 的 PromptManager 实现。 """ import logging from fastapi import APIRouter, Query, HTTPException from pydantic import ValidationError from core.debug.prompt_manager import PromptManager, CHAINS, PROMPT_FILE_MAP, PROMPT_CHAIN_MAP logger = logging.getLogger(__name__) # 全局 PromptManager 实例 _prompt_manager = None def _get_manager() -> PromptManager: """获取或创建 PromptManager 单例""" global _prompt_manager if _prompt_manager is None: _prompt_manager = PromptManager() return _prompt_manager def register_routes(router: APIRouter): """在指定 router 上注册提示词管理路由""" @router.get( '/api/prompts', summary='获取提示词列表', description='获取所有提示词及其版本信息，支持链路筛选和名称搜索。', ) async def list_prompts( chain: str = Query(default=None, description='链路筛选'), search: str = Query(default=None, description='名称搜索'), page: int = Query(default=1, ge=1, description='页码'), page_size: int = Query(default=50, ge=1, le=200, description='每页条数'), ): """ 获取所有提示词及其版本信息。 """ try: manager = _get_manager() items = manager.get_all_prompts(chain_filter=chain, search=search) # 分页 total = len(items) start = (page - 1) * page_size paged_items = items[start:start + page_size] return { 'status': 'ok', 'total': total, 'page': page, 'page_size': page_size, 'items': paged_items, 'chains': CHAINS, } except Exception as e: logger.error('获取提示词列表失败: %s', e) raise HTTPException(status_code=500, detail=str(e)) @router.get( '/api/prompts/{name}', summary='获取提示词详情', description='获取指定提示词指定版本的完整详情，包括 system_prompt 和 user_prompt。', ) async def get_prompt_detail( name: str, version: str = Query(default=None, description='版本号，不指定则返回当前激活版本'), ): """ 获取指定提示词的完整详情，包括系统提示词和用户提示词模板。 """ try: manager = _get_manager() detail = manager.get_prompt_detail(name, version=version) if detail is None: raise HTTPException( status_code=404, detail=f'提示词不存在: {name}', ) return { 'status': 'ok', **detail, } except HTTPException: raise except Exception as e: logger.error('获取提示词详情失败: %s', e) raise HTTPException(status_code=500, detail=str(e)) @router.get( '/api/prompts/{name}/versions', summary='获取版本列表', description='获取指定提示词的所有历史版本。', ) async def list_prompt_versions(name: str): """ 获取指定提示词的所有历史版本。 """ try: manager = _get_manager() # 先检查提示词是否存在 if name not in PROMPT_FILE_MAP: detail = manager.get_prompt_detail(name) if detail is None: raise HTTPException( status_code=404, detail=f'提示词不存在: {name}', ) # 获取当前激活版本 current_info = manager.get_prompt_detail(name) current_version = current_info.get('version', '') if current_info else '' versions = manager.get_versions(name) chain = PROMPT_CHAIN_MAP.get(name, '') return { 'status': 'ok', 'name': name, 'chain': chain, 'current_version': current_version, 'versions': versions, } except HTTPException: raise except Exception as e: logger.error('获取版本列表失败: %s', e) raise HTTPException(status_code=500, detail=str(e)) @router.post( '/api/prompts/save', summary='保存新版本', description='保存提示词的新版本。将当前编辑内容保存为新版本，并可选设为当前激活版本。', ) async def save_prompt_version(body: dict): """ 保存提示词的新版本。 """ try: name = body.get('name', '') system_prompt = body.get('system_prompt', '') user_prompt = body.get('user_prompt', '') note = body.get('note', '') set_current = body.get('set_current', True) # 参数验证 if not name: raise HTTPException(status_code=422, detail='name 不能为空') if not system_prompt: raise HTTPException(status_code=422, detail='system_prompt 不能为空') if not user_prompt: raise HTTPException(status_code=422, detail='user_prompt 不能为空') manager = _get_manager() result = manager.save_new_version( name=name, system_prompt=system_prompt, user_prompt=user_prompt, note=note, set_current=set_current, ) return { 'success': True, 'name': result['name'], 'version': result['version'], 'time': result['time'], 'message': f'已保存新版本 {result["version"]}', } except HTTPException: raise except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error('保存新版本失败: %s', e) raise HTTPException(status_code=500, detail=str(e)) @router.post( '/api/prompts/compare', summary='版本对比', description='对比两个版本的差异（行级 Diff）。', ) async def compare_prompt_versions(body: dict): """ 对比两个版本的差异。 """ try: name = body.get('name', '') base_version = body.get('base_version', '') target_version = body.get('target_version', '') if not name or not base_version or not target_version: raise HTTPException( status_code=422, detail='name, base_version, target_version 不能为空', ) manager = _get_manager() result = manager.compare_versions(name, base_version, target_version) return { 'status': 'ok', **result, } except HTTPException: raise except FileNotFoundError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error('版本对比失败: %s', e) raise HTTPException(status_code=500, detail=str(e)) @router.post( '/api/prompts/activate', summary='激活版本', description='将指定版本设为当前激活版本（覆盖写入主 YAML 文件）。', ) async def activate_prompt_version(body: dict): """ 将指定版本设为当前激活版本。 """ try: name = body.get('name', '') version = body.get('version', '') if not name or not version: raise HTTPException( status_code=422, detail='name, version 不能为空', ) manager = _get_manager() result = manager.activate_version(name, version) return { 'success': result['success'], 'name': result['name'], 'version': result['version'], 'message': f'已激活版本 {result["version"]}', } except HTTPException: raise except (ValueError, FileNotFoundError) as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error('激活版本失败: %s', e) raise HTTPException(status_code=500, detail=str(e)) @router.post( '/api/prompts/rollback', summary='回滚版本', description='回滚到指定历史版本（等同于将该版本内容设为当前激活版本）。', ) async def rollback_prompt_version(body: dict): """ 回滚到指定历史版本。 """ try: name = body.get('name', '') version = body.get('version', '') if not name or not version: raise HTTPException( status_code=422, detail='name, version 不能为空', ) manager = _get_manager() result = manager.rollback_version(name, version) return { 'success': result['success'], 'name': result['name'], 'version': result['version'], 'message': f'已回滚到版本 {result["version"]}', } except HTTPException: raise except (ValueError, FileNotFoundError) as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: logger.error('回滚版本失败: %s', e) raise HTTPException(status_code=500, detail=str(e))