Function process_stream_events

tutorials/10_sse_streaming.py:414–519

将 StreamEvent 列表转换为 AssistantEvent 列表。 这个函数模拟了 Claude Code 中从 API 流式事件 到 Agentic Loop 可用事件的转换过程。 对应源码: - api/client.rs:538-562 (MessageStream::next_event) - conversation.rs:353-390 (build_assistant_message)

(stream_events: list[StreamEvent])

def process_stream_events(stream_events: list[StreamEvent]) -> list[AssistantEvent]:
    """
    将 StreamEvent 列表转换为 AssistantEvent 列表。

    这个函数模拟了 Claude Code 中从 API 流式事件
    到 Agentic Loop 可用事件的转换过程。

    对应源码:
      - api/client.rs:538-562 (MessageStream::next_event)
      - conversation.rs:353-390 (build_assistant_message)
    """
    assistant_events = []
    # 正在构建中的内容块 (key = index)
    building_blocks: dict[int, ContentBlockState] = {}

    for event in stream_events:
        event_type = event.event_type

        if event_type == "message_start":
            # 消息开始，通常包含元信息（model, id 等）
            # 我们这里不需要特别处理
            pass

        elif event_type == "content_block_start":
            # 一个新的内容块开始了
            index = event.data["index"]
            block = event.data["content_block"]
            block_type = block["type"]  # "text" or "tool_use"

            state = ContentBlockState(
                index=index,
                block_type=block_type,
            )

            if block_type == "tool_use":
                state.tool_id = block.get("id", "")
                state.tool_name = block.get("name", "")

            if block_type == "text" and block.get("text"):
                # 有些 content_block_start 自带初始文本
                state.text = block["text"]
                assistant_events.append(AssistantEvent(
                    event_type=AssistantEventType.TEXT_DELTA,
                    text=block["text"],
                ))

            building_blocks[index] = state

        elif event_type == "content_block_delta":
            # 增量更新
            index = event.data["index"]
            delta = event.data["delta"]
            delta_type = delta["type"]

            state = building_blocks.get(index)
            if state is None:
                continue