On-Device RAG with Foundry Local: My Microsoft AI Innovators Project

In April I wrote here that I had been accepted into the Microsoft AI Innovators program. That post was 180 words, contained no code, and ended with the sentence "I'll write more about the process." Three months later, this is that post.

It is not an impression of the program. It is the architecture of the thing I actually built during the internship: rag-assistant, an assistant that answers questions over your own documents, on your own machine, with no internet connection. The repository is public — github.com/msgxr/rag-assistant. I started from Microsoft's Building Your First Local RAG Application with Foundry Local walkthrough and diverged from there. Every code block below is copied verbatim from the repo, which is why the comments are in Turkish.

Why an on-device model instead of a cloud API#

This looks like a technology preference, but it came out of four concrete constraints.

First, key distribution. The project is shared and demoed inside a team, and putting an API key on everyone's machine means owning the problem of that key leaking. Second, demo day: I did not want to depend on the room's Wi-Fi. Third, token cost — I run the evaluation set dozens of times, 26 questions per run. Fourth, documents never leaving the device.

The Foundry Local documentation lists four motivations under "Motivation for on-device AI": keeping data on the device, limited connectivity/offline operation, lowering token cost, and low latency. Three of them overlap with mine; key distribution was my own constraint, and low latency was never my priority. The runtime handles model acquisition, hardware acceleration and inference itself, running on ONNX Runtime; it picks a GPU or NPU when one is available and falls back to CPU. It adds roughly 20 MB to the application package.

What I pay for that is unambiguous: a quality ceiling. You accept the limits of a model small enough to sit on a laptop. That is also why I chose the aliases empirically rather than by reputation — qwen2.5-0.5b was faster but too weak in Turkish, and phi-3.5-mini climbed to a couple of minutes per answer on CPU. I settled on a Qwen2.5 1.5B class chat model with Qwen3-Embedding-0.6B for vectors, and pinned temperature at 0.2.

Writing this post also surfaced two mistakes in my own README. The prerequisites table says "Python 3.10+", but the SDK on PyPI requires 3.11 or newer. And I wrote "Windows or Apple Silicon Mac", while the documentation today also lists Linux. Both are on my correction list.

Five layers, one point of contact#

[User / UI]       ui_streamlit.py  ·  main.py (CLI)
      ↓
[Application]     generation.answer_query()
      ↓               ↳ prompts.build_user_message()
[RAG Retrieval]   retrieval.get_top_chunks()  ←  ingest.py (one-time)
      ↓                                              ↳ chunk_text()
[Data Layer]      rag.db (SQLite)  ←  db.py
      ↓
[AI Layer]        foundry_client.chat() / get_embedding()
                      ↳ Foundry Local Runtime — 100% on-device, offline

The one rigid rule in that diagram: only foundry_client.py touches the SDK. At runtime none of db, retrieval, generation or ui_streamlit imports foundry_local_sdk. There is one exception: check_setup.py imports the SDK directly for setup diagnostics. While writing this post I noticed that the repo's own data/architecture.md claims foundry_client.py is "the only file in the project that imports the Foundry Local SDK" — that sentence is wrong and needs fixing. That was not a stylistic choice — the published package is marked alpha, and response shapes move between versions. So I read the embedding response defensively:

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}"
    )

I know this function is ugly. The alternative is the project breaking in four places at once when the SDK bumps a minor version. And when the shape is genuinely unrecognised I raise an error that names the function to edit, rather than quietly returning an empty vector that would poison retrieval scores without any visible symptom. The contract I expose upward is four functions: warm_up(), get_embedding(text), chat(messages), shutdown().

Chunking: the shortest function I thought about the longest#

The first version split on paragraphs and I moved on. Then I looked at similarity scores in the Streamlit debug panel and saw single-line pieces like ## Başlık ranking at the top. The reason makes sense in hindsight: a very short text produces a tightly focused embedding, so it scores high against a related query — while carrying no information at all. One of my three retrieved chunks was being wasted.

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

The numbers: MAX_CHARS = 800 as the target upper bound, OVERLAP = 100 so a long paragraph split mid-thought does not lose context, MIN_CHARS = 150 below which a piece merges with its neighbour.

The trade-off became visible too. Larger chunks grounded answers better, but per-question latency on CPU went up, because context length is charged directly to inference time. The fix is still open on my list: send two chunks instead of three for short questions.

Data layer: one file instead of a vector database#

The schema is deliberately boring:

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()

Vectors live in a TEXT column as JSON. No vector database, no SQLite vector extension. The reason: the knowledge base is seven Markdown files. At that scale, comparing the query vector against every stored vector in plain Python is fast enough and costs zero extra dependencies. JSON also means I can read an embedding straight from the sqlite3 shell, which mattered while debugging. I know where this breaks and it is written in the README: comparison cost grows linearly with chunk count, and past a few thousand chunks a real index becomes necessary. Nothing outside the standard library's sqlite3 module gets installed.

One thing in ingest.py I fixed after the fact: ingestion now runs inside a single transaction — BEGIN, clear the table, insert every chunk, commit only when all of it succeeded. Before that, an embedding call failing halfway left me with a partially populated table, which silently corrupted the next evaluation run.

Retrieval is equally plain. No numpy; I wrote cosine similarity by hand:

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)

The two distinct failure modes of a small model#

This is where I learned the most. A 1.5B chat model fails in two different ways, and they need opposite responses.

The first: instead of answering, it echoes the instructions I gave it back at me. The second: it says "I don't have that information" even when the retrieved context clearly contains the answer. In my first version I treated both identically and printed the best chunk in either case. That was wrong, because a model saying "I don't know" is a legitimate answer — overriding it is exactly how you invite hallucination back in.

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}

There are three gates. If the top chunk scores below 0.45 the model is never called and an honest fallback is returned — off-topic questions land here. If instruction echo is detected the answer is untrustworthy, so the best passage is shown with its source file. If the model refused, the score decides: above 0.60 I treat the refusal as a model failure and show the passage; below it, I respect the model's judgement rather than manufacturing an answer.

That <= 120 character condition is also a bug fix. Before it, a long answer that happened to quote the phrase "no information" was misclassified as a refusal and thrown away.

The UI, and who pays for the first question#

Only one thing in the Streamlit layer is technically interesting: models load once when the page opens, not during the first question.

@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 exists precisely for this — the returned object behaves like a singleton, so a rerun does not reload the model. The sidebar's "show retrieved chunks" and "show similarity scores" checkboxes are not decoration either; that panel was the only place the short-chunk problem above was visible at all.

Evaluation: 26 questions, and what they do not measure#

eval/questions.yaml holds 26 questions: 18 answerable, 4 unanswerable (things genuinely absent from the knowledge base — weather, budget), and 4 edge cases (empty input, a single word, an extremely general question, and a very long one). The last recorded run in the repo is dated 2026-07-27 and reports 22/26 passed, 44.53 seconds per question on average.

I am quoting both numbers as recorded, which obliges me to say what they do not mean:

• ›This is not an accuracy measurement. The grader checks whether an expected keyword appears in the answer. Three of the four failures are exactly that artefact: the model described hardware acceleration without using the word "GPU", and explained the threshold without printing "0.45". The answers were acceptable; the metric was crude.
• ›The average is inflated. run_eval.py never calls warm_up(), so model load time is charged to the first question. Row 01 reads 128.28 seconds; the rest sit largely in a 27–70 second band.
• ›The hardware is unspecified. The record says "CPU-only laptop" and nothing more, so I cannot generalise these timings to any machine.
• ›It is a single run. Even at temperature 0.2 there is run-to-run variance, which I observed across recorded runs.

So I am not attaching a score to this project. What I measured is 26 questions I wrote myself. A defensible measurement needs either multiple acceptable keywords per question or a second model grading the answers — both are on the list, neither is done.

What is still missing#

The honest list:

• ›No reranking stage. Cosine score is currently the final word. A cross-encoder second pass would likely close the remainder of the short-chunk problem.
• ›No incremental ingestion. ingest.py wipes and rebuilds the table on every run. Fine at seven files, wasteful at seven hundred.
• ›No retry on refusal. The model occasionally declines an answerable question; retrying once would cover most of that.
• ›No concurrency. SQLite here is single-user; concurrent writers are not supported.
• ›Mixed-language answers. Bilingual prompting with very small models sometimes produces Turkish and English in one answer.

My next step is probably fixing the evaluation rather than the pipeline, because I cannot claim to have improved something I cannot measure. The code is public; if you find something wrong in it, open an issue.