aps-agent/server/knowledge/retrieval.py

# ============================================================
# 轻量混合检索（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):               # 逐资产打分
        shared = q_bi & d_bi                           # 共有二元组
        if not shared:                                 # 无共有 → 未命中
            continue
        base = _bigram_sim(q_bi, d_bi)                 # 基础相似度
        # 纯拉丁噪声查询（如 quantum blockchain）易与英文技术词偶合：要求更高 base
        query_has_cjk = any("\u4e00" <= c <= "\u9fff" for c in query)
        if not query_has_cjk and base < 0.12:
            continue
        # idf 加权：查询与文档共有的高区分度二元组占比
        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                                        # 空列表 = 未命中（上游诚实告知）