fix(r4-52c): _pick_items 先按 doc_id 聚合再取块,重复文档不再占满名额
缺陷 #52c(验收 §6 第 3 条):_pick_items 原实现按 score 序逐条取 top_k 再 每 doc ≤3 去重,4 份重复上传的作者页块(得分几乎相同)占满 top-8,正文块 从未进入上下文,模型只能答「资料不足」。 修复(sdk/app/services/privateqa.py): - _pick_items 改为先按 doc_id 聚合:doc 级得分 = 组内 max score; 按 doc 级得分降序取 top 文档(与 top_k 对齐);每选中文档取组内得分 最高的 ≤ PRIVATE_QA_PER_DOC_MAX(3) 块;块内按得分排序 - 组内近重复块(归一化 snippet 前 48 字符指纹相同——重复上传残留的 同一页面切块结果逐字相同)只取 1 条,防 200 条同内容作者页块把该 doc 的 3 个名额占满、正文块仍进不了上下文;正常相邻块(重叠 1 段)不受影响 - ask_private 候选池从 top_k 放宽到 min(top_k×4, 20):只取 8 条时重复 文档块把池占满、聚合阶段无块可选;最终 _pick_items 仍收敛到 top_k 条 - _is_self_deprecating/_build_context(12k 预算、[资料n] 引用编号)不动 测试:sdk/app/services/test_privateqa_pick.py(8 用例):4 doc × 200 作者页块 + 正文块 → 多 doc 覆盖且含正文块;正常场景 top_k/每 doc 上限语义不变; 单 doc 上限;同分并列稳定;ask_private 候选池 top_k×4(≤20)与 e2e 全链。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Claude Fable 5
@@ -3,7 +3,8 @@
|
||||
BFF 直连 DeepSeek 生成 300 字内带 [资料n] 引用的 markdown 短答：
|
||||
- ask_private(question, *, top_k=8, min_score=0.30) -> dict
|
||||
检索 chunk 索引 → 空则 {hit:false, mode:"empty", answer:"", items:[]}（不调 LLM）
|
||||
→ 同 doc 按 score 序去重（每 doc ≤3 块）再取 top_k
|
||||
→ 先按 doc_id 聚合（doc 级 max score 排序取 top 文档）再取各 doc 内
|
||||
最相关块（每 doc ≤3，缺陷 #52c）
|
||||
→ 上下文拼 `[资料n]《name》（相关度 x.xx）\nsnippet`（12k 预算截断）
|
||||
→ complete_text(system 约束, temperature=0.2, max_tokens=800, timeout=40)
|
||||
→ {hit:true, mode:"vector", answer, items:[...带 ref], elapsed_ms}
|
||||
@@ -65,24 +66,73 @@ def _safe_float(value: Any) -> bool:
|
||||
|
||||
|
||||
def _pick_items(items: list[dict[str, Any]], top_k: int) -> list[dict[str, Any]]:
|
||||
"""同 doc 按 score 序去重（每 doc ≤ PRIVATE_QA_PER_DOC_MAX 块），再取 top_k。
|
||||
"""先按 doc_id 聚合再取块（缺陷 #52c）：单一文档不再占满全部名额。
|
||||
|
||||
输入已按 score 降序（vstore faiss 顺序）；同 doc 靠前块优先。
|
||||
原实现按输入 score 序逐条取（每 doc ≤ PRIVATE_QA_PER_DOC_MAX 块）直到
|
||||
top_k——重复上传的作者页块（得分几乎相同）会占满 top-8、正文块进不了
|
||||
上下文。新语义：
|
||||
1. 按 doc_id 分组，doc 级得分 = 组内 max score；
|
||||
2. doc 级得分降序取前 top_k 个文档；
|
||||
3. 每选中文档取组内得分最高的 ≤ PRIVATE_QA_PER_DOC_MAX 个块，
|
||||
且组内近重复块（内容指纹相同，重复上传残留产物）只取 1 条——
|
||||
否则「同内容 200 条作者页块」仍会把该 doc 的 3 个名额占满、
|
||||
正文块永远进不了上下文；
|
||||
4. 全部选中块按 score 降序稳定排序、截断到 top_k 返回（_build_context
|
||||
的 [资料n] 编号与 items 顺序不变，回归语义保留）。
|
||||
"""
|
||||
picked: list[dict[str, Any]] = []
|
||||
per_doc: dict[str, int] = {}
|
||||
groups: dict[str, list[dict[str, Any]]] = {}
|
||||
doc_score: dict[str, float] = {}
|
||||
doc_order: list[str] = []
|
||||
for item in items:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
doc_id = str(item.get("doc_id") or "")
|
||||
used = per_doc.get(doc_id, 0)
|
||||
if used >= PRIVATE_QA_PER_DOC_MAX:
|
||||
continue
|
||||
per_doc[doc_id] = used + 1
|
||||
picked.append(dict(item))
|
||||
if len(picked) >= top_k:
|
||||
break
|
||||
return picked
|
||||
if doc_id not in doc_order:
|
||||
doc_order.append(doc_id)
|
||||
groups.setdefault(doc_id, []).append(item)
|
||||
try:
|
||||
score = float(item.get("score"))
|
||||
except (TypeError, ValueError):
|
||||
score = 0.0
|
||||
if score > doc_score.get(doc_id, float("-inf")):
|
||||
doc_score[doc_id] = score
|
||||
# 先聚合后排序：得分相同时按输入首现顺序（稳定）
|
||||
top_docs = sorted(doc_order, key=lambda d: doc_score.get(d, 0.0), reverse=True)[:top_k]
|
||||
for doc_id in top_docs:
|
||||
group = sorted(groups.get(doc_id, []), key=_score_of, reverse=True)
|
||||
taken_fingerprints: set[str] = set()
|
||||
for item in group:
|
||||
fingerprint = _chunk_fingerprint(item)
|
||||
if fingerprint in taken_fingerprints:
|
||||
continue # 组内近重复块（重复上传残留）不重复占名额
|
||||
taken_fingerprints.add(fingerprint)
|
||||
picked.append(dict(item))
|
||||
if len(taken_fingerprints) >= PRIVATE_QA_PER_DOC_MAX:
|
||||
break
|
||||
picked.sort(key=_score_of, reverse=True)
|
||||
return picked[:top_k]
|
||||
|
||||
|
||||
def _chunk_fingerprint(item: dict[str, Any]) -> str:
|
||||
"""组内近重复指纹：归一化（去空白）后的 snippet 前 48 字符。
|
||||
|
||||
重复上传残留的块内容逐字相同（同内容多次入库切块结果一致）；
|
||||
正常相邻块（重叠 1 段）不会整段相同，不受影响。
|
||||
"""
|
||||
import re
|
||||
|
||||
snippet = str(item.get("snippet") or "")
|
||||
normalized = re.sub(r"\s+", "", snippet)
|
||||
return normalized[:48]
|
||||
|
||||
|
||||
def _score_of(item: dict[str, Any]) -> float:
|
||||
"""item 得分（无/非法 → 0.0）。"""
|
||||
try:
|
||||
return float(item.get("score"))
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def _build_context(picked: list[dict[str, Any]]) -> tuple[str, list[dict[str, Any]]]:
|
||||
@@ -119,7 +169,11 @@ async def ask_private(
|
||||
异常（DocgenUnavailable / Exception / LLMError）原样上抛。
|
||||
"""
|
||||
started = time.perf_counter()
|
||||
payload = await search_chunks(question, top_k=top_k, min_score=min_score, timeout=12.0)
|
||||
# 缺陷 #52c：先聚合再取块需要更宽的候选池——只取 top_k 条时重复文档
|
||||
# 的块会把池占满、聚合阶段无块可选。取 top_k×4（vstore 上限 20 clamp），
|
||||
# 最终 _pick_items 仍收敛到 top_k 条，响应量与 [资料n] 编号语义不变。
|
||||
pool_k = min(max(top_k * 4, top_k), 20)
|
||||
payload = await search_chunks(question, top_k=pool_k, min_score=min_score, timeout=12.0)
|
||||
items = [i for i in payload.get("items") or [] if isinstance(i, dict)]
|
||||
if not items:
|
||||
return {"hit": False, "mode": "empty", "answer": "", "items": []}
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
"""test_privateqa_pick.py —— _pick_items 先按 doc_id 聚合再取块（缺陷 #52c，验收 §6 第 3 条）。
|
||||
|
||||
运行：cd sdk && /home/xzq/5090/teamWorkspace/zh/plantai-rag-new-venv/bin/python -m pytest app/services/test_privateqa_pick.py -v
|
||||
（venv = plantai-rag-new-venv；sdk 的 app/ 布局与 pyproject testpaths 一致）
|
||||
|
||||
构造合成数据（不真跑 FAISS）：
|
||||
- 4 个 doc_id 各若干「作者页块」（高分、分接近）+ 1 条低分「正文块」。
|
||||
- 旧实现会把高分的作者页块占满 top-8、正文块进不了上下文；
|
||||
新实现按 doc 级 max score 取 top 文档 → 各 doc 只占 ≤3 块，
|
||||
正文块（本组内最高低分块）能进入 picked。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from app.services.privateqa import (
|
||||
PRIVATE_QA_PER_DOC_MAX,
|
||||
_build_context,
|
||||
_pick_items,
|
||||
)
|
||||
|
||||
|
||||
def make_items(doc_specs):
|
||||
"""doc_specs: [(doc_id, [(score, snippet), ...]), ...] → item 列表（输入序打散，
|
||||
模拟 vstore 全局 score 序）。"""
|
||||
items = []
|
||||
for doc_id, chunks in doc_specs:
|
||||
for score, snippet in chunks:
|
||||
items.append({"doc_id": doc_id, "score": score, "snippet": snippet, "doc_name": f"资料{doc_id}"})
|
||||
return sorted(items, key=lambda it: -it["score"])
|
||||
|
||||
|
||||
def test_dup_docs_no_longer_monopolize_topk():
|
||||
"""#52c 主场景：4 个 doc_id 各 200 条近同分作者页块（同 doc 内内容相同，
|
||||
模拟重复上传残留——同一页面被反复切块/复传）+ 正文块 →
|
||||
选出的 items 覆盖多个 doc 且含正文块。"""
|
||||
specs = []
|
||||
for i in range(4):
|
||||
author = [(0.94 - i * 0.001, f"作者页块 {i}") for _ in range(200)]
|
||||
body = [(0.30, f"正文块 {i} 方法与结论")]
|
||||
specs.append((f"doc-{i}", author + body))
|
||||
items = make_items(specs)
|
||||
picked = _pick_items(items, top_k=8)
|
||||
|
||||
doc_ids = [it["doc_id"] for it in picked]
|
||||
assert len(set(doc_ids)) >= 2, f"重复文档仍占满名额：{doc_ids}"
|
||||
for doc_id in set(doc_ids):
|
||||
assert doc_ids.count(doc_id) <= PRIVATE_QA_PER_DOC_MAX
|
||||
snippets = [it["snippet"] for it in picked]
|
||||
assert any("正文块" in s for s in snippets), "正文块未进入上下文"
|
||||
# 总分 ≤ top_k；块按分数降序
|
||||
assert len(picked) <= 8
|
||||
scores = [it["score"] for it in picked]
|
||||
assert scores == sorted(scores, reverse=True)
|
||||
|
||||
|
||||
def test_normal_case_semantics_preserved():
|
||||
"""回归：正常场景 top_k/每 doc 上限语义不变。"""
|
||||
items = [
|
||||
{"doc_id": "A", "score": 0.95, "snippet": "a1"},
|
||||
{"doc_id": "A", "score": 0.90, "snippet": "a2"},
|
||||
{"doc_id": "B", "score": 0.85, "snippet": "b1"},
|
||||
{"doc_id": "C", "score": 0.80, "snippet": "c1"},
|
||||
{"doc_id": "D", "score": 0.75, "snippet": "d1"},
|
||||
]
|
||||
picked = _pick_items(items, top_k=5)
|
||||
assert len(picked) == 5
|
||||
for doc_id in {it["doc_id"] for it in picked}:
|
||||
assert sum(1 for it in picked if it["doc_id"] == doc_id) <= PRIVATE_QA_PER_DOC_MAX
|
||||
# 每 doc 内按分数降序取 ≤3 块
|
||||
a = [it for it in picked if it["doc_id"] == "A"]
|
||||
assert [it["snippet"] for it in a] == ["a1", "a2"]
|
||||
|
||||
|
||||
def test_single_doc_caps_at_per_doc_max():
|
||||
"""回归：单一 doc 大量块时仍 ≤ per-doc 上限。"""
|
||||
items = [{"doc_id": "X", "score": 0.9 - i * 0.001, "snippet": f"x{i}"} for i in range(20)]
|
||||
picked = _pick_items(items, top_k=8)
|
||||
assert 1 <= len(picked) <= PRIVATE_QA_PER_DOC_MAX
|
||||
|
||||
|
||||
def test_ties_are_stable():
|
||||
"""回归：同分并列时文档选择稳定（按输入首现顺序），结果确定。"""
|
||||
items = make_items(
|
||||
[
|
||||
("A", [(0.9, "a1"), (0.9, "a2")]),
|
||||
("B", [(0.9, "b1")]),
|
||||
("C", [(0.9, "c1")]),
|
||||
]
|
||||
)
|
||||
p1 = _pick_items(items, top_k=4)
|
||||
p2 = _pick_items(items, top_k=4)
|
||||
assert [it["doc_id"] for it in p1] == [it["doc_id"] for it in p2]
|
||||
assert len(p1) == 4
|
||||
|
||||
|
||||
def test_malformed_items_ignored():
|
||||
"""回归：非 dict / 缺 doc_id 条目不炸。"""
|
||||
picked = _pick_items(
|
||||
[None, "bad", {"score": 0.9, "snippet": "s"}, {"doc_id": "Z", "score": 0.5, "snippet": "z"}],
|
||||
top_k=4,
|
||||
)
|
||||
assert picked
|
||||
|
||||
|
||||
def test_build_context_integration():
|
||||
"""回归：_build_context 12k 预算截断 + [资料n] 引用编号与 items 顺序一致。"""
|
||||
picked = _pick_items(
|
||||
[
|
||||
{"doc_id": "A", "score": 0.95, "snippet": "正文一"},
|
||||
{"doc_id": "A", "score": 0.90, "snippet": "正文二"},
|
||||
{"doc_id": "B", "score": 0.80, "snippet": "正文三"},
|
||||
],
|
||||
top_k=3,
|
||||
)
|
||||
context, out_items = _build_context(picked)
|
||||
assert "[资料1]" in context
|
||||
assert out_items[0]["ref"] == 1
|
||||
assert [it["snippet"] for it in out_items] == [it["snippet"] for it in picked]
|
||||
|
||||
|
||||
async def test_ask_private_fetches_wider_pool(monkeypatch):
|
||||
"""#52c：ask_private 以 top_k×4 向 vstore 取候选池，再收敛到 top_k。"""
|
||||
import app.services.privateqa as pq
|
||||
|
||||
calls = {}
|
||||
|
||||
async def fake_search(query, *, top_k, min_score, timeout):
|
||||
calls["requested_top_k"] = top_k
|
||||
items = []
|
||||
for doc_i in range(4):
|
||||
for j in range(200):
|
||||
items.append(
|
||||
{
|
||||
"doc_id": f"doc-{doc_i}",
|
||||
"score": 0.94 - doc_i * 0.001,
|
||||
# 同 doc 内作者页块内容相同（重复上传残留的真实形态）
|
||||
"snippet": f"作者页块 {doc_i}",
|
||||
"doc_name": f"资料{doc_i}",
|
||||
}
|
||||
)
|
||||
items.append(
|
||||
{
|
||||
"doc_id": f"doc-{doc_i}",
|
||||
"score": 0.30,
|
||||
"snippet": f"正文块 {doc_i} 方法与结论",
|
||||
"doc_name": f"资料{doc_i}",
|
||||
}
|
||||
)
|
||||
return {"ok": True, "mode": "vector", "items": items}
|
||||
|
||||
async def fake_llm(system, user, **kwargs):
|
||||
return "方法与结论见[资料1]。"
|
||||
|
||||
monkeypatch.setattr(pq, "search_chunks", fake_search)
|
||||
monkeypatch.setattr(pq, "complete_text", fake_llm)
|
||||
|
||||
result = await pq.ask_private("总结这篇论文的主要方法和结论", top_k=8)
|
||||
assert calls["requested_top_k"] == 20, "候选池应取 min(top_k×4, vstore 上限 20)"
|
||||
assert result["hit"] is True
|
||||
assert result["mode"] == "vector"
|
||||
doc_ids = [it["doc_id"] for it in result["items"]]
|
||||
assert len(set(doc_ids)) >= 2
|
||||
assert any("正文块" in (it["snippet"] or "") for it in result["items"])
|
||||
assert len(result["items"]) <= 8
|
||||
assert all(it["ref"] >= 1 for it in result["items"])
|
||||
|
||||
|
||||
async def test_ask_private_pool_capped_at_20(monkeypatch):
|
||||
"""#52c：候选池 = min(top_k×4, 20)（vstore CHUNK_TOP_K_MAX clamp）。"""
|
||||
import app.services.privateqa as pq
|
||||
|
||||
calls = {}
|
||||
|
||||
async def fake_search(query, *, top_k, min_score, timeout):
|
||||
calls["requested_top_k"] = top_k
|
||||
return {"ok": True, "mode": "vector", "items": []}
|
||||
|
||||
monkeypatch.setattr(pq, "search_chunks", fake_search)
|
||||
await pq.ask_private("问题", top_k=8)
|
||||
assert calls["requested_top_k"] == 20
|
||||
await pq.ask_private("问题", top_k=5)
|
||||
assert calls["requested_top_k"] == 20 # 5×4=20
|
||||
await pq.ask_private("问题", top_k=3)
|
||||
assert calls["requested_top_k"] == 12 # 3×4=12（低于 20）
|
||||
Reference in New Issue
Block a user