[docs]async def acompletion_with_retry(
    llm: Fireworks,
    use_retry: bool,
    *,
    run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
    **kwargs: Any,
) -> Any:
    """Use tenacity to retry the completion call."""
    import fireworks.client
    retry_decorator = _create_retry_decorator(llm, run_manager=run_manager)
    @conditional_decorator(use_retry, retry_decorator)
    async def _completion_with_retry(**kwargs: Any) -> Any:
        return await fireworks.client.Completion.acreate(
            **kwargs,
        )
    return await _completion_with_retry(**kwargs)
[docs]def completion_with_retry_batching(
    llm: Fireworks,
    use_retry: bool,
    *,
    run_manager: Optional[CallbackManagerForLLMRun] = None,
    **kwargs: Any,
) -> Any:
    """Use tenacity to retry the completion call."""
    import fireworks.client
    prompt = kwargs["prompt"]
    del kwargs["prompt"]
    retry_decorator = _create_retry_decorator(llm, run_manager=run_manager)
    @conditional_decorator(use_retry, retry_decorator)
    def _completion_with_retry(prompt: str) -> Any:
        return fireworks.client.Completion.create(**kwargs, prompt=prompt)
    def batch_sync_run() -> List:
        with ThreadPoolExecutor() as executor:
            results = list(executor.map(_completion_with_retry, prompt))
        return results
    return batch_sync_run()
[docs]async def acompletion_with_retry_batching(
    llm: Fireworks,
    use_retry: bool,
    *,
    run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
    **kwargs: Any,