From b4ca5d0bfe8b935b0f398d92f0e5afeeca8ad4d6 Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Tue, 21 Jul 2026 16:45:29 +0800 Subject: [PATCH] refactor: encapsulate the code related to tools (#17101) ### Summary As title --- internal/entity/models/baidu.go | 98 ++----------------------- internal/entity/models/base_model.go | 104 +++++++++++++++++++++++++++ internal/entity/models/deepseek.go | 79 ++------------------ internal/entity/models/minimax.go | 44 +----------- internal/entity/models/moonshot.go | 44 +----------- internal/entity/models/openai.go | 48 +------------ internal/entity/models/ragcon.go | 38 +--------- internal/entity/models/xiaomi.go | 61 +--------------- 8 files changed, 123 insertions(+), 393 deletions(-) diff --git a/internal/entity/models/baidu.go b/internal/entity/models/baidu.go index 46c60775b3..6f9da4bc63 100644 --- a/internal/entity/models/baidu.go +++ b/internal/entity/models/baidu.go @@ -25,7 +25,6 @@ import ( "io" "net/http" "ragflow/internal/common" - "sort" "strings" ) @@ -65,25 +64,10 @@ func (b *BaiduModel) ChatWithMessages(modelName string, messages []Message, apiC } url := fmt.Sprintf("%s/%s", resolvedBaseURL, b.baseModel.URLSuffix.Chat) - // Convert messages to API format - apiMessages := make([]map[string]any, len(messages)) - for i, msg := range messages { - apiMessages[i] = map[string]any{ - "role": msg.Role, - "content": msg.Content, - } - if msg.ToolCallID != "" { - apiMessages[i]["tool_call_id"] = msg.ToolCallID - } - if len(msg.ToolCalls) > 0 { - apiMessages[i]["tool_calls"] = msg.ToolCalls - } - } - // Build request body reqBody := map[string]interface{}{ "model": modelName, - "messages": apiMessages, + "messages": buildChatMessages(messages), "stream": false, "temperature": 1, } @@ -264,25 +248,10 @@ func (b *BaiduModel) ChatStreamlyWithSender(modelName string, messages []Message } url := fmt.Sprintf("%s/%s", strings.TrimSuffix(resolvedBaseURL, "/"), b.baseModel.URLSuffix.Chat) - // Convert messages to API format - apiMessages := make([]map[string]interface{}, len(messages)) - for i, msg := range messages { - apiMessages[i] = map[string]interface{}{ - "role": msg.Role, - "content": msg.Content, - } - if msg.ToolCallID != "" { - apiMessages[i]["tool_call_id"] = msg.ToolCallID - } - if len(msg.ToolCalls) > 0 { - apiMessages[i]["tool_calls"] = msg.ToolCalls - } - } - // Build request body with streaming enabled reqBody := map[string]interface{}{ "model": modelName, - "messages": apiMessages, + "messages": buildChatMessages(messages), "stream": true, } @@ -408,55 +377,7 @@ func (b *BaiduModel) ChatStreamlyWithSender(modelName string, messages []Message return nil } - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - tcMap, ok := tc.(map[string]interface{}) - if !ok { - continue - } - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - existing, hasExisting := accumulatedToolCalls[idx] - if !hasExisting { - accumulatedToolCalls[idx] = cloneMap(tcMap) - continue - } - if id, ok := tcMap["id"].(string); ok && id != "" { - if eid, ok := existing["id"].(string); ok { - existing["id"] = eid + id - } else { - existing["id"] = id - } - } - if typ, ok := tcMap["type"].(string); ok && typ != "" { - existing["type"] = typ - } - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - ef, ok := existing["function"].(map[string]interface{}) - if !ok { - ef = make(map[string]interface{}) - existing["function"] = ef - } - if name, ok := fn["name"].(string); ok && name != "" { - if en, ok := ef["name"].(string); ok { - ef["name"] = en + name - } else { - ef["name"] = name - } - } - if args, ok := fn["arguments"].(string); ok && args != "" { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } - } + accumulateToolCallDeltas(delta, accumulatedToolCalls) reasoningContent, ok := delta["reasoning_content"].(string) if ok && reasoningContent != "" { @@ -486,18 +407,7 @@ func (b *BaiduModel) ChatStreamlyWithSender(modelName string, messages []Message return fmt.Errorf("baidu: stream ended before [DONE] or finish_reason") } - if len(accumulatedToolCalls) > 0 && modelConfig != nil { - indices := make([]int, 0, len(accumulatedToolCalls)) - for idx := range accumulatedToolCalls { - indices = append(indices, idx) - } - sort.Ints(indices) - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, idx := range indices { - tcs = append(tcs, accumulatedToolCalls[idx]) - } - modelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(modelConfig, accumulatedToolCalls) // Send [DONE] marker for OpenAI compatibility endOfStream := "[DONE]" diff --git a/internal/entity/models/base_model.go b/internal/entity/models/base_model.go index 5dcde22c4d..b0cb2be191 100644 --- a/internal/entity/models/base_model.go +++ b/internal/entity/models/base_model.go @@ -24,6 +24,7 @@ import ( "fmt" "io" "net/http" + "sort" "strings" "time" ) @@ -219,3 +220,106 @@ func ReadErrorBody(r io.Reader) string { b, _ := io.ReadAll(r) return string(b) } + +// buildChatMessages converts internal messages to chat API payload items. +func buildChatMessages(messages []Message) []map[string]any { + apiMessages := make([]map[string]interface{}, len(messages)) + for i, msg := range messages { + apiMsg := map[string]interface{}{ + "role": msg.Role, + "content": msg.Content, + } + if msg.ToolCallID != "" { + apiMsg["tool_call_id"] = msg.ToolCallID + } + if len(msg.ToolCalls) > 0 { + apiMsg["tool_calls"] = msg.ToolCalls + } + apiMessages[i] = apiMsg + } + return apiMessages +} + +// setSortedToolCallsResult stores accumulated tool calls in index order. +func setSortedToolCallsResult(chatConfig *ChatConfig, accumulatedToolCalls map[int]map[string]any) { + if chatConfig == nil || len(accumulatedToolCalls) == 0 { + return + } + indices := make([]int, 0, len(accumulatedToolCalls)) + for idx := range accumulatedToolCalls { + indices = append(indices, idx) + } + sort.Ints(indices) + toolCalls := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) + for _, idx := range indices { + toolCalls = append(toolCalls, accumulatedToolCalls[idx]) + } + chatConfig.ToolCallsResult = &toolCalls +} + +// accumulateToolCallDeltas merges streaming tool-call deltas by index. +func accumulateToolCallDeltas(delta map[string]interface{}, accumulatedToolCalls map[int]map[string]any) bool { + toolCallDeltas, ok := delta["tool_calls"].([]interface{}) + if !ok { + return false + } + for _, toolCallDelta := range toolCallDeltas { + toolCall, ok := toolCallDelta.(map[string]interface{}) + if !ok { + continue + } + idxF, ok := toolCall["index"].(float64) + if !ok { + continue + } + idx := int(idxF) + existing, hasExisting := accumulatedToolCalls[idx] + if !hasExisting { + accumulatedToolCalls[idx] = cloneMap(toolCall) + continue + } + appendStringField(existing, toolCall, "id") + if typ, ok := toolCall["type"].(string); ok && typ != "" { + existing["type"] = typ + } + mergeToolCallFunction(existing, toolCall) + } + return true +} + +// appendStringField appends a non-empty string field from src into dst. +func appendStringField(dst, src map[string]interface{}, key string) { + value, ok := src[key].(string) + if !ok || value == "" { + return + } + if existing, ok := dst[key].(string); ok { + dst[key] = existing + value + } else { + dst[key] = value + } +} + +// mergeToolCallFunction merges streamed function name and arguments. +func mergeToolCallFunction(existing, delta map[string]interface{}) { + fn, ok := delta["function"].(map[string]interface{}) + if !ok { + return + } + existingFn, ok := existing["function"].(map[string]interface{}) + if !ok { + existingFn = make(map[string]interface{}) + existing["function"] = existingFn + } + appendStringField(existingFn, fn, "name") + appendStringField(existingFn, fn, "arguments") +} + +// CloneMap returns a shallow copy of m. +func cloneMap(m map[string]interface{}) map[string]interface{} { + cp := make(map[string]interface{}, len(m)) + for k, v := range m { + cp[k] = v + } + return cp +} diff --git a/internal/entity/models/deepseek.go b/internal/entity/models/deepseek.go index 21bfaa2b10..7d54adcefa 100644 --- a/internal/entity/models/deepseek.go +++ b/internal/entity/models/deepseek.go @@ -24,7 +24,6 @@ import ( "io" "net/http" "ragflow/internal/common" - "sort" "strconv" "strings" ) @@ -68,26 +67,10 @@ func (d *DeepSeekModel) ChatWithMessages(modelName string, messages []Message, a } url := fmt.Sprintf("%s/%s", resolvedBaseURL, d.baseModel.URLSuffix.Chat) - // Convert messages to the format expected by API - apiMessages := make([]map[string]interface{}, len(messages)) - for i, msg := range messages { - apiMsg := map[string]interface{}{ - "role": msg.Role, - "content": msg.Content, - } - if msg.ToolCallID != "" { - apiMsg["tool_call_id"] = msg.ToolCallID - } - if len(msg.ToolCalls) > 0 { - apiMsg["tool_calls"] = msg.ToolCalls - } - apiMessages[i] = apiMsg - } - // Build request body reqBody := map[string]interface{}{ "model": modelName, - "messages": apiMessages, + "messages": buildChatMessages(messages), "stream": false, "temperature": 1, } @@ -266,26 +249,10 @@ func (d *DeepSeekModel) ChatStreamlyWithSender(modelName string, messages []Mess } url := fmt.Sprintf("%s/chat/completions", resolvedBaseURL) - // Convert messages to API format - apiMessages := make([]map[string]interface{}, len(messages)) - for i, msg := range messages { - apiMsg := map[string]interface{}{ - "role": msg.Role, - "content": msg.Content, - } - if msg.ToolCallID != "" { - apiMsg["tool_call_id"] = msg.ToolCallID - } - if len(msg.ToolCalls) > 0 { - apiMsg["tool_calls"] = msg.ToolCalls - } - apiMessages[i] = apiMsg - } - // Build request body with streaming enabled reqBody := map[string]interface{}{ "model": modelName, - "messages": apiMessages, + "messages": buildChatMessages(messages), "stream": true, "temperature": 1, } @@ -411,34 +378,7 @@ func (d *DeepSeekModel) ChatStreamlyWithSender(modelName string, messages []Mess return nil } - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - tcMap, ok := tc.(map[string]interface{}) - if !ok { - continue - } - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - existing, hasExisting := accumulatedToolCalls[idx] - if hasExisting { - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - if args, ok := fn["arguments"].(string); ok { - if ef, ok := existing["function"].(map[string]interface{}); ok { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } - } else { - accumulatedToolCalls[idx] = cloneMap(tcMap) - } - } + if accumulateToolCallDeltas(delta, accumulatedToolCalls) { return nil } @@ -469,18 +409,7 @@ func (d *DeepSeekModel) ChatStreamlyWithSender(modelName string, messages []Mess return fmt.Errorf("deepseek: stream ended before [DONE] or finish_reason") } - if len(accumulatedToolCalls) > 0 && chatModelConfig != nil { - indices := make([]int, 0, len(accumulatedToolCalls)) - for idx := range accumulatedToolCalls { - indices = append(indices, idx) - } - sort.Ints(indices) - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, idx := range indices { - tcs = append(tcs, accumulatedToolCalls[idx]) - } - chatModelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) // Send [DONE] marker for OpenAI compatibility endOfStream := "[DONE]" diff --git a/internal/entity/models/minimax.go b/internal/entity/models/minimax.go index 20d2842f2b..e2f8db1e80 100644 --- a/internal/entity/models/minimax.go +++ b/internal/entity/models/minimax.go @@ -25,7 +25,6 @@ import ( "io" "net/http" "ragflow/internal/common" - "sort" "strings" ) @@ -355,35 +354,7 @@ func (m *MinimaxModel) ChatStreamlyWithSender(modelName string, messages []Messa return nil } - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - tcMap, ok := tc.(map[string]interface{}) - if !ok { - continue - } - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - existing, hasExisting := accumulatedToolCalls[idx] - if hasExisting { - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - if args, ok := fn["arguments"].(string); ok { - if ef, ok := existing["function"].(map[string]interface{}); ok { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } - } else { - accumulatedToolCalls[idx] = cloneMap(tcMap) - } - } - } + accumulateToolCallDeltas(delta, accumulatedToolCalls) content, ok := delta["content"].(string) if ok && content != "" { @@ -413,18 +384,7 @@ func (m *MinimaxModel) ChatStreamlyWithSender(modelName string, messages []Messa return fmt.Errorf("minimax: stream ended before [DONE] or finish_reason") } - if len(accumulatedToolCalls) > 0 && modelConfig != nil { - indices := make([]int, 0, len(accumulatedToolCalls)) - for idx := range accumulatedToolCalls { - indices = append(indices, idx) - } - sort.Ints(indices) - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, idx := range indices { - tcs = append(tcs, accumulatedToolCalls[idx]) - } - modelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(modelConfig, accumulatedToolCalls) // Send [DONE] marker for OpenAI compatibility endOfStream := "[DONE]" diff --git a/internal/entity/models/moonshot.go b/internal/entity/models/moonshot.go index 84567d1812..3e789d1a0c 100644 --- a/internal/entity/models/moonshot.go +++ b/internal/entity/models/moonshot.go @@ -24,7 +24,6 @@ import ( "io" "net/http" "ragflow/internal/common" - "sort" "strings" ) @@ -366,35 +365,7 @@ func (m *MoonshotModel) ChatStreamlyWithSender(modelName string, messages []Mess return nil } - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - tcMap, ok := tc.(map[string]interface{}) - if !ok { - continue - } - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - existing, hasExisting := accumulatedToolCalls[idx] - if hasExisting { - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - if args, ok := fn["arguments"].(string); ok { - if ef, ok := existing["function"].(map[string]interface{}); ok { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } - } else { - accumulatedToolCalls[idx] = cloneMap(tcMap) - } - } - } + accumulateToolCallDeltas(delta, accumulatedToolCalls) reasoningContent, ok := delta["reasoning_content"].(string) if ok && reasoningContent != "" { @@ -424,18 +395,7 @@ func (m *MoonshotModel) ChatStreamlyWithSender(modelName string, messages []Mess return fmt.Errorf("moonshot: stream ended before [DONE] or finish_reason") } - if len(accumulatedToolCalls) > 0 && chatModelConfig != nil { - indices := make([]int, 0, len(accumulatedToolCalls)) - for idx := range accumulatedToolCalls { - indices = append(indices, idx) - } - sort.Ints(indices) - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, idx := range indices { - tcs = append(tcs, accumulatedToolCalls[idx]) - } - chatModelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) // Send [DONE] marker for OpenAI compatibility endOfStream := "[DONE]" diff --git a/internal/entity/models/openai.go b/internal/entity/models/openai.go index 9f17895be0..22f47f8d91 100644 --- a/internal/entity/models/openai.go +++ b/internal/entity/models/openai.go @@ -393,36 +393,7 @@ func (o *OpenAIModel) ChatStreamlyWithSender(modelName string, messages []Messag continue } - // Accumulate streaming tool_call deltas (mirrors Python's - // async_chat_streamly_with_tools in rag/llm/chat_model.py:500-509). - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - if tcMap, ok := tc.(map[string]interface{}); ok { - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - existing, hasExisting := accumulatedToolCalls[idx] - if hasExisting { - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - if args, ok := fn["arguments"].(string); ok { - if ef, ok := existing["function"].(map[string]interface{}); ok { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } - } else { - accumulatedToolCalls[idx] = cloneMap(tcMap) - } - } - } - continue // tool_call deltas don't carry content - } + accumulateToolCallDeltas(delta, accumulatedToolCalls) reasoningContent, ok := delta["reasoning_content"].(string) if ok && reasoningContent != "" { @@ -450,14 +421,7 @@ func (o *OpenAIModel) ChatStreamlyWithSender(modelName string, messages []Messag return fmt.Errorf("openai: stream ended before [DONE] or finish_reason") } - // Populate ToolCallsResult with accumulated streaming tool_calls. - if len(accumulatedToolCalls) > 0 && chatModelConfig != nil { - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, tc := range accumulatedToolCalls { - tcs = append(tcs, tc) - } - chatModelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) // Populate UsageResult with the authoritative usage from the stream. if streamUsage != nil && chatModelConfig != nil { @@ -1110,11 +1074,3 @@ func extractUsageFromMap(raw map[string]interface{}) (int, int, int) { } return pt, ct, tt } - -func cloneMap(m map[string]interface{}) map[string]interface{} { - cp := make(map[string]interface{}, len(m)) - for k, v := range m { - cp[k] = v - } - return cp -} diff --git a/internal/entity/models/ragcon.go b/internal/entity/models/ragcon.go index 92e1f5c60c..be1e6cdfeb 100644 --- a/internal/entity/models/ragcon.go +++ b/internal/entity/models/ragcon.go @@ -310,35 +310,7 @@ func (r *RAGconModel) ChatStreamlyWithSender(modelName string, messages []Messag continue } - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - tcMap, ok := tc.(map[string]interface{}) - if !ok { - continue - } - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - if existing, hasExisting := accumulatedToolCalls[idx]; hasExisting { - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - if args, ok := fn["arguments"].(string); ok { - if ef, ok := existing["function"].(map[string]interface{}); ok { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } - } else { - accumulatedToolCalls[idx] = cloneMap(tcMap) - } - } - continue - } + accumulateToolCallDeltas(delta, accumulatedToolCalls) if reasoningContent, ok := delta["reasoning_content"].(string); ok && reasoningContent != "" { if err := sender(nil, &reasoningContent); err != nil { @@ -361,13 +333,7 @@ func (r *RAGconModel) ChatStreamlyWithSender(modelName string, messages []Messag return fmt.Errorf("ragcon: stream ended before [DONE] or finish_reason") } - if len(accumulatedToolCalls) > 0 && chatModelConfig != nil { - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, tc := range accumulatedToolCalls { - tcs = append(tcs, tc) - } - chatModelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(chatModelConfig, accumulatedToolCalls) if streamUsage != nil && chatModelConfig != nil { chatModelConfig.UsageResult = streamUsage } diff --git a/internal/entity/models/xiaomi.go b/internal/entity/models/xiaomi.go index 867fe985d0..e522952983 100644 --- a/internal/entity/models/xiaomi.go +++ b/internal/entity/models/xiaomi.go @@ -28,7 +28,6 @@ import ( "os" "path/filepath" "ragflow/internal/common" - "sort" "strings" ) @@ -356,51 +355,8 @@ func (x *XiaomiModel) ChatStreamlyWithSender(modelName string, messages []Messag return nil } - if tcs, ok := delta["tool_calls"].([]interface{}); ok { - for _, tc := range tcs { - tcMap, ok := tc.(map[string]interface{}) - if !ok { - continue - } - idxF, ok := tcMap["index"].(float64) - if !ok { - continue - } - idx := int(idxF) - existing, hasExisting := accumulatedToolCalls[idx] - if hasExisting { - if id, ok := tcMap["id"].(string); ok && id != "" { - existing["id"] = id - } - if typ, ok := tcMap["type"].(string); ok && typ != "" { - existing["type"] = typ - } - if fn, ok := tcMap["function"].(map[string]interface{}); ok { - ef, ok := existing["function"].(map[string]interface{}) - if !ok { - ef = make(map[string]interface{}) - existing["function"] = ef - } - if name, ok := fn["name"].(string); ok && name != "" { - if en, ok := ef["name"].(string); ok { - ef["name"] = en + name - } else { - ef["name"] = name - } - } - if args, ok := fn["arguments"].(string); ok { - if ea, ok := ef["arguments"].(string); ok { - ef["arguments"] = ea + args - } else { - ef["arguments"] = args - } - } - } - } else { - accumulatedToolCalls[idx] = cloneMap(tcMap) - } - } - } + accumulateToolCallDeltas(delta, accumulatedToolCalls) + reasoningContent, ok := delta["reasoning_content"].(string) if ok && reasoningContent != "" { if err = sender(nil, &reasoningContent); err != nil { @@ -420,18 +376,7 @@ func (x *XiaomiModel) ChatStreamlyWithSender(modelName string, messages []Messag return fmt.Errorf("failed to scan response body: %w", err) } - if len(accumulatedToolCalls) > 0 && modelConfig != nil { - indices := make([]int, 0, len(accumulatedToolCalls)) - for idx := range accumulatedToolCalls { - indices = append(indices, idx) - } - sort.Ints(indices) - tcs := make([]map[string]interface{}, 0, len(accumulatedToolCalls)) - for _, idx := range indices { - tcs = append(tcs, accumulatedToolCalls[idx]) - } - modelConfig.ToolCallsResult = &tcs - } + setSortedToolCallsResult(modelConfig, accumulatedToolCalls) // Send [DONE] marker for OpenAI compatibility endOfStream := "[DONE]"