Graph API Reference

from langgraph.graph import StateGraph
 
graph = StateGraph(
    state_schema,              # TypedDict, Pydantic model, or dataclass
    input_schema=None,         # Optional input schema
    output_schema=None,        # Optional output schema
    config_schema=None         # Optional runtime config schema
)

import { StateGraph, StateSchema } from "@langchain/langgraph";
 
const graph = new StateGraph({
  state: StateSchema,         // Required
  input: InputSchema,         // Optional
  output: OutputSchema,       // Optional
  context: ContextSchema      // Optional
});

graph.add_node(
    name: str,                 # Node name
    action: Callable,          # Node function
    retry_policy: RetryPolicy = None,  # Retry configuration
    cache_policy: CachePolicy = None   # Cache configuration
)

graph.addNode(
  name: string,
  action: Function,
  options?: {
    retryPolicy?: RetryPolicy,
    cachePolicy?: CachePolicy,
    ends?: string[]  // For Command routing
  }
);

graph.add_edge(from_node: str, to_node: str)

graph.add_conditional_edges(
    source: str,               # Source node
    path: Callable,            # Routing function
    path_map: list[str] | dict = None  # Optional destination mapping
)

graph.addConditionalEdges(
  source: string,
  path: Function,
  pathMap?: string[] | Record<string, string>
);

app = graph.compile(
    checkpointer: BaseCheckpointSaver = None,  # For persistence
    store: BaseStore = None,                   # For long-term memory
    interrupt_before: list[str] = None,        # Nodes to interrupt before
    interrupt_after: list[str] = None,         # Nodes to interrupt after
    debug: bool = False                        # Enable debug mode
)

const app = graph.compile({
  checkpointer?: BaseCheckpointSaver,
  store?: BaseStore,
  interruptBefore?: string[],
  interruptAfter?: string[],
  debug?: boolean
});

result = app.invoke(
    input: dict,               # Graph input
    config: RunnableConfig = None,  # Thread ID, etc.
    stream_mode: str | list[str] = "values",  # Stream mode(s)
    debug: bool = False
)

const result = await app.invoke(
  input: object,
  config?: {
    configurable?: { thread_id?: string },
    streamMode?: string | string[],
    recursionLimit?: number
  }
);

for chunk in app.stream(
    input: dict,
    config: RunnableConfig = None,
    stream_mode: str | list[str] = "values",
    subgraphs: bool = False
):
    process(chunk)

for await (const chunk of await app.stream(
  input: object,
  config?: {
    streamMode?: string | string[],
    subgraphs?: boolean
  }
)) {
  process(chunk);
}

state = app.get_state(config)
# state.values - Current state
# state.next - Next nodes to execute
# state.tasks - Pending tasks

for snapshot in app.get_state_history(config):
    print(snapshot.values)

app.update_state(
    config,
    values: dict,              # State updates
    as_node: str = None        # Treat as coming from this node
)

from langgraph.graph import START
 
graph.add_edge(START, "first_node")

from langgraph.graph import END
 
graph.add_edge("last_node", END)

from langgraph.graph import MessagesState
 
class MyState(MessagesState):
    custom_field: str

import { StateSchema, MessagesValue } from "@langchain/langgraph";
 
const MyState = new StateSchema({
  messages: MessagesValue,
  customField: z.string()
});

from langgraph.types import Command
 
return Command(
    update={"field": "value"},
    goto="next_node"
)

import { Command } from "@langchain/langgraph";
 
return new Command({
  update: { field: "value" },
  goto: "nextNode"
});

from langgraph.types import interrupt
 
response = interrupt({"message": "Approve?"})

import { interrupt } from "@langchain/langgraph";
 
const response = interrupt({ message: "Approve?" });

config = {
    "configurable": {
        "thread_id": "conversation-1",
        "checkpoint_id": "...",  # Optional
        "checkpoint_ns": ""      # Optional
    },
    "recursion_limit": 25,       # Max steps
    "tags": ["production"],      # For tracing
}

const config = {
  configurable: {
    thread_id: "conversation-1",
    checkpoint_id: "...",  // Optional
    checkpoint_ns: ""      // Optional
  },
  recursionLimit: 25,      // Max steps
  tags: ["production"]     // For tracing
};