diff --git a/docs/docs/concepts/index.md b/docs/docs/concepts/index.md index 83fa55b42..ac8e2e861 100644 --- a/docs/docs/concepts/index.md +++ b/docs/docs/concepts/index.md @@ -30,6 +30,7 @@ The conceptual guide does not cover step-by-step instructions or specific implem - [Streaming](streaming.md): Streaming is crucial for enhancing the responsiveness of applications built on LLMs. By displaying output progressively, even before a complete response is ready, streaming significantly improves user experience (UX), particularly when dealing with the latency of LLMs. - [Functional API](functional_api.md): `@entrypoint` and `@task` decorators that allow you to add LangGraph functionality to an existing codebase. - [Durable Execution](durable_execution.md): LangGraph's built-in [persistence](./persistence.md) layer provides durable execution for workflows, ensuring that the state of each execution step is saved to a durable store. +- [Pregel](pregel.md): Pregel is LangGraph's runtime, which is responsible for managing the execution of LangGraph applications. - [FAQ](faq.md): Frequently asked questions about LangGraph. ## LangGraph Platform diff --git a/docs/docs/concepts/pregel.md b/docs/docs/concepts/pregel.md new file mode 100644 index 000000000..1d4a1f2e7 --- /dev/null +++ b/docs/docs/concepts/pregel.md @@ -0,0 +1,347 @@ +# LangGraph's Runtime (Pregel) + +[Pregel][langgraph.pregel.Pregel] implements LangGraph's runtime, managing the execution of LangGraph applications. + +Compiling a [StateGraph][langgraph.graph.StateGraph] or creating an [entrypoint][langgraph.func.entrypoint] produces a [Pregel][langgraph.pregel.Pregel] instance that can be invoked with input. + +This guide explains the runtime at a high level and provides instructions for directly implementing applications with Pregel. + +> **Note:** The [Pregel][langgraph.pregel.Pregel] runtime is named after [Google's Pregel algorithm](https://research.google/pubs/pub37252/), which describes an efficient method for large-scale parallel computation using graphs. + +## Overview + +In LangGraph, Pregel combines [**actors**](https://en.wikipedia.org/wiki/Actor_model) and **channels** into a single application. **Actors** read data from channels and write data to channels. Pregel organizes the execution of the application into multiple steps, following the **Pregel Algorithm**/**Bulk Synchronous Parallel** model. + +Each step consists of three phases: + +- **Plan**: Determine which **actors** to execute in this step. For example, in the first step, select the **actors** that subscribe to the special **input** channels; in subsequent steps, select the **actors** that subscribe to channels updated in the previous step. +- **Execution**: Execute all selected **actors** in parallel, until all complete, or one fails, or a timeout is reached. During this phase, channel updates are invisible to actors until the next step. +- **Update**: Update the channels with the values written by the **actors** in this step. + +Repeat until no **actors** are selected for execution, or a maximum number of steps is reached. + +## Actors + +An **actor** is a [PregelNode][langgraph.pregel.read.PregelNode]. It subscribes to channels, reads data from them, and writes data to them. It can be thought of as an **actor** in the Pregel algorithm. [PregelNodes][langgraph.pregel.read.PregelNode] implement LangChain's Runnable interface. + +## Channels + +Channels are used to communicate between actors (PregelNodes). Each channel has a value type, an update type, and an update function – which takes a sequence of updates and modifies the stored value. Channels can be used to send data from one chain to another, or to send data from a chain to itself in a future step. LangGraph provides a number of built-in channels: + +### Basic channels: LastValue and Topic + +- [LastValue][langgraph.channels.LastValue]: The default channel, stores the last value sent to the channel, useful for input and output values, or for sending data from one step to the next. +- [Topic][langgraph.channels.Topic]: A configurable PubSub Topic, useful for sending multiple values between **actors**, or for accumulating output. Can be configured to deduplicate values or to accumulate values over the course of multiple steps. + +### Advanced channels: Context and BinaryOperatorAggregate + +- `Context`: exposes the value of a context manager, managing its lifecycle. Useful for accessing external resources that require setup and/or teardown; e.g., `client = Context(httpx.Client)`. +- [BinaryOperatorAggregate][langgraph.channels.BinaryOperatorAggregate]: stores a persistent value, updated by applying a binary operator to the current value and each update sent to the channel, useful for computing aggregates over multiple steps; e.g.,`total = BinaryOperatorAggregate(int, operator.add)` + +## Examples + +While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or +the [entrypoint][langgraph.func.entrypoint] decorator, it is possible to interact with Pregel directly. + +Below are a few different examples to give you a sense of the Pregel API. + +=== "Single node" + + ```python + + from langgraph.channels import EphemeralValue + from langgraph.pregel import Pregel, Channel + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | Channel.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"}) + ``` + + ```con + {'b': 'foofoo'} + ``` + +=== "Multiple nodes" + + ```python + from langgraph.channels import LastValue, EphemeralValue + from langgraph.pregel import Pregel, Channel + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | Channel.write_to("b") + ) + + node2 = ( + Channel.subscribe_to("b") + | (lambda x: x + x) + | Channel.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"}) + ``` + + ```con + {'b': 'foofoo', 'c': 'foofoofoofoo'} + ``` + +=== "Topic" + + ```python + from langgraph.channels import EphemeralValue, Topic + from langgraph.pregel import Pregel, Channel + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | { + "b": Channel.write_to("b"), + "c": Channel.write_to("c") + } + ) + + node2 = ( + Channel.subscribe_to("b") + | (lambda x: x + x) + | { + "c": Channel.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"}) + ``` + + ```pycon + {'c': ['foofoo', 'foofoofoofoo']} + ``` + +=== "BinaryOperatorAggregate" + + This examples demonstrates how to use the BinaryOperatorAggregate channel to implement a reducer. + + ```python + from langgraph.channels import EphemeralValue, BinaryOperatorAggregate + from langgraph.pregel import Pregel, Channel + + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | { + "b": Channel.write_to("b"), + "c": Channel.write_to("c") + } + ) + + node2 = ( + Channel.subscribe_to("b") + | (lambda x: x + x) + | { + "c": Channel.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"}) + ``` + + +=== "Cycle" + + This example demonstrates how to introduce a cycle in the graph, by having + a chain write to a channel it subscribes to. Execution will continue + until a None value is written to the channel. + + ```python + from langgraph.channels import EphemeralValue + from langgraph.pregel import Pregel, Channel, ChannelWrite, ChannelWriteEntry + + example_node = ( + Channel.subscribe_to("value") + | (lambda x: x + x if len(x) < 10 else None) + | ChannelWrite(writes=[ChannelWriteEntry(channel="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"}) + ``` + + ```pycon + {'value': 'aaaaaaaaaaaaaaaa'} + ``` + +## High-level API + +LangGraph provides two high-level APIs for creating a Pregel application: the [StateGraph (Graph API)](./low_level.md) and the [Functional API](functional_api.md). + + +=== "StateGraph (Graph API)" + + The [StateGraph (Graph API)][langgraph.graph.StateGraph] is a higher-level abstraction that simplifies the creation of Pregel applications. It allows you to define a graph of nodes and edges. When you compile the graph, the StateGraph API automatically creates the Pregel application for you. + + ```python + from typing import TypedDict, Optional + + from langgraph.constants import START + from langgraph.graph import StateGraph + + class Essay(TypedDict): + topic: str + content: Optional[str] + score: Optional[float] + + 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") + + # Compile the graph. + # This will return a Pregel instance. + graph = builder.compile() + ``` + + The compiled Pregel instance will be associated with a list of nodes and channels. You can inspect the nodes and channels by printing them. + + ```python + print(graph.nodes) + ``` + + You will see something like this: + + ```pycon + {'__start__': , + 'write_essay': , + 'score_essay': } + ``` + + ```python + print(graph.channels) + ``` + + You should see something like this + + ```pycon + {'topic': , + 'content': , + 'score': , + '__start__': , + 'write_essay': , + 'score_essay': , + 'branch:__start__:__self__:write_essay': , + 'branch:__start__:__self__:score_essay': , + 'branch:write_essay:__self__:write_essay': , + 'branch:write_essay:__self__:score_essay': , + 'branch:score_essay:__self__:write_essay': , + 'branch:score_essay:__self__:score_essay': , + 'start:write_essay': } + ``` + +=== "Functional API" + + In the [Functional API](functional_api.md), you can use an [`entrypoint`][langgraph.func.entrypoint] to create + a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output. + + ```python + from typing import TypedDict, Optional + + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.func import entrypoint + + class Essay(TypedDict): + topic: str + content: Optional[str] + score: Optional[float] + + + 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) + ``` + + ```pycon + Nodes: + {'write_essay': } + Channels: + {'__start__': , '__end__': , '__previous__': } + ``` \ No newline at end of file diff --git a/docs/docs/reference/pregel.md b/docs/docs/reference/pregel.md index a7374fb7a..c1896ff20 100644 --- a/docs/docs/reference/pregel.md +++ b/docs/docs/reference/pregel.md @@ -1,9 +1,7 @@ -::: langgraph.pregel.Pregel +# Pregel + +::: langgraph.pregel options: members: - - stream - - astream - - invoke - - ainvoke - - update_state - - aupdate_state + - Pregel + - PregelNode \ No newline at end of file diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index 2fb9a4350..68b2d1a83 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -271,6 +271,7 @@ nav: - concepts/streaming.md - concepts/functional_api.md - concepts/durable_execution.md + - concepts/pregel.md - LangGraph Platform: - LangGraph Platform: concepts#langgraph-platform - High Level: diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index d433fe4d7..df6784e67 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -200,10 +200,42 @@ class Channel: class Pregel(PregelProtocol): """Pregel manages the runtime behavior for LangGraph applications. + ## Overview + + Pregel combines [**actors**](https://en.wikipedia.org/wiki/Actor_model) + and **channels** into a single application. + **Actors** read data from channels and write data to channels. + Pregel organizes the execution of the application into multiple steps, + following the **Pregel Algorithm**/**Bulk Synchronous Parallel** model. + + Each step consists of three phases: + + - **Plan**: Determine which **actors** to execute in this step. For example, + in the first step, select the **actors** that subscribe to the special + **input** channels; in subsequent steps, + select the **actors** that subscribe to channels updated in the previous step. + - **Execution**: Execute all selected **actors** in parallel, + until all complete, or one fails, or a timeout is reached. During this + phase, channel updates are invisible to actors until the next step. + - **Update**: Update the channels with the values written by the **actors** + in this step. + + Repeat until no **actors** are selected for execution, or a maximum number of + steps is reached. + + ## Actors + + An **actor** is a [PregelNode][langgraph.pregel.read.PregelNode]. + It subscribes to channels, reads data from them, and writes data to them. + It can be thought of as an **actor** in the Pregel algorithm. + [PregelNodes][langgraph.pregel.read.PregelNode] implement LangChain's + Runnable interface. + ## Channels - Channels are used to communicate between chains. Each channel has a value type, - an update type, and an update function – which takes a sequence of updates and + Channels are used to communicate between actors (PregelNodes). + Each channel has a value type, an update type, and an update function – which + takes a sequence of updates and modifies the stored value. Channels can be used to send data from one chain to another, or to send data from a chain to itself in a future step. LangGraph provides a number of built-in channels: @@ -213,7 +245,7 @@ class Pregel(PregelProtocol): - `LastValue`: The default channel, stores the last value sent to the channel, useful for input and output values, or for sending data from one step to the next - `Topic`: A configurable PubSub Topic, useful for sending multiple values - between chains, or for accumulating output. Can be configured to deduplicate + between *actors*, or for accumulating output. Can be configured to deduplicate values, and/or to accumulate values over the course of multiple steps. ### Advanced channels: Context and BinaryOperatorAggregate @@ -226,30 +258,202 @@ class Pregel(PregelProtocol): sent to the channel, useful for computing aggregates over multiple steps. eg. `total = BinaryOperatorAggregate(int, operator.add)` - ## Chains + ## Examples - Chains are LCEL Runnables which subscribe to one or more channels, and write to - one or more channels. Any valid LCEL expression can be used as a chain. Chains - can be combined into a Pregel application, which coordinates the execution of the - chains across multiple steps. + Most users will interact with Pregel via a + [StateGraph (Graph API)][langgraph.graph.StateGraph] or via an + [entrypoint (Functional API)][langgraph.func.entrypoint]. - ## Pregel + However, for **advanced** use cases, Pregel can be used directly. If you're + not sure whether you need to use Pregel directly, then the answer is probably no + – you should use the Graph API or Functional API instead. These are higher-level + interfaces that will compile down to Pregel under the hood. - Pregel combines multiple chains (or actors) into a single application. It - coordinates the execution of the chains across multiple steps, following the - Pregel/Bulk Synchronous Parallel model. Each step consists of three phases: + Here are some examples to give you a sense of how it works: - - **Plan**: Determine which chains to execute in this step, ie. the chains that - subscribe to channels updated in the previous step (or, in the first step, - chains that subscribe to input channels) - - **Execution**: Execute those chains in parallel, until all complete, or one fails, - or a timeout is reached. Any channel updates are invisible to other - chains until the next step. - - **Update**: Update the channels with the values written by the - chains in this step. + Example: Single node application + ```python + from langgraph.channels import EphemeralValue + from langgraph.pregel import Pregel, Channel, ChannelWriteEntry - Repeat until no chains are planned for execution, or a maximum number of steps - is reached. + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | Channel.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"}) + ``` + + ```con + {'b': 'foofoo'} + ``` + + Example: Using multiple nodes and multiple output channels + ```python + from langgraph.channels import LastValue, EphemeralValue + from langgraph.pregel import Pregel, Channel, ChannelWriteEntry + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | Channel.write_to("b") + ) + + node2 = ( + Channel.subscribe_to("b") + | (lambda x: x + x) + | Channel.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"}) + ``` + + ```con + {'b': 'foofoo', 'c': 'foofoofoofoo'} + ``` + + Example: Using a Topic channel + ```python + from langgraph.channels import LastValue, EphemeralValue, Topic + from langgraph.pregel import Pregel, Channel, ChannelWriteEntry + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | { + "b": Channel.write_to("b"), + "c": Channel.write_to("c") + } + ) + + node2 = ( + Channel.subscribe_to("b") + | (lambda x: x + x) + | { + "c": Channel.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"}) + ``` + + ```pycon + {'c': ['foofoo', 'foofoofoofoo']} + ``` + + Example: Using a BinaryOperatorAggregate channel + ```python + from langgraph.channels import EphemeralValue, BinaryOperatorAggregate + from langgraph.pregel import Pregel, Channel + + + node1 = ( + Channel.subscribe_to("a") + | (lambda x: x + x) + | { + "b": Channel.write_to("b"), + "c": Channel.write_to("c") + } + ) + + node2 = ( + Channel.subscribe_to("b") + | (lambda x: x + x) + | { + "c": Channel.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"}) + ``` + + ```con + {'c': 'foofoo | foofoofoofoo'} + ``` + + Example: Introducing a cycle + This example demonstrates how to introduce a cycle in the graph, by having + a chain write to a channel it subscribes to. Execution will continue + until a None value is written to the channel. + + ```python + from langgraph.channels import EphemeralValue + from langgraph.pregel import Pregel, Channel, ChannelWrite, ChannelWriteEntry + + example_node = ( + Channel.subscribe_to("value") + | (lambda x: x + x if len(x) < 10 else None) + | ChannelWrite(writes=[ChannelWriteEntry(channel="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"}) + ``` + + ```con + {'value': 'aaaaaaaaaaaaaaaa'} + ``` """ nodes: dict[str, PregelNode]