Maas-Model-platform/backend/app/services/search_options_validator.py

"""
搜索选项验证器

提供搜索功能的模型兼容性验证和参数验证逻辑
需求: 10.1, 10.2, 10.4
"""
import logging
from typing import List, Optional, Set
from sqlalchemy.orm import Session
from app.schemas.llm_schema import SearchOptions
from app.models.model import ModelNew, ModelCategory

logger = logging.getLogger(__name__)


class SearchOptionsValidator:
    """搜索选项验证器"""
    
    def __init__(self, db: Optional[Session] = None):
        """
        初始化验证器
        
        Args:
            db: 数据库会话（可选，用于动态查询模型信息）
        """
        self.db = db
    
    # 支持联网搜索的模型列表（基于阿里云百炼文档）- 作为备用静态配置
    SUPPORTED_SEARCH_MODELS: Set[str] = {
        # 通义千问系列
        "qwen3-max", "qwen3-max-2025-09-23", "qwen3-max-preview",
        "qwen-max", "qwen-max-latest", "qwen-max-2024-09-19",
        "qwen-plus", "qwen-plus-latest", "qwen-plus-2025-07-14",
        "qwen-flash", "qwen-flash-2025-07-28",
        "qwen-turbo", "qwen-turbo-latest", "qwen-turbo-2025-07-15",
        "qwq-plus",
        
        # 第三方模型
        "deepseek-v3.2", "deepseek-v3.2-exp", "deepseek-v3.1",
        "deepseek-r1-0528", "deepseek-r1", "deepseek-v3",
        "Moonshot-Kimi-K2-Instruct"
    }
    
    # 支持时效性参数的模型（仅turbo策略）
    FRESHNESS_SUPPORTED_MODELS: Set[str] = {
        "qwen3-max", "qwen3-max-preview", 
        "qwen-plus", "qwen-flash"
    }
    
    # 有效的搜索策略
    VALID_SEARCH_STRATEGIES: Set[str] = {"turbo", "max", "agent"}
    
    # 有效的时效性值（天数）
    VALID_FRESHNESS_VALUES: Set[int] = {7, 30, 180, 365}
    
    # 有效的引用格式
    VALID_CITATION_FORMATS: Set[str] = {"[<number>]", "[ref_<number>]"}
    
    @classmethod
    def validate_model_compatibility(cls, model: str, search_options: SearchOptions, db: Optional[Session] = None) -> bool:
        """
        验证模型是否支持联网搜索功能
        
        Args:
            model: 模型名称
            search_options: 搜索选项
            db: 数据库会话（可选，用于动态查询）
            
        Returns:
            是否支持搜索功能
            
        Raises:
            ValueError: 当模型不支持搜索功能时
        """
        if not search_options.enable_search:
            return True  # 未启用搜索，无需验证
        
        # 优先从数据库查询
        is_supported = False
        if db:
            try:
                model_obj = db.query(ModelNew).filter(
                    ModelNew.model_code == model,
                    ModelNew.categories.any(int(ModelCategory.LLM))
                ).first()
                is_supported = model_obj.is_search if model_obj else False
            except Exception as e:
                logger.warning(f"数据库查询模型搜索支持状态失败，使用静态配置: {e}")
                is_supported = model in cls.SUPPORTED_SEARCH_MODELS
        else:
            # 使用静态配置作为备用
            is_supported = model in cls.SUPPORTED_SEARCH_MODELS
            
        if not is_supported:
            # 获取支持的模型列表用于错误提示
            supported_models = cls.get_supported_models(db) if db else sorted(cls.SUPPORTED_SEARCH_MODELS)
            raise ValueError(
                f"模型 '{model}' 不支持联网搜索功能。"
                f"支持的模型: {', '.join(supported_models)}"
            )
        
        # 验证时效性参数兼容性
        if (search_options.freshness is not None and 
            search_options.search_strategy == "turbo" and
            model not in cls.FRESHNESS_SUPPORTED_MODELS):
            logger.warning(
                f"模型 '{model}' 不支持时效性参数，将忽略 freshness 设置"
            )
        
        return True
    
    @classmethod
    def validate_search_params(cls, search_options: SearchOptions) -> SearchOptions:
        """
        验证和标准化搜索参数
        
        Args:
            search_options: 原始搜索选项
            
        Returns:
            验证并标准化后的搜索选项
            
        Raises:
            ValueError: 当参数组合不兼容时
        """
        if not search_options.enable_search:
            return search_options
        
        # 时效性参数仅对turbo策略生效
        if (search_options.freshness is not None and 
            search_options.search_strategy != "turbo"):
            logger.warning(
                f"时效性参数仅对 'turbo' 策略生效，当前策略: '{search_options.search_strategy}'，"
                "将忽略 freshness 设置"
            )
            # 创建新的对象，重置freshness
            search_options = SearchOptions(
                enable_search=search_options.enable_search,
                search_strategy=search_options.search_strategy,
                forced_search=search_options.forced_search,
                enable_search_extension=search_options.enable_search_extension,
                freshness=None,  # 重置为None
                enable_source=search_options.enable_source,
                enable_citation=search_options.enable_citation,
                citation_format=search_options.citation_format,
                prepend_search_result=search_options.prepend_search_result,
                intention_options=search_options.intention_options
            )
        
        # 验证引用相关参数的逻辑一致性
        if search_options.enable_citation and not search_options.enable_source:
            logger.warning(
                "启用角标引用时建议同时启用搜索来源 (enable_source=True)"
            )
        
        # 验证提前返回搜索结果的前提条件
        if search_options.prepend_search_result and not search_options.enable_source:
            raise ValueError(
                "启用提前返回搜索结果 (prepend_search_result=True) 时，"
                "必须同时启用搜索来源 (enable_source=True)"
            )
        
        # 验证自然语言搜索控制参数
        if search_options.intention_options:
            # 验证 prompt_intervene 参数
            if "prompt_intervene" in search_options.intention_options:
                prompt_intervene = search_options.intention_options["prompt_intervene"]
                if not isinstance(prompt_intervene, str) or not prompt_intervene.strip():
                    raise ValueError("prompt_intervene 必须是非空字符串")
                
                # 验证指导语句长度（避免过长的指导语句）
                if len(prompt_intervene.strip()) > 500:
                    raise ValueError("prompt_intervene 长度不能超过500个字符")
            
            # 验证不支持的字段
            supported_fields = {"prompt_intervene"}
            unsupported_fields = set(search_options.intention_options.keys()) - supported_fields
            if unsupported_fields:
                logger.warning(f"intention_options 中包含不支持的字段: {unsupported_fields}")
                # 创建新的对象，只保留支持的字段
                filtered_intention_options = {
                    k: v for k, v in search_options.intention_options.items() 
                    if k in supported_fields
                }
                search_options = SearchOptions(
                    enable_search=search_options.enable_search,
                    search_strategy=search_options.search_strategy,
                    forced_search=search_options.forced_search,
                    enable_search_extension=search_options.enable_search_extension,
                    freshness=search_options.freshness,
                    enable_source=search_options.enable_source,
                    enable_citation=search_options.enable_citation,
                    citation_format=search_options.citation_format,
                    prepend_search_result=search_options.prepend_search_result,
                    intention_options=filtered_intention_options if filtered_intention_options else None
                )
        
        return search_options
    
    @classmethod
    def get_supported_models(cls, db: Optional[Session] = None) -> List[str]:
        """
        获取支持搜索的模型列表
        
        Args:
            db: 数据库会话（可选，用于动态查询）
        
        Returns:
            支持搜索的模型名称列表（按字母顺序排序）
        """
        # 优先从数据库查询
        if db:
            try:
                models = db.query(ModelNew).filter(
                    ModelNew.categories.any(int(ModelCategory.LLM)),
                    ModelNew.is_search == True
                ).all()
                return sorted([model.model_code for model in models])
            except Exception as e:
                logger.warning(f"数据库查询支持搜索的模型失败，使用静态配置: {e}")
        
        # 使用静态配置作为备用
        return sorted(cls.SUPPORTED_SEARCH_MODELS)
    
    @classmethod
    def get_freshness_supported_models(cls) -> List[str]:
        """
        获取支持时效性参数的模型列表
        
        Returns:
            支持时效性参数的模型名称列表（按字母顺序排序）
        """
        return sorted(cls.FRESHNESS_SUPPORTED_MODELS)
    
    @classmethod
    def is_search_supported(cls, model: str, db: Optional[Session] = None) -> bool:
        """
        检查指定模型是否支持搜索功能
        
        Args:
            model: 模型名称
            db: 数据库会话（可选，用于动态查询）
            
        Returns:
            是否支持搜索功能
        """
        # 优先从数据库查询
        if db:
            try:
                model_obj = db.query(ModelNew).filter(
                    ModelNew.model_code == model,
                    ModelNew.categories.any(int(ModelCategory.LLM))
                ).first()
                return model_obj.is_search if model_obj else False
            except Exception as e:
                logger.warning(f"数据库查询模型搜索支持状态失败，使用静态配置: {e}")
        
        # 使用静态配置作为备用
        return model in cls.SUPPORTED_SEARCH_MODELS
    
    @classmethod
    def is_freshness_supported(cls, model: str) -> bool:
        """
        检查指定模型是否支持时效性参数
        
        Args:
            model: 模型名称
            
        Returns:
            是否支持时效性参数
        """
        return model in cls.FRESHNESS_SUPPORTED_MODELS
    
    @classmethod
    def create_default_search_options(cls) -> SearchOptions:
        """
        创建默认的搜索选项
        
        Returns:
            默认搜索选项实例
        """
        return SearchOptions(
            enable_search=False,
            search_strategy="turbo",
            forced_search=False,
            enable_search_extension=False,
            freshness=None,
            enable_source=False,
            enable_citation=False,
            citation_format="[<number>]",
            prepend_search_result=False,
            intention_options=None
        )
    
    @classmethod
    def validate_and_normalize(cls, model: str, search_options: Optional[SearchOptions], db: Optional[Session] = None) -> SearchOptions:
        """
        验证并标准化搜索选项的完整流程
        
        Args:
            model: 模型名称
            search_options: 搜索选项（可选）
            db: 数据库会话（可选，用于动态查询）
            
        Returns:
            验证并标准化后的搜索选项
            
        Raises:
            ValueError: 当模型不支持或参数无效时
        """
        # 如果未提供搜索选项，使用默认值
        if search_options is None:
            search_options = cls.create_default_search_options()
        
        # 验证模型兼容性
        cls.validate_model_compatibility(model, search_options, db)
        
        # 验证和标准化参数
        validated_options = cls.validate_search_params(search_options)
        
        logger.info(
            f"搜索选项验证完成: model={model}, "
            f"enable_search={validated_options.enable_search}, "
            f"strategy={validated_options.search_strategy}"
        )
        
        return validated_options