The Ultimate AI & Machine Learning Cheatsheet

You are an expert Python developer with 10 years of experience in data engineering.

[CONTEXT] I am building a data pipeline that processes CSV files from an S3 bucket,
transforms the data, and loads it into a PostgreSQL database.

[REQUEST] Write a Python script using pandas and psycopg2 that:
1. Reads CSV files from a local directory
2. Validates the schema
3. Inserts valid rows into a PostgreSQL table

[AUDIENCE] This will be used by a junior developer on my team. Include error handling
and inline comments explaining each step.

[TONE] Professional but accessible. Include type hints.

[FORMAT] Return only the Python code with comments. No markdown code fences.

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import pandas as pd

# Load data
df = pd.read_csv("customers.csv")
X = df.drop("churn", axis=1)
y = df["churn"]

# Split data
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Train model
model = RandomForestClassifier(
    n_estimators=200, max_depth=10, min_samples_split=5, random_state=42
)
model.fit(X_train, y_train)

# Evaluate
y_pred = model.predict(X_test)
print(classification_report(y_test, y_pred))

# Feature importance
for feat, imp in sorted(
    zip(X.columns, model.feature_importances_), key=lambda x: -x[1]
):
    print(f"  {feat}: {imp:.4f}")

from sklearn.linear_model import Lasso, Ridge, ElasticNet
from sklearn.model_selection import cross_val_score

# L1 Regularization (feature selection)
lasso = Lasso(alpha=0.01)
print("Lasso R2:", cross_val_score(lasso, X, y, cv=5).mean())

# L2 Regularization (shrink weights)
ridge = Ridge(alpha=1.0)
print("Ridge R2:", cross_val_score(ridge, X, y, cv=5).mean())

# Elastic Net (L1 + L2)
elastic = ElasticNet(alpha=0.01, l1_ratio=0.5)
print("ElasticNet R2:", cross_val_score(elastic, X, y, cv=5).mean())

import torch
import torch.nn as nn

class SimpleClassifier(nn.Module):
    def __init__(self, input_dim, hidden_dim, num_classes):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.BatchNorm1d(hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, num_classes),
        )

    def forward(self, x):
        return self.net(x)

# Usage
model = SimpleClassifier(input_dim=784, hidden_dim=256, num_classes=10)
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-4)

import tiktoken

# Tokenize text
enc = tiktoken.encoding_for_model("gpt-4o")
text = "Machine learning is transforming the world!"
tokens = enc.encode(text)
print(f"Tokens: {tokens}")
print(f"Count: {len(tokens)} tokens")
print(f"Decoded: {enc.decode(tokens)}")

# Count tokens for cost estimation
def estimate_cost(text: str, model: str = "gpt-4o") -> float:
    enc = tiktoken.encoding_for_model(model)
    n_tokens = len(enc.encode(text))
    # GPT-4o pricing: $2.50/1M input, $10/1M output
    input_cost = (n_tokens / 1_000_000) * 2.50
    return round(input_cost, 6)

print(f"Estimated input cost: ${estimate_cost(text)}")

# Example: Basic content safety check before processing
import re

def check_input_safety(prompt: str) -> tuple[bool, str]:
    """Basic input validation for AI applications."""
    blocked_patterns = [
        r"(ignore|bypass|override)s+(previous|safety|system)",
        r"(pretend|act|roleplay)s+(as|like)s+(admin|god|no one)",
        r"(jailbreak|DAN|hacked)",
    ]
    for pattern in blocked_patterns:
        if re.search(pattern, prompt, re.IGNORECASE):
            return False, "Input flagged by safety filter."

    # Length check
    if len(prompt) > 50000:
        return False, "Input exceeds maximum allowed length."

    return True, "Input passed safety checks."

# RAG answer grounding - require citations
def ground_answer(answer: str, sources: list[str]) -> str:
    """Ensure answer is grounded in provided sources."""
    return (
        f"{answer}

"
        f"Sources used: {', '.join(sources[:3])}
"
        f"Note: This answer is based on the provided documents "
        f"and may not reflect the most current information."
    )

from openai import OpenAI
import json

client = OpenAI()  # uses OPENAI_API_KEY env var

# ── 1. Chat Completion ──────────────────────────────
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain async/await in Python."},
    ],
    temperature=0.7,
    max_tokens=500,
)
print(response.choices[0].message.content)

# ── 2. Streaming Response ───────────────────────────
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a haiku about code."}],
    stream=True,
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

# ── 3. Function Calling (Tool Use) ─────────────────
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                    },
                },
                "required": ["city"],
            },
        },
    }
]

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Weather in Tokyo?"}],
    tools=tools,
    tool_choice="auto",
)

# Check if model wants to call a function
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    args = json.loads(tool_call.function.arguments)
    print(f"Call: {tool_call.function.name}({args})")

# ── 4. Embeddings for RAG ──────────────────────────
embedding = client.embeddings.create(
    model="text-embedding-3-small",
    input="Machine learning is a subset of AI.",
)
print(f"Dimensions: {len(embedding.data[0].embedding)}")
print(f"First 5: {embedding.data[0].embedding[:5]}")

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyPDFLoader

# 1. Load and split documents
loader = PyPDFLoader("company-handbook.pdf")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["

", "
", ". ", " ", ""],
)
chunks = splitter.split_documents(docs)

# 2. Create vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
)

# 3. Create RAG chain
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

template = """Answer the question based on these context documents.
If unsure, say "I don't have enough information."

Context:
{context}

Question: {question}

Provide a clear, concise answer with source references."""

prompt = ChatPromptTemplate.from_template(template)

def format_docs(docs):
    return "

".join(
        f"[Doc {i+1}] {d.page_content}" for i, d in enumerate(docs)
    )

chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
)

# 4. Query
answer = chain.invoke("What is the vacation policy?")
print(answer.content)

import hashlib
import json

# ── 1. Simple Response Cache ───────────────────────
_cache: dict[str, str] = {}

def cached_completion(prompt: str, model: str = "gpt-4o-mini") -> str:
    """Cache responses to avoid duplicate API calls."""
    key = hashlib.md5(f"{model}:{prompt}".encode()).hexdigest()
    if key in _cache:
        print("Cache hit!")
        return _cache[key]

    # Actual API call
    from openai import OpenAI
    client = OpenAI()
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    result = response.choices[0].message.content
    _cache[key] = result
    return result

# ── 2. Model Routing by Complexity ─────────────────
def route_model(prompt: str, system_prompt: str) -> str:
    """Use cheap model for simple tasks, expensive for complex."""
    # Simple heuristic: use mini for short prompts
    is_simple = len(prompt.split()) < 50

    from openai import OpenAI
    client = OpenAI()
    model = "gpt-4o-mini" if is_simple else "gpt-4o"

    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt},
        ],
        temperature=0,
    )
    return response.choices[0].message.content

# ── 3. Structured Output (JSON Mode) ──────────────
response_schema = {
    "type": "json_schema",
    "json_schema": {
        "name": "product_review",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "sentiment": {
                    "type": "string",
                    "enum": ["positive", "negative", "neutral"],
                },
                "rating": {"type": "number"},
                "summary": {"type": "string"},
            },
            "required": ["sentiment", "rating", "summary"],
            "additionalProperties": False,
        },
    },
}

# Use with response_format parameter in API call
# response = client.chat.completions.create(
#     model="gpt-4o-mini",
#     messages=[...],
#     response_format=response_schema,
# )