RAGFlow 多租户扩展服务
技术方案

1 方案演进：V1 → V2 架构对比

2 V2 核心设计理念

2.1 RAGFlow 现有租户隔离能力复用

RAGFlow v0.25.6 已具备的租户隔离能力：

3 总体架构设计

3.1 系统架构总览

3.2 请求流转路径

3.3 关键映射关系

4 扩展服务技术选型

4.1 项目结构

ragflow-extension/
├── app/
│   ├── main.py                    # FastAPI 入口
│   ├── config.py                  # 配置管理
│   ├── proxy/                     # 代理层
│   │   ├── router.py              # 代理路由（catch-all）
│   │   ├── middleware.py          # 鉴权/限流/审计中间件
│   │   ├── rate_limiter.py        # 滑动窗口限流器
│   │   ├── tenant_resolver.py     # 租户上下文解析
│   │   └── request_transformer.py # 请求改写（注入 tenant_id 等）
│   ├── management/                # 管理服务
│   │   ├── org_router.py          # 法人管理 API
│   │   ├── group_router.py        # 用户组管理 API
│   │   ├── team_router.py         # 团队管理 API
│   │   ├── role_router.py         # 角色/权限管理 API
│   │   ├── space_router.py        # 空间管理 API
│   │   ├── ratelimit_router.py    # 限流规则管理 API
│   │   └── doc_admin_router.py    # 系统管理员文档管理 API
│   ├── models/                    # 数据模型
│   │   ├── organization.py
│   │   ├── user_group.py
│   │   ├── team.py
│   │   ├── role.py
│   │   ├── permission.py
│   │   ├── space.py
│   │   ├── rate_limit_quota.py
│   │   └── audit_log.py
│   ├── services/                  # 业务逻辑
│   │   ├── org_service.py
│   │   ├── team_service.py
│   │   ├── permission_engine.py
│   │   ├── space_service.py
│   │   └── ragflow_client.py      # RAGFlow API 客户端
│   ├── db/                        # 数据库
│   │   ├── session.py
│   │   └── migrations/
│   └── schemas/                   # Pydantic 模型
├── docker/
│   ├── Dockerfile
│   └── docker-compose.yml         # 扩展服务 compose
├── tests/
├── pyproject.toml
└── README.mdProject Structure

5 代理层设计：限流与请求路由

5.1 代理层核心职责

5.2 代理路由实现

# app/proxy/router.py

from fastapi import APIRouter, Request, Depends
from fastapi.responses import StreamingResponse
import httpx

RAGFLOW_BASE = "http://ragflow-server:9380"

proxy_router = APIRouter()

MANAGEMENT_PREFIXES = [
    "/api/v1/organizations",
    "/api/v1/teams",
    "/api/v1/rate-limits",
    "/api/v1/admin/",
    "/api/v1/ext/",
]

@proxy_router.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
async def proxy_to_ragflow(request: Request, path: str):
    full_path = f"/{path}"

    if any(full_path.startswith(p) for p in MANAGEMENT_PREFIXES):
        return Response(status_code=404)

    tenant_id = getattr(request.state, "tenant_id", None)
    user = getattr(request.state, "user", None)

    headers = dict(request.headers)
    headers.pop("host", None)
    headers.pop("content-length", None)

    if tenant_id:
        headers["X-Tenant-ID"] = tenant_id

    body = await request.body()

    url = f"{RAGFLOW_BASE}{full_path}"
    if request.url.query:
        url += f"?{request.url.query}"

    async with httpx.AsyncClient(timeout=300.0) as client:
        if request.method == "GET":
            resp = await client.get(url, headers=headers)
        elif request.method == "POST":
            resp = await client.post(url, headers=headers, content=body)
        elif request.method == "PUT":
            resp = await client.put(url, headers=headers, content=body)
        elif request.method == "DELETE":
            resp = await client.delete(url, headers=headers)
        elif request.method == "PATCH":
            resp = await client.patch(url, headers=headers, content=body)
        else:
            return Response(status_code=405)

    return Response(
        content=resp.content,
        status_code=resp.status_code,
        headers=dict(resp.headers),
    )Python — Proxy Router

5.3 限流中间件实现

# app/proxy/rate_limiter.py

import time
from typing import Optional

class SlidingWindowRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client

    async def check(self, scope_type: str, scope_id: str,
                    api_path: str, max_requests: int, window: int = 60) -> bool:
        key = f"ratelimit:{scope_type}:{scope_id}:{api_path}"
        now = time.time()
        pipe = self.redis.pipeline()
        pipe.zremrangebyscore(key, 0, now - window)
        pipe.zadd(key, {str(now): now})
        pipe.zcard(key)
        pipe.expire(key, window + 1)
        results = await pipe.execute()
        count = results[2]
        return count <= max_requests

    async def get_applicable_quotas(self, user_id: str, tenant_id: Optional[str]):
        quotas = []
        quotas.extend(await self._query_quotas("user", user_id))
        if tenant_id:
            quotas.extend(await self._query_quotas("team", tenant_id))
            org_id = await self._get_org_id(tenant_id)
            if org_id:
                quotas.extend(await self._query_quotas("org", org_id))
        return quotas

    async def _query_quotas(self, scope_type, scope_id):
        pass

    async def _get_org_id(self, tenant_id):
        passPython — Rate Limiter

5.4 代理中间件链

# app/proxy/middleware.py

from starlette.middleware.base import BaseHTTPMiddleware

class ProxyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Step 1: 认证 — 验证 JWT / API Token
        user = await self._authenticate(request)
        if not user:
            return JSONResponse({"code": 401, "message": "Unauthorized"}, 401)
        request.state.user = user

        # Step 2: 租户解析 — 从 JWT/Header/默认值获取 tenant_id
        tenant_id = await self._resolve_tenant(request, user)
        request.state.tenant_id = tenant_id

        # Step 3: 权限校验 — 检查用户是否有权访问该资源
        if not await self._check_permission(user, request.url.path, request.method):
            return JSONResponse({"code": 403, "message": "Forbidden"}, 403)

        # Step 4: 限流 — 滑动窗口检查
        rate_result = await self._check_rate_limit(user, tenant_id, request.url.path)
        if not rate_result.allowed:
            return JSONResponse(
                {"code": 429, "message": "Rate limit exceeded"},
                429,
                headers={"Retry-After": str(rate_result.retry_after)}
            )

        # Step 5: 审计日志 — 异步记录
        await self._audit_log(user, tenant_id, request)

        # Step 6: 转发请求
        response = await call_next(request)
        return responsePython — Middleware Chain

5.5 主要代理接口清单

6 管理服务设计：法人/团队/权限

6.1 管理服务核心职责

管理服务提供 RAGFlow 不具备的扩展管理能力，通过调用 RAGFlow API 完成底层操作：

6.2 RAGFlow API 客户端

# app/services/ragflow_client.py

import httpx

class RAGFlowClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key
        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=300.0
        )

    async def create_tenant_for_team(self, team_name: str, owner_email: str):
        resp = await self.client.post("/api/v1/user/register", json={
            "email": owner_email,
            "nickname": team_name,
            "password": self._generate_temp_password()
        })
        return resp.json()

    async def invite_member(self, tenant_id: str, email: str):
        resp = await self.client.post("/api/v1/tenant/members", json={
            "email": email
        }, headers={"X-Tenant-ID": tenant_id})
        return resp.json()

    async def create_dataset(self, tenant_id: str, name: str, **kwargs):
        resp = await self.client.post("/api/v1/dataset", json={
            "name": name, **kwargs
        }, headers={"X-Tenant-ID": tenant_id})
        return resp.json()

    async def list_datasets(self, tenant_id: str):
        resp = await self.client.get("/api/v1/dataset",
            headers={"X-Tenant-ID": tenant_id})
        return resp.json()

    async def get_tenant_info(self, tenant_id: str):
        resp = await self.client.get("/api/v1/tenant/info",
            headers={"X-Tenant-ID": tenant_id})
        return resp.json()Python — RAGFlow Client

6.3 团队创建流程（管理服务 → RAGFlow API）

# app/services/team_service.py

class TeamService:
    def __init__(self, db_session, ragflow_client: RAGFlowClient):
        self.db = db_session
        self.rf = ragflow_client

    async def create_team(self, org_id: str, group_id: str,
                          name: str, owner_email: str) -> Team:
        team = Team(id=uuid4(), org_id=org_id, group_id=group_id, name=name)
        self.db.add(team)

        rf_result = await self.rf.create_tenant_for_team(name, owner_email)
        tenant_id = rf_result["data"]["tenant_id"]

        team.ragflow_tenant_id = tenant_id
        self.db.commit()

        await self._create_default_spaces(team.id, tenant_id, owner_email)

        return teamPython — Team Service

7 数据模型设计

7.1 核心表结构

7.1.1 Organization（法人实体表）

CREATE TABLE organization (
    id              VARCHAR(32)  PRIMARY KEY,
    name            VARCHAR(200) NOT NULL,
    code            VARCHAR(64)  UNIQUE NOT NULL,
    description     TEXT,
    logo            TEXT,
    status          CHAR(1)      DEFAULT '1',
    isolation_level VARCHAR(16)  DEFAULT 'logical',
    created_by      VARCHAR(32),
    created_at      DATETIME     DEFAULT CURRENT_TIMESTAMP,
    updated_at      DATETIME     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);SQL

7.1.2 UserGroup（用户组表）

CREATE TABLE user_group (
    id              VARCHAR(32)  PRIMARY KEY,
    org_id          VARCHAR(32)  NOT NULL,
    name            VARCHAR(200) NOT NULL,
    code            VARCHAR(64)  NOT NULL,
    parent_id       VARCHAR(32),
    status          CHAR(1)      DEFAULT '1',
    created_by      VARCHAR(32),
    created_at      DATETIME     DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_org_code (org_id, code)
);SQL

7.1.3 Team（团队表）— 核心映射表

CREATE TABLE team (
    id                  VARCHAR(32)  PRIMARY KEY,
    org_id              VARCHAR(32)  NOT NULL,
    group_id            VARCHAR(32)  NOT NULL,
    name                VARCHAR(200) NOT NULL,
    ragflow_tenant_id   VARCHAR(32)  NOT NULL,  -- 映射到 RAGFlow Tenant
    ragflow_owner_email VARCHAR(255),            -- RAGFlow Tenant owner
    status              CHAR(1)      DEFAULT '1',
    created_by          VARCHAR(32),
    created_at          DATETIME     DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_ragflow_tenant (ragflow_tenant_id)
);SQL

7.1.4 OrgUser / UserGroupMember

CREATE TABLE org_user (
    id          VARCHAR(32)  PRIMARY KEY,
    org_id      VARCHAR(32)  NOT NULL,
    user_id     VARCHAR(32)  NOT NULL,
    ragflow_user_id VARCHAR(32),         -- 映射到 RAGFlow User
    role        VARCHAR(32)  DEFAULT 'member',
    status      CHAR(1)      DEFAULT '1',
    created_at  DATETIME     DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_org_user (org_id, user_id)
);

CREATE TABLE user_group_member (
    id          VARCHAR(32)  PRIMARY KEY,
    group_id    VARCHAR(32)  NOT NULL,
    user_id     VARCHAR(32)  NOT NULL,
    role        VARCHAR(32)  DEFAULT 'member',
    status      CHAR(1)      DEFAULT '1',
    created_at  DATETIME     DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_group_user (group_id, user_id)
);SQL

7.1.5 Permission / Role / UserRole

CREATE TABLE permission (
    id            VARCHAR(32)  PRIMARY KEY,
    code          VARCHAR(64)  UNIQUE NOT NULL,
    name          VARCHAR(200) NOT NULL,
    resource_type VARCHAR(32),
    action        VARCHAR(32),
    description   TEXT
);

CREATE TABLE role (
    id          VARCHAR(32)  PRIMARY KEY,
    code        VARCHAR(64)  UNIQUE NOT NULL,
    name        VARCHAR(200) NOT NULL,
    scope       VARCHAR(16) DEFAULT 'team',
    description TEXT
);

CREATE TABLE role_permission (
    role_id       VARCHAR(32) NOT NULL,
    permission_id VARCHAR(32) NOT NULL,
    PRIMARY KEY (role_id, permission_id)
);

CREATE TABLE user_role (
    id          VARCHAR(32)  PRIMARY KEY,
    user_id     VARCHAR(32)  NOT NULL,
    role_id     VARCHAR(32)  NOT NULL,
    scope_type  VARCHAR(16),
    scope_id    VARCHAR(32),
    granted_by  VARCHAR(32),
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_user_role_scope (user_id, role_id, scope_type, scope_id)
);SQL

7.1.6 Space（空间表）

CREATE TABLE space (
    id          VARCHAR(32)  PRIMARY KEY,
    team_id     VARCHAR(32)  NOT NULL,
    name        VARCHAR(200) NOT NULL,
    space_type  VARCHAR(16)  NOT NULL,  -- public / shared / private
    description TEXT,
    created_by  VARCHAR(32),
    status      CHAR(1) DEFAULT '1',
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_team_space (team_id, name)
);

CREATE TABLE space_member (
    id          VARCHAR(32)  PRIMARY KEY,
    space_id    VARCHAR(32)  NOT NULL,
    user_id     VARCHAR(32)  NOT NULL,
    granted_by  VARCHAR(32),
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_space_user (space_id, user_id)
);SQL

7.1.7 RateLimitQuota + AuditLog

CREATE TABLE rate_limit_quota (
    id               VARCHAR(32)  PRIMARY KEY,
    scope_type       VARCHAR(16) NOT NULL,
    scope_id         VARCHAR(32) NOT NULL,
    api_path_pattern VARCHAR(255),
    max_requests     INT NOT NULL,
    window_seconds   INT NOT NULL DEFAULT 60,
    max_tokens       INT,
    max_storage_mb   INT,
    created_at       DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_scope_pattern (scope_type, scope_id, api_path_pattern)
);

CREATE TABLE audit_log (
    id          VARCHAR(32)  PRIMARY KEY,
    user_id     VARCHAR(32)  NOT NULL,
    tenant_id   VARCHAR(32),
    action      VARCHAR(64)  NOT NULL,
    resource    VARCHAR(64),
    resource_id VARCHAR(32),
    detail      JSON,
    ip_address  VARCHAR(64),
    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_time (user_id, created_at),
    INDEX idx_tenant_time (tenant_id, created_at)
);SQL

7.2 映射关系图

8 向量数据库租户分区

8.1 分区策略（通过 RAGFlow API 管理）

扩展服务不直接操作向量数据库，而是通过 RAGFlow 的 Dataset API 和配置接口管理分区策略：

8.2 代理层向量查询改写

# app/proxy/request_transformer.py

class RequestTransformer:
    async def transform_retrieval_request(self, request_body: dict,
                                           tenant_id: str, org_id: str) -> dict:
        if "metadata_filter" not in request_body:
            request_body["metadata_filter"] = {}
        request_body["metadata_filter"]["tenant_id"] = tenant_id
        request_body["metadata_filter"]["org_id"] = org_id
        return request_body

    async def transform_dataset_create(self, request_body: dict,
                                        team: Team, space: Space) -> dict:
        request_body["meta_fields"] = {
            "org_id": team.org_id,
            "group_id": team.group_id,
            "team_id": team.id,
            "space_type": space.space_type
        }
        return request_bodyPython — Request Transformer

9 空间模型与访问隔离

9.1 空间类型

9.2 空间访问控制（代理层实现）

# app/services/space_service.py

class SpaceService:
    async def filter_accessible_datasets(self, user_id: str,
                                          team_id: str, datasets: list) -> list:
        accessible = []
        for ds in datasets:
            space = await self._get_space_by_dataset(ds["id"], team_id)
            if not space:
                accessible.append(ds)
                continue

            if space.space_type == "public":
                accessible.append(ds)
            elif space.space_type == "shared":
                if await self._is_space_member(user_id, space.id):
                    accessible.append(ds)
            elif space.space_type == "private":
                if ds.get("created_by") == user_id:
                    accessible.append(ds)

        return accessiblePython — Space Access Control

9.3 代理层响应过滤

当用户通过代理层请求 GET /api/v1/dataset 时，代理层在返回 RAGFlow 响应前，根据空间权限过滤：

# app/proxy/response_filter.py

class ResponseFilter:
    async def filter_dataset_list(self, response_body: dict,
                                   user_id: str, team_id: str) -> dict:
        datasets = response_body.get("data", [])
        filtered = await self.space_service.filter_accessible_datasets(
            user_id, team_id, datasets
        )
        response_body["data"] = filtered
        return response_bodyPython — Response Filter

10 API 设计

10.1 扩展服务 API（/api/v1/ext/）

10.2 代理接口（透传到 RAGFlow）

所有非 /api/v1/ext/ 前缀的请求，经代理层鉴权+限流后转发到 RAGFlow：

11 部署架构

11.1 Docker Compose 部署

# docker/docker-compose.yml

version: "3.8"

services:
  ragflow-extension:
    build: ..
    container_name: ragflow-extension
    ports:
      - "9380:9380"       # 对外端口（替代原 RAGFlow 端口）
    environment:
      - RAGFLOW_INTERNAL_URL=http://ragflow-server:9380
      - RAGFLOW_API_KEY=${RAGFLOW_API_KEY}
      - DATABASE_URL=postgresql+asyncpg://ext:ext@ext-db:5432/ragflow_ext
      - REDIS_URL=redis://redis:6379/1
    depends_on:
      - ext-db
    networks:
      - ragflow-network

  ext-db:
    image: postgres:16-alpine
    container_name: ragflow-ext-db
    environment:
      POSTGRES_DB: ragflow_ext
      POSTGRES_USER: ext
      POSTGRES_PASSWORD: ext
    volumes:
      - ext-db-data:/var/lib/postgresql/data
    networks:
      - ragflow-network

  ragflow-server:
    # RAGFlow 原始服务，端口改为 9381（内部端口）
    # 或保持 9380，扩展服务监听 80/443
    ...

networks:
  ragflow-network:
    external: true   # 复用 RAGFlow 已有网络

volumes:
  ext-db-data:Docker Compose

11.2 端口规划

12 RAGFlow 升级兼容策略

12.1 升级兼容矩阵

12.2 API 契约测试

# tests/test_ragflow_compat.py

import pytest
from app.services.ragflow_client import RAGFlowClient

RAGFLOW_URL = "http://ragflow-server:9381"

@pytest.mark.asyncio
async def test_dataset_api_compat():
    client = RAGFlowClient(RAGFLOW_URL, "test-key")
    resp = await client.list_datasets(tenant_id="test")
    assert "data" in resp
    assert isinstance(resp["data"], list)

@pytest.mark.asyncio
async def test_retrieval_api_compat():
    client = RAGFlowClient(RAGFLOW_URL, "test-key")
    resp = await client.client.post("/api/v1/retrieval", json={
        "question": "test",
        "dataset_ids": [],
    }, headers={"X-Tenant-ID": "test"})
    assert resp.status_code in (200, 404)

@pytest.mark.asyncio
async def test_chat_api_compat():
    client = RAGFlowClient(RAGFLOW_URL, "test-key")
    resp = await client.client.post("/api/v1/chat/completions", json={
        "model": "test",
        "messages": [{"role": "user", "content": "hi"}],
    }, headers={"X-Tenant-ID": "test"})
    assert resp.status_code in (200, 400, 404)Python — Compatibility Tests

12.3 升级流程

13 数据迁移方案

13.1 迁移步骤

13.2 数据同步脚本

# scripts/migrate_existing_tenants.py

import asyncio
from app.services.ragflow_client import RAGFlowClient
from app.db.session import get_session
from app.models import Organization, UserGroup, Team, OrgUser

async def migrate():
    rf = RAGFlowClient("http://ragflow-server:9381", "admin-key")
    db = get_session()

    tenants = await rf.list_all_tenants()

    for t in tenants:
        org = Organization(
            id=t["id"], name=f"{t['name']}_org",
            code=f"org_{t['id']}", isolation_level="logical"
        )
        db.add(org)

        group = UserGroup(
            id=t["id"], org_id=org.id,
            name=f"{t['name']}_group", code=f"group_{t['id']}"
        )
        db.add(group)

        team = Team(
            id=t["id"], org_id=org.id, group_id=group.id,
            name=t["name"], ragflow_tenant_id=t["id"]
        )
        db.add(team)

        for member in t.get("members", []):
            if member["role"] == "owner":
                ou = OrgUser(org_id=org.id, user_id=member["user_id"],
                             role="org_admin")
                db.add(ou)

    db.commit()

asyncio.run(migrate())Python — Migration Script

14 实施路线图

15 关键风险与应对

15.1 降级策略

15.2 V1 vs V2 方案最终对比