From e87008e26cef68f27782e277dc46b61e56dd066a Mon Sep 17 00:00:00 2001 From: Hz_ Date: Mon, 20 Jul 2026 15:54:55 +0800 Subject: [PATCH] fix(go-agent): support Aliyun tool calls (#17099) ## Summary - Enable synchronous and streaming tool calls for Aliyun models. - Preserve provider-specific chat endpoints and prevent repeated qwen-flash tool calls. - Restrict retrieval tool inputs to model-provided query parameters. ## Testing - `bash build.sh --test ./internal/entity/models ./internal/agent/component ./internal/agent/tool` - Manual frontend UI testing passed. --- conf/models/aliyun.json | 5 +- internal/agent/component/agent.go | 9 +- internal/agent/component/llm.go | 83 ++--- internal/agent/component/model_driver_test.go | 47 +++ internal/agent/tool/retrieval.go | 35 --- internal/agent/tool/retrieval_test.go | 5 + internal/entity/models/aliyun.go | 215 ++++++++++--- internal/entity/models/aliyun_test.go | 296 ++++++++++++++++++ internal/service/agent.go | 2 +- internal/service/agent_run_e2e_test.go | 13 +- 10 files changed, 563 insertions(+), 147 deletions(-) create mode 100644 internal/agent/component/model_driver_test.go create mode 100644 internal/entity/models/aliyun_test.go diff --git a/conf/models/aliyun.json b/conf/models/aliyun.json index d62f12eae7..7511f28c38 100644 --- a/conf/models/aliyun.json +++ b/conf/models/aliyun.json @@ -18,7 +18,10 @@ "max_tokens": 995904, "model_types": [ "chat" - ] + ], + "tools": { + "support": true + } }, { "name": "text-embedding-v4", diff --git a/internal/agent/component/agent.go b/internal/agent/component/agent.go index 86f7a65c7f..233b63674f 100644 --- a/internal/agent/component/agent.go +++ b/internal/agent/component/agent.go @@ -727,14 +727,7 @@ func buildAgentChatModel(ctx context.Context, p AgentParam) (*models.EinoChatMod if driver == "" { driver = "dummy" } - baseURL := baseURLMapForDriver(driver, p.BaseURL) - // urlSuffix: see chatURLSuffixFor in llm.go for the rationale. - // The factory's NewModelDriver stores URLSuffix verbatim; the - // driver then appends URLSuffix.Chat to baseURL to build the - // chat-completions endpoint, so an empty suffix leaves the URL - // pointing at the v1 root (404). Seed the right suffix per - // driver so the agent's ReAct loop hits a working endpoint. - d, err := models.NewModelFactory().CreateModelDriver(driver, baseURL, chatURLSuffixFor(driver)) + d, err := newChatModelDriver(driver, p.BaseURL) if err != nil { return nil, fmt.Errorf("resolve driver %q: %w", driver, err) } diff --git a/internal/agent/component/llm.go b/internal/agent/component/llm.go index e596c631ee..414b2db37d 100644 --- a/internal/agent/component/llm.go +++ b/internal/agent/component/llm.go @@ -63,9 +63,9 @@ type LLMParam struct { // same line verbatim. FrequencyPenalty *float64 - // Driver is the provider driver to use (e.g. "openai", "dummy"). When - // empty, the default ChatInvoker will look up a driver from ModelID - // (e.g. by attempting NewDummyModel for unknown providers). + // Driver is the configured provider driver to use (e.g. "openai"). When + // empty, the default ChatInvoker derives it from ModelID or uses the explicit + // test/development-only dummy driver. Driver string // APIKey overrides the default empty key. Tests may set this; prod @@ -219,23 +219,10 @@ func (e *einoChatInvoker) Invoke(ctx context.Context, req ChatInvokeRequest) (*C if driver == "" { driver = "dummy" } - baseURL := baseURLMapForDriver(driver, req.BaseURL) - // urlSuffix: each driver appends URLSuffix.Chat to baseURL to form - // the chat-completions endpoint (e.g. "chat/completions" for - // openai-compatible drivers, "v1/messages" for anthropic). The - // factory's NewModelDriver accepts a zero URLSuffix and stores it - // as-is; the openai driver then builds `/` (with no path), - // which is the wrong endpoint for a v1-root base URL. We seed - // the right suffix per driver here so the factory and the - // openai driver's URL construction agree. - urlSuffix := chatURLSuffixFor(driver) - d, err := models.NewModelFactory().CreateModelDriver(driver, baseURL, urlSuffix) + d, err := newChatModelDriver(driver, req.BaseURL) if err != nil { return nil, fmt.Errorf("component: LLM: resolve driver %q: %w", driver, err) } - if d == nil { - return nil, fmt.Errorf("component: LLM: no driver for %q", driver) - } apiKey := req.APIKey cfg := &models.APIConfig{ApiKey: &apiKey} cm := models.NewChatModel(d, &modelName, cfg) @@ -309,42 +296,38 @@ func toEinoMessages(msgs []schema.Message) []*schema.Message { return out } -// chatURLSuffixFor returns the URLSuffix the factory should pass to -// the driver for the chat endpoint. Each driver's ChatWithMessages -// builds `baseURL/URLSuffix.Chat`, so the suffix has to match the -// provider's actual chat path. We seed the common ones here; for any -// driver the factory has no entry for, we fall through to a default -// "chat/completions" path (the openai-compatible default), which -// matches the dummy driver and any third-party openai-compatible -// gateway. -func chatURLSuffixFor(driver string) models.URLSuffix { - switch strings.ToLower(driver) { - case "anthropic": - return models.URLSuffix{Chat: "v1/messages"} - case "ollama": - return models.URLSuffix{Chat: "api/chat"} - default: - return models.URLSuffix{Chat: "chat/completions"} - } -} - -func baseURLMapForDriver(driver, override string) map[string]string { - if override != "" { - return map[string]string{"default": override} - } +// newChatModelDriver returns the provider-configured driver used by regular +// chat. Provider-specific endpoint suffixes remain owned by conf/models/*.json; +// a tenant base_url override replaces only the endpoint root. +func newChatModelDriver(driver, override string) (models.ModelDriver, error) { pm := models.GetProviderManager() - if pm == nil { - return nil + if pm != nil { + provider := pm.FindProvider(driver) + if provider != nil && provider.ModelDriver != nil { + modelDriver := provider.ModelDriver + if strings.TrimSpace(override) != "" { + modelDriver = modelDriver.NewInstance( + map[string]string{ + "default": strings.TrimRight(override, "/"), + }, + ) + if modelDriver == nil { + return nil, fmt.Errorf("provider does not support a custom base_url") + } + } + return modelDriver, nil + } } - provider := pm.FindProvider(driver) - if provider == nil || len(provider.URL) == 0 { - return nil + + // Dummy is an explicit test/development driver and has no provider config. + if strings.EqualFold(driver, "dummy") { + baseURL := map[string]string(nil) + if strings.TrimSpace(override) != "" { + baseURL = map[string]string{"default": strings.TrimRight(override, "/")} + } + return models.NewDummyModel(baseURL, models.URLSuffix{Chat: "chat/completions"}), nil } - baseURL := make(map[string]string, len(provider.URL)) - for region, url := range provider.URL { - baseURL[region] = url - } - return baseURL + return nil, fmt.Errorf("provider is not configured") } // NewLLMComponent builds an LLMComponent from raw params. diff --git a/internal/agent/component/model_driver_test.go b/internal/agent/component/model_driver_test.go new file mode 100644 index 0000000000..84fc1b288b --- /dev/null +++ b/internal/agent/component/model_driver_test.go @@ -0,0 +1,47 @@ +package component + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/cloudwego/eino/schema" + + "ragflow/internal/entity/models" +) + +func TestNewChatModelDriverPreservesProviderChatSuffix(t *testing.T) { + if err := models.InitProviderManager("../../../conf/models"); err != nil { + t.Fatalf("InitProviderManager: %v", err) + } + + requestPath := make(chan string, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestPath <- r.URL.Path + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"choices":[{"message":{"content":"ok"}}]}`)) + })) + defer server.Close() + + driver, err := newChatModelDriver("Tongyi-Qianwen", server.URL) + if err != nil { + t.Fatalf("newChatModelDriver: %v", err) + } + apiKey := "test-key" + modelName := "qwen-flash" + chatModel := models.NewEinoChatModel( + models.NewChatModel(driver, &modelName, &models.APIConfig{ApiKey: &apiKey}), + nil, + ) + response, err := chatModel.Generate(context.Background(), []*schema.Message{schema.UserMessage("hi")}) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if response.Content != "ok" { + t.Errorf("content = %q, want ok", response.Content) + } + if got := <-requestPath; got != "/compatible-mode/v1/chat/completions" { + t.Errorf("request path = %q, want /compatible-mode/v1/chat/completions", got) + } +} diff --git a/internal/agent/tool/retrieval.go b/internal/agent/tool/retrieval.go index 03c097e125..9f38959836 100644 --- a/internal/agent/tool/retrieval.go +++ b/internal/agent/tool/retrieval.go @@ -121,41 +121,6 @@ func (r *RetrievalTool) Info(_ context.Context) (*schema.ToolInfo, error) { Desc: "The keywords to search the dataset. The keywords should be the most important words/terms (including synonyms) from the original request.", Required: true, }, - "dataset_ids": { - Type: schema.Array, - Desc: "Optional list of dataset IDs to restrict the search to.", - Required: false, - }, - "kb_ids": { - Type: schema.Array, - Desc: "Optional list of knowledge base IDs to restrict the search to.", - Required: false, - }, - "top_n": { - Type: schema.Integer, - Desc: "Number of top chunks to return. Defaults to 8 if omitted.", - Required: false, - }, - "top_k": { - Type: schema.Integer, - Desc: "Maximum candidate chunks retrieved before final top_n trimming.", - Required: false, - }, - "keywords_similarity_weight": { - Type: schema.Number, - Desc: "Keyword similarity weight in [0,1]; vector similarity weight is 1 - this value.", - Required: false, - }, - "use_kg": { - Type: schema.Boolean, - Desc: "GraphRAG toggle. Not supported in Go Canvas (plan ); must be false.", - Required: false, - }, - "similarity_threshold": { - Type: schema.Number, - Desc: "Minimum similarity threshold for dataset retrieval.", - Required: false, - }, }), }, nil } diff --git a/internal/agent/tool/retrieval_test.go b/internal/agent/tool/retrieval_test.go index 97aa4b1926..16072a5c4f 100644 --- a/internal/agent/tool/retrieval_test.go +++ b/internal/agent/tool/retrieval_test.go @@ -93,6 +93,11 @@ func TestRetrieval_InfoMatchesPythonMeta(t *testing.T) { if !strings.Contains(string(raw), `"query"`) { t.Errorf("schema JSON does not contain 'query' key: %s", raw) } + for _, nodeConfig := range []string{"dataset_ids", "kb_ids", "top_n", "top_k", "similarity_threshold", "keywords_similarity_weight", "use_kg"} { + if strings.Contains(string(raw), `"`+nodeConfig+`"`) { + t.Errorf("schema JSON exposes Canvas node config %q to the model: %s", nodeConfig, raw) + } + } } func TestRetrieval_EmptyArgsIsHandled(t *testing.T) { diff --git a/internal/entity/models/aliyun.go b/internal/entity/models/aliyun.go index 0116dc2ca4..74822ef71d 100644 --- a/internal/entity/models/aliyun.go +++ b/internal/entity/models/aliyun.go @@ -23,8 +23,10 @@ import ( "fmt" "io" "net/http" - "ragflow/internal/common" + "sort" "strings" + + "ragflow/internal/common" ) // AliyunModel implements ModelDriver for Aliyun @@ -68,14 +70,7 @@ func (a *AliyunModel) ChatWithMessages(modelName string, messages []Message, api url := fmt.Sprintf("%s/%s", strings.TrimSuffix(baseURL, "/"), a.baseModel.URLSuffix.Chat) - // Convert messages to the format expected by API - apiMessages := make([]map[string]interface{}, len(messages)) - for i, msg := range messages { - apiMessages[i] = map[string]interface{}{ - "role": msg.Role, - "content": msg.Content, - } - } + apiMessages := aliyunChatMessages(messages) // Build request body reqBody := map[string]interface{}{ @@ -113,6 +108,11 @@ func (a *AliyunModel) ChatWithMessages(modelName string, messages []Message, api reqBody["enable_thinking"] = false } } + + if chatModelConfig.Tools != nil { + reqBody["tools"] = chatModelConfig.Tools + reqBody["tool_choice"] = aliyunToolChoice(modelName, messages, chatModelConfig.ToolChoice) + } } jsonData, err := json.Marshal(reqBody) @@ -167,9 +167,10 @@ func (a *AliyunModel) ChatWithMessages(modelName string, messages []Message, api return nil, fmt.Errorf("invalid message format") } - answer, ok := messageMap["content"].(string) - if !ok { - return nil, fmt.Errorf("invalid content format") + answer, hasAnswer := messageMap["content"].(string) + toolCalls := aliyunToolCalls(messageMap) + if !hasAnswer && len(toolCalls) == 0 { + return nil, fmt.Errorf("response contains neither content nor tool calls") } var reasonContent string @@ -187,6 +188,7 @@ func (a *AliyunModel) ChatWithMessages(modelName string, messages []Message, api chatResponse := &ChatResponse{ Answer: &answer, ReasonContent: &reasonContent, + ToolCalls: toolCalls, } return chatResponse, nil @@ -210,14 +212,7 @@ func (a *AliyunModel) ChatStreamlyWithSender(modelName string, messages []Messag url := fmt.Sprintf("%s/%s", strings.TrimSuffix(baseURL, "/"), a.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, - } - } + apiMessages := aliyunChatMessages(messages) // Build request body with streaming enabled reqBody := map[string]interface{}{ @@ -227,35 +222,39 @@ func (a *AliyunModel) ChatStreamlyWithSender(modelName string, messages []Messag "temperature": 1, } - if chatModelConfig.Stream != nil { - reqBody["stream"] = *chatModelConfig.Stream - } + if chatModelConfig != nil { + if chatModelConfig.Stream != nil && !*chatModelConfig.Stream { + return fmt.Errorf("stream must be true in ChatStreamlyWithSender") + } + chatModelConfig.ToolCallsResult = nil - if chatModelConfig.MaxTokens != nil { - reqBody["max_tokens"] = *chatModelConfig.MaxTokens - } + if chatModelConfig.MaxTokens != nil { + reqBody["max_tokens"] = *chatModelConfig.MaxTokens + } - if chatModelConfig.Temperature != nil { - reqBody["temperature"] = *chatModelConfig.Temperature - } + if chatModelConfig.Temperature != nil { + reqBody["temperature"] = *chatModelConfig.Temperature + } - if chatModelConfig.DoSample != nil { - reqBody["do_sample"] = *chatModelConfig.DoSample - } + if chatModelConfig.DoSample != nil { + reqBody["do_sample"] = *chatModelConfig.DoSample + } - if chatModelConfig.TopP != nil { - reqBody["top_p"] = *chatModelConfig.TopP - } + if chatModelConfig.TopP != nil { + reqBody["top_p"] = *chatModelConfig.TopP + } - if chatModelConfig.Stop != nil { - reqBody["stop"] = *chatModelConfig.Stop - } + if chatModelConfig.Stop != nil { + reqBody["stop"] = *chatModelConfig.Stop + } - if chatModelConfig.Thinking != nil { - if *chatModelConfig.Thinking { - reqBody["enable_thinking"] = true - } else { - reqBody["enable_thinking"] = false + if chatModelConfig.Thinking != nil { + reqBody["enable_thinking"] = *chatModelConfig.Thinking + } + + if chatModelConfig.Tools != nil { + reqBody["tools"] = chatModelConfig.Tools + reqBody["tool_choice"] = aliyunToolChoice(modelName, messages, chatModelConfig.ToolChoice) } } @@ -286,8 +285,9 @@ func (a *AliyunModel) ChatStreamlyWithSender(modelName string, messages []Messag return fmt.Errorf("API request failed with status %d: %s", resp.StatusCode, string(body)) } - // SSE parsing: read line by line - if _, err := ParseSSEStream[map[string]interface{}](resp.Body, func(event map[string]interface{}) error { + 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)) choices, ok := event["choices"].([]interface{}) @@ -299,12 +299,26 @@ func (a *AliyunModel) ChatStreamlyWithSender(modelName string, messages []Messag 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 toolCalls, ok := delta["tool_calls"].([]interface{}); ok { + for _, rawToolCall := range toolCalls { + toolCall, ok := rawToolCall.(map[string]interface{}) + if !ok { + continue + } + mergeAliyunToolCallDelta(accumulatedToolCalls, toolCall) + } + return nil + } + content, ok := delta["content"].(string) if ok && content != "" { if err := sender(&content, nil); err != nil { @@ -320,15 +334,124 @@ func (a *AliyunModel) ChatStreamlyWithSender(modelName string, messages []Messag } return nil - }); err != nil { + }) + if err != nil { return fmt.Errorf("failed to scan response body: %w", err) } + if !done && !sawTerminal { + return fmt.Errorf("aliyun: stream ended before [DONE] or finish_reason") + } + + if len(accumulatedToolCalls) > 0 && chatModelConfig != nil { + indices := make([]int, 0, len(accumulatedToolCalls)) + for index := range accumulatedToolCalls { + indices = append(indices, index) + } + sort.Ints(indices) + toolCalls := make([]map[string]interface{}, 0, len(indices)) + for _, index := range indices { + toolCalls = append(toolCalls, accumulatedToolCalls[index]) + } + chatModelConfig.ToolCallsResult = &toolCalls + } // Send [DONE] marker for OpenAI compatibility endOfStream := "[DONE]" return sender(&endOfStream, nil) } +func mergeAliyunToolCallDelta(accumulated map[int]map[string]interface{}, delta map[string]interface{}) { + indexValue, ok := delta["index"].(float64) + if !ok { + return + } + index := int(indexValue) + toolCall, exists := accumulated[index] + if !exists { + toolCall = map[string]interface{}{"index": indexValue} + accumulated[index] = toolCall + } + + for _, field := range []string{"id", "type"} { + if value, ok := delta[field].(string); ok && value != "" { + toolCall[field] = value + } + } + + functionDelta, ok := delta["function"].(map[string]interface{}) + if !ok { + return + } + function, ok := toolCall["function"].(map[string]interface{}) + if !ok { + function = make(map[string]interface{}) + toolCall["function"] = function + } + if name, ok := functionDelta["name"].(string); ok && name != "" { + function["name"] = name + } + if arguments, ok := functionDelta["arguments"].(string); ok { + currentArguments, _ := function["arguments"].(string) + function["arguments"] = currentArguments + arguments + } +} + +func aliyunChatMessages(messages []Message) []map[string]interface{} { + apiMessages := make([]map[string]interface{}, len(messages)) + for i, msg := range messages { + apiMessage := map[string]interface{}{ + "role": msg.Role, + "content": msg.Content, + } + if msg.ToolCallID != "" { + apiMessage["tool_call_id"] = msg.ToolCallID + } + if len(msg.ToolCalls) > 0 { + apiMessage["tool_calls"] = msg.ToolCalls + } + apiMessages[i] = apiMessage + } + return apiMessages +} + +// aliyunToolChoice prevents qwen-flash from repeatedly issuing another tool call +// after a tool result has already been supplied. With "auto", qwen-flash can +// keep emitting tool_calls even for a successful result until the ReAct graph +// exhausts its step limit. Other models, initial calls, and explicit choices +// retain their configured behavior. +func aliyunToolChoice(modelName string, messages []Message, configured *string) string { + choice := "auto" + if configured != nil && strings.TrimSpace(*configured) != "" { + choice = *configured + } + if !strings.EqualFold(strings.TrimSpace(choice), "auto") { + return choice + } + if !strings.HasPrefix(strings.ToLower(strings.TrimSpace(modelName)), "qwen-flash") { + return choice + } + for _, message := range messages { + if strings.EqualFold(message.Role, "tool") && message.ToolCallID != "" { + return "none" + } + } + return choice +} + +func aliyunToolCalls(message map[string]interface{}) []map[string]interface{} { + rawToolCalls, ok := message["tool_calls"].([]interface{}) + if !ok { + return nil + } + toolCalls := make([]map[string]interface{}, 0, len(rawToolCalls)) + for _, rawToolCall := range rawToolCalls { + if toolCall, ok := rawToolCall.(map[string]interface{}); ok { + toolCalls = append(toolCalls, toolCall) + } + } + return toolCalls +} + type aliyunEmbeddingResponse struct { Data []EmbeddingData `json:"data"` Model string `json:"model"` diff --git a/internal/entity/models/aliyun_test.go b/internal/entity/models/aliyun_test.go new file mode 100644 index 0000000000..dad23a9252 --- /dev/null +++ b/internal/entity/models/aliyun_test.go @@ -0,0 +1,296 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package models + +import ( + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "testing" +) + +func TestAliyunChatWithMessagesSupportsToolCalls(t *testing.T) { + requestBody := make(chan map[string]interface{}, 1) + requestPath := make(chan string, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]interface{} + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + requestPath <- r.URL.Path + requestBody <- body + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"choices":[{"message":{"content":null,"tool_calls":[{"id":"call-1","type":"function","function":{"name":"retrieval","arguments":"{\"query\":\"ragflow\"}"}}]}}]}`)) + })) + defer server.Close() + + model := NewAliyunModel( + map[string]string{"default": server.URL}, + URLSuffix{Chat: "compatible-mode/v1/chat/completions"}, + ) + apiKey := "test-key" + toolChoice := "auto" + tools := []map[string]interface{}{{ + "type": "function", + "function": map[string]interface{}{ + "name": "retrieval", + "parameters": map[string]interface{}{"type": "object"}, + }, + }} + messages := []Message{{Role: "user", Content: "find ragflow"}} + + response, err := model.ChatWithMessages( + "qwen-flash", + messages, + &APIConfig{ApiKey: &apiKey}, + &ChatConfig{Tools: tools, ToolChoice: &toolChoice}, + nil, + ) + if err != nil { + t.Fatalf("ChatWithMessages: %v", err) + } + if len(response.ToolCalls) != 1 { + t.Fatalf("tool calls = %d, want 1", len(response.ToolCalls)) + } + if response.ToolCalls[0]["id"] != "call-1" { + t.Errorf("tool call id = %v, want call-1", response.ToolCalls[0]["id"]) + } + + if got := <-requestPath; got != "/compatible-mode/v1/chat/completions" { + t.Errorf("request path = %q, want /compatible-mode/v1/chat/completions", got) + } + body := <-requestBody + if body["tool_choice"] != "auto" { + t.Errorf("tool_choice = %v, want auto for initial qwen-flash call", body["tool_choice"]) + } + if _, ok := body["tools"].([]interface{}); !ok { + t.Fatalf("tools = %T, want JSON array", body["tools"]) + } +} + +func TestAliyunChatWithMessagesStopsQwenFlashAfterToolResult(t *testing.T) { + requestBody := make(chan map[string]interface{}, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]interface{} + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + requestBody <- body + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"final answer"},"finish_reason":"stop"}]}`)) + })) + defer server.Close() + + model := NewAliyunModel( + map[string]string{"default": server.URL}, + URLSuffix{Chat: "compatible-mode/v1/chat/completions"}, + ) + apiKey := "test-key" + auto := "auto" + messages := []Message{ + {Role: "user", Content: "find ragflow"}, + { + Role: "assistant", + Content: "", + ToolCalls: []map[string]interface{}{{ + "id": "previous-call", + "type": "function", + "function": map[string]interface{}{ + "name": "retrieval", + "arguments": `{"query":"ragflow"}`, + }, + }}, + }, + {Role: "tool", Content: "retrieved text", ToolCallID: "previous-call"}, + } + response, err := model.ChatWithMessages( + "qwen-flash", + messages, + &APIConfig{ApiKey: &apiKey}, + &ChatConfig{ + Tools: []map[string]interface{}{{"type": "function"}}, + ToolChoice: &auto, + }, + nil, + ) + if err != nil { + t.Fatalf("ChatWithMessages: %v", err) + } + if response.Answer == nil || *response.Answer != "final answer" { + t.Fatalf("answer = %#v, want final answer", response.Answer) + } + + body := <-requestBody + if body["tool_choice"] != "none" { + t.Fatalf("tool_choice = %v, want none after qwen-flash tool result", body["tool_choice"]) + } + gotMessages, ok := body["messages"].([]interface{}) + if !ok || len(gotMessages) != 3 { + t.Fatalf("messages = %T len=%d, want 3", body["messages"], len(gotMessages)) + } + assistantMessage, _ := gotMessages[1].(map[string]interface{}) + if _, ok := assistantMessage["tool_calls"].([]interface{}); !ok { + t.Errorf("assistant tool_calls = %T, want JSON array", assistantMessage["tool_calls"]) + } + toolMessage, _ := gotMessages[2].(map[string]interface{}) + if toolMessage["tool_call_id"] != "previous-call" { + t.Errorf("tool_call_id = %v, want previous-call", toolMessage["tool_call_id"]) + } +} + +func TestAliyunToolChoiceStopsOnlyQwenFlashAfterToolResult(t *testing.T) { + auto := "auto" + required := "required" + toolResult := []Message{{Role: "tool", Content: "result", ToolCallID: "call-1"}} + + tests := []struct { + name string + model string + messages []Message + configured *string + want string + }{ + {name: "qwen initial call", model: "qwen-flash", configured: &auto, want: "auto"}, + {name: "qwen after tool result", model: "qwen-flash", messages: toolResult, configured: &auto, want: "none"}, + {name: "versioned qwen after tool result", model: "qwen-flash-2025-07-28", messages: toolResult, configured: &auto, want: "none"}, + {name: "other aliyun model", model: "qwen-plus", messages: toolResult, configured: &auto, want: "auto"}, + {name: "explicit choice", model: "qwen-flash", messages: toolResult, configured: &required, want: "required"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := aliyunToolChoice(tt.model, tt.messages, tt.configured); got != tt.want { + t.Fatalf("aliyunToolChoice() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestAliyunChatStreamlyWithSenderSupportsToolCalls(t *testing.T) { + requestBody := make(chan map[string]interface{}, 1) + requestPath := make(chan string, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]interface{} + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + requestPath <- r.URL.Path + requestBody <- body + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(`data: {"choices":[{"delta":{"tool_calls":[{"index":1,"id":"call-2","type":"function","function":{"name":"lookup","arguments":"{}"}}]},"finish_reason":null}]} + +data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"retrieval","arguments":"{\"query\":\""}}]},"finish_reason":null}]} + +data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ragflow\"}"}}]},"finish_reason":"tool_calls"}]} + +`)) + })) + defer server.Close() + + model := NewAliyunModel( + map[string]string{"default": server.URL}, + URLSuffix{Chat: "compatible-mode/v1/chat/completions"}, + ) + apiKey := "test-key" + config := &ChatConfig{ + Tools: []map[string]interface{}{{ + "type": "function", + "function": map[string]interface{}{ + "name": "retrieval", + "parameters": map[string]interface{}{"type": "object"}, + }, + }}, + } + var streamed []string + err := model.ChatStreamlyWithSender( + "qwen-flash", + []Message{{Role: "user", Content: "find ragflow"}}, + &APIConfig{ApiKey: &apiKey}, + config, + nil, + func(content, reasoning *string) error { + if reasoning != nil { + return errors.New("unexpected reasoning content") + } + if content != nil { + streamed = append(streamed, *content) + } + return nil + }, + ) + if err != nil { + t.Fatalf("ChatStreamlyWithSender: %v", err) + } + + if got := <-requestPath; got != "/compatible-mode/v1/chat/completions" { + t.Errorf("request path = %q, want /compatible-mode/v1/chat/completions", got) + } + body := <-requestBody + if body["stream"] != true { + t.Errorf("stream = %v, want true", body["stream"]) + } + if body["tool_choice"] != "auto" { + t.Errorf("tool_choice = %v, want auto", body["tool_choice"]) + } + if _, ok := body["tools"].([]interface{}); !ok { + t.Fatalf("tools = %T, want JSON array", body["tools"]) + } + if len(streamed) != 1 || streamed[0] != "[DONE]" { + t.Errorf("streamed content = %#v, want only [DONE]", streamed) + } + if config.ToolCallsResult == nil { + t.Fatal("ToolCallsResult is nil") + } + toolCalls := *config.ToolCallsResult + if len(toolCalls) != 2 { + t.Fatalf("tool calls = %d, want 2", len(toolCalls)) + } + if toolCalls[0]["id"] != "call-1" || toolCalls[1]["id"] != "call-2" { + t.Fatalf("tool call order = [%v, %v], want [call-1, call-2]", toolCalls[0]["id"], toolCalls[1]["id"]) + } + function, _ := toolCalls[0]["function"].(map[string]interface{}) + if function["name"] != "retrieval" { + t.Errorf("function name = %v, want retrieval", function["name"]) + } + if function["arguments"] != `{"query":"ragflow"}` { + t.Errorf("function arguments = %v, want complete JSON", function["arguments"]) + } +} + +func TestAliyunChatStreamlyWithSenderRejectsStreamFalse(t *testing.T) { + model := NewAliyunModel( + map[string]string{"default": "https://dashscope.example"}, + URLSuffix{Chat: "compatible-mode/v1/chat/completions"}, + ) + apiKey := "test-key" + stream := false + err := model.ChatStreamlyWithSender( + "qwen-flash", + []Message{{Role: "user", Content: "hello"}}, + &APIConfig{ApiKey: &apiKey}, + &ChatConfig{Stream: &stream}, + nil, + func(_, _ *string) error { return nil }, + ) + if err == nil || err.Error() != "stream must be true in ChatStreamlyWithSender" { + t.Fatalf("error = %v, want stream validation error", err) + } +} diff --git a/internal/service/agent.go b/internal/service/agent.go index ed15dcb39d..c0bbf8705c 100644 --- a/internal/service/agent.go +++ b/internal/service/agent.go @@ -1414,7 +1414,7 @@ func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanv return state, nil } s.markRunFailed(ctx2, runID, "invoke: "+err.Error()) - return nil, fmt.Errorf("canvas invoke: %w: %w", ErrAgentStorageError, err) + return nil, fmt.Errorf("canvas invoke: %w", err) } // Emit message + message_end (mirrors Python's ans dict). diff --git a/internal/service/agent_run_e2e_test.go b/internal/service/agent_run_e2e_test.go index 99b98a417b..e2141eac5b 100644 --- a/internal/service/agent_run_e2e_test.go +++ b/internal/service/agent_run_e2e_test.go @@ -1711,10 +1711,8 @@ func (i *categorizeResumeInvoker) Invoke(_ context.Context, req component.ChatIn // matching the Python canvas.py soft-fail). Parameter-binding // sites keep runtime.ResolveTemplate's loud-fail contract. // -// mapAgentError must classify the resulting wrapped -// ErrAgentStorageError as CodeServerError (500). The test asserts -// at the service layer; the handler-layer mapping is exercised by -// TestMapAgentError in handler/agent_test.go. +// A component invocation error must retain its invoke context without being +// misclassified as an agent storage failure. func TestRunAgent_RealCanvas_InvokeFails(t *testing.T) { testDB := setupServiceTestDB(t) if err := testDB.AutoMigrate( @@ -1775,8 +1773,11 @@ func TestRunAgent_RealCanvas_InvokeFails(t *testing.T) { if len(errs) == 0 { t.Fatal("expected error event from Invoke of DSL with bad component query ref") } - if !strings.Contains(errs[0].Message, "agent storage error") { - t.Errorf("error message %q does not mention sanitised label", errs[0].Message) + if !strings.Contains(errs[0].Message, "canvas invoke:") { + t.Errorf("error message %q does not mention invoke context", errs[0].Message) + } + if strings.Contains(errs[0].Message, "agent storage error") { + t.Errorf("error message %q misclassifies invoke failure as storage failure", errs[0].Message) } }