pipeline

package
v0.69.2 Latest Latest
Warning

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

 Go to latest Published: Sep 6, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package pipeline is the composable capture-to-transcript engine of SpeechKit. It implements the contracts declared in the root speechkit package — speechkit.AudioRecorder, speechkit.SegmentCollector, speechkit.Transcriber, speechkit.TranscriptOutput — and wires them into three reusable building blocks:

Hosts that only need the ready-made dictation flow should use github.com/kombifyio/SpeechKit/pkg/speechkit/dictation; this package is for hosts that compose their own pipeline or replace one stage.

Index

Constants

View Source 
const (
	DefaultDictationPause                  = 1500 * time.Millisecond
	DefaultDictationMinSegment             = 1200 * time.Millisecond
	DefaultDictationMinIntermediateSegment = 6 * time.Second
	DefaultDictationParagraphPause         = 4 * time.Second
	DefaultDictationPadding                = 480 * time.Millisecond
	DefaultDictationOverlap                = 200 * time.Millisecond
	// DefaultDictationMaxUtterance is the longest stretch of continuous speech
	// the segmenter lets accumulate before it closes the segment without
	// waiting for a pause. Zero disables the cap; hosts that stream segments
	// live (meeting capture) set one, because a call with no 1.5-second pause
	// otherwise yields one 13-minute transcript piece that no downstream
	// model window fits and that the notes cannot show until it ends.
	DefaultDictationMaxUtterance = 0 * time.Second
)
View Source 
const (
	// LiveCommitImmediate pastes each provider-final as soon as it arrives.
	LiveCommitImmediate = "immediate"
	// LiveCommitPhrase waits for one sentence (or a short pause) before paste.
	LiveCommitPhrase = "phrase"
	// LiveCommitPassage waits for about two sentences (or a longer pause) so
	// live field injection reads as prose rather than breath-sized fragments.
	LiveCommitPassage = "passage"
)

Variables

View Source 
var (
	ErrMissingRunner      = errors.New("speechkit: transcription worker requires a runner")
	ErrMissingTranscriber = errors.New("speechkit: transcription runner requires a transcriber")
	ErrWorkerClosed       = errors.New("speechkit: transcription worker is closed")
	ErrWorkerQueueFull    = errors.New("speechkit: transcription worker queue is full")
)

Functions

func CountTerminalSentences 

func CountTerminalSentences(text string) int

CountTerminalSentences counts fragments that already look like finished sentences so live commit can wait for a short paragraph instead of a breath.

func FallbackDictationSegments 

func FallbackDictationSegments(fullPCM []byte) []speechkit.AudioSegment

FallbackDictationSegments wraps all of fullPCM in a single segment. Used when VAD-based segmentation is unavailable or produces no output.

func JoinTranscriptFragments 

func JoinTranscriptFragments(parts ...string) string

JoinTranscriptFragments concatenates live transcript slices with a single separating space when the next slice would otherwise glue onto the previous word or sentence.

func LiveInjectFragment 

func LiveInjectFragment(previousSession uint64, previousTail, next string, session uint64) (fragment, tail string, nextSession uint64)

LiveInjectFragment returns the text that should be pasted for this fragment so consecutive live injects keep a word gap without rewriting earlier text.

func NeedsLiveInjectSpace 

func NeedsLiveInjectSpace(prev, next string) bool

NeedsLiveInjectSpace reports whether two adjacent live fragments need a separating space when pasted one after the other.

Types

type DictationSegmenter 

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

DictationSegmenter implements [SegmentCollector] using VAD-based pause detection to split continuous speech into discrete segments.

func NewDictationSegmenter 

func NewDictationSegmenter(detector speechkit.VoiceActivityDetector, pauseThreshold time.Duration) *DictationSegmenter

func (*DictationSegmenter) CollectStopSegments 

func (s *DictationSegmenter) CollectStopSegments(fullPCM []byte) ([]speechkit.AudioSegment, error)

func (*DictationSegmenter) DrainReadySegments 

func (s *DictationSegmenter) DrainReadySegments() []speechkit.AudioSegment

DrainReadySegments returns pause-bounded intermediate segments that were completed during FeedPCM calls. It leaves the active utterance in place so a later Stop() only flushes the remaining tail.

func (*DictationSegmenter) FeedPCM 

func (s *DictationSegmenter) FeedPCM(pcm []byte) error

func (*DictationSegmenter) IdleAudio 

func (s *DictationSegmenter) IdleAudio() (time.Duration, time.Time)

IdleAudio reports silence measured in audio time rather than wall-clock time: the cumulative duration of processed silent frames since the last detected speech, plus the wall-clock time of the most recently processed PCM frame. While speech is in progress the silence duration is zero.

Audio-time anchoring makes silence-based auto-stop robust against CPU starvation: when frame delivery stalls, the silence counter freezes instead of counting real seconds against a stale timestamp.

Satisfies the [AudioIdleObserver] contract consumed by RecordingController; preferred over [IdleSince] when available.

func (*DictationSegmenter) IdleSince 

func (s *DictationSegmenter) IdleSince() time.Time

IdleSince returns the wall-clock time at which the segmenter most recently transitioned out of speech (or, for a fresh session that has not yet seen speech, the construction time). Returns the zero value when speech is currently being captured — the poller treats zero as "user is actively speaking, silence timer should reset."

Satisfies the [IdleObserver] contract consumed by RecordingController to drive silence-based auto-stop.

func (*DictationSegmenter) SetMaxUtterance added in v0.68.6 

func (s *DictationSegmenter) SetMaxUtterance(d time.Duration)

SetMaxUtterance caps how long continuous speech may run before the segmenter emits it as an intermediate segment even without a pause. <=0 disables the cap (the default).

func (*DictationSegmenter) SetMinIntermediateSegment 

func (s *DictationSegmenter) SetMinIntermediateSegment(d time.Duration)

SetMinIntermediateSegment configures the minimum active utterance duration that can be emitted before Stop(). Shorter utterances stay merged across natural pauses so dictation does not over-fragment; <=0 emits on every pause-bounded segment.

type LiveCommitFlusher 

type LiveCommitFlusher interface {
	FlushLiveCommit(ctx context.Context) error
}

LiveCommitFlusher drains a grouped live-commit buffer. Recording stop calls this so a trailing sentence is not left behind when the stream ends.

type LiveCommitPolicy 

type LiveCommitPolicy struct {
	Mode         string
	MinSentences int
	Hold         time.Duration
}

LiveCommitPolicy groups provider-native finals before they reach a sink. Overlay drafts still pass through immediately.

func NormalizeLiveCommitPolicy 

func NormalizeLiveCommitPolicy(mode string) LiveCommitPolicy

NormalizeLiveCommitPolicy maps a host mode onto hold/sentence defaults. Empty or unknown modes disable grouping so existing hosts stay immediate.

type PooledPCMRecorder 

type PooledPCMRecorder interface {
	SetPooledPCMHandler(func(buf []byte, release func()))
}

PooledPCMRecorder is optionally implemented by AudioRecorders whose backend leases per-frame buffers from a pool instead of allocating a fresh copy per frame (~33 allocations/sec during capture). When the recorder satisfies this interface the controller installs the pool-aware handler and releases each buffer as soon as the frame has been fed to the collector/stream — the controller never retains a frame. Structurally matches internal/audio's SetPooledPCMHandler.

type ReadySegmentCollector 

type ReadySegmentCollector interface {
	speechkit.SegmentCollector
	DrainReadySegments() []speechkit.AudioSegment
}

ReadySegmentCollector is implemented by collectors that can hand completed pause-bounded segments to the transcription queue before recording stops.

type RecordingController 

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

RecordingController manages the start/stop lifecycle of a single recording session and hands audio segments to the submission queue.

func NewRecordingController 

func NewRecordingController(recorder speechkit.AudioRecorder, submitter speechkit.JobSubmitter, observer speechkit.RecordingObserver, segmenterFactory speechkit.SegmentCollectorFactory) *RecordingController

func (*RecordingController) Cancel 

func (c *RecordingController) Cancel(opts speechkit.RecordingCancelOptions) error

Cancel stops the active recorder and discards the captured audio. Hosts use this when the user switches modes mid-capture; submitting the old buffer would deliver stale speech through the newly selected mode.

func (*RecordingController) IsCapturing 

func (c *RecordingController) IsCapturing() bool

IsCapturing reports whether the microphone is physically open. Unlike [IsRecording] it excludes the stop/drain window, so hosts can distinguish "user is still dictating" (suppress post-capture UI states) from "capture ended, transcription in flight" (terminal states must display).

func (*RecordingController) IsRecording 

func (c *RecordingController) IsRecording() bool

func (*RecordingController) SetDictationStream 

func (c *RecordingController) SetDictationStream(provider speechkit.DictationStreamProvider, sink speechkit.DictationStreamSink)

SetDictationStream configures the optional provider-native live dictation path. Hosts can leave this unset to keep the public full-capture default.

func (*RecordingController) SetFragmentSegments 

func (c *RecordingController) SetFragmentSegments(enabled bool)

SetFragmentSegments controls whether Stop() submits the VAD-derived segments as the STT source (true) or the full captured audio as a single submission (false, default). The segmenter still drives silence-based auto-stop either way; this only changes what audio is sent to transcription.

func (*RecordingController) SetIdleWatchInterval 

func (c *RecordingController) SetIdleWatchInterval(d time.Duration)

SetIdleWatchInterval overrides the polling interval used by the silence-based auto-stop watcher. Tests use this to keep the unit tests fast (e.g. 5ms polling). Production should never touch this.

func (*RecordingController) SetStreamSegments 

func (c *RecordingController) SetStreamSegments(enabled bool)

SetStreamSegments controls whether completed pause-bounded segments are submitted during recording instead of waiting until Stop(). The default is false. This is intended for live-ish dictation surfaces; hosts that need the strongest protection against VAD excision should keep the full-capture default.

func (*RecordingController) Start 

func (c *RecordingController) Start(opts speechkit.RecordingStartOptions) error

func (*RecordingController) Stop 

func (c *RecordingController) Stop(opts speechkit.RecordingStopOptions) error

type TranscriptSessionLedger 

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

TranscriptSessionLedger suppresses duplicate final commits for progressive transcription. It tracks both in-flight and completed segments so repeated Stop/Finalize/provider events cannot paste the same text twice.

func NewTranscriptSessionLedger 

func NewTranscriptSessionLedger() *TranscriptSessionLedger

func (*TranscriptSessionLedger) Begin 

func (l *TranscriptSessionLedger) Begin(key speechkit.TranscriptSegmentKey) bool

func (*TranscriptSessionLedger) Commit 

func (l *TranscriptSessionLedger) Commit(key speechkit.TranscriptSegmentKey)

func (*TranscriptSessionLedger) EndRecordingSession 

func (l *TranscriptSessionLedger) EndRecordingSession(recordingSessionID int64)

EndRecordingSession releases deduplication state once a durable recording session has ended. Controllers reconstructed while the recording is active still share the ledger; completed meetings cannot grow it without bound.

func (*TranscriptSessionLedger) Release 

func (l *TranscriptSessionLedger) Release(key speechkit.TranscriptSegmentKey)

type TranscriptionRunner 

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

TranscriptionRunner transcribes audio submissions and persists results. Create one with NewTranscriptionRunner.

func NewTranscriptionRunner 

func NewTranscriptionRunner(transcriber speechkit.Transcriber, store speechkit.Persistence) *TranscriptionRunner

NewTranscriptionRunner creates a TranscriptionRunner backed by the given transcriber and persistence store. Either argument may be nil.

func (*TranscriptionRunner) Commit 

func (r *TranscriptionRunner) Commit(ctx context.Context, submission speechkit.Submission, transcript speechkit.Transcript) (speechkit.Completion, error)

func (*TranscriptionRunner) WithObserver 

func (r *TranscriptionRunner) WithObserver(observer speechkit.CommitObserver) *TranscriptionRunner

type TranscriptionWorker 

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

TranscriptionWorker processes [TranscriptionJob] values from an internal queue on a single goroutine. Start it with TranscriptionWorker.Start and submit work with TranscriptionWorker.Submit.

func NewTranscriptionWorker 

func NewTranscriptionWorker(cfg TranscriptionWorkerConfig) (*TranscriptionWorker, error)

func (*TranscriptionWorker) Close 

func (w *TranscriptionWorker) Close()

func (*TranscriptionWorker) EndRecordingSession 

func (w *TranscriptionWorker) EndRecordingSession(recordingSessionID int64)

EndRecordingSession releases deduplication history for a durable recording after its capture lifecycle has ended.

func (*TranscriptionWorker) HandleDictationStreamEvent 

func (w *TranscriptionWorker) HandleDictationStreamEvent(ctx context.Context, event speechkit.DictationStreamEvent, opts speechkit.DictationStreamSinkOptions) error

HandleDictationStreamEvent routes provider-native dictation events through the same final-commit path as batch transcription. Interim/draft events are UI/status-only and must never call output or persistence.

func (*TranscriptionWorker) Start 

func (w *TranscriptionWorker) Start(ctx context.Context)

func (*TranscriptionWorker) Submit 

func (w *TranscriptionWorker) Submit(job speechkit.TranscriptionJob) error

func (*TranscriptionWorker) Wait 

func (w *TranscriptionWorker) Wait()

type TranscriptionWorkerConfig 

type TranscriptionWorkerConfig struct {
	Timeout     time.Duration
	QueueSize   int
	Runner      *TranscriptionRunner
	Output      speechkit.TranscriptOutput
	Interceptor speechkit.TranscriptInterceptor
	Transformer speechkit.TranscriptTransformer
	Observer    speechkit.TranscriptionObserver
	Ledger      *TranscriptSessionLedger
	// LowConfidenceThreshold flags recognized words below this acoustic
	// confidence (0..1) so the host can surface likely-misrecognized terms.
	// <= 0 disables the check. Only providers that expose per-word confidence
	// (Deepgram, AssemblyAI) produce data here.
	LowConfidenceThreshold float64
}

TranscriptionWorkerConfig configures a TranscriptionWorker. Runner is required; all other fields are optional.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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