diff --git a/.env.example b/.env.example index c082eb455..1f5c6721b 100644 --- a/.env.example +++ b/.env.example @@ -139,14 +139,13 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}' #d json_object #x pipeline_mode=S2S #v EVA_MODEL__S2S_PARAMS='{"model": "gpt-realtime-mini", "api_key": ""}' - -# Smallest Hydra S2S (set EVA_FRAMEWORK=smallest_hydra). Hydra emits no transcript -# on the wire, so a batch STT transcribes each turn for the audit log; it defaults -# to Smallest Pulse keyed on the same api_key. Override via the "transcription" block -# (provider: smallest|openai|deepgram). voice: wren|sloane|marlowe|reed|knox|tate. -#x pipeline_mode=S2S -#v EVA_MODEL__S2S=hydra -#v EVA_MODEL__S2S_PARAMS='{"model": "hydra", "api_key": "", "voice": "wren", "generate_initial_response": true, "transcription": {"provider": "smallest", "model": "pulse-pro", "language": "en"}}' +# Alternatives (set EVA_FRAMEWORK to match): +# Hydra (smallest_hydra): batch STT transcribes turns; override via "transcription" (provider: smallest|openai|deepgram). +# EVA_MODEL__S2S=hydra +# EVA_MODEL__S2S_PARAMS='{"model": "hydra", "api_key": "", "voice": "wren", "generate_initial_response": true, "transcription": {"provider": "smallest", "model": "pulse-pro", "language": "en"}}' +# Cartesia Managed (cartesia): provision agents with scripts/provision_cartesia_agents.py; see docs/experiment_setup.md. +# EVA_MODEL__S2S=cartesia +# EVA_MODEL__S2S_PARAMS='{"model": "gpt-5.5", "api_key": "", "assistant_agent_id": ""}' # --- AudioLLM mode --- #i Audio-input LLM model name. @@ -162,7 +161,7 @@ EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}' # --- Framework (S2S / AudioLLM) --- #i Base framework for S2S or AudioLLM pipelines. #d enum -#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice,smallest_hydra +#e pipecat,openai_realtime,gemini_live,elevenlabs,grok_voice,smallest_hydra,cartesia #v EVA_FRAMEWORK=openai_realtime # ============================================== diff --git a/docs/experiment_setup.md b/docs/experiment_setup.md index 805f4d7e1..f11256bec 100644 --- a/docs/experiment_setup.md +++ b/docs/experiment_setup.md @@ -92,6 +92,32 @@ We use the default turn detection configurations for most framework in our exper EVA-Bench makes turn detection parameters and options configurable via the CLI, so practitioners can run experiments using the turn detection settings available to their chosen framework. The only exception is ElevenAgents, where users must register and configure their agents separately prior to evaluation. +### Cartesia Managed Agents + +`framework: cartesia` runs a Cartesia managed agent over a WebSocket connection. Managed Agents runs the whole agent loop: audio orchestration (listening, turn-taking, interruptions and speech) along with tool calling, reasoning orchestration and knowledge retrieval, around the reasoning model you choose. EVA's tools run as client tools, and EVA's system prompt and greeting are sent with each call as dynamic variables. One agent per domain is registered before evaluation, with its config in `scripts/provision_cartesia_agents.py`. Model fallback is disabled on these agents, so only the chosen model answers. + +Some agent settings can't be changed per call, so they are set when the agent is provisioned from EVA's options. The assistant server checks them against the run config and refuses to start if they don't match: + +| EVA option | Agent setting | +|---|---| +| `EVA_LANGUAGE` | `language.primary` (regional variants such as `fr-CA` use the base code) | +| `EVA_MODEL__PRE_TOOL_SPEECH` (`auto` or `off`) | each tool's `pre_tool_speech` | +| `EVA_TURN_END_FALLBACK_TIME` (5 to 60 seconds, or unset) | `turn.inactivity_check_in_secs` (unset turns check-ins off) | + +```bash +# Create or update the agents; --model picks the LLM (default gpt-5.5). Prints each domain's s2s_params. +python scripts/provision_cartesia_agents.py --pre-tool-speech auto + +EVA_FRAMEWORK=cartesia \ +EVA_MODEL__S2S=cartesia \ +EVA_MODEL__S2S_PARAMS='{"model": "gpt-5.5", "api_key": "", "assistant_agent_id": ""}' \ +python main.py --domain airline +``` + +`--voice` sets the agent's voice; it isn't checked against the run config. + +`s2s_params`: `api_key`, `model` and `assistant_agent_id` (the agent for the run's domain) are required, as for ElevenLabs. `model` must match the model configured on the agent; `alias` optionally labels the run. `base_url` (default `wss://api.cartesia.ai`) and `cartesia_version` are optional. `EVA_MODEL__LLM` must be unset, as for any S2S run. + ## Judge Models The table below lists the API-hosted models used as a judge. diff --git a/scripts/provision_cartesia_agents.py b/scripts/provision_cartesia_agents.py new file mode 100644 index 000000000..27c22db8d --- /dev/null +++ b/scripts/provision_cartesia_agents.py @@ -0,0 +1,153 @@ +"""Create or update one Cartesia managed agent per EVA domain, with EVA's tools as client tools. + +The agent passes EVA's system prompt and greeting through untouched: they are sent with each call as +the dynamic variables system_prompt and initial_message. Settings that cannot be changed per call are +set here from EVA's own options, and the assistant server refuses to run against an agent whose +settings do not match the run config: + + --language -> language.primary (EVA_LANGUAGE, default en; fr-CA uses fr) + --pre-tool-speech -> each tool's pre_tool_speech (EVA_MODEL__PRE_TOOL_SPEECH, default auto) + --turn-end-fallback-time -> turn.inactivity_check_in_secs (EVA_TURN_END_FALLBACK_TIME, default unset) + +--voice sets the agent's voice. It is not checked: like any agent setting that doesn't affect the +benchmark, it is identified by the run's assistant_agent_id. + +Model fallback is disabled, so only the chosen model answers. + +Usage: + python scripts/provision_cartesia_agents.py [--model gpt-5.5] [--language en] [--voice VOICE_ID] + [--pre-tool-speech auto|off] [--turn-end-fallback-time SECONDS] [--domains airline itsm medical_hr] + +Reads CARTESIA_API_KEY from the environment or .env. Tool and agent IDs are kept in +output/cartesia_agents.json so a rerun updates the same agents. Prints each domain's +EVA_MODEL__S2S_PARAMS (without the api_key) to use for the run. +""" + +import argparse +import json +import os +import sys +from functools import partial +from pathlib import Path + +from dotenv import dotenv_values + +from eva.models.agents import AgentConfig +from eva.models.config import LANGUAGE_DISPLAY_NAMES, ModelConfig, base_language +from eva.utils import cartesia_agents + +DOMAINS = ("airline", "itsm", "medical_hr") +DEFAULT_VOICE_ID = "f786b574-daa5-4673-aa0c-cbe3e8534c02" # Katie +STATE = Path("output/cartesia_agents.json") + + +def tool_body(fn: dict, pre_tool_speech: str) -> dict: + return { + "type": "client", + "name": fn["name"], + "description": fn["description"], + "parameters": fn["parameters"], + "expects_response": True, + "response_timeout_secs": 20, + "pre_tool_speech": pre_tool_speech, + "execution_mode": "immediate", + } + + +def agent_config( + model: str, tool_ids: list[str], turn_end_fallback_time: int | None, language: str, voice_id: str +) -> dict: + return { + "instructions": cartesia_agents.SYSTEM_PROMPT_VARIABLE, + "prompt_optimization": False, # use EVA's system prompt as is + "initial_message": cartesia_agents.INITIAL_MESSAGE_VARIABLE, + "speculative_generation": True, + "model": {"id": model, "fallback_strategy": None}, + "language": {"primary": base_language(language)}, + "turn": {"inactivity_check_in_secs": turn_end_fallback_time}, + "audio": { + "input": {"noise_suppression": "off", "asr_model": "auto"}, + "output": {"voice_id": voice_id}, + }, + "tools": [{"id": t} for t in tool_ids], + } + + +def check_model(request, model: str) -> None: + path = "/v1/agents/models" + while True: + page = request("GET", path) + if any(m.get("id") == model for m in page["data"]): + return + if not page.get("has_more"): + raise SystemExit(f"{model} is not in GET /v1/agents/models") + path = f"/v1/agents/models?starting_after={page['next_page']}" + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--model", default="gpt-5.5", help="Any model id from GET /v1/agents/models") + parser.add_argument( + "--language", + choices=[lang.value for lang in LANGUAGE_DISPLAY_NAMES], + default="en", + help="Must match EVA_LANGUAGE for the run", + ) + parser.add_argument("--voice", default=DEFAULT_VOICE_ID, help="Cartesia voice ID for the agent (default: Katie)") + parser.add_argument( + "--pre-tool-speech", + choices=["auto", "off"], + default=ModelConfig.model_fields["pre_tool_speech"].default, + help="Must match EVA_MODEL__PRE_TOOL_SPEECH for the run", + ) + parser.add_argument( + "--turn-end-fallback-time", + type=int, + default=None, + help="Must match EVA_TURN_END_FALLBACK_TIME for the run (5-60 seconds; unset disables check-ins)", + ) + parser.add_argument("--domains", nargs="+", default=list(DOMAINS), choices=DOMAINS) + args = parser.parse_args() + if args.turn_end_fallback_time is not None and not 5 <= args.turn_end_fallback_time <= 60: + raise SystemExit("Cartesia supports --turn-end-fallback-time between 5 and 60 seconds") + + key = os.environ.get("CARTESIA_API_KEY") or dotenv_values(".env").get("CARTESIA_API_KEY") + if not key: + raise SystemExit("Set CARTESIA_API_KEY in the environment or .env") + request = partial(cartesia_agents.request, api_key=key) + check_model(request, args.model) + + state = json.loads(STATE.read_text()) if STATE.exists() else {"tools": {}, "agents": {}} + STATE.parent.mkdir(parents=True, exist_ok=True) + for domain in args.domains: + # Tools keep EVA's order, the order the model sees them in. + tool_ids = [] + for tool in AgentConfig.from_yaml(f"configs/agents/{domain}_agent.yaml").build_tools_for_agent(): + fn = tool["function"] + key = f"{domain}/{fn['name']}" + body = tool_body(fn, args.pre_tool_speech) + if key in state["tools"]: + patch = {k: v for k, v in body.items() if k != "type"} # a tool's type cannot change + request("PATCH", f"/v1/agents/tools/{state['tools'][key]}", body=patch) + else: + state["tools"][key] = request("POST", "/v1/agents/tools", body=body)["id"] + STATE.write_text(json.dumps(state, indent=2)) + tool_ids.append(state["tools"][key]) + + name = f"eva-{domain.replace('_', '-')}-{args.model.replace('.', '').replace('-', '')}" + payload = { + "name": name, + "description": "EVA-Bench agent", + "config": agent_config(args.model, tool_ids, args.turn_end_fallback_time, args.language, args.voice), + } + if domain in state["agents"]: + request("PATCH", f"/v1/agents/{state['agents'][domain]}", body=payload) + else: + state["agents"][domain] = request("POST", "/v1/agents", body=payload)["id"] + STATE.write_text(json.dumps(state, indent=2)) + print(f"{domain}: {state['agents'][domain]} with {len(tool_ids)} tools", file=sys.stderr) + print(f"{domain}: " + json.dumps({"model": args.model, "assistant_agent_id": state["agents"][domain]})) + + +if __name__ == "__main__": + main() diff --git a/src/eva/assistant/cartesia_server.py b/src/eva/assistant/cartesia_server.py new file mode 100644 index 000000000..42e792103 --- /dev/null +++ b/src/eva/assistant/cartesia_server.py @@ -0,0 +1,522 @@ +"""Cartesia Managed Agents assistant server. + +Bridges the user simulator's Twilio-style WebSocket to a Cartesia managed agent over its WebSocket +API (/v1/agents/websocket/{agent_id}). The agent handles audio and turn-taking; this server moves +audio, runs EVA's tools as client tools, and records the conversation. EVA's system prompt and +greeting are sent with each call as dynamic variables, to agents created by +scripts/provision_cartesia_agents.py. Settings that cannot be set per call are checked against the +run config before the call starts. +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import time +from pathlib import Path + +try: + import audioop +except ImportError: + import audioop_lts as audioop + +import numpy as np +import soxr +import uvicorn +import websockets +from fastapi import FastAPI, WebSocket, WebSocketDisconnect + +from eva.assistant.base_server import AbstractAssistantServer +from eva.assistant.pipeline.observers import FrameworkLogWriter, MetricsLogWriter +from eva.models.agents import AgentConfig +from eva.models.config import ModelConfig +from eva.utils import cartesia_agents +from eva.utils.audio_utils import create_twilio_media_message, parse_twilio_media_message +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +_SAMPLE_RATE = 16000 # agent PCM in both directions + +MULAW_CHUNK_SIZE = 160 # 20 ms at 8 kHz +CHUNK_DURATION_S = 0.02 +PCM_CHUNK_SIZE = 640 # 20 ms at 16 kHz + +# When the simulator sends nothing for this long, feed real-time silence so the agent can end the +# caller's turn. Never ahead of real time: the simulator sends its own silence in fast bursts. +INPUT_SILENCE_GAP_S = 0.1 +_SILENCE_FRAME = b"\x00" * PCM_CHUNK_SIZE + + +def _now_ms() -> int: + return int(round(time.time() * 1000)) + + +def _pad_buffer_to_walltime(buffer: bytearray, elapsed_s: float, sample_rate: int) -> None: + """Pad a PCM16 buffer with silence so its length matches elapsed wall time.""" + target_bytes = int(elapsed_s * sample_rate) * 2 + if len(buffer) < target_bytes: + buffer.extend(b"\x00" * (target_bytes - len(buffer))) + + +def _mix_pcm16(track_a: bytes | bytearray, track_b: bytes | bytearray) -> bytes: + """eva.utils.audio_utils.pcm16_mix, vectorized so teardown does not block the event loop.""" + a = np.frombuffer(track_a, dtype=" None: + # Consecutive agent messages arrive unspaced. + if self.text[-1:] in (".", "!", "?") and text[:1].isalnum(): + self.end_piece() + text = " " + text + if self._piece_ms is None: + self._piece_ms = _now_ms() + self.text += text + self._piece += text + + def end_piece(self) -> None: + if self._piece.strip() and self._piece_ms is not None: + self._pieces.append((self._piece.strip(), self._piece_ms)) + self._piece, self._piece_ms = "", None + + def take_pieces(self) -> list[tuple[str, int]]: + """Closed pieces not yet logged, as (text, start_ms).""" + pieces, self._pieces = self._pieces, [] + return pieces + + +class CartesiaAssistantServer(AbstractAssistantServer): + def __init__( + self, + current_date_time: str, + pipeline_config: ModelConfig, + agent: AgentConfig, + agent_config_path: str, + scenario_db_path: str, + output_dir: Path, + port: int, + conversation_id: str, + language: str = "en", + turn_end_fallback_time: int | None = None, + ): + super().__init__( + current_date_time=current_date_time, + pipeline_config=pipeline_config, + agent=agent, + agent_config_path=agent_config_path, + scenario_db_path=scenario_db_path, + output_dir=output_dir, + port=port, + conversation_id=conversation_id, + language=language, + ) + self._audio_sample_rate = _SAMPLE_RATE + + params = self.pipeline_config.s2s_params or {} + self._params = params + self._api_key = params["api_key"] + self._base_url = params.get("base_url", "wss://api.cartesia.ai").rstrip("/") + self._cartesia_version = params.get("cartesia_version", cartesia_agents.DEFAULT_CARTESIA_VERSION) + self._turn_end_fallback_time = turn_end_fallback_time + self._model = params["model"] + self._cartesia_agent_id = params.get("assistant_agent_id") + if not self._cartesia_agent_id: + raise ValueError("Missing Cartesia assistant agent ID in s2s_params") + + self._system_prompt = self._build_system_prompt() + self._call_id: str | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._caller_hung_up: asyncio.Event | None = None + + def notify_conversation_ending(self, reason: str | None = None) -> None: + """Hang up when the caller does, so the agent stops talking into EVA's teardown.""" + try: + if self._loop is not None and self._caller_hung_up is not None: + self._loop.call_soon_threadsafe(self._caller_hung_up.set) + logger.info(f"Caller hung up ({reason}); closing the agent connection for {self.conversation_id}") + except RuntimeError: + pass + + async def start(self) -> None: + """Start the FastAPI WebSocket server (non-blocking).""" + if self._running: + logger.warning("Server already running") + return + + await asyncio.to_thread(self._check_agent) + + self.output_dir.mkdir(parents=True, exist_ok=True) + self._fw_log = FrameworkLogWriter(self.output_dir) + self._metrics_log = MetricsLogWriter(self.output_dir) + + self._app = FastAPI() + + @self._app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket): + await websocket.accept() + await self._handle_session(websocket) + + config = uvicorn.Config(self._app, host="0.0.0.0", port=self.port, log_level="warning", lifespan="off") + self._server = uvicorn.Server(config) + self._running = True + self._server_task = asyncio.create_task(self._server.serve()) + while not self._server.started: + await asyncio.sleep(0.01) + logger.info(f"Cartesia managed server started on ws://localhost:{self.port}") + + def _check_agent(self) -> None: + """Fetch the managed agent and refuse to run if it does not match the run config.""" + agent = cartesia_agents.fetch_agent(self._cartesia_agent_id, self._params) + cartesia_agents.check_agent_matches_config( + agent, + model=self._model, + language=str(self.language), + pre_tool_speech=self.pipeline_config.pre_tool_speech, + turn_end_fallback_time=self._turn_end_fallback_time, + tool_names=[t["function"]["name"] for t in self.agent.build_tools_for_agent() or []], + ) + + async def _shutdown(self) -> None: + if not self._running: + return + self._running = False + if self._server: + self._server.should_exit = True + if self._server_task: + try: + await asyncio.wait_for(self._server_task, timeout=5.0) + except TimeoutError: + self._server_task.cancel() + try: + await self._server_task + except asyncio.CancelledError: + pass + except (asyncio.CancelledError, KeyboardInterrupt): + pass + self._server = None + self._server_task = None + # Mix here so the base class skips its slower per-sample mix. + if self.user_audio_buffer and self.assistant_audio_buffer and not self._audio_buffer: + self._audio_buffer = bytearray(_mix_pcm16(self.user_audio_buffer, self.assistant_audio_buffer)) + logger.info(f"Cartesia managed server stopped on port {self.port} (call_id={self._call_id})") + + def _session_create(self) -> dict: + return { + "type": "session_create", + # Our input is continuous, so a gap means a stalled EVA process, not silence. + "audio": {"input_format": f"pcm_{_SAMPLE_RATE}", "continuous_input": True}, + "dynamic_variables": {"system_prompt": self._system_prompt, "initial_message": self.initial_message}, + } + + async def _handle_session(self, websocket: WebSocket) -> None: # noqa: C901 + """Bridge one simulator session to one agent call.""" + stream_sid = self.conversation_id + sim_connected = True + self._loop = asyncio.get_running_loop() + self._caller_hung_up = asyncio.Event() + + # Simulator turn timestamps (wall-clock ms strings, as the audit log expects). + user_speech_start_ts: str | None = None + user_speech_stop_ts: str | None = None + # The agent's own end of the caller's turn; a latency reference when it precedes user_speech_stop. + user_turn_end_ms: int | None = None + pad_user_on_turn_start = True + record_t0: float | None = None + last_sim_frame = time.monotonic() + input_fed_until = last_sim_frame # monotonic time up to which the agent has input audio + + # Agent output: PCM for the recording and mulaw for the simulator, drained in lockstep. + pcm_fifo = bytearray() + mulaw_fifo = bytearray() + output_ready = asyncio.Event() + # The resampler holds back a short tail; pace_output flushes it when a burst ends. + down = soxr.ResampleStream(_SAMPLE_RATE, 8000, 1, dtype="int16", quality="HQ") + down_pending = False + up_state = None + + # One segment per spoken agent turn, closed by its turn_ended. + segment: _SpeechSegment | None = None + # A model turn starts with the first agent audio after the caller speaks. + in_model_turn = False + last_output_audio = 0.0 + + def log_pieces(pieces: list[tuple[str, int]]) -> None: + for text, start_ms in pieces: + self.audit_log.append_assistant_output(text, timestamp_ms=str(start_ms)) + + def flush_segment(was_interrupted: bool = False) -> None: + nonlocal segment + if segment is None: + return + segment.end_piece() + pieces = segment.take_pieces() + if pieces and was_interrupted: + pieces[-1] = (f"{pieces[-1][0]} [interrupted]", pieces[-1][1]) + log_pieces(pieces) + if segment.text.strip(): + self._fw_log.llm_response(segment.text.strip(), timestamp_ms=segment.start_ms) + self._fw_log.turn_end(was_interrupted=was_interrupted) + segment = None + + def on_agent_audio(pcm: bytes) -> None: + nonlocal segment, in_model_turn, last_output_audio, down_pending, user_speech_stop_ts, user_turn_end_ms + if segment is None: + segment = _SpeechSegment(_now_ms()) + last_output_audio = time.monotonic() + pcm_fifo.extend(pcm) + samples = np.frombuffer(pcm, dtype=np.int16) + mulaw_fifo.extend(audioop.lin2ulaw(down.resample_chunk(samples).tobytes(), 2)) + down_pending = True + output_ready.set() + if not in_model_turn: + in_model_turn = True + self._fw_log.turn_start(timestamp_ms=segment.start_ms) + refs = [int(user_speech_stop_ts)] if user_speech_stop_ts else [] + if user_turn_end_ms: + refs.append(user_turn_end_ms) + if refs and self._metrics_log: + latency_ms = segment.start_ms - min(refs) + if 0 < latency_ms < 30_000: + self._metrics_log.write_latency("model_response", latency_ms / 1000, self._model) + user_speech_stop_ts = None + user_turn_end_ms = None + + def clear_output() -> None: + nonlocal down_pending + pcm_fifo.clear() + mulaw_fifo.clear() + down.clear() + down_pending = False + + def flush_resampler_tail() -> None: + nonlocal down_pending + tail = down.resample_chunk(np.zeros(0, dtype=np.int16), last=True) + down.clear() + down_pending = False + if len(tail): + mulaw_fifo.extend(audioop.lin2ulaw(tail.tobytes(), 2)) + + url = ( + f"{self._base_url}/v1/agents/websocket/{self._cartesia_agent_id}?cartesia_version={self._cartesia_version}" + ) + try: + async with websockets.connect( + url, additional_headers={"X-API-Key": self._api_key}, max_size=None + ) as agent_ws: + + async def send_agent(payload: dict) -> None: + await agent_ws.send(json.dumps(payload, ensure_ascii=False)) + + await send_agent(self._session_create()) + # Audio sent before session_ready is dropped. + ready = json.loads(await asyncio.wait_for(agent_ws.recv(), timeout=30)) + if ready.get("type") != "session_ready": + raise RuntimeError(f"Expected session_ready, got {ready}") + self._call_id = ready["call_id"] + logger.info(f"Agent call started: call_id={self._call_id}") + + async def send_input(pcm: bytes) -> None: + await send_agent({"type": "audio_input", "audio": base64.b64encode(pcm).decode()}) + + async def run_tool(call: dict) -> None: + name = call["tool_name"] + try: + result = await self.execute_tool(name, call["parameters"] or {}) + if not isinstance(result, str): + result = json.dumps(result, ensure_ascii=False, separators=(",", ":")) + is_error = False + except Exception as e: + logger.error(f"Tool {name} failed: {e}", exc_info=True) + result, is_error = str(e), True + if call["expects_response"]: + await send_agent( + { + "type": "client_tool_result", + "tool_call_id": call["tool_call_id"], + "result": result, + "is_error": is_error, + } + ) + + async def forward_user_audio() -> None: + nonlocal stream_sid, sim_connected, user_speech_start_ts, user_speech_stop_ts, in_model_turn + nonlocal pad_user_on_turn_start, record_t0, last_sim_frame, up_state, input_fed_until + try: + while sim_connected and self._running: + try: + raw = await asyncio.wait_for(websocket.receive_text(), timeout=1.0) + except TimeoutError: + continue + try: + msg = json.loads(raw) + except json.JSONDecodeError: + continue + event = msg.get("event") + if event == "start": + stream_sid = msg.get("start", {}).get("streamSid", stream_sid) + elif event == "stop": + sim_connected = False + break + elif event == "user_speech_start": + user_speech_start_ts = msg.get("timestamp_ms") + pad_user_on_turn_start = True + in_model_turn = False + elif event == "user_speech_stop": + user_speech_stop_ts = msg.get("timestamp_ms") + elif event == "media": + mulaw = parse_twilio_media_message(raw) + if not mulaw: + continue + pcm_8k = audioop.ulaw2lin(mulaw, 2) + pcm, up_state = audioop.ratecv(pcm_8k, 2, 1, 8000, _SAMPLE_RATE, up_state) + now = time.monotonic() + last_sim_frame = now + input_fed_until = max(input_fed_until, now) + len(mulaw) / 8000 + if record_t0 is None: + record_t0 = now + if pad_user_on_turn_start: + _pad_buffer_to_walltime(self.user_audio_buffer, now - record_t0, _SAMPLE_RATE) + pad_user_on_turn_start = False + self.user_audio_buffer.extend(pcm) + await send_input(pcm) + except WebSocketDisconnect: + logger.info("Simulator WebSocket disconnected") + finally: + sim_connected = False + + async def feed_input_silence() -> None: + nonlocal input_fed_until + next_tick = time.monotonic() + while sim_connected and self._running: + next_tick += CHUNK_DURATION_S + await asyncio.sleep(max(0.0, next_tick - time.monotonic())) + now = time.monotonic() + if now - last_sim_frame > INPUT_SILENCE_GAP_S and input_fed_until < now: + await send_input(_SILENCE_FRAME) + input_fed_until += CHUNK_DURATION_S + elif next_tick < now - 1.0: + next_tick = now + + async def receive_agent_events() -> None: + nonlocal user_turn_end_ms, segment + async for raw in agent_ws: + msg = json.loads(raw) + event = msg.get("type") + if event == "audio_output": + on_agent_audio(base64.b64decode(msg["audio"])) + elif event == "turn_output_text_delta": + if msg["role"] == "assistant": + if segment is None: + segment = _SpeechSegment(_now_ms()) + segment.add_text(msg["text"]) + log_pieces(segment.take_pieces()) + elif event == "turn_ended": + text = msg["text"].strip() + if msg["role"] == "user": + user_turn_end_ms = _now_ms() + if text: + self.audit_log.append_user_input( + text, timestamp_ms=user_speech_start_ts or str(user_turn_end_ms) + ) + # The agent also ends a textless turn per tool call; its speech continues through it. + elif text or not msg["tool_calls"]: + if segment is None and text: + segment = _SpeechSegment(_now_ms()) + segment.add_text(msg["text"]) + flush_segment(was_interrupted=msg["interrupted"]) + elif event == "client_tool_call": + # Words spoken before the call stay before it in the trace. + if segment is not None: + segment.end_piece() + log_pieces(segment.take_pieces()) + asyncio.create_task(run_tool(msg)) + elif event == "audio_output_clear": + clear_output() + elif event == "error": + logger.error(f"Agent error: {msg}") + logger.info( + f"Agent connection closed for {self.conversation_id}: " + f"code={agent_ws.close_code} reason={agent_ws.close_reason!r}" + ) + + async def pace_output() -> None: + nonlocal record_t0 + next_send = time.monotonic() + resume = True + while self._running: + if not mulaw_fifo: + if down_pending and time.monotonic() - last_output_audio > 2 * CHUNK_DURATION_S: + flush_resampler_tail() + if mulaw_fifo: + continue + output_ready.clear() + try: + await asyncio.wait_for(output_ready.wait(), timeout=CHUNK_DURATION_S) + except TimeoutError: + pass + if not mulaw_fifo: + resume = True + continue + now = time.monotonic() + if record_t0 is None: + record_t0 = now + # Place each burst at its real time; append contiguously within it. + if resume: + _pad_buffer_to_walltime(self.assistant_audio_buffer, now - record_t0, _SAMPLE_RATE) + resume = False + chunk = bytes(mulaw_fifo[:MULAW_CHUNK_SIZE]) + del mulaw_fifo[:MULAW_CHUNK_SIZE] + self.assistant_audio_buffer.extend(pcm_fifo[:PCM_CHUNK_SIZE]) + del pcm_fifo[:PCM_CHUNK_SIZE] + if sim_connected: + try: + await websocket.send_text(create_twilio_media_message(stream_sid, chunk)) + except Exception: + return + next_send = max(next_send, now) + CHUNK_DURATION_S + await asyncio.sleep(max(0.0, next_send - time.monotonic())) + + tasks = { + asyncio.create_task(forward_user_audio()): "user_audio", + asyncio.create_task(feed_input_silence()): "input_silence", + asyncio.create_task(receive_agent_events()): "agent_events", + asyncio.create_task(pace_output()): "audio_pacer", + asyncio.create_task(self._caller_hung_up.wait()): "caller_hung_up", + } + done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for task in done: + exc = task.exception() if not task.cancelled() else None + if exc: + logger.error(f"Task '{tasks[task]}' failed: {exc}", exc_info=exc) + else: + logger.info(f"Task '{tasks[task]}' completed") + for task in pending: + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + flush_segment() + except Exception as e: + logger.error(f"Cartesia managed session error: {e}", exc_info=True) + finally: + logger.info(f"Client disconnected from Cartesia managed server (call_id={self._call_id})") diff --git a/src/eva/assistant/pipeline/services.py b/src/eva/assistant/pipeline/services.py index 7eae49c36..4cc48c5b1 100644 --- a/src/eva/assistant/pipeline/services.py +++ b/src/eva/assistant/pipeline/services.py @@ -56,6 +56,7 @@ from eva.assistant.pipeline.nvidia_baseten import BasetenSTTService, BasetenTTSService from eva.assistant.pipeline.realtime_llm import InstrumentedRealtimeLLMService from eva.models.agents import AgentConfig +from eva.models.config import base_language # Conditional Gemini imports - may fail if google-genai package version is incompatible try: @@ -87,14 +88,6 @@ SAMPLE_RATE = 24000 -def _base_language(tag: str) -> str: - """Return the ISO 639-1 base code from a BCP 47 tag (e.g. 'es-MX' → 'es'). - - Used for providers that only accept the two-letter base code (e.g. Whisper/OpenAI STT). - """ - return tag.split("-")[0].split("_")[0] - - def _to_language_enum(tag: str) -> Language: """Convert a BCP 47 tag to a pipecat Language enum. @@ -104,7 +97,7 @@ def _to_language_enum(tag: str) -> Language: try: return Language(tag) except ValueError: - return Language(_base_language(tag)) + return Language(base_language(tag)) # Round-robin counters for load-balanced URLs (one per service type) @@ -191,7 +184,7 @@ def create_stt_service( api_key=api_key, base_url=url, model=params["model"], - language=_base_language(language_code), + language=base_language(language_code), sample_rate=SAMPLE_RATE, ) @@ -204,7 +197,7 @@ def create_stt_service( # Defaults to the record's language; an explicit `language_hints` in STT_PARAMS overrides it, # and setting it blank (e.g. [] or "") omits the parameter so Flux auto-detects. if params["model"] == "flux-general-multi": - language_hints = params.get("language_hints", [_base_language(language_code)]) + language_hints = params.get("language_hints", [base_language(language_code)]) if isinstance(language_hints, str): language_hints = [language_hints] language_hints = [h.strip() for h in language_hints or [] if h and h.strip()] @@ -241,7 +234,7 @@ def create_stt_service( sample_rate=SAMPLE_RATE, commit_strategy=CommitStrategy.VAD, settings=ElevenLabsRealtimeSTTService.Settings( - language=_base_language(language_code), + language=base_language(language_code), **elevenlabs_settings_kwargs, ), ) @@ -274,7 +267,7 @@ def create_stt_service( logger.info(f"Using OpenAI STT: {params['model']}") # Whisper only accepts ISO 639-1 base codes (e.g. "es", not "es-MX") # params["language"] takes precedence if explicitly set by the user - whisper_lang = params.get("language") or _base_language(language_code) + whisper_lang = params.get("language") or base_language(language_code) stt_service = OpenAISTTService( api_key=api_key, base_url=url, @@ -474,7 +467,7 @@ def create_tts_service( sample_rate=SAMPLE_RATE, settings=ElevenLabsTTSService.Settings( voice=params.get("voice_id", "hpp4J3VqNfWAUOO0d1Us"), - language=_base_language(language_code), + language=base_language(language_code), **elevenlabs_settings_kwargs, ), ) diff --git a/src/eva/models/config.py b/src/eva/models/config.py index 0bc846bc6..e13089b20 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -259,6 +259,9 @@ def pipeline_parts(self) -> dict[str, str]: "s2s": get_model_alias_from_params(self.s2s_params) or self.s2s, **_fetch_elevenlabs_agent_models(self.s2s_params), } + if self.s2s == "cartesia": + # The model is the agent's LLM; name the framework too + return {"s2s": f"cartesia-{get_model_alias_from_params(self.s2s_params)}"} return {"s2s": get_model_alias_from_params(self.s2s_params)} case PipelineType.CASCADE: return { @@ -353,8 +356,8 @@ def get_pipeline_type(model_data: dict) -> PipelineType: ``llm_model`` in a flat dict. """ if s2s_value := model_data.get("s2s"): - # ElevenLabs uses s2s_params for configuration but is a cascade pipeline internally - if s2s_value == "elevenlabs": + # These frameworks use s2s_params but report the text they speak, as cascade pipelines do + if s2s_value in ("elevenlabs", "cartesia"): return PipelineType.CASCADE # Ultravox uses s2s_params for plumbing but is an audio-LLM (audio in, text out, separate TTS) if s2s_value == "ultravox": @@ -414,6 +417,14 @@ class BehaviorType(StrEnum): } +def base_language(tag: str) -> str: + """Return the ISO 639-1 base code from a BCP 47 tag (e.g. 'es-MX' → 'es'). + + Used for providers that only accept the two-letter base code (e.g. Whisper/OpenAI STT, Cartesia Managed Agents). + """ + return tag.split("-")[0].split("_")[0] + + class PerturbationConfig(BaseModel): """Perturbations applied to the simulated user during a benchmark run. @@ -534,19 +545,20 @@ class ModelDeployment(DeploymentTypedDict): ) # Framework selection - framework: Literal["pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice", "smallest_hydra"] = ( - Field( - "pipecat", - description=( - "Agent framework to use for the assistant server." - "'pipecat' (default): Pipecat pipeline." - "'openai_realtime': OpenAI Realtime API directly." - "'gemini_live': Gemini Live API via google-genai." - "'elevenlabs': ElevenLabs Conversational AI API." - "'grok_voice': xAI Grok voice realtime API." - "'smallest_hydra': Smallest Hydra speech-to-speech API." - ), - ) + framework: Literal[ + "pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice", "smallest_hydra", "cartesia" + ] = Field( + "pipecat", + description=( + "Agent framework to use for the assistant server." + "'pipecat' (default): Pipecat pipeline." + "'openai_realtime': OpenAI Realtime API directly." + "'gemini_live': Gemini Live API via google-genai." + "'elevenlabs': ElevenLabs Conversational AI API." + "'grok_voice': xAI Grok voice realtime API." + "'smallest_hydra': Smallest Hydra speech-to-speech API." + "'cartesia': Cartesia Managed Agents over a WebSocket connection." + ), ) # Run identifier diff --git a/src/eva/orchestrator/worker.py b/src/eva/orchestrator/worker.py index 0826dce66..128b2f8eb 100644 --- a/src/eva/orchestrator/worker.py +++ b/src/eva/orchestrator/worker.py @@ -53,10 +53,14 @@ def _get_server_class(framework: str) -> type[AbstractAssistantServer]: from eva.assistant.smallest_hydra_server import SmallestHydraAssistantServer return SmallestHydraAssistantServer + elif framework == "cartesia": + from eva.assistant.cartesia_server import CartesiaAssistantServer + + return CartesiaAssistantServer else: raise ValueError( f"Unknown framework: {framework!r}. " - "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice, smallest_hydra" + "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice, smallest_hydra, cartesia" ) @@ -312,10 +316,11 @@ async def _start_assistant(self) -> None: server_cls = _get_server_class(self.config.framework) resolved_db_path = self._materialize_resolved_scenario_db() # The turn-end fallback applies to the Pipecat pipelines (cascade + audio-LLM), whose - # turn detection can drop a user turn. S2S servers handle turn-taking natively and - # don't accept this kwarg. + # turn detection can drop a user turn, and to Cartesia Managed Agents, whose agents must + # be provisioned with a matching check-in time. Other S2S servers handle turn-taking + # natively and don't accept this kwarg. server_kwargs: dict[str, Any] = {} - if self.config.framework == "pipecat": + if self.config.framework in ("pipecat", "cartesia"): server_kwargs["turn_end_fallback_time"] = self.config.turn_end_fallback_time self._assistant_server = server_cls( current_date_time=self.record.current_date_time, diff --git a/src/eva/utils/cartesia_agents.py b/src/eva/utils/cartesia_agents.py new file mode 100644 index 000000000..c228017f2 --- /dev/null +++ b/src/eva/utils/cartesia_agents.py @@ -0,0 +1,108 @@ +"""Cartesia Managed Agents REST helpers, shared by provisioning, config and the assistant server. + +A managed agent's settings live on Cartesia's side and cannot be overridden per call, so EVA sets +them when the agent is provisioned (scripts/provision_cartesia_agents.py) and checks them against +the run config before a conversation starts (``check_agent_matches_config``). +""" + +import json +import urllib.error +import urllib.request +from typing import Any + +from eva.models.config import base_language + +DEFAULT_REST_URL = "https://api.cartesia.ai" +DEFAULT_CARTESIA_VERSION = "2026-08-14" + +# The agent passes EVA's prompt and greeting through untouched; see provision_cartesia_agents.py. +SYSTEM_PROMPT_VARIABLE = "{{system_prompt}}" +INITIAL_MESSAGE_VARIABLE = "{{initial_message}}" + + +def rest_url_from_params(s2s_params: dict[str, Any]) -> str: + """REST base URL matching the WebSocket ``base_url`` in s2s_params.""" + base = s2s_params.get("base_url", "wss://api.cartesia.ai").rstrip("/") + return base.replace("wss://", "https://", 1).replace("ws://", "http://", 1) + + +def request( + method: str, + path: str, + api_key: str, + body: dict | None = None, + base_url: str = DEFAULT_REST_URL, + cartesia_version: str = DEFAULT_CARTESIA_VERSION, +) -> dict: + req = urllib.request.Request( + f"{base_url}{path}", + data=json.dumps(body).encode() if body is not None else None, + method=method, + headers={"X-API-Key": api_key, "Cartesia-Version": cartesia_version, "Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=60) as resp: + raw = resp.read() + return json.loads(raw) if raw else {} + except urllib.error.HTTPError as e: + raise RuntimeError(f"HTTP {e.code} {method} {path}: {e.read().decode(errors='replace')[:1500]}") from None + + +_agent_cache: dict[str, dict] = {} + + +def fetch_agent(agent_id: str, s2s_params: dict[str, Any]) -> dict: + """Fetch an agent with its tools resolved (``config.tools`` as full tool objects). Cached per agent ID.""" + if agent_id in _agent_cache: + return _agent_cache[agent_id] + kwargs = { + "api_key": s2s_params["api_key"], + "base_url": rest_url_from_params(s2s_params), + "cartesia_version": s2s_params.get("cartesia_version", DEFAULT_CARTESIA_VERSION), + } + agent = request("GET", f"/v1/agents/{agent_id}", **kwargs) + agent["config"]["tools"] = [ + request("GET", f"/v1/agents/tools/{t['id']}", **kwargs) for t in agent["config"].get("tools", []) + ] + _agent_cache[agent_id] = agent + return agent + + +def check_agent_matches_config( + agent: dict, + model: str, + language: str, + pre_tool_speech: str, + turn_end_fallback_time: int | None, + tool_names: list[str], +) -> None: + """Raise if the provisioned agent would not run the benchmark as EVA's config describes.""" + config = agent["config"] + problems = [] + if config.get("instructions") != SYSTEM_PROMPT_VARIABLE or config.get("prompt_optimization"): + problems.append("instructions must be exactly '{{system_prompt}}' with prompt_optimization off") + if config.get("initial_message") != INITIAL_MESSAGE_VARIABLE: + problems.append("initial_message must be exactly '{{initial_message}}'") + if config["model"]["id"] != model: + problems.append(f"model is {config['model']['id']!r}, but s2s_params.model is {model!r}") + agent_language = config.get("language", {}).get("primary") + if agent_language != base_language(language): + problems.append(f"language.primary is {agent_language!r}, but EVA_LANGUAGE is {language!r}") + if config["model"].get("fallback_strategy") is not None: + problems.append("model.fallback_strategy must be null, so only the reported model answers") + check_in = config.get("turn", {}).get("inactivity_check_in_secs") + if check_in != turn_end_fallback_time: + problems.append( + f"turn.inactivity_check_in_secs is {check_in}, but turn_end_fallback_time is {turn_end_fallback_time}" + ) + tools = config.get("tools", []) + if [t["name"] for t in tools] != tool_names: + problems.append(f"tools {[t['name'] for t in tools]} do not match the EVA agent's tools {tool_names}") + wrong_speech = [t["name"] for t in tools if t.get("pre_tool_speech") != pre_tool_speech] + if wrong_speech: + problems.append(f"pre_tool_speech must be '{pre_tool_speech}' on tools {wrong_speech}") + if problems: + raise ValueError( + f"Cartesia agent {agent['id']} does not match the EVA run config; re-run " + "scripts/provision_cartesia_agents.py with matching settings:\n - " + "\n - ".join(problems) + ) diff --git a/tests/unit/assistant/test_cartesia_server.py b/tests/unit/assistant/test_cartesia_server.py new file mode 100644 index 000000000..ac3c2d446 --- /dev/null +++ b/tests/unit/assistant/test_cartesia_server.py @@ -0,0 +1,234 @@ +"""Tests for the Cartesia managed server's session_create, transcript pieces, segments and audio mix.""" + +import asyncio +import base64 +import json +import time +from unittest.mock import patch + +import numpy as np + +from eva.assistant.cartesia_server import CartesiaAssistantServer, _mix_pcm16, _SpeechSegment +from eva.utils.audio_utils import pcm16_mix + + +def _bare_server() -> CartesiaAssistantServer: + srv = object.__new__(CartesiaAssistantServer) + srv._system_prompt = "you are a helpful agent" + srv.initial_message = "Hello, how can I help you today?" + return srv + + +def test_session_create_streams_pcm_and_sends_prompt_and_greeting(): + assert _bare_server()._session_create() == { + "type": "session_create", + "audio": {"input_format": "pcm_16000", "continuous_input": True}, + "dynamic_variables": { + "system_prompt": "you are a helpful agent", + "initial_message": "Hello, how can I help you today?", + }, + } + + +def test_speech_pieces_split_at_messages_and_tool_calls(monkeypatch): + segment = _SpeechSegment(1000) + segment.add_text("One moment.") + monkeypatch.setattr("eva.assistant.cartesia_server._now_ms", lambda: 2500) + segment.add_text("Checking now") + segment.end_piece() # tool call + monkeypatch.setattr("eva.assistant.cartesia_server._now_ms", lambda: 4000) + segment.add_text(" and you are all set.") + segment.end_piece() + assert segment.take_pieces() == [("One moment.", 1000), ("Checking now", 2500), ("and you are all set.", 4000)] + + +def test_mix_matches_eva_mix_with_clipping_and_unequal_lengths(): + rng = np.random.default_rng(0) + a = rng.integers(-32768, 32767, 1000, dtype=np.int16).tobytes() + b = rng.integers(-32768, 32767, 700, dtype=np.int16).tobytes() + assert _mix_pcm16(a, b) == pcm16_mix(a, b) + assert _mix_pcm16(b, a) == pcm16_mix(b, a) + + +class _Recorder: + """Records method calls, standing in for the audit and framework logs.""" + + def __init__(self): + self.calls = [] + + def __getattr__(self, name): + return lambda *args, **kwargs: self.calls.append((name, args, kwargs)) + + +class _Simulator: + """Sends (seconds from session start, message) pairs from the user simulator, then stays silent.""" + + def __init__(self, script): + self.script = list(script) + self.started = None + + async def receive_text(self): + self.started = self.started or time.monotonic() + if not self.script: + await asyncio.Future() + at, message = self.script[0] + await asyncio.sleep(max(0.0, self.started + at - time.monotonic())) + self.script.pop(0) + return json.dumps(message) + + async def send_text(self, text): + pass + + +class _Agent: + """Replays (delay seconds, event) pairs as the agent WebSocket, then closes shortly after.""" + + close_code, close_reason = 1000, "" + + def __init__(self, events): + self.events = events + self.sent = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + pass + + async def send(self, text): + self.sent.append(json.loads(text)) + + async def recv(self): + return json.dumps({"type": "session_ready", "call_id": "call_test"}) + + async def __aiter__(self): + for delay, event in self.events: + await asyncio.sleep(delay) + yield json.dumps(event) + await asyncio.sleep(0.1) + + +def _audio(seconds: float) -> dict: + pcm = np.full(int(seconds * 16000), 1000, dtype=np.int16).tobytes() + return {"type": "audio_output", "audio": base64.b64encode(pcm).decode()} + + +def _text(text: str) -> dict: + return {"type": "turn_output_text_delta", "turn": 2, "role": "assistant", "text": text} + + +def _ended(text: str, interrupted: bool = False, tool_calls: tuple = ()) -> dict: + calls = [{"name": name, "arguments": {}} for name in tool_calls] + return { + "type": "turn_ended", + "turn": 2, + "role": "assistant", + "text": text, + "interrupted": interrupted, + "tool_calls": calls, + } + + +async def _session(events, script=(), execute_tool=None, agent=None) -> list[tuple]: + """Runs one session against the replayed agent and simulator; returns the framework log calls.""" + srv = _bare_server() + srv.execute_tool = execute_tool + srv.conversation_id, srv._running, srv._model = "conv_test", True, "managed-test" + srv._base_url, srv._cartesia_agent_id, srv._cartesia_version, srv._api_key = ( + "ws://agent", + "agent_test", + "test", + "key", + ) + srv.user_audio_buffer, srv.assistant_audio_buffer = bytearray(), bytearray() + srv.audit_log, srv._fw_log, srv._metrics_log = _Recorder(), _Recorder(), _Recorder() + with patch("eva.assistant.cartesia_server.websockets.connect", return_value=agent or _Agent(events)): + await srv._handle_session(_Simulator(script)) + return srv._fw_log.calls + + +def _responses(calls) -> list[str]: + return [args[0] for name, args, _ in calls if name == "llm_response"] + + +def _turn_starts(calls) -> int: + return sum(name == "turn_start" for name, _, _ in calls) + + +async def test_each_spoken_agent_turn_is_one_response(): + # A pause inside a turn does not split it; speech before a tool call and the reply after it are separate + # responses within one model turn. + calls = await _session( + [ + (0, _audio(0.1)), + (0, _text("One")), + (0.6, _audio(0.1)), + (0, _text(" moment.")), + (0, _ended("", tool_calls=("lookup",))), + (0, _ended("One moment.")), + (0.2, _audio(0.1)), + (0, _text("You are all set.")), + (0, _ended("You are all set.")), + ] + ) + assert _responses(calls) == ["One moment.", "You are all set."] + assert _turn_starts(calls) == 1 + + +async def test_caller_speech_starts_a_new_turn_and_interruptions_are_marked(): + calls = await _session( + [ + (0, _audio(1.0)), + (0, _text("Let me explain the")), + (0.1, {"type": "audio_output_clear"}), + (0, _ended("Let me explain the", interrupted=True)), + (0.3, _audio(0.1)), + (0, _text("Sure.")), + (0, _ended("Sure.")), + ], + script=[(0.05, {"event": "user_speech_start", "timestamp_ms": "1"})], + ) + assert _responses(calls) == ["Let me explain the", "Sure."] + assert _turn_starts(calls) == 2 + assert [kwargs["was_interrupted"] for name, _, kwargs in calls if name == "turn_end"] == [True, False] + + +def _tool_call(call_id: str, name: str, parameters: dict, expects_response: bool = True) -> dict: + return { + "type": "client_tool_call", + "tool_call_id": call_id, + "tool_name": name, + "parameters": parameters, + "expects_response": expects_response, + } + + +async def test_tool_calls_run_eva_tools_and_return_results_and_errors(): + executed = [] + + async def execute_tool(name, params): + executed.append((name, params)) + if name == "broken": + raise RuntimeError("database unavailable") + return {"status": "success", "reservation": {"id": params["confirmation_number"]}} + + agent = _Agent( + [ + (0, _tool_call("t1", "get_reservation", {"confirmation_number": "ABC123"})), + (0, _tool_call("t2", "broken", {})), + (0, _tool_call("t3", "notify", {}, expects_response=False)), + ] + ) + await _session([], execute_tool=execute_tool, agent=agent) + + assert executed == [("get_reservation", {"confirmation_number": "ABC123"}), ("broken", {}), ("notify", {})] + results = {m["tool_call_id"]: m for m in agent.sent if m["type"] == "client_tool_result"} + assert set(results) == {"t1", "t2"} # no result for a tool that expects none + assert json.loads(results["t1"]["result"]) == {"status": "success", "reservation": {"id": "ABC123"}} + assert results["t1"]["is_error"] is False + assert results["t2"] == { + "type": "client_tool_result", + "tool_call_id": "t2", + "result": "database unavailable", + "is_error": True, + } diff --git a/tests/unit/models/test_config_pipeline_type.py b/tests/unit/models/test_config_pipeline_type.py index 85e25c8c5..46dbd077b 100644 --- a/tests/unit/models/test_config_pipeline_type.py +++ b/tests/unit/models/test_config_pipeline_type.py @@ -10,6 +10,9 @@ def test_cascade_from_llm_key(self): def test_s2s_from_s2s_key(self): assert get_pipeline_type({"s2s": "gpt-realtime-mini"}) == PipelineType.S2S + def test_cartesia_is_cascade(self): + assert get_pipeline_type({"s2s": "cartesia"}) == PipelineType.CASCADE + def test_audio_llm_from_audio_llm_key(self): assert get_pipeline_type({"audio_llm": "ultravox"}) == PipelineType.AUDIO_LLM @@ -18,3 +21,10 @@ def test_legacy_realtime_model_returns_s2s(self): def test_empty_dict_returns_cascade(self): assert get_pipeline_type({}) == PipelineType.CASCADE + + +def test_cartesia_run_id_names_the_framework(): + from eva.models.config import ModelConfig + + cfg = ModelConfig(s2s="cartesia", s2s_params={"api_key": "k", "model": "gpt-5.5", "assistant_agent_id": "a"}) + assert cfg.pipeline_parts == {"s2s": "cartesia-gpt-5.5"} diff --git a/tests/unit/orchestrator/test_framework_dispatch.py b/tests/unit/orchestrator/test_framework_dispatch.py index 828841155..4c056909c 100644 --- a/tests/unit/orchestrator/test_framework_dispatch.py +++ b/tests/unit/orchestrator/test_framework_dispatch.py @@ -20,3 +20,9 @@ def test_unknown_framework_error_lists_grok_voice(): with pytest.raises(ValueError) as exc_info: _get_server_class("nope") assert "grok_voice" in str(exc_info.value) + + +def test_cartesia_dispatch_returns_cartesia_class(): + from eva.assistant.cartesia_server import CartesiaAssistantServer + + assert _get_server_class("cartesia") is CartesiaAssistantServer diff --git a/tests/unit/utils/test_cartesia_agents.py b/tests/unit/utils/test_cartesia_agents.py new file mode 100644 index 000000000..27668039a --- /dev/null +++ b/tests/unit/utils/test_cartesia_agents.py @@ -0,0 +1,51 @@ +"""Tests for the Cartesia managed agent check against the EVA run config.""" + +import pytest + +from eva.utils.cartesia_agents import check_agent_matches_config + + +def _agent(**overrides) -> dict: + config = { + "instructions": "{{system_prompt}}", + "prompt_optimization": False, + "initial_message": "{{initial_message}}", + "model": {"id": "gpt-5.5", "fallback_strategy": None}, + "language": {"primary": "en"}, + "turn": {"inactivity_check_in_secs": None}, + "tools": [{"name": "get_reservation", "pre_tool_speech": "auto"}], + } + config.update(overrides) + return {"id": "agent_x", "config": config} + + +def _check(agent, language="en", pre_tool_speech="auto", turn_end_fallback_time=None): + check_agent_matches_config(agent, "gpt-5.5", language, pre_tool_speech, turn_end_fallback_time, ["get_reservation"]) + + +def test_matching_agent_passes(): + _check(_agent()) + _check(_agent(turn={"inactivity_check_in_secs": 10}), turn_end_fallback_time=10) + _check(_agent(language={"primary": "fr"}), language="fr-CA") # Cartesia has no regional variants + + +@pytest.mark.parametrize( + "overrides, match", + [ + ({"model": {"id": "gpt-5.5", "fallback_strategy": {"type": "auto"}}}, "fallback_strategy"), + ({"model": {"id": "gpt-5.4", "fallback_strategy": None}}, "s2s_params.model"), + ({"language": {"primary": "fr"}}, "EVA_LANGUAGE"), + ({"prompt_optimization": True}, "prompt_optimization"), + ({"instructions": "Be nice. {{system_prompt}}"}, "instructions"), + ({"turn": {"inactivity_check_in_secs": 15}}, "inactivity_check_in_secs"), + ({"tools": []}, "do not match"), + ], +) +def test_mismatched_agent_is_rejected(overrides, match): + with pytest.raises(ValueError, match=match): + _check(_agent(**overrides)) + + +def test_pre_tool_speech_follows_eva_setting(): + with pytest.raises(ValueError, match="pre_tool_speech"): + _check(_agent(), pre_tool_speech="off")