Event Handling: Real-time Capture and Production Consumer Architecture

Streaming turns an agent’s output into an event stream. Event handling is the layer that decides what each event means, how it is serialized, how consumers process it, and how the system survives slow clients, out-of-order delivery, and transport failures.

← Streaming Principles | Multimodal →

The Architectural Value of Streaming

Streaming is valuable because it turns a long-running generation into observable steps. In a synchronous call, the user only sees the final result. In a streaming system, the UI can display text immediately, tools can show progress, logs can record each step, and observability systems can measure latency.

Dimension	Synchronous Call	Streaming
Time to first token	Usually seconds	Often under 100 ms
User experience	Waiting and uncertainty	Immediate feedback
Interactivity	Limited	Stronger; easier to interrupt or control
Resource use	Blocking wait	Incremental consumption
Timeout risk	Higher for long outputs	Lower

The important insight is that streaming is not just a frontend animation. It is a data-flow model that affects logging, retries, tool execution, analytics, and recovery.

Streaming Data Flow

A complete streaming flow usually looks like this:

User input
    |
    v
+-------------+  Event Stream  +--------------+
| LLM / Agent | --------------> | Consumer     |
|  generating |  TextDelta     |  pipeline    |
|             |  ToolCall      |              |
|             |  ToolResult    |              |
|             |  Error/Done    |              |
+-------------+                +------+-------+
                                     |
                +--------------------+--------------------+
                v                    v                    v
        +--------------+     +-------------+     +-------------+
        | UI rendering |     | Logging     |     | State/metrics|
        +--------------+     +-------------+     +-------------+

Each arrow in that diagram is a place where latency, buffering, or failure can occur. A good event layer makes those places explicit.

Event Type System

An event type system is the vocabulary of streaming. Common event types include text deltas, tool calls, tool results, thinking traces, errors, completion, cancellation, and heartbeats.

type EventType int

const (
    EventTypeTextDelta   EventType = iota
    EventTypeToolCall
    EventTypeToolResult
    EventTypeFunctionCall
    EventTypeFunctionResult
    EventTypeThinking
    EventTypeError
    EventTypeDone
    EventTypeCancelled
    EventTypeHeartbeat
)

type Event interface {
    Type() EventType
    Timestamp() time.Time
    Sequence() int64
    Metadata() map[string]interface{}
}

type BaseEvent struct {
    eventType EventType
    timestamp time.Time
    sequence  int64
    metadata  map[string]interface{}
}

func (e *BaseEvent) Type() EventType { return e.eventType }
func (e *BaseEvent) Timestamp() time.Time { return e.timestamp }
func (e *BaseEvent) Sequence() int64 { return e.sequence }
func (e *BaseEvent) Metadata() map[string]interface{} { return e.metadata }
}

A concrete event usually embeds BaseEvent and adds typed payload fields. For example:

type TextDeltaEvent struct {
    BaseEvent
    Text    string
    Index   int
    IsFinal bool
}

type ToolCallEvent struct {
    BaseEvent
    ToolName string
    ToolID   string
    Args     map[string]interface{}
    RawArgs  string
}

type DoneEvent struct {
    BaseEvent
    FinalText    string
    TokenUsage   int
    FinishReason string
}

Serialization Protocol

Events must be serialized consistently so logs, clients, and replay systems can understand them. JSON is easy and readable. Protocol Buffers or another compact format can help when throughput or payload size becomes important.

type EventWrapper struct {
    Type      string                 `json:"type"`
    Timestamp int64                  `json:"timestamp"`
    Sequence  int64                  `json:"sequence"`
    Metadata  map[string]interface{} `json:"metadata,omitempty"`
    Payload   map[string]interface{} `json:"payload"`
}

func serializeJSON(event Event) ([]byte, error) {
    wrapper := EventWrapper{
        Type:      event.Type().String(),
        Timestamp: event.Timestamp().UnixMilli(),
        Sequence:  event.Sequence(),
        Metadata:  event.Metadata(),
        Payload:   map[string]interface{}{},
    }

    switch e := event.(type) {
    case *TextDeltaEvent:
        wrapper.Payload["text"] = e.Text
        wrapper.Payload["index"] = e.Index
        wrapper.Payload["is_final"] = e.IsFinal
    case *ToolCallEvent:
        wrapper.Payload["tool_name"] = e.ToolName
        wrapper.Payload["tool_id"] = e.ToolID
        wrapper.Payload["args"] = e.Args
    }

    return json.Marshal(wrapper)
}

The serialization contract should be stable. If clients depend on field names or sequence semantics, breaking changes need a versioned migration plan.

Consumer Architecture

A consumer turns raw events into visible work: UI updates, logs, metrics, persistence, or downstream integrations. In production, use a consumer chain so responsibilities stay separated.

type EventConsumer interface {
    Consume(ctx context.Context, event Event) error
    Close() error
}

type ConsumerChain struct {
    consumers []EventConsumer
}

func (c *ConsumerChain) Consume(ctx context.Context, event Event) error {
    for _, consumer := range c.consumers {
        if err := consumer.Consume(ctx, event); err != nil {
            return err
        }
    }
    return nil
}

A filtering consumer can route only relevant events to a handler. For example, a UI renderer may only care about TextDelta and Done, while an observability consumer may care about every event.

type FilteringConsumer struct {
    filter   func(Event) bool
    consumer EventConsumer
}

func (c *FilteringConsumer) Consume(ctx context.Context, event Event) error {
    if !c.filter(event) {
        return nil
    }
    return c.consumer.Consume(ctx, event)
}

func NonHeartbeatFilter(event Event) bool {
    return event.Type() != EventTypeHeartbeat
}

Backpressure Control

Backpressure happens when producers emit faster than consumers can process. In streaming, producers often include the model API, tool executors, and serialization loops. Consumers include the browser, metrics pipeline, log writer, or database. If one side stalls, the system must decide whether to wait, buffer, or drop.

type BackpressureConsumer struct {
    consumer     EventConsumer
    buffer       chan Event
    dropStrategy DropStrategy
    metrics      *ConsumerMetrics
}

type DropStrategy int

const (
    DropStrategyReject DropStrategy = iota
    DropStrategyOldest
    DropStrategyLatest
    DropStrategySample
)

func (c *BackpressureConsumer) Consume(ctx context.Context, event Event) error {
    select {
    case c.buffer <- event:
        return nil
    default:
        c.metrics.DropCounter.Inc()
        switch c.dropStrategy {
        case DropStrategyReject:
            return fmt.Errorf("buffer full")
        case DropStrategyLatest:
            return nil
        case DropStrategyOldest:
            select {
            case <-c.buffer:
                c.buffer <- event
                return nil
            default:
                return fmt.Errorf("buffer empty after drop attempt")
            }
        default:
            return fmt.Errorf("unknown drop strategy")
        }
    }
}

Choose the drop strategy by event importance. Dropping a heartbeat may be fine; dropping a final Done event is usually not.

Ordered Consumption

Sequences keep the system understandable when events arrive late or repeat. A robust ordered consumer tracks the next expected sequence and buffers future events only within a bounded size.

type OrderedConsumer struct {
    consumer  EventConsumer
    buffer    map[int64]Event
    nextSeq   int64
    maxBuffer int
    timeout   time.Duration
    mu        sync.Mutex
}

func (c *OrderedConsumer) Consume(ctx context.Context, event Event) error {
    c.mu.Lock()
    defer c.mu.Unlock()

    seq := event.Sequence()
    if seq < c.nextSeq {
        return nil
    }
    if seq == c.nextSeq {
        if err := c.consumer.Consume(ctx, event); err != nil {
            return err
        }
        c.nextSeq++
        return c.consumeBuffered(ctx)
    }

    if len(c.buffer) >= c.maxBuffer {
        return fmt.Errorf("ordered consumer buffer full")
    }
    c.buffer[seq] = event
    return nil
}

func (c *OrderedConsumer) consumeBuffered(ctx context.Context) error {
    for {
        if event, ok := c.buffer[c.nextSeq]; ok {
            if err := c.consumer.Consume(ctx, event); err != nil {
                return err
            }
            delete(c.buffer, c.nextSeq)
            c.nextSeq++
        } else {
            return nil
        }
    }
}

Ordered delivery prevents duplicated text, broken tool-call sequences, and analytics inconsistencies.

Complete Chat Backend Example

A compact chat backend combines the event stream, consumer chain, SSE response, and lifecycle control.

func handleChat(w http.ResponseWriter, r *http.Request) {
    var req ChatRequest
    if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }

    w.Header().Set("Content-Type", "text/event-stream")
    w.Header().Set("Cache-Control", "no-cache")
    w.Header().Set("X-Accel-Buffering", "no")

    flusher, ok := w.(http.Flusher)
    if !ok {
        http.Error(w, "streaming unsupported", http.StatusInternalServerError)
        return
    }

    ctx, cancel := context.WithCancel(r.Context())
    defer cancel()

    stream, err := agent.RunStream(ctx, req.Message)
    if err != nil {
        writeSSE(w, "error", map[string]string{"error": err.Error()})
        return
    }
    defer stream.Close()

    chain := NewConsumerChain(&LoggingConsumer{}, &MetricsConsumer{})
    for {
        event, err := stream.Recv()
        if err == io.EOF {
            writeSSE(w, "done", map[string]string{"status": "completed"})
            flusher.Flush()
            return
        }
        if err != nil {
            writeSSE(w, "error", map[string]string{"error": err.Error()})
            flusher.Flush()
            return
        }

        if err := chain.Consume(ctx, event); err != nil {
            log.Printf("consumer error: %v", err)
        }
        if err := writeStreamEvent(w, event); err != nil {
            return
        }
        flusher.Flush()
    }
}

Error Recovery

A resilient system needs three layers: client acknowledgements, server-side buffers, and durable persistence for important events. Sequence numbers make all three practical.

If a connection drops, the backend can record the last emitted sequence, the client can request replay from that point, and the server can re-send missing events instead of forcing the user to restart. For critical systems, important events can also be persisted to a queue or database with at-least-once delivery semantics.

type ResumableStream struct {
    lastSeq     int64
    reconnects  int
    maxAttempts int
    backoff     time.Duration
}

func (s *ResumableStream) Start(ctx context.Context, handler EventHandler) error {
    for {
        stream, err := agent.RunStream(ctx, prompt, agent.WithResumeFrom(s.lastSeq))
        if err != nil {
            if s.reconnects >= s.maxAttempts {
                return fmt.Errorf("max reconnects exceeded: %w", err)
            }
            time.Sleep(s.backoff * time.Duration(1<<s.reconnects))
            s.reconnects++
            continue
        }
        s.reconnects = 0

        for {
            event, err := stream.Recv()
            if err == io.EOF {
                return nil
            }
            if err != nil {
                break
            }
            s.lastSeq = event.Sequence()
            if err := handler.Handle(ctx, event); err != nil {
                return err
            }
        }
    }
}

Performance Optimization

Streaming has the same token generation cost as non-streaming generation, but it adds per-event overhead. Each event is serialized, transmitted, rendered, and often logged.

Optimize by batching small TextDelta events when the UI can tolerate it, reusing LLM provider connections, keeping buffers bounded, disabling unnecessary consumers under load, and exposing metrics such as first-token latency, event throughput, dropped events, and consumer latency.

Common Questions

Q: What should be done if an event is lost?

Use acknowledgements for important events, maintain a replay buffer on the server, and persist critical events to a durable queue when strong reliability is required.

Q: How should high concurrency be handled?

Pool provider connections, limit active streams, use backpressure, selectively disable non-critical consumers, and scale backend workers behind a load balancer.

Q: Does streaming consume more tokens?

The generated token count is the same, but streaming adds network, serialization, and rendering overhead. Batch small text deltas when latency budget allows.

Next Step

Event handling now has a production shape: typed events, stable serialization, consumer chains, backpressure, ordering, and recovery. The next tutorial covers multimodal streaming: audio, images, and video without losing real-time responsiveness.

Want to keep learning hands-on Go ADK? Follow the “Mengshou Programming” channel for weekly Go and AI programming posts.

Frequently Asked Questions

What are the core event types in streaming?

Common types include TextDelta, ToolCall, ToolResult, Thinking, Error, Done, and heartbeat events.

What should happen if an event is lost?

Use acknowledgements, server-side replay buffers, sequence numbers, and durable queues for at-least-once delivery where needed.

How should high concurrency be handled?

Use connection pooling, rate limits, selective degradation, backpressure, and horizontal scaling of backend workers.

Does streaming consume more tokens?

Generation token count is the same, but each event has network and serialization overhead. Batch small text deltas when possible.

How is ordered consumption guaranteed?

Assign monotonically increasing sequence numbers and use buffering or retry logic for out-of-order messages.

The Architectural Value of Streaming#

Streaming Data Flow#

Event Type System#

Serialization Protocol#

Consumer Architecture#

Backpressure Control#

Ordered Consumption#

Complete Chat Backend Example#

Error Recovery#

Performance Optimization#

Common Questions#

Next Step#