74fdf9bd
 tangwang
1.
|
1
2
3
4
5
6
7
8
9
|
"""
Product title exclusion detection for query understanding.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Set, Tuple
|
|
74fdf9bd
tangwang
1.
|
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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
154
155
156
157
158
159
160
|
def _dedupe_terms(terms: Iterable[str]) -> List[str]:
result: List[str] = []
seen: Set[str] = set()
for raw_term in terms:
term = normalize_query_text(raw_term)
if not term or term in seen:
continue
seen.add(term)
result.append(term)
return result
@dataclass(frozen=True)
class ProductTitleExclusionRule:
zh_trigger_terms: Tuple[str, ...]
en_trigger_terms: Tuple[str, ...]
zh_title_exclusions: Tuple[str, ...]
en_title_exclusions: Tuple[str, ...]
max_term_ngram: int = 3
@classmethod
def from_config_row(cls, row: Dict[str, Sequence[str]]) -> Optional["ProductTitleExclusionRule"]:
zh_trigger_terms = tuple(_dedupe_terms(row.get("zh_trigger_terms") or []))
en_trigger_terms = tuple(_dedupe_terms(row.get("en_trigger_terms") or []))
zh_title_exclusions = tuple(_dedupe_terms(row.get("zh_title_exclusions") or []))
en_title_exclusions = tuple(_dedupe_terms(row.get("en_title_exclusions") or []))
if not zh_title_exclusions and not en_title_exclusions:
return None
if not zh_trigger_terms and not en_trigger_terms:
return None
max_ngram = max(
[1]
+ [len(term.split()) for term in zh_trigger_terms]
+ [len(term.split()) for term in en_trigger_terms]
)
return cls(
zh_trigger_terms=zh_trigger_terms,
en_trigger_terms=en_trigger_terms,
zh_title_exclusions=zh_title_exclusions,
en_title_exclusions=en_title_exclusions,
max_term_ngram=max_ngram,
)
def match_candidates(self, candidates: Iterable[str]) -> Optional[str]:
normalized_candidates = {normalize_query_text(candidate) for candidate in candidates}
for term in self.zh_trigger_terms:
if term in normalized_candidates:
return term
for term in self.en_trigger_terms:
if term in normalized_candidates:
return term
return None
@dataclass(frozen=True)
class DetectedProductTitleExclusion:
matched_term: str
matched_query_text: str
zh_title_exclusions: Tuple[str, ...]
en_title_exclusions: Tuple[str, ...]
def to_dict(self) -> Dict[str, Any]:
return {
"matched_term": self.matched_term,
"matched_query_text": self.matched_query_text,
"zh_title_exclusions": list(self.zh_title_exclusions),
"en_title_exclusions": list(self.en_title_exclusions),
}
@dataclass(frozen=True)
class ProductTitleExclusionProfile:
query_variants: Tuple[TokenizedText, ...] = field(default_factory=tuple)
exclusions: Tuple[DetectedProductTitleExclusion, ...] = field(default_factory=tuple)
@property
def is_active(self) -> bool:
return bool(self.exclusions)
def to_dict(self) -> Dict[str, Any]:
return {
"active": self.is_active,
"exclusions": [item.to_dict() for item in self.exclusions],
"query_variants": [
{
"text": variant.text,
"normalized_text": variant.normalized_text,
"fine_tokens": list(variant.fine_tokens),
"coarse_tokens": list(variant.coarse_tokens),
"candidates": list(variant.candidates),
}
for variant in self.query_variants
],
}
def all_zh_title_exclusions(self) -> List[str]:
return _dedupe_terms(
term
for item in self.exclusions
for term in item.zh_title_exclusions
)
def all_en_title_exclusions(self) -> List[str]:
return _dedupe_terms(
term
for item in self.exclusions
for term in item.en_title_exclusions
)
class ProductTitleExclusionRegistry:
def __init__(
self,
rules: Sequence[ProductTitleExclusionRule],
*,
enabled: bool = True,
) -> None:
self.rules = tuple(rules)
self.enabled = bool(enabled)
self.max_term_ngram = max((rule.max_term_ngram for rule in self.rules), default=3)
@classmethod
def from_query_config(cls, query_config: Any) -> "ProductTitleExclusionRegistry":
raw_rules = getattr(query_config, "product_title_exclusion_rules", []) or []
rules: List[ProductTitleExclusionRule] = []
for row in raw_rules:
if not isinstance(row, dict):
continue
rule = ProductTitleExclusionRule.from_config_row(row)
if rule is not None:
rules.append(rule)
return cls(
rules,
enabled=bool(getattr(query_config, "product_title_exclusion_enabled", True)),
)
class ProductTitleExclusionDetector:
def __init__(
self,
registry: ProductTitleExclusionRegistry,
*,
tokenizer: Optional[Callable[[str], Any]] = None,
) -> None:
self.registry = registry
self.tokenizer = tokenizer
|
|
45b39796
tangwang
qp性能优化
|
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
def _tokenize_text(
self,
text: str,
*,
analysis_cache: Optional[QueryTextAnalysisCache] = None,
) -> TokenizedText:
if analysis_cache is not None:
return analysis_cache.get_tokenized_text(
text,
max_ngram=self.registry.max_term_ngram,
)
return tokenize_text(
text,
tokenizer=self.tokenizer,
max_ngram=self.registry.max_term_ngram,
)
|
|
74fdf9bd
tangwang
1.
|
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
texts = [
getattr(parsed_query, "original_query", None),
getattr(parsed_query, "query_normalized", None),
getattr(parsed_query, "rewritten_query", None),
]
translations = getattr(parsed_query, "translations", {}) or {}
if isinstance(translations, dict):
texts.extend(translations.values())
for raw_text in texts:
text = str(raw_text or "").strip()
if not text:
continue
normalized = normalize_query_text(text)
if not normalized or normalized in seen:
continue
seen.add(normalized)
variants.append(
|
|
74fdf9bd
tangwang
1.
|
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
|
)
)
return tuple(variants)
def detect(self, parsed_query: Any) -> ProductTitleExclusionProfile:
if not self.registry.enabled or not self.registry.rules:
return ProductTitleExclusionProfile()
query_variants = self._build_query_variants(parsed_query)
detected: List[DetectedProductTitleExclusion] = []
seen_keys = set()
for variant in query_variants:
for rule in self.registry.rules:
matched_term = rule.match_candidates(variant.candidates)
if not matched_term:
continue
key = (
tuple(rule.zh_title_exclusions),
tuple(rule.en_title_exclusions),
)
if key in seen_keys:
continue
seen_keys.add(key)
detected.append(
DetectedProductTitleExclusion(
matched_term=matched_term,
matched_query_text=variant.text,
zh_title_exclusions=rule.zh_title_exclusions,
en_title_exclusions=rule.en_title_exclusions,
)
)
return ProductTitleExclusionProfile(
query_variants=query_variants,
exclusions=tuple(detected),
)
|