v1.0

examples/minimal_worker.py

@@ -1,18 +1,28 @@
 import logging
 
+from dotenv import load_dotenv
 from livekit.agents import AutoSubscribe, JobContext, WorkerOptions, WorkerType, cli
+from livekit.agents.pipeline import ChatCLI, PipelineAgent, AgentTask
+from livekit.plugins import cartesia, deepgram, openai, silero
 
 logger = logging.getLogger("my-worker")
 logger.setLevel(logging.INFO)
 
+load_dotenv()
 
-async def entrypoint(ctx: JobContext):
-    logger.info("starting entrypoint")
-
-    await ctx.connect(auto_subscribe=AutoSubscribe.SUBSCRIBE_ALL)
 
-    logger.info("connected to the room")
-    # add your agent logic here!
+async def entrypoint(ctx: JobContext):
+    agent = PipelineAgent(
+        task=AgentTask(
+            instructions="Talk to me!",
+            llm=openai.realtime.RealtimeModel(),
+        )
+    )
+    agent.start()
+
+    # start a chat inside the CLI
+    chat_cli = ChatCLI(agent)
+    await chat_cli.run()
 
 
 if __name__ == "__main__":

livekit-agents/livekit/agents/__init__.py

@@ -21,11 +21,9 @@
     pipeline,
     stt,
     tokenize,
-    transcription,
     tts,
     utils,
     vad,
-    voice_assistant,
 )
 from ._exceptions import (
     APIConnectionError,
@@ -69,10 +67,8 @@
     "tokenize",
     "llm",
     "metrics",
-    "transcription",
     "pipeline",
     "multimodal",
-    "voice_assistant",
     "cli",
     "AssignmentTimeoutError",
     "APIConnectionError",

livekit-agents/livekit/agents/cli/cli.py

@@ -2,14 +2,14 @@
 import pathlib
 import signal
 import sys
+import threading
 
 import click
-from livekit.protocol import models
 
 from .. import utils
 from ..log import logger
 from ..plugin import Plugin
-from ..worker import Worker, WorkerOptions
+from ..worker import JobExecutorType, Worker, WorkerOptions
 from . import proto
 from .log import setup_logging
 
@@ -58,6 +58,7 @@ def start(
             log_level=log_level,
             devmode=False,
             asyncio_debug=False,
+            register=True,
             watch=False,
             drain_timeout=drain_timeout,
         )
@@ -105,7 +106,63 @@ def dev(
         asyncio_debug: bool,
         watch: bool,
     ) -> None:
-        _run_dev(opts, log_level, url, api_key, api_secret, asyncio_debug, watch)
+        opts.ws_url = url or opts.ws_url
+        opts.api_key = api_key or opts.api_key
+        opts.api_secret = api_secret or opts.api_secret
+        args = proto.CliArgs(
+            opts=opts,
+            log_level=log_level,
+            devmode=True,
+            asyncio_debug=asyncio_debug,
+            watch=watch,
+            drain_timeout=0,
+            register=False,
+        )
+
+        _run_dev(args)
+
+    @cli.command(help="Start a new chat")
+    @click.option(
+        "--url",
+        envvar="LIVEKIT_URL",
+        help="LiveKit server or Cloud project's websocket URL",
+    )
+    @click.option(
+        "--api-key",
+        envvar="LIVEKIT_API_KEY",
+        help="LiveKit server or Cloud project's API key",
+    )
+    @click.option(
+        "--api-secret",
+        envvar="LIVEKIT_API_SECRET",
+        help="LiveKit server or Cloud project's API secret",
+    )
+    def chat(
+        url: str,
+        api_key: str,
+        api_secret: str,
+    ) -> None:
+        # keep everything inside the same process when using the chat mode
+        opts.job_executor_type = JobExecutorType.THREAD
+        opts.ws_url = url or opts.ws_url
+        opts.api_key = api_key or opts.api_key
+        opts.api_secret = api_secret or opts.api_secret
+
+        chat_name = utils.shortuuid("chat_cli_")
+
+        args = proto.CliArgs(
+            opts=opts,
+            log_level="ERROR",
+            devmode=True,
+            asyncio_debug=False,
+            watch=False,
+            drain_timeout=0,
+            register=False,
+            simulate_job=proto.SimulateJobArgs(
+                room=chat_name,
+            ),
+        )
+        _run_dev(args)
 
     @cli.command(help="Connect to a specific room")
     @click.option(
@@ -155,18 +212,25 @@ def connect(
         room: str,
         participant_identity: str,
     ) -> None:
-        _run_dev(
-            opts,
-            log_level,
-            url,
-            api_key,
-            api_secret,
-            asyncio_debug,
-            watch,
-            room,
-            participant_identity,
+        opts.ws_url = url or opts.ws_url
+        opts.api_key = api_key or opts.api_key
+        opts.api_secret = api_secret or opts.api_secret
+        args = proto.CliArgs(
+            opts=opts,
+            log_level=log_level,
+            devmode=True,
+            register=False,
+            asyncio_debug=asyncio_debug,
+            watch=watch,
+            drain_timeout=0,
+            simulate_job=proto.SimulateJobArgs(
+                room=room,
+                participant_identity=participant_identity,
+            ),
         )
 
+        _run_dev(args)
+
     @cli.command(help="Download plugin dependency files")
     @click.option(
         "--log-level",
@@ -188,34 +252,12 @@ def download_files(log_level: str) -> None:
 
 
 def _run_dev(
-    opts: WorkerOptions,
-    log_level: str,
-    url: str,
-    api_key: str,
-    api_secret: str,
-    asyncio_debug: bool,
-    watch: bool,
-    room: str = "",
-    participant_identity: str = "",
+    args: proto.CliArgs,
 ):
-    opts.ws_url = url or opts.ws_url
-    opts.api_key = api_key or opts.api_key
-    opts.api_secret = api_secret or opts.api_secret
-    args = proto.CliArgs(
-        opts=opts,
-        log_level=log_level,
-        devmode=True,
-        asyncio_debug=asyncio_debug,
-        watch=watch,
-        drain_timeout=0,
-        room=room,
-        participant_identity=participant_identity,
-    )
-
-    if watch:
+    if args.watch:
         from .watcher import WatchServer
 
-        setup_logging(log_level, args.devmode)
+        setup_logging(args.log_level, args.devmode)
         main_file = pathlib.Path(sys.argv[0]).parent
 
         async def _run_loop():
@@ -236,27 +278,34 @@ def run_worker(args: proto.CliArgs) -> None:
     setup_logging(args.log_level, args.devmode)
     args.opts.validate_config(args.devmode)
 
-    loop = asyncio.get_event_loop()
-    worker = Worker(args.opts, devmode=args.devmode, loop=loop)
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    worker = Worker(args.opts, devmode=args.devmode, register=args.register, loop=loop)
 
     loop.set_debug(args.asyncio_debug)
     loop.slow_callback_duration = 0.1  # 100ms
     utils.aio.debug.hook_slow_callbacks(2)
 
-    if args.room and args.reload_count == 0:
-        # directly connect to a specific room
-        @worker.once("worker_registered")
-        def _connect_on_register(worker_id: str, server_info: models.ServerInfo):
-            logger.info("connecting to room %s", args.room)
-            loop.create_task(worker.simulate_job(args.room, args.participant_identity))
+    @worker.once("worker_started")
+    def _worker_started():
+        if args.simulate_job and args.reload_count == 0:
+            logger.info("connecting to room %s", args.simulate_job.room)
+            loop.create_task(
+                worker.simulate_job(
+                    args.simulate_job.room, args.simulate_job.participant_identity
+                )
+            )
 
     try:
 
         def _signal_handler():
             raise KeyboardInterrupt
 
-        for sig in (signal.SIGINT, signal.SIGTERM):
-            loop.add_signal_handler(sig, _signal_handler)
+        if threading.current_thread() is threading.main_thread():
+            for sig in (signal.SIGINT, signal.SIGTERM):
+                loop.add_signal_handler(sig, _signal_handler)
+
     except NotImplementedError:
         # TODO(theomonnom): add_signal_handler is not implemented on win
         pass

livekit-agents/livekit/agents/cli/proto.py

@@ -12,6 +12,12 @@
 from ..worker import WorkerOptions
 
 
+@dataclass
+class SimulateJobArgs:
+    room: str = ""
+    participant_identity: str = ""
+
+
 @dataclass
 class CliArgs:
     opts: WorkerOptions
@@ -20,8 +26,11 @@ class CliArgs:
     asyncio_debug: bool
     watch: bool
     drain_timeout: int
-    room: str = ""
-    participant_identity: str = ""
+
+    # register the worker to the worker pool
+    register: bool = True
+
+    simulate_job: SimulateJobArgs | None = None
 
     # amount of time this worker has been reloaded
     reload_count: int = 0

livekit-agents/livekit/agents/debug/__init__.py

@@ -0,0 +1,7 @@
+from .tracing import Tracing, TracingGraph, TracingHandle
+
+__all__ = [
+    "Tracing",
+    "TracingGraph",
+    "TracingHandle",
+]