tofl / functions.py
bai4578's picture
Update functions.py
6b3173f verified
Raw
History Blame Contribute Delete
2.66 kB
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:
# Make the API call
response = completion(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
# Return the generated content
return response.choices[0].message.content.strip()
except RateLimitError as e:
# Retry logic for rate limit errors
print("RateLimitError")
retries += 1
wait_time = base_delay * (2 ** (retries - 1)) # Exponential backoff
print(f"Rate limit hit. Retry {retries}/{max_retries} in {wait_time} seconds...")
time.sleep(wait_time)
except Exception as e:
# Handle other types of exceptions
print(f"An error occurred: {e}")
# Retry logic for rate limit errors
retries += 1
wait_time = base_delay * (2 ** (retries - 1)) # Exponential backoff
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" # default fallback
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()