avatar

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: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package avatar implements provider-independent LiveKit avatar sessions and bounded PCM transports.

The canonical inference flow is:

  1. Construct an InferenceSession.
  2. Start it before starting voice.AgentSession.
  3. WaitForJoin until the avatar publishes video.
  4. Start the voice session and always Close the avatar during shutdown.

Rooms created with voice/roomio should pass roomioadapter.ReadyWaiter to the session. The adapter lives in a subpackage so queue/WebSocket avatar plugins do not import the raw-media codec stack during process startup.

Package avatar provides bounded voice I/O and lifecycle primitives for LiveKit avatar workers.

Index

Constants

View Source 
const (
	RPCClearBuffer      = "lk.clear_buffer"
	RPCPlaybackFinished = "lk.playback_finished"
	RPCPlaybackStarted  = "lk.playback_started"
	AudioStreamTopic    = "lk.audio_stream"
)
View Source 
const DefaultClearBufferTimeout = 2 * time.Second
View Source 
const DefaultQueueAudioCapacity = 64
View Source 
const DefaultWaitForJoinTimeout = 30 * time.Second

Variables

View Source 
var (
	ErrDataStreamAudioOutputClosed = errors.New("avatar DataStream audio output is closed")
	ErrAvatarRTCBridgeRequired     = errors.New("avatar RTC bridge is required while waiting for a participant or track")
)
View Source 
var (
	ErrAvatarSessionActive = errors.New("avatar session is already active")
	ErrAvatarJoinTimeout   = errors.New("timed out waiting for avatar participant")
)
View Source 
var ErrQueueAudioOutputClosed = errors.New("avatar queue audio output is closed")

Functions

func MarshalPlaybackFinishedPayload 

func MarshalPlaybackFinishedPayload(event voice.PlaybackFinishedEvent) ([]byte, error)

MarshalPlaybackFinishedPayload emits the Python AvatarRunner's canonical snake_case protocol with playback_position represented in seconds.

func ParsePlaybackFinishedPayload 

func ParsePlaybackFinishedPayload(payload string) voice.PlaybackFinishedEvent

ParsePlaybackFinishedPayload is deliberately total: malformed input yields a safe zero event so an RPC handler can always release WaitForPlayout. The protocol-canonical snake_case fields are preferred, with camelCase accepted for JavaScript producers.

Types

type AudioReceiver 

type AudioReceiver interface {
	stream.Reader[QueueAudioOutputItem]
	Start(context.Context) error
	OnClearBuffer(func(QueueAudioOutputClearEvent)) func()
	NotifyPlaybackStarted(...time.Time)
	NotifyPlaybackFinished(time.Duration, bool) error
	Close() error
}

AudioReceiver is the provider-worker side of queue I/O. DataStream receivers require stronger RTC reader cancellation than server-sdk-go v2.18.1 exposes, while QueueAudioOutput implements this interface without background work.

type AudioSegmentEnd 

type AudioSegmentEnd struct{}

AudioSegmentEnd marks the ordered boundary after one flushed or interrupted audio segment.

type AvatarSession 

type AvatarSession[UserData any] struct {
	// contains filtered or unexported fields
}

AvatarSession owns the provider-independent avatar lifecycle. The generic parameter matches AgentSession user data and enables typed job shutdown registration without a process-global job context.

func NewAvatarSession 

func NewAvatarSession[UserData any](options AvatarSessionOptions) *AvatarSession[UserData]

func (*AvatarSession[UserData]) AClose 

func (s *AvatarSession[UserData]) AClose(ctx context.Context) error

AClose is the Python/TypeScript-compatible spelling.

func (*AvatarSession[UserData]) AvatarIdentity 

func (s *AvatarSession[UserData]) AvatarIdentity() string

func (*AvatarSession[UserData]) Close 

func (s *AvatarSession[UserData]) Close(ctx context.Context) error

Close removes the room participant when credentials/remover are available, then always rolls back local state. Not-found is benign; other removal errors are logged to match the pinned Python/TypeScript lifecycle contract.

func (*AvatarSession[UserData]) OnMetrics 

func (s *AvatarSession[UserData]) OnMetrics(fn func(metrics.Avatar)) func()

func (*AvatarSession[UserData]) Provider 

func (s *AvatarSession[UserData]) Provider() string

func (*AvatarSession[UserData]) RollbackStart 

func (s *AvatarSession[UserData]) RollbackStart(ctx context.Context) error

RollbackStart releases only local lifecycle state. It intentionally does not remove the participant and is safe after an ambiguous gateway create failure.

func (*AvatarSession[UserData]) Start 

func (s *AvatarSession[UserData]) Start(ctx context.Context, agent *voice.AgentSession[UserData], room *lksdk.Room) error

Start attaches lifecycle and metrics tracking. Provider implementations should call this before provisioning and RollbackStart if provisioning fails.

func (*AvatarSession[UserData]) WaitForJoin 

func (s *AvatarSession[UserData]) WaitForJoin(ctx context.Context, options ...WaitForJoinOptions) error

type AvatarSessionOptions 

type AvatarSessionOptions struct {
	AvatarIdentity    string
	Provider          string
	ParentContext     context.Context
	ReadyWaiter       ReadyWaiter
	RemoveParticipant ParticipantRemover
	Logger            *slog.Logger
}

type ByteStreamOptions 

type ByteStreamOptions struct {
	Name                string
	Topic               string
	DestinationIdentity string
	Attributes          map[string]string
}

type ByteStreamWriter 

type ByteStreamWriter interface {
	Write(context.Context, []byte) error
	Close(context.Context) error
}

type DataStreamAudioOutput 

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

DataStreamAudioOutput sends strict little-endian PCM16 in one bounded RTC byte stream per speech segment and uses avatar RPCs for playout accounting.

func NewDataStreamAudioOutput 

func NewDataStreamAudioOutput(options DataStreamAudioOutputOptions) (*DataStreamAudioOutput, error)

func (*DataStreamAudioOutput) AClose 

func (o *DataStreamAudioOutput) AClose(ctx context.Context) error

func (*DataStreamAudioOutput) CanPause 

func (o *DataStreamAudioOutput) CanPause() bool

func (*DataStreamAudioOutput) CaptureFrame 

func (o *DataStreamAudioOutput) CaptureFrame(ctx context.Context, frame agents.AudioFrame) error

func (*DataStreamAudioOutput) CapturedPlayoutSegments 

func (o *DataStreamAudioOutput) CapturedPlayoutSegments() uint64

func (*DataStreamAudioOutput) ClearBuffer 

func (o *DataStreamAudioOutput) ClearBuffer(ctx context.Context) error

func (*DataStreamAudioOutput) Close 

func (o *DataStreamAudioOutput) Close(ctx context.Context) error

func (*DataStreamAudioOutput) Flush 

func (o *DataStreamAudioOutput) Flush(ctx context.Context) error

func (*DataStreamAudioOutput) OnAttached 

func (o *DataStreamAudioOutput) OnAttached()

func (*DataStreamAudioOutput) OnDetached 

func (o *DataStreamAudioOutput) OnDetached()

func (*DataStreamAudioOutput) OnPlaybackFinished 

func (o *DataStreamAudioOutput) OnPlaybackFinished(fn func(voice.PlaybackFinishedEvent)) func()

func (*DataStreamAudioOutput) OnPlaybackStarted 

func (o *DataStreamAudioOutput) OnPlaybackStarted(fn func(voice.PlaybackStartedEvent)) func()

func (*DataStreamAudioOutput) Pause 

func (o *DataStreamAudioOutput) Pause(ctx context.Context) error

func (*DataStreamAudioOutput) PendingPlayoutSegments 

func (o *DataStreamAudioOutput) PendingPlayoutSegments() uint64

func (*DataStreamAudioOutput) Resume 

func (o *DataStreamAudioOutput) Resume(ctx context.Context) error

func (*DataStreamAudioOutput) SampleRate 

func (o *DataStreamAudioOutput) SampleRate() int

func (*DataStreamAudioOutput) SetAttached 

func (o *DataStreamAudioOutput) SetAttached(attached bool)

func (*DataStreamAudioOutput) WaitForPlayout 

func (o *DataStreamAudioOutput) WaitForPlayout(ctx context.Context) (voice.PlaybackFinishedEvent, error)

type DataStreamAudioOutputOptions 

type DataStreamAudioOutputOptions struct {
	Room                *lksdk.Room
	ReadyWaiter         ReadyWaiter
	Transport           DataStreamTransport
	DestinationIdentity string
	SampleRate          int
	WaitRemoteTrack     lksdk.TrackKind
	WaitPlaybackStart   bool
	ClearBufferTimeout  time.Duration
	DisableClearTimeout bool
}

type DataStreamTransport 

type DataStreamTransport interface {
	WaitReady(context.Context, string, lksdk.TrackKind) error
	OpenByteStream(context.Context, ByteStreamOptions) (ByteStreamWriter, error)
	PerformRPC(context.Context, string, string, string) (string, error)
	RegisterRPC(string, string, func(RPCInvocation) string) (func(), error)
}

DataStreamTransport isolates RTC mechanics for deterministic tests and custom room implementations. RegisterRPC must multiplex identities for the same method and return an idempotent unregister function.

type InferenceSession 

type InferenceSession[UserData any] struct {
	// contains filtered or unexported fields
}

InferenceSession is the one-step voice adapter around the cycle-free inference.AvatarSession gateway lifecycle.

func NewInferenceSession 

func NewInferenceSession[UserData any](options InferenceSessionOptions) (*InferenceSession[UserData], error)

func (*InferenceSession[UserData]) AClose 

func (s *InferenceSession[UserData]) AClose(ctx context.Context) error

func (*InferenceSession[UserData]) AudioOutput 

func (s *InferenceSession[UserData]) AudioOutput() *DataStreamAudioOutput

func (*InferenceSession[UserData]) AvatarIdentity 

func (s *InferenceSession[UserData]) AvatarIdentity() string

func (*InferenceSession[UserData]) Base 

func (s *InferenceSession[UserData]) Base() *AvatarSession[UserData]

func (*InferenceSession[UserData]) Close 

func (s *InferenceSession[UserData]) Close(ctx context.Context) error

func (*InferenceSession[UserData]) Gateway 

func (s *InferenceSession[UserData]) Gateway() *inference.AvatarSession

func (*InferenceSession[UserData]) Provider 

func (s *InferenceSession[UserData]) Provider() string

func (*InferenceSession[UserData]) Start 

func (s *InferenceSession[UserData]) Start(ctx context.Context, agent *voice.AgentSession[UserData], room *lksdk.Room, options ...InferenceSessionStartOptions) error

func (*InferenceSession[UserData]) WaitForJoin 

func (s *InferenceSession[UserData]) WaitForJoin(ctx context.Context, options ...WaitForJoinOptions) error

type InferenceSessionOptions 

type InferenceSessionOptions struct {
	Gateway           inference.AvatarSessionOptions
	ParentContext     context.Context
	ReadyWaiter       ReadyWaiter
	RemoveParticipant ParticipantRemover
	Logger            *slog.Logger
}

type InferenceSessionStartOptions 

type InferenceSessionStartOptions struct {
	LiveKitURL string
}

type ParticipantRemover 

type ParticipantRemover func(context.Context, string, string) error

type QueueAudioOutput 

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

QueueAudioOutput is a bounded, single-stream AudioOutput for provider WebSocket adapters and AvatarRunner implementations. Capture, Flush, and ClearBuffer are linearized so a clear can never erase the boundary of the next segment.

func NewQueueAudioOutput 

func NewQueueAudioOutput(options QueueAudioOutputOptions) (*QueueAudioOutput, error)

func (*QueueAudioOutput) AClose 

func (o *QueueAudioOutput) AClose(context.Context) error

AClose is the Python/TypeScript-compatible spelling.

func (*QueueAudioOutput) CanPause 

func (o *QueueAudioOutput) CanPause() bool

func (*QueueAudioOutput) CaptureFrame 

func (o *QueueAudioOutput) CaptureFrame(ctx context.Context, frame agents.AudioFrame) error

func (*QueueAudioOutput) CapturedPlayoutSegments 

func (o *QueueAudioOutput) CapturedPlayoutSegments() uint64

func (*QueueAudioOutput) ClearBuffer 

func (o *QueueAudioOutput) ClearBuffer(ctx context.Context) error

func (*QueueAudioOutput) Close 

func (o *QueueAudioOutput) Close() error

Close gracefully closes the readable stream after already accepted items. It is idempotent and unblocks producers currently applying backpressure.

func (*QueueAudioOutput) Closed 

func (o *QueueAudioOutput) Closed() bool

func (*QueueAudioOutput) Flush 

func (o *QueueAudioOutput) Flush(ctx context.Context) error

func (*QueueAudioOutput) NotifyPlaybackFinished 

func (o *QueueAudioOutput) NotifyPlaybackFinished(playbackPosition time.Duration, interrupted bool) error

NotifyPlaybackFinished mirrors the Python/TypeScript transport callback.

func (*QueueAudioOutput) NotifyPlaybackFinishedEvent 

func (o *QueueAudioOutput) NotifyPlaybackFinishedEvent(event voice.PlaybackFinishedEvent) error

NotifyPlaybackFinishedEvent additionally preserves synchronized transcript.

func (*QueueAudioOutput) NotifyPlaybackStarted 

func (o *QueueAudioOutput) NotifyPlaybackStarted(createdAt ...time.Time)

func (*QueueAudioOutput) OnAttached 

func (o *QueueAudioOutput) OnAttached()

func (*QueueAudioOutput) OnClearBuffer 

func (o *QueueAudioOutput) OnClearBuffer(fn func(QueueAudioOutputClearEvent)) func()

func (*QueueAudioOutput) OnDetached 

func (o *QueueAudioOutput) OnDetached()

func (*QueueAudioOutput) OnPlaybackFinished 

func (o *QueueAudioOutput) OnPlaybackFinished(fn func(voice.PlaybackFinishedEvent)) func()

func (*QueueAudioOutput) OnPlaybackStarted 

func (o *QueueAudioOutput) OnPlaybackStarted(fn func(voice.PlaybackStartedEvent)) func()

func (*QueueAudioOutput) Pause 

func (o *QueueAudioOutput) Pause(ctx context.Context) error

func (*QueueAudioOutput) PendingPlayoutSegments 

func (o *QueueAudioOutput) PendingPlayoutSegments() uint64

func (*QueueAudioOutput) Recv 

func (o *QueueAudioOutput) Recv(ctx context.Context) (QueueAudioOutputItem, error)

Recv reads from the one shared stream. Multiple concurrent readers split items rather than broadcast them and should only be used intentionally.

func (*QueueAudioOutput) Resume 

func (o *QueueAudioOutput) Resume(ctx context.Context) error

func (*QueueAudioOutput) SampleRate 

func (o *QueueAudioOutput) SampleRate() int

func (*QueueAudioOutput) SetAttached 

func (o *QueueAudioOutput) SetAttached(attached bool)

func (*QueueAudioOutput) Start 

func (o *QueueAudioOutput) Start(context.Context) error

Start satisfies the receiver side of the Python queue-I/O contract. Queue output is ready immediately and owns no background goroutine.

func (*QueueAudioOutput) WaitForPlayout 

func (o *QueueAudioOutput) WaitForPlayout(ctx context.Context) (voice.PlaybackFinishedEvent, error)

type QueueAudioOutputClearEvent 

type QueueAudioOutputClearEvent struct {
	WasCapturing bool
}

type QueueAudioOutputItem 

type QueueAudioOutputItem struct {
	Frame      agents.AudioFrame
	SegmentEnd bool
}

QueueAudioOutputItem is a type-safe union. Exactly one of Frame or SegmentEnd is meaningful; SegmentEnd avoids a pointer allocation per marker.

func AudioFrameItem 

func AudioFrameItem(frame agents.AudioFrame) QueueAudioOutputItem

func AudioSegmentEndItem 

func AudioSegmentEndItem() QueueAudioOutputItem

type QueueAudioOutputOptions 

type QueueAudioOutputOptions struct {
	SampleRate        int
	Capacity          int
	WaitPlaybackStart bool
}

type RPCInvocation 

type RPCInvocation struct {
	CallerIdentity string
	Payload        string
}

type ReadyWaiter 

type ReadyWaiter func(context.Context, *lksdk.Room, string, lksdk.TrackKind) error

ReadyWaiter bridges the server SDK's lack of post-construction Room callback registration. It is only needed when the destination has not already joined. voice/avatar/roomioadapter provides one for roomio.RTCBridge.

type WaitForJoinOptions 

type WaitForJoinOptions struct {
	// Timeout defaults to 30 seconds. A negative duration waits indefinitely;
	// zero explicitly performs a non-blocking deadline check.
	Timeout *time.Duration
}

Source Files

View all Source files

Directories

Path Synopsis
Package roomioadapter connects avatar readiness waits to a roomio RTCBridge.
 Package roomioadapter connects avatar readiness waits to a roomio RTCBridge.

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.