tests/test_search_rerank_window.py

from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Dict, List
import numpy as np
import yaml
from config import (
    ConfigLoader,
    FunctionScoreConfig,
    IndexConfig,
    QueryConfig,
    RerankConfig,
    SPUConfig,
    SearchConfig,
)
from context import create_request_context
from query.style_intent import DetectedStyleIntent, StyleIntentProfile
from search.searcher import Searcher
@dataclass
class _FakeParsedQuery:
    original_query: str
    query_normalized: str
    rewritten_query: str
    detected_language: str = "en"
    translations: Dict[str, str] = None
    query_vector: Any = None
    style_intent_profile: Any = None
    def text_for_rerank(self) -> str:
        from query.query_parser import rerank_query_text
        return rerank_query_text(
            self.original_query,
            detected_language=self.detected_language,
            translations=self.translations,
        )
    def to_dict(self) -> Dict[str, Any]:
        return {
            "original_query": self.original_query,
            "query_normalized": self.query_normalized,
            "rewritten_query": self.rewritten_query,
            "detected_language": self.detected_language,
            "translations": self.translations or {},
            "style_intent_profile": (
                self.style_intent_profile.to_dict() if self.style_intent_profile is not None else None
            ),
        }
def _build_style_intent_profile(intent_type: str, canonical_value: str, *dimension_aliases: str) -> StyleIntentProfile:
    aliases = dimension_aliases or (intent_type,)
    return StyleIntentProfile(
        intents=(
            DetectedStyleIntent(
                intent_type=intent_type,
                canonical_value=canonical_value,
                matched_term=canonical_value,
                matched_query_text=canonical_value,
                attribute_terms=(canonical_value,),
                dimension_aliases=tuple(aliases),
            ),
        )
    )
class _FakeQueryParser:
    def parse(
        self,
        query: str,
        tenant_id: str,
        generate_vector: bool,
        context: Any,
        target_languages: Any = None,
    ):
        return _FakeParsedQuery(
            original_query=query,
            query_normalized=query,
            rewritten_query=query,
            translations={},
        )
class _FakeQueryBuilder:
    def build_query(self, **kwargs):
        return {
            "query": {"match_all": {}},
            "size": kwargs["size"],
            "from": kwargs["from_"],
        }
    def build_facets(self, facets: Any):
        return {}
    def add_sorting(self, es_query: Dict[str, Any], sort_by: str, sort_order: str):
        return es_query
class _FakeESClient:
    def __init__(self, total_hits: int = 5000):
        self.calls: List[Dict[str, Any]] = []
        self.total_hits = total_hits
    @staticmethod
    def _apply_source_filter(src: Dict[str, Any], source_spec: Any) -> Dict[str, Any]:
        if source_spec is None:
            return dict(src)
        if source_spec is False:
            return {}
        if isinstance(source_spec, dict):
            includes = source_spec.get("includes") or []
        elif isinstance(source_spec, list):
            includes = source_spec
        else:
            includes = []
        if not includes:
            return dict(src)
        return {k: v for k, v in src.items() if k in set(includes)}
    @staticmethod
    def _full_source(doc_id: str) -> Dict[str, Any]:
        return {
            "spu_id": doc_id,
            "title": {"en": f"product-{doc_id}"},
            "brief": {"en": f"brief-{doc_id}"},
            "vendor": {"en": f"vendor-{doc_id}"},
            "skus": [],
        }
    def search(
        self,
        index_name: str,
        body: Dict[str, Any],
        size: int,
        from_: int,
        include_named_queries_score: bool = False,
    ):
        self.calls.append(
            {
                "index_name": index_name,
                "body": body,
                "size": size,
                "from_": from_,
                "include_named_queries_score": include_named_queries_score,
            }
        )
        ids_query = (((body or {}).get("query") or {}).get("ids") or {}).get("values")
        source_spec = (body or {}).get("_source")
        if isinstance(ids_query, list):
            # Return reversed order intentionally; caller should restore original ranking order.
            ids = [str(i) for i in ids_query][::-1]
            hits = []
            for doc_id in ids:
                src = self._apply_source_filter(self._full_source(doc_id), source_spec)
                hit = {"_id": doc_id, "_score": 1.0}
                if source_spec is not False:
                    hit["_source"] = src
                hits.append(hit)
        else:
            end = min(from_ + size, self.total_hits)
            hits = []
            for i in range(from_, end):
                doc_id = str(i)
                src = self._apply_source_filter(self._full_source(doc_id), source_spec)
                hit = {"_id": doc_id, "_score": float(self.total_hits - i)}
                if source_spec is not False:
                    hit["_source"] = src
                hits.append(hit)
        return {
            "took": 8,
            "hits": {
                "total": {"value": self.total_hits},
                "max_score": hits[0]["_score"] if hits else 0.0,
                "hits": hits,
            },
        }
def _build_search_config(*, rerank_enabled: bool = True, rerank_window: int = 384):
    return SearchConfig(
        field_boosts={"title.en": 3.0},
        indexes=[IndexConfig(name="default", label="default", fields=["title.en"])],
        query_config=QueryConfig(enable_text_embedding=False, enable_query_rewrite=False),
        function_score=FunctionScoreConfig(),
        rerank=RerankConfig(enabled=rerank_enabled, rerank_window=rerank_window),
        spu_config=SPUConfig(enabled=False),
        es_index_name="test_products",
        es_settings={},
    )
def _build_searcher(config: SearchConfig, es_client: _FakeESClient) -> Searcher:
    searcher = Searcher(
        es_client=es_client,
        config=config,
        query_parser=_FakeQueryParser(),
    )
    searcher.query_builder = _FakeQueryBuilder()
    return searcher
class _FakeTextEncoder:
    def __init__(self, vectors: Dict[str, List[float]]):
        self.vectors = {
            key: np.array(value, dtype=np.float32)
            for key, value in vectors.items()
        }
    def encode(self, sentences, priority: int = 0, **kwargs):
        if isinstance(sentences, str):
            sentences = [sentences]
        return np.array([self.vectors[text] for text in sentences], dtype=object)
def test_config_loader_rerank_enabled_defaults_true(tmp_path: Path):
    config_data = {
        "es_index_name": "test_products",
        "field_boosts": {"title.en": 3.0},
        "indexes": [{"name": "default", "label": "default", "fields": ["title.en"]}],
        "query_config": {"supported_languages": ["en"], "default_language": "en"},
        "services": {
            "translation": {
                "service_url": "http://localhost:6005",
                "timeout_sec": 3.0,
                "default_model": "dummy-model",
                "default_scene": "general",
                "cache": {
                    "ttl_seconds": 60,
                    "sliding_expiration": True,
                },
                "capabilities": {
                    "dummy-model": {
                        "enabled": True,
                        "backend": "llm",
                        "use_cache": True,
                        "model": "dummy-model",
                        "base_url": "http://localhost:6005/v1",
                        "timeout_sec": 3.0,
                    }
                },
            },
            "embedding": {
                "provider": "http",
                "providers": {
                    "http": {
                        "text_base_url": "http://localhost:6005",
                        "image_base_url": "http://localhost:6008",
                    }
                },
                "backend": "tei",
                "backends": {
                    "tei": {
                        "base_url": "http://localhost:8080",
                        "timeout_sec": 3.0,
                        "model_id": "dummy-embedding-model",
                    }
                },
            },
            "rerank": {
                "provider": "http",
                "providers": {
                    "http": {
                        "base_url": "http://localhost:6007",
                        "service_url": "http://localhost:6007/rerank",
                    }
                },
                "backend": "bge",
                "backends": {
                    "bge": {
                        "model_name": "dummy-rerank-model",
                        "device": "cpu",
                        "use_fp16": False,
                        "batch_size": 8,
                        "max_length": 128,
                        "cache_dir": "./model_cache",
                        "enable_warmup": False,
                    }
                },
            },
        },
        "spu_config": {"enabled": False},
        "function_score": {"score_mode": "sum", "boost_mode": "multiply", "functions": []},
        "rerank": {"rerank_window": 384},
    }
    config_path = tmp_path / "config.yaml"
    config_path.write_text(yaml.safe_dump(config_data), encoding="utf-8")
    loader = ConfigLoader(config_path)
    loaded = loader.load_config(validate=False)
    assert loaded.rerank.enabled is True
def test_searcher_reranks_top_window_by_default(monkeypatch):
    es_client = _FakeESClient()
    searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
    context = create_request_context(reqid="t1", uid="u1")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
    )
    called: Dict[str, Any] = {"count": 0, "docs": 0}
    def _fake_run_lightweight_rerank(**kwargs):
        hits = kwargs["es_hits"]
        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
        called["docs"] = len(kwargs["es_response"]["hits"]["hits"])
        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_=20,
        size=10,
        context=context,
        enable_rerank=None,
    )
    assert called["count"] == 1
    assert called["docs"] == searcher.config.rerank.rerank_window
    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 es_client.calls[0]["body"]["_source"] is False
    assert len(es_client.calls) == 3
    assert es_client.calls[1]["size"] == max(
        searcher.config.coarse_rank.output_window,
        searcher.config.rerank.rerank_window,
    )
    assert es_client.calls[1]["from_"] == 0
    assert es_client.calls[2]["size"] == 10
    assert es_client.calls[2]["from_"] == 0
    assert es_client.calls[2]["body"]["query"]["ids"]["values"] == [str(i) for i in range(20, 30)]
    assert len(result.results) == 10
    assert result.results[0].spu_id == "20"
    assert result.results[0].brief == "brief-20"
def test_searcher_rerank_prefetch_source_follows_doc_template(monkeypatch):
    es_client = _FakeESClient()
    searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
    context = create_request_context(reqid="t1b", uid="u1b")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
    )
    monkeypatch.setattr(
        "search.rerank_client.run_lightweight_rerank",
        lambda **kwargs: ([1.0] * len(kwargs["es_hits"]), {"stage": "fine"}, []),
    )
    monkeypatch.setattr("search.rerank_client.run_rerank", lambda **kwargs: (kwargs["es_response"], None, []))
    searcher.search(
        query="toy",
        tenant_id="162",
        from_=0,
        size=5,
        context=context,
        enable_rerank=None,
        rerank_doc_template="{title} {vendor} {brief}",
    )
    assert es_client.calls[0]["body"]["_source"] is False
    assert es_client.calls[1]["body"]["_source"] == {"includes": ["brief", "title", "vendor"]}
def test_searcher_rerank_prefetch_source_includes_sku_fields_when_style_intent_active(monkeypatch):
    es_client = _FakeESClient()
    searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
    context = create_request_context(reqid="t1c", uid="u1c")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
    )
    monkeypatch.setattr(
        "search.rerank_client.run_lightweight_rerank",
        lambda **kwargs: ([1.0] * len(kwargs["es_hits"]), {"stage": "fine"}, []),
    )
    monkeypatch.setattr(
        "search.rerank_client.run_rerank",
        lambda **kwargs: (kwargs["es_response"], None, []),
    )
    class _IntentQueryParser:
        text_encoder = None
        def parse(
            self,
            query: str,
            tenant_id: str,
            generate_vector: bool,
            context: Any,
            target_languages: Any = None,
        ):
            return _FakeParsedQuery(
                original_query=query,
                query_normalized=query,
                rewritten_query=query,
                translations={},
                style_intent_profile=_build_style_intent_profile(
                    "color", "black", "color", "colors", "颜色"
                ),
            )
    searcher.query_parser = _IntentQueryParser()
    searcher.search(
        query="black dress",
        tenant_id="162",
        from_=0,
        size=5,
        context=context,
        enable_rerank=None,
    )
    assert es_client.calls[0]["body"]["_source"] is False
    assert es_client.calls[1]["body"]["_source"] == {
        "includes": ["option1_name", "option2_name", "option3_name", "skus", "title"]
    }
def test_searcher_skips_rerank_when_request_explicitly_false(monkeypatch):
    es_client = _FakeESClient()
    searcher = _build_searcher(_build_search_config(rerank_enabled=True), es_client)
    context = create_request_context(reqid="t2", uid="u2")
    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}
    def _fake_run_rerank(**kwargs):
        called["count"] += 1
        return kwargs["es_response"], None, []
    monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
    searcher.search(
        query="toy",
        tenant_id="162",
        from_=20,
        size=10,
        context=context,
        enable_rerank=False,
    )
    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
def test_searcher_skips_rerank_when_page_exceeds_window(monkeypatch):
    es_client = _FakeESClient()
    searcher = _build_searcher(_build_search_config(rerank_enabled=True, rerank_window=384), es_client)
    context = create_request_context(reqid="t3", uid="u3")
    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}
    def _fake_run_rerank(**kwargs):
        called["count"] += 1
        return kwargs["es_response"], None, []
    monkeypatch.setattr("search.rerank_client.run_rerank", _fake_run_rerank)
    searcher.search(
        query="toy",
        tenant_id="162",
        from_=995,
        size=10,
        context=context,
        enable_rerank=None,
    )
    assert called["count"] == 0
    assert es_client.calls[0]["from_"] == 995
    assert es_client.calls[0]["size"] == 10
    assert es_client.calls[0]["include_named_queries_score"] is False
    assert len(es_client.calls) == 1
def test_searcher_promotes_sku_when_option1_matches_translated_query(monkeypatch):
    es_client = _FakeESClient(total_hits=1)
    searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
    context = create_request_context(reqid="sku-text", uid="u-sku-text")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en", "zh"]}),
    )
    class _TranslatedQueryParser:
        text_encoder = None
        def parse(
            self,
            query: str,
            tenant_id: str,
            generate_vector: bool,
            context: Any,
            target_languages: Any = None,
        ):
            return _FakeParsedQuery(
                original_query=query,
                query_normalized=query,
                rewritten_query=query,
                translations={"en": "black dress"},
                style_intent_profile=_build_style_intent_profile(
                    "color", "black", "color", "colors", "颜色"
                ),
            )
    searcher.query_parser = _TranslatedQueryParser()
    def _full_source_with_skus(doc_id: str) -> Dict[str, Any]:
        return {
            "spu_id": doc_id,
            "title": {"en": f"product-{doc_id}"},
            "brief": {"en": f"brief-{doc_id}"},
            "vendor": {"en": f"vendor-{doc_id}"},
            "option1_name": "Color",
            "image_url": "https://img/default.jpg",
            "skus": [
                {"sku_id": "sku-red", "option1_value": "Red", "image_src": "https://img/red.jpg"},
                {"sku_id": "sku-black", "option1_value": "Black", "image_src": "https://img/black.jpg"},
            ],
        }
    monkeypatch.setattr(_FakeESClient, "_full_source", staticmethod(_full_source_with_skus))
    result = searcher.search(
        query="黑色 连衣裙",
        tenant_id="162",
        from_=0,
        size=1,
        context=context,
        enable_rerank=False,
    )
    assert len(result.results) == 1
    assert result.results[0].skus[0].sku_id == "sku-black"
    assert result.results[0].image_url == "https://img/black.jpg"
def test_searcher_uses_first_text_match_without_comparing_all_matches(monkeypatch):
    es_client = _FakeESClient(total_hits=1)
    searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
    context = create_request_context(reqid="sku-first-text", uid="u-sku-first-text")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
    )
    class _TextMatchQueryParser:
        text_encoder = None
        def parse(
            self,
            query: str,
            tenant_id: str,
            generate_vector: bool,
            context: Any,
            target_languages: Any = None,
        ):
            return _FakeParsedQuery(
                original_query=query,
                query_normalized=query,
                rewritten_query=query,
                translations={},
                style_intent_profile=_build_style_intent_profile(
                    "color", "black", "color", "colors", "颜色"
                ),
            )
    searcher.query_parser = _TextMatchQueryParser()
    def _full_source_with_multiple_text_matches(doc_id: str) -> Dict[str, Any]:
        return {
            "spu_id": doc_id,
            "title": {"en": f"product-{doc_id}"},
            "brief": {"en": f"brief-{doc_id}"},
            "vendor": {"en": f"vendor-{doc_id}"},
            "option1_name": "Color",
            "image_url": "https://img/default.jpg",
            "skus": [
                {"sku_id": "sku-red", "option1_value": "Red", "image_src": "https://img/red.jpg"},
                {
                    "sku_id": "sku-gloss-black",
                    "option1_value": "Gloss Black",
                    "image_src": "https://img/gloss-black.jpg",
                },
                {"sku_id": "sku-black", "option1_value": "Black", "image_src": "https://img/black.jpg"},
            ],
        }
    monkeypatch.setattr(_FakeESClient, "_full_source", staticmethod(_full_source_with_multiple_text_matches))
    result = searcher.search(
        query="black dress",
        tenant_id="162",
        from_=0,
        size=1,
        context=context,
        enable_rerank=False,
    )
    assert len(result.results) == 1
    assert result.results[0].skus[0].sku_id == "sku-gloss-black"
    assert result.results[0].image_url == "https://img/gloss-black.jpg"
def test_searcher_skips_sku_selection_when_option_name_does_not_match_dimension_alias(monkeypatch):
    es_client = _FakeESClient(total_hits=1)
    searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
    context = create_request_context(reqid="sku-unresolved-dimension", uid="u-sku-unresolved-dimension")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en", "zh"]}),
    )
    class _UnresolvedDimensionQueryParser:
        text_encoder = None
        def parse(
            self,
            query: str,
            tenant_id: str,
            generate_vector: bool,
            context: Any,
            target_languages: Any = None,
        ):
            return _FakeParsedQuery(
                original_query=query,
                query_normalized=query,
                rewritten_query=query,
                translations={"en": "black dress"},
                style_intent_profile=_build_style_intent_profile(
                    "color", "black", "color", "colors", "颜色"
                ),
            )
    searcher.query_parser = _UnresolvedDimensionQueryParser()
    def _full_source_with_unmatched_option_name(doc_id: str) -> Dict[str, Any]:
        return {
            "spu_id": doc_id,
            "title": {"en": f"product-{doc_id}"},
            "brief": {"en": f"brief-{doc_id}"},
            "vendor": {"en": f"vendor-{doc_id}"},
            "option1_name": "Tone",
            "image_url": "https://img/default.jpg",
            "skus": [
                {"sku_id": "sku-red", "option1_value": "Red", "image_src": "https://img/red.jpg"},
                {"sku_id": "sku-black", "option1_value": "Black", "image_src": "https://img/black.jpg"},
            ],
        }
    monkeypatch.setattr(_FakeESClient, "_full_source", staticmethod(_full_source_with_unmatched_option_name))
    result = searcher.search(
        query="黑色 连衣裙",
        tenant_id="162",
        from_=0,
        size=1,
        context=context,
        enable_rerank=False,
    )
    assert len(result.results) == 1
    assert result.results[0].skus[0].sku_id == "sku-red"
    assert result.results[0].image_url == "https://img/default.jpg"
def test_searcher_promotes_sku_by_embedding_when_query_has_no_direct_option_match(monkeypatch):
    es_client = _FakeESClient(total_hits=1)
    searcher = _build_searcher(_build_search_config(rerank_enabled=False), es_client)
    context = create_request_context(reqid="sku-embed", uid="u-sku-embed")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en"]}),
    )
    encoder = _FakeTextEncoder(
        {
            "linen summer dress": [0.8, 0.2],
            "red": [1.0, 0.0],
            "blue": [0.0, 1.0],
        }
    )
    class _EmbeddingQueryParser:
        text_encoder = encoder
        def parse(
            self,
            query: str,
            tenant_id: str,
            generate_vector: bool,
            context: Any,
            target_languages: Any = None,
        ):
            return _FakeParsedQuery(
                original_query=query,
                query_normalized=query,
                rewritten_query=query,
                translations={},
                query_vector=np.array([0.0, 1.0], dtype=np.float32),
                style_intent_profile=_build_style_intent_profile(
                    "color", "blue", "color", "colors", "颜色"
                ),
            )
    searcher.query_parser = _EmbeddingQueryParser()
    def _full_source_with_skus(doc_id: str) -> Dict[str, Any]:
        return {
            "spu_id": doc_id,
            "title": {"en": f"product-{doc_id}"},
            "brief": {"en": f"brief-{doc_id}"},
            "vendor": {"en": f"vendor-{doc_id}"},
            "option1_name": "Color",
            "image_url": "https://img/default.jpg",
            "skus": [
                {"sku_id": "sku-red", "option1_value": "Red", "image_src": "https://img/red.jpg"},
                {"sku_id": "sku-blue", "option1_value": "Blue", "image_src": "https://img/blue.jpg"},
            ],
        }
    monkeypatch.setattr(_FakeESClient, "_full_source", staticmethod(_full_source_with_skus))
    result = searcher.search(
        query="linen summer dress",
        tenant_id="162",
        from_=0,
        size=1,
        context=context,
        enable_rerank=False,
    )
    assert len(result.results) == 1
    assert result.results[0].skus[0].sku_id == "sku-blue"
    assert result.results[0].image_url == "https://img/blue.jpg"
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)
    context = create_request_context(reqid="dbg", uid="u-dbg")
    monkeypatch.setattr(
        "search.searcher.get_tenant_config_loader",
        lambda: SimpleNamespace(get_tenant_config=lambda tenant_id: {"index_languages": ["en", "zh"]}),
    )
    result = searcher.search(
        query="toy",
        tenant_id="162",
        from_=0,
        size=2,
        context=context,
        enable_rerank=False,
        debug=True,
    )
    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
    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
    assert result.debug_info["per_result"][0]["es_score_normalized"] == 1.0
    assert result.debug_info["per_result"][1]["es_score_normalized"] == 2.0 / 3.0