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
|