At its core, LangGraph models agent workflows as graphs. One can define the behavior of their agents using three key components:

State: A shared data structure that represents the current snapshot of your application. It can be any data type but is typicall ydefined using a shared state schema (like a TypedDict).
Nodes: Functions that encode the logic of your agents. They receive the current state as input and perform some computation and return some updated state.
Edges: Functions that determine which Node to execute next based on the current state. They can be conditional branches or fixed transitions.

nodes do the work, edges tell what to do next.

You can initialise a graph using a StateGraph class. This is parametrized by a user-defined State object. You need to always .compile your graph before you can use it.
ex.

# showing a simple graph compilation step
agent_builder = StateGraph(MyState)
graph = agent_builder.compile()

State

This consists of the schema of the graph as well as reducer functions which specify who to apply updates to the state. The schema of the State will be input schema to all Nodes and Edges in the graph, and can be either a TypedDict or a Pydantic model. All Nodes will emit updates to the State which are then applied using the specified reducer function.

Schema

You will mostly be using a TypedDict here, or maybe a Pydantic BaseModel.
Example of how a schema is supposed to work and how nodes make changes:

from typing import TypedDict
from langgraph.graph import END, START, StateGraph
 
class InputState(TypedDict):
    user_input: str
 
class OutputState(TypedDict):
    graph_output: str
 
class OverallState(TypedDict):
    foo: str
    user_input: str
    graph_output: str
 
class PrivateState(TypedDict):
    bar: str
 
def node_1(state: InputState) -> OverallState:
    # Write to OverallState
    return {"foo": state["user_input"] + " name"}
 
def node_2(state: OverallState) -> PrivateState:
    # Read from OverallState, write to PrivateState
    return {"bar": state["foo"] + " is"}
 
def node_3(state: PrivateState) -> OutputState:
    # Read from PrivateState, write to OutputState
    return {"graph_output": state["bar"] + " Lance"}
 
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
graph = builder.compile()
graph.invoke({"user_input": "My"})
# {'graph_output': 'My name is Lance'}

Insights

We pass state: InputState in node_1, but we write out to foo, a channel in the OverallState. How can we write out to a state channel that is not included in the input schema? This is because a node can write to any state channel in the graph state. The graph state is theb union of the state channels defined at initialization, which includes OverallState, InputState and OutputState.
Also, we initialise the graph with
```
 StateGraph(
      OverallState,
      input_schema=InputState,
      output_schema=OutputState
  )
```
Then how can we write to PrivateState in the second node? Because nodes can declare additional state channels as long as the state schema definition exists.
n this case, the private state schema is defined, so we can add bar as a new state channel in the graph and write to it.

Reducers

Each key in our State has its own independent reducer function. If no reducer function is explicitly specified, LangGraph assumes that all updates should override it.
Every reducer is a binary function with two positional arguments:

left argument: the current value already stored in state for that key
right argument: update for that key returned by a node
When a node returns a partial update, LangGraph calls the reducer for each updated key and saves the return value in the new state value.

new_value = reducer(left=current_state[key], right=node_update[key])

ex. of a reducer

def append_strings(left: list[str], right: list[str]) -> list[str]:
    """Combine the existing state value (left) with a node update (right)."""
    return left + right
 
class State(TypedDict):
    tags: Annotated[list[str], append_strings]

You can use Annotated for ex. to make an attribute annotated inside your state. For ex. tags: Annotated[list[str], append_strings] ensures that if the state is {"tags": ["draft"]} for ex. and a node returns {"tags": ["review"]} LangGraph calls

append_strings(left=["draft"], right=["review"])  # returns ["draft", "review"]. The new state value is {"tags": ["draft", "review"]}.

You will also commonly see operator.add being the second argument in an annotated type.
In some cases, you may want to bypass a reducer and directly overwrite a state value using Overwrite.

Since having a list of messages in your state is very common, there exists a prebuilt state called MessagesState. It’s defined with a single messages key which is a list of AnyMessage objects and uses the add_messages reducer.

from langgraph.graph import MessagesState
 
class State(MessagesState):
    documents: list[str]

Nodes

A node is a python function that accepts the following arguments:

state: the state of the graph
config: a RunnableConfig object that contains configuration information like thread_id and tracing information like tags.
runtime: a Runtime object that contains runtime context and other information like store, stream_writer, etc. See the docs for more details.

You can add nodes to graphs using the add_node method.
ex.

def node_with_runtime(state: State, runtime: Runtime[Context]):
    print("In node: ", runtime.context.user_id)
    return {"results": f"Hello, {state['input']}!"}
 
builder.add_node("plain_node", plain_node)

The first argument allows you to give names to your nodes.

You can use tasks if a node contains multiple operations, instead of splitting the logic across multiple nodes.
ex.

@task
def _make_request(url: str):
    """Make a request."""
    return requests.get(url).text[:100]

Special nodes:

START: it represents the node that sends user input into the graph.

from langgraph.graph import START
graph.add_edge(START, "node_a")

END: it represents the terminal node.

from langgraph.graph import END
graph.add_edge("node_a", END)

As I am utilising JIT learning I won’t be able to talk about everything here, but if you want to look up node caching feel free to do so.

Edges

Edges define how logic is routed and how the graph decides to stop. There are a few types of these:

Normal edges: go directly from one node to the next
Conditional edges: call a function to determine which node(s) to go to next
Entry point: which node to call first when user input arrives
Conditional entry point: call a function to determine which node(s) to call first when user input arrivesA

Normal edges

If you always want to go from node A to node B, you can use normal edges.

graph.add_edge("node_a", "node_b")

Conditional edges

If you want to optionally route to one or more edges, or optionally terminate, you can use the add_conditional_edges method. This method accepts the name of a node and a “routing function” to call after that node is executed.

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})

Similar to nodes, this routing_function accepts the current state of the graph and returns a value.

Entrypoint

The first node(s) that are run when the graph starts. You can use the add_edge method from the START node.

from langgraph.graph import START
graph.add_edge(START, "node_a")

Conditional entry point

A conditinoal entry point lets you start at different nodes depending on custom logic.

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})

Misc.

Controlling graph execution using `Command`

We can use a Command object for controlling graph execution. It accepts four parameters:

update: apply state updates (similar to returning updates from a node)
goto: navigate to specific nodes (similar to conditional edges)
graph: target a parent graph while navigating from subgraphs
resume: provide a value to resume execution after an interrupt

It’s used in three contexts:

Return from nodes: use update, goto and graph to combine state updates with control flow
Input to invoke or stream: use resume to continue execution after an interrupt
Return from tools: similar to return from nodes, combine state updates and control flow inside a tool

Return from nodes
update and goto
Return Command objects from node functions to update state and route to the next node in a single step.

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

With Command you can also achieve dunamic control flow behavior, identical to conditional edges:

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

Use Command when you need to both update state and route to a different node. If you only need to route without updating state, use conditional edges instead.
An end-to-end example:

import random
from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START
from langgraph.types import Command
 
# Define graph state
class State(TypedDict):
    foo: str
 
# Define the nodes
def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
    print("Called A")
    value = random.choice(["b", "c"])
    # this is a replacement for a conditional edge function
    if value == "b":
        goto = "node_b"
    else:
        goto = "node_c"
 
    # note how Command allows you to BOTH update the graph state AND route to the next node
    return Command(
        # this is the state update
        update={"foo": value},
        # this is a replacement for an edge
        goto=goto,
    )
 
def node_b(state: State):
    print("Called B")
    return {"foo": state["foo"] + "b"}
 
def node_c(state: State):
    print("Called C")
    return {"foo": state["foo"] + "c"}
 
builder = StateGraph(State)
builder.add_edge(START, "node_a")
builder.add_node(node_a)
builder.add_node(node_b)
builder.add_node(node_c)
# NOTE: there are no edges between nodes A, B and C!
 
graph = builder.compile()
result = graph.invoke({"foo": "apple"})

graph
If you are using subgraphs, you can navigate from a ndoe within a subgraph to a different node in the parent graph by specifying graph=Command.PARENT in Command.

def my_node(state: State) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT
    )

Input to invoke or stream
Command(resume=) is the only Command pattern intended as input to invoke()/stream(). Don’t use Command(update=) alone as input to continue with multi turn conversations, because passing any Command as input resumes from the latest checkpoint (i.e. the last step that ran, not __start__). To continue a conversation, pass a raw input dict:

graph.invoke( {
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

resume
Use Command(resume=) to provide a value and resume graph execution after an interrupt. The value passed to resume becomes the return value of the interrupt() call inside the paused nodes:

from typing import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt
 
 
class State(TypedDict):
    messages: list[dict]
 
 
def human_review(state: State):
    # Pauses the graph and waits for a value
    answer = interrupt("Do you approve?")
    return {"messages": [{"role": "user", "content": answer}]}
 
 
graph = (
    StateGraph(State)
    .add_node("human_review", human_review)
    .add_edge(START, "human_review")
    .add_edge("human_review", END)
    .compile(checkpointer=InMemorySaver())
)
 
config = {"configurable": {"thread_id": "graph-api-resume"}}
 
# First run - hits the interrupt and pauses
stream = graph.stream_events({"messages": []}, config, version="v3")
_ = stream.output  # drive the stream to completion
print(stream.interrupts)
 
# Resume with a value - the interrupt() call returns "yes"
resumed = graph.stream_events(Command(resume="yes"), config, version="v3")
final = resumed.output

Return from tools
You can return Command from tools to update graph state and control flow. Use update to modify state (ex. saving customer information looked up during a conversation) and goto to route to a specific node after the tool completes.

Subgraphs

A subgraph is a graph that is used as a node in another graph. These are useful for building multi-agent systems, reusing a set of nodes in multiple graphs, distributing development, etc.

When adding subgraphs, you need to define how the parent graph and the subgraph communicate. Two patterns exist here:

Calling a subgraph inside a node

Use this when parent and subgraph have different state schemas (no shared keys), or you need to transform state between them. You can invoke the subgraph inside a node function. This is common when you want to keep a private message history for each agent in a multi-agent system. The node function transforms the parent state to the subgraph state before invoking the subgraph, and transforms the results back to the parent state before returning.

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
 
class SubgraphState(TypedDict):
    bar: str
 
# Subgraph
def subgraph_node_1(state: SubgraphState):
    return {"bar": "hi! " + state["bar"]}
 
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
 
# Parent graph
class State(TypedDict):
    foo: str
 
def call_subgraph(state: State):
    # This is a wrapper function
    # Transform the state to the subgraph state
    subgraph_output = subgraph.invoke({"bar": state["foo"]})
    # Transform response back to the parent state
    return {"foo": subgraph_output["bar"]}
 
builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()

Adding a subgraph as a node

When the parent graph and the subgraph share state keys, you can pass a compiled subgraph directly to add_node. No wrapper function is needed, the subgraph reads from and writes to the parent’s state channels automatically. For example, in multi-agent systems, the agents often communicate over a shared messages key.

If your subgraph shares state keys with the parent graph, you can follow these steps to add it to your graph.

Define subgraph workflow (subgraph_builder in the example below) and compile it.
Pass compiled subgraph to the add_node method when defining the parent graph workflow.

from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
 
class State(TypedDict):
    foo: str
 
# Subgraph
 
def subgraph_node_1(state: State):
    return {"foo": "hi! " + state["foo"]}
 
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
 
# Parent graph
 
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()

For HITL, interrupts, or MemorySaver, please read about persistence.

Structured output

This allows agents to return data in a specific, predictable format. Instead of parsing natural language responses, you can get structured data in the form of Pydantic models, JSON objects, or dataclasses. You can use LangChain’s with_structured_output for this.
ex.

from pydantic import BaseModel
 
class AnswerWithJustification(BaseModel):
    '''An answer to the user question along with justification for the answer.'''
    answer: str
    justification: str
 
model = ChatModel(model="model-name", temperature=0)
structured_model = model.with_structured_output(AnswerWithJustification)
 
structured_model.invoke(
    "What weighs more a pound of bricks or a pound of feathers"
)
 
# -> AnswerWithJustification(
#     answer='They weigh the same',
#     justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )

LangChains’ create_agent handles sturctured output automatically using a response_format argument. The user sets their desired structured output and when the model generates the structured data, it’s captured, validated and returned in the 'structured_response' key of the agent’s state.

It seems that more recently, this has been the “idiomatic” way to get structured output in LangChain.

Passing state between graphs

This is an interesting problem I stumbled upon. My parent state was essentially taking some input from the user and appending it to its own messages list. But I wanted to pass down some part of this input to a child subgraph. How could I do that? I have noted that two approaches exist here:

Shared state keys: if you make a shared state key in both the internal state of the child subgraph and the state of the parent graph, this state key would be shared among the two graphs.
ex.

## Vision/Image Subgraph
class VisionState(TypedDict):
    """State schema for the vision graph"""
    file_path: str # shared state key with parent
    extracted_info: str
    messages: Annotated[list[AnyMessage], operator.add]  # shared state key with parent
# ...
## The parent graph
class ParentState(TypedDict):
    """State for the parent graph"""
    messages: Annotated[list[AnyMessage], operator.add]
    file_path: Optional[str]

Here, messages and file_path are shared state keys between the two graphs. Now this approach is okay if you want to keep the child subgraph’s node separate from the parent graph, but another approach exists if you want to call the child subgraph entirely.

Calling a subgraph inside a node: Recall that two patterns exist for subgraphs - calling inside a node, or adding them as a node. For sharing state, you can simply keep the subgraph schema as it as and call it in a wrapper function inside the parent supervisor node. An example:

## Vision/Image Subgraph
class VisionState(TypedDict):
    """State schema for the vision graph"""
    # none of these are shared state keys
    file_path: str
    extracted_info: str
    vision_messages: Annotated[list[AnyMessage], operator.add]
# ...
## The parent graph
class ParentState(TypedDict):
    """State for the parent graph"""
    messages: Annotated[list[AnyMessage], operator.add]
 
def supervisor_node(state: ParentState):
    """Supervisor node which decides whether to route to coding subgraph or the vision subgraph based on the user's prompt"""
    latest_message = state["messages"][-1]
    model = MODEL.with_structured_output(SupervisorModel)
    result = model.invoke(input=f"Based on this message: {latest_message} just reply using one word for action - either Coding or Vision based on whether the task is related to coding or vision operations. If it's related to neither, just put None in action. If it's a task related to documents (i.e. vision), make sure that the file_path has the correct file path. Example of coding tasks: anything where code has to be written. Example of vision tasks: document parsing, ex. extracting entitites from a document, document path, etc.")
    print(f"[{supervisor_node.__name__}]: {result}")
    if result.action == "Coding":
        return Command(update={"messages": [SystemMessage(content="routing to coding model...")]}, goto="coding_subgraph_node")
 
    if result.action == "Vision":
        subgraph_output = vision_subgraph.invoke({"file_path": result.file_path, "vision_messages": state["messages"]})
        # Transform response back to the parent state
        return {"messages": subgraph_output["vision_messages"]}

Here we call the subgraph inside the supervisor_node and update the parent state with the subgraph’s output.

To read

Stuff I need to read to complete those drills
https://reference.langchain.com/python/langgraph/types/interrupt
https://docs.langchain.com/oss/python/langgraph/persistence#persistence

Hyperreality

Explorer

LangGraph

State

Schema

Reducers

Nodes

Edges

Normal edges

Conditional edges

Entrypoint

Conditional entry point

Misc.

Controlling graph execution using `Command`

Subgraphs

Calling a subgraph inside a node

Adding a subgraph as a node

Structured output

Passing state between graphs

To read

Graph View

Table of Contents

Hyperreality

Explorer

LangGraph

State

Schema

Reducers

Nodes

Edges

Normal edges

Conditional edges

Entrypoint

Conditional entry point

Misc.

Controlling graph execution using Command

Subgraphs

Calling a subgraph inside a node

Adding a subgraph as a node

Structured output

Passing state between graphs

To read

Graph View

Table of Contents

Controlling graph execution using `Command`