Mini Project: Build an AI Chat Assistant

ai_assistant/
│
├── .env                  # API Key configurations
├── .gitignore            # Exclude environment variables
├── requirements.txt      # Project dependencies
└── main.py               # Complete FastAPI backend codebase

fastapi==0.111.0
uvicorn==0.30.1
pydantic==2.7.4
httpx==0.27.0
google-generativeai==0.7.2
python-dotenv==1.0.1

pip install -r requirements.txt

GEMINI_API_KEY=AIzaSyYourActualAPIKeyHere
PORT=8000

.env
__pycache__/
venv/
*.pyc

import os
import time
import logging
from typing import Dict, List, Optional
from fastapi import FastAPI, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from dotenv import load_dotenv
import google.generativeai as genai
from google.generativeai.types import APIError

# Configure Logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# Load configuration values
load_dotenv()
API_KEY = os.getenv("GEMINI_API_KEY")

if not API_KEY or API_KEY.startswith("AIzaSyYour"):
    logging.warning("GEMINI_API_KEY is not set or is using the default placeholder value. API calls will fail.")

# Configure Gemini Client
genai.configure(api_key=API_KEY)

# Instantiate FastAPI application
app = FastAPI(
    title="Gemini AI Chat Assistant API",
    description="A production-grade chat assistant with session history and input validation.",
    version="1.0.0"
)

# Enable CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# --- Memory Storage and Schema Models ---

class ChatMessage(BaseModel):
    role: str = Field(..., description="'user' or 'model'")
    content: str = Field(..., description="Text content of the message")
    timestamp: float = Field(default_factory=time.time)

# Simulates database caching
session_history_store: Dict[str, List[genai.types.ContentDict]] = {}

class ChatRequest(BaseModel):
    session_id: str = Field(..., min_length=3, max_length=50, description="Unique session ID to track history")
    prompt: str = Field(..., min_length=1, max_length=4096, description="User instruction message")
    temperature: float = Field(0.7, ge=0.0, le=1.0, description="Model sampling temperature")

class ChatResponse(BaseModel):
    status: str
    session_id: str
    response_text: str
    history_length: int

# --- API Endpoints ---

@app.get("/health", tags=["System"])
def health_check():
    """Returns application status status verification."""
    return {"status": "healthy", "timestamp": time.time(), "api_configured": bool(API_KEY)}

@app.post("/api/v1/chat", response_model=ChatResponse, tags=["AI Core"])
async def send_chat_message(request: ChatRequest):
    """
    Sends a query prompt to the Gemini model, incorporating session history 
    to maintain context across multiple turns.
    """
    session_id = request.session_id
    prompt = request.prompt
    
    # 1. Fetch or initialize chat history for the session
    if session_id not in session_history_store:
        session_history_store[session_id] = []
        
    history = session_history_store[session_id]
    
    # 2. Configure model settings and parameters
    generation_config = {
        "temperature": request.temperature,
        "max_output_tokens": 2048,
    }
    
    try:
        # 3. Instantiate model target
        model = genai.GenerativeModel(
            model_name="gemini-1.5-flash",
            generation_config=generation_config,
            system_instruction="You are a professional, polite, and technical AI assistant. Keep responses structured and concise."
        )
        
        # 4. Initialize session-backed chat object
        chat = model.start_chat(history=history)
        
        # 5. Send message and await response
        logging.info(f"Sending prompt for session '{session_id}' to Gemini API.")
        response = await chat.send_message_async(prompt)
        
        # 6. Update local history cache with the updated thread history
        session_history_store[session_id] = chat.history
        
        return ChatResponse(
            status="success",
            session_id=session_id,
            response_text=response.text,
            history_length=len(chat.history)
        )
        
    except APIError as e:
        logging.error(f"Gemini API Error occurred: {e}")
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"Upstream AI Provider Error: {e.message}"
        )
    except Exception as e:
        logging.error(f"Unexpected error: {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="An unexpected system error occurred while processing the request."
        )

@app.delete("/api/v1/chat/{session_id}", tags=["AI Core"])
def clear_chat_session(session_id: str):
    """Clears history cache for the specified session ID."""
    if session_id in session_history_store:
        del session_history_store[session_id]
        return {"status": "success", "message": f"Session '{session_id}' cleared successfully."}
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail="Session not found in active records."
    )

if __name__ == "__main__":
    import uvicorn
    # Starts the local development server
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

python main.py

curl -X POST "http://127.0.0.1:8000/api/v1/chat" \
     -H "Content-Type: application/json" \
     -d '{
       "session_id": "session_test_abc",
       "prompt": "Hello! What is Python in one sentence?",
       "temperature": 0.5
     }'

{
  "status": "success",
  "session_id": "session_test_abc",
  "response_text": "Python is a high-level, interpreted programming language known for its readability and versatile applications.",
  "history_length": 2
}

curl -X POST "http://127.0.0.1:8000/api/v1/chat" \
     -H "Content-Type: application/json" \
     -d '{
       "session_id": "session_test_abc",
       "prompt": "What did I just ask you?",
       "temperature": 0.5
     }'