增加多模态模型开放集识别。

2025-11-13 11:56:51 +08:00
parent b4b2b18ccd
commit ced0e4f7ef
7 changed files with 628 additions and 21 deletions
@@ -1,9 +1,12 @@
 import json
 import re
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple

 from .vlm_providers.base import VLMProvider
-from .prompts import build_closedset_prompt
+from .prompts import build_closedset_prompt, build_openset_prompt
+from .dish_name_cleaner import normalize_dish_name
+from .text_embedder import OllamaEmbedder
+from .vector_matcher import DishNameMatcher


 def _extract_json_obj(text: str):
@@ -93,3 +96,133 @@ def classify_image(
    return {"label": "Unknown", "confidence": 0.0}


+def classify_image_openset(
+    image_path: str,
+    provider: VLMProvider,
+    embedder: OllamaEmbedder,
+    matcher: DishNameMatcher,
+    top_k: int = 3,
+    min_match_score: float = 0.5,
+) -> Dict:
+    """
+    开放式菜品识别 + 向量匹配方案
+    
+    流程：
+        1. VLM开放式识别（不给候选列表）
+        2. 清洗菜品名
+        3. 文本向量化
+        4. FAISS检索Top-K候选
+        5. 综合置信度排序
+    
+    参数：
+        image_path: 图片路径
+        provider: VLM Provider实例
+        embedder: Ollama Embedder实例
+        matcher: 菜品名匹配器实例
+        top_k: 向量检索返回的候选数
+        min_match_score: 最低匹配分数阈值
+    
+    返回：
+        {
+            "raw_dish": "VLM原始输出",
+            "cleaned_dish": "清洗后的菜品名",
+            "vlm_confidence": 0.95,
+            "candidates": [
+                {"dish": "匹配的标准菜名", "match_score": 0.88, "final_score": 0.83},
+                ...
+            ],
+            "best_match": "最佳匹配菜名",
+            "final_confidence": 0.83
+        }
+    """
+    print(f"[VLM_Openset] Start classify image={image_path} top_k={top_k}")
+    
+    # 步骤1: VLM开放式识别
+    prompt = build_openset_prompt()
+    print(f"[VLM_Openset] Prompt length={len(prompt)}")
+    
+    print(f"[VLM_Openset] Calling {provider}.chat_vision()...")
+    text = provider.chat_vision(prompt, [image_path], temperature=0.1)
+    _preview = (str(text)[:200]).replace("\n", " ")
+    print(f"[VLM_Openset] Received text length={len(str(text))} preview={_preview}...")
+    
+    # 步骤2: 解析VLM输出
+    obj = _extract_json_obj(str(text))
+    if not obj or "dish" not in obj:
+        print("[VLM_Openset] JSON parse failed or missing 'dish' field")
+        return {
+            "raw_dish": str(text),
+            "cleaned_dish": "",
+            "vlm_confidence": 0.0,
+            "candidates": [],
+            "best_match": "Unknown",
+            "final_confidence": 0.0,
+        }
+    
+    raw_dish = obj.get("dish", "")
+    vlm_conf = obj.get("confidence", 0.0)
+    try:
+        vlm_conf = float(vlm_conf)
+    except Exception:
+        vlm_conf = 0.0
+    
+    print(f"[VLM_Openset] Parsed raw_dish='{raw_dish}' vlm_confidence={vlm_conf}")
+    
+    # 步骤3: 清洗菜品名
+    cleaned_dish = normalize_dish_name(raw_dish)
+    print(f"[VLM_Openset] Cleaned dish='{cleaned_dish}'")
+    
+    if not cleaned_dish:
+        print("[VLM_Openset] Cleaned dish is empty, return Unknown")
+        return {
+            "raw_dish": raw_dish,
+            "cleaned_dish": cleaned_dish,
+            "vlm_confidence": vlm_conf,
+            "candidates": [],
+            "best_match": "Unknown",
+            "final_confidence": 0.0,
+        }
+    
+    # 步骤4: 向量匹配
+    print(f"[VLM_Openset] Embedding text='{cleaned_dish}'...")
+    query_emb = embedder.encode_single(cleaned_dish)
+    
+    print(f"[VLM_Openset] Matching with FAISS index...")
+    matches = matcher.match(query_emb, top_k)
+    
+    # 步骤5: 构建候选列表（综合置信度 = VLM置信度 × 匹配分数）
+    candidates = []
+    for dish_name, match_score in matches:
+        if match_score < min_match_score:
+            print(f"[VLM_Openset] Skip candidate '{dish_name}' (score={match_score:.4f} < threshold={min_match_score})")
+            continue
+        
+        final_score = vlm_conf * match_score
+        candidates.append({
+            "dish": dish_name,
+            "match_score": match_score,
+            "final_score": final_score,
+        })
+    
+    # 按综合分数降序排序
+    candidates.sort(key=lambda x: x["final_score"], reverse=True)
+    
+    # 最佳匹配
+    if candidates:
+        best = candidates[0]
+        best_match = best["dish"]
+        final_conf = best["final_score"]
+        print(f"[VLM_Openset] Best match='{best_match}' final_confidence={final_conf:.4f}")
+    else:
+        best_match = "Unknown"
+        final_conf = 0.0
+        print("[VLM_Openset] No valid candidates, return Unknown")
+    
+    return {
+        "raw_dish": raw_dish,
+        "cleaned_dish": cleaned_dish,
+        "vlm_confidence": vlm_conf,
+        "candidates": candidates,
+        "best_match": best_match,
+        "final_confidence": final_conf,
+    }