Streamlit机器学习模型快速部署：零前端交付方案

1. 这不是又一个“部署教程”而是一套能立刻上线、被业务方点开就用的轻量级模型交付方案Streamlit 不是另一个需要配 Nginx、写 Dockerfile、搞反向代理、等 CI/CD 流水线跑完才能见人的“正经部署工具”。它是我过去三年在金融风控、电商推荐、医疗辅助三个垂直领域里反复验证过最省力、最抗压、最能让非技术同事真正“用起来”的模型落地入口。核心关键词就三个Streamlit、机器学习模型、快速部署——但重点不在“部署”这个词本身而在“让模型从 Jupyter Notebook 里走出来变成一个带按钮、能上传文件、会实时画图、还能被老板转发链接直接试用的网页应用”。它解决的从来不是“如何把模型塞进服务器”而是“怎么让数据科学家写的逻辑第一天就能被销售、运营、临床医生摸到、看懂、提反馈”。我见过太多团队花三周搭 Flask Vue 前后端最后上线的页面只有两个输入框和一个“预测”按钮而用户真正想要的是“把上周的客户名单 Excel 拖进来标红高风险客户顺便生成一页 PPT 报告”。Streamlit 就是干这个的它把 Python 数据处理、模型调用、可视化、交互控件全打包进一个.py文件里你streamlit run app.py一敲本地就起一个带热重载的开发服务器再用streamlit cloud一键发布或者扔进任意 Linux 服务器用nohup streamlit run app.py 守护进程跑着连域名都不用配——用户打开链接就是完整应用。它不替代 Kubernetes也不挑战 FastAPI 的高并发能力但它在“模型从实验室到业务一线”的最后一公里上效率碾压所有传统方案。适合谁适合手上有.pkl或.joblib训练好的 scikit-learn/XGBoost 模型、或 PyTorch/TensorFlow 导出的 ONNX 模型但没专职运维、没时间写前端、又急需让业务方看到效果的数据科学家也适合教学场景下学生交作业不再是.ipynb文件而是可点击、可交互、可分享的网页版模型演示。这不是玩具是我在某省级三甲医院部署的糖尿病视网膜病变初筛工具日均访问 200后台只跑着一台 2 核 4G 的阿里云轻量应用服务器至今没重启过。2. 整体设计思路为什么放弃 Flask/Django/FastAPI而选择 Streamlit 作为模型交付主干2.1 核心矛盾模型交付的本质不是“服务化”而是“可理解性交付”很多团队一说“部署模型”第一反应就是 REST API Swagger 文档 Postman 测试。这没错但错在起点——业务方根本不需要调 API。他们需要的是“我填个身份证号它告诉我信用分多少”“我拍张皮肤照片它圈出可能的病灶区域”“我上传 1000 行销售数据它标出下季度最可能断货的 SKU”。这些需求背后是输入方式多样化文本/文件/摄像头、输出形式具象化高亮/标注/图表/PDF、交互路径极简化单页完成无跳转。Flask 写一个上传 Excel 并返回结果的接口代码可能就 30 行但要让它支持拖拽上传、显示进度条、失败时给出中文错误提示、成功后自动生成带标题的柱状图并允许下载 PNG前端就得加 HTML/CSS/JS后端得处理文件流、内存管理、MIME 类型校验、临时文件清理……工作量指数级上升。而 Streamlit 天然内置了st.file_uploader、st.progress、st.pyplot、st.download_button你调用它们就像调用 Python 函数一样自然所有 UI 渲染、状态管理、前后端通信都由框架自动完成。我做过对比同样实现“上传 CSV → 模型预测 → 展示分类分布饼图 → 下载预测结果 CSV”Flask Bootstrap 方案总代码量 427 行含 HTML 模板 189 行Streamlit 方案仅 83 行纯 Python且全部逻辑集中在一个文件里。这不是偷懒是把工程师从“胶水代码”中解放出来专注在模型逻辑和业务规则上。2.2 架构选型单文件即应用彻底规避环境依赖地狱传统 Web 框架部署的核心痛点是环境隔离与依赖冲突。一个项目需要pandas1.3.5另一个需要pandas2.0.3系统级 Python 环境根本扛不住。Docker 能解决但代价是学习曲线陡峭、CI/CD 配置复杂、本地调试慢。Streamlit 的破局点在于它不要求你“部署一个服务”而是“运行一个脚本”。只要目标服务器装了 Python 3.8 和你的模型依赖scikit-learn,xgboost,torch等pip install streamlit后streamlit run app.py就能启动。它甚至自带依赖检查——当你import lightgbm但没装时页面会直接报红字错误而不是黑屏或 500。更关键的是Streamlit Cloud官方免费托管平台完全屏蔽了服务器概念你把代码推到 GitHub 仓库它自动检测requirements.txt构建环境运行streamlit run app.py给你分配一个https://yourname.streamlit.app/your-app的永久链接。我们团队给某银行做的反欺诈模型演示从本地写完到业务部门全员可用只用了 17 分钟Git Push → Streamlit Cloud 自动构建 → 分享链接。没有运维介入没有证书申请没有 DNS 解析等待。这种“零配置交付”能力在敏捷迭代场景下价值巨大——市场部今天说“想加个客户年龄段筛选”你改完st.multiselect两行代码Push 一下5 分钟后所有人看到新版。2.3 安全边界不碰生产核心只做“可信沙盒”前端有人质疑 Streamlit “不安全”担心它暴露内网模型或执行任意代码。这是误解。Streamlit 的设计哲学是“前端渲染层”而非“后端服务层”。它默认只监听localhost:8501不对外网开放所有模型加载、数据处理、预测计算都在 Python 进程内完成不涉及数据库直连、不调用 shell 命令除非你主动写os.system但那是你的责任不是框架漏洞。真正的安全实践是永远不在 Streamlit 应用里放敏感凭证、不连接生产数据库、不读取未授权文件路径。我们标准做法是——模型文件.pkl和预处理逻辑preprocessor.py放在应用同目录用joblib.load(model.pkl)加载原始数据由用户上传或从 S3/MinIO 等对象存储拉取通过boto3密钥走环境变量所有外部调用都封装在独立模块里Streamlit 脚本只负责 UI 和调度。这样即使 Streamlit 进程被攻破概率极低攻击者也只能拿到已加载的模型和当前会话的上传文件无法穿透到数据库或内网其他服务。它本质上是一个受控的、一次性的、无状态的“计算沙盒”比一个暴露/predict接口的 Flask 服务面更小、更易审计。我们所有面向客户的模型应用都部署在独立 VPC 子网内仅开放 8501 端口给公司内网完全符合金融行业等保三级对“应用层防护”的要求。3. 核心细节解析从模型加载到交互控件每个环节的实操要点与避坑指南3.1 模型加载别让joblib.load()成为性能瓶颈Streamlit 的默认行为是每次用户交互如点按钮、改滑块都会重新运行整个脚本。这意味着如果你把model joblib.load(model.pkl)写在脚本顶层每次用户上传新文件模型都会重新加载一遍——一个 500MB 的 XGBoost 模型加载耗时 3~5 秒体验直接崩坏。正确解法是st.cache_resource装饰器Streamlit 1.18 推荐旧版用st.cache。它告诉 Streamlit“这个对象创建代价高且不会随用户输入改变请缓存它后续复用”。import streamlit as st import joblib st.cache_resource def load_model(): return joblib.load(models/xgb_credit_risk_v3.pkl) # 在主逻辑中调用 model load_model() # 第一次调用加载之后所有会话共享同一实例提示st.cache_resource缓存的是对象本身如模型实例、数据库连接池而st.cache_data缓存的是函数返回值如pd.read_csv(data.csv)的 DataFrame。模型加载必须用st.cache_resource否则缓存失效。实测发现未加缓存时 10 次上传平均耗时 38.2 秒加缓存后首次加载 4.1 秒后续 9 次均为 0.03 秒内完成预测。另一个坑是模型路径硬编码。本地开发时model.pkl在当前目录但部署到 Streamlit Cloud 时工作目录是/app/your-repo/而模型文件可能在/app/your-repo/models/。务必用pathlib构建绝对路径from pathlib import Path MODEL_PATH Path(__file__).parent / models / xgb_credit_risk_v3.pkl st.cache_resource def load_model(): if not MODEL_PATH.exists(): st.error(f模型文件未找到{MODEL_PATH}) st.stop() return joblib.load(MODEL_PATH)3.2 输入处理文件上传、表单验证与数据清洗的无缝衔接业务方传来的数据永远是“脏”的Excel 有合并单元格、CSV 用分号分隔、列名大小写混乱、数值列混入字符串“N/A”。Streamlit 的st.file_uploader只负责接收二进制流剩下的全是你的事。关键技巧是把数据加载和清洗封装成带缓存的函数并在 UI 层做即时反馈。st.cache_data def load_and_clean_data(uploaded_file): if uploaded_file is None: return None try: # 自动识别文件类型 if uploaded_file.name.endswith(.csv): df pd.read_csv(uploaded_file, encodingutf-8) elif uploaded_file.name.endswith((.xls, .xlsx)): df pd.read_excel(uploaded_file, engineopenpyxl) else: st.error(仅支持 CSV 和 Excel 文件) return None # 基础清洗去空行、标准化列名 df df.dropna(howall) df.columns [col.strip().lower().replace( , _) for col in df.columns] # 关键字段存在性检查业务强相关 required_cols [customer_id, age, income] missing_cols [c for c in required_cols if c not in df.columns] if missing_cols: st.warning(f警告缺少必要字段 {missing_cols}将使用默认值填充) for c in missing_cols: df[c] 0 if c in [age, income] else UNKNOWN return df except Exception as e: st.error(f文件解析失败{str(e)}) return None # 在主界面调用 uploaded_file st.file_uploader(上传客户数据 (CSV/Excel), type[csv, xls, xlsx]) df load_and_clean_data(uploaded_file) if df is not None: st.success(f✅ 已加载 {len(df)} 条记录字段{list(df.columns)[:5]}...) st.dataframe(df.head(3)) # 显示前3行避免大数据集卡顿注意st.cache_data对uploaded_file的二进制内容做哈希缓存用户上传相同文件时load_and_clean_data不会重复执行。但若用户修改了文件内容再上传哈希值变函数重跑——这正是我们想要的。另外st.dataframe(df.head(3))是经验之谈直接st.dataframe(df)渲染万行数据会阻塞 UIhead()保证首屏秒开。3.3 模型预测状态管理、进度反馈与结果结构化呈现预测过程不能让用户盯着空白页面等。Streamlit 提供st.status1.22和st.progress两种方式。前者更现代支持嵌套步骤后者更直观适合长耗时任务。if st.button( 开始预测, typeprimary) and df is not None: # 创建状态容器支持多步反馈 with st.status(正在处理..., expandedTrue) as status: st.write(✅ 步骤 1数据预处理...) processed_df preprocess_data(df) # 自定义清洗函数 st.write(✅ 步骤 2模型加载中...) model load_model() # 复用缓存模型 st.write(✅ 步骤 3执行预测...) # 批量预测避免逐行调用慢 predictions model.predict(processed_df[FEATURE_COLS]) probabilities model.predict_proba(processed_df[FEATURE_COLS])[:, 1] st.write(✅ 步骤 4生成报告...) result_df pd.DataFrame({ customer_id: df[customer_id], risk_score: probabilities, risk_level: [高风险 if p 0.7 else 中风险 if p 0.3 else 低风险 for p in probabilities] }) status.update(label预测完成✅, statecomplete, expandedFalse) # 结果展示区 st.subheader( 预测结果概览) col1, col2 st.columns(2) with col1: st.metric(总客户数, len(result_df)) st.metric(高风险客户, len(result_df[result_df[risk_level]高风险])) with col2: fig, ax plt.subplots(figsize(5, 4)) result_df[risk_level].value_counts().plot(kindbar, axax) ax.set_title(风险等级分布) st.pyplot(fig) st.subheader( 详细结果) st.dataframe(result_df.style.background_gradient(subset[risk_score], cmapRdYlBu_r)) # 下载按钮 csv result_df.to_csv(indexFalse).encode(utf-8-sig) st.download_button( label 下载完整结果 (CSV), datacsv, file_namecredit_risk_prediction_result.csv, mimetext/csv )实操心得model.predict()必须传入二维数组n_samples x n_features切忌传入df[age]这样的 Series会报ValueError: Expected 2D array。务必用df[FEATURE_COLS]显式指定列顺序因为模型训练时特征顺序是固定的。FEATURE_COLS [age, income, loan_amount, ...]应定义为常量与训练脚本保持一致。4. 实操全流程从零开始搭建一个可商用的信贷风险评估应用4.1 环境准备与项目结构拒绝“脚本式混乱”拥抱工程化习惯哪怕只是一个 Streamlit 应用我也坚持清晰的项目结构。这不仅是为未来扩展更是为了降低协作门槛。我的标准目录如下credit-risk-app/ ├── app.py # 主应用入口只含 UI 逻辑 ├── models/ │ └── xgb_credit_risk_v3.pkl # 训练好的模型.pkl 或 .onnx ├── src/ │ ├── __init__.py │ ├── preprocessing.py # 数据清洗、特征工程函数 │ ├── features.py # 特征定义、列名映射 │ └── utils.py # 通用工具如日志、配置读取 ├── requirements.txt # 严格版本锁定 ├── README.md # 部署说明、业务逻辑简介 └── .streamlit/config.toml # Streamlit 配置可选requirements.txt是生命线必须精确到小版本streamlit1.32.0 pandas1.5.3 scikit-learn1.2.2 xgboost1.7.5 matplotlib3.7.1 joblib1.2.0 numpy1.23.5为什么锁死版本Streamlit 1.30 升级到 1.31 时st.experimental_rerun()行为变更导致我们一个应用出现无限重载循环pandas2.0 的read_csv默认引擎变更让某些 Excel 解析失败。线上环境绝不允许“最新版”这种模糊表述。pip install -r requirements.txt必须在任何环境都能复现完全一致的行为。4.2 核心应用代码app.py完整可运行的 127 行实现以下代码是经过生产验证的精简版已移除业务敏感字段保留全部关键模式import streamlit as st import pandas as pd import numpy as np import matplotlib.pyplot as plt import joblib from pathlib import Path import sys # 设置页面配置 st.set_page_config( page_title信贷风险智能评估, page_icon, layoutwide, initial_sidebar_stateexpanded ) # 1. 加载模型带缓存和错误处理 st.cache_resource def load_model(): model_path Path(__file__).parent / models / xgb_credit_risk_v3.pkl if not model_path.exists(): st.error(f❌ 模型文件缺失{model_path}) st.stop() try: model joblib.load(model_path) st.success(f✅ 模型加载成功XGBoost v3.2.1) return model except Exception as e: st.error(f❌ 模型加载失败{str(e)}) st.stop() # 2. 加载并清洗数据带缓存 st.cache_data def load_and_validate_data(uploaded_file): if uploaded_file is None: return None try: if uploaded_file.name.endswith(.csv): df pd.read_csv(uploaded_file, encodingutf-8) elif uploaded_file.name.endswith((.xls, .xlsx)): df pd.read_excel(uploaded_file, engineopenpyxl) else: st.error(❌ 仅支持 CSV 和 Excel 文件格式) return None # 强制转换关键数值列 for col in [age, income, loan_amount]: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) # 删除全空行 df df.dropna(howall) if len(df) 0: st.warning(⚠️ 上传文件为空或无有效数据) return None return df except Exception as e: st.error(f❌ 数据解析异常{str(e)}) return None # 3. 特征工程模拟实际应与训练脚本完全一致 def engineer_features(df): # 示例构造衍生特征 df df.copy() df[income_to_loan_ratio] df[income] / (df[loan_amount] 1e-6) df[age_group] pd.cut(df[age], bins[0, 25, 35, 45, 60, 100], labels[青年, 壮年, 中年, 中老年, 老年]) return df # 4. 主界面逻辑 st.title( 信贷风险智能评估系统) st.markdown( 本系统基于历史客户数据训练的 XGBoost 模型用于预测新客户违约风险。 **操作流程**上传客户数据 → 点击预测 → 查看风险分布与明细 → 下载结果。 ) # 侧边栏参数配置可选 with st.sidebar: st.header(⚙️ 配置选项) threshold st.slider(风险阈值, 0.0, 1.0, 0.5, 0.05, help高于此值判定为高风险) show_details st.checkbox(显示详细分析, valueTrue) # 主内容区 uploaded_file st.file_uploader( 上传客户信息表 (CSV/Excel), type[csv, xls, xlsx], help请确保包含customer_id, age, income, loan_amount 等字段 ) if uploaded_file is not None: df load_and_validate_data(uploaded_file) if df is not None and len(df) 0: st.info(f 已加载 {len(df)} 条客户记录检测到字段{list(df.columns)}) # 显示原始数据样例 with st.expander( 查看原始数据样例前5行): st.dataframe(df.head(5)) # 预测按钮 if st.button( 执行风险评估, typeprimary, use_container_widthTrue): model load_model() # 特征工程 processed_df engineer_features(df) # 定义训练时使用的特征列必须与训练脚本完全一致 FEATURE_COLS [age, income, loan_amount, income_to_loan_ratio] # 过滤缺失值 valid_mask processed_df[FEATURE_COLS].notna().all(axis1) valid_df processed_df[valid_mask].copy() if len(valid_df) 0: st.error(❌ 所有记录均缺少必要特征无法预测) else: # 执行预测 y_pred_proba model.predict_proba(valid_df[FEATURE_COLS])[:, 1] # 构建结果 result_df pd.DataFrame({ customer_id: valid_df[customer_id].values, risk_score: y_pred_proba, risk_level: [高风险 if s threshold else 低风险 for s in y_pred_proba] }) # 展示结果 st.subheader( 风险评估结果) col1, col2, col3 st.columns(3) col1.metric(总评估数, len(result_df)) col2.metric(高风险客户, len(result_df[result_df[risk_level]高风险])) col3.metric(平均风险分, f{result_df[risk_score].mean():.3f}) # 分布图 fig, ax plt.subplots(1, 2, figsize(12, 4)) result_df[risk_level].value_counts().plot(kindbar, axax[0], title风险等级分布) result_df[risk_score].hist(bins20, axax[1], title风险分分布) st.pyplot(fig) # 详细表格 if show_details: st.subheader( 完整预测明细) # 高亮高风险行 styled_df result_df.style.apply( lambda x: [background-color: #ffebee if v高风险 else for v in x], subset[risk_level] ) st.dataframe(styled_df) # 下载 csv result_df.to_csv(indexFalse).encode(utf-8-sig) st.download_button( label 下载预测结果 (CSV), datacsv, file_namefcredit_risk_result_{pd.Timestamp.now().strftime(%Y%m%d_%H%M%S)}.csv, mimetext/csv ) else: st.warning(⚠️ 数据加载失败请检查文件格式和内容) else: st.info( 请先在左侧上传客户数据文件)4.3 本地开发与调试热重载、状态调试与性能监控Streamlit 的--dev模式是开发利器。启动命令streamlit run app.py --server.port8501 --server.addresslocalhost --theme.baselight --logger.leveldebug--server.port: 指定端口避免冲突--theme.base: 强制浅色主题保护眼睛--logger.leveldebug: 输出详细日志定位st.cache命中/未命中情况调试核心技巧查看缓存状态在浏览器地址栏加?showSidebartrue右上角菜单 →Settings→Developer tools→Cache实时看到哪些函数被缓存、缓存大小、命中率。强制重载按CtrlR刷新页面或在终端按r键Streamlit CLI 支持。状态调试用st.session_state查看当前会话变量。例如在app.py末尾加# 调试用显示当前会话状态 # st.write(DEBUG session_state:, st.session_state)然后在 UI 上操作观察哪些变量被创建、更新。性能监控Streamlit 内置--server.enableCORSfalse和--server.maxUploadSize1000单位 MB可调优。对于大文件上传建议在config.toml中设置[server] enableCORS false maxUploadSize 500 # 允许跨域仅开发时 # enableCORS true5. 常见问题与排查技巧实录那些文档里不会写的血泪教训5.1 模型预测结果不一致检查特征顺序与缺失值处理这是最高频的线上事故。现象本地测试predict()结果和线上部署结果不同。根因几乎 100% 是特征列顺序不一致或缺失值填充策略不同。排查步骤在app.py中打印processed_df[FEATURE_COLS].columns.tolist()确认顺序在训练脚本中用print(X_train.columns.tolist())输出训练时的列顺序二者必须完全一致如果训练时用X_train df[[age,income,loan_amount]]则预测时也必须df[[age,income,loan_amount]]不能是df[[income,age,loan_amount]]检查缺失值训练时用SimpleImputer(strategymedian)则预测时也必须用同一个imputer实例缓存它不能用df.fillna(0)替代。我踩过的坑某次模型升级训练脚本新增了employment_length特征但app.py的FEATURE_COLS没同步更新导致model.predict()传入的数组少了一列XGBoost 报ValueError: feature_names mismatch。解决方案在app.py开头加断言assert len(FEATURE_COLS) model.n_features_in_, f特征数量不匹配期望{model.n_features_in_}实际{len(FEATURE_COLS)}5.2 页面白屏/加载失败优先检查静态资源与路径Streamlit 1.20 默认禁用st.image()加载本地图片出于安全但很多教程仍教st.image(logo.png)。正确方式是# ✅ 正确用 pathlib 读取二进制 logo_path Path(__file__).parent / assets / logo.png if logo_path.exists(): st.image(str(logo_path), width120) # ✅ 或者用 st.markdown 插入 base64适合小图标 import base64 with open(logo_path, rb) as f: data base64.b64encode(f.read()).decode() st.markdown(fimg srcdata:image/png;base64,{data} width120/, unsafe_allow_htmlTrue)另一个常见原因是requirements.txt里漏了依赖。比如用了plotly.express画图但requirements.txt只写了plotly没写plotly5.18.0Streamlit Cloud 构建时会装最新版而px.histogram()在 6.0 有 API 变更。永远用pip freeze requirements.txt在干净虚拟环境中生成依赖列表。5.3 Streamlit Cloud 部署失败看构建日志不是看应用日志Streamlit Cloud 的构建失败Build Failed和应用崩溃App Crashed是两回事。前者发生在pip install和streamlit run启动前后者发生在应用运行中。Build Failed 常见原因requirements.txt里有githttps://github.com/user/repo.git这种私有库地址而 Streamlit Cloud 无权限访问包名拼写错误如sklearn错应为scikit-learn对使用了conda专属包如cudatoolkit而 Streamlit Cloud 只支持pip。App Crashed 常见原因模型文件路径错误joblib.load()报FileNotFoundErrorst.file_uploader返回None后续代码未判空直接.shape触发AttributeError内存溢出上传 1GB Excelpd.read_excel()直接 OOM。解决方案在load_and_validate_data中加大小限制if uploaded_file.size 100 * 1024 * 1024: # 100MB st.error(❌ 文件过大100MB请压缩或分批上传) return None5.4 性能优化实战从 8 秒到 0.8 秒的预测提速某次上线后用户反馈“点预测按钮要等很久”。st.status显示步骤 3模型预测耗时 7.2 秒。排查发现模型是 XGBoost但predict_proba()调用的是 CPU 版本而服务器有 GPUprocessed_df有 5 万行但model.predict()默认是单线程。优化方案启用 GPU 加速XGBoost# 训练时保存为 GPU 模型 model xgb.XGBClassifier(tree_methodgpu_hist, gpu_id0) # 预测时自动使用 GPU批量预测并行化from sklearn.utils import parallel_backend from joblib import parallel_config # 使用 joblib 并行需模型支持 with parallel_config(threading, n_jobs4): y_pred_proba model.predict_proba(processed_df[FEATURE_COLS])[:, 1]最有效的一招提前采样。业务方其实只需要看“高风险 Top 100”不需要全量预测。加一个开关sample_size st.number_input(预测样本数0全量, min_value0, value1000) if sample_size 0 and len(processed_df) sample_size: processed_df processed_df.sample(nsample_size, random_state42)实测5 万行数据优化后预测耗时从 7.2 秒降至 0.78 秒用户体验质变。6. 进阶扩展不止于单页应用构建可持续演进的模型交付体系6.1 多页面应用用pages/目录管理复杂业务流当一个应用承载多个模型如信用分 反欺诈 营销响应单文件会失控。Streamlit 1.12 支持原生多页my-app/ ├── streamlit_app.py # 主入口可为空 ├── pages/ │ ├── 1__Home.py │ ├── 2__Credit_Score.py │ ├── 3_️_Fraud_Detection.py │ └── 4__Analytics_Dashboard.py每个pages/*.py是独立页面自动在左侧导航栏生成菜单。1__Home.py开头加import streamlit as st st.set_page_config(page_title首页, page_icon) st.title(欢迎来到模型中心)优势页面间状态隔离st.session_state不共享路由清晰团队可并行开发不同页面。6.2 与现有系统集成通过st.experimental_connection连接数据库和 APIStreamlit 1.22 的experimental_connection是连接外部系统的安全通道# 连接 PostgreSQL需安装 psycopg2-binary conn st.experimental_connection(postgresql, typesql) df conn.query(SELECT * FROM customers WHERE risk_score 0.8 LIMIT 100;) # 连接 REST API需安装 requests from streamlit.connections import ExperimentalBaseConnection class MyAPIDBConnection(ExperimentalBaseConnection[requests.Session]): def _connect(self, **kwargs) - requests.Session: session requests.Session() session.headers.update({Authorization: fBearer {self._secrets[api_key]}}) return session def fetch_data(self, endpoint: str): return self._instance.get(fhttps://api.example.com/{endpoint}).json() conn st.experimental_connection(my_api, typeMyAPIDBConnection, api_keyxxx) data conn.fetch_data(customers/high_risk)注意密钥必须存入 Streamlit Cloud 的SecretsSettings →