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

feat: Antigravity 端点签名迁移至 gemini:chat，流式 usage 提取重构与请求详情自动刷新

Browse Source 
- Antigravity 端点签名从 gemini:cli 统一为 gemini:chat，保留向后兼容，含 DB 迁移
- 流式处理中 usage/completion/text 提取抽离为 _update_ctx_from_provider_event，
  支持 envelope 解包后提取，避免格式转换流程中重复计数
- 新增 has_format_conversion 属性，区分真正的格式转换与 envelope rewrite，
  修正 usage 展示层的格式转换标记
- 请求详情抽屉对未完成请求支持自动轮询刷新，关闭时自动停止
- 按次计费样式调整；实际成本仅在倍率非 1.0 时显示；缓存日志降级为 trace
This commit is contained in:
fawney19
2026-02-07 02:59:02 +08:00
parent ea22f07046
commit 62f852b851
24 changed files with 481 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/api/admin/endpoints/keys.py
View File

@@ -1009,7 +1009,7 @@ class AdminRefreshProviderQuotaAdapter(AdminApiAdapter):
# 获取端点：
# - Codex: openai:cli
# - Antigravity: gemini:cli（用于触发 oauth 刷新 + 提供 auth_config.project_id）
# - Antigravity: gemini:chat（用于触发 oauth 刷新 + 提供 auth_config.project_id）
endpoint = None
if provider_type == ProviderType.CODEX:
for ep in provider.endpoints:
@@ -1019,12 +1019,16 @@ class AdminRefreshProviderQuotaAdapter(AdminApiAdapter):
if not endpoint:
raise InvalidRequestException("找不到有效的 openai:cli 端点")
else:
for ep in provider.endpoints:
if ep.api_format == "gemini:cli" and ep.is_active:
endpoint = ep
# Prefer the new signature, but keep backward-compat with existing DB rows.
for sig in ("gemini:chat", "gemini:cli"):
for ep in provider.endpoints:
if ep.api_format == sig and ep.is_active:
endpoint = ep
break
if endpoint is not None:
break
if not endpoint:
raise InvalidRequestException("找不到有效的 gemini:cli 端点")
raise InvalidRequestException("找不到有效的 gemini:chat/gemini:cli 端点")
results: list[dict] = []
success_count = 0

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

@@ -70,7 +70,7 @@ async def _resolve_key_auth(
auth_config: dict[str, Any] | None = None
if auth_type == "oauth":
endpoint_api_format = "gemini:cli" if provider_type == ProviderType.ANTIGRAVITY else None
endpoint_api_format = "gemini:chat" if provider_type == ProviderType.ANTIGRAVITY else None
try:
resolved = await resolve_oauth_access_token(
key_id=str(api_key.id),

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

@@ -1181,6 +1181,7 @@ class AdminUsageDetailAdapter(AdminApiAdapter):
"is_stream": usage_record.is_stream,
"status_code": usage_record.status_code,
"error_message": usage_record.error_message,
"status": usage_record.status,
"response_time_ms": usage_record.response_time_ms,
"first_byte_time_ms": usage_record.first_byte_time_ms, # 首字时间 (TTFB)
"created_at": usage_record.created_at.isoformat() if usage_record.created_at else None,

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

@@ -571,7 +571,7 @@ class BaseMessageHandler:
api_format = ctx.api_format
# 格式转换追踪
endpoint_api_format = ctx.provider_api_format or None
has_format_conversion = ctx.needs_conversion
has_format_conversion = ctx.has_format_conversion
# 如果 provider 为空，记录警告（不应该发生，但用于调试）
if not provider:

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

@@ -40,7 +40,7 @@ from src.api.handlers.base.base_handler import (
from src.api.handlers.base.parsers import get_parser_for_format
from src.api.handlers.base.request_builder import PassthroughRequestBuilder, get_provider_auth
from src.api.handlers.base.response_parser import ResponseParser
from src.api.handlers.base.stream_context import StreamContext
from src.api.handlers.base.stream_context import StreamContext, is_format_converted
from src.api.handlers.base.stream_processor import StreamProcessor
from src.api.handlers.base.stream_telemetry import StreamTelemetryRecorder
from src.api.handlers.base.upstream_stream_bridge import (
@@ -1322,7 +1322,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
client_response_headers=client_response_headers,
# 格式转换追踪
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
target_model=ctx.mapped_model,
)
@@ -1367,7 +1367,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
provider_request_body: dict[str, Any] | None = None
provider_api_format_for_error: str | None = None
client_api_format_for_error: str | None = None
needs_conversion_for_error: bool = False
needs_conversion_for_error: bool = False # 用于构建错误 payload（含 envelope rewrite）
provider_id: str | None = None # Provider ID（用于失败记录）
endpoint_id: str | None = None # Endpoint ID（用于失败记录）
key_id: str | None = None # Key ID（用于失败记录）
@@ -1795,7 +1795,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
api_format=api_format,
# 格式转换追踪
endpoint_api_format=provider_api_format_for_error or None,
has_format_conversion=needs_conversion_for_error,
has_format_conversion=is_format_converted(
provider_api_format_for_error, client_api_format_for_error
),
provider_id=provider_id,
provider_endpoint_id=endpoint_id,
provider_api_key_id=key_id,
@@ -1865,7 +1867,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
client_response_headers={"content-type": "application/json"},
# 格式转换追踪
endpoint_api_format=provider_api_format_for_error or None,
has_format_conversion=needs_conversion_for_error,
has_format_conversion=is_format_converted(
provider_api_format_for_error, client_api_format_for_error
),
target_model=mapped_model_result,
request_metadata=request_metadata,
)
@@ -1916,7 +1920,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
client_response_headers={"content-type": "application/json"},
# 格式转换追踪
endpoint_api_format=provider_api_format_for_error or None,
has_format_conversion=needs_conversion_for_error,
has_format_conversion=is_format_converted(
provider_api_format_for_error, client_api_format_for_error
),
# 模型映射信息
target_model=mapped_model_result,
request_metadata=request_metadata,

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

@@ -43,7 +43,7 @@ from src.api.handlers.base.request_builder import PassthroughRequestBuilder, get
from src.api.handlers.base.response_parser import (
ResponseParser,
)
from src.api.handlers.base.stream_context import StreamContext
from src.api.handlers.base.stream_context import StreamContext, is_format_converted
from src.api.handlers.base.upstream_stream_bridge import (
aggregate_upstream_stream_to_internal_response,
)
@@ -2459,7 +2459,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
target_model=ctx.mapped_model,
request_metadata=request_metadata,
)
@@ -2492,7 +2492,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
client_response_headers=client_response_headers,
# 格式转换追踪
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
# 模型映射信息
target_model=ctx.mapped_model,
request_metadata=request_metadata,
@@ -2552,7 +2552,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
api_format=ctx.api_format,
# 格式转换追踪
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
# Provider 侧追踪信息（用于记录真实成本）
provider_id=ctx.provider_id,
provider_endpoint_id=ctx.endpoint_id,
@@ -2709,7 +2709,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
client_response_headers=client_response_headers,
# 格式转换追踪
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
# 模型映射信息
target_model=ctx.mapped_model,
request_metadata=request_metadata,
@@ -3170,7 +3170,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
api_format=api_format,
# 格式转换追踪
endpoint_api_format=provider_api_format or None,
has_format_conversion=needs_conversion,
has_format_conversion=is_format_converted(provider_api_format, str(api_format)),
# Provider 侧追踪信息（用于记录真实成本）
provider_id=provider_id,
provider_endpoint_id=endpoint_id,
@@ -3249,7 +3249,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
client_response_headers={"content-type": "application/json"},
# 格式转换追踪
endpoint_api_format=provider_api_format or None,
has_format_conversion=needs_conversion,
has_format_conversion=is_format_converted(provider_api_format, str(api_format)),
# 模型映射信息
target_model=mapped_model_result,
request_metadata=request_metadata,
@@ -3364,7 +3364,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
first_byte_time_ms=ctx.first_byte_time_ms,
api_format=ctx.api_format,
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
)
except Exception as e:
logger.warning(f"[{self.request_id}] 同步更新 streaming 状态失败: {e}")

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

@@ -18,6 +18,18 @@ if TYPE_CHECKING:
from src.core.api_format.conversion.stream_state import StreamState
def is_format_converted(
provider_api_format: str | None,
client_api_format: str | None,
) -> bool:
"""client 与 provider 的 api_format 是否真正不同（用于 usage 展示层）"""
return bool(
provider_api_format
and client_api_format
and provider_api_format.strip().lower() != client_api_format.strip().lower()
)
@dataclass
class StreamContext:
"""
@@ -234,6 +246,15 @@ class StreamContext:
if self.first_byte_time_ms is None:
self.first_byte_time_ms = int((time.time() - start_time) * 1000)
@property
def has_format_conversion(self) -> bool:
"""是否发生了真正的格式转换（client 和 provider 的 api_format 不同）
区别于 needs_conversion：后者包含 envelope rewrite（如 Antigravity v1internal），
不代表客户端与上游的数据格式真正不同。此属性用于 usage 展示层。
"""
return is_format_converted(self.provider_api_format, self.client_api_format)
def is_success(self) -> bool:
"""检查请求是否成功"""
return self.status_code < 400

178
src/api/handlers/base/stream_processor.py
View File

@@ -108,6 +108,115 @@ class StreamProcessor:
pass
return self.default_parser
@staticmethod
def _maybe_mark_gemini_completion(ctx: StreamContext, data: dict[str, Any]) -> None:
"""Gemini: mark completion based on candidates[].finishReason."""
candidates = data.get("candidates")
if not isinstance(candidates, list) or not candidates:
return
for candidate in candidates:
if not isinstance(candidate, dict):
continue
finish_reason = candidate.get("finishReason")
if finish_reason is None:
continue
# UNSPECIFIED is the only clear "not done" sentinel across Gemini variants.
if str(finish_reason) != "FINISH_REASON_UNSPECIFIED":
ctx.has_completion = True
return
@staticmethod
def _extract_antigravity_usage_from_gemini_event(data: dict[str, Any]) -> dict[str, int] | None:
"""Antigravity: lenient Gemini usage extraction (totalTokenCount may be missing)."""
usage_metadata = data.get("usageMetadata", {})
if not isinstance(usage_metadata, dict) or not usage_metadata:
return None
def _as_int(v: Any) -> int:
try:
return int(v or 0)
except Exception:
return 0
prompt = _as_int(usage_metadata.get("promptTokenCount"))
cached = _as_int(usage_metadata.get("cachedContentTokenCount"))
candidates = _as_int(usage_metadata.get("candidatesTokenCount"))
thoughts = _as_int(usage_metadata.get("thoughtsTokenCount"))
# Align with Gemini billing convention: input_tokens includes cached content.
return {
"input_tokens": max(0, prompt),
"output_tokens": max(0, candidates + thoughts),
"cache_creation_tokens": 0,
"cache_read_tokens": max(0, cached),
}
def _unwrap_provider_envelope(self, ctx: StreamContext, data: dict[str, Any]) -> dict[str, Any]:
behavior = get_provider_behavior(
provider_type=str(getattr(ctx, "provider_type", "") or ""),
endpoint_sig=str(getattr(ctx, "provider_api_format", "") or ""),
)
envelope = behavior.envelope
if not envelope:
return data
try:
unwrapped = envelope.unwrap_response(data)
envelope.postprocess_unwrapped_response(
model=str(getattr(ctx, "model", "") or ""),
data=unwrapped,
)
return unwrapped if isinstance(unwrapped, dict) else data
except Exception:
return data
def _update_ctx_from_provider_event(
self,
ctx: StreamContext,
data: dict[str, Any],
*,
already_unwrapped: bool = False,
) -> None:
# Unwrap provider-specific envelopes (e.g. Antigravity v1internal wrapper)
if not already_unwrapped:
data = self._unwrap_provider_envelope(ctx, data)
parser = self.get_parser_for_provider(ctx)
# Provider usage extraction (best-effort)
provider_type = str(getattr(ctx, "provider_type", "") or "").lower()
provider_format = str(getattr(ctx, "provider_api_format", "") or "").strip().lower()
usage: dict[str, int] | None = None
if provider_type == "antigravity" and provider_format.startswith("gemini:"):
usage = self._extract_antigravity_usage_from_gemini_event(data)
if usage is None:
try:
usage = parser.extract_usage_from_response(data)
except Exception:
usage = None
if usage:
ctx.update_usage(
input_tokens=usage.get("input_tokens"),
output_tokens=usage.get("output_tokens"),
cached_tokens=usage.get("cache_read_tokens"),
cache_creation_tokens=usage.get("cache_creation_tokens"),
)
# Provider completion detection (Gemini doesn't emit response.completed).
if provider_format.startswith("gemini:"):
self._maybe_mark_gemini_completion(ctx, data)
# Provider text extraction (optional)
if self.collect_text:
try:
text = parser.extract_text_content(data)
except Exception:
text = ""
if text:
ctx.append_text(text)
def handle_sse_event(
self,
ctx: StreamContext,
@@ -115,6 +224,7 @@ class StreamProcessor:
data_str: str,
*,
skip_record: bool = False,
skip_ctx_update: bool = False,
) -> None:
"""
处理单个 SSE 事件
@@ -126,6 +236,7 @@ class StreamProcessor:
event_name: 事件名称
data_str: 事件数据字符串
skip_record: 是否跳过记录到 parsed_chunks（当需要格式转换时应为 True）
skip_ctx_update: 跳过 usage/completion/text 提取（由调用方统一处理时使用）
"""
if not data_str:
return
@@ -142,31 +253,17 @@ class StreamProcessor:
if not isinstance(data, dict):
return
# Update usage/completion/text from provider event (envelope-aware).
# 在 needs_conversion 正常流中由 _emit_converted_line 统一处理，此处跳过以避免重复。
if not skip_ctx_update:
self._update_ctx_from_provider_event(ctx, data)
# 统计数据事件数量（当需要格式转换时跳过，由 _emit_converted_line 统计/记录转换后的数据）
if not skip_record:
ctx.data_count += 1
if ctx.record_parsed_chunks:
ctx.parsed_chunks.append(data)
# 根据 Provider 格式选择解析器
parser = self.get_parser_for_provider(ctx)
# 使用解析器提取 usage
usage = parser.extract_usage_from_response(data)
if usage:
ctx.update_usage(
input_tokens=usage.get("input_tokens"),
output_tokens=usage.get("output_tokens"),
cached_tokens=usage.get("cache_read_tokens"),
cache_creation_tokens=usage.get("cache_creation_tokens"),
)
# 提取文本
if self.collect_text:
text = parser.extract_text_content(data)
if text:
ctx.append_text(text)
# 检查完成
event_type = event_name or data.get("type", "")
if event_type in ("response.completed", "message_stop"):
@@ -476,14 +573,31 @@ class StreamProcessor:
self.on_streaming_start()
streaming_started = True
def _process_line_with_perf(line: str, *, skip_record: bool = False) -> None:
def _process_line_with_perf(
line: str,
*,
skip_record: bool = False,
skip_ctx_update: bool = False,
) -> None:
nonlocal parse_time
if perf_capture:
t0 = time.perf_counter()
self._process_line(ctx, sse_parser, line, skip_record=skip_record)
self._process_line(
ctx,
sse_parser,
line,
skip_record=skip_record,
skip_ctx_update=skip_ctx_update,
)
parse_time += time.perf_counter() - t0
return
self._process_line(ctx, sse_parser, line, skip_record=skip_record)
self._process_line(
ctx,
sse_parser,
line,
skip_record=skip_record,
skip_ctx_update=skip_ctx_update,
)
def _build_stream_error_payload(message: str) -> dict:
if client_family == "openai":
@@ -589,6 +703,14 @@ class StreamProcessor:
data=data_obj,
)
# Update usage/completion/text based on the unwrapped provider event.
if isinstance(data_obj, dict):
self._update_ctx_from_provider_event(
ctx,
data_obj,
already_unwrapped=True,
)
try:
converted_events = registry.convert_stream_chunk(
data_obj,
@@ -648,7 +770,9 @@ class StreamProcessor:
if line:
# 需要格式转换时，跳过记录原始数据（由 _emit_converted_line 记录转换后的数据）
_process_line_with_perf(line, skip_record=True)
_process_line_with_perf(
line, skip_record=True, skip_ctx_update=True
)
normalized_line = line.rstrip("\r\n") if line else ""
out_chunks = _emit_converted_line(normalized_line)
if not out_chunks:
@@ -878,6 +1002,7 @@ class StreamProcessor:
line: str,
*,
skip_record: bool = False,
skip_ctx_update: bool = False,
) -> None:
"""
处理单行数据
@@ -887,6 +1012,7 @@ class StreamProcessor:
sse_parser: SSE 解析器
line: 原始行数据
skip_record: 是否跳过记录到 parsed_chunks（当需要格式转换时应为 True）
skip_ctx_update: 跳过 usage/completion/text 提取（由调用方统一处理时使用）
"""
# SSEEventParser 以"去掉换行符"的单行文本作为输入；这里统一剔除 CR/LF，
# 避免把空行误判成 "\n" 并导致事件边界解析错误。
@@ -898,7 +1024,11 @@ class StreamProcessor:
for event in events:
self.handle_sse_event(
ctx, event.get("event"), event.get("data") or "", skip_record=skip_record
ctx,
event.get("event"),
event.get("data") or "",
skip_record=skip_record,
skip_ctx_update=skip_ctx_update,
)
async def create_monitored_stream(

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

@@ -227,7 +227,7 @@ class StreamTelemetryRecorder:
request_type="chat",
metadata=metadata,
endpoint_api_format=ctx.provider_api_format,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
)
logger.debug(f"{self.format_id} 流式响应完成")
@@ -273,7 +273,7 @@ class StreamTelemetryRecorder:
request_type="chat",
metadata=metadata,
endpoint_api_format=ctx.provider_api_format,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
)
logger.debug(f"{self.format_id} 流式响应中断")
@@ -320,7 +320,7 @@ class StreamTelemetryRecorder:
request_type="chat",
metadata=metadata,
endpoint_api_format=ctx.provider_api_format,
has_format_conversion=ctx.needs_conversion,
has_format_conversion=ctx.has_format_conversion,
)
logger.debug(f"{self.format_id} 流式响应被客户端取消")

2
src/core/provider_templates/fixed_providers.py
View File

@@ -106,7 +106,7 @@ FIXED_PROVIDERS: dict[ProviderType, FixedProviderTemplate] = {
provider_type=ProviderType.ANTIGRAVITY,
display_name="Antigravity",
api_base_url=ANTIGRAVITY_PROD_URL,
endpoint_signatures=["gemini:cli"],
endpoint_signatures=["gemini:chat"],
oauth=FixedProviderOAuth(
authorize_url="https://accounts.google.com/o/oauth2/v2/auth",
token_url="https://oauth2.googleapis.com/token",

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

@@ -303,7 +303,9 @@ class ModelFetchScheduler:
# Use request_builder's lazy refresh logic and persist refreshed token back to DB.
# Endpoint signature is only used for tracing/debug; auth logic doesn't depend on it.
endpoint_api_format = (
"gemini:cli" if prepared.provider_type.lower() == ProviderType.ANTIGRAVITY else None
"gemini:chat"
if prepared.provider_type.lower() == ProviderType.ANTIGRAVITY
else None
)
try:
resolved = await resolve_oauth_access_token(

2
src/services/provider/adapters/antigravity/envelope.py
View File

@@ -1,6 +1,6 @@
"""Antigravity v1internal request/response envelope helpers.
Antigravity reuses the `gemini:cli` endpoint signature but wraps the actual
Antigravity reuses the `gemini:chat` endpoint signature but wraps the actual
wire format:
- Request: V1InternalRequest (top-level metadata + nested GeminiRequest)
- Response: V1InternalResponse (top-level responseId + nested GeminiResponse)

6
src/services/provider/adapters/antigravity/plugin.py
View File

@@ -249,7 +249,7 @@ async def fetch_models_antigravity(
"id": model_id,
"owned_by": "antigravity",
"display_name": display_name,
"api_format": "gemini:cli",
"api_format": "gemini:chat",
}
)
@@ -324,10 +324,14 @@ def register_all() -> None:
from src.services.provider.transport import register_transport_hook
# Envelope
register_envelope("antigravity", "gemini:chat", antigravity_v1internal_envelope)
# Backward compat: allow existing endpoints that still use the old signature.
register_envelope("antigravity", "gemini:cli", antigravity_v1internal_envelope)
register_envelope("antigravity", "", antigravity_v1internal_envelope)
# Transport
register_transport_hook("antigravity", "gemini:chat", build_antigravity_url)
# Backward compat: allow existing endpoints that still use the old signature.
register_transport_hook("antigravity", "gemini:cli", build_antigravity_url)
# Auth

4
src/utils/cache_decorator.py
View File

@@ -106,7 +106,7 @@ def cache_result(
if cached:
try:
result = json.loads(cached)
logger.debug(f"缓存命中: {cache_key}")
logger.trace(f"缓存命中: {cache_key}")
return result
except json.JSONDecodeError as e:
logger.warning(f"缓存解析失败，删除损坏缓存: {cache_key}, 错误: {e}")
@@ -123,7 +123,7 @@ def cache_result(
await redis_client.setex(
cache_key, ttl, json.dumps(result, ensure_ascii=False, default=str)
)
logger.debug(f"缓存已保存: {cache_key}, TTL: {ttl}s")
logger.trace(f"缓存已保存: {cache_key}, TTL: {ttl}s")
except Exception as e:
logger.warning(f"保存缓存失败: {e}")