requestBody = {
        "model": model,
        "prompt": prompt,
        "max_tokens": max_tokens,
        "temperature": temperature,
        "top_p": top_p,
    }
    requestHeaders = {
        "Authorization": f"Bearer {api_key}",
        "Accept": "application/json",
        "Content-Type": "application/json",
    }
    response = requests.post(requestUrl, headers=requestHeaders, json=requestBody)
    return response.text
[docs]def completion_with_retry(
    llm: Union[BaseFireworks, FireworksChat], **kwargs: Any
) -> Any:
    """Use tenacity to retry the completion call."""
    if "prompt" not in kwargs.keys():
        answers = []
        for i in range(len(kwargs["messages"])):
            result = kwargs["messages"][i]["content"]
            result = execute(
                result,
                kwargs["model"],
                llm.fireworks_api_key,
                llm.max_tokens,
                llm.temperature,
                llm.top_p,
            )
            curr_string = json.loads(result)["choices"][0]["text"]
            answers.append(curr_string)
    else:
        answers = []
        for i in range(len(kwargs["prompt"])):
            result = kwargs["prompt"][i]
            result = execute(
                result,
                kwargs["model"],
                llm.fireworks_api_key,
                llm.max_tokens,
                llm.temperature,
                llm.top_p,
            )
            curr_string = json.loads(result)["choices"][0]["text"]
            answers.append(curr_string)
    return answers
[docs]async def acompletion_with_retry(
    llm: Union[BaseFireworks, FireworksChat], **kwargs: Any