package services import ( "context" "encoding/json" "fmt" "sort" "strings" "sync" "code.tczkiot.com/wlw/ai-agent/contract" ) var BusinessActionToolService = &businessActionToolService{} type businessActionToolService struct { mu sync.RWMutex tools map[string]contract.BusinessActionTool } func SetBusinessActionTools(tools []contract.BusinessActionTool) error { registered := make(map[string]contract.BusinessActionTool, len(tools)) for _, tool := range tools { tool.Code = strings.TrimSpace(tool.Code) tool.Description = strings.TrimSpace(tool.Description) if tool.Code == "" { return fmt.Errorf("ai-agent: business action tool code is required") } if !strings.HasPrefix(tool.Code, "business/") { return fmt.Errorf("ai-agent: business action tool code must start with business/: %s", tool.Code) } if tool.Description == "" { return fmt.Errorf("ai-agent: business action tool description is required: %s", tool.Code) } if tool.Preview == nil || tool.Execute == nil { return fmt.Errorf("ai-agent: business action tool preview and executor are required: %s", tool.Code) } if _, exists := registered[tool.Code]; exists { return fmt.Errorf("ai-agent: duplicate business action tool code: %s", tool.Code) } if tool.InputSchema == nil { tool.InputSchema = map[string]any{"type": "object", "properties": map[string]any{}} } registered[tool.Code] = tool } BusinessActionToolService.mu.Lock() BusinessActionToolService.tools = registered BusinessActionToolService.mu.Unlock() return nil } func (s *businessActionToolService) ListForCustomerType(customerType string) []contract.BusinessActionTool { s.mu.RLock() defer s.mu.RUnlock() ret := make([]contract.BusinessActionTool, 0, len(s.tools)) for _, tool := range s.tools { if businessActionToolSupportsCustomerType(tool, customerType) { ret = append(ret, tool) } } sort.Slice(ret, func(i, j int) bool { return ret[i].Code < ret[j].Code }) return ret } func (s *businessActionToolService) ResolveForCustomerType(code, customerType string) (contract.BusinessActionTool, bool) { s.mu.RLock() defer s.mu.RUnlock() tool, ok := s.tools[strings.TrimSpace(code)] if !ok || !businessActionToolSupportsCustomerType(tool, customerType) { return contract.BusinessActionTool{}, false } return tool, true } func (s *businessActionToolService) Resolve(code string) (contract.BusinessActionTool, bool) { s.mu.RLock() defer s.mu.RUnlock() tool, ok := s.tools[strings.TrimSpace(code)] return tool, ok } func (s *businessActionToolService) Preview(ctx context.Context, tool contract.BusinessActionTool, businessContext contract.BusinessReadContext, arguments map[string]any) (string, error) { return tool.Preview(ctx, businessContext, arguments) } func (s *businessActionToolService) Execute(ctx context.Context, conversationID, aiAgentID int64, idempotencyKey string, tool contract.BusinessActionTool, businessContext contract.BusinessReadContext, arguments map[string]any) (*contract.BusinessActionResult, bool, error) { businessContext.CheckPointID = strings.TrimSpace(idempotencyKey) if tool.AuthorizeConfirmation != nil { if err := tool.AuthorizeConfirmation(ctx, businessContext, arguments, businessContext.CheckPointID); err != nil { return nil, false, err } } claim, err := AgentToolInvocationService.Claim(conversationID, aiAgentID, tool.Code, idempotencyKey) if err != nil { return nil, false, contract.NewBusinessActionError("操作请求记录创建失败,本次操作未执行,请稍后重试。", err) } if claim == nil || claim.Item == nil { err := fmt.Errorf("business action invocation could not be claimed") return nil, false, contract.NewBusinessActionError("操作请求无效,本次操作未执行,请重新发起。", err) } if claim.Completed { result := &contract.BusinessActionResult{} if err := json.Unmarshal([]byte(claim.Item.ResultData), result); err != nil { return nil, true, contract.NewBusinessActionError("操作已完成,但结果读取失败,请勿重复操作并联系人工客服核对。", err) } return result, true, nil } if claim.UnknownOutcome { err := fmt.Errorf("business action outcome requires reconciliation: %s", tool.Code) return nil, true, contract.NewUnknownOutcomeBusinessActionError("上次操作结果尚未确认,请勿重复操作,并联系人工客服核对。", err) } if !claim.Acquired { err := fmt.Errorf("business action is already running: %s", tool.Code) return nil, false, contract.NewBusinessActionError("操作正在处理中,请勿重复提交,请稍后查看结果。", err) } result, err := tool.Execute(ctx, businessContext, arguments) if err != nil { if businessActionFailureIsRetryable(ctx, err) { _ = AgentToolInvocationService.FailRetryable(claim.Item, err) } else { _ = AgentToolInvocationService.MarkUnknownOutcome(claim.Item, err) } return nil, false, err } if result == nil || strings.TrimSpace(result.Message) == "" { err = fmt.Errorf("business action returned an empty result: %s", tool.Code) _ = AgentToolInvocationService.MarkUnknownOutcome(claim.Item, err) return nil, false, contract.NewBusinessActionError("业务系统未返回操作结果,本次操作未完成,请联系人工客服核对。", err) } encoded, err := json.Marshal(result) if err != nil { // The host operation has already completed. Persist at least the customer // message and never mark the invocation retryable, which could execute the // same paid action twice. encoded, _ = json.Marshal(&contract.BusinessActionResult{Message: result.Message}) } if err := AgentToolInvocationService.Complete(claim.Item, string(encoded)); err != nil { // The external write may already have committed. Never leave the invocation // eligible for replay; persist an explicit reconciliation state whenever the // result cannot be durably recorded. _ = AgentToolInvocationService.MarkUnknownOutcome(claim.Item, err) return nil, false, contract.NewBusinessActionError("业务操作可能已经成功,但结果记录失败。请勿重复操作,并联系人工客服核对。", err) } return result, false, nil } func businessActionToolSupportsCustomerType(tool contract.BusinessActionTool, customerType string) bool { if len(tool.CustomerTypes) == 0 { return true } for _, candidate := range tool.CustomerTypes { if strings.EqualFold(strings.TrimSpace(candidate), strings.TrimSpace(customerType)) { return true } } return false }