(
model: Model,
encrypted_credentials: Dict,
input_text_list: List[str],
input_type: Optional[str],
)
| 34 | |
| 35 | # For POST /v1/text_embedding |
| 36 | async def text_embedding( |
| 37 | model: Model, |
| 38 | encrypted_credentials: Dict, |
| 39 | input_text_list: List[str], |
| 40 | input_type: Optional[str], |
| 41 | ) -> ResponseWrapper: |
| 42 | model_schema_id = model.model_schema_id |
| 43 | provider_model_id, properties = await __validate_model(model) |
| 44 | request_url = f"{CONFIG.TASKINGAI_INFERENCE_URL}/v1/text_embedding" |
| 45 | payload = { |
| 46 | "model_schema_id": model_schema_id, |
| 47 | "provider_model_id": provider_model_id, |
| 48 | "encrypted_credentials": encrypted_credentials, |
| 49 | "properties": properties, |
| 50 | "input": input_text_list, |
| 51 | } |
| 52 | if input_type: |
| 53 | payload["input_type"] = input_type |
| 54 | |
| 55 | async with aiohttp.ClientSession() as session: |
| 56 | response = await session.post(request_url, json=payload) |
| 57 | return ResponseWrapper(response.status, await response.json()) |
no test coverage detected