Trusted by companies shaping the future of agents – including Klarna, Replit, Elastic, and more – LangGraph is a low-level orchestration framework for building, managing, and deploying long-running, stateful agents.
Install LangGraph:
pip install -U langgraph
Create a simple workflow:
from langgraph.graph import START, StateGraph
from typing_extensions import TypedDict
class State(TypedDict):
text: str
def node_a(state: State) -> dict:
return {"text": state["text"] + "a"}
def node_b(state: State) -> dict:
return {"text": state["text"] + "b"}
graph = StateGraph(State)
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")
print(graph.compile().invoke({"text": ""}))
# {'text': 'ab'}Get started with the LangGraph Quickstart.
To quickly build agents with LangChain's create_agent (built on LangGraph), see the LangChain Agents documentation.
LangGraph provides low-level supporting infrastructure for any long-running, stateful workflow or agent. LangGraph does not abstract prompts or architecture, and provides the following central benefits:
While LangGraph can be used standalone, it also integrates seamlessly with any LangChain product, giving developers a full suite of tools for building agents. To improve your LLM application development, pair LangGraph with:
[!NOTE] Looking for the JS version of LangGraph? See the JS repo and the JS docs.
LangGraph is inspired by Pregel and Apache Beam. The public interface draws inspiration from NetworkX. LangGraph is built by LangChain Inc, the creators of LangChain, but can be used without LangChain.
Raised when the graph has exhausted the maximum number of steps.
Raised when attempting to update a channel with an invalid set of updates.
Raised when a subgraph is interrupted, suppressed by the root graph.
Raised when graph receives an empty input.
Raised when the executor is unable to find a task (for distributed mode).
Read-only execution info/metadata for the execution of current thread/run/node.
Convenience class that bundles run-scoped context and other runtime utilities.
Payload for a task start event.
Payload for a task result event.
A task entry within a CheckpointPayload.
Payload for a checkpoint event.
Stream part emitted for stream_mode="values".
Stream part emitted for stream_mode="updates".
Stream part emitted for stream_mode="messages".
Stream part emitted for stream_mode="custom".
Stream part emitted for stream_mode="checkpoints".
Stream part emitted for stream_mode="tasks".
Stream part emitted for stream_mode="debug".
Typed container returned by invoke() / ainvoke() with version="v2".
Configuration for retrying nodes.
Configuration for caching nodes.
Information about an interrupt that occurred in a node.
A Pregel task.
Cache key for a task.
Snapshot of the state of the graph at the beginning of a step.
A message or packet to send to a specific node in the graph.
One or more commands to update the graph's state and send messages to nodes.
Bypass a reducer and write the wrapped value directly to a BinaryOperatorAggregate channel.
A LangGraph specific deprecation warning.
A specific LangGraphDeprecationWarning subclass defining functionality deprecated since LangGraph v0.5.0
A specific LangGraphDeprecationWarning subclass defining functionality deprecated since LangGraph v1.0.0
A specific LangGraphDeprecationWarning subclass defining functionality deprecated since LangGraph v1.1.0
A configurable PubSub Topic.
Base class for all channels.
Stores the last value received, can receive at most one value per step.
Stores the last value received, but only made available after finish().
Stores the value received in the step immediately preceding, clears after.
Stores the last value received, assumes that if multiple values are
A channel that waits until all named values are received before making the value available.
A channel that waits until all named values are received before making the value ready to be made available. It is only made available after finish() is called.
Stores the result of applying a binary operator to the current value and each new value.
Stores the last value received, never checkpointed.
A graph whose nodes communicate by reading and writing to a shared state.
A message type for UI updates in LangGraph.
A message type for removing UI components in LangGraph.
Define a LangGraph workflow using the entrypoint decorator.
A primitive that can be returned from an entrypoint.
Get the nonlocal variables accessed of a function.
Get nonlocal variables accessed.
Pregel manages the runtime behavior for LangGraph applications.
Implements the logic for sending writes to CONFIG_KEY_SEND.
A callback handler that implements stream_mode=messages.
Implements the logic for reading state from CONFIG_KEY_READ.
A node in a Pregel graph. This won't be invoked as a runnable by the graph
A context manager that runs sync tasks in the background.
A context manager that runs async tasks in the background.
Exception raised when an error occurs in the remote graph.
The RemoteGraph class is a client implementation for calling remote
Protocol for objects containing writes to be applied to checkpoint.
Simplest implementation of WritesProtocol, for usage with writes that
Responsible for executing a set of Pregel tasks concurrently, committing
Tracks which subgraphs have already loaded their pre-replay checkpoint.
Async unbounded FIFO queue with a wait() method.
Semaphore subclass with a wait() method.
Unbounded FIFO queue with a wait() method.
Protocol to represent types that behave like TypedDicts
Protocol to represent types that behave like TypedDicts
Protocol to represent types that behave like dataclasses.
TypedDict to use for extra keyword arguments, enabling type checking warnings for deprecated arguments.
A string enum.
A much simpler version of RunnableLambda that requires sync and async functions.
Sequence of Runnable, where the output of each is the input of the next.
Raised by a node to interrupt execution.
A StateGraph where every node receives a list of messages as input and returns one or more messages as output.
Get the runtime for the current graph run.
Access LangGraph store from inside a graph node or entrypoint task at runtime.
Access LangGraph StreamWriter from inside a graph node or entrypoint task at runtime.
Interrupt the graph with a resumable exception from within a node.
Push a new UI message to update the UI state.
Delete a UI message by ID from the UI state.
Merge two lists of UI messages, supporting removing UI messages.
Merges two lists of messages, updating existing messages by ID.
Write a message manually to the messages / messages-tuple stream mode.
Define a LangGraph task using the task decorator.
Get subset of current_versions that are newer than previous_versions.
Get the nonlocal variables accessed by a function.
Check if the given string matches the format of xxh3_128_hexdigest.
Return the module and name of an object.
Create a checkpoint for the given channels.
Get channels from a checkpoint.
Produce "task" events for stream_mode=debug.
Return True if the payload already wraps multiple writes from the same channel.
Folds task writes into a result dict and aggregates multiple writes to the same channel.
Produce "task_result" events for stream_mode=debug.
Remove pregel-specific keys from the config.
Produce "checkpoint" events for stream_mode=debug.
Apply writes / subgraph states to tasks to be returned in a StateSnapshot.
Get colored text.
Get bolded text.
A coroutine that waits for a semaphore before running another coroutine.
A function that yields control to other threads before running another function.
Map input chunk to a sequence of pending writes in the form (channel, value).
Map input chunk to a sequence of pending writes in the form (channel, value).
Map pending writes (a sequence of tuples (channel, value)) to output chunk.
Map pending writes (a sequence of tuples (channel, value)) to output chunk.
Run a task with retries.
Run a task asynchronously with retries.
Check if the graph should be interrupted based on current state.
Function injected under CONFIG_KEY_READ in task config, to read current state.
Default channel versioning function, increments the current int version.
Apply writes from a set of tasks (usually the tasks from a Pregel step)
Prepare the set of tasks that will make up the next Pregel step.
Prepares a single task for the next Pregel step, given a task path, which
Prepare a push task with an attached caller. Used for the functional API.
Get the null version for the checkpoint, if available.
Generate a string representation of the task path.
Pop any values belonging to UntrackedValue channels in Send.arg for safe checkpointing.
Get the graph for this Pregel instance.
Add an edge to the graph.
Default cache key function that uses the arguments and keyword arguments to generate a hashable key.
Determine the default value for a field in a state schema.
Attempt to extract default values and descriptions from provided type, used for config schema.
Get Pydantic state update as a list of (key, value) tuples.
Return cached annotated keys for a Python class.
Remove task IDs from checkpoint namespace.
Merge multiple configs into one.
Patch a config with new values.
Get a callback manager for a config.
Get an async callback manager for a config.
Return a config with all keys, merging any provided configs.
Submit a coroutine object to a given event loop.
Set the child Runnable config + tracing context.
Check if a function is async.
Check if a function is an async generator.
Coerce a runnable-like object into a Runnable.
Get the field names of a Pydantic model.
Create a pydantic model with the given field definitions.
Check if a given "complex" type is supported by pydantic.
Exports package version.
LangGraph specific warnings.
Legacy utilities module, to be removed in v1.
Backwards compat imports for config utilities, to be removed in v1.
Backwards compat imports for runnable utilities, to be removed in v1.
Re-export types moved to langgraph.types