qyqy
5f82ac5107
评审意见 §1.3 要求的修法。背景（双方实测共同确认）：同一批集合名在两个开发环境里是两套不同 schema： 我方：knowledge_id / snippet（无 visibility），行数 106/177/73 架构师：doc_id / content / chapter / section / doc_no / visibility，行数 125/297/214 上一轮我把字段名硬编码成我方那套，在架构师环境会让 Milvus 报 field doc_id not exist → 三集合全失败 → 客服一律转人工（反向亦然）。硬编码任一套都会打挂另一套。 改法（采纳评审建议）： - 新增 app/core/knowledge_schema.py：describe_collection → 逻辑名到物理名映射，按集合缓存； 缺必需字段的集合明确判为不可用并如实记 degraded，不静默零召回 - 检索服务改为逐集合探测：output_fields 只请求实际存在的字段；visibility 过滤有该字段才拼 - KnowledgeHit 对外形状不变，检索逻辑（字面召回/父子块/去重/置信判定）一行未改 测试：新增 17 个探测单测；架构师的关键词召回测试参数化为两套 schema 各跑一遍。
217 lines
8.8 KiB
Python
217 lines
8.8 KiB
Python
"""知识集合的**运行时字段探测**：把"逻辑字段名"解析成该集合真实的物理字段名。
|
||
|
||
## 为什么需要它
|
||
|
||
同一批集合名（`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
|