Bygge en talesamtaleløkke
Opptak → transkribering → resonnering → tale, med håndtering av avbrytelser.
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 filepathStillhetsdeteksjon 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_RATEDen 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 sentenceKonseptet 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 TrueHå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.
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
- Tale-til-tekst med Whisper og Deepgram
- Tekst-til-tale i agentsvar
- Bygge en talesamtaleløkke
- Optimalisering av forsinkelse for taleagenter