backend: finish embedding processing queue

go-backend/handlers/cards.go

@@ -815,12 +815,15 @@ func (s *Handler) ChunkEmbedCard(userID, cardPK int) error {
 		log.Printf("error in chunking %v", err)
 		return err
 	}
-	embeddings, err := llms.GenerateEmbeddingsFromCard(s.DB, chunks)
-	if err != nil {
-		log.Printf("error generating embeddings: %v", err)
-		return err
-	}
-	llms.StoreEmbeddings(s.DB, userID, cardPK, embeddings)
+	for _, chunk := range chunks {
+		log.Printf("push chunk %v", chunk)
+		s.Server.LLMClient.EmbeddingQueue.Push(models.LLMRequest{
+			UserID: userID,
+			CardPK: cardPK,
+			Chunk:  chunk,
+		})
+	}
+	llms.StartProcessingQueue(s.Server.LLMClient)
 	return nil
 }
 

go-backend/llms/embeddings.go

@@ -10,6 +10,7 @@ import (
 	"log"
 	"net/http"
 	"os"
+	"time"
 
 	"github.com/pgvector/pgvector-go"
 	openai "github.com/sashabaranov/go-openai"
@@ -62,9 +63,25 @@ func ProcessQueue(c *models.LLMClient) {
 			if request.Retries < 4 {
 				c.EmbeddingQueue.Push(request)
 			}
+			continue
+		}
+		log.Printf("mainembeddings %v", request.Chunk)
+		err = StoreEmbeddings(
+			c.EmbeddingQueue.DB,
+			request.UserID,
+			request.CardPK,
+			[][]pgvector.Vector{embeddings},
+		)
+		if err != nil {
+			log.Printf("failed to store embed")
+			request.Retries += 1
+			if request.Retries < 4 {
+				c.EmbeddingQueue.Push(request)
+			}
+			continue
 		}
-		err = StoreEmbeddings(db, request.UserID, request.CardPK, embeddings)
 
+		time.Sleep(1000 * time.Millisecond)
 	}
 
 }

go-backend/main.go

@@ -132,7 +132,7 @@ func main() {
 	config := openai.DefaultConfig(os.Getenv("ZETTEL_LLM_KEY"))
 	config.BaseURL = os.Getenv("ZETTEL_LLM_ENDPOINT")
 
-	s.LLMClient = llms.NewClient(config)
+	s.LLMClient = llms.NewClient(s.DB, config)
 
 	go func() {
 		h.SyncStripePlans()

go-backend/models/chat.go

@@ -37,10 +37,10 @@ func NewEmbeddingQueue(db *sql.DB) *LLMRequestQueue {
 }
 
 // Push adds an email to the queue
-func (q *LLMRequestQueue) Push(email LLMRequest) {
+func (q *LLMRequestQueue) Push(request LLMRequest) {
 	q.Mu.Lock()
 	defer q.Mu.Unlock()
-	q.Queue = append(q.Queue, email)
+	q.Queue = append(q.Queue, request)
 }
 
 // Pop removes and returns the first email from the queue