LLM APIs, Embeddings and RAG

Calling LLM APIs from Python

Underneath the branding, every provider is the same shape: a POST with a list of messages and a JSON reply.

• The official SDKs (anthropic, openai) wrap this and handle retries – use them in real projects.
• Always set max_tokens and a timeout. An unbounded call is an unbounded bill.
• Temperature controls randomness: 0 for extraction and classification, higher only for creative drafting.
• Handle the error codes you will actually hit: 429 rate limit (back off and retry), 401 bad key, 529 overloaded.

import os, json, urllib.request

def ask(prompt: str, model: str = "claude-sonnet-5-5") -> str:
    body = json.dumps({
        "model": model,
        "max_tokens": 1024,
        "system": "You are a QA assistant. Be concise.",
        "messages": [{"role": "user", "content": prompt}],
    }).encode()
    req = urllib.request.Request(
        "https://api.anthropic.com/v1/messages",
        data=body,
        headers={
            "content-type": "application/json",
            "x-api-key": os.environ["ANTHROPIC_API_KEY"],   # never hard-coded
            "anthropic-version": "2023-06-01",
        },
    )
    with urllib.request.urlopen(req, timeout=60) as resp:
        data = json.loads(resp.read())
    return data["content"][0]["text"]

print(ask("Give me 3 boundary test cases for an age field that accepts 18 to 65."))

API keys, environment variables and secure configuration

The most common real-world AI incident is not a bad model. It is a leaked key.

• Keys live in environment variables or a secret manager. Not in code, not in a notebook, not in a screenshot.
• .env goes in .gitignore on day one. A key committed once is in the history forever.
• Separate keys for development and production, with separate spending limits.
• Rotate immediately if a key is exposed – and assume it was used.
• Never log the full prompt and response blindly: they contain whatever the user typed, including personal data.

# .env  (never committed - add it to .gitignore)
# ANTHROPIC_API_KEY=sk-ant-...

import os
from dotenv import load_dotenv

load_dotenv()
key = os.environ.get("ANTHROPIC_API_KEY")
if not key:
    raise RuntimeError("ANTHROPIC_API_KEY is not set - check your .env")

print("key loaded, length", len(key))   # print the length, never the key

Conversation history and memory patterns

The API is stateless. Every bit of “memory” is something your code chose to send again.

• To continue a conversation you resend the previous messages. That is the whole mechanism.
• History grows every turn, so cost and latency grow with it until you manage it.
• Trimming: keep the system prompt plus the last N turns.
• Summarising: replace older turns with a short summary – cheaper, but detail is lost and that loss is a testable risk.
• Persistent memory: store facts about the user in a database and inject only what is relevant.

MAX_TURNS = 10

def trim(messages):
    """Always keep the system message; keep only the most recent turns after it."""
    system = [m for m in messages if m["role"] == "system"]
    rest = [m for m in messages if m["role"] != "system"]
    return system + rest[-MAX_TURNS * 2:]

Embeddings and semantic similarity

Turning text into numbers so that meaning becomes measurable distance.

• An embedding is a vector – hundreds or thousands of numbers – that represents a piece of text.
• Similar meanings sit close together, even with no shared words: “cancel my policy” and “I want to terminate my cover”.
• Similarity is measured with cosine similarity (Module 3): 1 identical, 0 unrelated.
• Embeddings are produced by a separate, cheaper model than the one that writes answers.
• Everything must be embedded with the same model – mixing models makes the distances meaningless.

import numpy as np

def cosine(a, b):
    a, b = np.array(a), np.array(b)
    return float(a @ b / (np.linalg.norm(a) * np.linalg.norm(b)))

# embed() calls your provider's embedding endpoint
q = embed("how do I cancel my policy?")
for text in ["Policy cancellation process", "Premium payment options", "Claim settlement timeline"]:
    print(round(cosine(q, embed(text)), 3), text)

Vector databases and similarity search

Where the embeddings live so you can search millions of them quickly.

• A vector database stores vectors with their text and metadata, and finds nearest neighbours fast.
• Common choices: Chroma and FAISS for local work, Pinecone, Qdrant and Weaviate when hosted, pgvector if you already run Postgres.
• Store metadata with every vector – document name, section, date, access level. Without it you cannot cite or filter.
• Re-embedding the whole store is required if you change the embedding model. Plan for it.
• Index freshness is a real defect source: a policy updated yesterday that was never re-indexed will be answered from the old text.

Retrieval-Augmented Generation (RAG)

The standard way to make a model answer from your documents without retraining it.

• Index once: split documents into chunks, embed them, store them.
• Per question: embed the question, fetch the closest chunks, put them in the prompt, ask the model to answer only from them.
• The model supplies language; your retrieval supplies facts. Both can fail, and they fail differently.
• RAG beats fine-tuning for company knowledge: updating a document is a re-index, not a retraining run.
• Two failure modes to separate in every bug report: retrieval fetched the wrong chunks, or generation ignored the right ones.

Document loading, chunking and retrieval

Chunking decides the quality of the whole system, and it is almost always where the bugs are.

• Too small and the context is lost; too large and the real answer is diluted by surrounding noise.
• 800-1,200 characters with 10-20% overlap is a reasonable starting point – then measure.
• Split on structure (headings, clauses) rather than character count wherever the document has it.
• Tables and scanned PDFs need special handling; a table flattened into a paragraph usually becomes unusable.
• Keep the source and section in metadata so the answer can cite it.

def chunk(text: str, size: int = 800, overlap: int = 150):
    """Overlapping chunks so a sentence split across a boundary is still retrievable whole."""
    out, start = [], 0
    while start < len(text):
        out.append(text[start:start + size])
        start += size - overlap
    return out

parts = chunk(open("policy.txt", encoding="utf-8").read())
print(len(parts), "chunks, first one starts:", parts[0][:60])

Reranking, metadata filters and citations

Three upgrades that turn a demo-grade RAG into one people can trust.

• Rerank: fetch 20 candidates by vector similarity, then have a cross-encoder or a model reorder them. Nearly always worth the extra call.
• Metadata filters: restrict by product, region, language or date before the similarity search. This is also how access control is enforced.
• Citations: return the chunk id with every claim, and show the source to the user.
• Hybrid search: combine keyword search with vectors – exact ids, error codes and policy numbers need exact matching.
• Citations make automated testing possible: assert that the cited chunk really contains the quoted fact.

Evaluating retrieval quality and generated answers

You cannot improve what you do not measure – and RAG needs the two halves measured separately.

• Build a golden set: 30-50 real questions with the correct answer and the document that holds it.
• Retrieval metrics: was the right chunk in the top k (recall@k), and how high was it (MRR)?
• Generation metrics: is the answer supported by the retrieved text (groundedness), does it address the question (relevance), is it correct?
• Ragas and DeepEval automate much of this – Module 9 wires them into CI.
• Re-run the whole set on every prompt change, model change or re-index. This is your regression pack.

def recall_at_k(golden, retriever, k=5):
    hits = 0
    for question, expected_chunk_id in golden:
        ids = [c["id"] for c in retriever(question, k=k)]
        hits += expected_chunk_id in ids
    return hits / len(golden)

print("recall@5:", recall_at_k(GOLDEN, retrieve))   # below ~0.8 means fix retrieval, not the prompt

Project: a document chatbot over your own PDFs and policies

1. Collect five real documents – policies, handbooks, API docs. Extract the text.
2. Chunk them with overlap, embed them and store them with metadata for source and section.
3. Write the retrieve-then-answer loop with a system prompt that forbids answering outside the context and requires a citation.
4. Build a golden set of 30 questions, including five whose answers are deliberately not in the documents.
5. Measure recall@5 and count how many of the five unanswerable questions were refused correctly. Report both numbers – the refusals matter more than the answers.