From ed9d60db0d122fa35d051750e44dd90d917533a2 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Tue, 23 Jun 2026 09:34:08 +0800 Subject: [PATCH] Add workflow runtime execution and condition validation --- internal/ai/application/runtime/service.go | 90 +++- .../ai/application/runtime/tool_catalog.go | 4 +- .../application/runtime/tool_catalog_test.go | 57 ++- .../application/runtime/workflow_runtime.go | 38 +- internal/ai/runtime/workflow/executor.go | 408 ++++++++++++++++++ internal/ai/runtime/workflow/executor_test.go | 94 ++++ internal/ai/workflow/validator/validator.go | 75 ++++ .../ai/workflow/validator/validator_test.go | 54 +++ .../_components/workflow-editor.tsx | 218 +++++++++- .../_components/workflow-utils.ts | 33 +- 10 files changed, 1026 insertions(+), 45 deletions(-) create mode 100644 internal/ai/runtime/workflow/executor.go create mode 100644 internal/ai/runtime/workflow/executor_test.go diff --git a/internal/ai/application/runtime/service.go b/internal/ai/application/runtime/service.go index 6e3f3e7..d989f6e 100644 --- a/internal/ai/application/runtime/service.go +++ b/internal/ai/application/runtime/service.go @@ -2,9 +2,16 @@ package runtime import ( "context" + "encoding/json" + "time" "agent-desk/internal/ai/runtime/executor" + workflowexecutor "agent-desk/internal/ai/runtime/workflow" + "agent-desk/internal/models" "agent-desk/internal/pkg/utils" + "agent-desk/internal/repositories" + + "github.com/mlogclub/simple/sqls" ) type Service struct { @@ -24,32 +31,29 @@ func NewService() *Service { func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { req.UserMessage.Content = utils.BuildRuntimeMessageText(req.UserMessage.MessageType, req.UserMessage.Content) - aiAgent, err := prepareWorkflowAgent(req.AIAgent) + aiAgent, workflow, err := prepareWorkflowAgent(req.AIAgent) if err != nil { return nil, err } req.AIAgent = aiAgent - toolSet, err := s.prepare.prepareToolsForRun(req) - if err != nil { - return nil, err - } - req.ToolSet = toolSet - summary, err := s.runtime.ExecuteRun(ctx, executor.RunInput{ + workflowResult, err := workflowexecutor.NewExecutor().Execute(ctx, workflowexecutor.Input{ + Definition: workflow.Definition, Conversation: req.Conversation, UserMessage: req.UserMessage, AIAgent: req.AIAgent, AIConfig: req.AIConfig, - CheckPointID: req.CheckPointID, - ToolSet: req.ToolSet, }) if err != nil { - return toSummary(summary), err + return nil, err } - return toSummary(summary), nil + if err := writeWorkflowRun(req, workflow, workflowResult, ""); err != nil { + return nil, err + } + return toWorkflowSummary(workflowResult, req.AIConfig.ModelName), nil } func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) { - aiAgent, err := prepareWorkflowAgent(req.AIAgent) + aiAgent, _, err := prepareWorkflowAgent(req.AIAgent) if err != nil { return nil, err } @@ -72,3 +76,65 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro } return toSummary(summary), nil } + +func toWorkflowSummary(result *workflowexecutor.Result, modelName string) *Summary { + if result == nil { + return nil + } + trace := map[string]any{ + "status": result.Status, + "nodePath": result.NodePath, + } + traceData, _ := json.Marshal(trace) + return &Summary{ + Status: result.Status, + ReplyText: result.ReplyText, + ModelName: modelName, + PromptTokens: result.PromptTokens, + CompletionTokens: result.CompletionTokens, + RetrieverCount: result.RetrieverCount, + TraceData: string(traceData), + } +} + +func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) error { + if result == nil { + return nil + } + now := time.Now() + endedAt := now + nodeTypes := make(map[string]string, len(workflow.Definition.Nodes)) + for _, node := range workflow.Definition.Nodes { + nodeTypes[node.ID] = node.Type + } + return sqls.WithTransaction(func(ctx *sqls.TxContext) error { + run := &models.AIWorkflowRun{ + WorkflowID: workflow.WorkflowID, + WorkflowVersionID: workflow.VersionID, + ConversationID: req.Conversation.ID, + AIAgentID: req.AIAgent.ID, + MessageID: req.UserMessage.ID, + Status: 1, + StartedAt: now, + EndedAt: &endedAt, + ErrorMessage: errorMessage, + } + if err := repositories.AIWorkflowRunRepository.Create(ctx.Tx, run); err != nil { + return err + } + for _, nodeID := range result.NodePath { + nodeRun := &models.AIWorkflowNodeRun{ + WorkflowRunID: run.ID, + NodeID: nodeID, + NodeType: nodeTypes[nodeID], + Status: 1, + StartedAt: now, + EndedAt: &endedAt, + } + if err := repositories.AIWorkflowNodeRunRepository.Create(ctx.Tx, nodeRun); err != nil { + return err + } + } + return nil + }) +} diff --git a/internal/ai/application/runtime/tool_catalog.go b/internal/ai/application/runtime/tool_catalog.go index 1f27d95..a9aef2c 100644 --- a/internal/ai/application/runtime/tool_catalog.go +++ b/internal/ai/application/runtime/tool_catalog.go @@ -82,8 +82,8 @@ func (c *toolCatalog) parseAgentAllowedToolCodes(aiAgent models.AIAgent) []strin ret = append(ret, graphTools...) } } - if result, err := resolveAgentWorkflow(aiAgent); err == nil { - ret = append(ret, result.ToolCodes...) + if workflow, err := resolveAgentWorkflow(aiAgent); err == nil { + ret = append(ret, workflow.Compiled.ToolCodes...) } return toolx.NormalizeToolCodes(ret) } diff --git a/internal/ai/application/runtime/tool_catalog_test.go b/internal/ai/application/runtime/tool_catalog_test.go index 94717fe..88aabe2 100644 --- a/internal/ai/application/runtime/tool_catalog_test.go +++ b/internal/ai/application/runtime/tool_catalog_test.go @@ -1,6 +1,7 @@ package runtime import ( + "context" "encoding/json" "strings" "testing" @@ -104,7 +105,7 @@ func TestPrepareWorkflowAgentAppendsPublishedWorkflow(t *testing.T) { }, }) - agent, err := prepareWorkflowAgent(models.AIAgent{ + agent, _, err := prepareWorkflowAgent(models.AIAgent{ SystemPrompt: "Base prompt.", WorkflowVersionID: version.ID, }) @@ -120,7 +121,7 @@ func TestPrepareWorkflowAgentAppendsPublishedWorkflow(t *testing.T) { } func TestPrepareWorkflowAgentRejectsMissingPublishedWorkflow(t *testing.T) { - _, err := prepareWorkflowAgent(models.AIAgent{}) + _, _, err := prepareWorkflowAgent(models.AIAgent{}) if err == nil { t.Fatalf("expected missing workflow version error") } @@ -143,7 +144,7 @@ func TestPrepareWorkflowAgentRejectsDeletedPublishedWorkflow(t *testing.T) { t.Fatalf("delete workflow version: %v", err) } - _, err := prepareWorkflowAgent(models.AIAgent{WorkflowVersionID: version.ID}) + _, _, err := prepareWorkflowAgent(models.AIAgent{WorkflowVersionID: version.ID}) if err == nil { t.Fatalf("expected invalid workflow version error") } @@ -152,13 +153,61 @@ func TestPrepareWorkflowAgentRejectsDeletedPublishedWorkflow(t *testing.T) { } } +func TestServiceRunExecutesPublishedWorkflow(t *testing.T) { + setupWorkflowRuntimeTestDB(t) + version := createWorkflowRuntimeTestVersion(t, dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, + {ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Reply", Config: []byte(`{"staticReply":"workflow reply"}`)}, + {ID: "send_1", Type: workflowregistry.NodeTypeSendReply, Name: "Send", Inputs: map[string]dsl.VariableSelector{ + "replyText": {NodeID: "reply_1", Field: "replyText"}, + }}, + {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_reply", Source: "start_1", Target: "reply_1"}, + {ID: "edge_reply_send", Source: "reply_1", Target: "send_1"}, + {ID: "edge_send_end", Source: "send_1", Target: "end_1"}, + }, + }) + + summary, err := NewService().Run(context.Background(), Request{ + UserMessage: models.Message{Content: "hello"}, + AIAgent: models.AIAgent{ + WorkflowVersionID: version.ID, + }, + }) + if err != nil { + t.Fatalf("run workflow: %v", err) + } + if summary.ReplyText != "workflow reply" { + t.Fatalf("unexpected workflow reply: %q", summary.ReplyText) + } + var runCount int64 + if err := sqls.DB().Model(&models.AIWorkflowRun{}).Count(&runCount).Error; err != nil { + t.Fatalf("count workflow runs: %v", err) + } + if runCount != 1 { + t.Fatalf("expected one workflow run, got %d", runCount) + } + var nodeRunCount int64 + if err := sqls.DB().Model(&models.AIWorkflowNodeRun{}).Count(&nodeRunCount).Error; err != nil { + t.Fatalf("count workflow node runs: %v", err) + } + if nodeRunCount != 4 { + t.Fatalf("expected four workflow node runs, got %d", nodeRunCount) + } +} + func setupWorkflowRuntimeTestDB(t *testing.T) { t.Helper() db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{}) if err != nil { t.Fatalf("open sqlite db: %v", err) } - if err := db.AutoMigrate(&models.AIWorkflowVersion{}); err != nil { + if err := db.AutoMigrate(&models.AIWorkflowVersion{}, &models.AIWorkflowRun{}, &models.AIWorkflowNodeRun{}); err != nil { t.Fatalf("auto migrate: %v", err) } sqls.SetDB(db) diff --git a/internal/ai/application/runtime/workflow_runtime.go b/internal/ai/application/runtime/workflow_runtime.go index 3b702ee..fe5466c 100644 --- a/internal/ai/application/runtime/workflow_runtime.go +++ b/internal/ai/application/runtime/workflow_runtime.go @@ -14,35 +14,47 @@ import ( "github.com/mlogclub/simple/sqls" ) -func resolveAgentWorkflow(aiAgent models.AIAgent) (compiler.Result, error) { +type resolvedWorkflow struct { + Definition dsl.Definition + Compiled compiler.Result + WorkflowID int64 + VersionID int64 +} + +func resolveAgentWorkflow(aiAgent models.AIAgent) (resolvedWorkflow, error) { if aiAgent.WorkflowVersionID <= 0 { - return compiler.Result{}, errorsx.InvalidParam("workflow version is required") + return resolvedWorkflow{}, errorsx.InvalidParam("workflow version is required") } version := repositories.AIWorkflowVersionRepository.Get(sqls.DB(), aiAgent.WorkflowVersionID) if version == nil || version.Status != enums.StatusOk { - return compiler.Result{}, errorsx.InvalidParam("workflow version does not exist") + return resolvedWorkflow{}, errorsx.InvalidParam("workflow version does not exist") } var def dsl.Definition if err := json.Unmarshal([]byte(version.Definition), &def); err != nil { - return compiler.Result{}, errorsx.InvalidParam("workflow definition is invalid") + return resolvedWorkflow{}, errorsx.InvalidParam("workflow definition is invalid") } - return compiler.Compile(def), nil + return resolvedWorkflow{ + Definition: def, + Compiled: compiler.Compile(def), + WorkflowID: version.WorkflowID, + VersionID: version.ID, + }, nil } -func prepareWorkflowAgent(aiAgent models.AIAgent) (models.AIAgent, error) { - result, err := resolveAgentWorkflow(aiAgent) +func prepareWorkflowAgent(aiAgent models.AIAgent) (models.AIAgent, resolvedWorkflow, error) { + workflow, err := resolveAgentWorkflow(aiAgent) if err != nil { - return aiAgent, err + return aiAgent, resolvedWorkflow{}, err } - if strings.TrimSpace(result.Appendix) == "" { - return aiAgent, nil + if strings.TrimSpace(workflow.Compiled.Appendix) == "" { + return aiAgent, workflow, nil } prompt := strings.TrimSpace(aiAgent.SystemPrompt) - appendix := strings.TrimSpace(result.Appendix) + appendix := strings.TrimSpace(workflow.Compiled.Appendix) if prompt == "" { aiAgent.SystemPrompt = appendix - return aiAgent, nil + return aiAgent, workflow, nil } aiAgent.SystemPrompt = prompt + "\n\n" + appendix - return aiAgent, nil + return aiAgent, workflow, nil } diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go new file mode 100644 index 0000000..97dbd8c --- /dev/null +++ b/internal/ai/runtime/workflow/executor.go @@ -0,0 +1,408 @@ +package workflow + +import ( + "context" + "encoding/json" + "fmt" + "reflect" + "strconv" + "strings" + + "agent-desk/internal/ai" + "agent-desk/internal/ai/runtime/internal/impl/retrievers" + "agent-desk/internal/ai/workflow/dsl" + workflowregistry "agent-desk/internal/ai/workflow/registry" + "agent-desk/internal/models" + "agent-desk/internal/pkg/utils" +) + +const maxWorkflowSteps = 128 + +type Input struct { + Definition dsl.Definition + Conversation models.Conversation + UserMessage models.Message + AIAgent models.AIAgent + AIConfig models.AIConfig +} + +type Result struct { + Status string + ReplyText string + NodePath []string + PromptTokens int + CompletionTokens int + RetrieverCount int + TraceData string +} + +type Executor struct{} + +func NewExecutor() *Executor { + return &Executor{} +} + +type runState struct { + input Input + nodesByID map[string]dsl.Node + outgoing map[string][]dsl.Edge + vars map[string]map[string]any + result Result +} + +func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) { + state := newRunState(input) + currentID := strings.TrimSpace(input.Definition.EntryNodeID) + if currentID == "" { + return nil, fmt.Errorf("workflow entry node is required") + } + for step := 0; step < maxWorkflowSteps; step++ { + node, ok := state.nodesByID[currentID] + if !ok { + return nil, fmt.Errorf("workflow node does not exist: %s", currentID) + } + state.result.NodePath = append(state.result.NodePath, node.ID) + if err := e.executeNode(ctx, state, node); err != nil { + return nil, err + } + if node.Type == workflowregistry.NodeTypeEnd { + state.result.Status = "completed" + return &state.result, nil + } + nextID, ok, err := state.nextNodeID(node.ID) + if err != nil { + return nil, err + } + if !ok { + state.result.Status = "completed" + return &state.result, nil + } + currentID = nextID + } + return nil, fmt.Errorf("workflow exceeded max steps") +} + +func newRunState(input Input) *runState { + state := &runState{ + input: input, + nodesByID: make(map[string]dsl.Node, len(input.Definition.Nodes)), + outgoing: make(map[string][]dsl.Edge), + vars: make(map[string]map[string]any), + result: Result{ + Status: "started", + NodePath: make([]string, 0), + }, + } + for _, node := range input.Definition.Nodes { + node.ID = strings.TrimSpace(node.ID) + node.Type = strings.TrimSpace(node.Type) + if node.ID != "" { + state.nodesByID[node.ID] = node + } + } + for _, edge := range input.Definition.Edges { + state.outgoing[edge.Source] = append(state.outgoing[edge.Source], edge) + } + return state +} + +func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.Node) error { + switch node.Type { + case workflowregistry.NodeTypeStart: + state.setNodeVars(node.ID, map[string]any{ + "conversationId": state.input.Conversation.ID, + "messageId": state.input.UserMessage.ID, + "aiAgentId": state.input.AIAgent.ID, + "userMessage": strings.TrimSpace(state.input.UserMessage.Content), + "knowledgeBaseIds": utils.SplitInt64s(state.input.AIAgent.KnowledgeIDs), + "conversationState": state.input.Conversation.Status, + }) + case workflowregistry.NodeTypeKnowledgeRetrieve: + return e.executeKnowledgeRetrieve(ctx, state, node) + case workflowregistry.NodeTypeAnswerabilityGate: + return e.executeAnswerabilityGate(state, node) + case workflowregistry.NodeTypeCondition: + state.setNodeVars(node.ID, map[string]any{"matched": true}) + case workflowregistry.NodeTypeLLMReply: + return e.executeLLMReply(ctx, state, node) + case workflowregistry.NodeTypeSendReply: + replyText := strings.TrimSpace(toString(state.resolveInput(node, "replyText"))) + state.result.ReplyText = replyText + state.setNodeVars(node.ID, map[string]any{ + "sent": replyText != "", + "replyMessageId": int64(0), + }) + case workflowregistry.NodeTypeHandoffToHuman: + reason := strings.TrimSpace(toString(state.resolveInput(node, "reason"))) + replyText := strings.TrimSpace(readStringConfig(node.Config, "replyText")) + if replyText == "" { + replyText = "已为你转接人工客服，请稍候。" + } + state.result.ReplyText = replyText + state.setNodeVars(node.ID, map[string]any{ + "handoffId": int64(0), + "reason": reason, + }) + case workflowregistry.NodeTypeEnd: + state.setNodeVars(node.ID, map[string]any{"status": "completed"}) + default: + return fmt.Errorf("unsupported workflow node type: %s", node.Type) + } + return nil +} + +func (e *Executor) executeKnowledgeRetrieve(ctx context.Context, state *runState, node dsl.Node) error { + query := strings.TrimSpace(toString(state.resolveInput(node, "query"))) + retriever := retrievers.NewKnowledgeRetriever(state.input.AIAgent) + result, err := retriever.RetrieveContext(ctx, query) + if err != nil { + return err + } + items := make([]map[string]any, 0, len(result.ContextResults)) + for _, item := range result.ContextResults { + items = append(items, map[string]any{ + "knowledgeBaseId": item.KnowledgeBaseID, + "documentId": item.DocumentID, + "chunkId": item.ChunkID, + "content": item.Content, + "score": item.Score, + }) + } + state.result.RetrieverCount = len(result.Hits) + state.setNodeVars(node.ID, map[string]any{ + "items": items, + "summary": result.ContextText, + }) + return nil +} + +func (e *Executor) executeAnswerabilityGate(state *runState, node dsl.Node) error { + items := state.resolveInput(node, "knowledgeItems") + answerability := "unanswerable" + reason := "no retrieved knowledge items" + if hasItems(items) { + answerability = "answerable" + reason = "retrieved knowledge items are available" + } + state.setNodeVars(node.ID, map[string]any{ + "answerability": answerability, + "reason": reason, + }) + return nil +} + +func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node dsl.Node) error { + if staticReply := strings.TrimSpace(readStringConfig(node.Config, "staticReply")); staticReply != "" { + state.setNodeVars(node.ID, map[string]any{"replyText": staticReply}) + return nil + } + userPrompt := strings.TrimSpace(toString(state.resolveInput(node, "userMessage"))) + if userPrompt == "" { + userPrompt = strings.TrimSpace(state.input.UserMessage.Content) + } + knowledgeItems := toString(state.resolveInput(node, "knowledgeItems")) + systemPrompt := strings.TrimSpace(state.input.AIAgent.SystemPrompt) + if prompt := strings.TrimSpace(readStringConfig(node.Config, "prompt")); prompt != "" { + systemPrompt = strings.TrimSpace(systemPrompt + "\n\n" + prompt) + } + if knowledgeItems != "" { + userPrompt = userPrompt + "\n\nKnowledge context:\n" + knowledgeItems + } + result, err := ai.LLM.ChatWithConfig(ctx, state.input.AIConfig, systemPrompt, userPrompt) + if err != nil { + return err + } + state.result.PromptTokens += result.PromptTokens + state.result.CompletionTokens += result.CompletionTokens + state.setNodeVars(node.ID, map[string]any{"replyText": result.Content}) + return nil +} + +func (s *runState) nextNodeID(sourceNodeID string) (string, bool, error) { + edges := s.outgoing[sourceNodeID] + if len(edges) == 0 { + return "", false, nil + } + for _, edge := range edges { + if edge.Condition == nil { + continue + } + matched, err := s.evaluateCondition(edge.Condition) + if err != nil { + return "", false, err + } + if matched { + return strings.TrimSpace(edge.Target), true, nil + } + } + for _, edge := range edges { + if edge.Condition == nil { + return strings.TrimSpace(edge.Target), true, nil + } + } + return "", false, nil +} + +func (s *runState) evaluateCondition(condition *dsl.Condition) (bool, error) { + if condition == nil { + return true, nil + } + left := s.resolveSelector(condition.Left) + operator := strings.TrimSpace(condition.Operator) + if operator == "" && strings.TrimSpace(condition.Expression) != "" { + return false, fmt.Errorf("free-form workflow condition expressions are not supported") + } + switch operator { + case "eq", "equals": + return compareString(left, condition.Right) == 0, nil + case "neq", "not_equals": + return compareString(left, condition.Right) != 0, nil + case "contains": + return strings.Contains(toString(left), toString(condition.Right)), nil + case "exists": + return exists(left), nil + case "not_exists": + return !exists(left), nil + case "truthy", "is_true": + return truthy(left), nil + case "falsy", "is_false": + return !truthy(left), nil + case "gt": + return compareNumber(left, condition.Right) > 0, nil + case "gte": + return compareNumber(left, condition.Right) >= 0, nil + case "lt": + return compareNumber(left, condition.Right) < 0, nil + case "lte": + return compareNumber(left, condition.Right) <= 0, nil + default: + return false, fmt.Errorf("unsupported workflow condition operator: %s", operator) + } +} + +func (s *runState) setNodeVars(nodeID string, values map[string]any) { + s.vars[nodeID] = values +} + +func (s *runState) resolveInput(node dsl.Node, inputName string) any { + selector, ok := node.Inputs[inputName] + if !ok { + return nil + } + return s.resolveSelector(&selector) +} + +func (s *runState) resolveSelector(selector *dsl.VariableSelector) any { + if selector == nil { + return nil + } + fields := s.vars[strings.TrimSpace(selector.NodeID)] + if fields == nil { + return nil + } + return fields[strings.TrimSpace(selector.Field)] +} + +func readStringConfig(raw json.RawMessage, key string) string { + if len(raw) == 0 { + return "" + } + var cfg map[string]any + if err := json.Unmarshal(raw, &cfg); err != nil { + return "" + } + return toString(cfg[key]) +} + +func compareString(left any, right any) int { + return strings.Compare(toString(left), toString(right)) +} + +func compareNumber(left any, right any) int { + leftNum := toFloat(left) + rightNum := toFloat(right) + switch { + case leftNum > rightNum: + return 1 + case leftNum < rightNum: + return -1 + default: + return 0 + } +} + +func toString(value any) string { + switch v := value.(type) { + case nil: + return "" + case string: + return v + case fmt.Stringer: + return v.String() + case []map[string]any: + buf, _ := json.Marshal(v) + return string(buf) + default: + return strings.TrimSpace(fmt.Sprint(v)) + } +} + +func toFloat(value any) float64 { + switch v := value.(type) { + case int: + return float64(v) + case int64: + return float64(v) + case float64: + return v + case float32: + return float64(v) + case json.Number: + f, _ := v.Float64() + return f + case string: + f, _ := strconv.ParseFloat(strings.TrimSpace(v), 64) + return f + default: + return 0 + } +} + +func truthy(value any) bool { + switch v := value.(type) { + case nil: + return false + case bool: + return v + case string: + normalized := strings.ToLower(strings.TrimSpace(v)) + return normalized != "" && normalized != "false" && normalized != "0" + default: + return !reflect.ValueOf(value).IsZero() + } +} + +func exists(value any) bool { + if value == nil { + return false + } + switch v := value.(type) { + case string: + return strings.TrimSpace(v) != "" + default: + return true + } +} + +func hasItems(value any) bool { + if value == nil { + return false + } + rv := reflect.ValueOf(value) + switch rv.Kind() { + case reflect.Array, reflect.Slice, reflect.Map: + return rv.Len() > 0 + default: + return exists(value) + } +} diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go new file mode 100644 index 0000000..596c1e7 --- /dev/null +++ b/internal/ai/runtime/workflow/executor_test.go @@ -0,0 +1,94 @@ +package workflow + +import ( + "context" + "testing" + + "agent-desk/internal/ai/workflow/dsl" + workflowregistry "agent-desk/internal/ai/workflow/registry" + "agent-desk/internal/models" +) + +func TestExecutorRoutesByConditionEdge(t *testing.T) { + executor := NewExecutor() + result, err := executor.Execute(context.Background(), Input{ + Definition: conditionalReplyDefinition(), + UserMessage: models.Message{ + Content: "vip", + }, + }) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + if result.ReplyText != "VIP reply" { + t.Fatalf("unexpected reply: %q", result.ReplyText) + } + assertPath(t, result.NodePath, []string{"start_1", "condition_1", "vip_reply", "send_vip", "end_1"}) +} + +func TestExecutorUsesDefaultEdgeWhenConditionDoesNotMatch(t *testing.T) { + executor := NewExecutor() + result, err := executor.Execute(context.Background(), Input{ + Definition: conditionalReplyDefinition(), + UserMessage: models.Message{ + Content: "normal", + }, + }) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + if result.ReplyText != "Normal reply" { + t.Fatalf("unexpected reply: %q", result.ReplyText) + } + assertPath(t, result.NodePath, []string{"start_1", "condition_1", "normal_reply", "send_normal", "end_1"}) +} + +func conditionalReplyDefinition() dsl.Definition { + return dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, + {ID: "condition_1", Type: workflowregistry.NodeTypeCondition, Name: "Route"}, + {ID: "vip_reply", Type: workflowregistry.NodeTypeLLMReply, Name: "VIP", Config: []byte(`{"staticReply":"VIP reply"}`)}, + {ID: "normal_reply", Type: workflowregistry.NodeTypeLLMReply, Name: "Normal", Config: []byte(`{"staticReply":"Normal reply"}`)}, + {ID: "send_vip", Type: workflowregistry.NodeTypeSendReply, Name: "Send VIP", Inputs: map[string]dsl.VariableSelector{ + "replyText": {NodeID: "vip_reply", Field: "replyText"}, + }}, + {ID: "send_normal", Type: workflowregistry.NodeTypeSendReply, Name: "Send Normal", Inputs: map[string]dsl.VariableSelector{ + "replyText": {NodeID: "normal_reply", Field: "replyText"}, + }}, + {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_condition", Source: "start_1", Target: "condition_1"}, + { + ID: "edge_condition_vip", + Source: "condition_1", + Target: "vip_reply", + Condition: &dsl.Condition{ + Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, + Operator: "eq", + Right: "vip", + }, + }, + {ID: "edge_condition_default", Source: "condition_1", Target: "normal_reply"}, + {ID: "edge_vip_send", Source: "vip_reply", Target: "send_vip"}, + {ID: "edge_normal_send", Source: "normal_reply", Target: "send_normal"}, + {ID: "edge_send_vip_end", Source: "send_vip", Target: "end_1"}, + {ID: "edge_send_normal_end", Source: "send_normal", Target: "end_1"}, + }, + } +} + +func assertPath(t *testing.T, got []string, want []string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("unexpected path length: got %#v want %#v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("unexpected path: got %#v want %#v", got, want) + } + } +} diff --git a/internal/ai/workflow/validator/validator.go b/internal/ai/workflow/validator/validator.go index 5ce8ef1..8daebb1 100644 --- a/internal/ai/workflow/validator/validator.go +++ b/internal/ai/workflow/validator/validator.go @@ -56,6 +56,7 @@ func (v *definitionValidator) validate() { v.validateReachability() v.validateConfirmationGuards() v.validateVariableMappings() + v.validateConditions() } func (v *definitionValidator) validateNodes() { @@ -250,6 +251,80 @@ func (v *definitionValidator) validateInputSelector(nodeID string, input registr } } +func (v *definitionValidator) validateConditions() { + conditionalSources := make(map[string]bool) + defaultSources := make(map[string]bool) + for index, edge := range v.def.Edges { + field := fmt.Sprintf("edges[%d].condition", index) + sourceID := strings.TrimSpace(edge.Source) + if edge.Condition == nil { + if sourceID != "" { + defaultSources[sourceID] = true + } + continue + } + if sourceID != "" { + conditionalSources[sourceID] = true + } + v.validateCondition(field, sourceID, edge.Condition) + } + for sourceID := range conditionalSources { + if !defaultSources[sourceID] { + v.addError("edges."+sourceID, "conditional branch must include a default edge") + } + } +} + +func (v *definitionValidator) validateCondition(field string, sourceNodeID string, condition *dsl.Condition) { + if condition == nil { + return + } + operator := strings.TrimSpace(condition.Operator) + if operator == "" && strings.TrimSpace(condition.Expression) != "" { + v.addError(field+".expression", "free-form condition expressions are not supported") + return + } + if !isSupportedConditionOperator(operator) { + v.addError(field+".operator", "unsupported condition operator: "+operator) + return + } + if condition.Left == nil { + v.addError(field+".left", "condition left variable is required") + return + } + sourceSelectorNodeID := strings.TrimSpace(condition.Left.NodeID) + sourceField := strings.TrimSpace(condition.Left.Field) + if sourceSelectorNodeID == "" || sourceField == "" { + v.addError(field+".left", "condition left variable is required") + return + } + sourceNode, ok := v.nodesByID[sourceSelectorNodeID] + if !ok { + v.addError(field+".left", "condition source node does not exist: "+sourceSelectorNodeID) + return + } + if sourceNodeID != "" && !v.hasPath(sourceSelectorNodeID, sourceNodeID, make(map[string]struct{})) && sourceSelectorNodeID != sourceNodeID { + v.addError(field+".left", "condition source node is not available before branch: "+sourceSelectorNodeID) + return + } + sourceSpec, ok := v.registry.Get(sourceNode.Type) + if !ok { + return + } + if _, ok := findOutputSpec(sourceSpec.OutputSchema, sourceField); !ok { + v.addError(field+".left", "condition source field does not exist: "+sourceSelectorNodeID+"."+sourceField) + } +} + +func isSupportedConditionOperator(operator string) bool { + switch strings.TrimSpace(operator) { + case "eq", "equals", "neq", "not_equals", "contains", "exists", "not_exists", "truthy", "is_true", "falsy", "is_false", "gt", "gte", "lt", "lte": + return true + default: + return false + } +} + func (v *definitionValidator) hasPath(sourceID string, targetID string, visiting map[string]struct{}) bool { if sourceID == targetID { return false diff --git a/internal/ai/workflow/validator/validator_test.go b/internal/ai/workflow/validator/validator_test.go index 6435cda..9d3928f 100644 --- a/internal/ai/workflow/validator/validator_test.go +++ b/internal/ai/workflow/validator/validator_test.go @@ -194,6 +194,34 @@ func TestValidateDefinitionAcceptsMappedKnowledgeFlow(t *testing.T) { } } +func TestValidateDefinitionRejectsUnknownConditionOperator(t *testing.T) { + def := conditionDefinition() + def.Edges[1].Condition.Operator = "regex" + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected unknown condition operator to be invalid") + } + if !hasValidationMessage(result, "unsupported condition operator") { + t.Fatalf("expected condition operator error, got %#v", result.Errors) + } +} + +func TestValidateDefinitionRejectsUnknownConditionVariable(t *testing.T) { + def := conditionDefinition() + def.Edges[1].Condition.Left.Field = "missing" + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected unknown condition variable to be invalid") + } + if !hasValidationMessage(result, "condition source field does not exist") { + t.Fatalf("expected condition variable error, got %#v", result.Errors) + } +} + func minimalDefinition() dsl.Definition { return dsl.Definition{ SchemaVersion: 1, @@ -220,6 +248,32 @@ func mappedReplyDefinition() dsl.Definition { return def } +func conditionDefinition() dsl.Definition { + return dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: "start"}, + {ID: "condition_1", Type: "condition"}, + {ID: "end_1", Type: "end"}, + }, + Edges: []dsl.Edge{ + {ID: "e1", Source: "start_1", Target: "condition_1"}, + { + ID: "e2", + Source: "condition_1", + Target: "end_1", + Condition: &dsl.Condition{ + Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, + Operator: "eq", + Right: "hello", + }, + }, + {ID: "e3", Source: "condition_1", Target: "end_1"}, + }, + } +} + func hasValidationMessage(result validator.Result, want string) bool { for _, item := range result.Errors { if strings.Contains(item.Message, want) { diff --git a/web/app/dashboard/ai-workflows/_components/workflow-editor.tsx b/web/app/dashboard/ai-workflows/_components/workflow-editor.tsx index a0eea23..641205f 100644 --- a/web/app/dashboard/ai-workflows/_components/workflow-editor.tsx +++ b/web/app/dashboard/ai-workflows/_components/workflow-editor.tsx @@ -30,6 +30,9 @@ import { useCallback, useEffect, useMemo, useRef, useState } from "react" import { Badge } from "@/components/ui/badge" import { Button } from "@/components/ui/button" +import { Input } from "@/components/ui/input" +import { Label } from "@/components/ui/label" +import { OptionCombobox } from "@/components/option-combobox" import { ScrollArea } from "@/components/ui/scroll-area" import { Popover, @@ -47,10 +50,13 @@ import { getRequiredInputs, toApiDefinition, validateWorkflowDraft, + type WorkflowVariableRef, + type WorkflowVariableSelector, type WorkflowEditorEdge, type WorkflowEditorNode, } from "./workflow-utils" import { NodeConfigPanel } from "./node-config-panel" +import { VariableSelector } from "./variable-selector" type WorkflowNodeData = Record & { nodeType?: string @@ -67,6 +73,7 @@ type WorkflowNodeData = Record & { type WorkflowFlowNode = Node type WorkflowFlowEdge = Edge +type WorkflowEdgeCondition = NonNullable["condition"] type PendingNodeDrag = { spec: AIWorkflowNodeSpec @@ -117,6 +124,7 @@ function toFlowEdges(definition: AIWorkflowDefinition): WorkflowFlowEdge[] { id: edge.id, source: edge.source, target: edge.target, + label: edge.condition ? "条件" : undefined, data: edge.condition ? { condition: edge.condition } : undefined, })) } @@ -167,6 +175,8 @@ export function WorkflowEditor({ const [nodeLibraryResizing, setNodeLibraryResizing] = useState(false) const [pendingNodeDrag, setPendingNodeDrag] = useState(null) const [propertyPanelNode, setPropertyPanelNode] = useState(null) + const [selectedEdgeId, setSelectedEdgeId] = useState(null) + const [propertyPanelEdge, setPropertyPanelEdge] = useState(null) const [propertyPanelVisible, setPropertyPanelVisible] = useState(false) const editorRef = useRef(null) const canvasRef = useRef(null) @@ -176,6 +186,10 @@ export function WorkflowEditor({ () => nodes.find((node) => node.id === selectedNodeId) ?? null, [nodes, selectedNodeId] ) + const selectedEdge = useMemo( + () => edges.find((edge) => edge.id === selectedEdgeId) ?? null, + [edges, selectedEdgeId] + ) const draft = useMemo(() => toDraft(nodes, edges), [nodes, edges]) const validation = useMemo( () => validateWorkflowDraft(draft, nodeSpecs), @@ -193,6 +207,10 @@ export function WorkflowEditor({ () => (propertyPanelNode ? getAvailableVariables(draft, propertyPanelNode.id, nodeSpecs) : []), [draft, nodeSpecs, propertyPanelNode] ) + const propertyPanelEdgeVariables = useMemo( + () => (propertyPanelEdge ? getEdgeConditionVariables(draft, propertyPanelEdge.source, nodeSpecs) : []), + [draft, nodeSpecs, propertyPanelEdge] + ) useEffect(() => { onDefinitionChange(toApiDefinition(draft) as AIWorkflowDefinition) @@ -219,14 +237,24 @@ export function WorkflowEditor({ useEffect(() => { if (selectedNode) { setPropertyPanelNode(selectedNode) + setPropertyPanelEdge(null) + window.setTimeout(() => setPropertyPanelVisible(true), 0) + return + } + if (selectedEdge) { + setPropertyPanelEdge(selectedEdge) + setPropertyPanelNode(null) window.setTimeout(() => setPropertyPanelVisible(true), 0) return } setPropertyPanelVisible(false) - const timer = window.setTimeout(() => setPropertyPanelNode(null), 220) + const timer = window.setTimeout(() => { + setPropertyPanelNode(null) + setPropertyPanelEdge(null) + }, 220) return () => window.clearTimeout(timer) - }, [selectedNode]) + }, [selectedNode, selectedEdge]) const onConnect = useCallback( (connection: Connection) => { @@ -365,6 +393,20 @@ export function WorkflowEditor({ ) } + const updateEdgeCondition = (edgeId: string, condition?: WorkflowEdgeCondition) => { + setEdges((current) => + current.map((edge) => + edge.id === edgeId + ? { + ...edge, + label: condition ? "条件" : undefined, + data: condition ? { ...(edge.data as object), condition } : undefined, + } + : edge + ) + ) + } + const clampNodeLibraryWidth = useCallback((width: number) => { const containerWidth = editorRef.current?.getBoundingClientRect().width ?? 0 const maxWidth = containerWidth > 0 ? containerWidth * 0.34 : 520 @@ -513,9 +555,17 @@ export function WorkflowEditor({ onNodeClick={(event, node) => { event.stopPropagation() setSelectedNodeId(node.id) + setSelectedEdgeId(null) + }} + onEdgeClick={(event, edge) => { + event.stopPropagation() + setSelectedNodeId(null) + setSelectedEdgeId(edge.id) + }} + onPaneClick={() => { + setSelectedNodeId(null) + setSelectedEdgeId(null) }} - onEdgeClick={() => setSelectedNodeId(null)} - onPaneClick={() => setSelectedNodeId(null)} fitView fitViewOptions={fitViewOptions} minZoom={0.45} @@ -526,7 +576,7 @@ export function WorkflowEditor({ - {propertyPanelNode ? ( + {propertyPanelNode || propertyPanelEdge ? (