fix2
This commit is contained in:
parent
a18909f579
commit
9d7853c75a
|
|
@ -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")
|
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)
|
payload, err := json.Marshal(body)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
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("Authorization", "Bearer "+req.Credential.APIKey)
|
||||||
httpReq.Header.Set("Content-Type", "application/json")
|
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)
|
resp, err := p.client.Do(httpReq)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
@ -104,26 +108,70 @@ func (p *OpenAICompatible) GenerateText(ctx context.Context, req domai.GenerateR
|
||||||
}
|
}
|
||||||
defer resp.Body.Close()
|
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 {
|
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)
|
return nil, providerHTTPError("AI provider request failed", resp.StatusCode, resp.Status)
|
||||||
}
|
}
|
||||||
|
|
||||||
var chatResp openAIChatResponse
|
return accumulateChatStream(resp.Body)
|
||||||
if err := json.Unmarshal(data, &chatResp); err != nil {
|
}
|
||||||
return nil, app.For(code.AI).SvcThirdParty("failed to parse AI provider chat response")
|
|
||||||
|
// 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 {
|
if err := scanner.Err(); err != nil {
|
||||||
return nil, app.For(code.AI).SvcThirdParty("AI provider returned no choices")
|
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{
|
return &domai.GenerateResult{
|
||||||
Text: extractAssistantText(choice.Message),
|
Text: text,
|
||||||
FinishReason: choice.FinishReason,
|
FinishReason: finishReason,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -60,8 +60,8 @@ func TestOpenAICompatible_GenerateText_UsesContent(t *testing.T) {
|
||||||
if err := json.Unmarshal(body, &req); err != nil {
|
if err := json.Unmarshal(body, &req); err != nil {
|
||||||
t.Fatalf("unmarshal request: %v", err)
|
t.Fatalf("unmarshal request: %v", err)
|
||||||
}
|
}
|
||||||
if req.Stream {
|
if !req.Stream {
|
||||||
t.Fatal("expected non-stream request")
|
t.Fatal("expected stream request to avoid upstream 524 on long generations")
|
||||||
}
|
}
|
||||||
if req.Thinking == nil || req.Thinking.Type != "disabled" {
|
if req.Thinking == nil || req.Thinking.Type != "disabled" {
|
||||||
t.Fatalf("thinking = %+v, want disabled for deepseek on opencode", req.Thinking)
|
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 {
|
if req.MaxTokens == nil || *req.MaxTokens < 2048 {
|
||||||
t.Fatalf("max_tokens = %+v, want >= 2048 for deepseek on opencode", req.MaxTokens)
|
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()
|
defer server.Close()
|
||||||
|
|
||||||
|
|
@ -92,7 +95,9 @@ func TestOpenAICompatible_GenerateText_UsesContent(t *testing.T) {
|
||||||
|
|
||||||
func TestOpenAICompatible_GenerateText_FallsBackToReasoningContent(t *testing.T) {
|
func TestOpenAICompatible_GenerateText_FallsBackToReasoningContent(t *testing.T) {
|
||||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
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()
|
defer server.Close()
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue