From d87e482e1e058900d74e524594c1a49e2c17178c Mon Sep 17 00:00:00 2001 From: youpaishidifu Date: Mon, 3 Aug 2026 20:52:08 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E5=BE=AE=E4=BF=A1=E7=AB=AF?= =?UTF-8?q?=E7=9F=A5=E8=AF=86=E5=BA=93=E7=AD=94=E9=9D=9E=E6=89=80=E9=97=AE?= =?UTF-8?q?=20+=20QQ=20OCR=20=E6=BC=8F=E6=B6=88=E6=81=AF=EF=BC=88=E5=8F=8C?= =?UTF-8?q?=20bug=20=E4=BF=AE=E5=A4=8D=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit knowledge_base.py（问题2：微信端答非所问）： - 文档级标题语义匹配改为 token 级（原仅字面子串匹配），OCR 丢失"php"只剩"管理员端默认"也能正确命中 phpadmin - 新增"管理员+默认"组合强偏好：无"领地"时 phpadmin ×3.0、protect ×0.4，避免退化查询误选 protect/全局配置 - section 级重排新增 bonus +30：查询含"管理员+默认"时，偏好正文真正含"密码/登录/ypshidifu"的 section，压过"## 目录"目录项 monitor.py（问题1 + 问题2 上游）： - 新增 _split_and_denoise_segments：按消息边界预分割 + 分段噪声隔离，修复 Windows OCR 大杂烩把用户消息当噪声整段丢弃（QQ 漏消息） - 候选行剥离 bot 自身旧回复文本：OCR 把用户新提问与屏上 bot 旧回复拼成一行时，先剥离旧回复再走过滤，消除答非所问的上游拼接根因 测试：新增 debug_kb.py 复现退化查询、test_kb.py 回归；跨主题回归全部正确，py_compile 通过 --- debug_kb.py | 56 ++++++++++++++++++++++++ knowledge_base.py | 51 +++++++++++++--------- monitor.py | 109 +++++++++++++++++++++++++++++++++++++++++++++- test_kb.py | 40 +++++++++++++++++ 4 files changed, 235 insertions(+), 21 deletions(-) create mode 100644 debug_kb.py create mode 100644 test_kb.py diff --git a/debug_kb.py b/debug_kb.py new file mode 100644 index 0000000..56c205c --- /dev/null +++ b/debug_kb.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""调试知识库搜索 — 打印每个 section 的分数，定位为什么返回错误结果""" +import sys, os, re +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import jieba +from knowledge_base import KnowledgeBase + +def dump(query): + kb = KnowledgeBase("D:/桌面/源码/文档/docs") + kb.reload() + print(f"\n{'='*70}") + print(f"查询: {query!r}") + print(f"jieba tokens: {list(jieba.cut(query))}") + query_nouns = set(jieba.cut(query)) & {'密码','用户管理','领地','商品','CDK','工单','债券','在线'} + print(f"query core_nouns: {query_nouns}") + + results = kb.search(query, top_k=3) + print("\n--- 最终返回结果 (top_k=3) ---") + if results: + for i,(doc,score) in enumerate(results): + print(f" [{i+1}] {doc['title']} | {score:.2f} | heading={doc['heading'][:50]!r}") + print(f" body={doc['body'][:80]!r}") + else: + print(" 无结果") + + # 手动遍历所有 section，打印 combined_scores 近似（不精确，但看排序） + # 直接复用内部逻辑太复杂，这里只打印每个 doc 的 heading+body 片段供人工判断 + print("\n--- 知识库所有 section（含'密码'/'管理员'关键词的优先）---") + matches = [] + for d in kb.docs: + txt = d.get('heading','') + ' ' + d.get('body','') + if '密码' in txt or '管理员' in txt: + matches.append(d) + for d in matches: + print(f" [{d['title']}] heading={d['heading'][:40]!r}") + print(f" body={d['body'][:90]!r}") + +if __name__ == "__main__": + for q in [ + # 干净查询（对照组，应返回 phpadmin） + "PHP管理员端的默认密码是多少？", + # 假设 OCR 丢失"密码"二字 + "PHP管理员端的默认是多少？", + "php管理员端默认", + "管理员端默认", + "php管理员端默认是多少", + # 假设 OCR 把"密码"拆成"密 码"或加了空格 + "php 管理员 端 默认 密码 是 多少", + # 假设查询被拼接了 bot 可见的旧回复（全局配置） + "PHP管理员端的默认密码是多少？全局配置：管理员可通过该配置，配置默认用户拥有多少格领地，创建领地价格是多少", + # 假设 OCR 把 PHP 误读（大小写/空格） + "Php管理员端的默认密码是多少？", + "php 管理员端的默认密码是多少？", + ]: + dump(q) diff --git a/knowledge_base.py b/knowledge_base.py index 66da2da..e6d521d 100644 --- a/knowledge_base.py +++ b/knowledge_base.py @@ -466,30 +466,35 @@ class KnowledgeBase: # 弱匹配（score < min_score*2）不计入 doc_count，防低分堆叠 # ⭐ 文档标题语义匹配奖励：查询词与文档标题语义相关 → 巨幅奖励 # 如"php管理员"应优先返回phpadmin文档，"领地"应优先返回protect文档 - title_query_map = { - 'phpadmin': ['php管理员', 'phpadmin', 'php端管理员'], - 'protect': ['领地', 'protect', '保护'], - 'login': ['登录', 'login', '账号'], - 'beibao': ['背包', 'beibao', '仓库'], + # 改为 token 级匹配（而非字面子串），OCR 丢失"php"只剩"管理员端默认"也能命中 phpadmin + DOC_TOPIC_TOKENS = { + 'phpadmin': {'php', 'php管理员', '后端'}, + 'protect': {'领地', 'protect', '保护'}, + 'login': {'登录', 'login', '账号', 'sdf1'}, + 'beibao': {'背包', 'beibao', '仓库', '云背包'}, } + query_lc = query_stripped.lower() + # 查询 token 统一小写，与 DOC_TOPIC_TOKENS 对齐（OCR 常把 PHP→php/Php） + query_token_set2 = set(t.lower() for t in jieba.cut(query_stripped) if len(t) >= 2) + q_has_admin = '管理员' in query_token_set2 + q_has_default = '默认' in query_token_set2 for doc_title in doc_best: # 精确匹配（文档标题在查询中） - if doc_title.lower() in query_stripped.lower(): + if doc_title.lower() in query_lc: doc_best[doc_title] *= 5.0 - else: - # 语义匹配：查询词包含文档标题的同义词 - title_keywords = title_query_map.get(doc_title, [doc_title]) - # 精确短语匹配（如"php管理员"在查询中） - exact_phrase_match = any(kw in query_stripped.lower() for kw in title_keywords) - if exact_phrase_match: - # 精确短语匹配 → 高权重 - doc_best[doc_title] *= 4.0 - else: - # 部分匹配（如查询包含"php"但不包含"php管理员"） - # 这种情况下，更具体的文档（如phpadmin）应该比泛泛的文档（如php）权重更高 - partial_match = any(kw.split('_')[0] in query_stripped.lower() for kw in [doc_title] if '_' in kw) - if partial_match: - doc_best[doc_title] *= 1.5 + continue + # token 级语义匹配：文档主题 token 命中查询 → 加权（命中越多权重越高） + topic_tokens = DOC_TOPIC_TOKENS.get(doc_title, set()) + hits = topic_tokens & query_token_set2 + if hits: + doc_best[doc_title] *= (2.5 + 1.0 * len(hits)) + # ⭐ 管理员 + 默认 组合 → 强偏好 phpadmin 文档 + # 此时 protect/全局配置 虽含"管理员+默认"，但主题是领地；查询无"领地"时应让位 + if q_has_admin and q_has_default: + if 'phpadmin' in doc_best and '领地' not in query_token_set2: + doc_best['phpadmin'] *= 3.0 + if 'protect' in doc_best and '领地' not in query_token_set2: + doc_best['protect'] *= 0.4 sorted_docs = sorted( doc_best.items(), key=lambda x: (x[1], doc_count.get(x[0], 0)), @@ -597,6 +602,12 @@ class KnowledgeBase: body_noun_hits = sum(1 for n in core_nouns if n in body and n in query_stripped) if body_noun_hits >= 1: bonus += 15.0 # body 包含核心名词 → 优先返回 + # ⭐ 管理员+默认 组合（OCR 常丢"php/密码"）→ 偏好真正含"密码/登录"的 section + # 防止"## 目录 - 管理员端(当前)"这种目录项压过"## php管理员后端"(默认登录密码) + # 权重给足，必须压过目录项因 heading/phrase 命中的基础分差 + if '管理员' in query_tokens and '默认' in query_tokens: + if any(k in body for k in ('密码', '登录', 'ypshidifu')): + bonus += 30.0 # 文档头部惩罚：heading 是单层 # + 文档名（如 "# php管理员后端"），且无 body # ⭐ 例外：body 包含查询核心词时不惩罚（可能是精确答案） body_has_query = any(t in body for t in query_tokens if len(t) >= 2) diff --git a/monitor.py b/monitor.py index 2aa7e67..e36444e 100644 --- a/monitor.py +++ b/monitor.py @@ -2359,10 +2359,17 @@ class MessageMonitor: # 在 split 前先剥离这些行，避免 JSON 碎片污染用户消息 raw_text = self._strip_json_yaml_blocks(raw_text, platform) + # —— ⭐ 混合文本预分割：分段噪声隔离 —— + # Windows OCR 返回整屏文字的大杂烩（bot回复含
/图片URL + 用户消息）。 + # 若整体 _is_noise_text() 会因 HTML 标签把整段丢弃，用户消息跟着被杀。 + # 修复：先按消息边界分割成候选段，对各段单独做噪声检测， + # 只丢弃噪声段（bot回复），保留干净段（用户消息）。 + raw_text = self._split_and_denoise_segments(raw_text, platform) + # —— 可见时间戳新鲜性检查 —— # 若文本中有可见时间戳（HH:MM），但全部过期（>300s），视为历史消息直接丢弃 # 防止 OCR 混入旧消息内容误触发回复 - if not self._ocr_text_has_fresh_timestamp(raw_text, platform): + if not raw_text or not self._ocr_text_has_fresh_timestamp(raw_text, platform): logger.debug(f"[{platform}] OCR 历史消息(无可识别新鲜时间戳)，丢弃: {raw_text[:80]!r}") return [] @@ -3929,6 +3936,91 @@ class MessageMonitor: return True return False + def _split_and_denoise_segments(self, raw_text: str, platform: str) -> str: + """混合文本预分割 + 分段噪声隔离。 + + Windows OCR 返回整屏文字的大杂烩（bot 回复含
/图片URL + 用户消息）。 + 若整体调用 _is_noise_text() 会因 HTML 标签把整段丢弃，用户消息跟着被杀。 + + 策略： + 1. 按消息边界将文本分割成候选段 + 2. 对每个段单独做 _is_noise_text() 检查 + 3. 只保留干净的用户消息段，丢弃含噪声的 bot 回复段 + 4. 返回清理后的文本（空字符串 = 全部是噪声） + + 分割规则（按优先级）： + -
标签：bot markdown 回复的典型特征 + - 时间戳 HH:MM / HH:MM:SS：消息间天然边界 + - 图片 URL (https://img.ypshidifu.cn/...)：bot 回复中的图片引用 + - 长空白/间隔：消息间的自然分隔 + """ + if not raw_text: + return "" + + # 快速路径：文本本身不含噪声特征 → 直接原样返回，不做分割开销 + if not self._is_noise_text(raw_text) and ' 或
是 bot 回复中最强的噪声标记，优先在此处切分 + parts = re.split(r'(<\s*br\s*/?\s*>|\s*
\s*)', raw_text) + + for part in parts: + if not part or not part.strip(): + continue + part = part.strip() + + # 如果段落仍然很长且包含多种内容，尝试进一步切分 + if len(part) > 100 and ('https://' in part or re.search(r'\d{1,2}:\d{1,2}', part)): + # 按 URL 前后的问号句或时间戳进一步切分 + sub_parts = re.split( + r'(?<=。|？|！|\.)\s+(?=[^<])' + r'|(?<=[^:])\s+(?=\d{1,2}:\d{1,2})' + r'|(?<=\S)\s{2,}(?=\S)', + part + ) + for sp in sub_parts: + sp = sp.strip() + if sp: + segments.append(sp) + else: + segments.append(part) + + if not segments: + return raw_text # 分割失败，回退到原始文本 + + # ---- 对各段单独做噪声检测 ---- + clean_segments = [] + noise_count = 0 + for i, seg in enumerate(segments): + is_noise = self._is_noise_text(seg) + if is_noise: + noise_count += 1 + logger.debug(f"[{platform}] 预分割: 段[{i}] 含噪声特征，丢弃: {seg[:60]!r}") + else: + clean_segments.append(seg) + logger.debug(f"[{platform}] 预分割: 段[{i}] 干净，保留: {seg[:60]!r}") + + # ---- 结果汇总 ---- + if not clean_segments: + # 全部都是噪声 → 返回原始文本让后续流程正常丢弃 + logger.debug(f"[{platform}] 预分割: 全部{len(segments)}段均为噪声，回退原始文本") + return raw_text + + if noise_count == 0: + # 没有任何噪声段 → 原始文本本身可能就是干净的 + return raw_text + + # 有部分噪声被剥离，有部分干净段保留 + result = '\n'.join(clean_segments) + logger.info(f"[{platform}] 预分割: {len(segments)}段中剥离{noise_count}段噪声，保留{len(clean_segments)}段干净文本") + return result + @staticmethod def _is_noise_text(text: str) -> bool: """检测是否为OCR噪声文本（HTML标签、图片路径、OCR乱码等）。 @@ -4893,6 +4985,21 @@ class MessageMonitor: seen.add(ts) continue + # ⭐ 候选行包含 bot 自己的旧回复（OCR 把用户新消息 + bot 旧回复拼成一行） + # 例：用户问"PHP管理员端的默认密码"时屏上仍显示 bot 先前关于"全局配置"的回复， + # 两者被 OCR 拼成一条 → 直接拿去搜 KB 会命中 bot 旧回复（答非所问）。 + # 剥离 bot 旧回复文本，保留用户真正的提问再去走后续过滤与 KB 搜索。 + if len(ts) > 5: + stripped_bot = False + for r in replied: + if len(r) >= 6 and r in ts: + ts = re.sub(re.escape(r), ' ', ts) + stripped_bot = True + if stripped_bot: + ts = re.sub(r'\s+', ' ', ts).strip() + if not ts: + continue + # 剥离后继续走后续过滤（不 continue），让干净的用户提问正常进入 new_items # _replied_texts 优先检查（永不裁切，100% 拦住自己的回复） if ts in replied: continue diff --git a/test_kb.py b/test_kb.py new file mode 100644 index 0000000..f76daa4 --- /dev/null +++ b/test_kb.py @@ -0,0 +1,40 @@ +#!/usr/bin/env python3 +"""测试知识库搜索""" +import sys +import os +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from knowledge_base import KnowledgeBase + +def test_search(): + kb = KnowledgeBase("D:/桌面/源码/文档/docs") + kb.reload() + + if not kb.is_ready: + print("知识库未加载") + return + + print(f"知识库加载了 {len(kb.docs)} 个文档") + + # 测试查询 + queries = [ + "用户管理的部分可以查到哪些信息", + "用户管理", + "查询用户信息", + "php管理员后端", + ] + + for query in queries: + print(f"\n{'='*60}") + print(f"查询: {query}") + results = kb.search(query, top_k=3) + if results: + for i, (doc, score) in enumerate(results): + print(f" {i+1}. 标题: {doc['title']}, 匹配度: {score:.2f}") + print(f" heading: {doc['heading'][:60]}") + print(f" body片段: {doc['body'][:100].replace(chr(10), ' ')}...") + else: + print(" 无结果") + +if __name__ == "__main__": + test_search() \ No newline at end of file