Function build_assistant_message

mini_claude_code/runtime.py:164–206

源码: conversation.rs:353-390 返回: (assistant_message, optional_usage)

(
    events: list[AssistantEvent],
)

def build_assistant_message(
    events: list[AssistantEvent],
) -> tuple[Message, Optional[TokenUsage]]:
    """源码: conversation.rs:353-390

    返回: (assistant_message, optional_usage)
    """
    text_buffer: str = ""
    blocks: list = []
    finished: bool = False
    usage: Optional[TokenUsage] = None

    for event in events:
        if isinstance(event, TextDeltaEvent):
            text_buffer += event.text

        elif isinstance(event, ToolUseEvent):
            # flush 文本缓冲 — 源码: conversation.rs:392-398
            if text_buffer:
                blocks.append(TextContentBlock(text=text_buffer))
                text_buffer = ""
            blocks.append(ToolContentBlock(
                id=event.id,
                name=event.name,
                input=event.input,
            ))

        elif isinstance(event, MessageStopEvent):
            finished = True

    # 最后 flush 残留文本
    if text_buffer:
        blocks.append(TextContentBlock(text=text_buffer))

    if not finished:
        raise RuntimeError(
            "assistant stream ended without a message stop event"
        )
    if not blocks:
        raise RuntimeError("assistant stream produced no content")

    message = Message(role="assistant", content=blocks)
    return message, usage


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