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

> Learn to configure Celery Beat for scheduled LLM inference and batch processing. Automate your machine learning workflows with expert guidance. Master automated task scheduling today.

- Repository: [Haohao/lmforge-end-to-end-llmops-platform-for-multi-model-agents](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents)
- Tags: how-to-guide
- Published: 2026-03-03

---

**Add a `beat_schedule` dictionary to the `CELERY` configuration in [`api/config/config.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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.

```python

# 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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/demo_task.py) and [`api/internal/task/document_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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.

```python

# 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.

```python

# 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:

```bash
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:

```bash

# 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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/api/internal/task/demo_task.py) or [`api/internal/task/document_task.py`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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`](https://github.com/haohao-end/lmforge-end-to-end-llmops-platform-for-multi-model-agents/blob/main/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.