LangChain ile Otomatik E-posta Yanıtlama Sistemi Kurulumu

Yazar:komut.io ekibi
0 Yorum

E-posta yönetimi, sistem yöneticilerinin ve DevOps mühendislerinin zamanının büyük bir bölümünü yiyen, görünmez ama yorucu bir iştir. Özellikle büyük organizasyonlarda günde yüzlerce tekrarlayan soru, otomatik bildirim veya destek talebi geliyor ve bunların her birine manuel yanıt vermek hem zaman hem de enerji kaybı. Ben bu sorunu LangChain kullanarak otomatik bir e-posta yanıtlama sistemi kurarak çözdüm ve bu yazıda o deneyimi aktarmak istiyorum.

Sistem Mimarisine Genel Bakış

Kurduğumuz sistem temelde şu akış üzerine inşa edildi: Gelen e-postayı oku, içeriği analiz et, uygun bir yanıt üret, gerekirse insan onayına sun, onaylandıysa gönder. Kulağa basit geliyor ama şeytanın detaylarda olduğunu biliyorsunuz.

LangChain bu süreçte bize zincirleme (chaining) yeteneği sağlıyor. Tek bir LLM çağrısıyla her şeyi çözmek yerine, farklı adımları birbiriyle bağlayarak daha güvenilir ve yönetilebilir bir pipeline oluşturabiliyorsunuz. Bu da production ortamında işlerin sarpa sarması durumunda neyin nerede bozulduğunu anlamayı kolaylaştırıyor.

Mimari bileşenler şunlar:

  • E-posta Okuyucu: IMAP üzerinden gelen kutusu takibi
  • Sınıflandırıcı: E-postanın tipini ve önceliğini belirleyen zincir
  • Yanıt Üretici: Bağlama uygun yanıt oluşturan LLM zinciri
  • Onay Katmanı: Kritik yanıtlar için insan gözden geçirmesi
  • Gönderici: SMTP üzerinden yanıt gönderimi

Ortam Kurulumu

Önce gerekli bağımlılıkları kuralım:

pip install langchain langchain-openai langchain-community 
    imaplib2 python-dotenv pydantic 
    email-validator jinja2 redis

Ortam değişkenlerini ayarlayalım:

cat > .env << 'EOF'
OPENAI_API_KEY=sk-your-key-here
IMAP_SERVER=mail.sirketiniz.com
IMAP_PORT=993
[email protected]
IMAP_PASSWORD=guclu-bir-sifre
SMTP_SERVER=mail.sirketiniz.com
SMTP_PORT=587
[email protected]
SMTP_PASSWORD=guclu-bir-sifre
REDIS_URL=redis://localhost:6379/0
[email protected]
AUTO_REPLY_THRESHOLD=0.85
EOF

Redis’i ayağa kaldıralım, durumu takip etmek için kullanacağız:

docker run -d --name email-bot-redis 
    -p 6379:6379 
    --restart unless-stopped 
    redis:7-alpine redis-server --appendonly yes

E-posta Okuyucu Modülü

IMAP entegrasyonunu güvenli ve kararlı bir şekilde yazmak önemli. Bağlantı kesilmelerine, timeout’lara ve özellikle büyük ekleri olan e-postalara karşı dayanıklı olması gerekiyor:

cat > email_reader.py << 'PYEOF'
import imaplib
import email
import redis
import json
import hashlib
from email.header import decode_header
from datetime import datetime, timedelta
from typing import Optional
import os
from dotenv import load_dotenv

load_dotenv()

class EmailReader:
    def __init__(self):
        self.imap_server = os.getenv("IMAP_SERVER")
        self.imap_port = int(os.getenv("IMAP_PORT", 993))
        self.username = os.getenv("IMAP_USER")
        self.password = os.getenv("IMAP_PASSWORD")
        self.redis_client = redis.from_url(os.getenv("REDIS_URL"))
        self.connection = None

    def connect(self):
        try:
            self.connection = imaplib.IMAP4_SSL(
                self.imap_server, 
                self.imap_port
            )
            self.connection.login(self.username, self.password)
            print(f"[OK] IMAP baglantisi kuruldu: {self.imap_server}")
            return True
        except imaplib.IMAP4.error as e:
            print(f"[HATA] IMAP baglanti hatasi: {e}")
            return False

    def decode_subject(self, subject_header):
        decoded_parts = decode_header(subject_header)
        subject = ""
        for part, charset in decoded_parts:
            if isinstance(part, bytes):
                charset = charset or "utf-8"
                subject += part.decode(charset, errors="replace")
            else:
                subject += part
        return subject

    def get_email_body(self, msg):
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                content_type = part.get_content_type()
                if content_type == "text/plain":
                    payload = part.get_payload(decode=True)
                    if payload:
                        charset = part.get_content_charset() or "utf-8"
                        body += payload.decode(charset, errors="replace")
                    break
        else:
            payload = msg.get_payload(decode=True)
            if payload:
                charset = msg.get_content_charset() or "utf-8"
                body = payload.decode(charset, errors="replace")
        return body[:3000]

    def get_email_fingerprint(self, email_data):
        content = f"{email_data['from']}{email_data['subject']}{email_data['date']}"
        return hashlib.md5(content.encode()).hexdigest()

    def fetch_unread_emails(self, limit=20):
        if not self.connection:
            if not self.connect():
                return []

        try:
            self.connection.select("INBOX")
            _, messages = self.connection.search(None, "UNSEEN")
            email_ids = messages[0].split()
            
            if not email_ids:
                return []

            emails = []
            for email_id in email_ids[-limit:]:
                _, msg_data = self.connection.fetch(email_id, "(RFC822)")
                raw_email = msg_data[0][1]
                msg = email.message_from_bytes(raw_email)
                
                email_data = {
                    "id": email_id.decode(),
                    "from": msg.get("From", ""),
                    "subject": self.decode_subject(msg.get("Subject", "")),
                    "date": msg.get("Date", ""),
                    "body": self.get_email_body(msg),
                    "message_id": msg.get("Message-ID", ""),
                    "reply_to": msg.get("Reply-To", msg.get("From", ""))
                }
                
                fingerprint = self.get_email_fingerprint(email_data)
                processed_key = f"processed_email:{fingerprint}"
                
                if not self.redis_client.exists(processed_key):
                    emails.append(email_data)
                    self.redis_client.setex(
                        processed_key, 
                        timedelta(days=7), 
                        "1"
                    )

            return emails

        except Exception as e:
            print(f"[HATA] E-posta okuma hatasi: {e}")
            self.connection = None
            return []
PYEOF
echo "Email reader modulu olusturuldu"

LangChain Zincirleri Kurulumu

Şimdi asıl iş burada başlıyor. İki temel zincir kuruyoruz: biri sınıflandırma için, diğeri yanıt üretimi için. Bunları ayrı tutmak önemli çünkü sınıflandırma ucuz ve hızlı bir model ile yapılabilirken, yanıt üretimi daha güçlü bir model gerektiriyor:

cat > langchain_processor.py << 'PYEOF'
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain.output_parsers import PydanticOutputParser
from langchain.schema.runnable import RunnablePassthrough, RunnableParallel
from pydantic import BaseModel, Field
from typing import Optional, Literal
import os

class EmailClassification(BaseModel):
    category: Literal[
        "teknik_destek", 
        "fatura_odeme", 
        "genel_bilgi", 
        "sikayet", 
        "spam",
        "acil_sistem_alarmi"
    ] = Field(description="E-posta kategorisi")
    priority: Literal["dusuk", "orta", "yuksek", "kritik"] = Field(
        description="Oncelik seviyesi"
    )
    requires_human: bool = Field(
        description="Insan mudahalesi gerekiyor mu"
    )
    confidence: float = Field(
        description="0-1 arasi guven skoru",
        ge=0.0, le=1.0
    )
    summary: str = Field(description="E-posta ozeti (max 100 kelime)")

class EmailProcessor:
    def __init__(self):
        self.classifier_llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.responder_llm = ChatOpenAI(
            model="gpt-4o",
            temperature=0.3,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        self.classification_chain = self._build_classification_chain()
        self.response_chain = self._build_response_chain()

    def _build_classification_chain(self):
        parser = PydanticOutputParser(pydantic_object=EmailClassification)
        
        system_prompt = SystemMessagePromptTemplate.from_template(
            """Sen bir IT destek ekibinin e-posta siniflandirma asistanisin.
            Gelen e-postalari analiz edip siniflandiriyorsun.
            
            Onemli Kurallar:
            - Sistem alarmlari veya kesinti bildirimleri her zaman KRITIK
            - Banka/odeme bilgisi isteyen e-postalar spam olmali
            - Teknik sorunlarda detayli analiz yap
            
            {format_instructions}"""
        )
        
        human_prompt = HumanMessagePromptTemplate.from_template(
            """Gonderen: {sender}
            Konu: {subject}
            Icerik: {body}
            
            Bu e-postay siniflandir."""
        )
        
        prompt = ChatPromptTemplate.from_messages([
            system_prompt,
            human_prompt
        ]).partial(format_instructions=parser.get_format_instructions())
        
        return prompt | self.classifier_llm | parser

    def _build_response_chain(self):
        system_prompt = SystemMessagePromptTemplate.from_template(
            """Sen {company_name} sirketinin profesyonel IT destek uzmanindaki bir 
            e-posta yanit asistanisin.
            
            Yanit Kurallari:
            - Her zaman Turkce yaz
            - Profesyonel ama samimi bir ton kullan
            - Somut cozumler sun, muglak cevaplar verme
            - Gerekirse adim adim talimat ver
            - E-postay {response_length} kelimeyi gecmeyecek sekilde tut
            - Imza olarak sadece "{signature}" kullan"""
        )
        
        human_prompt = HumanMessagePromptTemplate.from_template(
            """Gonderen: {sender}
            Konu: {subject}
            Orijinal E-posta: {body}
            
            Kategori: {category}
            Ozet: {summary}
            
            Bu e-postaya uygun bir yanit yaz."""
        )
        
        prompt = ChatPromptTemplate.from_messages([
            system_prompt, 
            human_prompt
        ])
        
        return prompt | self.responder_llm

    def classify_email(self, email_data):
        return self.classification_chain.invoke({
            "sender": email_data["from"],
            "subject": email_data["subject"],
            "body": email_data["body"]
        })

    def generate_response(self, email_data, classification):
        result = self.response_chain.invoke({
            "company_name": "TechDestek A.S.",
            "response_length": "200",
            "signature": "IT Destek Ekibi",
            "sender": email_data["from"],
            "subject": email_data["subject"],
            "body": email_data["body"],
            "category": classification.category,
            "summary": classification.summary
        })
        return result.content
PYEOF
echo "LangChain islemci modulu olusturuldu"

Onay Mekanizması ve Gönderici

Production’da yapay zekanın ürettiği her yanıtı doğrudan göndermek riskli. Özellikle şikayet e-postaları veya kritik sistem alarmları için bir onay katmanı şart:

cat > email_sender.py << 'PYEOF'
import smtplib
import os
import json
import redis
import uuid
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import timedelta
from dotenv import load_dotenv

load_dotenv()

class EmailSender:
    def __init__(self):
        self.smtp_server = os.getenv("SMTP_SERVER")
        self.smtp_port = int(os.getenv("SMTP_PORT", 587))
        self.username = os.getenv("SMTP_USER")
        self.password = os.getenv("SMTP_PASSWORD")
        self.approval_email = os.getenv("APPROVAL_EMAIL")
        self.redis_client = redis.from_url(os.getenv("REDIS_URL"))
        self.threshold = float(os.getenv("AUTO_REPLY_THRESHOLD", 0.85))

    def _create_smtp_connection(self):
        conn = smtplib.SMTP(self.smtp_server, self.smtp_port)
        conn.ehlo()
        conn.starttls()
        conn.login(self.username, self.password)
        return conn

    def send_email(self, to_address, subject, body, in_reply_to=None):
        msg = MIMEMultipart("alternative")
        msg["From"] = self.username
        msg["To"] = to_address
        msg["Subject"] = f"Re: {subject}" if not subject.startswith("Re:") else subject
        
        if in_reply_to:
            msg["In-Reply-To"] = in_reply_to
            msg["References"] = in_reply_to

        msg.attach(MIMEText(body, "plain", "utf-8"))
        
        try:
            with self._create_smtp_connection() as conn:
                conn.sendmail(self.username, to_address, msg.as_string())
            print(f"[OK] E-posta gonderildi: {to_address}")
            return True
        except Exception as e:
            print(f"[HATA] E-posta gonderilemedi: {e}")
            return False

    def queue_for_approval(self, original_email, generated_response, classification):
        approval_id = str(uuid.uuid4())[:8]
        
        pending_data = {
            "approval_id": approval_id,
            "original_email": original_email,
            "generated_response": generated_response,
            "classification": classification.dict()
        }
        
        self.redis_client.setex(
            f"pending_approval:{approval_id}",
            timedelta(hours=24),
            json.dumps(pending_data, ensure_ascii=False)
        )
        
        approval_body = f"""
Otomatik E-posta Onay Talebi
Onay ID: {approval_id}
Kategori: {classification.category}
Oncelik: {classification.priority}
Guven Skoru: {classification.confidence:.2%}

--- ORJINAL E-POSTA ---
Gonderen: {original_email['from']}
Konu: {original_email['subject']}
{original_email['body'][:500]}

--- ONERILEN YANIT ---
{generated_response}

--- ONAY ---
Onayla: destek-bot onay {approval_id}
Reddet: destek-bot reddet {approval_id}
        """
        
        self.send_email(
            self.approval_email,
            f"[BOT ONAYI GEREKLI] {original_email['subject']}",
            approval_body
        )
        print(f"[INFO] Onay bekleniyor: ID={approval_id}")
        return approval_id

    def should_auto_reply(self, classification):
        auto_categories = ["genel_bilgi", "teknik_destek"]
        
        if classification.category == "spam":
            return False
        if classification.requires_human:
            return False
        if classification.priority in ["kritik"]:
            return False
        if classification.confidence < self.threshold:
            return False
        if classification.category not in auto_categories:
            return False
            
        return True
PYEOF
echo "Email sender modulu olusturuldu"

Ana Orkestrasyon Döngüsü

Bütün modülleri bir araya getiren ana script. Bunu systemd ile servis olarak çalıştıracağız:

cat > main.py << 'PYEOF'
import time
import schedule
import logging
from email_reader import EmailReader
from langchain_processor import EmailProcessor
from email_sender import EmailSender

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("/var/log/email-bot/bot.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class EmailBot:
    def __init__(self):
        self.reader = EmailReader()
        self.processor = EmailProcessor()
        self.sender = EmailSender()
        self.processed_count = 0
        self.error_count = 0

    def process_single_email(self, email_data):
        logger.info(f"Islenecek: {email_data['subject'][:50]}")
        
        classification = self.processor.classify_email(email_data)
        logger.info(
            f"Siniflandirma: {classification.category} "
            f"(guven: {classification.confidence:.2%})"
        )
        
        if classification.category == "spam":
            logger.info("Spam tespit edildi, atlaniyor")
            return
        
        response = self.processor.generate_response(email_data, classification)
        
        if self.sender.should_auto_reply(classification):
            self.sender.send_email(
                to_address=email_data["reply_to"],
                subject=email_data["subject"],
                body=response,
                in_reply_to=email_data.get("message_id")
            )
            logger.info("Otomatik yanit gonderildi")
        else:
            self.sender.queue_for_approval(
                email_data, response, classification
            )
            logger.info("Onay kuyruğuna alindi")

    def run_cycle(self):
        logger.info("E-posta kontrol dongüsü basliyor...")
        
        try:
            emails = self.reader.fetch_unread_emails(limit=20)
            
            if not emails:
                logger.info("Yeni e-posta yok")
                return
                
            logger.info(f"{len(emails)} adet yeni e-posta bulundu")
            
            for email_data in emails:
                try:
                    self.process_single_email(email_data)
                    self.processed_count += 1
                    time.sleep(2)
                except Exception as e:
                    self.error_count += 1
                    logger.error(f"E-posta isleme hatasi: {e}", exc_info=True)
                    
        except Exception as e:
            logger.error(f"Döngü hatasi: {e}", exc_info=True)

def main():
    logger.info("E-posta Bot baslatiliyor...")
    bot = EmailBot()
    
    schedule.every(5).minutes.do(bot.run_cycle)
    
    bot.run_cycle()
    
    while True:
        schedule.run_pending()
        time.sleep(30)

if __name__ == "__main__":
    main()
PYEOF

mkdir -p /var/log/email-bot
echo "Ana script olusturuldu"

Systemd Servis Olarak Kurulum

Botu systemd ile yönetmek en sağlıklı yaklaşım. Böylece sunucu yeniden başladığında otomatik devreye giriyor ve çökmeler durumunda kendini yeniden başlatıyor:

cat > /etc/systemd/system/email-bot.service << 'EOF'
[Unit]
Description=LangChain Otomatik E-posta Yanit Botu
After=network.target redis.service
Requires=redis.service

[Service]
Type=simple
User=emailbot
Group=emailbot
WorkingDirectory=/opt/email-bot
Environment="PATH=/opt/email-bot/venv/bin"
ExecStart=/opt/email-bot/venv/bin/python main.py
Restart=always
RestartSec=30
StartLimitIntervalSec=300
StartLimitBurst=5

StandardOutput=append:/var/log/email-bot/stdout.log
StandardError=append:/var/log/email-bot/stderr.log

[Install]
WantedBy=multi-user.target
EOF

useradd -r -s /bin/false -d /opt/email-bot emailbot
chown -R emailbot:emailbot /opt/email-bot /var/log/email-bot

systemctl daemon-reload
systemctl enable email-bot.service
systemctl start email-bot.service
systemctl status email-bot.service

Maliyet Takibi ve Rate Limiting

OpenAI API maliyetlerinin kontrolden çıkmaması için rate limiting ve maliyet takibi şart. Bunu ihmal eden sistemi daha sonra faturaya bakıp üzülmek zorunda kalıyor:

cat > cost_tracker.py << 'PYEOF'
import redis
import json
from datetime import datetime, timedelta
import os

class CostTracker:
    def __init__(self):
        self.redis_client = redis.from_url(os.getenv("REDIS_URL"))
        self.gpt4o_input_cost = 0.005
        self.gpt4o_output_cost = 0.015
        self.gpt4o_mini_cost = 0.0002
        self.daily_limit_usd = float(os.getenv("DAILY_COST_LIMIT_USD", 5.0))

    def get_today_key(self):
        return f"cost:{datetime.now().strftime('%Y-%m-%d')}"

    def add_usage(self, model, input_tokens, output_tokens):
        if "mini" in model:
            cost = (input_tokens + output_tokens) / 1000 * self.gpt4o_mini_cost
        else:
            cost = (input_tokens / 1000 * self.gpt4o_input_cost + 
                   output_tokens / 1000 * self.gpt4o_output_cost)
        
        key = self.get_today_key()
        today_cost = float(self.redis_client.get(key) or 0)
        today_cost += cost
        
        self.redis_client.setex(key, timedelta(days=30), str(today_cost))
        
        if today_cost > self.daily_limit_usd:
            raise Exception(
                f"Günlük maliyet limiti asildi: ${today_cost:.4f} > ${self.daily_limit_usd}"
            )
        
        return cost

    def get_daily_cost(self):
        key = self.get_today_key()
        return float(self.redis_client.get(key) or 0)

    def print_stats(self):
        daily_cost = self.get_daily_cost()
        print(f"Bugunun maliyeti: ${daily_cost:.4f}")
        print(f"Gunluk limit: ${self.daily_limit_usd:.2f}")
        print(f"Kalan: ${self.daily_limit_usd - daily_cost:.4f}")
PYEOF
echo "Maliyet takip modulu olusturuldu"

İzleme ve Hata Ayıklama

Sistemin gerçekten doğru çalışıp çalışmadığını kontrol etmek için log analizi ve durum kontrolü:

# Bot durumunu kontrol et
systemctl status email-bot.service

# Canli log takibi
journalctl -u email-bot.service -f

# Son 1 saatin loglarini goster
journalctl -u email-bot.service --since "1 hour ago"

# Redis'te onay bekleyenleri listele
redis-cli KEYS "pending_approval:*" | while read key; do
    echo "=== $key ==="
    redis-cli GET "$key" | python3 -c "
import sys, json
data = json.load(sys.stdin)
print('Konu:', data['original_email']['subject'])
print('Kategori:', data['classification']['category'])
print('Guven:', data['classification']['confidence'])
"
done

# Bugunun maliyet raporu
cd /opt/email-bot && 
    /opt/email-bot/venv/bin/python -c "
from cost_tracker import CostTracker
ct = CostTracker()
ct.print_stats()
"

# Islenen e-posta sayisini Redis'ten al
redis-cli KEYS "processed_email:*" | wc -l

Gerçek Dünya Senaryoları ve Dikkat Noktaları

Bu sistemi birkaç farklı organizasyonda kurdum ve her seferinde karşıma çıkan bazı kritik noktalar var.

Güven skoru eşiği seçimi belki de en önemli karar. %85’lik eşik iyi bir başlangıç noktası ama sektöre göre değişiyor. Sağlık veya finans gibi yüksek riskli alanlarda bu eşiği %95’e çekin ve insan onayı katmanını daha agresif kullanın.

Türkçe dil desteği konusunda GPT-4o Mini, Türkçe sınıflandırma için yeterince iyi. Ancak yanıt üretiminde GPT-4o’nun kalitesi belirgin şekilde daha yüksek. Maliyet ve kalite dengesini kendinize göre ayarlamanız gerekiyor.

IMAP bağlantısı uzun süreli çalışmalarda bazen kopuyor. Kod içinde reconnect mantığını yazdık ama yine de aralıklı bağlantı sorunlarını monitoring ile takip edin. Özellikle Office 365 ve Google Workspace, uzun süreli IMAP bağlantılarına zaman zaman toleranssız davranıyor.

Rate limiting konusu da ihmal edilemez. OpenAI API’sinde dakika başına token limitleri var. Yoğun e-posta trafiğinde bu limitlere çarpabilirsiniz. Her e-posta işlemi arasına koyduğumuz 2 saniyelik bekleme bu nedenle.

Kişisel veri ve KVKK uyumu meselesi çok önemli. E-posta içeriklerini üçüncü parti bir API’ye gönderiyorsunuz. OpenAI’ın veri işleme politikalarını gözden geçirin ve gerekirse kurumsal sözleşme yapın. Hassas verileri yanıt üretimine göndermeden önce bir anonimleştirme katmanı eklemek iyi bir pratik.

Yapay zekanın ürettiği yanıtların düzenli olarak insan tarafından gözden geçirilmesi de şart. İlk haftalar boyunca auto-reply eşiğini yüksek tutun, sistemi tanıyın, ardından kademeli olarak özerkliği artırın. Bunu yapmadan, modelin hatalı sınıflandırmasından kaynaklanan utanç verici yanıtlar gönderildiğini keşfetmek sizi zor durumda bırakabilir.

Sonuç

LangChain tabanlı bu sistem, e-posta yönetiminde ciddi bir iş yükü azalması sağlıyor. Genel bilgi ve rutin teknik destek sorularının otomatik yanıtlanması, ekibin gerçek problemlere daha fazla zaman ayırmasını mümkün kılıyor. Ancak sihirli bir çözüm değil. Yanlış ayarlanmış bir güven eşiği veya gözden kaçan bir sınıflandırma hatası, müşteri memnuniyetsizliğine yol açabilir.

Bu projeyi canlıya alırken adım adım ilerleyin: Önce sadece sınıflandırma yapın ve sonuçları inceleyin. Sonra onay mekanizmasını aktive edin ve bir hafta boyunca tüm yanıtları manuel onaylayın. Sisteme güven oluştuktan sonra otomatik yanıtlamayı düşük riskli kategorilerden başlatarak genişletin. Bu sabırlı yaklaşım, uzun vadede çok daha kararlı bir sistem ortaya çıkarıyor.

Bir yanıt yazın

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