修复2个问题：微信OCR hash碰撞(aHash→dHash+5秒强制刷新) + KB搜索动词+对象精准匹配防误召回

2026-07-28 22:28:29 +08:00
parent 1ccaa926fd
commit a093528888
2 changed files with 125 additions and 21 deletions
@@ -211,7 +211,38 @@ class KnowledgeBase:
    def is_ready(self) -> bool:
        return self._ready and len(self.docs) > 0

-    def search(self, query: str, top_k: int = 3, min_score: float = 0.05) -> List[Tuple[dict, float]]:
+    @staticmethod
+    def _extract_verb_object(query: str, action_verbs: set) -> List[Tuple[str, str]]:
+        """从查询中提取动词+对象对。
+        
+        例："怎么打印流水" → [("打印", "流水")]
+             "圈地工具用什么" → [("圈地", "工具")]
+        """
+        import re
+        pairs = []
+        # 模式1: 动词 + 对象（2+字动词 + 2+字名词）
+        for verb in action_verbs:
+            if verb in query:
+                # 找动词后面的内容作为对象
+                idx = query.index(verb) + len(verb)
+                rest = query[idx:].strip()
+                # 取动词后 2~6 字作为对象
+                if len(rest) >= 2:
+                    obj = rest[:6].strip()
+                    # 去掉常见无意义尾词
+                    for stop in ["的", "地", "得", "了", "吗", "呢", "吧", "啊"]:
+                        obj = obj.rstrip(stop)
+                    if len(obj) >= 2:
+                        pairs.append((verb, obj))
+        # 模式2: 2字动词 + 2字对象（从分词结果匹配）
+        if not pairs:
+            tokens = [t for t in jieba.cut(query) if len(t) >= 2]
+            for i, t in enumerate(tokens):
+                if t in action_verbs and i + 1 < len(tokens):
+                    pairs.append((t, tokens[i + 1]))
+        return pairs
+
+    def search(self, query: str, top_k: int = 3, min_score: float = 0.1) -> List[Tuple[dict, float]]:
        """搜索知识库，返回 (doc, score) 列表。

        策略：
@@ -310,6 +341,35 @@ class KnowledgeBase:
        has_def = bool(def_words & query_lower)
        def_headings = {"版本号", "版本", "区分", "定义"}

+        # 动作查询：提取动词+对象，对不含两者的文档降权
+        # 防止"圈地工具"匹配到含"工具"但不含"圈地"的文档（如寻宝插件）
+        action_verbs_all = {"打印", "查看", "设置", "创建", "使用", "查询", "导出", "执行", "怎么", "如何", "用法", "教程", "操作"}
+        verb_obj_pairs = self._extract_verb_object(query_stripped, action_verbs_all)
+        if verb_obj_pairs:
+            adjusted = []
+            for doc_title, doc_score in sorted_docs:
+                # 检查文档内容是否包含动词+对象
+                doc_text = ""
+                for i, doc in enumerate(self.docs):
+                    if doc["title"] == doc_title:
+                        doc_text += doc.get("heading", "") + " " + doc.get("content", "")
+                # 计算匹配度：动词+对象都包含 → 满分；只包含一个 → 降权；都不包含 → 大幅降权
+                match_count = 0
+                for verb, obj in verb_obj_pairs:
+                    v_in = verb in doc_text
+                    o_in = obj in doc_text
+                    if v_in and o_in:
+                        match_count += 2
+                    elif v_in or o_in:
+                        match_count += 1
+                if match_count == 0:
+                    doc_score *= 0.2  # 完全不匹配，大幅降权
+                elif match_count == 1:
+                    doc_score *= 0.5  # 部分匹配，降权
+                adjusted.append((doc_title, doc_score))
+            adjusted.sort(key=lambda x: x[1], reverse=True)
+            sorted_docs = adjusted
+
        # 定义类查询：检查每个文档是否有定义类 section，没有则降权
        if has_def:
            adjusted = []
@@ -344,22 +404,45 @@ class KnowledgeBase:

            # 同文档内，根据查询意图优先选合适的 section
            if has_action and len(doc_sections) > 1:
-                # 动作查询：按优先级匹配 heading 词
-                # "参数/指令/使用" > "方法" > "安装"
-                priority_order = [
-                    ({"参数", "指令", "使用"}),  # 最高优先级
-                    ({"方法"}),                   # 中等优先级（避免"安装方法"误匹配"参数"类）
-                    ({"安装"}),                   # 最低优先级
-                ]
-                for priority_level in priority_order:
+                # 动作查询：提取动词+对象，精准匹配 section
+                # 如"怎么打印流水" → verb="打印", obj="流水"，优先选 heading 同时含两者的 section
+                action_verbs = {"打印", "查看", "设置", "创建", "使用", "查询", "导出", "执行"}
+                verb_obj_pairs = self._extract_verb_object(query_stripped, action_verbs)
+                
+                if verb_obj_pairs:
+                    # 有动词+对象：找 heading 或 body 同时包含两者的 section
+                    best_idx = -1
+                    best_match = 0
                    for i in range(len(doc_sections)):
                        idx = doc_sections[i][0]
-                        if priority_level & self._heading_token_sets[idx]:
-                            doc_sections.insert(0, doc_sections.pop(i))
-                            break
-                    else:
-                        continue
-                    break  # 找到匹配就停止
+                        heading = self.docs[idx].get("heading", "")
+                        content = self.docs[idx].get("content", "")
+                        for verb, obj in verb_obj_pairs:
+                            v_in_h = verb in heading or verb in content
+                            o_in_h = obj in heading or obj in content
+                            match_score = int(v_in_h) + int(o_in_h)
+                            if match_score > best_match:
+                                best_match = match_score
+                                best_idx = i
+                    if best_idx > 0:
+                        doc_sections.insert(0, doc_sections.pop(best_idx))
+                else:
+                    # 无动词+对象：按优先级匹配 heading 词（旧逻辑）
+                    # "参数/指令/使用" > "方法" > "安装"
+                    priority_order = [
+                        ({"参数", "指令", "使用"}),  # 最高优先级
+                        ({"方法"}),                   # 中等优先级
+                        ({"安装"}),                   # 最低优先级
+                    ]
+                    for priority_level in priority_order:
+                        for i in range(len(doc_sections)):
+                            idx = doc_sections[i][0]
+                            if priority_level & self._heading_token_sets[idx]:
+                                doc_sections.insert(0, doc_sections.pop(i))
+                                break
+                        else:
+                            continue
+                        break  # 找到匹配就停止

            if has_def and len(doc_sections) > 1:
                # 定义查询：选 heading 中匹配最多定义类词的 section