Step 1: A validated endpoint
Fernbank sells answers from a hosted model to other companies' apps through an HTTP API. app.py is that API,
built with FastAPI: POST /v1/answer takes {"question": ..., "max_tokens": ...} and returns
{"answer": ..., "usage": {...}}. Behind it, llmsim.py plays the model provider (it is fast, predictable and
can fail on demand, which the last step needs). The Run button starts both and calls your API like a customer.
Every request costs money upstream, so reject bad input before it gets there. FastAPI validates the request body against a pydantic model and answers 422 with the reasons when it does not fit.
Do this
1. Constrain AskIn: question from 1 to 2,000 characters, max_tokens from 1 to 400 with a default of
200, using pydantic's Field.
2. Run and read the three responses: a valid question, an empty one, and one asking for 5,000 tokens.
app.py, the file you edit121 lines
"""Ask Fernbank: the backend that sells answers from a hosted model to other companies' apps.
Run it: uvicorn app:app --port 8000 (the model server llmsim.py must be running on UPSTREAM_URL)."""
import hashlib
import hmac
import json
import logging
import math
import os
import time
import uuid
from collections import defaultdict, deque
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse, StreamingResponse
from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI
from pydantic import BaseModel, Field
UPSTREAM = OpenAI(base_url=os.environ.get("UPSTREAM_URL", "http://127.0.0.1:8090/v1"), api_key="sim",
timeout=4.0, max_retries=0)
PLANS = {"free": {"per_minute": 5, "tokens_per_day": 2000}, "pro": {"per_minute": 60, "tokens_per_day": 200000}}
KEYS = json.load(open("keys.json")) # sha256(api key) -> {"key_id", "tenant", "plan", "revoked"?}
SYSTEM = "You are Fernbank's answer engine. Answer the question in plain English."
log = logging.getLogger("fernbank")
clock = time.time # the checks replace this with a fake clock
app = FastAPI(title="Ask Fernbank")
@app.middleware("http")
async def request_id(request: Request, call_next):
"""Every response carries an X-Request-ID, and every request writes one JSON log line."""
rid = request.headers.get("X-Request-ID") or uuid.uuid4().hex[:16]
request.state.request_id = rid
t0 = time.time()
response = await call_next(request)
response.headers["X-Request-ID"] = rid
log.info(json.dumps({"request_id": rid, "path": request.url.path, "status": response.status_code,
"key_id": getattr(request.state, "key_id", None), "ms": round((time.time() - t0) * 1000)}))
return response
# ---------- Step 1: a validated endpoint ----------
class AskIn(BaseModel):
# TODO (Step 1): constrain the fields with pydantic's Field: question 1 to 2000 characters;
# max_tokens between 1 and 400, default 200. Invalid requests then get a 422 before any model call.
question: str
max_tokens: int = 200
class AskOut(BaseModel):
answer: str
usage: dict
# ---------- Step 2: who is calling ----------
def authenticate(request: Request, authorization: str | None = Header(default=None)):
"""The caller's key record, from "Authorization: Bearer <key>". 401 if missing, unknown or revoked."""
request.state.key_id = "dev"
return {"key_id": "dev", "tenant": "dev", "plan": "pro"} # open to everyone until Step 2
# ---------- Step 3: limits ----------
_recent = defaultdict(deque) # key_id -> timestamps of requests in the last 60 seconds
_used = defaultdict(int) # (key_id, day) -> tokens used
def day(t):
return time.strftime("%Y-%m-%d", time.gmtime(t))
def enforce_limits(record):
"""Before a request: 429 if the key has made its plan's per_minute requests in the last 60 seconds
(with a Retry-After header in whole seconds), or has used its tokens_per_day today (UTC)."""
pass # no limits until Step 3
def record_usage(record, usage):
"""After a request: add its total tokens to today's count for the key."""
pass # usage is counted from Step 3
# ---------- Step 5: when the model fails ----------
def call_upstream(**kwargs):
"""UPSTREAM.chat.completions.create(**kwargs), retried once on a connection error or a 5xx. A timeout becomes a
504 and any other failure a 502, with a short message that does not repeat the upstream's error text."""
return UPSTREAM.chat.completions.create(**kwargs) # Step 5 handles failures
@app.exception_handler(HTTPException)
async def http_error(request: Request, exc: HTTPException):
return JSONResponse({"error": exc.detail, "request_id": getattr(request.state, "request_id", None)},
status_code=exc.status_code, headers=exc.headers)
def messages_for(question):
return [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
@app.post("/v1/answer", response_model=AskOut)
def answer(body: AskIn, record=Depends(authenticate)):
enforce_limits(record)
r = call_upstream(model="sim", messages=messages_for(body.question), max_tokens=body.max_tokens)
usage = r.usage.model_dump()
record_usage(record, usage)
return {"answer": r.choices[0].message.content, "usage": usage}
# ---------- Step 4: stream it ----------
@app.post("/v1/answer/stream")
def answer_stream(body: AskIn, record=Depends(authenticate)):
enforce_limits(record)
stream = call_upstream(model="sim", messages=messages_for(body.question), max_tokens=body.max_tokens,
stream=True, stream_options={"include_usage": True})
return StreamingResponse(sse_events(stream, record), media_type="text/event-stream",
headers={"Cache-Control": "no-cache"})
def sse_events(stream, record):
"""Yield 'data: {"text": ...}\\n\\n' for each piece of text as it arrives; when the upstream sends usage,
record_usage() it and yield 'data: {"usage": {...}}\\n\\n'; finish with 'data: [DONE]\\n\\n'."""
raise NotImplementedError("sse_events() arrives in Step 4")Provided for you:
keys.jsonllmsim.pytry_it.py