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

refactor: 提取 HandlerAdapterBase 基类，新增 api_family/endpoint_kind 结构化维度

Browse Source 
- 从 ChatAdapterBase 和 CliAdapterBase 提取公共逻辑到 HandlerAdapterBase，
  消除头部处理、异常处理、计费策略等重复代码
- Usage 记录链路全程透传 api_family / endpoint_kind / provider_api_family /
  provider_endpoint_kind 四个结构化字段，替代从 api_format 字符串解析
- 删除冗余的 ClaudeCliNormalizer、GeminiCliNormalizer、ClaudeCliResponseParser、
  GeminiCliResponseParser，改用 data_format_id 回退机制自动复用 Chat 版本
- build_endpoint_url 签名统一扩展 request_data / model_name 参数
- 新增 Alembic 迁移，含历史数据回填
This commit is contained in:
fawney19
2026-02-21 13:17:11 +08:00
parent 54988916e3
commit 0fe5346f4d
36 changed files with 967 additions and 1310 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -114,6 +114,8 @@ class BaseMessageHandler:
allowed_api_formats: list[str] | None = None,
adapter_detector: AdapterDetectorType | None = None,
perf_metrics: dict[str, Any] | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
) -> None:
self.db = db
self.user = user
@@ -127,6 +129,9 @@ class BaseMessageHandler:
self.primary_api_format = normalize_endpoint_signature(self.allowed_api_formats[0])
self.adapter_detector = adapter_detector
self.perf_metrics = perf_metrics
# 结构化格式维度（从 Adapter 层透传）
self.api_family = api_family
self.endpoint_kind = endpoint_kind
redis_client = get_redis_client_sync()
self.redis = redis_client

494
src/api/handlers/base/chat_adapter_base.py
View File

@@ -2,43 +2,31 @@
Chat Adapter 通用基类
提供 Chat 格式（进行请求验证和标准化）的通用适配器逻辑：
- 请求解析和验证
- 请求解析和验证（Pydantic）
- 审计日志记录
- 错误处理和响应格式化
- Handler 创建和调用
- 计费策略（支持不同 API 格式的差异化计费）
公共逻辑（异常处理、计费、头部构建等）继承自 HandlerAdapterBase。
子类只需提供：
- FORMAT_ID: API 格式标识
- HANDLER_CLASS: 对应的 ChatHandlerBase 子类
- _validate_request_body(): 可选覆盖请求验证逻辑
- _build_audit_metadata(): 可选覆盖审计元数据构建
- compute_total_input_context(): 可选覆盖总输入上下文计算（用于阶梯计费判定）
- _validate_request_body(): 请求验证逻辑
"""
from __future__ import annotations
import time
from abc import abstractmethod
from typing import Any, ClassVar
from typing import Any
import httpx
from fastapi import HTTPException, Request
from fastapi import HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from src.api.base.adapter import ApiAdapter, ApiMode
from src.api.base.adapter import ApiMode
from src.api.base.context import ApiRequestContext
from src.api.handlers.base.chat_handler_base import ChatHandlerBase
from src.core.api_format import (
ApiFamily,
EndpointKind,
build_adapter_base_headers_for_endpoint,
build_adapter_headers_for_endpoint,
get_adapter_protected_keys_for_endpoint,
get_auth_handler,
get_default_auth_method_for_endpoint,
)
from src.api.handlers.base.handler_adapter_base import HandlerAdapterBase
from src.core.exceptions import (
InvalidRequestException,
ModelNotSupportedException,
@@ -46,17 +34,13 @@ from src.core.exceptions import (
ProviderNotAvailableException,
ProviderRateLimitException,
ProviderTimeoutException,
ProxyException,
QuotaExceededException,
UpstreamClientException,
)
from src.core.logger import logger
from src.services.billing import calculate_request_cost as _calculate_request_cost
from src.services.request.result import RequestResult
from src.services.usage.recorder import UsageRecorder
class ChatAdapterBase(ApiAdapter):
class ChatAdapterBase(HandlerAdapterBase):
"""
Chat Adapter 通用基类
@@ -66,68 +50,12 @@ class ChatAdapterBase(ApiAdapter):
- name: 适配器名称
"""
# 子类必须覆盖
FORMAT_ID: str = "UNKNOWN"
HANDLER_CLASS: type[ChatHandlerBase]
# 新架构：结构化标识（逐步替代直接依赖 FORMAT_ID 的语义）
API_FAMILY: ClassVar[ApiFamily | None] = None
ENDPOINT_KIND: ClassVar[EndpointKind] = EndpointKind.CHAT
# 适配器配置
name: str = "chat.base"
mode = ApiMode.STANDARD
# 计费模板配置（子类可覆盖，如 "claude", "openai", "gemini"）
BILLING_TEMPLATE: str = "claude"
# 子类可以配置的特殊方法（用于check_endpoint）
@classmethod
def build_endpoint_url(cls, base_url: str) -> str:
"""构建端点URL，子类可以覆盖以自定义URL构建逻辑"""
# 默认实现：在base_url后添加特定路径
return base_url
@classmethod
def build_base_headers(cls, api_key: str) -> dict[str, str]:
"""构建基础请求头，使用统一的 headers.py 实现"""
return build_adapter_base_headers_for_endpoint(cls.FORMAT_ID, api_key)
@classmethod
def get_protected_header_keys(cls) -> tuple:
"""返回不应被extra_headers覆盖的头部key，使用统一的 headers.py 实现"""
return get_adapter_protected_keys_for_endpoint(cls.FORMAT_ID)
@classmethod
def build_headers_with_extra(
cls, api_key: str, extra_headers: dict[str, str] | None = None
) -> dict[str, str]:
"""构建完整请求头（包含 extra_headers），使用统一的 headers.py 实现"""
return build_adapter_headers_for_endpoint(cls.FORMAT_ID, api_key, extra_headers)
@classmethod
def build_request_body(cls, request_data: dict[str, Any] | None = None) -> dict[str, Any]:
"""构建测试请求体，使用转换器注册表自动处理格式转换
Args:
request_data: 可选的请求数据，会与默认测试请求合并
Returns:
转换为目标 API 格式的请求体
"""
from src.api.handlers.base.request_builder import build_test_request_body
return build_test_request_body(cls.FORMAT_ID, request_data)
def extract_api_key(self, request: Request) -> str | None:
"""从请求中提取 API 密钥，使用 AuthHandler 新流程"""
auth_method = get_default_auth_method_for_endpoint(self.FORMAT_ID)
handler = get_auth_handler(auth_method)
return handler.extract_credentials(request)
def __init__(self, allowed_api_formats: list[str] | None = None):
self.allowed_api_formats = allowed_api_formats or [self.FORMAT_ID]
async def handle(self, context: ApiRequestContext) -> Any:
"""处理 Chat API 请求"""
http_request = context.request
@@ -189,6 +117,8 @@ class ChatAdapterBase(ApiAdapter):
user_agent=user_agent,
start_time=start_time,
perf_metrics=context.extra.get("perf"),
api_family=self.API_FAMILY.value if self.API_FAMILY else None,
endpoint_kind=self.ENDPOINT_KIND.value if self.ENDPOINT_KIND else None,
)
# 处理请求
@@ -270,6 +200,8 @@ class ChatAdapterBase(ApiAdapter):
user_agent: str,
start_time: float,
perf_metrics: dict[str, Any] | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
) -> Any:
"""创建 Handler 实例 - 子类可覆盖"""
return self.HANDLER_CLASS(
@@ -283,60 +215,26 @@ class ChatAdapterBase(ApiAdapter):
allowed_api_formats=self.allowed_api_formats,
adapter_detector=self.detect_capability_requirements,
perf_metrics=perf_metrics,
api_family=api_family,
endpoint_kind=endpoint_kind,
)
def _merge_path_params(
self, original_request_body: dict[str, Any], path_params: dict[str, Any]
) -> dict[str, Any]:
"""
合并 URL 路径参数到请求体 - 子类可覆盖
默认实现：直接将 path_params 中的字段合并到请求体（不覆盖已有字段）
Args:
original_request_body: 原始请求体字典
path_params: URL 路径参数字典
Returns:
合并后的请求体字典
"""
merged = original_request_body.copy()
for key, value in path_params.items():
if key not in merged:
merged[key] = value
return merged
@abstractmethod
def _validate_request_body(
self, original_request_body: dict, path_params: dict | None = None
) -> None:
"""
验证请求体 - 子类必须实现
Args:
original_request_body: 原始请求体字典
path_params: URL 路径参数（如 Gemini 的 stream 通过 URL 端点传入）
Returns:
验证后的请求对象，或 JSONResponse 错误响应
"""
"""验证请求体 - 子类必须实现"""
pass
def _extract_message_count(self, payload: dict[str, Any], request_obj: Any) -> int:
"""
提取消息数量 - 子类可覆盖
默认实现：从 messages 字段提取
"""
"""提取消息数量 - 子类可覆盖"""
messages = payload.get("messages", [])
if hasattr(request_obj, "messages"):
messages = request_obj.messages
return len(messages) if isinstance(messages, list) else 0
def _build_audit_metadata(self, payload: dict[str, Any], request_obj: Any) -> dict[str, Any]:
"""
构建审计日志元数据 - 子类可覆盖
"""
"""构建审计日志元数据 - 子类可覆盖"""
model = getattr(request_obj, "model", payload.get("model", "unknown"))
stream = getattr(request_obj, "stream", payload.get("stream", False))
messages_count = self._extract_message_count(payload, request_obj)
@@ -351,363 +249,9 @@ class ChatAdapterBase(ApiAdapter):
"top_p": getattr(request_obj, "top_p", payload.get("top_p")),
}
async def _handle_provider_exception(
self,
e: Exception,
*,
db: Session,
user: Any,
api_key: Any,
model: str,
stream: bool,
start_time: float,
original_headers: dict[str, str],
original_request_body: dict[str, Any],
client_ip: str,
request_id: str,
) -> JSONResponse:
"""处理 Provider 相关异常"""
logger.debug(f"Caught provider exception: {type(e).__name__}")
response_time = int((time.time() - start_time) * 1000)
# 使用 RequestResult.from_exception 创建统一的失败结果
# 关键：api_format 从 FORMAT_ID 获取，确保始终有值
result = RequestResult.from_exception(
exception=e,
api_format=self.FORMAT_ID, # 使用 Adapter 的 FORMAT_ID 作为默认值
model=model,
response_time_ms=response_time,
is_stream=stream,
)
result.request_headers = original_headers
result.request_body = original_request_body
# 确定错误消息
if isinstance(e, ProviderAuthException):
error_message = (
"上游服务认证失败" if result.metadata.provider != "unknown" else "服务暂时不可用"
)
result.error_message = error_message
# 处理上游客户端错误（如图片处理失败）
if isinstance(e, UpstreamClientException):
# 返回 400 状态码和清晰的错误消息
result.status_code = e.status_code
result.error_message = e.message
# 使用 UsageRecorder 记录失败
recorder = UsageRecorder(
db=db,
user=user,
api_key=api_key,
client_ip=client_ip,
request_id=request_id,
)
await recorder.record_failure(result, original_headers, original_request_body)
# 根据异常类型确定错误类型
if isinstance(e, UpstreamClientException):
error_type = "invalid_request_error"
elif result.status_code == 503:
error_type = "internal_server_error"
else:
error_type = "rate_limit_exceeded"
return self._error_response(
status_code=result.status_code,
error_type=error_type,
message=result.error_message or str(e),
)
async def _handle_unexpected_exception(
self,
e: Exception,
*,
db: Session,
user: Any,
api_key: Any,
model: str,
stream: bool,
start_time: float,
original_headers: dict[str, str],
original_request_body: dict[str, Any],
client_ip: str,
request_id: str,
) -> JSONResponse:
"""处理未预期的异常"""
if isinstance(e, ProxyException):
logger.error(f"{self.FORMAT_ID} 请求处理业务异常: {type(e).__name__}: {e}")
else:
logger.opt(exception=e).error(
f"{self.FORMAT_ID} 请求处理意外异常: {type(e).__name__}: {e}"
)
response_time = int((time.time() - start_time) * 1000)
# 使用 RequestResult.from_exception 创建统一的失败结果
# 关键：api_format 从 FORMAT_ID 获取，确保始终有值
result = RequestResult.from_exception(
exception=e,
api_format=self.FORMAT_ID, # 使用 Adapter 的 FORMAT_ID 作为默认值
model=model,
response_time_ms=response_time,
is_stream=stream,
)
# 对于未预期的异常，强制设置状态码为 500
result.status_code = 500
result.error_type = "internal_error"
result.request_headers = original_headers
result.request_body = original_request_body
try:
# 使用 UsageRecorder 记录失败
recorder = UsageRecorder(
db=db,
user=user,
api_key=api_key,
client_ip=client_ip,
request_id=request_id,
)
await recorder.record_failure(result, original_headers, original_request_body)
except Exception as record_error:
logger.error(f"记录失败请求时出错: {record_error}")
return self._error_response(
status_code=500, error_type="internal_server_error", message="处理请求时发生内部错误"
)
def _error_response(self, status_code: int, error_type: str, message: str) -> JSONResponse:
"""生成错误响应 - 子类可覆盖以自定义格式"""
return JSONResponse(
status_code=status_code,
content={
"error": {
"type": error_type,
"message": message,
}
},
)
# =========================================================================
# 计费策略相关方法 - 子类可覆盖以实现不同 API 格式的差异化计费
# =========================================================================
def compute_total_input_context(
self,
input_tokens: int,
cache_read_input_tokens: int,
cache_creation_input_tokens: int = 0,
) -> int:
"""
计算总输入上下文（用于阶梯计费判定）
默认实现：input_tokens + cache_read_input_tokens
子类可覆盖此方法实现不同的计算逻辑
Args:
input_tokens: 输入 token 数
cache_read_input_tokens: 缓存读取 token 数
cache_creation_input_tokens: 缓存创建 token 数（部分格式可能需要）
Returns:
总输入上下文 token 数
"""
return input_tokens + cache_read_input_tokens
def compute_cost(
self,
input_tokens: int,
output_tokens: int,
cache_creation_input_tokens: int,
cache_read_input_tokens: int,
input_price_per_1m: float,
output_price_per_1m: float,
cache_creation_price_per_1m: float | None,
cache_read_price_per_1m: float | None,
price_per_request: float | None,
tiered_pricing: dict | None = None,
cache_ttl_minutes: int | None = None,
) -> dict[str, Any]:
"""
计算请求成本
使用 billing 模块的配置驱动计费。
子类可通过设置 BILLING_TEMPLATE 类属性来指定计费模板，
或覆盖此方法实现完全自定义的计费逻辑。
Args:
input_tokens: 输入 token 数
output_tokens: 输出 token 数
cache_creation_input_tokens: 缓存创建 token 数
cache_read_input_tokens: 缓存读取 token 数
input_price_per_1m: 输入价格（每 1M tokens）
output_price_per_1m: 输出价格（每 1M tokens）
cache_creation_price_per_1m: 缓存创建价格（每 1M tokens）
cache_read_price_per_1m: 缓存读取价格（每 1M tokens）
price_per_request: 按次计费价格
tiered_pricing: 阶梯计费配置
cache_ttl_minutes: 缓存时长（分钟）
Returns:
包含各项成本的字典：
{
"input_cost": float,
"output_cost": float,
"cache_creation_cost": float,
"cache_read_cost": float,
"cache_cost": float,
"request_cost": float,
"total_cost": float,
"tier_index": int | None, # 命中的阶梯索引
}
"""
# 计算总输入上下文（使用子类可覆盖的方法）
total_input_context = self.compute_total_input_context(
input_tokens, cache_read_input_tokens, cache_creation_input_tokens
)
return _calculate_request_cost(
input_tokens=input_tokens,
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens,
input_price_per_1m=input_price_per_1m,
output_price_per_1m=output_price_per_1m,
cache_creation_price_per_1m=cache_creation_price_per_1m,
cache_read_price_per_1m=cache_read_price_per_1m,
price_per_request=price_per_request,
tiered_pricing=tiered_pricing,
cache_ttl_minutes=cache_ttl_minutes,
total_input_context=total_input_context,
billing_template=self.BILLING_TEMPLATE,
)
# =========================================================================
# 模型列表查询 - 子类应覆盖此方法
# =========================================================================
@classmethod
async def fetch_models(
cls,
client: httpx.AsyncClient,
base_url: str,
api_key: str,
extra_headers: dict[str, str] | None = None,
) -> tuple[list, str | None]:
"""
查询上游 API 支持的模型列表
这是 Aether 内部发起的请求（非用户透传），用于：
- 管理后台查询提供商支持的模型
- 自动发现可用模型
Args:
client: httpx 异步客户端
base_url: API 基础 URL
api_key: API 密钥（已解密）
extra_headers: 端点配置的额外请求头
Returns:
(models, error): 模型列表和错误信息
- models: 模型信息列表，每个模型至少包含 id 字段
- error: 错误信息，成功时为 None
"""
# 默认实现返回空列表，子类应覆盖
return [], f"{cls.FORMAT_ID} adapter does not implement fetch_models"
@classmethod
async def check_endpoint(
cls,
client: httpx.AsyncClient,
base_url: str,
api_key: str,
request_data: dict[str, Any],
extra_headers: dict[str, str] | None = None,
# 端点规则参数
body_rules: list[dict[str, Any]] | None = None,
header_rules: list[dict[str, Any]] | None = None,
# 用量计算参数（现在强制记录）
db: Any | None = None,
user: Any | None = None,
provider_name: str | None = None,
provider_id: str | None = None,
api_key_id: str | None = None,
model_name: str | None = None,
# Provider 上下文（Chat 适配器忽略这些参数，仅保持签名兼容）
auth_type: str | None = None, # noqa: ARG003
provider_type: str | None = None, # noqa: ARG003
decrypted_auth_config: dict[str, Any] | None = None, # noqa: ARG003
# 代理参数（已解析，直接传递给 run_endpoint_check）
proxy_param: Any | None = None,
) -> dict[str, Any]:
"""
测试模型连接性（非流式）
Args:
client: httpx 异步客户端
base_url: API 基础 URL
api_key: API 密钥（已解密）
request_data: 请求数据
extra_headers: 端点配置的额外请求头
body_rules: 请求体规则（在格式转换后应用）
header_rules: 请求头规则（在请求头构建后应用）
db: 数据库会话
user: 用户对象
provider_name: 提供商名称
provider_id: 提供商ID
api_key_id: API Key ID
model_name: 模型名称
Returns:
测试响应数据
"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
from src.api.handlers.base.request_builder import apply_body_rules
from src.core.api_format.headers import HeaderBuilder
# 使用子类配置方法构建请求组件
url = cls.build_endpoint_url(base_url)
headers = cls.build_headers_with_extra(api_key, extra_headers)
body = cls.build_request_body(request_data)
# 应用请求体规则（在格式转换后应用，确保规则效果不被覆盖）
if body_rules:
body = apply_body_rules(body, body_rules)
# 应用请求头规则（在请求头构建后应用）
if header_rules:
# 获取认证头名称，防止被规则覆盖
from src.core.api_format import get_auth_config_for_endpoint
auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
protected_keys = {auth_header.lower(), "content-type"}
header_builder = HeaderBuilder()
header_builder.add_many(headers)
header_builder.apply_rules(header_rules, protected_keys)
headers = header_builder.build()
# 使用通用的endpoint checker执行请求
return await run_endpoint_check(
client=client,
url=url,
headers=headers,
json_body=body,
api_format=cls.FORMAT_ID,
# 用量计算参数（现在强制记录）
db=db,
user=user,
provider_name=provider_name,
provider_id=provider_id,
api_key_id=api_key_id,
model_name=model_name or request_data.get("model"),
proxy_param=proxy_param,
)
# =========================================================================
# Adapter 注册表 - 用于根据 API format 获取 Adapter 实例
# Adapter 注册表
# =========================================================================
_ADAPTER_REGISTRY: dict[str, type[ChatAdapterBase]] = {}

11
src/api/handlers/base/chat_handler_base.py
View File

@@ -141,6 +141,8 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
Callable[[dict[str, str], dict[str, Any] | None], dict[str, bool]]
) = None,
perf_metrics: dict[str, Any] | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
):
allowed = allowed_api_formats or [self.FORMAT_ID]
super().__init__(
@@ -154,6 +156,8 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
allowed_api_formats=allowed,
adapter_detector=adapter_detector,
perf_metrics=perf_metrics,
api_family=api_family,
endpoint_kind=endpoint_kind,
)
self._parser: ResponseParser | None = None
self._request_builder = PassthroughRequestBuilder()
@@ -483,7 +487,12 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
request_body_ref: dict[str, Any] = {"body": original_request_body}
# 创建类型安全的流式上下文
ctx = StreamContext(model=model, api_format=api_format)
ctx = StreamContext(
model=model,
api_format=api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
)
ctx.request_id = self.request_id
ctx.client_api_format = (
api_format.value if hasattr(api_format, "value") else str(api_format)

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

@@ -210,6 +210,8 @@ class ChatSyncExecutor:
is_stream=False,
provider_request_headers=ctx.provider_request_headers,
api_format=api_format,
api_family=handler.api_family,
endpoint_kind=handler.endpoint_kind,
# 格式转换追踪
endpoint_api_format=ctx.provider_api_format_for_error or None,
has_format_conversion=is_format_converted(
@@ -290,6 +292,8 @@ class ChatSyncExecutor:
error_message=str(e),
is_stream=False,
api_format=api_format,
api_family=handler.api_family,
endpoint_kind=handler.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
response_headers=ctx.response_headers,
client_response_headers={"content-type": "application/json"},
@@ -349,6 +353,8 @@ class ChatSyncExecutor:
provider_request_body=ctx.provider_request_body,
is_stream=False,
api_format=api_format,
api_family=handler.api_family,
endpoint_kind=handler.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
response_headers=error_response_headers,
# 非流式失败返回给客户端的是 JSON 错误响应
@@ -719,6 +725,8 @@ class ChatSyncExecutor:
provider_request_body=ctx.provider_request_body,
is_stream=True,
api_format=ctx.api_format,
api_family=handler.api_family,
endpoint_kind=handler.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,

668
src/api/handlers/base/cli_adapter_base.py
View File

@@ -4,9 +4,9 @@ CLI Adapter 通用基类
提供 CLI 格式（直接透传请求）的通用适配器逻辑：
- 请求解析和验证
- 审计日志记录
- 错误处理和响应格式化
- Handler 创建和调用
- 计费策略（支持不同 API 格式的差异化计费）
公共逻辑（异常处理、计费、头部构建等）继承自 HandlerAdapterBase。
子类只需提供：
- FORMAT_ID: API 格式标识
@@ -17,26 +17,16 @@ CLI Adapter 通用基类
from __future__ import annotations
import time
from typing import Any, ClassVar
from typing import Any
import httpx
from fastapi import HTTPException, Request
from fastapi import HTTPException
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from src.api.base.adapter import ApiAdapter, ApiMode
from src.api.base.adapter import ApiMode
from src.api.base.context import ApiRequestContext
from src.api.handlers.base.cli_handler_base import CliMessageHandlerBase
from src.core.api_format import (
ApiFamily,
EndpointKind,
build_adapter_base_headers_for_endpoint,
build_adapter_headers_for_endpoint,
get_adapter_protected_keys_for_endpoint,
get_auth_handler,
get_default_auth_method_for_endpoint,
)
from src.api.handlers.base.handler_adapter_base import HandlerAdapterBase
from src.core.api_format import EndpointKind
from src.core.exceptions import (
InvalidRequestException,
ModelNotSupportedException,
@@ -44,17 +34,13 @@ from src.core.exceptions import (
ProviderNotAvailableException,
ProviderRateLimitException,
ProviderTimeoutException,
ProxyException,
QuotaExceededException,
UpstreamClientException,
)
from src.core.logger import logger
from src.services.billing import calculate_request_cost as _calculate_request_cost
from src.services.request.result import RequestResult
from src.services.usage.recorder import UsageRecorder
class CliAdapterBase(ApiAdapter):
class CliAdapterBase(HandlerAdapterBase):
"""
CLI Adapter 通用基类
@@ -64,67 +50,15 @@ class CliAdapterBase(ApiAdapter):
- name: 适配器名称
"""
# 子类必须覆盖
FORMAT_ID: str = "UNKNOWN"
HANDLER_CLASS: type[CliMessageHandlerBase]
# 新架构：结构化标识（逐步替代直接依赖 FORMAT_ID 的语义）
API_FAMILY: ClassVar[ApiFamily | None] = None
ENDPOINT_KIND: ClassVar[EndpointKind] = EndpointKind.CLI
# CLI 端点类型覆盖（基类默认 CHAT）
ENDPOINT_KIND = EndpointKind.CLI
# 适配器配置
name: str = "cli.base"
mode = ApiMode.PROXY
# 计费模板配置（子类可覆盖，如 "claude", "openai", "gemini"）
BILLING_TEMPLATE: str = "claude"
def __init__(self, allowed_api_formats: list[str] | None = None):
self.allowed_api_formats = allowed_api_formats or [self.FORMAT_ID]
# =========================================================================
# API 格式与头部处理 - 使用统一的 headers.py 函数
# =========================================================================
def extract_api_key(self, request: Request) -> str | None:
"""
从请求中提取 API 密钥
使用 AuthHandler 新流程，根据 API 格式选择认证方式。
"""
auth_method = get_default_auth_method_for_endpoint(self.FORMAT_ID)
handler = get_auth_handler(auth_method)
return handler.extract_credentials(request)
@classmethod
def build_base_headers(cls, api_key: str) -> dict[str, str]:
"""
构建 CLI API 认证头
使用统一的头部处理函数。
"""
return build_adapter_base_headers_for_endpoint(cls.FORMAT_ID, api_key)
@classmethod
def build_headers_with_extra(
cls, api_key: str, extra_headers: dict[str, str] | None = None
) -> dict[str, str]:
"""
构建带额外头部的完整请求头
使用统一的头部处理函数，自动保护关键头部不被覆盖。
"""
return build_adapter_headers_for_endpoint(cls.FORMAT_ID, api_key, extra_headers)
@classmethod
def get_protected_header_keys(cls) -> tuple[str, ...]:
"""
返回 CLI API 的保护头部 key
使用统一的头部处理函数。
"""
return get_adapter_protected_keys_for_endpoint(cls.FORMAT_ID)
async def handle(self, context: ApiRequestContext) -> Any:
"""处理 CLI API 请求"""
http_request = context.request
@@ -137,7 +71,7 @@ class CliAdapterBase(ApiAdapter):
client_ip = context.client_ip
user_agent = context.user_agent
original_headers = context.original_headers
query_params = context.query_params # 获取查询参数
query_params = context.query_params
original_request_body = context.ensure_json_body()
@@ -192,6 +126,8 @@ class CliAdapterBase(ApiAdapter):
allowed_api_formats=self.allowed_api_formats,
adapter_detector=self.detect_capability_requirements,
perf_metrics=context.extra.get("perf"),
api_family=self.API_FAMILY.value if self.API_FAMILY else None,
endpoint_kind=self.ENDPOINT_KIND.value if self.ENDPOINT_KIND else None,
)
# 处理请求
@@ -218,7 +154,7 @@ class CliAdapterBase(ApiAdapter):
QuotaExceededException,
InvalidRequestException,
) as e:
logger.debug(f"客户端请求错误: {e.error_type}")
logger.debug("客户端请求错误: {}", e.error_type)
return self._error_response(
status_code=e.status_code,
error_type=("invalid_request_error" if e.status_code == 400 else "quota_exceeded"),
@@ -261,33 +197,8 @@ class CliAdapterBase(ApiAdapter):
request_id=request_id,
)
def _merge_path_params(
self, original_request_body: dict[str, Any], path_params: dict[str, Any]
) -> dict[str, Any]:
"""
合并 URL 路径参数到请求体 - 子类可覆盖
默认实现：直接将 path_params 中的字段合并到请求体（不覆盖已有字段）
Args:
original_request_body: 原始请求体字典
path_params: URL 路径参数字典
Returns:
合并后的请求体字典
"""
merged = original_request_body.copy()
for key, value in path_params.items():
if key not in merged:
merged[key] = value
return merged
def _extract_message_count(self, payload: dict[str, Any]) -> int:
"""
提取消息数量 - 子类可覆盖
默认实现：从 input 字段提取
"""
"""提取消息数量 - 子类可覆盖"""
if "input" not in payload:
return 0
input_data = payload["input"]
@@ -302,14 +213,7 @@ class CliAdapterBase(ApiAdapter):
payload: dict[str, Any],
path_params: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""
构建审计日志元数据 - 子类可覆盖
Args:
payload: 请求体
path_params: URL 路径参数（用于获取 model 等）
"""
# 优先从请求体获取 model，其次从 path_params
"""构建审计日志元数据 - 子类可覆盖"""
model = payload.get("model")
if model is None and path_params:
model = path_params.get("model", "unknown")
@@ -330,536 +234,9 @@ class CliAdapterBase(ApiAdapter):
"instructions_present": bool(payload.get("instructions")),
}
async def _handle_provider_exception(
self,
e: Exception,
*,
db: Session,
user: Any,
api_key: Any,
model: str,
stream: bool,
start_time: float,
original_headers: dict[str, str],
original_request_body: dict[str, Any],
client_ip: str,
request_id: str,
) -> JSONResponse:
"""处理 Provider 相关异常"""
logger.debug(f"Caught provider exception: {type(e).__name__}")
response_time = int((time.time() - start_time) * 1000)
# 使用 RequestResult.from_exception 创建统一的失败结果
# 关键：api_format 从 FORMAT_ID 获取，确保始终有值
result = RequestResult.from_exception(
exception=e,
api_format=self.FORMAT_ID, # 使用 Adapter 的 FORMAT_ID 作为默认值
model=model,
response_time_ms=response_time,
is_stream=stream,
)
result.request_headers = original_headers
result.request_body = original_request_body
# 确定错误消息
if isinstance(e, ProviderAuthException):
error_message = (
"上游服务认证失败" if result.metadata.provider != "unknown" else "服务暂时不可用"
)
result.error_message = error_message
# 处理上游客户端错误（如图片处理失败）
if isinstance(e, UpstreamClientException):
# 返回 400 状态码和清晰的错误消息
result.status_code = e.status_code
result.error_message = e.message
# 使用 UsageRecorder 记录失败
recorder = UsageRecorder(
db=db,
user=user,
api_key=api_key,
client_ip=client_ip,
request_id=request_id,
)
await recorder.record_failure(result, original_headers, original_request_body)
# 根据异常类型确定错误类型
if isinstance(e, UpstreamClientException):
error_type = "invalid_request_error"
elif result.status_code == 503:
error_type = "internal_server_error"
else:
error_type = "rate_limit_exceeded"
return self._error_response(
status_code=result.status_code,
error_type=error_type,
message=result.error_message or str(e),
)
async def _handle_unexpected_exception(
self,
e: Exception,
*,
db: Session,
user: Any,
api_key: Any,
model: str,
stream: bool,
start_time: float,
original_headers: dict[str, str],
original_request_body: dict[str, Any],
client_ip: str,
request_id: str,
) -> JSONResponse:
"""处理未预期的异常"""
if isinstance(e, ProxyException):
logger.error(f"{self.FORMAT_ID} 请求处理业务异常: {type(e).__name__}: {e}")
else:
logger.opt(exception=e).error(
f"{self.FORMAT_ID} 请求处理意外异常: {type(e).__name__}: {e}"
)
response_time = int((time.time() - start_time) * 1000)
# 使用 RequestResult.from_exception 创建统一的失败结果
# 关键：api_format 从 FORMAT_ID 获取，确保始终有值
result = RequestResult.from_exception(
exception=e,
api_format=self.FORMAT_ID, # 使用 Adapter 的 FORMAT_ID 作为默认值
model=model,
response_time_ms=response_time,
is_stream=stream,
)
# 对于未预期的异常，强制设置状态码为 500
result.status_code = 500
result.error_type = "internal_error"
result.request_headers = original_headers
result.request_body = original_request_body
# 使用 UsageRecorder 记录失败
recorder = UsageRecorder(
db=db,
user=user,
api_key=api_key,
client_ip=client_ip,
request_id=request_id,
)
await recorder.record_failure(result, original_headers, original_request_body)
return self._error_response(
status_code=500, error_type="internal_server_error", message="处理请求时发生内部错误"
)
def _error_response(self, status_code: int, error_type: str, message: str) -> JSONResponse:
"""生成错误响应"""
return JSONResponse(
status_code=status_code,
content={
"error": {
"type": error_type,
"message": message,
}
},
)
# =========================================================================
# 计费策略相关方法 - 子类可覆盖以实现不同 API 格式的差异化计费
# =========================================================================
def compute_total_input_context(
self,
input_tokens: int,
cache_read_input_tokens: int,
cache_creation_input_tokens: int = 0,
) -> int:
"""
计算总输入上下文（用于阶梯计费判定）
默认实现：input_tokens + cache_read_input_tokens
子类可覆盖此方法实现不同的计算逻辑
Args:
input_tokens: 输入 token 数
cache_read_input_tokens: 缓存读取 token 数
cache_creation_input_tokens: 缓存创建 token 数（部分格式可能需要）
Returns:
总输入上下文 token 数
"""
return input_tokens + cache_read_input_tokens
def compute_cost(
self,
input_tokens: int,
output_tokens: int,
cache_creation_input_tokens: int,
cache_read_input_tokens: int,
input_price_per_1m: float,
output_price_per_1m: float,
cache_creation_price_per_1m: float | None,
cache_read_price_per_1m: float | None,
price_per_request: float | None,
tiered_pricing: dict | None = None,
cache_ttl_minutes: int | None = None,
) -> dict[str, Any]:
"""
计算请求成本
使用 billing 模块的配置驱动计费。
子类可通过设置 BILLING_TEMPLATE 类属性来指定计费模板，
或覆盖此方法实现完全自定义的计费逻辑。
Args:
input_tokens: 输入 token 数
output_tokens: 输出 token 数
cache_creation_input_tokens: 缓存创建 token 数
cache_read_input_tokens: 缓存读取 token 数
input_price_per_1m: 输入价格（每 1M tokens）
output_price_per_1m: 输出价格（每 1M tokens）
cache_creation_price_per_1m: 缓存创建价格（每 1M tokens）
cache_read_price_per_1m: 缓存读取价格（每 1M tokens）
price_per_request: 按次计费价格
tiered_pricing: 阶梯计费配置
cache_ttl_minutes: 缓存时长（分钟）
Returns:
包含各项成本的字典
"""
# 计算总输入上下文（使用子类可覆盖的方法）
total_input_context = self.compute_total_input_context(
input_tokens, cache_read_input_tokens, cache_creation_input_tokens
)
return _calculate_request_cost(
input_tokens=input_tokens,
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens,
input_price_per_1m=input_price_per_1m,
output_price_per_1m=output_price_per_1m,
cache_creation_price_per_1m=cache_creation_price_per_1m,
cache_read_price_per_1m=cache_read_price_per_1m,
price_per_request=price_per_request,
tiered_pricing=tiered_pricing,
cache_ttl_minutes=cache_ttl_minutes,
total_input_context=total_input_context,
billing_template=self.BILLING_TEMPLATE,
)
# =========================================================================
# 模型列表查询 - 子类应覆盖此方法
# =========================================================================
@classmethod
async def fetch_models(
cls,
client: httpx.AsyncClient,
base_url: str,
api_key: str,
extra_headers: dict[str, str] | None = None,
) -> tuple[list, str | None]:
"""
查询上游 API 支持的模型列表
这是 Aether 内部发起的请求（非用户透传），用于：
- 管理后台查询提供商支持的模型
- 自动发现可用模型
Args:
client: httpx 异步客户端
base_url: API 基础 URL
api_key: API 密钥（已解密）
extra_headers: 端点配置的额外请求头
Returns:
(models, error): 模型列表和错误信息
- models: 模型信息列表，每个模型至少包含 id 字段
- error: 错误信息，成功时为 None
"""
# 默认实现返回空列表，子类应覆盖
return [], f"{cls.FORMAT_ID} adapter does not implement fetch_models"
@classmethod
async def check_endpoint(
cls,
client: httpx.AsyncClient,
base_url: str,
api_key: str,
request_data: dict[str, Any],
extra_headers: dict[str, str] | None = None,
# 端点规则参数
body_rules: list[dict[str, Any]] | None = None,
header_rules: list[dict[str, Any]] | None = None,
# 用量计算参数
db: Any | None = None,
user: Any | None = None,
provider_name: str | None = None,
provider_id: str | None = None,
api_key_id: str | None = None,
model_name: str | None = None,
# Provider 上下文（用于 OAuth 认证和 Antigravity 等特殊路由）
auth_type: str | None = None,
provider_type: str | None = None,
decrypted_auth_config: dict[str, Any] | None = None,
# 代理参数（已解析，直接传递给 run_endpoint_check）
proxy_param: Any | None = None,
) -> dict[str, Any]:
"""
测试模型连接性（非流式）
通用的CLI endpoint测试方法，使用配置方法模式：
- build_endpoint_url(): 构建请求URL
- build_base_headers(): 构建基础认证头
- get_protected_header_keys(): 获取受保护的头部key
- build_request_body(): 构建请求体
- get_cli_user_agent(): 获取CLI User-Agent（子类可覆盖）
Args:
client: httpx 异步客户端
base_url: API 基础 URL
api_key: API 密钥（已解密）
request_data: 请求数据
extra_headers: 端点配置的额外请求头
body_rules: 请求体规则（在格式转换后应用）
header_rules: 请求头规则（在请求头构建后应用）
db: 数据库会话
user: 用户对象
provider_name: 提供商名称
provider_id: 提供商ID
api_key_id: API密钥ID
model_name: 模型名称
auth_type: Key 认证类型（"api_key"/"oauth"/"vertex_ai"），
OAuth 类型自动使用 Authorization: Bearer 替代端点默认认证头
provider_type: 提供商类型（用于 Antigravity v1internal 等特殊路由）
decrypted_auth_config: 解密后的 OAuth 配置（Antigravity 需要 project_id）
Returns:
测试响应数据
"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
from src.api.handlers.base.request_builder import apply_body_rules
from src.core.api_format.headers import HeaderBuilder
from src.core.provider_types import ProviderType
is_antigravity = provider_type == ProviderType.ANTIGRAVITY
is_kiro = provider_type == ProviderType.KIRO
is_oauth = auth_type == "oauth"
# ---- URL ----
if is_kiro:
# Kiro 需要替换 base_url 中的 {region} 占位符，并使用专用路径
from src.services.provider.adapters.kiro.constants import (
DEFAULT_REGION,
KIRO_GENERATE_ASSISTANT_PATH,
)
region = (decrypted_auth_config or {}).get("region") or DEFAULT_REGION
effective_base_url = (
base_url.replace("{region}", region) if "{region}" in base_url else base_url
)
url = f"{str(effective_base_url).rstrip('/')}{KIRO_GENERATE_ASSISTANT_PATH}"
elif is_antigravity:
# Antigravity 走 v1internal 端点，模型名在请求体 envelope 中，不在 URL 路径里
from src.services.provider.adapters.antigravity.constants import (
V1INTERNAL_PATH_TEMPLATE,
)
from src.services.provider.adapters.antigravity.constants import (
get_http_user_agent as _get_antigravity_ua,
)
from src.services.provider.adapters.antigravity.envelope import wrap_v1internal_request
from src.services.provider.adapters.antigravity.url_availability import url_availability
ordered_urls = url_availability.get_ordered_urls(prefer_daily=True)
effective_base_url = ordered_urls[0] if ordered_urls else base_url
path = V1INTERNAL_PATH_TEMPLATE.format(action="generateContent")
url = f"{str(effective_base_url).rstrip('/')}{path}"
else:
url = cls.build_endpoint_url(base_url, request_data, model_name)
# ---- Headers ----
cli_extra = cls.get_cli_extra_headers(base_url=base_url)
merged_extra = dict(extra_headers) if extra_headers else {}
merged_extra.update(cli_extra)
# Antigravity 需要特定的 User-Agent
if is_antigravity:
merged_extra["User-Agent"] = _get_antigravity_ua()
# Kiro 需要特定的请求头
if is_kiro:
from src.services.provider.adapters.kiro.headers import build_generate_assistant_headers
from src.services.provider.adapters.kiro.models.credentials import KiroAuthConfig
from src.services.provider.adapters.kiro.token_manager import generate_machine_id
kiro_cfg = KiroAuthConfig.from_dict(decrypted_auth_config or {})
region = kiro_cfg.region or DEFAULT_REGION
machine_id = generate_machine_id(kiro_cfg)
kiro_headers = build_generate_assistant_headers(
host=f"q.{region}.amazonaws.com",
access_token=api_key,
machine_id=machine_id,
kiro_version=kiro_cfg.kiro_version,
system_version=kiro_cfg.system_version,
node_version=kiro_cfg.node_version,
)
merged_extra.update(kiro_headers)
headers = cls.build_headers_with_extra(api_key, merged_extra if merged_extra else None)
# OAuth 统一处理：替换端点默认认证头为 Authorization: Bearer
# （与 get_provider_auth 返回的 ProviderAuthInfo 行为一致）
if is_oauth:
from src.core.api_format import get_auth_config_for_endpoint
default_auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
if default_auth_header.lower() != "authorization":
headers.pop(default_auth_header, None)
headers["Authorization"] = f"Bearer {api_key}"
# ---- Body ----
body = cls.build_request_body(request_data, base_url=base_url)
if body_rules:
body = apply_body_rules(body, body_rules)
# Antigravity：用 v1internal envelope 包装请求体
if is_antigravity:
project_id = (decrypted_auth_config or {}).get("project_id", "")
effective_model = model_name or request_data.get("model", "")
body = wrap_v1internal_request(
body,
project_id=project_id,
model=effective_model,
)
# Kiro：用 conversationState envelope 包装请求体
if is_kiro:
from src.services.provider.adapters.kiro.converter import (
convert_claude_messages_to_conversation_state,
)
effective_model = model_name or request_data.get("model", "")
conversation_state = convert_claude_messages_to_conversation_state(
body,
model=effective_model,
)
body = {"conversationState": conversation_state}
if isinstance(kiro_cfg.profile_arn, str) and kiro_cfg.profile_arn.strip():
body["profileArn"] = kiro_cfg.profile_arn.strip()
# ---- Header Rules ----
if header_rules:
from src.core.api_format import get_auth_config_for_endpoint as _get_auth_cfg
# 保护实际使用的认证头，而非端点默认的
if is_oauth:
protected_keys = {"authorization", "content-type"}
else:
ep_auth_header, _ = _get_auth_cfg(cls.FORMAT_ID)
protected_keys = {ep_auth_header.lower(), "content-type"}
header_builder = HeaderBuilder()
header_builder.add_many(headers)
header_builder.apply_rules(header_rules, protected_keys)
headers = header_builder.build()
# ---- Execute ----
effective_model_name = model_name or request_data.get("model")
return await run_endpoint_check(
client=client,
url=url,
headers=headers,
json_body=body,
api_format=cls.FORMAT_ID,
db=db,
user=user,
provider_name=provider_name,
provider_id=provider_id,
api_key_id=api_key_id,
model_name=effective_model_name,
proxy_param=proxy_param,
)
# =========================================================================
# CLI Adapter 配置方法 - 子类应覆盖这些方法
# =========================================================================
@classmethod
def build_endpoint_url(
cls, base_url: str, request_data: dict[str, Any], model_name: str | None = None
) -> str:
"""
构建CLI API端点URL - 子类应覆盖
Args:
base_url: API基础URL
request_data: 请求数据
model_name: 模型名称（某些API需要，如Gemini）
Returns:
完整的端点URL
"""
raise NotImplementedError(f"{cls.FORMAT_ID} adapter must implement build_endpoint_url")
@classmethod
def build_request_body(
cls,
request_data: dict[str, Any] | None = None,
*,
base_url: str | None = None,
) -> dict[str, Any]:
"""构建测试请求体，使用转换器注册表自动处理格式转换
Args:
request_data: 可选的请求数据，会与默认测试请求合并
base_url: API 基础 URL，用于判断特殊端点（如 Codex）
Returns:
转换为目标 API 格式的请求体
"""
from src.api.handlers.base.request_builder import build_test_request_body
# 基类不使用 base_url，子类可覆盖以支持特殊端点
_ = base_url
return build_test_request_body(cls.FORMAT_ID, request_data)
@classmethod
def get_cli_user_agent(cls) -> str | None:
"""
获取CLI User-Agent - 子类可覆盖
Returns:
CLI User-Agent字符串，如果不需要则为None
"""
return None
@classmethod
def get_cli_extra_headers(cls, *, base_url: str | None = None) -> dict[str, str]:
"""
获取CLI额外请求头 - 子类可覆盖
用于 check_endpoint 测试请求时添加额外的头部。
默认实现只添加 User-Agent（如果有）。
Args:
base_url: API 基础 URL，子类可据此判断特殊端点（如 Codex）
Returns:
额外请求头字典
"""
headers: dict[str, str] = {}
cli_user_agent = cls.get_cli_user_agent()
if cli_user_agent:
headers["User-Agent"] = cli_user_agent
return headers
# =========================================================================
# CLI Adapter 注册表 - 用于根据 API format 获取 CLI Adapter 实例
# CLI Adapter 注册表
# =========================================================================
_CLI_ADAPTER_REGISTRY: dict[str, type[CliAdapterBase]] = {}
@@ -888,7 +265,6 @@ def _ensure_cli_adapters_loaded() -> None:
if _CLI_ADAPTERS_LOADED:
return
# 导入各个 CLI Adapter 模块以触发 @register_cli_adapter 装饰器
try:
from src.api.handlers.claude_cli import adapter as _ # noqa: F401
except ImportError:
@@ -906,15 +282,7 @@ def _ensure_cli_adapters_loaded() -> None:
def get_cli_adapter_class(api_format: str) -> type[CliAdapterBase] | None:
"""
根据 API format 获取 CLI Adapter 类
Args:
api_format: API 格式标识（如 "openai:cli", "claude:cli", "gemini:cli"）
Returns:
对应的 CLI Adapter 类，如果未找到返回 None
"""
"""根据 API format 获取 CLI Adapter 类"""
_ensure_cli_adapters_loaded()
return _CLI_ADAPTER_REGISTRY.get(api_format.upper()) if api_format else None

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

@@ -79,6 +79,8 @@ class CliMessageHandlerBase(
Callable[[dict[str, str], dict[str, Any] | None], dict[str, bool]]
) = None,
perf_metrics: dict[str, Any] | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
):
allowed = allowed_api_formats or [self.FORMAT_ID]
super().__init__(
@@ -92,6 +94,8 @@ class CliMessageHandlerBase(
allowed_api_formats=allowed,
adapter_detector=adapter_detector,
perf_metrics=perf_metrics,
api_family=api_family,
endpoint_kind=endpoint_kind,
)
self._parser: ResponseParser | None = None
self._request_builder = PassthroughRequestBuilder()

8
src/api/handlers/base/cli_monitor_mixin.py
View File

@@ -234,6 +234,8 @@ class CliMonitorMixin:
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
input_tokens=ctx.input_tokens,
@@ -267,6 +269,8 @@ class CliMonitorMixin:
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
# 预估 token 信息（来自 message_start 事件）
@@ -353,6 +357,8 @@ class CliMonitorMixin:
is_stream=True,
provider_request_headers=ctx.provider_request_headers,
api_format=ctx.api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
# 格式转换追踪
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.has_format_conversion,
@@ -499,6 +505,8 @@ class CliMonitorMixin:
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
response_headers=ctx.response_headers,

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

@@ -108,6 +108,8 @@ class CliStreamMixin:
ctx = StreamContext(
model=model,
api_format=self.allowed_api_formats[0],
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
request_id=self.request_id,
user_id=self.user.id,
api_key_id=self.api_key.id,

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

@@ -538,6 +538,8 @@ class CliSyncMixin:
is_stream=False,
provider_request_headers=provider_request_headers,
api_format=api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
# 格式转换追踪
endpoint_api_format=provider_api_format or None,
has_format_conversion=is_format_converted(provider_api_format, str(api_format)),
@@ -579,6 +581,8 @@ class CliSyncMixin:
error_message=str(e),
is_stream=False,
api_format=api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
request_metadata=request_metadata or None,
)
raise
@@ -615,6 +619,8 @@ class CliSyncMixin:
provider_request_body=provider_request_body,
is_stream=False,
api_format=api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
provider_request_headers=provider_request_headers,
response_headers=error_response_headers,
# 非流式失败返回给客户端的是 JSON 错误响应

518
src/api/handlers/base/handler_adapter_base.py Normal file
View File

@@ -0,0 +1,518 @@
"""
Handler Adapter 公共基类
从 ChatAdapterBase 和 CliAdapterBase 提取的共享逻辑：
- API 格式与头部处理
- 异常处理和错误响应
- 计费策略
- 模型列表查询和端点测试
- 路径参数合并
子类（ChatAdapterBase / CliAdapterBase）只需关注各自的 handle() 流程差异。
"""
from __future__ import annotations
import time
from typing import Any, ClassVar
import httpx
from fastapi import Request
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from src.api.base.adapter import ApiAdapter
from src.core.api_format import (
ApiFamily,
EndpointKind,
build_adapter_base_headers_for_endpoint,
build_adapter_headers_for_endpoint,
get_adapter_protected_keys_for_endpoint,
get_auth_handler,
get_default_auth_method_for_endpoint,
)
from src.core.exceptions import (
ProviderAuthException,
ProxyException,
UpstreamClientException,
)
from src.core.logger import logger
from src.services.billing import calculate_request_cost as _calculate_request_cost
from src.services.request.result import RequestResult
from src.services.usage.recorder import UsageRecorder
class HandlerAdapterBase(ApiAdapter):
"""
Chat/CLI Adapter 的公共基类
封装两者共享的逻辑：
- API 格式与头部处理
- 异常处理和错误响应
- 计费策略
- 模型列表查询和端点测试
子类（ChatAdapterBase / CliAdapterBase）只需实现 handle() 和格式特有的方法。
"""
# 子类必须覆盖
FORMAT_ID: str = "UNKNOWN"
# 结构化标识
API_FAMILY: ClassVar[ApiFamily | None] = None
ENDPOINT_KIND: ClassVar[EndpointKind] = EndpointKind.CHAT
# 计费模板配置（子类可覆盖，如 "claude", "openai", "gemini"）
BILLING_TEMPLATE: str = "claude"
def __init__(self, allowed_api_formats: list[str] | None = None):
self.allowed_api_formats = allowed_api_formats or [self.FORMAT_ID]
# =========================================================================
# API 格式与头部处理
# =========================================================================
def extract_api_key(self, request: Request) -> str | None:
"""从请求中提取 API 密钥"""
auth_method = get_default_auth_method_for_endpoint(self.FORMAT_ID)
handler = get_auth_handler(auth_method)
return handler.extract_credentials(request)
@classmethod
def build_base_headers(cls, api_key: str) -> dict[str, str]:
"""构建基础认证头"""
return build_adapter_base_headers_for_endpoint(cls.FORMAT_ID, api_key)
@classmethod
def build_headers_with_extra(
cls, api_key: str, extra_headers: dict[str, str] | None = None
) -> dict[str, str]:
"""构建带额外头部的完整请求头"""
return build_adapter_headers_for_endpoint(cls.FORMAT_ID, api_key, extra_headers)
@classmethod
def get_protected_header_keys(cls) -> tuple[str, ...]:
"""返回不应被 extra_headers 覆盖的头部 key"""
return get_adapter_protected_keys_for_endpoint(cls.FORMAT_ID)
# =========================================================================
# 路径参数合并
# =========================================================================
def _merge_path_params(
self, original_request_body: dict[str, Any], path_params: dict[str, Any]
) -> dict[str, Any]:
"""合并 URL 路径参数到请求体 - 子类可覆盖"""
merged = original_request_body.copy()
for key, value in path_params.items():
if key not in merged:
merged[key] = value
return merged
# =========================================================================
# 异常处理
# =========================================================================
async def _handle_provider_exception(
self,
e: Exception,
*,
db: Session,
user: Any,
api_key: Any,
model: str,
stream: bool,
start_time: float,
original_headers: dict[str, str],
original_request_body: dict[str, Any],
client_ip: str,
request_id: str,
) -> JSONResponse:
"""处理 Provider 相关异常"""
logger.debug("Caught provider exception: {}", type(e).__name__)
response_time = int((time.time() - start_time) * 1000)
result = RequestResult.from_exception(
exception=e,
api_format=self.FORMAT_ID,
model=model,
response_time_ms=response_time,
is_stream=stream,
)
result.request_headers = original_headers
result.request_body = original_request_body
if isinstance(e, ProviderAuthException):
error_message = (
"上游服务认证失败" if result.metadata.provider != "unknown" else "服务暂时不可用"
)
result.error_message = error_message
if isinstance(e, UpstreamClientException):
result.status_code = e.status_code
result.error_message = e.message
recorder = UsageRecorder(
db=db,
user=user,
api_key=api_key,
client_ip=client_ip,
request_id=request_id,
)
await recorder.record_failure(result, original_headers, original_request_body)
if isinstance(e, UpstreamClientException):
error_type = "invalid_request_error"
elif result.status_code == 503:
error_type = "internal_server_error"
else:
error_type = "rate_limit_exceeded"
return self._error_response(
status_code=result.status_code,
error_type=error_type,
message=result.error_message or str(e),
)
async def _handle_unexpected_exception(
self,
e: Exception,
*,
db: Session,
user: Any,
api_key: Any,
model: str,
stream: bool,
start_time: float,
original_headers: dict[str, str],
original_request_body: dict[str, Any],
client_ip: str,
request_id: str,
) -> JSONResponse:
"""处理未预期的异常"""
if isinstance(e, ProxyException):
logger.error("{} 请求处理业务异常: {}: {}", self.FORMAT_ID, type(e).__name__, e)
else:
logger.opt(exception=e).error(
"{} 请求处理意外异常: {}: {}", self.FORMAT_ID, type(e).__name__, e
)
response_time = int((time.time() - start_time) * 1000)
result = RequestResult.from_exception(
exception=e,
api_format=self.FORMAT_ID,
model=model,
response_time_ms=response_time,
is_stream=stream,
)
result.status_code = 500
result.error_type = "internal_error"
result.request_headers = original_headers
result.request_body = original_request_body
try:
recorder = UsageRecorder(
db=db,
user=user,
api_key=api_key,
client_ip=client_ip,
request_id=request_id,
)
await recorder.record_failure(result, original_headers, original_request_body)
except Exception as record_error:
logger.error("记录失败请求时出错: {}", record_error)
return self._error_response(
status_code=500, error_type="internal_server_error", message="处理请求时发生内部错误"
)
def _error_response(self, status_code: int, error_type: str, message: str) -> JSONResponse:
"""生成错误响应 - 子类可覆盖以自定义格式"""
return JSONResponse(
status_code=status_code,
content={
"error": {
"type": error_type,
"message": message,
}
},
)
# =========================================================================
# 计费策略
# =========================================================================
def compute_total_input_context(
self,
input_tokens: int,
cache_read_input_tokens: int,
cache_creation_input_tokens: int = 0,
) -> int:
"""计算总输入上下文（用于阶梯计费判定）- 子类可覆盖"""
return input_tokens + cache_read_input_tokens
def compute_cost(
self,
input_tokens: int,
output_tokens: int,
cache_creation_input_tokens: int,
cache_read_input_tokens: int,
input_price_per_1m: float,
output_price_per_1m: float,
cache_creation_price_per_1m: float | None,
cache_read_price_per_1m: float | None,
price_per_request: float | None,
tiered_pricing: dict | None = None,
cache_ttl_minutes: int | None = None,
) -> dict[str, Any]:
"""计算请求成本"""
total_input_context = self.compute_total_input_context(
input_tokens, cache_read_input_tokens, cache_creation_input_tokens
)
return _calculate_request_cost(
input_tokens=input_tokens,
output_tokens=output_tokens,
cache_creation_input_tokens=cache_creation_input_tokens,
cache_read_input_tokens=cache_read_input_tokens,
input_price_per_1m=input_price_per_1m,
output_price_per_1m=output_price_per_1m,
cache_creation_price_per_1m=cache_creation_price_per_1m,
cache_read_price_per_1m=cache_read_price_per_1m,
price_per_request=price_per_request,
tiered_pricing=tiered_pricing,
cache_ttl_minutes=cache_ttl_minutes,
total_input_context=total_input_context,
billing_template=self.BILLING_TEMPLATE,
)
# =========================================================================
# 模型列表查询与端点测试
# =========================================================================
@classmethod
async def fetch_models(
cls,
client: httpx.AsyncClient,
base_url: str,
api_key: str,
extra_headers: dict[str, str] | None = None,
) -> tuple[list, str | None]:
"""查询上游 API 支持的模型列表 - 子类应覆盖"""
return [], f"{cls.FORMAT_ID} adapter does not implement fetch_models"
@classmethod
def build_request_body(
cls,
request_data: dict[str, Any] | None = None,
*,
base_url: str | None = None,
) -> dict[str, Any]:
"""构建测试请求体，使用转换器注册表自动处理格式转换"""
from src.api.handlers.base.request_builder import build_test_request_body
_ = base_url
return build_test_request_body(cls.FORMAT_ID, request_data)
@classmethod
async def check_endpoint(
cls,
client: httpx.AsyncClient,
base_url: str,
api_key: str,
request_data: dict[str, Any],
extra_headers: dict[str, str] | None = None,
# 端点规则参数
body_rules: list[dict[str, Any]] | None = None,
header_rules: list[dict[str, Any]] | None = None,
# 用量计算参数
db: Any | None = None,
user: Any | None = None,
provider_name: str | None = None,
provider_id: str | None = None,
api_key_id: str | None = None,
model_name: str | None = None,
# Provider 上下文（用于 OAuth 认证和特殊路由）
auth_type: str | None = None,
provider_type: str | None = None,
decrypted_auth_config: dict[str, Any] | None = None,
# 代理参数
proxy_param: Any | None = None,
) -> dict[str, Any]:
"""
测试模型连接性（非流式）
统一的 endpoint 测试方法，支持 OAuth/Antigravity/Kiro 等特殊路由。
"""
from src.api.handlers.base.endpoint_checker import run_endpoint_check
from src.api.handlers.base.request_builder import apply_body_rules
from src.core.api_format.headers import HeaderBuilder
from src.core.provider_types import ProviderType
is_antigravity = provider_type == ProviderType.ANTIGRAVITY
is_kiro = provider_type == ProviderType.KIRO
is_oauth = auth_type == "oauth"
# ---- URL ----
if is_kiro:
from src.services.provider.adapters.kiro.constants import (
DEFAULT_REGION,
KIRO_GENERATE_ASSISTANT_PATH,
)
region = (decrypted_auth_config or {}).get("region") or DEFAULT_REGION
effective_base_url = (
base_url.replace("{region}", region) if "{region}" in base_url else base_url
)
url = f"{str(effective_base_url).rstrip('/')}{KIRO_GENERATE_ASSISTANT_PATH}"
elif is_antigravity:
from src.services.provider.adapters.antigravity.constants import (
V1INTERNAL_PATH_TEMPLATE,
)
from src.services.provider.adapters.antigravity.constants import (
get_http_user_agent as _get_antigravity_ua,
)
from src.services.provider.adapters.antigravity.url_availability import (
url_availability,
)
ordered_urls = url_availability.get_ordered_urls(prefer_daily=True)
effective_base_url = ordered_urls[0] if ordered_urls else base_url
path = V1INTERNAL_PATH_TEMPLATE.format(action="generateContent")
url = f"{str(effective_base_url).rstrip('/')}{path}"
else:
url = cls.build_endpoint_url(base_url, request_data, model_name)
# ---- Headers ----
cli_extra = cls.get_cli_extra_headers(base_url=base_url)
merged_extra = dict(extra_headers) if extra_headers else {}
merged_extra.update(cli_extra)
if is_antigravity:
merged_extra["User-Agent"] = _get_antigravity_ua()
if is_kiro:
from src.services.provider.adapters.kiro.headers import (
build_generate_assistant_headers,
)
from src.services.provider.adapters.kiro.models.credentials import KiroAuthConfig
from src.services.provider.adapters.kiro.token_manager import generate_machine_id
kiro_cfg = KiroAuthConfig.from_dict(decrypted_auth_config or {})
region = kiro_cfg.region or DEFAULT_REGION
machine_id = generate_machine_id(kiro_cfg)
kiro_headers = build_generate_assistant_headers(
host=f"q.{region}.amazonaws.com",
access_token=api_key,
machine_id=machine_id,
kiro_version=kiro_cfg.kiro_version,
system_version=kiro_cfg.system_version,
node_version=kiro_cfg.node_version,
)
merged_extra.update(kiro_headers)
headers = cls.build_headers_with_extra(api_key, merged_extra if merged_extra else None)
if is_oauth:
from src.core.api_format import get_auth_config_for_endpoint
default_auth_header, _ = get_auth_config_for_endpoint(cls.FORMAT_ID)
if default_auth_header.lower() != "authorization":
headers.pop(default_auth_header, None)
headers["Authorization"] = f"Bearer {api_key}"
# ---- Body ----
body = cls.build_request_body(request_data, base_url=base_url)
if body_rules:
body = apply_body_rules(body, body_rules)
if is_antigravity:
from src.services.provider.adapters.antigravity.envelope import (
wrap_v1internal_request,
)
project_id = (decrypted_auth_config or {}).get("project_id", "")
effective_model = model_name or request_data.get("model", "")
body = wrap_v1internal_request(
body,
project_id=project_id,
model=effective_model,
)
if is_kiro:
from src.services.provider.adapters.kiro.converter import (
convert_claude_messages_to_conversation_state,
)
effective_model = model_name or request_data.get("model", "")
conversation_state = convert_claude_messages_to_conversation_state(
body,
model=effective_model,
)
body = {"conversationState": conversation_state}
if isinstance(kiro_cfg.profile_arn, str) and kiro_cfg.profile_arn.strip():
body["profileArn"] = kiro_cfg.profile_arn.strip()
# ---- Header Rules ----
if header_rules:
from src.core.api_format import get_auth_config_for_endpoint as _get_auth_cfg
if is_oauth:
protected_keys = {"authorization", "content-type"}
else:
ep_auth_header, _ = _get_auth_cfg(cls.FORMAT_ID)
protected_keys = {ep_auth_header.lower(), "content-type"}
header_builder = HeaderBuilder()
header_builder.add_many(headers)
header_builder.apply_rules(header_rules, protected_keys)
headers = header_builder.build()
# ---- Execute ----
effective_model_name = model_name or request_data.get("model")
return await run_endpoint_check(
client=client,
url=url,
headers=headers,
json_body=body,
api_format=cls.FORMAT_ID,
db=db,
user=user,
provider_name=provider_name,
provider_id=provider_id,
api_key_id=api_key_id,
model_name=effective_model_name,
proxy_param=proxy_param,
)
# =========================================================================
# CLI Adapter 配置方法 - 子类可覆盖
# =========================================================================
@classmethod
def build_endpoint_url(
cls,
base_url: str,
request_data: dict[str, Any] | None = None,
model_name: str | None = None,
) -> str:
"""构建 API 端点 URL - 子类应覆盖"""
return base_url
@classmethod
def get_cli_user_agent(cls) -> str | None:
"""获取 CLI User-Agent - 子类可覆盖"""
return None
@classmethod
def get_cli_extra_headers(cls, *, base_url: str | None = None) -> dict[str, str]:
"""获取额外请求头 - 子类可覆盖"""
headers: dict[str, str] = {}
cli_user_agent = cls.get_cli_user_agent()
if cli_user_agent:
headers["User-Agent"] = cli_user_agent
return headers

24
src/api/handlers/base/parsers.py
View File

@@ -525,17 +525,6 @@ class ClaudeResponseParser(ResponseParser):
return is_error
class ClaudeCliResponseParser(ClaudeResponseParser):
"""Claude CLI 格式响应解析器"""
API_FORMAT = "claude:cli"
def __init__(self) -> None:
super().__init__()
self.name = self.API_FORMAT
self.api_format = self.API_FORMAT
class GeminiResponseParser(ResponseParser):
"""Gemini 格式响应解析器"""
@@ -694,17 +683,6 @@ class GeminiResponseParser(ResponseParser):
return bool(self._parser.is_error_event(response))
class GeminiCliResponseParser(GeminiResponseParser):
"""Gemini CLI 格式响应解析器"""
API_FORMAT = "gemini:cli"
def __init__(self) -> None:
super().__init__()
self.name = self.API_FORMAT
self.api_format = self.API_FORMAT
# 注册解析器到 core 层注册表（供 services 层通过 format_id 获取）
from src.core.stream_types import get_parser_for_format, register_parser
@@ -735,9 +713,7 @@ __all__ = [
"OpenAIResponseParser",
"OpenAICliResponseParser",
"ClaudeResponseParser",
"ClaudeCliResponseParser",
"GeminiResponseParser",
"GeminiCliResponseParser",
"register_default_parsers",
"get_parser_for_format",
"is_cli_format",

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

@@ -58,6 +58,8 @@ class StreamContext:
# 请求基本信息
model: str
api_format: str
api_family: str | None = None # 协议族（从 Adapter 层透传）
endpoint_kind: str | None = None # 端点类型（从 Adapter 层透传）
# 请求标识信息（CLI handler 需要）
request_id: str = ""

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

@@ -229,6 +229,8 @@ class StreamTelemetryRecorder:
is_stream=True,
provider_request_headers=ctx.provider_request_headers,
api_format=ctx.api_format,
api_family=ctx.api_family,
endpoint_kind=ctx.endpoint_kind,
provider_id=ctx.provider_id,
provider_endpoint_id=ctx.endpoint_id,
provider_api_key_id=ctx.key_id,
@@ -273,6 +275,8 @@ class StreamTelemetryRecorder:
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
api_family=ctx.api_family,
endpoint_kind=ctx.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
input_tokens=ctx.input_tokens,
@@ -328,6 +332,8 @@ class StreamTelemetryRecorder:
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
api_family=ctx.api_family,
endpoint_kind=ctx.endpoint_kind,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
input_tokens=ctx.input_tokens,

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

@@ -248,7 +248,12 @@ class ClaudeChatAdapter(ChatAdapterBase):
return [], error_msg
@classmethod
def build_endpoint_url(cls, base_url: str) -> str:
def build_endpoint_url(
cls,
base_url: str,
request_data: dict[str, Any] | None = None,
model_name: str | None = None,
) -> str:
"""构建Claude API端点URL"""
base_url = base_url.rstrip("/")
if base_url.endswith("/v1"):

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

@@ -237,7 +237,12 @@ class GeminiChatAdapter(ChatAdapterBase):
return [], error_msg
@classmethod
def build_endpoint_url(cls, base_url: str) -> str:
def build_endpoint_url(
cls,
base_url: str,
request_data: dict[str, Any] | None = None,
model_name: str | None = None,
) -> str:
"""构建Gemini API端点URL"""
base_url = base_url.rstrip("/")
if base_url.endswith("/v1beta"):
@@ -292,6 +297,8 @@ class GeminiChatAdapter(ChatAdapterBase):
if is_antigravity:
from src.services.provider.adapters.antigravity.constants import (
V1INTERNAL_PATH_TEMPLATE,
)
from src.services.provider.adapters.antigravity.constants import (
get_http_user_agent as _get_antigravity_ua,
)
from src.services.provider.adapters.antigravity.url_availability import url_availability

7
src/api/handlers/openai/adapter.py
View File

@@ -148,7 +148,12 @@ class OpenAIChatAdapter(ChatAdapterBase):
return [], error_msg
@classmethod
def build_endpoint_url(cls, base_url: str) -> str:
def build_endpoint_url(
cls,
base_url: str,
request_data: dict[str, Any] | None = None,
model_name: str | None = None,
) -> str:
"""构建OpenAI API端点URL"""
base_url = base_url.rstrip("/")
if base_url.endswith("/v1"):

4
src/api/handlers/openai/video_handler.py
View File

@@ -982,6 +982,8 @@ class OpenAIVideoHandler(VideoHandlerBase):
cache_creation_input_tokens=0,
cache_read_input_tokens=0,
api_format=self.FORMAT_ID,
api_family=self.API_FAMILY.value if self.API_FAMILY else None,
endpoint_kind=self.ENDPOINT_KIND.value if self.ENDPOINT_KIND else None,
endpoint_api_format=None,
has_format_conversion=False,
is_stream=False,
@@ -1098,6 +1100,8 @@ class OpenAIVideoHandler(VideoHandlerBase):
cache_creation_input_tokens=0,
cache_read_input_tokens=0,
api_format=self.FORMAT_ID,
api_family=self.API_FAMILY.value if self.API_FAMILY else None,
endpoint_kind=self.ENDPOINT_KIND.value if self.ENDPOINT_KIND else None,
endpoint_api_format=None,
has_format_conversion=False,
is_stream=False,

17
src/core/api_format/conversion/normalizers/claude_cli.py
View File

@@ -1,17 +0,0 @@
"""
Claude CLI Normalizer
CLAUDE_CLI 的请求/响应 body 与 CLAUDE 一致（Anthropic Messages API），差异主要在认证头。
因此这里复用 ClaudeNormalizer 的转换逻辑，仅更换 FORMAT_ID。
如需 CLI 特殊处理，可覆盖 request_from_internal / request_to_internal 等方法。
"""
from src.core.api_format.conversion.normalizers.claude import ClaudeNormalizer
class ClaudeCliNormalizer(ClaudeNormalizer):
FORMAT_ID = "claude:cli"
__all__ = ["ClaudeCliNormalizer"]

17
src/core/api_format/conversion/normalizers/gemini_cli.py
View File

@@ -1,17 +0,0 @@
"""
Gemini CLI Normalizer
GEMINI_CLI 的请求/响应 body 与 GEMINI 一致（Google Gemini API），差异主要在鉴权/UA 等请求层。
因此这里复用 GeminiNormalizer 的转换逻辑，仅更换 FORMAT_ID。
如需 CLI 特殊处理，可覆盖 request_from_internal / request_to_internal 等方法。
"""
from src.core.api_format.conversion.normalizers.gemini import GeminiNormalizer
class GeminiCliNormalizer(GeminiNormalizer):
FORMAT_ID = "gemini:cli"
__all__ = ["GeminiCliNormalizer"]

17
src/core/api_format/conversion/registry.py
View File

@@ -54,7 +54,22 @@ class FormatConversionRegistry:
logger.info(f"[FormatConversionRegistry] 注册 normalizer: {normalizer.FORMAT_ID}")
def get_normalizer(self, format_id: str) -> FormatNormalizer | None:
return self._normalizers.get(str(format_id).upper())
key = str(format_id).upper()
# 1. 精确匹配
normalizer = self._normalizers.get(key)
if normalizer is not None:
return normalizer
# 2. data_format_id 回退：如 "claude:cli" (dfid="claude") -> ClaudeNormalizer (dfid="claude")
from src.core.api_format.metadata import get_data_format_id_for_endpoint
target_dfid = get_data_format_id_for_endpoint(format_id)
if not target_dfid:
return None
for reg_key, reg_normalizer in self._normalizers.items():
reg_dfid = get_data_format_id_for_endpoint(reg_key)
if reg_dfid == target_dfid:
return reg_normalizer
return None
def _require_normalizer(self, format_id: str) -> FormatNormalizer:
normalizer = self.get_normalizer(format_id)

16
src/core/stream_types.py
View File

@@ -219,6 +219,16 @@ def get_parser_for_format(format_id: str) -> ResponseParser:
)
normalized = normalize_signature_key(format_id)
if normalized not in _PARSER_REGISTRY:
raise KeyError(f"Unknown format: {normalized}")
return _PARSER_REGISTRY[normalized]()
# 1. 精确匹配
if normalized in _PARSER_REGISTRY:
return _PARSER_REGISTRY[normalized]()
# 2. data_format_id 回退：如 "claude:cli" (dfid="claude") -> ClaudeResponseParser (dfid="claude")
from src.core.api_format.metadata import get_data_format_id_for_endpoint
target_dfid = get_data_format_id_for_endpoint(normalized)
if target_dfid:
for reg_key, parser_cls in _PARSER_REGISTRY.items():
reg_dfid = get_data_format_id_for_endpoint(reg_key)
if reg_dfid == target_dfid:
return parser_cls()
raise KeyError(f"Unknown format: {normalized}")

4
src/models/database.py
View File

@@ -366,7 +366,11 @@ class Usage(Base):
# 请求详情
request_type = Column(String(50)) # chat, completion, embedding等
api_format = Column(String(50), nullable=True) # API 格式: CLAUDE, OPENAI 等（用户请求格式）
api_family = Column(String(50), nullable=True) # 协议族: claude, openai, gemini
endpoint_kind = Column(String(50), nullable=True) # 端点类型: chat, cli, video
endpoint_api_format = Column(String(50), nullable=True) # 端点原生 API 格式
provider_api_family = Column(String(50), nullable=True) # 提供商协议族
provider_endpoint_kind = Column(String(50), nullable=True) # 提供商端点类型
has_format_conversion = Column(Boolean, nullable=True, default=False) # 是否发生了格式转换
is_stream = Column(Boolean, default=False) # 是否为流式请求
status_code = Column(Integer)

4
src/models/usage.py
View File

@@ -100,7 +100,11 @@ class Usage(Base):
# 请求详情
request_type = Column(String(50)) # chat, completion, embedding等
api_format = Column(String(50), nullable=True) # API 格式: CLAUDE, OPENAI 等（用户请求格式）
api_family = Column(String(50), nullable=True) # 协议族: claude, openai, gemini
endpoint_kind = Column(String(50), nullable=True) # 端点类型: chat, cli, video
endpoint_api_format = Column(String(50), nullable=True) # 端点原生 API 格式
provider_api_family = Column(String(50), nullable=True) # 提供商协议族
provider_endpoint_kind = Column(String(50), nullable=True) # 提供商端点类型
has_format_conversion = Column(Boolean, nullable=True, default=False) # 是否发生了格式转换
is_stream = Column(Boolean, default=False) # 是否为流式请求
status_code = Column(Integer)

10
src/services/request/result.py
View File

@@ -52,6 +52,10 @@ class RequestMetadata:
provider: str = "unknown"
model: str = "unknown"
# 结构化格式维度（从 Adapter 层透传，优先于从 api_format 字符串解析）
api_family: str | None = None # 协议族: claude, openai, gemini
endpoint_kind: str | None = None # 端点类型: chat, cli, video
# Provider 追踪信息
provider_id: str | None = None
provider_endpoint_id: str | None = None
@@ -80,6 +84,8 @@ class RequestMetadata:
api_format=self.api_format,
provider=provider,
model=self.model,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
provider_id=provider_id,
provider_endpoint_id=provider_endpoint_id,
provider_api_key_id=provider_api_key_id,
@@ -96,6 +102,8 @@ class RequestMetadata:
api_format=self.api_format,
provider=self.provider,
model=self.model,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
provider_id=self.provider_id,
provider_endpoint_id=self.provider_endpoint_id,
provider_api_key_id=self.provider_api_key_id,
@@ -274,6 +282,8 @@ class RequestResult:
api_format=get_meta_value(existing_metadata, "api_format") or api_format,
provider=get_meta_value(existing_metadata, "provider", "unknown") or "unknown",
model=get_meta_value(existing_metadata, "model", model) or model,
api_family=get_meta_value(existing_metadata, "api_family"),
endpoint_kind=get_meta_value(existing_metadata, "endpoint_kind"),
provider_id=get_meta_value(existing_metadata, "provider_id"),
provider_endpoint_id=get_meta_value(existing_metadata, "provider_endpoint_id"),
provider_api_key_id=get_meta_value(existing_metadata, "provider_api_key_id"),

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

@@ -135,6 +135,8 @@ class UsageBillingIntegrationMixin:
cache_read_input_tokens=params.cache_read_input_tokens,
request_type=params.request_type,
api_format=params.api_format,
api_family=params.api_family,
endpoint_kind=params.endpoint_kind,
endpoint_api_format=params.endpoint_api_format,
has_format_conversion=params.has_format_conversion,
is_stream=params.is_stream,

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

@@ -10,6 +10,17 @@ from src.services.system.config import SystemConfigService
from src.services.usage._types import UsageCostInfo
from src.services.usage.error_classifier import classify_error
def _parse_format_dimensions(api_format: str | None) -> tuple[str | None, str | None]:
"""从 api_format (如 'claude:chat') 解析出 (api_family, endpoint_kind)"""
if not api_format:
return None, None
parts = api_format.lower().split(":", 1)
if len(parts) == 2:
return parts[0], parts[1]
return parts[0], None
# Metadata pruning configuration (ordered by priority - drop first to last)
METADATA_PRUNE_KEYS: tuple[str, ...] = (
"raw_response_ref",
@@ -45,7 +56,11 @@ def build_usage_params(
cache_read_input_tokens: int,
request_type: str,
api_format: str | None,
api_family: str | None = None,
endpoint_kind: str | None = None,
endpoint_api_format: str | None,
provider_api_family: str | None = None,
provider_endpoint_kind: str | None = None,
has_format_conversion: bool,
is_stream: bool,
response_time_ms: int | None,
@@ -160,6 +175,16 @@ def build_usage_params(
if status_code >= 400 or error_message or status in {"failed", "cancelled"}:
error_category = classify_error(status_code, error_message, status).value
# 从 api_format / endpoint_api_format 解析 api_family + endpoint_kind
# 优先使用透传值，fallback 到字符串解析
parsed_family, parsed_kind = _parse_format_dimensions(api_format)
client_family = api_family or parsed_family
client_kind = endpoint_kind or parsed_kind
parsed_ep_family, parsed_ep_kind = _parse_format_dimensions(endpoint_api_format)
ep_family = provider_api_family or parsed_ep_family
ep_kind = provider_endpoint_kind or parsed_ep_kind
return {
"user_id": user.id if user else None,
"api_key_id": api_key.id if api_key else None,
@@ -196,7 +221,11 @@ def build_usage_params(
"price_per_request": request_price,
"request_type": request_type,
"api_format": api_format,
"api_family": client_family,
"endpoint_kind": client_kind,
"endpoint_api_format": endpoint_api_format,
"provider_api_family": ep_family,
"provider_endpoint_kind": ep_kind,
"has_format_conversion": has_format_conversion,
"is_stream": is_stream,
"status_code": status_code,
@@ -228,7 +257,11 @@ def update_existing_usage(
existing_usage.model = usage_params["model"]
existing_usage.request_type = usage_params["request_type"]
existing_usage.api_format = usage_params["api_format"]
existing_usage.api_family = usage_params.get("api_family")
existing_usage.endpoint_kind = usage_params.get("endpoint_kind")
existing_usage.endpoint_api_format = usage_params["endpoint_api_format"]
existing_usage.provider_api_family = usage_params.get("provider_api_family")
existing_usage.provider_endpoint_kind = usage_params.get("provider_endpoint_kind")
existing_usage.has_format_conversion = usage_params["has_format_conversion"]
existing_usage.is_stream = usage_params["is_stream"]
existing_usage.status = usage_params["status"]

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

@@ -1,6 +1,6 @@
from __future__ import annotations
from dataclasses import dataclass, field
from dataclasses import dataclass
from typing import Any
from sqlalchemy.orm import Session
@@ -23,6 +23,8 @@ class UsageRecordParams:
cache_read_input_tokens: int
request_type: str
api_format: str | None
api_family: str | None # 协议族（从 Adapter 层透传）
endpoint_kind: str | None # 端点类型（从 Adapter 层透传）
endpoint_api_format: str | None # 端点原生 API 格式
has_format_conversion: bool # 是否发生了格式转换
is_stream: bool

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

@@ -70,6 +70,8 @@ def _event_to_record(event: UsageEvent) -> dict[str, Any]:
"cache_read_input_tokens": data.get("cache_read_input_tokens") or 0,
"request_type": data.get("request_type") or "chat",
"api_format": data.get("api_format"),
"api_family": data.get("api_family"),
"endpoint_kind": data.get("endpoint_kind"),
"endpoint_api_format": data.get("endpoint_api_format"),
"has_format_conversion": data.get("has_format_conversion"),
"is_stream": data.get("is_stream", True),

4
src/services/usage/recorder.py
View File

@@ -112,6 +112,8 @@ class UsageRecorder:
cache_read_input_tokens=usage.cache_read_input_tokens,
request_type="chat",
api_format=metadata.api_format,
api_family=metadata.api_family,
endpoint_kind=metadata.endpoint_kind,
is_stream=result.is_stream,
response_time_ms=result.response_time_ms,
status_code=200,
@@ -181,6 +183,8 @@ class UsageRecorder:
output_tokens=0,
request_type="chat",
api_format=metadata.api_format,
api_family=metadata.api_family,
endpoint_kind=metadata.endpoint_kind,
is_stream=result.is_stream,
response_time_ms=result.response_time_ms,
status_code=result.status_code,

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

@@ -67,6 +67,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_read_input_tokens: int = 0,
request_type: str = "chat",
api_format: str | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
endpoint_api_format: str | None = None,
has_format_conversion: bool = False,
is_stream: bool = False,
@@ -116,6 +118,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_read_input_tokens=cache_read_input_tokens,
request_type=request_type,
api_format=api_format,
api_family=api_family,
endpoint_kind=endpoint_kind,
endpoint_api_format=endpoint_api_format,
has_format_conversion=has_format_conversion,
is_stream=is_stream,
@@ -189,6 +193,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_read_input_tokens: int = 0,
request_type: str = "chat",
api_format: str | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
endpoint_api_format: str | None = None,
has_format_conversion: bool = False,
is_stream: bool = False,
@@ -240,6 +246,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_read_input_tokens=cache_read_input_tokens,
request_type=request_type,
api_format=api_format,
api_family=api_family,
endpoint_kind=endpoint_kind,
endpoint_api_format=endpoint_api_format,
has_format_conversion=has_format_conversion,
is_stream=is_stream,
@@ -380,6 +388,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_creation_input_tokens: int = 0,
cache_read_input_tokens: int = 0,
api_format: str | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
endpoint_api_format: str | None = None,
has_format_conversion: bool = False,
is_stream: bool = False,
@@ -443,6 +453,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_read_input_tokens=cache_read_input_tokens,
request_type=request_type,
api_format=api_format,
api_family=api_family,
endpoint_kind=endpoint_kind,
endpoint_api_format=endpoint_api_format,
has_format_conversion=has_format_conversion,
is_stream=is_stream,
@@ -744,6 +756,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
cache_read_input_tokens=int(record.get("cache_read_input_tokens") or 0),
request_type=record.get("request_type") or "chat",
api_format=record.get("api_format"),
api_family=record.get("api_family"),
endpoint_kind=record.get("endpoint_kind"),
endpoint_api_format=record.get("endpoint_api_format"),
has_format_conversion=bool(record.get("has_format_conversion")),
is_stream=bool(record.get("is_stream", True)),

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

@@ -42,6 +42,9 @@ class StreamUsageTracker:
provider_api_key_id: str | None = None,
# API 格式（用于选择正确的响应解析器）
api_format: str | None = None,
# 结构化格式维度（从 Adapter 层透传）
api_family: str | None = None,
endpoint_kind: str | None = None,
# 格式转换信息
endpoint_api_format: str | None = None,
has_format_conversion: bool = False,
@@ -85,6 +88,8 @@ class StreamUsageTracker:
# API 格式和响应解析器
self.api_format = api_format or "claude:chat"
self.api_family = api_family
self.endpoint_kind = endpoint_kind
self.endpoint_api_format = endpoint_api_format
self.has_format_conversion = has_format_conversion
self.response_parser = get_parser_for_format(self.api_format)
@@ -759,6 +764,8 @@ class StreamUsageTracker:
cache_read_input_tokens=self.cache_read_input_tokens,
request_type="chat",
api_format=self.api_format,
api_family=self.api_family,
endpoint_kind=self.endpoint_kind,
endpoint_api_format=self.endpoint_api_format,
has_format_conversion=self.has_format_conversion,
is_stream=True,
@@ -854,6 +861,9 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
provider_api_key_id: str | None = None,
# API 格式（用于选择正确的响应解析器）
api_format: str | None = None,
# 结构化格式维度（从 Adapter 层透传）
api_family: str | None = None,
endpoint_kind: str | None = None,
# 格式转换信息
endpoint_api_format: str | None = None,
has_format_conversion: bool = False,
@@ -873,6 +883,8 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
provider_endpoint_id,
provider_api_key_id,
api_format,
api_family,
endpoint_kind,
endpoint_api_format,
has_format_conversion,
)
@@ -1126,6 +1138,9 @@ def create_stream_tracker(
provider_api_key_id: str | None = None,
# API 格式（用于选择正确的响应解析器）
api_format: str | None = None,
# 结构化格式维度（从 Adapter 层透传）
api_family: str | None = None,
endpoint_kind: str | None = None,
# 格式转换信息
endpoint_api_format: str | None = None,
has_format_conversion: bool = False,
@@ -1171,6 +1186,8 @@ def create_stream_tracker(
provider_endpoint_id,
provider_api_key_id,
api_format,
api_family,
endpoint_kind,
endpoint_api_format,
has_format_conversion,
)
@@ -1190,6 +1207,8 @@ def create_stream_tracker(
provider_endpoint_id,
provider_api_key_id,
api_format,
api_family,
endpoint_kind,
endpoint_api_format,
has_format_conversion,
)

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

@@ -81,6 +81,9 @@ class MessageTelemetry:
provider_endpoint_id: str | None = None,
provider_api_key_id: str | None = None,
api_format: str | None = None,
# 结构化格式维度（从 Adapter 层透传）
api_family: str | None = None,
endpoint_kind: str | None = None,
# 格式转换追踪
endpoint_api_format: str | None = None, # 端点原生 API 格式
has_format_conversion: bool = False, # 是否发生了格式转换
@@ -110,6 +113,8 @@ class MessageTelemetry:
cache_read_input_tokens=cache_read_tokens,
request_type="chat",
api_format=api_format,
api_family=api_family,
endpoint_kind=endpoint_kind,
endpoint_api_format=endpoint_api_format,
has_format_conversion=has_format_conversion,
is_stream=is_stream,
@@ -167,6 +172,8 @@ class MessageTelemetry:
request_headers: dict[str, Any],
is_stream: bool,
api_format: str | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
# 预估 token 信息（来自 message_start 事件，用于中断请求的成本估算）
@@ -222,6 +229,8 @@ class MessageTelemetry:
cache_read_input_tokens=cache_read_tokens,
request_type="chat",
api_format=api_format,
api_family=api_family,
endpoint_kind=endpoint_kind,
endpoint_api_format=endpoint_api_format,
has_format_conversion=has_format_conversion,
is_stream=is_stream,
@@ -259,6 +268,8 @@ class MessageTelemetry:
request_headers: dict[str, Any],
is_stream: bool,
api_format: str | None = None,
api_family: str | None = None,
endpoint_kind: str | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
input_tokens: int = 0,
@@ -299,6 +310,8 @@ class MessageTelemetry:
cache_read_input_tokens=cache_read_tokens,
request_type="chat",
api_format=api_format,
api_family=api_family,
endpoint_kind=endpoint_kind,
endpoint_api_format=endpoint_api_format,
has_format_conversion=has_format_conversion,
is_stream=is_stream,

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

@@ -225,6 +225,10 @@ class QueueTelemetryWriter(TelemetryWriter):
data["request_type"] = request_type
if kwargs.get("api_format"):
data["api_format"] = kwargs["api_format"]
if kwargs.get("api_family"):
data["api_family"] = kwargs["api_family"]
if kwargs.get("endpoint_kind"):
data["endpoint_kind"] = kwargs["endpoint_kind"]
if kwargs.get("endpoint_api_format"):
data["endpoint_api_format"] = kwargs["endpoint_api_format"]
if kwargs.get("has_format_conversion"):