diff --git a/internal/agent/component/agent.go b/internal/agent/component/agent.go index fd079a5b6f..a72928672e 100644 --- a/internal/agent/component/agent.go +++ b/internal/agent/component/agent.go @@ -84,8 +84,11 @@ type AgentParam struct { SubAgents []SubAgentTool MCP any // Saved MCP selections, decoded and validated when tools are built. ToolTimeout time.Duration + MaxRetries int // retry attempts after a failed model call + DelayAfterError time.Duration // initial retry backoff MaxRounds int - MessageHistoryWindowSize int // number of prior conversation turns to include; zero disables history + MessageHistoryWindowSize int // number of prior conversation turns to include; zero disables history + VisualFilesVar string // canvas-state ref containing image data URIs for the model OptimizeMultiTurn bool OptimizeHistoryWindow int // Meta is the OpenAI-style function-call schema the Agent exposes @@ -117,6 +120,7 @@ type AgentParam struct { // outputs.structured; when set, Invoke populates outputs["structured"] // so downstream Message templates like {Agent:X@structured} resolve. OutputStructure map[string]any + visualFiles []string // image URIs resolved from VisualFilesVar for this invocation } // SubAgentTool is a child Agent exposed to a parent Agent as an Eino tool. @@ -195,7 +199,12 @@ func runEinoReActAgent(ctx context.Context, p AgentParam) (*schema.Message, erro } agent, err := react.NewAgent(ctx, &react.AgentConfig{ - ToolCallingModel: &roundLimitedChatModel{ToolCallingChatModel: chatModel, remaining: maxSteps / 2}, + ToolCallingModel: &roundLimitedChatModel{ + ToolCallingChatModel: chatModel, + remaining: maxSteps / 2, + maxRetries: p.MaxRetries, + initialDelay: p.DelayAfterError, + }, ToolsConfig: compose.ToolsNodeConfig{ Tools: tools, }, @@ -277,7 +286,9 @@ func runEinoReActAgent(ctx context.Context, p AgentParam) (*schema.Message, erro // The reserved final model step must answer rather than schedule more tools. type roundLimitedChatModel struct { model.ToolCallingChatModel - remaining int + remaining int + maxRetries int + initialDelay time.Duration } func (m *roundLimitedChatModel) WithTools(tools []*schema.ToolInfo) (model.ToolCallingChatModel, error) { @@ -285,7 +296,12 @@ func (m *roundLimitedChatModel) WithTools(tools []*schema.ToolInfo) (model.ToolC if err != nil { return nil, err } - return &roundLimitedChatModel{ToolCallingChatModel: bound, remaining: m.remaining}, nil + return &roundLimitedChatModel{ + ToolCallingChatModel: bound, + remaining: m.remaining, + maxRetries: m.maxRetries, + initialDelay: m.initialDelay, + }, nil } func (m *roundLimitedChatModel) roundOptions(opts []model.Option) []model.Option { @@ -298,12 +314,69 @@ func (m *roundLimitedChatModel) roundOptions(opts []model.Option) []model.Option func (m *roundLimitedChatModel) Generate(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.Message, error) { input = m.roundInput(input) - return m.ToolCallingChatModel.Generate(ctx, input, m.roundOptions(opts)...) + opts = m.roundOptions(opts) + var response *schema.Message + err := common.RetryWithBackoff(ctx, m.maxRetries, m.initialDelay, func() error { + var err error + response, err = m.ToolCallingChatModel.Generate(ctx, input, opts...) + return err + }) + if err != nil { + return nil, err + } + return response, nil } func (m *roundLimitedChatModel) Stream(ctx context.Context, input []*schema.Message, opts ...model.Option) (*schema.StreamReader[*schema.Message], error) { input = m.roundInput(input) - return m.ToolCallingChatModel.Stream(ctx, input, m.roundOptions(opts)...) + opts = m.roundOptions(opts) + return retryChatModelStream(ctx, m.ToolCallingChatModel, input, opts, m.maxRetries, m.initialDelay) +} + +// retryChatModelStream retries only before the first message reaches the +// caller, so a failed partial response is never replayed as duplicate output. +func retryChatModelStream(ctx context.Context, inner model.ToolCallingChatModel, input []*schema.Message, opts []model.Option, maxRetries int, initialDelay time.Duration) (*schema.StreamReader[*schema.Message], error) { + if maxRetries <= 0 { + return inner.Stream(ctx, input, opts...) + } + + reader, writer := schema.Pipe[*schema.Message](1) + go func() { + defer writer.Close() + emitted := false + readerClosed := false + err := common.RetryWithBackoff(ctx, maxRetries, initialDelay, func() error { + emitted = false + stream, err := inner.Stream(ctx, input, opts...) + if err != nil { + return err + } + defer stream.Close() + for { + message, err := stream.Recv() + if errors.Is(err, io.EOF) { + return nil + } + if err != nil { + return err + } + if message == nil { + continue + } + emitted = true + if writer.Send(message, nil) { + readerClosed = true + return context.Canceled + } + } + }, func(error) bool { + return !emitted && !readerClosed + }) + if err != nil { + writer.Send(nil, err) + } + }() + return reader, nil } func (m *roundLimitedChatModel) roundInput(input []*schema.Message) []*schema.Message { @@ -389,6 +462,7 @@ func buildAgentInputMessages(ctx context.Context, p AgentParam) []*schema.Messag } } } + images = dedupStrings(append(images, p.visualFiles...)) current := schema.Message{Role: schema.User, Content: userText} if len(images) > 0 { current = userMessageWithImages(userText, images) @@ -414,6 +488,42 @@ func buildAgentInputMessages(ctx context.Context, p AgentParam) []*schema.Messag return input } +func resolveAgentVisualFiles(state *runtime.CanvasState, variable string) ([]string, error) { + variable = strings.TrimSpace(variable) + if variable == "" { + return nil, nil + } + if state == nil { + return nil, fmt.Errorf("canvas state is required for %q", variable) + } + value, err := state.GetVar(variable) + if err != nil { + return nil, err + } + if value == nil { + return nil, nil + } + var files []string + switch v := value.(type) { + case string: + files = []string{v} + case []string: + files = v + case []any: + files = make([]string, 0, len(v)) + for _, item := range v { + file, ok := item.(string) + if !ok { + return nil, fmt.Errorf("%q must resolve to a string or string array", variable) + } + files = append(files, file) + } + default: + return nil, fmt.Errorf("%q must resolve to a string or string array, got %T", variable, value) + } + return extractDataImages(files), nil +} + func emitAgentModelStreams(ctx context.Context, future react.MessageFuture, cite bool, holdContent bool) <-chan error { done := make(chan error, 1) go func() { @@ -1059,10 +1169,17 @@ func (c *AgentComponent) invokeNow(ctx context.Context, db *gorm.DB, inputs map[ p.UserPrompt = resolved } } + p.visualFiles, err = resolveAgentVisualFiles(state, p.VisualFilesVar) + if err != nil { + return nil, fmt.Errorf("Agent: resolve visual_files_var: %w", err) + } + var imageFiles []string if state != nil { - if err := rejectUnsupportedImages(ctx, db, state, originalModelID, p.ModelID); err != nil { - return nil, err - } + _, imageFiles = collectSysFiles(state) + } + imageFiles = append(imageFiles, p.visualFiles...) + if err := rejectUnsupportedImageFiles(ctx, db, imageFiles, originalModelID, p.ModelID); err != nil { + return nil, err } if hasRuntimeUserPrompt { p.UserPrompt = formatAgentRuntimePrompt(inputs, p.UserPrompt) @@ -1727,6 +1844,15 @@ func mergeAgentParam(base AgentParam, inputs map[string]any) AgentParam { if v, ok := intFrom(inputs, "tool_timeout"); ok { p.ToolTimeout = time.Duration(v) * time.Second } + if v, ok := intFrom(inputs, "max_retries"); ok { + p.MaxRetries = v + } + if v, ok := floatFrom(inputs, "delay_after_error"); ok { + p.DelayAfterError = time.Duration(v * float64(time.Second)) + } + if v, ok := stringFrom(inputs, "visual_files_var"); ok { + p.VisualFilesVar = v + } if tools, params, subAgents, ok := agentToolsFrom(inputs, "tools"); ok { p.Tools = tools p.SubAgents = subAgents diff --git a/internal/agent/component/agent_retry_test.go b/internal/agent/component/agent_retry_test.go new file mode 100644 index 0000000000..fd406839f3 --- /dev/null +++ b/internal/agent/component/agent_retry_test.go @@ -0,0 +1,155 @@ +package component + +import ( + "context" + "errors" + "io" + "testing" + "time" + + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/schema" +) + +type retryTestChatModel struct { + generateCalls int + streamCalls int + failStreamAfterFirstChunk bool +} + +func (m *retryTestChatModel) WithTools([]*schema.ToolInfo) (model.ToolCallingChatModel, error) { + return m, nil +} + +func (m *retryTestChatModel) Generate(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.Message, error) { + m.generateCalls++ + if m.generateCalls == 1 { + return nil, errors.New("temporary model failure") + } + return &schema.Message{Role: schema.Assistant, Content: "recovered"}, nil +} + +func (m *retryTestChatModel) Stream(_ context.Context, _ []*schema.Message, _ ...model.Option) (*schema.StreamReader[*schema.Message], error) { + m.streamCalls++ + reader, writer := schema.Pipe[*schema.Message](1) + go func(attempt int) { + defer writer.Close() + if attempt == 1 { + if m.failStreamAfterFirstChunk { + writer.Send(&schema.Message{Role: schema.Assistant, Content: "partial"}, nil) + writer.Send(nil, errors.New("stream failed after output")) + return + } + writer.Send(nil, errors.New("temporary stream failure")) + return + } + writer.Send(&schema.Message{Role: schema.Assistant, Content: "recovered"}, nil) + }(m.streamCalls) + return reader, nil +} + +func TestRoundLimitedChatModelDoesNotRetryStreamAfterOutput(t *testing.T) { + inner := &retryTestChatModel{failStreamAfterFirstChunk: true} + chatModel := &roundLimitedChatModel{ + ToolCallingChatModel: inner, + remaining: 1, + maxRetries: 2, + } + + stream, err := chatModel.Stream(t.Context(), []*schema.Message{schema.UserMessage("hello")}) + if err != nil { + t.Fatalf("Stream: %v", err) + } + defer stream.Close() + message, err := stream.Recv() + if err != nil || message.Content != "partial" { + t.Fatalf("first Recv = (%v, %v), want partial message", message, err) + } + if _, err = stream.Recv(); err == nil || err.Error() != "stream failed after output" { + t.Fatalf("second Recv error = %v, want original stream error", err) + } + if inner.streamCalls != 1 { + t.Fatalf("stream called %d times, want no retry after output", inner.streamCalls) + } +} + +func TestRoundLimitedChatModelWithToolsPreservesRetrySettings(t *testing.T) { + chatModel := &roundLimitedChatModel{ + ToolCallingChatModel: &retryTestChatModel{}, + remaining: 2, + maxRetries: 3, + initialDelay: 250 * time.Millisecond, + } + + bound, err := chatModel.WithTools([]*schema.ToolInfo{{Name: "search"}}) + if err != nil { + t.Fatalf("WithTools: %v", err) + } + got, ok := bound.(*roundLimitedChatModel) + if !ok { + t.Fatalf("bound model type = %T, want *roundLimitedChatModel", bound) + } + if got.maxRetries != 3 || got.initialDelay != 250*time.Millisecond || got.remaining != 2 { + t.Fatalf("bound retry settings = (%d, %s, %d), want (3, 250ms, 2)", got.maxRetries, got.initialDelay, got.remaining) + } +} + +func TestRoundLimitedChatModelRetriesGenerate(t *testing.T) { + inner := &retryTestChatModel{} + chatModel := &roundLimitedChatModel{ + ToolCallingChatModel: inner, + remaining: 1, + maxRetries: 1, + } + + got, err := chatModel.Generate(t.Context(), []*schema.Message{schema.UserMessage("hello")}) + if err != nil { + t.Fatalf("Generate: %v", err) + } + if got.Content != "recovered" || inner.generateCalls != 2 { + t.Fatalf("Generate = (%q, %d calls), want (recovered, 2 calls)", got.Content, inner.generateCalls) + } +} + +func TestRoundLimitedChatModelRetriesStreamBeforeFirstMessage(t *testing.T) { + inner := &retryTestChatModel{} + chatModel := &roundLimitedChatModel{ + ToolCallingChatModel: inner, + remaining: 1, + maxRetries: 1, + } + + stream, err := chatModel.Stream(t.Context(), []*schema.Message{schema.UserMessage("hello")}) + if err != nil { + t.Fatalf("Stream: %v", err) + } + defer stream.Close() + got, err := stream.Recv() + if err != nil { + t.Fatalf("Recv: %v", err) + } + if got.Content != "recovered" || inner.streamCalls != 2 { + t.Fatalf("Stream = (%q, %d calls), want (recovered, 2 calls)", got.Content, inner.streamCalls) + } + if _, err := stream.Recv(); !errors.Is(err, io.EOF) { + t.Fatalf("second Recv error = %v, want EOF", err) + } +} + +func TestMergeAgentParamReadsAdvancedSettings(t *testing.T) { + p := mergeAgentParam(AgentParam{}, map[string]any{ + "max_retries": float64(4), + "delay_after_error": float64(0.25), + "visual_files_var": "Begin:attachments", + }) + + if p.MaxRetries != 4 { + t.Errorf("MaxRetries = %d, want 4", p.MaxRetries) + } + if p.DelayAfterError != 250*time.Millisecond { + t.Errorf("DelayAfterError = %s, want 250ms", p.DelayAfterError) + } + if p.VisualFilesVar != "Begin:attachments" { + t.Errorf("VisualFilesVar = %q, want Begin:attachments", p.VisualFilesVar) + } +} diff --git a/internal/agent/component/llm_credentials.go b/internal/agent/component/llm_credentials.go index b826947aca..e803319e93 100644 --- a/internal/agent/component/llm_credentials.go +++ b/internal/agent/component/llm_credentials.go @@ -117,10 +117,16 @@ func rejectUnsupportedImages(ctx context.Context, db *gorm.DB, state *runtime.Ca if state == nil { return nil } - if _, images := collectSysFiles(state); len(images) > 0 { - if supports, known := modelImageCapability(ctx, db, modelRef, modelName); known && !supports { - return runtime.NewUserFacingError(fmt.Sprintf(`Image input is not supported by the selected model %q. Please select a vision-capable model.`, modelRef)) - } + _, images := collectSysFiles(state) + return rejectUnsupportedImageFiles(ctx, db, images, modelRef, modelName) +} + +func rejectUnsupportedImageFiles(ctx context.Context, db *gorm.DB, images []string, modelRef, modelName string) error { + if len(images) == 0 { + return nil + } + if supports, known := modelImageCapability(ctx, db, modelRef, modelName); known && !supports { + return runtime.NewUserFacingError(fmt.Sprintf(`Image input is not supported by the selected model %q. Please select a vision-capable model.`, modelRef)) } return nil } diff --git a/internal/agent/component/vision_test.go b/internal/agent/component/vision_test.go index 2b6f116071..df50be4e77 100644 --- a/internal/agent/component/vision_test.go +++ b/internal/agent/component/vision_test.go @@ -450,3 +450,53 @@ func TestBuildAgentInputMessagesSysFilesTextOnly(t *testing.T) { len(messages[0].UserInputMultiContent)) } } + +func TestAgentVisualFilesVarResolvesImageVariable(t *testing.T) { + uri := "data:image/png;base64,iVBORw0KGgo=" + state := runtime.NewCanvasState("run-agent-selected-file", "task-agent-selected-file") + state.SetVar("FileNode", "attachments", []any{"notes.txt", uri}) + + images, err := resolveAgentVisualFiles(state, "FileNode@attachments") + if err != nil { + t.Fatalf("resolveAgentVisualFiles: %v", err) + } + if !reflect.DeepEqual(images, []string{uri}) { + t.Fatalf("resolved images = %#v, want [%q]", images, uri) + } + + messages := buildAgentInputMessages(runtime.WithState(t.Context(), state), AgentParam{ + UserPrompt: "describe this", + visualFiles: images, + }) + if len(messages) != 1 || len(messages[0].UserInputMultiContent) != 2 { + t.Fatalf("messages = %#v, want one user message with text and image parts", messages) + } + image := messages[0].UserInputMultiContent[1] + if image.Type != schema.ChatMessagePartTypeImageURL || image.Image == nil || image.Image.URL == nil || *image.Image.URL != uri { + t.Fatalf("selected image part = %+v, want URI %q", image, uri) + } +} + +func TestAgentInvokeUsesVisualFilesVar(t *testing.T) { + uri := "data:image/jpeg;base64,/9j/4AAQ" + state := runtime.NewCanvasState("run-agent-selected-file", "task-agent-selected-file") + state.SetVar("FileNode", "attachments", []string{uri}) + var captured []string + withAgentRunner(t, func(_ context.Context, p AgentParam) (*schema.Message, error) { + captured = p.visualFiles + return &schema.Message{Role: schema.Assistant, Content: "described"}, nil + }) + + component := NewAgentComponent(agentParamFromMap(map[string]any{ + "model_id": "stub", + "visual_files_var": "FileNode@attachments", + })) + if _, err := component.Invoke(runtime.WithState(t.Context(), state), nil, map[string]any{ + "user_prompt": "describe this image", + }); err != nil { + t.Fatalf("Invoke: %v", err) + } + if !reflect.DeepEqual(captured, []string{uri}) { + t.Fatalf("agent runner visual files = %#v, want [%q]", captured, uri) + } +} diff --git a/internal/agent/tool/registry.go b/internal/agent/tool/registry.go index fee987bf1a..b6a53b8846 100644 --- a/internal/agent/tool/registry.go +++ b/internal/agent/tool/registry.go @@ -669,6 +669,27 @@ func buildTavilyTool(params map[string]any) (einotool.BaseTool, error) { } defaults.Days = days } + if value, ok := params["topic"]; ok { + topic, valid := value.(string) + if !valid || (topic != "general" && topic != "news") { + return nil, fmt.Errorf("agent tool: tool %q has unsupported topic %v", "tavily", value) + } + defaults.Topic = topic + } + for _, key := range []string{"include_domains", "exclude_domains"} { + domains, present, err := stringSliceParam(params, key) + if err != nil { + return nil, fmt.Errorf("agent tool: tool %q requires string-list node-level param %s: %w", "tavily", key, err) + } + if !present { + continue + } + if key == "include_domains" { + defaults.IncludeDomains = domains + } else { + defaults.ExcludeDomains = domains + } + } for _, key := range []string{"include_answer", "include_raw_content", "include_images", "include_image_descriptions"} { value, ok := params[key] if !ok { diff --git a/internal/agent/tool/tavily_test.go b/internal/agent/tool/tavily_test.go index f419232584..6d4d131424 100644 --- a/internal/agent/tool/tavily_test.go +++ b/internal/agent/tool/tavily_test.go @@ -21,6 +21,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "reflect" "strings" "testing" @@ -318,8 +319,11 @@ func TestTavily_BuildByNameUsesNodeDefaults(t *testing.T) { built, err := BuildByName("tavily", map[string]any{ "api_key": "stored-key", "search_depth": "advanced", + "topic": "news", "max_results": float64(12), "days": float64(7), + "include_domains": []any{"example.com", "news.example"}, + "exclude_domains": []any{"blocked.example"}, "include_answer": true, "include_raw_content": true, "include_images": true, @@ -337,6 +341,9 @@ func TestTavily_BuildByNameUsesNodeDefaults(t *testing.T) { if tavily.defaults.APIKey != "stored-key" || tavily.defaults.SearchDepth != "advanced" || tavily.defaults.MaxResults != 12 || tavily.defaults.Days != 7 { t.Fatalf("defaults = %+v", tavily.defaults) } + if tavily.defaults.Topic != "news" || !reflect.DeepEqual(tavily.defaults.IncludeDomains, []string{"example.com", "news.example"}) || !reflect.DeepEqual(tavily.defaults.ExcludeDomains, []string{"blocked.example"}) { + t.Fatalf("topic/domain defaults = %+v", tavily.defaults) + } if !tavily.defaults.IncludeAnswer || !tavily.defaults.IncludeRawContent || !tavily.defaults.IncludeImages || !tavily.defaults.IncludeImageDescriptions { t.Fatalf("boolean defaults = %+v", tavily.defaults) } @@ -357,6 +364,7 @@ func TestTavily_ExplicitFlagsOverrideNodeDefaults(t *testing.T) { helper := NewHTTPHelper().WithClient(&http.Client{Transport: rewriteHostTransport(srv.URL)}) tavily := newTavilyTool(helper, func() string { return "" }, tavilyParams{ APIKey: "stored-key", IncludeAnswer: true, + Topic: "news", IncludeDomains: []string{"example.com"}, ExcludeDomains: []string{"blocked.example"}, }) if _, err := tavily.InvokableRun(ctx, `{"query":"ragflow"}`); err != nil { t.Fatalf("InvokableRun(node defaults): %v", err) @@ -364,6 +372,9 @@ func TestTavily_ExplicitFlagsOverrideNodeDefaults(t *testing.T) { if gotBody["include_answer"] != true { t.Fatalf("node default include_answer was not sent: %#v", gotBody) } + if gotBody["topic"] != "news" || !reflect.DeepEqual(gotBody["include_domains"], []any{"example.com"}) || !reflect.DeepEqual(gotBody["exclude_domains"], []any{"blocked.example"}) { + t.Fatalf("node topic/domain defaults were not sent: %#v", gotBody) + } if gotBody["include_raw_content"] != false || gotBody["include_images"] != false { t.Fatalf("include_raw_content/images should default to false: %#v", gotBody) } @@ -415,11 +426,13 @@ func TestTavily_BuildByNameRejectsInvalidNodeDefaults(t *testing.T) { cases := []map[string]any{ {"api_key": 1}, {"search_depth": "deep"}, + {"topic": "finance"}, {"max_results": 0}, {"max_results": 21}, {"max_results": 1.5}, {"days": 0}, {"include_answer": "yes"}, + {"include_domains": []any{map[string]any{"value": "example.com"}}}, } for _, params := range cases { if _, err := BuildByName("tavily", params); err == nil { diff --git a/web/src/pages/agent/utils/__tests__/build-dsl-components.test.ts b/web/src/pages/agent/utils/__tests__/build-dsl-components.test.ts index 58ceb4bbdf..61142bf2d9 100644 --- a/web/src/pages/agent/utils/__tests__/build-dsl-components.test.ts +++ b/web/src/pages/agent/utils/__tests__/build-dsl-components.test.ts @@ -91,3 +91,29 @@ describe('buildDslComponentsByGraph ManualChunker params', () => { }); }); }); + +describe('buildDslComponentsByGraph DataOperations params', () => { + it('flattens form rows into the backend string-array shape', () => { + const node = { + id: 'DataOperations:TestId', + type: 'dataOperationsNode', + position: { x: 0, y: 0 }, + data: { + label: Operator.DataOperations, + name: 'Data Operations', + form: { + query: [{ input: 'Begin:query' }, { input: 'Parser:result' }], + select_keys: [{ name: 'title' }, { name: 'content' }], + remove_keys: [{ name: 'secret' }], + }, + }, + } as any; + + const components = buildDslComponentsByGraph([node], [], {}) as any; + const params = components['DataOperations:TestId'].obj.params; + + expect(params.query).toEqual(['Begin:query', 'Parser:result']); + expect(params.select_keys).toEqual(['title', 'content']); + expect(params.remove_keys).toEqual(['secret']); + }); +});