Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
- API
  
  Reference documentation for the pkg.go.dev API
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

processor

package

Go to main page

Versions in this module

v0

v0.1.0

Aug 30, 2026

Changes in this version

+ const DefaultPauseUntilReadyTimeout

+ const EventAfterProcessFrame

+ const EventAfterPushFrame

+ const EventBeforeProcessFrame

+ const EventBeforePushFrame

+ const EventError

+ const EventUsableChanged

+ func FrameIs[T frames.Frame]() func(frames.Frame) bool

+ func IdentityTransformer(f frames.Frame) frames.Frame

+ func IsSource(p Processor) bool

+ func NotifyProcessorSetup(observers []Observer, data ProcessorSetUp)

type Base

+ func (b *Base) AudioInSampleRate() int

+ func (b *Base) AudioOutSampleRate() int

+ func (b *Base) BeginTTFB() bool

+ func (b *Base) Broadcast(ctx context.Context, build func() frames.Frame) error

+ func (b *Base) BroadcastInterruption(ctx context.Context) error

+ func (b *Base) CanGenerateMetrics() bool

+ func (b *Base) EntryProcessors() []Processor

+ func (b *Base) Events() *events.Registry

+ func (b *Base) FlushPipeline(ctx context.Context) error

+ func (b *Base) HasQueuedFrame(match func(frames.Frame) bool) bool

+ func (b *Base) PauseProcessingAllFramesUntil(ready func(ctx context.Context), timeout time.Duration)

+ func (b *Base) PauseProcessingFrames()

+ func (b *Base) PauseProcessingSystemFrames()

+ func (b *Base) Processors() []Processor

+ func (b *Base) ProcessorsWithMetrics() []Processor

+ func (b *Base) PushErrorFrame(ctx context.Context, ef *frames.ErrorFrame, forceTreatAsPermanent bool)

+ func (b *Base) PushTokenUsage(ctx context.Context, model string, u frames.LLMTokenUsage) error

+ func (b *Base) ResumeProcessingFrames()

+ func (b *Base) ResumeProcessingSystemFrames()

+ func (b *Base) Running() Running

+ func (b *Base) Self() Processor

+ func (b *Base) SetUsable(ctx context.Context, usable bool)

+ func (b *Base) StartSpan(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span)

+ func (b *Base) Tracing() *tracing.TracingContext

+ func (b *Base) TracingEnabled() bool

+ func (b *Base) TrackFlushProbe(f *frames.PipelineFlushFrame)

+ func (b *Base) TypeName() string

+ func (b *Base) Usable() bool

+ type ConsumerOption func(*ConsumerProcessor)

+ func WithConsumerDirection(d Direction) ConsumerOption

+ func WithConsumerTransformer(t FrameTransformer) ConsumerOption

+ type ConsumerProcessor struct

+ func NewConsumerProcessor(name string, producer *ProducerProcessor, opts ...ConsumerOption) *ConsumerProcessor

+ func (c *ConsumerProcessor) Cleanup(ctx context.Context) error

+ func (c *ConsumerProcessor) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type ErrorClassifier interface

+ ClassifyError func(err error) errs.Category

+ type ErrorOption func(*errorOptions)

+ func ForceTreatAsPermanent() ErrorOption

+ func WithErrorCategory(c errs.Category) ErrorOption

+ type FilterOption func(*FunctionFilter)

+ func WithFilterSystemFrames() FilterOption

+ type FrameFilter struct

+ func NewFrameFilter(name string, allowed ...func(frames.Frame) bool) *FrameFilter

+ func (f *FrameFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type FrameProcessed struct

+ Direction Direction

+ Frame frames.Frame

+ Processor Processor

+ Timestamp time.Duration

+ type FramePushed struct

+ Destination Processor

+ Direction Direction

+ Frame frames.Frame

+ Source Processor

+ Timestamp time.Duration

+ type FrameTransformer func(frames.Frame) frames.Frame

+ type IdentityFilter struct

+ func NewIdentityFilter(name string) *IdentityFilter

+ func (f *IdentityFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type IdleFrameProcessor struct

+ func NewIdleFrameProcessor(name string, timeout time.Duration, callback func(*IdleFrameProcessor), ...) *IdleFrameProcessor

+ func (p *IdleFrameProcessor) Cleanup(ctx context.Context) error

+ func (p *IdleFrameProcessor) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type NullFilter struct

+ func NewNullFilter(name string) *NullFilter

+ func (f *NullFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type Observer interface

+ OnPushFrame func(data FramePushed)

+ type PipelineStartedObserver interface

+ OnPipelineStarted func()

+ type ProcessObserver interface

+ OnProcessFrame func(data FrameProcessed)

type Processor

+ CanGenerateMetrics func() bool

+ EntryProcessors func() []Processor

+ Events func() *events.Registry

+ Processors func() []Processor

+ ProcessorsWithMetrics func() []Processor

+ PushErrorFrame func(ctx context.Context, ef *frames.ErrorFrame, forceTreatAsPermanent bool)

+ SetUsable func(ctx context.Context, usable bool)

+ Usable func() bool

+ type ProcessorSetUp struct

+ FinishedAt time.Time

+ Processor Processor

+ StartedAt time.Time

+ func (d ProcessorSetUp) Duration() time.Duration

+ type ProducerOption func(*ProducerProcessor)

+ func WithProducerTransformer(t FrameTransformer) ProducerOption

+ func WithoutPassthrough() ProducerOption

+ type ProducerProcessor struct

+ func NewProducerProcessor(name string, filter func(frames.Frame) bool, opts ...ProducerOption) *ProducerProcessor

+ func (p *ProducerProcessor) AddConsumer() *frameQueue

+ func (p *ProducerProcessor) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type Running interface

+ Flush func(ctx context.Context) error

+ TrackFlushProbe func(f *frames.PipelineFlushFrame)

+ TurnTracker func() TurnTracker

type Setup

+ AudioInSampleRate int

+ AudioOutSampleRate int

+ EnableMetrics bool

+ EnableUsageMetrics bool

+ Observers []Observer

+ ReportOnlyInitialTTFB bool

+ Running Running

+ Tracing *tracing.TracingContext

+ TracingEnabled bool

+ type SetupObserver interface

+ OnProcessorSetup func(data ProcessorSetUp)

+ type SetupStartedObserver interface

+ OnPipelineSetupStarted func(at time.Time)

+ type StatelessTextTransformer struct

+ func NewStatelessTextTransformer(name string, transform func(string) string) *StatelessTextTransformer

+ func (p *StatelessTextTransformer) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type TurnTracker interface

+ OnTurnEnded func(fn func(turn int, duration time.Duration, interrupted bool))

+ type WakeNotifierFilter struct

+ func NewWakeNotifierFilter(name string, notifier notify.Notifier, filter func(frames.Frame) bool, ...) *WakeNotifierFilter

+ func (f *WakeNotifierFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

v0.0.5

Jul 13, 2026

v0.0.4

Jul 12, 2026

v0.0.3

Jul 12, 2026

v0.0.2

Jul 2, 2026

v0.0.1

Jun 29, 2026

Changes in this version

+ type Base struct

+ func New(name string, self Processor, opts ...Option) *Base

+ func (b *Base) Cleanup(ctx context.Context) error

+ func (b *Base) Clock() clock.Clock

+ func (b *Base) ID() uint64

+ func (b *Base) Link(next Processor)

+ func (b *Base) MetricsEnabled() bool

+ func (b *Base) Name() string

+ func (b *Base) Next() Processor

+ func (b *Base) Prev() Processor

+ func (b *Base) ProcessFrame(ctx context.Context, f frames.Frame, dir Direction) error

+ func (b *Base) PushError(ctx context.Context, msg string, err error, fatal bool)

+ func (b *Base) PushFrame(ctx context.Context, f frames.Frame, dir Direction) error

+ func (b *Base) QueueFrame(ctx context.Context, f frames.Frame, dir Direction) error

+ func (b *Base) Setup(ctx context.Context, s Setup) error

+ func (b *Base) UsageMetricsEnabled() bool

+ type Direction int

+ const Downstream

+ const Upstream

+ func (d Direction) String() string

+ type FilterFunc func(frames.Frame) bool

+ type FunctionFilter struct

+ func NewFunctionFilter(name string, dir Direction, allow FilterFunc) *FunctionFilter

+ func (f *FunctionFilter) ProcessFrame(ctx context.Context, frame frames.Frame, dir Direction) error

+ type HandlerFunc func(ctx context.Context, f frames.Frame, dir Direction) error

+ type Option func(*Base)

+ func WithDirectMode() Option

+ type Processor interface

+ Cleanup func(ctx context.Context) error

+ ID func() uint64

+ Link func(next Processor)

+ Name func() string

+ Next func() Processor

+ Prev func() Processor

+ ProcessFrame func(ctx context.Context, f frames.Frame, dir Direction) error

+ PushFrame func(ctx context.Context, f frames.Frame, dir Direction) error

+ QueueFrame func(ctx context.Context, f frames.Frame, dir Direction) error

+ Setup func(ctx context.Context, s Setup) error

+ func NewSink(name string, downstream HandlerFunc) Processor

+ func NewSource(name string, upstream HandlerFunc) Processor

+ type Setup struct

+ Clock clock.Clock