package runtime import ( "context" "fmt" "log/slog" "strconv" "strings" "time" "code.tczkiot.com/wlw/ai-agent/contract" applicationruntime "code.tczkiot.com/wlw/ai-agent/internal/ai/application/runtime" "code.tczkiot.com/wlw/ai-agent/internal/models" "code.tczkiot.com/wlw/ai-agent/internal/pkg/enums" "code.tczkiot.com/wlw/ai-agent/internal/pkg/tracex" svc "code.tczkiot.com/wlw/ai-agent/internal/services" "github.com/mlogclub/simple/sqls" ) const aiReplyFailedReply = "已收到您的消息,但本次智能处理没有完成。请稍后重试,或回复“人工客服”继续处理。" const businessIdentityMenuPrefix = "identity_menu" const aiReplyInvocationToolCode = "runtime/ai_reply" const aiReplyInvocationRecoveryGrace = 30 * time.Second func (s *aiReplyService) resolveReplyTimeout(aiAgent models.AIAgent) time.Duration { if aiAgent.ReplyTimeoutSeconds <= 0 { return time.Duration(defaultAIReplyAsyncTimeoutSeconds) * time.Second } if aiAgent.ReplyTimeoutSeconds > maxAIReplyAsyncTimeoutSeconds { return time.Duration(maxAIReplyAsyncTimeoutSeconds) * time.Second } return time.Duration(aiAgent.ReplyTimeoutSeconds) * time.Second } func (s *aiReplyService) TriggerReplyAsync(requestContext context.Context, conversation models.Conversation, message models.Message) { aiAgent := svc.AIAgentService.Get(conversation.AIAgentID) if aiAgent == nil || aiAgent.Status != enums.StatusOk { return } timeout := s.resolveReplyTimeout(*aiAgent) invocationKey := fmt.Sprintf("message:%d:revision:%d", message.ID, aiAgent.PublishedRevisionID) claim, err := svc.AgentToolInvocationService.ClaimRecoverable( conversation.ID, aiAgent.ID, aiReplyInvocationToolCode, invocationKey, time.Now().Add(-(timeout + aiReplyInvocationRecoveryGrace)), ) if err != nil { slog.Error("failed to claim ai reply run", "requestId", message.RequestID, "conversation_id", conversation.ID, "message_id", message.ID, "revision_id", aiAgent.PublishedRevisionID, "error", err) return } if claim == nil || claim.Item == nil || !claim.Acquired { return } if committedAIReply(conversation.ID, message.ID) != nil { if err := svc.AgentToolInvocationService.Complete(claim.Item, `{}`); err != nil { slog.Error("failed to reconcile recovered ai reply claim", "conversation_id", conversation.ID, "message_id", message.ID, "error", err) } return } proofContext := contract.BindCustomerAccessProofToMessage(requestContext, conversation.ID, message.ID, message.RequestID) proof, hasProof := contract.CustomerAccessProofFromContext(proofContext) go func() { startedAt := time.Now() ctx := tracex.ContextWithRequestID(context.Background(), message.RequestID) if hasProof { ctx = contract.WithCustomerAccessProof(ctx, proof) } ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() defer func() { if recovered := recover(); recovered != nil { err := fmt.Errorf("ai reply panic: %v", recovered) _ = svc.AgentToolInvocationService.FailRetryable(claim.Item, err) slog.Error("panic while triggering ai reply", "requestId", message.RequestID, "conversation_id", conversation.ID, "message_id", message.ID, "error", err) s.commitFailureReplyIfNeeded(conversation, message, *aiAgent, err) } }() var triggerErr error if s.triggerReply != nil { triggerErr = s.triggerReply(ctx, conversation, message, *aiAgent) } else { triggerErr = s.TriggerReply(ctx, conversation, message, *aiAgent) } if triggerErr != nil { _ = svc.AgentToolInvocationService.FailRetryable(claim.Item, triggerErr) slog.Error("failed to trigger ai reply", "requestId", message.RequestID, "message_id", message.ID, "timeout_ms", timeout.Milliseconds(), "elapsed_ms", time.Since(startedAt).Milliseconds(), "error", triggerErr) s.commitFailureReplyIfNeeded(conversation, message, *aiAgent, triggerErr) return } if err := svc.AgentToolInvocationService.Complete(claim.Item, `{}`); err != nil { slog.Error("failed to complete ai reply run claim", "requestId", message.RequestID, "conversation_id", conversation.ID, "message_id", message.ID, "error", err) } }() } func committedAIReply(conversationID, messageID int64) *models.Message { for _, prefix := range []string{"ai_reply", "identity_prompt", "ai_interrupt", "ai_interrupt_expired", "ai_resume", "ai_interrupt_resume"} { clientMsgID := fmt.Sprintf("%s_%d", prefix, messageID) if existing := svc.MessageService.FindOne(sqls.NewCnd().Eq("conversation_id", conversationID).Eq("client_msg_id", clientMsgID)); existing != nil { return existing } } return nil } func (s *aiReplyService) commitFailureReplyIfNeeded(conversation models.Conversation, message models.Message, aiAgent models.AIAgent, cause error) { clientMsgID := fmt.Sprintf("ai_error_%d", message.ID) if existing := svc.MessageService.FindOne( sqls.NewCnd().Eq("conversation_id", conversation.ID).Eq("client_msg_id", clientMsgID), ); existing != nil { return } if _, err := s.commit.CommitAIReply(replyCommitInput{ Conversation: conversation, Message: message, AIAgent: aiAgent, ReplyText: aiReplyFailureText(cause), ClientPrefix: "ai_error", }); err != nil { slog.Error("failed to commit ai error reply", "requestId", message.RequestID, "conversation_id", conversation.ID, "message_id", message.ID, "error", err) } } func aiReplyFailureText(cause error) string { if cause == nil { return aiReplyFailedReply } message := strings.ToLower(cause.Error()) switch { case strings.Contains(message, "ai 请求标识未设置"), strings.Contains(message, "invalid_ai_request_id"): return "系统内置 AI 请求失败:AI 请求标识无效。请联系管理员检查网关配置,或回复“人工客服”继续处理。" case strings.Contains(message, "insufficient_ai_balance"), strings.Contains(message, "ai 额度不足"): return "系统内置 AI 额度不足,请充值后重试,或回复“人工客服”继续处理。" case strings.Contains(message, "invalid_ai_key"), strings.Contains(message, "missing_ai_key"), strings.Contains(message, "ai key 格式无效"), strings.Contains(message, "ai 授权凭证无效"): return "系统内置 AI Key 无效或已撤销,请联系管理员检查客服设置。" case strings.Contains(message, "ai_gateway_not_configured"), strings.Contains(message, "system built-in llm model is not configured"), strings.Contains(message, "系统内置模型尚未配置"): return "系统内置 AI 模型尚未配置,请联系管理员完成配置。" case strings.Contains(message, "context deadline exceeded"), strings.Contains(message, "request timeout"), strings.Contains(message, "client.timeout"): return "系统内置 AI 请求超时,请稍后重试,或回复“人工客服”继续处理。" case strings.Contains(message, "internal_error"), strings.Contains(message, "网关内部异常"): if requestID := extractAIGatewayRequestID(cause.Error()); requestID != "" { return "系统内置 AI 网关内部异常,请稍后重试;排查编号:" + requestID + "。如仍失败,请联系管理员或回复“人工客服”。" } return "系统内置 AI 网关内部异常,请稍后重试;如仍失败,请联系管理员或回复“人工客服”。" case isAIUpstreamError(message): return aiUpstreamFailureText(message) default: return aiReplyFailedReply } } func extractAIGatewayRequestID(message string) string { lowerMessage := strings.ToLower(message) for _, marker := range []string{"request_id:", "request_id="} { start := strings.Index(lowerMessage, marker) if start < 0 { continue } value := strings.TrimSpace(message[start+len(marker):]) value = strings.TrimLeft(value, "(\"'") end := 0 for end < len(value) { char := value[end] if (char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || char == '-' || char == '_' { end++ continue } break } if end > 0 { return value[:end] } } return "" } func isAIUpstreamError(message string) bool { if strings.Contains(message, "ai_upstream_failed") || strings.Contains(message, "ai_request_failed") { return true } providerMentioned := strings.Contains(message, "qwen") || strings.Contains(message, "千问") || strings.Contains(message, "deepseek") || strings.Contains(message, "dashscope") if !providerMentioned { return false } for _, marker := range []string{ "status code", "bad request", "unauthorized", "forbidden", "too many requests", "returned 4", "returned 5", "返回 4", "返回 5", "invalidparameter", "invalid_parameter", "throttling", "arrearage", "accessdenied", "error", } { if strings.Contains(message, marker) { return true } } return false } func aiUpstreamFailureText(message string) string { provider := "上游模型" if strings.Contains(message, "qwen") || strings.Contains(message, "千问") || strings.Contains(message, "dashscope") { provider = "千问" } else if strings.Contains(message, "deepseek") { provider = "DeepSeek" } switch { case strings.Contains(message, "model not exist"), strings.Contains(message, "model_not_found"), strings.Contains(message, "invalid model"), strings.Contains(message, "model.accessdenied"), strings.Contains(message, "model access denied"), strings.Contains(message, "模型不存在"): return "系统内置 AI 调用失败:" + provider + "模型不存在或暂不可用,也可能尚未开通,请联系管理员检查模型名称和开通状态。" case strings.Contains(message, "authentication"), strings.Contains(message, "invalid api key"), strings.Contains(message, "invalid_api_key"), strings.Contains(message, "invalidapikey"), strings.Contains(message, "unauthorized"): return "系统内置 AI 调用失败:" + provider + " API Key 无效或无权限,请联系管理员检查官网模型配置。" case strings.Contains(message, "rate limit"), strings.Contains(message, "rate_limit"), strings.Contains(message, "too many requests"), strings.Contains(message, "throttling"): return "系统内置 AI 调用失败:" + provider + "请求过于频繁,请稍后重试。" case strings.Contains(message, "insufficient balance"), strings.Contains(message, "insufficient quota"), strings.Contains(message, "arrearage"), strings.Contains(message, "quota"): return "系统内置 AI 调用失败:" + provider + "账户额度不足或已欠费,请联系管理员处理。" case strings.Contains(message, "maximum context length"), strings.Contains(message, "context_length"), strings.Contains(message, "input length"), strings.Contains(message, "tokens exceed"), strings.Contains(message, "too many tokens"): return "系统内置 AI 调用失败:" + provider + "请求内容超出模型上下文长度,请缩短消息后重试。" case strings.Contains(message, "does not support tools"), strings.Contains(message, "tool calling is not supported"), strings.Contains(message, "function calling is not supported"), strings.Contains(message, "unsupported tool"), strings.Contains(message, "unsupported function"): return "系统内置 AI 调用失败:当前" + provider + "模型不支持客服工具调用,请联系管理员更换可用模型。" case strings.Contains(message, "data_inspection_failed"), strings.Contains(message, "content_filter"), strings.Contains(message, "inappropriate content"): return "系统内置 AI 调用失败:" + provider + "拒绝了本次内容,请调整表述后重试。" case strings.Contains(message, "invalidparameter"), strings.Contains(message, "invalid_parameter"), strings.Contains(message, "bad request"), strings.Contains(message, "返回 400"), strings.Contains(message, "returned 400"): return "系统内置 AI 调用失败:" + provider + "请求参数不兼容,请联系管理员检查模型与客服工具配置。" default: return "系统内置 AI 调用失败:" + provider + "服务返回错误,请联系管理员在 AI 回复记录中查看详细原因。" } } func (s *aiReplyService) TriggerReply(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent) (retErr error) { var summary *applicationruntime.RunResult replyCtx := aiReplyContext{ Conversation: conversation, Message: message, AIAgent: aiAgent, SummaryRef: &summary, } if err := ctx.Err(); err != nil { return err } if s.eligibility != nil && !s.eligibility.CanReply(conversation, message, aiAgent) { return nil } if !IsAIAgentRolloutEligible(conversation, aiAgent, svc.ChannelService.Get(conversation.ChannelID)) { return nil } identityResolution, identityErr := resolveGuestBusinessIdentity(ctx, conversation, message) if identityErr != nil || identityResolution.NeedsPrompt { _, err := s.commit.CommitAIReply(replyCommitInput{ Conversation: conversation, Message: message, AIAgent: aiAgent, ReplyText: guestBusinessIdentityPrompt(identityResolution, identityErr), ClientPrefix: "identity_prompt", }) return err } replyCtx.Conversation = identityResolution.Conversation if pendingInterrupt := svc.ConversationInterruptService.FindLatestPendingByConversationID(conversation.ID); pendingInterrupt != nil { replyCtx.PendingInterrupt = pendingInterrupt return s.resumePendingInterrupt(ctx, replyCtx) } if identityResolution.CandidateProvided && isBusinessIdentityOnlyMessage(message) { handled, err := s.sendBusinessIdentityMenu(ctx, replyCtx) if handled || err != nil { return err } } if selection, ok := businessIdentityMenuSelection(message); ok { latest := latestAIMessage(conversation.ID) if latest != nil && strings.HasPrefix(latest.ClientMsgID, businessIdentityMenuPrefix+"_") { matched, aiMessage, err := svc.CustomerQuickActionService.ExecuteSelectedReply( ctx, &replyCtx.Conversation, selection, message.RequestID, message.ID, ) if matched || err != nil { return s.finishQuickActionReply(ctx, replyCtx, matched, aiMessage, err) } } if actionCode, matched := legacyCardMenuActionCode(latest, selection); matched { actionMatched, aiMessage, err := svc.CustomerQuickActionService.ExecuteActionReply( ctx, &replyCtx.Conversation, actionCode, message.RequestID, message.ID, ) if actionMatched || err != nil { return s.finishQuickActionReply(ctx, replyCtx, actionMatched, aiMessage, err) } } } if matched, err := svc.CustomerQuickActionService.ExecuteMatchedReply( ctx, &replyCtx.Conversation, message.Content, message.RequestID, message.ID, ); matched || err != nil { return err } return s.executeReply(ctx, replyCtx) } func (s *aiReplyService) sendBusinessIdentityMenu(ctx context.Context, replyCtx aiReplyContext) (bool, error) { actions, err := svc.CustomerQuickActionService.ListForConversation(ctx, &replyCtx.Conversation) if err != nil || len(actions) == 0 { return false, err } objectLabel := "业务对象" switch replyCtx.Conversation.CustomerType { case "card": objectLabel = "卡号" case "device": objectLabel = "设备号" case "mall_user": objectLabel = "商城用户" } var builder strings.Builder builder.WriteString("已识别") builder.WriteString(objectLabel) if identifier := strings.TrimSpace(replyCtx.Conversation.CustomerExternalID); identifier != "" { builder.WriteString(":") builder.WriteString(identifier) } builder.WriteString("。\n\n请回复序号选择需要的服务:") for index, action := range actions { builder.WriteString(fmt.Sprintf("\n%d. %s", index+1, action.Title)) } builder.WriteString("\n\n也可以直接输入要咨询的问题。") _, err = svc.MessageService.SendAutomaticServiceMessageWithRequestID( replyCtx.Conversation.ID, fmt.Sprintf("%s_%d", businessIdentityMenuPrefix, replyCtx.Message.ID), builder.String(), replyCtx.Message.RequestID, ) return true, err } func isBusinessIdentityOnlyMessage(message models.Message) bool { content := businessIdentityMessageContent(message) candidates, _ := businessIdentityCandidates(content) for _, candidate := range candidates { content = strings.ReplaceAll(content, candidate, "") } for _, marker := range []string{"卡号", "卡板", "设备号", "设备", "iccid", "imei"} { content = strings.ReplaceAll(strings.ToLower(content), marker, "") } content = strings.Map(func(r rune) rune { if r == ' ' || r == ' ' || r == '\n' || r == '\r' || r == '\t' || r == ' ' { return -1 } switch r { case ':', ':', ',', ',', '。', '.', ';', ';', '-', '_': return -1 default: return r } }, content) return content == "" } func businessIdentityMenuSelection(message models.Message) (int, bool) { content := strings.TrimSpace(businessIdentityMessageContent(message)) selection, err := strconv.Atoi(content) return selection, err == nil && selection > 0 } func latestAIMessage(conversationID int64) *models.Message { return svc.MessageService.FindOne(sqls.NewCnd(). Eq("conversation_id", conversationID). Eq("sender_type", enums.IMSenderTypeAI). Desc("id")) } func legacyCardMenuActionCode(message *models.Message, selection int) (string, bool) { if message == nil || selection <= 0 { return "", false } content := businessIdentityMessageContent(*message) if strings.Contains(content, "卡片提示停机") && strings.Contains(content, "无法上网") && strings.Contains(content, "无信号") && strings.Contains(content, "已充值但未恢复") { if selection >= 1 && selection <= 4 { return "card/network_diagnosis", true } return "", false } if strings.Contains(content, "卡片状态") && strings.Contains(content, "网络连接") && strings.Contains(content, "套餐") && strings.Contains(content, "其他问题") { actions := map[int]string{ 1: "card/status", 2: "card/network_diagnosis", 3: "card/package", } code, ok := actions[selection] return code, ok } return "", false } func (s *aiReplyService) finishQuickActionReply( ctx context.Context, replyCtx aiReplyContext, matched bool, aiMessage string, err error, ) error { if err != nil || !matched { return err } if strings.TrimSpace(aiMessage) == "" { return nil } replyCtx.Message.Content = aiMessage replyCtx.Message.MessageType = enums.IMMessageTypeText return s.executeReply(ctx, replyCtx) } func (s *aiReplyService) resumePendingInterrupt(ctx context.Context, replyCtx aiReplyContext) error { return s.interrupts.ResumePendingInterrupt(ctx, s, replyCtx) } func (s *aiReplyService) executeReply(ctx context.Context, replyCtx aiReplyContext) error { summary, err := s.executor.Run(ctx, runtimeReplyRunInput{ Conversation: replyCtx.Conversation, Message: replyCtx.Message, AIAgent: replyCtx.AIAgent, }) replyCtx.setSummary(summary) if err != nil { return err } if summary != nil && summary.Interrupted { return s.interrupts.HandleInterruptedSummary(s, replyCtx, summary) } if summary != nil && summary.HandoffRequested { if _, err := svc.ConversationHumanDispatchService.HandoffByAIWithRequestID( replyCtx.Conversation.ID, replyCtx.AIAgent, summary.HandoffReason, replyCtx.Message.RequestID, ); err != nil { return err } return nil } if summary != nil && strings.TrimSpace(summary.ReplyText) != "" { _, err := s.commit.CommitAIReply(replyCommitInput{ Conversation: replyCtx.Conversation, Message: replyCtx.Message, AIAgent: replyCtx.AIAgent, ReplyText: summary.ReplyText, ClientPrefix: "ai_reply", }) if err != nil { return err } } return nil }