be52af70
tangwang
first commit
|
1
2
3
4
5
6
7
8
|
"""
Query parser - main module for query processing.
Handles query rewriting, translation, and embedding generation.
"""
from typing import Dict, List, Optional, Any
import numpy as np
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
9
|
import logging
|
7bc756c5
tangwang
优化 ES 查询构建
|
10
11
|
import re
import hanlp
|
be52af70
tangwang
first commit
|
12
|
|
be52af70
tangwang
first commit
|
13
|
from embeddings import BgeEncoder
|
9f96d6f3
tangwang
短query不用语义搜索
|
14
|
from config import SearchConfig
|
be52af70
tangwang
first commit
|
15
16
17
18
|
from .language_detector import LanguageDetector
from .translator import Translator
from .query_rewriter import QueryRewriter, QueryNormalizer
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
19
20
|
logger = logging.getLogger(__name__)
|
be52af70
tangwang
first commit
|
21
22
23
24
25
26
27
28
29
30
31
32
|
class ParsedQuery:
"""Container for parsed query results."""
def __init__(
self,
original_query: str,
normalized_query: str,
rewritten_query: Optional[str] = None,
detected_language: str = "unknown",
translations: Dict[str, str] = None,
query_vector: Optional[np.ndarray] = None,
|
7bc756c5
tangwang
优化 ES 查询构建
|
33
34
35
36
37
|
domain: str = "default",
keywords: str = "",
token_count: int = 0,
is_short_query: bool = False,
is_long_query: bool = False
|
be52af70
tangwang
first commit
|
38
39
40
41
42
43
44
45
|
):
self.original_query = original_query
self.normalized_query = normalized_query
self.rewritten_query = rewritten_query or normalized_query
self.detected_language = detected_language
self.translations = translations or {}
self.query_vector = query_vector
self.domain = domain
|
7bc756c5
tangwang
优化 ES 查询构建
|
46
47
48
49
50
|
# Query analysis fields
self.keywords = keywords
self.token_count = token_count
self.is_short_query = is_short_query
self.is_long_query = is_long_query
|
be52af70
tangwang
first commit
|
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
|
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary representation."""
result = {
"original_query": self.original_query,
"normalized_query": self.normalized_query,
"rewritten_query": self.rewritten_query,
"detected_language": self.detected_language,
"translations": self.translations,
"domain": self.domain,
"has_vector": self.query_vector is not None
}
return result
class QueryParser:
"""
Main query parser that processes queries through multiple stages:
1. Normalization
2. Query rewriting (brand/category mappings, synonyms)
3. Language detection
4. Translation to target languages
5. Text embedding generation (for semantic search)
"""
def __init__(
self,
|
9f96d6f3
tangwang
短query不用语义搜索
|
78
|
config: SearchConfig,
|
be52af70
tangwang
first commit
|
79
80
81
82
83
84
85
|
text_encoder: Optional[BgeEncoder] = None,
translator: Optional[Translator] = None
):
"""
Initialize query parser.
Args:
|
9f96d6f3
tangwang
短query不用语义搜索
|
86
|
config: SearchConfig instance
|
be52af70
tangwang
first commit
|
87
88
89
|
text_encoder: Text embedding encoder (lazy loaded if not provided)
translator: Translator instance (lazy loaded if not provided)
"""
|
9f96d6f3
tangwang
短query不用语义搜索
|
90
|
self.config = config
|
be52af70
tangwang
first commit
|
91
92
93
94
95
96
|
self._text_encoder = text_encoder
self._translator = translator
# Initialize components
self.normalizer = QueryNormalizer()
self.language_detector = LanguageDetector()
|
9f96d6f3
tangwang
短query不用语义搜索
|
97
|
self.rewriter = QueryRewriter(config.query_config.rewrite_dictionary)
|
7bc756c5
tangwang
优化 ES 查询构建
|
98
99
100
101
102
103
104
|
# Initialize HanLP components at startup
logger.info("Initializing HanLP components...")
self._tok = hanlp.load(hanlp.pretrained.tok.CTB9_TOK_ELECTRA_BASE_CRF)
self._tok.config.output_spans = True
self._pos_tag = hanlp.load(hanlp.pretrained.pos.CTB9_POS_ELECTRA_SMALL)
logger.info("HanLP components initialized")
|
be52af70
tangwang
first commit
|
105
106
107
108
|
@property
def text_encoder(self) -> BgeEncoder:
"""Lazy load text encoder."""
|
9f96d6f3
tangwang
短query不用语义搜索
|
109
|
if self._text_encoder is None and self.config.query_config.enable_text_embedding:
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
110
|
logger.info("Initializing text encoder (lazy load)...")
|
be52af70
tangwang
first commit
|
111
112
113
114
115
116
|
self._text_encoder = BgeEncoder()
return self._text_encoder
@property
def translator(self) -> Translator:
"""Lazy load translator."""
|
9f96d6f3
tangwang
短query不用语义搜索
|
117
|
if self._translator is None and self.config.query_config.enable_translation:
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
118
|
logger.info("Initializing translator (lazy load)...")
|
be52af70
tangwang
first commit
|
119
|
self._translator = Translator(
|
9f96d6f3
tangwang
短query不用语义搜索
|
120
|
api_key=self.config.query_config.translation_api_key,
|
522a3964
tangwang
多语言搜索翻译的优化(deepL添...
|
121
|
use_cache=True,
|
9f96d6f3
tangwang
短query不用语义搜索
|
122
123
|
glossary_id=self.config.query_config.translation_glossary_id,
translation_context=self.config.query_config.translation_context
|
be52af70
tangwang
first commit
|
124
125
|
)
return self._translator
|
7bc756c5
tangwang
优化 ES 查询构建
|
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
def _extract_keywords(self, query: str) -> str:
"""Extract keywords (nouns with length > 1) from query."""
tok_result = self._tok(query)
if not tok_result:
return ""
words = [x[0] for x in tok_result]
pos_tags = self._pos_tag(words)
keywords = []
for word, pos in zip(words, pos_tags):
if len(word) > 1 and pos.startswith('N'):
keywords.append(word)
return " ".join(keywords)
def _get_token_count(self, query: str) -> int:
"""Get token count using HanLP."""
tok_result = self._tok(query)
return len(tok_result) if tok_result else 0
def _analyze_query_type(self, query: str, token_count: int) -> tuple:
"""Analyze query type: (is_short_query, is_long_query)."""
is_quoted = query.startswith('"') and query.endswith('"')
is_short = is_quoted or ((token_count <= 2 or len(query) <= 4) and ' ' not in query)
is_long = token_count >= 4
return is_short, is_long
|
be52af70
tangwang
first commit
|
154
|
|
16c42787
tangwang
feat: implement r...
|
155
|
def parse(self, query: str, generate_vector: bool = True, context: Optional[Any] = None) -> ParsedQuery:
|
be52af70
tangwang
first commit
|
156
157
158
159
160
161
|
"""
Parse query through all processing stages.
Args:
query: Raw query string
generate_vector: Whether to generate query embedding
|
16c42787
tangwang
feat: implement r...
|
162
|
context: Optional request context for tracking and logging
|
be52af70
tangwang
first commit
|
163
164
165
166
|
Returns:
ParsedQuery object with all processing results
"""
|
16c42787
tangwang
feat: implement r...
|
167
168
169
170
171
172
173
174
|
# Initialize logger if context provided
logger = context.logger if context else None
if logger:
logger.info(
f"开始查询解析 | 原查询: '{query}' | 生成向量: {generate_vector}",
extra={'reqid': context.reqid, 'uid': context.uid}
)
|
16c42787
tangwang
feat: implement r...
|
175
|
def log_info(msg):
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
176
177
|
if context and hasattr(context, 'logger'):
context.logger.info(msg, extra={'reqid': context.reqid, 'uid': context.uid})
|
16c42787
tangwang
feat: implement r...
|
178
|
else:
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
179
|
logger.info(msg)
|
16c42787
tangwang
feat: implement r...
|
180
181
|
def log_debug(msg):
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
182
183
|
if context and hasattr(context, 'logger'):
context.logger.debug(msg, extra={'reqid': context.reqid, 'uid': context.uid})
|
16c42787
tangwang
feat: implement r...
|
184
|
else:
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
185
|
logger.debug(msg)
|
be52af70
tangwang
first commit
|
186
187
188
|
# Stage 1: Normalize
normalized = self.normalizer.normalize(query)
|
16c42787
tangwang
feat: implement r...
|
189
190
191
|
log_debug(f"标准化完成 | '{query}' -> '{normalized}'")
if context:
context.store_intermediate_result('normalized_query', normalized)
|
be52af70
tangwang
first commit
|
192
193
194
|
# Extract domain if present (e.g., "brand:Nike" -> domain="brand", query="Nike")
domain, query_text = self.normalizer.extract_domain_query(normalized)
|
16c42787
tangwang
feat: implement r...
|
195
196
197
198
|
log_debug(f"域提取 | 域: '{domain}', 查询: '{query_text}'")
if context:
context.store_intermediate_result('extracted_domain', domain)
context.store_intermediate_result('domain_query', query_text)
|
be52af70
tangwang
first commit
|
199
200
201
|
# Stage 2: Query rewriting
rewritten = None
|
9f96d6f3
tangwang
短query不用语义搜索
|
202
|
if self.config.query_config.rewrite_dictionary: # Enable rewrite if dictionary exists
|
be52af70
tangwang
first commit
|
203
204
|
rewritten = self.rewriter.rewrite(query_text)
if rewritten != query_text:
|
16c42787
tangwang
feat: implement r...
|
205
|
log_info(f"查询重写 | '{query_text}' -> '{rewritten}'")
|
be52af70
tangwang
first commit
|
206
|
query_text = rewritten
|
16c42787
tangwang
feat: implement r...
|
207
208
209
|
if context:
context.store_intermediate_result('rewritten_query', rewritten)
context.add_warning(f"查询被重写: {query_text}")
|
be52af70
tangwang
first commit
|
210
211
212
|
# Stage 3: Language detection
detected_lang = self.language_detector.detect(query_text)
|
16c42787
tangwang
feat: implement r...
|
213
214
215
|
log_info(f"语言检测 | 检测到语言: {detected_lang}")
if context:
context.store_intermediate_result('detected_language', detected_lang)
|
be52af70
tangwang
first commit
|
216
|
|
6e0e310c
tangwang
1. Translator 类增强
|
217
|
# Stage 4: Translation (async mode - only returns cached results, missing ones translated in background)
|
be52af70
tangwang
first commit
|
218
|
translations = {}
|
9f96d6f3
tangwang
短query不用语义搜索
|
219
|
if self.config.query_config.enable_translation:
|
16c42787
tangwang
feat: implement r...
|
220
221
|
try:
# Determine target languages for translation
|
bf89b597
tangwang
feat(search): ada...
|
222
223
|
# Simplified: always translate to Chinese and English
target_langs_for_translation = ['zh', 'en']
|
16c42787
tangwang
feat: implement r...
|
224
225
226
227
228
229
230
|
target_langs = self.translator.get_translation_needs(
detected_lang,
target_langs_for_translation
)
if target_langs:
|
522a3964
tangwang
多语言搜索翻译的优化(deepL添...
|
231
|
# Use e-commerce context for better disambiguation
|
0064e946
tangwang
feat: 增量索引服务、租户配置...
|
232
233
234
235
236
|
translation_context = self.config.query_config.translation_context
# For query translation, we use a general prompt (not language-specific)
# Since translate_multi uses same prompt for all languages, we use default
query_prompt = self.config.query_config.translation_prompts.get('query_zh') or \
self.config.query_config.translation_prompts.get('default_zh')
|
6e0e310c
tangwang
1. Translator 类增强
|
237
|
# Use async mode: returns cached translations immediately, missing ones translated in background
|
16c42787
tangwang
feat: implement r...
|
238
239
240
|
translations = self.translator.translate_multi(
query_text,
target_langs,
|
522a3964
tangwang
多语言搜索翻译的优化(deepL添...
|
241
|
source_lang=detected_lang,
|
6e0e310c
tangwang
1. Translator 类增强
|
242
|
context=translation_context,
|
0064e946
tangwang
feat: 增量索引服务、租户配置...
|
243
244
|
async_mode=True,
prompt=query_prompt
|
16c42787
tangwang
feat: implement r...
|
245
|
)
|
6e0e310c
tangwang
1. Translator 类增强
|
246
247
248
249
250
251
252
253
|
# Filter out None values (missing translations that are being processed async)
translations = {k: v for k, v in translations.items() if v is not None}
if translations:
log_info(f"翻译完成(缓存命中) | 结果: {translations}")
else:
log_debug(f"翻译未命中缓存,异步翻译中...")
|
16c42787
tangwang
feat: implement r...
|
254
255
256
257
258
259
260
261
262
263
264
|
if context:
context.store_intermediate_result('translations', translations)
for lang, translation in translations.items():
if translation:
context.store_intermediate_result(f'translation_{lang}', translation)
except Exception as e:
error_msg = f"翻译失败 | 错误: {str(e)}"
log_info(error_msg)
if context:
context.add_warning(error_msg)
|
be52af70
tangwang
first commit
|
265
|
|
7bc756c5
tangwang
优化 ES 查询构建
|
266
267
268
269
270
271
272
273
274
275
276
277
278
279
|
# Stage 5: Query analysis (keywords, token count, query type)
keywords = self._extract_keywords(query_text)
token_count = self._get_token_count(query_text)
is_short_query, is_long_query = self._analyze_query_type(query_text, token_count)
log_debug(f"查询分析 | 关键词: {keywords} | token数: {token_count} | "
f"短查询: {is_short_query} | 长查询: {is_long_query}")
if context:
context.store_intermediate_result('keywords', keywords)
context.store_intermediate_result('token_count', token_count)
context.store_intermediate_result('is_short_query', is_short_query)
context.store_intermediate_result('is_long_query', is_long_query)
# Stage 6: Text embedding (only for non-short queries)
|
be52af70
tangwang
first commit
|
280
|
query_vector = None
|
7bc756c5
tangwang
优化 ES 查询构建
|
281
282
|
should_generate_embedding = (
generate_vector and
|
9f96d6f3
tangwang
短query不用语义搜索
|
283
|
self.config.query_config.enable_text_embedding and
|
7bc756c5
tangwang
优化 ES 查询构建
|
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
|
domain == "default" and
not is_short_query
)
if should_generate_embedding:
try:
log_debug("开始生成查询向量")
query_vector = self.text_encoder.encode([query_text])[0]
log_debug(f"查询向量生成完成 | 形状: {query_vector.shape}")
if context:
context.store_intermediate_result('query_vector_shape', query_vector.shape)
except Exception as e:
error_msg = f"查询向量生成失败 | 错误: {str(e)}"
log_info(error_msg)
if context:
context.add_warning(error_msg)
|
be52af70
tangwang
first commit
|
300
301
302
303
304
305
306
307
308
|
# Build result
result = ParsedQuery(
original_query=query,
normalized_query=normalized,
rewritten_query=rewritten,
detected_language=detected_lang,
translations=translations,
query_vector=query_vector,
|
7bc756c5
tangwang
优化 ES 查询构建
|
309
310
311
312
313
|
domain=domain,
keywords=keywords,
token_count=token_count,
is_short_query=is_short_query,
is_long_query=is_long_query
|
be52af70
tangwang
first commit
|
314
315
|
)
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
316
317
|
if context and hasattr(context, 'logger'):
context.logger.info(
|
16c42787
tangwang
feat: implement r...
|
318
319
320
321
322
323
|
f"查询解析完成 | 原查询: '{query}' | 最终查询: '{rewritten or query_text}' | "
f"语言: {detected_lang} | 域: {domain} | "
f"翻译数量: {len(translations)} | 向量: {'是' if query_vector is not None else '否'}",
extra={'reqid': context.reqid, 'uid': context.uid}
)
else:
|
325eec03
tangwang
1. 日志、配置基础设施,使用优化
|
324
325
326
327
|
logger.info(
f"查询解析完成 | 原查询: '{query}' | 最终查询: '{rewritten or query_text}' | "
f"语言: {detected_lang} | 域: {domain}"
)
|
16c42787
tangwang
feat: implement r...
|
328
|
|
be52af70
tangwang
first commit
|
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
|
return result
def get_search_queries(self, parsed_query: ParsedQuery) -> List[str]:
"""
Get list of queries to search (original + translations).
Args:
parsed_query: Parsed query object
Returns:
List of query strings to search
"""
queries = [parsed_query.rewritten_query]
# Add translations
for lang, translation in parsed_query.translations.items():
if translation and translation != parsed_query.rewritten_query:
queries.append(translation)
return queries
def update_rewrite_rules(self, rules: Dict[str, str]) -> None:
"""
Update query rewrite rules.
Args:
rules: Dictionary of pattern -> replacement mappings
"""
for pattern, replacement in rules.items():
self.rewriter.add_rule(pattern, replacement)
def get_rewrite_rules(self) -> Dict[str, str]:
"""Get current rewrite rules."""
return self.rewriter.get_rules()
|