feat(knowledge): 知识检索增加产品名的字面兜底召回
问题：客户问「季季盈90天的起投金额是多少」会被引导到人工客服，而知识库里明明有答案。 实测根因不是阈值拍错了，而是专有名词在 embedding 空间里不占优势——该问句的向量 top1 只有 0.6291，够不到 0.75 硬门槛，只能靠与次优的差值勉强通过；而同一次查询用 title like "%季季盈%" 是唯一命中 PROD-007。既然客户已经说出了产品名，就不该再赌相似度。 做法（三条边界都是实测逼出来的，不是设想）： 1. 只对产品集合做字面匹配。客户问「季季盈90天的起投金额是多少」与通用 FAQ 标题 「基金起投金额是多少?」有 7 个字连续重合；把 FAQ 纳入字面匹配会让它和真正的产品块 一起拿到满分、差距归零，反而又退化成"转人工"。 2. 字面命中只在向量结果不够确定时采用。客户问「基金赎回几天到账」时向量已给出正确答案 （FAQ-0016 得 0.8060），但手册章节标题「5.2 基金赎回流程」与问句也有 4 个字连续重合， 无条件采纳会把"操作步骤"顶掉客户真正问的"到账时间"。 3. 重叠门槛取 6 字而不是 4 字："基金赎回"这类业务动作词正好 4 字，会骗过 4 字门槛； 产品名（"南方季季盈90天"）更长，6 字能同时保住产品名、挡住动作词。 未改动任何转人工判定阈值；VECTOR_CONFIDENT_SCORE 与 Agent 的 HIGH_SCORE 由单测锁定一致， 避免两处各自漂移出"谁都答不出来"的死角。 验证：季季盈类问法由"转人工"变为正确答出，基金赎回问法仍答 FAQ-0016； ruff / mypy(113 文件) / 453 unit+contract / 29 integration 全绿。
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
"""知识检索的「字面兜底召回」单元测试。
|
||||
|
||||
用例全部来自实测，不是设想：
|
||||
|
||||
1. 客户问「季季盈90天的起投金额是多少」，纯向量 top1 只有 0.6291，够不到 0.75 门槛；
|
||||
而它与产品块标题的最长公共子串是 6 个字（就是产品名本身），字面匹配能唯一锁定。
|
||||
2. 客户问「基金赎回几天到账」，纯向量 top1 是 0.8060 的**正确答案**，但它与手册章节
|
||||
标题「5.2 基金赎回流程」也有 4 个字连续重合——所以字面匹配必须同时满足
|
||||
"重叠够长（6 字）"与"向量不够确定"两个条件，否则会把已经答对的题顶掉。
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app.service.knowledge_search_service import (
|
||||
MIN_KEYWORD_OVERLAP,
|
||||
PRODUCT_COLLECTION,
|
||||
VECTOR_CONFIDENT_SCORE,
|
||||
KnowledgeSearchService,
|
||||
)
|
||||
|
||||
PRODUCT_TITLE = "南方科技有限公司 个人理财产品手册 · 二、银行理财产品 · 2.1 南方季季盈90天"
|
||||
FLOW_TITLE = "南方科技有限公司 个人理财产品手册 · 五、申购赎回操作流程 · 5.2 基金赎回流程"
|
||||
|
||||
|
||||
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]:
|
||||
"""构造 pymilvus 的 `{distance, entity}` 行（`search()` 的返回形状）。"""
|
||||
return {
|
||||
"distance": score,
|
||||
"entity": {
|
||||
"doc_id": doc_id, "title": title, "content": content,
|
||||
"source_file": "x.md", "visibility": "public", "doc_no": "",
|
||||
"version": "", "chapter": "",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _flat(doc_id: str, title: str, content: str = "正文") -> dict[str, Any]:
|
||||
"""构造 pymilvus 的扁平行（`query()` 的返回形状，字段直接挂在顶层）。
|
||||
|
||||
与 `search()` 的 `{distance, entity}` 不是同一种形状，所以这里刻意分成两个构造函数：
|
||||
本文件第一版把 `_row` 用在了标量查询上，于是业务代码解析出空 content 并跳过它，
|
||||
测试红了一次——是测试写错，不是业务代码有问题（业务代码"没有正文就不作答"是对的）。
|
||||
"""
|
||||
return {
|
||||
"doc_id": doc_id, "title": title, "content": content,
|
||||
"source_file": "x.md", "visibility": "public", "doc_no": "",
|
||||
"version": "", "chapter": "",
|
||||
}
|
||||
|
||||
|
||||
class FakeClient:
|
||||
"""向量检索与标量查询的替身；记录 query 调用次数以便断言"有没有走字面匹配"。"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
vector_rows: list[dict[str, Any]],
|
||||
product_titles: list[tuple[str, str]],
|
||||
product_rows: dict[str, dict[str, Any]],
|
||||
) -> None:
|
||||
self._vector_rows = vector_rows
|
||||
self._product_titles = product_titles
|
||||
self._product_rows = product_rows
|
||||
self.query_calls = 0
|
||||
|
||||
def search(self, **kwargs: Any) -> Any:
|
||||
return [self._vector_rows]
|
||||
|
||||
def query(self, **kwargs: Any) -> Any:
|
||||
self.query_calls += 1
|
||||
if "content" not in kwargs.get("output_fields", []):
|
||||
return [{"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]
|
||||
|
||||
|
||||
def _service(client: FakeClient) -> KnowledgeSearchService:
|
||||
return KnowledgeSearchService(client, _embed, collections=[PRODUCT_COLLECTION])
|
||||
|
||||
|
||||
def test_overlap_length_matches_measured_facts() -> None:
|
||||
"""锁定实测到的两段重叠长度，防止有人把阈值当成"拍脑袋的数"随手改掉。"""
|
||||
overlap = KnowledgeSearchService._overlap_length
|
||||
assert overlap("季季盈90天的起投金额是多少", PRODUCT_TITLE) == 6
|
||||
assert overlap("基金赎回几天到账", FLOW_TITLE) == 4
|
||||
assert overlap("客户想了解开户材料", PRODUCT_TITLE) == 0
|
||||
assert overlap("", PRODUCT_TITLE) == 0
|
||||
# 4 字的业务动作词必须被 6 字门槛挡在外面
|
||||
assert overlap("基金赎回几天到账", FLOW_TITLE) < MIN_KEYWORD_OVERLAP
|
||||
|
||||
|
||||
def test_gate_value_matches_agent_high_score() -> None:
|
||||
"""检索层的"向量够确定了"门槛与客服 Agent 的高置信门槛必须一致。
|
||||
|
||||
两处各自漂移的话，会出现"Agent 认为不够确定要转人工，检索层却认为够确定不给兜底"
|
||||
这种谁都答不出来的死角。
|
||||
"""
|
||||
from app.service.agent.implementations.customer_service import HIGH_SCORE
|
||||
|
||||
assert VECTOR_CONFIDENT_SCORE == HIGH_SCORE
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_literal_match_rescues_weak_vector_result() -> None:
|
||||
"""向量给不出高置信答案时，字面命中的产品块以确定分胜出（季季盈实测）。"""
|
||||
client = FakeClient(
|
||||
vector_rows=[_row("PROD-901", "某无关章节", 0.62)],
|
||||
product_titles=[("PROD-007", PRODUCT_TITLE)],
|
||||
product_rows={"PROD-007": _flat("PROD-007", PRODUCT_TITLE, "产品正文")},
|
||||
)
|
||||
outcome = await _service(client).search("季季盈90天的起投金额是多少")
|
||||
|
||||
assert outcome.hits[0].doc_id == "PROD-007"
|
||||
assert outcome.hits[0].score == 1.0
|
||||
assert client.query_calls == 2 # 先取标题表，再取命中块的正文
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_literal_match_stays_out_when_vector_is_confident() -> None:
|
||||
"""向量已给出高置信答案时，字面匹配不得介入（"基金赎回流程"实测反例）。"""
|
||||
client = FakeClient(
|
||||
vector_rows=[_row("FAQ-0016", "基金赎回到账需要多长时间?", 0.806)],
|
||||
product_titles=[("PROD-015", FLOW_TITLE)],
|
||||
product_rows={"PROD-015": _flat("PROD-015", FLOW_TITLE, "操作步骤")},
|
||||
)
|
||||
outcome = await _service(client).search("基金赎回几天到账")
|
||||
|
||||
assert outcome.hits[0].doc_id == "FAQ-0016"
|
||||
assert client.query_calls == 0 # 一次标量查询都不该发生
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_without_query_support_degrades_silently() -> None:
|
||||
"""客户端（如精简替身）不支持标量查询时，字面匹配静默跳过，不影响向量召回。"""
|
||||
|
||||
class NoQueryClient:
|
||||
def search(self, **kwargs: Any) -> Any:
|
||||
return [[_row("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
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_literal_lookup_failure_does_not_break_search() -> None:
|
||||
"""标量查询抛异常时字面匹配返回空，向量结果照常返回。"""
|
||||
|
||||
class BrokenQueryClient(FakeClient):
|
||||
def query(self, **kwargs: Any) -> Any:
|
||||
raise RuntimeError("milvus 标量查询挂了")
|
||||
|
||||
client = BrokenQueryClient(
|
||||
vector_rows=[_row("PROD-007", PRODUCT_TITLE, 0.62)],
|
||||
product_titles=[("PROD-007", PRODUCT_TITLE)],
|
||||
product_rows={},
|
||||
)
|
||||
outcome = await _service(client).search("季季盈90天的起投金额是多少")
|
||||
|
||||
assert [hit.doc_id for hit in outcome.hits] == ["PROD-007"]
|
||||
assert outcome.degraded is False
|
||||
Reference in New Issue
Block a user