1800 lines
67 KiB
Python
1800 lines
67 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
from services.ai_voice_runtime_service.audiosocket import normalize_media_uuid
|
|
from services.ai_voice_runtime_service.media_runtime import AudioSocketMediaRuntime, MediaRegistration
|
|
from services.ai_voice_runtime_service.providers.asr import build_asr_provider, build_streaming_asr_provider
|
|
from services.ai_voice_runtime_service.runtime_tts_provider import RuntimeConfiguredTTSProvider
|
|
from services.ai_orchestrator_service import operator_persona as persona
|
|
from services.shared.ai_operator_config import (
|
|
load_effective_ai_operator_config,
|
|
sync_ai_operator_config_from_code,
|
|
)
|
|
from services.shared.core import Role, new_id, utc_now_iso
|
|
from services.shared.db import get_session
|
|
from services.shared.models import (
|
|
HealthResponse,
|
|
VoiceAIHandoffRequestIn,
|
|
VoiceAIMediaBridgeEventIn,
|
|
VoiceAISessionCreateIn,
|
|
VoiceAISessionOut,
|
|
VoiceAIStartIn,
|
|
VoiceAITelephonyEventIn,
|
|
VoiceAITurnDecisionOut,
|
|
VoiceAITurnIn,
|
|
)
|
|
from services.shared.security import issue_app_token, require_roles
|
|
from services.shared.sql_init import init_sql_schema
|
|
from services.shared.sql_models import AsteriskCallLinkRow, VoiceAISessionRow, VoiceTranscriptSegmentRow
|
|
from services.shared.voice_transcripts import add_transcript_segment, next_transcript_sequence
|
|
|
|
init_sql_schema()
|
|
|
|
|
|
LOGGER = logging.getLogger("uvicorn.error")
|
|
_VOICE_START_THREADS_LOCK = threading.Lock()
|
|
_VOICE_START_THREADS: set[threading.Thread] = set()
|
|
|
|
|
|
def _bool_env(name: str, default: bool) -> bool:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
return raw.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _int_env(name: str, default: int) -> int:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return int(raw.strip())
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def _float_env(name: str, default: float) -> float:
|
|
raw = os.getenv(name)
|
|
if raw is None:
|
|
return default
|
|
try:
|
|
return float(raw.strip())
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def _ai_orchestrator_service_url() -> str:
|
|
return os.getenv("AI_ORCHESTRATOR_SERVICE_URL", "http://localhost:8017").rstrip("/")
|
|
|
|
|
|
def _asterisk_bridge_service_url() -> str:
|
|
return os.getenv("ASTERISK_BRIDGE_SERVICE_URL", "http://localhost:8016").rstrip("/")
|
|
|
|
|
|
def _interaction_service_url() -> str:
|
|
return os.getenv("INTERACTION_SERVICE_URL", "http://localhost:8004").rstrip("/")
|
|
|
|
|
|
def _sales_service_url() -> str:
|
|
return os.getenv("SALES_SERVICE_URL", "http://localhost:8020").rstrip("/")
|
|
|
|
|
|
def _sales_voice_sync_tenant_id() -> str | None:
|
|
return os.getenv("SALES_VOICE_SYNC_TENANT_ID", "").strip() or None
|
|
|
|
|
|
def _asr_provider_name() -> str:
|
|
return os.getenv("AI_VOICE_ASR_PROVIDER", "yandex").strip() or "yandex"
|
|
|
|
|
|
def _tts_provider_name() -> str:
|
|
return os.getenv("AI_VOICE_TTS_PROVIDER", "yandex").strip() or "yandex"
|
|
|
|
|
|
def _handoff_timeout_seconds() -> float:
|
|
return max(3.0, _float_env("AI_VOICE_HANDOFF_TIMEOUT_SECONDS", 8.0))
|
|
|
|
|
|
def _audiosocket_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_AUDIOSOCKET_ENABLED", False)
|
|
|
|
|
|
def _audiosocket_host() -> str:
|
|
return os.getenv("AI_VOICE_AUDIOSOCKET_HOST", "0.0.0.0").strip() or "0.0.0.0"
|
|
|
|
|
|
def _audiosocket_port() -> int:
|
|
return max(_int_env("AI_VOICE_AUDIOSOCKET_PORT", 9019), 1)
|
|
|
|
|
|
def _vad_frame_ms() -> int:
|
|
return 20
|
|
|
|
|
|
def _vad_min_speech_ms() -> int:
|
|
return max(_int_env("AI_VOICE_VAD_MIN_SPEECH_MS", 300), _vad_frame_ms())
|
|
|
|
|
|
def _vad_trailing_silence_ms() -> int:
|
|
return max(_int_env("AI_VOICE_VAD_TRAILING_SILENCE_MS", 400), _vad_frame_ms())
|
|
|
|
|
|
def _turn_max_ms() -> int:
|
|
return max(_int_env("AI_VOICE_TURN_MAX_MS", 10000), _vad_frame_ms())
|
|
|
|
|
|
def _vad_rms_threshold() -> int:
|
|
return max(_int_env("AI_VOICE_VAD_RMS_THRESHOLD", 250), 1)
|
|
|
|
|
|
def _media_idle_timeout_seconds() -> float:
|
|
return max(_float_env("AI_VOICE_MEDIA_IDLE_TIMEOUT_SECONDS", 15.0), 5.0)
|
|
|
|
|
|
def _media_registration_wait_timeout_seconds() -> float:
|
|
# `call.started` still performs cross-service writes before the AudioSocket
|
|
# registration event can be processed, so keep the media socket open long
|
|
# enough for the registration to arrive.
|
|
return max(_float_env("AI_VOICE_MEDIA_REGISTRATION_WAIT_TIMEOUT_SECONDS", 12.0), 0.0)
|
|
|
|
|
|
def _voice_v2_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_V2_ENABLED", False)
|
|
|
|
|
|
def _voice_v2_queue_codes() -> set[str]:
|
|
raw = str(os.getenv("AI_VOICE_V2_QUEUE_CODES", "voice_lab_ai") or "voice_lab_ai").strip()
|
|
return {item.strip() for item in raw.split(",") if item.strip()}
|
|
|
|
|
|
def _voice_v2_ack_mode() -> str:
|
|
return str(os.getenv("AI_VOICE_V2_ACK_MODE", "immediate_short") or "immediate_short").strip().lower()
|
|
|
|
|
|
def _voice_v2_streaming_tts_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_V2_STREAMING_TTS", True)
|
|
|
|
|
|
def _voice_v2_partial_asr_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_V2_PARTIAL_ASR", True)
|
|
|
|
|
|
def _voice_v2_duplex_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_V2_DUPLEX_ENABLED", True)
|
|
|
|
|
|
def _voice_v2_streaming_asr_backend() -> str:
|
|
return str(os.getenv("AI_VOICE_V2_STREAMING_ASR_BACKEND", "elevenlabs_realtime") or "elevenlabs_realtime").strip().lower()
|
|
|
|
|
|
def _voice_v2_prebaked_ack_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_V2_PREBAKED_ACK_ENABLED", True)
|
|
|
|
|
|
def _voice_v2_emotive_ack_enabled() -> bool:
|
|
return _bool_env("AI_VOICE_V2_EMOTIVE_ACK_ENABLED", True)
|
|
|
|
|
|
def _voice_v2_emotive_ack_ru_only() -> bool:
|
|
return _bool_env("AI_VOICE_V2_EMOTIVE_ACK_RU_ONLY", True)
|
|
|
|
|
|
def _service_headers(*, tenant_id: str | None = None) -> dict[str, str]:
|
|
token = issue_app_token(
|
|
subject="svc:ai-voice-runtime",
|
|
username="ai-voice-runtime",
|
|
role="admin",
|
|
auth_source="service",
|
|
provider="ai-voice-runtime",
|
|
tenant_id=tenant_id,
|
|
ttl_seconds=300,
|
|
)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _is_sqlite_locked(exc: Exception) -> bool:
|
|
return "database is locked" in str(exc).lower()
|
|
|
|
|
|
def _retry_db_write(fn, *, attempts: int = 6, base_delay_seconds: float = 0.05):
|
|
last_exc: Exception | None = None
|
|
for attempt in range(attempts):
|
|
try:
|
|
return fn()
|
|
except OperationalError as exc:
|
|
last_exc = exc
|
|
if not _is_sqlite_locked(exc) or attempt >= attempts - 1:
|
|
raise
|
|
time.sleep(base_delay_seconds * (attempt + 1))
|
|
if last_exc is not None:
|
|
raise last_exc
|
|
raise RuntimeError("DB write retry exhausted without exception")
|
|
|
|
|
|
def _request(
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
payload: dict[str, Any] | None = None,
|
|
timeout: float = 10.0,
|
|
tenant_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
with httpx.Client(timeout=timeout) as client:
|
|
response = client.request(
|
|
method,
|
|
url,
|
|
json=payload,
|
|
headers=_service_headers(tenant_id=tenant_id),
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _orchestrator_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 10.0) -> dict[str, Any]:
|
|
return _request(method, f"{_ai_orchestrator_service_url()}{path}", payload=payload, timeout=timeout)
|
|
|
|
|
|
def _register_voice_start_thread(thread: threading.Thread) -> None:
|
|
with _VOICE_START_THREADS_LOCK:
|
|
_VOICE_START_THREADS.add(thread)
|
|
|
|
|
|
def _unregister_voice_start_thread(thread: threading.Thread) -> None:
|
|
with _VOICE_START_THREADS_LOCK:
|
|
_VOICE_START_THREADS.discard(thread)
|
|
|
|
|
|
def _run_voice_start_thread(session_id: str, start_payload: VoiceAIStartIn) -> None:
|
|
current = threading.current_thread()
|
|
try:
|
|
_complete_voice_ai_session_start(session_id, start_payload)
|
|
finally:
|
|
_unregister_voice_start_thread(current)
|
|
|
|
|
|
def _wait_for_background_voice_start_threads(timeout_seconds: float = 5.0) -> None:
|
|
deadline = time.monotonic() + max(timeout_seconds, 0.0)
|
|
while True:
|
|
with _VOICE_START_THREADS_LOCK:
|
|
threads = [thread for thread in _VOICE_START_THREADS if thread.is_alive()]
|
|
if not threads:
|
|
return
|
|
remaining = max(deadline - time.monotonic(), 0.0)
|
|
if remaining <= 0:
|
|
return
|
|
for thread in threads:
|
|
thread.join(timeout=min(0.25, remaining))
|
|
|
|
|
|
def _bridge_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 10.0) -> dict[str, Any]:
|
|
return _request(method, f"{_asterisk_bridge_service_url()}{path}", payload=payload, timeout=timeout)
|
|
|
|
|
|
def _interaction_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 5.0) -> dict[str, Any]:
|
|
return _request(method, f"{_interaction_service_url()}{path}", payload=payload, timeout=timeout)
|
|
|
|
|
|
def _sales_request(method: str, path: str, *, payload: dict[str, Any] | None = None, timeout: float = 8.0) -> dict[str, Any]:
|
|
return _request(
|
|
method,
|
|
f"{_sales_service_url()}{path}",
|
|
payload=payload,
|
|
timeout=timeout,
|
|
tenant_id=_sales_voice_sync_tenant_id(),
|
|
)
|
|
|
|
|
|
def _resolved_voice_language(language_hint: str | None, fallback: str | None = None) -> str:
|
|
return str(language_hint or fallback or "ru").strip() or "ru"
|
|
|
|
|
|
def _voice_start_name_prompt(language: str | None) -> str:
|
|
if str(language or "").strip() == "kz":
|
|
return "Сәлеметсіз бе. Атыңызды атаңызшы."
|
|
return "Здравствуйте. Назовите, пожалуйста, ваше имя."
|
|
|
|
|
|
def _default_voice_greeting(language: str | None, *, agent_profile: str = "voice_support", operator_config: Any | None = None) -> str:
|
|
if str(agent_profile or "").strip() == "voice_start":
|
|
return _voice_start_name_prompt(language)
|
|
return persona.voice_greeting(str(language or "ru").strip() or "ru", operator_config)
|
|
|
|
|
|
_ASR_PROVIDER = build_asr_provider(_asr_provider_name())
|
|
_STREAMING_ASR_PROVIDER = build_streaming_asr_provider(_voice_v2_streaming_asr_backend())
|
|
_TTS_PROVIDER = RuntimeConfiguredTTSProvider(default_provider_name=_tts_provider_name())
|
|
|
|
|
|
def _current_tts_provider_name() -> str:
|
|
return _TTS_PROVIDER.current_provider_name()
|
|
|
|
|
|
def _load_voice_session(session, session_id: str) -> VoiceAISessionRow:
|
|
row = session.execute(
|
|
select(VoiceAISessionRow).where(VoiceAISessionRow.session_id == session_id)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="Voice AI session not found")
|
|
return row
|
|
|
|
|
|
def _load_call_link(session, call_id: str) -> AsteriskCallLinkRow | None:
|
|
return session.execute(
|
|
select(AsteriskCallLinkRow).where(AsteriskCallLinkRow.call_id == call_id)
|
|
).scalar_one_or_none()
|
|
|
|
|
|
def _latest_caller_transcript_text(session, session_id: str) -> str | None:
|
|
row = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "caller")
|
|
.where(VoiceTranscriptSegmentRow.is_final.is_(True))
|
|
.order_by(VoiceTranscriptSegmentRow.sequence_no.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
return None
|
|
text = str(row.text or "").strip()
|
|
return text or None
|
|
|
|
|
|
def _sales_transcript_snapshot(session, session_id: str, *, max_chars: int = 6000) -> str | None:
|
|
rows = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.is_final.is_(True))
|
|
.order_by(VoiceTranscriptSegmentRow.sequence_no.asc())
|
|
).scalars().all()
|
|
lines: list[str] = []
|
|
for row in rows:
|
|
text = str(row.text or "").strip()
|
|
if not text:
|
|
continue
|
|
speaker = "customer" if str(row.speaker or "").strip() == "caller" else str(row.speaker or "assistant").strip()
|
|
lines.append(f"{speaker}: {text}")
|
|
if not lines:
|
|
return None
|
|
selected: list[str] = []
|
|
total = 0
|
|
for line in reversed(lines):
|
|
line_len = len(line) + (1 if selected else 0)
|
|
if selected and total + line_len > max_chars:
|
|
break
|
|
if not selected and len(line) > max_chars:
|
|
selected.append(line[-max_chars:])
|
|
total = len(selected[0])
|
|
break
|
|
selected.append(line)
|
|
total += line_len
|
|
return "\n".join(reversed(selected)) or None
|
|
|
|
|
|
def _sync_sales_voice_session(
|
|
session_id: str,
|
|
*,
|
|
caller_number: str | None = None,
|
|
caller_name: str | None = None,
|
|
queue_code: str | None = None,
|
|
summary: str | None = None,
|
|
include_transcript: bool = False,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
link = _load_call_link(session, voice_session.call_id)
|
|
resolved_summary = (
|
|
str(summary or "").strip()
|
|
or _latest_caller_transcript_text(session, session_id)
|
|
or str(voice_session.handoff_reason or "").strip()
|
|
or None
|
|
)
|
|
transcript_text = _sales_transcript_snapshot(session, session_id) if include_transcript else None
|
|
payload = {
|
|
"call_id": voice_session.call_id,
|
|
"interaction_id": str(voice_session.interaction_id or getattr(link, "interaction_id", "") or "").strip() or None,
|
|
"queue_id": str(voice_session.queue_id or getattr(link, "queue_id", "") or "").strip() or None,
|
|
"queue_code": str(getattr(link, "queue_code", "") or queue_code or "").strip() or None,
|
|
"caller_number": str(getattr(link, "caller_number", "") or caller_number or "").strip() or None,
|
|
"caller_name": str(getattr(link, "caller_name", "") or caller_name or "").strip() or None,
|
|
"voice_session_id": voice_session.session_id,
|
|
"ai_session_id": voice_session.ai_session_id,
|
|
"ai_state": voice_session.status,
|
|
"handoff_reason": voice_session.handoff_reason,
|
|
"telephony_status": str(getattr(link, "telephony_status", "") or voice_session.media_status or "").strip() or None,
|
|
"call_status": str(getattr(link, "status", "") or voice_session.status or "").strip() or None,
|
|
"started_at": voice_session.started_at,
|
|
"ended_at": voice_session.ended_at,
|
|
"summary": resolved_summary,
|
|
"transcript_text": transcript_text,
|
|
"metadata": {
|
|
"linked_id": voice_session.linked_id,
|
|
"agent_profile": voice_session.agent_profile,
|
|
"language": voice_session.language,
|
|
"asr_provider": voice_session.asr_provider,
|
|
"tts_provider": voice_session.tts_provider,
|
|
"handoff_target_queue_id": voice_session.handoff_target_queue_id,
|
|
"media_status": voice_session.media_status,
|
|
"media_uuid": voice_session.media_uuid,
|
|
"media_connected_at": voice_session.media_connected_at,
|
|
"media_ended_at": voice_session.media_ended_at,
|
|
"last_media_frame_at": voice_session.last_media_frame_at,
|
|
"last_user_utterance_at": voice_session.last_user_utterance_at,
|
|
"last_ai_reply_at": voice_session.last_ai_reply_at,
|
|
"voice_start_language": voice_session.voice_start_language,
|
|
"customer_name_status": voice_session.customer_name_status,
|
|
"customer_name_value": voice_session.customer_name_value,
|
|
"customer_name_source": voice_session.customer_name_source,
|
|
"customer_name_resolved_at": voice_session.customer_name_resolved_at,
|
|
"claimed_by_user": getattr(link, "claimed_by_user", None),
|
|
"claimed_at": getattr(link, "claimed_at", None),
|
|
"operator_extension": getattr(link, "operator_extension", None),
|
|
"channel_name": getattr(link, "channel_name", None),
|
|
**(metadata or {}),
|
|
},
|
|
}
|
|
finally:
|
|
session.close()
|
|
|
|
try:
|
|
_sales_request("POST", "/internal/sales-sync/voice", payload=payload, timeout=8.0)
|
|
except Exception:
|
|
LOGGER.exception(
|
|
"voice_runtime.sales_sync_failed session_id=%s call_id=%s ai_state=%s",
|
|
payload.get("voice_session_id"),
|
|
payload.get("call_id"),
|
|
payload.get("ai_state"),
|
|
)
|
|
|
|
|
|
def _next_sequence(session, session_id: str) -> int:
|
|
return next_transcript_sequence(session, session_id)
|
|
|
|
|
|
def _record_segment(
|
|
session,
|
|
*,
|
|
voice_session: VoiceAISessionRow,
|
|
speaker: str,
|
|
source_type: str,
|
|
text: str,
|
|
sequence_no: int,
|
|
confidence: float | None = None,
|
|
payload: dict[str, Any] | None = None,
|
|
is_final: bool = True,
|
|
) -> None:
|
|
add_transcript_segment(
|
|
session,
|
|
session_id=voice_session.session_id,
|
|
call_id=voice_session.call_id,
|
|
interaction_id=voice_session.interaction_id,
|
|
speaker=speaker,
|
|
source_type=source_type,
|
|
text=text,
|
|
confidence=confidence,
|
|
payload=payload,
|
|
is_final=is_final,
|
|
sequence_no=sequence_no,
|
|
)
|
|
|
|
|
|
def _normalize_voice_text_key(text: str | None) -> str:
|
|
compact = str(text or "").strip().lower()
|
|
for char in ",.!?;:\"'()[]{}":
|
|
compact = compact.replace(char, " ")
|
|
return " ".join(compact.split())
|
|
|
|
|
|
def _is_low_signal_voice_text(text: str | None) -> bool:
|
|
normalized = _normalize_voice_text_key(text)
|
|
if not normalized:
|
|
return True
|
|
return normalized in {
|
|
"\u0430\u043b\u043b\u043e",
|
|
"\u0430\u0433\u0430",
|
|
"\u0434\u0430",
|
|
"\u0434\u043e\u0431\u0440\u044b\u0439 \u0434\u0435\u043d\u044c",
|
|
"\u0437\u0434\u0440\u0430\u0432\u0441\u0442\u0432\u0443\u0439\u0442\u0435",
|
|
"\u043b\u0430\u0434\u043d\u043e",
|
|
"\u043d\u0435\u0442",
|
|
"\u043d\u0435\u0430",
|
|
"\u043e\u0439",
|
|
"\u043e\u043a",
|
|
"\u043f\u0440\u0438\u0432\u0435\u0442",
|
|
"\u0441\u043b\u044b\u0448\u043d\u043e",
|
|
"\u0441\u043b\u044b\u0448\u0443",
|
|
"\u0443\u0433\u0443",
|
|
"\u0445\u043e\u0440\u043e\u0448\u043e",
|
|
"\u044f\u0441\u043d\u043e",
|
|
}
|
|
|
|
|
|
def _is_intent_bearing_turn(transcript_text: str | None, intent: str | None) -> bool:
|
|
normalized_intent = str(intent or "").strip().lower()
|
|
if normalized_intent in {
|
|
"schedule",
|
|
"address",
|
|
"price",
|
|
"status",
|
|
"problem",
|
|
"operator_request",
|
|
"handoff_request",
|
|
"kb_answer",
|
|
}:
|
|
return True
|
|
normalized_text = _normalize_voice_text_key(transcript_text)
|
|
if not normalized_text or _is_low_signal_voice_text(normalized_text):
|
|
return False
|
|
markers = (
|
|
"\u0433\u0440\u0430\u0444\u0438\u043a",
|
|
"\u0432\u0440\u0435\u043c\u044f \u0440\u0430\u0431\u043e\u0442\u044b",
|
|
"\u0440\u0435\u0436\u0438\u043c \u0440\u0430\u0431\u043e\u0442\u044b",
|
|
"\u0430\u0434\u0440\u0435\u0441",
|
|
"\u0444\u0438\u043b\u0438\u0430\u043b",
|
|
"\u0433\u043e\u0440\u043e\u0434",
|
|
"\u0437\u0430\u044f\u0432\u043a",
|
|
"\u0441\u0442\u0430\u0442\u0443\u0441",
|
|
"\u0442\u0430\u0440\u0438\u0444",
|
|
"\u0446\u0435\u043d\u0430",
|
|
"\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u044c",
|
|
"\u043e\u043f\u0435\u0440\u0430\u0442\u043e\u0440",
|
|
"\u043c\u0435\u043d\u0435\u0434\u0436\u0435\u0440",
|
|
"\u043d\u0435 \u0440\u0430\u0431\u043e\u0442\u0430\u0435\u0442",
|
|
"\u043e\u0448\u0438\u0431\u043a",
|
|
"\u043f\u0440\u043e\u0431\u043b\u0435\u043c",
|
|
)
|
|
return any(marker in normalized_text for marker in markers)
|
|
|
|
|
|
def _annotate_latest_caller_segment(
|
|
session,
|
|
*,
|
|
session_id: str,
|
|
provider_name: str,
|
|
transcript_text: str,
|
|
intent: str | None,
|
|
metadata: dict[str, Any] | None,
|
|
) -> None:
|
|
row = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "caller")
|
|
.order_by(VoiceTranscriptSegmentRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
return
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
payload["provider"] = str(provider_name or "").strip() or payload.get("provider")
|
|
payload["intent_bearing"] = _is_intent_bearing_turn(transcript_text, intent)
|
|
if intent:
|
|
payload["decision_intent"] = str(intent).strip()
|
|
if isinstance(metadata, dict) and metadata:
|
|
merged_metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
|
|
merged_metadata.update(metadata)
|
|
payload["metadata"] = merged_metadata
|
|
row.payload_json = json.dumps(payload, ensure_ascii=False)
|
|
|
|
|
|
def _load_greeting_segment(session, session_id: str) -> VoiceTranscriptSegmentRow | None:
|
|
rows = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "assistant")
|
|
.order_by(VoiceTranscriptSegmentRow.id.asc())
|
|
).scalars().all()
|
|
for row in rows:
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
if str(payload.get("kind") or "").strip() == "greeting":
|
|
return row
|
|
return None
|
|
|
|
|
|
def _load_latest_greeting_segment(session, session_id: str) -> VoiceTranscriptSegmentRow | None:
|
|
rows = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "assistant")
|
|
.order_by(VoiceTranscriptSegmentRow.id.desc())
|
|
).scalars().all()
|
|
for row in rows:
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
if str(payload.get("kind") or "").strip() == "greeting":
|
|
return row
|
|
return None
|
|
|
|
|
|
def _ensure_greeting_segment(session, voice_session: VoiceAISessionRow, greeting_text: str) -> None:
|
|
if not str(greeting_text or "").strip():
|
|
return
|
|
if _load_greeting_segment(session, voice_session.session_id) is not None:
|
|
return
|
|
_record_segment(
|
|
session,
|
|
voice_session=voice_session,
|
|
speaker="assistant",
|
|
source_type="tts",
|
|
text=greeting_text,
|
|
sequence_no=_next_sequence(session, voice_session.session_id),
|
|
payload={
|
|
"kind": "greeting",
|
|
"provider": _current_tts_provider_name(),
|
|
"delivery_status": "planned",
|
|
"delivery_state": "planned",
|
|
},
|
|
is_final=False,
|
|
)
|
|
|
|
|
|
def _upsert_greeting_segment(session, voice_session: VoiceAISessionRow, greeting_text: str) -> None:
|
|
normalized = str(greeting_text or "").strip()
|
|
if not normalized:
|
|
return
|
|
row = _load_latest_greeting_segment(session, voice_session.session_id)
|
|
if row is None:
|
|
_ensure_greeting_segment(session, voice_session, normalized)
|
|
return
|
|
row.text = normalized
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
payload["kind"] = "greeting"
|
|
payload.setdefault("delivery_status", "planned")
|
|
payload.setdefault("delivery_state", str(payload.get("delivery_status") or "planned"))
|
|
payload.setdefault("provider", _current_tts_provider_name())
|
|
row.payload_json = json.dumps(payload, ensure_ascii=False)
|
|
row.is_final = False
|
|
|
|
|
|
def _queue_new_greeting_segment(session, voice_session: VoiceAISessionRow, greeting_text: str) -> None:
|
|
normalized = str(greeting_text or "").strip()
|
|
if not normalized:
|
|
return
|
|
_record_segment(
|
|
session,
|
|
voice_session=voice_session,
|
|
speaker="assistant",
|
|
source_type="tts",
|
|
text=normalized,
|
|
sequence_no=_next_sequence(session, voice_session.session_id),
|
|
payload={
|
|
"kind": "greeting",
|
|
"provider": _current_tts_provider_name(),
|
|
"delivery_status": "planned",
|
|
"delivery_state": "planned",
|
|
},
|
|
is_final=False,
|
|
)
|
|
|
|
|
|
def _voice_start_metadata(payload: dict[str, Any] | None) -> dict[str, str | None]:
|
|
metadata = payload if isinstance(payload, dict) else {}
|
|
return {
|
|
"voice_start_language": str(
|
|
metadata.get("voice_start_language") or metadata.get("language") or ""
|
|
).strip()
|
|
or None,
|
|
"customer_name_status": str(metadata.get("customer_name_status") or "").strip() or None,
|
|
"customer_name_value": str(metadata.get("customer_name_value") or "").strip() or None,
|
|
"customer_name_source": str(metadata.get("customer_name_source") or "").strip() or None,
|
|
"customer_name_resolved_at": str(metadata.get("customer_name_resolved_at") or "").strip() or None,
|
|
"downstream_queue_code": str(metadata.get("downstream_queue_code") or "").strip() or None,
|
|
}
|
|
|
|
|
|
def _apply_voice_start_metadata(voice_session: VoiceAISessionRow, metadata: dict[str, Any] | None) -> None:
|
|
payload = metadata if isinstance(metadata, dict) else {}
|
|
values = _voice_start_metadata(payload)
|
|
if "voice_start_language" in payload or "language" in payload:
|
|
voice_session.voice_start_language = values["voice_start_language"]
|
|
if "customer_name_status" in payload:
|
|
voice_session.customer_name_status = values["customer_name_status"]
|
|
if "customer_name_value" in payload:
|
|
voice_session.customer_name_value = values["customer_name_value"]
|
|
if "customer_name_source" in payload:
|
|
voice_session.customer_name_source = values["customer_name_source"]
|
|
if "customer_name_resolved_at" in payload:
|
|
voice_session.customer_name_resolved_at = values["customer_name_resolved_at"]
|
|
|
|
|
|
def _voice_start_metadata_from_start_response(started: dict[str, Any] | None) -> dict[str, Any]:
|
|
payload = dict((started or {}).get("metadata") or {})
|
|
result = (started or {}).get("start_result")
|
|
if isinstance(result, dict):
|
|
if result.get("language") and not payload.get("voice_start_language"):
|
|
payload["voice_start_language"] = result.get("language")
|
|
if "customer_name_status" in result and "customer_name_status" not in payload:
|
|
payload["customer_name_status"] = result.get("customer_name_status")
|
|
if "customer_name_value" in result and "customer_name_value" not in payload:
|
|
payload["customer_name_value"] = result.get("customer_name_value")
|
|
if "customer_name_source" in result and "customer_name_source" not in payload:
|
|
payload["customer_name_source"] = result.get("customer_name_source")
|
|
if result.get("resolved_at") and "customer_name_resolved_at" not in payload:
|
|
payload["customer_name_resolved_at"] = result.get("resolved_at")
|
|
if result.get("downstream_queue_code") and "downstream_queue_code" not in payload:
|
|
payload["downstream_queue_code"] = result.get("downstream_queue_code")
|
|
return payload
|
|
|
|
|
|
def _mark_last_assistant_segment_interrupted(session, session_id: str) -> None:
|
|
row = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "assistant")
|
|
.where(VoiceTranscriptSegmentRow.is_final.is_(False))
|
|
.order_by(VoiceTranscriptSegmentRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
row = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "assistant")
|
|
.order_by(VoiceTranscriptSegmentRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if row is not None:
|
|
row.barge_in_interrupted = True
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
payload["delivery_status"] = "interrupted"
|
|
payload["delivery_state"] = "interrupted"
|
|
row.payload_json = json.dumps(payload, ensure_ascii=False)
|
|
|
|
|
|
def _append_interaction_timeline(interaction_id: str | None, action: str, metadata: dict[str, Any] | None = None) -> None:
|
|
if not str(interaction_id or "").strip():
|
|
return
|
|
_interaction_request(
|
|
"POST",
|
|
f"/interactions/{interaction_id}/timeline",
|
|
payload={"action": action, "metadata": metadata or {}},
|
|
timeout=3.0,
|
|
)
|
|
|
|
|
|
def _push_bridge_call_state(
|
|
voice_session: VoiceAISessionRow,
|
|
*,
|
|
ai_state: str,
|
|
handoff_reason: str | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> None:
|
|
_bridge_request(
|
|
"POST",
|
|
f"/internal/voice-ai/calls/{voice_session.call_id}/state",
|
|
payload={
|
|
"voice_session_id": voice_session.session_id,
|
|
"ai_session_id": voice_session.ai_session_id,
|
|
"ai_state": ai_state,
|
|
"handoff_reason": handoff_reason,
|
|
"metadata": metadata or {},
|
|
},
|
|
timeout=3.0,
|
|
)
|
|
|
|
|
|
def _media_registration_from_row(row: VoiceAISessionRow, *, queue_code: str | None = None) -> MediaRegistration:
|
|
normalized_queue_code = str(queue_code or "").strip() or None
|
|
voice_v2_for_session = bool(
|
|
_voice_v2_enabled() and normalized_queue_code and normalized_queue_code in _voice_v2_queue_codes()
|
|
)
|
|
streaming_backend = _voice_v2_streaming_asr_backend() if voice_v2_for_session else None
|
|
return MediaRegistration(
|
|
voice_session_id=row.session_id,
|
|
call_id=row.call_id,
|
|
interaction_id=str(row.interaction_id or "").strip(),
|
|
ai_session_id=str(row.ai_session_id or "").strip() or None,
|
|
language=str(row.language or "").strip() or None,
|
|
media_uuid=str(row.media_uuid or "").strip() or None,
|
|
queue_code=normalized_queue_code,
|
|
queue_id=str(row.queue_id or "").strip() or None,
|
|
agent_profile=str(row.agent_profile or "").strip() or None,
|
|
voice_v2_enabled=voice_v2_for_session,
|
|
voice_v2_ack_mode=_voice_v2_ack_mode() if voice_v2_for_session else "disabled",
|
|
voice_v2_streaming_tts=bool(voice_v2_for_session and _voice_v2_streaming_tts_enabled()),
|
|
voice_v2_partial_asr=bool(voice_v2_for_session and _voice_v2_partial_asr_enabled()),
|
|
voice_v2_duplex=bool(voice_v2_for_session and _voice_v2_duplex_enabled()),
|
|
voice_v2_streaming_asr_backend=streaming_backend,
|
|
voice_v2_prebaked_ack=bool(voice_v2_for_session and _voice_v2_prebaked_ack_enabled()),
|
|
# Deliberately NOT gated on voice_v2_for_session: phrase-variant rotation only
|
|
# needs a text pool + live TTS, not the v2 duplex/partial-ASR pipeline, so calls
|
|
# outside the v2 queue allowlist still get varied fillers instead of always the
|
|
# single fixed "Секунду." fallback string.
|
|
voice_v2_emotive_ack=bool(_voice_v2_emotive_ack_enabled()),
|
|
voice_v2_emotive_ack_ru_only=bool(_voice_v2_emotive_ack_ru_only()),
|
|
)
|
|
|
|
|
|
def _load_media_registration_by_uuid(media_uuid: str) -> MediaRegistration | None:
|
|
session = get_session()
|
|
try:
|
|
normalized = normalize_media_uuid(media_uuid)
|
|
row = session.execute(
|
|
select(VoiceAISessionRow)
|
|
.where(VoiceAISessionRow.media_uuid == normalized)
|
|
.order_by(VoiceAISessionRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
return None
|
|
call_link = session.execute(
|
|
select(AsteriskCallLinkRow)
|
|
.where(AsteriskCallLinkRow.call_id == row.call_id)
|
|
.order_by(AsteriskCallLinkRow.id.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
queue_code = str(call_link.queue_code or "").strip() or None if call_link is not None else None
|
|
return _media_registration_from_row(row, queue_code=queue_code)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _persist_media_bridge_event(session_id: str, payload: VoiceAIMediaBridgeEventIn) -> VoiceAISessionRow:
|
|
def _write() -> VoiceAISessionRow:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
if str(voice_session.call_id or "").strip() != str(payload.call_id or "").strip():
|
|
raise HTTPException(status_code=409, detail="call_id mismatch")
|
|
media_uuid = normalize_media_uuid(payload.media_uuid)
|
|
now = utc_now_iso()
|
|
if payload.event_type == "requested":
|
|
voice_session.media_uuid = media_uuid
|
|
voice_session.media_status = "requested"
|
|
voice_session.media_connected_at = None
|
|
voice_session.media_ended_at = None
|
|
else:
|
|
voice_session.media_uuid = voice_session.media_uuid or media_uuid
|
|
voice_session.media_status = "ended"
|
|
voice_session.media_ended_at = now
|
|
voice_session.updated_at = now
|
|
session.commit()
|
|
session.refresh(voice_session)
|
|
return voice_session
|
|
finally:
|
|
session.close()
|
|
|
|
return _retry_db_write(_write)
|
|
|
|
|
|
def _mark_media_connected(session_id: str, media_uuid: str) -> None:
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
voice_session.media_uuid = normalize_media_uuid(media_uuid)
|
|
voice_session.media_status = "connected"
|
|
voice_session.media_connected_at = now
|
|
voice_session.updated_at = now
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _mark_media_ended(session_id: str, reason: str) -> None:
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
voice_session.media_status = "ended"
|
|
voice_session.media_ended_at = now
|
|
voice_session.updated_at = now
|
|
if voice_session.status in {"queued", "greeting", "listening", "thinking", "speaking", "active"}:
|
|
voice_session.status = "completed"
|
|
voice_session.ended_at = voice_session.ended_at or now
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _touch_media_frame(session_id: str) -> None:
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
voice_session.last_media_frame_at = now
|
|
voice_session.updated_at = now
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _load_pending_greeting_text(session_id: str) -> str | None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
if voice_session.disclosure_played_at:
|
|
return None
|
|
rows = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "assistant")
|
|
.order_by(VoiceTranscriptSegmentRow.id.desc())
|
|
).scalars().all()
|
|
for row in rows:
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
if str(payload.get("kind") or "").strip() == "greeting":
|
|
return str(row.text or "").strip() or None
|
|
return None
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _update_reply_delivery_status(
|
|
session,
|
|
*,
|
|
session_id: str,
|
|
text: str,
|
|
is_greeting: bool,
|
|
phase: str | None,
|
|
status: str,
|
|
) -> VoiceAISessionRow:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
pending_rows = session.execute(
|
|
select(VoiceTranscriptSegmentRow)
|
|
.where(VoiceTranscriptSegmentRow.session_id == session_id)
|
|
.where(VoiceTranscriptSegmentRow.speaker == "assistant")
|
|
.order_by(VoiceTranscriptSegmentRow.id.desc())
|
|
).scalars().all()
|
|
matched_row = None
|
|
normalized_text = str(text or "").strip()
|
|
expected_phase = str(phase or "").strip() or None
|
|
for row in pending_rows:
|
|
try:
|
|
payload = json.loads(row.payload_json or "{}")
|
|
except Exception:
|
|
payload = {}
|
|
row_kind = str(payload.get("kind") or "").strip()
|
|
row_phase = str(payload.get("metadata", {}).get("phase") or "").strip() or None
|
|
if is_greeting:
|
|
if row_kind != "greeting":
|
|
continue
|
|
elif str(row.text or "").strip() != normalized_text:
|
|
continue
|
|
if expected_phase and row_phase not in {expected_phase, None}:
|
|
continue
|
|
matched_row = row
|
|
payload["delivery_status"] = status
|
|
payload["delivery_state"] = status
|
|
payload.setdefault("provider", _current_tts_provider_name())
|
|
row.payload_json = json.dumps(payload, ensure_ascii=False)
|
|
break
|
|
if matched_row is not None:
|
|
matched_row.is_final = status == "delivered"
|
|
if status in {"interrupted", "discarded"}:
|
|
matched_row.barge_in_interrupted = status == "interrupted"
|
|
if is_greeting and status == "delivered" and not voice_session.disclosure_played_at:
|
|
voice_session.disclosure_played_at = now
|
|
if status == "delivered":
|
|
voice_session.last_ai_reply_at = now
|
|
voice_session.updated_at = now
|
|
return voice_session
|
|
|
|
|
|
def _mark_reply_started(session_id: str, text: str, is_greeting: bool, phase: str | None = None) -> None:
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
_update_reply_delivery_status(
|
|
session,
|
|
session_id=session_id,
|
|
text=text,
|
|
is_greeting=is_greeting,
|
|
phase=phase,
|
|
status="started",
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _mark_reply_delivered(session_id: str, text: str, is_greeting: bool, phase: str | None = None) -> None:
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
_update_reply_delivery_status(
|
|
session,
|
|
text=text,
|
|
session_id=session_id,
|
|
is_greeting=is_greeting,
|
|
phase=phase,
|
|
status="delivered",
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _record_latency_metric(session_id: str, metric: str, latency_ms: int) -> None:
|
|
normalized_metric = str(metric or "").strip()
|
|
if not normalized_metric:
|
|
return
|
|
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
_record_segment(
|
|
session,
|
|
voice_session=voice_session,
|
|
speaker="system",
|
|
source_type="runtime",
|
|
text=normalized_metric,
|
|
sequence_no=_next_sequence(session, voice_session.session_id),
|
|
payload={
|
|
"kind": "latency_metric",
|
|
"metric": normalized_metric,
|
|
"latency_ms": int(latency_ms),
|
|
},
|
|
is_final=True,
|
|
)
|
|
voice_session.updated_at = utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _mark_reply_discarded(session_id: str, text: str, phase: str | None = None, status: str = "discarded") -> None:
|
|
normalized_status = str(status or "").strip().lower() or "discarded"
|
|
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
_update_reply_delivery_status(
|
|
session,
|
|
session_id=session_id,
|
|
text=text,
|
|
is_greeting=False,
|
|
phase=phase,
|
|
status=normalized_status,
|
|
)
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _record_runtime_reply_planned(
|
|
session_id: str,
|
|
text: str,
|
|
metadata: dict[str, Any] | None = None,
|
|
kind: str = "reply",
|
|
) -> None:
|
|
normalized_text = str(text or "").strip()
|
|
if not normalized_text:
|
|
return
|
|
|
|
def _write() -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
_record_segment(
|
|
session,
|
|
voice_session=voice_session,
|
|
speaker="assistant",
|
|
source_type="tts",
|
|
text=normalized_text,
|
|
sequence_no=_next_sequence(session, voice_session.session_id),
|
|
payload={
|
|
"provider": _current_tts_provider_name(),
|
|
"delivery_status": "planned",
|
|
"delivery_state": "planned",
|
|
"kind": kind,
|
|
"metadata": metadata or {},
|
|
},
|
|
is_final=False,
|
|
)
|
|
voice_session.updated_at = utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
|
|
_retry_db_write(_write)
|
|
|
|
|
|
def _set_voice_session_state(
|
|
session_id: str,
|
|
ai_state: str,
|
|
handoff_reason: str | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> None:
|
|
def _write() -> VoiceAISessionRow:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
voice_session.status = ai_state
|
|
if handoff_reason is not None:
|
|
voice_session.handoff_reason = str(handoff_reason).strip() or None
|
|
_apply_voice_start_metadata(voice_session, metadata)
|
|
voice_session.updated_at = now
|
|
if ai_state in {"completed", "closed"}:
|
|
voice_session.ended_at = voice_session.ended_at or now
|
|
session.commit()
|
|
session.refresh(voice_session)
|
|
return voice_session
|
|
finally:
|
|
session.close()
|
|
|
|
voice_session = _retry_db_write(_write)
|
|
_push_bridge_call_state(
|
|
voice_session,
|
|
ai_state=ai_state,
|
|
handoff_reason=handoff_reason,
|
|
metadata=metadata,
|
|
)
|
|
_sync_sales_voice_session(
|
|
voice_session.session_id,
|
|
include_transcript=ai_state in {"handoff_requested", "human_owned", "completed", "closed", "error"},
|
|
metadata=metadata,
|
|
)
|
|
if ai_state == "error":
|
|
_append_interaction_timeline(
|
|
voice_session.interaction_id,
|
|
"ai.error",
|
|
{
|
|
"call_id": voice_session.call_id,
|
|
"voice_session_id": voice_session.session_id,
|
|
"ai_session_id": voice_session.ai_session_id,
|
|
"error": str(handoff_reason or "").strip()[:500],
|
|
**(metadata or {}),
|
|
},
|
|
)
|
|
|
|
|
|
def _request_runtime_handoff_payload(
|
|
session_id: str,
|
|
*,
|
|
reason: str,
|
|
summary: dict[str, Any] | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
if not str(voice_session.interaction_id or "").strip():
|
|
raise RuntimeError("Voice AI session is missing interaction_id")
|
|
handoff_payload = VoiceAIHandoffRequestIn(
|
|
voice_session_id=voice_session.session_id,
|
|
ai_session_id=voice_session.ai_session_id,
|
|
interaction_id=voice_session.interaction_id,
|
|
target_queue_id=voice_session.handoff_target_queue_id,
|
|
reason=reason,
|
|
summary=summary or {},
|
|
metadata=metadata or {},
|
|
)
|
|
LOGGER.warning(
|
|
"voice_runtime.handoff_request call_id=%s session_id=%s ai_session_id=%s interaction_id=%s target_queue_id=%s reason=%s",
|
|
voice_session.call_id,
|
|
voice_session.session_id,
|
|
voice_session.ai_session_id,
|
|
voice_session.interaction_id,
|
|
voice_session.handoff_target_queue_id,
|
|
reason[:300],
|
|
)
|
|
try:
|
|
_bridge_request(
|
|
"POST",
|
|
f"/internal/voice-ai/calls/{voice_session.call_id}/handoff",
|
|
payload=handoff_payload.model_dump(),
|
|
timeout=_handoff_timeout_seconds(),
|
|
)
|
|
except Exception as exc:
|
|
LOGGER.warning(
|
|
"voice_runtime.handoff_failed call_id=%s session_id=%s ai_session_id=%s error=%s",
|
|
voice_session.call_id,
|
|
voice_session.session_id,
|
|
voice_session.ai_session_id,
|
|
str(exc)[:500],
|
|
)
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _request_runtime_handoff(
|
|
session_id: str,
|
|
customer_request_text: str,
|
|
decision: VoiceAITurnDecisionOut,
|
|
) -> None:
|
|
handoff_summary = {
|
|
"customer_request_text": customer_request_text,
|
|
"ai_outcome_text": decision.summary_text or decision.reply_text or decision.handoff_reason or "",
|
|
"recommended_next_step": "Continue the call manually and confirm the collected context.",
|
|
}
|
|
if decision.metadata:
|
|
for key in (
|
|
"customer_name_status",
|
|
"customer_name_value",
|
|
"customer_name_source",
|
|
"voice_start_language",
|
|
):
|
|
value = decision.metadata.get(key)
|
|
if value:
|
|
handoff_summary[key] = value
|
|
return _request_runtime_handoff_payload(
|
|
session_id,
|
|
reason=decision.handoff_reason or "AI requested human handoff.",
|
|
summary=handoff_summary,
|
|
metadata=decision.metadata,
|
|
)
|
|
|
|
|
|
def _handle_media_error(session_id: str, message: str, metadata: dict[str, Any] | None = None) -> None:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
previous_status = str(voice_session.status or "").strip()
|
|
now = utc_now_iso()
|
|
voice_session.status = "error"
|
|
voice_session.handoff_reason = str(message or "media_error")[:1000]
|
|
voice_session.media_status = "error"
|
|
voice_session.media_ended_at = now
|
|
voice_session.updated_at = now
|
|
session.commit()
|
|
_push_bridge_call_state(
|
|
voice_session,
|
|
ai_state="error",
|
|
handoff_reason=voice_session.handoff_reason,
|
|
metadata=metadata,
|
|
)
|
|
_append_interaction_timeline(
|
|
voice_session.interaction_id,
|
|
"ai.error",
|
|
{
|
|
"call_id": voice_session.call_id,
|
|
"voice_session_id": voice_session.session_id,
|
|
"ai_session_id": voice_session.ai_session_id,
|
|
"error": voice_session.handoff_reason,
|
|
**(metadata or {}),
|
|
},
|
|
)
|
|
if previous_status not in {"human_owned", "completed", "closed", "handoff_requested", "handoff_required"} and str(voice_session.interaction_id or "").strip():
|
|
handoff_payload = VoiceAIHandoffRequestIn(
|
|
voice_session_id=voice_session.session_id,
|
|
ai_session_id=voice_session.ai_session_id,
|
|
interaction_id=voice_session.interaction_id,
|
|
target_queue_id=voice_session.handoff_target_queue_id,
|
|
reason=voice_session.handoff_reason or "Voice AI media error.",
|
|
summary={
|
|
"customer_request_text": "Не удалось завершить AI-аудио-сценарий.",
|
|
"ai_outcome_text": voice_session.handoff_reason or "Voice AI media error.",
|
|
"recommended_next_step": "Продолжить звонок вручную и проверить состояние Voice AI runtime.",
|
|
},
|
|
)
|
|
try:
|
|
_bridge_request(
|
|
"POST",
|
|
f"/internal/voice-ai/calls/{voice_session.call_id}/handoff",
|
|
payload=handoff_payload.model_dump(),
|
|
timeout=_handoff_timeout_seconds(),
|
|
)
|
|
except Exception:
|
|
pass
|
|
_sync_sales_voice_session(
|
|
voice_session.session_id,
|
|
include_transcript=True,
|
|
metadata=metadata,
|
|
)
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _process_voice_ai_turn_sync(
|
|
session_id: str,
|
|
payload: VoiceAITurnIn,
|
|
*,
|
|
auto_handoff: bool,
|
|
) -> VoiceAITurnDecisionOut:
|
|
session = get_session()
|
|
voice_session = None
|
|
payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {}
|
|
reply_phase = str(payload_metadata.get("reply_phase") or "").strip().lower()
|
|
early_plan_only = reply_phase == "early_plan"
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
if not early_plan_only and voice_session.status in {"human_owned", "completed", "error"}:
|
|
raise HTTPException(status_code=409, detail="Voice AI session is no longer active")
|
|
now = utc_now_iso()
|
|
if not early_plan_only:
|
|
voice_session.status = "thinking"
|
|
voice_session.last_user_utterance_at = now
|
|
voice_session.updated_at = now
|
|
if payload.barge_in:
|
|
_mark_last_assistant_segment_interrupted(session, voice_session.session_id)
|
|
_record_segment(
|
|
session,
|
|
voice_session=voice_session,
|
|
speaker="caller",
|
|
source_type="asr",
|
|
text=payload.transcript_text,
|
|
sequence_no=max(payload.sequence_no, _next_sequence(session, voice_session.session_id)),
|
|
payload={
|
|
"provider": _ASR_PROVIDER.name,
|
|
"metadata": payload_metadata,
|
|
"barge_in": payload.barge_in,
|
|
"intent_bearing": False,
|
|
},
|
|
)
|
|
session.commit()
|
|
|
|
decision_payload = _orchestrator_request(
|
|
"POST",
|
|
f"/ai/voice/sessions/{session_id}/turns",
|
|
payload=payload.model_dump(),
|
|
timeout=12.0,
|
|
)
|
|
decision = VoiceAITurnDecisionOut.model_validate(decision_payload)
|
|
if not early_plan_only:
|
|
voice_session.language = decision.language or voice_session.language
|
|
voice_session.handoff_reason = decision.handoff_reason
|
|
_apply_voice_start_metadata(voice_session, decision.metadata)
|
|
_annotate_latest_caller_segment(
|
|
session,
|
|
session_id=voice_session.session_id,
|
|
provider_name=_ASR_PROVIDER.name,
|
|
transcript_text=payload.transcript_text,
|
|
intent=decision.intent,
|
|
metadata=payload_metadata,
|
|
)
|
|
voice_session.status = decision.status or ("handoff_requested" if decision.needs_handoff else "active")
|
|
voice_session.updated_at = utc_now_iso()
|
|
if decision.reply_text and not bool(payload_metadata.get("runtime_defer_reply_planned")):
|
|
_record_segment(
|
|
session,
|
|
voice_session=voice_session,
|
|
speaker="assistant",
|
|
source_type="tts",
|
|
text=decision.reply_text,
|
|
sequence_no=_next_sequence(session, voice_session.session_id),
|
|
payload={
|
|
"intent": decision.intent,
|
|
"kb_refs": decision.kb_refs,
|
|
"model": decision.model,
|
|
"provider": _current_tts_provider_name(),
|
|
"delivery_status": "planned",
|
|
"delivery_state": "planned",
|
|
"metadata": decision.metadata,
|
|
},
|
|
is_final=False,
|
|
)
|
|
session.commit()
|
|
if auto_handoff and decision.needs_handoff and not early_plan_only:
|
|
_request_runtime_handoff(session_id, payload.transcript_text, decision)
|
|
if not early_plan_only:
|
|
_sync_sales_voice_session(
|
|
session_id,
|
|
include_transcript=True,
|
|
summary=decision.summary_text or decision.reply_text or payload.transcript_text,
|
|
metadata=decision.metadata,
|
|
)
|
|
return decision
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
if voice_session is not None:
|
|
voice_session.status = "error"
|
|
voice_session.handoff_reason = str(exc)[:1000]
|
|
voice_session.updated_at = utc_now_iso()
|
|
session.commit()
|
|
raise HTTPException(status_code=502, detail=f"Voice AI runtime turn failed: {exc}") from exc
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def _media_process_turn(
|
|
session_id: str,
|
|
transcript_text: str,
|
|
language: str | None,
|
|
barge_in: bool,
|
|
metadata: dict[str, Any] | None,
|
|
) -> VoiceAITurnDecisionOut:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
interaction_id = str(voice_session.interaction_id or "").strip()
|
|
call_id = str(voice_session.call_id or "").strip()
|
|
finally:
|
|
session.close()
|
|
if not interaction_id or not call_id:
|
|
raise RuntimeError("Voice AI session is missing call linkage")
|
|
|
|
payload = VoiceAITurnIn(
|
|
voice_session_id=session_id,
|
|
call_id=call_id,
|
|
interaction_id=interaction_id,
|
|
transcript_text=transcript_text,
|
|
language=language,
|
|
sequence_no=1,
|
|
barge_in=barge_in,
|
|
metadata=metadata or {},
|
|
)
|
|
return _process_voice_ai_turn_sync(session_id, payload, auto_handoff=False)
|
|
|
|
|
|
def _complete_voice_ai_session_start(session_id: str, start_payload: VoiceAIStartIn) -> None:
|
|
try:
|
|
started = _orchestrator_request(
|
|
"POST",
|
|
f"/ai/voice/sessions/{session_id}/start",
|
|
payload=start_payload.model_dump(),
|
|
timeout=10.0,
|
|
)
|
|
except Exception as exc:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
voice_session.handoff_reason = str(exc)[:1000]
|
|
voice_session.updated_at = utc_now_iso()
|
|
session.commit()
|
|
finally:
|
|
session.close()
|
|
return
|
|
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
greeting_text = str(started.get("greeting_text") or "").strip()
|
|
handoff_metadata = _voice_start_metadata_from_start_response(started)
|
|
needs_handoff = bool(started.get("needs_handoff"))
|
|
handoff_reason = str(started.get("handoff_reason") or "").strip() or None
|
|
voice_session.ai_session_id = str(started.get("session_id") or voice_session.ai_session_id or "").strip() or None
|
|
voice_session.language = _resolved_voice_language(started.get("language"), voice_session.language)
|
|
_apply_voice_start_metadata(voice_session, handoff_metadata)
|
|
if greeting_text:
|
|
_upsert_greeting_segment(session, voice_session, greeting_text)
|
|
if needs_handoff:
|
|
voice_session.status = "handoff_requested"
|
|
elif greeting_text or _load_greeting_segment(session, session_id) is not None:
|
|
voice_session.status = "greeting"
|
|
elif not str(voice_session.status or "").strip():
|
|
voice_session.status = "active"
|
|
else:
|
|
voice_session.status = "active"
|
|
voice_session.handoff_reason = handoff_reason
|
|
voice_session.updated_at = now
|
|
session.commit()
|
|
session.refresh(voice_session)
|
|
finally:
|
|
session.close()
|
|
_push_bridge_call_state(
|
|
voice_session,
|
|
ai_state=voice_session.status,
|
|
handoff_reason=voice_session.handoff_reason,
|
|
metadata=handoff_metadata,
|
|
)
|
|
_sync_sales_voice_session(
|
|
voice_session.session_id,
|
|
include_transcript=False,
|
|
metadata=handoff_metadata,
|
|
)
|
|
if needs_handoff:
|
|
handoff_summary = {
|
|
"customer_request_text": "voice_start",
|
|
"ai_outcome_text": str(started.get("summary_text") or "").strip(),
|
|
"recommended_next_step": "Continue the next voice stage using the language and customer name handoff.",
|
|
}
|
|
for key in (
|
|
"customer_name_status",
|
|
"customer_name_value",
|
|
"customer_name_source",
|
|
"voice_start_language",
|
|
):
|
|
value = handoff_metadata.get(key)
|
|
if value:
|
|
handoff_summary[key] = value
|
|
_request_runtime_handoff_payload(
|
|
session_id,
|
|
reason=handoff_reason or "voice_start_completed",
|
|
summary=handoff_summary,
|
|
metadata=handoff_metadata,
|
|
)
|
|
|
|
|
|
_MEDIA_RUNTIME = AudioSocketMediaRuntime(
|
|
enabled=_audiosocket_enabled(),
|
|
host=_audiosocket_host(),
|
|
port=_audiosocket_port(),
|
|
frame_ms=_vad_frame_ms(),
|
|
idle_timeout_seconds=_media_idle_timeout_seconds(),
|
|
registration_wait_timeout_seconds=_media_registration_wait_timeout_seconds(),
|
|
min_speech_ms=_vad_min_speech_ms(),
|
|
trailing_silence_ms=_vad_trailing_silence_ms(),
|
|
max_turn_ms=_turn_max_ms(),
|
|
vad_rms_threshold=_vad_rms_threshold(),
|
|
asr_provider=_ASR_PROVIDER,
|
|
streaming_asr_provider=_STREAMING_ASR_PROVIDER,
|
|
tts_provider=_TTS_PROVIDER,
|
|
load_registration_by_media_uuid=_load_media_registration_by_uuid,
|
|
mark_media_connected=_mark_media_connected,
|
|
mark_media_ended=_mark_media_ended,
|
|
touch_media_frame=_touch_media_frame,
|
|
set_state=_set_voice_session_state,
|
|
get_pending_greeting=_load_pending_greeting_text,
|
|
mark_reply_started=_mark_reply_started,
|
|
mark_reply_delivered=_mark_reply_delivered,
|
|
mark_reply_discarded=_mark_reply_discarded,
|
|
plan_reply=_record_runtime_reply_planned,
|
|
record_latency=_record_latency_metric,
|
|
process_turn=_media_process_turn,
|
|
request_handoff=_request_runtime_handoff,
|
|
handle_media_error=_handle_media_error,
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _lifespan(_: FastAPI):
|
|
_sync_session = get_session()
|
|
try:
|
|
sync_ai_operator_config_from_code(_sync_session)
|
|
finally:
|
|
_sync_session.close()
|
|
await _MEDIA_RUNTIME.start()
|
|
try:
|
|
yield
|
|
finally:
|
|
await _MEDIA_RUNTIME.stop()
|
|
|
|
|
|
app = FastAPI(title="ai-voice-runtime-service", version="1.0.0", lifespan=_lifespan)
|
|
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
def health() -> HealthResponse:
|
|
return HealthResponse(status="ok", service="ai-voice-runtime-service", version="v1")
|
|
|
|
|
|
@app.post("/internal/voice-ai/sessions", response_model=VoiceAISessionOut)
|
|
def create_voice_ai_session(
|
|
payload: VoiceAISessionCreateIn,
|
|
_: dict = Depends(require_roles(Role.ADMIN)),
|
|
) -> VoiceAISessionOut:
|
|
session = get_session()
|
|
voice_session = None
|
|
should_start_async = False
|
|
try:
|
|
voice_session = session.execute(
|
|
select(VoiceAISessionRow).where(VoiceAISessionRow.call_id == payload.call_id)
|
|
).scalar_one_or_none()
|
|
now = utc_now_iso()
|
|
language = _resolved_voice_language(
|
|
payload.language_hint,
|
|
voice_session.language if voice_session is not None else None,
|
|
)
|
|
payload_metadata = payload.metadata if isinstance(payload.metadata, dict) else {}
|
|
effective_tts_provider_name = _current_tts_provider_name()
|
|
is_voice_start = (
|
|
str(payload.agent_profile or "").strip() == "voice_start"
|
|
or str(payload_metadata.get("stage") or "").strip() == "voice_start"
|
|
)
|
|
operator_config = load_effective_ai_operator_config(session)
|
|
if voice_session is None:
|
|
voice_session = VoiceAISessionRow(
|
|
session_id=new_id("avs"),
|
|
call_id=payload.call_id,
|
|
linked_id=payload.linked_id,
|
|
interaction_id=payload.interaction_id,
|
|
customer_id=None,
|
|
queue_id=payload.queue_id,
|
|
ai_session_id=None,
|
|
agent_profile=payload.agent_profile,
|
|
language=language,
|
|
asr_provider=_ASR_PROVIDER.name,
|
|
tts_provider=effective_tts_provider_name,
|
|
status="greeting",
|
|
handoff_reason=None,
|
|
handoff_target_queue_id=payload.handoff_queue_id or payload.queue_id,
|
|
media_uuid=None,
|
|
media_status=None,
|
|
media_connected_at=None,
|
|
media_ended_at=None,
|
|
last_media_frame_at=None,
|
|
disclosure_played_at=None,
|
|
last_user_utterance_at=None,
|
|
last_ai_reply_at=None,
|
|
started_at=now,
|
|
updated_at=now,
|
|
ended_at=None,
|
|
)
|
|
session.add(voice_session)
|
|
should_start_async = True
|
|
else:
|
|
stage_changed = (
|
|
str(voice_session.queue_id or "").strip() != str(payload.queue_id or "").strip()
|
|
or str(voice_session.agent_profile or "").strip() != str(payload.agent_profile or "").strip()
|
|
)
|
|
voice_session.linked_id = payload.linked_id
|
|
voice_session.interaction_id = payload.interaction_id
|
|
voice_session.queue_id = payload.queue_id
|
|
voice_session.agent_profile = payload.agent_profile
|
|
voice_session.language = language
|
|
voice_session.tts_provider = effective_tts_provider_name
|
|
voice_session.status = "greeting"
|
|
voice_session.handoff_reason = None
|
|
voice_session.handoff_target_queue_id = payload.handoff_queue_id or voice_session.handoff_target_queue_id or payload.queue_id
|
|
if stage_changed:
|
|
voice_session.ai_session_id = None
|
|
voice_session.disclosure_played_at = None
|
|
voice_session.ended_at = None
|
|
should_start_async = stage_changed or not str(voice_session.ai_session_id or "").strip()
|
|
voice_session.updated_at = now
|
|
_apply_voice_start_metadata(voice_session, payload_metadata)
|
|
greeting_text = _default_voice_greeting(
|
|
language,
|
|
agent_profile=payload.agent_profile,
|
|
operator_config=operator_config,
|
|
)
|
|
if not is_voice_start:
|
|
if voice_session.ai_session_id is None or should_start_async:
|
|
_queue_new_greeting_segment(session, voice_session, greeting_text)
|
|
else:
|
|
_upsert_greeting_segment(session, voice_session, greeting_text)
|
|
session.commit()
|
|
|
|
start_payload = VoiceAIStartIn(
|
|
voice_session_id=voice_session.session_id,
|
|
call_id=payload.call_id,
|
|
interaction_id=payload.interaction_id,
|
|
customer_id=None,
|
|
language_hint=language,
|
|
agent_profile=payload.agent_profile,
|
|
metadata=payload_metadata,
|
|
)
|
|
if should_start_async:
|
|
voice_start_thread = threading.Thread(
|
|
target=_run_voice_start_thread,
|
|
args=(voice_session.session_id, start_payload),
|
|
name=f"voice-start-{voice_session.session_id}",
|
|
daemon=True,
|
|
)
|
|
_register_voice_start_thread(voice_start_thread)
|
|
voice_start_thread.start()
|
|
_sync_sales_voice_session(
|
|
voice_session.session_id,
|
|
caller_number=payload.caller_number,
|
|
caller_name=payload.caller_name,
|
|
queue_code=str(payload_metadata.get("queue_code") or "").strip() or None,
|
|
include_transcript=False,
|
|
metadata=payload_metadata,
|
|
)
|
|
return VoiceAISessionOut(
|
|
voice_session_id=voice_session.session_id,
|
|
ai_session_id=voice_session.ai_session_id,
|
|
status=voice_session.status, # type: ignore[arg-type]
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as exc:
|
|
if voice_session is not None:
|
|
voice_session.status = "error"
|
|
voice_session.handoff_reason = str(exc)[:1000]
|
|
voice_session.updated_at = utc_now_iso()
|
|
session.commit()
|
|
raise HTTPException(status_code=502, detail=f"Voice AI session start failed: {exc}") from exc
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@app.post("/internal/voice-ai/sessions/{session_id}/media-bridge", response_model=VoiceAISessionOut)
|
|
def register_voice_ai_media_bridge(
|
|
session_id: str,
|
|
payload: VoiceAIMediaBridgeEventIn,
|
|
_: dict = Depends(require_roles(Role.ADMIN)),
|
|
) -> VoiceAISessionOut:
|
|
voice_session = _persist_media_bridge_event(session_id, payload)
|
|
if payload.event_type == "ended":
|
|
_MEDIA_RUNTIME.close_session_sync(session_id, reason=payload.reason or "media_bridge_ended")
|
|
return VoiceAISessionOut(
|
|
voice_session_id=voice_session.session_id,
|
|
ai_session_id=voice_session.ai_session_id,
|
|
status=voice_session.status, # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
@app.post("/internal/voice-ai/sessions/{session_id}/telephony-events", response_model=VoiceAISessionOut)
|
|
def push_voice_ai_telephony_event(
|
|
session_id: str,
|
|
payload: VoiceAITelephonyEventIn,
|
|
_: dict = Depends(require_roles(Role.ADMIN)),
|
|
) -> VoiceAISessionOut:
|
|
session = get_session()
|
|
try:
|
|
voice_session = _load_voice_session(session, session_id)
|
|
now = utc_now_iso()
|
|
if payload.event_type == "call.connected":
|
|
if voice_session.status in {"queued", "greeting", "active", "thinking", "speaking"}:
|
|
voice_session.status = "listening"
|
|
elif payload.event_type == "operator.connected":
|
|
voice_session.status = "human_owned"
|
|
elif payload.event_type == "call.ended":
|
|
voice_session.status = "completed" if voice_session.status != "error" else "error"
|
|
voice_session.ended_at = voice_session.ended_at or now
|
|
voice_session.media_status = voice_session.media_status or "ended"
|
|
voice_session.media_ended_at = voice_session.media_ended_at or now
|
|
voice_session.updated_at = now
|
|
session.commit()
|
|
ai_session_id = voice_session.ai_session_id
|
|
status = str(voice_session.status or "queued")
|
|
finally:
|
|
session.close()
|
|
|
|
if payload.event_type == "call.ended":
|
|
_MEDIA_RUNTIME.close_session_sync(session_id, reason="call_ended")
|
|
try:
|
|
_orchestrator_request(
|
|
"POST",
|
|
f"/ai/voice/sessions/{session_id}/close",
|
|
payload={},
|
|
timeout=5.0,
|
|
)
|
|
except Exception:
|
|
pass
|
|
_sync_sales_voice_session(
|
|
session_id,
|
|
include_transcript=payload.event_type in {"operator.connected", "call.ended"},
|
|
metadata={"telephony_event_type": payload.event_type, **(payload.payload or {})},
|
|
)
|
|
|
|
return VoiceAISessionOut(
|
|
voice_session_id=session_id,
|
|
ai_session_id=ai_session_id,
|
|
status=status, # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
@app.post("/internal/voice-ai/sessions/{session_id}/turns", response_model=VoiceAITurnDecisionOut)
|
|
def process_voice_ai_turn(
|
|
session_id: str,
|
|
payload: VoiceAITurnIn,
|
|
_: dict = Depends(require_roles(Role.ADMIN)),
|
|
) -> VoiceAITurnDecisionOut:
|
|
return _process_voice_ai_turn_sync(session_id, payload, auto_handoff=True)
|