mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 22:11:58 +00:00
ThearchyHelios
0d9f1d9339
- 实测 /api/evals/calibers、/api/schedules/trigger-options 按 Workflow.name 排序遇同名并列时 SQLite、PostgreSQL 顺序不同。用 AST 逐个过了一遍 order_by：最后一个排序键是 created_at / updated_at / last_active_at / at / name / 各种 *_at 之类、不唯一的列，又不带 DISTINCT、GROUP BY 的，补「, 模型.id」（降序补 id.desc()），共 104 处，分布在 api、apps、auth、blueprints、chat、data、engine、eval、ledger、look、memory、process、providers、scheduler（runs.py 里的三处在上一个提交里） - 按唯一列（seq、version、唯一的 name）排的不动；按聚合、coalesce 之类表达式排的说明见 db/postgres
384 lines
20 KiB
Python
384 lines
20 KiB
Python
"""一份口径的旧版本被谁用过（docs/design/next-round/02-口径账本.md 做法 7 一）：口径发布新版本之后，召回要追的那些答案。
|
||
|
||
**为什么单独一个模块。** 召回（ledger/recall）要按参数去重成组、重算，发布弹窗要预告规模，版本历史要写「这一版被多少问数
|
||
回答、导出、召回用过」——三处要的是同一份「用过」的清单。各数各的，预告的规模和实际召回的就对不上（验收要求两者一致）。
|
||
|
||
反查的对象（工作流 W 的旧版本 V，只看 V < 新版本）：
|
||
- **问数回答**（answer）：caliber_match 工件的 meta 里记着 W@V 等价已证明（含助手选用）的探索运行，取这次运行现在这次封存
|
||
的最新一份比对（ledger/store.latest：之后补比过、按新口径不再等价的不算），每条等价的查询一份参数（绑定）；
|
||
- **W@V 的正式运行**：定时任务发起的（schedule，每个定时任务最近 PERIODS 期）、应用发起的（app，每个应用最近 PERIODS 次）、
|
||
别处发起的（formal：手动、接口、按口径出具……），参数是那次运行的输入；
|
||
- **固定了 W@V 的工作流版本发起的正式运行**（pinned：子工作流或口径卡 caliber_from 固定在 W@V，apps/versions.pinned_upstream
|
||
同一个认法），参数是那次运行的输入里 W 的入口有的字段；
|
||
- **固定这一版的应用**（apps/versions.pinning_apps）：应用发起过的正式运行在上面；还没有运行的只列出来，没有可重算的参数；
|
||
- **导出记录**：上面这些运行上的 report.exported 审计（第 2 波），导出人也是答案的主人；
|
||
- **问数轮次**：运行 origin 里封存的 conversation_id / turn_id（origin_ref，第 2 波），老运行退回 conversation_turns.run_id。
|
||
|
||
**答案的主人**（owners，召回待办派给谁）：对话的 created_by（没有的退回运行的发起人）、导出人、定时任务的 created_by、应用的
|
||
管理者（授权里的 manage，没有就是建应用的人）；别处发起的正式运行是它的发起人。只记 (user|group, 名字)，组由开待办时展开。
|
||
|
||
只读应用库：不查业务库、不调模型。系统派生的运行（评测、对账、召回，engine/origin.counted_runs）不算「用过」。数量有上限
|
||
（MAX_ANSWERS、MAX_FORMAL），按新的在前取，超出的照实算进 truncated。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Iterable, Mapping
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from sqlalchemy import and_, func, or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.db.models import Artifact, Run, Workflow, WorkflowVersion
|
||
|
||
#: 用法的种类。answer 问数回答；schedule 定时任务发起的正式运行；app 应用发起的；formal 别处发起的；pinned 固定了这一版的
|
||
#: 别的工作流版本发起的
|
||
KINDS = ("answer", "schedule", "app", "formal", "pinned")
|
||
#: 每个定时任务、每个应用最近几期（细案「定时任务最近 K 期」）
|
||
PERIODS = 4
|
||
#: 最多看多少次问数回答、多少次正式运行（各按新的在前）
|
||
MAX_ANSWERS = 2000
|
||
MAX_FORMAL = 500
|
||
#: 种类的叫法（版本用量、召回批次、待办里写「来自问数据」……）
|
||
KIND_TEXT = {"answer": "问数回答", "schedule": "定时任务", "app": "应用", "formal": "正式运行",
|
||
"pinned": "固定了它的工作流"}
|
||
#: 导出审计的动作（api/evidence.py、api/bundle.py 写的）
|
||
EXPORTED = "report.exported"
|
||
|
||
|
||
@dataclass
|
||
class Use:
|
||
"""一次「用过」：哪次运行、旧版本几、参数（原样，没补缺省值）、什么时候、谁是答案的主人。"""
|
||
|
||
kind: str
|
||
run_id: str
|
||
version: int
|
||
binding: dict[str, Any]
|
||
at: datetime | None = None
|
||
owners: list[tuple[str, str]] = field(default_factory=list)
|
||
conversation_id: str | None = None
|
||
turn_id: str | None = None
|
||
schedule_id: str | None = None
|
||
app_id: str | None = None
|
||
#: 问数回答：等价的是哪条查询（查询快照）、口径的哪个取数节点
|
||
query: str | None = None
|
||
node_id: str | None = None
|
||
#: 固定了这一版的是哪个工作流的哪一版（pinned）
|
||
via_workflow: str | None = None
|
||
via_version: int | None = None
|
||
exporters: list[str] = field(default_factory=list)
|
||
|
||
def add_owner(self, kind: str, name: str | None) -> None:
|
||
if name and (kind, name) not in self.owners:
|
||
self.owners.append((kind, name))
|
||
|
||
def describe(self) -> dict[str, Any]:
|
||
"""记进召回运行 origin 的那一份（召回结束时据此开待办、在原答案上记账）：只放编号、参数的来处和主人。"""
|
||
out: dict[str, Any] = {"kind": self.kind, "run_id": self.run_id, "version": self.version,
|
||
"owners": [list(o) for o in self.owners]}
|
||
for key in ("conversation_id", "turn_id", "schedule_id", "app_id", "query", "node_id", "via_workflow",
|
||
"via_version"):
|
||
if (value := getattr(self, key)) is not None:
|
||
out[key] = value
|
||
if self.exporters:
|
||
out["exporters"] = list(self.exporters)
|
||
if self.at is not None:
|
||
out["at"] = self.at.isoformat()
|
||
return out
|
||
|
||
|
||
@dataclass
|
||
class Found:
|
||
"""反查的结果：用法、固定着旧版本的应用（{版本: [应用]}）、有没有因为上限少看了。"""
|
||
|
||
uses: list[Use]
|
||
apps: dict[int, list[dict[str, Any]]] = field(default_factory=dict)
|
||
truncated: bool = False
|
||
|
||
def versions(self) -> list[int]:
|
||
return sorted({u.version for u in self.uses} | set(self.apps))
|
||
|
||
|
||
# ==========================================================================
|
||
# 反查
|
||
# ==========================================================================
|
||
|
||
|
||
async def _answers(session: AsyncSession, workflow_id: str, below: int | None, only: int | None
|
||
) -> tuple[list[Use], bool]:
|
||
"""问数回答：比对工件 meta 里等价已证明的那几份口径里有 W@V 的探索运行（见模块说明）。"""
|
||
from app.engine.origin import counted_runs
|
||
from app.ledger import store
|
||
from app.ledger.labels import caliber_key, candidates_of
|
||
|
||
prefix = f"{workflow_id}@"
|
||
rows = (await session.execute(
|
||
select(Artifact.run_id, Artifact.meta, Run.finished_at).join(Run, Run.id == Artifact.run_id)
|
||
.where(Artifact.kind == store.KIND, Run.status == "succeeded", counted_runs())
|
||
.order_by(Run.finished_at.desc(), Run.id.desc()))).all()
|
||
wanted: dict[str, datetime | None] = {}
|
||
for run_id, meta, at in rows:
|
||
keys = [k for k in ((meta or {}).get("calibers") or []) if isinstance(k, str) and k.startswith(prefix)]
|
||
if not keys or run_id in wanted:
|
||
continue
|
||
versions = {int(k.rsplit("@", 1)[1]) for k in keys if k.rsplit("@", 1)[1].isdigit()}
|
||
if any((only is None or v == only) and (below is None or v < below) for v in versions):
|
||
wanted[run_id] = at
|
||
truncated = len(wanted) > MAX_ANSWERS
|
||
out: list[Use] = []
|
||
for run_id, at in list(wanted.items())[:MAX_ANSWERS]:
|
||
recorded = await store.latest(run_id)
|
||
content = recorded.content if recorded is not None else None
|
||
if not isinstance(content, dict):
|
||
continue
|
||
seen: set[tuple[int, str]] = set()
|
||
for query in content.get("queries") or []:
|
||
# 逐格证明了的也是用了这份口径（ledger/labels.candidates_of）
|
||
for cand in candidates_of(query):
|
||
if cand.get("status") != "proven" or cand.get("workflow_id") != workflow_id:
|
||
continue
|
||
version = cand.get("version")
|
||
if not isinstance(version, int) or (only is not None and version != only) \
|
||
or (below is not None and version >= below):
|
||
continue
|
||
if cand.get("caliber") != caliber_key(workflow_id, version):
|
||
continue
|
||
binding = dict(cand.get("binding") or {})
|
||
key = (version, _key(binding))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(Use(kind="answer", run_id=run_id, version=version, binding=binding, at=at,
|
||
query=query.get("artifact"), node_id=cand.get("node_id")))
|
||
await _chat_owners(session, [u for u in out])
|
||
return out, truncated
|
||
|
||
|
||
def _key(binding: Mapping[str, Any]) -> str:
|
||
from app.apps.resolve import input_key
|
||
|
||
return input_key(dict(binding))
|
||
|
||
|
||
async def _chat_owners(session: AsyncSession, found: list[Use]) -> None:
|
||
"""问数回答的轮次和主人：运行 origin 里封存的对话、轮次（origin_ref），老运行退回 conversation_turns.run_id；主人是
|
||
对话的 created_by，没有的是运行的发起人。"""
|
||
from app.db.models import Conversation, ConversationTurn
|
||
|
||
ids = sorted({u.run_id for u in found})
|
||
if not ids:
|
||
return
|
||
runs = {rid: (origin, started) for rid, origin, started in (await session.execute(
|
||
select(Run.id, Run.origin, Run.started_by).where(Run.id.in_(ids)))).all()}
|
||
turns: dict[str, tuple[str, str]] = {}
|
||
for rid, (origin, _) in runs.items():
|
||
if isinstance(origin, dict) and origin.get("conversation_id") and origin.get("turn_id"):
|
||
turns[rid] = (str(origin["conversation_id"]), str(origin["turn_id"]))
|
||
missing = [rid for rid in ids if rid not in turns]
|
||
if missing:
|
||
for turn_id, conv, rid in (await session.execute(
|
||
select(ConversationTurn.id, ConversationTurn.conversation_id, ConversationTurn.run_id)
|
||
.where(ConversationTurn.run_id.in_(missing)).order_by(ConversationTurn.created_at, ConversationTurn.id))).all():
|
||
turns.setdefault(rid, (conv, turn_id))
|
||
convs = sorted({c for c, _ in turns.values()})
|
||
owners = dict((await session.execute(select(Conversation.id, Conversation.created_by)
|
||
.where(Conversation.id.in_(convs)))).all()) if convs else {}
|
||
for use in found:
|
||
if use.run_id in turns:
|
||
use.conversation_id, use.turn_id = turns[use.run_id]
|
||
owner = owners.get(use.conversation_id) if use.conversation_id else None
|
||
use.add_owner("user", owner or runs.get(use.run_id, (None, None))[1])
|
||
|
||
|
||
async def _app_managers(session: AsyncSession, app_ids: Iterable[str]) -> dict[str, list[tuple[str, str]]]:
|
||
"""应用的管理者：授权里的 manage（人或组）；没有授权管理者的是建应用的人。"""
|
||
from app.db.models_apps import App, AppGrant
|
||
|
||
ids = sorted({a for a in app_ids if a})
|
||
if not ids:
|
||
return {}
|
||
out: dict[str, list[tuple[str, str]]] = {a: [] for a in ids}
|
||
for app_id, kind, name in (await session.execute(
|
||
select(AppGrant.app_id, AppGrant.principal_kind, AppGrant.principal)
|
||
.where(AppGrant.app_id.in_(ids), AppGrant.role == "manage"))).all():
|
||
if kind in ("user", "group") and name and (kind, name) not in out[app_id]:
|
||
out[app_id].append((kind, name))
|
||
for app_id, created_by in (await session.execute(select(App.id, App.created_by).where(App.id.in_(ids)))).all():
|
||
if not out[app_id] and created_by:
|
||
out[app_id].append(("user", created_by))
|
||
return out
|
||
|
||
|
||
async def _formal(session: AsyncSession, clause: Any, *, kind_of: Any) -> tuple[list[Use], bool]:
|
||
"""满足 clause 的已封存、成功的正式运行（新的在前，最多 MAX_FORMAL 次）→ 用法。定时任务、应用各只留最近 PERIODS 期。
|
||
kind_of(run) 给出这次运行算哪一种（schedule / app / formal / pinned）和旧版本。"""
|
||
from app.db.models import Schedule
|
||
from app.engine.origin import counted_runs
|
||
|
||
rows = list((await session.execute(
|
||
select(Run).where(clause, Run.run_class == "formal", Run.status == "succeeded",
|
||
Run.manifest_hash.is_not(None), counted_runs())
|
||
.order_by(Run.finished_at.desc(), Run.id.desc()).limit(MAX_FORMAL + 1))).scalars())
|
||
truncated = len(rows) > MAX_FORMAL
|
||
per: dict[tuple[str, str], int] = {}
|
||
out: list[Use] = []
|
||
for run in rows[:MAX_FORMAL]:
|
||
kind, version, via = kind_of(run)
|
||
group = ("schedule", run.schedule_id) if run.schedule_id else ("app", run.app_id) if run.app_id else None
|
||
if group is not None:
|
||
per[group] = per.get(group, 0) + 1
|
||
if per[group] > PERIODS:
|
||
continue
|
||
use = Use(kind=kind, run_id=run.id, version=version, binding=dict(run.input or {}), at=run.finished_at,
|
||
schedule_id=run.schedule_id, app_id=run.app_id)
|
||
if via is not None:
|
||
use.via_workflow, use.via_version = via
|
||
out.append(use)
|
||
schedules = sorted({u.schedule_id for u in out if u.schedule_id})
|
||
makers = dict((await session.execute(select(Schedule.id, Schedule.created_by)
|
||
.where(Schedule.id.in_(schedules)))).all()) if schedules else {}
|
||
managers = await _app_managers(session, [u.app_id for u in out if u.app_id])
|
||
starters = dict((await session.execute(select(Run.id, Run.started_by).where(
|
||
Run.id.in_([u.run_id for u in out])))).all()) if out else {}
|
||
for use in out:
|
||
if use.schedule_id:
|
||
use.add_owner("user", makers.get(use.schedule_id) or starters.get(use.run_id))
|
||
elif use.app_id:
|
||
for owner in managers.get(use.app_id) or []:
|
||
use.add_owner(*owner)
|
||
else:
|
||
use.add_owner("user", starters.get(use.run_id))
|
||
return out, truncated
|
||
|
||
|
||
def _pins(graph: Any, workflow_id: str) -> set[int]:
|
||
"""一张图里固定了 workflow_id 的哪几版（子工作流、口径卡的 caliber_from；同 apps/versions.pinned_upstream 的认法，
|
||
直接读 JSON，不为了这个把整张图解析成 GraphSpec）。"""
|
||
out: set[int] = set()
|
||
for node in (graph or {}).get("nodes") or [] if isinstance(graph, dict) else []:
|
||
if not isinstance(node, dict):
|
||
continue
|
||
cfg = (node.get("data") or {}).get("config") if isinstance(node.get("data"), dict) else None
|
||
if not isinstance(cfg, dict):
|
||
continue
|
||
if node.get("type") == "subgraph":
|
||
wid, version = cfg.get("workflow_id"), cfg.get("workflow_version")
|
||
elif node.get("type") == "metrics" and isinstance(cfg.get("caliber_from"), dict):
|
||
wid, version = cfg["caliber_from"].get("workflow_id"), cfg["caliber_from"].get("workflow_version")
|
||
else:
|
||
continue
|
||
if wid == workflow_id and str(version or "").isdigit():
|
||
out.add(int(version))
|
||
return out
|
||
|
||
|
||
async def pinned_versions(session: AsyncSession, workflow_id: str) -> dict[tuple[str, int], set[int]]:
|
||
"""别的工作流的已发布版本里固定了 workflow_id 的：{(工作流, 版本): 固定的 W 的版本}。"""
|
||
rows = (await session.execute(
|
||
select(WorkflowVersion.workflow_id, WorkflowVersion.version, WorkflowVersion.graph)
|
||
.where(WorkflowVersion.workflow_id != workflow_id, WorkflowVersion.level.is_not(None)))).all()
|
||
return {(wid, int(v)): pins for wid, v, graph in rows if (pins := _pins(graph, workflow_id))}
|
||
|
||
|
||
async def find_uses(session: AsyncSession, workflow_id: str, *, below: int | None = None,
|
||
only: int | None = None) -> Found:
|
||
"""W 的旧版本被谁用过（见模块说明）。below：只看版本号小于它的（召回：新版本）；only：只看这一版（版本用量）。"""
|
||
from app.apps.versions import pinning_apps
|
||
|
||
def version_ok(v: Any) -> bool:
|
||
return isinstance(v, int) and (only is None or v == only) and (below is None or v < below)
|
||
|
||
answers, cut_a = await _answers(session, workflow_id, below, only)
|
||
own = Run.workflow_id == workflow_id
|
||
if only is not None:
|
||
own = and_(own, Run.version == only)
|
||
elif below is not None:
|
||
own = and_(own, Run.version < below)
|
||
formal, cut_f = await _formal(session, own, kind_of=lambda r: (
|
||
"schedule" if r.schedule_id else "app" if r.app_id else "formal", int(r.version), None))
|
||
pins = await pinned_versions(session, workflow_id)
|
||
pinned: list[Use] = []
|
||
cut_p = False
|
||
targets = {key: sorted(v for v in versions if version_ok(v)) for key, versions in pins.items()}
|
||
targets = {key: vs for key, vs in targets.items() if vs}
|
||
if targets:
|
||
clause = or_(*(and_(Run.workflow_id == wid, Run.version == v) for (wid, v) in targets))
|
||
pinned, cut_p = await _formal(session, clause, kind_of=lambda r: (
|
||
"pinned", targets[(r.workflow_id, int(r.version))][-1], (r.workflow_id, int(r.version))))
|
||
uses = [*answers, *formal, *pinned]
|
||
await _exporters(session, uses)
|
||
apps: dict[int, list[dict[str, Any]]] = {}
|
||
for item in await pinning_apps(session, workflow_id):
|
||
if version_ok(item.get("version")):
|
||
apps.setdefault(int(item["version"]), []).append(item)
|
||
return Found(uses=uses, apps=apps, truncated=cut_a or cut_f or cut_p)
|
||
|
||
|
||
async def _exporters(session: AsyncSession, uses: list[Use]) -> None:
|
||
"""这些运行上的报告导出审计：导出人进答案的主人。"""
|
||
from app.db.models import AuditEvent
|
||
|
||
ids = sorted({u.run_id for u in uses})
|
||
if not ids:
|
||
return
|
||
by: dict[str, list[str]] = {}
|
||
for rid, actor in (await session.execute(select(AuditEvent.subject_id, AuditEvent.actor).where(
|
||
AuditEvent.action == EXPORTED, AuditEvent.subject_kind == "run",
|
||
AuditEvent.subject_id.in_(ids)).order_by(AuditEvent.at, AuditEvent.id))).all():
|
||
if actor and actor not in by.setdefault(rid, []):
|
||
by[rid].append(actor)
|
||
for use in uses:
|
||
use.exporters = list(by.get(use.run_id) or [])
|
||
for actor in use.exporters:
|
||
use.add_owner("user", actor)
|
||
|
||
|
||
# ==========================================================================
|
||
# 版本用量（版本历史、发布弹窗）
|
||
# ==========================================================================
|
||
|
||
|
||
async def exports_count(session: AsyncSession, run_ids: Iterable[str]) -> int:
|
||
from app.db.models import AuditEvent
|
||
|
||
ids = sorted(set(run_ids))
|
||
if not ids:
|
||
return 0
|
||
return int(await session.scalar(select(func.count()).select_from(AuditEvent).where(
|
||
AuditEvent.action == EXPORTED, AuditEvent.subject_kind == "run", AuditEvent.subject_id.in_(ids))) or 0)
|
||
|
||
|
||
def summarize(found: Found) -> dict[str, Any]:
|
||
"""用法的计数（不含参数和主人）：问数回答几次（运行数）、导出几次、定时任务几个、应用几个、各种正式运行几次。"""
|
||
answers = {u.run_id for u in found.uses if u.kind == "answer"}
|
||
schedules = {u.schedule_id for u in found.uses if u.schedule_id}
|
||
apps = {u.app_id for u in found.uses if u.app_id} | {a["app_id"] for items in found.apps.values() for a in items}
|
||
return {"answers": len(answers), "schedules": len(schedules), "apps": len(apps),
|
||
"formal": len({u.run_id for u in found.uses if u.kind in ("formal", "schedule", "app")}),
|
||
"pinned": len({u.run_id for u in found.uses if u.kind == "pinned"}),
|
||
"exports": sum(len(u.exporters) for u in {u.run_id: u for u in found.uses}.values()),
|
||
"truncated": found.truncated}
|
||
|
||
|
||
async def version_usage(session: AsyncSession, workflow_id: str, version: int) -> dict[str, Any]:
|
||
"""一个版本的用法计数（账本接口 GET /api/ledger/workflow-versions/{wf}/{v} 加的那几项）：问数回答、导出、召回。
|
||
|
||
召回分两头：这一版作为新版本召回过几组（recall_runs：以它为新版本发起过的召回运行；changed：其中数变了的）、作为旧版本
|
||
被召回过几组（recalled_from）。"""
|
||
from app.ledger import recall
|
||
|
||
found = await find_uses(session, workflow_id, only=version)
|
||
out = summarize(found)
|
||
out["exports"] = await exports_count(session, [u.run_id for u in found.uses])
|
||
out["recall"] = await recall.version_recall_counts(session, workflow_id, version)
|
||
return out
|
||
|
||
|
||
async def workflow_name(session: AsyncSession, workflow_id: str) -> str:
|
||
return str(await session.scalar(select(Workflow.name).where(Workflow.id == workflow_id)) or "")
|
||
|
||
|
||
__all__ = ["EXPORTED", "Found", "KINDS", "KIND_TEXT", "MAX_ANSWERS", "MAX_FORMAL", "PERIODS", "Use", "exports_count",
|
||
"find_uses", "pinned_versions", "summarize", "version_usage", "workflow_name"]
|