voice_assistant From livekit-agents cannot be imported

Question

I already download package livekit-agents and when I want to use, I can't find module voice_assistant. Is there a problem of my livekit-agents?

I already try upgrade my livekit-agents package to current version, but still not working.

• Community: Please provide enough code so others can better understand or reproduce the problem.

Answer 1

This quickstart on the LiveKit.io site is a great way to get started: https://docs.livekit.io/agents/start/voice-ai/

Here is a good way to get started. Use this for your requirements.txt

livekit-plugins-noise-cancellation
livekit-agents[google,openai,silero,deepgram,cartesia,elevenlabs,turn-detector,bey,tavus,bithuman,rime,speechmatics,groq,assemblyai]~=1.2

This this is a basic_agent.py here

import logging
from collections.abc import AsyncIterable
from dotenv import load_dotenv
from livekit import rtc
from livekit.agents import (
 Agent,
 AgentSession,
 JobContext,
 JobProcess,
 MetricsCollectedEvent,
 ModelSettings,
 RoomInputOptions,
 RoomOutputOptions,
 RunContext,
 WorkerOptions,
 cli,
 metrics,
)
from livekit.agents.llm import function_tool
from livekit.plugins import deepgram, openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
# uncomment to enable Krisp background voice/noise cancellation
# from livekit.plugins import noise_cancellation
logger = logging.getLogger("basic-agent")
load_dotenv()
class MyAgent(Agent):
 def __init__(self) -> None:
 super().__init__(
 instructions="Your name is Kelly. You would interact with users via voice."
 "with that in mind keep your responses concise and to the point."
 "do not use emojis, asterisks, markdown, or other special characters in your responses."
 "You are curious and friendly, and have a sense of humor."
 "you will speak english to the user",
 )
 async def on_enter(self):
 # when the agent is added to the session, it'll generate a reply
 # according to its instructions
 self.session.generate_reply()
 async def tts_node(
 self, text: AsyncIterable[str], model_settings: ModelSettings
 ) -> AsyncIterable[rtc.AudioFrame]:
 # Markdown and emoji filters are enabled in `Agent.tts_node`, markdown symbols
 # and emojis will be removed from the text sent to the TTS model
 # To disable these filters, customize the `tts_node` method and
 # use `return Agent.default.tts_node(self, text, model_settings)` instead
 return super().tts_node(text, model_settings)
 # all functions annotated with @function_tool will be passed to the LLM when this
 # agent is active
 @function_tool
 async def lookup_weather(
 self, context: RunContext, location: str, latitude: str, longitude: str
 ):
 """Called when the user asks for weather related information.
 Ensure the user's location (city or region) is provided.
 When given a location, please estimate the latitude and longitude of the location and
 do not ask the user for them.
 Args:
 location: The location they are asking for
 latitude: The latitude of the location, do not ask user for it
 longitude: The longitude of the location, do not ask user for it
 """
 logger.info(f"Looking up weather for {location}")
 return "sunny with a temperature of 70 degrees."
def prewarm(proc: JobProcess):
 proc.userdata["vad"] = silero.VAD.load()
async def entrypoint(ctx: JobContext):
 # each log entry will include these fields
 ctx.log_context_fields = {
 "room": ctx.room.name,
 }
 session = AgentSession(
 vad=ctx.proc.userdata["vad"],
 # any combination of STT, LLM, TTS, or realtime API can be used
 llm=openai.LLM(model="gpt-4o-mini"),
 stt=openai.STT(),
 tts=openai.TTS(voice="ash"),
 # allow the LLM to generate a response while waiting for the end of turn
 preemptive_generation=True,
 # sometimes background noise could interrupt the agent session, these are considered false positive interruptions
 # when it's detected, you may resume the agent's speech
 resume_false_interruption=True,
 false_interruption_timeout=1.0,
 min_interruption_duration=0.2, # with false interruption resume, interruption can be more sensitive
 # use LiveKit's turn detection model
 turn_detection=MultilingualModel(),
 )
 # log metrics as they are emitted, and total usage after session is over
 usage_collector = metrics.UsageCollector()
 @session.on("metrics_collected")
 def _on_metrics_collected(ev: MetricsCollectedEvent):
 metrics.log_metrics(ev.metrics)
 usage_collector.collect(ev.metrics)
 async def log_usage():
 summary = usage_collector.get_summary()
 logger.info(f"Usage: {summary}")
 # shutdown callbacks are triggered when the session is over
 ctx.add_shutdown_callback(log_usage)
 await session.start(
 agent=MyAgent(),
 room=ctx.room,
 room_input_options=RoomInputOptions(
 # uncomment to enable Krisp BVC noise cancellation
 # noise_cancellation=noise_cancellation.BVC(),
 ),
 room_output_options=RoomOutputOptions(transcription_enabled=True),
 )
if __name__ == "__main__":
 cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))

You will need to make sure you have your the following environment variables

These come from you LiveKit account here https://cloud.livekit.io/projects/p_/settings/project

You can skip this if you just run local with console

LIVEKIT_URL

LIVEKIT_API_KEY

LIVEKIT_API_SECRET

Get OpenAI API key or set a different LLM provider above

OPENAI_API_KEY

Now install the dependencies:

pip install -r requirements.txt

then run your agent (console only)

python basic_agent.py console

or connect from a web client

python basic_agent.py dev and then open https://agents-playground.livekit.io to connect

That should get you to a working agent quickly. If you get stuck I can help you more in our community Slack channel here: https://livekit.io/join-slack

Good Luck.