m1saka/zhanlu_proxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add OpenAI Responses API (/v1/responses) endpoint

Browse Source 
Translate Responses API requests (input→messages, instructions→system,
max_output_tokens→max_tokens, text.format→response_format, flat tools→nested
{function:{…}}) to upstream chat/completions, then convert responses back to
Responses format (streaming SSE event lifecycle + non-streaming JSON).

Verified against OpenAI migration guide and Python SDK Response model:
- Echo back required fields parallel_tool_calls/tool_choice/tools
- Include content:[] in reasoning items, logprobs:[] in output_text parts
- Support function_call/function_call_output multi-turn input items
- Map usage fields prompt_tokens→input_tokens, completion_tokens→output_tokens
This commit is contained in:
m1saka
2026-08-20 10:02:11 +08:00
parent c8938cb514
commit 2517b4f730
6 changed files with 1576 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/responses.go
+591
View File
@@ -0,0 +1,591 @@
package server
import (
"bufio"
"encoding/json"
"fmt"
"io"
"net/http"
"time"
"git.misaka.ren/M1saka/zhanlu_proxy/internal/openai"
)
// responses handles POST /v1/responses — the OpenAI Responses API. It
// translates the request to a chat-completions request for the upstream
// gateway, then translates the response back into the Responses format.
func (s *Server) responses(w http.ResponseWriter, r *http.Request) {
creds, err := s.currentCredentials()
if err != nil {
writeOpenAIError(w, http.StatusUnauthorized, "zhanlu credentials are not configured; open /login first", "auth_error", "missing_credentials")
return
}
body, err := io.ReadAll(r.Body)
if err != nil {
writeOpenAIError(w, http.StatusBadRequest, err.Error(), "invalid_request_error", "bad_body")
return
}
chatReq, err := openai.ParseResponsesRequest(body)
if err != nil {
writeOpenAIError(w, http.StatusBadRequest, err.Error(), "invalid_request_error", "bad_json")
return
}
meta := openai.ParseResponsesMeta(body)
if chatReq.Model == "" {
chatReq.Model = "zhanlu/auto"
}
start := time.Now()
clientWantsStream := chatReq.Stream
chatReq.Stream = true // upstream always streams
upstreamBody, err := chatReq.MarshalForUpstream()
if err != nil {
writeOpenAIError(w, http.StatusBadRequest, err.Error(), "invalid_request_error", "bad_body")
return
}
if !creds.HasAPIKey() {
creds, err = s.provisionCredentials(r.Context(), creds)
if err != nil {
writeOpenAIError(w, http.StatusBadGateway, "zhanlu api key provisioning failed: "+err.Error(), "auth_error", "zhanlu_provision_failed")
return
}
s.cfg.Credentials = creds
_ = s.st.SaveCredentials(creds)
}
modelBaseURL := firstNonEmpty(creds.ModelBaseURL, s.cfg.MobileModelBaseURL)
client, err := s.zhanluClientWithBase(modelBaseURL)
if err != nil {
writeOpenAIError(w, http.StatusInternalServerError, err.Error(), "sign_error", "signer_init_failed")
return
}
resp, err := client.ChatCompletions(r.Context(), creds.APIKey, upstreamBody)
if err != nil {
msg := "zhanlu upstream request failed"
if s.cfg.Debug {
msg = redactSensitive(err.Error())
}
writeOpenAIError(w, http.StatusBadGateway, msg, "upstream_error", "zhanlu_request_failed")
s.record(chatReq.Model, clientWantsStream, nil, "upstream_error", start)
return
}
defer resp.Body.Close()
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
msg := fmt.Sprintf("zhanlu upstream returned %d", resp.StatusCode)
if s.cfg.Debug && len(b) > 0 {
msg += ": " + string(b)
}
writeOpenAIError(w, http.StatusBadGateway, msg, "upstream_error", "zhanlu_bad_status")
s.record(chatReq.Model, clientWantsStream, nil, "upstream_error", start)
return
}
if clientWantsStream {
usage, status := s.proxyResponsesStream(w, resp, chatReq.Model, meta)
s.record(chatReq.Model, true, usage, status, start)
return
}
usage, status := s.aggregateResponsesStream(w, resp, chatReq.Model, meta)
s.record(chatReq.Model, false, usage, status, start)
}
// ---------------------------------------------------------------- non-streaming
// aggregateResponsesStream reads the upstream SSE, aggregates it (same logic
// as aggregateStream), then emits a Responses API JSON object.
func (s *Server) aggregateResponsesStream(w http.ResponseWriter, resp *http.Response, model string, meta openai.ResponsesMeta) (any, string) {
var content, reasoning, id string
var usage any
finishReason := "stop"
type toolCall struct {
ID string `json:"id"`
Type string `json:"type"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
}
toolCalls := map[int]*toolCall{}
err := forEachSSEChunk(resp.Body, func(chunk []byte) error {
var event struct {
ID string `json:"id"`
Choices []struct {
Delta struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Reasoning string `json:"reasoning"`
ToolCalls []struct {
Index int `json:"index"`
ID string `json:"id"`
Type string `json:"type"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
} `json:"tool_calls"`
} `json:"delta"`
FinishReason *string `json:"finish_reason"`
} `json:"choices"`
Usage any `json:"usage"`
}
if err := json.Unmarshal(chunk, &event); err != nil {
return err
}
if event.ID != "" {
id = event.ID
}
if event.Usage != nil {
usage = event.Usage
}
if len(event.Choices) > 0 {
content += event.Choices[0].Delta.Content
reasoning += event.Choices[0].Delta.ReasoningContent + event.Choices[0].Delta.Reasoning
for _, part := range event.Choices[0].Delta.ToolCalls {
call := toolCalls[part.Index]
if call == nil {
call = &toolCall{Type: "function"}
toolCalls[part.Index] = call
}
if part.ID != "" {
call.ID = part.ID
}
if part.Type != "" {
call.Type = part.Type
}
call.Function.Name += part.Function.Name
call.Function.Arguments += part.Function.Arguments
}
if event.Choices[0].FinishReason != nil {
finishReason = *event.Choices[0].FinishReason
}
}
return nil
})
if err != nil {
writeOpenAIError(w, http.StatusBadGateway, err.Error(), "upstream_error", "zhanlu_stream_error")
return nil, "upstream_error"
}
_ = id
_ = finishReason
respID := "resp_" + randomRequestID()
output := []any{}
if reasoning != "" {
output = append(output, map[string]any{
"type": "reasoning", "id": "rs_" + randomRequestID(), "status": "completed",
"content": []any{},
"summary": []map[string]any{{"type": "summary_text", "text": reasoning}},
})
}
// Message output item — included when there is text content.
if content != "" {
output = append(output, map[string]any{
"type": "message", "id": "msg_" + randomRequestID(), "status": "completed",
"role": "assistant",
"content": []map[string]any{{
"type": "output_text", "text": content,
"annotations": []any{}, "logprobs": []any{},
}},
})
}
if len(toolCalls) > 0 {
ordered := make([]*toolCall, 0, len(toolCalls))
for i := 0; i < len(toolCalls); i++ {
if call := toolCalls[i]; call != nil {
ordered = append(ordered, call)
}
}
for _, tc := range ordered {
output = append(output, map[string]any{
"type": "function_call", "id": "fc_" + randomRequestID(),
"call_id": tc.ID, "name": tc.Function.Name,
"arguments": tc.Function.Arguments, "status": "completed",
})
}
}
// If there was no content and no tool calls, still emit an empty message
// so the response always has at least one output item.
if len(output) == 0 {
output = append(output, map[string]any{
"type": "message", "id": "msg_" + randomRequestID(), "status": "completed",
"role": "assistant", "content": []map[string]any{},
})
}
result := map[string]any{
"id": respID, "object": "response", "created_at": time.Now().Unix(),
"model": model, "status": "completed", "output": output,
"parallel_tool_calls": meta.ParallelToolCalls,
"tool_choice": meta.ToolChoice,
"tools": meta.Tools,
}
if usage != nil {
result["usage"] = openai.UsageToResponses(usage)
}
writeJSON(w, http.StatusOK, result)
return usage, "success"
}
// ------------------------------------------------------------------- streaming
// responsesStreamState tracks the lifecycle of output items while converting
// Chat Completions SSE into Responses API SSE events.
type responsesStreamState struct {
w http.ResponseWriter
flusher http.Flusher
respID string
msgID string
rsID string
meta openai.ResponsesMeta
outputIdx int
reasoningOn bool
reasonPartOn bool
messageOn bool
partOn bool
fullContent string
fullReasoning string
usage any
status string
// tool call tracking
tools map[int]*streamToolCall
toolOrder []int
}
type streamToolCall struct {
itemID string
callID string
name string
args string
started bool
doneIdx int // assigned output index when started
}
func newResponsesStreamState(w http.ResponseWriter, model string, meta openai.ResponsesMeta) *responsesStreamState {
flusher, _ := w.(http.Flusher)
w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
w.Header().Set("X-Accel-Buffering", "no")
w.WriteHeader(http.StatusOK)
st := &responsesStreamState{
w: w,
flusher: flusher,
respID: "resp_" + randomRequestID(),
msgID: "msg_" + randomRequestID(),
rsID: "rs_" + randomRequestID(),
meta: meta,
tools: map[int]*streamToolCall{},
status: "success",
}
baseResp := func(st2 string) map[string]any {
return map[string]any{
"id": st.respID, "object": "response", "created_at": time.Now().Unix(),
"model": model, "status": st2, "output": []any{},
"parallel_tool_calls": meta.ParallelToolCalls,
"tool_choice": meta.ToolChoice,
"tools": meta.Tools,
}
}
st.emit("response.created", map[string]any{"type": "response.created", "response": baseResp("in_progress")})
st.emit("response.in_progress", map[string]any{"type": "response.in_progress", "response": baseResp("in_progress")})
return st
}
func (st *responsesStreamState) emit(event string, data any) {
b, _ := json.Marshal(data)
fmt.Fprintf(st.w, "event: %s\ndata: %s\n\n", event, b)
if st.flusher != nil {
st.flusher.Flush()
}
}
// closeReasoning closes the reasoning output item if it is open.
func (st *responsesStreamState) closeReasoning() {
if !st.reasoningOn {
return
}
if st.reasonPartOn {
st.emit("response.reasoning_summary_text.done", map[string]any{
"type": "response.reasoning_summary_text.done", "item_id": st.rsID,
"output_index": st.outputIdx, "summary_index": 0, "text": st.fullReasoning,
})
st.emit("response.reasoning_summary_part.done", map[string]any{
"type": "response.reasoning_summary_part.done", "item_id": st.rsID,
"output_index": st.outputIdx, "summary_index": 0,
"part": map[string]any{"type": "summary_text", "text": st.fullReasoning},
})
st.reasonPartOn = false
}
st.emit("response.output_item.done", map[string]any{
"type": "response.output_item.done", "output_index": st.outputIdx,
"item": map[string]any{
"type": "reasoning", "id": st.rsID, "status": "completed",
"content": []any{},
"summary": []map[string]any{{"type": "summary_text", "text": st.fullReasoning}},
},
})
st.reasoningOn = false
st.outputIdx++
}
// closeMessage closes the message output item if it is open.
func (st *responsesStreamState) closeMessage() {
if !st.messageOn {
return
}
if st.partOn {
st.emit("response.output_text.done", map[string]any{
"type": "response.output_text.done", "item_id": st.msgID,
"output_index": st.outputIdx, "content_index": 0, "text": st.fullContent,
})
st.emit("response.content_part.done", map[string]any{
"type": "response.content_part.done", "item_id": st.msgID,
"output_index": st.outputIdx, "content_index": 0,
"part": map[string]any{"type": "output_text", "text": st.fullContent, "annotations": []any{}, "logprobs": []any{}},
})
st.partOn = false
}
st.emit("response.output_item.done", map[string]any{
"type": "response.output_item.done", "output_index": st.outputIdx,
"item": map[string]any{
"type": "message", "id": st.msgID, "status": "completed", "role": "assistant",
"content": []map[string]any{{"type": "output_text", "text": st.fullContent, "annotations": []any{}, "logprobs": []any{}}},
},
})
st.messageOn = false
st.outputIdx++
}
// handleReasoning processes a reasoning content delta.
func (st *responsesStreamState) handleReasoning(delta string) {
if !st.reasoningOn {
st.reasoningOn = true
st.emit("response.output_item.added", map[string]any{
"type": "response.output_item.added", "output_index": st.outputIdx,
"item": map[string]any{
"type": "reasoning", "id": st.rsID, "status": "in_progress",
"content": []any{}, "summary": []any{},
},
})
st.emit("response.reasoning_summary_part.added", map[string]any{
"type": "response.reasoning_summary_part.added", "item_id": st.rsID,
"output_index": st.outputIdx, "summary_index": 0,
"part": map[string]any{"type": "summary_text", "text": ""},
})
st.reasonPartOn = true
}
st.fullReasoning += delta
st.emit("response.reasoning_summary_text.delta", map[string]any{
"type": "response.reasoning_summary_text.delta", "item_id": st.rsID,
"output_index": st.outputIdx, "summary_index": 0, "delta": delta,
})
}
// handleContent processes a text content delta.
func (st *responsesStreamState) handleContent(delta string) {
// Close reasoning if open — text content comes after reasoning.
st.closeReasoning()
if !st.messageOn {
st.messageOn = true
st.emit("response.output_item.added", map[string]any{
"type": "response.output_item.added", "output_index": st.outputIdx,
"item": map[string]any{
"type": "message", "id": st.msgID, "status": "in_progress", "role": "assistant", "content": []any{},
},
})
st.emit("response.content_part.added", map[string]any{
"type": "response.content_part.added", "item_id": st.msgID,
"output_index": st.outputIdx, "content_index": 0,
"part": map[string]any{"type": "output_text", "text": "", "annotations": []any{}, "logprobs": []any{}},
})
st.partOn = true
}
st.fullContent += delta
st.emit("response.output_text.delta", map[string]any{
"type": "response.output_text.delta", "item_id": st.msgID,
"output_index": st.outputIdx, "content_index": 0, "delta": delta,
})
}
// handleToolCall processes a tool call delta from the chat completion stream.
func (st *responsesStreamState) handleToolCall(index int, id, name, args string) {
// Close reasoning/message if open — tool calls are separate output items.
st.closeReasoning()
st.closeMessage()
call := st.tools[index]
if call == nil {
call = &streamToolCall{itemID: "fc_" + randomRequestID()}
st.tools[index] = call
st.toolOrder = append(st.toolOrder, index)
}
if id != "" {
call.callID = id
}
if name != "" {
call.name += name
}
if !call.started {
call.started = true
call.doneIdx = st.outputIdx
st.emit("response.output_item.added", map[string]any{
"type": "response.output_item.added", "output_index": st.outputIdx,
"item": map[string]any{
"type": "function_call", "id": call.itemID, "call_id": call.callID,
"name": call.name, "arguments": "", "status": "in_progress",
},
})
st.outputIdx++
}
if args != "" {
call.args += args
st.emit("response.function_call_arguments.delta", map[string]any{
"type": "response.function_call_arguments.delta", "item_id": call.itemID,
"output_index": call.doneIdx, "delta": args,
})
}
}
// finish closes all open items and sends the response.completed event.
func (st *responsesStreamState) finish(model string) {
st.closeReasoning()
st.closeMessage()
for _, idx := range st.toolOrder {
call := st.tools[idx]
st.emit("response.function_call_arguments.done", map[string]any{
"type": "response.function_call_arguments.done", "item_id": call.itemID,
"output_index": call.doneIdx, "arguments": call.args,
})
st.emit("response.output_item.done", map[string]any{
"type": "response.output_item.done", "output_index": call.doneIdx,
"item": map[string]any{
"type": "function_call", "id": call.itemID, "call_id": call.callID,
"name": call.name, "arguments": call.args, "status": "completed",
},
})
}
// Build the final output array for the completed event.
finalOutput := []any{}
if st.fullReasoning != "" {
finalOutput = append(finalOutput, map[string]any{
"type": "reasoning", "id": st.rsID, "status": "completed",
"content": []any{},
"summary": []map[string]any{{"type": "summary_text", "text": st.fullReasoning}},
})
}
if st.fullContent != "" {
finalOutput = append(finalOutput, map[string]any{
"type": "message", "id": st.msgID, "status": "completed", "role": "assistant",
"content": []map[string]any{{"type": "output_text", "text": st.fullContent, "annotations": []any{}, "logprobs": []any{}}},
})
}
for _, idx := range st.toolOrder {
call := st.tools[idx]
finalOutput = append(finalOutput, map[string]any{
"type": "function_call", "id": call.itemID, "call_id": call.callID,
"name": call.name, "arguments": call.args, "status": "completed",
})
}
// Ensure at least one output item exists.
if len(finalOutput) == 0 {
finalOutput = append(finalOutput, map[string]any{
"type": "message", "id": st.msgID, "status": "completed", "role": "assistant",
"content": []map[string]any{},
})
}
completedResp := map[string]any{
"id": st.respID, "object": "response", "created_at": time.Now().Unix(),
"model": model, "status": "completed", "output": finalOutput,
"parallel_tool_calls": st.meta.ParallelToolCalls,
"tool_choice": st.meta.ToolChoice,
"tools": st.meta.Tools,
}
if st.usage != nil {
completedResp["usage"] = openai.UsageToResponses(st.usage)
}
st.emit("response.completed", map[string]any{"type": "response.completed", "response": completedResp})
}
// proxyResponsesStream reads the upstream Chat Completions SSE and emits
// Responses API SSE events via a responsesStreamState state machine.
func (s *Server) proxyResponsesStream(w http.ResponseWriter, resp *http.Response, model string, meta openai.ResponsesMeta) (any, string) {
st := newResponsesStreamState(w, model, meta)
reader := bufio.NewReader(resp.Body)
for {
line, err := reader.ReadString('\n')
if line != "" {
payload := sseDataPayload(line)
if payload != "" && payload != "[DONE]" {
var evt struct {
State string `json:"state"`
ErrorMessage string `json:"errorMessage"`
Usage any `json:"usage"`
Choices []struct {
Delta struct {
Content string `json:"content"`
ReasoningContent string `json:"reasoning_content"`
Reasoning string `json:"reasoning"`
ToolCalls []struct {
Index int `json:"index"`
ID string `json:"id"`
Function struct {
Name string `json:"name"`
Arguments string `json:"arguments"`
} `json:"function"`
} `json:"tool_calls"`
} `json:"delta"`
} `json:"choices"`
}
if jErr := json.Unmarshal([]byte(payload), &evt); jErr == nil {
if evt.State == "ERROR" {
st.status = "upstream_error"
}
if evt.Usage != nil {
st.usage = evt.Usage
}
if len(evt.Choices) > 0 {
delta := evt.Choices[0].Delta
reasoningDelta := delta.ReasoningContent + delta.Reasoning
if reasoningDelta != "" {
st.handleReasoning(reasoningDelta)
}
if delta.Content != "" {
st.handleContent(delta.Content)
}
for _, tc := range delta.ToolCalls {
st.handleToolCall(tc.Index, tc.ID, tc.Function.Name, tc.Function.Arguments)
}
}
}
}
}
if err != nil {
if err != io.EOF {
b, _ := json.Marshal(map[string]any{
"message": err.Error(), "type": "upstream_error", "code": "zhanlu_stream_error",
})
fmt.Fprintf(w, "data: %s\n\n", b)
if st.flusher != nil {
st.flusher.Flush()
}
st.status = "upstream_error"
}
break
}
}
st.finish(model)
return st.usage, st.status
}