langgraph-fundamentals

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

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

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

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

// 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" };
};

from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
 
def plain_node(state: State):
    return {"results": "done"}
 
def node_with_config(state: State, config: RunnableConfig):
    thread_id = config["configurable"]["thread_id"]
    return {"results": f"Thread: {thread_id}"}
 
def node_with_runtime(state: State, runtime: Runtime[Context]):
    user_id = runtime.context.user_id
    return {"results": f"User: {user_id}"}

import { GraphNode, StateSchema } from "@langchain/langgraph";
 
const plainNode: GraphNode<typeof State> = (state) => {
  return { results: "done" };
};
 
const nodeWithConfig: GraphNode<typeof State> = (state, config) => {
  const threadId = config?.configurable?.thread_id;
  return { results: `Thread: ${threadId}` };
};

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"

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"

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

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

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

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

from langgraph.types import Send
from typing import Annotated
import operator
 
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"]]
 
def worker(state: dict) -> dict:
    return {"results": [f"Completed: {state['task']}"]}
 
def synthesize(state: OrchestratorState) -> dict:
    return {"summary": f"Processed {len(state['results'])} tasks"}
 
graph = (
    StateGraph(OrchestratorState)
    .add_node("worker", worker)
    .add_node("synthesize", synthesize)
    .add_conditional_edges(START, orchestrator, ["worker"])
    .add_edge("worker", "synthesize")
    .add_edge("synthesize", END)
    .compile()
)
 
result = graph.invoke({"tasks": ["Task A", "Task B", "Task C"]})

import { Send, StateGraph, StateSchema, ReducedValue, START, END } from "@langchain/langgraph";
import { z } from "zod";
 
const State = new StateSchema({
  tasks: z.array(z.string()),
  results: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (curr, upd) => curr.concat(upd) }
  ),
  summary: z.string().default(""),
});
 
const orchestrator = (state: typeof State.State) => {
  return state.tasks.map((task) => new Send("worker", { task }));
};
 
const worker = async (state: { task: string }) => {
  return { results: [`Completed: ${state.task}`] };
};
 
const synthesize = async (state: typeof State.State) => {
  return { summary: `Processed ${state.results.length} tasks` };
};
 
const graph = new StateGraph(State)
  .addNode("worker", worker)
  .addNode("synthesize", synthesize)
  .addConditionalEdges(START, orchestrator, ["worker"])
  .addEdge("worker", "synthesize")
  .addEdge("synthesize", END)
  .compile();

# WRONG: No reducer - last worker overwrites
class State(TypedDict):
    results: list
 
# CORRECT
class State(TypedDict):
    results: Annotated[list, operator.add]  # Accumulates

// WRONG: No reducer
const State = new StateSchema({ results: z.array(z.string()) });
 
// CORRECT
const State = new StateSchema({
  results: new ReducedValue(z.array(z.string()).default(() => []), { reducer: (curr, upd) => curr.concat(upd) }),
});

result = graph.invoke({"input": "hello"})
# With config (for persistence, tags, etc.)
result = graph.invoke({"input": "hello"}, {"configurable": {"thread_id": "1"}})

const result = await graph.invoke({ input: "hello" });
// With config
const result = await graph.invoke({ input: "hello" }, { configurable: { thread_id: "1" } });

for chunk in graph.stream(
    {"messages": [HumanMessage("Hello")]},
    stream_mode="messages"
):
    token, metadata = chunk
    if hasattr(token, "content"):
        print(token.content, end="", flush=True)

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

from langgraph.config import get_stream_writer
 
def my_node(state):
    writer = get_stream_writer()
    writer("Processing step 1...")
    # Do work
    writer("Complete!")
    return {"result": "done"}
 
for chunk in graph.stream({"data": "test"}, stream_mode="custom"):
    print(chunk)

import { getWriter } from "@langchain/langgraph";
 
const myNode = async (state: typeof State.State) => {
  const writer = getWriter();
  writer("Processing step 1...");
  // Do work
  writer("Complete!");
  return { result: "done" };
};
 
for await (const chunk of graph.stream({ data: "test" }, { streamMode: "custom" })) {
  console.log(chunk);
}

from langgraph.types import RetryPolicy
 
workflow.add_node(
    "search_documentation",
    search_documentation,
    retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)

workflow.addNode(
  "searchDocumentation",
  searchDocumentation,
  {
    retryPolicy: { maxAttempts: 3, initialInterval: 1.0 },
  },
);

from langgraph.prebuilt import ToolNode
 
tool_node = ToolNode(tools, handle_tool_errors=True)
 
workflow.add_node("tools", tool_node)

import { ToolNode } from "@langchain/langgraph/prebuilt";
 
const toolNode = new ToolNode(tools, { handleToolErrors: true });
 
workflow.addNode("tools", toolNode);

# WRONG
builder.invoke({"input": "test"})  # AttributeError!
 
# CORRECT
graph = builder.compile()
graph.invoke({"input": "test"})

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

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

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

# 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

// 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"] });