graph-api

133 notes in this subject, read out of this brain and free to use. This is page 2 of 3.

Relation between LangChain and LangGraph

LangChain is the agent framework providing abstractions and integrations for models, tools, and agent loops. LangGraph is the orchestration runtime providing durable execution, streaming, human-in-the-loop, and persistence. LangChain agent abstractions are built on top of LangGraph.

Mix deterministic and agentic steps in LangGraph

One of LangGraph's core strengths is the ability to mix deterministic steps with LLM-driven agentic steps in a single graph. This lets you build workflows where parts of the logic are fully predictable and auditable (deterministic steps) while other parts are flexible and model-driven (agentic steps), giving fine-grained control over exactly where and how AI is applied.

Topic channel type for PubSub and accumulation

Topic is a configurable PubSub channel useful for sending multiple values between actors or accumulating output across steps. It can be configured to deduplicate values or to accumulate all values written during a run.

Pregel runtime manages LangGraph execution

Pregel implements LangGraph's runtime, managing the execution of LangGraph applications. Compiling a StateGraph or creating an @entrypoint produces a Pregel instance that can be invoked with input. The runtime is named after Google's Pregel algorithm, which describes an efficient method for large-scale parallel computation using graphs.

Pregel algorithm: Plan, Execution, Update phases

Pregel organizes execution into multiple steps, each consisting of three phases: Plan determines which actors to execute in this step (first step selects actors subscribing to input channels; subsequent steps select actors subscribing to channels updated in the previous step). Execution runs all selected actors in parallel until all complete, one fails, or a timeout is reached (channel updates are invisible to actors until the next step). Update writes channel values from the actors in this step. The process repeats until no actors are selected for execution or a maximum number of steps is reached.

Actors and channels in Pregel

In Pregel, actors read data from channels and write data to channels. An actor is a PregelNode that subscribes to channels and implements LangChain's Runnable interface. Channels are used to communicate between actors, each with a value type, an update type, and an update function that takes a sequence of updates and modifies the stored value.

LastValue channel type

LastValue is the default channel type. It stores the last value written to it, overwriting any previous value. Use it for input and output values, or for passing data from one step to the next.

BinaryOperatorAggregate channel for running aggregates

BinaryOperatorAggregate stores a persistent value that is updated by applying a binary operator to the current value and each new update. Use it to compute running aggregates across steps.

Pregel single node example

Example showing a single node that subscribes to channel 'a', doubles the string value, and writes to channel 'b': Python: ```python from langgraph.channels import EphemeralValue from langgraph.pregel import Pregel, NodeBuilder node1 = ( NodeBuilder().subscribe_only("a") .do(lambda x: x + x) .write_to("b") ) app = Pregel( nodes={"node1": node1}, channels={ "a": EphemeralValue(str), "b": EphemeralValue(str), }, input_channels=["a"], output_channels=["b"], ) app.invoke({"a": "foo"}) ``` Output: `{'b': 'foofoo'}` TypeScript: ```typescript import { EphemeralValue } from "@langchain/langgraph/channels"; import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel"; const node1 = new NodeBuilder() .subscribeOnly("a") .do((x: string) => x + x) .writeTo("b"); const app = new Pregel({ nodes: { node1 }, channels: { a: new EphemeralValue<string>(), b: new EphemeralValue<string>(), }, inputChannels: ["a"], outputChannels: ["b"], }); await app.invoke({ a: "foo" }); ``` Output: `{ b: 'foofoo' }`

Pregel multiple nodes example

Example showing two nodes where node1 reads from 'a', doubles the value, writes to 'b'; node2 reads from 'b', doubles the value, writes to 'c': Python: ```python from langgraph.channels import LastValue, EphemeralValue from langgraph.pregel import Pregel, NodeBuilder node1 = ( NodeBuilder().subscribe_only("a") .do(lambda x: x + x) .write_to("b") ) node2 = ( NodeBuilder().subscribe_only("b") .do(lambda x: x + x) .write_to("c") ) app = Pregel( nodes={"node1": node1, "node2": node2}, channels={ "a": EphemeralValue(str), "b": LastValue(str), "c": EphemeralValue(str), }, input_channels=["a"], output_channels=["b", "c"], ) app.invoke({"a": "foo"}) ``` Output: `{'b': 'foofoo', 'c': 'foofoofoofoo'}` TypeScript: ```typescript import { LastValue, EphemeralValue } from "@langchain/langgraph/channels"; import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel"; const node1 = new NodeBuilder() .subscribeOnly("a") .do((x: string) => x + x) .writeTo("b"); const node2 = new NodeBuilder() .subscribeOnly("b") .do((x: string) => x + x) .writeTo("c"); const app = new Pregel({ nodes: { node1, node2 }, channels: { a: new EphemeralValue<string>(), b: new LastValue<string>(), c: new EphemeralValue<string>(), }, inputChannels: ["a"], outputChannels: ["b", "c"], }); await app.invoke({ a: "foo" }); ``` Output: `{ b: 'foofoo', c: 'foofoofoofoo' }`

Pregel Topic channel accumulate example

Example showing Topic channel with accumulation where node1 writes to both 'b' and 'c', and node2 reads from 'b' and writes to 'c', collecting all writes: Python: ```python from langgraph.channels import EphemeralValue, Topic from langgraph.pregel import Pregel, NodeBuilder node1 = ( NodeBuilder().subscribe_only("a") .do(lambda x: x + x) .write_to("b", "c") ) node2 = ( NodeBuilder().subscribe_to("b") .do(lambda x: x["b"] + x["b"]) .write_to("c") ) app = Pregel( nodes={"node1": node1, "node2": node2}, channels={ "a": EphemeralValue(str), "b": EphemeralValue(str), "c": Topic(str, accumulate=True), }, input_channels=["a"], output_channels=["c"], ) app.invoke({"a": "foo"}) ``` Output: `{'c': ['foofoo', 'foofoofoofoo']}` TypeScript: ```typescript import { EphemeralValue, Topic } from "@langchain/langgraph/channels"; import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel"; const node1 = new NodeBuilder() .subscribeOnly("a") .do((x: string) => x + x) .writeTo("b", "c"); const node2 = new NodeBuilder() .subscribeTo("b") .do((x: { b: string }) => x.b + x.b) .writeTo("c"); const app = new Pregel({ nodes: { node1, node2 }, channels: { a: new EphemeralValue<string>(), b: new EphemeralValue<string>(), c: new Topic<string>({ accumulate: true }), }, inputChannels: ["a"], outputChannels: ["c"], }); await app.invoke({ a: "foo" }); ``` Output: `{ c: ['foofoo', 'foofoofoofoo'] }`

Pregel BinaryOperatorAggregate reducer example

Example showing BinaryOperatorAggregate to implement a custom reducer that concatenates strings with a separator: Python: ```python from langgraph.channels import EphemeralValue, BinaryOperatorAggregate from langgraph.pregel import Pregel, NodeBuilder node1 = ( NodeBuilder().subscribe_only("a") .do(lambda x: x + x) .write_to("b", "c") ) node2 = ( NodeBuilder().subscribe_only("b") .do(lambda x: x + x) .write_to("c") ) def reducer(current, update): if current: return current + " | " + update else: return update app = Pregel( nodes={"node1": node1, "node2": node2}, channels={ "a": EphemeralValue(str), "b": EphemeralValue(str), "c": BinaryOperatorAggregate(str, operator=reducer), }, input_channels=["a"], output_channels=["c"], ) app.invoke({"a": "foo"}) ``` Output: `{ 'c': 'foofoo | foofoofoofoo' }` TypeScript: ```typescript import { EphemeralValue, BinaryOperatorAggregate } from "@langchain/langgraph/channels"; import { Pregel, NodeBuilder } from "@langchain/langgraph/pregel"; const node1 = new NodeBuilder() .subscribeOnly("a") .do((x: string) => x + x) .writeTo("b", "c"); const node2 = new NodeBuilder() .subscribeOnly("b") .do((x: string) => x + x) .writeTo("c"); const reducer = (current: string, update: string) => { if (current) { return current + " | " + update; } else { return update; } }; const app = new Pregel({ nodes: { node1, node2 }, channels: { a: new EphemeralValue<string>(), b: new EphemeralValue<string>(), c: new BinaryOperatorAggregate<string>({ operator: reducer }), }, inputChannels: ["a"], outputChannels: ["c"], }); await app.invoke({ a: "foo" }); ```

Pregel cycle with skip_none example

Example showing how to introduce a cycle in the graph by having a node write to a channel it subscribes to. Execution continues until a None value is written to the channel: Python: ```python from langgraph.channels import EphemeralValue from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry example_node = ( NodeBuilder().subscribe_only("value") .do(lambda x: x + x if len(x) < 10 else None) .write_to(ChannelWriteEntry("value", skip_none=True)) ) app = Pregel( nodes={"example_node": example_node}, channels={ "value": EphemeralValue(str), }, input_channels=["value"], output_channels=["value"], ) app.invoke({"value": "a"}) ``` Output: `{'value': 'aaaaaaaaaaaaaaaa'}` TypeScript: ```typescript import { EphemeralValue } from "@langchain/langgraph/channels"; import { Pregel, NodeBuilder, ChannelWriteEntry } from "@langchain/langgraph/pregel"; const exampleNode = new NodeBuilder() .subscribeOnly("value") .do((x: string) => x.length < 10 ? x + x : null) .writeTo(new ChannelWriteEntry("value", { skipNone: true })); const app = new Pregel({ nodes: { exampleNode }, channels: { value: new EphemeralValue<string>(), }, inputChannels: ["value"], outputChannels: ["value"], }); await app.invoke({ value: "a" }); ``` Output: `{ value: 'aaaaaaaaaaaaaaaa' }`

StateGraph compiles to Pregel instance

The StateGraph (Graph API) is a higher-level abstraction that simplifies creation of Pregel applications. When you compile a StateGraph, it automatically creates the underlying Pregel application. The compiled Pregel instance is associated with a list of nodes and channels that can be inspected via the .nodes and .channels properties.

StateGraph example with add_node and add_edge

Example of creating a StateGraph, adding nodes, and adding edges: Python: ```python from typing import TypedDict from langgraph.constants import START from langgraph.graph import StateGraph class Essay(TypedDict): topic: str content: str | None score: float | None def write_essay(essay: Essay): return { "content": f"Essay about {essay['topic']}", } def score_essay(essay: Essay): return { "score": 10 } builder = StateGraph(Essay) builder.add_node(write_essay) builder.add_node(score_essay) builder.add_edge(START, "write_essay") builder.add_edge("write_essay", "score_essay") graph = builder.compile() ``` TypeScript: ```typescript import { START, StateGraph } from "@langchain/langgraph"; interface Essay { topic: string; content?: string; score?: number; } const writeEssay = (essay: Essay) => { return { content: `Essay about ${essay.topic}`, }; }; const scoreEssay = (essay: Essay) => { return { score: 10 }; }; const builder = new StateGraph<Essay>({ channels: { topic: null, content: null, score: null, } }) .addNode("writeEssay", writeEssay) .addNode("scoreEssay", scoreEssay) .addEdge(START, "writeEssay") .addEdge("writeEssay", "scoreEssay"); const graph = builder.compile(); ```

Functional API @entrypoint creates Pregel instance

In the Functional API, you can use an @entrypoint decorator to create a Pregel application. The entrypoint decorator allows you to define a function that takes input and returns output. It produces a Pregel instance with inspectable nodes and channels.

Functional API @entrypoint example with checkpointer

Example of using @entrypoint decorator to create a Pregel application: Python: ```python from typing import TypedDict from langgraph.checkpoint.memory import InMemorySaver from langgraph.func import entrypoint class Essay(TypedDict): topic: str content: str | None score: float | None checkpointer = InMemorySaver() @entrypoint(checkpointer=checkpointer) def write_essay(essay: Essay): return { "content": f"Essay about {essay['topic']}", } print("Nodes: ") print(write_essay.nodes) print("Channels: ") print(write_essay.channels) ``` TypeScript: ```typescript import { MemorySaver } from "@langchain/langgraph"; import { entrypoint } from "@langchain/langgraph/func"; interface Essay { topic: string; content?: string; score?: number; } const checkpointer = new MemorySaver(); const writeEssay = entrypoint( { checkpointer, name: "writeEssay" }, async (essay: Essay) => { return { content: `Essay about ${essay.topic}`, }; } ); console.log("Nodes: "); console.log(writeEssay.nodes); console.log("Channels: "); console.log(writeEssay.channels); ```

Database tools for SQL agent - sql_db_list_tables

The sql_db_list_tables tool takes an empty string as input and outputs a comma-separated list of tables in the database. This tool should be called first to verify which tables are available before querying their schemas.

Database tools for SQL agent - sql_db_schema

The sql_db_schema tool takes a comma-separated list of table names as input (for example: table1, table2, table3) and outputs the schema and sample rows for those tables. The tool description warns to verify tables actually exist by calling sql_db_list_tables first, and to use this tool to query correct table fields if encountering unknown column errors.

Database tools for SQL agent - sql_db_query

The sql_db_query tool takes a detailed and correct SQL query as input and outputs a result from the database. If the query is not correct, an error message is returned. The tool description instructs to rewrite and check the query if an error is returned, and if an unknown column error occurs, use sql_db_schema to query the correct table fields.

Resume paused graph execution with Command

After a graph is interrupted for human review, execution can be resumed using LangGraph's Command API. The Command allows accepting the proposed tool call, editing its arguments, or providing alternative instructions before resuming the graph execution.

LangGraph SQL agent vs prebuilt LangChain SQL agent

LangChain offers built-in agent implementations using LangGraph primitives. For deeper customization, agents can be implemented directly in LangGraph. The LangGraph approach allows enforcing a higher degree of control through dedicated nodes for specific tool-calls, whereas the prebuilt agent relies on system prompts to constrain behavior such as instructing the agent to always start with listing tables and always run a query-checker before executing.

SQL agent with LangGraph - full example

A complete SQL agent implementation in LangGraph consists of the following components: (1) an LLM that supports tool-calling, (2) database tools for sql_db_list_tables, sql_db_schema, and sql_db_query, (3) dedicated nodes for listing DB tables, calling the schema tool, generating queries, and checking queries, (4) conditional edges that route to the query checker if a query is generated or end if no tool calls are present. The graph enforces a higher degree of control than prebuilt agents by using explicit nodes for specific tool-calls rather than relying on system prompts.

Security warning for SQL agents

Building Q&A systems that execute model-generated SQL queries has inherent risks. Database connection permissions should always be scoped as narrowly as possible for the agent's needs. This will mitigate, though not eliminate, the risks of building a model-driven system. Application-specific validation should be added before executing model-generated SQL.

Graph execution flow in email agent example

Graph starts at START -> read_email -> classify_intent -> (branches to search_documentation, bug_tracking, human_review, or draft_response based on classification) -> draft_response -> (routes to human_review or send_reply based on urgency/intent) -> send_reply -> END. Routing decisions happen inside nodes via Command, not graph edges.

Minimal graph structure with dynamic routing in nodes

Define only essential edges (START to first node, fixed routing, last node to END). Dynamic routing happens inside nodes via Command objects. This keeps graph structure minimal and control flow explicit—you can understand what agent does next by looking at current node.

LangGraph agents decompose into nodes, state, and edges

When building an agent with LangGraph, break the process into discrete steps called nodes. Describe decisions and transitions between nodes. Connect nodes through shared state that each node can read from and write to.

Five steps to build a LangGraph agent

Step 1: Map out workflow as discrete steps (identify nodes and connections). Step 2: Identify what each step needs to do (LLM steps, data steps, action steps, user input steps). Step 3: Design your state (raw data only, not formatted text). Step 4: Build your nodes (functions that take state and return updates). Step 5: Wire it together (connect nodes with edges and compile with checkpointer).

Example: calling StateGraph from Functional API

Example defining State TypedDict with foo: int, building StateGraph with double node, compiling to graph. Workflow invokes graph.invoke({'foo': x}) and returns transformed result. Shows integration of Graph API and Functional API in same application.

Example: calling another entrypoint multiply subworkflow

Example showing nested entrypoint: multiply entrypoint multiplies two numbers, main entrypoint calls multiply.invoke() with specific inputs and returns result. Demonstrates entrypoint composition and checkpointer inheritance.

Example: task caching with CachePolicy ttl

Example with @task(cache_policy=CachePolicy(ttl=120)) slow_add(x) that sleeps 1 second. When called twice with same input, second call returns cached result. Shows ttl in seconds for cache invalidation.

Calling Graph API graphs from Functional API

The Functional API and Graph API can be used together in the same application as they share the same underlying runtime. You can invoke compiled StateGraph instances from within entrypoint functions using .invoke().

Functional API: key features and decorator-based approach

The Functional API allows you to add LangGraph's key features (persistence, memory, human-in-the-loop, and streaming) to your applications with minimal changes to your existing code. It uses decorators like @entrypoint and @task to define workflows.

Entrypoint decorator: basic usage with checkpointer

The @entrypoint decorator marks a function as the entry point of a workflow. It accepts a checkpointer parameter for persistence. Input is restricted to the first argument of the function; to pass multiple inputs, use a dictionary.

Task decorator: defining reusable workflow steps

The @task decorator marks a function as a reusable task within a workflow. Tasks can be invoked from entrypoints or other tasks, and they support parallel execution and result retrieval via .result() method.

Parallel execution of tasks in Functional API

Tasks can be executed in parallel by invoking them concurrently without waiting, then collecting results. In Python, create a list of futures and call .result() on each. In JavaScript, use Promise.all() to await multiple task invocations.

Calling other entrypoints from entrypoints

You can call other entrypoints from within an entrypoint or task using .invoke(). Child entrypoints without an explicit checkpointer will automatically use the checkpointer from the parent entrypoint.

CachePolicy for task caching

Use @task(cache_policy=CachePolicy(ttl=seconds)) to cache task results. The ttl parameter specifies time-to-live in seconds after which the cache is invalidated. This is useful for avoiding redundant computation when the same task is called multiple times with identical inputs.

Example: simple workflow with is_even task

Example showing @task decorator for is_even(number) returning bool, format_message(is_even) returning str, and @entrypoint workflow combining them. Uses InMemorySaver() for persistence and thread_id config for state management. Demonstrates .result() to get task output.

Example: compose essay with LLM task

Example showing @task compose_essay(topic) using init_chat_model('gpt-3.5-turbo') to generate essay via model.invoke(). @entrypoint workflow calls the task with .result(). Demonstrates LLM integration within Functional API and persistence with InMemorySaver().

Example: parallel LLM calls generating paragraphs

Example showing generate_paragraph task called multiple times in a list comprehension, then collecting results via f.result() loop. Demonstrates parallel execution of IO-bound LLM calls and joining results into single output.

langgraph-docs skill primary use cases

The langgraph-docs skill is used when the user asks about LangGraph, graph agents, state machines, agent orchestration, LangGraph API, or needs LangGraph implementation guidance.

When NOT to use LangGraph

Do not use LangGraph for a simple tool-calling agent (use LangChain agents instead with less boilerplate for common patterns) or for a batteries-included agent with planning and subagents (use Deep Agents instead). LangGraph is the orchestration layer and should only be used when you need fine-grained control over agent behavior.

LangGraph definition and purpose

LangGraph is a low-level orchestration framework and runtime for building, managing, and deploying long-running, stateful agents. It provides durable execution, streaming, human-in-the-loop interactions, and time-travel debugging.

Python installation command

To install LangGraph for Python, run: pip install -U langgraph

JavaScript/TypeScript installation command

To install LangGraph for JavaScript/TypeScript, run: npm install @langchain/langgraph @langchain/core

LangGraph compatibility requirements

LangGraph requires Python 3.10+ or Node.js 22+.

Graph API example with StateGraph and MessagesState

Example showing how to use the Graph API with StateGraph and MessagesState: ```python from langgraph.graph import StateGraph, MessagesState, START, END def my_node(state: MessagesState): return {"messages": [{"role": "ai", "content": "hello world"}]} graph = StateGraph(MessagesState) graph.add_node(my_node) graph.add_edge(START, "my_node") graph.add_edge("my_node", END) graph = graph.compile() result = graph.invoke( {"messages": [{"role": "user", "content": "Hello!"}]} ) ```

Functional API example with entrypoint and task

Example showing how to use the Functional API with entrypoint and task decorators: ```python from langgraph.func import entrypoint, task @task def step_one(input: str) -> str: return f"processed: {input}" @entrypoint() def pipeline(input: str) -> str: return step_one(input).result() ```

compile() method purpose

The compile() method compiles a graph builder into an executable graph.

Graph API vs Functional API usage

Use the Graph API for complex workflows where you need explicit control over graph structure. Use the Functional API for simple linear pipelines.

LangGraph related skills

LangGraph is related to langchain (core building blocks for models, tools, and simple agents), deep-agents (high-level agent harness built on LangGraph), and langsmith (for tracing, evaluating, and deploying LangGraph agents).

When to use LangGraph

Use LangGraph when you need to design custom agent workflows with explicit graph-based control flow, add durable execution so agents survive failures and restarts, implement human-in-the-loop with interrupts and approval steps, build multi-agent systems with state shared across agents, stream intermediate results from long-running agent tasks, or time-travel debug by replaying agent execution from any checkpoint.

LangGraph core components: State, Nodes, Edges

LangGraph models agent workflows as graphs with three key components: State (a shared data structure representing the current snapshot), Nodes (functions that receive state as input, perform computation, and return updated state), and Edges (functions that determine which node to execute next based on current state).

Message passing and super-steps in LangGraph

LangGraph uses message passing to define a general program. When a Node completes, it sends messages along one or more edges to other nodes. The program proceeds in discrete super-steps, where a super-step is a single iteration over graph nodes. Nodes that run in parallel are part of the same super-step, while nodes that run sequentially belong to separate super-steps. Nodes begin in an inactive state, become active when receiving a new message on incoming edges, run their function, then vote to halt by marking themselves inactive at the end of each super-step. Graph execution terminates when all nodes are inactive and no messages are in transit.

StateGraph class

StateGraph is the main graph class to use in LangGraph, parameterized by a user-defined State object.

Graph compilation is required

You must compile your graph before you can use it. Compiling performs basic checks on the structure of the graph (e.g., no orphaned nodes) and is where you can specify runtime arguments like checkpointers and breakpoints. Compilation is done by calling the .compile() method.

TypeScript type utilities for graph definition

LangGraph provides several TypeScript type utilities for better type safety: GraphNode for typing node functions defined outside the graph builder, State.Node shorthand provided by each StateSchema instance, ConditionalEdgeRouter for typing routing functions in conditional edges, and StateSchema.State and StateSchema.Update for extracting the state and update types from a schema.

Runtime context schema definition and usage

When creating a graph, you can specify a context_schema (Python) or contextSchema (JavaScript) for runtime context passed to nodes. This is useful for passing information to nodes that is not part of the graph state, such as dependencies like model name or database connection. In Python, define a dataclass with context fields and pass it as context_schema=ContextSchema to StateGraph. Access context in nodes via Runtime[ContextSchema] parameter and runtime.context property. In JavaScript, define context with Zod schema and pass via StateGraph constructor, then access via config.context in node functions. Pass context at runtime using the context parameter of invoke method.

Recursion limit default values and behavior

The recursion limit sets the maximum number of super-steps a graph can execute during a single execution. In Python, the default recursion limit is 1000 steps (starting in version 1.0.6). In JavaScript, the default recursion limit is 25 steps. Once the limit is reached, LangGraph raises GraphRecursionError. The recursion_limit (Python) or recursionLimit (JavaScript) is a standalone config key and should not be passed inside the configurable key.

Give your agent this brain