TimeoutError
asyncio.exceptions.TimeoutError
Stack trace
Traceback (most recent call last):
File "flow.py", line 42, in run_ai_task
response = await client.chat.completions.create(...)
File "asyncio/tasks.py", line 481, in wait_for
raise asyncio.exceptions.TimeoutError
asyncio.exceptions.TimeoutError Why it happens
Prefect enforces time limits on tasks to prevent runaway executions. When an AI model call, such as an OpenAI chat completion, takes longer than the specified timeout, Prefect raises a TimeoutError to abort the task. This often happens due to slow network, large prompt sizes, or model latency.
Detection
Monitor Prefect task logs for asyncio.exceptions.TimeoutError and track task durations exceeding the configured timeout threshold before failure.
Causes & fixes
The AI model call exceeds the Prefect task timeout limit due to slow response or large input.
Increase the task timeout parameter in Prefect's task decorator or flow run configuration to allow more time for the AI call.
Network latency or intermittent connectivity slows down the AI API response.
Implement retry logic with exponential backoff around the AI call to handle transient network delays.
The AI prompt or request is too large or complex, causing longer processing times.
Optimize and shorten the prompt or split the request into smaller chunks to reduce model processing time.
Code: broken vs fixed
from openai import OpenAI
import prefect
@prefect.task(timeout_seconds=10)
async def run_ai_task():
client = OpenAI(api_key="sk-...") # Hardcoded key
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Hello"}]
) # This line triggers TimeoutError
return responseimport os
from openai import OpenAI
import prefect
@prefect.task(timeout_seconds=60) # Increased timeout
async def run_ai_task():
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"]) # Use env var for key
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Hello"}]
) # Fixed: longer timeout prevents error
return response
if __name__ == "__main__":
from prefect import flow
@flow
def main_flow():
result = run_ai_task()
print(result)
main_flow()Workaround
Wrap the AI call in a try/except block catching asyncio.exceptions.TimeoutError, then retry the call with a longer timeout or fallback logic.
Prevention
Configure Prefect task timeouts based on expected AI model latency and implement retries with backoff to handle transient delays, ensuring stable AI workflow execution.