flowchart TB
    Start([ragflow_server.py 启动]) --> InitLogger[初始化日志系统]
    InitLogger --> InitSettings[加载配置文件<br/>settings.init_settings]
    InitSettings --> InitDB[初始化数据库连接池<br/>init_web_db]
    InitDB --> InitData[初始化默认数据<br/>init_web_data]
    InitData --> InitRuntimeConfig[初始化运行时配置<br/>RuntimeConfig.init_env]
    InitRuntimeConfig --> RegisterBlueprints[注册路由 Blueprint<br/>扫描 apps/ 目录]
    RegisterBlueprints --> InitPlugins[加载插件<br/>GlobalPluginManager.load_plugins]
    InitPlugins --> StartBackgroundThread[启动后台线程<br/>update_progress]
    StartBackgroundThread --> StartHTTPServer[启动 HTTP 服务器<br/>Werkzeug run_simple]
    StartHTTPServer --> Ready([服务就绪])

    subgraph "配置初始化详情"
        InitSettings --> LoadYAML[加载 service_conf.yaml]
        LoadYAML --> LoadLLMFactories[加载 llm_factories.json]
        LoadLLMFactories --> ConnectDB[连接 MySQL/PostgreSQL]
        ConnectDB --> ConnectRedis[连接 Redis]
        ConnectRedis --> ConnectES[连接 Elasticsearch/Infinity]
        ConnectES --> ConnectMinIO[连接 MinIO/S3]
        ConnectMinIO --> InitRetriever[初始化 Retriever]
    end

flowchart TB
    subgraph "外部请求层"
        WebUI[Web UI<br/>React SPA]
        SDK[Python SDK]
        HTTP[第三方 HTTP Client]
    end

    subgraph "API 层（Flask 应用）"
        Server[ragflow_server.py<br/>Flask App 入口]
        
        subgraph "认证与中间件层"
            LoginManager[login_manager<br/>Flask-Login]
            TokenAuth[token_required<br/>JWT 校验装饰器]
            OAuth[OAuth/OIDC<br/>第三方登录适配器]
            CORS[CORS 跨域]
            ErrorHandler[全局异常处理器]
        end
        
        subgraph "路由层（Blueprint）"
            direction LR
            UserApp[user_app<br/>用户管理]
            KBApp[kb_app<br/>知识库管理]
            DocApp[document_app<br/>文档管理]
            DialogApp[dialog_app<br/>对话配置]
            ConvApp[conversation_app<br/>会话管理]
            CanvasApp[canvas_app<br/>Agent 画布]
            ChunkApp[chunk_app<br/>切片管理]
            SearchApp[search_app<br/>检索管理]
            SDKRoutes[sdk/*<br/>SDK 专用路由]
        end
        
        subgraph "Service 业务服务层"
            direction TB
            UserSvc[UserService<br/>用户/租户]
            KBSvc[KnowledgebaseService<br/>知识库]
            DocSvc[DocumentService<br/>文档]
            FileSvc[FileService<br/>文件]
            DialogSvc[DialogService<br/>对话配置]
            ConvSvc[ConversationService<br/>会话]
            TaskSvc[TaskService<br/>任务调度]
            LLMSvc[LLMService<br/>LLM 调用封装]
            TenantLLMSvc[TenantLLMService<br/>租户 LLM 配置]
        end
        
        subgraph "数据访问层（ORM）"
            Models[(db_models.py<br/>Peewee Models<br/>User/Tenant/KB/Doc/...)]
        end
    end

    subgraph "核心引擎层"
        direction TB
        RAGEngine[RAG Engine<br/>检索增强生成]
        RAGRetriever[Retriever<br/>混合检索]
        RAGRerank[Reranker<br/>重排序]
        RAGGenerator[Generator<br/>LLM 生成]
        
        AgentEngine[Agent Engine<br/>工作流执行]
        AgentCanvas[Canvas<br/>DAG 图执行器]
        AgentComponents[Components<br/>LLM/Tool/Switch/Iteration]
    end

    subgraph "存储层"
        MySQL[(MySQL/PG<br/>元数据存储)]
        Redis[(Redis<br/>Session/Queue/Cache)]
        ES[(Elasticsearch/Infinity<br/>向量+全文检索)]
        MinIO[(MinIO/S3<br/>文件对象存储)]
    end

    subgraph "外部依赖层"
        LLMProviders[LLM 提供商<br/>OpenAI/Ollama/DeepSeek/Azure]
        EmbeddingProviders[Embedding 模型<br/>bge-large-zh/bce-embedding]
        RerankProviders[Rerank 模型<br/>bge-reranker-v2-m3]
    end

    WebUI --> Server
    SDK --> Server
    HTTP --> Server

    Server --> LoginManager
    Server --> TokenAuth
    Server --> OAuth
    Server --> CORS
    Server --> ErrorHandler

    LoginManager --> UserApp
    TokenAuth --> SDKRoutes

    UserApp --> UserSvc
    KBApp --> KBSvc
    DocApp --> DocSvc
    DocApp --> FileSvc
    DialogApp --> DialogSvc
    ConvApp --> ConvSvc
    ConvApp --> DialogSvc
    CanvasApp --> AgentEngine
    ChunkApp --> DocSvc
    SearchApp --> RAGEngine

    UserSvc --> Models
    KBSvc --> Models
    DocSvc --> Models
    DialogSvc --> Models
    ConvSvc --> Models
    FileSvc --> Models
    TaskSvc --> Models

    Models --> MySQL

    ConvSvc --> DialogSvc
    DialogSvc --> RAGEngine
    DialogSvc --> LLMSvc
    LLMSvc --> TenantLLMSvc

    RAGEngine --> RAGRetriever
    RAGEngine --> RAGRerank
    RAGEngine --> RAGGenerator
    RAGRetriever --> ES
    RAGGenerator --> LLMProviders
    RAGRerank --> RerankProviders

    AgentEngine --> AgentCanvas
    AgentCanvas --> AgentComponents
    AgentComponents --> LLMProviders

    TaskSvc -.推送任务.-> Redis
    FileSvc --> MinIO
    TenantLLMSvc --> Redis
    LoginManager --> Redis

    LLMSvc --> EmbeddingProviders

classDiagram
    class User {
        +String id [PK]
        +String email [UQ]
        +String password_hash
        +String nickname
        +String avatar
        +String access_token [UQ]
        +DateTime create_time
        +login()
        +to_json()
    }

    class Tenant {
        +String id [PK]
        +String name
        +String llm_id
        +String embd_id
        +JSONField llm_config
        +JSONField asr_config
        +Int credit
    }

    class UserTenant {
        +String id [PK]
        +String tenant_id [FK]
        +String user_id [FK]
        +String role [ENUM]
        +String invited_by
        +DateTime create_time
    }

    class Knowledgebase {
        +String id [PK]
        +String tenant_id [FK]
        +String name [UQ per tenant]
        +String avatar
        +String description
        +String embd_id
        +String parser_id
        +JSONField parser_config
        +String language
        +String permission
        +Int doc_num
        +Int token_num
        +Int chunk_num
        +DateTime create_time
    }

    class Document {
        +String id [PK]
        +String kb_id [FK]
        +String name
        +String location
        +Int size
        +String type
        +String parser_id
        +JSONField parser_config
        +String source_type
        +Float progress
        +Int chunk_num
        +Int token_num
        +String status
        +DateTime create_time
    }

    class File {
        +String id [PK]
        +String tenant_id [FK]
        +String created_by [FK]
        +String name
        +String location
        +String type
        +String source_type
        +Int size
        +DateTime create_time
    }

    class File2Document {
        +String id [PK]
        +String file_id [FK]
        +String document_id [FK]
    }

    class Task {
        +String id [PK]
        +String doc_id [FK]
        +Int from_page
        +Int to_page
        +String status
        +Float progress
        +String progress_msg
        +DateTime begin_at
        +DateTime create_time
    }

    class Dialog {
        +String id [PK]
        +String tenant_id [FK]
        +String name
        +String description
        +String icon
        +ListField kb_ids
        +String llm_id
        +JSONField llm_setting
        +JSONField prompt_config
        +Float similarity_threshold
        +Float vector_similarity_weight
        +Int top_n
        +String rerank_id
        +DateTime create_time
    }

    class Conversation {
        +String id [PK]
        +String dialog_id [FK]
        +String name
        +String message [JSON]
        +String reference [JSON]
        +Int tokens_num
        +DateTime create_time
    }

    User "1" --> "N" UserTenant : belongs to
    Tenant "1" --> "N" UserTenant : has
    Tenant "1" --> "N" Knowledgebase : owns
    Knowledgebase "1" --> "N" Document : contains
    Document "1" --> "N" Task : has tasks
    Document "1" --> "1" File2Document : maps to
    File "1" --> "1" File2Document : maps to
    Tenant "1" --> "N" Dialog : owns
    Dialog "1" --> "N" Conversation : has
    User "1" --> "N" File : uploads

{
    "email": "user@example.com",    # 用户邮箱，必填
    "password": "<encrypted>"        # RSA 加密后的密码，必填
}

{
    "code": 0,
    "message": "Welcome back!",
    "data": {
        "id": "user_uuid",
        "email": "user@example.com",
        "nickname": "张三",
        "avatar": "http://minio/avatar.png",
        "access_token": "new_uuid_token",
        "create_time": 1697000000000
    }
}

@manager.route("/login", methods=["POST"])
def login():
    """用户登录"""
    if not request.json:
        return get_json_result(data=False, code=settings.RetCode.AUTHENTICATION_ERROR, message="Unauthorized!")
    
    email = request.json.get("email", "")
    users = UserService.query(email=email)
    if not users:
        return get_json_result(
            data=False,
            code=settings.RetCode.AUTHENTICATION_ERROR,
            message=f"Email: {email} is not registered!",
        )
    
    # 解密密码
    password = request.json.get("password")
    password = decrypt(password)  # RSA 解密
    
    # 校验密码
    user = UserService.query_user(email, password)
    
    # 检查账户状态
    if user and user.is_active == "0":
        return get_json_result(
            data=False,
            code=settings.RetCode.FORBIDDEN,
            message="This account has been disabled, please contact the administrator!",
        )
    elif user:
        # 登录成功，刷新 access_token
        response_data = user.to_json()
        user.access_token = get_uuid()  # 生成新 Token
        login_user(user)  # Flask-Login 记录登录态
        user.update_time = current_timestamp()
        user.save()  # 更新数据库
        return construct_response(data=response_data, auth=user.get_id(), message="Welcome back!")
    else:
        return get_json_result(
            data=False,
            code=settings.RetCode.AUTHENTICATION_ERROR,
            message="Email and password do not match!",
        )

# api/apps/__init__.py
@login_manager.request_loader
def load_user(web_request):
    """从 Authorization Header 或 Session 加载用户"""
    authorization = web_request.headers.get("Authorization")
    if authorization:
        # Token 认证（SDK 调用）
        jwt = Serializer(secret_key=settings.SECRET_KEY)
        access_token = str(jwt.loads(authorization))  # 解析 JWT
        user = UserService.query(access_token=access_token, status=StatusEnum.VALID.value)
        return user[0] if user else None
    else:
        # Session 认证（Web UI）
        return None  # Flask-Login 会从 Session 读取

# api/db/services/user_service.py
class UserService(CommonService):
    @classmethod
    def query_user(cls, email, password):
        """通过邮箱和密码查询用户"""
        users = cls.query(email=email, status=StatusEnum.VALID.value)
        if not users:
            return None
        user = users[0]
        # 校验密码哈希
        if check_password_hash(user.password, password):
            return user
        return None

sequenceDiagram
    autonumber
    participant C as 客户端（前端/SDK）
    participant API as /v1/user/login
    participant UserSvc as UserService
    participant MySQL as MySQL
    participant Redis as Redis

    C->>API: POST 登录请求（email + encrypted_password）
    API->>API: decrypt(password) - RSA 解密

    API->>UserSvc: query(email)
    UserSvc->>MySQL: SELECT * FROM user WHERE email=?
    MySQL-->>UserSvc: 返回用户记录

    alt 用户不存在
        API-->>C: 401 Unauthorized
    else 用户存在
        API->>UserSvc: query_user(email, password)
        UserSvc->>UserSvc: check_password_hash()
        
        alt 密码错误
            API-->>C: 401 密码错误
        else 密码正确
            UserSvc-->>API: 返回 User 对象
            
            API->>API: user.access_token = get_uuid()
            API->>API: login_user(user) - 写入 Session
            API->>Redis: 存储 Session
            Redis-->>API: OK
            
            API->>MySQL: UPDATE user SET access_token=?, update_time=?
            MySQL-->>API: OK
            
            API-->>C: 200 OK + user_data + access_token
        end
    end

{
    "email": "new@example.com",
    "nickname": "张三",
    "password": "<encrypted>",
    "captcha": "123456"  # 可选，邮箱验证码
}

{
    "code": 0,
    "message": "Register successfully!",
    "data": {
        "id": "user_uuid",
        "email": "new@example.com",
        "access_token": "uuid_token"
    }
}

@manager.route("/register", methods=["POST"])
def register():
    req = request.json
    
    # 校验邮箱是否已注册
    if UserService.query(email=req["email"]):
        return get_data_error_result(message="Email already registered!")
    
    # 校验验证码（如果启用）
    if settings.REQUIRE_EMAIL_VERIFICATION:
        if not verify_otp(req["email"], req["captcha"]):
            return get_data_error_result(message="Invalid captcha!")
    
    # 创建用户
    user_id = get_uuid()
    tenant_id = get_uuid()
    password_hash = generate_password_hash(decrypt(req["password"]))
    
    # 1. 创建 Tenant
    TenantService.save(
        id=tenant_id,
        name=req["nickname"],
        llm_id=settings.DEFAULT_LLM,
        embd_id=settings.DEFAULT_EMBEDDING
    )
    
    # 2. 创建 User
    UserService.save(
        id=user_id,
        email=req["email"],
        nickname=req["nickname"],
        password=password_hash,
        access_token=get_uuid()
    )
    
    # 3. 关联 User 与 Tenant
    UserTenantService.save(
        tenant_id=tenant_id,
        user_id=user_id,
        role=UserTenantRole.OWNER.value
    )
    
    # 4. 初始化 LLM 配置
    for llm in get_init_tenant_llm():
        TenantLLMService.save(tenant_id=tenant_id, **llm)
    
    return get_json_result(data=user.to_json())

class CreateDatasetReq(BaseModel):
    name: str = Field(..., min_length=1, max_length=128, description="知识库名称")
    avatar: Optional[str] = Field(None, description="Base64 编码的头像")
    description: Optional[str] = Field("", description="描述")
    embedding_model: Optional[str] = Field(None, description="Embedding 模型名称")
    permission: Optional[str] = Field("me", description="权限：me（私有）、team（团队）")
    chunk_method: str = Field("naive", description="切片方法")
    parser_config: Optional[Dict] = Field(None, description="解析器配置")

{
    "code": 0,
    "message": "success",
    "data": {
        "id": "kb_uuid",
        "name": "我的知识库",
        "avatar": "base64...",
        "embedding_model": "BAAI/bge-large-zh-v1.5",
        "chunk_method": "naive",
        "parser_config": {
            "chunk_token_count": 128,
            "layout_recognize": true,
            "delimiter": "\\n!?。；！？"
        },
        "permission": "me",
        "doc_num": 0,
        "chunk_num": 0,
        "create_time": 1697000000000
    }
}

@manager.route("/datasets", methods=["POST"])
@token_required
def create(tenant_id):
    """创建知识库"""
    req, err = validate_and_parse_json_request(request, CreateDatasetReq)
    if err is not None:
        return get_error_argument_result(err)
    
    # 1. 校验名称唯一性
    if KnowledgebaseService.get_or_none(name=req["name"], tenant_id=tenant_id, status=StatusEnum.VALID.value):
        return get_error_operating_result(message=f"Dataset name '{req['name']}' already exists")
    
    # 2. 生成默认配置
    req["parser_config"] = get_parser_config(req["parser_id"], req["parser_config"])
    req["id"] = get_uuid()
    req["tenant_id"] = tenant_id
    req["created_by"] = tenant_id
    
    # 3. 继承租户默认 Embedding 模型
    ok, t = TenantService.get_by_id(tenant_id)
    if not req.get("embd_id"):
        req["embd_id"] = t.embd_id
    else:
        # 校验 Embedding 模型是否可用
        ok, err = verify_embedding_availability(req["embd_id"], tenant_id)
        if not ok:
            return err
    
    # 4. 保存到数据库
    if not KnowledgebaseService.save(**req):
        return get_error_data_result(message="Create dataset error.(Database error)")
    
    # 5. 返回创建结果
    ok, k = KnowledgebaseService.get_by_id(req["id"])
    response_data = remap_dictionary_keys(k.to_dict())
    return get_result(data=response_data)

# api/utils/api_utils.py
def get_parser_config(parser_id, user_config):
    """获取解析器配置"""
    # 从预定义模板加载默认配置
    default_config = DEFAULT_PARSER_CONFIG.get(parser_id, {})
    if user_config:
        # 用户配置覆盖默认配置
        return deep_merge(default_config, user_config)
    return default_config

# api/db/services/knowledgebase_service.py
class KnowledgebaseService(CommonService):
    model = Knowledgebase
    
    @classmethod
    def save(cls, **kwargs):
        """保存知识库"""
        # 设置默认值
        kwargs.setdefault("doc_num", 0)
        kwargs.setdefault("chunk_num", 0)
        kwargs.setdefault("token_num", 0)
        kwargs.setdefault("create_time", current_timestamp())
        kwargs.setdefault("update_time", current_timestamp())
        return super().save(**kwargs)

sequenceDiagram
    autonumber
    participant C as 客户端（SDK/Web UI）
    participant API as /api/v1/datasets
    participant TokenAuth as token_required
    participant KBSvc as KnowledgebaseService
    participant TenantSvc as TenantService
    participant MySQL as MySQL

    C->>API: POST 创建知识库请求 + Authorization Header
    API->>TokenAuth: 校验 Token
    TokenAuth->>MySQL: 查询 User (access_token=?)
    MySQL-->>TokenAuth: 返回 User
    TokenAuth-->>API: 返回 tenant_id

    API->>API: validate_and_parse_json_request()
    API->>KBSvc: get_or_none(name, tenant_id)
    KBSvc->>MySQL: SELECT * FROM knowledgebase WHERE name=? AND tenant_id=?
    MySQL-->>KBSvc: 返回空（名称未占用）

    API->>TenantSvc: get_by_id(tenant_id)
    TenantSvc->>MySQL: SELECT * FROM tenant WHERE id=?
    MySQL-->>TenantSvc: 返回 Tenant
    TenantSvc-->>API: 返回默认 embd_id

    API->>API: get_parser_config(parser_id, parser_config)
    API->>KBSvc: save(id, name, embd_id, parser_config, ...)
    KBSvc->>MySQL: INSERT INTO knowledgebase VALUES (...)
    MySQL-->>KBSvc: OK

    API->>KBSvc: get_by_id(kb_id)
    KBSvc->>MySQL: SELECT * FROM knowledgebase WHERE id=?
    MySQL-->>KBSvc: 返回 Knowledgebase
    KBSvc-->>API: 返回 KB 对象

    API->>API: remap_dictionary_keys()
    API-->>C: 200 OK + KB 数据

class UpdateDatasetReq(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=128)
    avatar: Optional[str] = None
    description: Optional[str] = None
    embedding_model: Optional[str] = None
    permission: Optional[str] = None
    chunk_method: Optional[str] = None
    parser_config: Optional[Dict] = None
    pagerank: Optional[int] = Field(None, ge=0, le=10, description="PageRank 权重")

@manager.route("/datasets/<dataset_id>", methods=["PUT"])
@token_required
def update(tenant_id, dataset_id):
    """更新知识库"""
    req, err = validate_and_parse_json_request(request, UpdateDatasetReq, exclude_unset=True)
    if not req:
        return get_error_argument_result(message="No properties were modified")
    
    # 1. 校验权限
    kb = KnowledgebaseService.get_or_none(id=dataset_id, tenant_id=tenant_id)
    if kb is None:
        return get_error_permission_result(message=f"User '{tenant_id}' lacks permission for dataset '{dataset_id}'")
    
    # 2. 深度合并 parser_config
    if req.get("parser_config"):
        req["parser_config"] = deep_merge(kb.parser_config, req["parser_config"])
    
    # 3. 校验 embedding_model 修改
    if "embd_id" in req:
        if kb.chunk_num != 0 and req["embd_id"] != kb.embd_id:
            return get_error_data_result(
                message=f"When chunk_num ({kb.chunk_num}) > 0, embedding_model must remain {kb.embd_id}")
    
    # 4. 更新 PageRank（同步更新 ES）
    if "pagerank" in req and req["pagerank"] != kb.pagerank:
        if req["pagerank"] > 0:
            settings.docStoreConn.update(
                {"kb_id": kb.id}, 
                {PAGERANK_FLD: req["pagerank"]}, 
                search.index_name(kb.tenant_id), 
                kb.id
            )
        else:
            settings.docStoreConn.update(
                {"exists": PAGERANK_FLD}, 
                {"remove": PAGERANK_FLD}, 
                search.index_name(kb.tenant_id), 
                kb.id
            )
    
    # 5. 更新数据库
    if not KnowledgebaseService.update_by_id(kb.id, req):
        return get_error_data_result(message="Update dataset error.(Database error)")
    
    # 6. 返回更新结果
    ok, k = KnowledgebaseService.get_by_id(kb.id)
    response_data = remap_dictionary_keys(k.to_dict())
    return get_result(data=response_data)

{
    "ids": ["kb_uuid_1", "kb_uuid_2"]  // null 表示删除全部，[] 表示不删除
}

{
    "code": 0,
    "message": "Successfully deleted 2 datasets",
    "data": {
        "success_count": 2,
        "errors": []
    }
}

@manager.route("/datasets", methods=["DELETE"])
@token_required
def delete(tenant_id):
    """批量删除知识库"""
    req, err = validate_and_parse_json_request(request, DeleteDatasetReq)
    
    # 1. 获取待删除的知识库列表
    kb_id_instance_pairs = []
    if req["ids"] is None:
        # null 表示删除全部
        kbs = KnowledgebaseService.query(tenant_id=tenant_id)
        for kb in kbs:
            kb_id_instance_pairs.append((kb.id, kb))
    else:
        # 指定 ID 列表
        for kb_id in req["ids"]:
            kb = KnowledgebaseService.get_or_none(id=kb_id, tenant_id=tenant_id)
            if kb is None:
                return get_error_permission_result(message=f"User '{tenant_id}' lacks permission for datasets: '{kb_id}'")
            kb_id_instance_pairs.append((kb_id, kb))
    
    # 2. 逐个删除知识库
    errors = []
    success_count = 0
    for kb_id, kb in kb_id_instance_pairs:
        # 2.1 删除所有文档
        for doc in DocumentService.query(kb_id=kb_id):
            if not DocumentService.remove_document(doc, tenant_id):
                errors.append(f"Remove document '{doc.id}' error for dataset '{kb_id}'")
                continue
            # 2.2 删除文件关联
            f2d = File2DocumentService.get_by_document_id(doc.id)
            FileService.filter_delete([File.id == f2d[0].file_id])
            File2DocumentService.delete_by_document_id(doc.id)
        
        # 2.3 删除知识库
        if not KnowledgebaseService.delete_by_id(kb_id):
            errors.append(f"Delete dataset error for {kb_id}")
            continue
        success_count += 1
    
    # 3. 返回结果
    if not errors:
        return get_result()
    return get_result(data={"success_count": success_count, "errors": errors[:5]})

{
    "code": 0,
    "message": "success",
    "data": {
        "doc_id": "doc_uuid",
        "name": "example.pdf",
        "size": 1024000,
        "type": "pdf",
        "progress": 0.0,
        "chunk_num": 0,
        "create_time": 1697000000000
    }
}

@manager.route("/documents/upload", methods=["POST"])
@login_required
def upload():
    """上传文档"""
    kb_id = request.form.get("kb_id")
    if not kb_id:
        return get_error_argument_result(message="kb_id is required")
    
    # 1. 校验知识库权限
    kb = KnowledgebaseService.get_or_none(id=kb_id, tenant_id=current_user.id)
    if not kb:
        return get_error_permission_result(message="Knowledge base not found")
    
    # 2. 获取上传文件
    file = request.files.get("file")
    if not file:
        return get_error_argument_result(message="file is required")
    
    # 3. 校验文件大小
    file_size = len(file.read())
    file.seek(0)
    if file_size > settings.MAX_FILE_SIZE:
        return get_error_argument_result(message=f"File size exceeds {settings.MAX_FILE_SIZE} bytes")
    
    # 4. 上传到 MinIO
    file_location = f"{current_user.id}/{kb_id}/{get_uuid()}/{file.filename}"
    STORAGE_IMPL.put(bucket=settings.MINIO_BUCKET, key=file_location, data=file.read())
    
    # 5. 创建 File 记录
    file_id = get_uuid()
    FileService.save(
        id=file_id,
        tenant_id=current_user.id,
        created_by=current_user.id,
        name=file.filename,
        location=file_location,
        type=get_file_type(file.filename),
        size=file_size
    )
    
    # 6. 创建 Document 记录
    doc_id = get_uuid()
    DocumentService.save(
        id=doc_id,
        kb_id=kb_id,
        name=file.filename,
        location=file_location,
        size=file_size,
        type=get_file_type(file.filename),
        parser_id=kb.parser_id,
        parser_config=kb.parser_config,
        progress=0.0,
        chunk_num=0
    )
    
    # 7. 创建 File2Document 关联
    File2DocumentService.save(file_id=file_id, document_id=doc_id)
    
    # 8. 推送解析任务（如果 run=1）
    if request.form.get("run") == "1":
        TaskService.push_parsing_task(doc_id, kb_id, current_user.id)
    
    return get_result(data={"doc_id": doc_id})

sequenceDiagram
    autonumber
    participant C as 客户端
    participant API as /v1/documents/upload
    participant MinIO as MinIO
    participant MySQL as MySQL
    participant Redis as Redis 队列
    participant TaskExec as TaskExecutor

    C->>API: POST 上传文件（multipart/form-data）
    API->>API: 校验文件大小

    API->>MinIO: PUT 上传文件
    MinIO-->>API: 返回存储路径

    API->>MySQL: INSERT INTO file
    API->>MySQL: INSERT INTO document
    API->>MySQL: INSERT INTO file2document
    MySQL-->>API: OK

    alt run=1（立即解析）
        API->>Redis: 推送解析任务
        Redis-->>API: OK
        API-->>C: 201 Created + doc_id
        
        Redis->>TaskExec: 拉取任务
        TaskExec->>MinIO: 下载文件
        TaskExec->>TaskExec: 调用 DeepDoc 解析
        TaskExec->>MySQL: 更新 progress
    else run=0（不解析）
        API-->>C: 201 Created + doc_id
    end

{
    "name": "技术支持助手",
    "description": "回答技术问题的助手",
    "avatar": "base64...",
    "dataset_ids": ["kb_uuid_1", "kb_uuid_2"],
    "llm": {
        "model_name": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 2048,
        "top_p": 0.9
    },
    "prompt": {
        "prompt": "你是一个专业的技术支持助手...",
        "opener": "你好！我是技术支持助手，有什么可以帮您？",
        "show_quote": true,
        "variables": [{"key": "knowledge", "optional": false}]
    },
    "similarity_threshold": 0.2,
    "keywords_similarity_weight": 0.3,
    "top_n": 6,
    "rerank_model": "BAAI/bge-reranker-v2-m3"
}

{
    "code": 0,
    "message": "success",
    "data": {
        "id": "chat_uuid",
        "name": "技术支持助手",
        "dataset_ids": ["kb_uuid_1", "kb_uuid_2"],
        "llm": {"model_name": "gpt-4", ...},
        "prompt": {...},
        "create_time": 1697000000000
    }
}

@manager.route("/chats", methods=["POST"])
@token_required
def create(tenant_id):
    """创建对话配置"""
    req = request.json
    
    # 1. 校验知识库权限与状态
    ids = req.get("dataset_ids", [])
    for kb_id in ids:
        if not KnowledgebaseService.accessible(kb_id=kb_id, user_id=tenant_id):
            return get_error_data_result(f"You don't own the dataset {kb_id}")
        kbs = KnowledgebaseService.query(id=kb_id)
        if kbs[0].chunk_num == 0:
            return get_error_data_result(f"The dataset {kb_id} doesn't own parsed file")
    
    # 2. 校验 Embedding 模型一致性
    kbs = KnowledgebaseService.get_by_ids(ids) if ids else []
    embd_ids = [kb.embd_id for kb in kbs]
    if len(set(embd_ids)) > 1:
        return get_result(message='Datasets use different embedding models.', code=settings.RetCode.AUTHENTICATION_ERROR)
    
    # 3. 处理 LLM 配置
    llm = req.get("llm", {})
    if "model_name" in llm:
        req["llm_id"] = llm.pop("model_name")
    req["llm_setting"] = llm
    
    # 4. 处理 Prompt 配置（字段映射）
    prompt = req.get("prompt", {})
    key_mapping = {
        "parameters": "variables",
        "prologue": "opener",
        "quote": "show_quote",
        "system": "prompt"
    }
    for new_key, old_key in key_mapping.items():
        if old_key in prompt:
            prompt[new_key] = prompt.pop(old_key)
    req["prompt_config"] = prompt
    
    # 5. 校验 Prompt 变量
    for p in req["prompt_config"]["parameters"]:
        if not p["optional"] and req["prompt_config"]["system"].find("{%s}" % p["key"]) < 0:
            return get_error_data_result(message=f"Parameter '{p['key']}' is not used")
    
    # 6. 保存到数据库
    req["id"] = get_uuid()
    req["tenant_id"] = tenant_id
    req["kb_ids"] = ids
    if not DialogService.save(**req):
        return get_error_data_result(message="Fail to new a chat!")
    
    return get_result(data=req)

{
    "conversation_id": "conv_uuid",  // 可选，不提供则创建新会话
    "question": "RAGFlow 如何部署？",
    "stream": true  // 是否流式返回
}

{
    "code": 0,
    "message": "success",
    "data": {
        "answer": "RAGFlow 部署步骤如下：\n1. 安装 Docker...",
        "reference": [
            {
                "doc_id": "doc_uuid",
                "doc_name": "部署文档.pdf",
                "chunk_id": "chunk_uuid",
                "content": "...部署步骤...",
                "similarity": 0.85
            }
        ],
        "conversation_id": "conv_uuid"
    }
}

data: {"answer": "RAGFlow", "reference": []}
data: {"answer": " 部署", "reference": []}
data: {"answer": "步骤如下", "reference": []}
...
data: {"answer": "", "reference": [...]}  // 最后一条包含完整 reference

@manager.route("/chats/<chat_id>/completions", methods=["POST"])
@token_required
def completions(tenant_id, chat_id):
    """对话补全"""
    req = request.json
    
    # 1. 校验对话权限
    if not DialogService.query(tenant_id=tenant_id, id=chat_id):
        return get_error_data_result(message="You do not own the chat")
    
    # 2. 获取或创建会话
    conversation_id = req.get("conversation_id")
    if not conversation_id:
        conversation_id = get_uuid()
        ConversationService.save(
            id=conversation_id,
            dialog_id=chat_id,
            name=req["question"][:50],  # 首轮问题作为会话名
            message=[],
            reference=[]
        )
    
    # 3. 调用 RAG 引擎
    from rag.app.conversation import chat
    dialog = DialogService.get_by_id(chat_id)
    conv = ConversationService.get_by_id(conversation_id)
    
    if req.get("stream"):
        # 流式返回（SSE）
        def generate():
            for chunk in chat(
                dialog=dialog[1],
                messages=conv[1].message,
                question=req["question"],
                stream=True,
                **dialog[1].to_dict()
            ):
                yield f"data: {json.dumps(chunk)}\n\n"
        return Response(generate(), mimetype="text/event-stream")
    else:
        # 非流式返回
        answer, reference = chat(
            dialog=dialog[1],
            messages=conv[1].message,
            question=req["question"],
            stream=False,
            **dialog[1].to_dict()
        )
        
        # 4. 保存对话记录
        conv[1].message.append({"role": "user", "content": req["question"]})
        conv[1].message.append({"role": "assistant", "content": answer})
        conv[1].reference = reference
        conv[1].save()
        
        return get_result(data={
            "answer": answer,
            "reference": reference,
            "conversation_id": conversation_id
        })

graph TB
    Client[客户端] -->|1. POST /v1/user/login| UserApp[user_app.py::login]
    
    UserApp -->|2. request.json.get| RequestData{解析请求体}
    RequestData -->|3. email| UserService1[UserService.query<br/>查询用户是否存在]
    UserService1 -->|4. SELECT| MySQL1[(MySQL)]
    
    RequestData -->|5. password| DecryptFunc[decrypt 函数<br/>RSA 解密密码]
    
    UserService1 -->|6. 用户存在| UserService2[UserService.query_user<br/>校验密码]
    UserService2 -->|7. check_password_hash| PasswordCheck{密码校验}
    
    PasswordCheck -->|8. 密码正确| GenerateToken[get_uuid<br/>生成新 access_token]
    GenerateToken -->|9. login_user| FlaskLogin[Flask-Login<br/>写入 Session]
    FlaskLogin -->|10. Session.set| Redis1[(Redis)]
    
    GenerateToken -->|11. user.save| MySQL2[(MySQL)]
    MySQL2 -->|12. UPDATE user| UpdateToken[更新 access_token<br/>和 update_time]
    
    UpdateToken -->|13. to_json| ResponseData[构造响应数据]
    ResponseData -->|14. 200 OK| Client

    style UserApp fill:#e1f5ff
    style UserService1 fill:#fff4e1
    style UserService2 fill:#fff4e1
    style DecryptFunc fill:#ffe1f5
    style FlaskLogin fill:#e1ffe1
    style Redis1 fill:#f0f0f0
    style MySQL1 fill:#f0f0f0
    style MySQL2 fill:#f0f0f0

# api/apps/user_app.py: 67-138
@manager.route("/login", methods=["POST"])
def login():
    # 步骤 1: 获取请求数据
    email = request.json.get("email", "")
    password = request.json.get("password")
    
    # 步骤 2: 查询用户（调用 Service 层）
    users = UserService.query(email=email)
    if not users:
        return get_json_result(code=settings.RetCode.AUTHENTICATION_ERROR, 
                               message=f"Email: {email} is not registered!")
    
    # 步骤 3: 解密密码（调用工具函数）
    password = decrypt(password)  # RSA 私钥解密
    
    # 步骤 4: 验证密码（调用 Service 层）
    user = UserService.query_user(email, password)
    
    # 步骤 5: 生成新 Token 并登录
    user.access_token = get_uuid()  # 生成 UUID Token
    login_user(user)  # Flask-Login 写入 Session（存储到 Redis）
    
    # 步骤 6: 更新数据库
    user.update_time = current_timestamp()
    user.save()  # Peewee ORM 执行 UPDATE
    
    # 步骤 7: 返回响应
    return construct_response(data=user.to_json(), auth=user.get_id())

# api/db/services/user_service.py
class UserService(CommonService):
    model = User  # Peewee Model
    
    @classmethod
    @DB.connection_context()
    def query(cls, **kwargs):
        """查询用户（步骤 2）"""
        # 调用 Peewee ORM 查询
        users = cls.model.select().where(
            cls.model.email == kwargs.get('email')
        ).execute()
        return list(users)
    
    @classmethod
    def query_user(cls, email, password):
        """验证用户密码（步骤 4）"""
        users = cls.query(email=email, status=StatusEnum.VALID.value)
        if not users:
            return None
        user = users[0]
        # 使用 werkzeug.security 校验密码哈希
        if check_password_hash(user.password, password):
            return user
        return None

# api/db/db_models.py
class User(BaseModel):
    id = CharField(max_length=32, primary_key=True)
    email = CharField(max_length=128, unique=True)
    password = CharField(max_length=256, null=True)
    access_token = CharField(max_length=32, unique=True)
    nickname = CharField(max_length=128)
    avatar = CharField(max_length=1024, null=True)
    is_active = CharField(max_length=1, default='1')
    create_time = BigIntegerField()
    update_time = BigIntegerField()
    
    def save(self, *args, **kwargs):
        """保存用户（步骤 6）"""
        # Peewee 自动生成 SQL：
        # UPDATE user SET access_token=?, update_time=? WHERE id=?
        return super().save(*args, **kwargs)

# api/apps/__init__.py: 145-177
@login_manager.request_loader
def load_user(web_request):
    """从 Authorization Header 或 Session 加载用户"""
    authorization = web_request.headers.get("Authorization")
    if authorization:
        # Token 认证（SDK 调用）
        jwt = Serializer(secret_key=settings.SECRET_KEY)
        access_token = str(jwt.loads(authorization))
        user = UserService.query(access_token=access_token, 
                                 status=StatusEnum.VALID.value)
        return user[0] if user else None
    else:
        # Session 认证（Web UI），Flask-Login 自动从 Redis 读取
        return None

sequenceDiagram
    autonumber
    participant C as 客户端
    participant R as user_app::login
    participant US as UserService
    participant Crypt as decrypt 函数
    participant FL as Flask-Login
    participant M as MySQL
    participant Red as Redis

    C->>R: POST /v1/user/login<br/>{email, encrypted_password}
    
    R->>US: UserService.query(email=email)
    US->>M: SELECT * FROM user<br/>WHERE email=? AND status='1'
    M-->>US: User 记录
    US-->>R: users 列表
    
    alt 用户不存在
        R-->>C: 401 Unauthorized<br/>"Email not registered"
    end
    
    R->>Crypt: decrypt(password)
    Crypt->>Crypt: 使用 RSA 私钥解密
    Crypt-->>R: 明文密码
    
    R->>US: UserService.query_user(email, password)
    US->>US: check_password_hash(user.password, password)
    
    alt 密码错误
        US-->>R: None
        R-->>C: 401 Unauthorized<br/>"Password mismatch"
    end
    
    US-->>R: User 对象
    
    R->>R: user.access_token = get_uuid()
    R->>FL: login_user(user)
    FL->>Red: SET flask_session:<session_id><br/>{user_id: xxx}
    Red-->>FL: OK
    
    R->>R: user.update_time = now()
    R->>M: UPDATE user<br/>SET access_token=?, update_time=?<br/>WHERE id=?
    M-->>R: OK
    
    R->>R: response = user.to_json()
    R-->>C: 200 OK<br/>{id, email, access_token, ...}