在這篇文章中，我將演示如何用 Streamlit 快速構建一個輕量的對話機器人 UI，并通過 LangChain / LangGraph 調用 LLM，實現簡單的對話功能。通過將前端和后端分離，你可以單獨測試模型調用和 UI 顯示。
為什么選擇 Streamlit？
Streamlit 是一個專為 Python 數據應用設計的前端框架，特點是：
極簡化前端開發，只需 Python 代碼即可構建 Web 應用。
與 Python 生態兼容，方便集成機器學習、LLM 等工具。
交互組件豐富，如表單、滑塊、下拉框等。
通過 Streamlit，我們可以專注于業務邏輯，而不用寫復雜的 HTML/CSS/JS。
系統架構
我們將系統拆分為兩部分：
后端 (
backend.py)管理對話狀態（狀態圖）
調用 LLM（如 ChatTongyi）
處理對話記憶和存儲（SQLite）
前端 (
app.py)使用 Streamlit 顯示對話界面
負責收集用戶輸入
調用后端生成模型回復
這種架構有幾個好處：
UI 與后端解耦，可單獨測試
后端邏輯可復用到其他應用或接口
數據存儲統一管理，方便擴展
后端實現 (backend.py)
import sqlite3
from typing import Annotated, Listfrom langchain_core.messages import AnyMessage, HumanMessage
from langchain_community.chat_models import ChatTongyifrom langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from typing_extensions import TypedDict# ============ 狀態定義 ============
class ChatState(TypedDict):messages: Annotated[List[AnyMessage], add_messages]# ============ 節點函數 ============
def make_call_model(llm: ChatTongyi):"""返回一個節點函數，讀取 state -> 調用 LLM -> 更新 state.messages"""def call_model(state: ChatState) -> ChatState:response = llm.invoke(state["messages"])return {"messages": [response]}return call_model# ============ 后端核心類 ============
class ChatBackend:def __init__(self, api_key: str, model_name: str = "qwen-plus", temperature: float = 0.7, db_path: str = "memory.sqlite"):self.llm = ChatTongyi(model=model_name, temperature=temperature, api_key=api_key)self.builder = StateGraph(ChatState)self.builder.add_node("model", make_call_model(self.llm))self.builder.set_entry_point("model")self.builder.add_edge("model", END)# SQLite 檢查點conn = sqlite3.connect(db_path, check_same_thread=False)self.checkpointer = SqliteSaver(conn)self.app = self.builder.compile(checkpointer=self.checkpointer)def chat(self, user_input: str, thread_id: str):config = {"configurable": {"thread_id": thread_id}}final_state = self.app.invoke({"messages": [HumanMessage(content=user_input)]}, config)ai_text = final_state["messages"][-1].contentreturn ai_textif __name__ == '__main__':from backend import ChatBackendbackend = ChatBackend(api_key="YOUR_KEY")print(backend.chat("你好", "thread1"))前端實現 (app_ui.py)
import os
import uuid
import streamlit as st
from backend import ChatBackendst.set_page_config(page_title="Chatbot", page_icon="💬", layout="wide")# ======== Streamlit CSS =========
st.markdown("""<style>.main {padding: 2rem 2rem;}.chat-card {background: rgba(255,255,255,0.65); backdrop-filter: blur(8px); border-radius: 20px; padding: 1.25rem; box-shadow: 0 10px 30px rgba(0,0,0,0.08);} .msg {border-radius: 16px; padding: 0.8rem 1rem; margin: 0.35rem 0; line-height: 1.5;}.human {background: #eef2ff;}.ai {background: #ecfeff;}.small {font-size: 0.86rem; color: #6b7280;}.tag {display:inline-block; padding: .25rem .6rem; border-radius: 9999px; border: 1px solid #e5e7eb; margin-right:.4rem; cursor:pointer}.tag:hover {background:#f3f4f6}.footer-note {color:#9ca3af; font-size:.85rem}[data-testid="stSidebarHeader"] {margin-bottom: 0px}/* 讓主標題上移，和側邊欄對齊 */.block-container {padding-top: 1.5rem !important;}</style>""",unsafe_allow_html=True,
)# ======== 側邊欄配置 =========
with st.sidebar:st.markdown("## ?? 配置")if "thread_id" not in st.session_state:st.session_state.thread_id = str(uuid.uuid4())if "chat_display" not in st.session_state:st.session_state.chat_display = []api_key = st.text_input("DashScope API Key", type="password", value=os.getenv("DASHSCOPE_API_KEY", ""))model_name = st.selectbox("選擇模型", ["qwen-plus", "qwen-turbo", "qwen-max"], index=0)temperature = st.slider("Temperature", 0.0, 1.0, 0.7, 0.05)if st.button("? 新建會話"):st.session_state.thread_id = str(uuid.uuid4())st.session_state.chat_display = []st.rerun()if st.button("🗑? 清空當前會話"):st.session_state.chat_display = []st.rerun()# ======== 主區域 =========
st.markdown("# 💬 Chatbot")
if not api_key:st.warning("請先在左側輸入 DashScope API Key 才能開始對話。")
else:backend = ChatBackend(api_key=api_key, model_name=model_name, temperature=temperature)st.markdown('<div class="chat-card">', unsafe_allow_html=True)if not st.session_state.chat_display:st.info("開始對話吧！")else:for role, content in st.session_state.chat_display:css_class = "human" if role == "user" else "ai"avatar = "🧑?💻" if role == "user" else "🤖"st.markdown(f"<div class='msg {css_class}'><span class='small'>{avatar} {role}</span><br/>{content}</div>", unsafe_allow_html=True)with st.form("chat-form", clear_on_submit=True):user_input = st.text_area("輸入你的問題/指令：", height=100, placeholder="比如：幫我寫一個二分查找函數。")submitted = st.form_submit_button("發送 ?")if submitted and user_input.strip():st.session_state.chat_display.append(("user", user_input))with st.spinner("思考中..."):ai_text = backend.chat(user_input, st.session_state.thread_id)st.session_state.chat_display.append(("assistant", ai_text))st.rerun()st.markdown('</div>', unsafe_allow_html=True)st.markdown(f"<div class='footer-note'>Session: <code>{st.session_state.thread_id}</code></div>", unsafe_allow_html=True)
運行效果
運行后端服務：無需額外啟動，
backend.py被app.py調用即可。在瀏覽器中訪問 Streamlit 頁面：
streamlit run app_ui.py
左側輸入 API Key，選擇模型和溫度。
在主區域輸入問題或指令，點擊“發送”，AI 回復將顯示在聊天窗口。
效果示例：
總結與擴展
通過這套架構，我們實現了：
前后端解耦：UI 與 LLM 調用分離，可單獨測試
對話記憶管理：使用 SQLite 保存會話狀態
可擴展性：后端可替換不同 LLM 或添加多輪對話邏輯
其它
如果你的模型支持流式輸出
backend_stream.py
import os
import sqlite3
import time
from typing import Annotated, Listfrom dotenv import load_dotenv
from langchain_core.messages import AnyMessage, HumanMessage
from langchain_community.chat_models import ChatTongyi
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from typing_extensions import TypedDict
from langchain.callbacks.base import BaseCallbackHandler# ===== 狀態定義 =====
class ChatState(TypedDict):messages: Annotated[List[AnyMessage], add_messages]# ===== 節點函數（流式調用） =====
def make_call_model(llm: ChatTongyi):def call_model(state: ChatState) -> ChatState:# ? 這里改為流式responses = []for chunk in llm.stream(state["messages"]):responses.append(chunk)return {"messages": responses}return call_model# ===== 流式回調 =====
class StreamCallbackHandler(BaseCallbackHandler):def __init__(self):self.chunks = []def on_llm_new_token(self, token: str, **kwargs):self.chunks.append(token)# ===== 后端核心 =====
class ChatBackend:def __init__(self, api_key: str, model_name: str = "qwen-plus", temperature: float = 0.7, db_path: str = "memory.sqlite"):self.llm = ChatTongyi(model=model_name,temperature=temperature,api_key=api_key,streaming=True, # 開啟流式)self.builder = StateGraph(ChatState)self.builder.add_node("model", make_call_model(self.llm))self.builder.set_entry_point("model")self.builder.add_edge("model", END)conn = sqlite3.connect(db_path, check_same_thread=False)self.checkpointer = SqliteSaver(conn)self.app = self.builder.compile(checkpointer=self.checkpointer)def chat_stream_simulated(self, user_input: str):"""模擬流式輸出（調試用）"""response = f"AI 正在回答你的問題: {user_input}"for ch in response:yield chtime.sleep(0.05)def chat_stream(self, user_input: str, thread_id: str):"""真正的流式輸出"""config = {"configurable": {"thread_id": thread_id}}# 用 llm.stream() 獲取流式輸出for chunk in self.llm.stream([HumanMessage(content=user_input)]):token = chunk.text()if token:yield token
app_stream_ui.py
# app_stream_ui_no_thinking.py
import os
import time
import uuid
import streamlit as st
from backend_stream import ChatBackend # 請確保 backend_stream.chat_stream 返回逐 token 字符串st.set_page_config(page_title="Chatbot", page_icon="💬", layout="wide")# ======== 原有 CSS（保持不變） =========
st.markdown("""<style>.main {padding: 2rem 2rem;}.chat-card {background: rgba(255,255,255,0.65); backdrop-filter: blur(8px); border-radius: 20px; padding: 1.25rem; box-shadow: 0 10px 30px rgba(0,0,0,0.08);} .msg {border-radius: 16px; padding: 0.8rem 1rem; margin: 0.35rem 0; line-height: 1.5;}.human {background: #eef2ff;}.ai {background: #ecfeff;}.small {font-size: 0.86rem; color: #6b7280;}.footer-note {color:#9ca3af; font-size:.85rem}[data-testid="stSidebarHeader"] { margin-bottom: 0px }.block-container { padding-top: 1.5rem !important; }</style>""",unsafe_allow_html=True,
)# ======== 側邊欄配置 =========
with st.sidebar:st.markdown("## ?? 配置")if "thread_id" not in st.session_state:st.session_state.thread_id = str(uuid.uuid4())if "chat_display" not in st.session_state:st.session_state.chat_display = []api_key = st.text_input("DashScope API Key", type="password", value=os.getenv("DASHSCOPE_API_KEY", ""))model_name = st.selectbox("選擇模型", ["qwen-plus", "qwen-turbo", "qwen-max"], index=0)temperature = st.slider("Temperature", 0.0, 1.0, 0.7, 0.05)if st.button("? 新建會話"):st.session_state.thread_id = str(uuid.uuid4())st.session_state.chat_display = []st.experimental_rerun()if st.button("🗑? 清空當前會話"):st.session_state.chat_display = []st.experimental_rerun()# ======== 主區域 =========
st.markdown("# 💬 Chatbot")if not api_key:st.warning("請先在左側輸入 DashScope API Key 才能開始對話。")
else:# 可以緩存 backend 到 session_state，但為簡單起見這里每次實例化（如需優化我可以幫你加緩存）backend = ChatBackend(api_key=api_key, model_name=model_name, temperature=temperature)# 聊天歷史占位（位于輸入上方）chat_area = st.container()history_ph = chat_area.empty()def render_history(ph):"""渲染 st.session_state.chat_display（使用你原來的 HTML + CSS 樣式）"""html = '<div class="chat-card">'for role, content in st.session_state.chat_display:css_class = "human" if role == "user" else "ai"avatar = "🧑?💻" if role == "user" else "🤖"html += f"<div class='msg {css_class}'><span class='small'>{avatar} {role}</span><br/>{content}</div>"html += "</div>"ph.markdown(html, unsafe_allow_html=True)# 初始渲染歷史render_history(history_ph)# 用戶輸入表單（表單在歷史下方，因此輸出始終在上面）with st.form("chat-form", clear_on_submit=True):user_input = st.text_area("輸入你的問題/指令：", height=100, placeholder="比如：幫我寫一個帶注釋的二分查找函數。")submitted = st.form_submit_button("發送 ?")if submitted and user_input.strip():# 1) 把用戶消息寫入歷史（立即可見）st.session_state.chat_display.append(("user", user_input))# 2) 插入 assistant 占位（空字符串），保證后續輸出顯示在上方歷史st.session_state.chat_display.append(("assistant", ""))ai_index = len(st.session_state.chat_display) - 1# 立刻渲染一次，使用戶看到自己的消息和空 assistant 氣泡（輸出會填充此氣泡）render_history(history_ph)# 3) 開始流式生成并實時寫回歷史full_text = ""try:for token in backend.chat_stream(user_input, st.session_state.thread_id):# token: 每次 yield 的字符串（可能是字符或片段）full_text += token# 更新 session 中的 assistant 占位內容（注意不要添加"思考中"）st.session_state.chat_display[ai_index] = ("assistant", full_text)# 重新渲染歷史，保證輸出始終在輸入上方render_history(history_ph)# 小睡短暫時間幫助 Streamlit 刷新（按需調整或刪除）time.sleep(0.01)except Exception as e:# 如果生成出錯，把錯誤消息放到 assistant 氣泡里（替代原邏輯）st.session_state.chat_display[ai_index] = ("assistant", f"[生成出錯] {e}")render_history(history_ph)else:# 生成完成（full_text 已包含最終結果），已在循環中寫回，無需額外操作pass# 頁腳（session id）st.markdown(f"<div class='footer-note'>Session: <code>{st.session_state.thread_id}</code></div>", unsafe_allow_html=True)
和生命期管理)
 - 繪制折線圖的利刃)
