|
|
| def generate_text(prompt, model="gpt-4o-mini", system_prompt="You are a helpful assistant.", |
| temperature=0.7, max_tokens=4500, max_retries=4, base_delay=10): |
| """ |
| Generates text using litellm's completion API with retry handling for rate limit errors. |
| |
| Args: |
| prompt (str): The user input prompt. |
| model (str): The model to use for text generation. |
| system_prompt (str): The system message for context. |
| temperature (float): Sampling temperature. |
| max_tokens (int): Maximum number of tokens to generate. |
| max_retries (int): Maximum retry attempts for rate limit errors. |
| base_delay (int): Base delay in seconds for exponential backoff. |
| |
| Returns: |
| str or None: The generated text, or None if retries fail. |
| """ |
| messages = [ |
| {"role": "system", "content": system_prompt}, |
| {"role": "user", "content": prompt} |
| ] |
|
|
| retries = 0 |
|
|
| while retries <= max_retries: |
| try: |
| |
| response = completion( |
| model=model, |
| messages=messages, |
| temperature=temperature, |
| max_tokens=max_tokens |
| ) |
| |
| return response.choices[0].message.content.strip() |
|
|
| except RateLimitError as e: |
| |
| print("RateLimitError") |
| retries += 1 |
| wait_time = base_delay * (2 ** (retries - 1)) |
| print(f"Rate limit hit. Retry {retries}/{max_retries} in {wait_time} seconds...") |
| time.sleep(wait_time) |
|
|
| except Exception as e: |
| |
| print(f"An error occurred: {e}") |
| |
| retries += 1 |
| wait_time = base_delay * (2 ** (retries - 1)) |
| print(f"Rate limit hit. Retry {retries}/{max_retries} in {wait_time} seconds...") |
| time.sleep(wait_time) |
| |
|
|
| |
|
|
| import mimetypes |
|
|
| def transcribe_with_deepgram(api_key, audio_path): |
| endpoint = "https://api.deepgram.com/v1/listen?model=nova-3&smart_format=true" |
| mime_type, _ = mimetypes.guess_type(audio_path) |
| if mime_type is None: |
| mime_type = "audio/wav" |
| |
| headers = { |
| "Authorization": f"Token {api_key}", |
| "Content-Type": mime_type |
| } |
|
|
| with open(audio_path, "rb") as f: |
| response = requests.post(endpoint, headers=headers, data=f) |
|
|
| response.raise_for_status() |
| return response.json() |
|
|
|
|