opencode/picoclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(openai_compat): add configurable streaming for custom APIs

Browse source
This commit is contained in:
Anton Chen 2026-03-19 08:58:08 +08:00
parent e73d9d959e
commit 4d4d898a92
No known key found for this signature in database
GPG key ID: 44D9CF9904BB2AAE
14 changed files with 319 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
config/config.example.json
View file

@ -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",

3
docs/migration/model-list-migration.md
View file

@ -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:

32
docs/providers.md
View file

@ -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

1
pkg/config/config.go
View file

@ -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.)

3
pkg/providers/factory_provider.go
View file

@ -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":

4
pkg/providers/http_provider.go
View file

@ -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),
),
}
}

165
pkg/providers/openai_compat/provider.go
View file

@ -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 {

72
pkg/providers/openai_compat/provider_test.go
View file

@ -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("<!DOCTYPE html><html><body>"), bytes.Repeat([]byte("A"), 2048)...)
body = append(body, []byte("</body></html>")...)

2
web/backend/api/models.go
View file

@ -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,

1
web/frontend/src/api/models.ts
View file

@ -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

12
web/frontend/src/components/models/add-model-sheet.tsx
View file

@ -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({
/>
</Field>
<SwitchCardField
label={t("models.field.stream")}
hint={t("models.field.streamHint")}
checked={form.stream}
onCheckedChange={(checked) =>
setForm((f) => ({ ...f, stream: checked }))
}
/>
<Field
label={t("models.field.authMethod")}
hint={t("models.field.authMethodHint")}

13
web/frontend/src/components/models/edit-model-sheet.tsx
View file

@ -25,6 +25,7 @@ interface EditForm {
apiKey: string
apiBase: string
proxy: string
stream: boolean
authMethod: string
connectMode: string
workspace: string
@ -52,6 +53,7 @@ export function EditModelSheet({
apiKey: "",
apiBase: "",
proxy: "",
stream: false,
authMethod: "",
connectMode: "",
workspace: "",
@ -70,6 +72,7 @@ export function EditModelSheet({
apiKey: "",
apiBase: model.api_base ?? "",
proxy: model.proxy ?? "",
stream: model.stream ?? false,
authMethod: model.auth_method ?? "",
connectMode: model.connect_mode ?? "",
workspace: model.workspace ?? "",
@ -100,6 +103,7 @@ export function EditModelSheet({
api_base: form.apiBase || undefined,
api_key: form.apiKey || undefined,
proxy: form.proxy || undefined,
stream: form.stream || undefined,
auth_method: form.authMethod || undefined,
connect_mode: form.connectMode || undefined,
workspace: form.workspace || undefined,
@ -193,6 +197,15 @@ export function EditModelSheet({
/>
</Field>
<SwitchCardField
label={t("models.field.stream")}
hint={t("models.field.streamHint")}
checked={form.stream}
onCheckedChange={(checked) =>
setForm((f) => ({ ...f, stream: checked }))
}
/>
<Field
label={t("models.field.authMethod")}
hint={t("models.field.authMethodHint")}

2
web/frontend/src/i18n/locales/en.json
View file

@ -195,6 +195,8 @@
"apiKeyPlaceholderSet": "Leave blank to keep existing key",
"proxy": "HTTP Proxy",
"proxyHint": "Optional. e.g. http://127.0.0.1:7890",
"stream": "Force Streaming",
"streamHint": "Send chat/completions requests with stream=true and parse SSE responses.",
"authMethod": "Auth Method",
"authMethodHint": "Authentication method: oauth, token. Leave blank for API key auth.",
"connectMode": "Connect Mode",

2
web/frontend/src/i18n/locales/zh.json
View file

@ -195,6 +195,8 @@
"apiKeyPlaceholderSet": "留空保持原有 Key 不变",
"proxy": "HTTP 代理",
"proxyHint": "可选。例如 http://127.0.0.1:7890",
"stream": "强制流式返回",
"streamHint": "以 stream=true 发送 chat/completions 请求，并解析 SSE 响应。",
"authMethod": "认证方式",
"authMethodHint": "认证方式：oauth、token。留空表示使用 API Key 认证。",
"connectMode": "连接模式",