LangChain ile Sesli Asistan Geliştirme: Konuşmadan Yanıta

Yazar:komut.io ekibi LangChain ile AI Uygulama Geliştirme
0 Yorum

Sesli asistanlar artık sadece büyük tech şirketlerinin tekelinde değil. LangChain, OpenAI API’leri ve birkaç Python kütüphanesiyle kendi sesli asistanını sıfırdan inşa edebilirsin. Bu yazıda gerçek dünya senaryoları üzerinden, üretim ortamına taşınabilir bir sesli asistan geliştireceğiz. Sadece “merhaba dünya” seviyesinde değil, bağlam takibi yapan, hata yöneten ve ölçeklenebilir bir yapı kuracağız.

Mimariye Genel Bakış

Sesli asistan geliştirmek aslında üç temel bileşenin birbirine bağlanmasından ibaret. Speech-to-Text (STT) mikrofon girdisini yazıya çeviriyor, LangChain + LLM bu metni işleyip yanıt üretiyor, Text-to-Speech (TTS) ise yanıtı sese dönüştürüyor.

Bu pipeline’da dikkat edilmesi gereken nokta gecikme süresi. Kullanıcı konuştuğunda 2-3 saniye içinde yanıt gelmesi gerekiyor, aksi halde deneyim berbat oluyor. Bu yüzden her katmanda optimizasyon kararları alacağız.

Kullanacağımız stack:

  • OpenAI Whisper – STT için (lokal ya da API)
  • LangChain – LLM orkestrasyon katmanı
  • GPT-4o-mini – Hız/maliyet dengesi için ideal
  • ElevenLabs veya pyttsx3 – TTS için
  • PyAudio – Mikrofon girişi

Ortam Kurulumu

Önce sanal ortamı hazırlayıp bağımlılıkları yükleyelim. Production ortamında bu adımı bir Dockerfile’a taşıman önerilir.

python -m venv sesli-asistan-env
source sesli-asistan-env/bin/activate  # Windows'ta: .sesli-asistan-envScriptsactivate

pip install langchain langchain-openai openai
pip install SpeechRecognition pyaudio
pip install openai-whisper
pip install pyttsx3 elevenlabs
pip install python-dotenv numpy

# PyAudio için sistem bağımlılıkları (Ubuntu/Debian)
sudo apt-get install portaudio19-dev python3-dev
pip install pyaudio

# macOS için
brew install portaudio
pip install pyaudio

.env dosyamızı hazırlayalım:

cat > .env << 'EOF'
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxx
ELEVENLABS_API_KEY=xxxxxxxxxxxxxxxxxx
WHISPER_MODEL=base  # tiny, base, small, medium, large
LLM_MODEL=gpt-4o-mini
MAX_TOKENS=500
TEMPERATURE=0.7
EOF

Temel Ses Yakalama Modülü

Mikrofon girişini yönetecek bir sınıf yazalım. Burada önemli nokta sessizlik tespiti – kullanıcı konuşmayı bitirdiğinde otomatik olarak kayıt durmalı.

# audio_capture.py
import speech_recognition as sr
import numpy as np
import time
from typing import Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AudioCapture:
    def __init__(
        self,
        energy_threshold: int = 300,
        pause_threshold: float = 1.5,
        phrase_time_limit: int = 30
    ):
        self.recognizer = sr.Recognizer()
        self.recognizer.energy_threshold = energy_threshold
        self.recognizer.pause_threshold = pause_threshold
        self.recognizer.dynamic_energy_threshold = True
        self.phrase_time_limit = phrase_time_limit
        self.microphone = sr.Microphone()
        self._calibrate()

    def _calibrate(self):
        """Arka plan gürültüsüne göre kalibrasyon yap"""
        logger.info("Mikrofon kalibre ediliyor, sessiz olun...")
        with self.microphone as source:
            self.recognizer.adjust_for_ambient_noise(source, duration=2)
        logger.info(f"Kalibrasyon tamamlandı. Enerji eşiği: {self.recognizer.energy_threshold}")

    def listen(self) -> Optional[sr.AudioData]:
        """Ses girişini yakala ve AudioData döndür"""
        try:
            with self.microphone as source:
                logger.info("Dinleniyor...")
                audio = self.recognizer.listen(
                    source,
                    timeout=5,
                    phrase_time_limit=self.phrase_time_limit
                )
            return audio
        except sr.WaitTimeoutError:
            logger.warning("Ses girişi zaman aşımı")
            return None
        except Exception as e:
            logger.error(f"Ses yakalama hatası: {e}")
            return None

Whisper ile Speech-to-Text

API yerine lokal Whisper kullanmak hem maliyet hem de gizlilik açısından avantajlı. Kurumsal ortamlarda ses verilerini dışarıya göndermek istemeyebilirsin.

# stt_engine.py
import whisper
import numpy as np
import speech_recognition as sr
from typing import Optional, Literal
import torch
import logging

logger = logging.getLogger(__name__)

class WhisperSTT:
    def __init__(
        self,
        model_size: Literal["tiny", "base", "small", "medium", "large"] = "base",
        language: str = "tr",
        device: Optional[str] = None
    ):
        self.language = language
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
        
        logger.info(f"Whisper modeli yükleniyor: {model_size} - Cihaz: {self.device}")
        self.model = whisper.load_model(model_size, device=self.device)
        logger.info("Model yüklendi")

    def transcribe_audio_data(self, audio_data: sr.AudioData) -> Optional[str]:
        """SpeechRecognition AudioData'sını metne çevir"""
        try:
            # AudioData'yı numpy array'e dönüştür
            raw_data = audio_data.get_raw_data(
                convert_rate=16000,
                convert_width=2
            )
            audio_array = np.frombuffer(raw_data, dtype=np.int16)
            audio_float = audio_array.astype(np.float32) / 32768.0

            result = self.model.transcribe(
                audio_float,
                language=self.language,
                fp16=self.device == "cuda",
                task="transcribe"
            )
            
            text = result["text"].strip()
            logger.info(f"Transkripsiyon: '{text}'")
            return text if text else None
            
        except Exception as e:
            logger.error(f"Transkripsiyon hatası: {e}")
            return None

    def transcribe_file(self, file_path: str) -> Optional[str]:
        """Ses dosyasını metne çevir"""
        try:
            result = self.model.transcribe(
                file_path,
                language=self.language
            )
            return result["text"].strip()
        except Exception as e:
            logger.error(f"Dosya transkripsiyon hatası: {e}")
            return None

LangChain Konuşma Zinciri

Şimdi işin kalbi olan LangChain entegrasyonunu kuralım. Burada konuşma geçmişini bellekte tutacak ve sistem prompt’u ile asistana kişilik kazandıracağız.

# llm_engine.py
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import SystemMessage
from typing import Optional
import os
import logging

logger = logging.getLogger(__name__)

SYSTEM_PROMPT = """Sen yardımcı bir sesli asistansın. Adın Kaya.

Kuralların:
- Yanıtların kısa ve net olsun, maksimum 2-3 cümle
- Teknik konularda uzman gibi konuş ama anlaşılır ol
- Sesli yanıt vereceğin için markdown, liste veya özel karakterler kullanma
- Türkçe konuş, gerektiğinde teknik terimleri kullan
- Önceki konuşmayı hatırla ve bağlam kur
"""

class ConversationEngine:
    def __init__(
        self,
        model_name: str = "gpt-4o-mini",
        temperature: float = 0.7,
        max_tokens: int = 500,
        memory_window: int = 10
    ):
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=temperature,
            max_tokens=max_tokens,
            openai_api_key=os.getenv("OPENAI_API_KEY"),
            request_timeout=10  # 10 saniye timeout
        )
        
        self.memory = ConversationBufferWindowMemory(
            k=memory_window,
            return_messages=True,
            memory_key="history"
        )
        
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(content=SYSTEM_PROMPT),
            MessagesPlaceholder(variable_name="history"),
            ("human", "{input}")
        ])
        
        self.chain = ConversationChain(
            llm=self.llm,
            memory=self.memory,
            prompt=prompt,
            verbose=False
        )

    def get_response(self, user_input: str) -> Optional[str]:
        """Kullanıcı girdisine yanıt üret"""
        try:
            logger.info(f"LLM'e gönderilen: '{user_input}'")
            response = self.chain.predict(input=user_input)
            logger.info(f"LLM yanıtı: '{response}'")
            return response
        except Exception as e:
            logger.error(f"LLM yanıt hatası: {e}")
            return "Üzgünüm, şu anda yanıt üretemiyorum. Lütfen tekrar deneyin."

    def clear_memory(self):
        """Konuşma geçmişini temizle"""
        self.memory.clear()
        logger.info("Konuşma geçmişi temizlendi")

    def get_conversation_history(self) -> list:
        """Mevcut konuşma geçmişini döndür"""
        return self.memory.chat_memory.messages

Text-to-Speech Katmanı

TTS için iki seçenek sunuyorum: ElevenLabs (yüksek kalite, ücretli) ve pyttsx3 (lokal, ücretsiz). Production’da ElevenLabs’ı, geliştirme ortamında pyttsx3’ü tercih edebilirsin.

# tts_engine.py
import pyttsx3
import os
import tempfile
import pygame
import logging
from abc import ABC, abstractmethod
from typing import Optional
from elevenlabs import ElevenLabs, VoiceSettings

logger = logging.getLogger(__name__)

class BaseTTS(ABC):
    @abstractmethod
    def speak(self, text: str) -> bool:
        pass

class LocalTTS(BaseTTS):
    """pyttsx3 ile lokal TTS - geliştirme ortamı için"""
    
    def __init__(self, rate: int = 185, volume: float = 0.9):
        self.engine = pyttsx3.init()
        self.engine.setProperty("rate", rate)
        self.engine.setProperty("volume", volume)
        
        # Türkçe ses seç (varsa)
        voices = self.engine.getProperty("voices")
        for voice in voices:
            if "turkish" in voice.name.lower() or "tr" in voice.id.lower():
                self.engine.setProperty("voice", voice.id)
                logger.info(f"Türkçe ses seçildi: {voice.name}")
                break

    def speak(self, text: str) -> bool:
        try:
            self.engine.say(text)
            self.engine.runAndWait()
            return True
        except Exception as e:
            logger.error(f"TTS hatası: {e}")
            return False

class ElevenLabsTTS(BaseTTS):
    """ElevenLabs ile yüksek kaliteli TTS - production için"""
    
    def __init__(
        self,
        voice_id: str = "pNInz6obpgDQGcFmaJgB",  # Adam sesi
        model_id: str = "eleven_multilingual_v2",
        stability: float = 0.5,
        similarity_boost: float = 0.75
    ):
        self.client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
        self.voice_id = voice_id
        self.model_id = model_id
        self.voice_settings = VoiceSettings(
            stability=stability,
            similarity_boost=similarity_boost
        )
        pygame.mixer.init()

    def speak(self, text: str) -> bool:
        try:
            audio_stream = self.client.text_to_speech.convert(
                voice_id=self.voice_id,
                text=text,
                model_id=self.model_id,
                voice_settings=self.voice_settings
            )
            
            with tempfile.NamedTemporaryFile(
                suffix=".mp3",
                delete=False
            ) as tmp_file:
                for chunk in audio_stream:
                    tmp_file.write(chunk)
                tmp_path = tmp_file.name
            
            pygame.mixer.music.load(tmp_path)
            pygame.mixer.music.play()
            
            while pygame.mixer.music.get_busy():
                pygame.time.Clock().tick(10)
            
            os.unlink(tmp_path)
            return True
            
        except Exception as e:
            logger.error(f"ElevenLabs TTS hatası: {e}")
            return False

def get_tts_engine(use_elevenlabs: bool = False) -> BaseTTS:
    """Ortama göre TTS engine seç"""
    if use_elevenlabs and os.getenv("ELEVENLABS_API_KEY"):
        logger.info("ElevenLabs TTS kullanılıyor")
        return ElevenLabsTTS()
    logger.info("Lokal TTS kullanılıyor")
    return LocalTTS()

Ana Asistan Sınıfı ve Pipeline

Tüm bileşenleri bir araya getirip tam pipeline’ı oluşturalım. Komut kelimesi (“hey kaya”) tespiti ve graceful shutdown da ekliyoruz.

# assistant.py
import time
import threading
import logging
import os
from dotenv import load_dotenv
from audio_capture import AudioCapture
from stt_engine import WhisperSTT
from llm_engine import ConversationEngine
from tts_engine import get_tts_engine

load_dotenv()
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class VoiceAssistant:
    WAKE_WORDS = ["hey kaya", "kaya", "asistan"]
    EXIT_COMMANDS = ["çıkış", "güle güle", "kapat", "dur artık"]
    
    def __init__(self, use_elevenlabs: bool = False):
        logger.info("Sesli asistan başlatılıyor...")
        
        self.audio = AudioCapture(
            energy_threshold=300,
            pause_threshold=1.5
        )
        self.stt = WhisperSTT(
            model_size=os.getenv("WHISPER_MODEL", "base"),
            language="tr"
        )
        self.llm = ConversationEngine(
            model_name=os.getenv("LLM_MODEL", "gpt-4o-mini"),
            temperature=float(os.getenv("TEMPERATURE", "0.7")),
            max_tokens=int(os.getenv("MAX_TOKENS", "500"))
        )
        self.tts = get_tts_engine(use_elevenlabs=use_elevenlabs)
        
        self.is_running = False
        self.wake_word_mode = True  # True: uyku modu, False: aktif mod
        self.active_timeout = 30  # 30 saniye sessizlik sonrası uyku moduna geç
        self.last_interaction = time.time()
        
        logger.info("Asistan hazır!")

    def _check_wake_word(self, text: str) -> bool:
        """Uyandırma kelimesini kontrol et"""
        text_lower = text.lower()
        return any(wake in text_lower for wake in self.WAKE_WORDS)

    def _check_exit_command(self, text: str) -> bool:
        """Çıkış komutunu kontrol et"""
        text_lower = text.lower()
        return any(exit_cmd in text_lower for exit_cmd in self.EXIT_COMMANDS)

    def _process_single_turn(self, text: str) -> bool:
        """Tek konuşma turunu işle, False döndürürse çık"""
        if self._check_exit_command(text):
            self.tts.speak("Görüşmek üzere!")
            return False
        
        if text.strip() == "geçmişi temizle":
            self.llm.clear_memory()
            self.tts.speak("Konuşma geçmişi temizlendi.")
            return True
        
        response = self.llm.get_response(text)
        if response:
            self.tts.speak(response)
        
        return True

    def _activity_monitor(self):
        """Arka planda aktivite takibi"""
        while self.is_running:
            if (not self.wake_word_mode and 
                time.time() - self.last_interaction > self.active_timeout):
                logger.info("Aktivite yok, uyku moduna geçiliyor")
                self.wake_word_mode = True
            time.sleep(5)

    def run(self):
        """Ana asistan döngüsünü başlat"""
        self.is_running = True
        
        # Aktivite monitörünü arka planda başlat
        monitor_thread = threading.Thread(
            target=self._activity_monitor,
            daemon=True
        )
        monitor_thread.start()
        
        self.tts.speak("Merhaba! Hey Kaya diyerek beni uyandırabilirsin.")
        
        try:
            while self.is_running:
                audio_data = self.audio.listen()
                
                if audio_data is None:
                    continue
                
                text = self.stt.transcribe_audio_data(audio_data)
                
                if not text:
                    continue
                
                logger.info(f"Algılanan: '{text}'")
                
                if self.wake_word_mode:
                    if self._check_wake_word(text):
                        self.wake_word_mode = False
                        self.last_interaction = time.time()
                        self.tts.speak("Evet, sizi dinliyorum.")
                        logger.info("Uyandırma kelimesi algılandı, aktif moda geçildi")
                else:
                    self.last_interaction = time.time()
                    should_continue = self._process_single_turn(text)
                    if not should_continue:
                        self.is_running = False
                        
        except KeyboardInterrupt:
            logger.info("Kullanıcı tarafından durduruldu")
        finally:
            self.is_running = False
            logger.info("Asistan kapatıldı")

if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="LangChain Sesli Asistan")
    parser.add_argument(
        "--elevenlabs",
        action="store_true",
        help="ElevenLabs TTS kullan"
    )
    parser.add_argument(
        "--no-wake-word",
        action="store_true",
        help="Uyandırma kelimesi olmadan direkt başlat"
    )
    args = parser.parse_args()
    
    assistant = VoiceAssistant(use_elevenlabs=args.elevenlabs)
    
    if args.no_wake_word:
        assistant.wake_word_mode = False
    
    assistant.run()

Performans Optimizasyonu ve Gecikme Yönetimi

Gerçek dünya senaryosunda en büyük sorun gecikme. Birkaç pratik optimizasyon:

# Whisper modelini önceden GPU'ya yükle ve önbellekte tut
# Ayrıca LLM yanıtlarını streaming modunda işle

cat > streaming_response.py << 'EOF'
from langchain_openai import ChatOpenAI
from langchain_core.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# Streaming ile ilk token gelir gelmez TTS başlar
llm_streaming = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    temperature=0.7
)

# Cümle bazlı TTS için basit bir buffer
class SentenceBuffer:
    def __init__(self, tts_engine):
        self.buffer = ""
        self.tts = tts_engine
        self.sentence_endings = [".", "!", "?", "..."]
    
    def add_token(self, token: str):
        self.buffer += token
        for ending in self.sentence_endings:
            if ending in self.buffer:
                parts = self.buffer.split(ending, 1)
                sentence = parts[0] + ending
                if len(sentence.strip()) > 10:
                    self.tts.speak(sentence.strip())
                self.buffer = parts[1] if len(parts) > 1 else ""
    
    def flush(self):
        if self.buffer.strip():
            self.tts.speak(self.buffer.strip())
            self.buffer = ""
EOF

Sistemd Servisi Olarak Dağıtım

Asistanı her zaman arka planda çalışacak şekilde bir systemd servisi haline getirelim:

# /etc/systemd/system/voice-assistant.service
sudo tee /etc/systemd/system/voice-assistant.service << 'EOF'
[Unit]
Description=LangChain Voice Assistant
After=network.target sound.target
Wants=sound.target

[Service]
Type=simple
User=pi
Group=audio
WorkingDirectory=/opt/voice-assistant
Environment=DISPLAY=:0
Environment=PULSE_SERVER=unix:/run/user/1000/pulse/native
EnvironmentFile=/opt/voice-assistant/.env
ExecStart=/opt/voice-assistant/sesli-asistan-env/bin/python assistant.py
Restart=on-failure
RestartSec=10
StandardOutput=append:/var/log/voice-assistant/app.log
StandardError=append:/var/log/voice-assistant/error.log

[Install]
WantedBy=multi-user.target
EOF

# Log dizini oluştur
sudo mkdir -p /var/log/voice-assistant
sudo chown pi:pi /var/log/voice-assistant

# Servisi etkinleştir ve başlat
sudo systemctl daemon-reload
sudo systemctl enable voice-assistant
sudo systemctl start voice-assistant

# Durumu kontrol et
sudo systemctl status voice-assistant
journalctl -u voice-assistant -f

Gerçek Dünya Senaryosu: Sunucu Odası Asistanı

Bu yapıyı bir adım öteye taşıyalım. Sysadmin olarak sunucu odanızda çalışırken ellerin doluyken “Hey Kaya, disk kullanımını söyle” diyebilirsin:

  • Özel araçlar ekle: LangChain’in tool/agent yapısıyla df -h, top, systemctl status gibi komutları çalıştıran araçlar bağlanabilir
  • Alerting entegrasyonu: Prometheus/Grafana alertleri sesli bildirime dönüştürülebilir
  • Log analizi: Tail edilen log satırlarını LLM’e göndererek sesli özetleme yapılabilir
  • Çoklu dil desteği: Whisper’ın çok dilli yapısı sayesinde İngilizce komutlar da algılanır
  • Raspberry Pi dağıtımı: Küçük formlu bir Raspberry Pi 4 ile sunucu odası köşesine kurabilirsin

Güvenlik ve Gizlilik Notları

Sesli asistan geliştirirken göz ardı edilmemesi gereken birkaç nokta var. Uyandırma kelimesi olmayan modda mikrofon sürekli aktiftir, bu ciddi bir gizlilik sorunudur. Wake word modelini mutlaka etkinleştir.

API anahtarlarını asla kod içine gömme, her zaman .env kullan. Kurumsal ortamlarda ses verilerini bulut STT’ye göndermek yerine lokal Whisper tercih et. Konuşma geçmişini memory buffer’dan düzenli temizle, hassas bilgiler kalıcı olarak loglanmamalı. Servis account yetkilerini minimum tutarak çalıştır.

Sorun Giderme

En sık karşılaşılan sorunlar ve çözümleri:

  • PyAudio kurulum hatası: portaudio sistem kütüphanesinin kurulu olduğundan emin ol. Ubuntu’da sudo apt-get install portaudio19-dev yeterli
  • Yüksek gecikme: Whisper tiny veya base modelini dene, medium ve large CPU’da çok yavaş
  • Mikrofon algılanmıyor: python -c "import speech_recognition as sr; print(sr.Microphone.list_microphone_names())" ile cihaz listesini kontrol et
  • LLM timeout: request_timeout parametresini artır ama 15 saniyeyi geçme, aksi halde kullanıcı deneyimi bozulur
  • TTS sesi yok: pygame.mixer.init() çağrıldığından emin ol, özellikle headless ortamlarda DISPLAY değişkeni gerekebilir

Sonuç

LangChain ile sesli asistan geliştirmek göründüğünden çok daha erişilebilir. Temel pipeline üç bileşenden oluşuyor: ses yakalama, dil modeli ve ses sentezi. Bunların her birini bağımsız olarak geliştirip test edebilirsin.

Bu yazıda kurduğumuz yapı üzerine daha fazlasını inşa edebilirsin. LangChain’in agent framework’ü ile gerçek araçlara bağlamak, RAG entegrasyonuyla kişisel bilgi tabanı oluşturmak ya da birden fazla kullanıcıyı destekleyecek şekilde genişletmek mümkün. Sysadmin perspektifinden bakıldığında, bu tür araçlar elle dolduğunda veya terminal açık değilken bile sistemlerle etkileşim kurma imkânı veriyor. Raspberry Pi üzerinde çalışan, sunucu odası kapısına monte edilmiş bir sesli asistan kulağa hobi projesi gibi gelebilir, ama üretimde ciddi iş görebilir.

Kod örneklerinin tamamını GitHub’a koyacağım, takipte kal.

Bir yanıt yazın

E-posta adresiniz yayınlanmayacak. Gerekli alanlar * ile işaretlenmişlerdir