tzt
374dc765c2
- T17 模型池：PoolStore（local/budget/premium 条目 + architect/worker/agent 角色指派），
/pool CRUD+连通测试+模型探测端点；build_v2_pipeline 池指派优先（测试 override 最后）；
V2Stats 新增 by_model 按 token/成本分账
- T18 智能体：OpenAI 兼容工具调用客户端（transport 可注入）+ AgentService
（事件落盘 agent_runs/{id}/events.jsonl）+ /agent 提交/status/events/SSE stream
+ 工作区浏览/读取端点（越界 400）；轮数与 token 双护栏，模型经池 agent 角色或经典回退
- 新增测试 16 项，全量 262 passed（httpx 假注入，不依赖真实模型/key）
97 lines
4.1 KiB
Python
97 lines
4.1 KiB
Python
"""v2 统计聚合器（V2Stats）—— token 计量与账单（纯标准库）。
|
|
|
|
T9：每请求 API token 记账；聚合快路径命中率、回合数分布、熔断次数、累计 token/成本。
|
|
配合 /metrics 对外透出（论文 E1 token 经济学数据来源之一）。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
class V2Stats:
|
|
"""线程安全的 v2 运行统计。"""
|
|
|
|
def __init__(self):
|
|
self._lock = threading.Lock()
|
|
self._total = 0
|
|
self._fast_path = 0
|
|
self._breach = 0
|
|
self._by_status: Dict[str, int] = {}
|
|
self._rounds: List[int] = []
|
|
self._api_input_tokens = 0
|
|
self._api_output_tokens = 0
|
|
self._api_cost_usd = 0.0
|
|
self._by_model: Dict[str, Dict[str, Any]] = {}
|
|
self._recent: List[Dict[str, Any]] = []
|
|
|
|
def record(self, result) -> None:
|
|
"""记录一次 PipelineResult。"""
|
|
with self._lock:
|
|
self._total += 1
|
|
if getattr(result, "fast_path", False):
|
|
self._fast_path += 1
|
|
status = getattr(result, "status", "?")
|
|
self._by_status[status] = self._by_status.get(status, 0) + 1
|
|
self._rounds.append(getattr(result, "rounds_used", 0))
|
|
if "breach" in " ".join(getattr(result, "route", [])):
|
|
self._breach += 1
|
|
self._api_input_tokens += getattr(result, "api_input_tokens", 0)
|
|
self._api_output_tokens += getattr(result, "api_output_tokens", 0)
|
|
self._api_cost_usd += getattr(result, "cost_est", 0.0)
|
|
# 按模型分账（token / 成本 / 次数）
|
|
model = getattr(result, "model_used", None) or "unknown"
|
|
in_tok = getattr(result, "api_input_tokens", 0)
|
|
out_tok = getattr(result, "api_output_tokens", 0)
|
|
bucket = self._by_model.setdefault(model, {
|
|
"requests": 0, "input_tokens": 0, "output_tokens": 0, "cost_est_usd": 0.0,
|
|
})
|
|
bucket["requests"] += 1
|
|
bucket["input_tokens"] += in_tok
|
|
bucket["output_tokens"] += out_tok
|
|
bucket["cost_est_usd"] = round(bucket["cost_est_usd"] + getattr(result, "cost_est", 0.0), 6)
|
|
self._recent.append({
|
|
"request_id": getattr(result, "request_id", ""),
|
|
"status": status,
|
|
"fast_path": getattr(result, "fast_path", False),
|
|
"api_input_tokens": getattr(result, "api_input_tokens", 0),
|
|
"api_output_tokens": getattr(result, "api_output_tokens", 0),
|
|
"rounds_used": getattr(result, "rounds_used", 0),
|
|
})
|
|
if len(self._recent) > 200:
|
|
self._recent = self._recent[-200:]
|
|
|
|
def summary(self) -> Dict[str, Any]:
|
|
with self._lock:
|
|
total = self._total
|
|
rounds = self._rounds
|
|
return {
|
|
"total_requests": total,
|
|
"fast_path_rate": round(self._fast_path / total, 4) if total else 0.0,
|
|
"status_distribution": dict(self._by_status),
|
|
"breach_count": self._breach,
|
|
"rounds_used": {
|
|
"avg": round(sum(rounds) / len(rounds), 2) if rounds else 0.0,
|
|
"max": max(rounds) if rounds else 0,
|
|
"distribution": _histogram(rounds),
|
|
},
|
|
"api_tokens": {
|
|
"input": self._api_input_tokens,
|
|
"output": self._api_output_tokens,
|
|
"total": self._api_input_tokens + self._api_output_tokens,
|
|
"cost_est_usd": round(self._api_cost_usd, 6),
|
|
},
|
|
"by_model": {
|
|
m: {**b, "cost_est_usd": round(b["cost_est_usd"], 6)}
|
|
for m, b in self._by_model.items()
|
|
},
|
|
}
|
|
|
|
|
|
def _histogram(values: List[int], max_bucket: int = 10) -> Dict[str, int]:
|
|
out: Dict[str, int] = {}
|
|
for v in values:
|
|
key = str(v) if v <= max_bucket else f">{max_bucket}"
|
|
out[key] = out.get(key, 0) + 1
|
|
return out
|