i also got the dedicated service for embeddings!!
lass CustomAPIEmbedding(BaseEmbedding):
API_URL =
headers = {
"Accept": "application/json",
"Content-Type": "application/json"
}
def query(self, payload):
response = requests.post(self.API_URL, headers=self.headers, json=payload)
output = response.json()
return output
async def _aget_query_embedding(self, query: str) -> List[float]:
return self._get_query_embedding(query)
async def _aget_text_embedding(self, text: str) -> List[float]:
return self._get_text_embedding(text)
def _get_query_embedding(self, query: str) -> List[float]:
response = self.query({
"inputs": [query],
"parameters": {}
})
# Access the 'embeddings' key from the response
embeddings = response['embeddings'][0]
return embeddings
def _get_text_embedding(self, text: str) -> List[List[float]]:
response = self.query({
"inputs": [text],
"parameters": {}
})
# Access the 'embeddings' key from the response
embeddings = response['embeddings'] # Adjust based on actual API response structure
return embeddings
def _get_text_embeddings(self, texts: List[str]) -> List[List[float]]:
embeddings = []
for text in texts:
response = self.query({
"inputs": [text],
"parameters": {}
})
try:
# Adjust this line based on the actual structure of the API response
embeddings_list = response['embeddings'] # Hypothetical correct key
embeddings.append(embeddings_list[0]) # Assuming the first element is what you need
except KeyError:
print(f"Unexpected response structure: {response}")
raise
return embeddings