mirror of
https://github.com/aljazceru/Tutorial-Codebase-Knowledge.git
synced 2025-12-19 15:34:23 +01:00
init push
This commit is contained in:
zachary62
306
docs/LangGraph/01_graph___stategraph.md
Normal file
306
docs/LangGraph/01_graph___stategraph.md
Normal file
@@ -0,0 +1,306 @@
|
||||
# Chapter 1: Graph / StateGraph - The Blueprint of Your Application
|
||||
|
||||
Welcome to the LangGraph tutorial! We're excited to help you learn how to build powerful, stateful applications with Large Language Models (LLMs).
|
||||
|
||||
Imagine you're building an application, maybe a chatbot, an agent that performs tasks, or something that processes data in multiple steps. As these applications get more complex, just calling an LLM once isn't enough. You need a way to structure the flow – maybe call an LLM, then a tool, then another LLM based on the result. How do you manage this sequence of steps and the information passed between them?
|
||||
|
||||
That's where **Graphs** come in!
|
||||
|
||||
## What Problem Do Graphs Solve?
|
||||
|
||||
Think of a complex task like baking a cake. You don't just throw all the ingredients in the oven. There's a sequence: mix dry ingredients, mix wet ingredients, combine them, pour into a pan, bake, cool, frost. Each step depends on the previous one.
|
||||
|
||||
LangGraph helps you define these steps and the order they should happen in. It provides a way to create a **flowchart** or a **blueprint** for your application's logic.
|
||||
|
||||
The core idea is to break down your application into:
|
||||
|
||||
1. **Nodes:** These are the individual steps or actions (like "mix dry ingredients" or "call the LLM").
|
||||
2. **Edges:** These are the connections or transitions between the steps, defining the order (after mixing dry ingredients, mix wet ingredients).
|
||||
|
||||
LangGraph provides different types of graphs, but the most common and useful one for building stateful applications is the `StateGraph`.
|
||||
|
||||
## Core Concepts: `Graph`, `StateGraph`, and `MessageGraph`
|
||||
|
||||
Let's look at the main types of graphs you'll encounter:
|
||||
|
||||
1. **`Graph` (The Basic Blueprint)**
|
||||
* This is the most fundamental type. You define nodes (steps) and edges (connections).
|
||||
* It's like a basic flowchart diagram.
|
||||
* You explicitly define how information passes from one node to the next.
|
||||
* While foundational, you'll often use the more specialized `StateGraph` for convenience.
|
||||
|
||||
```python
|
||||
# This is a conceptual example - we usually use StateGraph
|
||||
from langgraph.graph import Graph
|
||||
|
||||
# Define simple functions or Runnables as nodes
|
||||
def step_one(input_data):
|
||||
print("Running Step 1")
|
||||
return input_data * 2
|
||||
|
||||
def step_two(processed_data):
|
||||
print("Running Step 2")
|
||||
return processed_data + 5
|
||||
|
||||
# Create a basic graph
|
||||
basic_graph_builder = Graph()
|
||||
|
||||
# Add nodes
|
||||
basic_graph_builder.add_node("A", step_one)
|
||||
basic_graph_builder.add_node("B", step_two)
|
||||
|
||||
# Add edges (connections)
|
||||
basic_graph_builder.add_edge("A", "B") # Run B after A
|
||||
basic_graph_builder.set_entry_point("A") # Start at A
|
||||
# basic_graph_builder.set_finish_point("B") # Not needed for this simple Graph type
|
||||
```
|
||||
|
||||
2. **`StateGraph` (The Collaborative Whiteboard)**
|
||||
* This is the workhorse for most LangGraph applications. It's a specialized `Graph`.
|
||||
* **Key Idea:** Nodes communicate *implicitly* by reading from and writing to a shared **State** object.
|
||||
* **Analogy:** Imagine a central whiteboard (the State). Each node (person) can read what's on the whiteboard, do some work, and then update the whiteboard with new information or changes.
|
||||
* You define the *structure* of this shared state first (e.g., what keys it holds).
|
||||
* Each node receives the *current* state and returns a *dictionary* containing only the parts of the state it wants to *update*. LangGraph handles merging these updates into the main state.
|
||||
|
||||
3. **`MessageGraph` (The Chatbot Specialist)**
|
||||
* This is a further specialization of `StateGraph`, designed specifically for building chatbots or conversational agents.
|
||||
* It automatically manages a `messages` list within its state.
|
||||
* Nodes typically take the current list of messages and return new messages to be added.
|
||||
* It uses a special function (`add_messages`) to append messages while handling potential duplicates or updates based on message IDs. This makes building chat flows much simpler.
|
||||
|
||||
For the rest of this chapter, we'll focus on `StateGraph` as it introduces the core concepts most clearly.
|
||||
|
||||
## Building a Simple `StateGraph`
|
||||
|
||||
Let's build a tiny application that takes a number, adds 1 to it, and then multiplies it by 2.
|
||||
|
||||
**Step 1: Define the State**
|
||||
|
||||
First, we define the "whiteboard" – the structure of the data our graph will work with. We use Python's `TypedDict` for this.
|
||||
|
||||
```python
|
||||
from typing import TypedDict
|
||||
|
||||
class MyState(TypedDict):
|
||||
# Our state will hold a single number called 'value'
|
||||
value: int
|
||||
```
|
||||
|
||||
This tells our `StateGraph` that the shared information will always contain an integer named `value`.
|
||||
|
||||
**Step 2: Define the Nodes**
|
||||
|
||||
Nodes are functions (or LangChain Runnables) that perform the work. They take the current `State` as input and return a dictionary containing the *updates* to the state.
|
||||
|
||||
```python
|
||||
# Node 1: Adds 1 to the value
|
||||
def add_one(state: MyState) -> dict:
|
||||
print("--- Running Adder Node ---")
|
||||
current_value = state['value']
|
||||
new_value = current_value + 1
|
||||
print(f"Input value: {current_value}, Output value: {new_value}")
|
||||
# Return *only* the key we want to update
|
||||
return {"value": new_value}
|
||||
|
||||
# Node 2: Multiplies the value by 2
|
||||
def multiply_by_two(state: MyState) -> dict:
|
||||
print("--- Running Multiplier Node ---")
|
||||
current_value = state['value']
|
||||
new_value = current_value * 2
|
||||
print(f"Input value: {current_value}, Output value: {new_value}")
|
||||
# Return the update
|
||||
return {"value": new_value}
|
||||
```
|
||||
|
||||
Notice how each function takes `state` and returns a `dict` specifying which part of the state (`"value"`) should be updated and with what new value.
|
||||
|
||||
**Step 3: Create the Graph and Add Nodes/Edges**
|
||||
|
||||
Now we assemble our blueprint using `StateGraph`.
|
||||
|
||||
```python
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
|
||||
# Create a StateGraph instance linked to our state definition
|
||||
workflow = StateGraph(MyState)
|
||||
|
||||
# Add the nodes to the graph
|
||||
workflow.add_node("adder", add_one)
|
||||
workflow.add_node("multiplier", multiply_by_two)
|
||||
|
||||
# Set the entry point --> where does the flow start?
|
||||
workflow.set_entry_point("adder")
|
||||
|
||||
# Add edges --> how do the nodes connect?
|
||||
workflow.add_edge("adder", "multiplier") # After adder, run multiplier
|
||||
|
||||
# Set the finish point --> where does the flow end?
|
||||
# We use the special identifier END
|
||||
workflow.add_edge("multiplier", END)
|
||||
```
|
||||
|
||||
* `StateGraph(MyState)`: Creates the graph, telling it to use our `MyState` structure.
|
||||
* `add_node("name", function)`: Registers our functions as steps in the graph with unique names.
|
||||
* `set_entry_point("adder")`: Specifies that the `adder` node should run first. This implicitly creates an edge from a special `START` point to `adder`.
|
||||
* `add_edge("adder", "multiplier")`: Creates a connection. After `adder` finishes, `multiplier` will run.
|
||||
* `add_edge("multiplier", END)`: Specifies that after `multiplier` finishes, the graph execution should stop. `END` is a special marker for the graph's conclusion.
|
||||
|
||||
**Step 4: Compile the Graph**
|
||||
|
||||
Before we can run it, we need to `compile` the graph. This finalizes the structure and makes it executable.
|
||||
|
||||
```python
|
||||
# Compile the workflow into an executable object
|
||||
app = workflow.compile()
|
||||
```
|
||||
|
||||
**Step 5: Run It!**
|
||||
|
||||
Now we can invoke our compiled graph (`app`) with some initial state.
|
||||
|
||||
```python
|
||||
# Define the initial state
|
||||
initial_state = {"value": 5}
|
||||
|
||||
# Run the graph
|
||||
final_state = app.invoke(initial_state)
|
||||
|
||||
# Print the final result
|
||||
print("\n--- Final State ---")
|
||||
print(final_state)
|
||||
```
|
||||
|
||||
**Expected Output:**
|
||||
|
||||
```text
|
||||
--- Running Adder Node ---
|
||||
Input value: 5, Output value: 6
|
||||
--- Running Multiplier Node ---
|
||||
Input value: 6, Output value: 12
|
||||
|
||||
--- Final State ---
|
||||
{'value': 12}
|
||||
```
|
||||
|
||||
As you can see, the graph executed the nodes in the defined order (`adder` then `multiplier`), automatically passing the updated state between them!
|
||||
|
||||
## How Does `StateGraph` Work Under the Hood?
|
||||
|
||||
You defined the nodes and edges, but what actually happens when you call `invoke()`?
|
||||
|
||||
1. **Initialization:** LangGraph takes your initial input (`{"value": 5}`) and puts it onto the "whiteboard" (the internal state).
|
||||
2. **Execution Engine:** A powerful internal component called the [Pregel Execution Engine](05_pregel_execution_engine.md) takes over. It looks at the current state and the graph structure.
|
||||
3. **Following Edges:** It starts at the `START` node and follows the edge to the entry point (`adder`).
|
||||
4. **Node Execution:** It runs the `adder` function, passing it the current state (`{"value": 5}`).
|
||||
5. **State Update:** The `adder` function returns `{"value": 6}`. The Pregel engine uses special mechanisms called [Channels](03_channels.md) to update the value associated with the `"value"` key on the "whiteboard". The state is now `{"value": 6}`.
|
||||
6. **Next Step:** The engine sees the edge from `adder` to `multiplier`.
|
||||
7. **Node Execution:** It runs the `multiplier` function, passing it the *updated* state (`{"value": 6}`).
|
||||
8. **State Update:** `multiplier` returns `{"value": 12}`. The engine updates the state again via the [Channels](03_channels.md). The state is now `{"value": 12}`.
|
||||
9. **Following Edges:** The engine sees the edge from `multiplier` to `END`.
|
||||
10. **Finish:** Reaching `END` signals the execution is complete. The final state (`{"value": 12}`) is returned.
|
||||
|
||||
Here's a simplified visual:
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant User
|
||||
participant App (CompiledGraph)
|
||||
participant State
|
||||
participant AdderNode as adder
|
||||
participant MultiplierNode as multiplier
|
||||
|
||||
User->>App: invoke({"value": 5})
|
||||
App->>State: Initialize state = {"value": 5}
|
||||
App->>AdderNode: Execute(state)
|
||||
AdderNode->>State: Read value (5)
|
||||
AdderNode-->>App: Return {"value": 6}
|
||||
App->>State: Update state = {"value": 6}
|
||||
App->>MultiplierNode: Execute(state)
|
||||
MultiplierNode->>State: Read value (6)
|
||||
MultiplierNode-->>App: Return {"value": 12}
|
||||
App->>State: Update state = {"value": 12}
|
||||
App->>User: Return final state {"value": 12}
|
||||
```
|
||||
|
||||
Don't worry too much about the details of Pregel or Channels yet – we'll cover them in later chapters. The key takeaway is that `StateGraph` manages the state and orchestrates the execution based on your defined nodes and edges.
|
||||
|
||||
## A Peek at the Code (`graph/state.py`, `graph/graph.py`)
|
||||
|
||||
Let's briefly look at the code snippets provided to see how these concepts map to the implementation:
|
||||
|
||||
* **`StateGraph.__init__` (`graph/state.py`)**:
|
||||
```python
|
||||
# Simplified view
|
||||
class StateGraph(Graph):
|
||||
def __init__(self, state_schema: Optional[Type[Any]] = None, ...):
|
||||
super().__init__()
|
||||
# ... stores the state_schema ...
|
||||
self.schema = state_schema
|
||||
# ... analyzes the schema to understand state keys and how to update them ...
|
||||
self._add_schema(state_schema)
|
||||
# ... sets up internal dictionaries for channels, nodes etc. ...
|
||||
```
|
||||
This code initializes the graph, crucially storing the `state_schema` you provide. It analyzes this schema to figure out the "keys" on your whiteboard (like `"value"`) and sets up the internal structures ([Channels](03_channels.md)) needed to manage updates to each key.
|
||||
|
||||
* **`StateGraph.add_node` (`graph/state.py`)**:
|
||||
```python
|
||||
# Simplified view
|
||||
def add_node(self, node: str, action: RunnableLike, ...):
|
||||
# ... basic checks for name conflicts, reserved names (START, END) ...
|
||||
if node in self.channels: # Cannot use a state key name as a node name
|
||||
raise ValueError(...)
|
||||
# ... wrap the provided action (function/runnable) ...
|
||||
runnable = coerce_to_runnable(action, ...)
|
||||
# ... store the node details (runnable, input type etc.) ...
|
||||
self.nodes[node] = StateNodeSpec(runnable, ..., input=input or self.schema, ...)
|
||||
return self
|
||||
```
|
||||
When you add a node, it stores the associated function (`action`) and links it to the provided `node` name. It also figures out what input schema the node expects (usually the main graph state schema).
|
||||
|
||||
* **`Graph.add_edge` (`graph/graph.py`)**:
|
||||
```python
|
||||
# Simplified view from the base Graph class
|
||||
def add_edge(self, start_key: str, end_key: str):
|
||||
# ... checks for invalid edges (e.g., starting from END) ...
|
||||
# ... basic validation ...
|
||||
# Stores the connection as a simple pair
|
||||
self.edges.add((start_key, end_key))
|
||||
return self
|
||||
```
|
||||
Adding an edge is relatively simple – it just records the `(start_key, end_key)` pair in a set, representing the connection.
|
||||
|
||||
* **`StateGraph.compile` (`graph/state.py`)**:
|
||||
```python
|
||||
# Simplified view
|
||||
def compile(self, ...):
|
||||
# ... validation checks ...
|
||||
self.validate(...)
|
||||
# ... create the CompiledStateGraph instance ...
|
||||
compiled = CompiledStateGraph(builder=self, ...)
|
||||
# ... add nodes, edges, branches to the compiled version ...
|
||||
for key, node in self.nodes.items():
|
||||
compiled.attach_node(key, node)
|
||||
for start, end in self.edges:
|
||||
compiled.attach_edge(start, end)
|
||||
# ... more setup for branches, entry/exit points ...
|
||||
# ... finalize and return the compiled graph ...
|
||||
return compiled.validate()
|
||||
```
|
||||
Compilation takes your defined nodes and edges and builds the final, executable `CompiledStateGraph`. It sets up the internal machinery ([Pregel](05_pregel_execution_engine.md), [Channels](03_channels.md)) based on your blueprint.
|
||||
|
||||
## Conclusion
|
||||
|
||||
You've learned the fundamental concept in LangGraph: the **Graph**.
|
||||
|
||||
* Graphs define the structure and flow of your application using **Nodes** (steps) and **Edges** (connections).
|
||||
* **`StateGraph`** is the most common type, where nodes communicate implicitly by reading and updating a shared **State** object (like a whiteboard).
|
||||
* **`MessageGraph`** is a specialized `StateGraph` for easily building chatbots.
|
||||
* You define the state structure, write node functions that update parts of the state, connect them with edges, and `compile` the graph to make it runnable.
|
||||
|
||||
Now that you understand how to define the overall *structure* of your application using `StateGraph`, the next step is to dive deeper into what constitutes a **Node**.
|
||||
|
||||
Let's move on to [Chapter 2: Nodes (`PregelNode`)](02_nodes___pregelnode__.md) to explore how individual steps are defined and executed.
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
224
docs/LangGraph/02_nodes___pregelnode__.md
Normal file
224
docs/LangGraph/02_nodes___pregelnode__.md
Normal file
@@ -0,0 +1,224 @@
|
||||
# Chapter 2: Nodes (`PregelNode`) - The Workers of Your Graph
|
||||
|
||||
In [Chapter 1: Graph / StateGraph](01_graph___stategraph.md), we learned how `StateGraph` acts as a blueprint or a flowchart for our application. It defines the overall structure and the shared "whiteboard" (the State) that holds information.
|
||||
|
||||
But who actually does the work? If the `StateGraph` is the assembly line blueprint, who are the workers on the line?
|
||||
|
||||
That's where **Nodes** come in!
|
||||
|
||||
## What Problem Do Nodes Solve?
|
||||
|
||||
Think back to our cake baking analogy from Chapter 1. We had steps like "mix dry ingredients," "mix wet ingredients," "combine," etc. Each of these distinct actions needs to be performed by someone or something.
|
||||
|
||||
In LangGraph, **Nodes** represent these individual units of work or computation steps within your graph.
|
||||
|
||||
* **Analogy:** Imagine chefs in a kitchen (the graph). Each chef (node) has a specific task: one chops vegetables, another mixes the sauce, another cooks the main course. They all work with shared ingredients (the state) from the pantry and fridge, and they put their finished components back for others to use.
|
||||
|
||||
Nodes are the core building blocks that perform the actual logic of your application.
|
||||
|
||||
## Key Concepts: What Makes a Node?
|
||||
|
||||
1. **The Action:** At its heart, a node is usually a Python function or a LangChain Runnable. This is the code that gets executed when the node runs.
|
||||
2. **Input:** A node typically reads data it needs from the shared graph **State**. It receives the *current* state when it's invoked. In our `StateGraph` example from Chapter 1, both `add_one` and `multiply_by_two` received the `state` dictionary containing the current `value`.
|
||||
3. **Execution:** The node runs its defined logic (the function or Runnable).
|
||||
4. **Output:** After executing, a node in a `StateGraph` returns a dictionary. This dictionary specifies *which parts* of the shared state the node wants to *update* and what the new values should be. LangGraph takes care of merging these updates back into the main state.
|
||||
|
||||
## Adding Nodes to Your Graph (`add_node`)
|
||||
|
||||
How do we tell our `StateGraph` about these workers? We use the `add_node` method.
|
||||
|
||||
Let's revisit the code from Chapter 1:
|
||||
|
||||
**Step 1: Define the Node Functions**
|
||||
|
||||
These are our "workers". They take the state and return updates.
|
||||
|
||||
```python
|
||||
from typing import TypedDict
|
||||
|
||||
# Define the state structure (the whiteboard)
|
||||
class MyState(TypedDict):
|
||||
value: int
|
||||
|
||||
# Node 1: Adds 1 to the value
|
||||
def add_one(state: MyState) -> dict:
|
||||
print("--- Running Adder Node ---")
|
||||
current_value = state['value']
|
||||
new_value = current_value + 1
|
||||
print(f"Input value: {current_value}, Output value: {new_value}")
|
||||
# Return *only* the key we want to update
|
||||
return {"value": new_value}
|
||||
|
||||
# Node 2: Multiplies the value by 2
|
||||
def multiply_by_two(state: MyState) -> dict:
|
||||
print("--- Running Multiplier Node ---")
|
||||
current_value = state['value']
|
||||
new_value = current_value * 2
|
||||
print(f"Input value: {current_value}, Output value: {new_value}")
|
||||
# Return the update
|
||||
return {"value": new_value}
|
||||
```
|
||||
|
||||
**Step 2: Create the Graph and Add Nodes**
|
||||
|
||||
Here's where we hire our workers and assign them names on the assembly line.
|
||||
|
||||
```python
|
||||
from langgraph.graph import StateGraph
|
||||
|
||||
# Create the graph builder linked to our state
|
||||
workflow = StateGraph(MyState)
|
||||
|
||||
# Add the first node:
|
||||
# Give it the name "adder" and tell it to use the 'add_one' function
|
||||
workflow.add_node("adder", add_one)
|
||||
|
||||
# Add the second node:
|
||||
# Give it the name "multiplier" and tell it to use the 'multiply_by_two' function
|
||||
workflow.add_node("multiplier", multiply_by_two)
|
||||
|
||||
# (Edges like set_entry_point, add_edge, etc. define the flow *between* nodes)
|
||||
# ... add edges and compile ...
|
||||
```
|
||||
|
||||
* `workflow.add_node("adder", add_one)`: This line registers the `add_one` function as a node within the `workflow` graph. We give it the unique name `"adder"`. When the graph needs to execute the "adder" step, it will call our `add_one` function.
|
||||
* `workflow.add_node("multiplier", multiply_by_two)`: Similarly, this registers the `multiply_by_two` function under the name `"multiplier"`.
|
||||
|
||||
It's that simple! You define what a step does (the function) and then register it with `add_node`, giving it a name so you can connect it using edges later.
|
||||
|
||||
## How Do Nodes Actually Run? (Under the Hood)
|
||||
|
||||
You've defined the functions and added them as nodes. What happens internally when the graph executes?
|
||||
|
||||
1. **Triggering:** The [Pregel Execution Engine](05_pregel_execution_engine.md) (LangGraph's internal coordinator) determines which node should run next based on the graph's structure (edges) and the current state. For example, after the `START` point, it knows to run the entry point node ("adder" in our example).
|
||||
2. **Reading State:** Before running the node's function (`add_one`), the engine reads the necessary information from the shared state. It knows what the function needs (the `MyState` dictionary). This reading happens via mechanisms called [Channels](03_channels.md), which manage the shared state.
|
||||
3. **Invoking the Function:** The engine calls the node's function (e.g., `add_one`), passing the state it just read (`{'value': 5}`).
|
||||
4. **Executing Logic:** Your function's code runs (e.g., `5 + 1`).
|
||||
5. **Receiving Updates:** The engine receives the dictionary returned by the function (e.g., `{'value': 6}`).
|
||||
6. **Writing State:** The engine uses [Channels](03_channels.md) again to update the shared state with the information from the returned dictionary. The state on the "whiteboard" is now modified (e.g., becomes `{'value': 6}`).
|
||||
7. **Next Step:** The engine then looks for the next edge originating from the completed node ("adder") to determine what runs next ("multiplier").
|
||||
|
||||
Here's a simplified view of the "adder" node executing:
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Engine as Pregel Engine
|
||||
participant State (via Channels)
|
||||
participant AdderNode as adder (add_one func)
|
||||
|
||||
Engine->>State (via Channels): Read 'value' (current state is {'value': 5})
|
||||
State (via Channels)-->>Engine: Returns {'value': 5}
|
||||
Engine->>AdderNode: Invoke add_one({'value': 5})
|
||||
Note over AdderNode: Function executes: 5 + 1 = 6
|
||||
AdderNode-->>Engine: Return {'value': 6}
|
||||
Engine->>State (via Channels): Write update: 'value' = 6
|
||||
State (via Channels)-->>Engine: Acknowledge (state is now {'value': 6})
|
||||
Engine->>Engine: Find next node based on edge from "adder"
|
||||
```
|
||||
|
||||
## A Peek at the Code (`graph/state.py`, `pregel/read.py`)
|
||||
|
||||
Let's look at simplified snippets to see how this maps to the code:
|
||||
|
||||
* **`StateGraph.add_node` (`graph/state.py`)**:
|
||||
```python
|
||||
# Simplified view
|
||||
class StateGraph(Graph):
|
||||
# ... (other methods) ...
|
||||
def add_node(
|
||||
self,
|
||||
node: str, # The name you give the node (e.g., "adder")
|
||||
action: RunnableLike, # The function or Runnable (e.g., add_one)
|
||||
*,
|
||||
# ... other optional parameters ...
|
||||
input: Optional[Type[Any]] = None, # Optional: specific input type for this node
|
||||
) -> Self:
|
||||
# ... (checks for valid name, etc.) ...
|
||||
if node in self.channels: # Can't use a state key name as a node name
|
||||
raise ValueError(...)
|
||||
|
||||
# Converts your function into a standard LangChain Runnable if needed
|
||||
runnable = coerce_to_runnable(action, ...)
|
||||
|
||||
# Stores the node's details, including the runnable and input schema
|
||||
self.nodes[node] = StateNodeSpec(
|
||||
runnable=runnable,
|
||||
metadata=None, # Optional metadata
|
||||
input=input or self.schema, # Default to graph's main state schema
|
||||
# ... other details ...
|
||||
)
|
||||
return self
|
||||
```
|
||||
When you call `add_node`, LangGraph stores your function (`action`) under the given `node` name. It wraps your function into a standard `Runnable` object (`coerce_to_runnable`) and keeps track of what input schema it expects (usually the graph's main state schema). This stored information is a `StateNodeSpec`.
|
||||
|
||||
* **`CompiledStateGraph.attach_node` (`graph/state.py`)**:
|
||||
```python
|
||||
# Simplified view (during graph.compile())
|
||||
class CompiledStateGraph(CompiledGraph):
|
||||
# ... (other methods) ...
|
||||
def attach_node(self, key: str, node: Optional[StateNodeSpec]) -> None:
|
||||
# ... (handles START node specially) ...
|
||||
if node is not None:
|
||||
# Determine what parts of the state this node needs to read
|
||||
input_schema = node.input
|
||||
input_values = list(self.builder.schemas[input_schema]) # Keys to read
|
||||
|
||||
# Create the internal representation: PregelNode
|
||||
self.nodes[key] = PregelNode(
|
||||
triggers=[f"branch:to:{key}"], # When should this node run? (Connected via Channels)
|
||||
channels=input_values, # What state keys does it read?
|
||||
mapper=_pick_mapper(...), # How to format the input state for the function
|
||||
writers=[ChannelWrite(...)], # How to write the output back to state (via Channels)
|
||||
bound=node.runnable, # The actual function/Runnable to execute!
|
||||
# ... other internal details ...
|
||||
)
|
||||
# ...
|
||||
```
|
||||
During the `compile()` step, the information stored in `StateNodeSpec` is used to create the actual operational node object, which is internally called `PregelNode`. This `PregelNode` is the real "worker" managed by the execution engine.
|
||||
|
||||
* **`PregelNode` (`pregel/read.py`)**:
|
||||
```python
|
||||
# Simplified view
|
||||
class PregelNode(Runnable):
|
||||
channels: Union[list[str], Mapping[str, str]] # State keys to read as input
|
||||
triggers: list[str] # Channel updates that activate this node
|
||||
mapper: Optional[Callable[[Any], Any]] # Function to format input state
|
||||
writers: list[Runnable] # Runnables to write output back to Channels
|
||||
bound: Runnable[Any, Any] # << THE ACTUAL FUNCTION/RUNNABLE YOU PROVIDED >>
|
||||
# ... other attributes like retry policy, tags, etc. ...
|
||||
|
||||
def __init__(self, *, channels, triggers, writers, bound, ...) -> None:
|
||||
self.channels = channels
|
||||
self.triggers = list(triggers)
|
||||
self.writers = writers or []
|
||||
self.bound = bound # Your code lives here!
|
||||
# ... initialize other attributes ...
|
||||
|
||||
# ... (methods for execution, handled by the Pregel engine) ...
|
||||
```
|
||||
The `PregelNode` object encapsulates everything needed to run your node:
|
||||
* `bound`: This holds the actual function or Runnable you passed to `add_node`.
|
||||
* `channels`: Specifies which parts of the state (managed by [Channels](03_channels.md)) to read as input.
|
||||
* `triggers`: Specifies which [Channels](03_channels.md) must be updated to make this node eligible to run.
|
||||
* `writers`: Defines how the output of `bound` should be written back to the state using [Channels](03_channels.md).
|
||||
|
||||
Don't worry too much about `PregelNode` details right now. The key idea is that `add_node` registers your function, and `compile` turns it into an executable component (`PregelNode`) that the graph engine can manage, telling it when to run, what state to read, and how to write results back.
|
||||
|
||||
## Conclusion
|
||||
|
||||
You've now learned about the "workers" in your LangGraph application: **Nodes**.
|
||||
|
||||
* Nodes are the individual computational steps defined by Python functions or LangChain Runnables.
|
||||
* They read from the shared `StateGraph` state.
|
||||
* They execute their logic.
|
||||
* They return dictionaries specifying updates to the state.
|
||||
* You add them to your graph using `graph.add_node("node_name", your_function)`.
|
||||
* Internally, they are represented as `PregelNode` objects, managed by the execution engine.
|
||||
|
||||
We have the blueprint (`StateGraph`) and the workers (`Nodes`). But how exactly does information get passed around? How does the "adder" node's output (`{'value': 6}`) reliably get to the "multiplier" node? How is the state managed efficiently?
|
||||
|
||||
That's the role of [Chapter 3: Channels](03_channels.md), the communication system of the graph.
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
426
docs/LangGraph/03_channels.md
Normal file
426
docs/LangGraph/03_channels.md
Normal file
@@ -0,0 +1,426 @@
|
||||
# Chapter 3: Channels - The Communication System
|
||||
|
||||
In [Chapter 1: Graph / StateGraph](01_graph___stategraph.md), we learned about the `StateGraph` as the blueprint for our application, holding the shared "whiteboard" or state. In [Chapter 2: Nodes (`PregelNode`)](02_nodes___pregelnode__.md), we met the "workers" or Nodes that perform tasks and read/write to this whiteboard.
|
||||
|
||||
But how does this "whiteboard" *actually* work? How does the information written by one node reliably get seen by the next? What happens if multiple nodes try to write to the *same part* of the whiteboard at roughly the same time?
|
||||
|
||||
This is where **Channels** come in. They are the fundamental mechanism for communication and state management within a `StateGraph`.
|
||||
|
||||
## What Problem Do Channels Solve?
|
||||
|
||||
Imagine our simple graph from Chapter 1:
|
||||
|
||||
```python
|
||||
# State: {'value': int}
|
||||
# Node 1: adder (reads 'value', returns {'value': value + 1})
|
||||
# Node 2: multiplier (reads 'value', returns {'value': value * 2})
|
||||
# Flow: START -> adder -> multiplier -> END
|
||||
```
|
||||
|
||||
When `adder` runs with `{'value': 5}`, it returns `{'value': 6}`. How does this update the central state so that `multiplier` receives `{'value': 6}` and not the original `{'value': 5}`?
|
||||
|
||||
Furthermore, what if we had a more complex graph where two different nodes, say `node_A` and `node_B`, both finished their work and *both* wanted to update the `value` key in the same step? Should the final `value` be the one from `node_A`, the one from `node_B`, their sum, or something else?
|
||||
|
||||
**Channels** solve these problems by defining:
|
||||
|
||||
1. **Storage:** How the value for a specific key in the state is stored.
|
||||
2. **Update Logic:** How incoming updates for that key are combined or processed.
|
||||
|
||||
## Channels: Mailboxes for Your State
|
||||
|
||||
Think of the shared state (our "whiteboard") not as one big surface, but as a collection of **mailboxes**.
|
||||
|
||||
* **Each key in your state dictionary (`MyState`) gets its own dedicated mailbox.** In our example, there's a mailbox labeled `"value"`.
|
||||
* When a Node finishes and returns a dictionary (like `{'value': 6}`), the [Pregel Execution Engine](05_pregel_execution_engine.md) acts like a mail carrier. It takes the value `6` and puts it into the mailbox labeled `"value"`.
|
||||
* When another Node needs to read the state, the engine goes to the relevant mailboxes (like `"value"`) and gets the current contents.
|
||||
|
||||
This mailbox concept ensures that updates intended for `"value"` only affect `"value"`, and updates for another key (say, `"messages"`) would go into *its* own separate mailbox.
|
||||
|
||||
**Crucially, each mailbox (Channel) has specific rules about how incoming mail (updates) is handled.** Does the new mail replace the old one? Is it added to a list? Is it mathematically combined with the previous value? These rules are defined by the **Channel Type**.
|
||||
|
||||
## How Channels Work: The Update Cycle
|
||||
|
||||
Here's a step-by-step view of how channels manage state during graph execution:
|
||||
|
||||
1. **Node Returns Update:** A node (e.g., `adder`) finishes and returns a dictionary (e.g., `{'value': 6}`).
|
||||
2. **Engine Routes Update:** The [Pregel Execution Engine](05_pregel_execution_engine.md) sees the key `"value"` and routes the update `6` to the Channel associated with `"value"`.
|
||||
3. **Channel Receives Update(s):** The `"value"` Channel receives `6`. If other nodes also returned updates for `"value"` in the same step, the Channel would receive all of them in a sequence (e.g., `[6, maybe_another_update]`).
|
||||
4. **Channel Applies Update Logic:** The Channel uses its specific rule (its type) to process the incoming update(s). For example, a `LastValue` channel would just keep the *last* update it received in the sequence. A `BinaryOperatorAggregate` channel might *sum* all the updates with its current value.
|
||||
5. **State is Updated:** The Channel now holds the new, processed value.
|
||||
6. **Node Reads State:** When the next node (e.g., `multiplier`) needs the state, the Engine queries the relevant Channels (e.g., the `"value"` Channel).
|
||||
7. **Channel Provides Value:** The Channel provides its current stored value (e.g., `6`) to the Engine, which passes it to the node.
|
||||
|
||||
This ensures that state updates are handled consistently according to predefined rules for each piece of state.
|
||||
|
||||
## Common Channel Types: Defining the Rules
|
||||
|
||||
LangGraph provides several types of Channels, each with different update logic. You usually define which channel type to use for a state key when you define your state `TypedDict`, often using `typing.Annotated`.
|
||||
|
||||
Here are the most common ones:
|
||||
|
||||
1. **`LastValue[T]`** (The Default Overwriter)
|
||||
* **Rule:** Keeps only the **last** value it received. If multiple updates arrive in the same step, the final value is simply the last one in the sequence processed by the engine.
|
||||
* **Analogy:** Like a standard variable assignment (`my_variable = new_value`). The old value is discarded.
|
||||
* **When to Use:** This is the **default** for keys in your `TypedDict` state unless you specify otherwise with `Annotated`. It's perfect for state values that should be replaced entirely, like the current step's result or a user's latest query.
|
||||
* **Code:** `langgraph.channels.LastValue` (from `channels/last_value.py`)
|
||||
|
||||
```python
|
||||
# channels/last_value.py (Simplified)
|
||||
class LastValue(Generic[Value], BaseChannel[Value, Value, Value]):
|
||||
# ... (initializer, etc.)
|
||||
value: Any = MISSING # Stores the single, last value
|
||||
|
||||
def update(self, values: Sequence[Value]) -> bool:
|
||||
if len(values) == 0: # No updates this step
|
||||
return False
|
||||
# If multiple updates in one step, only the last one matters!
|
||||
# Example: if values = [update1, update2], self.value becomes update2
|
||||
self.value = values[-1]
|
||||
return True
|
||||
|
||||
def get(self) -> Value:
|
||||
if self.value is MISSING:
|
||||
raise EmptyChannelError()
|
||||
return self.value # Return the currently stored last value
|
||||
```
|
||||
* **How to Use (Implicitly):**
|
||||
```python
|
||||
from typing import TypedDict
|
||||
|
||||
class MyState(TypedDict):
|
||||
# Because we didn't use Annotated, LangGraph defaults to LastValue[int]
|
||||
value: int
|
||||
user_query: str # Also defaults to LastValue[str]
|
||||
```
|
||||
|
||||
2. **`BinaryOperatorAggregate[T]`** (The Combiner)
|
||||
* **Rule:** Takes an initial "identity" value (like `0` for addition, `1` for multiplication) and a **binary operator** function (e.g., `+`, `*`, `operator.add`). When it receives updates, it applies the operator between its current value and each new update, accumulating the result.
|
||||
* **Analogy:** Like a running total (`total += new_number`).
|
||||
* **When to Use:** Useful for accumulating scores, counts, or combining numerical results.
|
||||
* **Code:** `langgraph.channels.BinaryOperatorAggregate` (from `channels/binop.py`)
|
||||
|
||||
```python
|
||||
# channels/binop.py (Simplified)
|
||||
import operator
|
||||
from typing import Callable
|
||||
|
||||
class BinaryOperatorAggregate(Generic[Value], BaseChannel[Value, Value, Value]):
|
||||
# ... (initializer stores the operator and identity value)
|
||||
value: Any = MISSING
|
||||
operator: Callable[[Value, Value], Value]
|
||||
|
||||
def update(self, values: Sequence[Value]) -> bool:
|
||||
if not values:
|
||||
return False
|
||||
# Start with the first value if the channel was empty
|
||||
if self.value is MISSING:
|
||||
self.value = values[0]
|
||||
values = values[1:]
|
||||
# Apply the operator for all subsequent values
|
||||
for val in values:
|
||||
self.value = self.operator(self.value, val)
|
||||
return True
|
||||
|
||||
def get(self) -> Value:
|
||||
# ... (return self.value, handling MISSING)
|
||||
```
|
||||
* **How to Use (Explicitly with `Annotated`):**
|
||||
```python
|
||||
import operator
|
||||
from typing import TypedDict, Annotated
|
||||
from langgraph.channels import BinaryOperatorAggregate
|
||||
|
||||
class AgentState(TypedDict):
|
||||
# Use Annotated to specify the channel type and operator
|
||||
total_score: Annotated[int, BinaryOperatorAggregate(int, operator.add)]
|
||||
# ^^^ state key 'total_score' will use BinaryOperatorAggregate with addition
|
||||
```
|
||||
|
||||
3. **`Topic[T]`** (The Collector)
|
||||
* **Rule:** Collects all updates it receives into a **list**. By default (`accumulate=False`), it clears the list after each step, so `get()` returns only the updates from the *immediately preceding* step. If `accumulate=True`, it keeps adding to the list across multiple steps.
|
||||
* **Analogy:** Like appending to a log file or a list (`my_list.append(new_item)`).
|
||||
* **When to Use:** Great for gathering messages in a conversation (`MessageGraph` uses this internally!), collecting events, or tracking a sequence of results.
|
||||
* **Code:** `langgraph.channels.Topic` (from `channels/topic.py`)
|
||||
|
||||
```python
|
||||
# channels/topic.py (Simplified)
|
||||
from typing import Sequence, List, Union
|
||||
|
||||
class Topic(Generic[Value], BaseChannel[Sequence[Value], Union[Value, list[Value]], list[Value]]):
|
||||
# ... (initializer sets accumulate flag)
|
||||
values: list[Value]
|
||||
accumulate: bool
|
||||
|
||||
def update(self, updates: Sequence[Union[Value, list[Value]]]) -> bool:
|
||||
old_len = len(self.values)
|
||||
# Clear list if not accumulating
|
||||
if not self.accumulate:
|
||||
self.values = []
|
||||
# Flatten and extend the list with new updates
|
||||
new_values = list(flatten(updates)) # flatten handles list-of-lists
|
||||
self.values.extend(new_values)
|
||||
return len(self.values) != old_len # Return True if list changed
|
||||
|
||||
def get(self) -> Sequence[Value]:
|
||||
# ... (return list(self.values), handling empty)
|
||||
```
|
||||
* **How to Use (Explicitly with `Annotated`):**
|
||||
```python
|
||||
from typing import TypedDict, Annotated, List
|
||||
from langgraph.channels import Topic
|
||||
|
||||
class ChatState(TypedDict):
|
||||
# Use Annotated to specify the Topic channel
|
||||
# The final type hint for the state is List[str]
|
||||
chat_history: Annotated[List[str], Topic(str, accumulate=True)]
|
||||
# ^^^ state key 'chat_history' will use Topic to accumulate strings
|
||||
```
|
||||
|
||||
There are other specialized channels like `EphemeralValue` (clears after reading) and `Context` (allows passing values down without modifying state), but `LastValue`, `BinaryOperatorAggregate`, and `Topic` are the most fundamental.
|
||||
|
||||
## Channels in Action: Our Simple Graph Revisited
|
||||
|
||||
Let's trace our `adder` -> `multiplier` graph again, focusing on the implicit `LastValue` channel for the `"value"` key:
|
||||
|
||||
```python
|
||||
from typing import TypedDict
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
|
||||
# State uses implicit LastValue[int] for 'value'
|
||||
class MyState(TypedDict):
|
||||
value: int
|
||||
|
||||
# Nodes (same as before)
|
||||
def add_one(state: MyState) -> dict:
|
||||
return {"value": state['value'] + 1}
|
||||
|
||||
def multiply_by_two(state: MyState) -> dict:
|
||||
return {"value": state['value'] * 2}
|
||||
|
||||
# Graph setup (same as before)
|
||||
workflow = StateGraph(MyState)
|
||||
workflow.add_node("adder", add_one)
|
||||
workflow.add_node("multiplier", multiply_by_two)
|
||||
workflow.set_entry_point("adder")
|
||||
workflow.add_edge("adder", "multiplier")
|
||||
workflow.add_edge("multiplier", END)
|
||||
app = workflow.compile()
|
||||
|
||||
# Execution with initial state {"value": 5}
|
||||
initial_state = {"value": 5}
|
||||
final_state = app.invoke(initial_state)
|
||||
```
|
||||
|
||||
Here's the flow with the Channel involved:
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant User
|
||||
participant App as CompiledGraph
|
||||
participant Engine as Pregel Engine
|
||||
participant ValueChannel as "value" (LastValue)
|
||||
participant AdderNode as adder
|
||||
participant MultiplierNode as multiplier
|
||||
|
||||
User->>App: invoke({"value": 5})
|
||||
App->>Engine: Start execution
|
||||
Engine->>ValueChannel: Initialize/Set state from input (value = 5)
|
||||
App->>Engine: Entry point is "adder"
|
||||
Engine->>ValueChannel: Read current value (5)
|
||||
ValueChannel-->>Engine: Returns 5
|
||||
Engine->>AdderNode: Execute(state={'value': 5})
|
||||
AdderNode-->>Engine: Return {"value": 6}
|
||||
Engine->>ValueChannel: Update with [6]
|
||||
Note over ValueChannel: LastValue rule: value becomes 6
|
||||
ValueChannel-->>Engine: Acknowledge update
|
||||
Engine->>Engine: Follow edge "adder" -> "multiplier"
|
||||
Engine->>ValueChannel: Read current value (6)
|
||||
ValueChannel-->>Engine: Returns 6
|
||||
Engine->>MultiplierNode: Execute(state={'value': 6})
|
||||
MultiplierNode-->>Engine: Return {"value": 12}
|
||||
Engine->>ValueChannel: Update with [12]
|
||||
Note over ValueChannel: LastValue rule: value becomes 12
|
||||
ValueChannel-->>Engine: Acknowledge update
|
||||
Engine->>Engine: Follow edge "multiplier" -> END
|
||||
Engine->>ValueChannel: Read final value (12)
|
||||
ValueChannel-->>Engine: Returns 12
|
||||
Engine->>App: Execution finished, final state {'value': 12}
|
||||
App->>User: Return final state {'value': 12}
|
||||
```
|
||||
|
||||
The `LastValue` channel ensures that the output of `adder` correctly overwrites the initial state before `multiplier` reads it.
|
||||
|
||||
## Example: Using `BinaryOperatorAggregate` Explicitly
|
||||
|
||||
Let's modify the state to *sum* values instead of overwriting them.
|
||||
|
||||
```python
|
||||
import operator
|
||||
from typing import TypedDict, Annotated
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
# Import the channel type
|
||||
from langgraph.channels import BinaryOperatorAggregate
|
||||
|
||||
# Define state with an explicitly configured channel
|
||||
class SummingState(TypedDict):
|
||||
# Use Annotated to specify the channel and its operator (addition)
|
||||
value: Annotated[int, BinaryOperatorAggregate(int, operator.add)]
|
||||
|
||||
# Node 1: Returns 5 to be ADDED to the current value
|
||||
def add_five(state: SummingState) -> dict:
|
||||
print(f"--- Running Adder Node 1 (current value: {state.get('value', 0)}) ---")
|
||||
# Note: We return the *increment*, not the new total
|
||||
return {"value": 5}
|
||||
|
||||
# Node 2: Returns 10 to be ADDED to the current value
|
||||
def add_ten(state: SummingState) -> dict:
|
||||
print(f"--- Running Adder Node 2 (current value: {state['value']}) ---")
|
||||
# Note: We return the *increment*, not the new total
|
||||
return {"value": 10}
|
||||
|
||||
# Create graph
|
||||
workflow = StateGraph(SummingState)
|
||||
workflow.add_node("adder1", add_five)
|
||||
workflow.add_node("adder2", add_ten)
|
||||
workflow.set_entry_point("adder1")
|
||||
workflow.add_edge("adder1", "adder2")
|
||||
workflow.add_edge("adder2", END)
|
||||
|
||||
app = workflow.compile()
|
||||
|
||||
# Run with initial state value = 0 (BinaryOperatorAggregate defaults int to 0)
|
||||
print("Invoking graph...")
|
||||
# You could also provide an initial value: app.invoke({"value": 100})
|
||||
final_state = app.invoke({})
|
||||
|
||||
print("\n--- Final State ---")
|
||||
print(final_state)
|
||||
```
|
||||
|
||||
**Expected Output:**
|
||||
|
||||
```text
|
||||
Invoking graph...
|
||||
--- Running Adder Node 1 (current value: 0) ---
|
||||
--- Running Adder Node 2 (current value: 5) ---
|
||||
|
||||
--- Final State ---
|
||||
{'value': 15}
|
||||
```
|
||||
|
||||
Because we used `Annotated[int, BinaryOperatorAggregate(int, operator.add)]`, the `"value"` channel now *adds* incoming updates (`5` then `10`) to its current state, resulting in a final sum of `15`.
|
||||
|
||||
## How `StateGraph` Finds the Right Channel
|
||||
|
||||
You might wonder how `StateGraph` knows whether to use `LastValue` or something else. When you initialize `StateGraph(MyState)`, it inspects your state schema (`MyState`).
|
||||
|
||||
* It uses Python's `get_type_hints(MyState, include_extras=True)` to look at each field (like `value`).
|
||||
* If a field has `Annotated[SomeType, SomeChannelConfig]`, it uses `SomeChannelConfig` (e.g., `BinaryOperatorAggregate(...)`, `Topic(...)`) to create the channel for that key.
|
||||
* If a field is just `SomeType` (like `value: int`), it defaults to creating a `LastValue[SomeType]` channel for that key.
|
||||
|
||||
This logic is primarily handled within the `StateGraph._add_schema` method, which calls internal helpers like `_get_channels`.
|
||||
|
||||
```python
|
||||
# graph/state.py (Simplified view of channel detection)
|
||||
|
||||
def _get_channels(schema: Type[dict]) -> tuple[...]:
|
||||
# ... gets type hints including Annotated metadata ...
|
||||
type_hints = get_type_hints(schema, include_extras=True)
|
||||
all_keys = {}
|
||||
for name, typ in type_hints.items():
|
||||
# Checks if the annotation specifies a channel or binop
|
||||
if channel := _is_field_channel(typ) or _is_field_binop(typ):
|
||||
channel.key = name
|
||||
all_keys[name] = channel
|
||||
else:
|
||||
# Default case: Use LastValue
|
||||
fallback = LastValue(typ)
|
||||
fallback.key = name
|
||||
all_keys[name] = fallback
|
||||
# ... separate BaseChannel instances from ManagedValueSpec ...
|
||||
return channels, managed_values, type_hints
|
||||
|
||||
def _is_field_channel(typ: Type[Any]) -> Optional[BaseChannel]:
|
||||
# Checks if Annotated metadata contains a BaseChannel instance or class
|
||||
if hasattr(typ, "__metadata__"):
|
||||
meta = typ.__metadata__
|
||||
if len(meta) >= 1 and isinstance(meta[-1], BaseChannel):
|
||||
return meta[-1] # Return the channel instance directly
|
||||
# ... (handle channel classes too) ...
|
||||
return None
|
||||
|
||||
def _is_field_binop(typ: Type[Any]) -> Optional[BinaryOperatorAggregate]:
|
||||
# Checks if Annotated metadata contains a callable (the reducer function)
|
||||
if hasattr(typ, "__metadata__"):
|
||||
meta = typ.__metadata__
|
||||
if len(meta) >= 1 and callable(meta[-1]):
|
||||
# ... (validate function signature) ...
|
||||
return BinaryOperatorAggregate(typ, meta[-1]) # Create binop channel
|
||||
return None
|
||||
|
||||
# --- In StateGraph.__init__ ---
|
||||
# self._add_schema(state_schema) # This calls _get_channels
|
||||
```
|
||||
|
||||
## Under the Hood: `BaseChannel`
|
||||
|
||||
All channel types inherit from a base class called `BaseChannel`. This class defines the common interface that the [Pregel Execution Engine](05_pregel_execution_engine.md) uses to interact with any channel.
|
||||
|
||||
```python
|
||||
# channels/base.py (Simplified Abstract Base Class)
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Generic, Sequence, TypeVar
|
||||
|
||||
Value = TypeVar("Value") # The type of the stored state
|
||||
Update = TypeVar("Update") # The type of incoming updates
|
||||
Checkpoint = TypeVar("Checkpoint") # The type of saved state
|
||||
|
||||
class BaseChannel(Generic[Value, Update, Checkpoint], ABC):
|
||||
# ... (init, type properties) ...
|
||||
|
||||
@abstractmethod
|
||||
def update(self, values: Sequence[Update]) -> bool:
|
||||
"""Combines the sequence of updates with the current channel value."""
|
||||
# Must be implemented by subclasses (like LastValue, Topic)
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get(self) -> Value:
|
||||
"""Returns the current value of the channel."""
|
||||
# Must be implemented by subclasses
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def checkpoint(self) -> Checkpoint:
|
||||
"""Returns a serializable representation of the channel's state."""
|
||||
# Used by the Checkpointer
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def from_checkpoint(self, checkpoint: Checkpoint) -> Self:
|
||||
"""Creates a new channel instance from a saved checkpoint."""
|
||||
# Used by the Checkpointer
|
||||
pass
|
||||
```
|
||||
|
||||
The specific logic for `LastValue`, `Topic`, `BinaryOperatorAggregate`, etc., is implemented within their respective `update` and `get` methods, adhering to this common interface. The `checkpoint` and `from_checkpoint` methods are crucial for saving and loading the graph's state, which we'll explore more in [Chapter 6: Checkpointer (`BaseCheckpointSaver`)](06_checkpointer___basecheckpointsaver__.md).
|
||||
|
||||
## Conclusion
|
||||
|
||||
You've learned about **Channels**, the crucial communication and state management system within LangGraph's `StateGraph`.
|
||||
|
||||
* Channels act like **mailboxes** for each key in your graph's state.
|
||||
* They define **how updates are combined** when nodes write to the state.
|
||||
* The default channel is **`LastValue`**, which overwrites the previous value.
|
||||
* You can use `typing.Annotated` in your state definition to specify other channel types like **`BinaryOperatorAggregate`** (for combining values, e.g., summing) or **`Topic`** (for collecting updates into a list).
|
||||
* `StateGraph` automatically creates the correct channel for each state key based on your type hints.
|
||||
|
||||
Understanding channels helps you control precisely how information flows and accumulates in your stateful applications.
|
||||
|
||||
Now that we know how the state is managed (Channels) and how work gets done (Nodes), how do we control the *flow* of execution? What if we want to go to different nodes based on the current state? That's where conditional logic comes in.
|
||||
|
||||
Let's move on to [Chapter 4: Control Flow Primitives (`Branch`, `Send`, `Interrupt`)](04_control_flow_primitives___branch____send____interrupt__.md) to learn how to direct the traffic within our graph.
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
@@ -0,0 +1,608 @@
|
||||
# Chapter 4: Control Flow Primitives (`Branch`, `Send`, `Interrupt`)
|
||||
|
||||
In [Chapter 3: Channels](03_channels.md), we saw how information is stored and updated in our graph's shared state using Channels. We have the blueprint ([`StateGraph`](01_graph___stategraph.md)), the workers ([`Nodes`](02_nodes___pregelnode__.md)), and the communication system ([Channels](03_channels.md)).
|
||||
|
||||
But what if we don't want our graph to follow a single, fixed path? What if we need it to make decisions? For example, imagine a chatbot: sometimes it needs to use a tool (like a search engine), and other times it can answer directly. How do we tell the graph *which* path to take based on the current situation?
|
||||
|
||||
This is where **Control Flow Primitives** come in. They are special mechanisms that allow you to dynamically direct the execution path of your graph, making it much more flexible and powerful.
|
||||
|
||||
## What Problem Do Control Flow Primitives Solve?
|
||||
|
||||
Think of our graph like a train system. So far, we've only built tracks that go in a straight line from one station (node) to the next. Control flow primitives are like the **switches** and **signals** that allow the train (our execution flow) to:
|
||||
|
||||
1. **Choose a path:** Decide whether to go left or right at a junction based on some condition (like an "if" statement).
|
||||
2. **Dispatch specific trains:** Send a specific piece of cargo directly to a particular station, maybe even multiple pieces to the same station to be processed in parallel.
|
||||
3. **Wait for instructions:** Pause the train journey until an external signal (like human approval) is given.
|
||||
|
||||
LangGraph provides three main primitives for this:
|
||||
|
||||
* **`Branch`**: Acts like a conditional router or switch ("if/else"). It directs the flow to different nodes based on the current state.
|
||||
* **`Send`**: Allows a node to directly trigger another node with specific input, useful for parallel processing patterns like map-reduce.
|
||||
* **`Interrupt`**: Pauses the graph execution, usually to wait for external input (like a human clicking "Approve") before continuing.
|
||||
|
||||
Let's explore each one.
|
||||
|
||||
## 1. `Branch` - The Conditional Router
|
||||
|
||||
Imagine our chatbot needs to decide: "Should I use the search tool, or can I answer from my knowledge?" This decision depends on the conversation history or the user's specific question stored in the graph's state.
|
||||
|
||||
The `Branch` primitive allows us to implement this kind of conditional logic. You add it using the `graph.add_conditional_edges()` method.
|
||||
|
||||
**How it Works:**
|
||||
|
||||
1. You define a regular node (let's call it `should_i_search`).
|
||||
2. You define a separate **routing function**. This function takes the current state and decides *which node* should run next. It returns the name of the next node (or a list of names).
|
||||
3. You connect the `should_i_search` node to the routing function using `add_conditional_edges`. You tell it: "After `should_i_search` finishes, call this routing function to decide where to go next."
|
||||
4. You provide a mapping (a dictionary) that links the possible return values of your routing function to the actual node names in your graph.
|
||||
|
||||
**Example: Chatbot Deciding to Search**
|
||||
|
||||
Let's build a tiny graph that decides whether to go to a `search_tool` node or a `respond_directly` node.
|
||||
|
||||
**Step 1: Define State**
|
||||
|
||||
```python
|
||||
from typing import TypedDict, Annotated, List
|
||||
import operator
|
||||
|
||||
class ChatState(TypedDict):
|
||||
user_query: str
|
||||
# We'll store the decision here
|
||||
next_action: str
|
||||
# Keep track of intermediate results
|
||||
search_result: Annotated[List[str], operator.add] # Use Topic or add if accumulating
|
||||
final_response: str
|
||||
```
|
||||
|
||||
Our state holds the user's query and a field `next_action` to store the decision.
|
||||
|
||||
**Step 2: Define Nodes**
|
||||
|
||||
```python
|
||||
# Node that decides the next step
|
||||
def determine_action(state: ChatState) -> dict:
|
||||
print("--- Determining Action ---")
|
||||
query = state['user_query']
|
||||
if "weather" in query.lower():
|
||||
print("Decision: Need to use search tool for weather.")
|
||||
return {"next_action": "USE_TOOL"}
|
||||
else:
|
||||
print("Decision: Can respond directly.")
|
||||
return {"next_action": "RESPOND"}
|
||||
|
||||
# Node representing the search tool
|
||||
def run_search_tool(state: ChatState) -> dict:
|
||||
print("--- Using Search Tool ---")
|
||||
query = state['user_query']
|
||||
# Simulate finding a result
|
||||
result = f"Search result for '{query}': It's sunny!"
|
||||
# We return the result to be ADDED to the state list
|
||||
return {"search_result": [result]} # Return as list for operator.add
|
||||
|
||||
# Node that generates a final response
|
||||
def generate_response(state: ChatState) -> dict:
|
||||
print("--- Generating Response ---")
|
||||
if state.get("search_result"):
|
||||
response = f"Based on my search: {state['search_result'][-1]}"
|
||||
else:
|
||||
response = f"Responding directly to: {state['user_query']}"
|
||||
return {"final_response": response}
|
||||
```
|
||||
|
||||
**Step 3: Define the Routing Function**
|
||||
|
||||
This function reads the `next_action` from the state and returns the *key* we'll use in our mapping.
|
||||
|
||||
```python
|
||||
def route_based_on_action(state: ChatState) -> str:
|
||||
print("--- Routing ---")
|
||||
action = state['next_action']
|
||||
print(f"Routing based on action: {action}")
|
||||
if action == "USE_TOOL":
|
||||
return "route_to_tool" # This key must match our path_map
|
||||
else:
|
||||
return "route_to_respond" # This key must match our path_map
|
||||
```
|
||||
|
||||
**Step 4: Build the Graph with Conditional Edges**
|
||||
|
||||
```python
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
|
||||
workflow = StateGraph(ChatState)
|
||||
|
||||
workflow.add_node("decider", determine_action)
|
||||
workflow.add_node("search_tool", run_search_tool)
|
||||
workflow.add_node("responder", generate_response)
|
||||
|
||||
workflow.set_entry_point("decider")
|
||||
|
||||
# After 'decider', call 'route_based_on_action' to choose the next step
|
||||
workflow.add_conditional_edges(
|
||||
"decider", # Start node
|
||||
route_based_on_action, # The routing function
|
||||
{
|
||||
# Map the routing function's output to actual node names
|
||||
"route_to_tool": "search_tool",
|
||||
"route_to_respond": "responder"
|
||||
}
|
||||
)
|
||||
|
||||
# Define what happens *after* the conditional paths
|
||||
workflow.add_edge("search_tool", "responder") # After searching, generate response
|
||||
workflow.add_edge("responder", END) # After responding, end
|
||||
|
||||
# Compile
|
||||
app = workflow.compile()
|
||||
```
|
||||
|
||||
* `add_conditional_edges("decider", route_based_on_action, ...)`: This is the key part. It tells LangGraph: after the "decider" node runs, execute the `route_based_on_action` function.
|
||||
* `path_map = {"route_to_tool": "search_tool", ...}`: This dictionary maps the string returned by `route_based_on_action` to the actual next node to execute.
|
||||
|
||||
**Step 5: Run It!**
|
||||
|
||||
```python
|
||||
# Scenario 1: Query needs the tool
|
||||
print("--- Scenario 1: Weather Query ---")
|
||||
input1 = {"user_query": "What's the weather like?"}
|
||||
final_state1 = app.invoke(input1)
|
||||
print("Final State 1:", final_state1)
|
||||
|
||||
print("\n--- Scenario 2: Direct Response ---")
|
||||
# Scenario 2: Query doesn't need the tool
|
||||
input2 = {"user_query": "Tell me a joke."}
|
||||
final_state2 = app.invoke(input2)
|
||||
print("Final State 2:", final_state2)
|
||||
```
|
||||
|
||||
**Expected Output:**
|
||||
|
||||
```text
|
||||
--- Scenario 1: Weather Query ---
|
||||
--- Determining Action ---
|
||||
Decision: Need to use search tool for weather.
|
||||
--- Routing ---
|
||||
Routing based on action: USE_TOOL
|
||||
--- Using Search Tool ---
|
||||
--- Generating Response ---
|
||||
Final State 1: {'user_query': "What's the weather like?", 'next_action': 'USE_TOOL', 'search_result': ["Search result for 'What's the weather like?': It's sunny!"], 'final_response': "Based on my search: Search result for 'What's the weather like?': It's sunny!"}
|
||||
|
||||
--- Scenario 2: Direct Response ---
|
||||
--- Determining Action ---
|
||||
Decision: Can respond directly.
|
||||
--- Routing ---
|
||||
Routing based on action: RESPOND
|
||||
--- Generating Response ---
|
||||
Final State 2: {'user_query': 'Tell me a joke.', 'next_action': 'RESPOND', 'search_result': [], 'final_response': 'Responding directly to: Tell me a joke.'}
|
||||
```
|
||||
|
||||
See how the graph took different paths based on the `next_action` set by the `decider` node and interpreted by the `route_based_on_action` function!
|
||||
|
||||
**Visualizing the Branch:**
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
Start[START] --> Decider(decider);
|
||||
Decider -- route_based_on_action --> Route{Routing Logic};
|
||||
Route -- "route_to_tool" --> Search(search_tool);
|
||||
Route -- "route_to_respond" --> Respond(responder);
|
||||
Search --> Respond;
|
||||
Respond --> End(END);
|
||||
```
|
||||
|
||||
**Internals (`graph/branch.py`)**
|
||||
|
||||
* When you call `add_conditional_edges`, LangGraph stores a `Branch` object (`graph/branch.py`). This object holds your routing function (`path`) and the mapping (`path_map` / `ends`).
|
||||
* During execution, after the source node ("decider") finishes, the [Pregel Execution Engine](05_pregel_execution_engine.md) runs the `Branch` object.
|
||||
* The `Branch.run()` method eventually calls your routing function (`_route` or `_aroute` internally) with the current state.
|
||||
* It takes the return value (e.g., "route_to_tool"), looks it up in the `ends` dictionary to get the actual node name ("search_tool"), and tells the engine to schedule that node next.
|
||||
|
||||
```python
|
||||
# graph/branch.py (Simplified view)
|
||||
class Branch(NamedTuple):
|
||||
path: Runnable # Your routing function wrapped as a Runnable
|
||||
ends: Optional[dict[Hashable, str]] # Your path_map
|
||||
# ... other fields ...
|
||||
|
||||
def _route(self, input: Any, config: RunnableConfig, ...) -> Runnable:
|
||||
# ... reads current state if needed ...
|
||||
value = ... # Get the state
|
||||
result = self.path.invoke(value, config) # Call your routing function
|
||||
# ... determines destination node(s) using self.ends mapping ...
|
||||
destinations = [self.ends[r] for r in result]
|
||||
# ... tells the engine (via writer) which node(s) to run next ...
|
||||
return writer(destinations, config) or input # writer is a callback to the engine
|
||||
|
||||
# graph/state.py (Simplified view)
|
||||
class StateGraph(Graph):
|
||||
# ...
|
||||
def add_conditional_edges(self, source, path, path_map, ...):
|
||||
# ... wrap 'path' into a Runnable ...
|
||||
runnable_path = coerce_to_runnable(path, ...)
|
||||
# Create and store the Branch object
|
||||
self.branches[source][name] = Branch.from_path(runnable_path, path_map, ...)
|
||||
return self
|
||||
```
|
||||
|
||||
## 2. `Send` - Directing Specific Traffic
|
||||
|
||||
Sometimes, you don't just want to choose *one* path, but you want to trigger a *specific* node with *specific* data, possibly multiple times. This is common in "map-reduce" patterns where you split a task into smaller pieces, process each piece independently, and then combine the results.
|
||||
|
||||
The `Send` primitive allows a node (or a conditional edge function) to directly "send" a piece of data to another node, telling the engine: "Run *this* node next, and give it *this* input."
|
||||
|
||||
**How it Works:**
|
||||
|
||||
1. You import `Send` from `langgraph.graph` (or `langgraph.types`).
|
||||
2. In a node or a conditional edge function, instead of just returning a state update or a node name, you return `Send(target_node_name, data_for_that_node)`.
|
||||
3. You can return a list of `Send` objects to trigger multiple node executions, potentially in parallel (depending on the executor).
|
||||
|
||||
**Example: Simple Map-Reduce**
|
||||
|
||||
Let's imagine we want to process a list of items. One node splits the list, another node processes each item individually (the "map" step), and a final node aggregates the results (the "reduce" step).
|
||||
|
||||
**Step 1: Define State**
|
||||
|
||||
```python
|
||||
from typing import TypedDict, List, Annotated
|
||||
import operator
|
||||
|
||||
class MapReduceState(TypedDict):
|
||||
items_to_process: List[str]
|
||||
# Use Topic or operator.add to collect results from worker nodes
|
||||
processed_items: Annotated[List[str], operator.add]
|
||||
final_result: str
|
||||
```
|
||||
|
||||
**Step 2: Define Nodes**
|
||||
|
||||
```python
|
||||
# Node to prepare items (not really needed here, but shows the flow)
|
||||
def prepare_items(state: MapReduceState) -> dict:
|
||||
print("--- Preparing Items (No change) ---")
|
||||
# In a real scenario, this might fetch or generate the items
|
||||
return {}
|
||||
|
||||
# Node to process a single item (Our "Worker")
|
||||
def process_single_item(state: dict) -> dict:
|
||||
# Note: This node receives the dict passed via Send, NOT the full MapReduceState
|
||||
item = state['item']
|
||||
print(f"--- Processing Item: {item} ---")
|
||||
processed = f"Processed_{item.upper()}"
|
||||
# Return the processed item to be ADDED to the list in the main state
|
||||
return {"processed_items": [processed]} # Return list for operator.add
|
||||
|
||||
# Node to aggregate results
|
||||
def aggregate_results(state: MapReduceState) -> dict:
|
||||
print("--- Aggregating Results ---")
|
||||
all_processed = state['processed_items']
|
||||
final = ", ".join(all_processed)
|
||||
return {"final_result": final}
|
||||
```
|
||||
|
||||
**Step 3: Define the Dispatching Function (using `Send`)**
|
||||
|
||||
This function will run after `prepare_items` and will use `Send` to trigger `process_single_item` for each item.
|
||||
|
||||
```python
|
||||
from langgraph.graph import Send # Import Send
|
||||
|
||||
def dispatch_work(state: MapReduceState) -> List[Send]:
|
||||
print("--- Dispatching Work ---")
|
||||
items = state['items_to_process']
|
||||
send_packets = []
|
||||
for item in items:
|
||||
print(f"Sending item '{item}' to worker node.")
|
||||
# Create a Send object for each item
|
||||
# Target node: "worker"
|
||||
# Data payload: a dictionary {'item': current_item}
|
||||
packet = Send("worker", {"item": item})
|
||||
send_packets.append(packet)
|
||||
return send_packets # Return a list of Send objects
|
||||
```
|
||||
|
||||
**Step 4: Build the Graph**
|
||||
|
||||
```python
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
|
||||
workflow = StateGraph(MapReduceState)
|
||||
|
||||
workflow.add_node("preparer", prepare_items)
|
||||
workflow.add_node("worker", process_single_item) # The node targeted by Send
|
||||
workflow.add_node("aggregator", aggregate_results)
|
||||
|
||||
workflow.set_entry_point("preparer")
|
||||
|
||||
# After 'preparer', call 'dispatch_work' which returns Send packets
|
||||
workflow.add_conditional_edges("preparer", dispatch_work)
|
||||
# NOTE: We don't need a path_map here because dispatch_work directly
|
||||
# returns Send objects specifying the target node.
|
||||
|
||||
# The 'worker' node outputs are aggregated implicitly by the 'processed_items' channel.
|
||||
# We need an edge to tell the graph when to run the aggregator.
|
||||
# Let's wait until ALL workers triggered by Send are done.
|
||||
# We can achieve this implicitly if the aggregator reads state written by workers.
|
||||
# A simple edge ensures aggregator runs *after* the step involving workers.
|
||||
# (More complex aggregation might need explicit barrier channels)
|
||||
workflow.add_edge("worker", "aggregator")
|
||||
|
||||
workflow.add_edge("aggregator", END)
|
||||
|
||||
# Compile
|
||||
app = workflow.compile()
|
||||
```
|
||||
|
||||
**Step 5: Run It!**
|
||||
|
||||
```python
|
||||
input_state = {"items_to_process": ["apple", "banana", "cherry"]}
|
||||
final_state = app.invoke(input_state)
|
||||
print("\nFinal State:", final_state)
|
||||
```
|
||||
|
||||
**Expected Output (order of processing might vary):**
|
||||
|
||||
```text
|
||||
--- Preparing Items (No change) ---
|
||||
--- Dispatching Work ---
|
||||
Sending item 'apple' to worker node.
|
||||
Sending item 'banana' to worker node.
|
||||
Sending item 'cherry' to worker node.
|
||||
--- Processing Item: apple ---
|
||||
--- Processing Item: banana ---
|
||||
--- Processing Item: cherry ---
|
||||
--- Aggregating Results ---
|
||||
|
||||
Final State: {'items_to_process': ['apple', 'banana', 'cherry'], 'processed_items': ['Processed_APPLE', 'Processed_BANANA', 'Processed_CHERRY'], 'final_result': 'Processed_APPLE, Processed_BANANA, Processed_CHERRY'}
|
||||
```
|
||||
|
||||
The `dispatch_work` function returned three `Send` objects. The LangGraph engine then scheduled the "worker" node to run three times, each time with a different input dictionary (`{'item': 'apple'}`, `{'item': 'banana'}`, `{'item': 'cherry'}`). The results were automatically collected in `processed_items` thanks to the `operator.add` reducer on our `Annotated` state key. Finally, the `aggregator` ran.
|
||||
|
||||
**Internals (`types.py`, `constants.py`)**
|
||||
|
||||
* `Send(node, arg)` is a simple data class defined in `langgraph/types.py`.
|
||||
* When a node or branch returns `Send` objects, the engine collects them. Internally, these are often associated with a special channel key like `TASKS` (defined in `langgraph/constants.py`).
|
||||
* The [Pregel Execution Engine](05_pregel_execution_engine.md) processes these `TASKS`. For each `Send(node, arg)`, it schedules the target `node` to run in the *next* step, passing `arg` as its input.
|
||||
* This allows for dynamic, data-driven invocation of nodes outside the standard edge connections.
|
||||
|
||||
```python
|
||||
# types.py (Simplified view)
|
||||
class Send:
|
||||
__slots__ = ("node", "arg")
|
||||
node: str # Target node name
|
||||
arg: Any # Data payload for the node
|
||||
|
||||
def __init__(self, /, node: str, arg: Any) -> None:
|
||||
self.node = node
|
||||
self.arg = arg
|
||||
# ... repr, eq, hash ...
|
||||
|
||||
# constants.py (Simplified view)
|
||||
TASKS = sys.intern("__pregel_tasks") # Internal key for Send objects
|
||||
|
||||
# pregel/algo.py (Conceptual idea during task processing)
|
||||
# if write is for TASKS channel:
|
||||
# packet = write_value # This is the Send object
|
||||
# # Schedule packet.node to run in the next step with packet.arg
|
||||
# schedule_task(node=packet.node, input=packet.arg, ...)
|
||||
```
|
||||
|
||||
## 3. `Interrupt` - Pausing for Instructions
|
||||
|
||||
Sometimes, your graph needs to stop and wait for external input before proceeding. A common case is Human-in-the-Loop (HITL), where an AI agent proposes a plan or an action, and a human needs to approve it.
|
||||
|
||||
The `Interrupt` primitive allows a node to pause the graph's execution and wait. This requires a [Checkpointer](06_checkpointer___basecheckpointsaver__.md) to be configured, as the graph needs to save its state to be resumable later.
|
||||
|
||||
**How it Works:**
|
||||
|
||||
1. You import `interrupt` from `langgraph.types`.
|
||||
2. Inside a node, you call `interrupt(value_to_send_to_client)`.
|
||||
3. This immediately raises a special `GraphInterrupt` exception.
|
||||
4. The LangGraph engine catches this, saves the current state using the checkpointer, and returns control to your calling code, often signaling that an interrupt occurred. The `value_to_send_to_client` is included in the information returned.
|
||||
5. Later, you can resume the graph execution by providing a value. This is typically done by invoking the compiled graph again with a special `Command(resume=value_for_interrupt)` object (from `langgraph.types`) and the same configuration (including the thread ID for the checkpointer).
|
||||
6. When resumed, the graph loads the saved state. The execution engine restarts the *interrupted node from the beginning*. When the code reaches the `interrupt()` call again, instead of raising an exception, it *returns* the `value_for_interrupt` that you provided when resuming. The node then continues executing from that point.
|
||||
|
||||
**Example: Human Approval Step**
|
||||
|
||||
Let's create a graph where a node plans an action, another node presents it for human approval (using `interrupt`), and a final node executes it if approved.
|
||||
|
||||
**Step 1: Define State**
|
||||
|
||||
```python
|
||||
from typing import TypedDict, Optional
|
||||
|
||||
class ApprovalState(TypedDict):
|
||||
plan: str
|
||||
# We'll use the resume value to implicitly know if approved
|
||||
feedback: Optional[str] # Store feedback/approval status
|
||||
```
|
||||
|
||||
**Step 2: Define Nodes (including interrupt)**
|
||||
|
||||
```python
|
||||
from langgraph.types import interrupt, Command # Import interrupt and Command
|
||||
|
||||
# Node that creates a plan
|
||||
def create_plan(state: ApprovalState) -> dict:
|
||||
print("--- Creating Plan ---")
|
||||
plan = "Plan: Execute risky action X."
|
||||
return {"plan": plan}
|
||||
|
||||
# Node that requests human approval using interrupt
|
||||
def request_approval(state: ApprovalState) -> dict:
|
||||
print("--- Requesting Human Approval ---")
|
||||
plan = state['plan']
|
||||
print(f"Proposed Plan: {plan}")
|
||||
# Call interrupt, passing the plan to the client
|
||||
# Execution STOPS here on the first run.
|
||||
feedback_or_approval = interrupt(plan)
|
||||
# --- Execution RESUMES here on the second run ---
|
||||
print(f"--- Resumed with feedback: {feedback_or_approval} ---")
|
||||
# Store the feedback received from the resume command
|
||||
return {"feedback": str(feedback_or_approval)} # Ensure it's a string
|
||||
|
||||
# Node that executes the plan (only if approved implicitly by resuming)
|
||||
def execute_plan(state: ApprovalState) -> dict:
|
||||
print("--- Executing Plan ---")
|
||||
if state.get("feedback"): # Check if we got feedback (meaning we resumed)
|
||||
print(f"Executing '{state['plan']}' based on feedback: {state['feedback']}")
|
||||
return {} # No state change needed
|
||||
else:
|
||||
# This path shouldn't be hit if interrupt works correctly
|
||||
print("Execution skipped (no feedback received).")
|
||||
return{}
|
||||
|
||||
```
|
||||
|
||||
**Step 3: Build the Graph (with Checkpointer!)**
|
||||
|
||||
```python
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
# Need a checkpointer for interrupts!
|
||||
from langgraph.checkpoint.memory import MemorySaver
|
||||
|
||||
workflow = StateGraph(ApprovalState)
|
||||
|
||||
workflow.add_node("planner", create_plan)
|
||||
workflow.add_node("approval_gate", request_approval)
|
||||
workflow.add_node("executor", execute_plan)
|
||||
|
||||
workflow.set_entry_point("planner")
|
||||
workflow.add_edge("planner", "approval_gate")
|
||||
workflow.add_edge("approval_gate", "executor") # Runs after interrupt is resolved
|
||||
workflow.add_edge("executor", END)
|
||||
|
||||
# Create checkpointer and compile
|
||||
memory_saver = MemorySaver()
|
||||
app = workflow.compile(checkpointer=memory_saver)
|
||||
```
|
||||
|
||||
**Step 4: Run and Resume**
|
||||
|
||||
```python
|
||||
import uuid
|
||||
|
||||
# Unique ID for this conversation thread is needed for the checkpointer
|
||||
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
|
||||
|
||||
print("--- Initial Invocation ---")
|
||||
# Start the graph. It should interrupt at the approval node.
|
||||
interrupt_info = None
|
||||
for chunk in app.stream({"plan": ""}, config=config):
|
||||
print(chunk)
|
||||
# Check if the chunk contains interrupt information
|
||||
if "__interrupt__" in chunk:
|
||||
interrupt_info = chunk["__interrupt__"]
|
||||
print("\n!! Graph Interrupted !!")
|
||||
break # Stop processing stream after interrupt
|
||||
|
||||
# The client code inspects the interrupt value (the plan)
|
||||
if interrupt_info:
|
||||
print(f"Interrupt Value (Plan): {interrupt_info[0].value}") # interrupt_info is a tuple
|
||||
|
||||
# --- Simulate human interaction ---
|
||||
human_decision = "Approved, proceed with caution."
|
||||
print(f"\n--- Resuming with Decision: '{human_decision}' ---")
|
||||
|
||||
# Resume execution with the human's feedback/approval
|
||||
# We pass the decision using Command(resume=...)
|
||||
for chunk in app.stream(Command(resume=human_decision), config=config):
|
||||
print(chunk)
|
||||
|
||||
else:
|
||||
print("Graph finished without interruption.")
|
||||
```
|
||||
|
||||
**Expected Output:**
|
||||
|
||||
```text
|
||||
--- Initial Invocation ---
|
||||
{'planner': {'plan': 'Plan: Execute risky action X.'}}
|
||||
{'approval_gate': None} # Node starts execution
|
||||
--- Requesting Human Approval ---
|
||||
Proposed Plan: Plan: Execute risky action X.
|
||||
{'__interrupt__': (Interrupt(value='Plan: Execute risky action X.', resumable=True, ns=..., when='during'),)} # Interrupt occurs
|
||||
|
||||
!! Graph Interrupted !!
|
||||
Interrupt Value (Plan): Plan: Execute risky action X.
|
||||
|
||||
--- Resuming with Decision: 'Approved, proceed with caution.' ---
|
||||
{'approval_gate': {'feedback': 'Approved, proceed with caution.'}} # Node resumes and finishes
|
||||
--- Resumed with feedback: Approved, proceed with caution. ---
|
||||
{'executor': {}} # Executor node runs
|
||||
--- Executing Plan ---
|
||||
Executing 'Plan: Execute risky action X.' based on feedback: Approved, proceed with caution.
|
||||
{'__end__': {'plan': 'Plan: Execute risky action X.', 'feedback': 'Approved, proceed with caution.'}} # Graph finishes
|
||||
```
|
||||
|
||||
The graph paused at `request_approval` after printing the plan. We then resumed it by sending `Command(resume="Approved, proceed with caution.")`. The `request_approval` node restarted, the `interrupt()` call returned our resume value, which was stored in the state, and finally, the `executor` node ran using that feedback.
|
||||
|
||||
**Internals (`types.py`, `errors.py`, Checkpointer)**
|
||||
|
||||
* The `interrupt(value)` function (in `langgraph/types.py`) checks if a resume value is available for the current step within the node.
|
||||
* If no resume value exists (first run), it raises a `GraphInterrupt` exception (`langgraph/errors.py`) containing an `Interrupt` object (`langgraph/types.py`) which holds the `value`.
|
||||
* The [Pregel Execution Engine](05_pregel_execution_engine.md) catches `GraphInterrupt`.
|
||||
* If a [Checkpointer](06_checkpointer___basecheckpointsaver__.md) is present, the engine saves the current state (including which node was interrupted) and passes the `Interrupt` object back to the caller.
|
||||
* When you resume with `Command(resume=resume_value)`, the engine loads the checkpoint.
|
||||
* It knows which node was interrupted and provides the `resume_value` to it (often via a special `RESUME` entry written to the state channels, managed internally via `PregelScratchpad` in `pregel/algo.py`).
|
||||
* The node restarts. When `interrupt()` is called again, it finds the `resume_value` (provided via the scratchpad or internal state) and returns it instead of raising an exception.
|
||||
|
||||
```python
|
||||
# types.py (Simplified view)
|
||||
def interrupt(value: Any) -> Any:
|
||||
# ... access internal config/scratchpad ...
|
||||
scratchpad = conf[CONFIG_KEY_SCRATCHPAD]
|
||||
idx = scratchpad.interrupt_counter()
|
||||
|
||||
# Check if resume value already exists for this interrupt index
|
||||
if scratchpad.resume and idx < len(scratchpad.resume):
|
||||
return scratchpad.resume[idx] # Return existing resume value
|
||||
|
||||
# Check if a new global resume value was provided
|
||||
v = scratchpad.get_null_resume(consume=True)
|
||||
if v is not None:
|
||||
# Store and return the new resume value
|
||||
scratchpad.resume.append(v)
|
||||
conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)]) # Update state internally
|
||||
return v
|
||||
|
||||
# No resume value - raise the interrupt exception
|
||||
raise GraphInterrupt(
|
||||
(Interrupt(value=value, resumable=True, ns=...),)
|
||||
)
|
||||
|
||||
# types.py (Simplified view)
|
||||
@dataclasses.dataclass
|
||||
class Interrupt:
|
||||
value: Any # The value passed to interrupt()
|
||||
resumable: bool = True
|
||||
# ... other fields ...
|
||||
|
||||
# types.py (Simplified view)
|
||||
@dataclasses.dataclass
|
||||
class Command:
|
||||
# ... other fields like update, goto ...
|
||||
resume: Optional[Any] = None # Value to provide to a pending interrupt
|
||||
|
||||
# errors.py (Simplified view)
|
||||
class GraphInterrupt(Exception): # Base class for interrupts
|
||||
pass
|
||||
```
|
||||
|
||||
## Conclusion
|
||||
|
||||
You've learned about the essential tools for controlling the flow of execution in your LangGraph applications:
|
||||
|
||||
* **`Branch`** (`add_conditional_edges`): Used to create conditional paths, like `if/else` statements, directing the flow based on the current state. Requires a routing function and often a path map.
|
||||
* **`Send`**: Used to directly trigger a specific node with specific data, bypassing normal edges. Essential for patterns like map-reduce where you want to invoke the same worker node multiple times with different inputs.
|
||||
* **`Interrupt`** (`langgraph.types.interrupt`): Used to pause graph execution, typically for human-in-the-loop scenarios. Requires a checkpointer and is resumed using `Command(resume=...)`.
|
||||
|
||||
These primitives transform your graph from a simple linear sequence into a dynamic, decision-making process capable of handling complex, real-world workflows.
|
||||
|
||||
Now that we understand how nodes execute, how state is managed via channels, and how control flow directs traffic, let's look at the engine that orchestrates all of this behind the scenes.
|
||||
|
||||
Next up: [Chapter 5: Pregel Execution Engine](05_pregel_execution_engine.md)
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
223
docs/LangGraph/05_pregel_execution_engine.md
Normal file
223
docs/LangGraph/05_pregel_execution_engine.md
Normal file
@@ -0,0 +1,223 @@
|
||||
# Chapter 5: Pregel Execution Engine - The Engine Room
|
||||
|
||||
In the previous chapters, we learned how to build the blueprint of our application using [`StateGraph`](01_graph___stategraph.md), define the workers with [`Nodes`](02_nodes___pregelnode__.md), manage the shared state with [`Channels`](03_channels.md), and direct the traffic using [Control Flow Primitives](04_control_flow_primitives___branch____send____interrupt__.md).
|
||||
|
||||
But what actually takes all these pieces – the blueprint, the workers, the communication rules, the traffic signals – and makes them *run*? What ensures Node A runs, its output updates the state correctly via channels, and then Node B (or maybe Node C based on a Branch) runs with that updated state?
|
||||
|
||||
Meet the **Pregel Execution Engine**. This is the heart of LangGraph, the engine room that drives your graph forward.
|
||||
|
||||
## What Problem Does Pregel Solve?
|
||||
|
||||
Imagine you've designed a complex assembly line (your `StateGraph`). You have different stations (Nodes) where specific tasks are done, conveyor belts (Channels) moving parts between stations, and switches (Branches) directing parts down different paths.
|
||||
|
||||
How do you ensure the line runs smoothly? You need a manager! Someone who:
|
||||
|
||||
1. Knows the overall plan (the graph structure).
|
||||
2. Knows which station should work next based on what just finished.
|
||||
3. Delivers the right parts (state) to the right station.
|
||||
4. Collects the finished work from a station.
|
||||
5. Updates the central inventory (the shared state via Channels).
|
||||
6. Deals with decisions (Branches) and special instructions (Sends, Interrupts).
|
||||
7. Handles multiple stations working at the same time if possible (parallelism).
|
||||
8. Keeps track of progress and can save the state (Checkpointing).
|
||||
|
||||
The **Pregel Execution Engine** is this assembly line manager for your LangGraph application. It takes your compiled graph definition and orchestrates its execution step-by-step.
|
||||
|
||||
## Key Concepts: How Pregel Manages the Flow
|
||||
|
||||
Pregel is inspired by a system developed at Google for processing large graphs. LangGraph adapts these ideas for executing AI agents and multi-step workflows. Here's how it works conceptually:
|
||||
|
||||
1. **Step-by-Step Execution ("Supersteps"):** Pregel runs the graph in discrete steps, often called "supersteps." Think of it like turns in a board game.
|
||||
2. **Scheduling Nodes:** In each step, Pregel looks at the current state and the graph structure (edges, branches) to figure out which [Nodes (`PregelNode`)](02_nodes___pregelnode__.md) should run *in this turn*. This could be the entry point node at the start, nodes triggered by the previous step's output, or nodes activated by a `Send` command.
|
||||
3. **Executing Nodes:** It runs the scheduled nodes. If multiple nodes are scheduled for the same step and they don't directly depend on each other *within that step*, Pregel might run them in parallel using background threads or asyncio tasks.
|
||||
4. **Gathering Updates:** As each node finishes, it returns a dictionary of updates (like `{"value": 6}`). Pregel collects all these updates from all the nodes that ran in the current step.
|
||||
5. **Updating State via Channels:** Pregel takes the collected updates and applies them to the shared state using the appropriate [`Channels`](03_channels.md). For example, it sends `6` to the `"value"` channel, which might overwrite the old value (if it's `LastValue`) or add to it (if it's `BinaryOperatorAggregate`).
|
||||
6. **Looping:** After updating the state, Pregel checks if there are more nodes to run (e.g., nodes connected by edges from the ones that just finished) or if the graph has reached the `END`. If there's more work, it starts the next step (superstep).
|
||||
7. **Handling Control Flow:** It seamlessly integrates [Control Flow Primitives](04_control_flow_primitives___branch____send____interrupt__.md). When a `Branch` needs to run, Pregel executes the routing function and schedules the next node accordingly. When `Send` is used, Pregel schedules the target node with the specific data. When `Interrupt` occurs, Pregel pauses execution (and relies on a [Checkpointer](06_checkpointer___basecheckpointsaver__.md) to save state).
|
||||
8. **Checkpointing:** At configurable points (often after each step), Pregel interacts with the [Checkpointer (`BaseCheckpointSaver`)](06_checkpointer___basecheckpointsaver__.md) to save the current state of all channels. This allows the graph to be paused and resumed later.
|
||||
|
||||
Essentially, Pregel is the **orchestrator** that manages the entire lifecycle of a graph's execution.
|
||||
|
||||
## How Pregel Executes Our Simple Graph
|
||||
|
||||
Let's revisit the simple `adder -> multiplier` graph from [Chapter 1: Graph / StateGraph](01_graph___stategraph.md) and see how Pregel runs it when you call `app.invoke({"value": 5})`.
|
||||
|
||||
**Graph:**
|
||||
|
||||
* State: `{'value': int}` (uses `LastValue` channel by default)
|
||||
* Nodes: `adder` (value+1), `multiplier` (value*2)
|
||||
* Edges: `START -> adder`, `adder -> multiplier`, `multiplier -> END`
|
||||
|
||||
**Execution Flow:**
|
||||
|
||||
1. **Start:** `app.invoke({"value": 5})` is called. The Pregel engine inside the compiled `app` takes over.
|
||||
2. **Initialization:** Pregel sets the initial state in the `"value"` [Channel](03_channels.md) to `5`. `step = 0`.
|
||||
3. **Step 1 Begins:**
|
||||
* **Scheduling:** Pregel sees the edge from `START` to `adder`. It schedules the `adder` node to run in this step.
|
||||
* **Execution:** Pregel retrieves the current state (`{'value': 5}`) from the [Channel](03_channels.md) and runs the `add_one` function associated with the `adder` node.
|
||||
* **Gathering Updates:** The `add_one` function returns `{"value": 6}`. Pregel gathers this write.
|
||||
* **Applying Updates:** Pregel sends the update `6` to the `"value"` [Channel](03_channels.md). Since it's a `LastValue` channel, its state becomes `6`.
|
||||
* **(Checkpointing):** If a checkpointer is configured (and enabled for this step), Pregel saves the state (`{'value': 6}`).
|
||||
* `step` increments to `1`.
|
||||
4. **Step 2 Begins:**
|
||||
* **Scheduling:** Pregel looks at edges originating from nodes that completed in Step 1 (`adder`). It finds the edge `adder -> multiplier`. It schedules the `multiplier` node.
|
||||
* **Execution:** Pregel retrieves the current state (`{'value': 6}`) from the `"value"` [Channel](03_channels.md) and runs the `multiply_by_two` function.
|
||||
* **Gathering Updates:** The `multiply_by_two` function returns `{"value": 12}`. Pregel gathers this write.
|
||||
* **Applying Updates:** Pregel sends the update `12` to the `"value"` [Channel](03_channels.md). The channel's state becomes `12`.
|
||||
* **(Checkpointing):** Pregel saves the state (`{'value': 12}`).
|
||||
* `step` increments to `2`.
|
||||
5. **Step 3 Begins:**
|
||||
* **Scheduling:** Pregel looks at edges from `multiplier`. It finds the edge `multiplier -> END`. Reaching `END` means no more application nodes are scheduled.
|
||||
* **(Execution, Gathering, Applying):** No application nodes run.
|
||||
* **(Checkpointing):** Pregel saves the final state (`{'value': 12}`).
|
||||
6. **Finish:** Pregel detects the `END` state. Execution halts.
|
||||
7. **Return:** The final state (`{'value': 12}`) is read from the channels and returned by `app.invoke()`.
|
||||
|
||||
**Visualizing the Flow:**
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant User
|
||||
participant App as CompiledGraph
|
||||
participant PregelEngine as Pregel Engine
|
||||
participant StateChannels as Channels
|
||||
participant AdderNode as adder
|
||||
participant MultiplierNode as multiplier
|
||||
|
||||
User->>App: invoke({"value": 5})
|
||||
App->>PregelEngine: Start Execution
|
||||
PregelEngine->>StateChannels: Initialize state {"value": 5}
|
||||
Note over PregelEngine: Step 1
|
||||
PregelEngine->>PregelEngine: Schedule 'adder' (from START)
|
||||
PregelEngine->>StateChannels: Read state ({'value': 5})
|
||||
PregelEngine->>AdderNode: Run add_one({'value': 5})
|
||||
AdderNode-->>PregelEngine: Return {"value": 6}
|
||||
PregelEngine->>StateChannels: Apply update {"value": 6}
|
||||
StateChannels-->>PregelEngine: State is now {'value': 6}
|
||||
Note over PregelEngine: Step 2
|
||||
PregelEngine->>PregelEngine: Schedule 'multiplier' (from 'adder')
|
||||
PregelEngine->>StateChannels: Read state ({'value': 6})
|
||||
PregelEngine->>MultiplierNode: Run multiply_by_two({'value': 6})
|
||||
MultiplierNode-->>PregelEngine: Return {"value": 12}
|
||||
PregelEngine->>StateChannels: Apply update {"value": 12}
|
||||
StateChannels-->>PregelEngine: State is now {'value': 12}
|
||||
Note over PregelEngine: Step 3
|
||||
PregelEngine->>PregelEngine: Check edges from 'multiplier' (sees END)
|
||||
PregelEngine->>PregelEngine: No more nodes to schedule. Finish.
|
||||
PregelEngine->>StateChannels: Read final state ({'value': 12})
|
||||
PregelEngine->>App: Return final state {'value': 12}
|
||||
App->>User: Return {'value': 12}
|
||||
```
|
||||
|
||||
Pregel acts as the hidden conductor ensuring each part plays at the right time with the right information.
|
||||
|
||||
## Internal Implementation: A Glimpse Under the Hood
|
||||
|
||||
You don't typically interact with the Pregel engine directly; it's encapsulated within the compiled graph object you get from `graph.compile()`. However, understanding its core components helps clarify how LangGraph works. The main logic resides in the `langgraph/pregel/` directory.
|
||||
|
||||
1. **Compilation:** When you call `graph.compile()`, LangGraph analyzes your nodes, edges, branches, and state schema. It translates your high-level graph definition into an internal representation suitable for the Pregel engine. This includes creating the actual [`PregelNode`](02_nodes___pregelnode__.md) objects which contain information about which channels to read, which function to run, and how to write outputs back.
|
||||
2. **The Loop (`pregel/loop.py`):** The core execution happens within a loop (managed by classes like `SyncPregelLoop` or `AsyncPregelLoop`). Each iteration of this loop represents one "superstep".
|
||||
3. **Task Preparation (`pregel/algo.py::prepare_next_tasks`):** At the start of each step, this function determines which tasks (nodes) are ready to run. It checks:
|
||||
* Which [Channels](03_channels.md) were updated in the previous step.
|
||||
* Which nodes are triggered by those updated channels (based on edges and branches).
|
||||
* Are there any pending `Send` messages ([Control Flow Primitives](04_control_flow_primitives___branch____send____interrupt__.md)) targeting specific nodes?
|
||||
* It uses internal versioning on channels to avoid re-running nodes unnecessarily if their inputs haven't changed.
|
||||
4. **Task Execution (`pregel/runner.py::PregelRunner`):** This component takes the list of tasks scheduled for the current step and executes them.
|
||||
* It uses an executor (like Python's `concurrent.futures.ThreadPoolExecutor` for sync code or `asyncio` for async code) to potentially run independent tasks in parallel.
|
||||
* For each task, it reads the required state from the [Channels](03_channels.md), executes the node's function/Runnable, and collects the returned writes (the update dictionary).
|
||||
* It handles retries if configured for a node.
|
||||
5. **Applying Writes (`pregel/algo.py::apply_writes`):** After tasks in a step complete (or fail), this function gathers all the writes returned by those tasks.
|
||||
* It groups writes by channel name.
|
||||
* It calls the `.update()` method on each corresponding [Channel](03_channels.md) object, passing the collected updates for that channel. The channel itself enforces its update logic (e.g., `LastValue` overwrites, `Topic` appends).
|
||||
* It updates the internal checkpoint state with new channel versions.
|
||||
6. **Checkpointing (`pregel/loop.py`, `checkpoint/base.py`):** The loop interacts with the configured [Checkpointer (`BaseCheckpointSaver`)](06_checkpointer___basecheckpointsaver__.md) to save the graph's state (the values and versions of all channels) at appropriate times (e.g., after each step).
|
||||
7. **Interrupt Handling (`pregel/loop.py`, `types.py::interrupt`):** If a node calls `interrupt()`, the `PregelRunner` catches the `GraphInterrupt` exception. The `PregelLoop` then coordinates with the [Checkpointer](06_checkpointer___basecheckpointsaver__.md) to save state and pause execution, returning control to the user. Resuming involves loading the checkpoint and providing the resume value back to the waiting `interrupt()` call.
|
||||
|
||||
**Simplified Code Snippets:**
|
||||
|
||||
* **Task Preparation (Conceptual):**
|
||||
```python
|
||||
# pregel/algo.py (Simplified Concept)
|
||||
def prepare_next_tasks(checkpoint, processes, channels, config, step, ...):
|
||||
tasks = {}
|
||||
# Check PUSH tasks (from Send)
|
||||
for packet in checkpoint["pending_sends"]:
|
||||
# ... create task if node exists ...
|
||||
task = create_task_for_send(packet, ...)
|
||||
tasks[task.id] = task
|
||||
|
||||
# Check PULL tasks (from edges/triggers)
|
||||
for name, proc in processes.items():
|
||||
# Check if any trigger channel for 'proc' was updated since last seen
|
||||
if _triggers(channels, checkpoint["channel_versions"], proc):
|
||||
# ... read input for the node ...
|
||||
task = create_task_for_pull(name, proc, ...)
|
||||
tasks[task.id] = task
|
||||
return tasks
|
||||
```
|
||||
This function checks both explicit `Send` commands and regular node triggers based on updated channels to build the list of tasks for the next step.
|
||||
|
||||
* **Applying Writes (Conceptual):**
|
||||
```python
|
||||
# pregel/algo.py (Simplified Concept)
|
||||
def apply_writes(checkpoint, channels, tasks: list[PregelExecutableTask], get_next_version):
|
||||
# ... (sort tasks for determinism, update seen versions) ...
|
||||
pending_writes_by_channel = defaultdict(list)
|
||||
for task in tasks:
|
||||
for chan, val in task.writes: # task.writes is the dict returned by the node
|
||||
if chan in channels:
|
||||
pending_writes_by_channel[chan].append(val)
|
||||
# ... (handle TASKS, PUSH, managed values etc.) ...
|
||||
|
||||
updated_channels = set()
|
||||
# Apply writes to channels
|
||||
for chan_name, values_to_update in pending_writes_by_channel.items():
|
||||
channel_obj = channels[chan_name]
|
||||
if channel_obj.update(values_to_update): # Channel applies its logic here!
|
||||
# If updated, bump the version in the checkpoint
|
||||
checkpoint["channel_versions"][chan_name] = get_next_version(...)
|
||||
updated_channels.add(chan_name)
|
||||
|
||||
# ... (handle channels that weren't written to but need bumping) ...
|
||||
return updated_channels
|
||||
```
|
||||
This function takes the results from all nodes in a step and uses the `channel.update()` method to modify the state according to each channel's rules.
|
||||
|
||||
* **The Main Loop (Conceptual):**
|
||||
```python
|
||||
# pregel/loop.py (Simplified Concept - SyncPregelLoop/AsyncPregelLoop)
|
||||
class PregelLoop:
|
||||
def run(self): # Simplified invoke/stream logic
|
||||
with self: # Enters context (loads checkpoint, sets up channels)
|
||||
while self.tick(): # tick executes one step
|
||||
# Start tasks for the current step using PregelRunner
|
||||
runner = PregelRunner(...)
|
||||
for _ in runner.tick(self.tasks):
|
||||
# Yield control back, allowing writes/outputs to be streamed
|
||||
pass # (actual stream logic happens via callbacks)
|
||||
return self.output # Return final result
|
||||
```
|
||||
The loop repeatedly calls `tick()`. Inside `tick()`, it prepares tasks, runs them using `PregelRunner`, applies the resulting writes, handles checkpoints/interrupts, and determines if another step is needed.
|
||||
|
||||
You don't need to know the deep implementation details, but understanding this step-by-step process managed by Pregel helps visualize how your graph comes alive.
|
||||
|
||||
## Conclusion
|
||||
|
||||
The **Pregel Execution Engine** is the powerful, yet hidden, coordinator that runs your LangGraph graphs.
|
||||
|
||||
* It executes the graph **step-by-step** (supersteps).
|
||||
* In each step, it **schedules** which nodes run based on the graph structure and current state.
|
||||
* It **runs** the nodes (potentially in parallel).
|
||||
* It **gathers** node outputs and **updates** the shared state using [`Channels`](03_channels.md).
|
||||
* It seamlessly integrates [`Control Flow Primitives`](04_control_flow_primitives___branch____send____interrupt__.md) like `Branch`, `Send`, and `Interrupt`.
|
||||
* It works with a [`Checkpointer`](06_checkpointer___basecheckpointsaver__.md) to save and resume state.
|
||||
|
||||
Think of it as the engine ensuring your application's logic flows correctly, state is managed reliably, and complex operations are orchestrated smoothly.
|
||||
|
||||
We've mentioned checkpointing several times – the ability to save and load the graph's state. This is crucial for long-running processes, human-in-the-loop workflows, and resilience. How does that work?
|
||||
|
||||
Let's dive into [Chapter 6: Checkpointer (`BaseCheckpointSaver`)](06_checkpointer___basecheckpointsaver__.md) to understand how LangGraph persists and resumes state.
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
391
docs/LangGraph/06_checkpointer___basecheckpointsaver__.md
Normal file
391
docs/LangGraph/06_checkpointer___basecheckpointsaver__.md
Normal file
@@ -0,0 +1,391 @@
|
||||
# Chapter 6: Checkpointer (`BaseCheckpointSaver`) - Saving Your Progress
|
||||
|
||||
In [Chapter 5: Pregel Execution Engine](05_pregel_execution_engine.md), we saw how the engine runs our graph step-by-step. But what happens if a graph takes hours to run, or if it needs to pause and wait for a human? If the program crashes or we need to stop it, do we lose all the progress?
|
||||
|
||||
That's where **Checkpointers** come to the rescue!
|
||||
|
||||
## What Problem Do Checkpointers Solve?
|
||||
|
||||
Imagine you're playing a long video game. You wouldn't want to start from the very beginning every time you stop playing, right? Games have save points or checkpoints that record your progress.
|
||||
|
||||
LangGraph's **Checkpointer** does the same thing for your graph execution. It automatically saves the graph's state at certain points, usually after each step completed by the [Pregel Execution Engine](05_pregel_execution_engine.md).
|
||||
|
||||
This is incredibly useful for:
|
||||
|
||||
1. **Long-Running Processes:** If your graph involves many steps or calls to slow tools/LLMs, you can stop it and resume later without losing work.
|
||||
2. **Resilience:** If your program crashes unexpectedly, you can restart it from the last saved checkpoint.
|
||||
3. **Human-in-the-Loop (HITL):** As we saw with `Interrupt` in [Chapter 4: Control Flow Primitives](04_control_flow_primitives___branch____send____interrupt__.md), pausing the graph requires saving its state so it can be perfectly restored when the human provides input. Checkpointers are essential for this.
|
||||
|
||||
**Analogy:** Think of a checkpointer as an automatic "Save" button for your graph's progress. It takes snapshots of the shared "whiteboard" ([Channels](03_channels.md)) so you can always pick up where you left off.
|
||||
|
||||
## Key Concepts
|
||||
|
||||
1. **What is Saved?** The checkpointer saves the current value and version of every [Channel](03_channels.md) in your graph's state. It also keeps track of which step the graph was on and any pending tasks (like those created by `Send`).
|
||||
2. **When is it Saved?** The [Pregel Execution Engine](05_pregel_execution_engine.md) typically triggers the checkpointer to save after each "superstep" (a round of node executions and state updates).
|
||||
3. **Where is it Saved?** This depends on the specific checkpointer implementation you choose. LangGraph provides several:
|
||||
* `MemorySaver`: Stores checkpoints in your computer's RAM. Simple for testing, but **lost when your script ends**.
|
||||
* `SqliteSaver`: Stores checkpoints in a local SQLite database file, making them persistent across script runs.
|
||||
* Other savers might store checkpoints in cloud databases or other persistent storage.
|
||||
4. **`thread_id` (The Save Slot Name):** To save and load progress correctly, you need a way to identify *which* specific run of the graph you want to work with. Think of this like naming your save file in a game. In LangGraph, this identifier is called the `thread_id`. You provide it in the `config` when you run the graph. Each unique `thread_id` represents an independent "conversation" or execution history.
|
||||
|
||||
## How to Use a Checkpointer
|
||||
|
||||
Using a checkpointer is straightforward. You just need to tell LangGraph *which* checkpointer to use when you compile your graph.
|
||||
|
||||
**Step 1: Import a Checkpointer**
|
||||
|
||||
Let's start with the simplest one, `MemorySaver`.
|
||||
|
||||
```python
|
||||
# Import the simplest checkpointer
|
||||
from langgraph.checkpoint.memory import MemorySaver
|
||||
```
|
||||
|
||||
**Step 2: Instantiate the Checkpointer**
|
||||
|
||||
```python
|
||||
# Create an instance of the memory checkpointer
|
||||
memory_saver = MemorySaver()
|
||||
```
|
||||
|
||||
**Step 3: Compile Your Graph with the Checkpointer**
|
||||
|
||||
Let's reuse our simple `adder -> multiplier` graph. The graph definition itself doesn't change.
|
||||
|
||||
```python
|
||||
# --- Define State and Nodes (same as Chapter 1) ---
|
||||
from typing import TypedDict
|
||||
from langgraph.graph import StateGraph, END, START
|
||||
|
||||
class MyState(TypedDict):
|
||||
value: int
|
||||
|
||||
def add_one(state: MyState) -> dict:
|
||||
print(f"Adder: Adding 1 to {state['value']}")
|
||||
return {"value": state['value'] + 1}
|
||||
|
||||
def multiply_by_two(state: MyState) -> dict:
|
||||
print(f"Multiplier: Doubling {state['value']}")
|
||||
return {"value": state['value'] * 2}
|
||||
|
||||
# --- Build the Graph (same as Chapter 1) ---
|
||||
workflow = StateGraph(MyState)
|
||||
workflow.add_node("adder", add_one)
|
||||
workflow.add_node("multiplier", multiply_by_two)
|
||||
workflow.set_entry_point("adder")
|
||||
workflow.add_edge("adder", "multiplier")
|
||||
workflow.add_edge("multiplier", END)
|
||||
|
||||
# --- Compile WITH the checkpointer ---
|
||||
# Pass the checkpointer instance to the compile method
|
||||
app = workflow.compile(checkpointer=memory_saver)
|
||||
```
|
||||
|
||||
That's it! By passing `checkpointer=memory_saver` to `compile()`, you've enabled automatic checkpointing for this graph.
|
||||
|
||||
**Step 4: Run with a `thread_id`**
|
||||
|
||||
To use the checkpointer, you need to provide a configuration dictionary (`config`) containing a unique identifier for this specific execution thread.
|
||||
|
||||
```python
|
||||
import uuid
|
||||
|
||||
# Create a unique ID for this run
|
||||
thread_id = str(uuid.uuid4())
|
||||
config = {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
# Define the initial state
|
||||
initial_state = {"value": 5}
|
||||
|
||||
print("--- Running Graph (First Time) ---")
|
||||
# Run the graph with the config
|
||||
final_state = app.invoke(initial_state, config=config)
|
||||
|
||||
print("\n--- Final State (First Run) ---")
|
||||
print(final_state)
|
||||
```
|
||||
|
||||
**Expected Output (First Run):**
|
||||
|
||||
```text
|
||||
--- Running Graph (First Time) ---
|
||||
Adder: Adding 1 to 5
|
||||
Multiplier: Doubling 6
|
||||
|
||||
--- Final State (First Run) ---
|
||||
{'value': 12}
|
||||
```
|
||||
|
||||
Behind the scenes, `MemorySaver` saved the state after the `adder` step and after the `multiplier` step, associating it with the `thread_id` you provided.
|
||||
|
||||
**Step 5: Resume the Graph**
|
||||
|
||||
Now, let's imagine we stopped the process. If we run the *same graph* with the *same `thread_id`*, the checkpointer allows the [Pregel Execution Engine](05_pregel_execution_engine.md) to load the last saved state and continue. Since the first run finished completely, running `invoke` again will just load the final state.
|
||||
|
||||
```python
|
||||
print("\n--- Running Graph Again with SAME thread_id ---")
|
||||
# Use the SAME config (containing the same thread_id)
|
||||
# Provide NO initial state, as it will be loaded from the checkpoint
|
||||
resumed_state = app.invoke(None, config=config)
|
||||
|
||||
print("\n--- Final State (Resumed Run) ---")
|
||||
print(resumed_state)
|
||||
|
||||
# Let's check the saved states using the checkpointer directly
|
||||
print("\n--- Checkpoints Saved ---")
|
||||
for checkpoint in memory_saver.list(config):
|
||||
print(checkpoint)
|
||||
```
|
||||
|
||||
**Expected Output (Second Run):**
|
||||
|
||||
```text
|
||||
--- Running Graph Again with SAME thread_id ---
|
||||
# Notice: No node printouts because the graph already finished!
|
||||
# It just loads the final saved state.
|
||||
|
||||
--- Final State (Resumed Run) ---
|
||||
{'value': 12}
|
||||
|
||||
--- Checkpoints Saved ---
|
||||
# You'll see checkpoint objects representing saved states
|
||||
CheckpointTuple(config={'configurable': {'thread_id': '...'}}, checkpoint={'v': 1, 'ts': '...', 'id': '...', 'channel_values': {'value': 6}, 'channel_versions': {'adder': 1}, 'versions_seen': {'adder': {}}}, metadata={'source': 'loop', 'step': 1, ...}, ...)
|
||||
CheckpointTuple(config={'configurable': {'thread_id': '...'}}, checkpoint={'v': 1, 'ts': '...', 'id': '...', 'channel_values': {'value': 12}, 'channel_versions': {'adder': 1, 'multiplier': 2}, 'versions_seen': {'adder': {}, 'multiplier': {'adder': 1}}}, metadata={'source': 'loop', 'step': 2, ...}, ...)
|
||||
CheckpointTuple(config={'configurable': {'thread_id': '...'}}, checkpoint={'v': 1, 'ts': '...', 'id': '...', 'channel_values': {'value': 12}, 'channel_versions': {'adder': 1, 'multiplier': 2}, 'versions_seen': {'adder': {}, 'multiplier': {'adder': 1}}}, metadata={'source': 'loop', 'step': 3, ...}, ...)
|
||||
```
|
||||
|
||||
The checkpointer successfully loaded the final state (`{'value': 12}`) associated with that `thread_id`.
|
||||
|
||||
**Checkpointers and `Interrupt` (Human-in-the-Loop)**
|
||||
|
||||
Remember the `Interrupt` example from [Chapter 4](04_control_flow_primitives___branch____send____interrupt__.md)?
|
||||
|
||||
```python
|
||||
# (Simplified HITL example from Chapter 4)
|
||||
from langgraph.types import interrupt, Command
|
||||
# ... (State, Nodes: create_plan, request_approval, execute_plan) ...
|
||||
|
||||
# Compile WITH checkpointer (REQUIRED for interrupt)
|
||||
memory_saver_hitl = MemorySaver()
|
||||
app_hitl = workflow.compile(checkpointer=memory_saver_hitl)
|
||||
|
||||
# Run, get interrupted
|
||||
config_hitl = {"configurable": {"thread_id": str(uuid.uuid4())}}
|
||||
for chunk in app_hitl.stream({"plan": ""}, config=config_hitl):
|
||||
# ... (detect interrupt) ...
|
||||
print("Graph interrupted!")
|
||||
break
|
||||
|
||||
# Resume after human decision
|
||||
human_decision = "Approved"
|
||||
for chunk in app_hitl.stream(Command(resume=human_decision), config=config_hitl):
|
||||
# ... (process remaining steps) ...
|
||||
print("Graph resumed and finished!")
|
||||
```
|
||||
|
||||
When `interrupt()` was called inside the `request_approval` node, the [Pregel Execution Engine](05_pregel_execution_engine.md) automatically used the `memory_saver_hitl` checkpointer to save the *exact state* of the graph at that moment (including the plan). When we called `stream` again with `Command(resume=...)` and the *same* `config_hitl`, the engine loaded that saved state using the checkpointer, allowing the graph to continue exactly where it left off, now with the human's feedback.
|
||||
|
||||
**Without a checkpointer, `Interrupt` cannot work.**
|
||||
|
||||
## How Checkpointing Works Internally
|
||||
|
||||
What happens behind the scenes when a checkpointer is configured?
|
||||
|
||||
**Saving:**
|
||||
|
||||
1. **Step Complete:** The [Pregel Execution Engine](05_pregel_execution_engine.md) finishes a step (e.g., after running the `adder` node and updating the state).
|
||||
2. **Signal Checkpointer:** The engine tells the configured checkpointer (`MemorySaver` in our example) that it's time to save.
|
||||
3. **Gather State:** The checkpointer (or the engine on its behalf) accesses all the active [Channels](03_channels.md).
|
||||
4. **Serialize State:** For each channel, it calls the channel's internal `checkpoint()` method to get a serializable representation of its current value (e.g., the number `6` for the `"value"` channel).
|
||||
5. **Store Checkpoint:** The checkpointer bundles the serialized channel values, their versions, the current step number, and other metadata into a `Checkpoint` object. It then stores this `Checkpoint` associated with the current `thread_id` provided in the `config`. `MemorySaver` stores it in a dictionary in RAM; `SqliteSaver` writes it to a database table.
|
||||
|
||||
**Loading (Resuming):**
|
||||
|
||||
1. **Invoke with `thread_id`:** You call `app.invoke(None, config=config)` where `config` contains a `thread_id` that has been previously saved.
|
||||
2. **Request Checkpoint:** The [Pregel Execution Engine](05_pregel_execution_engine.md) asks the checkpointer to load the latest checkpoint for the given `thread_id`.
|
||||
3. **Retrieve Checkpoint:** The checkpointer retrieves the saved `Checkpoint` object (e.g., from its memory dictionary or the database).
|
||||
4. **Restore State:** The engine takes the saved channel values from the checkpoint. For each channel, it calls the channel's `from_checkpoint()` method (or similar internal logic) to restore its state. The "whiteboard" ([Channels](03_channels.md)) is now exactly as it was when the checkpoint was saved.
|
||||
5. **Continue Execution:** The engine looks at the saved step number and metadata to figure out where to resume execution, typically by preparing the tasks for the *next* step.
|
||||
|
||||
Here's a simplified view of the interaction:
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant User
|
||||
participant App as CompiledGraph
|
||||
participant Engine as Pregel Engine
|
||||
participant Saver as Checkpointer (e.g., MemorySaver)
|
||||
participant Storage as Underlying Storage (RAM, DB)
|
||||
|
||||
%% Saving %%
|
||||
Engine->>Engine: Finishes Step N
|
||||
Engine->>Saver: Save checkpoint for config (thread_id)
|
||||
Saver->>Engine: Request current channel states & versions
|
||||
Engine-->>Saver: Provides states & versions
|
||||
Saver->>Storage: Store Checkpoint(Step N, states, versions) linked to thread_id
|
||||
Storage-->>Saver: Acknowledge Save
|
||||
Saver-->>Engine: Save Complete
|
||||
|
||||
%% Loading %%
|
||||
User->>App: invoke(None, config with thread_id)
|
||||
App->>Engine: Start/Resume Execution
|
||||
Engine->>Saver: Get latest checkpoint for config (thread_id)
|
||||
Saver->>Storage: Retrieve Checkpoint linked to thread_id
|
||||
Storage-->>Saver: Returns Checkpoint(Step N, states, versions)
|
||||
Saver-->>Engine: Provides Checkpoint
|
||||
Engine->>Engine: Restore channel states from checkpoint
|
||||
Engine->>Engine: Prepare tasks for Step N+1
|
||||
Engine->>App: Continue execution...
|
||||
```
|
||||
|
||||
## A Peek at the Code (`checkpoint/base.py`, `checkpoint/memory.py`, `pregel/loop.py`)
|
||||
|
||||
Let's look at the core components:
|
||||
|
||||
* **`BaseCheckpointSaver` (`checkpoint/base.py`)**: This is the abstract base class (like a template) that all checkpointers must implement. It defines the essential methods the engine needs.
|
||||
|
||||
```python
|
||||
# checkpoint/base.py (Highly Simplified)
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, Mapping, Optional, Sequence, Tuple, TypedDict
|
||||
|
||||
# Represents a saved checkpoint
|
||||
class Checkpoint(TypedDict):
|
||||
channel_values: Mapping[str, Any] # Saved state of channels
|
||||
channel_versions: Mapping[str, int] # Internal versions
|
||||
versions_seen: Mapping[str, Mapping[str, int]] # Tracking for node execution
|
||||
# ... other metadata like v, ts, id, pending_sends ...
|
||||
|
||||
# Represents the checkpoint tuple retrieved from storage
|
||||
class CheckpointTuple(NamedTuple):
|
||||
config: dict # The config used (includes thread_id)
|
||||
checkpoint: Checkpoint
|
||||
metadata: dict
|
||||
# ... other fields like parent_config, pending_writes ...
|
||||
|
||||
class BaseCheckpointSaver(ABC):
|
||||
# --- Sync Methods ---
|
||||
@abstractmethod
|
||||
def get_tuple(self, config: dict) -> Optional[CheckpointTuple]:
|
||||
"""Load the checkpoint tuple for the given config."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def put(self, config: dict, checkpoint: Checkpoint, metadata: dict) -> dict:
|
||||
"""Save a checkpoint."""
|
||||
...
|
||||
|
||||
# --- Async Methods (similar structure) ---
|
||||
@abstractmethod
|
||||
async def aget_tuple(self, config: dict) -> Optional[CheckpointTuple]:
|
||||
"""Async load the checkpoint tuple."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def aput(self, config: dict, checkpoint: Checkpoint, metadata: dict) -> dict:
|
||||
"""Async save a checkpoint."""
|
||||
...
|
||||
|
||||
# --- Other methods (list, put_writes) omitted for brevity ---
|
||||
```
|
||||
The key methods are `get_tuple` (to load) and `put` (to save), along with their async counterparts (`aget_tuple`, `aput`). Any specific checkpointer (like `MemorySaver`, `SqliteSaver`) must provide concrete implementations for these methods.
|
||||
|
||||
* **`MemorySaver` (`checkpoint/memory.py`)**: A simple implementation that uses an in-memory dictionary.
|
||||
|
||||
```python
|
||||
# checkpoint/memory.py (Highly Simplified)
|
||||
import threading
|
||||
from collections import defaultdict
|
||||
|
||||
class MemorySaver(BaseCheckpointSaver):
|
||||
def __init__(self):
|
||||
# Use a dictionary to store checkpoints in RAM
|
||||
# Key: thread_id, Value: List of CheckpointTuples
|
||||
self._checkpoints: defaultdict[str, list[CheckpointTuple]] = defaultdict(list)
|
||||
self._lock = threading.RLock() # To handle multiple threads safely
|
||||
|
||||
def get_tuple(self, config: dict) -> Optional[CheckpointTuple]:
|
||||
thread_id = config["configurable"]["thread_id"]
|
||||
with self._lock:
|
||||
if checkpoints := self._checkpoints.get(thread_id):
|
||||
# Return the latest checkpoint for this thread_id
|
||||
return checkpoints[-1]
|
||||
return None
|
||||
|
||||
def put(self, config: dict, checkpoint: Checkpoint, metadata: dict) -> dict:
|
||||
thread_id = config["configurable"]["thread_id"]
|
||||
with self._lock:
|
||||
# Append the new checkpoint to the list for this thread_id
|
||||
self._checkpoints[thread_id].append(
|
||||
CheckpointTuple(config, checkpoint, metadata)
|
||||
)
|
||||
return {"configurable": {"thread_id": thread_id}}
|
||||
|
||||
# ... async methods (aget_tuple, aput) are similar using the same dict ...
|
||||
# ... list method iterates through the dictionary ...
|
||||
```
|
||||
As you can see, `MemorySaver` just uses a standard Python dictionary (`self._checkpoints`) to store the `CheckpointTuple` for each `thread_id`. This is simple but not persistent.
|
||||
|
||||
* **Integration (`pregel/loop.py`)**: The [Pregel Execution Engine](05_pregel_execution_engine.md) (`PregelLoop` classes) interacts with the checkpointer during its execution cycle.
|
||||
|
||||
```python
|
||||
# pregel/loop.py (Conceptual Snippets)
|
||||
|
||||
class PregelLoop: # Base class for Sync/Async loops
|
||||
def __init__(self, ..., checkpointer: Optional[BaseCheckpointSaver], ...):
|
||||
self.checkpointer = checkpointer
|
||||
# ... other init ...
|
||||
|
||||
def _put_checkpoint(self, metadata: CheckpointMetadata) -> None:
|
||||
# Called by the loop after a step or input processing
|
||||
if self.checkpointer:
|
||||
# 1. Create the Checkpoint object from current channels/state
|
||||
checkpoint_data = create_checkpoint(self.checkpoint, self.channels, ...)
|
||||
|
||||
# 2. Call the checkpointer's put method (sync or async)
|
||||
# (Uses self.submit to potentially run in background)
|
||||
self.submit(self.checkpointer.put, self.checkpoint_config, checkpoint_data, metadata)
|
||||
|
||||
# 3. Update internal config with the new checkpoint ID
|
||||
self.checkpoint_config = {"configurable": {"thread_id": ..., "checkpoint_id": checkpoint_data["id"]}}
|
||||
|
||||
def __enter__(self): # Or __aenter__ for async
|
||||
# Called when the loop starts
|
||||
if self.checkpointer:
|
||||
# 1. Try to load an existing checkpoint tuple
|
||||
saved = self.checkpointer.get_tuple(self.checkpoint_config)
|
||||
else:
|
||||
saved = None
|
||||
|
||||
if saved:
|
||||
# 2. Restore state from the loaded checkpoint
|
||||
self.checkpoint = saved.checkpoint
|
||||
self.checkpoint_config = saved.config
|
||||
# ... restore channels from saved.checkpoint['channel_values'] ...
|
||||
else:
|
||||
# Initialize with an empty checkpoint
|
||||
self.checkpoint = empty_checkpoint()
|
||||
|
||||
# ... setup channels based on restored or empty checkpoint ...
|
||||
return self
|
||||
```
|
||||
The `PregelLoop` uses the checkpointer's `get_tuple` method when it starts (in `__enter__` or `__aenter__`) to load any existing state. It uses the `put` method (inside `_put_checkpoint`) during execution to save progress.
|
||||
|
||||
## Conclusion
|
||||
|
||||
You've learned about **Checkpointers (`BaseCheckpointSaver`)**, the mechanism that gives your LangGraph applications memory and resilience.
|
||||
|
||||
* Checkpointers **save** the state of your graph's [Channels](03_channels.md) periodically.
|
||||
* They **load** saved states to resume execution.
|
||||
* This is crucial for **long-running graphs**, **human-in-the-loop** workflows (using `Interrupt`), and **recovering from failures**.
|
||||
* You enable checkpointing by passing a `checkpointer` instance (like `MemorySaver` or `SqliteSaver`) to `graph.compile()`.
|
||||
* You manage different execution histories using a unique `thread_id` in the `config`.
|
||||
* `MemorySaver` is simple for testing but lost when the script ends; use database savers (like `SqliteSaver`) for true persistence.
|
||||
|
||||
This chapter concludes our tour of the core concepts in LangGraph! You now understand the fundamental building blocks: the blueprint ([`StateGraph`](01_graph___stategraph.md)), the workers ([`Nodes`](02_nodes___pregelnode__.md)), the communication system ([`Channels`](03_channels.md)), the traffic signals ([Control Flow Primitives](04_control_flow_primitives___branch____send____interrupt__.md)), the engine room ([Pregel Execution Engine](05_pregel_execution_engine.md)), and the save system ([Checkpointer](06_checkpointer___basecheckpointsaver__.md)).
|
||||
|
||||
With these concepts, you're well-equipped to start building your own sophisticated, stateful applications with LangGraph! Explore the documentation for more examples, advanced patterns, and different checkpointer implementations. Happy building!
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
40
docs/LangGraph/index.md
Normal file
40
docs/LangGraph/index.md
Normal file
@@ -0,0 +1,40 @@
|
||||
# Tutorial: LangGraph
|
||||
|
||||
LangGraph helps you build complex **stateful applications**, like chatbots or agents, using a *graph-based approach*.
|
||||
You define your application's logic as a series of steps (**Nodes**) connected by transitions (**Edges**) in a **Graph**.
|
||||
The system manages the application's *shared state* using **Channels** and executes the graph step-by-step with its **Pregel engine**, handling things like branching, interruptions, and saving progress (**Checkpointing**).
|
||||
|
||||
|
||||
**Source Repository:** [https://github.com/langchain-ai/langgraph/tree/55f922cf2f3e63600ed8f0d0cd1262a75a991fdc/libs/langgraph/langgraph](https://github.com/langchain-ai/langgraph/tree/55f922cf2f3e63600ed8f0d0cd1262a75a991fdc/libs/langgraph/langgraph)
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A0["Pregel Execution Engine"]
|
||||
A1["Graph / StateGraph"]
|
||||
A2["Channels"]
|
||||
A3["Nodes (PregelNode)"]
|
||||
A4["Checkpointer (BaseCheckpointSaver)"]
|
||||
A5["Control Flow Primitives (Branch, Send, Interrupt)"]
|
||||
A0 -- "Executes" --> A1
|
||||
A1 -- "Contains" --> A3
|
||||
A3 -- "Updates State Via" --> A2
|
||||
A0 -- "Manages State Via" --> A2
|
||||
A0 -- "Uses Checkpointer" --> A4
|
||||
A1 -- "Defines Control Flow With" --> A5
|
||||
A5 -- "Directs Execution Of" --> A0
|
||||
A4 -- "Saves State Of" --> A2
|
||||
```
|
||||
|
||||
## Chapters
|
||||
|
||||
1. [Graph / StateGraph](01_graph___stategraph.md)
|
||||
2. [Nodes (`PregelNode`)](02_nodes___pregelnode__.md)
|
||||
3. [Channels](03_channels.md)
|
||||
4. [Control Flow Primitives (`Branch`, `Send`, `Interrupt`)](04_control_flow_primitives___branch____send____interrupt__.md)
|
||||
5. [Pregel Execution Engine](05_pregel_execution_engine.md)
|
||||
6. [Checkpointer (`BaseCheckpointSaver`)](06_checkpointer___basecheckpointsaver__.md)
|
||||
|
||||
|
||||
---
|
||||
|
||||
Generated by [AI Codebase Knowledge Builder](https://github.com/The-Pocket/Tutorial-Codebase-Knowledge)
|
||||
Reference in New Issue
Block a user