mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 16:41:59 +00:00
没做完的模型调用也计一次：挑表超时被取消的那次不再漏掉
真实模型验收里，挑表（look/pick 的 asyncio.wait_for）超时被取消的那次调用一点都不计：LangChain 的 ainvoke 被取消时 卡在 agenerate 的 gather 上，on_llm_error 根本走不到；astream 被取消时会回调，可一段都没收到的又被当成「请求刚发出去 就被拒了」不计。问数据一轮、评测按角色的调用次数因此少一次。 providers/usage.Meter 现在多记一项 unfinished（汇总里是 unfinished_calls）：被取消、超时（按异常类型认，含各家 SDK 的 …Timeout）、流读到一半被关掉的调用都记一次调用、标未完成，token 只计已经报了的（一段都没收到的记 0、算没报全， 金额也就估不出）；开始了、到汇总（和嵌套并入外层）时还没有结束回调的调用一并算未完成。请求一发出去就被拒的仍不计。 评测汇总相加（eval/answer._add_usage）带上这一项；评测页按角色的用量写「其中 N 次未完成」；问数据页的这一轮用量 相加时也留着它。 测试：test_usage_meter.py::test_a_call_cancelled_before_any_result_still_counts_as_an_unfinished_call（ainvoke 被 wait_for 取消、look/pick._ask 超时、astream 收到一段输入用量后被取消、被拒的不计）；by_model 的形状多了一项， test_metering_counts_by_model_and_prices_from_the_catalog 和评测同口径那条跟着改。去掉修复回跑失败。
This commit is contained in:
ThearchyHelios
@@ -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
|
||||
|
||||
@@ -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)}
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -441,7 +441,9 @@ export const EVAL_TEXT = {
|
||||
timing: { build_ms: '建图', run_wall_ms: '运行（墙钟）', run_active_ms: '运行（执行）', review_ms: '复核', total_ms: '整题' } as Record<string, string>,
|
||||
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<string, string>,
|
||||
passRates: '重复评的题的通过率',
|
||||
outcomes: '各期望的结论',
|
||||
|
||||
@@ -278,7 +278,7 @@ function Overview({ run }: { run: EvalRun }) {
|
||||
<li key={role} data-eval-usage-role={role}>
|
||||
{O.role[role]}：{usageCost(u)}
|
||||
{u.input_tokens != null && <span className="text-faint"> · {O.tokens(formatTokens(u.input_tokens, { compact: true }), formatTokens(u.output_tokens ?? 0, { compact: true }))}</span>}
|
||||
{u.calls != null && <span className="text-faint"> · {O.callsN(u.calls)}</span>}
|
||||
{u.calls != null && <span className="text-faint"> · {O.callsN(u.calls, u.unfinished_calls ?? 0)}</span>}
|
||||
</li>
|
||||
)
|
||||
})}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -415,7 +415,10 @@ export interface EvalUsage {
|
||||
calls?: number
|
||||
cost_usd?: number | null
|
||||
unpriced_calls?: number
|
||||
by_model?: Record<string, { input_tokens: number; output_tokens: number; calls: number; cost_usd: number; unpriced_calls: number }>
|
||||
/** 没做完的调用（被取消、超时）：算在 calls 里，token 只计了已报的 */
|
||||
unfinished_calls?: number
|
||||
by_model?: Record<string, { input_tokens: number; output_tokens: number; calls: number; cost_usd: number; unpriced_calls: number
|
||||
unfinished_calls?: number }>
|
||||
}
|
||||
|
||||
export interface EvalTiming {
|
||||
|
||||
Reference in New Issue
Block a user