diff --git a/backend/app/eval/answer.py b/backend/app/eval/answer.py index 9d281947..fd1cfa0b 100644 --- a/backend/app/eval/answer.py +++ b/backend/app/eval/answer.py @@ -107,19 +107,21 @@ class Turn: "timings": dict(self.timings), "own_models": list(self.own_models)} +#: 用量汇总里相加的计数(providers/usage 的 summary):unfinished_calls 是没做完的调用(被取消、超时,token 只计了已报的) +_COUNTED = ("input_tokens", "output_tokens", "calls", "unpriced_calls", "unfinished_calls") + + def _add_usage(into: dict[str, Any] | None, more: Mapping[str, Any] | None) -> dict[str, Any]: """两份 providers/usage 汇总相加(by_model 逐个模型加)。""" - out = copy.deepcopy(into) if into else {"by_model": {}, "input_tokens": 0, "output_tokens": 0, "calls": 0, - "cost_usd": 0.0, "unpriced_calls": 0} + out = copy.deepcopy(into) if into else {"by_model": {}, **dict.fromkeys(_COUNTED, 0), "cost_usd": 0.0} if not more: return out - for key in ("input_tokens", "output_tokens", "calls", "unpriced_calls"): + for key in _COUNTED: out[key] = int(out.get(key) or 0) + int(more.get(key) or 0) out["cost_usd"] = round(float(out.get("cost_usd") or 0) + float(more.get("cost_usd") or 0), 6) for model, row in (more.get("by_model") or {}).items(): - mine = out["by_model"].setdefault(model, {"input_tokens": 0, "output_tokens": 0, "calls": 0, "cost_usd": 0.0, - "unpriced_calls": 0}) - for key in ("input_tokens", "output_tokens", "calls", "unpriced_calls"): + mine = out["by_model"].setdefault(model, {**dict.fromkeys(_COUNTED, 0), "cost_usd": 0.0}) + for key in _COUNTED: mine[key] = int(mine.get(key) or 0) + int(row.get(key) or 0) mine["cost_usd"] = round(float(mine.get("cost_usd") or 0) + float(row.get("cost_usd") or 0), 6) return out diff --git a/backend/app/providers/usage.py b/backend/app/providers/usage.py index 56298a2c..35079236 100644 --- a/backend/app/providers/usage.py +++ b/backend/app/providers/usage.py @@ -18,6 +18,12 @@ async generator:with 块跨着 yield 的话,变量在两次 yield 之间留 流末尾那一段用量收得到;只有模型在那一行之后还往下写超过 DRAIN_CHARS 字的闲话时才当场关流(不为闲话付钱),连接断了也一样, 这一次只拿得到已经报了的部分:照样计次数、计已报的 token,记成估不出,不悄悄少算。 +**没做完的调用**(unfinished_calls)。被取消(挑表超时被 asyncio.wait_for 取消、调用方停了)、超时、流读到一半被关掉的调用 +都记一次调用、标「未完成」:token 只计已经报了的(一段都没收到的记 0,当作不知道,记成估不出),调用次数照实。以前挑表超时 +的那一次一点都不计,问数据一轮、评测按角色的调用次数就少一次。LangChain 的 ainvoke 被取消时不一定回调 on_llm_error +(agenerate 卡在 gather 上被取消,后面的回调走不到):开始了、到汇总时还没有结束回调的调用(_pending 里剩下的)一并算未完成。 +请求刚发出去就被拒了的(鉴权错、参数不对,一段都没收到)不算一次调用:模型没干活。 + 模型名按发起调用时的参数取(invocation_params 的 model / model_name,就是接入里配的模型 id,价格目录按它查);调用参数里 没有的(以前的演示模型就没报,评测汇总里助手用量的模型因此记成 unknown),看 LangChain 照模型对象给回调的 metadata 里的 ls_model_name;再取不到看回复里的 model_name,都没有才记 UNKNOWN_MODEL。OpenAI 兼容、Anthropic 的模型在调用参数里就带着 @@ -37,6 +43,17 @@ from langchain_core.tracers.context import register_configure_hook #: 取不到模型名的调用记在这个名下 UNKNOWN_MODEL = "unknown" +#: 每个模型累计的几项(unreported 用量没报全的次数、unfinished 没做完的次数) +_COUNTERS = ("input_tokens", "output_tokens", "calls", "unreported", "unfinished") + + +def _abandoned(error: BaseException) -> bool: + """调用是半路放弃的(被取消、超时),不是一发出去就被拒:模型可能已经在干活、已经计费。""" + import asyncio + + if isinstance(error, (asyncio.CancelledError, TimeoutError)): + return True + return any(cls.__name__.endswith(("Timeout", "TimeoutError")) for cls in type(error).__mro__) class Meter(BaseCallbackHandler): @@ -51,6 +68,10 @@ class Meter(BaseCallbackHandler): self._pending: dict[UUID, str] = {} self._models: dict[str, dict[str, int]] = {} + def _row(self, model: str) -> dict[str, int]: + """这个模型的累计(调用方持锁)。unreported:用量没报全的次数;unfinished:没做完的次数(见模块开头)。""" + return self._models.setdefault(model, dict.fromkeys(_COUNTERS, 0)) + # ------------------------------------------------------------------ 回调 def _started(self, run_id: UUID, serialized: dict[str, Any] | None, kwargs: dict[str, Any]) -> None: @@ -83,7 +104,7 @@ class Meter(BaseCallbackHandler): with self._lock: model = self._pending.pop(run_id, "") or str(meta.get("model_name") or meta.get("model") or "") \ or UNKNOWN_MODEL - row = self._models.setdefault(model, {"input_tokens": 0, "output_tokens": 0, "calls": 0, "unreported": 0}) + row = self._row(model) row["calls"] += 1 if usage: row["input_tokens"] += int(usage.get("input_tokens") or 0) @@ -92,8 +113,9 @@ class Meter(BaseCallbackHandler): row["unreported"] += 1 def on_llm_error(self, error: BaseException, *, run_id: UUID, **kwargs: Any) -> None: - """流读到一半停了(生成过程在最后一行之后读到太多闲话就关流,或者连接断了):LangChain 照样交来已经收到的那几段。 - 有收到东西的照计一次、记已报的 token,算用量没报全;一段都没收到的(请求刚发出去就被拒了)不计。""" + """调用没做完:流读到一半停了(生成过程在最后一行之后读到太多闲话就关流,或者连接断了)、被取消、超时。LangChain + 照样交来已经收到的那几段。有收到东西的、被取消或超时的照计一次、记已报的 token,算用量没报全、未完成;一段都没收到 + 的报错(请求刚发出去就被拒了)不计。""" from langchain_core.outputs import ChatGenerationChunk response = kwargs.get("response") @@ -105,36 +127,48 @@ class Meter(BaseCallbackHandler): partial = None with self._lock: model = self._pending.pop(run_id, "") - if partial is None: + if partial is None and not _abandoned(error): return meta = getattr(partial, "response_metadata", None) or {} model = model or str(meta.get("model_name") or meta.get("model") or "") or UNKNOWN_MODEL - row = self._models.setdefault(model, {"input_tokens": 0, "output_tokens": 0, "calls": 0, "unreported": 0}) + row = self._row(model) row["calls"] += 1 row["unreported"] += 1 + row["unfinished"] += 1 reported = getattr(partial, "usage_metadata", None) or {} row["input_tokens"] += int(reported.get("input_tokens") or 0) row["output_tokens"] += int(reported.get("output_tokens") or 0) + def _rows(self) -> dict[str, dict[str, int]]: + """累计的一份拷贝,开始了还没结束回调的调用算成没做完的(调用方持锁)。""" + rows = {k: dict(v) for k, v in self._models.items()} + for model in self._pending.values(): + row = rows.setdefault(model or UNKNOWN_MODEL, dict.fromkeys(_COUNTERS, 0)) + row["calls"] += 1 + row["unreported"] += 1 + row["unfinished"] += 1 + return rows + # ------------------------------------------------------------------ 合并、汇总 def merge(self, other: Meter) -> None: """把另一次计量并进来(嵌套的计量出块时并进外层)。""" with other._lock: - rows = {k: dict(v) for k, v in other._models.items()} + rows = other._rows() with self._lock: for model, row in rows.items(): - mine = self._models.setdefault(model, dict.fromkeys(row, 0)) + mine = self._row(model) for key, value in row.items(): mine[key] = mine.get(key, 0) + value def summary(self) -> dict[str, Any]: - """{by_model: {模型: {input_tokens, output_tokens, calls, cost_usd, unpriced_calls}}, input_tokens, output_tokens, - calls, cost_usd, unpriced_calls}。cost_usd 只是估得出的那部分。""" + """{by_model: {模型: {input_tokens, output_tokens, calls, cost_usd, unpriced_calls, unfinished_calls}}, input_tokens, + output_tokens, calls, cost_usd, unpriced_calls, unfinished_calls}。cost_usd 只是估得出的那部分;unfinished_calls 次 + 没做完(token 只计了已报的)。""" from app.providers.catalog import estimate_cost, find_pricing with self._lock: - rows = {k: dict(v) for k, v in self._models.items()} + rows = self._rows() by_model: dict[str, dict[str, Any]] = {} for model, row in sorted(rows.items()): pricing = find_pricing(model) @@ -146,9 +180,9 @@ class Meter(BaseCallbackHandler): by_model[model] = {"input_tokens": row["input_tokens"], "output_tokens": row["output_tokens"], "calls": row["calls"], "cost_usd": round(estimate_cost(model, row["input_tokens"], row["output_tokens"]), 6), - "unpriced_calls": unpriced} + "unpriced_calls": unpriced, "unfinished_calls": row["unfinished"]} total = {key: sum(r[key] for r in by_model.values()) - for key in ("input_tokens", "output_tokens", "calls", "unpriced_calls")} + for key in ("input_tokens", "output_tokens", "calls", "unpriced_calls", "unfinished_calls")} return {"by_model": by_model, **total, "cost_usd": round(sum(r["cost_usd"] for r in by_model.values()), 6)} diff --git a/backend/tests/test_usage_meter.py b/backend/tests/test_usage_meter.py index 828a8130..894a63cf 100644 --- a/backend/tests/test_usage_meter.py +++ b/backend/tests/test_usage_meter.py @@ -96,7 +96,8 @@ async def test_metering_counts_by_model_and_prices_from_the_catalog(): await Scripted(model=FREE, report=False).ainvoke("你好") got = m.summary() assert got["by_model"][PRICED] == {"input_tokens": 2000, "output_tokens": 400, "calls": 2, - "cost_usd": round(estimate_cost(PRICED, 2000, 400), 6), "unpriced_calls": 0} + "cost_usd": round(estimate_cost(PRICED, 2000, 400), 6), "unpriced_calls": 0, + "unfinished_calls": 0} assert got["by_model"][UNPRICED]["unpriced_calls"] == 1 and got["by_model"][UNPRICED]["cost_usd"] == 0 # 目录外的模型名(多了后缀)照样算估不出;目录价是 0 的本地模型没报用量也不算估不出 assert got["by_model"][PRICED + "-silent"]["unpriced_calls"] == 1 @@ -184,6 +185,64 @@ async def test_the_eval_driver_on_the_demo_model_records_its_name(monkeypatch): assert set(got) == {"mock-fast"} and got["mock-fast"]["calls"] >= 1, got +class Slow(Scripted): + """回得很慢的模型:调用方等不及(挑表超时被 asyncio.wait_for 取消)。partial=True 时先报一段输入用量再卡住。""" + + delay: float = 5.0 + partial: bool = False + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: + self.calls += 1 + await asyncio.sleep(self.delay) + return ChatResult(generations=[ChatGeneration(message=AIMessage(content="晚了"))]) + + async def _astream(self, messages, stop=None, run_manager=None, **kwargs): + self.calls += 1 + if self.partial: + yield ChatGenerationChunk(message=AIMessageChunk(content="{", usage_metadata={ + "input_tokens": 900, "output_tokens": 0, "total_tokens": 900})) + await asyncio.sleep(self.delay) + yield ChatGenerationChunk(message=AIMessageChunk(content="晚了")) + + +class Refused(Scripted): + """请求刚发出去就被拒了(鉴权错、参数不对):模型没干活,不算一次调用。""" + + async def _agenerate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult: + raise ValueError("401 鉴权失败") + + +async def test_a_call_cancelled_before_any_result_still_counts_as_an_unfinished_call(): + """挑表超时被取消的那次调用以前一点都不计(LangChain 取消时连 on_llm_error 都不一定回调):调用次数少了一次。现在 + 记一次调用、token 不知道(没有部分结果的记 0、算没报全),标「未完成」;有部分用量的照实记已报的;被拒的仍不计。""" + from app.look import pick + + m = usage.meter() + with usage.metering(m): + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(Slow(model=PRICED).ainvoke("挑表"), timeout=0.05) + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(pick._ask(Slow(model=UNPRICED), [("human", "挑表")]), timeout=0.05) + + async def read(): + async for _ in Slow(model=PRICED + "-partial", partial=True).astream("挑表"): + pass + + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(read(), timeout=0.05) + with pytest.raises(ValueError): + await Refused(model=PRICED + "-refused").ainvoke("挑表") + got = m.summary() + rows = got["by_model"] + assert (rows[PRICED]["calls"], rows[PRICED]["unfinished_calls"], rows[PRICED]["input_tokens"]) == (1, 1, 0), rows + assert rows[PRICED]["unpriced_calls"] == 1, "token 不知道,金额也就估不出" + assert (rows[UNPRICED]["calls"], rows[UNPRICED]["unfinished_calls"]) == (1, 1), "挑表(look/pick._ask)超时同样计" + partial = rows[PRICED + "-partial"] + assert (partial["calls"], partial["unfinished_calls"], partial["input_tokens"]) == (1, 1, 900), partial + assert PRICED + "-refused" not in rows, "被拒的不算一次调用" + assert got["calls"] == 3 and got["unfinished_calls"] == 3 + + async def test_outside_the_meter_nothing_is_counted(): m = usage.meter() with usage.metering(m): diff --git a/frontend/src/lib/termsEval.ts b/frontend/src/lib/termsEval.ts index d09caa4c..5b9d3acb 100644 --- a/frontend/src/lib/termsEval.ts +++ b/frontend/src/lib/termsEval.ts @@ -441,7 +441,9 @@ export const EVAL_TEXT = { timing: { build_ms: '建图', run_wall_ms: '运行(墙钟)', run_active_ms: '运行(执行)', review_ms: '复核', total_ms: '整题' } as Record, usage: '按角色的用量', tokens: (input: string, output: string) => `输入 ${input} · 输出 ${output}`, - callsN: (n: number) => `${formatNumber(n)} 次调用`, + /** unfinished:没做完的调用(被取消、超时,token 只计了已报的;服务端 providers/usage 的 unfinished_calls) */ + callsN: (n: number, unfinished = 0) => + `${formatNumber(n)} 次调用${unfinished > 0 ? `(其中 ${formatNumber(unfinished)} 次未完成:被取消或超时,token 只计了已报的部分)` : ''}`, role: { assistant: '助手', writer: '写作', judge: '结论裁判', gold: '算期望' } as Record, passRates: '重复评的题的通过率', outcomes: '各期望的结论', diff --git a/frontend/src/pages/evals/EvaluationPage.tsx b/frontend/src/pages/evals/EvaluationPage.tsx index 581725fa..ee2cf788 100644 --- a/frontend/src/pages/evals/EvaluationPage.tsx +++ b/frontend/src/pages/evals/EvaluationPage.tsx @@ -278,7 +278,7 @@ function Overview({ run }: { run: EvalRun }) {
  • {O.role[role]}:{usageCost(u)} {u.input_tokens != null && · {O.tokens(formatTokens(u.input_tokens, { compact: true }), formatTokens(u.output_tokens ?? 0, { compact: true }))}} - {u.calls != null && · {O.callsN(u.calls)}} + {u.calls != null && · {O.callsN(u.calls, u.unfinished_calls ?? 0)}}
  • ) })} diff --git a/frontend/src/store/chat.ts b/frontend/src/store/chat.ts index 1f8f2bed..11413482 100644 --- a/frontend/src/store/chat.ts +++ b/frontend/src/store/chat.ts @@ -174,6 +174,8 @@ export interface UsageTotal { cost_usd?: number | null unpriced_calls?: number calls?: number + /** 没做完的调用(被取消、超时):算在 calls 里,token 只计了已报的(服务端 providers/usage 的 unfinished_calls) */ + unfinished_calls?: number input_tokens?: number output_tokens?: number } @@ -498,7 +500,7 @@ function buildMeta(t: ChatTurn): TurnMeta { } } -const USAGE_KEYS = ['cost_usd', 'unpriced_calls', 'calls', 'input_tokens', 'output_tokens'] as const +const USAGE_KEYS = ['cost_usd', 'unpriced_calls', 'calls', 'unfinished_calls', 'input_tokens', 'output_tokens'] as const /** 两份用量相加(缺的当 0)。b 不是对象时原样交回 a */ export function addUsage(a: UsageTotal | undefined, b: UsageTotal | null | undefined): UsageTotal | undefined { diff --git a/frontend/src/typesEval.ts b/frontend/src/typesEval.ts index 8ed3e80a..1b8a0c9c 100644 --- a/frontend/src/typesEval.ts +++ b/frontend/src/typesEval.ts @@ -415,7 +415,10 @@ export interface EvalUsage { calls?: number cost_usd?: number | null unpriced_calls?: number - by_model?: Record + /** 没做完的调用(被取消、超时):算在 calls 里,token 只计了已报的 */ + unfinished_calls?: number + by_model?: Record } export interface EvalTiming {