mirror of
https://github.com/NCBM/plyngent.git
synced 2026-07-23 05:55:16 +08:00
core/lmproto: await async AsyncResponse.content
niquests exposes content as an async property; reading it without await left a coroutine warning and skipped the body. Only load body on HTTP errors for streams so SSE is not consumed early.
This commit is contained in:
@@ -2,7 +2,7 @@ from typing import TYPE_CHECKING, Literal, overload
|
|||||||
|
|
||||||
import msgspec
|
import msgspec
|
||||||
|
|
||||||
from ...openai_compatible.client import BaseOpenAIClient
|
from ...openai_compatible.client import BaseOpenAIClient, read_response_body
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from collections.abc import AsyncIterator
|
from collections.abc import AsyncIterator
|
||||||
@@ -45,5 +45,12 @@ class DeepseekOpenAIClient(BaseOpenAIClient):
|
|||||||
headers={"Content-Type": "application/json"},
|
headers={"Content-Type": "application/json"},
|
||||||
stream=False,
|
stream=False,
|
||||||
)
|
)
|
||||||
assert resp.content is not None
|
await self._ensure_ok(resp)
|
||||||
return self.decoder.decode(resp.content)
|
body = await read_response_body(resp)
|
||||||
|
if body is None:
|
||||||
|
msg = "chat completions response body is empty"
|
||||||
|
raise RuntimeError(msg)
|
||||||
|
if not isinstance(body, (bytes, bytearray)):
|
||||||
|
msg = f"chat completions response body has unexpected type {type(body)!r}"
|
||||||
|
raise TypeError(msg)
|
||||||
|
return self.decoder.decode(bytes(body))
|
||||||
|
|||||||
@@ -41,6 +41,18 @@ async def _close_async_response(resp: AsyncResponse) -> None:
|
|||||||
return
|
return
|
||||||
|
|
||||||
|
|
||||||
|
async def read_response_body(resp: object) -> bytes | str | None:
|
||||||
|
"""Read response body; niquests ``AsyncResponse.content`` is an async property."""
|
||||||
|
content = getattr(resp, "content", None)
|
||||||
|
if inspect.isawaitable(content):
|
||||||
|
content = await content
|
||||||
|
if isinstance(content, bytes | bytearray):
|
||||||
|
return bytes(content)
|
||||||
|
if isinstance(content, str):
|
||||||
|
return content
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def sse_data_payload(line: bytes) -> bytes | None | Literal[False]:
|
def sse_data_payload(line: bytes) -> bytes | None | Literal[False]:
|
||||||
"""Parse one SSE line.
|
"""Parse one SSE line.
|
||||||
|
|
||||||
@@ -98,12 +110,11 @@ class BaseOpenAIClient:
|
|||||||
status = getattr(resp, "status_code", None)
|
status = getattr(resp, "status_code", None)
|
||||||
if not isinstance(status, int):
|
if not isinstance(status, int):
|
||||||
return
|
return
|
||||||
content = getattr(resp, "content", None)
|
# Only pull the body for error responses — success streams must not
|
||||||
|
# await content (that would consume the SSE body).
|
||||||
body: bytes | str | None = None
|
body: bytes | str | None = None
|
||||||
if isinstance(content, bytes | bytearray):
|
if status >= _HTTP_ERROR:
|
||||||
body = bytes(content)
|
body = await read_response_body(resp)
|
||||||
elif isinstance(content, str):
|
|
||||||
body = content
|
|
||||||
msg = http_error_message(status, body)
|
msg = http_error_message(status, body)
|
||||||
if msg is None:
|
if msg is None:
|
||||||
return
|
return
|
||||||
@@ -163,8 +174,14 @@ class OpenAIClient(BaseOpenAIClient):
|
|||||||
stream=False,
|
stream=False,
|
||||||
)
|
)
|
||||||
await self._ensure_ok(resp)
|
await self._ensure_ok(resp)
|
||||||
assert resp.content is not None
|
body = await read_response_body(resp)
|
||||||
return self.decoder.decode(resp.content)
|
if body is None:
|
||||||
|
msg = "chat completions response body is empty"
|
||||||
|
raise RuntimeError(msg)
|
||||||
|
if not isinstance(body, (bytes, bytearray)):
|
||||||
|
msg = f"chat completions response body has unexpected type {type(body)!r}"
|
||||||
|
raise TypeError(msg)
|
||||||
|
return self.decoder.decode(bytes(body))
|
||||||
|
|
||||||
|
|
||||||
def merge_stream_tool_calls(deltas: list[StreamToolCallDelta]) -> list[AssistantFunctionToolCall]:
|
def merge_stream_tool_calls(deltas: list[StreamToolCallDelta]) -> list[AssistantFunctionToolCall]:
|
||||||
|
|||||||
@@ -1,6 +1,10 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from plyngent.lmproto.openai_compatible.client import http_error_message, sse_data_payload
|
from plyngent.lmproto.openai_compatible.client import (
|
||||||
|
http_error_message,
|
||||||
|
read_response_body,
|
||||||
|
sse_data_payload,
|
||||||
|
)
|
||||||
from plyngent.lmproto.openai_compatible.model import ChatCompletionChunk
|
from plyngent.lmproto.openai_compatible.model import ChatCompletionChunk
|
||||||
|
|
||||||
|
|
||||||
@@ -69,3 +73,20 @@ def test_http_error_message() -> None:
|
|||||||
assert err503 is not None
|
assert err503 is not None
|
||||||
assert "503" in err503
|
assert "503" in err503
|
||||||
assert "unavailable" in err503
|
assert "unavailable" in err503
|
||||||
|
|
||||||
|
|
||||||
|
async def test_response_body_awaits_async_content() -> None:
|
||||||
|
class AsyncContentResp:
|
||||||
|
@property
|
||||||
|
async def content(self) -> bytes:
|
||||||
|
return b'{"ok":true}'
|
||||||
|
|
||||||
|
body = await read_response_body(AsyncContentResp())
|
||||||
|
assert body == b'{"ok":true}'
|
||||||
|
|
||||||
|
|
||||||
|
async def test_response_body_sync_bytes() -> None:
|
||||||
|
class SyncContentResp:
|
||||||
|
content = b"plain"
|
||||||
|
|
||||||
|
assert await read_response_body(SyncContentResp()) == b"plain"
|
||||||
|
|||||||
Reference in New Issue
Block a user