170 lines
6.6 KiB
Python
170 lines
6.6 KiB
Python
import asyncio
|
|
import textwrap
|
|
import unittest.mock
|
|
|
|
import pytest
|
|
from livekit.agents import AgentSession, inference
|
|
|
|
from agent import InterviewData, PastExperienceAgent, SelfIntroAgent
|
|
|
|
|
|
def _llm() -> inference.LLM:
|
|
return inference.LLM(model="openai/gpt-4.1-mini")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_self_intro_greeting() -> None:
|
|
"""SelfIntroAgent greets the candidate and invites a self-introduction."""
|
|
async with (
|
|
_llm() as llm,
|
|
AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session,
|
|
):
|
|
await session.start(SelfIntroAgent())
|
|
|
|
result = await session.run(user_input="Hello")
|
|
|
|
await (
|
|
result.expect.next_event()
|
|
.is_message(role="assistant")
|
|
.judge(
|
|
llm,
|
|
intent=textwrap.dedent(
|
|
"""\
|
|
Greets the candidate warmly and invites them to introduce themselves,
|
|
share their background, or describe their experience and motivations.
|
|
"""
|
|
),
|
|
)
|
|
)
|
|
result.expect.no_more_events()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_intro_complete_handoff() -> None:
|
|
"""Highest-priority: after a complete self-introduction, intro_complete fires once
|
|
and control transfers to PastExperienceAgent."""
|
|
async with (
|
|
_llm() as llm,
|
|
AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session,
|
|
):
|
|
await session.start(SelfIntroAgent())
|
|
|
|
result = await session.run(
|
|
user_input=(
|
|
"Hi, I'm Sarah. I've been a software engineer for six years, "
|
|
"mostly at fintech startups. My most recent role was at a payments company "
|
|
"where I led backend API development. I'm interested in this position "
|
|
"because I want to work on larger-scale distributed systems. "
|
|
"That's my background!"
|
|
)
|
|
)
|
|
|
|
# intro_complete must be called somewhere in this turn
|
|
fnc = result.expect.next_event(type="function_call")
|
|
assert fnc.event().item.name == "intro_complete"
|
|
# Handoff to PastExperienceAgent must occur (position-independent search)
|
|
result.expect.contains_agent_handoff(new_agent_type=PastExperienceAgent)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_premature_handoff() -> None:
|
|
"""A bare greeting must not trigger intro_complete or hand off to PastExperienceAgent."""
|
|
async with (
|
|
_llm() as llm,
|
|
AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session,
|
|
):
|
|
await session.start(SelfIntroAgent())
|
|
|
|
result = await session.run(user_input="Hi")
|
|
|
|
# contains_agent_handoff should raise AssertionError when no handoff is present
|
|
with pytest.raises(AssertionError):
|
|
result.expect.contains_agent_handoff(new_agent_type=PastExperienceAgent)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_watchdog_cancelled_on_normal_handoff() -> None:
|
|
"""After intro_complete fires normally, on_exit cancels both watchdogs —
|
|
no second transition replaces the resulting PastExperienceAgent."""
|
|
async with (
|
|
_llm() as llm,
|
|
AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session,
|
|
):
|
|
intro_agent = SelfIntroAgent(silence_budget=3.0, stage_budget=30.0)
|
|
await session.start(intro_agent)
|
|
|
|
result = await session.run(
|
|
user_input=(
|
|
"Hi, I'm Sarah. I've been a software engineer for six years, "
|
|
"mostly at fintech startups. My most recent role was at a payments company "
|
|
"where I led backend API development. That's my background!"
|
|
)
|
|
)
|
|
result.expect.contains_agent_handoff(new_agent_type=PastExperienceAgent)
|
|
|
|
# Give the event loop a tick to process on_exit after the handoff.
|
|
await asyncio.sleep(0.1)
|
|
|
|
# on_exit must have cancelled and cleared both watchdog tasks.
|
|
assert intro_agent._silence_task is None, (
|
|
"on_exit did not clear _silence_task after intro_complete fired"
|
|
)
|
|
assert intro_agent._stage_task is None, (
|
|
"on_exit did not clear _stage_task after intro_complete fired"
|
|
)
|
|
assert isinstance(session.current_agent, PastExperienceAgent)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_handoff() -> None:
|
|
"""Silence fallback: when the silence budget elapses without intro_complete
|
|
firing, control transfers to PastExperienceAgent. Stage budget is set high so
|
|
it does not interfere with this silence-path test."""
|
|
async with (
|
|
_llm() as llm,
|
|
AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session,
|
|
):
|
|
# In text-test mode user_state is always "listening", so the silence timer
|
|
# acts as a flat wall-clock countdown. 0.5 s fires quickly; 2 s headroom.
|
|
await session.start(SelfIntroAgent(silence_budget=0.5, stage_budget=30.0))
|
|
await asyncio.sleep(2.0)
|
|
|
|
assert isinstance(session.current_agent, PastExperienceAgent), (
|
|
f"Expected PastExperienceAgent after silence budget elapsed; "
|
|
f"got {type(session.current_agent).__name__}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_silence_reset_on_speech() -> None:
|
|
"""Code-path check: the silence timer resets while user_state is 'speaking', so it
|
|
does not fire during candidate speech.
|
|
NOTE: this test mocks user_state and does NOT prove real-VAD behavior — the live
|
|
25 s continuous-talk run is the acceptance criterion for VAD-driven resets."""
|
|
async with (
|
|
_llm() as llm,
|
|
AgentSession[InterviewData](llm=llm, userdata=InterviewData()) as session,
|
|
):
|
|
await session.start(SelfIntroAgent(silence_budget=0.5, stage_budget=30.0))
|
|
|
|
# Simulate candidate speaking for 0.7 s — past the 0.5 s budget.
|
|
# Without the reset, the silence timer would have fired at 0.5 s.
|
|
with unittest.mock.patch.object(
|
|
type(session),
|
|
"user_state",
|
|
new_callable=unittest.mock.PropertyMock,
|
|
return_value="speaking",
|
|
):
|
|
await asyncio.sleep(0.7)
|
|
|
|
# Immediately after "speech" ends: timer has not fired yet (counter restarted).
|
|
assert isinstance(session.current_agent, SelfIntroAgent), (
|
|
"Silence timer fired during speech — reset logic not working"
|
|
)
|
|
|
|
# Wait for full silence budget + poll headroom + say() → should fire.
|
|
await asyncio.sleep(1.5)
|
|
assert isinstance(session.current_agent, PastExperienceAgent), (
|
|
"Silence timer did not fire after full budget of silence post-speech"
|
|
)
|