Bài 10: Streamlit — dashboard và data app

Sau bài này bạn sẽ:

• Hiểu Streamlit dùng hợp lý trong bài toán nào và bài toán nào nên dùng công cụ khác.
• Biết execution model của Streamlit và tại sao nó ảnh hưởng đến cách viết code.
• Dùng đúng st.session_state, @st.cache_data, @st.cache_resource.
• Xây được dashboard monitoring model và chatbot RAG có streaming.
• Biết cách tổ chức multi-page app.

Streamlit phù hợp với các trường hợp sau:

• Dashboard data — hiển thị chart, table, metric card, filter sidebar. Người dùng chọn điều kiện, dashboard cập nhật.
• Internal tool / data exploration — team data dùng để khám phá dataset, xem phân phối, debug feature.
• Monitoring model — theo dõi latency, error rate, drift qua thời gian với filter model version và date range.
• Demo multi-step workflow — người dùng upload file, cấu hình tham số, xem kết quả theo từng bước.
• Chatbot / conversational app — có lịch sử hội thoại nhiều turn, streaming token.

Khi nào không dùng Streamlit:

• Chỉ cần một form đơn giản nhận input và trả output — Gradio (bài 9) phù hợp hơn vì ít boilerplate hơn.
• Cần deploy public API endpoint cho client khác gọi — dùng FastAPI (Module 1).
• Cần giao diện web thực sự với auth, routing phức tạp, custom CSS nhiều — Next.js hoặc một web framework đầy đủ hơn.

pip install streamlit          # Streamlit 1.x, current 1.36+
streamlit --version            # kiểm tra version

Tạo file app.py:

import streamlit as st

st.title("Hello Streamlit")
st.write("Đây là app đầu tiên.")

name = st.text_input("Tên bạn là gì?")
if name:
    st.write(f"Xin chào, {name}!")

Chạy:

streamlit run app.py

Streamlit tự mở http://localhost:8501 trên trình duyệt. Hot-reload tự động khi lưu file.

Cấu trúc tối thiểu: viết Python tuần tự từ trên xuống dưới. Mỗi st.xxx() call render ra một UI element theo đúng thứ tự trong code. Không có template, không có router, không có callback phức tạp ở giai đoạn này.

Đây là điểm quan trọng nhất khi làm việc với Streamlit, và là nguồn gốc của phần lớn bug mà người mới mắc phải.

Mỗi lần user thao tác — click button, thay đổi slider, gõ vào text input — Streamlit chạy lại toàn bộ script app.py từ đầu đến cuối.

Điều này có nghĩa:

• Biến Python thông thường bị reset về giá trị ban đầu sau mỗi lần rerun.
• Model hoặc dữ liệu load bên ngoài cache sẽ bị load lại mỗi click — rất chậm.
• Logic phải được thiết kế cho việc chạy lại, không phải chạy 1 lần.

Giữ state với st.session_state

st.session_state là dict tồn tại xuyên suốt các lần rerun trong cùng một browser session:

import streamlit as st

# Sai: biến thông thường bị reset mỗi rerun
# count = 0  # luôn = 0 sau mỗi click

# Đúng: dùng session_state
if "count" not in st.session_state:
    st.session_state.count = 0

if st.button("Tăng"):
    st.session_state.count += 1

st.write(f"Đã click: {st.session_state.count} lần")

Cache để tránh tính toán lặp

Khi script chạy lại, các function được đánh dấu cache sẽ không chạy lại nếu argument không đổi — Streamlit trả về kết quả từ cache. Chi tiết ở mục 9.

Text

st.title("Tiêu đề lớn")
st.header("Header")
st.subheader("Subheader")
st.write("Bất kỳ object nào: string, dict, DataFrame, ...")
st.markdown("**Bold**, *italic*, `code`")
st.code("print('hello')", language="python")

Input widgets

name = st.text_input("Tên", placeholder="Nhập tên...")
number = st.number_input("Số", min_value=0, max_value=100, value=50)
slider_val = st.slider("Threshold", 0.0, 1.0, 0.5, step=0.01)
option = st.selectbox("Model version", ["v1.0", "v1.1", "v2.0"])
versions = st.multiselect("Chọn nhiều", ["v1.0", "v1.1", "v2.0"])
uploaded = st.file_uploader("Upload CSV", type=["csv"])
user_msg = st.chat_input("Nhập tin nhắn...")

Display data

import pandas as pd

df = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
st.dataframe(df)              # interactive, sortable
st.table(df)                  # static table
st.line_chart(df)
st.bar_chart(df)
st.image("path/to/image.png", caption="Caption")

# Metric card
st.metric(label="Avg latency", value="142 ms", delta="-8 ms")

Layout

# Columns
col1, col2, col3 = st.columns(3)
with col1:
    st.metric("Total requests", 12_450)
with col2:
    st.metric("Avg latency", "142 ms")
with col3:
    st.metric("Error rate", "0.3%")

# Tabs
tab1, tab2 = st.tabs(["Overview", "Details"])
with tab1:
    st.write("Nội dung tab 1")
with tab2:
    st.write("Nội dung tab 2")

# Expander
with st.expander("Xem chi tiết"):
    st.write("Nội dung ẩn khi collapsed")

# Sidebar
model_version = st.sidebar.selectbox("Model", ["v1", "v2"])
st.sidebar.slider("Confidence threshold", 0.0, 1.0, 0.5)

Dashboard đọc CSV log inference, hiển thị metric tổng quan, chart latency theo thời gian, và cho phép filter theo date range + model version.

Giả sử inference_log.csv có cột: timestamp, model_version, latency_ms, status_code.

# dashboard.py
import streamlit as st
import pandas as pd

st.set_page_config(page_title="Model Monitoring", layout="wide")
st.title("Model Monitoring Dashboard")

# --- Load data (cache để không đọc lại mỗi rerun) ---
@st.cache_data
def load_log(path: str) -> pd.DataFrame:
    df = pd.read_csv(path, parse_dates=["timestamp"])
    return df

df = load_log("inference_log.csv")

# --- Sidebar filter ---
with st.sidebar:
    st.header("Filters")
    versions = st.multiselect(
        "Model version",
        options=df["model_version"].unique().tolist(),
        default=df["model_version"].unique().tolist(),
    )
    date_range = st.date_input(
        "Date range",
        value=(df["timestamp"].min().date(), df["timestamp"].max().date()),
    )

# --- Apply filter ---
mask = (
    df["model_version"].isin(versions)
    & (df["timestamp"].dt.date >= date_range[0])
    & (df["timestamp"].dt.date <= date_range[1])
)
filtered = df[mask]

# --- Metrics row ---
total = len(filtered)
avg_lat = filtered["latency_ms"].mean() if total else 0
error_rate = (
    (filtered["status_code"] != 200).sum() / total * 100 if total else 0
)

col1, col2, col3 = st.columns(3)
col1.metric("Total requests", f"{total:,}")
col2.metric("Avg latency", f"{avg_lat:.0f} ms")
col3.metric("Error rate", f"{error_rate:.2f}%")

# --- Latency chart ---
st.subheader("Latency theo thời gian")
chart_df = (
    filtered.set_index("timestamp")["latency_ms"]
    .resample("1h")
    .mean()
    .reset_index()
    .rename(columns={"latency_ms": "avg_latency_ms"})
)
st.line_chart(chart_df.set_index("timestamp"))

# --- Raw data (limit display) ---
with st.expander("Raw log (hiển thị tối đa 500 dòng)"):
    st.dataframe(filtered.head(500))

Chạy: streamlit run dashboard.py

Lưu ý về @st.cache_data: function load_log chỉ đọc file 1 lần. Khi user kéo filter sidebar, Streamlit rerun script nhưng không gọi lại load_log vì argument path không đổi. Nếu muốn reload file (ví dụ file được ghi thêm), có thể thêm ttl: @st.cache_data(ttl=60) để cache hết hạn sau 60 giây.

Chatbot lưu lịch sử hội thoại trong session_state và stream token từ LLM về trực tiếp. @st.cache_resource đảm bảo model và vector store chỉ load 1 lần dù script chạy lại nhiều lần.

# chatbot.py  —  yêu cầu: streamlit>=1.30, openai>=1.0
import streamlit as st
from openai import OpenAI

st.set_page_config(page_title="RAG Chatbot")
st.title("RAG Chatbot")

# --- Load resource 1 lần, share cho mọi user session ---
@st.cache_resource
def get_openai_client() -> OpenAI:
    # key lấy từ st.secrets hoặc env var
    return OpenAI(api_key=st.secrets["OPENAI_API_KEY"])

@st.cache_resource
def get_vector_store():
    # Ví dụ: ChromaDB local
    import chromadb
    client = chromadb.PersistentClient(path="./chroma_db")
    return client.get_collection("docs")

openai_client = get_openai_client()
collection = get_vector_store()

# --- Khởi tạo lịch sử nếu chưa có ---
if "messages" not in st.session_state:
    st.session_state.messages = []

# --- Render lịch sử hội thoại ---
for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# --- Nhận input từ user ---
if prompt := st.chat_input("Hỏi gì đó..."):
    # Hiển thị tin nhắn user ngay
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    # Retrieve context từ vector store
    results = collection.query(query_texts=[prompt], n_results=3)
    context = "\n".join(results["documents"][0])

    system_msg = f"Trả lời dựa trên context sau:\n{context}"
    history = [{"role": "system", "content": system_msg}] + st.session_state.messages

    # Stream response từ LLM
    with st.chat_message("assistant"):
        stream = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=history,
            stream=True,
        )
        # st.write_stream nhận generator và stream trực tiếp lên UI
        response_text = st.write_stream(
            chunk.choices[0].delta.content or ""
            for chunk in stream
            if chunk.choices[0].delta.content
        )

    st.session_state.messages.append({"role": "assistant", "content": response_text})

Hai điểm quan trọng:

• st.write_stream(generator) (Streamlit 1.30+) nhận Python generator và hiển thị từng phần tử ngay khi generator yield — đây là cách duy nhất để stream text trong Streamlit. Dùng st.write() thông thường sẽ chờ toàn bộ response rồi mới render.
• @st.cache_resource đảm bảo OpenAI() client và ChromaDB collection chỉ khởi tạo 1 lần, dù Streamlit rerun script hàng trăm lần. Instance đó được share giữa mọi tab và mọi user session — phù hợp cho connection pool và model loader.

Cách 1: thư mục pages/ (từ Streamlit 1.10)

Đặt file Python vào thư mục pages/ — Streamlit tự tạo sidebar navigation:

my_app/
├── app.py                  # trang chính (Home)
└── pages/
    ├── 01_Dashboard.py     # Streamlit dùng tên file làm label nav
    └── 02_Chatbot.py

Chạy: streamlit run app.py. Sidebar sẽ tự có link đến "Dashboard" và "Chatbot".

Quy tắc đặt tên: prefix số (01_, 02_) quyết định thứ tự; dấu gạch dưới (_) hiển thị thành space trong nav label. Ví dụ 01_Model_Monitoring.py → "Model Monitoring".

Cách 2: st.navigation() API (Streamlit 1.36+)

st.navigation() cho phép cấu hình tường minh và linh hoạt hơn, không phụ thuộc vào tên file:

# app.py — dùng st.navigation() (1.36+)
import streamlit as st

pg = st.navigation([
    st.Page("views/dashboard.py", title="Dashboard", icon=":material/dashboard:"),
    st.Page("views/chatbot.py", title="Chatbot", icon=":material/chat:"),
    st.Page("views/settings.py", title="Settings", icon=":material/settings:"),
])
pg.run()

Với cách này, file Python không cần nằm trong pages/ và có thể đặt tên tự do.

Session state giữa các page

st.session_state tồn tại xuyên suốt khi user chuyển page trong cùng session. Đây là cách chia sẻ trạng thái (ví dụ: dữ liệu đã upload, model đã chọn) giữa Dashboard page và Chatbot page mà không cần database.

Streamlit cung cấp hai decorator cache với mục đích khác nhau:

@st.cache_data — cho data

Dùng cho function trả về data: DataFrame, dict, list, string, số. Mỗi lần gọi với argument khác nhau, Streamlit hash argument để tạo cache key khác nhau. Kết quả được copy (serialized/deserialized) khi trả về — mỗi lần gọi với cùng argument nhận bản copy riêng, không phải object chung.

@st.cache_data
def fetch_metrics(model_version: str, date: str) -> pd.DataFrame:
    # gọi API hoặc đọc DB
    return df

# TTL: cache hết hạn sau 5 phút
@st.cache_data(ttl=300)
def fetch_live_stats() -> dict:
    return call_monitoring_api()

# Max entries: giữ tối đa 10 bộ kết quả khác nhau
@st.cache_data(max_entries=10)
def embed_text(text: str) -> list:
    return embedding_model.encode(text).tolist()

@st.cache_resource — cho resource

Dùng cho function trả về resource không thể copy: model PyTorch/sklearn, database connection, HTTP client. Instance được share giữa mọi rerun và mọi user session — không bị copy. Phù hợp cho model loader, vector DB client, connection pool.

@st.cache_resource
def load_model(model_path: str):
    import torch
    model = torch.load(model_path, map_location="cpu")
    model.eval()
    return model

@st.cache_resource
def get_db_connection():
    import psycopg2
    return psycopg2.connect(st.secrets["DATABASE_URL"])

So sánh nhanh

Lỗi thường gặp với cache

Streamlit hash argument để tạo cache key. Nếu argument là object không hashable (ví dụ list, dict, object tự định nghĩa không implement __hash__), Streamlit sẽ báo lỗi hoặc bỏ qua cache. Giải pháp: chuyển argument sang kiểu hashable (tuple thay cho list), hoặc dùng hash_funcs parameter để tự định nghĩa cách hash.

# Sai: list không hashable
@st.cache_data
def process(items: list) -> pd.DataFrame:  # lỗi cache khi gọi
    ...

# Đúng: convert sang tuple trước khi truyền
result = process(tuple(my_list))
# hoặc dùng hash_funcs
@st.cache_data(hash_funcs={list: lambda x: str(x)})
def process(items: list) -> pd.DataFrame:
    ...

1. Quên cache model → mỗi click load lại

Model PyTorch 500MB load mỗi lần rerun — app gần như unusable. Luôn bọc model loader trong @st.cache_resource.

2. Lưu state trong biến global

# Sai: biến global bị reset mỗi rerun
history = []  # luôn rỗng

def add_message(msg):
    history.append(msg)  # thêm vào rồi mất sau rerun tiếp theo

# Đúng: session_state
if "history" not in st.session_state:
    st.session_state.history = []

3. st.dataframe với 100k+ row

Streamlit serialize toàn bộ DataFrame sang JSON để gửi về browser. 100k row thường mất vài giây và browser render chậm. Luôn limit số row hiển thị:

st.dataframe(df.head(500))  # hoặc st.dataframe(df.sample(500))
st.caption(f"Hiển thị 500/{len(df):,} dòng")

4. Dùng st.write() cho streaming text

st.write() không stream — nó chờ toàn bộ chuỗi rồi mới render. Để stream token từ LLM, phải dùng st.write_stream(generator) (1.30+).

5. Widget không bọc trong st.form → rerun quá nhiều

Mỗi widget (text_input, selectbox, v.v.) mặc định trigger rerun ngay khi thay đổi. Nếu form có nhiều widget, mỗi keystroke đều rerun — tốn tài nguyên. Bọc trong st.form để chỉ rerun khi user bấm submit:

with st.form("search_form"):
    keyword = st.text_input("Từ khóa")
    top_k = st.slider("Top K", 1, 20, 5)
    submitted = st.form_submit_button("Tìm kiếm")

if submitted:
    results = search(keyword, top_k)
    st.dataframe(results)

6. Không dùng uvicorn để serve Streamlit

Streamlit dùng Tornado (không phải ASGI) làm HTTP server nội bộ. Không thể serve bằng uvicorn app:app như FastAPI. Để deploy, chỉ dùng streamlit run app.py (hoặc platform hỗ trợ sẵn như HF Spaces, Streamlit Community Cloud).

Hai lựa chọn phổ biến nhất cho Streamlit app:

Streamlit Community Cloud

Link GitHub repo → deploy. Free tier cho public repo. App phải có requirements.txt. Secrets quản lý qua dashboard (không hard-code key trong code). Phù hợp để share demo với người ngoài tổ chức.

Giới hạn: 1 GB RAM, 1 vCPU cho free tier. Không phù hợp cho model lớn.

Hugging Face Spaces

Chọn SDK = Streamlit khi tạo Space. Push code lên repo Space trên HF. Hỗ trợ GPU Space (có phí). Phù hợp cho demo model ML/NLP vì cùng hệ sinh thái với Hugging Face Hub. Deploy chi tiết ở bài 12.

Self-host

# Chạy trong container (không dùng uvicorn)
streamlit run app.py --server.port 8501 --server.headless true

Wrap trong Docker, expose port 8501. Không cần ASGI server wrapper. Bài 32–33 (Module 6) sẽ hướng dẫn Dockerfile chi tiết.

Bài 11: Khi nào dùng Gradio, khi nào dùng Streamlit — so sánh kỹ thuật: execution model, state management, component library, use case điển hình và trade-off khi chọn một trong hai.