mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-06-10 16:22:59 +08:00
959067edfb
- Replaced `NewUsageReporter` with `NewExecutorUsageReporter` to include executor type in usage records. - Updated all executors to use the new reporter implementation. - Extended `UsageReporter` to track and publish executor type. - Added tests to validate proper executor type recording and handling. - Enhanced RedisQueue plugin and payload schema with executor type support.
795 lines
27 KiB
Go
795 lines
27 KiB
Go
package executor
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"mime"
|
|
"mime/multipart"
|
|
"net/http"
|
|
"net/textproto"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/router-for-me/CLIProxyAPI/v7/internal/config"
|
|
"github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps"
|
|
"github.com/router-for-me/CLIProxyAPI/v7/internal/thinking"
|
|
"github.com/router-for-me/CLIProxyAPI/v7/internal/util"
|
|
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/sjson"
|
|
)
|
|
|
|
const (
|
|
openAICompatImageHandlerType = "openai-image"
|
|
openAICompatImagesGenerationsPath = "/images/generations"
|
|
openAICompatImagesEditsPath = "/images/edits"
|
|
openAICompatDefaultImageEndpoint = openAICompatImagesGenerationsPath
|
|
openAICompatMultipartMemory int64 = 32 << 20
|
|
)
|
|
|
|
// OpenAICompatExecutor implements a stateless executor for OpenAI-compatible providers.
|
|
// It performs request/response translation and executes against the provider base URL
|
|
// using per-auth credentials (API key) and per-auth HTTP transport (proxy) from context.
|
|
type OpenAICompatExecutor struct {
|
|
provider string
|
|
cfg *config.Config
|
|
}
|
|
|
|
// NewOpenAICompatExecutor creates an executor bound to a provider key (e.g., "openrouter").
|
|
func NewOpenAICompatExecutor(provider string, cfg *config.Config) *OpenAICompatExecutor {
|
|
return &OpenAICompatExecutor{provider: provider, cfg: cfg}
|
|
}
|
|
|
|
// Identifier implements cliproxyauth.ProviderExecutor.
|
|
func (e *OpenAICompatExecutor) Identifier() string { return e.provider }
|
|
|
|
// PrepareRequest injects OpenAI-compatible credentials into the outgoing HTTP request.
|
|
func (e *OpenAICompatExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error {
|
|
if req == nil {
|
|
return nil
|
|
}
|
|
_, apiKey := e.resolveCredentials(auth)
|
|
if strings.TrimSpace(apiKey) != "" {
|
|
req.Header.Set("Authorization", "Bearer "+apiKey)
|
|
}
|
|
var attrs map[string]string
|
|
if auth != nil {
|
|
attrs = auth.Attributes
|
|
}
|
|
util.ApplyCustomHeadersFromAttrs(req, attrs)
|
|
return nil
|
|
}
|
|
|
|
// HttpRequest injects OpenAI-compatible credentials into the request and executes it.
|
|
func (e *OpenAICompatExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) {
|
|
if req == nil {
|
|
return nil, fmt.Errorf("openai compat executor: request is nil")
|
|
}
|
|
if ctx == nil {
|
|
ctx = req.Context()
|
|
}
|
|
httpReq := req.WithContext(ctx)
|
|
if err := e.PrepareRequest(httpReq, auth); err != nil {
|
|
return nil, err
|
|
}
|
|
httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0)
|
|
return httpClient.Do(httpReq)
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) {
|
|
if endpointPath := openAICompatImageEndpointPath(opts); endpointPath != "" {
|
|
return e.executeImages(ctx, auth, req, opts, endpointPath)
|
|
}
|
|
|
|
baseModel := thinking.ParseSuffix(req.Model).ModelName
|
|
|
|
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
|
|
defer reporter.TrackFailure(ctx, &err)
|
|
|
|
baseURL, apiKey := e.resolveCredentials(auth)
|
|
if baseURL == "" {
|
|
err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}
|
|
return
|
|
}
|
|
|
|
from := opts.SourceFormat
|
|
to := sdktranslator.FromString("openai")
|
|
endpoint := "/chat/completions"
|
|
if opts.Alt == "responses/compact" {
|
|
to = sdktranslator.FromString("openai-response")
|
|
endpoint = "/responses/compact"
|
|
}
|
|
originalPayloadSource := req.Payload
|
|
if len(opts.OriginalRequest) > 0 {
|
|
originalPayloadSource = opts.OriginalRequest
|
|
}
|
|
originalPayload := originalPayloadSource
|
|
originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, opts.Stream)
|
|
translated := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, opts.Stream)
|
|
|
|
translated, err = thinking.ApplyThinking(translated, req.Model, from.String(), to.String(), e.Identifier())
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
|
|
requestedModel := helps.PayloadRequestedModel(opts, req.Model)
|
|
requestPath := helps.PayloadRequestPath(opts)
|
|
translated = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", translated, originalTranslated, requestedModel, requestPath, opts.Headers)
|
|
if opts.Alt == "responses/compact" {
|
|
if updated, errDelete := sjson.DeleteBytes(translated, "stream"); errDelete == nil {
|
|
translated = updated
|
|
}
|
|
translated = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "openai compat executor", translated)
|
|
}
|
|
reporter.SetTranslatedReasoningEffort(translated, to.String())
|
|
|
|
url := strings.TrimSuffix(baseURL, "/") + endpoint
|
|
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(translated))
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
httpReq.Header.Set("Content-Type", "application/json")
|
|
if apiKey != "" {
|
|
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
|
|
}
|
|
httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")
|
|
var attrs map[string]string
|
|
if auth != nil {
|
|
attrs = auth.Attributes
|
|
}
|
|
util.ApplyCustomHeadersFromAttrs(httpReq, attrs)
|
|
var authID, authLabel, authType, authValue string
|
|
if auth != nil {
|
|
authID = auth.ID
|
|
authLabel = auth.Label
|
|
authType, authValue = auth.AccountInfo()
|
|
}
|
|
helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{
|
|
URL: url,
|
|
Method: http.MethodPost,
|
|
Headers: httpReq.Header.Clone(),
|
|
Body: translated,
|
|
Provider: e.Identifier(),
|
|
AuthID: authID,
|
|
AuthLabel: authLabel,
|
|
AuthType: authType,
|
|
AuthValue: authValue,
|
|
})
|
|
|
|
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("openai compat executor: close response body error: %v", errClose)
|
|
}
|
|
}()
|
|
helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())
|
|
if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {
|
|
b, _ := io.ReadAll(httpResp.Body)
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, b)
|
|
helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b))
|
|
err = statusErr{code: httpResp.StatusCode, msg: string(b)}
|
|
return resp, err
|
|
}
|
|
body, err := io.ReadAll(httpResp.Body)
|
|
if err != nil {
|
|
helps.RecordAPIResponseError(ctx, e.cfg, err)
|
|
return resp, err
|
|
}
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, body)
|
|
reporter.Publish(ctx, helps.ParseOpenAIUsage(body))
|
|
// Ensure we at least record the request even if upstream doesn't return usage
|
|
reporter.EnsurePublished(ctx)
|
|
// Translate response back to source format when needed
|
|
var param any
|
|
out := sdktranslator.TranslateNonStream(ctx, to, from, req.Model, opts.OriginalRequest, translated, body, ¶m)
|
|
resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()}
|
|
return resp, nil
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) executeImages(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, endpointPath string) (resp cliproxyexecutor.Response, err error) {
|
|
baseModel := thinking.ParseSuffix(req.Model).ModelName
|
|
|
|
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
|
|
defer reporter.TrackFailure(ctx, &err)
|
|
|
|
baseURL, apiKey := e.resolveCredentials(auth)
|
|
if baseURL == "" {
|
|
err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}
|
|
return resp, err
|
|
}
|
|
|
|
payload, contentType, errPrepare := prepareOpenAICompatImagesPayload(req.Payload, baseModel, opts.Headers.Get("Content-Type"), false)
|
|
if errPrepare != nil {
|
|
err = errPrepare
|
|
return resp, err
|
|
}
|
|
if contentType == "" {
|
|
contentType = "application/json"
|
|
}
|
|
reporter.SetTranslatedReasoningEffort(payload, "openai")
|
|
|
|
url := strings.TrimSuffix(baseURL, "/") + endpointPath
|
|
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
|
|
if err != nil {
|
|
return resp, err
|
|
}
|
|
httpReq.Header.Set("Content-Type", contentType)
|
|
if apiKey != "" {
|
|
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
|
|
}
|
|
httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")
|
|
var attrs map[string]string
|
|
if auth != nil {
|
|
attrs = auth.Attributes
|
|
}
|
|
util.ApplyCustomHeadersFromAttrs(httpReq, attrs)
|
|
var authID, authLabel, authType, authValue string
|
|
if auth != nil {
|
|
authID = auth.ID
|
|
authLabel = auth.Label
|
|
authType, authValue = auth.AccountInfo()
|
|
}
|
|
helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{
|
|
URL: url,
|
|
Method: http.MethodPost,
|
|
Headers: httpReq.Header.Clone(),
|
|
Body: payload,
|
|
Provider: e.Identifier(),
|
|
AuthID: authID,
|
|
AuthLabel: authLabel,
|
|
AuthType: authType,
|
|
AuthValue: authValue,
|
|
})
|
|
|
|
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("openai compat executor: close response body error: %v", errClose)
|
|
}
|
|
}()
|
|
helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())
|
|
|
|
body, errRead := io.ReadAll(httpResp.Body)
|
|
if errRead != nil {
|
|
helps.RecordAPIResponseError(ctx, e.cfg, errRead)
|
|
err = errRead
|
|
return resp, err
|
|
}
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, body)
|
|
|
|
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"), body))
|
|
err = statusErr{code: httpResp.StatusCode, msg: string(body)}
|
|
return resp, err
|
|
}
|
|
|
|
reporter.Publish(ctx, helps.ParseOpenAIUsage(body))
|
|
reporter.EnsurePublished(ctx)
|
|
resp = cliproxyexecutor.Response{Payload: body, Headers: httpResp.Header.Clone()}
|
|
return resp, nil
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) {
|
|
if endpointPath := openAICompatImageEndpointPath(opts); endpointPath != "" {
|
|
return e.executeImagesStream(ctx, auth, req, opts, endpointPath)
|
|
}
|
|
|
|
baseModel := thinking.ParseSuffix(req.Model).ModelName
|
|
|
|
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
|
|
defer reporter.TrackFailure(ctx, &err)
|
|
|
|
baseURL, apiKey := e.resolveCredentials(auth)
|
|
if baseURL == "" {
|
|
err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}
|
|
return nil, err
|
|
}
|
|
|
|
from := opts.SourceFormat
|
|
to := sdktranslator.FromString("openai")
|
|
originalPayloadSource := req.Payload
|
|
if len(opts.OriginalRequest) > 0 {
|
|
originalPayloadSource = opts.OriginalRequest
|
|
}
|
|
originalPayload := originalPayloadSource
|
|
originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, true)
|
|
translated := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, true)
|
|
|
|
translated, err = thinking.ApplyThinking(translated, req.Model, from.String(), to.String(), e.Identifier())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
requestedModel := helps.PayloadRequestedModel(opts, req.Model)
|
|
requestPath := helps.PayloadRequestPath(opts)
|
|
translated = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", translated, originalTranslated, requestedModel, requestPath, opts.Headers)
|
|
|
|
// Request usage data in the final streaming chunk so that token statistics
|
|
// are captured even when the upstream is an OpenAI-compatible provider.
|
|
translated, _ = sjson.SetBytes(translated, "stream_options.include_usage", true)
|
|
reporter.SetTranslatedReasoningEffort(translated, to.String())
|
|
|
|
url := strings.TrimSuffix(baseURL, "/") + "/chat/completions"
|
|
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(translated))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
httpReq.Header.Set("Content-Type", "application/json")
|
|
if apiKey != "" {
|
|
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
|
|
}
|
|
httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")
|
|
var attrs map[string]string
|
|
if auth != nil {
|
|
attrs = auth.Attributes
|
|
}
|
|
util.ApplyCustomHeadersFromAttrs(httpReq, attrs)
|
|
httpReq.Header.Set("Accept", "text/event-stream")
|
|
httpReq.Header.Set("Cache-Control", "no-cache")
|
|
var authID, authLabel, authType, authValue string
|
|
if auth != nil {
|
|
authID = auth.ID
|
|
authLabel = auth.Label
|
|
authType, authValue = auth.AccountInfo()
|
|
}
|
|
helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{
|
|
URL: url,
|
|
Method: http.MethodPost,
|
|
Headers: httpReq.Header.Clone(),
|
|
Body: translated,
|
|
Provider: e.Identifier(),
|
|
AuthID: authID,
|
|
AuthLabel: authLabel,
|
|
AuthType: authType,
|
|
AuthValue: authValue,
|
|
})
|
|
|
|
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, err
|
|
}
|
|
helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())
|
|
if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {
|
|
b, _ := io.ReadAll(httpResp.Body)
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, b)
|
|
helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b))
|
|
if errClose := httpResp.Body.Close(); errClose != nil {
|
|
log.Errorf("openai compat executor: close response body error: %v", errClose)
|
|
}
|
|
err = statusErr{code: httpResp.StatusCode, msg: string(b)}
|
|
return nil, err
|
|
}
|
|
out := make(chan cliproxyexecutor.StreamChunk)
|
|
go func() {
|
|
defer close(out)
|
|
defer func() {
|
|
if errClose := httpResp.Body.Close(); errClose != nil {
|
|
log.Errorf("openai compat executor: close response body error: %v", errClose)
|
|
}
|
|
}()
|
|
scanner := bufio.NewScanner(httpResp.Body)
|
|
scanner.Buffer(nil, 52_428_800) // 50MB
|
|
var param any
|
|
for scanner.Scan() {
|
|
line := scanner.Bytes()
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, line)
|
|
if detail, ok := helps.ParseOpenAIStreamUsage(line); ok {
|
|
reporter.Publish(ctx, detail)
|
|
}
|
|
trimmedLine := bytes.TrimSpace(line)
|
|
if len(trimmedLine) == 0 {
|
|
continue
|
|
}
|
|
|
|
if !bytes.HasPrefix(trimmedLine, []byte("data:")) {
|
|
if bytes.HasPrefix(trimmedLine, []byte(":")) || bytes.HasPrefix(trimmedLine, []byte("event:")) ||
|
|
bytes.HasPrefix(trimmedLine, []byte("id:")) || bytes.HasPrefix(trimmedLine, []byte("retry:")) {
|
|
continue
|
|
}
|
|
if bytes.HasPrefix(trimmedLine, []byte("{")) || bytes.HasPrefix(trimmedLine, []byte("[")) {
|
|
streamErr := statusErr{code: http.StatusBadGateway, msg: string(trimmedLine)}
|
|
helps.RecordAPIResponseError(ctx, e.cfg, streamErr)
|
|
reporter.PublishFailure(ctx, streamErr)
|
|
select {
|
|
case out <- cliproxyexecutor.StreamChunk{Err: streamErr}:
|
|
case <-ctx.Done():
|
|
}
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
// OpenAI-compatible streams must use SSE data lines.
|
|
chunks := sdktranslator.TranslateStream(ctx, to, from, req.Model, opts.OriginalRequest, translated, bytes.Clone(trimmedLine), ¶m)
|
|
for i := range chunks {
|
|
select {
|
|
case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
if errScan := scanner.Err(); errScan != nil {
|
|
helps.RecordAPIResponseError(ctx, e.cfg, errScan)
|
|
reporter.PublishFailure(ctx, errScan)
|
|
select {
|
|
case out <- cliproxyexecutor.StreamChunk{Err: errScan}:
|
|
case <-ctx.Done():
|
|
}
|
|
} else {
|
|
// In case the upstream close the stream without a terminal [DONE] marker.
|
|
// Feed a synthetic done marker through the translator so pending
|
|
// response.completed events are still emitted exactly once.
|
|
chunks := sdktranslator.TranslateStream(ctx, to, from, req.Model, opts.OriginalRequest, translated, []byte("data: [DONE]"), ¶m)
|
|
for i := range chunks {
|
|
select {
|
|
case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
// Ensure we record the request if no usage chunk was ever seen
|
|
reporter.EnsurePublished(ctx)
|
|
}()
|
|
return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) executeImagesStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, endpointPath string) (_ *cliproxyexecutor.StreamResult, err error) {
|
|
baseModel := thinking.ParseSuffix(req.Model).ModelName
|
|
|
|
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
|
|
defer reporter.TrackFailure(ctx, &err)
|
|
|
|
baseURL, apiKey := e.resolveCredentials(auth)
|
|
if baseURL == "" {
|
|
err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}
|
|
return nil, err
|
|
}
|
|
|
|
payload, contentType, errPrepare := prepareOpenAICompatImagesPayload(req.Payload, baseModel, opts.Headers.Get("Content-Type"), true)
|
|
if errPrepare != nil {
|
|
err = errPrepare
|
|
return nil, err
|
|
}
|
|
if contentType == "" {
|
|
contentType = "application/json"
|
|
}
|
|
reporter.SetTranslatedReasoningEffort(payload, "openai")
|
|
|
|
url := strings.TrimSuffix(baseURL, "/") + endpointPath
|
|
httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
httpReq.Header.Set("Content-Type", contentType)
|
|
httpReq.Header.Set("Accept", "text/event-stream")
|
|
httpReq.Header.Set("Cache-Control", "no-cache")
|
|
if apiKey != "" {
|
|
httpReq.Header.Set("Authorization", "Bearer "+apiKey)
|
|
}
|
|
httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")
|
|
var attrs map[string]string
|
|
if auth != nil {
|
|
attrs = auth.Attributes
|
|
}
|
|
util.ApplyCustomHeadersFromAttrs(httpReq, attrs)
|
|
var authID, authLabel, authType, authValue string
|
|
if auth != nil {
|
|
authID = auth.ID
|
|
authLabel = auth.Label
|
|
authType, authValue = auth.AccountInfo()
|
|
}
|
|
helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{
|
|
URL: url,
|
|
Method: http.MethodPost,
|
|
Headers: httpReq.Header.Clone(),
|
|
Body: payload,
|
|
Provider: e.Identifier(),
|
|
AuthID: authID,
|
|
AuthLabel: authLabel,
|
|
AuthType: authType,
|
|
AuthValue: authValue,
|
|
})
|
|
|
|
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, err
|
|
}
|
|
helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())
|
|
|
|
if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {
|
|
body, errRead := io.ReadAll(httpResp.Body)
|
|
if errClose := httpResp.Body.Close(); errClose != nil {
|
|
log.Errorf("openai compat executor: close response body error: %v", errClose)
|
|
}
|
|
if errRead != nil {
|
|
helps.RecordAPIResponseError(ctx, e.cfg, errRead)
|
|
return nil, errRead
|
|
}
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, body)
|
|
helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), body))
|
|
return nil, statusErr{code: httpResp.StatusCode, msg: string(body)}
|
|
}
|
|
|
|
out := make(chan cliproxyexecutor.StreamChunk)
|
|
go func() {
|
|
defer close(out)
|
|
defer func() {
|
|
if errClose := httpResp.Body.Close(); errClose != nil {
|
|
log.Errorf("openai compat executor: close response body error: %v", errClose)
|
|
}
|
|
reporter.EnsurePublished(ctx)
|
|
}()
|
|
buffer := make([]byte, 32*1024)
|
|
for {
|
|
n, errRead := httpResp.Body.Read(buffer)
|
|
if n > 0 {
|
|
chunk := bytes.Clone(buffer[:n])
|
|
helps.AppendAPIResponseChunk(ctx, e.cfg, chunk)
|
|
select {
|
|
case out <- cliproxyexecutor.StreamChunk{Payload: chunk}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
if errRead != nil {
|
|
if errRead != io.EOF {
|
|
helps.RecordAPIResponseError(ctx, e.cfg, errRead)
|
|
reporter.PublishFailure(ctx, errRead)
|
|
select {
|
|
case out <- cliproxyexecutor.StreamChunk{Err: errRead}:
|
|
case <-ctx.Done():
|
|
}
|
|
}
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) CountTokens(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (cliproxyexecutor.Response, error) {
|
|
baseModel := thinking.ParseSuffix(req.Model).ModelName
|
|
|
|
from := opts.SourceFormat
|
|
to := sdktranslator.FromString("openai")
|
|
translated := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, false)
|
|
|
|
modelForCounting := baseModel
|
|
|
|
translated, err := thinking.ApplyThinking(translated, req.Model, from.String(), to.String(), e.Identifier())
|
|
if err != nil {
|
|
return cliproxyexecutor.Response{}, err
|
|
}
|
|
|
|
enc, err := helps.TokenizerForModel(modelForCounting)
|
|
if err != nil {
|
|
return cliproxyexecutor.Response{}, fmt.Errorf("openai compat executor: tokenizer init failed: %w", err)
|
|
}
|
|
|
|
count, err := helps.CountOpenAIChatTokens(enc, translated)
|
|
if err != nil {
|
|
return cliproxyexecutor.Response{}, fmt.Errorf("openai compat executor: token counting failed: %w", err)
|
|
}
|
|
|
|
usageJSON := helps.BuildOpenAIUsageJSON(count)
|
|
translatedUsage := sdktranslator.TranslateTokenCount(ctx, to, from, count, usageJSON)
|
|
return cliproxyexecutor.Response{Payload: translatedUsage}, nil
|
|
}
|
|
|
|
// Refresh is a no-op for API-key based compatibility providers.
|
|
func (e *OpenAICompatExecutor) Refresh(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) {
|
|
log.Debugf("openai compat executor: refresh called")
|
|
if refreshed, handled, err := helps.RefreshAuthViaHome(ctx, e.cfg, auth); handled {
|
|
return refreshed, err
|
|
}
|
|
return auth, nil
|
|
}
|
|
|
|
func openAICompatImageEndpointPath(opts cliproxyexecutor.Options) string {
|
|
if opts.SourceFormat.String() != openAICompatImageHandlerType {
|
|
return ""
|
|
}
|
|
path := helps.PayloadRequestPath(opts)
|
|
if strings.HasSuffix(path, "/images/edits") {
|
|
return openAICompatImagesEditsPath
|
|
}
|
|
if strings.HasSuffix(path, "/images/generations") {
|
|
return openAICompatImagesGenerationsPath
|
|
}
|
|
return openAICompatDefaultImageEndpoint
|
|
}
|
|
|
|
func prepareOpenAICompatImagesPayload(payload []byte, model string, contentType string, stream bool) ([]byte, string, error) {
|
|
model = strings.TrimSpace(model)
|
|
contentType = strings.TrimSpace(contentType)
|
|
if json.Valid(payload) {
|
|
if model != "" {
|
|
payload, _ = sjson.SetBytes(payload, "model", model)
|
|
}
|
|
if stream {
|
|
payload, _ = sjson.SetBytes(payload, "stream", true)
|
|
} else {
|
|
payload, _ = sjson.DeleteBytes(payload, "stream")
|
|
}
|
|
return payload, "application/json", nil
|
|
}
|
|
|
|
mediaType, params, errParse := mime.ParseMediaType(contentType)
|
|
if errParse != nil || !strings.HasPrefix(strings.ToLower(strings.TrimSpace(mediaType)), "multipart/") {
|
|
return payload, contentType, nil
|
|
}
|
|
boundary := strings.TrimSpace(params["boundary"])
|
|
if boundary == "" {
|
|
return nil, "", fmt.Errorf("multipart boundary is missing")
|
|
}
|
|
return rewriteOpenAICompatImagesMultipartPayload(payload, model, boundary, stream)
|
|
}
|
|
|
|
func cloneOpenAICompatMIMEHeader(src textproto.MIMEHeader) textproto.MIMEHeader {
|
|
dst := make(textproto.MIMEHeader, len(src))
|
|
for key, values := range src {
|
|
dst[key] = append([]string(nil), values...)
|
|
}
|
|
return dst
|
|
}
|
|
|
|
func rewriteOpenAICompatImagesMultipartPayload(payload []byte, model string, boundary string, stream bool) ([]byte, string, error) {
|
|
reader := multipart.NewReader(bytes.NewReader(payload), boundary)
|
|
form, errRead := reader.ReadForm(openAICompatMultipartMemory)
|
|
if errRead != nil {
|
|
return nil, "", fmt.Errorf("read multipart form failed: %w", errRead)
|
|
}
|
|
defer func() {
|
|
if errRemove := form.RemoveAll(); errRemove != nil {
|
|
log.Errorf("openai compat executor: remove multipart form files error: %v", errRemove)
|
|
}
|
|
}()
|
|
|
|
var body bytes.Buffer
|
|
writer := multipart.NewWriter(&body)
|
|
if model != "" {
|
|
if errWrite := writer.WriteField("model", model); errWrite != nil {
|
|
return nil, "", fmt.Errorf("write model field failed: %w", errWrite)
|
|
}
|
|
}
|
|
if stream {
|
|
if errWrite := writer.WriteField("stream", "true"); errWrite != nil {
|
|
return nil, "", fmt.Errorf("write stream field failed: %w", errWrite)
|
|
}
|
|
}
|
|
for key, values := range form.Value {
|
|
if key == "model" || key == "stream" {
|
|
continue
|
|
}
|
|
for _, value := range values {
|
|
if errWrite := writer.WriteField(key, value); errWrite != nil {
|
|
return nil, "", fmt.Errorf("write form field %s failed: %w", key, errWrite)
|
|
}
|
|
}
|
|
}
|
|
for key, files := range form.File {
|
|
for _, fileHeader := range files {
|
|
if fileHeader == nil {
|
|
continue
|
|
}
|
|
header := cloneOpenAICompatMIMEHeader(fileHeader.Header)
|
|
header.Set("Content-Disposition", multipart.FileContentDisposition(key, fileHeader.Filename))
|
|
if header.Get("Content-Type") == "" {
|
|
header.Set("Content-Type", "application/octet-stream")
|
|
}
|
|
part, errCreate := writer.CreatePart(header)
|
|
if errCreate != nil {
|
|
return nil, "", fmt.Errorf("create file field %s failed: %w", key, errCreate)
|
|
}
|
|
src, errOpen := fileHeader.Open()
|
|
if errOpen != nil {
|
|
return nil, "", fmt.Errorf("open upload file failed: %w", errOpen)
|
|
}
|
|
_, errCopy := io.Copy(part, src)
|
|
if errClose := src.Close(); errClose != nil {
|
|
log.Errorf("openai compat executor: close upload file error: %v", errClose)
|
|
if errCopy == nil {
|
|
errCopy = errClose
|
|
}
|
|
}
|
|
if errCopy != nil {
|
|
return nil, "", fmt.Errorf("copy upload file failed: %w", errCopy)
|
|
}
|
|
}
|
|
}
|
|
if errClose := writer.Close(); errClose != nil {
|
|
return nil, "", fmt.Errorf("close multipart writer failed: %w", errClose)
|
|
}
|
|
return body.Bytes(), writer.FormDataContentType(), nil
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) resolveCredentials(auth *cliproxyauth.Auth) (baseURL, apiKey string) {
|
|
if auth == nil {
|
|
return "", ""
|
|
}
|
|
if auth.Attributes != nil {
|
|
baseURL = strings.TrimSpace(auth.Attributes["base_url"])
|
|
apiKey = strings.TrimSpace(auth.Attributes["api_key"])
|
|
}
|
|
return
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) resolveCompatConfig(auth *cliproxyauth.Auth) *config.OpenAICompatibility {
|
|
if auth == nil || e.cfg == nil {
|
|
return nil
|
|
}
|
|
candidates := make([]string, 0, 3)
|
|
if auth.Attributes != nil {
|
|
if v := strings.TrimSpace(auth.Attributes["compat_name"]); v != "" {
|
|
candidates = append(candidates, v)
|
|
}
|
|
if v := strings.TrimSpace(auth.Attributes["provider_key"]); v != "" {
|
|
candidates = append(candidates, v)
|
|
}
|
|
}
|
|
if v := strings.TrimSpace(auth.Provider); v != "" {
|
|
candidates = append(candidates, v)
|
|
}
|
|
for i := range e.cfg.OpenAICompatibility {
|
|
compat := &e.cfg.OpenAICompatibility[i]
|
|
if compat.Disabled {
|
|
continue
|
|
}
|
|
for _, candidate := range candidates {
|
|
if candidate != "" && strings.EqualFold(strings.TrimSpace(candidate), compat.Name) {
|
|
return compat
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (e *OpenAICompatExecutor) overrideModel(payload []byte, model string) []byte {
|
|
if len(payload) == 0 || model == "" {
|
|
return payload
|
|
}
|
|
payload, _ = sjson.SetBytes(payload, "model", model)
|
|
return payload
|
|
}
|
|
|
|
type statusErr struct {
|
|
code int
|
|
msg string
|
|
retryAfter *time.Duration
|
|
}
|
|
|
|
func (e statusErr) Error() string {
|
|
if e.msg != "" {
|
|
return e.msg
|
|
}
|
|
return fmt.Sprintf("status %d", e.code)
|
|
}
|
|
func (e statusErr) StatusCode() int { return e.code }
|
|
func (e statusErr) RetryAfter() *time.Duration { return e.retryAfter }
|