diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go index 97dbd8c..7481ef5 100644 --- a/internal/ai/runtime/workflow/executor.go +++ b/internal/ai/runtime/workflow/executor.go @@ -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) diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go index 596c1e7..0de337e 100644 --- a/internal/ai/runtime/workflow/executor_test.go +++ b/internal/ai/runtime/workflow/executor_test.go @@ -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) { diff --git a/internal/ai/workflow/registry/registry.go b/internal/ai/workflow/registry/registry.go index fa17763..102b9e8 100644 --- a/internal/ai/workflow/registry/registry.go +++ b/internal/ai/workflow/registry/registry.go @@ -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{