實作課程 4 — 事件¶
目標: 理解Copilot SDK工作階段事件生命週期:SDK實際上發出什麼事件,事件到達的順序以及對常見任務(如串流、監控、完成和錯誤)而言哪些事件是重要的。
時間: ~20分鐘
必要條件: 實作課程 3 完成。
第1步 — 執行事件範例¶
從 src/AgentOrchestrator-python執行Copilot SDK實驗範例:
新增 --model <id> 覆蓋模型:
預期輸出:
== 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-5) — 工作階段開始,待處理的訊息和技能載入,系統訊息出現,工具以待命訊息、技能和工具的順序宣佈。
session.tools_updated - 使用者回合(6-9) — 使用者訊息被接受,掛鉤事件在帶帶執行,工作階段標題可以改變。
- 助理回合開始(10-12) — 幫助助理回合開始,使用資訊表面,模型呼叫開始
- 串流 (13–28) — 串流和推理差異到達,隨後是
assistant.message_start - 完成 (29–32) — 助理使用報告,最終助理 訊息到達,推理最終化,助理輪次結束
- 拆卸和空閒 (33–37) — 另一個掛鉤對執行,使用被 檢查點,助理變得空閒,然後整個工作階段變得空閒
- 關閉 (38–40) — 發生在
async with session:解壓
掛鉤事件是同一有序流的一部分。如果你正在探索治理掛鉤,那順序很重要,因為 hook.start 和 hook.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_delta 和 assistant.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_delta和assistant.message_delta不同的事件- 空閒狀態未關閉——當新的事件到達時,工作階段會繼續擴充套件
async with工作階段的解包過程 - 每條記錄事件都是有用的,但對應用程式碼來說是噪音
- 僅僅等待助理訊息是不夠的;完成時請務必
session.idle - 忽略
session.error工作階段的等待時間會因為全時超時而被您的呼叫者等待 - 處理程式被呼叫 由SDK ——保持它們快速,而不是在內部做慢工作
💡 延伸挑戰¶
- 更改範例以僅列印與工具相關的事件——那些與工具相關的事件
evt.type.value開始於tool. - 測量時間到第一個令牌之前
time.perf_counter()停止在您關心的第一個差分之前session.send(...)單獨計數與事件相關的事件,並記錄它們出現的位置 - 將事件分組到上述生命週期階段中的七個階段中,而不是列印一個平鋪的數字清單
- 列印
- 與
evt.raw_type一起evt.type.value並檢視兩個之間的差異
✅ 檢查點¶
現在你可以解釋:
- [x] SDK 發射的有序工作階段生命週期
- [x] Python 使用了一個
SessionEvent與evt.typeenum, 而不像 C# - ✓ 那
session.on傳回一個取消訂閱的可呼叫 - ✓ 為什麼真實的應用程式處理了所有發出事件的很小一部分
- ✓ 與之不同的
streaming_delta和message_delta - ✓ 為什麼
session.idle✓ 完成操作,為什麼空閒≠關閉 - ✓ 為什麼錯誤事件必須失敗等待的未來,為什麼超時很重要
- ✓ 掛鉤和使用事件出現在生命週期中
相關¶
- 上一個: 實作課程 3 — 工具
- 下一頁: 實作課程 05 — 工作階段
- 示範:Copilot SDK 整合
- 額外 — 治理掛鉤
- 掛鉤和治理
- 疑難排解