Search⌘ K
AI Features

Getting Slow Calls Off the Request Path

Explore how to move slow AI model calls off the HTTP request path using FastAPI's background tasks. Learn to return immediate acknowledgments while processing data asynchronously, enabling better client experience and avoiding timeouts. Understand the limits of background tasks and when to consider more durable solutions like task queues for critical workloads.

Every endpoint built so far in this chapter has done the same thing: accept a request, wait for the model to respond, and only then send back a result. That is fine for a quick classification, but a call that takes several seconds, sometimes longer if a repair attempt is needed, leaves a caller sitting on an open HTTP connection the entire time. This lesson moves the slow part off the request path entirely.

Already configured setup code

The block below is carried over so this file runs entirely on its own: the schema a valid result must match, the prompt builder, the code fence stripper, a call_llm that runs with no API key using a mock response by default, and the pipeline tying it all together. None of this is new.

import json
from enum import Enum
from pydantic import BaseModel, Field, ValidationError
class Sentiment(str, Enum):
POSITIVE = "positive"
NEUTRAL = "neutral"
NEGATIVE = "negative"
class FeedbackCategory(str, Enum):
PRODUCT_QUALITY = "product_quality"
CUSTOMER_SERVICE = "customer_service"
PRICING = "pricing"
OTHER = "other"
class Confidence(str, Enum):
HIGH = "high"
LOW = "low"
class Feedback(BaseModel):
sentiment: Sentiment
category: FeedbackCategory
severity: int = Field(ge=1, le=5, default=1)
confidence: Confidence
summary: str
def build_feedback_prompt(feedback_text: str) -> str:
return (
"Analyze the customer feedback below. Respond with a JSON object "
"in exactly this shape, and nothing else:\n\n"
"{\n"
' "sentiment": <one of: "positive", "neutral", "negative">,\n'
' "category": <one of: "product_quality", "customer_service", '
'"pricing", "other">,\n'
' "severity": <integer from 1 to 5>,\n'
' "confidence": <one of: "high", "low">,\n'
' "summary": <string, one short sentence>\n'
"}\n\n"
f'Feedback: "{feedback_text}"'
)
def build_repair_prompt(original_prompt: str, bad_response: str, error: ValidationError) -> str:
return (
f"{original_prompt}\n\nYour previous response did not match the "
f"required format. Previous response:\n{bad_response}\n\n"
f"Validation error:\n{error}\n\nRespond again with only the "
"corrected JSON object."
)
def strip_code_fence(text: str) -> str:
stripped = text.strip()
if stripped.startswith("```"):
lines = stripped.split("\n")[1:]
if lines and lines[-1].strip() == "```":
lines = lines[:-1]
return "\n".join(lines).strip()
return stripped
# USE_MOCK is True by default, so this file runs immediately with no
# API key. Set it to False and fill in a real provider block below to
# use an actual model.
USE_MOCK = True
def call_llm(prompt: str) -> str:
if USE_MOCK:
return (
'{"sentiment": "negative", "category": "customer_service", '
'"severity": 4, "confidence": "high", '
'"summary": "Support took days to reply."}'
)
# from openai import OpenAI
# client = OpenAI(api_key="your-openai-api-key")
# response = client.chat.completions.create(
# model="your-chosen-openai-model",
# messages=[{"role": "user", "content": prompt}],
# )
# return response.choices[0].message.content
raise NotImplementedError("Fill in a real provider block, or set USE_MOCK back to True.")
def get_validated_feedback(feedback_text: str, max_attempts: int = 2) -> Feedback:
prompt = build_feedback_prompt(feedback_text)
current_prompt = prompt
for attempt in range(1, max_attempts + 1):
raw_response = call_llm(current_prompt)
cleaned = strip_code_fence(raw_response)
try:
parsed = json.loads(cleaned)
except json.JSONDecodeError:
continue
try:
return Feedback.model_validate(parsed)
except ValidationError as error:
current_prompt = build_repair_prompt(prompt, cleaned, error)
return Feedback(
sentiment=Sentiment.NEUTRAL,
category=FeedbackCategory.OTHER,
severity=3,
confidence=Confidence.LOW,
summary="Could not be automatically assessed.",
)

The shape of the problem

An HTTP request has a caller on the other end actively waiting for a response. The longer that wait, the worse the experience, and past a certain point, many clients and load balancers will simply time out and give up, even if the model call would have eventually succeeded. The fix is not to make the model faster, since we already covered in Making the Call from Python that latency is a real constraint we do not fully control. The fix is to stop making the caller wait for it at all.

Returning immediately, finishing the work after

...