From dd002346227d0d27546c79c83f75d7fef74338b0 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 10:18:12 +0800 Subject: [PATCH] =?UTF-8?q?=E7=B3=BB=E7=BB=9F=E6=B4=BE=E7=94=9F=E7=9A=84?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E4=B8=8D=E8=BF=9B=E4=BD=BF=E7=94=A8=E7=BB=9F?= =?UTF-8?q?=E8=AE=A1=EF=BC=88=E6=80=BB=E6=96=B9=E6=A1=88=20X1=E3=80=81X2?= =?UTF-8?q?=EF=BC=8C=E7=BB=86=E6=A1=88=2001=20=E5=81=9A=E6=B3=95=204?= =?UTF-8?q?=EF=BC=89=EF=BC=9A=E8=AF=84=E6=B5=8B=E4=B8=80=E8=B7=91=EF=BC=8C?= =?UTF-8?q?=E8=81=9A=E7=B1=BB=E3=80=81=E7=99=BD=E5=90=8D=E5=8D=95=E3=80=81?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E4=BD=BF=E7=94=A8=E7=83=AD=E5=BA=A6=E9=83=BD?= =?UTF-8?q?=E4=BC=9A=E8=A2=AB=E5=88=B7=E9=AB=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - engine/origin.py：SYSTEM_ORIGINS = (eval, reconcile, recall) 和 counted_runs()（写法同 hidden_runs，coalesce 防 NULL：来源为空的老运行、外连接没对上运行的行照常计入）；ledger_issue 是替用户正式出具，照常计入 - 取问题、聚类全站只留一份：新增 app/chat/questions.py（recent_questions / cluster_questions）。只取问题原文、不取整份输入（以前把含 history 的 input JSON 拿去聚，同一段对话里的不同问题会聚成一类）；问数据按轮次一轮一条（conversation_turns.question），同一轮里的自动重来只是换了挂着的运行，不多算；别的探索运行取 input.question；排除系统运行和别人的私有运行。/governance/exploratory-clusters 改用它，多给 turn_ids 和 embedder（本地哈希向量的结果只是下限） - 工具白名单衰减、目录的使用次数和「被哪些运行用过」、审阅队列的使用权重（都走 catalog.query_snapshot_tables）只算 counted_runs 圈进来的运行 - 记录页：没按来源筛时默认不列这三类，按来源点名就列；新增 GET /api/runs/unlisted（同一份筛选，_filtered 和列表共用）照实给出没列出几次、各是哪种来源。界面在列表底下（空列表时也写）写「另有 N 次系统运行未列出（评测 2 次、口径对账 1 次）」，来源下拉的「全部」改说「全部来源（不含系统运行）」；文案进 lib/terms.ts 的 SYSTEM_RUNS_TEXT；check-runs 补一段 - test_personal_apps 里代表「别的探索运行」的两条运行改成画布试跑：问数据的问题按轮次取，没挂在任何一轮上的问数据运行不单独算 - 新增 tests/test_eval_origin.py（去掉 counted_runs 的过滤回跑，5 项都会失败） --- backend/app/api/governance.py | 71 +++----- backend/app/api/runs.py | 50 +++++- backend/app/chat/__init__.py | 5 + backend/app/chat/questions.py | 115 ++++++++++++ backend/app/data/catalog.py | 7 +- backend/app/engine/origin.py | 18 ++ backend/tests/test_eval_origin.py | 232 +++++++++++++++++++++++++ backend/tests/test_personal_apps.py | 5 +- frontend/src/api/client.ts | 9 + frontend/src/lib/terms.ts | 15 +- frontend/src/pages/RunsPage.tsx | 5 +- frontend/src/pages/runs/RunList.tsx | 54 ++++-- frontend/src/pages/runs/useRunsData.ts | 27 ++- frontend/src/typesApp.ts | 9 + scripts/check-runs.mjs | 20 +++ 15 files changed, 570 insertions(+), 72 deletions(-) create mode 100644 backend/app/chat/__init__.py create mode 100644 backend/app/chat/questions.py create mode 100644 backend/tests/test_eval_origin.py diff --git a/backend/app/api/governance.py b/backend/app/api/governance.py index 710e1707..a5a73074 100644 --- a/backend/app/api/governance.py +++ b/backend/app/api/governance.py @@ -1,6 +1,5 @@ from __future__ import annotations -import json from collections import defaultdict from typing import Any @@ -45,11 +44,14 @@ async def tool_usage( merged.extend(agent.get("tools") or []) granted[node.id] = sorted(set(merged)) + from app.engine.origin import counted_runs + + # 只算计入统计的运行（engine/origin.counted_runs）：评测、口径对账、召回替人跑的运行用到了哪些工具不说明白名单该不该收 run_ids = [ rid for (rid,) in await session.execute( select(Run.id) - .where(Run.workflow_id == workflow_id) + .where(Run.workflow_id == workflow_id, counted_runs()) .order_by(Run.created_at.desc()) .limit(limit_runs) ) @@ -95,62 +97,39 @@ async def exploratory_clusters( ) -> dict[str, Any]: """探索性问题聚类：同一个问题被反复问，就该长成模板里的固定一节。 - 这就是模板生长的触发器。用本地 embedding 做贪心聚类， - 出现 ≥3 次的簇标记为晋升候选。 + 这就是模板生长的触发器。取问题、聚类全站只有一份（app/chat/questions.py）：只取问题原文（不含 history），问数据一轮 + 一条（自动重来不多算），评测这类系统派生的运行、别人的个人应用刷新和个人蓝图试运行都不进。出现 ≥3 次的簇标记为晋升 + 候选。embedder 是聚类用的向量模型：本地哈希的语义能力弱，结果只是下限。 """ - from sqlalchemy import not_ - from app.api.deps import current_principal - from app.auth.access import hidden_runs - from app.memory.embeddings import cosine, embed_texts + from app.chat.questions import cluster_questions, embedder, recent_questions - rows = list( - await session.execute( - select(Run.id, Run.input, Run.workflow_name, Run.created_at) - # 别人的个人应用刷新、个人蓝图试运行只给拥有者（同运行列表），不进聚类 - .where(Run.run_class != "formal", not_(hidden_runs(current_principal()))) - .order_by(Run.created_at.desc()) - .limit(limit) - ) - ) - items = [] - for rid, input_payload, wf_name, created in rows: - text = json.dumps(input_payload or {}, ensure_ascii=False, sort_keys=True) - if len(text) <= 4: # 空输入没有聚类意义 - continue - items.append({"run_id": rid, "text": text, "workflow": wf_name, - "at": created.isoformat() if created else None}) - if not items: - return {"clusters": [], "considered": 0} + questions = await recent_questions(session, current_principal(), limit=limit) + if not questions: + return {"clusters": [], "considered": 0, "embedder": embedder()} + clusters = await cluster_questions(questions, threshold=threshold) - vectors = await embed_texts([i["text"] for i in items]) - clusters: list[dict[str, Any]] = [] - for i, item in enumerate(items): - placed = False - for cluster in clusters: - if cosine(vectors[i], vectors[cluster["seed"]]) >= threshold: - cluster["members"].append(item) - placed = True - break - if not placed: - clusters.append({"seed": i, "members": [item]}) + def _at(value: Any) -> str | None: + return value.isoformat() if value else None out = [ { - "count": len(c["members"]), - "representative": c["members"][0]["text"][:200], - "workflow": c["members"][0]["workflow"], - "run_ids": [m["run_id"] for m in c["members"]][:20], - "first_at": c["members"][-1]["at"], - "last_at": c["members"][0]["at"], - "promote_candidate": len(c["members"]) >= 3, + "count": c.count, + "representative": c.members[0].text[:200], + "workflow": c.members[0].workflow, + "run_ids": [m.run_id for m in c.members if m.run_id][:20], + "turn_ids": [m.turn_id for m in c.members if m.turn_id][:20], + "first_at": _at(c.members[-1].at), + "last_at": _at(c.members[0].at), + "promote_candidate": c.count >= 3, } for c in clusters - if len(c["members"]) >= 2 + if c.count >= 2 ] out.sort(key=lambda c: c["count"], reverse=True) return { "clusters": out, - "considered": len(items), + "considered": len(questions), + "embedder": embedder(), "hint": "被标记为可固化的问题反复出现，建议用「提取模板」把它固定为工作流中的一步", } diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index 3a35364e..c6b38a3e 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -171,13 +171,54 @@ async def list_runs( 以前只有 limit：待审批的运行排在第 145 位，列表只取 100 条，点了徽标也找不到它。 筛选都走服务端，翻页用 created_at 做游标（before 传上一页最后一条的 created_at）。 别人的个人应用刷新、个人蓝图试运行不列（只给拥有者，auth/access.hidden_runs），管理员也一样。 + 没按来源筛时，系统派生的运行（评测、口径对账、口径召回：engine/origin.SYSTEM_ORIGINS）默认不列——几十道评测题的运行 + 会把真实用户的运行挤出第一页；按来源点名就列。没列出几次由 GET /api/runs/unlisted 照实说，界面写在列表底下。 """ + stmt = _filtered(select(Run), workflow_id=workflow_id, status=status, run_class=run_class, q=q, + schedule_id=schedule_id, origin_kind=origin_kind) + stmt = stmt.order_by(Run.created_at.desc()).limit(limit) + if before is not None: + stmt = stmt.where(Run.created_at < before) + return list((await session.execute(stmt)).scalars()) + + +@router.get("/unlisted") +async def unlisted_runs( + workflow_id: str | None = Query(default=None), + status: str | None = Query(default=None), + run_class: str | None = None, + q: str | None = Query(default=None), + schedule_id: str | None = Query(default=None), + origin_kind: str | None = Query(default=None), + session: AsyncSession = Depends(get_session), +) -> dict[str, Any]: + """同样的筛选下，运行列表因为「系统派生的运行默认不列」而没列出的有几次：{total, by_origin: {来源: 次数}}。 + + 按来源筛了的（origin_kind 给了）什么都不藏，total 为 0。别人的个人应用刷新、个人蓝图试运行不算在内：它们不是「没列出」， + 是读不了。""" + from sqlalchemy import func + + from app.engine.origin import SYSTEM_ORIGINS + + if _csv(origin_kind): + return {"total": 0, "by_origin": {}} + stmt = _filtered(select(Run.origin_kind, func.count()), workflow_id=workflow_id, status=status, run_class=run_class, + q=q, schedule_id=schedule_id, origin_kind=",".join(SYSTEM_ORIGINS)) + rows = (await session.execute(stmt.group_by(Run.origin_kind))).all() + by_origin = {kind: int(n) for kind, n in rows if kind and n} + return {"total": sum(by_origin.values()), "by_origin": by_origin} + + +def _filtered(stmt: Any, *, workflow_id: str | None, status: str | None, run_class: str | None, q: str | None, + schedule_id: str | None, origin_kind: str | None) -> Any: + """运行列表的筛选条件（列表和「没列出几次」共用一份）。""" from sqlalchemy import not_ from app.api.deps import current_principal from app.auth.access import hidden_runs + from app.engine.origin import counted_runs - stmt = select(Run).where(not_(hidden_runs(current_principal()))).order_by(Run.created_at.desc()).limit(limit) + stmt = stmt.where(not_(hidden_runs(current_principal()))) if workflow_id == UNSAVED: stmt = stmt.where(Run.workflow_id.is_(None)) elif workflow_id: @@ -193,9 +234,10 @@ async def list_runs( if origins := _csv(origin_kind): kinds = [o for o in origins if o != NO_ORIGIN] stmt = stmt.where(or_(Run.origin_kind.in_(kinds), *([Run.origin_kind.is_(None)] if NO_ORIGIN in origins else []))) - if before is not None: - stmt = stmt.where(Run.created_at < before) - return list((await session.execute(stmt)).scalars()) + else: + # 没点名来源：只列计入统计的运行，系统派生的（评测、口径对账、召回）不列；来源为空的老运行照常列出 + stmt = stmt.where(counted_runs()) + return stmt def _csv(value: str | None) -> list[str]: diff --git a/backend/app/chat/__init__.py b/backend/app/chat/__init__.py new file mode 100644 index 00000000..508c2ca1 --- /dev/null +++ b/backend/app/chat/__init__.py @@ -0,0 +1,5 @@ +"""问数据（一轮：提问 → 建图 → 运行 → 复核）在后端的部分。 + +- turn.py：一轮里的几个「决定」（要不要运行、输入怎么填、要不要带原图重来、重来时说什么），问数据页和评测走同一套； +- questions.py：问过的问题原文和它们的聚类，全站只有这一份（治理接口、以后的口径账本、评测都调它）。 +""" diff --git a/backend/app/chat/questions.py b/backend/app/chat/questions.py new file mode 100644 index 00000000..6e7e9201 --- /dev/null +++ b/backend/app/chat/questions.py @@ -0,0 +1,115 @@ +"""问过的问题和它们的聚类：全站只有这一份（docs/design/next-round/00-总方案.md X2）。 + +同一个问题被反复问，就该长成模板里的固定一节、沉淀成口径——探索问题聚类（/governance/exploratory-clusters）、以后口径 +账本的「可沉淀的问题」、评测的问题重复度（E3）量的都是这件事，各算一份的话几个数对不上，所以取问题和聚类只写在这里。 + +**取什么。** 只取问题原文，不取整份输入：以前把运行的整份 input（含 history）拿去聚，同一段对话里前几轮的问答会把不同的 +问题聚成一类，重复度偏高。 + +- 问数据：一轮一条（conversation_turns.question，用户原本问的那句话）。同一轮里的自动重来、用户点的「重试」只是换了这一轮 + 挂着的运行，不多出一条——按运行数的话，答坏了自动重来一次就被算成问了两遍。没挂运行的轮次（只回了一句、助手反问） + 也算：问题照样被问过。 +- 别的探索运行（画布试跑、蓝图试运行、接口……）：输入里有 question 这个字段的取它；没有的不收（输入里别的字段不是问题）。 + 问数据的运行由上面一轮一条代表，这里不再数。 + +**不收什么。** 系统派生的运行（评测、口径对账、口径召回：engine/origin.counted_runs）不收，评测题不能把聚类刷高； +这个人读不了的只给拥有者的运行（auth/access.hidden_runs）不收；正式运行不收（探索才谈得上「沉淀」）。 + +聚类用 memory/embeddings 的向量贪心聚：按时间从新到旧，和某一类的种子足够像就并进去，否则自立一类。本地向量是词频 +哈希、语义能力弱，结果是下限——调用方据 embedder 照实写明用的是哪个向量模型。 +""" +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from sqlalchemy import func, not_, select +from sqlalchemy.ext.asyncio import AsyncSession + +#: 缺省的相似度阈值（同治理接口一直以来的 0.62） +DEFAULT_THRESHOLD = 0.62 + + +@dataclass(frozen=True) +class Question: + """问过的一个问题。turn_id 是问数据的那一轮（别的探索运行没有）；run_id 是它挂着的运行（只回了一句的轮次没有）。""" + + text: str + at: datetime | None + run_id: str | None = None + turn_id: str | None = None + workflow: str | None = None + origin: str | None = None + + +@dataclass +class QuestionCluster: + """一类问题：members 按时间从新到旧，第一个是种子。""" + + members: list[Question] = field(default_factory=list) + + @property + def count(self) -> int: + return len(self.members) + + +async def recent_questions(session: AsyncSession, principal: Any, *, limit: int) -> list[Question]: + """最近问过的问题，从新到旧，最多 limit 条（取法见模块开头）。principal 用来去掉这个人读不了的只给拥有者的运行。""" + from app.auth.access import hidden_runs + from app.db.models import Conversation, ConversationTurn, Run + from app.engine.origin import counted_runs + + hidden = hidden_runs(principal) + turns = (await session.execute( + select(ConversationTurn.id, ConversationTurn.question, ConversationTurn.created_at, ConversationTurn.run_id, + Run.workflow_name) + .join(Conversation, Conversation.id == ConversationTurn.conversation_id) + .outerjoin(Run, Run.id == ConversationTurn.run_id) + .where(Conversation.kind == "chat", func.trim(ConversationTurn.question) != "", counted_runs(), not_(hidden)) + .order_by(ConversationTurn.created_at.desc()).limit(limit) + )).all() + runs = (await session.execute( + select(Run.id, Run.input, Run.workflow_name, Run.created_at, Run.origin_kind) + .where(Run.run_class != "formal", func.coalesce(Run.origin_kind, "") != "chat", counted_runs(), not_(hidden), + func.json_type(Run.input, "$.question") == "text") + .order_by(Run.created_at.desc()).limit(limit) + )).all() + out = [Question(text=question.strip(), at=at, run_id=run_id, turn_id=turn_id, workflow=workflow, origin="chat") + for turn_id, question, at, run_id, workflow in turns] + for run_id, payload, workflow, at, origin in runs: + 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) + return out[:limit] + + +async def cluster_questions(questions: Sequence[Question], *, threshold: float = DEFAULT_THRESHOLD, + ) -> list[QuestionCluster]: + """向量贪心聚类（顺序同 questions：从新到旧）。""" + from app.memory import embeddings + + if not questions: + return [] + vectors = await embeddings.embed_texts([q.text for q in questions]) + clusters: list[tuple[int, QuestionCluster]] = [] + for i, question in enumerate(questions): + for seed, cluster in clusters: + if embeddings.cosine(vectors[i], vectors[seed]) >= threshold: + cluster.members.append(question) + break + else: + clusters.append((i, QuestionCluster(members=[question]))) + return [cluster for _, cluster in clusters] + + +def embedder() -> str: + """聚类用的是哪个向量模型（memory/embeddings.embedder_id）：本地哈希的结果只是下限，调用方照实写出来。""" + from app.memory import embeddings + + return embeddings.embedder_id() + + +__all__ = ["DEFAULT_THRESHOLD", "Question", "QuestionCluster", "cluster_questions", "embedder", "recent_questions"] diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 6ca00287..3247aea4 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -2408,15 +2408,18 @@ async def query_snapshot_tables(session: AsyncSession, source: Any) -> list[tupl 使用次数（table_usage）和「这张表被哪些运行用过」（data/catalog_review.py）共用这一份取法：数据源查询工具每次成功 查询都存一份查询快照工件（tools/datasource.py），工件引用表 artifacts 里按 (内容哈希, 运行) 各记一行；只收属于某次运行 - （runs 里有这次运行）、属于这个源的行，工具库里的试用（不属于任何运行）不算。 + （runs 里有这次运行）、属于这个源的行，工具库里的试用（不属于任何运行）不算；系统派生的运行（评测、口径对账、召回， + engine/origin.counted_runs）也不算——一跑评测，常用表的排序、起草的默认范围、审阅队列的权重就跟着被评测题刷高了。 表名在查询当时就记进了引用的 meta（query_snapshot_meta），这里只读 artifacts 一张表；升级前的老引用没有 meta，退回读 工件内容里的源名和 SQL，读过的按工件 id 缓存（内容永不改变）。SQL 里写的表名去掉 schema 前缀后按 name_key（只把 ASCII 字母转小写，和 SQLite 比较标识符的口径一致）对到表结构上，对不上的不计。 """ + from app.engine.origin import counted_runs + rows = (await session.execute( select(Artifact.id, Artifact.meta, Artifact.run_id).join(Run, Run.id == Artifact.run_id) - .where(Artifact.kind == QUERY_SNAPSHOT_KIND) + .where(Artifact.kind == QUERY_SNAPSHOT_KIND, counted_runs()) )).all() tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} lookup: dict[str, str] = {} diff --git a/backend/app/engine/origin.py b/backend/app/engine/origin.py index 6e418db8..25081ff7 100644 --- a/backend/app/engine/origin.py +++ b/backend/app/engine/origin.py @@ -34,6 +34,24 @@ ORIGIN_KINDS = ( #: 客户端在 POST /api/runs 里能声明的来源 CLIENT_ORIGINS = ("chat", "canvas", "launcher", "api") +#: 系统派生的运行（docs/design/next-round/00-总方案.md X1）：评测、口径对账、口径召回都是服务端为了量、为了比替人跑的， +#: 不是有人在用。它们是完整的运行（证据、封存都在、点得开），但不进使用统计——探索问题聚类、工具白名单衰减、数据目录的 +#: 使用次数和「被哪些运行用过」、审阅队列的使用权重、以后的账本统计——否则一跑评测，治理信号和目录的使用热度就失真了。 +#: 记录页默认也不列（筛选里能选，列表照实写有几次没列出）。ledger_issue 是替用户正式出具，照常计入 +SYSTEM_ORIGINS = ("eval", "reconcile", "recall") + + +def counted_runs() -> Any: + """SQL 条件：计入使用统计的运行（排除 SYSTEM_ORIGINS）。所有使用统计都只算它圈进来的运行，别处不另写一份过滤。 + + 来源按非空比较（coalesce，同 auth/access.hidden_runs）：加来源这一列之前的老运行 origin_kind 为空，NOT IN 碰上 NULL + 会把它们一起筛掉；外连接没对上运行的行（问数据只回了一句、没有运行的那一轮）同理照常计入。""" + from sqlalchemy import func + + from app.db.models import Run + + return func.coalesce(Run.origin_kind, "").not_in(SYSTEM_ORIGINS) + def origin_note(kind: str, **data: Any) -> dict[str, Any]: """run.origin 事件的数据：kind 加上调用方给的字段（app_id、app_version、spec_hash、entry / view、hook、 diff --git a/backend/tests/test_eval_origin.py b/backend/tests/test_eval_origin.py new file mode 100644 index 00000000..951b12be --- /dev/null +++ b/backend/tests/test_eval_origin.py @@ -0,0 +1,232 @@ +"""系统派生的运行不进使用统计（docs/design/next-round/00-总方案.md X1，细案 01 做法 4）。 + +评测、口径对账、口径召回（engine/origin.SYSTEM_ORIGINS）是服务端替人跑的完整运行，证据点得开，但不是有人在用。这里守： + +- 客户端声明不了这几种来源；ledger_issue（替用户正式出具）照常计入； +- counted_runs() 只去掉这三种，来源为空的老运行照常计入（防 NULL）； +- 探索问题聚类只取问题原文（不含 history），问数据一轮一条（自动重来不多算），评测运行不进； +- 工具白名单衰减、目录的使用次数和「被哪些运行用过」、审阅队列的使用权重都不计评测运行； +- 记录页默认不列这三种，按来源点名就列；GET /api/runs/unlisted 照实说没列出几次。 + +表名、问题、工作流名都是虚构的。 +""" +from __future__ import annotations + +import uuid +from datetime import datetime, timedelta, timezone + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, select + +from app.data import catalog, catalog_review +from app.db.base import SessionLocal +from app.db.models import Artifact, Conversation, ConversationTurn, DataSource, Run, RunEvent, Workflow +from app.engine.origin import CLIENT_ORIGINS, ORIGIN_KINDS, SYSTEM_ORIGINS, counted_runs +from app.main import app +from tests.fixtures.sources import drop_source + +BASE = datetime(2026, 9, 1, tzinfo=timezone.utc) +MARK = f"计入统计-{uuid.uuid4().hex[:6]}" + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture(autouse=True) +async def _clean(): + yield + async with SessionLocal() as session: + await session.execute(delete(Run).where(Run.workflow_name.like(f"{MARK}%"))) + await session.execute(delete(Conversation).where(Conversation.title.like(f"{MARK}%"))) + await session.commit() + + +async def _run(origin: str | None, *, minutes: int = 0, **extra) -> str: + rid = uuid.uuid4().hex + async with SessionLocal() as session: + session.add(Run(id=rid, graph={}, workflow_name=f"{MARK}-{origin}", run_class="exploratory", status="succeeded", + origin_kind=origin, created_at=BASE + timedelta(minutes=minutes), **extra)) + await session.commit() + return rid + + +def test_system_origins_are_server_only_and_ledger_issue_counts(): + assert set(SYSTEM_ORIGINS) == {"eval", "reconcile", "recall"} + assert set(SYSTEM_ORIGINS) <= set(ORIGIN_KINDS) + assert not set(SYSTEM_ORIGINS) & set(CLIENT_ORIGINS) + assert "ledger_issue" in ORIGIN_KINDS and "ledger_issue" not in SYSTEM_ORIGINS, "替用户正式出具照常计入" + + +async def test_a_client_cannot_claim_a_system_origin(client): + for origin in (*SYSTEM_ORIGINS, "ledger_issue"): + r = await client.post("/api/runs", json={"graph": {"nodes": [], "edges": []}, "origin": origin}) + assert r.status_code == 422 and r.json()["code"] == "origin_not_allowed", (origin, r.text) + + +async def test_counted_runs_keeps_legacy_null_origins(): + ids = {origin: await _run(origin) for origin in (None, "chat", "ledger_issue", *SYSTEM_ORIGINS)} + async with SessionLocal() as session: + counted = set((await session.execute( + select(Run.id).where(Run.id.in_(list(ids.values())), counted_runs()))).scalars()) + assert counted == {ids[None], ids["chat"], ids["ledger_issue"]} + + +# -------------------------------------------------------------------------- +# 探索问题聚类 +# -------------------------------------------------------------------------- + + +async def _turn(conversation_id: str, question: str, run_id: str | None, seq: int, minutes: int) -> str: + async with SessionLocal() as session: + turn = ConversationTurn(conversation_id=conversation_id, seq=seq, question=question, run_id=run_id, + status="done", created_at=BASE + timedelta(minutes=minutes)) + session.add(turn) + await session.commit() + return turn.id + + +async def _conversation() -> str: + async with SessionLocal() as session: + row = Conversation(title=f"{MARK}-对话", kind="chat") + session.add(row) + await session.commit() + return row.id + + +async def test_clusters_take_only_the_question_once_per_turn_and_skip_eval(client, monkeypatch): + from app.memory import embeddings + + question = f"{MARK} 上周各门店的销售额" + + async def by_text(texts): + # 同一句话同一个向量，别的话各不相同：聚类只看取了什么文字 + return [[1.0, 0.0] if t == question else [0.0, float(i + 1)] for i, t in enumerate(texts)] + + monkeypatch.setattr(embeddings, "embed_texts", by_text) + conv = await _conversation() + # 第一轮：答坏了自动重来一次——两次运行，轮次只挂着后一次；输入里的 history 各不相同也不影响 + first = await _run("chat", minutes=1, input={"question": question, "history": "问：别的\n答：别的"}) + retried = await _run("chat", minutes=2, input={"question": question, "history": "问：别的\n答：别的"}) + turn_a = await _turn(conv, question, retried, 1, 2) + # 第二轮：同一个问题又问了一次，这次只回了一句、没有运行 + turn_b = await _turn(conv, question, None, 2, 3) + # 评测作答的运行：同一个问题，不进聚类 + graded = await _run("eval", minutes=4, input={"question": question}) + # 画布试跑：输入里有 question 的照样收 + canvas = await _run("canvas", minutes=5, input={"question": question}) + + body = (await client.get("/api/governance/exploratory-clusters", params={"limit": 500})).json() + mine = [c for c in body["clusters"] if c["representative"] == question] + assert len(mine) == 1, body + cluster = mine[0] + assert cluster["count"] == 3, "一轮一条（自动重来不多算）+ 画布那一次；评测不算" + assert set(cluster["turn_ids"]) == {turn_a, turn_b} + assert set(cluster["run_ids"]) == {retried, canvas} + assert first not in cluster["run_ids"] and graded not in cluster["run_ids"] + assert cluster["promote_candidate"] is True + assert body["embedder"] + + +async def test_the_history_never_reaches_the_clusterer(monkeypatch): + from app.api.deps import current_principal + from app.chat.questions import recent_questions + + conv = await _conversation() + rid = await _run("chat", minutes=10, input={"question": f"{MARK} 本月退款", "history": "问：上周\n答：100"}) + await _turn(conv, f"{MARK} 本月退款", rid, 1, 10) + async with SessionLocal() as session: + got = [q for q in await recent_questions(session, current_principal(), limit=500) if MARK in q.text] + assert [q.text for q in got] == [f"{MARK} 本月退款"] + + +# -------------------------------------------------------------------------- +# 工具白名单衰减 +# -------------------------------------------------------------------------- + + +async def test_tool_usage_ignores_eval_runs(client): + graph = {"nodes": [{"id": "a", "type": "agent", "position": {"x": 0, "y": 0}, + "data": {"label": "a", "config": {"tools": ["calc", "kb_search"]}}}], "edges": []} + async with SessionLocal() as session: + wf = Workflow(name=f"{MARK}-白名单", graph=graph) + session.add(wf) + await session.commit() + wf_id = wf.id + try: + used = await _run("canvas", workflow_id=wf_id) + graded = await _run("eval", workflow_id=wf_id) + async with SessionLocal() as session: + session.add(RunEvent(run_id=used, seq=1, type="tool.start", node_id="a", data={"tool": "calc"}, ts=0.0)) + session.add(RunEvent(run_id=graded, seq=1, type="tool.start", node_id="a", data={"tool": "kb_search"}, ts=0.0)) + await session.commit() + body = (await client.get("/api/governance/tool-usage", params={"workflow_id": wf_id})).json() + assert body["runs_considered"] == 1 + node = body["nodes"][0] + assert node["used"] == {"calc": 1} and node["unused"] == ["kb_search"], "评测用过的工具不能让白名单显得都在用" + finally: + async with SessionLocal() as session: + await session.execute(delete(Workflow).where(Workflow.id == wf_id)) + await session.commit() + + +# -------------------------------------------------------------------------- +# 目录的使用次数、被哪些运行用过、审阅队列的权重 +# -------------------------------------------------------------------------- + + +async def test_catalog_usage_ignores_eval_runs(): + name = f"eo_{uuid.uuid4().hex[:8]}" + tables = {"orders": {"qualified": "orders", "primary_key": ["id"], "comment": None, "is_view": False, + "foreign_keys": [], "unique": [], + "columns": [{"name": "id", "type": "INTEGER", "nullable": True, "comment": None}]}} + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=":memory:", schema_cache={"tables": tables}) + session.add(row) + await session.commit() + sid = row.id + try: + used = await _run("chat") + graded = await _run("eval") + legacy = await _run(None) + async with SessionLocal() as session: + for rid in (used, graded, legacy): + session.add(Artifact(id=uuid.uuid4().hex, run_id=rid, kind=catalog.QUERY_SNAPSHOT_KIND, + meta={"source_id": sid, "source": name, "tables": ["orders"]})) + await session.commit() + source = await session.get(DataSource, sid) + assert await catalog.table_usage(session, source) == {"orders": 2} + counts = await catalog_review.table_usage_counts(session, source) + assert counts.queries == {"orders": 2} and counts.runs == {"orders": 2} + listed = await catalog_review.table_runs(session, source, "orders") + assert listed["total"] == 2 and {r["run_id"] for r in listed["runs"]} == {used, legacy} + finally: + await drop_source(sid) + + +# -------------------------------------------------------------------------- +# 记录页 +# -------------------------------------------------------------------------- + + +async def test_the_run_list_hides_system_runs_by_default_and_says_how_many(client): + chat = await _run("chat") + graded = [await _run("eval", minutes=i) for i in range(2)] + recon = await _run("reconcile") + legacy = await _run(None) + q = {"q": MARK, "limit": 200} + + listed = {r["id"] for r in (await client.get("/api/runs", params=q)).json()} + assert {chat, legacy} <= listed and not (set(graded) | {recon}) & listed + unlisted = (await client.get("/api/runs/unlisted", params={"q": MARK})).json() + assert unlisted == {"total": 3, "by_origin": {"eval": 2, "reconcile": 1}} + + # 按来源点名就列，此时没有藏起来的 + only = {r["id"] for r in (await client.get("/api/runs", params={**q, "origin_kind": "eval"})).json()} + assert only == set(graded) + assert (await client.get("/api/runs/unlisted", params={"q": MARK, "origin_kind": "eval"})).json()["total"] == 0 + # 同样的筛选一起作用：按状态筛时只数这个状态的 + assert (await client.get("/api/runs/unlisted", params={"q": MARK, "status": "failed"})).json()["total"] == 0 diff --git a/backend/tests/test_personal_apps.py b/backend/tests/test_personal_apps.py index 8439a8d7..ff25c447 100644 --- a/backend/tests/test_personal_apps.py +++ b/backend/tests/test_personal_apps.py @@ -964,9 +964,10 @@ async def test_the_other_doors_to_a_private_run_are_shut_too(monkeypatch): bob_trial = await wait(r.json()["run_id"]) question = {"question": "上周各门店的销售额"} async with SessionLocal() as session: - # 聚类只列两条以上的簇：别的探索运行放两条 + # 聚类只列两条以上的簇：别的探索运行放两条。用画布试跑：问数据的问题按轮次取（app/chat/questions.py）， + # 没挂在任何一轮上的问数据运行不单独算 shared, also = (Run(workflow_name="共享的运行", graph=KPI, status="succeeded", run_class="exploratory", - origin_kind="chat", started_by="bob", input=question) for _ in range(2)) + origin_kind="canvas", started_by="bob", input=question) for _ in range(2)) session.add_all([shared, also]) await session.flush() for run_id in (trial.id, refresh.id, bob_trial.id): diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 832089de..eafdec5b 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -13,6 +13,7 @@ import type { RunEvent, RunStatus, Skill, ToolChange, ToolInfo, ToolTrust, UpgradeResult, UploadDecision, UploadMixedColumn, UploadResult, UploadShapeReason, ValidationIssue, VarIssue, Variable, Workflow, WorkflowVersion, } from '../types' +import type { UnlistedRuns } from '../typesApp' import { ApiError, BASE, NETWORK_MESSAGE, actorHeader, codeOf, del, describeDetail, failure, get, isGatewayFailure, parseBody, patch, post, put, qs, report, request, upload, type RequestOptions, type UploadOptions, @@ -563,6 +564,14 @@ export const api = { /** 按运行来源筛（X3）；none 是来源未记录的老运行 */ origin_kind?: string }) => get(`/runs${qs(params)}`), + /** + * 同样的筛选下，列表因为「系统派生的运行默认不列」而没列出的次数（评测、口径对账、口径召回，next-round X1）。按来源 + * 筛了的什么都不藏，total 为 0。老后端没有这个接口（404），调用方当作 0 + */ + unlisted: (params?: { + workflow_id?: string; status?: RunStatus | RunStatus[] | string; run_class?: 'formal' | 'exploratory' + q?: string; origin_kind?: string + }) => get(`/runs/unlisted${qs(params)}`), get: (id: string) => get(`/runs/${id}`), /** 这次运行实际跑的那张图（运行时的快照），不是工作流现在的样子 */ graph: (id: string) => get<{ graph: GraphSpec; workflow_id: string | null; version: number | null }>( diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 51fb6dba..fd44bf8d 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -5124,7 +5124,8 @@ export const ORIGIN_TEXT = { unknown: '来源未记录', unknownHint: '这次运行发起时还没有记录来源', hint: '这次运行从哪里发起。发起时写进运行记录，在封存范围内', - filterAll: '全部来源', + // 不点名来源时系统派生的运行（评测、口径对账、口径召回）不列，「全部」得照实说不含它们 + filterAll: '全部来源（不含系统运行）', filterLabel: '按运行来源筛选', /** 时间线里 run.origin 那一行 */ step: (label: string) => `来源：${label}`, @@ -5135,6 +5136,18 @@ export const ORIGIN_TEXT = { actor: (who: string, verified: boolean) => `${verified ? '已登录用户' : '署名'}「${who}」`, } +/** + * 记录页没列出的系统派生运行（评测、口径对账、口径召回：服务端 engine/origin.SYSTEM_ORIGINS，next-round X1）。它们是完整的 + * 运行、不进使用统计，默认不列；照实写有几次没列出、在哪儿能看 + */ +export const SYSTEM_RUNS_TEXT = { + /** parts 是「评测 2 次、口径对账 1 次」 */ + unlisted: (total: number, parts: string) => + `另有 ${total} 次系统运行未列出（${parts}），在来源筛选中选择对应来源即可查看`, + part: (label: string, n: number) => `${label} ${n} 次`, + hint: '评测、口径对账、口径召回是系统代为发起的运行：证据仍可查看，但不计入使用统计，记录页默认不列', +} + /** 来源标签：老运行没有来源时写「来源未记录」 */ export const originLabel = (kind: string | null | undefined): string => kind && kind in ORIGIN_LABEL ? ORIGIN_LABEL[kind as OriginKind] : ORIGIN_TEXT.unknown diff --git a/frontend/src/pages/RunsPage.tsx b/frontend/src/pages/RunsPage.tsx index af6fdf58..10a3f3fc 100644 --- a/frontend/src/pages/RunsPage.tsx +++ b/frontend/src/pages/RunsPage.tsx @@ -18,7 +18,7 @@ import { } from './runs/model' import { ORIGIN_TEXT, TASK_LOG_TEXT, UNSAVED_NAME, originLabel } from '../lib/terms' import { RunTabs, type TabItem } from './runs/parts' -import { useApprovalQueue, useRunList, useStatusCounts, useTaskQueue } from './runs/useRunsData' +import { useApprovalQueue, useRunList, useStatusCounts, useTaskQueue, useUnlistedRuns } from './runs/useRunsData' import './runs/runs.css' /** @@ -95,6 +95,8 @@ export function RunsPage() { const filters = { codes, runClass, workflowId, q: parsed.q, origin } const list = useRunList(filters, tab !== 'approvals' && !impossible) + // 系统派生的运行（评测、口径对账、召回）默认不列：列表底下照实写有几次没列出 + const unlisted = useUnlistedRuns(filters, epoch, tab !== 'approvals' && !impossible) // 计数两份：页签和分段上的跟着当前的类别、工作流、名称条件走；右侧总览说的 // 是全局——筛掉了不代表事情没了 const narrowed = !!(runClass || workflowId || parsed.q || origin) @@ -251,6 +253,7 @@ export function RunsPage() { emptyBody={tab === 'all' ? '在问数据或画布中发起运行后，完整轨迹会记录在这里。' : undefined} filtered={filterNote} onClear={clearFilters} + unlisted={unlisted} /> )} diff --git a/frontend/src/pages/runs/RunList.tsx b/frontend/src/pages/runs/RunList.tsx index 01e9e699..0c0989cc 100644 --- a/frontend/src/pages/runs/RunList.tsx +++ b/frontend/src/pages/runs/RunList.tsx @@ -7,12 +7,12 @@ import { formatCost, formatDateTime, formatDay, formatSpan, formatTime, formatTokens, parseServerTime, shortId, } from '../../lib/format' import { - ORIGIN_LABEL, ORIGIN_TEXT, RUN_CLASS_LABEL, TASK_KIND_LABEL, TASK_LOG_TEXT, UNSAVED_HINT, UNSAVED_NAME, isUnsaved, originLabel, - runName, + ORIGIN_LABEL, ORIGIN_TEXT, RUN_CLASS_LABEL, SYSTEM_RUNS_TEXT, TASK_KIND_LABEL, TASK_LOG_TEXT, UNSAVED_HINT, UNSAVED_NAME, isUnsaved, + originLabel, runName, } from '../../lib/terms' import { summarizeRun } from '../../run/decode' import type { Approval, Run, Workflow } from '../../types' -import type { OriginKind, Task } from '../../typesApp' +import type { OriginKind, Task, UnlistedRuns } from '../../typesApp' import { UNSAVED_WORKFLOW_ID } from '../../api/client' import { explainRunError } from '../../lib/explain' import { @@ -195,7 +195,7 @@ function Segment({ on, onClick, children, data, title }: { // ------------------------------------------------------------------------- export function RunRows({ - list, codeOf, pendingOf, selectedId, onOpen, dupNames, emptyTitle, emptyBody, filtered, onClear, + list, codeOf, pendingOf, selectedId, onOpen, dupNames, emptyTitle, emptyBody, filtered, onClear, unlisted, }: { list: RunList codeOf: (run: Run) => StatusCode @@ -208,6 +208,8 @@ export function RunRows({ /** 带着筛选条件：空了要说"没有匹配"，不能说"还没有运行记录" */ filtered: string | null onClear: () => void + /** 没列出的系统派生运行（评测、口径对账、口径召回）。没有或点名了来源时不给 */ + unlisted?: UnlistedRuns | null }) { const changed = useChangedRows(list.rows) const now = useTicker(60_000) @@ -221,18 +223,25 @@ export function RunRows({ return void list.reload()} compact className="m-3" /> } const rows = list.rows + // 空着的时候也照实说有几次没列出：不能让「还没有运行记录」盖住几十次评测运行 + const note = unlistedNote(unlisted) if (!rows.length) { - return filtered - ? ( - } - title="没有匹配的运行" - body={filtered} - offline={false} - action={} - /> - ) - : } title={emptyTitle} body={emptyBody} /> + return ( + <> + {filtered + ? ( + } + title="没有匹配的运行" + body={filtered} + offline={false} + action={} + /> + ) + : } title={emptyTitle} body={emptyBody} />} + {note} + + ) } // 按天分组，组头吸顶：翻到哪天一眼可见 @@ -278,10 +287,25 @@ export function RunRows({ )} {!list.hasMore && rows.length > 0 && · 已全部加载} + {note} ) } +/** 「另有 N 次系统运行未列出（评测 2 次、口径对账 1 次）……」：没有没列出的就不画 */ +function unlistedNote(unlisted: UnlistedRuns | null | undefined): ReactNode { + if (!unlisted?.total) return null + const parts = (Object.keys(ORIGIN_LABEL) as OriginKind[]) + .filter((k) => (unlisted.by_origin[k] ?? 0) > 0) + .map((k) => SYSTEM_RUNS_TEXT.part(ORIGIN_LABEL[k], unlisted.by_origin[k] ?? 0)) + .join('、') + return ( +

+ {SYSTEM_RUNS_TEXT.unlisted(unlisted.total, parts)} +

+ ) +} + /** * 实时刷新时哪几行的状态刚变了：给它们闪一下底色。值是第几次变化，行里那层 * 闪光按它换 key——同一行接连变两次（停到审批上、又被批掉接着跑）会重新挂载、 diff --git a/frontend/src/pages/runs/useRunsData.ts b/frontend/src/pages/runs/useRunsData.ts index e96a3a4a..cc8f8f6e 100644 --- a/frontend/src/pages/runs/useRunsData.ts +++ b/frontend/src/pages/runs/useRunsData.ts @@ -3,7 +3,7 @@ import { api } from '../../api/client' import { toast } from '../../components/ui' import type { StatusCode } from '../../lib/status' import type { Approval, Run } from '../../types' -import type { Task } from '../../typesApp' +import type { Task, UnlistedRuns } from '../../typesApp' import { useCatalog } from '../../store/catalog' import { runCode, serverStatuses } from './model' @@ -123,6 +123,31 @@ export function useRunList(filters: ListFilters, enabled: boolean): RunList { return { rows, state, error, hasMore, loadingMore, reload, loadMore, patchRow, removeRow } } +// ------------------------------------------------------------------------- +// 没列出的系统运行 +// ------------------------------------------------------------------------- + +/** + * 同样的筛选下，列表没列出的系统派生运行（评测、口径对账、口径召回）有几次。点名了来源的什么都不藏，不用问；老后端 + * 没有这个接口时当作 0（不报错：列表本身照常） + */ +export function useUnlistedRuns(filters: ListFilters, epoch: number, enabled: boolean): UnlistedRuns | null { + const [got, setGot] = useState(null) + const params = paramsOf(filters) + const key = JSON.stringify(params) + const ask = enabled && !params.origin_kind + + useEffect(() => { + if (!ask) { setGot(null); return } + let live = true + api.runs.unlisted(JSON.parse(key)) + .then((r) => { if (live) setGot(r) }, () => { if (live) setGot(null) }) + return () => { live = false } + }, [key, epoch, ask]) + + return ask ? got : null +} + // ------------------------------------------------------------------------- // 计数 // ------------------------------------------------------------------------- diff --git a/frontend/src/typesApp.ts b/frontend/src/typesApp.ts index df2b6c2a..949d192a 100644 --- a/frontend/src/typesApp.ts +++ b/frontend/src/typesApp.ts @@ -705,6 +705,15 @@ export type OriginKind = | 'trial' | 'personal_app' | 'eval' | 'ledger_issue' | 'reconcile' | 'recall' +/** + * GET /api/runs/unlisted：同样的筛选下，记录页因为「系统派生的运行默认不列」而没列出的次数。系统派生的是评测、口径对账、 + * 口径召回（服务端 engine/origin.SYSTEM_ORIGINS）：不进使用统计，按来源点名才列 + */ +export interface UnlistedRuns { + total: number + by_origin: Partial> +} + /** 查看的对象。caliber 是已发布口径卡上的指标（蓝图助手复用口径时看，服务端 look/calibers.py） */ export interface LookObject { kind: 'table' | 'capability' | 'workflow_version' | 'caliber' diff --git a/scripts/check-runs.mjs b/scripts/check-runs.mjs index 7e82526f..80c41fe3 100644 --- a/scripts/check-runs.mjs +++ b/scripts/check-runs.mjs @@ -2407,6 +2407,26 @@ await section('运行来源：标签与筛选', async () => { check('只剩那一种来源', (await ids()).join(',') === 'fake0origin0canvas', (await ids()).join(',')) await sel.selectOption('') await settle() + // next-round X1：系统派生的运行（评测、口径对账、召回）默认不列，列表底下照实写有几次没列出；点名来源就不再说 + const unlistedAsked = [] + fakes.set('GET /api/runs/unlisted', (url) => { + unlistedAsked.push(url.searchParams.get('origin_kind')) + return { status: 200, json: { total: 3, by_origin: { eval: 2, reconcile: 1 } } } + }) + await page.goto(`${WEB}/runs`, { waitUntil: 'networkidle' }) + await page.waitForTimeout(300) + const note = page.locator('[data-runs-unlisted]') + const noteText = await note.innerText().catch(() => '') + check('列表底下写明另有几次系统运行未列出、各是什么', await note.count() === 1 + && noteText.includes('另有 3 次系统运行未列出') && noteText.includes('评测 2 次') && noteText.includes('口径对账 1 次'), noteText) + check('来源下拉的「全部」照实说不含系统运行', (await sel.locator('option').first().innerText()).includes('不含系统运行')) + await sel.selectOption('eval') + await settle() + check('点名了来源：不再问、不再写没列出', await page.locator('[data-runs-unlisted]').count() === 0 + && !unlistedAsked.includes('eval'), unlistedAsked.join(',')) + await sel.selectOption('') + await settle() + fakes.delete('GET /api/runs/unlisted') fakes.delete('GET /api/runs') if (base) { await page.goto(`${WEB}/runs/${base.id}`, { waitUntil: 'networkidle' })