把单一换到丰富时踩过的坑

def preprocess_query(query: str) -> dict:
    """查询预处理"""
    result = {
        'original': query,
        'cleaned': query.strip(),
        'tokens': [],
        'entities': []
    }

    # 基础清洗
    query = query.strip().lower()

    # 分词（中文需要专门的分词器）
    tokens = tokenize_chinese(query)
    result['tokens'] = tokens

    # 命名实体识别（识别人名、地名、技术名词等）
    entities = extract_entities(query)
    result['entities'] = entities

    # 停用词处理（保留核心意思）
    tokens = [t for t in tokens if t not in STOP_WORDS]
    result['core_tokens'] = tokens

    return result

from openai import OpenAI
import json

client = OpenAI()

def understand_query(query: str) -> dict:
    """理解查询的语义"""
    prompt = f"""
    分析以下查询的语义，提取关键信息：

    查询：{query}

    请返回JSON格式：
    {{
        "main_intent": "主要意图（比如搜索、学习、解决问题）",
        "keywords": ["关键词1", "关键词2"],
        "synonyms": {{
            "关键词1": ["同义词1", "同义词2"],
            "关键词2": ["同义词1", "同义词2"]
        }},
        "context": "上下文或场景",
        "domain": "技术领域"
    }}
    """

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是一个专业的搜索查询分析师。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3
    )

    try:
        result = json.loads(response.choices[0].message.content)
        return result
    except json.JSONDecodeError:
        return {
            "main_intent": "未知",
            "keywords": [],
            "synonyms": {},
            "context": "",
            "domain": "通用"
        }

def expand_with_synonyms(query_analysis: dict) -> list[str]:
    """同义词扩展"""
    expanded_queries = [query_analysis['original']]

    # 对每个关键词进行同义词扩展
    for keyword, synonyms in query_analysis['synonyms'].items():
        for synonym in synonyms:
            # 替换关键词生成新查询
            new_query = query_analysis['original'].replace(keyword, synonym)
            expanded_queries.append(new_query)

    return list(set(expanded_queries))  # 去重

def expand_with_context(query_analysis: dict) -> list[str]:
    """上下文扩展 - 添加相关概念"""
    context_expansions = {
        "前端开发": ["JavaScript", "TypeScript", "React", "Vue", "Angular"],
        "后端开发": ["Node.js", "Python", "Java", "Go", "API设计"],
        "数据库": ["MySQL", "PostgreSQL", "MongoDB", "Redis", "SQL优化"],
        # ... 更多上下文
    }

    domain = query_analysis.get('domain', '')
    if domain in context_expansions:
        concepts = context_expansions[domain]
        return [f"{query_analysis['original']} {concept}" for concept in concepts]
    return []

def expand_with_domain_terms(query_analysis: dict) -> list[str]:
    """领域术语扩展 - 添加领域相关的专业术语"""
    domain_terms = {
        "Vue": ["组件通信", "状态管理", "生命周期", "指令", "路由"],
        "React": ["Hooks", "Redux", "Context", "Component", "Props"],
        "Python": ["异步编程", "装饰器", "生成器", "多线程", "协程"],
        # ... 更多领域术语
    }

    expanded = []
    tokens = query_analysis.get('keywords', [])

    for token in tokens:
        for domain, terms in domain_terms.items():
            if domain.lower() in token.lower():
                expanded.extend([f"{query_analysis['original']} {term}" for term in terms])

    return expanded

def rewrite_and_fuse(original_query: str, expansions: list[list[str]]) -> list[str]:
    """查询重写与融合"""
    all_queries = [original_query]

    # 收集所有扩展查询
    for expansion_list in expansions:
        all_queries.extend(expansion_list)

    # 去重，但保持顺序
    seen = set()
    unique_queries = []
    for query in all_queries:
        normalized = query.lower().strip()
        if normalized not in seen:
            seen.add(normalized)
            unique_queries.append(query)

    # 控制扩展数量（避免太多）
    MAX_QUERIES = 10
    if len(unique_queries) > MAX_QUERIES:
        # 优先保留原始查询和最相关的扩展
        unique_queries = unique_queries[:MAX_QUERIES]

    return unique_queries

def ai_query_expansion(query: str) -> dict:
    """完整的AI查询扩展流程"""
    # 1. 预处理
    processed = preprocess_query(query)

    # 2. 语义理解
    query_analysis = understand_query(query)

    # 3. 多策略扩展
    synonyms_expansions = expand_with_synonyms(query_analysis)
    context_expansions = expand_with_context(query_analysis)
    domain_expansions = expand_with_domain_terms(query_analysis)

    # 4. 查询重写与融合
    all_expansions = [synonyms_expansions, context_expansions, domain_expansions]
    final_queries = rewrite_and_fuse(query, all_expansions)

    return {
        'original_query': query,
        'processed_query': processed,
        'query_analysis': query_analysis,
        'expanded_queries': final_queries,
        'expansion_count': len(final_queries) - 1
    }

import time
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_expand(query: str) -> dict:
    """带缓存的查询扩展"""
    return ai_query_expansion(query)

def expand_with_timeout(query: str, timeout: float = 2.0) -> dict:
    """带超时的查询扩展"""
    try:
        # 先查缓存
        cached = cached_expand(query)
        if cached:
            return cached

        # 设置超时
        start_time = time.time()
        result = ai_query_expansion(query)
        elapsed = time.time() - start_time

        if elapsed > timeout:
            print(f"警告：查询扩展耗时 {elapsed:.2f}秒，超过 {timeout}秒")

        return result

    except Exception as e:
        print(f"查询扩展失败：{e}")
        return {
            'original_query': query,
            'expanded_queries': [query],
            'expansion_count': 0
        }

def handle_mixed_language(query: str) -> dict:
    """处理中英文混合查询"""
    # 识别中英文部分
    chinese_parts = []
    english_parts = []

    for char in query:
        if '一' <= char <= '鿿':
            chinese_parts.append(char)
        else:
            english_parts.append(char)

    chinese_text = ''.join(chinese_parts).strip()
    english_text = ''.join(english_parts).strip()

    # 分别处理
    result = {
        'original': query,
        'chinese': chinese_text,
        'english': english_text,
        'expansions': []
    }

    if chinese_text:
        result['expansions'].extend(expand_chinese(chinese_text))
    if english_text:
        result['expansions'].extend(expand_english(english_text))

    return result

DOMAIN_CONFIG = {
    "前端": {
        "expansion_strategies": ["synonyms", "framework_specific"],
        "max_expansions": 8,
        "min_relevance": 0.7
    },
    "后端": {
        "expansion_strategies": ["synonyms", "database_specific"],
        "max_expansions": 10,
        "min_relevance": 0.6
    },
    "AI/ML": {
        "expansion_strategies": ["synonyms", "context", "domain_terms"],
        "max_expansions": 12,
        "min_relevance": 0.8
    }
}

def get_domain_config(domain: str) -> dict:
    """获取领域配置"""
    return DOMAIN_CONFIG.get(domain, DOMAIN_CONFIG["通用"])

def domain_aware_expansion(query: str, domain: str) -> dict:
    """领域感知的查询扩展"""
    config = get_domain_config(domain)

    result = ai_query_expansion(query)

    # 根据领域配置过滤扩展结果
    if config.get('max_expansions'):
        result['expanded_queries'] = result['expanded_queries'][:config['max_expansions']]

    return result