From a16bd2e96ecfbaa2f75185346694f0cb667c1b7a Mon Sep 17 00:00:00 2001 From: sauhardjain Date: Wed, 14 Oct 2026 16:58:50 -0700 Subject: [PATCH 1/3] Add Cartesia Managed Agents assistant server --- .gitignore | 5 + docs/experiment_setup.md | 13 + scripts/cartesia_managed/provision_agents.py | 124 +++++ scripts/cartesia_managed/run.sh | 48 ++ src/eva/assistant/cartesia_managed_server.py | 505 ++++++++++++++++++ src/eva/models/config.py | 31 +- src/eva/orchestrator/worker.py | 6 +- .../assistant/test_cartesia_managed_server.py | 190 +++++++ .../unit/models/test_config_pipeline_type.py | 3 + .../orchestrator/test_framework_dispatch.py | 6 + 10 files changed, 915 insertions(+), 16 deletions(-) create mode 100644 scripts/cartesia_managed/provision_agents.py create mode 100755 scripts/cartesia_managed/run.sh create mode 100644 src/eva/assistant/cartesia_managed_server.py create mode 100644 tests/unit/assistant/test_cartesia_managed_server.py diff --git a/.gitignore b/.gitignore index bf3094b3c..2f8a4f229 100644 --- a/.gitignore +++ b/.gitignore @@ -93,6 +93,11 @@ local/ scripts/leaderboard_runs.json scripts/extract_and_generate_leaderboard.py +# Only the public Managed Agents setup belongs in version control. +scripts/cartesia_managed/* +!scripts/cartesia_managed/provision_agents.py +!scripts/cartesia_managed/run.sh + # Website data generation scripts (local only) website/scripts/ diff --git a/docs/experiment_setup.md b/docs/experiment_setup.md index 805f4d7e1..6452af2f7 100644 --- a/docs/experiment_setup.md +++ b/docs/experiment_setup.md @@ -92,6 +92,19 @@ 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_managed` 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 `provision_agents.py`: + +```bash +uv run python scripts/cartesia_managed/provision_agents.py # create or update; --model picks the LLM (default gpt-5.5) +scripts/cartesia_managed/run.sh airline # likewise itsm, medical_hr +``` + +Both read `CARTESIA_API_KEY` from the environment or `.env`. + +`s2s_params`: `api_key` and `agent_ids` (EVA agent id to Cartesia agent id) are required; `model` and `alias` label the run; `base_url` (default `wss://api.cartesia.ai`) and `cartesia_version` are optional. + ## Judge Models The table below lists the API-hosted models used as a judge. diff --git a/scripts/cartesia_managed/provision_agents.py b/scripts/cartesia_managed/provision_agents.py new file mode 100644 index 000000000..2d9d640a7 --- /dev/null +++ b/scripts/cartesia_managed/provision_agents.py @@ -0,0 +1,124 @@ +"""Create or update one Cartesia managed agent per EVA domain, with EVA's tools as client tools. + +EVA's system prompt and greeting are sent with each call as the dynamic variables system_prompt and +initial_message. Ids are kept in output/cartesia_managed_agents.json, which run.sh reads, so a rerun +updates the same agents. + + uv run python scripts/cartesia_managed/provision_agents.py [--model gpt-5.5] + +Reads CARTESIA_API_KEY from the environment or .env. +""" + +import argparse +import json +import os +import sys +import urllib.error +import urllib.request +from pathlib import Path + +from dotenv import dotenv_values + +from eva.models.agents import AgentConfig + +API = "https://api.cartesia.ai" +CARTESIA_VERSION = "2026-08-14" +DOMAINS = ("airline", "itsm", "medical_hr") +VOICE_ID = "f786b574-daa5-4673-aa0c-cbe3e8534c02" # Katie +STATE = Path("output/cartesia_managed_agents.json") + + +def request(method: str, path: str, body: dict | None = None) -> dict: + 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") + req = urllib.request.Request( + f"{API}{path}", + data=json.dumps(body).encode() if body is not None else None, + method=method, + headers={"X-API-Key": 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 SystemExit(f"HTTP {e.code} {method} {path}: {e.read().decode(errors='replace')[:1500]}") from None + + +def tool_body(fn: dict) -> dict: + return { + "type": "client", + "name": fn["name"], + "description": fn["description"], + "parameters": fn["parameters"], + "expects_response": True, + "response_timeout_secs": 20, + "pre_tool_speech": "force", + "execution_mode": "immediate", + } + + +def agent_config(model: str, tool_ids: list[str]) -> dict: + return { + "instructions": "{{system_prompt}}", + "prompt_optimization": False, # use EVA's system prompt as is + "initial_message": "{{initial_message}}", + "speculative_generation": True, + "model": {"id": model, "fallback_strategy": {"type": "auto"}}, + "language": {"primary": "en"}, + "turn": {"inactivity_check_in_secs": 15}, + "audio": { + "input": {"noise_suppression": "off", "asr_model": "auto"}, + "output": {"voice_id": VOICE_ID}, + }, + "tools": [{"id": t} for t in tool_ids], + } + + +def check_model(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") + args = parser.parse_args() + check_model(args.model) + + state = json.loads(STATE.read_text()) if STATE.exists() else {"tools": {}, "agents": {}} + state["model"] = args.model + STATE.parent.mkdir(parents=True, exist_ok=True) + for domain in 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']}" + if key in state["tools"]: + patch = {k: v for k, v in tool_body(fn).items() if k != "type"} # a tool's type cannot change + request("PATCH", f"/v1/agents/tools/{state['tools'][key]}", patch) + else: + state["tools"][key] = request("POST", "/v1/agents/tools", tool_body(fn))["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)} + if domain in state["agents"]: + request("PATCH", f"/v1/agents/{state['agents'][domain]}", payload) + else: + state["agents"][domain] = request("POST", "/v1/agents", payload)["id"] + STATE.write_text(json.dumps(state, indent=2)) + print(f"{domain}: {state['agents'][domain]} with {len(tool_ids)} tools", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/scripts/cartesia_managed/run.sh b/scripts/cartesia_managed/run.sh new file mode 100755 index 000000000..8fc5921fc --- /dev/null +++ b/scripts/cartesia_managed/run.sh @@ -0,0 +1,48 @@ +#!/usr/bin/env bash +# Run one EVA domain against the agents made by provision_agents.py. +# +# scripts/cartesia_managed/run.sh [extra eva args...] +# +# CARTESIA_API_KEY comes from the environment or .env; everything else is read from .env as for any EVA run. +# +# Our ElevenLabs simulator agents used Eric for male callers and Hope (tnSpp4vdxKPjI9w0GnoV) for female +# callers, since ElevenLabs no longer allows Natalee Champlin on new agents. +set -euo pipefail + +DOMAIN="$1" +shift + +EVA_MODEL__S2S_PARAMS="$(uv run python - <<'EOF' +import json +import os + +from dotenv import dotenv_values + +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") +state = json.load(open("output/cartesia_managed_agents.json")) +print(json.dumps({ + "api_key": key, + "model": f"managed-{state['model']}", + "alias": f"{state['model']}+cartesia-managed-agents", + "agent_ids": {f"agent_{domain}": agent_id for domain, agent_id in state["agents"].items()}, +})) +EOF +)" +export EVA_MODEL__S2S_PARAMS +export EVA_FRAMEWORK=cartesia_managed +export EVA_MODEL__LLM="" +# Managed Agents owns these components; exclude unused local settings from run artifacts. +export EVA_MODEL__STT="" +export EVA_MODEL__STT_PARAMS=null +export EVA_MODEL__TTS="" +export EVA_MODEL__TTS_PARAMS=null +export EVA_MODEL__AUDIO_LLM="" +export EVA_MODEL__AUDIO_LLM_PARAMS=null +export EVA_MODEL__S2S=cartesia_managed +# Pre-tool speech is configured per tool on the agent. +export EVA_MODEL__PRE_TOOL_SPEECH=off +export EVA_DOMAIN="$DOMAIN" + +exec uv run eva --domain "$DOMAIN" "$@" diff --git a/src/eva/assistant/cartesia_managed_server.py b/src/eva/assistant/cartesia_managed_server.py new file mode 100644 index 000000000..e167b4b1c --- /dev/null +++ b/src/eva/assistant/cartesia_managed_server.py @@ -0,0 +1,505 @@ +"""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/cartesia_managed/provision_agents.py. +""" + +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.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 +_DEFAULT_CARTESIA_VERSION = "2026-08-14" + +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 CartesiaManagedAssistantServer(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", + ): + 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._api_key = params["api_key"] + self._model = params["model"] + self._base_url = params.get("base_url", "wss://api.cartesia.ai").rstrip("/") + self._cartesia_version = params.get("cartesia_version", _DEFAULT_CARTESIA_VERSION) + agent_ids = params.get("agent_ids") or {} + if agent.id not in agent_ids: + raise ValueError(f"s2s_params.agent_ids has no Cartesia agent for EVA agent {agent.id!r}") + self._cartesia_agent_id = agent_ids[agent.id] + + 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 + + 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}") + + 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/models/config.py b/src/eva/models/config.py index 0bc846bc6..1fcf8986d 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -353,8 +353,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_managed"): return PipelineType.CASCADE # Ultravox uses s2s_params for plumbing but is an audio-LLM (audio in, text out, separate TTS) if s2s_value == "ultravox": @@ -534,19 +534,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_managed" + ] = 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_managed': 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..0d5cd2117 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_managed": + from eva.assistant.cartesia_managed_server import CartesiaManagedAssistantServer + + return CartesiaManagedAssistantServer 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_managed" ) diff --git a/tests/unit/assistant/test_cartesia_managed_server.py b/tests/unit/assistant/test_cartesia_managed_server.py new file mode 100644 index 000000000..34769d16a --- /dev/null +++ b/tests/unit/assistant/test_cartesia_managed_server.py @@ -0,0 +1,190 @@ +"""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_managed_server import CartesiaManagedAssistantServer, _mix_pcm16, _SpeechSegment +from eva.utils.audio_utils import pcm16_mix + + +def _bare_server() -> CartesiaManagedAssistantServer: + srv = object.__new__(CartesiaManagedAssistantServer) + 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_managed_server._now_ms", lambda: 2500) + segment.add_text("Checking now") + segment.end_piece() # tool call + monkeypatch.setattr("eva.assistant.cartesia_managed_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 + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + pass + + async def send(self, text): + pass + + 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=()) -> list[tuple]: + """Runs one session against the replayed agent and simulator; returns the framework log calls.""" + srv = _bare_server() + 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_managed_server.websockets.connect", return_value=_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 turns. + 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] diff --git a/tests/unit/models/test_config_pipeline_type.py b/tests/unit/models/test_config_pipeline_type.py index 85e25c8c5..c1a157b3c 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_managed_is_cascade(self): + assert get_pipeline_type({"s2s": "cartesia_managed"}) == PipelineType.CASCADE + def test_audio_llm_from_audio_llm_key(self): assert get_pipeline_type({"audio_llm": "ultravox"}) == PipelineType.AUDIO_LLM diff --git a/tests/unit/orchestrator/test_framework_dispatch.py b/tests/unit/orchestrator/test_framework_dispatch.py index 828841155..2e65422ac 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_managed_dispatch_returns_cartesia_class(): + from eva.assistant.cartesia_managed_server import CartesiaManagedAssistantServer + + assert _get_server_class("cartesia_managed") is CartesiaManagedAssistantServer From 4f7c6f011c27f59b4aafc6323d050f9cab11264a Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Tue, 20 Oct 2026 10:38:26 -0700 Subject: [PATCH 2/3] cleanup and integrate with benchmark native style --- .env.example | 17 +-- .gitignore | 5 - docs/experiment_setup.md | 22 ++- scripts/cartesia_managed/provision_agents.py | 124 ---------------- scripts/cartesia_managed/run.sh | 48 ------ scripts/provision_cartesia_agents.py | 140 ++++++++++++++++++ ...a_managed_server.py => cartesia_server.py} | 34 +++-- src/eva/models/config.py | 9 +- src/eva/orchestrator/worker.py | 15 +- src/eva/utils/cartesia_agents.py | 98 ++++++++++++ ...aged_server.py => test_cartesia_server.py} | 62 ++++++-- .../unit/models/test_config_pipeline_type.py | 11 +- .../orchestrator/test_framework_dispatch.py | 6 +- tests/unit/utils/test_cartesia_agents.py | 48 ++++++ 14 files changed, 414 insertions(+), 225 deletions(-) delete mode 100644 scripts/cartesia_managed/provision_agents.py delete mode 100755 scripts/cartesia_managed/run.sh create mode 100644 scripts/provision_cartesia_agents.py rename src/eva/assistant/{cartesia_managed_server.py => cartesia_server.py} (95%) create mode 100644 src/eva/utils/cartesia_agents.py rename tests/unit/assistant/{test_cartesia_managed_server.py => test_cartesia_server.py} (71%) create mode 100644 tests/unit/utils/test_cartesia_agents.py 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/.gitignore b/.gitignore index 2f8a4f229..bf3094b3c 100644 --- a/.gitignore +++ b/.gitignore @@ -93,11 +93,6 @@ local/ scripts/leaderboard_runs.json scripts/extract_and_generate_leaderboard.py -# Only the public Managed Agents setup belongs in version control. -scripts/cartesia_managed/* -!scripts/cartesia_managed/provision_agents.py -!scripts/cartesia_managed/run.sh - # Website data generation scripts (local only) website/scripts/ diff --git a/docs/experiment_setup.md b/docs/experiment_setup.md index 6452af2f7..9fb13cf5a 100644 --- a/docs/experiment_setup.md +++ b/docs/experiment_setup.md @@ -94,16 +94,26 @@ EVA-Bench makes turn detection parameters and options configurable via the CLI, ### Cartesia Managed Agents -`framework: cartesia_managed` 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 `provision_agents.py`: +`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_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 -uv run python scripts/cartesia_managed/provision_agents.py # create or update; --model picks the LLM (default gpt-5.5) -scripts/cartesia_managed/run.sh airline # likewise itsm, medical_hr -``` +# 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 -Both read `CARTESIA_API_KEY` from the environment or `.env`. +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 +``` -`s2s_params`: `api_key` and `agent_ids` (EVA agent id to Cartesia agent id) are required; `model` and `alias` label the run; `base_url` (default `wss://api.cartesia.ai`) and `cartesia_version` are optional. +`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 diff --git a/scripts/cartesia_managed/provision_agents.py b/scripts/cartesia_managed/provision_agents.py deleted file mode 100644 index 2d9d640a7..000000000 --- a/scripts/cartesia_managed/provision_agents.py +++ /dev/null @@ -1,124 +0,0 @@ -"""Create or update one Cartesia managed agent per EVA domain, with EVA's tools as client tools. - -EVA's system prompt and greeting are sent with each call as the dynamic variables system_prompt and -initial_message. Ids are kept in output/cartesia_managed_agents.json, which run.sh reads, so a rerun -updates the same agents. - - uv run python scripts/cartesia_managed/provision_agents.py [--model gpt-5.5] - -Reads CARTESIA_API_KEY from the environment or .env. -""" - -import argparse -import json -import os -import sys -import urllib.error -import urllib.request -from pathlib import Path - -from dotenv import dotenv_values - -from eva.models.agents import AgentConfig - -API = "https://api.cartesia.ai" -CARTESIA_VERSION = "2026-08-14" -DOMAINS = ("airline", "itsm", "medical_hr") -VOICE_ID = "f786b574-daa5-4673-aa0c-cbe3e8534c02" # Katie -STATE = Path("output/cartesia_managed_agents.json") - - -def request(method: str, path: str, body: dict | None = None) -> dict: - 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") - req = urllib.request.Request( - f"{API}{path}", - data=json.dumps(body).encode() if body is not None else None, - method=method, - headers={"X-API-Key": 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 SystemExit(f"HTTP {e.code} {method} {path}: {e.read().decode(errors='replace')[:1500]}") from None - - -def tool_body(fn: dict) -> dict: - return { - "type": "client", - "name": fn["name"], - "description": fn["description"], - "parameters": fn["parameters"], - "expects_response": True, - "response_timeout_secs": 20, - "pre_tool_speech": "force", - "execution_mode": "immediate", - } - - -def agent_config(model: str, tool_ids: list[str]) -> dict: - return { - "instructions": "{{system_prompt}}", - "prompt_optimization": False, # use EVA's system prompt as is - "initial_message": "{{initial_message}}", - "speculative_generation": True, - "model": {"id": model, "fallback_strategy": {"type": "auto"}}, - "language": {"primary": "en"}, - "turn": {"inactivity_check_in_secs": 15}, - "audio": { - "input": {"noise_suppression": "off", "asr_model": "auto"}, - "output": {"voice_id": VOICE_ID}, - }, - "tools": [{"id": t} for t in tool_ids], - } - - -def check_model(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") - args = parser.parse_args() - check_model(args.model) - - state = json.loads(STATE.read_text()) if STATE.exists() else {"tools": {}, "agents": {}} - state["model"] = args.model - STATE.parent.mkdir(parents=True, exist_ok=True) - for domain in 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']}" - if key in state["tools"]: - patch = {k: v for k, v in tool_body(fn).items() if k != "type"} # a tool's type cannot change - request("PATCH", f"/v1/agents/tools/{state['tools'][key]}", patch) - else: - state["tools"][key] = request("POST", "/v1/agents/tools", tool_body(fn))["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)} - if domain in state["agents"]: - request("PATCH", f"/v1/agents/{state['agents'][domain]}", payload) - else: - state["agents"][domain] = request("POST", "/v1/agents", payload)["id"] - STATE.write_text(json.dumps(state, indent=2)) - print(f"{domain}: {state['agents'][domain]} with {len(tool_ids)} tools", file=sys.stderr) - - -if __name__ == "__main__": - main() diff --git a/scripts/cartesia_managed/run.sh b/scripts/cartesia_managed/run.sh deleted file mode 100755 index 8fc5921fc..000000000 --- a/scripts/cartesia_managed/run.sh +++ /dev/null @@ -1,48 +0,0 @@ -#!/usr/bin/env bash -# Run one EVA domain against the agents made by provision_agents.py. -# -# scripts/cartesia_managed/run.sh [extra eva args...] -# -# CARTESIA_API_KEY comes from the environment or .env; everything else is read from .env as for any EVA run. -# -# Our ElevenLabs simulator agents used Eric for male callers and Hope (tnSpp4vdxKPjI9w0GnoV) for female -# callers, since ElevenLabs no longer allows Natalee Champlin on new agents. -set -euo pipefail - -DOMAIN="$1" -shift - -EVA_MODEL__S2S_PARAMS="$(uv run python - <<'EOF' -import json -import os - -from dotenv import dotenv_values - -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") -state = json.load(open("output/cartesia_managed_agents.json")) -print(json.dumps({ - "api_key": key, - "model": f"managed-{state['model']}", - "alias": f"{state['model']}+cartesia-managed-agents", - "agent_ids": {f"agent_{domain}": agent_id for domain, agent_id in state["agents"].items()}, -})) -EOF -)" -export EVA_MODEL__S2S_PARAMS -export EVA_FRAMEWORK=cartesia_managed -export EVA_MODEL__LLM="" -# Managed Agents owns these components; exclude unused local settings from run artifacts. -export EVA_MODEL__STT="" -export EVA_MODEL__STT_PARAMS=null -export EVA_MODEL__TTS="" -export EVA_MODEL__TTS_PARAMS=null -export EVA_MODEL__AUDIO_LLM="" -export EVA_MODEL__AUDIO_LLM_PARAMS=null -export EVA_MODEL__S2S=cartesia_managed -# Pre-tool speech is configured per tool on the agent. -export EVA_MODEL__PRE_TOOL_SPEECH=off -export EVA_DOMAIN="$DOMAIN" - -exec uv run eva --domain "$DOMAIN" "$@" diff --git a/scripts/provision_cartesia_agents.py b/scripts/provision_cartesia_agents.py new file mode 100644 index 000000000..c1824da83 --- /dev/null +++ b/scripts/provision_cartesia_agents.py @@ -0,0 +1,140 @@ +"""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: + + --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) + +Model fallback is disabled, so only the chosen model answers. + +Usage: + python scripts/provision_cartesia_agents.py [--model gpt-5.5] [--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 ModelConfig +from eva.utils import cartesia_agents + +DOMAINS = ("airline", "itsm", "medical_hr") +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) -> 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": "en"}, + "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( + "--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=15, + 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), + } + 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_managed_server.py b/src/eva/assistant/cartesia_server.py similarity index 95% rename from src/eva/assistant/cartesia_managed_server.py rename to src/eva/assistant/cartesia_server.py index e167b4b1c..d8ee27bfb 100644 --- a/src/eva/assistant/cartesia_managed_server.py +++ b/src/eva/assistant/cartesia_server.py @@ -4,7 +4,8 @@ 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/cartesia_managed/provision_agents.py. +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 @@ -30,13 +31,13 @@ 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 -_DEFAULT_CARTESIA_VERSION = "2026-08-14" MULAW_CHUNK_SIZE = 160 # 20 ms at 8 kHz CHUNK_DURATION_S = 0.02 @@ -103,7 +104,7 @@ def take_pieces(self) -> list[tuple[str, int]]: return pieces -class CartesiaManagedAssistantServer(AbstractAssistantServer): +class CartesiaAssistantServer(AbstractAssistantServer): def __init__( self, current_date_time: str, @@ -115,6 +116,7 @@ def __init__( port: int, conversation_id: str, language: str = "en", + turn_end_fallback_time: int | None = None, ): super().__init__( current_date_time=current_date_time, @@ -130,14 +132,15 @@ def __init__( self._audio_sample_rate = _SAMPLE_RATE params = self.pipeline_config.s2s_params or {} + self._params = params self._api_key = params["api_key"] - self._model = params["model"] self._base_url = params.get("base_url", "wss://api.cartesia.ai").rstrip("/") - self._cartesia_version = params.get("cartesia_version", _DEFAULT_CARTESIA_VERSION) - agent_ids = params.get("agent_ids") or {} - if agent.id not in agent_ids: - raise ValueError(f"s2s_params.agent_ids has no Cartesia agent for EVA agent {agent.id!r}") - self._cartesia_agent_id = agent_ids[agent.id] + 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 @@ -159,6 +162,8 @@ async def start(self) -> None: 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) @@ -178,6 +183,17 @@ async def websocket_endpoint(websocket: WebSocket): 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, + 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 diff --git a/src/eva/models/config.py b/src/eva/models/config.py index 1fcf8986d..7616f9422 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 { @@ -354,7 +357,7 @@ def get_pipeline_type(model_data: dict) -> PipelineType: """ if s2s_value := model_data.get("s2s"): # These frameworks use s2s_params but report the text they speak, as cascade pipelines do - if s2s_value in ("elevenlabs", "cartesia_managed"): + 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": @@ -535,7 +538,7 @@ class ModelDeployment(DeploymentTypedDict): # Framework selection framework: Literal[ - "pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice", "smallest_hydra", "cartesia_managed" + "pipecat", "openai_realtime", "gemini_live", "elevenlabs", "grok_voice", "smallest_hydra", "cartesia" ] = Field( "pipecat", description=( @@ -546,7 +549,7 @@ class ModelDeployment(DeploymentTypedDict): "'elevenlabs': ElevenLabs Conversational AI API." "'grok_voice': xAI Grok voice realtime API." "'smallest_hydra': Smallest Hydra speech-to-speech API." - "'cartesia_managed': Cartesia Managed Agents over a WebSocket connection." + "'cartesia': Cartesia Managed Agents over a WebSocket connection." ), ) diff --git a/src/eva/orchestrator/worker.py b/src/eva/orchestrator/worker.py index 0d5cd2117..128b2f8eb 100644 --- a/src/eva/orchestrator/worker.py +++ b/src/eva/orchestrator/worker.py @@ -53,14 +53,14 @@ def _get_server_class(framework: str) -> type[AbstractAssistantServer]: from eva.assistant.smallest_hydra_server import SmallestHydraAssistantServer return SmallestHydraAssistantServer - elif framework == "cartesia_managed": - from eva.assistant.cartesia_managed_server import CartesiaManagedAssistantServer + elif framework == "cartesia": + from eva.assistant.cartesia_server import CartesiaAssistantServer - return CartesiaManagedAssistantServer + return CartesiaAssistantServer else: raise ValueError( f"Unknown framework: {framework!r}. " - "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice, smallest_hydra, cartesia_managed" + "Supported: pipecat, openai_realtime, gemini_live, elevenlabs, grok_voice, smallest_hydra, cartesia" ) @@ -316,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..3dc753cdb --- /dev/null +++ b/src/eva/utils/cartesia_agents.py @@ -0,0 +1,98 @@ +"""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 + +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, 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}") + 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_managed_server.py b/tests/unit/assistant/test_cartesia_server.py similarity index 71% rename from tests/unit/assistant/test_cartesia_managed_server.py rename to tests/unit/assistant/test_cartesia_server.py index 34769d16a..ac3c2d446 100644 --- a/tests/unit/assistant/test_cartesia_managed_server.py +++ b/tests/unit/assistant/test_cartesia_server.py @@ -8,12 +8,12 @@ import numpy as np -from eva.assistant.cartesia_managed_server import CartesiaManagedAssistantServer, _mix_pcm16, _SpeechSegment +from eva.assistant.cartesia_server import CartesiaAssistantServer, _mix_pcm16, _SpeechSegment from eva.utils.audio_utils import pcm16_mix -def _bare_server() -> CartesiaManagedAssistantServer: - srv = object.__new__(CartesiaManagedAssistantServer) +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 @@ -33,10 +33,10 @@ def test_session_create_streams_pcm_and_sends_prompt_and_greeting(): def test_speech_pieces_split_at_messages_and_tool_calls(monkeypatch): segment = _SpeechSegment(1000) segment.add_text("One moment.") - monkeypatch.setattr("eva.assistant.cartesia_managed_server._now_ms", lambda: 2500) + 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_managed_server._now_ms", lambda: 4000) + 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)] @@ -87,6 +87,7 @@ class _Agent: def __init__(self, events): self.events = events + self.sent = [] async def __aenter__(self): return self @@ -95,7 +96,7 @@ async def __aexit__(self, *args): pass async def send(self, text): - pass + self.sent.append(json.loads(text)) async def recv(self): return json.dumps({"type": "session_ready", "call_id": "call_test"}) @@ -128,9 +129,10 @@ def _ended(text: str, interrupted: bool = False, tool_calls: tuple = ()) -> dict } -async def _session(events, script=()) -> list[tuple]: +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", @@ -140,7 +142,7 @@ async def _session(events, script=()) -> list[tuple]: ) 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_managed_server.websockets.connect", return_value=_Agent(events)): + 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 @@ -154,7 +156,8 @@ def _turn_starts(calls) -> int: 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 turns. + # 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)), @@ -188,3 +191,44 @@ async def test_caller_speech_starts_a_new_turn_and_interruptions_are_marked(): 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 c1a157b3c..46dbd077b 100644 --- a/tests/unit/models/test_config_pipeline_type.py +++ b/tests/unit/models/test_config_pipeline_type.py @@ -10,8 +10,8 @@ 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_managed_is_cascade(self): - assert get_pipeline_type({"s2s": "cartesia_managed"}) == PipelineType.CASCADE + 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 @@ -21,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 2e65422ac..4c056909c 100644 --- a/tests/unit/orchestrator/test_framework_dispatch.py +++ b/tests/unit/orchestrator/test_framework_dispatch.py @@ -22,7 +22,7 @@ def test_unknown_framework_error_lists_grok_voice(): assert "grok_voice" in str(exc_info.value) -def test_cartesia_managed_dispatch_returns_cartesia_class(): - from eva.assistant.cartesia_managed_server import CartesiaManagedAssistantServer +def test_cartesia_dispatch_returns_cartesia_class(): + from eva.assistant.cartesia_server import CartesiaAssistantServer - assert _get_server_class("cartesia_managed") is CartesiaManagedAssistantServer + 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..f1c123b66 --- /dev/null +++ b/tests/unit/utils/test_cartesia_agents.py @@ -0,0 +1,48 @@ +"""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}, + "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, pre_tool_speech="auto", turn_end_fallback_time=None): + check_agent_matches_config(agent, "gpt-5.5", 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) + + +@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"), + ({"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") From 87ca901f8c28d9194d990916a2d7c73b698da2a6 Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Tue, 20 Oct 2026 13:18:58 -0700 Subject: [PATCH 3/3] multilang provisioning --- docs/experiment_setup.md | 3 +++ scripts/provision_cartesia_agents.py | 31 +++++++++++++++++------- src/eva/assistant/cartesia_server.py | 1 + src/eva/assistant/pipeline/services.py | 21 ++++++---------- src/eva/models/config.py | 8 ++++++ src/eva/utils/cartesia_agents.py | 12 ++++++++- tests/unit/utils/test_cartesia_agents.py | 7 ++++-- 7 files changed, 57 insertions(+), 26 deletions(-) diff --git a/docs/experiment_setup.md b/docs/experiment_setup.md index 9fb13cf5a..f11256bec 100644 --- a/docs/experiment_setup.md +++ b/docs/experiment_setup.md @@ -100,6 +100,7 @@ Some agent settings can't be changed per call, so they are set when the agent is | 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) | @@ -113,6 +114,8 @@ EVA_MODEL__S2S_PARAMS='{"model": "gpt-5.5", "api_key": "", "assist 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 diff --git a/scripts/provision_cartesia_agents.py b/scripts/provision_cartesia_agents.py index c1824da83..27c22db8d 100644 --- a/scripts/provision_cartesia_agents.py +++ b/scripts/provision_cartesia_agents.py @@ -5,14 +5,18 @@ 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] [--pre-tool-speech auto|off] - [--turn-end-fallback-time SECONDS] [--domains airline itsm medical_hr] + 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 @@ -29,11 +33,11 @@ from dotenv import dotenv_values from eva.models.agents import AgentConfig -from eva.models.config import ModelConfig +from eva.models.config import LANGUAGE_DISPLAY_NAMES, ModelConfig, base_language from eva.utils import cartesia_agents DOMAINS = ("airline", "itsm", "medical_hr") -VOICE_ID = "f786b574-daa5-4673-aa0c-cbe3e8534c02" # Katie +DEFAULT_VOICE_ID = "f786b574-daa5-4673-aa0c-cbe3e8534c02" # Katie STATE = Path("output/cartesia_agents.json") @@ -50,18 +54,20 @@ def tool_body(fn: dict, pre_tool_speech: str) -> dict: } -def agent_config(model: str, tool_ids: list[str], turn_end_fallback_time: int | None) -> dict: +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": "en"}, + "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}, + "output": {"voice_id": voice_id}, }, "tools": [{"id": t} for t in tool_ids], } @@ -81,6 +87,13 @@ def check_model(request, model: str) -> None: 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"], @@ -90,7 +103,7 @@ def main() -> None: parser.add_argument( "--turn-end-fallback-time", type=int, - default=15, + 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) @@ -125,7 +138,7 @@ def main() -> None: payload = { "name": name, "description": "EVA-Bench agent", - "config": agent_config(args.model, tool_ids, args.turn_end_fallback_time), + "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) diff --git a/src/eva/assistant/cartesia_server.py b/src/eva/assistant/cartesia_server.py index d8ee27bfb..42e792103 100644 --- a/src/eva/assistant/cartesia_server.py +++ b/src/eva/assistant/cartesia_server.py @@ -189,6 +189,7 @@ def _check_agent(self) -> None: 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 []], 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 7616f9422..e13089b20 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -417,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. diff --git a/src/eva/utils/cartesia_agents.py b/src/eva/utils/cartesia_agents.py index 3dc753cdb..c228017f2 100644 --- a/src/eva/utils/cartesia_agents.py +++ b/src/eva/utils/cartesia_agents.py @@ -10,6 +10,8 @@ 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" @@ -67,7 +69,12 @@ def fetch_agent(agent_id: str, s2s_params: dict[str, Any]) -> dict: def check_agent_matches_config( - agent: dict, model: str, pre_tool_speech: str, turn_end_fallback_time: int | None, tool_names: list[str] + 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"] @@ -78,6 +85,9 @@ def check_agent_matches_config( 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") diff --git a/tests/unit/utils/test_cartesia_agents.py b/tests/unit/utils/test_cartesia_agents.py index f1c123b66..27668039a 100644 --- a/tests/unit/utils/test_cartesia_agents.py +++ b/tests/unit/utils/test_cartesia_agents.py @@ -11,6 +11,7 @@ def _agent(**overrides) -> dict: "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"}], } @@ -18,13 +19,14 @@ def _agent(**overrides) -> dict: return {"id": "agent_x", "config": config} -def _check(agent, pre_tool_speech="auto", turn_end_fallback_time=None): - check_agent_matches_config(agent, "gpt-5.5", pre_tool_speech, turn_end_fallback_time, ["get_reservation"]) +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( @@ -32,6 +34,7 @@ def test_matching_agent_passes(): [ ({"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"),