From d1f39ae57e9178fd4bd3232589952835fb9e2809 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Mon, 22 Jun 2026 18:33:39 +0800 Subject: [PATCH] feat: Enhance AI Workflow with Variable Contracts and Input Mapping - Added ConfigSchema, InputSchema, OutputSchema, and DefaultInputs to AIWorkflowNodeSpecResponse. - Implemented BuildAIWorkflowNodeSpecs to include variable contracts for start and send_reply nodes. - Introduced applyAutoInputMappings to automatically map inputs based on node connections. - Enhanced validation to check for required input mappings in workflows. - Updated workflow editor to support variable selection for node inputs. - Translated node names and labels to Chinese for better localization. - Added tests for variable mapping and validation logic. --- internal/ai/workflow/dsl/types.go | 21 +- internal/ai/workflow/registry/registry.go | 182 +++++++++++++- .../ai/workflow/registry/registry_test.go | 60 +++++ internal/ai/workflow/registry/spec.go | 38 ++- internal/ai/workflow/validator/validator.go | 111 +++++++++ .../ai/workflow/validator/validator_test.go | 109 ++++++++- internal/builders/ai_workflow_builder.go | 4 + internal/builders/ai_workflow_builder_test.go | 40 ++++ .../pkg/dto/response/ai_workflow_response.go | 16 +- .../ai_agent_workflow_service_test.go | 15 ++ internal/services/ai_workflow_service.go | 21 +- internal/services/ai_workflow_service_test.go | 4 +- .../_components/config-workbench.tsx | 46 +++- .../_components/node-config-panel.tsx | 138 +++++++++-- .../_components/variable-selector.tsx | 35 +++ .../_components/workflow-editor.tsx | 163 ++++++++++++- .../_components/workflow-utils.test.mjs | 149 ++++++++++++ .../_components/workflow-utils.ts | 226 +++++++++++++++++- web/lib/api/admin.ts | 27 +++ 19 files changed, 1332 insertions(+), 73 deletions(-) create mode 100644 internal/ai/workflow/registry/registry_test.go create mode 100644 internal/builders/ai_workflow_builder_test.go create mode 100644 web/app/dashboard/ai-workflows/_components/variable-selector.tsx diff --git a/internal/ai/workflow/dsl/types.go b/internal/ai/workflow/dsl/types.go index 1812bed..5cd9a98 100644 --- a/internal/ai/workflow/dsl/types.go +++ b/internal/ai/workflow/dsl/types.go @@ -10,11 +10,12 @@ type Definition struct { } type Node struct { - ID string `json:"id"` - Type string `json:"type"` - Name string `json:"name"` - Position Position `json:"position"` - Config json.RawMessage `json:"config"` + 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"` } type Position struct { @@ -30,5 +31,13 @@ type Edge struct { } type Condition struct { - Expression string `json:"expression"` + Expression string `json:"expression,omitempty"` + Left *VariableSelector `json:"left,omitempty"` + Operator string `json:"operator,omitempty"` + Right any `json:"right,omitempty"` +} + +type VariableSelector struct { + NodeID string `json:"nodeId"` + Field string `json:"field"` } diff --git a/internal/ai/workflow/registry/registry.go b/internal/ai/workflow/registry/registry.go index 13465c7..fa17763 100644 --- a/internal/ai/workflow/registry/registry.go +++ b/internal/ai/workflow/registry/registry.go @@ -1,5 +1,7 @@ package registry +import "agent-desk/internal/ai/workflow/dsl" + const ( NodeTypeStart = "start" NodeTypeKnowledgeRetrieve = "knowledge_retrieve" @@ -17,17 +19,173 @@ const ( func DefaultRegistry() *Registry { return NewRegistry( - NodeSpec{Type: NodeTypeStart, Title: "Start", Description: "Conversation workflow entry.", RiskLevel: NodeRiskLevelLow}, - NodeSpec{Type: NodeTypeKnowledgeRetrieve, Title: "Knowledge Retrieve", Description: "Retrieve knowledge for the current user message.", RiskLevel: NodeRiskLevelLow}, - NodeSpec{Type: NodeTypeAnswerabilityGate, Title: "Answerability Gate", Description: "Decide whether retrieved knowledge is enough to answer.", RiskLevel: NodeRiskLevelLow}, - NodeSpec{Type: NodeTypeLLMReply, Title: "LLM Reply", Description: "Generate a reply or structured analysis with the configured model.", RiskLevel: NodeRiskLevelMedium}, - NodeSpec{Type: NodeTypeCondition, Title: "Condition", Description: "Route by controlled workflow variables.", RiskLevel: NodeRiskLevelLow}, - NodeSpec{Type: NodeTypeAnalyzeConversation, Title: "Analyze Conversation", Description: "Analyze intent, risk, and recommended next action.", RiskLevel: NodeRiskLevelLow}, - NodeSpec{Type: NodeTypePrepareTicketDraft, Title: "Prepare Ticket Draft", Description: "Build a ticket draft from conversation context.", RiskLevel: NodeRiskLevelMedium}, - NodeSpec{Type: NodeTypeHumanConfirm, Title: "Human Confirm", Description: "Interrupt and wait for explicit user confirmation.", RiskLevel: NodeRiskLevelMedium, Interruptible: true}, - NodeSpec{Type: NodeTypeCreateTicket, Title: "Create Ticket", Description: "Create a ticket from a confirmed draft.", RiskLevel: NodeRiskLevelHigh, RequiresConfirmationPredecessor: true}, - NodeSpec{Type: NodeTypeHandoffToHuman, Title: "Handoff To Human", Description: "Transfer the conversation to human support.", RiskLevel: NodeRiskLevelHigh, RequiresConfirmationPredecessor: true}, - NodeSpec{Type: NodeTypeSendReply, Title: "Send Reply", Description: "Return or commit customer-visible reply text.", RiskLevel: NodeRiskLevelLow}, - NodeSpec{Type: NodeTypeEnd, Title: "End", Description: "End workflow execution.", RiskLevel: NodeRiskLevelLow}, + NodeSpec{ + Type: NodeTypeStart, + Title: "Start", + Description: "Conversation workflow entry.", + RiskLevel: NodeRiskLevelLow, + OutputSchema: []VariableSpec{ + output("conversationId", VariableTypeInteger, "Conversation ID."), + output("messageId", VariableTypeInteger, "Current user message ID."), + output("aiAgentId", VariableTypeInteger, "AI Agent ID."), + output("userMessage", VariableTypeString, "Current user message content."), + output("knowledgeBaseIds", VariableTypeIntegerArray, "Knowledge bases bound to the AI Agent."), + }, + }, + NodeSpec{ + Type: NodeTypeKnowledgeRetrieve, + Title: "Knowledge Retrieve", + Description: "Retrieve knowledge for the current user message.", + RiskLevel: NodeRiskLevelLow, + InputSchema: []VariableSpec{ + requiredInput("query", VariableTypeString, "Search query."), + }, + OutputSchema: []VariableSpec{ + output("items", VariableTypeObjectArray, "Retrieved knowledge items."), + output("summary", VariableTypeString, "Short retrieval summary."), + }, + DefaultInputs: map[string]dsl.VariableSelector{ + "query": {NodeID: "start_1", Field: "userMessage"}, + }, + }, + NodeSpec{ + Type: NodeTypeAnswerabilityGate, + Title: "Answerability Gate", + Description: "Decide whether retrieved knowledge is enough to answer.", + RiskLevel: NodeRiskLevelLow, + InputSchema: []VariableSpec{ + requiredInput("userMessage", VariableTypeString, "Current user message content."), + requiredInput("knowledgeItems", VariableTypeObjectArray, "Retrieved knowledge items."), + }, + OutputSchema: []VariableSpec{ + output("answerability", VariableTypeString, "Answerability decision."), + output("reason", VariableTypeString, "Decision reason."), + }, + }, + NodeSpec{ + Type: NodeTypeLLMReply, + Title: "LLM Reply", + Description: "Generate a reply or structured analysis with the configured model.", + RiskLevel: NodeRiskLevelMedium, + InputSchema: []VariableSpec{ + requiredInput("userMessage", VariableTypeString, "Current user message content."), + optionalInput("knowledgeItems", VariableTypeObjectArray, "Retrieved knowledge items."), + }, + OutputSchema: []VariableSpec{ + output("replyText", VariableTypeString, "Generated reply text."), + }, + }, + NodeSpec{ + Type: NodeTypeCondition, + Title: "Condition", + Description: "Route by controlled workflow variables.", + RiskLevel: NodeRiskLevelLow, + OutputSchema: []VariableSpec{ + output("matched", VariableTypeBoolean, "Whether the condition matched."), + }, + }, + NodeSpec{ + Type: NodeTypeAnalyzeConversation, + Title: "Analyze Conversation", + Description: "Analyze intent, risk, and recommended next action.", + RiskLevel: NodeRiskLevelLow, + InputSchema: []VariableSpec{ + requiredInput("userMessage", VariableTypeString, "Current user message content."), + }, + OutputSchema: []VariableSpec{ + output("intent", VariableTypeString, "Detected user intent."), + output("riskLevel", VariableTypeString, "Detected risk level."), + output("needTicket", VariableTypeBoolean, "Whether a ticket is recommended."), + output("needHumanHandoff", VariableTypeBoolean, "Whether human handoff is recommended."), + }, + }, + NodeSpec{ + Type: NodeTypePrepareTicketDraft, + Title: "Prepare Ticket Draft", + Description: "Build a ticket draft from conversation context.", + RiskLevel: NodeRiskLevelMedium, + InputSchema: []VariableSpec{ + requiredInput("issue", VariableTypeString, "Issue summary."), + }, + OutputSchema: []VariableSpec{ + output("ticketDraft", VariableTypeObject, "Draft ticket payload."), + }, + }, + NodeSpec{ + Type: NodeTypeHumanConfirm, + Title: "Human Confirm", + Description: "Interrupt and wait for explicit user confirmation.", + RiskLevel: NodeRiskLevelMedium, + Interruptible: true, + InputSchema: []VariableSpec{ + requiredInput("prompt", VariableTypeString, "Confirmation prompt."), + }, + OutputSchema: []VariableSpec{ + output("confirmed", VariableTypeBoolean, "Whether the user confirmed."), + output("responseText", VariableTypeString, "Confirmation response text."), + }, + }, + NodeSpec{ + Type: NodeTypeCreateTicket, + Title: "Create Ticket", + Description: "Create a ticket from a confirmed draft.", + RiskLevel: NodeRiskLevelHigh, + RequiresConfirmationPredecessor: true, + InputSchema: []VariableSpec{ + requiredInput("ticketDraft", VariableTypeObject, "Confirmed draft ticket payload."), + requiredInput("confirmed", VariableTypeBoolean, "Confirmation result."), + }, + OutputSchema: []VariableSpec{ + output("ticketId", VariableTypeInteger, "Created ticket ID."), + }, + }, + NodeSpec{ + Type: NodeTypeHandoffToHuman, + Title: "Handoff To Human", + Description: "Transfer the conversation to human support.", + RiskLevel: NodeRiskLevelHigh, + RequiresConfirmationPredecessor: true, + InputSchema: []VariableSpec{ + requiredInput("reason", VariableTypeString, "Handoff reason."), + requiredInput("confirmed", VariableTypeBoolean, "Confirmation result."), + }, + OutputSchema: []VariableSpec{ + output("handoffId", VariableTypeInteger, "Handoff operation ID."), + }, + }, + NodeSpec{ + Type: NodeTypeSendReply, + Title: "Send Reply", + Description: "Return or commit customer-visible reply text.", + RiskLevel: NodeRiskLevelLow, + InputSchema: []VariableSpec{ + requiredInput("replyText", VariableTypeString, "Customer-visible reply text."), + }, + OutputSchema: []VariableSpec{ + output("sent", VariableTypeBoolean, "Whether the reply was sent."), + output("replyMessageId", VariableTypeInteger, "Reply message ID."), + }, + }, + NodeSpec{ + Type: NodeTypeEnd, + Title: "End", + Description: "End workflow execution.", + RiskLevel: NodeRiskLevelLow, + OutputSchema: []VariableSpec{ + output("status", VariableTypeString, "Workflow terminal status."), + }, + }, ) } + +func requiredInput(name string, variableType VariableType, description string) VariableSpec { + return VariableSpec{Name: name, Type: variableType, Required: true, Description: description} +} + +func optionalInput(name string, variableType VariableType, description string) VariableSpec { + return VariableSpec{Name: name, Type: variableType, Description: description} +} + +func output(name string, variableType VariableType, description string) VariableSpec { + return VariableSpec{Name: name, Type: variableType, Description: description} +} diff --git a/internal/ai/workflow/registry/registry_test.go b/internal/ai/workflow/registry/registry_test.go new file mode 100644 index 0000000..6c26275 --- /dev/null +++ b/internal/ai/workflow/registry/registry_test.go @@ -0,0 +1,60 @@ +package registry + +import "testing" + +func TestDefaultRegistryExposesStartOutputs(t *testing.T) { + spec, ok := DefaultRegistry().Get(NodeTypeStart) + if !ok { + t.Fatalf("start node spec not found") + } + if !hasVariable(spec.OutputSchema, "userMessage", VariableTypeString) { + t.Fatalf("expected start output userMessage:string, got %#v", spec.OutputSchema) + } + if !hasVariable(spec.OutputSchema, "knowledgeBaseIds", VariableTypeIntegerArray) { + t.Fatalf("expected start output knowledgeBaseIds:array, got %#v", spec.OutputSchema) + } +} + +func TestDefaultRegistryExposesKnowledgeRetrieveInputsAndOutputs(t *testing.T) { + spec, ok := DefaultRegistry().Get(NodeTypeKnowledgeRetrieve) + if !ok { + t.Fatalf("knowledge_retrieve node spec not found") + } + if !hasRequiredVariable(spec.InputSchema, "query", VariableTypeString) { + t.Fatalf("expected knowledge_retrieve required input query:string, got %#v", spec.InputSchema) + } + if !hasVariable(spec.OutputSchema, "items", VariableTypeObjectArray) { + t.Fatalf("expected knowledge_retrieve output items:array, got %#v", spec.OutputSchema) + } +} + +func TestDefaultRegistryExposesSendReplyRequiredInput(t *testing.T) { + spec, ok := DefaultRegistry().Get(NodeTypeSendReply) + if !ok { + t.Fatalf("send_reply node spec not found") + } + if !hasRequiredVariable(spec.InputSchema, "replyText", VariableTypeString) { + t.Fatalf("expected send_reply required input replyText:string, got %#v", spec.InputSchema) + } + if !hasVariable(spec.OutputSchema, "sent", VariableTypeBoolean) { + t.Fatalf("expected send_reply output sent:boolean, got %#v", spec.OutputSchema) + } +} + +func hasRequiredVariable(items []VariableSpec, name string, variableType VariableType) bool { + for _, item := range items { + if item.Name == name && item.Type == variableType && item.Required { + return true + } + } + return false +} + +func hasVariable(items []VariableSpec, name string, variableType VariableType) bool { + for _, item := range items { + if item.Name == name && item.Type == variableType { + return true + } + } + return false +} diff --git a/internal/ai/workflow/registry/spec.go b/internal/ai/workflow/registry/spec.go index 0f98acb..d492a3f 100644 --- a/internal/ai/workflow/registry/spec.go +++ b/internal/ai/workflow/registry/spec.go @@ -1,5 +1,7 @@ package registry +import "agent-desk/internal/ai/workflow/dsl" + type NodeRiskLevel string const ( @@ -8,13 +10,37 @@ const ( NodeRiskLevelHigh NodeRiskLevel = "high" ) +type VariableType string + +const ( + VariableTypeString VariableType = "string" + VariableTypeInteger VariableType = "integer" + VariableTypeBoolean VariableType = "boolean" + VariableTypeObject VariableType = "object" + VariableTypeStringArray VariableType = "array" + VariableTypeIntegerArray VariableType = "array" + VariableTypeObjectArray VariableType = "array" + VariableTypeAny VariableType = "any" +) + +type VariableSpec struct { + Name string `json:"name"` + Type VariableType `json:"type"` + Required bool `json:"required,omitempty"` + Description string `json:"description"` +} + 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"` + 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 Registry struct { diff --git a/internal/ai/workflow/validator/validator.go b/internal/ai/workflow/validator/validator.go index 2c9e7e7..5ce8ef1 100644 --- a/internal/ai/workflow/validator/validator.go +++ b/internal/ai/workflow/validator/validator.go @@ -55,6 +55,7 @@ func (v *definitionValidator) validate() { v.validateEntry() v.validateReachability() v.validateConfirmationGuards() + v.validateVariableMappings() } func (v *definitionValidator) validateNodes() { @@ -182,6 +183,116 @@ func (v *definitionValidator) validateConfirmationGuards() { } } +func (v *definitionValidator) validateVariableMappings() { + for id, node := range v.nodesByID { + spec, ok := v.registry.Get(node.Type) + if !ok { + continue + } + for _, input := range spec.InputSchema { + 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) + continue + } + v.validateInputSelector(id, input, selector) + } + 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 + } + if _, ok := findInputSpec(spec.InputSchema, inputName); ok { + continue + } + sourceNode, sourceOK := v.nodesByID[strings.TrimSpace(selector.NodeID)] + if !sourceOK { + v.addError("nodes."+id+".inputs."+inputName, "input source node does not exist: "+selector.NodeID) + 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) + } + } + } +} + +func (v *definitionValidator) validateInputSelector(nodeID string, input registry.VariableSpec, selector dsl.VariableSelector) { + sourceNodeID := strings.TrimSpace(selector.NodeID) + sourceField := strings.TrimSpace(selector.Field) + sourceNode, ok := v.nodesByID[sourceNodeID] + if !ok { + v.addError("nodes."+nodeID+".inputs."+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) + return + } + sourceSpec, ok := v.registry.Get(sourceNode.Type) + if !ok { + return + } + output, ok := findOutputSpec(sourceSpec.OutputSchema, sourceField) + if !ok { + v.addError("nodes."+nodeID+".inputs."+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)) + } +} + +func (v *definitionValidator) hasPath(sourceID string, targetID string, visiting map[string]struct{}) bool { + if sourceID == targetID { + return false + } + if _, seen := visiting[sourceID]; seen { + return false + } + visiting[sourceID] = struct{}{} + for _, next := range v.outgoing[sourceID] { + if next == targetID { + return true + } + if v.hasPath(next, targetID, visiting) { + return true + } + } + return false +} + +func findInputSpec(items []registry.VariableSpec, name string) (registry.VariableSpec, bool) { + name = strings.TrimSpace(name) + for _, item := range items { + if item.Name == name { + return item, true + } + } + return registry.VariableSpec{}, false +} + +func findOutputSpec(items []registry.VariableSpec, name string) (registry.VariableSpec, bool) { + name = strings.TrimSpace(name) + for _, item := range items { + if item.Name == name { + return item, true + } + } + return registry.VariableSpec{}, false +} + +func variableTypesCompatible(input registry.VariableType, output registry.VariableType) bool { + return input == registry.VariableTypeAny || output == registry.VariableTypeAny || input == output +} + func (v *definitionValidator) hasConfirmationPredecessor(nodeID string, visiting map[string]struct{}) bool { if _, seen := visiting[nodeID]; seen { return false diff --git a/internal/ai/workflow/validator/validator_test.go b/internal/ai/workflow/validator/validator_test.go index 065674f..6435cda 100644 --- a/internal/ai/workflow/validator/validator_test.go +++ b/internal/ai/workflow/validator/validator_test.go @@ -83,9 +83,16 @@ func TestValidateDefinitionAcceptsConfirmedCreateTicket(t *testing.T) { EntryNodeID: "start_1", Nodes: []dsl.Node{ {ID: "start_1", Type: "start"}, - {ID: "draft_1", Type: "prepare_ticket_draft"}, - {ID: "confirm_1", Type: "human_confirm"}, - {ID: "create_1", Type: "create_ticket"}, + {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{ @@ -103,13 +110,99 @@ func TestValidateDefinitionAcceptsConfirmedCreateTicket(t *testing.T) { } } +func TestValidateDefinitionRejectsMissingRequiredInputMapping(t *testing.T) { + def := minimalDefinition() + def.Nodes[1].Inputs = nil + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected missing required input mapping to be invalid") + } + if !hasValidationMessage(result, "required input mapping is missing") { + t.Fatalf("expected required-input error, got %#v", result.Errors) + } +} + +func TestValidateDefinitionRejectsUnknownInputSourceNode(t *testing.T) { + def := mappedReplyDefinition() + def.Nodes[1].Inputs["replyText"] = dsl.VariableSelector{NodeID: "missing_1", Field: "replyText"} + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected unknown input source node to be invalid") + } + if !hasValidationMessage(result, "input source node does not exist") { + t.Fatalf("expected source-node error, got %#v", result.Errors) + } +} + +func TestValidateDefinitionRejectsUnknownInputSourceField(t *testing.T) { + def := mappedReplyDefinition() + def.Nodes[1].Inputs["replyText"] = dsl.VariableSelector{NodeID: "start_1", Field: "missing"} + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected unknown input source field to be invalid") + } + if !hasValidationMessage(result, "input source field does not exist") { + t.Fatalf("expected source-field error, got %#v", result.Errors) + } +} + +func TestValidateDefinitionRejectsIncompatibleInputType(t *testing.T) { + def := mappedReplyDefinition() + def.Nodes[1].Inputs["replyText"] = dsl.VariableSelector{NodeID: "start_1", Field: "conversationId"} + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if result.Valid { + t.Fatalf("expected incompatible input type to be invalid") + } + if !hasValidationMessage(result, "input type mismatch") { + t.Fatalf("expected type-mismatch error, got %#v", result.Errors) + } +} + +func TestValidateDefinitionAcceptsMappedKnowledgeFlow(t *testing.T) { + def := dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + 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"}, + }, + 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"}, + }, + } + + result := validator.ValidateDefinition(def, registry.DefaultRegistry()) + + if !result.Valid { + t.Fatalf("expected mapped knowledge flow to be valid, got %#v", result.Errors) + } +} + func minimalDefinition() dsl.Definition { return dsl.Definition{ SchemaVersion: 1, EntryNodeID: "start_1", Nodes: []dsl.Node{ {ID: "start_1", Type: "start"}, - {ID: "reply_1", Type: "send_reply", Config: json.RawMessage(`{"text":"hello"}`)}, + {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"}, }, Edges: []dsl.Edge{ @@ -119,6 +212,14 @@ func minimalDefinition() dsl.Definition { } } +func mappedReplyDefinition() dsl.Definition { + def := minimalDefinition() + def.Nodes[1].Inputs = map[string]dsl.VariableSelector{ + "replyText": {NodeID: "start_1", Field: "userMessage"}, + } + return def +} + 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.go b/internal/builders/ai_workflow_builder.go index 13ac177..75b45d7 100644 --- a/internal/builders/ai_workflow_builder.go +++ b/internal/builders/ai_workflow_builder.go @@ -78,6 +78,10 @@ func BuildAIWorkflowNodeSpecs(list []workflowregistry.NodeSpec) []response.AIWor RiskLevel: item.RiskLevel, Interruptible: item.Interruptible, RequiresConfirmationPredecessor: item.RequiresConfirmationPredecessor, + ConfigSchema: item.ConfigSchema, + InputSchema: item.InputSchema, + OutputSchema: item.OutputSchema, + DefaultInputs: item.DefaultInputs, }) } return ret diff --git a/internal/builders/ai_workflow_builder_test.go b/internal/builders/ai_workflow_builder_test.go new file mode 100644 index 0000000..c643591 --- /dev/null +++ b/internal/builders/ai_workflow_builder_test.go @@ -0,0 +1,40 @@ +package builders + +import ( + "testing" + + workflowregistry "agent-desk/internal/ai/workflow/registry" +) + +func TestBuildAIWorkflowNodeSpecsIncludesVariableContracts(t *testing.T) { + specs := BuildAIWorkflowNodeSpecs(workflowregistry.DefaultRegistry().List()) + + var startFound bool + var sendReplyFound bool + for _, spec := range specs { + switch spec.Type { + case workflowregistry.NodeTypeStart: + startFound = true + if !hasResponseVariable(spec.OutputSchema, "userMessage") { + t.Fatalf("expected start output userMessage, got %#v", spec.OutputSchema) + } + case workflowregistry.NodeTypeSendReply: + sendReplyFound = true + if !hasResponseVariable(spec.InputSchema, "replyText") { + t.Fatalf("expected send_reply input replyText, got %#v", spec.InputSchema) + } + } + } + if !startFound || !sendReplyFound { + t.Fatalf("expected start and send_reply specs in response") + } +} + +func hasResponseVariable(items []workflowregistry.VariableSpec, name string) bool { + for _, item := range items { + if item.Name == name { + return true + } + } + return false +} diff --git a/internal/pkg/dto/response/ai_workflow_response.go b/internal/pkg/dto/response/ai_workflow_response.go index a45ffe3..07302c3 100644 --- a/internal/pkg/dto/response/ai_workflow_response.go +++ b/internal/pkg/dto/response/ai_workflow_response.go @@ -42,10 +42,14 @@ type AIWorkflowValidationResponse struct { } type AIWorkflowNodeSpecResponse struct { - Type string `json:"type"` - Title string `json:"title"` - Description string `json:"description"` - RiskLevel workflowregistry.NodeRiskLevel `json:"riskLevel"` - Interruptible bool `json:"interruptible"` - RequiresConfirmationPredecessor bool `json:"requiresConfirmationPredecessor"` + Type string `json:"type"` + Title string `json:"title"` + Description string `json:"description"` + RiskLevel workflowregistry.NodeRiskLevel `json:"riskLevel"` + Interruptible bool `json:"interruptible"` + RequiresConfirmationPredecessor bool `json:"requiresConfirmationPredecessor"` + 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"` } diff --git a/internal/services/ai_agent_workflow_service_test.go b/internal/services/ai_agent_workflow_service_test.go index dd1d8fc..47a4c3d 100644 --- a/internal/services/ai_agent_workflow_service_test.go +++ b/internal/services/ai_agent_workflow_service_test.go @@ -50,6 +50,12 @@ func TestAIAgentServiceCreatesDefaultWorkflow(t *testing.T) { if stored.EntryNodeID == "" { t.Fatalf("expected default draft definition") } + if len(stored.Nodes) < 5 { + t.Fatalf("expected product-ready default workflow, got nodes: %#v", stored.Nodes) + } + if !workflowHasNodeType(stored, "knowledge_retrieve") || !workflowHasNodeType(stored, "llm_reply") || !workflowHasNodeType(stored, "send_reply") { + t.Fatalf("expected default workflow to include retrieve, llm reply, and send reply nodes: %#v", stored.Nodes) + } } func TestAIWorkflowServicePublishAgentWorkflowBindsAgentVersion(t *testing.T) { @@ -169,3 +175,12 @@ func aiAgentWorkflowTestOperator() *dto.AuthPrincipal { Nickname: "agent-workflow-tester", } } + +func workflowHasNodeType(def dsl.Definition, nodeType string) bool { + for _, node := range def.Nodes { + if node.Type == nodeType { + return true + } + } + return false +} diff --git a/internal/services/ai_workflow_service.go b/internal/services/ai_workflow_service.go index f4f7bee..da268a9 100644 --- a/internal/services/ai_workflow_service.go +++ b/internal/services/ai_workflow_service.go @@ -337,10 +337,25 @@ func defaultAgentWorkflowDefinition() dsl.Definition { SchemaVersion: 1, EntryNodeID: "start_1", Nodes: []dsl.Node{ - {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start", Position: dsl.Position{X: 0, Y: 80}}, - {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End", Position: dsl.Position{X: 360, Y: 80}}, + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "开始", Position: dsl.Position{X: 0, Y: 120}}, + {ID: "retrieve_1", Type: workflowregistry.NodeTypeKnowledgeRetrieve, Name: "知识检索", Position: dsl.Position{X: 260, Y: 120}, Inputs: map[string]dsl.VariableSelector{ + "query": {NodeID: "start_1", Field: "userMessage"}, + }}, + {ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "AI 回复", Position: dsl.Position{X: 520, Y: 120}, 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: 780, Y: 120}, Inputs: map[string]dsl.VariableSelector{ + "replyText": {NodeID: "reply_1", Field: "replyText"}, + }}, + {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "结束", Position: dsl.Position{X: 1040, Y: 120}}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_retrieve", Source: "start_1", Target: "retrieve_1"}, + {ID: "edge_retrieve_reply", Source: "retrieve_1", Target: "reply_1"}, + {ID: "edge_reply_send", Source: "reply_1", Target: "send_1"}, + {ID: "edge_send_end", Source: "send_1", Target: "end_1"}, }, - Edges: []dsl.Edge{{ID: "edge_start_end", Source: "start_1", Target: "end_1"}}, } } diff --git a/internal/services/ai_workflow_service_test.go b/internal/services/ai_workflow_service_test.go index 73f33d8..fad4a94 100644 --- a/internal/services/ai_workflow_service_test.go +++ b/internal/services/ai_workflow_service_test.go @@ -174,7 +174,9 @@ func validAIWorkflowDefinition() dsl.Definition { EntryNodeID: "start_1", Nodes: []dsl.Node{ {ID: "start_1", Type: "start"}, - {ID: "reply_1", Type: "send_reply", Config: json.RawMessage(`{"text":"hello"}`)}, + {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"}, }, Edges: []dsl.Edge{ diff --git a/web/app/dashboard/ai-agents/_components/config-workbench.tsx b/web/app/dashboard/ai-agents/_components/config-workbench.tsx index 580dd4a..07a9652 100644 --- a/web/app/dashboard/ai-agents/_components/config-workbench.tsx +++ b/web/app/dashboard/ai-agents/_components/config-workbench.tsx @@ -89,19 +89,55 @@ const emptyDefinition: AIWorkflowDefinition = { { id: "start_1", type: "start", - name: "Start", - position: { x: 0, y: 80 }, + name: "开始", + position: { x: 0, y: 120 }, config: {}, }, + { + id: "retrieve_1", + type: "knowledge_retrieve", + name: "知识检索", + position: { x: 260, y: 120 }, + config: {}, + inputs: { + query: { nodeId: "start_1", field: "userMessage" }, + }, + }, + { + id: "reply_1", + type: "llm_reply", + name: "AI 回复", + position: { x: 520, y: 120 }, + config: {}, + inputs: { + userMessage: { nodeId: "start_1", field: "userMessage" }, + knowledgeItems: { nodeId: "retrieve_1", field: "items" }, + }, + }, + { + id: "send_1", + type: "send_reply", + name: "发送回复", + position: { x: 780, y: 120 }, + config: {}, + inputs: { + replyText: { nodeId: "reply_1", field: "replyText" }, + }, + }, { id: "end_1", type: "end", - name: "End", - position: { x: 360, y: 80 }, + name: "结束", + position: { x: 1040, y: 120 }, config: {}, }, ], - edges: [{ id: "edge_start_end", source: "start_1", target: "end_1" }], + edges: [ + { id: "edge_start_retrieve", source: "start_1", target: "retrieve_1" }, + { id: "edge_retrieve_reply", source: "retrieve_1", target: "reply_1" }, + { id: "edge_reply_send", source: "reply_1", target: "send_1" }, + { id: "edge_send_end", source: "send_1", target: "end_1" }, + ], } function toText(value: string | number | undefined | null) { 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 7689c12..a7e676f 100644 --- a/web/app/dashboard/ai-workflows/_components/node-config-panel.tsx +++ b/web/app/dashboard/ai-workflows/_components/node-config-panel.tsx @@ -7,51 +7,85 @@ 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 { VariableSelector } from "./variable-selector" +import type { + WorkflowNodeSpec, + WorkflowVariableRef, + WorkflowVariableSelector, +} from "./workflow-utils" type WorkflowNodeData = Record & { nodeType?: string name?: string config?: Record + inputs?: Record } export function NodeConfigPanel({ node, + nodeSpec, + availableVariables, onChange, }: { node: Node | null + nodeSpec?: WorkflowNodeSpec + availableVariables: WorkflowVariableRef[] onChange: (nodeId: string, data: WorkflowNodeData) => void }) { if (!node) { return (
- Select a node to edit its properties. + 选择一个节点后，可以配置输入映射并查看输出变量。
) } - return + return ( + + ) } function NodeConfigForm({ node, + nodeSpec, + availableVariables, onChange, }: { node: Node + nodeSpec?: WorkflowNodeSpec + availableVariables: WorkflowVariableRef[] 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 inputSchema = nodeSpec?.inputSchema ?? [] + const outputSchema = nodeSpec?.outputSchema ?? [] + + const commitChange = (next: Partial) => { + onChange(node.id, { + ...node.data, + name: name.trim() || node.data.nodeType || node.id, + config: node.data.config ?? {}, + inputs, + ...next, + }) + } const handleApply = () => { try { const parsed = JSON.parse(configText || "{}") as Record setError("") - onChange(node.id, { - ...node.data, - name: name.trim() || node.data.nodeType || node.id, - config: parsed, - }) + commitChange({ config: parsed }) } catch { setError("Config must be valid JSON.") } @@ -64,24 +98,90 @@ function NodeConfigForm({
{node.id}
- + setName(event.target.value)} + onBlur={() => commitChange({ name: name.trim() || node.data.nodeType || node.id })} />
-
- -