跳轉到內容

實作課程 4 — 事件

目標: 理解Copilot SDK工作階段事件生命週期:SDK實際上發出什麼事件,事件到達的順序以及對常見任務(如串流、監控、完成和錯誤)而言哪些事件是重要的。

時間: ~20分鐘

必要條件: 實作課程 3 完成。

第1步 — 執行事件範例

src/AgentOrchestrator-python執行Copilot SDK實驗範例:

cd src/AgentOrchestrator-python
uv run python -m sdk_labs events

新增 --model <id> 覆蓋模型:

uv run python -m sdk_labs events --model gpt-5-mini

預期輸出:

== Lab 04: events ==

Model: claude-haiku-4.5
Prompt: Name two retail KPIs. One line each.

  1. session.start
  2. pending_messages.modified
  3. session.skills_loaded
  4. system.message
  5. session.tools_updated
  6. user.message
  7. hook.start
  8. session.title_changed
  9. hook.end
 10. assistant.turn_start
 11. session.usage_info
 12. model.call_start
 13. assistant.streaming_delta
 14. assistant.reasoning_delta
 15. assistant.streaming_delta
 16. assistant.reasoning_delta
 17. assistant.streaming_delta
 18. assistant.reasoning_delta
 19. assistant.streaming_delta
 20. assistant.reasoning_delta
 21. assistant.streaming_delta
 22. assistant.reasoning_delta
 23. assistant.streaming_delta
 24. assistant.reasoning_delta
 25. assistant.streaming_delta
 26. assistant.message_start
 27. assistant.streaming_delta
 28. assistant.streaming_delta
 29. assistant.usage
 30. assistant.message
     content: 1. **Sales per Square Foot** — Revenue generated per unit of retail floor space;…
     (preceded by 3 delta events)
 31. assistant.reasoning
 32. assistant.turn_end
 33. hook.start
 34. hook.end
 35. session.usage_checkpoint
 36. assistant.idle
 37. session.idle

Total delta events: 3
 38. session.shutdown
 39. session.background_tasks_changed
 40. session.background_tasks_changed

重要的驚喜是體積。一個簡單的提示交換髮出遠多於"使用者訊息,助理訊息,完成"。

⚠️ 事件38-40在總結行之後出現。 Total delta events: 3 工作階段報告空閒時列印,但塊尚未退出。在離開時, teardown發出三個額外事件。這是一個有用的提醒,說明 async with session: 空閒不是關閉的同義詞 空閒與關閉不同.

⚠️ 上述數字是觀察到的一次執行,不是合同。 確切的計數和順序取決於模型、提示和SDK版本——讀取其形狀,而不是作為固定規格。

第2步 — 走生命週期階段

如果將事件流按階段分組,更容易記住。

  1. 工作階段設定(1-5) — 工作階段開始,待處理的訊息和技能載入,系統訊息出現,工具以待命訊息、技能和工具的順序宣佈。 session.tools_updated
  2. 使用者回合(6-9) — 使用者訊息被接受,掛鉤事件在帶帶執行,工作階段標題可以改變。
  3. 助理回合開始(10-12) — 幫助助理回合開始,使用資訊表面,模型呼叫開始
  4. 串流 (13–28) — 串流和推理差異到達,隨後是 assistant.message_start
  5. 完成 (29–32) — 助理使用報告,最終助理 訊息到達,推理最終化,助理輪次結束
  6. 拆卸和空閒 (33–37) — 另一個掛鉤對執行,使用被 檢查點,助理變得空閒,然後整個工作階段變得空閒
  7. 關閉 (38–40) — 發生在 async with session: 解壓

掛鉤事件是同一有序流的一部分。如果你正在探索治理掛鉤,那順序很重要,因為 hook.starthook.end 出現在工作而不是在單獨的側通道中。請參閱 額外 — 治理掛鉤掛鉤和治理.

使用也有自己的事件: session.usage_info, assistant.usage,以及 session.usage_checkpoint. 這些是需要檢查以獲取令牌和成本資料的事件,而不是文字內容。

第 3 步 — 訂閱 session.on

開啟 events_sample.py 和找到訂閱:

            waiter = IdleWaiter()
            counters = {"order": 0, "deltas": 0}

            def on_event(evt: SessionEvent) -> None:
                # Deltas arrive in a flood; count them instead of printing each one.
                if evt.type is SessionEventType.ASSISTANT_MESSAGE_DELTA:
                    counters["deltas"] += 1
                    return

                counters["order"] += 1
                # Unlike C#, the event type is a value on the event rather than
                # a subclass, so this prints evt.type instead of a class name.
                print(f"{counters['order']:3d}. {evt.type.value}")

                if evt.type is SessionEventType.ASSISTANT_MESSAGE:
                    print(f"     content: {trim(evt.data.content)}")
                    print(f"     (preceded by {counters['deltas']} delta events)")
                elif evt.type is SessionEventType.SESSION_ERROR:
                    print(f"     ERROR: {evt.data.message}")

                waiter.handle(evt)

            session.on(on_event)

實驗專案附加 PyPI github-copilot-sdk 1.0.9, 從 copilot匯入,要求 Python 3.11 或更高版本。

⚠️ 這是 .NET SDK 與 .NET SDK 最大的結構差異。 在 C# 中,每個事件都是自己的類,你模式比對在子類上:

// .NET — one class per event
session.On<SessionEvent>(evt => Console.WriteLine(evt.GetType().Name));

在 Python 中,只有一個 SessionEvent 資料類別。事件型別是物件上的一個,而不是物件本身的型別。

這就是為什麼轉錄列印的原因 assistant.message (列舉的 .value) 在.NET實驗列印 AssistantMessageEvent (類名)。使用 is 進行比較—— SessionEventType 成員是單例。

SessionEvent 攜帶 data, id, timestamp, type, agent_id, ephemeral, parent_id,以及 raw_type.形狀取決於 evt.data 取決於 evt.type,因此範例唯讀取 evt.data.content 內部 ASSISTANT_MESSAGE 分支。

💡 session.on(handler) 傳回一個取消訂閱的可呼叫.保持它,如果你需要在工作階段結束前停止監聽;例如,儲存結果 並稍後呼叫它。 session.on(on_event) 範例幾乎記錄了所有內容,讓你可以學習生命週期。真正的應用程式 不需要所有這些。

第4步 — 比較應用處理的內容

和檢視其處理程式。它只對四種事件型別做出反應:

開啟 copilot_chat.py 那是一個合理的生產選擇。對於瀏覽器串流,應用程式需要 文字塊、最終訊息記錄、完成訊號和錯誤路徑。它不需要在每 個設定、掛鉤、推理或 遙測 事件上分支。

                        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))

注意,兩個機制一起工作:文字塊被新增到一個流中, 這樣它們可以立即被串流,而空閒和錯誤 解決一個 停止的 指示器。這個分隔是

示範 01 asyncio.Queue 這樣它們可以立即串流,當空閒和錯誤時 Future 那告訴生成器何時停止。那分隔物是 示範 01.

⚠️ 有兩個不同的delta家族。 assistant.streaming_deltaassistant.message_delta 相同的事件。 ASSISTANT_MESSAGE_DELTA範例只抑制 Total delta events: 3 與許多可見的 assistant.streaming_delta 行在上面的轉錄中。如果你訂閱了錯誤的,你可能會看到預期之外的少得多的片段。在假設名稱意味著相同的東西之前,測量你的場景實際發出的內容。

第5步 — 在空閒時完成,錯誤時失敗

session.idle 範例和應用程式使用相同的完成訊號。 唯讀回呼 因為事件是—沒有非同步迭代器 await — 你需要一個處理程式,該處理程式在回呼中解決。範例共享 IdleWaiter:

class IdleWaiter:
    """Resolves when the session reports idle, or raises on session error."""

    def __init__(self) -> None:
        self._future: asyncio.Future[None] = asyncio.get_event_loop().create_future()

    def handle(self, evt: SessionEvent) -> bool:
        """Returns True when the event was a terminal (idle/error) event."""
        if evt.type is SessionEventType.SESSION_IDLE:
            if not self._future.done():
                self._future.set_result(None)
            return True
        if evt.type is SessionEventType.SESSION_ERROR:
            if not self._future.done():
                self._future.set_exception(RuntimeError(evt.data.message))
            return True
        return False

    async def wait(self) -> None:
        await asyncio.wait_for(self._future, timeout=TIMEOUT_SECONDS)

在事件範例中,回呼上面的呼叫 waiter.handle(evt) 每個事件;在傳送提示後,範例等待 waiter.wait() 列印總結之前

這是C#的 TaskCompletionSource.

⚠️ 總是處理錯誤事件。 如果工作階段失敗且沒有設定異常, await waiter.wait() 等待超時到期。這種失敗模式很容易被忽略,因為完美路徑的工作方式。

⚠️ 總是設定一個超時。 IdleWaiter.wait() 將未來包裹在 asyncio.wait_for(..., timeout=180). 一個傳輸丟失意味著 idle錯誤可能永遠不會到達,而沒有超時,你的協程將永遠掛起。 💡 如果你只想要回復,而不需要關心生命週期,SDK有一個快捷方式,這會等待你:

有一個 await session.send_and_wait(prompt, timeout=180).

⚠️ 警告

  • 資料類別;根據 SessionEvent ,而不是根據子類 evt.type
  • assistant.streaming_deltaassistant.message_delta 不同的事件
  • 空閒狀態未關閉——當新的事件到達時,工作階段會繼續擴充套件 async with 工作階段的解包過程
  • 每條記錄事件都是有用的,但對應用程式碼來說是噪音
  • 僅僅等待助理訊息是不夠的;完成時請務必 session.idle
  • 忽略 session.error 工作階段的等待時間會因為全時超時而被您的呼叫者等待
  • 處理程式被呼叫 由SDK ——保持它們快速,而不是在內部做慢工作

💡 延伸挑戰

  1. 更改範例以僅列印與工具相關的事件——那些與工具相關的事件 evt.type.value 開始於 tool.
  2. 測量時間到第一個令牌之前 time.perf_counter() 停止在您關心的第一個差分之前 session.send(...) 單獨計數與事件相關的事件,並記錄它們出現的位置
  3. 將事件分組到上述生命週期階段中的七個階段中,而不是列印一個平鋪的數字清單
  4. 列印
  5. evt.raw_type 一起 evt.type.value 並檢視兩個之間的差異

✅ 檢查點

現在你可以解釋:

  • [x] SDK 發射的有序工作階段生命週期
  • [x] Python 使用了一個 SessionEventevt.type enum, 而不像 C#
  • ✓ 那 session.on 傳回一個取消訂閱的可呼叫
  • ✓ 為什麼真實的應用程式處理了所有發出事件的很小一部分
  • ✓ 與之不同的 streaming_deltamessage_delta
  • ✓ 為什麼 session.idle ✓ 完成操作,為什麼空閒≠關閉
  • ✓ 為什麼錯誤事件必須失敗等待的未來,為什麼超時很重要
  • ✓ 掛鉤和使用事件出現在生命週期中