INVOKE THIS SKILL when writing ANY LangGraph code. Covers StateGraph, state schemas, nodes, edges, Command, Send, invoke, streaming, and error handling.
INVOKE THIS SKILL when writing ANY LangGraph code. Covers StateGraph, state schemas, nodes, edges, Command, Send, invoke, streaming, and error handling.
LangGraph models agent workflows as **directed graphs**:
StateGraph: Main class for building stateful graphs
Nodes: Functions that perform work and update state
Edges: Define execution order (static or conditional)
START/END: Special nodes marking entry and exit points
State with Reducers: Control how state updates are merged
Graphs must be compile()d before execution.
Designing a LangGraph application
Follow these 5 steps when building a new graph:
Map out discrete steps — sketch a flowchart of your workflow. Each step becomes a node.
Identify what each step does — categorize nodes: LLM step, data step, action step, or user input step. For each, determine static context (prompt), dynamic context (from state), retry strategy, and desired outcome.
Design your state — state is shared memory for all nodes. Store raw data, format prompts on-demand inside nodes.
Build your nodes — implement each step as a function that takes state and returns partial updates.
Wire it together — connect nodes with edges, add conditional routing, compile with a checkpointer if needed.
Use LangGraph When
Use Alternatives When
Need fine-grained control over agent orchestration
Quick prototyping → LangChain agents
Building complex workflows with branching/loops
Simple stateless workflows → LangChain direct
Define state schema with reducers for accumulating lists and summing integers.
```python
from typing_extensions import TypedDict, Annotated
import operator
Without a reducer, returning a list overwrites previous values.
```python
# WRONG: List will be OVERWRITTEN
class State(TypedDict):
messages: list # No reducer!
Nodes must return partial updates, not mutate and return full state.
```python
# WRONG: Returning entire state object
def my_node(state: State) -> State:
state["field"] = "updated"
return state # Don't mutate and return!
Simple two-node graph with linear edges.
```python
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
Route to different nodes based on state with conditional edges.
```python
from typing import Literal
from langgraph.graph import StateGraph, START, END
Command lets you update state AND choose next node in one return.
```python
from langgraph.types import Command
from typing import Literal
Fan out tasks to parallel workers using the Send API and aggregate results.
```python
from langgraph.types import Send
from typing import Annotated
import operator
Use a reducer to accumulate parallel worker results (otherwise last worker overwrites).
```python
# WRONG: No reducer - last worker overwrites
class State(TypedDict):
results: list
```python
result = graph.invoke({"input": "hello"})
# With config (for persistence, tags, etc.)
result = graph.invoke({"input": "hello"}, {"configurable": {"thread_id": "1"}})
```
```typescript
const result = await graph.invoke({ input: "hello" });
// With config
const result = await graph.invoke({ input: "hello" }, { configurable: { thread_id: "1" } });
```
Stream LLM tokens in real-time for chat UI display.
```python
for chunk in graph.stream(
{"messages": [HumanMessage("Hello")]},
stream_mode="messages"
):
token, metadata = chunk
if hasattr(token, "content"):
print(token.content, end="", flush=True)
```
Stream LLM tokens in real-time for chat UI display.
```typescript
for await (const chunk of graph.stream(
{ messages: [new HumanMessage("Hello")] },
{ streamMode: "messages" }
)) {
const [token, metadata] = chunk;
if (token.content) {
process.stdout.write(token.content);
}
}
```
Emit custom progress updates from within nodes using the stream writer.
```python
from langgraph.config import get_stream_writer
Use RetryPolicy for transient errors (network issues, rate limits).
```python
from langgraph.types import RetryPolicy
Use ToolNode from langgraph.prebuilt to handle tool execution and errors. When handle_tool_errors=True, errors are returned as ToolMessages so the LLM can recover.
```python
from langgraph.prebuilt import ToolNode
Must compile() to get executable graph.
```python
# WRONG
builder.invoke({"input": "test"}) # AttributeError!
Provide conditional path to END to avoid infinite loops.
```python
# WRONG: Loops forever
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_a")
Other common mistakes:
```python
# Router must return names of nodes that exist in the graph
builder.add_node("my_node", func) # Add node BEFORE referencing in edges
builder.add_conditional_edges("node_a", router, ["my_node"])
### What You Should NOT Do
Require human-in-the-loop, persistence
Batteries-included features → Deep Agents
State Management
Need
Solution
Example
Overwrite value
No reducer (default)
Simple fields like counters
Append to list
Reducer (operator.add / concat)
Message history, logs
Custom logic
Custom reducer function
Complex merging
class State(TypedDict):
name: str # Default: overwrites on update
messages: Annotated[list, operator.add] # Appends to list
total: Annotated[int, operator.add] # Sums integers
</python>
<typescript>
Use StateSchema with ReducedValue for accumulating arrays.
```typescript
import { StateSchema, ReducedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
name: z.string(), // Default: overwrites
messages: MessagesValue, // Built-in for messages
items: new ReducedValue(
z.array(z.string()).default(() => []),
{ reducer: (current, update) => current.concat(update) }
),
});
Node 1 returns: {"messages": ["A"]}
Node 2 returns: {"messages": ["B"]}
Final: {"messages": ["B"]} # "A" is LOST!
CORRECT: Use Annotated with operator.add
from typing import Annotated
import operator
class State(TypedDict):
messages: Annotated[list, operator.add]
Final: {"messages": ["A", "B"]}
</python>
<typescript>
Without ReducedValue, arrays are overwritten not appended.
```typescript
// WRONG: Array will be overwritten
const State = new StateSchema({
items: z.array(z.string()), // No reducer!
});
// Node 1: { items: ["A"] }, Node 2: { items: ["B"] }
// Final: { items: ["B"] } // A is lost!
// CORRECT: Use ReducedValue
const State = new StateSchema({
items: new ReducedValue(
z.array(z.string()).default(() => []),
{ reducer: (current, update) => current.concat(update) }
),
});
// Final: { items: ["A", "B"] }
</python>
<typescript>
Return Command with update and goto to combine state change with routing.
```typescript
import { StateGraph, StateSchema, START, END, Command } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
count: z.number().default(0),
result: z.string().default(""),
});
const nodeA = async (state: typeof State.State) => {
const newCount = state.count + 1;
if (newCount > 5) {
return new Command({ update: { count: newCount }, goto: "node_c" });
}
return new Command({ update: { count: newCount }, goto: "node_b" });
};
const graph = new StateGraph(State)
.addNode("node_a", nodeA, { ends: ["node_b", "node_c"] })
.addNode("node_b", async () => ({ result: "B" }))
.addNode("node_c", async () => ({ result: "C" }))
.addEdge(START, "node_a")
.addEdge("node_b", END)
.addEdge("node_c", END)
.compile();
Python: Use Command[Literal["node_a", "node_b"]] as the return type annotation to declare valid goto destinations.
TypeScript: Pass { ends: ["node_a", "node_b"] } as the third argument to addNode to declare valid goto destinations.
Warning: Command only adds dynamic edges — static edges defined with add_edge / addEdge still execute. If node_a returns Command(goto="node_c") and you also have graph.add_edge("node_a", "node_b"), bothnode_b and node_c will run.
Send API
Fan-out with Send: return [Send("worker", {...})] from a conditional edge to spawn parallel workers. Requires a reducer on the results field.
class OrchestratorState(TypedDict):
tasks: list[str]
results: Annotated[list, operator.add]
summary: str
def orchestrator(state: OrchestratorState):
"""Fan out tasks to workers."""
return [Send("worker", {"task": task}) for task in state["tasks"]]
</python>
<typescript>
Use ToolNode from @langchain/langgraph/prebuilt to handle tool execution and errors. When handleToolErrors is true, errors are returned as ToolMessages so the LLM can recover.
```typescript
import { ToolNode } from "@langchain/langgraph/prebuilt";
const toolNode = new ToolNode(tools, { handleToolErrors: true });
workflow.addNode("tools", toolNode);