core/lmproto: add OpenAI Responses API models and client

POST/GET/DELETE /responses with stream events; helpers for output_text and function_call items.
This commit is contained in:
2026-07-15 20:43:11 +08:00
parent 61582ffd7d
commit 3e20b1559c
5 changed files with 590 additions and 24 deletions
+2 -1
View File
@@ -34,7 +34,8 @@ lmproto/deepseek/openai_compat/ ← Extends base via inheritance + extra fields
```
- **`openai_compatible/model.py`** — tagged chat messages (`SystemChatMessage`, `UserChatMessage`, …), tools, request/response, streaming chunks.
- **`openai_compatible/client.py`** — `BaseOpenAIClient` / `OpenAIClient` via `niquests` async + SSE.
- **`openai_compatible/responses_model.py`** — OpenAI Responses API (`ResponsesCreateParam`, `Response`, function_call items, stream events).
- **`openai_compatible/client.py`** — `BaseOpenAIClient` / `OpenAIClient` via `niquests` async + SSE; `chat_completions`, `models`, `responses` / `get_response` / `delete_response`.
- **`openai_compatible/config.py`** — `OpenAIConfig` (token + base URL).
- **DeepSeek** — `DeepseekOpenAIClient`; models add `reasoning_content`, `prefix`, `ThinkingOptions`. Config default model ids: `deepseek-v4-flash`, `deepseek-v4-pro`.
@@ -40,3 +40,14 @@ from .model import ToolFunctionItem as ToolFunctionItem
from .model import UserChatMessage as UserChatMessage
from .model import Verbosity as Verbosity
from .model import VoiceName as VoiceName
from .responses_model import Response as Response
from .responses_model import ResponseDeleted as ResponseDeleted
from .responses_model import ResponseFunctionTool as ResponseFunctionTool
from .responses_model import ResponseFunctionToolCall as ResponseFunctionToolCall
from .responses_model import ResponseFunctionToolCallOutput as ResponseFunctionToolCallOutput
from .responses_model import ResponseOutputMessage as ResponseOutputMessage
from .responses_model import ResponseOutputText as ResponseOutputText
from .responses_model import ResponsesCreateParam as ResponsesCreateParam
from .responses_model import ResponseStreamEvent as ResponseStreamEvent
from .responses_model import response_function_calls as response_function_calls
from .responses_model import response_output_text as response_output_text
@@ -18,6 +18,7 @@ from .model import (
ModelsResponse,
StreamToolCallDelta,
)
from .responses_model import Response, ResponseDeleted, ResponsesCreateParam, ResponseStreamEvent
if TYPE_CHECKING:
from collections.abc import AsyncIterator
@@ -77,11 +78,16 @@ def sse_data_payload(line: bytes) -> bytes | None | Literal[False]:
return payload
def http_error_message(status_code: int, body: bytes | str | None = None) -> str | None:
def http_error_message(
status_code: int,
body: bytes | str | None = None,
*,
what: str = "request",
) -> str | None:
"""Return an error string for HTTP ``status_code`` >= 400, else ``None``."""
if status_code < _HTTP_ERROR:
return None
msg = f"chat completions HTTP {status_code}"
msg = f"{what} HTTP {status_code}"
if body is None:
return msg
if isinstance(body, (bytes, bytearray)):
@@ -97,6 +103,9 @@ class BaseOpenAIClient:
decoder: msgspec.json.Decoder[ChatCompletionResponse]
chunk_decoder: msgspec.json.Decoder[ChatCompletionChunk]
models_decoder: msgspec.json.Decoder[ModelsResponse]
response_decoder: msgspec.json.Decoder[Response]
response_deleted_decoder: msgspec.json.Decoder[ResponseDeleted]
response_event_decoder: msgspec.json.Decoder[ResponseStreamEvent]
def __init__(self, config: OpenAIConfig) -> None:
self.session = niquests.AsyncSession(
@@ -107,6 +116,9 @@ class BaseOpenAIClient:
self.decoder = msgspec.json.Decoder(ChatCompletionResponse)
self.chunk_decoder = msgspec.json.Decoder(ChatCompletionChunk)
self.models_decoder = msgspec.json.Decoder(ModelsResponse)
self.response_decoder = msgspec.json.Decoder(Response)
self.response_deleted_decoder = msgspec.json.Decoder(ResponseDeleted)
self.response_event_decoder = msgspec.json.Decoder(ResponseStreamEvent)
async def models(self) -> list[str]:
"""List model ids via OpenAI-compatible ``GET /models``.
@@ -114,19 +126,12 @@ class BaseOpenAIClient:
Returns sorted unique ``id`` values from the response ``data`` array.
"""
resp = await self.session.get("/models", stream=False)
await self._ensure_ok(resp)
body = await read_response_body(resp)
if body is None:
msg = "models response body is empty"
raise RuntimeError(msg)
if not isinstance(body, (bytes, bytearray)):
msg = f"models response body has unexpected type {type(body)!r}"
raise TypeError(msg)
parsed = self.models_decoder.decode(bytes(body))
body = await self._read_json_body(resp, what="models")
parsed = self.models_decoder.decode(body)
ids = {item.id for item in parsed.data if item.id}
return sorted(ids)
async def _ensure_ok(self, resp: object) -> None:
async def _ensure_ok(self, resp: object, *, what: str = "request") -> None:
"""Raise if the HTTP response is an error (stream or non-stream)."""
status = getattr(resp, "status_code", None)
if not isinstance(status, int):
@@ -136,7 +141,7 @@ class BaseOpenAIClient:
body: bytes | str | None = None
if status >= _HTTP_ERROR:
body = await read_response_body(resp)
msg = http_error_message(status, body)
msg = http_error_message(status, body, what=what)
if msg is None:
return
if hasattr(resp, "close") or hasattr(resp, "aclose"):
@@ -146,7 +151,7 @@ class BaseOpenAIClient:
async def _parse_sse(self, resp: AsyncResponse) -> AsyncIterator[ChatCompletionChunk]:
"""Yield SSE chunks; stop at ``data: [DONE]`` so we do not hang on keep-alive."""
try:
await self._ensure_ok(resp)
await self._ensure_ok(resp, what="chat completions")
async for raw in resp.iter_lines():
if not raw:
continue
@@ -159,6 +164,33 @@ class BaseOpenAIClient:
finally:
await _close_async_response(resp)
async def _parse_response_sse(self, resp: AsyncResponse) -> AsyncIterator[ResponseStreamEvent]:
"""Yield Responses API SSE events; stop at ``data: [DONE]``."""
try:
await self._ensure_ok(resp, what="responses")
async for raw in resp.iter_lines():
if not raw:
continue
parsed = sse_data_payload(bytes(raw))
if parsed is None:
continue
if parsed is False:
break
yield self.response_event_decoder.decode(parsed)
finally:
await _close_async_response(resp)
async def _read_json_body(self, resp: object, *, what: str) -> bytes:
await self._ensure_ok(resp, what=what)
body = await read_response_body(resp)
if body is None:
msg = f"{what} response body is empty"
raise RuntimeError(msg)
if not isinstance(body, (bytes, bytearray)):
msg = f"{what} response body has unexpected type {type(body)!r}"
raise TypeError(msg)
return bytes(body)
class OpenAIClient(BaseOpenAIClient):
def __init__(self, config: OpenAIConfig) -> None:
@@ -194,15 +226,53 @@ class OpenAIClient(BaseOpenAIClient):
headers={"Content-Type": "application/json"},
stream=False,
)
await self._ensure_ok(resp)
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))
body = await self._read_json_body(resp, what="chat completions")
return self.decoder.decode(body)
@overload
async def responses(
self, param: ResponsesCreateParam, *, stream: Literal[False] = False
) -> Response: ...
@overload
async def responses(
self, param: ResponsesCreateParam, *, stream: Literal[True]
) -> AsyncIterator[ResponseStreamEvent]: ...
async def responses(
self, param: ResponsesCreateParam, *, stream: bool = False
) -> Response | AsyncIterator[ResponseStreamEvent]:
"""Create a model response via OpenAI ``POST /responses``."""
param = msgspec.structs.replace(param, stream=stream)
data = self.encoder.encode(param)
if stream:
resp = await self.session.post(
"/responses",
data=data,
headers={"Content-Type": "application/json"},
stream=True,
)
return self._parse_response_sse(resp)
resp = await self.session.post(
"/responses",
data=data,
headers={"Content-Type": "application/json"},
stream=False,
)
body = await self._read_json_body(resp, what="responses")
return self.response_decoder.decode(body)
async def get_response(self, response_id: str) -> Response:
"""Retrieve a stored response via ``GET /responses/{id}``."""
resp = await self.session.get(f"/responses/{response_id}", stream=False)
body = await self._read_json_body(resp, what="responses")
return self.response_decoder.decode(body)
async def delete_response(self, response_id: str) -> ResponseDeleted:
"""Delete a stored response via ``DELETE /responses/{id}``."""
resp = await self.session.delete(f"/responses/{response_id}", stream=False)
body = await self._read_json_body(resp, what="responses")
return self.response_deleted_decoder.decode(body)
def merge_stream_tool_calls(deltas: list[StreamToolCallDelta]) -> list[AssistantFunctionToolCall]:
@@ -0,0 +1,326 @@
"""OpenAI Responses API (``POST /responses``) msgspec models.
Covers the common text + function-calling surface used by agents. Built-in
tools (web_search, file_search, …) may appear as extra fields on decode when
``omit_defaults`` / unknown fields allow; request construction for those is
left to callers via generic dicts where needed.
"""
from __future__ import annotations
from typing import Any, Literal, cast
import msgspec
from msgspec import UNSET, Struct, field
from plyngent.typedef import JSONSchema, Unset # noqa: TC001
from .model import ServiceTier # noqa: TC001
type ResponseStatus = Literal[
"completed",
"failed",
"in_progress",
"cancelled",
"queued",
"incomplete",
]
type ResponseItemStatus = Literal["in_progress", "completed", "incomplete"]
type ReasoningEffort = Literal["none", "minimal", "low", "medium", "high", "xhigh"]
type TruncationMode = Literal["auto", "disabled"]
type PromptCacheRetention = Literal["in_memory", "24h"]
# --- Input / output content blocks -------------------------------------------------
class ResponseInputText(Struct, tag_field="type", tag="input_text"):
text: str
class ResponseOutputText(Struct, tag_field="type", tag="output_text"):
text: str
annotations: list[dict[str, Any]] = field(default_factory=list)
logprobs: list[dict[str, Any]] | Unset = UNSET
class ResponseOutputRefusal(Struct, tag_field="type", tag="refusal"):
refusal: str
type ResponseMessageContent = ResponseOutputText | ResponseOutputRefusal | ResponseInputText
# --- Message-shaped items (input or output) ----------------------------------------
class ResponseEasyInputMessage(Struct, tag_field="type", tag="message"):
"""Message item used in ``input`` (role + content string or parts)."""
role: Literal["system", "developer", "user", "assistant"]
content: str | list[dict[str, Any]]
# Output messages from the API also use type=message with id/status.
id: str | Unset = UNSET
status: ResponseItemStatus | Unset = UNSET
phase: Literal["commentary", "final_answer"] | Unset = UNSET
class ResponseOutputMessage(Struct, tag_field="type", tag="message"):
"""Assistant message item in ``response.output``."""
id: str
content: list[ResponseMessageContent]
role: Literal["assistant"] = "assistant"
status: ResponseItemStatus = "completed"
phase: Literal["commentary", "final_answer"] | Unset = UNSET
# --- Function calling (Responses shape: flat, not nested under ``function``) -------
class ResponseFunctionTool(Struct, tag_field="type", tag="function"):
"""Function tool definition for Responses ``tools`` array."""
name: str
description: str | Unset = UNSET
parameters: JSONSchema | Unset = UNSET
strict: bool | Unset = UNSET
class ResponseFunctionToolCall(Struct, tag_field="type", tag="function_call"):
"""Model-emitted function call item."""
call_id: str
name: str
arguments: str
id: str | Unset = UNSET
status: ResponseItemStatus | Unset = UNSET
namespace: str | Unset = UNSET
class ResponseFunctionToolCallOutput(Struct, tag_field="type", tag="function_call_output"):
"""Tool result item passed back in subsequent ``input``."""
call_id: str
output: str
id: str | Unset = UNSET
status: ResponseItemStatus | Unset = UNSET
# --- Reasoning item ----------------------------------------------------------------
class ResponseReasoningItem(Struct, tag_field="type", tag="reasoning"):
id: str
content: list[dict[str, Any]] = field(default_factory=list)
summary: list[dict[str, Any]] = field(default_factory=list)
status: ResponseItemStatus | Unset = UNSET
encrypted_content: str | Unset = UNSET
# --- Catch-all for other output item types (web_search_call, …) ---------------------
class ResponseUnknownItem(Struct):
"""Fallback for output/input items we do not model in full yet."""
type: str
id: str | Unset = UNSET
type ResponseOutputItem = (
ResponseOutputMessage
| ResponseFunctionToolCall
| ResponseReasoningItem
| ResponseFunctionToolCallOutput
| ResponseEasyInputMessage
)
# --- Request -----------------------------------------------------------------------
class ResponseReasoningConfig(Struct, omit_defaults=True):
effort: ReasoningEffort | Unset = UNSET
summary: Literal["auto", "concise", "detailed"] | Unset = UNSET
class ResponseTextFormatJsonSchema(Struct, tag_field="type", tag="json_schema"):
name: str
schema: JSONSchema
strict: bool | Unset = UNSET
description: str | Unset = UNSET
class ResponseTextFormatText(Struct, tag_field="type", tag="text"):
pass
class ResponseTextFormatJsonObject(Struct, tag_field="type", tag="json_object"):
pass
type ResponseTextFormat = ResponseTextFormatText | ResponseTextFormatJsonObject | ResponseTextFormatJsonSchema
class ResponseTextConfig(Struct, omit_defaults=True):
format: ResponseTextFormat | Unset = UNSET
verbosity: Literal["low", "medium", "high"] | Unset = UNSET
class ResponseStreamOptions(Struct, omit_defaults=True):
include_obfuscation: bool | Unset = UNSET
class ResponsesCreateParam(Struct, omit_defaults=True):
"""Body for ``POST /responses``."""
model: str
# string, or list of message / function_call_output / … items
input: str | list[dict[str, Any] | ResponseEasyInputMessage | ResponseFunctionToolCallOutput]
instructions: str | Unset = UNSET
tools: list[ResponseFunctionTool | dict[str, Any]] | Unset = UNSET
tool_choice: str | dict[str, Any] | Unset = UNSET
parallel_tool_calls: bool | Unset = UNSET
previous_response_id: str | Unset = UNSET
store: bool | Unset = UNSET
stream: bool = False
stream_options: ResponseStreamOptions | Unset = UNSET
temperature: float | Unset = UNSET
top_p: float | Unset = UNSET
max_output_tokens: int | Unset = UNSET
max_tool_calls: int | Unset = UNSET
metadata: dict[str, str] | Unset = UNSET
reasoning: ResponseReasoningConfig | Unset = UNSET
text: ResponseTextConfig | Unset = UNSET
truncation: TruncationMode | Unset = UNSET
service_tier: ServiceTier | Unset = UNSET
user: str | Unset = UNSET
safety_identifier: str | Unset = UNSET
prompt_cache_key: str | Unset = UNSET
prompt_cache_retention: PromptCacheRetention | Unset = UNSET
include: list[str] | Unset = UNSET
background: bool | Unset = UNSET
# --- Response object ---------------------------------------------------------------
class ResponseIncompleteDetails(Struct, omit_defaults=True):
reason: Literal["max_output_tokens", "content_filter"] | Unset = UNSET
class ResponseError(Struct, omit_defaults=True):
code: str | Unset = UNSET
message: str | Unset = UNSET
class ResponseUsage(Struct, omit_defaults=True):
input_tokens: int | Unset = UNSET
output_tokens: int | Unset = UNSET
total_tokens: int | Unset = UNSET
input_tokens_details: dict[str, Any] | Unset = UNSET
output_tokens_details: dict[str, Any] | Unset = UNSET
class Response(Struct, omit_defaults=True):
"""``object: response`` from create/retrieve."""
id: str
created_at: float | int
model: str
# Keep items as dicts so unknown built-in tool item types still decode.
output: list[dict[str, Any]]
object: Literal["response"] = "response"
status: ResponseStatus | Unset = UNSET
error: ResponseError | None | Unset = UNSET
incomplete_details: ResponseIncompleteDetails | None | Unset = UNSET
instructions: str | list[dict[str, Any]] | None | Unset = UNSET
metadata: dict[str, str] | None | Unset = UNSET
parallel_tool_calls: bool | Unset = UNSET
temperature: float | None | Unset = UNSET
top_p: float | None | Unset = UNSET
tools: list[dict[str, Any]] | Unset = UNSET
tool_choice: str | dict[str, Any] | Unset = UNSET
# Prefer loose dict — nested usage details vary by model/API version.
usage: dict[str, Any] | None | Unset = UNSET
previous_response_id: str | None | Unset = UNSET
service_tier: ServiceTier | None | Unset = UNSET
truncation: TruncationMode | None | Unset = UNSET
text: dict[str, Any] | Unset = UNSET
reasoning: dict[str, Any] | None | Unset = UNSET
store: bool | Unset = UNSET
user: str | None | Unset = UNSET
class ResponseDeleted(Struct):
id: str
object: Literal["response.deleted"] = "response.deleted"
deleted: bool = True
# --- Streaming events (subset) -----------------------------------------------------
class ResponseStreamEvent(Struct, omit_defaults=True):
"""Generic SSE event for Responses streaming.
OpenAI emits many event types; we keep common fields optional and let
callers branch on ``type``. Nested ``response`` is a dict so unknown
fields and evolving shapes still decode.
"""
type: str
response: dict[str, Any] | Unset = UNSET
item_id: str | Unset = UNSET
output_index: int | Unset = UNSET
content_index: int | Unset = UNSET
delta: str | Unset = UNSET
text: str | Unset = UNSET
item: dict[str, Any] | Unset = UNSET
sequence_number: int | Unset = UNSET
# function call argument streaming
name: str | Unset = UNSET
arguments: str | Unset = UNSET
call_id: str | Unset = UNSET
# --- Helpers -----------------------------------------------------------------------
def response_output_text(response: Response) -> str:
"""Concatenate all ``output_text`` blocks from message items (SDK-style helper)."""
parts: list[str] = []
for raw in response.output:
if raw.get("type") != "message":
continue
content = raw.get("content")
if not isinstance(content, list):
continue
for block_obj in content:
if not isinstance(block_obj, dict):
continue
block_map = cast("dict[str, object]", block_obj)
if block_map.get("type") != "output_text":
continue
text = block_map.get("text")
if isinstance(text, str):
parts.append(text)
return "".join(parts)
def response_function_calls(response: Response) -> list[ResponseFunctionToolCall]:
"""Decode ``function_call`` items from ``response.output``."""
result: list[ResponseFunctionToolCall] = []
for raw in response.output:
if raw.get("type") != "function_call":
continue
try:
result.append(msgspec_convert_function_call(raw))
except (TypeError, ValueError, KeyError, msgspec.ValidationError):
continue
return result
def msgspec_convert_function_call(raw: dict[str, Any]) -> ResponseFunctionToolCall:
return msgspec.convert(raw, ResponseFunctionToolCall)
+158
View File
@@ -0,0 +1,158 @@
from __future__ import annotations
import msgspec
import pytest
from plyngent.lmproto.openai_compatible.client import OpenAIClient
from plyngent.lmproto.openai_compatible.config import OpenAIConfig
from plyngent.lmproto.openai_compatible.responses_model import (
Response,
ResponseDeleted,
ResponseFunctionTool,
ResponseFunctionToolCall,
ResponsesCreateParam,
ResponseStreamEvent,
response_function_calls,
response_output_text,
)
def _sample_response_body() -> bytes:
return (
b'{"id":"resp_1","object":"response","created_at":1,"model":"gpt-test",'
b'"status":"completed","output":['
b'{"id":"rs_1","type":"reasoning","content":[],"summary":[]},'
b'{"id":"msg_1","type":"message","role":"assistant","status":"completed",'
b'"content":[{"type":"output_text","text":"hello world","annotations":[]}]},'
b'{"type":"function_call","call_id":"call_1","name":"add",'
b'"arguments":"{\\"a\\":1}","status":"completed"}'
b"]}"
)
def test_response_decode_and_helpers() -> None:
parsed = msgspec.json.decode(_sample_response_body(), type=Response)
assert parsed.id == "resp_1"
assert parsed.object == "response"
assert response_output_text(parsed) == "hello world"
calls = response_function_calls(parsed)
assert len(calls) == 1
assert isinstance(calls[0], ResponseFunctionToolCall)
assert calls[0].name == "add"
assert calls[0].call_id == "call_1"
def test_create_param_encode_omits_defaults() -> None:
param = ResponsesCreateParam(
model="gpt-test",
input="hi",
tools=[ResponseFunctionTool(name="add", parameters={"type": "object"})],
)
raw = msgspec.json.encode(param)
data = msgspec.json.decode(raw)
assert data["model"] == "gpt-test"
assert data["input"] == "hi"
# omit_defaults: default stream=False is not encoded until client sets stream
assert "stream" not in data
assert data["tools"][0]["type"] == "function"
assert data["tools"][0]["name"] == "add"
def test_stream_event_decode() -> None:
raw = (
b'{"type":"response.output_text.delta","item_id":"msg_1",'
b'"output_index":0,"content_index":0,"delta":"hel"}'
)
event = msgspec.json.decode(raw, type=ResponseStreamEvent)
assert event.type == "response.output_text.delta"
assert event.delta == "hel"
@pytest.mark.asyncio
async def test_client_responses_create(monkeypatch: pytest.MonkeyPatch) -> None:
client = OpenAIClient(OpenAIConfig(access_key_or_token="sk", base_url="https://example/v1"))
class _Resp:
status_code = 200
@property
def content(self) -> bytes:
return _sample_response_body()
async def fake_post(path: str, **kwargs: object) -> _Resp:
assert path == "/responses"
assert kwargs.get("stream") is False
return _Resp()
monkeypatch.setattr(client.session, "post", fake_post)
result = await client.responses(ResponsesCreateParam(model="gpt-test", input="hi"))
assert isinstance(result, Response)
assert response_output_text(result) == "hello world"
@pytest.mark.asyncio
async def test_client_responses_stream(monkeypatch: pytest.MonkeyPatch) -> None:
client = OpenAIClient(OpenAIConfig(access_key_or_token="sk", base_url="https://example/v1"))
class _Resp:
status_code = 200
async def iter_lines(self):
yield (
b'data: {"type":"response.output_text.delta","delta":"a",'
b'"item_id":"m","output_index":0,"content_index":0}'
)
yield b"data: [DONE]"
def close(self) -> None:
return None
async def fake_post(path: str, **kwargs: object) -> _Resp:
assert path == "/responses"
assert kwargs.get("stream") is True
return _Resp()
monkeypatch.setattr(client.session, "post", fake_post)
stream = await client.responses(
ResponsesCreateParam(model="gpt-test", input="hi"),
stream=True,
)
events = [event async for event in stream]
assert len(events) == 1
assert events[0].type == "response.output_text.delta"
assert events[0].delta == "a"
@pytest.mark.asyncio
async def test_client_get_delete_response(monkeypatch: pytest.MonkeyPatch) -> None:
client = OpenAIClient(OpenAIConfig(access_key_or_token="sk", base_url="https://example/v1"))
class _Get:
status_code = 200
@property
def content(self) -> bytes:
return _sample_response_body()
class _Del:
status_code = 200
@property
def content(self) -> bytes:
return b'{"id":"resp_1","object":"response.deleted","deleted":true}'
async def fake_get(path: str, **kwargs: object) -> _Get:
assert path == "/responses/resp_1"
return _Get()
async def fake_delete(path: str, **kwargs: object) -> _Del:
assert path == "/responses/resp_1"
return _Del()
monkeypatch.setattr(client.session, "get", fake_get)
monkeypatch.setattr(client.session, "delete", fake_delete)
got = await client.get_response("resp_1")
assert got.id == "resp_1"
deleted = await client.delete_response("resp_1")
assert isinstance(deleted, ResponseDeleted)
assert deleted.deleted is True