How to Build
Chatbots &
Voice Agents

from setfit import SetFitModel
from openai import OpenAI

# Option A — Fine-tuned local classifier (fast, cheap at scale)
model = SetFitModel.from_pretrained("your-org/intent-classifier-v2")

def classify_intent_local(text: str) -> str:
    predictions = model.predict([text])
    return predictions[0]  # e.g. "cancel_subscription"

# Option B — LLM-based classification (flexible, no training needed)
client = OpenAI()
INTENTS = ["billing", "technical_support", "cancel", "upgrade", "other"]

def classify_intent_llm(text: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content":
                f"Classify into one of: {INTENTS}. "
                 "Respond ONLY with JSON: {{\"intent\": \"...\", \"confidence\": 0.0}}"},
            {"role": "user", "content": text}
        ]
    )
    import json
    return json.loads(resp.choices[0].message.content)["intent"]

import redis, json
from openai import OpenAI

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
client = OpenAI()

SYSTEM_PROMPT = """You are a helpful support agent for Acme Corp.
Answer using only verified facts from context. Be concise.
If you don't know, say so — never guess."""

def get_history(session_id: str) -> list:
    raw = r.get(f"session:{session_id}")
    return json.loads(raw) if raw else []

def save_history(session_id: str, history: list):
    r.setex(f"session:{session_id}", 3600, json.dumps(history))

def chat_turn(session_id: str, user_msg: str, context: str = "") -> str:
    history = get_history(session_id)
    history.append({"role": "user", "content": user_msg})

    messages = [
        {"role": "system", "content": SYSTEM_PROMPT +
            (f"\n\nContext:\n{context}" if context else "")},
        *history[-10:]  # Keep last 10 turns to control token usage
    ]

    resp = client.chat.completions.create(
        model="gpt-4o", messages=messages,
        temperature=0.3, max_tokens=600
    )
    assistant_msg = resp.choices[0].message.content
    history.append({"role": "assistant", "content": assistant_msg})
    save_history(session_id, history)
    return assistant_msg

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI

app = FastAPI()
client = OpenAI()

@app.post("/chat/stream")
async def stream_chat(session_id: str, message: str):
    async def token_generator():
        stream = client.chat.completions.create(
            model="gpt-4o", stream=True,
            messages=[{"role": "user", "content": message}]
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content or ""
            if delta:
                yield f"data: {delta}\n\n"  # SSE format
        yield "data: [DONE]\n\n"

    return StreamingResponse(token_generator(),
                              media_type="text/event-stream")

import asyncio
from deepgram import DeepgramClient, LiveOptions

dg_client = DeepgramClient(api_key="your-deepgram-key")

async def stream_stt(audio_queue: asyncio.Queue, on_transcript):
    options = LiveOptions(
        model="nova-3",
        language="en-US",
        encoding="mulaw",          # G.711 telephony codec
        sample_rate=8000,
        channels=1,
        punctuate=True,
        interim_results=True,      # Stream partials for low latency
        endpointing=300,           # ms silence = end of utterance
        smart_format=True
    )

    async with dg_client.listen.asyncwebsocket.v("1") as connection:
        await connection.start(options)

        async def send_audio():
            while True:
                audio_chunk = await audio_queue.get()
                await connection.send(audio_chunk)

        async def receive_transcript():
            async for result in connection:
                transcript = (result.channel.alternatives[0].transcript
                              if result.is_final else None)
                if transcript:
                    await on_transcript(transcript)  # trigger dialogue

        await asyncio.gather(send_audio(), receive_transcript())

import re, asyncio
from elevenlabs.client import AsyncElevenLabs

eleven = AsyncElevenLabs(api_key="your-elevenlabs-key")

SENTENCE_BOUNDARY = re.compile(r'(?<=[.!?])\s+')

async def stream_llm_to_tts(llm_stream, audio_sink):
    buffer = ""

    async for chunk in llm_stream:
        token = chunk.choices[0].delta.content or ""
        buffer += token

        # Flush on sentence boundary — don't wait for full response
        parts = SENTENCE_BOUNDARY.split(buffer)
        if len(parts) > 1:
            sentence, buffer = parts[0], " ".join(parts[1:])
            await synthesise_and_play(sentence, audio_sink)

    if buffer.strip():  # Flush remaining text
        await synthesise_and_play(buffer, audio_sink)

async def synthesise_and_play(text: str, audio_sink):
    audio_stream = await eleven.text_to_speech.stream(
        text=text,
        voice_id="your-voice-id",
        model_id="eleven_turbo_v2_5",  # <200ms first chunk
        output_format="ulaw_8000"         # telephony-compatible
    )
    async for audio_chunk in audio_stream:
        await audio_sink.write(audio_chunk)  # pipe to SIP/WebRTC

import webrtcvad, asyncio

vad = webrtcvad.Vad(aggressiveness=2)  # 0=permissive, 3=strict

class VoiceSession:
    def __init__(self):
        self.agent_speaking = False
        self.tts_task: asyncio.Task | None = None
        self.stt_queue = asyncio.Queue()

    async def on_audio_chunk(self, chunk: bytes):
        is_speech = vad.is_speech(chunk, sample_rate=8000)

        if is_speech and self.agent_speaking:
            # Barge-in: cancel current TTS, start listening
            if self.tts_task and not self.tts_task.done():
                self.tts_task.cancel()
            self.agent_speaking = False
            await self.stt_queue.put(chunk)

        elif is_speech:
            await self.stt_queue.put(chunk)  # normal speech input

    async def speak(self, text: str, audio_sink):
        self.agent_speaking = True
        self.tts_task = asyncio.create_task(
            synthesise_and_play(text, audio_sink)
        )
        try:
            await self.tts_task
        except asyncio.CancelledError:
            pass  # Barge-in cancelled — normal flow
        finally:
            self.agent_speaking = False

CHATBOT_SYSTEM = """You are a support agent for Acme Corp.
- Answer only from the provided context. Never guess or fabricate.
- Use markdown: **bold** for key terms, bullet lists for steps.
- If unsure, say: "I don't have that information — let me connect you."
- Keep responses under 150 words unless a detailed procedure is needed.
- Always cite the source document at the end: [Source: doc_name]"""

VOICE_AGENT_SYSTEM = """You are a voice assistant for Acme Corp.
CRITICAL RULES FOR VOICE OUTPUT:
- Write as spoken language only. No markdown, no bullet points, no lists.
- Use short sentences. Maximum 2 sentences per turn.
- Spell out numbers: say "twenty five dollars" not "$25".
- Use natural transitions: "Sure, let me check that for you."
- If you need to list items, use: "First... then... and finally..."
- Never start with "Great!" or "Absolutely!" — sound natural, not scripted.
- If you don't know, say: "I'm not sure about that — let me transfer you."
"""

# Tool calling — same API for chatbot and voice agent
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order_status",
            "description": "Look up a customer's order status by order ID",
            "parameters": {
                "type": "object",
                "properties": {
                    "order_id": {"type": "string",
                                  "description": "The order ID from the conversation"}
                },
                "required": ["order_id"]
            }
        }
    }
]

from fastapi import FastAPI, WebSocket, Request
from fastapi.responses import Response
import base64, json

app = FastAPI()

@app.post("/incoming-call")
async def handle_call(request: Request):
    # TwiML tells Twilio to open a media stream to our WS endpoint
    twiml = """<Response>
        <Connect>
            <Stream url="wss://your-domain.com/media-stream"/>
        </Connect>
    </Response>"""
    return Response(content=twiml, media_type="text/xml")

@app.websocket("/media-stream")
async def media_stream(ws: WebSocket):
    await ws.accept()
    session = VoiceSession()

    async def audio_sink(chunk: bytes):
        # Send TTS audio back to Twilio in base64
        payload = base64.b64encode(chunk).decode()
        await ws.send_text(json.dumps({
            "event": "media",
            "media": {"payload": payload}
        }))

    async for raw in ws.iter_text():
        msg = json.loads(raw)
        if msg["event"] == "media":
            audio = base64.b64decode(msg["media"]["payload"])
            await session.on_audio_chunk(audio)

from openai import OpenAI
from pinecone import Pinecone
import asyncio

client = OpenAI()
pc = Pinecone(api_key="your-pinecone-key")
index = pc.Index("knowledge-base")

async def retrieve_context(query: str, top_k: int = 3) -> str:
    # For voice: top_k=3 keeps context small and latency low
    embedding = client.embeddings.create(
        model="text-embedding-3-small", input=query
    ).data[0].embedding

    results = index.query(
        vector=embedding, top_k=top_k, include_metadata=True
    )
    return "\n".join(
        m['metadata']['text'] for m in results['matches']
    )

async def answer_with_rag(query: str, voice_mode: bool = False) -> str:
    # Retrieve context in parallel with any other prep work
    context = await retrieve_context(query)
    system = VOICE_AGENT_SYSTEM if voice_mode else CHATBOT_SYSTEM
    prompt = f"Context:\n{context}\n\nQuestion: {query}"

    resp = client.chat.completions.create(
        model="gpt-4o-mini" if voice_mode else "gpt-4o",
        messages=[{"role": "system", "content": system},
                  {"role": "user",   "content": prompt}],
        temperature=0.2,
        stream=voice_mode  # Stream for voice, sync for chatbot
    )
    return resp if voice_mode else resp.choices[0].message.content