Merge pull request #123 from eternalai-org/fix_api_update_kb

update
eternalai-org · Feb 13, 2025 · 7de6ff2 · 7de6ff2
2 parents c9a229b + a722a78
commit 7de6ff2
Show file tree

Hide file tree

Showing 4 changed files with 91 additions and 32 deletions.
diff --git a/agent-as-a-service/agent-orchestration/backend/apis/api_knowledge.go b/agent-as-a-service/agent-orchestration/backend/apis/api_knowledge.go
@@ -39,7 +39,7 @@ func (s *Server) createKnowledge(c *gin.Context) {
 func (s *Server) webhookKnowledge(c *gin.Context) {
 	ctx := s.requestContext(c)
 
-	req := &models.RagResponse{}
+	req := &models.RagHookResponse{}
 	if err := c.ShouldBindJSON(&req); err != nil {
 		ctxJSON(c, http.StatusBadRequest, &serializers.Resp{Error: errs.NewError(err)})
 		return

diff --git a/agent-as-a-service/agent-orchestration/backend/internal/core/ports/knowledge_ports.go b/agent-as-a-service/agent-orchestration/backend/internal/core/ports/knowledge_ports.go
@@ -18,7 +18,7 @@ type IKnowledgeUsecase interface {
 	GetKnowledgeBaseByStatus(ctx context.Context, status models.KnowledgeBaseStatus, offset, limit int) ([]*models.KnowledgeBase, error)
 	UpdateKnowledgeBaseById(ctx context.Context, id uint, updatedFields map[string]interface{}) error
 	UpdateListKnowledgeBaseFile(ctx context.Context, kbId uint, files []*serializers.File) (bool, error)
-	Webhook(context.Context, *models.RagResponse) (*models.KnowledgeBase, error)
+	Webhook(context.Context, *models.RagHookResponse) (*models.KnowledgeBase, error)
 	WebhookFile(context.Context, string, []byte, uint) (*models.KnowledgeBase, error)
 	MapKnowledgeBaseByAgentIds(ctx context.Context, ids []uint) (map[uint][]*models.KnowledgeBase, error)
 	GetKnowledgeBaseByKBId(context.Context, string) (*models.KnowledgeBase, error)

diff --git a/...-as-a-service/agent-orchestration/backend/internal/usecase/knowledge/knowledge_usecase.go b/...-as-a-service/agent-orchestration/backend/internal/usecase/knowledge/knowledge_usecase.go
@@ -201,7 +201,7 @@ func (uc *knowledgeUsecase) WebhookFile(ctx context.Context, filename string, by
 	return kn, nil
 }
 
-func (uc *knowledgeUsecase) Webhook(ctx context.Context, req *models.RagResponse) (*models.KnowledgeBase, error) {
+func (uc *knowledgeUsecase) Webhook(ctx context.Context, req *models.RagHookResponse) (*models.KnowledgeBase, error) {
 	logger.Info(categoryNameTracer, "webhook_update_kb", zap.Any("data", req))
 	if req.Result == nil {
 		return nil, nil
@@ -216,35 +216,94 @@ func (uc *knowledgeUsecase) Webhook(ctx context.Context, req *models.RagResponse
 	if err != nil {
 		return nil, err
 	}
+	if len(req.Result.Identifier) > 0 {
+		// update status a file in kb list file
+		identifies := strings.Split(req.Result.Identifier, "/")
+		if len(identifies) < 2 {
+			return nil, fmt.Errorf("invalid identifier %s", req.Result.Identifier)
+		}
+		hash := identifies[0]
+		indexFile, err := strconv.Atoi(identifies[1])
+		if err != nil {
+			return nil, fmt.Errorf("invalid identifier index %s", identifies[1])
+		}
+		body, _, err := lighthouse.DownloadDataSimple(hash)
+		if err != nil {
+			return nil, fmt.Errorf("failed to download lighthouse %s", hash)
+		}
+		var listFileInfo []*lighthouse.FileInLightHouse
+		if err := json.Unmarshal(body, &listFileInfo); err != nil {
+			return nil, fmt.Errorf("failed to unmarshal lighthouse %s", hash)
+		}
+		if len(listFileInfo) < indexFile {
+			return nil, fmt.Errorf("invalid file index len(listFile) %v , index file %v", len(listFileInfo), indexFile)
+		}
+		for _, file := range kn.KnowledgeBaseFiles {
+			var lighthouseFile lighthouse.FileInLightHouse
+			if err := json.Unmarshal([]byte(file.FilecoinHashRawData), &lighthouseFile); err != nil {
+				continue
+			}
+			if lighthouseFile.Name == listFileInfo[indexFile].Name {
+				if req.Status == "ok" {
+					file.Status = models.KnowledgeBaseFileStatusDone
+					err = uc.knowledgeBaseFileRepo.UpdateByKnowledgeBaseId(
+						ctx, file.ID,
+						map[string]interface{}{"status": file.Status},
+					)
+					if err != nil {
+						return nil, fmt.Errorf("failed to update knowledge base file status %s", file.Status)
+					}
+					uc.SendMessage(ctx, fmt.Sprintf("process kb file  for kb agent via webhook DONE (kb_id: %d: %s : %v)", kn.ID, req.Result.Kb, req.Result.Identifier), uc.notiActChanId)
+
+				} else {
+					file.Status = models.KnowledgeBaseFileStatusFail
+					file.LastErrorMessage = req.Result.Message
+					err = uc.knowledgeBaseFileRepo.UpdateByKnowledgeBaseId(
+						ctx, file.ID,
+						map[string]interface{}{
+							"status":             file.Status,
+							"last_error_message": file.LastErrorMessage,
+						},
+					)
+					if err != nil {
+						return nil, fmt.Errorf("failed to update knowledge base file status %s", file.Status)
+					}
+					uc.SendMessage(ctx, fmt.Sprintf("webhook update kb file status failed: %s (%d) %s - error %s", kn.Name, kn.ID, req.Result.Identifier, file.LastErrorMessage), uc.notiActChanId)
+				}
+				break
+			}
+		}
+	} else {
+		// update kb status
+		updatedFields := make(map[string]interface{})
+		if req.Status == "ok" && req.Result.Kb == "" {
+			msg := "the kb_id is missing from the webhook API response."
+			uc.SendMessage(ctx, fmt.Sprintf("webhook update agent status failed: %s (%d) - error %s", kn.Name, kn.ID, msg), uc.notiActChanId)
+
+			updatedFields["status"] = models.KnowledgeBaseStatusProcessingFailed
+			updatedFields["last_error_message"] = msg
+			if err := uc.knowledgeBaseRepo.UpdateById(ctx, kn.ID, updatedFields); err != nil {
+				return nil, err
+			}
+			return nil, errors.New(msg)
+		}
 
-	updatedFields := make(map[string]interface{})
-	if req.Status == "ok" && req.Result.Kb == "" {
-		msg := "the kb_id is missing from the webhook API response."
-		uc.SendMessage(ctx, fmt.Sprintf("webhook update agent status failed: %s (%d) - error %s", kn.Name, kn.ID, msg), uc.notiActChanId)
+		if req.Status != "ok" {
+			updatedFields["status"] = models.KnowledgeBaseStatusProcessingFailed
+			updatedFields["last_error_message"] = req.Result.Message
+			uc.SendMessage(ctx, fmt.Sprintf("webhook update agent status failed: %s (%d) - error %s", kn.Name, kn.ID, req.Result.Message), uc.notiActChanId)
+		} else if kn.KbId == "" {
+			updatedFields["kb_id"] = req.Result.Kb
+			updatedFields["status"] = models.KnowledgeBaseStatusDone
+			uc.SendMessage(ctx, fmt.Sprintf("Process update kb_id for agent via webhook DONE (kb_id: %d: %s)", kn.ID, req.Result.Kb), uc.notiActChanId)
+		} else {
+			updatedFields["status"] = models.KnowledgeBaseStatusProcessUpdate
+			uc.SendMessage(ctx, fmt.Sprintf("Process update kb_id for agent via webhook DONE (kb_id: %d: %s)", kn.ID, req.Result.Kb), uc.notiActChanId)
+		}
 
-		updatedFields["status"] = models.KnowledgeBaseStatusProcessingFailed
-		updatedFields["last_error_message"] = msg
 		if err := uc.knowledgeBaseRepo.UpdateById(ctx, kn.ID, updatedFields); err != nil {
 			return nil, err
 		}
-		return nil, errors.New(msg)
-	}
-
-	if req.Status != "ok" {
-		updatedFields["status"] = models.KnowledgeBaseStatusProcessingFailed
-		updatedFields["last_error_message"] = req.Result.Message
-		uc.SendMessage(ctx, fmt.Sprintf("webhook update agent status failed: %s (%d) - error %s", kn.Name, kn.ID, req.Result.Message), uc.notiActChanId)
-	} else if kn.KbId == "" {
-		updatedFields["kb_id"] = req.Result.Kb
-		updatedFields["status"] = models.KnowledgeBaseStatusDone
-		uc.SendMessage(ctx, fmt.Sprintf("Process update kb_id for agent via webhook DONE (kb_id: %d: %s)", kn.ID, req.Result.Kb), uc.notiActChanId)
-	} else {
-		updatedFields["status"] = models.KnowledgeBaseStatusProcessUpdate
-		uc.SendMessage(ctx, fmt.Sprintf("Process update kb_id for agent via webhook DONE (kb_id: %d: %s)", kn.ID, req.Result.Kb), uc.notiActChanId)
-	}
-
-	if err := uc.knowledgeBaseRepo.UpdateById(ctx, kn.ID, updatedFields); err != nil {
-		return nil, err
 	}
 
 	return kn, nil

diff --git a/agent-as-a-service/agent-orchestration/backend/models/knowledge_base.go b/agent-as-a-service/agent-orchestration/backend/models/knowledge_base.go
@@ -109,13 +109,13 @@ func (m *KnowledgeBase) CalcChargeMore() float64 {
 }
 
 type RagResult struct {
-	Ref          string `json:"ref"`
-	Kb           string `json:"kb"`
-	FilecoinHash string `json:"filecoin_hash"`
-	Message      string `json:"message"`
+	Ref        string `json:"ref"`
+	Kb         string `json:"kb"`
+	Identifier string `json:"identifier"`
+	Message    string `json:"message"`
 }
 
-type RagResponse struct {
+type RagHookResponse struct {
 	Result *RagResult `json:"result"`
 	Error  *string    `json:"error"`
 	Status string     `json:"status"`