wehub-resource-sync
360 lines
12 KiB
Python
360 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from openai.types.realtime.realtime_session_create_request import (
|
|
RealtimeSessionCreateRequest,
|
|
)
|
|
from openai.types.realtime.session_update_event import SessionUpdateEvent
|
|
|
|
from agents.handoffs import Handoff
|
|
from agents.realtime.agent import RealtimeAgent
|
|
from agents.realtime.config import RealtimeRunConfig, RealtimeSessionModelSettings
|
|
from agents.realtime.handoffs import realtime_handoff
|
|
from agents.realtime.model import RealtimeModelConfig
|
|
from agents.realtime.openai_realtime import (
|
|
OpenAIRealtimeSIPModel,
|
|
OpenAIRealtimeWebSocketModel,
|
|
_build_model_settings_from_agent,
|
|
_collect_enabled_handoffs,
|
|
)
|
|
from agents.run_context import RunContextWrapper
|
|
from agents.tool import FunctionTool, function_tool
|
|
|
|
|
|
def _disabled_billing_realtime_handoff(*, is_enabled: Any = False) -> Handoff[Any, Any]:
|
|
return realtime_handoff(
|
|
RealtimeAgent(name="billing"),
|
|
tool_name_override="transfer_to_billing",
|
|
is_enabled=is_enabled,
|
|
)
|
|
|
|
|
|
def _disabled_billing_realtime_tool(*, is_enabled: Any = False) -> FunctionTool:
|
|
return function_tool(
|
|
lambda: "ok",
|
|
name_override="transfer_to_billing",
|
|
is_enabled=is_enabled,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_collect_enabled_handoffs_filters_disabled() -> None:
|
|
parent = RealtimeAgent(name="parent")
|
|
disabled = realtime_handoff(
|
|
RealtimeAgent(name="child_disabled"),
|
|
is_enabled=lambda ctx, agent: False,
|
|
)
|
|
parent.handoffs = [disabled, RealtimeAgent(name="child_enabled")]
|
|
|
|
enabled = await _collect_enabled_handoffs(parent, RunContextWrapper(None))
|
|
|
|
assert len(enabled) == 1
|
|
assert isinstance(enabled[0], Handoff)
|
|
assert enabled[0].agent_name == "child_enabled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_model_settings_from_agent_merges_agent_fields(monkeypatch: pytest.MonkeyPatch):
|
|
agent = RealtimeAgent(name="root", prompt={"id": "prompt-id"})
|
|
monkeypatch.setattr(agent, "get_system_prompt", AsyncMock(return_value="sys"))
|
|
|
|
@function_tool
|
|
def helper() -> str:
|
|
"""Helper tool for testing."""
|
|
return "ok"
|
|
|
|
monkeypatch.setattr(agent, "get_all_tools", AsyncMock(return_value=[helper]))
|
|
agent.handoffs = [RealtimeAgent(name="handoff-child")]
|
|
base_settings: RealtimeSessionModelSettings = {"model_name": "gpt-realtime-2.1"}
|
|
starting_settings: RealtimeSessionModelSettings = {"voice": "verse"}
|
|
run_config: RealtimeRunConfig = {"tracing_disabled": True}
|
|
|
|
merged = await _build_model_settings_from_agent(
|
|
agent=agent,
|
|
context_wrapper=RunContextWrapper(None),
|
|
base_settings=base_settings,
|
|
starting_settings=starting_settings,
|
|
run_config=run_config,
|
|
)
|
|
|
|
assert merged["prompt"] == {"id": "prompt-id"}
|
|
assert merged["instructions"] == "sys"
|
|
assert merged["tools"][0].name == helper.name
|
|
assert merged["handoffs"][0].agent_name == "handoff-child"
|
|
assert merged["voice"] == "verse"
|
|
assert merged["model_name"] == "gpt-realtime-2.1"
|
|
assert merged["tracing"] is None
|
|
assert base_settings == {"model_name": "gpt-realtime-2.1"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_model_settings_filters_disabled_starting_handoff_name_conflict():
|
|
tool = function_tool(lambda: "ok", name_override="transfer_to_billing")
|
|
disabled_handoff = _disabled_billing_realtime_handoff()
|
|
agent = RealtimeAgent(name="parent", tools=[tool])
|
|
|
|
merged = await _build_model_settings_from_agent(
|
|
agent=agent,
|
|
context_wrapper=RunContextWrapper(None),
|
|
base_settings={},
|
|
starting_settings={"handoffs": [disabled_handoff]},
|
|
run_config=None,
|
|
)
|
|
|
|
assert merged["tools"] == [tool]
|
|
assert merged["handoffs"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_model_settings_filters_disabled_starting_tool_name_conflict():
|
|
disabled_tool = _disabled_billing_realtime_tool()
|
|
handoff = _disabled_billing_realtime_handoff(is_enabled=True)
|
|
agent = RealtimeAgent(name="parent", handoffs=[handoff])
|
|
|
|
merged = await _build_model_settings_from_agent(
|
|
agent=agent,
|
|
context_wrapper=RunContextWrapper(None),
|
|
base_settings={},
|
|
starting_settings={"tools": [disabled_tool]},
|
|
run_config=None,
|
|
)
|
|
|
|
assert merged["tools"] == []
|
|
assert merged["handoffs"] == [handoff]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_model_settings_evaluates_starting_tool_is_enabled_callable():
|
|
calls: list[tuple[RunContextWrapper[Any], RealtimeAgent[Any]]] = []
|
|
|
|
async def is_enabled(ctx: RunContextWrapper[Any], agent_arg: RealtimeAgent[Any]) -> bool:
|
|
calls.append((ctx, agent_arg))
|
|
return False
|
|
|
|
disabled_tool = _disabled_billing_realtime_tool(is_enabled=is_enabled)
|
|
agent = RealtimeAgent(name="parent")
|
|
context_wrapper = RunContextWrapper(None)
|
|
|
|
merged = await _build_model_settings_from_agent(
|
|
agent=agent,
|
|
context_wrapper=context_wrapper,
|
|
base_settings={},
|
|
starting_settings={"tools": [disabled_tool]},
|
|
run_config=None,
|
|
)
|
|
|
|
assert merged["tools"] == []
|
|
assert calls == [(context_wrapper, agent)]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_model_settings_does_not_reevaluate_agent_handoff_without_override():
|
|
call_count = 0
|
|
|
|
async def is_enabled(ctx: RunContextWrapper[Any], agent_arg: RealtimeAgent[Any]) -> bool:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return call_count == 1
|
|
|
|
handoff = cast(
|
|
Handoff[Any, Any],
|
|
realtime_handoff(RealtimeAgent(name="billing"), is_enabled=is_enabled),
|
|
)
|
|
agent = RealtimeAgent(name="parent", handoffs=[handoff])
|
|
|
|
merged = await _build_model_settings_from_agent(
|
|
agent=agent,
|
|
context_wrapper=RunContextWrapper(None),
|
|
base_settings={},
|
|
starting_settings={"voice": "verse"},
|
|
run_config=None,
|
|
)
|
|
|
|
assert merged["handoffs"] == [handoff]
|
|
assert call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sip_model_build_initial_session_payload(monkeypatch: pytest.MonkeyPatch):
|
|
agent = RealtimeAgent(name="parent", prompt={"id": "prompt-99"})
|
|
child_agent = RealtimeAgent(name="child")
|
|
agent.handoffs = [child_agent]
|
|
|
|
@function_tool
|
|
def ping() -> str:
|
|
"""Ping tool used for session payload building."""
|
|
return "pong"
|
|
|
|
monkeypatch.setattr(agent, "get_system_prompt", AsyncMock(return_value="parent-system"))
|
|
monkeypatch.setattr(agent, "get_all_tools", AsyncMock(return_value=[ping]))
|
|
|
|
model_config: RealtimeModelConfig = {
|
|
"initial_model_settings": {
|
|
"model_name": "gpt-realtime-mini",
|
|
"voice": "verse",
|
|
}
|
|
}
|
|
run_config: RealtimeRunConfig = {
|
|
"model_settings": {"output_modalities": ["text"]},
|
|
"tracing_disabled": True,
|
|
}
|
|
overrides: RealtimeSessionModelSettings = {
|
|
"audio": {"input": {"format": {"type": "audio/pcmu"}}},
|
|
"output_audio_format": "g711_ulaw",
|
|
}
|
|
|
|
payload = await OpenAIRealtimeSIPModel.build_initial_session_payload(
|
|
agent,
|
|
context={"user": "abc"},
|
|
model_config=model_config,
|
|
run_config=run_config,
|
|
overrides=overrides,
|
|
)
|
|
|
|
assert isinstance(payload, RealtimeSessionCreateRequest)
|
|
assert payload.model == "gpt-realtime-mini"
|
|
assert payload.output_modalities == ["text"]
|
|
assert payload.audio is not None
|
|
audio = payload.audio
|
|
assert audio.input is not None
|
|
assert audio.input.format is not None
|
|
assert audio.input.format.type == "audio/pcmu"
|
|
assert audio.output is not None
|
|
assert audio.output.format is not None
|
|
assert audio.output.format.type == "audio/pcmu"
|
|
assert audio.output.voice == "verse"
|
|
assert payload.instructions == "parent-system"
|
|
assert payload.prompt is not None and payload.prompt.id == "prompt-99"
|
|
tool_names: set[str] = set()
|
|
for tool in payload.tools or []:
|
|
name = getattr(tool, "name", None)
|
|
if name:
|
|
tool_names.add(name)
|
|
assert ping.name in tool_names
|
|
assert f"transfer_to_{child_agent.name}" in tool_names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sip_initial_session_payload_filters_disabled_initial_model_settings_handoff():
|
|
tool = function_tool(lambda: "ok", name_override="transfer_to_billing")
|
|
disabled_handoff = _disabled_billing_realtime_handoff()
|
|
agent = RealtimeAgent(name="parent", tools=[tool])
|
|
|
|
payload = await OpenAIRealtimeSIPModel.build_initial_session_payload(
|
|
agent,
|
|
model_config={"initial_model_settings": {"handoffs": [disabled_handoff]}},
|
|
)
|
|
|
|
tool_names = [getattr(tool, "name", None) for tool in payload.tools or []]
|
|
assert tool_names.count("transfer_to_billing") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sip_initial_session_payload_filters_disabled_initial_model_settings_tool():
|
|
disabled_tool = _disabled_billing_realtime_tool()
|
|
agent = RealtimeAgent(
|
|
name="parent",
|
|
handoffs=[_disabled_billing_realtime_handoff(is_enabled=True)],
|
|
)
|
|
|
|
payload = await OpenAIRealtimeSIPModel.build_initial_session_payload(
|
|
agent,
|
|
model_config={"initial_model_settings": {"tools": [disabled_tool]}},
|
|
)
|
|
|
|
tool_names = [getattr(tool, "name", None) for tool in payload.tools or []]
|
|
assert tool_names == ["transfer_to_billing"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sip_initial_session_payload_filters_disabled_override_handoff():
|
|
tool = function_tool(lambda: "ok", name_override="transfer_to_billing")
|
|
disabled_handoff = _disabled_billing_realtime_handoff()
|
|
agent = RealtimeAgent(name="parent", tools=[tool])
|
|
|
|
payload = await OpenAIRealtimeSIPModel.build_initial_session_payload(
|
|
agent,
|
|
overrides={"handoffs": [disabled_handoff]},
|
|
)
|
|
|
|
tool_names = [getattr(tool, "name", None) for tool in payload.tools or []]
|
|
assert tool_names.count("transfer_to_billing") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sip_initial_session_payload_filters_disabled_override_tool():
|
|
disabled_tool = _disabled_billing_realtime_tool()
|
|
agent = RealtimeAgent(
|
|
name="parent",
|
|
handoffs=[_disabled_billing_realtime_handoff(is_enabled=True)],
|
|
)
|
|
|
|
payload = await OpenAIRealtimeSIPModel.build_initial_session_payload(
|
|
agent,
|
|
overrides={"tools": [disabled_tool]},
|
|
)
|
|
|
|
tool_names = [getattr(tool, "name", None) for tool in payload.tools or []]
|
|
assert tool_names == ["transfer_to_billing"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sip_initial_session_payload_does_not_reevaluate_agent_handoff_without_override():
|
|
call_count = 0
|
|
|
|
async def is_enabled(ctx: RunContextWrapper[Any], agent_arg: RealtimeAgent[Any]) -> bool:
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return call_count == 1
|
|
|
|
handoff = cast(
|
|
Handoff[Any, Any],
|
|
realtime_handoff(RealtimeAgent(name="billing"), is_enabled=is_enabled),
|
|
)
|
|
agent = RealtimeAgent(name="parent", handoffs=[handoff])
|
|
|
|
payload = await OpenAIRealtimeSIPModel.build_initial_session_payload(
|
|
agent,
|
|
overrides={"voice": "verse"},
|
|
)
|
|
|
|
tool_names = [getattr(tool, "name", None) for tool in payload.tools or []]
|
|
assert "transfer_to_billing" in tool_names
|
|
assert call_count == 1
|
|
|
|
|
|
def test_call_id_session_update_omits_null_audio_formats() -> None:
|
|
model = OpenAIRealtimeWebSocketModel()
|
|
model._call_id = "call_123"
|
|
|
|
session_config = model._get_session_config({})
|
|
payload = SessionUpdateEvent(type="session.update", session=session_config).model_dump(
|
|
exclude_unset=True
|
|
)
|
|
|
|
audio = payload["session"]["audio"]
|
|
assert "format" not in audio["input"]
|
|
assert "format" not in audio["output"]
|
|
|
|
|
|
def test_call_id_session_update_includes_explicit_audio_formats() -> None:
|
|
model = OpenAIRealtimeWebSocketModel()
|
|
model._call_id = "call_123"
|
|
|
|
session_config = model._get_session_config(
|
|
{
|
|
"input_audio_format": "g711_ulaw",
|
|
"output_audio_format": "g711_ulaw",
|
|
}
|
|
)
|
|
payload = SessionUpdateEvent(type="session.update", session=session_config).model_dump(
|
|
exclude_unset=True
|
|
)
|
|
|
|
audio = payload["session"]["audio"]
|
|
assert audio["input"]["format"]["type"] == "audio/pcmu"
|
|
assert audio["output"]["format"]["type"] == "audio/pcmu"
|