Implement handoff to human functionality in workflow executor and update registry output schema

2026-06-23 21:48:20 +08:00
parent ec9e4a852a
commit 8992f263aa
3 changed files with 247 additions and 10 deletions
@@ -14,6 +14,7 @@ import (
 	workflowregistry "agent-desk/internal/ai/workflow/registry"
 	"agent-desk/internal/models"
 	"agent-desk/internal/pkg/utils"
+	"agent-desk/internal/services"
 )

 const maxWorkflowSteps = 128
@@ -133,16 +134,7 @@ func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.No
 			"replyMessageId": int64(0),
 		})
 	case workflowregistry.NodeTypeHandoffToHuman:
-		reason := strings.TrimSpace(toString(state.resolveInput(node, "reason")))
-		replyText := strings.TrimSpace(readStringConfig(node.Config, "replyText"))
-		if replyText == "" {
-			replyText = "已为你转接人工客服，请稍候。"
-		}
-		state.result.ReplyText = replyText
-		state.setNodeVars(node.ID, map[string]any{
-			"handoffId": int64(0),
-			"reason":    reason,
-		})
+		return e.executeHandoffToHuman(state, node)
 	case workflowregistry.NodeTypeEnd:
 		state.setNodeVars(node.ID, map[string]any{"status": "completed"})
 	default:
@@ -151,6 +143,35 @@ func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.No
 	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(
+		state.input.Conversation.ID,
+		state.input.AIAgent,
+		reason,
+		strings.TrimSpace(state.input.UserMessage.RequestID),
+	)
+	if err != nil {
+		return err
+	}
+	output := map[string]any{
+		"handoffId":  int64(0),
+		"reason":     reason,
+		"decision":   "",
+		"teamId":     int64(0),
+		"assigneeId": int64(0),
+		"message":    "",
+	}
+	if result != nil {
+		output["decision"] = string(result.Decision)
+		output["teamId"] = result.TeamID
+		output["assigneeId"] = result.AssigneeID
+		output["message"] = strings.TrimSpace(result.Message)
+	}
+	state.setNodeVars(node.ID, output)
+	return nil
+}
+
 func (e *Executor) executeKnowledgeRetrieve(ctx context.Context, state *runState, node dsl.Node) error {
 	query := strings.TrimSpace(toString(state.resolveInput(node, "query")))
 	retriever := retrievers.NewKnowledgeRetriever(state.input.AIAgent)
@@ -2,11 +2,20 @@ package workflow

 import (
 	"context"
+	"strings"
 	"testing"
+	"time"

 	"agent-desk/internal/ai/workflow/dsl"
 	workflowregistry "agent-desk/internal/ai/workflow/registry"
 	"agent-desk/internal/models"
+	"agent-desk/internal/pkg/enums"
+	"agent-desk/internal/services"
+
+	"github.com/glebarez/sqlite"
+	"github.com/mlogclub/simple/sqls"
+	"gorm.io/gorm"
+	"gorm.io/gorm/schema"
 )

 func TestExecutorRoutesByConditionEdge(t *testing.T) {
@@ -43,6 +52,46 @@ func TestExecutorUsesDefaultEdgeWhenConditionDoesNotMatch(t *testing.T) {
 	assertPath(t, result.NodePath, []string{"start_1", "condition_1", "normal_reply", "send_normal", "end_1"})
 }

+func TestExecutorHandoffToHumanRunsRealDispatchAction(t *testing.T) {
+	db := setupWorkflowExecutorHandoffDB(t)
+	aiAgent := createWorkflowExecutorHandoffAIAgent(t, db, "1")
+	createWorkflowExecutorHandoffTeam(t, db, 1, "售后支持组")
+	createWorkflowExecutorHandoffActiveSchedule(t, db, 1)
+	createWorkflowExecutorHandoffAgentProfile(t, db, 101, 1)
+	conversation := createWorkflowExecutorHandoffConversation(t, db, aiAgent.ID)
+	userMessage := createWorkflowExecutorCustomerMessage(t, db, conversation.ID, "需要人工处理")
+
+	result, err := NewExecutor().Execute(context.Background(), Input{
+		Definition:   handoffWorkflowDefinition(),
+		Conversation: conversation,
+		UserMessage:  userMessage,
+		AIAgent:      aiAgent,
+	})
+	if err != nil {
+		t.Fatalf("execute workflow: %v", err)
+	}
+	if strings.TrimSpace(result.ReplyText) != "" {
+		t.Fatalf("expected workflow handoff node to avoid duplicate reply text, got %q", result.ReplyText)
+	}
+	assertPath(t, result.NodePath, []string{"start_1", "handoff_1", "assigned_end"})
+
+	current := services.ConversationService.Get(conversation.ID)
+	if current.Status != enums.IMConversationStatusActive {
+		t.Fatalf("expected active conversation, got status=%d", current.Status)
+	}
+	if current.CurrentAssigneeID != 101 || current.CurrentTeamID != 1 {
+		t.Fatalf("unexpected assignment: assignee=%d team=%d", current.CurrentAssigneeID, current.CurrentTeamID)
+	}
+	if current.HandoffAt == nil || current.HandoffReason != "需要人工处理" {
+		t.Fatalf("expected handoff metadata, got at=%v reason=%q", current.HandoffAt, current.HandoffReason)
+	}
+
+	notice := services.MessageService.FindOne(sqls.NewCnd().Eq("conversation_id", conversation.ID).Eq("sender_type", enums.IMSenderTypeAI).Desc("id"))
+	if notice == nil || strings.TrimSpace(notice.Content) == "" {
+		t.Fatalf("expected handoff service to send ai notice, got %+v", notice)
+	}
+}
+
 func conditionalReplyDefinition() dsl.Definition {
 	return dsl.Definition{
 		SchemaVersion: 1,
@@ -81,6 +130,168 @@ func conditionalReplyDefinition() dsl.Definition {
 	}
 }

+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: "assigned_end", Type: workflowregistry.NodeTypeEnd, Name: "Assigned"},
+			{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
+		},
+		Edges: []dsl.Edge{
+			{ID: "edge_start_handoff", Source: "start_1", Target: "handoff_1"},
+			{
+				ID:     "edge_handoff_assigned",
+				Source: "handoff_1",
+				Target: "assigned_end",
+				Condition: &dsl.Condition{
+					Left:     &dsl.VariableSelector{NodeID: "handoff_1", Field: "decision"},
+					Operator: "eq",
+					Right:    string(services.HandoffDecisionAssigned),
+				},
+			},
+			{ID: "edge_handoff_default", Source: "handoff_1", Target: "default_end"},
+		},
+	}
+}
+
+func setupWorkflowExecutorHandoffDB(t *testing.T) *gorm.DB {
+	t.Helper()
+	dbName := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name())
+	db, err := gorm.Open(sqlite.Open("file:"+dbName+"?mode=memory&cache=shared"), &gorm.Config{
+		NamingStrategy: schema.NamingStrategy{
+			TablePrefix:   "t_",
+			SingularTable: true,
+		},
+	})
+	if err != nil {
+		t.Fatalf("open sqlite error = %v", err)
+	}
+	t.Cleanup(func() {
+		sqlDB, err := db.DB()
+		if err == nil {
+			_ = sqlDB.Close()
+		}
+	})
+	if err := db.AutoMigrate(
+		&models.User{},
+		&models.AIAgent{},
+		&models.AgentTeam{},
+		&models.AgentTeamSchedule{},
+		&models.AgentProfile{},
+		&models.Conversation{},
+		&models.ConversationAssignment{},
+		&models.ConversationEventLog{},
+		&models.ConversationReadState{},
+		&models.Message{},
+		&models.ChannelMessageOutbox{},
+	); err != nil {
+		t.Fatalf("auto migrate error = %v", err)
+	}
+	sqls.SetDB(db)
+	return db
+}
+
+func createWorkflowExecutorHandoffAIAgent(t *testing.T, db *gorm.DB, teamIDs string) models.AIAgent {
+	t.Helper()
+	item := models.AIAgent{
+		Name:        "测试AI",
+		ServiceMode: enums.IMConversationServiceModeAIFirst,
+		TeamIDs:     teamIDs,
+		Status:      enums.StatusOk,
+	}
+	if err := db.Create(&item).Error; err != nil {
+		t.Fatalf("create ai agent error = %v", err)
+	}
+	return item
+}
+
+func createWorkflowExecutorHandoffTeam(t *testing.T, db *gorm.DB, id int64, name string) {
+	t.Helper()
+	if err := db.Create(&models.AgentTeam{ID: id, Name: name, Status: enums.StatusOk}).Error; err != nil {
+		t.Fatalf("create team error = %v", err)
+	}
+}
+
+func createWorkflowExecutorHandoffActiveSchedule(t *testing.T, db *gorm.DB, teamID int64) {
+	t.Helper()
+	now := time.Now()
+	if err := db.Create(&models.AgentTeamSchedule{
+		TeamID:  teamID,
+		StartAt: now.Add(-time.Hour),
+		EndAt:   now.Add(time.Hour),
+		Status:  enums.StatusOk,
+	}).Error; err != nil {
+		t.Fatalf("create schedule error = %v", err)
+	}
+}
+
+func createWorkflowExecutorHandoffAgentProfile(t *testing.T, db *gorm.DB, userID int64, teamID int64) {
+	t.Helper()
+	if err := db.Create(&models.User{
+		ID:       userID,
+		Username: "agent",
+		Nickname: "客服",
+		Status:   enums.StatusOk,
+	}).Error; err != nil {
+		t.Fatalf("create user error = %v", err)
+	}
+	if err := db.Create(&models.AgentProfile{
+		UserID:             userID,
+		TeamID:             teamID,
+		AgentCode:          "A001",
+		DisplayName:        "客服",
+		ServiceStatus:      enums.ServiceStatusIdle,
+		MaxConcurrentCount: 3,
+		AutoAssignEnabled:  true,
+		Status:             enums.StatusOk,
+	}).Error; err != nil {
+		t.Fatalf("create profile error = %v", err)
+	}
+}
+
+func createWorkflowExecutorHandoffConversation(t *testing.T, db *gorm.DB, aiAgentID int64) models.Conversation {
+	t.Helper()
+	now := time.Now()
+	item := models.Conversation{
+		AIAgentID:     aiAgentID,
+		ChannelID:     1,
+		CustomerID:    1,
+		CustomerName:  "测试访客",
+		Status:        enums.IMConversationStatusAIServing,
+		ServiceMode:   enums.IMConversationServiceModeAIFirst,
+		LastMessageAt: now,
+		LastActiveAt:  now,
+	}
+	if err := db.Create(&item).Error; err != nil {
+		t.Fatalf("create conversation error = %v", err)
+	}
+	return item
+}
+
+func createWorkflowExecutorCustomerMessage(t *testing.T, db *gorm.DB, conversationID int64, content string) models.Message {
+	t.Helper()
+	now := time.Now()
+	item := models.Message{
+		ConversationID: conversationID,
+		ClientMsgID:    "customer-message",
+		SenderType:     enums.IMSenderTypeCustomer,
+		MessageType:    enums.IMMessageTypeText,
+		Content:        content,
+		SeqNo:          1,
+		SendStatus:     enums.IMMessageStatusSent,
+		SentAt:         &now,
+	}
+	if err := db.Create(&item).Error; err != nil {
+		t.Fatalf("create message error = %v", err)
+	}
+	return item
+}
+
 func assertPath(t *testing.T, got []string, want []string) {
 	t.Helper()
 	if len(got) != len(want) {
@@ -151,6 +151,11 @@ func DefaultRegistry() *Registry {
 			},
 			OutputSchema: []VariableSpec{
 				output("handoffId", VariableTypeInteger, "Handoff operation ID."),
+				output("reason", VariableTypeString, "Handoff reason."),
+				output("decision", VariableTypeString, "Handoff dispatch decision."),
+				output("teamId", VariableTypeInteger, "Assigned or pending team ID."),
+				output("assigneeId", VariableTypeInteger, "Assigned agent user ID."),
+				output("message", VariableTypeString, "Customer-visible handoff notice."),
 			},
 		},
 		NodeSpec{