fix: apply agent advanced settings and Tavily defaults (#22104)

This commit is contained in:
buua436 authored and GitHub committed 2026-11-09 14:36:25 +08:00
1 parent cf0551433d
commit 0bacfb2939
7 files changed
+410 -13

No files matched your search

+135 -9
View File
@@ -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
@@ -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)
}
}
+10 -4
View File
@@ -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
}
+50
View File
@@ -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)
}
}
+21
View File
@@ -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 {
+13
View File
@@ -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 {
@@ -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']);
});
});