Run a Local AI Agent in Production for $0 in API Fees: FastAPI, Ollama and ChromaDB

Most agent prototypes start the same way: an OpenAI or Anthropic key, a few prompts, a demo that works. Then the agent goes to production and every user request turns into several model calls. Retrieval, reasoning steps and the final answer all bill per token, and the bill grows with traffic and with every extra step your agent takes.

Running the model yourself changes the cost model. You pay for hardware (or a GPU server), not per token. Your documents also never leave your machine. This tutorial builds a small but complete local agent:

• Ollama runs the chat model and the embedding model locally.
• ChromaDB stores your documents and handles semantic search.
• FastAPI exposes /ingest and /ask endpoints.
• A minimal agent loop lets the model decide when to search the knowledge base, using Ollama's native tool calling.

No LangChain, no LlamaIndex: about 150 lines of plain Python you can read in one sitting.

Architecture

client ──HTTP──► FastAPI (/ingest, /ask)
                    │
                    ├── agent loop ──► Ollama /api/chat   (llama3.1:8b, tool calling)
                    │        │
                    │        └─ search_docs tool ──► ChromaDB (persistent, cosine)
                    │
                    └── embeddings ──► Ollama /api/embed  (nomic-embed-text)

The model never touches the database directly. It asks for a search_docs call, our code runs the search, and the results go back into the conversation as a tool message. That boundary is what keeps the agent debuggable.

Requirements

• Python 3.11+ (tested on 3.13)
• Ollama installed and running (ollama serve)
• About 8 GB of free RAM or VRAM for an 8B model at the default 4-bit quantization. It runs on CPU too, just slower.

Pull the two models:

ollama pull llama3.1:8b        # chat model with tool-calling support
ollama pull nomic-embed-text   # embedding model

Pin the Python dependencies (requirements.txt):

fastapi==0.149.0
uvicorn[standard]==0.54.0
httpx==0.28.1
chromadb==1.5.9
pydantic==2.14.1

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

The project is four files: ollama_client.py, store.py, agent.py, main.py.

Step 1: A small async client for Ollama

Talk to Ollama's HTTP API directly with httpx. One shared client gives you connection pooling, and an explicit timeout object means a stuck generation can't hang a worker forever.

# ollama_client.py
import asyncio
import os

import httpx

OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434")
CHAT_MODEL = os.getenv("CHAT_MODEL", "llama3.1:8b")
EMBED_MODEL = os.getenv("EMBED_MODEL", "nomic-embed-text")

# One shared client: connection pooling, explicit timeouts.
# Generation can be slow on CPU, so the read timeout is generous.
_client = httpx.AsyncClient(
    base_url=OLLAMA_URL,
    timeout=httpx.Timeout(connect=5.0, read=120.0, write=10.0, pool=10.0),
)
# Cap in-flight generations; extra requests wait here instead of piling up on the GPU.
_gpu_slots = asyncio.Semaphore(int(os.getenv("MAX_PARALLEL_LLM", "2")))


async def embed(texts: list[str]) -> list[list[float]]:
    r = await _client.post("/api/embed", json={"model": EMBED_MODEL, "input": texts})
    r.raise_for_status()
    return r.json()["embeddings"]


async def chat(messages: list[dict], tools: list[dict] | None = None) -> dict:
    payload = {
        "model": CHAT_MODEL,
        "messages": messages,
        "stream": False,
        "keep_alive": "30m",  # keep weights loaded between requests
        "options": {"temperature": 0.2, "num_ctx": 8192},
    }
    if tools:
        payload["tools"] = tools
    async with _gpu_slots:
        r = await _client.post("/api/chat", json=payload)
    r.raise_for_status()
    return r.json()["message"]

Two details matter in production:

• The semaphore. A single GPU can only run a few generations at once. Without a cap, 50 concurrent requests all hit Ollama together, and every one of them gets slow. With MAX_PARALLEL_LLM=2, extra requests queue inside your app, where you can time them out cleanly. Match it to Ollama's OLLAMA_NUM_PARALLEL setting.
• keep_alive. By default Ollama unloads a model after a few idle minutes, and the next request pays the load time again. Keeping it loaded for 30 minutes avoids that cold start on a dedicated box.

Step 2: ChromaDB as the knowledge base

Chroma can embed text for you, but its default embedding function downloads a model on first use. Here we compute embeddings with Ollama and pass the vectors in ourselves. One embedding model, fully local, and the same one for documents and queries.

# store.py
import asyncio

import chromadb

from ollama_client import embed

_db = chromadb.PersistentClient(path="./chroma_data")
# embedding_function=None: we pass our own vectors from Ollama,
# so Chroma never downloads its default embedding model.
_docs = _db.get_or_create_collection(
    "docs", embedding_function=None, metadata={"hnsw:space": "cosine"}
)


async def add_documents(ids: list[str], texts: list[str], source: str) -> None:
    vectors = await embed(texts)
    await asyncio.to_thread(
        _docs.upsert,
        ids=ids,
        documents=texts,
        embeddings=vectors,
        metadatas=[{"source": source} for _ in ids],
    )


async def search(query: str, k: int = 4) -> list[dict]:
    [vector] = await embed([query])
    res = await asyncio.to_thread(_docs.query, query_embeddings=[vector], n_results=k)
    return [
        {"text": doc, "source": meta["source"], "distance": round(dist, 3)}
        for doc, meta, dist in zip(res["documents"][0], res["metadatas"][0], res["distances"][0])
    ]

PersistentClient writes to ./chroma_data, so the index survives restarts. Chroma's Python client is synchronous, so the calls run in asyncio.to_thread to avoid blocking the event loop while a search runs. upsert makes re-ingesting the same document IDs safe.

Step 3: The agent loop

This is the "agent" part: the model gets a tool, decides whether to call it, sees the results and either calls it again or answers. Ollama returns tool calls in message.tool_calls for models that support them (Llama 3.1, Qwen 2.5 and others).

# agent.py
import json

from ollama_client import chat
from store import search

SYSTEM = (
    "You are a support agent for an internal knowledge base. "
    "Use the search_docs tool before answering factual questions. "
    "Answer only from the retrieved text and cite the source. "
    "If the documents do not contain the answer, say you don't know."
)

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "search_docs",
            "description": "Semantic search over the knowledge base.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string", "description": "What to look up"}},
                "required": ["query"],
            },
        },
    }
]


async def run_agent(question: str, max_steps: int = 4) -> dict:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
    sources: set[str] = set()

    for _ in range(max_steps):
        msg = await chat(messages, tools=TOOLS)
        messages.append(msg)
        calls = msg.get("tool_calls") or []
        if not calls:
            return {"answer": msg.get("content", ""), "sources": sorted(sources)}

        for call in calls:
            fn = call["function"]
            args = fn.get("arguments") or {}
            if isinstance(args, str):  # some models return a JSON string
                args = json.loads(args)
            if fn["name"] == "search_docs":
                hits = await search(args.get("query", question))
                sources.update(h["source"] for h in hits)
                result = json.dumps(hits)
            else:
                result = json.dumps({"error": f"unknown tool {fn['name']}"})
            messages.append({"role": "tool", "content": result})

    return {"answer": "Stopped: too many tool calls.", "sources": sorted(sources)}

The loop is deliberately boring:

• max_steps caps how often the model can call tools. A confused model will otherwise search forever.
• Sources are collected from the actual search hits, not from what the model claims, so the citations you return are real.
• Unknown tool names come back as an error message instead of crashing. The model usually recovers on the next step.

Adding a tool means adding one schema entry to TOOLS and one elif branch. Keep tools narrow and read-only until you trust the model's choices.

Step 4: The FastAPI service

# main.py
import asyncio

import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from agent import run_agent
from ollama_client import CHAT_MODEL
from store import add_documents

app = FastAPI(title="Local AI agent")


class Doc(BaseModel):
    id: str
    text: str = Field(min_length=1)


class IngestRequest(BaseModel):
    source: str
    docs: list[Doc] = Field(min_length=1, max_length=256)


class AskRequest(BaseModel):
    question: str = Field(min_length=3, max_length=2000)


class AskResponse(BaseModel):
    answer: str
    sources: list[str]


@app.post("/ingest")
async def ingest(req: IngestRequest):
    await add_documents([d.id for d in req.docs], [d.text for d in req.docs], req.source)
    return {"ingested": len(req.docs)}


@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest):
    try:
        # Hard ceiling for the whole agent loop, independent of per-call timeouts.
        return await asyncio.wait_for(run_agent(req.question), timeout=180)
    except asyncio.TimeoutError:
        raise HTTPException(504, "Agent timed out")
    except httpx.HTTPError as e:
        raise HTTPException(502, f"Ollama error: {e.__class__.__name__}")


@app.get("/health")
async def health():
    return {"status": "ok", "model": CHAT_MODEL}

Run it:

uvicorn main:app --port 8000

Load some documents and ask a question:

curl -s -X POST localhost:8000/ingest -H 'content-type: application/json' -d '{
  "source": "handbook.md",
  "docs": [
    {"id": "refunds-1", "text": "Refunds are issued within 14 days of purchase, no questions asked."},
    {"id": "support-1", "text": "Support replies within one business day, Monday to Friday."}
  ]
}'

curl -s -X POST localhost:8000/ask -H 'content-type: application/json' \
  -d '{"question": "How long do I have to ask for a refund?"}'

You get JSON back: the answer plus a sources list such as ["handbook.md"]. The first request is slower because Ollama loads the model; after that, latency depends on your hardware and model size.

For real documents, split them into chunks of a few hundred words before ingesting (one chunk per Doc), with IDs like file.md#3. Retrieval quality depends far more on sensible chunks than on any setting in this code.

Production notes

Concurrency. Run uvicorn with one worker per GPU box at first. Each worker has its own semaphore, so with 4 workers and MAX_PARALLEL_LLM=2 you can have 8 generations in flight. Size it on purpose. FastAPI's async handlers can hold many waiting requests cheaply; the GPU is the real bottleneck.

Timeouts at three levels. A per-call read timeout (httpx), a ceiling for the whole agent loop (asyncio.wait_for), and a timeout at your reverse proxy. Return 504 rather than letting clients hang.

Model choice. Start with an 8B instruct model that supports tool calling (llama3.1:8b, qwen2.5:7b). Check whether it calls the tool when it should and refuses when the docs don't contain the answer. Only move to a larger model if the small one fails on your real questions, because bigger models cost a lot more memory and latency.

Memory. Model weights dominate memory use. num_ctx matters too: the KV cache grows with context length and with the number of parallel requests. If Ollama starts swapping or falls back to CPU, lower num_ctx or MAX_PARALLEL_LLM before buying hardware. Chroma's own footprint is small for thousands of chunks.

Security. This service has no authentication. Before you expose it, put it behind an API key or your existing auth, add per-client rate limiting, and cap request sizes (the Pydantic max_length limits are a start). An open LLM endpoint is an easy way to let strangers use your GPU.

Observability. Log per request: steps taken, tool calls, retrieved sources and total latency. When an answer is wrong, this tells you whether retrieval or the model failed.

What does it actually cost?

"$0 in API fees" is true. "$0" is not: you pay for the machine. A simple way to compare:

• Hosted API per month: input tokens × input price + output tokens × output price.
• Self-hosted per month: server or GPU rental (or hardware amortized), plus power and your maintenance time. This is roughly flat no matter how many tokens you process, up to the hardware's throughput.

As an example, take 100,000 agent runs a month at about 1,500 input and 500 output tokens each: 150M input and 50M output tokens. With OpenAI's list prices when the original version of this post was written (GPT-4o at $2.50 / $10 per million input/output tokens, GPT-4o-mini at $0.15 / $0.60), that is about $875/month for GPT-4o and about $52/month for GPT-4o-mini. Check current pricing before you decide; it changes often. A rented GPU server in the ~$120/month range sits between the two.

So be honest with yourself about why you're going local:

• If a small hosted model is good enough, the API is often cheaper than a GPU at moderate volume.
• Local wins when you would otherwise need a large model at high volume, when costs must be predictable, or when data can't leave your infrastructure. In many companies, that last reason decides it.

Wrapping up

You now have a local agent that ingests documents, decides when to search, answers with real sources and fails cleanly when Ollama is slow or down. Everything runs on one machine with five pinned dependencies.

I build small, zero-dependency developer tools at PyForge. If you mix local and hosted models, LLMCostGuard adds spending guardrails for LLM calls, and RateShield handles the rate limiting this service needs before it goes public. All of them are included in the All-Access Pass.