feat(knowledge): 产品知识拆到表格行级,并按问句选粒度

问题(客户实测反馈):同一会话里问「季季盈90天起投多少」和「那它风险高吗」,两次回答
**一模一样**——都是整个产品小节的表格。客户问的是风险,收到的是整张说明书,看起来像
客服没听懂问题。

根因是切分粒度:原来"一个叶子标题 = 一块",产品手册里就是整个产品小节(表格 + 说明)
成一块。这既让两个不同的问题命中同一块,也让整节几百字的向量成了"整节的混合语义",
与"起投多少"这种具体小问题相似度天然偏低(实测该问句向量 top1 仅 0.6291,够不到 0.75
硬门槛,只能靠与次优的差值勉强通过)。

改动三处:
1. 切分:Markdown 表格的每一行额外生成一个**自解释**的小块("南方季季盈90天:起投金额
   1万元"),挂在父块 doc_id 下(PROD-007-04),父块照旧保留。知识块 160 → 631。
   效果:该问句的命中分从 0.6291 升到 0.869,命中的正是"起投金额"那一行。
2. 检索:命中行级子块时把它的整节父块一并带回(分数按 0.9 折算),供调用方按问句选粒度。
   整节块保底占最后一个名额,且不参与 top1/top2 判定——实测它挤到第 2 位会把 gap 从
   0.090 压到 0.076,几乎跌破 0.07 的转人工门槛。
3. 客服:命中的是行级子块时,看问句与子块标签是否真的对得上——「起投多少」对「起投金额」
   对得上,用那一行;「介绍一下」对不上,换成整节。

过程中两次判据写错并已修正(都固化进了测试):用"含连字符"认子块时,整节块自己的编号
PROD-901 被误判成子块;用"不含两位数字后缀"认整节块时,FAQ 块全被误判成整节块排到后面,
把正确答案挤出 top1、害得「基金赎回几天到账」转人工。

验证:起投/管理费等字段问法给出聚焦的单行答案;"介绍一下"给出整节;FAQ 与政策问法不受
影响(换话题、指代追问等此前修好的场景复测通过);
ruff / mypy(113 文件) / 468 unit+contract / 29 integration 全绿。
This commit is contained in:
2026-09-10 22:29:23 +08:00
parent a6c09fa3c4
commit 4c2b147793
5 changed files with 769 additions and 13 deletions
@@ -15,6 +15,8 @@
|`chitchat` → 模型生成(提示词走发布配置)|其余与异常 → 引导人工客服
"""
from typing import Any
from app.core.contracts import (
AgentDefinition,
AgentRequest,
@@ -146,6 +148,9 @@ class CustomerServiceAgent(BaseAgent):
if not confident and not (score >= MID_SCORE and gap >= MIN_GAP):
return self._guide_to_human(f"置信度不足:score={score:.3f} gap={gap:.3f}")
# 命中的是行级子块时,先分辨客户问的是"某个字段"还是"整个产品":
# 子块让「起投多少」拿到聚焦答案,但「介绍一下」会被某一行抢答。
best = self._prefer_section(request.message, best, hits)
content = str(best.get("content") or "").strip()
if not content:
return self._guide_to_human("命中内容为空")
@@ -162,6 +167,35 @@ class CustomerServiceAgent(BaseAgent):
intent=self._classified_intent,
)
@staticmethod
def _prefer_section(message: str, best: dict[str, Any], hits: list[Any]) -> dict[str, Any]:
"""客户问整个产品时,用整节块替换掉抢答的那一行。
行级子块是为了让「起投多少」拿到聚焦答案,但「介绍一下」会被某一行抢答
(实测返回了"产品期限 90天封闭期",而客户要的是整个产品)。
判据不用问句分类器,而是看问句与子块标签是否真的对得上。标签取自子块的
section 末段("起投金额""风险等级"):「起投多少」含"起投"、「风险高吗」含"风险",
都算对得上;「介绍一下」与任何标签都不重合,说明客户要的是整节。
父块由检索层按子块分数的 0.9 折算后一并带回。万一没带回来就仍用子块——
宁可答得窄一点,也不要拿不相干的块去搪塞。
"""
doc_id = str(best.get("doc_id") or "")
# 末段恰为 2 位数字才是行级子块(PROD-007-04);整节块自己的编号形如 PROD-901,
# 用"含连字符"判断会把整节块误判成子块。
parent_id, _, tail = doc_id.rpartition("-")
if not (parent_id and tail.isdigit() and len(tail) == 2):
return best # 命中的本来就是整节
# 标签取自子块 title 的末段(title 由 " · " 连接),如"起投金额""风险等级"
label = str(best.get("title") or "").split(" · ")[-1].strip()
if label and label[:2] in message:
return best # 客户问的正是这个字段
for hit in hits:
if isinstance(hit, dict) and str(hit.get("doc_id") or "") == parent_id:
return hit
return best
# ---- 出口二:闲聊(提示词走发布配置) ----
async def _chitchat(self, request: AgentRequest) -> CoreResult:
@@ -214,22 +248,43 @@ class CustomerServiceAgent(BaseAgent):
# 短到这种程度的问题,自己通常不构成完整意图("起投多少""风险高吗")
_MIN_STANDALONE_CHARS = 8
@staticmethod
def _topic_of(answer: str) -> str:
"""从客服上一轮的回答里取出"这一轮在说哪个产品",取不出来就返回空串。
回答是我们自己组装的,只有两种形状,都取自知识块的字段:
- 行级块:`南方季季盈90天:起投金额 1万元`(首个冒号前就是产品名)
- 整节块:`### 2.1 南方季季盈90天`(首个标题行去掉 # 号)
取不出来时宁可返回空、退化成不带主语,也不要拿兜底话术当主语:
「抱歉,这个问题我暂时无法给出准确答复」这种句子拿去检索只会把问题带偏。
"""
first = next((line.strip() for line in answer.splitlines() if line.strip()), "")
first = first.lstrip("#").strip()
topic = first.split(":", 1)[0].strip() if ":" in first else first
if not topic or len(topic) > 20 or "," in topic or "。" in topic:
return ""
return topic
@classmethod
def _search_query(cls, request: AgentRequest) -> str:
"""构造交给检索的查询串。
只有当客户这一句**自己说不清楚**时,才把上文接进来。
什么时候带上文(实测决定,两条都不能少):
为什么不能无条件拼接(实测):同一会话先问「季季盈90天的起投金额是多少」,
再问「基金赎回几天到账」,若把上一轮问题拼进检索词,第二问会命中季季盈的产品块、
答出产品介绍——**答非所问**。在金融场景里这比"引导转人工"糟得多:客户问 A 得到 B
的答案,会直接失去对客服的信任,而"答不了"至少是诚实的。
1. **客户这一句自己说不清楚时**才带。同一会话先问「季季盈90天的起投金额是多少」、
再问「基金赎回几天到账」,若无条件带上文,第二问会命中季季盈的产品块、
答出产品介绍——**答非所问**。在金融场景里这比"引导转人工"糟得多。
2. **带上文时只带"在说哪个产品",不带上一轮的原话**。把上一轮整句拼进来会让
检索词语义变"宽",反而只能命中粗粒度的整节块:实测「季季盈90天起投多少
那它风险高吗」命中的是整个产品小节,客户问的"风险"完全没有被聚焦;
换成「南方季季盈90天 那它风险高吗」才命中"风险等级"那一行。
所以拼接只留给两种真正需要上文的情况:句子里有指代词,或短到不足以独立表达意图。
其余一律以当前问题检索,客户换话题就不会被上一轮拖回去。
换句话说:上文的作用是**补主语**,不是**补内容**。
只取客户的话、不取 Agent 自己的回答:把 Agent 的措辞也拼进来会让检索偏向自己
上一轮的说法,而客户的真实意图可能已经在下一句里被修正过。
只取客户的话、不取 Agent 自己的回答当内容:把 Agent 的措辞也拼进来会让检索偏向
自己上一轮的说法,而客户的真实意图可能已经在下一句里被修正过。这里从回答里取的
只有产品名这一个"主语",不是它的论述。
"""
message = request.message.strip()
needs_context = (
@@ -238,8 +293,13 @@ class CustomerServiceAgent(BaseAgent):
)
if not needs_context:
return message[:500]
recent_user = [turn.content for turn in request.history if turn.role == "user"][-2:]
return " ".join([*recent_user, message])[:500]
for turn in reversed(request.history):
if turn.role == "assistant":
topic = cls._topic_of(turn.content)
if topic:
return f"{topic} {message}"[:500]
break
return message[:500]
# ---- 出口三:引导人工客服(不做工单,只回话并留痕) ----
+104 -1
View File
@@ -39,6 +39,10 @@ KEYWORD_MATCH_SCORE = 1.0 # 字面命中的确定分,压过任何相似度分
MIN_KEYWORD_OVERLAP = 6
KEYWORD_SCAN_LIMIT = 500 # 一次最多扫描多少条标题;知识库到上千块后应改为倒排索引
# 行级子块命中时,其父块(整节)按子块分数的这个比例一并返回:排在子块之后,
# 既不抢"起投多少"这类聚焦答案,又不至于把差距压到转人工门槛之下。
PARENT_SCORE_RATIO = 0.9
# 字面匹配只在向量结果**不够确定**时介入。
#
# 这条门禁是实测逼出来的:客户问「基金赎回几天到账」,向量已给出正确答案
@@ -168,6 +172,10 @@ class KnowledgeSearchService:
if best_vector_score < VECTOR_CONFIDENT_SCORE:
collected.extend(self._product_keyword_hits(client, targets, text, expression))
# 命中行级子块时把父块(整节)一并带回,供调用方按问句选粒度:
# 「起投多少」要那一行,「介绍一下」要整节。
collected.extend(self._parent_hits(client, targets, collected, expression))
if not collected and failures == len(targets) and targets:
# 三个集合全查失败:是链路故障,不是"知识库里没有"
return KnowledgeSearchOutcome(
@@ -183,8 +191,25 @@ class KnowledgeSearchService:
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(deduped[: max(1, top_k)]),
hits=tuple(selected),
degraded=failures > 0,
reason="partial_collection_failure" if failures else "",
searched_collections=targets,
@@ -287,6 +312,84 @@ class KnowledgeSearchService:
# 是不是"只有一个明确候选"交给上层的 gap 判定,这里不替它做选择。
return hits
def _parent_hits(
self, client: Any, targets: Sequence[str], 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 in targets:
try:
rows = lookup(
collection_name=collection,
filter=self._anded(expression, f"doc_id in [{quoted}]"),
output_fields=list(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_id") or "")
content = str(row.get("content") or "")
if not content or doc_id not in parent_scores:
continue
found.append(KnowledgeHit(
doc_id=doc_id,
title=str(row.get("title") or ""),
content=content,
score=parent_scores[doc_id],
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 ""),
))
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 _parse(raw: Any, collection: str) -> list[KnowledgeHit]:
"""把 pymilvus 的 `[[{id, distance, entity}]]` 折叠成命中列表(纯函数,不抛异常)。"""
File diff suppressed because one or more lines are too long
@@ -0,0 +1,58 @@
"""知识块粒度选择的单元测试。
背景是实测的三次翻车,每条判据都对应其中一次:
1. 客户问「起投多少」和「风险高吗」时命中同一块(整个产品小节),拿到**完全相同**的
整节内容,看起来像客服没听懂问题——所以把表格行拆成了行级子块。
2. 拆细之后「介绍一下」又被某一行抢答(返回"产品期限 90天封闭期")——所以要能换回整节。
3. 两次判据写错:用"含连字符"认子块时,整节块自己的编号 PROD-901 被误判成子块;
用"不含两位数字后缀"认整节块时,FAQ 块全被误判成整节块、把正确答案挤出了 top1。
"""
from app.service.agent.implementations.customer_service import CustomerServiceAgent
from app.service.knowledge_search_service import KnowledgeSearchService
def test_parent_of_recognises_row_blocks() -> None:
assert KnowledgeSearchService._parent_of("PROD-007-04") == "PROD-007"
def test_parent_of_rejects_section_blocks() -> None:
"""整节块的编号本身就含连字符(PROD-901),不能被当成子块。"""
assert KnowledgeSearchService._parent_of("PROD-901") is None
assert KnowledgeSearchService._parent_of("FAQ-0016") is None
assert KnowledgeSearchService._parent_of("HNW-003") is None
def test_section_chosen_for_overview_question() -> None:
"""客户问整节时,用父块替掉抢答的那一行。"""
child = {"doc_id": "PROD-007-05", "title": "手册 · 2.1 南方季季盈90天 · 产品期限"}
parent = {"doc_id": "PROD-007", "content": "整节内容"}
assert CustomerServiceAgent._prefer_section(
"南方季季盈90天介绍一下", child, [child, parent]
) is parent
def test_row_kept_when_question_names_the_field() -> None:
"""客户问的正是那一行时不能换成整节,否则"聚焦"就白做了。"""
child = {"doc_id": "PROD-007-04", "title": "手册 · 2.1 南方季季盈90天 · 起投金额"}
parent = {"doc_id": "PROD-007", "content": "整节内容"}
assert CustomerServiceAgent._prefer_section(
"季季盈90天起投多少", child, [child, parent]
) is child
def test_missing_parent_falls_back_to_row() -> None:
"""父块没带回来时仍用子块:宁可答得窄,也不要拿不相干的块搪塞。"""
child = {"doc_id": "PROD-007-05", "title": "手册 · 2.1 南方季季盈90天 · 产品期限"}
assert CustomerServiceAgent._prefer_section("介绍一下", child, [child]) is child
def test_plain_block_is_not_treated_as_section() -> None:
"""FAQ 这类独立块没有子块挂在下面,不该被当成整节块。"""
plain = {"doc_id": "FAQ-0016", "title": "基金赎回到账需要多长时间?"}
assert CustomerServiceAgent._prefer_section("基金赎回几天到账", plain, [plain]) is plain
+65 -1
View File
@@ -132,6 +132,50 @@ def chunk_qa(text: str) -> list[dict[str, str]]:
return chunks
TABLE_ROW = re.compile(r"^\|(.+)\|\s*$")
LEADING_NUMBER = re.compile(r"^\d+(?:\.\d+)*\s*")
def expand_table_rows(chunk: dict[str, str], parent_id: str) -> list[dict[str, object]]:
"""把 Markdown 表格的每一行拆成自解释的小块(父块照旧保留)。
为什么需要:现在的粒度是"一个叶子标题 = 一块",产品手册里就是**整个产品小节**
(表格 + 说明)成一块。于是客户问「起投多少」和问「风险高吗」命中同一块、拿到
**完全相同**的整节内容——客户会觉得客服没听懂问题,只是把说明书重贴一遍。
顺带地,整节几百字的向量是"整节的混合语义",与"起投多少"这种具体小问题相似度
天然偏低(实测该问句向量 top1 只有 0.6291,够不到 0.75 门槛)。
小块必须**自解释**:只回「1万元」客户不知道说的是哪个产品,所以带上产品名与行标签。
父块保留,客户问「这个产品怎么样」时仍要能拿到完整一节。
"""
blocks: list[dict[str, object]] = []
header: list[str] = []
name = LEADING_NUMBER.sub("", chunk["section"]).strip() or chunk["section"]
for line in chunk["content"].splitlines():
match = TABLE_ROW.match(line.strip())
if not match:
continue
cells = [cell.strip() for cell in match.group(1).split("|")]
if all(set(cell) <= {"-", ":", " "} for cell in cells):
continue # 分隔行 |---|
if not header:
header = cells
continue
if len(cells) < 2:
continue
label, value = cells[0], cells[1]
if not label or not value:
continue
blocks.append({
"title": f"{chunk['title']} · {label}",
"chapter": chunk["chapter"],
"section": f"{name} · {label}",
"content": f"{name}:{label} {value}",
"parent_id": parent_id,
})
return blocks
records: list[dict[str, object]] = []
for relative, config in SOURCES.items():
text = (Path("knowledge") / relative).read_text(encoding="utf-8")
@@ -144,8 +188,9 @@ for relative, config in SOURCES.items():
if any(chunk["chapter"].startswith(prefix) for prefix in allowed)
]
for order, chunk in enumerate(chunks, 1):
parent_id = f"{config['prefix']}-{order:03d}"
records.append({
"doc_id": f"{config['prefix']}-{order:03d}",
"doc_id": parent_id,
"collection": config["collection"],
"title": chunk["title"],
"content": chunk["content"],
@@ -160,6 +205,25 @@ for relative, config in SOURCES.items():
"visibility": config["visibility"],
"chars": len(chunk["content"]),
})
# 行级子块:挂在父块 doc_id 下(PROD-007-01 这种),父块编号不受新增子块影响,
# 因此反复重跑本脚本得到的 doc_id 是稳定的。
for row_order, block in enumerate(expand_table_rows(chunk, parent_id), 1):
records.append({
"doc_id": f"{parent_id}-{row_order:02d}",
"collection": config["collection"],
"title": block["title"],
"content": block["content"],
"chapter": block["chapter"],
"section": block["section"],
"tags": config["tags"],
"doc_no": config["doc_no"],
"version": config["version"],
"effective_date": config["effective_date"],
"expire_date": "", "source_url": "", "reviewer": "",
"source_file": relative,
"visibility": config["visibility"],
"chars": len(str(block["content"])),
})
for order, chunk in enumerate(
chunk_qa((Path("knowledge") / "faq/高频问答对.txt").read_text(encoding="utf-8")), 1