| from typing import List |
| import openai |
| from tenacity import retry, stop_after_attempt, wait_random_exponential |
|
|
| from models.Base import BaseModel |
|
|
|
|
| class OpenAIModel(BaseModel): |
| def __init__(self, |
| model_id="GPT4o", |
| model_api_version='2024-06-01', |
| api_key=None): |
| assert api_key is not None, "no api key is provided." |
| self.model_id = model_id |
| self.model_api_version = model_api_version |
|
|
| url = 'https://llm-api.amd.com' |
| headers = { |
| 'Ocp-Apim-Subscription-Key': api_key |
| } |
| model_api_version = '2024-06-01' |
| |
|
|
| self.client = openai.AzureOpenAI( |
| api_key='dummy', |
| api_version=self.model_api_version, |
| base_url=url, |
| default_headers=headers |
| ) |
| self.client.base_url = '{0}/openai/deployments/{1}'.format(url, self.model_id) |
| |
| @retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(5)) |
| def generate(self, |
| messages: List, |
| temperature=0, |
| presence_penalty=0, |
| frequency_penalty=0, |
| max_tokens=5000) -> str: |
| response = self.client.chat.completions.create( |
| model=self.model_id, |
| messages=messages, |
| temperature=temperature, |
| n=1, |
| stream=False, |
| stop=None, |
| max_tokens=max_tokens, |
| presence_penalty=presence_penalty, |
| frequency_penalty=frequency_penalty, |
| logit_bias=None, |
| user=None |
| ) |
| |
| |
| if not response or not hasattr(response, 'choices') or len(response.choices) == 0: |
| raise ValueError("No response choices returned from the API.") |
|
|
| return response.choices[0].message.content |
| |
