diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go index 7481ef5..7e6a492 100644 --- a/internal/ai/runtime/workflow/executor.go +++ b/internal/ai/runtime/workflow/executor.go @@ -9,6 +9,7 @@ import ( "strings" "agent-desk/internal/ai" + "agent-desk/internal/ai/runtime/graphs" "agent-desk/internal/ai/runtime/internal/impl/retrievers" "agent-desk/internal/ai/workflow/dsl" workflowregistry "agent-desk/internal/ai/workflow/registry" @@ -124,6 +125,8 @@ func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.No return e.executeAnswerabilityGate(state, node) case workflowregistry.NodeTypeCondition: state.setNodeVars(node.ID, map[string]any{"matched": true}) + case workflowregistry.NodeTypeAnalyzeConversation: + return e.executeAnalyzeConversation(ctx, state, node) case workflowregistry.NodeTypeLLMReply: return e.executeLLMReply(ctx, state, node) case workflowregistry.NodeTypeSendReply: @@ -143,6 +146,48 @@ func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.No return nil } +func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runState, node dsl.Node) error { + userMessage := strings.TrimSpace(toString(state.resolveInput(node, "userMessage"))) + input := graphs.AnalyzeConversationInput{ + ObservedIssue: userMessage, + } + if strings.TrimSpace(readStringConfig(node.Config, "goal")) != "" { + input.Goal = strings.TrimSpace(readStringConfig(node.Config, "goal")) + } + if readBoolConfig(node.Config, "needTicket") { + input.NeedTicket = true + } + if readBoolConfig(node.Config, "needHumanHandoff") { + input.NeedHumanHandoff = true + } + if readBoolConfig(node.Config, "needQualityCheck") { + input.NeedQualityCheck = true + } + if strings.TrimSpace(readStringConfig(node.Config, "additionalContext")) != "" { + input.AdditionalContext = strings.TrimSpace(readStringConfig(node.Config, "additionalContext")) + } + args, err := json.Marshal(input) + if err != nil { + return err + } + raw, err := graphs.NewAnalyzeConversationGraph(state.input.Conversation).Run(ctx, string(args)) + if err != nil { + return err + } + var result graphs.AnalyzeConversationResult + if err := json.Unmarshal([]byte(raw), &result); err != nil { + return err + } + nextAction := strings.TrimSpace(result.RecommendedNextAction) + state.setNodeVars(node.ID, map[string]any{ + "intent": strings.TrimSpace(result.UserIntent), + "riskLevel": strings.TrimSpace(result.RiskLevel), + "needTicket": nextAction == "prepare_ticket", + "needHumanHandoff": nextAction == "handoff_to_human", + }) + return nil +} + func (e *Executor) executeHandoffToHuman(state *runState, node dsl.Node) error { reason := strings.TrimSpace(toString(state.resolveInput(node, "reason"))) result, err := services.ConversationHumanDispatchService.HandoffByAIWithRequestID( @@ -335,6 +380,17 @@ func readStringConfig(raw json.RawMessage, key string) string { return toString(cfg[key]) } +func readBoolConfig(raw json.RawMessage, key string) bool { + if len(raw) == 0 { + return false + } + var cfg map[string]any + if err := json.Unmarshal(raw, &cfg); err != nil { + return false + } + return truthy(cfg[key]) +} + func compareString(left any, right any) int { return strings.Compare(toString(left), toString(right)) } diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go index 0de337e..8d6f1bd 100644 --- a/internal/ai/runtime/workflow/executor_test.go +++ b/internal/ai/runtime/workflow/executor_test.go @@ -92,6 +92,24 @@ func TestExecutorHandoffToHumanRunsRealDispatchAction(t *testing.T) { } } +func TestExecutorAnalyzeConversationOutputsBranchVariables(t *testing.T) { + db := setupWorkflowExecutorHandoffDB(t) + aiAgent := createWorkflowExecutorHandoffAIAgent(t, db, "1") + conversation := createWorkflowExecutorHandoffConversation(t, db, aiAgent.ID) + userMessage := createWorkflowExecutorCustomerMessage(t, db, conversation.ID, "你们重复扣费了,我要投诉并转人工") + + result, err := NewExecutor().Execute(context.Background(), Input{ + Definition: analyzeConversationWorkflowDefinition(), + Conversation: conversation, + UserMessage: userMessage, + AIAgent: aiAgent, + }) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + assertPath(t, result.NodePath, []string{"start_1", "analyze_1", "handoff_end"}) +} + func conditionalReplyDefinition() dsl.Definition { return dsl.Definition{ SchemaVersion: 1, @@ -130,6 +148,34 @@ func conditionalReplyDefinition() dsl.Definition { } } +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: "handoff_end", Type: workflowregistry.NodeTypeEnd, Name: "Handoff"}, + {ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_analyze", Source: "start_1", Target: "analyze_1"}, + { + ID: "edge_analyze_handoff", + Source: "analyze_1", + Target: "handoff_end", + Condition: &dsl.Condition{ + Left: &dsl.VariableSelector{NodeID: "analyze_1", Field: "needHumanHandoff"}, + Operator: "is_true", + }, + }, + {ID: "edge_analyze_default", Source: "analyze_1", Target: "default_end"}, + }, + } +} + func handoffWorkflowDefinition() dsl.Definition { return dsl.Definition{ SchemaVersion: 1,