Many job descriptions for AI engineers name LangChain or LangGraph, and interviewers often hand you a short snippet and ask what each line does. You also review code like this at work, written by a teammate or by an AI assistant. The questions that separate candidates are specific: what a reducer does, why the loop ends, what a checkpointer saves, and what happens when a paused run resumes. If you can answer those on a 20-line graph, you can pick up any other framework quickly, because they all wrap the same loop.

You do not need to write this code from memory, but you must be able to read it and say what each line does. LangChain gives you a ready-made agent. LangGraph is the engine underneath, where the agent is drawn as a graph.

Frameworks to recognise

AutoGen itself is now effectively in maintenance mode, so say "Microsoft Agent Framework" if asked about it.

To choose, name the one constraint that dominates your project and take the framework built around it. Then check that it covers what every production agent needs: saved state, a pause for approval, step limits and tracing. If you need none of these, plain SDK code is the right choice.

How the LangChain products fit together

LangChain's agents are built on LangGraph, so a LangChain agent is a LangGraph graph inside.

LangChain words you will see in code

• Chat model: one interface for every provider. Changing provider is changing a string.
• Messages: system, user, AI and tool messages. A conversation is a list of these.
• Tool: a normal Python function. Its name, docstring and type hints become the schema the model sees.
• create_agent: takes a model, tools and a system prompt, and returns a ready agent with the tool-calling loop built in.
• Middleware: hooks that run around model and tool calls. Guardrails, retries, summarising and approval steps are added this way.
• Structured output: asking the model to return a typed object, not free text.
• Loader, splitter, embeddings, vector store, retriever: the RAG pieces from the Vector DBs and RAG lesson.
• invoke, stream, batch: run once, stream the output as it is produced, or run many inputs.
• Pipe syntax (prompt | model | parser): the older chain style, still common in existing codebases. Each piece passes its output to the next.

Reading a LangChain agent

# needs: langchain
from langchain.agents import create_agent

def get_order_status(order_id: str) -> str:
    """Look up the delivery status of an order."""
    return lookup(order_id)

agent = create_agent(
    model="provider:model-name",
    tools=[get_order_status],
    system_prompt="You are a support assistant.",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Where is order 123?"}]}
)

1. The function is the tool. The docstring tells the model when to use it; the type hints define its arguments.
2. create_agent wires the model, the tools and the prompt into a loop.
3. invoke sends the user message. The agent calls the model, runs any tool the model asks for, feeds the result back, and repeats until the model answers without a tool call.
4. result["messages"] holds the whole exchange, including tool calls. The last message is the answer.

LangGraph words you will see in code

• State: a shared, typed dictionary. Every node reads it and returns only the part it changes.
• Reducer: the rule for merging an update into a state key. The default overwrites. add_messages appends, which is why chat history grows.
• Node: a Python function that takes the state and returns an update. It can be an LLM call or plain code.
• Edge: a fixed rule, "after A, go to B".
• Conditional edge: a function that looks at the state and returns the name of the next node. Branches and loops come from here.
• START and END: the entry and exit points.
• compile: turns the definition into something you can run. The checkpointer is attached here.
• Checkpointer: saves the state after every step, per thread_id. It gives conversation memory, recovery after a crash, human-in-the-loop and replay. In-memory for development; SQLite or Postgres for production. The LangGraph docs suggest SQLite for local workflows and Postgres for a production service.
• interrupt: pauses inside a node and waits for a person. You resume by invoking the graph with Command(resume=...). It needs a checkpointer.
• Command: lets a node return a state update and the next node together.
• Send: starts many parallel runs of a node, one per item. Reducers merge the results. This is map-reduce.
• Subgraph: a graph used as a node inside another graph. Multi-agent systems are built this way.
• ToolNode: a prebuilt node that executes the tool calls in the last AI message.
• Store: long-term memory shared across threads. The checkpointer is memory within one thread.

The agent loop as a graph

The LangGraph agent loop from START through the model node, the tool calls decision and the tools node to END, with a checkpointer saving the state after every step

The loop is START, model node, a "tool calls?" decision, then either END or the tools node, whose results join the state before the model runs again. A checkpointer saves the state after every step, keyed by thread_id.

The conditional edge after the model node is the whole agent: tool calls send the run to the tools node and back, and no tool calls send it to END.

Reading a LangGraph agent

# needs: langgraph
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
    messages: Annotated[list, add_messages]

def call_model(state: State):
    reply = model_with_tools.invoke(state["messages"])
    return {"messages": [reply]}

def route(state: State):
    last = state["messages"][-1]
    return "tools" if last.tool_calls else END

builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route)
builder.add_edge("tools", "model")

graph = builder.compile(checkpointer=InMemorySaver())
graph.invoke(
    {"messages": [{"role": "user", "content": "Where is order 123?"}]},
    config={"configurable": {"thread_id": "user-42"}},
)

1. State: one key, messages. add_messages means new messages are appended, not overwritten. Annotated[list, add_messages] is how a reducer is attached to a key.
2. call_model: sends all messages to the LLM and returns its reply as the update. model_with_tools is defined outside the snippet: it is the chat model with the tool schemas attached, usually by model.bind_tools(tools), so the model knows which tools it may ask for.
3. route: if the reply asks for a tool, go to tools. Otherwise finish.
4. add_node: registers the two nodes. ToolNode runs the requested tools and adds their results as tool messages.
5. Edges: START to model; model to wherever route says; tools back to model. That last edge creates the loop.
6. compile with a checkpointer: the state is saved after every step.
7. invoke with a thread_id: calling again with the same id continues the same conversation.

If the model never stops asking for tools, the recursion limit ends the run. It caps the steps of one run, raises GraphRecursionError when reached, and is set per call with config={"recursion_limit": 25}.

What a reducer does: a two-node example

Take a state with one key, results, and two nodes. Node A returns {"results": ["a"]}. Node B returns {"results": ["b"]}.

# needs: langgraph
import operator
from typing import Annotated, TypedDict

class State(TypedDict):
    results: Annotated[list, operator.add]   # reducer: old + new
    status: str                               # no reducer: the new value replaces the old one

add_messages is a reducer written for chat history. It appends new messages, and when an update carries the id of an existing message it replaces that message.

What a checkpointer stores per thread

A thread is the chain of checkpoints saved under one thread_id. Each checkpoint is a snapshot with these parts:

graph.get_state(config) returns the latest snapshot of a thread and graph.get_state_history(config) returns all of them, newest first. Replaying from an older checkpoint skips the nodes before it and runs the later ones again, LLM calls included.

Reading a human approval step

# needs: langgraph
from langgraph.types import interrupt, Command

def approve_refund(state: State):
    decision = interrupt({"refund": state["amount"]})
    return {"approved": decision == "yes"}

graph.invoke(Command(resume="yes"), config)

In this snippet State also has the keys amount and approved, and config is the same thread_id config that started the run.

interrupt saves the state and stops the run. Hours later, the second call resumes it with the person's answer. On resume the node runs again from its first line, so anything before the interrupt must be safe to repeat.

How the resume works, step by step:

1. The node calls interrupt(payload). LangGraph saves a checkpoint and the first invoke returns, with the payload under result["__interrupt__"].
2. Your application shows the payload to a person. Nothing is running while it waits.
3. The application calls invoke with Command(resume="yes") and the same thread_id.
4. LangGraph loads the checkpoint and restarts the node. This time interrupt returns "yes" and the node finishes.

Two rules follow. Put side effects after the interrupt, never before it. Do not wrap interrupt in a bare try/except, because the pause is raised as an exception.

When someone else, or an AI, writes the code: five things to check

1. State and reducers: which keys exist, and which append instead of overwrite?
2. Exits: where can the loop end, and what stops it if the model keeps calling tools?
3. Checkpointer: is one attached, and is it a production one, not in-memory?
4. Tools: does each have a clear docstring and typed arguments?
5. Side effects: is there an approval step before anything that spends money or cannot be undone?

Tracing one thread. Run the graph above on "Where is order 123?" with thread_id "user-42". The input is merged first, so the state starts with 1 message.

route ran twice. After step 1 the last message had a tool call, so it returned "tools". After step 3 it had none, so it returned END.

The last column counts one checkpoint per node step, as the small engine below does. Real LangGraph also saves two at the start of a run, an empty one and one that holds the input, so get_state_history returns 5 snapshots after this first call.

Second call, same thread. The user sends "Thanks!" with thread_id "user-42". The engine loads checkpoint 3 (4 messages), appends the new message (5), and runs the model once (6 messages, next is END). The thread now holds 4 checkpoints. A call with thread_id "user-7" starts from an empty state and ends with 2 messages: threads share nothing.

The same run without the reducer. Remove add_messages, so messages is overwritten. Step 1 replaces the user's question with the AI message: 1 message. Step 2 replaces that with the tool result: 1 message. At step 3 the model sees only "out for delivery" and has no idea what was asked. One missing annotation breaks the agent, and nothing raises an error.

The loop that never ends. If route always returns "tools", the run goes model, tools, model, tools. With a recursion limit of 4 it is stopped after step 4, with 4 checkpoints saved that show where it was going wrong.

The code example below is this exact run on a small engine written from scratch, and it prints the same numbers.

"""A tiny state-graph engine in the style of LangGraph, standard library only.

Nodes read the state and return an update. A reducer merges each update. A conditional edge picks
the next node. A checkpointer saves the state after every step, keyed by thread id. No network.
"""
import copy

START, END = "__start__", "__end__"


class StateGraph:
    def __init__(self, reducers):
        self.reducers = reducers                      # key -> merge rule; other keys are overwritten
        self.nodes, self.edges, self.routers = {}, {}, {}

    def add_node(self, name, fn):
        self.nodes[name] = fn

    def add_edge(self, src, dst):                     # fixed: after src, always go to dst
        self.edges[src] = dst

    def add_conditional_edges(self, src, router):     # router(state) returns the next node's name
        self.routers[src] = router

    def merge(self, state, update):
        for key, value in update.items():
            reducer = self.reducers.get(key)
            state[key] = reducer(state[key], value) if reducer and key in state else value
        return state

    def invoke(self, update, thread_id, checkpointer, recursion_limit=10):
        history = checkpointer.setdefault(thread_id, [])              # one checkpoint list per thread
        state = copy.deepcopy(history[-1]["state"]) if history else {}
        state = self.merge(state, update)                             # the input is merged like any update
        node, step = self.edges[START], 0
        while node != END:
            step += 1
            if step > recursion_limit:
                raise RecursionError(f"no END after {recursion_limit} steps (recursion limit)")
            state = self.merge(state, self.nodes[node](state))        # run the node, merge what it returns
            nxt = self.routers[node](state) if node in self.routers else self.edges[node]
            history.append({"node": node, "next": nxt, "state": copy.deepcopy(state)})   # checkpoint
            last = state["messages"][-1]
            print(f"  step {step}: ran {node:5} next {nxt:7} | {len(state['messages'])} messages, "
                  f"model_calls={state['model_calls']} | last = {last['role']}: {last.get('tool_calls') or last['content']}")
            node = nxt
        return state


def add_messages(old, new):                           # reducer: append, do not overwrite
    return old + new


def call_model(state):                                # scripted fake model: decides from the last message
    last = state["messages"][-1]
    if last["role"] == "user" and "order" in last["content"]:
        reply = {"role": "ai", "content": "", "tool_calls": [{"name": "get_order_status", "args": {"order_id": "123"}}]}
    elif last["role"] == "tool":
        reply = {"role": "ai", "content": f"Order 123 is {last['content']}."}
    else:
        reply = {"role": "ai", "content": "You are welcome."}
    return {"messages": [reply], "model_calls": state.get("model_calls", 0) + 1}


def run_tools(state):                                 # like ToolNode: run every call in the last AI message
    tools = {"get_order_status": lambda order_id: "out for delivery"}
    calls = state["messages"][-1].get("tool_calls", [])
    return {"messages": [{"role": "tool", "content": tools[c["name"]](**c["args"])} for c in calls]}


def route(state):                                     # the conditional edge: tools, or finish
    return "tools" if state["messages"][-1].get("tool_calls") else END


def build(router):
    g = StateGraph(reducers={"messages": add_messages})   # model_calls has no reducer: it is overwritten
    g.add_node("model", call_model)
    g.add_node("tools", run_tools)
    g.add_edge(START, "model")
    g.add_conditional_edges("model", router)
    g.add_edge("tools", "model")
    return g


if __name__ == "__main__":
    saver, graph = {}, build(route)                   # saver: in-memory checkpointer, thread_id -> checkpoints
    for thread, text in [("user-42", "Where is order 123?"), ("user-42", "Thanks!"), ("user-7", "Hello")]:
        print(f"thread {thread}: {text}")
        out = graph.invoke({"messages": [{"role": "user", "content": text}]}, thread, saver)
    assert len(saver["user-42"][-1]["state"]["messages"]) == 6 and len(out["messages"]) == 2
    print("checkpoints per thread:", {t: len(h) for t, h in saver.items()})
    try:                                              # a broken router that never returns END
        build(lambda state: "tools").invoke({"messages": [{"role": "user", "content": "Hi"}]}, "user-9", saver, 4)
    except RecursionError as exc:
        print("stopped:", exc)