diff --git a/config/config.example.json b/config/config.example.json index c214f26fa..8b231af38 100644 --- a/config/config.example.json +++ b/config/config.example.json @@ -59,6 +59,14 @@ "api_key": "your-azure-api-key", "api_base": "https://your-resource.openai.azure.com" }, + { + "_comment": "Some OpenAI-compatible relays require chat/completions requests with stream=true", + "model_name": "custom-relay-gpt-5.4", + "model": "openai/gpt-5.4", + "api_key": "sk-your-relay-key", + "api_base": "https://relay.example.com/v1", + "stream": true + }, { "model_name": "loadbalanced-gpt-5.4", "model": "openai/gpt-5.4", diff --git a/docs/migration/model-list-migration.md b/docs/migration/model-list-migration.md index eed228d4d..f141a7e9a 100644 --- a/docs/migration/model-list-migration.md +++ b/docs/migration/model-list-migration.md @@ -113,6 +113,7 @@ The `model` field uses a protocol prefix format: `[protocol/]model-identifier` | `api_base` | No | API endpoint URL | | `api_key` | No* | API authentication key | | `proxy` | No | HTTP proxy URL | +| `stream` | No | Force `chat/completions` requests to use `stream=true` and parse SSE responses | | `auth_method` | No | Authentication method: `oauth`, `token` | | `connect_mode` | No | Connection mode for CLI providers: `stdio`, `grpc` | | `rpm` | No | Requests per minute limit | @@ -121,6 +122,8 @@ The `model` field uses a protocol prefix format: `[protocol/]model-identifier` *`api_key` is required for HTTP-based protocols unless `api_base` points to a local server. +Use `stream: true` for OpenAI-compatible relays that reject non-streaming requests or always return `text/event-stream`. + ## Load Balancing Configure multiple endpoints for the same model to distribute load: diff --git a/docs/providers.md b/docs/providers.md index e62cbb969..e1d7a2d38 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -39,6 +39,26 @@ This design also enables **multi-agent support** with flexible provider selectio - **Load balancing**: Distribute requests across multiple endpoints - **Centralized configuration**: Manage all providers in one place +#### Model Fields + +| Field | Required | Description | +| ----- | -------- | ----------- | +| `model_name` | Yes | User-facing alias for the model | +| `model` | Yes | Protocol and model identifier (for example `openai/gpt-5.4`) | +| `api_base` | No | API endpoint URL | +| `api_key` | No* | API authentication key | +| `proxy` | No | HTTP proxy URL | +| `stream` | No | Force `chat/completions` requests to send `stream=true` and parse SSE responses. Useful for OpenAI-compatible relays that reject non-streaming requests. | +| `auth_method` | No | Authentication method: `oauth`, `token` | +| `connect_mode` | No | Connection mode for CLI providers: `stdio`, `grpc` | +| `workspace` | No | Working directory for CLI-based providers | +| `rpm` | No | Requests per minute limit | +| `max_tokens_field` | No | Override the request field name for max tokens | +| `request_timeout` | No | HTTP request timeout in seconds; `<=0` uses the default timeout | +| `thinking_level` | No | Extended thinking budget: `off`, `low`, `medium`, `high`, `xhigh`, `adaptive` | + +*`api_key` is required for HTTP-based protocols unless `api_base` points to a local server. + #### 📋 All Supported Vendors | Vendor | `model` Prefix | Default API Base | Protocol | API Key | @@ -196,6 +216,18 @@ For direct Anthropic API access or custom endpoints that only support Anthropic' } ``` +If your relay rejects non-streaming `chat/completions` calls with errors such as `Stream must be set to true`, enable `stream`: + +```json +{ + "model_name": "my-custom-model", + "model": "openai/custom-model", + "api_base": "https://my-proxy.com/v1", + "api_key": "sk-...", + "stream": true +} +``` + **LiteLLM Proxy** ```json diff --git a/pkg/config/config.go b/pkg/config/config.go index 739f8d373..994bbffc3 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -607,6 +607,7 @@ type ModelConfig struct { APIKey string `json:"api_key"` // API authentication key (single key) APIKeys []string `json:"api_keys,omitempty"` // API authentication keys (multiple keys for failover) Proxy string `json:"proxy,omitempty"` // HTTP proxy URL + Stream bool `json:"stream,omitempty"` // Force Chat Completions SSE streaming Fallbacks []string `json:"fallbacks,omitempty"` // Fallback model names for failover // Special providers (CLI-based, OAuth, etc.) diff --git a/pkg/providers/factory_provider.go b/pkg/providers/factory_provider.go index dbb5db5cb..22cb03064 100644 --- a/pkg/providers/factory_provider.go +++ b/pkg/providers/factory_provider.go @@ -93,6 +93,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err cfg.Proxy, cfg.MaxTokensField, cfg.RequestTimeout, + cfg.Stream, ), modelID, nil case "azure", "azure-openai": @@ -131,6 +132,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err cfg.Proxy, cfg.MaxTokensField, cfg.RequestTimeout, + cfg.Stream, ), modelID, nil case "anthropic": @@ -156,6 +158,7 @@ func CreateProviderFromConfig(cfg *config.ModelConfig) (LLMProvider, string, err cfg.Proxy, cfg.MaxTokensField, cfg.RequestTimeout, + cfg.Stream, ), modelID, nil case "anthropic-messages": diff --git a/pkg/providers/http_provider.go b/pkg/providers/http_provider.go index 4d823630e..05f330057 100644 --- a/pkg/providers/http_provider.go +++ b/pkg/providers/http_provider.go @@ -24,12 +24,13 @@ func NewHTTPProvider(apiKey, apiBase, proxy string) *HTTPProvider { } func NewHTTPProviderWithMaxTokensField(apiKey, apiBase, proxy, maxTokensField string) *HTTPProvider { - return NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(apiKey, apiBase, proxy, maxTokensField, 0) + return NewHTTPProviderWithMaxTokensFieldAndRequestTimeout(apiKey, apiBase, proxy, maxTokensField, 0, false) } func NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( apiKey, apiBase, proxy, maxTokensField string, requestTimeoutSeconds int, + forceStream bool, ) *HTTPProvider { return &HTTPProvider{ delegate: openai_compat.NewProvider( @@ -38,6 +39,7 @@ func NewHTTPProviderWithMaxTokensFieldAndRequestTimeout( proxy, openai_compat.WithMaxTokensField(maxTokensField), openai_compat.WithRequestTimeout(time.Duration(requestTimeoutSeconds)*time.Second), + openai_compat.WithStreaming(forceStream), ), } } diff --git a/pkg/providers/openai_compat/provider.go b/pkg/providers/openai_compat/provider.go index 463db83c9..7c85b487f 100644 --- a/pkg/providers/openai_compat/provider.go +++ b/pkg/providers/openai_compat/provider.go @@ -1,12 +1,15 @@ package openai_compat import ( + "bufio" "bytes" "context" "encoding/json" "fmt" + "io" "net/http" "net/url" + "strconv" "strings" "time" @@ -30,6 +33,7 @@ type ( type Provider struct { apiKey string apiBase string + forceStream bool maxTokensField string // Field name for max tokens (e.g., "max_completion_tokens" for o1/glm models) httpClient *http.Client } @@ -44,6 +48,12 @@ func WithMaxTokensField(maxTokensField string) Option { } } +func WithStreaming(forceStream bool) Option { + return func(p *Provider) { + p.forceStream = forceStream + } +} + func WithRequestTimeout(timeout time.Duration) Option { return func(p *Provider) { if timeout > 0 { @@ -102,6 +112,10 @@ func (p *Provider) Chat( "model": model, "messages": common.SerializeMessages(messages), } + if p.forceStream { + requestBody["stream"] = true + requestBody["stream_options"] = map[string]any{"include_usage": true} + } // When fallback uses a different provider (e.g. DeepSeek), that provider must not inject web_search_preview. nativeSearch, _ := options["native_search"].(bool) @@ -175,9 +189,160 @@ func (p *Provider) Chat( return nil, common.HandleErrorResponse(resp, p.apiBase) } + if p.forceStream || strings.Contains(strings.ToLower(resp.Header.Get("Content-Type")), "text/event-stream") { + return p.readStreamResponse(resp.Body) + } + return common.ReadAndParseResponse(resp, p.apiBase) } +func (p *Provider) readStreamResponse(body io.Reader) (*LLMResponse, error) { + scanner := bufio.NewScanner(body) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + var out LLMResponse + var toolCalls []streamToolCall + + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || !strings.HasPrefix(line, "data:") { + continue + } + + data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if data == "" { + continue + } + if data == "[DONE]" { + break + } + + if err := mergeStreamChunk(data, &out, &toolCalls); err != nil { + return nil, err + } + } + + if err := scanner.Err(); err != nil { + return nil, fmt.Errorf("failed to read SSE response: %w", err) + } + + if len(toolCalls) > 0 { + out.ToolCalls = finalizeStreamToolCalls(toolCalls) + } + if out.FinishReason == "" { + out.FinishReason = "stop" + } + return &out, nil +} + +type streamChunk struct { + 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 json.RawMessage `json:"arguments"` + } `json:"function"` + } `json:"tool_calls"` + } `json:"delta"` + FinishReason *string `json:"finish_reason"` + } `json:"choices"` + Usage *UsageInfo `json:"usage"` +} + +type streamToolCall struct { + ID string + Type string + Name string + Arguments strings.Builder +} + +func mergeStreamChunk(data string, out *LLMResponse, toolCalls *[]streamToolCall) error { + var chunk streamChunk + if err := json.Unmarshal([]byte(data), &chunk); err != nil { + return fmt.Errorf("failed to decode SSE chunk: %w", err) + } + + if chunk.Usage != nil { + out.Usage = chunk.Usage + } + + for _, choice := range chunk.Choices { + out.Content += choice.Delta.Content + out.ReasoningContent += choice.Delta.ReasoningContent + out.Reasoning += choice.Delta.Reasoning + + for _, tc := range choice.Delta.ToolCalls { + index := len(*toolCalls) + if tc.Index != nil && *tc.Index >= 0 { + index = *tc.Index + } + for len(*toolCalls) <= index { + *toolCalls = append(*toolCalls, streamToolCall{}) + } + + current := &(*toolCalls)[index] + if tc.ID != "" { + current.ID = tc.ID + } + if tc.Type != "" { + current.Type = tc.Type + } + if tc.Function != nil { + if tc.Function.Name != "" { + current.Name = tc.Function.Name + } + if len(tc.Function.Arguments) > 0 { + current.Arguments.WriteString(streamArgumentText(tc.Function.Arguments)) + } + } + } + + if choice.FinishReason != nil && *choice.FinishReason != "" { + out.FinishReason = *choice.FinishReason + } + } + + return nil +} + +func finalizeStreamToolCalls(streamCalls []streamToolCall) []ToolCall { + result := make([]ToolCall, 0, len(streamCalls)) + for i, tc := range streamCalls { + if tc.ID == "" && tc.Name == "" && tc.Arguments.Len() == 0 { + continue + } + id := tc.ID + if id == "" { + id = "call_" + strconv.Itoa(i) + } + result = append(result, ToolCall{ + ID: id, + Type: tc.Type, + Name: tc.Name, + Arguments: common.DecodeToolCallArguments(json.RawMessage(tc.Arguments.String()), tc.Name), + }) + } + return result +} + +func streamArgumentText(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + var s string + if err := json.Unmarshal(raw, &s); err == nil { + return s + } + return string(raw) +} + func normalizeModel(model, apiBase string) string { before, after, ok := strings.Cut(model, "/") if !ok { diff --git a/pkg/providers/openai_compat/provider_test.go b/pkg/providers/openai_compat/provider_test.go index efb03ccb8..64b99f576 100644 --- a/pkg/providers/openai_compat/provider_test.go +++ b/pkg/providers/openai_compat/provider_test.go @@ -367,6 +367,78 @@ func TestProviderChat_SuccessResponseUsesStreamingDecoder(t *testing.T) { } } +func TestProviderChat_ForceStreamRequestsSSE(t *testing.T) { + var requestBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(&requestBody); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + })) + defer server.Close() + + p := NewProvider("key", server.URL, "", WithStreaming(true)) + out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "gpt-4o", nil) + if err != nil { + t.Fatalf("Chat() error = %v", err) + } + + if requestBody["stream"] != true { + t.Fatalf("stream = %v, want true", requestBody["stream"]) + } + streamOptions, ok := requestBody["stream_options"].(map[string]any) + if !ok || streamOptions["include_usage"] != true { + t.Fatalf("stream_options = %#v, want include_usage=true", requestBody["stream_options"]) + } + if out.Content != "ok" { + t.Fatalf("Content = %q, want ok", out.Content) + } +} + +func TestProviderChat_ParsesStreamedToolCalls(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"Hello \"}}]}\n\n") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"world\"}}]}\n\n") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"get_weather\",\"arguments\":\"{\\\"city\\\":\\\"\"}}]}}]}\n\n") + fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"SF\\\"}\"}}],\"reasoning_content\":\"thinking\"},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":5,\"total_tokens\":15}}\n\n") + fmt.Fprint(w, "data: [DONE]\n\n") + })) + defer server.Close() + + p := NewProvider("key", server.URL, "", WithStreaming(true)) + out, err := p.Chat(t.Context(), []Message{{Role: "user", Content: "hi"}}, nil, "gpt-4o", nil) + if err != nil { + t.Fatalf("Chat() error = %v", err) + } + + if out.Content != "Hello world" { + t.Fatalf("Content = %q, want %q", out.Content, "Hello world") + } + if out.ReasoningContent != "thinking" { + t.Fatalf("ReasoningContent = %q, want thinking", out.ReasoningContent) + } + if out.FinishReason != "tool_calls" { + t.Fatalf("FinishReason = %q, want tool_calls", out.FinishReason) + } + if out.Usage == nil || out.Usage.TotalTokens != 15 { + t.Fatalf("Usage = %#v, want total_tokens=15", out.Usage) + } + if len(out.ToolCalls) != 1 { + t.Fatalf("len(ToolCalls) = %d, want 1", len(out.ToolCalls)) + } + if out.ToolCalls[0].Name != "get_weather" { + t.Fatalf("ToolCalls[0].Name = %q, want get_weather", out.ToolCalls[0].Name) + } + if out.ToolCalls[0].Arguments["city"] != "SF" { + t.Fatalf("ToolCalls[0].Arguments[city] = %v, want SF", out.ToolCalls[0].Arguments["city"]) + } +} + func TestProviderChat_LargeHTMLResponsePreviewIsTruncated(t *testing.T) { body := append([]byte(""), bytes.Repeat([]byte("A"), 2048)...) body = append(body, []byte("")...) diff --git a/web/backend/api/models.go b/web/backend/api/models.go index 7f3d29c77..32b9a49ac 100644 --- a/web/backend/api/models.go +++ b/web/backend/api/models.go @@ -29,6 +29,7 @@ type modelResponse struct { APIBase string `json:"api_base,omitempty"` APIKey string `json:"api_key"` Proxy string `json:"proxy,omitempty"` + Stream bool `json:"stream,omitempty"` AuthMethod string `json:"auth_method,omitempty"` // Advanced fields ConnectMode string `json:"connect_mode,omitempty"` @@ -74,6 +75,7 @@ func (h *Handler) handleListModels(w http.ResponseWriter, r *http.Request) { APIBase: m.APIBase, APIKey: maskAPIKey(m.APIKey), Proxy: m.Proxy, + Stream: m.Stream, AuthMethod: m.AuthMethod, ConnectMode: m.ConnectMode, Workspace: m.Workspace, diff --git a/web/frontend/src/api/models.ts b/web/frontend/src/api/models.ts index 8e49b48b4..457994415 100644 --- a/web/frontend/src/api/models.ts +++ b/web/frontend/src/api/models.ts @@ -9,6 +9,7 @@ export interface ModelInfo { api_base?: string api_key: string proxy?: string + stream?: boolean auth_method?: string // Advanced fields connect_mode?: string diff --git a/web/frontend/src/components/models/add-model-sheet.tsx b/web/frontend/src/components/models/add-model-sheet.tsx index c760bc672..3e96604aa 100644 --- a/web/frontend/src/components/models/add-model-sheet.tsx +++ b/web/frontend/src/components/models/add-model-sheet.tsx @@ -27,6 +27,7 @@ interface AddForm { apiBase: string apiKey: string proxy: string + stream: boolean authMethod: string connectMode: string workspace: string @@ -42,6 +43,7 @@ const EMPTY_ADD_FORM: AddForm = { apiBase: "", apiKey: "", proxy: "", + stream: false, authMethod: "", connectMode: "", workspace: "", @@ -120,6 +122,7 @@ export function AddModelSheet({ api_base: form.apiBase.trim() || undefined, api_key: form.apiKey.trim() || undefined, proxy: form.proxy.trim() || undefined, + stream: form.stream || undefined, auth_method: form.authMethod.trim() || undefined, connect_mode: form.connectMode.trim() || undefined, workspace: form.workspace.trim() || undefined, @@ -225,6 +228,15 @@ export function AddModelSheet({ /> + + setForm((f) => ({ ...f, stream: checked })) + } + /> + + + setForm((f) => ({ ...f, stream: checked })) + } + /> +