Responses API 實戰（三）：串流、背景任務與失敗恢復

Responses 串流的連線狀態、模型生成狀態與應用工作狀態各自獨立。連線可能在生成完成前中斷，使用者取消時也可能已有工具正在執行；背景工作則需要保存 response ID，讓瀏覽器重新整理後仍能查詢結果。恢復流程因此需要記錄事件、終止狀態與工具執行進度。

技術文件助理的長任務包含文件分析、工具等待、答案整理與結果保存。回應項目和上下文管理見回應解析與對話狀態，工具參數、結果回填及業務驗證見工具呼叫與結構化輸出。

連線、回應與工作狀態

文件分析同時涉及 HTTP／SSE 連線、OpenAI response 與應用工作三種狀態。連線可處於讀取或中斷，response 可處於生成、完成或未完成，應用工作則可能正在查文件、等待工具或寫入結果。各自的識別與生命週期需要分別保存，才能查得一次中斷影響的範圍。

層次	應保存的識別與狀態	常見誤判
傳輸	請求紀錄、開始時間、最後事件時間、斷線原因	把 EOF、Abort 或 HTTP 200 當成答案完成
模型回應	response ID、status、error、incomplete_details、完整 output	只看到一個文字 done 就提交整份結果
業務工作	應用程式 job ID、使用者範圍、工具執行紀錄、結果版本	重試模型請求時，連已完成的寫入操作也重新執行

response.completed 表示本次模型生成完成。若 output 包含 function_call，應用工作仍處於工具執行與回填階段，UI 可顯示等待查詢，工作紀錄也維持待處理狀態，直到取得最終答案。

串流終止狀態

程式使用 OpenAI Python SDK，模型為 gpt-6-astra，推理程度為 low。執行環境需安裝 openai 並設定 OPENAI_API_KEY，OPENAI_RESPONSES_MODEL 可指定相容模型。SDK 自動重試已停用，斷線與逾時會交回應用程式處理，模型請求與工具操作也各自使用獨立的重試政策。

import os
import sys
from openai import OpenAI, APIConnectionError, APITimeoutError

client = OpenAI(timeout=120.0, max_retries=0)
MODEL = os.getenv("OPENAI_RESPONSES_MODEL", "gpt-6-astra")


class StreamTurnError(RuntimeError):
    def __init__(self, message, response_id, parts, response=None):
        super().__init__(message)
        self.response_id = response_id
        self.partial = {key: "".join(text) for key, text in parts.items()}
        self.response = response


def stream_document_answer(question):
    response_id = None
    final_response = None
    parts = {}
    try:
        with client.responses.create(
            model=MODEL,
            reasoning={"effort": "low"},
            instructions="以繁體中文回答技術文件問題，缺少資料時指出需要補充的設定。",
            input=question,
            stream=True,
            store=False,
        ) as events:
            for event in events:
                if event.type == "response.created":
                    response_id = event.response.id
                    print(f"response_id={response_id}", file=sys.stderr)
                elif event.type == "response.output_text.delta":
                    key = (event.output_index, event.content_index)
                    parts.setdefault(key, []).append(event.delta)
                    sys.stdout.write(event.delta)
                    sys.stdout.flush()
                elif event.type in {
                    "response.completed", "response.incomplete", "response.failed"
                }:
                    final_response = event.response
                elif event.type == "error":
                    raise StreamTurnError(
                        f"stream error: {event.code}: {event.message}", response_id, parts
                    )
    except (APITimeoutError, APIConnectionError) as error:
        # 保留 partial 與識別供追查，不把它寫成完成的業務結果。
        raise StreamTurnError(
            f"串流中斷；response_id={response_id}", response_id, parts
        ) from error

    if final_response is None:
        raise StreamTurnError("連線結束但沒有整輪終止事件", response_id, parts)
    if final_response.status != "completed":
        raise StreamTurnError(
            f"status={final_response.status}; error={final_response.error}; "
            f"incomplete={final_response.incomplete_details}",
            response_id, parts, final_response,
        )
    if any(item.type == "function_call" for item in final_response.output):
        raise StreamTurnError(
            "模型正在等待工具結果，請接上工具處理流程",
            response_id, parts, final_response,
        )
    return final_response


if __name__ == "__main__":
    result = stream_document_answer("說明 Django ASGI 部署時應區分哪些逾時與連線狀態。")
    print("\n回應已完成；可依 output 型別進行拒絕、內容與業務驗證。")

StreamTurnError 會帶出 response ID、已累積文字與可取得的終止回應，供應用程式保存 partial 與診斷資料。文字 delta 依 output_index 和 content_index 分組，維持各文字項目的邊界。終端機即時印出片段，正式前端則需分別管理訊息與內容項目；對話重播仍使用完整 output，包括 reasoning 與 assistant phase。

response.output_text.done 只表示一個文字內容完成，後續仍可能出現其他內容、工具項目或整輪終止事件。官方串流文件定義了各階段的 typed events，Node SDK 的串流說明也記載，乾淨的 EOF 可能只留下 partial response，finalResponse 仍須依 status 判斷完成與否。

輸出不完整與串流錯誤診斷

文字顯示中斷、JSON 解析失敗及工具未執行，可能分別來自生成預算、解析時機或傳輸層。事件序列可顯示哪個階段已完成、最後收到哪個事件，以及 response 是否已進入終止狀態，據此定位輸出或連線問題。

現象	要查看的紀錄	處理方式
文字已出現，最後卻是 incomplete	整輪狀態、incomplete_details、輸出及 reasoning 用量	保留片段並標記未完成，調整任務或預算；不當成完整文件保存
工具 arguments 解析失敗	item_id／output_index、arguments delta 與 done	分呼叫累積，等完整參數後驗證，不對每個片段做 json.loads
直接連 API 會持續出字，經應用程式後一次出現	API、應用伺服器、代理、瀏覽器各層的第一個事件時間	逐層檢查緩衝與 flush；不要在後端先把整輪讀完才回給前端
沒有任何終止事件就 EOF	最後事件、response ID、SDK 錯誤與網路紀錄	標記狀態未確認；有可取回的 response 才查詢，不自動提交 partial

Structured Outputs 的完整解析需要已完成的輸出。Python SDK #2486 與併入的 #3263 記錄過未完成 JSON 在解析階段拋出例外的情況，其中 #2486 的回報版本為 1.86.0。原始狀態、SDK 版本與 ValidationError 應一併保留，才能辨認例外發生前的生成結果。

取消與工具執行狀態

停止讀取串流或取消背景生成，不會回滾工具已造成的資料變更。尤其在收到 function_call 後中斷的流程中，應用程式可能已開始執行工具，遠端對話卻尚未保存相應項目，因此取消處理須同時核對兩邊的實際狀態。

Python SDK #3561 以 2.45.0 回報：程式已收到工具呼叫，取消後該呼叫沒有保存在 conversation，下一輪回填得到「No tool call found」。這個重現有特定版本與取消時機，恢復流程則可依實際保存的呼叫、結果與執行紀錄判斷，不依賴該問題在其他版本是否存在。

取消時的位置	應用程式保留什麼	下一步
工具尚未形成完整參數	partial、最後事件與取消原因	不執行工具；重新開始或從完整且一致的對話位置接續
呼叫已完成，工具尚未執行	完整 call item、response ID、執行計畫	核對對話是否含該呼叫，再決定執行或結束該輪
工具已執行，結果尚未回填	業務操作鍵、執行結果、call_id	讀取既有結果，核對後回填；不因重試模型就再做一次操作
結果已回填，答案尚未完成	包含工具結果的完整上下文與工作狀態	延續一致的鏈；不要回到尚未回填的舊 response

工具的執行鍵由應用程式的業務工作與操作目的決定。call_id 關聯本次模型呼叫，重新生成可能得到新的 call_id；同一個業務操作因此需要自己的去重鍵，才能在模型重試時沿用既有執行結果。

背景工作與輪詢

一般串流適合需要即時顯示的短回答，較長的文件分析可使用 background mode。建立回應後，應用程式將 response ID 與自己的工作 ID 一併保存，再透過輪詢或 Webhook 接收結果。自訂工具的執行、使用者授權與產品工作紀錄仍由應用程式管理。

背景工作程式可獨立執行，工作識別保存在本機檔案，輪詢期間處理 queued／in_progress。等待時間用盡後，識別仍保留，下一次可取回同一份 response。多使用者服務則將這些欄位存入資料庫，按工作隔離紀錄。

import json
import os
import time
from pathlib import Path
from uuid import uuid4
from openai import OpenAI

client = OpenAI(timeout=120.0, max_retries=0)
job_id = str(uuid4())
response = client.responses.create(
    model=os.getenv("OPENAI_RESPONSES_MODEL", "gpt-6-astra"),
    reasoning={"effort": "low"},
    instructions="根據提供的文件整理部署檢查項目，以繁體中文完整說明。",
    input="分析 Django ASGI 的應用工作、HTTP 連線與代理逾時如何分別設計。",
    background=True,
    store=True,
    metadata={"app_job_id": job_id},
)
state_path = Path(f"response-job-{job_id}.json")
state_path.write_text(json.dumps({
    "job_id": job_id, "response_id": response.id,
}, ensure_ascii=False), encoding="utf-8")

deadline = time.monotonic() + 180
while response.status in {"queued", "in_progress"}:
    if time.monotonic() >= deadline:
        raise TimeoutError(f"等待上限已到，之後用 {state_path} 取回同一個 response")
    time.sleep(2)
    response = client.responses.retrieve(response.id)

if response.status != "completed":
    raise RuntimeError(f"{response.status}: {response.error or response.incomplete_details}")
print(response.output_text)

三分鐘是程式設定的等待上限，與 API 的統一期限無關。使用者取消背景生成時，可呼叫 client.responses.cancel(response_id) 並保存回傳的終止狀態。官方 cancel 操作可重複呼叫並取得最終 response；工具操作已產生的結果仍按業務流程另外處理。

store=True 的保存期限依資源政策決定，response、conversation、檔案與產品資料庫各自管理。背景執行另有供輪詢使用的暫存規則，因此產品仍需保存原始工作紀錄與必要結果。資源保留條件見Background mode及資料控制文件。

事件續接與游標

背景串流支援依事件位置續接，原始 response 必須在建立時啟用 background。一般同步串流、背景串流與 WebSocket 各有不同的恢復條件，普通串流中斷後無法直接套用背景事件的游標。

原始請求需要同時設定 background=True 與 stream=True，只有 background 的輪詢回應不適用串流續接。下面的程式會建立獨立背景串流並保存 response ID，首次連線中斷後，該 ID 可傳給事件庫函式。背景事件的 sequence_number 作為續接位置，游標對應已提交的事件紀錄；若游標比文字更早提交，中途故障可能造成資料遺失。SQLite 以 response 與事件序號的唯一鍵去重，游標由已提交紀錄取得。函式回傳可保存的 response 字典，完整事件則供前端重建畫面；呼叫端仍依 status 與 output 型別決定後續工作。

import json
import os
from pathlib import Path
from uuid import uuid4
from openai import OpenAI

client = OpenAI(timeout=120.0, max_retries=0)
stream_job_id = str(uuid4())
with client.responses.create(
    model=os.getenv("OPENAI_RESPONSES_MODEL", "gpt-6-astra"),
    reasoning={"effort": "low"},
    input="整理 Django ASGI 文件中與長連線、取消和背景工作有關的部署檢查項目。",
    background=True, stream=True, store=True,
    metadata={"app_job_id": stream_job_id},
) as events:
    for event in events:
        if event.type == "response.created":
            Path(f"background-stream-{stream_job_id}.json").write_text(
                json.dumps({"response_id": event.response.id}), encoding="utf-8"
            )
        print(event.type)
# 若途中斷線，從上述檔案取得 ID，再呼叫後面的 resume_background。
# 此段只列印事件型別，完整事件可由後面的持久化流程重播。

import json
import sqlite3


def resume_background(client, response_id, database="response-events.sqlite3"):
    database_connection = sqlite3.connect(database)
    try:
        database_connection.execute("""
            CREATE TABLE IF NOT EXISTS response_events (
                response_id TEXT NOT NULL,
                sequence_number INTEGER NOT NULL,
                payload TEXT NOT NULL,
                PRIMARY KEY (response_id, sequence_number)
            )
        """)
        cursor = database_connection.execute(
            "SELECT MAX(sequence_number) FROM response_events WHERE response_id = ?",
            (response_id,),
        ).fetchone()[0]
        last = database_connection.execute(
            "SELECT payload FROM response_events WHERE response_id = ? "
            "ORDER BY sequence_number DESC LIMIT 1", (response_id,),
        ).fetchone()
        terminal = {"response.completed", "response.incomplete", "response.failed"}
        if last is not None:
            saved = json.loads(last[0])
            if saved["type"] in terminal:
                return saved["response"]
        options = {} if cursor is None else {"starting_after": cursor}
        final_response = None
        with client.responses.stream(response_id=response_id, **options) as events:
            for event in events:
                # keepalive 的序號不是已保存事件的續接位置。
                if event.type == "keepalive":
                    continue
                payload = json.dumps(event.model_dump(mode="json"), ensure_ascii=False)
                with database_connection:
                    database_connection.execute(
                        "INSERT OR IGNORE INTO response_events VALUES (?, ?, ?)",
                        (response_id, event.sequence_number, payload),
                    )
                if event.type in {
                    "response.completed", "response.incomplete", "response.failed"
                }:
                    final_response = event.response.model_dump(mode="json")
        if final_response is None:
            raise RuntimeError("尚未收到終止事件，保留事件庫並取回同一份背景工作")
        return final_response
    finally:
        database_connection.close()

事件庫保存同一個 response 的事件，重新連線使用 retrieve，無須重新 create。前端依 response ID 與序號去重並組合文字項目，工具結果則與自己的操作紀錄核對。終止事件若已保存，可直接讀取完成紀錄。接續條件與 SDK 用法見背景串流文件。

WebSocket 提供持續連線與連線內上下文快取。previous_response_id 用來延續上下文，背景事件游標則定位待續接的事件，兩者使用不同的識別與恢復機制。連線中斷後還需考慮可取得的持久狀態，具體條件見WebSocket 文件。

Webhook 驗簽與 Inbox 處理

背景工作完成後可透過 Webhook 通知應用程式，同一通知可能重送。HTTP handler 驗證原始 body 並保存事件，提交成功後回覆；retrieve、資料轉換及後續業務工作由工作者處理。成功回覆早於持久化時，程序中斷可能讓已確認的事件遺失。

兩段程式分別是 Django model 與 view，需放進既有 app、建立 migration、設定 URL 和 OPENAI_WEBHOOK_SECRET。收件表提供可重試的 inbox，事件唯一鍵去重，資料庫工作者讀取 pending 紀錄。工作者透過交易領取紀錄並保存狀態，程序重啟後可繼續處理未完成工作。

# models.py
from django.db import models


class ResponseWebhookInbox(models.Model):
    event_id = models.CharField(max_length=255, primary_key=True)
    payload = models.JSONField()
    state = models.CharField(max_length=16, default="pending")
    received_at = models.DateTimeField(auto_now_add=True)

# views.py
import json
import os
from django.db import transaction
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from openai import OpenAI, InvalidWebhookSignatureError
from .models import ResponseWebhookInbox

client = OpenAI(webhook_secret=os.environ["OPENAI_WEBHOOK_SECRET"])


@csrf_exempt
@require_POST
def response_webhook(request):
    raw_body = request.body
    try:
        client.webhooks.unwrap(raw_body, request.headers)
        payload = json.loads(raw_body)
        event_id = request.headers["webhook-id"]
    except (InvalidWebhookSignatureError, ValueError, KeyError):
        return HttpResponse(status=400)

    with transaction.atomic():
        ResponseWebhookInbox.objects.get_or_create(
            event_id=event_id,
            defaults={"payload": payload},
        )
    return HttpResponse(status=200)

簽章驗證使用原始 body，重新編碼 JSON 會改變驗證所需的內容。webhook-id 依官方 Webhook 文件作為去重鍵，重送事件在資料庫只保存一筆。Handler 回覆 200 表示事件已提交，後續的文件分析與業務操作則有各自的完成狀態。

工作者使用已驗證事件中的 data.id 取回 response，核對應用工作與狀態，將結果寫入結果表後才把 inbox 標為已處理。Webhook 事件與業務結果各有去重需求，不同事件可能指向同一個工作，結果表因此也須設定業務工作或 response 的唯一識別。

重試上限與請求紀錄

SDK、反向代理、工作佇列與前端的重試次數會互相累乘，等待時間也可能超過單一呼叫設定的期限。整個工作可等待的時間、模型嘗試次數及工具操作鍵，應作為各層設定的共同依據。重新建立 response 可能再次生成或觸發工具，取回既有 response 則是查詢狀態，兩者的費用與副作用不同。

失敗情境	處理重點
建立回應時逾時，還沒有取得 ID	記為結果未確認，保留自己的 job ID、開始時間與請求紀錄；有副作用流程先查操作狀態，不直接重做整段
已取得背景 response ID，之後讀取中斷	取回同一個 ID 或從已保存游標續接，避免再次 create
429 或暫時性服務錯誤	在工作總預算內退避重試並記錄次數，避免 SDK、佇列與前端同時無上限重試
400 的資料形狀或工具配對錯誤	修正 input／call_id／上下文資料；相同請求原樣重送不會補回遺漏的工具結果
工具完成但回填失敗	保留已完成操作與結果，恢復一致的對話後回填，不重複執行業務操作

工作紀錄包含應用 job ID、response ID、可取得的 request ID、模型、SDK 版本、首個事件時間、終止事件時間、status、用量及工具操作鍵。這些資料可用來比對模型生成、網路等待、工具執行與代理緩衝的耗時；文字、文件及其他敏感輸入的保存範圍另依產品資料政策管理。

中斷與恢復測試

固定事件與工具結果可驗證文件助理的狀態轉換，包括下列中斷與恢復分支。程式分支測試不必每次呼叫模型，實際模型的輸出品質、延遲與用量則由另一組測試記錄，兩者分別驗證程式分支與模型行為。

測試輸入	應通過的條件
只有 delta 和文字 done，沒有整輪終止事件	文字可以顯示為 partial，但不提交完成結果
輸出最後成為 incomplete	保留原因與用量，不進入正常 JSON／業務成功分支
收到完整工具呼叫後取消	核對是否執行與是否保存，不直接把半套上下文接成下一輪
同一個背景事件重送兩次	事件庫與 UI 去重，游標只反映已持久化的位置
同一個 Webhook 重送兩次	inbox 只增加一筆，未完成的工作仍可被工作者重試
程序在保存事件後重新啟動	由既有 response ID 與事件紀錄恢復，不另建重複工作

文字解析、對話歷史及工具回填可分別參考回應與對話狀態、工具與資料驗證；既有 Chat Completions 程式的欄位與端點差異見遷移對照。