When an authenticated user, create a Milvus database and upload a file with URL;
Enter OtherEmbedding.aencode() with base_url ;
class OtherEmbedding(BaseEmbeddingModel):
def __init__(self, **kwargs) -> None:
super().__init__(**kwargs)
self.headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
def build_payload(self, message: list[str] | str) -> dict:
return {"model": self.model, "input": message}
def encode(self, message: list[str] | str) -> list[list[float]]:
payload = self.build_payload(message)
try:
response = requests.post(self.base_url, json=payload, headers=self.headers, timeout=60)
response.raise_for_status()
result = response.json()
if not isinstance(result, dict) or "data" not in result:
raise ValueError(f"Other Embedding failed: Invalid response format {result}")
return [item["embedding"] for item in result["data"]]
except (requests.RequestException, json.JSONDecodeError) as e:
logger.error(f"Other Embedding request failed: {e}, {payload}")
raise ValueError(f"Other Embedding request failed: {e}")
async def aencode(self, message: list[str] | str) -> list[list[float]]:
payload = self.build_payload(message)
async with httpx.AsyncClient() as client:
try:
response = await client.post(self.base_url, json=payload, headers=self.headers, timeout=60)
response.raise_for_status()
result = response.json()
if not isinstance(result, dict) or "data" not in result:
raise ValueError(f"Other Embedding failed: Invalid response format {result}")
return [item["embedding"] for item in result["data"]]
except (httpx.RequestError, json.JSONDecodeError) as e:
raise ValueError(f"Other Embedding async request failed: {e}, {payload}, {self.base_url=}")
