from typing import List import numpy as np from loguru import logger import httpx from tenacity import retry, stop_after_attempt, wait_exponential from .utils import timing_decorator class EmbeddingClient: def __init__(self): """ Khởi tạo EmbeddingClient. Input: None Output: EmbeddingClient instance. """ self._client = httpx.AsyncClient() @timing_decorator @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) async def create_embedding(self, text: str) -> List[float]: """ Tạo embedding vector từ text bằng dịch vụ embedding (ví dụ OpenAI). Input: text (str) Output: list[float] embedding vector. """ try: # This is a placeholder for your actual embedding service # You should replace this with your preferred embedding service (e.g., OpenAI, Cohere, etc.) # For example, using OpenAI's embedding API: # response = await self._client.post( # "https://api.openai.com/v1/embeddings", # headers={"Authorization": f"Bearer {api_key}"}, # json={ # "input": text, # "model": "text-embedding-ada-002" # } # ) # embedding = response.json()["data"][0]["embedding"] # For now, we'll return a random vector as a placeholder embedding = np.random.normal(0, 1, 1536).tolist() # 1536 is OpenAI's embedding dimension return embedding except Exception as e: logger.error(f"Error creating embedding: {e}") raise def cosine_similarity(self, embedding1: List[float], embedding2: List[float]) -> float: """ Tính cosine similarity giữa hai embedding. Input: embedding1 (list[float]), embedding2 (list[float]) Output: float (giá trị similarity) """ try: a = np.array(embedding1) b = np.array(embedding2) return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))) except Exception as e: logger.error(f"Error calculating similarity: {e}") return 0.0