← work

conversational chatbot

an async conversational backend where redis can disappear and users never notice.

repo ↗

what i built

  • auth — signup and login with bcrypt hashing and 30-minute jwt access tokens.
  • per-user conversations — users, conversations and messages persisted through async sqlalchemy. every read checks that the conversation belongs to the caller.
  • layered structure — routes call repositories, repositories own the queries, and the gemini and redis clients sit behind services.
  • cache that can fail safely — normalized prompts cached for an hour. if redis is down, requests skip the cache and still succeed.
  • shipped with ci — multi-stage dockerfile, plus a workflow that starts redis 7, runs pytest and builds the image on every push.

how it fits together

clientfastapibot.pyredis cachenormalized prompts · 1 hgemini servicerepositoriessqliteusers · conversations · messageslookupmisshit · skips gemini
figure 1a cache hit returns without touching gemini. either way the reply is saved to the database.(scroll the figure sideways)

don't block the event loop

bcrypt is deliberately slow and cpu-bound. called directly inside an async def, it freezes every other request while it runs. so hashing and verification go to the thread pool, and the loop keeps serving others.

hashedpassword = await asyncio.to_thread(secure_pwd, request.password) is_valid = await asyncio.to_thread(verify_pwd, request.password, user.hashedpassword)
bot.py

the same applies to the synchronous redis and gemini clients in /chat.

one client, built once

the gemini client is expensive to construct. it's created inside fastapi's lifespan hook, stored on app.state, and injected with a dependency. it's built once at startup, and a bad key fails the boot instead of the first user's request.

@asynccontextmanager async def lifespan(app: FastAPI): await init_db() app.state.gemini = Geminiservice() yield
bot.py

redis is an optimisation, not a dependency

both the lookup and the write are wrapped separately. a lookup failure sets cached = None and falls through to gemini. a write failure is logged and ignored. redis being down makes responses slower, not wrong.

try: normalized_key = normalize_key(request.message) cached = await asyncio.to_thread(get_cached_response, normalized_key) except Exception: logger.warning("Redis unavailable") cached = None normalized_key = None
bot.py

metrics

metricmeasuredhow
throughput42.5 req/slocust load against /chat
median latency110 mssame run
success rate99.86%same run
from the readme's load test. the median is mostly cache hits, so it says little about a cold gemini call — on a miss, gemini latency is what limits throughput.
parametervalue
access token lifetime30 minutes
cache ttl1 hour, keyed on the normalized prompt
password hashingbcrypt, off the event loop
ciredis 7, pytest and an image build on every push

the event loop survived. character development.