Add AnalyzeConversation node execution and corresponding test case

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))
 }

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,