實作課程 02 — 你的第一個串流聊天

目標： 遵循一個提示詞，從Python 技術堆疊的頂部到底部——HTTP使用者端 → FastAPI → Copilot SDK → 模型 → 後端 —— 並理解為什麼模型探索是執行時資料而不是寫死的清單。

時間： ~20分鐘

必要條件： 實作課程 01 完成，FastAPI伺服器執行在埠5070。

第 1 步 — 啟動或確認伺服器

從存放庫根目錄：

cd src/AgentOrchestrator-python
uv run uvicorn app.main:app --port 5070

開啟 http://localhost:5070 如果你想稍後檢視UI。這個過程不僅適用於API，也適用於靜態UI。

確認API是否健康:

curl http://localhost:5070/api/chat/health

預期結果：

{"status":"healthy","service":"CopilotChat","availableModels":["claude-haiku-4.5","gpt-4.1","gpt-5","claude-sonnet-4.5","claude-opus-4.5","gemini-2.5-pro"]}

那個備用清單故意很小。真正的模型選擇器詢問你的已登入的Copilot帳戶它能使用。

第 2 步 — 觀察傳輸格式

傳送提示並觀察原始的串流事件:

curl -sN -X POST http://localhost:5070/api/chat/stream \
    -H 'Content-Type: application/json' \
    -d '{"prompt":"Reply with exactly: streaming works","model":"claude-haiku-4.5"}'

捕獲輸出:

data: {"content": "streaming works"}

data: [DONE]

注意：每個幀是 data: 加上JSON，然後是 空行多個幀的多個答案會因為分段而分散；流的結束以 data: [DONE].

⚠️ 請求模型是 prompt, model, 可選 systemMessage的鍵 message 被無聲地忽略——你將收到一個通用的問候，而不是一個驗證錯誤。

第 3 步 — 找到伺服器端

開啟 app/routers/chat.py 和定位 stream_chat.

stream_chat 傳回一個 StreamingResponse 媒體型別 text/event-stream. 在 event_stream, 每個 SDK 分段被序列化為一個 SSE 幀:

yield f"data: {json.dumps({'content': chunk})}\n\n"

然後寫 data: [DONE] 終結者

⚠️ 空行不是裝飾。它是事件分隔符號。刪除它並 許多 SSE 使用者端將保持緩衝，因為它們從未看到完整的事件。

現在看看 except 塊。錯誤寫入 到流 作為 data: {"error": "..."}. 一旦 SSE 回應開始，狀態行和 標頭已經消失了，因此 HTTP 500 對使用者端來說已經不再有用。

第 4 步 — 找到 SDK 整合

開啟 app/services/copilot_chat.py.

CopilotChatService.chat_stream 開啟串流的工作階段:

session = await self._client.create_session(
    model=model,
    streaming=True,
    system_message=(
        {"mode": "append", "content": system_message} if system_message else None
    ),
)

然後它訂閱事件:

def on_event(evt: SessionEvent) -> None:
    # Unlike .NET, every event arrives as one SessionEvent
    # carrying a `type` enum and a `data` payload, so this
    # dispatches on `evt.type` rather than on subclasses.
    if evt.type is SessionEventType.ASSISTANT_MESSAGE_DELTA:
        queue.put_nowait(evt.data.delta_content or "")
    elif evt.type is SessionEventType.ASSISTANT_MESSAGE:
        logger.info(
            "Assistant response complete: %d chars",
            len(evt.data.content or ""),
        )
    elif evt.type is SessionEventType.SESSION_IDLE:
        if not done.done():
            done.set_result(None)
    elif evt.type is SessionEventType.SESSION_ERROR:
        logger.error("Session error: %s", evt.data.message)
        if not done.done():
            done.set_exception(RuntimeError(evt.data.message))

佇列是 SDK 的回呼風格與 FastAPI 的非同步回應生成器之間的橋樑。回呼將分塊推送到 asyncio.Queue; 路由等待佇列並傳回 SSE 幀。

第 5 步 — 理解 Python 事件

這是第一個重要的 Python-vs-.NET 差異。

.NET 樣例模式比對事件子類，如 AssistantMessageDeltaEvent 和 SessionIdleEvent. Python 給你一個 SessionEvent 資料類別，具有欄位如:

• type
• data
• id
• timestamp

事件 type 欄位是 SessionEventType 列舉，所以 Python 程式碼分支如:

if evt.type is SessionEventType.ASSISTANT_MESSAGE_DELTA:
    ...
elif evt.type is SessionEventType.SESSION_ERROR:
    ...

事件也是 唯讀回呼. session.on(handler) 傳回一個 取消訂閱可呼叫；沒有直接的非同步迭代器可以迴圈。

在 sdk_labs/_common.py 使用此模式來等待 SESSION_IDLE 或丟擲錯誤 SESSION_ERROR.

第 6 步 — 詢問 API 哪些模型

curl -s http://localhost:5070/api/chat/models | jq -r '.[].id'

/api/chat/models 傳回一個 JSON 清單 由形狀像:

{"id":"...","name":"...","description":"..."}

該路由首先詢問 CopilotChatService.list_models(), 誰呼叫 await client.list_models(). 那會傳回 ModelInfo 具有 .id 和 .name 欄位。

⚠️ list_models() 有一個已知的上游錯誤: 它可以丟擲 ValueError: Missing required field 'multiplier' in ModelBilling (github/copilot-sdk#1302). 路由器廣泛地捕獲並退回到靜態目錄，因此 UI 仍然有選擇。

第 7 步 — 切換模型並比較

從你的即時清單中選擇一個模型 ID:

MODEL=$(curl -s http://localhost:5070/api/chat/models | jq -r '.[0].id')
echo "Using $MODEL"

curl -sN -X POST http://localhost:5070/api/chat/stream \
  -H 'Content-Type: application/json' \
  -d "{\"prompt\":\"In one sentence, what is customer churn?\",\"model\":\"$MODEL\"}"

重複使用不同的 ID 並比較延遲、風格和語氣。

在 sdk_labs/model_picker.py 使用相同的主意: 優先 claude-haiku-4.5, 但退回到一個具體的模型 --model <id>.

第 8 步 — 用系統訊息塑造回應

API 接受可選 systemMessage. 服務將它傳送到 附加 模式 因此它補充了工作階段內建的指令，而不是替換它們。

curl -sN -X POST http://localhost:5070/api/chat/stream \
  -H 'Content-Type: application/json' \
  -d '{
    "prompt":"Which segment has the lowest retention?",
    "model":"claude-haiku-4.5",
    "systemMessage":"You are a retail analytics assistant. Context: 4 segments — High Value (92% retention), Regular (78%), At Risk (45%), New (65%). Answer in one sentence."
  }'

預期行為: 一個基於實際的答覆 有流失風險 在 45% 的風險

第 9 步 — 知道基本工作階段呼叫

Python SDK 物件支援 async with所以，樣例會自動清理，同時為使用者端和工作階段提供確定性。

對於非流式一次性提示，SDK還暴露了：

await session.send_and_wait(prompt, timeout=60.0)

FastAPI服務使用 send(prompt) 因為它處理每個差分到達時。

第 10 步 — 嘗試 UI 路徑

回到瀏覽器中，在 http://localhost:5070選擇一個模型，問 ‘請列出三個零售KPI。一條一行。'，並觀察訊息分塊渲染。如果串流失敗，檢查 data: {"error": "..."} 幀。

完成的對話看起來如下——使用的 SSE 訊框與上述 curl 輸出相同，由 app.js 解析並轉譯為 Markdown：

💡 串流期間，助理訊息泡泡會顯示動態輸入指示器；收到 [DONE] 終止標記後，便會以轉譯完成的 Markdown 取代。

⚠️ 靜態 UI 會傳送 { prompt, model }。此物件由 app/static/app.js 建立，並與 ChatRequest 相符。早期版本誤傳 message；由於 Pydantic 會忽略未知鍵而非拒絕要求，瀏覽器會針對空白提示串流回應，卻不會明確回報錯誤。tests/test_chat_contract.py 現在會驗證 app.js 傳送的欄位與 API 讀取的欄位一致。

✅ 檢查點

現在你可以解釋：

• [x] 錯誤被串流而不是作為HTTP狀態程式碼傳回的原因
• [x] 如何FastAPI的
• 包裝了SDK流 StreamingResponse 的
• 為什麼Python在分支上 evt.type 而不是事件子類
• 為什麼模型在執行時被發現
• 為什麼系統訊息將助理與零售領域聯絡起來

💡 延伸挑戰

開啟 app/services/copilot_chat.py 並暫時記錄每個事件型別 在鏈之前 if 傳送一個簡短的提示詞並比較事件序列與助理範例:

uv run python -m sdk_labs events

那個更深層次的事件生命週期範例是 sdk_labs/events_sample.py.

相關

• 上一個: 實作課程 01 — 環境設定
• 下一頁: 實作課程 3 — 工具
• 示範：Copilot SDK 整合