feat: harden public OpenAI API contracts
This commit is contained in:
Xin Wang
@@ -0,0 +1,314 @@
|
||||
"""Pydantic contract for the supported Responses API surface."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Annotated, Any, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
|
||||
|
||||
ItemStatus = Literal["in_progress", "completed", "incomplete"]
|
||||
ResponseStatus = Literal[
|
||||
"in_progress",
|
||||
"completed",
|
||||
"failed",
|
||||
"cancelled",
|
||||
"incomplete",
|
||||
]
|
||||
|
||||
|
||||
class _StrictModel(BaseModel):
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
|
||||
class InputText(_StrictModel):
|
||||
type: Literal["input_text"] = "input_text"
|
||||
text: str
|
||||
|
||||
|
||||
class OutputTextInput(_StrictModel):
|
||||
type: Literal["output_text"] = "output_text"
|
||||
text: str
|
||||
annotations: list[dict[str, Any]] = Field(default_factory=list)
|
||||
|
||||
|
||||
MessageContent = Annotated[
|
||||
InputText | OutputTextInput,
|
||||
Field(discriminator="type"),
|
||||
]
|
||||
|
||||
|
||||
class ResponseInputMessage(_StrictModel):
|
||||
type: Literal["message"] = "message"
|
||||
role: Literal["user", "assistant"] = "user"
|
||||
content: str | list[MessageContent]
|
||||
|
||||
|
||||
class ResponseFunctionCallOutputInput(_StrictModel):
|
||||
type: Literal["function_call_output"] = "function_call_output"
|
||||
call_id: str
|
||||
output: Any
|
||||
|
||||
|
||||
ResponseInputItem = Annotated[
|
||||
ResponseInputMessage | ResponseFunctionCallOutputInput,
|
||||
Field(discriminator="type"),
|
||||
]
|
||||
|
||||
|
||||
class InteractiveMediaRequestExtension(_StrictModel):
|
||||
version: Literal["1"] | None = Field(
|
||||
default=None,
|
||||
description="Interactive Media extension version. Omit or use `1`.",
|
||||
)
|
||||
variables: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
|
||||
class ResponseCreateRequest(BaseModel):
|
||||
"""Supported request fields; extras are preserved for semantic rejection."""
|
||||
|
||||
model_config = ConfigDict(
|
||||
extra="allow",
|
||||
json_schema_extra={
|
||||
"examples": [
|
||||
{
|
||||
"model": "assistant:asst_xxx",
|
||||
"input": "请介绍一下这个产品",
|
||||
"store": True,
|
||||
},
|
||||
{
|
||||
"previous_response_id": "resp_xxx",
|
||||
"input": "继续说明",
|
||||
"stream": True,
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
model: str | None = Field(
|
||||
default=None,
|
||||
examples=["assistant:asst_xxx"],
|
||||
description="Required on the first turn; inherited on later turns.",
|
||||
)
|
||||
previous_response_id: str | None = None
|
||||
input: str | list[ResponseInputItem]
|
||||
store: bool | None = Field(
|
||||
default=None,
|
||||
description="Stateful subset: omit this field or set it to true.",
|
||||
)
|
||||
stream: bool = False
|
||||
x_interactive_media: InteractiveMediaRequestExtension | None = None
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def default_message_discriminator(cls, value: Any) -> Any:
|
||||
"""Accept the official SDK's easy input message with omitted `type`."""
|
||||
|
||||
if not isinstance(value, dict) or not isinstance(value.get("input"), list):
|
||||
return value
|
||||
copied = dict(value)
|
||||
copied["input"] = [
|
||||
({"type": "message", **item} if isinstance(item, dict) and "type" not in item else item)
|
||||
for item in value["input"]
|
||||
]
|
||||
return copied
|
||||
|
||||
|
||||
class ResponseOutputText(_StrictModel):
|
||||
type: Literal["output_text"] = "output_text"
|
||||
text: str
|
||||
annotations: list[dict[str, Any]] = Field(default_factory=list)
|
||||
logprobs: list[dict[str, Any]] | None = None
|
||||
|
||||
|
||||
class ResponseOutputMessage(_StrictModel):
|
||||
id: str
|
||||
type: Literal["message"] = "message"
|
||||
status: ItemStatus
|
||||
role: Literal["assistant"] = "assistant"
|
||||
content: list[ResponseOutputText]
|
||||
|
||||
|
||||
class ResponseFunctionCall(_StrictModel):
|
||||
id: str
|
||||
type: Literal["function_call"] = "function_call"
|
||||
status: ItemStatus
|
||||
call_id: str
|
||||
name: str
|
||||
arguments: str
|
||||
|
||||
|
||||
ResponseOutputItem = Annotated[
|
||||
ResponseOutputMessage | ResponseFunctionCall,
|
||||
Field(discriminator="type"),
|
||||
]
|
||||
|
||||
|
||||
class ResponseError(_StrictModel):
|
||||
code: str
|
||||
message: str
|
||||
|
||||
|
||||
class ResponseUsage(_StrictModel):
|
||||
input_tokens: int
|
||||
output_tokens: int
|
||||
total_tokens: int
|
||||
input_tokens_details: dict[str, Any] = Field(default_factory=dict)
|
||||
output_tokens_details: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
|
||||
class ResponseObject(_StrictModel):
|
||||
model_config = ConfigDict(
|
||||
extra="forbid",
|
||||
json_schema_extra={
|
||||
"examples": [
|
||||
{
|
||||
"id": "resp_xxx",
|
||||
"object": "response",
|
||||
"created_at": 1788739200,
|
||||
"completed_at": 1788739201,
|
||||
"status": "completed",
|
||||
"error": None,
|
||||
"incomplete_details": None,
|
||||
"instructions": None,
|
||||
"model": "assistant:asst_xxx",
|
||||
"output": [
|
||||
{
|
||||
"id": "msg_xxx",
|
||||
"type": "message",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{
|
||||
"type": "output_text",
|
||||
"text": "这是一个实时对话助手。",
|
||||
"annotations": [],
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
"parallel_tool_calls": False,
|
||||
"previous_response_id": None,
|
||||
"store": True,
|
||||
"tool_choice": "auto",
|
||||
"tools": [],
|
||||
"metadata": {},
|
||||
"usage": None,
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
id: str
|
||||
object: Literal["response"] = "response"
|
||||
created_at: int
|
||||
completed_at: int | None = None
|
||||
status: ResponseStatus
|
||||
error: ResponseError | None = None
|
||||
incomplete_details: dict[str, Any] | None = None
|
||||
instructions: None = None
|
||||
model: str | None
|
||||
output: list[ResponseOutputItem]
|
||||
parallel_tool_calls: Literal[False] = False
|
||||
previous_response_id: str | None = None
|
||||
store: Literal[True] = True
|
||||
tool_choice: Literal["auto"] = "auto"
|
||||
tools: list[Any] = Field(default_factory=list)
|
||||
metadata: dict[str, str] = Field(default_factory=dict)
|
||||
usage: ResponseUsage | None = None
|
||||
|
||||
|
||||
class ResponseInputMessageItem(_StrictModel):
|
||||
id: str
|
||||
type: Literal["message"] = "message"
|
||||
role: Literal["user", "assistant"]
|
||||
content: list[MessageContent]
|
||||
status: ItemStatus = "completed"
|
||||
|
||||
|
||||
class ResponseFunctionCallOutputItem(_StrictModel):
|
||||
id: str
|
||||
type: Literal["function_call_output"] = "function_call_output"
|
||||
call_id: str
|
||||
output: str
|
||||
status: ItemStatus = "completed"
|
||||
|
||||
|
||||
ResponseStoredInputItem = Annotated[
|
||||
ResponseInputMessageItem | ResponseFunctionCallOutputItem,
|
||||
Field(discriminator="type"),
|
||||
]
|
||||
|
||||
|
||||
class ResponseInputItemList(_StrictModel):
|
||||
model_config = ConfigDict(
|
||||
extra="forbid",
|
||||
json_schema_extra={
|
||||
"examples": [
|
||||
{
|
||||
"object": "list",
|
||||
"data": [
|
||||
{
|
||||
"id": "msg_input_xxx",
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "你好"}
|
||||
],
|
||||
"status": "completed",
|
||||
}
|
||||
],
|
||||
"first_id": "msg_input_xxx",
|
||||
"last_id": "msg_input_xxx",
|
||||
"has_more": False,
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
object: Literal["list"] = "list"
|
||||
data: list[ResponseStoredInputItem]
|
||||
first_id: str | None = None
|
||||
last_id: str | None = None
|
||||
has_more: bool
|
||||
|
||||
|
||||
class ModelObject(_StrictModel):
|
||||
id: str
|
||||
object: Literal["model"] = "model"
|
||||
created: int
|
||||
owned_by: str = "interactive-media"
|
||||
name: str
|
||||
capabilities: list[Literal["responses", "realtime"]]
|
||||
assistant_id: str
|
||||
assistant_type: str
|
||||
runtime_mode: str
|
||||
|
||||
|
||||
class ModelList(_StrictModel):
|
||||
model_config = ConfigDict(
|
||||
extra="forbid",
|
||||
json_schema_extra={
|
||||
"examples": [
|
||||
{
|
||||
"object": "list",
|
||||
"data": [
|
||||
{
|
||||
"id": "assistant:asst_xxx",
|
||||
"object": "model",
|
||||
"created": 1788739200,
|
||||
"owned_by": "interactive-media",
|
||||
"name": "产品顾问",
|
||||
"capabilities": ["responses", "realtime"],
|
||||
"assistant_id": "asst_xxx",
|
||||
"assistant_type": "prompt",
|
||||
"runtime_mode": "pipeline",
|
||||
}
|
||||
],
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
object: Literal["list"] = "list"
|
||||
data: list[ModelObject]
|
||||
Reference in New Issue
Block a user