diff --git a/conf/models/volcengine.json b/conf/models/volcengine.json index c260154c9..e4200ce57 100644 --- a/conf/models/volcengine.json +++ b/conf/models/volcengine.json @@ -14,7 +14,11 @@ "max_tokens": 262144, "model_types": [ "chat" - ] + ], + "thinking": { + "default_value": true, + "clear_thinking": true + } } ] } \ No newline at end of file diff --git a/internal/entity/models/volcengine.go b/internal/entity/models/volcengine.go index f203412ca..c18985436 100644 --- a/internal/entity/models/volcengine.go +++ b/internal/entity/models/volcengine.go @@ -17,9 +17,14 @@ package models import ( + "bufio" + "bytes" + "encoding/json" "fmt" "io" "net/http" + "ragflow/internal/logger" + "strings" "time" ) @@ -53,7 +58,152 @@ func (z *VolcEngine) Name() string { // Chat sends a message and returns response func (z *VolcEngine) Chat(modelName, message *string, apiConfig *APIConfig, modelConfig *ChatConfig) (*ChatResponse, error) { - return nil, fmt.Errorf("%s, no such method", z.Name()) + if message == nil { + return nil, fmt.Errorf("message is nil") + } + + var region = "default" + if apiConfig.Region != nil { + region = *apiConfig.Region + } + + url := fmt.Sprintf("%s/%s", z.BaseURL[region], z.URLSuffix.Chat) + + //Build request body + reqBody := map[string]interface{}{ + "model": modelName, + "messages": []map[string]string{ + {"role": "user", "content": *message}, + }, + "stream": false, + "temperature": 1, + } + + if modelConfig.Stream != nil { + reqBody["stream"] = *modelConfig.Stream + } + + if modelConfig.MaxTokens != nil { + reqBody["max_tokens"] = *modelConfig.MaxTokens + } + + if modelConfig.Temperature != nil { + reqBody["temperature"] = *modelConfig.Temperature + } + + if modelConfig.TopP != nil { + reqBody["top_p"] = *modelConfig.TopP + } + // TODO VolcEngine has `auto` mode + if modelConfig.Thinking != nil { + if *modelConfig.Thinking { + var thinkingFlag string + switch *modelConfig.Effort { + case "none", "minimal": + thinkingFlag = "disabled" + reqBody["reasoning_effort"] = "minimal" + break + case "low": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "low" + break + case "medium": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "medium" + break + case "auto", "default": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "medium" + break + case "high": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "high" + break + default: + return nil, fmt.Errorf("invalid effort level") + } + reqBody["thinking"] = map[string]interface{}{ + "type": thinkingFlag, + } + } else { + reqBody["thinking"] = map[string]interface{}{ + "type": "disabled", + } + } + } + + jsonData, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + + req, err := http.NewRequest("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 := z.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 != 200 { + return nil, fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) + } + + // Parse response + var result map[string]interface{} + if err = json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("failed to unmarshal response: %w", err) + } + + choices, ok := result["choices"].([]interface{}) + if !ok || len(choices) == 0 { + return nil, fmt.Errorf("no choices in responses") + } + + 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) + if !ok { + return nil, fmt.Errorf("invalid content format") + } + + var reasonContent string + if modelConfig.Thinking != nil && *modelConfig.Thinking { + reasonContent, ok = messageMap["reasoning_content"].(string) + if !ok { + return nil, fmt.Errorf("invalid reasonContent format") + } + // if first char of reasonContent is \n remove the \n + if reasonContent != "" && reasonContent[0] == '\n' { + reasonContent = reasonContent[1:] + } + } + + chatResponse := &ChatResponse{ + Answer: &content, + ReasonContent: &reasonContent, + } + + return chatResponse, nil } // ChatWithMessages sends multiple messages with roles and returns response @@ -63,7 +213,177 @@ func (z *VolcEngine) ChatWithMessages(modelName string, apiKey *string, messages // ChatStreamlyWithSender sends a message and streams response via sender function (best performance, no channel) func (z *VolcEngine) ChatStreamlyWithSender(modelName, message *string, apiConfig *APIConfig, modelConfig *ChatConfig, sender func(*string, *string) error) error { - return fmt.Errorf("%s, no such method", z.Name()) + var region = "default" + + if apiConfig.Region != nil { + region = *apiConfig.Region + } + + url := fmt.Sprintf("%s/chat/completions", z.BaseURL[region]) + + // Build request body with streaming enabled + reqBody := map[string]interface{}{ + "model": modelName, + "messages": []map[string]interface{}{ + {"role": "user", "content": *message}, + }, + "stream": true, + "temperature": 1, + } + + if modelConfig.Stream != nil { + reqBody["stream"] = *modelConfig.Stream + } + + if modelConfig.MaxTokens != nil { + reqBody["max_tokens"] = *modelConfig.MaxTokens + } + + if modelConfig.Temperature != nil { + reqBody["temperature"] = *modelConfig.Temperature + } + + if modelConfig.TopP != nil { + reqBody["top_p"] = *modelConfig.TopP + } + + if modelConfig.DoSample != nil { + reqBody["do_sample"] = *modelConfig.DoSample + } + + if modelConfig.Stop != nil { + reqBody["stop"] = *modelConfig.Stop + } + + // TODO VolcEngine has `auto` mode + if modelConfig.Thinking != nil { + if *modelConfig.Thinking { + var thinkingFlag string + switch *modelConfig.Effort { + case "none", "minimal": + thinkingFlag = "disabled" + reqBody["reasoning_effort"] = "minimal" + break + case "low": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "low" + break + case "medium": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "medium" + break + case "auto", "default": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "medium" + break + case "high": + thinkingFlag = "enabled" + reqBody["reasoning_effort"] = "high" + break + default: + return fmt.Errorf("invalid effort level") + } + reqBody["thinking"] = map[string]interface{}{ + "type": thinkingFlag, + } + } else { + reqBody["thinking"] = map[string]interface{}{ + "type": "disabled", + } + } + } + + jsonData, err := json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("failed to marshal request: %w", err) + } + + req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonData)) + if err != nil { + return 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 := z.httpClient.Do(req) + if err != nil { + return fmt.Errorf("failed to send request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) + } + + // SSE parsing: read line by line + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + line := scanner.Text() + logger.Info(line) + + // SSE data line start with data: + if !strings.HasPrefix(line, "data:") { + continue + } + + // Extract JSON after data: + data := strings.TrimSpace(line[5:]) + + // [DONE] marks the end of stream + if data == "[DONE]" { + break + } + + // Parse the JSON event + var event map[string]interface{} + if err := json.Unmarshal([]byte(data), &event); err != nil { + continue + } + + choices, ok := event["choices"].([]interface{}) + if !ok || len(choices) == 0 { + continue + } + + firstChoice, ok := choices[0].(map[string]interface{}) + if !ok { + continue + } + + delta, ok := firstChoice["delta"].(map[string]interface{}) + if !ok { + continue + } + + 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 != "" { + break + } + } + + // Send [DONE] marker for OpenAI compatibility + endOfStream := "[DONE]" + if err = sender(&endOfStream, nil); err != nil { + return err + } + + return scanner.Err() } // Encode encodes a list of texts into embeddings