feat(search): 统一重排逻辑，仅由 ai_search 控制并调用外部 BGE 重排服务

- API：新增请求参数 ai_search，开启时在窗口内走重排流程 - 配置：RerankConfig 移除 enabled/expression/description，仅保留 rerank_window 及 service_url/timeout_sec/weight_es/weight_ai；默认超时 15s - 重排流程：ai_search 且 from+size<=rerank_window 时，ES 取前 rerank_window 条，调用外部 /rerank 服务，融合 ES 与重排分数后按 from/size 分页；否则不重排 - search/rerank_client：新增模块，封装 build_docs、call_rerank_service、 fuse_scores_and_resort、run_rerank；超时单独捕获并简短日志 - search/searcher：移除 RerankEngine，enable_rerank=ai_search，使用 config.rerank 参数 - 删除 search/rerank_engine.py（本地表达式重排），统一为外部服务一种实现 - 文档：搜索 API 对接指南补充 ai_search 与 relevance_score 说明 - 测试：conftest 中 rerank 配置改为新结构 Co-authored-by: Cursor <cursoragent@cursor.com>

feat(search): 统一重排逻辑，仅由 ai_search 控制并调用外部 BGE 重排服务
- API：新增请求参数 ai_search，开启时在窗口内走重排流程 - 配置：RerankConfig 移除 enabled/expression/description，仅保留 rerank_window 及 service_url/timeout_sec/weight_es/weight_ai；默认超时 15s - 重排流程：ai_search 且 from+size<=rerank_window 时，ES 取前 rerank_window 条，调用外部 /rerank 服务，融合 ES 与重排分数后按 from/size 分页；否则不重排 - search/rerank_client：新增模块，封装 build_docs、call_rerank_service、 fuse_scores_and_resort、run_rerank；超时单独捕获并简短日志 - search/searcher：移除 RerankEngine，enable_rerank=ai_search，使用 config.rerank 参数 - 删除 search/rerank_engine.py（本地表达式重排），统一为外部服务一种实现 - 文档：搜索 API 对接指南补充 ai_search 与 relevance_score 说明 - 测试：conftest 中 rerank 配置改为新结构 Co-authored-by: Cursor <cursoragent@cursor.com>
tangwang
1 parent d90e7428
Showing 10 changed files with 362 additions and 203 deletions Show diff stats
api/models.py
api/routes/search.py
config/config.yaml
config/config_loader.py
docs/搜索API对接指南.md
search/__init__.py
search/rerank_client.py
search/rerank_engine.py
search/searcher.py
tests/conftest.py
@@ -151,6 +151,10 @@ class SearchRequest(BaseModel):
     min_score: Optional[float] = Field(None, ge=0, description="最小相关性分数阈值")
     highlight: bool = Field(False, description="是否高亮搜索关键词（暂不实现）")
     debug: bool = Field(False, description="是否返回调试信息")
+    ai_search: bool = Field(
+        False,
+        description="是否开启 AI 搜索（调用本地重排服务对 ES 结果进行二次排序）"
+    )
  
     # SKU筛选参数
     sku_filter_dimension: Optional[List[str]] = Field(
@@ -84,6 +84,7 @@ async def search(request: SearchRequest, http_request: Request):
             f"min_score: {request.min_score} | "
             f"language: {request.language} | "
             f"debug: {request.debug} | "
+            f"ai_search: {request.ai_search} | "
             f"sku_filter_dimension: {request.sku_filter_dimension} | "
             f"filters: {request.filters} | "
             f"range_filters: {request.range_filters} | "
@@ -111,6 +112,7 @@ async def search(request: SearchRequest, http_request: Request):
             debug=request.debug,
             language=request.language,
             sku_filter_dimension=request.sku_filter_dimension,
+            ai_search=request.ai_search,
         )
  
         # Include performance summary in response
@@ -133,11 +133,14 @@ function_score:
   boost_mode: "multiply"
   functions: []
  
-# Rerank配置（本地重排，当前禁用）
+# 重排配置（唯一实现：外部 BGE 重排服务，由请求参数 ai_search 控制是否执行）
+# ai_search 且 from+size<=rerank_window 时：从 ES 取前 rerank_window 条、重排后再按 from/size 分页
 rerank:
-  enabled: false
-  expression: ""
-  description: "Local reranking (disabled, use ES function_score instead)"
+  rerank_window: 1000
+  # service_url: "http://127.0.0.1:6007/rerank"  # 可选，不填则用默认端口 6007
+  timeout_sec: 15.0   # 文档多时重排耗时长，可按需调大
+  weight_es: 0.4
+  weight_ai: 0.6
  
 # SPU配置（已启用，使用嵌套skus）
 spu_config:
@@ -88,10 +88,14 @@ class RankingConfig:
  
 @dataclass
 class RerankConfig:
-    """本地重排配置（当前禁用）"""
-    enabled: bool = False
-    expression: str = ""
-    description: str = ""
+    """重排配置（唯一实现：调用外部 BGE 重排服务，由请求参数 ai_search 控制是否执行）"""
+    # 重排窗口：ai_search 且 from+size<=rerank_window 时，从 ES 取前 rerank_window 条重排后再分页
+    rerank_window: int = 1000
+    # 可选：重排服务 URL，为空时使用 reranker 模块默认端口 6007
+    service_url: Optional[str] = None
+    timeout_sec: float = 15.0
+    weight_es: float = 0.4
+    weight_ai: float = 0.6
  
  
 @dataclass
@@ -263,12 +267,14 @@ class ConfigLoader:
             functions=fs_data.get("functions") or []
         )
  
-        # Parse Rerank configuration
+        # Parse Rerank configuration（唯一实现：外部重排服务，由 ai_search 控制）
         rerank_data = config_data.get("rerank", {})
         rerank = RerankConfig(
-            enabled=rerank_data.get("enabled", False),
-            expression=rerank_data.get("expression") or "",
-            description=rerank_data.get("description") or ""
+            rerank_window=int(rerank_data.get("rerank_window", 1000)),
+            service_url=rerank_data.get("service_url") or None,
+            timeout_sec=float(rerank_data.get("timeout_sec", 15.0)),
+            weight_es=float(rerank_data.get("weight_es", 0.4)),
+            weight_ai=float(rerank_data.get("weight_ai", 0.6)),
         )
  
         # Parse SPU config
@@ -399,9 +405,11 @@ class ConfigLoader:
                 "functions": config.function_score.functions
             },
             "rerank": {
-                "enabled": config.rerank.enabled,
-                "expression": config.rerank.expression,
-                "description": config.rerank.description
+                "rerank_window": config.rerank.rerank_window,
+                "service_url": config.rerank.service_url,
+                "timeout_sec": config.rerank.timeout_sec,
+                "weight_es": config.rerank.weight_es,
+                "weight_ai": config.rerank.weight_ai,
             },
             "spu_config": {
                 "enabled": config.spu_config.enabled,
@@ -167,6 +167,7 @@ curl -X POST &quot;http://120.76.41.98:6002/search/&quot; \
   "min_score": 0.0,
   "sku_filter_dimension": ["string"],
   "debug": false,
+  "ai_search": false,
   "user_id": "string",
   "session_id": "string"
 }
@@ -188,6 +189,7 @@ curl -X POST &quot;http://120.76.41.98:6002/search/&quot; \
 | `min_score` | float | N | null | 最小相关性分数阈值 |
 | `sku_filter_dimension` | array[string] | N | null | 子SKU筛选维度列表（见[SKU筛选维度](#35-sku筛选维度)） |
 | `debug` | boolean | N | false | 是否返回调试信息 |
+| `ai_search` | boolean | N | false | 是否开启 AI 搜索（调用本地重排服务对 ES 结果进行二次排序） |
 | `user_id` | string | N | null | 用户ID（用于个性化，预留） |
 | `session_id` | string | N | null | 会话ID（用于分析，预留） |
  
@@ -787,7 +789,7 @@ curl &quot;http://localhost:6002/search/12345&quot;
 | `option3_name` | string | 选项3名称 |
 | `specifications` | array[object] | 规格列表（与ES specifications字段对应） |
 | `skus` | array | SKU 列表 |
-| `relevance_score` | float | 相关性分数 |
+| `relevance_score` | float | 相关性分数（默认为 ES 原始分数；当开启 AI 搜索时为融合后的最终分数） |
  
 ### 4.4 SkuResult字段说明
  
@@ -2,14 +2,12 @@
  
 from .boolean_parser import BooleanParser, QueryNode
 from .es_query_builder import ESQueryBuilder
-from .rerank_engine import RerankEngine
 from .searcher import Searcher, SearchResult
  
 __all__ = [
     'BooleanParser',
     'QueryNode',
     'ESQueryBuilder',
-    'RerankEngine',
     'Searcher',
     'SearchResult',
 ]
@@ -0,0 +1,244 @@
+"""
+重排客户端：调用外部 BGE 重排服务，并对 ES 分数与重排分数进行融合。
+
+流程：
+1. 从 ES hits 构造用于重排的文档文本列表
+2. POST 请求到重排服务 /rerank，获取每条文档的 relevance 分数
+3. 将 ES 分数（归一化）与重排分数线性融合，写回 hit["_score"] 并重排序
+"""
+
+from typing import Dict, Any, List, Optional, Tuple
+import logging
+
+logger = logging.getLogger(__name__)
+
+# 默认融合权重：ES 归一化分数权重、重排分数权重（相加为 1）
+DEFAULT_WEIGHT_ES = 0.4
+DEFAULT_WEIGHT_AI = 0.6
+# 重排服务默认超时（文档较多时需更大，建议 config 中 timeout_sec 调大）
+DEFAULT_TIMEOUT_SEC = 15.0
+
+
+def build_docs_from_hits(
+    es_hits: List[Dict[str, Any]],
+    language: str = "zh",
+) -> List[str]:
+    """
+    从 ES 命中结果构造重排服务所需的文档文本列表（与 hits 一一对应）。
+
+    文本由 title、brief、description、vendor、category_path 等多语言字段拼接，
+    按 language 优先选取对应语言；若无内容则用 spu_id 兜底。
+
+    Args:
+        es_hits: ES 返回的 hits 列表，每项含 _source
+        language: 语言代码，如 "zh"、"en"
+
+    Returns:
+        与 es_hits 等长的字符串列表，用于 POST /rerank 的 docs
+    """
+    lang = (language or "zh").strip().lower()
+    if lang not in ("zh", "en"):
+        lang = "zh"
+
+    def pick_lang_text(obj: Any) -> str:
+        if obj is None:
+            return ""
+        if isinstance(obj, dict):
+            return str(obj.get(lang) or obj.get("zh") or obj.get("en") or "").strip()
+        return str(obj).strip()
+
+    docs: List[str] = []
+    for hit in es_hits:
+        src = hit.get("_source") or {}
+        parts: List[str] = []
+        for key in ("title", "brief", "description", "vendor", "category_path"):
+            parts.append(pick_lang_text(src.get(key)))
+        text = " ".join(p for p in parts if p).strip()
+        if not text:
+            text = str(src.get("spu_id", ""))
+        docs.append(text)
+    return docs
+
+
+def call_rerank_service(
+    query: str,
+    docs: List[str],
+    service_url: str,
+    timeout_sec: float = DEFAULT_TIMEOUT_SEC,
+) -> Tuple[Optional[List[float]], Optional[Dict[str, Any]]]:
+    """
+    调用重排服务 POST /rerank，返回分数列表与 meta。
+
+    Args:
+        query: 搜索查询字符串
+        docs: 文档文本列表（与 ES hits 顺序一致）
+        service_url: 完整 URL，如 http://127.0.0.1:6007/rerank
+        timeout_sec: 请求超时秒数
+
+    Returns:
+        (scores, meta)：成功时 scores 与 docs 等长，meta 为服务返回的 meta；
+        失败时返回 (None, None)
+    """
+    if not docs:
+        return [], {}
+    try:
+        import requests
+        payload = {"query": (query or "").strip(), "docs": docs}
+        response = requests.post(service_url, json=payload, timeout=timeout_sec)
+        if response.status_code != 200:
+            logger.warning(
+                "Rerank service HTTP %s: %s",
+                response.status_code,
+                (response.text or "")[:200],
+            )
+            return None, None
+        data = response.json()
+        scores = data.get("scores")
+        if not isinstance(scores, list):
+            return None, None
+        return scores, data.get("meta") or {}
+    except (requests.exceptions.ReadTimeout, requests.exceptions.ConnectTimeout) as e:
+        logger.warning(
+            "Rerank request timed out after %.1fs (docs=%d); returning ES order. %s",
+            timeout_sec, len(docs), e,
+        )
+        return None, None
+    except Exception as e:
+        logger.warning("Rerank request failed: %s", e, exc_info=True)
+        return None, None
+
+
+def fuse_scores_and_resort(
+    es_hits: List[Dict[str, Any]],
+    rerank_scores: List[float],
+    weight_es: float = DEFAULT_WEIGHT_ES,
+    weight_ai: float = DEFAULT_WEIGHT_AI,
+) -> List[Dict[str, Any]]:
+    """
+    将 ES 分数与重排分数线性融合，写回每条 hit 的 _score，并按融合分数降序重排。
+
+    对每条 hit 会写入：
+    - _original_score: 原始 ES 分数
+    - _ai_rerank_score: 重排服务返回的分数
+    - _fused_score: 融合分数
+    - _score: 置为融合分数（供后续 ResultFormatter 使用）
+
+    Args:
+        es_hits: ES hits 列表（会被原地修改）
+        rerank_scores: 与 es_hits 等长的重排分数列表
+        weight_es: ES 归一化分数权重
+        weight_ai: 重排分数权重
+
+    Returns:
+        每条文档的融合调试信息列表，用于 debug_info
+    """
+    n = len(es_hits)
+    if n == 0 or len(rerank_scores) != n:
+        return []
+
+    # 收集 ES 原始分数
+    es_scores: List[float] = []
+    for hit in es_hits:
+        raw = hit.get("_score")
+        try:
+            es_scores.append(float(raw) if raw is not None else 0.0)
+        except (TypeError, ValueError):
+            es_scores.append(0.0)
+
+    max_es = max(es_scores) if es_scores else 0.0
+    fused_debug: List[Dict[str, Any]] = []
+
+    for idx, hit in enumerate(es_hits):
+        es_score = es_scores[idx]
+        ai_score_raw = rerank_scores[idx]
+        try:
+            ai_score = float(ai_score_raw)
+        except (TypeError, ValueError):
+            ai_score = 0.0
+
+        es_norm = (es_score / max_es) if max_es > 0 else 0.0
+        fused = weight_es * es_norm + weight_ai * ai_score
+
+        hit["_original_score"] = hit.get("_score")
+        hit["_ai_rerank_score"] = ai_score
+        hit["_fused_score"] = fused
+        hit["_score"] = fused
+
+        fused_debug.append({
+            "doc_id": hit.get("_id"),
+            "es_score": es_score,
+            "es_score_norm": es_norm,
+            "ai_rerank_score": ai_score,
+            "fused_score": fused,
+        })
+
+    # 按融合分数降序重排
+    es_hits.sort(
+        key=lambda h: h.get("_fused_score", h.get("_score", 0.0)),
+        reverse=True,
+    )
+    return fused_debug
+
+
+def run_rerank(
+    query: str,
+    es_response: Dict[str, Any],
+    language: str = "zh",
+    service_url: Optional[str] = None,
+    timeout_sec: float = DEFAULT_TIMEOUT_SEC,
+    weight_es: float = DEFAULT_WEIGHT_ES,
+    weight_ai: float = DEFAULT_WEIGHT_AI,
+) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]], List[Dict[str, Any]]]:
+    """
+    完整重排流程：从 es_response 取 hits -> 构造 docs -> 调服务 -> 融合分数并重排 -> 更新 max_score。
+
+    Args:
+        query: 搜索查询
+        es_response: ES 原始响应（其中的 hits["hits"] 会被原地修改）
+        language: 文档文本使用的语言
+        service_url: 重排服务 URL，为 None 时使用默认 127.0.0.1:6007
+        timeout_sec: 请求超时
+        weight_es: ES 分数权重
+        weight_ai: 重排分数权重
+
+    Returns:
+        (es_response, rerank_meta, fused_debug):
+        - es_response: 已更新 hits 与 max_score 的响应（同一引用）
+        - rerank_meta: 重排服务返回的 meta，失败时为 None
+        - fused_debug: 每条文档的融合信息，供 debug 使用
+    """
+    try:
+        from reranker.config import CONFIG as RERANKER_CONFIG
+    except Exception:
+        RERANKER_CONFIG = None
+
+    url = service_url
+    if not url and RERANKER_CONFIG is not None:
+        url = f"http://127.0.0.1:{RERANKER_CONFIG.PORT}/rerank"
+    if not url:
+        url = "http://127.0.0.1:6007/rerank"
+
+    hits = es_response.get("hits", {}).get("hits") or []
+    if not hits:
+        return es_response, None, []
+
+    docs = build_docs_from_hits(hits, language=language)
+    scores, meta = call_rerank_service(query, docs, url, timeout_sec=timeout_sec)
+
+    if scores is None or len(scores) != len(hits):
+        return es_response, None, []
+
+    fused_debug = fuse_scores_and_resort(
+        hits,
+        scores,
+        weight_es=weight_es,
+        weight_ai=weight_ai,
+    )
+
+    # 更新 max_score 为融合后的最高分
+    if hits:
+        top = hits[0].get("_fused_score", hits[0].get("_score", 0.0)) or 0.0
+        if "hits" in es_response:
+            es_response["hits"]["max_score"] = top
+
+    return es_response, meta, fused_debug
@@ -1,171 +0,0 @@
-"""
-Reranking engine for post-processing search result scoring.
-
-本地重排引擎，用于ES返回结果后的二次排序。
-当前状态：已禁用，优先使用ES的function_score。
-
-Supports expression-based ranking with functions like:
-- bm25(): Base BM25 text relevance score
-- text_embedding_relevance(): KNN embedding similarity
-- field_value(field): Use field value in scoring
-- timeliness(date_field): Time decay function
-"""
-
-import re
-from typing import Dict, Any, List, Optional
-import math
-
-
-class RerankEngine:
-    """
-    本地重排引擎（当前禁用）
-    
-    功能：对ES返回的结果进行二次打分和排序
-    用途：复杂的自定义排序逻辑、实时个性化等
-    """
-
-    def __init__(self, ranking_expression: str, enabled: bool = False):
-        """
-        Initialize rerank engine.
-
-        Args:
-            ranking_expression: Ranking expression string
-                              Example: "bm25() + 0.2*text_embedding_relevance() + general_score*2"
-            enabled: Whether local reranking is enabled (default: False)
-        """
-        self.enabled = enabled
-        self.expression = ranking_expression
-        self.parsed_terms = []
-        
-        if enabled:
-            self.parsed_terms = self._parse_expression(ranking_expression)
-
-    def _parse_expression(self, expression: str) -> List[Dict[str, Any]]:
-        """
-        Parse ranking expression into terms.
-
-        Args:
-            expression: Ranking expression
-
-        Returns:
-            List of term dictionaries
-        """
-        terms = []
-
-        # Pattern to match: coefficient * function() or field_name
-        # Example: "0.2*text_embedding_relevance()" or "general_score*2"
-        pattern = r'([+-]?\s*\d*\.?\d*)\s*\*?\s*([a-zA-Z_]\w*(?:\([^)]*\))?)'
-
-        for match in re.finditer(pattern, expression):
-            coef_str = match.group(1).strip()
-            func_str = match.group(2).strip()
-
-            # Parse coefficient
-            if coef_str in ['', '+']:
-                coefficient = 1.0
-            elif coef_str == '-':
-                coefficient = -1.0
-            else:
-                try:
-                    coefficient = float(coef_str)
-                except ValueError:
-                    coefficient = 1.0
-
-            # Check if function or field
-            if '(' in func_str:
-                # Function call
-                func_name = func_str[:func_str.index('(')]
-                args_str = func_str[func_str.index('(') + 1:func_str.rindex(')')]
-                args = [arg.strip() for arg in args_str.split(',')] if args_str else []
-
-                terms.append({
-                    'type': 'function',
-                    'name': func_name,
-                    'args': args,
-                    'coefficient': coefficient
-                })
-            else:
-                # Field reference
-                terms.append({
-                    'type': 'field',
-                    'name': func_str,
-                    'coefficient': coefficient
-                })
-
-        return terms
-
-    def calculate_score(
-        self,
-        hit: Dict[str, Any],
-        base_score: float,
-        knn_score: Optional[float] = None
-    ) -> float:
-        """
-        Calculate final score for a search result.
-
-        Args:
-            hit: ES hit document
-            base_score: Base BM25 score
-            knn_score: KNN similarity score (if available)
-
-        Returns:
-            Final calculated score
-        """
-        if not self.enabled:
-            return base_score
-        
-        score = 0.0
-        source = hit.get('_source', {})
-
-        for term in self.parsed_terms:
-            term_value = 0.0
-
-            if term['type'] == 'function':
-                func_name = term['name']
-
-                if func_name == 'bm25':
-                    term_value = base_score
-
-                elif func_name == 'text_embedding_relevance':
-                    term_value = knn_score if knn_score is not None else 0.0
-
-                elif func_name == 'timeliness':
-                    # Time decay function
-                    if term['args']:
-                        date_field = term['args'][0]
-                        if date_field in source:
-                            # Simple time decay (would need actual implementation)
-                            term_value = 1.0
-                    else:
-                        term_value = 1.0
-
-                elif func_name == 'field_value':
-                    # Get field value
-                    if term['args'] and term['args'][0] in source:
-                        field_value = source[term['args'][0]]
-                        try:
-                            term_value = float(field_value)
-                        except (ValueError, TypeError):
-                            term_value = 0.0
-
-            elif term['type'] == 'field':
-                # Direct field reference
-                field_name = term['name']
-                if field_name in source:
-                    try:
-                        term_value = float(source[field_name])
-                    except (ValueError, TypeError):
-                        term_value = 0.0
-
-            score += term['coefficient'] * term_value
-
-        return score
-
-    def get_expression(self) -> str:
-        """Get ranking expression."""
-        return self.expression
-
-    def get_terms(self) -> List[Dict[str, Any]]:
-        """Get parsed expression terms."""
-        return self.parsed_terms
-
@@ -13,7 +13,6 @@ from query import QueryParser, ParsedQuery
 from embeddings import CLIPImageEncoder
 from .boolean_parser import BooleanParser, QueryNode
 from .es_query_builder import ESQueryBuilder
-from .rerank_engine import RerankEngine
 from config import SearchConfig
 from config.tenant_config_loader import get_tenant_config_loader
 from config.utils import get_match_fields_for_index
@@ -99,7 +98,6 @@ class Searcher:
  
         # Initialize components
         self.boolean_parser = BooleanParser()
-        self.rerank_engine = RerankEngine(config.ranking.expression, enabled=False)
  
         # Get match fields from config
         self.match_fields = get_match_fields_for_index(config, "default")
@@ -137,6 +135,7 @@ class Searcher:
         debug: bool = False,
         language: str = "en",
         sku_filter_dimension: Optional[List[str]] = None,
+        ai_search: bool = False,
     ) -> SearchResult:
         """
         Execute search query (外部友好格式).
@@ -168,15 +167,21 @@ class Searcher:
         index_langs = tenant_cfg.get("index_languages") or []
         enable_translation = len(index_langs) > 0
         enable_embedding = self.config.query_config.enable_text_embedding
-        enable_rerank = False  # Temporarily disabled
+        # 重排仅由请求参数 ai_search 控制，唯一实现为调用外部 BGE 重排服务
+        enable_rerank = bool(ai_search)
+        rerank_window = self.config.rerank.rerank_window or 1000
+        # 若开启重排且请求范围在窗口内：从 ES 取前 rerank_window 条、重排后再按 from/size 分页；否则不重排，按原 from/size 查 ES
+        in_rerank_window = enable_rerank and (from_ + size) <= rerank_window
+        es_fetch_from = 0 if in_rerank_window else from_
+        es_fetch_size = rerank_window if in_rerank_window else size
  
         # Start timing
         context.start_stage(RequestContextStage.TOTAL)
  
         context.logger.info(
             f"开始搜索请求 | 查询: '{query}' | 参数: size={size}, from_={from_}, "
-            f"enable_translation={enable_translation}, enable_embedding={enable_embedding}, "
-            f"enable_rerank={enable_rerank}, min_score={min_score}",
+            f"enable_rerank={enable_rerank}, in_rerank_window={in_rerank_window}, es_fetch=({es_fetch_from},{es_fetch_size}) | "
+            f"enable_translation={enable_translation}, enable_embedding={enable_embedding}, min_score={min_score}",
             extra={'reqid': context.reqid, 'uid': context.uid}
         )
  
@@ -184,6 +189,9 @@ class Searcher:
         context.metadata['search_params'] = {
             'size': size,
             'from_': from_,
+            'es_fetch_from': es_fetch_from,
+            'es_fetch_size': es_fetch_size,
+            'in_rerank_window': in_rerank_window,
             'filters': filters,
             'range_filters': range_filters,
             'facets': facets,
@@ -287,8 +295,8 @@ class Searcher:
                 filters=filters,
                 range_filters=range_filters,
                 facet_configs=facets,
-                size=size,
-                from_=from_,
+                size=es_fetch_size,
+                from_=es_fetch_from,
                 enable_knn=enable_embedding and parsed_query.query_vector is not None,
                 min_score=min_score,
                 parsed_query=parsed_query
@@ -336,12 +344,12 @@ class Searcher:
         # Step 4: Elasticsearch search
         context.start_stage(RequestContextStage.ELASTICSEARCH_SEARCH)
         try:
-            # Use tenant-specific index name
+            # Use tenant-specific index name（开启重排且在窗口内时已用 es_fetch_size/es_fetch_from）
             es_response = self.es_client.search(
                 index_name=index_name,
                 body=body_for_es,
-                size=size,
-                from_=from_
+                size=es_fetch_size,
+                from_=es_fetch_from
             )
  
             # Store ES response in context
@@ -365,6 +373,69 @@ class Searcher:
         finally:
             context.end_stage(RequestContextStage.ELASTICSEARCH_SEARCH)
  
+        # Optional Step 4.5: AI reranking（仅当请求范围在重排窗口内时执行）
+        if enable_rerank and in_rerank_window:
+            context.start_stage(RequestContextStage.RERANKING)
+            try:
+                from .rerank_client import run_rerank
+
+                rerank_query = parsed_query.original_query if parsed_query else query
+                rc = self.config.rerank
+                es_response, rerank_meta, fused_debug = run_rerank(
+                    query=rerank_query,
+                    es_response=es_response,
+                    language=language,
+                    service_url=rc.service_url,
+                    timeout_sec=rc.timeout_sec,
+                    weight_es=rc.weight_es,
+                    weight_ai=rc.weight_ai,
+                )
+
+                if rerank_meta is not None:
+                    try:
+                        from reranker.config import CONFIG as RERANKER_CONFIG
+                        rerank_url = f"http://127.0.0.1:{RERANKER_CONFIG.PORT}/rerank"
+                    except Exception:
+                        rerank_url = "http://127.0.0.1:6007/rerank"
+                    context.metadata.setdefault("rerank_info", {})
+                    context.metadata["rerank_info"].update({
+                        "service_url": rerank_url,
+                        "docs": len(es_response.get("hits", {}).get("hits") or []),
+                        "meta": rerank_meta,
+                    })
+                    context.store_intermediate_result("rerank_scores", fused_debug)
+                    context.logger.info(
+                        f"重排完成 | docs={len(fused_debug)} | meta={rerank_meta}",
+                        extra={'reqid': context.reqid, 'uid': context.uid}
+                    )
+            except Exception as e:
+                context.add_warning(f"Rerank failed: {e}")
+                context.logger.warning(
+                    f"调用重排服务失败 | error: {e}",
+                    extra={'reqid': context.reqid, 'uid': context.uid},
+                    exc_info=True,
+                )
+            finally:
+                context.end_stage(RequestContextStage.RERANKING)
+
+        # 当本次请求在重排窗口内时：已从 ES 取了 rerank_window 条并可能已重排，需按请求的 from/size 做分页切片
+        if in_rerank_window:
+            hits = es_response.get("hits", {}).get("hits") or []
+            sliced = hits[from_ : from_ + size]
+            es_response.setdefault("hits", {})["hits"] = sliced
+            if sliced:
+                slice_max = max((h.get("_score") for h in sliced), default=0.0)
+                try:
+                    es_response["hits"]["max_score"] = float(slice_max)
+                except (TypeError, ValueError):
+                    es_response["hits"]["max_score"] = 0.0
+            else:
+                es_response["hits"]["max_score"] = 0.0
+            context.logger.info(
+                f"重排分页切片 | from={from_}, size={size}, 返回={len(sliced)}条",
+                extra={'reqid': context.reqid, 'uid': context.uid}
+            )
+
         # Step 5: Result processing
         context.start_stage(RequestContextStage.RESULT_PROCESSING)
         try:
@@ -379,7 +450,7 @@ class Searcher:
                 total_value = total.get('value', 0)
             else:
                 total_value = total
-
+            # max_score 会在启用 AI 搜索时被更新为融合分数的最大值
             max_score = es_response.get('hits', {}).get('max_score') or 0.0
  
             # Format results using ResultFormatter
@@ -191,9 +191,7 @@ def temp_config_file() -&gt; Generator[str, None, None]:
             "functions": []
         },
         "rerank": {
-            "enabled": False,
-            "expression": "",
-            "description": ""
+            "rerank_window": 1000
         }
     }