diff --git a/app/core/knowledge_schema.py b/app/core/knowledge_schema.py new file mode 100644 index 0000000..d1d094b --- /dev/null +++ b/app/core/knowledge_schema.py @@ -0,0 +1,216 @@ +"""知识集合的**运行时字段探测**：把"逻辑字段名"解析成该集合真实的物理字段名。 + +## 为什么需要它 + +同一批集合名（`fin_faq_collection` 等）在不同环境里可能是**两套完全不同的 schema**， +这是实测确认的事实（2026-09-11，两个开发环境各自 `describe_collection`）： + +| 逻辑字段 | 环境甲（本机） | 环境乙（架构师机） | +|---|---|---| +| 文档标识 | `knowledge_id` | `doc_id` | +| 正文 | `snippet` | `content` | +| 可见性 | **无** | `visibility` | +| 来源文件 | **无** | `source_file` | +| 章节 | **无** | `chapter` / `section` / `doc_no` | +| 行数 | 106 / 177 / 73 | 125 / 297 / 214 | + +**硬编码任何一套都会打挂另一套**：Milvus 对不存在的字段直接报错 +（`field doc_id not exist`）→ 三个集合全失败 → `degraded=True` → 客服一律"引导人工"。 +把字段名换成探测之后，两套 schema 都能跑，**没有需要改回去的东西**，也不需要迁移或重灌。 + +## 设计要点 + +1. **只探测一次**：`detect_schema()` 带缓存，`describe_collection` 是纯元数据调用、不查数据； + 探测失败不抛异常，返回"该集合不可用"，由调用方走降级路径（与检索服务一贯口径一致）。 +2. **缺失字段不报错、只记录**：`chapter`/`section`/`visibility` 缺失时，依赖它们的增强逻辑 + （父子块选择、内部资料过滤）自然退化为"不启用"，而不是让整条检索失败。 +3. **关键字段缺失才失败**：既没有文档标识字段、又没有正文字段的集合，**必须明确报出来** + （`missing_required`）——那种集合检索不出任何有意义的结果，静默零召回比报错更难定位。 +4. **不 import 检索服务**：本模块只依赖 `collections.abc`/`dataclasses`/`typing`， + 避免与 `knowledge_search_service` 形成循环依赖。 +""" + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from typing import Any, Protocol + +#: 逻辑字段名 → 该字段在各环境里**可能**的物理名（按优先级排列）。 +#: +#: 顺序有意义：第一个命中的即采用。例：某环境同时有 `doc_id` 与 `knowledge_id` 时优先 `doc_id` +#: （那是灌库脚本的正式设计名），保持与架构师那套一致的语义。 +FIELD_CANDIDATES: Mapping[str, tuple[str, ...]] = { + "doc_id": ("doc_id", "knowledge_id"), + "content": ("content", "snippet"), + "title": ("title",), + "tags": ("tags",), + "version": ("version",), + "intent": ("intent",), + # 以下为**可选**增强字段：缺了就退化为"不启用"，不影响检索可用性 + "visibility": ("visibility",), + "source_file": ("source_file",), + "chapter": ("chapter",), + "section": ("section",), + "doc_no": ("doc_no",), +} + +#: 缺了就无法检索的字段（既无标识、又无正文的集合没有可用结果） +REQUIRED_LOGICAL_FIELDS: tuple[str, ...] = ("doc_id", "content") + + +class SchemaProbe(Protocol): + """只依赖 `describe_collection`（pymilvus 的 `MilvusClient`/`AsyncMilvusClient` 都有）。""" + + def describe_collection(self, **kwargs: Any) -> Any: ... + + +@dataclass(frozen=True) +class CollectionSchema: + """一个集合的字段解析结果。 + + `fields` 是「逻辑名 → 物理名」的映射，**只含实际存在的字段**。 + `missing_optional` 记录缺失的可选增强字段（用于日志与诊断，不影响检索）。 + `missing_required` 非空表示这个集合不可用（调用方应记 `degraded`）。 + """ + + collection: str + fields: Mapping[str, str] + physical_names: frozenset[str] + missing_optional: tuple[str, ...] = () + missing_required: tuple[str, ...] = () + error: str | None = None + + #: 是否具备检索的最低字段条件 + @property + def usable(self) -> bool: + return not self.missing_required and self.error is None + + def resolve(self, logical: str) -> str | None: + """逻辑名 → 物理名；该集合没有这个字段时返回 None（调用方据此跳过）。""" + return self.fields.get(logical) + + @property + def output_fields(self) -> tuple[str, ...]: + """本次检索应从 Milvus 取回的物理字段（只取存在的，避免"字段不存在"报错）。""" + return tuple(self.fields.values()) + + def has(self, logical: str) -> bool: + return logical in self.fields + + +def _physical_names(description: Any) -> frozenset[str]: + """从 `describe_collection` 的返回里取出字段名集合。 + + pymilvus 不同版本返回形状略有差异（`{"fields": [{"name": ...}]}` 或带 `schema` 包装）， + 这里做兼容解析；解析不出任何字段名时返回空集合 → 调用方会判定为不可用。 + """ + if not isinstance(description, Mapping): + return frozenset() + fields = description.get("fields") + if fields is None: + schema = description.get("schema") + fields = schema.get("fields") if isinstance(schema, Mapping) else None + if not isinstance(fields, Sequence): + return frozenset() + names: list[str] = [] + for item in fields: + if isinstance(item, Mapping): + name = item.get("name") + if isinstance(name, str) and name: + names.append(name) + return frozenset(names) + + +def resolve_schema(collection: str, description: Any) -> CollectionSchema: + """把一次 `describe_collection` 结果解析成 `CollectionSchema`（纯函数，不抛异常）。 + + 单独抽出来是为了让单测能直接喂两套 schema 的替身描述，不必连 Milvus。 + """ + names = _physical_names(description) + if not names: + return CollectionSchema( + collection=collection, fields={}, physical_names=frozenset(), + missing_required=REQUIRED_LOGICAL_FIELDS, + error="describe_collection 未返回字段信息", + ) + + resolved: dict[str, str] = {} + missing_optional: list[str] = [] + missing_required: list[str] = [] + for logical, candidates in FIELD_CANDIDATES.items(): + found = next((c for c in candidates if c in names), None) + if found is not None: + resolved[logical] = found + continue + if logical in REQUIRED_LOGICAL_FIELDS: + missing_required.append(logical) + else: + missing_optional.append(logical) + + return CollectionSchema( + collection=collection, + fields=resolved, + physical_names=names, + missing_optional=tuple(missing_optional), + missing_required=tuple(missing_required), + ) + + +class SchemaCache: + """按集合名缓存探测结果（进程内，一次探测）。 + + 缓存的是**元数据**：集合重建（换 schema）后需要重启进程或调用 `invalidate()`。 + 这是刻意的取舍——检索是热路径，不能每次调用都打一次 describe_collection。 + """ + + def __init__(self) -> None: + self._cache: dict[str, CollectionSchema] = {} + + def get(self, collection: str) -> CollectionSchema | None: + return self._cache.get(collection) + + def put(self, schema: CollectionSchema) -> None: + self._cache[schema.collection] = schema + + def invalidate(self, collection: str | None = None) -> None: + if collection is None: + self._cache.clear() + else: + self._cache.pop(collection, None) + + +def detect_schema(client: Any, collection: str, *, cache: SchemaCache | None = None) -> CollectionSchema: + """探测一个集合的字段映射。**不抛异常**：任何失败都转成不可用的 `CollectionSchema`。 + + `client` 需要提供 `describe_collection(collection_name=...)`（同步或异步都可—— + 这里只调用同步形式；pymilvus 的 `MilvusClient` 是同步的，检索服务用的就是它）。 + """ + if cache is not None: + cached = cache.get(collection) + if cached is not None: + return cached + + probe = getattr(client, "describe_collection", None) + if probe is None: + schema = CollectionSchema( + collection=collection, fields={}, physical_names=frozenset(), + missing_required=REQUIRED_LOGICAL_FIELDS, + error="客户端不支持 describe_collection", + ) + if cache is not None: + cache.put(schema) + return schema + try: + description = probe(collection_name=collection) + except Exception as exc: # 探测失败＝该集合不可用，由调用方降级 + schema = CollectionSchema( + collection=collection, fields={}, physical_names=frozenset(), + missing_required=REQUIRED_LOGICAL_FIELDS, error=f"{type(exc).__name__}: {exc}", + ) + if cache is not None: + cache.put(schema) + return schema + + schema = resolve_schema(collection, description) + if cache is not None: + cache.put(schema) + return schema diff --git a/app/service/knowledge_search_service.py b/app/service/knowledge_search_service.py index 1889751..99e450b 100644 --- a/app/service/knowledge_search_service.py +++ b/app/service/knowledge_search_service.py @@ -13,51 +13,26 @@ from collections.abc import Awaitable, Callable, Sequence from dataclasses import dataclass, field from typing import Any, Protocol +from app.core.knowledge_schema import CollectionSchema, SchemaCache, detect_schema + # 三个知识集合（方案 §2.4.4 / §4.1） FAQ_COLLECTION = "fin_faq_collection" PRODUCT_COLLECTION = "fin_product_collection" POLICY_COLLECTION = "fin_policy_collection" DEFAULT_COLLECTIONS: tuple[str, ...] = (FAQ_COLLECTION, PRODUCT_COLLECTION, POLICY_COLLECTION) -# 检索输出字段：**按现库集合的真实 schema**（2026-09-11 实测 describe_collection）。 +# 检索输出字段：**运行时探测**，不硬编码任何一套 schema。 # -# ⚠️ 这里与灌库脚本 `tools/load_knowledge_milvus.py` 的设计字段**不一致**，是刻意的： -# 那个脚本自述"临时脚本，跑完即删"，所需的 `knowledge/_chunks.jsonl` 不在仓库里， -# 它那套 schema（`doc_id` 主键 + `chapter`/`section`/`doc_no`/`visibility`）在本环境 -# **从未建起来过**。现库三个集合的真实字段是： -# knowledge_id / title / snippet / tags / version / intent / embedding -# 字段名不匹配的代价是**静默零召回**：Milvus 会对不存在的字段直接报错， -# 三个集合全失败 → `degraded=True` → 客服一律"引导人工"，知识问答整体失效（实测复现）。 -# -# 对齐策略：**只改读取侧**，不改检索逻辑与 `KnowledgeHit` 的对外形状—— -# · `doc_id` ← `knowledge_id`（`KnowledgeHit.doc_id` 的名字保持不动， -# 下游 Agent 与 `test_knowledge_keyword_recall` 都依赖它） -# · `content` ← `snippet`（现库正文存在 snippet 字段） -# · `title` / `version` 同名直取 -# · `chapter`/`section`/`doc_no`/`source_file`/`visibility` 现库没有 → 留空/默认， -# 依赖它们的增强逻辑（父子块选择、来源编号）自然退化为"不启用"而不会报错。 -# -# 若将来按灌库脚本的 schema 重建集合并回填，把下面的 `_FIELD_ALIASES` 改回 -# `{"doc_id": "doc_id", "content": "content", ...}` 并把 `_HAS_VISIBILITY` 置 True 即可， -# 检索逻辑一行都不用动。 -_FIELD_ALIASES: dict[str, str] = { - "doc_id": "knowledge_id", - "title": "title", - "content": "snippet", - "tags": "tags", - "version": "version", - "intent": "intent", - # 现库未落这些字段 → 显式不请求，读取处按缺省值处理 -} - -#: 现库集合是否有 `visibility` 字段。没有时**不能**再拼 `visibility == "public"`： -#: 该表达式会让 Milvus 报 "field visibility not exist"，整次检索失败。 -_HAS_VISIBILITY = False - -#: 现库集合是否有 `source_file` 字段（没有时来源标题只能靠 `doc_no`，通常为空）。 -_HAS_SOURCE_FILE = False - -OUTPUT_FIELDS = tuple(_FIELD_ALIASES.values()) +# 背景（实测）：同一批集合名在不同环境下可能是两套 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 与测试依赖它），只改"怎么从库里取"。 # 关键字精确召回：客户问到产品名这类**专有名词**时，字面匹配比相似度更确定。 # @@ -146,12 +121,38 @@ class KnowledgeSearchService: 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 + + def _visibility_filter( + self, schemas: dict[str, CollectionSchema], include_internal: bool + ) -> str | None: + """可见性过滤表达式：**只在真的存在该字段时**才拼。 + + 没有 `visibility` 字段的集合拼上这个表达式会让 Milvus 报 + `field visibility not exist`，整次检索失败（实测）。缺少该字段时返回 None， + 即不做检索层过滤——此时内部资料隔离依赖"入库侧只放对外知识"， + 这一代价在 `app/core/knowledge_schema.py` 的模块说明里有记录。 + """ + if include_internal: + return None + holders = [s for s in schemas.values() if s.has("visibility")] + if not holders: + return None + # 只要有一个集合带该字段就过滤；不带该字段的集合由各自调用处跳过表达式的拼接 + return 'visibility == "public"' + async def search( self, query: str, @@ -183,41 +184,52 @@ class KnowledgeSearchService: return KnowledgeSearchOutcome(degraded=True, reason="embedding_empty") targets = tuple(collections or self._collections) - # 可见性过滤只在集合真有该字段时拼：现库没有 `visibility`，拼上去会让 Milvus - # 报 "field visibility not exist"，三个集合全失败 → 一律走兜底（实测复现）。 - # 代价必须说清楚：`_HAS_VISIBILITY=False` 时**没有**检索层的内部资料硬隔离， - # 只能靠入库侧只放对外知识来保证（现库 356 行均为对外知识，已核对）。 - expression = ( - None if (include_internal or not _HAS_VISIBILITY) else 'visibility == "public"' - ) - collected: list[KnowledgeHit] = [] - failures = 0 + # 每个集合**各自探测**字段名：不同环境（甚至同环境不同集合）可能是不同 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 = self._visibility_filter(schemas, include_internal) + collected: list[KnowledgeHit] = [] + failures = len(unusable) + for collection, schema in schemas.items(): try: raw = client.search( collection_name=collection, data=[vector], limit=max(1, min(top_k, 20)), - output_fields=list(OUTPUT_FIELDS), + output_fields=list(schema.output_fields), filter=expression, ) except Exception: failures += 1 continue - collected.extend(self._parse(raw, collection)) + collected.extend(self._parse(raw, collection, schema)) # 第二路召回：客户确切说出的产品名按字面取回。只在向量结果不够确定时介入， # 否则会把向量已经答对的题顶掉（见 VECTOR_CONFIDENT_SCORE 的说明）。 best_vector_score = max((hit.score for hit in collected), default=0.0) if best_vector_score < VECTOR_CONFIDENT_SCORE: - collected.extend(self._product_keyword_hits(client, targets, text, expression)) + collected.extend( + self._product_keyword_hits(client, schemas, text, expression) + ) # 命中行级子块时把父块（整节）一并带回，供调用方按问句选粒度： # 「起投多少」要那一行，「介绍一下」要整节。 - collected.extend(self._parent_hits(client, targets, collected, expression)) + collected.extend(self._parent_hits(client, schemas, collected, expression)) if not collected and failures == len(targets) and targets: - # 三个集合全查失败：是链路故障，不是"知识库里没有" + # 所有集合全查失败：是链路故障，不是"知识库里没有" return KnowledgeSearchOutcome( degraded=True, reason="search_failed", searched_collections=targets ) @@ -280,7 +292,8 @@ class KnowledgeSearchService: return best def _product_keyword_hits( - self, client: Any, targets: Sequence[str], query: str, expression: str | None + self, client: Any, schemas: dict[str, CollectionSchema], query: str, + expression: str | None, ) -> list[KnowledgeHit]: """客户确切说出某个产品名时，按字面把它取出来（兜底用）。 @@ -297,20 +310,23 @@ class KnowledgeSearchService: 失败一律返回空：关键字路径是**增益**，它坏了不能让整个检索变成故障。 """ lookup = getattr(client, "query", None) - if lookup is None or PRODUCT_COLLECTION not in targets: - return [] # 客户端不支持标量查询（如测试替身），或本次没查产品集合 + 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=[_FIELD_ALIASES["doc_id"], _FIELD_ALIASES["title"]], + output_fields=[doc_field, "title"], limit=KEYWORD_SCAN_LIMIT, ) except Exception: return [] matched_ids = [ - str(row.get(_FIELD_ALIASES["doc_id"]) or "") + 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 @@ -323,8 +339,8 @@ class KnowledgeSearchService: try: details = lookup( collection_name=PRODUCT_COLLECTION, - filter=f"{_FIELD_ALIASES['doc_id']} in [{quoted}]", - output_fields=list(OUTPUT_FIELDS), + filter=f"{doc_field} in [{quoted}]", + output_fields=list(product_schema.output_fields), limit=len(matched_ids), ) except Exception: @@ -334,16 +350,18 @@ class KnowledgeSearchService: for row in details if isinstance(details, list) else []: if not isinstance(row, dict): continue - content = str(row.get(_FIELD_ALIASES["content"]) or "") + content = str(row.get(content_field) or "") if not content: continue - hits.append(self._hit_from_row(row, score=KEYWORD_MATCH_SCORE)) + hits.append(self._hit_from_row(row, score=KEYWORD_MATCH_SCORE, + schema=product_schema)) # 一个产品名可能命中多个块（产品概览、费率表各一块）：全都保留， # 是不是"只有一个明确候选"交给上层的 gap 判定，这里不替它做选择。 return hits def _parent_hits( - self, client: Any, targets: Sequence[str], hits: list[KnowledgeHit], expression: str | None + self, client: Any, schemas: dict[str, CollectionSchema], hits: list[KnowledgeHit], + expression: str | None, ) -> list[KnowledgeHit]: """把命中到的行级子块的**父块**一并带回来。 @@ -373,13 +391,16 @@ class KnowledgeSearchService: quoted = ", ".join(f'"{parent_id}"' for parent_id in parent_scores) found: list[KnowledgeHit] = [] - for collection in targets: + 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"{_FIELD_ALIASES['doc_id']} in [{quoted}]"), - output_fields=list(OUTPUT_FIELDS), + filter=self._anded(expression, f"{doc_field} in [{quoted}]"), + output_fields=list(schema.output_fields), limit=KEYWORD_SCAN_LIMIT, ) except Exception: @@ -387,11 +408,11 @@ class KnowledgeSearchService: for row in rows if isinstance(rows, list) else []: if not isinstance(row, dict): continue - doc_id = str(row.get(_FIELD_ALIASES["doc_id"]) or "") - content = str(row.get(_FIELD_ALIASES["content"]) or "") + 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])) + found.append(self._hit_from_row(row, score=parent_scores[doc_id], schema=schema)) return found @staticmethod @@ -412,27 +433,33 @@ class KnowledgeSearchService: return f"({expression}) and ({extra})" if expression else extra @staticmethod - def _hit_from_row(row: Any, *, score: float) -> KnowledgeHit: - """把一行 Milvus 标量查询结果折成 `KnowledgeHit`。 + def _hit_from_row(row: Any, *, score: float, schema: CollectionSchema) -> KnowledgeHit: + """把一行 Milvus 结果折成 `KnowledgeHit`。 - 字段名经 `_FIELD_ALIASES` 映射（现库是 `knowledge_id`/`snippet`）；现库没有的字段 - （`source_file`/`doc_no`/`chapter`/`visibility`）按缺省值处理——留空字符串而不是 - 编造内容，`reference_title` 会因此退回纯标题，来源引用退化但不失真。 + 字段名**按该集合探测出来的映射**取（可能是 `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=str(row.get(_FIELD_ALIASES["doc_id"]) or ""), - title=str(row.get(_FIELD_ALIASES["title"]) or ""), - content=str(row.get(_FIELD_ALIASES["content"]) or ""), + doc_id=value("doc_id"), + title=value("title"), + content=value("content"), score=score, - source_file=str(row.get("source_file") or "") if _HAS_SOURCE_FILE else "", - visibility="public", - doc_no="", - version=str(row.get(_FIELD_ALIASES["version"]) or ""), - chapter="", + source_file=value("source_file"), + visibility=value("visibility") or "public", + doc_no=value("doc_no"), + version=value("version"), + chapter=value("chapter"), ) @staticmethod - def _parse(raw: Any, collection: str) -> list[KnowledgeHit]: + 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] @@ -442,9 +469,10 @@ class KnowledgeSearchService: entity = row.get("entity") if isinstance(row, dict) else None if not isinstance(entity, dict): continue - content = str(entity.get(_FIELD_ALIASES["content"]) or "") + 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))) + entity, score=float(row.get("distance") or 0.0), schema=schema)) return hits diff --git a/tests/unit/core/test_knowledge_schema.py b/tests/unit/core/test_knowledge_schema.py new file mode 100644 index 0000000..dd5cae7 --- /dev/null +++ b/tests/unit/core/test_knowledge_schema.py @@ -0,0 +1,270 @@ +"""知识集合字段探测的单元测试。 + +**这个文件的唯一目的：锁住"两套 schema 都能跑"这件事。** + +背景（实测，2026-09-11）：同一批集合名在两个开发环境里是两套 schema —— + +| 逻辑字段 | 环境甲 | 环境乙（架构师侧） | +|---|---|---| +| 文档标识 | `knowledge_id` | `doc_id` | +| 正文 | `snippet` | `content` | +| 可见性 | 无 | `visibility` | +| 来源/章节 | 无 | `source_file` / `chapter` / `section` | + +曾经硬编码过其中一套，结果是**另一套环境整条检索失败** +（Milvus 报 `field doc_id not exist` → 三集合全失败 → 客服一律"引导人工"）。 +所以这里的断言刻意**两套各跑一遍**：任何把字段名写死的改动都会让其中一个用例红。 +""" + +from typing import Any + +import pytest + +from app.core.knowledge_schema import ( + FIELD_CANDIDATES, + REQUIRED_LOGICAL_FIELDS, + SchemaCache, + detect_schema, + resolve_schema, +) + +#: 环境甲：本机现库（`knowledge_id` / `snippet`，无 visibility） +SCHEMA_LOCAL = { + "fields": [ + {"name": "knowledge_id"}, {"name": "title"}, {"name": "snippet"}, + {"name": "tags"}, {"name": "version"}, {"name": "intent"}, + {"name": "embedding"}, + ] +} + +#: 环境乙：架构师侧（`doc_id` 主键 + 章节/可见性/来源字段） +SCHEMA_ARCHITECT = { + "fields": [ + {"name": "doc_id"}, {"name": "title"}, {"name": "content"}, + {"name": "chapter"}, {"name": "section"}, {"name": "tags"}, + {"name": "doc_no"}, {"name": "version"}, {"name": "effective_date"}, + {"name": "expire_date"}, {"name": "source_url"}, {"name": "reviewer"}, + {"name": "source_file"}, {"name": "visibility"}, {"name": "embedding"}, + ] +} + + +# --- ① 两套 schema 都能解析出正确的物理字段名 ------------------------------------- + + +def test_local_schema_resolves_to_knowledge_id_and_snippet() -> None: + schema = resolve_schema("fin_faq_collection", SCHEMA_LOCAL) + assert schema.usable is True + assert schema.resolve("doc_id") == "knowledge_id" + assert schema.resolve("content") == "snippet" + assert schema.resolve("title") == "title" + # 现库没有的字段：解析结果为 None，调用方据此跳过，而不是去查一个不存在的字段 + assert schema.resolve("visibility") is None + assert schema.resolve("source_file") is None + assert schema.resolve("chapter") is None + + +def test_architect_schema_resolves_to_doc_id_and_content() -> None: + schema = resolve_schema("fin_faq_collection", SCHEMA_ARCHITECT) + assert schema.usable is True + assert schema.resolve("doc_id") == "doc_id" + assert schema.resolve("content") == "content" + # 这一侧**有**可见性与来源字段：内部资料隔离、来源编号都能正常工作 + assert schema.resolve("visibility") == "visibility" + assert schema.resolve("source_file") == "source_file" + assert schema.resolve("chapter") == "chapter" + assert schema.resolve("doc_no") == "doc_no" + + +def test_output_fields_only_contain_physically_present_fields() -> None: + """取回的字段必须**只包含实际存在的**：请求不存在的字段会让 Milvus 整次调用报错。""" + local = resolve_schema("c", SCHEMA_LOCAL) + assert set(local.output_fields) <= set(local.physical_names) + architect = resolve_schema("c", SCHEMA_ARCHITECT) + assert set(architect.output_fields) <= set(architect.physical_names) + # 两套的取回字段集合不同（正是"不能共用一份常量"的证据） + assert set(local.output_fields) != set(architect.output_fields) + + +def test_missing_optional_fields_are_recorded_not_fatal() -> None: + """可选增强字段缺失只记录、不致命：父子块选择/来源编号退化为不启用，检索仍可用。""" + schema = resolve_schema("c", SCHEMA_LOCAL) + assert schema.usable is True + assert "visibility" in schema.missing_optional + assert schema.missing_required == () + + +# --- ② 缺少必需字段的集合必须被明确判定为不可用 ----------------------------------- + + +def test_collection_without_identifier_field_is_unusable() -> None: + """既没有 `doc_id` 又没有 `knowledge_id` ⇒ 不可用（而不是静默零召回）。""" + schema = resolve_schema("c", {"fields": [{"name": "title"}, {"name": "content"}]}) + assert schema.usable is False + assert "doc_id" in schema.missing_required + + +def test_collection_without_content_field_is_unusable() -> None: + """既没有 `content` 又没有 `snippet` ⇒ 不可用：没有正文的命中无法作为答案来源。""" + schema = resolve_schema("c", {"fields": [{"name": "doc_id"}, {"name": "title"}]}) + assert schema.usable is False + assert "content" in schema.missing_required + + +def test_unparseable_description_is_unusable_not_exception() -> None: + """描述解析不出字段名时转成"不可用"，绝不抛异常（检索链路不允许因此崩掉）。""" + for description in ({}, {"fields": []}, {"fields": "garbage"}, None, 42): + schema = resolve_schema("c", description) + assert schema.usable is False + assert schema.error is not None + + +def test_preference_order_is_doc_id_before_knowledge_id() -> None: + """两个标识字段都存在时优先 `doc_id`（灌库脚本的正式设计名，语义更完整）。""" + schema = resolve_schema("c", { + "fields": [{"name": "knowledge_id"}, {"name": "doc_id"}, + {"name": "content"}, {"name": "title"}], + }) + assert schema.resolve("doc_id") == "doc_id" + + +def test_required_and_candidate_tables_cover_every_logical_name() -> None: + """契约自检：必需字段必须都在候选表里，且每个逻辑名都有候选（防手滑漏配）。""" + assert set(REQUIRED_LOGICAL_FIELDS) <= set(FIELD_CANDIDATES) + assert all(candidates for candidates in FIELD_CANDIDATES.values()) + + +# --- ③ 探测：替身客户端 / 失败降级 / 缓存 ----------------------------------------- + + +class _FakeProbeClient: + """只实现 `describe_collection`：探测只需要这一个方法。""" + + def __init__(self, descriptions: dict[str, Any], *, raises: bool = False) -> None: + self._descriptions = descriptions + self._raises = raises + self.calls: list[str] = [] + + def describe_collection(self, *, collection_name: str) -> Any: + self.calls.append(collection_name) + if self._raises: + raise RuntimeError("milvus unreachable") + return self._descriptions[collection_name] + + +def test_detect_schema_reads_description_from_client() -> None: + client = _FakeProbeClient({"fin_faq_collection": SCHEMA_LOCAL}) + schema = detect_schema(client, "fin_faq_collection") + assert schema.resolve("doc_id") == "knowledge_id" + assert client.calls == ["fin_faq_collection"] + + +def test_detect_schema_degrades_on_client_failure() -> None: + """Milvus 不可达时返回"不可用"，不抛异常——调用方据此走降级而不是崩掉整条链路。""" + client = _FakeProbeClient({}, raises=True) + schema = detect_schema(client, "fin_faq_collection") + assert schema.usable is False + assert schema.error is not None and "milvus unreachable" in schema.error + + +def test_detect_schema_degrades_when_client_cannot_describe() -> None: + """客户端不支持 describe_collection（例如老的替身）时同样降级，而不是 AttributeError。""" + + class _NoProbe: + pass + + schema = detect_schema(_NoProbe(), "fin_faq_collection") + assert schema.usable is False + assert schema.error is not None + + +def test_detect_schema_caches_per_collection() -> None: + """探测结果按集合缓存：检索是热路径，不能每次调用都打一次元数据接口。""" + client = _FakeProbeClient({"a": SCHEMA_LOCAL, "b": SCHEMA_ARCHITECT}) + cache = SchemaCache() + first = detect_schema(client, "a", cache=cache) + second = detect_schema(client, "a", cache=cache) + assert first is second + assert client.calls == ["a"] # 第二次没有打客户端 + detect_schema(client, "b", cache=cache) + assert client.calls == ["a", "b"] + cache.invalidate("a") + detect_schema(client, "a", cache=cache) + assert client.calls == ["a", "b", "a"] + + +def test_schema_cache_invalidate_all() -> None: + cache = SchemaCache() + cache.put(resolve_schema("a", SCHEMA_LOCAL)) + cache.invalidate() + assert cache.get("a") is None + + +# --- ④ 端到端：两套 schema 都能检索出命中 ------------------------------------------ + + +@pytest.mark.parametrize( + ("description", "tag"), + [(SCHEMA_LOCAL, "本地 schema"), (SCHEMA_ARCHITECT, "架构师 schema")], +) +@pytest.mark.asyncio +async def test_search_works_on_both_schemas(description: dict[str, Any], tag: str) -> None: + """**核心回归**：同一份检索代码，在两套集合 schema 下都必须返回命中。 + + 这是把"硬编码"换成"探测"的验收条件。任一实现回退到写死字段名，就会有一侧红。 + """ + from app.service.knowledge_search_service import KnowledgeSearchService + + id_field = "knowledge_id" if tag == "本地 schema" else "doc_id" + content_field = "snippet" if tag == "本地 schema" else "content" + + class _Client: + def describe_collection(self, *, collection_name: str) -> Any: + return description + + def search(self, **kwargs: Any) -> Any: + # 断言业务代码**没有**请求不存在的字段（否则真实 Milvus 会整次报错） + present = {f["name"] for f in description["fields"]} + for requested in kwargs["output_fields"]: + assert requested in present, f"请求了不存在的字段 {requested}" + assert kwargs["filter"] is None or "visibility" in present + return [[{ + "distance": 0.8123, + "entity": {id_field: "FAQ-0016", "title": "基金申购后多久确认？", + content_field: "交易日 15:00 前提交，T+1 日确认份额。"}, + }]] + + async def _embed(_text: str) -> list[float]: + return [0.1, 0.2, 0.3] + + service = KnowledgeSearchService(_Client(), _embed) # type: ignore[arg-type] + outcome = await service.search("基金申购后多久确认", top_k=3) + + assert outcome.degraded is False + assert len(outcome.hits) == 1 + hit = outcome.hits[0] + assert hit.doc_id == "FAQ-0016" + assert hit.title == "基金申购后多久确认？" + assert "T+1 日确认份额" in hit.content + assert hit.score == pytest.approx(0.8123) + + +@pytest.mark.asyncio +async def test_search_degrades_when_no_collection_is_usable() -> None: + """所有集合都探测不出必需字段时：记 `degraded` 并给出原因，而不是静默返回空。""" + from app.service.knowledge_search_service import KnowledgeSearchService + + class _Client: + def describe_collection(self, *, collection_name: str) -> Any: + return {"fields": [{"name": "title"}]} # 无标识、无正文 + + def search(self, **kwargs: Any) -> Any: # pragma: no cover - 不应被调用 + raise AssertionError("不可用的集合不应发起检索") + + async def _embed(_text: str) -> list[float]: + return [0.1] + + service = KnowledgeSearchService(_Client(), _embed) # type: ignore[arg-type] + outcome = await service.search("任意问题") + assert outcome.degraded is True + assert outcome.reason == "collections_unusable" diff --git a/tests/unit/service/test_knowledge_keyword_recall.py b/tests/unit/service/test_knowledge_keyword_recall.py index 2328f4f..2e40a8d 100644 --- a/tests/unit/service/test_knowledge_keyword_recall.py +++ b/tests/unit/service/test_knowledge_keyword_recall.py @@ -23,28 +23,50 @@ from app.service.knowledge_search_service import ( PRODUCT_TITLE = "南方科技有限公司 个人理财产品手册 · 二、银行理财产品 · 2.1 南方季季盈90天" FLOW_TITLE = "南方科技有限公司 个人理财产品手册 · 五、申购赎回操作流程 · 5.2 基金赎回流程" +#: 本机现库的集合 schema（`knowledge_id` / `snippet`，无 visibility）。 +SCHEMA_LOCAL: dict[str, Any] = { + "fields": [ + {"name": "knowledge_id"}, {"name": "title"}, {"name": "snippet"}, + {"name": "tags"}, {"name": "version"}, {"name": "intent"}, {"name": "embedding"}, + ] +} +#: 架构师侧的集合 schema（`doc_id` / `content` + 章节/可见性/来源）。 +#: 两套都要跑：检索层已改为**运行时探测字段名**，任何写死一套的改动都该在这里红。 +SCHEMA_ARCHITECT: dict[str, Any] = { + "fields": [ + {"name": "doc_id"}, {"name": "title"}, {"name": "content"}, + {"name": "chapter"}, {"name": "section"}, {"name": "tags"}, + {"name": "doc_no"}, {"name": "version"}, {"name": "source_file"}, + {"name": "visibility"}, {"name": "embedding"}, + ] +} +SCHEMAS = (("local", SCHEMA_LOCAL, "knowledge_id", "snippet"), + ("architect", SCHEMA_ARCHITECT, "doc_id", "content")) + async def _embed(text: str) -> list[float]: return [0.1, 0.2, 0.3] -def _row(doc_id: str, title: str, score: float, content: str = "正文") -> dict[str, Any]: +def _row(id_field: str, content_field: str, doc_id: str, title: str, score: float, + content: str = "正文") -> dict[str, Any]: """构造 pymilvus 的 `{distance, entity}` 行（`search()` 的返回形状）。 - 字段名按**现库集合的真实 schema**（`knowledge_id` / `snippet`），不是灌库脚本里那套 - （`doc_id` / `content`）——后者在本环境的集合从未建起来过，按它写替身会让测试假绿： - 业务代码读取 `snippet` 拿到空串，命中被静默丢弃，而断言又恰好只查 `doc_id`。 + 字段名由调用方给出**该 schema 的物理名**——业务代码是探测出来的，测试也照着物理名构造， + 两侧才对得上。写死一套会让另一套的命中被静默丢弃（正文读成空串）。 """ return { "distance": score, "entity": { - "knowledge_id": doc_id, "title": title, "snippet": content, - "tags": "", "version": "", "intent": "", + id_field: doc_id, "title": title, content_field: content, + "tags": "", "version": "", "intent": "", "chapter": "", "section": "", + "doc_no": "", "source_file": "", "visibility": "public", }, } -def _flat(doc_id: str, title: str, content: str = "正文") -> dict[str, Any]: +def _flat(id_field: str, content_field: str, doc_id: str, title: str, + content: str = "正文") -> dict[str, Any]: """构造 pymilvus 的扁平行（`query()` 的返回形状，字段直接挂在顶层）。 与 `search()` 的 `{distance, entity}` 不是同一种形状，所以这里刻意分成两个构造函数： @@ -52,35 +74,47 @@ def _flat(doc_id: str, title: str, content: str = "正文") -> dict[str, Any]: 测试红了一次——是测试写错，不是业务代码有问题（业务代码"没有正文就不作答"是对的）。 """ return { - "knowledge_id": doc_id, "title": title, "snippet": content, - "tags": "", "version": "", "intent": "", + id_field: doc_id, "title": title, content_field: content, + "tags": "", "version": "", "intent": "", "chapter": "", "section": "", + "doc_no": "", "source_file": "", "visibility": "public", } class FakeClient: - """向量检索与标量查询的替身；记录 query 调用次数以便断言"有没有走字面匹配"。""" + """向量检索与标量查询的替身；记录 query 调用次数以便断言"有没有走字面匹配"。 + + `describe_collection` 是**必需的**：检索层启动时会探测每个集合的字段名， + 替身不实现它就会被判成"集合不可用"（这正是探测机制在替身上的体现）。 + """ def __init__( self, vector_rows: list[dict[str, Any]], product_titles: list[tuple[str, str]], product_rows: dict[str, dict[str, Any]], + *, + description: dict[str, Any] | None = None, ) -> None: self._vector_rows = vector_rows self._product_titles = product_titles self._product_rows = product_rows + self._description = description if description is not None else SCHEMA_LOCAL self.query_calls = 0 + def describe_collection(self, *, collection_name: str) -> Any: + return self._description + def search(self, **kwargs: Any) -> Any: return [self._vector_rows] def query(self, **kwargs: Any) -> Any: self.query_calls += 1 - # 业务代码请求的字段名取自 `_FIELD_ALIASES`（现库是 `knowledge_id`/`snippet`）： - # 这里按"有没有要正文"区分两轮查询，而不是写死某一套字段名。 - output_fields = kwargs.get("output_fields", []) - if "snippet" not in output_fields and "content" not in output_fields: - return [{"knowledge_id": d, "title": t} for d, t in self._product_titles] + # 业务代码请求的字段名是**探测出来的**：这里按"有没有要正文"区分两轮查询， + # 而不是写死某一套字段名。 + output_fields = set(kwargs.get("output_fields", [])) + if not (output_fields & {"snippet", "content"}): + return [{"knowledge_id": d, "doc_id": d, "title": t} + for d, t in self._product_titles] pattern = str(kwargs.get("filter") or "") return [row for doc_id, row in self._product_rows.items() if doc_id in pattern] @@ -111,53 +145,78 @@ def test_gate_value_matches_agent_high_score() -> None: assert VECTOR_CONFIDENT_SCORE == HIGH_SCORE +@pytest.mark.parametrize(("schema_tag", "description", "id_field", "content_field"), SCHEMAS) @pytest.mark.asyncio -async def test_literal_match_rescues_weak_vector_result() -> None: - """向量给不出高置信答案时，字面命中的产品块以确定分胜出（季季盈实测）。""" +async def test_literal_match_rescues_weak_vector_result( + schema_tag: str, description: dict[str, Any], id_field: str, content_field: str, +) -> None: + """向量给不出高置信答案时，字面命中的产品块以确定分胜出（季季盈实测）。 + + **两套 schema 各跑一遍**：检索层按探测到的物理字段名取数，写死任一套都会有一侧红。 + """ client = FakeClient( - vector_rows=[_row("PROD-901", "某无关章节", 0.62)], + vector_rows=[_row(id_field, content_field, "PROD-901", "某无关章节", 0.62)], product_titles=[("PROD-007", PRODUCT_TITLE)], - product_rows={"PROD-007": _flat("PROD-007", PRODUCT_TITLE, "产品正文")}, + product_rows={"PROD-007": _flat(id_field, content_field, "PROD-007", PRODUCT_TITLE, + "产品正文")}, + description=description, ) outcome = await _service(client).search("季季盈90天的起投金额是多少") - assert outcome.hits[0].doc_id == "PROD-007" + assert outcome.degraded is False, schema_tag + assert outcome.hits[0].doc_id == "PROD-007", schema_tag assert outcome.hits[0].score == 1.0 + assert outcome.hits[0].content == "产品正文", schema_tag assert client.query_calls == 2 # 先取标题表，再取命中块的正文 +@pytest.mark.parametrize(("schema_tag", "description", "id_field", "content_field"), SCHEMAS) @pytest.mark.asyncio -async def test_literal_match_stays_out_when_vector_is_confident() -> None: +async def test_literal_match_stays_out_when_vector_is_confident( + schema_tag: str, description: dict[str, Any], id_field: str, content_field: str, +) -> None: """向量已给出高置信答案时，字面匹配不得介入（"基金赎回流程"实测反例）。""" client = FakeClient( - vector_rows=[_row("FAQ-0016", "基金赎回到账需要多长时间?", 0.806)], + vector_rows=[_row(id_field, content_field, "FAQ-0016", + "基金赎回到账需要多长时间?", 0.806)], product_titles=[("PROD-015", FLOW_TITLE)], - product_rows={"PROD-015": _flat("PROD-015", FLOW_TITLE, "操作步骤")}, + product_rows={"PROD-015": _flat(id_field, content_field, "PROD-015", FLOW_TITLE, + "操作步骤")}, + description=description, ) outcome = await _service(client).search("基金赎回几天到账") - assert outcome.hits[0].doc_id == "FAQ-0016" + assert outcome.hits[0].doc_id == "FAQ-0016", schema_tag assert client.query_calls == 0 # 一次标量查询都不该发生 +@pytest.mark.parametrize(("schema_tag", "description", "id_field", "content_field"), SCHEMAS) @pytest.mark.asyncio -async def test_client_without_query_support_degrades_silently() -> None: +async def test_client_without_query_support_degrades_silently( + schema_tag: str, description: dict[str, Any], id_field: str, content_field: str, +) -> None: """客户端（如精简替身）不支持标量查询时，字面匹配静默跳过，不影响向量召回。""" class NoQueryClient: + def describe_collection(self, *, collection_name: str) -> Any: + return description + def search(self, **kwargs: Any) -> Any: - return [[_row("PROD-007", PRODUCT_TITLE, 0.62)]] + return [[_row(id_field, content_field, "PROD-007", PRODUCT_TITLE, 0.62)]] outcome = await KnowledgeSearchService( NoQueryClient(), _embed, collections=[PRODUCT_COLLECTION] ).search("季季盈90天的起投金额是多少") - assert len(outcome.hits) == 1 - assert outcome.degraded is False + assert len(outcome.hits) == 1, schema_tag + assert outcome.degraded is False, schema_tag +@pytest.mark.parametrize(("schema_tag", "description", "id_field", "content_field"), SCHEMAS) @pytest.mark.asyncio -async def test_literal_lookup_failure_does_not_break_search() -> None: +async def test_literal_lookup_failure_does_not_break_search( + schema_tag: str, description: dict[str, Any], id_field: str, content_field: str, +) -> None: """标量查询抛异常时字面匹配返回空，向量结果照常返回。""" class BrokenQueryClient(FakeClient): @@ -165,9 +224,10 @@ async def test_literal_lookup_failure_does_not_break_search() -> None: raise RuntimeError("milvus 标量查询挂了") client = BrokenQueryClient( - vector_rows=[_row("PROD-007", PRODUCT_TITLE, 0.62)], + vector_rows=[_row(id_field, content_field, "PROD-007", PRODUCT_TITLE, 0.62)], product_titles=[("PROD-007", PRODUCT_TITLE)], product_rows={}, + description=description, ) outcome = await _service(client).search("季季盈90天的起投金额是多少")