跳至内容

实验 04 — 事件

目标:理解Copilot SDK会话事件生命周期：SDK实际上会发出什么事件，事件到达的顺序是什么，以及对于诸如流式传输、遥测、完成和错误等常见任务，哪些事件是重要的。

时间: ~20分钟

先决条件： 实验室 03 已完成。

步骤 1 — 运行事件示例

从 src/AgentOrchestrator-python，运行 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

The 重要惊喜在于消息的体积。一个简单的单提示交流发出的消息远多于“用户消息，助手消息，完成”。

⚠️ 第38到第40条事件在总结行之后到达。 Total delta events: 3 一旦会话报告为空闲，就会立即打印出来，但 async with session: 块还没有退出。清理在离开时又发出三条事件。这提醒我们 空闲并不等于关闭.

⚠️ 这些数字是观察到的一次运行结果，而非契约规定。 模型、提示词和 SDK 版本的不同会导致具体的计数和排序变化 —— 请查看序列的结构，而非将其视为固定不变的规格。

步骤 2 — 走完生命周期阶段

事件流更容易记住，如果你按阶段对其进行分组：

  1. 会话配置 (1–5) — 会话开始，等待的消息和技能加载，系统消息出现，工具通过 session.tools_updated 发布
  2. 用户轮到发言 (6–9) — 用户的消息被接受，钩子事件在-band运行，且会话标题可以改变
  3. 助手轮到发言 (10–12) — 助手的发言开始，使用信息显现，模型调用开始
  4. 流式传输 (13–28) — 流式传输和推理差异到达，随后是 assistant.message_start
  5. 完成阶段（29–32） — assistant 使用被报告，最终助手消息到达，推理被最终确定，助手回合结束
  6. 清理和空闲（33–37） — 另一组钩子运行，使用情况被检查点，助手变得空闲，然后整个会话变得空闲
  7. 关闭 (38–40) — 在 async with session: 逆向执行时发出

Hook events are part of the same ordered stream. If you are exploring 政府治理钩子，那里的顺序很重要，因为 hook.start 和 hook.end 出现在工作周围，而不是在一个单独的侧信道中。参见 额外 — 治理钩子 和 钩子和治理。

Usage 也有自己的事件：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 最大的结构差异。 在 C# 中，每个事件都是一个独立的类，并且您通过模式匹配在子类上进行：

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

In 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 — 比较应用处理的内容

The 示例日志几乎记录了所有内容，以便您可以学习生命周期。 实际的应用程序并不需要那么多。

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

这是一个合理的生产选择。对于浏览器流式传输，该应用需要文本块、最终消息日志、完成信号，以及错误路径。它不需要在每个设置、钩子、推理或遥测事件上分支。

注意这两个机制的协同工作：文本块被放到一个 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() 在打印摘要之前。

这是 Python 的 TaskCompletionSource 的类似物。

⚠️ 始终处理错误事件。 如果会话失败且没有任何设置异常，则 await waiter.wait() 等待超时过期。这种失败模式容易被忽略，因为乐观路径工作得很好。

⚠️ 始终设置超时。 IdleWaiter.wait() 将未来包裹在 asyncio.wait_for(..., timeout=180) 中。连接丢失意味着空闲和错误可能永远不会到达，而且没有超时，您的协程将永远挂起。

💡 如果你只需要回复而不需要关心生命周期，SDK 提供了一个为你省去等待的快捷方式： await session.send_and_wait(prompt, timeout=180).

⚠️ 捕捉

  • 有一个SessionEvent数据类；根据evt.type分支，而不是根据子类分支>
  • assistant.streaming_delta 和 assistant.message_delta 是不同的事件
  • 空闲状态不关闭——更多事件随着async with块的 unwind 消失而到达
  • 记录每个事件对于学习是有用的，但对于应用程序代码来说噪音较大
  • 仅等待助手的消息是不够的；必须在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] The 有序会话生命周期由 SDK 发出
  • [x] 为什么 Python 使用一个 SessionEvent 而不是像 C# 那样使用一个 evt.type 枚举
  • [x] 那 session.on 返回一个取消订阅的可调用函数
  • [x] 实现中的应用只处理一部分已发出的事件
  • [x] 代码中的streaming_delta和message_delta的区别
  • [x] 为什么 session.idle 完成操作，以及为什么 idle ≠ closed
  • [x] 为什么错误事件必须失败等待的未来，以及为什么超时很重要
  • [x] 在生命周期中钩子和使用事件出现的位置