fix(avatar): normalize embedding API endpoint

digital-avatar-app/backend/embeddings.py

@@ -18,6 +18,14 @@ EMBED_DIM = 256
 MODEL = os.getenv("EMBEDDING_MODEL", "mock-hash-embed-v1")
 
 
+def _embedding_endpoint(api_url):
+    """Accept either an OpenAI-compatible base URL or its full endpoint."""
+    api_url = (api_url or "").strip().rstrip("/")
+    if not api_url or api_url.endswith("/embeddings"):
+        return api_url
+    return f"{api_url}/embeddings"
+
+
 def _tokenize(text):
     text = (text or "").lower()
     # 英文/数字按词，CJK 逐字（中文无空格，需拆到字级才能命中子词）
@@ -47,7 +55,7 @@ def embed(texts):
     """返回 list[list[float]]，与输入顺序一致。"""
     if not texts:
         return []
-    api_url = os.getenv("EMBEDDING_API_URL")
+    api_url = _embedding_endpoint(os.getenv("EMBEDDING_API_URL"))
     if api_url:
         api_key = os.getenv("EMBEDDING_API_KEY", "")
         model = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")

digital-avatar-app/backend/tests/test_embeddings.py

@@ -48,9 +48,11 @@ class RemoteEmbeddingTests(unittest.TestCase):
     def test_large_input_is_split_into_provider_safe_batches(self):
         texts = [f"chunk-{index}" for index in range(14)]
         batch_sizes = []
+        requested_urls = []
 
         def fake_urlopen(request, timeout):
             self.assertEqual(timeout, 30)
+            requested_urls.append(request.full_url)
             payload = json.loads(request.data.decode("utf-8"))
             batch_sizes.append(len(payload["input"]))
             return FakeResponse({
@@ -61,7 +63,7 @@ class RemoteEmbeddingTests(unittest.TestCase):
             })
 
         with patch.dict(os.environ, {
-            "EMBEDDING_API_URL": "https://embedding.example/v1/embeddings",
+            "EMBEDDING_API_URL": "https://embedding.example/v1",
             "EMBEDDING_API_KEY": "test-key",
             "EMBEDDING_MODEL": "text-embedding-v4",
             "EMBEDDING_BATCH_SIZE": "10",
@@ -69,8 +71,18 @@ class RemoteEmbeddingTests(unittest.TestCase):
             result = embeddings.embed(texts)
 
         self.assertEqual(batch_sizes, [10, 4])
+        self.assertEqual(requested_urls, [
+            "https://embedding.example/v1/embeddings",
+            "https://embedding.example/v1/embeddings",
+        ])
         self.assertEqual(result, [[float(index)] for index in range(14)])
 
+    def test_full_embedding_endpoint_is_not_modified(self):
+        self.assertEqual(
+            embeddings._embedding_endpoint("https://embedding.example/v1/embeddings/"),
+            "https://embedding.example/v1/embeddings",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()