"""评测里的一轮作答：无头地走一遍问数据页走的那一轮（细案 01 做法 1、9；总方案 2.1）。 **量的就是用户拿到的那一轮。** 不另写一份编排：一轮的流程是 app/chat/drive.TurnDriver（问数据的一轮在服务端跑也是它， app/chat/job.py），生成过程是 api/copilot.answer_ops，几个「决定」是 chat/turn 的纯函数，复核是 api/copilot.review_of_run。这里 只写评测和问数据不一样的那几项（Driver 是 TurnDriver 的子类）： answer_ops → final.next（chat/turn.next_step）→ launch_run(origin=eval) → 等终态 → review_of_run → 要的话带原图重来一次 - 生成：问法就是 instruction，intent=answer，限定的数据源同题目；助手模型按变体覆盖（GenerateIn.provider / model，挑表、 复核跟着换）；读哪份目录按变体（answer_ops 的 catalog_view，查看的表详情、自查的 SQL 检查器都按它）；分析工具开不开按 变体（answer_ops 的 analysis_tools，经 TurnDriver 透传：关的那一臂工具清单不列分析工具，节点手册不讲分析卡）。 - 结局：reply 直接回答、asked 反问、fail 判失败（建图失败）、ready 只建不跑、retry（老服务端那条路：没自查过的图带着原因 重来一次）、run 运行。生成过程交出 error（助手本轮未完成）、开流前就拒绝（模型没配好、数据源不在）都是建图失败，另记 「助手调用失败」——连续几题都这样就止损（runner.py）；只有生成过程一个操作都没给（只写了思考，error.next=retry）时按 同一个规矩先带着原因重来一次，还不行才记建图失败。每次重来的原因照实记在 retries。 - 运行：fill_input 填入口字段（单轮，history 为空），来源 eval，名字「评测：题集名·第 n 题」，选项带目录视图（Run.options 的 catalog_view：运行时的 db_schema、冻结快照、自查都按它）、只读和遮罩（同使用者的问数据运行：评测页会显示答案里的数， 遮罩值不能从这里露出来；评测不写业务库），有分析期望的题集另带这次评测的分析设置快照（判定用的阈值和算期望同一份）。 写作模型按变体写进图的 defaults：节点自己点了模型的不覆盖，照实记在 own_models。 - 终态：停在审批上（危险工具、人工节点）的直接取消、记「评测不代批」；被驳回的（cancelled + end_reason=rejected，总方案 X9）记路径「被驳回」，不算运行失败；失败、被停止记运行失败。 - 复核：总是等完（问数据页超过 25 秒不等，评测里照等，复核用时另记，超过 25 秒的单列，见 stats.py）。复核给了 retry_instruction 就带着这次运行的图重来一次（chat/turn 的规则，最多试 MAX_ATTEMPTS 次）。 用量按角色记（做法 3、11）：助手 = 生成（answer_ops 末尾的 usage）+ 复核（review_of_run 的 usage）；写作 = 答案运行里的 llm.end（按模型分）。重来的话两次都算——那都是这一题花的钱。 时限、停止由调用方管（runner.py 用 within 包住 drive，排队等名额的那一段不算）：这里随时把正在跑的那次运行记在 current_run，调用方 超时、停止时据此取消它（stop_run）。 """ from __future__ import annotations import copy import logging from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field from typing import Any from sqlalchemy import select # 一轮的流程和结果抽到了 chat/drive.py（问数据、评测共用），这里留着老名字：runner、stats 和测试照旧从这里取 from app.chat.drive import CANCEL_GRACE_S, Turn, TurnDriver, add_usage # noqa: F401 logger = logging.getLogger(__name__) #: 问数据页最多等复核这么久（chat/job.REVIEW_WAIT_S，和以前页面上的 REVIEW_TIMEOUT_MS 同一个数）：评测照等，超过的单列 REVIEW_PAGE_WAIT_S = 25.0 #: 运行停在审批上：评测不代批 APPROVAL_REASON = "运行停在审批上（危险工具、人工节点）：评测不代批，已取消这次运行" #: 每一步在做什么（EvalResult.current） PHASE_TEXT = {"gold": "正在算期望", "build": "正在建图", "run": "运行中", "review": "正在复核", "score": "正在评分", "gold_after": "作答后复算期望", "judge": "正在请裁判判断"} @dataclass class TurnContext: """这一轮作答的来历（记进运行 origin）和要用的东西。""" suite_id: str suite_name: str eval_run_id: str case_id: str | None case_seq: int | None actor: str | None arm: str | None = None repeat: int = 0 #: 变体：助手模型 {provider, model}、写作模型 {provider, model} assistant: Mapping[str, Any] = field(default_factory=dict) writer: Mapping[str, Any] = field(default_factory=dict) #: 目录视图（Run.options['catalog_view'] 的形状） view: Mapping[str, Any] | None = None #: 这次评测的分析设置快照（有分析期望的题集才有）：作答的运行按它判（Run.options['analysis']），和算期望同一份 analysis: Mapping[str, Any] | None = None #: 变体：分析工具开不开（variants.analysis_tools）。关的那一臂建图时助手看不到分析工具（TurnDriver 的 analysis_tools） analysis_tools: bool = True #: 变体：本体穿透开不开（variants.ontology_tools）。关的那一臂建图时助手看不到本体工具与邻居提示 ontology_tools: bool = True async def writer_usage(run_ids: list[str]) -> dict[str, Any]: """答案运行里的模型调用（llm.end，按模型分）：写作这一角色的用量，形状同 providers/usage 的汇总。封存之后追加的 （按需裁判、质询）不算：那是事后另记的账（裁判记在评测的账上另算，见 judging.py）。""" from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import RunEvent from app.providers.catalog import unpriced_call out = add_usage(None, None) if not run_ids: return out async with SessionLocal() as session: rows = (await session.execute(select(RunEvent.data).where( RunEvent.run_id.in_(run_ids), RunEvent.type == EventType.LLM_END))).scalars().all() for data in rows: data = data if isinstance(data, dict) else {} if data.get("post_seal"): continue model = str(data.get("model") or "unknown") piece = {"input_tokens": int(data.get("input_tokens") or 0), "output_tokens": int(data.get("output_tokens") or 0), "calls": int(data.get("calls") or 1), "cost_usd": float(data.get("cost_usd") or 0.0), "unpriced_calls": 1 if unpriced_call(data) else 0} out = add_usage(out, {**piece, "by_model": {model: piece}}) return out def with_writer(graph: dict[str, Any], writer: Mapping[str, Any]) -> tuple[dict[str, Any], list[str]]: """写作模型按变体写进图的 defaults（只写给了的那几项）。返回 (图, 自己点了模型、不会被覆盖的节点 id)。""" provider, model = writer.get("provider"), writer.get("model") if not (provider or model): return graph, [] out = copy.deepcopy(graph) defaults = dict(out.get("defaults") or {}) if provider: defaults["provider"] = provider if model: defaults["model"] = model out["defaults"] = defaults own = [] for node in out.get("nodes") or []: config = ((node or {}).get("data") or {}).get("config") or {} if isinstance(config, Mapping) and (config.get("provider") or config.get("model")): own.append(str(node.get("id"))) return out, own def _options(view: Mapping[str, Any] | None, analysis: Mapping[str, Any] | None = None) -> dict[str, Any]: """作答运行带的选项：目录视图、只读、遮罩（见模块说明）；有分析期望的题集另带这次评测的分析设置快照（发起时 engine/launch 见调用方给了就照它的，不再现读设置）：判「主要来自」「超出常规波动」的阈值和算期望的是同一份。""" from app.analysis.contract import SETTINGS_KEY from app.data.catalog_view import OPTION_KEY from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION out: dict[str, Any] = {MASK_RESULTS_OPTION: True, READ_ONLY_OPTION: True} if view: out[OPTION_KEY] = dict(view) if analysis is not None: out[SETTINGS_KEY] = dict(analysis) return out def run_name(suite_name: str, case_seq: int | None) -> str: return f"评测：{suite_name or '题集'}·第 {case_seq or '?'} 题" async def stop_run(run_id: str | None, actor: str | None = None) -> None: """取消一次还没结束的运行、等它收尾（超时、停止评测、停在审批上时）。已经结束的什么都不做。""" from app.engine.runner import run_manager if not run_id: return try: await run_manager.cancel(run_id, actor=actor) await run_manager.wait_idle(run_id, timeout=CANCEL_GRACE_S) except Exception: # noqa: BLE001 - 取消是收尾，出错只记日志 logger.exception("评测：取消运行 %s 时出错", run_id) class Driver(TurnDriver): """一轮作答（流程在 TurnDriver）。drive() 跑完一轮，结果在 turn 上；on_phase(阶段) 在每一步开始时调（runner 据此写 「正在做什么」）。和问数据不一样的：助手模型、目录视图、分析工具开不开按变体；写作模型按变体写进图；来源 eval、只读和 遮罩；停在审批上直接取消（评测不代批）；复核一直等。""" def __init__(self, question: str, *, datasource_ids: list[str], ctx: TurnContext, on_phase: Callable[[str], Awaitable[None]] | None = None) -> None: super().__init__(question, datasource_ids=datasource_ids, on_phase=on_phase, provider=ctx.assistant.get("provider"), model=ctx.assistant.get("model"), analysis_tools=ctx.analysis_tools, ontology_tools=ctx.ontology_tools) self.ctx = ctx def catalog_view(self) -> Any: return self.ctx.view def prepare_graph(self, graph: dict[str, Any]) -> dict[str, Any]: graph, own = with_writer(graph, self.ctx.writer) self.turn.own_models = sorted(set(self.turn.own_models) | set(own)) return graph async def launch_args(self, session: Any, graph: dict[str, Any]) -> dict[str, Any]: origin = {k: v for k, v in { "suite_id": self.ctx.suite_id, "eval_run_id": self.ctx.eval_run_id, "case_id": self.ctx.case_id, "role": "answer", "arm": self.ctx.arm, "repeat": self.ctx.repeat, "attempt": self.turn.attempts}.items() if v is not None} return {"actor": self.ctx.actor, "options": _options(self.ctx.view, self.ctx.analysis), "origin_kind": "eval", "origin": origin, "workflow_name": run_name(self.ctx.suite_name, self.ctx.case_seq)} async def on_interrupted(self, run_id: str, row: Any) -> Any | None: t = self.turn waiting = await self._pending_approvals(run_id) await stop_run(run_id, self.ctx.actor) t.path = "run_failed" t.reason = APPROVAL_REASON if waiting else "运行停在断点上（服务可能重启过）：评测不继续这次运行，已取消" return None async def drive_workflow(self, graph: dict[str, Any], inputs: dict[str, Any]) -> Turn: """工作流题：钉住那一版的图按输入跑一次探索运行（来源 eval）。不建图、不复核：考的是这一版本身。不发正式运行—— 正式运行会被 pick_base 和应用视图的「最近一次封存」取到（同期望探针不发正式运行的理由）。""" import time t = self.turn started = time.monotonic() try: await self._phase("run") if await self._run(graph, inputs) is not None: t.path = "ran" return t finally: t.timings["total_ms"] = int((time.monotonic() - started) * 1000) __all__ = ["APPROVAL_REASON", "Driver", "PHASE_TEXT", "REVIEW_PAGE_WAIT_S", "Turn", "TurnContext", "add_usage", "run_name", "stop_run", "with_writer", "writer_usage"]