Building Real AI Features

from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from openai import OpenAI from typing import Optional import os app = FastAPI() client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # CORS — required for frontend calls app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:3000"], # restrict in production allow_methods=["*"], allow_headers=["*"]) class SummarizeRequest(BaseModel): text: str max_words: Optional[int] = 150 format: Optional[str] = "paragraph" # or "bullets" class SummarizeResponse(BaseModel): summary: str tokens_used: int @app.post("/summarize", response_model=SummarizeResponse) async def summarize(req: SummarizeRequest): if not req.text.strip(): raise HTTPException(400, "Text cannot be empty") try: resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": f"Summarize in {req.max_words} words or fewer. Format: {req.format}."}, {"role": "user", "content": req.text} ], max_tokens=500, temperature=0.3 ) return SummarizeResponse( summary=resp.choices[0].message.content, tokens_used=resp.usage.total_tokens ) except Exception as e: raise HTTPException(500, str(e)) # Start: uvicorn main:app --reload # Docs: http://localhost:8000/docs

# Client-side: maintain history array messages = [ {"role": "system", "content": "You are a helpful assistant."}, ] def chat(user_input: str) -> str: messages.append({"role": "user", "content": user_input}) resp = client.chat.completions.create( model="gpt-4o-mini", messages=messages, max_tokens=500 ) reply = resp.choices[0].message.content messages.append({"role": "assistant", "content": reply}) return reply

import tiktoken def sliding_window(messages, max_tokens=3000, model="gpt-4o-mini"): """Keep system prompt + most recent messages within token budget.""" enc = tiktoken.encoding_for_model(model) system = [m for m in messages if m["role"] == "system"] convo = [m for m in messages if m["role"] != "system"] def count(msgs): return sum(len(enc.encode(m["content"])) + 4 for m in msgs) + 3 while count(system + convo) > max_tokens and len(convo) > 2: convo.pop(0) # drop oldest user/assistant pair return system + convo

import tiktoken def chunk_text(text: str, chunk_tokens: int = 2000, model: str = "gpt-4o-mini"): enc = tiktoken.encoding_for_model(model) tokens = enc.encode(text) chunks = [] for i in range(0, len(tokens), chunk_tokens): chunks.append(enc.decode(tokens[i:i + chunk_tokens])) return chunks def summarize_chunk(chunk: str) -> str: resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "Summarize this section concisely."}, {"role": "user", "content": chunk} ], max_tokens=300 ) return resp.choices[0].message.content def map_reduce_summarize(long_text: str) -> str: chunks = chunk_text(long_text) # MAP — summarize each chunk summaries = [summarize_chunk(c) for c in chunks] # REDUCE — combine summaries resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "Synthesize these summaries into one coherent summary."}, {"role": "user", "content": "\n\n---\n\n".join(summaries)} ], max_tokens=500 ) return resp.choices[0].message.content # Rule of thumb: text < 3000 tokens → single pass. Larger → map-reduce.

import json def classify_sentiment(text: str) -> dict: resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ { "role": "system", "content": """Classify the sentiment of the text. Respond with ONLY valid JSON in this exact format: {"label": "positive"|"negative"|"neutral", "confidence": 0.0-1.0, "reasoning": "one sentence explanation"}""" }, {"role": "user", "content": text} ], response_format={"type": "json_object"}, # ← forces valid JSON temperature=0.0, # ← deterministic for classification max_tokens=150 ) result = json.loads(resp.choices[0].message.content) return { "label": result["label"], # "positive" "confidence": result["confidence"], # 0.92 "reasoning": result["reasoning"], # "..." }

from unittest.mock import MagicMock, patch from fastapi.testclient import TestClient from main import app test_client = TestClient(app) def mock_response(content: str, tokens: int = 100): m = MagicMock() m.choices[0].message.content = content m.choices[0].finish_reason = "stop" m.usage.total_tokens = tokens m.usage.prompt_tokens = tokens - 30 m.usage.completion_tokens = 30 return m # Test normal operation @patch("main.client.chat.completions.create") def test_summarize_success(mock_create): mock_create.return_value = mock_response("This is a summary.") resp = test_client.post("/summarize", json={"text": "Long text here..."}) assert resp.status_code == 200 assert resp.json()["summary"] == "This is a summary." # Test input validation def test_summarize_empty_text(): resp = test_client.post("/summarize", json={"text": ""}) assert resp.status_code == 400 # Test cost guard — make sure you're not accidentally using gpt-4o @patch("main.client.chat.completions.create") def test_uses_mini_model(mock_create): mock_create.return_value = mock_response("summary") test_client.post("/summarize", json={"text": "test"}) call_kwargs = mock_create.call_args[1] assert call_kwargs["model"] == "gpt-4o-mini" # ← cost guard