diff --git a/.github/workflows/docker-build-check.yml b/.github/workflows/docker-build-check.yml index ba144bd7..9b272295 100644 --- a/.github/workflows/docker-build-check.yml +++ b/.github/workflows/docker-build-check.yml @@ -59,8 +59,8 @@ jobs: push: ${{ github.event_name == 'workflow_dispatch' && inputs.publish == true }} tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max + cache-from: type=gha,scope=api-check + cache-to: type=gha,scope=api-check,mode=max platforms: linux/amd64 build-worker: @@ -105,8 +105,8 @@ jobs: push: ${{ github.event_name == 'workflow_dispatch' && inputs.publish == true }} tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max + cache-from: type=gha,scope=worker-check + cache-to: type=gha,scope=worker-check,mode=max platforms: linux/amd64 build-args: | INSTALL_EXTRAS=qualitative-voice,reports diff --git a/app/api/v1/routes/evaluator_results.py b/app/api/v1/routes/evaluator_results.py index 3d8f5e55..08a8884e 100644 --- a/app/api/v1/routes/evaluator_results.py +++ b/app/api/v1/routes/evaluator_results.py @@ -4,7 +4,7 @@ from sqlalchemy.orm import Session from sqlalchemy import and_ from uuid import UUID -from typing import List, Optional +from typing import List, Optional, Dict, Any from app.database import get_db from app.dependencies import get_organization_id, get_api_key @@ -21,6 +21,115 @@ router = APIRouter(prefix="/evaluator-results", tags=["evaluator-results"]) +def _derive_speaker_segments_from_call_data( + call_data: Optional[Dict[str, Any]], + provider_platform: Optional[str], +) -> Optional[List[Dict[str, Any]]]: + """Derive speaker segments from provider call_data without persisting duplicates.""" + if not isinstance(call_data, dict): + return None + + platform = (provider_platform or "").lower() + segments: List[Dict[str, Any]] = [] + + def _append_segment(speaker: str, text: str, start: float, end: float): + if not text or not str(text).strip(): + return + segments.append( + { + "speaker": speaker, + "text": str(text).strip(), + "start": float(start or 0), + "end": float(end or start or 0), + } + ) + + if platform == "vapi": + transcript_object = call_data.get("transcript_object", []) + if isinstance(transcript_object, list) and transcript_object: + for entry in transcript_object: + role = entry.get("role", "") + if role == "user": + speaker = "Speaker 1" + elif role in ("agent", "assistant", "bot"): + speaker = "Speaker 2" + else: + continue + start = entry.get("seconds_from_start", 0) + duration_ms = entry.get("duration_ms", 0) + _append_segment(speaker, entry.get("content", ""), start, start + ((duration_ms or 0) / 1000)) + else: + artifact = call_data.get("artifact", {}) + messages = call_data.get("messages", []) or (artifact.get("messages", []) if isinstance(artifact, dict) else []) + if isinstance(messages, list): + for msg in messages: + role = msg.get("role", "") + if role == "user": + speaker = "Speaker 1" + elif role in ("agent", "assistant", "bot"): + speaker = "Speaker 2" + else: + continue + start = msg.get("secondsFromStart", 0) + duration_ms = msg.get("duration", 0) + content = msg.get("message", "") or msg.get("content", "") + _append_segment(speaker, content, start, start + ((duration_ms or 0) / 1000)) + + elif platform == "retell": + transcript_object = call_data.get("transcript_object", []) + if isinstance(transcript_object, list) and transcript_object: + for entry in transcript_object: + role = entry.get("role", "") + speaker = "Speaker 1" if role == "user" else "Speaker 2" + start = entry.get("start_time", entry.get("timestamp", 0)) or 0 + end = entry.get("end_time", start) or start + _append_segment(speaker, entry.get("content", "") or entry.get("text", ""), start, end) + else: + transcript = call_data.get("transcript", "") + if isinstance(transcript, str): + for line in transcript.split("\n"): + line = line.strip() + if not line: + continue + if line.lower().startswith("user:"): + _append_segment("Speaker 1", line.split(":", 1)[1], 0, 0) + elif line.lower().startswith("agent:"): + _append_segment("Speaker 2", line.split(":", 1)[1], 0, 0) + + elif platform == "elevenlabs": + transcript_object = call_data.get("transcript_object", []) + if isinstance(transcript_object, list): + for entry in transcript_object: + speaker_raw = str(entry.get("speaker", "")).lower() + speaker = "Speaker 2" if speaker_raw in ("agent", "assistant", "ai") else "Speaker 1" + _append_segment( + speaker, + entry.get("text", ""), + entry.get("start", 0), + entry.get("end", entry.get("start", 0)), + ) + elif isinstance(call_data.get("transcript"), list): + for entry in call_data.get("transcript", []): + role = entry.get("role", "") + speaker = "Speaker 2" if role in ("agent", "assistant", "ai") else "Speaker 1" + t = entry.get("time_in_call_secs", 0) + _append_segment(speaker, entry.get("message", "") or entry.get("text", ""), t, t) + + return segments or None + + +def _resolve_speaker_segments(result: EvaluatorResult) -> Optional[List[Dict[str, Any]]]: + """ + Prefer deriving speaker segments from provider call_data for provider-linked results. + Falls back to persisted speaker_segments for non-provider/manual results. + """ + if result.provider_platform and isinstance(result.call_data, dict): + derived = _derive_speaker_segments_from_call_data(result.call_data, result.provider_platform) + if derived: + return derived + return result.speaker_segments + + @router.get("", response_model=List[EvaluatorResultResponse]) def list_evaluator_results( skip: int = 0, @@ -86,7 +195,7 @@ def list_evaluator_results( "status": result.status, "audio_s3_key": result.audio_s3_key, "transcription": result.transcription, - "speaker_segments": result.speaker_segments, + "speaker_segments": _resolve_speaker_segments(result), "metric_scores": result.metric_scores, "celery_task_id": result.celery_task_id, "error_message": result.error_message, @@ -171,7 +280,7 @@ def get_evaluator_result( "status": result.status, "audio_s3_key": result.audio_s3_key, "transcription": result.transcription, - "speaker_segments": result.speaker_segments, + "speaker_segments": _resolve_speaker_segments(result), "metric_scores": result.metric_scores, "celery_task_id": result.celery_task_id, "error_message": result.error_message, @@ -484,6 +593,10 @@ def re_evaluate_result( call_data = result.call_data or {} platform = (result.provider_platform or "").lower() recording_urls = call_data.get("recording_urls", {}) + provider_payload = call_data.get("provider_payload", {}) + artifact = call_data.get("artifact", {}) if isinstance(call_data, dict) else {} + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} audio_bytes = None decrypted_key = None @@ -511,9 +624,16 @@ def re_evaluate_result( audio_bytes = resp.content elif platform == "vapi": audio_url = ( - recording_urls.get("combined_url") + call_data.get("recordingUrl") + or call_data.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or recording_urls.get("combined_url") or recording_urls.get("stereo_url") or call_data.get("recordingUrl") + or provider_payload.get("recordingUrl") + or provider_payload.get("stereoRecordingUrl") ) if audio_url: resp = _http.get(audio_url, timeout=120) diff --git a/app/api/v1/routes/observability.py b/app/api/v1/routes/observability.py index 9f8b0dfe..c4cfeec1 100644 --- a/app/api/v1/routes/observability.py +++ b/app/api/v1/routes/observability.py @@ -482,8 +482,6 @@ async def evaluate_call( raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to generate unique result ID") transcript = _messages_to_transcript(messages) - speaker_segments = _messages_to_speaker_segments(messages) - duration_seconds: Optional[float] = None if call_data.get("startedAt") and call_data.get("endedAt"): try: @@ -504,7 +502,8 @@ async def evaluate_call( duration_seconds=duration_seconds, status=EvaluatorResultStatus.QUEUED.value, transcription=transcript, - speaker_segments=speaker_segments, + # Keep transcript structure in provider call_data; derive speaker segments on read. + speaker_segments=None, provider_call_id=call_recording.provider_call_id, provider_platform=call_recording.provider_platform, call_data=call_data, diff --git a/app/api/v1/routes/playground.py b/app/api/v1/routes/playground.py index f5604262..25ceb1ae 100644 --- a/app/api/v1/routes/playground.py +++ b/app/api/v1/routes/playground.py @@ -51,14 +51,15 @@ def extract_transcript_from_call_data(call_data: Dict[str, Any], provider_platfo provider_platform_lower = provider_platform.lower() if provider_platform else "" if provider_platform_lower == "vapi": - # Vapi: transcript is in call_data["transcript"] or build from transcript_object + # Vapi: keep provider payload raw and derive transcript from transcript/messages. transcript_text = call_data.get("transcript", "") # Get structured messages for speaker segments transcript_object = call_data.get("transcript_object", []) if not transcript_object: # Try messages array - messages = call_data.get("messages", []) + artifact = call_data.get("artifact", {}) if isinstance(call_data, dict) else {} + messages = call_data.get("messages", []) or artifact.get("messages", []) for msg in messages: role = msg.get("role", "unknown") content = msg.get("message", "") or msg.get("content", "") @@ -261,9 +262,14 @@ def poll_call_metrics( db.commit() db.refresh(call_recording) - # Check if call is complete (has end_timestamp or call_status indicates completion) - call_status = call_metrics.get("call_status", "") - end_timestamp = call_metrics.get("end_timestamp") + # Check if call is complete (supports raw + normalized payloads) + call_status = ( + call_metrics.get("call_status") + or call_metrics.get("status") + or "" + ) + call_status = str(call_status).lower() + end_timestamp = call_metrics.get("end_timestamp") or call_metrics.get("endedAt") # If call is complete, stop polling if end_timestamp or call_status in ["ended", "completed", "failed", "end-of-call-report", "done"]: @@ -282,7 +288,7 @@ def poll_call_metrics( logger.info(f"[Poll Call Metrics] Call complete, creating EvaluatorResult for call {provider_call_id}") # Extract transcript and speaker segments from call_data - transcript_text, speaker_segments = extract_transcript_from_call_data( + transcript_text, _ = extract_transcript_from_call_data( call_metrics, provider_platform ) @@ -298,8 +304,8 @@ def poll_call_metrics( duration_seconds = call_metrics.get("duration_seconds", 0) if not duration_seconds: # Try to calculate from timestamps - start_ts = call_metrics.get("start_timestamp") - end_ts = call_metrics.get("end_timestamp") + start_ts = call_metrics.get("start_timestamp") or call_metrics.get("startedAt") + end_ts = call_metrics.get("end_timestamp") or call_metrics.get("endedAt") if start_ts and end_ts: try: from dateutil import parser @@ -333,10 +339,17 @@ def poll_call_metrics( if resp.status_code == 200: audio_bytes = resp.content elif plat == "vapi": + artifact = call_metrics.get("artifact", {}) + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} audio_url = ( - recording_urls.get("combined_url") + call_metrics.get("recordingUrl") + or call_metrics.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or recording_urls.get("combined_url") or recording_urls.get("stereo_url") - or call_metrics.get("recordingUrl") ) if audio_url: resp = _http.get(audio_url, timeout=120) @@ -371,7 +384,6 @@ def poll_call_metrics( status=EvaluatorResultStatus.QUEUED.value, audio_s3_key=audio_s3_key, transcription=transcript_text, - speaker_segments=speaker_segments if speaker_segments else None, provider_call_id=provider_call_id, provider_platform=provider_platform, call_data=call_metrics, @@ -906,31 +918,55 @@ async def re_evaluate_call_recording( decrypted_key = decrypt_api_key(integration.api_key) - recording_urls = call_data.get("recording_urls", {}) - audio_bytes = None - - if platform == "elevenlabs": - audio_url = recording_urls.get("conversation_audio") - if audio_url: - resp = http_requests.get(audio_url, headers={"xi-api-key": decrypted_key}, timeout=120) - if resp.status_code == 200: - audio_bytes = resp.content - elif platform == "retell": - audio_url = call_data.get("recording_url") - if audio_url: - resp = http_requests.get(audio_url, timeout=120) - if resp.status_code == 200: - audio_bytes = resp.content - elif platform == "vapi": - audio_url = ( - recording_urls.get("combined_url") - or recording_urls.get("stereo_url") - or call_data.get("recordingUrl") - ) - if audio_url: - resp = http_requests.get(audio_url, timeout=120) - if resp.status_code == 200: - audio_bytes = resp.content + def _download_audio_from_payload(payload: Dict[str, Any]): + payload_urls = payload.get("recording_urls", {}) if isinstance(payload, dict) else {} + artifact = payload.get("artifact", {}) if isinstance(payload, dict) else {} + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} + url = None + headers = None + if platform == "elevenlabs": + url = payload_urls.get("conversation_audio") + headers = {"xi-api-key": decrypted_key} + elif platform == "retell": + url = payload.get("recording_url") + elif platform == "vapi": + url = ( + payload.get("recordingUrl") + or payload.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or payload_urls.get("combined_url") + or payload_urls.get("stereo_url") + ) + if not url: + return None, None + response = http_requests.get(url, headers=headers, timeout=120) + if response.status_code != 200: + return None, response + return response.content, response + + audio_bytes, resp = _download_audio_from_payload(call_data) + + # Retry once with fresh provider payload (new signed URL) using provider_call_id + if not audio_bytes and call_recording.provider_call_id: + try: + provider_class = get_voice_provider(platform) + provider_kwargs: Dict[str, Any] = {"api_key": decrypted_key} + if platform == "vapi" and integration.public_key: + provider_kwargs["public_key"] = integration.public_key + provider = provider_class(**provider_kwargs) + if hasattr(provider, "retrieve_call_metrics"): + refreshed_call_data = provider.retrieve_call_metrics(call_recording.provider_call_id) + if isinstance(refreshed_call_data, dict) and refreshed_call_data: + call_data = refreshed_call_data + call_recording.call_data = refreshed_call_data + db.commit() + logger.info(f"[Re-evaluate] Refreshed provider call data for call {call_recording.provider_call_id}") + audio_bytes, resp = _download_audio_from_payload(call_data) + except Exception as refresh_err: + logger.warning(f"[Re-evaluate] Provider audio URL refresh failed: {refresh_err}") if not audio_bytes: raise HTTPException( @@ -952,14 +988,16 @@ async def re_evaluate_call_recording( logger.info(f"[Re-evaluate] Reusing existing S3 audio: {audio_s3_key}") # --- Extract transcript from existing call data ------------------------ - transcript_text, speaker_segments = extract_transcript_from_call_data(call_data, platform) + transcript_text, _ = extract_transcript_from_call_data(call_data, platform) # --- Create or reset EvaluatorResult ----------------------------------- if existing_result: existing_result.status = EvaluatorResultStatus.QUEUED.value existing_result.audio_s3_key = audio_s3_key + # Preserve full provider payload on re-evaluation so debug/inspection data + # is not reduced to call_analysis-only shape. + existing_result.call_data = call_data if isinstance(call_data, dict) else existing_result.call_data existing_result.transcription = transcript_text or existing_result.transcription - existing_result.speaker_segments = speaker_segments if speaker_segments else existing_result.speaker_segments existing_result.metric_scores = None existing_result.error_message = None existing_result.celery_task_id = None @@ -971,6 +1009,15 @@ async def re_evaluate_call_recording( agent = db.query(Agent).filter(Agent.id == call_recording.agent_id).first() result_id = generate_unique_result_id(db) duration_seconds = call_data.get("duration_seconds", 0) + if not duration_seconds: + start_ts = call_data.get("start_timestamp") or call_data.get("startedAt") + end_ts = call_data.get("end_timestamp") or call_data.get("endedAt") + if start_ts and end_ts: + try: + from dateutil import parser + duration_seconds = (parser.parse(end_ts) - parser.parse(start_ts)).total_seconds() + except Exception: + duration_seconds = 0 result_name = f"Voice AI Call - {agent.name}" if agent else "Voice AI Call" evaluator_result = EvaluatorResult( @@ -985,7 +1032,6 @@ async def re_evaluate_call_recording( status=EvaluatorResultStatus.QUEUED.value, audio_s3_key=audio_s3_key, transcription=transcript_text, - speaker_segments=speaker_segments if speaker_segments else None, provider_call_id=call_recording.provider_call_id, provider_platform=platform, call_data=call_data, @@ -1046,8 +1092,16 @@ async def stream_call_audio( # For Retell / Vapi the URL is public – redirect directly if platform in ("retell", "vapi"): + artifact = call_data.get("artifact", {}) + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} url = ( - recording_urls.get("combined_url") + call_data.get("recordingUrl") + or call_data.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or recording_urls.get("combined_url") or recording_urls.get("stereo_url") or call_data.get("recording_url") ) diff --git a/app/migrations/014_cleanup_provider_evaluator_speaker_segments.py b/app/migrations/014_cleanup_provider_evaluator_speaker_segments.py new file mode 100644 index 00000000..5e92c30a --- /dev/null +++ b/app/migrations/014_cleanup_provider_evaluator_speaker_segments.py @@ -0,0 +1,32 @@ +""" +Migration: Remove duplicated speaker_segments for provider-linked evaluator results. + +Provider integrations now keep transcript structure in call_data and derive +speaker segments on read, so persisting speaker_segments duplicates data. +""" + +from sqlalchemy import text +from sqlalchemy.orm import Session + +description = "Clear duplicated speaker_segments on provider-linked evaluator_results" + + +def upgrade(db: Session): + result = db.execute( + text( + """ + UPDATE evaluator_results + SET speaker_segments = NULL + WHERE provider_platform IS NOT NULL + AND speaker_segments IS NOT NULL + """ + ) + ) + db.commit() + print(f"Cleared speaker_segments for {result.rowcount or 0} provider-linked evaluator_results rows") + + +def downgrade(db: Session): + # No safe automatic rollback: removed segments are derived dynamically from call_data. + print("No-op downgrade for 014_cleanup_provider_evaluator_speaker_segments") + db.commit() diff --git a/app/services/audio/qualitative_voice_service.py b/app/services/audio/qualitative_voice_service.py index fdd6b792..de515dd7 100644 --- a/app/services/audio/qualitative_voice_service.py +++ b/app/services/audio/qualitative_voice_service.py @@ -15,6 +15,7 @@ """ import os +import shutil from typing import Dict, Any, Optional, List, Set, Tuple import numpy as np @@ -39,6 +40,74 @@ _transformers_model = None _speechbrain_encoder = None +# ffmpeg availability (used by transformers/librosa backends for audio decode) +_ffmpeg_checked = False +_ffmpeg_available = False +_ffmpeg_path = None + + +def _ensure_ffmpeg_available() -> bool: + """ + Ensure `ffmpeg` executable is available. + + Strategy: + 1. Use system ffmpeg from PATH if present. + 2. Lazily download bundled ffmpeg via imageio-ffmpeg and expose as `ffmpeg` + on PATH for libraries that call the `ffmpeg` binary directly. + """ + global _ffmpeg_checked, _ffmpeg_available, _ffmpeg_path + + if _ffmpeg_checked: + return _ffmpeg_available + + _ffmpeg_checked = True + + # Fast path: system ffmpeg already installed. + existing = shutil.which("ffmpeg") + if existing: + _ffmpeg_available = True + _ffmpeg_path = existing + logger.info(f"[QualitativeVoice] Using system ffmpeg: {existing}") + return True + + # Fallback: lazy download with imageio-ffmpeg. + try: + import imageio_ffmpeg + + downloaded_exe = imageio_ffmpeg.get_ffmpeg_exe() + if downloaded_exe and os.path.exists(downloaded_exe): + shim_dir = os.path.join("/tmp", "efficientai_ffmpeg") + os.makedirs(shim_dir, exist_ok=True) + shim_path = os.path.join(shim_dir, "ffmpeg") + + # Create a stable `ffmpeg` command path for subprocess callers. + if not os.path.exists(shim_path): + try: + os.symlink(downloaded_exe, shim_path) + except OSError: + # Fallback if symlink isn't allowed in this runtime. + shutil.copy2(downloaded_exe, shim_path) + os.chmod(shim_path, 0o755) + + current_path = os.environ.get("PATH", "") + if shim_dir not in current_path.split(os.pathsep): + os.environ["PATH"] = shim_dir + os.pathsep + current_path + + resolved = shutil.which("ffmpeg") + if resolved: + _ffmpeg_available = True + _ffmpeg_path = resolved + logger.info(f"[QualitativeVoice] Bootstrapped ffmpeg lazily: {resolved}") + return True + except Exception as e: + logger.warning(f"[QualitativeVoice] Lazy ffmpeg bootstrap failed: {e}") + + logger.warning( + "[QualitativeVoice] ffmpeg not available. Emotion/advanced audio metrics may be missing. " + "Install system ffmpeg or include imageio-ffmpeg." + ) + return False + def _check_availability(): """Check library availability on first use (lazy).""" @@ -85,6 +154,7 @@ def _check_availability(): logger.debug(f"speechbrain not available: {e}") _SPEECHMOS_AVAILABLE = _TORCH_AVAILABLE + _ensure_ffmpeg_available() def _get_torch(): @@ -196,6 +266,7 @@ def _load_audio(self, audio_path: str, target_sr: int = 16000) -> Optional[Tuple """ try: _check_availability() + _ensure_ffmpeg_available() librosa = _get_librosa() if librosa is not None: audio, sr = librosa.load(audio_path, sr=target_sr, mono=True) @@ -357,6 +428,10 @@ def calculate_emotion_category(self, audio_path: str) -> Tuple[Optional[str], Op Classify the dominant emotion in the audio. """ try: + if not _ensure_ffmpeg_available(): + logger.warning("[QualitativeVoice] Skipping emotion classification: ffmpeg unavailable") + return None, None + classifier = self._get_emotion_classifier() if classifier is None: return None, None @@ -382,6 +457,10 @@ def calculate_valence_arousal(self, audio_path: str) -> Tuple[Optional[float], O Calculate Valence and Arousal scores. """ try: + if not _ensure_ffmpeg_available(): + logger.warning("[QualitativeVoice] Skipping valence/arousal: ffmpeg unavailable") + return None, None + model = self._get_valence_arousal_model() if model is None or self._valence_arousal_processor is None: return None, None diff --git a/app/services/audio/voice_quality_service.py b/app/services/audio/voice_quality_service.py index 48ffe3e2..664eb298 100644 --- a/app/services/audio/voice_quality_service.py +++ b/app/services/audio/voice_quality_service.py @@ -61,7 +61,22 @@ def get_recording_url(call_data: Optional[Dict[str, Any]], provider_platform: Op if provider_platform == "vapi": # Vapi stores recording URLs in recording_urls object recording_urls = call_data.get("recording_urls", {}) - return recording_urls.get("combined_url") or recording_urls.get("stereo_url") or call_data.get("recordingUrl") + provider_payload = call_data.get("provider_payload", {}) + artifact = call_data.get("artifact", {}) if isinstance(call_data, dict) else {} + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} + return ( + call_data.get("recordingUrl") + or call_data.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or recording_urls.get("combined_url") + or recording_urls.get("stereo_url") + or call_data.get("recordingUrl") + or provider_payload.get("recordingUrl") + or provider_payload.get("stereoRecordingUrl") + ) elif provider_platform == "retell": # Retell stores recording URL directly return call_data.get("recording_url") diff --git a/app/services/testing/test_agent_bridge_service.py b/app/services/testing/test_agent_bridge_service.py index 9416a697..d8f4cd6e 100644 --- a/app/services/testing/test_agent_bridge_service.py +++ b/app/services/testing/test_agent_bridge_service.py @@ -845,8 +845,13 @@ async def _poll_call_results( continue # Check call status - call_status = call_metrics.get("call_status", "") - end_timestamp = call_metrics.get("end_timestamp") + call_status = ( + call_metrics.get("call_status") + or call_metrics.get("status") + or "" + ) + call_status = str(call_status).lower() + end_timestamp = call_metrics.get("end_timestamp") or call_metrics.get("endedAt") transcript = call_metrics.get("transcript", "") logger.info( @@ -861,7 +866,7 @@ async def _poll_call_results( continue # If call is complete, process results - if end_timestamp or call_status in ["ended", "completed", "failed", "done"]: + if end_timestamp or call_status in ["ended", "completed", "failed", "done", "end-of-call-report"]: call_completed = True logger.info(f"[Bridge Poll] ✅ Call completed: status={call_status}") @@ -880,17 +885,26 @@ async def _poll_call_results( result.call_data = call_metrics logger.info(f"[Bridge Poll] ✅ Stored call_data with {len(call_metrics)} keys: {list(call_metrics.keys())}") - # Extract duration (Retell uses duration_ms, Vapi uses duration_seconds) + # Extract duration (provider-specific) duration_ms = call_metrics.get("duration_ms") duration_seconds = call_metrics.get("duration_seconds") if duration_ms: result.duration_seconds = duration_ms / 1000.0 elif duration_seconds: result.duration_seconds = duration_seconds + else: + started_at = call_metrics.get("start_timestamp") or call_metrics.get("startedAt") + ended_at = call_metrics.get("end_timestamp") or call_metrics.get("endedAt") + if started_at and ended_at: + try: + from dateutil import parser + result.duration_seconds = (parser.parse(ended_at) - parser.parse(started_at)).total_seconds() + except Exception: + pass if result.duration_seconds: logger.info(f"[Bridge Poll] Duration: {result.duration_seconds:.1f}s") - # Extract transcript and speaker segments + # Extract transcript from provider call data transcript_text, speaker_segments = self._extract_transcript_from_call_data(call_metrics, provider_platform) if transcript_text: @@ -900,8 +914,7 @@ async def _poll_call_results( logger.warning("[Bridge Poll] ⚠️ No transcript extracted from call_data") if speaker_segments: - result.speaker_segments = speaker_segments - logger.info(f"[Bridge Poll] ✅ Extracted {len(speaker_segments)} speaker segments") + logger.info(f"[Bridge Poll] ✅ Derived {len(speaker_segments)} speaker segments from call_data") # Download call audio from provider and upload to S3 audio_s3_key = None @@ -910,6 +923,7 @@ async def _poll_call_results( import uuid as _uuid recording_urls = call_metrics.get("recording_urls", {}) + provider_payload = call_metrics.get("provider_payload", {}) audio_bytes = None plat = provider_platform.lower() @@ -926,10 +940,20 @@ async def _poll_call_results( if resp.status_code == 200: audio_bytes = resp.content elif plat == "vapi": + artifact = call_metrics.get("artifact", {}) if isinstance(call_metrics, dict) else {} + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} audio_url = ( - recording_urls.get("combined_url") + call_metrics.get("recordingUrl") + or call_metrics.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or recording_urls.get("combined_url") or recording_urls.get("stereo_url") or call_metrics.get("recordingUrl") + or provider_payload.get("recordingUrl") + or provider_payload.get("stereoRecordingUrl") ) if audio_url: resp = _http.get(audio_url, timeout=120) @@ -1042,11 +1066,12 @@ def _extract_transcript_from_call_data(self, call_data: dict, provider_platform: elif provider_platform == "vapi": # Vapi format + artifact = call_data.get("artifact", {}) if isinstance(call_data, dict) else {} transcript_text = call_data.get("transcript", "") # Get structured transcript object (preferred) or fall back to messages transcript_obj = call_data.get("transcript_object", []) - messages = call_data.get("messages", []) + messages = call_data.get("messages", []) or artifact.get("messages", []) # Use transcript_object if available if transcript_obj: diff --git a/app/services/voice_providers/vapi.py b/app/services/voice_providers/vapi.py index 53d8901a..97b81a47 100644 --- a/app/services/voice_providers/vapi.py +++ b/app/services/voice_providers/vapi.py @@ -242,193 +242,26 @@ def retrieve_call_metrics(self, call_id: str) -> Dict[str, Any]: data = response.json() - # Log raw response for debugging - logger.info(f"[VapiProvider] Retrieved call {call_id}, status: {data.get('status')}, ended_reason: {data.get('endedReason')}") + logger.info( + f"[VapiProvider] Retrieved call {call_id}, status: {data.get('status')}, " + f"ended_reason: {data.get('endedReason')}" + ) logger.debug(f"[VapiProvider] Raw call data keys: {list(data.keys())}") - - # Extract nested fields safely - artifact = data.get("artifact") or {} - analysis = data.get("analysis") or {} - cost_breakdown = data.get("costBreakdown") or {} - - # Get transcript from multiple possible locations - transcript = data.get("transcript") or artifact.get("transcript", "") - - # Get messages (structured conversation) from multiple possible locations - messages = data.get("messages") or artifact.get("messages", []) or [] - - # Filter out system messages for display (keep for raw_data) - display_messages = [m for m in messages if m.get("role") != "system"] - - logger.info(f"[VapiProvider] Call {call_id}: transcript_len={len(transcript) if transcript else 0}, messages_count={len(display_messages)}") - - # Log end reason for debugging - if data.get("endedReason"): - logger.info(f"[VapiProvider] Call ended reason: {data.get('endedReason')}") - - # === Timestamps and Duration === - started_at = data.get("startedAt") - ended_at = data.get("endedAt") - - duration_seconds = 0 - if started_at and ended_at: - from dateutil import parser - try: - start_time = parser.parse(started_at) - end_time = parser.parse(ended_at) - duration_seconds = (end_time - start_time).total_seconds() - except Exception as e: - logger.warning(f"[VapiProvider] Failed to parse timestamps: {e}") - - # === Recording URLs === - recording = artifact.get("recording") or {} - mono_recording = recording.get("mono") or {} - - recording_urls = { - "combined_url": data.get("recordingUrl") or artifact.get("recordingUrl") or mono_recording.get("combinedUrl"), - "stereo_url": data.get("stereoRecordingUrl") or artifact.get("stereoRecordingUrl") or recording.get("stereoUrl"), - "assistant_url": mono_recording.get("assistantUrl"), - "customer_url": mono_recording.get("customerUrl"), - } - - # === Performance Metrics (from Vapi's artifact) === - perf_metrics = artifact.get("performanceMetrics") or {} - turn_latencies = perf_metrics.get("turnLatencies") or [] - - # Use Vapi's pre-calculated averages if available - latency_stats = {} - if perf_metrics: - latency_stats = { - "model_latency_avg": perf_metrics.get("modelLatencyAverage"), - "voice_latency_avg": perf_metrics.get("voiceLatencyAverage"), - "transcriber_latency_avg": perf_metrics.get("transcriberLatencyAverage"), - "endpointing_latency_avg": perf_metrics.get("endpointingLatencyAverage"), - "turn_latency_avg": perf_metrics.get("turnLatencyAverage"), - "from_transport_latency_avg": perf_metrics.get("fromTransportLatencyAverage"), - "to_transport_latency_avg": perf_metrics.get("toTransportLatencyAverage"), - "num_assistant_interrupted": perf_metrics.get("numAssistantInterrupted", 0), - "turn_latencies": turn_latencies, - } - - # Calculate additional latency percentiles if we have turn latencies - if turn_latencies: - import numpy as np - total_latencies = [t.get("turnLatency", 0) for t in turn_latencies if t.get("turnLatency")] - if total_latencies: - # Convert NumPy types to native Python types for JSON serialization - latency_stats.update({ - "p50": float(round(np.percentile(total_latencies, 50), 2)), - "p90": float(round(np.percentile(total_latencies, 90), 2)), - "p95": float(round(np.percentile(total_latencies, 95), 2)), - "p99": float(round(np.percentile(total_latencies, 99), 2)), - "max": float(round(np.max(total_latencies), 2)), - "min": float(round(np.min(total_latencies), 2)), - "num_turns": int(len(total_latencies)) - }) - - # === Interruption Count === - interruption_count = perf_metrics.get("numAssistantInterrupted", 0) - - # If not provided by Vapi, calculate manually - if not interruption_count and messages: - for i, msg in enumerate(messages): - if msg.get("role") == "user" and i > 0: - prev_msg = messages[i-1] - if prev_msg.get("role") in ["assistant", "bot"]: - prev_agent_end = (prev_msg.get("secondsFromStart") or 0) + ((prev_msg.get("duration") or 0) / 1000) - user_start = msg.get("secondsFromStart") or 0 - if user_start < (prev_agent_end - 0.5): - interruption_count += 1 - - # === Cost Breakdown (detailed) === - analysis_cost = cost_breakdown.get("analysisCostBreakdown") or {} - - normalized_cost_breakdown = { - "transport": cost_breakdown.get("transport", 0), - "stt": cost_breakdown.get("stt", 0), - "llm": cost_breakdown.get("llm", 0), - "tts": cost_breakdown.get("tts", 0), - "vapi": cost_breakdown.get("vapi", 0), - "total": cost_breakdown.get("total", 0), - # Token usage - "llm_prompt_tokens": cost_breakdown.get("llmPromptTokens", 0), - "llm_completion_tokens": cost_breakdown.get("llmCompletionTokens", 0), - "llm_cached_prompt_tokens": cost_breakdown.get("llmCachedPromptTokens", 0), - "tts_characters": cost_breakdown.get("ttsCharacters", 0), - # Analysis costs - "analysis": { - "summary": analysis_cost.get("summary", 0), - "success_evaluation": analysis_cost.get("successEvaluation", 0), - "structured_data": analysis_cost.get("structuredData", 0), - }, - } - - # === Analysis (summary, success evaluation) === - summary = analysis.get("summary") or data.get("summary") or "" - success_evaluation = analysis.get("successEvaluation") - - normalized_analysis = { - "summary": summary, - "success_evaluation": success_evaluation, - "latency_stats": latency_stats, - "interruption_count": interruption_count, - } - # === Build Transcript Object (structured with timing) === - # Filter to only user and bot messages for the transcript object - transcript_object = [] - for msg in display_messages: - role = msg.get("role", "unknown") - content = msg.get("message", "") or msg.get("content", "") - - if not content or role == "system": - continue - - # Map roles for consistency - if role in ["bot", "assistant"]: - normalized_role = "agent" - elif role == "user": - normalized_role = "user" - else: - continue - - transcript_entry = { - "role": normalized_role, - "content": content, - "seconds_from_start": msg.get("secondsFromStart", 0), - "duration_ms": msg.get("duration", 0), - "end_time_ms": msg.get("endTime"), - "time_ms": msg.get("time"), + # Store provider payload as-is and only add generated metadata. + result = dict(data) + generated = result.get("generated") + if not isinstance(generated, dict): + generated = {} + generated.update( + { + "provider": "vapi", + "schema_mode": "raw_provider_payload", + "generated_by": "efficientai", } - - # Include word-level confidence if available - metadata = msg.get("metadata") or {} - if metadata.get("wordLevelConfidence"): - transcript_entry["words"] = metadata["wordLevelConfidence"] - - transcript_object.append(transcript_entry) + ) + result["generated"] = generated - result = { - "call_id": data.get("id"), - "call_status": data.get("status"), - "start_timestamp": started_at, - "end_timestamp": ended_at, - "duration_seconds": duration_seconds, - "cost": data.get("cost", 0), - "cost_breakdown": normalized_cost_breakdown, - "transcript": transcript, - "transcript_object": transcript_object, - "messages": display_messages, - "analysis": normalized_analysis, - "recording_urls": recording_urls, - "monitor": data.get("monitor"), - "ended_reason": data.get("endedReason") or data.get("reason"), - "metadata": data.get("metadata"), - "assistant_id": data.get("assistantId"), - "call_type": data.get("type"), - "raw_data": data - } - # Ensure all values are JSON serializable (convert NumPy types, etc.) return self._make_json_serializable(result) except Exception as e: diff --git a/app/workers/tasks/process_evaluator_result.py b/app/workers/tasks/process_evaluator_result.py index 21eea88e..815c2900 100644 --- a/app/workers/tasks/process_evaluator_result.py +++ b/app/workers/tasks/process_evaluator_result.py @@ -1,6 +1,7 @@ """Celery task: process evaluator result (transcribe and evaluate metrics).""" import time +import uuid as _uuid from uuid import UUID from loguru import logger @@ -194,6 +195,156 @@ def _categorize_metrics(enabled_metrics, has_audio): return llm_metrics, audio_metrics, skipped_scores +def _normalize_platform(platform: object) -> str: + """Normalize provider platform enum/string into lowercase string.""" + if not platform: + return "" + if hasattr(platform, "value"): + return str(platform.value).lower() + return str(platform).lower() + + +def _extract_audio_url(call_data: dict, platform: str) -> str | None: + """Extract provider-specific audio URL from call data.""" + recording_urls = call_data.get("recording_urls", {}) if isinstance(call_data, dict) else {} + provider_payload = call_data.get("provider_payload", {}) if isinstance(call_data, dict) else {} + artifact = call_data.get("artifact", {}) if isinstance(call_data, dict) else {} + recording = artifact.get("recording", {}) if isinstance(artifact, dict) else {} + mono_recording = recording.get("mono", {}) if isinstance(recording, dict) else {} + if platform == "elevenlabs": + return recording_urls.get("conversation_audio") + if platform == "retell": + return call_data.get("recording_url") + if platform == "vapi": + return ( + call_data.get("recordingUrl") + or call_data.get("stereoRecordingUrl") + or artifact.get("recordingUrl") + or artifact.get("stereoRecordingUrl") + or mono_recording.get("combinedUrl") + or recording_urls.get("combined_url") + or recording_urls.get("stereo_url") + or call_data.get("recordingUrl") + or provider_payload.get("recordingUrl") + or provider_payload.get("stereoRecordingUrl") + ) + return None + + +def _recover_missing_audio_for_result(result, db, refresh_call_data: bool = True) -> bool: + """ + Attempt to recover missing audio from provider, upload to S3, and persist key. + + Returns True when a new S3 key is successfully stored. + """ + import requests as _http + + from app.core.encryption import decrypt_api_key + from app.models.database import Agent, Integration + from app.services.storage.s3_service import s3_service + from app.services.voice_providers import get_voice_provider + + platform = _normalize_platform(result.provider_platform) + if platform not in {"retell", "vapi", "elevenlabs"}: + return False + if not result.provider_call_id: + return False + + agent = db.query(Agent).filter(Agent.id == result.agent_id).first() if result.agent_id else None + integration = None + decrypted_key = None + if agent and agent.voice_ai_integration_id: + integration = db.query(Integration).filter( + Integration.id == agent.voice_ai_integration_id, + Integration.organization_id == result.organization_id, + ).first() + if integration: + try: + decrypted_key = decrypt_api_key(integration.api_key) + except Exception as decrypt_err: + logger.warning( + f"[EvaluatorResult {result.result_id}] Unable to decrypt integration key " + f"for audio recovery: {decrypt_err}" + ) + + call_data = result.call_data or {} + if refresh_call_data and decrypted_key: + try: + provider_class = get_voice_provider(platform) + provider_kwargs = {"api_key": decrypted_key} + if platform == "vapi" and integration and getattr(integration, "public_key", None): + provider_kwargs["public_key"] = integration.public_key + provider = provider_class(**provider_kwargs) + if hasattr(provider, "retrieve_call_metrics"): + refreshed = provider.retrieve_call_metrics(result.provider_call_id) + if isinstance(refreshed, dict) and refreshed: + call_data = refreshed + result.call_data = refreshed + db.commit() + except Exception as refresh_err: + logger.warning( + f"[EvaluatorResult {result.result_id}] Audio recovery could not refresh provider metrics: " + f"{refresh_err}" + ) + + audio_url = _extract_audio_url(call_data, platform) + if not audio_url: + logger.warning( + f"[EvaluatorResult {result.result_id}] Audio recovery failed: no provider recording URL available" + ) + return False + + headers = {"xi-api-key": decrypted_key} if platform == "elevenlabs" and decrypted_key else None + try: + response = _http.get(audio_url, headers=headers, timeout=120) + except Exception as download_err: + logger.warning( + f"[EvaluatorResult {result.result_id}] Audio recovery download failed: {download_err}" + ) + return False + + if response.status_code != 200 or not response.content: + logger.warning( + f"[EvaluatorResult {result.result_id}] Audio recovery download returned " + f"status={response.status_code}" + ) + return False + + content_type = response.headers.get("content-type", "audio/mpeg") + ext = "wav" if "wav" in content_type else "mp3" + org_id = str(result.organization_id) + s3_key = ( + f"audio/organizations/{org_id}/evaluations/" + f"{result.provider_call_id}/{_uuid.uuid4()}.{ext}" + ) + + try: + s3_service.upload_file_by_key(response.content, s3_key, content_type=content_type) + except Exception as upload_err: + logger.warning( + f"[EvaluatorResult {result.result_id}] Audio recovery upload failed: {upload_err}" + ) + return False + + result.audio_s3_key = s3_key + db.commit() + db.refresh(result) + logger.info( + f"[EvaluatorResult {result.result_id}] Recovered missing audio and stored at {s3_key}" + ) + return True + + +def _all_audio_scores_download_failed(audio_scores: dict[str, dict[str, object]]) -> bool: + """Check whether every audio metric failed due to missing/unreadable S3 object.""" + if not audio_scores: + return False + return all( + isinstance(score, dict) and score.get("error") == "audio_download_failed" + for score in audio_scores.values() + ) + + @celery_app.task(name="process_evaluator_result", bind=True, max_retries=3) def process_evaluator_result_task(self, result_id: str): """ @@ -229,9 +380,11 @@ def process_evaluator_result_task(self, result_id: str): result.celery_task_id = self.request.id db.commit() - has_existing_transcript = bool(result.transcription) - try: + if not result.audio_s3_key: + _recover_missing_audio_for_result(result, db, refresh_call_data=True) + + has_existing_transcript = bool(result.transcription) if not result.audio_s3_key and not has_existing_transcript: raise ValueError("No audio S3 key or existing transcript found") @@ -259,7 +412,9 @@ def process_evaluator_result_task(self, result_id: str): result, ai_providers, db ) result.transcription = transcription - result.speaker_segments = speaker_segments if speaker_segments else None + # Avoid duplicating transcript structure when provider call_data already carries it. + if not result.call_data: + result.speaker_segments = speaker_segments if speaker_segments else None db.commit() # Step 2: Load and categorize metrics @@ -285,6 +440,20 @@ def process_evaluator_result_task(self, result_id: str): audio_metrics=audio_metrics, result_id=result.result_id, ) + + if _all_audio_scores_download_failed(audio_scores): + logger.warning( + f"[EvaluatorResult {result.result_id}] Existing S3 audio unavailable; " + "attempting provider audio recovery" + ) + recovered = _recover_missing_audio_for_result(result, db, refresh_call_data=True) + if recovered and result.audio_s3_key: + audio_scores = evaluate_audio_metrics( + audio_s3_key=result.audio_s3_key, + audio_metrics=audio_metrics, + result_id=result.result_id, + ) + metric_scores.update(audio_scores) except Exception as audio_err: logger.error( @@ -344,7 +513,14 @@ def process_evaluator_result_task(self, result_id: str): scenario=scenario, ) if call_analysis: - result.call_data = {"call_analysis": call_analysis} + existing_call_data = dict(result.call_data) if isinstance(result.call_data, dict) else {} + existing_call_data["call_analysis"] = call_analysis + generated = existing_call_data.get("generated", {}) + if not isinstance(generated, dict): + generated = {} + generated["call_analysis"] = call_analysis + existing_call_data["generated"] = generated + result.call_data = existing_call_data except Exception as analysis_err: logger.warning( f"[EvaluatorResult {result.result_id}] Call analysis failed (non-fatal): {analysis_err}" diff --git a/docker/Dockerfile.api b/docker/Dockerfile.api index 9badfd4e..689369f8 100644 --- a/docker/Dockerfile.api +++ b/docker/Dockerfile.api @@ -10,8 +10,9 @@ RUN apt-get update && apt-get install -y \ && apt-get install -y nodejs \ && rm -rf /var/lib/apt/lists/* -# Install uv -COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv +# Install uv via pip to get the glibc-linked binary (the musl-linked binary +# from ghcr.io/astral-sh/uv has DNS resolution issues inside Docker) +RUN pip install --no-cache-dir uv # Set working directory WORKDIR /app @@ -29,6 +30,11 @@ RUN --mount=type=cache,target=/root/.cache/uv \ --mount=type=cache,target=/root/.cache/pip \ uv pip install --system -e . +# Security: fail build if compromised litellm .pth file is present (CVE: supply chain attack on litellm 1.82.8) +RUN if find /usr/local/lib/ -name "litellm_init.pth" 2>/dev/null | grep -q .; then \ + echo "SECURITY: Malicious litellm_init.pth detected! Aborting build." && exit 1; \ + fi + # ============================================================ # LAYER 2: Frontend dependencies (cached unless package.json changes) # ============================================================ diff --git a/docker/Dockerfile.worker b/docker/Dockerfile.worker index 1d582edf..f6923a65 100644 --- a/docker/Dockerfile.worker +++ b/docker/Dockerfile.worker @@ -21,8 +21,9 @@ RUN apt-get update && apt-get install -y \ shared-mime-info \ && rm -rf /var/lib/apt/lists/* -# Install uv -COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv +# Install uv via pip to get the glibc-linked binary (the musl-linked binary +# from ghcr.io/astral-sh/uv has DNS resolution issues inside Docker) +RUN pip install --no-cache-dir uv # Set working directory WORKDIR /app @@ -60,6 +61,11 @@ RUN --mount=type=cache,target=/root/.cache/uv \ uv pip install --system -e .; \ fi +# Security: fail build if compromised litellm .pth file is present (CVE: supply chain attack on litellm 1.82.8) +RUN if find /usr/local/lib/ -name "litellm_init.pth" 2>/dev/null | grep -q .; then \ + echo "SECURITY: Malicious litellm_init.pth detected! Aborting build." && exit 1; \ + fi + # ============================================================ # LAYER 2: Application code (rebuilds on code changes only) # ============================================================ diff --git a/frontend/src/components/call-recordings/TestVoiceAgentResultDetails.tsx b/frontend/src/components/call-recordings/TestVoiceAgentResultDetails.tsx index b72fbe13..74c6daef 100644 --- a/frontend/src/components/call-recordings/TestVoiceAgentResultDetails.tsx +++ b/frontend/src/components/call-recordings/TestVoiceAgentResultDetails.tsx @@ -1,4 +1,4 @@ -import { useState } from 'react' +import { useState, type ReactNode } from 'react' import { Clock, MessageSquare, TrendingUp, Download, Server, BarChart3, CheckCircle, XCircle, HelpCircle, Brain, Sparkles, AudioWaveform } from 'lucide-react' @@ -11,71 +11,91 @@ const METRIC_INFO: Record = { 'Pitch Variance': { - description: 'F0 variation measuring prosodic expressiveness.', + description: 'F0 variation measuring prosodic expressiveness. Higher values indicate more expressive speech.', ideal: '20-50 Hz (natural speech)', unit: 'Hz', category: 'acoustic' }, 'Jitter': { - description: 'Cycle-to-cycle pitch period variation.', + description: 'Cycle-to-cycle pitch period variation indicating vocal stability. Lower is better.', ideal: '< 1% (healthy voice)', unit: '%', category: 'acoustic' }, 'Shimmer': { - description: 'Amplitude perturbation measuring voice quality.', + description: 'Amplitude perturbation measuring voice quality consistency. Lower is better.', ideal: '< 3% (clear voice)', unit: '%', category: 'acoustic' }, 'HNR': { - description: 'Harmonics-to-Noise Ratio measuring signal clarity.', - ideal: '> 20 dB (clear voice)', + description: 'Harmonics-to-Noise Ratio measuring signal clarity. Higher indicates cleaner voice.', + ideal: '> 20 dB (clear, non-breathy)', unit: 'dB', category: 'acoustic' }, 'MOS Score': { - description: 'Mean Opinion Score (1-5 scale).', - ideal: '4.0+ (studio quality)', + description: 'Mean Opinion Score predicting human perception of audio quality (1-5 scale).', + ideal: '4.0+ (studio quality), 3.0 (phone quality), <2.0 (poor/robotic)', + category: 'ai_voice' + }, + 'Emotion Category': { + description: 'Dominant emotion detected in the voice (angry, happy, sad, neutral, fearful, etc.).', + ideal: 'Depends on context - should match expected tone', category: 'ai_voice' }, 'Emotion Confidence': { - description: 'Confidence score for detected emotion.', + description: 'Confidence score for the detected emotion category.', ideal: '> 0.7 (high confidence)', category: 'ai_voice' }, 'Valence': { - description: 'Emotional positivity/negativity scale.', - ideal: '-1.0 to +1.0', + description: 'Emotional positivity/negativity scale. Negative = sad/angry, Positive = happy/excited.', + ideal: '-1.0 to +1.0 (context dependent)', category: 'ai_voice' }, 'Arousal': { - description: 'Emotional intensity/energy level.', - ideal: '0.3-0.6 (engaged)', + description: 'Emotional intensity/energy level. Low = calm/sleepy, High = excited/energetic.', + ideal: '0.3-0.6 (engaged but not agitated)', + category: 'ai_voice' + }, + 'Speaker Consistency': { + description: 'Voice identity stability throughout the call. Detects if voice changed mid-call (glitch).', + ideal: '> 0.8 (same voice), < 0.5 indicates voice glitch', category: 'ai_voice' }, 'Prosody Score': { - description: 'Expressiveness/drama score.', - ideal: '0.4-0.7 (natural)', + description: 'Expressiveness/drama score. Low = monotone/flat, High = expressive/dynamic.', + ideal: '0.4-0.7 (natural expressiveness)', category: 'ai_voice' }, 'Follow Instructions': { - description: 'How well the agent followed instructions.', + description: 'How well the agent followed the given instructions and guidelines.', ideal: '> 0.8 (80%+)', category: 'llm' }, 'Problem Resolution': { - description: 'Whether the agent resolved the problem.', - ideal: 'true', + description: 'Whether the agent successfully resolved the customer\'s problem or query.', + ideal: '> 0.8 (80%+)', category: 'llm' }, 'Professionalism': { - description: 'Professional demeanor and language.', + description: 'Professional demeanor, appropriate language, and courteous behavior.', ideal: '> 0.85 (85%+)', category: 'llm' }, 'Clarity and Empathy': { - description: 'Clear communication with empathy.', + description: 'Clear communication combined with understanding and acknowledgment of customer feelings.', + ideal: '> 0.8 (80%+)', + category: 'llm' + }, + 'Objective Achieved': { + description: 'Whether the conversation\'s primary objective was successfully achieved.', + ideal: 'Yes/True', + category: 'llm' + }, + 'Overall Quality': { + description: 'Holistic assessment of the entire conversation quality.', ideal: '> 0.8 (80%+)', category: 'llm' }, @@ -83,6 +103,70 @@ const METRIC_INFO: Record METRIC_INFO[metricName] || null +const SECTION_INFO: Record<'conversation' | 'ai_voice' | 'acoustic', string> = { + conversation: 'LLM-based evaluation of how well the agent handled intent, instructions, and resolution quality.', + ai_voice: 'ML-based voice quality metrics on naturalness, affect, consistency, and expressiveness.', + acoustic: 'Signal-level acoustic measurements from the recording (pitch stability, perturbation, noise ratio).', +} + +const MetricTooltip = ({ metricName }: { metricName: string }) => { + const [isVisible, setIsVisible] = useState(false) + const info = getMetricInfo(metricName) + + if (!info) return null + + return ( +
+ + {isVisible && ( +
+
{metricName}
+

{info.description}

+
+ Ideal: + {info.ideal} +
+
+
+ )} +
+ ) +} + +const SectionTooltip = ({ section }: { section: 'conversation' | 'ai_voice' | 'acoustic' }) => { + const [isVisible, setIsVisible] = useState(false) + + return ( +
+ + {isVisible && ( +
+

{SECTION_INFO[section]}

+
+
+ )} +
+ ) +} + interface TestVoiceAgentResultData { id?: string result_id?: string @@ -97,7 +181,16 @@ interface TestVoiceAgentResultData { start: number end: number }> | null - metric_scores?: Record | null + metric_scores?: Record< + string, + { + value: any + type: string + metric_name: string + skipped?: string + error?: string | null + } + > | null call_analysis?: { call_summary?: string user_sentiment?: string @@ -249,33 +342,115 @@ export default function TestVoiceAgentResultDetails({ resultData }: TestVoiceAge ) // Helper function to format metric values - const formatMetricValue = (value: any, type: string, metricName: string): string => { - if (value === null || value === undefined) return 'N/A' - if (type === 'boolean') return value ? 'Yes' : 'No' - if (type === 'rating') return `${(value * 100).toFixed(0)}%` - if (typeof value === 'number') { + const formatMetricValue = (value: any, type: string, metricName: string): ReactNode => { + if (value === null || value === undefined) return N/A + + const normalizedType = type?.toLowerCase() + + if (metricName === 'Emotion Category') { + const emotion = String(value).toLowerCase() + const emotionConfig: Record = { + neutral: { emoji: '😐', color: 'text-gray-700', bg: 'bg-gray-100' }, + happy: { emoji: '😊', color: 'text-green-700', bg: 'bg-green-100' }, + sad: { emoji: '😢', color: 'text-blue-700', bg: 'bg-blue-100' }, + angry: { emoji: '😠', color: 'text-red-700', bg: 'bg-red-100' }, + fearful: { emoji: '😨', color: 'text-purple-700', bg: 'bg-purple-100' }, + fear: { emoji: '😨', color: 'text-purple-700', bg: 'bg-purple-100' }, + surprised: { emoji: '😲', color: 'text-yellow-700', bg: 'bg-yellow-100' }, + surprise: { emoji: '😲', color: 'text-yellow-700', bg: 'bg-yellow-100' }, + disgusted: { emoji: '🤢', color: 'text-green-800', bg: 'bg-green-200' }, + disgust: { emoji: '🤢', color: 'text-green-800', bg: 'bg-green-200' }, + calm: { emoji: '😌', color: 'text-teal-700', bg: 'bg-teal-100' }, + } + const config = emotionConfig[emotion] || { emoji: '🎭', color: 'text-gray-700', bg: 'bg-gray-100' } + + return ( +
+ {config.emoji} + {value} +
+ ) + } + + if (normalizedType === 'boolean') { + const boolValue = value === true || value === 1 || value === '1' || value === 'true' + return boolValue ? ( +
+ + Yes +
+ ) : ( +
+ + No +
+ ) + } + + if (normalizedType === 'rating') { + if (typeof value === 'string' && isNaN(parseFloat(value))) { + return ( + + {value} + + ) + } + + const numValue = typeof value === 'number' ? value : parseFloat(value) + if (isNaN(numValue)) return N/A + + const normalizedValue = Math.max(0, Math.min(1, numValue)) + const percentage = Math.round(normalizedValue * 100) + const getBarColor = (pct: number): string => { + if (pct >= 70) return 'bg-green-500' + if (pct >= 50) return 'bg-yellow-500' + return 'bg-red-500' + } + + return ( +
+ {percentage}% +
+
+
+
+ ) + } + + if (normalizedType === 'number') { + const numValue = typeof value === 'number' ? value : parseFloat(value) + if (isNaN(numValue)) return N/A + const info = getMetricInfo(metricName) - if (info?.unit) return `${value.toFixed(2)} ${info.unit}` - return value.toFixed(2) + if (info?.category === 'acoustic' || info?.category === 'ai_voice') { + return ( +
+ {numValue.toFixed(2)} + {info.unit || ''} +
+ ) + } + return {numValue.toFixed(1)} } - return String(value) + + return {String(value)} } // Helper to check if metric has a valid value const hasValidValue = (metric: any): boolean => { - return metric?.value !== null && metric?.value !== undefined && !metric?.error + const val = metric?.value + if (val === null || val === undefined) return false + if (val === '') return false + if (typeof val === 'string' && val.toLowerCase() === 'n/a') return false + if (typeof val === 'string' && val.toLowerCase() === 'na') return false + if (typeof val === 'string' && val.trim() === '') return false + return true } - // Helper to get metric icon based on value quality - const getMetricIcon = (value: any, type: string) => { - if (value === null || value === undefined) return - if (type === 'boolean') return value ? : - if (type === 'rating') { - if (value >= 0.8) return - if (value >= 0.6) return - return - } - return null + const isAudioCategoryMetric = (metricName?: string): boolean => { + if (!metricName) return false + const info = getMetricInfo(metricName) + return info?.category === 'acoustic' || info?.category === 'ai_voice' } const MetricsCard = () => { @@ -300,6 +475,14 @@ export default function TestVoiceAgentResultDetails({ resultData }: TestVoiceAge const info = getMetricInfo(m.metric_name) return info?.category === 'ai_voice' }) + const unavailableAudioMetrics = metrics.filter(([, m]) => { + if (!isAudioCategoryMetric(m?.metric_name)) return false + if (m?.skipped === 'audio_required') return true + if (typeof m?.error === 'string' && m.error.trim().length > 0) return true + return false + }) + const hasAnyAudioMetric = metrics.some(([, m]) => isAudioCategoryMetric(m?.metric_name)) + const hasAudioUnavailableNotice = hasAnyAudioMetric && unavailableAudioMetrics.length > 0 return (
@@ -309,23 +492,33 @@ export default function TestVoiceAgentResultDetails({ resultData }: TestVoiceAge
+ {hasAudioUnavailableNotice && ( +
+ {unavailableAudioMetrics.length === 1 + ? 'An audio metric is unavailable for this run. Audio analysis could not complete for that metric.' + : 'Some audio metrics are unavailable for this run. Audio analysis could not fully complete for one or more metrics.'} +
+ )} + {/* LLM Conversation Metrics */} {llmMetrics.length > 0 && (
- -

Conversation Quality

+ +

Conversation Quality

+ + LLM Evaluation
-
+
{llmMetrics.map(([id, metric]) => ( -
-
-

{metric.metric_name}

- {getMetricIcon(metric.value, metric.type)} +
+
+ {metric.metric_name} +
-

+

{formatMetricValue(metric.value, metric.type, metric.metric_name)} -

+
))}
@@ -338,19 +531,19 @@ export default function TestVoiceAgentResultDetails({ resultData }: TestVoiceAge

AI Voice Quality

+ + ML Analysis
-
+
{aiVoiceMetrics.map(([id, metric]) => ( -
-

{metric.metric_name}

-

+

+
+ {metric.metric_name} + +
+
{formatMetricValue(metric.value, metric.type, metric.metric_name)} -

- {getMetricInfo(metric.metric_name)?.ideal && ( -

- Ideal: {getMetricInfo(metric.metric_name)?.ideal} -

- )} +
))}
@@ -361,21 +554,21 @@ export default function TestVoiceAgentResultDetails({ resultData }: TestVoiceAge {acousticMetrics.length > 0 && (
- -

Acoustic Analysis

+ +

Acoustic Analysis

+ + Signal Analysis
-
+
{acousticMetrics.map(([id, metric]) => ( -
-

{metric.metric_name}

-

+

+
+ {metric.metric_name} + +
+
{formatMetricValue(metric.value, metric.type, metric.metric_name)} -

- {getMetricInfo(metric.metric_name)?.ideal && ( -

- Ideal: {getMetricInfo(metric.metric_name)?.ideal} -

- )} +
))}
diff --git a/frontend/src/components/call-recordings/VapiCallDetails.tsx b/frontend/src/components/call-recordings/VapiCallDetails.tsx index 1a6f00cf..21f90480 100644 --- a/frontend/src/components/call-recordings/VapiCallDetails.tsx +++ b/frontend/src/components/call-recordings/VapiCallDetails.tsx @@ -107,6 +107,7 @@ interface VapiCallDetailsProps { } const COLORS = ['#8b5cf6', '#06b6d4', '#f59e0b', '#ef4444', '#10b981']; +type LatencyStats = NonNullable['latency_stats']> export default function VapiCallDetails({ callData, hideTranscript = false }: VapiCallDetailsProps) { const [activeTab, setActiveTab] = useState<'overview' | 'transcript'>('overview') @@ -123,17 +124,88 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va return new Date(timestamp).toLocaleString() } + const raw = callData as any + const artifact = raw.artifact || {} + const artifactRecording = artifact.recording || {} + const artifactMono = artifactRecording.mono || {} + const rawCostBreakdown = raw.costBreakdown || {} + const rawAnalysis = raw.analysis || {} + const perf = artifact.performanceMetrics || {} + + const callId = callData.call_id || raw.id + const assistantId = callData.assistant_id || raw.assistantId + const startTimestamp = callData.start_timestamp || raw.startedAt + const endTimestamp = callData.end_timestamp || raw.endedAt + const callStatus = callData.call_status || raw.status + const callType = callData.call_type || raw.type + const endedReason = callData.ended_reason || raw.endedReason || raw.reason + + const computedDurationSeconds = (() => { + if (typeof callData.duration_seconds === 'number') return callData.duration_seconds + if (startTimestamp && endTimestamp) { + const startMs = Date.parse(startTimestamp) + const endMs = Date.parse(endTimestamp) + if (!Number.isNaN(startMs) && !Number.isNaN(endMs) && endMs >= startMs) { + return (endMs - startMs) / 1000 + } + } + return undefined + })() + + const recordingLinks = { + combined: + callData.recording_urls?.combined_url || + raw.recordingUrl || + artifact.recordingUrl || + artifactMono.combinedUrl, + stereo: + callData.recording_urls?.stereo_url || + raw.stereoRecordingUrl || + artifact.stereoRecordingUrl || + artifactRecording.stereoUrl, + assistant: callData.recording_urls?.assistant_url || artifactMono.assistantUrl, + customer: callData.recording_urls?.customer_url || artifactMono.customerUrl, + } + // Get recording URL - prefer combined, then stereo - const recordingUrl = callData.recording_urls?.combined_url || - callData.recording_urls?.stereo_url || - callData.raw_data?.recordingUrl || - callData.raw_data?.stereoRecordingUrl - - // Prepare transcript for display - const transcriptEntries = callData.transcript_object || [] - + const recordingUrl = recordingLinks.combined || recordingLinks.stereo + + // Prepare transcript for display (prefer normalized transcript_object if present) + const transcriptEntries: VapiTranscriptEntry[] = + (callData.transcript_object && callData.transcript_object.length > 0 + ? callData.transcript_object + : (raw.messages || artifact.messages || []) + .filter((msg: any) => msg?.role !== 'system') + .map((msg: any) => ({ + role: msg.role === 'bot' ? 'agent' : msg.role, + content: msg.message || msg.content || '', + seconds_from_start: msg.secondsFromStart, + duration_ms: msg.duration, + end_time_ms: msg.endTime, + time_ms: msg.time, + words: msg.metadata?.wordLevelConfidence, + })) + .filter((entry: VapiTranscriptEntry) => !!entry.content)) + // Prepare latency data for chart - const latencyStats = callData.analysis?.latency_stats + const latencyStats: LatencyStats = callData.analysis?.latency_stats || { + model_latency_avg: perf.modelLatencyAverage, + voice_latency_avg: perf.voiceLatencyAverage, + transcriber_latency_avg: perf.transcriberLatencyAverage, + endpointing_latency_avg: perf.endpointingLatencyAverage, + turn_latency_avg: perf.turnLatencyAverage, + from_transport_latency_avg: perf.fromTransportLatencyAverage, + to_transport_latency_avg: perf.toTransportLatencyAverage, + num_assistant_interrupted: perf.numAssistantInterrupted, + turn_latencies: perf.turnLatencies, + p50: undefined, + p90: undefined, + p95: undefined, + p99: undefined, + max: undefined, + min: undefined, + num_turns: undefined, + } const latencyData = [ { name: 'Model', avg: latencyStats?.model_latency_avg, p50: latencyStats?.p50 }, { name: 'Voice', avg: latencyStats?.voice_latency_avg }, @@ -143,7 +215,7 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va ].filter(item => item.avg !== undefined && item.avg !== null) // Prepare cost data for pie chart - const costBreakdown = callData.cost_breakdown + const costBreakdown = callData.cost_breakdown || rawCostBreakdown const costData = [ { name: 'Transport', value: costBreakdown?.transport || 0 }, { name: 'STT', value: costBreakdown?.stt || 0 }, @@ -153,8 +225,9 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va ].filter(item => item.value > 0) const SummaryCard = () => { - const analysis = callData.analysis - const successEval = analysis?.success_evaluation + const summary = rawAnalysis.summary || raw.summary || callData.analysis?.summary + const successEval = rawAnalysis.successEvaluation ?? callData.analysis?.success_evaluation + const interruptionCount = perf.numAssistantInterrupted ?? callData.analysis?.interruption_count ?? 0 const isSuccessful = successEval === true || successEval === 'true' return ( @@ -164,13 +237,13 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va Call Analysis - {analysis ? ( + {summary || successEval !== undefined ? (
- {analysis.summary && ( + {summary && (

Summary

- {analysis.summary} + {summary}

)} @@ -200,21 +273,21 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va

Interruptions

- {analysis.interruption_count ?? 0} + {interruptionCount}

Ended Reason

- {callData.ended_reason?.replace(/-/g, ' ') || 'Normal'} + {endedReason?.replace(/-/g, ' ') || 'Normal'}

Call Type

- {callData.call_type || 'Web Call'} + {callType || 'Web Call'}
@@ -272,11 +345,11 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va
) }) - ) : callData.transcript ? ( + ) : (callData.transcript || raw.transcript) ? ( // Fallback to plain transcript if no transcript_object 
-              {callData.transcript}
+              {callData.transcript || raw.transcript}
) : ( @@ -299,10 +372,10 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va

Total Cost

- ${callData.cost?.toFixed(4) || costBreakdown?.total?.toFixed(4) || '0.0000'} + ${(callData.cost ?? raw.cost)?.toFixed?.(4) || costBreakdown?.total?.toFixed(4) || '0.0000'}

- Duration: {formatDuration(callData.duration_seconds)} + Duration: {formatDuration(computedDurationSeconds)}

@@ -338,26 +411,26 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va )}
{/* Token Usage */} - {(costBreakdown?.llm_prompt_tokens || costBreakdown?.tts_characters) && ( + {(costBreakdown?.llm_prompt_tokens || costBreakdown?.llmPromptTokens || costBreakdown?.tts_characters || costBreakdown?.ttsCharacters) && (

Usage

- {costBreakdown.llm_prompt_tokens && ( + {(costBreakdown.llm_prompt_tokens || costBreakdown.llmPromptTokens) && (
LLM Prompt Tokens - {costBreakdown.llm_prompt_tokens.toLocaleString()} + {(costBreakdown.llm_prompt_tokens || costBreakdown.llmPromptTokens).toLocaleString()}
)} - {costBreakdown.llm_completion_tokens && ( + {(costBreakdown.llm_completion_tokens || costBreakdown.llmCompletionTokens) && (
LLM Completion Tokens - {costBreakdown.llm_completion_tokens.toLocaleString()} + {(costBreakdown.llm_completion_tokens || costBreakdown.llmCompletionTokens).toLocaleString()}
)} - {costBreakdown.tts_characters && ( + {(costBreakdown.tts_characters || costBreakdown.ttsCharacters) && (
TTS Characters - {costBreakdown.tts_characters.toLocaleString()} + {(costBreakdown.tts_characters || costBreakdown.ttsCharacters).toLocaleString()}
)}
@@ -472,38 +545,38 @@ export default function VapiCallDetails({ callData, hideTranscript = false }: Va

Call ID

-

{callData.call_id}

+

{callId}

Assistant ID

-

{callData.assistant_id}

+

{assistantId}

Start Time

-

{formatTimestamp(callData.start_timestamp)}

+

{formatTimestamp(startTimestamp)}

End Time

-

{formatTimestamp(callData.end_timestamp)}

+

{formatTimestamp(endTimestamp)}

Duration

-

{formatDuration(callData.duration_seconds)}

+

{formatDuration(computedDurationSeconds)}

Status

-

{callData.call_status}

+

{callStatus}

{/* Recording URLs */} - {callData.recording_urls && Object.values(callData.recording_urls).some(Boolean) && ( + {Object.values(recordingLinks).some(Boolean) && (

Recordings

- {callData.recording_urls.combined_url && ( + {recordingLinks.combined && ( )} - {callData.recording_urls.stereo_url && ( + {recordingLinks.stereo && ( )} - {callData.recording_urls.assistant_url && ( + {recordingLinks.assistant && ( )} - {callData.recording_urls.customer_url && ( + {recordingLinks.customer && ( { return true } +const isAudioCategoryMetric = (metricName?: string): boolean => { + if (!metricName) return false + const info = getMetricInfo(metricName) + return info?.category === 'acoustic' || info?.category === 'ai_voice' +} + function EvaluationStepper({ status }: { status: string }) { const [dots, setDots] = useState('') @@ -603,9 +609,30 @@ export default function CallRecordingDetail() { return !info || info.category === 'llm' } ) + const unavailableAudioMetrics = Object.entries(metricScores).filter( + ([, metric]: [string, any]) => { + const metricName = metric.metric_name || '' + if (!isAudioCategoryMetric(metricName)) return false + if (metric.skipped === 'audio_required') return true + if (typeof metric.error === 'string' && metric.error.trim().length > 0) return true + return false + } + ) + const hasAnyAudioMetric = Object.entries(metricScores).some( + ([, metric]: [string, any]) => isAudioCategoryMetric(metric.metric_name || '') + ) + const hasAudioUnavailableNotice = hasAnyAudioMetric && unavailableAudioMetrics.length > 0 return (
+ {hasAudioUnavailableNotice && ( +
+ {unavailableAudioMetrics.length === 1 + ? 'An audio metric is unavailable for this run. We retried fetching call audio from the provider, but audio analysis could not complete.' + : 'Some audio metrics are unavailable for this run. We retried fetching call audio from the provider, but audio analysis could not fully complete.'} +
+ )} + {/* AI Voice Quality Metrics */} {aiVoiceMetrics.length > 0 && (
diff --git a/pyproject.toml b/pyproject.toml index de34701a..25271249 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,7 +38,7 @@ dependencies = [ "croniter>=2.0.0", "pytz>=2024.1", "jiwer>=3.0.0", - "litellm>=1.60.0", + "litellm>=1.60.0,<=1.82.6", "jinja2>=3.1.0", "wait-for2>=0.4.0", "docstring-parser>=0.15", @@ -93,6 +93,7 @@ qualitative-voice = [ "torchaudio>=2.1.0", "transformers>=4.41.0", "librosa>=0.10.0", + "imageio-ffmpeg>=0.5.1", "scipy>=1.11.0", "pyannote.audio>=3.1.0", "praat-parselmouth>=0.4.3", diff --git a/schema_er_diagram.png b/schema_er_diagram.png index 769c3623..b571447e 100644 Binary files a/schema_er_diagram.png and b/schema_er_diagram.png differ