fix duplicate message

app/embedding.py

@@ -2,9 +2,8 @@ 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_async, timing_decorator_sync
+from .utils import timing_decorator_async, timing_decorator_sync, call_endpoint_with_retry
 
 class EmbeddingClient:
     def __init__(self):
@@ -16,52 +15,27 @@ class EmbeddingClient:
         self._client = httpx.AsyncClient()
 
     @timing_decorator_async
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True)
     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.
         """
+        url = "https://vietcat-vietnameseembeddingv2.hf.space/embed"
+        payload = {"text": text}
         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
-
-            logger.info(f"[DEBUG] Bắt đầu tạo embedding: {text}")
-            response = await self._client.post(
-                # "https://vietcat-bgem3node.hf.space/embed",
-                "https://vietcat-vietnameseembeddingv2.hf.space/embed",
-                json={"text": text}
-            )
-            response.raise_for_status()
-            data = response.json()
-            logger.info(f"[DEBUG] Embedding API response: {data['embedding'][:10]}...{data['embedding'][-10:]}")
-            return data["embedding"]
+            response = await call_endpoint_with_retry(self._client, url, payload)
+            if response is not None:
+                data = response.json()
+                logger.info(f"[DEBUG] Embedding API response: {data['embedding'][:10]}...{data['embedding'][-10:]}")
+                return data["embedding"]
+            else:
+                logger.error("Embedding API response is None")
+                raise RuntimeError("Embedding API response is None")
         except Exception as e:
             logger.error(f"Error creating embedding: {e}")
-            if 'response' in locals():
-                try:
-                    logger.error(f"Embedding API status: {response.status_code}, content: {await response.aread()}")
-                except Exception as ex:
-                    logger.error(f"Error reading response content: {ex}")
             raise
 
-    @timing_decorator_sync
     def cosine_similarity(self, embedding1: List[float], embedding2: List[float]) -> float:
         """
         Tính cosine similarity giữa hai embedding.

app/llm.py

@@ -5,7 +5,7 @@ from loguru import logger
 from tenacity import retry, stop_after_attempt, wait_exponential
 import os
 
-from .utils import timing_decorator_async, timing_decorator_sync
+from .utils import timing_decorator_async, timing_decorator_sync, call_endpoint_with_retry
 
 class LLMClient:
     """
@@ -118,125 +118,60 @@ class LLMClient:
             logger.error(f"[LLM] Error generating text with {self.provider}: {e}")
             raise
     
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True)
     async def _generate_openai(self, prompt: str, system_prompt: Optional[str] = None, **kwargs) -> str:
-        """Generate text với OpenAI API."""
-        messages = []
-        
-        if system_prompt:
-            messages.append({"role": "system", "content": system_prompt})
-        
-        messages.append({"role": "user", "content": prompt})
-        
-        payload = {
-            "model": kwargs.get("model", self.model),
-            "messages": messages,
-            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
-            "temperature": kwargs.get("temperature", self.temperature),
-            "stream": False
-        }
-        
-        headers = {
-            "Authorization": f"Bearer {self.api_key}",
-            "Content-Type": "application/json"
-        }
-        
-        response = await self._client.post(
-            f"{self.base_url}/chat/completions",
-            headers=headers,
-            json=payload
-        )
-        response.raise_for_status()
-        
-        data = response.json()
-        return data["choices"][0]["message"]["content"]
+        url = f"{self.base_url}/chat/completions"
+        payload = {"model": kwargs.get("model", self.model), "messages": [{"role": "system", "content": system_prompt or ""}, {"role": "user", "content": prompt}], "max_tokens": kwargs.get("max_tokens", self.max_tokens), "temperature": kwargs.get("temperature", self.temperature), "stream": False}
+        headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
+        response = await call_endpoint_with_retry(self._client, url, payload, headers=headers)
+        if response is not None:
+            data = response.json()
+            return data["choices"][0]["message"]["content"]
+        else:
+            logger.error("OpenAI API response is None")
+            raise RuntimeError("OpenAI API response is None")
     
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True)
     async def _generate_huggingface(self, prompt: str, **kwargs) -> str:
-        """Generate text với HuggingFace API."""
-        payload = {
-            "inputs": prompt,
-            "parameters": {
-                "max_new_tokens": kwargs.get("max_tokens", self.max_tokens),
-                "temperature": kwargs.get("temperature", self.temperature),
-                "return_full_text": False
-            }
-        }
-        
-        headers = {
-            "Authorization": f"Bearer {self.api_key}",
-            "Content-Type": "application/json"
-        }
-        
-        response = await self._client.post(
-            f"{self.base_url}/models/{self.model}",
-            headers=headers,
-            json=payload
-        )
-        response.raise_for_status()
-        
-        data = response.json()
-        return data[0]["generated_text"]
+        url = f"{self.base_url}/generate"
+        payload = {"inputs": prompt}
+        response = await call_endpoint_with_retry(self._client, url, payload)
+        if response is not None:
+            data = response.json()
+            return data[0]["generated_text"]
+        else:
+            logger.error("HuggingFace API response is None")
+            raise RuntimeError("HuggingFace API response is None")
     
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True)
     async def _generate_local(self, prompt: str, **kwargs) -> str:
-        """Generate text với local model."""
-        payload = {
-            "prompt": prompt,
-            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
-            "temperature": kwargs.get("temperature", self.temperature),
-            "model": kwargs.get("model", self.model)
-        }
-        
-        response = await self._client.post(
-            f"{self.base_url}/generate",
-            json=payload
-        )
-        response.raise_for_status()
-        
-        data = response.json()
-        return data.get("text", "")
+        url = f"{self.base_url}/generate"
+        payload = {"prompt": prompt}
+        response = await call_endpoint_with_retry(self._client, url, payload)
+        if response is not None:
+            data = response.json()
+            return data.get("text", "")
+        else:
+            logger.error("Local API response is None")
+            raise RuntimeError("Local API response is None")
     
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True)
     async def _generate_custom(self, prompt: str, **kwargs) -> str:
-        """Generate text với custom provider."""
-        payload = {
-            "prompt": prompt,
-            "max_tokens": kwargs.get("max_tokens", self.max_tokens),
-            "temperature": kwargs.get("temperature", self.temperature),
-            "model": kwargs.get("model", self.model)
-        }
-        
-        headers = {}
-        if self.api_key:
-            headers["Authorization"] = f"Bearer {self.api_key}"
-        
-        response = await self._client.post(
-            f"{self.base_url}/generate",
-            headers=headers,
-            json=payload
-        )
-        response.raise_for_status()
-        
-        data = response.json()
-        return data.get("text", "")
+        url = f"{self.base_url}/custom"
+        payload = {"prompt": prompt}
+        response = await call_endpoint_with_retry(self._client, url, payload)
+        if response is not None:
+            data = response.json()
+            return data.get("text", "")
+        else:
+            logger.error("Custom API response is None")
+            raise RuntimeError("Custom API response is None")
     
-    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True)
     async def _generate_hfs(self, prompt: str, **kwargs) -> str:
-        """Generate text với HFS provider."""
         endpoint = f"{self.base_url}/purechat"
         payload = {"prompt": prompt}
-        logger.info(f"[LLM][FLOW] Chuẩn bị gửi request tới HFS endpoint: {endpoint}")
-        logger.info(f"[LLM][FLOW] Payload gửi đi: {payload}")
         headers = {}
         if self.api_key:
             headers["Authorization"] = f"Bearer {self.api_key}"
-        try:
-            response = await self._client.post(endpoint, headers=headers, json=payload)
-            logger.info(f"[LLM][FLOW] Đã nhận response từ HFS, status: {response.status_code}")
-            response.raise_for_status()
+        response = await call_endpoint_with_retry(self._client, endpoint, payload, 3, 300, headers=headers)
+        if response is not None:
             data = response.json()
-            logger.info(f"[LLM][FLOW] Response data: {data}")
             if 'response' in data:
                 return data['response']
             elif 'result' in data:
@@ -244,9 +179,9 @@ class LLMClient:
             elif 'data' in data and isinstance(data['data'], list):
                 return data['data'][0]
             return str(data)
-        except Exception as e:
-            logger.error(f"[LLM][FLOW][ERROR] Lỗi khi gọi HFS endpoint: {endpoint} | Exception: {e}")
-            raise
+        else:
+            logger.error("HFS API response is None")
+            raise RuntimeError("HFS API response is None")
 
     @timing_decorator_async
     async def chat(

app/utils.py

@@ -3,6 +3,8 @@ from functools import wraps
 from loguru import logger
 from typing import Any, Callable
 import os
+import asyncio
+import httpx
 
 def timing_decorator_async(func: Callable) -> Callable:
     """
@@ -98,4 +100,29 @@ def validate_config(settings) -> None:
         if not getattr(settings, field, None):
             missing.append(field)
     if missing:
-        raise RuntimeError(f"Missing config: {', '.join(missing)}") 
+        raise RuntimeError(f"Missing config: {', '.join(missing)}")
+
+def get_logger():
+    return logger
+
+async def call_endpoint_with_retry(client, url, payload, max_retries=3, base_timeout=30, headers=None):
+    logger = get_logger()
+    timeout = base_timeout
+    for attempt in range(1, max_retries + 1):
+        try:
+            response = await client.post(url, json=payload, timeout=timeout, headers=headers)
+            response.raise_for_status()
+            return response
+        except httpx.TimeoutException as e:
+            if attempt == max_retries:
+                raise
+            else:
+                logger.warning(f"Timeout (attempt {attempt}/{max_retries}), retrying with timeout={timeout * 2}s...")
+                timeout *= 2
+                await asyncio.sleep(1)
+        except httpx.HTTPStatusError as e:
+            logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
+            raise
+        except Exception as e:
+            logger.error(f"Other error: {e}")
+            raise