From 87f3fae1aad84b9983f76b81062329cf7b128f43 Mon Sep 17 00:00:00 2001 From: lacerate551 <128470311+lacerate551@users.noreply.github.com> Date: Wed, 17 Jun 2026 17:49:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(server-release):=20=E4=BB=8E=20main=20?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=E6=A3=80=E7=B4=A2=E5=A2=9E=E5=BC=BA=EF=BC=88?= =?UTF-8?q?=E7=AB=A0=E8=8A=82=E8=81=9A=E7=B1=BB=E6=8F=90=E5=8D=87=20+=20BM?= =?UTF-8?q?25=20=E8=BF=BD=E5=8A=A0=E4=BF=AE=E5=A4=8D=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 main 分支选择性提取检索质量增强代码，不含端口/API 配置变更： - knowledge/base.py: BM25Index.add_documents 改为追加+去重模式， 修复多文件上传后仅保留最后文件切片的覆盖 bug - core/engine.py: 章节聚类提升（_section_cluster_boost）、词法匹配 辅助种子资格（_chunk_lexical_score）、扩展参数优化 （CONTEXT_EXPANSION_AFTER 5→8, MAX_EXPANDED_NEIGHBORS 4→8） - api/chat_routes.py: 路由层词法匹配救援（_rescue_lexical_match）、 章节聚类救援安全网（_rescue_section_cluster） 部署后需重建 BM25 索引：KnowledgeBaseManager().rebuild_bm25_index('public_kb') --- api/chat_routes.py | 205 ++++++++++++++++++++++++++++++++++++++++++++- core/engine.py | 171 +++++++++++++++++++++++++++++++++++-- knowledge/base.py | 28 +++++-- 3 files changed, 390 insertions(+), 14 deletions(-) diff --git a/api/chat_routes.py b/api/chat_routes.py index ac59de0..e1ee619 100644 --- a/api/chat_routes.py +++ b/api/chat_routes.py @@ -292,6 +292,198 @@ def _process_table_doc(doc: str, meta: Dict) -> str: return doc +def _rescue_lexical_match(contexts: List[Dict], retrieval_query: str, + min_score: float) -> List[Dict]: + """ + 词法匹配救援：当切片文本精确包含查询的核心关键词但 CrossEncoder 评分很低时， + 将其分数提升至保底值，并同时救援同 source 下 chunk_index 相邻的切片。 + + 适用场景： + 1. 某个 section 只有一个正确切片（无法触发聚类救援） + 2. 枚举类问题的 header 切片被词法匹配救援后，其后续子条目也应被一并保留 + + Args: + contexts: 全部上下文切片 + retrieval_query: 检索查询 + min_score: 最低分数阈值 + + Returns: + 修改后的 contexts + """ + if not contexts or not retrieval_query: + return contexts + + import re + from config import CLUSTER_RESCUE_FLOOR + + # 清理查询：去除 markdown 格式和标点 + clean_query = re.sub(r'\*+|#+|`', '', retrieval_query) + clean_query = re.sub(r'[？?！!。，,、；;：:"""\'\s]+', ' ', clean_query).strip() + + if len(clean_query) < 2: + return contexts + + # 提取查询中的有意义 bigram（连续两字组） + query_bigrams = set() + for i in range(len(clean_query) - 1): + w = clean_query[i:i+2].strip() + if len(w) == 2: + query_bigrams.add(w) + + if not query_bigrams: + return contexts + + # Phase 1: 词法匹配救援——找到高分匹配的切片 + rescued_seeds = [] # [(source, chunk_index)] + for ctx in contexts: + if ctx.get('score', 0) >= min_score: + continue + + doc = ctx.get('doc', '') or '' + meta = ctx.get('meta', {}) + section = meta.get('section', '') or meta.get('section_path', '') + combined = doc + ' ' + section + + matched = sum(1 for w in query_bigrams if w in combined) + match_ratio = matched / len(query_bigrams) + + if match_ratio > 0.35: + ctx['score'] = max(ctx.get('score', 0), CLUSTER_RESCUE_FLOOR) + source = meta.get('source', '') + chunk_index = meta.get('chunk_index') + if source and chunk_index is not None: + try: + rescued_seeds.append((source, int(chunk_index))) + except (ValueError, TypeError): + pass + + # Phase 2: 邻居救援——对每个被词法匹配救援的种子， + # 同时救援同 source 下 chunk_index 后续相邻的切片（枚举子条目） + if rescued_seeds: + for source, seed_idx in rescued_seeds: + for ctx in contexts: + if ctx.get('score', 0) >= min_score: + continue + meta = ctx.get('meta', {}) + if meta.get('source') != source: + continue + try: + n_idx = int(meta.get('chunk_index', -1)) + except (ValueError, TypeError): + continue + # 救援种子后续 8 个相邻切片（覆盖大多数枚举/条款模式） + if seed_idx < n_idx <= seed_idx + 8: + ctx['score'] = max(ctx.get('score', 0), CLUSTER_RESCUE_FLOOR) + + return contexts + + +def _normalize_section_for_rescue(section_path: str, levels: int = 2) -> str: + """归一化 section_path：取前 N 级路径用于分组""" + if not section_path: + return '' + parts = [p.strip() for p in section_path.split('>')] + return ' > '.join(parts[:levels]) + + +def _rescue_section_cluster(contexts: List[Dict], retrieval_query: str, + min_score: float) -> List[Dict]: + """ + 路由层章节聚类救援：当某个 section 有多个候选切片但全部低于 min_score 时， + 给该 section 的切片分配保底分数，使其通过后续的 min_score 过滤。 + + 作为引擎层 _section_cluster_boost 的二次安全网，防止极端情况下所有正确切片被过滤。 + + Args: + contexts: engine 返回的全部上下文切片（每个含 doc, meta, score） + retrieval_query: 改写后的检索查询 + min_score: 当前的最低分数阈值 + + Returns: + 修改后的 contexts（部分切片 score 被提升至保底分数） + """ + if not contexts: + return contexts + + from config import ( + CLUSTER_MIN_MEMBERS, CLUSTER_MIN_TYPES, CLUSTER_RESCUE_FLOOR, + CLUSTER_MAX_SECTIONS, CLUSTER_MAX_RESCUE_PER_SECTION, + CLUSTER_SECTION_PREFIX_LEVELS, + ) + + # 1. 按 (source, normalized_section) 分组 + from collections import defaultdict + section_groups = defaultdict(list) # key → [index_in_contexts] + + for i, ctx in enumerate(contexts): + meta = ctx.get('meta', {}) + source = meta.get('source', '') + section_path = meta.get('section', '') or meta.get('section_path', '') + norm_section = _normalize_section_for_rescue(section_path, CLUSTER_SECTION_PREFIX_LEVELS) + if not source or not norm_section: + continue + key = (source, norm_section) + section_groups[key].append(i) + + # 2. 检测"全灭 section"并计算聚类强度 + rescue_candidates = [] # (strength, key, member_indices) + + for key, indices in section_groups.items(): + if len(indices) < CLUSTER_MIN_MEMBERS: + continue + + # 检查是否所有成员都低于 min_score（"全灭"） + scores = [contexts[i].get('score', 0) for i in indices] + if any(s >= min_score for s in scores): + continue # 已有成员通过阈值，无需救援 + + # 计算聚类强度 + chunk_types = set(contexts[i].get('meta', {}).get('chunk_type', 'text') for i in indices) + type_diversity = len(chunk_types) + if type_diversity < CLUSTER_MIN_TYPES: + continue # 类型不够多样，可能是噪音 + + # 查询与 section_path 的字符重叠率 + source, norm_section = key + query_chars = set(retrieval_query) + section_chars = set(norm_section) + overlap = len(query_chars & section_chars) / max(len(query_chars), 1) + + # 聚类强度 = 成员数 × 类型多样性 × (1 + 查询匹配度) + strength = len(indices) * type_diversity * (1.0 + overlap) + rescue_candidates.append((strength, key, indices)) + + # 3. 按强度降序，救援 top-N section + rescue_candidates.sort(key=lambda x: x[0], reverse=True) + + rescued_sections = [] + total_rescued = 0 + + for strength, key, indices in rescue_candidates[:CLUSTER_MAX_SECTIONS]: + # 对组内切片分配保底分数（仅低于 min_score 的） + rescue_count = 0 + for idx in indices: + if rescue_count >= CLUSTER_MAX_RESCUE_PER_SECTION: + break + ctx = contexts[idx] + if ctx.get('score', 0) < min_score: + ctx['score'] = CLUSTER_RESCUE_FLOOR + rescue_count += 1 + total_rescued += 1 + + if rescue_count > 0: + source, norm_section = key + rescued_sections.append({ + 'source': source, + 'section': norm_section, + 'members': len(indices), + 'rescued': rescue_count, + 'strength': round(strength, 2) + }) + + return contexts + + def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks: int, min_score: float = 0.0) -> List[Dict]: """ @@ -1635,7 +1827,8 @@ def rag(): DIRECT_CONTEXT_MAX_CHARS, RERANK_CONTEXT_MIN_SCORE, CONTEXT_MAX_CHARS, CONTEXT_SOFT_LIMIT, - CONFIDENCE_WARN_THRESHOLD, CONFIDENCE_CAUTION_THRESHOLD + CONFIDENCE_WARN_THRESHOLD, CONFIDENCE_CAUTION_THRESHOLD, + SECTION_CLUSTER_RESCUE_ENABLED ) data = request.json or {} @@ -2140,6 +2333,16 @@ def rag(): # 4. 构建 prompt（Phase 6：LLM 图片感知） # Bug 1 修复：文本切片用于 top 5 名额竞争，图片描述不参与竞争 + + # 词法匹配救援：当切片文本精确包含查询关键词但 CrossEncoder 评分低时， + # 提升分数使其通过 min_score 过滤（适用于独立切片无法触发聚类救援的场景） + contexts = _rescue_lexical_match(contexts, retrieval_query, RERANK_CONTEXT_MIN_SCORE) + + # 章节聚类救援（路由层安全网）：在 min_score 过滤前， + # 检测"全灭 section"并分配保底分数 + if SECTION_CLUSTER_RESCUE_ENABLED: + contexts = _rescue_section_cluster(contexts, retrieval_query, RERANK_CONTEXT_MIN_SCORE) + text_contexts = _order_text_contexts_for_prompt(contexts, message, MAX_CONTEXT_CHUNKS, min_score=RERANK_CONTEXT_MIN_SCORE) # Phase 2：按字符预算构建上下文 diff --git a/core/engine.py b/core/engine.py index 07d0918..a4c855a 100644 --- a/core/engine.py +++ b/core/engine.py @@ -77,6 +77,10 @@ try: CONTEXT_EXPANSION_MAX_CHUNKS, ENUM_QUERY_DISABLE_TOPK_SHRINK, ENUM_QUERY_MMR_LAMBDA, # Phase 3 扩展精细化 EXPANSION_SCORE_THRESHOLD, MAX_EXPANDED_NEIGHBORS, + # 章节聚类救援 + SECTION_CLUSTER_BOOST_ENABLED, CLUSTER_MIN_MEMBERS, CLUSTER_MIN_TYPES, + CLUSTER_SEED_FLOOR, CLUSTER_MAX_BOOST_PER_SECTION, CLUSTER_MAX_SECTIONS, + CLUSTER_SECTION_PREFIX_LEVELS, # 上下文与生成 LLM_TEMPERATURE, LLM_MAX_TOKENS, RECALL_MULTIPLIER, # FAQ 与黑名单 @@ -102,10 +106,18 @@ except ImportError: MMR_TOP_K = 30 CONTEXT_EXPANSION_ENABLED = True CONTEXT_EXPANSION_BEFORE = 1 - CONTEXT_EXPANSION_AFTER = 5 + CONTEXT_EXPANSION_AFTER = 8 CONTEXT_EXPANSION_MAX_CHUNKS = 24 EXPANSION_SCORE_THRESHOLD = 0.3 - MAX_EXPANDED_NEIGHBORS = 4 + MAX_EXPANDED_NEIGHBORS = 8 + # 章节聚类救援默认值 + SECTION_CLUSTER_BOOST_ENABLED = True + CLUSTER_MIN_MEMBERS = 3 + CLUSTER_MIN_TYPES = 2 + CLUSTER_SEED_FLOOR = 0.35 + CLUSTER_MAX_BOOST_PER_SECTION = 8 + CLUSTER_MAX_SECTIONS = 3 + CLUSTER_SECTION_PREFIX_LEVELS = 1 ENUM_QUERY_DISABLE_TOPK_SHRINK = True ENUM_QUERY_MMR_LAMBDA = 0.85 DYNAMIC_RRF_ENABLED = True @@ -733,11 +745,19 @@ class RAGEngine: # 时间衰减（Time Decay） fused_results = self._apply_time_decay(fused_results) + # 提前附加 _debug，使聚类提升能写入调试步骤 + fused_results['_debug'] = _debug + + # ========== 章节聚类提升：在扩展前将低分但聚类的切片提升至种子阈值 ========== + if SECTION_CLUSTER_BOOST_ENABLED: + fused_results = self._section_cluster_boost(fused_results, query) + # ========== 上下文扩展：补充强命中切片周围的连续文本（rerank 之后，防止被截断）========== # Phase 3：仅对高分种子扩展邻居 before_exp = len(fused_results['ids'][0]) if fused_results.get('ids') else 0 fused_results = self._expand_contiguous_chunks(fused_results, top_k=top_k, - min_score=EXPANSION_SCORE_THRESHOLD) + min_score=EXPANSION_SCORE_THRESHOLD, + query=query) after_exp = len(fused_results['ids'][0]) if fused_results.get('ids') else 0 _debug['steps'].append({'name': 'context_expansion', 'before': before_exp, 'after': after_exp}) @@ -1162,13 +1182,14 @@ class RAGEngine: return self.collection def _expand_contiguous_chunks(self, results: dict, top_k: int = None, - min_score: float = 0.0) -> dict: + min_score: float = 0.0, query: str = '') -> dict: """Add same-source same-section neighbor text chunks around strong hits. Args: results: 检索结果 top_k: 最大切片数 min_score: Phase 3 最低分数阈值，仅对 Rerank 分数高于此值的种子扩展 + query: 查询文本，用于词法匹配辅助种子资格判定 """ if not CONTEXT_EXPANSION_ENABLED: return results @@ -1198,7 +1219,7 @@ class RAGEngine: seeds = [ (doc_id, doc, meta, dist) for doc_id, doc, meta, dist in items[:base_limit] - if meta.get('chunk_type', 'text') == 'text' + if (meta.get('chunk_type', 'text') == 'text' or meta.get('_cluster_boosted')) and meta.get('source') and self._to_int(meta.get('chunk_index')) is not None ] @@ -1210,7 +1231,11 @@ class RAGEngine: # Phase 3：跳过分数低于阈值的种子（仅当 min_score > 0 时生效） if min_score > 0 and seed_dist < min_score: - continue + # 词法匹配豁免：CrossEncoder 低分但关键词重叠度高时仍允许作为种子 + if query and self._chunk_lexical_score(_seed_doc, query) > 0.3: + pass # 词法匹配度高，允许作为种子 + else: + continue source = seed_meta.get('source') section = seed_meta.get('section', '') or seed_meta.get('section_path', '') @@ -1316,6 +1341,130 @@ class RAGEngine: expanded[key] = results[key] return expanded + @staticmethod + def _normalize_section_path(section_path: str, levels: int = None) -> str: + """归一化 section_path：取前 N 级路径，容忍 MinerU 标题检测误差。 + + 例如: "第三章 吸烟场所的功能设置 > 第三条 文明吸烟..." → "第三章 吸烟场所的功能设置" + """ + if not section_path: + return '' + if levels is None: + levels = CLUSTER_SECTION_PREFIX_LEVELS + parts = [p.strip() for p in section_path.split('>')] + return ' > '.join(parts[:levels]) + + def _section_cluster_boost(self, results: dict, query: str = '') -> dict: + """章节聚类提升：当同一 section 下多个切片（text+table）同时出现在候选集中， + 即使单个切片 CrossEncoder 分数很低，也将整组提升到种子阈值。 + + 核心洞察：单个低分切片不可信，但同一 section 多个切片同时出现是强信号。 + 提升后的切片可以作为 _expand_contiguous_chunks 的种子，触发邻居扩展。 + + Args: + results: rerank 后的检索结果 + query: 用户查询（用于后续扩展） + + Returns: + 修改后的 results（distances 被调整，meta 中标记 _cluster_boosted） + """ + if not results.get('ids') or not results['ids'][0]: + return results + + ids = results['ids'][0] + metas = results.get('metadatas', [[]])[0] + distances = results.get('distances', [[]])[0] if results.get('distances') else None + + if not distances: + return results + + # 1. 按 (source, normalized_section) 分组 + from collections import defaultdict + section_groups = defaultdict(list) # key → [(index, meta, dist)] + + for i, (meta, dist) in enumerate(zip(metas, distances)): + source = meta.get('source', '') + section_path = meta.get('section', '') or meta.get('section_path', '') + norm_section = self._normalize_section_path(section_path) + if not source or not norm_section: + continue + key = (source, norm_section) + section_groups[key].append((i, meta, dist)) + + # 2. 检测聚类信号并提升 + boost_target_dist = 1.0 - CLUSTER_SEED_FLOOR # score=0.35 → dist=0.65 + boosted_sections = [] + total_boosted = 0 + + # 按组成员数降序排列，优先处理最大聚类 + sorted_groups = sorted(section_groups.items(), key=lambda x: len(x[1]), reverse=True) + + for (source, norm_section), members in sorted_groups: + if len(boosted_sections) >= CLUSTER_MAX_SECTIONS: + break + + # 聚类信号检测：成员数 >= 阈值 且 类型多样性 >= 阈值 + chunk_types = set(m[1].get('chunk_type', 'text') for m in members) + if len(members) < CLUSTER_MIN_MEMBERS or len(chunk_types) < CLUSTER_MIN_TYPES: + continue + + # 提升组内切片分数（仅提升低于阈值的） + boost_count = 0 + for idx, meta, dist in members: + if boost_count >= CLUSTER_MAX_BOOST_PER_SECTION: + break + # 只提升分数低于种子阈值的切片（高分切片不需要） + if dist > boost_target_dist: + distances[idx] = boost_target_dist + meta['_cluster_boosted'] = True + boost_count += 1 + total_boosted += 1 + + if boost_count > 0: + boosted_sections.append({ + 'source': source, + 'section': norm_section, + 'members': len(members), + 'types': list(chunk_types), + 'boosted': boost_count + }) + + # 3. 写 debug 信息 + if boosted_sections: + debug_info = results.get('_debug', {}) + if 'steps' not in debug_info: + debug_info['steps'] = [] + debug_info['steps'].append({ + 'name': 'section_cluster_boost', + 'sections': boosted_sections, + 'total_boosted': total_boosted + }) + results['_debug'] = debug_info + logger.info(f"[章节聚类提升] 提升 {total_boosted} 个切片，" + f"涉及 {len(boosted_sections)} 个 section: " + f"{[s['section'][:30] for s in boosted_sections]}") + + return results + + @staticmethod + def _chunk_lexical_score(chunk_text: str, query: str) -> float: + """计算切片文本与查询的词法重叠度（bigram 命中率），用于辅助种子资格判定。""" + if not chunk_text or not query: + return 0.0 + import re + clean_q = re.sub(r'[？?！!。，,、；;：:"""\'\s*#`]+', ' ', query).strip() + if len(clean_q) < 2: + return 0.0 + bigrams = set() + for i in range(len(clean_q) - 1): + w = clean_q[i:i+2].strip() + if len(w) == 2: + bigrams.add(w) + if not bigrams: + return 0.0 + matched = sum(1 for w in bigrams if w in chunk_text) + return matched / len(bigrams) + def _search_with_sub_queries( self, original_query, sub_queries, top_k=5, allowed_levels=None, role=None, department=None, collections=None, source_filter=None @@ -1627,11 +1776,19 @@ class RAGEngine: # 时间衰减 fused_results = self._apply_time_decay(fused_results) + # 提前附加 _debug，使聚类提升能写入调试步骤 + fused_results['_debug'] = _debug + + # ========== 章节聚类提升：在扩展前将低分但聚类的切片提升至种子阈值 ========== + if SECTION_CLUSTER_BOOST_ENABLED: + fused_results = self._section_cluster_boost(fused_results, query) + # ========== 上下文扩展：补充强命中切片周围的连续文本（rerank 之后，防止被截断）========== # Phase 3：仅对高分种子扩展邻居 before_exp = len(fused_results['ids'][0]) if fused_results.get('ids') else 0 fused_results = self._expand_contiguous_chunks(fused_results, top_k=top_k, - min_score=EXPANSION_SCORE_THRESHOLD) + min_score=EXPANSION_SCORE_THRESHOLD, + query=query) after_exp = len(fused_results['ids'][0]) if fused_results.get('ids') else 0 if _debug is not None: _debug['steps'].append({'name': 'context_expansion', 'before': before_exp, 'after': after_exp}) diff --git a/knowledge/base.py b/knowledge/base.py index 814207b..5210c27 100644 --- a/knowledge/base.py +++ b/knowledge/base.py @@ -382,18 +382,34 @@ class BM25Index: def add_documents(self, ids: List[str], documents: List[str], metadatas: List[dict]) -> None: """ - 添加文档到索引（会覆盖原有索引） + 添加文档到索引（追加模式，自动去重） + + 如果 ID 已存在则更新对应文档，否则追加新文档。 + 添加后自动重建 BM25 索引。 Args: ids: 文档 ID 列表 documents: 文档内容列表 metadatas: 文档元数据列表 """ - self.ids = ids - self.documents = documents - self.metadatas = metadatas - if documents: - tokenized = [self.tokenize(doc) for doc in documents] + # 建立已有 ID -> 索引位置 的映射，用于去重 + existing_map = {doc_id: idx for idx, doc_id in enumerate(self.ids)} + + for i, doc_id in enumerate(ids): + if doc_id in existing_map: + # 更新已有文档 + pos = existing_map[doc_id] + self.documents[pos] = documents[i] + self.metadatas[pos] = metadatas[i] + else: + # 追加新文档 + existing_map[doc_id] = len(self.ids) + self.ids.append(doc_id) + self.documents.append(documents[i]) + self.metadatas.append(metadatas[i]) + + if self.documents: + tokenized = [self.tokenize(doc) for doc in self.documents] self.bm25 = BM25Okapi(tokenized) def search(self, query: str, top_k: int = 10) -> Tuple[List[str], List[str], List[dict], List[float]]: