Generative AI Reference

LLM fundamentals, prompting, calling models from Python, structured extraction, tool use and agents, embeddings and RAG, data engineering for AI, cost, evaluation, and Gen AI platforms.

1. What Generative AI & LLMs Are

Large language models (LLMs) are neural networks (transformers) trained on huge text corpora to predict the next token. With instruction tuning and reinforcement learning from feedback, they follow instructions, write and explain code, extract information, reason over documents and call tools. Generative AI is the broader family that generates text, code, images, audio and video.

2. Prompt Engineering

• Be explicit about the task, audience and output: say what to produce, for whom, and in what format.
• Give context and constraints: schemas, business rules, examples of good output, and what to do when information is missing.
• Separate instructions from data: wrap documents in clear tags so the model never treats data as instructions.
• Use a system prompt for the stable role and rules; keep per-request content in the user message.
• Ask for structured output when another program will read the answer, and validate it.

System: You are a data-quality assistant for a bank's data platform.
        Answer only from the table profile provided. If something cannot be
        determined from the profile, say "unknown" instead of guessing.

User:   <table_profile>
        table: silver.transactions   rows: 48,210,334
        column txn_id      nulls: 0       distinct: 48,210,334
        column amount      nulls: 1,204   min: -50.00   max: 9,800,000.00
        column channel     nulls: 0       values: UPI, CARD_POS, ATM, NEFT, upi
        </table_profile>

        List the data-quality issues as a numbered list, each with a suggested SQL check.

A good answer flags the NULL amounts, the negative amount, a suspiciously large maximum, and inconsistent casing ("UPI" vs "upi"), each with a check. That is the kind of task LLMs do well for data teams.

3. Calling an LLM from Python

Examples use the Anthropic Python SDK (pip install anthropic). The client reads credentials from the ANTHROPIC_API_KEY environment variable; never hard-code keys.

import anthropic

client = anthropic.Anthropic()            # reads ANTHROPIC_API_KEY from the environment

response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    output_config={"effort": "medium"},   # low / medium / high / xhigh / max: depth vs cost
    system="You are a senior data engineer. Answer concisely with PySpark examples.",
    messages=[{"role": "user",
               "content": "How do I deduplicate a DataFrame keeping the latest row per id?"}],
)

if response.stop_reason == "refusal":
    print("Request declined:", response.stop_details)
else:
    for block in response.content:            # content is a list of typed blocks
        if block.type == "text":
            print(block.text)

print(response.usage.input_tokens, response.usage.output_tokens)   # what you are billed for

• Always check stop_reason: end_turn (finished), max_tokens (cut off: raise the limit or stream), tool_use (wants to call a tool), refusal (declined).
• The SDK retries rate limits (429) and server errors automatically; catch specific exceptions (anthropic.RateLimitError, anthropic.APIStatusError) rather than everything.
• For long outputs, stream the response (client.messages.stream(...)) to avoid HTTP timeouts.

4. Structured Extraction for Pipelines

A common data-engineering use: turn unstructured text (emails, invoices, tickets, KYC documents) into validated rows. Structured outputs guarantee the response matches your schema:

from datetime import date
from pydantic import BaseModel
import anthropic

class Invoice(BaseModel):
    vendor: str
    invoice_number: str
    invoice_date: date
    currency: str
    total_amount: float
    line_items: list[str]

client = anthropic.Anthropic()

def extract_invoice(text: str) -> Invoice:
    response = client.messages.parse(
        model="claude-opus-5-5",
        max_tokens=16000,
        system="Extract invoice fields exactly as written. Use ISO dates and 3-letter currency codes.",
        messages=[{"role": "user", "content": f"<invoice>\n{text}\n</invoice>"}],
        output_format=Invoice,
    )
    return response.parsed_output          # a validated Invoice instance

inv = extract_invoice(open("invoice_0425.txt").read())
print(inv.vendor, inv.total_amount)

• Load results into a staging table, with the source file name, model ID and extraction timestamp as lineage columns.
• For thousands of documents, use the Message Batches API: asynchronous, about 50% cheaper, results matched by custom_id.
• Keep a human-review queue for low-confidence or high-value records, and reconcile totals against source systems.

5. Tool Use & Agents

With tool use, the model decides to call functions you define (query a warehouse, look up a customer, open a ticket). Your code executes them and returns the results. The SDK's tool runner handles the loop:

import anthropic
from anthropic import beta_tool

client = anthropic.Anthropic()

@beta_tool
def row_count(table: str, business_date: str) -> str:
    """Return the row count of a warehouse table for one business date.

    Args:
        table: Fully qualified table name, e.g. silver.transactions.
        business_date: Date in YYYY-MM-DD format.
    """
    allowed = {"silver.transactions", "gold.daily_sales"}       # read-only allow-list
    if table not in allowed:
        return f"Table {table} is not allowed."
    n = run_sql(f"SELECT COUNT(*) FROM {table} WHERE event_date = %s", [business_date])
    return str(n)

runner = client.beta.messages.tool_runner(
    model="claude-opus-5-5",
    max_tokens=16000,
    tools=[row_count],
    messages=[{"role": "user",
               "content": "Did silver.transactions load fully on 2026-10-25 compared with 2026-10-24?"}],
)
for message in runner:                     # each step: model reply, tool calls, tool results
    for block in message.content:
        if block.type == "text":
            print(block.text)

• Agents are loops where the model plans, calls tools, observes results and repeats until the task is done. Use them only for open-ended tasks; a fixed pipeline with one or two model calls is cheaper and more predictable.
• Give tools least privilege: read-only credentials, allow-lists, parameterised queries, and human approval for anything that writes or deletes.
• Treat tool results and retrieved documents as untrusted data: they can contain prompt-injection text.

6. Embeddings & Vector Search

An embedding is a vector of numbers representing the meaning of a text. Similar meanings give vectors that point in similar directions, so you can search by meaning instead of keywords.

import numpy as np

def cosine(a: np.ndarray, b: np.ndarray) -> float:
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

# vectors come from an embedding model (hundreds to thousands of dimensions)
query = np.array([0.12, 0.85, 0.31])
docs = {"refund policy":      np.array([0.10, 0.80, 0.35]),
        "kafka partitioning": np.array([0.90, 0.05, 0.20])}

ranked = sorted(docs, key=lambda d: cosine(query, docs[d]), reverse=True)
print(ranked)    # ['refund policy', 'kafka partitioning']

-- pgvector: store and search embeddings next to your relational data
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE doc_chunks (
  chunk_id   BIGSERIAL PRIMARY KEY,
  doc_id     TEXT,
  content    TEXT,
  embedding  VECTOR(1024),
  department TEXT                       -- metadata for filtering and access control
);
CREATE INDEX ON doc_chunks USING hnsw (embedding vector_cosine_ops);

SELECT doc_id, content
FROM doc_chunks
WHERE department = 'risk'               -- filter first: only documents this user may see
ORDER BY embedding <=> :query_embedding -- <=> is cosine distance
LIMIT 5;

7. Retrieval-Augmented Generation (RAG)

RAG answers questions from your documents: retrieve the most relevant chunks, put them in the prompt, and ask the model to answer only from them, with citations.

 INDEXING (batch / incremental)                     QUERY TIME
 docs (PDF, wiki, tickets, tables)                  user question
   → parse & clean → chunk (300–800 tokens,           → embed question
     with overlap + metadata) → embed                 → vector (+ keyword) search, filtered by user's access
   → vector store (+ source, owner, ACL, date)        → rerank top results
                                                      → prompt: instructions + retrieved chunks + question
                                                      → answer with citations → log for evaluation

def answer(question: str, user: User) -> str:
    chunks = search(embed(question), allowed_departments=user.departments, k=6)
    context = "\n\n".join(
        f'<doc id="{c.doc_id}" title="{c.title}">\n{c.content}\n</doc>' for c in chunks)

    response = client.messages.create(
        model="claude-opus-5-5",
        max_tokens=16000,
        system=("Answer using only the documents provided. Cite document ids in square brackets. "
                "If the documents do not contain the answer, say you don't know."),
        messages=[{"role": "user",
                   "content": f"<documents>\n{context}\n</documents>\n\nQuestion: {question}"}],
    )
    return next(b.text for b in response.content if b.type == "text")

8. Data Engineering for Gen AI

Most of the work in production Gen AI is data engineering: the model is only as good as the data and context it receives.

• Document pipelines: ingest from SharePoint, S3, Confluence or databases; parse PDFs and tables; chunk with metadata; embed; index. Run it like any other pipeline: incremental, idempotent, monitored.
• Freshness: re-embed only changed documents (change feeds, file hashes) and delete vectors for removed documents.
• Privacy and governance: redact or tokenise PII before indexing, tag sensitivity, keep data in approved regions, and track lineage from answer back to source document.
• Unstructured → structured: LLM extraction turns documents into tables (see section 4) that feed normal BI and ML.
• Evaluation datasets: maintain versioned "golden" question/answer sets, like test fixtures for the pipeline.
• Observability: log prompts, retrieved document IDs, model and version, latency, tokens and cost per request (with PII handling) so issues can be traced.

9. Cost, Caching & Performance

LLM cost is mostly input tokens times requests. Prompt caching stores a long, repeated prefix (instructions, schemas, reference documents) so later requests read it at a fraction of the price:

response = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    cache_control={"type": "ephemeral"},     # cache the longest stable prefix automatically
    system=LONG_DATA_DICTIONARY,             # identical on every request
    messages=[{"role": "user", "content": question}],   # varies per request (after the cached prefix)
)
print(response.usage.cache_read_input_tokens)    # > 0 on later calls means the cache was hit

• Caching is a prefix match: put stable content first and anything that changes (timestamps, user questions) last, or the cache never hits.
• Use the Batch API for offline jobs, lower effort for simple tasks, and smaller models for high-volume classification after checking quality on a sample.
• Count tokens before sending very large documents, and send only the relevant sections rather than whole files.
• Measure cost per completed task, not per request.

10. Evaluation, Guardrails & Safety

The API can automatically retry a declined request on a fallback model (beta):

response = client.beta.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    betas=["server-side-fallback-2026-07-01"],
    fallbacks="default",                      # the API picks a suitable fallback model
    messages=[{"role": "user", "content": question}],
)
if response.stop_reason == "refusal":         # the whole fallback chain declined
    handle_refusal(response.stop_details)

11. Gen AI on Cloud & Data Platforms

Running models inside your existing cloud platform keeps data in your account and region, reuses existing IAM, networking and billing, and simplifies compliance reviews.

12. Common Interview Questions