package workflow import ( "context" "encoding/json" "fmt" "html" "reflect" "regexp" "strconv" "strings" "time" "code.tczkiot.com/wlw/ai-agent/internal/ai" "code.tczkiot.com/wlw/ai-agent/internal/ai/runtime/graphs" "code.tczkiot.com/wlw/ai-agent/internal/ai/runtime/readtools" aitooling "code.tczkiot.com/wlw/ai-agent/internal/ai/tooling" "code.tczkiot.com/wlw/ai-agent/internal/ai/workflow/dsl" workflowregistry "code.tczkiot.com/wlw/ai-agent/internal/ai/workflow/registry" "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/toolx" "code.tczkiot.com/wlw/ai-agent/internal/services" ) const maxWorkflowSteps = 128 var workflowHTMLTagPattern = regexp.MustCompile(`<[^>]+>`) var workflowTemplateVariablePattern = regexp.MustCompile(`\{\{\s*([a-zA-Z0-9_]+)\s*\}\}`) type Input struct { Definition dsl.Definition Conversation models.Conversation UserMessage models.Message AIAgent models.AIAgent AIConfig models.AIConfig Debug bool } type Result struct { Status string ReplyText string NodePath []string NodeTraces []NodeTrace PromptTokens int CompletionTokens int RetrieverCount int TraceData string CheckPointID string CheckPointData string Interrupted bool Interrupts []InterruptSummary } type NodeTrace struct { NodeID string NodeType string Status string InputPreview string OutputPreview string ErrorMessage string DurationMS int } type InterruptSummary struct { Type string ID string InfoPreview string } type Executor struct{} func NewExecutor() *Executor { return &Executor{} } type runState struct { input Input nodesByID map[string]dsl.Node outgoing map[string][]dsl.Edge vars map[string]map[string]any branchDecisions map[string]branchDecision result Result } type workflowCheckPoint struct { Definition dsl.Definition `json:"definition"` ConfirmNodeID string `json:"confirmNodeId"` Vars map[string]map[string]any `json:"vars"` } type branchDecision struct { SelectedEdgeID string `json:"selectedEdgeId,omitempty"` SelectedBranchID string `json:"selectedBranchId,omitempty"` SelectedBranchName string `json:"selectedBranchName,omitempty"` SelectedTargetNodeID string `json:"selectedTargetNodeId,omitempty"` Reason string `json:"reason"` Evaluations []conditionEvaluation `json:"evaluations,omitempty"` } type conditionEvaluation struct { EdgeID string `json:"edgeId"` BranchID string `json:"branchId,omitempty"` BranchName string `json:"branchName,omitempty"` TargetNodeID string `json:"targetNodeId"` SourceNodeID string `json:"sourceNodeId,omitempty"` SourceField string `json:"sourceField,omitempty"` Operator string `json:"operator,omitempty"` LeftValue any `json:"leftValue,omitempty"` RightValue any `json:"rightValue,omitempty"` Matched bool `json:"matched"` } func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) { state := newRunState(input) currentID := state.startNodeID() if currentID == "" { return nil, fmt.Errorf("workflow entry node is required") } return e.executeFrom(ctx, state, currentID) } func (e *Executor) Resume(ctx context.Context, input Input, checkPointData string, resumeText string) (*Result, error) { var checkpoint workflowCheckPoint if err := json.Unmarshal([]byte(strings.TrimSpace(checkPointData)), &checkpoint); err != nil { return nil, fmt.Errorf("invalid workflow checkpoint: %w", err) } if len(checkpoint.Definition.Nodes) > 0 { input.Definition = checkpoint.Definition } state := newRunState(input) state.vars = checkpoint.Vars if state.vars == nil { state.vars = make(map[string]map[string]any) } confirmNodeID := strings.TrimSpace(checkpoint.ConfirmNodeID) if confirmNodeID == "" { return nil, fmt.Errorf("workflow checkpoint confirm node is required") } decision := graphs.ParseConfirmationDecision(resumeText) if decision == "" { node, ok := state.nodesByID[confirmNodeID] if !ok { return nil, fmt.Errorf("workflow node does not exist: %s", confirmNodeID) } if err := e.executeHumanConfirm(state, node); err != nil { return nil, err } state.result.Status = "interrupted" return &state.result, nil } state.setNodeVars(confirmNodeID, map[string]any{ "confirmed": decision == graphs.ConfirmationDecisionConfirm, "responseText": strings.TrimSpace(resumeText), }) nextID, ok, err := state.nextNodeID(confirmNodeID) if err != nil { return nil, err } if !ok { state.result.Status = "completed" return &state.result, nil } return e.executeFrom(ctx, state, nextID) } func (e *Executor) executeFrom(ctx context.Context, state *runState, currentID string) (*Result, error) { for step := 0; step < maxWorkflowSteps; step++ { node, ok := state.nodesByID[currentID] if !ok { err := fmt.Errorf("workflow node does not exist: %s", currentID) state.result.Status = "error" return &state.result, err } state.result.NodePath = append(state.result.NodePath, node.ID) trace := NodeTrace{ NodeID: node.ID, NodeType: node.Type, Status: "running", InputPreview: workflowPreviewJSON(state.nodeInputPreview(node)), } startedAt := time.Now() if err := e.executeNode(ctx, state, node); err != nil { trace.Status = "failed" trace.ErrorMessage = err.Error() trace.DurationMS = int(time.Since(startedAt).Milliseconds()) state.result.NodeTraces = append(state.result.NodeTraces, trace) state.result.Status = "error" return &state.result, err } trace.DurationMS = int(time.Since(startedAt).Milliseconds()) if state.result.Interrupted { trace.OutputPreview = workflowPreviewJSON(state.nodeOutputPreview(node.ID)) trace.Status = "interrupted" state.result.NodeTraces = append(state.result.NodeTraces, trace) state.result.Status = "interrupted" return &state.result, nil } if node.Type == workflowregistry.NodeTypeEnd { trace.OutputPreview = workflowPreviewJSON(state.nodeOutputPreview(node.ID)) trace.Status = "completed" state.result.NodeTraces = append(state.result.NodeTraces, trace) state.result.Status = "completed" return &state.result, nil } nextID, ok, err := state.nextNodeID(node.ID) if err != nil { trace.OutputPreview = workflowPreviewJSON(state.nodeOutputPreview(node.ID)) trace.Status = "failed" trace.ErrorMessage = err.Error() trace.DurationMS = int(time.Since(startedAt).Milliseconds()) state.result.NodeTraces = append(state.result.NodeTraces, trace) state.result.Status = "error" return &state.result, err } trace.OutputPreview = workflowPreviewJSON(state.nodeOutputPreview(node.ID)) trace.Status = "completed" state.result.NodeTraces = append(state.result.NodeTraces, trace) if !ok { state.result.Status = "completed" return &state.result, nil } currentID = nextID } err := fmt.Errorf("workflow exceeded max steps") state.result.Status = "error" return &state.result, err } func newRunState(input Input) *runState { state := &runState{ input: input, nodesByID: make(map[string]dsl.Node, len(input.Definition.Nodes)), outgoing: make(map[string][]dsl.Edge), vars: make(map[string]map[string]any), branchDecisions: make(map[string]branchDecision), result: Result{ Status: "started", NodePath: make([]string, 0), NodeTraces: make([]NodeTrace, 0), }, } for _, node := range input.Definition.Nodes { node.ID = strings.TrimSpace(node.ID) node.Type = strings.TrimSpace(node.Type) if node.ID != "" { state.nodesByID[node.ID] = node } } for _, edge := range input.Definition.Edges { state.outgoing[strings.TrimSpace(edge.SourceNodeID)] = append(state.outgoing[strings.TrimSpace(edge.SourceNodeID)], edge) } return state } func (s *runState) startNodeID() string { for _, node := range s.nodesByID { if strings.TrimSpace(node.Type) == workflowregistry.NodeTypeStart { return strings.TrimSpace(node.ID) } } return "" } func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.Node) error { switch node.Type { case workflowregistry.NodeTypeStart: userMessage := strings.TrimSpace(state.input.UserMessage.Content) state.setNodeVars(node.ID, map[string]any{ "conversationId": state.input.Conversation.ID, "messageId": state.input.UserMessage.ID, "aiAgentId": state.input.AIAgent.ID, "userMessage": userMessage, "query": userMessage, "conversationState": state.input.Conversation.Status, }) case workflowregistry.NodeTypeConversationUnderstanding: return e.executeConversationUnderstanding(state, node) case workflowregistry.NodeTypeReplyPolicy: return e.executeReplyPolicy(state, node) case workflowregistry.NodeTypeKnowledgeRetrieve: return e.executeKnowledgeRetrieve(ctx, state, node) case workflowregistry.NodeTypeAnswerabilityGate: 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.NodeTypePrepareTicketDraft: return e.executePrepareTicketDraft(ctx, state, node) case workflowregistry.NodeTypeHumanConfirm: return e.executeHumanConfirm(state, node) case workflowregistry.NodeTypeCreateTicket: return e.executeCreateTicket(state, node) case workflowregistry.NodeTypeLLMReply: return e.executeLLMReply(ctx, state, node) case workflowregistry.NodeTypeLLM: return e.executeOfficialLLM(ctx, state, node) case workflowregistry.NodeTypeSendReply: replyText := strings.TrimSpace(toString(state.resolveInput(node, "replyText"))) state.result.ReplyText = replyText state.setNodeVars(node.ID, map[string]any{ "sent": replyText != "", "replyMessageId": int64(0), }) case workflowregistry.NodeTypeHandoffToHuman: return e.executeHandoffToHuman(state, node) case workflowregistry.NodeTypeEnd: outputs := state.resolvedInputs(node) outputs["status"] = "completed" state.setNodeVars(node.ID, outputs) if state.result.ReplyText == "" { state.result.ReplyText = strings.TrimSpace(toString(outputs["result"])) } default: return fmt.Errorf("unsupported workflow node type: %s", node.Type) } return nil } func (e *Executor) executeOfficialLLM(ctx context.Context, state *runState, node dsl.Node) error { systemPrompt := strings.TrimSpace(toString(state.resolveInput(node, "systemPrompt"))) if systemPrompt == "" { systemPrompt = strings.TrimSpace(state.input.AIAgent.SystemPrompt) } userPrompt := strings.TrimSpace(toString(state.resolveInput(node, "prompt"))) if userPrompt == "" { userPrompt = strings.TrimSpace(state.input.UserMessage.Content) } result, err := ai.LLM.ChatWithConfig(ctx, state.input.AIConfig, systemPrompt, userPrompt) if err != nil { return err } state.result.PromptTokens += result.PromptTokens state.result.CompletionTokens += result.CompletionTokens state.result.ReplyText = strings.TrimSpace(result.Content) state.setNodeVars(node.ID, map[string]any{"result": result.Content}) return nil } func (e *Executor) executeConversationUnderstanding(state *runState, node dsl.Node) error { rawMessage := strings.TrimSpace(toString(state.resolveInput(node, "userMessage"))) if rawMessage == "" { rawMessage = state.input.UserMessage.Content } understanding := understandConversationMessage(rawMessage) state.setNodeVars(node.ID, map[string]any{ "normalizedMessage": understanding.NormalizedMessage, "messageIntent": understanding.MessageIntent, "answerScope": understanding.AnswerScope, "confidence": understanding.Confidence, "riskSignals": understanding.RiskSignals, "reason": understanding.Reason, }) return nil } func (e *Executor) executeReplyPolicy(state *runState, node dsl.Node) error { intent := strings.TrimSpace(toString(state.resolveInput(node, "messageIntent"))) scope := strings.TrimSpace(toString(state.resolveInput(node, "answerScope"))) userMessage := normalizeWorkflowUserMessage(toString(state.resolveInput(node, "userMessage"))) decision := decideWorkflowReplyPolicy(state.input.AIAgent, workflowReplyPolicyInput{ MessageIntent: intent, AnswerScope: scope, UserMessage: userMessage, Answerability: strings.TrimSpace(toString(state.resolveInput(node, "answerability"))), }) state.setNodeVars(node.ID, map[string]any{ "action": decision.Action, "replyText": decision.ReplyText, "reason": decision.Reason, "requiresFlow": decision.RequiresFlow, "targetFlow": decision.TargetFlow, "finalReplySource": decision.FinalReplySource, }) return nil } func (e *Executor) executeCreateTicket(state *runState, node dsl.Node) error { confirmed := truthy(state.resolveInput(node, "confirmed")) if !confirmed { state.setNodeVars(node.ID, map[string]any{ "ticketId": int64(0), "created": false, }) return nil } if state.input.Debug { state.setNodeVars(node.ID, map[string]any{ "ticketId": int64(0), "ticketNo": "", "created": false, "message": "调试运行不会创建工单。", "skipped": true, }) return nil } draft := asMap(state.resolveInput(node, "ticketDraft")) title := strings.TrimSpace(toString(draft["title"])) description := strings.TrimSpace(toString(draft["description"])) tagIDs := toInt64Slice(state.resolveInput(node, "tagIds")) assigneeID := toInt64(state.resolveInput(node, "assigneeId")) result, err := services.BusinessToolExecutor.Execute(context.Background(), services.BusinessToolInput{ Conversation: state.input.Conversation, AIAgent: state.input.AIAgent, ToolCode: toolx.GraphCreateTicketConfirm.Code, Arguments: map[string]any{ "title": title, "description": description, "tagIds": tagIDs, "assigneeId": assigneeID, }, IdempotencyKey: workflowToolIdempotencyKey(state, node), Confirmed: true, }) if err != nil { return err } var output struct { TicketID int64 `json:"ticketId"` TicketNo string `json:"ticketNo"` Created bool `json:"created"` } if err := json.Unmarshal([]byte(result.ResultData), &output); err != nil { return err } item := &models.Ticket{ID: output.TicketID, TicketNo: output.TicketNo} state.setNodeVars(node.ID, map[string]any{ "ticketId": item.ID, "ticketNo": item.TicketNo, "created": output.Created, "message": buildTicketCreatedMessage(item), }) return nil } func buildTicketCreatedMessage(item *models.Ticket) string { if item == nil { return "工单已创建。" } ticketNo := strings.TrimSpace(item.TicketNo) if ticketNo == "" { return fmt.Sprintf("工单已创建,工单 ID:%d。", item.ID) } return "工单已创建,工单号:" + ticketNo + "。" } type workflowConversationUnderstanding struct { NormalizedMessage string MessageIntent string AnswerScope string Confidence float64 RiskSignals []string Reason string } type workflowReplyPolicyInput struct { MessageIntent string AnswerScope string UserMessage string Answerability string } type workflowReplyPolicyDecision struct { Action string ReplyText string Reason string RequiresFlow bool TargetFlow string FinalReplySource string } func understandConversationMessage(rawMessage string) workflowConversationUnderstanding { message := normalizeWorkflowUserMessage(rawMessage) ret := workflowConversationUnderstanding{ NormalizedMessage: message, MessageIntent: "unknown", AnswerScope: "needs_clarification", Confidence: 0.5, Reason: "message intent is unclear", } if message == "" { ret.MessageIntent = "unknown" ret.AnswerScope = "needs_clarification" ret.Confidence = 0.9 ret.Reason = "empty message" return ret } lower := strings.ToLower(message) switch { case isGreetingMessage(lower): ret.MessageIntent = "greeting" ret.AnswerScope = "direct_reply" ret.Confidence = 0.98 ret.Reason = "matched greeting phrase" case containsAnyWorkflowText(lower, "谢谢", "感谢", "多谢", "辛苦了", "thank"): ret.MessageIntent = "thanks" ret.AnswerScope = "direct_reply" ret.Confidence = 0.95 ret.Reason = "matched thanks phrase" case containsAnyWorkflowText(lower, "再见", "拜拜", "不用了", "没事了", "结束"): ret.MessageIntent = "end_conversation" ret.AnswerScope = "direct_reply" ret.Confidence = 0.9 ret.Reason = "matched ending phrase" case containsAnyWorkflowText(lower, "人工", "转人工", "真人", "客服"): ret.MessageIntent = "handoff_request" ret.AnswerScope = "needs_handoff" ret.Confidence = 0.95 ret.RiskSignals = append(ret.RiskSignals, "handoff_requested") ret.Reason = "matched handoff phrase" case containsAnyWorkflowText(lower, "投诉", "举报", "差评", "曝光", "起诉", "律师", "12315"): ret.MessageIntent = "complaint" ret.AnswerScope = "needs_handoff" ret.Confidence = 0.92 ret.RiskSignals = append(ret.RiskSignals, "complaint_escalation") ret.Reason = "matched complaint phrase" case containsAnyWorkflowText(lower, "工单", "报障", "售后", "登记问题", "记录问题"): ret.MessageIntent = "ticket_request" ret.AnswerScope = "needs_ticket" ret.Confidence = 0.9 ret.RiskSignals = append(ret.RiskSignals, "ticket_expected") ret.Reason = "matched ticket phrase" case containsAnyWorkflowText(lower, "确认", "可以", "好的", "好", "是的", "取消"): ret.MessageIntent = "confirmation" ret.AnswerScope = "direct_reply" ret.Confidence = 0.8 ret.Reason = "matched confirmation phrase" case isAmbiguousWorkflowQuestion(lower): ret.MessageIntent = "ambiguous_question" ret.AnswerScope = "needs_clarification" ret.Confidence = 0.82 ret.Reason = "message lacks a concrete business object" default: ret.MessageIntent = "business_question" ret.AnswerScope = "needs_knowledge" ret.Confidence = 0.7 ret.Reason = "default business question policy" } return ret } func decideWorkflowReplyPolicy(aiAgent models.AIAgent, input workflowReplyPolicyInput) workflowReplyPolicyDecision { intent := strings.TrimSpace(input.MessageIntent) scope := strings.TrimSpace(input.AnswerScope) if answerability := strings.TrimSpace(input.Answerability); answerability != "" && answerability != "answerable" { return workflowReplyPolicyDecision{ Action: "knowledge_fallback", ReplyText: workflowKnowledgeFallbackReply(aiAgent), Reason: "knowledge is not sufficient for business answer", FinalReplySource: "knowledge_fallback", } } switch { case intent == "greeting": return workflowReplyPolicyDecision{Action: "direct_reply", ReplyText: "您好,请问有什么可以帮您?", Reason: "greeting can be answered directly", FinalReplySource: "direct_reply"} case intent == "thanks": return workflowReplyPolicyDecision{Action: "direct_reply", ReplyText: "不客气,如有其他问题可以继续告诉我。", Reason: "thanks can be answered directly", FinalReplySource: "direct_reply"} case intent == "end_conversation": return workflowReplyPolicyDecision{Action: "end_conversation", ReplyText: "好的,如后续还有问题可以随时联系。", Reason: "conversation ending phrase", FinalReplySource: "direct_reply"} case intent == "confirmation": return workflowReplyPolicyDecision{Action: "direct_reply", ReplyText: "好的,请继续补充需要处理的问题。", Reason: "confirmation without pending interrupt", FinalReplySource: "direct_reply"} case intent == "handoff_request" || scope == "needs_handoff": return workflowReplyPolicyDecision{Action: "handoff_to_human", Reason: "user requested human support or risk requires handoff", RequiresFlow: true, TargetFlow: "handoff_to_human", FinalReplySource: "handoff_notice"} case intent == "ticket_request" || scope == "needs_ticket": return workflowReplyPolicyDecision{Action: "prepare_ticket", Reason: "user requested ticket handling", RequiresFlow: true, TargetFlow: "prepare_ticket", FinalReplySource: "ticket_result"} case intent == "ambiguous_question" || scope == "needs_clarification": return workflowReplyPolicyDecision{Action: "clarify", ReplyText: "请补充具体的产品、场景、报错信息或你希望处理的结果,我再继续帮你确认。", Reason: "message needs clarification", FinalReplySource: "clarification"} case scope == "needs_knowledge": return workflowReplyPolicyDecision{Action: "retrieve_knowledge", Reason: "business question should be answered with knowledge evidence", RequiresFlow: true, TargetFlow: "knowledge", FinalReplySource: "knowledge_answer"} default: return workflowReplyPolicyDecision{Action: "clarify", ReplyText: "请补充更具体的问题,我再继续帮你处理。", Reason: "fallback to clarification for unclear policy input", FinalReplySource: "clarification"} } } func normalizeWorkflowUserMessage(value string) string { value = strings.TrimSpace(value) if value == "" { return "" } value = workflowHTMLTagPattern.ReplaceAllString(value, " ") value = html.UnescapeString(value) value = strings.Join(strings.Fields(value), " ") return strings.TrimSpace(value) } func isGreetingMessage(value string) bool { trimmed := strings.Trim(value, " \r\n。.!!??~~") return containsAnyWorkflowText(trimmed, "你好", "您好", "在吗", "在不在") || trimmed == "hello" || trimmed == "hi" } func isAmbiguousWorkflowQuestion(value string) bool { trimmed := strings.Trim(value, " \r\n。.!!??~~") return containsAnyWorkflowText(trimmed, "怎么弄", "怎么办", "怎么处理", "帮我看看", "有问题") || len([]rune(trimmed)) <= 3 } func containsAnyWorkflowText(value string, needles ...string) bool { for _, needle := range needles { if strings.Contains(value, needle) { return true } } return false } func (e *Executor) executeHumanConfirm(state *runState, node dsl.Node) error { prompt := strings.TrimSpace(toString(state.resolveInput(node, "prompt"))) if prompt == "" { prompt = "请确认是否继续。" } infoPreview, err := json.Marshal(map[string]string{"message": prompt}) if err != nil { return err } state.result.Interrupted = true state.result.CheckPointID = buildWorkflowCheckPointID(state.input, node.ID) checkpoint, err := json.Marshal(workflowCheckPoint{ Definition: state.input.Definition, ConfirmNodeID: node.ID, Vars: state.vars, }) if err != nil { return err } state.result.CheckPointData = string(checkpoint) state.result.Interrupts = []InterruptSummary{ { Type: workflowregistry.NodeTypeHumanConfirm, ID: node.ID, InfoPreview: string(infoPreview), }, } return nil } func buildWorkflowCheckPointID(input Input, nodeID string) string { return fmt.Sprintf("workflow:%d:%d:%s", input.Conversation.ID, input.UserMessage.ID, strings.TrimSpace(nodeID)) } func (e *Executor) executePrepareTicketDraft(ctx context.Context, state *runState, node dsl.Node) error { issue := strings.TrimSpace(toString(state.resolveInput(node, "issue"))) input := graphs.PrepareTicketDraftInput{ Issue: issue, } if title := strings.TrimSpace(readStringConfig(node.Data.Config, "title")); title != "" { input.Title = title } if description := strings.TrimSpace(readStringConfig(node.Data.Config, "description")); description != "" { input.Description = description } if impact := strings.TrimSpace(readStringConfig(node.Data.Config, "impact")); impact != "" { input.Impact = impact } if expectedOutcome := strings.TrimSpace(readStringConfig(node.Data.Config, "expectedOutcome")); expectedOutcome != "" { input.ExpectedOutcome = expectedOutcome } if currentAttempt := strings.TrimSpace(readStringConfig(node.Data.Config, "currentAttempt")); currentAttempt != "" { input.CurrentAttempt = currentAttempt } _, raw, err := readtools.ExecuteGraphTool(ctx, state.input.Conversation, toolx.GraphPrepareTicketDraft.Code, map[string]any{ "title": input.Title, "description": input.Description, "issue": input.Issue, "impact": input.Impact, "expectedOutcome": input.ExpectedOutcome, "currentAttempt": input.CurrentAttempt, }, workflowReadToolPolicy(toolx.GraphPrepareTicketDraft.Code)) if err != nil { return err } var result graphs.PrepareTicketDraftResult if err := json.Unmarshal([]byte(raw), &result); err != nil { return err } state.setNodeVars(node.ID, map[string]any{ "ticketDraft": ticketDraftWorkflowOutput(result), "ready": result.Ready, "title": strings.TrimSpace(result.Title), "description": strings.TrimSpace(result.Description), "missingFields": result.MissingFields, "followUpQuestions": result.FollowUpQuestions, "conversationFacts": result.ConversationFacts, }) return nil } func ticketDraftWorkflowOutput(result graphs.PrepareTicketDraftResult) map[string]any { return map[string]any{ "ready": result.Ready, "title": strings.TrimSpace(result.Title), "description": strings.TrimSpace(result.Description), "missingFields": result.MissingFields, "followUpQuestions": result.FollowUpQuestions, "conversationFacts": result.ConversationFacts, } } 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.Data.Config, "goal")) != "" { input.Goal = strings.TrimSpace(readStringConfig(node.Data.Config, "goal")) } if readBoolConfig(node.Data.Config, "needTicket") { input.NeedTicket = true } if readBoolConfig(node.Data.Config, "needHumanHandoff") { input.NeedHumanHandoff = true } if readBoolConfig(node.Data.Config, "needQualityCheck") { input.NeedQualityCheck = true } if strings.TrimSpace(readStringConfig(node.Data.Config, "additionalContext")) != "" { input.AdditionalContext = strings.TrimSpace(readStringConfig(node.Data.Config, "additionalContext")) } _, raw, err := readtools.ExecuteGraphTool(ctx, state.input.Conversation, toolx.GraphAnalyzeConversation.Code, map[string]any{ "goal": input.Goal, "observedIssue": input.ObservedIssue, "needTicket": input.NeedTicket, "needHumanHandoff": input.NeedHumanHandoff, "needQualityCheck": input.NeedQualityCheck, "additionalContext": input.AdditionalContext, }, workflowReadToolPolicy(toolx.GraphAnalyzeConversation.Code)) 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 workflowReadToolPolicy(toolCode string) aitooling.Policy { return aitooling.Policy{ AllowedToolCodes: []string{toolCode}, AllowedRiskLevels: []string{aitooling.RiskLevelRead}, Confirmed: true, } } func (e *Executor) executeHandoffToHuman(state *runState, node dsl.Node) error { if _, hasConfirmedInput := node.Data.InputsValues["confirmed"]; hasConfirmedInput && !truthy(state.resolveInput(node, "confirmed")) { state.setNodeVars(node.ID, map[string]any{ "handoffId": int64(0), "reason": strings.TrimSpace(toString(state.resolveInput(node, "reason"))), "decision": "cancelled", "teamId": int64(0), "assigneeId": int64(0), "message": "", "skipped": true, }) return nil } if state.input.Debug { state.setNodeVars(node.ID, map[string]any{ "handoffId": int64(0), "reason": strings.TrimSpace(toString(state.resolveInput(node, "reason"))), "decision": "cancelled", "teamId": int64(0), "assigneeId": int64(0), "message": "调试运行不会转人工。", "skipped": true, }) return nil } reason := strings.TrimSpace(toString(state.resolveInput(node, "reason"))) result, err := services.BusinessToolExecutor.Execute(context.Background(), services.BusinessToolInput{ Conversation: state.input.Conversation, AIAgent: state.input.AIAgent, ToolCode: toolx.GraphHandoffConversation.Code, Arguments: map[string]any{"reason": reason}, IdempotencyKey: workflowToolIdempotencyKey(state, node), Confirmed: true, }) if err != nil { return err } var handoff struct { Decision string `json:"decision"` TeamID int64 `json:"teamId"` AssigneeID int64 `json:"assigneeId"` Message string `json:"message"` } if err := json.Unmarshal([]byte(result.ResultData), &handoff); err != nil { return err } output := map[string]any{ "handoffId": int64(0), "reason": reason, "decision": "", "teamId": int64(0), "assigneeId": int64(0), "message": "", } output["decision"] = handoff.Decision output["teamId"] = handoff.TeamID output["assigneeId"] = handoff.AssigneeID output["message"] = strings.TrimSpace(handoff.Message) state.setNodeVars(node.ID, output) return nil } func workflowToolIdempotencyKey(state *runState, node dsl.Node) string { requestID := strings.TrimSpace(state.input.UserMessage.RequestID) if requestID != "" { return fmt.Sprintf("workflow:%d:node:%s:request:%s", state.input.Conversation.ID, node.ID, requestID) } return fmt.Sprintf("workflow:%d:node:%s:message:%d", state.input.Conversation.ID, node.ID, state.input.UserMessage.ID) } func (e *Executor) executeKnowledgeRetrieve(ctx context.Context, state *runState, node dsl.Node) error { query := strings.TrimSpace(toString(state.resolveInput(node, "query"))) knowledgeBaseIDs := readInt64ArrayConfig(node.Data.Config, "knowledgeBaseIds") if len(knowledgeBaseIDs) == 0 { return fmt.Errorf("knowledge retrieve node requires knowledgeBaseIds") } _, result, err := readtools.RetrieveKnowledge(ctx, state.input.AIAgent, knowledgeBaseIDs, query, workflowReadToolPolicy(toolx.BuiltinKnowledgeRetrieve.Code)) if err != nil { return err } items := make([]map[string]any, 0, len(result.ContextResults)) for _, item := range result.ContextResults { items = append(items, map[string]any{ "knowledgeBaseId": item.KnowledgeBaseID, "documentId": item.DocumentID, "chunkId": item.ChunkID, "content": item.Content, "score": item.Score, }) } state.result.RetrieverCount = len(result.Hits) state.setNodeVars(node.ID, map[string]any{ "items": items, "summary": result.ContextText, }) return nil } func (e *Executor) executeAnswerabilityGate(state *runState, node dsl.Node) error { items := state.resolveInput(node, "knowledgeItems") answerability := "unanswerable" reason := "no retrieved knowledge items" if hasItems(items) { answerability = "answerable" reason = "retrieved knowledge items are available" } state.setNodeVars(node.ID, map[string]any{ "answerability": answerability, "reason": reason, }) return nil } func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node dsl.Node) error { if staticReply := strings.TrimSpace(readStringConfig(node.Data.Config, "staticReply")); staticReply != "" { state.setNodeVars(node.ID, map[string]any{"replyText": renderWorkflowTemplate(staticReply, state.resolvedInputs(node))}) return nil } userPrompt := strings.TrimSpace(toString(state.resolveInput(node, "userMessage"))) if userPrompt == "" { userPrompt = strings.TrimSpace(state.input.UserMessage.Content) } knowledgeItems := toString(state.resolveInput(node, "knowledgeItems")) systemPrompt := strings.TrimSpace(state.input.AIAgent.SystemPrompt) if prompt := strings.TrimSpace(readStringConfig(node.Data.Config, "prompt")); prompt != "" { systemPrompt = strings.TrimSpace(systemPrompt + "\n\n" + prompt) } if _, declaresKnowledge := node.Data.InputsValues["knowledgeItems"]; declaresKnowledge && !hasItems(state.resolveInput(node, "knowledgeItems")) { state.setNodeVars(node.ID, map[string]any{"replyText": workflowKnowledgeFallbackReply(state.input.AIAgent)}) return nil } if knowledgeItems != "" { userPrompt = userPrompt + "\n\nKnowledge context:\n" + knowledgeItems } result, err := ai.LLM.ChatWithConfig(ctx, state.input.AIConfig, systemPrompt, userPrompt) if err != nil { return err } state.result.PromptTokens += result.PromptTokens state.result.CompletionTokens += result.CompletionTokens state.setNodeVars(node.ID, map[string]any{"replyText": result.Content}) return nil } func workflowKnowledgeFallbackReply(aiAgent models.AIAgent) string { if reply := strings.TrimSpace(aiAgent.FallbackMessage); reply != "" { return reply } if aiAgent.FallbackMode == enums.AIAgentFallbackModeSuggestRetry { return "当前知识库里没有找到足够明确的信息,你可以换个更具体的问法再试一次。" } return "当前知识库暂无明确信息。" } func (s *runState) nextNodeID(sourceNodeID string) (string, bool, error) { edges := s.outgoing[sourceNodeID] if len(edges) == 0 { return "", false, nil } node := s.nodesByID[sourceNodeID] if strings.TrimSpace(node.Type) != workflowregistry.NodeTypeCondition { return strings.TrimSpace(edges[0].TargetNodeID), true, nil } if rawConditions, ok := node.Data.Extra["conditions"]; ok { return s.nextFlowGramConditionNodeID(sourceNodeID, rawConditions) } config := dsl.ConditionConfig{} if len(node.Data.Config) > 0 { if err := json.Unmarshal(node.Data.Config, &config); err != nil { return "", false, fmt.Errorf("invalid condition node config: %w", err) } } evaluations := make([]conditionEvaluation, 0) for _, branch := range config.Branches { if branch.Default { continue } matched, evaluation, err := s.evaluateConditionBranch(sourceNodeID, branch) if err != nil { return "", false, err } evaluations = append(evaluations, evaluation) if matched { targetNodeID := strings.TrimSpace(branch.TargetNodeID) s.branchDecisions[sourceNodeID] = branchDecision{ SelectedEdgeID: s.edgeIDForTarget(sourceNodeID, targetNodeID), SelectedBranchID: strings.TrimSpace(branch.ID), SelectedBranchName: strings.TrimSpace(branch.Name), SelectedTargetNodeID: targetNodeID, Reason: "condition branch matched", Evaluations: evaluations, } return targetNodeID, true, nil } } for _, branch := range config.Branches { if !branch.Default { continue } targetNodeID := strings.TrimSpace(branch.TargetNodeID) s.branchDecisions[sourceNodeID] = branchDecision{ SelectedEdgeID: s.edgeIDForTarget(sourceNodeID, targetNodeID), SelectedBranchID: strings.TrimSpace(branch.ID), SelectedBranchName: strings.TrimSpace(branch.Name), SelectedTargetNodeID: targetNodeID, Reason: "no condition branch matched; selected default branch", Evaluations: evaluations, } return targetNodeID, true, nil } s.branchDecisions[sourceNodeID] = branchDecision{ Reason: "no condition branch matched and no default branch exists", Evaluations: evaluations, } return "", false, nil } func (s *runState) nextFlowGramConditionNodeID(sourceNodeID string, raw json.RawMessage) (string, bool, error) { var conditions []dsl.FlowGramConditionItem if err := json.Unmarshal(raw, &conditions); err != nil { return "", false, fmt.Errorf("invalid FlowGram condition data: %w", err) } evaluations := make([]conditionEvaluation, 0, len(conditions)) for _, item := range conditions { matched, evaluation, err := s.evaluateFlowGramCondition(sourceNodeID, item) if err != nil { return "", false, err } evaluations = append(evaluations, evaluation) if !matched { continue } if edge, ok := s.edgeForSourcePort(sourceNodeID, item.Key); ok { targetNodeID := strings.TrimSpace(edge.TargetNodeID) s.branchDecisions[sourceNodeID] = branchDecision{ SelectedEdgeID: strings.TrimSpace(item.Key), SelectedBranchID: strings.TrimSpace(item.Key), SelectedTargetNodeID: targetNodeID, Reason: "condition branch matched", Evaluations: evaluations, } return targetNodeID, true, nil } } if edge, ok := s.edgeForSourcePort(sourceNodeID, "else"); ok { targetNodeID := strings.TrimSpace(edge.TargetNodeID) s.branchDecisions[sourceNodeID] = branchDecision{ SelectedEdgeID: "else", SelectedBranchID: "else", SelectedTargetNodeID: targetNodeID, Reason: "no condition branch matched; selected else branch", Evaluations: evaluations, } return targetNodeID, true, nil } return "", false, nil } func (s *runState) evaluateFlowGramCondition(sourceNodeID string, item dsl.FlowGramConditionItem) (bool, conditionEvaluation, error) { left := s.resolveValue(item.Value.Left) right := s.resolveValue(item.Value.Right) operator := strings.TrimSpace(item.Value.Operator) evaluation := conditionEvaluation{ EdgeID: strings.TrimSpace(item.Key), BranchID: strings.TrimSpace(item.Key), SourceNodeID: sourceNodeID, Operator: operator, LeftValue: left, RightValue: right, } evaluation.SourceNodeID, evaluation.SourceField, _ = item.Value.Left.Ref() var matched bool switch operator { case "eq", "equals": matched = compareString(left, right) == 0 case "neq", "not_equals": matched = compareString(left, right) != 0 case "contains": matched = strings.Contains(toString(left), toString(right)) case "exists": matched = exists(left) case "not_exists": matched = !exists(left) case "truthy", "is_true": matched = truthy(left) case "falsy", "is_false": matched = !truthy(left) case "gt": matched = compareNumber(left, right) > 0 case "gte": matched = compareNumber(left, right) >= 0 case "lt": matched = compareNumber(left, right) < 0 case "lte": matched = compareNumber(left, right) <= 0 default: return false, evaluation, fmt.Errorf("unsupported workflow condition operator: %s", operator) } evaluation.Matched = matched return matched, evaluation, nil } func (s *runState) edgeForSourcePort(sourceNodeID string, sourcePortID string) (dsl.Edge, bool) { for _, edge := range s.outgoing[sourceNodeID] { if strings.TrimSpace(edge.SourcePortID) == strings.TrimSpace(sourcePortID) { return edge, true } } return dsl.Edge{}, false } func (s *runState) evaluateConditionBranch(sourceNodeID string, branch dsl.ConditionBranch) (bool, conditionEvaluation, error) { condition := branch.Condition targetNodeID := strings.TrimSpace(branch.TargetNodeID) evaluation := conditionEvaluation{ EdgeID: s.edgeIDForTarget(sourceNodeID, targetNodeID), BranchID: strings.TrimSpace(branch.ID), BranchName: strings.TrimSpace(branch.Name), TargetNodeID: targetNodeID, } if condition == nil { evaluation.Matched = true return true, evaluation, nil } operator := strings.TrimSpace(condition.Operator) var left any if condition.Left != nil { left = s.resolveValue(*condition.Left) evaluation.SourceNodeID, evaluation.SourceField, _ = condition.Left.Ref() evaluation.SourceNodeID = strings.TrimSpace(evaluation.SourceNodeID) evaluation.SourceField = strings.TrimSpace(evaluation.SourceField) } evaluation.Operator = operator evaluation.LeftValue = left evaluation.RightValue = condition.Right if operator == "" && strings.TrimSpace(condition.Expression) != "" { return false, evaluation, fmt.Errorf("free-form workflow condition expressions are not supported") } var matched bool switch operator { case "eq", "equals": matched = compareString(left, condition.Right) == 0 case "neq", "not_equals": matched = compareString(left, condition.Right) != 0 case "contains": matched = strings.Contains(toString(left), toString(condition.Right)) case "exists": matched = exists(left) case "not_exists": matched = !exists(left) case "truthy", "is_true": matched = truthy(left) case "falsy", "is_false": matched = !truthy(left) case "gt": matched = compareNumber(left, condition.Right) > 0 case "gte": matched = compareNumber(left, condition.Right) >= 0 case "lt": matched = compareNumber(left, condition.Right) < 0 case "lte": matched = compareNumber(left, condition.Right) <= 0 default: return false, evaluation, fmt.Errorf("unsupported workflow condition operator: %s", operator) } evaluation.Matched = matched return matched, evaluation, nil } func (s *runState) edgeIDForTarget(sourceNodeID string, targetNodeID string) string { for _, edge := range s.outgoing[sourceNodeID] { if strings.TrimSpace(edge.TargetNodeID) == targetNodeID { if edge.SourcePortID != "" { return strings.TrimSpace(edge.SourcePortID) } return strings.TrimSpace(edge.SourceNodeID + "->" + edge.TargetNodeID) } } return "" } func (s *runState) setNodeVars(nodeID string, values map[string]any) { s.vars[nodeID] = values } func (s *runState) resolveInput(node dsl.Node, inputName string) any { value, ok := node.Data.InputsValues[inputName] if !ok { return nil } return s.resolveValue(value) } func (s *runState) resolvedInputs(node dsl.Node) map[string]any { inputs := make(map[string]any, len(node.Data.InputsValues)) for name, value := range node.Data.InputsValues { inputs[name] = s.resolveValue(value) } return inputs } func (s *runState) nodeInputPreview(node dsl.Node) map[string]any { inputs := s.resolvedInputs(node) ret := map[string]any{ "inputs": inputs, } if len(node.Data.Config) > 0 { var cfg any if err := json.Unmarshal(node.Data.Config, &cfg); err == nil { ret["config"] = cfg } else { ret["config"] = string(node.Data.Config) } } return ret } func renderWorkflowTemplate(template string, values map[string]any) string { if strings.TrimSpace(template) == "" || len(values) == 0 { return template } return workflowTemplateVariablePattern.ReplaceAllStringFunc(template, func(match string) string { parts := workflowTemplateVariablePattern.FindStringSubmatch(match) if len(parts) < 2 { return match } name := strings.TrimSpace(parts[1]) value, ok := values[name] if !ok { return "" } return workflowTemplateValueString(value) }) } func workflowTemplateValueString(value any) string { switch v := value.(type) { case nil: return "" case string: return strings.TrimSpace(v) case []string: return strings.TrimSpace(strings.Join(v, "\n")) case []any: parts := make([]string, 0, len(v)) for _, item := range v { if text := workflowTemplateValueString(item); text != "" { parts = append(parts, text) } } return strings.Join(parts, "\n") default: raw, err := json.Marshal(v) if err != nil { return strings.TrimSpace(fmt.Sprint(v)) } return strings.TrimSpace(string(raw)) } } func (s *runState) nodeOutputPreview(nodeID string) map[string]any { ret := map[string]any{ "outputs": s.vars[nodeID], } if decision, ok := s.branchDecisions[nodeID]; ok { ret["branchDecision"] = decision } return ret } func workflowPreviewJSON(value any) string { raw, err := json.Marshal(value) if err != nil { return "" } const maxPreviewBytes = 2000 if len(raw) <= maxPreviewBytes { return string(raw) } return string(raw[:maxPreviewBytes]) } func (s *runState) resolveValue(value dsl.Value) any { switch value.Type { case dsl.ValueTypeRef: nodeID, field, ok := value.Ref() if !ok { return nil } fields := s.vars[strings.TrimSpace(nodeID)] if fields == nil { return nil } return fields[strings.TrimSpace(field)] case dsl.ValueTypeConstant: return value.ConstantContent case dsl.ValueTypeTemplate: if len(value.Content) > 0 { return value.Content[0] } return nil default: return nil } } func readStringConfig(raw json.RawMessage, key string) string { if len(raw) == 0 { return "" } var cfg map[string]any if err := json.Unmarshal(raw, &cfg); err != nil { return "" } 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 readInt64ArrayConfig(raw json.RawMessage, key string) []int64 { if len(raw) == 0 { return nil } var cfg map[string]any if err := json.Unmarshal(raw, &cfg); err != nil { return nil } items, ok := cfg[key].([]any) if !ok { return nil } ret := make([]int64, 0, len(items)) for _, item := range items { switch value := item.(type) { case float64: ret = append(ret, int64(value)) case int64: ret = append(ret, value) case int: ret = append(ret, int64(value)) } } return ret } func compareString(left any, right any) int { return strings.Compare(toString(left), toString(right)) } func compareNumber(left any, right any) int { leftNum := toFloat(left) rightNum := toFloat(right) switch { case leftNum > rightNum: return 1 case leftNum < rightNum: return -1 default: return 0 } } func toString(value any) string { switch v := value.(type) { case nil: return "" case string: return v case fmt.Stringer: return v.String() case []map[string]any: buf, _ := json.Marshal(v) return string(buf) default: return strings.TrimSpace(fmt.Sprint(v)) } } func toFloat(value any) float64 { switch v := value.(type) { case int: return float64(v) case int64: return float64(v) case float64: return v case float32: return float64(v) case json.Number: f, _ := v.Float64() return f case string: f, _ := strconv.ParseFloat(strings.TrimSpace(v), 64) return f default: return 0 } } func toInt64(value any) int64 { return int64(toFloat(value)) } func toInt64Slice(value any) []int64 { rv := reflect.ValueOf(value) if !rv.IsValid() || (rv.Kind() != reflect.Array && rv.Kind() != reflect.Slice) { return nil } ret := make([]int64, 0, rv.Len()) for index := 0; index < rv.Len(); index++ { if id := toInt64(rv.Index(index).Interface()); id > 0 { ret = append(ret, id) } } return ret } func asMap(value any) map[string]any { switch v := value.(type) { case map[string]any: return v case map[string]string: ret := make(map[string]any, len(v)) for key, item := range v { ret[key] = item } return ret case string: var ret map[string]any if err := json.Unmarshal([]byte(strings.TrimSpace(v)), &ret); err == nil { return ret } } return map[string]any{} } func truthy(value any) bool { switch v := value.(type) { case nil: return false case bool: return v case string: normalized := strings.ToLower(strings.TrimSpace(v)) return normalized != "" && normalized != "false" && normalized != "0" default: return !reflect.ValueOf(value).IsZero() } } func exists(value any) bool { if value == nil { return false } switch v := value.(type) { case string: return strings.TrimSpace(v) != "" default: return true } } func hasItems(value any) bool { if value == nil { return false } rv := reflect.ValueOf(value) switch rv.Kind() { case reflect.Array, reflect.Slice, reflect.Map: return rv.Len() > 0 default: return exists(value) } }