跳到主要內容

AG Day 41 效能總檢與壓測

AG Day 41 效能總檢與壓測

執行需求：CPU 可跑。前幾篇把研究助理從 CLI 工具推進到 HTTP 服務、容器化部署、Streamlit 前端，整個系統看起來已經完整。但任何 production 服務在沒做過壓力測試之前，都只能說「能跑」，不能說「能扛」。今天的任務是用 AG Day 35 已經接好的 Langfuse trace 與 AG Day 39 已經跑起來的 docker compose 環境，設計一份壓測腳本（k6 或 locust 都可），把「十個並行使用者同時送出研究問題」、「LLM API 慢回應」、「SQLite 寫入瓶頸」這三個情境在可控條件下重現出來，並從 trace 中找出瓶頸節點。我們會用 Python 的 asyncio 與 httpx 寫一份輕量壓測器，搭配 matplotlib 把結果繪成圖表，這樣 AG Day 42 文件化時可以一起附上。

引言

壓力測試的目的不是「證明系統很快」，而是「找出系統在壓力下會先從哪裡崩潰」。對 research-agent 來說，瓶頸點可能散落在三個地方：FastAPI 的同步/非同步橋接（AG Day 38 已經用 run_in_executor 處理）、SQLite 寫入的鎖定（AG Day 31 留下來的問題）、以及最慢的元件——LLM API 本身的延遲。前兩個是我們自己程式碼的問題，可以靠最佳化與架構調整解決；第三個是外部依賴的問題，我們只能透過快取（AG Day 36）、模型分級、與併行處理去緩解。今天我們要在不修改 production 程式的情況下，從外部觀察這三個瓶頸的嚴重程度。

壓測工具的選擇很多：k6 適合純 HTTP 層的吞吐測試、locust 適合模擬真實使用者行為、ApacheBench 適合最簡單的單端點基準測試。我們今天刻意選用 Python 的 asyncio + httpx 自製壓測器，理由有三：第一是設定最簡單、不需安裝額外工具；第二是與既有 tests/ 結構一致、容易在 CI 跑；第三是我們可以同時打 FastAPI 與讀 Langfuse trace，方便做端對端分析。對單機 CPU 環境，這個自製壓測器每秒發 50–100 個請求已經足夠；真的要測更高負載，未來可以改用 k6。

今天的具體目標有三個。第一是「基準測試」：在沒有壓力的情況下，POST /runs、GET /runs/{run_id}、POST /runs/{run_id}/resume 三個端點的平均延遲、P95 延遲、與吞吐量。第二是「並行測試」：模擬 5 個、10 個、20 個並行使用者同時送出研究問題，觀察系統的吞吐量與錯誤率曲線。第三是「故障注入」：刻意把 LLM API 換成慢回應版本（例如每個呼叫加 2 秒延遲），觀察系統在依賴變慢時的退化模式，並找出 AG Day 38 留下的 run_in_executor 在慢回應下會不會阻塞整個 event loop。所有測試結果會被寫進 reports/perf-2026-08-24.json，方便 AG Day 42 的文件化章節引用。

原理／觀念

壓測的三個層次

壓力測試通常分三個層次。第一層是「基準測試（baseline）」：在沒有其他負載的情況下量測單一請求的延遲分布，這是後續比較的基準。第二層是「負載測試（load test）」：用預期的 production 流量（通常是基準測試的 5-10 倍）持續打一段時間，觀察系統是否能穩定維持目標吞吐量與延遲。第三層是「壓力測試（stress test）」：把流量推到超過預期峰值（10-20 倍），觀察系統從哪裡開始失敗、失敗模式是 graceful degradation 還是 catastrophic failure。我們今天會做第一層與第二層，第三層留給 production 環境的維運團隊。

延遲的百分位數比平均值有意義

壓測結果最常見的誤用是「平均延遲」。平均值會被幾個極端的長尾值拉高或拉低，無法反映真實使用者體驗。實務上我們看的是 P50（中位數）、P95（95% 的請求都比這快）、P99（99% 的請求都比這快）。對研究助理來說，「平均 3 秒完成」沒有意義，因為有 5% 的請求可能卡了 15 秒，使用者會以為系統壞了。AG Day 35 的 Langfuse 已經會自動計算每個 span 的 P50/P95/P99，今天的壓測腳本也會把這層資訊印出來，與 Langfuse 的數字交叉比對。

Little's Law 與系統容量估算

Little's Law 是一個簡單但實用的公式：系統中的平均請求數 = 吞吐量 × 平均延遲。對我們的系統來說，如果平均延遲是 5 秒、吞吐量目標是每秒 2 個請求，那系統中「同時在跑的請求數」大約是 10 個。這告訴我們 FastAPI 的 run_in_executor thread pool 至少要 10 個 worker 才能撐住目標吞吐量；AG Day 38 我們用預設的 None（表示用 asyncio 預設的 thread pool，5 * CPU 數量），對 8 核機器是 40 個 worker，足夠應付今天的壓測。AG Day 44 維運手冊會把這個估算寫成正式公式。

故障注入是測試彈性設計的捷徑

今天的故障注入不會真的去把 LLM API 弄壞（那需要 mock server），而是在壓測腳本裡把 RESEARCH_AGENT_DRY_RUN 環境變數打開、並把 LangGraph 圖換成一個「每次 invoke 都 sleep 2 秒」的假實作。這樣我們可以在不需要 API 金鑰的情況下，模擬 LLM API 慢回應的場景，觀察系統的退化模式。故障注入的價值在於：當外部依賴出問題時，你的系統是「慢慢變慢」還是「整個停擺」，這兩種故障模式對使用者體驗與維運成本完全不同。

完整實作

今天的程式集中在兩個檔案：scripts/loadtest.py（壓測器本體）與 scripts/summarize_perf.py（壓測結果彙整）。我們先建立檔案：

mkdir -p research-agent/scripts
touch research-agent/scripts/loadtest.py
touch research-agent/scripts/summarize_perf.py

第一步：寫壓測器本體。我們用 asyncio.Semaphore 控制並行數、用 httpx.AsyncClient 送請求、用簡單的 list 收集每次請求的延遲：

# research-agent/scripts/loadtest.py
from __future__ import annotations
import argparse
import asyncio
import json
import statistics
import time
from pathlib import Path

import httpx


async def _one_request(client: httpx.AsyncClient, base: str, question: str) -> dict:
    """送出一次研究問題，並量測端對端延遲。"""
    started = time.perf_counter()
    try:
        resp = await client.post(f"{base}/runs", json={"question": question}, timeout=30)
        elapsed = (time.perf_counter() - started) * 1000
        return {
            "status": resp.status_code,
            "elapsed_ms": elapsed,
            "run_id": resp.json().get("run_id") if resp.status_code == 201 else None,
            "ok": resp.status_code == 201,
        }
    except httpx.HTTPError as exc:
        return {"status": "error", "elapsed_ms": -1, "ok": False, "error": str(exc)}


async def _poll_until_done(client: httpx.AsyncClient, base: str, run_id: str, max_wait: float = 30.0) -> dict:
    """輪詢直到 run 完成或逾時，回傳最終狀態與總耗時。"""
    started = time.perf_counter()
    while (time.perf_counter() - started) < max_wait:
        resp = await client.get(f"{base}/runs/{run_id}", timeout=10)
        if resp.status_code != 200:
            return {"status": "error", "elapsed_ms": -1}
        data = resp.json()
        if data["status"] in {"completed", "failed", "interrupted"}:
            elapsed = (time.perf_counter() - started) * 1000
            return {"status": data["status"], "elapsed_ms": elapsed, "ok": data["status"] == "completed"}
        await asyncio.sleep(0.5)
    return {"status": "timeout", "elapsed_ms": max_wait * 1000, "ok": False}


async def run_scenario(base: str, *, total: int, concurrency: int, scenario: str) -> dict:
    """執行一個壓測情境：發出 total 個請求，concurrency 同時進行。"""
    semaphore = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        async def worker(idx: int) -> dict:
            async with semaphore:
                question = f"[{scenario}] 壓測問題 {idx}：請研究某個示範主題。"
                create = await _one_request(client, base, question)
                if create["ok"] and create["run_id"]:
                    polled = await _poll_until_done(client, base, create["run_id"])
                    return {**create, "final": polled}
                return create
        tasks = [asyncio.create_task(worker(i)) for i in range(total)]
        results = await asyncio.gather(*tasks)

    ok_count = sum(1 for r in results if r.get("ok"))
    elapsed = [r["elapsed_ms"] for r in results if r["elapsed_ms"] > 0]
    p50 = statistics.median(elapsed) if elapsed else 0
    p95 = sorted(elapsed)[int(len(elapsed) * 0.95)] if len(elapsed) >= 20 else 0
    return {
        "scenario": scenario,
        "total": total,
        "concurrency": concurrency,
        "ok": ok_count,
        "fail": total - ok_count,
        "p50_ms": round(p50, 1),
        "p95_ms": round(p95, 1),
        "max_ms": round(max(elapsed) if elapsed else 0, 1),
    }


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base", default="http://127.0.0.1:8000")
    parser.add_argument("--total", type=int, default=20)
    parser.add_argument("--concurrency", type=int, default=5)
    parser.add_argument("--scenario", default="baseline")
    parser.add_argument("--out", default="reports/perf-latest.json")
    args = parser.parse_args()

    summary = asyncio.run(run_scenario(
        args.base,
        total=args.total,
        concurrency=args.concurrency,
        scenario=args.scenario,
    ))
    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(summary, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

第二步：寫故障注入版的圖實作。我們新增 src/research_agent/fake_graph.py，提供可控制延遲的假 LangGraph 圖，專門給壓測與故障演練使用：

# research-agent/src/research_agent/fake_graph.py
"""壓測用的假 LangGraph 圖：用環境變數控制延遲，模擬 LLM API 慢回應。"""
from __future__ import annotations
import asyncio
import os

from langchain_core.messages import AIMessage

from research_agent.security import redact


LATENCY_MS = int(os.environ.get("RESEARCH_AGENT_FAKE_LATENCY_MS", "0"))
FAIL_RATE = float(os.environ.get("RESEARCH_AGENT_FAKE_FAIL_RATE", "0.0"))


class FakeGraph:
    """延遲可控、隨機失敗的假圖；呼叫介面與 LangGraph 編譯後的圖一致。"""

    def __init__(self) -> None:
        self.invocations: list[str] = []

    def invoke(self, payload, config):
        import random
        self.invocations.append(str(config))
        if FAIL_RATE > 0 and random.random() < FAIL_RATE:
            raise RuntimeError("fake graph 隨機失敗（測試用）")
        asyncio.get_event_loop().run_until_complete(asyncio.sleep(LATENCY_MS / 1000))
        question = payload["messages"][0].content
        return {
            "messages": payload["messages"] + [
                AIMessage(content=redact(f"（fake）對「{question}」的示範回應。")),
            ]
        }

    def get_state(self, config):
        class Snap:
            values = {"messages": []}
            next = ()
        return Snap()

第三步：在 server.py 加一個環境變數 hook，讓 RESEARCH_AGENT_FAKE_GRAPH=true 時 build_graph() 自動換成假實作：

# research-agent/src/research_agent/server.py（節錄）
def _build():
    if os.environ.get("RESEARCH_AGENT_FAKE_GRAPH") == "true":
        from research_agent.fake_graph import FakeGraph
        return FakeGraph()
    return build_graph_from_core()


# 把原本所有 _invoke_graph 內的 build_graph(checkpointer=...) 換成 _build()

第四步：寫結果彙整腳本，把多個壓測情境的結果彙整成一張圖表。我們用 matplotlib 畫 P50/P95 延遲對比、用文字模式列出錯誤率：

# research-agent/scripts/summarize_perf.py
"""把多個壓測情境的 JSON 結果彙整成一張延遲對比圖。"""
from __future__ import annotations
import argparse
import json
from pathlib import Path


def load_summaries(path: Path) -> list[dict]:
    out = []
    for fp in sorted(path.glob("perf-*.json")):
        out.append({"file": fp.name, **json.loads(fp.read_text(encoding="utf-8"))})
    return out


def print_table(rows: list[dict]) -> None:
    print(f"{'scenario':18s} {'total':>6s} {'conc':>5s} {'ok':>5s} {'p50(ms)':>10s} {'p95(ms)':>10s}")
    for r in rows:
        print(f"{r['scenario']:18s} {r['total']:>6d} {r['concurrency']:>5d} {r['ok']:>5d} {r['p50_ms']:>10.1f} {r['p95_ms']:>10.1f}")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dir", default="reports", help="壓測 JSON 結果目錄")
    args = parser.parse_args()
    rows = load_summaries(Path(args.dir))
    if not rows:
        print("找不到 perf-*.json；請先跑 loadtest.py 產生結果")
        return
    print_table(rows)


if __name__ == "__main__":
    main()

第五步：用一段 orchestrator 腳本，依序跑「baseline」「concurrency 5」「concurrency 10」「slow backend」四個情境，把結果都寫進 reports/perf-*.json：

# research-agent/scripts/run_all_scenarios.sh
#!/usr/bin/env bash
set -euo pipefail

BASE="http://127.0.0.1:8000"
OUT_DIR="reports"

mkdir -p "$OUT_DIR"

echo "[1/4] baseline (1 user, 10 reqs)..."
uv run python scripts/loadtest.py --base "$BASE" --total 10 --concurrency 1 \
    --scenario baseline --out "$OUT_DIR/perf-baseline.json"

echo "[2/4] concurrency 5 (5 users, 20 reqs)..."
uv run python scripts/loadtest.py --base "$BASE" --total 20 --concurrency 5 \
    --scenario conc-5 --out "$OUT_DIR/perf-conc-5.json"

echo "[3/4] concurrency 10 (10 users, 20 reqs)..."
uv run python scripts/loadtest.py --base "$BASE" --total 20 --concurrency 10 \
    --scenario conc-10 --out "$OUT_DIR/perf-conc-10.json"

echo "[4/4] slow backend (RESEARCH_AGENT_FAKE_LATENCY_MS=2000)..."
RESEARCH_AGENT_FAKE_GRAPH=true \
RESEARCH_AGENT_FAKE_LATENCY_MS=2000 \
uv run python scripts/loadtest.py --base "$BASE" --total 10 --concurrency 5 \
    --scenario slow-backend --out "$OUT_DIR/perf-slow.json"

echo "全部完成；執行 summarize_perf.py："
uv run python scripts/summarize_perf.py --dir "$OUT_DIR"

離線示範輸出（用 fake graph 模擬，所有延遲都是 fake 的，僅示範格式）：

$ bash scripts/run_all_scenarios.sh
[1/4] baseline (1 user, 10 reqs)...
{
  "scenario": "baseline",
  "total": 10,
  "concurrency": 1,
  "ok": 10,
  "fail": 0,
  "p50_ms": 142.7,
  "p95_ms": 198.4,
  "max_ms": 215.0
}
[2/4] concurrency 5 ...
[3/4] concurrency 10 ...
[4/4] slow backend ...

scenario             total  conc     ok    p50(ms)    p95(ms)
baseline                10     1     10      142.7      198.4
conc-5                  20     5     20      158.2      220.5
conc-10                 20    10     20      162.4      235.1
slow-backend            10     5     10     2032.4     2085.0

第六步：用 Langfuse 的查詢 API 讀出壓測期間的 trace，計算每個 span 的 P50/P95 延遲。我們把這個分析也寫成一支小腳本，把 Langfuse 數字與壓測結果交叉比對：

# research-agent/scripts/langfuse_perf.py
"""從 Langfuse 讀出壓測期間的 trace，計算每個 span 的 P50/P95。"""
from __future__ import annotations
import os
import statistics
from typing import Iterable

from langfuse import Langfuse


LANGFUSE = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)


def percentiles(values: Iterable[float]) -> dict:
    vs = sorted(values)
    if not vs:
        return {"p50": 0, "p95": 0}
    p50 = vs[len(vs) // 2]
    p95 = vs[int(len(vs) * 0.95)] if len(vs) >= 20 else vs[-1]
    return {"p50": round(p50, 1), "p95": round(p95, 1)}


def main() -> None:
    spans = LANGFUSE.fetch_spans(tags=["loadtest"])
    by_name: dict[str, list[float]] = {}
    for span in spans:
        duration_ms = (span.end_time - span.start_time).total_seconds() * 1000
        by_name.setdefault(span.name, []).append(duration_ms)

    print(f"{'span':24s} {'count':>6s} {'p50(ms)':>10s} {'p95(ms)':>10s}")
    for name, durations in sorted(by_name.items()):
        stats = percentiles(durations)
        print(f"{name:24s} {len(durations):>6d} {stats['p50']:>10.1f} {stats['p95']:>10.1f}")


if __name__ == "__main__":
    main()

這支腳本會在壓測結束後跑一次，把 Langfuse 觀察到的 span 等級延遲印出來。我們可以從中看出：到底是 FastAPI 的入口慢、還是 LangGraph 的節點慢、還是模型呼叫本身慢，進而決定要最佳化哪一塊。

常見錯誤與踩雷

第一個雷是「壓測環境與 production 環境不一致」。我們今天刻意用 docker compose 跑 production 版的服務（AG Day 39 的 image），確保壓測結果能反映真實情況；但很多團隊壓測時跑的是開發版（例如 uvicorn --reload 模式），這個模式多了檔案監聽、多了 debug middleware，量到的數字會比 production 好很多。建議 CI 上的壓測永遠跑 production image，這樣數字才有意義。如果你的 production 用 Kubernetes，也可以用 kind 或 minikube 在 CI 上跑一份輕量的壓測環境；不必追求 100% 等價，只要確保網路拓樸與資源配置接近即可。

第二個雷是「忘記 warmup」。冷啟動的服務第一次呼叫通常比較慢（imports、連線池初始化），這個 cold start 不該算在平均值裡。我們今天的 loadtest.py 預設就把所有請求算進去；如果你想要更精準，可以先發 5 個 warmup 請求再開始計時。

第三個雷是「壓測結果寫進了 production 資料庫」。我們的壓測腳本送出的問題會真的進入 LangGraph 圖、被真的檢索、被真的寫進 SQLite。對 demo 沒問題，但對 production 會污染真實資料。建議在壓測時用獨立的 RESEARCH_AGENT_DRY_RUN=true 環境變數、或獨立的 named volume mount，讓壓測資料與 production 資料完全隔離。如果你的 production 還在用同一個 Langfuse 專案接收 trace，也要記得用 tag 把壓測 trace 過濾掉，否則一段時間後就會被假資料淹沒。

第四個雷是「壓測本身造成 LLM API 帳單飆升」。如果 production 環境真的在接 OpenAI API，每次壓測都會消耗 token。我們今天刻意用 fake graph 繞過 LLM API，這也是為什麼 fake graph 是必要的；真實 production 壓測前務必確認 RESEARCH_AGENT_FAKE_GRAPH=true 已經設好。

第五個雷是「只看吞吐量、不看資源使用」。今天我們只量了延遲與錯誤率，但 CPU、記憶體、網路 I/O 的數字也很重要。建議同時跑 docker stats，把容器資源使用率也記錄下來，這樣 AG Day 44 維運手冊裡的容量規劃才有依據。

效能與實務提醒

實務上最容易踩到的效能問題是「asyncio 預設 thread pool 不夠大」。AG Day 38 我們用 loop.run_in_executor(None, ...) 跑同步的 LangGraph 圖，None 表示用 asyncio 預設的 executor。這個預設 executor 大小是 min(32, CPU + 4)，對 8 核機器是 12。如果壓測發現 FastAPI 端開始拒絕服務、但 LangGraph 還在跑，第一個要查的就是 thread pool 是否被打滿。可以透過 loop.set_default_executor(ThreadPoolExecutor(max_workers=64)) 調大。

第二個提醒是「SQLite 在高並行下的鎖定」。AG Day 31 的 SQLite checkpointer 在單一寫入下沒問題，但 10 個並行使用者同時寫入時，會頻繁出現 database is locked 錯誤。壓測時如果發現這個錯誤率隨並行數線性上升，就該考慮換成 Postgres 或加連線池。AG Day 39 的 docker compose 已經預留 Postgres 設定，必要時只要把 RESEARCH_AGENT_CHECKPOINTER_URL 換成 postgresql://... 就能切換。

第三個提醒是「壓測結果要搭配 Langfuse trace 看」。單純的 HTTP 延遲數字看不出瓶頸在 FastAPI、LangGraph 還是 LLM API。今天我們用 Langfuse 的 span 等級資料交叉比對，這對調校最有幫助。實務上我們會在 Langfuse 上設一個「慢於 5 秒的 trace 自動告警」規則，這樣壓測期間如果某個 span 突然變慢，會立刻收到通知。如果發現「每個 span 都很快、但總延遲很長」，那瓶頸通常出在「模型間的等待」或「外部 API 的限流」，這時候 Langfuse 的 timeline 視圖會比單純的 span 等級數據更有用。

第四個提醒是「壓測腳本本身可能會被當成 production 流量」。如果你的 Langfuse 設定是「所有 trace 都寫入」，壓測的 trace 也會被寫進去，造成 Langfuse 介面被假資料淹沒。建議在 Langfuse SDK 加上 tags=["loadtest"] 標記，並在 Langfuse 介面上用 tag filter 把壓測 trace 過濾掉。今天的範例已經加上 tag，記得在 production 環境也沿用這個習慣。

第五個提醒是「定期重跑壓測」。效能會隨著依賴版本、模型版本、流量模式而變化，建議每個月或每次重大改版後跑一次今天寫的壓測腳本。AG Day 44 維運手冊會把這個排程寫進 CI 流程（每月第一個週一凌晨自動跑），壓測結果自動發到維運頻道。我們今天刻意把 perf_regression_check.py 寫成可單獨執行的 CLI，這樣日後接到 CI runner 不需要改任何程式碼，直接加一個 cron job 就能用。

小結

今天我們用 Python 的 asyncio 與 httpx 自製了一份壓測器，搭配 FakeGraph 做故障注入，並把壓測期間的 trace 與 Langfuse 交叉比對找出瓶頸節點。整個流程包含「基準測試 → 並行測試 → 故障注入」三個層次，結果會自動寫進 reports/perf-*.json 並用 summarize_perf.py 彙整成表格。這份壓測腳本是 AG Day 44 維運手冊裡「容量規劃」一節的核心輸入。

今天新增的關鍵詞：基準測試（baseline）——沒有其他負載下的單一請求延遲；負載測試（load test）——用預期峰值的流量持續打系統；百分位數（P50/P95/P99）——比平均值更能反映真實使用者體驗的延遲指標；Little's Law——吞吐量、平均延遲、同時在跑的請求數三者的關係公式；故障注入（fault injection）——刻意模擬依賴失敗來測試系統彈性的方法。

第七步：寫一個簡單的 pytest，把今天的壓測器本體做最小化測試。我們確保 _one_request() 與 _poll_until_done() 兩個核心函式在錯誤情境下也能穩定回傳結構化資料：

# research-agent/tests/test_loadtest.py
import pytest

from research_agent.scripts.loadtest import _poll_until_done  # type: ignore  # path-based import


class FakeClient:
    """模擬 httpx.AsyncClient 的極簡版。"""

    def __init__(self, responses):
        self.responses = responses
        self.calls = 0

    async def get(self, url, timeout):
        self.calls += 1
        if self.calls > len(self.responses):
            raise RuntimeError("no more responses")
        return self.responses[self.calls - 1]


class FakeResp:
    def __init__(self, payload, status=200):
        self._payload = payload
        self.status_code = status

    def json(self):
        return self._payload


@pytest.mark.asyncio
async def test_poll_returns_when_completed():
    client = FakeClient([FakeResp({"status": "completed"}), FakeResp({"status": "completed"})])
    out = await _poll_until_done(client, "http://x", "run-1", max_wait=2.0)
    assert out["status"] == "completed"
    assert out["ok"] is True


@pytest.mark.asyncio
async def test_poll_times_out(monkeypatch):
    import asyncio
    async def fake_sleep(_):
        return None
    monkeypatch.setattr(asyncio, "sleep", fake_sleep)

    client = FakeClient([FakeResp({"status": "running"})] * 5)
    out = await _poll_until_done(client, "http://x", "run-1", max_wait=0.1)
    assert out["status"] == "timeout"
    assert out["ok"] is False

第八步：把 reports/perf-*.json 接到 GitHub Actions（或其他 CI），做成自動化的「效能回歸測試」。我們比對新一次的 P95 與基準的差距，超過 20% 就讓 CI 紅燈：

# research-agent/scripts/perf_regression_check.py
"""比對最新壓測結果與基準，超過門檻則視為效能回歸。"""
from __future__ import annotations
import json
import sys
from pathlib import Path


BASELINE = Path("reports/perf-baseline.json")
LATEST = Path("reports/perf-conc-10.json")
THRESHOLD_PCT = 20.0


def main() -> int:
    if not BASELINE.exists() or not LATEST.exists():
        print("找不到基準或最新結果，跳過回歸檢查")
        return 0
    base = json.loads(BASELINE.read_text(encoding="utf-8"))
    latest = json.loads(LATEST.read_text(encoding="utf-8"))
    delta = (latest["p95_ms"] - base["p95_ms"]) / base["p95_ms"] * 100
    if delta > THRESHOLD_PCT:
        print(f"[FAIL] P95 延遲比基準慢 {delta:.1f}%，超過 {THRESHOLD_PCT}% 門檻")
        return 1
    print(f"[OK] P95 延遲差距 {delta:.1f}%，未超過門檻")
    return 0


if __name__ == "__main__":
    sys.exit(main())

結語

今天的壓測把研究助理的能見度從「能跑」推進到「能扛」。但任何一個 production 系統都必須有人能接手維運。接下來 AG Day 42 會進入「文件與交接：README 與架構圖」：把所有今天的設定、壓測結果、部署指令彙整成一份新工程師能在半天內接手的 README，並畫一張 ASCII 架構圖把整個系統的元件關係視覺化。今天 reports/perf-*.json 的內容會直接被 README 的「容量規劃」段落引用，這是文件化章節的第一個具體產出。AG Day 42 結束時，你手上應該會有一份能直接交給同事或未來自己的自己完整文件。順帶把今天動過的檔案做個總結：scripts/loadtest.py（新增，壓測器本體）、scripts/summarize_perf.py（新增，結果彙整）、scripts/run_all_scenarios.sh（新增，批次跑四個情境）、scripts/langfuse_perf.py（新增，Langfuse 跨對）、scripts/perf_regression_check.py（新增，CI 門檻）、src/research_agent/fake_graph.py（新增，故障注入假圖）、tests/test_loadtest.py（新增，壓測器單元測試）、src/research_agent/server.py（小幅 hook 環境變數）。

明天，我們會進入「AG Day 42 文件與交接：README 與架構圖」，把今天產出的壓測數據、AG Day 39 的 docker compose 設定、AG Day 40 的前端指令，全部彙整進一份新工程師能在半天內接手的 README，並畫一張 Mermaid 架構圖把整個系統的元件關係視覺化。

延伸資源

  • Python asyncio 官方文件：https://docs.python.org/3/library/asyncio.html。Semaphore、gather、run_in_executor 的完整介面。
  • httpx 官方文件：https://www.python-httpx.org/。AsyncClient 與 timeout 設定。
  • k6 官方文件：https://k6.io/docs/。若未來需要更高負載的壓測，可改用 k6。
  • Langfuse 官方文件：https://langfuse.com/docs。fetch_spans 與 tag 過濾的查詢介面。
標籤：

留言

張貼留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中，變數與資料型別是處理資料的基礎。變數是存放資料的容器，資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別，是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數，以及常見的資料型別，包括整數、浮點數、字串、布林值，還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議，以及如何用 type() 檢查資料型別。 什麼是變數？如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數，我們可以給一筆資料一個名字，並在程式的其他地方用這個名字取用該筆資料。在 Python 中，變數不需要事先宣告型別，因為 Python 是動態型別語言，變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單，只要用賦值符號 = 把值指定給變數即可。例如： x = 5 # 定義變數 x，並把整數 5 賦值給它 name = "Alice" # 定義變數 name，並把字串 "Alice" 賦值給它 在這裡，x 是一個變數，被賦予整數 5；name 是另一個變數，被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改，也就是說，我們可以在程式的不同地方給同一個變數新的值。例如： x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求，在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同，定義變數時不需要宣告型別。賦值時，Python 會根據值自動判斷變數的型別。例如： x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值，這是 Python 的彈性之一。 常見資料型別 在 Python 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡，程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者，或是想踏入人工智慧（AI）領域，學會寫程式都能明顯提升你的競爭力。在眾多程式語言中，Python 因為語法簡單、功能強大、應用範圍廣泛，成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢，並一步步教你在不同系統上安裝與設定 Python 開發環境，最後寫出第一支 Python 程式。 為什麼選擇 Python？ Python 是一種高階程式語言，由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性，並用縮排來定義程式區塊，這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇；就算是經驗豐富的開發者，也能用它完成複雜的專案。 Python 的優勢如下： 簡單易學 ：Python 的語法清楚、結構簡潔，初學者很快就能上手。和其他語言相比，學習曲線相對平緩，不需要先弄懂一堆複雜觀念，就能開始寫程式。 應用範圍廣泛 ：從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲，Python 都有大量開源函式庫與工具支援，而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 ：Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas；開發網站有 Django、Flask；做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件，讓開發更有效率。 跨平台支援 ：Python 支援 Windows、macOS、Linux 等作業系統，程式通常不需要太多修改就能跨平台執行，讓開發與部署更有彈性。 活躍的社群 ：Python 擁有龐大的開發者社群。學習或開發上遇到問題，幾乎都能在社群與論壇（例如 Stack Overflow）找到答案，對初學者來說是很強的後盾，也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能，讓許多領域都開始大量使用它。以下是幾個常見的應用方向： 資料科學 ：隨著大數據與人工智慧興起，資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...
張貼留言
閱讀完整內容

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門 隨著人工智慧（AI）與深度學習（Deep Learning）快速發展，越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者，還是已經有程式基礎的開發者，學好 Python 與深度學習框架（例如 PyTorch），都能為你打開更多可能。 為什麼選擇 Python？ Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手，而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化，還是建立機器學習與深度學習模型，Python 都能勝任。對想進入 AI 或資料科學領域的人來說，它幾乎是必備工具。 PyTorch 是什麼？ PyTorch 是由 Meta（原 Facebook）AI 研究團隊開發的開源深度學習框架，以易用、靈活和動態計算圖著稱，是許多 AI 研究人員與開發者的首選。相較於其他框架，PyTorch 的寫法更貼近原生 Python，對初學者相對友善。無論是簡單的實驗，還是複雜的深度學習模型，PyTorch 都能提供強大的支援。 這個系列能帶給你什麼？ 這個系列會從 Python 的基礎開始，帶你一步一步學習，最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式，也能跟著文章的節奏累積技能，理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎：從變數、條件判斷到函式與模組。 資料處理工具：用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化：用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎：線性代數、微積分與機率。 PyTorch 入門：理解張量、模型建構與 GPU 加速。 基礎深度學習模型：CNN 與 RNN 的實作應用。 深度學習專案實戰：從資料前處理到模型部署的端到端流程。 誰適合這個系列？ 程式初學者 ：如果你對 AI 充滿好奇，卻還沒寫過程式，系列的第一部分會帶你快速上手 Python，並幫助你理解深度學習的基本觀念。 資料科學愛好者 ：如果你已經熟悉一些資料處理方法，進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容