Function lesson_3_async_iterator

tutorials/17_async_await_for_sse.py:182–237

SSE 流式响应 = 一个"异步迭代器" 每次 yield 一个事件，直到流结束。 就像: 普通迭代器: for item in [1, 2, 3] → 立即可用 异步迭代器: async for item in stream → 可能要等网络

()

# ============================================================

def lesson_3_async_iterator():
    """
    SSE 流式响应 = 一个"异步迭代器"
    每次 yield 一个事件，直到流结束。

    就像:
    普通迭代器: for item in [1, 2, 3]  → 立即可用
    异步迭代器: async for item in stream → 可能要等网络
    """
    print("\n" + "=" * 60)
    print("第三课: 异步迭代器（SSE 流的基础）")
    print("=" * 60)

    # 用 async generator 模拟 SSE 流
    async def fake_sse_stream() -> AsyncIterator[dict]:
        """模拟 Anthropic API 的 SSE 流"""
        events = [
            {"type": "message_start", "message": {"id": "msg_123"}},
            {"type": "content_block_start", "index": 0},
            {"type": "content_block_delta", "delta": {"text": "Hello"}},
            {"type": "content_block_delta", "delta": {"text": " World"}},
            {"type": "content_block_delta", "delta": {"text": "!"}},
            {"type": "content_block_stop", "index": 0},
            {"type": "message_stop"},
        ]
        for event in events:
            await asyncio.sleep(0.1)  # 模拟网络延迟
            yield event  # 像 return，但不终止函数

    async def consume_stream():
        """消费 SSE 流，逐字拼接文本"""
        full_text = ""
        async for event in fake_sse_stream():
            event_type = event["type"]
            if event_type == "content_block_delta":
                chunk = event["delta"]["text"]
                full_text += chunk
                print(f"    收到: '{chunk}'  (累计: '{full_text}')")
            else:
                print(f"    事件: {event_type}")
        return full_text

    print("\n  模拟 SSE 流消费:")
    result = asyncio.run(consume_stream())
    print(f"\n  最终文本: '{result}'")

    print("""
    关键语法:
      async def stream() -> AsyncIterator:
          yield item             # async generator

      async for item in stream():
          process(item)          # async for 循环

    这就是 Claude Code 的 MessageStream.next_event() 的 Python 版。
    """)