diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index cdad6346..2fe472ec 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -249,7 +249,10 @@ async def certify_dismiss(payload: DismissIn, session: AsyncSession = Depends(ge """「不是口径」:记一条审计(ledger.certify.dismiss),这一项之后不再进队。""" from app.ledger import certify + from app.api.deps import current_principal + if payload.kind is not None and payload.kind not in certify.KINDS: raise HTTPException(422, "可沉淀的问题只有三种:确定性重复、问题聚类、被反复用的答案") - await certify.dismiss(session, payload.item_id, kind=payload.kind, note=payload.note) + await certify.dismiss(session, payload.item_id, kind=payload.kind, note=payload.note, + principal=current_principal()) return {"ok": True, "item_id": payload.item_id} diff --git a/backend/app/chat/questions.py b/backend/app/chat/questions.py index 0c99e042..99d61240 100644 --- a/backend/app/chat/questions.py +++ b/backend/app/chat/questions.py @@ -18,6 +18,12 @@ 聚类用 memory/embeddings 的向量贪心聚:按时间从新到旧,和某一类的种子足够像就并进去,否则自立一类。本地向量是词频 哈希、语义能力弱,结果是下限——调用方据 embedder 照实写明用的是哪个向量模型。 +**类的 id 由成员定。** 每个问题有稳定的 id(Question.key:问数据是那一轮,别的探索运行是那次运行),一类的 id +(QuestionCluster.id)是成员的 id 排好之后取的哈希:同一批问题两次读取,类的 id 一样。以前按种子问题的原文算,种子是一类里 +最新的那一句,同一批问题时间相同的几句先后不定、来了一句新的问法种子就换了,「不是口径」按 id 记就失效了。同一时刻问的 +几句按 id 排出先后(recent_questions),不靠库里读出来的先后。来了新成员 id 照样会变:认证队列按成员记「不是口径」 +(ledger/certify),不只按 id。 + **向量只取一次。** 同一批问题会被聚很多遍:评测页「口径命中与重复度」一次打开,/api/evals/signals 按窗口调 compute_stats (缺省 6 个窗口,每个都聚一次),页面另调 /api/ledger/stats;认证队列、治理页的探索问题聚类也各聚一遍。配了远程向量模型时 每聚一遍都把几百句问题重新发一次,页面慢、还花钱。所以问题原文 → 向量在进程里留一份(_vectors:最多 VECTOR_CACHE_MAX 句, @@ -28,6 +34,7 @@ from __future__ import annotations import asyncio +import hashlib import weakref from collections import OrderedDict from collections.abc import Sequence @@ -55,6 +62,16 @@ class Question: workflow: str | None = None origin: str | None = None + @property + def key(self) -> str: + """稳定的 id:问数据的那一轮(turn:…),别的探索运行是那次运行(run:…);两样都没有的按原文和时间。""" + if self.turn_id: + return f"turn:{self.turn_id}" + if self.run_id: + return f"run:{self.run_id}" + stamp = self.at.isoformat() if self.at is not None else "" + return "text:" + hashlib.sha256(f"{self.text}\n{stamp}".encode()).hexdigest()[:16] + @dataclass class QuestionCluster: @@ -66,6 +83,16 @@ class QuestionCluster: def count(self) -> int: return len(self.members) + @property + def keys(self) -> list[str]: + """成员的稳定 id,排好的。""" + return sorted(q.key for q in self.members) + + @property + def id(self) -> str: + """类的 id:成员的稳定 id 排好之后取哈希(见模块说明)。同一批问题两次读取一样。""" + return hashlib.sha256("\n".join(self.keys).encode()).hexdigest()[:16] + async def recent_questions(session: AsyncSession, principal: Any, *, limit: int) -> list[Question]: """最近问过的问题,从新到旧,最多 limit 条(取法见模块开头)。principal 用来去掉这个人读不了的只给拥有者的运行。""" @@ -94,7 +121,8 @@ async def recent_questions(session: AsyncSession, principal: Any, *, limit: int) text = (payload or {}).get("question") if isinstance(payload, dict) else None if isinstance(text, str) and text.strip(): out.append(Question(text=text.strip(), at=at, run_id=run_id, workflow=workflow, origin=origin)) - out.sort(key=lambda q: q.at.timestamp() if q.at is not None else 0.0, reverse=True) + # 同一时刻问的几句按稳定的 id 排:库里读出来的先后不定,种子、聚类的结果就跟着变(见模块说明) + out.sort(key=lambda q: (q.at.timestamp() if q.at is not None else 0.0, q.key), reverse=True) return out[:limit] diff --git a/backend/app/ledger/certify.py b/backend/app/ledger/certify.py index 3892a763..049c9ba7 100644 --- a/backend/app/ledger/certify.py +++ b/backend/app/ledger/certify.py @@ -11,7 +11,9 @@ 位置(按规范形里的先后对齐)是参数候选,不变的是口径常量。展示的是代表那条的规范文本(字面量都换成了「?」,不带取值)。 这一项的 id 按(数据源、规范形)算;以前按规范文本记的「不是口径」照样认(组里任何一条的规范文本对得上就算); - 问题聚类(cluster):app/chat/questions 全站唯一的聚类(只取问题原文、不含 history、排除系统派生的运行),≥ REPEAT_RUNS - 个问题的类; + 个问题的类。这一项的 id 是类的 id(成员的稳定 id 排好取哈希:同一批问题两次读取一样);「不是口径」记下当时的成员,之后 + 这一类过半的成员是那时记下的就照样不进队(来了几句新问法 id 会变,按成员认就不失效)。以前按种子问题原文算的 id 记的 + 照样认,记的成员读不懂的照实忽略; - 被反复用的答案(reused):导出过 ≥ 2 次(report.exported 审计,第 2 波)、做成过应用(蓝图的来源是这次运行)、在画布里 打开过或提取过(工作流的 origin.run_id 是这次运行)的轮次。 每项给代表问题、轮次链接、(shape 的)SQL 形状和参数候选、有没有相关口径(可以对账的那份)、是不是已经固化过(frozen_into, @@ -241,6 +243,27 @@ async def _dismissed(session: Any) -> set[str]: AuditEvent.action == DISMISS_ACTION, AuditEvent.subject_kind == SUBJECT_KIND))).all() if sid} +async def _dismissed_members(session: Any) -> list[set[str]]: + """问题聚类上点过「不是口径」时记下的成员(每条一组稳定 id)。老记录没记成员、记的读不懂(不是字符串列表)的不算。""" + from app.db.models import AuditEvent + + out: list[set[str]] = [] + for (data,) in (await session.execute(select(AuditEvent.data).where( + AuditEvent.action == DISMISS_ACTION, AuditEvent.subject_kind == SUBJECT_KIND))).all(): + members = data.get("members") if isinstance(data, dict) else None + if isinstance(members, list) and members and all(isinstance(m, str) and m for m in members): + out.append(set(members)) + return out + + +def _cluster_dismissed(cluster: Any, dismissed: set[str], members: list[set[str]]) -> bool: + """这一类点过「不是口径」没有:类的 id 记过;以前按种子问题原文算的 id 记过;或者过半的成员是某一次记下的。""" + if f"cluster:{cluster.id}" in dismissed or item_id("cluster", " ".join(cluster.members[0].text.split())) in dismissed: + return True + keys = set(cluster.keys) + return any(len(keys & seen) * 2 > len(keys) for seen in members) + + async def _frozen(session: Any) -> dict[str, list[dict[str, Any]]]: """来源运行 → 从它固化出的工作流 [{workflow_id, name, status, published_version}](origin.sources)。""" from app.db.models import Workflow @@ -279,6 +302,7 @@ async def queue(session: Any, *, principal: Any) -> dict[str, Any]: from app.ledger import store dismissed = await _dismissed(session) + dismissed_members = await _dismissed_members(session) frozen = await _frozen(session) kinds = {name: kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind))).all()} runs = list((await session.execute(select(Run).where( @@ -360,8 +384,8 @@ async def queue(session: Any, *, principal: Any) -> dict[str, Any]: if cluster.count < REPEAT_RUNS: continue seed = cluster.members[0] - iid = item_id("cluster", " ".join(seed.text.split())) - if iid in dismissed: + iid = f"cluster:{cluster.id}" + if _cluster_dismissed(cluster, dismissed, dismissed_members): continue items.append({"id": iid, "kind": "cluster", "question": seed.text, "count": cluster.count, "turns": [{"run_id": q.run_id, "turn_id": q.turn_id, "question": q.text} @@ -415,12 +439,25 @@ async def _reused(session: Any, runs: list[Run], turns: dict[str, dict[str, Any] return out -async def dismiss(session: Any, iid: str, *, kind: str | None = None, note: str | None = None) -> None: - """「不是口径」:写一条审计,之后这一项不再进队。""" +async def dismiss(session: Any, iid: str, *, kind: str | None = None, note: str | None = None, + principal: Any = None) -> None: + """「不是口径」:写一条审计,之后这一项不再进队。问题聚类另记下这一类现在的成员(服务端按同一批问题重新聚一次找到 + 这一类,不收客户端传的成员;找不到——这会儿问题又多了、聚出来的已经不是这一类——就只记 id,照实)。""" from app.auth.audit import record + members: list[str] = [] + if principal is not None and (kind == "cluster" or iid.startswith("cluster:")): + from app.chat.questions import cluster_questions, recent_questions + + try: + questions = await recent_questions(session, principal, limit=QUESTION_LIMIT) + found = next((c for c in await cluster_questions(questions) if f"cluster:{c.id}" == iid), None) \ + if questions else None + members = found.keys if found is not None else [] + except Exception: # noqa: BLE001 - 取向量出错:只记 id,「不是口径」照样记上 + logger.exception("「不是口径」时重新聚类出错,只按 id 记") await record(session, DISMISS_ACTION, SUBJECT_KIND, iid, - data={k: v for k, v in {"kind": kind, "note": note}.items() if v}) + data={k: v for k, v in {"kind": kind, "note": note, "members": members}.items() if v}) await session.commit() diff --git a/backend/tests/test_ledger_certify.py b/backend/tests/test_ledger_certify.py index 1431921c..5ecbbbd5 100644 --- a/backend/tests/test_ledger_certify.py +++ b/backend/tests/test_ledger_certify.py @@ -347,3 +347,78 @@ async def test_a_not_a_caliber_recorded_by_the_old_text_id_still_counts(client, res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": old, "kind": "shape"}) assert res.status_code == 200 assert shape_items((await client.get("/api/ledger/certify-queue")).json()) == [] + + +# -------------------------------------------------------------------------- +# 问题聚类的 id 由成员定;「不是口径」按成员记 +# -------------------------------------------------------------------------- + + +async def test_cluster_ids_come_from_the_members_and_do_not_change_between_reads(): + from datetime import datetime, timezone + + from app.chat.questions import Question, QuestionCluster, cluster_questions + + at = datetime(2026, 10, 1, tzinfo=timezone.utc) + # 同一时刻问的三句(时间相同,库里读出来的先后不定):两次读出来先后不同,种子就不同,类的 id 仍然一样 + texts = ["仓库乙本月出库多少件", "仓库乙本月出库多少件?", "仓库乙本月出库了多少件"] + questions = [Question(text=t, at=at, turn_id=f"t-{i}") for i, t in enumerate(texts)] + first = await cluster_questions(questions) + again = await cluster_questions(list(reversed(questions))) + assert len(first) == 1 and len(again) == 1 and first[0].members[0].text != again[0].members[0].text + assert [c.id for c in first] == [c.id for c in again] + assert first[0].keys == ["turn:t-0", "turn:t-1", "turn:t-2"] + assert QuestionCluster(members=list(reversed(questions))).id == first[0].id + # 成员不同,id 就不同 + assert QuestionCluster(members=questions[:2]).id != first[0].id + + +async def test_not_a_caliber_on_a_cluster_survives_a_new_member_and_old_records_still_count(client, monkeypatch): + import uuid + + from app.auth.audit import record + + text = f"仓库甲本周入库了多少件({uuid.uuid4().hex[:6]})" + a, b = await conversation(), await conversation() + asked = {(await ask(monkeypatch, conv, weekly(*WEEKS[0]), question=text)).id for conv in (a, a, b)} + + async def mine() -> list[dict]: + body = (await client.get("/api/ledger/certify-queue")).json() + return [i for i in body["items"] if i["kind"] == "cluster" and asked & {t["run_id"] for t in i["turns"]}] + + [item] = await mine() + [again] = await mine() + assert item["id"] == again["id"], "同一批问题两次读取,类的 id 一样" + # 读不懂的老记录照实忽略:不报错、不把这一类藏起来 + async with SessionLocal() as session: + await record(session, certify.DISMISS_ACTION, certify.SUBJECT_KIND, "cluster:0000000000000000", + data={"kind": "cluster", "members": "看不懂"}) + await session.commit() + assert [i["id"] for i in await mine()] == [item["id"]] + + res = await client.post("/api/ledger/certify-queue/dismiss", json={"item_id": item["id"], "kind": "cluster"}) + assert res.status_code == 200 + assert await mine() == [] + # 再换个说法问一句:成员多了一个、种子换成了新的这句,类的 id 变了;过半的成员是点「不是口径」时记下的,照样不进队 + newer = await ask(monkeypatch, b, weekly(*WEEKS[1]), question=text + "?") + asked.add(newer.id) + assert await mine() == [] + + +async def test_a_not_a_caliber_recorded_by_the_old_seed_text_id_still_counts(client, monkeypatch): + import uuid + + from app.auth.audit import record + + text = f"仓库丙上周退货多少件({uuid.uuid4().hex[:6]})" + a, b = await conversation(), await conversation() + for conv in (a, a, b): + await ask(monkeypatch, conv, weekly(*WEEKS[0]), question=text) + body = (await client.get("/api/ledger/certify-queue")).json() + assert [i for i in body["items"] if i["kind"] == "cluster" and i["question"] == text] + async with SessionLocal() as session: + await record(session, certify.DISMISS_ACTION, certify.SUBJECT_KIND, certify.item_id("cluster", text), + data={"kind": "cluster"}) + await session.commit() + body = (await client.get("/api/ledger/certify-queue")).json() + assert not [i for i in body["items"] if i["kind"] == "cluster" and i["question"] == text]