A chat service with a tool-using AI assistant that streams responses progressively.
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
cp .env.example .env
# Edit .env with your LLM_API_KEY
uvicorn app.main:app --reloadOpen http://localhost:8000 in your browser.
| Variable | Default | Description |
|---|---|---|
LLM_API_KEY |
(required) | API key for your LLM provider |
LLM_MODEL |
google/gemini-3-flash-preview |
Model identifier |
LLM_BASE_URL |
https://openrouter.ai/api/v1 |
OpenAI-compatible API base URL |
Works with any OpenAI-compatible provider — change LLM_BASE_URL to https://api.openai.com/v1 for OpenAI, or point at a local Ollama instance.
pytestAll tests run offline with no API key — the LLM is replaced by a mock adapter in tests.
┌──────────────┐ POST /chat ┌────────────────────────────────┐
│ │ {messages: [...]} │ FastAPI Server │
│ Browser │ ──────────────────────────► │ │
│ │ │ ┌─────────┐ ┌───────────┐ │
│ index.html │ ◄────────────────────────── │ │ LLM │───►│ Tools │ │
│ chat.js │ SSE stream │ │ Adapter │ │ Registry │ │
│ styles.css │ (text, tool, done events) │ └────┬────┘ └───────────┘ │
│ │ │ │ │
└──────────────┘ └───────┼────────────────────────┘
│
▼
┌─────────────────┐
│ LLM Provider │
│ (OpenRouter / │
│ OpenAI / etc) │
└─────────────────┘
Components:
- Browser — sends full conversation history, reads SSE stream via
fetch+ReadableStream - FastAPI Server — stateless HTTP handler, validates requests, returns a
StreamingResponse - LLM Adapter (
app/llm.py) — calls any OpenAI-compatible API, handles the tool loop (call LLM → execute tools → call LLM again with results) - Tools Registry (
app/tools/) — calculator (ast-safe math), weather (Open-Meteo API), converter (unit conversion) - SSE Stream (
app/stream.py) — formats events asdata: {json}\n\n, wraps errors so the client always gets adoneevent
See specs/ for the full design specifications.
Real LLM with tool use via function calling. The LLM decides when to invoke tools — no keyword matching or rule engine. This makes the assistant genuinely useful and demonstrates the complete tool-use protocol.
Class-based tools with async execute. One tool (weather) makes real HTTP calls to Open-Meteo. The async interface lets all tools share a common ABC without forcing synchronous tools into unnecessary complexity.
SSE over WebSocket. The interaction is request/response — user sends a message, server streams back. There's no server-initiated communication outside of a response, so WebSocket's bidirectional channel adds complexity (connection lifecycle, heartbeats, reconnection logic) without a clear benefit. SSE keeps the server simple (a standard HTTP handler returning a streaming response) and the client simple (a fetch call with a ReadableStream reader). We use POST + ReadableStream rather than EventSource specifically because EventSource only supports GET and can't send a JSON body with conversation history.
Stateless server, history in the browser. Each request sends the full conversation. Scales horizontally — any instance can serve any request.
Mock adapter over recorded fixtures for testing. Tests swap in a FakeLLM that yields scripted events via FastAPI dependency injection. This tests the code we wrote (streaming logic, tool dispatch, SSE formatting) without coupling to the provider's wire format — which the openai SDK already handles. Recorded fixtures would be more realistic but brittle to maintain and harder to simulate edge cases like tool errors or max-iteration breaches.
Tool loop with safety cap. The LLM can chain multiple tool calls per response. A MAX_TOOL_ITERATIONS guard (5) prevents infinite loops from malformed model output.
How would you keep responses fast when expensive operations (e.g. image or video generation) spike under load?
This can be handled on different layes
- Queue requests — don't send every request to the LLM at once. Buffer with priority so a traffic spike degrades gracefully instead of killing everyone's experience. We might want to implement backpressue as well to keep the queue size manageable
- Decouple expensive operations — for slow tools (image gen, video), return a tool_start event immediately and push the job to a background worker. Deliver results asynchronously rather than blocking the stream for 30+ seconds
- Model tiering — route simple queries to a cheaper/faster model, reserve expensive models for complex reasoning. A lightweight classifier at the edge decides.
How would you handle a long-running task so it doesn't block the user's chat?
The current design blocks the stream while a tool executes. For tools that take seconds (web scraping, DB queries), run them concurrently and stream a "thinking" indicator. The SSE protocol already supports this — emit tool_start, then tool_result when ready.
How would you store and scale conversation data for millions of users?
My assumption is that our use case has a high write to read ratio (people are chatting more than retrieveing old conversations), hence we need a write optimized data store. For millions of users: store conversations in a distributed store (DynamoDB, Redis) keyed by session ID.
How would you choose between a cheaper and a more capable model — and how would you know quality held up?
Route by query complexity: classify intent cheaply, then dispatch to the right model tier. Evaluate quality with automated evals (expected tool calls, response coherence), LLM-as-a-Judge and human feedback loops.