feat: enhance condition node functionality with branches and validation

- Introduced WorkflowConditionBranch and WorkflowNodeConfig types to manage condition branches in nodes.
- Updated NodeConfigPanel and ConditionNodePanel components to support adding, editing, and deleting branches.
- Implemented validation for condition nodes to ensure at least one branch exists and that default branches are correctly configured.
- Modified workflow-utils to handle condition branches and updated toApiDefinition and fromApiDefinition functions to maintain branch integrity during serialization.
- Removed edge condition handling from WorkflowEditor and related components, simplifying edge management.
- Added tests to ensure condition branches are preserved in API definitions.
This commit is contained in:
mlogclub
2026-06-25 21:08:59 +08:00
parent 072c2f0105
commit b806a2208b
12 changed files with 745 additions and 511 deletions
+55 -26
View File
@@ -86,6 +86,8 @@ type workflowCheckPoint struct {
type branchDecision struct {
SelectedEdgeID string `json:"selectedEdgeId,omitempty"`
SelectedBranchID string `json:"selectedBranchId,omitempty"`
SelectedBranchName string `json:"selectedBranchName,omitempty"`
SelectedTargetNodeID string `json:"selectedTargetNodeId,omitempty"`
Reason string `json:"reason"`
Evaluations []conditionEvaluation `json:"evaluations,omitempty"`
@@ -93,6 +95,8 @@ type branchDecision struct {
type conditionEvaluation struct {
EdgeID string `json:"edgeId"`
BranchID string `json:"branchId,omitempty"`
BranchName string `json:"branchName,omitempty"`
TargetNodeID string `json:"targetNodeId"`
SourceNodeID string `json:"sourceNodeId,omitempty"`
SourceField string `json:"sourceField,omitempty"`
@@ -588,53 +592,69 @@ func (s *runState) nextNodeID(sourceNodeID string) (string, bool, error) {
if len(edges) == 0 {
return "", false, nil
}
node := s.nodesByID[sourceNodeID]
if strings.TrimSpace(node.Type) != workflowregistry.NodeTypeCondition {
return strings.TrimSpace(edges[0].Target), true, nil
}
config := dsl.ConditionConfig{}
if len(node.Config) > 0 {
if err := json.Unmarshal(node.Config, &config); err != nil {
return "", false, fmt.Errorf("invalid condition node config: %w", err)
}
}
evaluations := make([]conditionEvaluation, 0)
for _, edge := range edges {
if edge.Condition == nil {
for _, branch := range config.Branches {
if branch.Default {
continue
}
matched, evaluation, err := s.evaluateCondition(edge)
matched, evaluation, err := s.evaluateConditionBranch(sourceNodeID, branch)
if err != nil {
return "", false, err
}
evaluations = append(evaluations, evaluation)
if matched {
targetNodeID := strings.TrimSpace(branch.TargetNodeID)
s.branchDecisions[sourceNodeID] = branchDecision{
SelectedEdgeID: strings.TrimSpace(edge.ID),
SelectedTargetNodeID: strings.TrimSpace(edge.Target),
Reason: "conditional edge matched",
SelectedEdgeID: s.edgeIDForTarget(sourceNodeID, targetNodeID),
SelectedBranchID: strings.TrimSpace(branch.ID),
SelectedBranchName: strings.TrimSpace(branch.Name),
SelectedTargetNodeID: targetNodeID,
Reason: "condition branch matched",
Evaluations: evaluations,
}
return strings.TrimSpace(edge.Target), true, nil
return targetNodeID, true, nil
}
}
for _, edge := range edges {
if edge.Condition == nil {
if len(evaluations) > 0 {
s.branchDecisions[sourceNodeID] = branchDecision{
SelectedEdgeID: strings.TrimSpace(edge.ID),
SelectedTargetNodeID: strings.TrimSpace(edge.Target),
Reason: "no conditional edge matched; selected default edge",
Evaluations: evaluations,
}
}
return strings.TrimSpace(edge.Target), true, nil
for _, branch := range config.Branches {
if !branch.Default {
continue
}
}
if len(evaluations) > 0 {
targetNodeID := strings.TrimSpace(branch.TargetNodeID)
s.branchDecisions[sourceNodeID] = branchDecision{
Reason: "no conditional edge matched and no default edge exists",
Evaluations: evaluations,
SelectedEdgeID: s.edgeIDForTarget(sourceNodeID, targetNodeID),
SelectedBranchID: strings.TrimSpace(branch.ID),
SelectedBranchName: strings.TrimSpace(branch.Name),
SelectedTargetNodeID: targetNodeID,
Reason: "no condition branch matched; selected default branch",
Evaluations: evaluations,
}
return targetNodeID, true, nil
}
s.branchDecisions[sourceNodeID] = branchDecision{
Reason: "no condition branch matched and no default branch exists",
Evaluations: evaluations,
}
return "", false, nil
}
func (s *runState) evaluateCondition(edge dsl.Edge) (bool, conditionEvaluation, error) {
condition := edge.Condition
func (s *runState) evaluateConditionBranch(sourceNodeID string, branch dsl.ConditionBranch) (bool, conditionEvaluation, error) {
condition := branch.Condition
targetNodeID := strings.TrimSpace(branch.TargetNodeID)
evaluation := conditionEvaluation{
EdgeID: strings.TrimSpace(edge.ID),
TargetNodeID: strings.TrimSpace(edge.Target),
EdgeID: s.edgeIDForTarget(sourceNodeID, targetNodeID),
BranchID: strings.TrimSpace(branch.ID),
BranchName: strings.TrimSpace(branch.Name),
TargetNodeID: targetNodeID,
}
if condition == nil {
evaluation.Matched = true
@@ -683,6 +703,15 @@ func (s *runState) evaluateCondition(edge dsl.Edge) (bool, conditionEvaluation,
return matched, evaluation, nil
}
func (s *runState) edgeIDForTarget(sourceNodeID string, targetNodeID string) string {
for _, edge := range s.outgoing[sourceNodeID] {
if strings.TrimSpace(edge.Target) == targetNodeID {
return strings.TrimSpace(edge.ID)
}
}
return ""
}
func (s *runState) setNodeVars(nodeID string, values map[string]any) {
s.vars[nodeID] = values
}
+78 -69
View File
@@ -2,6 +2,7 @@ package workflow
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
@@ -18,7 +19,15 @@ import (
"gorm.io/gorm/schema"
)
func TestExecutorRoutesByConditionEdge(t *testing.T) {
func mustMarshalWorkflowTestConfig(value any) json.RawMessage {
raw, err := json.Marshal(value)
if err != nil {
panic(err)
}
return raw
}
func TestExecutorRoutesByConditionNodeBranch(t *testing.T) {
executor := NewExecutor()
result, err := executor.Execute(context.Background(), Input{
Definition: conditionalReplyDefinition(),
@@ -52,6 +61,7 @@ func TestExecutorConditionNodeTraceExplainsMatchedEdge(t *testing.T) {
}
for _, want := range []string{
`"selectedEdgeId":"edge_condition_vip"`,
`"selectedBranchId":"vip"`,
`"selectedTargetNodeId":"vip_reply"`,
`"operator":"eq"`,
`"leftValue":"vip"`,
@@ -80,8 +90,9 @@ func TestExecutorConditionNodeTraceExplainsDefaultEdge(t *testing.T) {
}
for _, want := range []string{
`"selectedEdgeId":"edge_condition_default"`,
`"selectedBranchId":"default"`,
`"selectedTargetNodeId":"normal_reply"`,
`"reason":"no conditional edge matched; selected default edge"`,
`"reason":"no condition branch matched; selected default branch"`,
`"leftValue":"normal"`,
`"matched":false`,
} {
@@ -129,7 +140,7 @@ func TestExecutorHandoffToHumanRunsRealDispatchAction(t *testing.T) {
if strings.TrimSpace(result.ReplyText) != "" {
t.Fatalf("expected workflow handoff node to avoid duplicate reply text, got %q", result.ReplyText)
}
assertPath(t, result.NodePath, []string{"start_1", "handoff_1", "assigned_end"})
assertPath(t, result.NodePath, []string{"start_1", "handoff_1", "handoff_route_1", "assigned_end"})
current := services.ConversationService.Get(conversation.ID)
if current.Status != enums.IMConversationStatusActive {
@@ -207,7 +218,7 @@ func TestExecutorAnalyzeConversationOutputsBranchVariables(t *testing.T) {
if err != nil {
t.Fatalf("execute workflow: %v", err)
}
assertPath(t, result.NodePath, []string{"start_1", "analyze_1", "handoff_end"})
assertPath(t, result.NodePath, []string{"start_1", "analyze_1", "analyze_route_1", "handoff_end"})
}
func TestExecutorPrepareTicketDraftOutputsDraftVariable(t *testing.T) {
@@ -225,7 +236,7 @@ func TestExecutorPrepareTicketDraftOutputsDraftVariable(t *testing.T) {
if err != nil {
t.Fatalf("execute workflow: %v", err)
}
assertPath(t, result.NodePath, []string{"start_1", "draft_1", "ready_end"})
assertPath(t, result.NodePath, []string{"start_1", "draft_1", "draft_route_1", "ready_end"})
}
func TestExecutorLLMReplyUsesAgentFallbackWhenDeclaredKnowledgeIsEmpty(t *testing.T) {
@@ -314,7 +325,7 @@ func TestExecutorResumeHumanConfirmContinuesWithConfirmedVariable(t *testing.T)
if result.Interrupted {
t.Fatalf("expected workflow resume to complete")
}
assertPath(t, result.NodePath, []string{"end_1"})
assertPath(t, result.NodePath, []string{"confirm_route_1", "end_1"})
}
func TestExecutorResumeCreatesTicketAfterHumanConfirmation(t *testing.T) {
@@ -349,7 +360,7 @@ func TestExecutorResumeCreatesTicketAfterHumanConfirmation(t *testing.T) {
if result.Interrupted {
t.Fatalf("expected workflow to complete")
}
assertPath(t, result.NodePath, []string{"create_ticket_1", "end_1"})
assertPath(t, result.NodePath, []string{"confirm_route_1", "create_ticket_1", "end_1"})
var ticket models.Ticket
if err := db.First(&ticket, "conversation_id = ?", conversation.ID).Error; err != nil {
@@ -403,7 +414,14 @@ func conditionalReplyDefinition() dsl.Definition {
EntryNodeID: "start_1",
Nodes: []dsl.Node{
{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
{ID: "condition_1", Type: workflowregistry.NodeTypeCondition, Name: "Route"},
{ID: "condition_1", Type: workflowregistry.NodeTypeCondition, Name: "Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "vip", Name: "VIP", TargetNodeID: "vip_reply", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
Operator: "eq",
Right: "vip",
}},
{ID: "default", Name: "Default", TargetNodeID: "normal_reply", Default: true},
}})},
{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{
@@ -416,16 +434,7 @@ func conditionalReplyDefinition() dsl.Definition {
},
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_vip", Source: "condition_1", Target: "vip_reply"},
{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"},
@@ -448,6 +457,13 @@ func createTicketWorkflowDefinition() dsl.Definition {
{ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{
"prompt": {NodeID: "prompt_1", Field: "replyText"},
}},
{ID: "confirm_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Confirm Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "yes", Name: "Yes", TargetNodeID: "create_ticket_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"},
Operator: "is_true",
}},
{ID: "default", Name: "Cancel", TargetNodeID: "cancel_end", Default: true},
}})},
{ID: "create_ticket_1", Type: workflowregistry.NodeTypeCreateTicket, Name: "Create Ticket", Inputs: map[string]dsl.VariableSelector{
"ticketDraft": {NodeID: "draft_1", Field: "ticketDraft"},
"confirmed": {NodeID: "confirm_1", Field: "confirmed"},
@@ -459,16 +475,9 @@ func createTicketWorkflowDefinition() dsl.Definition {
{ID: "edge_start_draft", Source: "start_1", Target: "draft_1"},
{ID: "edge_draft_prompt", Source: "draft_1", Target: "prompt_1"},
{ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"},
{
ID: "edge_confirm_create",
Source: "confirm_1",
Target: "create_ticket_1",
Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"},
Operator: "is_true",
},
},
{ID: "edge_confirm_cancel", Source: "confirm_1", Target: "cancel_end"},
{ID: "edge_confirm_route", Source: "confirm_1", Target: "confirm_route_1"},
{ID: "edge_confirm_create", Source: "confirm_route_1", Target: "create_ticket_1"},
{ID: "edge_confirm_cancel", Source: "confirm_route_1", Target: "cancel_end"},
{ID: "edge_create_end", Source: "create_ticket_1", Target: "end_1"},
},
}
@@ -484,22 +493,22 @@ func humanConfirmWorkflowDefinition() dsl.Definition {
{ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{
"prompt": {NodeID: "prompt_1", Field: "replyText"},
}},
{ID: "confirm_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Confirm Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "yes", Name: "Yes", TargetNodeID: "end_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"},
Operator: "is_true",
}},
{ID: "default", Name: "Cancel", TargetNodeID: "cancel_end", Default: true},
}})},
{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
{ID: "cancel_end", Type: workflowregistry.NodeTypeEnd, Name: "Cancel"},
},
Edges: []dsl.Edge{
{ID: "edge_start_prompt", Source: "start_1", Target: "prompt_1"},
{ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"},
{
ID: "edge_confirm_yes",
Source: "confirm_1",
Target: "end_1",
Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"},
Operator: "is_true",
},
},
{ID: "edge_confirm_cancel", Source: "confirm_1", Target: "cancel_end"},
{ID: "edge_confirm_route", Source: "confirm_1", Target: "confirm_route_1"},
{ID: "edge_confirm_yes", Source: "confirm_route_1", Target: "end_1"},
{ID: "edge_confirm_cancel", Source: "confirm_route_1", Target: "cancel_end"},
},
}
}
@@ -513,21 +522,21 @@ func prepareTicketDraftWorkflowDefinition() dsl.Definition {
{ID: "draft_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft", Inputs: map[string]dsl.VariableSelector{
"issue": {NodeID: "start_1", Field: "userMessage"},
}},
{ID: "draft_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Draft Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "ready", Name: "Ready", TargetNodeID: "ready_end", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "draft_1", Field: "ticketDraft"},
Operator: "exists",
}},
{ID: "default", Name: "Default", TargetNodeID: "default_end", Default: true},
}})},
{ID: "ready_end", Type: workflowregistry.NodeTypeEnd, Name: "Ready"},
{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
},
Edges: []dsl.Edge{
{ID: "edge_start_draft", Source: "start_1", Target: "draft_1"},
{
ID: "edge_draft_ready",
Source: "draft_1",
Target: "ready_end",
Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "draft_1", Field: "ticketDraft"},
Operator: "exists",
},
},
{ID: "edge_draft_default", Source: "draft_1", Target: "default_end"},
{ID: "edge_draft_route", Source: "draft_1", Target: "draft_route_1"},
{ID: "edge_draft_ready", Source: "draft_route_1", Target: "ready_end"},
{ID: "edge_draft_default", Source: "draft_route_1", Target: "default_end"},
},
}
}
@@ -541,21 +550,21 @@ func analyzeConversationWorkflowDefinition() dsl.Definition {
{ID: "analyze_1", Type: workflowregistry.NodeTypeAnalyzeConversation, Name: "Analyze", Inputs: map[string]dsl.VariableSelector{
"userMessage": {NodeID: "start_1", Field: "userMessage"},
}},
{ID: "analyze_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Analyze Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "handoff", Name: "Handoff", TargetNodeID: "handoff_end", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "analyze_1", Field: "needHumanHandoff"},
Operator: "is_true",
}},
{ID: "default", Name: "Default", TargetNodeID: "default_end", Default: true},
}})},
{ID: "handoff_end", Type: workflowregistry.NodeTypeEnd, Name: "Handoff"},
{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
},
Edges: []dsl.Edge{
{ID: "edge_start_analyze", Source: "start_1", Target: "analyze_1"},
{
ID: "edge_analyze_handoff",
Source: "analyze_1",
Target: "handoff_end",
Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "analyze_1", Field: "needHumanHandoff"},
Operator: "is_true",
},
},
{ID: "edge_analyze_default", Source: "analyze_1", Target: "default_end"},
{ID: "edge_analyze_route", Source: "analyze_1", Target: "analyze_route_1"},
{ID: "edge_analyze_handoff", Source: "analyze_route_1", Target: "handoff_end"},
{ID: "edge_analyze_default", Source: "analyze_route_1", Target: "default_end"},
},
}
}
@@ -569,22 +578,22 @@ func handoffWorkflowDefinition() dsl.Definition {
{ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff", Inputs: map[string]dsl.VariableSelector{
"reason": {NodeID: "start_1", Field: "userMessage"},
}},
{ID: "handoff_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Handoff Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "assigned", Name: "Assigned", TargetNodeID: "assigned_end", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "handoff_1", Field: "decision"},
Operator: "eq",
Right: string(services.HandoffDecisionAssigned),
}},
{ID: "default", Name: "Default", TargetNodeID: "default_end", Default: true},
}})},
{ID: "assigned_end", Type: workflowregistry.NodeTypeEnd, Name: "Assigned"},
{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
},
Edges: []dsl.Edge{
{ID: "edge_start_handoff", Source: "start_1", Target: "handoff_1"},
{
ID: "edge_handoff_assigned",
Source: "handoff_1",
Target: "assigned_end",
Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "handoff_1", Field: "decision"},
Operator: "eq",
Right: string(services.HandoffDecisionAssigned),
},
},
{ID: "edge_handoff_default", Source: "handoff_1", Target: "default_end"},
{ID: "edge_handoff_route", Source: "handoff_1", Target: "handoff_route_1"},
{ID: "edge_handoff_assigned", Source: "handoff_route_1", Target: "assigned_end"},
{ID: "edge_handoff_default", Source: "handoff_route_1", Target: "default_end"},
},
}
}
+15 -4
View File
@@ -24,10 +24,21 @@ type Position struct {
}
type Edge struct {
ID string `json:"id"`
Source string `json:"source"`
Target string `json:"target"`
Condition *Condition `json:"condition,omitempty"`
ID string `json:"id"`
Source string `json:"source"`
Target string `json:"target"`
}
type ConditionConfig struct {
Branches []ConditionBranch `json:"branches,omitempty"`
}
type ConditionBranch struct {
ID string `json:"id"`
Name string `json:"name,omitempty"`
TargetNodeID string `json:"targetNodeId"`
Condition *Condition `json:"condition,omitempty"`
Default bool `json:"default,omitempty"`
}
type Condition struct {
+58 -16
View File
@@ -1,6 +1,7 @@
package validator
import (
"encoding/json"
"fmt"
"strings"
@@ -268,31 +269,60 @@ 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
}
for index, node := range v.def.Nodes {
if strings.TrimSpace(node.Type) != registry.NodeTypeCondition {
continue
}
if sourceID != "" {
conditionalSources[sourceID] = true
field := fmt.Sprintf("nodes[%d].config.branches", index)
config := dsl.ConditionConfig{}
if len(node.Config) > 0 {
if err := json.Unmarshal(node.Config, &config); err != nil {
v.addError(field, "condition branches config must be valid JSON")
continue
}
}
v.validateCondition(field, sourceID, edge.Condition)
}
for sourceID := range conditionalSources {
if !defaultSources[sourceID] {
v.addError("edges."+sourceID, "conditional branch must include a default edge")
if len(config.Branches) == 0 {
v.addError(field, "condition node must include at least one branch")
continue
}
defaultCount := 0
seenBranchIDs := make(map[string]struct{}, len(config.Branches))
for branchIndex, branch := range config.Branches {
branchField := fmt.Sprintf("%s[%d]", field, branchIndex)
branchID := strings.TrimSpace(branch.ID)
if branchID == "" {
v.addError(branchField+".id", "condition branch id is required")
} else if _, exists := seenBranchIDs[branchID]; exists {
v.addError(branchField+".id", "duplicate condition branch id: "+branchID)
}
seenBranchIDs[branchID] = struct{}{}
targetNodeID := strings.TrimSpace(branch.TargetNodeID)
if targetNodeID == "" {
v.addError(branchField+".targetNodeId", "condition branch target node is required")
} else if _, ok := v.nodesByID[targetNodeID]; !ok {
v.addError(branchField+".targetNodeId", "condition branch target node does not exist: "+targetNodeID)
}
if !v.hasEdgeTo(strings.TrimSpace(node.ID), targetNodeID) {
v.addError(branchField+".targetNodeId", "condition branch target must have an outgoing edge: "+targetNodeID)
}
if branch.Default {
defaultCount++
if branch.Condition != nil {
v.addError(branchField+".condition", "default condition branch must not define a condition")
}
continue
}
v.validateCondition(branchField+".condition", strings.TrimSpace(node.ID), branch.Condition)
}
if defaultCount != 1 {
v.addError(field, "condition node must include exactly one default branch")
}
}
}
func (v *definitionValidator) validateCondition(field string, sourceNodeID string, condition *dsl.Condition) {
if condition == nil {
v.addError(field, "condition branch condition is required")
return
}
operator := strings.TrimSpace(condition.Operator)
@@ -360,6 +390,18 @@ func (v *definitionValidator) hasPath(sourceID string, targetID string, visiting
return false
}
func (v *definitionValidator) hasEdgeTo(sourceID string, targetID string) bool {
if sourceID == "" || targetID == "" {
return true
}
for _, edge := range v.def.Edges {
if strings.TrimSpace(edge.Source) == sourceID && strings.TrimSpace(edge.Target) == targetID {
return true
}
}
return false
}
func findInputSpec(items []registry.VariableSpec, name string) (registry.VariableSpec, bool) {
name = strings.TrimSpace(name)
for _, item := range items {
@@ -260,7 +260,16 @@ func TestValidateDefinitionAcceptsMappedKnowledgeFlow(t *testing.T) {
func TestValidateDefinitionRejectsUnknownConditionOperator(t *testing.T) {
def := conditionDefinition()
def.Edges[1].Condition.Operator = "regex"
var config dsl.ConditionConfig
if err := json.Unmarshal(def.Nodes[1].Config, &config); err != nil {
t.Fatalf("unmarshal condition config: %v", err)
}
config.Branches[0].Condition.Operator = "regex"
raw, err := json.Marshal(config)
if err != nil {
t.Fatalf("marshal condition config: %v", err)
}
def.Nodes[1].Config = raw
result := validator.ValidateDefinition(def, registry.DefaultRegistry())
@@ -274,7 +283,16 @@ func TestValidateDefinitionRejectsUnknownConditionOperator(t *testing.T) {
func TestValidateDefinitionRejectsUnknownConditionVariable(t *testing.T) {
def := conditionDefinition()
def.Edges[1].Condition.Left.Field = "missing"
var config dsl.ConditionConfig
if err := json.Unmarshal(def.Nodes[1].Config, &config); err != nil {
t.Fatalf("unmarshal condition config: %v", err)
}
config.Branches[0].Condition.Left.Field = "missing"
raw, err := json.Marshal(config)
if err != nil {
t.Fatalf("marshal condition config: %v", err)
}
def.Nodes[1].Config = raw
result := validator.ValidateDefinition(def, registry.DefaultRegistry())
@@ -313,26 +331,37 @@ func mappedReplyDefinition() dsl.Definition {
}
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"},
conditionConfig, _ := json.Marshal(dsl.ConditionConfig{
Branches: []dsl.ConditionBranch{
{
ID: "e2",
Source: "condition_1",
Target: "end_1",
ID: "hello",
Name: "Hello",
TargetNodeID: "end_1",
Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
Operator: "eq",
Right: "hello",
},
},
{
ID: "default",
Name: "Default",
TargetNodeID: "end_1",
Default: true,
},
},
})
return dsl.Definition{
SchemaVersion: 1,
EntryNodeID: "start_1",
Nodes: []dsl.Node{
{ID: "start_1", Type: "start"},
{ID: "condition_1", Type: "condition", Config: conditionConfig},
{ID: "end_1", Type: "end"},
},
Edges: []dsl.Edge{
{ID: "e1", Source: "start_1", Target: "condition_1"},
{ID: "e2", Source: "condition_1", Target: "end_1"},
{ID: "e3", Source: "condition_1", Target: "end_1"},
},
}
@@ -73,10 +73,10 @@ func TestAIAgentServiceCreatesDefaultWorkflow(t *testing.T) {
t.Fatalf("expected default workflow to include %s node: %#v", nodeType, stored.Nodes)
}
}
assertConditionEdgeToNodeType(t, stored, "route_intent_1", workflowregistry.NodeTypeHandoffToHuman, "contains", "人工")
assertConditionEdgeToNodeType(t, stored, "route_intent_1", workflowregistry.NodeTypePrepareTicketDraft, "contains", "工单")
assertConditionEdgeToNodeType(t, stored, "answerability_1", workflowregistry.NodeTypeLLMReply, "eq", "answerable")
assertDefaultEdgeToNodeType(t, stored, "answerability_1", workflowregistry.NodeTypeLLMReply)
assertConditionBranchToNodeType(t, stored, "route_intent_1", workflowregistry.NodeTypeHandoffToHuman, "contains", "人工")
assertConditionBranchToNodeType(t, stored, "route_intent_1", workflowregistry.NodeTypePrepareTicketDraft, "contains", "工单")
assertConditionBranchToNodeID(t, stored, "answerability_route_1", "reply_1", "eq", "answerable")
assertDefaultBranchToNodeID(t, stored, "answerability_route_1", "fallback_reply_1")
if !workflowEdgeExists(stored, "create_ticket_1", "ticket_result_reply_1") {
t.Fatalf("expected create_ticket to flow into a customer-visible result reply")
}
@@ -235,29 +235,57 @@ func nodeTypeByID(def dsl.Definition, nodeID string) string {
return ""
}
func assertConditionEdgeToNodeType(t *testing.T, def dsl.Definition, sourceID string, targetType string, operator string, right any) {
func assertConditionBranchToNodeType(t *testing.T, def dsl.Definition, sourceID string, targetType string, operator string, right any) {
t.Helper()
nodeTypes := workflowNodeTypeMap(def)
for _, edge := range def.Edges {
if edge.Source != sourceID || nodeTypes[edge.Target] != targetType || edge.Condition == nil {
for _, branch := range conditionBranches(t, def, sourceID) {
if nodeTypes[branch.TargetNodeID] != targetType || branch.Condition == nil {
continue
}
if edge.Condition.Operator == operator && edge.Condition.Right == right {
if branch.Condition.Operator == operator && branch.Condition.Right == right {
return
}
}
t.Fatalf("expected %s conditional edge from %s to %s with right=%v, got %#v", operator, sourceID, targetType, right, def.Edges)
t.Fatalf("expected %s condition branch from %s to %s with right=%v", operator, sourceID, targetType, right)
}
func assertDefaultEdgeToNodeType(t *testing.T, def dsl.Definition, sourceID string, targetType string) {
func assertConditionBranchToNodeID(t *testing.T, def dsl.Definition, sourceID string, targetID string, operator string, right any) {
t.Helper()
nodeTypes := workflowNodeTypeMap(def)
for _, edge := range def.Edges {
if edge.Source == sourceID && nodeTypes[edge.Target] == targetType && edge.Condition == nil {
for _, branch := range conditionBranches(t, def, sourceID) {
if branch.TargetNodeID != targetID || branch.Condition == nil {
continue
}
if branch.Condition.Operator == operator && branch.Condition.Right == right {
return
}
}
t.Fatalf("expected default edge from %s to %s, got %#v", sourceID, targetType, def.Edges)
t.Fatalf("expected %s condition branch from %s to %s with right=%v", operator, sourceID, targetID, right)
}
func assertDefaultBranchToNodeID(t *testing.T, def dsl.Definition, sourceID string, targetID string) {
t.Helper()
for _, branch := range conditionBranches(t, def, sourceID) {
if branch.TargetNodeID == targetID && branch.Default {
return
}
}
t.Fatalf("expected default branch from %s to %s", sourceID, targetID)
}
func conditionBranches(t *testing.T, def dsl.Definition, nodeID string) []dsl.ConditionBranch {
t.Helper()
for _, node := range def.Nodes {
if node.ID != nodeID {
continue
}
var config dsl.ConditionConfig
if err := json.Unmarshal(node.Config, &config); err != nil {
t.Fatalf("unmarshal condition config for %s: %v", nodeID, err)
}
return config.Branches
}
t.Fatalf("condition node not found: %s", nodeID)
return nil
}
func workflowEdgeExists(def dsl.Definition, sourceID string, targetID string) bool {
+31 -32
View File
@@ -427,7 +427,13 @@ func defaultAgentWorkflowDefinition() dsl.Definition {
EntryNodeID: "start_1",
Nodes: []dsl.Node{
{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "开始", Position: dsl.Position{X: 0, Y: 260}},
{ID: "route_intent_1", Type: workflowregistry.NodeTypeCondition, Name: "意图分流", Position: dsl.Position{X: 260, Y: 260}},
{ID: "route_intent_1", Type: workflowregistry.NodeTypeCondition, Name: "意图分流", Position: dsl.Position{X: 260, Y: 260}, Config: mustMarshalWorkflowConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "handoff", Name: "需要转人工", TargetNodeID: "handoff_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, Operator: "contains", Right: "人工"}},
{ID: "ticket", Name: "需要建单", TargetNodeID: "draft_ticket_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, Operator: "contains", Right: "工单"}},
{ID: "complaint", Name: "投诉建单", TargetNodeID: "draft_ticket_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, Operator: "contains", Right: "投诉"}},
{ID: "incident", Name: "报障建单", TargetNodeID: "draft_ticket_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, Operator: "contains", Right: "报障"}},
{ID: "default", Name: "默认知识库回复", TargetNodeID: "retrieve_1", Default: true},
}})},
{ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "转人工", Position: dsl.Position{X: 560, Y: 80}, Inputs: map[string]dsl.VariableSelector{
"reason": {NodeID: "start_1", Field: "userMessage"},
}},
@@ -441,6 +447,10 @@ func defaultAgentWorkflowDefinition() dsl.Definition {
{ID: "ticket_confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "确认建单", Position: dsl.Position{X: 1160, Y: 240}, Inputs: map[string]dsl.VariableSelector{
"prompt": {NodeID: "ticket_confirm_prompt_1", Field: "replyText"},
}},
{ID: "ticket_confirm_route_1", Type: workflowregistry.NodeTypeCondition, Name: "建单确认分流", Position: dsl.Position{X: 1310, Y: 240}, Config: mustMarshalWorkflowConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "confirmed", Name: "已确认", TargetNodeID: "create_ticket_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "ticket_confirm_1", Field: "confirmed"}, Operator: "is_true"}},
{ID: "default", Name: "取消或未确认", TargetNodeID: "ticket_cancel_reply_1", Default: true},
}})},
{ID: "create_ticket_1", Type: workflowregistry.NodeTypeCreateTicket, Name: "创建工单", Position: dsl.Position{X: 1460, Y: 180}, Inputs: map[string]dsl.VariableSelector{
"ticketDraft": {NodeID: "draft_ticket_1", Field: "ticketDraft"},
"confirmed": {NodeID: "ticket_confirm_1", Field: "confirmed"},
@@ -461,6 +471,10 @@ func defaultAgentWorkflowDefinition() dsl.Definition {
"userMessage": {NodeID: "start_1", Field: "userMessage"},
"knowledgeItems": {NodeID: "retrieve_1", Field: "items"},
}},
{ID: "answerability_route_1", Type: workflowregistry.NodeTypeCondition, Name: "可回答分流", Position: dsl.Position{X: 1010, Y: 500}, Config: mustMarshalWorkflowConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
{ID: "answerable", Name: "可以回答", TargetNodeID: "reply_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "answerability_1", Field: "answerability"}, Operator: "eq", Right: "answerable"}},
{ID: "default", Name: "兜底追问", TargetNodeID: "fallback_reply_1", Default: true},
}})},
{ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "AI 回复", Position: dsl.Position{X: 1160, Y: 440}, Inputs: map[string]dsl.VariableSelector{
"userMessage": {NodeID: "start_1", Field: "userMessage"},
"knowledgeItems": {NodeID: "retrieve_1", Field: "items"},
@@ -479,46 +493,23 @@ func defaultAgentWorkflowDefinition() dsl.Definition {
},
Edges: []dsl.Edge{
{ID: "edge_start_route_intent", Source: "start_1", Target: "route_intent_1"},
{ID: "edge_intent_handoff", Source: "route_intent_1", Target: "handoff_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
Operator: "contains",
Right: "人工",
}},
{ID: "edge_intent_ticket", Source: "route_intent_1", Target: "draft_ticket_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
Operator: "contains",
Right: "工单",
}},
{ID: "edge_intent_complaint", Source: "route_intent_1", Target: "draft_ticket_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
Operator: "contains",
Right: "投诉",
}},
{ID: "edge_intent_incident", Source: "route_intent_1", Target: "draft_ticket_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
Operator: "contains",
Right: "报障",
}},
{ID: "edge_intent_handoff", Source: "route_intent_1", Target: "handoff_1"},
{ID: "edge_intent_ticket", Source: "route_intent_1", Target: "draft_ticket_1"},
{ID: "edge_intent_knowledge_default", Source: "route_intent_1", Target: "retrieve_1"},
{ID: "edge_handoff_end", Source: "handoff_1", Target: "handoff_end_1"},
{ID: "edge_draft_ticket_confirm_prompt", Source: "draft_ticket_1", Target: "ticket_confirm_prompt_1"},
{ID: "edge_ticket_prompt_confirm", Source: "ticket_confirm_prompt_1", Target: "ticket_confirm_1"},
{ID: "edge_ticket_confirm_create", Source: "ticket_confirm_1", Target: "create_ticket_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "ticket_confirm_1", Field: "confirmed"},
Operator: "is_true",
}},
{ID: "edge_ticket_confirm_cancel", Source: "ticket_confirm_1", Target: "ticket_cancel_reply_1"},
{ID: "edge_ticket_confirm_route", Source: "ticket_confirm_1", Target: "ticket_confirm_route_1"},
{ID: "edge_ticket_confirm_create", Source: "ticket_confirm_route_1", Target: "create_ticket_1"},
{ID: "edge_ticket_confirm_cancel", Source: "ticket_confirm_route_1", Target: "ticket_cancel_reply_1"},
{ID: "edge_create_ticket_result", Source: "create_ticket_1", Target: "ticket_result_reply_1"},
{ID: "edge_ticket_result_end", Source: "ticket_result_reply_1", Target: "end_1"},
{ID: "edge_ticket_cancel_send", Source: "ticket_cancel_reply_1", Target: "send_ticket_cancel_1"},
{ID: "edge_ticket_cancel_end", Source: "send_ticket_cancel_1", Target: "end_1"},
{ID: "edge_retrieve_answerability", Source: "retrieve_1", Target: "answerability_1"},
{ID: "edge_answerability_reply", Source: "answerability_1", Target: "reply_1", Condition: &dsl.Condition{
Left: &dsl.VariableSelector{NodeID: "answerability_1", Field: "answerability"},
Operator: "eq",
Right: "answerable",
}},
{ID: "edge_answerability_fallback", Source: "answerability_1", Target: "fallback_reply_1"},
{ID: "edge_answerability_route", Source: "answerability_1", Target: "answerability_route_1"},
{ID: "edge_answerability_reply", Source: "answerability_route_1", Target: "reply_1"},
{ID: "edge_answerability_fallback", Source: "answerability_route_1", Target: "fallback_reply_1"},
{ID: "edge_reply_send", Source: "reply_1", Target: "send_1"},
{ID: "edge_fallback_send", Source: "fallback_reply_1", Target: "send_fallback_1"},
{ID: "edge_send_end", Source: "send_1", Target: "end_1"},
@@ -527,6 +518,14 @@ func defaultAgentWorkflowDefinition() dsl.Definition {
}
}
func mustMarshalWorkflowConfig(value any) json.RawMessage {
raw, err := json.Marshal(value)
if err != nil {
panic(err)
}
return raw
}
func defaultAgentWorkflowName(agentName string) string {
agentName = strings.TrimSpace(agentName)
if agentName == "" {