Files
group_fqcd_jr/app/service/knowledge_search_service.py
qyqy 5f82ac5107 feat(knowledge): 检索字段名改为运行时探测,两套集合 schema 都能跑
评审意见 §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 各跑一遍。
2026-09-11 18:42:42 +08:00

479 lines
23 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""知识库检索:把客户问题向量化后在三个知识集合里检索(只读)。
为什么不复用记忆那套 `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_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。
#
# 背景(实测):同一批集合名在不同环境下可能是两套 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 = ""
@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
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,
*,
collections: Sequence[str] | None = None,
top_k: int = 5,
include_internal: bool = False,
) -> KnowledgeSearchOutcome:
"""检索知识库。
`include_internal=False`(默认)时在 Milvus 侧就过滤掉 `visibility=internal` 的块,
内部资料不进入面向客户的答案——这是检索层的硬隔离,不依赖提示词约束。
"""
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 = 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(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)
if best_vector_score < VECTOR_CONFIDENT_SCORE:
collected.extend(
self._product_keyword_hits(client, schemas, text, 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
)
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)
# "整节块"= **有行级子块挂在它下面**的块。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:
"""两段文本的最长公共子串长度。
用最长公共子串而不是分词:知识库标题是「南方科技有限公司 个人理财产品手册 ·
2.1 南方季季盈90天」这种没有词边界的长串,任何分词器都得先养一份自定义词典,
而词典会和手册一起过期。子串匹配不需要词典,手册改版也不会失效。
"""
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 _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"),
)
@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