agents

package module
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: 40 Imported by: 0

README

LiveKit Agents for Go

Production-oriented Go SDK for realtime, programmable LiveKit agents. Its public surface follows @livekit/agents and livekit-agents, expressed with idiomatic Go contexts, interfaces, typed options, and explicit error returns.

The compatibility baseline is:

  • livekit/agents-js commit 128f3f6a230616e960325b112067861ce1f1a17f (@livekit/agents 1.7.1)
  • livekit/agents commit cdb37ade6f8e80822e6c5ec4e6de457f2dcaf637
  • LiveKit protocol 1.50.4
  • LiveKit Go server/RTC SDK 2.18.1

The core package intentionally avoids provider SDKs and eager initialization. Providers are isolated in subpackages so unused integrations do not affect binary size or process startup. The only bundled provider is ElevenLabs.

go get github.com/infinityscroll/livekit-agents-go@v0.1.0

Production voice agent

package main

import (
    "context"
    "os"

    agents "github.com/infinityscroll/livekit-agents-go"
    "github.com/infinityscroll/livekit-agents-go/agentscli"
    "github.com/infinityscroll/livekit-agents-go/llm"
    "github.com/infinityscroll/livekit-agents-go/voice"
    voicekit "github.com/infinityscroll/livekit-agents-go/voice/livekit"
)

func main() {
    options := agents.WorkerOptions[struct{}]{
        JobEntrypoint: func(ctx context.Context, job *agents.JobContext[struct{}]) error {
            session, err := voice.NewAgentSession(voice.AgentSessionOptions[struct{}]{
                ParentContext: job.Context(),
                STTModel:      "deepgram/nova-3:en",
                LLMModel:      "openai/gpt-4.1-mini",
                TTSModel:      "cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
            })
            if err != nil {
                return err
            }
            agent, err := voice.NewAgent(voice.AgentOptions[struct{}]{
                ID: "assistant",
                Instructions: llm.NewInstructions(
                    "You are a concise, helpful voice assistant.",
                    "You are a concise, helpful text assistant.",
                ),
            })
            if err != nil {
                return err
            }
            _, err = voicekit.Start(ctx, session, agent, voicekit.StartOptions[struct{}]{
                Job: job,
            })
            return err
        },
    }
    os.Exit(agentscli.Main(options, os.Args[1:]))
}

voicekit.Start is the one-step equivalent of the Python/TypeScript session start path: it connects the room, installs RoomIO, claims the primary session, resolves inherited recording/redaction, configures bounded report capture and LiveKit Cloud observability, starts RecorderIO when requested, and registers ordered shutdown/report finalization with the job. Partial startup rolls back in reverse order. Secondary sessions are returned to the caller for explicit closure and cannot accidentally claim primary recording.

Set LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET, then build and run your-agent dev locally or your-agent start in production. The embedded CLI also supports connect, console [--record], download-files, version, and help, matching the upstream workflow. Applications embedding their own command layer can call agents.Run(ctx, options) directly. The normal production executor uses a process per job; select ExecutorModeInProcess only when shared-process behavior is intentional.

Installation and native audio

The worker, model, telemetry, ElevenLabs, and voice-session packages are normal Go modules. RTC RoomIO additionally uses cgo through the pinned LiveKit media stack:

# Debian / Ubuntu
sudo apt-get install pkg-config libogg-dev libopus-dev libopusfile-dev libsoxr-dev

# macOS (Homebrew)
brew install pkg-config libogg opus opusfile soxr

Text-only/core consumers can keep RoomIO out of their import graph. See docs/capability-gaps.md for the exact native and raw-video boundaries.

What is included

  • process-isolated worker/job lifecycle, health, reconnect, drain, memory and authenticated bounded IPC;
  • LiveKit Inference LLM/STT/TTS, turn detection, VAD, adaptive interruption, alignment and avatar gateway adapters;
  • cascaded and realtime voice sessions, RoomIO, recorder, background audio, remote/console sessions, AMD, tools, handoffs, workflows and telemetry;
  • STT/TTS streaming and fallback adapters, tokenization/transcription, metrics, provider-format conversion, and the ElevenLabs plugin;
  • machine-checked API and cross-language parity manifests, dependency/license policy, fuzz/race/static-analysis gates, cold-start probes and benchmarks.

See docs/compatibility.md and docs/parity-manifest.json for the exact TypeScript/Python-to-Go mapping, docs/performance.md for reproducible startup, CPU, allocation, and memory measurements, and docs/capability-gaps.md for the explicit boundaries imposed by the pinned Go RTC/native stack.

Supported Go

The module targets Go 1.26 or newer, matching the current LiveKit Go RTC SDK. CI also builds with the current stable Go toolchain.

License

Apache-2.0.

Documentation

Index

Constants

View Source 
const (
	// UserdataTimedTranscript stores timed transcripts on an AudioFrame.
	UserdataTimedTranscript = "lk.timed_transcripts"
	// UserdataTTSStartedTime stores when text was first sent to TTS.
	UserdataTTSStartedTime = "lk.tts_started_time"

	AttributeSimulator         = "lk.simulator"
	AttributeSimulatorDispatch = "lk.simulator.dispatch"
	AttributeSimulationEnabled = "lk.simulation.enabled"
	AttributeSimulationRunID   = "lk.simulation.run_id"
	AttributeSimulationJobID   = "lk.simulation.job_id"
	AttributeRedactionEnabled  = "lk.redaction.enabled"

	AttributeTranscriptionExpression = "lk.expression"
)
View Source 
const (
	CPUCountEnvironment      = "NUM_CPUS"
	DefaultCPUSampleInterval = 500 * time.Millisecond
	DefaultCGroupV1CPUCount  = 2.0
)
View Source 
const (
	// FFmpegPathEnvironment is the TypeScript-compatible executable override.
	FFmpegPathEnvironment = "LIVEKIT_FFMPEG_PATH"
	// FFmpegPathEnv is retained as the shorter compatibility spelling.
	FFmpegPathEnv = FFmpegPathEnvironment
)
View Source 
const (
	SubscribeAll AutoSubscribe = iota
	SubscribeNone
	SubscribeVideoOnly
	SubscribeAudioOnly

	// Upper-case aliases make mechanical TypeScript/Python migrations easier.
	SUBSCRIBE_ALL  = SubscribeAll
	SUBSCRIBE_NONE = SubscribeNone
	VIDEO_ONLY     = SubscribeVideoOnly
	AUDIO_ONLY     = SubscribeAudioOnly
)
View Source 
const (
	LogFormatJSON LogFormat = "json"
	LogFormatText LogFormat = "text"

	LogLevelTrace  slog.Level = slog.LevelDebug - 4
	LogLevelSilent slog.Level = slog.LevelError + 100
)
View Source 
const (
	SimulationModeUnspecified = livekit.SimulationMode_SIMULATION_MODE_UNSPECIFIED
	SimulationModeText        = livekit.SimulationMode_SIMULATION_MODE_TEXT
	SimulationModeAudio       = livekit.SimulationMode_SIMULATION_MODE_AUDIO

	// Upper-case aliases ease mechanical TypeScript/protobuf migrations.
	SIMULATION_MODE_UNSPECIFIED = SimulationModeUnspecified
	SIMULATION_MODE_TEXT        = SimulationModeText
	SIMULATION_MODE_AUDIO       = SimulationModeAudio
)
View Source 
const (
	DefaultMaxReconnects          = 10
	DefaultAssignmentTimeout      = 7500 * time.Millisecond
	DefaultStatusUpdateInterval   = 2500 * time.Millisecond
	DefaultDrainTimeout           = time.Hour
	DefaultShutdownProcessTimeout = time.Minute
	DefaultInitializeTimeout      = 10 * time.Second
	DefaultJobMemoryWarnMB        = 1000
	DefaultWorkerPort             = 8081

	AttributeAgentName = "lk.agent.name"
)
View Source 
const (
	ServerTypeRoom      = livekit.JobType_JT_ROOM
	ServerTypePublisher = livekit.JobType_JT_PUBLISHER
)
View Source 
const DefaultAudioEnergyThreshold = 0.004
View Source 
const DefaultInferenceInitializeTimeout = 5 * time.Minute

Variables

View Source 
var (
	ErrInvalidAudioFormat = errors.New("invalid audio format")
	ErrIncompletePCMFrame = errors.New("incomplete PCM16 frame")
)
View Source 
var (
	// ErrInferenceProcessUnavailable reports that the supervised runner child
	// cannot accept a request. It is safe to use with errors.Is.
	ErrInferenceProcessUnavailable = errors.New("agents: inference process is unavailable")
	// ErrInferenceProcessCrashed reports an unexpected runner-child exit.
	ErrInferenceProcessCrashed = errors.New("agents: inference process crashed")
	// ErrInferenceConcurrencyLimit reports that the fixed-size subprocess
	// request budget is exhausted. Requests are not placed on an unbounded queue.
	ErrInferenceConcurrencyLimit = errors.New("agents: inference concurrency limit reached")
	// ErrInferenceRegistrationMismatch means the parent and same-binary child
	// did not execute the same startup registration path.
	ErrInferenceRegistrationMismatch = errors.New("agents: inference runner registrations differ in child process")
)
View Source 
var (
	ErrInferenceRunnerExists  = errors.New("agents: inference runner is already registered")
	ErrInferenceRunnerUnknown = errors.New("agents: inference runner is not registered")
)
View Source 
var (
	ErrJobRequestAnswered = errors.New("agents: job request has already been answered")
	ErrRoomNotConnected   = errors.New("agents: room is not connected")
	ErrFunctionExists     = errors.New("agents: function has already been registered")
)
View Source 
var (
	// ErrPrimarySessionRecording is returned when a secondary session
	// explicitly enables recording. Exactly one session owns recording and the
	// automatic shutdown/report lifecycle for a job.
	ErrPrimarySessionRecording = errors.New("agents: only the primary session may enable recording")
	ErrJobSessionFinishing     = errors.New("agents: job session lifecycle is already finishing")
)
View Source 
var (
	ErrNoJobCapacity       = errors.New("agents: worker has no available job capacity")
	ErrJobNotFound         = errors.New("agents: job was not found")
	ErrReservationReleased = errors.New("agents: job reservation was released")
	ErrExecutorClosed      = errors.New("agents: job executor is closed")
)
View Source 
var DefaultAPIConnectOptions = APIConnectOptions{
	MaxRetries:    3,
	RetryInterval: 2 * time.Second,
	Timeout:       10 * time.Second,
}
View Source 
var DefaultSessionConnectOptions = SessionConnectOptions{
	STT:                    DefaultAPIConnectOptions,
	LLM:                    DefaultAPIConnectOptions,
	TTS:                    DefaultAPIConnectOptions,
	MaxUnrecoverableErrors: 3,
}
View Source 
var ErrCPUUsageUnavailable = errors.New("CPU usage is unavailable")
View Source 
var ErrInferenceExecutorClosed = errors.New("agents: inference executor is closed")
View Source 
var ErrRTCBridgeRequired = errors.New("agents: a construction-time RTCBridge is required for event-driven room waits")
View Source 
var ErrSimulatorVerdictUnavailable = errors.New("simulator verdict is only available after the simulation completes")
View Source 
var KnownLanguageCodes = []LanguageCode{
	"af", "am", "ar", "as", "az", "be", "bg", "bn", "bs", "ca", "cs", "cy", "da", "de",
	"el", "en", "es", "et", "eu", "fa", "ff", "fi", "fr", "ga", "gl", "gu", "ha", "he",
	"hi", "hr", "hu", "hy", "id", "ig", "is", "it", "ja", "jv", "ka", "kk", "km", "kn",
	"ko", "ku", "ky", "lb", "lg", "ln", "lo", "lt", "lv", "mi", "mk", "ml", "mn", "mr",
	"ms", "mt", "my", "ne", "nl", "no", "ny", "oc", "or", "pa", "pl", "ps", "pt", "ro",
	"ru", "sd", "sk", "sl", "sn", "so", "sq", "sr", "sv", "sw", "ta", "te", "tg", "th",
	"tl", "tr", "uk", "ur", "uz", "vi", "wo", "xh", "yo", "zh", "zu",
}
View Source 
var Version = "v0.1.0"

Version is the semantic version of this SDK. Downstream development builds may override it with -ldflags "-X github.com/infinityscroll/livekit-agents-go.Version=vX.Y.Z".

Functions

func AreLanguagesEquivalent 

func AreLanguagesEquivalent(left, right string) bool

func BaseLanguage 

func BaseLanguage(language string) string

func CalculateAudioDuration 

func CalculateAudioDuration(frames []AudioFrame) time.Duration

func CalculateAudioDurationSeconds 

func CalculateAudioDurationSeconds(frame AudioFrame) float64

func ConfigureFFmpeg 

func ConfigureFFmpeg() string

ConfigureFFmpeg mirrors agents-js's explicit setup hook. Go callers do not need process-global configuration, so the resolved executable is returned for dependency injection instead.

func Dedent 

func Dedent(value string) string

Dedent strips a common leading run of spaces/tabs from non-empty lines, removes one initial newline, and trims trailing Unicode whitespace. It is the Go string equivalent of the agents-js dedent tagged template helper.

func FFmpegCommand 

func FFmpegCommand(ctx context.Context, args ...string) *exec.Cmd

FFmpegCommand constructs a cancellation-aware command using the standard SDK resolution rules. Starting the command reports the usual exec error if no executable is installed.

func ISOLanguage 

func ISOLanguage(language string) string

func InferenceExecutorFromContext 

func InferenceExecutorFromContext(ctx context.Context) (ipc.InferenceExecutor, bool)

InferenceExecutorFromContext returns the worker-global executor propagated to a job entrypoint without requiring the caller to know JobContext's user-data type. Model adapters use this to share native runners across isolated jobs.

func InitializeLogger 

func InitializeLogger(options LoggerOptions) (*slog.Logger, error)

InitializeLogger installs the logger returned by Log. It performs no work until called and never changes slog.Default unless SetDefault is true.

func IsAPIError 

func IsAPIError(err error) bool

func IsAgent 

func IsAgent(value any) bool

IsAgent reports whether value was produced as a Go AgentDefinition. The private marker deliberately rejects look-alike structs.

func IsCloud 

func IsCloud(rawURL string) bool

IsCloud reports whether rawURL is hosted on a LiveKit Cloud production or run domain. Malformed and hostname-less URLs return false.

func IsDevMode 

func IsDevMode() bool

IsDevMode reports whether the process was launched in a LiveKit development mode such as dev, connect, or console.

func IsHosted 

func IsHosted() bool

IsHosted reports whether the worker is hosted by LiveKit Cloud. Presence is significant even when the environment variable is explicitly empty.

func LanguageName 

func LanguageName(language string) (string, bool)

func LanguageRegion 

func LanguageRegion(language string) (string, bool)

func Log 

func Log() *slog.Logger

Log returns the framework logger or slog.Default when the application has not initialized one. This makes library use safe outside the CLI.

func MustRegisterInferenceRunner 

func MustRegisterInferenceRunner(method string, factory InferenceRunnerFactory)

MustRegisterInferenceRunner is intended for package-level plugin registration before the executor pool starts and panics on a duplicate or invalid method. Package initialization naturally runs in the same-binary inference child required by process mode.

func OnPluginRegistered 

func OnPluginRegistered(fn func(PluginRegisteredEvent)) func()

func ParseLogLevel 

func ParseLogLevel(value string) (slog.Level, error)

func PutResource 

func PutResource[T any](bag *ResourceBag, key string, value T) error

func RegisterInferenceRunner 

func RegisterInferenceRunner(method string, factory InferenceRunnerFactory) error

RegisterInferenceRunner registers a local inference method. Registration is concurrency-safe, rejects duplicates, and allocates the registry only after the first feature package is actually imported.

Register every production method before the worker executor pool starts. Process mode re-executes the same binary and requires the factory to be linked and registered along the child startup path as well; a runtime-only closure cannot be serialized into that process. ExecutorModeInProcess is the explicit development escape hatch for parent-only dynamic registrations.

func RegisterPlugin 

func RegisterPlugin(plugin Plugin) error

func RegisteredInferenceRunners 

func RegisteredInferenceRunners() map[string]InferenceRunnerFactory

RegisteredInferenceRunners returns a defensive snapshot. An empty registry returns nil without allocating, keeping core-only process startup lean.

func ResolveFFmpegPath 

func ResolveFFmpegPath() string

ResolveFFmpegPath resolves LIVEKIT_FFMPEG_PATH first and then ffmpeg on PATH. It performs no work at package initialization and returns an empty string when no executable can be found.

func Resource 

func Resource[T any](bag *ResourceBag, key string) (T, bool)

func Run 

func Run[T any](ctx context.Context, options WorkerOptions[T]) error

Run constructs a worker, handles SIGINT/SIGTERM, drains, and closes it.

func RunConsoleJob 

func RunConsoleJob[UserData any](ctx context.Context, options ConsoleJobOptions[UserData]) (resultErr error)

RunConsoleJob executes the normal prewarm, entrypoint, job-context, and shutdown-callback lifecycle in-process. Cancellation is the console lifetime signal and is treated as a clean shutdown.

func SetLogger 

func SetLogger(logger *slog.Logger) error

SetLogger installs an application-owned logger without mutating the global slog default. Passing nil is rejected instead of creating a latent panic.

func ShortUUID 

func ShortUUID(prefix string) string

ShortUUID returns a URL-safe, process-independent identifier with at least 100 bits of randomness. crypto/rand.Text panics only if OS entropy fails.

func WaitForParticipantAttribute 

func WaitForParticipantAttribute(
	ctx context.Context,
	room *lksdk.Room,
	bridge *rtcbridge.RTCBridge,
	identity, attribute, value string,
) error

WaitForParticipantAttribute waits without polling until a remote participant's attribute equals value. bridge must be the callback bridge installed when room was constructed; server-sdk-go v2.18.1 has no safe post-construction callback registration API.

func WaitForTrackPublication 

func WaitForTrackPublication(
	ctx context.Context,
	room *lksdk.Room,
	bridge *rtcbridge.RTCBridge,
	options WaitForTrackPublicationOptions,
) (lksdk.TrackPublication, error)

WaitForTrackPublication waits without polling for a matching local or remote publication and returns the SDK-owned publication handle.

func WithConnectionResult 

func WithConnectionResult[T comparable, R any](ctx context.Context, pool *ConnectionPool[T], fn func(context.Context, T) (R, error)) (R, error)

func WithInferenceExecutor 

func WithInferenceExecutor(ctx context.Context, executor ipc.InferenceExecutor) context.Context

WithInferenceExecutor explicitly propagates an executor to model/session constructors outside a worker entrypoint. It is useful for console hosts and tests; a typed nil executor leaves the context unchanged.

Types

type APIConnectOptions 

type APIConnectOptions struct {
	MaxRetries    int
	RetryInterval time.Duration
	Timeout       time.Duration
}

APIConnectOptions controls timeout and retry behavior for provider calls. Zero fields resolve to DefaultAPIConnectOptions. Set MaxRetries to a negative value to explicitly disable retries.

func (APIConnectOptions) Resolve 

func (o APIConnectOptions) Resolve() APIConnectOptions

Resolve returns a validated, fully populated copy.

func (APIConnectOptions) RetryDelay 

func (o APIConnectOptions) RetryDelay(n int) time.Duration

RetryDelay returns the delay before retry number n. The first retry is fast, matching the Python and TypeScript SDKs.

type APIConnectionError 

type APIConnectionError struct{ *APIError }

func NewAPIConnectionError 

func NewAPIConnectionError(message string, retryable bool, cause error) *APIConnectionError

func (*APIConnectionError) Unwrap 

func (e *APIConnectionError) Unwrap() error

type APIError 

type APIError struct {
	Message       string
	Body          any
	RetryableFlag bool
	Cause         error
}

APIError is returned by STT, LLM, TTS, VAD, and inference providers.

func NewAPIError 

func NewAPIError(message string, body any, retryable bool, cause error) *APIError

func (*APIError) Error 

func (e *APIError) Error() string

func (*APIError) Retryable 

func (e *APIError) Retryable() bool

func (*APIError) Unwrap 

func (e *APIError) Unwrap() error

type APIStatusError 

type APIStatusError struct {
	*APIError
	StatusCode int
	RequestID  string
}

func NewAPIStatusError 

func NewAPIStatusError(message string, statusCode int, requestID string, body any, retryable bool, cause error) *APIStatusError

func (*APIStatusError) Error 

func (e *APIStatusError) Error() string

func (*APIStatusError) Unwrap 

func (e *APIStatusError) Unwrap() error

type APITimeoutError 

type APITimeoutError struct{ *APIConnectionError }

func NewAPITimeoutError 

func NewAPITimeoutError(message string, retryable bool, cause error) *APITimeoutError

func (*APITimeoutError) Unwrap 

func (e *APITimeoutError) Unwrap() error

type AgentDefinition 

type AgentDefinition[T any] struct {
	// contains filtered or unexported fields
}

AgentDefinition is the validated worker module used by agents-js defineAgent. Its fields are immutable after construction so it can be shared safely between server setup and process executors.

func DefineAgent 

func DefineAgent[T any](options AgentDefinitionOptions[T]) (*AgentDefinition[T], error)

func MustDefineAgent 

func MustDefineAgent[T any](options AgentDefinitionOptions[T]) *AgentDefinition[T]

func (*AgentDefinition[T]) Entrypoint 

func (d *AgentDefinition[T]) Entrypoint() JobEntrypoint[T]

func (*AgentDefinition[T]) Prewarm 

func (d *AgentDefinition[T]) Prewarm() PrewarmFunc[T]

func (*AgentDefinition[T]) SimulationEnd 

func (d *AgentDefinition[T]) SimulationEnd() SimulationEndFunc[T]

type AgentDefinitionOptions 

type AgentDefinitionOptions[T any] struct {
	Entrypoint      JobEntrypoint[T]
	Prewarm         PrewarmFunc[T]
	OnSimulationEnd SimulationEndFunc[T]
}

type AgentServer 

type AgentServer[T any] struct {
	// contains filtered or unexported fields
}

AgentServer owns worker registration, job reservations, executors, health serving, reconnection, and graceful draining.

func NewAgentServer 

func NewAgentServer[T any](options ServerOptions[T]) (*AgentServer[T], error)

func (*AgentServer[T]) ActiveJobs 

func (s *AgentServer[T]) ActiveJobs() []RunningJobInfo

func (*AgentServer[T]) Close 

func (s *AgentServer[T]) Close(contexts ...context.Context) error

Close stops registration, the health server, and all remaining executors. It is idempotent. With no context argument it uses a background context.

func (*AgentServer[T]) Drain 

func (s *AgentServer[T]) Drain(ctx context.Context) error

Drain immediately advertises FULL and waits for accepted jobs to finish.

func (*AgentServer[T]) Draining 

func (s *AgentServer[T]) Draining() bool

func (*AgentServer[T]) Events 

func (s *AgentServer[T]) Events() *EventEmitter[WorkerEvent]

func (*AgentServer[T]) HTTPAddress 

func (s *AgentServer[T]) HTTPAddress() string

HTTPAddress returns the bound health endpoint after Run has started it.

func (*AgentServer[T]) ID 

func (s *AgentServer[T]) ID() string

func (*AgentServer[T]) RegisterAgent 

func (s *AgentServer[T]) RegisterAgent(name string, entrypoint JobEntrypoint[T]) error

RegisterAgent mirrors Python's pre-run AgentServer registration style.

func (*AgentServer[T]) RegisterRTCSession 

func (s *AgentServer[T]) RegisterRTCSession(name string, entrypoint JobEntrypoint[T]) error

RegisterRTCSession is the Python-compatible name for RegisterAgent.

func (*AgentServer[T]) Registered 

func (s *AgentServer[T]) Registered() bool

func (*AgentServer[T]) Run 

func (s *AgentServer[T]) Run(ctx context.Context) error

Run blocks until the worker is closed, its context is canceled, or a fatal registration/reconnection error occurs. A canceled context triggers a drain.

func (*AgentServer[T]) SimulateJob 

func (s *AgentServer[T]) SimulateJob(ctx context.Context, roomName, participantIdentity string) error

SimulateJob asks the connected server to synthesize a job request.

type AssignmentTimeoutError 

type AssignmentTimeoutError struct{ Message string }

func (*AssignmentTimeoutError) Error 

func (e *AssignmentTimeoutError) Error() string

type AudioByteStream 

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

AudioByteStream packetizes little-endian PCM16 without repeatedly copying the accumulated input. It is not safe for concurrent use.

func NewAudioByteStream 

func NewAudioByteStream(sampleRate, channels, samplesPerChannel int) (*AudioByteStream, error)

func (*AudioByteStream) Flush 

func (s *AudioByteStream) Flush() ([]AudioFrame, error)

func (*AudioByteStream) Reset 

func (s *AudioByteStream) Reset()

func (*AudioByteStream) Write 

func (s *AudioByteStream) Write(data []byte) []AudioFrame

type AudioDecodeError 

type AudioDecodeError struct {
	Path   string
	Codec  string
	Stderr string
	Cause  error
}

func (*AudioDecodeError) Error 

func (e *AudioDecodeError) Error() string

func (*AudioDecodeError) Unwrap 

func (e *AudioDecodeError) Unwrap() error

type AudioDecodeOptions 

type AudioDecodeOptions struct {
	SampleRate int
	Channels   int
	// NumChannels is the agents-js compatibility name for Channels.
	NumChannels    int
	Format         string
	FrameDuration  time.Duration
	StreamCapacity int
	// FFmpegPath overrides the lazily discovered ffmpeg executable. Native
	// matching-format PCM16 WAV decoding does not start or require ffmpeg.
	FFmpegPath string
}

type AudioEnergyFilter 

type AudioEnergyFilter struct {
	Cooldown  time.Duration
	Threshold float64
	// contains filtered or unexported fields
}

AudioEnergyFilter holds speech-active state for Cooldown after the most recent frame whose normalized PCM RMS exceeds Threshold. PushFrame performs no allocations and is intended for one audio actor/goroutine.

func NewAudioEnergyFilter 

func NewAudioEnergyFilter(cooldown time.Duration) *AudioEnergyFilter

func (*AudioEnergyFilter) PushFrame 

func (f *AudioEnergyFilter) PushFrame(frame AudioFrame) bool

func (*AudioEnergyFilter) Reset 

func (f *AudioEnergyFilter) Reset()

type AudioFrame 

type AudioFrame struct {
	Data              []int16
	SampleRate        int
	Channels          int
	SamplesPerChannel int
	UserData          map[string]any
}

AudioFrame is interleaved signed 16-bit PCM. Data is owned by the frame and remains valid until the frame is discarded.

func MergeFrames 

func MergeFrames(frames []AudioFrame) (AudioFrame, error)

MergeFrames combines same-format frames with one allocation.

func NewAudioFrame 

func NewAudioFrame(data []int16, sampleRate, channels int) (AudioFrame, error)

func (AudioFrame) Duration 

func (f AudioFrame) Duration() time.Duration

type AudioFrameStream 

type AudioFrameStream interface {
	stream.Reader[AudioFrame]
	Close() error
	Wait(context.Context) error
}

func AudioFramesFromFile 

func AudioFramesFromFile(ctx context.Context, path string, options AudioDecodeOptions) (AudioFrameStream, error)

AudioFramesFromFile decodes a file to bounded 16-bit PCM frames. Matching PCM16 WAV files use the native streaming parser (fast startup, no subprocess); other formats and resampling use ffmpeg, matching agents-js's codec surface.

func LoopAudioFramesFromFile 

func LoopAudioFramesFromFile(ctx context.Context, path string, options AudioDecodeOptions) (AudioFrameStream, error)

LoopAudioFramesFromFile repeatedly decodes a file without retaining the entire file in memory. An empty file terminates with io.ErrUnexpectedEOF instead of spinning at 100% CPU.

type AutoSubscribe 

type AutoSubscribe uint8

AutoSubscribe controls which remote tracks JobContext.Connect subscribes to. The values intentionally follow the TypeScript and Python SDK ordering.

type CGroupV1CPUMonitor 

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

CGroupV1CPUMonitor reads the legacy cpu/cpuacct controllers.

func NewCGroupV1CPUMonitor 

func NewCGroupV1CPUMonitor(root string) *CGroupV1CPUMonitor

func (*CGroupV1CPUMonitor) CPUCount 

func (m *CGroupV1CPUMonitor) CPUCount() float64

func (*CGroupV1CPUMonitor) CPUPercent 

func (m *CGroupV1CPUMonitor) CPUPercent(ctx context.Context, interval time.Duration) (float64, error)

type CGroupV2CPUMonitor 

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

CGroupV2CPUMonitor reads unified-hierarchy CPU quota and usage.

func NewCGroupV2CPUMonitor 

func NewCGroupV2CPUMonitor(root string) *CGroupV2CPUMonitor

func (*CGroupV2CPUMonitor) CPUCount 

func (m *CGroupV2CPUMonitor) CPUCount() float64

func (*CGroupV2CPUMonitor) CPUPercent 

func (m *CGroupV2CPUMonitor) CPUPercent(ctx context.Context, interval time.Duration) (float64, error)

type CPUMonitor 

type CPUMonitor interface {
	CPUCount() float64
	CPUPercent(context.Context, time.Duration) (float64, error)
}

CPUMonitor reports normalized load in [0,1] and the effective CPU quota. CPUPercent uses no helper goroutine and always honors ctx cancellation.

func GetCPUMonitor 

func GetCPUMonitor() CPUMonitor

GetCPUMonitor selects Linux cgroup accounting when available, then falls back to Go runtime CPU-class metrics. Detection is explicit and lazy.

type ConnectOptions 

type ConnectOptions struct {
	AutoSubscribe AutoSubscribe
	RoomOptions   []lksdk.ConnectOption
}

ConnectOptions controls room connection and automatic subscription.

type ConnectionPool 

type ConnectionPool[T comparable] struct {
	// contains filtered or unexported fields
}

ConnectionPool serializes establishment, reuses connections in insertion order, and never returns an expired/failed connection.

func NewConnectionPool 

func NewConnectionPool[T comparable](options ConnectionPoolOptions[T]) (*ConnectionPool[T], error)

func (*ConnectionPool[T]) Close 

func (p *ConnectionPool[T]) Close(ctx context.Context) error

func (*ConnectionPool[T]) Get 

func (p *ConnectionPool[T]) Get(ctx context.Context) (T, error)

func (*ConnectionPool[T]) Invalidate 

func (p *ConnectionPool[T]) Invalidate()

func (*ConnectionPool[T]) Len 

func (p *ConnectionPool[T]) Len() int

func (*ConnectionPool[T]) Prewarm 

func (p *ConnectionPool[T]) Prewarm(ctx context.Context)

func (*ConnectionPool[T]) Put 

func (p *ConnectionPool[T]) Put(conn T) bool

func (*ConnectionPool[T]) Remove 

func (p *ConnectionPool[T]) Remove(ctx context.Context, conn T) error

func (*ConnectionPool[T]) WithConnection 

func (p *ConnectionPool[T]) WithConnection(ctx context.Context, fn func(context.Context, T) error) error

type ConnectionPoolOptions 

type ConnectionPoolOptions[T comparable] struct {
	MaxSessionDuration time.Duration
	MarkRefreshedOnGet bool
	Connect            func(context.Context) (T, error)
	Close              func(context.Context, T) error
	ConnectTimeout     time.Duration
}

type ConsoleJobOptions 

type ConsoleJobOptions[UserData any] struct {
	Server ServerOptions[UserData]
	Record bool
	// SessionDirectory is shared with AgentSession recording/report lifecycle.
	// An empty value uses the normal traversal-safe temporary job directory.
	SessionDirectory string
	ShutdownTimeout  time.Duration
	// Setup runs after prewarm and before the entrypoint. The CLI uses it to
	// bind process-local console IO without introducing a root-package import
	// cycle with voice.
	Setup func(context.Context, *JobContext[UserData], SimulationEndFunc[UserData]) error
}

ConsoleJobOptions configures the in-process fake job used by the console command. It deliberately does not require LiveKit credentials or initialize a worker websocket.

type Entrypoint 

type Entrypoint[T any] = JobEntrypoint[T]

type EventEmitter 

type EventEmitter[T any] struct {
	// contains filtered or unexported fields
}

EventEmitter is a small synchronous typed emitter. Callbacks execute in registration order and outside the emitter lock. Subscribe and unsubscribe are safe to call concurrently with Emit.

func (*EventEmitter[T]) Emit 

func (e *EventEmitter[T]) Emit(event T)

func (*EventEmitter[T]) Len 

func (e *EventEmitter[T]) Len() int

func (*EventEmitter[T]) Subscribe 

func (e *EventEmitter[T]) Subscribe(fn func(T)) (unsubscribe func())

type ExecutorMode 

type ExecutorMode uint8

ExecutorMode selects job isolation. Process isolation is the production default and preserves the Python/TypeScript crash and memory boundary. 

const (
	ExecutorModeProcess ExecutorMode = iota
	ExecutorModeInProcess
)

type ExpFilter 

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

ExpFilter performs exponentially weighted smoothing. It is intended to be owned by one actor/goroutine, avoiding a lock on load/audio hot paths.

func NewExpFilter 

func NewExpFilter(options ExpFilterOptions) (*ExpFilter, error)

func (*ExpFilter) Apply 

func (f *ExpFilter) Apply(exponent, sample float64) float64

func (*ExpFilter) Reset 

func (f *ExpFilter) Reset(initial *float64)

func (*ExpFilter) SetAlpha 

func (f *ExpFilter) SetAlpha(alpha float64) error

func (*ExpFilter) Value 

func (f *ExpFilter) Value() (float64, bool)

type ExpFilterOptions 

type ExpFilterOptions struct {
	Alpha   float64
	Initial *float64
	Min     *float64
	Max     *float64
}

type FlushSentinel 

type FlushSentinel struct{}

FlushSentinel is the Go equivalent of the JS FlushSentinel symbol. A model stream emits a value with Flush=true instead of relying on an untyped marker.

type FunctionExistsError 

type FunctionExistsError struct{ Message string }

FunctionExistsError reports duplicate callback registration.

func (*FunctionExistsError) Error 

func (e *FunctionExistsError) Error() string

func (*FunctionExistsError) Unwrap 

func (e *FunctionExistsError) Unwrap() error

type IdleTimeoutError 

type IdleTimeoutError struct{ Message string }

func (*IdleTimeoutError) Error 

func (e *IdleTimeoutError) Error() string

type InferenceProcessError 

type InferenceProcessError struct {
	Code    string
	Message string
}

InferenceProcessError is a stable error category returned by a supervised inference subprocess. Arbitrary native errors cannot preserve their concrete Go type across an OS process boundary, but their message and category do.

func (*InferenceProcessError) Error 

func (e *InferenceProcessError) Error() string

func (*InferenceProcessError) Is 

func (e *InferenceProcessError) Is(target error) bool

type InferenceRunner 

type InferenceRunner interface {
	Initialize(context.Context) error
	Run(context.Context, any) (any, error)
	Close(context.Context) error
}

InferenceRunner is the context-aware Go contract for a local model runner. A runner instance is owned by one inference executor and must tolerate an Initialize/Close pair even when no Run call succeeds.

func NewInferenceRunner 

func NewInferenceRunner(method string) (InferenceRunner, error)

NewInferenceRunner creates one registered runner without holding the registry lock across user code.

type InferenceRunnerFactory 

type InferenceRunnerFactory func() (InferenceRunner, error)

InferenceRunnerFactory constructs a fresh runner in the process that will execute it. Factories replace JavaScript import paths because Go links the implementation into both the parent and supervised child binary.

type InferenceRunnerFuncs 

type InferenceRunnerFuncs struct {
	InitializeFunc func(context.Context) error
	RunFunc        func(context.Context, any) (any, error)
	CloseFunc      func(context.Context) error
}

InferenceRunnerFuncs adapts three functions to InferenceRunner.

func (InferenceRunnerFuncs) Close 

func (r InferenceRunnerFuncs) Close(ctx context.Context) error

func (InferenceRunnerFuncs) Initialize 

func (r InferenceRunnerFuncs) Initialize(ctx context.Context) error

func (InferenceRunnerFuncs) Run 

func (r InferenceRunnerFuncs) Run(ctx context.Context, input any) (any, error)

type JobAcceptArguments 

type JobAcceptArguments = JobAcceptOptions

type JobAcceptOptions 

type JobAcceptOptions struct {
	Name       string
	Identity   string
	Metadata   string
	Attributes map[string]string
}

JobAcceptOptions controls the participant created for an accepted job.

type JobContext 

type JobContext[T any] struct {
	// contains filtered or unexported fields
}

JobContext is the job and room environment passed to an entrypoint.

func JobFromContext 

func JobFromContext[T any](ctx context.Context) (*JobContext[T], bool)

JobFromContext returns the typed job explicitly propagated by the executor.

func (*JobContext[T]) AddParticipantEntrypoint 

func (j *JobContext[T]) AddParticipantEntrypoint(callback ParticipantEntrypoint[T]) error

AddParticipantEntrypoint registers a bounded, asynchronous participant hook. The same function cannot be added twice.

func (*JobContext[T]) AddShutdownCallback 

func (j *JobContext[T]) AddShutdownCallback(callback ShutdownCallback) error

func (*JobContext[T]) Agent 

func (j *JobContext[T]) Agent() *lksdk.LocalParticipant

func (*JobContext[T]) Connect 

func (j *JobContext[T]) Connect(ctx context.Context, opts ConnectOptions) error

Connect joins the assigned room. It is idempotent, safe for concurrent use, and applies audio/video-only subscription rules to existing and future tracks.

func (*JobContext[T]) Context 

func (j *JobContext[T]) Context() context.Context

func (*JobContext[T]) DeleteRoom 

func (j *JobContext[T]) DeleteRoom(ctx context.Context, name string) error

DeleteRoom deletes the assigned room, or name when provided. Fake jobs are deliberately a no-op.

func (*JobContext[T]) Done 

func (j *JobContext[T]) Done() <-chan struct{}

func (*JobContext[T]) InferenceExecutor 

func (j *JobContext[T]) InferenceExecutor() ipc.InferenceExecutor

InferenceExecutor returns the worker-global local inference executor. In process-isolated jobs this is a cancellation-aware IPC client backed by the same shared runner instances as every other job.

func (*JobContext[T]) Info 

func (j *JobContext[T]) Info() RunningJobInfo

func (*JobContext[T]) IsConnected 

func (j *JobContext[T]) IsConnected() bool

func (*JobContext[T]) IsFakeJob 

func (j *JobContext[T]) IsFakeJob() bool

func (*JobContext[T]) Job 

func (j *JobContext[T]) Job() *livekit.Job

func (*JobContext[T]) Process 

func (j *JobContext[T]) Process() *JobProcess[T]

func (*JobContext[T]) RTCBridge 

func (j *JobContext[T]) RTCBridge() *rtcbridge.RTCBridge

RTCBridge returns the construction-time callback bridge installed on Room. Higher-level RoomIO uses it to subscribe without polling or unsafe callback replacement. The JobContext owns and closes the bridge.

func (*JobContext[T]) RegisterSession 

func (j *JobContext[T]) RegisterSession(options JobSessionRegistrationOptions) (*JobSessionRegistration, error)

RegisterSession claims the primary session slot atomically. An omitted recording choice inherits the dispatch setting. A secondary omitted choice is silently demoted to recording-off; an explicit recording-on choice fails.

func (*JobContext[T]) Room 

func (j *JobContext[T]) Room() *lksdk.Room

func (*JobContext[T]) SessionDirectory 

func (j *JobContext[T]) SessionDirectory() string

SessionDirectory is the traversal-safe per-job location for recordings and session artifacts. The directory is created lazily by the component that writes an artifact.

func (*JobContext[T]) Shutdown 

func (j *JobContext[T]) Shutdown(reason string)

Shutdown requests graceful job termination. Calling it more than once is a no-op and preserves the first reason.

func (*JobContext[T]) ShutdownReason 

func (j *JobContext[T]) ShutdownReason() string

func (*JobContext[T]) SimulationContext 

func (j *JobContext[T]) SimulationContext() (*SimulationContext[T], bool)

SimulationContext resolves and caches the simulation dispatch carried on the assigned job. Malformed or incomplete attributes are ignored after one warning, matching the TypeScript/Python entrypoint behavior.

func (*JobContext[T]) State 

func (j *JobContext[T]) State() *T

func (*JobContext[T]) WaitForParticipant 

func (j *JobContext[T]) WaitForParticipant(ctx context.Context, identity string, kinds ...lksdk.ParticipantKind) (*lksdk.RemoteParticipant, error)

WaitForParticipant waits for a non-agent remote participant. If kinds is empty, every participant kind except agent is accepted.

func (*JobContext[T]) WaitForParticipantAttribute 

func (j *JobContext[T]) WaitForParticipantAttribute(ctx context.Context, identity, attribute, value string) error

WaitForParticipantAttribute uses the JobContext's construction-time bridge.

func (*JobContext[T]) WaitForTrackPublication 

func (j *JobContext[T]) WaitForTrackPublication(ctx context.Context, options WaitForTrackPublicationOptions) (lksdk.TrackPublication, error)

WaitForTrackPublication uses the JobContext's construction-time bridge.

func (*JobContext[T]) WorkerID 

func (j *JobContext[T]) WorkerID() string

type JobEntrypoint 

type JobEntrypoint[T any] func(context.Context, *JobContext[T]) error

type JobProcess 

type JobProcess[T any] struct {
	// contains filtered or unexported fields
}

JobProcess is process-local state initialized by a prewarm callback. In process-executor mode PID is the child PID; in explicit in-process mode it is the current process PID.

func (*JobProcess[T]) PID 

func (p *JobProcess[T]) PID() int

func (*JobProcess[T]) State 

func (p *JobProcess[T]) State() *T

State returns the typed, process-local value shared by Prewarm and the job.

func (*JobProcess[T]) UserData 

func (p *JobProcess[T]) UserData() *T

UserData is a compatibility alias for State.

type JobRejectOptions 

type JobRejectOptions struct{}

JobRejectOptions is reserved for protocol-compatible rejection metadata. The current worker protocol carries only the availability decision.

type JobRequest 

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

JobRequest is the immutable view given to the request handler. Exactly one Accept or Reject call succeeds.

func (*JobRequest) Accept 

func (r *JobRequest) Accept(ctx context.Context, opts JobAcceptOptions) error

func (*JobRequest) AgentName 

func (r *JobRequest) AgentName() string

func (*JobRequest) Answered 

func (r *JobRequest) Answered() bool

func (*JobRequest) ID 

func (r *JobRequest) ID() string

func (*JobRequest) Job 

func (r *JobRequest) Job() *livekit.Job

func (*JobRequest) Publisher 

func (r *JobRequest) Publisher() *livekit.ParticipantInfo

func (*JobRequest) Reject 

func (r *JobRequest) Reject(ctx context.Context, opts JobRejectOptions) error

func (*JobRequest) Resuming 

func (r *JobRequest) Resuming() bool

func (*JobRequest) Room 

func (r *JobRequest) Room() *livekit.Room

type JobSessionLifecycle 

type JobSessionLifecycle interface {
	CloseSession(context.Context) error
	FinalizeSession(context.Context) error
}

JobSessionLifecycle is the import-cycle-free bridge between JobContext and a high-level agent session. CloseSession runs before the room disconnects; FinalizeSession then persists/uploads its report. A finalization failure is logged and does not turn an otherwise successful agent job into a failed assignment, matching the TypeScript/Python lifecycle.

type JobSessionLifecycleFuncs 

type JobSessionLifecycleFuncs struct {
	Close    func(context.Context) error
	Finalize func(context.Context) error
}

JobSessionLifecycleFuncs adapts functions without requiring a wrapper type.

func (JobSessionLifecycleFuncs) CloseSession 

func (f JobSessionLifecycleFuncs) CloseSession(ctx context.Context) error

func (JobSessionLifecycleFuncs) FinalizeSession 

func (f JobSessionLifecycleFuncs) FinalizeSession(ctx context.Context) error

type JobSessionRegistration 

type JobSessionRegistration struct {
	Primary          bool
	RecordingEnabled bool
	RedactionEnabled bool
	// contains filtered or unexported fields
}

JobSessionRegistration is the resolved claim returned to a session. Release rolls back a primary claim when Start fails; it is idempotent.

func (*JobSessionRegistration) Release 

func (r *JobSessionRegistration) Release()

Release rolls back this registration. Successful primary sessions normally remain registered until JobContext shutdown so the job runner can close and finalize them in the correct order.

type JobSessionRegistrationOptions 

type JobSessionRegistrationOptions struct {
	Lifecycle JobSessionLifecycle
	Recording Override[bool]
}

JobSessionRegistrationOptions controls the primary-session claim. Recording inherits livekit.Job.enable_recording when Recording is zero-valued. Use Use(false) or Disable[bool]() to explicitly disable it.

type LanguageCode 

type LanguageCode string

func AsLanguageCode 

func AsLanguageCode(language string) LanguageCode

func NormalizeLanguage 

func NormalizeLanguage(language string) LanguageCode

type LoadFunc 

type LoadFunc[T any] func(context.Context, *AgentServer[T]) (float64, error)

type LocalInferenceExecutor 

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

LocalInferenceExecutor owns one instance of every registered inference runner. A worker shares this executor across jobs so a native model is loaded once per worker host rather than once per job process.

Initialize eagerly constructs all registered runners. DoInference also initializes a runner lazily, which keeps standalone and console use useful without imposing import-time work. Close is idempotent and waits for accepted initialization and inference calls before closing runners.

func NewLocalInferenceExecutor 

func NewLocalInferenceExecutor(factories map[string]InferenceRunnerFactory) (*LocalInferenceExecutor, error)

NewLocalInferenceExecutor snapshots factories. A nil map uses the process registry; an explicitly empty map creates a valid executor that reports unknown methods without allocating runner state.

func (*LocalInferenceExecutor) Close 

func (e *LocalInferenceExecutor) Close(ctx context.Context) error

Close rejects new work, cancels initialization waiters, waits for accepted calls, then closes initialized runners concurrently. A caller deadline is reported, while cleanup continues and a later Close observes the final error.

func (*LocalInferenceExecutor) DoInference 

func (e *LocalInferenceExecutor) DoInference(ctx context.Context, method string, data any) (result any, err error)

DoInference dispatches one request to a registered runner. Runner calls may execute concurrently, matching the TypeScript/Python shared executor. The request context is propagated through initialization and Run.

func (*LocalInferenceExecutor) HasInferenceRunners 

func (e *LocalInferenceExecutor) HasInferenceRunners() bool

HasInferenceRunners reports whether the executor has any configured method without initializing it.

func (*LocalInferenceExecutor) Healthy 

func (e *LocalInferenceExecutor) Healthy() error

Healthy reports deterministic initialization/closure failures without triggering lazy initialization. Workers include it in readiness checks.

func (*LocalInferenceExecutor) Initialize 

func (e *LocalInferenceExecutor) Initialize(ctx context.Context) error

Initialize eagerly initializes all runners concurrently. It is safe to call repeatedly and concurrently. Partial initialization is retained so Close can release every successfully initialized runner; callers should close the executor when Initialize returns an error.

type LogFormat 

type LogFormat string

type LoggerOptions 

type LoggerOptions struct {
	Level       slog.Leveler
	Format      LogFormat
	Writer      io.Writer
	Handler     slog.Handler
	AddSource   bool
	SetDefault  bool
	ReplaceAttr func([]string, slog.Attr) slog.Attr
}

LoggerOptions configures the process-local framework logger. Handler wins over Writer/Format/Level. The zero value creates JSON logs on stdout without replacing slog.Default.

type MissingCredentialsError 

type MissingCredentialsError struct{ Name string }

func (*MissingCredentialsError) Error 

func (e *MissingCredentialsError) Error() string

type Override 

type Override[T any] struct {
	// contains filtered or unexported fields
}

Override models the Python/TypeScript inherit/use/disable tri-state. Its zero value inherits.

func Disable 

func Disable[T any]() Override[T]

func Use 

func Use[T any](value T) Override[T]

func (Override[T]) IsDisabled 

func (o Override[T]) IsDisabled() bool

func (Override[T]) IsInherited 

func (o Override[T]) IsInherited() bool

func (Override[T]) Resolve 

func (o Override[T]) Resolve(inherited T, inheritedEnabled bool) (T, bool)

Resolve returns the selected value and whether the component is enabled.

func (Override[T]) Value 

func (o Override[T]) Value() (T, bool)

type ParticipantEntrypoint 

type ParticipantEntrypoint[T any] func(context.Context, *JobContext[T], *lksdk.RemoteParticipant) error

type ParticipantNotFoundError 

type ParticipantNotFoundError struct{ Identity string }

ParticipantNotFoundError reports a participant that was absent before an attribute wait could be armed.

func (*ParticipantNotFoundError) Error 

func (e *ParticipantNotFoundError) Error() string

type ParticipantWaitDisconnectedError 

type ParticipantWaitDisconnectedError struct {
	Identity  string
	Attribute string
	Room      bool
}

ParticipantWaitDisconnectedError reports a requested participant or the room disappearing before a wait condition was satisfied.

func (*ParticipantWaitDisconnectedError) Error 

func (e *ParticipantWaitDisconnectedError) Error() string

type Plugin 

type Plugin interface {
	Title() string
	Version() string
	Package() string
	DownloadFiles(context.Context) error
}

func RegisteredPlugins 

func RegisteredPlugins() []Plugin

type PluginRegisteredEvent 

type PluginRegisteredEvent struct{ Plugin Plugin }

type PrewarmFunc 

type PrewarmFunc[T any] func(context.Context, *JobProcess[T]) error

type RequestHandler 

type RequestHandler func(context.Context, *JobRequest) error

type ResolvedLoggerOptions 

type ResolvedLoggerOptions struct {
	Level      slog.Level
	Format     LogFormat
	AddSource  bool
	SetDefault bool
}

func CurrentLoggerOptions 

func CurrentLoggerOptions() (ResolvedLoggerOptions, bool)

type ResourceBag 

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

ResourceBag carries process-prewarmed dependencies without serialization or reflection on access. It is safe for concurrent jobs in in-process mode.

func NewResourceBag 

func NewResourceBag() *ResourceBag

func (*ResourceBag) Delete 

func (b *ResourceBag) Delete(key string)

func (*ResourceBag) Put 

func (b *ResourceBag) Put(key string, value any) error

type RunningJobInfo 

type RunningJobInfo struct {
	AcceptArguments JobAcceptOptions
	Job             *livekit.Job
	URL             string
	Token           SecretString
	WorkerID        string
	APIKey          SecretString
	APISecret       SecretString
	FakeJob         bool
	// SessionDirectory overrides the per-job directory used for recordings and
	// reports. Production assignments normally leave this empty and receive a
	// traversal-safe directory below os.TempDir. Console runners set it to their
	// user-visible session directory.
	SessionDirectory string
}

RunningJobInfo is the complete, immutable assignment passed to an executor. API credentials are redacted when formatted or marshaled.

func (RunningJobInfo) Clone 

func (i RunningJobInfo) Clone() RunningJobInfo

Clone returns a deep copy suitable for crossing ownership boundaries.

type RuntimeCPUMonitor 

type RuntimeCPUMonitor struct{}

RuntimeCPUMonitor uses the stable runtime/metrics CPU classes. It measures this Go process rather than unrelated host work and is portable without cgo.

func NewRuntimeCPUMonitor 

func NewRuntimeCPUMonitor() *RuntimeCPUMonitor

func (*RuntimeCPUMonitor) CPUCount 

func (*RuntimeCPUMonitor) CPUCount() float64

func (*RuntimeCPUMonitor) CPUPercent 

func (*RuntimeCPUMonitor) CPUPercent(ctx context.Context, interval time.Duration) (float64, error)

type Scenario 

type Scenario = livekit.Scenario

type ScenarioGroup 

type ScenarioGroup = livekit.ScenarioGroup

type ScenarioUserdata 

type ScenarioUserdata map[string]any

type SecretString 

type SecretString string

func NewSecretString 

func NewSecretString(value string) SecretString

func (SecretString) GoString 

func (SecretString) GoString() string

func (SecretString) MarshalJSON 

func (SecretString) MarshalJSON() ([]byte, error)

func (SecretString) MarshalText 

func (SecretString) MarshalText() ([]byte, error)

func (SecretString) Reveal 

func (s SecretString) Reveal() string

func (SecretString) String 

func (SecretString) String() string

type ServerOptions 

type ServerOptions[T any] struct {
	// Agent is the defineAgent-compatible immutable module. Direct entrypoint
	// fields remain supported for idiomatic Go construction.
	Agent         *AgentDefinition[T]
	JobEntrypoint JobEntrypoint[T]
	// Entrypoint is the legacy compatibility name. JobEntrypoint is canonical.
	Entrypoint     JobEntrypoint[T]
	RequestHandler RequestHandler
	// RequestFunc is the TypeScript-compatible alias for RequestHandler.
	RequestFunc     RequestHandler
	Prewarm         PrewarmFunc[T]
	OnSimulationEnd SimulationEndFunc[T]

	AgentName      string
	AgentNameIsEnv bool
	ServerType     ServerType
	Deployment     string

	ExecutorMode      ExecutorMode
	MaxConcurrentJobs int
	NumIdleProcesses  int

	LoadFunc      LoadFunc[T]
	LoadThreshold float64
	Production    bool
	Simulation    bool

	DrainTimeout             time.Duration
	ShutdownProcessTimeout   time.Duration
	InitializeProcessTimeout time.Duration
	AssignmentTimeout        time.Duration
	StatusUpdateInterval     time.Duration
	MaxReconnects            int
	MaxReconnectsSet         bool

	Permissions    WorkerPermissions
	PermissionsSet bool

	URL         string
	WSURL       string
	APIKey      SecretString
	APISecret   SecretString
	WorkerToken SecretString

	Host string
	Port int

	JobMemoryWarnMB    int
	JobMemoryWarnMBSet bool
	JobMemoryLimitMB   int
	Logger             *slog.Logger
}

ServerOptions configures the worker runtime. Zero-valued optional fields use the same production/development defaults as the Python and TypeScript SDKs.

type ServerType 

type ServerType = livekit.JobType

type SessionConnectOptions 

type SessionConnectOptions struct {
	STT                    APIConnectOptions
	LLM                    APIConnectOptions
	TTS                    APIConnectOptions
	MaxUnrecoverableErrors int
}

func (SessionConnectOptions) Resolve 

func (o SessionConnectOptions) Resolve() SessionConnectOptions

type ShutdownCallback 

type ShutdownCallback func(context.Context, string) error

type SimulationContext 

type SimulationContext[T any] struct {
	// contains filtered or unexported fields
}

SimulationContext carries the immutable dispatch plus final verdict state for one simulated job. The user verdict can veto a simulator success but cannot turn a simulator failure into a success.

func NewSimulationContext 

func NewSimulationContext[T any](dispatch *SimulationDispatch, job *JobContext[T]) (*SimulationContext[T], error)

func (*SimulationContext[T]) BeginFinalize 

func (s *SimulationContext[T]) BeginFinalize(verdict SimulationVerdict, run *SimulationRun)

BeginFinalize is called by the session host before the on-simulation-end hook. It makes the simulator verdict and run visible atomically.

func (*SimulationContext[T]) Dispatch 

func (s *SimulationContext[T]) Dispatch() *SimulationDispatch

func (*SimulationContext[T]) EffectiveVerdict 

func (s *SimulationContext[T]) EffectiveVerdict() (SimulationVerdict, error)

func (*SimulationContext[T]) Fail 

func (s *SimulationContext[T]) Fail(reason string)

Fail records the user's veto. The last call wins.

func (*SimulationContext[T]) JobContext 

func (s *SimulationContext[T]) JobContext() *JobContext[T]

func (*SimulationContext[T]) JobID 

func (s *SimulationContext[T]) JobID() string

func (*SimulationContext[T]) Mode 

func (s *SimulationContext[T]) Mode() SimulationMode

func (*SimulationContext[T]) Run 

func (s *SimulationContext[T]) Run() (*SimulationRun, bool)

func (*SimulationContext[T]) RunID 

func (s *SimulationContext[T]) RunID() string

func (*SimulationContext[T]) Scenario 

func (s *SimulationContext[T]) Scenario() *Scenario

func (*SimulationContext[T]) SimulatorVerdict 

func (s *SimulationContext[T]) SimulatorVerdict() (SimulationVerdict, error)

func (*SimulationContext[T]) UserVerdict 

func (s *SimulationContext[T]) UserVerdict() (SimulationVerdict, bool)

func (*SimulationContext[T]) Userdata 

func (s *SimulationContext[T]) Userdata() (ScenarioUserdata, error)

Userdata decodes the scenario's JSON object. It returns an empty map for an empty field and preserves integer tokens as json.Number.

type SimulationDispatch 

type SimulationDispatch = livekit.SimulationDispatch

func ParseSimulationDispatch 

func ParseSimulationDispatch(raw string) (*SimulationDispatch, error)

ParseSimulationDispatch decodes the proto-JSON job attribute. Unknown fields are discarded so a newer LiveKit server can extend the message without breaking an older agent binary.

type SimulationEndFunc 

type SimulationEndFunc[T any] func(context.Context, *SimulationContext[T]) error

type SimulationMode 

type SimulationMode = livekit.SimulationMode

type SimulationRun 

type SimulationRun = livekit.SimulationRun

type SimulationRunJob 

type SimulationRunJob = livekit.SimulationRun_Job

type SimulationVerdict 

type SimulationVerdict struct {
	Success bool   `json:"success"`
	Reason  string `json:"reason"`
}

SimulationVerdict is a pass/fail result with a human-readable reason.

type TimedString 

type TimedString struct {
	Text            string
	StartTime       *time.Duration
	EndTime         *time.Duration
	Confidence      *float64
	StartTimeOffset *time.Duration
	SpeakerID       *string
}

func CreateTimedString 

func CreateTimedString(options TimedStringOptions) TimedString

func NewTimedString 

func NewTimedString(text string, start, end time.Duration) TimedString

type TimedStringOptions 

type TimedStringOptions struct {
	Text            string
	StartTime       *time.Duration
	EndTime         *time.Duration
	Confidence      *float64
	StartTimeOffset *time.Duration
	SpeakerID       *string
}

type UnexpectedModelBehavior 

type UnexpectedModelBehavior struct {
	Message string
	Cause   error
}

func (*UnexpectedModelBehavior) Error 

func (e *UnexpectedModelBehavior) Error() string

func (*UnexpectedModelBehavior) Unwrap 

func (e *UnexpectedModelBehavior) Unwrap() error

type WaitForTrackPublicationOptions 

type WaitForTrackPublicationOptions struct {
	Identity            string
	Kind                lksdk.TrackKind
	IncludeLocal        bool
	WaitForSubscription bool
	EventCapacity       int
}

WaitForTrackPublicationOptions matches the TypeScript/Python helper. Empty Identity and Kind values mean any participant and any kind respectively. Local publications resolve when published and ignore WaitForSubscription.

type Worker 

type Worker[T any] = AgentServer[T]

Worker is the deprecated name for AgentServer.

func NewWorker 

func NewWorker[T any](options WorkerOptions[T]) (*Worker[T], error)

NewWorker is the deprecated constructor name for NewAgentServer.

type WorkerError 

type WorkerError struct {
	Message string
	Cause   error
}

func (*WorkerError) Error 

func (e *WorkerError) Error() string

func (*WorkerError) Unwrap 

func (e *WorkerError) Unwrap() error

type WorkerEvent 

type WorkerEvent struct {
	Type       WorkerEventType
	WorkerID   string
	ServerInfo *livekit.ServerInfo
	Message    *livekit.WorkerMessage
	Error      error
}

type WorkerEventType 

type WorkerEventType string
const (
	WorkerEventRegistered WorkerEventType = "worker_registered"
	WorkerEventMessage    WorkerEventType = "worker_msg"
	WorkerEventClosed     WorkerEventType = "worker_closed"
)

type WorkerOptions 

type WorkerOptions[T any] = ServerOptions[T]

WorkerOptions is retained as a source-compatible alias.

type WorkerPermissions 

type WorkerPermissions struct {
	CanPublish            bool
	CanSubscribe          bool
	CanPublishData        bool
	CanUpdateMetadata     bool
	CanPublishSources     []livekit.TrackSource
	Hidden                bool
	CanSubscribeMetrics   bool
	CanManageAgentSession bool
}

WorkerPermissions are applied to every agent participant created by this worker. Use DefaultWorkerPermissions when modifying individual fields.

func DefaultWorkerPermissions 

func DefaultWorkerPermissions() WorkerPermissions

Source Files

View all Source files

Directories

Path Synopsis
Package agentscli provides the embedded LiveKit Agents command-line runtime.
 Package agentscli provides the embedded LiveKit Agents command-line runtime.
benchmarks
coldstart/core command
coldstart/elevenlabs command
coldstart/measure command
coldstart/voice command
Package beta retains the temporary agents-js compatibility surface.
 Package beta retains the temporary agents-js compatibility surface.
tools
Package tools contains the beta LiveKit agent tools.
 Package tools contains the beta LiveKit agent tools.
workflows
Package workflows contains deprecated compatibility aliases.
 Package workflows contains deprecated compatibility aliases.
Package inference implements the LiveKit Cloud Inference adapters.
 Package inference implements the LiveKit Cloud Inference adapters.
internal
apimanifest
Package apimanifest produces the deterministic public-API snapshot used by release checks.
 Package apimanifest produces the deterministic public-API snapshot used by release checks.
cmd/apimanifest command
cmd/depaudit command
Command depaudit verifies the reviewed dependency and licensing policy for this module.
 Command depaudit verifies the reviewed dependency and licensing policy for this module.
paritymanifest
Package paritymanifest validates the checked-in cross-language API parity ledger.
 Package paritymanifest validates the checked-in cross-language API parity ledger.
releaseversion command
Command releaseversion prints the SDK version without a leading v.
 Command releaseversion prints the SDK version without a leading v.
workerprotocol
Package workerprotocol implements the LiveKit agent worker WebSocket wire protocol.
 Package workerprotocol implements the LiveKit agent worker WebSocket wire protocol.
Package ipc contains the public cross-process inference contract.
 Package ipc contains the public cross-process inference contract.
llm
providerformat
Package providerformat converts LiveKit chat contexts into the wire-neutral request shapes consumed by OpenAI-compatible, Google Gemini, and Mistral APIs.
 Package providerformat converts LiveKit chat contexts into the wire-neutral request shapes consumed by OpenAI-compatible, Google Gemini, and Mistral APIs.
plugins
elevenlabs
Package elevenlabs provides production ElevenLabs speech-to-text and text-to-speech clients for LiveKit Agents.
 Package elevenlabs provides production ElevenLabs speech-to-text and text-to-speech clients for LiveKit Agents.
Package rtcbridge multiplexes one LiveKit RoomCallback into bounded, context-first subscriptions.
 Package rtcbridge multiplexes one LiveKit RoomCallback into bounded, context-first subscriptions.
Package stream provides cancellation-aware, bounded streams used throughout the SDK.
 Package stream provides cancellation-aware, bounded streams used throughout the SDK.
stt
testing
Package stttest contains deterministic STT test doubles.
 Package stttest contains deterministic STT test doubles.
Package tokenize provides the sentence, word, paragraph, and English hyphenation primitives used by streaming transcription and TTS.
 Package tokenize provides the sentence, word, paragraph, and English hyphenation primitives used by streaming transcription and TTS.
Package transcription contains the legacy public text/audio synchronizer.
 Package transcription contains the legacy public text/audio synchronizer.
Package voice implements LiveKit's agent-session state machine and media pipeline.
 Package voice implements LiveKit's agent-session state machine and media pipeline.
avatar
Package avatar implements provider-independent LiveKit avatar sessions and bounded PCM transports.
 Package avatar implements provider-independent LiveKit avatar sessions and bounded PCM transports.
avatar/roomioadapter
Package roomioadapter connects avatar readiness waits to a roomio RTCBridge.
 Package roomioadapter connects avatar readiness waits to a roomio RTCBridge.
backgroundaudio
Package backgroundaudio provides the background-audio player shipped by the LiveKit Agents TypeScript SDK, adapted to context-first Go APIs.
 Package backgroundaudio provides the background-audio player shipped by the LiveKit Agents TypeScript SDK, adapted to context-first Go APIs.
livekit
Package livekit binds a voice.AgentSession to the current job's LiveKit room, recording, observability, report, and primary-session lifecycle.
 Package livekit binds a voice.AgentSession to the current job's LiveKit room, recording, observability, report, and primary-session lifecycle.
recorderio
Package recorderio records the user input and agent output sides of a voice session into a synchronized stereo Ogg/Opus file.
 Package recorderio records the user input and agent output sides of a voice session into a synchronized stereo Ogg/Opus file.
roomio
testing
Package voicetest provides deterministic model doubles and assertion helpers for voice-agent tests.
 Package voicetest provides deterministic model doubles and assertion helpers for voice-agent tests.
transcription
Package transcription provides streaming transcript synchronization and text transforms for voice sessions.
 Package transcription provides streaming transcript synchronization and text transforms for voice sessions.
Package workflows contains reusable, context-first agent workflows.
 Package workflows contains reusable, context-first agent workflows.
livekit
Package livekit provides the optional server-sdk-go/RoomIO backend for the stable warm-transfer workflow.
 Package livekit provides the optional server-sdk-go/RoomIO backend for the stable warm-transfer workflow.

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.