params: Optional[Dict] = None,
) -> Any:
    """Generate text from the model."""
    params = params or {}
    headers = {
        "Content-Type": "application/json",
        "ApiKey": f"{self.nebula_api_key}",
    }
    body = {
        "prompt": {
            "instruction": instruction,
            "conversation": {"text": f"{conversation}"},
        }
    }
    # add params to body
    for key, value in params.items():
        body[key] = value
    # make request
    response = requests.post(url, headers=headers, json=body)
    if response.status_code != 200:
        raise Exception(
            f"Request failed with status code {response.status_code}"
            f" and message {response.text}"
        )
    return json.loads(response.text)
def _create_retry_decorator(llm: Nebula) -> Callable[[Any], Any]:
    min_seconds = 4
    max_seconds = 10
    # Wait 2^x * 1 second between each retry starting with
    # 4 seconds, then up to 10 seconds, then 10 seconds afterward
    max_retries = llm.max_retries if llm.max_retries is not None else 3
    return retry(
        reraise=True,
        stop=stop_after_attempt(max_retries),
        wait=wait_exponential(multiplier=1, min=min_seconds, max=max_seconds),
        retry=(
            retry_if_exception_type((RequestException, ConnectTimeout, ReadTimeout))
        ),
        before_sleep=before_sleep_log(logger, logging.WARNING),
    )
[docs]def completion_with_retry(llm: Nebula, **kwargs: Any) -> Any: