forked from livekit-examples/agent-starter-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathagent.py
More file actions
137 lines (111 loc) · 5.48 KB
/
Copy pathagent.py
File metadata and controls
137 lines (111 loc) · 5.48 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
import logging
from dotenv import load_dotenv
from livekit.agents import (
NOT_GIVEN,
Agent,
AgentFalseInterruptionEvent,
AgentSession,
JobContext,
JobProcess,
MetricsCollectedEvent,
RoomInputOptions,
RunContext,
WorkerOptions,
cli,
metrics,
)
from livekit.agents.llm import function_tool
from livekit.plugins import cartesia, deepgram, noise_cancellation, openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
logger = logging.getLogger("agent")
load_dotenv(".env.local")
class Assistant(Agent):
def __init__(self) -> None:
super().__init__(
instructions="""You are a helpful voice AI assistant.
You eagerly assist users with their questions by providing information from your extensive knowledge.
Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
You are curious, friendly, and have a sense of humor.""",
)
# 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):
"""Use this tool to look up current weather information in the given location.
If the location is not supported by the weather service, the tool will indicate this. You must tell the user the location's weather is unavailable.
Args:
location: The location to look up weather information for (e.g. city name)
"""
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):
# Logging setup
# Add any other context you want in all log entries here
ctx.log_context_fields = {
"room": ctx.room.name,
}
# Set up a voice AI pipeline using OpenAI, Cartesia, Deepgram, and the LiveKit turn detector
session = AgentSession(
# A Large Language Model (LLM) is your agent's brain, processing user input and generating a response
# See all providers at https://docs.livekit.io/agents/integrations/llm/
llm=openai.LLM(model="gpt-4o-mini"),
# Speech-to-text (STT) is your agent's ears, turning the user's speech into text that the LLM can understand
# See all providers at https://docs.livekit.io/agents/integrations/stt/
stt=deepgram.STT(model="nova-3", language="multi"),
# Text-to-speech (TTS) is your agent's voice, turning the LLM's text into speech that the user can hear
# See all providers at https://docs.livekit.io/agents/integrations/tts/
tts=cartesia.TTS(voice="6f84f4b8-58a2-430c-8c79-688dad597532"),
# VAD and turn detection are used to determine when the user is speaking and when the agent should respond
# See more at https://docs.livekit.io/agents/build/turns
turn_detection=MultilingualModel(),
vad=ctx.proc.userdata["vad"],
# 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=True,
)
# To use a realtime model instead of a voice pipeline, use the following session setup instead:
# session = AgentSession(
# # See all providers at https://docs.livekit.io/agents/integrations/realtime/
# llm=openai.realtime.RealtimeModel(voice="marin")
# )
# 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
@session.on("agent_false_interruption")
def _on_agent_false_interruption(ev: AgentFalseInterruptionEvent):
logger.info("false positive interruption, resuming")
session.generate_reply(instructions=ev.extra_instructions or NOT_GIVEN)
# Metrics collection, to measure pipeline performance
# For more information, see https://docs.livekit.io/agents/build/metrics/
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}")
ctx.add_shutdown_callback(log_usage)
# # Add a virtual avatar to the session, if desired
# # For other providers, see https://docs.livekit.io/agents/integrations/avatar/
# avatar = hedra.AvatarSession(
# avatar_id="...", # See https://docs.livekit.io/agents/integrations/avatar/hedra
# )
# # Start the avatar and wait for it to join
# await avatar.start(session, room=ctx.room)
# Start the session, which initializes the voice pipeline and warms up the models
await session.start(
agent=Assistant(),
room=ctx.room,
room_input_options=RoomInputOptions(
# LiveKit Cloud enhanced noise cancellation
# - If self-hosting, omit this parameter
# - For telephony applications, use `BVCTelephony` for best results
noise_cancellation=noise_cancellation.BVC(),
),
)
# Join the room and connect to the user
await ctx.connect()
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm))