"""知识集合的**运行时字段探测**:把"逻辑字段名"解析成该集合真实的物理字段名。 ## 为什么需要它 同一批集合名(`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