Perform chat completion :param model: the model object :param messages: a list of chat completion messages :param configs: the model configurations :param function_call: the function call parameter :param functions: the list of functions :return: the chat completion data
(
model: Model,
messages: List[Dict],
configs: Dict,
function_call: Optional[str] = None,
functions: Optional[List[Dict]] = None,
)
| 42 | |
| 43 | |
| 44 | async def chat_completion( |
| 45 | model: Model, |
| 46 | messages: List[Dict], |
| 47 | configs: Dict, |
| 48 | function_call: Optional[str] = None, |
| 49 | functions: Optional[List[Dict]] = None, |
| 50 | ) -> Dict: |
| 51 | """ |
| 52 | Perform chat completion |
| 53 | :param model: the model object |
| 54 | :param messages: a list of chat completion messages |
| 55 | :param configs: the model configurations |
| 56 | :param function_call: the function call parameter |
| 57 | :param functions: the list of functions |
| 58 | :return: the chat completion data |
| 59 | """ |
| 60 | |
| 61 | model_schema_id = model.model_schema_id |
| 62 | provider_model_id, properties = await __validate_model(model) |
| 63 | request_url = f"{CONFIG.TASKINGAI_INFERENCE_URL}/v1/chat_completion" |
| 64 | payload = { |
| 65 | "model_schema_id": model_schema_id, |
| 66 | "provider_model_id": provider_model_id, |
| 67 | "messages": messages, # List of message dicts |
| 68 | "encrypted_credentials": model.encrypted_credentials, |
| 69 | "properties": properties, |
| 70 | "configs": configs, |
| 71 | "function_call": function_call, |
| 72 | "functions": functions, |
| 73 | "stream": False, |
| 74 | } |
| 75 | |
| 76 | async with aiohttp.ClientSession() as session: |
| 77 | response = await session.post(request_url, json=payload) |
| 78 | response_wrapper = ResponseWrapper(response.status, await response.json()) |
| 79 | check_http_error(response_wrapper) |
| 80 | return response_wrapper.json()["data"] |
| 81 | |
| 82 | |
| 83 | async def stream_chat_completion( |
no test coverage detected