streaming

package
v1.5.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 16, 2026 License: Apache-2.0 Imports: 12 Imported by: 2

Documentation

Overview

Package streaming provides functionality for accumulating streaming chunks and other chunk-related workflows

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AccumulatedData 

type AccumulatedData struct {
	RequestID             string
	Model                 string
	Status                string
	Stream                bool
	Latency               int64 // in milliseconds
	TimeToFirstToken      int64 // Time to first token in milliseconds (streaming only)
	StartTimestamp        time.Time
	EndTimestamp          time.Time
	OutputMessage         *schemas.ChatMessage
	OutputMessages        []schemas.ResponsesMessage // For responses API
	ToolCalls             []schemas.ChatAssistantMessageToolCall
	ErrorDetails          *schemas.BifrostError
	TokenUsage            *schemas.BifrostLLMUsage
	CacheDebug            *schemas.BifrostCacheDebug
	Cost                  *float64
	AudioOutput           *schemas.BifrostSpeechResponse
	TranscriptionOutput   *schemas.BifrostTranscriptionResponse
	ImageGenerationOutput *schemas.BifrostImageGenerationResponse
	PassthroughOutput     *schemas.BifrostPassthroughResponse // For passthrough streaming
	FinishReason          *string
	LogProbs              *schemas.BifrostLogProbs
	RawResponse           *string
}

AccumulatedData contains the accumulated data for a stream

type Accumulator 

type Accumulator struct {
	// contains filtered or unexported fields
}

Accumulator manages accumulation of streaming chunks

func NewAccumulator 

func NewAccumulator(pricingManager *modelcatalog.ModelCatalog, logger schemas.Logger) *Accumulator

NewAccumulator creates a new accumulator

func (*Accumulator) Cleanup 

func (a *Accumulator) Cleanup()

Cleanup cleans up the accumulator

func (*Accumulator) CleanupStreamAccumulator added in v1.1.2 

func (a *Accumulator) CleanupStreamAccumulator(requestID string) error

CleanupStreamAccumulator decrements the reference counter for a stream accumulator. The accumulator is only cleaned up when the reference counter reaches 0. This function is idempotent - calling it after cleanup has already happened is safe.

func (*Accumulator) ClearPausedStreamBuffer added in v1.4.4 

func (a *Accumulator) ClearPausedStreamBuffer(traceID string) error

ClearPausedStreamBuffer drops chunks buffered while a stream is paused.

func (*Accumulator) CreateStreamAccumulator 

func (a *Accumulator) CreateStreamAccumulator(requestID string, startTimestamp time.Time) *StreamAccumulator

CreateStreamAccumulator creates a new stream accumulator for a request It increments the reference counter atomically for concurrent access tracking

func (*Accumulator) EndStream added in v1.4.0 

func (a *Accumulator) EndStream(traceID string, err *schemas.BifrostError)

EndStream is the Tracer-level entry point for terminating a stream. Any buffered chunks are flushed first; then if err is non-nil it is delivered as a terminal error chunk. After EndStream, further provider chunks are dropped.

func (*Accumulator) ForceCleanupStreamAccumulator added in v1.4.0 

func (a *Accumulator) ForceCleanupStreamAccumulator(requestID string)

ForceCleanupStreamAccumulator reaps a stream accumulator regardless of its reference counter. It is the guaranteed end-of-stream backstop: callers invoke it from the stream's terminal lifecycle hook (the provider goroutine's finalizer), at which point the stream has stopped delivering chunks and the per-plugin refcount handshake may be incomplete (e.g. a client abort that never produced a terminal chunk, or multiple plugins that each Create but not all Cleanup). Force-reaping here mirrors the TTL sweep (cleanupOldAccumulators), which already deletes with forceEndGate=true. Idempotent and safe to call after CleanupStreamAccumulator has already freed the entry.

func (*Accumulator) GateSend added in v1.4.0 

func (a *Accumulator) GateSend(traceID string, chunk *schemas.BifrostStreamChunk, isFinal, isHardErr bool, ch chan *schemas.BifrostStreamChunk, ctx *schemas.BifrostContext) bool

GateSend is the Tracer-level entry point for delivering a stream chunk through the pause/resume/end gate. See Tracer.GateSend in core/schemas for behavior. Returns true if the chunk was handled (delivered or buffered), false if the caller should stop sending.

func (*Accumulator) GetAccumulatedResponse added in v1.4.0 

func (a *Accumulator) GetAccumulatedResponse(traceID string) *schemas.BifrostResponse

GetAccumulatedResponse returns a snapshot *schemas.BifrostResponse built from chunks accumulated so far for traceID. Built on demand each call; no caching. Detects stream type by which per-type chunk slice is populated. Returns nil if:

  • traceID is empty
  • no accumulator exists for traceID (never started or already cleaned up)
  • no chunks have been accumulated yet
  • the populated slices are ambiguous (more than one type — should not happen in normal flow, but defensive)
  • the per-type build returns no data

Note: ExtraFields.Provider / OriginalModelRequested / ResolvedModelUsed are not preserved on the StreamAccumulator (they're per-chunk, not per-stream), so the returned response will have empty values for those fields. The body of the response (Choices/Message/etc.) is fully populated.

func (*Accumulator) IsStreamEnded added in v1.4.0 

func (a *Accumulator) IsStreamEnded(traceID string) bool

IsStreamEnded reports whether the gate for traceID is in the Ended state. Read-only: does NOT create an accumulator if one doesn't exist.

func (*Accumulator) IsStreamPaused added in v1.4.0 

func (a *Accumulator) IsStreamPaused(traceID string) bool

IsStreamPaused reports whether the gate for traceID is currently Paused. Read-only: does NOT create an accumulator if one doesn't exist.

func (*Accumulator) PauseStream added in v1.4.0 

func (a *Accumulator) PauseStream(traceID string)

PauseStream is the Tracer-level entry point for pausing a stream. Forwards to the per-stream accumulator entry keyed by traceID.

func (*Accumulator) ProcessStreamingResponse 

func (a *Accumulator) ProcessStreamingResponse(ctx *schemas.BifrostContext, result *schemas.BifrostResponse, bifrostErr *schemas.BifrostError) (*ProcessedStreamResponse, error)

ProcessStreamingResponse processes a streaming response It handles chat, audio, and responses streaming responses

func (*Accumulator) ResumeStream added in v1.4.0 

func (a *Accumulator) ResumeStream(traceID string)

ResumeStream is the Tracer-level entry point for resuming a paused stream.

func (*Accumulator) WaitForFlusher added in v1.4.0 

func (a *Accumulator) WaitForFlusher(traceID string)

WaitForFlusher is the Tracer-level entry point for blocking until the gate flusher for traceID has fully drained and exited. Read-only: does NOT create an accumulator if one doesn't exist (no flusher = nothing to wait for).

type AudioStreamChunk 

type AudioStreamChunk struct {
	Timestamp          time.Time                            // When chunk was received
	Delta              *schemas.BifrostSpeechStreamResponse // The actual delta content
	FinishReason       *string                              // If this is the final chunk
	TokenUsage         *schemas.SpeechUsage                 // Token usage if available
	SemanticCacheDebug *schemas.BifrostCacheDebug           // Semantic cache debug if available
	Cost               *float64                             // Cost in dollars from pricing plugin
	ErrorDetails       *schemas.BifrostError                // Error if any
	ChunkIndex         int                                  // Index of the chunk in the stream
	RawResponse        *string
}

AudioStreamChunk represents a single streaming chunk

type ChatStreamChunk 

type ChatStreamChunk struct {
	Timestamp          time.Time                              // When chunk was received
	Delta              *schemas.ChatStreamResponseChoiceDelta // The actual delta content
	FinishReason       *string                                // If this is the final chunk
	LogProbs           *schemas.BifrostLogProbs               // LogProbs if available
	TokenUsage         *schemas.BifrostLLMUsage               // Token usage if available
	SemanticCacheDebug *schemas.BifrostCacheDebug             // Semantic cache debug if available
	Cost               *float64                               // Cost in dollars from pricing plugin
	ErrorDetails       *schemas.BifrostError                  // Error if any
	ChunkIndex         int                                    // Index of the chunk in the stream
	RawResponse        *string                                // Raw response if available
}

ChatStreamChunk represents a single streaming chunk

type ImageStreamChunk added in v1.2.9 

type ImageStreamChunk struct {
	Timestamp          time.Time                                     // When chunk was received
	Delta              *schemas.BifrostImageGenerationStreamResponse // The actual stream response
	FinishReason       *string                                       // If this is the final chunk
	ChunkIndex         int                                           // Index of the chunk in the stream
	ImageIndex         int                                           // Index of the image in the stream
	ErrorDetails       *schemas.BifrostError                         // Error if any
	Cost               *float64                                      // Cost in dollars from pricing plugin
	SemanticCacheDebug *schemas.BifrostCacheDebug                    // Semantic cache debug if available
	TokenUsage         *schemas.ImageUsage                           // Token usage if available
	RawResponse        *string                                       // Raw response if available
}

ImageStreamChunk represents a single image streaming chunk

type ProcessedStreamResponse 

type ProcessedStreamResponse struct {
	RequestID      string
	StreamType     StreamType
	Provider       schemas.ModelProvider
	RequestedModel string // original model requested by the caller
	ResolvedModel  string // actual model used by the provider (equals RequestedModel when no alias mapping exists)
	RoutingInfo    schemas.RoutingInfo
	Data           *AccumulatedData
	RawRequest     *interface{}
}

ProcessedStreamResponse represents a processed streaming response

func (*ProcessedStreamResponse) ToBifrostResponse 

func (p *ProcessedStreamResponse) ToBifrostResponse() *schemas.BifrostResponse

ToBifrostResponse converts a ProcessedStreamResponse to a BifrostResponse

type ResponsesStreamChunk added in v1.1.6 

type ResponsesStreamChunk struct {
	Timestamp          time.Time                               // When chunk was received
	StreamResponse     *schemas.BifrostResponsesStreamResponse // The actual stream response
	FinishReason       *string                                 // If this is the final chunk
	TokenUsage         *schemas.BifrostLLMUsage                // Token usage if available
	SemanticCacheDebug *schemas.BifrostCacheDebug              // Semantic cache debug if available
	Cost               *float64                                // Cost in dollars from pricing plugin
	ErrorDetails       *schemas.BifrostError                   // Error if any
	ChunkIndex         int                                     // Index of the chunk in the stream
	RawResponse        *string
}

ResponsesStreamChunk represents a single responses streaming chunk

type StreamAccumulator 

type StreamAccumulator struct {
	RequestID           string
	StartTimestamp      time.Time
	FirstChunkTimestamp time.Time // Timestamp when the first chunk was received (for TTFT calculation)
	ChatStreamChunks    []*ChatStreamChunk

	ResponsesStreamChunks     []*ResponsesStreamChunk
	TranscriptionStreamChunks []*TranscriptionStreamChunk
	AudioStreamChunks         []*AudioStreamChunk
	ImageStreamChunks         []*ImageStreamChunk

	// De-dup maps to prevent chunk loss on out-of-order arrival
	ChatChunksSeen          map[int]struct{}
	ResponsesChunksSeen     map[int]struct{}
	TranscriptionChunksSeen map[int]struct{}
	AudioChunksSeen         map[int]struct{}
	ImageChunksSeen         map[string]struct{} // Composite key: "imageIndex:chunkIndex" to scope de-dup per image

	// Track highest ChunkIndex for metadata extraction (TokenUsage, Cost, FinishReason)
	MaxChatChunkIndex          int
	MaxResponsesChunkIndex     int
	MaxTranscriptionChunkIndex int
	MaxAudioChunkIndex         int

	// TerminalErrorChunkIndex holds the reserved chunk index for the terminal error (-1 = unset); reused across plugin calls for correct dedup.
	TerminalErrorChunkIndex int
	// TerminalResponseChunkIndex holds the reserved index for a terminal response event when providers omit or reuse sequence numbers.
	TerminalResponseChunkIndex int

	// Passthrough streaming accumulation
	PassthroughBody       []byte            // Accumulated body bytes from passthrough streaming chunks
	PassthroughStatusCode int               // Status code from passthrough response
	PassthroughHeaders    map[string]string // Headers from passthrough response
	PassthroughPath       string            // Stripped provider path, e.g. "/v1/chat/completions"

	IsComplete     bool
	FinalTimestamp time.Time

	Timestamp time.Time
	// contains filtered or unexported fields
}

StreamAccumulator manages accumulation of streaming chunks

func (*StreamAccumulator) ClearPausedBuffer added in v1.4.4 

func (sa *StreamAccumulator) ClearPausedBuffer() error

ClearPausedBuffer removes replay chunks captured while the gate is paused. If a terminal chunk was among the dropped chunks, delivering a replacement terminal becomes the caller's responsibility.

func (*StreamAccumulator) End added in v1.4.0 

func (sa *StreamAccumulator) End(err *schemas.BifrostError)

End transitions the gate to Ended. Any buffered chunks are flushed by the flusher (if running) before exit; if err is non-nil it is delivered as a terminal error chunk after the flush. Idempotent.

If no flusher is running but the gate has a cached delivery target (i.e. a prior GateSend captured ch+ctx) and there is work to do — buffered chunks or a synthetic terminal error from err — a flusher is started so the terminal chunk reaches the client. When ch+ctx were never cached (no chunks ever sent), the error is dropped: there is no consumer to deliver it to.

func (*StreamAccumulator) GateSend added in v1.4.0 

func (sa *StreamAccumulator) GateSend(chunk *schemas.BifrostStreamChunk, isFinal, isHardErr bool, ch chan *schemas.BifrostStreamChunk, ctx *schemas.BifrostContext) bool

GateSend implements the per-chunk delivery gate.

  • Active state: chunk is forwarded to ch (with ctx.Done() guard).
  • Paused state: chunk is buffered for later replay; flusher started lazily.
  • Ended state: chunk is dropped.

Final chunks (isFinal) and hard provider errors (isHardErr) bypass Active and force-flush + transition to Ended; if a flusher is running or chunks are buffered, the final chunk is appended to the buffer so the flusher delivers it in order.

func (*StreamAccumulator) Pause added in v1.4.0 

func (sa *StreamAccumulator) Pause()

Pause transitions the gate from Active to Paused. Idempotent.

func (*StreamAccumulator) Resume added in v1.4.0 

func (sa *StreamAccumulator) Resume()

Resume transitions the gate from Paused back to Active and wakes the flusher to drain buffered chunks. Idempotent.

func (*StreamAccumulator) WaitForFlusher added in v1.4.0 

func (sa *StreamAccumulator) WaitForFlusher()

WaitForFlusher blocks until the currently-running flusher goroutine, if any, has fully drained and exited. Returns immediately if no flusher is active. Useful before closing the response channel so the gate can finalize ordered delivery without racing the producer's close.

type StreamState added in v1.4.0 

type StreamState int8

StreamState represents the delivery state of a streaming response. 

const (
	// StreamStateActive: chunks are forwarded to the client as they arrive.
	StreamStateActive StreamState = iota
	// StreamStatePaused: chunks are buffered for later replay; not delivered.
	StreamStatePaused
	// StreamStateEnded: gate is closed; further chunks are dropped.
	StreamStateEnded
)

type StreamType 

type StreamType string
const (
	StreamTypeText          StreamType = "text.completion"
	StreamTypeChat          StreamType = "chat.completion"
	StreamTypeAudio         StreamType = "audio.speech"
	StreamTypeImage         StreamType = "image.generation"
	StreamTypeTranscription StreamType = "audio.transcription"
	StreamTypeResponses     StreamType = "responses"
	StreamTypePassthrough   StreamType = "passthrough"
)

type TranscriptionStreamChunk 

type TranscriptionStreamChunk struct {
	Timestamp          time.Time                                   // When chunk was received
	Delta              *schemas.BifrostTranscriptionStreamResponse // The actual delta content
	FinishReason       *string                                     // If this is the final chunk
	TokenUsage         *schemas.TranscriptionUsage                 // Token usage if available
	SemanticCacheDebug *schemas.BifrostCacheDebug                  // Semantic cache debug if available
	Cost               *float64                                    // Cost in dollars from pricing plugin
	ErrorDetails       *schemas.BifrostError                       // Error if any
	ChunkIndex         int                                         // Index of the chunk in the stream
	RawResponse        *string
}

TranscriptionStreamChunk represents a single transcription streaming chunk

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.