mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-11-10 21:32:43 +08:00
Refactor: better to use component display name (#20431)
This commit is contained in:
1 parent
285b2c2903
commit
69ff9a9129
9 files changed
+106
-26
No files matched your search
@@ -62,10 +62,13 @@ type Canvas struct {
|
||||
// CanvasComponent is the in-memory DSL node. The Obj.ComponentName
|
||||
// matches agent/component/<name>.py's class name (case-insensitive,
|
||||
// per Python v1 DSL semantics).
|
||||
// DisplayName is the editable graph node name used in execution logs. It is
|
||||
// runtime metadata rather than part of the executable component topology.
|
||||
type CanvasComponent struct {
|
||||
Obj CanvasComponentObj `json:"obj"`
|
||||
Downstream []string `json:"downstream"`
|
||||
Upstream []string `json:"upstream"`
|
||||
Obj CanvasComponentObj `json:"obj"`
|
||||
Downstream []string `json:"downstream"`
|
||||
Upstream []string `json:"upstream"`
|
||||
DisplayName string `json:"-"`
|
||||
}
|
||||
|
||||
type CanvasComponentObj struct {
|
||||
|
||||
@@ -16,7 +16,10 @@
|
||||
|
||||
package canvas
|
||||
|
||||
import "fmt"
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// DecodeFromDSL converts a canonical canvas DSL map into a Canvas.
|
||||
// It accepts both canonical IMPORT shape (`obj.component_name`) and the
|
||||
@@ -33,6 +36,7 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) {
|
||||
Components: make(map[string]CanvasComponent, len(rawComps)),
|
||||
NodeParents: make(map[string]string),
|
||||
}
|
||||
displayNames := make(map[string]string)
|
||||
if p, ok := dsl["path"].([]any); ok {
|
||||
c.Path = make([]string, 0, len(p))
|
||||
for _, v := range p {
|
||||
@@ -58,6 +62,11 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) {
|
||||
if id != "" && parentID != "" {
|
||||
c.NodeParents[id] = parentID
|
||||
}
|
||||
data, _ := node["data"].(map[string]any)
|
||||
displayName, _ := data["name"].(string)
|
||||
if id != "" && strings.TrimSpace(displayName) != "" {
|
||||
displayNames[id] = displayName
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -71,6 +80,7 @@ func DecodeFromDSL(dsl map[string]any) (*Canvas, error) {
|
||||
return nil, fmt.Errorf("agent: component %q has empty component_name", cpnID)
|
||||
}
|
||||
c.Components[cpnID] = CanvasComponent{
|
||||
DisplayName: displayNames[cpnID],
|
||||
Obj: CanvasComponentObj{
|
||||
ComponentName: name,
|
||||
Params: params,
|
||||
|
||||
@@ -85,3 +85,32 @@ func TestDecodeFromDSLSkipsMalformedHistoryEntries(t *testing.T) {
|
||||
t.Fatalf("decoded history = %#v, want one assistant entry", decoded.History)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeFromDSLComponentDisplayName(t *testing.T) {
|
||||
dsl := map[string]any{
|
||||
"components": map[string]any{
|
||||
"CodeExec:SweetMooseTalk": map[string]any{
|
||||
"obj": map[string]any{
|
||||
"component_name": "CodeExec",
|
||||
"params": map[string]any{},
|
||||
},
|
||||
},
|
||||
},
|
||||
"graph": map[string]any{
|
||||
"nodes": []any{
|
||||
map[string]any{
|
||||
"id": "CodeExec:SweetMooseTalk",
|
||||
"data": map[string]any{"name": "Analyze sales"},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
decoded, err := DecodeFromDSL(dsl)
|
||||
if err != nil {
|
||||
t.Fatalf("DecodeFromDSL: %v", err)
|
||||
}
|
||||
if got := decoded.Components["CodeExec:SweetMooseTalk"].DisplayName; got != "Analyze sales" {
|
||||
t.Fatalf("display name = %q, want Analyze sales", got)
|
||||
}
|
||||
}
|
||||
@@ -255,7 +255,7 @@ func buildSubWorkflow(
|
||||
DeferAgentToMessage: deferToMessage,
|
||||
SuppressAgentMessageEvents: strings.EqualFold(name, "Agent") && !deferToMessage,
|
||||
}
|
||||
body, err := buildNodeBodyWithOptions(ctx, cpnID, name, c.Components[cpnID].Obj.Params, nodeOpts)
|
||||
body, err := buildNodeBodyWithOptions(ctx, cpnID, name, c.Components[cpnID].DisplayName, c.Components[cpnID].Obj.Params, nodeOpts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -120,11 +120,11 @@ func applyOverrideParams(params, cpnOverride map[string]any) map[string]any {
|
||||
return out
|
||||
}
|
||||
|
||||
func buildNodeBody(ctx context.Context, cpnID, name string, params map[string]any) (nodeBodyFn, error) {
|
||||
return buildNodeBodyWithOptions(ctx, cpnID, name, params, runtime.ComponentExecutionOptions{})
|
||||
func buildNodeBody(ctx context.Context, cpnID, name, displayName string, params map[string]any) (nodeBodyFn, error) {
|
||||
return buildNodeBodyWithOptions(ctx, cpnID, name, displayName, params, runtime.ComponentExecutionOptions{})
|
||||
}
|
||||
|
||||
func buildNodeBodyWithOptions(ctx context.Context, cpnID, name string, params map[string]any, opts runtime.ComponentExecutionOptions) (nodeBodyFn, error) {
|
||||
func buildNodeBodyWithOptions(ctx context.Context, cpnID, name, displayName string, params map[string]any, opts runtime.ComponentExecutionOptions) (nodeBodyFn, error) {
|
||||
if overrides := overrideParamsFromContext(ctx); len(overrides) > 0 {
|
||||
// overrides is keyed by cpnID; a component only sees its own
|
||||
// entry. Components absent from the map are left untouched.
|
||||
@@ -158,7 +158,7 @@ func buildNodeBodyWithOptions(ctx context.Context, cpnID, name string, params ma
|
||||
// without the runtime.Component interface needing to expose Name().
|
||||
// The factory returns the class name as the DSL's `component_name`
|
||||
// field, which is also what ComponentBase.Name() would have returned.
|
||||
return realComponentBodyWithOptions(cpnID, name, comp, opts), nil
|
||||
return realComponentBodyWithOptions(cpnID, name, displayName, comp, opts), nil
|
||||
}
|
||||
// Fallback: no factory registered. This path is only exercised by
|
||||
// canvas-only unit tests; production wiring always installs a
|
||||
@@ -216,19 +216,22 @@ func legacyNoOpBody(cpnID string) nodeBodyFn {
|
||||
// dataflow-result UI can show per-node timing without each
|
||||
// component repeating the bookkeeping.
|
||||
//
|
||||
// Timeout errors are surfaced as `timeout after Xs: <wrapped>`;
|
||||
// parent-context cancellation as `cancelled: <wrapped>`; all other
|
||||
// errors wrap the component's own error with the cpn_id for diagnostics.
|
||||
// Invocation errors identify the component by display name, falling back to the
|
||||
// cpn_id when no display name is available.
|
||||
//
|
||||
// The output map is tagged with __cpn_id__ before return so statePost
|
||||
// can attribute the result; if the component already populated that
|
||||
// key it is overwritten with the canvas-controlled value to keep
|
||||
// attribution authoritative.
|
||||
func realComponentBody(cpnID, componentClass string, comp runtime.Component) nodeBodyFn {
|
||||
return realComponentBodyWithOptions(cpnID, componentClass, comp, runtime.ComponentExecutionOptions{})
|
||||
func realComponentBody(cpnID, componentClass, displayName string, comp runtime.Component) nodeBodyFn {
|
||||
return realComponentBodyWithOptions(cpnID, componentClass, displayName, comp, runtime.ComponentExecutionOptions{})
|
||||
}
|
||||
|
||||
func realComponentBodyWithOptions(cpnID, componentClass string, comp runtime.Component, opts runtime.ComponentExecutionOptions) nodeBodyFn {
|
||||
func realComponentBodyWithOptions(cpnID, componentClass, displayName string, comp runtime.Component, opts runtime.ComponentExecutionOptions) nodeBodyFn {
|
||||
logName := strings.TrimSpace(displayName)
|
||||
if logName == "" {
|
||||
logName = cpnID
|
||||
}
|
||||
return func(ctx context.Context, in map[string]any) (map[string]any, error) {
|
||||
// The framework imposes no wall-clock deadline on component execution.
|
||||
// Long-running components (notably knowledge compilation, which fans out
|
||||
@@ -261,12 +264,12 @@ func realComponentBodyWithOptions(cpnID, componentClass string, comp runtime.Com
|
||||
common.Debug("agent: component invoke cancelled",
|
||||
zap.String("component_id", cpnID),
|
||||
zap.String("component_class", componentClass))
|
||||
return nil, fmt.Errorf("agent: component %q invoke: cancelled: %w", cpnID, invokeErr)
|
||||
return nil, fmt.Errorf("agent: component %q invoke: cancelled: %w", logName, invokeErr)
|
||||
case errors.Is(invokeErr, context.DeadlineExceeded):
|
||||
common.Error("agent: component invoke failed", invokeErr,
|
||||
zap.String("component_id", cpnID),
|
||||
zap.String("component_class", componentClass))
|
||||
return nil, fmt.Errorf("agent: component %q invoke: context deadline exceeded: %w", cpnID, invokeErr)
|
||||
return nil, fmt.Errorf("agent: component %q invoke: context deadline exceeded: %w", logName, invokeErr)
|
||||
default:
|
||||
// Surface the failure as a structured log line. The wrapped error
|
||||
// already carries the full cause chain (e.g. deepseek DNS/timeout),
|
||||
@@ -275,7 +278,7 @@ func realComponentBodyWithOptions(cpnID, componentClass string, comp runtime.Com
|
||||
common.Error("agent: component invoke failed", invokeErr,
|
||||
zap.String("component_id", cpnID),
|
||||
zap.String("component_class", componentClass))
|
||||
return nil, fmt.Errorf("agent: component %q invoke: %w", cpnID, invokeErr)
|
||||
return nil, fmt.Errorf("agent: component %q invoke: %w", logName, invokeErr)
|
||||
}
|
||||
}
|
||||
if out == nil {
|
||||
|
||||
@@ -56,8 +56,8 @@ func TestRealComponentBody_BindsFractionToNodeID(t *testing.T) {
|
||||
got = append(got, [2]interface{}{component, fraction})
|
||||
})
|
||||
|
||||
bodyA := realComponentBody("Parser:aaa", "TestFraction", &fractionComponent{fraction: 0.25})
|
||||
bodyB := realComponentBody("Parser:bbb", "TestFraction", &fractionComponent{fraction: 0.75})
|
||||
bodyA := realComponentBody("Parser:aaa", "TestFraction", "", &fractionComponent{fraction: 0.25})
|
||||
bodyB := realComponentBody("Parser:bbb", "TestFraction", "", &fractionComponent{fraction: 0.75})
|
||||
if _, err := bodyA(ctx, map[string]any{}); err != nil {
|
||||
t.Fatalf("bodyA: %v", err)
|
||||
}
|
||||
@@ -79,7 +79,7 @@ func TestRealComponentBody_BindsFractionToNodeID(t *testing.T) {
|
||||
// TestRealComponentBody_NoFractionSinkIsNoop verifies a headless run (no
|
||||
// run-level callback) pays nothing and components still report freely.
|
||||
func TestRealComponentBody_NoFractionSinkIsNoop(t *testing.T) {
|
||||
body := realComponentBody("test-cpn", "TestFraction", &fractionComponent{fraction: 0.5})
|
||||
body := realComponentBody("test-cpn", "TestFraction", "", &fractionComponent{fraction: 0.5})
|
||||
if _, err := body(t.Context(), map[string]any{}); err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
||||
@@ -66,7 +66,7 @@ func TestBuildNodeBody_OverrideParams(t *testing.T) {
|
||||
ctx := WithComponentFactory(t.Context(), factory)
|
||||
ctx = withOverrideParams(ctx, override)
|
||||
|
||||
body, err := buildNodeBody(ctx, "cpn-parser", "Parser", baseParams)
|
||||
body, err := buildNodeBody(ctx, "cpn-parser", "Parser", "", baseParams)
|
||||
if err != nil {
|
||||
t.Fatalf("buildNodeBody: %v", err)
|
||||
}
|
||||
@@ -109,7 +109,7 @@ func TestBuildNodeBody_OverrideParamsNilIsNoOp(t *testing.T) {
|
||||
baseParams := map[string]any{"name": "x"}
|
||||
|
||||
ctx := WithComponentFactory(t.Context(), factory)
|
||||
body, err := buildNodeBody(ctx, "cpn", "Parser", baseParams)
|
||||
body, err := buildNodeBody(ctx, "cpn", "Parser", "", baseParams)
|
||||
if err != nil {
|
||||
t.Fatalf("buildNodeBody: %v", err)
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ package canvas
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"ragflow/internal/agent/runtime"
|
||||
@@ -37,6 +38,21 @@ import (
|
||||
// cancelled.
|
||||
type blockingComponent struct{}
|
||||
|
||||
type failingComponent struct{}
|
||||
|
||||
func (f *failingComponent) Name() string { return "failing" }
|
||||
|
||||
func (f *failingComponent) Invoke(_ context.Context, _ *gorm.DB, _ map[string]any) (map[string]any, error) {
|
||||
return nil, errors.New("boom")
|
||||
}
|
||||
|
||||
func (f *failingComponent) Stream(_ context.Context, _ map[string]any) (<-chan map[string]any, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (f *failingComponent) Inputs() map[string]string { return nil }
|
||||
func (f *failingComponent) Outputs() map[string]string { return nil }
|
||||
|
||||
func (b *blockingComponent) Name() string { return "blocking" }
|
||||
|
||||
func (b *blockingComponent) Invoke(ctx context.Context, _ *gorm.DB, _ map[string]any) (map[string]any, error) {
|
||||
@@ -57,7 +73,7 @@ func (b *blockingComponent) Outputs() map[string]string { return nil }
|
||||
// the only mechanism that stops a long-running component.
|
||||
func TestRealComponentBody_RespectsParentCancellation(t *testing.T) {
|
||||
comp := &blockingComponent{}
|
||||
body := realComponentBody("test-cpn", "TestBlocking", comp)
|
||||
body := realComponentBody("test-cpn", "TestBlocking", "", comp)
|
||||
|
||||
parentCtx, cancel := context.WithCancel(t.Context())
|
||||
cancel() // pre-cancel
|
||||
@@ -69,13 +85,31 @@ func TestRealComponentBody_RespectsParentCancellation(t *testing.T) {
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("expected context.Canceled wrapped error, got: %v", err)
|
||||
}
|
||||
if !strings.Contains(err.Error(), `component "test-cpn" invoke`) {
|
||||
t.Errorf("error = %q, want component ID fallback", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRealComponentBody_UsesDisplayNameInError(t *testing.T) {
|
||||
body := realComponentBody("CodeExec:SweetMooseTalk", "CodeExec", "Analyze sales", &failingComponent{})
|
||||
|
||||
_, err := body(t.Context(), nil)
|
||||
if err == nil {
|
||||
t.Fatal("expected error, got nil")
|
||||
}
|
||||
if !strings.Contains(err.Error(), `component "Analyze sales" invoke`) {
|
||||
t.Errorf("error = %q, want display name", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "CodeExec:SweetMooseTalk") {
|
||||
t.Errorf("error = %q, should not contain component ID when display name is set", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRealComponentBody_NoTimeoutWhenFast verifies that a component returning
|
||||
// immediately does not incur any timeout-induced latency or error wrapping.
|
||||
func TestRealComponentBody_NoTimeoutWhenFast(t *testing.T) {
|
||||
comp := &echoComponent{}
|
||||
body := realComponentBody("test-cpn", "TestEcho", comp)
|
||||
body := realComponentBody("test-cpn", "TestEcho", "", comp)
|
||||
|
||||
out, err := body(t.Context(), map[string]any{"x": 1})
|
||||
if err != nil {
|
||||
@@ -113,5 +147,6 @@ func (e *echoComponent) Outputs() map[string]string { return nil }
|
||||
// Compile-time check that the stubs satisfy the interface.
|
||||
var (
|
||||
_ runtime.Component = (*blockingComponent)(nil)
|
||||
_ runtime.Component = (*failingComponent)(nil)
|
||||
_ runtime.Component = (*echoComponent)(nil)
|
||||
)
|
||||
@@ -557,7 +557,7 @@ func BuildWorkflow(ctx context.Context, c *Canvas) (*compose.Workflow[map[string
|
||||
DeferAgentToMessage: deferToMessage,
|
||||
SuppressAgentMessageEvents: strings.EqualFold(name, "Agent") && !deferToMessage,
|
||||
}
|
||||
body, err := buildNodeBodyWithOptions(ctx, cpnID, name, c.Components[cpnID].Obj.Params, nodeOpts)
|
||||
body, err := buildNodeBodyWithOptions(ctx, cpnID, name, c.Components[cpnID].DisplayName, c.Components[cpnID].Obj.Params, nodeOpts)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user