- 代理发给上游时,仍然会补 stream_options.include_usage=true,所以 token 统计还在。
- 下游如果没有显式传 stream_options.include_usage=true,代理会把“纯 usage 的 SSE chunk”吞掉,不再往客户端发。 - 如果下游本来就显式要求 include_usage=true,那仍然会原样透传 usage chunk。
This commit is contained in:
1 parent
cae49a5dd0
commit
4131a8e7f2
3 files changed
+21
-1
No files matched your search
Vendored
+11
@@ -21,6 +21,7 @@ def build_proxy_response(
|
||||
stream_context_func: Any,
|
||||
logger: Logger,
|
||||
on_complete: Optional[Callable[[Dict[str, Any]], None]] = None,
|
||||
forward_stream_usage: bool = False,
|
||||
) -> Response:
|
||||
"""基于上游响应与输出钩子构建 Flask 响应。"""
|
||||
hook_func = getattr(hook, "output_body_hook", None) if hook else None
|
||||
@@ -62,6 +63,14 @@ def build_proxy_response(
|
||||
return None
|
||||
return stripped[5:].strip()
|
||||
|
||||
def is_usage_only_stream_chunk(payload: Any) -> bool:
|
||||
if not isinstance(payload, dict):
|
||||
return False
|
||||
if not isinstance(payload.get("usage"), dict):
|
||||
return False
|
||||
choices = payload.get("choices")
|
||||
return choices is None or (isinstance(choices, list) and len(choices) == 0)
|
||||
|
||||
def process_sse_event(event_text: str) -> Iterator[bytes]:
|
||||
# Preserve non-data SSE fields to avoid breaking strict clients.
|
||||
normalized_text = event_text.replace("\r\n", "\n").replace("\r", "\n")
|
||||
@@ -97,6 +106,8 @@ def build_proxy_response(
|
||||
|
||||
if isinstance(data, dict):
|
||||
update_meta_from_payload(data)
|
||||
if not forward_stream_usage and is_usage_only_stream_chunk(data):
|
||||
return
|
||||
|
||||
if hook_func:
|
||||
modified = hook_func(ctx, data)
|
||||
|
||||
@@ -71,6 +71,9 @@ class ProxyController:
|
||||
)
|
||||
|
||||
stream_options = request_data.get("stream_options")
|
||||
client_requested_usage_chunk = isinstance(stream_options, dict) and (
|
||||
stream_options.get("include_usage") is True
|
||||
)
|
||||
if not isinstance(stream_options, dict):
|
||||
stream_options = {}
|
||||
if stream_options.get("include_usage") is not True:
|
||||
@@ -109,7 +112,11 @@ class ProxyController:
|
||||
)
|
||||
|
||||
result, status_code = self._proxy_service.proxy_request(
|
||||
provider, request_data, headers, on_complete=on_proxy_complete
|
||||
provider,
|
||||
request_data,
|
||||
headers,
|
||||
on_complete=on_proxy_complete,
|
||||
forward_stream_usage=client_requested_usage_chunk,
|
||||
)
|
||||
if result is None:
|
||||
self._logger.error(f"Proxy failed after retries: model={model_name!r}, ip={client_ip}")
|
||||
|
||||
@@ -30,6 +30,7 @@ class ProxyService:
|
||||
request_data: Dict[str, Any],
|
||||
request_headers: Dict[str, str],
|
||||
on_complete: Optional[Callable[[Dict[str, Any]], None]] = None,
|
||||
forward_stream_usage: bool = False,
|
||||
) -> Tuple[Optional[Response], int]:
|
||||
"""代理请求到目标 provider,并处理重试与输出钩子。"""
|
||||
target_url = provider.api
|
||||
@@ -123,6 +124,7 @@ class ProxyService:
|
||||
stream_with_context,
|
||||
self._logger,
|
||||
on_complete=on_complete,
|
||||
forward_stream_usage=forward_stream_usage,
|
||||
)
|
||||
self._logger.info(
|
||||
"Upstream request completed: provider=%s status=%s stream=%s",
|
||||
|
||||
Reference in new issue
Block a user