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.
This commit is contained in:
mlogclub
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user