Issue with bad requests and API timeout in Embedding calls

[saicharans]

We have been using “text-embedding-ada-002” model for over 1.5 year now, recently we are experiencing many time outs and bad request in the embedding calls.
Can anyone let us know how to resolve this
thank you.

[_j]

“Resolve this”? Probably the first thing, that might be the reason why you post here:

1. report the issue to OpenAI

If you were expecting advice from fellow users that frequent the forum, to make your code more robust, then you just need a strategy for “handle”.

First, you can see if the failing inputs follow a particular pattern in what you are sending that might no longer be performed correctly, such as length of text, length of lists sent, etc, and re-evaluate how to submit that data in a way more tolerated.

Then, of course, reduce the timeout parameter, and add some retries.

Then, the ultimate goal might be to directly replace the API embeddings call with a more robust retrying method that automatically launches some more attempts – while the first still runs. Thus you get first-to-reply.

An example of doing this in Python might be useful for all reading.

To implement a mechanism that adds additional API calls if the initial request doesn’t respond within an expected timeframe? Here’s how you might approach it:

• Set a Threshold Time: Determine a reasonable time by which most successful API calls complete. For example, if the majority of your calls respond within 3 seconds, you can use this as your threshold.

• Monitor the API Call: After initiating an API call, monitor its progress. If the call hasn’t completed within the 3-second threshold, this might indicate a potential timeout or failure.

• Initiate Additional Requests: If the initial call is taking too long, you can start additional identical API requests. This employs a first-to-respond technique, where you accept the earliest successful response and disregard the others.

• Implement a Failure Strategy: To avoid unnecessary resource usage, set a limit on how many additional requests you will send or a maximum time after which you cease trying. This ensures that your system doesn’t persist indefinitely in the face of persistent failures.

Since the failure rate could be relatively low (around 5% perhaps?), launching multiple parallel calls for every request might not be the most efficient approach due to the associated costs of API calls. Therefore, initially sending one API call and only initiating additional requests when there’s a delay can strike a balance between reliability and resource management.

By adopting this strategy, you can enhance the resilience of your application against occasional timeouts and failures, while keeping additional costs in check.

---

To implement a fault-tolerant API calling strategy as described, using Python and its methods as a presentation format, we can create a new asynchronous function robust_embeddings_create that wraps around the original client.embeddings.create method. This function will handle launching additional API calls if the initial call doesn’t respond within a specified threshold (3 seconds by default), and it will manage the logic for maximum retries and total allowable time.

Here’s how we can approach this:

1. Use Asynchronous Tasks: Leverage Python’s asyncio library to manage asynchronous tasks. This allows us to start multiple API calls concurrently and await the first one to complete.

2. Implement Timing Logic: Keep track of when to start new API calls based on your thresholds (e.g., after 3 seconds if no response). Also, monitor the total elapsed time to ensure we don’t exceed the maximum allowed time (e.g., 20 seconds).

3. Manage Task Completion and Cancellation: When any API call completes successfully, use its result and cancel all other pending tasks to avoid unnecessary resource usage.

4. Handle Exceptions and Failures: If all retries fail within the allowed time, raise an exception or handle the error as per your requirements.

Below is the implementation of a robust_embeddings_create function, basically something that could drop in at a similar point in code that uses SDK embeddings and either your string or your list of strings input that embeddings can accept:

import openai
import asyncio
import time
from typing import Any, List, Union

async def robust_embeddings_create(
    client,
    model: str,
    input: Union[str, List[str]],
    encoding_format: str = "float",
    initial_delay: float = 3.0,
    max_retries: int = 5,
    max_total_time: float = 20.0
) -> Any:
    """
    Create embeddings with fault tolerance by launching additional API calls if the initial call is slow.

    Args:
        client: The AsyncOpenAI client instance.
        model (str): The model to use for generating embeddings.
        input (Union[str, List[str]]): The input string or list of strings.
        encoding_format (str, optional): Encoding format for the embeddings. Defaults to "float".
        initial_delay (float, optional): Time in seconds to wait before starting an additional API call. Defaults to 3.0.
        max_retries (int, optional): Maximum number of API call retries. Defaults to 5.
        max_total_time (float, optional): Maximum total time in seconds to wait for responses. Defaults to 20.0.

    Returns:
        Any: The embeddings result from the first successful API call.

    Raises:
        Exception: If embeddings creation fails after exceeding max retries or total time.
    """
    start_time = time.time()
    tasks = set()
    retry_intervals = [initial_delay] * (max_retries - 1)  # Intervals between retries
    scheduled_times = [start_time + sum(retry_intervals[:i + 1]) for i in range(len(retry_intervals))]
    total_deadline = start_time + max_total_time

    # Start the initial API call task
    initial_task = asyncio.create_task(client.embeddings.create(
        model=model,
        input=input,
        encoding_format=encoding_format
    ))
    tasks.add(initial_task)

    while True:
        now = time.time()
        if scheduled_times:
            next_retry_time = scheduled_times[0]
            time_until_next_retry = max(next_retry_time - now, 0)
        else:
            time_until_next_retry = None

        time_until_total_deadline = total_deadline - now
        if time_until_total_deadline <= 0:
            # Exceeded maximum total time
            for task in tasks:
                task.cancel()
            raise Exception("Embedding creation failed after exceeding the maximum total time.")

        # Determine the timeout for asyncio.wait
        if time_until_next_retry is None:
            timeout = time_until_total_deadline
        else:
            timeout = min(time_until_next_retry, time_until_total_deadline)
            if timeout <= 0:
                timeout = 0  # Ensure non-negative timeout

        # Wait for any task to complete or for the timeout
        done, pending = await asyncio.wait(tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)

        if done:
            # A task has completed
            for task in done:
                try:
                    result = task.result()
                    # Cancel all other pending tasks
                    for p in pending:
                        p.cancel()
                    return result
                except Exception:
                    # Task failed; remove it from the set and continue
                    tasks.discard(task)
        else:
            # Timeout occurred; check if it's time to start a new retry
            if scheduled_times and now >= scheduled_times[0]:
                # Start a new API call task
                print("-- additional embeddings API call added --") 
                new_task = asyncio.create_task(client.embeddings.create(
                    model=model,
                    input=input,
                    encoding_format=encoding_format
                ))
                tasks.add(new_task)
                scheduled_times.pop(0)
            elif not tasks:
                # No tasks are pending; all tasks have failed
                raise Exception("All embeddings tasks have failed.")

# Usage example
# Replace the list with your actual input
async def main():
    client = openai.AsyncOpenAI()
    string_or_list_of_strings = ["Big bugs bite", "terrorism"]
    try:
        response_obj = await robust_embeddings_create(
            client=client,
            model="text-embedding-ada-002",
            input=string_or_list_of_strings,
            encoding_format="float"
        )
        print("Embedding creation succeeded:",
              response_obj.data[0].embedding[:5])
    except Exception as e:
        print("Embedding creation failed:", str(e))

# To run the asynchronous main function
asyncio.run(main())

Explanation of the Code:

• Function Parameters:

  • initial_delay: Time to wait before starting an additional API call if the previous hasn’t completed (default is 3 seconds).
  • max_retries: Maximum number of API calls to attempt (initial call plus retries, default is 5).
  • max_total_time: Maximum total time to wait before giving up (default is 20 seconds).

• Task Management:

  • We start by creating the initial API call as an asynchronous task.
  • We maintain a set tasks of all ongoing tasks.
  • The scheduled_times list contains the times at which new retries should be initiated.

• Main Loop:

  • The loop continues until a result is returned or an exception is raised.
  • We calculate the timeout for asyncio.wait based on the next scheduled retry and the total deadline.
  • If any task completes, we attempt to get its result.
    • On success, we cancel all other tasks and return the result.
    • On failure, we remove the failed task from the set and continue.
  • If the timeout expires, we check if it’s time to start a new retry.
    • If so, we initiate a new API call and add it to the tasks set.
  • If we exceed the maximum total time or all tasks have failed, we raise an exception.

• Error Handling:

  • If all retries fail or we exceed the maximum allowed time, we raise an exception indicating the failure.

• Usage:

  • Replace string_or_list_of_strings with your actual input.
  • The main function demonstrates how to use the robust_embeddings_create function within an asynchronous context.

This implementation ensures that:

• Efficiency: Additional API calls are only made if necessary, preserving resources and costs.
• Responsiveness: The first successful response is used, minimizing wait times for the user.
• Robustness: The strategy handles intermittent failures and network issues gracefully, improving the reliability of your application.

Notes:

• There are two print() for your own viewing that may be places to log instead.
• This is asyncio, so not a complete drop-in for every code base,
• Adjust the parameters (initial_delay, max_retries, max_total_time) as needed based on your specific use case and performance considerations.
• Ensure proper exception handling in your production code to capture and log errors better, that can help you see if tuning is needed.

---

More analysis of the coded solution and its operation

The provided Python code implements a fault-tolerant mechanism for creating embeddings using an asynchronous OpenAI API client. The function robust_embeddings_create is designed to handle potential delays or failures in API calls by launching additional concurrent API requests if the initial call is slow or unresponsive. Below is a detailed analysis of the algorithms and logic used in the code, along with examples demonstrating its operation under various adverse conditions.

---

Function Overview

The function aims to achieve the following:

• Fault Tolerance: Handle slow or failed API calls by initiating additional calls after specified delays.
• Concurrency Management: Run multiple API call tasks concurrently and return the result of the first successful one.
• Time Constraints: Limit the total time spent waiting for API calls to complete and control the intervals between retries.

---

Detailed Logic Examination

Initialization

start_time = time.time()
tasks = set()
retry_intervals = [initial_delay] * (max_retries - 1)
scheduled_times = [start_time + sum(retry_intervals[:i + 1]) for i in range(len(retry_intervals))]
total_deadline = start_time + max_total_time

• start_time: Records the time at which the function starts execution.
• tasks: A set to keep track of all ongoing asyncio tasks representing API calls.
• retry_intervals: A list of intervals between retries. Each interval is set to initial_delay.
• scheduled_times: A list of absolute times when new retries should be initiated.
• total_deadline: The absolute time by which the function should return a result or fail.

Starting the Initial API Call

initial_task = asyncio.create_task(client.embeddings.create(
    model=model,
    input=input,
    encoding_format=encoding_format
))
tasks.add(initial_task)

• Initiates the first API call as an asyncio task and adds it to the tasks set.

Main Event Loop

while True:
    # Calculate time-related variables
    # Determine the timeout for asyncio.wait
    # Wait for any task to complete or for the timeout
    # Handle completed tasks or initiate retries

• Purpose: Continuously manage tasks, handle task completions, initiate retries, and enforce time constraints.

Time Calculations

now = time.time()
if scheduled_times:
    next_retry_time = scheduled_times[0]
    time_until_next_retry = max(next_retry_time - now, 0)
else:
    time_until_next_retry = None

time_until_total_deadline = total_deadline - now
if time_until_total_deadline <= 0:
    # Handle total time exceeded

• time_until_next_retry: Time remaining before the next scheduled retry.
• time_until_total_deadline: Time remaining before the total allowed time expires.

Determining Wait Timeout

if time_until_next_retry is None:
    timeout = time_until_total_deadline
else:
    timeout = min(time_until_next_retry, time_until_total_deadline)
    if timeout <= 0:
        timeout = 0  # Ensure non-negative timeout

• timeout: The maximum time asyncio.wait should wait for tasks to complete.

Waiting for Tasks

done, pending = await asyncio.wait(tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)

• done: A set of tasks that have completed.
• pending: A set of tasks that are still running.

Handling Completed Tasks

if done:
    for task in done:
        try:
            result = task.result()
            # Cancel all other pending tasks
            for p in pending:
                p.cancel()
            return result
        except Exception:
            # Task failed; remove it from the set and continue
            tasks.discard(task)

• Success: If a task completes successfully, cancel all pending tasks and return the result.
• Failure: If a task raises an exception, remove it from tasks.

Initiating Retries

else:
    # Timeout occurred; check if it's time to start a new retry
    if scheduled_times and now >= scheduled_times[0]:
        # Start a new API call task
        print("-- additional embeddings API call added --") 
        new_task = asyncio.create_task(client.embeddings.create(
            model=model,
            input=input,
            encoding_format=encoding_format
        ))
        tasks.add(new_task)
        scheduled_times.pop(0)
    elif not tasks:
        # No tasks are pending; all tasks have failed
        raise Exception("All embeddings tasks have failed.")

• Retry Condition: If the timeout occurred and it’s time for a retry, initiate a new API call.
• No Pending Tasks: If all tasks have been completed and none are pending, raise an exception.

---

Examples of Operation Against Adversity

Example 1: Initial API Call Succeeds Quickly

• Conditions: The initial API call completes successfully within initial_delay.
• Flow:
  • The initial task completes before any retries are initiated.
  • The result is returned immediately.
  • No additional tasks are created.
• Outcome:
  • Efficient completion with minimal resource usage.
  • Quick response to the caller.

Example 2: Initial API Call Is Slow, Second Call Succeeds

• Conditions: The initial API call is slow and does not complete before initial_delay.
• Flow:
  • After initial_delay, a second API call is initiated.
  • The second API call completes successfully.
  • Both the initial and second tasks are running concurrently.
  • Upon completion, pending tasks are canceled.
• Outcome:
  • Fault tolerance ensures timely result despite the initial delay.
  • Additional resource usage due to the concurrent task.

Example 3: Multiple Retries Needed Before Success

• Conditions: Initial and subsequent API calls fail due to transient errors.
• Flow:
  • Each API call raises an exception upon completion.
  • Retries are initiated at each initial_delay interval.
  • After several failures, a subsequent call succeeds within max_total_time.
  • All pending tasks are canceled upon success.
• Outcome:
  • The function overcomes multiple failures to deliver a result.
  • Demonstrates resilience against intermittent API issues.

Example 4: All API Calls Fail Before Total Deadline

• Conditions: Every API call fails, and retries are exhausted.
• Flow:
  • All tasks raise exceptions.
  • After the last retry, tasks becomes empty.
  • The function raises an exception indicating failure.
• Outcome:
  • The function correctly identifies that all attempts have failed.
  • Returns a clear error to the caller.

Example 5: No Tasks Complete Before Total Deadline

• Conditions: API calls hang and do not complete before max_total_time.
• Flow:
  • Tasks remain pending without completion.
  • Total time exceeds max_total_time.
  • All tasks are canceled.
  • The function raises an exception due to timeout.
• Outcome:
  • Prevents indefinite waiting by enforcing a time limit.
  • Provides a timeout mechanism for unresponsive services.

---

Potential Issues and Recommendations

1. Use of time.time() for Timing

• Issue: time.time() may be affected by system clock changes (e.g., NTP adjustments).
• Recommendation: Use time.monotonic() for measuring elapsed time, which is immune to system clock updates.

Implementation:

start_time = time.monotonic()
# ... replace all instances of time.time() with time.monotonic()

2. Exception Handling Granularity

• Issue: All exceptions are treated equally; specific handling for different exceptions is lacking.
• Recommendation: Differentiate between recoverable and non-recoverable exceptions (e.g., network errors vs. authentication errors) to decide whether to retry or fail immediately.

Implementation Example:

except SomeRecoverableException:
    # Keep the task in the set to allow for retries
    pass
except Exception as e:
    # Remove task and perhaps fail fast for critical errors
    tasks.discard(task)
    if isinstance(e, CriticalException):
        # Cancel all tasks and raise immediately
        for t in tasks:
            t.cancel()
        raise

3. Rate Limiting and API Throttling

• Issue: Concurrent retries may violate API rate limits or lead to throttling.
• Recommendation: Implement exponential backoff for retries and respect the API’s rate limit guidelines.

Implementation Example:

retry_intervals = [initial_delay * (2 ** i) for i in range(max_retries - 1)]

4. Task Cancellation Handling

• Issue: Canceling tasks may not immediately release resources; proper cleanup is necessary.
• Recommendation: Ensure that the API client properly handles cancellation and that resources such as network connections are released.

Implementation Example:

• Confirm that the OpenAI client supports cancellation and clean-up of pending requests.

5. Logging and Monitoring

• Issue: Lack of detailed logging makes debugging difficult.
• Recommendation: Add logging statements to track the status of tasks, retries, and exceptions.

Implementation Example:

import logging

logger = logging.getLogger(__name__)

# Add logging in strategic places
logger.debug("Starting initial API call.")
# ... inside loops and exception handlers
logger.warning("API call failed with exception: %s", str(e))

---

Final Thoughts

The code provides a robust mechanism for making fault-tolerant API calls to create embeddings. By launching additional API requests when the initial call is slow or fails, it increases the chances of obtaining a result within acceptable time frames.

Key strengths of the code include:

• Concurrency Management: Efficiently manages multiple asynchronous tasks without overwhelming resources.
• Time Constraints: Ensures that operations complete within a specified total time, preventing indefinite hangs.
• Simplicity: The logic is straightforward, making it easier to understand and maintain.

Areas for improvement involve refining timing mechanisms, enhancing exception handling, and ensuring compliance with API usage policies. By addressing these considerations, the function can become more reliable and efficient in various operational scenarios.

---

Summary

The robust_embeddings_create function is designed to handle adversity by:

• Initiating retries when API calls are slow or fail.
• Using concurrent tasks to minimize overall wait time.
• Enforcing a maximum total time to prevent indefinite operation.
• Canceling unnecessary tasks to conserve resources.

Through careful scheduling and management of asynchronous tasks, the function aims to provide a reliable way to create embeddings even under adverse conditions such as network delays, API timeouts, or intermittent failures.

My response greatly enhanced by AI and AI enhanced by my hand, of course.

[merefield]

Relying on any API call working first time is asking for trouble.

Always wrap these things in a job that can automatically retry when there is any risk of the service being unreliable, slow or temporarily unavailable.

I’m coding in Ruby on Rails with Sidekiq, which I use for such jobs.