Blame view

search/searcher.py 31.2 KB
be52af70   tangwang   first commit
1
2
3
  """
  Main Searcher module - executes search queries against Elasticsearch.
  
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
4
  Handles query parsing, ranking, and result formatting.
be52af70   tangwang   first commit
5
6
  """
  
6aa246be   tangwang   问题:Pydantic 应该能自动...
7
  from typing import Dict, Any, List, Optional, Union
d1d356f8   tangwang   脚本优化
8
  import os
99bea633   tangwang   add logs
9
  import time, json
325eec03   tangwang   1. 日志、配置基础设施,使用优化
10
  import logging
28e57bb1   tangwang   日志体系优化
11
  import hashlib
be52af70   tangwang   first commit
12
  
be52af70   tangwang   first commit
13
14
  from utils.es_client import ESClient
  from query import QueryParser, ParsedQuery
07cf5a93   tangwang   START_EMBEDDING=...
15
  from embeddings.image_encoder import CLIPImageEncoder
be52af70   tangwang   first commit
16
  from .es_query_builder import ESQueryBuilder
9f96d6f3   tangwang   短query不用语义搜索
17
  from config import SearchConfig
345d960b   tangwang   1. 删除全局 enable_tr...
18
  from config.tenant_config_loader import get_tenant_config_loader
ed948666   tangwang   tidy
19
  from context.request_context import RequestContext, RequestContextStage
13320ac6   tangwang   分面接口修改:
20
  from api.models import FacetResult, FacetValue, FacetConfig
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
21
  from api.result_formatter import ResultFormatter
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
22
  from indexer.mapping_generator import get_tenant_index_name
be52af70   tangwang   first commit
23
  
325eec03   tangwang   1. 日志、配置基础设施,使用优化
24
  logger = logging.getLogger(__name__)
28e57bb1   tangwang   日志体系优化
25
26
27
28
29
30
31
32
33
  backend_verbose_logger = logging.getLogger("backend.verbose")
  
  
  def _log_backend_verbose(payload: Dict[str, Any]) -> None:
      if not backend_verbose_logger.handlers:
          return
      backend_verbose_logger.info(
          json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
      )
325eec03   tangwang   1. 日志、配置基础设施,使用优化
34
  
be52af70   tangwang   first commit
35
36
  
  class SearchResult:
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
37
      """Container for search results (外部友好格式)."""
be52af70   tangwang   first commit
38
39
40
  
      def __init__(
          self,
cadc77b6   tangwang   索引字段名、变量名、API数据结构...
41
          results: List[Any],  # List[SpuResult]
be52af70   tangwang   first commit
42
43
44
          total: int,
          max_score: float,
          took_ms: int,
6aa246be   tangwang   问题:Pydantic 应该能自动...
45
          facets: Optional[List[FacetResult]] = None,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
46
          query_info: Optional[Dict[str, Any]] = None,
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
47
48
          suggestions: Optional[List[str]] = None,
          related_searches: Optional[List[str]] = None,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
49
          debug_info: Optional[Dict[str, Any]] = None
be52af70   tangwang   first commit
50
      ):
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
51
          self.results = results
be52af70   tangwang   first commit
52
53
54
          self.total = total
          self.max_score = max_score
          self.took_ms = took_ms
6aa246be   tangwang   问题:Pydantic 应该能自动...
55
          self.facets = facets
be52af70   tangwang   first commit
56
          self.query_info = query_info or {}
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
57
58
          self.suggestions = suggestions or []
          self.related_searches = related_searches or []
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
59
          self.debug_info = debug_info
43f1139f   tangwang   refactor: ES查询结构重...
60
  
be52af70   tangwang   first commit
61
62
      def to_dict(self) -> Dict[str, Any]:
          """Convert to dictionary representation."""
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
63
          result = {
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
64
              "results": [r.model_dump() if hasattr(r, 'model_dump') else r for r in self.results],
be52af70   tangwang   first commit
65
66
67
              "total": self.total,
              "max_score": self.max_score,
              "took_ms": self.took_ms,
6aa246be   tangwang   问题:Pydantic 应该能自动...
68
              "facets": [f.model_dump() for f in self.facets] if self.facets else None,
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
69
70
71
              "query_info": self.query_info,
              "suggestions": self.suggestions,
              "related_searches": self.related_searches
be52af70   tangwang   first commit
72
          }
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
73
74
75
          if self.debug_info is not None:
              result["debug_info"] = self.debug_info
          return result
be52af70   tangwang   first commit
76
77
78
79
80
81
82
83
  
  
  class Searcher:
      """
      Main search engine class.
  
      Handles:
      - Query parsing and translation
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
84
      - Dynamic multi-language text recall planning
be52af70   tangwang   first commit
85
86
87
88
89
90
      - ES query building
      - Result ranking and formatting
      """
  
      def __init__(
          self,
be52af70   tangwang   first commit
91
          es_client: ESClient,
9f96d6f3   tangwang   短query不用语义搜索
92
          config: SearchConfig,
26b910bd   tangwang   refactor service ...
93
94
          query_parser: Optional[QueryParser] = None,
          image_encoder: Optional[CLIPImageEncoder] = None,
be52af70   tangwang   first commit
95
96
97
98
99
      ):
          """
          Initialize searcher.
  
          Args:
be52af70   tangwang   first commit
100
              es_client: Elasticsearch client
9f96d6f3   tangwang   短query不用语义搜索
101
              config: SearchConfig instance
be52af70   tangwang   first commit
102
              query_parser: Query parser (created if not provided)
26b910bd   tangwang   refactor service ...
103
              image_encoder: Optional pre-initialized image encoder
be52af70   tangwang   first commit
104
          """
be52af70   tangwang   first commit
105
          self.es_client = es_client
9f96d6f3   tangwang   短query不用语义搜索
106
          self.config = config
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
107
          # Index name is now generated dynamically per tenant, no longer stored here
9f96d6f3   tangwang   短query不用语义搜索
108
          self.query_parser = query_parser or QueryParser(config)
9f96d6f3   tangwang   短query不用语义搜索
109
          self.text_embedding_field = config.query_config.text_embedding_field or "title_embedding"
26b910bd   tangwang   refactor service ...
110
111
112
113
114
115
          self.image_embedding_field = config.query_config.image_embedding_field
          if self.image_embedding_field and image_encoder is None:
              self.image_encoder = CLIPImageEncoder()
          else:
              self.image_encoder = image_encoder
          self.source_fields = config.query_config.source_fields
be52af70   tangwang   first commit
116
  
f0d020c3   tangwang   多语言查询改为只支持中英文两种,f...
117
          # Query builder - simplified single-layer architecture
f0d020c3   tangwang   多语言查询改为只支持中英文两种,f...
118
          self.query_builder = ESQueryBuilder(
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
119
120
121
122
123
              match_fields=[],
              field_boosts=self.config.field_boosts,
              multilingual_fields=self.config.query_config.multilingual_fields,
              shared_fields=self.config.query_config.shared_fields,
              core_multilingual_fields=self.config.query_config.core_multilingual_fields,
be52af70   tangwang   first commit
124
              text_embedding_field=self.text_embedding_field,
13377199   tangwang   接口优化
125
              image_embedding_field=self.image_embedding_field,
9f96d6f3   tangwang   短query不用语义搜索
126
              source_fields=self.source_fields,
7bc756c5   tangwang   优化 ES 查询构建
127
              function_score_config=self.config.function_score,
70dab99f   tangwang   add logs
128
              default_language=self.config.query_config.default_language,
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
129
130
131
132
133
134
              knn_boost=self.config.query_config.knn_boost,
              base_minimum_should_match=self.config.query_config.base_minimum_should_match,
              translation_minimum_should_match=self.config.query_config.translation_minimum_should_match,
              translation_boost=self.config.query_config.translation_boost,
              translation_boost_when_source_missing=self.config.query_config.translation_boost_when_source_missing,
              source_boost_when_missing=self.config.query_config.source_boost_when_missing,
bcada818   tangwang   last
135
136
137
              original_query_fallback_boost_when_translation_missing=(
                  self.config.query_config.original_query_fallback_boost_when_translation_missing
              ),
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
138
139
140
141
              keywords_boost=self.config.query_config.keywords_boost,
              enable_phrase_query=self.config.query_config.enable_phrase_query,
              tie_breaker_base_query=self.config.query_config.tie_breaker_base_query,
              tie_breaker_keywords=self.config.query_config.tie_breaker_keywords,
be52af70   tangwang   first commit
142
143
          )
  
26b910bd   tangwang   refactor service ...
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
      def _apply_source_filter(self, es_query: Dict[str, Any]) -> None:
          """
          Apply tri-state _source semantics:
          - None: do not set _source (return full source)
          - []: _source=false (return no source fields)
          - [..]: _source.includes=[..]
          """
          if self.source_fields is None:
              return
          if not isinstance(self.source_fields, list):
              raise ValueError("query_config.source_fields must be null or list[str]")
          if len(self.source_fields) == 0:
              es_query["_source"] = False
              return
          es_query["_source"] = {"includes": self.source_fields}
  
be52af70   tangwang   first commit
160
161
162
      def search(
          self,
          query: str,
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
163
          tenant_id: str,
be52af70   tangwang   first commit
164
165
166
          size: int = 10,
          from_: int = 0,
          filters: Optional[Dict[str, Any]] = None,
6aa246be   tangwang   问题:Pydantic 应该能自动...
167
          range_filters: Optional[Dict[str, Any]] = None,
13320ac6   tangwang   分面接口修改:
168
          facets: Optional[List[FacetConfig]] = None,
16c42787   tangwang   feat: implement r...
169
          min_score: Optional[float] = None,
c86c8237   tangwang   支持聚合。过滤项补充了逻辑,但是有问题
170
          context: Optional[RequestContext] = None,
c86c8237   tangwang   支持聚合。过滤项补充了逻辑,但是有问题
171
          sort_by: Optional[str] = None,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
172
          sort_order: Optional[str] = "desc",
577ec972   tangwang   返回给前端的字段、格式适配。主要包...
173
          debug: bool = False,
2739b281   tangwang   多语言索引调整
174
          language: str = "en",
a3a5d41b   tangwang   (sku_filter_dimen...
175
          sku_filter_dimension: Optional[List[str]] = None,
ff32d894   tangwang   rerank
176
177
178
          enable_rerank: bool = False,
          rerank_query_template: Optional[str] = None,
          rerank_doc_template: Optional[str] = None,
be52af70   tangwang   first commit
179
180
      ) -> SearchResult:
          """
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
181
          Execute search query (外部友好格式).
be52af70   tangwang   first commit
182
183
184
  
          Args:
              query: Search query string
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
185
              tenant_id: Tenant ID (required for filtering)
be52af70   tangwang   first commit
186
187
              size: Number of results to return
              from_: Offset for pagination
6aa246be   tangwang   问题:Pydantic 应该能自动...
188
189
190
              filters: Exact match filters
              range_filters: Range filters for numeric fields
              facets: Facet configurations for faceted search
be52af70   tangwang   first commit
191
              min_score: Minimum score threshold
16c42787   tangwang   feat: implement r...
192
              context: Request context for tracking (created if not provided)
c86c8237   tangwang   支持聚合。过滤项补充了逻辑,但是有问题
193
194
              sort_by: Field name for sorting
              sort_order: Sort order: 'asc' or 'desc'
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
195
              debug: Enable debug information output
be52af70   tangwang   first commit
196
197
  
          Returns:
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
198
              SearchResult object with formatted results
be52af70   tangwang   first commit
199
          """
16c42787   tangwang   feat: implement r...
200
          if context is None:
ed948666   tangwang   tidy
201
              raise ValueError("context is required")
16c42787   tangwang   feat: implement r...
202
  
345d960b   tangwang   1. 删除全局 enable_tr...
203
204
205
          # 根据租户配置决定翻译开关(离线/在线统一)
          tenant_loader = get_tenant_config_loader()
          tenant_cfg = tenant_loader.get_tenant_config(tenant_id)
038e4e2f   tangwang   refactor(i18n): t...
206
207
          index_langs = tenant_cfg.get("index_languages") or []
          enable_translation = len(index_langs) > 0
9f96d6f3   tangwang   短query不用语义搜索
208
          enable_embedding = self.config.query_config.enable_text_embedding
ff32d894   tangwang   rerank
209
210
          # 重排仅由请求参数 enable_rerank 控制,唯一实现为调用外部 BGE 重排服务
          do_rerank = bool(enable_rerank)
506c39b7   tangwang   feat(search): 统一重...
211
212
          rerank_window = self.config.rerank.rerank_window or 1000
          # 若开启重排且请求范围在窗口内:从 ES 取前 rerank_window 条、重排后再按 from/size 分页;否则不重排,按原 from/size 查 ES
ff32d894   tangwang   rerank
213
          in_rerank_window = do_rerank and (from_ + size) <= rerank_window
506c39b7   tangwang   feat(search): 统一重...
214
215
          es_fetch_from = 0 if in_rerank_window else from_
          es_fetch_size = rerank_window if in_rerank_window else size
16c42787   tangwang   feat: implement r...
216
217
218
219
220
221
  
          # Start timing
          context.start_stage(RequestContextStage.TOTAL)
  
          context.logger.info(
              f"开始搜索请求 | 查询: '{query}' | 参数: size={size}, from_={from_}, "
ff32d894   tangwang   rerank
222
              f"enable_rerank={do_rerank}, in_rerank_window={in_rerank_window}, es_fetch=({es_fetch_from},{es_fetch_size}) | "
506c39b7   tangwang   feat(search): 统一重...
223
              f"enable_translation={enable_translation}, enable_embedding={enable_embedding}, min_score={min_score}",
16c42787   tangwang   feat: implement r...
224
225
226
227
228
229
230
              extra={'reqid': context.reqid, 'uid': context.uid}
          )
  
          # Store search parameters in context
          context.metadata['search_params'] = {
              'size': size,
              'from_': from_,
506c39b7   tangwang   feat(search): 统一重...
231
232
233
              'es_fetch_from': es_fetch_from,
              'es_fetch_size': es_fetch_size,
              'in_rerank_window': in_rerank_window,
ff32d894   tangwang   rerank
234
235
              'rerank_query_template': rerank_query_template,
              'rerank_doc_template': rerank_doc_template,
16c42787   tangwang   feat: implement r...
236
              'filters': filters,
6aa246be   tangwang   问题:Pydantic 应该能自动...
237
238
              'range_filters': range_filters,
              'facets': facets,
16c42787   tangwang   feat: implement r...
239
240
              'enable_translation': enable_translation,
              'enable_embedding': enable_embedding,
ff32d894   tangwang   rerank
241
              'enable_rerank': do_rerank,
c86c8237   tangwang   支持聚合。过滤项补充了逻辑,但是有问题
242
              'min_score': min_score,
c86c8237   tangwang   支持聚合。过滤项补充了逻辑,但是有问题
243
244
              'sort_by': sort_by,
              'sort_order': sort_order
16c42787   tangwang   feat: implement r...
245
          }
be52af70   tangwang   first commit
246
  
16c42787   tangwang   feat: implement r...
247
248
249
          context.metadata['feature_flags'] = {
              'translation_enabled': enable_translation,
              'embedding_enabled': enable_embedding,
ff32d894   tangwang   rerank
250
              'rerank_enabled': do_rerank
16c42787   tangwang   feat: implement r...
251
          }
be52af70   tangwang   first commit
252
253
  
          # Step 1: Parse query
16c42787   tangwang   feat: implement r...
254
255
256
257
          context.start_stage(RequestContextStage.QUERY_PARSING)
          try:
              parsed_query = self.query_parser.parse(
                  query,
345d960b   tangwang   1. 删除全局 enable_tr...
258
                  tenant_id=tenant_id,
16c42787   tangwang   feat: implement r...
259
260
261
262
263
264
                  generate_vector=enable_embedding,
                  context=context
              )
              # Store query analysis results in context
              context.store_query_analysis(
                  original_query=parsed_query.original_query,
3a5fda00   tangwang   1. ES字段 skus的 ima...
265
                  query_normalized=parsed_query.query_normalized,
16c42787   tangwang   feat: implement r...
266
267
268
269
270
                  rewritten_query=parsed_query.rewritten_query,
                  detected_language=parsed_query.detected_language,
                  translations=parsed_query.translations,
                  query_vector=parsed_query.query_vector.tolist() if parsed_query.query_vector is not None else None,
                  domain=parsed_query.domain,
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
271
                  is_simple_query=True
16c42787   tangwang   feat: implement r...
272
              )
be52af70   tangwang   first commit
273
  
16c42787   tangwang   feat: implement r...
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
              context.logger.info(
                  f"查询解析完成 | 原查询: '{parsed_query.original_query}' | "
                  f"重写后: '{parsed_query.rewritten_query}' | "
                  f"语言: {parsed_query.detected_language} | "
                  f"域: {parsed_query.domain} | "
                  f"向量: {'是' if parsed_query.query_vector is not None else '否'}",
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
          except Exception as e:
              context.set_error(e)
              context.logger.error(
                  f"查询解析失败 | 错误: {str(e)}",
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
              raise
          finally:
              context.end_stage(RequestContextStage.QUERY_PARSING)
  
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
292
          # Step 2: Query building
16c42787   tangwang   feat: implement r...
293
294
          context.start_stage(RequestContextStage.QUERY_BUILDING)
          try:
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
295
296
              # Generate tenant-specific index name
              index_name = get_tenant_index_name(tenant_id)
2739b281   tangwang   多语言索引调整
297
              # index_name = "search_products"
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
298
299
              
              # No longer need to add tenant_id to filters since each tenant has its own index
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
300
  
f0d020c3   tangwang   多语言查询改为只支持中英文两种,f...
301
              es_query = self.query_builder.build_query(
3a5fda00   tangwang   1. ES字段 skus的 ima...
302
                  query_text=parsed_query.rewritten_query or parsed_query.query_normalized,
16c42787   tangwang   feat: implement r...
303
                  query_vector=parsed_query.query_vector if enable_embedding else None,
16c42787   tangwang   feat: implement r...
304
                  filters=filters,
6aa246be   tangwang   问题:Pydantic 应该能自动...
305
                  range_filters=range_filters,
c581becd   tangwang   feat: 实现 Multi-Se...
306
                  facet_configs=facets,
506c39b7   tangwang   feat(search): 统一重...
307
308
                  size=es_fetch_size,
                  from_=es_fetch_from,
16c42787   tangwang   feat: implement r...
309
                  enable_knn=enable_embedding and parsed_query.query_vector is not None,
7bc756c5   tangwang   优化 ES 查询构建
310
311
                  min_score=min_score,
                  parsed_query=parsed_query
16c42787   tangwang   feat: implement r...
312
              )
be52af70   tangwang   first commit
313
  
6aa246be   tangwang   问题:Pydantic 应该能自动...
314
315
316
317
318
319
320
              # Add facets for faceted search
              if facets:
                  facet_aggs = self.query_builder.build_facets(facets)
                  if facet_aggs:
                      if "aggs" not in es_query:
                          es_query["aggs"] = {}
                      es_query["aggs"].update(facet_aggs)
16c42787   tangwang   feat: implement r...
321
  
c86c8237   tangwang   支持聚合。过滤项补充了逻辑,但是有问题
322
323
324
325
              # Add sorting if specified
              if sort_by:
                  es_query = self.query_builder.add_sorting(es_query, sort_by, sort_order)
  
16c42787   tangwang   feat: implement r...
326
327
328
329
330
331
332
              # Extract size and from from body for ES client parameters
              body_for_es = {k: v for k, v in es_query.items() if k not in ['size', 'from']}
  
              # Store ES query in context
              context.store_intermediate_result('es_query', es_query)
              context.store_intermediate_result('es_body_for_search', body_for_es)
  
28e57bb1   tangwang   日志体系优化
333
334
335
336
337
              # Serialize ES query to compute a compact size + stable digest for correlation
              es_query_compact = json.dumps(es_query, 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)
              vector_dims = int(len(parsed_query.query_vector)) if parsed_query.query_vector is not None else 0
99bea633   tangwang   add logs
338
  
16c42787   tangwang   feat: implement r...
339
              context.logger.info(
28e57bb1   tangwang   日志体系优化
340
341
342
343
344
345
                  "ES query built | size: %s chars | digest: %s | KNN: %s | vector_dims: %s | facets: %s",
                  len(es_query_compact),
                  es_query_digest,
                  "yes" if knn_enabled else "no",
                  vector_dims,
                  "yes" if facets else "no",
16c42787   tangwang   feat: implement r...
346
347
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
28e57bb1   tangwang   日志体系优化
348
349
350
351
352
353
354
355
356
357
358
359
              _log_backend_verbose({
                  "event": "es_query_built",
                  "reqid": context.reqid,
                  "uid": context.uid,
                  "tenant_id": tenant_id,
                  "size_chars": len(es_query_compact),
                  "sha256_16": es_query_digest,
                  "knn_enabled": knn_enabled,
                  "vector_dims": vector_dims,
                  "has_facets": bool(facets),
                  "query": es_query,
              })
16c42787   tangwang   feat: implement r...
360
361
362
363
364
365
366
367
368
369
370
371
372
          except Exception as e:
              context.set_error(e)
              context.logger.error(
                  f"ES查询构建失败 | 错误: {str(e)}",
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
              raise
          finally:
              context.end_stage(RequestContextStage.QUERY_BUILDING)
  
          # Step 4: Elasticsearch search
          context.start_stage(RequestContextStage.ELASTICSEARCH_SEARCH)
          try:
506c39b7   tangwang   feat(search): 统一重...
373
              # Use tenant-specific index name(开启重排且在窗口内时已用 es_fetch_size/es_fetch_from)
16c42787   tangwang   feat: implement r...
374
              es_response = self.es_client.search(
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
375
                  index_name=index_name,
16c42787   tangwang   feat: implement r...
376
                  body=body_for_es,
506c39b7   tangwang   feat(search): 统一重...
377
378
                  size=es_fetch_size,
                  from_=es_fetch_from
be52af70   tangwang   first commit
379
380
              )
  
16c42787   tangwang   feat: implement r...
381
382
              # Store ES response in context
              context.store_intermediate_result('es_response', es_response)
be52af70   tangwang   first commit
383
  
16c42787   tangwang   feat: implement r...
384
385
386
387
388
              # Extract timing from ES response
              es_took = es_response.get('took', 0)
              context.logger.info(
                  f"ES搜索完成 | 耗时: {es_took}ms | "
                  f"命中数: {es_response.get('hits', {}).get('total', {}).get('value', 0)} | "
25d3e81d   tangwang   fix指定sort项时候的bug
389
                  f"最高分: {(es_response.get('hits', {}).get('max_score') or 0):.3f}",
16c42787   tangwang   feat: implement r...
390
391
392
393
394
395
396
397
398
399
400
401
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
          except Exception as e:
              context.set_error(e)
              context.logger.error(
                  f"ES搜索执行失败 | 错误: {str(e)}",
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
              raise
          finally:
              context.end_stage(RequestContextStage.ELASTICSEARCH_SEARCH)
  
506c39b7   tangwang   feat(search): 统一重...
402
          # Optional Step 4.5: AI reranking(仅当请求范围在重排窗口内时执行)
ff32d894   tangwang   rerank
403
          if do_rerank and in_rerank_window:
506c39b7   tangwang   feat(search): 统一重...
404
405
406
407
408
409
              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
ff32d894   tangwang   rerank
410
411
                  effective_query_template = rerank_query_template or rc.rerank_query_template
                  effective_doc_template = rerank_doc_template or rc.rerank_doc_template
506c39b7   tangwang   feat(search): 统一重...
412
413
414
415
                  es_response, rerank_meta, fused_debug = run_rerank(
                      query=rerank_query,
                      es_response=es_response,
                      language=language,
506c39b7   tangwang   feat(search): 统一重...
416
417
418
                      timeout_sec=rc.timeout_sec,
                      weight_es=rc.weight_es,
                      weight_ai=rc.weight_ai,
ff32d894   tangwang   rerank
419
420
                      rerank_query_template=effective_query_template,
                      rerank_doc_template=effective_doc_template,
506c39b7   tangwang   feat(search): 统一重...
421
422
423
                  )
  
                  if rerank_meta is not None:
42e3aea6   tangwang   tidy
424
425
                      from config.services_config import get_rerank_service_url
                      rerank_url = get_rerank_service_url()
506c39b7   tangwang   feat(search): 统一重...
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
                      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}
              )
  
16c42787   tangwang   feat: implement r...
465
466
467
          # Step 5: Result processing
          context.start_stage(RequestContextStage.RESULT_PROCESSING)
          try:
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
468
469
              # Extract ES hits
              es_hits = []
16c42787   tangwang   feat: implement r...
470
              if 'hits' in es_response and 'hits' in es_response['hits']:
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
471
                  es_hits = es_response['hits']['hits']
be52af70   tangwang   first commit
472
  
16c42787   tangwang   feat: implement r...
473
474
475
476
477
478
              # Extract total and max_score
              total = es_response.get('hits', {}).get('total', {})
              if isinstance(total, dict):
                  total_value = total.get('value', 0)
              else:
                  total_value = total
506c39b7   tangwang   feat(search): 统一重...
479
              # max_score 会在启用 AI 搜索时被更新为融合分数的最大值
25d3e81d   tangwang   fix指定sort项时候的bug
480
              max_score = es_response.get('hits', {}).get('max_score') or 0.0
be52af70   tangwang   first commit
481
  
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
482
              # Format results using ResultFormatter
577ec972   tangwang   返回给前端的字段、格式适配。主要包...
483
484
485
              formatted_results = ResultFormatter.format_search_results(
                  es_hits,
                  max_score,
ca91352a   tangwang   更新文档
486
487
                  language=language,
                  sku_filter_dimension=sku_filter_dimension
577ec972   tangwang   返回给前端的字段、格式适配。主要包...
488
              )
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
489
490
491
492
493
494
  
              # Format facets
              standardized_facets = None
              if facets:
                  standardized_facets = ResultFormatter.format_facets(
                      es_response.get('aggregations', {}),
c581becd   tangwang   feat: 实现 Multi-Se...
495
496
                      facets,
                      filters
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
497
498
499
500
501
502
                  )
  
              # Generate suggestions and related searches
              query_text = parsed_query.original_query if parsed_query else query
              suggestions = ResultFormatter.generate_suggestions(query_text, formatted_results)
              related_searches = ResultFormatter.generate_related_searches(query_text, formatted_results)
be52af70   tangwang   first commit
503
  
16c42787   tangwang   feat: implement r...
504
              context.logger.info(
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
505
                  f"结果处理完成 | 返回: {len(formatted_results)}条 | 总计: {total_value}条",
16c42787   tangwang   feat: implement r...
506
507
508
509
510
511
512
513
514
515
516
517
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
  
          except Exception as e:
              context.set_error(e)
              context.logger.error(
                  f"结果处理失败 | 错误: {str(e)}",
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
              raise
          finally:
              context.end_stage(RequestContextStage.RESULT_PROCESSING)
be52af70   tangwang   first commit
518
  
16c42787   tangwang   feat: implement r...
519
520
521
          # End total timing and build result
          total_duration = context.end_stage(RequestContextStage.TOTAL)
          context.performance_metrics.total_duration = total_duration
be52af70   tangwang   first commit
522
  
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
523
524
525
526
527
528
          # Collect debug information if requested
          debug_info = None
          if debug:
              debug_info = {
                  "query_analysis": {
                      "original_query": context.query_analysis.original_query,
3a5fda00   tangwang   1. ES字段 skus的 ima...
529
                      "query_normalized": context.query_analysis.query_normalized,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
530
531
532
533
534
                      "rewritten_query": context.query_analysis.rewritten_query,
                      "detected_language": context.query_analysis.detected_language,
                      "translations": context.query_analysis.translations,
                      "has_vector": context.query_analysis.query_vector is not None,
                      "is_simple_query": context.query_analysis.is_simple_query,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
                      "domain": context.query_analysis.domain
                  },
                  "es_query": context.get_intermediate_result('es_query', {}),
                  "es_response": {
                      "took_ms": es_response.get('took', 0),
                      "total_hits": total_value,
                      "max_score": max_score,
                      "shards": es_response.get('_shards', {})
                  },
                  "feature_flags": context.metadata.get('feature_flags', {}),
                  "stage_timings": {
                      k: round(v, 2) for k, v in context.performance_metrics.stage_timings.items()
                  },
                  "search_params": context.metadata.get('search_params', {})
              }
  
be52af70   tangwang   first commit
551
552
          # Build result
          result = SearchResult(
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
553
              results=formatted_results,
be52af70   tangwang   first commit
554
555
              total=total_value,
              max_score=max_score,
16c42787   tangwang   feat: implement r...
556
              took_ms=int(total_duration),
6aa246be   tangwang   问题:Pydantic 应该能自动...
557
              facets=standardized_facets,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
558
              query_info=parsed_query.to_dict(),
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
559
560
              suggestions=suggestions,
              related_searches=related_searches,
1f071951   tangwang   补充调试信息,记录包括各个阶段的 ...
561
              debug_info=debug_info
be52af70   tangwang   first commit
562
563
          )
  
16c42787   tangwang   feat: implement r...
564
565
          # Log complete performance summary
          context.log_performance_summary()
be52af70   tangwang   first commit
566
567
568
569
570
571
  
          return result
  
      def search_by_image(
          self,
          image_url: str,
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
572
          tenant_id: str,
be52af70   tangwang   first commit
573
          size: int = 10,
6aa246be   tangwang   问题:Pydantic 应该能自动...
574
575
          filters: Optional[Dict[str, Any]] = None,
          range_filters: Optional[Dict[str, Any]] = None
be52af70   tangwang   first commit
576
577
      ) -> SearchResult:
          """
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
578
          Search by image similarity (外部友好格式).
be52af70   tangwang   first commit
579
580
581
  
          Args:
              image_url: URL of query image
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
582
              tenant_id: Tenant ID (required for filtering)
be52af70   tangwang   first commit
583
              size: Number of results
6aa246be   tangwang   问题:Pydantic 应该能自动...
584
585
              filters: Exact match filters
              range_filters: Range filters for numeric fields
be52af70   tangwang   first commit
586
587
  
          Returns:
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
588
              SearchResult object with formatted results
be52af70   tangwang   first commit
589
590
591
592
593
          """
          if not self.image_embedding_field:
              raise ValueError("Image embedding field not configured")
  
          # Generate image embedding
26b910bd   tangwang   refactor service ...
594
595
596
          if self.image_encoder is None:
              raise RuntimeError("Image encoder is not initialized at startup")
          image_vector = self.image_encoder.encode_image_from_url(image_url)
be52af70   tangwang   first commit
597
598
599
600
  
          if image_vector is None:
              raise ValueError(f"Failed to encode image: {image_url}")
  
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
601
602
603
604
          # Generate tenant-specific index name
          index_name = get_tenant_index_name(tenant_id)
          
          # No longer need to add tenant_id to filters since each tenant has its own index
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
605
  
be52af70   tangwang   first commit
606
607
608
609
610
611
612
613
614
615
616
          # Build KNN query
          es_query = {
              "size": size,
              "knn": {
                  "field": self.image_embedding_field,
                  "query_vector": image_vector.tolist(),
                  "k": size,
                  "num_candidates": size * 10
              }
          }
  
26b910bd   tangwang   refactor service ...
617
618
          # Apply source filtering semantics (None / [] / list)
          self._apply_source_filter(es_query)
13377199   tangwang   接口优化
619
  
6aa246be   tangwang   问题:Pydantic 应该能自动...
620
621
622
          if filters or range_filters:
              filter_clauses = self.query_builder._build_filters(filters, range_filters)
              if filter_clauses:
7fbca0d7   tangwang   启动脚本优化
623
624
625
626
627
628
629
                  if len(filter_clauses) == 1:
                      es_query["knn"]["filter"] = filter_clauses[0]
                  else:
                      es_query["knn"]["filter"] = {
                          "bool": {
                              "filter": filter_clauses
                          }
6aa246be   tangwang   问题:Pydantic 应该能自动...
630
                      }
be52af70   tangwang   first commit
631
632
633
  
          # Execute search
          es_response = self.es_client.search(
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
634
              index_name=index_name,
be52af70   tangwang   first commit
635
636
637
638
              body=es_query,
              size=size
          )
  
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
639
640
          # Extract ES hits
          es_hits = []
be52af70   tangwang   first commit
641
          if 'hits' in es_response and 'hits' in es_response['hits']:
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
642
              es_hits = es_response['hits']['hits']
be52af70   tangwang   first commit
643
  
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
644
          # Extract total and max_score
be52af70   tangwang   first commit
645
646
647
648
649
650
          total = es_response.get('hits', {}).get('total', {})
          if isinstance(total, dict):
              total_value = total.get('value', 0)
          else:
              total_value = total
  
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
651
652
653
          max_score = es_response.get('hits', {}).get('max_score') or 0.0
  
          # Format results using ResultFormatter
ca91352a   tangwang   更新文档
654
655
656
          formatted_results = ResultFormatter.format_search_results(
              es_hits, 
              max_score,
2739b281   tangwang   多语言索引调整
657
              language="en",  # Default language for image search
ca91352a   tangwang   更新文档
658
659
              sku_filter_dimension=None  # Image search doesn't support SKU filtering
          )
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
660
  
be52af70   tangwang   first commit
661
          return SearchResult(
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
662
              results=formatted_results,
be52af70   tangwang   first commit
663
              total=total_value,
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
664
              max_score=max_score,
be52af70   tangwang   first commit
665
              took_ms=es_response.get('took', 0),
1f6d15fa   tangwang   重构:SPU级别索引、统一索引架构...
666
667
668
669
              facets=None,
              query_info={'image_url': image_url, 'search_type': 'image_similarity'},
              suggestions=[],
              related_searches=[]
be52af70   tangwang   first commit
670
671
          )
  
b926f678   tangwang   多语言查询
672
673
      def get_domain_summary(self) -> Dict[str, Any]:
          """
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
674
          Get summary of dynamic text retrieval configuration.
b926f678   tangwang   多语言查询
675
676
  
          Returns:
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
677
              Dictionary with language-aware field information
b926f678   tangwang   多语言查询
678
          """
bd96cead   tangwang   1. 动态多语言字段与统一策略配置
679
680
681
682
683
684
685
          return {
              "mode": "dynamic_language_fields",
              "multilingual_fields": self.config.query_config.multilingual_fields,
              "shared_fields": self.config.query_config.shared_fields,
              "core_multilingual_fields": self.config.query_config.core_multilingual_fields,
              "field_boosts": self.config.field_boosts,
          }
b926f678   tangwang   多语言查询
686
  
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
687
      def get_document(self, tenant_id: str, doc_id: str) -> Optional[Dict[str, Any]]:
be52af70   tangwang   first commit
688
689
690
691
          """
          Get single document by ID.
  
          Args:
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
692
              tenant_id: Tenant ID (required to determine which index to query)
be52af70   tangwang   first commit
693
694
695
696
697
698
              doc_id: Document ID
  
          Returns:
              Document or None if not found
          """
          try:
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
699
              index_name = get_tenant_index_name(tenant_id)
be52af70   tangwang   first commit
700
              response = self.es_client.client.get(
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
701
                  index=index_name,
be52af70   tangwang   first commit
702
703
704
705
                  id=doc_id
              )
              return response.get('_source')
          except Exception as e:
e4a39cc8   tangwang   索引隔离。 不同的tenant_i...
706
              logger.error(f"Failed to get document {doc_id} from tenant {tenant_id}: {e}", exc_info=True)
be52af70   tangwang   first commit
707
              return None
6aa246be   tangwang   问题:Pydantic 应该能自动...
708
709
710
711
  
      def _standardize_facets(
          self,
          es_aggregations: Dict[str, Any],
ff5325fa   tangwang   修复:直接在 Searcher 层...
712
          facet_configs: Optional[List[Union[str, Any]]],
6aa246be   tangwang   问题:Pydantic 应该能自动...
713
714
715
716
717
718
719
          current_filters: Optional[Dict[str, Any]]
      ) -> Optional[List[FacetResult]]:
          """
           ES 聚合结果转换为标准化的分面格式(返回 Pydantic 模型)。
          
          Args:
              es_aggregations: ES 原始聚合结果
ff5325fa   tangwang   修复:直接在 Searcher 层...
720
              facet_configs: 分面配置列表(str  FacetConfig
6aa246be   tangwang   问题:Pydantic 应该能自动...
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
              current_filters: 当前应用的过滤器
          
          Returns:
              标准化的分面结果列表(FacetResult 对象)
          """
          if not es_aggregations or not facet_configs:
              return None
          
          standardized_facets: List[FacetResult] = []
          
          for config in facet_configs:
              # 解析配置
              if isinstance(config, str):
                  field = config
                  facet_type = "terms"
              else:
ff5325fa   tangwang   修复:直接在 Searcher 层...
737
738
739
                  # FacetConfig 对象
                  field = config.field
                  facet_type = config.type
6aa246be   tangwang   问题:Pydantic 应该能自动...
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
              
              agg_name = f"{field}_facet"
              
              if agg_name not in es_aggregations:
                  continue
              
              agg_result = es_aggregations[agg_name]
              
              # 获取当前字段的选中值
              selected_values = set()
              if current_filters and field in current_filters:
                  filter_value = current_filters[field]
                  if isinstance(filter_value, list):
                      selected_values = set(filter_value)
                  else:
                      selected_values = {filter_value}
              
              # 转换 buckets 为 FacetValue 对象
              facet_values: List[FacetValue] = []
              if 'buckets' in agg_result:
                  for bucket in agg_result['buckets']:
                      value = bucket.get('key')
                      count = bucket.get('doc_count', 0)
                      
                      facet_values.append(FacetValue(
                          value=value,
                          label=str(value),
                          count=count,
                          selected=value in selected_values
                      ))
              
              # 构建 FacetResult 对象
              facet_result = FacetResult(
                  field=field,
e7ad2b4a   tangwang   测试页面分页配置
774
                  label=field,
6aa246be   tangwang   问题:Pydantic 应该能自动...
775
776
777
778
779
780
781
                  type=facet_type,
                  values=facet_values
              )
              
              standardized_facets.append(facet_result)
          
          return standardized_facets if standardized_facets else None