79 lines
4.6 KiB
Python
79 lines
4.6 KiB
Python
# ============================================================
|
||
# 轻量混合检索（moduleId: knowledge-retrieval, 可重生 ✅, 黄金测试 tests/golden/test_m3_knowledge.py）
|
||
# plan.md §8.2 的 M3 降配实现（诚实声明）：
|
||
# 向量检索 → 字符二元组(bigram)相似度代替（中文无分词友好、零依赖）
|
||
# BM25 → 词频-逆文档频率的简化打分
|
||
# 元数据 → 标签精确命中加权
|
||
# 接口不变：后续换真向量库/重排器时调用方零改动。
|
||
# 硬规则（§8.1）：命中必须带出处（assetId/title/version）。
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import math # idf 对数
|
||
import re # 清洗
|
||
from typing import Any # 类型标注
|
||
|
||
# 非中文字母数字的清洗模式（保留语义字符）
|
||
_CLEAN = re.compile(r"[^\w\u4e00-\u9fff]+")
|
||
|
||
|
||
def _bigrams(text: str) -> set[str]:
|
||
"""字符二元组集合：中文场景下无需分词的稳健相似度基元。"""
|
||
t = _CLEAN.sub("", text.lower()) # 去符号统一小写
|
||
if len(t) < 2: # 太短退化为单字
|
||
return set(t)
|
||
return {t[i:i + 2] for i in range(len(t) - 1)} # 相邻两字符滑窗
|
||
|
||
|
||
def _bigram_sim(q: set[str], d: set[str]) -> float:
|
||
"""Dice 相似度：2|交|/(|q|+|d|)，对长度差异更稳健。"""
|
||
if not q or not d: # 空集合无相似
|
||
return 0.0
|
||
return 2 * len(q & d) / (len(q) + len(d)) # Dice 系数
|
||
|
||
|
||
def search(assets: list[dict[str, Any]], query: str, top_k: int = 3,
|
||
min_score: float = 0.05) -> list[dict[str, Any]]:
|
||
"""混合检索：bigram 相似 + 标签命中加权 + 标题命中加权（P0 只读）。
|
||
|
||
Args:
|
||
assets: 知识资产列表（KnowledgeStore.assets）
|
||
query: 用户问题
|
||
top_k: 返回条数
|
||
min_score: 分数下限（低于视为未命中，触发"知识库没有相关内容"的诚实回答）
|
||
Returns:
|
||
[{asset 元信息, score, snippet}] 按分数降序；必带出处字段
|
||
"""
|
||
q_bi = _bigrams(query) # 查询的二元组
|
||
# 简化 idf：出现某二元组的文档越少，其区分度越高
|
||
docs_bi = [_bigrams(a["title"] + a["content"] + " ".join(a["tags"])) for a in assets] # 各文档二元组
|
||
n_docs = max(len(assets), 1) # 文档数
|
||
df: dict[str, int] = {} # 文档频率表
|
||
for bi in docs_bi: # 统计每个二元组出现的文档数
|
||
for g in bi:
|
||
df[g] = df.get(g, 0) + 1
|
||
scored: list[tuple[float, dict[str, Any]]] = [] # (分数, 资产)
|
||
for a, d_bi in zip(assets, docs_bi): # 逐资产打分
|
||
base = _bigram_sim(q_bi, d_bi) # 基础相似度
|
||
# idf 加权：查询与文档共有的高区分度二元组占比
|
||
shared = q_bi & d_bi # 共有二元组
|
||
idf_bonus = sum(math.log(1 + n_docs / df.get(g, 1)) for g in shared) / (len(q_bi) + 1) # 区分度加成
|
||
tag_bonus = 0.3 if any(t in query for t in a["tags"]) else 0.0 # 标签精确命中加权
|
||
title_bonus = 0.4 if any(ch in a["title"] for ch in [query]) or a["title"] in query else 0.0 # 标题整体命中
|
||
score = base + 0.1 * idf_bonus + tag_bonus + title_bonus # 混合分
|
||
if score >= min_score: # 过滤噪声
|
||
scored.append((score, a))
|
||
scored.sort(key=lambda x: -x[0]) # 按分数降序
|
||
hits = [] # 命中结果（带出处）
|
||
for score, a in scored[:top_k]: # 取前 K
|
||
hits.append({
|
||
"assetId": a["assetId"], # 出处：资产 ID
|
||
"title": a["title"], # 出处：标题
|
||
"kind": a["kind"], # 类别
|
||
"version": a["version"], # 出处：版本（§8.1 版本化）
|
||
"score": round(score, 3), # 相关度（调试与排序展示）
|
||
"snippet": a["content"][:120] + ("…" if len(a["content"]) > 120 else ""), # 摘录
|
||
"content": a["content"], # 全文（LLM 拼上下文用）
|
||
})
|
||
return hits # 空列表 = 未命中（上游诚实告知）
|