langgraph-fundamentals

name	LangGraph Fundamentals
description	INVOKE THIS SKILL when writing ANY LangGraph code. Covers StateGraph creation, node functions, edges, state schemas with reducers (Annotated), and the Command API.

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.

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
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

Define state schema with reducers for accumulating lists and summing integers. ```python from typing_extensions import TypedDict, Annotated import operator

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) }
  ),
});

Without a reducer, returning a list overwrites previous values. ```python # WRONG: List will be OVERWRITTEN class State(TypedDict): messages: list # No reducer!

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"] }

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!

CORRECT: Return dict with only the updates

def my_node(state: State) -> dict: return {"field": "updated"}

</python>
<typescript>
Return partial updates only, not the full state object.
```typescript
// WRONG: Returning entire state
const myNode = async (state: typeof State.State) => {
  state.field = "updated";
  return state;  // Don't do this!
};

// CORRECT: Return partial updates
const myNode = async (state: typeof State.State) => {
  return { field: "updated" };
};

---

Building Graphs

Need	Edge Type	When to Use
Always go to same node	add_edge()	Fixed, deterministic flow
Route based on state	add_conditional_edges()	Dynamic branching
Update state AND route	Command	Combine logic in single node
Fan-out to multiple nodes	Send	Parallel processing with dynamic inputs

Simple two-node graph with linear edges. ```python from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict

class State(TypedDict): input: str output: str

def process_input(state: State) -> dict: return {"output": f"Processed: {state['input']}"}

def finalize(state: State) -> dict: return {"output": state["output"].upper()}

graph = ( StateGraph(State) .add_node("process", process_input) .add_node("finalize", finalize) .add_edge(START, "process") .add_edge("process", "finalize") .add_edge("finalize", END) .compile() )

result = graph.invoke({"input": "hello"}) print(result["output"]) # "PROCESSED: HELLO"

</python>
<typescript>
Chain nodes with addEdge and compile before invoking.
```typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  input: z.string(),
  output: z.string().default(""),
});

const processInput = async (state: typeof State.State) => {
  return { output: `Processed: ${state.input}` };
};

const finalize = async (state: typeof State.State) => {
  return { output: state.output.toUpperCase() };
};

const graph = new StateGraph(State)
  .addNode("process", processInput)
  .addNode("finalize", finalize)
  .addEdge(START, "process")
  .addEdge("process", "finalize")
  .addEdge("finalize", END)
  .compile();

const result = await graph.invoke({ input: "hello" });
console.log(result.output);  // "PROCESSED: HELLO"

Route to different nodes based on state with conditional edges. ```python from typing import Literal from langgraph.graph import StateGraph, START, END

class State(TypedDict): query: str route: str result: str

def classify(state: State) -> dict: if "weather" in state["query"].lower(): return {"route": "weather"} return {"route": "general"}

def route_query(state: State) -> Literal["weather", "general"]: return state["route"]

graph = ( StateGraph(State) .add_node("classify", classify) .add_node("weather", lambda s: {"result": "Sunny, 72F"}) .add_node("general", lambda s: {"result": "General response"}) .add_edge(START, "classify") .add_conditional_edges("classify", route_query, ["weather", "general"]) .add_edge("weather", END) .add_edge("general", END) .compile() )

</python>
<typescript>
addConditionalEdges routes based on function return value.
```typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  query: z.string(),
  route: z.string().default(""),
  result: z.string().default(""),
});

const classify = async (state: typeof State.State) => {
  if (state.query.toLowerCase().includes("weather")) {
    return { route: "weather" };
  }
  return { route: "general" };
};

const routeQuery = (state: typeof State.State) => state.route;

const graph = new StateGraph(State)
  .addNode("classify", classify)
  .addNode("weather", async () => ({ result: "Sunny, 72F" }))
  .addNode("general", async () => ({ result: "General response" }))
  .addEdge(START, "classify")
  .addConditionalEdges("classify", routeQuery, ["weather", "general"])
  .addEdge("weather", END)
  .addEdge("general", END)
  .compile();

Command lets you update state AND choose next node in one return. ```python from langgraph.types import Command from typing import Literal

class State(TypedDict): count: int result: str

def node_a(state: State) -> Command[Literal["node_b", "node_c"]]: """Update state AND decide next node in one return.""" new_count = state["count"] + 1 if new_count > 5: return Command(update={"count": new_count}, goto="node_c") return Command(update={"count": new_count}, goto="node_b")

graph = ( StateGraph(State) .add_node("node_a", node_a) .add_node("node_b", lambda s: {"result": "B"}) .add_node("node_c", lambda s: {"result": "C"}) .add_edge(START, "node_a") .add_edge("node_b", END) .add_edge("node_c", END) .compile() )

</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();

Fan-out with Send: return `[Send("worker", {...})]` from a conditional edge to spawn parallel workers. Requires a reducer on the results field. A deep understanding of Send and execution patterns is essential for complex workflows. Conditional edge to END prevents infinite loops. ```python from langgraph.graph import StateGraph, START, END

class State(TypedDict): count: int max_iterations: int

def increment(state: State) -> dict: return {"count": state["count"] + 1}

def should_continue(state: State) -> str: if state["count"] >= state["max_iterations"]: return END return "increment"

graph = ( StateGraph(State) .add_node("increment", increment) .add_edge(START, "increment") .add_conditional_edges("increment", should_continue) .compile() )

result = graph.invoke({"count": 0, "max_iterations": 5}) print(result["count"]) # 5

</python>
<typescript>
Return END from conditional function to terminate loop.
```typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  count: z.number().default(0),
  maxIterations: z.number(),
});

const increment = async (state: typeof State.State) => {
  return { count: state.count + 1 };
};

const shouldContinue = (state: typeof State.State) => {
  if (state.count >= state.maxIterations) return END;
  return "increment";
};

const graph = new StateGraph(State)
  .addNode("increment", increment)
  .addEdge(START, "increment")
  .addConditionalEdges("increment", shouldContinue)
  .compile();

const result = await graph.invoke({ count: 0, maxIterations: 5 });
console.log(result.count);  // 5

Must compile() to get executable graph. ```python # WRONG builder.invoke({"input": "test"}) # AttributeError!

CORRECT

graph = builder.compile() graph.invoke({"input": "test"})

</python>
<typescript>
Must compile() to get executable graph.
```typescript
// WRONG
await builder.invoke({ input: "test" });

// CORRECT
const graph = builder.compile();
await graph.invoke({ input: "test" });

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")

CORRECT

def should_continue(state): return END if state["count"] > 10 else "node_b" builder.add_conditional_edges("node_a", should_continue)

</python>
<typescript>
Use conditional edges with END return to break loops.
```typescript
// WRONG: Loops forever
builder.addEdge("node_a", "node_b").addEdge("node_b", "node_a");

// CORRECT
builder.addConditionalEdges("node_a", (state) => state.count > 10 ? END : "node_b");

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"])

Command return type needs Literal for routing destinations (Python)

def node_a(state) -> Command[Literal["node_b", "node_c"]]: return Command(goto="node_b")

START is entry-only - cannot route back to it

builder.add_edge("node_a", START) # WRONG! builder.add_edge("node_a", "entry") # Use a named entry node instead

Reducer expects matching types

return {"items": ["item"]} # List for list reducer, not a string

```typescript
// Always await graph.invoke() - it returns a Promise
const result = await graph.invoke({ input: "test" });

// TS Command nodes need { ends } to declare routing destinations
builder.addNode("router", routerFn, { ends: ["node_b", "node_c"] });

### What You CAN Configure

• Define custom state schemas with TypedDict/StateSchema
• Add reducers to control how state updates are merged
• Create nodes (any function) and add static/conditional edges
• Use Command for combined state update + routing
• Create loops with conditional termination

What You CANNOT Configure

• Modify START/END behavior
• Access state outside node functions
• Modify state directly (must return partial update dicts)