Deep learning, from the neuron up to LoRA, GRPO & agents.

neuron.pypython
import numpy as np

# one neuron, 3 inputs
x = np.array([1.0, 1.0, 0.0])          # weather, free time, tired
w = np.array([2.0, 3.0, -4.0])         # learned weights
b = -1.0                               # learned bias

def relu(z):                           # a common activation (see §5)
    return max(0.0, z)

z = np.dot(w, x) + b                   # weighted sum + bias  -> 4.0
y = relu(z)                            # activation           -> 4.0
print(f"pre-activation z = {z},  output y = {y}")
# A whole layer is just many neurons stacked: a matrix multiply.
W = np.array([[2., 3., -4.],           # neuron 1
              [1., -1., 1.]])          # neuron 2
b_vec = np.array([-1.0, 0.5])
layer_out = np.maximum(0.0, W @ x + b_vec)   # ReLU over the vector
print("layer output:", layer_out)

tensors.pypython
import torch

# create tensors
x = torch.randn(32, 784)        # batch of 32 flattened 28x28 images
W = torch.randn(784, 128)       # weight matrix: 784 in -> 128 out
b = torch.randn(128)            # bias vector

# a linear layer, by hand
y = x @ W + b                   # @ is matmul; b is BROADCAST over 32 rows
print(x.shape, "@", W.shape, "->", y.shape)   # (32,784) @ (784,128) -> (32,128)

# reshaping / moving axes (constant in real code)
imgs = torch.randn(32, 3, 224, 224)           # (batch, channels, H, W)
flat = imgs.reshape(32, -1)                   # -> (32, 150528)
moved = imgs.permute(0, 2, 3, 1)              # NCHW -> NHWC: (32,224,224,3)

# everything runs on a GPU by moving the tensor there
device = "cuda" if torch.cuda.is_available() else "cpu"
x = x.to(device)                              # same API, faster hardware

mlp_pytorch.pyPyTorch
import torch.nn as nn

model = nn.Sequential(
    nn.Linear(3, 16),   # input 3  -> hidden 16
    nn.ReLU(),
    nn.Linear(16, 16),  # hidden   -> hidden 16
    nn.ReLU(),
    nn.Linear(16, 1),   # hidden   -> output 1 (price)
)
# forward pass: prediction = model(x)

mlp_keras.pyTensorFlow / Keras
import tensorflow as tf
from tensorflow import keras

model = keras.Sequential([
    keras.layers.Dense(16, activation="relu", input_shape=(3,)),
    keras.layers.Dense(16, activation="relu"),
    keras.layers.Dense(1),                 # linear output for regression
])
# forward pass: prediction = model(x)

activations.pypython
import torch, torch.nn.functional as F

z = torch.tensor([2.0, 1.0, 0.1])

F.relu(z)        # tensor([2.0, 1.0, 0.1])      negatives -> 0
torch.sigmoid(z) # tensor([0.88, 0.73, 0.52])   each squashed to (0,1)
torch.tanh(z)    # tensor([0.96, 0.76, 0.10])   squashed to (-1,1)
F.softmax(z, 0)  # tensor([0.66, 0.24, 0.10])   a probability distribution
F.gelu(z)        # smooth ReLU-like curve used in BERT/GPT

losses.pypython
import torch, torch.nn as nn

# regression
pred = torch.tensor([3.2, 5.0]); target = torch.tensor([3.0, 4.5])
mse = nn.MSELoss()(pred, target)            # -> 0.145

# multi-class classification (3 classes, batch of 2)
logits = torch.tensor([[2.0, 0.5, 0.1],     # raw scores, NOT softmaxed
                       [0.2, 0.1, 3.0]])
labels = torch.tensor([0, 2])               # true class indices
ce = nn.CrossEntropyLoss()(logits, labels)  # applies softmax internally
print(mse.item(), ce.item())

autograd.pyPyTorch
import torch

# w is a parameter we want gradients for
w = torch.tensor([2.0], requires_grad=True)
x = torch.tensor([3.0])

y    = w * x          # forward: 6.0
loss = (y - 10) ** 2  # target is 10 -> loss = (6-10)^2 = 16

loss.backward()       # BACKPROP: fills w.grad with d(loss)/dw
print(w.grad)         # tensor([-24.])  -> negative: increasing w lowers loss

# you almost never call backward yourself on toy expressions;
# in real training it's one line inside the loop (see section 10).

optimizers.pyPyTorch
import torch

# pass the model's parameters and a learning rate
opt = torch.optim.Adam(model.parameters(), lr=1e-3)

# one optimization step (inside the training loop):
opt.zero_grad()     # clear old gradients (they accumulate otherwise!)
loss.backward()     # backprop: compute new gradients
opt.step()          # apply the update rule to every weight

# Keras equivalent:
# model.compile(optimizer=keras.optimizers.Adam(1e-3), loss="mse")

regularization.pyPyTorch
import torch.nn as nn

model = nn.Sequential(
    nn.Linear(784, 256),
    nn.BatchNorm1d(256),    # batch normalization
    nn.ReLU(),
    nn.Dropout(p=0.3),      # zero 30% of activations during training
    nn.Linear(256, 10),
)

# weight decay (L2) is set on the optimizer:
opt = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-2)

# IMPORTANT: dropout/batchnorm behave differently in train vs eval!
model.train()   # enables dropout + batchnorm updates
# ... training ...
model.eval()    # disables dropout, freezes batchnorm stats for inference

train.pyPyTorch
import torch, torch.nn as nn
from torch.utils.data import DataLoader

device = "cuda" if torch.cuda.is_available() else "cpu"
model  = MyModel().to(device)
loss_fn = nn.CrossEntropyLoss()
opt    = torch.optim.AdamW(model.parameters(), lr=1e-3)

for epoch in range(EPOCHS):
    # ---- train ----
    model.train()
    for xb, yb in train_loader:                 # mini-batches
        xb, yb = xb.to(device), yb.to(device)
        preds  = model(xb)                       # 1. forward
        loss   = loss_fn(preds, yb)              # 2. loss
        opt.zero_grad()                          #    clear old grads
        loss.backward()                          # 3. backward
        opt.step()                               # 4. update weights

    # ---- validate ----
    model.eval()
    correct = 0
    with torch.no_grad():                        # no gradients needed
        for xb, yb in val_loader:
            xb, yb = xb.to(device), yb.to(device)
            correct += (model(xb).argmax(1) == yb).sum().item()
    print(f"epoch {epoch}: val acc = {correct/len(val_loader.dataset):.3f}")

train_keras.pyTensorFlow / Keras
# Keras wraps the loop in .fit() — convenient, less explicit
model.compile(optimizer="adamw", loss="sparse_categorical_crossentropy",
              metrics=["accuracy"])
model.fit(train_ds, validation_data=val_ds, epochs=EPOCHS)

cnn.pyPyTorch
import torch.nn as nn

cnn = nn.Sequential(
    nn.Conv2d(3, 32, kernel_size=3, padding=1),  # 3 RGB in -> 32 feature maps
    nn.ReLU(),
    nn.MaxPool2d(2),                             # halve spatial size
    nn.Conv2d(32, 64, kernel_size=3, padding=1), # 32 -> 64 feature maps
    nn.ReLU(),
    nn.MaxPool2d(2),
    nn.Flatten(),
    nn.Linear(64 * 8 * 8, 128), nn.ReLU(),       # dense head
    nn.Linear(128, 10),                          # 10 classes
)
# For real projects you rarely build from scratch — you fine-tune a
# pretrained ResNet/EfficientNet (see Section 15 on transfer learning).

lstm.pyPyTorch
import torch, torch.nn as nn

# classify the sentiment of a sequence of word-embeddings
class SentimentLSTM(nn.Module):
    def __init__(self, vocab, dim=128, hidden=256):
        super().__init__()
        self.embed = nn.Embedding(vocab, dim)        # token id -> vector
        self.lstm  = nn.LSTM(dim, hidden, batch_first=True)
        self.head  = nn.Linear(hidden, 2)            # pos / neg
    def forward(self, x):                            # x: (batch, seq_len)
        e = self.embed(x)                            # (batch, seq, dim)
        out, (h_n, c_n) = self.lstm(e)               # h_n = final memory
        return self.head(h_n[-1])                    # classify from it

attention.pyPyTorch
import torch, torch.nn.functional as F

def self_attention(x, Wq, Wk, Wv):
    # x: (seq_len, dim)  — one sequence of token vectors
    Q, K, V = x @ Wq, x @ Wk, x @ Wv          # learned projections
    d_k = Q.size(-1)
    scores = (Q @ K.transpose(-2, -1)) / d_k ** 0.5   # (seq, seq)
    weights = F.softmax(scores, dim=-1)        # how much each token attends
    return weights @ V                         # weighted blend of values

# In practice you use the built-in, optimized version:
# torch.nn.MultiheadAttention, or just load a pretrained transformer (§16).

embeddings.py🤗 sentence-transformers
from sentence_transformers import SentenceTransformer, util

model = SentenceTransformer("all-MiniLM-L6-v2")   # small, fast embedder
docs = ["How do I reset my password?",
        "Ways to recover account access",
        "What's the weather in Paris?"]
emb = model.encode(docs, convert_to_tensor=True)  # (3, 384) vectors

query = model.encode("I forgot my login", convert_to_tensor=True)
scores = util.cos_sim(query, emb)                 # cosine similarity
print(scores)  # highest for the first two docs, low for the weather one

finetune_vision.pyPyTorch / torchvision
import torch, torch.nn as nn
from torchvision import models

# 1. load a model pretrained on ImageNet
net = models.resnet50(weights="IMAGENET1K_V2")

# 2. freeze the body (feature extraction)
for p in net.parameters():
    p.requires_grad = False

# 3. replace the head for OUR 10 bird classes (this part trains)
net.fc = nn.Linear(net.fc.in_features, 10)

# 4. train only the new head with the standard loop from Section 10
opt = torch.optim.Adam(net.fc.parameters(), lr=1e-3)
# To FINE-TUNE instead: unfreeze later layers and use a tiny LR (e.g. 1e-5).

pipeline.py🤗 Transformers
from transformers import pipeline

clf = pipeline("sentiment-analysis")          # downloads a model for you
print(clf("I absolutely loved this guide!"))
# [{'label': 'POSITIVE', 'score': 0.9998}]

# the same API covers dozens of tasks:
pipeline("summarization")
pipeline("question-answering")
pipeline("text-generation", model="gpt2")
pipeline("zero-shot-classification")          # classify into labels you invent
pipeline("automatic-speech-recognition")      # audio -> text

model_direct.py🤗 Transformers
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

name = "distilbert-base-uncased-finetuned-sst-2-english"
tok   = AutoTokenizer.from_pretrained(name)
model = AutoModelForSequenceClassification.from_pretrained(name)

inputs = tok("Transformers make this easy.", return_tensors="pt")
with torch.no_grad():
    logits = model(**inputs).logits          # raw scores
probs = logits.softmax(-1)                    # -> probabilities
print(model.config.id2label[probs.argmax().item()])   # 'POSITIVE'

finetune_text.py🤗 Transformers
from transformers import (AutoTokenizer, AutoModelForSequenceClassification,
                          TrainingArguments, Trainer)
from datasets import load_dataset

ds  = load_dataset("imdb")                                  # 1. data
tok = AutoTokenizer.from_pretrained("distilbert-base-uncased")
ds  = ds.map(lambda b: tok(b["text"], truncation=True), batched=True)

model = AutoModelForSequenceClassification.from_pretrained(  # 2. model
    "distilbert-base-uncased", num_labels=2)

args = TrainingArguments(output_dir="out", num_train_epochs=2,  # 3. config
    per_device_train_batch_size=16, eval_strategy="epoch",
    learning_rate=2e-5, fp16=True)

Trainer(model=model, args=args,                              # 4. train
        train_dataset=ds["train"], eval_dataset=ds["test"],
        tokenizer=tok).train()

generate_image.py🤗 diffusers
from diffusers import StableDiffusionPipeline
import torch

pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5", torch_dtype=torch.float16
).to("cuda")

image = pipe("a cat in a spacesuit, digital art",
             num_inference_steps=30,        # how many denoising steps
             guidance_scale=7.5).images[0]  # how strongly to follow the prompt
image.save("cat.png")

deploy.pyPyTorch + FastAPI
# --- save & load weights ---
torch.save(model.state_dict(), "model.pt")
model.load_state_dict(torch.load("model.pt"))
model.eval()                                  # ALWAYS eval before serving

# --- minimal inference API ---
from fastapi import FastAPI
app = FastAPI()

@app.post("/predict")
def predict(payload: dict):
    x = preprocess(payload["text"])           # same prep as training!
    with torch.no_grad():
        logits = model(x)
    return {"label": int(logits.argmax(-1))}
# run: uvicorn deploy:app --host 0.0.0.0 --port 8000

load_4bit.pyTransformers + bitsandbytes
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
import torch

# 4-bit NF4 config: the recipe behind QLoRA
bnb = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",          # NormalFloat-4, tuned for weight distributions
    bnb_4bit_use_double_quant=True,     # quantize the quantization constants too
    bnb_4bit_compute_dtype=torch.bfloat16,  # math runs in bf16
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B-Instruct",
    quantization_config=bnb,
    device_map="auto",                  # place layers across available GPUs/CPU
)
tok = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-8B-Instruct")
# The 8B model now occupies ~5 GB instead of ~16 GB.

qlora_finetune.pyPEFT + TRL + Transformers
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer, SFTConfig
import torch

model_id = "meta-llama/Llama-3.1-8B"

# 1. load base in 4-bit (the "Q" in QLoRA) -- frozen
bnb = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4",
                         bnb_4bit_use_double_quant=True,
                         bnb_4bit_compute_dtype=torch.bfloat16)
model = AutoModelForCausalLM.from_pretrained(model_id, quantization_config=bnb,
                                             device_map="auto")
model = prepare_model_for_kbit_training(model)

# 2. attach LoRA adapters (the only thing that trains)
lora = LoraConfig(
    r=16, lora_alpha=32, lora_dropout=0.05, bias="none", task_type="CAUSAL_LM",
    target_modules=["q_proj","k_proj","v_proj","o_proj",      # attention
                    "gate_proj","up_proj","down_proj"],        # MLP -> "all-linear"
)
model = get_peft_model(model, lora)
model.print_trainable_parameters()   # e.g. "trainable: 0.42% of all params"

# 3. train like normal -- TRL handles the SFT loop
trainer = SFTTrainer(model=model, train_dataset=my_dataset,
                     args=SFTConfig(output_dir="out", per_device_train_batch_size=2,
                                    gradient_accumulation_steps=8, bf16=True,
                                    num_train_epochs=1, learning_rate=2e-4))
trainer.train()

# 4. save the tiny adapter (a few dozen MB), or merge for zero-latency deploy:
model.save_pretrained("my-lora-adapter")
# merged = model.merge_and_unload()   # fold B*A into W for serving

ppo_rlhf.pyTRL (Transformer Reinforcement Learning)
from trl import PPOTrainer, PPOConfig, AutoModelForCausalLMWithValueHead
from transformers import AutoTokenizer, pipeline

# policy = SFT model + a value head (the critic PPO needs)
policy = AutoModelForCausalLMWithValueHead.from_pretrained("my-sft-model")
ref    = AutoModelForCausalLMWithValueHead.from_pretrained("my-sft-model")  # frozen anchor
tok    = AutoTokenizer.from_pretrained("my-sft-model")

# a trained reward model scores responses (here via a text-classification pipeline)
reward_fn = pipeline("text-classification", model="my-reward-model")

ppo = PPOTrainer(PPOConfig(batch_size=32, learning_rate=1e-5), policy, ref, tok)

for batch in dataloader:
    queries   = batch["input_ids"]
    responses = ppo.generate(queries, max_new_tokens=128)      # policy acts
    texts     = [tok.decode(r) for r in responses]
    rewards   = [out["score"] for out in reward_fn(texts)]     # RM judges
    # PPO step: push policy toward higher reward, clipped + KL-penalized vs ref
    ppo.step(queries, responses, rewards)

preference_opt.pyTRL
# ---------- DPO: offline, from preference pairs ----------
from trl import DPOTrainer, DPOConfig
from transformers import AutoModelForCausalLM, AutoTokenizer

model = AutoModelForCausalLM.from_pretrained("my-sft-model")
tok   = AutoTokenizer.from_pretrained("my-sft-model")
# dataset columns: {"prompt", "chosen", "rejected"}
dpo = DPOTrainer(model=model, args=DPOConfig(beta=0.1, output_dir="dpo-out"),
                 train_dataset=pref_dataset, processing_class=tok)
dpo.train()                       # no reward model, no rollouts -- just stable training

# ---------- GRPO: online RL with a verifiable reward ----------
from trl import GRPOTrainer, GRPOConfig

def reward_correct(completions, answer, **kw):
    # +1 if the model's final answer matches ground truth, else 0  (RLVR)
    return [1.0 if extract(c) == a else 0.0 for c, a in zip(completions, answer)]

grpo = GRPOTrainer(
    model="my-sft-model",
    reward_funcs=reward_correct,
    args=GRPOConfig(num_generations=8,      # G samples per prompt -> the "group"
                    output_dir="grpo-out", learning_rate=1e-6),
    train_dataset=math_dataset,
)
grpo.train()

modern_block.pyPyTorch + Transformers
import torch, torch.nn as nn, torch.nn.functional as F

class RMSNorm(nn.Module):                      # cheaper than LayerNorm
    def __init__(self, d, eps=1e-6):
        super().__init__(); self.g = nn.Parameter(torch.ones(d)); self.eps = eps
    def forward(self, x):
        return self.g * x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + self.eps)

class SwiGLU(nn.Module):                        # gated FFN used by Llama/Qwen
    def __init__(self, d, hidden):
        super().__init__()
        self.w_gate = nn.Linear(d, hidden, bias=False)
        self.w_up   = nn.Linear(d, hidden, bias=False)
        self.w_down = nn.Linear(hidden, d, bias=False)
    def forward(self, x):
        return self.w_down(F.silu(self.w_gate(x)) * self.w_up(x))

# FlashAttention is exact attention done IO-efficiently. In practice you just
# enable the fused kernel -- PyTorch picks it automatically:
out = F.scaled_dot_product_attention(q, k, v, is_causal=True)   # uses Flash when available

# ...or tell Hugging Face to use it when loading a model:
from transformers import AutoModelForCausalLM
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B", attn_implementation="flash_attention_2",
    torch_dtype=torch.bfloat16)

train_at_scale.pyTransformers + Accelerate
from transformers import TrainingArguments

args = TrainingArguments(
    output_dir="out",
    per_device_train_batch_size=2,      # small micro-batch that fits in memory
    gradient_accumulation_steps=16,     # ...but train as if batch = 2 * 16 * n_gpus
    bf16=True,                          # mixed precision (Ampere/Hopper+)
    gradient_checkpointing=True,        # recompute activations -> big memory save
    optim="adamw_torch_fused",
    # multi-GPU sharding via FSDP (or use a DeepSpeed ZeRO-3 config file):
    fsdp="full_shard auto_wrap",
    fsdp_config={"transformer_layer_cls_to_wrap": "LlamaDecoderLayer"},
    logging_steps=10, num_train_epochs=1, learning_rate=2e-5,
)
# Launch across GPUs with:   accelerate launch train_at_scale.py
# (Accelerate / torchrun handle the distributed process group for you.)

serve.pyvLLM + Transformers
# ---------- High-throughput serving with vLLM ----------
from vllm import LLM, SamplingParams

llm = LLM(model="meta-llama/Llama-3.1-8B-Instruct",
          quantization="awq",          # quantized weights
          gpu_memory_utilization=0.9)  # PagedAttention + continuous batching are automatic
out = llm.generate(["Explain KV cache in one line."],
                   SamplingParams(temperature=0.7, max_tokens=128))
print(out[0].outputs[0].text)
# Or expose an OpenAI-compatible server:
#   vllm serve meta-llama/Llama-3.1-8B-Instruct

# ---------- Speculative decoding with a draft model ----------
from transformers import AutoModelForCausalLM, AutoTokenizer
target = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.1-70B-Instruct")
draft  = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3.2-1B-Instruct")
tok = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-70B-Instruct")
ids = tok("The capital of France is", return_tensors="pt").input_ids
out = target.generate(ids, assistant_model=draft, max_new_tokens=40)  # same output, faster
print(tok.decode(out[0]))

rag_and_agent.pysentence-transformers + an LLM client
# ---------- RAG: retrieve, then generate grounded on real docs ----------
from sentence_transformers import SentenceTransformer
import numpy as np

emb = SentenceTransformer("all-MiniLM-L6-v2")
docs = ["Our refund window is 30 days.", "Support hours are 9-5 GMT.", ...]
doc_vecs = emb.encode(docs, normalize_embeddings=True)

def retrieve(query, k=3):
    q = emb.encode([query], normalize_embeddings=True)[0]
    scores = doc_vecs @ q                       # cosine similarity
    return [docs[i] for i in np.argsort(scores)[-k:][::-1]]

def answer(query, llm):
    context = "\n".join(retrieve(query))         # top-k relevant chunks
    prompt  = f"Use ONLY this context:\n{context}\n\nQuestion: {query}"
    return llm(prompt)                           # grounded, less hallucination

# ---------- Agent loop: LLM decides which tool to call, then observes ----------
TOOLS = {"search": web_search, "python": run_code}   # name -> callable

def agent(goal, llm, max_steps=6):
    history = [f"Goal: {goal}"]
    for _ in range(max_steps):
        step = llm("\n".join(history) +
                   "\nThink, then output: TOOL <name> <args>  OR  FINAL <answer>")
        history.append(step)
        if step.startswith("FINAL"):
            return step[6:]
        name, args = parse(step)                 # e.g. "search", "vLLM throughput"
        observation = TOOLS[name](args)          # act
        history.append(f"Observation: {observation}")   # observe -> loop
    return "stopped: step budget exhausted"

01_langchain_agent.pyLangChain — create_agent
# pip install -qU langchain "langchain[anthropic]"
from langchain.agents import create_agent

def get_weather(city: str) -> str:
    """Get the weather for a given city."""        # docstring = the tool description
    return f"It's always sunny in {city}!"

agent = create_agent(
    model="claude-sonnet-4-6",     # swap to "openai:gpt-5.4" or "google_genai:gemini-3.5-flash"
    tools=[get_weather],
    system_prompt="You are a helpful assistant.",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in San Francisco?"}]}
)
print(result["messages"][-1].content_blocks)
# create_agent runs the loop: model -> (maybe call tool) -> observe -> repeat -> final answer.

02_langchain_rag.pyLangChain — agentic RAG
# pip install -qU langchain langchain-community langchain-chroma "langchain[anthropic]"
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain.embeddings import init_embeddings
from langchain.tools.retriever import create_retriever_tool
from langchain.agents import create_agent

# 1. INGEST: load -> split into chunks -> embed -> store in a vector DB
docs   = WebBaseLoader("https://example.com/handbook").load()
chunks = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=150).split_documents(docs)
store  = Chroma.from_documents(chunks, init_embeddings("openai:text-embedding-3-small"))

# 2. turn the retriever into a TOOL the agent can choose to call
retriever_tool = create_retriever_tool(
    store.as_retriever(search_kwargs={"k": 4}),
    name="company_handbook",
    description="Search the company handbook for policies, benefits, and procedures.",
)

# 3. the agent now decides WHEN to retrieve -> this is "agentic RAG"
agent = create_agent(model="claude-sonnet-4-6", tools=[retriever_tool],
                     system_prompt="Answer using the handbook. Cite what you find.")
print(agent.invoke({"messages": [{"role": "user", "content": "How many vacation days do I get?"}]}))

03_langgraph_graph.pyLangGraph — StateGraph
# pip install -U langgraph
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver

# nodes are just functions: (state) -> partial state update
def retrieve(state: MessagesState):
    query = state["messages"][-1].content
    hits  = store.as_retriever().invoke(query)          # your vector store
    return {"messages": [{"role": "system", "content": f"Context:\n{hits}"}]}

def generate(state: MessagesState):
    answer = model.invoke(state["messages"])            # your chat model
    return {"messages": [answer]}

# wire the graph: START -> retrieve -> generate -> END
g = StateGraph(MessagesState)
g.add_node(retrieve); g.add_node(generate)
g.add_edge(START, "retrieve")
g.add_edge("retrieve", "generate")
g.add_edge("generate", END)

# a checkpointer gives durable execution + conversation memory across turns
graph = g.compile(checkpointer=InMemorySaver())
graph.invoke({"messages": [{"role": "user", "content": "Summarize the refund policy."}]},
             config={"configurable": {"thread_id": "user-42"}})   # same thread = remembered
# Add g.add_conditional_edges(...) for the grade/retry branch, or interrupt() for human approval.

04_deep_agent.pyDeep Agents — create_deep_agent
# pip install -qU deepagents langchain-anthropic
from deepagents import create_deep_agent

def web_search(query: str) -> str:
    """Search the web and return the top results."""
    return run_search(query)        # plug in Tavily/SerpAPI/your own retriever here

# create_deep_agent bundles planning (write_todos), a virtual filesystem with
# automatic context compression, subagent spawning (the `task` tool), and memory.
agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[web_search],
    system_prompt=(
        "You are an expert researcher. Plan first, delegate sub-questions to "
        "subagents, save findings to files, then write a cited report."
    ),
)

# Give it a big, multi-step task -- it will plan, spawn subagents, and manage its own context.
result = agent.invoke({"messages": [{"role": "user", "content":
    "Compare QLoRA, DoRA and full fine-tuning on cost and quality. Write a sourced summary."}]})
print(result["messages"][-1].content)

01_lc_core.pyLangChain — create_agent
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import BaseModel, Field

# 1. a tool is a typed function; its docstring is the description the model reads
def search_flights(origin: str, destination: str, date: str) -> str:
    """Search available flights between two cities on a given date."""
    return f"3 flights from {origin} to {destination} on {date}: ..."

# 2. structured output: ask for parsed, validated data instead of prose
class FlightPick(BaseModel):
    airline: str = Field(description="chosen airline")
    price_usd: float
    depart_time: str

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[search_flights],
    system_prompt="You are a travel assistant. Pick the best-value flight.",
    response_format=FlightPick,             # -> result["structured_response"] is a FlightPick
    checkpointer=InMemorySaver(),           # -> short-term memory across turns
)

cfg = {"configurable": {"thread_id": "trip-1"}}
res = agent.invoke({"messages": [{"role": "user",
        "content": "Cheapest flight SFO to JFK on Dec 5?"}]}, config=cfg)
print(res["structured_response"])           # FlightPick(airline=..., price_usd=..., ...)
# Because of the thread_id, a follow-up "what about Dec 6?" remembers the context.

02_lc_middleware.pyLangChain — built-in middleware
from langchain.agents import create_agent
from langchain.agents.middleware import (
    SummarizationMiddleware,    # compress old history near the context limit
    ToolCallLimitMiddleware,    # cap tool calls (cost / runaway protection)
    ModelFallbackMiddleware,    # retry on a backup model if the primary fails
    PIIMiddleware,              # detect & redact sensitive data
    HumanInTheLoopMiddleware,   # pause for approval on risky tools
)
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="claude-sonnet-4-6",
    tools=[search_tool, send_email_tool],
    checkpointer=InMemorySaver(),                       # required by HITL
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        SummarizationMiddleware(model="gpt-5.4-mini",
                                trigger=("fraction", 0.8), keep=("messages", 20)),
        ToolCallLimitMiddleware(thread_limit=20, run_limit=8),
        ModelFallbackMiddleware("gpt-5.4-mini", "openai:gpt-5.4"),
        HumanInTheLoopMiddleware(interrupt_on={          # ask a human before sending
            "send_email_tool": {"allowed_decisions": ["approve", "edit", "reject"]},
            "search_tool": False,                        # auto-run safe tools
        }),
    ],
)
# Each middleware handles ONE concern and composes with the others -- no agent rewrite.

03_lc_custom_mw.pyLangChain — @hooks
from langchain.agents import create_agent
from langchain.agents.middleware import before_model, dynamic_prompt

# inject a fresh, per-request system prompt (e.g. user preferences from a store)
@dynamic_prompt
def personalized_prompt(request) -> str:
    user = request.runtime.context.get("user_name", "there")
    return f"You are a concise assistant. Address the user as {user}."

# a guardrail that runs right before every model call
@before_model
def log_and_guard(state) -> dict | None:
    last = state["messages"][-1].content
    if "wire transfer" in last.lower():
        return {"jump_to": "end"}        # short-circuit the loop
    return None                          # None -> proceed normally

agent = create_agent(
    model="claude-sonnet-4-6", tools=[...],
    middleware=[personalized_prompt, log_and_guard],
    context_schema=dict,                 # lets us pass runtime context
)
agent.invoke({"messages": [{"role": "user", "content": "hi"}]},
             context={"user_name": "Sara"})

01_lg_graph.pyLangGraph — StateGraph
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver

# 1. STATE: messages append (reducer); attempts overwrite by default
class State(TypedDict):
    messages: Annotated[list, add_messages]   # reducer -> appends, never clobbers
    attempts: int

# 2. NODES: (state) -> partial update
def call_model(state: State):
    reply = model.invoke(state["messages"])
    return {"messages": [reply], "attempts": state.get("attempts", 0) + 1}

def run_tools(state: State):
    results = execute_tool_calls(state["messages"][-1])
    return {"messages": results}

# 3. ROUTER for a conditional edge: keep looping or stop
def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls and state["attempts"] < 5:
        return "tools"
    return END

# 4. WIRE the graph (note the cycle tools -> agent)
g = StateGraph(State)
g.add_node("agent", call_model)
g.add_node("tools", run_tools)
g.add_edge(START, "agent")
g.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
g.add_edge("tools", "agent")              # loop back

graph = g.compile(checkpointer=InMemorySaver())
graph.invoke({"messages": [{"role": "user", "content": "Plan my week."}], "attempts": 0},
             config={"configurable": {"thread_id": "u1"}})

02_lg_hitl.pyLangGraph — interrupt / Command
from langgraph.types import Command, interrupt

def approve_purchase(state: State):
    # pause the graph and surface a payload to the caller
    decision = interrupt({"action": "buy", "item": state["item"], "cost": state["cost"]})
    if decision == "approve":
        return {"messages": [{"role": "system", "content": "Purchase approved."}]}
    return {"messages": [{"role": "system", "content": "Purchase cancelled."}]}

# ... add_node("approve", approve_purchase) ... compile(checkpointer=...)

cfg = {"configurable": {"thread_id": "order-9"}}
result = graph.invoke({...}, config=cfg)
# Execution pauses at interrupt(); result surfaces the pending interrupt payload:
print(result["__interrupt__"])     # (Interrupt(value={'action': 'buy', ...}),)

# A human reviews, then we RESUME -- the resume value becomes interrupt()'s return:
graph.invoke(Command(resume="approve"), config=cfg)   # continues exactly where it paused

03_lg_command_stream.pyLangGraph — Command / stream / Postgres
from langgraph.types import Command

# A node that BOTH updates state AND picks the next node -- no conditional edge needed
def supervisor(state: State) -> Command:
    nxt = "researcher" if needs_research(state) else "writer"
    return Command(goto=nxt, update={"messages": [route_note(nxt)]})

# Stream intermediate progress instead of waiting for the final result
for mode, chunk in graph.stream(inputs, config=cfg, stream_mode=["updates", "messages"]):
    print(mode, chunk)        # "updates" = per-node state deltas; "messages" = token stream

# Durable, multi-instance memory for production: swap the checkpointer
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://...") as saver:
    graph = g.compile(checkpointer=saver)   # survives restarts, scales across servers

01_deepagent.pyDeep Agents — create_deep_agent
from deepagents import create_deep_agent

def web_search(query: str) -> str:
    """Search the web and return results."""
    return run_tavily(query)

# specialized subagents: each gets its OWN isolated context window
research_subagent = {
    "name": "researcher",
    "description": "Deeply researches one focused sub-question and returns a summary.",
    "system_prompt": "You are a meticulous researcher. Cite sources.",
    "tools": [web_search],
}
critic_subagent = {
    "name": "critic",
    "description": "Reviews a draft for accuracy and gaps.",
    "system_prompt": "You are a sharp editor. List concrete fixes.",
}

agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[web_search],
    system_prompt=("You are a research lead. First write_todos to plan. "
                   "Delegate sub-questions to the researcher subagent, save findings "
                   "to files, then have the critic review before the final report."),
    subagents=[research_subagent, critic_subagent],
)

result = agent.invoke({"messages": [{"role": "user", "content":
    "Write a sourced report comparing LoRA, QLoRA, and DoRA on cost and quality."}]})
print(result["messages"][-1].content)
# Under the hood: write_todos -> task(researcher) x N -> write_file -> task(critic) -> report

02_deepagent_backends.pyDeep Agents — backends + HITL
from deepagents import create_deep_agent
from deepagents.backends import LocalShellBackend     # filesystem + real shell `execute`
from deepagents.middleware import FilesystemMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

# A coding-style agent: real local files + shell, long-term memory, approval gates
agent = create_deep_agent(
    model="anthropic:claude-sonnet-4-6",
    tools=[],
    backend=LocalShellBackend(workspace_root="/workspace"),  # ls/read/write/edit + execute
    system_prompt="You are a coding agent. Plan, edit files, run tests, then summarize.",
    checkpointer=InMemorySaver(),     # durable + resumable (also enables interrupts)
    store=InMemoryStore(),            # long-term memory across threads (swap for a DB)
    interrupt_on={"execute": {"allowed_decisions": ["approve", "reject"]}},  # gate shell cmds
)

# Because it returns a compiled LangGraph graph, you can stream subagent activity:
for mode, chunk in agent.stream({"messages": [{"role": "user",
        "content": "Fix the failing test in utils.py"}]},
        config={"configurable": {"thread_id": "dev-1"}}, stream_mode="updates"):
    print(mode, chunk)

for epoch in range(epochs):
    for xb, yb in loader:
        preds = model(xb)         # forward
        loss  = loss_fn(preds, yb)
        opt.zero_grad()           # clear grads  ← don't forget!
        loss.backward()           # backprop
        opt.step()                # update