# Core Components of the RAGFlow Backend API: Architecture and Interaction Patterns

> Explore the core components of the RAGFlow backend API including Quart server, Blueprints, service classes, Peewee ORM, and Redis queues. Learn about their architecture and interaction patterns.

- Repository: [InfiniFlow/ragflow](https://github.com/infiniflow/ragflow)
- Tags: architecture
- Published: 2026-02-23

---

**The RAGFlow backend API is built on a Quart web server that uses Blueprint modules for routing, delegates business logic to service classes, and persists data through Peewee ORM models, with background tasks managed via Redis queues and distributed locks.**

RAGFlow is an open-source RAG (Retrieval-Augmented Generation) engine that provides a modular backend API for document processing and knowledge base management. The architecture follows a clean separation of concerns, with distinct layers handling HTTP routing, business logic, and data persistence. This design enables high concurrency, extensibility through plugins, and independent scaling of background workers.

## Entry Point and Server Bootstrap

The backend lifecycle begins in [`api/ragflow_server.py`](https://github.com/infiniflow/ragflow/blob/main/api/ragflow_server.py), which serves as the application entry point. This file initializes the Quart application, sets up logging and signal handling, and launches the HTTP server.

```python

# From api/ragflow_server.py

def main():
    # Initialize database connections

    init_db()
    # Load plugins via GlobalPluginManager

    plugin_manager.load_plugins()
    # Start background progress updater thread

    start_progress_updater()
    # Launch Quart server

    app.run(host=HOST, port=PORT)

```

The server also spawns a background thread that periodically calls `DocumentService.update_progress()`, protected by a **Redis distributed lock** to prevent concurrent updates across multiple server instances.

## Application Factory and Blueprint Registration

The global Quart instance is created in [`api/apps/__init__.py`](https://github.com/infiniflow/ragflow/blob/main/api/apps/__init__.py), which acts as an application factory. This module registers CORS handlers, custom JSON encoders, and dynamically loads all API modules.

```python

# From api/apps/__init__.py

def create_app():
    app = Quart(__name__)
    # Register CORS and error handlers

    register_extensions(app)
    # Dynamically load all *_app.py modules as Blueprints

    for module in glob('api/apps/*_app.py'):
        import_module(module)
    return app

```

The `register_page` helper function converts each module into a Blueprint with an appropriate URL prefix, enabling modular routing where each functional area (users, knowledge bases, files) maintains its own route definitions.

## HTTP Layer: Blueprint Modules

Individual Blueprint modules such as [`kb_app.py`](https://github.com/infiniflow/ragflow/blob/main/kb_app.py), [`file_app.py`](https://github.com/infiniflow/ragflow/blob/main/file_app.py), and [`user_app.py`](https://github.com/infiniflow/ragflow/blob/main/user_app.py) define the RESTful endpoints for each domain. These modules handle HTTP request parsing, parameter validation, and response serialization.

For example, the knowledge base endpoint in [`api/apps/kb_app.py`](https://github.com/infiniflow/ragflow/blob/main/api/apps/kb_app.py) handles listing knowledge bases:

```python

# From api/apps/kb_app.py

@app.route('/api/v1/kb', methods=['GET'])
async def list_knowledgebases():
    # Extract query parameters

    tenant_id = request.args.get('tenant_id')
    # Delegate to service layer

    result = KnowledgebaseService.get_list(tenant_id=tenant_id)
    # Return JSON response

    return jsonify(result)

```

## Business Logic Layer: Service Classes

The service layer in `api/db/services/*.py` encapsulates all business rules and data access logic. Classes like `DocumentService`, `KnowledgebaseService`, and `UserService` provide high-level methods that operate on Peewee ORM models.

For instance, `DocumentService` manages document lifecycle operations and task queuing:

```python

# From api/db/services/document_service.py

class DocumentService:
    @staticmethod
    def update_progress(doc_id):
        # Protected by RedisDistributedLock

        with RedisDistributedLock(f"doc_{doc_id}"):
            # Aggregate task progress from Redis queue

            progress = calculate_progress(doc_id)
            # Update database via Peewee model

            Document.update(progress=progress).where(Document.id == doc_id).execute()
    
    @staticmethod
    def queue_raptor_o_graphrag_tasks(doc_id):
        # Enqueue long-running GraphRAG task to Redis

        task = {"doc_id": doc_id, "type": "graphrag"}
        RedisConn.lpush("task_queue", json.dumps(task))

```

## Data Access Layer: Peewee ORM Models

Database entities are defined in [`api/db/db_models.py`](https://github.com/infiniflow/ragflow/blob/main/api/db/db_models.py) using the Peewee ORM. The `BaseDataBase` class establishes a pooled, retry-aware connection to MySQL, PostgreSQL, or OceanBase, while individual models like `User`, `Knowledgebase`, and `Document` map to relational tables.

```python

# From api/db/db_models.py

class BaseDataBase:
    def __init__(self):
        # Pooled connection with retry logic

        self.database = PooledMySQLDatabase(
            database=settings.DB_NAME,
            host=settings.DB_HOST,
            max_connections=20,
            stale_timeout=300
        )

class Document(BaseModel):
    id = CharField(primary_key=True)
    kb_id = CharField(index=True)
    name = CharField()
    progress = FloatField(default=0.0)
    status = CharField(default="pending")
    # Additional fields...

```

## Background Processing: Redis Task Queues and Distributed Locks

Long-running operations such as document parsing, embedding generation, and GraphRAG processing are decoupled from the HTTP request cycle through a Redis-backed task queue. The [`common/utils/redis_conn.py`](https://github.com/infiniflow/ragflow/blob/main/common/utils/redis_conn.py) module provides the Redis client, while `DocumentService` uses `RedisDistributedLock` to prevent race conditions during progress updates.

The background thread spawned in [`ragflow_server.py`](https://github.com/infiniflow/ragflow/blob/main/ragflow_server.py) periodically aggregates task status from the queue and updates the database, enabling clients to poll for progress via the document endpoints.

## Configuration and Utilities

Central configuration resides in [`common/settings.py`](https://github.com/infiniflow/ragflow/blob/main/common/settings.py), which loads environment variables and defines database connections, storage backends (MinIO), and security parameters such as JWT secrets. Utility modules in `api/utils/*` provide JSON encoding, command registration, and logging helpers that are imported across the application.

## Practical Code Examples

### List Knowledge Bases (Python Client)

```python
import requests

API_URL = "http://localhost:8080/api/v1/kb"
TOKEN = "Bearer <your-access-token>"   # obtained from /api/v1/login

resp = requests.get(API_URL, headers={"Authorization": TOKEN})
print(resp.json())

```

*The endpoint is defined in [`api/apps/kb_app.py`](https://github.com/infiniflow/ragflow/blob/main/api/apps/kb_app.py) → `@app.route("/kb", methods=["GET"])` and ultimately calls `KnowledgebaseService.get_list(...)`.*

### Upload a Document (cURL)

```bash
curl -X POST http://localhost:8080/api/v1/file \
  -H "Authorization: Bearer $TOKEN" \
  -F "file=@/path/to/report.pdf" \
  -F "kb_id=kb_12345"

```

*[`file_app.py`](https://github.com/infiniflow/ragflow/blob/main/file_app.py) receives the multipart request, stores the file via `settings.STORAGE_IMPL`, creates a `File` record, then creates a `Document` record through `DocumentService.insert()`.*

### Trigger a GraphRAG Task (Python)

```python
from ragflow import RAGFlowClient

client = RAGFlowClient(base_url="http://localhost:8080", token=TOKEN)
doc_id = "doc_abc123"
task_id = client.graph_rag.create_task(doc_id)   # internally calls queue_raptor_o_graphrag_tasks()

print("Task queued:", task_id)

```

*The helper `queue_raptor_o_graphrag_tasks` lives in [`api/db/services/document_service.py`](https://github.com/infiniflow/ragflow/blob/main/api/db/services/document_service.py) and pushes a task description onto the Redis queue.*

## Summary

- **Entry Point**: [`ragflow_server.py`](https://github.com/infiniflow/ragflow/blob/main/ragflow_server.py) bootstraps the Quart application, initializes the database, loads plugins, and spawns a background progress updater thread protected by Redis distributed locks.
- **HTTP Routing**: [`api/apps/__init__.py`](https://github.com/infiniflow/ragflow/blob/main/api/apps/__init__.py) creates the global Quart instance and dynamically registers Blueprint modules (`*_app.py`) that define RESTful endpoints.
- **Business Logic**: Service classes in `api/db/services/*.py` encapsulate domain logic, validation, and database operations, using Peewee ORM models defined in [`api/db/db_models.py`](https://github.com/infiniflow/ragflow/blob/main/api/db/db_models.py).
- **Data Persistence**: Peewee models with pooled, retry-aware database connections to MySQL, PostgreSQL, or OceanBase.
- **Background Processing**: Redis-backed task queues and distributed locks enable asynchronous document processing (parsing, embedding, GraphRAG) decoupled from HTTP requests.
- **Configuration**: Centralized settings in [`common/settings.py`](https://github.com/infiniflow/ragflow/blob/main/common/settings.py) manage database, storage, and security parameters.

## Frequently Asked Questions

### How does RAGFlow handle concurrent document processing without data races?

RAGFlow uses **RedisDistributedLock** to protect critical sections during progress updates. The background thread in [`ragflow_server.py`](https://github.com/infiniflow/ragflow/blob/main/ragflow_server.py) acquires a lock specific to each document before calling `DocumentService.update_progress()`, ensuring that only one instance updates the database row at a time, even when multiple workers process chunks of the same document concurrently.

### What database backends does RAGFlow support?

According to the `BaseDataBase` implementation in [`api/db/db_models.py`](https://github.com/infiniflow/ragflow/blob/main/api/db/db_models.py), RAGFlow supports **MySQL**, **PostgreSQL**, and **OceanBase** through Peewee's pooled connection interface. The specific driver is selected via environment variables defined in [`common/settings.py`](https://github.com/infiniflow/ragflow/blob/main/common/settings.py), with connection pooling configured to handle up to 20 concurrent connections by default.

### How are long-running tasks like GraphRAG decoupled from HTTP requests?

When a client triggers a GraphRAG operation, the Blueprint handler calls `DocumentService.queue_raptor_o_graphrag_tasks()`, which serializes the task description and pushes it onto a **Redis list** acting as a task queue. The HTTP request returns immediately with a task ID, while separate worker processes (outside the API server) consume the queue and execute the heavy processing. The API server's background thread periodically aggregates worker progress and updates the `Document` table for client polling.

### Where does RAGFlow load plugins, and how do they integrate with the API?

RAGFlow initializes the **GlobalPluginManager** during server startup in [`ragflow_server.py`](https://github.com/infiniflow/ragflow/blob/main/ragflow_server.py) (lines 44-45). Plugins are dynamically loaded from the [`agent/plugin.py`](https://github.com/infiniflow/ragflow/blob/main/agent/plugin.py) module and can register additional routes or background tasks with the global Quart `app` instance. This allows custom retrievers, connectors, or API endpoints to integrate seamlessly without modifying core Blueprint files.