Function lesson_6_queue

tutorials/17_async_await_for_sse.py:415–463

在 agent 系统中，经常需要一个组件产生数据，另一个组件消费数据。 比如: SSE 解析器 → 产出事件 → agentic loop 消费事件

()

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

def lesson_6_queue():
    """
    在 agent 系统中，经常需要一个组件产生数据，另一个组件消费数据。
    比如: SSE 解析器 → 产出事件 → agentic loop 消费事件
    """
    print("\n" + "=" * 60)
    print("第六课: asyncio.Queue（生产者/消费者）")
    print("=" * 60)

    async def sse_producer(queue: asyncio.Queue):
        """生产者: 模拟 SSE 事件到达"""
        events = ["message_start", "delta:Hello", "delta: World", "message_stop"]
        for event in events:
            await asyncio.sleep(0.1)
            await queue.put(event)
            print(f"    [生产者] 放入: {event}")
        await queue.put(None)  # 结束信号

    async def ui_consumer(queue: asyncio.Queue):
        """消费者: 模拟 UI 逐字显示"""
        text = ""
        while True:
            event = await queue.get()
            if event is None:
                break
            if event.startswith("delta:"):
                chunk = event[6:]
                text += chunk
                print(f"    [消费者] 显示: '{chunk}' (累计: '{text}')")
            else:
                print(f"    [消费者] 处理: {event}")

    async def demo():
        queue = asyncio.Queue()
        # 生产者和消费者同时运行
        await asyncio.gather(
            sse_producer(queue),
            ui_consumer(queue),
        )

    print("\n  生产者(SSE)和消费者(UI)同时运行:")
    asyncio.run(demo())

    print("""
    这个模式在 Claude Code 中的应用:
    - SSE 解析器 → Queue → 消息构建器
    - Bash 命令输出 → Queue → 进度报告器
    - 多个 Worker agent → Queue → Leader agent 收集结果
    """)


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