Foundry Local ile Cihaz Üzerinde RAG: Microsoft AI Innovators Projem

Nisan ayında bu blogda Microsoft AI Innovators programına kabul aldığımı yazmıştım. O yazı 180 kelimeydi, tek bir kod satırı içermiyordu ve "süreçle ilgili daha fazla yazacağım" cümlesiyle bitiyordu. Aradan üç ay geçti; bu yazı o cümlenin karşılığı.

Anlatacağım şey program hakkında bir izlenim değil, staj boyunca kurduğum projenin mimarisi. Proje adı rag-assistant: kendi bilgisayarında, internet bağlantısı olmadan, kendi dökümanlarının üzerinde soru-cevap yapan bir asistan. Depo herkese açık — github.com/msgxr/rag-assistant. Başlangıç noktam Microsoft'un Building Your First Local RAG Application with Foundry Local yazısıydı; sonrası kendi kararlarımla ayrıştı. Aşağıdaki kod blokları depodan birebir alındı.

Neden bulut API'si değil de cihaz üzerinde model#

Bu bir teknoloji tercihi gibi görünüyor ama aslında dört tane somut kısıttan çıktı.

Birincisi anahtar dağıtımı. Proje ekip içinde paylaşılan ve demo edilen bir şey; herkesin makinesine bir API anahtarı koymak, o anahtarın sızmasını yönetmek anlamına geliyor. İkincisi demo günü: salonun internetine güvenmek istemedim. Üçüncüsü token maliyeti — değerlendirme setini onlarca kez koşturuyorum, her koşu 26 soru. Dördüncüsü verinin cihazdan çıkmaması.

Foundry Local dokümantasyonu "Motivation for on-device AI" başlığı altında dört gerekçe sayıyor: veriyi cihazda tutmak, kısıtlı bağlantı/çevrimdışı çalışmak, token maliyetini düşürmek, düşük gecikme. Üçü benim listemle örtüşüyor; anahtar dağıtımı benim kendi kısıtım, düşük gecikme ise benim önceliğim değildi, o yüzden doğru araca denk geldiğimi düşünüyorum. Çalışma zamanı model indirmeyi, donanım hızlandırmayı ve çıkarımı kendi üstlenip ONNX Runtime üzerinden koşuyor; GPU/NPU varsa onu, yoksa CPU'yu seçiyor. Uygulama paketine eklediği yük yaklaşık 20 MB.

Karşılığında ödediğim bedel net: kalite tavanı. Cihazda koşan bir modelin sınırlarını kabul ediyorsun. Alias seçimini de bu yüzden deneyerek yaptım — qwen2.5-0.5b daha hızlıydı ama Türkçede zayıf kaldı, phi-3.5-mini CPU'da cevap başına birkaç dakikaya çıktı. Sonuçta Qwen2.5 1.5B sınıfı sohbet modeli ve Qwen3-Embedding-0.6B gömme modelinde kaldım, sıcaklığı 0.2'ye çektim.

Bu yazıyı yazarken kendi README'imde iki hata buldum: gereksinimler tablosunda "Python 3.10+" yazıyor, ama PyPI'daki paket Python 3.11 ve üstünü istiyor; ayrıca ben "Windows veya Apple Silicon Mac" yazmışım, dokümantasyon bugün Linux'u da sayıyor. İkisi de düzeltme listemde.

Beş katman ve tek temas noktası#

[Kullanıcı / UI]  ui_streamlit.py  ·  main.py (CLI)
      ↓
[Uygulama]        generation.answer_query()
      ↓               ↳ prompts.build_user_message()
[RAG Getirme]     retrieval.get_top_chunks()  ←  ingest.py (tek seferlik)
      ↓                                              ↳ chunk_text()
[Veri]            rag.db (SQLite)  ←  db.py
      ↓
[AI]              foundry_client.chat() / get_embedding()
                      ↳ Foundry Local Runtime — %100 cihaz üzerinde, çevrimdışı

Bu diyagramın tek katı kuralı şu: SDK'ya yalnızca foundry_client.py dokunuyor. Çalışma zamanında db, retrieval, generation, ui_streamlit hiçbiri foundry_local_sdk'yı import etmiyor. Tek istisna check_setup.py: kurulum tanılaması için SDK'yı doğrudan import ediyor. Bu yazıyı yazarken deponun data/architecture.md dosyasında "projede SDK'yı import eden tek dosya" dediğimi gördüm — o cümle yanlış ve düzeltilmesi gerekiyor. Bunu baştan koymam şansa dayanmıyordu — SDK'nın PyPI sürümü alpha etiketli ve yanıt şekilleri sürümler arasında oynuyor. Bu yüzden gömme yanıtını savunmacı okuyorum:

def _extract_embedding(resp) -> list[float]:
    """
    Embedding response şekli SDK sürümüne göre değişebilir.
    Yaygın şekilleri sırayla dener; tanıyamazsa anlaşılır hata verir.
    """
    data = getattr(resp, "data", None)
    if data:
        first = data[0]
        emb = getattr(first, "embedding", None)
        if emb is not None:
            return list(emb)
        if isinstance(first, dict) and "embedding" in first:
            return list(first["embedding"])
    emb = getattr(resp, "embedding", None)
    if emb is not None:
        return list(emb)
    if isinstance(resp, (list, tuple)):
        return list(resp)
    raise RuntimeError(
        "Embedding response şekli tanınamadı; _extract_embedding'i SDK sürümüne "
        f"göre güncelle. Gelen tip: {type(resp)!r}"
    )

Bu fonksiyon çirkin, farkındayım. Ama alternatifi, SDK bir küçük sürüm atladığında projenin dört yerinden birden kırılması. Şeklin tanınmadığı durumda sessizce boş liste döndürmek yerine nerede bakılacağını söyleyen bir hata fırlatmayı tercih ettim. Dışa açtığım sözleşme sadece dört fonksiyon: warm_up(), get_embedding(text), chat(messages), shutdown().

Parçalama: en kısa fonksiyon, en çok düşündüğüm yer#

İlk sürümde metni paragraflara bölüp geçtim. Sonra Streamlit'in debug panelinde skorlara bakınca şunu gördüm: ## Başlık gibi tek satırlık parçalar üst sıralara çıkıyordu. Sebebi mantıklı — kısa bir metnin gömme vektörü çok odaklı oluyor, dolayısıyla ilgili sorguya yüksek benzerlik veriyor; ama taşıdığı bilgi sıfır. Getirilen üç parçadan biri boşa gidiyordu.

def chunk_text(text: str, max_chars: int = MAX_CHARS, overlap: int = OVERLAP,
               min_chars: int = MIN_CHARS) -> list[str]:
    """Önce paragraflara böl; çok uzun paragrafları örtüşmeli pencerelerle parçala."""
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    pieces: list[str] = []
    for para in paragraphs:
        if len(para) <= max_chars:
            pieces.append(para)
        else:
            start = 0
            while start < len(para):
                end = start + max_chars
                piece = para[start:end].strip()
                if piece:
                    pieces.append(piece)
                if end >= len(para):
                    break
                start = end - overlap

    # Tek başına anlamsız kalan kısa parçalar ("## Başlık", kod satırı) retrieval'ı
    # yanıltır: kısa metnin embedding'i çok odaklı olduğu için üst sıraya çıkar ama
    # bilgi taşımaz. Bu yüzden kısa parçaları bir önceki parçayla birleştiriyoruz.
    chunks: list[str] = []
    for piece in pieces:
        if chunks and (len(chunks[-1]) < min_chars or len(piece) < min_chars) \
                and len(chunks[-1]) + len(piece) + 2 <= max_chars + min_chars:
            chunks[-1] += "\n\n" + piece
        else:
            chunks.append(piece)
    return chunks

Sayılar şöyle: MAX_CHARS = 800 hedef üst sınır, OVERLAP = 100 uzun paragraf bölünürken bağlam kopmasın diye, MIN_CHARS = 150 bunun altındaki parçalar komşusuyla birleşiyor.

Takas de görünür oldu: parçalar büyüdüğünde cevaplar kaynağa daha iyi dayandı, ama CPU'da soru başına gecikme arttı. Bağlam uzunluğu doğrudan çıkarım süresine yazılıyor. Bunun çözümü hâlâ yapılacaklar listemde — kısa sorular için üç parça yerine iki parça göndermek.

Veri katmanı: vektör veritabanı yerine tek dosya#

Şema kasten sıkıcı:

def init_db(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS documents (
            id        INTEGER PRIMARY KEY AUTOINCREMENT,
            source    TEXT NOT NULL,
            content   TEXT NOT NULL,
            embedding TEXT NOT NULL
        )
        """
    )
    conn.commit()

Vektörler TEXT sütununda JSON metni olarak duruyor. Vektör veritabanı da, SQLite vektör eklentisi de kullanmadım. Gerekçe: bilgi tabanı yedi Markdown dosyası. Bu ölçekte sorgu vektörünü saklanan her vektörle Python içinde karşılaştırmak hem yeterince hızlı hem de sıfır ek bağımlılık. Ayrıca JSON metni sqlite3 komut satırından okunabiliyor, hata ayıklarken bu işe yaradı. Sınırı biliyorum ve README'de yazılı: karşılaştırma parça sayısıyla doğrusal büyüyor, birkaç bin parçadan sonra gerçek bir indeks gerekir. Standart kütüphanenin sqlite3 modülü dışında hiçbir şey kurulmuyor.

ingest.py içinde bir şeyi sonradan düzelttim: tüm ingestion artık tek transaction içinde. Önce BEGIN, sonra tabloyu temizle, parçaları yaz, hepsi bittiğinde commit. Öncesinde bir parçanın gömme işlemi ortada patladığında elimde yarı dolu bir tablo kalıyordu ve bu, sonraki değerlendirme koşusunu sessizce bozuyordu.

Getirme katmanı da aynı ölçüde yalın — numpy yok, kosinüs benzerliğini elle yazdım:

def _cosine(a: list[float], b: list[float]) -> float:
    dot = sum(x * y for x, y in zip(a, b))
    na = math.sqrt(sum(x * x for x in a))
    nb = math.sqrt(sum(y * y for y in b))
    if na == 0.0 or nb == 0.0:
        return 0.0
    return dot / (na * nb)

Küçük modelin iki ayrı bozulma biçimi#

Projede en çok öğrendiğim yer burası. 1.5B'lik bir sohbet modeli iki farklı şekilde bozuluyor ve ikisi farklı müdahale istiyor.

Birincisi: model cevap vermek yerine kendisine verdiğim talimatı geri yazıyor. İkincisi: bağlam gerçekten iyiyken bile "elimde bilgi yok" diyip pes ediyor. İlk sürümde ikisini aynı kutuya koyup her ikisinde de en iyi parçayı cevap olarak basıyordum. Bu yanlıştı, çünkü modelin "bilmiyorum" demesi meşru bir cevap; onu ezmek halüsinasyona kapı açıyor.

TOP_K = 3
MIN_RELEVANCE_SCORE = 0.45
STRONG_SCORE = 0.60   # bu skorun üstünde retrieval'a güven: model pes etse bile pasajı göster
FALLBACK_ANSWER = "Bu konuda elimdeki dökümanlarda bilgi yok."

# ...

    # Önce echo kontrolü: talimat tekrarı içinde fallback cümlesi de geçebilir
    if _matches(answer, ECHO_MARKERS):
        answer = _context_answer(chunks)
    elif _matches(answer, REFUSAL_MARKERS) and len(answer.strip()) <= 120:
        # Kısa cevap + refusal ifadesi = model "bilmiyorum" diyor. (Uzun cevaplar
        # "bilgi yok" ifadesini alıntılayan açıklamalar olabilir, onlara dokunma.)
        if _top_score(chunks) >= STRONG_SCORE:
            # Retrieval çok güçlüyken modelin pes etmesi model hatasıdır:
            # cevap uydurmadan, en alakalı pasajı kaynağıyla göster.
            answer = _context_answer(chunks)
        else:
            return {"answer": FALLBACK_ANSWER, "sources": [], "used_chunks": chunks}

Üç kapı var. En iyi parçanın skoru 0.45'in altındaysa model hiç çağrılmıyor, dürüst fallback dönüyor — alakasız sorular buraya düşüyor. Talimat tekrarı görülürse cevap güvenilmez, en iyi pasaj kaynağıyla gösteriliyor. Model pes ettiyse skora bakılıyor: 0.60 üstündeyse bu bir model hatası sayılıp pasaj gösteriliyor, altındaysa modelin kararına saygı duyuluyor.

O <= 120 karakter koşulu da bir hata düzeltmesi. Öncesinde "bilgi yok" ifadesini alıntılayarak açıklama yapan uzun cevapları fallback sanıp atıyordum.

Arayüz ve ilk sorunun bedeli#

Streamlit arayüzünde bir tek şey teknik olarak ilginç: modeller sayfa açılışında bir kez yükleniyor, ilk soru sırasında değil.

@st.cache_resource
def _warm_up() -> bool:
    """Modelleri sayfa açılırken bir kez yükler; ilk soru gecikmesiz cevaplanır."""
    fc.warm_up()
    return True

st.cache_resource tam bu iş için var — dönen nesne tekil davranıyor, her yeniden çizimde model tekrar yüklenmiyor. Kenar çubuğundaki "getirilen parçaları ve skorları göster" kutuları da dekor değil; yukarıda anlattığım kısa-parça sorununu görebildiğim tek yer oydu.

Ölçüm: 26 soru, ve neyi ölçmediği#

eval/questions.yaml içinde 26 soru var: 18 cevaplanabilir, 4 cevaplanamaz (bilgi tabanında olmayan şeyler — hava durumu, bütçe), 4 kenar durumu (boş girdi, tek kelime, çok genel, çok uzun soru). Depodaki son kayıtlı koşu 2026-07-27 tarihli ve şunu yazıyor: 22/26 geçti, ortalama 44.53 saniye/soru.

Bu iki sayıyı olduğu gibi yazıyorum, ama neyi ölçmediklerini de yazmam gerekiyor:

• ›Bu bir doğruluk ölçümü değil. Değerlendirici, cevapta beklenen bir anahtar kelimenin geçip geçmediğine bakıyor. Dört başarısızlığın üçü tam olarak bu yüzden: model donanım hızlandırmayı "GPU" kelimesini kullanmadan anlattı, eşiği "0.45" sayısını yazmadan açıkladı. Cevaplar kabul edilebilirdi, ölçüt kabaydı.
• ›Ortalama süre şişkin. run_eval.py başta warm_up() çağırmıyor, yani model yükleme süresi ilk soruya yazılıyor. Tabloda birinci satır 128.28 saniye, geri kalanlar büyük ölçüde 27–70 saniye bandında.
• ›Donanım belirsiz. Kayıt "CPU-only laptop" diyor, daha fazlası yok. O yüzden bu süreleri hiçbir donanım için genelleyemem.
• ›Tek koşu. Sıcaklık 0.2 olsa bile koşudan koşuya oynama var; bunu koşu kayıtlarında gözlemledim.

Bu yüzden bu projeye bir "başarı puanı" iliştirmiyorum. Ölçtüğüm şey, benim yazdığım 26 sorudan başkası değil. Düzgün bir ölçüm, ikinci bir modelle notlandırma ya da soru başına birden fazla kabul edilebilir anahtar kelime gerektiriyor; ikisi de yapılacaklar listemde.

Şimdi ne eksik#

Dürüst liste:

• ›Yeniden sıralama (rerank) katmanı yok. Şu an kosinüs skoru son söz. Cross-encoder tipi bir ikinci geçiş, kısa-parça probleminin kalan kısmını da çözebilir.
• ›Artımlı ingest yok. ingest.py her koşuda tabloyu siliyor ve baştan kuruyor. Yedi dosyada sorun değil, yedi yüzde israf.
• ›Pes etme durumunda tekrar deneme yok. Model nadiren cevaplanabilir bir soruyu reddediyor; bir kez yeniden denemek bunu büyük ölçüde kapatır.
• ›Eşzamanlılık yok. SQLite burada tek kullanıcılı; birden çok yazıcı desteklenmiyor.
• ›Karışık dil. Çok küçük modellerle iki dilli promptlama zaman zaman Türkçe-İngilizce karışık cevap üretiyor.

Bir sonraki adım muhtemelen değerlendirmeyi düzeltmek, çünkü ölçemediğim şeyi iyileştirdiğimi iddia edemem. Kod açık; yanlış bulan olursa depoda issue açabilir.