Celery Task Queue Integration with Flask for Async LLM Operations: Architecture and Implementation

The lmforge platform implements a Flask-aware Celery extension that wraps tasks in application context, enabling background workers to access Flask configurations, database sessions, and dependency-injected services while processing long-running LLM operations asynchronously.

The lmforge-end-to-end-llmops-platform-for-multi-model-agents repository implements a robust Celery task queue integration with Flask to handle asynchronous LLM operations without blocking HTTP requests. This architecture decouples client-facing API endpoints from computationally intensive workloads like document indexing and agent application creation, ensuring responsive user experiences while leveraging Flask's ecosystem within background workers.

Flask-Aware Celery Extension with Application Context

The core integration resides in api/internal/extension/celery_extension.py, which defines a custom FlaskTask class that subclasses celery.Task. This class overrides the __call__ method to wrap task execution within app.app_context(), ensuring that each worker process has access to Flask configurations, database connections, and other initialized extensions.

The extension creates a Celery instance using task_cls=FlaskTask, configures it from app.config["CELERY"], and registers it in app.extensions["celery"] for global access throughout the application.

Application Bootstrap and Initialization Order

During startup, the Http class in api/internal/server/http.py orchestrates extension initialization. The Celery extension initializes after database, Weaviate, migrations, logging, and Redis extensions to ensure that background tasks can leverage these resources.

The initialization sequence ensures that when workers spawn, they inherit a fully configured Flask application context with active database sessions and vector store connections ready for LLM operations.

Defining Asynchronous Tasks for LLM Operations

Background jobs are defined as standard Celery tasks using the @shared_task decorator, located in modules like api/internal/task/document_task.py and api/internal/task/app_task.py. These tasks follow a lazy import pattern to avoid circular dependencies and minimize worker startup time.

For example, the build_documents task in document_task.py imports the IndexingService from the Flask dependency injector only when the task executes:

@shared_task
def build_documents(document_ids: list[UUID]) -> None:
    from app.http.module import injector
    from internal.service.indexing_service import IndexingService
    indexing_service = injector.get(IndexingService)
    indexing_service.build_documents(document_ids)

Similarly, auto_create_app in app_task.py handles asynchronous agent application creation, allowing the HTTP layer to return immediately while the worker processes LLM configuration and vector store indexing.

Triggering Tasks from Business Logic

The integration demonstrates practical usage in api/internal/service/assistant_agent_service.py, where LangChain tools enqueue background work. The create_app tool accepts parameters from an LLM agent and immediately delegates to auto_create_app.delay(), returning a confirmation message to the agent without waiting for completion:

@tool("create_app", args_schema=CreateAppInput)
def create_app(name: str, description: str) -> str:
    auto_create_app.delay(name, description, account_id)
    return f"已调用后端异步任务创建Agent应用。\n应用名称: {name}\n应用描述: {description}"

This pattern ensures that conversational AI interfaces remain responsive while complex LLM operations execute in dedicated worker processes.

Worker Configuration and Broker Setup

Celery configuration resides in api/config/default_config.py, specifying Redis as both broker and result backend:

"CELERY_BROKER_DB": 1,
"CELERY_RESULT_BACKEND_DB": 1,

The configuration dictionary maps to standard Celery settings, allowing seamless switching to RabbitMQ or Amazon SQS by updating configuration values without modifying application code. Workers consume from the Redis database specified, ensuring isolated task queues for the LLMOps platform.

Summary

• Flask-aware task context: The custom FlaskTask class in celery_extension.py ensures every worker executes within the Flask application context, providing access to configurations and database sessions.
• Lazy-loaded dependencies: Tasks in document_task.py and app_task.py use runtime imports and dependency injection to avoid circular imports and keep workers lightweight.
• Async workflow integration: Services like AssistantAgentService trigger tasks via .delay() to offload LLM operations, maintaining responsive HTTP endpoints while background workers handle document indexing and app creation.
• Redis-backed configuration: Default settings in default_config.py use Redis for both broker and result backend, with configurable databases for environment isolation.

Frequently Asked Questions

How does the Flask application context get passed to Celery workers?

The FlaskTask class defined in api/internal/extension/celery_extension.py subclasses celery.Task and overrides the __call__ method. When a worker executes a task, this wrapper method enters the Flask application context using with app.app_context(): before running the actual task logic, ensuring database connections and configurations are available.

Why do Celery tasks use lazy imports instead of top-level imports?

Tasks in files like api/internal/task/document_task.py import services inside the function body rather than at the module level to prevent circular dependency issues during worker startup. Since the Flask application context initializes fully before tasks execute, lazy imports combined with the dependency injector (injector.get) ensure services are resolved only when needed within the active application context.

What happens when an LLM agent triggers a long-running operation?

When the Assistant Agent service in api/internal/service/assistant_agent_service.py receives a tool call requiring heavy processing (like creating an application), it immediately calls auto_create_app.delay() to enqueue the task. The HTTP response returns to the LLM agent instantly with a confirmation message, while a separate Celery worker process picks up the task from the Redis broker and executes the lengthy LLM configuration and indexing operations in the background.

Can the Celery broker be switched from Redis to RabbitMQ or SQS?

Yes. The configuration in api/config/default_config.py uses standard Celery configuration keys for the broker and result backend. By updating the CELERY dictionary in the Flask config—specifically the broker URL and result backend settings—you can switch to RabbitMQ, Amazon SQS, or other supported brokers without modifying the task definitions or extension code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}