stage

package

Go to main page

Versions in this module

v1

v1.5.5

Jul 11, 2026

v1.5.4

Jul 2, 2026

Changes in this version

type DuplexProviderStage

+ func (s *DuplexProviderStage) SetSessionObserver(fn func(providers.StreamInputSession))

type IncrementalSaveConfig

+ PersistReasoning bool

+ type ReasoningDelta struct

+ Opaque []types.OpaqueReasoning

+ Text string

type StreamElement

+ Reasoning *ReasoningDelta

v1.5.3

Jun 26, 2026

Changes in this version

type AudioTurnConfig

+ EmitEndOfTurn bool

type ProviderConfig

+ Streaming bool

type StreamElement

+ EndOfTurn bool

+ Interrupt bool

+ func NewEndOfTurnElement() StreamElement

+ func NewInterruptElement() StreamElement

v1.5.2

Jun 22, 2026

v1.5.1

Jun 17, 2026

v1.5.0

Jun 15, 2026

Changes in this version

+ func NewCompositionStepExecutor(deps CompositionExecutorDeps) engine.StepExecutor

+ func NewFileSchemaResolver(configDir string) func(path string) (json.RawMessage, error)

+ type CompositionExecutorDeps struct

+ BaseMetadata map[string]interface{}

+ BaseVariables map[string]string

+ Emitter *events.Emitter

+ HookRegistry *hooks.Registry

+ PromptRegistry *prompt.Registry

+ Provider providers.Provider

+ SchemaResolver func(path string) (json.RawMessage, error)

+ ToolRegistry *tools.Registry

+ type CompositionRecorder struct

+ func NewCompositionRecorder() *CompositionRecorder

+ func (r *CompositionRecorder) CompositionMetadata() map[string]any

+ func (r *CompositionRecorder) RecordBranch(id, taken string)

+ func (r *CompositionRecorder) RecordParallel(id string, branches []engine.NamedOutput)

+ func (r *CompositionRecorder) RecordStepCompleted(id, kind string, input, output json.RawMessage, attempt int, err error)

+ func (r *CompositionRecorder) RecordStepStarted(id, kind string, input json.RawMessage)

+ func (r *CompositionRecorder) Reset()

+ func (r *CompositionRecorder) SetEmitter(em *events.Emitter)

+ func (r *CompositionRecorder) Snapshot() *CompositionSnapshot

+ type CompositionSnapshot struct

+ Branches map[string]string

+ Parallel map[string]string

+ Steps []CompositionStepRecord

+ type CompositionStage struct

+ func NewCompositionStage(name string, comp *composition.Composition, deps CompositionExecutorDeps) *CompositionStage

+ func NewCompositionStageWithRecorder(name string, comp *composition.Composition, deps CompositionExecutorDeps, ...) *CompositionStage

+ func (s *CompositionStage) Name() string

+ func (s *CompositionStage) Process(ctx context.Context, in <-chan StreamElement, out chan<- StreamElement) error

+ func (s *CompositionStage) Type() StageType

+ type CompositionStepRecord struct

+ Attempt int

+ ID string

+ Kind string

+ Output json.RawMessage

v1.4.16

Jun 11, 2026

Changes in this version

type PipelineConfig

+ ClassifyRegistry *classify.Registry

v1.4.15

Jun 9, 2026

v1.4.14

Jun 7, 2026

v1.4.13

Jun 6, 2026

v1.4.12

May 28, 2026

v1.4.11

May 22, 2026

v1.4.10

May 19, 2026

v1.4.9

May 18, 2026

v1.4.8

May 15, 2026

v1.4.7

May 15, 2026

Changes in this version

type AudioFormat

+ func (af AudioFormat) BytesPerSample() int

+ type AudioPacingStage struct

+ func NewAudioPacingStage() *AudioPacingStage

+ func NewNamedAudioPacingStage(name string) *AudioPacingStage

+ func (s *AudioPacingStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

type ContextAssemblyStage

+ func NewContextAssemblyStageWithTurnState(config *ContextAssemblyConfig, turnState *TurnState) *ContextAssemblyStage

type ContextBuilderStage

+ func NewContextBuilderStageWithTurnState(policy *ContextBuilderPolicy, turnState *TurnState) *ContextBuilderStage

type DuplexProviderStage

+ func NewDuplexProviderStageWithTurnState(provider providers.StreamInputSupport, ...) *DuplexProviderStage

+ type ElementMetadata struct

+ AllResponsesReceived bool

+ ContextTruncated bool

+ EnableCacheBreakpoints bool

+ FinishReason *string

+ FromHistory bool

+ Interrupted bool

+ InterruptedTurnComplete bool

+ MediaExtract *MediaExtractInfo

+ MergeInputIndex *int

+ Passthrough bool

+ PendingTools []tools.PendingToolExecution

+ TokenCount int

+ ToolResponses []providers.ToolResponse

+ ToolResultMessages []types.Message

+ Transcription *Transcription

+ TurnID *string

+ VideoFrames *VideoFramesInfo

type ExecutionResult

+ PendingTools []tools.PendingToolExecution

type IncrementalSaveStage

+ func NewIncrementalSaveStageWithTurnState(config *IncrementalSaveConfig, turnState *TurnState) *IncrementalSaveStage

+ type MediaExtractInfo struct

+ MediaType string

+ MessageID string

+ OriginalMessage *types.Message

+ PartIndex int

+ TotalParts int

type MemoryRetrievalStage

+ func NewMemoryRetrievalStageWithTurnState(retriever memory.Retriever, store memory.Store, scope map[string]string, ...) *MemoryRetrievalStage

+ func (s *MemoryRetrievalStage) WithContextFormatter(fn memory.ContextFormatter) *MemoryRetrievalStage

type PromptAssemblyStage

+ func NewPromptAssemblyStageWithTurnState(promptRegistry *prompt.Registry, taskType string, ...) *PromptAssemblyStage

type ProviderStage

+ func NewProviderStageWithTurnState(provider providers.Provider, toolRegistry *tools.Registry, ...) *ProviderStage

type StateStoreLoadStage

+ func NewStateStoreLoadStageWithTurnState(config *pipeline.StateStoreConfig, turnState *TurnState) *StateStoreLoadStage

type StreamElement

+ Meta ElementMetadata

type TemplateStage

+ func NewTemplateStageWithTurnState(emitter *events.Emitter, turnState *TurnState) *TemplateStage

type TokenBudgetStage

+ func NewTokenBudgetStageWithTurnState(config *TokenBudgetConfig, turnState *TurnState) *TokenBudgetStage

+ type Transcription struct

+ Text string

+ type TurnState struct

+ AllowedTools []string

+ ConversationID string

+ ProviderRequestMetadata map[string]interface{}

+ SystemPrompt string

+ Template *prompt.Template

+ UserID string

+ Validators []prompt.ValidatorConfig

+ Variables map[string]string

+ func NewTurnState() *TurnState

type VariableProviderStage

+ func NewVariableProviderStageWithVarsAndTurnState(staticVars map[string]string, variableProviders []variables.Provider, ...) *VariableProviderStage

+ type VideoFramesInfo struct

+ FrameIndex int

+ OriginalVideo *VideoData

+ TotalFrames int

+ VideoID string

v1.4.6

Apr 20, 2026

v1.4.5

Apr 15, 2026

Changes in this version

type ProviderConfig

+ ToolSelector selection.Selector

v1.4.4

Apr 12, 2026

v1.4.3

Apr 12, 2026

v1.4.2

Apr 10, 2026

Changes in this version

type STTStageConfig

+ Retry stt.RetryConfig

type TTSStageWithInterruptionConfig

+ Retry tts.RetryConfig

v1.4.1

Apr 4, 2026

Changes in this version

type StreamElement

+ func StreamMediaToElement(media *providers.StreamMediaData) StreamElement

v1.4.0

Mar 8, 2026

retracted

v1.3.30

Apr 4, 2026

v1.3.29

Apr 3, 2026

v1.3.28

Apr 2, 2026

v1.3.27

Apr 1, 2026

v1.3.26

Apr 1, 2026

Changes in this version

type TemplateStage

+ func NewTemplateStageWithEmitter(emitter *events.Emitter) *TemplateStage

type VariableProviderStage

+ func NewVariableProviderStageWithVars(staticVars map[string]string, variableProviders []variables.Provider) *VariableProviderStage

v1.3.25

Apr 1, 2026

v1.3.24

Mar 29, 2026

Changes in this version

+ const DefaultBudgetTokens

+ type CompactResult struct

+ CompactedTokens int

+ Messages []types.Message

+ MessagesFolded int

+ OriginalTokens int

+ type CompactionContext struct

+ Budget int

+ Current int

+ Messages []types.Message

+ PinBoundary int

+ type CompactionRule interface

+ CanFold func(msg *types.Message, idx int, ctx *CompactionContext) bool

+ Fold func(msg *types.Message, idx int, ctx *CompactionContext) (string, []int)

+ Name func() string

+ func CollapsePairs() CompactionRule

+ func FoldToolResults() CompactionRule

+ type CompactionStrategy interface

+ Compact func(messages []types.Message, lastInputTokens int) CompactResult

+ TokenBudget func() int

+ type ContextCompactor struct

+ BudgetTokens int

+ PinRecentCount int

+ Rules []CompactionRule

+ Threshold float64

+ func (c *ContextCompactor) Compact(messages []types.Message, lastInputTokens int) CompactResult

+ func (c *ContextCompactor) TokenBudget() int

type IncrementalSaveConfig

+ MessageLog statestore.MessageLog

+ type MemoryExtractionStage struct

+ func NewMemoryExtractionStage(extractor memory.Extractor, store memory.Store, scope map[string]string) *MemoryExtractionStage

+ func (s *MemoryExtractionStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type MemoryRetrievalStage struct

+ func NewMemoryRetrievalStage(retriever memory.Retriever, store memory.Store, scope map[string]string) *MemoryRetrievalStage

+ func (s *MemoryRetrievalStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

type ProviderConfig

+ Compactor CompactionStrategy

+ MessageLog statestore.MessageLog

+ MessageLogConvID string

v1.3.23

Mar 27, 2026

Changes in this version

+ const DefaultIdleTimeoutSeconds

+ var ErrIdleTimeout = errors.New("pipeline idle timeout: no activity detected")

+ var ErrInvalidIdleTimeout = errors.New("idle timeout must be non-negative")

+ func ResetIdleFromContext(ctx context.Context)

type PipelineConfig

+ IdleTimeout time.Duration

+ func (c *PipelineConfig) WithIdleTimeout(timeout time.Duration) *PipelineConfig

type RecordingStageConfig

+ IncludeStreamingText bool

v1.3.22

Mar 25, 2026

v1.3.21

Mar 24, 2026

v1.3.20

Mar 23, 2026

Changes in this version

type ProviderConfig

+ Source string

v1.3.19

Mar 20, 2026

v1.3.18

Mar 19, 2026

Changes in this version

type StreamPipeline

+ BaseMetadata map[string]interface{}

v1.3.17

Mar 18, 2026

v1.3.16

Mar 16, 2026

v1.3.15

Mar 15, 2026

v1.3.14

Mar 14, 2026

v1.3.13

Mar 13, 2026

Changes in this version

type ProviderConfig

+ Labels map[string]string

v1.3.12

Mar 10, 2026

v1.3.11

Mar 9, 2026

v1.3.10

Mar 8, 2026

Changes in this version

+ const DefaultAudioChannelBufferSize

+ const DefaultMaxMessages

+ const DefaultWarningThreshold

type ContextAssemblyConfig

+ HasContextWindow bool

+ HasTokenBudget bool

+ MaxMessages int

+ WarningThreshold int

+ type TokenBudgetConfig struct

+ LargeConversationThreshold int

+ MaxTokens int

+ ReserveTokens int

+ TokenCounter *tokenizer.HeuristicTokenCounter

+ type TokenBudgetStage struct

+ func NewTokenBudgetStage(config *TokenBudgetConfig) *TokenBudgetStage

+ func (s *TokenBudgetStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

v1.3.9

Mar 4, 2026

v1.3.8

Mar 4, 2026

v1.3.7

Mar 3, 2026

v1.3.6

Mar 3, 2026

v1.3.5

Mar 1, 2026

v1.3.4

Mar 1, 2026

v1.3.3

Mar 1, 2026

v1.3.2

Feb 26, 2026

Changes in this version

type ProviderStage

+ func NewProviderStageWithHooks(provider providers.Provider, toolRegistry *tools.Registry, ...) *ProviderStage

v1.3.1

Feb 23, 2026

Changes in this version

+ type ContextAssemblyConfig struct

+ MessageIndex statestore.MessageIndex

+ RecentMessages int

+ RetrievalTopK int

+ StateStoreConfig *rtpipeline.StateStoreConfig

+ type ContextAssemblyStage struct

+ func NewContextAssemblyStage(config *ContextAssemblyConfig) *ContextAssemblyStage

+ func (s *ContextAssemblyStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

type ContextBuilderPolicy

+ Summarizer statestore.Summarizer

+ type IncrementalSaveConfig struct

+ MessageIndex statestore.MessageIndex

+ StateStoreConfig *rtpipeline.StateStoreConfig

+ SummarizeBatchSize int

+ SummarizeThreshold int

+ Summarizer statestore.Summarizer

+ type IncrementalSaveStage struct

+ func NewIncrementalSaveStage(config *IncrementalSaveConfig) *IncrementalSaveStage

+ func (s *IncrementalSaveStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

v1.2.0

Feb 15, 2026

v1.1.11

Feb 15, 2026

v1.1.10

Jan 27, 2026

Changes in this version

+ type MediaConvertConfig struct

+ AudioConverterConfig media.AudioConverterConfig

+ ImageResizeConfig media.ImageResizeConfig

+ PassthroughOnError bool

+ TargetAudioFormats []string

+ TargetImageFormats []string

+ TargetVideoFormats []string

+ func DefaultMediaConvertConfig() MediaConvertConfig

+ type MediaConvertStage struct

+ func NewMediaConvertStage(config *MediaConvertConfig) *MediaConvertStage

+ func (s *MediaConvertStage) GetConfig() MediaConvertConfig

+ func (s *MediaConvertStage) InputCapabilities() Capabilities

+ func (s *MediaConvertStage) OutputCapabilities() Capabilities

+ func (s *MediaConvertStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

v1.1.9

Jan 24, 2026

Changes in this version

type ContextBuilderPolicy

+ TokenCounter tokenizer.TokenCounter

v1.1.8

Jan 15, 2026

Changes in this version

+ const DefaultCompletionTimeout

+ const DefaultFFmpegPath

+ const DefaultFFmpegTimeout

+ const DefaultFrameInterval

+ const DefaultFrameRateLimitFPS

+ const DefaultFramesCompletionTimeout

+ const DefaultMaxFrames

+ const DefaultMaxFramesPerMessage

+ const DefaultOutputFormat

+ const DefaultOutputQuality

+ const DefaultTargetFPS

+ const MediaExtractMediaTypeKey

+ const MediaExtractMessageIDKey

+ const MediaExtractOriginalMessageKey

+ const MediaExtractPartIndexKey

+ const MediaExtractTotalPartsKey

+ const OutputFormatPNG

+ const VideoFramesFrameIndexKey

+ const VideoFramesOriginalVideoKey

+ const VideoFramesTimestampKey

+ const VideoFramesTotalFramesKey

+ const VideoFramesVideoIDKey

+ var ErrFFmpegFailed = errors.New("ffmpeg execution failed")

+ var ErrFFmpegNotFound = errors.New("ffmpeg not found")

+ var ErrFFmpegTimeout = errors.New("ffmpeg execution timeout")

+ var ErrInvalidVideoFormat = errors.New("invalid or unsupported video format")

+ var ErrNoFramesExtracted = errors.New("no frames extracted from video")

+ var ErrVideoDataRequired = errors.New("video data required but not available")

+ type DropStrategy int

+ const DropStrategyKeepLatest

+ const DropStrategyUniform

+ func (s DropStrategy) String() string

+ type FrameExtractionMode int

+ const FrameExtractionFPS

+ const FrameExtractionInterval

+ const FrameExtractionKeyframes

+ func (m FrameExtractionMode) String() string

+ type FrameRateLimitConfig struct

+ DropStrategy DropStrategy

+ PassthroughAudio bool

+ PassthroughNonMedia bool

+ TargetFPS float64

+ func DefaultFrameRateLimitConfig() FrameRateLimitConfig

+ type FrameRateLimitStage struct

+ func NewFrameRateLimitStage(config FrameRateLimitConfig) *FrameRateLimitStage

+ func (s *FrameRateLimitStage) GetConfig() FrameRateLimitConfig

+ func (s *FrameRateLimitStage) GetStats() (emitted, dropped int64)

+ func (s *FrameRateLimitStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type FrameSelectionStrategy int

+ const FrameSelectionFirst

+ const FrameSelectionLast

+ const FrameSelectionUniform

+ func (s FrameSelectionStrategy) String() string

+ type FramesToMessageConfig struct

+ CompletionTimeout time.Duration

+ FrameSelectionStrategy FrameSelectionStrategy

+ MaxFramesPerMessage int

+ StorageService storage.MediaStorageService

+ func DefaultFramesToMessageConfig() FramesToMessageConfig

+ type FramesToMessageStage struct

+ func NewFramesToMessageStage(config FramesToMessageConfig) *FramesToMessageStage

+ func (s *FramesToMessageStage) GetConfig() FramesToMessageConfig

+ func (s *FramesToMessageStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

type ImageData

+ FrameNum int64

+ StorageRef storage.Reference

+ Timestamp time.Time

+ func (d *ImageData) EnsureLoaded(ctx context.Context, store storage.MediaStorageService) ([]byte, error)

+ func (d *ImageData) Externalize(ctx context.Context, store storage.MediaStorageService, ...) error

+ func (d *ImageData) IsExternalized() bool

+ func (d *ImageData) Load(ctx context.Context, store storage.MediaStorageService) error

+ type ImagePreprocessConfig struct

+ EnableResize bool

+ Resize ImageResizeStageConfig

+ func DefaultImagePreprocessConfig() ImagePreprocessConfig

+ type ImagePreprocessStage struct

+ func NewImagePreprocessStage(config ImagePreprocessConfig) *ImagePreprocessStage

+ func (s *ImagePreprocessStage) GetConfig() ImagePreprocessConfig

+ func (s *ImagePreprocessStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type ImageResizeStage struct

+ func NewImageResizeStage(config ImageResizeStageConfig) *ImageResizeStage

+ func (s *ImageResizeStage) GetConfig() ImageResizeStageConfig

+ func (s *ImageResizeStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type ImageResizeStageConfig = media.ImageResizeConfig

+ func DefaultImageResizeStageConfig() ImageResizeStageConfig

+ type MediaComposeConfig struct

+ CompletionTimeout time.Duration

+ StorageService storage.MediaStorageService

+ func DefaultMediaComposeConfig() MediaComposeConfig

+ type MediaComposeStage struct

+ func NewMediaComposeStage(config MediaComposeConfig) *MediaComposeStage

+ func (s *MediaComposeStage) GetConfig() MediaComposeConfig

+ func (s *MediaComposeStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type MediaExtractConfig struct

+ ExtractImages bool

+ ExtractVideos bool

+ PreserveStorageRefs bool

+ StorageService storage.MediaStorageService

+ func DefaultMediaExtractConfig() MediaExtractConfig

+ type MediaExtractStage struct

+ func NewMediaExtractStage(config MediaExtractConfig) *MediaExtractStage

+ func (s *MediaExtractStage) GetConfig() MediaExtractConfig

+ func (s *MediaExtractStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

type ProviderConfig

+ ResponseFormat *providers.ResponseFormat

type ProviderStage

+ func NewProviderStageWithEmitter(provider providers.Provider, toolRegistry *tools.Registry, ...) *ProviderStage

type VideoData

+ FrameNum int64

+ StorageRef storage.Reference

+ func (d *VideoData) EnsureLoaded(ctx context.Context, store storage.MediaStorageService) ([]byte, error)

+ func (d *VideoData) Externalize(ctx context.Context, store storage.MediaStorageService, ...) error

+ func (d *VideoData) IsExternalized() bool

+ func (d *VideoData) Load(ctx context.Context, store storage.MediaStorageService) error

+ type VideoToFramesConfig struct

+ FFmpegPath string

+ FFmpegTimeout time.Duration

+ Interval time.Duration

+ MaxFrames int

+ Mode FrameExtractionMode

+ OutputFormat string

+ OutputQuality int

+ OutputWidth int

+ StorageService storage.MediaStorageService

+ TargetFPS float64

+ func DefaultVideoToFramesConfig() VideoToFramesConfig

+ type VideoToFramesStage struct

+ func NewVideoToFramesStage(config VideoToFramesConfig) *VideoToFramesStage

+ func (s *VideoToFramesStage) GetConfig() VideoToFramesConfig

+ func (s *VideoToFramesStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

v1.1.7

Jan 10, 2026

Changes in this version

type PipelineConfig

+ PrometheusAddr string

+ PrometheusEnabled bool

+ func (c *PipelineConfig) WithPrometheusExporter(addr string) *PipelineConfig

v1.1.6

Dec 23, 2025

Changes in this version

+ const DefaultChannelBufferSize

+ const DefaultExecutionTimeoutSeconds

+ const DefaultGracefulShutdownTimeoutSeconds

+ const DefaultMaxConcurrentPipelines

+ var ErrCyclicDependency = errors.New("cyclic dependency detected in pipeline")

+ var ErrDuplicateStageName = errors.New("duplicate stage name")

+ var ErrInvalidChannelBufferSize = errors.New("channel buffer size must be non-negative")

+ var ErrInvalidExecutionTimeout = errors.New("execution timeout must be non-negative")

+ var ErrInvalidGracefulShutdownTimeout = errors.New("graceful shutdown timeout must be non-negative")

+ var ErrInvalidMaxConcurrentPipelines = errors.New("max concurrent pipelines must be non-negative")

+ var ErrInvalidPipeline = errors.New("invalid pipeline configuration")

+ var ErrNoStages = errors.New("pipeline must have at least one stage")

+ var ErrPipelineShuttingDown = errors.New("pipeline is shutting down")

+ var ErrShutdownTimeout = errors.New("shutdown timeout exceeded")

+ var ErrStageNotFound = errors.New("stage not found")

+ func BatchEmbeddingTexts(texts []string, batchSize int) [][]string

+ func CosineSimilarity(a, b []float32) float64

+ func DescribeCapabilities(stage Stage) string

+ func GetTraceInfo(elem *StreamElement) (traceID string, stageTimes map[string]time.Time)

+ func NormalizeEmbedding(embedding []float32) []float32

+ func PutElement(elem *StreamElement)

+ func ValidateCapabilities(stages []Stage, edges map[string][]string)

+ type AudioCapability struct

+ Channels []int

+ Formats []AudioFormat

+ SampleRates []int

+ func (ac *AudioCapability) AcceptsAudio(audio *AudioData) bool

+ func (ac *AudioCapability) AcceptsChannels(channels int) bool

+ func (ac *AudioCapability) AcceptsFormat(format AudioFormat) bool

+ func (ac *AudioCapability) AcceptsSampleRate(rate int) bool

+ type AudioData struct

+ Channels int

+ Duration time.Duration

+ Encoding string

+ Format AudioFormat

+ SampleRate int

+ Samples []byte

+ type AudioFormat int

+ const AudioFormatAAC

+ const AudioFormatFloat32

+ const AudioFormatMP3

+ const AudioFormatOpus

+ const AudioFormatPCM16

+ func (af AudioFormat) String() string

+ type AudioResampleConfig struct

+ PassthroughIfSameRate bool

+ TargetSampleRate int

+ func DefaultAudioResampleConfig() AudioResampleConfig

+ type AudioResampleStage struct

+ func NewAudioResampleStage(config AudioResampleConfig) *AudioResampleStage

+ func (s *AudioResampleStage) GetConfig() AudioResampleConfig

+ func (s *AudioResampleStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type AudioTurnConfig struct

+ InterruptionHandler *audio.InterruptionHandler

+ MaxTurnDuration time.Duration

+ MinSpeechDuration time.Duration

+ SampleRate int

+ SilenceDuration time.Duration

+ TurnDetector audio.TurnDetector

+ VAD audio.VADAnalyzer

+ func DefaultAudioTurnConfig() AudioTurnConfig

+ type AudioTurnStage struct

+ func NewAudioTurnStage(config AudioTurnConfig) (*AudioTurnStage, error)

+ func (s *AudioTurnStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type BaseStage struct

+ func NewBaseStage(name string, stageType StageType) BaseStage

+ func (b *BaseStage) Name() string

+ func (b *BaseStage) Type() StageType

+ type BroadcastRouter struct

+ func NewBroadcastRouter(name string) *BroadcastRouter

+ func (r *BroadcastRouter) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (r *BroadcastRouter) RegisterOutput(name string, output chan<- StreamElement)

+ type ByOriginalIndex []ScoredMessage

+ func (s ByOriginalIndex) Len() int

+ func (s ByOriginalIndex) Less(i, j int) bool

+ func (s ByOriginalIndex) Swap(i, j int)

+ type Capabilities struct

+ Audio *AudioCapability

+ ContentTypes []ContentType

+ func AnyCapabilities() Capabilities

+ func AudioCapabilities(formats []AudioFormat, sampleRates, channels []int) Capabilities

+ func MessageCapabilities() Capabilities

+ func TextCapabilities() Capabilities

+ func (c *Capabilities) AcceptsContentType(ct ContentType) bool

+ func (c *Capabilities) AcceptsElement(elem *StreamElement) bool

+ type ContentRouter struct

+ func NewContentRouter(name string, rules ...RoutingRule) *ContentRouter

+ func (r *ContentRouter) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (r *ContentRouter) RegisterOutput(name string, output chan<- StreamElement)

+ type ContentType int

+ const ContentTypeAny

+ const ContentTypeAudio

+ const ContentTypeImage

+ const ContentTypeMessage

+ const ContentTypeText

+ const ContentTypeToolCall

+ const ContentTypeVideo

+ func (ct ContentType) String() string

+ type ContextBuilderPolicy struct

+ CacheBreakpoints bool

+ RelevanceConfig *RelevanceConfig

+ ReserveForOutput int

+ Strategy TruncationStrategy

+ TokenBudget int

+ type ContextBuilderStage struct

+ func NewContextBuilderStage(policy *ContextBuilderPolicy) *ContextBuilderStage

+ func (s *ContextBuilderStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type DebugStage struct

+ func NewDebugStage(stageName string) *DebugStage

+ func (s *DebugStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type DuplexProviderStage struct

+ func NewDuplexProviderStage(provider providers.StreamInputSupport, ...) *DuplexProviderStage

+ func NewDuplexProviderStageWithEmitter(provider providers.StreamInputSupport, ...) *DuplexProviderStage

+ func (s *DuplexProviderStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type EndInputter interface

+ EndInput func()

+ type ExecutionResult struct

+ CostInfo types.CostInfo

+ Messages []types.Message

+ Metadata map[string]interface{}

+ Response *Response

+ Trace ExecutionTrace

+ type ExecutionTrace struct

+ CompletedAt *time.Time

+ Duration time.Duration

+ StartedAt time.Time

+ type FilterStage struct

+ func NewFilterStage(name string, predicate func(StreamElement) bool) *FilterStage

+ func (fs *FilterStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type FormatCapable interface

+ InputCapabilities func() Capabilities

+ OutputCapabilities func() Capabilities

+ type HashRouter struct

+ func NewHashRouter(name string, outputNames []string, keyFunc func(StreamElement) string) *HashRouter

+ func (r *HashRouter) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (r *HashRouter) RegisterOutput(name string, output chan<- StreamElement)

+ type ImageData struct

+ Data []byte

+ Format string

+ Height int

+ MIMEType string

+ Width int

+ type MapStage struct

+ func NewMapStage(name string, mapFunc func(StreamElement) (StreamElement, error)) *MapStage

+ func (ms *MapStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type MediaExternalizerConfig struct

+ ConversationID string

+ DefaultPolicy string

+ Enabled bool

+ RunID string

+ SessionID string

+ SizeThresholdKB int64

+ StorageService storage.MediaStorageService

+ type MediaExternalizerStage struct

+ func NewMediaExternalizerStage(config *MediaExternalizerConfig) *MediaExternalizerStage

+ func (s *MediaExternalizerStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type MergeStage struct

+ func NewMergeStage(name string, inputCount int) *MergeStage

+ func (s *MergeStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (s *MergeStage) ProcessMultiple(ctx context.Context, inputs []<-chan StreamElement, ...) error

+ type MetricsStage struct

+ func NewMetricsStage(wrappedStage Stage) *MetricsStage

+ func (s *MetricsStage) GetMetrics() StageMetrics

+ func (s *MetricsStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type PassthroughStage struct

+ func NewPassthroughStage(name string) *PassthroughStage

+ func (ps *PassthroughStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type PipelineBuilder struct

+ func NewPipelineBuilder() *PipelineBuilder

+ func NewPipelineBuilderWithConfig(config *PipelineConfig) *PipelineBuilder

+ func (b *PipelineBuilder) AddStage(stage Stage) *PipelineBuilder

+ func (b *PipelineBuilder) Branch(fromStage string, toStages ...string) *PipelineBuilder

+ func (b *PipelineBuilder) Build() (*StreamPipeline, error)

+ func (b *PipelineBuilder) Chain(stages ...Stage) *PipelineBuilder

+ func (b *PipelineBuilder) Clone() *PipelineBuilder

+ func (b *PipelineBuilder) Connect(fromStage, toStage string) *PipelineBuilder

+ func (b *PipelineBuilder) WithConfig(config *PipelineConfig) *PipelineBuilder

+ func (b *PipelineBuilder) WithEventEmitter(emitter *events.Emitter) *PipelineBuilder

+ type PipelineConfig struct

+ ChannelBufferSize int

+ EnableMetrics bool

+ EnableTracing bool

+ ExecutionTimeout time.Duration

+ GracefulShutdownTimeout time.Duration

+ MaxConcurrentPipelines int

+ PriorityQueueEnabled bool

+ func DefaultPipelineConfig() *PipelineConfig

+ func (c *PipelineConfig) Validate() error

+ func (c *PipelineConfig) WithChannelBufferSize(size int) *PipelineConfig

+ func (c *PipelineConfig) WithExecutionTimeout(timeout time.Duration) *PipelineConfig

+ func (c *PipelineConfig) WithGracefulShutdownTimeout(timeout time.Duration) *PipelineConfig

+ func (c *PipelineConfig) WithMaxConcurrentPipelines(maxPipelines int) *PipelineConfig

+ func (c *PipelineConfig) WithMetrics(enabled bool) *PipelineConfig

+ func (c *PipelineConfig) WithPriorityQueue(enabled bool) *PipelineConfig

+ func (c *PipelineConfig) WithTracing(enabled bool) *PipelineConfig

+ type Priority int

+ const PriorityCritical

+ const PriorityHigh

+ const PriorityLow

+ const PriorityNormal

+ type PriorityChannel struct

+ func NewPriorityChannel(capacity int) *PriorityChannel

+ func (pc *PriorityChannel) Close()

+ func (pc *PriorityChannel) Len() int

+ func (pc *PriorityChannel) Receive(ctx context.Context) (StreamElement, bool, error)

+ func (pc *PriorityChannel) Send(ctx context.Context, elem StreamElement) error

+ type PromptAssemblyStage struct

+ func NewPromptAssemblyStage(promptRegistry *prompt.Registry, taskType string, ...) *PromptAssemblyStage

+ func (s *PromptAssemblyStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type ProviderConfig struct

+ DisableTrace bool

+ MaxTokens int

+ Seed *int

+ Temperature float32

+ type ProviderStage struct

+ func NewProviderStage(provider providers.Provider, toolRegistry *tools.Registry, ...) *ProviderStage

+ func (s *ProviderStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type QuerySourceType string

+ const QuerySourceCustom

+ const QuerySourceLastN

+ const QuerySourceLastUser

+ type RandomRouter struct

+ func NewRandomRouter(name string, outputNames []string) *RandomRouter

+ func (r *RandomRouter) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (r *RandomRouter) RegisterOutput(name string, output chan<- StreamElement)

+ type RecordingPosition string

+ const RecordingPositionInput

+ const RecordingPositionOutput

+ type RecordingStage struct

+ func NewRecordingStage(eventBus *events.EventBus, config RecordingStageConfig) *RecordingStage

+ func (rs *RecordingStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (rs *RecordingStage) WithConversationID(conversationID string) *RecordingStage

+ func (rs *RecordingStage) WithSessionID(sessionID string) *RecordingStage

+ type RecordingStageConfig struct

+ ConversationID string

+ IncludeAudio bool

+ IncludeImages bool

+ IncludeVideo bool

+ Position RecordingPosition

+ SessionID string

+ func DefaultRecordingStageConfig() RecordingStageConfig

+ type RelevanceConfig struct

+ AlwaysKeepSystemRole bool

+ CacheEmbeddings bool

+ CustomQuery string

+ EmbeddingProvider providers.EmbeddingProvider

+ LastNCount int

+ MinRecentMessages int

+ QuerySource QuerySourceType

+ SimilarityThreshold float64

+ type Response struct

+ Content string

+ FinalResponse string

+ Parts []types.ContentPart

+ Role string

+ ToolCalls []types.MessageToolCall

+ type ResponseVADConfig struct

+ MaxWaitDuration time.Duration

+ SampleRate int

+ SilenceDuration time.Duration

+ VAD audio.VADAnalyzer

+ func DefaultResponseVADConfig() ResponseVADConfig

+ type ResponseVADStage struct

+ func NewResponseVADStage(config ResponseVADConfig) (*ResponseVADStage, error)

+ func (s *ResponseVADStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type RoundRobinRouter struct

+ func NewRoundRobinRouter(name string, outputNames []string) *RoundRobinRouter

+ func (r *RoundRobinRouter) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (r *RoundRobinRouter) RegisterOutput(name string, output chan<- StreamElement)

+ type RouterFunc func(elem *StreamElement) []string

+ type RouterStage struct

+ func NewRouterStage(name string, routerFunc RouterFunc) *RouterStage

+ func (s *RouterStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (s *RouterStage) RegisterOutput(name string, output chan<- StreamElement)

+ type RoutingRule struct

+ Name string

+ Output string

+ Predicate func(StreamElement) bool

+ func RouteAudio(output string, format AudioFormat) RoutingRule

+ func RouteContentType(output string, ct ContentType) RoutingRule

+ func RouteWhen(output string, predicate func(StreamElement) bool) RoutingRule

+ type STTStage struct

+ func NewSTTStage(service stt.Service, config STTStageConfig) *STTStage

+ func (s *STTStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type STTStageConfig struct

+ Language string

+ MinAudioBytes int

+ SkipEmpty bool

+ func DefaultSTTStageConfig() STTStageConfig

+ type ScoredMessage struct

+ Index int

+ IsProtected bool

+ Message types.Message

+ Score float64

+ TokenCount int

+ type ScoredMessages []ScoredMessage

+ func (s ScoredMessages) Len() int

+ func (s ScoredMessages) Less(i, j int) bool

+ func (s ScoredMessages) Swap(i, j int)

+ type Stage interface

+ Name func() string

+ Process func(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ Type func() StageType

+ type StageError struct

+ Err error

+ StageName string

+ StageType StageType

+ func NewStageError(stageName string, stageType StageType, err error) *StageError

+ func (e *StageError) Error() string

+ func (e *StageError) Unwrap() error

+ type StageFunc struct

+ func NewStageFunc(name string, stageType StageType, ...) *StageFunc

+ func (sf *StageFunc) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type StageMetrics struct

+ AvgLatency time.Duration

+ ElementsErrored int64

+ ElementsIn int64

+ ElementsOut int64

+ LastUpdated time.Time

+ MaxLatency time.Duration

+ MinLatency time.Duration

+ StageName string

+ TotalLatency time.Duration

+ func NewStageMetrics(stageName string) *StageMetrics

+ func (m *StageMetrics) GetMetrics() StageMetrics

+ func (m *StageMetrics) RecordElement(latency time.Duration, hasError bool)

+ func (m *StageMetrics) Reset()

+ type StageType int

+ const StageTypeAccumulate

+ const StageTypeBidirectional

+ const StageTypeGenerate

+ const StageTypeSink

+ const StageTypeTransform

+ func (st StageType) String() string

+ type StateStoreLoadStage struct

+ func NewStateStoreLoadStage(config *pipeline.StateStoreConfig) *StateStoreLoadStage

+ func (s *StateStoreLoadStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type StateStoreSaveStage struct

+ func NewStateStoreSaveStage(config *pipeline.StateStoreConfig) *StateStoreSaveStage

+ func (s *StateStoreSaveStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type StreamElement struct

+ Audio *AudioData

+ EndOfStream bool

+ Error error

+ Image *ImageData

+ MediaData *types.MediaContent

+ Message *types.Message

+ Metadata map[string]interface{}

+ Part *types.ContentPart

+ Priority Priority

+ Sequence int64

+ Source string

+ Text *string

+ Timestamp time.Time

+ ToolCall *types.MessageToolCall

+ Video *VideoData

+ func GetAudioElement(audio *AudioData) *StreamElement

+ func GetElement() *StreamElement

+ func GetEndOfStreamElement() *StreamElement

+ func GetErrorElement(err error) *StreamElement

+ func GetImageElement(image *ImageData) *StreamElement

+ func GetMessageElement(msg *types.Message) *StreamElement

+ func GetTextElement(text string) *StreamElement

+ func GetVideoElement(video *VideoData) *StreamElement

+ func NewAudioElement(audio *AudioData) StreamElement

+ func NewEndOfStreamElement() StreamElement

+ func NewErrorElement(err error) StreamElement

+ func NewImageElement(image *ImageData) StreamElement

+ func NewMessageElement(msg *types.Message) StreamElement

+ func NewTextElement(text string) StreamElement

+ func NewVideoElement(video *VideoData) StreamElement

+ func (e *StreamElement) GetMetadata(key string) interface{}

+ func (e *StreamElement) HasContent() bool

+ func (e *StreamElement) IsControl() bool

+ func (e *StreamElement) IsEmpty() bool

+ func (e *StreamElement) Reset()

+ func (e *StreamElement) WithMetadata(key string, value interface{}) *StreamElement

+ func (e *StreamElement) WithPriority(priority Priority) *StreamElement

+ func (e *StreamElement) WithSequence(seq int64) *StreamElement

+ func (e *StreamElement) WithSource(source string) *StreamElement

+ type StreamPipeline struct

+ func (p *StreamPipeline) Execute(ctx context.Context, input <-chan StreamElement) (<-chan StreamElement, error)

+ func (p *StreamPipeline) ExecuteSync(ctx context.Context, input ...StreamElement) (*ExecutionResult, error)

+ func (p *StreamPipeline) Shutdown(ctx context.Context) error

+ type TTSConfig struct

+ MinTextLength int

+ SkipEmpty bool

+ func DefaultTTSConfig() TTSConfig

+ type TTSService interface

+ MIMEType func() string

+ Synthesize func(ctx context.Context, text string) ([]byte, error)

+ type TTSStage struct

+ func NewTTSStage(tts TTSService, config TTSConfig) *TTSStage

+ func (s *TTSStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type TTSStageWithInterruption struct

+ func NewTTSStageWithInterruption(service tts.Service, config TTSStageWithInterruptionConfig) *TTSStageWithInterruption

+ func (s *TTSStageWithInterruption) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type TTSStageWithInterruptionConfig struct

+ InterruptionHandler *audio.InterruptionHandler

+ MinTextLength int

+ SkipEmpty bool

+ Speed float64

+ Voice string

+ func DefaultTTSStageWithInterruptionConfig() TTSStageWithInterruptionConfig

+ type TemplateStage struct

+ func NewTemplateStage() *TemplateStage

+ func (s *TemplateStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type TracingStage struct

+ func NewTracingStage(wrappedStage Stage, traceIDGen func() string) *TracingStage

+ func (s *TracingStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type Transcriber interface

+ Transcribe func(ctx context.Context, audio []byte) (string, error)

+ type TruncationStrategy string

+ const TruncateFail

+ const TruncateLeastRelevant

+ const TruncateOldest

+ const TruncateSummarize

+ type VADAccumulatorStage struct

+ func NewVADAccumulatorStage(analyzer audio.VADAnalyzer, transcriber Transcriber, config VADConfig) *VADAccumulatorStage

+ func (s *VADAccumulatorStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type VADConfig struct

+ MaxTurnDuration time.Duration

+ MinSpeechDuration time.Duration

+ SilenceDuration time.Duration

+ Threshold float64

+ func DefaultVADConfig() VADConfig

+ type ValidationStage struct

+ func NewValidationStage(registry *validators.Registry, suppressExceptions bool) *ValidationStage

+ func (s *ValidationStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type VariableProviderStage struct

+ func NewVariableProviderStage(providers ...variables.Provider) *VariableProviderStage

+ func (s *VariableProviderStage) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ type VideoData struct

+ Data []byte

+ Duration time.Duration

+ Format string

+ FrameRate float64

+ Height int

+ IsKeyFrame bool

+ MIMEType string

+ Timestamp time.Time

+ Width int

+ type WeightedRouter struct

+ func NewWeightedRouter(name string, weights map[string]float64) *WeightedRouter

+ func (r *WeightedRouter) Process(ctx context.Context, input <-chan StreamElement, output chan<- StreamElement) error

+ func (r *WeightedRouter) RegisterOutput(name string, output chan<- StreamElement)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL