From d0082b464562821096e0ec22efa14fec93a127a9 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 21:41:30 +0800 Subject: [PATCH] =?UTF-8?q?=E9=97=AE=E9=A2=98=E8=81=9A=E7=B1=BB=E7=9A=84?= =?UTF-8?q?=E5=90=91=E9=87=8F=E5=8F=AA=E5=8F=96=E4=B8=80=E6=AC=A1=EF=BC=9A?= =?UTF-8?q?=E8=AF=84=E6=B5=8B=E9=A1=B5=E3=80=8C=E5=8F=A3=E5=BE=84=E5=91=BD?= =?UTF-8?q?=E4=B8=AD=E4=B8=8E=E9=87=8D=E5=A4=8D=E5=BA=A6=E3=80=8D=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E6=8C=89=E7=AA=97=E5=8F=A3=E6=8A=8A=E9=97=AE=E9=A2=98?= =?UTF-8?q?=E9=87=8D=E5=8F=91=E7=BB=99=E5=90=91=E9=87=8F=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=EF=BC=88=E5=A4=8D=E6=A0=B8=20W4R-14=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：这个页签一次打开，/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 条失败；去掉共用的锁：并行那条失败。 --- backend/app/api/evals.py | 3 +- backend/app/chat/questions.py | 70 ++++++++++++- backend/tests/test_evals_ui_api.py | 33 +++++++ backend/tests/test_question_vectors.py | 130 +++++++++++++++++++++++++ 4 files changed, 232 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_question_vectors.py diff --git a/backend/app/api/evals.py b/backend/app/api/evals.py index 28c6addc..3c722aaa 100644 --- a/backend/app/api/evals.py +++ b/backend/app/api/evals.py @@ -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 diff --git a/backend/app/chat/questions.py b/backend/app/chat/questions.py index 6e7e9201..0c99e042 100644 --- a/backend/app/chat/questions.py +++ b/backend/app/chat/questions.py @@ -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"] diff --git a/backend/tests/test_evals_ui_api.py b/backend/tests/test_evals_ui_api.py index f80871ec..2307c84d 100644 --- a/backend/tests/test_evals_ui_api.py +++ b/backend/tests/test_evals_ui_api.py @@ -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 diff --git a/backend/tests/test_question_vectors.py b/backend/tests/test_question_vectors.py new file mode 100644 index 00000000..f17cee7b --- /dev/null +++ b/backend/tests/test_question_vectors.py @@ -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]