Building an Agent

# === FULLY LOCAL VOICE AGENT ===

# 1. Start local LLM server (llama.cpp)
./llama-server -m models/llama-3.1-8b-instruct.gguf \
  --host 0.0.0.0 --port 8080 -c 4096 -ngl 35

# 2. Python agent script:
pip install faster-whisper piper-tts pyaudio numpy

# agent_local.py — see full code in lab exercises

# agent_local.py — Minimal local voice agent skeleton
import pyaudio, numpy as np, requests, subprocess, io, wave
from faster_whisper import WhisperModel

# Init STT
stt_model = WhisperModel("/opt/faster-whisper/models/base.en", device="cpu", compute_type="int8")

# Audio settings
RATE, CHUNK = 16000, 1024
audio = pyaudio.PyAudio()
stream = audio.open(format=pyaudio.paInt16, channels=1, rate=RATE,
                    input=True, frames_per_buffer=CHUNK)

def transcribe(audio_data):
    """Local STT with faster-whisper"""
    segments, _ = stt_model.transcribe(audio_data, beam_size=5, vad_filter=True)
    return " ".join([s.text for s in segments])

def think(text, history):
    """Local LLM via llama.cpp server (OpenAI-compatible)"""
    messages = history + [{"role": "user", "content": text}]
    resp = requests.post("http://localhost:8080/v1/chat/completions", json={
        "messages": messages, "max_tokens": 150, "stream": False
    })
    return resp.json()["choices"][0]["message"]["content"]

def speak(text):
    """Local TTS with Piper"""
    proc = subprocess.run(
        ["piper", "--model", "en_US-lessac-medium.onnx", "--output_raw"],
        input=text.encode(), capture_output=True
    )
    # Play proc.stdout as raw audio...

print("Agent ready. Speak into microphone...")
# Main loop: listen → transcribe → think → speak

# Full LiveKit agent with function calling (Agents 1.0+ API)
from livekit import agents
from livekit.agents import AgentServer, AgentSession, Agent, room_io, function_tool
from livekit.plugins import silero

class CustomerAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="""You are a customer service agent.
            Keep responses short and conversational.
            Use tools to look up accounts and transfer calls."""
        )

    @function_tool(description="Look up a customer account by phone number")
    async def lookup_account(self, phone_number: str) -> str:
        # Your database lookup logic here
        return "Account found: John Doe, Balance: $1,234.56"

    @function_tool(description="Transfer the call to a human agent")
    async def transfer_to_human(self, reason: str) -> str:
        # Trigger call transfer logic
        return f"Transferring to human agent. Reason: {reason}"

server = AgentServer()

@server.rtc_session(agent_name="customer-service")
async def entrypoint(ctx: agents.JobContext):
    session = AgentSession(
        stt="deepgram/nova-3",
        llm="openai/gpt-4.1-mini",
        tts="cartesia/sonic-3",
        vad=silero.VAD.load(),
    )
    await session.start(room=ctx.room, agent=CustomerAgent())
    await session.generate_reply(
        instructions="Thank the caller and ask how you can help."
    )

if __name__ == "__main__":
    agents.cli.run_app(server)