问题聚类的向量只取一次:评测页「口径命中与重复度」不再按窗口把问题重发给向量模型(复核 W4R-14)

为什么:这个页签一次打开,/api/evals/signals 按窗口调 compute_stats(缺省 6 个窗口,每个都聚一次问题),页面另调
/api/ledger/stats,一共聚 8 遍;chat/questions.cluster_questions 每遍都把窗口里的问题原文整批发给 embed_texts,
一次里重复的问法也照发。配了远程向量模型时每打开一次就是几百到几千次向量调用,慢、花钱。复现:本地 3 句问题各问 3 次,
一次打开每句被取了 9 次向量。

做法(改在全站唯一的取问题、聚类处 app/chat/questions.py,不动 ledger/stats 的统计口径):
- 问题原文 → 向量在进程里留一份(_vectors,最多 VECTOR_CACHE_MAX=1000 句、最近用到的留着),同一句话只取一次,
  一次里重复的问法去重后只发一句;/signals、/ledger/stats、认证队列、治理页的探索问题聚类都走它;
- 缓存认「哪一个 embedder、哪一个取向量的函数」算的:设置页换了向量模型(configure 换了对象)整份丢掉,不把两个
  向量空间的坐标混着比;取的这一会儿换了的,这一批照用、不留;
- 取向量按事件循环串行(每个循环一把锁,同项目里别处的写法):页面并行取的两个接口不把同一批问题各发一遍。

测试:test_evals_ui_api 新增「一次打开每句问题只取一次向量、再打开一次都不取、两个接口并行也一样」;
新文件 tests/test_question_vectors.py(去重、换模型丢缓存、取到一半换模型、上限与最近使用、并行只发一批)。
去掉缓存回跑:每句 9 次;去掉换模型的核对:2 条失败;去掉共用的锁:并行那条失败。
This commit is contained in:
ThearchyHelios
2026-10-16 21:41:30 +08:00
parent 4a81ad25bf
commit d0082b4645
4 changed files with 232 additions and 4 deletions
+2 -1
View File
@@ -34,7 +34,8 @@
结果的列和前几行(界面上点选格或列);用到遮罩列的照保存时那一道拒收、照实写原因(不回 422);
- GET /signals:「口径命中与重复度」页签的判据(细案 copilot-look-and-caliber-ledger §3.8 的参考线:E1 ≥30% / <10%,
E3 ≥20%)、当前窗口和往前几个窗口各自落在哪一档、成员不少于 3 的问题类(可以加入题集或提取模板)。数都出自
ledger/stats.compute_stats 和 chat/questions(全站各一份),这里只按窗口调它们、按判据分档。
ledger/stats.compute_stats 和 chat/questions(全站各一份),这里只按窗口调它们、按判据分档。每个窗口都要聚一次问题,页面
另调 /api/ledger/stats:问题的向量由 chat/questions 在进程里留一份,同一句话只取一次(配了远程向量模型时不按窗口重发)。
"""
from __future__ import annotations
+67 -3
View File
@@ -17,9 +17,19 @@
聚类用 memory/embeddings 的向量贪心聚:按时间从新到旧,和某一类的种子足够像就并进去,否则自立一类。本地向量是词频
哈希、语义能力弱,结果是下限——调用方据 embedder 照实写明用的是哪个向量模型。
**向量只取一次。** 同一批问题会被聚很多遍:评测页「口径命中与重复度」一次打开,/api/evals/signals 按窗口调 compute_stats
(缺省 6 个窗口,每个都聚一次),页面另调 /api/ledger/stats;认证队列、治理页的探索问题聚类也各聚一遍。配了远程向量模型时
每聚一遍都把几百句问题重新发一次,页面慢、还花钱。所以问题原文 → 向量在进程里留一份(_vectors:最多 VECTOR_CACHE_MAX 句,
最近用到的留着),同一句话只取一次;一次里重复的问法(聚类量的正是这个)也只发一句。缓存认的是「哪一个 embedder、哪一个取向量
的函数」算的:设置页换了向量模型(memory/embeddings.configure 换了对象)就整份丢掉,不把两个向量空间的坐标混着比。取向量按事件
循环串行:页面并行取的两个接口不会把同一批问题各发一遍。
"""
from __future__ import annotations
import asyncio
import weakref
from collections import OrderedDict
from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import datetime
@@ -30,6 +40,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
#: 缺省的相似度阈值(同治理接口一直以来的 0.62)
DEFAULT_THRESHOLD = 0.62
#: 进程里最多留几句问题的向量:取问题的地方一次最多 500 句(ledger/stats、治理接口的上限),留两批的量,窗口来回切不用重取
VECTOR_CACHE_MAX = 1000
@dataclass(frozen=True)
@@ -86,14 +98,65 @@ async def recent_questions(session: AsyncSession, principal: Any, *, limit: int)
return out[:limit]
#: 问题原文 → 向量(最近用到的在后)和它们是谁算的(见模块说明「向量只取一次」)
_VECTORS: "OrderedDict[str, Any]" = OrderedDict()
_VECTORS_OF: tuple[Any, Any] | None = None
_VECTOR_LOCKS: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock]" = weakref.WeakKeyDictionary()
def _vector_lock() -> asyncio.Lock:
"""取向量的锁,按事件循环各一把(asyncio.Lock 不能跨循环用)。"""
loop = asyncio.get_running_loop()
lock = _VECTOR_LOCKS.get(loop)
if lock is None:
lock = _VECTOR_LOCKS[loop] = asyncio.Lock()
return lock
def _owner() -> tuple[Any, Any]:
from app.memory import embeddings
return embeddings.get_embedder(), embeddings.embed_texts
def _same_owner(a: tuple[Any, Any] | None, b: tuple[Any, Any]) -> bool:
return a is not None and a[0] is b[0] and a[1] is b[1]
async def _vectors(texts: Sequence[str]) -> list[Any]:
"""这些问题的向量(顺序同 texts):进程里有的直接用,没有的去重后一次取齐、留下来。"""
global _VECTORS_OF
async with _vector_lock():
owner = _owner()
if not _same_owner(_VECTORS_OF, owner):
_VECTORS.clear()
_VECTORS_OF = owner
got = {t: _VECTORS[t] for t in texts if t in _VECTORS}
missing = [t for t in dict.fromkeys(texts) if t not in got]
if missing:
fresh = await owner[1](missing)
got.update(zip(missing, fresh))
# 取的这一会儿设置页换了向量模型:这一批照用(同一次聚类里坐标是一致的),不留给下一次
if not _same_owner(_VECTORS_OF, _owner()):
return [got[t] for t in texts]
for t in missing:
_VECTORS[t] = got[t]
for t in dict.fromkeys(texts):
_VECTORS.move_to_end(t)
while len(_VECTORS) > VECTOR_CACHE_MAX:
_VECTORS.popitem(last=False)
return [got[t] for t in texts]
async def cluster_questions(questions: Sequence[Question], *, threshold: float = DEFAULT_THRESHOLD,
) -> list[QuestionCluster]:
"""向量贪心聚类(顺序同 questions:从新到旧)。"""
"""向量贪心聚类(顺序同 questions:从新到旧)。向量取自 _vectors:同一句话在进程里只取一次。"""
from app.memory import embeddings
if not questions:
return []
vectors = await embeddings.embed_texts([q.text for q in questions])
vectors = await _vectors([q.text for q in questions])
clusters: list[tuple[int, QuestionCluster]] = []
for i, question in enumerate(questions):
for seed, cluster in clusters:
@@ -112,4 +175,5 @@ def embedder() -> str:
return embeddings.embedder_id()
__all__ = ["DEFAULT_THRESHOLD", "Question", "QuestionCluster", "cluster_questions", "embedder", "recent_questions"]
__all__ = ["DEFAULT_THRESHOLD", "VECTOR_CACHE_MAX", "Question", "QuestionCluster", "cluster_questions", "embedder",
"recent_questions"]
+33
View File
@@ -194,6 +194,39 @@ async def test_signals_give_criteria_trend_and_repeated_question_clusters(client
assert (await client.get("/api/evals/signals", params={"days": 0})).status_code == 422
async def test_one_visit_to_the_signals_tab_embeds_each_question_once(client, monkeypatch):
"""「口径命中与重复度」页签一次打开:/signals 按窗口调 compute_stats(每个窗口都聚一次问题),页面另调 /ledger/stats——
同一句问题原文不管在几个窗口、几个接口里出现,都只取一次向量;再打开一遍一次都不取(配了远程向量模型时每取一次都花钱)。
两个接口同时到(页面是并行取的)也一样。"""
import asyncio
from collections import Counter
from app.memory import embeddings
real = embeddings.embed_texts
seen: Counter[str] = Counter()
async def counting(texts):
seen.update(texts)
return await real(texts)
monkeypatch.setattr(embeddings, "embed_texts", counting)
mark = uuid.uuid4().hex[:6]
asked = [f"虚构门店 {mark} 第 {i} 问的销售额是多少?" for i in range(3)]
for question in asked:
await _asked(question, 3)
a, b = await asyncio.gather(client.get("/api/evals/signals", params={"days": 30, "points": 6}),
client.get("/api/ledger/stats"))
assert a.status_code == 200 and b.status_code == 200
mine = {t: n for t, n in seen.items() if mark in t}
assert set(mine) == set(asked) and set(mine.values()) == {1}, mine
assert sum(c["count"] for c in a.json()["clusters"] if mark in c["representative"]) >= 9, "聚类照旧:九次提问都在"
seen.clear()
await client.get("/api/evals/signals", params={"days": 30, "points": 6})
await client.get("/api/ledger/stats")
assert not [t for t in seen if mark in t], "再打开一遍:向量都在进程里,一次都不取"
def test_bands_follow_the_reference_lines():
from app.api import evals
+130
View File
@@ -0,0 +1,130 @@
"""问题聚类的向量只取一次(app/chat/questions.py 模块说明「向量只取一次」)。
评测页「口径命中与重复度」一次打开要按窗口聚好几遍问题,页面另调账本统计;配了远程向量模型时每遍都把几百句问题重发一次。
这里守进程里那一份问题原文 → 向量:
- 同一句话只取一次,一次里重复的问法也只发一句;顺序、对应关系照旧;
- 换了向量模型(或取向量的函数)就整份丢掉,不把两个向量空间的坐标混着比;取的这一会儿换了的,这一批照用、不留;
- 最多留 VECTOR_CACHE_MAX 句,最近用到的留着;一次超过上限也照样每句拿到自己的向量。
问题都是虚构的。
"""
from __future__ import annotations
import pytest
from app.chat import questions as qmod
from app.chat.questions import Question, cluster_questions
from app.memory import embeddings
@pytest.fixture(autouse=True)
def fresh_cache():
qmod._VECTORS.clear()
qmod._VECTORS_OF = None
yield
qmod._VECTORS.clear()
qmod._VECTORS_OF = None
DIM = 32
def _fake(calls: list[list[str]], *, scale: float = 1.0, made: dict[str, list[float]] | None = None):
"""每句话一个确定的向量:「同类」开头的几句彼此相同(聚得到一起),别的各占一个方向(互不相像)。made 记下给过的向量。"""
seen: dict[str, int] = {}
async def embed(texts):
calls.append(list(texts))
out = []
for t in texts:
vec = [0.0] * DIM
vec[0 if t.startswith("同类") else 1 + seen.setdefault(t, len(seen))] = scale
out.append(vec)
if made is not None:
made[t] = vec
return out
return embed
def _qs(*texts: str) -> list[Question]:
return [Question(text=t, at=None) for t in texts]
async def test_each_text_is_embedded_once_and_duplicates_in_one_call_are_sent_once(monkeypatch):
calls: list[list[str]] = []
monkeypatch.setattr(embeddings, "embed_texts", _fake(calls))
got = await cluster_questions(_qs("同类:上周销售额", "同类:上周销售额", "别的:库存", "同类:上周销售额"))
assert calls == [["同类:上周销售额", "别的:库存"]], "一次里重复的问法只发一句"
assert [c.count for c in got] == [3, 1]
again = await cluster_questions(_qs("别的:库存", "同类:上周销售额", "新的:退款率"))
assert calls[1:] == [["新的:退款率"]], "已经取过的不再取"
assert [c.count for c in again] == [1, 1, 1]
async def test_a_new_embedder_drops_every_vector(monkeypatch):
first: list[list[str]] = []
second: list[list[str]] = []
monkeypatch.setattr(embeddings, "embed_texts", _fake(first))
await cluster_questions(_qs("同类:本月退款率", "别的:毛利"))
monkeypatch.setattr(embeddings, "embed_texts", _fake(second, scale=2.0))
await cluster_questions(_qs("同类:本月退款率", "别的:毛利"))
assert second == [["同类:本月退款率", "别的:毛利"]], "换了取向量的函数:整份重取"
assert qmod._VECTORS["同类:本月退款率"][0] == 2.0
# 设置页换了向量模型(configure 换了 embedder 对象)同理
other = embeddings.LocalEmbedder()
monkeypatch.setattr(embeddings, "get_embedder", lambda: other)
await cluster_questions(_qs("同类:本月退款率"))
assert second[1:] == [["同类:本月退款率"]]
async def test_switching_embedder_mid_call_keeps_the_batch_but_not_the_cache(monkeypatch):
calls: list[list[str]] = []
later: list[list[str]] = []
async def switching(texts):
calls.append(list(texts))
# 取的这一会儿设置页换了向量模型
monkeypatch.setattr(embeddings, "embed_texts", _fake(later))
return [[1.0] + [0.0] * (DIM - 1) for _ in texts]
monkeypatch.setattr(embeddings, "embed_texts", switching)
got = await cluster_questions(_qs("同类:华东门店", "同类:华北门店"))
assert [c.count for c in got] == [2], "这一批照用旧向量算完"
assert not qmod._VECTORS, "旧向量不留给下一次"
await cluster_questions(_qs("同类:华东门店"))
assert later == [["同类:华东门店"]]
async def test_the_cache_is_bounded_and_keeps_the_recently_used(monkeypatch):
calls: list[list[str]] = []
made: dict[str, list[float]] = {}
monkeypatch.setattr(embeddings, "embed_texts", _fake(calls, made=made))
monkeypatch.setattr(qmod, "VECTOR_CACHE_MAX", 3)
texts = [f"别的:第{i}问" for i in range(5)]
got = await qmod._vectors(texts)
assert got == [made[t] for t in texts] and len({tuple(v) for v in got}) == 5, "一次超过上限也每句拿到自己的向量"
assert list(qmod._VECTORS) == texts[2:], "只留最近的 3 句"
await qmod._vectors([texts[2]])
await qmod._vectors(["别的:又一问"])
assert list(qmod._VECTORS) == [texts[4], texts[2], "别的:又一问"], "最近用到的留着,最久没用的先丢"
async def test_two_callers_at_once_send_the_batch_once(monkeypatch):
"""页面并行取 /api/evals/signals 和 /api/ledger/stats:同一批问题不各发一遍。"""
import asyncio
calls: list[list[str]] = []
inner = _fake(calls)
async def slow(texts):
await asyncio.sleep(0.02)
return await inner(texts)
monkeypatch.setattr(embeddings, "embed_texts", slow)
batch = _qs("同类:上周客单价", "别的:会员数")
a, b = await asyncio.gather(cluster_questions(batch), cluster_questions(batch))
assert calls == [["同类:上周客单价", "别的:会员数"]]
assert [c.count for c in a] == [c.count for c in b] == [1, 1]