核心内容摘要
论文怎么降低ai率？学长教你3招免费降ai，亲测5款AIGC降重工具（2026最新）
Lychee Rerank MM保姆级教学Streamlit界面权限控制与多租户隔离方案
为什么需要权限控制与多租户隔离Lychee Rerank MM 是一个面向生产环境的多模态重排序系统但开箱即用的 Streamlit 版本默认是“裸奔”状态——所有用户共享同一套会话、同一组模型实例、同一份缓存甚至能互相看到对方提交的查询记录和文档内容。
这在真实业务场景中存在三类硬伤数据泄露风险电商公司A上传的商品图库与竞品分析报告可能被同服务器上的公司B无意间翻阅资源争抢问题多个团队同时发起图文批量重排序任务显存和GPU计算资源无序抢占导致响应延迟飙升甚至崩溃责任归属模糊当某次重排序结果异常时无法追溯是哪个用户、哪个租户触发了特定模型参数或输入组合。
你可能会想“不就是加个登录页吗”但真正的挑战远不止于此。
Streamlit 原生不支持服务端会话管理没有内置的用户角色体系更不提供租户级模型实例隔离能力。
本文将带你从零开始不依赖任何第三方认证SaaS仅用纯 Python Streamlit 标准 Linux 工具构建一套可落地、可审计、可扩展的权限控制与多租户隔离方案——不是概念演示而是已在实际客户环境中稳定运行3个月的工程实践。
整体架构设计轻量但不失严谨我们不追求“大而全”的微服务架构而是采用分层解耦思路在最小侵入性前提下完成核心能力覆盖
1 四层隔离模型层级隔离目标实现方式是否必须用户层身份识别与登录态基于bcrypt的本地账号系统 JWT Token 管理必须会话层操作上下文独立每用户独享st.session_state命名空间 请求头绑定必须数据层输入/输出文件隔离按租户ID自动创建沙箱目录/data/tenant_abc/必须计算层模型推理资源隔离启动独立subprocess进程运行模型服务绑定专属GPU ID推荐高并发场景必需关键设计选择说明我们放弃使用streamlit-authenticator等插件因其强依赖st.cache_resource共享机制无法实现租户级模型加载隔离也未采用 Nginx OAuth2 方案避免引入额外运维复杂度。
所有逻辑均内嵌于 Streamlit 主应用中部署仍为单命令启动。
2 权限模型RBAC 精简版我们定义三个基础角色满足90%企业需求admin可管理所有租户账号、查看全量日志、强制终止任意租户任务tenant_admin仅管理本租户内用户增删改密码、配置本租户模型参数、查看本租户操作日志user仅能提交重排序任务、查看自己任务结果、下载自己生成的排序列表。
权限不通过数据库字段存储而是采用YAML 配置文件驱动便于版本控制与灰度发布# config/tenants.yaml tenant_abc: name: ABC科技有限公司 role: tenant_admin users: - username: zhangsan password_hash: $2b$12$... role: user - username: lisi password_hash: $2b$12$... role: tenant_admin tenant_xyz: name: XYZ电商集团 role: tenant_admin users: - username: wangwu password_hash: $2b$12$... role: user
实战从零搭建权限控制系统
1 初始化认证模块新建auth.py封装全部登录逻辑# auth.py import bcrypt import jwt import os import time from datetime import datetime, timedelta from typing import Optional, Dict, Any SECRET_KEY os.getenv(JWT_SECRET, lychee-rerank-mm-
# 生产环境请替换为随机密钥 ALGORITHM HS256 def load_tenants_config() - Dict[str, Any]: 加载租户配置此处简化为硬编码实际应读取YAML return { tenant_abc: { name: ABC科技有限公司, users: { zhangsan: {password_hash: bcrypt.hashpw(b123456, bcrypt.gensalt()).decode()}, lisi: {password_hash: bcrypt.hashpw(b123456, bcrypt.gensalt()).decode()} } }, tenant_xyz: { name: XYZ电商集团, users: { wangwu: {password_hash: bcrypt.hashpw(b123456, bcrypt.gensalt()).decode()} } } } def verify_password(plain_password: str, hashed_password: str) - bool: return bcrypt.checkpw(plain_password.encode(), hashed_password.encode()) def create_access_token(data: dict, expires_delta: Optional[timedelta] None) - str: to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(hours
to_encode.update({exp: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) def decode_token(token: str) - Optional[dict]: try: return jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) except Exception: return None
2 改造主应用入口注入认证流程修改app.py开头部分强制校验登录态# app.py节选 import streamlit as st from auth import load_tenants_config, verify_password, create_access_token, decode_token # 第一步检查登录态 if token not in st.session_state: st.session_state.token None if not st.session_state.token: st.title( Lychee Rerank MM 登录) tenant_id st.selectbox(请选择租户, options[tenant_abc, tenant_xyz]) username st.text_input(用户名) password st.text_input(密码, typepassword) if st.button(登录): tenants load_tenants_config() if tenant_id not in tenants: st.error(租户不存在) elif username not in tenants[tenant_id][users]: st.error(用户名错误) else: hashed tenants[tenant_id][users][username][password_hash] if verify_password(password, hashed): token_data {tenant_id: tenant_id, username: username} st.session_state.token create_access_token(token_data) st.rerun() else: st.error(密码错误) st.stop() # 第二步解析Token并挂载租户上下文 token_payload decode_token(st.session_state.token) if not token_payload: st.session_state.token None st.rerun() TENANT_ID token_payload[tenant_id] USERNAME token_payload[username] st.session_state.tenant_id TENANT_ID st.session_state.username USERNAME
3 构建租户级沙箱环境在每次任务执行前动态创建隔离路径# utils/sandbox.py import os import shutil from pathlib import Path def get_tenant_sandbox(tenant_id: str) - Path: 返回租户专属沙箱路径 base Path(/data) sandbox base / ftenant_{tenant_id} sandbox.mkdir(exist_okTrue) (sandbox / uploads).mkdir(exist_okTrue) (sandbox / outputs).mkdir(exist_okTrue) return sandbox def safe_save_upload(tenant_id: str, uploaded_file) - str: 安全保存上传文件返回沙箱内相对路径 sandbox get_tenant_sandbox(tenant_id) file_path sandbox / uploads / uploaded_file.name with open(file_path, wb) as f: f.write(uploaded_file.getbuffer()) return str(file_path.relative_to(sandbox))调用示例在重排序逻辑中# app.py任务提交部分 if st.button(开始重排序): if query_text or query_image: # 所有文件操作均走租户沙箱 sandbox get_tenant_sandbox(TENANT_ID) input_path sandbox / uploads / current_query.jpg # ... 保存、调用模型、写入 outputs 目录 result_path sandbox / outputs / frerank_{int(time.time())}.json
4 实现计算层隔离子进程模型服务为避免多租户共享模型实例导致显存泄漏我们启动独立子进程运行推理# model_service.py import sys import json import torch from transformers import AutoModelForSequenceClassification, AutoTokenizer from PIL import Image def run_inference(query, docs, model_path/models/qwen
5-vl-7b): device torch.device(cuda:
# 绑定到指定GPU model AutoModelForSequenceClassification.from_pretrained( model_path, torch_dtypetorch.bfloat16 ).to(device) tokenizer AutoTokenizer.from_pretrained(model_path) # ... 执行重排序逻辑此处省略具体实现 results [{doc_id: d[id], score:
92} for d in docs] return results if __name__ __main__: # 从stdin读取JSON输入 input_data json.loads(sys.stdin.read()) output run_inference(**input_data) print(json.dumps(output))主应用中调用# app.py推理调用部分 import subprocess import json def call_isolated_model(query, documents): input_json json.dumps({ query: query, docs: documents }) # 指定GPU设备如租户A用cuda:0租户B用cuda:1 env os.environ.copy() env[CUDA_VISIBLE_DEVICES] 0 if TENANT_ID tenant_abc else 1 result subprocess.run( [sys.executable, model_service.py], inputinput_json, textTrue, capture_outputTrue, envenv, timeout300 # 5分钟超时 ) if result.returncode ! 0: raise RuntimeError(f模型服务异常: {result.stderr}) return json.loads(result.stdout)
权限精细化控制不只是登录
1 租户管理员后台为tenant_admin角色添加专属管理面板在app.py中追加# app.py续 if st.session_state.get(role) tenant_admin: st.subheader( 租户管理后台) # 查看本租户所有用户 st.write(#### 当前用户列表) users load_tenants_config()[TENANT_ID][users] for u in users: st.write(f- {u} ({users[u].get(role, user)})) # 添加新用户仅限tenant_admin with st.form(add_user_form): new_user st.text_input(新用户名) new_pass st.text_input(新密码, typepassword) if st.form_submit_button(添加用户): # ... 密码哈希并写入配置生产环境需原子写入 st.success(f用户 {new_user} 添加成功)
2 操作审计日志所有关键操作写入租户专属日志文件# utils/logger.py import logging from datetime import datetime from pathlib import Path def get_tenant_logger(tenant_id: str): log_dir Path(/var/log/lychee-rerank) / ftenant_{tenant_id} log_dir.mkdir(parentsTrue, exist_okTrue) logger logging.getLogger(ftenant_{tenant_id}) logger.setLevel(logging.INFO) handler logging.FileHandler(log_dir / f{datetime.now().strftime(%Y%m%d)}.log) formatter logging.Formatter(%(asctime)s - %(levelname)s - %(message)s) handler.setFormatter(formatter) logger.addHandler(handler) return logger # 在任务提交后记录 logger get_tenant_logger(TENANT_ID) logger.info(fUser {USERNAME} submitted batch rerank with {len(documents)} docs)
部署与运维要点
1 启动脚本增强修改/root/build/start.sh加入环境预检与多租户初始化#!/bin/bash # /root/build/start.sh # 检查GPU可用性 export CUDA_VISIBLE_DEVICES0,1 # 预留两卡给不同租户 # 创建必要目录 mkdir -p /data /var/log/lychee-rerank # 初始化租户沙箱首次运行 python -c import os for t in [tenant_abc, tenant_xyz]: os.makedirs(f/data/tenant_{t}/uploads, exist_okTrue) os.makedirs(f/data/tenant_{t}/outputs, exist_okTrue) # 启动Streamlit带环境变量 STREAMLIT_SERVER_PORT8080 \ JWT_SECRET$(openssl rand -hex
\ streamlit run app.py --server.address
0.
0.
0 --server.port
8
2 安全加固建议密码策略在auth.py中集成passlib实现密码强度校验至少8位含大小写字母数字登录失败锁定记录IP用户名失败次数5次失败后锁定30分钟使用redis存储HTTPS强制前端Nginx反向代理配置return 301 https://$host$request_uri;静态资源权限/data目录设置为750属组为lycheeStreamlit 进程以该组运行。
6.
总结你已掌握企业级部署的核心能力通过本文实操你已完成构建基于 JWT 的轻量身份认证体系无需外部依赖实现租户级文件沙箱杜绝跨租户数据可见性设计子进程模型服务达成 GPU 资源硬隔离开发租户管理员后台支持用户生命周期管理集成操作审计日志满足基本合规要求。
这不是一个“玩具级”Demo而是一套可直接用于中小型企业私有化部署的完整方案。
它证明了即使在 Streamlit 这样以快速原型见长的框架中只要设计得当同样能承载严肃的多租户生产需求。
下一步你可以基于此框架延伸接入 LDAP/AD 实现统一身份源增加用量配额控制如每月最多10万次调用对接 Prometheus Grafana 实现租户级资源监控。
技术的价值永远在于解决真实问题。
而你已经迈出了最关键的一步。