Build a local AI agent on a Raspberry Pi 5 with VectorAI DB

This article was originally published on the Actian blog.

In this tutorial, you'll build an AI agent that runs entirely on a Raspberry Pi 5 and remembers what you told it in earlier sessions, even after the database restarts. You'll need a Pi 5 with 8GB of RAM, an NVMe SSD, Docker, and uv.

The stack is two containers and one Python script. Ollama serves Qwen 2.5 1.5B for chat and nomic-embed-text for embeddings, and Actian VectorAI DB stores the agent's memory. In the test run, retrieval took 1.8 ms and the model answered in 3.6 seconds on the Pi's CPU, with no cloud API involved. The complete code is on GitHub.

Prerequisites

Part	What you need
Board	Raspberry Pi 5, 8GB
Storage	NVMe SSD on a PCIe HAT
Cooling	Official Active Cooler or equivalent
Power	Official 27W USB-C supply
OS	Raspberry Pi OS 64-bit (Trixie)

• You need Docker with Compose and uv for Python 3.11 or later.
• The 10,000 and 100,000 vector benchmarks need a VectorAI DB license or the 30-day trial, because the Community Edition caps collections below 5,000 vectors. The agent itself runs fine on Community.

How it works

Query flow from embedding through retrieval, generation, and storage, all on the board

• Ollama runs in a container capped at 4GB and used 1.58 GiB with both models loaded.
• VectorAI DB runs in a container capped at 2GB, used 125 MiB, and keeps vectors in an HNSW index on disk.
• agent.py is under 100 lines of plain Python with no agent framework.

Each message goes through five stages. The agent embeds it, retrieves related memories, injects them into the system prompt, generates an answer, and stores the message.

1. Enable memory limits and create the project

Raspberry Pi OS boots with the memory cgroup disabled, so Docker ignores mem_limit. Enable it and reboot.

sudo sed -i '1 s/$/ cgroup_enable=memory/' /boot/firmware/cmdline.txt && sudo reboot

Create the project folders. VectorAI DB runs as UID 999, so the data folder needs that owner.

mkdir -p pi5-vectorai-agent && cd pi5-vectorai-agent
mkdir -p data ollama results
sudo chown 999:999 data

Create docker-compose.yml.

services:
  vectorai:
    image: actian/vectorai:latest
    container_name: vectorai
    # the licence is tied to the host identity; a fixed hostname keeps it valid when the container is recreated
    hostname: vectorai
    ports:
      - "6573:6573" # REST
      - "6574:6574" # gRPC
    volumes:
      # vector data persists across restarts
      - ./data:/var/lib/actian-vectorai
    environment:
      - VECTORAI_LOG_LEVEL=info
      - ACTIAN_VECTORAI_ACCEPT_EULA=YES
    mem_limit: 2g
    restart: unless-stopped

  ollama:
    image: ollama/ollama:latest
    container_name: ollama
    ports:
      - "11434:11434"
    volumes:
      - ./ollama:/root/.ollama
    mem_limit: 4g
    restart: unless-stopped

The fixed hostname keeps a VectorAI DB license valid when the container is recreated. Start both services.

docker compose up -d
docker ps

VectorAI DB takes about 15 seconds to start. For other install options, see the Docker install guide.

2. Pull the models

docker exec ollama ollama pull qwen2.5:1.5b
docker exec ollama ollama pull nomic-embed-text

qwen2.5:1.5b is about 1.1GB and nomic-embed-text is about 274MB. Ollama keeps a model loaded for five minutes after its last use by default.

3. Install the Python dependencies

curl -LsSf https://astral.sh/uv/install.sh | sh
source $HOME/.local/bin/env

Create pyproject.toml

[project]
name = "pi5-vectorai-agent"
version = "0.1.0"
description = "Offline AI agent with semantic memory on a Raspberry Pi 5: Ollama + Actian VectorAI DB"
requires-python = ">=3.11"
# `uv run agent.py` reads this, creates .venv and installs these on first run
dependencies = [
    "requests",                # Ollama's HTTP API
    "actian-vectorai-client",  # VectorAI DB client (gRPC); brings numpy, protobuf, grpcio
    "numpy",                   # benchmark vectors
    "matplotlib",              # benchmark chart only
]

numpyand matplotlib are only for the benchmark. Install everything into .venv.

uv sync

4. Smoke test both services

uv run python -c 'from actian_vectorai import VectorAIClient; print(VectorAIClient("localhost:6574").health_check())'
curl -s localhost:11434/api/embed -d '{"model":"nomic-embed-text","input":"hello"}' | head -c 200; echo
curl -s localhost:11434/api/generate -d '{"model":"qwen2.5:1.5b","prompt":"Say hello in five words.","stream":false}'

The VectorAI DB health check reported version 1.0.2 in the test run. The other two calls confirm that Ollama can embed and generate. Check the memory each container uses.

docker stats --no-stream --format "{{.Name}}: {{.MemUsage}}"

The test run showed 1.58 GiB for Ollama and 125 MiB for VectorAI DB.

5. Create the memory collection

from actian_vectorai import Distance, VectorAIClient, VectorParams

with VectorAIClient("localhost:6574") as db:
    db.collections.create("agent_memory", vectors_config=VectorParams(size=768, distance=Distance.Cosine))

The collection uses 768-dimensional cosine vectors to match nomic-embed-text. This step is optional, because the agent creates the collection on its first run.

6. Build the agent

Createagent.py.

"""Local AI agent with semantic memory: Ollama (Qwen 2.5 1.5B) + Actian VectorAI DB."""
import os
import sys
import time

import requests
from actian_vectorai import Distance, PointStruct, VectorAIClient, VectorParams

OLLAMA = "http://localhost:11434"
VECTORAI = "localhost:6574"
CHAT_MODEL = "qwen2.5:1.5b"
EMBED_MODEL = "nomic-embed-text"
EMBED_DIM = 768            # nomic-embed-text output size
COLLECTION = "agent_memory"
TOP_K = 3
MIN_SCORE = 0.6            # unrelated memories score ~0.3-0.55 with nomic-embed-text, related ones 0.7+


def embed(text: str) -> list[float]:
    r = requests.post(f"{OLLAMA}/api/embed", json={"model": EMBED_MODEL, "input": text}, timeout=60)
    r.raise_for_status()
    return r.json()["embeddings"][0]


def chat(messages: list[dict]) -> str:
    # Ollama exposes an OpenAI-compatible endpoint, so any OpenAI client works here too
    r = requests.post(f"{OLLAMA}/v1/chat/completions", timeout=300,
                      json={"model": CHAT_MODEL, "messages": messages, "temperature": 0.3})
    r.raise_for_status()
    return r.json()["choices"][0]["message"]["content"]


# Tell the model what it runs on, so "what hardware is this?" gets a real answer
MODEL_FILE = "/proc/device-tree/model"
BOARD = open(MODEL_FILE).read().strip("\x00\n ") if os.path.exists(MODEL_FILE) else "unknown board"
HARDWARE = f"{BOARD}, {os.sysconf('SC_PAGE_SIZE') * os.sysconf('SC_PHYS_PAGES') / 1024**3:.0f} GB RAM"


def step(db: VectorAIClient, user_msg: str) -> str:
    # 1. Embed the incoming query on the Pi CPU
    t0 = time.perf_counter()
    qvec = embed(user_msg)
    t1 = time.perf_counter()

    # 2. Retrieve the closest past exchanges from VectorAI DB (HNSW index)
    hits = db.points.search(COLLECTION, vector=qvec, limit=TOP_K, with_payload=True)
    t2 = time.perf_counter()
    memories = [h.payload["text"] for h in hits if h.score >= MIN_SCORE]

    # 3. Inject retrieved memories into the system prompt, labelled as past notes
    #    so a small model doesn't mistake them for the current message
    context = "\n".join(f"- {m}" for m in memories) or "- (none)"
    messages = [
        {"role": "system", "content": f"You are a helpful assistant running fully offline on {HARDWARE} "
         f"with {CHAT_MODEL} via Ollama and Actian VectorAI DB for memory.\n"
         "Every message the user sends is saved to your memory automatically. If the user tells you "
         "a fact, confirm in one sentence that you will remember it.\n"
         "Answer in 1-3 sentences. Only use the notes below if they help answer the current message.\n\n"
         "Notes from earlier conversations:\n" + context},
        {"role": "user", "content": user_msg},
    ]

    # 4. Local inference through Ollama
    answer = chat(messages)
    t3 = time.perf_counter()

    # 5. Write the user's message back so future sessions can recall it. Only the user's
    #    side is stored: the facts come from the user, and storing the model's own replies
    #    would feed any wrong answer back into later prompts. The query vector is reused.
    db.points.upsert(COLLECTION, [PointStruct(
        id=time.time_ns() // 1000,  # microsecond timestamp as a unique integer ID
        vector=qvec,
        payload={"text": f"User said: {user_msg}", "ts": time.time()},
    )])
    t4 = time.perf_counter()

    print(f"  [embed {1000 * (t1 - t0):.0f} ms | retrieval {1000 * (t2 - t1):.1f} ms "
          f"({len(memories)} memories) | inference {t3 - t2:.1f} s | write-back {1000 * (t4 - t3):.0f} ms]")
    return answer


def main() -> None:
    with VectorAIClient(VECTORAI) as db:
        if db.collections.exists(COLLECTION):
            # VectorAI DB keeps collections on disk but only loads them when opened;
            # without this, search and count fail after every DB restart
            db.vde.open_collection(COLLECTION)
        else:
            db.collections.create(COLLECTION, vectors_config=VectorParams(size=EMBED_DIM, distance=Distance.Cosine))
        print(f"Agent ready on {HARDWARE} | LLM: {CHAT_MODEL} | "
              f"{db.points.count(COLLECTION)} memories in VectorAI DB. Ctrl+C to quit.")
        if len(sys.argv) > 1:  # one-shot mode: python agent.py "your question"
            print(step(db, " ".join(sys.argv[1:])))
            return
        try:  # Ctrl+C quits cleanly, even mid-answer
            while True:
                user_msg = input("\nyou> ").strip()
                if user_msg:
                    print(f"agent> {step(db, user_msg)}")
        except (EOFError, KeyboardInterrupt):
            print()


if __name__ == "__main__":
    main()

Here's what each stage does in the test run.

• Embed turns the message into a 768-dimensional vector in about 80 ms.
• Retrieve asks VectorAI DB for the three closest memories (TOP_K = 3) and keeps those scoring at least 0.6. Related memories scored 0.7 or higher, and unrelated ones scored 0.3 to 0.55.
• Inject adds the memories to the system prompt as notes from earlier conversations.
• Generate sends the prompt to Qwen through Ollama's OpenAI-compatible endpoint, at 12.1 tokens per second.
• Store writes the user's message back to VectorAI DB in about 6 ms, reusing the query vector.

The agent only stores the user's side of the conversation. Storing the model's replies would feed any wrong answer back into later prompts.

open_collection() matters after a restart. VectorAI DB keeps collections on disk and only loads them when opened, so search and count fail after a database restart without that call.

Run the agent.

uv run agent.py

To test memory, tell it something in one session, such as a greenhouse sensor reading and a pump in zone B going offline. Quit, restart the database if you like, and ask about it in a new session. In the test run, the agent recalled both facts with 1.8 ms of retrieval and a 3.6-second answer.

The benchmark

The benchmark loads synthetic, unit-normalized 768-dimensional vectors with cosine distance and default HNSW settings. It runs 10 warm-up queries, then times 100 top-5 searches at each corpus size.

"""Retrieval latency benchmark for VectorAI DB on Raspberry Pi 5.

Loads 1K / 10K / 100K vectors (768-dim, same size as nomic-embed-text), runs
100 timed top-k searches per corpus size, and writes p50/p95/p99 to CSV.

Vectors are synthetic (random, unit-normalized) so the 100K load finishes in
minutes instead of hours of on-device embedding. Random vectors are a harder
case for HNSW than real text embeddings, so treat these numbers as conservative.

The free Community edition caps the server at 5,000 vectors; without a licence
run it with --sizes 1000 2000 4000.

Usage: python benchmark.py [--sizes 1000 10000 100000] [--queries 100]
"""
import argparse
import csv
import platform
import statistics
import subprocess
import time
from pathlib import Path

import numpy as np
from actian_vectorai import Distance, PointStruct, VectorAIClient, VectorParams

DIM = 768
TOP_K = 5
BATCH = 500
WARMUP = 10


def hardware() -> dict:
    def read(path: str) -> str:
        try:
            return Path(path).read_text().strip("\x00\n ")
        except OSError:
            return "unknown"

    def cmd(*args: str) -> str:
        try:
            return subprocess.run(args, capture_output=True, text=True, timeout=5).stdout.strip()
        except (OSError, subprocess.SubprocessError):
            return "unknown"

    mem_kb = next((line.split()[1] for line in read("/proc/meminfo").splitlines() if line.startswith("MemTotal")), "0")
    # ask Docker where the vectorai container keeps its data, so the CSV names the drive
    # the database actually writes to (the NVMe), wherever this script was cloned
    data_dir = cmd("docker", "inspect", "-f", '{{range .Mounts}}{{if eq .Destination "/var/lib/actian-vectorai"}}'
                   '{{.Source}}{{end}}{{end}}', "vectorai") or str(Path(__file__).resolve().parent / "data")
    return {
        "board": read("/proc/device-tree/model"),
        "arch": platform.machine(),
        "kernel": platform.release(),
        "ram_gb": round(int(mem_kb) / 1024 / 1024, 1),
        "storage": cmd("findmnt", "-no", "SOURCE", "-T", data_dir) or "unknown",
        # vcgencmd prints "temp=48.3'C" / "throttled=0x0"; keep only the value
        "temp": cmd("vcgencmd", "measure_temp").split("=")[-1],
        "throttled": cmd("vcgencmd", "get_throttled").split("=")[-1],  # 0x0 means no throttling since boot
    }


def unit_vectors(n: int, rng: np.random.Generator) -> np.ndarray:
    v = rng.standard_normal((n, DIM)).astype(np.float32)
    return v / np.linalg.norm(v, axis=1, keepdims=True)


def load(db: VectorAIClient, name: str, n: int, rng: np.random.Generator) -> float:
    if db.collections.exists(name):
        db.collections.delete(name)
    db.collections.create(name, vectors_config=VectorParams(size=DIM, distance=Distance.Cosine))
    t0 = time.perf_counter()
    for start in range(0, n, BATCH):
        vecs = unit_vectors(min(BATCH, n - start), rng)
        db.points.upsert(name, [
            PointStruct(id=start + i, vector=v.tolist(), payload={"i": start + i})
            for i, v in enumerate(vecs)
        ])
    return time.perf_counter() - t0


def measure(db: VectorAIClient, name: str, queries: int, rng: np.random.Generator) -> list[float]:
    qs = unit_vectors(queries + WARMUP, rng).tolist()
    for q in qs[:WARMUP]:  # first searches load the collection into memory
        db.points.search(name, vector=q, limit=TOP_K)
    lat = []
    for q in qs[WARMUP:]:
        t0 = time.perf_counter()
        db.points.search(name, vector=q, limit=TOP_K)
        lat.append((time.perf_counter() - t0) * 1000)
    return lat


def pct(values: list[float], p: float) -> float:
    return float(np.percentile(values, p))


def main() -> None:
    ap = argparse.ArgumentParser()
    # Community edition (no licence) is capped at 5K vectors: use --sizes 1000 2000 4000
    ap.add_argument("--sizes", type=int, nargs="+", default=[1_000, 10_000, 100_000])
    ap.add_argument("--queries", type=int, default=100)
    ap.add_argument("--host", default="localhost:6574")
    ap.add_argument("--out", default="results/benchmark.csv")
    ap.add_argument("--keep", action="store_true", help="keep benchmark collections afterwards")
    args = ap.parse_args()

    hw = hardware()
    print("Hardware:", ", ".join(f"{k}={v}" for k, v in hw.items()))
    rng = np.random.default_rng(42)
    rows = []

    with VectorAIClient(args.host) as db:
        for n in args.sizes:
            name = f"bench_{n}"
            print(f"\n== {n:,} vectors ==")
            load_s = load(db, name, n, rng)
            print(f"  loaded in {load_s:.1f} s ({n / load_s:,.0f} vectors/s)")
            lat = measure(db, name, args.queries, rng)
            row = {
                "vectors": n, "dim": DIM, "top_k": TOP_K, "queries": len(lat),
                "p50_ms": round(pct(lat, 50), 2), "p95_ms": round(pct(lat, 95), 2),
                "p99_ms": round(pct(lat, 99), 2), "mean_ms": round(statistics.mean(lat), 2),
                "max_ms": round(max(lat), 2), "load_s": round(load_s, 1),
            }
            rows.append(row)
            print(f"  p50 {row['p50_ms']} ms | p95 {row['p95_ms']} ms | p99 {row['p99_ms']} ms")
            if not args.keep:
                db.collections.delete(name)

    hw_after = hardware()
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0]) + ["board", "ram_gb", "storage", "temp_end", "throttled_end"])
        w.writeheader()
        for r in rows:
            w.writerow({**r, "board": hw["board"], "ram_gb": hw["ram_gb"], "storage": hw["storage"],
                        "temp_end": hw_after["temp"], "throttled_end": hw_after["throttled"]})
    print(f"\nWrote {out}. End state: temp={hw_after['temp']}, throttled={hw_after['throttled']}")


if __name__ == "__main__":

main()

uv run benchmark.py            # 1K / 10K / 100K vectors, 100 queries each
uv run plot_benchmark.py       # writes results/benchmark.png

# Community edition (no licence) is capped at 5,000 vectors
uv run benchmark.py --sizes 1000 2000 4000

Vectors	p50 (ms)	p95 (ms)	p99 (ms)	Load time
1,000	2	1.97	2.01	0.4 s
10,000	3	3.12	3.47	28.7 s
100,000	4	4.75	4.98	526 s (8.8 min)

Figure 2: p50 and p99 latency vs. corpus size

A 100x larger corpus raised median latency from 1.90 ms to 4.17 ms, and p99 stayed under 5 ms at 100,000 vectors. Loading ran at about 190 vectors per second. Random vectors are harder for HNSW than embeddings of text, so expect equal or better numbers with your own data.

Hardware notes

• The whole stack used about 2.38 GiB of RAM, with the OS taking about 630 MB. A 4GB board wasn't tested.
• Qwen reloaded in 3.1 s from a warm cache and in 4.4 s after a reboot.
• The benchmark ran with the NVMe link at PCIe Gen 3, which you can enable with the command below.

echo "dtparam=pciex1_gen=3" | sudo tee -a /boot/firmware/config.txt && sudo reboot

• During the 100,000-vector run, the SoC went from 46.6°C to 68.6°C with no throttling. One run peaked at 77.4°C, and an underpowered supply triggered undervoltage (0x50000), so use the official 27W supply.

When You Hit the Pi's Limits

• Running several agents roughly doubles response time.
• A 7B model at 4-bit needs about 4.5GB for its weights, which exceeds the 4GB Ollama cap.
• The 2GB VectorAI DB cap is below Actian's general guidance of 8GB.
• The agent stores every user message, including questions, so memory grows with use.
• For production edge deployments, consider a Jetson Orin. You can also move inference to a box on your LAN by pointing OLLAMA at it.

OLLAMA = "http://192.168.1.50:11434"   # inference box on your LAN

Wrapping up

You built an offline agent on a Raspberry Pi 5 that combines local models, semantic memory, and persistent storage in VectorAI DB. The same arm64 image and API run on a Jetson Orin or an on-premises server, so the code carries over when you outgrow the board.

What's next

Start with VectorAI DB Community Edition and clone the companion repo.
Read the VectorAI DB Docker install guide for other platforms.
Ask questions in the Actian Discord.