相关性调参、enrich内容解析优化

tangwang
1 parent 8140e942
Showing 7 changed files with 185 additions and 48 deletions Show diff stats
api/routes/indexer.py
config/config.yaml
docs/TODO.txt
indexer/document_transformer.py
indexer/product_enrich.py
search/es_query_builder.py
suggestion/builder.py
@@ -449,7 +449,7 @@ def _run_enrich_content(tenant_id: str, items: List[Dict[str, str]], languages: 
     同步执行内容理解：调用 product_enrich.analyze_products，按语言批量跑 LLM，
     再聚合成每 SPU 的 qanchors、semantic_attributes、tags。供 run_in_executor 调用。
     """
-    from indexer.product_enrich import analyze_products
+    from indexer.product_enrich import analyze_products, split_multi_value_field
  
     llm_langs = list(dict.fromkeys(languages)) or ["en"]
  
@@ -510,10 +510,7 @@ def _run_enrich_content(tenant_id: str, items: List[Dict[str, str]], languages: 
                 raw = row.get(name)
                 if not raw:
                     continue
-                for part in re.split(r"[,;|/\n\t]+", str(raw)):
-                    value = part.strip()
-                    if not value:
-                        continue
+                for value in split_multi_value_field(str(raw)):
                     rec["semantic_attributes"].append({"lang": lang, "name": name, "value": value})
                     if name == "tags":
                         rec["tags"].append(value)
@@ -75,16 +75,15 @@ es_settings:
 # 若需要按某个语言单独调权，也可以加显式 key（例如 title.de: 3.2）。
 field_boosts:
   title: 3.0
+  qanchors: 2.5
+  tags: 2.0
+  category_name_text: 2.0
+  category_path: 2.0
   brief: 1.5
-  description: 1.0
-  qanchors: 1.5
-  vendor: 1.5
-  category_path: 1.5
-  category_name_text: 1.5
-  tags: 1.0
-  option1_values: 0.6
-  option2_values: 0.4
-  option3_values: 0.4
+  description: 1.5
+  option1_values: 1.5
+  option2_values: 1.5
+  option3_values: 1.5
  
 # Query Configuration（查询配置）
 query_config:
@@ -122,11 +121,11 @@ query_config:
   search_fields:
     multilingual_fields:
       - "title"
-      - "brief"
-      - "description"
-      - "vendor"
+      - "qanchors"
       - "category_path"
       - "category_name_text"
+      - "brief"
+      - "description"
     shared_fields:
       - "tags"
       - "option1_values"
@@ -135,15 +134,14 @@ query_config:
     core_multilingual_fields:
       - "title"
       - "brief"
-      - "vendor"
       - "category_name_text"
  
   # 统一文本召回策略（主查询 + 翻译查询）
   text_query_strategy:
     base_minimum_should_match: "75%"
     translation_minimum_should_match: "75%"
-    translation_boost: 0.6
-    tie_breaker_base_query: 0.9
+    translation_boost: 0.75
+    tie_breaker_base_query: 0.5
  
   # Embedding字段名称
   text_embedding_field: "title_embedding"
  
+
+
+本地部署一个7b Q4量化的大模型
+es需要licence的两个功能，如果费用低，开通下licence，或者改es源码定制开发下，支持 rank.rrf，reranker
+
+
+
 把knn跟文本相关性的融合方式修改为  "rank": {"rrf": {} }需要licence，可以帮我修改源码支持吗？
  
   knn_boost: 2.0  
@@ -13,8 +20,6 @@
 }
  
  
-
-
 "image_embedding": {
   "type": "nested",
   "properties": {
@@ -58,34 +63,70 @@ image_embedding改为，一个spu有多个sku向量，每个向量内部properti
 2. ES支持reranker pipline？
  
  
-@reranker/backends/qwen3_vllm.py 单次 generate 前有进程内锁，同一进程里不会并行多路 vLLM 推理，这个锁有必要吗？是否会影响性能？是否能够打开，使得性能更好？比如这个场景，我一次请求 400 条，分成每64个一个batch，基于我现在的gpu配置，可以再提高并发度吗？
-测试了，让每个批次都并发地进行，耗时没有变化
+
+
+
  
 增加款式意图识别模块
  
-意图类型： 颜色，尺寸（目前只需要支持这两种）
+意图类型： 颜色，尺码（目前只需要支持这两种）
  
 意图召回层：
 每种意图，有一个召回词集合
 对query（包括原始query、各种翻译query 都做匹配）
  
-意图识别层：
-如果召回 判断有款式需求，
+以颜色意图为例：
+有一个词表，每一行 都逗号分割，互为同义词，行内第一个为标准化词
+query匹配了其中任何一个词，都认为，具有颜色意图
+匹配规则： 用细粒度、粗粒度分词，看是否有在词表中的。原始query分词、和每种翻译的分词，都要用。
+
+意图判断： 暂时留空，直接返回true。目前没有模型，即只要召回了（词表匹配了），即认为有该维度款式需求。
+
+
+
+意图使用：
+
+我们第一阶段，使用 参与ES提权。
+
+一、参与ES提权
+
+
+二、参与reranker
  
  
-是否有：
-颜色需求
-尺码需求
 如果有： 先做sku筛选，然后把最优的拼接到名称中，参与reranker。
  
  
 现在在reranker、分页之后、做填充的时候，已经有做sku的筛选。
 需要优化：
 现在是，先做包含的判断，找到第一个 option_value被query包含的，则直接认为匹配。改为
-1. 第一轮：遍历完，如果有且仅有一个才这样。
-2. 第二轮：如果有多个，跳到3。如果没有，对每个词都走泛化词表进行匹配。
+1. 第一轮：遍历完，如果有且仅有一个被query包含，那么认为匹配。
+2. 第二轮：如果有多个符合（被query包含），跳到3。如果没有，对每个词都走泛化词表进行匹配。
 3. 第三轮：如果有多个，那么对这多个，走embedding相关性取最高的。如果一个也没有，则对所有的走embedding相关性取最高的
-这个sku筛选也需要提取为一个独立的模块
+这个sku筛选也需要提取为一个独立的模块。
+
+
+另外：现在是reranker、分页之后做sku筛选，要改为：
+1. 有款式意图的时候，才做sku筛选
+2. sku筛选的时机，改为在reranker之前，对所有内容做sku筛选，然后
+3. 从仅 option1 扩展到多个维度，识别的意图，包含意图的维度名（color）和维度名的泛化词list（color、颜色、colour、olors、、、、），遍历option1_name,option2_name,option3_name，看哪个能匹配上意图的维度名list，哪个匹配上了，则在这个维度筛选。
+4. Rerank doc （有款式意图的时候）要带上属性后缀，拼接到title后面。在调用 run_rerank 前，对每条 hit 生成「用于重排的 doc 文本」（标题 + 可选后缀）
+5. TODO ： 还有一个问题。  目前，sku只返回一个维度（店铺主维度。默认应该是option1，不是所有维度的sku信息都返回的。所以，如果有款式意图，但是主维度是颜色，那么拿不到全的款式sku）
+
+
+
+
+当前项目功能已经较多，但是有清晰的框架，请务必基于现有框架进行改造，不要进行补丁式的修改，避免代码逻辑分叉。
+
+请一步一步来，先设计意图识别模块，仔细思考需求，意图识别模块需要提供哪些内容，用于返回数据接口的定义，深度思考，定义一个合理的接口后，再给出合理的模块设计。
+
+
+
+
+
+
+
+
  
  
  
@@ -398,6 +439,31 @@ embeddings/image_encoder.py：requests.post(..., timeout=self.timeout_sec)
  
  
  
+
+
+
+
+
+
+
+多reranker：
+
+改 reranker 服务，一次请求返回多路分
+服务启动时 加载多个 backend（或按请求懒加载），/rerank 响应扩展为例如
+scores: [...]（兼容主后端）+ scores_by_backend: { "bge": [...], "qwen3_vllm": [...] }。
+搜索侧解析多路分，再融合或只透传 debug。
+优点：搜索侧仍只调一个 URL。缺点：单进程多大模型 显存压力很大；
+
+融合层要注意的一点
+fuse_scores_and_resort 目前只消费 一条 rerank_scores 序列，并写入 _rerank_score
+多 backend 之后需要rerank_scores 都参与融合
+
+
+
+
+
+
+
 product_enrich : Partial Mode   :   done
 https://help.aliyun.com/zh/model-studio/partial-mode?spm=a2c4g.11186623.help-menu-2400256.d_0_3_0_7.74a630119Ct6zR
 需在messages 数组中将最后一条消息的 role 设置为 assistant，并在其 content 中提供前缀，在此消息中设置参数 "partial": true。messages格式如下：
@@ -13,7 +13,7 @@ import numpy as np
 import logging
 import re
 from typing import Dict, Any, Optional, List
-from indexer.product_enrich import analyze_products
+from indexer.product_enrich import analyze_products, split_multi_value_field
  
 logger = logging.getLogger(__name__)
  
@@ -121,7 +121,7 @@ class SPUDocumentTransformer:
         # Tags
         if pd.notna(spu_row.get('tags')):
             tags_str = str(spu_row['tags'])
-            doc['tags'] = [tag.strip() for tag in tags_str.split(',') if tag.strip()]
+            doc['tags'] = split_multi_value_field(tags_str)
  
         # Category相关字段
         self._fill_category_fields(doc, spu_row)
@@ -282,11 +282,7 @@ class SPUDocumentTransformer:
                 raw = row.get(name)
                 if not raw:
                     continue
-                parts = re.split(r"[,;|/\n\t]+", str(raw))
-                for part in parts:
-                    value = part.strip()
-                    if not value:
-                        continue
+                for value in split_multi_value_field(str(raw)):
                     semantic_list.append({"lang": lang, "name": name, "value": value})
  
             if qanchors_obj:
@@ -703,11 +699,7 @@ class SPUDocumentTransformer:
                 raw = row.get(name)
                 if not raw:
                     continue
-                parts = re.split(r"[,;|/\n\t]+", str(raw))
-                for part in parts:
-                    value = part.strip()
-                    if not value:
-                        continue
+                for value in split_multi_value_field(str(raw)):
                     semantic_list.append(
                         {
                             "lang": lang,
@@ -144,6 +144,20 @@ if _missing_prompt_langs:
     )
  
  
+# 多值字段分隔：英文逗号、中文逗号、顿号，及历史约定的 ; | / 与空白
+_MULTI_VALUE_FIELD_SPLIT_RE = re.compile(r"[，、,;|/\n\t]+")
+
+
+def split_multi_value_field(text: Optional[str]) -> List[str]:
+    """将 LLM/业务中的多值字符串拆成短语列表（strip 后去空）。"""
+    if text is None:
+        return []
+    s = str(text).strip()
+    if not s:
+        return []
+    return [p.strip() for p in _MULTI_VALUE_FIELD_SPLIT_RE.split(s) if p.strip()]
+
+
 def _normalize_space(text: str) -> str:
     return re.sub(r"\s+", " ", (text or "").strip())
  
@@ -38,6 +38,10 @@ class ESQueryBuilder:
         translation_boost: float = 0.4,
         tie_breaker_base_query: float = 0.9,
         mixed_script_merged_field_boost_scale: float = 0.6,
+        phrase_match_base_fields: Optional[Tuple[str, ...]] = None,
+        phrase_match_slop: int = 2,
+        phrase_match_tie_breaker: float = 0.4,
+        phrase_match_boost: float = 3.0,
     ):
         """
         Initialize query builder.
@@ -73,6 +77,10 @@ class ESQueryBuilder:
         self.translation_boost = float(translation_boost)
         self.tie_breaker_base_query = float(tie_breaker_base_query)
         self.mixed_script_merged_field_boost_scale = float(mixed_script_merged_field_boost_scale)
+        self.phrase_match_base_fields = tuple(phrase_match_base_fields or ("title", "qanchors"))
+        self.phrase_match_slop = int(phrase_match_slop)
+        self.phrase_match_tie_breaker = float(phrase_match_tie_breaker)
+        self.phrase_match_boost = float(phrase_match_boost)
  
     def _apply_source_filter(self, es_query: Dict[str, Any]) -> None:
         """
@@ -453,6 +461,49 @@ class ESQueryBuilder:
         """Format (field_path, boost) pairs for Elasticsearch multi_match ``fields``."""
         return [self._format_field_with_boost(path, boost) for path, boost in specs]
  
+    def _build_phrase_match_fields(self, language: str) -> List[str]:
+        """Fields for phrase multi_match: base names × ``.{lang}`` with ``field_boosts``."""
+        lang = (language or "").strip().lower()
+        if not lang:
+            return []
+        out: List[str] = []
+        for base in self.phrase_match_base_fields:
+            path = f"{base}.{lang}"
+            boost = self._get_field_boost(base, lang)
+            out.append(self._format_field_with_boost(path, boost))
+        return out
+
+    def _append_phrase_should_clause(
+        self,
+        should_clauses: List[Dict[str, Any]],
+        lang: str,
+        lang_query: str,
+        clause_name: str,
+        is_source: bool,
+    ) -> None:
+        text = (lang_query or "").strip()
+        if not text:
+            return
+        phrase_fields = self._build_phrase_match_fields(lang)
+        if not phrase_fields:
+            return
+        boost = (
+            self.phrase_match_boost
+            if is_source
+            else self.phrase_match_boost * float(self.translation_boost)
+        )
+        should_clauses.append({
+            "multi_match": {
+                "_name": f"{clause_name}_phrase",
+                "query": lang_query,
+                "type": "phrase",
+                "fields": phrase_fields,
+                "slop": self.phrase_match_slop,
+                "tie_breaker": self.phrase_match_tie_breaker,
+                "boost": boost,
+            }
+        })
+
     def _merge_supplemental_lang_field_specs(
         self,
         specs: List[MatchFieldSpec],
@@ -590,6 +641,9 @@ class ESQueryBuilder:
             should_clauses.append({
                 "multi_match": clause["multi_match"]
             })
+            self._append_phrase_should_clause(
+                should_clauses, lang, lang_query, clause_name, is_source
+            )
  
         if base_query_text:
             append_clause(source_lang, base_query_text, "base_query", True)
@@ -606,7 +660,7 @@ class ESQueryBuilder:
         # Fallback to a simple query when language fields cannot be resolved.
         if not should_clauses:
             fallback_fields = self.match_fields or ["title.en^1.0"]
-            return {
+            fallback_lexical = {
                 "multi_match": {
                     "_name": "base_query_fallback",
                     "query": query_text,
@@ -615,6 +669,22 @@ class ESQueryBuilder:
                     "tie_breaker": self.tie_breaker_base_query,
                 }
             }
+            fb_should: List[Dict[str, Any]] = [fallback_lexical]
+            self._append_phrase_should_clause(
+                fb_should,
+                self.default_language,
+                query_text,
+                "base_query_fallback",
+                True,
+            )
+            if len(fb_should) == 1:
+                return fallback_lexical
+            return {
+                "bool": {
+                    "should": fb_should,
+                    "minimum_should_match": 1,
+                }
+            }
  
         # Return bool query with should clauses
         if len(should_clauses) == 1:
@@ -147,7 +147,7 @@ class SuggestionIndexBuilder:
         raw = str(value).strip()
         if not raw:
             return []
-        parts = re.split(r"[,;|/\n\t]+", raw)
+        parts = re.split(r"[，、,;|/\n\t]+", raw)
         out = [p.strip() for p in parts if p and p.strip()]
         if not out:
             return [raw]
@@ -162,7 +162,7 @@ class SuggestionIndexBuilder:
         s = str(raw).strip()
         if not s:
             return []
-        parts = re.split(r"[,;|/\n\t]+", s)
+        parts = re.split(r"[，、,;|/\n\t]+", s)
         out = [p.strip() for p in parts if p and p.strip()]
         return out if out else [s]