From 3cdd31786af9636c6f16d688fbccb13f38c6c74c Mon Sep 17 00:00:00 2001 From: mlogclub Date: Tue, 14 Apr 2026 19:05:05 +0800 Subject: [PATCH] feat: enhance event consumption logic and add success reply handling --- .../ai/runtime/executor/event_consumer.go | 8 ++++ .../runtime/executor/event_consumer_test.go | 41 +++++++++++++++++++ internal/ai/runtime/graphs/handoff_graph.go | 6 ++- .../ai/runtime/graphs/handoff_graph_test.go | 10 +++++ internal/services/message_service.go | 11 ++++- internal/services/message_service_test.go | 29 +++++++++++++ 6 files changed, 103 insertions(+), 2 deletions(-) create mode 100644 internal/ai/runtime/executor/event_consumer_test.go create mode 100644 internal/services/message_service_test.go diff --git a/internal/ai/runtime/executor/event_consumer.go b/internal/ai/runtime/executor/event_consumer.go index f130507..41280a1 100644 --- a/internal/ai/runtime/executor/event_consumer.go +++ b/internal/ai/runtime/executor/event_consumer.go @@ -4,6 +4,8 @@ import ( "strings" "cs-agent/internal/ai/runtime/internal/impl/callbacks" + "cs-agent/internal/pkg/enums" + "cs-agent/internal/pkg/toolx" "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/schema" @@ -56,6 +58,12 @@ func consumeAgentEvents(events *adk.AsyncIterator[*adk.AgentEvent], summary *Run toolCode = strings.TrimSpace(mappedCode) } summary.InvokedToolCodes = appendIfMissing(summary.InvokedToolCodes, toolCode) + if strings.TrimSpace(summary.ReplyText) == "" && toolx.ResolveToolSourceType(toolCode) == enums.ToolSourceTypeGraph { + toolReplyText := strings.TrimSpace(messageOutput.Message.Content) + if toolReplyText != "" { + summary.ReplyText = toolReplyText + } + } } } if summary.Status == "started" { diff --git a/internal/ai/runtime/executor/event_consumer_test.go b/internal/ai/runtime/executor/event_consumer_test.go new file mode 100644 index 0000000..2e2de1b --- /dev/null +++ b/internal/ai/runtime/executor/event_consumer_test.go @@ -0,0 +1,41 @@ +package executor + +import ( + "testing" + + "cs-agent/internal/pkg/toolx" + + "github.com/cloudwego/eino/adk" + "github.com/cloudwego/eino/schema" +) + +func TestConsumeAgentEventsUsesGraphToolTextAsReplyFallback(t *testing.T) { + summary := &RunResult{ + Status: "started", + InvokedToolCodes: make([]string, 0), + } + events, gen := adk.NewAsyncIteratorPair[*adk.AgentEvent]() + gen.Send(&adk.AgentEvent{ + Output: &adk.AgentOutput{ + MessageOutput: &adk.MessageVariant{ + Role: schema.Tool, + ToolName: toolx.GraphHandoffConversation.Name, + Message: &schema.Message{ + Content: "已为你转接人工客服，请稍候。，请稍候。", + }, + }, + }, + }) + gen.Close() + + consumeAgentEvents(events, summary, nil, map[string]string{ + toolx.GraphHandoffConversation.Name: toolx.GraphHandoffConversation.Code, + }) + + if summary.ReplyText != "已为你转接人工客服，请稍候。，请稍候。" { + t.Fatalf("unexpected reply text: %q", summary.ReplyText) + } + if summary.Status != "completed" { + t.Fatalf("unexpected summary status: %q", summary.Status) + } +} diff --git a/internal/ai/runtime/graphs/handoff_graph.go b/internal/ai/runtime/graphs/handoff_graph.go index 7fd401a..8f11acd 100644 --- a/internal/ai/runtime/graphs/handoff_graph.go +++ b/internal/ai/runtime/graphs/handoff_graph.go @@ -82,7 +82,7 @@ func (g *HandoffGraph) Run(ctx context.Context, argumentsInJSON string) (string, if err := services.ConversationService.HandoffByAI(g.conversation.ID, g.aiAgent, state.Reason); err != nil { return "", err } - return "已为你转接人工客服，请稍候。", nil + return g.buildSuccessReply(), nil case ConfirmationDecisionCancel: return CancelHandoffReply, nil default: @@ -112,6 +112,10 @@ func (g *HandoffGraph) buildConfirmationPrompt(reason string) string { return fmt.Sprintf("我准备为你转接人工客服。\n原因：%s\n请直接回复“确认”或“取消”。", strings.TrimSpace(reason)) } +func (g *HandoffGraph) buildSuccessReply() string { + return "已为你转接人工客服，请稍候。，请稍候。" +} + func parseHandoffDecision(value string) ConfirmationDecision { return ParseConfirmationDecision(value) } diff --git a/internal/ai/runtime/graphs/handoff_graph_test.go b/internal/ai/runtime/graphs/handoff_graph_test.go index b6f469d..25bf8ee 100644 --- a/internal/ai/runtime/graphs/handoff_graph_test.go +++ b/internal/ai/runtime/graphs/handoff_graph_test.go @@ -29,3 +29,13 @@ func TestHandoffGraphBuildReasonFallback(t *testing.T) { t.Fatalf("unexpected fallback reason: %q", reason) } } + +func TestHandoffGraphBuildSuccessReply(t *testing.T) { + graph := NewHandoffGraph(&models.Conversation{ID: 1}, &models.AIAgent{Name: "AI"}) + + got := graph.buildSuccessReply() + want := "已为你转接人工客服，请稍候。，请稍候。" + if got != want { + t.Fatalf("unexpected success reply: %q", got) + } +} diff --git a/internal/services/message_service.go b/internal/services/message_service.go index 9fb422f..bb226fa 100644 --- a/internal/services/message_service.go +++ b/internal/services/message_service.go @@ -514,7 +514,7 @@ func (s *messageService) ValidateConversationSender(conversationID int64, sender if operator == nil { return nil, errorsx.Unauthorized("未登录或登录已过期") } - if conversation.Status != enums.IMConversationStatusAIServing { + if conversation.Status != enums.IMConversationStatusAIServing && !allowAIMessageOnPendingHandoff(conversation) { return nil, errorsx.Forbidden("当前会话不处于 AI 接待状态") } if conversation.CurrentAssigneeID != 0 { @@ -530,6 +530,15 @@ func (s *messageService) ValidateConversationSender(conversationID int64, sender return conversation, nil } +func allowAIMessageOnPendingHandoff(conversation *models.Conversation) bool { + if conversation == nil { + return false + } + return conversation.Status == enums.IMConversationStatusPending && + conversation.HandoffAt != nil && + conversation.CurrentAssigneeID == 0 +} + func suffixFilenameForSummary(filename string) string { filename = strings.TrimSpace(filename) if filename == "" { diff --git a/internal/services/message_service_test.go b/internal/services/message_service_test.go new file mode 100644 index 0000000..d1f4e6b --- /dev/null +++ b/internal/services/message_service_test.go @@ -0,0 +1,29 @@ +package services + +import ( + "testing" + "time" + + "cs-agent/internal/models" + "cs-agent/internal/pkg/enums" +) + +func TestAllowAIMessageOnPendingHandoff(t *testing.T) { + conversation := &models.Conversation{ + Status: enums.IMConversationStatusPending, + CurrentAssigneeID: 0, + HandoffAt: ptrTime(time.Now()), + } + if !allowAIMessageOnPendingHandoff(conversation) { + t.Fatalf("expected pending handoff conversation to allow ai handoff notice") + } + + conversation.Status = enums.IMConversationStatusAIServing + if allowAIMessageOnPendingHandoff(conversation) { + t.Fatalf("expected ai serving conversation not to use pending handoff allowance") + } +} + +func ptrTime(v time.Time) *time.Time { + return &v +}