Chapter 5 · Event Stream Subscription and Streaming Output

Chapter goals: Master agent.subscribe() and all event types, understand the message_update / text_delta incremental mechanism, and build typewriter-style real-time output.

5.1 Why Event Streams

An LLM generating a long response may take ten-plus seconds. Rather than leaving the user staring at a blank screen, push the generation process out in real time — that is the purpose of event streams. Agent models everything that happens within a round as events:

// Subscribe: returns an unsubscribe function
const unsubscribe = agent.subscribe((event) => {
  console.log(event.type);
});

// Remember to unsubscribe when done (component unmount / session end)
unsubscribe();

Listeners are awaited in registration order — async handling of agent_end also counts toward round settlement, providing a reliable barrier for "wrap-up work."

5.2 Complete Event Sequence of prompt()

A prompt("Hello") with no tool calls emits in sequence:

prompt("Hello")
├─ agent_start                      # Round begins
├─ turn_start                       # One turn = one LLM call + tool execution
├─ message_start { userMessage }    # User message queued
├─ message_end   { userMessage }
├─ message_start { assistantMessage }
├─ message_update { partial... }    # ★ Streaming increments, possibly hundreds or thousands
├─ message_update { partial... }
├─ message_end   { assistantMessage } # Complete response
├─ turn_end      { message, toolResults: [] }
└─ agent_end      { messages: [...] } # Final event of the entire run

With tool calls, tool_execution_start / update / end are interspersed between turns (detailed in Chapters 6 and 7).

5.3 message_update and text_delta

Note the nested structure: message_update is an Agent-layer event; it carries assistantMessageEvent (the raw stream event from the pi-ai layer) inside:

// Standard pattern for a typewriter effect
agent.subscribe((event) => {
  if (event.type === "message_update") {
    const inner = event.assistantMessageEvent;
    switch (inner.type) {
      case "text_delta":
        // Output only the newly added text fragment
        process.stdout.write(inner.delta);
        break;
      case "thinking_delta":
        // Thinking process increment from reasoning-capable models
        process.stdout.write(`\x1b[2m${inner.delta}\x1b[0m`);
        break;
    }
  }
});

5.4 Complete Event Type Table

Event	Description	Key Properties
agent_start	Agent begins processing this run	—
agent_end	Final event of the run; no loop events follow	messages: [...]
turn_start	New turn begins (1 LLM call + tool execution)	—
turn_end	Turn ends	message, toolResults
message_start	Any message begins (user/assistant/toolResult)	message
message_update	Assistant messages only, carries incremental events	assistantMessageEvent
message_end	Message completes	message
tool_execution_start	Tool begins executing	toolCallId, toolName, args
tool_execution_update	Tool streaming progress (when the tool reports progress)	Partial results
tool_execution_end	Tool execution completes	toolCallId, result

5.5 In Practice: A Structured Terminal UI

// Combine multiple events to build readable terminal output
agent.subscribe((event) => {
  switch (event.type) {
    case "turn_start":
      process.stdout.write("\n🤖 ");
      break;
    case "tool_execution_start":
      // Clear hint when a tool starts
      console.log(`\n🔧 Calling tool ${event.toolName} args=${JSON.stringify(event.args)}`);
      break;
    case "tool_execution_end":
      console.log(`✅ Tool done (${event.toolCallId})`);
      break;
    case "message_update": {
      const inner = event.assistantMessageEvent;
      if (inner.type === "text_delta") process.stdout.write(inner.delta);
      break;
    }
    case "agent_end":
      console.log(`\n\n📊 This round had ${event.messages.length} messages total`);
      break;
  }
});

await agent.prompt("Search for Node.js 22 new features and summarize");

5.6 Chapter Summary

• subscribe() awaits listeners in registration order; the return value is used to unsubscribe;
• Event hierarchy: agent → turn → message → tool execution;
• message_update.assistantMessageEvent.text_delta is the key to implementing a typewriter effect;
• agent_end is the run barrier event, suitable for persistence and other wrap-up work.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. To implement a typewriter effect, which event should you listen to and which field should you read?

2. Regarding the agent_end event, which statement is correct?

3. Which event is emitted when a tool begins executing?

4. What does the return value of `agent.subscribe()` represent?

🛠️ Hands-on Practice

1. Add a complete colored terminal UI to the Agent from Chapter 4: thinking in gray italics, body text in white, tool calls with yellow prompts.
2. Count the occurrences of each event type in a round with tool calls and verify the event sequence diagram.
3. Register two subscribers (one printing, one counting) and verify they both receive events in registration order.

Next Chapter 6 enters the most core topic of this course: tool definition and execution.