AI260626/group_fqcd_jr
13
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(knowledge): 检索字段名改为运行时探测，两套集合 schema 都能跑

Browse Source 
评审意见 §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 各跑一遍。
This commit is contained in:
qyqy
2026-09-11 18:42:42 +08:00
parent e342670130
commit 5f82ac5107
4 changed files with 689 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/service/knowledge_search_service.py
+113 -85
View File
@@ -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