wehub-resource-sync
141 lines
5.2 KiB
Python
141 lines
5.2 KiB
Python
import logging
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
from livekit.agents import (
|
||
Agent,
|
||
AgentServer,
|
||
AgentSession,
|
||
JobContext,
|
||
MetricsCollectedEvent,
|
||
RunContext,
|
||
TurnHandlingOptions,
|
||
cli,
|
||
inference,
|
||
metrics,
|
||
room_io,
|
||
text_transforms,
|
||
)
|
||
from livekit.agents.beta import EndCallTool
|
||
from livekit.agents.llm import function_tool
|
||
|
||
# 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, built by LiveKit. 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 over voice.",
|
||
tools=[EndCallTool()],
|
||
)
|
||
|
||
async def on_enter(self) -> None:
|
||
# when the agent is added to the session, it'll generate a reply
|
||
# according to its instructions
|
||
self.session.generate_reply(instructions="greet the user and introduce yourself")
|
||
|
||
# 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
|
||
) -> 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."
|
||
|
||
|
||
server = AgentServer()
|
||
|
||
|
||
@server.rtc_session()
|
||
async def entrypoint(ctx: JobContext) -> None:
|
||
# each log entry will include these fields
|
||
ctx.log_context_fields = {
|
||
"room": ctx.room.name,
|
||
}
|
||
session: AgentSession = AgentSession(
|
||
# Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand
|
||
# See all available models at https://docs.livekit.io/agents/models/stt/
|
||
stt=inference.STT("deepgram/nova-3", language="multi"),
|
||
# A Large Language Model (LLM) is your agent's brain, processing user input and generating a response
|
||
# See all available models at https://docs.livekit.io/agents/models/llm/
|
||
llm=inference.LLM("openai/gpt-4.1-mini"),
|
||
# Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear
|
||
# See all available models as well as voice selections at https://docs.livekit.io/agents/models/tts/
|
||
tts=inference.TTS("cartesia/sonic-3", voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc"),
|
||
turn_handling=TurnHandlingOptions(
|
||
interruption={
|
||
# 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,
|
||
},
|
||
# allow the LLM to generate a response while waiting for the end of turn
|
||
# See more at https://docs.livekit.io/agents/build/audio/#preemptive-generation
|
||
preemptive_generation={"enabled": True, "max_retries": 3},
|
||
),
|
||
# blocks interruptions for a few seconds after the agent starts speaking to allow client to calibrate AEC
|
||
aec_warmup_duration=3.0,
|
||
tts_text_transforms=[
|
||
"filter_emoji",
|
||
"filter_markdown",
|
||
text_transforms.replace({"LiveKit": "<<ˈ|l|aɪ|v|k|ɪ|t>>"}),
|
||
],
|
||
# automatically detect keyterms and apply them to the STT per user turn
|
||
keyterms_options={
|
||
"keyterms": ["LiveKit"],
|
||
"keyterm_detection": {
|
||
"enabled": True,
|
||
"turn_interval": 1, # increase to reduce LLM API calls
|
||
},
|
||
},
|
||
)
|
||
|
||
@session.on("metrics_collected")
|
||
def _on_metrics_collected(ev: MetricsCollectedEvent) -> None:
|
||
if ev.metrics.type == "stt_metrics":
|
||
return
|
||
metrics.log_metrics(ev.metrics)
|
||
|
||
async def log_usage():
|
||
logger.info(f"Usage: {session.usage}")
|
||
|
||
# shutdown callbacks are triggered when the session is over
|
||
ctx.add_shutdown_callback(log_usage)
|
||
|
||
await session.start(
|
||
agent=MyAgent(),
|
||
room=ctx.room,
|
||
room_options=room_io.RoomOptions(
|
||
audio_input=room_io.AudioInputOptions(
|
||
# uncomment to enable the Krisp BVC noise cancellation
|
||
# noise_cancellation=noise_cancellation.BVC(),
|
||
),
|
||
),
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cli.run_app(server)
|