💡 導讀：五張研究申請表，為什麼不能全部擠到同一個窗口？

Day 17，我們學習從歷史 Atlas 摘要辨認單一 SNV 的 AVI 與特徵歸因；Day 18，我們學習讀取同一變異的歷史 REF／ALT 剪接預測。現在研究員可能會問：「如果我有幾百筆候選變異，要讓 Python 一筆一筆等嗎？服務忽然忙碌又該怎麼辦？」

想像五個人到櫃台辦事：窗口有限、有人文件有誤、有人碰上系統暫停。如果每個錯誤都無限制地重新排隊，服務可能更擁擠。程式需要會排隊、辨認錯誤、控制重試次數、保留失敗理由。這正是今天的 asyncio 練習。

本篇先讓 Google AI Studio 做一個看得見的「批次任務狀態面板」，再用 Python 反向拆解面板背後的工作流程。這是研究用原型，不是已可部署的 AlphaGenome 正式批次客戶端。

一、核心概念：四種情況，四種處理方式

要特別區分三個容易混淆的概念：

• asyncio：讓程式在等待期間安排其他任務；不代表免費增加官方 API 配額。
• asyncio.Semaphore(2)：最多同時安排兩筆本地模擬請求；這是並行上限，不是每秒最多兩筆的速率限制。
• Exponential backoff（指數退避）：失敗後逐次增加等待時間；本文用 0.01 → 0.02 秒這類極短的教學時間，不是 Google 官方建議的正式參數。

本篇 429、503 與逾時均由我們自行寫的 Fake API 產生。 Reboot 已有 Atlas／Model API 的歷史成功證據，但目前提供的稽核資料沒有證明這次的 503、429 重試流程曾在真實 AlphaGenome 服務成功執行。

二、Google AI Studio：先建立批次任務 MVP

在 AI Studio 建立畫面前，可以先貼入以下提示詞：

請建立繁體中文、Research Use Only 的「基因變異批次查詢研究面板」。
先只使用 SYNTHETIC / MOCK 資料，不連真實 AlphaGenome API。

畫面包含：
- 五筆人工變異清單（必須標明 chrToy，非真實基因組座標）
- 每筆的狀態：WAITING / RUNNING / RETRYING / OK / FAILED
- 嘗試次數、模擬 HTTP 狀態、等待秒數、失敗原因
- 最大並行數與最大重試次數設定
- 最終 JSON 審閱紀錄與來源標示

所有模擬分數標為 SYNTHETIC_SCORE；未取得分數顯示 null，不補零。
不使用任何病人資料，不展示致病風險、ACMG/AMP 診斷或治療建議。
不要把 API Key 放進前端；即使有可輸入 Key 的欄位，也先停用 LIVE 模式。

接著從畫面倒推 Python 模組：

AI Studio 產生 React／網頁原型之後，不是把網頁程式一鍵轉成 Python；而是看懂每個欄位的資料契約，再用 Python 重新做出相同流程。

💻 三、今天動手做：完整的離線 Python Mini Demo

請將以下全部內容存成 day19_offline_asyncio_lab.py。只使用 Python 3.12 標準函式庫。它沒有 HTTP 客戶端、沒有 alphagenome 匯入、不讀取 .env，也沒有能啟用 LIVE 的開關。

"""Day 19: deterministic OFFLINE-only asyncio retry/limit laboratory.

Research education only. No SDK, HTTP client, credentials, or live API access.
HTTP-like 429/503/400 statuses and TIMEOUT are entirely simulated.
"""
from __future__ import annotations

import argparse
import asyncio
import json
from pathlib import Path

# These variants and numbers are invented. They are NOT GRCh38 or API outputs.
SCENARIOS = {
    "chrToy:101:A>G": [
        {"code": 429, "retry_after_s": 0.03},
        {"code": 200, "synthetic_score": 0.12},
    ],
    "chrToy:102:C>T": [
        {"code": 503}, {"code": 503},
        {"code": 200, "synthetic_score": -0.21},
    ],
    "chrToy:103:G>A": [{"code": 400}],
    "chrToy:104:T>C": [{"code": 503}, {"code": 503}, {"code": 503}],
    "chrToy:105:A>C": [
        {"code": "TIMEOUT"}, {"code": 200, "synthetic_score": 0.07},
    ],
}
RETRYABLE = {429, 503, "TIMEOUT"}


class FakeAPI:
    """One preprogrammed local response per attempt; never makes network calls."""

    def __init__(self, scenarios=None):
        self.scenarios = SCENARIOS if scenarios is None else scenarios
        self.in_flight = 0
        self.max_in_flight = 0

    async def query(self, variant, attempt):
        self.in_flight += 1
        self.max_in_flight = max(self.max_in_flight, self.in_flight)
        try:
            # A tiny local delay permits asyncio tasks to overlap for teaching.
            await asyncio.sleep(0.005)
            outcomes = self.scenarios[variant]
            if attempt > len(outcomes):
                raise ValueError("The teaching scenario has no such attempt")
            return outcomes[attempt - 1]
        finally:
            self.in_flight -= 1


def delay_for(code, attempt, retry_after_s=None):
    """Teaching backoff: 0.01, 0.02,... seconds, max 0.08 seconds."""
    delay = min(0.08, 0.01 * (2 ** (attempt - 1)))
    if code == 429 and isinstance(retry_after_s, (int, float)) and retry_after_s >= 0:
        delay = max(delay, retry_after_s)
    return round(delay, 3)


async def query_with_retry(variant, fake_api, semaphore, max_attempts=3, sleeper=asyncio.sleep):
    """Retry only retryable mock errors; release semaphore while waiting."""
    history = []
    waits = []
    for attempt in range(1, max_attempts + 1):
        # This semaphore limits simultaneous requests, NOT requests per second.
        async with semaphore:
            response = await fake_api.query(variant, attempt)
        code = response["code"]
        history.append(code)

        if code == 200:
            return {"variant": variant, "status": "OK", "attempts": attempt,
                    "history": history, "retry_wait_s": waits,
                    "synthetic_score": response["synthetic_score"]}
        if code not in RETRYABLE:
            return {"variant": variant, "status": "NOT_RETRYABLE", "attempts": attempt,
                    "history": history, "retry_wait_s": waits,
                    "synthetic_score": None}
        if attempt == max_attempts:
            return {"variant": variant, "status": "RETRY_EXHAUSTED", "attempts": attempt,
                    "history": history, "retry_wait_s": waits,
                    "synthetic_score": None}

        wait = delay_for(code, attempt, response.get("retry_after_s"))
        waits.append(wait)
        await sleeper(wait)

    raise AssertionError("Unreachable")


async def run_batch(max_concurrency=2, max_attempts=3, scenarios=None, sleeper=asyncio.sleep):
    """Simulate a batch without keys, SDK, internet, or patient records."""
    if not 1 <= max_concurrency <= 8:
        raise ValueError("max_concurrency must be 1..8")
    if not 1 <= max_attempts <= 5:
        raise ValueError("max_attempts must be 1..5")
    api = FakeAPI(scenarios)
    semaphore = asyncio.Semaphore(max_concurrency)
    # gather() returns results in input order even if tasks complete differently.
    rows = await asyncio.gather(*(
        query_with_retry(v, api, semaphore, max_attempts, sleeper)
        for v in api.scenarios
    ))
    return {
        "evidence_class": "SYNTHETIC_FIXTURE",
        "mode": "OFFLINE_ONLY",
        "sdk_used": False,
        "real_api_calls_this_run": 0,
        "clinical_interpretation": False,
        "scenario_count": len(rows),
        "success_count": sum(r["status"] == "OK" for r in rows),
        "failure_count": sum(r["status"] != "OK" for r in rows),
        "total_attempts": sum(r["attempts"] for r in rows),
        "max_concurrency_limit": max_concurrency,
        "observed_max_mock_in_flight": api.max_in_flight,
        "sum_scheduled_retry_wait_s": round(
            sum(sum(r["retry_wait_s"]) for r in rows), 3),
        "results": rows,
    }


def main(argv=None):
    parser = argparse.ArgumentParser(description="Day19 OFFLINE asyncio mock retry lab")
    parser.add_argument("--mode", choices=["offline"], default="offline")
    parser.add_argument("--max-concurrency", type=int, default=2)
    parser.add_argument("--max-attempts", type=int, default=3)
    parser.add_argument("--output-dir", default="day19_output")
    args = parser.parse_args(argv)
    try:
        payload = asyncio.run(run_batch(args.max_concurrency, args.max_attempts))
    except ValueError as exc:
        parser.error(str(exc))
    folder = Path(args.output_dir)
    folder.mkdir(parents=True, exist_ok=True)
    (folder / "day19_mock_batch.json").write_text(
        json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    summary = {"mode": payload["mode"], "scenario_count": payload["scenario_count"],
               "success_count": payload["success_count"],
               "failure_count": payload["failure_count"],
               "total_attempts": payload["total_attempts"],
               "real_api_calls_this_run": payload["real_api_calls_this_run"],
               "output_file": str(folder / "day19_mock_batch.json")}
    print(json.dumps(summary, ensure_ascii=False, sort_keys=True))


if __name__ == "__main__":
    main()

▶ 執行方式

將 .py 儲存在 Windows Downloads，於 WSL Ubuntu 執行：

cd /mnt/c/Users/user/Downloads
python3.12 day19_offline_asyncio_lab.py \
  --mode offline \
  --max-concurrency 2 \
  --max-attempts 3 \
  --output-dir day19_output

在 Downloads 會產生 day19_output/day19_mock_batch.json，保存五筆人工請求的所有嘗試、模擬錯誤與等待紀錄。沒有真實 Atlas AVI 或 RNA 預測分數。

▶ 實測輸出與如何閱讀

我在本次工作環境以 Python 3.13.5 執行上述邏輯，stdout 為：

{"failure_count": 2, "mode": "OFFLINE_ONLY", "output_file": "day19_output/day19_mock_batch.json", "real_api_calls_this_run": 0, "scenario_count": 5, "success_count": 3, "total_attempts": 11}

你可以在 JSON 中找到以下五筆結果：

合計 5 筆情境、11 次本地模擬嘗試、3 筆成功、2 筆失敗。這裡的「成功」是 Fake API 腳本按預先設定回傳 200，不是 3 筆變異完成 AlphaGenome 模型預測。

sum_scheduled_retry_wait_s=0.10 是所有任務安排的等待秒數加總，不等於程式的實際牆鐘執行時間；不同任務可能同時等待。observed_max_mock_in_flight=2 則證明這次模擬最多同時處理兩個查詢，但不代表符合任何真實官方配額。

▶ 讀者可以自己改的兩個參數

# 降低最大並行數，觀察輸出中的 observed_max_mock_in_flight
python3.12 day19_offline_asyncio_lab.py --max-concurrency 1 --output-dir day19_one_at_a_time

# 縮短重試上限；部分原本第三次才成功的情境將失敗
python3.12 day19_offline_asyncio_lab.py --max-attempts 2 --output-dir day19_only_two_attempts

# 這支腳本不支援 LIVE：應拒絕參數，而不是偷偷連線
python3.12 day19_offline_asyncio_lab.py --mode live

最後一行的預期行為是命令列參數錯誤（非零退出碼），不是呼叫真實 API。

四、從「教學 Fake API」升級為真實 API 還差什麼？

此刻我們不把 FakeAPI.query() 直接替換成 client.query_variant() 就宣稱完成。原因包括：

1. SDK 呼叫模式不同： 需核對正式 alphagenome==0.10.0 安裝版函式簽名、同步／非同步行為及回傳資料結構。不能假設官方函式本身就是 async def。
2. 錯誤契約需實測： 本文的 429／503 是一般 HTTP 教學情境；真實 SDK 如何呈現認證錯誤、資源耗盡、超時或 503，尚無本輪 LIVE 證據。
3. 真正的速率限制： Semaphore 只有並行上限；正式批次客戶端還須考慮每秒／每分鐘額度、伺服器 Retry-After、抖動（jitter）、最大總等待時間與重複查詢的成本。
4. 紀錄來源： 必須保存模型版本、assembly、座標制度、ontology、scorer、SDK 版本、輸出型態、來源雜湊與錯誤類型；不能只保存畫面上的成功數字。
5. 隱私與用途： 真實患者 VCF、基因資料與第三方服務傳輸須先有對應核准；此系列研究用工具不能自動宣稱臨床可用。

這也是為什麼 Reboot SDK 0.9.0 的歷史成功，與新的 Bridge 0.10.0 Mock PASS 要分開記錄。兩者不能拼接成一次「新版 API 已經完成批次負載測試」。

✅ 本日 Checklist

• [ ] 我能說明 asyncio 與 Semaphore 的作用不同於真正的每秒配額。
• [ ] 我能指出何時適合重試（429／503／TIMEOUT）及不應盲目重試（400／401）。
• [ ] 我能解釋 RETRY_EXHAUSTED 為何應寫進報告，而不能直接消失。
• [ ] 我能產生 day19_mock_batch.json，找到五筆人工歷程，確認 real_api_calls_this_run=0。
• [ ] 我知道本次結果來自 SYNTHETIC_FIXTURE，不是新一輪 LIVE 查詢。

五、科學證據與限制

證據分層： Day 17／18 使用的是 Reboot 歷史 LIVE 執行後的衍生摘要；本篇 429／503／TIMEOUT、三個合成分數與任務狀態則全部是新寫的 SYNTHETIC_FIXTURE。目前沒有任何證據支持「曾完成官方 API 的批次性能測試」，也沒有證據支持由此計算出模型正確率或器官致病易感指數。

品質驗證： 本次只於 Python 3.13.5 執行離線範例與 12 項單元測試並通過；未在本次環境執行 Python 3.12.3。程式語法使用 Python 3.12 支援的標準函式庫，仍應由作者在 WSL Python 3.12 重新確認。

任何正式 Atlas／Model API 的 LIVE 操作都需要另行明確授權；不使用真實患者資料；GitHub 維持 PRIVATE／PUBLICATION_ON_HOLD。

六、明日預告

Day 20 將從「如何安全管理一批查詢」走到「怎麼正確定位要查哪段基因座」：使用人工 Motif 區間練習 1-based／0-based 座標換算，理解不小心偏移一個位置為何會改變註解結果。