aigenrec/luminabackend

1from langchain_together import TogetherEmbeddings2from config.settings import settings3from typing import List4from utils.logger import logger5import os6import asyncio7 8class EmbeddingService:9    def __init__(self):10        os.environ['TOGETHER_API_KEY'] = settings.TOGETHER_API_KEY11        self.embeddings = TogetherEmbeddings(12            model=settings.EMBEDDING_MODEL,13            together_api_key=settings.TOGETHER_API_KEY14        )15    16    async def generate_embeddings(self, texts: List[str]) -> List[List[float]]:17        """Generate embeddings for multiple texts"""18        try:19            # LangChain handles batching internally usually, but explicit batching is safer20            embeddings = []21            loop = asyncio.get_running_loop()22            batch_size = 2523            for i in range(0, len(texts), batch_size):24                batch = texts[i:i + batch_size]25                batch_embeddings = await loop.run_in_executor(26                    None,27                    lambda: self.embeddings.embed_documents(batch)28                )29                embeddings.extend(batch_embeddings)30            return embeddings31        except Exception as e:32            logger.error(f"Error generating embeddings: {str(e)}")33            raise34    35    async def generate_embedding(self, text: str) -> List[float]:36        """Generate embedding for a single text"""37        try:38            return self.embeddings.embed_query(text)39        except Exception as e:40            logger.error(f"Error generating embedding: {str(e)}")41            raise42 43embedding_service = EmbeddingService()