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 jsonfrom enum import Enumfrom pydantic import BaseModel, Field, ValidationErrorclass 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: Sentimentcategory: FeedbackCategoryseverity: int = Field(ge=1, le=5, default=1)confidence: Confidencesummary: strdef 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 = Truedef 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.contentraise 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 = promptfor 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:continuetry: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.