processor

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 30, 2026 License: BSD-2-Clause Imports: 15 Imported by: 0

Documentation

Overview

Package processor defines the frame processor: the building block of a jargo pipeline. Processors link into a chain, receive frames, process them, and push them on to the next or previous processor. Each processor handles system frames with priority, processes data and control frames in order on its own goroutine, and can be interrupted.

Index

Constants

View Source 
const (
	// EventUsableChanged fires with the new value of Usable whenever a
	// processor stops or starts being able to do its job.
	//
	//	events.On(p.Events(), processor.EventUsableChanged,
	//	    func(ctx context.Context, usable bool) { … })
	EventUsableChanged = "on_usable_changed"
	// EventError fires with the error frame a processor reports, before the
	// frame travels. A handler reading Source.Usable() therefore sees the
	// verdict that came with the error it is handling.
	EventError = "on_error"
	// EventBeforeProcessFrame fires with a frame this processor is about to
	// handle, before it has been handled.
	EventBeforeProcessFrame = "on_before_process_frame"
	// EventAfterProcessFrame fires with a frame this processor has handled. It
	// does not fire for a frame whose handling failed, which raises EventError
	// instead.
	EventAfterProcessFrame = "on_after_process_frame"
	// EventBeforePushFrame fires with a frame this processor is about to send to
	// a neighbor, before the neighbor has it.
	EventBeforePushFrame = "on_before_push_frame"
	// EventAfterPushFrame fires with a frame this processor has sent to a
	// neighbor.
	EventAfterPushFrame = "on_after_push_frame"
)

The events every processor raises.

View Source 
const DefaultPauseUntilReadyTimeout = 5 * time.Second

DefaultPauseUntilReadyTimeout is how long a processor holds frames waiting for a readiness condition before giving up and resuming. See PauseProcessingAllFramesUntil.

Variables

This section is empty.

Functions

func FrameIs added in v0.1.0 

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

FrameIs builds a matcher for one frame type, for passing to NewFrameFilter.

func IdentityTransformer added in v0.1.0 

func IdentityTransformer(f frames.Frame) frames.Frame

IdentityTransformer passes a frame along unchanged. It is what a producer or consumer built without a transformer uses.

func IsSource added in v0.1.0 

func IsSource(p Processor) bool

IsSource reports whether p is a pipeline source: the endpoint a pipeline wraps the head of its chain in so frames can be fed in and taken out at the edges.

It is plumbing rather than a step of the pipeline's work, which is what an observer measuring what each processor cost uses to leave it out of the measurement.

func NotifyProcessorSetup added in v0.1.0 

func NotifyProcessorSetup(observers []Observer, data ProcessorSetUp)

NotifyProcessorSetup reports a processor having been set up to every observer listening for it. A pipeline calls it as each of its processors is set up.

Types

type Base 

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

Base implements Processor. Embed it in a concrete processor and pass the concrete value as self so the base can dispatch to the overridden ProcessFrame: 

type Echo struct{ *processor.Base }

func NewEcho() *Echo {
    e := &Echo{}
    e.Base = processor.New("Echo", e)
    return e
}

func (e *Echo) ProcessFrame(ctx context.Context, f frames.Frame, dir processor.Direction) error {
    if err := e.Base.ProcessFrame(ctx, f, dir); err != nil {
        return err
    }
    return e.PushFrame(ctx, f, dir)
}

func New 

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

New builds a Base named name. self is the embedding processor, used to dispatch to its ProcessFrame; pass nil for a plain pass-through that does not override ProcessFrame.

func (*Base) AudioInSampleRate added in v0.1.0 

func (b *Base) AudioInSampleRate() int

AudioInSampleRate is the pipeline's input audio sample rate in Hz, which a service that was not given one of its own takes as its own. It is known from the moment the processor is set up.

func (*Base) AudioOutSampleRate added in v0.1.0 

func (b *Base) AudioOutSampleRate() int

AudioOutSampleRate is the pipeline's output audio sample rate in Hz. See AudioInSampleRate.

func (*Base) BeginTTFB added in v0.1.0 

func (b *Base) BeginTTFB() bool

BeginTTFB reports whether a time-to-first-byte measurement should be started, and records that one was. A service calls it where it would start the clock, and measures nothing when it reports false.

It answers true every time unless the StartFrame asked for only the initial TTFB, in which case the first measurement is the only one armed: a caller who wants the figure the call opened with gets it once rather than on every turn.

func (*Base) Broadcast added in v0.1.0 

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

Broadcast sends a frame both downstream and upstream, so an event that the whole pipeline has to see reaches processors on either side of this one.

build is called once per direction: the two halves are distinct frames paired by BroadcastSiblingID. The directions are processed on separate goroutines, so a single shared frame would be mutated concurrently, and a consumer that sees both halves can recognize the pair rather than reporting the event twice.

func (*Base) BroadcastInterruption added in v0.1.0 

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

BroadcastInterruption interrupts the pipeline from this processor: it drops the work this processor had queued and sends an InterruptionFrame both ways, so every processor on either side hears that the turn was cut off.

Use it wherever something other than the user's voice interrupts the bot: a client typing over it, a tool deciding the answer is stale, a supervisor stopping the turn.

It returns as soon as the frames are away, and this processor's own goroutine is left running, so a caller can carry on and push what it interrupted for. That is why the queue is emptied rather than the goroutine canceled: canceling here would cancel the call asking for the interruption.

func (*Base) CanGenerateMetrics added in v0.1.0 

func (b *Base) CanGenerateMetrics() bool

CanGenerateMetrics implements Processor. A processor reports no metrics unless it is a service, which overrides this.

func (*Base) Cleanup 

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

Cleanup implements Processor. It stops the process and input goroutines and waits for the event handlers still running, so a caller reading what a handler collected does not race it.

func (*Base) Clock 

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

Clock returns the pipeline clock, available after Setup.

func (*Base) EntryProcessors added in v0.1.0 

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

EntryProcessors implements Processor. A plain processor has none; a compound processor overrides this.

func (*Base) Events added in v0.1.0 

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

Events returns the registry of events this processor raises, for attaching handlers to them.

func (*Base) FlushPipeline added in v0.1.0 

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

FlushPipeline blocks until every frame queued ahead of the call has traveled the whole pipeline, along with anything a processor started by pushing upstream. Use it to let the pipeline settle, after an interruption say, before injecting new work.

The wait is bounded by the flush itself, which gives up once the pipeline has gone quiet without the probe coming back; a pipeline that keeps working keeps the wait alive however long it takes.

Never call it from the goroutine that processes frames: the probe it waits on has to pass through this processor to complete its trip, so a processor blocking its own frame path would wait forever. A processor that needs this runs it from a goroutine of its own.

It is a no-op for a processor driven outside a pipeline worker, which has nothing to drain.

func (*Base) HasQueuedFrame added in v0.1.0 

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

HasQueuedFrame reports whether a frame satisfying match is still waiting in this processor's in-order queue, behind the one being handled now. A processor uses it to tell that more of the same work is already on its way, so it can hold off on an action until the last of it arrives rather than repeating the action once per frame.

Only data and control frames are considered. A system frame is handled the moment it is queued and so never waits.

func (*Base) ID 

func (b *Base) ID() uint64

ID implements Processor. 

func (b *Base) Link(next Processor)

Link implements Processor.

func (*Base) MetricsEnabled 

func (b *Base) MetricsEnabled() bool

MetricsEnabled reports whether performance-metrics collection was enabled by the StartFrame. It is valid once the processor has received its StartFrame.

func (*Base) Name 

func (b *Base) Name() string

Name implements Processor.

func (*Base) Next 

func (b *Base) Next() Processor

Next implements Processor.

func (*Base) PauseProcessingAllFramesUntil added in v0.1.0 

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

PauseProcessingAllFramesUntil holds the frames arriving at this processor until ready returns.

It is for a processor that cannot act on frames until some condition holds, such as one opening a connection in the background. The frames wait in the processor's queues and are handled in order once the condition resolves, so nothing is lost.

The frame being handled when this is called is unaffected: the pause takes hold from the next frame on. A processor pausing while it handles its StartFrame still passes that frame downstream, so starting the pipeline is not delayed.

Both queues are held, so a processor left paused could not handle the frames that shut it down. The pause is therefore always lifted: when ready returns, when timeout elapses, or at cleanup, whichever comes first. A timeout of zero takes DefaultPauseUntilReadyTimeout.

ready is called on a goroutine of its own and should return once the processor can work, or when ctx ends.

func (*Base) PauseProcessingFrames added in v0.1.0 

func (b *Base) PauseProcessingFrames()

PauseProcessingFrames holds the processor before it handles its next data or control frame. Held frames stay queued, in order, and are handled once ResumeProcessingFrames is called. System frames are unaffected; pause those with PauseProcessingSystemFrames.

The pause is one-shot: a resume releases the processor, which then keeps handling frames until it is paused again.

func (*Base) PauseProcessingSystemFrames added in v0.1.0 

func (b *Base) PauseProcessingSystemFrames()

PauseProcessingSystemFrames holds the processor before it handles its next system frame. Because the same goroutine feeds the queue that data and control frames are handled from, holding it also stops those frames being handed on.

func (*Base) Prev 

func (b *Base) Prev() Processor

Prev implements Processor.

func (*Base) ProcessFrame 

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

ProcessFrame implements Processor. It handles the system frames that drive a processor's lifecycle: StartFrame, InterruptionFrame and CancelFrame. A concrete processor overrides this, calls the base first, then forwards the frame with PushFrame.

func (*Base) Processors added in v0.1.0 

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

Processors implements Processor. A plain processor contains none; a compound processor overrides this.

func (*Base) ProcessorsWithMetrics added in v0.1.0 

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

ProcessorsWithMetrics implements Processor. A plain processor contains nothing, so it reports nothing; a compound processor overrides this and collects from what it contains.

func (*Base) PushError 

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

PushError builds an ErrorFrame for msg and pushes it upstream.

func (*Base) PushErrorFrame added in v0.1.0 

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

PushErrorFrame settles the error frame's category and the processor's usability, tells the error handlers, and pushes the frame upstream.

forceTreatAsPermanent reports the error as one that will keep recurring. Leaving it false does not keep the processor usable: a permanent category costs it its usability either way.

func (*Base) PushFrame 

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

PushFrame implements Processor. It forwards a frame to the neighbor in dir.

A frame pushed at a neighbor that has not started yet is not dropped: it waits in that neighbor's queue until its StartFrame arrives, which is what lets a processor that connects while the pipeline is being set up push what it receives straight away.

func (*Base) PushTokenUsage added in v0.1.0 

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

PushTokenUsage reports LLM token usage measured by a service that does not run through the LLM base: a realtime (speech-to-speech) service that receives a usage event from its provider. It records the aggregate token counts as metrics and emits a MetricsFrame downstream for in-band consumers (e.g. an RTVI client). The caller passes the model id and gates the call on UsageMetricsEnabled, so the conversion from the provider's usage shape happens only when metrics are collected.

The gen_ai.usage.* attributes go on the span already covering the work the usage was measured for, which the caller supplies through ctx. A service that reports usage outside any span of its own records it in metrics alone: usage belongs to the operation that incurred it, and a span raised here just to hold it would have nothing to say about what the model actually did.

func (*Base) QueueFrame 

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

QueueFrame implements Processor.

func (*Base) ResumeProcessingFrames added in v0.1.0 

func (b *Base) ResumeProcessingFrames()

ResumeProcessingFrames releases a processor paused with PauseProcessingFrames.

func (*Base) ResumeProcessingSystemFrames added in v0.1.0 

func (b *Base) ResumeProcessingSystemFrames()

ResumeProcessingSystemFrames releases a processor paused with PauseProcessingSystemFrames.

func (*Base) Running added in v0.1.0 

func (b *Base) Running() Running

Running is the pipeline this processor is part of, or nil for a processor driven outside one.

func (*Base) Self added in v0.1.0 

func (b *Base) Self() Processor

Self is the processor this base belongs to: the concrete value passed to New, or the base itself when none was.

Push through it, rather than through the embedded base, whenever a frame leaving a processor should go through whatever the outer type does on its way out. A type embedding another processor overrides PushFrame to inspect, rewrite or drop what leaves it, and a push made on the inner value would go straight past that.

func (*Base) SetUsable added in v0.1.0 

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

SetUsable sets whether this processor can be given work, raising EventUsableChanged when the value moves.

Call it to bring back a processor that became unusable, once whatever stopped it working has been dealt with: new credentials, or a provider that has come back up. Services also do this for themselves when their settings change, since new settings may be the fix.

func (*Base) Setup 

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

Setup implements Processor. It stores the shared components. The goroutines are not started here: nothing is drained until the StartFrame arrives, so a processor never acts on a frame before it has been started.

func (*Base) StartSpan added in v0.1.0 

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

StartSpan opens a span for work this processor is doing, parented to the turn being spoken (or to the conversation between turns), and returns it with a context carrying it.

On a pipeline that is not traced the span is a no-op and nothing is recorded or exported, so a caller opens one unconditionally and sets attributes on it without guarding: the cost of an untraced pipeline is the branch taken here.

func (*Base) Tracing added in v0.1.0 

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

Tracing returns the session's tracing state, available after Setup. It is nil when the pipeline is not traced, which its methods handle: parent a span with Tracing().Parent(ctx) without checking.

func (*Base) TracingEnabled added in v0.1.0 

func (b *Base) TracingEnabled() bool

TracingEnabled reports whether this pipeline is traced. A processor that raises spans of its own opens them through StartSpan, which checks this.

func (*Base) TrackFlushProbe added in v0.1.0 

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

TrackFlushProbe reports a flush probe from another worker entering the pipeline this processor runs in, so the worker waiting on it hears that this pipeline is working on it. It is a no-op for a processor driven outside a pipeline worker, which answers no probes.

func (*Base) TypeName added in v0.1.0 

func (b *Base) TypeName() string

TypeName is the name the processor was built with, without the instance number Name appends ("OpenAILLM", where Name is "OpenAILLM#3"). It names the kind of processor rather than this one, which is what identifies the provider behind a service on its spans and what a metric is grouped by.

func (*Base) Usable added in v0.1.0 

func (b *Base) Usable() bool

Usable reports whether this processor can still do its job.

A processor stays usable through failures it might recover from, and becomes unusable once its work can no longer succeed: a provider has rejected its API key, model or voice, or it has failed enough times to stop trying. Sending it more work would only produce more of the same error, so services stop accepting work and stop reconnecting once this is false.

Errors set this as they are reported, so an error handler reading ErrorFrame.Source.Usable() sees the verdict that came with the error it is handling.

func (*Base) UsageMetricsEnabled 

func (b *Base) UsageMetricsEnabled() bool

UsageMetricsEnabled reports whether usage-metrics collection was enabled by the StartFrame. It is valid once the processor has received its StartFrame.

type ConsumerOption added in v0.1.0 

type ConsumerOption func(*ConsumerProcessor)

ConsumerOption configures a ConsumerProcessor.

func WithConsumerDirection added in v0.1.0 

func WithConsumerDirection(d Direction) ConsumerOption

WithConsumerDirection sets which way the consumed frames travel. They go downstream by default.

func WithConsumerTransformer added in v0.1.0 

func WithConsumerTransformer(t FrameTransformer) ConsumerOption

WithConsumerTransformer rewrites each frame before it is put into the pipeline.

type ConsumerProcessor added in v0.1.0 

type ConsumerProcessor struct {
	*Base
	// contains filtered or unexported fields
}

ConsumerProcessor puts the frames a ProducerProcessor picked into the pipeline at the point it sits, while passing everything reaching it along untouched.

func NewConsumerProcessor added in v0.1.0 

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

NewConsumerProcessor builds a consumer of the frames producer picks.

func (*ConsumerProcessor) Cleanup added in v0.1.0 

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

Cleanup implements Processor.

func (*ConsumerProcessor) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type Direction 

type Direction int

Direction is the direction a frame flows through the pipeline. 

const (
	// Downstream is the direction from input toward output.
	Downstream Direction = iota
	// Upstream is the direction from output back toward input.
	Upstream
)

func (Direction) String 

func (d Direction) String() string

String returns "downstream" or "upstream".

type ErrorClassifier added in v0.1.0 

type ErrorClassifier interface {
	// ClassifyError returns the category of err, or the zero category to let
	// the shared classification decide.
	ClassifyError(err error) errs.Category
}

ErrorClassifier is implemented by a processor that knows the shape of the failures its provider raises, so it can say what one means where the shared classification cannot.

Implement it on a service whose provider signals failures through errors of its own rather than an HTTP status, or whose credentials can be rejected for a reason a reconnection would clear. Returning the zero category falls back to the shared classification.

type ErrorOption added in v0.1.0 

type ErrorOption func(*errorOptions)

ErrorOption adjusts what PushError reports beyond the message. The options stand for the arguments upstream passes by keyword, so an ordinary failure stays a four-argument call.

func ForceTreatAsPermanent added in v0.1.0 

func ForceTreatAsPermanent() ErrorOption

ForceTreatAsPermanent reports the error as one that will keep recurring, leaving the processor unable to do any more work: having failed too many times to keep trying, say. It is only needed for a failure the category does not already convey, since a permanent category costs the processor its usability on its own.

func WithErrorCategory added in v0.1.0 

func WithErrorCategory(c errs.Category) ErrorOption

WithErrorCategory reports why the error occurred, when the caller knows. Leaving it unset lets the category be worked out from the error; passing errors.Unknown reports a failure whose cause cannot be attributed, an unexpected one caught by a catch-all, say, which may not have come from this processor at all.

type FilterFunc 

type FilterFunc func(frames.Frame) bool

FilterFunc reports whether a frame is allowed to pass through a FunctionFilter.

type FilterOption added in v0.1.0 

type FilterOption func(*FunctionFilter)

FilterOption configures a FunctionFilter.

func WithFilterSystemFrames added in v0.1.0 

func WithFilterSystemFrames() FilterOption

WithFilterSystemFrames has the predicate decide system frames too, rather than passing them through. The lifecycle frames still always pass.

type FrameFilter added in v0.1.0 

type FrameFilter struct {
	*Base
	// contains filtered or unexported fields
}

FrameFilter forwards only the frame types it was built for, and drops the rest.

The lifecycle and system frames always pass whatever the list says. A processor downstream of a filter still has to be started, stopped and told about an interruption, so gating those would break the pipeline rather than filter it.

func NewFrameFilter added in v0.1.0 

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

NewFrameFilter builds a filter passing only the frames that match one of the given types. Build a matcher with FrameIs.

func (*FrameFilter) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type FrameProcessed added in v0.1.0 

type FrameProcessed struct {
	// Processor is the processor handling the frame.
	Processor Processor
	// Frame is the frame being handled.
	Frame frames.Frame
	// Direction is which way it is going.
	Direction Direction
	// Timestamp is when it was handed over, on the pipeline clock.
	Timestamp time.Duration
}

FrameProcessed is one frame reaching a processor.

type FramePushed added in v0.1.0 

type FramePushed struct {
	// Source is the processor sending the frame.
	Source Processor
	// Destination is the processor receiving it.
	Destination Processor
	// Frame is the frame being handed over.
	Frame frames.Frame
	// Direction is which way it is going.
	Direction Direction
	// Timestamp is when it was pushed, on the pipeline clock.
	Timestamp time.Duration
}

FramePushed is one frame moving from one processor to the next.

type FrameTransformer added in v0.1.0 

type FrameTransformer func(frames.Frame) frames.Frame

FrameTransformer rewrites a frame on its way from a producer to a consumer.

type FunctionFilter 

type FunctionFilter struct {
	*Base
	// contains filtered or unexported fields
}

FunctionFilter forwards frames, dropping those a predicate rejects. It runs in direct mode, deciding on the caller's goroutine, and is the building block a ServiceSwitcher uses to gate a branch on or off.

Two kinds of frame are never dropped, whatever the predicate says. The lifecycle frames (start, end and cancel) always pass, because a branch that is gated off still has to be started and shut down with the rest of the pipeline. Every other system frame passes too, unless the filter was built to decide those as well.

func NewFunctionFilter 

func NewFunctionFilter(name string, dir *Direction, allow FilterFunc, opts ...FilterOption) *FunctionFilter

NewFunctionFilter builds a filter that gates frames using allow. dir is the direction the predicate decides; frames traveling the other way pass through untouched. A nil dir has the predicate decide both directions.

func (*FunctionFilter) ProcessFrame 

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

ProcessFrame drops a frame the predicate rejects and forwards everything else. The predicate is consulted for every frame, including the ones that pass regardless, because deciding is how a predicate watching the stream keeps up with it.

type HandlerFunc 

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

HandlerFunc handles a frame that reaches the edge of a pipeline. A source uses it for upstream frames, a sink for downstream frames.

type IdentityFilter added in v0.1.0 

type IdentityFilter struct {
	*Base
}

IdentityFilter forwards every frame unchanged.

It is a processor that does nothing, which is exactly what makes it useful for building a branch of a ParallelPipeline that only has to carry frames through: the branch is then a place in the pipeline rather than a transformation, and no frame should come out of it twice.

func NewIdentityFilter added in v0.1.0 

func NewIdentityFilter(name string) *IdentityFilter

NewIdentityFilter builds an IdentityFilter.

func (*IdentityFilter) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type IdleFrameProcessor added in v0.1.0 

type IdleFrameProcessor struct {
	*Base
	// contains filtered or unexported fields
}

IdleFrameProcessor calls back when nothing it is watching for has come through for a while.

It forwards every frame untouched. What it adds is a clock: each frame it watches for restarts it, and the callback runs whenever the clock runs out. Watching for nothing in particular means any frame restarts it, which measures the pipeline going quiet; naming frame types measures the absence of those, which is how a bot notices a caller who has stopped speaking to it.

The clock keeps running after the callback, so a pipeline that stays idle calls back once per timeout rather than only the first time.

func NewIdleFrameProcessor added in v0.1.0 

func NewIdleFrameProcessor(
	name string,
	timeout time.Duration,
	callback func(*IdleFrameProcessor),
	types ...func(frames.Frame) bool,
) *IdleFrameProcessor

NewIdleFrameProcessor builds a processor calling back when timeout passes with none of types coming through. With no types given, any frame at all restarts the clock. Build a matcher with FrameIs.

func (*IdleFrameProcessor) Cleanup added in v0.1.0 

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

Cleanup implements Processor.

func (*IdleFrameProcessor) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type NullFilter added in v0.1.0 

type NullFilter struct {
	*Base
}

NullFilter drops every frame except the ones the pipeline needs to keep working: the end of the run and the system frames.

It stops a stretch of the pipeline dead without taking it out, which is what makes it useful for holding a branch silent while the pipeline runs.

func NewNullFilter added in v0.1.0 

func NewNullFilter(name string) *NullFilter

NewNullFilter builds a NullFilter.

func (*NullFilter) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type Observer added in v0.1.0 

type Observer interface {
	// OnPushFrame reports one frame handed from one processor to the next.
	OnPushFrame(data FramePushed)
}

Observer watches frames flowing through a pipeline without modifying them, to derive turn, latency or startup metrics, to log the stream, or to report events to a client.

Every handover between processors is reported, not only what reaches the ends of the pipeline, so an observer sees where each frame came from. That is what lets it tell a frame that has been through the output transport, and so carries real playback timing, from the same frame earlier in the pipeline.

Observers must be safe for concurrent use: a pipeline's processors each run on their own goroutine, so the methods may be called from any of them.

type Option 

type Option func(*Base)

Option configures a Base at construction.

func WithDirectMode 

func WithDirectMode() Option

WithDirectMode makes a processor process frames immediately on the caller's goroutine instead of queueing them. It is used for routing processors (a pipeline and its source and sink) that only forward frames.

type PipelineStartedObserver added in v0.1.0 

type PipelineStartedObserver interface {
	Observer
	// OnPipelineStarted reports that the pipeline has started.
	OnPipelineStarted()
}

PipelineStartedObserver is an optional interface an Observer implements to hear that the pipeline has fully started, which is the StartFrame having been handled by every processor, including the branches of a parallel pipeline.

It is reported in order with the frames, so an observer that sets itself up here has done so before the first frame of the conversation reaches it.

type ProcessObserver added in v0.1.0 

type ProcessObserver interface {
	Observer
	// OnProcessFrame reports one frame reaching a processor.
	OnProcessFrame(data FrameProcessed)
}

ProcessObserver is an optional interface an Observer implements to also see a frame as it reaches a processor, before that processor has handled it.

type Processor 

type Processor interface {
	// ID is a process-unique identifier for this processor.
	ID() uint64
	// Name is a human-readable label, "<name>#<id>".
	Name() string

	// Next is the downstream processor, or nil.
	Next() Processor
	// Prev is the upstream processor, or nil.
	Prev() Processor

	// Processors are the sub-processors this processor contains. Only a
	// compound processor (a pipeline, a parallel pipeline) has any; every
	// other processor reports none.
	Processors() []Processor
	// EntryProcessors are the processors a frame entering a compound
	// processor reaches first. A pipeline is a processor itself, so an entry
	// processor can be a pipeline in turn. Every other processor reports none.
	EntryProcessors() []Processor
	// ProcessorsWithMetrics are the processors below this one that report
	// metrics, collected recursively.
	ProcessorsWithMetrics() []Processor
	// CanGenerateMetrics reports whether this processor reports metrics. It is
	// false for everything but a service.
	CanGenerateMetrics() bool
	// Link sets next as this processor's downstream neighbor and this
	// processor as next's upstream neighbor.
	Link(next Processor)

	// Setup wires the processor with shared components and starts its
	// goroutines. It must be called before frames are queued.
	Setup(ctx context.Context, s Setup) error
	// Cleanup stops the processor's goroutines and releases resources.
	Cleanup(ctx context.Context) error

	// QueueFrame hands a frame to this processor for processing.
	QueueFrame(ctx context.Context, f frames.Frame, dir Direction) error
	// ProcessFrame processes a frame. The base implementation handles system
	// lifecycle frames; concrete processors override it and call the base
	// first.
	ProcessFrame(ctx context.Context, f frames.Frame, dir Direction) error
	// PushFrame sends a frame to the neighboring processor in dir.
	PushFrame(ctx context.Context, f frames.Frame, dir Direction) error
	// PushErrorFrame settles the error's category and this processor's
	// usability, tells the error handlers, and pushes the frame upstream.
	PushErrorFrame(ctx context.Context, ef *frames.ErrorFrame, forceTreatAsPermanent bool)

	// Usable reports whether this processor can still do its job. See
	// [Base.Usable].
	Usable() bool
	// SetUsable sets whether this processor can be given work, raising
	// EventUsableChanged when the value moves.
	SetUsable(ctx context.Context, usable bool)
	// Events returns the registry of events this processor raises.
	Events() *events.Registry
}

Processor is a node in a pipeline. Concrete processors embed *Base, which provides every method here except a custom ProcessFrame.

func NewSink 

func NewSink(name string, downstream HandlerFunc) Processor

NewSink returns a pipeline sink. Downstream frames reaching it are passed to downstream; upstream frames are forwarded back along the chain.

func NewSource 

func NewSource(name string, upstream HandlerFunc) Processor

NewSource returns a pipeline source. Upstream frames reaching it are passed to upstream; downstream frames are forwarded along the chain.

type ProcessorSetUp added in v0.1.0 

type ProcessorSetUp struct {
	// Processor is the processor that was set up.
	Processor Processor
	// StartedAt is when the processor's Setup began.
	StartedAt time.Time
	// FinishedAt is when it returned.
	FinishedAt time.Time
}

ProcessorSetUp is one processor having been set up.

Processors are set up concurrently and before any frame flows, so this is what a timing observer measures the work a processor does to get ready by. The times are wall-clock readings carrying a monotonic reading, since the pipeline clock is not what these are offsets from.

func (ProcessorSetUp) Duration added in v0.1.0 

func (d ProcessorSetUp) Duration() time.Duration

Duration is what the processor's Setup cost.

type ProducerOption added in v0.1.0 

type ProducerOption func(*ProducerProcessor)

ProducerOption configures a ProducerProcessor.

func WithProducerTransformer added in v0.1.0 

func WithProducerTransformer(t FrameTransformer) ProducerOption

WithProducerTransformer rewrites each frame on its way to the consumers. The frame carrying on down the pipeline is not the rewritten one.

func WithoutPassthrough added in v0.1.0 

func WithoutPassthrough() ProducerOption

WithoutPassthrough keeps a frame the predicate picked from carrying on down the pipeline, so it reaches the consumers only. A frame the predicate did not pick still passes.

type ProducerProcessor added in v0.1.0 

type ProducerProcessor struct {
	*Base
	// contains filtered or unexported fields
}

ProducerProcessor picks frames out of the stream passing through it and hands copies to any number of consumers elsewhere in the pipeline.

It is how a frame reaches a part of the pipeline it does not flow through: a branch of a ParallelPipeline can watch what another branch is saying without being wired to it. The frames it picks are chosen by a predicate, and may be rewritten on the way out.

func NewProducerProcessor added in v0.1.0 

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

NewProducerProcessor builds a producer handing every frame filter picks to its consumers. By default the frame carries on down the pipeline as well.

func (*ProducerProcessor) AddConsumer added in v0.1.0 

func (p *ProducerProcessor) AddConsumer() *frameQueue

AddConsumer registers a consumer and returns the queue its frames arrive on. A ConsumerProcessor calls it for you when the pipeline starts.

func (*ProducerProcessor) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type Running added in v0.1.0 

type Running interface {
	// Flush blocks until every frame queued ahead of the call has traveled the
	// whole pipeline, so the pipeline has settled. Never call it from the
	// goroutine that processes frames: the probe has to pass through this
	// processor to complete its round trip.
	Flush(ctx context.Context) error
	// TrackFlushProbe reports that a flush probe from another worker has entered
	// this pipeline, so that whoever is waiting on it, out of sight in the
	// pipeline that started it, is told this one is still working. It is called
	// by whatever brings the probe in, since only that knows it is really coming
	// in: every worker on the bus sees it, but most have nowhere to put it.
	TrackFlushProbe(f *frames.PipelineFlushFrame)
	// TurnTracker follows the conversation's turns, and is nil when the worker
	// is not tracking them. A processor that has to know where a turn ended
	// subscribes to it; the interface is narrow because the observer that
	// implements it already depends on this package.
	TurnTracker() TurnTracker
}

Running is the pipeline a processor belongs to, for the few things a processor needs from it that the frame path cannot express on its own. It is an interface rather than the concrete task because the task is built on top of processors, so naming that type here would be circular.

type Setup 

type Setup struct {
	// AudioInSampleRate is the pipeline's input audio sample rate in Hz.
	AudioInSampleRate int
	// AudioOutSampleRate is the pipeline's output audio sample rate in Hz.
	AudioOutSampleRate int
	// Clock is the pipeline clock used for timing.
	Clock clock.Clock
	// EnableMetrics turns on performance metrics collection.
	EnableMetrics bool
	// EnableUsageMetrics turns on usage metrics collection.
	EnableUsageMetrics bool
	// ReportOnlyInitialTTFB asks for the first time-to-first-byte of the run and
	// nothing after it.
	ReportOnlyInitialTTFB bool
	// Observers watch every frame handed between processors.
	Observers []Observer
	// Running is the pipeline this processor is part of. Nil for a processor
	// driven outside a pipeline task.
	Running Running
	// Tracing is the session's tracing state: the conversation span the trace
	// hangs from and the turn being spoken. Processors parent their spans to it,
	// so a span raised away from the frame path still lands under the turn it
	// belongs to. Nil when the pipeline is not traced.
	Tracing *tracing.TracingContext
	// TracingEnabled reports whether this pipeline is traced, and is what the
	// processors gate their spans on. It is separate from Tracing being present
	// because an installed TracerProvider is not on its own a request to trace
	// the pipeline: an application that traces its own server would otherwise
	// get a service span per turn with nothing to hang it from.
	TracingEnabled bool
}

Setup carries the shared components a processor needs, propagated down the pipeline when it is set up.

type SetupObserver added in v0.1.0 

type SetupObserver interface {
	Observer
	// OnProcessorSetup reports one processor having been set up.
	OnProcessorSetup(data ProcessorSetUp)
}

SetupObserver is an optional interface an Observer implements to also hear that a processor has been set up.

A processor connects and does its other slow start-up work there, so this is where that cost can be measured. Processors are set up concurrently, so these arrive in the order they finish rather than in pipeline order.

type SetupStartedObserver added in v0.1.0 

type SetupStartedObserver interface {
	Observer
	// OnPipelineSetupStarted reports the instant the pipeline began setting its
	// processors up.
	OnPipelineSetupStarted(at time.Time)
}

SetupStartedObserver is an optional interface an Observer implements to hear that the pipeline has begun setting its processors up.

It arrives before any processor has been set up, so an observer timing the start of a session measures from here. Processors connect while they are set up, so this is earlier than the StartFrame and is what the pipeline clock runs from.

type StatelessTextTransformer added in v0.1.0 

type StatelessTextTransformer struct {
	*Base
	// contains filtered or unexported fields
}

StatelessTextTransformer rewrites the text of every TextFrame passing through it, and forwards everything else untouched.

It holds nothing between frames, so it suits a rewrite that depends only on the text in hand: changing case, substituting a term, stripping a marker. A rewrite that has to see a whole sentence belongs in a text aggregator or a filter on the synthesizer instead, where the text has been gathered.

func NewStatelessTextTransformer added in v0.1.0 

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

NewStatelessTextTransformer builds a transformer applying transform to the text of each TextFrame.

func (*StatelessTextTransformer) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

type TurnTracker added in v0.1.0 

type TurnTracker interface {
	// OnTurnEnded adds a listener called when a turn ends, with the turn number,
	// how long it lasted, and whether an interruption cut it short.
	OnTurnEnded(fn func(turn int, duration time.Duration, interrupted bool))
}

TurnTracker reports where the conversation's turns end.

type WakeNotifierFilter added in v0.1.0 

type WakeNotifierFilter struct {
	*Base
	// contains filtered or unexported fields
}

WakeNotifierFilter forwards every frame, and signals a notifier when a frame of one of the types it watches satisfies its predicate.

It is how a condition seen at one point in the pipeline releases something held at another. Nothing is filtered out despite the name: what it decides is whether to signal, not whether the frame goes on.

func NewWakeNotifierFilter added in v0.1.0 

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

NewWakeNotifierFilter builds a filter signaling notifier when a frame matching one of types also satisfies filter. Build a matcher with FrameIs.

func (*WakeNotifierFilter) ProcessFrame added in v0.1.0 

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

ProcessFrame implements Processor.

Source Files

View all Source files

Directories

Path Synopsis
Package aggregators assembles the conversation around an LLM.
 Package aggregators assembles the conversation around an LLM.
Package audiobuffer records a conversation's audio.
 Package audiobuffer records a conversation's audio.
Package dtmf aggregates the DTMF keypresses a caller makes into a string a language model can read.
 Package dtmf aggregates the DTMF keypresses a caller makes into a string a language model can read.
Package ivr navigates automated phone menus (IVR systems).
 Package ivr navigates automated phone menus (IVR systems).
Package langchain bridges an external "chain" — any streaming text generator, such as a LangChain-style runnable or a custom agent — into a jargo pipeline.
 Package langchain bridges an external "chain" — any streaming text generator, such as a LangChain-style runnable or a custom agent — into a jargo pipeline.
Package rtvi implements the RTVI protocol over a transport's messaging channel: a JSON message format and a processor that completes the client handshake and reports pipeline events to the client.
 Package rtvi implements the RTVI protocol over a transport's messaging channel: a JSON message format and a processor that completes the client handshake and reports pipeline events to the client.
Package turns manages the user-turn lifecycle, ported from Pipecat's turns subsystem.
 Package turns manages the user-turn lifecycle, ported from Pipecat's turns subsystem.
Package vadproc is the voice-activity-detection pipeline processor.
 Package vadproc is the voice-activity-detection pipeline processor.
Package voicemail detects whether an outbound call reached a person or a voicemail system.
 Package voicemail detects whether an outbound call reached a person or a voicemail system.

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.