From 56ca53ad35a28a3488a8254fcbeaa324b48563ca Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 19:41:57 +0800 Subject: [PATCH] =?UTF-8?q?=E9=97=AE=E6=95=B0=E6=8D=AE=E5=BB=BA=E5=9B=BE?= =?UTF-8?q?=E7=94=A8=E9=87=8F=E4=B8=8D=E5=86=8D=E6=98=AF=E4=B8=8B=E9=99=90?= =?UTF-8?q?=EF=BC=9B=E8=AF=84=E6=B5=8B=E9=80=9A=E7=9F=A5=E9=93=BE=E5=88=B0?= =?UTF-8?q?=E8=AF=84=E6=B5=8B=E8=AF=A6=E6=83=85=EF=BC=9B=E8=AF=84=E6=B5=8B?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E8=A1=A5=E7=95=8C=E9=9D=A2=E8=A6=81=E7=9A=84?= =?UTF-8?q?=E5=87=A0=E6=A0=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用量（第 1 波遗留）：查清楚了，丢的不是浏览器那边——问数据页一直读到 SSE 结束、usage 那一条收得到——而是服务端生成过程 （api/copilot_gen._iter_ops）读到 done、look、ask 那一行就关掉模型的流，流末尾那一段才报的用量（Anthropic 的输出 token、 OpenAI 兼容系的 stream_usage）一次都拿不到，每次建图都记成「估不出」，这一轮的花费多半是下限。改成交出那一行之后先把流读完 再关（_drain，内容丢掉）：照约定那一行就是最后一行，读完只多收用量那一段；模型在那一行之后还往下写超过 DRAIN_CHARS 字闲话的 当场关流、照实记成没报全（不为闲话付钱，也不当成报全了）。评测的作答（app/eval/answer.Driver）走同一个 answer_ops， 用量跟着报全，和问数据页同一个口径。test_usage_meter：原先断言「输出 token 为 0、记成估不出」的那条改成报全；另加 look 后、 闲话超限、评测和问数据页收到同一份用量三条；去掉修复回跑，三条都抓得住。providers/usage.py 的说明跟着改。 通知：notify_eval 的链接从定时任务的触发历史改成这次评测的详情页 /evals/runs/{id}（评测页这一波上线）；测试设上对外访问地址、 断言链接（原来的断言在没设地址时对空列表恒真）。 评测接口（只加不改，app/api/evals.py）：题集列表和详情带 latest（最近一次评测）、trend（同一变体最近几次评完的准确率，旧的 在前，换了模型、目录的不连成一条线）、schedules（目标是它的定时任务）；评测详情带 actions（停止、只重试没做完的、重新评分、 钉住 / 取消基线能不能做，和接口拒收的条件同一份）和没做完的项数；POST /suites/{id}/cases/preview 写 SQL 定期望的预览 （同期望的 SQL 探针，交回列和前 50 行，遮罩列照保存时那一道拒收、没过守卫的指明是哪一格，都不回 422）；GET /signals 「口径命中与重复度」的判据和分档、窗口趋势（数只出自 ledger/stats.compute_stats）、成员不少于 3 的问题类（只出自 chat/questions，评测运行不进）。为什么放服务端：界面层不放逻辑，统计、能不能做都由服务端给。tests/test_evals_ui_api.py。 --- backend/app/api/copilot_gen.py | 23 ++- backend/app/api/evals.py | 250 +++++++++++++++++++++++++++- backend/app/notify/service.py | 5 +- backend/app/providers/usage.py | 9 +- backend/tests/test_eval_schedule.py | 41 ++++- backend/tests/test_evals_ui_api.py | 202 ++++++++++++++++++++++ backend/tests/test_usage_meter.py | 114 +++++++++++-- 7 files changed, 618 insertions(+), 26 deletions(-) create mode 100644 backend/tests/test_evals_ui_api.py diff --git a/backend/app/api/copilot_gen.py b/backend/app/api/copilot_gen.py index 2acbeab4..bd49edee 100644 --- a/backend/app/api/copilot_gen.py +++ b/backend/app/api/copilot_gen.py @@ -41,6 +41,11 @@ _ASK_OPS = ("ask", "bp_ask") _STOP_OPS = ("done", "look", *_ASK_OPS) #: 不占次数的查看回话（need 没写、次数用完）连着来几次就不再接着写：模型在原地打转，再要下去只是花钱 _IDLE_LOOKS = 2 +#: 读到停下的那一行（done、look、ask）之后，最多再收多少字。照约定那一行就是这一次输出的最后一行，后面只剩流末尾那一段 +#: 用量（Anthropic 的输出 token、OpenAI 兼容系开了 stream_usage 的用量都在最后一段才报）：读完它，这一次的花费才报得全， +#: 问数据页这一轮的花费、评测按角色的成本才不是下限。超过这么多字说明模型还在往下写闲话：不为它付钱，当场关流，这一次的 +#: 用量记成没报全（providers/usage.py 照实记估不出，不当成报全了） +DRAIN_CHARS = 400 # ========================================================================== @@ -72,9 +77,24 @@ class CopilotIncomplete(Exception): self.detail = detail +async def _drain(stream: Any, already: int) -> None: + """停下的那一行之后把模型的流读完（内容一概不要），只为收到流末尾才报的用量；读到超过 DRAIN_CHARS 字（模型还在写闲话） + 就不读了，交回调用方关流。already 是缓冲里那一行后面已经收到的字数。""" + from app.engine.state import message_text as _text + from app.engine.state import thinking_text + + seen = already + async for chunk in stream: + seen += len(_text(chunk)) + len(thinking_text(chunk) or "") + if seen > DRAIN_CHARS: + return + + async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None = None): """流式读模型输出，按行切出操作。done、look、ask 之后即停——后面就算有闲话也不要了；look 之后停下也就是 - 「读到这一行就停止本次输出」，模型那条流随之关掉。 + 「读到这一行就停止本次输出」。停下之后不马上关流：交出那一行、调用方接着要下一条时先把流读完（_drain，内容丢掉，最多 + DRAIN_CHARS 字），流末尾那一段用量才收得到——以前读到这一行就关流，输出 token 一个都拿不到，问数据这一轮的花费多半是下限 + （第 1 波遗留）。 除了操作，还把模型的思考原样转出去（thinking_text：Claude 的 thinking 块、OpenAI 兼容系的 reasoning_content）， 不把非 JSON 行当成思考。lines 给了的话，认出来的操作行原样记进去：接着写时作为 assistant 消息交回模型。 @@ -101,6 +121,7 @@ async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None lines.append(line.strip()) yield op if op.get("op") in _STOP_OPS: + await _drain(stream, len(buffer)) return cut = _cut_off(chunk) or cut op = _parse_op_line(buffer) diff --git a/backend/app/api/evals.py b/backend/app/api/evals.py index dc5e9419..ca834aa8 100644 --- a/backend/app/api/evals.py +++ b/backend/app/api/evals.py @@ -24,6 +24,17 @@ GET /evaluations/{id}/compare?base=（和上一次或点名的那一次比：指纹差异、逐题转移和归因）；GET /pairs/{pair_id}（A/B 配对表和 精确 McNemar）；POST /suites/{id}/cleanup（按保留规则清理更早的明细）。「口径命中与重复度」的 E1–E4 直接用口径账本的 GET /api/ledger/stats（总方案 X2：只有一份实现）。 + +第 4 波评测页（细案 01 第 18 项）缺的几样，补在这里（统计、能不能做都由这里给，界面只显示）： +- GET /suites 每个题集另带 latest（最近一次评测的状态和准确率）、trend（同一变体最近几次评完的准确率，旧的在前）、schedules + （目标是这个题集的定时任务）——列表页不再为每个题集再取一遍历次评测； +- GET /evaluations/{id} 另带 actions：停止、只重试没做完的、重新评分、钉住 / 取消基线现在能不能做（和那几个接口拒收的条件 + 同一份），以及 open（没做完的项数）； +- POST /suites/{id}/cases/preview：写 SQL 定期望时的第一步——按此刻渲染、过只读守卫、跑一次和期望同一条路的 SQL 探针，交回 + 结果的列和前几行（界面上点选格或列）；用到遮罩列的照保存时那一道拒收、照实写原因（不回 422）； +- GET /signals：「口径命中与重复度」页签的判据（细案 copilot-look-and-caliber-ledger §3.8 的参考线：E1 ≥30% / <10%， + E3 ≥20%）、当前窗口和往前几个窗口各自落在哪一档、成员不少于 3 的问题类（可以加入题集或提取模板）。数都出自 + ledger/stats.compute_stats 和 chat/questions（全站各一份），这里只按窗口调它们、按判据分档。 """ from __future__ import annotations @@ -37,7 +48,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.coded import CodedHTTPException from app.api.deps import actor_of -from app.db.base import get_session, new_id, utcnow +from app.db.base import SessionLocal, get_session, new_id, utcnow from app.db.models_eval import EvalCase, EvalCatalogSnapshot, EvalRun, EvalSuite from app.eval import cases as case_rules from app.eval import gold as gold_rules @@ -223,8 +234,70 @@ async def _insert_case(session: AsyncSession, suite: EvalSuite, fields: dict[str # ========================================================================== +#: 列表里每个题集最多看最近几次评测（最近一次、趋势都从这几次里取）；趋势最多几个点 +OVERVIEW_RUNS = 30 +TREND_POINTS = 8 + + +def _rate_of(run: EvalRun) -> dict[str, Any] | None: + """一次评测总表里的准确率（stats.rate 的形状：passed、n、rate、ci）；还没有总表的为 None。""" + acc = (run.summary or {}).get("accuracy") + return dict(acc) if isinstance(acc, dict) else None + + +async def _overview(session: AsyncSession, suite_ids: list[str]) -> dict[str, dict[str, Any]]: + """每个题集：最近一次评测（不论结束没有）、同一变体最近几次评完的准确率（趋势，旧的在前）、目标是它的定时任务。 + + 趋势只连同一变体（fingerprint.variant_key，和上一次比挑基准同一个认法）的几次：换了模型、换了目录的那几次不连成一条线， + 免得把「换了配置」读成「变好了 / 变差了」。A/B 的几臂各是一次评测，同一变体的那一臂才进趋势。""" + from app.db.models import Schedule + from app.eval import runner + from app.scheduler import cron + + out: dict[str, dict[str, Any]] = {sid: {"latest": None, "trend": [], "schedules": []} for sid in suite_ids} + if not suite_ids: + return out + ranked = select(EvalRun, func.row_number().over(partition_by=EvalRun.suite_id, + order_by=EvalRun.created_at.desc()).label("n")) \ + .where(EvalRun.suite_id.in_(suite_ids)).subquery() + from sqlalchemy.orm import aliased + + run_alias = aliased(EvalRun, ranked) + rows = list((await session.execute(select(run_alias).where(ranked.c.n <= OVERVIEW_RUNS) + .order_by(run_alias.created_at.desc()))).scalars()) + await runner.reconcile_stale(session, rows) + by_suite: dict[str, list[EvalRun]] = {} + for row in rows: + by_suite.setdefault(row.suite_id, []).append(row) + for sid, runs in by_suite.items(): + latest = runs[0] + out[sid]["latest"] = {"id": latest.id, "status": latest.status, "running": runner.running(latest), + "arm": latest.arm, "pair_id": latest.pair_id, "trigger": latest.trigger, + "created_at": _iso(latest.created_at), "finished_at": _iso(latest.finished_at), + "counts": dict(latest.counts or {}), "accuracy": _rate_of(latest), + "traced_accuracy": dict((latest.summary or {}).get("traced_accuracy") or {}) or None} + finished = [r for r in runs if r.status in ("succeeded", "partial") and _rate_of(r) is not None] + if finished: + key = (finished[0].fingerprint or {}).get("variant_key") + same = [r for r in finished if (r.fingerprint or {}).get("variant_key") == key][:TREND_POINTS] + out[sid]["trend"] = [{"id": r.id, "created_at": _iso(r.created_at), "arm": r.arm, **(_rate_of(r) or {})} + for r in reversed(same)] + schedules = (await session.execute(select(Schedule).where( + Schedule.target_kind == "eval_suite", Schedule.target_id.in_(suite_ids)) + .order_by(Schedule.created_at))).scalars() + for row in schedules: + try: + text = cron.describe(row.cron, row.timezone) + except cron.CronError: + text = f"按 cron 表达式 {row.cron}（{row.timezone}）" + out[row.target_id]["schedules"].append({"id": row.id, "name": row.name, "enabled": row.enabled, + "description_text": text, "next_run_at": _iso(row.next_run_at)}) + return out + + @router.get("/suites") async def list_suites(archived: bool = False, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """题集（归档的不列，archived=true 时一起列）。每个题集另带 latest、trend、schedules（_overview）。""" query = select(EvalSuite).order_by(EvalSuite.updated_at.desc()) if not archived: query = query.where(EvalSuite.archived.is_(False)) @@ -239,7 +312,8 @@ async def list_suites(archived: bool = False, session: AsyncSession = Depends(ge counts[suite_id]["cases"] += n if enabled: counts[suite_id]["enabled"] += n - return {"suites": [_suite_out(r, counts=counts[r.id], site=site) for r in rows]} + overview = await _overview(session, [r.id for r in rows]) + return {"suites": [{**_suite_out(r, counts=counts[r.id], site=site), **overview[r.id]} for r in rows]} @router.post("/suites", status_code=201) @@ -266,6 +340,7 @@ async def get_suite(suite_id: str, session: AsyncSession = Depends(get_session)) active = [c for c in out if not c["archived"]] return {**_suite_out(suite, counts={"cases": len(active), "enabled": sum(1 for c in active if c["enabled"])}, site=await _site(session)), + **(await _overview(session, [suite_id]))[suite_id], "cases": out, "pending_upgrades": sum(1 for c in active if (c["upgrade"] or {}).get("pending"))} @@ -342,6 +417,62 @@ async def trial_unsaved(suite_id: str, request: Request, session: AsyncSession = return {"trial": (await _trial(session, suite, fields, actor_of(x_actor))).to_dict()} +#: 预览最多交回几行（界面上点选格或列用；多行期望的取值仍由试算按整份结果算） +PREVIEW_ROWS = 50 + + +@router.post("/suites/{suite_id}/cases/preview") +async def preview_sql(suite_id: str, request: Request, session: AsyncSession = Depends(get_session), + x_actor: str | None = Header(default=None)) -> dict[str, Any]: + """写 SQL 定期望的第一步：{datasource_id, sql}（SQL 可以带 {{ schedule.* }}）。按此刻渲染、过只读守卫、跑一次和期望同一条路 + 的 SQL 探针（裸图探索运行，来源 eval），交回结果的列和前 PREVIEW_ROWS 行，界面上据此点选格或列、再去试算。 + + 结果里有遮罩列、取不到结果、运行失败、超时的，照保存时那一道拒收、照实写原因（status 不是 ok，reason、code 同试算）——不回 + 422：这是在写题的途中，界面把原因落在 SQL 那一格下面。{status, reason, code, columns, rows, row_count, truncated, run_id, + rendered}。""" + from app.core import artifact_store + from app.db.models import Artifact + + suite = await _suite(session, suite_id) + body = await _body(request) + probe = {"question": "（预览）", "gold": {"kind": "sql", "datasource_id": body.get("datasource_id"), + "sql": body.get("sql")}, + # 只为跑一次探针：「没有数据」这一种期望不取任何格，取不到值时下面照样交回结果 + "expectations": [{"id": "preview", "kind": "none"}]} + try: + fields = normalize_case(probe) + await resolve_case(session, fields, tz=suite.timezone) + except CaseInvalid as e: + # 没选数据源、SQL 空着、变量写错、没过只读守卫：照实写原因、指明是哪一格 + return {"status": "invalid", "reason": e.message, "code": e.code, "field": e.field, "columns": [], "rows": [], + "row_count": None, "truncated": False, "run_id": None, "rendered": {}} + outcome = await _trial(session, suite, fields, actor_of(x_actor)) + out: dict[str, Any] = {"status": outcome.status, "reason": outcome.reason, "code": outcome.code, "columns": [], + "rows": [], "row_count": None, "truncated": False, + "run_id": outcome.run_ids[0] if outcome.run_ids else None, + "rendered": dict(outcome.rendered or {})} + if outcome.status not in ("ok", "no_value") or not outcome.run_ids: + return out + async with SessionLocal() as read: + refs = (await read.execute(select(Artifact.id).where( + Artifact.run_id == outcome.run_ids[0], Artifact.kind == "query_snapshot", + Artifact.node_id == gold_rules.QUERY_NODE).order_by(Artifact.created_at.desc()))).scalars().all() + snapshot = None + for ref in refs: + try: + snapshot = artifact_store.load(ref) + except ValueError: + snapshot = None + if isinstance(snapshot, dict): + break + if not isinstance(snapshot, dict): + return {**out, "status": "failed", "reason": "预览的运行没有留下查询结果", "code": gold_rules.GOLD_FAILED} + rows = [list(r) if isinstance(r, (list, tuple)) else [r] for r in snapshot.get("rows") or []] + return {**out, "status": "ok", "reason": None, "code": None, "columns": [str(c) for c in snapshot.get("columns") or []], + "rows": rows[:PREVIEW_ROWS], "row_count": snapshot.get("row_count", len(rows)), + "truncated": bool(snapshot.get("truncated")) or len(rows) > PREVIEW_ROWS, "shown": min(len(rows), PREVIEW_ROWS)} + + @router.patch("/suites/{suite_id}/cases/{case_id}") async def update_case(suite_id: str, case_id: str, request: Request, session: AsyncSession = Depends(get_session), x_actor: str | None = Header(default=None)) -> dict[str, Any]: @@ -859,7 +990,29 @@ async def get_evaluation(evaluation_id: str, session: AsyncSession = Depends(get .where(EvalResult.eval_run_id == row.id, EvalResult.status == "running") .order_by(EvalResult.seq))).all() return {**_evaluation_out(row), - "current": [{"case_seq": seq, "repeat": rep, "arm": arm, "doing": doing} for seq, rep, arm, doing in running]} + "current": [{"case_seq": seq, "repeat": rep, "arm": arm, "doing": doing} for seq, rep, arm, doing in running], + "actions": await _actions(session, row)} + + +async def _actions(session: AsyncSession, row: EvalRun) -> dict[str, Any]: + """这次评测现在能做哪几件事（界面照它摆按钮，不自己判）：和 stop / retry / rescore / pin 几个接口拒收的条件同一份。 + open 是这次评测（A/B 的话几臂合起来）还没做完的项数：停止、中断、出错、到了上限没开评的，重试就是把它们接着做。""" + from app.core import job_runtime as runtime + from app.db.models_eval import EvalResult + from app.eval import runner + + active = row.status in runtime.ACTIVE + leader = runner.leader_of(row) + arms = list((await session.execute(select(EvalRun.id).where( + (EvalRun.id == leader) | (EvalRun.pair_id == leader)))).scalars()) + open_items = int(await session.scalar(select(func.count()).select_from(EvalResult).where( + EvalResult.eval_run_id.in_(arms), EvalResult.status != "ok")) or 0) + suite = await session.get(EvalSuite, row.suite_id) + cleaned = row.cleaned_at is not None + finished = row.status in ("succeeded", "partial") + return {"stop": active, "retry": not active and not cleaned and open_items > 0 and suite is not None + and not suite.archived, "rescore": not active and not cleaned, + "pin": finished and not row.pinned, "unpin": bool(row.pinned), "open": open_items, "cleaned": cleaned} @router.get("/evaluations/{evaluation_id}/results") @@ -997,6 +1150,97 @@ async def get_pair(pair_id: str, session: AsyncSession = Depends(get_session)) - return {"pair_id": pair_id, "evaluations": [_evaluation_out(r) for r in rows], "table": table, "cleaned": cleaned} +# ========================================================================== +# 口径命中与重复度（E1、E3）：判据、分档、窗口趋势、成员不少于 3 的问题类 +# ========================================================================== + +#: 判据（docs/design/copilot-look-and-caliber-ledger.md §3.8，已修订为持续统计的参考线）：E1 等价命中率 ≥30% 值得按口径走、 +#: <10% 先只做对账、介于两者之间先加强规范化；E3 问题重复度 ≥20% 沉淀有价值。只作参考，不卡门 +E1_ROUTE, E1_LOW, E3_WORTH = 0.30, 0.10, 0.20 +SIGNAL_CRITERIA = { + "e1": [{"band": "route", "at_least": E1_ROUTE, "text": "等价命中率 ≥ 30%：值得按已治理的口径走"}, + {"band": "normalize", "at_least": E1_LOW, "text": "10%–30%：先加强规范化，让写法不同的等价 SQL 认得出"}, + {"band": "reconcile", "at_least": 0.0, "text": "< 10%：先只做口径对账"}], + "e3": [{"band": "worth", "at_least": E3_WORTH, "text": "问题重复度 ≥ 20%：把反复问的问题沉淀成口径有价值"}, + {"band": "not_yet", "at_least": 0.0, "text": "< 20%：反复问的问题还不多"}], + "note": "参考线出自口径账本的设计文档，只作参考，不卡门", +} +SIGNAL_DAYS_MAX = 365 +SIGNAL_POINTS_MAX = 12 +#: 每类最多列几个问题 +CLUSTER_MEMBERS = 10 + + +def _e1_band(rate: float | None) -> str | None: + if rate is None: + return None + return "route" if rate >= E1_ROUTE else "normalize" if rate >= E1_LOW else "reconcile" + + +def _e3_band(share: float | None) -> str | None: + if share is None: + return None + return "worth" if share >= E3_WORTH else "not_yet" + + +def _point(stats: dict[str, Any], since: Any, until: Any) -> dict[str, Any]: + """一个窗口的 E1、E3（取自 ledger/stats.compute_stats 的结果，不另算）和各自落在哪一档。""" + e1, e3 = stats.get("e1") or {}, stats.get("e3") or {} + rate = (e1.get("rates") or {}).get("proven") + share = e3.get("cluster_share") + return {"since": _iso(since), "until": _iso(until), + "e1": {"rate": rate, "n": int(e1.get("with_candidates") or 0), "queries": int(e1.get("queries") or 0), + "band": _e1_band(rate)}, + "e3": {"share": share, "n": int(e3.get("questions") or 0), "lower_bound": e3.get("repeated_share"), + "band": _e3_band(share)}} + + +@router.get("/signals") +async def signals(days: int = 30, points: int = 6, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """「口径命中与重复度」页签：判据、最近 days 天和往前 points - 1 个同样长的窗口各自的 E1、E3（趋势，旧的在前），以及最近 + days 天里成员不少于 3 的问题类。 + + E1、E3 的数只出自口径账本的 compute_stats（总方案 X2，全站一份；评测运行不计入，见 engine/origin.counted_runs），问题类只出自 + chat/questions（和 E3、认证队列同一份：只取问题原文、不含 history、排除系统派生的运行）。界面上的详细分布另调 + GET /api/ledger/stats。每一类给代表问题、最近几个成员和最近一个挂着运行的成员（「提取模板」就从那次运行提取）。""" + from datetime import timedelta + + from app.api.deps import current_principal + from app.chat.questions import cluster_questions, embedder, recent_questions + from app.ledger.stats import QUESTION_LIMIT, REPEAT_RUNS, compute_stats + + if not 1 <= days <= SIGNAL_DAYS_MAX: + raise CodedHTTPException(422, f"窗口需要是 1 到 {SIGNAL_DAYS_MAX} 天", case_rules.CASE_INVALID, field="days") + if not 1 <= points <= SIGNAL_POINTS_MAX: + raise CodedHTTPException(422, f"趋势需要是 1 到 {SIGNAL_POINTS_MAX} 个窗口", case_rules.CASE_INVALID, + field="points") + principal = current_principal() + now = utcnow() + span = timedelta(days=days) + trend = [] + for i in reversed(range(points)): + until = now - span * i + since = until - span + trend.append(_point(await compute_stats(session, since=since, until=until, principal=principal), since, until)) + since = now - span + questions = [q for q in await recent_questions(session, principal, limit=QUESTION_LIMIT) + if q.at is None or q.at >= since] + clusters = [] + for c in await cluster_questions(questions) if questions else []: + if c.count < REPEAT_RUNS: + continue + with_run = next((m for m in c.members if m.run_id), None) + clusters.append({"count": c.count, "representative": c.members[0].text[:500], + "first_at": _iso(c.members[-1].at), "last_at": _iso(c.members[0].at), + "members": [{"text": m.text[:500], "at": _iso(m.at), "run_id": m.run_id, "turn_id": m.turn_id} + for m in c.members[:CLUSTER_MEMBERS]], + "run_id": with_run.run_id if with_run else None}) + clusters.sort(key=lambda c: c["count"], reverse=True) + return {"window": {"since": _iso(since), "until": _iso(now), "days": days}, "criteria": SIGNAL_CRITERIA, + "current": trend[-1], "trend": trend, "clusters": clusters, "questions": len(questions), + "threshold": REPEAT_RUNS, "embedder": embedder()} + + @router.post("/suites/{suite_id}/cleanup") async def cleanup_suite(suite_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: """按保留规则清理这个题集更早的评测明细（最近 keep_runs 次、钉住的基线、A/B 的另几臂留着；总表永久保留）。""" diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index 8d0fea86..a62d1b76 100644 --- a/backend/app/notify/service.py +++ b/backend/app/notify/service.py @@ -463,7 +463,8 @@ async def notify_source_changed(source_id: str) -> int: async def notify_eval(eval_run_id: str, event: str, *, passed: int, graded: int, regressed: int) -> int: """定时评测的通知（eval.regressed 有一道题从通过变成不通过、eval.failed 没评完：失败、到了上限、止损）。只推订阅了它的 全局渠道。消息里只有题集名、状态、通过几题、退步几题、评测编号和链接（细案 01 做法 15、总方案 X6）：不放期望值、答案里的数、 - 题目原文。链接去发起它的那条定时任务的触发历史（评测页在第 4 波）。""" + 题目原文。链接去这次评测的详情页（/evals/runs/{评测 id}：进度、逐题结果、和上一次比；第 4 波评测页上线之前是定时任务的 + 触发历史），退步的那几题在详情页上点得开。""" from app.db.models_eval import EvalRun, EvalSuite async with SessionLocal() as session: @@ -485,7 +486,7 @@ async def notify_eval(eval_run_id: str, event: str, *, passed: int, graded: int, if run.arm: lines.append(("对照", run.arm[:40])) lines.append(("评测", f"#{eval_run_id[:6]}")) - links = [("查看定时任务", f"{base}/schedules/{run.schedule_id}?tab=history")] if base and run.schedule_id else [] + links = [("查看评测", f"{base}/evals/runs/{eval_run_id}")] if base else [] message = {"event": event, "status": run.status, "ok": False, "title": f"定时评测「{name}」{'有退步' if regressed_event else '没有评完'}", "lines": lines, "links": links, "subject": {"eval_run_id": eval_run_id, "suite_id": run.suite_id, "schedule_id": run.schedule_id}} diff --git a/backend/app/providers/usage.py b/backend/app/providers/usage.py index 37563e6b..9404039a 100644 --- a/backend/app/providers/usage.py +++ b/backend/app/providers/usage.py @@ -13,9 +13,10 @@ async generator：with 块跨着 yield 的话，变量在两次 yield 之间留 **金额。** 按 providers/catalog.estimate_cost 算，和运行的 llm.end 一个口径。估不出的调用记 unpriced_calls：模型不在价格 目录里，或者用量没报全而目录价又不是 0。估不出的不当成 0 元，调用方据此写「≥」或「金额无法估算」。用量没报全有两种： -回复没带用量（有的网关流式不给）；读到一半就不读了——生成过程读到 done、look、ask 那一行就关掉模型的流（后面的闲话不要、 -也不再为它付钱），用量在流的末尾才报全（Anthropic 的输出 token 在最后一条，OpenAI 兼容系开了 stream_usage 也在最后）， -关掉的那一次只拿得到已经报了的部分。这一次照样计次数、计已报的 token，记成估不出，不悄悄少算。 +回复没带用量（有的网关流式不给）；读到一半就不读了——用量在流的末尾才报全（Anthropic 的输出 token 在最后一条，OpenAI +兼容系开了 stream_usage 也在最后）。生成过程读到 done、look、ask 那一行之后照样把流读完再关（api/copilot_gen._drain）， +流末尾那一段用量收得到；只有模型在那一行之后还往下写超过 DRAIN_CHARS 字的闲话时才当场关流（不为闲话付钱），连接断了也一样， +这一次只拿得到已经报了的部分：照样计次数、计已报的 token，记成估不出，不悄悄少算。 模型名按发起调用时的参数取（invocation_params 的 model / model_name，就是接入里配的模型 id，价格目录按它查），取不到再 看回复里的 model_name。 @@ -85,7 +86,7 @@ class Meter(BaseCallbackHandler): row["unreported"] += 1 def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None: - """流读到一半停了（生成过程读到 done / look / ask 就关流，或者连接断了）：LangChain 照样交来已经收到的那几段。 + """流读到一半停了（生成过程在最后一行之后读到太多闲话就关流，或者连接断了）：LangChain 照样交来已经收到的那几段。 有收到东西的照计一次、记已报的 token，算用量没报全；一段都没收到的（请求刚发出去就被拒了）不计。""" from langchain_core.outputs import ChatGenerationChunk diff --git a/backend/tests/test_eval_schedule.py b/backend/tests/test_eval_schedule.py index 441d4644..1bf3c7b3 100644 --- a/backend/tests/test_eval_schedule.py +++ b/backend/tests/test_eval_schedule.py @@ -73,6 +73,34 @@ def received(monkeypatch): return got +async def _public_url(url: str): + """临时设对外访问地址（通知里的链接按它拼），交回复原的函数：别的测试不受影响。""" + from app.core import public_url + from app.db.models import Setting + + async with SessionLocal() as session: + row = await session.get(Setting, "notify") + before = dict(row.value or {}) if row is not None else None + if row is None: + session.add(Setting(key="notify", value={"public_base_url": url})) + else: + row.value = {**(row.value or {}), "public_base_url": url} + await session.commit() + await public_url.refresh(session) + + async def restore() -> None: + async with SessionLocal() as session: + row = await session.get(Setting, "notify") + if before is None and row is not None: + await session.delete(row) + elif row is not None: + row.value = before + await session.commit() + await public_url.refresh(session) + + return restore + + def stub(monkeypatch, *, delay: float = 0.0, path: str = "asked"): async def drive(self): await asyncio.sleep(delay) @@ -212,8 +240,12 @@ async def test_an_unfinished_scheduled_evaluation_pushes_failed(client, monkeypa suite_id = await es.make_suite("none", cases=ASK * 4, name="虚构失败题集", budget={"max_parallel": 1}) body = await _schedule(client, suite_id) await _channel(["eval.failed"]) - fire = await service.fire_once(body["id"], utcnow().replace(microsecond=0), "manual") - await es.wait_eval(fire.eval_run_id) + restore = await _public_url("https://agentlab.example.com") + try: + fire = await service.fire_once(body["id"], utcnow().replace(microsecond=0), "manual") + await es.wait_eval(fire.eval_run_id) + finally: + await restore() async with SessionLocal() as session: run = await session.get(EvalRun, fire.eval_run_id) skipped = (await session.execute(select(EvalResult).where( @@ -223,8 +255,9 @@ async def test_an_unfinished_scheduled_evaluation_pushes_failed(client, monkeypa message = json.loads(request.content) assert message["event"] == "eval.failed" and message["title"] == "定时评测「虚构失败题集」没有评完" assert "通过：0 / 3 题" in message["text"] - # 链接去那条定时任务的触发历史（设了对外访问地址才有链接） - assert all("/schedules/" in link["url"] for link in message["links"]) + # 链接去这次评测的详情页（第 4 波评测页上线之前去定时任务的触发历史）；设了对外访问地址才有链接 + assert [link["url"] for link in message["links"]] == [f"https://agentlab.example.com/evals/runs/{fire.eval_run_id}"], \ + message["links"] def test_the_eval_events_say_they_are_for_scheduled_evaluations(): diff --git a/backend/tests/test_evals_ui_api.py b/backend/tests/test_evals_ui_api.py new file mode 100644 index 00000000..3a32da41 --- /dev/null +++ b/backend/tests/test_evals_ui_api.py @@ -0,0 +1,202 @@ +"""评测页（第 4 波，细案 01 第 18 项）在 app/api/evals.py 里补的几样：统计、能不能做都由服务端给，界面只显示。 + +- 题集列表每个题集带最近一次评测、同一变体最近几次评完的准确率（旧的在前，换了变体的不连成一条线）、目标是它的定时任务； +- 评测详情带 actions：只重试没做完的、重新评分、钉住 / 取消基线现在能不能做，和接口拒收的条件同一份； +- 写 SQL 定期望的预览：交回结果的列和前几行；用到遮罩列的照保存时那一道拒收，没过只读守卫的指明是哪一格（都不回 422）； +- 「口径命中与重复度」：判据、窗口趋势（出自 ledger/stats.compute_stats）、成员不少于 3 的问题类（出自 chat/questions）， + 评测运行不进问题类。 + +库、表、题集名、问法都是虚构的。 +""" +from __future__ import annotations + +import uuid +from datetime import timedelta + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete + +from app.db.base import SessionLocal, utcnow +from app.db.models import Conversation, ConversationTurn, Run, Schedule +from app.db.models_eval import EvalResult, EvalRun +from app.engine.runner import run_manager +from app.main import app +from tests import eval_samples as es +from tests.fixtures import dual_check_shop as shop +from tests.fixtures.sources import drop_source + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +async def source(tmp_path): + name, sid = await shop.add_source(shop.build(tmp_path / "shop.db")) + yield {"name": name, "id": sid} + await drop_source(sid) + + +def _summary(passed: int, n: int) -> dict: + return {"accuracy": {"passed": passed, "n": n, "rate": round(passed / n, 4) if n else None, "ci": None}, + "traced_accuracy": {"passed": passed, "n": n, "rate": round(passed / n, 4) if n else None, "ci": None}} + + +async def _evaluation(suite_id: str, *, status: str, key: str, passed: int, n: int, minutes_ago: int, + arm: str | None = None, pinned: bool = False, cleaned: bool = False) -> str: + async with SessionLocal() as session: + row = EvalRun(suite_id=suite_id, status=status, trigger="manual", variant={}, resolved_models={}, + fingerprint={"variant_key": key}, summary=_summary(passed, n) if status != "queued" else {}, + counts={"total": n, "done": n}, arm=arm, pinned=pinned, + cleaned_at=utcnow() if cleaned else None) + session.add(row) + await session.flush() + row.created_at = utcnow() - timedelta(minutes=minutes_ago) + await session.commit() + return row.id + + +async def test_the_suite_list_carries_the_latest_run_a_trend_of_the_same_variant_and_its_schedules(client, source): + suite_id = await es.make_suite(source["id"], name=f"虚构题集_{uuid.uuid4().hex[:6]}") + old = await _evaluation(suite_id, status="succeeded", key="A", passed=1, n=4, minutes_ago=50) + other = await _evaluation(suite_id, status="succeeded", key="B", passed=4, n=4, minutes_ago=40) + failed = await _evaluation(suite_id, status="failed", key="A", passed=0, n=4, minutes_ago=30) + mid = await _evaluation(suite_id, status="partial", key="A", passed=2, n=4, minutes_ago=20) + new = await _evaluation(suite_id, status="succeeded", key="A", passed=3, n=4, minutes_ago=10) + async with SessionLocal() as session: + sched = Schedule(name="虚构每周回归", workflow_id="", workflow_name="虚构题集", target_kind="eval_suite", + target_id=suite_id, cron="0 9 * * 1", timezone="Asia/Shanghai", enabled=True) + session.add(sched) + await session.commit() + sched_id = sched.id + try: + listed = {s["id"]: s for s in (await client.get("/api/evals/suites")).json()["suites"]} + got = listed[suite_id] + assert got["latest"]["id"] == new and got["latest"]["accuracy"]["rate"] == 0.75 + # 趋势只连同一变体、评完了的：B 那一次、失败的那一次都不进；旧的在前 + assert [p["id"] for p in got["trend"]] == [old, mid, new], got["trend"] + assert [p["rate"] for p in got["trend"]] == [0.25, 0.5, 0.75] + assert other not in [p["id"] for p in got["trend"]] and failed not in [p["id"] for p in got["trend"]] + [schedule] = got["schedules"] + assert schedule["id"] == sched_id and schedule["enabled"] is True and "09:00" in schedule["description_text"] + detail = (await client.get(f"/api/evals/suites/{suite_id}")).json() + assert detail["latest"]["id"] == new and [s["id"] for s in detail["schedules"]] == [sched_id] + finally: + async with SessionLocal() as session: + await session.execute(delete(Schedule).where(Schedule.id == sched_id)) + await session.commit() + await es.drop_suites() + + +async def test_the_evaluation_says_what_can_be_done_now(client, source): + suite_id = await es.make_suite(source["id"], name=f"虚构题集_{uuid.uuid4().hex[:6]}") + try: + run_id = await _evaluation(suite_id, status="partial", key="A", passed=1, n=2, minutes_ago=5) + async with SessionLocal() as session: + session.add(EvalResult(eval_run_id=run_id, seq=1, case_id=None, case_seq=1, case_hash="h", repeat=0, + status="ok", verdict="pass")) + session.add(EvalResult(eval_run_id=run_id, seq=2, case_id=None, case_seq=2, case_hash="h", repeat=0, + status="cancelled", error="评测已停止")) + await session.commit() + actions = (await client.get(f"/api/evals/evaluations/{run_id}")).json()["actions"] + assert actions == {"stop": False, "retry": True, "rescore": True, "pin": True, "unpin": False, "open": 1, + "cleaned": False} + pinned = await client.post(f"/api/evals/evaluations/{run_id}/pin", json={"pinned": True}) + assert pinned.status_code == 200 + actions = (await client.get(f"/api/evals/evaluations/{run_id}")).json()["actions"] + assert actions["pin"] is False and actions["unpin"] is True + + done = await _evaluation(suite_id, status="succeeded", key="A", passed=2, n=2, minutes_ago=4, cleaned=True) + actions = (await client.get(f"/api/evals/evaluations/{done}")).json()["actions"] + # 明细清理了的：不能重试、不能重评（和接口拒收的条件一致） + assert actions["retry"] is False and actions["rescore"] is False and actions["cleaned"] is True + assert (await client.post(f"/api/evals/evaluations/{done}/rescore")).status_code == 409 + finally: + await es.drop_suites() + + +async def test_previewing_an_expected_sql_gives_columns_and_rows(client, source): + made = await client.post("/api/evals/suites", json={"name": f"虚构题集_{uuid.uuid4().hex[:6]}"}) + suite_id = made.json()["id"] + try: + body = {"datasource_id": source["id"], + "sql": "SELECT status AS 状态, SUM(amount) AS gmv FROM orders GROUP BY status ORDER BY status"} + got = (await client.post(f"/api/evals/suites/{suite_id}/cases/preview", json=body)).json() + assert got["status"] == "ok" and got["columns"] == ["状态", "gmv"], got + assert got["rows"] and all(len(r) == 2 for r in got["rows"]) and got["run_id"] + assert got["rendered"]["sql"].startswith("SELECT status") + # 没过只读守卫：指明是 SQL 那一格，不回 422 + res = await client.post(f"/api/evals/suites/{suite_id}/cases/preview", + json={"datasource_id": source["id"], "sql": "DELETE FROM orders"}) + assert res.status_code == 200 and res.json()["status"] == "invalid" and res.json()["field"] == "gold.sql" + res = await client.post(f"/api/evals/suites/{suite_id}/cases/preview", json={"sql": "SELECT 1"}) + assert res.json()["status"] == "invalid" and res.json()["field"] == "gold.datasource_id" + finally: + await es.drop_suites() + + +async def test_a_preview_touching_a_masked_column_is_refused(client, tmp_path): + name, sid = await shop.add_source(shop.build(tmp_path / "masked.db"), options={"mask_columns": ["region"]}) + made = await client.post("/api/evals/suites", json={"name": f"虚构题集_{uuid.uuid4().hex[:6]}"}) + try: + got = (await client.post(f"/api/evals/suites/{made.json()['id']}/cases/preview", json={ + "datasource_id": sid, "sql": "SELECT region AS 地区, SUM(amount) AS gmv FROM orders GROUP BY region"})).json() + assert got["status"] == "refused" and got["code"] == "eval_gold_masked" and "遮罩" in got["reason"], got + assert got["rows"] == [] and got["columns"] == [], "遮罩列的值一格都不能交出去" + finally: + await es.drop_suites() + await drop_source(sid) + + +async def _asked(question: str, n: int, *, origin: str = "chat") -> list[str]: + async with SessionLocal() as session: + conv = Conversation(kind="chat", title="虚构对话") + session.add(conv) + await session.flush() + ids = [] + for i in range(n): + run = Run(workflow_name="问数据", status="succeeded", graph={}, input={"question": question}, + origin_kind=origin, run_class="exploratory") + session.add(run) + await session.flush() + session.add(ConversationTurn(conversation_id=conv.id, seq=i, question=question, run_id=run.id, + status="done")) + ids.append(run.id) + await session.commit() + return ids + + +async def test_signals_give_criteria_trend_and_repeated_question_clusters(client): + question = f"虚构门店 {uuid.uuid4().hex[:6]} 上周的销售额是多少？" + runs = await _asked(question, 3) + eval_question = f"虚构评测题 {uuid.uuid4().hex[:6]} 本月退款率？" + await _asked(eval_question, 3, origin="eval") + got = (await client.get("/api/evals/signals", params={"days": 30, "points": 3})).json() + assert [c["band"] for c in got["criteria"]["e1"]] == ["route", "normalize", "reconcile"] + assert got["criteria"]["e3"][0]["at_least"] == 0.2 + assert len(got["trend"]) == 3 and got["current"] == got["trend"][-1] + assert got["trend"][0]["until"] <= got["trend"][1]["since"] or got["trend"][0]["until"] == got["trend"][1]["since"] + assert got["current"]["e3"]["n"] >= 3 and got["current"]["e3"]["band"] in ("worth", "not_yet") + mine = [c for c in got["clusters"] if c["representative"] == question] + assert mine and mine[0]["count"] == 3 and mine[0]["run_id"] in runs, got["clusters"] + assert not [c for c in got["clusters"] if c["representative"] == eval_question], "评测运行不进问题类" + assert got["threshold"] == 3 and got["embedder"] + assert (await client.get("/api/evals/signals", params={"days": 0})).status_code == 422 + + +def test_bands_follow_the_reference_lines(): + from app.api import evals + + assert [evals._e1_band(x) for x in (None, 0.05, 0.1, 0.29, 0.3)] == [None, "reconcile", "normalize", "normalize", + "route"] + assert [evals._e3_band(x) for x in (None, 0.19, 0.2)] == [None, "not_yet", "worth"] diff --git a/backend/tests/test_usage_meter.py b/backend/tests/test_usage_meter.py index c69259f6..ed0c35b5 100644 --- a/backend/tests/test_usage_meter.py +++ b/backend/tests/test_usage_meter.py @@ -5,7 +5,10 @@ 不当成 0 元； - metered 只计生成过程自己的调用（含它在一步里起的任务）；调用方在两次 yield 之间做的事（发起运行）不计——运行的用量 另有 llm.end，包进来就重复了；嵌套时内层并进外层； -- answer_ops 末尾的 usage 记下这一轮建图的每一次调用（起草、自查修正各一次）；复核的结果带 usage。 +- answer_ops 末尾的 usage 记下这一轮建图的每一次调用（起草、自查修正各一次）；复核的结果带 usage； +- 读到 done / look / ask 那一行之后照样把模型的流读完再关（读完只为拿到流末尾才报的用量）：输出 token 报全了、金额估得出， + 不再是下限；模型在那一行之后还往下写一大段闲话的，读到上限就当场关流、照实记成没报全； +- 评测的作答（app/eval/answer.Driver）记的助手用量就是问数据页收到的那条 usage：同一个口径。 模型都是剧本，不调真接口；价格用目录里登记过的模型 id，名字之外都是虚构的。 """ @@ -190,9 +193,19 @@ async def client(): yield c +async def _sse(client, body: dict) -> list[dict]: + events = [] + async with client.stream("POST", "/api/copilot/generate-stream", json=body) as r: + async for line in r.aiter_lines(): + if line.startswith("data: "): + events.append(json.loads(line[6:])) + return events + + async def test_answer_ops_ends_with_the_usage_of_every_build_call(client, monkeypatch): - """起草、自查修正各一次。生成过程读到 done 就关流（后面的闲话不要），流末尾才报的输出 token 拿不到：这两次照样计次数、 - 计已报的输入 token，记成估不出（unpriced_calls），金额只是估得出的那部分——不悄悄少算。""" + """起草、自查修正各一次。读到 done 那一行之后照样把模型的流读完（模型照约定 done 就是最后一行，读完只多收流末尾那一段 + 用量）：输出 token 报全了、金额估得出，这一轮的花费不再是下限（第 1 波的遗留：以前读到 done 就关流，输出 token 一个都 + 拿不到，两次都记成估不出）。""" import app.api.copilot as copilot model = Scripted(model=PRICED, replies=[GRAPH, FIX], tokens=(3000, 500)) @@ -201,19 +214,96 @@ async def test_answer_ops_ends_with_the_usage_of_every_build_call(client, monkey return model, PRICED monkeypatch.setattr(copilot, "get_chat_model", _model) - events = [] - async with client.stream("POST", "/api/copilot/generate-stream", json={"instruction": "数一数", "intent": "answer"}) as r: - async for line in r.aiter_lines(): - if line.startswith("data: "): - events.append(json.loads(line[6:])) + events = await _sse(client, {"instruction": "数一数", "intent": "answer"}) assert [e["op"] for e in events if e["op"] == "check"] and events[-2]["op"] == "final" spent = events[-1] assert spent["op"] == "usage", spent assert model.calls == 2, spent # 起草一次、自查修正一次 - assert spent["by_model"][PRICED]["calls"] == 2, spent - assert spent["by_model"][PRICED]["input_tokens"] == 6000 and spent["by_model"][PRICED]["output_tokens"] == 0 - assert spent["cost_usd"] == pytest.approx(estimate_cost(PRICED, 6000, 0), abs=1e-6) - assert spent["unpriced_calls"] == 2, "关流前没报全的两次记成估不出" + row = spent["by_model"][PRICED] + assert row["calls"] == 2, spent + assert (row["input_tokens"], row["output_tokens"]) == (6000, 1000), "流末尾才报的输出 token 也收到了" + assert spent["unpriced_calls"] == 0, "读完了的流用量报全了，不记估不出" + assert spent["cost_usd"] == pytest.approx(estimate_cost(PRICED, 6000, 1000), abs=1e-6) + + +async def test_a_look_line_also_waits_for_the_usage_at_the_end_of_the_stream(client, monkeypatch): + """读到 look 停下查看、ask 等用户回答：同样把这一次的流读完再关，查看前那一次的输出 token 不丢。""" + import app.api.copilot as copilot + + looked = _ops({"op": "look", "kind": "table", "id": "不存在的表", "need": "看看有哪些列"}) + model = Scripted(model=PRICED, replies=[looked, _ops({"op": "ask", "question": "要按哪一天算？"})], tokens=(1200, 80)) + + async def _model(*_a, **_k): + return model, PRICED + + monkeypatch.setattr(copilot, "get_chat_model", _model) + events = await _sse(client, {"instruction": "看一下再问", "intent": "answer"}) + spent = events[-1] + assert spent["op"] == "usage" and model.calls == 2, [e["op"] for e in events] + row = spent["by_model"][PRICED] + assert (row["input_tokens"], row["output_tokens"], row["unpriced_calls"]) == (2400, 160, 0), spent + + +class Chatty(Scripted): + """done 那一行之后还往下写一大段闲话（不守约定），用量照样在最后一段才报。""" + + chatter: int = 40 + + async def _astream(self, messages, stop=None, run_manager=None, **kwargs): + lines = self._next() + inp, out = self.tokens + for i, line in enumerate(lines): + meta = {"input_tokens": inp, "output_tokens": 0, "total_tokens": inp} if i == 0 else None + yield ChatGenerationChunk(message=AIMessageChunk(content=line + "\n", **({"usage_metadata": meta} if meta + else {}))) + for _ in range(self.chatter): + yield ChatGenerationChunk(message=AIMessageChunk(content="这是一段写在最后一行之后、没人要的闲话。\n")) + yield ChatGenerationChunk(message=AIMessageChunk(content="", usage_metadata={ + "input_tokens": 0, "output_tokens": out, "total_tokens": out})) + + +async def test_chatter_after_the_last_line_is_cut_off_and_recorded_as_not_fully_reported(client, monkeypatch): + """读完流只为拿用量，不为闲话付钱：最后一行之后又收到超过上限的字就当场关流。这一次的用量没报全，照实记成估不出，不当成 + 报全了，也不悄悄当成 0 元。""" + import app.api.copilot as copilot + from app.api import copilot_gen + + model = Chatty(model=PRICED, replies=[_ops({"op": "ask", "question": "要按哪一天算？"})], tokens=(500, 300)) + + async def _model(*_a, **_k): + return model, PRICED + + monkeypatch.setattr(copilot, "get_chat_model", _model) + assert model.chatter * len("这是一段写在最后一行之后、没人要的闲话。\n") > copilot_gen.DRAIN_CHARS + events = await _sse(client, {"instruction": "按天看销售额", "intent": "answer"}) + assert events[-1]["op"] == "usage" and model.calls == 1, [e["op"] for e in events] + row = events[-1]["by_model"][PRICED] + assert (row["calls"], row["input_tokens"], row["output_tokens"], row["unpriced_calls"]) == (1, 500, 0, 1), row + + +async def test_the_eval_driver_records_the_same_usage_the_chat_page_receives(client, monkeypatch): + """评测的作答（app/eval/answer.Driver._generate → _take）收下的助手用量，就是问数据页经 SSE 收到的那条 usage：同一个 + 生成过程（answer_ops）、同一份计量，读完流之后两边都是报全了的数。""" + import app.api.copilot as copilot + from app.eval import answer + + replies = [_ops({"op": "ask", "question": "要按哪一天算？"})] + model = Scripted(model=PRICED, replies=replies, tokens=(700, 90)) + + async def _model(*_a, **_k): + return model, PRICED + + monkeypatch.setattr(copilot, "get_chat_model", _model) + chat = (await _sse(client, {"instruction": "按天看销售额", "intent": "answer"}))[-1] + assert chat["op"] == "usage" + driver = answer.Driver("按天看销售额", datasource_ids=[], ctx=answer.TurnContext( + suite_id="s", suite_name="虚构题集", eval_run_id="e", case_id=None, case_seq=1, actor=None)) + ops, refused = await driver._generate("按天看销售额", None) + assert refused is None and any(o.get("op") == "ask" for o in ops) + driver._take(ops) + got = driver.turn.usage["assistant"] + assert got["by_model"] == chat["by_model"] and got["cost_usd"] == chat["cost_usd"], (got, chat) + assert got["by_model"][PRICED]["output_tokens"] == 90 and got["unpriced_calls"] == 0 async def test_a_build_on_an_unpriced_model_says_it_cannot_be_priced(client, monkeypatch):