LangChain

01 — Orientation & Setup

Know what LangChain is, its package boundaries, when to use it, and how to spot legacy tutorials.

LangChain is an open-source framework of standard interfaces and building blocks around model calls: messages, prompts, tools, structured output, retrieval, agents, execution, streaming, persistence, middleware, tracing, and evaluation. Its value is composition without locking the app to one provider.

You do not need LangChain just to call an LLM. A few direct SDK calls are clearer for single-turn scripts. Use LangChain when the app becomes a multi-component pipeline/agent, or when you want standardized streaming, tool calling, persistence, middleware, tracing, and evaluation.

LangChain vs. LangGraph vs. LangSmith

Complementary, not exclusive. create_agent is built on LangGraph, so you get persistence/streaming without writing graph code until control flow demands it.

Package layout (1.x)

Important: do not memorize "loader X lives in package Y." The integration surface changes faster than the core API. Check the current integration docs for the exact package.

Install

# Python 3.10+
python -m venv .venv
source .venv/bin/activate
pip install -U langchain
pip install -U langchain-openai langchain-anthropic  # as needed
export OPENAI_API_KEY="sk-..."

Old-tutorial radar

• langgraph.prebuilt.create_react_agent → langchain.agents.create_agent.
• Legacy chains / old memory classes / former langchain.retrievers namespace → compatibility surfaces under langchain-classic.
• Agent prompt=... (old API) → system_prompt=... in create_agent.
• Old chain/memory tutorials may still teach concepts, but do not copy them as 1.x reference code. Identify the generation first; import path first, version second.

Gotchas

• API churn: verify imports/versions against current docs before production copy-paste.
• Framework ≠ requirement: extra abstraction without multi-step needs is cost, not value.

Try it yourself: print langchain.__version__, make one direct SDK call and one init_chat_model call, and state what the framework added.

Primary sources: v1 release notes, v1 migration guide, Overview.

02 — Chat Models

Instantiate any provider behind one interface; drive it with invoke / stream / batch.

A chat model wraps a provider API behind one contract: same constructor style, same .invoke()/.stream()/.batch() (+ async twins), same message types out. Provider becomes configuration, not architecture.

Two ways to get a model

from langchain.chat_models import init_chat_model
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

model = init_chat_model("openai:gpt-5.5")               # needs langchain-openai
model = init_chat_model("anthropic:claude-sonnet-4-6") # needs langchain-anthropic

model = ChatOpenAI(model="gpt-5.5")
model = ChatAnthropic(model="claude-sonnet-4-6")

Use the factory when provider is config-driven; use the class for provider-specific options. Common params: model, temperature, max_tokens, timeout (s), max_retries.

Messages in, AIMessage out

from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, HumanMessage

model = init_chat_model("openai:gpt-5.5")
response = model.invoke([
    SystemMessage("You are a helpful assistant."),
    HumanMessage("Translate to French: I love programming."),
])
print(type(response))  # AIMessage
print(response.text)   # "J'adore la programmation."

Roles: SystemMessage (steering), HumanMessage (user), AIMessage (model turns + tool-call requests), ToolMessage (tool results, Lesson 10). Useful AIMessage attrs: .text, .tool_calls, .usage_metadata, .content_blocks (provider-agnostic multimodal/reasoning view).

Standard interface

response = model.invoke("Why is the sky blue?")
for chunk in model.stream("Tell me a short story."):  # yields AIMessageChunk
    print(chunk.text, end="", flush=True)
responses = model.batch(["Q1?", "Q2?"], config={"max_concurrency": 5})
# async: ainvoke, astream, abatch. Reuse one thread-safe instance.

Gotchas

• .content may be str or list of content blocks (images/reasoning). Use .text for plain string.
• Newest reasoning models may reject non-default temperature with 400 — a retired knob, not a tunable one.
• invoke returns AIMessage, not str — extract via .text or StrOutputParser.
• Prefer explicit "provider:model" over bare names; pin IDs over moving aliases.

Try it yourself: inspect model.invoke("Hello!").usage_metadata.

Primary sources: Models, Messages reference.

03 — Prompt Templates

Build parameterized, reproducible prompts — including few-shot — instead of f-strings.

Hand-concatenated prompts invite injection bugs and spaghetti. ChatPromptTemplate declares structure + variables up front; invoking fills holes into a message list. Missing variables fail before tokens are spent.

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder, FewShotChatMessagePromptTemplate
from langchain.messages import HumanMessage, AIMessage

prompt = ChatPromptTemplate.from_messages([
    ("system", "You answer in {style} style."),
    ("human", "{question}"),
])
prompt.invoke({"style": "pirate", "question": "What is a transformer?"}).to_messages()

# History splicing
hist_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are concise."),
    MessagesPlaceholder("history"),
    ("human", "{question}"),
])
hist_prompt.invoke({"question": "And in Rust?", "history": [HumanMessage("How to read a file in Python?"), AIMessage("open('f').read()")]})

import datetime
p = ChatPromptTemplate.from_messages([("human", "Today is {date}: {q}")])
p = p.partial(date=datetime.date(2026, 9, 1).isoformat())
p.invoke({"q": "what day is it?"})

examples = [{"input": "2+2", "output": "4"}, {"input": "2+3", "output": "5"}]
ex = ChatPromptTemplate.from_messages([("human", "{input}"), ("ai", "{output}")])
few = FewShotChatMessagePromptTemplate(examples=examples, example_prompt=ex)
full = ChatPromptTemplate.from_messages([("system", "Answer with just the number."), few, ("human", "{input}")])
full.invoke({"input": "2+9"}).to_messages()

Gotchas

• Literal braces (JSON) break f-string templates — double them or use template_format="mustache".
• Untrusted values belong inside messages, never concatenated into the system prompt.
• For simple agents a plain system_prompt="..." string is fine; templates pay off with multiple variables / chain composition.

Try it yourself: build a template with a history placeholder over a 3-turn conversation.

04 — LCEL — Composing Chains

Pipe prompts/models/code into one Runnable that streams, batches, and traces.

A Runnable has invoke/stream/batch (+ async). prompt | model | parser builds a RunnableSequence: output feeds next input. Value is composite behavior — end-to-end streaming, batching, retries, tracing — not brevity.

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough, chain

model = init_chat_model("openai:gpt-5.5")
prompt = ChatPromptTemplate.from_messages([("system", "Translate to {language}."), ("human", "{text}")])
chain_ = prompt | model | StrOutputParser()
chain_.invoke({"language": "Spanish", "text": "the weather is nice"})
for t in chain_.stream({"language": "Spanish", "text": "the weather is nice"}):
    print(t, end="", flush=True)

# Primitives: fan-out + carry-through
enriched = RunnablePassthrough().assign(topic_len=lambda d: len(d["topic"]))
enriched.invoke({"topic": "rag"})  # {'topic':'rag','topic_len':3}
retrieval_step = RunnableParallel(context=lambda d: fetch_docs(d["question"]), question=RunnablePassthrough())

# Resilience + traceable function
robust = prompt | model.with_retry(stop_after_attempt=3).with_fallbacks([backup_model]) | StrOutputParser()
@chain
def extract_topic(text: str) -> dict:
    return {"topic": text.split()[-1]}

Gotchas

• Enforce inter-step shape contracts; convert AIMessage→str explicitly.
• Any buffering RunnableLambda breaks end-to-end streaming by construction.
• Raw dicts coerce to RunnableParallel — keep the schema deliberate.
• Debug with chain.get_graph().print_ascii() / LangSmith, not prints.
• Plain Python is fine for plain logic; LCEL when you need streaming/batching/tracing/fallbacks.

Try it yourself: add .with_retry() and confirm the chain still streams.

Primary sources: LCEL guide, Runnable reference.

05 — Structured Output

Get validated objects, not prose; know provider-native vs tool-calling strategies and failure handling.

Apps want records. .with_structured_output(schema) returns a model-like runnable emitting typed objects. Capabilities/guarantees are provider-dependent — LangChain adapts a common interface to them. Schema validity ≠ factual truth.

from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model

class Movie(BaseModel):
    title: str = Field(description="Movie title")
    year: int = Field(description="Release year")
    director: str = Field(description="Director's name")

model = init_chat_model("openai:gpt-5.5")
obj = model.with_structured_output(Movie).invoke("Inception was directed by Christopher Nolan, released 2010.")
print(obj.title, obj.model_dump())
debug = model.with_structured_output(Movie, include_raw=True).invoke("...")
# {"raw": AIMessage, "parsed": Movie, "parsing_error": None} — keep raw for prod debugging

Methods: method="json_schema" (provider-native enforcement where supported), "function_calling" (schema as tool, works on tool-capable models), "json_mode" (valid JSON only — you describe schema in prompt, weakest). Prefer Pydantic (validation + typed access); TypedDict/JSON Schema give plain dicts.

from langchain.agents import create_agent
from langchain.agents.structured_output import ProviderStrategy, ToolStrategy

class TripPlan(BaseModel):
    destination: str
    summary: str

agent = create_agent(model="openai:gpt-5.5", tools=[], response_format=TripPlan)  # auto strategy
agent_native = create_agent(model="openai:gpt-5.5", tools=[], response_format=ProviderStrategy(TripPlan))
agent_tool = create_agent(model="anthropic:claude-sonnet-4-6", tools=[], response_format=ToolStrategy(TripPlan, handle_errors=True))
result = agent.invoke({"messages": [{"role": "user", "content": "Plan a weekend in Tokyo."}]})
print(result["structured_response"])

Gotchas

• Field descriptions are prompts — undescribed fields get guesses.
• Keep schemas small/flat where enforcement is weak; log/surface errors, define retry-or-fallback policy.
• StrOutputParser() remains the normal LCEL text tail; specialized parsers are the fallback when native structured output is unavailable. Parsers aren't obsolete — native first when supported.

Try it yourself: extract ArticleSummary(title, summary, topics: list[str]) from 3 articles incl. one adversarial input; compare raw vs parsed.

Primary sources: Structured output, Strategies reference.

06 — Memory & Conversation State

Distinguish history / agent state / checkpointer / runtime context / long-term store — and use each correctly.

from langchain.chat_models import init_chat_model
from langchain.messages import SystemMessage, HumanMessage
model = init_chat_model("openai:gpt-5.5")
history = [SystemMessage("You are helpful.")]
history += [HumanMessage("Hi, I'm Sam."), model.invoke(history + [HumanMessage("Hi, I'm Sam.")])]
print(model.invoke(history + [HumanMessage("What's my name?")]).text)

from langchain.agents import create_agent, AgentState
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(model="openai:gpt-5.5", tools=[], checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "conversation-1"}}
agent.invoke({"messages": [{"role": "user", "content": "My name is Sam."}]}, config=cfg)
print(agent.invoke({"messages": [{"role": "user", "content": "What's my name?"}]}, config=cfg)["messages"][-1].text)

class CustomState(AgentState):
    user_id: str
    preferences: dict
agent2 = create_agent(model="openai:gpt-5.5", tools=[], state_schema=CustomState, checkpointer=InMemorySaver())

from dataclasses import dataclass
@dataclass
class Context: user_id: str; plan: str
agent3 = create_agent(model="openai:gpt-5.5", tools=[], context_schema=Context)
agent3.invoke({"messages": [{"role": "user", "content": "Help me."}]}, context=Context(user_id="u-123", plan="pro"))

InMemorySaver is process-local (tests/learning). Production: durable checkpointer (e.g. Postgres) + same thread_id model; long-term facts go in a separate store, not the checkpointer. Manage growth via trim_messages / delete / summarization middleware — long context raises cost/latency and can bury signal.

Gotchas

• Silent stateless bug: thread_id without checkpointer=... = stateless, no error.
• Thread isolation: never share one user's thread_id with another.
• Checkpointer is persistence, not the definition of memory; state is not a queryable DB; long-term memory needs write/read/delete policy.

Try it yourself: two threads + custom user_id + runtime plan; state which persists vs per-run.

Primary sources: Short-term memory, Agents, create_agent.

07 — Document Loaders & Text Splitters

Turn source data into well-formed Documents and retrieval-friendly chunks.

RAG quality starts before embeddings. Pipeline: source → parsing/extraction → normalization → Document+metadata → chunking → embedding → index. A Document is just page_content + metadata; loaders are conveniences, not requirements.

from langchain_core.documents import Document
from pypdf import PdfReader
from langchain_text_splitters import RecursiveCharacterTextSplitter

reader = PdfReader("report.pdf")
docs = [Document(page_content=(pg.extract_text() or ""), metadata={"source": "report.pdf", "page": i+1, "document_id": "report-2026-q3"}) for i, pg in enumerate(reader.pages)]
chunks = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200, add_start_index=True).split_documents(docs)

RecursiveCharacterTextSplitter is a baseline (paragraph → sentence → word), not a universal optimum. 1000/200 and "10–20% overlap" are starting points — heading/code/table-aware splitting often wins on technical docs. Preserve provenance (source, page, section, doc ID, tenant, timestamp) at ingest or citations/filters become impossible.

Gotchas

• Chars ≠ tokens (EN ~4 chars/token — approximation, pass a tokenizer to length_function under token budgets).
• Don't flatten structure blindly; measure chunking by retrieval results, not blog defaults.

Try it yourself: compare 3 chunkings on one PDF/MD; judge which chunks stand alone.

Primary sources: Retrieval, v1 notes.

08 — Embeddings & Vector Stores

Build a small semantic index correctly; avoid the mistakes that make vector search unreliable.

Documents → Embedding model → Vectors+metadata → Index → Query embedding → Nearest-neighbor → Documents

from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.documents import Document

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")  # dims are model-configurable; don't hard-code
store = InMemoryVectorStore(embeddings)
store.add_documents([Document(page_content="LangChain builds LLM apps."), Document(page_content="Gradient descent minimizes loss."), Document(page_content="Eiffel Tower is in Paris.")])
print(store.similarity_search("How to build around an LLM?", k=2)[0].page_content)

# Persistent alternative (same high-level interface, different ops):
# pip install -U langchain-chroma
from langchain_chroma import Chroma
persistent = Chroma(collection_name="docs", embedding_function=embeddings, persist_directory="./chroma_db")
persistent.add_documents(store.similarity_search("LLM", k=2))

Gotchas

• Scores need interpretation: score may be similarity, distance, or normalized value — never copy thresholds across backends.
• One embedding space per index: changing models normally requires re-embedding; cache embeddings, batch via embed_documents.
• Similarity ≠ authorization: enforce tenant/user filters; vector search doesn't do access control.
• Semantic ≠ keyword: IDs/rare names may need lexical/hybrid too.

Try it yourself: 5-doc corpus + similarity_search_with_score + metadata-filtered retrieval; explain a similar-but-invalid hit.

Primary sources: Models/embeddings, Retrieval.

09 — Retrievers & RAG

Build baseline RAG, separate retrieval vs generation failures, know when to go agentic.

RAG = retrieve evidence → place in context → generate grounded answer. A vector DB is one retrieval mechanism; production may add keyword, filters, reranking, rewriting, hierarchy. A retriever is the interface query → list[Document].

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

chunks = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200).split_documents([Document(page_content=long_text, metadata={"source": "notes.txt"})])
store = InMemoryVectorStore(OpenAIEmbeddings()); store.add_documents(chunks)
retriever = store.as_retriever(search_type="similarity", search_kwargs={"k": 3})
def format_docs(docs): return "\n\n".join(d.page_content for d in docs)
model = init_chat_model("openai:gpt-5.5")
prompt = ChatPromptTemplate.from_messages([("system", "Answer using the provided context. If it lacks the answer, say you do not know.\n\nContext:\n{context}"), ("human", "{question}")])
rag_chain = (RunnableParallel(context=lambda d: format_docs(retriever.invoke(d["question"])), question=RunnablePassthrough()) | prompt | model | StrOutputParser())
print(rag_chain.invoke({"question": "What does the text say about X?"}))

What can go wrong?

Fixed vs agentic retrieval: fixed = every question retrieves once (predictable, cheap — start here). Agentic = search wrapped as tool; model decides if/when/how often (flexible, costlier). Beyond baseline: better chunking, filters, hybrid lexical+semantic, rewriting, reranking — then re-evaluate.

Gotchas

• Higher k adds noise + burns budget; "answer only from context" without a miss-policy breeds hallucination.
• Retrieved content is untrusted data — it can carry injection/malicious instructions. Never let it authorize actions.
• Keep provenance (source/page/IDs) or no citations; stale index = stale answers.

Try it yourself: 20-doc corpus × 5 questions (lookup, paraphrase, multi-doc, unsupported, conflicting); inspect chunks before answers.

Primary sources: RAG guide, Retrieval, Agents.

10 — Tools & Tool Calling

Expose functions to a model; run the bind → request → execute → reply loop by hand once.

The model never runs code — it emits a structured call; your code executes and returns a ToolMessage. Same provider feature that powers structured output.

from langchain.tools import tool
from pydantic import BaseModel, Field
@tool
def get_weather(location: str) -> str:
    """Get current weather for a location."""
    return f"It's always sunny in {location}!"

class WeatherInput(BaseModel):
    location: str = Field(description="City, e.g. 'San Francisco'")
    units: str = Field(default="celsius", description="Units")
@tool("fetch_weather", args_schema=WeatherInput)
def fetch_weather(location: str, units: str = "celsius") -> str: ...

from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage, ToolMessage
mwt = init_chat_model("openai:gpt-5.5").bind_tools([get_weather])
reply = mwt.invoke("What's the weather in Paris?")  # reply.tool_calls: [{name, args, id}]
msgs = [HumanMessage("What's the weather in Paris?"), reply]
for call in reply.tool_calls:
    msgs.append(ToolMessage(content=get_weather.invoke(call["args"]), tool_call_id=call["id"]))  # id MUST match
print(mwt.invoke(msgs).text)

Gotchas

• Type hints required; vague docstring = unusable tool (model decides from name+description+schema).
• Validate args at boundary — model output is user input before it touches DB/shell.
• Handle parallel tool_calls (iterate all); every call needs a paired ToolMessage or next call fails.

Try it yourself: add an irrelevant calculator; verify routing.

11 — Agents

Use create_agent correctly; know state/context/middleware and where LangGraph begins.

An agent is a decision loop: model sees context+tools → calls tool or answers → result re-enters state → repeat until stop. 1.x entry point is create_agent (LangGraph under the hood → persistence/streaming/interrupts free).

from langchain.agents import create_agent
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"It's always sunny in {city}!"
agent = create_agent(model="openai:gpt-5.5", tools=[get_weather], system_prompt="You are helpful. Use tools when useful.")
print(agent.invoke({"messages": [{"role": "user", "content": "Weather in SF?"}]})["messages"][-1].text)

Middleware (the 1.x control plane)

from pydantic import BaseModel
class FinalAnswer(BaseModel): answer: str; confidence: float
a2 = create_agent(model="openai:gpt-5.5", tools=[get_weather], response_format=FinalAnswer)
print(a2.invoke({"messages": [{"role": "user", "content": "Weather in SF?"}]})["structured_response"])

Use agents when step count/order is genuinely variable; use chains/explicit workflows when the path is known (cheaper, deterministic). Bound execution (time/steps/tool perms); never pre-bind tools before create_agent (pass raw model + tools=[...]). Reach for LangGraph directly for branches, parallel paths, custom transitions, long resumable interrupts, subgraphs/sub-agents.

Gotchas

• Agent ≠ autonomous program — it lives inside your tools/state/middleware/stops.
• Tool errors need policy (validate, catch, return error strings, cap retries); persistence needs checkpointer; per-run identity belongs in context, durable facts in a store.

Try it yourself: calculator + weather + error-returning tool; explain each routing decision from messages.

Primary sources: Agents, create_agent, Middleware, Migration.

12 — Streaming & Execution Events

Choose the right stream shape: tokens vs state updates vs events.

for chunk in model.stream("Explain transformers in 3 sentences."):  # AIMessageChunk
    print(chunk.text, end="", flush=True)
full = None
for c in model.stream("Hi"): full = c if full is None else full + c

for chunk in agent.stream({"messages": [{"role": "user", "content": "Weather in Boston?"}]}, stream_mode="messages", version="v2"):
    if chunk["type"] == "messages": print(chunk["data"][0].text, end="", flush=True)
# modes: messages (tokens/tool-call chunks) | updates (graph-node state) | custom (app progress). Combine as needed.
async for ev in model.astream_events("Hello"):
    if ev["event"] == "on_chat_model_stream": print(ev["data"]["chunk"].text, end="")
chain_ = prompt | model | StrOutputParser()
for t in chain_.stream({"question": "Why RAG?"}): print(t, end="", flush=True)

Callbacks vs middleware: callbacks = passive observability/logging; middleware = active control (state, guardrails, routing). New agent behavior belongs in middleware. Streaming aids time-to-first-output, not total compute; design renderers for text/tool/state chunks; use async streaming in async apps; never copy cross-generation event loops without checking shape.

Try it yourself: same agent via invoke vs token stream vs updates; decide what the frontend sees.

Primary sources: Streaming, Runnable.

13 — Observability with LangSmith

Trace runs to answer what the model saw, why tools fired, and where time went.

Input/output pairs can't explain multi-step failures. LangSmith (optional, not required to run LangChain) records steps, rendered inputs, outputs, latency, tokens, errors, metadata. Custom code needs @traceable.

pip install -U langsmith
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="ls__..."
export LANGSMITH_PROJECT="my-project"  # set regional endpoint if non-US

Tracing ≠ evaluation: tracing records what happened; evaluation judges quality. Flow: trace → inspect → dataset example → evaluate → compare. Traces contain prompts/docs/PII — redact, set retention/access/region deliberately. Local prints/callback tracers suffice for tiny scripts; LangSmith pays off for shared traces, experiments, and prod monitoring.

Try it yourself: trace Lesson 09 chain; name chunks, input, latency, output; save one failure as eval example.

Primary sources: Tracing quickstart, Overview.

14 — Testing & Evaluation

Test determinism with unit tests; judge quality with evals; score agents on outcome + trajectory.

def test_prompt(): assert prompt.invoke({"question": "What is RAG?"}).to_messages()[-1].content == "What is RAG?"
def test_tool_schema(): assert "location" in get_weather.args["properties"]

from langsmith import Client
client = Client()
def target(inputs: dict): return {"answer": my_rag_chain.invoke({"question": inputs["question"]})}
def exact_match(outputs, reference_outputs): return outputs["answer"] == reference_outputs["answer"]
results = client.evaluate(target, data="my-rag-dataset", evaluators=[exact_match], experiment_prefix="rag-v1", max_concurrency=2)

Structure is dataset → target → evaluators. Build datasets from real usage/failures (easy, hard, unsupported, ambiguous, adversarial); score retrieval and groundedness separately so model isn't blamed for retrieval. Value is version comparison (prompt/model/retriever A vs B), not one score. Small datasets lie; LLM judges need rubrics + hand-labeled calibration; a right answer after a dangerous tool call is still a failure.

Try it yourself: 20-example RAG set (5 unsupported + 5 hard retrieval); change retriever, rerun.

Primary sources: Eval quickstart, Agents.

15 — Production Readiness

Ship a demo as a dependable service: reliability, security, data, observability, eval gates, ops.

from langchain.chat_models import init_chat_model
from langchain.rate_limiters import InMemoryRateLimiter
model = init_chat_model("openai:gpt-5.5", timeout=30, max_retries=6,
    rate_limiter=InMemoryRateLimiter(requests_per_second=0.5, check_every_n_seconds=0.1, max_bucket_size=10))
resilient = model.with_retry(stop_after_attempt=3).with_fallbacks([backup_model])  # retries/fallbacks multiply cost + paths

InMemoryRateLimiter is process-local — multi-worker needs gateway/queue/shared limiter + provider quotas. Bound agents (time/steps/tool perms) via middleware (limits, HITL approval, guardrails); validate + authorize every tool call (untrusted input → schema → policy → execute → sanitized result); never assume model-chosen = safe. Templates don't sanitize untrusted content — injection defense is architecture: least privilege, instruction/data separation, sandboxing, output controls, approval for impact.

Persistence: durable checkpointer (not process memory) for threads; separate long-term store for durable facts. Manage context (trim/summarize/selective retrieval/compact tool outputs) — it affects correctness, not just cost. Measure tokens/latency first; smaller models, fewer calls, caching, streaming for perceived latency. Still need FastAPI/auth/secrets/queues/logging/CI-CD/rollback.

Try it yourself: one-page prod design for your Lesson 11 agent covering all rows above.

Primary sources: Model resilience, Middleware, Memory, Tracing.