GenAI Integration Patterns & Asynchronous Workflows

                  ┌──────────────────────────────────────────────┐
                  │           Client Request Ingress             │
                  └──────────────────────┬───────────────────────┘
                                         │
        ┌────────────────────────┼────────────────────────┬────────────────────────┐
        ▼                        ▼                        ▼                        ▼
┌───────────────┐        ┌───────────────┐        ┌───────────────┐        ┌───────────────┐
│ Synchronous   │        │ Asynchronous  │        │ Event-Driven  │        │     Batch     │
│  API Gateway  │        │  Task Queue   │        │   Streaming   │        │  Processing   │
└───────────────┘        └───────────────┘        └───────────────┘        └───────────────┘

docker run -d -p 6379:6379 redis:alpine

pip install fastapi celery redis openai pydantic uvicorn

import os
import time
import logging
from typing import Optional, Dict, Any
from pydantic import BaseModel, Field
from fastapi import FastAPI, HTTPException, status
from celery import Celery
from openai import OpenAI
 
# 1. Initialize Logger and Configurations
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AsyncIntegration")
 
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
 
# 2. Celery Worker Configuration
# Broker: Enqueues tasks; Backend: Stores execution states/results
celery_app = Celery(
    "tasks",
    broker=REDIS_URL,
    backend=REDIS_URL
)
 
celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_time_limit=300,  # Hard timeout of 5 minutes
    task_soft_time_limit=240  # Soft timeout of 4 minutes
)
 
# 3. Pydantic Models for Input Validation
class JobSubmission(BaseModel):
    user_id: str = Field(..., description="Alphanumeric identifier for tracking costs")
    prompt: str = Field(..., min_length=5, description="Input prompt for LLM processing")
    model: str = Field("gpt-4o-mini", description="Target model name")
 
class JobStatusResponse(BaseModel):
    task_id: str = Field(..., description="Unique identifier for the asynchronous task")
    status: str = Field(..., description="Task state: PENDING, STARTED, SUCCESS, FAILURE")
    result: Optional[str] = Field(None, description="Model response payload if successful")
    error: Optional[str] = Field(None, description="Exception trace details if task failed")
    duration_seconds: Optional[float] = Field(None, description="Total execution time")
 
# 4. Celery Task Definition
@celery_app.task(bind=True, max_retries=3)
def process_llm_job(self, user_id: str, prompt: str, model: str) -> Dict[str, Any]:
    """
    Executes the LLM request inside a background worker thread.
    Retries automatically on transient API failures.
    """
    start_time = time.time()
    logger.info(f"Starting Celery task {self.request.id} for user {user_id}")
    
    # Initialize client inside the worker to avoid thread serialization issues
    client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "mock-key-if-testing"))
    
    try:
        # Standard synchronous OpenAI SDK call (executes inside worker process pool)
        response = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "You are a production assistant executing background analysis."},
                {"role": "user", "content": prompt}
            ],
            timeout=60.0
        )
        
        duration = time.time() - start_time
        logger.info(f"Task {self.request.id} completed successfully in {duration:.2f}s")
        
        return {
            "result": response.choices[0].message.content,
            "duration": round(duration, 2)
        }
        
    except Exception as exc:
        logger.error(f"Task {self.request.id} encountered exception: {str(exc)}")
        # Retry logic for transient network or rate-limiting errors
        try:
            self.retry(exc=exc, countdown=2 ** self.request.retries)
        except self.MaxRetriesExceededError:
            logger.error(f"Task {self.request.id} exceeded maximum retries. Marking as failed.")
            raise exc
 
# 5. FastAPI Gateway Endpoints
app = FastAPI(
    title="GenAI Asynchronous Gateway",
    description="Production-grade API endpoints for queuing and monitoring LLM tasks.",
    version="1.0.0"
)
 
@app.post(
    "/v1/jobs",
    status_code=status.HTTP_202_ACCEPTED,
    response_model=Dict[str, str],
    summary="Queue a new LLM task"
)
async def submit_job(payload: JobSubmission):
    """
    Submits a task to the Redis broker queue.
    Immediately returns a task ID to release the HTTP connection.
    """
    try:
        # Trigger Celery task asynchronously using .delay()
        task = process_llm_job.delay(payload.user_id, payload.prompt, payload.model)
        logger.info(f"Enqueued prompt for user {payload.user_id}. Task ID: {task.id}")
        return {"task_id": task.id}
    except Exception as e:
        logger.critical(f"Failed to submit task to broker: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Message broker offline. Unable to queue task."
        )
 
@app.get(
    "/v1/jobs/{task_id}",
    response_model=JobStatusResponse,
    summary="Get job execution status and results"
)
async def get_job_status(task_id: str):
    """
    Queries the Celery state backend to retrieve task progress.
    """
    # Query Redis state database
    result = celery_app.AsyncResult(task_id)
    
    response = JobStatusResponse(
        task_id=task_id,
        status=result.status
    )
    
    if result.status == "SUCCESS":
        task_output = result.get()
        response.result = task_output.get("result")
        response.duration_seconds = task_output.get("duration")
    elif result.status == "FAILURE":
        response.error = str(result.result)
        
    return response

                       ┌────────────────────────────┐
                       │   Worker Finishes Job      │
                       └─────────────┬──────────────┘
                                     ▼
                       ┌────────────────────────────┐
                       │   Query Client Webhook URL │
                       │    from Database Registries │
                       └─────────────┬──────────────┘
                                     ▼
                       ┌────────────────────────────┐
                       │  HTTP Client POST Request   │
                       │    to registered callback  │
                       └─────────────┬──────────────┘
                                     ▼
                       ┌────────────────────────────┐
                       │   Client Server Receives   │
                       │     Payload & Processes    │
                       └────────────────────────────┘