Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
15 kB · 382 lines
Go
at commit e8e39526
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383package executor
import ( "bytes" "context" "fmt" "io" "net/http" "strings" "time"
"github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps" cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth" cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor" sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" "github.com/tidwall/sjson")
func (e *XAIExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return e.executeCompact(ctx, auth, req, opts) } if endpointPath := xaiImageEndpointPath(opts); endpointPath != "" { return e.executeImages(ctx, auth, req, endpointPath) } if xaiIsVideoRequest(opts) { return e.executeVideos(ctx, auth, req, opts) }
token, _ := xaiCreds(auth) baseURL := xaiChatBaseURL(auth) logXAIResolvedBaseURL(ctx, baseURL)
prepared, err := e.prepareResponsesRequest(ctx, req, opts, true) if err != nil { return resp, err }
reporter := helps.NewExecutorUsageReporter(ctx, e, prepared.baseModel, auth) defer reporter.TrackFailure(ctx, &err) reporter.SetTranslatedReasoningEffort(prepared.body, e.Identifier())
url := strings.TrimSuffix(baseURL, "/") + "/responses" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(prepared.body)) if err != nil { return resp, err } applyXAIChatHeaders(httpReq, auth, token, true, prepared.sessionID) e.recordXAIRequest(ctx, auth, url, httpReq.Header.Clone(), prepared.body)
httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpClient = reporter.TrackHTTPClient(httpClient) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("xai executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { data, errRead := io.ReadAll(httpResp.Body) if errRead != nil { helps.RecordAPIResponseError(ctx, e.cfg, errRead) return resp, errRead } helps.AppendAPIResponseChunk(ctx, e.cfg, data) helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), data)) return resp, xaiStatusErr(httpResp.StatusCode, data) }
data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data)
outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte responseFilter := newXAIInternalXSearchResponseFilter(prepared.filterInternalXSearch, prepared.clientDeclaredTools) for _, line := range bytes.Split(data, []byte("\n")) { if !bytes.HasPrefix(line, xaiDataTag) { continue } eventData := xaiNormalizeReasoningSummaryData(bytes.TrimSpace(line[len(xaiDataTag):])) eventData = restoreXAINamespaceToolCalls(eventData, prepared.namespaceTools) eventData = responseFilter.apply(eventData) if len(eventData) == 0 { continue } switch gjson.GetBytes(eventData, "type").String() { case "response.output_item.done": xaiCollectOutputItemDone(eventData, outputItemsByIndex, &outputItemsFallback) case "response.completed": if detail, ok := helps.ParseCodexUsage(eventData); ok { reporter.Publish(ctx, detail) } completedData := xaiPatchCompletedOutput(eventData, outputItemsByIndex, outputItemsFallback) completedData = xaiNormalizeReasoningSummaryData(completedData) cacheXAIReasoningReplayFromCompleted(ctx, prepared.replayScope, completedData) var param any out := sdktranslator.TranslateNonStream(ctx, prepared.to, prepared.responseFormat, req.Model, prepared.originalPayload, prepared.body, completedData, ¶m) return cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()}, nil } }
return resp, statusErr{code: http.StatusRequestTimeout, msg: "xai stream error: stream disconnected before response.completed"}}
func (e *XAIExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { prepared, data, headers, errCompact := e.executeCompactRequest(ctx, auth, req, opts) if errCompact != nil { return resp, errCompact }
var param any out := sdktranslator.TranslateNonStream(ctx, prepared.to, prepared.responseFormat, req.Model, prepared.originalPayload, prepared.body, data, ¶m) return cliproxyexecutor.Response{Payload: out, Headers: headers}, nil}
func (e *XAIExecutor) executeCompactRequest(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (*xaiPreparedRequest, []byte, http.Header, error) { token, _ := xaiCreds(auth) // Compact must not use xaiChatBaseURL: CLI chat-proxy returns 404 for // /responses/compact and a 404 cools down the whole xAI auth pool. baseURL := xaiCompactBaseURL(auth) logXAIResolvedBaseURL(ctx, baseURL)
prepared, err := e.prepareResponsesRequestTo(ctx, req, opts, false, sdktranslator.FormatOpenAIResponse) if err != nil { return nil, nil, nil, err } prepared.body, _ = sjson.DeleteBytes(prepared.body, "stream") prepared.body, _ = sjson.DeleteBytes(prepared.body, "tools") for _, field := range []string{"max_output_tokens", "temperature", "top_p", "top_k", "stop"} { prepared.body, _ = sjson.DeleteBytes(prepared.body, field) } prepared.body = xaiRemoveInputItemsByType(prepared.body, "compaction_trigger")
reporter := helps.NewExecutorUsageReporter(ctx, e, prepared.baseModel, auth) defer reporter.TrackFailure(ctx, &err) reporter.SetTranslatedReasoningEffort(prepared.body, e.Identifier())
requestURL := strings.TrimSuffix(baseURL, "/") + "/responses/compact" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, requestURL, bytes.NewReader(prepared.body)) if err != nil { return nil, nil, nil, err } // Official API / custom compact endpoints use standard API headers, not CLI // chat-proxy identity headers (which applyXAIChatHeaders may still attach for OAuth chat). applyXAIHeaders(httpReq, auth, token, false, prepared.sessionID) e.recordXAIRequest(ctx, auth, requestURL, httpReq.Header.Clone(), prepared.body)
httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpClient = reporter.TrackHTTPClient(httpClient) httpResp, err := httpClient.Do(httpReq) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return nil, nil, nil, err } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("xai executor: close response body error: %v", errClose) } }() helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())
data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return nil, nil, nil, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data)
if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), data)) err = xaiStatusErr(httpResp.StatusCode, data) return nil, nil, nil, err }
reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) reporter.EnsurePublished(ctx) clearXAIReasoningReplayAfterCompaction(ctx, prepared.replayScope) return prepared, data, httpResp.Header.Clone(), nil}
func (e *XAIExecutor) executeCompactionTriggerStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (*cliproxyexecutor.StreamResult, error) { prepared, data, headers, err := e.executeCompactRequest(ctx, auth, req, opts) if err != nil { return nil, err }
headers = headers.Clone() if headers == nil { headers = make(http.Header) } headers.Set("Content-Type", "text/event-stream")
chunks := xaiBuildCompactionTriggerStreamChunks(prepared, data) out := make(chan cliproxyexecutor.StreamChunk, len(chunks)) for _, chunk := range chunks { out <- cliproxyexecutor.StreamChunk{Payload: chunk} } close(out) return &cliproxyexecutor.StreamResult{Headers: headers, Chunks: out}, nil}
func xaiInputHasItemType(body []byte, itemType string) bool { input := gjson.GetBytes(body, "input") if !input.IsArray() { return false } for _, item := range input.Array() { if item.Get("type").String() == itemType { return true } } return false}
func xaiRemoveInputItemsByType(body []byte, itemType string) []byte { input := gjson.GetBytes(body, "input") if !input.IsArray() { return body }
var buf bytes.Buffer buf.WriteByte('[') kept := 0 for _, item := range input.Array() { if item.Get("type").String() == itemType { continue } if kept > 0 { buf.WriteByte(',') } buf.WriteString(item.Raw) kept++ } buf.WriteByte(']')
updated, err := sjson.SetRawBytes(body, "input", buf.Bytes()) if err != nil { return body } return updated}
func xaiBuildCompactionTriggerStreamChunks(prepared *xaiPreparedRequest, compactData []byte) [][]byte { responseID := xaiCompactionResponseID(compactData) now := time.Now().Unix() createdAt := gjson.GetBytes(compactData, "created_at").Int() if createdAt == 0 { createdAt = now } completedAt := gjson.GetBytes(compactData, "completed_at").Int() if completedAt == 0 { completedAt = now }
item := xaiCompactionOutputItem(compactData, responseID) output := make([]byte, 0, len(item)+2) output = append(output, '[') output = append(output, item...) output = append(output, ']')
createdResponse := xaiBuildCompactionBaseResponse(prepared, compactData, responseID, createdAt, "in_progress") inProgressResponse := xaiBuildCompactionBaseResponse(prepared, compactData, responseID, createdAt, "in_progress") completedResponse := xaiBuildCompactionBaseResponse(prepared, compactData, responseID, createdAt, "completed") completedResponse, _ = sjson.SetBytes(completedResponse, "completed_at", completedAt) completedResponse, _ = sjson.SetRawBytes(completedResponse, "output", output) if usage := gjson.GetBytes(compactData, "usage"); usage.Exists() { completedResponse, _ = sjson.SetRawBytes(completedResponse, "usage", []byte(usage.Raw)) }
createdPayload := []byte(`{"type":"response.created","sequence_number":0}`) createdPayload, _ = sjson.SetRawBytes(createdPayload, "response", createdResponse) inProgressPayload := []byte(`{"type":"response.in_progress","sequence_number":1}`) inProgressPayload, _ = sjson.SetRawBytes(inProgressPayload, "response", inProgressResponse) addedPayload := []byte(`{"type":"response.output_item.added","sequence_number":2,"output_index":0}`) addedPayload, _ = sjson.SetRawBytes(addedPayload, "item", item) keepalivePayload := []byte(`{"type":"keepalive","sequence_number":3}`) donePayload := []byte(`{"type":"response.output_item.done","sequence_number":4,"output_index":0}`) donePayload, _ = sjson.SetRawBytes(donePayload, "item", item) completedPayload := []byte(`{"type":"response.completed","sequence_number":5}`) completedPayload, _ = sjson.SetRawBytes(completedPayload, "response", completedResponse)
return [][]byte{ xaiBuildSSEFrame("response.created", createdPayload), xaiBuildSSEFrame("response.in_progress", inProgressPayload), xaiBuildSSEFrame("response.output_item.added", addedPayload), xaiBuildSSEFrame("keepalive", keepalivePayload), xaiBuildSSEFrame("response.output_item.done", donePayload), xaiBuildSSEFrame("response.completed", completedPayload), }}
func xaiBuildCompactionBaseResponse(prepared *xaiPreparedRequest, compactData []byte, responseID string, createdAt int64, status string) []byte { response := []byte(`{"id":"","object":"response","created_at":0,"status":"","background":false,"error":null,"incomplete_details":null,"output":[]}`) response, _ = sjson.SetBytes(response, "id", responseID) response, _ = sjson.SetBytes(response, "created_at", createdAt) response, _ = sjson.SetBytes(response, "status", status) if model := gjson.GetBytes(compactData, "model").String(); model != "" { response, _ = sjson.SetBytes(response, "model", model) } else if prepared != nil && prepared.baseModel != "" { response, _ = sjson.SetBytes(response, "model", prepared.baseModel) }
if prepared == nil { return response } for _, field := range []string{ "instructions", "max_output_tokens", "max_tool_calls", "parallel_tool_calls", "previous_response_id", "prompt_cache_key", "reasoning", "text", "tool_choice", "tools", "top_logprobs", "top_p", "truncation", "user", "metadata", } { if value := gjson.GetBytes(prepared.body, field); value.Exists() { response, _ = sjson.SetRawBytes(response, field, []byte(value.Raw)) } } return response}
func xaiCompactionOutputItem(compactData []byte, responseID string) []byte { itemResult := gjson.GetBytes(compactData, "output.0") item := []byte(`{"type":"compaction"}`) if itemResult.Exists() && itemResult.Type == gjson.JSON { item = []byte(itemResult.Raw) } if !gjson.GetBytes(item, "type").Exists() { item, _ = sjson.SetBytes(item, "type", "compaction") } if !gjson.GetBytes(item, "id").Exists() { item, _ = sjson.SetBytes(item, "id", xaiCompactionItemID(responseID)) } return item}
func xaiCompactionResponseID(compactData []byte) string { if responseID := strings.TrimSpace(gjson.GetBytes(compactData, "id").String()); responseID != "" { if strings.HasPrefix(responseID, "resp_") { return responseID } return "resp_" + strings.TrimPrefix(responseID, "cmp_") } return fmt.Sprintf("resp_xai_compaction_%d", time.Now().UnixNano())}
func xaiCompactionItemID(responseID string) string { if suffix := strings.TrimPrefix(responseID, "resp_"); suffix != "" && suffix != responseID { return "cmp_" + suffix } return "cmp_" + responseID}
func xaiBuildSSEFrame(eventName string, data []byte) []byte { out := make([]byte, 0, len(eventName)+len(data)+16) out = append(out, "event: "...) out = append(out, eventName...) out = append(out, '\n') out = append(out, "data: "...) out = append(out, data...) out = append(out, '\n', '\n') return out}