group_fqcd_jr/app/service/knowledge_search_service.py

"""知识库检索：把客户问题向量化后在四个知识集合里检索（只读）。

为什么不复用记忆那套 `VectorMemoryAdapter`：它只把命中折叠成 `(memory_uuid, score)`，
会把知识块的标题与正文丢掉。而客服回答必须能把**原文与来源**一起交给客户——金融场景
里「答案出自哪份文件的哪一条」本身就是交付物的一部分，丢了正文等于没法给来源引用。

失败语义与基座一致：**任何一步失败都不抛异常给主链路**，而是返回 `degraded=True`
的空结果，由调用方（客服 Agent）据此走「引导客户致电人工客服」的兜底路径。
金融场景下"答不了"是可接受的结果，"答错"不是。
"""

from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass, field
from typing import Any, Protocol

from app.core.knowledge_contracts import visibility_expression
from app.core.knowledge_schema import CollectionSchema, SchemaCache, detect_schema

# 四个知识集合（方案 §2.4.4 / §4.1 + `乙-7` 第 4 集合）
BASIC_COLLECTION = "fin_basic_collection"  # 金融行业基础信息（行业通用常识）
FAQ_COLLECTION = "fin_faq_collection"
PRODUCT_COLLECTION = "fin_product_collection"
POLICY_COLLECTION = "fin_policy_collection"

#: 默认检索面 = **三个主集合**。第 4 集合**不在其中**，这是实测出来的结论：
#:
#: 2026-09-19 把 `fin_basic_collection` 加进默认检索面后复跑 46 条金标，
#: `M-1` 出口准确率 **100% → 91.3%**（`A-03`/`A-07`/`B-06`/`E-04` 由 `E3`/`E5b` 掉成 `E4`）。
#: 根因不是"答错了"，而是**通用常识文本天生与很多问题都像**：它把**次优分**抬到 0.69–0.79，
#: 而 `MIN_GAP=0.07` 的"top1 领先不足"判据据此把本该直答的题推给证据约束生成。
#: 换句话说，它是**抢答**，不是**补位** —— 而 FAQ 里已有的专条（如 `FAQ-0065` 的
#: 「什么是T日、T+1？」0.703）本来就比第 4 集合的通用段落更贴题。
#:
#: 因此本轮的落法：**集合建好、灌好、契约放行，但不进默认检索面**。
#: 它是"主集合答不出来时的兜底素材"，要真正启用需要先做**跨集合分数归一 / 重排**，
#: 否则每个集合的分数量纲不同，混在一起比大小本身就不成立。
#: 启用前先复跑 `_eval_harness`，`M-1` 必须仍是 100%。
PRIMARY_COLLECTIONS: tuple[str, ...] = (FAQ_COLLECTION, PRODUCT_COLLECTION, POLICY_COLLECTION)
DEFAULT_COLLECTIONS: tuple[str, ...] = PRIMARY_COLLECTIONS
#: 补位集合（当前**不在**默认检索面内，见上方实测记录）。
FALLBACK_COLLECTIONS: tuple[str, ...] = (BASIC_COLLECTION,)

# 检索输出字段：**运行时探测**，不硬编码任何一套 schema。
#
# 背景（实测）：同一批集合名在不同环境下可能是两套 schema ——
#   环境甲：`knowledge_id` / `snippet`（无 visibility）
#   环境乙：`doc_id` / `content` / `visibility` / `chapter` / `source_file`
# 硬编码任一套都会把另一套打挂（Milvus 对不存在的字段直接报错 → 三集合全失败 →
# `degraded=True` → 客服一律"引导人工"）。因此字段名一律经
# `app/core/knowledge_schema.py` 的 `detect_schema()` 探测得出：
#   · 逻辑名 `doc_id` → 物理名 `doc_id` 或 `knowledge_id`（谁在就用谁）
#   · 逻辑名 `content` → 物理名 `content` 或 `snippet`
#   · `visibility` / `source_file` / `chapter` 等存在就用、不存在就跳过
# `KnowledgeHit` 的对外形状**保持不变**（下游 Agent 与测试依赖它），只改"怎么从库里取"。

# 关键字精确召回：客户问到产品名这类**专有名词**时，字面匹配比相似度更确定。
#
# 为什么需要：专有名词在 embedding 空间里不占优势。实测 160 条知识，客户问
# 「季季盈90天的起投金额是多少」向量 top1 = PROD-007 仅 0.6291（够不到 0.75 的硬门槛，
# 只能靠与次优的差值勉强通过）；而同一次查询用 `title like "%季季盈%"` 是**唯一命中**
# PROD-007。既然客户已经明确说出了产品名，就不该再让相似度去赌。
KEYWORD_MATCH_SCORE = 1.0  # 字面命中的确定分，压过任何相似度分
# 与标题的最长公共子串至少要这么长，才算"客户确切提到了它"。
# 为什么不是 4：实测客户问「基金赎回几天到账」，与手册章节标题「5.2 基金赎回流程」
# 正好有 4 个字连续重合——但"基金赎回"是业务动作词，不是专有名词。产品名
# （"南方季季盈90天"）通常比业务动作词长，取 6 字能同时保住产品名、挡住动作词。
MIN_KEYWORD_OVERLAP = 6
KEYWORD_SCAN_LIMIT = 500  # 一次最多扫描多少条标题；知识库到上千块后应改为倒排索引

# 行级子块命中时，其父块（整节）按子块分数的这个比例一并返回：排在子块之后，
# 既不抢"起投多少"这类聚焦答案，又不至于把差距压到转人工门槛之下。
PARENT_SCORE_RATIO = 0.9

# 字面匹配只在向量结果**不够确定**时介入。
#
# 这条门禁是实测逼出来的：客户问「基金赎回几天到账」，向量已给出正确答案
# （FAQ-0016「基金赎回到账需要多长时间?」得 0.8060），但产品手册里的章节标题
# 「五、申购赎回操作指南」与问句也有 6 个字连续重合，无条件字面匹配会把它顶到第一，
# 用操作步骤替换掉客户真正问的到账时间。所以字面匹配是**兜底**，不是优先。
#
# 门槛必须与客服 Agent 的高置信门槛一致，tests/unit 有断言锁定两者相等。
VECTOR_CONFIDENT_SCORE = 0.75


class VectorSearcher(Protocol):
    """只依赖用到的两个方法，便于测试替身注入。"""

    def search(self, **kwargs: Any) -> Any: ...


Embedder = Callable[[str], Awaitable[list[float]]]


@dataclass(frozen=True)
class KnowledgeHit:
    """一条知识命中；`score` 为 COSINE 相似度（越大越相似）。"""

    doc_id: str
    title: str
    content: str
    score: float
    source_file: str = ""
    visibility: str = "public"
    doc_no: str = ""
    version: str = ""
    chapter: str = ""
    #: v1.4（2026-09-18）：支撑「同族合并」（`family_id`）与「计算型回答定位参数位」
    #: （`param_class`）与意图标签（`intent`）。集合/切片件缺这三个字段时留空串，
    #: 消费方据此退化为"不启用"，不编造值。
    family_id: str = ""
    param_class: str = ""
    intent: str = ""

    @property
    def reference_title(self) -> str:
        """给客户看的来源标题：优先带内部文件编号，便于人工核对。"""
        return f"{self.title}（{self.doc_no}）" if self.doc_no else self.title


@dataclass(frozen=True)
class KnowledgeSearchOutcome:
    """检索结果；`degraded=True` 表示检索链路故障，调用方必须走兜底而非当'没找到'。"""

    hits: tuple[KnowledgeHit, ...] = ()
    degraded: bool = False
    reason: str = ""
    searched_collections: tuple[str, ...] = field(default_factory=tuple)

    @property
    def best(self) -> KnowledgeHit | None:
        return self.hits[0] if self.hits else None

    @property
    def top_score(self) -> float:
        return self.hits[0].score if self.hits else 0.0


class KnowledgeSearchService:
    def __init__(
        self,
        client: VectorSearcher | None,
        embedder: Embedder | None,
        *,
        collections: Sequence[str] = DEFAULT_COLLECTIONS,
    ) -> None:
        self._client = client
        self._embedder = embedder
        self._collections = tuple(collections)
        # 字段探测结果按集合缓存（进程内一次）：`describe_collection` 是元数据调用，
        # 但检索是热路径，不该每次调用都打一次。集合重建后需重启或 `invalidate`。
        self._schemas = SchemaCache()

    @property
    def available(self) -> bool:
        """向量库与向量化能力是否都在位；缺任一项都不做检索，直接走兜底。"""
        return self._client is not None and self._embedder is not None

    def _schema_for(self, client: Any, collection: str) -> CollectionSchema | None:
        """该集合的字段映射；探测失败或缺少必需字段时返回 None（调用方跳过该集合）。"""
        schema = detect_schema(client, collection, cache=self._schemas)
        return schema if schema.usable else None

    async def search(
        self,
        query: str,
        *,
        tiers: frozenset[str],
        collections: Sequence[str] | None = None,
        top_k: int = 5,
        literal_parallel: bool = False,
    ) -> KnowledgeSearchOutcome:
        """检索知识库。

        `tiers` 是**主体被允许的档位集合**，必填、**无默认值** —— 遗漏即 `TypeError`，
        不会静默放行（`D2.4` §5.4 fail-closed 的第一道防线）。求值入口是
        `app.core.knowledge_contracts.tiers_for_roles(context.roles)`，**业务代码里
        不得手写档位字面量**。

        过滤在 Milvus 侧完成（`visibility in [...]`），是**检索层硬隔离**，
        不依赖提示词约束。
        """
        text = query.strip()
        if not text:
            return KnowledgeSearchOutcome(reason="empty_query")
        if not self.available:
            return KnowledgeSearchOutcome(degraded=True, reason="vector_backend_unavailable")

        client, embedder = self._client, self._embedder
        if client is None or embedder is None:
            # 与 available 重复，但这里需要类型收窄（mypy 不跨属性判断 Optional）
            return KnowledgeSearchOutcome(degraded=True, reason="vector_backend_unavailable")
        try:
            vector = await embedder(text)
        except Exception:
            return KnowledgeSearchOutcome(degraded=True, reason="embedding_failed")
        if not vector:
            return KnowledgeSearchOutcome(degraded=True, reason="embedding_empty")

        targets = tuple(collections or self._collections)
        # 每个集合**各自探测**字段名：不同环境（甚至同环境不同集合）可能是不同 schema，
        # 用统一的一套字段名去查会让整次检索失败（Milvus 对不存在的字段直接报错）。
        schemas: dict[str, CollectionSchema] = {}
        unusable: list[str] = []
        for collection in targets:
            schema = self._schema_for(client, collection)
            if schema is None:
                unusable.append(collection)
                continue
            schemas[collection] = schema
        if not schemas:
            # 一个集合都用不了：是链路/配置故障，不是"知识库里没有"——如实标记降级
            return KnowledgeSearchOutcome(
                degraded=True, reason="collections_unusable", searched_collections=targets
            )
        expression = visibility_expression(tiers)
        # 档位隔离的前提是集合**有** `visibility` 字段。缺字段的集合无法证明隔离，
        # 受限档位下一律不放行（fail-closed）—— 这正是 `K-07`「缺字段即放行」的根治点。
        # 只有请求档位等于全集（`expression is None`）时才允许无过滤读取。
        isolated: dict[str, CollectionSchema] = {}
        barred: list[str] = []
        for collection, schema in schemas.items():
            if expression is not None and not schema.has("visibility"):
                barred.append(collection)
                continue
            isolated[collection] = schema
        collected: list[KnowledgeHit] = []
        failures = len(unusable) + len(barred)
        for collection, schema in isolated.items():
            try:
                raw = client.search(
                    collection_name=collection,
                    data=[vector],
                    limit=max(1, min(top_k, 20)),
                    output_fields=list(schema.output_fields),
                    filter=expression,
                )
            except Exception:
                failures += 1
                continue
            collected.extend(self._parse(raw, collection, schema))

        # 第二路召回：客户确切说出的产品名按字面取回。只在向量结果不够确定时介入，
        # 否则会把向量已经答对的题顶掉（见 VECTOR_CONFIDENT_SCORE 的说明）。
        best_vector_score = max((hit.score for hit in collected), default=0.0)
        # `literal_parallel=True`（`W34` 会签 21）：把字面通道从"条件触发的兜底"
        # 升为**可并行召回的独立通道** —— 不再被 `VECTOR_CONFIDENT_SCORE` 挡住。
        #
        # ⚠️ 默认 `False` ⇒ 本行以下与改动前**逐字相同**，既有调用方零影响。
        # 🔴 打开它**必须与精排同批启用**：字面命中给满分 `1.0` 会把 `gap` 压平 ⇒
        #    触发"领先不足"判据 ⇒ 一律转人工（这正是本通道此前被闸门挡住的原因）。
        if literal_parallel or best_vector_score < VECTOR_CONFIDENT_SCORE:
            collected.extend(
                self._product_keyword_hits(client, isolated, text, expression)
            )

        # 命中行级子块时把父块（整节）一并带回，供调用方按问句选粒度：
        # 「起投多少」要那一行，「介绍一下」要整节。
        collected.extend(self._parent_hits(client, isolated, collected, expression))

        if not collected and failures == len(targets) and targets:
            # 所有集合全查失败：是链路故障，不是"知识库里没有"
            return KnowledgeSearchOutcome(
                degraded=True, reason="search_failed", searched_collections=targets
            )

        collected.sort(key=lambda hit: hit.score, reverse=True)
        # 同一内容可能同时存在于产品手册与问答对里，按 doc_id 去重保留最高分
        deduped: list[KnowledgeHit] = []
        seen: set[str] = set()
        for hit in collected:
            if hit.doc_id in seen:
                continue
            seen.add(hit.doc_id)
            deduped.append(hit)

        # 同一节的**兄弟子块**只保留最高分的那一条（见方法 docstring）。
        # 必须在 doc_id 去重之后：两者挡的是不同的重复形态。
        deduped = self._merge_sibling_subblocks(deduped)

        # "整节块"= **有行级子块挂在它下面**的块。FAQ/政策/公司信息的块没有子块，
        # 它们本身就是细粒度答案，不能和产品手册的整节块混为一谈：第一版用"doc_id 不含
        # 两位数字后缀"判断，把 FAQ 块全当成整节块排到最后，直接害得「基金赎回几天到账」
        # 转人工（正确答案被挤出了 top1）。
        section_ids = {
            parent for parent in (self._parent_of(hit.doc_id) for hit in deduped) if parent
        }
        plain = [hit for hit in deduped if hit.doc_id not in section_ids]
        sections = [hit for hit in deduped if hit.doc_id in section_ids]
        selected = plain[: max(1, top_k)]
        if sections:
            # 整节块保底占最后一个名额：它按分数容易被 top_k 截掉，而「介绍一下」这类
            # 概括问句只能靠它拿到整节（实测被截后客户只收到一行"产品期限 90天封闭期"）。
            # 放末尾是为了不让它参与 top1/top2 判定：实测它挤到第 2 位时 gap 会从
            # 0.090 掉到 0.076，几乎跌破 0.07 的转人工门槛。
            selected = [*selected[: max(0, top_k - 1)], sections[0]]
        return KnowledgeSearchOutcome(
            hits=tuple(selected),
            degraded=failures > 0,
            reason="partial_collection_failure" if failures else "",
            searched_collections=targets,
        )

    # ---- 关键字精确召回（字面匹配） ----

    @staticmethod
    def _overlap_length(left: str, right: str) -> int:
        """两段文本的最长公共子串长度。

        用最长公共子串而不是分词：知识库标题是「南方基金管理股份有限公司 公募基金与专户产品手册 ·
        1.2 南方稳健增利债券 A」这种没有词边界的长串，任何分词器都得先养一份自定义词典，
        而词典会和手册一起过期。子串匹配不需要词典，手册改版也不会失效。
        """
        if not left or not right:
            return 0
        previous = [0] * (len(right) + 1)
        best = 0
        for i in range(1, len(left) + 1):
            current = [0] * (len(right) + 1)
            for j in range(1, len(right) + 1):
                if left[i - 1] == right[j - 1]:
                    current[j] = previous[j - 1] + 1
                    if current[j] > best:
                        best = current[j]
            previous = current
        return best

    def _product_keyword_hits(
        self, client: Any, schemas: dict[str, CollectionSchema], query: str,
        expression: str | None,
    ) -> list[KnowledgeHit]:
        """客户确切说出某个产品名时，按字面把它取出来（兜底用）。

        两处边界都是实测逼出来的：

        1. **只对产品集合做**。客户问「季季盈90天的起投金额是多少」，与通用 FAQ 标题
           「基金起投金额是多少?」的最长公共子串有 7 个字；若把 FAQ 也纳入字面匹配，
           它会和真正的产品块一起拿到满分、差距归零，反而又退化成"转人工"。
        2. **命中不能无条件优先**。产品手册的标题里不只有产品名，还有章节名：客户问
           「基金赎回几天到账」时，「五、申购赎回操作指南」那一块与问句也有 6 个字连续
           重合。所以本方法产出什么是一回事，是否采用由调用方按"向量是否已经足够确定"
           决定（见 VECTOR_CONFIDENT_SCORE）。

        失败一律返回空：关键字路径是**增益**，它坏了不能让整个检索变成故障。
        """
        lookup = getattr(client, "query", None)
        product_schema = schemas.get(PRODUCT_COLLECTION)
        doc_field = product_schema.resolve("doc_id") if product_schema else None
        content_field = product_schema.resolve("content") if product_schema else None
        if lookup is None or product_schema is None or doc_field is None or content_field is None:
            return []  # 客户端不支持标量查询（如测试替身），或本次没查/不能查产品集合
        try:
            rows = lookup(
                collection_name=PRODUCT_COLLECTION,
                filter=expression,
                output_fields=[doc_field, "title"],
                limit=KEYWORD_SCAN_LIMIT,
            )
        except Exception:
            return []

        matched_ids = [
            str(row.get(doc_field) or "")
            for row in (rows if isinstance(rows, list) else [])
            if isinstance(row, dict)
            and self._overlap_length(query, str(row.get("title") or "")) >= MIN_KEYWORD_OVERLAP
        ]
        matched_ids = [doc_id for doc_id in matched_ids if doc_id]
        if not matched_ids:
            return []

        quoted = ", ".join(f'"{doc_id}"' for doc_id in matched_ids)
        try:
            details = lookup(
                collection_name=PRODUCT_COLLECTION,
                filter=f"{doc_field} in [{quoted}]",
                output_fields=list(product_schema.output_fields),
                limit=len(matched_ids),
            )
        except Exception:
            return []

        hits: list[KnowledgeHit] = []
        for row in details if isinstance(details, list) else []:
            if not isinstance(row, dict):
                continue
            content = str(row.get(content_field) or "")
            if not content:
                continue
            hits.append(self._hit_from_row(row, score=KEYWORD_MATCH_SCORE,
                                           schema=product_schema))
        # 一个产品名可能命中多个块（产品概览、费率表各一块）：全都保留，
        # 是不是"只有一个明确候选"交给上层的 gap 判定，这里不替它做选择。
        return hits

    def _parent_hits(
        self, client: Any, schemas: dict[str, CollectionSchema], hits: list[KnowledgeHit],
        expression: str | None,
    ) -> list[KnowledgeHit]:
        """把命中到的行级子块的**父块**一并带回来。

        行级子块让「起投多少」拿到了聚焦答案，但客户问「介绍一下」时会被某一行抢答
        （实测返回了"产品期限 90天封闭期"，而客户要的是整个产品的介绍）。父块是同一
        产品的整节内容，一并带回来，由调用方按问句自己选粒度——检索层不猜客户想听多细。

        分数按子块的 0.9 折算：既排在子块之后（不抢聚焦答案），又不会把差距压到转人工
        门槛之下（0.869 折算成 0.782，与子块差 0.087，仍在 0.07 之上）。

        失败一律返回空：父块是**补充**，取不到不影响子块结果。
        """
        # 只取"得分最高的那个子块"的父块：整节只可能来自一个产品，把命中到的子块的父块
        # 全带上只会挤占 top_k 名额（实测带上多个后，父块反而被截断在门外、整节拿不到）。
        best_child = max(
            (hit for hit in hits if self._parent_of(hit.doc_id) is not None),
            key=lambda hit: hit.score,
            default=None,
        )
        lookup = getattr(client, "query", None)
        if best_child is None:
            return []
        parent_id = self._parent_of(best_child.doc_id)
        if not parent_id or lookup is None:
            return []
        parent_scores = {parent_id: best_child.score * PARENT_SCORE_RATIO}

        quoted = ", ".join(f'"{parent_id}"' for parent_id in parent_scores)
        found: list[KnowledgeHit] = []
        for collection, schema in schemas.items():
            doc_field = schema.resolve("doc_id")
            content_field = schema.resolve("content")
            if doc_field is None or content_field is None:
                continue  # 该集合缺少必需字段（探测阶段已记为不可用）
            try:
                rows = lookup(
                    collection_name=collection,
                    filter=self._anded(expression, f"{doc_field} in [{quoted}]"),
                    output_fields=list(schema.output_fields),
                    limit=KEYWORD_SCAN_LIMIT,
                )
            except Exception:
                continue  # 某个集合查不到不影响其它集合
            for row in rows if isinstance(rows, list) else []:
                if not isinstance(row, dict):
                    continue
                doc_id = str(row.get(doc_field) or "")
                content = str(row.get(content_field) or "")
                if not content or doc_id not in parent_scores:
                    continue
                found.append(self._hit_from_row(row, score=parent_scores[doc_id], schema=schema))
        return found

    @staticmethod
    def _merge_sibling_subblocks(hits: list[KnowledgeHit]) -> list[KnowledgeHit]:
        """同一父块的行级子块**只保留最高分的那一条**，其余兄弟子块不进候选。

        ## 为什么（实测：这是"客服答不出来"的主要来源之一）

        行级子块的粒度设计是对的（整节几百字是混合语义，与「起投多少」这类具体小问题
        相似度天然偏低），但它有个副作用：**同一节的子块彼此分数贴得极近**，而它们
        **全都是"同一个答案"的不同细节，不是并列的多个答案**。

        `doc_id` 去重接不住这种情形 —— `POL-AST-009-07` 与 `POL-AST-009-12` 是两个
        不同的 doc_id，却都来自同一节。实测 `个人投资者适当性管理指南` 第九条被切成
        94 块（16 张问卷表格），问「风险评估问卷怎么评分」时 top5 全是这一节的子块，
        top1 与次优差 **0.002**，客服按"中置信必须领先次优 ≥0.07"判定并列 → **转人工**。

        归并后的实测（同一批向量，离线复算）：
        「风险评估问卷怎么评分」gap 从 **0.002 → 0.093**（转人工 → 可答）；
        10 个问句回归 **0 个变坏**，其中「南方季季盈90天的起投金额」这类靠子块精确
        命中的问题 top1 与答案**完全不变**（gap 反而从 0.113 升到 0.205），
        库外问句「今天天气怎么样」仍正确转人工。

        ## 为什么不会丢信息

        被丢掉的兄弟子块**不是答案本身**，只是同一节的其它细节；本节内容由
        `_parent_hits` 带回来的**父块（整节）**兜底 —— 它就在 `section_ids` 里按保底
        名额占最后一席（见下方 `sections[0]`）。所以客户拿到的仍是"最相关的那一行 +
        整节说明"，与设计意图一致。

        ## 边界

        - 只对"有父块的行级子块"生效（`_parent_of` 非空即形如 `PROD-007-04`）；
        - 父块自身（整节块）、FAQ/政策/公司信息这类**本身就是细粒度答案**的块
          （`_parent_of` 为 None）一律不合并 —— 它们之间打平是真的存在多个候选，
          该转人工就得转人工（实测「高净值客户有什么权益」gap 0.0075 属此类，
          归并不改它，也不能改）。
        - 入参必须**已按分数降序**（调用方刚排过序），否则"保留最高分"不成立。
        """
        kept: list[KnowledgeHit] = []
        best_parent: set[str] = set()
        for hit in hits:
            parent = KnowledgeSearchService._parent_of(hit.doc_id)
            if parent is not None:
                if parent in best_parent:
                    continue
                best_parent.add(parent)
            kept.append(hit)
        return kept

    @staticmethod
    def _parent_of(doc_id: str) -> str | None:
        """行级子块的父块 doc_id；整节块返回 None。

        判据是"末段恰为 2 位数字"（PROD-007-04 → PROD-007），**不是**"含连字符"：
        整节块自己的编号就形如 PROD-901，用连字符判断会把整节块误判成子块。
        """
        head, _, tail = doc_id.rpartition("-")
        if head and tail.isdigit() and len(tail) == 2:
            return head
        return None

    @staticmethod
    def _anded(expression: str | None, extra: str) -> str:
        """把可见性过滤与 doc_id 过滤合成一个 Milvus 表达式。"""
        return f"({expression}) and ({extra})" if expression else extra

    @staticmethod
    def _hit_from_row(row: Any, *, score: float, schema: CollectionSchema) -> KnowledgeHit:
        """把一行 Milvus 结果折成 `KnowledgeHit`。

        字段名**按该集合探测出来的映射**取（可能是 `doc_id`/`content`，也可能是
        `knowledge_id`/`snippet`）；该集合没有的字段留空字符串——**不编造内容**，
        `reference_title` 会因此退回纯标题，来源引用退化但不失真。
        """
        def value(logical: str) -> str:
            field_name = schema.resolve(logical)
            if field_name is None:
                return ""
            return str(row.get(field_name) or "")

        return KnowledgeHit(
            doc_id=value("doc_id"),
            title=value("title"),
            content=value("content"),
            score=score,
            source_file=value("source_file"),
            visibility=value("visibility") or "public",
            doc_no=value("doc_no"),
            version=value("version"),
            chapter=value("chapter"),
            family_id=value("family_id"),
            param_class=value("param_class"),
            intent=value("intent"),
        )

    @staticmethod
    def _parse(raw: Any, collection: str, schema: CollectionSchema) -> list[KnowledgeHit]:
        """把 pymilvus 的 `[[{id, distance, entity}]]` 折叠成命中列表（纯函数，不抛异常）。"""
        hits: list[KnowledgeHit] = []
        groups = raw if isinstance(raw, (list, tuple)) else [raw]
        for group in groups:
            rows = group if isinstance(group, (list, tuple)) else [group]
            for row in rows:
                entity = row.get("entity") if isinstance(row, dict) else None
                if not isinstance(entity, dict):
                    continue
                content_field = schema.resolve("content")
                content = str(entity.get(content_field) or "") if content_field else ""
                if not content:
                    continue  # 没有正文的命中无法作为答案来源，直接丢弃而不是猜造
                hits.append(KnowledgeSearchService._hit_from_row(
                    entity, score=float(row.get("distance") or 0.0), schema=schema))
        return hits