How to Configure Celery Beat for Scheduled LLM Inference and Batch Processing Tasks

Add a beat_schedule dictionary to the CELERY configuration in api/config/config.py, define periodic tasks using @shared_task decorators in the task modules, and launch the worker with the -B flag to enable Celery Beat scheduling for automated LLM inference and batch processing pipelines.

The LMForge end-to-end LLMOps platform for multi-model agents relies on Celery for asynchronous background work. To automate recurring workloads such as nightly document summarization or hourly batch inference queues, you must configure Celery Beat for scheduled LLM inference and batch processing tasks by extending the existing Flask-Celery integration.

Prerequisites: The Existing Celery Infrastructure

Before enabling the scheduler, understand that LMForge already initializes a Flask-aware Celery instance in api/internal/extension/celery_extension.py. This extension creates a Celery object using task_cls=FlaskTask and registers it under app.extensions["celery"]. The configuration builder in api/config/config.py populates broker URL and result backend settings from environment variables, providing the foundation for adding Beat configuration.

Step 1: Configure the Beat Schedule in Flask Config

The core of scheduling lies in the CELERY dictionary within api/config/config.py.

Adding Periodic Tasks to beat_schedule

Extend the configuration dictionary with a beat_schedule key containing named entries for each recurring job. Each entry specifies the task import path, schedule interval, and optional arguments.

# api/config/config.py

from celery.schedules import crontab

self.CELERY = {
    "broker_url": _get_env("CELERY_BROKER_URL"),
    "result_backend": _get_env("CELERY_RESULT_BACKEND"),
    "task_ignore_result": _get_bool_env("CELERY_TASK_IGNORE_RESULT"),
    "result_expires": int(_get_env("CELERY_RESULT_EXPIRES")),
    "broker_connection_retry_on_startup": _get_bool_env(
        "CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP"
    ),
    "beat_schedule": {
        "hourly-llm-inference": {
            "task": "api.internal.task.demo_task.run_scheduled_inference",
            "schedule": 3600,  # seconds

            "args": [],
        },
        "nightly-index-rebuild": {
            "task": "api.internal.task.document_task.rebuild_index",
            "schedule": crontab(minute=30, hour=2),
            "args": [],
        },
    },
}

Step 2: Implement Scheduled Task Functions

Create or reuse @shared_task decorated functions in the task modules. The platform already contains task files such as api/internal/task/demo_task.py and api/internal/task/document_task.py where you can add periodic workload handlers.

Batch LLM Inference Task

Implement a task that processes pending inference requests from the database on an hourly schedule.

# api/internal/task/demo_task.py

from celery import shared_task
from datetime import datetime

@shared_task
def run_scheduled_inference():
    """
    Periodic task that pulls pending LLM inference requests
    from the database and dispatches them to the inference service.
    """
    from app.http.module import injector
    from internal.service.inference_service import InferenceService

    inference_srv = injector.get(InferenceService)
    inference_srv.process_pending_requests()
    print(f"[Scheduled] LLM inference batch run at {datetime.utcnow()}")

Vector Index Rebuild Task

For weekly or nightly batch processing of document indexes, add a task that rebuilds vector stores.

# api/internal/task/document_task.py

from celery import shared_task

@shared_task
def rebuild_index():
    """Re-create the vector index for all datasets – ideal for nightly runs."""
    from app.http.app import injector
    from internal.service.indexing_service import IndexingService

    indexing_srv = injector.get(IndexingService)
    indexing_srv.rebuild_all_indexes()

Step 3: Start Workers with Beat Enabled

After configuring schedules and tasks, launch the Celery processes. You can run Beat inside the worker process for development or as a separate service for production.

Development Mode (Worker + Beat Combined)

Use the -B flag to start the scheduler alongside the worker:

celery -A api.internal.extension.celery_extension.celery_app worker -B --loglevel=INFO

Production Deployment (Separate Processes)

For better reliability and scalability, run the Beat scheduler as a dedicated service separate from workers:

# Terminal 1: Worker

celery -A api.internal.extension.celery_extension.celery_app worker --loglevel=INFO

# Terminal 2: Beat scheduler

celery -A api.internal.extension.celery_extension.celery_app beat --loglevel=INFO

Managing Schedule Persistence and Updates

Because LMForge stores the schedule in the Flask configuration dictionary rather than a persistent database backend, updating CELERY["beat_schedule"] and restarting the Beat process applies changes immediately. This approach eliminates database migrations for schedule adjustments but requires a process restart to pick up new intervals. Monitor task execution through Celery logs using --loglevel=INFO to verify that periodic tasks enqueue at the expected intervals.

Summary

• Extend api/config/config.py: Add the beat_schedule dictionary to the existing CELERY configuration to define intervals using seconds or crontab expressions.
• Create task handlers: Implement @shared_task functions in api/internal/task/demo_task.py or api/internal/task/document_task.py to handle LLM inference batches and index rebuilds.
• Use dependency injection: Access services like InferenceService via the Flask injector pattern already established in the platform.
• Launch with -B: Start the Beat scheduler using either the combined worker flag or as a standalone process depending on your environment.
• Restart to reload: Apply schedule changes by restarting the Beat process, as the configuration loads at startup.

Frequently Asked Questions

Where does LMForge store the Celery Beat schedule configuration?

The schedule is defined directly in the Python configuration within api/config/config.py as part of the CELERY["beat_schedule"] dictionary. This file-based configuration loads at startup, meaning you must restart the Beat process to apply any changes to task intervals or new scheduled jobs.

Can I use a database-backed scheduler with LMForge?

Yes, while the default setup uses the static configuration approach, you can install celery-sqlalchemy-scheduler or similar extensions and update api/internal/extension/celery_extension.py to initialize the database scheduler. However, this requires additional database tables and migrations beyond the current Flask-Celery integration.

How do I pass dynamic arguments to scheduled LLM inference tasks?

Include the args or kwargs keys in your beat_schedule entry within api/config/config.py. For example, "args": [100] passes the integer 100 as the first positional argument to your task function. For truly dynamic parameters fetched at runtime, implement the logic inside the task body using inference_srv to query the database for pending items.

Why should I run Beat separately from workers in production?

Running celery beat as a standalone process prevents task scheduling from stopping if a worker crashes or is restarted. It also allows horizontal scaling of workers without running duplicate Beat instances, which would cause tasks to execute multiple times. Use the separate process model when deploying LMForge in Kubernetes or Docker Swarm environments.

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