livekit--agents
wehub-resource-sync
398 行
15 KiB
Python
398 行
15 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import datetime
|
|
import logging
|
|
import os
|
|
import sys
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from typing import Literal
|
|
from zoneinfo import ZoneInfo
|
|
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import simulation
|
|
from calendar_api import (
|
|
AvailableSlot,
|
|
CalComCalendar,
|
|
Calendar,
|
|
FakeCalendar,
|
|
SlotUnavailableError,
|
|
)
|
|
from dotenv import load_dotenv
|
|
from ui_view import UIView
|
|
|
|
from livekit.agents import (
|
|
Agent,
|
|
AgentServer,
|
|
AgentSession,
|
|
JobContext,
|
|
RunContext,
|
|
SimulationContext,
|
|
ToolError,
|
|
beta,
|
|
cli,
|
|
function_tool,
|
|
get_job_context,
|
|
inference,
|
|
mock_tools,
|
|
)
|
|
from livekit.agents.evals import (
|
|
JudgeGroup,
|
|
accuracy_judge,
|
|
coherence_judge,
|
|
conciseness_judge,
|
|
handoff_judge,
|
|
relevancy_judge,
|
|
safety_judge,
|
|
task_completion_judge,
|
|
tool_use_judge,
|
|
)
|
|
from livekit.agents.voice import UserStateChangedEvent
|
|
|
|
load_dotenv()
|
|
|
|
|
|
@dataclass
|
|
class Userdata:
|
|
cal: Calendar
|
|
slot_unavailable_count: int = 0
|
|
# Optional UI for the LiveKit Playground. ``None`` when the agent
|
|
# is running anywhere else — the tool handlers no-op on it and
|
|
# the rest of the code stays oblivious to the playground.
|
|
ui: UIView | None = None
|
|
|
|
|
|
logger = logging.getLogger("front-desk")
|
|
|
|
|
|
class FrontDeskAgent(Agent):
|
|
def __init__(
|
|
self, *, timezone: str, now: Callable[[], datetime.datetime] | None = None
|
|
) -> None:
|
|
self.tz = ZoneInfo(timezone)
|
|
# the calendar's clock, so the agent's sense of "today" matches the
|
|
# availability it sees. Defaults to wall-clock; the simulation entrypoint
|
|
# passes the calendar's pinned clock. Exposed to the model via the
|
|
# get_current_time tool rather than baked into the (cached) instructions.
|
|
self._now = now or (lambda: datetime.datetime.now(self.tz))
|
|
|
|
super().__init__(
|
|
instructions=(
|
|
# Outcome — what a great interaction looks like.
|
|
"You are Front-Desk, a helpful and efficient voice assistant. "
|
|
"A great interaction ends with the user booked into an appointment slot that works "
|
|
"for them, reached through a warm, flowing conversation with as little "
|
|
"back-and-forth as possible. "
|
|
# The current date/time is not baked in (it would break the prompt cache);
|
|
# pull it from get_current_time whenever you need to reason about dates.
|
|
"You do not inherently know the current date or time — call get_current_time "
|
|
"whenever you need to reason about dates, such as interpreting a request like "
|
|
"'next Tuesday' or checking whether a date the caller mentions has already passed. "
|
|
# Voice & personality — keep it short and human.
|
|
"Your output is synthesized directly to speech, so produce a natural verbatim "
|
|
"transcript, not polished text. Start responses with real reactions (oh, hmm, ah) "
|
|
'and fillers (um, uh, like) rather than "Absolutely" or "Certainly", with '
|
|
"mid-sentence fillers (like, you know, I mean) where they’d naturally fall. Mirror "
|
|
"the user's formality: if they're casual, use informal phrasing (gotcha, alright, "
|
|
"gonna, kinda, lemme, yeah); if they're more formal, keep your speech cleaner. Vary "
|
|
"your openers across turns — if you opened the last turn with 'gotcha', pick "
|
|
"'alright' or 'okay' this turn; don't repeat the same opener back-to-back. "
|
|
# How to work — be proactive, acknowledge before acting, stop when you can move forward.
|
|
"Be proactive: when the user greets you, use it to move things forward (e.g. "
|
|
"'Would you like to book a time?') rather than just greeting back. Before a tool "
|
|
"call that takes a moment, give a brief spoken acknowledgment so there’s no dead "
|
|
"air. After each result, check whether you can now move the user toward a booking: "
|
|
"if so, do it; if you're missing something, ask for just that. "
|
|
# Speaking about times — constraints that keep it natural over voice.
|
|
"When talking about availability, call list_available_slots and offer a few clear "
|
|
"options at a time, then pause for a response and guide the user to confirm. Say "
|
|
"times like 'Monday at 2' — avoid timezones, timestamps, and the words 'AM'/'PM'; "
|
|
"use natural phrases like 'in the morning' or 'in the evening', and don’t mention "
|
|
"the year unless it differs from the current one. When listing several times in the "
|
|
"same window, group them ('in the evening at 4, 5, or 6') instead of repeating the "
|
|
"time-of-day qualifier on each slot. If a chosen time is no longer available, let "
|
|
"them know gently and offer the next options."
|
|
)
|
|
)
|
|
|
|
self._slots_map: dict[str, AvailableSlot] = {}
|
|
|
|
async def on_enter(self) -> None:
|
|
hour = self._now().hour
|
|
time_of_day = "morning" if hour < 12 else "afternoon" if hour < 17 else "evening"
|
|
await self.session.generate_reply(
|
|
instructions=(
|
|
f"Say hello and welcome to the caller — it's currently {time_of_day} their time. "
|
|
"You're the front desk of an office and you're here to help them schedule a visit. "
|
|
"Invite them to book an appointment to visit, and ask what time works. "
|
|
"Keep it warm and brief."
|
|
)
|
|
)
|
|
|
|
@function_tool
|
|
async def get_current_time(self) -> str:
|
|
"""Get the current date and time.
|
|
|
|
Call this whenever you need to reason about dates — to interpret relative
|
|
requests like "next Tuesday", or to check whether a date the caller
|
|
mentions has already passed.
|
|
"""
|
|
# Kept out of the (cached) system instructions and served on demand, so the
|
|
# prompt-cache prefix stays stable and the time is always current.
|
|
return f"The current date and time is {self._now():%A, %B %d, %Y at %H:%M %Z}."
|
|
|
|
@function_tool
|
|
async def schedule_appointment(
|
|
self,
|
|
ctx: RunContext[Userdata],
|
|
slot_id: str,
|
|
) -> str | None:
|
|
"""
|
|
Schedule an appointment at the given slot.
|
|
|
|
Args:
|
|
slot_id: The identifier for the selected time slot (as shown in the list of available slots).
|
|
"""
|
|
if not (slot := self._slots_map.get(slot_id)):
|
|
raise ToolError(f"error: slot {slot_id} was not found")
|
|
|
|
email_result = await beta.workflows.GetEmailTask(chat_ctx=self.chat_ctx)
|
|
|
|
if ctx.speech_handle.interrupted:
|
|
return None
|
|
|
|
ctx.disallow_interruptions()
|
|
|
|
try:
|
|
await ctx.userdata.cal.schedule_appointment(
|
|
start_time=slot.start_time, attendee_email=email_result.email_address
|
|
)
|
|
except SlotUnavailableError:
|
|
ctx.userdata.slot_unavailable_count += 1
|
|
try:
|
|
get_job_context().tagger.add(
|
|
"slot:unavailable",
|
|
metadata={"count": ctx.userdata.slot_unavailable_count},
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
# exceptions other than ToolError are treated as "An internal error occurred" for the LLM.
|
|
# Tell the LLM this slot isn't available anymore
|
|
raise ToolError("This slot isn't available anymore") from None
|
|
|
|
# the booking is recorded by the calendar (the system of record); no
|
|
# parallel bookkeeping here that the simulation mock would have to mirror
|
|
local = slot.start_time.astimezone(self.tz)
|
|
try:
|
|
get_job_context().tagger.add(
|
|
"appointment:booked",
|
|
metadata={"time": local.isoformat()},
|
|
)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
if ctx.userdata.ui is not None:
|
|
ctx.userdata.ui.appointment_booked(slot, self.tz)
|
|
|
|
return f"The appointment was successfully scheduled for {local.strftime('%A, %B %d, %Y at %H:%M %Z')}."
|
|
|
|
@function_tool
|
|
async def list_available_slots(
|
|
self, ctx: RunContext[Userdata], range: Literal["+2week", "+1month", "+3month", "default"]
|
|
) -> str:
|
|
"""
|
|
Return a plain-text list of available slots, one per line.
|
|
|
|
<slot_id> – <Weekday>, <Month> <Day>, <Year> at <HH:MM> <TZ> (<relative time>)
|
|
|
|
You must infer the appropriate ``range`` implicitly from the
|
|
conversational context and **must not** prompt the user to pick a value
|
|
explicitly.
|
|
|
|
Args:
|
|
range: Determines how far ahead to search for free time slots.
|
|
"""
|
|
current_time = self._now()
|
|
lines: list[str] = []
|
|
|
|
if range == "+2week" or range == "default":
|
|
range_days = 14
|
|
elif range == "+1month":
|
|
range_days = 30
|
|
elif range == "+3month":
|
|
range_days = 90
|
|
|
|
slots = await ctx.userdata.cal.list_available_slots(
|
|
start_time=current_time, end_time=current_time + datetime.timedelta(days=range_days)
|
|
)
|
|
|
|
for slot in slots:
|
|
local = slot.start_time.astimezone(self.tz)
|
|
delta = local - current_time
|
|
days = delta.days
|
|
seconds = delta.seconds
|
|
|
|
if local.date() == current_time.date():
|
|
if seconds < 3600:
|
|
rel = "in less than an hour"
|
|
else:
|
|
rel = "later today"
|
|
elif local.date() == (current_time.date() + datetime.timedelta(days=1)):
|
|
rel = "tomorrow"
|
|
elif days < 7:
|
|
rel = f"in {days} days"
|
|
elif days < 14:
|
|
rel = "in 1 week"
|
|
else:
|
|
rel = f"in {days // 7} weeks"
|
|
|
|
lines.append(
|
|
f"{slot.unique_hash} – {local.strftime('%A, %B %d, %Y')} at "
|
|
f"{local:%H:%M} {local.tzname()} ({rel})"
|
|
)
|
|
self._slots_map[slot.unique_hash] = slot
|
|
|
|
if ctx.userdata.ui is not None:
|
|
ctx.userdata.ui.slots_listed(slots, current_time, self.tz, range_days)
|
|
|
|
return "\n".join(lines) or "No slots available at the moment."
|
|
|
|
|
|
server = AgentServer()
|
|
|
|
|
|
async def on_simulation_end(ctx: SimulationContext) -> None:
|
|
# grade the run on final calendar state; a mismatch vetoes the run
|
|
userdata = ctx.userdata()
|
|
if "expected_booking" not in userdata:
|
|
return # scenario graded on conversation only
|
|
|
|
cal: FakeCalendar = ctx.job_context.primary_session.userdata.cal
|
|
booked = cal.scheduled_appointments
|
|
|
|
def speak(dt: datetime.datetime) -> str:
|
|
return dt.astimezone(cal.tz).isoformat()
|
|
|
|
if (expected_raw := userdata["expected_booking"]) is None:
|
|
if booked:
|
|
times = ", ".join(speak(b.slot.start_time) for b in booked)
|
|
ctx.fail(reason=f"no booking was expected, but the agent booked: {times}")
|
|
return
|
|
|
|
expected = simulation.parse_slot(expected_raw, cal.tz)
|
|
if len(booked) != 1 or booked[0].slot.start_time != expected:
|
|
times = ", ".join(speak(b.slot.start_time) for b in booked) or "nothing"
|
|
ctx.fail(reason=f"expected a single booking at {speak(expected)}, got {times}")
|
|
|
|
|
|
async def on_session_end(ctx: JobContext) -> None:
|
|
# `on_session_end` runs even if the job crashed before the AgentSession
|
|
# started (e.g. a bad timezone, a calendar fault) — make_session_report
|
|
# raises in that case, and there's nothing to evaluate anyway.
|
|
try:
|
|
report = ctx.make_session_report()
|
|
except RuntimeError:
|
|
return
|
|
|
|
# Skip evaluation for very short conversations
|
|
chat = report.chat_history.copy(exclude_function_call=True, exclude_instructions=True)
|
|
if len(chat.items) < 3:
|
|
return
|
|
|
|
judges = JudgeGroup(
|
|
llm="openai/gpt-4o-mini",
|
|
judges=[
|
|
task_completion_judge(),
|
|
accuracy_judge(),
|
|
tool_use_judge(),
|
|
handoff_judge(),
|
|
safety_judge(),
|
|
relevancy_judge(),
|
|
coherence_judge(),
|
|
conciseness_judge(),
|
|
],
|
|
)
|
|
|
|
await judges.evaluate(report.chat_history)
|
|
|
|
userdata = ctx.primary_session.userdata
|
|
if userdata.cal.scheduled_appointments:
|
|
ctx.tagger.success()
|
|
else:
|
|
ctx.tagger.fail(reason="Appointment was not booked")
|
|
|
|
logger.info("session tags: %s", ctx.tagger.tags)
|
|
|
|
|
|
@server.rtc_session(on_session_end=on_session_end, on_simulation_end=on_simulation_end)
|
|
async def frontdesk_agent(ctx: JobContext):
|
|
await ctx.connect()
|
|
|
|
timezone = "UTC"
|
|
tool_mocks: dict[str, Callable] = {}
|
|
|
|
if sim := ctx.simulation_context():
|
|
# the scenario's userdata seeds the calendar (pinned to the scenario's
|
|
# clock so its absolute dates line up); the tools run mocked
|
|
cal = simulation.fake_calendar(sim, timezone=timezone)
|
|
tool_mocks = simulation.tool_mocks(cal, ZoneInfo(timezone))
|
|
elif cal_api_key := os.getenv("CAL_API_KEY", None):
|
|
logger.info("CAL_API_KEY detected, using cal.com calendar")
|
|
cal = CalComCalendar(api_key=cal_api_key, timezone=timezone)
|
|
else:
|
|
logger.warning(
|
|
"CAL_API_KEY is not set. Falling back to FakeCalendar; set CAL_API_KEY to enable Cal.com integration."
|
|
)
|
|
cal = FakeCalendar(timezone=timezone)
|
|
|
|
await cal.initialize()
|
|
|
|
userdata = Userdata(cal=cal, ui=UIView(ctx))
|
|
|
|
session = AgentSession[Userdata](
|
|
userdata=userdata,
|
|
stt=inference.STT("deepgram/nova-3"),
|
|
llm=inference.LLM("google/gemma-4-31b-it"),
|
|
tts=inference.TTS(
|
|
"inworld/inworld-tts-2",
|
|
voice="Nadia",
|
|
extra_kwargs={"delivery_mode": "CREATIVE", "speaking_rate": 1.1},
|
|
),
|
|
# leave max_tool_steps at the default (3) so a turn can chain
|
|
# get_current_time -> list_available_slots
|
|
# Flip user_state to "away" after 10s of mutual silence so we can
|
|
# check whether they're still there (default is 15s).
|
|
user_away_timeout=10.0,
|
|
)
|
|
|
|
idle_task: asyncio.Task[None] | None = None
|
|
|
|
async def _nudge_while_idle() -> None:
|
|
# Nudge every 10s until the user speaks again — speaking flips
|
|
# user_state out of "away", which cancels this task below.
|
|
while True:
|
|
logger.info("user idle — checking if they're still there")
|
|
await session.generate_reply(
|
|
instructions="The user has been idle, see if they're still there"
|
|
)
|
|
await asyncio.sleep(10)
|
|
|
|
@session.on("user_state_changed")
|
|
def _on_user_state_changed(ev: UserStateChangedEvent) -> None:
|
|
nonlocal idle_task
|
|
if ev.new_state == "away":
|
|
if idle_task is None or idle_task.done():
|
|
idle_task = asyncio.create_task(_nudge_while_idle())
|
|
elif idle_task is not None:
|
|
idle_task.cancel()
|
|
idle_task = None
|
|
|
|
mock_tools(FrontDeskAgent, tool_mocks, session=session)
|
|
await session.start(agent=FrontDeskAgent(timezone=timezone, now=cal.now), room=ctx.room)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli.run_app(server)
|