LangGraph Graph API components
A LangGraph graph consists of three core components: State (shared data that nodes read and update), Nodes (functions that take the current state, run a step like calling a model or tool, and return state updates), and Edges (connections that define which node runs next, including conditional edges that branch based on state).
Migrating from Graph to Functional API
When a graph becomes overly complex for simple linear processes, you can simplify by migrating to the Functional API. This works well when the workflow is primarily sequential with minimal branching, reducing the overhead of defining state schemas and graph structures.
Functional API entrypoint decorator
The @entrypoint decorator in the Functional API marks the main workflow function. It accepts parameters like checkpointer and enables checkpointing and other LangGraph persistence features for the entire workflow defined in standard Python/TypeScript control flow.
Graph API for complex decision trees
The Graph API uses a declarative approach where you define nodes, edges, and shared state to create a visual graph structure. When your workflow has multiple decision points that depend on various conditions, the Graph API makes these branches explicit and easy to visualize using conditional edges.
Both APIs share core LangGraph features
Both the Graph API and Functional API provide the same core LangGraph features including persistence, streaming, human-in-the-loop, and memory, but package them in different paradigms to suit different development styles and use cases.
Combining Graph and Functional APIs
Both APIs can be used together in the same application. This is useful when different parts of your system have different requirements. For example, you can use the Graph API for complex multi-agent coordination and the Functional API for simple data processing, then call the functional API from within a graph node.
Functional API linear workflows with simple logic
The Functional API is suitable for workflows that are primarily sequential with straightforward conditional logic. Tasks are called using .result() in Python or await in TypeScript, with if/else statements for branching, and the interrupt() function can be used for human-in-the-loop checkpoints.
Functional API example with entrypoint in TypeScript
Example showing Functional API with entrypoint in TypeScript:
```typescript
import { task, entrypoint } from "@langchain/langgraph";
const processUserInput = task(
"processUserInput",
async (userInput: string) => {
return { processed: userInput.toLowerCase().trim() };
}
);
const workflow = entrypoint(
{ checkpointer },
async (userInput: string) => {
const processed = await processUserInput(userInput);
let response: string;
if (processed.processed.includes("urgent")) {
response = await handleUrgentRequest(processed);
} else {
response = await handleNormalRequest(processed);
}
return response;
}
);
```
This shows how the Functional API in TypeScript uses task() and entrypoint() functions with async/await.
Graph API vs Functional API overview
LangGraph provides two APIs for building agent workflows: the Graph API and the Functional API. Both share the same underlying runtime and can be used together in the same application, but they are designed for different use cases and development preferences.
When to use Graph API
Use the Graph API when you need: complex workflow visualization for debugging and documentation, explicit state management with shared data across multiple nodes, conditional branching with multiple decision points, parallel execution paths that need to merge later, or team collaboration where visual representation aids understanding.
Graph API team development example in Python
Example showing Graph API with clear separation of concerns for team development:
```python
workflow.add_node("data_ingestion", data_team_function)
workflow.add_node("ml_processing", ml_team_function)
workflow.add_node("business_logic", product_team_function)
workflow.add_node("output_formatting", frontend_team_function)
```
This shows how the Graph API's visual structure with named nodes allows different team members to work on different components independently.
Combining Graph and Functional APIs example in Python
Example showing how to use both APIs together in Python:
```python
from langgraph.graph import StateGraph
from langgraph.func import entrypoint
# Complex multi-agent coordination using Graph API
coordination_graph = StateGraph(CoordinationState)
coordination_graph.add_node("orchestrator", orchestrator_node)
coordination_graph.add_node("agent_a", agent_a_node)
coordination_graph.add_node("agent_b", agent_b_node)
# Simple data processing using Functional API
@entrypoint()
def data_processor(raw_data: dict) -> dict:
cleaned = clean_data(raw_data).result()
transformed = transform_data(cleaned).result()
return transformed
# Use the functional API result in the graph
def orchestrator_node(state):
processed_data = data_processor.invoke(state["raw_data"])
return {"processed_data": processed_data}
```
This shows calling a Functional API workflow from within a Graph API node using .invoke().
Migrating from Functional to Graph API
When a functional workflow grows complex with multiple decision points and conditional branches, you can migrate to the Graph API by converting tasks to nodes, replacing if/else conditionals with conditional_edges, and defining a TypedDict or StateSchema for state management.
Functional API task decorator
The @task decorator in the Functional API marks a function as a task that can be awaited or used with .result(). It enables LangGraph features like persistence and checkpointing on individual functions within a larger workflow.
Functional API example with entrypoint in Python
Example showing Functional API with entrypoint in Python:
```python
from langgraph.func import entrypoint, task
@task
def process_user_input(user_input: str) -> dict:
return {"processed": user_input.lower().strip()}
@entrypoint(checkpointer=checkpointer)
def workflow(user_input: str) -> str:
processed = process_user_input(user_input).result()
if "urgent" in processed["processed"]:
response = handle_urgent_request(processed).result()
else:
response = handle_normal_request(processed).result()
return response
```
This shows how the Functional API uses @task and @entrypoint decorators to wrap standard Python procedural code.
Graph API parallel processing example in Python
Example showing Graph API parallel processing in Python:
```python
from langgraph.graph import StateGraph, START
workflow.add_node("fetch_news", fetch_news)
workflow.add_node("fetch_weather", fetch_weather)
workflow.add_node("fetch_stocks", fetch_stocks)
workflow.add_node("combine_data", combine_all_data)
# All fetch operations run in parallel
workflow.add_edge(START, "fetch_news")
workflow.add_edge(START, "fetch_weather")
workflow.add_edge(START, "fetch_stocks")
# Combine waits for all parallel operations to complete
workflow.add_edge("fetch_news", "combine_data")
workflow.add_edge("fetch_weather", "combine_data")
workflow.add_edge("fetch_stocks", "combine_data")
```
This shows how multiple edges from START and convergence at a combine node enables parallel execution.
Functional API with human-in-the-loop using interrupt
Example showing Functional API with human review checkpoint in Python:
```python
from langgraph.func import entrypoint, interrupt
@entrypoint(checkpointer=checkpointer)
def essay_workflow(topic: str) -> dict:
outline = create_outline(topic).result()
if len(outline["points"]) < 3:
outline = expand_outline(outline).result()
draft = write_draft(outline).result()
# Human review checkpoint
feedback = interrupt({"draft": draft, "action": "Please review"})
if feedback == "approve":
final_essay = draft
else:
final_essay = revise_essay(draft, feedback).result()
return {"essay": final_essay}
```
This shows how interrupt() pauses workflow execution and returns data for human review, then resumes with the human's response.
When to use Functional API
Use the Functional API when you want: minimal code changes to existing procedural code, standard control flow (if/else, loops, function calls), function-scoped state without explicit state management, rapid prototyping with less boilerplate, or linear workflows with simple branching logic.
Graph API example with conditional edges in Python
Example showing Graph API with conditional branching in Python:
```python
from langgraph.graph import StateGraph
from typing import TypedDict
class AgentState(TypedDict):
messages: list
current_tool: str
retry_count: int
def should_continue(state):
if state["retry_count"] > 3:
return "end"
elif state["current_tool"] == "search":
return "process_search"
else:
return "call_llm"
workflow = StateGraph(AgentState)
workflow.add_node("call_llm", call_llm_node)
workflow.add_node("process_search", search_node)
workflow.add_conditional_edges("call_llm", should_continue)
```
This shows how to define state with TypedDict, create nodes, and add conditional edges that route based on state values.
Graph API example with conditional edges in TypeScript
Example showing Graph API with conditional branching in TypeScript:
```typescript
import * as z from "zod";
import {
StateGraph,
StateSchema,
MessagesValue,
START,
END,
type GraphNode,
type ConditionalEdgeRouter,
} from "@langchain/langgraph";
const AgentState = new StateSchema({
messages: MessagesValue,
currentTool: z.string(),
retryCount: z.number().default(0),
});
const shouldContinue: ConditionalEdgeRouter<typeof AgentState> = (state) => {
if (state.retryCount > 3) {
return END;
} else if (state.currentTool === "search") {
return "processSearch";
} else {
return "callLlm";
}
};
const workflow = new StateGraph(AgentState)
.addNode("callLlm", callLlmNode)
.addNode("processSearch", searchNode)
.addConditionalEdges("callLlm", shouldContinue);
```
This shows how to define state with StateSchema, create nodes, and add conditional edges using ConditionalEdgeRouter.
Functional API for existing procedural code
The Functional API uses an imperative approach that integrates LangGraph features into standard procedural code. When you have existing code that uses standard control flow and want to add LangGraph features with minimal refactoring, the Functional API enables this with decorators like @task and @entrypoint.
Resuming execution with Command primitive
Resuming an execution after an interrupt can be done by passing a resume value to the Command primitive. The config must contain the same thread_id as the paused execution. Methods available are invoke with Command(resume=some_resume_value), ainvoke, stream_events, and astream_events.
Injectable parameters available in entrypoint
When declaring an entrypoint, you can request access to additional parameters that will be injected automatically at runtime: 'previous' provides access to the state associated with the previous checkpoint for the given thread (useful for short-term memory); 'store' provides an instance of BaseStore for long-term memory; 'writer' provides access to the StreamWriter when working with Async Python < 3.11; 'config' provides access to run time configuration (RunnableConfig). Parameters must be declared with appropriate name and type annotation.
Entrypoint execution methods
Using the @entrypoint decorator yields a Pregel object that can be executed using the invoke, ainvoke, stream, and astream methods. All execution methods require a config dictionary with configurable thread_id. invoke and ainvoke wait for or await the result synchronously or asynchronously. stream and astream_events return an event stream that can be iterated over.
Functional API vs Graph API control flow differences
The Functional API does not require thinking about graph structure and allows using standard Python constructs to define workflows, usually trimming the amount of code needed. The Graph API requires declaring a State and may require defining reducers to manage updates to the graph state. @entrypoint and @tasks do not require explicit state management as their state is scoped to the function and is not shared across functions.
Visualization support in Functional API vs Graph API
The Graph API makes it easy to visualize the workflow as a graph which can be useful for debugging, understanding the workflow, and sharing with others. The Functional API does not support visualization as the graph is dynamically generated during runtime.
Entrypoint serialization requirement
The inputs and outputs of entrypoints must be JSON-serializable to support checkpointing. Use Python primitives like dictionaries, lists, strings, numbers, and booleans to ensure inputs and outputs are serializable.
Non-deterministic control flow pitfall example
Incorrect example using current time for control flow:
```python
@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
t0 = inputs["t0"]
t1 = time.time() # Non-deterministic
delta_t = t1 - t0
if delta_t > 1:
result = slow_task(1).result()
value = interrupt("question")
else:
result = slow_task(2).result()
value = interrupt("question")
return {"result": result, "value": value}
```
Correct approach using a task to capture time:
```python
@task
def get_time() -> float:
return time.time()
@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
t0 = inputs["t0"]
t1 = get_time().result() # Deterministic - same value on resume
delta_t = t1 - t0
if delta_t > 1:
result = slow_task(1).result()
value = interrupt("question")
else:
result = slow_task(2).result()
value = interrupt("question")
return {"result": result, "value": value}
```
Entrypoint function definition requirements
An entrypoint is defined by decorating a function with the @entrypoint decorator (Python) or calling the entrypoint function with configuration and a function (JavaScript). The function must accept a single positional argument which serves as the workflow input. If multiple pieces of data need to be passed, use a dictionary or object as the input type for the first argument. Decorating a function with an entrypoint produces a Pregel instance which manages workflow execution including streaming, resumption, and checkpointing.
Serialization requirements for entrypoint and task
There are two key aspects to serialization in LangGraph: entrypoint inputs and outputs must be JSON-serializable; task outputs must be JSON-serializable. These requirements are necessary for enabling checkpointing and workflow resumption. Use Python primitives like dictionaries, lists, strings, numbers and booleans to ensure inputs and outputs are serializable. Providing non-serializable inputs or outputs will result in a runtime error when a workflow is configured with a checkpointer.
Side effects handling pitfall
Side effects should be encapsulated in tasks to ensure they are not executed multiple times when resuming a workflow. If a side effect like writing to a file or sending an email is directly included in the entrypoint it will be executed a second time when resuming the workflow. The correct approach is to wrap side effects in tasks so their execution is tracked and not repeated on resume.
Side effects encapsulation pitfall example
Incorrect example with side effect in entrypoint:
```python
@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
# This code will be executed a second time when resuming the workflow.
with open("output.txt", "w") as f:
f.write("Side effect executed")
value = interrupt("question")
return value
```
Correct approach encapsulating side effect in task:
```python
from langgraph.func import task
@task
def write_to_file():
with open("output.txt", "w") as f:
f.write("Side effect executed")
@entrypoint(checkpointer=checkpointer)
def my_workflow(inputs: dict) -> int:
# The side effect is now encapsulated in a task.
write_to_file().result()
value = interrupt("question")
return value
```
Functional API building blocks: @entrypoint and @task
The Functional API uses two key building blocks: @entrypoint decorator marks a function as the starting point of a workflow, encapsulating logic and managing execution flow including handling long-running tasks and interrupts; @task represents a discrete unit of work such as an API call or data processing step that can be executed asynchronously within an entrypoint and returns a future-like object that can be awaited or resolved synchronously.
Functional API resume after interrupt example
To resume an execution after an interrupt using Command primitive:
```python
from langgraph.types import Command
config = {
"configurable": {
"thread_id": "some_thread_id"
}
}
my_workflow.invoke(Command(resume=some_resume_value), config)
```
For async invocation:
```python
await my_workflow.ainvoke(Command(resume=some_resume_value), config)
```
For streaming resume:
```python
stream = my_workflow.stream_events(Command(resume=some_resume_value), config, version="v3")
for message in stream.messages:
for token in message.text:
print(token, end="", flush=True)
```
Functional API example: essay workflow with interrupt and resume
Example Python workflow that writes an essay and interrupts for human review:
```python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint, task
from langgraph.types import interrupt
@task
def write_essay(topic: str) -> str:
"""Write an essay about the given topic."""
time.sleep(1) # A placeholder for a long-running task.
return f"An essay about topic: {topic}"
@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str) -> dict:
"""A simple workflow that writes an essay and asks for a review."""
essay = write_essay("cat").result()
is_approved = interrupt({
"essay": essay,
"action": "Please approve/reject the essay",
})
return {
"essay": essay,
"is_approved": is_approved,
}
```
This workflow will write an essay about the topic "cat" and then pause to get a review from a human. The workflow can be interrupted for an indefinite amount of time until a review is provided. When the workflow is resumed it executes from the very start but because the result of the write_essay task was already saved the task result will be loaded from the checkpoint instead of being recomputed.
Non-deterministic control flow pitfall
Operations that might give different results each time like getting current time or random numbers should be encapsulated in tasks to ensure that on resume the same result is returned. If not in a task: Get random number (5) → interrupt → resume → get new random number (7). If in a task: Get random number (5) → interrupt → resume → (returns 5 again). This is especially important when using human-in-the-loop workflows with multiple interrupt calls because LangGraph keeps a list of resume values for each task/entrypoint. When an interrupt is encountered it's matched with the corresponding resume value. This matching is strictly index-based so the order of resume values should match the order of interrupts. If the order of execution is not maintained when resuming one interrupt call may be matched with the wrong resume value leading to incorrect results.
Functional API overview and capabilities
The Functional API allows you to add LangGraph's key features (persistence, memory, human-in-the-loop, and streaming) to applications with minimal changes to existing code. It is designed to integrate these features into existing code that may use standard language primitives for branching and control flow such as if statements, for loops, and function calls, without enforcing a rigid execution model.
Approval workflow example
Example of approval workflow using interrupt():
Python:
```python
from typing import Literal
from langgraph.types import interrupt, Command
def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
# Pause execution; payload shows up on stream.interrupts (with stream_events) or result["__interrupt__"] (with invoke)
is_approved = interrupt({
"question": "Do you want to proceed with this action?",
"details": state["action_details"]
})
# Route based on the response
if is_approved:
return Command(goto="proceed")
else:
return Command(goto="cancel")
```
JavaScript:
```typescript
import { interrupt, Command } from "@langchain/langgraph";
const approvalNode: typeof State.Node = (state) => {
// Pause execution; payload surfaces in result.__interrupt__
const isApproved = interrupt({
question: "Do you want to proceed?",
details: state.actionDetails
});
// Route based on the response
if (isApproved) {
return new Command({ goto: "proceed" });
} else {
return new Command({ goto: "cancel" });
}
}
```
Do not pass Command(update=...) or Command(goto=...) as input to invoke/stream
Command(resume=...) is the only Command pattern intended as input to invoke()/stream()/stream_events(). The other Command parameters (update, goto, graph) are designed for returning from node functions. Do not pass Command(update=...) as input to continue multi-turn conversations—pass a plain input dict instead.
Static interrupts for debugging with interrupt_before and interrupt_after
To debug and test a graph, you can use static interrupts as breakpoints to step through graph execution one node at a time. Static interrupts are triggered at defined points either before or after a node executes.
At compile time (Python):
```python
graph = builder.compile(
interrupt_before=["node_a"],
interrupt_after=["node_b", "node_c"],
checkpointer=checkpointer,
)
config = {"configurable": {"thread_id": "some_thread"}}
graph.invoke(inputs, config=config) # Run until breakpoint
graph.invoke(None, config=config) # Resume
```
At runtime (Python):
```python
config = {"configurable": {"thread_id": "some_thread"}}
graph.invoke(
inputs,
interrupt_before=["node_a"],
interrupt_after=["node_b", "node_c"],
config=config,
)
graph.invoke(None, config=config) # Resume
```
At compile time (JavaScript):
```typescript
const graph = builder.compile({
interruptBefore: ["node_a"],
interruptAfter: ["node_b", "node_c"],
checkpointer,
});
const config = {configurable: {thread_id: "some_thread"}};
await graph.invoke(inputs, config); // Run until breakpoint
await graph.invoke(null, config); // Resume
```
At runtime (JavaScript):
```typescript
await graph.invoke(inputs, {
interruptBefore: ["node_a"],
interruptAfter: ["node_b", "node_c"],
configurable: {thread_id: "some_thread"}
});
await graph.invoke(null, config); // Resume
```
interrupt() function pauses graph execution
The interrupt() function pauses graph execution at the point it is called and waits indefinitely for external input. It accepts any JSON-serializable value as a payload that is surfaced to the caller. When resuming with Command(resume=...), that value becomes the return value of the interrupt() call.
Interrupt within tool functions
You can place interrupt() directly inside tool functions to pause before tool execution and allow for human review and editing of the tool call before it executes. This makes the tool itself pause for approval whenever it's called. The interrupt payload can include action details, and the resume value can override inputs before executing.
Python:
```python
from langchain.tools import tool
from langgraph.types import interrupt
@tool
def send_email(to: str, subject: str, body: str):
"""Send an email to a recipient."""
# Pause before sending; payload surfaces on stream.interrupts when using event streaming
response = interrupt({
"action": "send_email",
"to": to,
"subject": subject,
"body": body,
"message": "Approve sending this email?"
})
if response.get("action") == "approve":
# Resume value can override inputs before executing
final_to = response.get("to", to)
final_subject = response.get("subject", subject)
final_body = response.get("body", body)
return f"Email sent to {final_to} with subject '{final_subject}'"
return "Email cancelled by user"
```
JavaScript:
```typescript
import { tool } from "@langchain/core/tools";
import { interrupt } from "@langchain/langgraph";
import * as z from "zod";
const sendEmailTool = tool(
async ({ to, subject, body }) => {
// Pause before sending; payload surfaces in result.__interrupt__
const response = interrupt({
action: "send_email",
to,
subject,
body,
message: "Approve sending this email?",
});
if (response?.action === "approve") {
// Resume value can override inputs before executing
const finalTo = response.to ?? to;
const finalSubject = response.subject ?? subject;
const finalBody = response.body ?? body;
return `Email sent to ${finalTo} with subject '${finalSubject}'`;
}
return "Email cancelled by user";
},
{
name: "send_email",
description: "Send an email to a recipient",
schema: z.object({
to: z.string(),
subject: z.string(),
body: z.string(),
}),
},
);
```
Review and edit state pattern with interrupt()
Example showing how to let a human review and edit graph state:
Python:
```python
from langgraph.types import interrupt
def review_node(state: State):
# Pause and show the current content for review (payload surfaces on stream.interrupts)
edited_content = interrupt({
"instruction": "Review and edit this content",
"content": state["generated_text"]
})
# Update the state with the edited version
return {"generated_text": edited_content}
```
JavaScript:
```typescript
import { interrupt } from "@langchain/langgraph";
const reviewNode: typeof State.Node = (state) => {
// Pause and show the current content for review (surfaces in result.__interrupt__)
const editedContent = interrupt({
instruction: "Review and edit this content",
content: state.generatedText
});
// Update the state with the edited version
return { generatedText: editedContent };
}
```
Resuming interrupt with Command(resume=...)
After an interrupt pauses execution, resume the graph by invoking it again with a Command that contains the resume value. Python: graph.stream_events(Command(resume=value), config=config, version="v3") or graph.invoke(Command(resume=value), config=config). JavaScript: await graph.invoke(new Command({resume: value}), config). The resume value must use the same thread_id that was used when the interrupt occurred, and any JSON-serializable value can be passed as the resume value.
interrupt() example - approval node
Example showing interrupt() in a node:
Python:
```python
from langgraph.types import interrupt
def approval_node(state: State):
# Pause and ask for approval
approved = interrupt("Do you approve this action?")
# When you resume, Command(resume=...) returns that value here
return {"approved": approved}
```
JavaScript:
```typescript
import { interrupt } from "@langchain/langgraph";
async function approvalNode(state: State) {
// Pause and ask for approval
const approved = interrupt("Do you approve this action?");
// Command({ resume: ... }) provides the value returned into this variable
return { approved };
}
```
Side effects before interrupt() must be idempotent
Any side effects (API calls, database operations) called before interrupt() should be idempotent, meaning they can be applied multiple times without changing the result beyond the initial execution. Because nodes re-run from the beginning on resume, non-idempotent operations will execute multiple times, potentially causing duplicate records or unintended overwrites. Place side effects after interrupt() calls or separate them into different nodes when possible.
Only pass JSON-serializable values to interrupt()
Pass only JSON-serializable values to interrupt() - simple types (strings, numbers, booleans) and dictionaries/objects with simple values. Do not pass functions, class instances, or other complex objects that cannot be serialized, as this will fail depending on which checkpointer is used.
Use conditional edges for input validation with interrupts
The correct pattern for validating human input is: 1) Store the re-prompt question in state (e.g. pending_question); 2) In the node, call interrupt() exactly once, passing the current question from state; 3) If the answer is invalid, return the updated pending_question so the next invocation re-prompts; 4) Use add_conditional_edges (Python) or addConditionalEdges (JavaScript) to route back to the node until a valid value is collected. This avoids exponential re-execution that occurs with while True + interrupt() loops inside a single node.
Do not conditionally skip interrupt() calls or loop them with non-deterministic logic
LangGraph keeps a list of resume values specific to the task executing the node. Interrupt matching is strictly index-based, so the order of interrupt() calls within a node is important. Do not conditionally skip interrupt() calls based on state, as this changes the order. Do not loop interrupt() calls using logic that isn't deterministic across executions (like while True validation loops). Use conditional edges instead for validation re-prompting.
Do not wrap interrupt() in try/except or try/catch blocks
The interrupt() function pauses execution by throwing a special exception. If you wrap the interrupt() call in a try/except (Python) or try/catch (JavaScript) block, you will catch this exception and the interrupt will not be passed back to the graph. Keep interrupt() calls separate from error-prone code.
Node restarts from the beginning when resumed from interrupt
When a node is resumed from an interrupt, the entire node re-executes from the beginning. Any code that ran before the interrupt() call will execute again. This means side effects before interrupt() calls must be idempotent.
Resume multiple simultaneous interrupts with interrupt ID mapping
When parallel branches interrupt simultaneously (fan-out to multiple nodes that each call interrupt()), resume multiple interrupts in a single invocation by mapping each interrupt ID to its resume value. This ensures each response is paired with the correct interrupt at runtime. Access interrupt IDs from the __interrupt__ field.
Deep Agents provides agent harness on top of LangGraph
Deep Agents is an agent harness built on top of LangGraph that provides planning, subagents, filesystem tools, and context management capabilities beyond the core LangGraph orchestration layer.
Install LangGraph JavaScript/TypeScript
LangGraph for JavaScript can be installed using npm with: npm install @langchain/langgraph @langchain/core. Alternative package managers include pnpm (pnpm add @langchain/langgraph @langchain/core), yarn (yarn add @langchain/langgraph @langchain/core), or bun (bun add @langchain/langgraph @langchain/core).
Hello world example in TypeScript
import { StateSchema, MessagesValue, type GraphNode, StateGraph, START, END } from "@langchain/langgraph";
const State = new StateSchema({
messages: MessagesValue,
});
const mockLlm: GraphNode<typeof State> = (state) => {
return { messages: [{ role: "ai", content: "hello world" }] };
};
const graph = new StateGraph(State)
.addNode("mock_llm", mockLlm)
.addEdge(START, "mock_llm")
.addEdge("mock_llm", END)
.compile();
await graph.invoke({ messages: [{ role: "user", content: "hi!" }] });
This example shows how to create a basic LangGraph with state schema, a node, edges from START to the node and from the node to END, compile it, and invoke it with input.
Hello world example in Python
from langgraph.graph import StateGraph, MessagesState, START, END
def mock_llm(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}
graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()
graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
This example shows how to create a basic LangGraph with state, a node, edges from START to the node and from the node to END, compile it, and invoke it with input.
LangGraph is focused on agent orchestration, not abstractions
LangGraph is very low-level and focused entirely on agent orchestration. It does not abstract prompts or architecture. Users do not need to use LangChain to use LangGraph, though LangChain components can be used throughout LangGraph applications to integrate models and tools.
LangGraph vs higher-level abstractions
LangGraph is very low-level. If you are just getting started with agents or want a higher-level abstraction, LangChain's agents are recommended as they provide prebuilt architectures for common LLM and tool-calling loops. LangGraph is recommended for users who need fine-grained control over agent orchestration.
LangGraph core strengths and capabilities
LangGraph's core strengths are: durable execution, streaming, human-in-the-loop interaction, persistence through failures, the ability to mix deterministic steps with LLM-driven agentic steps in a single graph, and comprehensive memory for both short-term working memory and long-term memory across sessions.
LangGraph is a low-level orchestration framework for building stateful agents
LangGraph is a low-level orchestration framework and runtime for building, managing, and deploying long-running, stateful agents. It gives fine-grained control to mix deterministic, hand-coded steps with LLM-driven agentic steps in the same graph, enabling bespoke agents that behave exactly as required by the application.