Add Deepgram Voice Agent framework support

[weiz9]

Summary

Adds a new deepgram assistant-server framework backed by Deepgram's Voice Agent API (unified STT→LLM→TTS over a single WebSocket), so the Deepgram agent can be benchmarked like the existing S2S frameworks. Modeled on the Gemini Live server and the docs/assistant_server_contract.md contract.

Although it is configured through s2s_params, the Voice Agent is a cascade internally (STT→LLM→TTS), so it is scored as a CASCADE pipeline — STT WER, transcription accuracy, and speakability metrics all run, and the run_id/folder exposes the three component models ({stt, llm, tts}).

Changes

• New server src/eva/assistant/deepgram_server.py (DeepgramAssistantServer): bridges the Twilio-framed user-simulator WebSocket ↔ client.agent.v1.connect(), with the standard audit-log / framework-log / metrics-log / audio-buffer handling.
• Registration: deepgram added to worker._get_server_class() and the framework Literal in models/config.py.
• Pipeline mapping: get_pipeline_type() maps deepgram → CASCADE for scoring; pipeline_parts exposes {stt: listen_model, llm: alias-or-model, tts: speak_model} via the shared _param_alias() helper and shared DEFAULT_LISTEN_MODEL / DEFAULT_SPEAK_MODEL constants.
• Bring-your-own (BYO) think provider (opt-in): drive the LLM step with your own credentials/endpoint instead of Deepgram's managed provider — for long prompts that exceed managed limits and for an apples-to-apples comparison with the cascade pipeline. Supports aws_bedrock (IAM/STS creds, auto-built from AWS_* env vars when omitted), Anthropic/OpenAI/Google via think_endpoint, plus think_params and context_length passthrough. Managed config is unchanged.
• Custom Google think endpoint: handles Google's requirement that the model live in the endpoint URL (.../models/<model>:streamGenerateContent) and be omitted from settings, by sending the raw settings dict past the SDK's typed-model validation for that case.
• Fail-loud on think outages: a provider Error, or a greeting-only conversation (caller spoke but the model never replied), is logged at ERROR with guidance instead of silently producing a clean-looking 1-turn run.
• Tests: tests/unit/assistant/test_deepgram_server.py (settings + tool-conversion + BYO-think coverage) and a dispatch case in test_framework_dispatch.py.
• Docs: Deepgram reference section in assistant_server_contract.md; deepgram added to the framework enum in .env.example.
• Version: simulation_version 2.0.0 → 2.0.1. This only adds a new framework; it does not change any existing framework's outputs, so old simulations do not need re-running.

Implementation notes (found via live testing)

• Event parsing: the server dispatches on the raw WebSocket JSON type field rather than the SDK's typed event iterator, whose response-union deserialization mis-routed events (dropping transcripts and tool-call requests) — parsing the JSON directly is deterministic. Binary frames (TTS audio) are handled unchanged.
• KeepAlive: a periodic KeepAlive task prevents Deepgram's ~10s input-audio timeout from closing the session while the half-duplex agent is speaking.
• Latency: model_response latency is measured from the server-side receipt time of the simulator's user_speech_stop event (the simulator emits it on a monotonic clock).
• request_id: the Welcome event's request_id is logged at INFO so each conversation can be traced in Deepgram's server-side logs / support tickets.

Known limitation — BYO Bedrock model coverage

Deepgram's aws_bedrock think integration only supports Claude 3.5-era models (e.g. anthropic/claude-3-5-haiku-20240307-v1:0). Haiku 4.5 is rejected with "Invalid agent.think settings - model not available", so BYO Bedrock cannot match the cascade pipeline's Bedrock Haiku 4.5 — use managed or BYO Anthropic-direct (claude-haiku-4-5) for Haiku 4.5.

Config example

EVA_FRAMEWORK=deepgram
EVA_MODEL__S2S=deepgram
EVA_MODEL__S2S_PARAMS='{"api_key":"<deepgram-key>","model":"gpt-4o-mini"}'

Optional s2s_params (defaults in parentheses): think_provider (open_ai), listen_model (nova-3), speak_model (aura-2-thalia-en), alias (short metrics/run_id label; Deepgram still receives model). BYO think (all optional): think_credentials, think_endpoint, think_params, context_length, think_region. The conversation language comes from the run's language, not s2s_params.

Testing

• ruff + ruff format clean; pre-commit run --all-files passes.
• Unit suite passes across assistant / orchestrator / models, including the new Deepgram settings, tool-conversion, framework-dispatch, and BYO-think cases.
• Live end-to-end runs across airline, itsm, and medical_hr reached valid conversation ends with tool calls executing scenario-DB mutations and no framework-level errors (disconnects, audio timeouts, or event-processing failures).

[JosephMarinier]

Instead of introducing listen_model and speak_model in s2s_params, I think it would be possible to use the regular model in stt_params and tts_params. Have you considered that? I'm not sure whether that's a good idea. Let's brainstorm. What do you think?

Suggested change

	self._listen_model: str = s2s_params.get("listen_model", _DEFAULT_LISTEN_MODEL)
	self._speak_model: str = s2s_params.get("speak_model", _DEFAULT_SPEAK_MODEL)
	self._listen_model: str = (self.pipeline_config.stt_params or {}).get("model", _DEFAULT_LISTEN_MODEL)
	self._speak_model: str = (self.pipeline_config.tts_params or {}).get("model", _DEFAULT_SPEAK_MODEL)

[weiz9]

Yeah, I considered it. Downside is that Deepgram Voice Agent always uses Deepgram for STT/TTS on the same api_key: stt_params/tts_params normally pair with the stt/tts provider fields, so populating them here implies a swap-ability that doesn't exist. Keeping listen_model/speak_model in s2s_params keeps all the Voice Agent config in one validated place. Not strongly attached though, happy to switch if you prefer the consistency. WDYT?

[JosephMarinier]

Good point about the api_key; we wouldn't want to have it required in stt_params and tts_params. But, it wouldn't be, since _check_companion_services() won't check stt_params and tts_params if pipeline_type is S2S. So I think we're good; we can define model (and even alias) in stt_params and tts_params without defining api_key.

I'll update my suggestion to make sure we're imagining the same thing.

There are multiple possibilities. I'm not sure which one is the clearest. Maybe it's yours. What do you think?

[JosephMarinier]

Cool! I don't understand all the implementation details, but that looks very complete. 👍

[JosephMarinier]

Here's one possibility: reuse the existing stt/tts_params.get("model") (and even stt/tts_params.get("alias") for the config's pipeline_parts).

[JosephMarinier]

Suggested change

	# s2s_params is guaranteed non-None by _check_companion_services().
	return {
	"stt": self.s2s_params.get("listen_model", DEFAULT_LISTEN_MODEL),
	"llm": _param_alias(self.s2s_params),
	"tts": self.s2s_params.get("speak_model", DEFAULT_SPEAK_MODEL),
	# s2s_params is guaranteed non-None by _check_companion_services(), but not stt_params and tts_params.
	return {
	"stt": _param_alias(self.stt_params or {}) or DEFAULT_LISTEN_MODEL,
	"llm": _param_alias(self.s2s_params),
	"tts": _param_alias(self.tts_params or {}) or DEFAULT_SPEAK_MODEL,

[JosephMarinier]

Suggested change

	return params.get("alias") or params["model"]
	return params.get("alias") or params.get("model")

[JosephMarinier]

Suggested change

	self._listen_model: str = s2s_params.get("listen_model", DEFAULT_LISTEN_MODEL)
	self._speak_model: str = s2s_params.get("speak_model", DEFAULT_SPEAK_MODEL)
	self._listen_model: str = (self.pipeline_config.stt_params or {}).get("model", DEFAULT_LISTEN_MODEL)
	self._speak_model: str = (self.pipeline_config.tts_params or {}).get("model", DEFAULT_SPEAK_MODEL)

[JosephMarinier]

And here's another even simpler possibility: reuse the existing stt/tts.

[JosephMarinier]

Suggested change

	"stt": self.s2s_params.get("listen_model", DEFAULT_LISTEN_MODEL),
	"llm": _param_alias(self.s2s_params),
	"tts": self.s2s_params.get("speak_model", DEFAULT_SPEAK_MODEL),
	"stt": self.stt or DEFAULT_LISTEN_MODEL,
	"llm": _param_alias(self.s2s_params),
	"tts": self.tts or DEFAULT_SPEAK_MODEL,

[JosephMarinier]

Suggested change

	self._listen_model: str = s2s_params.get("listen_model", DEFAULT_LISTEN_MODEL)
	self._speak_model: str = s2s_params.get("speak_model", DEFAULT_SPEAK_MODEL)
	self._listen_model: str = self.pipeline_config.stt or DEFAULT_LISTEN_MODEL
	self._speak_model: str = self.pipeline_config.tts or DEFAULT_SPEAK_MODEL