| | 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 |
| | |
