270 lines
8.2 KiB
Go
270 lines
8.2 KiB
Go
package gateway
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"gitea.rutua.cn/kate522/CrossRelay/internal/chatcmpl"
|
|
"gitea.rutua.cn/kate522/CrossRelay/internal/config"
|
|
"gitea.rutua.cn/kate522/CrossRelay/internal/responses"
|
|
)
|
|
|
|
// handleResponses serves POST /v1/responses in both streaming and
|
|
// non-streaming modes.
|
|
func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) {
|
|
body, err := io.ReadAll(io.LimitReader(r.Body, 64<<20))
|
|
if err != nil {
|
|
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: "+err.Error())
|
|
return
|
|
}
|
|
var req responses.Request
|
|
if err := json.Unmarshal(body, &req); err != nil {
|
|
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: "+err.Error())
|
|
return
|
|
}
|
|
provider := s.store.ActiveCodex()
|
|
if provider == nil {
|
|
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error",
|
|
"no active provider: activate one in the CrossRelay app")
|
|
return
|
|
}
|
|
model := provider.ResolveModel(req.Model)
|
|
rec := recordFrom(r)
|
|
if provider.Protocol == config.ProtocolOpenAIResponses {
|
|
rec.update(func(e *LogEntry) {
|
|
e.Model = model
|
|
e.InputTokens = (len(body) + 3) / 4
|
|
})
|
|
s.proxyNativeResponses(w, r, provider, body, req.Model, model, req.Stream, rec)
|
|
return
|
|
}
|
|
if provider.Protocol != config.ProtocolOpenAIChat {
|
|
writeResponsesError(w, http.StatusNotImplemented, "unsupported_protocol",
|
|
fmt.Sprintf("upstream protocol %q is not supported for Codex Responses requests", provider.Protocol))
|
|
return
|
|
}
|
|
|
|
chatReq, err := responses.ConvertRequest(&req, model)
|
|
if err != nil {
|
|
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
|
|
rec.update(func(e *LogEntry) {
|
|
e.Model = model
|
|
e.InputTokens = responses.EstimateTokens(chatReq)
|
|
})
|
|
|
|
client := chatcmpl.NewClient(provider.BaseURL, provider.APIKey)
|
|
if req.Stream {
|
|
s.handleResponsesStream(w, r, client, chatReq, &req, model, rec)
|
|
return
|
|
}
|
|
|
|
resp, err := client.Create(r.Context(), chatReq)
|
|
if err != nil {
|
|
s.failUpstreamResponses(w, rec, err)
|
|
return
|
|
}
|
|
out := responses.ConvertResponse(resp, req.Model)
|
|
rec.update(func(e *LogEntry) {
|
|
if out.Usage != nil {
|
|
e.OutputTokens = out.Usage.OutputTokens
|
|
}
|
|
})
|
|
writeJSON(w, http.StatusOK, out)
|
|
}
|
|
|
|
func (s *Server) proxyNativeResponses(w http.ResponseWriter, r *http.Request, provider *config.Provider,
|
|
body []byte, requestedModel, model string, stream bool, rec *record) {
|
|
if model != requestedModel {
|
|
var payload map[string]json.RawMessage
|
|
if err := json.Unmarshal(body, &payload); err != nil || payload == nil {
|
|
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", "invalid request body: expected a JSON object")
|
|
return
|
|
}
|
|
encodedModel, err := json.Marshal(model)
|
|
if err != nil {
|
|
writeResponsesError(w, http.StatusInternalServerError, "api_error", "encode model: "+err.Error())
|
|
return
|
|
}
|
|
payload["model"] = encodedModel
|
|
body, err = json.Marshal(payload)
|
|
if err != nil {
|
|
writeResponsesError(w, http.StatusInternalServerError, "api_error", "encode request: "+err.Error())
|
|
return
|
|
}
|
|
}
|
|
|
|
upstreamReq, err := http.NewRequestWithContext(r.Context(), http.MethodPost,
|
|
strings.TrimRight(provider.BaseURL, "/")+"/responses", bytes.NewReader(body))
|
|
if err != nil {
|
|
writeResponsesError(w, http.StatusBadGateway, "api_error", "create upstream request: "+err.Error())
|
|
return
|
|
}
|
|
for name, values := range r.Header {
|
|
if isHopByHopHeader(name) || strings.EqualFold(name, "Authorization") || strings.EqualFold(name, "X-API-Key") || strings.EqualFold(name, "Content-Length") {
|
|
continue
|
|
}
|
|
for _, value := range values {
|
|
upstreamReq.Header.Add(name, value)
|
|
}
|
|
}
|
|
upstreamReq.Header.Set("Content-Type", "application/json")
|
|
if r.Header.Get("Accept") == "" {
|
|
if stream {
|
|
upstreamReq.Header.Set("Accept", "text/event-stream")
|
|
} else {
|
|
upstreamReq.Header.Set("Accept", "application/json")
|
|
}
|
|
}
|
|
if provider.APIKey != "" {
|
|
upstreamReq.Header.Set("Authorization", "Bearer "+provider.APIKey)
|
|
}
|
|
|
|
upstreamResp, err := (&http.Client{Timeout: 30 * time.Minute}).Do(upstreamReq)
|
|
if err != nil {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
writeResponsesError(w, http.StatusBadGateway, "api_error", fmt.Sprintf("upstream request failed: %v", err))
|
|
return
|
|
}
|
|
defer upstreamResp.Body.Close()
|
|
|
|
for name, values := range upstreamResp.Header {
|
|
if isHopByHopHeader(name) {
|
|
continue
|
|
}
|
|
for _, value := range values {
|
|
w.Header().Add(name, value)
|
|
}
|
|
}
|
|
|
|
if stream {
|
|
w.WriteHeader(upstreamResp.StatusCode)
|
|
if err := copyStreamingResponse(w, upstreamResp.Body); err != nil {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
}
|
|
return
|
|
}
|
|
|
|
responseBody, err := io.ReadAll(io.LimitReader(upstreamResp.Body, 64<<20))
|
|
if err != nil {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
writeResponsesError(w, http.StatusBadGateway, "api_error", "read upstream response: "+err.Error())
|
|
return
|
|
}
|
|
if upstreamResp.StatusCode >= http.StatusBadRequest {
|
|
rec.update(func(e *LogEntry) {
|
|
e.Error = fmt.Sprintf("upstream %d: %s", upstreamResp.StatusCode, strings.TrimSpace(string(responseBody)))
|
|
})
|
|
} else {
|
|
var result struct {
|
|
Usage *responses.Usage `json:"usage"`
|
|
}
|
|
if json.Unmarshal(responseBody, &result) == nil && result.Usage != nil {
|
|
rec.update(func(e *LogEntry) {
|
|
e.InputTokens = result.Usage.InputTokens
|
|
e.OutputTokens = result.Usage.OutputTokens
|
|
})
|
|
}
|
|
}
|
|
w.WriteHeader(upstreamResp.StatusCode)
|
|
_, _ = w.Write(responseBody)
|
|
}
|
|
|
|
func copyStreamingResponse(w http.ResponseWriter, body io.Reader) error {
|
|
buf := make([]byte, 32<<10)
|
|
for {
|
|
n, readErr := body.Read(buf)
|
|
if n > 0 {
|
|
if _, err := w.Write(buf[:n]); err != nil {
|
|
return err
|
|
}
|
|
if flusher, ok := w.(http.Flusher); ok {
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
if errors.Is(readErr, io.EOF) {
|
|
return nil
|
|
}
|
|
if readErr != nil {
|
|
return readErr
|
|
}
|
|
}
|
|
}
|
|
|
|
func isHopByHopHeader(name string) bool {
|
|
switch strings.ToLower(name) {
|
|
case "connection", "keep-alive", "proxy-authenticate", "proxy-authorization", "proxy-connection",
|
|
"te", "trailer", "transfer-encoding", "upgrade":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// handleResponsesStream relays a Chat Completions SSE stream as
|
|
// Responses events.
|
|
func (s *Server) handleResponsesStream(w http.ResponseWriter, r *http.Request, client *chatcmpl.Client,
|
|
chatReq *chatcmpl.Request, req *responses.Request, model string, rec *record) {
|
|
|
|
resp, err := client.CreateStream(r.Context(), chatReq)
|
|
if err != nil {
|
|
s.failUpstreamResponses(w, rec, err)
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
conv := responses.NewStreamConverter(w, req.Model, responses.EstimateTokens(chatReq))
|
|
if err := conv.WriteStart(); err != nil {
|
|
return
|
|
}
|
|
stream := chatcmpl.NewStream(resp.Body)
|
|
for {
|
|
chunk, err := stream.Next()
|
|
if errors.Is(err, io.EOF) {
|
|
break
|
|
}
|
|
if err != nil {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
conv.WriteError("api_error", err.Error())
|
|
return
|
|
}
|
|
if err := conv.OnChunk(chunk); err != nil {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
conv.WriteError("api_error", err.Error())
|
|
return
|
|
}
|
|
}
|
|
if err := conv.Finish(); err != nil {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
}
|
|
}
|
|
|
|
// failUpstreamResponses maps an upstream failure onto a Responses
|
|
// error envelope.
|
|
func (s *Server) failUpstreamResponses(w http.ResponseWriter, rec *record, err error) {
|
|
rec.update(func(e *LogEntry) { e.Error = err.Error() })
|
|
var ue *chatcmpl.UpstreamError
|
|
if errors.As(err, &ue) {
|
|
switch {
|
|
case ue.StatusCode == http.StatusUnauthorized || ue.StatusCode == http.StatusForbidden:
|
|
writeResponsesError(w, http.StatusUnauthorized, "authentication_error", ue.Message)
|
|
case ue.StatusCode == http.StatusTooManyRequests:
|
|
writeResponsesError(w, http.StatusTooManyRequests, "rate_limit_error", ue.Message)
|
|
case ue.StatusCode == http.StatusBadRequest:
|
|
writeResponsesError(w, http.StatusBadRequest, "invalid_request_error", ue.Message)
|
|
default:
|
|
writeResponsesError(w, http.StatusBadGateway, "api_error", ue.Message)
|
|
}
|
|
return
|
|
}
|
|
writeResponsesError(w, http.StatusBadGateway, "api_error", fmt.Sprintf("upstream request failed: %v", err))
|
|
}
|