diff --git a/backend/internal/model/ai/provider/openai_compatible.go b/backend/internal/model/ai/provider/openai_compatible.go index 7762995..91667c4 100644 --- a/backend/internal/model/ai/provider/openai_compatible.go +++ b/backend/internal/model/ai/provider/openai_compatible.go @@ -84,7 +84,11 @@ func (p *OpenAICompatible) GenerateText(ctx context.Context, req domai.GenerateR return nil, app.For(code.AI).InputMissingRequired("missing AI provider token") } - body := p.buildChatCompletionRequest(req, false) + // Stream even for one-shot generation: long outputs (e.g. knowledge graph / + // research map) can exceed the provider's Cloudflare ~100s origin timeout in + // non-stream mode and surface as HTTP 524. Streaming keeps bytes flowing and + // we accumulate them back into a single result. + body := p.buildChatCompletionRequest(req, true) payload, err := json.Marshal(body) if err != nil { return nil, err @@ -96,7 +100,7 @@ func (p *OpenAICompatible) GenerateText(ctx context.Context, req domai.GenerateR } httpReq.Header.Set("Authorization", "Bearer "+req.Credential.APIKey) httpReq.Header.Set("Content-Type", "application/json") - httpReq.Header.Set("Accept", "application/json") + httpReq.Header.Set("Accept", "text/event-stream") resp, err := p.client.Do(httpReq) if err != nil { @@ -104,26 +108,70 @@ func (p *OpenAICompatible) GenerateText(ctx context.Context, req domai.GenerateR } defer resp.Body.Close() - data, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) - if err != nil { - return nil, err - } if resp.StatusCode < 200 || resp.StatusCode >= 300 { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4096)) return nil, providerHTTPError("AI provider request failed", resp.StatusCode, resp.Status) } - var chatResp openAIChatResponse - if err := json.Unmarshal(data, &chatResp); err != nil { - return nil, app.For(code.AI).SvcThirdParty("failed to parse AI provider chat response") + return accumulateChatStream(resp.Body) +} + +// accumulateChatStream consumes an OpenAI-compatible SSE stream and folds the +// chunks back into a single GenerateResult. Final answer content and reasoning +// are kept apart so reasoning never gets prepended to (and corrupts) the parsed +// answer; reasoning is only used as a fallback when no content was emitted. +func accumulateChatStream(body io.Reader) (*domai.GenerateResult, error) { + scanner := bufio.NewScanner(body) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + var content strings.Builder + var reasoning strings.Builder + finishReason := "" + + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || !strings.HasPrefix(line, "data:") { + continue + } + data := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if data == "[DONE]" { + break + } + var chunk openAIStreamChunk + if err := json.Unmarshal([]byte(data), &chunk); err != nil { + return nil, app.For(code.AI).SvcThirdParty("failed to parse AI provider chat response") + } + for _, choice := range chunk.Choices { + delta := choice.Delta + if delta.Content != "" { + content.WriteString(delta.Content) + } else if delta.Text != "" { + content.WriteString(delta.Text) + } + if delta.ReasoningContent != "" { + reasoning.WriteString(delta.ReasoningContent) + } else if delta.Reasoning != "" { + reasoning.WriteString(delta.Reasoning) + } + if choice.FinishReason != "" { + finishReason = choice.FinishReason + } + } } - if len(chatResp.Choices) == 0 { - return nil, app.For(code.AI).SvcThirdParty("AI provider returned no choices") + if err := scanner.Err(); err != nil { + return nil, app.For(code.AI).SvcThirdParty("AI provider stream read failed: " + err.Error()) } - choice := chatResp.Choices[0] + text := content.String() + if strings.TrimSpace(text) == "" { + text = reasoning.String() + } + if strings.TrimSpace(text) == "" { + return nil, app.For(code.AI).SvcThirdParty("AI provider returned no choices") + } return &domai.GenerateResult{ - Text: extractAssistantText(choice.Message), - FinishReason: choice.FinishReason, + Text: text, + FinishReason: finishReason, }, nil } diff --git a/backend/internal/model/ai/provider/openai_compatible_test.go b/backend/internal/model/ai/provider/openai_compatible_test.go index 5ce9038..9eeef0c 100644 --- a/backend/internal/model/ai/provider/openai_compatible_test.go +++ b/backend/internal/model/ai/provider/openai_compatible_test.go @@ -60,8 +60,8 @@ func TestOpenAICompatible_GenerateText_UsesContent(t *testing.T) { if err := json.Unmarshal(body, &req); err != nil { t.Fatalf("unmarshal request: %v", err) } - if req.Stream { - t.Fatal("expected non-stream request") + if !req.Stream { + t.Fatal("expected stream request to avoid upstream 524 on long generations") } if req.Thinking == nil || req.Thinking.Type != "disabled" { t.Fatalf("thinking = %+v, want disabled for deepseek on opencode", req.Thinking) @@ -69,7 +69,10 @@ func TestOpenAICompatible_GenerateText_UsesContent(t *testing.T) { if req.MaxTokens == nil || *req.MaxTokens < 2048 { t.Fatalf("max_tokens = %+v, want >= 2048 for deepseek on opencode", req.MaxTokens) } - _, _ = w.Write([]byte(`{"choices":[{"message":{"content":"hello"},"finish_reason":"stop"}]}`)) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hel\"}}]}\n\n")) + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"lo\"},\"finish_reason\":\"stop\"}]}\n\n")) + _, _ = w.Write([]byte("data: [DONE]\n\n")) })) defer server.Close() @@ -92,7 +95,9 @@ func TestOpenAICompatible_GenerateText_UsesContent(t *testing.T) { func TestOpenAICompatible_GenerateText_FallsBackToReasoningContent(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - _, _ = w.Write([]byte(`{"choices":[{"message":{"content":"","reasoning_content":"thinking only"},"finish_reason":"length"}]}`)) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"thinking only\"},\"finish_reason\":\"length\"}]}\n\n")) + _, _ = w.Write([]byte("data: [DONE]\n\n")) })) defer server.Close()