contexta_be/app/services/embeddings.py

from openai import OpenAI
from app.config import settings
from typing import List
import logging

logger = logging.getLogger(__name__)

_openai_client = None


def get_openai_client() -> OpenAI:
    global _openai_client
    if _openai_client is None:
        _openai_client = OpenAI(api_key=settings.openai_api_key)
    return _openai_client


class EmbeddingService:
    def __init__(self):
        self.model = settings.embedding_model

    def embed_text(self, text: str) -> List[float]:
        """Generate embedding for a single text"""
        client = get_openai_client()
        try:
            response = client.embeddings.create(
                model=self.model,
                input=text,
            )
            return response.data[0].embedding
        except Exception as e:
            logger.error(f"Embedding error: {e}")
            raise

    def embed_batch(self, texts: List[str]) -> List[List[float]]:
        """Generate embeddings for multiple texts"""
        client = get_openai_client()
        try:
            # Clean texts
            cleaned = [t.replace("\n", " ").strip() for t in texts if t.strip()]
            if not cleaned:
                return []

            response = client.embeddings.create(
                model=self.model,
                input=cleaned,
            )
            return [item.embedding for item in response.data]
        except Exception as e:
            logger.error(f"Batch embedding error: {e}")
            raise


embedding_service = EmbeddingService()