RAGFlow 多租户多团队
企业级增强技术方案

1 现状分析

1.1 RAGFlow 当前架构概览

RAGFlow v0.25.6 采用 Python Quart 后端 + React/TypeScript 前端 + Docker 微服务架构，核心组件包括：

1.2 当前数据模型（User-Tenant-UserTenant 三表关系）

• User：用户表，存储 email、password、nickname、access_token 等
• Tenant：租户表，每个用户注册时自动创建一个以其名字命名的默认租户
• UserTenant：用户-租户关联表，记录 role（owner/member）、invited_by、status
• Knowledgebase：知识库表，通过 tenant_id FK 归属租户，permission 字段控制可见性（me/team）
• Document：文档表，通过 kb_id FK 归属知识库
• Dialog / Agent：对话/智能体，通过 tenant_id 归属租户

1.3 当前团队模型局限

2 目标需求映射

3 总体架构设计

3.1 分层架构总览

3.2 核心设计原则

4 数据模型设计

4.1 多层租户模型（法人 → 用户组 → 团队）

4.2 新增/修改表结构

4.2.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',  -- logical / physical
    created_by      VARCHAR(32),
    created_at      DATETIME     DEFAULT CURRENT_TIMESTAMP,
    updated_at      DATETIME     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);SQL

4.2.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,
    description     TEXT,
    parent_id       VARCHAR(32),       -- 支持树形结构
    status          CHAR(1)      DEFAULT '1',
    created_by      VARCHAR(32),
    created_at      DATETIME     DEFAULT CURRENT_TIMESTAMP,
    updated_at      DATETIME     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_org_code (org_id, code)
);SQL

4.2.3 Tenant 表扩展

ALTER TABLE tenant ADD COLUMN org_id        VARCHAR(32);
ALTER TABLE tenant ADD COLUMN user_group_id VARCHAR(32);
ALTER TABLE tenant ADD COLUMN tenant_type   VARCHAR(16) DEFAULT 'team';
ALTER TABLE tenant ADD COLUMN space_config  JSON;SQL

4.2.4 UserTenant 角色扩展

ALTER TABLE usertenant ADD COLUMN role VARCHAR(32) DEFAULT 'member';
-- 角色枚举：owner / team_admin / member / viewerSQL

4.2.5 OrgUser（法人-用户关联表）— 新增

CREATE TABLE org_user (
    id          VARCHAR(32)  PRIMARY KEY,
    org_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_org_user (org_id, user_id)
);SQL

4.2.6 UserGroupMember（用户组成员表）— 新增

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

4.2.7 Permission / Role 权限表 — 新增

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

4.2.8 RateLimitQuota（限流配额表）— 新增

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,
    updated_at       DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_scope_pattern (scope_type, scope_id, api_path_pattern)
);SQL

4.2.9 Space（空间表）— 新增

CREATE TABLE space (
    id          VARCHAR(32)  PRIMARY KEY,
    tenant_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_tenant_space (tenant_id, name)
);SQL

4.2.10 Knowledgebase 表扩展

ALTER TABLE knowledgebase ADD COLUMN space_id      VARCHAR(32);
ALTER TABLE knowledgebase ADD COLUMN org_id        VARCHAR(32);
ALTER TABLE knowledgebase ADD COLUMN user_group_id VARCHAR(32);SQL

4.3 完整 ER 关系图

5 核心模块详细设计

5.1 角色与权限体系（RBAC + ABAC）

5.1.1 角色定义

5.1.2 权限矩阵

* 仅限本作用域内资源

5.1.3 权限引擎实现

class PermissionEngine:
    def __init__(self, db_session):
        self.db = db_session

    async def check_permission(self, user_id: str, resource_type: ResourceType,
                                action: Action, scope_id: str = None) -> bool:
        user_roles = await self._get_user_roles(user_id, resource_type, scope_id)
        for ur in user_roles:
            role_perms = await self._get_role_permissions(ur.role_id)
            for perm in role_perms:
                if perm.resource_type == resource_type.value and perm.action == action.value:
                    if ur.scope_id is None or ur.scope_id == scope_id:
                        return True
        return await self._check_abac_policy(user_id, resource_type, action, scope_id)

def require_permission(resource_type: ResourceType, action: Action):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            user = get_current_user()
            engine = PermissionEngine(get_db_session())
            scope_id = kwargs.get("tenant_id") or kwargs.get("kb_id")
            if not await engine.check_permission(user.id, resource_type, action, scope_id):
                raise HTTPException(403, "Permission denied")
            return await func(*args, **kwargs)
        return wrapper
    return decoratorPython

5.2 租户上下文中间件

class TenantContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if any(request.url.path.startswith(p) for p in EXEMPT_PATHS):
            return await call_next(request)

        user = getattr(request.state, "user", None)
        if not user:
            return JSONResponse({"code": 401, "message": "Unauthorized"}, 401)

        tenant_id = request.headers.get("X-Tenant-ID")
        if not tenant_id:
            tenant_id = self._resolve_default_tenant(user.id)

        if not await self._validate_tenant_membership(user.id, tenant_id):
            return JSONResponse({"code": 403, "message": "Not a member"}, 403)

        request.state.tenant_id = tenant_id
        request.state.org_id = await self._resolve_org_id(tenant_id)
        request.state.user_group_id = await self._resolve_user_group_id(tenant_id)

        return await call_next(request)Python

5.3 用户限流模块

class RateLimiterMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, redis_client):
        super().__init__(app)
        self.redis = redis_client

    async def _sliding_window_check(self, key, max_requests, window):
        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)
        results = await pipe.execute()
        return results[2] <= max_requestsPython

5.4 向量数据库租户分区策略

5.4.1 分区策略对比

5.4.2 推荐方案：Collection 级分区 + Partition Key 混合策略

5.5 空间模型设计

5.6 数据逻辑隔离实现

5.6.1 数据库行级安全（PostgreSQL RLS）

ALTER TABLE knowledgebase ENABLE ROW LEVEL SECURITY;
ALTER TABLE document ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation_policy ON knowledgebase
    USING (tenant_id = current_setting('app.current_tenant'));

CREATE POLICY org_isolation_policy ON knowledgebase
    USING (org_id = current_setting('app.current_org'));SQL

5.6.2 Redis Key 隔离

def redis_key(scope: str, scope_id: str, *parts) -> str:
    return f"ragflow:{scope}:{scope_id}:{':'.join(parts)}"

# ragflow:tenant:abc123:session:xyz789
# ragflow:org:org001:config:llm_settings
# ragflow:user:uid001:token_usage:20260608Python

5.6.3 MinIO 对象存储隔离

class TenantStorage:
    def get_bucket_name(self, org_id, tenant_id=None):
        org_config = self._get_org_config(org_id)
        if org_config.isolation_level == 'physical':
            return f"ragflow-org-{org_id}"
        else:
            return f"ragflow-tenant-{tenant_id}"

    def get_object_path(self, tenant_id, kb_id, doc_id, filename):
        return f"{tenant_id}/{kb_id}/{doc_id}/{filename}"Python

6 API 设计

6.1 新增 API 端点

6.2 现有 API 兼容性

7 前端改造要点

8 数据迁移方案

8.1 迁移步骤

8.2 回填 SQL 示例

-- Step 1: 为每个现有 tenant 创建默认法人
INSERT INTO organization (id, name, code, status, isolation_level, created_by)
SELECT id, CONCAT(name, '_org'), CONCAT('org_', id), '1', 'logical', id
FROM tenant;

-- Step 2: 为每个现有 tenant 创建默认用户组
INSERT INTO user_group (id, org_id, name, code, status, created_by)
SELECT id, id, CONCAT(name, '_group'), CONCAT('group_', id), '1', id
FROM tenant;

-- Step 3: 回填 tenant 扩展字段
UPDATE tenant t SET org_id = t.id, user_group_id = t.id, tenant_type = 'team';

-- Step 4: 回填 knowledgebase 扩展字段
UPDATE knowledgebase kb
JOIN tenant t ON kb.tenant_id = t.id
SET kb.org_id = t.org_id, kb.user_group_id = t.user_group_id;

-- Step 5: 为现有 owner 角色升级为 team_admin
UPDATE usertenant SET role = 'team_admin' WHERE role = 'owner';SQL

9 可复用性设计

9.1 中间件抽象

9.2 配置驱动

tenant:
  default_isolation_level: logical
  auto_create_org_on_register: true
  auto_create_group_on_register: true
  max_teams_per_org: 100
  max_members_per_team: 500

vector_db:
  partition_strategy: collection
  auto_create_partition: true
  index_refresh_interval: 5s

rate_limit:
  enabled: true
  default_window: 60
  default_max_requests: 100
  storage_quota_default_mb: 5120YAML

10 安全增强

10.1 认证体系

10.2 安全措施清单

11 实施路线图

12 关键风险与应对

附录 A：RAGFlow v0.25.6 关键特性参考

附录 B：向量数据库选型建议