1. rerank漏斗配置优化

2. +service_enabled_by_config() { reranker|reranker-fine|translator 如果被关闭，则run.sh all 不启动该服务

1. rerank漏斗配置优化
2. +service_enabled_by_config() { reranker|reranker-fine|translator 如果被关闭，则run.sh all 不启动该服务
tangwang
1 parent 984f14f9
Showing 8 changed files with 509 additions and 178 deletions Show diff stats
api/models.py
config/config.yaml
docs/issues/a
docs/issues/issue-2026-04-12-test-env.md
scripts/service_ctl.sh
search/searcher.py
tests/test_embedding_pipeline.py
tests/test_search_rerank_window.py
@@ -154,7 +154,8 @@ class SearchRequest(BaseModel):
     enable_rerank: Optional[bool] = Field(
         None,
         description=(
-            "是否开启重排（调用外部重排服务对 ES 结果进行二次排序）。"
+            "是否开启最终重排（调用外部 rerank 服务改写上一阶段顺序）。"
+            "关闭时仍保留 coarse/fine 流程，仅在 rerank 阶段保序透传。"
             "不传则使用服务端配置 rerank.enabled（默认开启）。"
         )
     )
@@ -292,8 +292,8 @@ function_score:
 # 粗排配置（仅融合 ES 文本/向量信号，不调用模型）
 coarse_rank:
   enabled: true
-  input_window: 700
-  output_window: 240
+  input_window: 480
+  output_window: 160
   fusion:
     es_bias: 10.0
     es_exponent: 0.05
@@ -309,6 +309,7 @@ coarse_rank:
     knn_exponent: 0.4
 # 精排配置（轻量 reranker）
+# enabled=false 时仍进入 fine 阶段，但保序透传，不调用 fine 模型服务
 fine_rank:
   enabled: false
   input_window: 160
@@ -319,6 +320,7 @@ fine_rank:
   service_profile: fine
 # 重排配置（provider/URL 在 services.rerank）
+# enabled=false 时仍进入 rerank 阶段，但保序透传，不调用最终 rerank 服务
 rerank:
   enabled: true
   rerank_window: 160
@@ -510,7 +512,7 @@ services:
       default:
         host: 0.0.0.0
         port: 6007
-        backend: qwen3_vllm_score
+        backend: bge
         runtime_dir: ./.runtime/reranker/default
       fine:
         host: 0.0.0.0
@@ -0,0 +1,16 @@
+120.76.41.98   端口22 用户名和密码：
+tw  twtw@123 （有sudo权限）
+这台机器上的目录/home/tw/saas-search 已经部署了本项目
+请帮我运行项目
+1. 帮我checkout一个test环境的分支，这个分支，把重排、翻译模型 都关闭掉，因为这台机gpu显存较小（embedding模型可以保留）
+2. 在这个分支，把服务都启动起来
+3. 使用docker，安装一个ES，参考本项目的文档 ES9*.md。因为这台机器已经有一个系统的elasticsearch，为了不相互干扰，将本项目依赖的es9安装到docker，并且在测试环境配置的es地址做适配的工作
+
+
+1. 不是要禁用6005，而是6005端口已经有对应的文本服务了，直接用就行
+2. 6005其实就是本项目的一个历史早期版本启动起来的，在另外一个目录：/home/tw/SearchEngine，请看他的启动配置
+nohup bash scripts/start_embedding_service.sh > log.start_embedding_service.0412 2>&1 &
+是这样启动起来的
+看他陪的文本是用的哪套方案、哪个模型，跟他对齐（我指的是当前的测试分支）
+
+
@@ -20,6 +20,7 @@ CORE_SERVICES=(&quot;backend&quot; &quot;indexer&quot; &quot;frontend&quot; &quot;eval-web&quot;)
 OPTIONAL_SERVICES=("tei" "cnclip" "embedding" "embedding-image" "translator" "reranker")
 FULL_SERVICES=("${OPTIONAL_SERVICES[@]}" "${CORE_SERVICES[@]}")
 STOP_ORDER_SERVICES=("frontend" "eval-web" "indexer" "backend" "reranker" "translator" "embedding-image" "embedding" "cnclip" "tei")
+declare -Ag SERVICE_ENABLED_CACHE=()
 all_services() {
   echo "${FULL_SERVICES[@]}"
@@ -33,6 +34,72 @@ config_python_bin() {
   fi
 }
+service_enabled_by_config() {
+  local service="$1"
+  case "${service}" in
+    reranker|reranker-fine|translator)
+      ;;
+    *)
+      return 0
+      ;;
+  esac
+
+  if [ -n "${SERVICE_ENABLED_CACHE[${service}]+x}" ]; then
+    [ "${SERVICE_ENABLED_CACHE[${service}]}" = "1" ]
+    return
+  fi
+
+  local pybin
+  pybin="$(config_python_bin)"
+
+  local enabled
+  if ! enabled="$(
+    SERVICE_NAME="${service}" \
+    PYTHONPATH="${PROJECT_ROOT}${PYTHONPATH:+:${PYTHONPATH}}" \
+    "${pybin}" - <<'PY'
+from config.loader import get_app_config
+import os
+
+service = os.environ["SERVICE_NAME"]
+cfg = get_app_config()
+
+enabled = True
+if service == "reranker":
+    enabled = bool(cfg.search.rerank.enabled)
+elif service == "reranker-fine":
+    enabled = bool(cfg.search.fine_rank.enabled)
+elif service == "translator":
+    capabilities = dict(cfg.services.translation.capabilities or {})
+    enabled = any(bool((value or {}).get("enabled", True)) for value in capabilities.values())
+
+print("1" if enabled else "0")
+PY
+  )"; then
+    echo "[warn] failed to read config state for ${service}; defaulting to enabled" >&2
+    enabled="1"
+  fi
+
+  SERVICE_ENABLED_CACHE["${service}"]="${enabled}"
+  [ "${enabled}" = "1" ]
+}
+
+filter_disabled_targets() {
+  local targets="$1"
+  local verbose="${2:-quiet}"
+  local out=""
+  local svc
+
+  for svc in ${targets}; do
+    if service_enabled_by_config "${svc}"; then
+      out="${out} ${svc}"
+    elif [ "${verbose}" = "verbose" ]; then
+      echo "[skip] ${svc} disabled by config" >&2
+    fi
+  done
+
+  echo "${out# }"
+}
+
 reranker_instance_for_service() {
   local service="$1"
   case "${service}" in
@@ -468,6 +535,16 @@ stop_monitor_daemon() {
 start_monitor_daemon() {
   local targets="$1"
+  if [ -z "${targets}" ]; then
+    if is_monitor_daemon_running; then
+      echo "[info] no enabled services to monitor; stopping monitor daemon"
+      stop_monitor_daemon
+    else
+      echo "[info] no enabled services to monitor"
+    fi
+    return 0
+  fi
+
   local pf
   pf="$(monitor_pid_file)"
   local tf
@@ -581,6 +658,10 @@ wait_for_startup_health() {
 start_one() {
   local service="$1"
   cd "${PROJECT_ROOT}"
+  if ! service_enabled_by_config "${service}"; then
+    echo "[skip] ${service} disabled by config"
+    return 0
+  fi
   local cmd
   if ! cmd="$(service_start_cmd "${service}")"; then
     echo "[error] unknown service: ${service}" >&2
@@ -953,6 +1034,7 @@ main() {
   load_env_file "${PROJECT_ROOT}/.env"
   local targets=""
+  local effective_targets=""
   local monitor_was_running=0
   local monitor_prev_targets=""
   local auto_monitor_on_start="${SERVICE_CTL_AUTO_MONITOR_ON_START:-1}"
@@ -976,12 +1058,23 @@ main() {
       ;;
   esac
+  effective_targets="${targets}"
+  case "${action}" in
+    up|start|restart|monitor|monitor-start)
+      effective_targets="$(filter_disabled_targets "${targets}" "verbose")"
+      ;;
+  esac
+
   case "${action}" in
     up)
-      for svc in ${targets}; do
+      if [ -z "${effective_targets}" ]; then
+        echo "[info] no enabled services in target set"
+        exit 0
+      fi
+      for svc in ${effective_targets}; do
         start_one "${svc}"
       done
-      start_monitor_daemon "${targets}"
+      start_monitor_daemon "${effective_targets}"
       ;;
     down)
       stop_monitor_daemon
@@ -990,11 +1083,15 @@ main() {
       done
       ;;
     start)
-      for svc in ${targets}; do
+      if [ -z "${effective_targets}" ]; then
+        echo "[info] no enabled services in target set"
+        exit 0
+      fi
+      for svc in ${effective_targets}; do
         start_one "${svc}"
       done
       if [ "${auto_monitor_on_start}" = "1" ]; then
-        start_monitor_daemon "$(merge_targets "$(monitor_current_targets)" "${targets}")"
+        start_monitor_daemon "$(merge_targets "$(monitor_current_targets)" "${effective_targets}")"
       fi
       ;;
     stop)
@@ -1025,16 +1122,17 @@ main() {
       for svc in ${restart_stop_targets}; do
         stop_one "${svc}"
       done
-      for svc in ${targets}; do
+      for svc in ${effective_targets}; do
         start_one "${svc}"
       done
       if [ "${monitor_was_running}" -eq 1 ]; then
         monitor_prev_targets="$(normalize_targets "${monitor_prev_targets}")"
+        monitor_prev_targets="$(filter_disabled_targets "${monitor_prev_targets}" "quiet")"
         monitor_prev_targets="$(apply_target_order monitor "${monitor_prev_targets}")"
-        [ -z "${monitor_prev_targets}" ] && monitor_prev_targets="${targets}"
+        [ -z "${monitor_prev_targets}" ] && monitor_prev_targets="${effective_targets}"
         start_monitor_daemon "${monitor_prev_targets}"
       elif [ "${auto_monitor_on_start}" = "1" ]; then
-        start_monitor_daemon "$(merge_targets "$(monitor_current_targets)" "${targets}")"
+        start_monitor_daemon "$(merge_targets "$(monitor_current_targets)" "${effective_targets}")"
       fi
       ;;
     status)
@@ -1044,10 +1142,14 @@ main() {
       monitor_daemon_status
       ;;
     monitor)
-      monitor_services "${targets}"
+      if [ -z "${effective_targets}" ]; then
+        echo "[info] no enabled services in target set"
+        exit 0
+      fi
+      monitor_services "${effective_targets}"
       ;;
     monitor-start)
-      start_monitor_daemon "${targets}"
+      start_monitor_daemon "${effective_targets}"
       ;;
     monitor-stop)
       stop_monitor_daemon
@@ -401,7 +401,9 @@ class Searcher:
             language: Response / field selection language hint (e.g. zh, en)
             sku_filter_dimension: SKU grouping dimensions for per-SPU variant pick
             enable_rerank: If None, use ``config.rerank.enabled``; if set, overrides
-                whether the rerank provider is invoked (subject to rerank window).
+                whether the final rerank provider is invoked (subject to rank window).
+                When false, the ranking pipeline still runs and rerank stage becomes
+                pass-through.
             rerank_query_template: Override for rerank query text template; None uses
                 ``config.rerank.rerank_query_template`` (e.g. ``"{query}"``).
             rerank_doc_template: Override for per-hit document text passed to rerank;
@@ -430,15 +432,16 @@ class Searcher:
         # 重排开关优先级：请求参数显式传值 > 服务端配置（默认开启）
         rerank_enabled_by_config = bool(rc.enabled)
         do_rerank = rerank_enabled_by_config if enable_rerank is None else bool(enable_rerank)
+        fine_enabled = bool(fine_cfg.enabled)
         rerank_window = rc.rerank_window
         coarse_input_window = max(rerank_window, int(coarse_cfg.input_window))
         coarse_output_window = max(rerank_window, int(coarse_cfg.output_window))
         fine_input_window = max(rerank_window, int(fine_cfg.input_window))
         fine_output_window = max(rerank_window, int(fine_cfg.output_window))
-        # 若开启重排且请求范围在窗口内：从 ES 取前 rerank_window 条、重排后再按 from/size 分页；否则不重排，按原 from/size 查 ES
-        in_rerank_window = do_rerank and (from_ + size) <= rerank_window
-        es_fetch_from = 0 if in_rerank_window else from_
-        es_fetch_size = coarse_input_window if in_rerank_window else size
+        # 多阶段排序窗口独立于最终 rerank 开关：即使关闭最终 rerank，也保留 coarse/fine 流程。
+        in_rank_window = (from_ + size) <= rerank_window
+        es_fetch_from = 0 if in_rank_window else from_
+        es_fetch_size = coarse_input_window if in_rank_window else size
         es_score_normalization_factor: Optional[float] = None
         initial_ranks_by_doc: Dict[str, int] = {}
@@ -455,7 +458,8 @@ class Searcher:
         context.logger.info(
             f"开始搜索请求 | 查询: '{query}' | 参数: size={size}, from_={from_}, "
             f"enable_rerank(request)={enable_rerank}, enable_rerank(config)={rerank_enabled_by_config}, "
-            f"enable_rerank(effective)={do_rerank}, in_rerank_window={in_rerank_window}, "
+            f"fine_enabled(config)={fine_enabled}, "
+            f"enable_rerank(effective)={do_rerank}, in_rank_window={in_rank_window}, "
             f"es_fetch=({es_fetch_from},{es_fetch_size}) | "
             f"index_languages={index_langs} | "
             f"enable_translation={enable_translation}, enable_embedding={enable_embedding}, min_score={min_score}",
@@ -468,8 +472,9 @@ class Searcher:
             'from_': from_,
             'es_fetch_from': es_fetch_from,
             'es_fetch_size': es_fetch_size,
-            'in_rerank_window': in_rerank_window,
+            'in_rank_window': in_rank_window,
             'rerank_enabled_by_config': rerank_enabled_by_config,
+            'fine_enabled': fine_enabled,
             'enable_rerank_request': enable_rerank,
             'rerank_query_template': effective_query_template,
             'rerank_doc_template': effective_doc_template,
@@ -494,6 +499,7 @@ class Searcher:
         context.metadata['feature_flags'] = {
             'translation_enabled': enable_translation,
             'embedding_enabled': enable_embedding,
+            'fine_enabled': fine_enabled,
             'rerank_enabled': do_rerank,
             'style_intent_enabled': bool(self.style_intent_registry.enabled),
         }
@@ -526,7 +532,7 @@ class Searcher:
                 f"语言: {parsed_query.detected_language} | "
                 f"关键词: {parsed_query.keywords_queries} | "
                 f"文本向量: {'是' if parsed_query.query_vector is not None else '否'} | "
-                f"图片向量: {'是' if getattr(parsed_query, 'image_query_vector', None) is not None else '否'}",
+                f"图片向量: {'是' if parsed_query.image_query_vector is not None else '否'}",
                 extra={'reqid': context.reqid, 'uid': context.uid}
             )
         except Exception as e:
@@ -545,17 +551,16 @@ class Searcher:
             # Generate tenant-specific index name
             index_name = get_tenant_index_name(tenant_id)
             # index_name = "search_products"
-            
+
             # No longer need to add tenant_id to filters since each tenant has its own index
+            image_query_vector = None
+            if enable_embedding:
+                image_query_vector = parsed_query.image_query_vector
             es_query = self.query_builder.build_query(
                 query_text=parsed_query.rewritten_query or parsed_query.query_normalized,
                 query_vector=parsed_query.query_vector if enable_embedding else None,
-                image_query_vector=(
-                    getattr(parsed_query, "image_query_vector", None)
-                    if enable_embedding
-                    else None
-                ),
+                image_query_vector=image_query_vector,
                 filters=filters,
                 range_filters=range_filters,
                 facet_configs=facets,
@@ -563,7 +568,7 @@ class Searcher:
                 from_=es_fetch_from,
                 enable_knn=enable_embedding and (
                     parsed_query.query_vector is not None
-                    or getattr(parsed_query, "image_query_vector", None) is not None
+                    or image_query_vector is not None
                 ),
                 min_score=min_score,
                 parsed_query=parsed_query,
@@ -587,8 +592,7 @@ class Searcher:
             # In multi-stage rank window, first pass only needs score signals for coarse rank.
             es_query_for_fetch = es_query
-            rerank_prefetch_source = None
-            if in_rerank_window:
+            if in_rank_window:
                 es_query_for_fetch = dict(es_query)
                 es_query_for_fetch["_source"] = False
@@ -597,31 +601,28 @@ class Searcher:
             # Store ES query in context
             context.store_intermediate_result('es_query', es_query)
-            if in_rerank_window and rerank_prefetch_source is not None:
-                context.store_intermediate_result('es_query_rerank_prefetch_source', rerank_prefetch_source)
             # Serialize ES query to compute a compact size + stable digest for correlation
             es_query_compact = json.dumps(es_query_for_fetch, ensure_ascii=False, separators=(",", ":"))
             es_query_digest = hashlib.sha256(es_query_compact.encode("utf-8")).hexdigest()[:16]
             knn_enabled = bool(enable_embedding and (
                 parsed_query.query_vector is not None
-                or getattr(parsed_query, "image_query_vector", None) is not None
+                or image_query_vector is not None
             ))
             vector_dims = int(len(parsed_query.query_vector)) if parsed_query.query_vector is not None else 0
             image_vector_dims = (
-                int(len(parsed_query.image_query_vector))
-                if getattr(parsed_query, "image_query_vector", None) is not None
+                int(len(image_query_vector))
+                if image_query_vector is not None
                 else 0
             )
             context.logger.info(
-                "ES query built | size: %s chars | digest: %s | KNN: %s | vector_dims: %s | image_vector_dims: %s | facets: %s | rerank_prefetch_source: %s",
+                "ES query built | size: %s chars | digest: %s | KNN: %s | vector_dims: %s | image_vector_dims: %s | facets: %s",
                 len(es_query_compact),
                 es_query_digest,
                 "yes" if knn_enabled else "no",
                 vector_dims,
                 image_vector_dims,
                 "yes" if facets else "no",
-                rerank_prefetch_source,
                 extra={'reqid': context.reqid, 'uid': context.uid}
             )
             _log_backend_verbose({
@@ -656,7 +657,7 @@ class Searcher:
                 body=body_for_es,
                 size=es_fetch_size,
                 from_=es_fetch_from,
-                include_named_queries_score=bool(do_rerank and in_rerank_window),
+                include_named_queries_score=bool(in_rank_window),
             )
             # Store ES response in context
@@ -698,10 +699,177 @@ class Searcher:
             context.end_stage(RequestContextStage.ELASTICSEARCH_SEARCH_PRIMARY)
         style_intent_decisions: Dict[str, SkuSelectionDecision] = {}
-        if do_rerank and in_rerank_window:
+        if in_rank_window:
             from dataclasses import asdict
             from config.services_config import get_rerank_backend_config, get_rerank_service_url
             from .rerank_client import coarse_resort_hits, run_lightweight_rerank, run_rerank
+            coarse_fusion_debug = asdict(coarse_cfg.fusion)
+            stage_fusion_debug = asdict(rc.fusion)
+
+            def _rank_map(stage_hits: List[Dict[str, Any]]) -> Dict[str, int]:
+                return {
+                    str(hit.get("_id")): rank
+                    for rank, hit in enumerate(stage_hits, 1)
+                    if hit.get("_id") is not None
+                }
+
+            def _stage_debug_info(
+                *,
+                enabled: bool,
+                applied: bool,
+                skipped_reason: Optional[str],
+                service_profile: Optional[str],
+                query_template: str,
+                doc_template: str,
+                docs_in: int,
+                docs_out: int,
+                top_n: int,
+                meta: Optional[Dict[str, Any]] = None,
+                backend: Optional[str] = None,
+                backend_model_name: Optional[str] = None,
+                service_url: Optional[str] = None,
+                model: Optional[str] = None,
+                fusion: Optional[Dict[str, Any]] = None,
+            ) -> Dict[str, Any]:
+                return {
+                    "enabled": enabled,
+                    "applied": applied,
+                    "passthrough": not applied,
+                    "skipped_reason": skipped_reason,
+                    "service_profile": service_profile,
+                    "service_url": service_url,
+                    "backend": backend,
+                    "model": model,
+                    "backend_model_name": backend_model_name,
+                    "query_template": query_template,
+                    "doc_template": doc_template,
+                    "query_text": str(query_template).format_map({"query": rerank_query}),
+                    "docs_in": docs_in,
+                    "docs_out": docs_out,
+                    "top_n": top_n,
+                    "meta": meta,
+                    "fusion": fusion,
+                }
+
+            def _run_optional_stage(
+                *,
+                stage: RequestContextStage,
+                stage_label: str,
+                enabled: bool,
+                stage_hits: List[Dict[str, Any]],
+                input_limit: int,
+                output_limit: int,
+                service_profile: Optional[str],
+                query_template: str,
+                doc_template: str,
+                top_n: int,
+                debug_key: Optional[str],
+                runner,
+            ) -> tuple[List[Dict[str, Any]], Dict[str, int], Optional[Dict[str, Any]]]:
+                context.start_stage(stage)
+                try:
+                    input_hits = list(stage_hits[:input_limit])
+                    output_hits = list(stage_hits[:output_limit])
+                    applied = False
+                    skip_reason: Optional[str] = None
+                    meta: Optional[Dict[str, Any]] = None
+                    debug_rows: Optional[List[Dict[str, Any]]] = None
+
+                    if enabled and input_hits:
+                        output_hits_candidate, applied, meta, debug_rows = runner(input_hits)
+                        if applied:
+                            output_hits = list((output_hits_candidate or input_hits)[:output_limit])
+                        else:
+                            skip_reason = "service_returned_none"
+                    else:
+                        skip_reason = "disabled" if not enabled else "no_hits"
+
+                    ranks = _rank_map(output_hits) if debug else {}
+                    stage_info = None
+                    if debug:
+                        if applied:
+                            backend_name, backend_cfg = get_rerank_backend_config(service_profile)
+                            stage_info = _stage_debug_info(
+                                enabled=True,
+                                applied=True,
+                                skipped_reason=None,
+                                service_profile=service_profile,
+                                service_url=get_rerank_service_url(profile=service_profile),
+                                backend=backend_name,
+                                backend_model_name=backend_cfg.get("model_name"),
+                                model=meta.get("model") if isinstance(meta, dict) else None,
+                                query_template=query_template,
+                                doc_template=doc_template,
+                                docs_in=len(input_hits),
+                                docs_out=len(output_hits),
+                                top_n=top_n,
+                                meta=meta,
+                                fusion=stage_fusion_debug,
+                            )
+                            if debug_key is not None and debug_rows is not None:
+                                context.store_intermediate_result(debug_key, debug_rows)
+                        else:
+                            stage_info = _stage_debug_info(
+                                enabled=enabled,
+                                applied=False,
+                                skipped_reason=skip_reason,
+                                service_profile=service_profile,
+                                query_template=query_template,
+                                doc_template=doc_template,
+                                docs_in=len(input_hits),
+                                docs_out=len(output_hits),
+                                top_n=top_n,
+                                fusion=stage_fusion_debug,
+                            )
+
+                    if applied:
+                        context.logger.info(
+                            "%s完成 | docs=%s | top_n=%s | meta=%s",
+                            stage_label,
+                            len(output_hits),
+                            top_n,
+                            meta,
+                            extra={'reqid': context.reqid, 'uid': context.uid}
+                        )
+                    else:
+                        context.logger.info(
+                            "%s透传 | reason=%s | docs=%s | top_n=%s",
+                            stage_label,
+                            skip_reason,
+                            len(output_hits),
+                            top_n,
+                            extra={'reqid': context.reqid, 'uid': context.uid}
+                        )
+                    return output_hits, ranks, stage_info
+                except Exception as e:
+                    output_hits = list(stage_hits[:output_limit])
+                    ranks = _rank_map(output_hits) if debug else {}
+                    stage_info = None
+                    if debug:
+                        stage_info = _stage_debug_info(
+                            enabled=enabled,
+                            applied=False,
+                            skipped_reason="error",
+                            service_profile=service_profile,
+                            query_template=query_template,
+                            doc_template=doc_template,
+                            docs_in=min(len(stage_hits), input_limit),
+                            docs_out=len(output_hits),
+                            top_n=top_n,
+                            meta={"error": str(e)},
+                            fusion=stage_fusion_debug,
+                        )
+                    context.add_warning(f"{stage_label} failed: {e}")
+                    context.logger.warning(
+                        "调用%s服务失败 | error: %s",
+                        stage_label,
+                        e,
+                        extra={'reqid': context.reqid, 'uid': context.uid},
+                        exc_info=True,
+                    )
+                    return output_hits, ranks, stage_info
+                finally:
+                    context.end_stage(stage)
             rerank_query = parsed_query.text_for_rerank() if parsed_query else query
             hits = es_response.get("hits", {}).get("hits") or []
@@ -716,17 +884,12 @@ class Searcher:
                 hits = hits[:coarse_output_window]
                 es_response.setdefault("hits", {})["hits"] = hits
                 if debug:
-                    coarse_ranks_by_doc = {
-                        str(hit.get("_id")): rank
-                        for rank, hit in enumerate(hits, 1)
-                        if hit.get("_id") is not None
+                    coarse_ranks_by_doc = _rank_map(hits)
+                    coarse_debug_info = {
+                        "docs_in": es_fetch_size,
+                        "docs_out": len(hits),
+                        "fusion": coarse_fusion_debug,
                     }
-                    if debug:
-                        coarse_debug_info = {
-                            "docs_in": es_fetch_size,
-                            "docs_out": len(hits),
-                            "fusion": asdict(coarse_cfg.fusion),
-                        }
                     context.store_intermediate_result("coarse_rank_scores", coarse_debug)
                 context.logger.info(
                     "粗排完成 | docs_in=%s | docs_out=%s",
@@ -777,72 +940,42 @@ class Searcher:
                         extra={'reqid': context.reqid, 'uid': context.uid}
                     )
-            fine_scores: Optional[List[float]] = None
-            hits = es_response.get("hits", {}).get("hits") or []
-            if fine_cfg.enabled and hits:
-                context.start_stage(RequestContextStage.FINE_RANKING)
-                try:
-                    fine_scores, fine_meta, fine_debug_rows = run_lightweight_rerank(
-                        query=rerank_query,
-                        es_hits=hits[:fine_input_window],
-                        language=language,
-                        timeout_sec=fine_cfg.timeout_sec,
-                        rerank_query_template=fine_query_template,
-                        rerank_doc_template=fine_doc_template,
-                        top_n=fine_output_window,
-                        debug=debug,
-                        fusion=rc.fusion,
-                        style_intent_selected_sku_boost=self.config.query_config.style_intent_selected_sku_boost,
-                        service_profile=fine_cfg.service_profile,
-                    )
-                    if fine_scores is not None:
-                        hits = hits[:fine_output_window]
-                        es_response["hits"]["hits"] = hits
-                        if debug:
-                            fine_ranks_by_doc = {
-                                str(hit.get("_id")): rank
-                                for rank, hit in enumerate(hits, 1)
-                                if hit.get("_id") is not None
-                            }
-                            fine_backend_name, fine_backend_cfg = get_rerank_backend_config(fine_cfg.service_profile)
-                            fine_debug_info = {
-                                "service_profile": fine_cfg.service_profile,
-                                "service_url": get_rerank_service_url(profile=fine_cfg.service_profile),
-                                "backend": fine_backend_name,
-                                "model": fine_meta.get("model") if isinstance(fine_meta, dict) else None,
-                                "backend_model_name": fine_backend_cfg.get("model_name"),
-                                "query_template": fine_query_template,
-                                "doc_template": fine_doc_template,
-                                "query_text": str(fine_query_template).format_map({"query": rerank_query}),
-                                "docs_in": min(len(fine_scores), fine_input_window),
-                                "docs_out": len(hits),
-                                "top_n": fine_output_window,
-                                "meta": fine_meta,
-                                "fusion": asdict(rc.fusion),
-                            }
-                            context.store_intermediate_result("fine_rank_scores", fine_debug_rows)
-                        context.logger.info(
-                            "精排完成 | docs=%s | top_n=%s | meta=%s",
-                            len(hits),
-                            fine_output_window,
-                            fine_meta,
-                            extra={'reqid': context.reqid, 'uid': context.uid}
-                        )
-                except Exception as e:
-                    context.add_warning(f"Fine rerank failed: {e}")
-                    context.logger.warning(
-                        f"调用精排服务失败 | error: {e}",
-                        extra={'reqid': context.reqid, 'uid': context.uid},
-                        exc_info=True,
-                    )
-                finally:
-                    context.end_stage(RequestContextStage.FINE_RANKING)
+            def _run_fine_stage(stage_input: List[Dict[str, Any]]):
+                fine_scores, fine_meta, fine_debug_rows = run_lightweight_rerank(
+                    query=rerank_query,
+                    es_hits=stage_input,
+                    language=language,
+                    timeout_sec=fine_cfg.timeout_sec,
+                    rerank_query_template=fine_query_template,
+                    rerank_doc_template=fine_doc_template,
+                    top_n=fine_output_window,
+                    debug=debug,
+                    fusion=rc.fusion,
+                    style_intent_selected_sku_boost=self.config.query_config.style_intent_selected_sku_boost,
+                    service_profile=fine_cfg.service_profile,
+                )
+                return stage_input, fine_scores is not None, fine_meta, fine_debug_rows
+
+            hits, fine_ranks_by_doc, fine_debug_info = _run_optional_stage(
+                stage=RequestContextStage.FINE_RANKING,
+                stage_label="精排",
+                enabled=fine_enabled,
+                stage_hits=es_response.get("hits", {}).get("hits") or [],
+                input_limit=fine_input_window,
+                output_limit=fine_output_window,
+                service_profile=fine_cfg.service_profile,
+                query_template=fine_query_template,
+                doc_template=fine_doc_template,
+                top_n=fine_output_window,
+                debug_key="fine_rank_scores",
+                runner=_run_fine_stage,
+            )
+            es_response["hits"]["hits"] = hits
-            context.start_stage(RequestContextStage.RERANKING)
-            try:
-                final_hits = es_response.get("hits", {}).get("hits") or []
-                final_input = final_hits[:rerank_window]
-                es_response["hits"]["hits"] = final_input
+            def _run_rerank_stage(stage_input: List[Dict[str, Any]]):
+                nonlocal es_response
+
+                es_response["hits"]["hits"] = stage_input
                 es_response, rerank_meta, fused_debug = run_rerank(
                     query=rerank_query,
                     es_response=es_response,
@@ -858,48 +991,31 @@ class Searcher:
                     service_profile=rc.service_profile,
                     style_intent_selected_sku_boost=self.config.query_config.style_intent_selected_sku_boost,
                 )
-
-                if rerank_meta is not None:
-                    if debug:
-                        rerank_ranks_by_doc = {
-                            str(hit.get("_id")): rank
-                            for rank, hit in enumerate(es_response.get("hits", {}).get("hits") or [], 1)
-                            if hit.get("_id") is not None
-                        }
-                        rerank_backend_name, rerank_backend_cfg = get_rerank_backend_config(rc.service_profile)
-                        rerank_debug_info = {
-                            "service_profile": rc.service_profile,
-                            "service_url": get_rerank_service_url(profile=rc.service_profile),
-                            "backend": rerank_backend_name,
-                            "model": rerank_meta.get("model") if isinstance(rerank_meta, dict) else None,
-                            "backend_model_name": rerank_backend_cfg.get("model_name"),
-                            "query_template": effective_query_template,
-                            "doc_template": effective_doc_template,
-                            "query_text": str(effective_query_template).format_map({"query": rerank_query}),
-                            "docs_in": len(final_input),
-                            "docs_out": len(es_response.get("hits", {}).get("hits") or []),
-                            "top_n": from_ + size,
-                            "meta": rerank_meta,
-                            "fusion": asdict(rc.fusion),
-                        }
-                        context.store_intermediate_result("rerank_scores", fused_debug)
-                    context.logger.info(
-                        f"重排完成 | docs={len(es_response.get('hits', {}).get('hits') or [])} | "
-                        f"top_n={from_ + size} | 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,
+                return (
+                    es_response.get("hits", {}).get("hits") or [],
+                    rerank_meta is not None,
+                    rerank_meta,
+                    fused_debug,
                 )
-            finally:
-                context.end_stage(RequestContextStage.RERANKING)
-        # 当本次请求在重排窗口内时：已按多阶段排序产出前 rerank_window 条，需按请求的 from/size 做分页切片
-        if in_rerank_window:
+            hits, rerank_ranks_by_doc, rerank_debug_info = _run_optional_stage(
+                stage=RequestContextStage.RERANKING,
+                stage_label="重排",
+                enabled=do_rerank,
+                stage_hits=es_response.get("hits", {}).get("hits") or [],
+                input_limit=rerank_window,
+                output_limit=rerank_window,
+                service_profile=rc.service_profile,
+                query_template=effective_query_template,
+                doc_template=effective_doc_template,
+                top_n=from_ + size,
+                debug_key="rerank_scores",
+                runner=_run_rerank_stage,
+            )
+            es_response["hits"]["hits"] = hits
+
+        # 当本次请求在排序窗口内时：已按多阶段排序产出前 rerank_window 条，需按请求的 from/size 做分页切片
+        if in_rank_window:
             hits = es_response.get("hits", {}).get("hits") or []
             sliced = hits[from_ : from_ + size]
             es_response.setdefault("hits", {})["hits"] = sliced
@@ -961,12 +1077,12 @@ class Searcher:
                         context.end_stage(RequestContextStage.ELASTICSEARCH_PAGE_FILL)
             context.logger.info(
-                f"重排分页切片 | from={from_}, size={size}, 返回={len(sliced)}条",
+                f"排序窗口分页切片 | from={from_}, size={size}, 返回={len(sliced)}条",
                 extra={'reqid': context.reqid, 'uid': context.uid}
             )
         # 非重排窗口：款式意图在 result_processing 之前执行，便于单独计时且与 ES 召回阶段衔接
-        if self._has_style_intent(parsed_query) and not in_rerank_window:
+        if self._has_style_intent(parsed_query) and not in_rank_window:
             es_hits_pre = es_response.get("hits", {}).get("hits") or []
             style_intent_decisions = self._apply_style_intent_to_hits(
                 es_hits_pre,
@@ -1259,7 +1375,7 @@ class Searcher:
         # Collect debug information if requested
         debug_info = None
         if debug:
-            query_tokens = getattr(parsed_query, "query_tokens", []) if parsed_query else []
+            query_tokens = parsed_query.query_tokens if parsed_query else []
             token_count = len(query_tokens)
             text_knn_is_long = token_count >= 5
             text_knn_k = self.query_builder.knn_text_k_long if text_knn_is_long else self.query_builder.knn_text_k
@@ -1279,7 +1395,7 @@ class Searcher:
                     "translations": context.query_analysis.translations,
                     "keywords_queries": context.query_analysis.keywords_queries,
                     "has_vector": context.query_analysis.query_vector is not None,
-                    "has_image_vector": getattr(parsed_query, "image_query_vector", None) is not None,
+                    "has_image_vector": parsed_query.image_query_vector is not None,
                     "query_tokens": query_tokens,
                     "intent_detection": context.get_intermediate_result("style_intent_profile"),
                 },
@@ -1298,9 +1414,10 @@ class Searcher:
                     },
                     "image_knn": {
                         "enabled": bool(
-                            enable_embedding
+                            self.image_embedding_field
+                            and enable_embedding
                             and parsed_query
-                            and getattr(parsed_query, "image_query_vector", None) is not None
+                            and image_query_vector is not None
                         ),
                         "k": self.query_builder.knn_image_k,
                         "num_candidates": self.query_builder.knn_image_num_candidates,
@@ -1311,9 +1428,8 @@ class Searcher:
                 "es_query_context": {
                     "es_fetch_from": es_fetch_from,
                     "es_fetch_size": es_fetch_size,
-                    "in_rerank_window": in_rerank_window,
-                    "rerank_prefetch_source": context.get_intermediate_result('es_query_rerank_prefetch_source'),
-                    "include_named_queries_score": bool(do_rerank and in_rerank_window),
+                    "in_rank_window": in_rank_window,
+                    "include_named_queries_score": bool(in_rank_window),
                 },
                 "es_response": {
                     "took_ms": es_response.get('took', 0),
@@ -1369,10 +1485,10 @@ class Searcher:
                     "retrieval_plan": debug_info["retrieval_plan"],
                     "ranking_windows": {
                         "es_fetch_size": es_fetch_size,
-                        "coarse_output_window": coarse_output_window if do_rerank and in_rerank_window else None,
-                        "fine_input_window": fine_input_window if do_rerank and in_rerank_window else None,
-                        "fine_output_window": fine_output_window if do_rerank and in_rerank_window else None,
-                        "rerank_window": rerank_window if do_rerank and in_rerank_window else None,
+                        "coarse_output_window": coarse_output_window if in_rank_window else None,
+                        "fine_input_window": fine_input_window if in_rank_window else None,
+                        "fine_output_window": fine_output_window if in_rank_window else None,
+                        "rerank_window": rerank_window if in_rank_window else None,
                         "page_from": from_,
                         "page_size": size,
                     },
+from dataclasses import asdict
 from typing import Any, Dict, List, Optional
 import numpy as np
 from __future__ import annotations
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, Dict, List
@@ -30,7 +30,10 @@ class _FakeParsedQuery:
     rewritten_query: str
     detected_language: str = "en"
     translations: Dict[str, str] = None
+    keywords_queries: Dict[str, str] = field(default_factory=dict)
     query_vector: Any = None
+    image_query_vector: Any = None
+    query_tokens: List[str] = field(default_factory=list)
     style_intent_profile: Any = None
     def text_for_rerank(self) -> str:
@@ -89,6 +92,15 @@ class _FakeQueryParser:
 class _FakeQueryBuilder:
+    knn_text_k = 120
+    knn_text_k_long = 160
+    knn_text_num_candidates = 400
+    knn_text_num_candidates_long = 500
+    knn_text_boost = 20.0
+    knn_image_k = 120
+    knn_image_num_candidates = 400
+    knn_image_boost = 20.0
+
     def build_query(self, **kwargs):
         return {
             "query": {"match_all": {}},
@@ -583,7 +595,7 @@ def test_searcher_rerank_prefetch_source_includes_sku_fields_when_style_intent_a
     }
-def test_searcher_skips_rerank_when_request_explicitly_false(monkeypatch):
+def test_searcher_keeps_previous_stage_order_when_request_explicitly_disables_rerank(monkeypatch):
     es_client = _FakeESClient()
     searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
     context = create_request_context(reqid="t2", uid="u2")
@@ -593,28 +605,95 @@ def test_searcher_skips_rerank_when_request_explicitly_false(monkeypatch):
         lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
     )
-    called: Dict[str, int] = {"count": 0}
+    called: Dict[str, int] = {"count": 0, "fine": 0}
+
+    def _fake_run_lightweight_rerank(**kwargs):
+        called["fine"] += 1
+        hits = kwargs["es_hits"]
+        for idx, hit in enumerate(hits):
+            hit["_fine_score"] = float(idx + 1)
+        hits.reverse()
+        return [hit["_fine_score"] for hit in hits], {"stage": "fine"}, []
     def _fake_run_rerank(**kwargs):
         called["count"] += 1
         return kwargs["es_response"], None, []
+    monkeypatch.setattr("search.rerank_client.run_lightweight_rerank", _fake_run_lightweight_rerank)
     monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
-    searcher.search(
+    result = searcher.search(
         query="toy",
         tenant_id="162",
         from_=20,
         size=10,
         context=context,
         enable_rerank=False,
+        debug=True,
     )
     assert called["count"] == 0
-    assert es_client.calls[0]["from_"] == 20
-    assert es_client.calls[0]["size"] == 10
-    assert es_client.calls[0]["include_named_queries_score"] is False
-    assert len(es_client.calls) == 1
+    assert called["fine"] == 1
+    assert es_client.calls[0]["from_"] == 0
+    assert es_client.calls[0]["size"] == searcher.config.coarse_rank.input_window
+    assert es_client.calls[0]["include_named_queries_score"] is True
+    assert len(es_client.calls) == 3
+    assert es_client.calls[2]["body"]["query"]["ids"]["values"] == [str(i) for i in range(363, 353, -1)]
+    assert len(result.results) == 10
+    assert [item.spu_id for item in result.results[:3]] == ["363", "362", "361"]
+    assert result.debug_info["rerank"]["enabled"] is False
+    assert result.debug_info["rerank"]["applied"] is False
+    assert result.debug_info["rerank"]["skipped_reason"] == "disabled"
+    assert result.debug_info["per_result"][0]["ranking_funnel"]["rerank"]["rank"] == 21
+
+
+def test_searcher_keeps_previous_stage_order_when_config_disables_rerank(monkeypatch):
+    es_client = _FakeESClient()
+    searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
+    context = create_request_context(reqid="t2b", uid="u2b")
+
+    monkeypatch.setattr(
+        "search.searcher.get_tenant_config_loader",
+        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
+    )
+
+    called: Dict[str, int] = {"count": 0, "fine": 0}
+
+    def _fake_run_lightweight_rerank(**kwargs):
+        called["fine"] += 1
+        hits = kwargs["es_hits"]
+        hits.reverse()
+        for idx, hit in enumerate(hits):
+            hit["_fine_score"] = float(len(hits) - idx)
+        return [hit["_fine_score"] for hit in hits], {"stage": "fine"}, []
+
+    def _fake_run_rerank(**kwargs):
+        called["count"] += 1
+        return kwargs["es_response"], None, []
+
+    monkeypatch.setattr("search.rerank_client.run_lightweight_rerank", _fake_run_lightweight_rerank)
+    monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
+
+    result = searcher.search(
+        query="toy",
+        tenant_id="162",
+        from_=0,
+        size=5,
+        context=context,
+        enable_rerank=None,
+        debug=True,
+    )
+
+    assert called["count"] == 0
+    assert called["fine"] == 1
+    assert es_client.calls[0]["from_"] == 0
+    assert es_client.calls[0]["size"] == searcher.config.coarse_rank.input_window
+    assert es_client.calls[0]["include_named_queries_score"] is True
+    assert len(result.results) == 5
+    assert [item.spu_id for item in result.results] == ["383", "382", "381", "380", "379"]
+    assert result.debug_info["rerank"]["enabled"] is False
+    assert result.debug_info["rerank"]["applied"] is False
+    assert result.debug_info["rerank"]["skipped_reason"] == "disabled"
 def test_searcher_skips_rerank_when_page_exceeds_window(monkeypatch):
@@ -919,7 +998,8 @@ def test_searcher_promotes_sku_by_embedding_when_query_has_no_direct_option_matc
 def test_searcher_debug_info_uses_initial_es_max_score_for_normalization(monkeypatch):
     es_client = _FakeESClient(total_hits=3)
-    searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
+    cfg = _build_search_config(rerank_enabled=False)
+    searcher = _build_searcher(cfg, es_client)
     context = create_request_context(reqid="dbg", uid="u-dbg")
     monkeypatch.setattr(
@@ -939,7 +1019,8 @@ def test_searcher_debug_info_uses_initial_es_max_score_for_normalization(monkeyp
     assert result.debug_info["query_analysis"]["index_languages"] == ["en", "zh"]
     assert result.debug_info["query_analysis"]["query_tokens"] == []
-    assert result.debug_info["es_query_context"]["es_fetch_size"] == 2
+    expected_es_fetch = max(cfg.rerank.rerank_window, cfg.coarse_rank.input_window)
+    assert result.debug_info["es_query_context"]["es_fetch_size"] == expected_es_fetch
     assert result.debug_info["es_response"]["es_score_normalization_factor"] == 3.0
     assert result.debug_info["per_result"][0]["initial_rank"] == 1
     assert result.debug_info["per_result"][0]["final_rank"] == 1
@@ -970,6 +1051,12 @@ def test_searcher_rerank_rank_change_falls_back_to_coarse_rank_when_fine_disable
         lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
     )
+    fine_called: Dict[str, int] = {"count": 0}
+
+    def _fake_run_lightweight_rerank(**kwargs):
+        fine_called["count"] += 1
+        return [], {"stage": "fine"}, []
+
     def _fake_run_rerank(**kwargs):
         hits = kwargs["es_response"]["hits"]["hits"]
         hits.reverse()
@@ -994,6 +1081,7 @@ def test_searcher_rerank_rank_change_falls_back_to_coarse_rank_when_fine_disable
             )
         return kwargs["es_response"], {"model": "final-reranker"}, fused_debug
+    monkeypatch.setattr("search.rerank_client.run_lightweight_rerank", _fake_run_lightweight_rerank)
     monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
     result = searcher.search(
@@ -1008,7 +1096,12 @@ def test_searcher_rerank_rank_change_falls_back_to_coarse_rank_when_fine_disable
     per_result = {row["spu_id"]: row for row in result.debug_info["per_result"]}
     moved = per_result["4"]["ranking_funnel"]
-    assert moved["fine_rank"]["rank"] is None
+    assert fine_called["count"] == 0
+    assert result.debug_info["fine_rank"]["enabled"] is False
+    assert result.debug_info["fine_rank"]["applied"] is False
+    assert result.debug_info["fine_rank"]["skipped_reason"] == "disabled"
+    assert moved["fine_rank"]["rank"] == 5
+    assert moved["fine_rank"]["rank_change"] == 0
     assert moved["rerank"]["rank"] == 1
     assert moved["rerank"]["rank_change"] == 4
     assert moved["final_page"]["rank_change"] == 0