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: 拆分 usage 记录的请求体/响应体为客户端侧与提供商侧

Browse Source 
将 request_body/response_body 语义明确为客户端原始请求体和提供商原始响应体，
新增 provider_request_body（格式转换后发给提供商的请求体）和 client_response_body
（格式转换后返回给客户端的响应体），支持跨格式转换场景下分别查看两侧数据。

- 数据库新增 provider_request_body/client_response_body 及对应压缩字段
- 全链路（telemetry/recording/handler/stream_context）传递新字段
- 维护调度器同步支持新字段的压缩与清理
- 前端请求详情抽屉支持请求体/响应体/响应头的客户端/提供商视图切换
This commit is contained in:
fawney19
2026-02-21 01:56:28 +08:00
parent 4c5dac603f
commit 314e4a497d
23 changed files with 449 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1278,9 +1278,11 @@ class AdminUsageDetailAdapter(AdminApiAdapter):
"request_headers": usage_record.request_headers,
"request_body": usage_record.get_request_body(),
"provider_request_headers": usage_record.provider_request_headers,
"provider_request_body": usage_record.get_provider_request_body(),
"response_headers": usage_record.response_headers,
"client_response_headers": usage_record.client_response_headers,
"response_body": usage_record.get_response_body(),
"client_response_body": usage_record.get_client_response_body(),
"metadata": usage_record.request_metadata,
"tiered_pricing": tiered_pricing_info,
"video_billing": video_billing_info,

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

@@ -1046,6 +1046,10 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
output_state = {"started": False}
# 保留提供商原始响应到 provider_parsed_chunks
if ctx.record_parsed_chunks and isinstance(response_json, dict):
ctx.provider_parsed_chunks.append(response_json)
async def _streamified() -> AsyncGenerator[bytes]:
for ev in iter_internal_response_as_stream_events(internal_resp):
converted_events = tgt_norm.stream_event_from_internal(ev, state)

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

@@ -70,6 +70,7 @@ class SyncRequestContext:
key_id: str | None = None
mapped_model_result: str | None = None
sync_proxy_info: dict[str, Any] | None = None
provider_response_json: dict[str, Any] | None = None # 格式转换前的提供商原始响应
class ChatSyncExecutor:
@@ -182,8 +183,6 @@ class ChatSyncExecutor:
cache_creation_tokens = usage_info.get("cache_creation_input_tokens", 0)
cached_tokens = usage_info.get("cache_read_input_tokens", 0)
actual_request_body = ctx.provider_request_body or original_request_body
# 非流式成功时，返回给客户端的是提供商响应头（透传）
# JSONResponse 会自动设置 content-type，但我们记录实际返回的完整头
client_response_headers = filter_proxy_response_headers(ctx.response_headers)
@@ -200,10 +199,12 @@ class ChatSyncExecutor:
response_time_ms=response_time_ms,
status_code=ctx.status_code,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
response_body=ctx.response_json,
response_body=ctx.provider_response_json or ctx.response_json,
client_response_body=ctx.response_json if ctx.provider_response_json else None,
provider_request_body=ctx.provider_request_body,
cache_creation_tokens=cache_creation_tokens,
cache_read_tokens=cached_tokens,
is_stream=False,
@@ -242,7 +243,6 @@ class ChatSyncExecutor:
# Thinking 签名错误：TaskService 层已处理整流重试但仍失败
# 记录实际发送给 Provider 的请求体，便于排查问题根因
response_time_ms = handler.elapsed_ms()
actual_request_body = ctx.provider_request_body or original_request_body
request_metadata = handler._build_request_metadata() or {}
if ctx.sync_proxy_info:
request_metadata["proxy"] = ctx.sync_proxy_info
@@ -252,7 +252,8 @@ class ChatSyncExecutor:
response_time_ms=response_time_ms,
status_code=e.status_code or 400,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
provider_request_body=ctx.provider_request_body,
error_message=str(e),
is_stream=False,
provider_id=ctx.provider_id,
@@ -275,7 +276,6 @@ class ChatSyncExecutor:
except UpstreamClientException as e:
response_time_ms = handler.elapsed_ms()
actual_request_body = ctx.provider_request_body or original_request_body
request_metadata = handler._build_request_metadata() or {}
if ctx.sync_proxy_info:
request_metadata["proxy"] = ctx.sync_proxy_info
@@ -285,7 +285,8 @@ class ChatSyncExecutor:
response_time_ms=response_time_ms,
status_code=_get_error_status_code(e),
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
provider_request_body=ctx.provider_request_body,
error_message=str(e),
is_stream=False,
api_format=api_format,
@@ -327,8 +328,6 @@ class ChatSyncExecutor:
elif isinstance(e, ProviderTimeoutException):
status_code = 504
actual_request_body = ctx.provider_request_body or original_request_body
# 尝试从异常中提取响应头
error_response_headers: dict[str, str] = {}
if isinstance(e, ProviderRateLimitException) and e.response_headers:
@@ -346,7 +345,8 @@ class ChatSyncExecutor:
status_code=status_code,
error_message=extract_client_error_message(e),
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
provider_request_body=ctx.provider_request_body,
is_stream=False,
api_format=api_format,
provider_request_headers=ctx.provider_request_headers,
@@ -667,6 +667,7 @@ class ChatSyncExecutor:
# 跨格式：响应转换回 client_format（失败触发 failover）
if needs_conversion and isinstance(ctx.response_json, dict):
ctx.provider_response_json = ctx.response_json.copy()
registry = get_format_converter_registry()
ctx.response_json = registry.convert_response(
ctx.response_json,
@@ -700,8 +701,6 @@ class ChatSyncExecutor:
elif isinstance(error, ProviderTimeoutException):
status_code = 504
actual_request_body = ctx.provider_request_body or original_request_body
# 失败时返回给客户端的是 JSON 错误响应
client_response_headers = {"content-type": "application/json"}
@@ -716,7 +715,8 @@ class ChatSyncExecutor:
status_code=status_code,
error_message=extract_client_error_message(error),
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
provider_request_body=ctx.provider_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,

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

@@ -80,6 +80,10 @@ class CliEventMixin:
ctx.data_count += 1
if ctx.record_parsed_chunks:
ctx.parsed_chunks.append(data)
else:
# 格式转换场景：保留提供商原始数据
if ctx.record_parsed_chunks:
ctx.provider_parsed_chunks.append(data)
event_type = event_name or data.get("type", "")

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

@@ -211,19 +211,8 @@ class CliMonitorMixin:
bg_db, user, api_key, ctx.request_id, self.client_ip
)
response_body = {
"chunks": ctx.parsed_chunks,
"metadata": {
"stream": True,
"total_chunks": len(ctx.parsed_chunks),
"data_count": ctx.data_count,
"has_completion": ctx.has_completion,
"response_time_ms": response_time_ms,
},
}
# 使用实际发送给 Provider 的请求体（如果有），否则用原始请求体
actual_request_body = ctx.provider_request_body or original_request_body
response_body = ctx.build_response_body(response_time_ms)
client_response_body = ctx.build_client_response_body(response_time_ms)
# 根据状态码决定记录成功还是失败
# 499 = 客户端取消（不算系统失败）；其他 4xx/5xx 视为失败
@@ -242,15 +231,17 @@ class CliMonitorMixin:
first_byte_time_ms=ctx.first_byte_time_ms,
status_code=ctx.status_code,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
response_body=response_body,
client_response_body=client_response_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
endpoint_api_format=ctx.provider_api_format or None,
@@ -273,16 +264,18 @@ class CliMonitorMixin:
status_code=ctx.status_code,
error_message=ctx.error_message or f"HTTP {ctx.status_code}",
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
# 预估 token 信息（来自 message_start 事件）
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
response_body=response_body,
client_response_body=client_response_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
# 格式转换追踪
@@ -319,7 +312,10 @@ class CliMonitorMixin:
and ctx.input_tokens == 0
and ctx.output_tokens == 0
):
self._estimate_tokens_for_incomplete_stream(ctx, actual_request_body)
# 用实际发给 Provider 的请求体估算 token（格式转换时与客户端请求体不同）
self._estimate_tokens_for_incomplete_stream(
ctx, ctx.provider_request_body or original_request_body
)
# 流式成功时，返回给客户端的是提供商响应头 + SSE 必需头
client_response_headers = filter_proxy_response_headers(ctx.response_headers)
@@ -346,10 +342,12 @@ class CliMonitorMixin:
first_byte_time_ms=ctx.first_byte_time_ms, # 传递首字时间
status_code=ctx.status_code,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
response_body=response_body,
client_response_body=client_response_body,
provider_request_body=ctx.provider_request_body,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
is_stream=True,
@@ -487,9 +485,6 @@ class CliMonitorMixin:
ctx.status_code = status_code
ctx.error_message = str(error)
# 使用实际发送给 Provider 的请求体（如果有），否则用原始请求体
actual_request_body = ctx.provider_request_body or original_request_body
# 失败时返回给客户端的是 JSON 错误响应
client_response_headers = {"content-type": "application/json"}
@@ -501,10 +496,11 @@ class CliMonitorMixin:
status_code=status_code,
error_message=extract_client_error_message(error),
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
# 格式转换追踪

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

@@ -243,6 +243,7 @@ class CliStreamMixin:
# 重置上下文状态（重试时清除之前的数据，避免累积）
ctx.parsed_chunks = []
ctx.provider_parsed_chunks = []
ctx.chunk_count = 0
ctx.data_count = 0
ctx.has_completion = False

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

@@ -480,6 +480,7 @@ class CliSyncMixin:
response_json = {}
# 跨格式：响应转换回 client_format（失败不触发 failover，保守回退为原始响应）
provider_response_json: dict[str, Any] | None = None
if (
needs_conversion
and provider_api_format
@@ -487,6 +488,7 @@ class CliSyncMixin:
and isinstance(response_json, dict)
):
try:
provider_response_json = response_json.copy()
registry = get_format_converter_registry()
response_json = registry.convert_response(
response_json,
@@ -499,6 +501,7 @@ class CliSyncMixin:
)
except Exception as conv_err:
logger.warning("非流式响应格式转换失败，使用原始响应: {}", conv_err)
provider_response_json = None
# 使用解析器提取 usage
usage = self.parser.extract_usage_from_response(response_json)
@@ -509,9 +512,6 @@ class CliSyncMixin:
output_text = self.parser.extract_text_content(response_json)[:200]
# 使用实际发送给 Provider 的请求体（如果有），否则用原始请求体
actual_request_body = provider_request_body or original_request_body
# 非流式成功时，返回给客户端的是提供商响应头（透传）
client_response_headers = filter_proxy_response_headers(response_headers)
client_response_headers["content-type"] = "application/json"
@@ -527,10 +527,12 @@ class CliSyncMixin:
response_time_ms=response_time_ms,
status_code=status_code,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
response_headers=response_headers,
client_response_headers=client_response_headers,
response_body=response_json,
response_body=provider_response_json or response_json,
client_response_body=response_json if provider_response_json else None,
provider_request_body=provider_request_body,
cache_creation_tokens=cache_creation_tokens,
cache_read_tokens=cached_tokens,
is_stream=False,
@@ -563,7 +565,6 @@ class CliSyncMixin:
# Thinking 签名错误：TaskService 层已处理整流重试但仍失败
# 记录实际发送给 Provider 的请求体，便于排查问题根因
response_time_ms = int((time.time() - sync_start_time) * 1000)
actual_request_body = provider_request_body or original_request_body
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
@@ -573,7 +574,8 @@ class CliSyncMixin:
response_time_ms=response_time_ms,
status_code=e.status_code or 400,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
provider_request_body=provider_request_body,
error_message=str(e),
is_stream=False,
api_format=api_format,
@@ -592,9 +594,6 @@ class CliSyncMixin:
elif isinstance(e, ProviderTimeoutException):
status_code = 504
# 使用实际发送给 Provider 的请求体（如果有），否则用原始请求体
actual_request_body = provider_request_body or original_request_body
# 尝试从异常中提取响应头
error_response_headers: dict[str, str] = {}
if isinstance(e, ProviderRateLimitException) and e.response_headers:
@@ -612,7 +611,8 @@ class CliSyncMixin:
status_code=status_code,
error_message=extract_client_error_message(e),
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
provider_request_body=provider_request_body,
is_stream=False,
api_format=api_format,
provider_request_headers=provider_request_headers,

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

@@ -121,6 +121,8 @@ class StreamContext:
data_count: int = 0
chunk_count: int = 0
parsed_chunks: list[dict[str, Any]] = field(default_factory=list)
# 格式转换时保留提供商原始 chunks（转换前的数据）
provider_parsed_chunks: list[dict[str, Any]] = field(default_factory=list)
# 是否记录 parsed_chunks（可用于降低高并发/长流式响应的内存占用）
record_parsed_chunks: bool = True
@@ -143,6 +145,7 @@ class StreamContext:
保留 model 和 api_format，重置其他所有状态。
"""
self.parsed_chunks = []
self.provider_parsed_chunks = []
self.chunk_count = 0
self.data_count = 0
self.has_completion = False
@@ -299,7 +302,29 @@ class StreamContext:
构建响应体元数据
用于记录到 Usage 表的 response_body 字段。
当有格式转换时，返回提供商原始 chunks；否则返回 parsed_chunks。
"""
chunks = self.provider_parsed_chunks if self.provider_parsed_chunks else self.parsed_chunks
return {
"chunks": chunks,
"metadata": {
"stream": True,
"total_chunks": len(chunks),
"data_count": self.data_count,
"has_completion": self.has_completion,
"response_time_ms": response_time_ms,
},
}
def build_client_response_body(self, response_time_ms: int) -> dict[str, Any] | None:
"""
构建客户端侧响应体元数据
仅当有格式转换时返回（parsed_chunks 是转换后的客户端格式）；
无格式转换时返回 None（此时 parsed_chunks 已在 response_body 中）。
"""
if not self.provider_parsed_chunks:
return None
return {
"chunks": self.parsed_chunks,
"metadata": {

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

@@ -263,6 +263,10 @@ class StreamProcessor:
ctx.data_count += 1
if ctx.record_parsed_chunks:
ctx.parsed_chunks.append(data)
else:
# 格式转换场景：保留提供商原始数据
if ctx.record_parsed_chunks:
ctx.provider_parsed_chunks.append(data)
# 检查完成
event_type = event_name or data.get("type", "")

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

@@ -100,8 +100,6 @@ class StreamTelemetryRecorder:
writer = await self._get_telemetry_writer(bg_db, ctx, response_time_ms)
if writer is None:
return
actual_request_body = ctx.provider_request_body or original_request_body
# 兜底估算：流未正常完成且 token 均为 0 时，从请求体粗略估算
# 覆盖 Chat Handler 路径（CLI Handler 在更早的位置已做估算，
# 若已估算过则 token > 0，此处条件不会触发）
@@ -112,7 +110,10 @@ class StreamTelemetryRecorder:
and ctx.input_tokens == 0
and ctx.output_tokens == 0
):
self._estimate_tokens_for_incomplete_stream(ctx, actual_request_body)
# 用实际发给 Provider 的请求体估算 token（格式转换时与客户端请求体不同）
self._estimate_tokens_for_incomplete_stream(
ctx, ctx.provider_request_body or original_request_body
)
should_log_body = SystemConfigService.should_log_body(bg_db)
include_bodies = (
@@ -123,6 +124,9 @@ class StreamTelemetryRecorder:
response_body = (
ctx.build_response_body(response_time_ms) if include_bodies else None
)
client_response_body = (
ctx.build_client_response_body(response_time_ms) if include_bodies else None
)
try:
await self._dispatch_record(
@@ -130,9 +134,10 @@ class StreamTelemetryRecorder:
writer,
ctx,
original_headers,
actual_request_body,
original_request_body,
response_body,
response_time_ms,
client_response_body=client_response_body,
)
except Exception as writer_error:
if not isinstance(writer, QueueTelemetryWriter):
@@ -151,14 +156,17 @@ class StreamTelemetryRecorder:
return
if response_body is None and should_log_body:
response_body = ctx.build_response_body(response_time_ms)
if client_response_body is None and should_log_body:
client_response_body = ctx.build_client_response_body(response_time_ms)
await self._dispatch_record(
bg_db,
db_writer,
ctx,
original_headers,
actual_request_body,
original_request_body,
response_body,
response_time_ms,
client_response_body=client_response_body,
)
# 更新候选记录状态
@@ -180,9 +188,10 @@ class StreamTelemetryRecorder:
writer: TelemetryWriter,
ctx: StreamContext,
original_headers: dict[str, str],
actual_request_body: dict[str, Any],
original_request_body: dict[str, Any],
response_body: dict[str, Any] | None,
response_time_ms: int,
client_response_body: dict[str, Any] | None = None,
) -> None:
"""记录成功的请求"""
# 流式成功时，返回给客户端的是提供商响应头 + SSE 必需头
@@ -209,10 +218,12 @@ class StreamTelemetryRecorder:
first_byte_time_ms=ctx.first_byte_time_ms, # 传递首字时间
status_code=ctx.status_code,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
response_body=response_body,
client_response_body=client_response_body,
provider_request_body=ctx.provider_request_body,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
is_stream=True,
@@ -236,9 +247,10 @@ class StreamTelemetryRecorder:
writer: TelemetryWriter,
ctx: StreamContext,
original_headers: dict[str, str],
actual_request_body: dict[str, Any],
original_request_body: dict[str, Any],
response_body: dict[str, Any] | None,
response_time_ms: int,
client_response_body: dict[str, Any] | None = None,
) -> None:
"""记录失败的请求"""
# 失败时返回给客户端的是 JSON 错误响应，如果没有设置则使用默认值
@@ -258,15 +270,17 @@ class StreamTelemetryRecorder:
status_code=ctx.status_code,
error_message=ctx.error_message or f"HTTP {ctx.status_code}",
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
response_body=response_body,
client_response_body=client_response_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
provider_id=ctx.provider_id,
@@ -289,9 +303,10 @@ class StreamTelemetryRecorder:
writer: TelemetryWriter,
ctx: StreamContext,
original_headers: dict[str, str],
actual_request_body: dict[str, Any],
original_request_body: dict[str, Any],
response_body: dict[str, Any] | None,
response_time_ms: int,
client_response_body: dict[str, Any] | None = None,
) -> None:
"""记录客户端取消的请求"""
client_response_headers = ctx.client_response_headers or {
@@ -310,15 +325,17 @@ class StreamTelemetryRecorder:
first_byte_time_ms=ctx.first_byte_time_ms,
status_code=ctx.status_code,
request_headers=original_headers,
request_body=actual_request_body,
request_body=original_request_body,
is_stream=True,
api_format=ctx.api_format,
provider_request_headers=ctx.provider_request_headers,
provider_request_body=ctx.provider_request_body,
input_tokens=ctx.input_tokens,
output_tokens=ctx.output_tokens,
cache_creation_tokens=ctx.cache_creation_tokens,
cache_read_tokens=ctx.cached_tokens,
response_body=response_body,
client_response_body=client_response_body,
response_headers=ctx.response_headers,
client_response_headers=client_response_headers,
provider_id=ctx.provider_id,
@@ -481,9 +498,10 @@ class StreamTelemetryRecorder:
writer: TelemetryWriter,
ctx: StreamContext,
original_headers: dict[str, str],
actual_request_body: dict[str, Any],
original_request_body: dict[str, Any],
response_body: dict[str, Any] | None,
response_time_ms: int,
client_response_body: dict[str, Any] | None = None,
) -> None:
"""根据上下文状态分发到对应的记录方法"""
if ctx.is_success():
@@ -491,9 +509,10 @@ class StreamTelemetryRecorder:
writer,
ctx,
original_headers,
actual_request_body,
original_request_body,
response_body,
response_time_ms,
client_response_body=client_response_body,
)
# Queue writer 异步落库可能造成 UI 延迟，先直接更新 Usage 状态
if isinstance(writer, QueueTelemetryWriter):
@@ -508,9 +527,10 @@ class StreamTelemetryRecorder:
writer,
ctx,
original_headers,
actual_request_body,
original_request_body,
response_body,
response_time_ms,
client_response_body=client_response_body,
)
# Queue writer 异步落库可能造成 UI 延迟，先直接更新 Usage 状态
if isinstance(writer, QueueTelemetryWriter):
@@ -525,9 +545,10 @@ class StreamTelemetryRecorder:
writer,
ctx,
original_headers,
actual_request_body,
original_request_body,
response_body,
response_time_ms,
client_response_body=client_response_body,
)
# Queue writer 异步落库可能造成 UI 延迟，先直接更新 Usage 状态
if isinstance(writer, QueueTelemetryWriter):

36
src/models/database.py
View File

@@ -392,15 +392,19 @@ class Usage(Base):
# 完整请求和响应记录
request_headers = Column(JSON, nullable=True) # 客户端请求头
request_body = Column(JSON, nullable=True) # 请求体（7天内未压缩）
request_body = Column(JSON, nullable=True) # 客户端原始请求体（7天内未压缩）
provider_request_headers = Column(JSON, nullable=True) # 向提供商发送的请求头
provider_request_body = Column(JSON, nullable=True) # 发给提供商的请求体（格式转换后）
response_headers = Column(JSON, nullable=True) # 提供商响应头
response_body = Column(JSON, nullable=True) # 提供商原始响应体（7天内未压缩）
client_response_headers = Column(JSON, nullable=True) # 返回给客户端的响应头
response_body = Column(JSON, nullable=True) # 响应体（7天内未压缩）
client_response_body = Column(JSON, nullable=True) # 返回给客户端的响应体（格式转换后）
# 压缩存储字段（7天后自动压缩到这里）
request_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的请求体
response_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的响应体
request_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的客户端请求体
provider_request_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的提供商请求体
response_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的提供商响应体
client_response_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的客户端响应体
# 元数据
request_metadata = Column(JSON, nullable=True) # 存储额外信息
@@ -421,7 +425,7 @@ class Usage(Base):
provider_api_key = relationship("ProviderAPIKey")
def get_request_body(self) -> Any:
"""获取请求体（自动解压）"""
"""获取客户端原始请求体（自动解压）"""
if self.request_body is not None:
return self.request_body
if self.request_body_compressed is not None:
@@ -430,8 +434,18 @@ class Usage(Base):
return decompress_json(self.request_body_compressed)
return None
def get_provider_request_body(self) -> Any:
"""获取发给提供商的请求体（自动解压）"""
if self.provider_request_body is not None:
return self.provider_request_body
if self.provider_request_body_compressed is not None:
from src.utils.compression import decompress_json
return decompress_json(self.provider_request_body_compressed)
return None
def get_response_body(self) -> Any:
"""获取响应体（自动解压）"""
"""获取提供商原始响应体（自动解压）"""
if self.response_body is not None:
return self.response_body
if self.response_body_compressed is not None:
@@ -440,6 +454,16 @@ class Usage(Base):
return decompress_json(self.response_body_compressed)
return None
def get_client_response_body(self) -> Any:
"""获取返回给客户端的响应体（自动解压）"""
if self.client_response_body is not None:
return self.client_response_body
if self.client_response_body_compressed is not None:
from src.utils.compression import decompress_json
return decompress_json(self.client_response_body_compressed)
return None
class UserQuota(Base):
"""用户配额历史记录"""

36
src/models/usage.py
View File

@@ -126,15 +126,19 @@ class Usage(Base):
# 完整请求和响应记录
request_headers = Column(JSON, nullable=True) # 客户端请求头
request_body = Column(JSON, nullable=True) # 请求体（7天内未压缩）
request_body = Column(JSON, nullable=True) # 客户端原始请求体（7天内未压缩）
provider_request_headers = Column(JSON, nullable=True) # 向提供商发送的请求头
provider_request_body = Column(JSON, nullable=True) # 发给提供商的请求体（格式转换后）
response_headers = Column(JSON, nullable=True) # 提供商响应头
response_body = Column(JSON, nullable=True) # 提供商原始响应体（7天内未压缩）
client_response_headers = Column(JSON, nullable=True) # 返回给客户端的响应头
response_body = Column(JSON, nullable=True) # 响应体（7天内未压缩）
client_response_body = Column(JSON, nullable=True) # 返回给客户端的响应体（格式转换后）
# 压缩存储字段（7天后自动压缩到这里）
request_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的请求体
response_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的响应体
request_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的客户端请求体
provider_request_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的提供商请求体
response_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的提供商响应体
client_response_body_compressed = Column(LargeBinary, nullable=True) # gzip压缩的客户端响应体
# 元数据
request_metadata = Column(JSON, nullable=True) # 存储额外信息
@@ -155,7 +159,7 @@ class Usage(Base):
provider_api_key = relationship("ProviderAPIKey")
def get_request_body(self) -> Any:
"""获取请求体（自动解压）"""
"""获取客户端原始请求体（自动解压）"""
if self.request_body is not None:
return self.request_body
if self.request_body_compressed is not None:
@@ -164,8 +168,18 @@ class Usage(Base):
return decompress_json(self.request_body_compressed)
return None
def get_provider_request_body(self) -> Any:
"""获取发给提供商的请求体（自动解压）"""
if self.provider_request_body is not None:
return self.provider_request_body
if self.provider_request_body_compressed is not None:
from src.utils.compression import decompress_json
return decompress_json(self.provider_request_body_compressed)
return None
def get_response_body(self) -> Any:
"""获取响应体（自动解压）"""
"""获取提供商原始响应体（自动解压）"""
if self.response_body is not None:
return self.response_body
if self.response_body_compressed is not None:
@@ -174,6 +188,16 @@ class Usage(Base):
return decompress_json(self.response_body_compressed)
return None
def get_client_response_body(self) -> Any:
"""获取返回给客户端的响应体（自动解压）"""
if self.client_response_body is not None:
return self.client_response_body
if self.client_response_body_compressed is not None:
from src.utils.compression import decompress_json
return decompress_json(self.client_response_body_compressed)
return None
class RequestCandidate(Base):
"""请求候选记录 - 追踪所有候选（包括未使用的）"""

65
src/services/system/maintenance_scheduler.py
View File

@@ -928,9 +928,20 @@ class MaintenanceScheduler:
# 1. 查询需要压缩的记录
# 注意：排除已经是 NULL 或 JSON null 的记录
records = (
batch_db.query(Usage.id, Usage.request_body, Usage.response_body)
batch_db.query(
Usage.id,
Usage.request_body,
Usage.response_body,
Usage.provider_request_body,
Usage.client_response_body,
)
.filter(Usage.created_at < cutoff_time)
.filter((Usage.request_body.isnot(None)) | (Usage.response_body.isnot(None)))
.filter(
(Usage.request_body.isnot(None))
| (Usage.response_body.isnot(None))
| (Usage.provider_request_body.isnot(None))
| (Usage.client_response_body.isnot(None))
)
.limit(batch_size)
.all()
)
@@ -940,9 +951,12 @@ class MaintenanceScheduler:
# 过滤掉实际值为 None 的记录（JSON null 被解析为 Python None）
valid_records = [
(rid, req, resp)
for rid, req, resp in records
if req is not None or resp is not None
r
for r in records
if r.request_body is not None
or r.response_body is not None
or r.provider_request_body is not None
or r.client_response_body is not None
]
if not valid_records:
@@ -950,17 +964,22 @@ class MaintenanceScheduler:
logger.warning(
f"检测到 {len(records)} 条记录的 body 字段为 JSON null，进行清理"
)
for record_id, _, _ in records:
for r in records:
batch_db.execute(
update(Usage)
.where(Usage.id == record_id)
.values(request_body=null(), response_body=null())
.where(Usage.id == r.id)
.values(
request_body=null(),
response_body=null(),
provider_request_body=null(),
client_response_body=null(),
)
)
batch_db.commit()
continue
# 检测是否有重复的 ID（说明更新未生效）
current_ids = {r[0] for r in valid_records}
current_ids = {r.id for r in valid_records}
repeated_ids = current_ids & processed_ids
if repeated_ids:
logger.error(
@@ -972,28 +991,40 @@ class MaintenanceScheduler:
batch_success = 0
# 2. 逐条更新（确保每条都正确处理）
for record_id, req_body, resp_body in valid_records:
for r in valid_records:
try:
# 使用 null() 确保设置的是 SQL NULL 而不是 JSON null
result = batch_db.execute(
update(Usage)
.where(Usage.id == record_id)
.where(Usage.id == r.id)
.values(
request_body=null(),
response_body=null(),
provider_request_body=null(),
client_response_body=null(),
request_body_compressed=(
compress_json(req_body) if req_body else None
compress_json(r.request_body) if r.request_body else None
),
response_body_compressed=(
compress_json(resp_body) if resp_body else None
compress_json(r.response_body) if r.response_body else None
),
provider_request_body_compressed=(
compress_json(r.provider_request_body)
if r.provider_request_body
else None
),
client_response_body_compressed=(
compress_json(r.client_response_body)
if r.client_response_body
else None
),
)
)
if result.rowcount > 0:
batch_success += 1
processed_ids.add(record_id)
processed_ids.add(r.id)
except Exception as e:
logger.warning(f"压缩记录 {record_id} 失败: {e}")
logger.warning(f"压缩记录 {r.id} 失败: {e}")
continue
batch_db.commit()
@@ -1050,6 +1081,8 @@ class MaintenanceScheduler:
.filter(
(Usage.request_body_compressed.isnot(None))
| (Usage.response_body_compressed.isnot(None))
| (Usage.provider_request_body_compressed.isnot(None))
| (Usage.client_response_body_compressed.isnot(None))
)
.limit(batch_size)
.all()
@@ -1067,6 +1100,8 @@ class MaintenanceScheduler:
.values(
request_body_compressed=null(),
response_body_compressed=null(),
provider_request_body_compressed=null(),
client_response_body_compressed=null(),
)
)

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

@@ -146,9 +146,11 @@ class UsageBillingIntegrationMixin:
request_headers=params.request_headers,
request_body=params.request_body,
provider_request_headers=params.provider_request_headers,
provider_request_body=params.provider_request_body,
response_headers=params.response_headers,
client_response_headers=params.client_response_headers,
response_body=params.response_body,
client_response_body=params.client_response_body,
request_id=params.request_id,
provider_id=params.provider_id,
provider_endpoint_id=params.provider_endpoint_id,

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

@@ -56,9 +56,11 @@ def build_usage_params(
request_headers: dict[str, Any] | None,
request_body: Any | None,
provider_request_headers: dict[str, Any] | None,
provider_request_body: Any | None,
response_headers: dict[str, Any] | None,
client_response_headers: dict[str, Any] | None,
response_body: Any | None,
client_response_body: Any | None,
request_id: str,
provider_id: str | None,
provider_endpoint_id: str | None,
@@ -103,16 +105,26 @@ def build_usage_params(
# 处理请求体和响应体（可能需要截断）
processed_request_body = None
processed_provider_request_body = None
processed_response_body = None
processed_client_response_body = None
if should_log_body:
if request_body:
processed_request_body = SystemConfigService.truncate_body(
db, request_body, is_request=True
)
if provider_request_body:
processed_provider_request_body = SystemConfigService.truncate_body(
db, provider_request_body, is_request=True
)
if response_body:
processed_response_body = SystemConfigService.truncate_body(
db, response_body, is_request=False
)
if client_response_body:
processed_client_response_body = SystemConfigService.truncate_body(
db, client_response_body, is_request=False
)
# 处理响应头
processed_response_headers = None
@@ -197,9 +209,11 @@ def build_usage_params(
"request_headers": processed_request_headers,
"request_body": processed_request_body,
"provider_request_headers": processed_provider_request_headers,
"provider_request_body": processed_provider_request_body,
"response_headers": processed_response_headers,
"client_response_headers": processed_client_response_headers,
"response_body": processed_response_body,
"client_response_body": processed_client_response_body,
}
@@ -231,9 +245,12 @@ def update_existing_usage(
existing_usage.request_body = usage_params["request_body"]
if usage_params["provider_request_headers"] is not None:
existing_usage.provider_request_headers = usage_params["provider_request_headers"]
if usage_params["provider_request_body"] is not None:
existing_usage.provider_request_body = usage_params["provider_request_body"]
existing_usage.response_body = usage_params["response_body"]
existing_usage.response_headers = usage_params["response_headers"]
existing_usage.client_response_headers = usage_params["client_response_headers"]
existing_usage.client_response_body = usage_params["client_response_body"]
# 更新 token 和费用信息
existing_usage.input_tokens = usage_params["input_tokens"]

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

@@ -34,9 +34,11 @@ class UsageRecordParams:
request_headers: dict[str, Any] | None
request_body: Any | None
provider_request_headers: dict[str, Any] | None
provider_request_body: Any | None
response_headers: dict[str, Any] | None
client_response_headers: dict[str, Any] | None
response_body: Any | None
client_response_body: Any | None
request_id: str
provider_id: str | None
provider_endpoint_id: str | None

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

@@ -81,9 +81,11 @@ def _event_to_record(event: UsageEvent) -> dict[str, Any]:
"request_headers": data.get("request_headers"),
"request_body": _parse_body(data.get("request_body")),
"provider_request_headers": data.get("provider_request_headers"),
"provider_request_body": _parse_body(data.get("provider_request_body")),
"response_headers": data.get("response_headers"),
"client_response_headers": data.get("client_response_headers"),
"response_body": _parse_body(data.get("response_body")),
"client_response_body": _parse_body(data.get("client_response_body")),
"provider_id": data.get("provider_id"),
"provider_endpoint_id": data.get("provider_endpoint_id"),
"provider_api_key_id": data.get("provider_api_key_id"),
@@ -464,9 +466,11 @@ class UsageQueueConsumer:
request_headers=data.get("request_headers"),
request_body=_parse_body(data.get("request_body")),
provider_request_headers=data.get("provider_request_headers"),
provider_request_body=_parse_body(data.get("provider_request_body")),
response_headers=data.get("response_headers"),
client_response_headers=data.get("client_response_headers"),
response_body=_parse_body(data.get("response_body")),
client_response_body=_parse_body(data.get("client_response_body")),
request_id=event.request_id,
provider_id=data.get("provider_id"),
provider_endpoint_id=data.get("provider_endpoint_id"),

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

@@ -78,9 +78,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers: dict[str, Any] | None = None,
request_body: Any | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
response_body: Any | None = None,
client_response_body: Any | None = None,
request_id: str | None = None,
provider_id: str | None = None,
provider_endpoint_id: str | None = None,
@@ -125,9 +127,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers,
provider_request_body=provider_request_body,
response_headers=response_headers,
client_response_headers=client_response_headers,
response_body=response_body,
client_response_body=client_response_body,
request_id=request_id,
provider_id=provider_id,
provider_endpoint_id=provider_endpoint_id,
@@ -196,9 +200,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers: dict[str, Any] | None = None,
request_body: Any | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
response_body: Any | None = None,
client_response_body: Any | None = None,
request_id: str | None = None,
provider_id: str | None = None,
provider_endpoint_id: str | None = None,
@@ -245,9 +251,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers,
provider_request_body=provider_request_body,
response_headers=response_headers,
client_response_headers=client_response_headers,
response_body=response_body,
client_response_body=client_response_body,
request_id=request_id,
provider_id=provider_id,
provider_endpoint_id=provider_endpoint_id,
@@ -383,9 +391,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers: dict[str, Any] | None = None,
request_body: Any | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
response_body: Any | None = None,
client_response_body: Any | None = None,
request_id: str | None = None,
provider_id: str | None = None,
provider_endpoint_id: str | None = None,
@@ -444,9 +454,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers,
provider_request_body=provider_request_body,
response_headers=response_headers,
client_response_headers=client_response_headers,
response_body=response_body,
client_response_body=client_response_body,
request_id=request_id,
provider_id=provider_id,
provider_endpoint_id=provider_endpoint_id,
@@ -743,9 +755,11 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
request_headers=record.get("request_headers"),
request_body=record.get("request_body"),
provider_request_headers=record.get("provider_request_headers"),
provider_request_body=record.get("provider_request_body"),
response_headers=record.get("response_headers"),
client_response_headers=record.get("client_response_headers"),
response_body=record.get("response_body"),
client_response_body=record.get("client_response_body"),
request_id=request_id,
provider_id=record.get("provider_id"),
provider_endpoint_id=record.get("provider_endpoint_id"),

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

@@ -72,6 +72,8 @@ class MessageTelemetry:
cache_read_tokens: int = 0,
is_stream: bool = False,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
client_response_body: Any | None = None,
# 时间指标
first_byte_time_ms: int | None = None, # 首字时间/TTFB
# Provider 侧追踪信息（用于记录真实成本）
@@ -117,9 +119,11 @@ class MessageTelemetry:
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers or {},
provider_request_body=provider_request_body,
response_headers=response_headers,
client_response_headers=client_response_headers,
response_body=response_body,
client_response_body=client_response_body,
request_id=self.request_id,
# Provider 侧追踪信息（用于记录真实成本）
provider_id=provider_id,
@@ -164,6 +168,7 @@ class MessageTelemetry:
is_stream: bool,
api_format: str | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
# 预估 token 信息（来自 message_start 事件，用于中断请求的成本估算）
input_tokens: int = 0,
output_tokens: int = 0,
@@ -172,6 +177,7 @@ class MessageTelemetry:
response_body: dict[str, Any] | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
client_response_body: Any | None = None,
# Provider 侧追踪信息（用于 curl 复现等场景）
provider_id: str | None = None,
provider_endpoint_id: str | None = None,
@@ -225,9 +231,11 @@ class MessageTelemetry:
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers or {},
provider_request_body=provider_request_body,
response_headers=response_headers or {},
client_response_headers=client_response_headers,
response_body=response_body or {"error": error_message},
client_response_body=client_response_body,
request_id=self.request_id,
# Provider 侧追踪信息
provider_id=provider_id,
@@ -252,6 +260,7 @@ class MessageTelemetry:
is_stream: bool,
api_format: str | None = None,
provider_request_headers: dict[str, Any] | None = None,
provider_request_body: Any | None = None,
input_tokens: int = 0,
output_tokens: int = 0,
cache_creation_tokens: int = 0,
@@ -259,6 +268,7 @@ class MessageTelemetry:
response_body: dict[str, Any] | None = None,
response_headers: dict[str, Any] | None = None,
client_response_headers: dict[str, Any] | None = None,
client_response_body: Any | None = None,
# Provider 侧追踪信息
provider_id: str | None = None,
provider_endpoint_id: str | None = None,
@@ -299,9 +309,11 @@ class MessageTelemetry:
request_headers=request_headers,
request_body=request_body,
provider_request_headers=provider_request_headers or {},
provider_request_body=provider_request_body,
response_headers=response_headers or {},
client_response_headers=client_response_headers,
response_body=response_body or {},
client_response_body=client_response_body,
request_id=self.request_id,
# Provider 侧追踪信息
provider_id=provider_id,

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

@@ -268,14 +268,28 @@ class QueueTelemetryWriter(TelemetryWriter):
max_size=self._max_request_body_size,
is_request=True,
)
provider_request_body = self._truncate_body(
kwargs.get("provider_request_body"),
max_size=self._max_request_body_size,
is_request=True,
)
response_body = self._truncate_body(
kwargs.get("response_body"),
max_size=self._max_response_body_size,
is_request=False,
)
client_response_body = self._truncate_body(
kwargs.get("client_response_body"),
max_size=self._max_response_body_size,
is_request=False,
)
if request_body is not None:
data["request_body"] = request_body
if provider_request_body is not None:
data["provider_request_body"] = provider_request_body
if response_body is not None:
data["response_body"] = response_body
if client_response_body is not None:
data["client_response_body"] = client_response_body
return data