diff --git a/internal/ai/application/runtime/workflow_summary_test.go b/internal/ai/application/runtime/workflow_summary_test.go index 724504b..8aa2967 100644 --- a/internal/ai/application/runtime/workflow_summary_test.go +++ b/internal/ai/application/runtime/workflow_summary_test.go @@ -49,13 +49,12 @@ func TestToWorkflowSummaryPreservesInterruptCheckpoint(t *testing.T) { func TestPrepareWorkflowAgentDoesNotInjectWorkflowAppendix(t *testing.T) { db := setupWorkflowResumeTestDB(t) definitionJSON := mustMarshalDefinition(t, dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start", + SchemaVersion: 2, Nodes: []dsl.Node{ - {ID: "start", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "handoff", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff"}, + runtimeTestNode("start", workflowregistry.NodeTypeStart, "Start", nil, nil), + runtimeTestNode("handoff", workflowregistry.NodeTypeHandoffToHuman, "Handoff", nil, nil), }, - Edges: []dsl.Edge{{ID: "edge_start_handoff", Source: "start", Target: "handoff"}}, + Edges: []dsl.Edge{runtimeTestEdge("edge_start_handoff", "start", "handoff")}, }) version := models.AIWorkflowVersion{ WorkflowID: 1, @@ -219,14 +218,13 @@ func TestServiceResumeReusesInterruptedWorkflowRun(t *testing.T) { func TestServiceRunWritesFailedWorkflowRun(t *testing.T) { db := setupWorkflowResumeTestDB(t) def := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: 2, Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "bad_1", Type: "unsupported_node", Name: "Bad"}, + runtimeTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + runtimeTestNode("bad_1", "unsupported_node", "Bad", nil, nil), }, Edges: []dsl.Edge{ - {ID: "edge_start_bad", Source: "start_1", Target: "bad_1"}, + runtimeTestEdge("edge_start_bad", "start_1", "bad_1"), }, } version := models.AIWorkflowVersion{ @@ -335,24 +333,39 @@ func setupWorkflowResumeTestDB(t *testing.T) *gorm.DB { func runtimeHumanConfirmDefinition() dsl.Definition { return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: 2, Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认"}`)}, - {ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{ - "prompt": {NodeID: "prompt_1", Field: "replyText"}, - }}, - {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + runtimeTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + runtimeTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", []byte(`{"staticReply":"请确认"}`), nil), + runtimeTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", nil, map[string]dsl.Value{ + "prompt": dsl.RefValue("prompt_1", "replyText"), + }), + runtimeTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), }, 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_end", Source: "confirm_1", Target: "end_1"}, + runtimeTestEdge("edge_start_prompt", "start_1", "prompt_1"), + runtimeTestEdge("edge_prompt_confirm", "prompt_1", "confirm_1"), + runtimeTestEdge("edge_confirm_end", "confirm_1", "end_1"), }, } } +func runtimeTestNode(id string, nodeType string, title string, config []byte, inputs map[string]dsl.Value) dsl.Node { + return dsl.Node{ + ID: id, + Type: nodeType, + Data: dsl.NodeData{ + Title: title, + Config: config, + InputsValues: inputs, + }, + } +} + +func runtimeTestEdge(id string, source string, target string) dsl.Edge { + return dsl.Edge{SourceNodeID: source, TargetNodeID: target, SourcePortID: id} +} + func mustMarshalDefinition(t *testing.T, def dsl.Definition) string { t.Helper() buf, err := json.Marshal(def) diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go index b266e85..dacd607 100644 --- a/internal/ai/runtime/workflow/executor.go +++ b/internal/ai/runtime/workflow/executor.go @@ -112,7 +112,7 @@ type conditionEvaluation struct { func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) { state := newRunState(input) - currentID := strings.TrimSpace(input.Definition.EntryNodeID) + currentID := state.startNodeID() if currentID == "" { return nil, fmt.Errorf("workflow entry node is required") } @@ -247,11 +247,20 @@ func newRunState(input Input) *runState { } } for _, edge := range input.Definition.Edges { - state.outgoing[edge.Source] = append(state.outgoing[edge.Source], edge) + state.outgoing[strings.TrimSpace(edge.SourceNodeID)] = append(state.outgoing[strings.TrimSpace(edge.SourceNodeID)], edge) } return state } +func (s *runState) startNodeID() string { + for _, node := range s.nodesByID { + if strings.TrimSpace(node.Type) == workflowregistry.NodeTypeStart { + return strings.TrimSpace(node.ID) + } + } + return "" +} + func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.Node) error { switch node.Type { case workflowregistry.NodeTypeStart: @@ -587,19 +596,19 @@ func (e *Executor) executePrepareTicketDraft(ctx context.Context, state *runStat input := graphs.PrepareTicketDraftInput{ Issue: issue, } - if title := strings.TrimSpace(readStringConfig(node.Config, "title")); title != "" { + if title := strings.TrimSpace(readStringConfig(node.Data.Config, "title")); title != "" { input.Title = title } - if description := strings.TrimSpace(readStringConfig(node.Config, "description")); description != "" { + if description := strings.TrimSpace(readStringConfig(node.Data.Config, "description")); description != "" { input.Description = description } - if impact := strings.TrimSpace(readStringConfig(node.Config, "impact")); impact != "" { + if impact := strings.TrimSpace(readStringConfig(node.Data.Config, "impact")); impact != "" { input.Impact = impact } - if expectedOutcome := strings.TrimSpace(readStringConfig(node.Config, "expectedOutcome")); expectedOutcome != "" { + if expectedOutcome := strings.TrimSpace(readStringConfig(node.Data.Config, "expectedOutcome")); expectedOutcome != "" { input.ExpectedOutcome = expectedOutcome } - if currentAttempt := strings.TrimSpace(readStringConfig(node.Config, "currentAttempt")); currentAttempt != "" { + if currentAttempt := strings.TrimSpace(readStringConfig(node.Data.Config, "currentAttempt")); currentAttempt != "" { input.CurrentAttempt = currentAttempt } args, err := json.Marshal(input) @@ -632,20 +641,20 @@ func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runSta input := graphs.AnalyzeConversationInput{ ObservedIssue: userMessage, } - if strings.TrimSpace(readStringConfig(node.Config, "goal")) != "" { - input.Goal = strings.TrimSpace(readStringConfig(node.Config, "goal")) + if strings.TrimSpace(readStringConfig(node.Data.Config, "goal")) != "" { + input.Goal = strings.TrimSpace(readStringConfig(node.Data.Config, "goal")) } - if readBoolConfig(node.Config, "needTicket") { + if readBoolConfig(node.Data.Config, "needTicket") { input.NeedTicket = true } - if readBoolConfig(node.Config, "needHumanHandoff") { + if readBoolConfig(node.Data.Config, "needHumanHandoff") { input.NeedHumanHandoff = true } - if readBoolConfig(node.Config, "needQualityCheck") { + if readBoolConfig(node.Data.Config, "needQualityCheck") { input.NeedQualityCheck = true } - if strings.TrimSpace(readStringConfig(node.Config, "additionalContext")) != "" { - input.AdditionalContext = strings.TrimSpace(readStringConfig(node.Config, "additionalContext")) + if strings.TrimSpace(readStringConfig(node.Data.Config, "additionalContext")) != "" { + input.AdditionalContext = strings.TrimSpace(readStringConfig(node.Data.Config, "additionalContext")) } args, err := json.Marshal(input) if err != nil { @@ -670,7 +679,7 @@ func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runSta } func (e *Executor) executeHandoffToHuman(state *runState, node dsl.Node) error { - if _, hasConfirmedInput := node.Inputs["confirmed"]; hasConfirmedInput && !truthy(state.resolveInput(node, "confirmed")) { + if _, hasConfirmedInput := node.Data.InputsValues["confirmed"]; hasConfirmedInput && !truthy(state.resolveInput(node, "confirmed")) { state.setNodeVars(node.ID, map[string]any{ "handoffId": int64(0), "reason": strings.TrimSpace(toString(state.resolveInput(node, "reason"))), @@ -751,7 +760,7 @@ func (e *Executor) executeAnswerabilityGate(state *runState, node dsl.Node) erro } func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node dsl.Node) error { - if staticReply := strings.TrimSpace(readStringConfig(node.Config, "staticReply")); staticReply != "" { + if staticReply := strings.TrimSpace(readStringConfig(node.Data.Config, "staticReply")); staticReply != "" { state.setNodeVars(node.ID, map[string]any{"replyText": staticReply}) return nil } @@ -761,10 +770,10 @@ func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node ds } knowledgeItems := toString(state.resolveInput(node, "knowledgeItems")) systemPrompt := strings.TrimSpace(state.input.AIAgent.SystemPrompt) - if prompt := strings.TrimSpace(readStringConfig(node.Config, "prompt")); prompt != "" { + if prompt := strings.TrimSpace(readStringConfig(node.Data.Config, "prompt")); prompt != "" { systemPrompt = strings.TrimSpace(systemPrompt + "\n\n" + prompt) } - if _, declaresKnowledge := node.Inputs["knowledgeItems"]; declaresKnowledge && len(utils.SplitInt64s(state.input.AIAgent.KnowledgeIDs)) > 0 && !hasItems(state.resolveInput(node, "knowledgeItems")) { + if _, declaresKnowledge := node.Data.InputsValues["knowledgeItems"]; declaresKnowledge && len(utils.SplitInt64s(state.input.AIAgent.KnowledgeIDs)) > 0 && !hasItems(state.resolveInput(node, "knowledgeItems")) { state.setNodeVars(node.ID, map[string]any{"replyText": workflowKnowledgeFallbackReply(state.input.AIAgent)}) return nil } @@ -798,11 +807,11 @@ func (s *runState) nextNodeID(sourceNodeID string) (string, bool, error) { } node := s.nodesByID[sourceNodeID] if strings.TrimSpace(node.Type) != workflowregistry.NodeTypeCondition { - return strings.TrimSpace(edges[0].Target), true, nil + return strings.TrimSpace(edges[0].TargetNodeID), true, nil } config := dsl.ConditionConfig{} - if len(node.Config) > 0 { - if err := json.Unmarshal(node.Config, &config); err != nil { + if len(node.Data.Config) > 0 { + if err := json.Unmarshal(node.Data.Config, &config); err != nil { return "", false, fmt.Errorf("invalid condition node config: %w", err) } } @@ -864,11 +873,13 @@ func (s *runState) evaluateConditionBranch(sourceNodeID string, branch dsl.Condi evaluation.Matched = true return true, evaluation, nil } - left := s.resolveSelector(condition.Left) operator := strings.TrimSpace(condition.Operator) + var left any if condition.Left != nil { - evaluation.SourceNodeID = strings.TrimSpace(condition.Left.NodeID) - evaluation.SourceField = strings.TrimSpace(condition.Left.Field) + left = s.resolveValue(*condition.Left) + evaluation.SourceNodeID, evaluation.SourceField, _ = condition.Left.Ref() + evaluation.SourceNodeID = strings.TrimSpace(evaluation.SourceNodeID) + evaluation.SourceField = strings.TrimSpace(evaluation.SourceField) } evaluation.Operator = operator evaluation.LeftValue = left @@ -909,8 +920,11 @@ func (s *runState) evaluateConditionBranch(sourceNodeID string, branch dsl.Condi 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) + if strings.TrimSpace(edge.TargetNodeID) == targetNodeID { + if edge.SourcePortID != "" { + return strings.TrimSpace(edge.SourcePortID) + } + return strings.TrimSpace(edge.SourceNodeID + "->" + edge.TargetNodeID) } } return "" @@ -921,27 +935,27 @@ func (s *runState) setNodeVars(nodeID string, values map[string]any) { } func (s *runState) resolveInput(node dsl.Node, inputName string) any { - selector, ok := node.Inputs[inputName] + value, ok := node.Data.InputsValues[inputName] if !ok { return nil } - return s.resolveSelector(&selector) + return s.resolveValue(value) } func (s *runState) nodeInputPreview(node dsl.Node) map[string]any { - inputs := make(map[string]any, len(node.Inputs)) - for name, selector := range node.Inputs { - inputs[name] = s.resolveSelector(&selector) + inputs := make(map[string]any, len(node.Data.InputsValues)) + for name, value := range node.Data.InputsValues { + inputs[name] = s.resolveValue(value) } ret := map[string]any{ "inputs": inputs, } - if len(node.Config) > 0 { + if len(node.Data.Config) > 0 { var cfg any - if err := json.Unmarshal(node.Config, &cfg); err == nil { + if err := json.Unmarshal(node.Data.Config, &cfg); err == nil { ret["config"] = cfg } else { - ret["config"] = string(node.Config) + ret["config"] = string(node.Data.Config) } } return ret @@ -969,15 +983,28 @@ func workflowPreviewJSON(value any) string { return string(raw[:maxPreviewBytes]) } -func (s *runState) resolveSelector(selector *dsl.VariableSelector) any { - if selector == nil { +func (s *runState) resolveValue(value dsl.Value) any { + switch value.Type { + case dsl.ValueTypeRef: + nodeID, field, ok := value.Ref() + if !ok { + return nil + } + fields := s.vars[strings.TrimSpace(nodeID)] + if fields == nil { + return nil + } + return fields[strings.TrimSpace(field)] + case dsl.ValueTypeConstant: + return value.ConstantContent + case dsl.ValueTypeTemplate: + if len(value.Content) > 0 { + return value.Content[0] + } + return nil + default: 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 { diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go index b5729e2..9893d1f 100644 --- a/internal/ai/runtime/workflow/executor_test.go +++ b/internal/ai/runtime/workflow/executor_test.go @@ -435,286 +435,254 @@ func findNodeTrace(items []NodeTrace, nodeID string) *NodeTrace { } func emptyKnowledgeReplyDefinition() dsl.Definition { - return 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", Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - "knowledgeItems": {NodeID: "missing_retrieve", Field: "items"}, - }}, - {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"}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("reply_1", workflowregistry.NodeTypeLLMReply, "Reply", map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), + "knowledgeItems": dsl.RefValue("missing_retrieve", "items"), + }, nil), + wfTestNode("send_1", workflowregistry.NodeTypeSendReply, "Send", wfTestInputs("replyText", "reply_1", "replyText"), nil), + wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), }, - 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"}, + []dsl.Edge{ + wfTestEdge("start_1", "reply_1", "edge_start_reply"), + wfTestEdge("reply_1", "send_1", "edge_reply_send"), + wfTestEdge("send_1", "end_1", "edge_send_end"), }, - } + ) } func policyFirstWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "understanding_1", Type: workflowregistry.NodeTypeConversationUnderstanding, Name: "Understanding", Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "policy_1", Type: workflowregistry.NodeTypeReplyPolicy, Name: "Policy", Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - "messageIntent": {NodeID: "understanding_1", Field: "messageIntent"}, - "answerScope": {NodeID: "understanding_1", Field: "answerScope"}, - "riskSignals": {NodeID: "understanding_1", Field: "riskSignals"}, - "knowledgeItems": {NodeID: "retrieve_1", Field: "items"}, - }}, - {ID: "policy_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Policy Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ - {ID: "direct", Name: "Direct", TargetNodeID: "send_direct_1", Condition: &dsl.Condition{ - Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, - Operator: "eq", - Right: "direct_reply", - }}, - {ID: "knowledge", Name: "Knowledge", TargetNodeID: "retrieve_end", Condition: &dsl.Condition{ - Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, - Operator: "eq", - Right: "retrieve_knowledge", - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("understanding_1", workflowregistry.NodeTypeConversationUnderstanding, "Understanding", wfTestInputs("userMessage", "start_1", "userMessage"), nil), + wfTestNode("policy_1", workflowregistry.NodeTypeReplyPolicy, "Policy", map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), + "messageIntent": dsl.RefValue("understanding_1", "messageIntent"), + "answerScope": dsl.RefValue("understanding_1", "answerScope"), + "riskSignals": dsl.RefValue("understanding_1", "riskSignals"), + "knowledgeItems": dsl.RefValue("retrieve_1", "items"), + }, nil), + wfTestNode("policy_route_1", workflowregistry.NodeTypeCondition, "Policy Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("direct", "Direct", "send_direct_1", "policy_1", "action", "eq", "direct_reply"), + wfTestConditionBranch("knowledge", "Knowledge", "retrieve_end", "policy_1", "action", "eq", "retrieve_knowledge"), {ID: "default", Name: "Default", TargetNodeID: "end_1", Default: true}, - }})}, - {ID: "send_direct_1", Type: workflowregistry.NodeTypeSendReply, Name: "Send Direct", Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "policy_1", Field: "replyText"}, - }}, - {ID: "retrieve_end", Type: workflowregistry.NodeTypeEnd, Name: "Retrieve"}, - {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + }}), + wfTestNode("send_direct_1", workflowregistry.NodeTypeSendReply, "Send Direct", wfTestInputs("replyText", "policy_1", "replyText"), nil), + wfTestNode("retrieve_end", workflowregistry.NodeTypeEnd, "Retrieve", nil, nil), + wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), }, - Edges: []dsl.Edge{ - {ID: "edge_start_understanding", Source: "start_1", Target: "understanding_1"}, - {ID: "edge_understanding_policy", Source: "understanding_1", Target: "policy_1"}, - {ID: "edge_policy_route", Source: "policy_1", Target: "policy_route_1"}, - {ID: "edge_policy_direct", Source: "policy_route_1", Target: "send_direct_1"}, - {ID: "edge_policy_knowledge", Source: "policy_route_1", Target: "retrieve_end"}, - {ID: "edge_policy_default", Source: "policy_route_1", Target: "end_1"}, - {ID: "edge_send_direct_end", Source: "send_direct_1", Target: "end_1"}, + []dsl.Edge{ + wfTestEdge("start_1", "understanding_1", "edge_start_understanding"), + wfTestEdge("understanding_1", "policy_1", "edge_understanding_policy"), + wfTestEdge("policy_1", "policy_route_1", "edge_policy_route"), + wfTestEdge("policy_route_1", "send_direct_1", "edge_policy_direct"), + wfTestEdge("policy_route_1", "retrieve_end", "edge_policy_knowledge"), + wfTestEdge("policy_route_1", "end_1", "edge_policy_default"), + wfTestEdge("send_direct_1", "end_1", "edge_send_direct_end"), }, - } + ) } 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", 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", - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("condition_1", workflowregistry.NodeTypeCondition, "Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("vip", "VIP", "vip_reply", "start_1", "userMessage", "eq", "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{ - "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"}, + }}), + wfTestNode("vip_reply", workflowregistry.NodeTypeLLMReply, "VIP", nil, map[string]any{"staticReply": "VIP reply"}), + wfTestNode("normal_reply", workflowregistry.NodeTypeLLMReply, "Normal", nil, map[string]any{"staticReply": "Normal reply"}), + wfTestNode("send_vip", workflowregistry.NodeTypeSendReply, "Send VIP", wfTestInputs("replyText", "vip_reply", "replyText"), nil), + wfTestNode("send_normal", workflowregistry.NodeTypeSendReply, "Send Normal", wfTestInputs("replyText", "normal_reply", "replyText"), nil), + wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), }, - Edges: []dsl.Edge{ - {ID: "edge_start_condition", Source: "start_1", Target: "condition_1"}, - {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"}, - {ID: "edge_send_vip_end", Source: "send_vip", Target: "end_1"}, - {ID: "edge_send_normal_end", Source: "send_normal", Target: "end_1"}, + []dsl.Edge{ + wfTestEdge("start_1", "condition_1", "edge_start_condition"), + wfTestEdge("condition_1", "vip_reply", "edge_condition_vip"), + wfTestEdge("condition_1", "normal_reply", "edge_condition_default"), + wfTestEdge("vip_reply", "send_vip", "edge_vip_send"), + wfTestEdge("normal_reply", "send_normal", "edge_normal_send"), + wfTestEdge("send_vip", "end_1", "edge_send_vip_end"), + wfTestEdge("send_normal", "end_1", "edge_send_normal_end"), }, - } + ) } func createTicketWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "draft_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft", Inputs: map[string]dsl.VariableSelector{ - "issue": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认创建工单"}`)}, - {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", - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("draft_1", workflowregistry.NodeTypePrepareTicketDraft, "Draft", wfTestInputs("issue", "start_1", "userMessage"), nil), + wfTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", nil, map[string]any{"staticReply": "请确认创建工单"}), + wfTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", wfTestInputs("prompt", "prompt_1", "replyText"), nil), + wfTestNode("confirm_route_1", workflowregistry.NodeTypeCondition, "Confirm Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("yes", "Yes", "create_ticket_1", "confirm_1", "confirmed", "is_true", nil), {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"}, - }}, - {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, - {ID: "cancel_end", Type: workflowregistry.NodeTypeEnd, Name: "Cancel"}, + }}), + wfTestNode("create_ticket_1", workflowregistry.NodeTypeCreateTicket, "Create Ticket", map[string]dsl.Value{ + "ticketDraft": dsl.RefValue("draft_1", "ticketDraft"), + "confirmed": dsl.RefValue("confirm_1", "confirmed"), + }, nil), + wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), + wfTestNode("cancel_end", workflowregistry.NodeTypeEnd, "Cancel", nil, nil), }, - Edges: []dsl.Edge{ - {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_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"}, + []dsl.Edge{ + wfTestEdge("start_1", "draft_1", "edge_start_draft"), + wfTestEdge("draft_1", "prompt_1", "edge_draft_prompt"), + wfTestEdge("prompt_1", "confirm_1", "edge_prompt_confirm"), + wfTestEdge("confirm_1", "confirm_route_1", "edge_confirm_route"), + wfTestEdge("confirm_route_1", "create_ticket_1", "edge_confirm_create"), + wfTestEdge("confirm_route_1", "cancel_end", "edge_confirm_cancel"), + wfTestEdge("create_ticket_1", "end_1", "edge_create_end"), }, - } + ) } func humanConfirmWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认创建工单"}`)}, - {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", - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", nil, map[string]any{"staticReply": "请确认创建工单"}), + wfTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", wfTestInputs("prompt", "prompt_1", "replyText"), nil), + wfTestNode("confirm_route_1", workflowregistry.NodeTypeCondition, "Confirm Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("yes", "Yes", "end_1", "confirm_1", "confirmed", "is_true", nil), {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"}, + }}), + wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), + wfTestNode("cancel_end", workflowregistry.NodeTypeEnd, "Cancel", nil, nil), }, - 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_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"}, + []dsl.Edge{ + wfTestEdge("start_1", "prompt_1", "edge_start_prompt"), + wfTestEdge("prompt_1", "confirm_1", "edge_prompt_confirm"), + wfTestEdge("confirm_1", "confirm_route_1", "edge_confirm_route"), + wfTestEdge("confirm_route_1", "end_1", "edge_confirm_yes"), + wfTestEdge("confirm_route_1", "cancel_end", "edge_confirm_cancel"), }, - } + ) } func prepareTicketDraftWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {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", - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("draft_1", workflowregistry.NodeTypePrepareTicketDraft, "Draft", wfTestInputs("issue", "start_1", "userMessage"), nil), + wfTestNode("draft_route_1", workflowregistry.NodeTypeCondition, "Draft Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("ready", "Ready", "ready_end", "draft_1", "ticketDraft", "exists", nil), {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"}, + }}), + wfTestNode("ready_end", workflowregistry.NodeTypeEnd, "Ready", nil, nil), + wfTestNode("default_end", workflowregistry.NodeTypeEnd, "Default", nil, nil), }, - Edges: []dsl.Edge{ - {ID: "edge_start_draft", Source: "start_1", Target: "draft_1"}, - {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"}, + []dsl.Edge{ + wfTestEdge("start_1", "draft_1", "edge_start_draft"), + wfTestEdge("draft_1", "draft_route_1", "edge_draft_route"), + wfTestEdge("draft_route_1", "ready_end", "edge_draft_ready"), + wfTestEdge("draft_route_1", "default_end", "edge_draft_default"), }, - } + ) } func analyzeConversationWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {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", - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("analyze_1", workflowregistry.NodeTypeAnalyzeConversation, "Analyze", wfTestInputs("userMessage", "start_1", "userMessage"), nil), + wfTestNode("analyze_route_1", workflowregistry.NodeTypeCondition, "Analyze Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("handoff", "Handoff", "handoff_end", "analyze_1", "needHumanHandoff", "is_true", nil), {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"}, + }}), + wfTestNode("handoff_end", workflowregistry.NodeTypeEnd, "Handoff", nil, nil), + wfTestNode("default_end", workflowregistry.NodeTypeEnd, "Default", nil, nil), }, - Edges: []dsl.Edge{ - {ID: "edge_start_analyze", Source: "start_1", Target: "analyze_1"}, - {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"}, + []dsl.Edge{ + wfTestEdge("start_1", "analyze_1", "edge_start_analyze"), + wfTestEdge("analyze_1", "analyze_route_1", "edge_analyze_route"), + wfTestEdge("analyze_route_1", "handoff_end", "edge_analyze_handoff"), + wfTestEdge("analyze_route_1", "default_end", "edge_analyze_default"), }, - } + ) } func handoffWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {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), - }}, + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("handoff_1", workflowregistry.NodeTypeHandoffToHuman, "Handoff", wfTestInputs("reason", "start_1", "userMessage"), nil), + wfTestNode("handoff_route_1", workflowregistry.NodeTypeCondition, "Handoff Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + wfTestConditionBranch("assigned", "Assigned", "assigned_end", "handoff_1", "decision", "eq", 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"}, + }}), + wfTestNode("assigned_end", workflowregistry.NodeTypeEnd, "Assigned", nil, nil), + wfTestNode("default_end", workflowregistry.NodeTypeEnd, "Default", nil, nil), }, - Edges: []dsl.Edge{ - {ID: "edge_start_handoff", Source: "start_1", Target: "handoff_1"}, - {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"}, + []dsl.Edge{ + wfTestEdge("start_1", "handoff_1", "edge_start_handoff"), + wfTestEdge("handoff_1", "handoff_route_1", "edge_handoff_route"), + wfTestEdge("handoff_route_1", "assigned_end", "edge_handoff_assigned"), + wfTestEdge("handoff_route_1", "default_end", "edge_handoff_default"), + }, + ) +} + +func handoffAfterConfirmationWorkflowDefinition() dsl.Definition { + return wfTestDefinition( + []dsl.Node{ + wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil), + wfTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", nil, map[string]any{"staticReply": "请确认转人工"}), + wfTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", wfTestInputs("prompt", "prompt_1", "replyText"), nil), + wfTestNode("handoff_1", workflowregistry.NodeTypeHandoffToHuman, "Handoff", map[string]dsl.Value{ + "reason": dsl.RefValue("start_1", "userMessage"), + "confirmed": dsl.RefValue("confirm_1", "confirmed"), + }, nil), + wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil), + }, + []dsl.Edge{ + wfTestEdge("start_1", "prompt_1", "edge_start_prompt"), + wfTestEdge("prompt_1", "confirm_1", "edge_prompt_confirm"), + wfTestEdge("confirm_1", "handoff_1", "edge_confirm_handoff"), + wfTestEdge("handoff_1", "end_1", "edge_handoff_end"), + }, + ) +} + +func wfTestDefinition(nodes []dsl.Node, edges []dsl.Edge) dsl.Definition { + return dsl.Definition{SchemaVersion: dsl.SchemaVersion, Nodes: nodes, Edges: edges} +} + +func wfTestNode(id string, nodeType string, title string, inputs map[string]dsl.Value, config any) dsl.Node { + return dsl.Node{ + ID: id, + Type: nodeType, + Meta: dsl.NodeMeta{Position: dsl.Position{X: 0, Y: 0}}, + Data: dsl.NodeData{ + Title: title, + InputsValues: inputs, + Config: mustMarshalWorkflowTestConfig(config), }, } } -func handoffAfterConfirmationWorkflowDefinition() dsl.Definition { - return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, - {ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认转人工"}`)}, - {ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{ - "prompt": {NodeID: "prompt_1", Field: "replyText"}, - }}, - {ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff", Inputs: map[string]dsl.VariableSelector{ - "reason": {NodeID: "start_1", Field: "userMessage"}, - "confirmed": {NodeID: "confirm_1", Field: "confirmed"}, - }}, - {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, - }, - 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_handoff", Source: "confirm_1", Target: "handoff_1"}, - {ID: "edge_handoff_end", Source: "handoff_1", Target: "end_1"}, +func wfTestInputs(name string, nodeID string, field string) map[string]dsl.Value { + return map[string]dsl.Value{name: dsl.RefValue(nodeID, field)} +} + +func wfTestEdge(source string, target string, id string) dsl.Edge { + return dsl.Edge{SourceNodeID: source, TargetNodeID: target, SourcePortID: id} +} + +func wfTestConditionBranch(id string, name string, targetNodeID string, nodeID string, field string, operator string, right any) dsl.ConditionBranch { + return dsl.ConditionBranch{ + ID: id, + Name: name, + TargetNodeID: targetNodeID, + Condition: &dsl.Condition{ + Left: &dsl.Value{Type: dsl.ValueTypeRef, Content: []string{nodeID, field}}, + Operator: operator, + Right: right, }, } } diff --git a/internal/ai/workflow/dsl/types.go b/internal/ai/workflow/dsl/types.go index 7a18265..d193648 100644 --- a/internal/ai/workflow/dsl/types.go +++ b/internal/ai/workflow/dsl/types.go @@ -2,20 +2,34 @@ package dsl import "encoding/json" +const SchemaVersion = 2 + type Definition struct { SchemaVersion int `json:"schemaVersion"` - EntryNodeID string `json:"entryNodeId"` Nodes []Node `json:"nodes"` Edges []Edge `json:"edges"` } type Node struct { - ID string `json:"id"` - Type string `json:"type"` - Name string `json:"name"` - Position Position `json:"position"` - Config json.RawMessage `json:"config"` - Inputs map[string]VariableSelector `json:"inputs,omitempty"` + ID string `json:"id"` + Type string `json:"type"` + Meta NodeMeta `json:"meta"` + Data NodeData `json:"data"` + Blocks []Node `json:"blocks,omitempty"` + Edges []Edge `json:"edges,omitempty"` +} + +type NodeMeta struct { + Position Position `json:"position"` +} + +type NodeData struct { + Title string `json:"title,omitempty"` + Config json.RawMessage `json:"config,omitempty"` + Inputs json.RawMessage `json:"inputs,omitempty"` + Outputs json.RawMessage `json:"outputs,omitempty"` + InputsValues map[string]Value `json:"inputsValues,omitempty"` + Extra map[string]json.RawMessage `json:"-"` } type Position struct { @@ -24,9 +38,25 @@ type Position struct { } type Edge struct { - ID string `json:"id"` - Source string `json:"source"` - Target string `json:"target"` + SourceNodeID string `json:"sourceNodeID"` + TargetNodeID string `json:"targetNodeID"` + SourcePortID string `json:"sourcePortID,omitempty"` + TargetPortID string `json:"targetPortID,omitempty"` +} + +type ValueType string + +const ( + ValueTypeConstant ValueType = "constant" + ValueTypeRef ValueType = "ref" + ValueTypeTemplate ValueType = "template" +) + +type Value struct { + Type ValueType `json:"type"` + Content []string `json:"content,omitempty"` + ConstantContent any `json:"-"` + RawContent json.RawMessage `json:"-"` } type ConditionConfig struct { @@ -42,13 +72,118 @@ type ConditionBranch struct { } type Condition struct { - Expression string `json:"expression,omitempty"` - Left *VariableSelector `json:"left,omitempty"` - Operator string `json:"operator,omitempty"` - Right any `json:"right,omitempty"` + Expression string `json:"expression,omitempty"` + Left *Value `json:"left,omitempty"` + Operator string `json:"operator,omitempty"` + Right any `json:"right,omitempty"` } -type VariableSelector struct { - NodeID string `json:"nodeId"` - Field string `json:"field"` +func RefValue(nodeID string, field string) Value { + return Value{Type: ValueTypeRef, Content: []string{nodeID, field}} +} + +func ConstantValue(value any) Value { + raw, _ := json.Marshal(value) + return Value{Type: ValueTypeConstant, ConstantContent: value, RawContent: raw} +} + +func TemplateValue(value string) Value { + return Value{Type: ValueTypeTemplate, Content: []string{value}} +} + +func (v Value) Ref() (nodeID string, field string, ok bool) { + if v.Type != ValueTypeRef || len(v.Content) < 2 { + return "", "", false + } + return v.Content[0], v.Content[1], true +} + +func (v *Value) UnmarshalJSON(data []byte) error { + type alias struct { + Type ValueType `json:"type"` + Content json.RawMessage `json:"content"` + } + var parsed alias + if err := json.Unmarshal(data, &parsed); err != nil { + return err + } + v.Type = parsed.Type + v.RawContent = append(v.RawContent[:0], parsed.Content...) + switch parsed.Type { + case ValueTypeRef: + var content []string + if len(parsed.Content) > 0 { + if err := json.Unmarshal(parsed.Content, &content); err != nil { + return err + } + } + v.Content = content + case ValueTypeTemplate: + var content string + if len(parsed.Content) > 0 { + if err := json.Unmarshal(parsed.Content, &content); err != nil { + return err + } + } + v.Content = []string{content} + case ValueTypeConstant: + if len(parsed.Content) > 0 { + if err := json.Unmarshal(parsed.Content, &v.ConstantContent); err != nil { + return err + } + } + default: + if len(parsed.Content) > 0 { + var content []string + if err := json.Unmarshal(parsed.Content, &content); err == nil { + v.Content = content + } + } + } + return nil +} + +func (v Value) MarshalJSON() ([]byte, error) { + type alias struct { + Type ValueType `json:"type"` + Content any `json:"content,omitempty"` + } + var content any + switch v.Type { + case ValueTypeRef: + content = v.Content + case ValueTypeTemplate: + if len(v.Content) > 0 { + content = v.Content[0] + } + case ValueTypeConstant: + content = v.ConstantContent + default: + if len(v.Content) > 0 { + content = v.Content + } + } + return json.Marshal(alias{Type: v.Type, Content: content}) +} + +func (d *NodeData) UnmarshalJSON(data []byte) error { + type alias NodeData + var parsed alias + if err := json.Unmarshal(data, &parsed); err != nil { + return err + } + extra := make(map[string]json.RawMessage) + if err := json.Unmarshal(data, &extra); err != nil { + return err + } + delete(extra, "title") + delete(extra, "config") + delete(extra, "inputs") + delete(extra, "outputs") + delete(extra, "inputsValues") + *d = NodeData(parsed) + if len(extra) > 0 { + d.Extra = extra + } + return nil } diff --git a/internal/ai/workflow/dsl/types_test.go b/internal/ai/workflow/dsl/types_test.go new file mode 100644 index 0000000..2b6f30e --- /dev/null +++ b/internal/ai/workflow/dsl/types_test.go @@ -0,0 +1,83 @@ +package dsl_test + +import ( + "encoding/json" + "testing" + + "agent-desk/internal/ai/workflow/dsl" +) + +func TestDefinitionUnmarshalsFlowGramStyleSchema(t *testing.T) { + raw := []byte(`{ + "schemaVersion": 2, + "nodes": [{ + "id": "send_1", + "type": "send_reply", + "meta": { + "position": { "x": 360, "y": 120 } + }, + "data": { + "title": "发送回复", + "config": { "text": "hello" }, + "inputs": { + "type": "object", + "properties": { + "replyText": { "type": "string" } + }, + "required": ["replyText"] + }, + "outputs": { + "type": "object", + "properties": { + "sent": { "type": "boolean" } + } + }, + "inputsValues": { + "replyText": { + "type": "ref", + "content": ["start_1", "userMessage"] + } + } + } + }], + "edges": [{ + "sourceNodeID": "start_1", + "targetNodeID": "send_1", + "sourcePortID": "default" + }] + }`) + + var def dsl.Definition + if err := json.Unmarshal(raw, &def); err != nil { + t.Fatalf("unmarshal definition: %v", err) + } + + if def.SchemaVersion != 2 { + t.Fatalf("unexpected schema version: %d", def.SchemaVersion) + } + node := def.Nodes[0] + if node.ID != "send_1" || node.Type != "send_reply" { + t.Fatalf("unexpected node identity: %#v", node) + } + if node.Meta.Position.X != 360 || node.Meta.Position.Y != 120 { + t.Fatalf("unexpected node position: %#v", node.Meta.Position) + } + if node.Data.Title != "发送回复" { + t.Fatalf("unexpected node title: %q", node.Data.Title) + } + var config map[string]string + if err := json.Unmarshal(node.Data.Config, &config); err != nil { + t.Fatalf("unmarshal config: %v", err) + } + if config["text"] != "hello" { + t.Fatalf("unexpected config: %s", node.Data.Config) + } + replyText := node.Data.InputsValues["replyText"] + if replyText.Type != dsl.ValueTypeRef || len(replyText.Content) != 2 || replyText.Content[0] != "start_1" || replyText.Content[1] != "userMessage" { + t.Fatalf("unexpected replyText value: %#v", replyText) + } + edge := def.Edges[0] + if edge.SourceNodeID != "start_1" || edge.TargetNodeID != "send_1" || edge.SourcePortID != "default" { + t.Fatalf("unexpected edge: %#v", edge) + } +} diff --git a/internal/ai/workflow/registry/registry.go b/internal/ai/workflow/registry/registry.go index 3715aba..f4ae791 100644 --- a/internal/ai/workflow/registry/registry.go +++ b/internal/ai/workflow/registry/registry.go @@ -67,8 +67,8 @@ func DefaultRegistry() *Registry { output("riskSignals", VariableTypeStringArray, "Detected risk signals."), output("reason", VariableTypeString, "Decision reason."), }, - DefaultInputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, + DefaultInputs: map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), }, }, NodeSpec{ @@ -123,8 +123,8 @@ func DefaultRegistry() *Registry { output("items", VariableTypeObjectArray, "Retrieved knowledge items."), output("summary", VariableTypeString, "Short retrieval summary."), }, - DefaultInputs: map[string]dsl.VariableSelector{ - "query": {NodeID: "start_1", Field: "userMessage"}, + DefaultInputs: map[string]dsl.Value{ + "query": dsl.RefValue("start_1", "userMessage"), }, }, NodeSpec{ diff --git a/internal/ai/workflow/registry/spec.go b/internal/ai/workflow/registry/spec.go index f5a35e1..c279704 100644 --- a/internal/ai/workflow/registry/spec.go +++ b/internal/ai/workflow/registry/spec.go @@ -41,16 +41,16 @@ type VariableValueOption struct { } type NodeSpec struct { - Type string `json:"type"` - Title string `json:"title"` - Description string `json:"description"` - RiskLevel NodeRiskLevel `json:"riskLevel"` - Interruptible bool `json:"interruptible"` - RequiresConfirmationPredecessor bool `json:"requiresConfirmationPredecessor"` - ConfigSchema any `json:"configSchema,omitempty"` - InputSchema []VariableSpec `json:"inputSchema,omitempty"` - OutputSchema []VariableSpec `json:"outputSchema,omitempty"` - DefaultInputs map[string]dsl.VariableSelector `json:"defaultInputs,omitempty"` + Type string `json:"type"` + Title string `json:"title"` + Description string `json:"description"` + RiskLevel NodeRiskLevel `json:"riskLevel"` + Interruptible bool `json:"interruptible"` + RequiresConfirmationPredecessor bool `json:"requiresConfirmationPredecessor"` + ConfigSchema any `json:"configSchema,omitempty"` + InputSchema []VariableSpec `json:"inputSchema,omitempty"` + OutputSchema []VariableSpec `json:"outputSchema,omitempty"` + DefaultInputs map[string]dsl.Value `json:"defaultInputs,omitempty"` } type Registry struct { diff --git a/internal/ai/workflow/validator/validator.go b/internal/ai/workflow/validator/validator.go index f84b169..6035476 100644 --- a/internal/ai/workflow/validator/validator.go +++ b/internal/ai/workflow/validator/validator.go @@ -53,7 +53,6 @@ type definitionValidator struct { func (v *definitionValidator) validate() { v.validateNodes() v.validateEdges() - v.validateEntry() v.validateReachability() v.validateConfirmationGuards() v.validateVariableMappings() @@ -98,53 +97,29 @@ func (v *definitionValidator) validateNodes() { } func (v *definitionValidator) validateEdges() { - seen := make(map[string]struct{}, len(v.def.Edges)) for index, edge := range v.def.Edges { - edge.ID = strings.TrimSpace(edge.ID) - edge.Source = strings.TrimSpace(edge.Source) - edge.Target = strings.TrimSpace(edge.Target) + source := strings.TrimSpace(edge.SourceNodeID) + target := strings.TrimSpace(edge.TargetNodeID) field := fmt.Sprintf("edges[%d]", index) - if edge.ID == "" { - v.addError(field+".id", "edge id is required") - } else if _, exists := seen[edge.ID]; exists { - v.addError(field+".id", "duplicate edge id: "+edge.ID) + if source == "" { + v.addError(field+".sourceNodeID", "edge source node is required") + } else if _, ok := v.nodesByID[source]; !ok { + v.addError(field+".sourceNodeID", "edge source node does not exist: "+source) } - seen[edge.ID] = struct{}{} - if edge.Source == "" { - v.addError(field+".source", "edge source is required") - } else if _, ok := v.nodesByID[edge.Source]; !ok { - v.addError(field+".source", "edge source node does not exist: "+edge.Source) + if target == "" { + v.addError(field+".targetNodeID", "edge target node is required") + } else if _, ok := v.nodesByID[target]; !ok { + v.addError(field+".targetNodeID", "edge target node does not exist: "+target) } - if edge.Target == "" { - v.addError(field+".target", "edge target is required") - } else if _, ok := v.nodesByID[edge.Target]; !ok { - v.addError(field+".target", "edge target node does not exist: "+edge.Target) + if source != "" && target != "" { + v.outgoing[source] = append(v.outgoing[source], target) + v.incoming[target] = append(v.incoming[target], source) } - if edge.Source != "" && edge.Target != "" { - v.outgoing[edge.Source] = append(v.outgoing[edge.Source], edge.Target) - v.incoming[edge.Target] = append(v.incoming[edge.Target], edge.Source) - } - } -} - -func (v *definitionValidator) validateEntry() { - entryNodeID := strings.TrimSpace(v.def.EntryNodeID) - if entryNodeID == "" { - v.addError("entryNodeId", "entry node id is required") - return - } - entry, ok := v.nodesByID[entryNodeID] - if !ok { - v.addError("entryNodeId", "entry node does not exist: "+entryNodeID) - return - } - if entry.Type != registry.NodeTypeStart { - v.addError("entryNodeId", "entry node must be the start node") } } func (v *definitionValidator) validateReachability() { - entryNodeID := strings.TrimSpace(v.def.EntryNodeID) + entryNodeID := v.entryNodeID() if entryNodeID == "" { return } @@ -187,17 +162,17 @@ func (v *definitionValidator) validateConfirmationGuards() { } func (v *definitionValidator) validateConfirmedInput(nodeID string, node dsl.Node) { - selector, ok := node.Inputs["confirmed"] - if !ok || strings.TrimSpace(selector.NodeID) == "" || strings.TrimSpace(selector.Field) == "" { + value, ok := node.Data.InputsValues["confirmed"] + sourceNodeID, sourceField, refOK := value.Ref() + if !ok || !refOK || strings.TrimSpace(sourceNodeID) == "" || strings.TrimSpace(sourceField) == "" { return } - sourceNodeID := strings.TrimSpace(selector.NodeID) sourceNode, ok := v.nodesByID[sourceNodeID] if !ok { return } - if sourceNode.Type != registry.NodeTypeHumanConfirm || strings.TrimSpace(selector.Field) != "confirmed" { - v.addError("nodes."+nodeID+".inputs.confirmed", "confirmed input must come from human_confirm.confirmed") + if sourceNode.Type != registry.NodeTypeHumanConfirm || strings.TrimSpace(sourceField) != "confirmed" { + v.addError("nodes."+nodeID+".data.inputsValues.confirmed", "confirmed input must come from human_confirm.confirmed") } } @@ -211,47 +186,55 @@ func (v *definitionValidator) validateVariableMappings() { if !input.Required { continue } - selector, ok := node.Inputs[input.Name] - if !ok || strings.TrimSpace(selector.NodeID) == "" || strings.TrimSpace(selector.Field) == "" { - v.addError("nodes."+id+".inputs."+input.Name, "required input mapping is missing: "+input.Name) + value, ok := node.Data.InputsValues[input.Name] + if !ok { + v.addError("nodes."+id+".data.inputsValues."+input.Name, "required input mapping is missing: "+input.Name) continue } - v.validateInputSelector(id, input, selector) + v.validateInputValue(id, input, value) } - for inputName, selector := range node.Inputs { - if strings.TrimSpace(selector.NodeID) == "" || strings.TrimSpace(selector.Field) == "" { - v.addError("nodes."+id+".inputs."+inputName, "input mapping source is required") - continue - } + for inputName, value := range node.Data.InputsValues { if _, ok := findInputSpec(spec.InputSchema, inputName); ok { continue } - sourceNode, sourceOK := v.nodesByID[strings.TrimSpace(selector.NodeID)] + sourceNodeID, sourceField, refOK := value.Ref() + if !refOK { + continue + } + sourceNode, sourceOK := v.nodesByID[strings.TrimSpace(sourceNodeID)] if !sourceOK { - v.addError("nodes."+id+".inputs."+inputName, "input source node does not exist: "+selector.NodeID) + v.addError("nodes."+id+".data.inputsValues."+inputName, "input source node does not exist: "+sourceNodeID) continue } sourceSpec, sourceSpecOK := v.registry.Get(sourceNode.Type) if !sourceSpecOK { continue } - if _, ok := findOutputSpec(sourceSpec.OutputSchema, selector.Field); !ok { - v.addError("nodes."+id+".inputs."+inputName, "input source field does not exist: "+selector.NodeID+"."+selector.Field) + if _, ok := findOutputSpec(sourceSpec.OutputSchema, sourceField); !ok { + v.addError("nodes."+id+".data.inputsValues."+inputName, "input source field does not exist: "+sourceNodeID+"."+sourceField) } } } } -func (v *definitionValidator) validateInputSelector(nodeID string, input registry.VariableSpec, selector dsl.VariableSelector) { - sourceNodeID := strings.TrimSpace(selector.NodeID) - sourceField := strings.TrimSpace(selector.Field) +func (v *definitionValidator) validateInputValue(nodeID string, input registry.VariableSpec, value dsl.Value) { + sourceNodeID, sourceField, ok := value.Ref() + if !ok { + if value.Type == dsl.ValueTypeConstant || value.Type == dsl.ValueTypeTemplate { + return + } + v.addError("nodes."+nodeID+".data.inputsValues."+input.Name, "input mapping source is required") + return + } + sourceNodeID = strings.TrimSpace(sourceNodeID) + sourceField = strings.TrimSpace(sourceField) sourceNode, ok := v.nodesByID[sourceNodeID] if !ok { - v.addError("nodes."+nodeID+".inputs."+input.Name, "input source node does not exist: "+sourceNodeID) + v.addError("nodes."+nodeID+".data.inputsValues."+input.Name, "input source node does not exist: "+sourceNodeID) return } if !v.hasPath(sourceNodeID, nodeID, make(map[string]struct{})) { - v.addError("nodes."+nodeID+".inputs."+input.Name, "input source node is not available before current node: "+sourceNodeID) + v.addError("nodes."+nodeID+".data.inputsValues."+input.Name, "input source node is not available before current node: "+sourceNodeID) return } sourceSpec, ok := v.registry.Get(sourceNode.Type) @@ -260,11 +243,11 @@ func (v *definitionValidator) validateInputSelector(nodeID string, input registr } output, ok := findOutputSpec(sourceSpec.OutputSchema, sourceField) if !ok { - v.addError("nodes."+nodeID+".inputs."+input.Name, "input source field does not exist: "+sourceNodeID+"."+sourceField) + v.addError("nodes."+nodeID+".data.inputsValues."+input.Name, "input source field does not exist: "+sourceNodeID+"."+sourceField) return } if !variableTypesCompatible(input.Type, output.Type) { - v.addError("nodes."+nodeID+".inputs."+input.Name, fmt.Sprintf("input type mismatch: %s expects %s but %s.%s is %s", input.Name, input.Type, sourceNodeID, sourceField, output.Type)) + v.addError("nodes."+nodeID+".data.inputsValues."+input.Name, fmt.Sprintf("input type mismatch: %s expects %s but %s.%s is %s", input.Name, input.Type, sourceNodeID, sourceField, output.Type)) } } @@ -275,8 +258,8 @@ func (v *definitionValidator) validateConditions() { } 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 { + if len(node.Data.Config) > 0 { + if err := json.Unmarshal(node.Data.Config, &config); err != nil { v.addError(field, "condition branches config must be valid JSON") continue } @@ -341,9 +324,10 @@ func (v *definitionValidator) validateCondition(field string, sourceNodeID strin 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 == "" { + sourceSelectorNodeID, sourceField, leftOK := condition.Left.Ref() + sourceSelectorNodeID = strings.TrimSpace(sourceSelectorNodeID) + sourceField = strings.TrimSpace(sourceField) + if !leftOK || sourceSelectorNodeID == "" || sourceField == "" { v.addError(field+".left", "condition left variable is required") return } @@ -462,13 +446,20 @@ func (v *definitionValidator) hasEdgeTo(sourceID string, targetID string) bool { return true } for _, edge := range v.def.Edges { - if strings.TrimSpace(edge.Source) == sourceID && strings.TrimSpace(edge.Target) == targetID { + if strings.TrimSpace(edge.SourceNodeID) == sourceID && strings.TrimSpace(edge.TargetNodeID) == targetID { return true } } return false } +func (v *definitionValidator) entryNodeID() string { + if len(v.startNodeIDs) != 1 { + return "" + } + return v.startNodeIDs[0] +} + func findInputSpec(items []registry.VariableSpec, name string) (registry.VariableSpec, bool) { name = strings.TrimSpace(name) for _, item := range items { diff --git a/internal/ai/workflow/validator/validator_test.go b/internal/ai/workflow/validator/validator_test.go index 6abb0d8..cbd0586 100644 --- a/internal/ai/workflow/validator/validator_test.go +++ b/internal/ai/workflow/validator/validator_test.go @@ -10,7 +10,7 @@ import ( "agent-desk/internal/ai/workflow/validator" ) -func TestValidateDefinitionAcceptsMinimalConversationFlow(t *testing.T) { +func TestValidateDefinitionAcceptsMinimalFlowGramStyleFlow(t *testing.T) { result := validator.ValidateDefinition(minimalDefinition(), registry.DefaultRegistry()) if !result.Valid { @@ -21,8 +21,8 @@ func TestValidateDefinitionAcceptsMinimalConversationFlow(t *testing.T) { func TestValidateDefinitionRejectsMissingStart(t *testing.T) { def := minimalDefinition() def.Nodes = []dsl.Node{ - {ID: "reply_1", Type: "send_reply", Config: json.RawMessage(`{"text":"hello"}`)}, - {ID: "end_1", Type: "end"}, + node("reply_1", "send_reply", inputs("replyText", dsl.RefValue("start_1", "userMessage")), nil), + node("end_1", "end", nil, nil), } result := validator.ValidateDefinition(def, registry.DefaultRegistry()) @@ -35,148 +35,9 @@ func TestValidateDefinitionRejectsMissingStart(t *testing.T) { } } -func TestValidateDefinitionRejectsUnknownNodeType(t *testing.T) { +func TestValidateDefinitionRejectsMissingRequiredInputValue(t *testing.T) { def := minimalDefinition() - def.Nodes = append(def.Nodes, dsl.Node{ID: "unknown_1", Type: "unknown_node"}) - def.Edges = append(def.Edges, dsl.Edge{ID: "e3", Source: "reply_1", Target: "unknown_1"}) - - result := validator.ValidateDefinition(def, registry.DefaultRegistry()) - - if result.Valid { - t.Fatalf("expected unknown node type to be invalid") - } - if !hasValidationMessage(result, "unknown node type") { - t.Fatalf("expected unknown-node error, got %#v", result.Errors) - } -} - -func TestValidateDefinitionRejectsUnguardedCreateTicket(t *testing.T) { - def := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "draft_1", Type: "prepare_ticket_draft"}, - {ID: "create_1", Type: "create_ticket"}, - {ID: "end_1", Type: "end"}, - }, - Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "draft_1"}, - {ID: "e2", Source: "draft_1", Target: "create_1"}, - {ID: "e3", Source: "create_1", Target: "end_1"}, - }, - } - - result := validator.ValidateDefinition(def, registry.DefaultRegistry()) - - if result.Valid { - t.Fatalf("expected unguarded create_ticket to be invalid") - } - if !hasValidationMessage(result, "requires human_confirm") { - t.Fatalf("expected confirmation guard error, got %#v", result.Errors) - } -} - -func TestValidateDefinitionAcceptsConfirmedCreateTicket(t *testing.T) { - def := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "draft_1", Type: "prepare_ticket_draft", Inputs: map[string]dsl.VariableSelector{ - "issue": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "confirm_1", Type: "human_confirm", Inputs: map[string]dsl.VariableSelector{ - "prompt": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "create_1", Type: "create_ticket", Inputs: map[string]dsl.VariableSelector{ - "ticketDraft": {NodeID: "draft_1", Field: "ticketDraft"}, - "confirmed": {NodeID: "confirm_1", Field: "confirmed"}, - }}, - {ID: "end_1", Type: "end"}, - }, - Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "draft_1"}, - {ID: "e2", Source: "draft_1", Target: "confirm_1"}, - {ID: "e3", Source: "confirm_1", Target: "create_1"}, - {ID: "e4", Source: "create_1", Target: "end_1"}, - }, - } - - result := validator.ValidateDefinition(def, registry.DefaultRegistry()) - - if !result.Valid { - t.Fatalf("expected confirmed create_ticket to be valid, got %#v", result.Errors) - } -} - -func TestValidateDefinitionAcceptsDirectHandoffToHuman(t *testing.T) { - def := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "handoff_1", Type: "handoff_to_human", Inputs: map[string]dsl.VariableSelector{ - "reason": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "end_1", Type: "end"}, - }, - Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "handoff_1"}, - {ID: "e2", Source: "handoff_1", Target: "end_1"}, - }, - } - - result := validator.ValidateDefinition(def, registry.DefaultRegistry()) - - if !result.Valid { - t.Fatalf("expected direct handoff_to_human to be valid, got %#v", result.Errors) - } -} - -func TestValidateDefinitionRejectsConfirmedInputFromNonConfirmNode(t *testing.T) { - def := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", - Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "analysis_1", Type: "analyze_conversation", Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "draft_1", Type: "prepare_ticket_draft", Inputs: map[string]dsl.VariableSelector{ - "issue": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "confirm_1", Type: "human_confirm", Inputs: map[string]dsl.VariableSelector{ - "prompt": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "create_1", Type: "create_ticket", Inputs: map[string]dsl.VariableSelector{ - "ticketDraft": {NodeID: "draft_1", Field: "ticketDraft"}, - "confirmed": {NodeID: "analysis_1", Field: "needTicket"}, - }}, - {ID: "end_1", Type: "end"}, - }, - Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "analysis_1"}, - {ID: "e2", Source: "analysis_1", Target: "draft_1"}, - {ID: "e3", Source: "draft_1", Target: "confirm_1"}, - {ID: "e4", Source: "confirm_1", Target: "create_1"}, - {ID: "e5", Source: "create_1", Target: "end_1"}, - }, - } - - result := validator.ValidateDefinition(def, registry.DefaultRegistry()) - - if result.Valid { - t.Fatalf("expected confirmed input from non-confirm node to be invalid") - } - if !hasValidationMessage(result, "confirmed input must come from human_confirm.confirmed") { - t.Fatalf("expected confirmed-source error, got %#v", result.Errors) - } -} - -func TestValidateDefinitionRejectsMissingRequiredInputMapping(t *testing.T) { - def := minimalDefinition() - def.Nodes[1].Inputs = nil + def.Nodes[1].Data.InputsValues = nil result := validator.ValidateDefinition(def, registry.DefaultRegistry()) @@ -189,8 +50,8 @@ func TestValidateDefinitionRejectsMissingRequiredInputMapping(t *testing.T) { } func TestValidateDefinitionRejectsUnknownInputSourceNode(t *testing.T) { - def := mappedReplyDefinition() - def.Nodes[1].Inputs["replyText"] = dsl.VariableSelector{NodeID: "missing_1", Field: "replyText"} + def := minimalDefinition() + def.Nodes[1].Data.InputsValues["replyText"] = dsl.RefValue("missing_1", "replyText") result := validator.ValidateDefinition(def, registry.DefaultRegistry()) @@ -202,9 +63,25 @@ func TestValidateDefinitionRejectsUnknownInputSourceNode(t *testing.T) { } } +func TestValidateDefinitionRejectsUnavailableInputSourceNode(t *testing.T) { + def := minimalDefinition() + def.Nodes = append(def.Nodes, node("late_1", "llm_reply", inputs("userMessage", dsl.RefValue("reply_1", "sent")), nil)) + def.Edges = append(def.Edges, edge("reply_1", "late_1")) + def.Nodes[1].Data.InputsValues["replyText"] = dsl.RefValue("late_1", "replyText") + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected downstream input source to be invalid") + } + if !hasValidationMessage(result, "input source node is not available before current node") { + t.Fatalf("expected source availability error, got %#v", result.Errors) + } +} + func TestValidateDefinitionRejectsUnknownInputSourceField(t *testing.T) { - def := mappedReplyDefinition() - def.Nodes[1].Inputs["replyText"] = dsl.VariableSelector{NodeID: "start_1", Field: "missing"} + def := minimalDefinition() + def.Nodes[1].Data.InputsValues["replyText"] = dsl.RefValue("start_1", "missing") result := validator.ValidateDefinition(def, registry.DefaultRegistry()) @@ -217,8 +94,8 @@ func TestValidateDefinitionRejectsUnknownInputSourceField(t *testing.T) { } func TestValidateDefinitionRejectsIncompatibleInputType(t *testing.T) { - def := mappedReplyDefinition() - def.Nodes[1].Inputs["replyText"] = dsl.VariableSelector{NodeID: "start_1", Field: "conversationId"} + def := minimalDefinition() + def.Nodes[1].Data.InputsValues["replyText"] = dsl.RefValue("start_1", "conversationId") result := validator.ValidateDefinition(def, registry.DefaultRegistry()) @@ -230,69 +107,83 @@ func TestValidateDefinitionRejectsIncompatibleInputType(t *testing.T) { } } -func TestValidateDefinitionAcceptsMappedKnowledgeFlow(t *testing.T) { +func TestValidateDefinitionAcceptsConfirmedCreateTicket(t *testing.T) { def := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "retrieve_1", Type: "knowledge_retrieve", Inputs: map[string]dsl.VariableSelector{ - "query": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "reply_1", Type: "send_reply", Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "end_1", Type: "end"}, + node("start_1", "start", nil, nil), + node("draft_1", "prepare_ticket_draft", inputs("issue", dsl.RefValue("start_1", "userMessage")), nil), + node("confirm_1", "human_confirm", inputs("prompt", dsl.RefValue("start_1", "userMessage")), nil), + node("create_1", "create_ticket", map[string]dsl.Value{ + "ticketDraft": dsl.RefValue("draft_1", "ticketDraft"), + "confirmed": dsl.RefValue("confirm_1", "confirmed"), + }, nil), + node("end_1", "end", nil, nil), }, Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "retrieve_1"}, - {ID: "e2", Source: "retrieve_1", Target: "reply_1"}, - {ID: "e3", Source: "reply_1", Target: "end_1"}, + edge("start_1", "draft_1"), + edge("draft_1", "confirm_1"), + edge("confirm_1", "create_1"), + edge("create_1", "end_1"), }, } result := validator.ValidateDefinition(def, registry.DefaultRegistry()) if !result.Valid { - t.Fatalf("expected mapped knowledge flow to be valid, got %#v", result.Errors) + t.Fatalf("expected confirmed create_ticket to be valid, got %#v", result.Errors) } } -func TestValidateDefinitionRejectsUnknownConditionOperator(t *testing.T) { - def := conditionDefinition() - var config dsl.ConditionConfig - if err := json.Unmarshal(def.Nodes[1].Config, &config); err != nil { - t.Fatalf("unmarshal condition config: %v", err) +func TestValidateDefinitionRejectsConfirmedInputFromNonConfirmNode(t *testing.T) { + def := minimalDefinition() + def.Nodes = []dsl.Node{ + node("start_1", "start", nil, nil), + node("draft_1", "prepare_ticket_draft", inputs("issue", dsl.RefValue("start_1", "userMessage")), nil), + node("create_1", "create_ticket", map[string]dsl.Value{ + "ticketDraft": dsl.RefValue("draft_1", "ticketDraft"), + "confirmed": dsl.RefValue("start_1", "userMessage"), + }, nil), + node("end_1", "end", nil, nil), } - config.Branches[0].Condition.Operator = "regex" - raw, err := json.Marshal(config) - if err != nil { - t.Fatalf("marshal condition config: %v", err) + def.Edges = []dsl.Edge{ + edge("start_1", "draft_1"), + edge("draft_1", "create_1"), + edge("create_1", "end_1"), } - def.Nodes[1].Config = raw result := validator.ValidateDefinition(def, registry.DefaultRegistry()) if result.Valid { - t.Fatalf("expected unknown condition operator to be invalid") + t.Fatalf("expected confirmed input from non-confirm node to be invalid") } - if !hasValidationMessage(result, "unsupported condition operator") { - t.Fatalf("expected condition operator error, got %#v", result.Errors) + if !hasValidationMessage(result, "confirmed input must come from human_confirm.confirmed") { + t.Fatalf("expected confirmed-source error, got %#v", result.Errors) + } +} + +func TestValidateDefinitionRejectsConditionBranchTargetWithoutEdge(t *testing.T) { + def := conditionDefinition() + def.Edges = []dsl.Edge{edge("start_1", "condition_1")} + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected condition branch target without edge to be invalid") + } + if !hasValidationMessage(result, "condition branch target must have an outgoing edge") { + t.Fatalf("expected branch edge error, got %#v", result.Errors) } } func TestValidateDefinitionRejectsUnknownConditionVariable(t *testing.T) { def := conditionDefinition() var config dsl.ConditionConfig - if err := json.Unmarshal(def.Nodes[1].Config, &config); err != nil { + if err := json.Unmarshal(def.Nodes[1].Data.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 + config.Branches[0].Condition.Left = &dsl.Value{Type: dsl.ValueTypeRef, Content: []string{"start_1", "missing"}} + def.Nodes[1].Data.Config = mustJSON(config) result := validator.ValidateDefinition(def, registry.DefaultRegistry()) @@ -304,123 +195,30 @@ func TestValidateDefinitionRejectsUnknownConditionVariable(t *testing.T) { } } -func TestValidateDefinitionRejectsInvalidConditionEnumValue(t *testing.T) { - def := policyConditionDefinition("unknown_action") - - result := validator.ValidateDefinition(def, registry.DefaultRegistry()) - - if result.Valid { - t.Fatalf("expected invalid enum condition value to be rejected") - } - if !hasValidationMessage(result, "condition comparison value is not allowed") { - t.Fatalf("expected condition enum value error, got %#v", result.Errors) - } -} - -func TestValidateDefinitionRejectsConditionDefaultBranchBeforeLast(t *testing.T) { - def := conditionDefinition() - 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], config.Branches[1] = config.Branches[1], config.Branches[0] - 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()) - - if result.Valid { - t.Fatalf("expected default branch before last to be invalid") - } - if !hasValidationMessage(result, "default condition branch must be last") { - t.Fatalf("expected default branch order error, got %#v", result.Errors) - } -} - func minimalDefinition() dsl.Definition { return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "reply_1", Type: "send_reply", Config: json.RawMessage(`{"text":"hello"}`), Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "end_1", Type: "end"}, + node("start_1", "start", nil, nil), + node("reply_1", "send_reply", inputs("replyText", dsl.RefValue("start_1", "userMessage")), map[string]any{"text": "hello"}), + node("end_1", "end", nil, nil), }, Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "reply_1"}, - {ID: "e2", Source: "reply_1", Target: "end_1"}, + edge("start_1", "reply_1"), + edge("reply_1", "end_1"), }, } } -func policyConditionDefinition(action any) dsl.Definition { - conditionConfig, _ := json.Marshal(dsl.ConditionConfig{ - Branches: []dsl.ConditionBranch{ - { - ID: "direct", - Name: "Direct", - TargetNodeID: "end_1", - Condition: &dsl.Condition{ - Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, - Operator: "eq", - Right: action, - }, - }, - { - 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: "understanding_1", Type: "conversation_understanding", Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "policy_1", Type: "reply_policy", Inputs: map[string]dsl.VariableSelector{ - "messageIntent": {NodeID: "understanding_1", Field: "messageIntent"}, - "answerScope": {NodeID: "understanding_1", Field: "answerScope"}, - }}, - {ID: "condition_1", Type: "condition", Config: conditionConfig}, - {ID: "end_1", Type: "end"}, - }, - Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "understanding_1"}, - {ID: "e2", Source: "understanding_1", Target: "policy_1"}, - {ID: "e3", Source: "policy_1", Target: "condition_1"}, - {ID: "e4", Source: "condition_1", Target: "end_1"}, - {ID: "e5", Source: "condition_1", Target: "end_1"}, - }, - } -} - -func mappedReplyDefinition() dsl.Definition { - def := minimalDefinition() - def.Nodes[1].Inputs = map[string]dsl.VariableSelector{ - "replyText": {NodeID: "start_1", Field: "userMessage"}, - } - return def -} - func conditionDefinition() dsl.Definition { - conditionConfig, _ := json.Marshal(dsl.ConditionConfig{ + conditionConfig := dsl.ConditionConfig{ Branches: []dsl.ConditionBranch{ { ID: "hello", Name: "Hello", TargetNodeID: "end_1", Condition: &dsl.Condition{ - Left: &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"}, + Left: &dsl.Value{Type: dsl.ValueTypeRef, Content: []string{"start_1", "userMessage"}}, Operator: "eq", Right: "hello", }, @@ -432,23 +230,54 @@ func conditionDefinition() dsl.Definition { Default: true, }, }, - }) + } return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "condition_1", Type: "condition", Config: conditionConfig}, - {ID: "end_1", Type: "end"}, + node("start_1", "start", nil, nil), + node("condition_1", "condition", nil, conditionConfig), + node("end_1", "end", nil, nil), }, 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"}, + edge("start_1", "condition_1"), + edge("condition_1", "end_1"), + edge("condition_1", "end_1"), }, } } +func node(id string, nodeType string, inputValues map[string]dsl.Value, config any) dsl.Node { + return dsl.Node{ + ID: id, + Type: nodeType, + Meta: dsl.NodeMeta{Position: dsl.Position{X: 0, Y: 0}}, + Data: dsl.NodeData{ + Title: nodeType, + Config: mustJSON(config), + InputsValues: inputValues, + }, + } +} + +func edge(source string, target string) dsl.Edge { + return dsl.Edge{SourceNodeID: source, TargetNodeID: target} +} + +func inputs(name string, value dsl.Value) map[string]dsl.Value { + return map[string]dsl.Value{name: value} +} + +func mustJSON(value any) json.RawMessage { + if value == nil { + return nil + } + raw, err := json.Marshal(value) + if err != nil { + panic(err) + } + return raw +} + func hasValidationMessage(result validator.Result, want string) bool { for _, item := range result.Errors { if strings.Contains(item.Message, want) { diff --git a/internal/builders/ai_workflow_builder_test.go b/internal/builders/ai_workflow_builder_test.go index 1d254df..144a710 100644 --- a/internal/builders/ai_workflow_builder_test.go +++ b/internal/builders/ai_workflow_builder_test.go @@ -99,13 +99,12 @@ func TestBuildAIWorkflowRunIncludesAuditDisplayFields(t *testing.T) { func TestBuildAIWorkflowRunDetailIncludesPublishedDefinitionSnapshot(t *testing.T) { definition := dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "开始"}, - {ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "运行时回复"}, + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Data: dsl.NodeData{Title: "开始"}}, + {ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Data: dsl.NodeData{Title: "运行时回复"}}, }, - Edges: []dsl.Edge{{ID: "edge_start_reply", Source: "start_1", Target: "reply_1"}}, + Edges: []dsl.Edge{{SourceNodeID: "start_1", TargetNodeID: "reply_1", SourcePortID: "edge_start_reply"}}, } buf, err := json.Marshal(definition) if err != nil { @@ -120,13 +119,13 @@ func TestBuildAIWorkflowRunDetailIncludesPublishedDefinitionSnapshot(t *testing. &models.AIAgent{Name: "售后 Agent"}, ) - if resp.Definition.EntryNodeID != "start_1" { + if resp.Definition.SchemaVersion != dsl.SchemaVersion { t.Fatalf("expected run detail definition from published version, got %#v", resp.Definition) } - if len(resp.Definition.Nodes) != 2 || resp.Definition.Nodes[1].Name != "运行时回复" { + if len(resp.Definition.Nodes) != 2 || resp.Definition.Nodes[1].Data.Title != "运行时回复" { t.Fatalf("expected published definition nodes, got %#v", resp.Definition.Nodes) } - if len(resp.Definition.Edges) != 1 || resp.Definition.Edges[0].ID != "edge_start_reply" { + if len(resp.Definition.Edges) != 1 || resp.Definition.Edges[0].SourcePortID != "edge_start_reply" { t.Fatalf("expected published definition edges, got %#v", resp.Definition.Edges) } } diff --git a/internal/pkg/dto/response/ai_workflow_response.go b/internal/pkg/dto/response/ai_workflow_response.go index 083070f..d40b66a 100644 --- a/internal/pkg/dto/response/ai_workflow_response.go +++ b/internal/pkg/dto/response/ai_workflow_response.go @@ -51,7 +51,7 @@ type AIWorkflowNodeSpecResponse struct { ConfigSchema any `json:"configSchema,omitempty"` InputSchema []workflowregistry.VariableSpec `json:"inputSchema,omitempty"` OutputSchema []workflowregistry.VariableSpec `json:"outputSchema,omitempty"` - DefaultInputs map[string]dsl.VariableSelector `json:"defaultInputs,omitempty"` + DefaultInputs map[string]dsl.Value `json:"defaultInputs,omitempty"` } type AIWorkflowRunResponse struct { diff --git a/internal/services/ai_agent_workflow_service_test.go b/internal/services/ai_agent_workflow_service_test.go index b18b679..c5d11f8 100644 --- a/internal/services/ai_agent_workflow_service_test.go +++ b/internal/services/ai_agent_workflow_service_test.go @@ -49,7 +49,7 @@ func TestAIAgentServiceCreatesDefaultWorkflow(t *testing.T) { if err := json.Unmarshal([]byte(workflow.DraftDefinition), &stored); err != nil { t.Fatalf("unmarshal draft definition: %v", err) } - if stored.EntryNodeID == "" { + if stored.SchemaVersion != dsl.SchemaVersion || nodeTypeByID(stored, "start_1") != workflowregistry.NodeTypeStart { t.Fatalf("expected default draft definition") } validation := workflowvalidator.ValidateDefinition(stored, workflowregistry.DefaultRegistry()) @@ -93,7 +93,7 @@ func TestAIAgentServiceCreatesDefaultWorkflow(t *testing.T) { func TestAIWorkflowServiceDefaultAgentWorkflowDefinitionIsValid(t *testing.T) { definition := AIWorkflowService.DefaultAgentWorkflowDefinition() - if definition.EntryNodeID == "" { + if definition.SchemaVersion != dsl.SchemaVersion || nodeTypeByID(definition, "start_1") != workflowregistry.NodeTypeStart { t.Fatalf("expected default workflow definition") } validation := workflowvalidator.ValidateDefinition(definition, workflowregistry.DefaultRegistry()) @@ -296,7 +296,7 @@ func conditionBranches(t *testing.T, def dsl.Definition, nodeID string) []dsl.Co continue } var config dsl.ConditionConfig - if err := json.Unmarshal(node.Config, &config); err != nil { + if err := json.Unmarshal(node.Data.Config, &config); err != nil { t.Fatalf("unmarshal condition config for %s: %v", nodeID, err) } return config.Branches @@ -307,7 +307,7 @@ func conditionBranches(t *testing.T, def dsl.Definition, nodeID string) []dsl.Co func workflowEdgeExists(def dsl.Definition, sourceID string, targetID string) bool { for _, edge := range def.Edges { - if edge.Source == sourceID && edge.Target == targetID { + if edge.SourceNodeID == sourceID && edge.TargetNodeID == targetID { return true } } @@ -329,10 +329,10 @@ func assertWorkflowLayoutDoesNotOverlap(t *testing.T, def dsl.Definition) { width, height := defaultWorkflowNodeRenderSize(node.Type) boxes = append(boxes, workflowLayoutBox{ NodeID: node.ID, - Left: node.Position.X, - Top: node.Position.Y, - Right: node.Position.X + width, - Bottom: node.Position.Y + height, + Left: node.Meta.Position.X, + Top: node.Meta.Position.Y, + Right: node.Meta.Position.X + width, + Bottom: node.Meta.Position.Y + height, }) } const minGap = 32.0 diff --git a/internal/services/ai_workflow_service.go b/internal/services/ai_workflow_service.go index 1543e62..2515c35 100644 --- a/internal/services/ai_workflow_service.go +++ b/internal/services/ai_workflow_service.go @@ -423,120 +423,132 @@ func (s *aiWorkflowService) createDefaultAgentWorkflow(db *gorm.DB, agent *model func defaultAgentWorkflowDefinition() dsl.Definition { return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "开始", Position: dsl.Position{X: 0, Y: 520}}, - {ID: "understanding_1", Type: workflowregistry.NodeTypeConversationUnderstanding, Name: "会话理解", Position: dsl.Position{X: 320, Y: 520}, Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "policy_1", Type: workflowregistry.NodeTypeReplyPolicy, Name: "回复策略", Position: dsl.Position{X: 640, Y: 520}, Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - "messageIntent": {NodeID: "understanding_1", Field: "messageIntent"}, - "answerScope": {NodeID: "understanding_1", Field: "answerScope"}, - "riskSignals": {NodeID: "understanding_1", Field: "riskSignals"}, - }}, - {ID: "policy_route_1", Type: workflowregistry.NodeTypeCondition, Name: "策略分流", Position: dsl.Position{X: 960, Y: 520}, Config: mustMarshalWorkflowConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ - {ID: "direct", Name: "直接回复", TargetNodeID: "policy_reply_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, Operator: "eq", Right: "direct_reply"}}, - {ID: "clarify", Name: "追问澄清", TargetNodeID: "policy_reply_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, Operator: "eq", Right: "clarify"}}, - {ID: "end_conversation", Name: "结束语", TargetNodeID: "policy_reply_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, Operator: "eq", Right: "end_conversation"}}, - {ID: "handoff", Name: "转人工", TargetNodeID: "handoff_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, Operator: "eq", Right: "handoff_to_human"}}, - {ID: "ticket", Name: "创建工单", TargetNodeID: "draft_ticket_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, Operator: "eq", Right: "prepare_ticket"}}, - {ID: "knowledge", Name: "知识库回复", TargetNodeID: "retrieve_1", Condition: &dsl.Condition{Left: &dsl.VariableSelector{NodeID: "policy_1", Field: "action"}, Operator: "eq", Right: "retrieve_knowledge"}}, + workflowNode("start_1", workflowregistry.NodeTypeStart, "开始", 0, 520, nil, nil), + workflowNode("understanding_1", workflowregistry.NodeTypeConversationUnderstanding, "会话理解", 320, 520, workflowInputs("userMessage", "start_1", "userMessage"), nil), + workflowNode("policy_1", workflowregistry.NodeTypeReplyPolicy, "回复策略", 640, 520, map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), + "messageIntent": dsl.RefValue("understanding_1", "messageIntent"), + "answerScope": dsl.RefValue("understanding_1", "answerScope"), + "riskSignals": dsl.RefValue("understanding_1", "riskSignals"), + }, nil), + workflowNode("policy_route_1", workflowregistry.NodeTypeCondition, "策略分流", 960, 520, nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + workflowConditionBranch("direct", "直接回复", "policy_reply_1", "policy_1", "action", "eq", "direct_reply"), + workflowConditionBranch("clarify", "追问澄清", "policy_reply_1", "policy_1", "action", "eq", "clarify"), + workflowConditionBranch("end_conversation", "结束语", "policy_reply_1", "policy_1", "action", "eq", "end_conversation"), + workflowConditionBranch("handoff", "转人工", "handoff_1", "policy_1", "action", "eq", "handoff_to_human"), + workflowConditionBranch("ticket", "创建工单", "draft_ticket_1", "policy_1", "action", "eq", "prepare_ticket"), + workflowConditionBranch("knowledge", "知识库回复", "retrieve_1", "policy_1", "action", "eq", "retrieve_knowledge"), {ID: "default", Name: "默认澄清", TargetNodeID: "policy_reply_1", Default: true}, - }})}, - {ID: "policy_reply_1", Type: workflowregistry.NodeTypeSendReply, Name: "发送策略回复", Position: dsl.Position{X: 1280, Y: 0}, Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "policy_1", Field: "replyText"}, - }}, - {ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "转人工", Position: dsl.Position{X: 1280, Y: 220}, Inputs: map[string]dsl.VariableSelector{ - "reason": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "handoff_end_1", Type: workflowregistry.NodeTypeEnd, Name: "结束", Position: dsl.Position{X: 1600, Y: 220}}, - {ID: "draft_ticket_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "整理工单草稿", Position: dsl.Position{X: 1280, Y: 440}, Inputs: map[string]dsl.VariableSelector{ - "issue": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "ticket_confirm_prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "建单确认文案", Position: dsl.Position{X: 1600, Y: 440}, Config: json.RawMessage(`{"staticReply":"我已整理工单草稿。请回复“确认”创建工单，或回复“取消”放弃。"}`), Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "ticket_confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "确认建单", Position: dsl.Position{X: 1920, Y: 440}, 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: 2240, Y: 440}, 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"}}, + }}), + workflowNode("policy_reply_1", workflowregistry.NodeTypeSendReply, "发送策略回复", 1280, 0, workflowInputs("replyText", "policy_1", "replyText"), nil), + workflowNode("handoff_1", workflowregistry.NodeTypeHandoffToHuman, "转人工", 1280, 220, workflowInputs("reason", "start_1", "userMessage"), nil), + workflowNode("handoff_end_1", workflowregistry.NodeTypeEnd, "结束", 1600, 220, nil, nil), + workflowNode("draft_ticket_1", workflowregistry.NodeTypePrepareTicketDraft, "整理工单草稿", 1280, 440, workflowInputs("issue", "start_1", "userMessage"), nil), + workflowNode("ticket_confirm_prompt_1", workflowregistry.NodeTypeLLMReply, "建单确认文案", 1600, 440, workflowInputs("userMessage", "start_1", "userMessage"), map[string]any{"staticReply": "我已整理工单草稿。请回复“确认”创建工单，或回复“取消”放弃。"}), + workflowNode("ticket_confirm_1", workflowregistry.NodeTypeHumanConfirm, "确认建单", 1920, 440, workflowInputs("prompt", "ticket_confirm_prompt_1", "replyText"), nil), + workflowNode("ticket_confirm_route_1", workflowregistry.NodeTypeCondition, "建单确认分流", 2240, 440, nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + workflowConditionBranch("confirmed", "已确认", "create_ticket_1", "ticket_confirm_1", "confirmed", "is_true", nil), {ID: "default", Name: "取消或未确认", TargetNodeID: "ticket_cancel_reply_1", Default: true}, - }})}, - {ID: "create_ticket_1", Type: workflowregistry.NodeTypeCreateTicket, Name: "创建工单", Position: dsl.Position{X: 2560, Y: 320}, Inputs: map[string]dsl.VariableSelector{ - "ticketDraft": {NodeID: "draft_ticket_1", Field: "ticketDraft"}, - "confirmed": {NodeID: "ticket_confirm_1", Field: "confirmed"}, - }}, - {ID: "ticket_result_reply_1", Type: workflowregistry.NodeTypeSendReply, Name: "发送建单结果", Position: dsl.Position{X: 2880, Y: 320}, Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "create_ticket_1", Field: "message"}, - }}, - {ID: "ticket_cancel_reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "取消建单提示", Position: dsl.Position{X: 2560, Y: 560}, Config: json.RawMessage(`{"staticReply":"已取消创建工单。你可以继续补充问题，我会继续帮你处理。"}`), Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "send_ticket_cancel_1", Type: workflowregistry.NodeTypeSendReply, Name: "发送取消提示", Position: dsl.Position{X: 2880, Y: 560}, Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "ticket_cancel_reply_1", Field: "replyText"}, - }}, - {ID: "retrieve_1", Type: workflowregistry.NodeTypeKnowledgeRetrieve, Name: "知识检索", Position: dsl.Position{X: 1280, Y: 860}, Inputs: map[string]dsl.VariableSelector{ - "query": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "answerability_1", Type: workflowregistry.NodeTypeAnswerabilityGate, Name: "可回答判断", Position: dsl.Position{X: 1600, Y: 860}, Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - "knowledgeItems": {NodeID: "retrieve_1", Field: "items"}, - }}, - {ID: "answerability_route_1", Type: workflowregistry.NodeTypeCondition, Name: "可回答分流", Position: dsl.Position{X: 1920, Y: 860}, 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"}}, + }}), + workflowNode("create_ticket_1", workflowregistry.NodeTypeCreateTicket, "创建工单", 2560, 320, map[string]dsl.Value{ + "ticketDraft": dsl.RefValue("draft_ticket_1", "ticketDraft"), + "confirmed": dsl.RefValue("ticket_confirm_1", "confirmed"), + }, nil), + workflowNode("ticket_result_reply_1", workflowregistry.NodeTypeSendReply, "发送建单结果", 2880, 320, workflowInputs("replyText", "create_ticket_1", "message"), nil), + workflowNode("ticket_cancel_reply_1", workflowregistry.NodeTypeLLMReply, "取消建单提示", 2560, 560, workflowInputs("userMessage", "start_1", "userMessage"), map[string]any{"staticReply": "已取消创建工单。你可以继续补充问题，我会继续帮你处理。"}), + workflowNode("send_ticket_cancel_1", workflowregistry.NodeTypeSendReply, "发送取消提示", 2880, 560, workflowInputs("replyText", "ticket_cancel_reply_1", "replyText"), nil), + workflowNode("retrieve_1", workflowregistry.NodeTypeKnowledgeRetrieve, "知识检索", 1280, 860, workflowInputs("query", "start_1", "userMessage"), nil), + workflowNode("answerability_1", workflowregistry.NodeTypeAnswerabilityGate, "可回答判断", 1600, 860, map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), + "knowledgeItems": dsl.RefValue("retrieve_1", "items"), + }, nil), + workflowNode("answerability_route_1", workflowregistry.NodeTypeCondition, "可回答分流", 1920, 860, nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{ + workflowConditionBranch("answerable", "可以回答", "reply_1", "answerability_1", "answerability", "eq", "answerable"), {ID: "default", Name: "兜底追问", TargetNodeID: "fallback_reply_1", Default: true}, - }})}, - {ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "AI 回复", Position: dsl.Position{X: 2240, Y: 780}, Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - "knowledgeItems": {NodeID: "retrieve_1", Field: "items"}, - }}, - {ID: "send_1", Type: workflowregistry.NodeTypeSendReply, Name: "发送回复", Position: dsl.Position{X: 2560, Y: 780}, Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "reply_1", Field: "replyText"}, - }}, - {ID: "fallback_reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "兜底追问", Position: dsl.Position{X: 2240, Y: 1040}, Inputs: map[string]dsl.VariableSelector{ - "userMessage": {NodeID: "start_1", Field: "userMessage"}, - "knowledgeItems": {NodeID: "retrieve_1", Field: "items"}, - }}, - {ID: "send_fallback_1", Type: workflowregistry.NodeTypeSendReply, Name: "发送兜底", Position: dsl.Position{X: 2560, Y: 1040}, Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "fallback_reply_1", Field: "replyText"}, - }}, - {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "结束", Position: dsl.Position{X: 3200, Y: 780}}, + }}), + workflowNode("reply_1", workflowregistry.NodeTypeLLMReply, "AI 回复", 2240, 780, map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), + "knowledgeItems": dsl.RefValue("retrieve_1", "items"), + }, nil), + workflowNode("send_1", workflowregistry.NodeTypeSendReply, "发送回复", 2560, 780, workflowInputs("replyText", "reply_1", "replyText"), nil), + workflowNode("fallback_reply_1", workflowregistry.NodeTypeLLMReply, "兜底追问", 2240, 1040, map[string]dsl.Value{ + "userMessage": dsl.RefValue("start_1", "userMessage"), + "knowledgeItems": dsl.RefValue("retrieve_1", "items"), + }, nil), + workflowNode("send_fallback_1", workflowregistry.NodeTypeSendReply, "发送兜底", 2560, 1040, workflowInputs("replyText", "fallback_reply_1", "replyText"), nil), + workflowNode("end_1", workflowregistry.NodeTypeEnd, "结束", 3200, 780, nil, nil), }, Edges: []dsl.Edge{ - {ID: "edge_start_understanding", Source: "start_1", Target: "understanding_1"}, - {ID: "edge_understanding_policy", Source: "understanding_1", Target: "policy_1"}, - {ID: "edge_policy_route", Source: "policy_1", Target: "policy_route_1"}, - {ID: "edge_policy_reply", Source: "policy_route_1", Target: "policy_reply_1"}, - {ID: "edge_policy_handoff", Source: "policy_route_1", Target: "handoff_1"}, - {ID: "edge_policy_ticket", Source: "policy_route_1", Target: "draft_ticket_1"}, - {ID: "edge_policy_knowledge", Source: "policy_route_1", Target: "retrieve_1"}, - {ID: "edge_policy_reply_end", Source: "policy_reply_1", Target: "end_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_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_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"}, - {ID: "edge_send_fallback_end", Source: "send_fallback_1", Target: "end_1"}, + workflowEdge("start_1", "understanding_1"), + workflowEdge("understanding_1", "policy_1"), + workflowEdge("policy_1", "policy_route_1"), + workflowEdge("policy_route_1", "policy_reply_1"), + workflowEdge("policy_route_1", "handoff_1"), + workflowEdge("policy_route_1", "draft_ticket_1"), + workflowEdge("policy_route_1", "retrieve_1"), + workflowEdge("policy_reply_1", "end_1"), + workflowEdge("handoff_1", "handoff_end_1"), + workflowEdge("draft_ticket_1", "ticket_confirm_prompt_1"), + workflowEdge("ticket_confirm_prompt_1", "ticket_confirm_1"), + workflowEdge("ticket_confirm_1", "ticket_confirm_route_1"), + workflowEdge("ticket_confirm_route_1", "create_ticket_1"), + workflowEdge("ticket_confirm_route_1", "ticket_cancel_reply_1"), + workflowEdge("create_ticket_1", "ticket_result_reply_1"), + workflowEdge("ticket_result_reply_1", "end_1"), + workflowEdge("ticket_cancel_reply_1", "send_ticket_cancel_1"), + workflowEdge("send_ticket_cancel_1", "end_1"), + workflowEdge("retrieve_1", "answerability_1"), + workflowEdge("answerability_1", "answerability_route_1"), + workflowEdge("answerability_route_1", "reply_1"), + workflowEdge("answerability_route_1", "fallback_reply_1"), + workflowEdge("reply_1", "send_1"), + workflowEdge("fallback_reply_1", "send_fallback_1"), + workflowEdge("send_1", "end_1"), + workflowEdge("send_fallback_1", "end_1"), }, } } +func workflowNode(id string, nodeType string, title string, x float64, y float64, inputs map[string]dsl.Value, config any) dsl.Node { + return dsl.Node{ + ID: id, + Type: nodeType, + Meta: dsl.NodeMeta{Position: dsl.Position{X: x, Y: y}}, + Data: dsl.NodeData{ + Title: title, + Config: mustMarshalWorkflowConfig(config), + InputsValues: inputs, + }, + } +} + +func workflowInputs(name string, nodeID string, field string) map[string]dsl.Value { + return map[string]dsl.Value{name: dsl.RefValue(nodeID, field)} +} + +func workflowConditionBranch(id string, name string, targetNodeID string, nodeID string, field string, operator string, right any) dsl.ConditionBranch { + return dsl.ConditionBranch{ + ID: id, + Name: name, + TargetNodeID: targetNodeID, + Condition: &dsl.Condition{ + Left: &dsl.Value{Type: dsl.ValueTypeRef, Content: []string{nodeID, field}}, + Operator: operator, + Right: right, + }, + } +} + +func workflowEdge(source string, target string) dsl.Edge { + return dsl.Edge{SourceNodeID: source, TargetNodeID: target} +} + func mustMarshalWorkflowConfig(value any) json.RawMessage { + if value == nil { + return nil + } raw, err := json.Marshal(value) if err != nil { panic(err) diff --git a/internal/services/ai_workflow_service_test.go b/internal/services/ai_workflow_service_test.go index dfcd820..65eb78f 100644 --- a/internal/services/ai_workflow_service_test.go +++ b/internal/services/ai_workflow_service_test.go @@ -20,16 +20,15 @@ import ( func TestAIWorkflowServiceValidateDefinitionReportsErrors(t *testing.T) { setupAIWorkflowTestDB(t) result := AIWorkflowService.ValidateDefinition(dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "create_1", Type: "create_ticket"}, - {ID: "end_1", Type: "end"}, + workflowServiceTestNode("start_1", "start", nil, nil), + workflowServiceTestNode("create_1", "create_ticket", nil, nil), + workflowServiceTestNode("end_1", "end", nil, nil), }, Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "create_1"}, - {ID: "e2", Source: "create_1", Target: "end_1"}, + workflowServiceTestEdge("start_1", "create_1"), + workflowServiceTestEdge("create_1", "end_1"), }, }) @@ -79,7 +78,7 @@ func TestAIWorkflowServicePublishCreatesImmutableVersion(t *testing.T) { if err := json.Unmarshal([]byte(version.Definition), &stored); err != nil { t.Fatalf("unmarshal stored definition: %v", err) } - if stored.EntryNodeID != "start_1" { + if stored.SchemaVersion != dsl.SchemaVersion || len(stored.Nodes) == 0 { t.Fatalf("unexpected stored definition: %+v", stored) } } @@ -131,16 +130,15 @@ func TestAIWorkflowServicePublishRejectsInvalidDSL(t *testing.T) { _, err = AIWorkflowService.PublishWorkflow(request.PublishAIWorkflowRequest{ WorkflowID: workflow.ID, Definition: dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "create_1", Type: "create_ticket"}, - {ID: "end_1", Type: "end"}, + workflowServiceTestNode("start_1", "start", nil, nil), + workflowServiceTestNode("create_1", "create_ticket", nil, nil), + workflowServiceTestNode("end_1", "end", nil, nil), }, Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "create_1"}, - {ID: "e2", Source: "create_1", Target: "end_1"}, + workflowServiceTestEdge("start_1", "create_1"), + workflowServiceTestEdge("create_1", "end_1"), }, }, }, operator) @@ -164,7 +162,7 @@ func TestAIWorkflowServiceRunListAndDetail(t *testing.T) { t.Fatalf("create workflow: %v", err) } versionDefinition := validAIWorkflowDefinition() - versionDefinition.Nodes[1].Name = "运行时回复" + versionDefinition.Nodes[1].Data.Title = "运行时回复" versionDefinitionJSON, err := json.Marshal(versionDefinition) if err != nil { t.Fatalf("marshal version definition: %v", err) @@ -278,22 +276,49 @@ func setupAIWorkflowTestDB(t *testing.T) { func validAIWorkflowDefinition() dsl.Definition { return dsl.Definition{ - SchemaVersion: 1, - EntryNodeID: "start_1", + SchemaVersion: dsl.SchemaVersion, Nodes: []dsl.Node{ - {ID: "start_1", Type: "start"}, - {ID: "reply_1", Type: "send_reply", Config: json.RawMessage(`{"text":"hello"}`), Inputs: map[string]dsl.VariableSelector{ - "replyText": {NodeID: "start_1", Field: "userMessage"}, - }}, - {ID: "end_1", Type: "end"}, + workflowServiceTestNode("start_1", "start", nil, nil), + workflowServiceTestNode("reply_1", "send_reply", map[string]dsl.Value{ + "replyText": dsl.RefValue("start_1", "userMessage"), + }, map[string]any{"text": "hello"}), + workflowServiceTestNode("end_1", "end", nil, nil), }, Edges: []dsl.Edge{ - {ID: "e1", Source: "start_1", Target: "reply_1"}, - {ID: "e2", Source: "reply_1", Target: "end_1"}, + workflowServiceTestEdge("start_1", "reply_1"), + workflowServiceTestEdge("reply_1", "end_1"), }, } } +func workflowServiceTestNode(id string, nodeType string, inputs map[string]dsl.Value, config any) dsl.Node { + return dsl.Node{ + ID: id, + Type: nodeType, + Meta: dsl.NodeMeta{Position: dsl.Position{X: 0, Y: 0}}, + Data: dsl.NodeData{ + Title: nodeType, + InputsValues: inputs, + Config: mustMarshalWorkflowServiceTestConfig(config), + }, + } +} + +func workflowServiceTestEdge(source string, target string) dsl.Edge { + return dsl.Edge{SourceNodeID: source, TargetNodeID: target} +} + +func mustMarshalWorkflowServiceTestConfig(value any) json.RawMessage { + if value == nil { + return nil + } + raw, err := json.Marshal(value) + if err != nil { + panic(err) + } + return raw +} + func aiWorkflowTestOperator() *dto.AuthPrincipal { return &dto.AuthPrincipal{ UserID: 1, diff --git a/web/app/dashboard/ai-agents/_components/config-workbench.tsx b/web/app/dashboard/ai-agents/_components/config-workbench.tsx index 3b76e45..0de1190 100644 --- a/web/app/dashboard/ai-agents/_components/config-workbench.tsx +++ b/web/app/dashboard/ai-agents/_components/config-workbench.tsx @@ -70,6 +70,7 @@ import { IMConversationServiceMode, Status, } from "@/lib/generated/enums" +import { useWorkflowDefinitionHistory } from "../../ai-workflows/_components/use-workflow-definition-history" import { WorkflowEditor } from "../../ai-workflows/_components/workflow-editor" type DirectToolItem = CreateAIAgentPayload["directTools"][number] @@ -88,25 +89,22 @@ type SectionKey = | "workflow" const fallbackDefinition: AIWorkflowDefinition = { - schemaVersion: 1, - entryNodeId: "start_1", + schemaVersion: 2, nodes: [ { id: "start_1", type: "start", - name: "开始", - position: { x: 0, y: 80 }, - config: {}, + meta: { position: { x: 0, y: 80 } }, + data: { title: "开始", config: {}, inputsValues: {} }, }, { id: "end_1", type: "end", - name: "结束", - position: { x: 260, y: 80 }, - config: {}, + meta: { position: { x: 260, y: 80 } }, + data: { title: "结束", config: {}, inputsValues: {} }, }, ], - edges: [{ id: "edge_start_end", source: "start_1", target: "end_1" }], + edges: [{ sourceNodeID: "start_1", targetNodeID: "end_1", sourcePortID: "edge_start_end" }], } function toText(value: string | number | undefined | null) { @@ -156,8 +154,16 @@ export function AIAgentConfigWorkbench({ const [selectedSkillIds, setSelectedSkillIds] = useState([]) const [directTools, setDirectTools] = useState([]) - const [definition, setDefinition] = useState(fallbackDefinition) - const [workflowEditorKey, setWorkflowEditorKey] = useState(0) + const { + definition, + revision: workflowRevision, + canUndo: canUndoWorkflow, + canRedo: canRedoWorkflow, + replace: replaceWorkflowHistory, + update: updateWorkflowDefinition, + undo: undoWorkflowDefinition, + redo: redoWorkflowDefinition, + } = useWorkflowDefinitionHistory(fallbackDefinition) const [aiConfigs, setAIConfigs] = useState([]) const [knowledgeBases, setKnowledgeBases] = useState([]) @@ -175,9 +181,8 @@ export function AIAgentConfigWorkbench({ }, [agentId]) const replaceWorkflowDefinition = useCallback((nextDefinition: AIWorkflowDefinition) => { - setDefinition(nextDefinition) - setWorkflowEditorKey((current) => current + 1) - }, []) + replaceWorkflowHistory(nextDefinition) + }, [replaceWorkflowHistory]) const loadData = useCallback(async () => { setLoading(true) @@ -811,10 +816,14 @@ export function AIAgentConfigWorkbench({ {activeSection === "workflow" ? ( & { - nodeId: string - nodeType: string - name: string - executed: boolean - statusName?: string - durationMs?: number - errorMessage?: string - selected?: boolean -} - -type AuditNode = Node -type AuditEdge = Edge<{ executed?: boolean }> - -type BranchDecision = { - selectedEdgeId?: string - selectedBranchId?: string - selectedBranchName?: string - selectedTargetNodeId?: string - reason?: string - evaluations?: BranchEvaluation[] -} - -type BranchEvaluation = { - edgeId?: string - branchId?: string - branchName?: string - targetNodeId?: string - sourceNodeId?: string - sourceField?: string - operator?: string - leftValue?: unknown - rightValue?: unknown - matched?: boolean -} - -const auditNodeTypes = { - auditNode: AuditCanvasNode, -} - -const auditEdgeTypes = { - auditEdge: AuditCanvasEdge, -} - -const fitViewOptions = { - padding: 0.12, - minZoom: 0.32, - maxZoom: 1, -} - -const defaultEdgeOptions = { - type: "auditEdge", - markerEnd: { - type: MarkerType.ArrowClosed, - }, -} - -const auditLayoutScale = { - x: 1.35, - y: 1.15, -} +import { useFlowgramEditorProps } from "../../ai-workflows/_components/flowgram-editor-provider" export function WorkflowRunAuditGraph({ run }: { run: AIWorkflowRun }) { const nodeRuns = run.nodes ?? [] - const nodeRunByNodeId = useMemo(() => { - const map = new Map() - for (const node of nodeRuns) { - map.set(node.nodeId, node) - } - return map - }, [nodeRuns]) - const activeEdgeIds = useMemo(() => buildActiveEdgeIds(run.definition, nodeRuns), [run.definition, nodeRuns]) - const firstExecutedNodeId = nodeRuns[0]?.nodeId ?? run.definition?.entryNodeId ?? "" - const [selectedNodeId, setSelectedNodeId] = useState(firstExecutedNodeId) - - const nodes = useMemo(() => { - return (run.definition?.nodes ?? []).map((node) => { - const nodeRun = nodeRunByNodeId.get(node.id) - return { - id: node.id, - type: "auditNode", - position: scaleAuditPosition(node.position), - data: { - nodeId: node.id, - nodeType: node.type, - name: node.name || node.id, - executed: Boolean(nodeRun), - statusName: nodeRun?.statusName, - durationMs: nodeRun?.durationMs, - errorMessage: nodeRun?.errorMessage, - selected: selectedNodeId === node.id, - }, - } - }) - }, [nodeRunByNodeId, run.definition?.nodes, selectedNodeId]) - - const edges = useMemo(() => { - return (run.definition?.edges ?? []).map((edge) => ({ - id: edge.id, - source: edge.source, - target: edge.target, - type: "auditEdge", - data: { - executed: activeEdgeIds.has(edge.id), - }, - })) - }, [activeEdgeIds, run.definition?.edges]) - - const selectedNodeRun = selectedNodeId ? nodeRunByNodeId.get(selectedNodeId) : undefined - const selectedDefinitionNode = run.definition?.nodes?.find((node) => node.id === selectedNodeId) - - if (!run.definition?.nodes?.length) { - return ( -
- 流程定义快照缺失，仍可查看下方节点运行明细。 -
- ) - } - - return ( -
-
- setSelectedNodeId(node.id)} - > - - - -
- -
- ) -} - -function buildActiveEdgeIds(definition: AIWorkflowDefinition | undefined, nodeRuns: AIWorkflowNodeRun[]) { - const active = new Set() - const edges = definition?.edges ?? [] - for (let i = 0; i < nodeRuns.length - 1; i += 1) { - const source = nodeRuns[i]?.nodeId - const target = nodeRuns[i + 1]?.nodeId - const edge = edges.find((item) => item.source === source && item.target === target) - if (edge) { - active.add(edge.id) - } - } - return active -} - -function AuditCanvasEdge({ - id, - sourceX, - sourceY, - targetX, - targetY, - sourcePosition, - targetPosition, - markerEnd, - data, -}: EdgeProps) { - const [edgePath] = getBezierPath({ - sourceX, - sourceY, - sourcePosition, - targetX, - targetY, - targetPosition, - curvature: 0.18, + const firstNodeId = nodeRuns[0]?.nodeId ?? run.definition?.nodes?.[0]?.id ?? "" + const [selectedNodeId, setSelectedNodeId] = useState(firstNodeId) + const selectedNodeRun = nodeRuns.find((item) => item.nodeId === selectedNodeId) ?? nodeRuns[0] + const executedNodeIds = useMemo(() => new Set(nodeRuns.map((item) => item.nodeId)), [nodeRuns]) + const editorProps = useFlowgramEditorProps({ + definition: run.definition ?? { schemaVersion: 2, nodes: [], edges: [] }, + nodeSpecs: [], + readonly: true, }) - const executed = Boolean(data?.executed) + return ( - - ) -} - -function AuditCanvasNode({ data }: NodeProps) { - const executed = Boolean(data.executed) - const failed = data.statusName === "failed" || Boolean(data.errorMessage) - const interrupted = data.statusName === "interrupted" - const selected = Boolean(data.selected) - const condition = data.nodeType === "condition" - const toneClass = failed - ? "border-destructive bg-destructive/5 text-destructive" - : interrupted - ? "border-amber-500 bg-amber-500/10 text-amber-700" - : executed - ? "border-emerald-500 bg-emerald-500/10 text-emerald-700" - : "border-border bg-muted/40 text-muted-foreground" - - if (condition) { - return ( -
- -
-
- -
{data.name}
-
{data.statusName || "未执行"}
+
+
+ + + +
+ + + 已执行 {executedNodeIds.size} + + {run.errorMessage ? ( + + + 异常 + + ) : null}
-
- ) - } - - return ( -
- -
-
- {failed ? : } -
-
{data.name}
-
{data.nodeType}
+
-
- {data.statusName || "未执行"} - {executed ? ( - - - {data.durationMs ?? 0} ms - - ) : null} -
- +
) } -function AuditSidePanel({ - definitionNode, - nodeRun, -}: { - definitionNode?: AIWorkflowDefinition["nodes"][number] - nodeRun?: AIWorkflowNodeRun -}) { - const inputValue = safeParseJSON(nodeRun?.inputPreview ?? "") - const outputValue = safeParseJSON(nodeRun?.outputPreview ?? "") - const branchDecision = extractBranchDecision(outputValue) - - return ( - -
-
-
- -

{definitionNode?.name || nodeRun?.nodeId || "节点详情"}

-
-
- {definitionNode?.id || nodeRun?.nodeId || "-"} · {definitionNode?.type || nodeRun?.nodeType || "unknown"} -
-
- - {nodeRun ? ( -
- - - - -
- ) : ( -
- 本次运行没有执行该节点。 -
- )} - - {nodeRun?.errorMessage ? ( -
- {nodeRun.errorMessage} -
- ) : null} - - {branchDecision ? : null} - - - -
- - ) -} - -function AuditMeta({ label, value }: { label: string; value: string }) { - return ( -
-
{label}
-
{value}
-
- ) -} - -function scaleAuditPosition(position: AIWorkflowDefinition["nodes"][number]["position"] | undefined) { - return { - x: Math.round((position?.x ?? 0) * auditLayoutScale.x), - y: Math.round((position?.y ?? 0) * auditLayoutScale.y), +function NodeRunPreview({ nodeRun }: { nodeRun?: AIWorkflowNodeRun }) { + if (!nodeRun) { + return
暂无节点执行记录。
} -} - -function BranchDecisionBlock({ decision }: { decision: BranchDecision }) { return ( -
-
-
分支决策
- {decision.selectedBranchName || decision.selectedBranchId || "default"} -
-
-
目标节点：{decision.selectedTargetNodeId || "-"}
-
原因：{decision.reason || "-"}
-
- {decision.evaluations?.length ? ( -
- {decision.evaluations.map((item, index) => ( -
-
- {item.branchName || item.branchId || item.edgeId || `条件 ${index + 1}`} - {item.matched ? "命中" : "未命中"} -
-
- {item.sourceNodeId}.{item.sourceField} {item.operator} {formatUnknown(item.rightValue)} -
-
- 实际值：{formatUnknown(item.leftValue)} -
-
- ))} +
+ {nodeRun.errorMessage ? ( +
+ {nodeRun.errorMessage}
) : null} + +
) } -function PreviewBlock({ title, raw, value }: { title: string; raw: string; value: unknown }) { +function PreviewBlock({ title, value }: { title: string; value?: string }) { return ( -
-
{title}
- {value !== null ? ( - - ) : raw.trim() ? ( - 
-          {raw}
-
- ) : ( -
-
- )} +
+
{title}
+ {value ? :
无
}
) } -function extractBranchDecision(value: unknown): BranchDecision | null { - if (!value || typeof value !== "object") { - return null - } - const record = value as Record - const decision = record.branchDecision - if (!decision || typeof decision !== "object") { - return null - } - return decision as BranchDecision -} - -function safeParseJSON(raw: string): unknown | null { - const trimmed = raw.trim() - if (!trimmed) { - return null - } +function parsePreview(value: string) { try { - return JSON.parse(trimmed) + return JSON.parse(value) } catch { - return null - } -} - -function formatUnknown(value: unknown) { - if (typeof value === "string") { return value } - if (value === null || value === undefined) { - return "-" - } - try { - return JSON.stringify(value) - } catch { - return String(value) - } } diff --git a/web/app/dashboard/ai-workflows/_components/flowgram-editor-provider.tsx b/web/app/dashboard/ai-workflows/_components/flowgram-editor-provider.tsx new file mode 100644 index 0000000..2f36d5c --- /dev/null +++ b/web/app/dashboard/ai-workflows/_components/flowgram-editor-provider.tsx @@ -0,0 +1,68 @@ +"use client" + +import { useMemo } from "react" + +import { + type FreeLayoutProps, + type WorkflowJSON, +} from "@flowgram.ai/free-layout-editor" + +import type { AIWorkflowDefinition, AIWorkflowNodeSpec } from "@/lib/api/admin" + +import { FlowgramNodeRenderer } from "./flowgram-node-renderer" +import { buildFlowgramNodeRegistries } from "./flowgram-node-registries" + +export function useFlowgramEditorProps({ + definition, + nodeSpecs, + readonly = false, + onDefinitionChange, +}: { + definition: AIWorkflowDefinition + nodeSpecs: AIWorkflowNodeSpec[] + readonly?: boolean + onDefinitionChange?: (definition: AIWorkflowDefinition) => void +}) { + return useMemo( + () => ({ + background: true, + readonly, + initialData: definition as WorkflowJSON, + nodeRegistries: buildFlowgramNodeRegistries(nodeSpecs), + materials: { + renderDefaultNode: FlowgramNodeRenderer, + }, + nodeEngine: { + enable: false, + }, + history: { + enable: !readonly, + enableChangeNode: !readonly, + }, + canDeleteNode: (_ctx, node) => { + const type = String(node.flowNodeType ?? "") + return type !== "start" && type !== "end" + }, + canDeleteLine: () => !readonly, + onContentChange: (ctx) => { + if (readonly) { + return + } + onDefinitionChange?.(ctx.document.toJSON() as AIWorkflowDefinition) + }, + onAllLayersRendered: (ctx) => { + void ctx.tools.fitView(false) + }, + getNodeDefaultRegistry(type) { + return { + type, + meta: { + defaultExpanded: true, + }, + } + }, + plugins: () => [], + }), + [definition, nodeSpecs, onDefinitionChange, readonly] + ) +} diff --git a/web/app/dashboard/ai-workflows/_components/flowgram-node-registries.tsx b/web/app/dashboard/ai-workflows/_components/flowgram-node-registries.tsx new file mode 100644 index 0000000..2f5e1c8 --- /dev/null +++ b/web/app/dashboard/ai-workflows/_components/flowgram-node-registries.tsx @@ -0,0 +1,41 @@ +import type { WorkflowNodeRegistry } from "@flowgram.ai/free-layout-editor" + +import type { AIWorkflowNodeSpec } from "@/lib/api/admin" + +export function buildFlowgramNodeRegistries(nodeSpecs: AIWorkflowNodeSpec[]): WorkflowNodeRegistry[] { + const seen = new Set() + const specs = nodeSpecs.length > 0 + ? nodeSpecs + : [ + { type: "start", title: "开始" }, + { type: "end", title: "结束" }, + ] + + return specs + .filter((spec) => { + if (!spec.type || seen.has(spec.type)) { + return false + } + seen.add(spec.type) + return true + }) + .map((spec) => ({ + type: spec.type, + meta: { + defaultExpanded: true, + deleteDisable: spec.type === "start" || spec.type === "end", + copyDisable: spec.type === "start" || spec.type === "end", + defaultPorts: defaultPortsForNodeType(spec.type), + }, + })) +} + +function defaultPortsForNodeType(type: string) { + if (type === "start") { + return [{ type: "output" as const }] + } + if (type === "end") { + return [{ type: "input" as const }] + } + return [{ type: "input" as const }, { type: "output" as const }] +} diff --git a/web/app/dashboard/ai-workflows/_components/flowgram-node-renderer.tsx b/web/app/dashboard/ai-workflows/_components/flowgram-node-renderer.tsx new file mode 100644 index 0000000..8480574 --- /dev/null +++ b/web/app/dashboard/ai-workflows/_components/flowgram-node-renderer.tsx @@ -0,0 +1,63 @@ +import "@flowgram.ai/free-layout-editor/index.css" + +import { + useNodeRender, + WorkflowNodeRenderer, + type WorkflowNodeProps, +} from "@flowgram.ai/free-layout-editor" +import { + BotIcon, + CircleStopIcon, + DatabaseIcon, + GitBranchIcon, + MessageSquareTextIcon, + SendIcon, + UserRoundIcon, +} from "lucide-react" +import type { ComponentType } from "react" + +import { cn } from "@/lib/utils" + +const iconByType: Record> = { + start: UserRoundIcon, + conversation_understanding: BotIcon, + reply_policy: MessageSquareTextIcon, + condition: GitBranchIcon, + knowledge_retrieve: DatabaseIcon, + answerability_gate: GitBranchIcon, + llm_reply: BotIcon, + human_confirm: UserRoundIcon, + create_ticket: MessageSquareTextIcon, + handoff_to_human: UserRoundIcon, + send_reply: SendIcon, + end: CircleStopIcon, +} + +export function FlowgramNodeRenderer(props: WorkflowNodeProps) { + const { selected, node } = useNodeRender() + const nodeType = String(node.flowNodeType ?? "") + const Icon = iconByType[nodeType] ?? BotIcon + const nodeJSON = node.toJSON?.() as { data?: { title?: string }; title?: string } | undefined + const title = nodeJSON?.data?.title || nodeJSON?.title || nodeType || "节点" + + return ( + +
+
+ +
+
+
{title}
+
{nodeType}
+
+
+ + ) +} diff --git a/web/app/dashboard/ai-workflows/_components/node-config-panel.tsx b/web/app/dashboard/ai-workflows/_components/node-config-panel.tsx index ef7a493..523f7c1 100644 --- a/web/app/dashboard/ai-workflows/_components/node-config-panel.tsx +++ b/web/app/dashboard/ai-workflows/_components/node-config-panel.tsx @@ -1,577 +1,363 @@ "use client" -import { useState } from "react" -import type { Node } from "@xyflow/react" +import { useEffect, useMemo, useState } from "react" +import { Trash2Icon } from "lucide-react" import { Button } from "@/components/ui/button" import { Input } from "@/components/ui/input" import { Label } from "@/components/ui/label" import { Textarea } from "@/components/ui/textarea" import { OptionCombobox } from "@/components/option-combobox" -import { VariableSelector } from "./variable-selector" -import type { - WorkflowConditionBranch, - WorkflowNodeSpec, - WorkflowNodeConfig, - WorkflowVariableRef, - WorkflowVariableSpec, - WorkflowVariableSelector, -} from "./workflow-utils" +import type { AIWorkflowDefinition, AIWorkflowNodeSpec } from "@/lib/api/admin" +import { cn } from "@/lib/utils" -type WorkflowNodeData = Record & { - nodeType?: string - name?: string - title?: string - config?: WorkflowNodeConfig - inputs?: Record -} +import { VariableSelector } from "./variable-selector" +import { + createConditionBranchID, + isRefValue, + normalizeNodeConfig, + type WorkflowConditionBranch, + type WorkflowVariableRef, +} from "./workflow-utils" export type WorkflowBranchSummary = { branchId: string - targetNodeId: string - targetName: string - conditionLabel: string - conditionSet: boolean - isDefault: boolean + targetNodeId?: string + targetName?: string } export function NodeConfigPanel({ node, nodeSpec, + nodes, availableVariables, - branchSummaries = [], onChange, + onDelete, }: { - node: Node | null - nodeSpec?: WorkflowNodeSpec - availableVariables: WorkflowVariableRef[] + node: AIWorkflowDefinition["nodes"][number] | null + nodeSpec?: AIWorkflowNodeSpec + nodes: AIWorkflowDefinition["nodes"] + availableVariables?: WorkflowVariableRef[] branchSummaries?: WorkflowBranchSummary[] - onChange: (nodeId: string, data: WorkflowNodeData) => void + onChange: (nodeId: string, data: AIWorkflowDefinition["nodes"][number]["data"]) => void + onDelete?: (nodeId: string) => void }) { + const [configText, setConfigText] = useState("{}") + + useEffect(() => { + setConfigText(JSON.stringify(node?.data?.config ?? {}, null, 2)) + }, [node?.id, node?.data?.config]) + + const configError = useMemo(() => { + if (!node) { + return "" + } + try { + const parsed = JSON.parse(configText || "{}") + return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? "" : "配置必须是 JSON 对象" + } catch { + return "JSON 格式错误" + } + }, [configText, node]) + if (!node) { return ( -
- 选择一个节点后，可以配置输入映射并查看输出变量。 +
+ 未选择节点
) } - return ( - - ) -} - -function NodeConfigForm({ - node, - nodeSpec, - availableVariables, - branchSummaries, - onChange, -}: { - node: Node - nodeSpec?: WorkflowNodeSpec - availableVariables: WorkflowVariableRef[] - branchSummaries: WorkflowBranchSummary[] - onChange: (nodeId: string, data: WorkflowNodeData) => void -}) { - const [name, setName] = useState(node.data.name ?? "") - const [configText, setConfigText] = useState(JSON.stringify(node.data.config ?? {}, null, 2)) - const [inputs, setInputs] = useState>( - node.data.inputs ?? {} - ) - const [error, setError] = useState("") + const inputsValues = node.data?.inputsValues ?? {} const inputSchema = nodeSpec?.inputSchema ?? [] - const outputSchema = nodeSpec?.outputSchema ?? [] - const isConditionNode = node.data.nodeType === "condition" - const fallbackNodeName = nodeSpec?.title || node.data.title || node.data.nodeType || node.id - const panelTitle = name.trim() || node.data.name?.trim() || fallbackNodeName + const canDelete = node.type !== "start" && node.type !== "end" + const config = normalizeNodeConfig(node.data?.config) + const branches = config.branches ?? [] - const commitChange = (next: Partial) => { + const updateData = (data: Partial) => { onChange(node.id, { - ...node.data, - name: name.trim() || fallbackNodeName, - config: node.data.config ?? {}, - inputs, - ...next, + ...(node.data ?? {}), + ...data, + }) + } + const updateConfig = (nextConfig: Record) => updateData({ config: nextConfig }) + const updateBranch = (branch: WorkflowConditionBranch) => { + const nextBranches = branches.some((item) => item.id === branch.id) + ? branches.map((item) => (item.id === branch.id ? branch : item)) + : [...branches, branch] + updateConfig({ ...config, branches: nextBranches }) + } + const deleteBranch = (branchId: string) => { + updateConfig({ ...config, branches: branches.filter((branch) => branch.id !== branchId) }) + } + const addBranch = () => { + const targetNodeId = nodes.find((item) => item.id !== node.id && item.type !== "start")?.id ?? "" + updateBranch({ + id: createConditionBranchID(branches), + name: "新分支", + targetNodeId, + condition: { + operator: "eq", + }, }) } - const handleApply = () => { - try { - const parsed = JSON.parse(configText || "{}") as Record - setError("") - commitChange({ config: parsed }) - } catch { - setError("Config must be valid JSON.") - } - } - return ( -
-
-
- setName(event.target.value)} - onBlur={() => commitChange({ name: name.trim() || node.data.nodeType || node.id })} - className="h-8 border-0 bg-transparent px-0 text-sm font-semibold uppercase shadow-none focus-visible:ring-0" - aria-label="节点名称" - /> -
- {node.data.nodeType && node.data.nodeType !== panelTitle - ? `${node.id} · ${node.data.nodeType}` - : node.id} +
+
+
+
+
{node.data?.title || nodeSpec?.title || node.type}
+
{node.id}
+ {canDelete ? ( + + ) : null}
-
- {isConditionNode ? ( - commitChange({ config: { ...(node.data.config ?? {}), branches } })} +
+
+ + updateData({ title: event.target.value })} /> - ) : ( - <> - {inputSchema.length > 0 ? ( -
-
输入映射
- {availableVariables.length === 0 ? ( -
- 当前节点前面还没有可用变量，请先连接上游节点。 -
- ) : null} - {inputSchema.map((input) => ( +
+ + {inputSchema.length > 0 ? ( +
+
输入
+ {inputSchema.map((input) => { + const value = inputsValues[input.name] + return (
-
- - {input.type} -
+ { - const nextInputs = { - ...inputs, - [input.name]: value, - } - setInputs(nextInputs) - commitChange({ - inputs: nextInputs, + value={isRefValue(value) ? value : undefined} + variables={availableVariables ?? []} + placeholder="选择变量" + onChange={(next) => { + updateData({ + inputsValues: { + ...inputsValues, + [input.name]: next, + }, }) }} /> - {inputs[input.name] ? ( -
- 已选择：{inputs[input.name].nodeId}.{inputs[input.name].field} -
- ) : null} {input.description ? (
{input.description}
) : null}
- ))} -
- ) : null} -
- 高级配置 JSON -
-