i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 17:30:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(usage): 改进请求追踪与缓存预热

Browse Source 
- 区分请求重试与故障转移：新增 has_retry 标识亲和缓存重试
- 修正候选 TTFB 计算：记录候选自身的首字节时间而非全局时间
- Streaming 状态同步 rate_multiplier，支持按 API 格式配置
- 新增缓存预热服务：启动时预热仪表盘统计、热力图、每日统计
- 优化缓存 TTL 配置：仪表盘统计 2 分钟、热力图和每日统计 10 分钟
- 自动刷新间隔从 10 秒调整为 5 秒
This commit is contained in:
fawney19
2026-01-15 11:45:46 +08:00
parent 24b65d76b5
commit 514bf7e3ed
15 changed files with 386 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
src/api/admin/usage/routes.py
View File

@@ -1,5 +1,6 @@
"""管理员使用情况统计路由。"""
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@@ -717,19 +718,42 @@ class AdminUsageRecordsAdapter(AdminApiAdapter):
request_ids = [usage.request_id for usage, _, _, _, _ in records if usage.request_id]
fallback_map = {}
retry_map = {}
if request_ids:
# 查询每个请求的候选执行情况
# 只统计实际执行的候选（success 或 failed），不包括 skipped/pending/available
executed_counts = (
db.query(RequestCandidate.request_id, func.count(RequestCandidate.id))
executed_candidates = (
db.query(
RequestCandidate.request_id,
RequestCandidate.candidate_index,
RequestCandidate.retry_index,
)
.filter(
RequestCandidate.request_id.in_(request_ids),
RequestCandidate.status.in_(["success", "failed"]),
)
.group_by(RequestCandidate.request_id)
.all()
)
# 如果实际执行的候选数 > 1，说明发生了 Provider 切换
fallback_map = {req_id: count > 1 for req_id, count in executed_counts}
# 按 request_id 分组分析
request_candidates: dict[str, list[tuple[int, int]]] = defaultdict(list)
for req_id, candidate_idx, retry_idx in executed_candidates:
request_candidates[req_id].append((candidate_idx, retry_idx))
for req_id, candidates in request_candidates.items():
# 提取所有不同的 candidate_index
unique_candidates = set(c[0] for c in candidates)
# 如果有多个不同的 candidate_index，说明发生了 Fallback（Provider 切换）
fallback_map[req_id] = len(unique_candidates) > 1
# 检查是否有重试：同一个 candidate_index 有多个 retry_index
has_retry = False
for candidate_idx in unique_candidates:
retry_indices = [c[1] for c in candidates if c[0] == candidate_idx]
if len(retry_indices) > 1 or (retry_indices and max(retry_indices) > 0):
has_retry = True
break
retry_map[req_id] = has_retry
context.add_audit_metadata(
action="usage_records",
@@ -809,6 +833,7 @@ class AdminUsageRecordsAdapter(AdminApiAdapter):
"error_message": usage.error_message,
"status": usage.status, # 请求状态: pending, streaming, completed, failed
"has_fallback": fallback_map.get(usage.request_id, False),
"has_retry": retry_map.get(usage.request_id, False),
"api_format": usage.api_format
or (endpoint.api_format if endpoint and endpoint.api_format else None),
"api_key_name": provider_api_key.name if provider_api_key else None,

5
src/api/dashboard/routes.py
View File

@@ -11,6 +11,7 @@ from sqlalchemy.orm import Session
from src.api.base.adapter import ApiAdapter, ApiMode
from src.api.base.admin_adapter import AdminApiAdapter
from src.api.base.pipeline import ApiRequestPipeline
from src.config.constants import CacheTTL
from src.core.enums import UserRole
from src.database import get_db
from src.models.database import ApiKey, Provider, RequestCandidate, StatsDaily, StatsDailyModel, Usage
@@ -177,7 +178,7 @@ class DashboardStatsAdapter(DashboardAdapter):
class AdminDashboardStatsAdapter(AdminApiAdapter):
@cache_result(key_prefix="dashboard:admin:stats", ttl=60, user_specific=False)
@cache_result(key_prefix="dashboard:admin:stats", ttl=CacheTTL.DASHBOARD_STATS, user_specific=False)
async def handle(self, context): # type: ignore[override]
"""管理员仪表盘统计 - 使用预聚合数据优化性能"""
from zoneinfo import ZoneInfo
@@ -786,7 +787,7 @@ class DashboardProviderStatusAdapter(DashboardAdapter):
class DashboardDailyStatsAdapter(DashboardAdapter):
days: int
@cache_result(key_prefix="dashboard:daily:stats", ttl=300, user_specific=True)
@cache_result(key_prefix="dashboard:daily:stats", ttl=CacheTTL.DASHBOARD_DAILY, user_specific=True)
async def handle(self, context): # type: ignore[override]
from zoneinfo import ZoneInfo
from src.services.system.stats_aggregator import APP_TIMEZONE

2
src/api/handlers/base/base_handler.py
View File

@@ -432,6 +432,7 @@ class BaseMessageHandler:
endpoint_id = ctx.endpoint_id
key_id = ctx.key_id
first_byte_time_ms = ctx.first_byte_time_ms
api_format = ctx.api_format
# 如果 provider 为空，记录警告（不应该发生，但用于调试）
if not provider:
@@ -455,6 +456,7 @@ class BaseMessageHandler:
provider_endpoint_id=endpoint_id,
provider_api_key_id=key_id,
first_byte_time_ms=first_byte_time_ms,
api_format=api_format,
)
finally:
db.close()

18
src/api/handlers/base/cli_handler_base.py
View File

@@ -1432,6 +1432,16 @@ class CliMessageHandlerBase(BaseMessageHandler):
if ctx.attempt_id:
from src.services.request.candidate import RequestCandidateService
# 计算候选自身的 TTFB
candidate_first_byte_time_ms: Optional[int] = None
if ctx.first_byte_time_ms is not None:
candidate_first_byte_time_ms = RequestCandidateService.calculate_candidate_ttfb(
db=bg_db,
candidate_id=ctx.attempt_id,
request_start_time=self.start_time,
global_first_byte_time_ms=ctx.first_byte_time_ms,
)
# 根据状态码决定是成功还是失败
# 499 = 客户端断开连接，应标记为失败
# 503 = 服务不可用（如流中断），应标记为失败
@@ -1443,8 +1453,8 @@ class CliMessageHandlerBase(BaseMessageHandler):
"chunk_count": ctx.chunk_count,
"data_count": ctx.data_count,
}
if ctx.first_byte_time_ms is not None:
extra_data["first_byte_time_ms"] = ctx.first_byte_time_ms
if candidate_first_byte_time_ms is not None:
extra_data["first_byte_time_ms"] = candidate_first_byte_time_ms
RequestCandidateService.mark_candidate_failed(
db=bg_db,
candidate_id=ctx.attempt_id,
@@ -1462,8 +1472,8 @@ class CliMessageHandlerBase(BaseMessageHandler):
"chunk_count": ctx.chunk_count,
"data_count": ctx.data_count,
}
if ctx.first_byte_time_ms is not None:
extra_data["first_byte_time_ms"] = ctx.first_byte_time_ms
if candidate_first_byte_time_ms is not None:
extra_data["first_byte_time_ms"] = candidate_first_byte_time_ms
RequestCandidateService.mark_candidate_success(
db=bg_db,
candidate_id=ctx.attempt_id,

12
src/api/handlers/base/stream_telemetry.py
View File

@@ -132,7 +132,7 @@ class StreamTelemetryRecorder:
)
# 更新候选记录状态
await self._update_candidate_status(bg_db, ctx, response_time_ms)
await self._update_candidate_status(bg_db, ctx, response_time_ms, start_time)
finally:
if bg_db:
@@ -234,6 +234,7 @@ class StreamTelemetryRecorder:
db: Session,
ctx: StreamContext,
response_time_ms: int,
request_start_time: float,
) -> None:
"""更新候选记录状态"""
if not ctx.attempt_id:
@@ -246,7 +247,14 @@ class StreamTelemetryRecorder:
"data_count": ctx.data_count,
}
if ctx.first_byte_time_ms is not None:
extra_data["first_byte_time_ms"] = ctx.first_byte_time_ms
# 计算候选自身的 TTFB
first_byte_time_ms = RequestCandidateService.calculate_candidate_ttfb(
db=db,
candidate_id=ctx.attempt_id,
request_start_time=request_start_time,
global_first_byte_time_ms=ctx.first_byte_time_ms,
)
extra_data["first_byte_time_ms"] = first_byte_time_ms
if ctx.is_success():
RequestCandidateService.mark_candidate_success(