streaming

func StreamFromContext ¶

func StreamFromContext(ctx context.Context, baseStream types.ChatCompletionStream) types.ChatCompletionStream

StreamFromContext creates a context-aware stream from a base stream with cancellation support.

This function wraps a ChatCompletionStream with context awareness, enabling proper SSE (Server-Sent Events) cancellation when the context is canceled or times out. It is critical for external gateways that need to manage streaming lifecycle and forward error events to clients.

The wrapped stream monitors the context and returns properly formatted error chunks that can be forwarded as SSE error events. The error categorization includes:

• context.DeadlineExceeded -> timeout errors
• context.Canceled -> client cancellation errors
• Other errors -> categorized by type (network, API, rate limit, etc.)

Parameters:

• ctx: The context to monitor for cancellation. Use context.WithTimeout or context.WithCancel to enable cancellation.
• baseStream: The underlying ChatCompletionStream to wrap.

Returns:

• A ChatCompletionStream that respects context cancellation and returns error chunks formatted for SSE forwarding.

Example:

// Create a stream with 30 second timeout
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

// Get stream from provider
baseStream := provider.ChatCompletion(ctx, request)

// Wrap with context for SSE cancellation support
stream := streaming.StreamFromContext(ctx, baseStream)

// Stream responses with context awareness
for {
    chunk, err := stream.Next()
    if err != nil {
        // Error chunk contains SSE-formatted error for client forwarding
        if chunk.Error != "" {
            // Forward to client as SSE error event
            sseError := fmt.Sprintf("event: error\ndata: %s\n\n", chunk.Error)
            fmt.Fprint(w, sseError)
        }
        break
    }

    // Forward normal chunk to client
    forwardChunkToClient(w, chunk)

    if chunk.Done {
        break
    }
}