如何将Streamlit与服务器端Jupyter Lab Notebook连接并实现交互？

下面是几个可行的落地方案，根据你的需求和服务器环境选择即可：

方案1：用共享文件做数据中转（最简单无依赖） #

核心思路是借助服务器上的共享文件作为中间介质，Streamlit写入输入数据，Notebook监听文件变化后执行处理逻辑，再把结果写回文件，最后Streamlit读取结果。

Streamlit 端代码 #

import streamlit as st
import json
import time

# 获取用户输入
user_input = st.text_input("输入需要处理的数据")
if st.button("提交处理"):
    # 将输入数据写入JSON文件（替换为你服务器上的共享路径）
    with open("/opt/shared/data_input.json", "w") as f:
        json.dump({"input": user_input}, f)

# 轮询读取处理结果
if st.button("获取结果") or st.session_state.get("polling", False):
    st.session_state["polling"] = True
    try:
        with open("/opt/shared/data_output.json", "r") as f:
            result = json.load(f)
        st.success(f"处理结果：{result['output']}")
        st.session_state["polling"] = False
    except FileNotFoundError:
        st.info("等待Notebook处理中...")
        time.sleep(2)
        st.experimental_rerun()

Jupyter Notebook 端代码 #

import json
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 你的处理函数集合
def func1(data):
    return data.upper()

def func2(data):
    return data + "_processed"

def main_process(input_data):
    step1 = func1(input_data)
    step2 = func2(step1)
    return step2

# 监听输入文件变化的处理器
class InputFileHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path == "/opt/shared/data_input.json":
            # 读取输入数据
            with open(event.src_path, "r") as f:
                input_data = json.load(f)["input"]
            # 执行处理逻辑
            result = main_process(input_data)
            # 写入结果文件
            with open("/opt/shared/data_output.json", "w") as f:
                json.dump({"output": result}, f)

# 启动文件监听
event_handler = InputFileHandler()
observer = Observer()
observer.schedule(event_handler, path="/opt/shared/", recursive=False)
observer.start()

# 保持Notebook运行状态
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

注意：需先安装文件监听依赖：pip install watchdog，共享路径要确保Streamlit和Jupyter进程都拥有读写权限。

方案2：用Redis做实时内存缓存（适合高实时性场景） #

如果需要更快的响应速度，可利用Redis作为中间缓存，Streamlit和Notebook通过Redis的键值对传递数据，避免文件IO的延迟。

准备工作 #

在服务器上安装Redis并启动服务，然后安装Python客户端：pip install redis

Streamlit 端代码 #

import streamlit as st
import redis
import time

# 连接Redis（根据你的Redis配置调整参数）
r = redis.Redis(host="localhost", port=6379, db=0)

user_input = st.text_input("输入需要处理的数据")
if st.button("提交处理"):
    # 生成唯一任务ID，避免冲突
    task_id = f"task_{int(time.time())}"
    r.set(task_id, user_input)
    st.session_state["current_task"] = task_id

# 轮询获取处理结果
if "current_task" in st.session_state:
    task_id = st.session_state["current_task"]
    result = r.get(f"{task_id}_result")
    if result:
        st.success(f"处理结果：{result.decode('utf-8')}")
        # 清理缓存数据
        r.delete(task_id)
        r.delete(f"{task_id}_result")
        del st.session_state["current_task"]
    else:
        st.info("处理中...")
        time.sleep(1)
        st.experimental_rerun()

Jupyter Notebook 端代码 #

import redis
import time

# 你的处理函数（和方案1一致）
def func1(data):
    return data.upper()

def func2(data):
    return data + "_processed"

def main_process(input_data):
    step1 = func1(input_data)
    step2 = func2(step1)
    return step2

# 连接Redis
r = redis.Redis(host="localhost", port=6379, db=0)

# 轮询监听新任务
while True:
    # 匹配所有task_开头的任务键
    task_keys = r.keys("task_*")
    for key in task_keys:
        task_id = key.decode('utf-8')
        input_data = r.get(key).decode('utf-8')
        # 执行处理逻辑
        result = main_process(input_data)
        # 写入结果缓存
        r.set(f"{task_id}_result", result)
    time.sleep(0.5)

方案3：把Notebook逻辑封装成API（最规范的生产级方案） #

将Notebook中的处理函数提取为独立的Python模块，用FastAPI编写HTTP接口，Streamlit通过发送HTTP请求调用接口获取结果，这是长期维护项目的首选方案。

第一步：提取处理函数到单独文件（processor.py） #

# processor.py
def func1(data):
    return data.upper()

def func2(data):
    return data + "_processed"

def main_process(input_data):
    step1 = func1(input_data)
    step2 = func2(step1)
    return step2

第二步：编写FastAPI接口（api.py） #

from fastapi import FastAPI
from pydantic import BaseModel
import processor

app = FastAPI()

# 定义输入数据结构
class InputData(BaseModel):
    input: str

@app.post("/process")
def process_data(data: InputData):
    result = processor.main_process(data.input)
    return {"output": result}

第三步：启动API服务 #

在服务器上运行：uvicorn api:app --host 0.0.0.0 --port 8000

第四步：Streamlit 端调用API #

import streamlit as st
import requests

user_input = st.text_input("输入需要处理的数据")
if st.button("提交处理"):
    response = requests.post(
        "http://localhost:8000/process",
        json={"input": user_input}
    )
    if response.status_code == 200:
        result = response.json()["output"]
        st.success(f"处理结果：{result}")
    else:
        st.error("处理失败，请重试")

各方案适用场景：

• 方案1：适合快速测试验证，无需额外服务，缺点是实时性差、存在文件读写冲突风险
• 方案2：适合实时性要求高的场景，Redis性能优异，支持高频交互
• 方案3：适合长期维护的项目，接口规范易扩展，是生产环境的标准做法

内容的提问来源于stack exchange，提问作者shriya sharma