tzt
8820e8da48
- inference.py（/chat/legacy 热路径）：kb.match 循环不变量外提（原每步全量重扫+重排序， 最坏 O(steps×rules×patterns)）；fired 查重 list→set - knowledge.py：Rule patterns 注册侧懒缓存小写副本（原每条规则每次匹配重复 lower）； match() 文本只 lower 一次（原逐规则重复）；load() 的 yaml 文件名集合提到循环外 - worker.py：本地端点生成器 httpx.AsyncClient 懒建复用（原每步新建/销毁连接， 对齐 ArchitectClient 惯用法；协作循环最多 10 次生成免重复建连） - pipeline.py（协作循环）：plan_by_id O(1) 步定义查找；done 集合增量维护 （原每轮重建 progress+archive 扫描）；领域只解析一次（原 _artifact_name 每步 全领域 kb.match）；_deps_done 支持传入预填集合（保持旧签名兼容） - v2stats.py：回合数分布改增量聚合（sum/max/分桶计数），summary() O(n)→O(1)， 不再持有无界 list（修长时运行内存增长） - gateway/agent.py + api.py：AgentService 运行计数 O(1) 化（原 register 全量扫描）， 状态迁移收敛到 _transition_state 单一入口（api.py cancel/异常两处绕过点一并接入， 消除计数与状态脱节隐患）；21 项 agent 测试全绿（两轮全量 230 passed 复核）
110 lines
4.7 KiB
Python
110 lines
4.7 KiB
Python
"""v2 统计聚合器（V2Stats）—— token 计量与账单（纯标准库）。
|
||
|
||
T9：每请求 API token 记账；聚合快路径命中率、回合数分布、熔断次数、累计 token/成本。
|
||
配合 /metrics 对外透出（论文 E1 token 经济学数据来源之一）。
|
||
|
||
性能设计（2026-09 优化）：回合数分布以"和/最大值/分桶计数"增量维护，
|
||
summary() 从每次全量重算 O(n) 降为 O(桶数)，且不再持有无界 list。
|
||
"""
|
||
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] = {}
|
||
# 回合数增量聚合（等价于全量 list 的 sum/max/histogram，内存 O(桶数)）
|
||
self._rounds_count = 0
|
||
self._rounds_sum = 0
|
||
self._rounds_max = 0
|
||
self._rounds_hist: Dict[str, 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
|
||
rounds = getattr(result, "rounds_used", 0)
|
||
self._rounds_count += 1
|
||
self._rounds_sum += rounds
|
||
if rounds > self._rounds_max:
|
||
self._rounds_max = rounds
|
||
bucket = str(rounds) if rounds <= 10 else ">10"
|
||
self._rounds_hist[bucket] = self._rounds_hist.get(bucket, 0) + 1
|
||
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_n = self._rounds_count
|
||
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(self._rounds_sum / rounds_n, 2) if rounds_n else 0.0,
|
||
"max": self._rounds_max if rounds_n else 0,
|
||
"distribution": dict(self._rounds_hist),
|
||
},
|
||
"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
|