diff --git a/internal/entity/models/groq.go b/internal/entity/models/groq.go index 25081df728..fa84d9b4a1 100644 --- a/internal/entity/models/groq.go +++ b/internal/entity/models/groq.go @@ -123,67 +123,12 @@ func (g *GroqModel) ChatWithMessages(ctx context.Context, modelName string, mess reqBody := buildRequestBody(chatModelConfig, modelName, messages, false) applyGroqReasoningRequestParams(reqBody, modelName, chatModelConfig) - jsonData, err := json.Marshal(reqBody) + body, err := g.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout) if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) + return nil, err } - ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout) - defer cancel() - - req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(jsonData)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey)) - - resp, err := g.baseModel.httpClient.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to send request: %w", err) - } - defer resp.Body.Close() - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read response: %w", err) - } - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) - } - - return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, _ *ChatConfig) (chatResponseParts, error) { - var result groqChatResponse - if err := json.Unmarshal(body, &result); err != nil { - return chatResponseParts{}, fmt.Errorf("failed to parse response: %w", err) - } - if result.Error != nil { - return chatResponseParts{}, fmt.Errorf("groq: upstream error: %v", result.Error) - } - if len(result.Choices) == 0 { - return chatResponseParts{}, fmt.Errorf("no choices in response") - } - - content := result.Choices[0].Message.Content - reasonContent := result.Choices[0].Message.ReasoningContent - if reasonContent == "" { - reasonContent = result.Choices[0].Message.Reasoning - } - - parts := chatResponseParts{ - RequestID: result.ID, - Content: &content, - ReasonContent: &reasonContent, - } - if result.Usage != nil { - parts.Usage = &TokenUsage{ - PromptTokens: result.Usage.PromptTokens, - CompletionTokens: result.Usage.CompletionTokens, - TotalTokens: result.Usage.TotalTokens, - } - } - return parts, nil - }) + return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig) } func (g *GroqModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error { @@ -211,6 +156,7 @@ func (g *GroqModel) ChatStreamlyWithSender(ctx context.Context, modelName string reqBody := buildRequestBody(chatModelConfig, modelName, messages, true) applyGroqReasoningRequestParams(reqBody, modelName, chatModelConfig) + reqBody["stream_options"] = map[string]any{"include_usage": true} jsonData, err := json.Marshal(reqBody) if err != nil { return fmt.Errorf("failed to marshal request: %w", err) @@ -238,66 +184,7 @@ func (g *GroqModel) ChatStreamlyWithSender(ctx context.Context, modelName string return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) } - sawTerminal := false - accumulatedToolCalls := make(map[int]map[string]interface{}) - done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error { - common.Info(fmt.Sprintf("%v", event)) - - tokenUsage, found, usageErr := decodeOpenAICompatibleStreamUsage(event) - if usageErr != nil { - return usageErr - } - if found { - applyStreamUsage(chatModelConfig, modelUsage, tokenUsage) - } - - choices, ok := event["choices"].([]interface{}) - if !ok || len(choices) == 0 { - return nil - } - - firstChoice, ok := choices[0].(map[string]interface{}) - if !ok { - return nil - } - - delta, ok := firstChoice["delta"].(map[string]interface{}) - if !ok { - return nil - } - - accumulateToolCallDeltas(delta, accumulatedToolCalls) - - content, ok := delta["content"].(string) - if ok && content != "" { - if err := sender(&content, nil); err != nil { - return err - } - } - - reasoningContent, ok := delta["reasoning_content"].(string) - if ok && reasoningContent != "" { - if err := sender(nil, &reasoningContent); err != nil { - return err - } - } - - finishReason, ok := firstChoice["finish_reason"].(string) - if ok && finishReason != "" { - sawTerminal = true - } - return nil - }) - if err != nil { - return fmt.Errorf("failed to scan response body: %w", err) - } - if !done && !sawTerminal { - return fmt.Errorf("deepseek: stream ended before [DONE] or finish_reason") - } - - setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) - endOfStream := "[DONE]" - return sender(&endOfStream, nil) + return HandleStreamingResponse(resp.Body, modelUsage, chatModelConfig, OpenAIParserConfig, sender) } type groqModelInfo struct { diff --git a/internal/entity/models/longcat.go b/internal/entity/models/longcat.go index 275ca78dc5..846a9a2d52 100644 --- a/internal/entity/models/longcat.go +++ b/internal/entity/models/longcat.go @@ -98,75 +98,12 @@ func (l *LongCatModel) ChatWithMessages(ctx context.Context, modelName string, m reqBody := buildRequestBody(chatModelConfig, modelName, messages, false) delete(reqBody, "stop") - jsonData, err := json.Marshal(reqBody) + body, err := l.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout) if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) + return nil, err } - ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout) - defer cancel() - - req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey)) - - resp, err := l.baseModel.httpClient.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to send request: %w", err) - } - defer resp.Body.Close() - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read response: %w", err) - } - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) - } - - return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, chatConfig *ChatConfig) (chatResponseParts, error) { - var result LongCatChatResponse - if err := json.Unmarshal(body, &result); err != nil { - return chatResponseParts{}, fmt.Errorf("failed to parse response: %w", err) - } - if len(result.Choices) == 0 { - return chatResponseParts{}, fmt.Errorf("no choices in response") - } - - choice := &result.Choices[0] - content := choice.Message.Content - // LongCat-Flash-Thinking may emit all output as reasoning_content - // with content left empty. That is a valid response — only reject when - // there is neither content, reasoning, nor tool calls. - if content == "" && choice.Message.ReasoningContent == "" && len(choice.Message.ToolCalls) == 0 { - return chatResponseParts{}, fmt.Errorf("invalid content format") - } - - // reasoning_content is typically prefixed with a leading newline. - reasonContent := choice.Message.ReasoningContent - if reasonContent != "" && reasonContent[0] == '\n' { - reasonContent = reasonContent[1:] - } - - usage := &TokenUsage{ - PromptTokens: result.Usage.PromptTokens, - CompletionTokens: result.Usage.CompletionTokens, - TotalTokens: result.Usage.TotalTokens, - } - - return chatResponseParts{ - RequestID: result.ID, - Content: &content, - ReasonContent: &reasonContent, - ToolCalls: choice.Message.ToolCalls, - Usage: usage, - }, nil - }) + return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig) } // ChatStreamlyWithSender sends messages and streams the response via the @@ -206,7 +143,9 @@ func (l *LongCatModel) ChatStreamlyWithSender(ctx context.Context, modelName str return fmt.Errorf("failed to marshal request: %w", err) } - // SSE streams are long-lived. + ctx, cancel := context.WithTimeout(ctx, streamCallTimeout) + defer cancel() + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData)) if err != nil { return fmt.Errorf("failed to create request: %w", err) @@ -226,74 +165,7 @@ func (l *LongCatModel) ChatStreamlyWithSender(ctx context.Context, modelName str return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) } - sawTerminal := false - accumulatedToolCalls := make(map[int]map[string]any) - done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error { - if apiErr, ok := event["error"]; ok { - return fmt.Errorf("longcat: upstream stream error: %v", apiErr) - } - - tokenUsage, found, usageErr := decodeOpenAICompatibleStreamUsage(event) - if usageErr != nil { - return usageErr - } - if found { - applyStreamUsage(chatModelConfig, modelUsage, tokenUsage) - // Aggregate counts only — the full event carries content/reasoning_content. - common.Info(fmt.Sprintf("longcat: usage prompt=%d completion=%d total=%d", - tokenUsage.PromptTokens, tokenUsage.CompletionTokens, tokenUsage.TotalTokens)) - } - - choices, ok := event["choices"].([]interface{}) - if !ok || len(choices) == 0 { - return nil - } - - firstChoice, ok := choices[0].(map[string]interface{}) - if !ok { - return nil - } - - delta, ok := firstChoice["delta"].(map[string]interface{}) - if !ok { - return nil - } - - accumulateToolCallDeltas(delta, accumulatedToolCalls) - - if r, ok := delta["reasoning_content"].(string); ok && r != "" { - if err := sender(nil, &r); err != nil { - return err - } - } - - content, ok := delta["content"].(string) - if ok && content != "" { - if err := sender(&content, nil); err != nil { - return err - } - } - - finishReason, ok := firstChoice["finish_reason"].(string) - if ok && finishReason != "" { - sawTerminal = true - } - return nil - }) - if err != nil { - return fmt.Errorf("failed to scan response body: %w", err) - } - setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) - if !done && !sawTerminal { - return fmt.Errorf("longcat: stream ended before [DONE] or finish_reason") - } - - endOfStream := "[DONE]" - if err := sender(&endOfStream, nil); err != nil { - return err - } - - return nil + return HandleStreamingResponse(resp.Body, modelUsage, chatModelConfig, OpenAIParserConfig, sender) } type longCatModelInfo struct { diff --git a/internal/entity/models/nvidia.go b/internal/entity/models/nvidia.go index ab10b38de4..1d08d704a2 100644 --- a/internal/entity/models/nvidia.go +++ b/internal/entity/models/nvidia.go @@ -94,81 +94,11 @@ func (n *NvidiaModel) ChatWithMessages(ctx context.Context, modelName string, me } } - jsonData, err := json.Marshal(reqBody) + body, err := n.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout) if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) + return nil, err } - - ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout) - defer cancel() - - req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey)) - - resp, err := n.baseModel.httpClient.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to send request: %w", err) - } - defer resp.Body.Close() - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read response: %w", err) - } - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) - } - - var result map[string]interface{} - if err = json.Unmarshal(body, &result); err != nil { - return nil, fmt.Errorf("failed to parse response: %w", err) - } - - choices, ok := result["choices"].([]interface{}) - if !ok || len(choices) == 0 { - return nil, fmt.Errorf("no choices in response") - } - - firstChoice, ok := choices[0].(map[string]interface{}) - if !ok { - return nil, fmt.Errorf("invalid choice format") - } - - messageMap, ok := firstChoice["message"].(map[string]interface{}) - if !ok { - return nil, fmt.Errorf("invalid message format") - } - - content, ok := messageMap["content"].(string) - toolCalls := extractToolCalls(messageMap) - if !ok && len(toolCalls) == 0 { - return nil, fmt.Errorf("invalid content format") - } - - var reasonContent string - if chatModelConfig != nil && chatModelConfig.Thinking != nil && *chatModelConfig.Thinking { - reasonContent, ok = messageMap["reasoning_content"].(string) - if !ok { - return nil, fmt.Errorf("invalid content format") - } - if reasonContent != "" && reasonContent[0] == '\n' { - reasonContent = reasonContent[1:] - } - } - - chatResponse := &ChatResponse{ - Answer: &content, - ReasonContent: &reasonContent, - ToolCalls: toolCalls, - } - - return chatResponse, nil + return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig) } func (n *NvidiaModel) ChatStreamlyWithSender(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, modelConfig *ChatConfig, modelUsage *common.ModelUsage, sender func(*string, *string) error) error { @@ -203,6 +133,7 @@ func (n *NvidiaModel) ChatStreamlyWithSender(ctx context.Context, modelName stri } } + reqBody["stream_options"] = map[string]any{"include_usage": true} jsonData, err := json.Marshal(reqBody) if err != nil { return fmt.Errorf("failed to marshal request: %w", err) @@ -230,52 +161,7 @@ func (n *NvidiaModel) ChatStreamlyWithSender(ctx context.Context, modelName stri return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) } - accumulatedToolCalls := make(map[int]map[string]any) - if _, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error { - choices, ok := event["choices"].([]interface{}) - if !ok || len(choices) == 0 { - return nil - } - - firstChoice, ok := choices[0].(map[string]interface{}) - if !ok { - return nil - } - - delta, ok := firstChoice["delta"].(map[string]interface{}) - if !ok { - return nil - } - - accumulateToolCallDeltas(delta, accumulatedToolCalls) - - reasoningContent, ok := delta["reasoning_content"].(string) - if ok && reasoningContent != "" { - if err := sender(nil, &reasoningContent); err != nil { - return err - } - } - - content, ok := delta["content"].(string) - if ok && content != "" { - if err := sender(&content, nil); err != nil { - return err - } - } - - return nil - }); err != nil { - return fmt.Errorf("failed to scan response body: %w", err) - } - - setSortedToolCallsResult(modelConfig, accumulatedToolCalls) - - endOfStream := "[DONE]" - if err = sender(&endOfStream, nil); err != nil { - return err - } - - return nil + return HandleStreamingResponse(resp.Body, modelUsage, modelConfig, OpenAIParserConfig, sender) } type nvidiaEmbeddingResponse struct { diff --git a/internal/entity/models/siliconflow.go b/internal/entity/models/siliconflow.go index c760c7d06c..efb0ae71fb 100644 --- a/internal/entity/models/siliconflow.go +++ b/internal/entity/models/siliconflow.go @@ -55,38 +55,6 @@ func (s *SiliconflowModel) Name() string { return "SILICONFLOW" } -// SiliconflowChatResponse mirrors the response returned by SiliconFlow's -// POST /chat/completions endpoint. SiliconFlow uses the OpenAI-compatible -// schema and includes reasoning and cache token details when available. -type SiliconflowChatResponse struct { - ID string `json:"id"` - Object string `json:"object"` - Created int64 `json:"created"` - Model string `json:"model"` - Choices []struct { - FinishReason string `json:"finish_reason"` - Index int `json:"index"` - Message struct { - Content string `json:"content"` - ReasoningContent string `json:"reasoning_content"` - Role string `json:"role"` - ToolCalls []map[string]any `json:"tool_calls"` - } `json:"message"` - } `json:"choices"` - SystemFingerprint string `json:"system_fingerprint"` - Usage struct { - CompletionTokens int `json:"completion_tokens"` - PromptTokens int `json:"prompt_tokens"` - TotalTokens int `json:"total_tokens"` - CompletionTokensDetails struct { - ReasoningTokens int `json:"reasoning_tokens"` - } `json:"completion_tokens_details"` - PromptTokensDetails struct { - CachedTokens int `json:"cached_tokens"` - } `json:"prompt_tokens_details"` - } `json:"usage"` -} - // ChatWithMessages sends multiple messages with roles and returns response func (s *SiliconflowModel) ChatWithMessages(ctx context.Context, modelName string, messages []Message, apiConfig *APIConfig, chatModelConfig *ChatConfig, modelUsage *common.ModelUsage) (*ChatResponse, error) { if err := s.baseModel.APIConfigCheck(apiConfig); err != nil { @@ -115,79 +83,11 @@ func (s *SiliconflowModel) ChatWithMessages(ctx context.Context, modelName strin reqBody["enable_thinking"] = false } - jsonData, err := json.Marshal(reqBody) + body, err := s.baseModel.doRequest(ctx, url, apiConfig, reqBody, nonStreamCallTimeout) if err != nil { - return nil, fmt.Errorf("failed to marshal request: %w", err) + return nil, err } - - ctx, cancel := context.WithTimeout(ctx, nonStreamCallTimeout) - defer cancel() - - req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData)) - if err != nil { - return nil, fmt.Errorf("failed to create request: %w", err) - } - - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", *apiConfig.ApiKey)) - - resp, err := s.baseModel.httpClient.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to send request: %w", err) - } - defer resp.Body.Close() - - body, err := io.ReadAll(resp.Body) - if err != nil { - return nil, fmt.Errorf("failed to read response: %w", err) - } - - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) - } - - return parseChatCompletionResponse(body, chatModelConfig, modelUsage, func(body []byte, chatConfig *ChatConfig) (chatResponseParts, error) { - var result SiliconflowChatResponse - if err := json.Unmarshal(body, &result); err != nil { - return chatResponseParts{}, fmt.Errorf("failed to parse response: %w", err) - } - if len(result.Choices) == 0 { - return chatResponseParts{}, fmt.Errorf("no choices in response") - } - - choice := &result.Choices[0] - content := choice.Message.Content - if content == "" && len(choice.Message.ToolCalls) == 0 { - return chatResponseParts{}, fmt.Errorf("invalid content format") - } - - reasonContent := "" - if chatConfig != nil && chatConfig.Thinking != nil && *chatConfig.Thinking { - reasonContent = choice.Message.ReasoningContent - if reasonContent == "" { - reasoning, answer := GetThinkingAndAnswer(chatConfig.ModelClass, &content) - if reasoning != nil { - reasonContent = *reasoning - content = *answer - } - } - if reasonContent != "" && reasonContent[0] == '\n' { - reasonContent = reasonContent[1:] - } - } - - return chatResponseParts{ - RequestID: result.ID, - Content: &content, - ReasonContent: &reasonContent, - ToolCalls: choice.Message.ToolCalls, - Usage: &TokenUsage{ - PromptTokens: result.Usage.PromptTokens, - CompletionTokens: result.Usage.CompletionTokens, - TotalTokens: result.Usage.TotalTokens, - }, - }, nil - }) + return HandleNonStreamingResponse(body, modelUsage, chatModelConfig, OpenAIParserConfig) } // ChatStreamlyWithSender sends messages and streams response via sender function (best performance, no channel) @@ -246,72 +146,7 @@ func (s *SiliconflowModel) ChatStreamlyWithSender(ctx context.Context, modelName return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) } - // SSE parsing: read line by line - sawTerminal := false - accumulatedToolCalls := make(map[int]map[string]interface{}) - done, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error { - common.Info(fmt.Sprintf("%v", event)) - - tokenUsage, found, usageErr := decodeOpenAICompatibleStreamUsage(event) - if usageErr != nil { - return usageErr - } - if found { - applyStreamUsage(chatModelConfig, modelUsage, tokenUsage) - } - - choices, ok := event["choices"].([]interface{}) - if !ok || len(choices) == 0 { - return nil - } - - firstChoice, ok := choices[0].(map[string]interface{}) - if !ok { - return nil - } - if finishReason, ok := firstChoice["finish_reason"].(string); ok && finishReason != "" { - sawTerminal = true - } - - delta, ok := firstChoice["delta"].(map[string]interface{}) - if !ok { - return nil - } - if accumulateToolCallDeltas(delta, accumulatedToolCalls) { - return nil - } - - reasoningContent, ok := delta["reasoning_content"].(string) - if ok && reasoningContent != "" { - if err := sender(nil, &reasoningContent); err != nil { - return err - } - } - - content, ok := delta["content"].(string) - if ok && content != "" { - if err := sender(&content, nil); err != nil { - return err - } - } - - return nil - }) - if err != nil { - return fmt.Errorf("failed to scan response body: %w", err) - } - if !done && !sawTerminal { - return fmt.Errorf("siliconflow: stream ended before [DONE] or finish_reason") - } - setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) - - // Send [DONE] marker for OpenAI compatibility - endOfStream := "[DONE]" - if err = sender(&endOfStream, nil); err != nil { - return err - } - - return nil + return HandleStreamingResponse(resp.Body, modelUsage, chatModelConfig, OpenAIParserConfig, sender) } type siliconflowEmbeddingResponse struct {