"""本地知识库 — 加载 Markdown/TXT 文件，jieba 分词 + TF-IDF 搜索 核心改进：知识库按标题（## / ###）切分为独立的 section，每个 section 作为 一个独立的"文档"进行索引。搜索时直接找到最相关的 section，返回结果即为 最相关的片段，不再需要从整篇文档中二次提取。 """ import os import re import logging from pathlib import Path from typing import List, Tuple, Optional import jieba from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity import numpy as np class KnowledgeBase: """知识库引擎""" def __init__(self, root_path: str = ""): self.root_path: str = root_path self.docs: List[dict] = [] # [{path, title, content, heading, section_id}] self._vectorizer = TfidfVectorizer(tokenizer=self._tokenize, token_pattern=None, lowercase=False) self._tfidf_matrix = None self._ready = False self._heading_token_sets: List[set] = [] self._body_token_sets: List[set] = [] # 常见同义词映射（查询侧 → 文档侧） SYNONYMS = { "登陆": ["登录"], "版号": ["版本号"], "权力": ["权利"], # ⭐ 口语"下到/能下到"= 下载（OCR常见口语），映射到文档侧"下载" # 修复"在哪儿能下到那登录插件啊?"无法命中"## 4. 下载地址"section "下到": ["下载"], } @staticmethod def _tokenize(text: str) -> List[str]: """jieba 分词""" return list(jieba.cut(text)) @classmethod def _expand_query(cls, query: str) -> List[str]: """扩展查询：为每个词生成同义词变体，提高对错别字/同义词的容忍度。 返回多个候选查询字符串。 """ # 基础查询 queries = [query] # 为每个同义词替换生成变体 for src, dests in cls.SYNONYMS.items(): if src in query: for dest in dests: queries.append(query.replace(src, dest)) return queries @staticmethod def _parse_qa_segments(content: str) -> list: """解析 Q&A 格式文件为独立问答段。 格式：问题[N]：xxx\n答：xxx 返回 [{"question": str, "answer": str, "raw": str}] """ segments = [] cur_q = [] cur_a = [] state = None # None, 'q', 'a' def flush(): if not cur_q and not cur_a: return q = '\n'.join(cur_q).strip() a = '\n'.join(cur_a).strip() if q or a: segments.append({ "question": q, "answer": a, "raw": f"问题：{q}\n答：{a}", }) cur_q.clear() cur_a.clear() for line in content.split('\n'): qm = re.match(r'^问题\d*\s*[：:]\s*(.*)', line) am = re.match(r'^答\s*[：:]\s*(.*)', line) if qm: flush() state = 'q' cur_q.append(qm.group(1)) elif am: state = 'a' cur_a.append(am.group(1)) elif state == 'q': cur_q.append(line) elif state == 'a': cur_a.append(line) # else: 文件头空白/注释，忽略 flush() return segments @staticmethod def _split_into_sections(content: str, title: str) -> List[dict]: """把 Markdown 文档按 ## / ### 标题切分为多个 section。 每个 section 包含： - heading: 标题行（如 "## 可用参数"） - content: 标题 + 正文（用于展示） - body: 仅正文（用于 TF-IDF，避免标题词重复加权） 返回 section 列表。 """ sections = [] # ⭐ 保护代码块：```...``` 内的内容不参与标题切分 # 修复"### 2.2.7 菜单.txt"section被截断：JSON示例里的 "# 服务器菜单"、 # "# 使用类json格式" 注释行被误判为Markdown标题，导致： # 1) 菜单.txt section body 在 ```json 处截断，JSON配置内容丢失 # 2) "# 使用类json格式" 被误切成独立section，污染搜索结果 code_blocks = [] def _protect_code_block(m): code_blocks.append(m.group(0)) return f'\n__CODEBLOCK_{len(code_blocks) - 1}__\n' content_protected = re.sub( r'```.*?```', _protect_code_block, content, flags=re.DOTALL) def _restore_code_blocks(text): for j, cb in enumerate(code_blocks): text = text.replace(f'__CODEBLOCK_{j}__', cb) return text # 按行首的 ## 标题切分（前面可能有空白） parts = re.split(r'(?=^#{1,6}\s)', content_protected, flags=re.MULTILINE) section_id = 0 for i, part in enumerate(parts): part_stripped = part.strip() if not part_stripped or len(part_stripped) < 5: continue # 提取标题行 lines = part_stripped.split('\n') heading = lines[0].strip() body = '\n'.join(lines[1:]).strip() if len(lines) > 1 else "" # 检测body中的列表项，进行二次切分 # 列表项以 "- " 开头，每个列表项作为独立子section match_heading = None list_sections = [] if body: # ⭐ 列表项切分前先恢复代码块占位符，避免列表切分逻辑 # 把恢复后的多行代码块内容当作独立列表项切分 body_restored = _restore_code_blocks(body) # 按列表项切分 list_sections = KnowledgeBase._split_body_by_list_items(body_restored, heading) if list_sections: for ls in list_sections: sections.append({ "heading": ls["heading"], "content": _restore_code_blocks(ls["content"]), "body": _restore_code_blocks(ls["body"]), "section_id": section_id, }) section_id += 1 continue # ⭐ 列表项多为短内容被合并（_split_body_by_list_items返回[]）： # 把列表项文本并入match_heading仅用于搜索匹配（heading token集）， # 展示仍用原heading。恢复拆分时的heading匹配强度（如"- reg 注册账号" # 让"注册账号"查询命中heading），但section保持合并——TF-IDF被长文本 # 稀释，不会像单行碎片那样劫持无关查询。 # 修复"注册账号"/"怎么登录账号"被invite.md"违规处罚"section # （heading携带完整条目文本获得免费heading命中）压过的问题 item_texts = [] for _line in body_restored.split('\n'): _stripped = _line.strip() if _stripped.startswith('- ') and len(_stripped) > 2: item_texts.append(_stripped[2:].strip()) if item_texts: match_heading = heading + ' ' + ' '.join(item_texts) sections.append({ "heading": heading, "content": _restore_code_blocks(part_stripped), "body": _restore_code_blocks(body), "match_heading": match_heading, "section_id": section_id, }) section_id += 1 # 如果没有切分出多个 section，整个文档作为一段 if len(sections) <= 1: sections = [{ "heading": f"# {title}", "content": content.strip(), "body": content.strip(), "section_id": 0, }] return sections @staticmethod def _split_body_by_list_items(body: str, parent_heading: str) -> List[dict]: """将body中的列表项切分为独立子section。 列表项以 "- " 开头，每个列表项及其后续内容作为独立子section。 返回子section列表，如果列表项少于2个则返回空列表（不切分）。 ⭐ 修复：Markdown表格（以 | 开头的行）不应被切分到列表项中， 应作为独立section保留，确保表格内容可被搜索匹配。 ⭐ 修复：列表项内容过短（<10字符）时不切分，保持为完整section。 防止下载链接等短内容被切分成多个空section，导致TF-IDF得分过低。 """ lines = body.split('\n') # ⭐ 检测表格区域：找到所有表格行的范围 table_ranges = [] # [(start, end), ...] table_start = -1 for i, line in enumerate(lines): stripped = line.strip() if stripped.startswith('|') and '|' in stripped[1:]: if table_start == -1: table_start = i else: if table_start != -1: table_ranges.append((table_start, i - 1)) table_start = -1 if table_start != -1: table_ranges.append((table_start, len(lines) - 1)) # ⭐ 检测列表项是否多为短内容（指令枚举/下载链接），如果是则不切分 # ⭐ 修复(2026-08-10)：从"全部短"放宽为"≥70%短" # 原因：sdf1_login.md"## 3. 指令参数及介绍"13项中仅2项长（ # "sdf1_login radio reload 重载广播..."、"printer 打印流水数据..."）， # 原"全部短才保护"逻辑失效，导致该section被切成13个单行section， # 任何含"登录"的查询都会命中"- login 登录账号"这种孤立行并被回复。 # 放宽后（10/13≈77%≥70%），整块指令列表保留为一个section， # 回答包含完整上下文，不再返回单行碎片。 list_item_count = 0 short_items = 0 for line in lines: stripped = line.strip() if stripped.startswith('- ') and len(stripped) > 2: list_item_count += 1 # 检查列表项内容是否过短（去掉 "- " 前缀后） item_content = stripped[2:].strip() # ⭐ 修复：提高阈值到30字符，防止下载链接等短内容被切分 if len(item_content) < 30: short_items += 1 # 如果列表项大多是短内容（短项占比≥70%），不切分 if list_item_count >= 2 and short_items >= list_item_count * 0.7: return [] def is_in_table(line_idx): """检查行是否在表格区域内""" for t_start, t_end in table_ranges: if t_start <= line_idx <= t_end: return True return False list_items = [] current_item = [] current_item_start = -1 for i, line in enumerate(lines): stripped = line.strip() # ⭐ 跳过表格行，不作为列表项处理 if is_in_table(i): if current_item: # 表格开始，结束当前列表项 list_items.append({ 'start': current_item_start, 'end': i - 1, 'text': '\n'.join(current_item).strip(), 'title_line': current_item[0][2:].strip() }) current_item = [] current_item_start = -1 continue if stripped.startswith('- ') and len(stripped) > 2: # 新的列表项开始 if current_item: list_items.append({ 'start': current_item_start, 'end': i - 1, 'text': '\n'.join(current_item).strip(), 'title_line': current_item[0][2:].strip() # 去掉 "- " 前缀 }) current_item = [line] current_item_start = i elif current_item: current_item.append(line) # 处理最后一个列表项 if current_item: list_items.append({ 'start': current_item_start, 'end': len(lines) - 1, 'text': '\n'.join(current_item).strip(), 'title_line': current_item[0][2:].strip() }) # 如果列表项少于2个，不切分 if len(list_items) < 2: return [] sections = [] for item in list_items: # 提取列表项的标题（前20个字符） title_line = item['title_line'] # 去掉可能的链接和图片标记 title_clean = re.sub(r'\[([^\]]+)\]\([^\)]+\)', r'\1', title_line) title_clean = re.sub(r'!\[[^\]]*\]\([^\)]+\)', '', title_clean) title_clean = title_clean.strip() # 截取标题（最多30字符） if len(title_clean) > 30: title_clean = title_clean[:30] + '...' # 构建子section内容 sub_content = item['text'] sub_heading = f"{parent_heading} - {title_clean}" sections.append({ "heading": sub_heading, "content": sub_content, "body": sub_content, }) # ⭐ 添加独立的表格section（表格不在任何列表项中） for t_start, t_end in table_ranges: table_text = '\n'.join(lines[t_start:t_end + 1]).strip() if table_text: # 检查这个表格是否已经在某个列表项中 in_list_item = False for item in list_items: if t_start >= item['start'] and t_end <= item['end']: in_list_item = True break if not in_list_item: sections.append({ "heading": f"{parent_heading} - 表格", "content": table_text, "body": table_text, }) return sections def set_path(self, path: str): self.root_path = path self.reload() def reload(self): """重新加载知识库文件""" self.docs = [] if not self.root_path or not os.path.isdir(self.root_path): self._ready = False return root = Path(self.root_path) extensions = (".md", ".txt", ".markdown") for fp in sorted(root.rglob("*")): if fp.suffix.lower() in extensions and fp.is_file(): try: content = fp.read_text("utf-8", errors="ignore") title = fp.stem # 解析 Q&A 段 segments = self._parse_qa_segments(content) if segments: for seg in segments: self.docs.append({ "path": str(fp), "title": title, "content": seg["raw"], "answer": seg["answer"], "question": seg["question"], "heading": f"# {title}", "body": seg["raw"], }) else: # 非 Q&A 格式 → 按标题切分为多个 section sections = self._split_into_sections(content, title) for sec in sections: self.docs.append({ "path": str(fp), "title": title, "content": sec["content"], "heading": sec["heading"], "body": sec["body"], "match_heading": sec.get("match_heading"), }) except Exception: continue if not self.docs: self._ready = False return # 构建 TF-IDF（用 body 而非 content，避免标题词重复加权） # ⭐ 修复：排除body中的Markdown链接语法（如 [text](url)），链接文本是引用而非内容 corpus = [] for d in self.docs: body = d.get("body", d["content"]) # 移除链接语法 [text](url)，只保留非链接部分 body_clean = re.sub(r'\[([^\]]*)\]\([^\)]+\)', '', body) corpus.append(body_clean) try: self._tfidf_matrix = self._vectorizer.fit_transform(corpus) self._ready = True except Exception: self._ready = False return # 预计算分词集合，避免每次搜索重复 jieba self._heading_token_sets = [] self._body_token_sets = [] for doc in self.docs: # ⭐ match_heading（合并的短列表项文本）优先用于搜索匹配，展示仍用原heading heading = doc.get("match_heading") or doc.get("heading", "") body = doc.get("body", doc["content"]) # ⭐ 修复：排除heading中的Markdown链接文本（如 [sdf1_login登录插件](sdf1_login.md)） # 链接文本是引用，不是真正的内容，不应参与搜索匹配 heading_clean = re.sub(r'\[([^\]]*)\]\([^\)]+\)', r'\1', heading) # 保留链接文本用于显示，但标记为排除 # 实际上，链接文本应该被排除，因为它是引用而非内容 # 但为了兼容性，我们只排除链接语法本身，保留链接文本 # 更好的方案：heading_token_sets只包含非链接部分的token heading_no_links = re.sub(r'\[([^\]]*)\]\([^\)]+\)', '', heading) # 完全移除链接语法 self._heading_token_sets.append(set(t for t in jieba.cut(heading_no_links) if len(t) >= 2)) self._body_token_sets.append(set(t for t in jieba.cut(body) if len(t) >= 2)) @property def is_ready(self) -> bool: return self._ready and len(self.docs) > 0 @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 _build_parent_child_map(self, doc_idx_list: List[int]) -> dict: """构建 section 的 parent-child 关系映射。 根据 heading 层级（# / ## / ### / ####）推断父子关系。 返回 {child_idx: parent_idx} 映射。 """ import re parent_map = {} current_parent = {1: None, 2: None, 3: None, 4: None} for idx in doc_idx_list: if idx >= len(self.docs): continue doc = self.docs[idx] heading = doc.get("heading", "") # 提取 heading 层级 match = re.match(r'^(#{1,6})\s', heading) if not match: continue level = len(match.group(1)) # 找到当前级别及以上的 parent parent = None for l in range(level - 1, 0, -1): if current_parent.get(l) is not None: parent = current_parent[l] break parent_map[idx] = parent current_parent[level] = idx # 重置更低级别的 parent for l in range(level + 1, 5): current_parent[l] = None return parent_map def search(self, query: str, top_k: int = 3, min_score: float = 0.1) -> List[Tuple[dict, float]]: """搜索知识库，返回 (doc, score) 列表。 策略： 1. TF-IDF + 关键词命中计算每个 section 的分数 2. 按文档聚合：找到最佳匹配的文档 3. 在最佳文档内，根据查询意图找到最合适的 section 4. 补充其他文档的高分 section """ if not self.is_ready or not query.strip(): return [] query_stripped = query.strip() # 扩展查询（同义词替换） expanded_queries = self._expand_query(query_stripped) # 短文本检查 tokens = [t for t in jieba.cut(query_stripped) if len(t) >= 2] if len(query_stripped) < 2 or len(tokens) == 0: return [] # ⭐ 过滤查询停用词：通用名词（服务器、玩家、游戏等）会导致TF-IDF虚高 # 修复"服务器的公告"被player.md"运营方权利"压过index.md"近期公告"的问题 from auto_reply import AutoReply query_tokens_filtered = [t for t in tokens if t not in AutoReply._STOPWORDS] if not query_tokens_filtered: # 所有token都是停用词，回退到原始tokens query_tokens_filtered = tokens # 对每个扩展查询计算 TF-IDF，取最高分 tfidf_scores = np.zeros(len(self.docs)) for eq in expanded_queries: query_vec = self._vectorizer.transform([eq]) scores = cosine_similarity(query_vec, self._tfidf_matrix)[0] tfidf_scores = np.maximum(tfidf_scores, scores) # 关键词命中：使用过滤后的token expanded_tokens = set() for eq in expanded_queries: expanded_tokens.update(t for t in jieba.cut(eq) if len(t) >= 2 and t not in AutoReply._STOPWORDS) query_tokens_set = expanded_tokens if expanded_tokens else set(query_tokens_filtered) idf_dict = {} if hasattr(self._vectorizer, 'idf_') and self._vectorizer.idf_ is not None: feature_names = self._vectorizer.get_feature_names_out() for t in query_tokens_set: if t in feature_names: idx = list(feature_names).index(t) idf_dict[t] = self._vectorizer.idf_[idx] else: idf_dict[t] = 1.0 else: for t in query_tokens_set: idf_dict[t] = 1.0 heading_token_sets = self._heading_token_sets keyword_scores = np.zeros(len(self.docs)) for i, doc in enumerate(self.docs): heading_hits = sum(idf_dict.get(t, 1.0) for t in query_tokens_set if t in heading_token_sets[i]) body_hits = sum(idf_dict.get(t, 1.0) for t in query_tokens_set if t in self._body_token_sets[i]) keyword_scores[i] = heading_hits * 3 + body_hits combined_scores = 0.7 * tfidf_scores + 0.3 * keyword_scores # 内容质量调整：惩罚标题-only/正文过短的 section，奖励正文含查询词的 section # 避免"# Sdf1系列登录插件使用说明"这种纯标题段因为标题词匹配就排第一 # ⭐ 修复：排除heading中的Markdown链接语法（如 [text](url)）再做短语匹配 # 防止protect.md的"[sdf1_login登录插件](sdf1_login.md)"链接文本 # 误触发"登录插件"精确短语匹配，导致错误文档得分巨幅膨胀 _link_pattern = re.compile(r'\[([^\]]*)\]\([^\)]+\)') for i, doc in enumerate(self.docs): body = doc.get("body", "") body_len = len(body.strip()) heading = doc.get("heading", "") # 清理heading中的链接语法，保留链接文本（用于TF-IDF/关键词匹配） heading_clean = _link_pattern.sub(r'\1', heading) # 完整移除链接（含链接文本）（用于精确短语匹配，防止链接文字误匹配） # 如"[sdf1_login登录插件](sdf1_login.md)" → " "，防止"登录插件"链接文字触发5x加分 heading_no_link_text = _link_pattern.sub('', heading) full_text_check = heading_clean + " " + body full_text_clean = heading_no_link_text + " " + body # ⭐ 短语匹配用：移除整个链接（含文字），防止链接文字误匹配 full_text_phrase = heading_no_link_text + " " + body # 先计算正文含查询词情况 body_hits = sum(1 for t in query_tokens_set if t in self._body_token_sets[i]) # ⭐ 精确短语匹配检查（使用清理后的文本，排除链接文本） phrase_in_text = query_stripped in full_text_phrase # 单层 # 标题且无实质正文 → 大幅降权 if heading.startswith("# ") and body_len < 50: combined_scores[i] *= 0.2 elif body_len < 30 and not phrase_in_text and body_hits == 0: # 正文短但不含查询词 → 惩罚；含查询词的短section不惩罚（可能是精确答案） combined_scores[i] *= 0.5 # 正文含查询词 → 奖励（说明 section 真的在讲用户问的内容） if body_hits >= 2: combined_scores[i] *= 1.3 elif body_hits >= 1: combined_scores[i] *= 1.1 # ⭐ 下载意图加权（combined_scores阶段）：查询含"下载/下到"等口语， # heading 含"下载"的 section 加权——必须在文档聚合(doc_best)之前执行， # 否则下载section的文档分不够高，排不进 top_k 文档（rerank bonus救不了）。 # 修复"登录插件怎么下载"被skin.md"menu"（body含"下载皮肤文件"）压过的问题 # ⚠ 排除heading含"皮肤"的section（"### 1.下载皮肤文件"是皮肤教程，非插件下载）， # 否则皮肤文档会被放大到最高分，guide合并后压过真正的下载地址section if (any(w in query_stripped for w in ('下载', '下到', '哪下', '怎么下')) and '下载' in heading_clean and '皮肤' not in heading_clean): combined_scores[i] *= 4.0 # ⭐ 精确短语匹配：文档包含完整查询短语 → 巨幅奖励 # 防止"php端的管理员密码"匹配到不含"php"的云背包文档 # ⭐ 使用full_text_phrase（排除链接文字），防止链接文字如"sdf1_login登录插件"误匹配 if query_stripped in full_text_phrase: combined_scores[i] *= 5.0 elif len(query_stripped) >= 4: # 2字以上子串匹配：如"php端"、"管理员密码" sub_hits = 0 for qlen in range(min(len(query_stripped), 6), 3, -1): for start in range(len(query_stripped) - qlen + 1): phrase = query_stripped[start:start+qlen] if phrase in full_text_phrase: sub_hits += 1 if sub_hits >= 2: combined_scores[i] *= 2.5 elif sub_hits == 1: combined_scores[i] *= 1.5 # ⭐ 核心名词短语匹配：如"登录插件"、"资源包"等 # 防止"登录插件"被拆分为"登录"+"插件"分别匹配，导致错误文档得分更高 core_phrases = ['登录插件', '资源包', '下载地址', '配置文件'] for phrase in core_phrases: if phrase in query_stripped and phrase in full_text_phrase: combined_scores[i] *= 2.0 break # ⭐ 否定式查询标记：查询问"是否X"时，在reranking阶段惩罚不含X的section # 不在此处penalty——此处penalty会被后续loop的bonus覆盖 # ⭐ 核心名词匹配奖励：查询中的关键名词在 section 中出现 → 巨幅奖励 # 防止"PHP管理员端的默认密码"匹配到不含"密码"的"全局配置" section core_nouns = {'密码', '用户管理', '领地', '商品', 'CDK', '工单', '债券', '在线', '公告', '更新'} query_nouns = set(jieba.cut(query_stripped)) & core_nouns if query_nouns: # 查询包含核心名词，检查 section 是否包含这些名词 noun_hits = sum(1 for n in query_nouns if n in full_text_check) if noun_hits >= 1: combined_scores[i] *= 2.0 # 包含核心名词 → 奖励 else: combined_scores[i] *= 0.3 # 不包含核心名词 → 大幅惩罚 # ⭐ 表格内容奖励：当查询关于"支持/操作/方法/方式"时，与查询相关的表格 section 巨幅加分 # 防止"领地保护支持哪些操作"匹配到不含表格的"我的领地" section # ⚠ 修复：只奖励与查询语义相关的表格，避免"操作方式"表覆盖"权限级别"表 table_query_tokens = {'支持', '操作', '方法', '方式', '功能', '可以', '能够'} if table_query_tokens & query_tokens_set: body_stripped = body.strip() if body_stripped.startswith('|') and '|' in body_stripped[1:]: # ⚠ 导航/目录类表格不奖励：index.md"文档导航"body含 # "领地系统"等全部主题词，命中查询词即被5x放大到爆炸分， # 压过protect.md真实的"操作方式"表格。 # 此类 section heading 含导航/目录/关于/导航关键词，或title=index _nav_heading = heading _is_nav = ( any(k in _nav_heading for k in ('导航', '目录', '关于我们', '欢迎', '总览')) or doc.get("title", "").lower() == 'index' ) # 检查heading是否与查询语义相关 h_tokens = set(t for t in jieba.cut(heading) if len(t) >= 2) heading_relevant = bool(query_tokens_set & h_tokens) # 或body中含查询核心名词（收紧：必须是语义性名词，防"系统"等泛化词误触发） _tbl_core = {'操作', '领地', '保护', '支持', '方法', '方式', '模式', '配置', '下载'} body_core_hits = sum(1 for n in _tbl_core if n in body_stripped) body_relevant = body_core_hits >= 2 or any( n in body_stripped for n in query_tokens_set if n in _tbl_core) if (heading_relevant or body_relevant) and not _is_nav: combined_scores[i] *= 5.0 # 相关表格 → 巨幅奖励 # 短查询惩罚：分词数 ≤2 的短查询（如图标残片/短词）得分大幅降权 # 防止 OCR 碎片中的高 IDF 词（如"下载"）获得超过真实用户消息的分数 n_tokens = len(tokens) if n_tokens <= 2: combined_scores *= 0.3 elif n_tokens <= 3: combined_scores *= 0.6 # 按文档聚合：取各文档下的最高 section 分作为文档分 # 同时统计匹配 section 数量，用于后续排序（section 多的文档更可能是主文档） doc_best = {} doc_count = {} for i, doc in enumerate(self.docs): t = doc["title"] score = combined_scores[i] if score >= min_score: doc_best[t] = max(doc_best.get(t, 0), score) # 只统计强匹配 section（≥2×min_score），弱匹配不计入 bonus if score >= min_score * 2: doc_count[t] = doc_count.get(t, 0) + 1 else: doc_best[t] = max(doc_best.get(t, 0), score) if not doc_best: return [] # 文档级排序：主分优先，doc_count 仅做 tiebreaker（section 数更多→多角度匹配） # 弱匹配（score < min_score*2）不计入 doc_count，防低分堆叠 # ⭐ 文档标题语义匹配奖励：查询词与文档标题语义相关 → 巨幅奖励 # 如"php管理员"应优先返回phpadmin文档，"领地"应优先返回protect文档 # 改为 token 级匹配（而非字面子串），OCR 丢失"php"只剩"管理员端默认"也能命中 phpadmin DOC_TOPIC_TOKENS = { 'php': {'php', '后台', '玩家端', '登录模式'}, 'phpadmin': {'php', 'php管理员', '后端'}, 'protect': {'领地', 'protect', '保护'}, 'login': {'登录', 'login', '账号', 'sdf1'}, 'sdf1_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_lc: doc_best[doc_title] *= 5.0 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)) # ⭐ 核心名词文档匹配：查询包含"登录插件"时，优先返回sdf1_login文档 # 防止protect文档因包含"登录插件"链接文本而误排第一 if '登录插件' in query_stripped: if 'sdf1_login' in doc_best: doc_best['sdf1_login'] *= 5.0 # 优先返回登录插件文档 if 'login' in doc_best: doc_best['login'] *= 5.0 if 'protect' in doc_best: doc_best['protect'] *= 0.2 # 大幅降权protect文档（含链接文本） # ⭐ php登录模式查询：查询包含"php"+"登录模式"时，优先返回php文档 # 防止login/sdf1_login文档因含"登录"而压过真正含答案的php文档 if 'php' in query_token_set2 and '登录模式' in query_stripped: if 'php' in doc_best: doc_best['php'] *= 10.0 # 巨幅优先php文档 if 'login' in doc_best: doc_best['login'] *= 0.3 # 大幅降权login文档 if 'sdf1_login' in doc_best: doc_best['sdf1_login'] *= 0.3 # 大幅降权sdf1_login文档 if 'protect' in doc_best: doc_best['protect'] *= 0.1 # 大幅降权protect文档（含链接文本） # ⭐ 管理员 + 默认 组合 → 强偏好 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)), reverse=True ) results = [] # 查询意图检测 query_lower = {t.lower() for t in query_tokens_set} action_words = {"怎么", "如何", "用法", "教程", "操作"} def_words = {"版本", "版号", "版本号", "定义", "区分"} has_action = bool(action_words & query_lower) has_def = bool(def_words & query_lower) def_headings = {"版本号", "版本", "区分", "定义"} # ⭐ 下载意图：查询明确要下载/获取插件时，下载section就是答案。 # 后续的has_action优先级重排（"安装说明"含"安装"被抬到最前）和guide合并 # （numbered heading "## 2."/"## 4." 误判为教程步骤）都会覆盖下载section， # 必须整体跳过。修复"登录插件怎么下载"返回"## 2.安装说明"的问题 download_intent = any(w in query_stripped for w in ('下载', '下到', '哪下', '怎么下')) # 动作查询：提取动词+对象，对不含两者的文档降权 # 防止"圈地工具"匹配到含"工具"但不含"圈地"的文档（如寻宝插件） 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 = [] for doc_title, doc_score in sorted_docs: has_def_section = False for i, doc in enumerate(self.docs): if doc["title"] == doc_title and combined_scores[i] >= min_score / 2: if def_headings & self._heading_token_sets[i]: has_def_section = True break if not has_def_section: doc_score *= 0.3 # 没有定义类 section 的文档大幅降权 adjusted.append((doc_title, doc_score)) # 重新排序 adjusted.sort(key=lambda x: x[1], reverse=True) sorted_docs = adjusted # 从最佳文档开始，逐个文档取 section for doc_title, doc_score in sorted_docs: if len(results) >= top_k: break # 收集该文档的所有 section doc_sections = [] for i, doc in enumerate(self.docs): if doc["title"] == doc_title and combined_scores[i] >= min_score: doc_sections.append((i, doc, combined_scores[i])) doc_sections.sort(key=lambda x: x[2], reverse=True) # ⭐ 表格兜底策略：当最佳文档包含与查询相关的操作/支持表格时，强制加入并给超高分 # 防止"领地保护"因表格TF-IDF=0而丢失正确答案（表格是答案但不含查询词） # ⚠ 修复：只优先与查询语义相关的表格，避免"操作方式"表覆盖"权限级别"表 if len(doc_sections) > 0: # 检查文档是否已有与查询相关的表格section has_relevant_table = False for idx, _, _ in doc_sections: body = self.docs[idx].get("body", "").strip() heading = self.docs[idx].get("heading", "") if body.startswith('|') and ('操作' in body or '支持' in body): h_tokens = set(t for t in jieba.cut(heading) if len(t) >= 2) if query_tokens_set & h_tokens: has_relevant_table = True break if not has_relevant_table: # 强制加入与查询相关的操作/支持表格section for i, doc in enumerate(self.docs): if doc["title"] == doc_title: body = doc.get("body", "").strip() heading = doc.get("heading", "") if body.startswith('|') and '|' in body[1:] and ('操作' in body or '支持' in body): # 检查heading是否与查询语义相关 h_tokens = set(t for t in jieba.cut(heading) if len(t) >= 2) heading_relevant = bool(query_tokens_set & h_tokens) # ⚠ body相关：必须含查询核心名词（领地/操作/商品等）而非任意token # 修复"领地系统有哪两种操作模式?"误把sdf1_login"2.安装说明"表格 # （body含"系统"一词）当成"操作方式"答案，被max_score*2放大到89分 _core_nouns_tbl = {'操作', '领地', '保护', '支持', '方法', '方式', '模式'} body_core_hits = sum(1 for n in _core_nouns_tbl if n in body) body_relevant = body_core_hits >= 2 or any( n in body for n in query_tokens_set if n in _core_nouns_tbl) if heading_relevant or body_relevant: # 给表格section超高分（最高分×2），确保reranking后仍排第一 max_score = max(s for _, _, s in doc_sections) if doc_sections else min_score _tbl_score = max_score * 2 # ⭐ 答案级表格：查询意图是"操作/模式/方式/方法"，表格body含 # 操作方式/模式列 → 抬到30分。protect.md"操作方式|是否支持" # 表格是"领地系统有哪两种操作模式?"直接答案，但max_score*2 # 只有~10分，压不过php后台玩家端（body含"领地管理"+核心名词 # bonus 也得~27分）。答案级表格必须比误匹配section更高。 if any(w in query_stripped for w in ('操作', '模式', '方式', '方法')): if any(n in body for n in ('操作', '方式', '模式')): _tbl_score = max(_tbl_score, 30.0) doc_sections.append((i, doc, _tbl_score)) break doc_sections.sort(key=lambda x: x[2], reverse=True) else: # ⭐ 表格已在doc_sections中：has_relevant_table为True时不走强制加入， # 表格保持原分可能过低（protect"操作方式"表10.25压不过php后台玩家端27.29）。 # 答案级表格抬到30分，逻辑与强制加入分支保持一致。 for _ti, (_tidx, _tdoc, _tscore) in enumerate(doc_sections): _tbody = _tdoc.get("body", "").strip() if _tbody.startswith('|') and ('操作' in _tbody or '支持' in _tbody): if any(w in query_stripped for w in ('操作', '模式', '方式', '方法')): if any(n in _tbody for n in ('操作', '方式', '模式')): doc_sections[_ti] = (_tidx, _tdoc, max(_tscore, 30.0)) break if not doc_sections: continue # 同文档内，根据查询意图优先选合适的 section # 1. heading 包含查询关键词 → bonus；文档头部（仅含文档名）→ penalty query_tokens = set(t for t in jieba.cut(query_stripped) if len(t) >= 2) reranked = [] for idx, doc, score in doc_sections: heading = doc.get("heading", "") body = doc.get("body", doc.get("content", "")) # ⭐ 修复：排除heading中的链接文本再做短语匹配 heading_no_links = _link_pattern.sub('', heading) # 使用heading_no_links计算token，防止链接文字如"sdf1_login登录插件"误匹配 h_tokens = set(t for t in jieba.cut(heading_no_links) if len(t) >= 2) # heading 包含查询关键词 → 适度 bonus（2.0 每命中词） heading_hits = len(query_tokens & h_tokens) bonus = heading_hits * 2.0 # heading 包含查询连续短语 → 额外 bonus（但降低权重，防止链接文本误匹配） # ⭐ 修复：降低heading短语匹配bonus（6.0→2.0），防止protect.md的"登录插件"链接文本 # 压过sdf1_login.md的真实下载section；使用heading_no_links排除链接文本 for qlen in range(min(len(query_stripped), 8), 3, -1): for start in range(len(query_stripped) - qlen + 1): phrase = query_stripped[start:start+qlen] if len(phrase) >= 4 and phrase in heading_no_links: bonus += 2.0 break # body 包含查询连续短语 → 强 bonus（但降低权重，防止链接文本/引用文本误匹配） # ⭐ 修复：降低body短语匹配bonus（10.0→3.0），防止protect.md的"插件下载: 见github..." # 压过sdf1_login.md的真实下载section for qlen in range(min(len(query_stripped), 8), 3, -1): for start in range(len(query_stripped) - qlen + 1): phrase = query_stripped[start:start+qlen] if len(phrase) >= 4 and phrase in body: bonus += 3.0 break # ⭐ 下载意图加权：查询含"下载/下到"等口语，section heading 含"下载" → 高bonus # 修复"在哪儿能下到那登录插件啊?"被"插件设置"(body含"登录")压过， # 而真正的"## 4. 下载地址"section(3.93)排不进top3的问题 # ⚠ 排除heading含"皮肤"的section（皮肤教程非插件下载） # ⚠ 只检查真正的标题行：如果heading含
或列表标记，说明是复合标题， # 需检查"下载"是否出现在标题主体部分（
之前），而非列表项内容 _download_match = False if any(w in query_stripped for w in ('下载', '下到', '哪下', '怎么下')): if '下载' in heading_no_links and '皮肤' not in heading_no_links: # 检查是否是纯标题（不含
或列表标记） if '
' not in heading and '- [' not in heading and '|-' not in heading: _download_match = True else: # 复合标题：检查标题主体部分（
或列表之前）是否含"下载" _title_part = re.split(r'
|- |\|-', heading_no_links)[0] if '下载' in _title_part: _download_match = True if _download_match: bonus += 8.0 # ⭐ 配置文件section加权：查询含"配置/设置/怎么/如何"时， # heading 带 .txt/.json/.yml 后缀的 section 就是配置文件格式说明， # 是"XX如何配置"的直接答案。修复"登录插件的菜单如何配置?" # 时"### 2.2.7 菜单.txt"(JSON配置格式)7.89分被"自定义菜单物品" # 碎片和"指令参数及介绍"压过、sub_k截断丢弃的问题。 # ⚠ 条件三重：配置意图词 + 配置文件后缀 + heading命中查询的 # 非泛化词（排除登录/配置/设置/使用/插件——这些词到处出现无定位力）。 # "怎么登录账号"的"2.2.4 插件设置.txt"交集仅{登录}∈generic → 不触发 if any(w in query_stripped for w in ('配置', '设置', '怎么', '如何')): if re.search(r'\.(txt|json|yml|yaml|properties|cfg|conf)', heading_no_links): _cfg_h_tokens = set(t for t in jieba.cut(heading_no_links) if len(t) >= 2) _cfg_hit = query_tokens & _cfg_h_tokens _generic_cfg = {'登录', '配置', '设置', '使用', '插件'} if any(t not in _generic_cfg for t in _cfg_hit): bonus += 8.0 # ⭐ body 包含查询核心名词（如"密码"）→ 巨幅 bonus # ⚠ 按名词出现次数计分（1次+5，上限15），而非出现即+15整分： # 修复php后台玩家端body仅含"领地管理"一句（领地×2）就+15分整， # 压过protect.md真正含"操作方式"表格的答案section core_nouns = {'密码', '用户管理', '领地', '商品', 'CDK', '工单', '债券', '在线', '公告', '更新'} _noun_total = sum(body.count(n) for n in core_nouns if n in body and n in query_stripped) if _noun_total >= 1: bonus += min(15.0, 5.0 * _noun_total) # 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) if heading.startswith("# ") and (not body or len(body) < 50) and not body_has_query: bonus -= 15.0 # ⭐ 导航/目录类section全局惩罚：文档导航/目录/总览表格列出全部主题词， # TF-IDF+核心名词bonus会让它们得分虚高（如index"文档导航"对"皮肤更新" # 查询得28分压过skin文档）。导航section永远不是具体问题的答案 → 重罚。 # ⚠ "关于我们"除外：它是查询意图主体（index"关于我们"=服务器简介即为 # 精确答案），"关于我们"查询曾被该项误判-30打入-15.71负分丢弃。 _nav_penalty_kw = ('导航', '目录', '欢迎', '总览', '文档导航') _q_is_aboutus = '关于我们' in query_stripped and '关于我们' in heading_no_links if any(k in heading_no_links for k in _nav_penalty_kw) and not _q_is_aboutus: bonus -= 30.0 reranked.append((idx, doc, score + bonus)) # ⭐ 否定式查询惩罚（在reranking后执行，防止被bonus覆盖） _q_nopunct = re.sub(r'[？?！!。，,\s]', '', query_stripped) if re.search(r'是强制的还是可选|是否.*可选|强制.*可选|可选.*强制', _q_nopunct): _neg_core = None for phrase in ['资源包', '登录插件', 'CDK', '工单']: if phrase in _q_nopunct: _neg_core = phrase break if _neg_core: for j, (idx, doc, final_score) in enumerate(reranked): sec_text = doc.get("heading", "") + " " + (doc.get("body", "") or doc.get("content", "")) if _neg_core not in sec_text: reranked[j] = (idx, doc, final_score * 0.3) reranked.sort(key=lambda x: x[2], reverse=True) doc_sections = reranked # ⭐ Parent-child 精度提升：当 parent section body 是"空壳"（表格/短文本）， # 优先返回 child sections（更具体的子内容） # 如 "## 2.安装说明：" body是前置插件表格，应优先返回 "### 2.1 下载" 子sections if len(doc_sections) > 1: # 构建当前文档的 parent-child 关系（从全部docs中） all_doc_indices = list(range(len(self.docs))) parent_map = self._build_parent_child_map(all_doc_indices) # 检查是否有"空壳"parent section for i, (idx, doc, score) in enumerate(doc_sections): heading = doc.get("heading", "") body = doc.get("body", "") body_len = len(body.strip()) h_tokens = set(t for t in jieba.cut(heading) if len(t) >= 2) heading_has_query = bool(query_tokens & h_tokens) is_table = body.strip().startswith('|') and '|' in body[1:] # 条件1: parent heading含查询词 + body是表格 # 条件2: parent body是表格 + 任何child heading含查询词（即使parent不匹配） children = [c_idx for c_idx, p_idx in parent_map.items() if p_idx == idx] child_has_query_heading = False if children: for child_idx in children: child_h = self.docs[child_idx].get("heading", "") child_h_tokens = set(t for t in jieba.cut(child_h) if len(t) >= 2) if query_tokens & child_h_tokens: child_has_query_heading = True break if (heading_has_query and is_table) or (is_table and child_has_query_heading): if children: # 找到有实质性body的children good_children = [] for child_idx in children: child_doc = self.docs[child_idx] child_body = child_doc.get("body", "") child_body_len = len(child_body.strip()) # child有非空body且不是表格 → 候选 if child_body_len > 5 and not child_body.strip().startswith('|'): good_children.append((child_idx, child_doc)) if good_children: # 给parent减分（body是空壳表格） doc_sections[i] = (idx, doc, score * 0.3) # 将good children加入doc_sections（如果不在的话） existing_idx_set = {s[0] for s in doc_sections} for child_idx, child_doc in good_children: if child_idx not in existing_idx_set: # 计算child的加分：heading含查询词 → 额外加分 child_heading = child_doc.get("heading", "") child_h_tokens = set(t for t in jieba.cut(child_heading) if len(t) >= 2) child_heading_bonus = len(query_tokens & child_h_tokens) * 5.0 # child body含查询词 → 额外加分 child_body = child_doc.get("body", "") child_body_bonus = 10.0 if any(t in child_body for t in query_tokens if len(t) >= 2) else 0.0 child_score = score * 0.8 + child_heading_bonus + child_body_bonus doc_sections.append((child_idx, child_doc, child_score)) # 重新排序 doc_sections.sort(key=lambda x: x[2], reverse=True) break # ⭐ 表格优先策略（reranking后）：条件优先返回与查询语义相关的表格section # 防止reranking后表格被其他section超越（表格是答案但不含查询词） # ⚠ 修复：不能盲目优先含"操作"的表格——"操作方式"表（交互模式）会覆盖"权限级别"表 # 只有当表格section的heading与查询关键词有交集时才优先 if len(doc_sections) > 1: for i in range(len(doc_sections)): idx = doc_sections[i][0] body = self.docs[idx].get("body", "").strip() heading = self.docs[idx].get("heading", "") if body.startswith('|') and ('操作' in body or '支持' in body): # 检查heading是否与查询语义相关（heading含查询词 或 查询词含heading关键词） h_tokens = set(t for t in jieba.cut(heading) if len(t) >= 2) heading_relevant = bool(query_tokens & h_tokens) # ⚠ body相关收紧：必须含查询核心名词（操作/领地等）而非任意token # 修复"领地系统有哪两种操作模式?"在rerank后仍把sdf1_login # "2.安装说明"表格（body含"系统"）移到首位压过protect"操作方式"表 _core_nouns_tbl2 = {'操作', '领地', '保护', '支持', '方法', '方式', '模式'} body_core_hits = sum(1 for n in _core_nouns_tbl2 if n in body) body_relevant = body_core_hits >= 2 or any( n in body for n in query_tokens if n in _core_nouns_tbl2) if heading_relevant or body_relevant: doc_sections.insert(0, doc_sections.pop(i)) break # ⭐ 下载意图优先：查询包含"下载"时，优先返回heading含"下载"的section # 防止"在哪儿能下载到登录插件？"返回login指令而非下载链接 # ⚠ 排除heading含"皮肤"的section（"### 1.下载皮肤文件"是皮肤教程， # 会被误移到首位并导致guide合并返回"## menu"压过插件下载地址） if len(doc_sections) > 1 and '下载' in query_stripped: for i in range(len(doc_sections)): idx = doc_sections[i][0] heading = self.docs[idx].get("heading", "") _link_pattern = re.compile(r'\[([^\]]*)\]\([^\)]+\)') heading_no_links = _link_pattern.sub('', heading) if '下载' in heading_no_links and '皮肤' not in heading_no_links: doc_sections.insert(0, doc_sections.pop(i)) break # ⭐ php登录模式意图优先：查询包含"php"+"登录模式"时，优先返回php文档的section # 防止sdf1_login的login指令压过真正含答案的php后台玩家端section if len(doc_sections) > 1 and 'php' in query_token_set2 and '登录模式' in query_stripped: for i in range(len(doc_sections)): idx = doc_sections[i][0] doc = self.docs[idx] if doc.get("title") == "php": doc_sections.insert(0, doc_sections.pop(i)) break # ⭐ 下载意图查询跳过has_action重排：下载section已被下载优先级移到最前， # 重排会把"安装说明"（heading含"安装"）重新抬到首位，覆盖下载答案 if has_action and not download_intent and len(doc_sections) > 1: # 动作查询：提取动词+对象，精准匹配 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] 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 best_idx = -1 best_count = 0 for i in range(len(doc_sections)): idx = doc_sections[i][0] match_count = len(def_headings & self._heading_token_sets[idx]) if match_count > best_count: best_count = match_count best_idx = i if best_idx > 0: doc_sections.insert(0, doc_sections.pop(best_idx)) # 从每个文档取前 1-2 个 section # ⭐ 修复：当查询是操作指南（如何/怎么/教程）时，合并同一文档同一 guide 的所有相关 section # 避免"如何通过官网更新皮肤"只返回最后一个步骤 # ⭐ 避免混合不同 guide 的内容（如 url 方式和正版玩家方式混合） # ⭐ 下载意图查询跳过guide合并：numbered heading（"## 1.xxx"/"## 4.下载地址"） # 会被step_headings误判为教程步骤，guide合并选中"安装说明"guide覆盖下载答案 is_guide_query = any(w in query_stripped for w in ['如何', '怎么', '教程', '步骤', '操作', '指南']) if is_guide_query and not download_intent and len(doc_sections) > 1: # 检查是否为操作指南类文档（多个步骤 section） step_headings = {'1.', '2.', '3.', '4.', '5.', '步骤', '第一', '第二', '第三', '第四', '第五'} # ⭐ 只统计"### "级子标题作为教程步骤（如skin.md的"### 1.下载皮肤文件"）。 # 不统计"## 2."/"## 4."章节编号——sdf1_login.md的"## 3. 指令参数及介绍" # 等章节被误判为步骤，导致"怎么登录账号"触发guide合并，吞掉正确section # 返回"## 2.安装说明"guide。教程步骤一定是3级标题下的子节。 step_count = sum(1 for _, doc, _ in doc_sections if doc.get('heading', '').startswith('### ') and any(s in doc.get('heading', '') for s in step_headings)) if step_count >= 3: # ⭐ 语义命中守卫：若存在 section 的 heading 直接命中查询核心词 # （如"菜单.txt"命中"菜单"），说明查询有明确的section级答案， # 跳过guide合并，避免"登录插件的菜单如何配置?"被合并成"2.安装说明" # guide而吞掉真正的"### 2.2.7 菜单.txt"配置段。 # ⚠ 条件：heading token 命中 且 body 非空，防止纯目录section误触发 # ⚠ 记录分数最高的命中section（而非第一个），确保"菜单.txt"这类 # 高combine分的精确答案胜出，而不是被同文档先出现的低分section抢占 _heading_hit_section = None # (doc, score) _heading_hit_idx = None _heading_hit_score = -1.0 for s_idx, s_doc, s_score in doc_sections: s_heading = s_doc.get('heading', '') s_body = s_doc.get('body', '') s_h_tokens = set(t for t in jieba.cut(s_heading) if len(t) >= 2) _hit = query_tokens_set & s_h_tokens # 排除查询侧泛化词：登录/配置/设置/使用/插件 到处出现， # 命中它们无法定位答案section；真正能定位的是 # 如"菜单""皮肤"这类具体名词 _generic = {'登录', '配置', '设置', '使用', '插件'} _strong_hit = any(t not in _generic for t in _hit) if _hit and _strong_hit and len(s_body.strip()) > 10: if s_score > _heading_hit_score: _heading_hit_section = (s_doc, s_score) _heading_hit_idx = s_idx _heading_hit_score = s_score if _heading_hit_section is not None: logger = logging.getLogger("msgbot.kb") logger.debug( f"guide合并跳过: heading命中查询词 " f"{_heading_hit_section[0].get('heading', '')[:30]!r} " f"score={_heading_hit_section[1]:.2f}" ) # ⭐ 命中section提到doc_sections首位： # has_action重排可能已把"指令参数及介绍"等priority命中section抬到最前， # 若此处不提升，sub_k=2截断会丢弃真正的答案section # （"登录插件的菜单如何配置?"中菜单.txt 23.08分排第三被丢） for _ii, (_s_idx2, _, _) in enumerate(doc_sections): if _s_idx2 == _heading_hit_idx: doc_sections.insert(0, doc_sections.pop(_ii)) break else: # ⭐ 获取最佳文档的路径，查找该文档的所有 sections best_doc_path = doc_sections[0][1].get('path', '') if best_doc_path: # 查找同一文档的所有 sections all_doc_sections = [(i, d) for i, d in enumerate(self.docs) if d.get('path') == best_doc_path] if len(all_doc_sections) > 3: # 有多个 sections，可能是操作指南 # ⭐ 按 parent guide 分组 sections # 遇到 # 或 ## 级别标题时，视为新 guide 的开始 guides = [] # [(parent_heading, [(idx, doc), ...]), ...] current_guide = None current_sections = [] for idx, doc in all_doc_sections: heading = doc.get('heading', '') # 判断是否为 guide 级别标题（# 或 ##） if heading.startswith('# ') or heading.startswith('## '): # 保存前一个 guide if current_guide is not None and current_sections: guides.append((current_guide, current_sections)) current_guide = heading current_sections = [(idx, doc)] else: # 属于当前 guide 的子 section（### 及以下） if current_guide is not None: current_sections.append((idx, doc)) else: # 没有 guide 级别标题，作为独立 section if current_sections: guides.append((None, current_sections)) current_sections = [(idx, doc)] current_guide = None # 保存最后一个 guide if current_guide is not None and current_sections: guides.append((current_guide, current_sections)) elif current_sections: guides.append((None, current_sections)) # 选择最相关的 guide 进行合并 if len(guides) > 1: # ⭐ 根据查询关键词选择 guide # 查询含 "url"/"图床"/"直连"/"链接" → 选择非正版 guide（含 url 操作方式） # 查询含 "正版"/"官网"/"启动器"/"更新" → 选择正版 guide # 默认选择包含最多步骤的 guide url_keywords = {'url', '图床', '直连', '链接', '非正版'} genuine_keywords = {'正版', '官网', '启动器', 'minecraft.net'} best_guide_idx = 0 best_guide_score = -1 for i, (parent, sections) in enumerate(guides): if not sections: continue # 计算 guide 与查询的相关性 guide_text = ' '.join(doc.get('heading', '') + ' ' + doc.get('body', '') for _, doc in sections) score = 0 for kw in url_keywords: if kw in query_stripped and kw in guide_text: score += 10 for kw in genuine_keywords: if kw in query_stripped and kw in guide_text: score += 10 # 优先选择步骤最多的 guide（默认行为） # ⭐ 与step_count一致：只统计"### "级子标题步骤 step_in_guide = sum(1 for _, doc in sections if doc.get('heading', '').startswith('### ') and any(s in doc.get('heading', '') for s in step_headings)) score += step_in_guide * 2 if score > best_guide_score: best_guide_score = score best_guide_idx = i selected_guide = guides[best_guide_idx][1] else: # 只有一个 guide，直接使用 selected_guide = guides[0][1] if guides else [] # 合并选中 guide 的所有步骤 section if selected_guide: merged_body = "" for idx, doc in selected_guide: heading = doc.get('heading', '') body = doc.get('body', doc.get('content', '')) if body and len(body) > 10: merged_body += f"\n{heading}\n{body}\n" if merged_body: # 创建合并后的文档 merged_doc = selected_guide[0][1].copy() merged_doc['body'] = merged_body.strip() merged_doc['content'] = merged_body.strip() # 删除 answer 字段，使用 body if 'answer' in merged_doc: del merged_doc['answer'] results.append((merged_doc, float(doc_sections[0][2]))) continue # 默认：从每个文档取前 1-2 个 section sub_k = 2 if len(results) < top_k - 1 else top_k - len(results) for i, (idx, doc, score) in enumerate(doc_sections): if i >= sub_k: break # ⭐ php登录模式查询：php文档的section额外加分，确保排在sdf1_login之前 final_score = float(score) if 'php' in query_token_set2 and '登录模式' in query_stripped: if doc.get("title") == "php": final_score *= 10.0 # php文档section巨幅加分 results.append((doc, final_score)) # ⭐ 最终排序：按分数降序，确保最高分的结果排第一 results.sort(key=lambda x: x[1], reverse=True) return results[:top_k] if results: return results # Fallback: 子串匹配（查询包含在文档标题或内容中） q = query_stripped if len(q) >= 4: for doc in self.docs: if q in doc["content"] or q in doc["title"]: logger = logging.getLogger("msgbot.kb") logger.debug(f"子串匹配: \"{q}\" → {doc['title']}") return [(doc, min_score)] return results def get_doc_count(self) -> int: return len(self.docs)