AI-agenter · leksjon

Bygge en talesamtaleløkke

Opptak → transkribering → resonnering → tale, med håndtering av avbrytelser.

Leksjon 3 av 413 trinn

Bygge en talesamtaleløkke er en gratis leksjon i AI-agenter på CoddyKit. Dette er leksjon 3 av 4. Du kan lese hele leksjonen gratis nedenfor – og deretter øve praktisk i nettleseren med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i AI-agenter, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i AI-agenter inneholder totalt 4 leksjoner.

Sløyfen for talesamtaler

En fullstendig sløyfe for talesamtaler kobler mikrofoninngangen til agentens utdata i en kontinuerlig syklus: ta opp → transkriber → agent → TTS → spill av → gjenta.

Denne leksjonen dekker hver komponent og utfordringene som gjør talesløyfer annerledes enn tekstbaserte agenter: stillhetsdeteksjon, håndtering av avbrytelser og deteksjon av slutten på talen.

Opptak fra mikrofonen

Bruk sounddevice til å ta opp lyd fra standardmikrofonen. Ta opp enten i en fast tidsperiode eller til det oppdages stillhet. Ta alltid opp i 16 kHz mono – samplingsfrekvensen Whisper forventer.

Installer med pip install sounddevice soundfile numpy.

import sounddevice as sd
import numpy as np
import tempfile
import soundfile as sf

SAMPLE_RATE = 16000  # 16kHz mono — optimal for Whisper
DURATION = 5         # seconds

def record_audio(duration=DURATION, sample_rate=SAMPLE_RATE):
    print(f'Recording for {duration} seconds...')
    audio_data = sd.rec(
        int(duration * sample_rate),
        samplerate=sample_rate,
        channels=1,
        dtype='float32'
    )
    sd.wait()  # block until recording finishes
    print('Recording complete')
    return audio_data, sample_rate

def audio_to_file(audio_data, sample_rate, filepath='/tmp/recording.wav'):
    sf.write(filepath, audio_data, sample_rate)
    return filepath

Stillhetsdeteksjon for slutten av talen

Det er upraktisk å ta opp i en fast tidsperiode – brukeren må vente selv om vedkommende ble ferdig med å snakke etter 2 sekunder. Stillhetsdeteksjon stopper opptaket automatisk når brukeren har vært stille i en angitt tidsperiode.

import sounddevice as sd
import numpy as np
from collections import deque

SAMPLE_RATE = 16000
CHUNK_SIZE = 1024         # samples per chunk
SILENCE_THRESHOLD = 0.02  # RMS volume below this = silence
SILENCE_DURATION = 1.5   # seconds of silence to end recording
MAX_DURATION = 30         # max recording length in seconds

def record_until_silence():
    chunks = []
    silent_chunks = 0
    max_silent = int(SILENCE_DURATION * SAMPLE_RATE / CHUNK_SIZE)
    max_chunks = int(MAX_DURATION * SAMPLE_RATE / CHUNK_SIZE)

    print('Listening... (speak now)')
    with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
                        blocksize=CHUNK_SIZE, dtype='float32') as stream:
        while True:
            chunk, _ = stream.read(CHUNK_SIZE)
            chunks.append(chunk.copy())

            rms = np.sqrt(np.mean(chunk**2))
            if rms < SILENCE_THRESHOLD:
                silent_chunks += 1
            else:
                silent_chunks = 0  # speech detected — reset counter

            if silent_chunks >= max_silent and len(chunks) > max_silent:
                break
            if len(chunks) >= max_chunks:
                break

    audio = np.concatenate(chunks, axis=0)
    return audio, SAMPLE_RATE

Den fullstendige talesløyfen

Koble alle komponentene sammen i en kontinuerlig sløyfe. Etter at agenten har svart, begynner lyttingen igjen umiddelbart, slik at det oppstår en naturlig samtale frem og tilbake.

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

conversation_history = []

def voice_conversation_loop():
    print('Voice agent started. Say something (Ctrl+C to exit).')

    while True:
        # 1. Record user
        audio_data, sample_rate = record_until_silence()
        audio_path = audio_to_file(audio_data, sample_rate)

        # 2. Transcribe
        user_text = transcribe_file(audio_path)
        print(f'You: {user_text}')

        if not user_text.strip():
            continue  # empty — listen again

        # 3. Run agent
        conversation_history.append({'role': 'user', 'content': user_text})
        response = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'system', 'content': 'You are a helpful voice assistant. Keep responses concise.'}] + conversation_history[-10:]
        )
        agent_text = response.choices[0].message.content
        conversation_history.append({'role': 'assistant', 'content': agent_text})
        print(f'Agent: {agent_text}')

        # 4. TTS and play
        say(agent_text)  # speaks sentence by sentence

Konseptet med vekkeorddeteksjon

Kontinuerlig lytting er kostbart (kontinuerlige Whisper-kall) og et inngrep i personvernet. Vekkeorddeteksjon kjører en lett lokal modell som bare starter full behandling når en bestemt frase høres, for eksempel «Hei, agent».

# Wake word detection with pvporcupine (Picovoice)
# pip install pvporcupine sounddevice

import pvporcupine
import sounddevice as sd
import numpy as np
import os

def listen_for_wake_word(wake_word='computer'):
    porcupine = pvporcupine.create(
        access_key=os.getenv('PICOVOICE_KEY'),
        keywords=[wake_word]  # built-in: computer, hey barista, hey google, jarvis...
    )

    print(f'Listening for wake word: "{wake_word}"...')

    with sd.InputStream(
        samplerate=porcupine.sample_rate,
        channels=1,
        dtype='int16',
        blocksize=porcupine.frame_length
    ) as stream:
        while True:
            frame, _ = stream.read(porcupine.frame_length)
            pcm = frame.flatten().astype('int16')
            result = porcupine.process(pcm)
            if result >= 0:
                print(f'Wake word detected!')
                porcupine.delete()
                return True

Håndtering av avbrytelser

Brukere bør kunne avbryte agenten mens den snakker. Implementer håndtering av avbrytelser ved å kjøre avspillingen i en separat tråd og overvåke mikrofonvolumet. Hvis brukeren begynner å snakke, stopper De avspillingen umiddelbart.

import threading
import sounddevice as sd
import numpy as np

interrupt_event = threading.Event()

def monitor_for_interrupt(threshold=0.03):
    def audio_callback(indata, frames, time_info, status):
        rms = np.sqrt(np.mean(indata**2))
        if rms > threshold:
            print('Interrupt detected!')
            interrupt_event.set()

    with sd.InputStream(callback=audio_callback, channels=1,
                        samplerate=16000, blocksize=1024):
        while not interrupt_event.is_set():
            sd.sleep(50)

def speak_with_interrupt(text, voice='nova'):
    interrupt_event.clear()

    # Start interrupt monitor in background
    monitor_thread = threading.Thread(target=monitor_for_interrupt, daemon=True)
    monitor_thread.start()

    # Play audio sentence by sentence
    for sentence in split_into_sentences(text):
        if interrupt_event.is_set():
            print('Playback interrupted')
            break
        audio_path = cached_tts(sentence, voice)
        play_audio_file(audio_path)

Støyfiltrering

Mikrofonlyd inneholder ofte bakgrunnsstøy: vifter, tastaturklikk og romklang. Bruk en enkel støysperre for å dempe lyd under en terskel, og bruk eventuelt en taleaktivitetsdetektor (VAD) for bedre nøyaktighet.

import numpy as np

NOISE_GATE_THRESHOLD = 0.015  # RMS threshold

def apply_noise_gate(audio_data, threshold=NOISE_GATE_THRESHOLD):
    rms = np.sqrt(np.mean(audio_data**2))
    if rms < threshold:
        return np.zeros_like(audio_data)  # silence below threshold
    return audio_data

def has_speech_content(audio_data, threshold=0.02):
    rms_values = [
        np.sqrt(np.mean(audio_data[i:i+1600]**2))
        for i in range(0, len(audio_data), 1600)
    ]
    speech_frames = sum(1 for r in rms_values if r > threshold)
    speech_ratio = speech_frames / max(len(rms_values), 1)
    return speech_ratio > 0.1  # at least 10% of frames have speech

# Use in loop before transcribing
audio, sr = record_until_silence()
if has_speech_content(audio):
    text = transcribe_file(audio_to_file(audio, sr))
else:
    print('No speech detected — listening again')

Tilstandsmaskin for samtalesløyfen

En robust talesløyfe bruker en tilstandsmaskin til å holde oversikt over hva agenten gjør til enhver tid, slik at kappløpstilstander mellom avspilling og lytting unngås.

from enum import Enum

class AgentState(Enum):
    IDLE           = 'idle'
    LISTENING      = 'listening'
    TRANSCRIBING   = 'transcribing'
    THINKING       = 'thinking'
    SPEAKING       = 'speaking'

current_state = AgentState.IDLE

def set_state(new_state):
    global current_state
    print(f'State: {current_state.value} -> {new_state.value}')
    current_state = new_state

def voice_loop_with_state():
    while True:
        set_state(AgentState.LISTENING)
        audio, sr = record_until_silence()

        if not has_speech_content(audio):
            continue

        set_state(AgentState.TRANSCRIBING)
        text = transcribe_file(audio_to_file(audio, sr))
        if not text.strip():
            continue
        print(f'You: {text}')

        set_state(AgentState.THINKING)
        agent_response = run_agent(text)
        print(f'Agent: {agent_response}')

        set_state(AgentState.SPEAKING)
        speak_with_interrupt(agent_response)

Håndtering av samtalekontekst

Taleagenten opprettholder samtalehistorikken slik at den kan svare på oppfølgingsspørsmål som «Fortell meg mer om det» eller «Hva sa De at prisen var?» Behold historikken i minnet, og begrens den til de siste N samtaleturene for å unngå at konteksten blir for stor.

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

MAX_HISTORY_TURNS = 10
SYSTEM_PROMPT = 'You are a concise voice assistant. Keep responses under 3 sentences unless asked for detail.'

class VoiceAgent:
    def __init__(self):
        self.history = []

    def respond(self, user_text):
        self.history.append({'role': 'user', 'content': user_text})

        # Trim history to last N turns
        recent = self.history[-(MAX_HISTORY_TURNS * 2):]

        response = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=[{'role': 'system', 'content': SYSTEM_PROMPT}] + recent
        )
        agent_text = response.choices[0].message.content
        self.history.append({'role': 'assistant', 'content': agent_text})
        return agent_text

    def reset(self):
        self.history = []
        print('Conversation history cleared')

Kontrollert avslutning og opprydding

Håndter Ctrl+C og andre avslutningssignaler på en kontrollert måte: stopp lydstrømmen, rydd opp i midlertidige filer og si farvel før programmet avsluttes.

import signal
import sys
import glob
import os

TMP_DIR = '/tmp'
agent = None

def cleanup_temp_files():
    for f in glob.glob(os.path.join(TMP_DIR, 'recording_*.wav')):
        os.unlink(f)
    print('Temp files cleaned up')

def graceful_shutdown(signum, frame):
    print('\nShutting down voice agent...')
    if agent:
        say('Goodbye! Have a great day.')
    cleanup_temp_files()
    sys.exit(0)

# Register signal handler
signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)

# Main entry point
agent = VoiceAgent()
print('Voice agent ready. Press Ctrl+C to exit.')
voice_loop_with_state()

Logging av samtalen

Behold en varig logg over samtalen: tidsstempler, transkribert tekst, agentsvar og eventuelle feil. Dette er avgjørende for feilsøking av problemer med taleagenter og for å bygge funksjoner for gjennomgang av samtaler.

import json
import os
from datetime import datetime

LOG_DIR = '/tmp/voice_agent_logs'
os.makedirs(LOG_DIR, exist_ok=True)

conversation_log = []

def log_turn(speaker, text, latency_ms=None, error=None):
    entry = {
        'timestamp': datetime.now().isoformat(),
        'speaker': speaker,   # 'user' or 'agent'
        'text': text,
        'latency_ms': latency_ms
    }
    if error:
        entry['error'] = error
    conversation_log.append(entry)

def save_conversation_log(session_id=None):
    if not session_id:
        session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
    log_path = os.path.join(LOG_DIR, f'session_{session_id}.json')
    with open(log_path, 'w') as f:
        json.dump({
            'session_id': session_id,
            'turns': len(conversation_log),
            'log': conversation_log
        }, f, indent=2)
    print(f'Log saved: {log_path}')
    return log_path

# Usage in voice loop
log_turn('user', transcribed_text, latency_ms=420)
log_turn('agent', agent_response, latency_ms=850)

Kunnskapssjekk

Hva er formålet med stillhetsdeteksjon i en sløyfe for talesamtaler?

Oppsummering: Bygging av en sløyfe for talesamtaler

En sløyfe for talesamtaler: ta opp fra mikrofonen med stillhetsdeteksjon → transkriber med Whisper → kjør agenten med samtalehistorikk → TTS-svar setning for setning → spill av med overvåking av avbrytelser → gjenta.

Viktige komponenter er en tilstandsmaskin for å hindre kappløpstilstander, vekkeorddeteksjon for kontinuerlig lytting, en støysperre for lydkvalitet og begrensning av samtalehistorikken for konteksthåndtering. Håndter Ctrl+C kontrollert med opprydding og en farvelmelding.

Gratis å komme i gang

Lær deg AI-agenter med en AI-veileder – gratis

Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.

Kurs
60
Leksjoner
239

Ofte stilte spørsmål

Er leksjonen «Bygge en talesamtaleløkke» gratis?

Ja – hele teksten i «Bygge en talesamtaleløkke» er gratis å lese her på nettet. For å øve interaktivt med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt, og for å låse opp resten av AI-agenter-kurset, kan du oppgradere til CoddyKit PRO. Kurset i AI-agenter inneholder totalt 4 leksjoner.

Hva lærer jeg i «Bygge en talesamtaleløkke»?

Opptak → transkribering → resonnering → tale, med håndtering av avbrytelser. Du øver på AI-agenter med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.

Trenger jeg erfaring for å begynne med AI-agenter?

Ingen tidligere erfaring er nødvendig. AI-agenter på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 3 av 4.

Hvor lang tid tar leksjonen «Bygge en talesamtaleløkke»?

De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.

Kan jeg skrive og kjøre kode i denne AI-agenter-leksjonen?

Ja. Alle AI-agenter-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.

Alle leksjonene i dette kurset

  1. Tale-til-tekst med Whisper og Deepgram
  2. Tekst-til-tale i agentsvar
  3. Bygge en talesamtaleløkke
  4. Optimalisering av forsinkelse for taleagenter
← Tilbake til AI-agenter