Session 1 got you to AgentExecutor.
This is everything between there and shipping — graphs, memory, evaluation, cost, deployment, and the multi-agent patterns the field is converging on.
{sub}
You now have the mental models for the 10 things that separate hobby agents from shipped ones. None of these are deep on their own — but together they're 90% of what senior agent engineers do.
Pick one chapter challenge. Build it this week. The next session goes deep on whichever you find hardest.
The vocabulary collapses two distinct things. Generative systems are reactive: prompt in, content out. Agentic systems are proactive: they decompose a goal, pick tools, observe results, and revise their plan.
Think of LangChain as the library of parts (LLMs, prompts, retrievers, tools, output parsers) and LangGraph as the runtime for assembling them into stateful systems.
(state) → partial state update. Pure-ish function.{`from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
query: str
answer: str
def respond(state: State) -> dict:
return {"answer": llm.invoke(state["query"]).content}
g = StateGraph(State)
g.add_node("respond", respond)
g.add_edge(START, "respond")
g.add_edge("respond", END)
app = g.compile()
app.invoke({"query": "Hello"})`}Note the reducer: by default, returning {`{"answer": "..."}`} replaces the field. For lists you usually want operator.add (append) — that's how messages accumulate across nodes.
One node, then the next. The graph version of a chain.
{`g.add_edge(START, "outline")
g.add_edge("outline", "draft")
g.add_edge("draft", "polish")
g.add_edge("polish", END)`}Multiple edges from one node fire concurrently; LangGraph fan-ins automatically when all upstream nodes complete.
{`g.add_edge("split", "translate_fr")
g.add_edge("split", "translate_de")
g.add_edge("split", "translate_jp")
# All three feed merge:
g.add_edge("translate_fr", "merge")
g.add_edge("translate_de", "merge")
g.add_edge("translate_jp", "merge")`}A router function picks the next node based on state.
{`def route(s): return s["intent"] # "tech" | "sales" | "billing"
g.add_conditional_edges("classify", route, {
"tech": "tech_agent",
"sales": "sales_agent",
"billing": "billing_agent",
})`}The most useful pattern in production: generate → evaluate → loop until the evaluator approves (or you hit a max-iteration cap).
{`def should_continue(state):
if state["score"] >= 8 or state["iterations"] >= 5:
return END
return "generator"
g.add_conditional_edges("evaluator", should_continue,
{"generator": "generator", END: END})`}max_iterations, an evaluator that's too strict will loop forever and burn through your token budget while you sleep.
Without a checkpointer, every invoke starts from scratch. With one, state is automatically saved at each super-step and keyed by a thread_id you pass in the config.
{`from langgraph.checkpoint.memory import MemorySaver
# For prod: from langgraph.checkpoint.postgres import PostgresSaver
app = graph.compile(checkpointer=MemorySaver())
cfg_a = {"configurable": {"thread_id": "user_42_session_a"}}
app.invoke({"messages": [("user", "My name is Sam.")]}, cfg_a)
app.invoke({"messages": [("user", "What's my name?")]}, cfg_a)
# → "Your name is Sam." (loaded from checkpoint)
cfg_b = {"configurable": {"thread_id": "user_99_session_a"}}
app.invoke({"messages": [("user", "What's my name?")]}, cfg_b)
# → "I don't know your name." (different thread, fresh state)`}thread_id. Everything else — sessions, history, resume — is a property of those three things.
{lede}
The big shift from Session 1: instead of trusting the LLM to decide when to stop, you draw the control flow yourself. Each node receives the current state, returns an update, and a conditional edge picks the next node based on what changed.
TypedDict) that flows through the graph.state → partial state update.A → B) or conditional (A → fn(state) → B|C|END).thread_id, so you can pause, resume, or branch.This graph classifies a user message, optionally calls a tool, then responds. Watch the active node and how state grows turn by turn.
{`from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
messages: list
intent: str | None
next: str | None
graph = StateGraph(State)
graph.add_node("classify", classify_intent)
graph.add_node("tool_use", call_tool)
graph.add_node("respond", generate_response)
graph.set_entry_point("classify")
graph.add_conditional_edges(
"classify",
lambda s: s["next"], # router function
{"tool_use": "tool_use", "respond": "respond"},
)
graph.add_edge("tool_use", "respond")
graph.add_edge("respond", END)
app = graph.compile(checkpointer=MemorySaver())`}interrupt() inside any node and the graph pauses — the checkpointer freezes state, you ship the partial result to a UI, and resuming continues exactly where you left off. This is the whole substrate for approval flows.
The prebuilt create_react_agent(model, tools) is a 5-line LangGraph that you can crack open and modify. You start with the same convenience, but every line is now editable — add a guardrail node, swap models per branch, log to LangSmith from one place.
The model doesn't actually call your function. It emits a structured request — "I want to call get_weather with city='Tokyo'" — and your runtime executes it and feeds the result back. The LLM is the planner; your code is the doer.
search_internal_kb beats search. The model picks tools by name and docstring before it picks them by spec.request_id) so duplicates are detectable.{`from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""Returns current weather for a city. Use for any 'what's the weather' query."""
return f"{city}: 18°C, light rain"
agent = create_react_agent(model, tools=[get_weather, search_web, send_email])`}Three reasons to reach for a subgraph: encapsulation (the parent doesn't need to know how research happens), reuse (drop the same research subgraph into 4 different products), and state isolation (the subgraph has its own private state that doesn't pollute the parent).
{`# Build the subgraph independently
research = StateGraph(ResearchState)
research.add_node("search", search_node)
research.add_node("fact_check", fact_check_node)
research.add_edge("search", "fact_check")
research.add_edge("fact_check", END)
research.set_entry_point("search")
research_app = research.compile()
# Use it as a node in the main graph
main = StateGraph(MainState)
main.add_node("research_team", research_app) # ← subgraph as node
main.add_node("writer", writer_node)
main.add_edge("research_team", "writer")`}input and output mappers — but it adds a coordination cost. Keep state shapes aligned where possible.Watch them work in concert: a user shares preferences over 8 messages, the buffer slides, the summary forms, and 2 facts get written to the vector store for next time.
In LangGraph, the checkpointer keys state by thread_id. That's your conversation memory — it follows one user-session. Long-term memory lives outside the graph entirely (Postgres + pgvector, Pinecone, whatever) and is loaded into state at the start of each turn.
{`# At the top of each turn:
relevant = vectorstore.similarity_search(user_msg, k=3)
state["context"] = relevant + state["messages"][-6:]
# At the end:
if worth_remembering(user_msg):
vectorstore.add_texts([extract_fact(user_msg)])`}LLMs return text. Your code wants objects. Without structure, you write fragile regex and pray. With structure, the model itself is constrained to emit valid JSON matching your Pydantic schema, and the output is auto-parsed.
Three mechanisms, in roughly the order they appeared:
{`from pydantic import BaseModel
from typing import Literal
class Task(BaseModel):
title: str
owner: str
due_date: date
priority: Literal["low", "med", "high"]
tags: list[str] = []
budget_usd: int
structured_llm = llm.with_structured_output(Task)
task = structured_llm.invoke("Plan the Q2 review, owner alex@co, due June 15, high priority, $12.5k")
# task is a Task instance — typed, validated, ready to use`}text-embedding-3-small is the default; domain-tuned models (legal, medical) win on niche corpora.{`# Indexing (offline)
docs = TextLoader("policies/").load()
chunks = RecursiveCharacterTextSplitter(chunk_size=600, chunk_overlap=60).split_documents(docs)
vectorstore = PGVector.from_documents(chunks, OpenAIEmbeddings(), connection=DB)
# Query (per request)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt | llm | StrOutputParser()
)`}{`import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__..."
os.environ["LANGCHAIN_PROJECT"] = "agents-prod"
# That's it. Every chain, agent, tool call now traces automatically.
# View at smith.langchain.com.`}(input, expected) pairs. The single most valuable artifact in your repo.For open-ended outputs (summaries, translations, code), a small LLM scores the candidate against a rubric. The judge prompt is itself a versioned artifact — you eval the judge against human-labeled examples to make sure it agrees with you.
{`JUDGE = """You are evaluating a customer support response.
Rubric:
- 5: solves the issue, warm tone, no errors
- 3: addresses the issue, but tone is off or info incomplete
- 1: wrong, unhelpful, or unsafe
Question: {question}
Reference: {reference}
Candidate: {candidate}
Return JSON: {{"score": 1-5, "reason": "..."}}"""
judge = (ChatOpenAI(model="gpt-4o-mini")
.with_structured_output(Score))`}A real input rarely arrives clean. It might contain personal data you can't store, a jailbreak attempt, or just a query so abusive you don't want to spend tokens on it. Stage your defenses so the LLM is the last thing that sees the request.
(prompt_tokens, completion_tokens, model) on every call. You can't optimize what you can't see.Watch routing in action. 8 queries arrive — some easy, some hard, some repeats. The router picks a model (or hits cache) for each.
{`def route(query: str) -> str:
if cache_hit := semantic_cache.get(query):
return cache_hit # ~free
score = complexity_classifier(query) # 0..1
if score < 0.6:
return cheap_model.invoke(query) # haiku-class
return expensive_model.invoke(query) # gpt-4-class`}An LLM call takes 2–6 seconds. Without streaming, the user stares at a spinner. With token streaming and tool-call status banners, the same wait feels like progress.
chunk from the model API is forwarded immediately. Users see typing.json-stream, Outlines partial mode). Render fields the moment they're complete.{`# FastAPI + SSE
@app.get("/chat")
async def chat(q: str):
async def stream():
async for event in agent.astream_events({"q": q}, version="v2"):
if event["event"] == "on_chat_model_stream":
yield f"data: {event['data']['chunk'].content}\\n\\n"
elif event["event"] == "on_tool_start":
yield f"event: tool\\ndata: {event['name']}\\n\\n"
return StreamingResponse(stream(), media_type="text/event-stream")`}Every LangChain runnable has both invoke and ainvoke. Once you're in async-land, parallelize anything independent — fetches, retrievals, multiple model calls — with asyncio.gather.
{`# Sync — runs serially. Total ≈ sum(latencies).
user = fetch_user(uid)
orders = fetch_orders(uid)
recs = fetch_recommendations(uid)
# Async — runs in parallel. Total ≈ max(latencies).
user, orders, recs = await asyncio.gather(
fetch_user_a(uid),
fetch_orders_a(uid),
fetch_recommendations_a(uid),
)
# In LangGraph: nodes that don't depend on each other auto-parallelize
# when you use Send (fan-out). See "Map-reduce" below.`}requests) with async — they block the event loop. Use httpx in async contexts. (2) Concurrency limits matter — Anthropic gives you ~50 req/s. Use a Semaphore to cap your fan-out.{`from langgraph.types import interrupt
def review_node(state):
decision = interrupt({ # ← graph pauses, state checkpointed
"draft": state["draft"],
"ask": "approve, edit, or reject?",
})
return {"approved": decision == "approve",
"draft": decision.get("edited", state["draft"])}
# Resume by invoking with the human's response:
graph.invoke(Command(resume="approve"), config={"configurable": {"thread_id": "t1"}})`}{`# Get the history of a thread
history = list(graph.get_state_history(config))
# Pick any past checkpoint
target = history[2] # 3 steps ago
# Fork from there with optional state edit
graph.update_state(target.config, {"prompt_version": "v2"})
graph.invoke(None, target.config) # resumes from c2 with new state`}A node returns a list of Send objects, each addressing a downstream node with its own slice of state. The runtime fans them out, awaits them all, and the next node sees the aggregated results.
{`from langgraph.types import Send
def fan_out(state):
return [Send("classify", {"doc": d}) for d in state["docs"]]
def classify(state):
sentiment = llm.with_structured_output(Sentiment).invoke(state["doc"])
return {"results": [sentiment]} # reducer concatenates
graph.add_conditional_edges("split", fan_out, ["classify"])
graph.add_edge("classify", "aggregate")`}A web request that takes 30 seconds is a problem. An agent run that takes 30 seconds is normal. The architectural fix: the request returns a run_id, a worker picks up the actual job, and the client polls or subscribes for completion.
thread_id resumes survive deploys.Watch a supervisor coordinate three specialists on a multi-step task.
computer_use, OpenAI's Operator. The model sees screenshots and emits click(x,y), type(text), screenshot(). Useful when there's no API.{`# Code-execution sandbox (Modal, E2B, Pyodide)
@tool
def run_python(code: str) -> str:
"""Execute Python in a sandboxed environment. Returns stdout."""
return sandbox.run(code, timeout=30)
agent = create_react_agent(claude_4_5, tools=[run_python])
agent.invoke({"messages": [("user",
"Load this CSV and find the customer with highest LTV.")]})`}Most engineers reach for fine-tuning too early. The decision tree is short: