Building and Shipping a GenAI App: Architecture, Streamlit/FastAPI, Cost, Observability and Responsible AI (Capstone)

# app.py  -> streamlit run app.py
import streamlit as st, requests
st.title("NovaMart Help")
if "history" not in st.session_state: st.session_state.history = []
for m in st.session_state.history: st.chat_message(m["role"]).write(m["content"])
if q := st.chat_input("Ask about orders, returns, delivery…"):
    st.chat_message("user").write(q)
    r = requests.post("http://localhost:8000/chat", json={"session_id": st.session_state.get("sid", "demo"), "message": q}, timeout=60).json()
    st.chat_message("assistant").write(r["answer"] + ("\n\n" + " · ".join(r["sources"]) if r["sources"] else ""))
    st.session_state.history += [{"role": "user", "content": q}, {"role": "assistant", "content": r["answer"]}]
    c1, c2 = st.columns(8)[:2]
    c1.button("👍", key=f"up{len(st.session_state.history)}"); c2.button("👎", key=f"down{len(st.session_state.history)}")

# api.py -> uvicorn api:app
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import anthropic, time, logging
app, client = FastAPI(), anthropic.Anthropic()
SESSIONS: dict[str, list] = {}                         # use Redis or a database in production
class ChatIn(BaseModel):
    session_id: str = Field(max_length=64)
    message: str = Field(min_length=1, max_length=2000)
@app.post("/chat")
def chat(body: ChatIn):
    if injection_score(body.message)[0] >= 2: raise HTTPException(400, "Please rephrase your question.")
    prompt, ctx = build_prompt(body.message)            # RAG from lesson 5
    history = SESSIONS.setdefault(body.session_id, [])[-6:]          # keep the last 3 exchanges
    t0 = time.perf_counter()
    resp = client.messages.create(model="claude-sonnet-5-5", max_tokens=2000, system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],
                                  messages=history + [{"role": "user", "content": prompt}])
    answer = next(b.text for b in resp.content if b.type == "text")
    issues = check_output(answer)                        # lesson 9 guardrails
    if issues != ["ok"]: answer = "Let me connect you with a support agent."
    SESSIONS[body.session_id] = history + [{"role": "user", "content": body.message}, {"role": "assistant", "content": answer}]
    logging.info({"latency_ms": int(1000 * (time.perf_counter() - t0)), "in": resp.usage.input_tokens, "out": resp.usage.output_tokens, "issues": issues})
    return {"answer": answer, "sources": ctx.id.tolist()}

import pandas as pd, numpy as np
PRICE = {"claude-haiku-5-5": (0.10, 0.50), "claude-sonnet-5-5": (2.00, 10.00), "claude-opus-5-5": (4.00, 20.00)}   # USD per million tokens, Oct 2026 list prices
def monthly(conversations, turns, system_tok, ctx_tok, user_tok, out_tok, model, cache_hit=0.0, usd_inr=88):
    pin, pout = PRICE[model]; calls = conversations * turns
    cached = system_tok * cache_hit; uncached = system_tok - cached + ctx_tok + user_tok
    usd = calls * ((uncached * pin + cached * pin * 0.1) + out_tok * pout) / 1e6
    return round(usd * usd_inr)
base = dict(conversations=60_000, turns=4, system_tok=1500, ctx_tok=1200, user_tok=60, out_tok=180)
rows = []
for m in PRICE:
    for hit in (0.0, 0.9):
        rows.append({"model": m, "system prompt cache hit": f"{hit:.0%}", "₹ / month": monthly(model=m, cache_hit=hit, **base)})
plan = pd.DataFrame(rows); print(plan.to_string(index=False))
route = 0.8 * monthly(model="claude-haiku-5-5", cache_hit=.9, **base) + 0.2 * monthly(model="claude-sonnet-5-5", cache_hit=.9, **base)
print(f"\nrouting 80% of turns to Haiku and 20% to Sonnet: ≈ ₹{route:,.0f}/month for {base['conversations']:,} conversations ({route / base['conversations']:.2f} ₹ per conversation)")