From 570493e71cb6f574dfba085d5aa5f371669f76f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Thu, 10 Sep 2026 22:15:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(knowledge):=20=E7=9F=A5=E8=AF=86=E6=A3=80?= =?UTF-8?q?=E7=B4=A2=E5=A2=9E=E5=8A=A0=E4=BA=A7=E5=93=81=E5=90=8D=E7=9A=84?= =?UTF-8?q?=E5=AD=97=E9=9D=A2=E5=85=9C=E5=BA=95=E5=8F=AC=E5=9B=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题：客户问「季季盈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 全绿。 --- app/service/knowledge_search_service.py | 127 +++++++++++++ .../service/test_knowledge_keyword_recall.py | 169 ++++++++++++++++++ 2 files changed, 296 insertions(+) create mode 100644 tests/unit/service/test_knowledge_keyword_recall.py diff --git a/app/service/knowledge_search_service.py b/app/service/knowledge_search_service.py index 61b208c..99ff282 100644 --- a/app/service/knowledge_search_service.py +++ b/app/service/knowledge_search_service.py @@ -25,6 +25,30 @@ OUTPUT_FIELDS = ( "tags", "doc_no", "version", "source_file", "visibility", ) +# 关键字精确召回：客户问到产品名这类**专有名词**时，字面匹配比相似度更确定。 +# +# 为什么需要：专有名词在 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 # 一次最多扫描多少条标题；知识库到上千块后应改为倒排索引 + +# 字面匹配只在向量结果**不够确定**时介入。 +# +# 这条门禁是实测逼出来的：客户问「基金赎回几天到账」，向量已给出正确答案 +# （FAQ-0016「基金赎回到账需要多长时间?」得 0.8060），但产品手册里的章节标题 +# 「五、申购赎回操作指南」与问句也有 6 个字连续重合，无条件字面匹配会把它顶到第一， +# 用操作步骤替换掉客户真正问的到账时间。所以字面匹配是**兜底**，不是优先。 +# +# 门槛必须与客服 Agent 的高置信门槛一致，tests/unit 有断言锁定两者相等。 +VECTOR_CONFIDENT_SCORE = 0.75 + class VectorSearcher(Protocol): """只依赖用到的两个方法，便于测试替身注入。""" @@ -138,6 +162,12 @@ class KnowledgeSearchService: continue collected.extend(self._parse(raw, collection)) + # 第二路召回：客户确切说出的产品名按字面取回。只在向量结果不够确定时介入， + # 否则会把向量已经答对的题顶掉（见 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)) + if not collected and failures == len(targets) and targets: # 三个集合全查失败：是链路故障，不是"知识库里没有" return KnowledgeSearchOutcome( @@ -160,6 +190,103 @@ class KnowledgeSearchService: 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, targets: Sequence[str], query: str, expression: str | None + ) -> list[KnowledgeHit]: + """客户确切说出某个产品名时，按字面把它取出来（兜底用）。 + + 两处边界都是实测逼出来的： + + 1. **只对产品集合做**。客户问「季季盈90天的起投金额是多少」，与通用 FAQ 标题 + 「基金起投金额是多少?」的最长公共子串有 7 个字；若把 FAQ 也纳入字面匹配， + 它会和真正的产品块一起拿到满分、差距归零，反而又退化成"转人工"。 + 2. **命中不能无条件优先**。产品手册的标题里不只有产品名，还有章节名：客户问 + 「基金赎回几天到账」时，「五、申购赎回操作指南」那一块与问句也有 6 个字连续 + 重合。所以本方法产出什么是一回事，是否采用由调用方按"向量是否已经足够确定" + 决定（见 VECTOR_CONFIDENT_SCORE）。 + + 失败一律返回空：关键字路径是**增益**，它坏了不能让整个检索变成故障。 + """ + lookup = getattr(client, "query", None) + if lookup is None or PRODUCT_COLLECTION not in targets: + return [] # 客户端不支持标量查询（如测试替身），或本次没查产品集合 + try: + rows = lookup( + collection_name=PRODUCT_COLLECTION, + filter=expression, + output_fields=["doc_id", "title"], + limit=KEYWORD_SCAN_LIMIT, + ) + except Exception: + return [] + + matched_ids = [ + str(row.get("doc_id") 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_id in [{quoted}]", + output_fields=list(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") or "") + if not content: + continue + hits.append(KnowledgeHit( + doc_id=str(row.get("doc_id") or ""), + title=str(row.get("title") or ""), + content=content, + score=KEYWORD_MATCH_SCORE, + source_file=str(row.get("source_file") or ""), + visibility=str(row.get("visibility") or "public"), + doc_no=str(row.get("doc_no") or ""), + version=str(row.get("version") or ""), + chapter=str(row.get("chapter") or ""), + )) + # 一个产品名可能命中多个块（产品概览、费率表各一块）：全都保留， + # 是不是"只有一个明确候选"交给上层的 gap 判定，这里不替它做选择。 + return hits + @staticmethod def _parse(raw: Any, collection: str) -> list[KnowledgeHit]: """把 pymilvus 的 `[[{id, distance, entity}]]` 折叠成命中列表（纯函数，不抛异常）。""" diff --git a/tests/unit/service/test_knowledge_keyword_recall.py b/tests/unit/service/test_knowledge_keyword_recall.py new file mode 100644 index 0000000..a164d67 --- /dev/null +++ b/tests/unit/service/test_knowledge_keyword_recall.py @@ -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