TELEPHONE NETWORK
                                |
                           SIP Trunk
                                |
                    +-----------v-----------+
                    |      ASTERISK PBX      |
                    |                        |
                    |  1. Answer()           |
                    |  2. AGI(setup.agi)     |
                    |     - Generate UUID    |
                    |     - Write metadata   |
                    |  3. AudioSocket(       |
                    |       127.0.0.1:9099)  |
                    +-----------+------------+
                                |
                    TCP (AudioSocket Protocol)
                    8kHz 16-bit PCM, 20ms frames
                                |
                    +-----------v------------+
                    |   PYTHON VOICE AGENT    |
                    |   (asyncio TCP server)  |
                    |                         |
                    |  +-------------------+  |
                    |  | Audio Reader      |  |    +------------------+
          Caller    |  | - Read PCM frames |--------> Deepgram Nova-3 |
          speaks    |  | - Barge-in VAD    |  |    | Streaming STT    |
                    |  +-------------------+  |    | (WebSocket)      |
                    |                         |    +--------+---------+
                    |                         |             |
                    |                         |        Transcript
                    |                         |             |
                    |  +-------------------+  |    +--------v---------+
                    |  | Conversation      |  |    | Groq Llama 3.3   |
                    |  | Manager           |<------| 70B specdec      |
                    |  | - State machine   |  |    | Streaming LLM    |
                    |  | - Message history |  |    | (HTTP SSE)       |
                    |  | - Tool calls      |  |    +--------+---------+
                    |  +-------------------+  |             |
                    |                         |       Token stream
                    |                         |             |
                    |  +-------------------+  |    +--------v---------+
          Caller    |  | Audio Writer      |  |    | Cartesia Sonic-3 |
          hears  <-----| - Queue playback  |<------| Streaming TTS    |
                    |  | - 20ms pacing     |  |    | (WebSocket)      |
                    |  +-------------------+  |    +------------------+
                    |                         |
                    +-------------------------+

    Latency Budget (target: <250ms mouth-to-ear):
    ┌──────────────────────────────────────────────────────┐
    │ Deepgram STT final transcript:    ~150-200ms        │
    │ Groq LLM first token (TTFT):     ~30-50ms          │
    │ Cartesia TTS first audio (TTFB): ~50-80ms          │
    │ AudioSocket frame transmission:  ~20ms (1 frame)   │
    │                                                      │
    │ TOTAL:                            ~200-250ms        │
    └──────────────────────────────────────────────────────┘

pip install websockets aiohttp cartesia

asterisk -rx "module show like audiosocket"

Module              Description                       Use Count  Status
res_audiosocket.so  AudioSocket support               0          Running
app_audiosocket.so  AudioSocket application           0          Running
2 modules loaded

asterisk -rx "module load res_audiosocket.so"
asterisk -rx "module load app_audiosocket.so"

load = res_audiosocket.so
load = app_audiosocket.so

+--------+--------+--------+--- ... ---+
| Type   | Length (big-endian)| Payload  |
| 1 byte | 2 bytes           | N bytes  |
+--------+--------+--------+--- ... ---+

; ─── AI Voice Agent ─────────────────────────────────────────────
; Route inbound DID calls to the AI voice agent.
; The CALLED variable must be set by the inbound trunk context
; before transferring here (e.g., Set(CALLED=${EXTEN})).

[voice-agent]
exten => voice_agent,1,NoOp(VOICE AGENT: DID=${CALLED} CLI=${CALLERID(num)})
same => n,Answer()
same => n,AGI(voice_agent_setup.agi)
same => n,AudioSocket(${VA_UUID},127.0.0.1:9099)
same => n,NoOp(VOICE AGENT ENDED: ${DIALSTATUS})
same => n,Hangup()

[from-trunk]
; Match all inbound DIDs and route to voice agent
exten => _X.,1,NoOp(Inbound call to ${EXTEN} from ${CALLERID(num)})
same => n,Set(CALLED=${EXTEN})
same => n,Goto(voice-agent,voice_agent,1)

[from-trunk]
; Route specific DIDs to voice agent, others to normal handling
exten => 02012345678,1,Set(CALLED=${EXTEN})
same => n,Goto(voice-agent,voice_agent,1)

exten => 02087654321,1,Set(CALLED=${EXTEN})
same => n,Goto(voice-agent,voice_agent,1)

; Default: normal call handling
exten => _X.,1,Goto(default,${EXTEN},1)

#!/usr/bin/perl
# voice_agent_setup.agi
# Generate UUID and write metadata JSON for the voice agent.
# Called by Asterisk dialplan before AudioSocket().
use strict;
use warnings;
$| = 1;  # autoflush STDOUT -- critical for AGI protocol

# ── Read AGI environment variables ──
my %agi;
while (<STDIN>) {
    chomp;
    last if /^$/;
    if (/^agi_(\w+):\s*(.*)$/) {
        $agi{$1} = $2;
    }
}

# ── Generate UUID ──
# Use the kernel's random UUID generator (no external dependencies)
my $uuid = '';
if (open my $fh, '<', '/proc/sys/kernel/random/uuid') {
    $uuid = <$fh>;
    chomp $uuid;
    close $fh;
}

# ── Get call metadata ──
# CALLED is set in the dialplan before calling AGI
print "GET VARIABLE CALLED\n";
my $resp = <STDIN>;
my $did = '';
if ($resp =~ /\((.+)\)/) {
    $did = $1;
}
my $cli = $agi{'callerid'} || '';

# ── Write metadata JSON ──
if ($uuid) {
    my $file = "/tmp/va_${uuid}.json";
    if (open my $fh, '>', $file) {
        print $fh qq({"did":"$did","cli":"$cli"});
        close $fh;
    }
}

# ── Set channel variable for AudioSocket ──
print "SET VARIABLE VA_UUID $uuid\n";
my $result = <STDIN>;

chmod +x /var/lib/asterisk/agi-bin/voice_agent_setup.agi

#!/usr/bin/env python3.11
"""
Voice Agent -- AudioSocket + Deepgram STT + Groq LLM + Cartesia TTS
Token-streaming pipeline: ~200-250ms mouth-to-ear latency.
"""

import asyncio
import struct
import json
import os
import time
import logging
import audioop
from pathlib import Path

import aiohttp
from websockets.asyncio.client import connect as ws_connect
from cartesia import AsyncCartesia

# ── Configuration ──
LISTEN_HOST = "127.0.0.1"
LISTEN_PORT = 9099

# Audio: 8kHz 16-bit signed linear PCM, mono
# 320 bytes = 160 samples = 20ms per frame
CHUNK_SIZE = 320
SAMPLE_RATE = 8000
CHUNK_DURATION = CHUNK_SIZE / (SAMPLE_RATE * 2)  # 0.02s = 20ms

# Barge-in thresholds
BARGEIN_RMS_THRESHOLD = 800   # RMS energy level to detect speech
BARGEIN_DURATION = 0.3        # seconds of sustained speech to trigger

log = logging.getLogger("voice-agent")


class DeepgramSTT:
    """Streaming speech-to-text via Deepgram Nova-3 WebSocket."""
    ...

class GroqLLM:
    """Streaming LLM via Groq OpenAI-compatible API."""
    ...

class CartesiaTTS:
    """Streaming TTS via Cartesia Sonic-3 WebSocket."""
    ...

class VoiceAgent:
    """Main orchestrator: STT -> LLM -> TTS pipeline."""
    ...


async def handle_connection(reader, writer):
    """Handle one AudioSocket connection (one call)."""
    agent = VoiceAgent()
    await agent.handle_call(reader, writer)


async def main():
    server = await asyncio.start_server(
        handle_connection, LISTEN_HOST, LISTEN_PORT,
    )
    log.info("Voice Agent listening on %s:%d", LISTEN_HOST, LISTEN_PORT)
    async with server:
        await server.serve_forever()


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(name)s %(levelname)s %(message)s",
    )
    asyncio.run(main())

class VoiceAgent:
    def __init__(self):
        self.stt = DeepgramSTT()
        self.llm = GroqLLM()
        self.tts = CartesiaTTS()
        self.messages = []           # LLM conversation history
        self.call_context = None     # Company/DID context
        self.audio_out_queue = asyncio.Queue()  # PCM chunks to send to caller
        self.hangup_event = asyncio.Event()     # Signals call ended
        self.is_speaking = asyncio.Event()      # True while agent is talking
        self.barge_in_event = asyncio.Event()   # Caller interrupted
        self.call_uuid = None
        self._current_tts_ctx = None            # For barge-in cancellation

    async def handle_call(self, reader, writer):
        """Main call handler -- the entire call lifecycle."""
        # 1. Read UUID frame from AudioSocket
        # 2. Read metadata from /tmp/va_{uuid}.json
        # 3. Fetch company context from API
        # 4. Build system prompt
        # 5. Connect STT + TTS WebSockets
        # 6. Start background audio I/O tasks
        # 7. Speak greeting
        # 8. Conversation loop: listen -> think -> speak
        ...

# AudioSocket protocol constants
AS_TYPE_HANGUP = 0x00
AS_TYPE_UUID   = 0x01
AS_TYPE_AUDIO  = 0x10
AS_TYPE_ERROR  = 0xFF


async def read_as_frame(reader):
    """Read one AudioSocket frame: 1 byte type + 2 bytes length + payload."""
    header = await reader.readexactly(3)
    frame_type = header[0]
    length = struct.unpack(">H", header[1:3])[0]
    payload = b""
    if length > 0:
        payload = await reader.readexactly(length)
    return frame_type, payload

def make_as_frame(frame_type, payload):
    """Build an AudioSocket frame."""
    return struct.pack(">BH", frame_type, len(payload)) + payload

async def _audio_reader(self, reader):
    """Read audio from AudioSocket, forward to STT or check barge-in."""
    speech_energy_start = None

    while not self.hangup_event.is_set():
        frame_type, payload = await read_as_frame(reader)

        if frame_type == AS_TYPE_HANGUP:
            self.hangup_event.set()
            return

        if frame_type == AS_TYPE_ERROR:
            self.hangup_event.set()
            return

        if frame_type == AS_TYPE_AUDIO and payload:
            if self.is_speaking.is_set():
                # Agent is talking -- check for barge-in
                rms = audioop.rms(payload, 2)
                if rms > BARGEIN_RMS_THRESHOLD:
                    if speech_energy_start is None:
                        speech_energy_start = time.monotonic()
                    elif time.monotonic() - speech_energy_start >= BARGEIN_DURATION:
                        # Caller is interrupting
                        self.barge_in_event.set()
                        speech_energy_start = None
                        # Clear audio queue, cancel TTS
                        while not self.audio_out_queue.empty():
                            self.audio_out_queue.get_nowait()
                        if self._current_tts_ctx:
                            await self.tts.cancel_context(self._current_tts_ctx)
                        self.is_speaking.clear()
                        await self.stt.send_audio(payload)
                else:
                    speech_energy_start = None
            else:
                # Agent is silent -- forward to STT
                speech_energy_start = None
                await self.stt.send_audio(payload)

async def _audio_writer(self, writer):
    """Write queued audio to AudioSocket at real-time rate."""
    while not self.hangup_event.is_set():
        try:
            chunk = await asyncio.wait_for(
                self.audio_out_queue.get(), timeout=0.5
            )
            frame = make_as_frame(AS_TYPE_AUDIO, chunk)
            writer.write(frame)
            await writer.drain()
            await asyncio.sleep(CHUNK_DURATION)  # 20ms pacing
        except asyncio.TimeoutError:
            continue

class DeepgramSTT:
    def __init__(self):
        self.ws = None
        self.transcript_queue = asyncio.Queue()
        self._recv_task = None
        self._keepalive_task = None

    async def connect(self):
        params = (
            "encoding=linear16"
            "&sample_rate=8000"
            "&channels=1"
            "&model=nova-3"
            "&language=en-GB"
            "&smart_format=true"
            "&endpointing=300"
            "&interim_results=true"
            "&utterance_end_ms=1000"
            "&vad_events=true"
            "&keywords=postcode:2"
            "&keywords=plumber:2"
            "&keywords=callout:1"
        )
        url = f"wss://api.deepgram.com/v1/listen?{params}"
        headers = {"Authorization": f"Token {DEEPGRAM_API_KEY}"}

        self.ws = await ws_connect(
            url, additional_headers=headers,
            ping_interval=5, ping_timeout=10,
        )
        self._recv_task = asyncio.create_task(self._receive_loop())
        self._keepalive_task = asyncio.create_task(self._keepalive_loop())

async def _receive_loop(self):
    """Process transcript results from Deepgram."""
    async for msg in self.ws:
        data = json.loads(msg)

        if data.get("type") == "Results":
            alt = data["channel"]["alternatives"][0]
            transcript = alt.get("transcript", "").strip()
            is_final = data.get("is_final", False)
            speech_final = data.get("speech_final", False)

            if transcript and is_final:
                await self.transcript_queue.put({
                    "transcript": transcript,
                    "speech_final": speech_final,
                })

        elif data.get("type") == "UtteranceEnd":
            # Safety flush: if we have accumulated fragments, process them
            await self.transcript_queue.put({
                "transcript": "",
                "speech_final": True,
                "utterance_end": True,
            })

async def _keepalive_loop(self):
    while self.ws:
        await asyncio.sleep(8)
        await self.ws.send(json.dumps({"type": "KeepAlive"}))

async def send_audio(self, audio_data):
    """Forward raw PCM bytes to Deepgram."""
    if self.ws:
        await self.ws.send(audio_data)

async def close(self):
    if self._keepalive_task:
        self._keepalive_task.cancel()
    if self._recv_task:
        self._recv_task.cancel()
    if self.ws:
        await self.ws.send(json.dumps({"type": "CloseStream"}))
        await self.ws.close()

class GroqLLM:
    def __init__(self):
        self.session = None

    async def _ensure_session(self):
        if not self.session:
            self.session = aiohttp.ClientSession()

    async def generate(self, messages):
        """Stream completion tokens. Yields dicts with type 'text' or 'tool_call'."""
        await self._ensure_session()

        headers = {
            "Authorization": f"Bearer {GROQ_API_KEY}",
            "Content-Type": "application/json",
        }
        payload = {
            "model": GROQ_MODEL,
            "messages": messages,
            "tools": [BOOKING_TOOL],
            "stream": True,
            "temperature": 0.3,
            "max_tokens": 250,
        }

        tool_calls = {}
        t0 = time.monotonic()
        first_token = True

        async with self.session.post(
            "https://api.groq.com/openai/v1/chat/completions",
            headers=headers, json=payload,
        ) as resp:
            if resp.status != 200:
                body = await resp.text()
                log.error("Groq API error %d: %s", resp.status, body[:500])
                yield {"type": "text", "text": "Bear with me, I'm having a small technical issue."}
                return

            buffer = ""
            async for raw_line in resp.content:
                buffer += raw_line.decode()
                while "\n" in buffer:
                    line, buffer = buffer.split("\n", 1)
                    line = line.strip()
                    if not line or not line.startswith("data: "):
                        continue
                    if line == "data: [DONE]":
                        break

                    data = json.loads(line[6:])
                    choice = data["choices"][0]
                    delta = choice.get("delta", {})

                    # Text content -- yield immediately for token-streaming
                    content = delta.get("content")
                    if content:
                        if first_token:
                            log.info("LLM TTFT: %.0fms",
                                     (time.monotonic() - t0) * 1000)
                            first_token = False
                        yield {"type": "text", "text": content}

                    # Tool calls -- accumulate arguments across chunks
                    if "tool_calls" in delta:
                        for tc in delta["tool_calls"]:
                            idx = tc["index"]
                            if idx not in tool_calls:
                                tool_calls[idx] = {
                                    "id": tc.get("id", ""),
                                    "name": tc["function"]["name"],
                                    "arguments": "",
                                }
                            tool_calls[idx]["arguments"] += (
                                tc["function"].get("arguments", "")
                            )

                    # Tool call complete
                    if choice.get("finish_reason") == "tool_calls":
                        for tc in tool_calls.values():
                            yield {
                                "type": "tool_call",
                                "id": tc["id"],
                                "name": tc["name"],
                                "arguments": json.loads(tc["arguments"]),
                            }

class CartesiaTTS:
    def __init__(self):
        self.client = None
        self.connection = None

    async def connect(self):
        """Open persistent WebSocket (reused across utterances)."""
        self.client = AsyncCartesia(api_key=CARTESIA_API_KEY)
        self.connection = await self.client.tts.websocket_connect().__aenter__()
        log.info("Cartesia TTS connected")

async def synthesize_streaming(self, text, audio_out_queue):
    """Send complete text, stream audio chunks to queue."""
    ctx = self.connection.context()

    await ctx.send(
        model_id="sonic-3",
        transcript=text,
        voice={"mode": "id", "id": CARTESIA_VOICE_ID},
        output_format={
            "container": "raw",
            "encoding": "pcm_s16le",
            "sample_rate": 8000,
        },
        continue_=False,  # Complete utterance
    )

    async for response in ctx.receive():
        if response.type == "chunk" and response.audio:
            pcm_bytes = response.audio
            # Split into 320-byte chunks (20ms frames)
            for i in range(0, len(pcm_bytes), CHUNK_SIZE):
                chunk = pcm_bytes[i:i + CHUNK_SIZE]
                if len(chunk) < CHUNK_SIZE:
                    chunk += b'\x00' * (CHUNK_SIZE - len(chunk))
                await audio_out_queue.put(chunk)

async def stream_tokens(self, audio_out_queue, cancel_event):
    """Create a token-streaming TTS context.

    Returns (ctx, receive_task) -- the caller pushes tokens into ctx,
    and receive_task writes audio to audio_out_queue in the background.
    """
    ctx = self.connection.context()
    recv_task = asyncio.create_task(
        self._receive_audio(ctx, audio_out_queue, cancel_event)
    )
    return ctx, recv_task

async def _receive_audio(self, ctx, audio_out_queue, cancel_event):
    """Background: receive audio from Cartesia, chunk and queue."""
    first_audio = True
    t0 = time.monotonic()

    async for response in ctx.receive():
        if cancel_event.is_set():
            break

        if response.type == "chunk" and response.audio:
            if first_audio:
                log.info("TTS TTFB: %.0fms", (time.monotonic() - t0) * 1000)
                first_audio = False

            pcm_bytes = response.audio
            for i in range(0, len(pcm_bytes), CHUNK_SIZE):
                if cancel_event.is_set():
                    return
                chunk = pcm_bytes[i:i + CHUNK_SIZE]
                if len(chunk) < CHUNK_SIZE:
                    chunk += b'\x00' * (CHUNK_SIZE - len(chunk))
                await audio_out_queue.put(chunk)

# Each LLM token is sent immediately to Cartesia:
await ctx.send(
    model_id="sonic-3",
    transcript=token,          # e.g., "I", " can", " get"
    voice={"mode": "id", "id": VOICE_ID},
    output_format={...},
    continue_=True,            # <-- This is the key
)

# When the LLM finishes:
await ctx.no_more_inputs()

async def cancel_context(self, ctx):
    """Cancel in-progress TTS for barge-in."""
    await self.connection.send({
        "context_id": ctx._context_id,
        "cancel": True,
    })

async def _think_and_speak(self):
    """Stream LLM tokens directly to Cartesia TTS.
    No sentence-boundary detection -- tokens flow straight through."""
    full_response = []
    t0 = time.monotonic()
    self.barge_in_event.clear()

    # Create a TTS streaming context
    cancel_event = asyncio.Event()
    ctx, recv_task = await self.tts.stream_tokens(
        self.audio_out_queue, cancel_event
    )
    self._current_tts_ctx = ctx
    self.is_speaking.set()

    tts_voice = {"mode": "id", "id": CARTESIA_VOICE_ID}
    tts_format = {
        "container": "raw",
        "encoding": "pcm_s16le",
        "sample_rate": SAMPLE_RATE,
    }

    try:
        # Stream LLM tokens directly to TTS
        async for event in self.llm.generate(self.messages):
            if self.barge_in_event.is_set():
                log.info("Barge-in: aborting stream")
                cancel_event.set()
                break

            if event["type"] == "text":
                token = event["text"]
                full_response.append(token)

                # Send token directly to TTS (continuation mode)
                await ctx.send(
                    model_id=CARTESIA_MODEL,
                    transcript=token,
                    voice=tts_voice,
                    output_format=tts_format,
                    continue_=True,
                )

            elif event["type"] == "tool_call":
                # Handle tool call (see Section 15)
                ...

        # LLM finished -- flush TTS
        if not self.barge_in_event.is_set():
            await ctx.no_more_inputs()
            await recv_task  # Wait for all audio to be received

            # Wait for audio queue to drain (all chunks played)
            while not self.audio_out_queue.empty():
                await asyncio.sleep(0.02)

    finally:
        self.is_speaking.clear()
        self._current_tts_ctx = None
        await asyncio.sleep(0.1)  # Brief pause before listening

    return "".join(full_response)

User speaks -> STT -> [wait for full transcript] ->
LLM -> [wait for full response] ->
TTS -> [wait for full audio] ->
Play to user

Total: 1.5 - 3.0 seconds

User speaks -> STT (streaming) ->
             |-> LLM token 1 -> TTS -> Audio chunk 1 -> Play
             |-> LLM token 2 -> TTS -> Audio chunk 2 -> Play
             |-> LLM token 3 -> TTS -> Audio chunk 3 -> Play
             ...

Total: 200-250ms to first audio

Timeline:
  Agent speaking:  "I can get a plumber out to you within---"
  Caller speaks:                                  "yeah how much?"
  Agent detects:                                     [RMS > 800 for 300ms]
  Agent stops:                                       [audio queue cleared]
  Agent listens:                                     "yeah how much?"
  Agent responds:                                                    "There's a 49 pound callout..."

# Constants
BARGEIN_RMS_THRESHOLD = 800   # Energy threshold (0-32768 range for 16-bit audio)
BARGEIN_DURATION = 0.3        # Sustained speech duration in seconds

rms = audioop.rms(payload, 2)  # Calculate RMS of 20ms frame

if rms > BARGEIN_RMS_THRESHOLD:
    if speech_energy_start is None:
        speech_energy_start = time.monotonic()
    elif time.monotonic() - speech_energy_start >= BARGEIN_DURATION:
        # Confirmed barge-in
        self.barge_in_event.set()
        speech_energy_start = None

        # 1. Clear the audio output queue (stop pending playback)
        while not self.audio_out_queue.empty():
            self.audio_out_queue.get_nowait()

        # 2. Cancel the active TTS context
        if self._current_tts_ctx:
            await self.tts.cancel_context(self._current_tts_ctx)

        # 3. Resume STT (start listening again)
        self.is_speaking.clear()
        await self.stt.send_audio(payload)
else:
    speech_energy_start = None  # Reset if energy drops below threshold

if rms > 200:  # Log anything above ambient noise
    log.debug("RMS during speech: %d", rms)

Step 1: GREET
  "Hello, [Company Name], good [morning/afternoon/evening]."
  (Then silence -- wait for caller to state their problem)
       |
       v
Step 2: UNDERSTAND
  Caller states problem. Agent asks ONE follow-up if needed.
  "Is it leaking now?" / "Whole house or just one room?"
       |
       v
Step 3: QUOTE
  "I can get a [trade] out to you within thirty minutes to an hour.
   There's a [fee] pound callout. He'll quote on-site before starting."
  (Wait for agreement)
       |
       v
Step 4: POSTCODE
  "What's your postcode?"
  (Confirm with phonetic alphabet)
       |
       v
Step 5: ADDRESS
  "And the house number and street?"
       |
       v
Step 6: NAME
  "And your name?"
       |
       v
Step 7: BOOK
  [Tool call: create_booking]
  (Automatic -- no user-facing output)
       |
       v
Step 8: CONFIRM
  "That's booked. The [trade] will be with you within the hour.
   Thanks for calling."

SYSTEM_PROMPT_TEMPLATE = """You work at {company_name}, a UK {trade_type} \
company. You answer the phone. Casual British English -- "no worries", \
"lovely", "bear with me". Short replies, 1 sentence max. Never sound scripted.

Never use American English. Never say "sir", "madam", "dear" -- you don't \
know who's calling. Never say "How can I help you?" -- they're calling you, \
they'll tell you.

# Context
- Company: {company_name}
- Trade: {trade_type} / {trade_label}
- Callout: {callout_fee}
- Caller: {caller_id}
- Repeat: {is_repeat}

# Workflow -- follow this exact order, one step per reply

Step 1: Greet. Say "Hello, {company_name}, good [morning/afternoon/evening]."
Then shut up and listen.

Step 2: They tell you the problem. Ask ONE quick follow-up if needed.
Then move on. Don't interrogate.

Step 3: Quote. "I can get a {trade_label} out to you within thirty minutes
to an hour. There's a {callout_fee} pound callout for the {trade_label} to
come out, then he'll quote you on-site before starting anything."
Stop here. Wait for agreement.

Step 4: Postcode. "What's your postcode?" Confirm with phonetic alphabet.

Step 5: Address. "And the house number and street?"

Step 6: Name. "And your name?"

Step 7: Book using create_booking tool. Use caller_id from context -- never
ask for phone number.

Step 8: "That's booked. The {trade_label} will be with you within the hour.
Thanks for calling."

# NEVER do
- Never combine steps -- one thing per reply
- Never ask for phone number
- Never repeat their name back
- Never summarise at the end
- Never quote labour -- only callout fee
- Never give exact time -- "thirty minutes to an hour"
- Never say "call centre", "AI", or system names

# Objections
- Want total price: "Every job's different -- the {trade_label} will quote
  on-site before starting. No obligation."
- Too expensive: "No worries. Thanks for calling."
- Are you local: "Yes, we've got someone nearby."
- Abusive: "I'm not able to help further on this call."
"""

import datetime

hour = datetime.datetime.now().hour
tod = "good morning" if hour < 12 else "good afternoon" if hour < 18 else "good evening"
company = self.call_context.get("company_name", "Home Services")
greeting = f"Hello, {company}, {tod}."

await self._speak(greeting)
self.messages.append({"role": "assistant", "content": greeting})

BOOKING_TOOL = {
    "type": "function",
    "function": {
        "name": "create_booking",
        "description": (
            "Create a job booking after collecting customer name, postcode, "
            "address, and problem description. Call this ONLY after all "
            "details are collected."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "customer_name": {
                    "type": "string",
                    "description": "Customer full name",
                },
                "postcode": {
                    "type": "string",
                    "description": "UK postcode with space",
                },
                "address": {
                    "type": "string",
                    "description": "Full street address",
                },
                "problem_description": {
                    "type": "string",
                    "description": "One-line summary of the issue",
                },
            },
            "required": [
                "customer_name", "postcode",
                "address", "problem_description",
            ],
        },
    },
}

# Inside _think_and_speak:
elif event["type"] == "tool_call":
    # 1. Close current TTS context (no more text to speak)
    await ctx.no_more_inputs()
    await recv_task

    self.is_speaking.clear()
    self._current_tts_ctx = None

    if event["name"] == "create_booking":
        # 2. Call the booking API
        result = await self.create_booking(event["arguments"])

        # 3. Add tool call + result to message history
        self.messages.append({
            "role": "assistant",
            "content": None,
            "tool_calls": [{
                "id": event["id"],
                "type": "function",
                "function": {
                    "name": "create_booking",
                    "arguments": json.dumps(event["arguments"]),
                },
            }],
        })
        self.messages.append({
            "role": "tool",
            "tool_call_id": event["id"],
            "content": json.dumps(result or {"success": True}),
        })

        # 4. Get confirmation response from LLM
        confirm_parts = []
        async for ev in self.llm.generate(self.messages):
            if ev["type"] == "text":
                confirm_parts.append(ev["text"])

        confirmation = "".join(confirm_parts)
        if confirmation.strip():
            await self._speak(confirmation.strip())

<?php
// create_booking.php -- Receive booking from voice agent
header('Content-Type: application/json');

// Verify API key
$api_key = $_SERVER['HTTP_X_API_KEY'] ?? '';
if ($api_key !== getenv('VOICE_AGENT_API_KEY')) {
    http_response_code(401);
    echo json_encode(['error' => 'Unauthorized']);
    exit;
}

$data = json_decode(file_get_contents('php://input'), true);

// Validate required fields
$required = ['customer_name', 'postcode', 'address', 'problem_description'];
foreach ($required as $field) {
    if (empty($data[$field])) {
        http_response_code(400);
        echo json_encode(['error' => "Missing field: $field"]);
        exit;
    }
}

// Insert into database, send notifications, etc.
// ... your business logic here ...

echo json_encode([
    'success' => true,
    'booking_id' => uniqid('BK-'),
    'message' => 'Booking created successfully',
]);

CHECK_AVAILABILITY_TOOL = {
    "type": "function",
    "function": {
        "name": "check_availability",
        "description": "Check engineer availability for a given postcode area",
        "parameters": {
            "type": "object",
            "properties": {
                "postcode": {"type": "string"},
                "urgency": {"type": "string", "enum": ["emergency", "standard"]},
            },
            "required": ["postcode"],
        },
    },
}

# In the LLM payload:
"tools": [BOOKING_TOOL, CHECK_AVAILABILITY_TOOL],

<?php
// did_context.php -- Return company context for a DID
header('Content-Type: application/json');

$api_key = $_SERVER['HTTP_X_API_KEY'] ?? '';
if ($api_key !== getenv('VOICE_AGENT_API_KEY')) {
    http_response_code(401);
    echo json_encode(['error' => 'Unauthorized']);
    exit;
}

$data = json_decode(file_get_contents('php://input'), true);
$did = $data['did_number'] ?? '';
$cli = $data['caller_id'] ?? '';

// Look up company by DID
// This could be a database query, config file, etc.
$companies = [
    '02012345678' => [
        'company_name' => 'Quick Fix Plumbing',
        'trade_type'   => 'plumbing',
        'trade_label'  => 'plumber',
        'callout_fee'  => 49,
        'area'         => 'London',
    ],
    '01234567890' => [
        'company_name' => 'Spark Electrical',
        'trade_type'   => 'electrical',
        'trade_label'  => 'electrician',
        'callout_fee'  => 59,
        'area'         => 'Manchester',
    ],
];

$company = $companies[$did] ?? [
    'company_name' => 'Home Services',
    'trade_type'   => 'general maintenance',
    'trade_label'  => 'engineer',
    'callout_fee'  => 49,
    'area'         => 'your area',
];

// Check if repeat caller
$is_repeat = false;
// ... query your database for previous calls from $cli ...

$company['caller_id'] = $cli;
$company['is_repeat'] = $is_repeat;
$company['did_number'] = $did;
$company['greeting'] = "Hello, {$company['company_name']}, how can I help you?";

echo json_encode($company);

async def get_call_context(self, did, cli):
    """Fetch company context from our API."""
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(
                CONTEXT_API_URL,
                json={"did_number": did, "caller_id": cli},
                headers={"X-API-Key": CONTEXT_API_KEY},
                timeout=aiohttp.ClientTimeout(total=3),
            ) as resp:
                if resp.status == 200:
                    return await resp.json()
    except Exception as e:
        log.warning("Context API error: %s", e)

    # Fallback defaults if API fails
    return {
        "company_name": "Home Services",
        "trade_type": "plumbing",
        "trade_label": "plumber",
        "callout_fee": 49,
        "caller_id": cli,
        "is_repeat": False,
        "greeting": "Hello, how can I help you?",
    }

# Voice Agent API Keys -- chmod 600 this file
GROQ_API_KEY="gsk_YOUR_GROQ_API_KEY_HERE"
GROQ_MODEL="llama-3.3-70b-specdec"
DEEPGRAM_API_KEY="YOUR_DEEPGRAM_API_KEY_HERE"
CARTESIA_API_KEY="sk_car_YOUR_CARTESIA_KEY_HERE"
CARTESIA_VOICE_ID="a01c369f-6d2d-4185-bc20-b32c225eab70"
CARTESIA_MODEL="sonic-3"
CONTEXT_API_URL="http://127.0.0.1/api/did_context.php"
CONTEXT_API_KEY="YOUR_INTERNAL_API_KEY_HERE"
BOOKING_API_URL="http://127.0.0.1/api/create_booking.php"

chmod 600 /opt/voice-agent/voice_agent.env

[Unit]
Description=AI Voice Agent (AudioSocket + Deepgram STT + Groq LLM + Cartesia TTS)
After=network.target asterisk.service

[Service]
Type=simple
ExecStart=/usr/bin/python3.11 /opt/voice-agent/agent.py
WorkingDirectory=/opt/voice-agent
Restart=always
RestartSec=3
StandardOutput=journal
StandardError=journal
SyslogIdentifier=voice-agent

[Install]
WantedBy=multi-user.target

systemctl daemon-reload
systemctl enable voice-agent
systemctl start voice-agent

systemctl status voice-agent

● voice-agent.service - AI Voice Agent (AudioSocket + Deepgram STT + Groq LLM + Cartesia TTS)
   Loaded: loaded (/etc/systemd/system/voice-agent.service; enabled)
   Active: active (running) since ...

# Live logs
journalctl -u voice-agent -f

# Last 100 lines
journalctl -u voice-agent -n 100

systemctl restart voice-agent

LLM generates "I can get a plumber out to you."  -> 400ms
TTS synthesizes entire sentence                   -> 300ms
Total: 700ms before caller hears anything

LLM token "I"     -> TTS starts synthesizing     -> 80ms
LLM token " can"  -> TTS continues
LLM token " get"  -> Caller already hearing "I can..."
Total: ~80ms before caller hears first syllable

endpointing=500  (default) -- 500ms of silence before finalizing
endpointing=300  (our setting) -- 300ms, more responsive
endpointing=150  -- Very aggressive, may cut off mid-sentence

# Connect once at call start
await self.tts.connect()

# Reuse for every utterance
ctx = self.connection.context()  # New context, same connection

# BAD: Receive 24kHz, resample to 8kHz
import audioop
audio_8k = audioop.ratecv(audio_24k, 2, 1, 24000, 8000, None)[0]

# GOOD: Request 8kHz natively from Cartesia
output_format={
    "container": "raw",
    "encoding": "pcm_s16le",
    "sample_rate": 8000,  # Native 8kHz -- no resampling needed
}

# Connect in parallel
await asyncio.gather(
    self.stt.connect(),
    self.tts.connect(),
)
# Both ready -- no connection delay on first interaction

output_format={
    "container": "raw",
    "encoding": "pcm_s16le",    # Signed 16-bit little-endian
    "sample_rate": 8000,         # 8kHz, not 16kHz or 24kHz
}

await asyncio.sleep(CHUNK_DURATION)  # 0.02 seconds

await self.ws.send(json.dumps({"type": "KeepAlive"}))

# Check if agent is listening
ss -tlnp | grep 9099

# Check service status
systemctl status voice-agent

# Check logs for startup errors
journalctl -u voice-agent -n 50

log.debug("Barge-in check: RMS=%d threshold=%d", rms, BARGEIN_RMS_THRESHOLD)

# Test the context API directly
curl -X POST http://127.0.0.1/api/did_context.php \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"did_number":"02012345678","caller_id":"07700900000"}'

# In _audio_reader:
if self.recording_file:
    self.recording_file.write(payload)  # Caller's audio

# In _audio_writer:
if self.recording_file:
    self.recording_file.write(chunk)     # Agent's audio

import signal

def handle_shutdown(signum, frame):
    log.info("Shutdown signal received")
    # The asyncio server will stop accepting new connections
    # Active calls will finish naturally
    raise SystemExit(0)

signal.signal(signal.SIGTERM, handle_shutdown)

log.info("CALL_METRICS uuid=%s duration=%.1f turns=%d bargein_count=%d "
         "avg_latency_ms=%d booking=%s",
         self.call_uuid, duration, turn_count, bargein_count,
         avg_latency, booking_created)