LLM Router Örneği

from google import genai
import os
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from enum import Enum
from todo_manager import TodoManager
from weather_service import is_weather_question, handle_weather_question

# Adım 1: .env dosyasından ortam değişkenlerini yükle.
# Bu, API anahtarı gibi hassas bilgilerin güvenli şekilde alınmasını sağlar.
load_dotenv()

# Adım 2: Ortam değişkenlerinden Google AI API anahtarını al.
api_key = os.getenv("GOOGLE_AI_API_KEY")

# Adım 3: API anahtarı ile Google AI istemcisini başlat.
client = genai.Client(api_key=api_key)

# Adım 4: Mesaj yönlendirme sonuçlarını temsil eden bir enum tanımla.
# Bu enum, mesajın "Task", "Question" veya "Information" olarak sınıflandırılmasını sağlar.
class RoutingResult(Enum):
    TASK = "Task"
    QUESTION = "Question"
    INFORMATION = "Information"

# Adım 5: Mesaj yönlendirme sonucunu tutacak bir Pydantic modeli oluştur.
# Bu model, mesajın hangi kategoriye ait olduğunu belirtir.
class MessageForRouting(BaseModel):
    # Yönlendirme sonucu; "Task", "Question" veya "Information" değerlerinden biridir.
    result: RoutingResult = Field(description="Mesajın yönlendirme sonucu, ['Task', 'Question', 'Information']")
    
    # Adım 6: Modelin enum değerlerini direkt kullanabilmesi için yapılandır.
    class Config:
        use_enum_values = True

# Adım 7: Bir mesajı yönlendiren fonksiyonu tanımla.
def route_message(message: str) -> MessageForRouting:
    # Adım 7.1: AI'nın mesajı doğru kategoriye yönlendirmesi için bir prompt oluştur.
    prompt = f"""
    You are a helpful assistant that routes messages to the appropriate destination.
    The possible destinations are:
    - Task
    - Question
    - Information
    The message is: {message}
    """
    # Adım 7.2: Oluşturulan prompt'u, yanıtın JSON formatında ve MessageForRouting şemasına uygun olması beklentisiyle AI modeline gönder.
    response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=prompt,
        config={
            "response_mime_type": "application/json",
            "response_schema": MessageForRouting
        }
    )
    # Adım 7.3: Yanıtı ayrıştır ve yönlendirme sonucunu döndür.
    return response.parsed

# Adım 8: Genel soruları yanıtlamak için bir fonksiyon tanımla.
def answer_question(question: str) -> str:
    """Genel bir soruyu LLM kullanarak yanıtla."""
    # Adım 8.1: Soruyu yanıtlaması için AI'ya uygun bir prompt oluştur.
    prompt = f"""
    Please answer the following question concisely and accurately:
    {question}
    """
    # Adım 8.2: Daha yetenekli bir model kullanarak prompt'u gönder ve metin yanıtını döndür.
    response = client.models.generate_content(
        model="gemini-2.0-pro-exp-02-05",
        contents=prompt
    )
    return response.text

# Adım 9: Gelen mesajı yönlendirme ve uygun şekilde işleme fonksiyonunu tanımla.
def process_message(message: str):
    """Mesajı yönlendirerek uygun şekilde işle."""
    # Adım 9.1: Mesajı yönlendir ve mesajın tipi (Task, Question veya Information) belirle.
    routing_result = route_message(message)
    
    # Adım 9.2: Yönlendirme sonucuna göre ilgili işlemi yap.
    if routing_result.result == "Task":
        # Görev (Task) ise, mesajı to-do listesine ekle.
        todo_manager = TodoManager()
        todo_item = todo_manager.add_todo(message)
        return f"To-do listenize eklendi: {todo_item.task} (ID: {todo_item.id})"
    
    elif routing_result.result == "Question":
        # Soru (Question) ise, önce mesajın hava durumu sorusu olup olmadığını kontrol et.
        if is_weather_question(message):
            # Hava durumu sorusu ise, hava durumu servisini kullan.
            return handle_weather_question(message)
        else:
            # Aksi halde, genel soruyu LLM kullanarak yanıtla.
            return answer_question(message)
    
    elif routing_result.result == "Information":
        # Bilgi (Information) içeriyorsa, basitçe alınan bilgiyi onaylayan bir mesaj döndür.
        return "Bilginiz için teşekkürler: " + message
    
    else:
        # Tanımlanamayan durumlarda varsayılan yanıt.
        return "Bu mesajı nasıl işleyeceğimi bilmiyorum."

# Adım 10: Örnek kullanım - Fonksiyonları test etmek için.
if __name__ == "__main__":
    print("Soru örneği (hava durumu):")
    print(process_message("What is the weather in Istanbul?"))
    
    print("\nSoru örneği (genel):")
    print(process_message("What is the capital of France?"))
    
    print("\nGörev örneği:")
    print(process_message("Create a proposal for a new product"))
    
    print("\nBilgi örneği:")
    print(process_message("Tomorrow is your off day"))
    
    # Mevcut to-do listesini görüntüle.
    todo_manager = TodoManager()
    print("\nMevcut To-Do Listesi:")
    for todo in todo_manager.get_all_todos():
        status = "✓" if todo.completed else "□"
        print(f"{todo.id}. [{status}] {todo.task}")

from message_routing import process_message
from todo_manager import TodoManager
from weather_service import handle_weather_question
import argparse

# Adım 1: To-do listesindeki görevleri biçimli olarak ekrana yazdıran fonksiyonu tanımla.
def display_todos(todos):
    """To-do listesindeki görevleri biçimli olarak ekrana yazdırır."""
    # Eğer görev listesi boşsa bilgilendirici mesaj yazdır.
    if not todos:
        print("To-do listesinde hiç öğe yok.")
        return
    # Her bir görev için durum, oluşturulma tarihi ve tamamlanma tarihini biçimli olarak ekrana yazdır.
    for todo in todos:
        status = "✓" if todo.completed else "□"
        created = todo.created_at.strftime("%Y-%m-%d %H:%M")
        completed = todo.completed_at.strftime("%Y-%m-%d %H:%M") if todo.completed_at else "N/A"
        print(f"{todo.id}. [{status}] {todo.task}")
        print(f"   Oluşturulma: {created} | Tamamlanma: {completed}")

# Adım 2: Komut satırı argümanlarını işleyerek kullanıcıya to-do yönetimi, mesaj işleme ve hava durumu sorgulaması yapma imkanı sunan ana fonksiyonu oluştur.
def main():
    # Adım 2.1: Argüman ayrıştırıcıyı oluştur ve açıklamasını ekle.
    parser = argparse.ArgumentParser(description="To-Do Listesi ve Hava Durumu ile Akıllı Asistan")
    
    # Adım 2.2: Alt komutları tanımlamak için subparsers oluştur.
    subparsers = parser.add_subparsers(dest="command", help="Çalıştırılacak komut")
    
    # Adım 2.3: Yeni görev eklemek için "add" komutunu tanımla.
    add_parser = subparsers.add_parser("add", help="Yeni görev ekle")
    add_parser.add_argument("task", help="Görev açıklaması")
    
    # Adım 2.4: Bir görevi tamamlanmış olarak işaretlemek için "complete" komutunu tanımla.
    complete_parser = subparsers.add_parser("complete", help="Görevi tamamlandı olarak işaretle")
    complete_parser.add_argument("id", type=int, help="Tamamlanacak görev ID'si")
    
    # Adım 2.5: Bir görevi silmek için "delete" komutunu tanımla.
    delete_parser = subparsers.add_parser("delete", help="Görevi sil")
    delete_parser.add_argument("id", type=int, help="Silinecek görev ID'si")
    
    # Adım 2.6: Görevleri listelemek için "list" komutunu tanımla.
    list_parser = subparsers.add_parser("list", help="Görevleri listele")
    list_parser.add_argument("--all", action="store_true", help="Tüm görevleri göster")
    list_parser.add_argument("--completed", action="store_true", help="Sadece tamamlanmış görevleri göster")
    
    # Adım 2.7: Bir mesajı işleyip yönlendirmek için "process" komutunu tanımla.
    process_parser = subparsers.add_parser("process", help="Bir mesajı işle ve yönlendir")
    process_parser.add_argument("message", help="İşlenecek mesaj")
    
    # Adım 2.8: Belirtilen konum için hava durumu sorgulaması yapmak amacıyla "weather" komutunu tanımla.
    weather_parser = subparsers.add_parser("weather", help="Hava durumu bilgisini getir")
    weather_parser.add_argument("location", help="Hava durumu bilgisi istenen konum")
    
    # Adım 2.9: Doğrudan LLM'e soru sormak için "ask" komutunu tanımla.
    ask_parser = subparsers.add_parser("ask", help="LLM'e doğrudan soru sor")
    ask_parser.add_argument("question", help="Sorulacak soru")
    
    # Adım 3: Komut satırı argümanlarını ayrıştır.
    args = parser.parse_args()
    
    # Adım 4: To-do yönetimi için TodoManager örneğini oluştur.
    todo_manager = TodoManager()
    
    # Adım 5: Girilen komuta göre ilgili işlemleri gerçekleştir.
    if args.command == "add":
        # Yeni görev ekle.
        todo = todo_manager.add_todo(args.task)
        print(f"Eklendi: {todo.task} (ID: {todo.id})")
    
    elif args.command == "complete":
        # Belirtilen ID'li görevi tamamlandı olarak işaretle.
        todo = todo_manager.complete_todo(args.id)
        if todo:
            print(f"Tamamlandı olarak işaretlendi: {todo.task}")
        else:
            print(f"ID {args.id} ile eşleşen görev bulunamadı.")
    
    elif args.command == "delete":
        # Belirtilen ID'li görevi sil.
        success = todo_manager.delete_todo(args.id)
        if success:
            print(f"ID {args.id} numaralı görev silindi.")
        else:
            print(f"ID {args.id} ile eşleşen görev bulunamadı.")
    
    elif args.command == "list":
        # Listeleme: aktif, tamamlanmış veya tüm görevleri göster.
        if args.completed:
            print("Tamamlanmış Görevler:")
            display_todos(todo_manager.get_completed_todos())
        elif args.all:
            print("Tüm Görevler:")
            display_todos(todo_manager.get_all_todos())
        else:
            print("Aktif Görevler:")
            display_todos(todo_manager.get_active_todos())
    
    elif args.command == "process":
        # Mesajı işle ve yönlendir, ardından sonucu ekrana yazdır.
        result = process_message(args.message)
        print(result)
    
    elif args.command == "weather":
        # Belirtilen konum için hava durumu sorgusu yap.
        weather_question = f"What is the weather in {args.location}?"
        result = handle_weather_question(weather_question)
        print(result)
    
    elif args.command == "ask":
        # Doğrudan LLM'e soru sorarak yanıt al.
        result = process_message(args.question)
        print(result)
    
    else:
        # Hiçbir komut girilmediyse yardım mesajını göster.
        parser.print_help()

# Adım 6: Ana fonksiyonu çalıştır.
if __name__ == "__main__":
    main()

from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field
import json
import os

# Adım 1: Her bir to-do öğesini temsil eden TodoItem sınıfını tanımla.
# Bu sınıf, görev bilgilerini (id, görev açıklaması, oluşturulma tarihi, tamamlanma durumu, tamamlanma tarihi) saklar.
class TodoItem(BaseModel):
    id: int                           # Görevin benzersiz kimliği
    task: str                         # Görev açıklaması
    created_at: datetime = Field(default_factory=datetime.now)  # Görevin oluşturulma zamanı (varsayılan olarak anlık zaman)
    completed: bool = False           # Görevin tamamlanma durumu (varsayılan: tamamlanmadı)
    completed_at: Optional[datetime] = None  # Görevin tamamlanma zamanı (tamamlanmadıysa None)

    # Adım 2: JSON dönüşümü sırasında datetime nesnelerinin ISO formatına çevrilmesini sağlar.
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

# Adım 3: Görevleri yönetmek için TodoManager sınıfını tanımla.
class TodoManager:
    # Adım 3.1: Sınıfın örneği oluşturulurken kullanılacak dosya adını belirle ve mevcut görevleri yükle.
    def __init__(self, storage_file: str = "todos.json"):
        self.storage_file = storage_file  # Görevlerin saklanacağı dosya adı
        self.todos: List[TodoItem] = []     # Görev listesini başlat
        self.load_todos()                   # Var olan görevleri yükle

    # Adım 4: Görevleri saklama dosyasından yükleyen fonksiyonu tanımla.
    def load_todos(self):
        """Eğer mevcutsa, storage_file içerisindeki görevleri yükler."""
        if os.path.exists(self.storage_file):
            try:
                with open(self.storage_file, "r") as f:
                    todos_data = json.load(f)
                    self.todos = []
                    # Her bir kaydı TodoItem nesnesine dönüştür.
                    for todo_data in todos_data:
                        # String olarak saklanan tarih bilgilerini datetime nesnelerine çevir.
                        if "created_at" in todo_data and todo_data["created_at"]:
                            todo_data["created_at"] = datetime.fromisoformat(todo_data["created_at"])
                        if "completed_at" in todo_data and todo_data["completed_at"]:
                            todo_data["completed_at"] = datetime.fromisoformat(todo_data["completed_at"])
                        self.todos.append(TodoItem(**todo_data))
            except Exception as e:
                print(f"To-do'lar yüklenirken hata oluştu: {e}")
                self.todos = []

    # Adım 5: Görev listesini saklama dosyasına kaydeden fonksiyonu tanımla.
    def save_todos(self):
        """Görev listesini storage_file dosyasına kaydeder."""
        try:
            with open(self.storage_file, "w") as f:
                todos_json = [todo.dict() for todo in self.todos]
                json.dump(todos_json, f, default=str, indent=2)
        except Exception as e:
            print(f"To-do'lar kaydedilirken hata oluştu: {e}")

    # Adım 6: Yeni bir görev ekleyen fonksiyonu tanımla.
    def add_todo(self, task: str) -> TodoItem:
        """Yeni bir görev ekler ve eklenen görevi döner."""
        # Yeni bir ID oluşturmak için mevcut görevlerin varlığına göre basit artış yöntemi kullan.
        new_id = 1 if not self.todos else max(todo.id for todo in self.todos) + 1
        
        # Yeni görev nesnesini oluştur.
        todo = TodoItem(id=new_id, task=task)
        self.todos.append(todo)  # Görev listesini güncelle
        self.save_todos()        # Değişiklikleri dosyaya kaydet
        return todo

    # Adım 7: Belirtilen ID'ye sahip görevi tamamlandı olarak işaretleyen fonksiyonu tanımla.
    def complete_todo(self, todo_id: int) -> Optional[TodoItem]:
        """Bir görevi tamamlandı olarak işaretler ve güncellenmiş görevi döner."""
        for todo in self.todos:
            if todo.id == todo_id:
                todo.completed = True         # Görevin tamamlandığını işaretle
                todo.completed_at = datetime.now()  # Tamamlanma zamanını ayarla
                self.save_todos()             # Değişiklikleri kaydet
                return todo
        return None  # Belirtilen ID'ye sahip görev bulunamazsa None döner

    # Adım 8: Belirtilen ID'ye sahip görevi silen fonksiyonu tanımla.
    def delete_todo(self, todo_id: int) -> bool:
        """Belirtilen ID'ye sahip görevi siler ve silme işleminin başarılı olup olmadığını döner."""
        for i, todo in enumerate(self.todos):
            if todo.id == todo_id:
                del self.todos[i]  # Görevi listeden çıkar
                self.save_todos()  # Değişiklikleri kaydet
                return True
        return False

    # Adım 9: Tüm görevleri döndüren fonksiyonu tanımla.
    def get_all_todos(self) -> List[TodoItem]:
        """Tüm görevleri döner."""
        return self.todos

    # Adım 10: Tamamlanmamış (aktif) görevleri döndüren fonksiyonu tanımla.
    def get_active_todos(self) -> List[TodoItem]:
        """Aktif (tamamlanmamış) görevleri döner."""
        return [todo for todo in self.todos if not todo.completed]

    # Adım 11: Tamamlanmış görevleri döndüren fonksiyonu tanımla.
    def get_completed_todos(self) -> List[TodoItem]:
        """Tamamlanmış görevleri döner."""
        return [todo for todo in self.todos if todo.completed]

# Adım 12: Örnek kullanım - Bu bölüm, sınıfın işlevselliğini test etmek içindir.
if __name__ == "__main__":
    todo_manager = TodoManager()
    todo_manager.add_todo("Market alışverişi yap")
    todo_manager.add_todo("Proje raporunu tamamla")
    print("Tüm görevler:", todo_manager.get_all_todos())

import requests
from typing import Optional, Dict, Any
import re

# Adım 1: Hava durumu ile ilgili sorulardan konum bilgisini çıkarmak için fonksiyon.
def extract_location(question: str) -> Optional[str]:
    """Hava durumu sorusundan konum bilgisini çıkarır."""
    # Adım 1.1: Farklı kalıplara göre konum bilgisi çıkarma deseni tanımla.
    patterns = [
        r"weather\s+in\s+([A-Za-z\s]+)(?:\?|$)",
        r"temperature\s+in\s+([A-Za-z\s]+)(?:\?|$)",
        r"how\s+(?:hot|cold|warm)\s+is\s+(?:it\s+in\s+)?([A-Za-z\s]+)(?:\?|$)",
        r"what'?s\s+the\s+weather\s+(?:like\s+)?(?:in\s+)?([A-Za-z\s]+)(?:\?|$)"
    ]
    
    # Adım 1.2: Belirlenen kalıpları kullanarak soruda konum bilgisini ara.
    for pattern in patterns:
        match = re.search(pattern, question, re.IGNORECASE)
        if match:
            return match.group(1).strip()
    
    # Adım 1.3: Hiçbir eşleşme bulunamazsa None döndür.
    return None

# Adım 2: Konum adını enlem ve boylam bilgisine dönüştüren fonksiyon.
def geocode_location(location: str) -> Optional[Dict[str, float]]:
    """Konum adını, enlem ve boylam bilgisine çevirir."""
    url = "https://nominatim.openstreetmap.org/search"
    params = {
        "q": location,
        "format": "json",
        "limit": 1
    }
    # Adım 2.1: Nominatim servisi tarafından isteklerin kabul edilebilmesi için User-Agent belirle.
    headers = {
        "User-Agent": "WeatherApp/1.0"
    }
    
    try:
        response = requests.get(url, params=params, headers=headers)
        if response.status_code == 200:
            data = response.json()
            if data and len(data) > 0:
                # Adım 2.2: Enlem, boylam ve gösterim adını içeren veriyi döndür.
                return {
                    "latitude": float(data[0]["lat"]),
                    "longitude": float(data[0]["lon"]),
                    "display_name": data[0]["display_name"]
                }
    except Exception as e:
        print(f"Konum bilgisini geocode ederken hata oluştu: {e}")
    
    # Adım 2.3: İşlem başarısız olursa None döndür.
    return None

# Adım 3: Belirtilen enlem ve boylam için hava durumu verilerini getiren fonksiyon.
def get_weather(latitude: float, longitude: float) -> Optional[Dict[str, Any]]:
    """
    Belirtilen koordinatlar için Open-Meteo API'sinden bugünün hava durumu verilerini alır.
    """
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": latitude,
        "longitude": longitude,
        "daily": "temperature_2m_max,temperature_2m_min,precipitation_sum",
        "timezone": "auto",
        "forecast_days": 1  # Sadece bugünkü tahmin bilgilerini al
    }
    
    try:
        response = requests.get(url, params=params)
        if response.status_code == 200:
            data = response.json()
            # Adım 3.1: Bugünün tarihini ve hava durumu verilerini çıkar.
            date = data.get("daily", {}).get("time", [])[0]
            temp_max = data.get("daily", {}).get("temperature_2m_max", [])[0]
            temp_min = data.get("daily", {}).get("temperature_2m_min", [])[0]
            precipitation = data.get("daily", {}).get("precipitation_sum", [])[0]
            
            # Adım 3.2: Verileri ve birim bilgilerini içeren sonucu oluştur.
            result = {
                "date": date,
                "max_temperature": temp_max,
                "min_temperature": temp_min,
                "precipitation": precipitation,
                "units": {
                    "temperature": data.get("daily_units", {}).get("temperature_2m_max", "°C"),
                    "precipitation": data.get("daily_units", {}).get("precipitation_sum", "mm")
                }
            }
            return result
        else:
            print(f"Hava durumu API isteği, {response.status_code} durum kodu ile başarısız oldu.")
    except Exception as e:
        print(f"Hava durumu verileri alınırken hata oluştu: {e}")
    
    # Adım 3.3: Hata durumunda None döndür.
    return None

# Adım 4: Verilen sorunun hava durumu sorusu olup olmadığını kontrol eden fonksiyon.
def is_weather_question(question: str) -> bool:
    """Soru, hava durumu ile ilgili mi kontrol eder."""
    weather_keywords = [
        "weather", "temperature", "hot", "cold", "warm", "rain", "sunny", 
        "forecast", "climate", "humidity", "precipitation"
    ]
    question_lower = question.lower()
    # Adım 4.1: Sorunun içeriğinde hava durumu ile ilgili anahtar kelimeler var mı kontrol et.
    return any(keyword in question_lower for keyword in weather_keywords)

# Adım 5: Belirtilen konum için hava durumu verilerini döndüren fonksiyon.
def get_weather_for_location(location: str) -> Optional[Dict[str, Any]]:
    """Belirtilen konum için hava durumu verilerini getirir."""
    geo_data = geocode_location(location)
    if geo_data:
        weather_data = get_weather(geo_data["latitude"], geo_data["longitude"])
        if weather_data:
            # Adım 5.1: Hava durumu verilerine konum bilgisini ekle.
            weather_data["location"] = geo_data["display_name"]
            return weather_data
    return None

# Adım 6: Hava durumu verilerini, okunabilir bir metin halinde biçimlendiren fonksiyon.
def format_weather_response(weather_data: Dict[str, Any]) -> str:
    """Hava durumu verilerini, okunabilir bir metin haline getirir."""
    location = weather_data.get("location", "istenen konum")
    date = weather_data.get("date", "bugün")
    max_temp = weather_data.get("max_temperature")
    min_temp = weather_data.get("min_temperature")
    precipitation = weather_data.get("precipitation")
    temp_unit = weather_data.get("units", {}).get("temperature", "°C")
    precip_unit = weather_data.get("units", {}).get("precipitation", "mm")
    
    response = f"{location} için {date} hava durumu:\n"
    response += f"- Maksimum sıcaklık: {max_temp}{temp_unit}\n"
    response += f"- Minimum sıcaklık: {min_temp}{temp_unit}\n"
    response += f"- Yağış miktarı: {precipitation}{precip_unit}"
    
    return response

# Adım 7: Hava durumu ile ilgili soruyu işleyip, yanıt döndüren ana fonksiyon.
def handle_weather_question(question: str) -> str:
    """Hava durumu sorusunu işleyerek cevap döndürür."""
    # Adım 7.1: Sorudan konum bilgisini çıkar.
    location = extract_location(question)
    
    # Adım 7.2: Konum bilgisi çıkarılamadıysa bilgilendirici mesaj döndür.
    if not location:
        return "Hangi konum hakkında bilgi istediğinizi belirleyemedim. Lütfen bir şehir veya bölge belirtin."
    
    # Adım 7.3: Belirtilen konum için hava durumu verilerini al.
    weather_data = get_weather_for_location(location)
    
    # Adım 7.4: Hava durumu verileri alınabiliyorsa, biçimlendirilmiş yanıtı döndür.
    if weather_data:
        return format_weather_response(weather_data)
    else:
        return f"Üzgünüm, {location} için hava durumu bilgisi alınamadı. Lütfen başka bir konum deneyin."

# Adım 8: Örnek kullanım - Fonksiyonların test edilmesi için.
if __name__ == "__main__":
    test_questions = [
        "What is the weather in Tokyo?",
        "How hot is it in New York?",
        "Tell me the temperature in Paris",
        "Is it raining in London?"
    ]
    
    for question in test_questions:
        print(f"Soru: {question}")
        if is_weather_question(question):
            print(f"Yanıt: {handle_weather_question(question)}")
        else:
            print("Bu bir hava durumu sorusu değil.")
        print()