Senad Uka
714aa1782b
- AudioSocket server (asyncio TCP, Asterisk wire protocol) bridging calls into the same M2 agent used by chat - Call session engine: greeting → energy-VAD utterance collection → STT → agent turn → TTS playback with barge-in (120ms of caller speech stops playback); inactivity + max-duration guards; unintelligible audio reaches the agent as '[nerazumljivo]' so the two-attempt rule stays in the prompt - STTProvider (faster-whisper, lang hint 'sr', GPU/CPU via env) and TTSProvider (Azure Neural raw-8k PCM / ElevenLabs Flash) + test fakes - Recording (mixed caller+agent WAV), transcripts, per-turn latency trace, metering via record_agent_call (§12) - Internal dialplan API: /internal/calls/register (ring targets computed from working hours + ring settings §5.2), /answered (human outcome, no minutes), /hangup (abandoned → missed-call SMS §5.5); shared-token auth - Asterisk config generator (pjsip.conf + extensions.conf from DB): worker endpoints, per-tenant test-caller endpoint, register→Dial→AudioSocket dialplan with h-extension reporting — validated against a real Asterisk 20 container (modules, dialplan, endpoints all load) - docker-compose 'voice' profile (voice server + Asterisk), Dockerfile.voice, docs/VOICE_TESTING.md softphone runbook - 16 new tests incl. full fake-call e2e and a real-TCP AudioSocket wire test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
247 lines
9.0 KiB
Python
247 lines
9.0 KiB
Python
"""One agent-handled phone call: audio loop, barge-in, recording, metering (M4).
|
|
|
|
The transport hands us paced 20 ms SLIN frames from Asterisk (AudioSocket) and
|
|
accepts frames to play back. Flow per §6.2: greeting → collect utterance (VAD)
|
|
→ STT → agent turn (M2 loop, same tools as chat) → TTS → play, with barge-in
|
|
(caller speech stops playback). Unintelligible audio reaches the agent as
|
|
'[nerazumljivo]' so the two-attempts rule (§6.4/A.6) lives in the prompt.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
|
|
from sqlalchemy import select
|
|
|
|
from gogo.agent.loop import AgentConversation
|
|
from gogo.config import get_settings
|
|
from gogo.db import get_sessionmaker
|
|
from gogo.metering import record_agent_call
|
|
from gogo.models import Call, CallRegistration, Tenant, utcnow
|
|
from gogo.voice.audio import FRAME_BYTES, FRAME_MS, CallRecorder
|
|
from gogo.voice.stt import get_stt
|
|
from gogo.voice.tts import get_tts
|
|
from gogo.voice.vad import UtteranceCollector
|
|
|
|
log = logging.getLogger("gogo.voice.call")
|
|
|
|
UNCLEAR_MARKER = "[nerazumljivo]"
|
|
GOODBYE_TIMEOUT = (
|
|
"Izgleda da veza ne radi. Salon će vidjeti vaš broj i javiti vam se. Prijatno!"
|
|
)
|
|
|
|
BARGE_IN_FRAMES = 6 # 120 ms of speech during playback stops it
|
|
|
|
|
|
@dataclass
|
|
class CallLimits:
|
|
inactivity_s: float = 30.0
|
|
max_call_s: float = 600.0
|
|
max_turns: int = 30
|
|
|
|
|
|
class Transport:
|
|
"""What CallSession needs from the wire (implemented by AudioSocket + fakes)."""
|
|
|
|
async def read_frame(self) -> bytes | None:
|
|
"""Next inbound 20 ms frame, or None on hangup."""
|
|
raise NotImplementedError
|
|
|
|
async def send_audio(self, pcm: bytes) -> None:
|
|
raise NotImplementedError
|
|
|
|
async def hangup(self) -> None:
|
|
raise NotImplementedError
|
|
|
|
|
|
class CallSession:
|
|
def __init__(
|
|
self,
|
|
transport: Transport,
|
|
registration_id: uuid.UUID,
|
|
*,
|
|
limits: CallLimits | None = None,
|
|
pace: float = FRAME_MS / 1000,
|
|
):
|
|
self.transport = transport
|
|
self.registration_id = registration_id
|
|
self.limits = limits or CallLimits()
|
|
self.pace = pace
|
|
self.recorder = CallRecorder()
|
|
self.collector = UtteranceCollector()
|
|
self._started = time.monotonic()
|
|
|
|
async def run(self) -> None:
|
|
async with get_sessionmaker()() as session:
|
|
reg = (
|
|
await session.execute(
|
|
select(CallRegistration).where(CallRegistration.id == self.registration_id)
|
|
)
|
|
).scalar_one_or_none()
|
|
if reg is None:
|
|
log.error("no registration for call %s — hanging up", self.registration_id)
|
|
await self.transport.hangup()
|
|
return
|
|
tenant = (
|
|
await session.execute(select(Tenant).where(Tenant.id == reg.tenant_id))
|
|
).scalar_one()
|
|
reg.handled_by_agent = True
|
|
|
|
call = Call(
|
|
tenant_id=tenant.id,
|
|
caller_msisdn=reg.caller_msisdn,
|
|
started_at=utcnow(),
|
|
)
|
|
session.add(call)
|
|
await session.flush()
|
|
|
|
convo = AgentConversation(
|
|
session,
|
|
tenant,
|
|
channel="voice",
|
|
caller_phone=reg.caller_msisdn,
|
|
call_id=call.id,
|
|
)
|
|
tts, voice = get_tts(tenant.tts_voice)
|
|
stt = get_stt()
|
|
|
|
try:
|
|
greeting = await convo.greeting()
|
|
await session.commit()
|
|
await self._speak(tts, voice, greeting)
|
|
|
|
for _turn in range(self.limits.max_turns):
|
|
if time.monotonic() - self._started > self.limits.max_call_s:
|
|
log.warning("max call duration hit (tenant %s)", tenant.slug)
|
|
break
|
|
utterance = await self._collect_utterance()
|
|
if utterance is None: # hangup
|
|
break
|
|
if utterance == b"": # inactivity
|
|
await self._speak(tts, voice, GOODBYE_TIMEOUT)
|
|
break
|
|
|
|
t0 = time.monotonic()
|
|
text = await stt.transcribe(utterance)
|
|
stt_ms = int((time.monotonic() - t0) * 1000)
|
|
|
|
t0 = time.monotonic()
|
|
reply = await convo.user_turn(text or UNCLEAR_MARKER)
|
|
llm_ms = int((time.monotonic() - t0) * 1000)
|
|
await session.commit()
|
|
|
|
t0 = time.monotonic()
|
|
interrupted = await self._speak(tts, voice, reply)
|
|
tts_ms = int((time.monotonic() - t0) * 1000)
|
|
|
|
trace = call.trace or {}
|
|
turns = trace.setdefault("turns", [])
|
|
turns.append(
|
|
{"stt_ms": stt_ms, "llm_ms": llm_ms, "speak_ms": tts_ms,
|
|
"interrupted": interrupted, "chars": len(reply)}
|
|
)
|
|
call.trace = dict(trace)
|
|
|
|
if self._agent_closed(convo):
|
|
break
|
|
except Exception: # noqa: BLE001 — a crash must still finalize the call
|
|
log.exception("call session error (tenant %s)", tenant.slug)
|
|
finally:
|
|
await self._finalize(session, tenant, call, convo, reg)
|
|
await self.transport.hangup()
|
|
|
|
# -- audio helpers ------------------------------------------------------
|
|
|
|
async def _speak(self, tts, voice: str, text: str) -> bool:
|
|
"""Synthesize and play; returns True if the caller barged in."""
|
|
if not text:
|
|
return False
|
|
try:
|
|
pcm = await tts.synthesize(text, voice)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("TTS failed — continuing without audio")
|
|
return False
|
|
|
|
voiced_streak = 0
|
|
for i in range(0, len(pcm), FRAME_BYTES):
|
|
frame = pcm[i : i + FRAME_BYTES]
|
|
await self.transport.send_audio(frame)
|
|
self.recorder.add_agent(frame)
|
|
if self.pace:
|
|
await asyncio.sleep(self.pace)
|
|
# barge-in: watch inbound audio while we play
|
|
inbound = await self._poll_frame()
|
|
if inbound is not None:
|
|
self.recorder.add_caller(inbound)
|
|
if self.collector.is_speech(inbound):
|
|
voiced_streak += 1
|
|
if voiced_streak >= BARGE_IN_FRAMES:
|
|
self.collector.reset()
|
|
self.collector.feed(inbound)
|
|
return True
|
|
else:
|
|
voiced_streak = 0
|
|
return False
|
|
|
|
async def _poll_frame(self) -> bytes | None:
|
|
try:
|
|
return await asyncio.wait_for(self.transport.read_frame(), timeout=0.001)
|
|
except TimeoutError:
|
|
return None
|
|
|
|
async def _collect_utterance(self) -> bytes | None:
|
|
"""None = hangup, b'' = inactivity timeout, else utterance PCM."""
|
|
deadline = time.monotonic() + self.limits.inactivity_s
|
|
while True:
|
|
timeout = deadline - time.monotonic()
|
|
if timeout <= 0:
|
|
return b""
|
|
try:
|
|
frame = await asyncio.wait_for(self.transport.read_frame(), timeout=timeout)
|
|
except TimeoutError:
|
|
return b""
|
|
if frame is None:
|
|
return None
|
|
self.recorder.add_caller(frame)
|
|
utterance = self.collector.feed(frame)
|
|
if utterance is not None:
|
|
return utterance
|
|
|
|
def _agent_closed(self, convo: AgentConversation) -> bool:
|
|
"""The closing line ends the call from our side (short calls §2.5)."""
|
|
last = next((t.text for t in reversed(convo.turns) if t.role == "assistant"), "")
|
|
lowered = last.lower()
|
|
return ("prijatno" in lowered or "doviđenja" in lowered) and bool(
|
|
convo.executor.created_request_id or convo.executor.took_message
|
|
)
|
|
|
|
async def _finalize(self, session, tenant, call, convo, reg) -> None:
|
|
try:
|
|
call.duration_s = int(time.monotonic() - self._started)
|
|
call.agent_seconds_charged = call.duration_s
|
|
call.outcome = convo.outcome
|
|
call.transcript = convo.transcript
|
|
|
|
rec_dir = get_settings().recordings_dir
|
|
os.makedirs(rec_dir, exist_ok=True)
|
|
path = os.path.join(rec_dir, f"{call.id}.wav")
|
|
with open(path, "wb") as f:
|
|
f.write(self.recorder.to_wav())
|
|
call.recording_path = path
|
|
|
|
reg.finalized = True
|
|
await record_agent_call(session, tenant, call.agent_seconds_charged)
|
|
await session.commit()
|
|
log.info(
|
|
"call %s finalized: %ss outcome=%s tenant=%s",
|
|
call.id, call.duration_s, call.outcome, tenant.slug,
|
|
)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("failed to finalize call %s", call.id)
|
|
await session.rollback()
|