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

feat: 缓存计费细分、能力匹配优化、用户模型调用计数

Browse Source 
1. 缓存创建 tokens 区分 5min/1h TTL，支持按缓存时长差异化计费
   - Usage 表新增 cache_creation_input_tokens_5m/1h 字段
   - Claude handler 解析新格式 (ephemeral_5m/1h, claude_cache_creation_5/1h)
   - 计费规则支持 cache_ttl_pricing 覆盖 cache_creation 价格

2. 能力匹配机制优化
   - COMPATIBLE 能力不再硬过滤，改为排序阶段通过 capability_miss_count 优先级处理
   - cache_1h 改为 COMPATIBLE + REQUEST_PARAM（自动检测请求体中的 ttl=1h）
   - gemini_files 改为 EXCLUSIVE + REQUEST_PARAM（自动检测 fileData.fileUri）
   - 移除前端模型偏好/能力配置 UI（不再需要用户手动配置）

3. 新增用户-模型维度调用次数计数器 (UserModelUsageCount)
   - 原子递增，避免从 Usage 表聚合查询
   - 前端模型目录和用户可用模型列表展示调用次数

4. 其他改进
   - global_model_id 改为必填（NOT NULL），清理孤立模型
   - 模型映射对话框支持从上游获取模型列表并分组折叠
   - 端点测试不再依赖端点启用状态
   - 异步任务页面对普通用户隐藏用户信息列
   - Dashboard 响应式布局断点调整 (sm -> lg)
   - 号池管理仅展示已启用号池的提供商
This commit is contained in:
fawney19
2026-02-28 11:44:08 +08:00
parent 82bbed2720
commit ecb16d345a
59 changed files with 1053 additions and 608 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/api/admin/provider_query.py
View File

@@ -30,6 +30,7 @@ from src.services.model.fetch_scheduler import (
from src.services.model.upstream_fetcher import (
EndpointFetchConfig,
UpstreamModelsFetchContext,
UpstreamModelsFetcherRegistry,
build_format_to_config,
fetch_models_for_key,
get_adapter_for_format,
@@ -235,7 +236,15 @@ async def query_available_models(
# 构建 api_format -> EndpointFetchConfig 映射（纯数据，不依赖 ORM session）
format_to_endpoint = build_format_to_config(provider.endpoints)
if not format_to_endpoint:
# 检查是否有注册自定义 fetcher（如预设模型），有则不依赖活跃 endpoint
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
# 延迟导入避免循环依赖（与 upstream_fetcher.fetch_models_for_key 保持一致）
from src.services.provider.envelope import ensure_providers_bootstrapped
ensure_providers_bootstrapped()
has_custom_fetcher = UpstreamModelsFetcherRegistry.get(provider_type) is not None
if not format_to_endpoint and not has_custom_fetcher:
raise HTTPException(status_code=400, detail="No active endpoints found for this provider")
# 如果指定了 api_key_id，只获取该 Key 的模型
@@ -253,7 +262,6 @@ async def query_available_models(
raise HTTPException(status_code=400, detail="No active API Key found for this provider")
# Antigravity: 按 tier/可用性排序后逐个尝试，成功即停止
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
if provider_type == ProviderType.ANTIGRAVITY:
return await _fetch_models_antigravity_ordered(
provider=provider,
@@ -610,12 +618,12 @@ async def test_model(
raise HTTPException(status_code=404, detail="Provider not found")
# 构建 api_format -> endpoint 映射 和 id -> endpoint 映射
# 测试不依赖端点启用状态，禁用的端点也可以用于测试连通性
format_to_endpoint: dict[str, ProviderEndpoint] = {}
id_to_endpoint: dict[str, ProviderEndpoint] = {}
for ep in provider.endpoints:
if ep.is_active:
format_to_endpoint[ep.api_format] = ep
id_to_endpoint[ep.id] = ep
format_to_endpoint[ep.api_format] = ep
id_to_endpoint[ep.id] = ep
# 找到合适的端点和 API Key
endpoint = None
@@ -628,7 +636,7 @@ async def test_model(
if not endpoint:
raise HTTPException(
status_code=404,
detail=f"No active endpoint found for API format: {request.api_format}",
detail=f"No endpoint found for API format: {request.api_format}",
)
if request.api_key_id:
@@ -657,7 +665,7 @@ async def test_model(
# 使用指定的端点
endpoint = id_to_endpoint.get(request.endpoint_id)
if not endpoint:
raise HTTPException(status_code=404, detail="Endpoint not found or not active")
raise HTTPException(status_code=404, detail="Endpoint not found")
if request.api_key_id:
# 同时指定了 Key，需要校验是否支持该端点格式

7
src/api/admin/providers/models.py
View File

@@ -715,6 +715,7 @@ class AdminImportFromUpstreamAdapter(AdminApiAdapter):
# 1. 检查是否已存在同名的 ProviderModel
existing = (
db.query(Model)
.options(joinedload(Model.global_model))
.filter(
Model.provider_id == self.provider_id,
Model.provider_model_name == model_id,
@@ -727,10 +728,8 @@ class AdminImportFromUpstreamAdapter(AdminApiAdapter):
success.append(
ImportFromUpstreamSuccessItem(
model_id=model_id,
global_model_id=existing.global_model_id or "",
global_model_name=(
existing.global_model.name if existing.global_model else ""
),
global_model_id=existing.global_model_id,
global_model_name=existing.global_model.name,
provider_model_id=existing.id,
created_global_model=False,
)

1
src/api/admin/providers/summary.py
View File

@@ -319,7 +319,6 @@ def _build_provider_summary(db: Session, provider: Provider) -> ProviderWithEndp
.filter(
Model.provider_id == provider.id,
Model.is_active == True,
Model.global_model_id.isnot(None),
)
.distinct()
.all()

5
src/api/admin/system.py
View File

@@ -2556,16 +2556,18 @@ def _purge_stats_and_reset_counters(db: Session) -> None:
class AdminPurgeUsageAdapter(AdminApiAdapter):
async def handle(self, context: ApiRequestContext) -> Any: # type: ignore[override]
"""清空全部使用记录及相关统计数据"""
from src.models.database import RequestCandidate
from src.models.database import RequestCandidate, UserModelUsageCount
db = context.db
usage_count = db.query(Usage).count()
candidates_count = db.query(RequestCandidate).count()
usage_counts_count = db.query(UserModelUsageCount).count()
# 清空使用记录
db.query(RequestCandidate).delete()
db.query(Usage).delete()
db.query(UserModelUsageCount).delete()
_purge_stats_and_reset_counters(db)
db.commit()
@@ -2575,6 +2577,7 @@ class AdminPurgeUsageAdapter(AdminApiAdapter):
"deleted": {
"usage_records": usage_count,
"request_candidates": candidates_count,
"user_model_usage_counts": usage_counts_count,
},
}

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

@@ -1259,6 +1259,8 @@ class AdminUsageDetailAdapter(AdminApiAdapter):
},
"cache_creation_input_tokens": usage_record.cache_creation_input_tokens,
"cache_read_input_tokens": usage_record.cache_read_input_tokens,
"cache_creation_input_tokens_5m": usage_record.cache_creation_input_tokens_5m or 0,
"cache_creation_input_tokens_1h": usage_record.cache_creation_input_tokens_1h or 0,
"cache_creation_cost": getattr(usage_record, "cache_creation_cost_usd", 0.0),
"cache_read_cost": getattr(usage_record, "cache_read_cost_usd", 0.0),
"request_cost": getattr(usage_record, "request_cost_usd", 0.0),

4
src/api/handlers/base/chat_sync_executor.py
View File

@@ -184,6 +184,8 @@ class ChatSyncExecutor:
output_tokens = usage_info.get("output_tokens", 0)
cache_creation_tokens = usage_info.get("cache_creation_input_tokens", 0)
cached_tokens = usage_info.get("cache_read_input_tokens", 0)
cache_creation_tokens_5m = usage_info.get("cache_creation_input_tokens_5m", 0)
cache_creation_tokens_1h = usage_info.get("cache_creation_input_tokens_1h", 0)
# 非流式成功时，返回给客户端的是提供商响应头（透传）
# JSONResponse 会自动设置 content-type，但我们记录实际返回的完整头
@@ -211,6 +213,8 @@ class ChatSyncExecutor:
provider_request_body=ctx.provider_request_body,
cache_creation_tokens=cache_creation_tokens,
cache_read_tokens=cached_tokens,
cache_creation_tokens_5m=cache_creation_tokens_5m,
cache_creation_tokens_1h=cache_creation_tokens_1h,
is_stream=False,
provider_request_headers=ctx.provider_request_headers,
api_format=api_format,

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

@@ -171,7 +171,7 @@ async def _calculate_and_record_usage(
provider = db.query(Provider).filter(Provider.id == provider_api_key.provider_id).first()
if provider:
for ep in provider.endpoints:
if ep.api_format == api_format and ep.is_active:
if ep.api_format == api_format:
provider_endpoint = ep
break

4
src/api/handlers/base/stream_context.py
View File

@@ -86,6 +86,8 @@ class StreamContext:
output_tokens: int = 0
cached_tokens: int = 0
cache_creation_tokens: int = 0
cache_creation_tokens_5m: int = 0 # 5min TTL 缓存创建
cache_creation_tokens_1h: int = 0 # 1h TTL 缓存创建
# 响应内容
_collected_text_parts: list[str] = field(default_factory=list, repr=False)
@@ -159,6 +161,8 @@ class StreamContext:
self.output_tokens = 0
self.cached_tokens = 0
self.cache_creation_tokens = 0
self.cache_creation_tokens_5m = 0
self.cache_creation_tokens_1h = 0
self.error_message = None
self.upstream_response = None
self.status_code = 200

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

@@ -228,6 +228,8 @@ class StreamTelemetryRecorder:
provider_request_body=ctx.provider_request_body,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
cache_creation_tokens_5m=ctx.cache_creation_tokens_5m,
cache_creation_tokens_1h=ctx.cache_creation_tokens_1h,
is_stream=True,
provider_request_headers=ctx.provider_request_headers,
api_format=ctx.api_format,
@@ -285,6 +287,8 @@ class StreamTelemetryRecorder:
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
cache_creation_tokens_5m=ctx.cache_creation_tokens_5m,
cache_creation_tokens_1h=ctx.cache_creation_tokens_1h,
response_body=response_body,
client_response_body=client_response_body,
response_headers=ctx.response_headers,
@@ -342,6 +346,8 @@ class StreamTelemetryRecorder:
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
cache_creation_tokens_5m=ctx.cache_creation_tokens_5m,
cache_creation_tokens_1h=ctx.cache_creation_tokens_1h,
response_body=response_body,
client_response_body=client_response_body,
response_headers=ctx.response_headers,

28
src/api/handlers/base/utils.py
View File

@@ -94,6 +94,34 @@ def extract_cache_creation_tokens(usage: dict[str, Any]) -> int:
return old_format
def extract_cache_creation_tokens_detail(usage: dict[str, Any]) -> tuple[int, int, int]:
"""
提取缓存创建 tokens 细分（区分 5m 和 1h）
返回 (total, tokens_5m, tokens_1h) 三元组。
当无法区分时，tokens_5m 和 tokens_1h 均为 0，total 为合计值。
"""
# 1. 嵌套格式
cache_creation = usage.get("cache_creation")
if isinstance(cache_creation, dict) and (
"ephemeral_5m_input_tokens" in cache_creation
or "ephemeral_1h_input_tokens" in cache_creation
):
t5m = int(cache_creation.get("ephemeral_5m_input_tokens", 0))
t1h = int(cache_creation.get("ephemeral_1h_input_tokens", 0))
return t5m + t1h, t5m, t1h
# 2. 扁平新格式
if "claude_cache_creation_5_m_tokens" in usage or "claude_cache_creation_1_h_tokens" in usage:
t5m = int(usage.get("claude_cache_creation_5_m_tokens", 0))
t1h = int(usage.get("claude_cache_creation_1_h_tokens", 0))
return t5m + t1h, t5m, t1h
# 3. 旧格式：无法区分
old = int(usage.get("cache_creation_input_tokens", 0))
return old, 0, 0
def build_sse_headers(extra_headers: dict[str, str] | None = None) -> dict[str, str]:
"""
构建 SSE（text/event-stream）推荐响应头，用于减少代理缓冲带来的卡顿/成段输出。

55
src/api/handlers/claude/adapter.py
View File

@@ -31,14 +31,15 @@ class ClaudeCapabilityDetector:
request_body: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""
从 Claude 请求头检测能力需求
从 Claude 请求头和请求体检测能力需求
检测规则:
- anthropic-beta: context-1m-xxx -> context_1m: True
- 请求体中 cache_control.ttl = "1h" -> cache_1h: True
Args:
headers: 请求头字典
request_body: 请求体（Claude 不使用，保留用于接口统一）
request_body: 请求体（用于检测 cache_control.ttl）
"""
requirements: dict[str, bool] = {}
@@ -47,9 +48,59 @@ class ClaudeCapabilityDetector:
if beta_header and "context-1m" in beta_header.lower():
requirements["context_1m"] = True
# 从请求体检测 cache_1h
if request_body and _detect_cache_1h_in_body(request_body):
requirements["cache_1h"] = True
return requirements
def _has_cache_1h_ttl(block: dict[str, Any]) -> bool:
"""检查单个内容块是否包含 cache_control.ttl = '1h'"""
cache_control = block.get("cache_control")
if isinstance(cache_control, dict):
return cache_control.get("ttl") == "1h"
return False
def _detect_cache_1h_in_body(body: dict[str, Any]) -> bool:
"""
扫描 Claude 请求体，检测是否包含 cache_control.ttl = "1h"
检查位置：
- system[].cache_control.ttl
- messages[].content[].cache_control.ttl
- tools[].cache_control.ttl
"""
# 检查 system（数组格式）
system = body.get("system")
if isinstance(system, list):
for block in system:
if isinstance(block, dict) and _has_cache_1h_ttl(block):
return True
# 检查 messages
messages = body.get("messages")
if isinstance(messages, list):
for msg in messages:
if not isinstance(msg, dict):
continue
content = msg.get("content")
if isinstance(content, list):
for block in content:
if isinstance(block, dict) and _has_cache_1h_ttl(block):
return True
# 检查 tools
tools = body.get("tools")
if isinstance(tools, list):
for tool in tools:
if isinstance(tool, dict) and _has_cache_1h_ttl(tool):
return True
return False
@register_adapter
class ClaudeChatAdapter(ChatAdapterBase):
"""

7
src/api/handlers/claude/handler.py
View File

@@ -8,7 +8,7 @@ Claude Chat Handler - 基于通用 Chat Handler 基类的简化实现
from typing import Any
from src.api.handlers.base.chat_handler_base import ChatHandlerBase
from src.api.handlers.base.utils import extract_cache_creation_tokens
from src.api.handlers.base.utils import extract_cache_creation_tokens_detail
from src.core.api_format import ApiFamily, EndpointKind
@@ -103,12 +103,15 @@ class ClaudeChatHandler(ChatHandlerBase):
- 新格式：claude_cache_creation_5_m_tokens / claude_cache_creation_1_h_tokens
"""
usage = response.get("usage", {})
total, t5m, t1h = extract_cache_creation_tokens_detail(usage)
return {
"input_tokens": usage.get("input_tokens", 0),
"output_tokens": usage.get("output_tokens", 0),
"cache_creation_input_tokens": extract_cache_creation_tokens(usage),
"cache_creation_input_tokens": total,
"cache_read_input_tokens": usage.get("cache_read_input_tokens", 0),
"cache_creation_input_tokens_5m": t5m,
"cache_creation_input_tokens_1h": t1h,
}
def _normalize_response(self, response: dict[str, Any]) -> dict[str, Any]:

2
src/api/handlers/claude_cli/adapter.py
View File

@@ -46,7 +46,7 @@ class ClaudeCliAdapter(CliAdapterBase):
request_body: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""检测 Claude CLI 请求中隐含的能力需求"""
return ClaudeCapabilityDetector.detect_from_headers(headers)
return ClaudeCapabilityDetector.detect_from_headers(headers, request_body)
# =========================================================================
# Claude CLI 特定的计费逻辑

18
src/api/handlers/claude_cli/handler.py
View File

@@ -10,7 +10,7 @@ from src.api.handlers.base.cli_handler_base import (
CliMessageHandlerBase,
StreamContext,
)
from src.api.handlers.base.utils import extract_cache_creation_tokens
from src.api.handlers.base.utils import extract_cache_creation_tokens_detail
from src.core.api_format import ApiFamily, EndpointKind
@@ -114,9 +114,11 @@ class ClaudeCliMessageHandler(CliMessageHandlerBase):
if cache_read:
ctx.cached_tokens = cache_read
cache_creation = extract_cache_creation_tokens(usage)
if cache_creation:
ctx.cache_creation_tokens = cache_creation
total, t5m, t1h = extract_cache_creation_tokens_detail(usage)
if total:
ctx.cache_creation_tokens = total
ctx.cache_creation_tokens_5m = t5m
ctx.cache_creation_tokens_1h = t1h
# 处理文本增量
elif event_type == "content_block_delta":
@@ -140,9 +142,11 @@ class ClaudeCliMessageHandler(CliMessageHandlerBase):
ctx.cached_tokens = usage["cache_read_input_tokens"]
# 更新缓存创建 tokens
cache_creation = extract_cache_creation_tokens(usage)
if cache_creation > 0:
ctx.cache_creation_tokens = cache_creation
total, t5m, t1h = extract_cache_creation_tokens_detail(usage)
if total > 0:
ctx.cache_creation_tokens = total
ctx.cache_creation_tokens_5m = t5m
ctx.cache_creation_tokens_1h = t1h
# 检查是否结束
delta = data.get("delta", {})

29
src/api/handlers/gemini/adapter.py
View File

@@ -19,9 +19,30 @@ from src.core.api_format.enums import AuthMethod
from src.core.api_format.headers import BROWSER_FINGERPRINT_HEADERS
from src.core.logger import logger
from src.models.gemini import GeminiRequest
from src.services.gemini_files_mapping import extract_file_names_from_request
from src.services.provider.transport import redact_url_for_log
class GeminiCapabilityDetector:
"""Gemini API 能力检测器"""
@staticmethod
def detect_from_request(
headers: dict[str, str], # noqa: ARG004 - 预留
request_body: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""
从请求体检测 Gemini 能力需求
检测规则:
- fileData.fileUri -> gemini_files: True
"""
requirements: dict[str, bool] = {}
if request_body and extract_file_names_from_request(request_body):
requirements["gemini_files"] = True
return requirements
@register_adapter
class GeminiChatAdapter(ChatAdapterBase):
"""
@@ -62,11 +83,11 @@ class GeminiChatAdapter(ChatAdapterBase):
def detect_capability_requirements(
self,
headers: dict[str, str], # noqa: ARG002 - 预留
request_body: dict[str, Any] | None = None, # noqa: ARG002 - 预留
headers: dict[str, str],
request_body: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""Gemini API 无特殊能力要求"""
return {}
"""从请求体检测 Gemini 能力需求（fileData.fileUri -> gemini_files）"""
return GeminiCapabilityDetector.detect_from_request(headers, request_body)
def _merge_path_params(
self, original_request_body: dict[str, Any], path_params: dict[str, Any] # noqa: ARG002

10
src/api/handlers/gemini_cli/adapter.py
View File

@@ -13,7 +13,7 @@ from fastapi import Request
from src.api.handlers.base.cli_adapter_base import CliAdapterBase, register_cli_adapter
from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
from src.api.handlers.gemini.adapter import GeminiChatAdapter
from src.api.handlers.gemini.adapter import GeminiCapabilityDetector, GeminiChatAdapter
from src.config.settings import config
from src.core.api_format import ApiFamily, get_auth_handler
from src.core.api_format.enums import AuthMethod
@@ -53,6 +53,14 @@ class GeminiCliAdapter(CliAdapterBase):
handler = get_auth_handler(AuthMethod.GOOG_API_KEY)
return handler.extract_credentials(request)
def detect_capability_requirements(
self,
headers: dict[str, str],
request_body: dict[str, Any] | None = None,
) -> dict[str, bool]:
"""从请求体检测 Gemini 能力需求（fileData.fileUri -> gemini_files）"""
return GeminiCapabilityDetector.detect_from_request(headers, request_body)
def _merge_path_params(
self, original_request_body: dict[str, Any], path_params: dict[str, Any] # noqa: ARG002
) -> dict[str, Any]:

10
src/api/public/catalog.py
View File

@@ -312,18 +312,13 @@ class PublicProvidersAdapter(PublicApiAdapter):
providers = query.offset(self.skip).limit(self.limit).all()
result = []
for provider in providers:
models_count = (
db.query(Model)
.filter(Model.provider_id == provider.id, Model.global_model_id.isnot(None))
.count()
)
models_count = db.query(Model).filter(Model.provider_id == provider.id).count()
active_models_count = (
db.query(Model)
.filter(
and_(
Model.provider_id == provider.id,
Model.is_active.is_(True),
Model.global_model_id.isnot(None),
)
)
.count()
@@ -367,7 +362,6 @@ class PublicModelsAdapter(PublicApiAdapter):
and_(
Model.is_active.is_(True),
Provider.is_active.is_(True),
Model.global_model_id.isnot(None),
)
)
)
@@ -424,7 +418,6 @@ class PublicStatsAdapter(PublicApiAdapter):
and_(
Model.is_active.is_(True),
Provider.is_active.is_(True),
Model.global_model_id.isnot(None),
)
)
.count()
@@ -462,7 +455,6 @@ class PublicSearchModelsAdapter(PublicApiAdapter):
and_(
Model.is_active.is_(True),
Provider.is_active.is_(True),
Model.global_model_id.isnot(None),
)
)
)

19
src/api/user_me/routes.py
View File

@@ -32,7 +32,15 @@ from src.models.api import (
UpdatePreferencesRequest,
UpdateProfileRequest,
)
from src.models.database import ApiKey, GlobalModel, Model, Provider, Usage, User
from src.models.database import (
ApiKey,
GlobalModel,
Model,
Provider,
Usage,
User,
UserModelUsageCount,
)
from src.services.system.time_range import TimeRangeParams
from src.services.usage.service import UsageService
from src.services.user.apikey import ApiKeyService
@@ -1203,6 +1211,14 @@ class ListAvailableModelsAdapter(AuthenticatedApiAdapter):
.all()
)
# 查询当前用户的每模型调用次数
user_usage_rows = (
db.query(UserModelUsageCount.model, UserModelUsageCount.usage_count)
.filter(UserModelUsageCount.user_id == user.id)
.all()
)
user_usage_map: dict[str, int] = {row.model: row.usage_count for row in user_usage_rows}
# 转换为响应格式（复用 PublicGlobalModelResponse schema）
model_responses = [
PublicGlobalModelResponse(
@@ -1214,6 +1230,7 @@ class ListAvailableModelsAdapter(AuthenticatedApiAdapter):
default_tiered_pricing=gm.default_tiered_pricing,
supported_capabilities=gm.supported_capabilities,
config=gm.config,
usage_count=user_usage_map.get(gm.name, 0),
)
for gm in models
]

68
src/core/key_capabilities.py
View File

@@ -100,17 +100,15 @@ def check_capability_match(
匹配逻辑：
1. EXCLUSIVE（互斥）能力：
- 请求需要且 Key 有 → 通过
- 请求需要但 Key 没有 → 拒绝
- 请求不需要但 Key 有 → 拒绝（避免浪费高价资源）
- 请求不需要且 Key 没有 → 通过
- 请求未声明但 Key 有 → 拒绝（关键：未声明等同于不需要）
- 请求需要且 Key 有 -> 通过
- 请求需要但 Key 没有 -> 拒绝
- 请求不需要但 Key 有 -> 拒绝（避免浪费高价资源）
- 请求不需要且 Key 没有 -> 通过
- 请求未声明但 Key 有 -> 拒绝（关键：未声明等同于不需要）
2. COMPATIBLE（兼容）能力：
- 请求需要且 Key 有 → 通过
- 请求需要但 Key 没有 → 拒绝
- 请求不需要/未声明且 Key 有 → 通过（无额外成本，不浪费）
- 请求不需要/未声明且 Key 没有 → 通过
- 不做硬过滤，交由排序阶段通过 compute_capability_score() 处理
- 有能力的 Key 优先排序，没有的也不被排除
Args:
key_capabilities: Key 拥有的能力 {"cache_1h": True, ...}
@@ -136,9 +134,7 @@ def check_capability_match(
if not is_required and key_has_cap:
return False, f"不需要{cap_def.display_name}(避免浪费高价资源)"
elif cap_def.match_mode == CapabilityMatchMode.COMPATIBLE:
if is_required and not key_has_cap:
return False, f"需要{cap_def.display_name}但 Key 不支持"
# COMPATIBLE: 不做硬过滤
# 第二步：检查 Key 拥有的 EXCLUSIVE 能力是否被请求需要
# 如果 Key 有某个 EXCLUSIVE 能力，但请求没有声明需要，应该跳过这个 Key
@@ -158,6 +154,40 @@ def check_capability_match(
return True, None
def compute_capability_score(
key_capabilities: dict[str, bool] | None,
requirements: dict[str, bool] | None,
) -> int:
"""
计算 COMPATIBLE 能力不匹配数量
返回 0 表示完全匹配（或无 COMPATIBLE 需求），正数表示有 N 个 COMPATIBLE 能力不满足。
用于候选排序：得分越低越优先。
Args:
key_capabilities: Key 拥有的能力
requirements: 请求需要的能力
Returns:
不满足的 COMPATIBLE 能力数量
"""
key_caps = key_capabilities or {}
reqs = requirements or {}
miss_count = 0
for cap_name, is_required in reqs.items():
if not is_required:
continue
cap_def = _capabilities.get(cap_name)
if not cap_def:
continue
if cap_def.match_mode == CapabilityMatchMode.COMPATIBLE:
if not key_caps.get(cap_name, False):
miss_count += 1
return miss_count
def _match_error_patterns(error_msg: str, patterns: list[str]) -> bool:
"""检查错误信息是否匹配模式（所有关键词都要出现）"""
if not patterns:
@@ -220,20 +250,14 @@ class _CapabilityDefinitionsProxy:
CAPABILITY_DEFINITIONS = _CapabilityDefinitionsProxy()
# ============ 兼容旧的插件基类（逐步废弃） ============
CapabilityPlugin = CapabilityDefinition # 类型别名，兼容旧代码
# ============ 注册内置能力 ============
register_capability(
name="cache_1h",
display_name="1 小时缓存",
description="使用 1 小时缓存 TTL（价格更高，适合长对话）",
match_mode=CapabilityMatchMode.EXCLUSIVE,
config_mode=CapabilityConfigMode.USER_CONFIGURABLE,
match_mode=CapabilityMatchMode.COMPATIBLE,
config_mode=CapabilityConfigMode.REQUEST_PARAM,
short_name="1h缓存",
)
@@ -251,7 +275,7 @@ register_capability(
name="gemini_files",
display_name="Gemini 文件 API",
description="支持 Gemini Files API（文件上传/管理），仅 Google 官方 API 支持",
match_mode=CapabilityMatchMode.COMPATIBLE,
config_mode=CapabilityConfigMode.USER_CONFIGURABLE,
match_mode=CapabilityMatchMode.EXCLUSIVE,
config_mode=CapabilityConfigMode.REQUEST_PARAM,
short_name="文件API",
)

6
src/models/api.py
View File

@@ -577,7 +577,7 @@ class ModelResponse(BaseModel):
id: str
provider_id: str
global_model_id: str | None
global_model_id: str
provider_model_name: str
provider_model_mappings: list[dict] | None = None
@@ -612,7 +612,7 @@ class ModelResponse(BaseModel):
created_at: datetime
updated_at: datetime
# 关联的 GlobalModel 信息（如果有）
# 关联的 GlobalModel 信息
global_model_name: str | None = None
global_model_display_name: str | None = None
@@ -744,6 +744,8 @@ class PublicGlobalModelResponse(BaseModel):
supported_capabilities: list[str] | None = None
# 模型配置（JSON）
config: dict | None = None
# 调用次数
usage_count: int = 0
class PublicGlobalModelListResponse(BaseModel):

35
src/models/database.py
View File

@@ -1001,8 +1001,8 @@ class GlobalModel(ExportMixin, Base):
# "cache_creation_price_per_1m": 3.75, # 可选
# "cache_read_price_per_1m": 0.30, # 可选
# "cache_ttl_pricing": [ # 可选：按缓存时长分价格
# {"ttl_minutes": 5, "cache_read_price_per_1m": 0.30},
# {"ttl_minutes": 60, "cache_read_price_per_1m": 0.50}
# {"ttl_minutes": 5, "cache_creation_price_per_1m": 3.75, "cache_read_price_per_1m": 0.30},
# {"ttl_minutes": 60, "cache_creation_price_per_1m": 6.00, "cache_read_price_per_1m": 0.50}
# ]
# },
# {"up_to": null, "input_price_per_1m": 1.25, ...}
@@ -2700,5 +2700,36 @@ class GeminiFileMapping(Base):
)
class UserModelUsageCount(Base):
"""用户-模型维度调用次数计数器
每个用户对每个模型维护一个原子递增的计数器，
避免从 Usage 表聚合查询，查询性能 O(N) 其中 N 是用户使用过的模型数。
"""
__tablename__ = "user_model_usage_counts"
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
user_id = Column(String(36), ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
model = Column(String(100), nullable=False)
usage_count = Column(Integer, default=0, nullable=False)
created_at = Column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), nullable=False
)
updated_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
onupdate=lambda: datetime.now(timezone.utc),
nullable=False,
)
__table_args__ = (
UniqueConstraint("user_id", "model", name="uq_user_model_usage_count"),
Index("idx_user_model_usage_user", "user_id"),
Index("idx_user_model_usage_model", "model"),
)
# 导入扩展的数据库模型
from .database_extensions import ApiKeyProviderMapping, ProviderUsageTracking

12
src/models/model.py
View File

@@ -69,8 +69,8 @@ class GlobalModel(ExportMixin, Base):
# "cache_creation_price_per_1m": 3.75, # 可选
# "cache_read_price_per_1m": 0.30, # 可选
# "cache_ttl_pricing": [ # 可选：按缓存时长分价格
# {"ttl_minutes": 5, "cache_read_price_per_1m": 0.30},
# {"ttl_minutes": 60, "cache_read_price_per_1m": 0.50}
# {"ttl_minutes": 5, "cache_creation_price_per_1m": 3.75, "cache_read_price_per_1m": 0.30},
# {"ttl_minutes": 60, "cache_creation_price_per_1m": 6.00, "cache_read_price_per_1m": 0.50}
# ]
# },
# {"up_to": null, "input_price_per_1m": 1.25, ...}
@@ -132,9 +132,7 @@ class Model(ExportMixin, Base):
设计原则:
- Model 表示 Provider 对某个模型的具体实现
- global_model_id 可为空：
- 为空时：模型尚未关联到 GlobalModel，不参与路由
- 不为空时：模型已关联 GlobalModel，参与路由
- global_model_id 必填，必须关联到一个 GlobalModel
- provider_model_name 是 Provider 侧的实际模型名称 (可能与 GlobalModel.name 不同)
- 价格和能力配置可为空，为空时使用 GlobalModel 的默认值
"""
@@ -154,8 +152,8 @@ class Model(ExportMixin, Base):
id = Column(String(36), primary_key=True, default=lambda: str(uuid.uuid4()), index=True)
provider_id = Column(String(36), ForeignKey("providers.id"), nullable=False)
# 可为空：NULL 表示未关联，不参与路由；非 NULL 表示已关联，参与路由
global_model_id = Column(String(36), ForeignKey("global_models.id"), nullable=True, index=True)
# 必须关联一个 GlobalModel
global_model_id = Column(String(36), ForeignKey("global_models.id"), nullable=False, index=True)
# Provider 映射配置
provider_model_name = Column(String(200), nullable=False) # Provider 侧的主模型名称

7
src/models/pydantic_models.py
View File

@@ -19,6 +19,9 @@ class CacheTTLPricing(BaseModel):
cache_creation_price_per_1m: float = Field(
..., ge=0, description="该时长的缓存创建价格/M tokens"
)
cache_read_price_per_1m: float | None = Field(
None, ge=0, description="该时长的缓存读取价格/M tokens"
)
class PricingTier(BaseModel):
@@ -313,8 +316,8 @@ class ImportFromUpstreamSuccessItem(BaseModel):
model_id: str = Field(..., description="上游模型 ID")
provider_model_id: str = Field(..., description="Provider Model ID")
global_model_id: str | None = Field("", description="GlobalModel ID（如果已关联）")
global_model_name: str | None = Field("", description="GlobalModel 名称（如果已关联）")
global_model_id: str = Field(..., description="GlobalModel ID")
global_model_name: str = Field(..., description="GlobalModel 名称")
created_global_model: bool = Field(
False, description="是否新创建了 GlobalModel（始终为 false）"
)

2
src/models/usage.py
View File

@@ -71,6 +71,8 @@ class Usage(Base):
# 缓存相关 tokens (for Claude models)
cache_creation_input_tokens = Column(Integer, default=0)
cache_read_input_tokens = Column(Integer, default=0)
cache_creation_input_tokens_5m = Column(Integer, default=0) # 5min TTL 缓存创建
cache_creation_input_tokens_1h = Column(Integer, default=0) # 1h TTL 缓存创建
# 成本计算
input_cost_usd = Column(Float, default=0.0)

9
src/services/billing/default_rules.py
View File

@@ -231,7 +231,14 @@ class DefaultBillingRuleGenerator:
"source": "tiered",
"tier_key": tier_key,
"allow_zero": True,
"tiers": _tiers_for("cache_creation_price_per_1m", default_multiplier=1.25),
# TTL override supported when dims include cache_ttl_minutes
"ttl_key": "cache_ttl_minutes",
"ttl_value_key": "cache_creation_price_per_1m",
"tiers": _tiers_for(
"cache_creation_price_per_1m",
default_multiplier=1.25,
include_cache_ttl_pricing=True,
),
"default": base_cache_creation_price,
}
dimension_mappings["cache_read_price_per_1m"] = {

4
src/services/model/global_model.py
View File

@@ -545,8 +545,8 @@ class GlobalModelService:
models_to_delete: list[Model] = []
for model in models:
# 跳过没有关联 GlobalModel 的
if not model.global_model_id or not model.global_model:
# 跳过 global_model 关系未加载的
if not model.global_model:
continue
global_model = cast(GlobalModel, model.global_model)

21
src/services/model/service.py
View File

@@ -65,15 +65,14 @@ class ModelService:
db.commit()
db.refresh(model)
# 显式加载 global_model 关系
if model.global_model_id:
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import joinedload
model = (
db.query(Model)
.options(joinedload(Model.global_model))
.filter(Model.id == model.id)
.first()
)
model = (
db.query(Model)
.options(joinedload(Model.global_model))
.filter(Model.id == model.id)
.first()
)
logger.info(
f"创建模型成功: provider={provider.name}, model={model.provider_model_name}, global_model_id={model.global_model_id}"
@@ -226,7 +225,7 @@ class ModelService:
)
# 清除内存缓存（ModelMapperMiddleware 实例）
if model.provider_id and model.global_model_id:
if model.provider_id:
cache_service = get_cache_invalidation_service()
cache_service.on_model_changed(model.provider_id, model.global_model_id)
@@ -297,7 +296,7 @@ class ModelService:
)
# 清除内存缓存
if cache_info["provider_id"] and cache_info["global_model_id"]:
if cache_info["provider_id"]:
cache_service = get_cache_invalidation_service()
cache_service.on_model_changed(
cache_info["provider_id"], cache_info["global_model_id"]
@@ -338,7 +337,7 @@ class ModelService:
)
# 清除内存缓存（ModelMapperMiddleware 实例）
if model.provider_id and model.global_model_id:
if model.provider_id:
cache_service = get_cache_invalidation_service()
cache_service.on_model_changed(model.provider_id, model.global_model_id)

20
src/services/scheduling/candidate_builder.py
View File

@@ -19,7 +19,12 @@ from sqlalchemy.orm import Session, selectinload
from src.core.api_format.conversion.compatibility import is_format_compatible
from src.core.api_format.enums import EndpointKind
from src.core.api_format.signature import make_signature_key, parse_signature_key
from src.core.key_capabilities import check_capability_match
from src.core.key_capabilities import (
CapabilityMatchMode,
check_capability_match,
compute_capability_score,
get_capability,
)
from src.core.logger import logger
from src.core.model_permissions import check_model_allowed_with_mappings
from src.models.database import (
@@ -219,9 +224,13 @@ class CandidateBuilder:
# 检查模型是否支持所需的能力（在 Provider 级别检查，而不是 Key 级别）
# 只有当 model_supported_capabilities 非空时才进行检查
# 空列表意味着模型没有配置能力限制，默认支持所有能力
# COMPATIBLE 能力跳过模型级硬过滤（交由排序阶段处理）
if capability_requirements and model_supported_capabilities:
for cap_name, is_required in capability_requirements.items():
if is_required and cap_name not in model_supported_capabilities:
cap_def = get_capability(cap_name)
if cap_def and cap_def.match_mode == CapabilityMatchMode.COMPATIBLE:
continue
return (
False,
f"模型 {model_name} 不支持能力: {cap_name}",
@@ -617,6 +626,15 @@ class CandidateBuilder:
needs_conversion=needs_conversion,
provider_api_format=str(endpoint_format_str or ""),
output_limit=output_limit,
# is_skipped 候选不参与排序，miss_count 无意义，置 0 避免干扰
capability_miss_count=(
compute_capability_score(
key.capabilities or {},
capability_requirements,
)
if is_available
else 0
),
)
if needs_conversion:

52
src/services/scheduling/candidate_sorter.py
View File

@@ -18,6 +18,8 @@ from src.services.scheduling.utils import affinity_hash
from src.services.system.config import SystemConfigService
if TYPE_CHECKING:
from collections.abc import Callable
from sqlalchemy.orm import Session
from src.models.database import ProviderAPIKey
@@ -30,6 +32,28 @@ class CandidateSorter:
def __init__(self, config: SchedulingConfig) -> None:
self._config = config
@staticmethod
def _split_by_capability_match(
candidates: list[ProviderCandidate],
) -> tuple[list[ProviderCandidate], list[ProviderCandidate]]:
"""按 capability_miss_count 分组：完全匹配(0)在前，部分匹配(>0)在后"""
full_match = [c for c in candidates if c.capability_miss_count == 0]
partial_match = [c for c in candidates if c.capability_miss_count > 0]
return full_match, partial_match
def _with_capability_split(
self,
candidates: list[ProviderCandidate],
sort_fn: Callable[..., list[ProviderCandidate]],
*args: object,
**kwargs: object,
) -> list[ProviderCandidate]:
"""通用包装：先按 capability_miss_count 分组，再分别排序后合并"""
if not candidates:
return candidates
full_match, partial_match = self._split_by_capability_match(candidates)
return sort_fn(full_match, *args, **kwargs) + sort_fn(partial_match, *args, **kwargs)
def _apply_priority_mode_sort(
self,
candidates: list[ProviderCandidate],
@@ -40,7 +64,8 @@ class CandidateSorter:
"""
根据优先级模式对候选列表排序（数字越小越优先）
排序规则（受 keep_priority_on_conversion 配置影响）：
排序规则：
0. 按 capability_miss_count 分组：完全匹配(0)在前，部分匹配(>0)在后
1. 如果全局配置 keep_priority_on_conversion=True，所有候选保持原优先级
2. 否则，按 needs_conversion 和 provider.keep_priority_on_conversion 分组：
- 保持优先级的候选（exact 或 provider.keep_priority_on_conversion=True）按原优先级排序
@@ -52,6 +77,21 @@ class CandidateSorter:
if not candidates:
return candidates
return self._with_capability_split(
candidates, self._apply_priority_mode_sort_inner, db, affinity_key, api_format
)
def _apply_priority_mode_sort_inner(
self,
candidates: list[ProviderCandidate],
db: Session,
affinity_key: str | None = None,
api_format: str | None = None,
) -> list[ProviderCandidate]:
"""优先级模式排序的内部实现（不含 capability_miss_count 分组）"""
if not candidates:
return candidates
# 全局配置：如果开启，所有候选保持原优先级
global_keep_priority = SystemConfigService.is_keep_priority_on_conversion(db)
@@ -159,6 +199,7 @@ class CandidateSorter:
负载均衡模式：同优先级内随机轮换
排序逻辑：
0. 按 capability_miss_count 分组：完全匹配(0)在前，部分匹配(>0)在后
1. 按优先级分组（provider_priority, internal_priority 或 global_priority_by_format）
2. 同优先级组内随机打乱
3. 不考虑缓存亲和性
@@ -166,6 +207,15 @@ class CandidateSorter:
if not candidates:
return candidates
return self._with_capability_split(candidates, self._apply_load_balance_inner, api_format)
def _apply_load_balance_inner(
self, candidates: list[ProviderCandidate], api_format: str | None = None
) -> list[ProviderCandidate]:
"""负载均衡排序的内部实现（不含 capability_miss_count 分组）"""
if not candidates:
return candidates
priority_groups: dict[tuple, list[ProviderCandidate]] = defaultdict(list)
# 根据优先级模式选择分组方式

1
src/services/scheduling/schemas.py
View File

@@ -29,6 +29,7 @@ class ProviderCandidate:
needs_conversion: bool = False # 是否需要格式转换
provider_api_format: str = "" # Provider 端点实际格式（用于健康度/熔断 bucket）
output_limit: int | None = None # GlobalModel 配置的模型输出上限
capability_miss_count: int = 0 # COMPATIBLE 能力不匹配数（0=完全匹配，用于排序）
def _stable_order_key(self) -> tuple[int, int, str, str, str]:
"""

2
src/services/usage/_billing_integration.py
View File

@@ -133,6 +133,8 @@ class UsageBillingIntegrationMixin:
output_tokens=params.output_tokens,
cache_creation_input_tokens=params.cache_creation_input_tokens,
cache_read_input_tokens=params.cache_read_input_tokens,
cache_creation_input_tokens_5m=params.cache_creation_input_tokens_5m,
cache_creation_input_tokens_1h=params.cache_creation_input_tokens_1h,
request_type=params.request_type,
api_format=params.api_format,
api_family=params.api_family,

10
src/services/usage/_recording_helpers.py
View File

@@ -55,6 +55,8 @@ def build_usage_params(
output_tokens: int,
cache_creation_input_tokens: int,
cache_read_input_tokens: int,
cache_creation_input_tokens_5m: int = 0,
cache_creation_input_tokens_1h: int = 0,
request_type: str,
api_format: str | None,
api_family: str | None = None,
@@ -201,6 +203,8 @@ def build_usage_params(
"total_tokens": input_tokens + output_tokens,
"cache_creation_input_tokens": cache_creation_input_tokens,
"cache_read_input_tokens": cache_read_input_tokens,
"cache_creation_input_tokens_5m": cache_creation_input_tokens_5m,
"cache_creation_input_tokens_1h": cache_creation_input_tokens_1h,
"input_cost_usd": input_cost,
"output_cost_usd": output_cost,
"cache_cost_usd": cache_cost,
@@ -292,6 +296,12 @@ def update_existing_usage(
existing_usage.total_tokens = usage_params["total_tokens"]
existing_usage.cache_creation_input_tokens = usage_params["cache_creation_input_tokens"]
existing_usage.cache_read_input_tokens = usage_params["cache_read_input_tokens"]
existing_usage.cache_creation_input_tokens_5m = usage_params.get(
"cache_creation_input_tokens_5m", 0
)
existing_usage.cache_creation_input_tokens_1h = usage_params.get(
"cache_creation_input_tokens_1h", 0
)
existing_usage.input_cost_usd = usage_params["input_cost_usd"]
existing_usage.output_cost_usd = usage_params["output_cost_usd"]
existing_usage.cache_cost_usd = usage_params["cache_cost_usd"]

2
src/services/usage/_types.py
View File

@@ -49,6 +49,8 @@ class UsageRecordParams:
cache_ttl_minutes: int | None
use_tiered_pricing: bool
target_model: str | None
cache_creation_input_tokens_5m: int = 0
cache_creation_input_tokens_1h: int = 0
def __post_init__(self) -> None:
"""验证关键字段，确保数据完整性"""

2
src/services/usage/consumer_streams.py
View File

@@ -68,6 +68,8 @@ def _event_to_record(event: UsageEvent) -> dict[str, Any]:
"output_tokens": data.get("output_tokens") or 0,
"cache_creation_input_tokens": data.get("cache_creation_input_tokens") or 0,
"cache_read_input_tokens": data.get("cache_read_input_tokens") or 0,
"cache_creation_input_tokens_5m": data.get("cache_creation_input_tokens_5m") or 0,
"cache_creation_input_tokens_1h": data.get("cache_creation_input_tokens_1h") or 0,
"request_type": data.get("request_type") or "chat",
"api_format": data.get("api_format"),
"api_family": data.get("api_family"),

87
src/services/usage/recording.py
View File

@@ -7,7 +7,7 @@ from typing import Any
from sqlalchemy.orm import Session
from src.core.logger import logger
from src.models.database import ApiKey, Provider, Usage, User
from src.models.database import ApiKey, Provider, Usage, User, UserModelUsageCount
from src.services.usage._billing_integration import UsageBillingIntegrationMixin
from src.services.usage._recording_helpers import (
METADATA_KEEP_KEYS,
@@ -44,6 +44,31 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
"""更新已存在的 Usage 记录（委托到模块级函数）"""
update_existing_usage(existing_usage, usage_params, target_model)
@staticmethod
def _increment_user_model_usage(
db: Session, user: User | None, model: str, count: int = 1
) -> None:
"""原子递增用户-模型调用次数计数器"""
if user is None:
return
from sqlalchemy import func as sa_func
from sqlalchemy.dialects.postgresql import insert as pg_insert
stmt = pg_insert(UserModelUsageCount).values(
id=str(uuid.uuid4()),
user_id=user.id,
model=model,
usage_count=count,
)
stmt = stmt.on_conflict_do_update(
constraint="uq_user_model_usage_count",
set_={
"usage_count": UserModelUsageCount.usage_count + count,
"updated_at": sa_func.now(),
},
)
db.execute(stmt)
@classmethod
def _sanitize_request_metadata(cls, metadata: dict[str, Any]) -> dict[str, Any]:
"""元数据清理（委托到模块级函数）"""
@@ -65,6 +90,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens: int,
cache_creation_input_tokens: int = 0,
cache_read_input_tokens: int = 0,
cache_creation_input_tokens_5m: int = 0,
cache_creation_input_tokens_1h: int = 0,
request_type: str = "chat",
api_format: str | None = None,
api_family: str | None = None,
@@ -116,6 +143,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens,
cache_creation_input_tokens_5m=cache_creation_input_tokens_5m,
cache_creation_input_tokens_1h=cache_creation_input_tokens_1h,
request_type=request_type,
api_format=api_format,
api_family=api_family,
@@ -162,6 +191,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
.values(usage_count=GlobalModel.usage_count + 1)
)
# 更新用户-模型调用次数计数器
cls._increment_user_model_usage(db, user, model)
# 更新 Provider 月度使用量（原子操作）
if provider_id:
actual_total_cost = usage_params["actual_total_cost_usd"]
@@ -191,6 +223,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens: int,
cache_creation_input_tokens: int = 0,
cache_read_input_tokens: int = 0,
cache_creation_input_tokens_5m: int = 0,
cache_creation_input_tokens_1h: int = 0,
request_type: str = "chat",
api_format: str | None = None,
api_family: str | None = None,
@@ -244,6 +278,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens,
cache_creation_input_tokens_5m=cache_creation_input_tokens_5m,
cache_creation_input_tokens_1h=cache_creation_input_tokens_1h,
request_type=request_type,
api_format=api_format,
api_family=api_family,
@@ -347,6 +383,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
.values(usage_count=GlobalModel.usage_count + 1)
)
# 更新用户-模型调用次数计数器
cls._increment_user_model_usage(db, user, model)
# 更新 Provider 月度使用量
if provider_id:
actual_total_cost = usage_params["actual_total_cost_usd"]
@@ -387,6 +426,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens: int = 0,
cache_creation_input_tokens: int = 0,
cache_read_input_tokens: int = 0,
cache_creation_input_tokens_5m: int = 0,
cache_creation_input_tokens_1h: int = 0,
api_format: str | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
@@ -451,6 +492,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens,
cache_creation_input_tokens_5m=cache_creation_input_tokens_5m,
cache_creation_input_tokens_1h=cache_creation_input_tokens_1h,
request_type=request_type,
api_format=api_format,
api_family=api_family,
@@ -588,6 +631,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
.values(usage_count=GlobalModel.usage_count + 1)
)
# 更新用户-模型调用次数计数器
cls._increment_user_model_usage(db, user, model)
# 更新 Provider 月度使用量（使用 actual_total_cost）
if provider_id:
actual_total_cost = usage_params["actual_total_cost_usd"]
@@ -714,6 +760,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
lambda: {"requests": 0, "cost": 0.0, "is_standalone": False}
)
model_counts: dict[str, int] = defaultdict(int) # model -> count
user_model_counts: dict[tuple[str, str], int] = defaultdict(
int
) # (user_id, model) -> count
provider_costs: dict[str, float] = defaultdict(float) # provider_id -> cost
# 合并所有需要处理的记录（用于预取 user/api_key）
@@ -754,6 +803,12 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
output_tokens=int(record.get("output_tokens") or 0),
cache_creation_input_tokens=int(record.get("cache_creation_input_tokens") or 0),
cache_read_input_tokens=int(record.get("cache_read_input_tokens") or 0),
cache_creation_input_tokens_5m=int(
record.get("cache_creation_input_tokens_5m") or 0
),
cache_creation_input_tokens_1h=int(
record.get("cache_creation_input_tokens_1h") or 0
),
request_type=record.get("request_type") or "chat",
api_format=record.get("api_format"),
api_family=record.get("api_family"),
@@ -850,6 +905,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
# 聚合统计
model_name = record.get("model") or "unknown"
model_counts[model_name] += 1
if user:
user_model_counts[(str(user.id), model_name)] += 1
provider_id = record.get("provider_id")
if provider_id:
@@ -898,6 +955,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
# 聚合统计
model_name = record.get("model") or "unknown"
model_counts[model_name] += 1
if user:
user_model_counts[(str(user.id), model_name)] += 1
provider_id = record.get("provider_id")
if provider_id:
@@ -959,6 +1018,30 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
.values(usage_count=GlobalModel.usage_count + count)
)
# 批量更新用户-模型调用次数计数器
from sqlalchemy import func as sql_func
from sqlalchemy.dialects.postgresql import insert as pg_insert
if user_model_counts:
rows = [
{
"id": str(uuid.uuid4()),
"user_id": uid,
"model": model_name,
"usage_count": count,
}
for (uid, model_name), count in user_model_counts.items()
]
stmt = pg_insert(UserModelUsageCount).values(rows)
stmt = stmt.on_conflict_do_update(
constraint="uq_user_model_usage_count",
set_={
"usage_count": UserModelUsageCount.usage_count + stmt.excluded.usage_count,
"updated_at": sql_func.now(),
},
)
db.execute(stmt)
# 批量更新 Provider 月度使用量
for provider_id, cost in provider_costs.items():
if cost > 0:
@@ -969,8 +1052,6 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
)
# 批量更新用户使用量
from sqlalchemy import func as sql_func
for user_id, cost in user_costs.items():
if cost > 0:
db.execute(

24
src/services/usage/stream.py
View File

@@ -100,6 +100,8 @@ class StreamUsageTracker:
self.output_tokens = 0
self.cache_creation_input_tokens = 0
self.cache_read_input_tokens = 0
self.cache_creation_input_tokens_5m = 0
self.cache_creation_input_tokens_1h = 0
self.accumulated_content = ""
# 完整响应跟踪（仅用于内部统计，不记录到数据库）
@@ -477,6 +479,8 @@ class StreamUsageTracker:
"""
import time
from src.api.handlers.base.utils import extract_cache_creation_tokens_detail
self.start_time = time.time()
self.request_data = request_data # 保存请求数据
@@ -545,21 +549,17 @@ class StreamUsageTracker:
# 如果响应中包含准确的usage信息，使用它
self.input_tokens = usage.get("input_tokens", self.input_tokens)
self.output_tokens = usage.get("output_tokens", self.output_tokens)
self.cache_creation_input_tokens = usage.get(
"cache_creation_input_tokens", self.cache_creation_input_tokens
)
self.cache_read_input_tokens = usage.get(
"cache_read_input_tokens", self.cache_read_input_tokens
)
# 处理新的cache_creation格式
if "cache_creation" in usage:
cache_creation_data = usage.get("cache_creation", {})
# 如果没有cache_creation_input_tokens，尝试从cache_creation中获取
if not self.cache_creation_input_tokens:
self.cache_creation_input_tokens = cache_creation_data.get(
"ephemeral_5m_input_tokens", 0
) + cache_creation_data.get("ephemeral_1h_input_tokens", 0)
# 统一提取 cache_creation tokens（新格式优先于旧格式）
total, t5m, t1h = extract_cache_creation_tokens_detail(usage)
if total:
self.cache_creation_input_tokens = total
if t5m or t1h:
self.cache_creation_input_tokens_5m = t5m
self.cache_creation_input_tokens_1h = t1h
finally:
# 流结束后记录使用量
@@ -768,6 +768,8 @@ class StreamUsageTracker:
output_tokens=self.output_tokens,
cache_creation_input_tokens=self.cache_creation_input_tokens,
cache_read_input_tokens=self.cache_read_input_tokens,
cache_creation_input_tokens_5m=self.cache_creation_input_tokens_5m,
cache_creation_input_tokens_1h=self.cache_creation_input_tokens_1h,
request_type="chat",
api_format=self.api_format,
api_family=self.api_family,

12
src/services/usage/telemetry.py
View File

@@ -70,6 +70,8 @@ class MessageTelemetry:
client_response_headers: dict[str, Any] | None = None,
cache_creation_tokens: int = 0,
cache_read_tokens: int = 0,
cache_creation_tokens_5m: int = 0,
cache_creation_tokens_1h: int = 0,
is_stream: bool = False,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
@@ -111,6 +113,8 @@ class MessageTelemetry:
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_tokens,
cache_read_input_tokens=cache_read_tokens,
cache_creation_input_tokens_5m=cache_creation_tokens_5m,
cache_creation_input_tokens_1h=cache_creation_tokens_1h,
request_type="chat",
api_format=api_format,
api_family=api_family,
@@ -181,6 +185,8 @@ class MessageTelemetry:
output_tokens: int = 0,
cache_creation_tokens: int = 0,
cache_read_tokens: int = 0,
cache_creation_tokens_5m: int = 0,
cache_creation_tokens_1h: int = 0,
response_body: dict[str, Any] | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
@@ -227,6 +233,8 @@ class MessageTelemetry:
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_tokens,
cache_read_input_tokens=cache_read_tokens,
cache_creation_input_tokens_5m=cache_creation_tokens_5m,
cache_creation_input_tokens_1h=cache_creation_tokens_1h,
request_type="chat",
api_format=api_format,
api_family=api_family,
@@ -276,6 +284,8 @@ class MessageTelemetry:
output_tokens: int = 0,
cache_creation_tokens: int = 0,
cache_read_tokens: int = 0,
cache_creation_tokens_5m: int = 0,
cache_creation_tokens_1h: int = 0,
response_body: dict[str, Any] | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
@@ -308,6 +318,8 @@ class MessageTelemetry:
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_tokens,
cache_read_input_tokens=cache_read_tokens,
cache_creation_input_tokens_5m=cache_creation_tokens_5m,
cache_creation_input_tokens_1h=cache_creation_tokens_1h,
request_type="chat",
api_format=api_format,
api_family=api_family,

8
src/services/usage/telemetry_writer.py
View File

@@ -206,6 +206,14 @@ class QueueTelemetryWriter(TelemetryWriter):
if cache_read:
data["cache_read_input_tokens"] = cache_read
# 缓存 5m/1h 细分
cache_creation_5m = kwargs.get("cache_creation_tokens_5m", 0)
cache_creation_1h = kwargs.get("cache_creation_tokens_1h", 0)
if cache_creation_5m:
data["cache_creation_input_tokens_5m"] = cache_creation_5m
if cache_creation_1h:
data["cache_creation_input_tokens_1h"] = cache_creation_1h
# 时间指标
if kwargs.get("response_time_ms") is not None:
data["response_time_ms"] = kwargs["response_time_ms"]