Master Generative AI — Part 5: Career & Capstone Projects¶

# app.py — RAG Chatbot
import streamlit as st
from pathlib import Path
import chromadb
from sentence_transformers import SentenceTransformer
from openai import OpenAI
from PyPDF2 import PdfReader
import hashlib

# ── Configuration ─────────────────────────────────────────────────
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
TOP_K = 4
MODEL = "gpt-4o-mini"

embed_model = SentenceTransformer("all-MiniLM-L6-v2")
chroma_client = chromadb.PersistentClient(path="./chroma_db")
collection = chroma_client.get_or_create_collection("documents")
openai_client = OpenAI()

# ── Document Processing ───────────────────────────────────────────
def extract_text(file) -> str:
    if file.name.endswith(".pdf"):
        reader = PdfReader(file)
        return "\n".join(page.extract_text() for page in reader.pages)
    return file.read().decode("utf-8")

def chunk_text(text: str) -> list[str]:
    words = text.split()
    chunks = []
    for i in range(0, len(words), CHUNK_SIZE - CHUNK_OVERLAP):
        chunk = " ".join(words[i:i + CHUNK_SIZE])
        if chunk:
            chunks.append(chunk)
    return chunks

def index_document(file) -> int:
    text = extract_text(file)
    chunks = chunk_text(text)
    embeddings = embed_model.encode(chunks).tolist()

    ids = [hashlib.md5(f"{file.name}_{i}".encode()).hexdigest() for i in range(len(chunks))]
    collection.upsert(
        ids=ids,
        embeddings=embeddings,
        documents=chunks,
        metadatas=[{"source": file.name, "chunk": i} for i in range(len(chunks))]
    )
    return len(chunks)

# ── Query ─────────────────────────────────────────────────────────
def answer_question(question: str, history: list) -> tuple[str, list[str]]:
    # Retrieve relevant chunks
    q_embed = embed_model.encode([question]).tolist()
    results = collection.query(query_embeddings=q_embed, n_results=TOP_K)
    chunks = results["documents"][0]
    sources = list({m["source"] for m in results["metadatas"][0]})
    context = "\n\n".join(f"[{i+1}] {c}" for i, c in enumerate(chunks))

    # Build conversation history
    messages = [{
        "role": "system",
        "content": f"""Answer questions using ONLY the provided context.
If the context doesn't contain the answer, say: "I don't have information about that in the provided documents."
Always be concise and cite which context number supports your answer.

Context:
{context}"""
    }]
    for h in history[-6:]:  # last 3 turns
        messages.extend([
            {"role": "user", "content": h["question"]},
            {"role": "assistant", "content": h["answer"]}
        ])
    messages.append({"role": "user", "content": question})

    response = openai_client.chat.completions.create(
        model=MODEL, messages=messages, temperature=0, max_tokens=600
    )
    return response.choices[0].message.content, sources

# ── Streamlit UI ──────────────────────────────────────────────────
st.set_page_config(page_title="AI Document Q&A", page_icon="🤖", layout="wide")
st.title("🤖 AI Document Q&A Chatbot")

with st.sidebar:
    st.header("📄 Upload Documents")
    uploaded_files = st.file_uploader("Upload PDFs or text files",
                                       accept_multiple_files=True,
                                       type=["pdf", "txt"])
    if uploaded_files and st.button("Index Documents"):
        for f in uploaded_files:
            count = index_document(f)
            st.success(f"✓ {f.name}: {count} chunks indexed")

    doc_count = collection.count()
    st.metric("Chunks in knowledge base", doc_count)

if "history" not in st.session_state:
    st.session_state.history = []

for msg in st.session_state.history:
    with st.chat_message("user"):
        st.write(msg["question"])
    with st.chat_message("assistant"):
        st.write(msg["answer"])
        if msg.get("sources"):
            st.caption(f"Sources: {', '.join(msg['sources'])}")

if question := st.chat_input("Ask about your documents..."):
    if collection.count() == 0:
        st.warning("Please upload and index documents first.")
    else:
        with st.chat_message("user"):
            st.write(question)
        with st.chat_message("assistant"):
            with st.spinner("Searching..."):
                answer, sources = answer_question(question, st.session_state.history)
            st.write(answer)
            if sources:
                st.caption(f"Sources: {', '.join(sources)}")
        st.session_state.history.append(
            {"question": question, "answer": answer, "sources": sources}
        )

# content_generator/main.py
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import OpenAI
import anthropic
import json

app = FastAPI(title="AI Content Generator")
openai_client = OpenAI()
claude_client = anthropic.Anthropic()

class ContentRequest(BaseModel):
    product_name: str
    product_description: str
    target_audience: str
    tone: str = "professional"
    formats: list[str] = ["blog", "linkedin", "email_subject"]

class ContentResponse(BaseModel):
    format: str
    content: str
    word_count: int
    model_used: str

CONTENT_CONFIGS = {
    "blog": {
        "system": "Expert blog writer. SEO-optimized. Use headers, examples, CTA.",
        "instruction": "Write a 600-word blog post.",
        "model": "openai",
        "openai_model": "gpt-4o"
    },
    "linkedin": {
        "system": "LinkedIn content strategist. Professional, engaging, no hashtag spam.",
        "instruction": "Write a LinkedIn post (200 words, 3-5 relevant hashtags, strong hook first line).",
        "model": "anthropic",
        "anthropic_model": "claude-haiku-4-5-20251001"
    },
    "email_subject": {
        "system": "Email marketing expert. High open rates. A/B variants.",
        "instruction": "Write 5 email subject lines. Each max 50 chars. Return as numbered list.",
        "model": "openai",
        "openai_model": "gpt-4o-mini"
    },
    "twitter": {
        "system": "Twitter/X copywriter. Punchy. Conversational.",
        "instruction": "Write 3 tweets (max 280 chars each). Return each on a new line.",
        "model": "openai",
        "openai_model": "gpt-4o-mini"
    },
    "product_description": {
        "system": "E-commerce copywriter. Benefit-first. Converts browsers to buyers.",
        "instruction": "Write a product description (150 words, bullet benefits, CTA).",
        "model": "anthropic",
        "anthropic_model": "claude-haiku-4-5-20251001"
    },
}

def generate_content(config: dict, context: str) -> str:
    prompt = f"{config['instruction']}\n\nProduct brief:\n{context}"
    if config["model"] == "openai":
        response = openai_client.chat.completions.create(
            model=config["openai_model"],
            messages=[
                {"role": "system", "content": config["system"]},
                {"role": "user", "content": prompt}
            ],
            temperature=0.8
        )
        return response.choices[0].message.content
    else:
        response = claude_client.messages.create(
            model=config["anthropic_model"],
            max_tokens=1000,
            system=config["system"],
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

@app.post("/generate", response_model=list[ContentResponse])
async def generate(request: ContentRequest):
    context = f"""Product: {request.product_name}
Description: {request.product_description}
Target audience: {request.target_audience}
Tone: {request.tone}"""

    results = []
    for fmt in request.formats:
        if fmt not in CONTENT_CONFIGS:
            raise HTTPException(400, f"Unknown format: {fmt}")
        config = CONTENT_CONFIGS[fmt]
        content = generate_content(config, context)
        results.append(ContentResponse(
            format=fmt,
            content=content,
            word_count=len(content.split()),
            model_used=config.get("openai_model") or config.get("anthropic_model", "unknown")
        ))
    return results

@app.get("/formats")
def list_formats():
    return list(CONTENT_CONFIGS.keys())

# art_generator/main.py
import streamlit as st
from diffusers import FluxPipeline, StableDiffusionPipeline
import torch
from PIL import Image
import json
from pathlib import Path
from datetime import datetime

# Style presets — prompt templates for consistent aesthetics
STYLE_PRESETS = {
    "Photorealistic": "photorealistic, DSLR quality, 8K, detailed, natural lighting",
    "Oil Painting": "oil painting, brushstrokes, canvas texture, museum quality, impressionist",
    "Anime": "anime style, Studio Ghibli, vibrant colors, clean lines, manga influence",
    "Watercolor": "watercolor painting, soft edges, paper texture, delicate, artistic",
    "Pixel Art": "pixel art, 8-bit style, retro game aesthetic, limited color palette",
    "Cinematic": "cinematic shot, movie still, dramatic lighting, shallow depth of field, film grain",
    "Minimalist": "minimalist design, clean, simple, white background, modern, elegant",
}

NEGATIVE_PROMPT = "blurry, low quality, distorted, ugly, watermark, text, signature, deformed"

@st.cache_resource
def load_model(model_choice: str):
    if model_choice == "FLUX Schnell (Fast)":
        pipe = FluxPipeline.from_pretrained(
            "black-forest-labs/FLUX.1-schnell",
            torch_dtype=torch.bfloat16
        ).to("cuda")
        return pipe, "flux"
    else:
        pipe = StableDiffusionPipeline.from_pretrained(
            "runwayml/stable-diffusion-v1-5",
            torch_dtype=torch.float16
        ).to("cuda")
        return pipe, "sd"

def enhance_prompt(base_prompt: str, style: str, enhance: bool) -> str:
    """Optionally enhance the user's prompt with AI."""
    if not enhance:
        return f"{base_prompt}, {STYLE_PRESETS[style]}"
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"Enhance this image prompt for Stable Diffusion. Add descriptive details, lighting, mood. Keep it under 75 words. Style: {style}.\n\nPrompt: {base_prompt}"
        }]
    )
    enhanced = response.choices[0].message.content.strip()
    return f"{enhanced}, {STYLE_PRESETS[style]}"

def generate_image(pipe, model_type: str, prompt: str, **kwargs) -> Image.Image:
    if model_type == "flux":
        return pipe(prompt=prompt, num_inference_steps=4, guidance_scale=0.0, **kwargs).images[0]
    else:
        return pipe(prompt=prompt, negative_prompt=NEGATIVE_PROMPT, **kwargs).images[0]

def save_to_gallery(image: Image.Image, prompt: str, style: str) -> str:
    Path("gallery").mkdir(exist_ok=True)
    filename = f"gallery/{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
    image.save(filename)
    meta_file = filename.replace(".png", ".json")
    json.dump({"prompt": prompt, "style": style, "created": datetime.now().isoformat()},
               open(meta_file, "w"), indent=2)
    return filename

# ── UI ────────────────────────────────────────────────────────────
st.set_page_config(page_title="AI Art Generator", page_icon="🎨", layout="wide")
st.title("🎨 AI Art Generator")

col1, col2 = st.columns([2, 3])

with col1:
    st.header("Settings")
    model_choice = st.selectbox("Model", ["FLUX Schnell (Fast)", "Stable Diffusion 1.5"])
    style = st.selectbox("Style Preset", list(STYLE_PRESETS.keys()))
    prompt = st.text_area("Describe your image", placeholder="A serene Thai temple at sunset...", height=100)
    enhance = st.checkbox("✨ AI Prompt Enhancement", value=True)

    with st.expander("Advanced"):
        steps = st.slider("Steps", 4, 50, 20)
        guidance = st.slider("Guidance Scale", 1.0, 15.0, 7.5)
        width = st.select_slider("Width", [512, 768, 1024], value=768)
        height = st.select_slider("Height", [512, 768, 1024], value=768)

    generate_btn = st.button("🎨 Generate", type="primary", use_container_width=True)

with col2:
    st.header("Output")
    if generate_btn and prompt:
        with st.spinner("Loading model..."):
            pipe, model_type = load_model(model_choice)

        final_prompt = enhance_prompt(prompt, style, enhance)
        if enhance:
            st.info(f"**Enhanced prompt:** {final_prompt}")

        with st.spinner(f"Generating ({steps} steps)..."):
            image = generate_image(pipe, model_type, final_prompt,
                                   num_inference_steps=steps,
                                   guidance_scale=guidance,
                                   width=width, height=height)

        st.image(image, use_column_width=True)
        filename = save_to_gallery(image, final_prompt, style)
        st.success(f"Saved to gallery: {filename}")
        st.download_button("⬇️ Download", open(filename, "rb"), file_name="ai_art.png")

# Gallery view
st.divider()
st.header("🖼️ Gallery")
gallery_files = sorted(Path("gallery").glob("*.png"), reverse=True)[:12] if Path("gallery").exists() else []
if gallery_files:
    cols = st.columns(4)
    for i, img_path in enumerate(gallery_files):
        with cols[i % 4]:
            st.image(str(img_path), use_column_width=True)
            meta_path = str(img_path).replace(".png", ".json")
            if Path(meta_path).exists():
                meta = json.load(open(meta_path))
                st.caption(meta["style"])

import numpy as np
from sentence_transformers import SentenceTransformer

def semantic_search(query: str, documents: list[str], top_k: int = 3) -> list[tuple[float, str]]:
    model = SentenceTransformer("all-MiniLM-L6-v2")

    # Embed everything
    doc_embeddings = model.encode(documents, normalize_embeddings=True)
    query_embedding = model.encode([query], normalize_embeddings=True)

    # Cosine similarity = dot product when vectors are normalized
    scores = (doc_embeddings @ query_embedding.T).flatten()

    # Return top-k (score, document) pairs
    top_indices = np.argsort(scores)[::-1][:top_k]
    return [(float(scores[i]), documents[i]) for i in top_indices]

# Test
docs = [
    "Python is a versatile programming language.",
    "Machine learning requires large amounts of data.",
    "The Transformer architecture uses self-attention.",
    "Deep learning models have millions of parameters.",
]
results = semantic_search("How do neural networks learn?", docs)
for score, doc in results:
    print(f"[{score:.3f}] {doc}")

def chunk_document(text: str, chunk_size: int = 500, overlap: int = 50) -> list[dict]:
    """
    Chunk text into overlapping windows.
    Returns list of dicts with text and metadata.
    """
    words = text.split()
    chunks = []

    start = 0
    chunk_id = 0
    while start < len(words):
        end = min(start + chunk_size, len(words))
        chunk_text = " ".join(words[start:end])
        chunks.append({
            "id": chunk_id,
            "text": chunk_text,
            "word_count": end - start,
            "start_word": start,
            "end_word": end,
        })
        chunk_id += 1
        start += chunk_size - overlap  # overlap for continuity

        if end == len(words):
            break

    return chunks

# Test
sample = "The quick brown fox " * 200  # 800 words
chunks = chunk_document(sample, chunk_size=100, overlap=20)
print(f"Total chunks: {len(chunks)}")
for c in chunks[:3]:
    print(f"Chunk {c['id']}: words {c['start_word']}–{c['end_word']}")

import tiktoken
from dataclasses import dataclass, field

@dataclass
class TokenBudgetManager:
    model: str = "gpt-4o"
    max_context: int = 128_000   # model's context window
    response_reserve: int = 2_000  # reserve for response
    system_prompt: str = ""

    def __post_init__(self):
        self.enc = tiktoken.encoding_for_model(self.model)
        self.system_tokens = len(self.enc.encode(self.system_prompt))

    def count_tokens(self, text: str) -> int:
        return len(self.enc.encode(text))

    def available_for_history(self) -> int:
        return self.max_context - self.system_tokens - self.response_reserve

    def trim_history(self, messages: list[dict]) -> list[dict]:
        """Trim oldest messages to fit within token budget."""
        budget = self.available_for_history()

        # Count from newest to oldest
        kept = []
        used = 0
        for msg in reversed(messages):
            tokens = self.count_tokens(msg["content"]) + 4  # role overhead
            if used + tokens <= budget:
                kept.insert(0, msg)
                used += tokens
            else:
                break  # stop — older messages don't fit

        trimmed = len(messages) - len(kept)
        if trimmed > 0:
            print(f"[TokenBudget] Trimmed {trimmed} old messages ({used}/{budget} tokens used)")

        return kept

# Usage
manager = TokenBudgetManager(
    model="gpt-4o",
    system_prompt="You are a helpful assistant specialized in Python."
)

history = [{"role": "user" if i % 2 == 0 else "assistant",
             "content": f"Message {i} " * 100}
           for i in range(20)]

trimmed = manager.trim_history(history)
print(f"Kept {len(trimmed)}/{len(history)} messages")