youpaishidifu
87c0bd9f62
Bug1: 提问豁免（_is_question_message）让继承极老时间戳的消息绕过过滤，
如22:07的消息在次日10:43仍通过stage splitting放行。
修复：给提问豁免加24h硬上限，继承时间戳>24h不再豁免。
Bug2: protect.md heading含[sdf1_login登录插件](...)链接，
heading_no_link_text保留链接文字，导致phrase_in_text匹配触发5x加分。
修复：新增full_text_phrase（移除整个链接含文字），
所有短语匹配改为用full_text_phrase，链接文字不再参与评分。
1107 lines
56 KiB
Python
1107 lines
56 KiB
Python
"""本地知识库 — 加载 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 = {
|
||
"登陆": ["登录"],
|
||
"版号": ["版本号"],
|
||
"权力": ["权利"],
|
||
}
|
||
|
||
@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 = []
|
||
# 按行首的 ## 标题切分（前面可能有空白）
|
||
parts = re.split(r'(?=^#{1,6}\s)', content, 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
|
||
if body:
|
||
# 按列表项切分
|
||
list_sections = KnowledgeBase._split_body_by_list_items(body, heading)
|
||
if list_sections:
|
||
for ls in list_sections:
|
||
sections.append({
|
||
"heading": ls["heading"],
|
||
"content": ls["content"],
|
||
"body": ls["body"],
|
||
"section_id": section_id,
|
||
})
|
||
section_id += 1
|
||
continue
|
||
|
||
sections.append({
|
||
"heading": heading,
|
||
"content": part_stripped,
|
||
"body": body,
|
||
"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))
|
||
|
||
# ⭐ 检测列表项是否都是短内容（如下载链接），如果是则不切分
|
||
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
|
||
# 如果列表项都是短内容（如下载链接），不切分
|
||
if list_item_count >= 2 and short_items == list_item_count:
|
||
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"],
|
||
})
|
||
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:
|
||
heading = 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，取最高分
|
||
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)
|
||
|
||
# 关键词命中
|
||
expanded_tokens = set()
|
||
for eq in expanded_queries:
|
||
expanded_tokens.update(t for t in jieba.cut(eq) if len(t) >= 2)
|
||
query_tokens_set = expanded_tokens
|
||
|
||
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
|
||
# ⭐ 精确短语匹配：文档包含完整查询短语 → 巨幅奖励
|
||
# 防止"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)
|
||
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:]:
|
||
# 检查heading是否与查询语义相关
|
||
h_tokens = set(t for t in jieba.cut(heading) if len(t) >= 2)
|
||
heading_relevant = bool(query_tokens_set & h_tokens)
|
||
# 或body中含查询核心名词
|
||
body_relevant = any(n in body_stripped for n in query_tokens_set if len(n) >= 2)
|
||
if heading_relevant or body_relevant:
|
||
combined_scores[i] *= 3.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 = {
|
||
'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'] *= 3.0 # 优先返回登录插件文档
|
||
if 'login' in doc_best:
|
||
doc_best['login'] *= 3.0
|
||
if 'protect' in doc_best:
|
||
doc_best['protect'] *= 0.5 # 降权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 = {"版本号", "版本", "区分", "定义"}
|
||
|
||
# 动作查询：提取动词+对象，对不含两者的文档降权
|
||
# 防止"圈地工具"匹配到含"工具"但不含"圈地"的文档（如寻宝插件）
|
||
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_relevant = any(n in body for n in query_tokens_set if len(n) >= 2)
|
||
if heading_relevant or body_relevant:
|
||
# 给表格section超高分（最高分×2），确保reranking后仍排第一
|
||
max_score = max(s for _, _, s in doc_sections) if doc_sections else min_score
|
||
doc_sections.append((i, doc, max_score * 2))
|
||
break
|
||
doc_sections.sort(key=lambda x: x[2], reverse=True)
|
||
|
||
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)
|
||
h_tokens = set(t for t in jieba.cut(heading) 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
|
||
# ⭐ body 包含查询核心名词（如"密码"）→ 巨幅 bonus
|
||
core_nouns = {'密码', '用户管理', '领地', '商品', 'CDK', '工单', '债券', '在线'}
|
||
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)
|
||
if heading.startswith("# ") and (not body or len(body) < 50) and not body_has_query:
|
||
bonus -= 15.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 有 heading 命中但 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())
|
||
|
||
# 判断是否为"空壳"section：heading含查询词 + body是表格
|
||
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:]
|
||
|
||
if heading_has_query and is_table:
|
||
# 找到该section的children（从全部docs中）
|
||
children = [c_idx for c_idx, p_idx in parent_map.items() if p_idx == idx]
|
||
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_heading = child_doc.get("heading", "")
|
||
|
||
# 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中是否有查询核心名词
|
||
body_relevant = any(n in body for n in query_tokens if len(n) >= 2)
|
||
if heading_relevant or body_relevant:
|
||
doc_sections.insert(0, doc_sections.pop(i))
|
||
break
|
||
|
||
if has_action 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 方式和正版玩家方式混合）
|
||
is_guide_query = any(w in query_stripped for w in ['如何', '怎么', '教程', '步骤', '操作', '指南'])
|
||
if is_guide_query and len(doc_sections) > 1:
|
||
# 检查是否为操作指南类文档（多个步骤 section）
|
||
step_headings = {'1.', '2.', '3.', '4.', '5.', '步骤', '第一', '第二', '第三', '第四', '第五'}
|
||
step_count = sum(1 for _, doc, _ in doc_sections if any(s in doc.get('heading', '') for s in step_headings))
|
||
if step_count >= 3:
|
||
# ⭐ 获取最佳文档的路径，查找该文档的所有 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_in_guide = sum(1 for _, doc in sections
|
||
if 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
|
||
results.append((doc, float(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)
|