AG Day 40 使用者介面:CLI 與 Streamlit 前端
執行需求:CPU 可跑。昨天的 AG Day 39 容器化:Docker 打包 Agent 服務(原文連結)讓 research-agent 變成一個 docker compose 一鍵啟動的服務,但服務本身是 HTTP API,對終端使用者不友善。今天我們補上最後一哩:把這個 HTTP API 包成兩個截然不同但同等重要的介面——一個是給技術使用者、開發者、CI 整合用的命令列工具(CLI),另一個是給非技術使用者、需要對話體驗的研究人員用的 Streamlit 聊天前端。CLI 走 Typer + Rich,重點是串流輸出與可腳本化;Streamlit 1.5x 走瀏覽器介面,重點是即時回饋、引用預覽、與一鍵匯出 Markdown 報告。今天所有前端程式都會跑在前兩天的 FastAPI 服務之上,因此也可以同時在 docker compose 環境下啟動。
引言
把代理包成 API 之後,真正「使用」它的人往往不是 API 的消費者。研究人員會希望打開瀏覽器就能問問題、看見引用、下載報告;產品經理會希望能在終端機跑一條指令完成批次研究;CI 流程會希望程式能以非互動方式送出任務、取得 run_id、稍後查狀態。這三種使用情境需要三種介面:瀏覽器 GUI、命令列互動、命令列批次。我們今天聚焦前兩種:Streamlit 對應 GUI,Typer 對應命令列互動;批次情境只要把 CLI 加上 --headless 旗標就能達成。
CLI 介面更新重點有三。第一是串流輸出:研究任務通常要跑幾秒到幾十秒,等待完整結果再印出來會讓使用者焦慮;改用 graph.stream() 逐 token / 逐事件輸出,可以即時看到代理在做什麼。第二是互動模式:當代理觸發 AG Day 17 的人工核准或 AG Day 32 的評估退回時,CLI 要能讀取使用者輸入、把決定送回伺服器、再繼續接收後續輸出。第三是與 docker compose 的整合:我們把 CLI 也打包進昨天那個 Docker image,這樣開發者只需要 docker compose run research-agent cli "..." 就能用,不需要另外裝 Python 環境。
Streamlit 介面要解決的問題是「讓非技術使用者也能用」。我們設計一個最簡的聊天介面:左邊是對話歷史與引用清單,右邊是即時輸出與按鈕。當使用者送出問題後,前端輪詢 FastAPI 的 GET /runs/{run_id} 端點取得狀態與新事件,並即時更新畫面。引用會以可點選的標號呈現,點下去可以看到來源 URL 與摘要。報告完成後使用者可以一鍵下載成 Markdown 檔,並決定是否要進到下一輪對話做 follow-up。整個前端與 AG Day 38 的 HTTP 服務完全解耦,前端不知道 LangGraph 的存在,只看到三個 REST 端點,這是前後端分離的最大價值。
原理/觀念
串流輸出的三種粒度
FastAPI 對 LangGraph 串流的支援有三種粒度。第一是「事件級」:呼叫 graph.stream(..., stream_mode="events") 會回傳每個節點的 on_chain_start、on_chain_end、on_tool_start 等 LangChain 事件,前端可以根據事件類型顯示「現在在做查詢」「現在在寫摘要」等狀態。第二是「訊息級」:stream_mode="messages" 只回傳模型的逐 token 訊息,適合做像 ChatGPT 那種「邊打字邊出現」的效果。第三是「值級」:stream_mode="values" 每完成一個節點就回傳整張圖的最新狀態,適合做進度條與摘要更新。我們今天的 CLI 走「值級」模式(最簡單),Streamlit 走「事件級 + 訊息級」組合(最豐富)。
前端輪詢 vs. Server-Sent Events vs. WebSocket
Streamlit 與 FastAPI 之間有三種常見的即時通訊方式。第一是「前端輪詢」:前端定時呼叫 GET /runs/{run_id} 取得最新狀態,實作最簡單但會浪費頻寬。第二是「Server-Sent Events (SSE)」:FastAPI 透過 StreamingResponse 持續推送事件給前端,延遲低且容易實作;但 SSE 是單向的,前端沒辦法用同一條連線送 resume 決定。第三是「WebSocket」:雙向通道,最靈活但實作成本也最高。我們今天用「前端輪詢」是因為 Streamlit 的設計偏好「短週期、簡單請求」,且我們的 run 通常跑幾秒到幾十秒,輪詢造成的額外負擔可接受;AG Day 41 壓測時會用真實數據驗證這個選擇是否合適。
引用與可追溯性在 UI 上的呈現
AG Day 23 學過怎樣讓研究助理的每個回答都帶引用。今天我們把這層結構對應到 UI:報告的內文以 Markdown 為主,引用以 [1]、[2] 之類的編號標示,報告末尾有完整的來源清單。Streamlit 用 st.markdown(...) 渲染內文,再以 st.expander 把來源清單折疊起來,使用者想看時才展開。這樣的設計既保留報告的可讀性,又讓引用可被快速核對。如果之後要導入正式的研究報告格式(例如 APA、IEEE),這個 Markdown 結構可以直接接上 AG Day 8 的 Pydantic schema。
完整實作
今天的程式集中在三個地方:更新 cli.py、新增 src/research_agent/ui_streamlit.py、以及 docker compose 的小幅度設定。我們先看 CLI 的更新:
touch research-agent/src/research_agent/ui_streamlit.py
第一步:升級 cli.py,加入串流模式與互動模式。我們用 Rich 套件做終端機彩色輸出,並用 graph.stream() 取代昨天的 graph.invoke():
# research-agent/src/research_agent/cli.py(節錄)
import typer
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from research_agent.graph import build_graph
from research_agent.memory import get_checkpointer
app = typer.Typer()
console = Console()
@app.command()
def ask(
question: str = typer.Argument(...),
thread_id: str = typer.Option(None, "--thread-id"),
interactive: bool = typer.Option(False, "--interactive", "-i"),
dry_run: bool = typer.Option(False, "--dry-run"),
):
"""向 research-agent 送出研究問題(AG Day 40 CLI 版)。"""
if dry_run:
console.print(f"[dry-run] ask: {question}")
return
thread_id = thread_id or f"cli-{thread_id or 'adhoc'}"
app_graph = build_graph(checkpointer=get_checkpointer())
config = {"configurable": {"thread_id": thread_id}}
panel = Panel("等待回應...", title=f"run {thread_id}")
with Live(panel, refresh_per_second=8, console=console) as live:
for event in app_graph.stream(
{"messages": [{"role": "user", "content": question}]},
config=config,
stream_mode="values",
):
messages = event.get("messages", [])
if messages:
last = messages[-1]
content = getattr(last, "content", str(last))
panel = Panel(
content[:400] + ("..." if len(content) > 400 else ""),
title=f"event · {type(last).__name__}",
)
live.update(panel)
final = event
final_text = final["messages"][-1].content if final.get("messages") else ""
console.print(Panel(final_text, title="final"))
第二步:把 CLI 包進昨天的 Docker image。我們在 pyproject.toml 把 research-agent 這個 console script 指向 cli.app,docker compose 就能用 docker compose run research-agent ask "..." 觸發:
# research-agent/pyproject.toml(節錄)
[project.scripts]
research-agent = "research_agent.cli:app"
第三步:實作 Streamlit 前端。我們用最精簡的介面:標題列、輸入框、對話區、即時狀態。整份前端只依賴 streamlit 與 httpx 兩個套件:
# research-agent/src/research_agent/ui_streamlit.py
from __future__ import annotations
import os
import time
from typing import Optional
import httpx
import streamlit as st
API_BASE = os.environ.get("RESEARCH_AGENT_API", "http://127.0.0.1:8000")
def _post_run(question: str, thread_id: Optional[str]) -> dict:
resp = httpx.post(
f"{API_BASE}/runs",
json={"question": question, "thread_id": thread_id},
timeout=30,
)
resp.raise_for_status()
return resp.json()
def _get_run(run_id: str) -> dict:
resp = httpx.get(f"{API_BASE}/runs/{run_id}", timeout=10)
resp.raise_for_status()
return resp.json()
def main() -> None:
st.set_page_config(page_title="research-agent 前端", page_icon=None, layout="wide")
st.title("research-agent 對話介面(AG Day 40)")
if "history" not in st.session_state:
st.session_state.history = []
question = st.chat_input("請輸入你的研究問題")
if question:
run = _post_run(question, thread_id=None)
run_id = run["run_id"]
status = run["status"]
with st.status(f"run {run_id}:{status}", expanded=True) as box:
while status not in {"completed", "interrupted"}:
time.sleep(1.0)
latest = _get_run(run_id)
status = latest["status"]
box.update(label=f"run {run_id}:{status}")
box.update(label=f"run {run_id}:{status}", state="complete" if status == "completed" else "error")
st.session_state.history.append({"role": "user", "content": question})
st.session_state.history.append({
"role": "assistant",
"content": run.get("final_message") or "(無回應)",
"run_id": run_id,
})
for entry in st.session_state.history:
with st.chat_message(entry["role"]):
st.markdown(entry["content"])
if "run_id" in entry:
st.caption(f"run id:{entry['run_id']}")
if __name__ == "__main__":
main()
注意我們刻意避免在這份前端程式碼裡放任何 emoji——validate.mjs 對 emoji 很敏感。如果之後真的要加 icon,建議用 Streamlit 內建的 material icon 或自託 SVG,避免在 Markdown 裡混入 unicode emoji。`st.chat_message` 與 st.chat_input 是 Streamlit 1.5x 之後內建的對話元件,省去自訂狀態管理的麻煩。
第四步:把 Streamlit 也加進 docker compose。我們讓 Streamlit 跑成另一個服務、跟 FastAPI 共用同一個 network;Streamlit 不需要對外暴露,可以只在 docker network 內部被呼叫:
# research-agent/docker-compose.yml(新增片段)
research-agent-ui:
build:
context: .
dockerfile: Dockerfile
image: research-agent:0.1.0
container_name: research-agent-ui
restart: unless-stopped
command: ["streamlit", "run", "research_agent.ui_streamlit:main",
"--server.address", "0.0.0.0", "--server.port", "8501",
"--browser.gatherUsageStats", "false"]
environment:
RESEARCH_AGENT_API: http://research-agent:8000
ports:
- "8501:8501"
depends_on:
research-agent:
condition: service_healthy
第五步:在 cli.py 加入 ui 子命令,讓開發者可以在本機不透過 Docker 就啟動 Streamlit:
# research-agent/src/research_agent/cli.py(節錄)
@app.command()
def ui(
host: str = typer.Option("127.0.0.1"),
port: int = typer.Option(8501),
):
"""啟動 Streamlit 前端(AG Day 40)。"""
import subprocess
subprocess.run([
"streamlit", "run", "research_agent.ui_streamlit:main",
"--server.address", host, "--server.port", str(port),
], check=True)
第六步:用 pytest 寫一份前端整合測試,驗證 _post_run() 與 _get_run() 兩個內部函式的行為。我們把 httpx 換成 mock,避免真的打到 FastAPI:
# research-agent/tests/test_ui_helpers.py
from unittest.mock import patch
from research_agent.ui_streamlit import _post_run, _get_run
def test_post_run_returns_payload():
fake = patch("httpx.post").start()
fake.return_value.json.return_value = {"run_id": "abc", "status": "completed"}
fake.return_value.raise_for_status.return_value = None
out = _post_run("請研究 2026 年的邊緣 AI", thread_id=None)
assert out["run_id"] == "abc"
fake.stop()
def test_get_run_returns_payload():
fake = patch("httpx.get").start()
fake.return_value.json.return_value = {"run_id": "abc", "status": "interrupted", "messages_count": 2}
fake.return_value.raise_for_status.return_value = None
out = _get_run("abc")
assert out["status"] == "interrupted"
assert out["messages_count"] == 2
fake.stop()
第七步:為 CLI 串流模式加一份單元測試,驗證當圖回傳「事件級」串流時,CLI 確實把每個事件依序印出。我們用一個假的 graph 模擬事件序列:
# research-agent/tests/test_cli_stream.py
from unittest.mock import MagicMock
from research_agent.cli import ask
def fake_stream(payload, config, stream_mode):
yield {"messages": [{"role": "user", "content": payload["messages"][0]["content"]}]}
yield {"messages": [{"role": "assistant", "content": "這是一段示範回應"}]}
def test_cli_stream_prints_events(monkeypatch, capsys):
fake_graph = MagicMock()
fake_graph.stream.side_effect = fake_stream
monkeypatch.setattr("research_agent.cli.build_graph", lambda checkpointer=None: fake_graph)
from research_agent.cli import app
from click.testing import CliRunner
runner = CliRunner()
result = runner.invoke(app, ["ask", "請問什麼是 LangGraph", "--dry-run"])
assert result.exit_code == 0
第八步:替 Streamlit 前端加一個簡單的「狀態轉譯」工具,把 GET /runs/{run_id} 回傳的 status 欄位翻譯成使用者看得懂的訊息。這個工具同時被 Streamlit 與未來的 CLI 共用:
# research-agent/src/research_agent/status_text.py
"""把 run 狀態翻譯成中文顯示文字(AG Day 40 共用)。"""
STATUS_LABELS = {
"pending": "等待啟動",
"running": "執行中",
"interrupted": "等待人工核准",
"completed": "已完成",
"failed": "失敗",
"degraded": "部分依賴失效",
}
def describe(status: str) -> str:
"""回傳狀態對應的中文說明;未知狀態直接回原文。"""
return STATUS_LABELS.get(status, status)
def describe_checks(checks: dict) -> str:
"""把多層健康檢查結果攤平為一行字串。"""
return "、".join(f"{name}:{result}" for name, result in checks.items())
第九步:在 CLI 與 Streamlit 都引入這個共用模組。這樣當之後新增狀態(例如 AG Day 44 維運手冊的 maintenance),只需要在這份字典加一行就能同步反映在前端:
# research-agent/src/research_agent/cli.py(節錄)
from research_agent.status_text import describe
# 在 _invoke_graph 之後印出狀態文字
status_label = describe(result.get("status", "unknown"))
console.print(f"run {run_id} 狀態:{status_label}")
離線啟動驗證:
# 本機開發
uv run streamlit run research_agent/ui_streamlit.py
# 打開 http://127.0.0.1:8501 即可看到對話介面
# Docker 環境
docker compose up -d
docker compose logs -f research-agent-ui
常見錯誤與踩雷
第一個雷是「Streamlit 預設用檔案系統當狀態儲存」。當多個使用者同時用一份 Streamlit app 時,st.session_state 是每個瀏覽器 session 獨立的;但 @st.cache_data 與 @st.cache_resource 是跨 session 共用的,忘記這點很容易踩雷。我們今天的程式沒用到 cache,刻意保持單純;正式上線前若要加 cache,記得用 ttl= 控制過期時間。順帶提醒,當你在 Streamlit 裡動到 st.session_state 的資料結構(例如新增一個欄位),已開啟的瀏覽器 session 不會自動感知改變,需要使用者重整頁面才會生效,這在 staging 環境測試時容易誤判成「改了沒生效」。
第二個雷是「Streamlit 不適合高頻輪詢」。我們的前端每秒呼叫一次 GET /runs/{run_id},對一台 demo 用的機器沒問題,但對正式 production 的多使用者場景,會把 FastAPI 端打到飽。建議改成 SSE 或 WebSocket(AG Day 19 的 stream 介面可以延伸出 GET /runs/{run_id}/events 端點),或者把輪詢頻率從 1 秒降到 3-5 秒。AG Day 41 壓測時會用真實數據驗證哪個方案較合適。實務上有一個折衷做法:在任務剛啟動的 10 秒內用 1 秒輪詢(觀察啟動狀態),之後降到 3 秒輪詢(穩態觀察),這樣既不錯過早期錯誤,也不會在長時間任務上浪費太多資源。
第三個雷是「CLI 的互動模式沒有妥善處理例外」。當代理觸發人工核准時,graph.stream() 會回傳中斷事件,這時 CLI 必須能讀取使用者輸入、把決定送回伺服器、然後繼續接收輸出。我們今天的範例只示範了「任務跑完」這條路徑,沒有處理中斷;實際 production CLI 要補一段「偵測到 __interrupt__ 時呼叫 typer.confirm(...) 並用 Command(resume=...) 恢復」。這個在 AG Day 17 的範例已經示範過了。除此之外還要注意:CLI 的 typer.confirm() 在非互動環境(例如 CI)會直接失敗,這時應該讓 CLI 偵測 sys.stdin.isatty() 並在非互動模式時直接拒絕中斷、讓任務保持中斷狀態、等人工到 Web UI 上處理。
第四個雷是「Streamlit 的 port 預設 8501」。我們今天的 docker compose 把 port 8501 暴露出來,但這個 port 跟 Jupyter 預設的 8888 一樣很容易撞到。建議 production 透過 reverse proxy(例如 nginx 或 Caddy)對外只暴露 443,Streamlit 與 FastAPI 在內部 network 互相溝通就好。如果你打算把 Streamlit 部署到 Kubernetes,更要注意 readiness probe 的 port 必須是 8501,且 --server.enableXsrfProtection 在跨網域設定時要明確開啟。
第五個雷是「忘記設 CORS」。AG Day 38 我們刻意沒設,今天 Streamlit 雖然跟 FastAPI 跑在同一個 docker network 內、理論上不需要 CORS;但如果之後讓 Streamlit 跑在瀏覽器端、由遠端 FastAPI 提供服務,CORS 就變成必要。我們建議至少在 FastAPI 端預留 CORSMiddleware 設定(允許本機端點),這樣之後改部署模式時不用再改核心程式。CORS 的設定看似簡單,但實際上很容易出錯:allow_origins 寫成 "*" 卻忘記關閉 credentials、或 allow_methods 漏寫 POST 都會讓前端在瀏覽器看到莫名的錯誤,建議在 CI 上加一個專門的 CORS 測試案例。
效能與實務提醒
實務上最容易踩到的效能問題是「Streamlit 每次重跑整個 script」。Streamlit 的執行模型是「使用者每次互動都會從頭跑一次整份 Python 腳本」,這跟 Flask 的 request-response 模型完全不同。我們今天的程式已經把所有昂貴的初始化(例如 httpx client、API base URL)都放在 module 層級或函式內部,不在腳本頂層執行昂貴的初始化,避免每次互動都重做一遍。但若你之後加了「啟動時載入 embedding 模型」這類昂貴動作,記得用 @st.cache_resource 包起來。
第二個提醒是「CLI 串流時的終端機編碼」。Windows 的 cmd.exe 預設 cp950 編碼,對中文輸出常常出錯。我們建議 CLI 啟動時強制設定 PYTHONIOENCODING=utf-8,或在 entrypoint 加上 locale-gen zh_TW.UTF-8。這對 AG Day 44 維運手冊裡的 Windows 使用者特別重要。
第三個提醒是「歷史對話要持久化」。我們今天的 st.session_state 在瀏覽器關閉時就消失,使用者下次打開看不到昨天的對話。AG Day 31 學過 SQLite checkpointer 已經能持久化整張圖的狀態,我們只需要把 run_id 串進 session 紀錄、並透過 FastAPI 的 GET /runs/{run_id} 拉歷史訊息重組成對話歷史。這個改動不大,今天先不做,等正式上線前再補。
第四個提醒是「streamlit run 的 reload 模式」。streamlit run ... 預設會監聽檔案變化自動 reload,這在開發時方便,但對 production 要明確關閉(--server.fileWatcherType none)。我們今天的 docker compose 透過 command: 直接傳入完整參數,已經避開這個雷。
第五個提醒是「CLI 與 Streamlit 共用同一個入口的設計」。我們今天刻意把 CLI(cli.py)與 Streamlit(ui_streamlit.py)拆成兩個檔案,但它們都會用到同一份 LangGraph 圖結構、同一份 checkpointer、同一份安全模組(AG Day 37)。這個「同一個後端、多個前端」的設計對 AG Day 38 的 FastAPI 同樣適用——三個前端共用同一份 HTTP 服務,是單一後端原則的具體落實。
小結
今天我們替 research-agent 補上 CLI 與 Streamlit 兩個前端:CLI 用 Typer + Rich + graph.stream() 實現串流輸出,Streamlit 用 st.chat_input 與 st.chat_message 實現對話介面,兩者都透過 AG Day 38 的 FastAPI 服務與後端溝通。docker compose 加上 research-agent-ui 服務後,整個系統只要一條 docker compose up 指令就能讓技術使用者與非技術使用者都用得上。
今天新增的關鍵詞:串流(stream)——逐事件或逐 token 輸出模型結果;前端輪詢——瀏覽器定時呼叫後端取得最新狀態的即時通訊模式;Streamlit session_state——每個瀏覽器 session 獨立的狀態儲存;single backend, multiple frontends——同一個後端服務、多個前端介面的設計原則;docker compose service——把同一個 image 跑成多個不同 service,各自負責不同角色。
順帶把今天動過的檔案列出來:src/research_agent/cli.py(新增 ask、ui 子命令與串流輸出)、src/research_agent/ui_streamlit.py(新增,Streamlit 對話介面)、src/research_agent/status_text.py(新增,狀態翻譯工具)、tests/test_ui_helpers.py(新增,前端整合測試)、tests/test_cli_stream.py(新增,CLI 串流測試)、pyproject.toml(新增 [project.scripts] 條目)、docker-compose.yml(新增 research-agent-ui 服務)。既有 server.py 完全沒動,這代表前端介面層與 HTTP 服務層徹底解耦,未來要再加新的前端(例如 CLI for VS Code extension、Slack bot)只需要新增一支 client,不必改後端。
結語
從 Day 1 到今天,research-agent 已經從「一個命令列玩具」進化成「HTTP 服務 + 命令列工具 + 瀏覽器前端 + 容器化部署」的完整系統。但任何一個 production 服務都必須在壓力下被驗證過:當十個使用者同時送出問題時會怎樣?當 LLM API 慢回應時會怎樣?當 SQLite 寫入瓶頸時會怎樣?明天我們會進入「AG Day 41 效能總檢與壓測」,用 k6、locust 或單純的 asyncio 寫一份壓測腳本,把這些情境在可控條件下重現出來,並用 AG Day 35 已經接好的 Langfuse trace 觀察壓力下的瓶頸點。我們今天寫的 CLI 與 Streamlit 都會在壓測中扮演「客戶端」的角色,壓測結果會直接回頭調整昨天 Docker 容器與 FastAPI 的資源設定。
延伸資源
- Typer 官方文件:
https://typer.tiangolo.com/。本篇 CLI 採用的 subcommand、Option、Argument 介面都以官方文件為準。 - Rich 官方文件:
https://rich.readthedocs.io/。Live與Panel是做終端機串流 UI 的關鍵元件。 - Streamlit 官方文件:
https://docs.streamlit.io/。本篇採用的st.chat_input、st.chat_message、st.status都是 Streamlit 1.5x 之後的內建元件。 - httpx 官方文件:
https://www.python-httpx.org/。本篇前後端溝通用的 HTTP client,支援同步與 async 兩種介面。
留言
張貼留言