功能原理與機制
OpenAI Batch API 提供一種 非同步批次 處理請求的方式，用戶可以一次提交大量任務，讓模型在 24 小時內處理完畢並返回結果。核心運作流程如下：
- 準備批次任務：開發者先將多筆請求整理為
.jsonl（JSON Lines）格式的輸入檔，每一行是一個獨立的 JSON 任務物件，每個任務至少包含以下欄位：custom_id：自定義的任務識別碼，用來對應輸出結果中的哪一筆請求。由於批次執行後結果未必按原順序排列，透過custom_id可以正確對號入座每筆回應。method：HTTP 方法，通常為"POST"，代表這筆請求要呼叫的 API 方法（目前僅支援 POST 類型的 OpenAI API 端點）。url：API 路徑，例如"/v1/chat/completions"或"/v1/embeddings"，指定每個任務要呼叫的模型端點。注意：所有任務必須針對相同的模型與端點，因此url在一個批次中通常相同。body：HTTP 請求內容，包含模型參數和輸入資料。例如，聊天完成任務的 body 可能包括model（模型名稱）、messages（對話訊息陣列）等；Embedding 任務則包含input文本等欄位，body的結構與對應同步 API 的請求格式相同。
- 資料上傳與批次建立：準備好
.jsonl檔後，需透過 OpenAI 檔案上傳介面將其上傳，指定用途為"batch"。上傳成功會回傳一個file_id。接著呼叫批次建立端點，提供剛才的file_id和目標endpoint（例如/v1/chat/completions）來啟動批次作業。可選參數completion_window用來設定允許的最長執行時間，一般為預設的"24h"。成功建立批次後會獲得一個batch_id供後續查詢。 - 後端執行機制：Batch API 在後端會將提交的任務排入佇列進行處理。與即時同步請求不同，批次作業利用獨立的資源池和排程系統，將任務分配至閒置的運算資源進行處理。這意味著 OpenAI 可能在離峰時段或透過分散式計算同時處理多筆請求，以提高效率。換言之，Batch 作業被放入排程/佇列後，後端系統會按可用算力逐步取出任務執行，可能對任務做分片、並行處理，以確保在服務水平協議（SLA）內完成所有任務。
- 結果產生與資料流向：所有任務執行完畢或達到時間窗口後，Batch 作業會標記為完成，結果將以檔案形式提供。每筆輸出結果也是獨立的 JSON 行，包含當初的
custom_id以及response欄位。response中通常會有status_code（例如 200）以及body（對應該任務的 API回傳內容）等資訊。開發者可以透過批次查詢接口取得 output 檔案的 ID，再利用檔案下載介面抓取所有結果資料。整體資料流向是：用戶端→上傳請求檔→OpenAI服務列隊處理→生成結果檔→用戶端下載結果，資料在雲端存儲及處理過程中遵循 OpenAI 的安全與保留政策。 - 同步 vs. 批次架構差異：同步 API 屬於即送即答，每次呼叫直接由前端服務器分配模型計算資源處理，受制於嚴格的即時 Rate Limit；而 Batch API 採用非即時的後處理架構，有專門的批次作業處理管線。例如，OpenAI 會將批次請求放入高吞吐的任務佇列，由背景工作進程依序取出執行，不與同步請求爭用資源。這可能涉及內部的排程器、任務暫存，以及大量 GPU 分布式計算節點共同完成批次中成千上萬請求。由於有最長 24 小時的寬限時間，系統能夠彈性調度這些任務，在不影響即時服務的前提下充分利用整體算力資源。因此，相較同步 API 架構，Batch API 背後多了一層工作排程與結果彙整機制，確保即使大量請求也能在承諾時間內完成並產出統一的結果檔。
成本效益與效能評估
Batch API 在成本和效能上相對同步 API 有顯著差異：
- Token 成本優惠：使用 Batch API 可享有約 5 折的價格優惠。根據 OpenAI 說明，批次端點的計費是同步 API 的 50%。也就是說，無論是輸入或輸出 token，都只需付一般價格的一半。例如，假設某模型同步調用每 1,000 個 token 收費 $1，美金，那麼經由 Batch API 執行相同任務僅約 $0.5，美金。實際定價因模型而異，但折扣幅度一致。例如 GPT-4 Turbo 同步價格約每 1K tokens $0.003，而批次僅 $0.0015；又如
text-embedding-ada-002Embedding 模型同步價每 1K token $0.0001，美金，Batch 僅 $0.00005，美金。因此，大量任務透過 Batch 執行可節省近一半經費。許多使用者回報透過批次處理將每月 OpenAI API 花費從數百美元降至不到一半。 - Rate Limit 與吞吐量：Batch API 擁有獨立且遠高於同步 API的流量限制。OpenAI 為批次請求配置了專門的高頻管道，例如允許 GPT-4 Turbo 模型佇列高達 2.5 億個輸入 token的一次性作業量。這遠超同步 API 每分鐘僅數萬 token 的限制。更重要的是，Batch API 的使用不計入您平常 API 金鑰的 Rate Limit 配額，換言之，你可以同時進行批次作業和即時請求，兩者各自獨立計算流量。每個批次作業本身也有大小上限：官方建議單一批次最多包含 50,000 筆請求或檔案大小不超過 100MB（新的說明中甚至提到針對 Embeddings 任務可一次enqueue多達100萬筆請求）。如此高的容量允許一次處理龐大的資料集。相較之下，使用同步 API 若嘗試提交數萬請求，極易碰到 Rate Limit 瓶頸而被拒絕。因此，在吞吐量方面，Batch API 提供了大幅提升的並行處理能力。
- 延遲與完成時間：採用 Batch API 意味著單筆請求的即時回應延遲較高，因為結果須等整個批次任務處理完才能取得（最遲 24 小時內），但是從整體任務完成效率來看，Batch API 反而可能更佳。舉例來說，如果有 100,000 筆文本需要轉換為向量嵌入：
- 使用同步 API：假設每筆請求平均包含 100 個 token，總計需要處理 1,000 萬個 token。即使模型本身處理單筆僅需數百毫秒，Rate Limit 會是主要瓶頸。就算您的額度允許每分鐘 250,000 個 token，處理 1,000 萬 token 仍約需 40 分鐘，考慮網絡開銷和排隊，實際可能數小時才能全部完成。而且同步價格不打折，若每萬 token 收費 $0.001，美金，處理 1,000 萬 token 成本約 $10，美金。
- 使用 Batch API：你可以一次提交這 100,000 筆請求，OpenAI 後端會自動並行處理許多請求。許多用戶反饋批次任務通常在幾十分鐘到數小時內即提前完成，遠低於 24 小時上限。假定同樣在數小時內拿到結果，總成本只有同步的一半，約 $5，美金（依照 50% 折扣計算）。更重要的是，您無需自行控制請求節奏，OpenAI 會在其高並發架構下最大化吞吐量完成任務。在此例中，Batch API 提供更高的總體吞吐，而換取的是容忍數十分鐘以上的等待時間。
- 從平均延遲角度看，每筆請求透過 Batch 處理可能要等較久才拿到，但整批完成所需的時間對大規模任務往往更短或更可預期。因此，即時性 vs. 效率是取捨：即時互動場景仍需同步 API，而大量非即時任務改用 Batch 更划算。
- 成本效益綜合分析：總結而言，Batch API 用較長的允許時間換取成本與效能優勢。對於不需立即結果的任務，Batch 可大幅降低每 token 價格並且一次提交大量請求而不擔心 Rate Limit。許多企業發現將日常的批量分析工作轉為 Batch 後，每月 API 開支降低超過 50% 且不再需要維護複雜的併發控制機制。不過，需要即時回覆的情境下，Batch 的等待時間和非即時性就成為劣勢。因此應根據應用需求權衡：批次適合大量且可延遲的任務，同步適合少量需即時回應的請求。
使用情境與實務案例
Batch API 特別適用於大量資料處理且對即時性要求不高的情境。以下列舉三種最適合使用 Batch API 的應用場景，並說明為何同步呼叫不適用：
- 大規模內容分類與標注：例如電子商務平台要為數十萬商品描述自動加上標籤，或社群網站要對用戶貼文進行情緒分類。這類任務資料量巨大，但可以離線處理。使用 Batch API 可以一次提交所有文本讓模型批量分類，常見用途包括：
- 標籤（tagging）或分類內容，如為市集商品或部落格文章產生標籤與摘要。
- 批量情感分析，針對大量用戶回饋評估情緒傾向。
- 批量內容過濾或支援票據分類等。
- 如果用同步 API 逐筆處理，除了費用高昂（無折扣）外，Rate Limit 也很快成為瓶頸，需複雜的程式來分批等待。而 Batch API 天生適合這種資料密集型工作，不僅成本減半，還省去手動控管呼叫頻率的麻煩。唯一代價是結果需要稍晚（例如幾小時後）彙總，對於每日或定期的分類任務而言這通常可以接受。
- 模型評估與大量實驗：在研發或ML評估中，常需要對模型進行大規模評測。例如評估一個新 GPT 模型在上萬筆測試問答上的表現、或批量產生不同提示詞（prompt）的回應以比較效果。這些屬於批次推論（batch inference）的範疇。使用 Batch API，開發者可以一次提交整套評測任務讓模型離線跑完，24小時內保證拿到所有結果。相比之下，若用同步 API 實現：
- 需要撰寫程式反覆呼叫模型並監控速率，可能耗費大量時間管理請求併發。
- 模型產生成本更高（沒有折扣），長時間大量呼叫也增加失敗風險（例如網絡不穩定導致中斷）。
- 有了 Batch API，只需準備好測試集請求列表並提交，等待結果即可，簡化了模型評估的工作流。實務上，開源工具如 EleutherAI 的 eval harness、學術研究等都可將繁重的評估工作轉為批次執行，以節省時間與金錢。
- 知識庫向量嵌入（Embeddings）建置：許多應用需要對大規模文本資料（文件、維基百科文章、產品描述等）計算向量嵌入，以建構向量搜尋或推薦系統的索引。例如一家公司要將其知識庫上十萬篇文件轉換為 Embedding 存入向量資料庫。這種資料量大的嵌入計算非常適合用 Batch API：
- 您可以把所有文件的內容分批打包成 JSONL 任務（每行一段文字），用 Batch API 調用 OpenAI 的 Embedding 模型一次性算出所有向量。
- 相較之下，用同步 API 逐篇轉換不但花費時間長、成本也高（折扣差異明顯），而且需要自己控制請求速率以避免超額。
- 使用 Batch，好處是高吞吐且成本低，OpenAI 後端可能同時併發計算多個向量請求，以快速產生結果。例如有國外部落格分享了用 Batch API 對 15 萬關鍵字進行分群（Cluster）的案例：作者將關鍵字列表拆成兩個批次提交 GPT-4 Turbo 模型進行語意分群，成功在一天內拿到結果，並節省了約一半費用。可見，對於知識庫建置這種需要處理數十萬文本的任務，Batch API 在效能和成本上都有壓倒性優勢。
除了上述情境，Batch API 還被廣泛應用於內容產生與轉換（如批量摘要文件、大量翻譯）以及資料擴充（如為資料集生成額外資訊欄位）等場景。關鍵判斷標準是：如果任務可以離線執行且不要求立即得到結果，那通常就是 Batch API 大顯身手的地方。反之，像即問即答的聊天機器人、線上服務即時回覆用戶詢問等，就仍須使用同步 API。
實務案例：目前國內外已有許多企業和開源項目將 Batch API 應用於生產工作流：
- SEO 關鍵字分群： 有行銷技術部落格分享使用 Batch API 分析關鍵字，以 50% 成本完成大量關鍵字的語意 clustering。同步方式下這種規模本難以在短時間內完成，而 Batch 簡化了流程，適合數據科學批次分析。
- 資料標註與內容審查： 一些平台（如社群網站）需定期審核用戶產生內容。某些開源方案將每日新產生的內容彙集後，用 Batch API 做批量內容分類和違規偵測，確保在隔天上線前完成審查，同時減少人工成本。OpenAI 官方亦建議將內容標記/摘要類的工作轉為批次，如自動為商品產生說明、為客服票據分類並給出建議答覆等。
- 學術與企業的模型評測： 美國賓州大學的研究人員示範了如何利用 Batch API 批量詢問 GPT 模型，以從名人姓名列表中提取出身地等資訊。透過批次，他們在保證結果完整性的同時節省了一半成本並避免手動逐條查詢。類似地，許多開源評測框架開始支援調用 OpenAI Batch API 來加速大規模實驗。
與 LLMOps/MLOps 流程整合：Batch API 非同步、延遲執行的特性，使其非常適合整合到資料管道和機器學習工作流中。例如：
- 在 Apache Airflow 中，可以建立一個 DAG任務：先上傳 JSONL 檔並啟動 Batch 作業，接著使用 Airflow 的 sensor 等待批次完成，最後下載結果做後續處理。一位開發者在 Airflow 提交功能需求，希望 OpenAI 的提供者 Hook能支援 Batch API，以便在工作流中直接觸發批次預測並等待結果。目前 Airflow 已新增相應支持（提供 Deferrable Operator 來非阻塞等待 Batch 完成）。
- 在 Prefect、ZenML 等工作流中，編寫類似的流程也很直接。您可以使用 Prefect 的任務序列：一個任務上傳檔案並呼叫批次，然後設定一個輪詢任務每隔幾分鐘檢查狀態，最後一個任務處理結果檔案。ZenML 官方亦提供了範例整合 Batch API 進行 LLM 批量推論的指南，使得這種異步推理可以嵌入持續的機器學習管線中。
- 更專業的如 Flyte/Union.ai 等 MLOps 平台，甚至針對 OpenAI Batch API 開發了專門的 Agent。Union.ai 團隊推出的 Flyte OpenAI Batch agent 可以自動執行從資料抓取、JSONL 打包上傳、啟動批次到定時查詢結果等步驟，並在結果就緒時發出通知。透過這種整合，他們展示了如何在不增加容器執行開銷的情況下，以最少人工干預完成整個批次推理流程。這種自動化範本非常適合定期需要跑批次任務的情境，例如每晚處理當日累積的數據並寫入資料庫。
總之，Batch API 能與現有的排程與管線工具良好結合。最佳實務是將批次任務的觸發、監控與結果處理模組化嵌入整個 MLOps 流程中，透過如 Airflow task、Prefect flow 或專用 agent 來管理。在這樣的架構下，開發團隊可以輕鬆地自動化週期性的 GPT 任務（如每日摘要、每週數據分析），充分享受 Batch API 帶來的高效率與低成本優勢。
限制與風險控管
雖然 Batch API 功能強大，但在使用時需注意其限制並採取相應的風險控管措施：
- 常見錯誤原因與錯誤碼：
- 輸入格式錯誤：批次 JSONL 檔案格式要求嚴格，每行必須是獨立完備的 JSON。若存在格式問題（例如缺少大括號、引號未轉義等），建立批次時會出現類似 “This line is not parseable as valid JSON” 的錯誤。此時 Batch 作業不會啟動。解決策略是使用 JSON 工具驗證每行格式，或嘗試逐行縮小範圍定位問題行。
- URL 或參數錯誤：如果 JSONL 中的
url欄位填寫不正確（例如包含完整主機名稱、或路徑不符），也會導致錯誤。例如錯誤訊息 “The URL provided for this request does not prefix-match the batch endpoint” 表示 URL 格式不符預期。通常需要改為像"/v1/chat/completions"這樣的相對路徑。同樣地，如果body中缺少必要參數（如未提供model名稱）會引發驗證錯誤，應對照 API 文件修正。 - Custom ID 問題：每行的
custom_id必須唯一。重複的 ID 可能導致結果對應混亂，甚至造成批次執行異常。一些使用者曾懷疑因為 custom_id 重複而導致只有部分請求被執行的情況。最佳做法是在準備任務時，確保 custom_id 有清晰的規則（如使用遞增序號或UUID），避免衝突。 - 批次大小限制：如果提交的批次超過官方限制（例如單檔超過 100MB或超過 5萬行），OpenAI 會在建立批次時返回錯誤，指出檔案過大或請求數過多。這種錯誤通常代碼為
400 Bad Request，訊息提示超出限制。解決方法是將任務切分成較小的多個批次提交。 - 認證與權限錯誤：使用 Batch API 需要有效的 API 金鑰且該帳號有存取相應模型的權限。如果金鑰無效或沒有批次調用權限，將立即返回
401 Unauthorized或相關錯誤。另外，若body中指定了未支援批次的模型（或名稱拼寫錯誤），也可能回傳錯誤訊息提示模型無效。這類錯誤需確認帳戶權限、金鑰和模型名稱是否正確無誤。 - 執行超時/批次過期：Batch API 設計保證最長執行 24 小時。如果後端在 SLA 時限內未能完成所有請求，批次狀態將標記為 expired（過期），錯誤碼為
"batch_expired"。OpenAI 會取消尚未完成的請求，並將已完成部分的結果返回。此時開發者只會付費已完成部分的 token 消耗。Expired 狀況可能發生於任務量極大或當時系統負載過高導致無法及時處理完畢。為降低此風險，務必遵循官方建議的批次大小，或在臨近 24 小時仍未完成時主動取消重提（不過一般較少發生，因為系統會調整資源在期限內完成）。
- 錯誤排查策略：遇到批次執行問題時，可採取以下排查步驟：
- 檢查批次狀態與計數：透過批次檢索 API（
GET /batches/{id}）查看status以及request_counts等資訊。request_counts會列出 completed、failed、total 等數量。如果failed非零，可進一步下載 error 檔案（若有提供error_file_id）查看每行任務的錯誤訊息。若狀態為expired或failed，表示有全域性性問題，需要根據訊息調整後重跑。 - 驗證輸入檔格式：隨機抽取幾行 JSONL 嘗試用 JSON parser 讀取，確定沒有語法錯誤或控制字元問題。特別注意換行、引號等符號正確轉譯。必要時可縮小批次範圍，二分法測試找出問題行。
- 確認參數與環境：檢查 API Key 是否正確、模型名稱是否拼寫無誤且在支援列表內（例如不要用舊版模型名稱），
url是否以/v1/...開頭。也確認 openai 版本是否為最新。 - 瞭解部分完成的情況：如果發現批次完成但
completed計數遠小於 total（例如只有 5%請求成功，而 error file 無明確錯誤，可能是系統當時問題（例如 Azure 平台曾有類似回報）。
- 檢查批次狀態與計數：透過批次檢索 API（
- 隱私、資安與合規風險：批次處理涉及將大量資料上傳至雲端並長時間儲存，需注意：
- 資料留存與刪除：根據 OpenAI 的政策，API 請求與結果最長可能保留 30 天 作為防濫用用途。而透過 Files 介面上傳的批次檔案及其輸出檔，理論上會一直保存在帳號中直到使用者手動刪除。這意味著敏感資料會在雲端停留一段時間。若您的應用有嚴格的隱私要求（例如包含個資的資料集），建議在批次完成並拿到結果後，立即使用 API 刪除相關的上傳檔案與結果檔，減少資料外洩風險。對於高度機密的資料，也許需要評估是否適合上傳至第三方 AI服務處理。
- 資料加密與訪問控制：OpenAI 不支援上傳已加密的內容直接進行處理（模型需要明文輸入）。因此敏感資料在傳輸與儲存過程中應確保傳輸層安全（OpenAI API 本身走 HTTPS 傳輸）。儲存在 OpenAI 伺服器期間，雖然 OpenAI 承諾安全存放，但您可採取額外措施：例如對某些非關鍵敏感欄位進行去識別化處理後再提交，或者與 OpenAI 簽訂企業級資料保護協議。如果您的帳戶有 Zero Data Retention 選項（零資料留存，需要 OpenAI批准），務必啟用以確保批次內容不被保留。
- 存取權限管理：妥善保管 API 金鑰，避免在 Code Repository直接暴露。由於批次任務通常是自動化執行，可能涉及 CI/CD管線或雲端工作流，建議使用環境變數或秘密管理服務來傳遞 API 金鑰，並確保執行環境權限鎖定在必要範圍。針對批次上傳的檔案，本身綁定於您的 OpenAI 帳戶，外部無法直接存取，但還是要防範日誌中不記錄檔案 ID等資訊，避免被不當利用。
- 合規考量：若批次處理涉及個人資料（例如大量用戶訊息分析），需注意遵守相關法規如 GDPR。確保已在隱私政策中告知用戶資料將交由第三方（OpenAI）處理，並取得必要授權。可考慮對批次結果進行稽核，如定期抽查模型輸出內容，確認沒有洩漏不該提供的資訊或產生偏見。在資安方面，監控批次使用情況，防止 API 金鑰濫用或異常高額費用產生。
- 批次未完成的自動補救：如果遇到批次過期或部分任務未成功的情況，應實作機制自動補救：
- 結果校驗與重試：批次完成後，程式應檢查
completed是否等於total。若有缺失，可根據custom_id找出哪些請求沒有對應結果。將這些失敗/遺漏的任務重新打包成新的 JSONL，再次提交一個補充批次或改用同步 API 重跑。為降低再次失敗機率，建議檢視原先失敗原因並適度調整（例如將過大的任務切小）。這種重試機制可整合在您的工作流中，確保批次作業最終100%完成。 - 自動通知與警報：由於批次任務執行時間長，不可能人工盯住等待。最好配置自動通知系統：例如當批次狀態轉為 completed/failed/expired 時，由程式發送 Slack 訊息或 Email 給相關工程師。許多工作流工具提供簡單的 webhook 或郵件通知整合，您也可以直接在批次監控腳本中加入通知邏輯。一旦收到「批次過期未完成」的警報，團隊就能即時介入查看並啟動補救流程。
- 取消與資源清理：如果某批次任務發現輸入有誤或需求變更，可透過 API 主動取消 (
cancel) 批次，以避免浪費資源和費用。取消後已完成部分依然會產生結果檔，其餘將標記取消。在自動化環境中，應在適當情況下調用取消（例如偵測到明顯錯誤後立即取消批次，然後發通知）。此外，在批次完成/取消後，記得清理暫存的檔案資源（如刪除已無用的上傳檔案），保持帳戶檔案列表整潔並減少資安風險。
- 結果校驗與重試：批次完成後，程式應檢查
綜上，良好的風險控管需要嚴謹的檔案製備與驗證來防止執行前的錯誤，並在執行期間及事後實施監控、通知和補救策略。同時在隱私與合規上慎重評估，確保批次處理流程既高效又符合安全標準。
實作與最佳實務
為了更具體地說明 Batch API 的使用方法，以下提供一個 Python 實作範例，demo 如何自動上傳 JSONL、建立批次、監控狀態並下載結果。此外，我們也討論處理超大檔案或大量請求時的策略與開發建議。
範例：使用 Python 呼叫 Batch API
假設我們有數篇文章，需要請 gpt-4o 模型產生摘要。我們將這些任務打包成 JSONL 檔並透過 Batch API 執行。步驟如下：
import json
import time
import sys
from openai import OpenAI
from openai.error import OpenAIError
# 1. 初始化 client
client = OpenAI(api_key="你的 OpenAI API 金鑰")
# 2. 準備批次任務
articles = [ ... ] # 請自行填入文章列表
batch_requests = []
for idx, content in enumerate(articles):
batch_requests.append({
"custom_id": f"task_{idx}", # custom_id 必須為字串且唯一
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o",
"messages": [
{"role": "system", "content": "你是一個優秀的文章摘要助手。"},
{"role": "user", "content": f"請摘要以下文章內容：{content}"}
],
"temperature": 0.7,
"max_tokens": 4096
}
})
# 3. 將任務寫入 JSONL
input_filename = "batch_input.jsonl"
try:
with open(input_filename, "w", encoding="utf-8") as f:
for req in batch_requests:
f.write(json.dumps(req, ensure_ascii=False) + "\n")
except OSError as e:
print(f"寫入 {input_filename} 時發生錯誤：{e}", file=sys.stderr)
sys.exit(1)
# 4. 上傳 JSONL 檔案（purpose="batch"）
try:
with open(input_filename, "rb") as f:
upload_resp = client.files.create(file=f, purpose="batch")
file_id = upload_resp.id
print("已上傳檔案 ID：", file_id)
except OpenAIError as e:
print("檔案上傳失敗：", e, file=sys.stderr)
sys.exit(1)
# 5. 建立 Batch Job（completion_window 必需為 "24h"）
try:
batch_job = client.batches.create(
input_file_id=file_id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
batch_id = batch_job.id
print("Batch 工作 ID：", batch_id)
except OpenAIError as e:
print("建立 Batch 失敗：", e, file=sys.stderr)
sys.exit(1)
# 6. 輪詢監控狀態
while True:
try:
batch_job = client.batches.retrieve(batch_id=batch_id)
except OpenAIError as e:
print("查詢 Batch 狀態失敗：", e, file=sys.stderr)
break
status = batch_job.status
print("目前狀態：", status)
if status in ("completed", "failed", "expired"):
print(f"Batch 結束，狀態：{status}")
break
time.sleep(10) # 等待 10 秒再查一次
# 7. 若完成則下載並解析結果
if batch_job.status == "completed":
result_file_id = batch_job.output_file_id
try:
# 新版 SDK 下載檔案內容
content_bytes = client.files.retrieve_content(file_id=result_file_id)
decoded = content_bytes.decode("utf-8")
results = [json.loads(line) for line in decoded.splitlines()]
except OpenAIError as e:
print("下載結果失敗：", e, file=sys.stderr)
sys.exit(1)
# 列印前 3 筆摘要
for res in results[:3]:
cid = res["custom_id"]
summary = res["response"]["body"]["choices"][0]["message"]["content"]
print(f"{cid} -> 摘要：{summary}")
以下針對修正版程式碼逐行拆解：
import json
import time
import sys
from openai import OpenAI
from openai.error import OpenAIErrorimport json：載入內建的 JSON 模組，用來做 JSON 編碼／解碼。import time：載入時間模組，後續用sleep()實作輪詢（polling）。import sys：引入系統模組，可用來印錯誤到stderr並sys.exit()。from openai import OpenAI：從新版 SDK 建立OpenAIclient。from openai.error import OpenAIError：引入通用例外類別，用來捕捉 API 呼叫失敗時的錯誤。
# 1. 初始化 client
client = OpenAI(api_key="你的 OpenAI API 金鑰")這行建立一個 client 實例，並把你的 API 金鑰傳給它。之後所有跟 OpenAI 互動的呼叫，都透過這個 client 物件來處理。
# 2. 準備批次任務
articles = [ ... ] # 請自行填入文章列表
batch_requests = []
for idx, content in enumerate(articles):
batch_requests.append({
"custom_id": f"task_{idx}", # custom_id 必須為字串且唯一
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o",
"messages": [
{"role": "system", "content": "你是一個優秀的文章摘要助手。"},
{"role": "user", "content": f"請摘要以下文章內容：{content}"}
],
"temperature": 0.7,
"max_tokens": 4096
}
})articles = […]：把要處理的文章清單放這裡，元素個數決定 batch size。batch_requests = []：預先準備一個空陣列，用來存所有 batch 任務的設定。for idx, content in enumerate(articles):：loop 每篇文章，idx當作任務編號、content是文章內容。.append({...})：把格式符合 Batch API 要求的 dict 加進batch_requests。custom_id：給每筆任務一組獨一無二的 ID，好追蹤回傳結果。method、url：固定 POST、指向 Chat Completions 端點。body：塞進 model、對話訊息、temperature、max_tokens… 等參數。
# 3. 將任務寫入 JSONL
input_filename = "batch_input.jsonl"
try:
with open(input_filename, "w", encoding="utf-8") as f:
for req in batch_requests:
f.write(json.dumps(req, ensure_ascii=False) + "\n")
except OSError as e:
print(f"寫入 {input_filename} 時發生錯誤：{e}", file=sys.stderr)
sys.exit(1)input_filename = "batch_input.jsonl"：定義要輸出的檔名。with open(..., "w", encoding="utf-8") as f:：開一個檔案物件f，確保離開區塊時自動關閉。for req in batch_requests:：把每筆任務用json.dumps()轉成 JSON 字串並寫一行。except OSError：捕捉檔案 I/O 錯誤，印到標準錯誤流，並以sys.exit(1)結束程式。
# 4. 上傳 JSONL 檔案（purpose="batch"）
try:
with open(input_filename, "rb") as f:
upload_resp = client.files.create(file=f, purpose="batch")
file_id = upload_resp.id
print("已上傳檔案 ID：", file_id)
except OpenAIError as e:
print("檔案上傳失敗：", e, file=sys.stderr)
sys.exit(1)with open(..., "rb") as f:：以二進位讀取模式打開剛才產出的 JSONL。client.files.create(...)：呼叫 OpenAI 上傳檔案，purpose="batch"標明用途。file_id = upload_resp.id：取得回傳的檔案 ID，後續建立 batch job 用它。except OpenAIError：捕捉任何 API 呼叫錯誤，並輸出錯誤後結束。
# 5. 建立 Batch Job（completion_window 必需為 "24h"）
try:
batch_job = client.batches.create(
input_file_id=file_id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
batch_id = batch_job.id
print("Batch 工作 ID：", batch_id)
except OpenAIError as e:
print("建立 Batch 失敗：", e, file=sys.stderr)
sys.exit(1)client.batches.create(...)：用剛才取得的file_id建立 Batch Job，並指定要呼叫的 endpoint、完成窗口時間（24 小時）。batch_id = batch_job.id：存下批次工作的 ID，之後輪詢用。
# 6. 輪詢監控狀態
while True:
try:
batch_job = client.batches.retrieve(batch_id=batch_id)
except OpenAIError as e:
print("查詢 Batch 狀態失敗：", e, file=sys.stderr)
break
status = batch_job.status
print("目前狀態：", status)
if status in ("completed", "failed", "expired"):
print(f"Batch 結束，狀態：{status}")
break
time.sleep(10) # 等待 10 秒再查一次while True:：開始無限迴圈，不斷檢查狀態。client.batches.retrieve(...)：用batch_id向 API 詢問最新狀態。status = batch_job.status：目前狀態可能是"pending"、"running"、"completed"、"failed"、"expired"… 等。if status in (…)：一旦作業結束或失敗，就跳出迴圈。time.sleep(10)：每 10 秒才再問一次，避免過度呼叫 API。
# 7. 若完成則下載並解析結果
if batch_job.status == "completed":
result_file_id = batch_job.output_file_id
try:
content_bytes = client.files.retrieve_content(file_id=result_file_id)
decoded = content_bytes.decode("utf-8")
results = [json.loads(line) for line in decoded.splitlines()]
except OpenAIError as e:
print("下載結果失敗：", e, file=sys.stderr)
sys.exit(1)
# 列印前 3 筆摘要
for res in results[:3]:
cid = res["custom_id"]
summary = res["response"]["body"]["choices"][0]["message"]["content"]
print(f"{cid} -> 摘要：{summary}")if batch_job.status == "completed":：確認作業真的成功完成才處理結果。result_file_id = batch_job.output_file_id：取得後端產出的結果檔案 ID。retrieve_content(...)：新版 SDK 下載整個檔案內容，回傳bytes。.decode("utf-8")／.splitlines()：把 bytes 換成字串，再拆成一行行的 JSON。[json.loads(line) …]：把每行 JSON parse 成 dict，存到results。for res in results[:3]:：示範印出第一個到第三個任務的摘要結果。
透過上面逐行拆解，你就能清楚知道各區塊的用意、資源釋放、錯誤處理，以及新版 SDK 的介面差異。如有細節想再釐清，歡迎繼續討論！
自動化測試流程建議
開發 Batch API 功能時，應設計完善的自動化測試來確保流程穩定可靠，包括：
- 單元測試：對於組裝 JSONL、解析結果等函式邏輯，可以撰寫單元測試。例如：
- 測試 JSONL 生成函式：給定幾筆假資料，檢查輸出的檔案每行都能
json.loads成功，且 custom_id 是否符合預期格式。 - 測試結果解析函式：模擬幾筆批次結果的 JSON 行，確認解析後能正確提取所需欄位。
- 使用 Mock 模擬 OpenAI API 回應：由於真正呼叫 Batch API 可能費時且耗費額度，在單元測試中可使用模擬物件。比如用
monkeypatch替換openai.File.create等方法，使其返回預設的 file_id；模擬openai.Batch.create立即返回一個假的 batch_id；以及模擬openai.Batch.retrieve連續呼叫若干次後狀態變為 completed 並提供假 output_file_id。如此可在不實際連網的情況下，驗證我們的輪詢邏輯和錯誤處理分支是否正常運作。 - Edge cases 測試：例如測試當任務列表為空時是否正確處理、當生成的 JSONL 超過尺寸限制時函式是否有對應反應（比如拒絕或切分）。
- 測試 JSONL 生成函式：給定幾筆假資料，檢查輸出的檔案每行都能
- 整合測試：這類測試會實際調用 OpenAI API，但通常僅用少量任務來驗證端到端流程。例如：
- 提交一個包含 1-2 個簡單請求的小批次到 OpenAI 測試（可選擇廉價的模型
gpt-4o-mini或text-embedding-ada-002以降低花費），輪詢直到完成，並驗證返回結果內容是否合理。這可以確保我們的整個調用鏈路在 OpenAI 平台上能跑通。 - 測試錯誤處理：可以構造一個明知格式有誤的 JSONL，上傳並嘗試建立批次，確認我們的程式捕捉到了預期的錯誤（例如接收到
"parseable as valid JSON"的錯誤訊息），並且沒有無限等待。 - 環境隔離：整合測試通常不在每次 CI運行，因為涉及實際 API耗時耗錢。可將其配置為手動或定期在 Staging環境執行。此外，要注意不要在公用日誌中暴露敏感資訊（如 API Key或批次輸入內容）。
- 提交一個包含 1-2 個簡單請求的小批次到 OpenAI 測試（可選擇廉價的模型
- 測試數據與金鑰管理：將 OpenAI API 金鑰以環境變數注入測試環境，切勿寫死在程式中。對於外部依賴（網絡請求），使用 Test Double在單元測試中取代，在整合測試中則連通實際服務。確保在測試完成後，批次上傳的檔案和結果檔也刪除清理，以免佔用額度或洩漏資訊。
大檔案與大量請求的最佳實務
當面對超過 Batch API 單次限制的大型任務時，我們需要分批、分片策略來順利處理：
- 檔案分割策略：如前所述，OpenAI 建議單個批次檔案不超過 100MB 或 5 萬請求。如果您有更大的資料集（例如 20 萬行輸入），應分割成多個
.jsonl檔案。例如，可將 20 萬請求切成每檔 5 萬行，得到 4 個檔案分批上傳。可以在程式中以迴圈自動化這個過程。Medium 範例代碼。關鍵是確保每個檔案內部 custom_id 唯一，且跨檔案也不要重複，可以在 custom_id 加上檔案序號前綴避免衝突。 - 批次併發與節奏：OpenAI Batch API 雖允許一次 enqueue 大量任務，但對於超大規模（如上百萬請求），最好逐批請求而非一次全部上傳。根據官方，Embeddings 任務同時在處理的請求上限為 100 萬；對其他任務雖未硬性限制數量，但每個帳戶有其批次 token 配額。因此，如果切成多個批次檔案，建議控制同時啟動的批次數量。例如您有 10個檔案，或許每次提交 2-3個批次，同步監控，等前面的完成再繼續提交後面的，以免過度併發導致隊列擁塞或自身難以管理。這種流水線式處理能在保持高效的同時，減少單批過期風險。
- 任務拆分：有時單筆任務本身就很龐大（例如一個超長文本超過模型上下文長度）。Batch API 並不會自動拆分單一任務內容，仍需在準備資料時切分。舉例來說，如果要摘要一本小說，需要先將文本切章節，每章作為獨立任務，否則直接提交整本書的內容給一條請求會失敗或被截斷。在 Embedding 情境下，若一篇文章特別長，應在客戶端預先將文章切段（確保每段不超模型 token上限）後作為多個 embedding任務。總之，確保每行任務大小在模型可處理範圍內，這樣批次才能順利完成所有項目。
- 程式設計建議：處理大批量時，注意程式的記憶體與效能：
- 流式讀寫：不要一次性將百萬條記錄全部載入記憶體再序列化。應採取流式寫檔方式（如上例所示，一條條 write），這樣即使資料集極大也能處理，而不會撐爆記憶體。
- 檔案尺寸監控：可以在寫入時計算已寫入的 byte 大小，接近 100MB 就啟動下一檔案。例如 Python 可以用
f.tell()檢查檔案指標位置確定大小。如果每行字數相對固定，也可按行數估計（例如每 5 萬行切一檔）。 - 結果合併：當多批次分開處理完成後，需要彙總結果。可以在 custom_id 上體現分批資訊，或在結果拿回後按需要合併。例如最簡單就是把所有結果檔的內容合併在一起，再依 custom_id 排序或插入資料庫。若順序很重要，可以在 custom_id 中編碼原始順序索引，結果回來後依此排序。
- 錯誤重試自動化：在大量批次執行中，更要實現前述的自動補救機制。可以寫一個收尾腳本，統一匯總所有結果，對照原始清單檢查遺漏。如果有，直接輸出一個新的 JSONL 檔包含這些遺漏項，再次執行 Batch 或改走同步，以確保最終無遺漏。
- 日誌與監控：大規模批次可能跑好幾個小時，建議加入適當的日誌。例如每隔幾分鐘 log當前已完成幾筆，或每個批次完成時記錄耗時和簡要統計。這有助於日後分析性能瓶頸，並作為成本預估依據。也可利用 OpenAI 提供的
request_counts來分析批次執行過程中的成功/失敗比率等。
遵循上述經驗分享，你可以有效地運用 OpenAI Batch API 來處理超大規模的任務，同時保持程式穩定性和結果可靠性。在確保格式正確與流程順暢的前提下，Batch API 將為您的應用帶來高性價比的模型推理能力，讓大規模的 AI 任務變得更可管理、更經濟。
今天就先分享到這邊，我們會在後續文章中繼續討論各家 LLM vendors 所提供的效率/成本優化工具在 LLMOPs與 AgentOps 中的整合點與優化經驗。
