Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 53 additions & 0 deletions e2e/inference_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,59 @@ func TestE2E_Inference(t *testing.T) {
t.Logf("responses API: %q", resp.OutputText)
})

t.Run("ResponsesAPIStructuredOutput", func(t *testing.T) {
status, body := doJSON(t, http.MethodPost, serverURL+"/responses",
map[string]any{
"model": bc.model,
"input": "Return a JSON object with answer set to yes.",
"max_output_tokens": 64,
"text": map[string]any{
"format": map[string]any{
"type": "json_schema",
"name": "answer_payload",
"strict": true,
"schema": map[string]any{
"type": "object",
"properties": map[string]any{
"answer": map[string]any{
"type": "string",
"enum": []string{"yes"},
},
},
"required": []string{"answer"},
"additionalProperties": false,
},
},
},
})
if status != http.StatusOK {
t.Fatalf("responses structured output: status=%d body=%s", status, body)
}
var resp struct {
Status string `json:"status"`
OutputText string `json:"output_text"`
}
if err := json.Unmarshal(body, &resp); err != nil {
t.Fatalf("decode: %v (body=%s)", err, body)
}
if resp.Status != "completed" {
t.Errorf("expected status=completed, got %q", resp.Status)
}

var output struct {
Answer string `json:"answer"`
}
if err := json.Unmarshal([]byte(resp.OutputText), &output); err != nil {
t.Fatalf("output_text is not valid structured JSON: %v (output_text=%q)", err, resp.OutputText)
}
if output.Answer == "" {
t.Fatalf("answer field is empty in output_text=%q", resp.OutputText)
}
if output.Answer != "yes" {
t.Fatalf("answer = %q, want yes (output_text=%q)", output.Answer, resp.OutputText)
}
})

t.Run("AnthropicMessages", func(t *testing.T) {
status, body := doJSON(t, http.MethodPost, serverURL+"/anthropic/v1/messages",
map[string]any{
Expand Down
54 changes: 54 additions & 0 deletions pkg/responses/api.go
Original file line number Diff line number Diff line change
Expand Up @@ -91,9 +91,42 @@ type CreateRequest struct {
// MaxOutputTokens limits the response length.
MaxOutputTokens *int `json:"max_output_tokens,omitempty"`

// Store controls whether the response is stored for later retrieval.
Store *bool `json:"store,omitempty"`

// Text configures text output formatting.
Text *ResponseTextConfig `json:"text,omitempty"`

// Include requests additional hosted-tool output fields. Unsupported locally.
Include []string `json:"include,omitempty"`

// Stream enables streaming responses.
Stream bool `json:"stream,omitempty"`

// StreamOptions configures response streaming. Unsupported locally.
StreamOptions json.RawMessage `json:"stream_options,omitempty"`

// TopLogprobs requests token log probabilities. Unsupported locally.
TopLogprobs *int `json:"top_logprobs,omitempty"`

// Truncation configures OpenAI context truncation behavior.
Truncation string `json:"truncation,omitempty"`

// Background requests asynchronous background response execution.
Background *bool `json:"background,omitempty"`

// Conversation requests OpenAI hosted conversation state. Unsupported locally.
Conversation json.RawMessage `json:"conversation,omitempty"`

// Prompt requests an OpenAI hosted prompt template. Unsupported locally.
Prompt json.RawMessage `json:"prompt,omitempty"`

// ServiceTier selects OpenAI service tier. Unsupported locally.
ServiceTier string `json:"service_tier,omitempty"`

// SafetyIdentifier is an OpenAI safety identifier. Unsupported locally.
SafetyIdentifier string `json:"safety_identifier,omitempty"`

// Metadata is user-defined metadata for the response.
Metadata map[string]string `json:"metadata,omitempty"`

Expand All @@ -104,6 +137,21 @@ type CreateRequest struct {
User string `json:"user,omitempty"`
}

// ResponseTextConfig configures text output for a Responses API request.
type ResponseTextConfig struct {
Format *ResponseTextFormat `json:"format,omitempty"`
Verbosity string `json:"verbosity,omitempty"`
}

// ResponseTextFormat configures the output text format.
type ResponseTextFormat struct {
Type string `json:"type"`
Name string `json:"name,omitempty"`
Description string `json:"description,omitempty"`
Schema json.RawMessage `json:"schema,omitempty"`
Strict *bool `json:"strict,omitempty"`
}

// Response represents a complete response from the API.
type Response struct {
// ID is the unique identifier for this response.
Expand Down Expand Up @@ -160,6 +208,12 @@ type Response struct {
// MaxOutputTokens limit used.
MaxOutputTokens *int `json:"max_output_tokens"`

// Store indicates whether the response was stored.
Store *bool `json:"store,omitempty"`

// Text config used for the response.
Text *ResponseTextConfig `json:"text,omitempty"`

// PreviousResponseID is the ID of the previous response in the chain.
PreviousResponseID *string `json:"previous_response_id"`

Expand Down
116 changes: 96 additions & 20 deletions pkg/responses/handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -33,23 +33,25 @@ func NewHTTPHandler(log logging.Logger, schedulerHTTP http.Handler, allowedOrigi
maxRequestBodyBytes: 10 * 1024 * 1024, // Default to 10MB
}

// Register routes
h.router.HandleFunc("POST "+APIPrefix, h.handleCreate)
h.router.HandleFunc("GET "+APIPrefix+"/{id}", h.handleGet)
h.router.HandleFunc("GET "+APIPrefix+"/{id}/input_items", h.handleListInputItems)
h.router.HandleFunc("DELETE "+APIPrefix+"/{id}", h.handleDelete)
// Also register /v1/responses routes
h.router.HandleFunc("POST /v1"+APIPrefix, h.handleCreate)
h.router.HandleFunc("GET /v1"+APIPrefix+"/{id}", h.handleGet)
h.router.HandleFunc("GET /v1"+APIPrefix+"/{id}/input_items", h.handleListInputItems)
h.router.HandleFunc("DELETE /v1"+APIPrefix+"/{id}", h.handleDelete)
h.registerRoutes(APIPrefix)
h.registerRoutes("/v1" + APIPrefix)
h.registerRoutes("/engines" + APIPrefix)
h.registerRoutes("/engines/v1" + APIPrefix)
h.registerRoutes("/engines/{backend}/v1" + APIPrefix)

// Apply CORS middleware
h.httpHandler = middleware.CorsMiddleware(allowedOrigins, h.router)

return h
}

func (h *HTTPHandler) registerRoutes(prefix string) {
h.router.HandleFunc("POST "+prefix, h.handleCreate)
h.router.HandleFunc("GET "+prefix+"/{id}", h.handleGet)
h.router.HandleFunc("GET "+prefix+"/{id}/input_items", h.handleListInputItems)
h.router.HandleFunc("DELETE "+prefix+"/{id}", h.handleDelete)
}

// Close releases resources held by the handler, including the background
// store cleanup goroutine. It should be called when the handler is shut down.
func (h *HTTPHandler) Close() {
Expand Down Expand Up @@ -97,14 +99,21 @@ func (h *HTTPHandler) handleCreate(w http.ResponseWriter, r *http.Request) {
h.sendError(w, http.StatusBadRequest, "invalid_request", "model is required")
return
}
if err := validateUnsupportedRequestFields(&req); err != nil {
h.sendError(w, http.StatusBadRequest, "invalid_request", err.Error())
return
}

// Create a new response
respID := GenerateResponseID()
resp := NewResponse(respID, req.Model)
store := shouldStore(&req)
resp.Instructions = nilIfEmpty(req.Instructions)
resp.Temperature = req.Temperature
resp.TopP = req.TopP
resp.MaxOutputTokens = req.MaxOutputTokens
resp.Store = &store
resp.Text = req.Text
resp.Tools = req.Tools
resp.ToolChoice = req.ToolChoice
resp.ParallelToolCalls = req.ParallelToolCalls
Expand Down Expand Up @@ -134,7 +143,7 @@ func (h *HTTPHandler) handleCreate(w http.ResponseWriter, r *http.Request) {
}

// Create upstream request
upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, "/engines/v1/chat/completions", bytes.NewReader(chatBody))
upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost, chatCompletionPathForRequest(r), bytes.NewReader(chatBody))
if err != nil {
h.sendError(w, http.StatusInternalServerError, "internal_error", "Failed to create request")
return
Expand All @@ -147,24 +156,29 @@ func (h *HTTPHandler) handleCreate(w http.ResponseWriter, r *http.Request) {

if req.Stream {
// Handle streaming response
h.handleStreaming(w, upstreamReq, resp)
h.handleStreaming(w, upstreamReq, resp, store)
} else {
// Handle non-streaming response
h.handleNonStreaming(w, upstreamReq, resp)
h.handleNonStreaming(w, upstreamReq, resp, store)
}
}

// handleStreaming handles streaming responses.
func (h *HTTPHandler) handleStreaming(w http.ResponseWriter, upstreamReq *http.Request, resp *Response) {
func (h *HTTPHandler) handleStreaming(w http.ResponseWriter, upstreamReq *http.Request, resp *Response, store bool) {
responseStore := h.store
if !store {
responseStore = nil
}

// Create streaming writer
streamWriter := NewStreamingResponseWriter(w, resp, h.store)
streamWriter := NewStreamingResponseWriter(w, resp, responseStore)

// Forward to scheduler
h.schedulerHTTP.ServeHTTP(streamWriter, upstreamReq)
}

// handleNonStreaming handles non-streaming responses.
func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *http.Request, resp *Response) {
func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *http.Request, resp *Response, store bool) {
// Capture upstream response
capture := NewNonStreamingResponseCapture()

Expand All @@ -187,7 +201,9 @@ func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *htt
Code: errResp.Error.Code,
Message: errResp.Error.Message,
}
h.store.Save(resp)
if store {
h.store.Save(resp)
}
h.sendJSON(w, capture.StatusCode, resp)
return
}
Expand All @@ -197,7 +213,9 @@ func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *htt
Code: "upstream_error",
Message: capture.Body.String(),
}
h.store.Save(resp)
if store {
h.store.Save(resp)
}
h.sendJSON(w, capture.StatusCode, resp)
return
}
Expand All @@ -210,7 +228,9 @@ func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *htt
Code: "parse_error",
Message: "Failed to parse upstream response",
}
h.store.Save(resp)
if store {
h.store.Save(resp)
}
h.sendJSON(w, http.StatusInternalServerError, resp)
return
}
Expand All @@ -222,6 +242,8 @@ func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *htt
finalResp.Temperature = resp.Temperature
finalResp.TopP = resp.TopP
finalResp.MaxOutputTokens = resp.MaxOutputTokens
finalResp.Store = resp.Store
finalResp.Text = resp.Text
finalResp.Tools = resp.Tools
finalResp.ToolChoice = resp.ToolChoice
finalResp.ParallelToolCalls = resp.ParallelToolCalls
Expand All @@ -232,7 +254,9 @@ func (h *HTTPHandler) handleNonStreaming(w http.ResponseWriter, upstreamReq *htt
finalResp.CreatedAt = resp.CreatedAt

// Store the response
h.store.Save(finalResp)
if store {
h.store.Save(finalResp)
}

// Send response
h.sendJSON(w, http.StatusOK, finalResp)
Expand Down Expand Up @@ -332,6 +356,58 @@ func nilIfEmpty(s string) *string {
return &s
}

func shouldStore(req *CreateRequest) bool {
return req.Store == nil || *req.Store
}

func chatCompletionPathForRequest(r *http.Request) string {
if backend := r.PathValue("backend"); backend != "" {
return "/engines/" + backend + "/v1/chat/completions"
}
return "/engines/v1/chat/completions"
}

func validateUnsupportedRequestFields(req *CreateRequest) error {
if len(req.Include) > 0 {
return unsupportedFieldError("include")
}
if req.StreamOptions != nil && !isNullJSON(req.StreamOptions) {
return unsupportedFieldError("stream_options")
}
if req.TopLogprobs != nil {
return unsupportedFieldError("top_logprobs")
}
switch req.Truncation {
case "", "disabled":
default:
return fmt.Errorf("truncation value %q is not supported by Docker Model Runner Responses compatibility layer", req.Truncation)
}
if req.Background != nil && *req.Background {
return fmt.Errorf("background responses are not supported by Docker Model Runner Responses compatibility layer")
}
if req.Conversation != nil && !isNullJSON(req.Conversation) {
return fmt.Errorf("conversation is not supported by Docker Model Runner Responses compatibility layer; use previous_response_id instead")
}
if req.Prompt != nil && !isNullJSON(req.Prompt) {
return unsupportedFieldError("prompt")
}
if req.ServiceTier != "" {
return unsupportedFieldError("service_tier")
}
if req.SafetyIdentifier != "" {
return unsupportedFieldError("safety_identifier")
}
return nil
}

func unsupportedFieldError(field string) error {
return fmt.Errorf("%s is not supported by Docker Model Runner Responses compatibility layer", field)
}

func isNullJSON(raw json.RawMessage) bool {
return strings.TrimSpace(string(raw)) == "null"
}

// GetStore returns the response store (for testing).
func (h *HTTPHandler) GetStore() *Store {
return h.store
Expand Down
Loading
Loading