RAGFlow API Module

Overview

The RAGFlow API module is the HTTP/WebSocket adapter layer that connects external clients to the RAG (Retrieval-Augmented Generation) platform core. It implements:

• Async Web Server (Quart): High-concurrency request handling with Python async/await
• Multi-Tenant Isolation: Tenant ID filtering at query level, enforced across all data access
• Service Layer Pattern: 15+ domain services (DocumentService, DialogService, TaskService, etc.) encapsulating business logic and state mutations
• Implicit Task Queue: Redis-backed document processing with background executor coordination
• Hybrid Async+Sync: Async handlers + synchronous Peewee ORM, optimal for I/O-bound workloads
• Dual Auth Strategy: JWT tokens + session-based authentication with password hashing (bcrypt)

Architecture Diagram

Design Philosophy

Why Async Web Server + Sync ORM?

The hybrid approach achieves high concurrency for I/O-bound workloads (uploads, API calls to external LLMs) while keeping business logic simple and debuggable:

• Quart (async Flask): Handles concurrent requests without thread-per-request overhead
• Peewee (sync ORM): Simple query API, fewer abstractions to debug, connection pool (10-20) absorbs DB contention
• Task Queue (Redis): Document processing returns immediately; progress updates via WebSocket
• Avoids async ORM complexity: Async drivers (asyncpg, motor) add significant operational burden

Multi-Tenancy Enforcement

Every domain object carries a tenant_id. Auth layer loads the user's default tenant and passes it through the entire request lifecycle:

• Queries are filtered at ORM level, not application tier
• Missing tenant_id check = security vulnerability
• Services are responsible for enforcing tenant isolation (critical responsibility)

Service Layer: More Than CRUD

Domain services do not merely wrap database operations. They coordinate state transitions, manage soft deletes, queue tasks, and handle cross-table mutations atomically. Bypassing a service = bypassing multi-tenancy isolation.

Sub-Modules

Document Upload Lifecycle (Narrated Walkthrough)

Trace a PDF upload from HTTP request through task execution and progress reporting:

1. HTTP POST /api/documents/upload — Client sends multipart form with PDF file, document name, and knowledge base ID. Request lands in apps/document_api.py handler.
2. Auth Middleware Validates JWT — Token decoded in common/auth.py. User's default tenant extracted; tenant ID passed through request context.
3. Handler Calls DocumentService.create() — Business logic layer receives request. Validates: tenant isolation, knowledge base ownership, file format, size limits.
4. File Saved to Storage — PDF persisted to disk or cloud storage. File path stored in Document row.
5. Document Row Inserted — Peewee ORM inserts row into documents table. Tenant ID enforced at ORM level.
6. TaskService.queue() — DocumentService delegates heavy lifting to TaskService. Creates Task row with status=pending.
7. Redis Queue Entry Created — Task ID pushed to Redis list. Background executor polls this queue.
8. Handler Returns 202 Accepted — HTTP response sent immediately. Document created, processing queued.
9. Background Executor Polls Redis — Separate process (not blocking HTTP handler) dequeues task ID.
10. Parsing & Chunking — ParsingService extracts text. ChunkService breaks PDF into semantic chunks. Embeddings generated (external LLM call).
11. Progress Updated in Redis — Executor pushes progress events: { "task_id": "...", "progress": 45 } to Redis.
12. WebSocket Handler Fetches Status — Browser WebSocket listener polls /api/tasks/{task_id}/status or listens to server push.
13. Client Receives Progress — UI updates progress bar in real-time without blocking HTTP.
14. Task Completion — Executor marks task status=completed. Document now searchable.

Authentication & Multi-Tenancy

JWT Token Flow

All requests require a valid JWT token issued by the identity provider:

Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

Auth middleware in common/auth.py decodes the token and extracts:

• user_id — Unique user identifier
• tenant_id — Default tenant for this user
• permissions — Role-based access control (admin, editor, viewer)

Tenant Isolation

Multi-tenancy is enforced at three levels:

1. Auth Layer: User's tenant ID loaded from token
2. ORM Layer: All queries filtered by WHERE tenant_id = ?
3. Service Layer: Services validate tenant ownership before state mutations

Password Hashing

Passwords (if used for local auth) are hashed with bcrypt in apps/auth/password.py:

import bcrypt
hashed = bcrypt.hashpw(password.encode(), bcrypt.gensalt())
bcrypt.checkpw(password.encode(), hashed)  # verify

Architectural Decisions

Directory Structure

api/
├── apps/                       # Blueprint routing and REST handlers
│   ├── __init__.py            # Blueprint registration
│   ├── auth/                  # JWT decode, password hashing
│   ├── document_api.py        # /api/documents/* endpoints
│   ├── dialog_api.py          # /api/dialogs/* endpoints
│   ├── task_api.py            # /api/tasks/* endpoints (status polling)
│   └── websocket/             # WebSocket handlers for real-time updates
├── db/
│   ├── db_models.py           # Peewee ORM models (15+ tables)
│   ├── services/              # 15+ domain services
│   │   ├── document_service.py
│   │   ├── dialog_service.py
│   │   ├── task_service.py
│   │   ├── llm_service.py
│   │   └── ...
│   └── joint_services/        # Cross-domain services
│       ├── user_account_service.py
│       ├── memory_message_service.py
│       └── tenant_model_service.py
├── common/
│   ├── auth.py               # Auth decorators, JWT decode
│   ├── exceptions.py         # Custom exception classes
│   └── utils.py              # Shared utility functions
└── utils/
    ├── file_ops.py           # File I/O helpers
    ├── serialization.py      # JSON/pickle serialization
    └── logging.py            # Custom logging setup