AG Day 38 部署:FastAPI 包裝 Agent
執行需求:CPU 可跑。昨天的 AG Day 37 安全:prompt injection 與工具防護(原文連結)替 research-agent 補上了輸入分類、工具白名單與輸出遮罩,整個代理已經具備對外提供服務的最低安全門檻。今天我們把這個代理從「命令列工具」推進到「HTTP 服務」:用 FastAPI 把 LangGraph 圖包成幾個 REST 端點,讓其他應用(CLI、Streamlit、其他內部系統)能透過 HTTP 呼叫研究助理。我們會用 uvicorn 啟動服務、用 Pydantic v2 做請求與回應的 schema 驗證,並把 AG Day 37 的安全模組串進 FastAPI 的依賴注入(Depends)裡。所有程式碼都在本機 CPU 上能完整驗證,不需外部服務或 API 金鑰也能跑得動骨架。
引言
到目前為止,research-agent 都是命令列形式:uv run research-agent "研究問題" 執行一次就結束。這對開發者很方便,但對其他應用來說不是友善的介面——沒人想在自己的程式裡 subprocess.run() 一個 CLI;他們想要的是 HTTP 呼叫。FastAPI 是把 Python 程式對外暴露成 HTTP 服務最直接的選擇:它內建 Pydantic 驗證、型別註解自動產生 OpenAPI 文件、且原生支援 async,搭配 LangGraph 的串流介面特別合拍。今天的目標是新增 src/research_agent/server.py,提供三個端點:POST /runs 開啟一個新的研究任務、GET /runs/{run_id} 查詢任務狀態、POST /runs/{run_id}/resume 從中斷點(例如昨天的人工核准、或模型評估被拒)恢復執行。
FastAPI 雖然容易上手,但「把一個有狀態的代理放進無狀態的 HTTP 服務」本身有不少細節。今天我們會處理三個常見的雷:第一是 checkpointer 的生命週期——每個請求進來時要怎麼拿到對應的 checkpointer 與 thread_id;第二是同步與非同步的橋接——LangGraph 圖的 invoke() 在 FastAPI 的 async event loop 裡怎麼呼叫才不會阻塞整個服務;第三是請求驗證——AG Day 37 的 security.py 怎麼透過 FastAPI 的 Depends 機制接到所有進來的請求。我們也會用 httpx 寫一段整合測試,模擬「開新任務 → 查狀態 → 收到中斷 → 恢復執行」的完整流程。
這一篇不談容器化(AG Day 39 才會進到 Docker)、也不談正式部署的 HTTPS / reverse proxy(那是維運手冊的事)。今天的目標只有一個:讓 research-agent 能用 uv run uvicorn research_agent.server:app 啟動,並用 curl 或 httpx 完成幾次完整呼叫。明天我們會把這個服務推進到容器化階段。
原理/觀念
有狀態代理 vs. 無狀態 HTTP
HTTP 服務天生是無狀態的:每個請求都應該可以被任意一台機器處理。但我們的 LangGraph 圖是有狀態的:每次 invoke() 之後,圖的狀態會被寫進 checkpointer,下次同 thread_id 呼叫時從那裡繼續。把這兩者接起來有兩種主流做法。第一種是「thread_id 由呼叫端決定」:呼叫端在每次請求帶上一個 thread_id,伺服器端用這個 ID 從 checkpointer 讀出對應的圖狀態並恢復執行;同一 thread 上的多次呼叫會被視為同一個任務的不同階段。第二種是「run_id 由伺服器端決定」:呼叫端呼叫 POST /runs 時伺服器端隨機產生一個 run_id,呼叫端之後用這個 run_id 查狀態或恢復。今天我們採用第一種與第二種的混合:POST /runs 開新任務時伺服器端產生 run_id(同時也作為 thread_id),呼叫端拿到的就是一個可以之後查詢與恢復的識別碼。
同步介面包進 async event loop
LangGraph 的 invoke() 是同步函式(雖然底層模型呼叫可能用 async client)。當這個同步呼叫跑在 FastAPI 的 async event loop 裡,會卡住整個 worker thread,導致其他請求都沒辦法被處理。實務上有兩種解法:第一種是把同步呼叫丟到 run_in_executor,讓它在 thread pool 裡跑,event loop 繼續處理其他請求;第二種是把 LangGraph 圖改成 async 版本(ainvoke()),完全在 event loop 裡跑。我們今天採用第一種,因為既有的 LangGraph 介面都是同步寫的,改 async 牽動太大;對一個偶爾會跑幾秒鐘的研究任務來說,run_in_executor 已經足夠。
FastAPI 的依賴注入怎麼用於安全模組
FastAPI 的 Depends() 機制讓我們可以把「每個請求都會跑」的前處理邏輯抽成獨立函式,並掛在端點簽名上。AG Day 37 的 redact() 與 classify_segment() 自然就落在這裡:我們新增一個 security_dependency() 函式,每個端點宣告 Depends(security_dependency),FastAPI 會在處理該端點之前先呼叫這個函式,把請求內容通過分類與遮罩兩道關卡,再進入實際的業務邏輯。這比在每個端點函式裡手動呼叫安全函式更不容易漏,因為新增端點時只要記得加上 Depends(),就自動繼承了同樣的保護。
完整實作
今天的程式集中在兩個檔案:新增 src/research_agent/server.py,並小幅更新 cli.py 把今天的 server 子命令加進來。我們先建立檔案:
touch research-agent/src/research_agent/server.py
第一步:定義 Pydantic v2 的請求與回應 schema。這些 schema 既是 FastAPI 的驗證器、也是 OpenAPI 文件的來源、也可以在程式裡直接拿來型別化處理。我們用 Field 加上說明欄位,讓 OpenAPI 介面更友善:
# research-agent/src/research_agent/server.py
from __future__ import annotations
import asyncio
import uuid
from datetime import datetime, timezone
from typing import Any, Optional
from fastapi import Depends, FastAPI, HTTPException, status
from langchain_core.messages import HumanMessage
from langgraph.types import Command
from pydantic import BaseModel, Field
from research_agent.graph import build_graph
from research_agent.memory import get_checkpointer
from research_agent.security import classify_segment, redact, SourceKind
app = FastAPI(
title="research-agent API",
description="把 LangGraph 研究助理包成 HTTP 服務(AG Day 38)",
version="0.1.0",
)
class RunRequest(BaseModel):
"""啟動一個新的研究任務。"""
question: str = Field(min_length=1, max_length=2000, description="要研究的問題")
thread_id: Optional[str] = Field(default=None, description="沿用既有 thread,未提供則自動產生")
class RunResponse(BaseModel):
"""啟動任務的回應:拿到 run_id 之後可以用來查狀態或恢復。"""
run_id: str
status: str
created_at: datetime
final_message: Optional[str] = None
class StatusResponse(BaseModel):
run_id: str
status: str
messages_count: int
last_event: Optional[str] = None
class ResumeRequest(BaseModel):
"""從中斷點恢復執行;對應 AG Day 17 的人工核准與 AG Day 32 的評估退回。"""
decision: str = Field(description="例如 'approve'、'reject',依觸發中斷的工具決定")
第二步:實作安全依賴函式,把 AG Day 37 的 classify_segment() 與 redact() 接到 FastAPI。這個函式會在每個端點前自動執行,並把分類後的請求包成一個統一型別:
# research-agent/src/research_agent/server.py(接續)
class SecuredRequest(BaseModel):
"""經過安全模組處理後的請求物件。"""
question: str
thread_id: str
flagged: bool = False
def security_dependency(payload: RunRequest) -> SecuredRequest:
"""FastAPI Depends:對每個請求做分類與遮罩。"""
classified = classify_segment(SourceKind.USER, payload.question)
flagged = classified.kind.value == "mixed"
safe_question = redact(payload.question)
thread_id = payload.thread_id or f"run-{uuid.uuid4().hex[:12]}"
if flagged:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="問題文字疑似含指令注入字樣,請重新表述。",
)
return SecuredRequest(question=safe_question, thread_id=thread_id, flagged=flagged)
第三步:實作三個端點 POST /runs、GET /runs/{run_id}、POST /runs/{run_id}/resume。為了不阻塞 event loop,所有對 LangGraph 圖的呼叫都用 run_in_executor 丟到 thread pool 跑:
# research-agent/src/research_agent/server.py(接續)
async def _invoke_graph(thread_id: str, payload: dict) -> dict:
"""用 executor 跑同步的 graph.invoke,避免阻塞 event loop。"""
loop = asyncio.get_running_loop()
app_graph = build_graph(checkpointer=get_checkpointer())
config = {"configurable": {"thread_id": thread_id}}
return await loop.run_in_executor(
None,
lambda: app_graph.invoke(payload, config=config),
)
@app.post("/runs", response_model=RunResponse, status_code=status.HTTP_201_CREATED)
async def create_run(secured: SecuredRequest = Depends(security_dependency)) -> RunResponse:
"""啟動一個新的研究任務。"""
result = await _invoke_graph(
secured.thread_id,
{"messages": [HumanMessage(content=secured.question)]},
)
final_message = result["messages"][-1].content if result.get("messages") else None
return RunResponse(
run_id=secured.thread_id,
status=result.get("__interrupt__") and "interrupted" or "completed",
created_at=datetime.now(timezone.utc),
final_message=final_message,
)
@app.get("/runs/{run_id}", response_model=StatusResponse)
async def get_run(run_id: str) -> StatusResponse:
"""查詢任務狀態。"""
loop = asyncio.get_running_loop()
app_graph = build_graph(checkpointer=get_checkpointer())
snapshot = await loop.run_in_executor(
None,
lambda: app_graph.get_state({"configurable": {"thread_id": run_id}}),
)
return StatusResponse(
run_id=run_id,
status="interrupted" if snapshot.next else ("completed" if snapshot.values else "pending"),
messages_count=len(snapshot.values.get("messages", [])) if snapshot.values else 0,
last_event=str(snapshot.next) if snapshot.next else None,
)
@app.post("/runs/{run_id}/resume", response_model=RunResponse)
async def resume_run(run_id: str, body: ResumeRequest) -> RunResponse:
"""從中斷點恢復執行;對應 AG Day 17 的 HITL 與 AG Day 32 的評估退回。"""
result = await _invoke_graph(
run_id,
Command(resume=body.decision),
)
final_message = result["messages"][-1].content if result.get("messages") else None
return RunResponse(
run_id=run_id,
status=result.get("__interrupt__") and "interrupted" or "completed",
created_at=datetime.now(timezone.utc),
final_message=final_message,
)
@app.get("/healthz")
async def healthz() -> dict:
"""健康檢查端點:給 load balancer / Docker healthcheck 使用。"""
return {"status": "ok"}
第四步:用 httpx 寫一段整合測試,模擬「開新任務 → 查狀態 → 收到中斷 → 恢復執行」完整流程。我們用 httpx.ASGITransport 在測試裡直接呼叫 ASGI app,不需真的啟動 uvicorn:
# research-agent/tests/test_server.py
import pytest
from httpx import ASGITransport, AsyncClient
from research_agent.server import app
@pytest.fixture
def fake_graph(monkeypatch):
"""把 graph 換成假實作,避免真的呼叫模型。"""
class FakeGraph:
def __init__(self):
self.calls = []
def invoke(self, payload, config):
self.calls.append((payload, config))
# 第一次 invoke 回傳中斷,第二次(resume)回完成
if len(self.calls) == 1:
return {
"messages": [payload["messages"][0]],
"__interrupt__": [{"value": {"reason": "需核准"}}],
}
return {
"messages": payload.get("messages", []) + [
type("Msg", (), {"content": "(示範)已完成研究"})()
]
}
def get_state(self, config):
class Snap:
values = {"messages": ["hi"]}
next = ()
return Snap()
fake = FakeGraph()
monkeypatch.setattr("research_agent.server.build_graph", lambda checkpointer=None: fake)
return fake
@pytest.mark.asyncio
async def test_create_and_get_run(fake_graph):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
resp = await client.post("/runs", json={"question": "請研究 2026 年邊緣 AI 晶片"})
assert resp.status_code == 201
run_id = resp.json()["run_id"]
assert resp.json()["status"] == "interrupted"
status = await client.get(f"/runs/{run_id}")
assert status.status_code == 200
assert status.json()["status"] in {"interrupted", "completed", "pending"}
@pytest.mark.asyncio
async def test_resume_run(fake_graph):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
resp = await client.post("/runs", json={"question": "請研究向量資料庫"})
run_id = resp.json()["run_id"]
resumed = await client.post(f"/runs/{run_id}/resume", json={"decision": "approve"})
assert resumed.status_code == 200
assert resumed.json()["status"] == "completed"
@pytest.mark.asyncio
async def test_healthz():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
resp = await client.get("/healthz")
assert resp.status_code == 200
assert resp.json() == {"status": "ok"}
第五步:在 cli.py 補一個 serve 子命令,讓開發者可以用 uv run research-agent serve 啟動整個 HTTP 服務,並用 --port、--host 參數覆寫預設值:
# research-agent/src/research_agent/cli.py(新增子命令)
import uvicorn
import typer
@app.command()
def serve(
host: str = typer.Option("127.0.0.1", help="綁定的 host"),
port: int = typer.Option(8000, help="綁定的 port"),
reload: bool = typer.Option(False, help="開發模式自動 reload"),
):
"""啟動 FastAPI 服務(AG Day 38)。"""
uvicorn.run(
"research_agent.server:app",
host=host,
port=port,
reload=reload,
)
離線啟動驗證:
cd research-agent
uv run uvicorn research_agent.server:app --port 8000 --reload
# 在另一個終端機:
curl -s http://127.0.0.1:8000/healthz
# {"status":"ok"}
常見錯誤與踩雷
第一個雷是「FastAPI 預設會把同步端點函式丟到 thread pool,但你自己呼叫的阻塞程式碼不會」。很多人以為把函式宣告為 async def 就夠了,但若在 async 函式裡直接呼叫 graph.invoke()(同步),仍然會阻塞。我們今天用 run_in_executor 把同步呼叫丟回 thread pool,這在 LangGraph 還沒原生支援 async 之前是最穩定的做法。AG Day 19 的串流介面在 FastAPI 裡的整合也會碰到同樣的問題,請用同一招處理。
第二個雷是「checkpointer 的資源生命週期」。每個端點都呼叫 get_checkpointer() 重新生成一個 SQLite 連線雖然不會錯,但會讓 SQLite 在高並行下撞到鎖。建議把 get_checkpointer() 換成單例(singleton)模式,整個 process 共用同一個連線池,並透過 FastAPI 的 lifespan event 統一管理開關。我們今天為了簡化沒這樣做,但 production 上線前一定要處理。
第三個雷是「沒有正確區分建立任務(POST /runs)與恢復任務(POST /runs/{run_id}/resume)」。我們今天的設計是 POST /runs 永遠從頭啟動新任務(即使呼叫端帶了 thread_id 也是覆寫),POST /runs/{run_id}/resume 永遠從中斷點恢復。如果你混用,把 resume 設計成「沒中斷就當成新任務跑」,會遇到一種詭異的失敗:呼叫端以為在恢復、伺服器端卻從頭跑起,造成狀態錯亂。建議在 create_run 端點拒絕帶有已存在 thread_id 的請求,要求呼叫端明確用 resume 路徑。
第四個雷是「沒有設定 request 與 response 大小限制」。研究任務會把整份報告塞進 final_message,報告若很大(例如嵌入大量 base64 圖片),會讓單次回應超過幾 MB,擊穿 nginx 或 load balancer 的預設上限。請在 FastAPI 的中介層加上 request body 限制(例如 Request.max_body_size = 5 * 1024 * 1024),並把 final_message 改成「報告檔案路徑」而非「整段內文」,讓呼叫端另開 GET 端點下載完整報告。
第五個雷是「忘記處理 CORS」。如果之後要從 Streamlit(AG Day 40)或瀏覽器前端呼叫這個 API,必須設定 CORSMiddleware 允許的 origin。我們今天刻意沒加,是因為目前所有呼叫端都在伺服器端(CLI、內部服務),但 AG Day 40 會用到,到時候會把 CORSMiddleware 加進來並允許本機端點。
效能與實務提醒
實務上最容易踩到的效能問題是「每個請求都重新產生一個 LangGraph 圖實例」。build_graph() 內部會註冊節點、檢查 schema、建立 routing 表,這個成本雖然不大但也不該每次呼叫都重做。建議把 build_graph() 換成「懶載入單例」:第一次呼叫時建立、後續直接回傳同一個物件。checkpointer 同樣建議單例化。FastAPI 的 lifespan 機制是放這些單例的好地方。
第二個提醒是「/runs/{run_id}/resume 端點的冪等性」。如果呼叫端因為網路問題重試同一個 resume 請求,伺服器端可能會收到兩次相同的 resume,造成重複執行。我們今天的實作沒處理這個;建議在 resume 端點用 checkpointer 內的某個狀態(例如「上一個 event 的 id」)當冪等鍵,相同 id 的 resume 直接回傳上次的結果而不真的執行。這對長任務特別重要,因為重跑一次研究任務成本不低。
第三個提醒是「OpenAPI 文件應該跟程式碼同步」。FastAPI 的 OpenAPI 文件是從 Pydantic schema 自動產生的,這代表只要 RunRequest 的欄位改變、文件就會跟著變。建議把這個 OpenAPI endpoint(/openapi.json)串進 CI 流程,在 PR 上自動產生文件變更的 diff,這樣 reviewer 可以順便看到「介面改了什麼」。AG Day 42 會再談文件化,這裡先預告這個做法。
第四個提醒是「正式部署前要把 --reload 拿掉」。--reload 是 uvicorn 的開發模式,會監聽檔案系統變化並重啟服務,方便除錯,但生產環境應該關掉以避免不必要的負擔。AG Day 39 容器化時會把這個 flag 完全去掉,並透過環境變數(例如 RESEARCH_AGENT_RELOAD=false)傳入。
第五個提醒是「AG Day 31 的 SQLite 檢查點放在容器內是有狀態的」。今天我們把 checkpointer 沿用 AG Day 31 的 SQLite 路徑,這對單機本機測試沒問題,但容器化之後每次容器重啟 SQLite 檔案會不見,這個問題 AG Day 39 會用 volume mount 解決。在那之前請把這層不確定性記下來。
第六個提醒是「環境變數的管理」。FastAPI 服務啟動時通常需要讀取 API 金鑰、模型名稱、Langfuse 端點等設定。我們在 AG Day 3 已經建立 OPENAI_API_KEY、ANTHROPIC_API_KEY、TAVILY_API_KEY、LANGFUSE_PUBLIC_KEY、LANGFUSE_SECRET_KEY 等環境變數,今天在 server.py 內透過 os.environ 讀取,並在 FastAPI 啟動時把關鍵變數列出來,便於除錯。建議另寫一個 env.py 模組集中管理讀取邏輯:
# research-agent/src/research_agent/env.py
import os
from typing import Optional
def get_env(name: str, default: Optional[str] = None, *, required: bool = False) -> str:
"""集中讀取環境變數,缺少必要值時立刻拋錯,避免進入 LangGraph 才失敗。"""
value = os.environ.get(name, default)
if required and not value:
raise RuntimeError(f"環境變數 {name} 未設定,請在 .env 或部署環境注入。")
return value or ""
def require_keys() -> dict:
"""回傳研究助理啟動時必須有的金鑰字典。"""
return {
"model": get_env("RESEARCH_AGENT_MODEL", "gpt-placeholder"),
"openai": get_env("OPENAI_API_KEY", required=False),
"anthropic": get_env("ANTHROPIC_API_KEY", required=False),
"tavily": get_env("TAVILY_API_KEY", required=False),
"langfuse_public": get_env("LANGFUSE_PUBLIC_KEY", required=False),
"langfuse_secret": get_env("LANGFUSE_SECRET_KEY", required=False),
}
# 在 server.py 啟動時印出來,幫助除錯
@app.on_event("startup")
def announce_env() -> None:
keys = require_keys()
present = [k for k, v in keys.items() if v]
print(f"[research-agent] 已載入設定:{', '.join(present)}")
這段 env.py 把環境變數讀取集中起來,避免散落在各個端點裡。require_keys() 並不真的在缺少時讓服務啟動失敗——我們刻意用 required=False 預設,這樣在沒有任何金鑰的開發者機器上也能跑得動 /healthz。等到真的要呼叫模型時,llm.py 才會在缺金鑰的情況下報錯,這樣錯誤訊息更貼近真正出問題的地方。AG Day 39 容器化時,.env 檔案會透過 Docker secret 或環境變數傳入容器。
小結
今天我們把 research-agent 從 CLI 工具推進到 HTTP 服務。三個端點(POST /runs、GET /runs/{run_id}、POST /runs/{run_id}/resume)構成一個最小可用的研究任務 API,安全模組透過 FastAPI Depends 自動接到所有請求,/healthz 提供給 load balancer 與 Docker healthcheck 用。同步與 async 的橋接靠 run_in_executor 處理,並用 httpx 寫了完整的整合測試。
今天新增的關鍵詞:依賴注入(Depends)——FastAPI 在端點執行前自動呼叫的前處理函式;run_in_executor——把同步呼叫丟到 thread pool 的 asyncio API;OpenAPI——從 Pydantic schema 自動產生的 API 規格文件;ASGITransport——在測試裡直接呼叫 ASGI app、不需啟動 uvicorn 的機制。
順帶整理一下我們今天動過的檔案,方便你對照檢查:src/research_agent/server.py(新增,約 110 行)放了 FastAPI app 與三個核心端點;src/research_agent/env.py(新增)集中管理環境變數讀取;src/research_agent/cli.py(新增 serve 子命令)讓 uv run research-agent serve 也能啟動服務;tests/test_server.py(新增)用 httpx.ASGITransport 跑整合測試;既有 graph.py 與 memory.py 完全沒動,這代表今天的部署改動對核心圖邏輯零侵入,未來要升級 LangGraph 版本也不會卡在部署介面上。
另外,今天我們在 security_dependency() 裡選擇了「發現疑似注入字樣就回 400」這個策略,這對正式系統很合適;但在原型階段若你希望更寬鬆(例如只是想蒐集攻擊樣本),可以把那段換成「標記 flagged=True、照樣進入圖執行、並在 Langfuse 留下一個 warning span」。這個開關建議也寫成環境變數(例如 RESEARCH_AGENT_STRICT_SECURITY=true),方便在不同環境切換嚴格程度。
結語
現在 research-agent 已經是一個能對外提供 HTTP 服務的程式,但這個服務目前在開發者機器上跑得很順,到別的機器上卻未必——不同作業系統的 Python 版本、不同的系統套件版本、不同的環境變數設定都會造成差異。明天我們會進入「AG Day 39 容器化:Docker 打包 Agent 服務」,把今天寫的 FastAPI 服務整個打包進 Docker image,加上 uv 與 Python 的固定版本,再透過 docker compose 把 Langfuse、Tavily 這些外部依賴也串起來,讓整個系統在任何一台裝了 Docker 的機器上都能一鍵啟動。今天寫的 /healthz 端點正是 AG Day 39 會用到的 Docker healthcheck 介面,這個伏筆在部署章節會兌現。
延伸資源
- FastAPI 官方文件:
https://fastapi.tiangolo.com/。本篇採用的 Pydantic v2 schema、Depends、OpenAPI 自動產生都以官方文件為準。 - uvicorn 官方文件:
https://www.uvicorn.org/。--reload、--host、--port等參數的完整說明。 - httpx 官方文件:
https://www.python-httpx.org/。本篇測試使用的ASGITransport與AsyncClient介面。 - LangGraph 官方文件:Persistence 與 Interrupt 章節中關於
get_state()、Command(resume=...)的用法。
留言
張貼留言