lib0.1.17

Allow call to update_state without values

.github/workflows/size.yml

@@ -0,0 +1,28 @@
+name: Check File Size
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:
+
+jobs:
+  file-size-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Get changed files
+        id: changed-files
+        uses: tj-actions/changed-files@v44
+      - name: Filter by size
+        run: |
+          large_added_files=$(find ${{ steps.changed-files.outputs.added_files }} -maxdepth 0 -size +1M)
+          if [ -n "$large_added_files" ]; then
+            echo "Large files added: $large_added_files"
+            echo "# Large files added:" >> $GITHUB_STEP_SUMMARY
+            echo "$large_added_files" >> $GITHUB_STEP_SUMMARY
+            exit 1
+          fi

README.md

@@ -3,7 +3,6 @@
 ![Version](https://img.shields.io/pypi/v/langgraph)
 [![Downloads](https://static.pepy.tech/badge/langgraph/month)](https://pepy.tech/project/langgraph)
 [![Open Issues](https://img.shields.io/github/issues-raw/langchain-ai/langgraph)](https://github.com/langchain-ai/langgraph/issues)
-[![](https://dcbadge.vercel.app/api/server/6adMQxSpJS?compact=true&style=flat)](https://discord.com/channels/1038097195422978059/1170024642245832774)
 [![Docs](https://img.shields.io/badge/docs-latest-blue)](https://langchain-ai.github.io/langgraph/)
 
 ⚡ Building language agents as graphs ⚡

docs/docs/cloud/concepts/api.md

@@ -12,6 +12,10 @@ An assistant is a configured instance of a [`CompiledGraph`][compiledgraph]. It
 
 The LangGraph Cloud API provides several endpoints for creating and managing assistants. See the <a href="../reference/api/api_ref.html#tag/assistantscreate" target="_blank">API reference</a> for more details.
 
+#### Configuring Assistants
+
+You can save custom assistants from the same graph to set different default prompts, models, and other configurations without changing a line of code in your graph. This allows you the ability to quickly test out different configurations without having to rewrite your graph every time, and also give users the flexibility to select different configurations when using your LangGraph application. See <a href="https://langchain-ai.github.io/langgraph/cloud/how-tos/cloud_examples/configuration_cloud/">this</a> how-to for information on how to configure a deployed graph. 
+
 ### Threads
 
 A thread contains the accumulated state of a group of runs. If a run is executed on a thread, then the [state][state] of the underlying graph of the assistant will be persisted to the thread. A thread's current and historical state can be retrieved. To persist state, a thread must be created prior to executing a run.

docs/docs/cloud/deployment/graph_rebuild.md

@@ -0,0 +1,146 @@
+# Rebuild Graph at Runtime
+
+You might need to rebuild your graph with a different configuration for a new run. For example, you might need to use a different graph state or graph structure depending on the config. This guide shows how you can do this.
+
+!!! note "Note"
+    In most cases, customizing behavior based on the config should be handled by a single graph where each node can read a config and change its behavior based on it
+
+## Prerequisites
+
+Make sure to check out [this how-to guide](./setup.md) on setting up your app for deployment first.
+
+## Define graphs
+
+Let's say you have an app with a simple graph that calls an LLM and returns the response to the user. The app file directory looks like the following:
+
+```
+my-app/
+|-- requirements.txt
+|-- .env
+|-- openai_agent.py     # code for your graph
+```
+
+where the graph is defined in `openai_agent.py`. 
+
+### No rebuild
+
+In the standard LangGraph API configuration, the server uses the compiled graph instance that's defined at the top level of `openai_agent.py`, which looks like the following:
+
+```python
+from langchain_openai import ChatOpenAI
+from langgraph.graph import END, MessageGraph
+
+model = ChatOpenAI(temperature=0)
+
+graph_workflow = MessageGraph()
+
+graph_workflow.add_node("agent", model)
+graph_workflow.add_edge("agent", END)
+graph_workflow.set_entry_point("agent")
+
+agent = graph_workflow.compile()
+```
+
+To make the server aware of your graph, you need to specify a path to the variable that contains the `CompiledStateGraph` instance in your LangGraph API configuration (`langgraph.json`), e.g.:
+
+```
+{
+    "dependencies": ["."],
+    "graphs": {
+        "openai_agent": "./openai_agent.py:agent",
+    },
+    "env": "./.env"
+}
+```
+
+### Rebuild
+
+To make your graph rebuild on each new run with custom configuration, you need to rewrite `openai_agent.py` to instead provide a _function_ that takes a config and returns a graph (or compiled graph) instance. Let's say we want to return our existing graph for user ID '1', and a tool-calling agent for other users. We can modify `openai_agent.py` as follows:
+
+```python
+from typing import Annotated, TypedDict
+from langchain_openai import ChatOpenAI
+from langgraph.graph import END, MessageGraph
+from langgraph.graph.state import StateGraph
+from langgraph.graph.message import add_messages
+from langgraph.prebuilt import ToolNode
+from langchain_core.tools import tool
+from langchain_core.messages import BaseMessage
+from langchain_core.runnables import RunnableConfig
+
+
+class State(TypedDict):
+    messages: Annotated[list[BaseMessage], add_messages]
+
+
+model = ChatOpenAI(temperature=0)
+
+def make_default_graph():
+    """Make a simple LLM agent"""
+    graph_workflow = StateGraph(State)
+    def call_model(state):
+        return {"messages": [model.invoke(state["messages"])]}
+
+    graph_workflow.add_node("agent", call_model)
+    graph_workflow.add_edge("agent", END)
+    graph_workflow.set_entry_point("agent")
+
+    agent = graph_workflow.compile()
+    return agent
+
+
+def make_alternative_graph():
+    """Make a tool-calling agent"""
+
+    @tool
+    def add(a: float, b: float):
+        """Adds two numbers."""
+        return a + b
+
+    tool_node = ToolNode([add])
+    model_with_tools = model.bind_tools([add])
+    def call_model(state):
+        return {"messages": [model_with_tools.invoke(state["messages"])]}
+
+    def should_continue(state: State):
+        if state["messages"][-1].tool_calls:
+            return "tools"
+        else:
+            return END
+
+    graph_workflow = StateGraph(State)
+
+    graph_workflow.add_node("agent", call_model)
+    graph_workflow.add_node("tools", tool_node)
+    graph_workflow.add_edge("tools", "agent")
+    graph_workflow.set_entry_point("agent")
+    graph_workflow.add_conditional_edges("agent", should_continue)
+
+    agent = graph_workflow.compile()
+    return agent
+
+
+# this is the graph making function that will decide which graph to
+# build based on the provided config
+def make_graph(config: RunnableConfig):
+    user_id = config.get("configurable", {}).get("user_id")
+    # route to different graph state / structure based on the user ID
+    if user_id == "1":
+        return make_default_graph()
+    else:
+        return make_alternative_graph()
+```
+
+Finally, you need to specify the path to your graph-making function (`make_graph`) in `langgraph.json`:
+
+```
+{
+    "dependencies": ["."],
+    "graphs": {
+        "openai_agent": "./openai_agent.py:make_graph",
+    },
+    "env": "./.env"
+}
+```
+
+See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)

docs/docs/cloud/deployment/setup.md

@@ -88,7 +88,7 @@ agent = graph_workflow.compile()
 ```
 
 !!! warning "Assign `CompiledGraph` to Variable"
-    The build process for LangGraph Cloud requires that the `CompiledGraph` object be assigned to a variable at the top-level of a Python module.
+    The build process for LangGraph Cloud requires that the `CompiledGraph` object be assigned to a variable at the top-level of a Python module (alternatively, you can provide [a function that creates a graph](./graph_rebuild.md)).
 
 Example file directory:
 ```

docs/docs/cloud/how-tos/invoke_studio.md

@@ -8,6 +8,8 @@ The LangGraph Studio lets you test different configurations and inputs to your g
     1. Select `Submit` to invoke the selected assistant.
 1. View output of the invocation in the right-hand pane.
 
-The following GIF shows these exact steps being carried out:
+The following video shows these exact steps being carried out:
 
-![Using LangGraph Studio](./img/studio_input.gif)
+<video controls allowfullscreen="true" poster="../img/studio_input_poster.png">
+    <source src="../img/studio_input.mp4" type="video/mp4">
+</video>

docs/docs/cloud/how-tos/test_deployment.md

@@ -9,6 +9,8 @@ Starting from the <a href="https://smith.langchain.com/" target="_blank">LangSmi
 1. In the top-right corner, select `Open LangGraph Studio`.
 1. [Invoke an assistant](./invoke_studio.md) or [view an existing thread](./threads_studio.md).
 
-The following GIF shows these exact steps being carried out:
+The following video shows these exact steps being carried out:
 
-![Using LangGraph Studio](./img/studio_usage.gif)
+<video controls allowfullscreen="true" poster="../img/studio_usage_poster.png">
+    <source src="../img/studio_usage.mp4" type="video/mp4">
+</video>

docs/docs/cloud/how-tos/threads_studio.md

@@ -6,14 +6,18 @@
 1. View the state of the thread (i.e. the output) in the right-hand pane.
 1. To create a new thread, select `+ New Thread`.
 
-The following GIF shows these exact steps being carried out:
+The following video shows these exact steps being carried out:
 
-![Using LangGraph Studio](./img/studio_threads.gif)
+<video controls="true" allowfullscreen="true" poster="../img/studio_threads_poster.png">
+    <source src="../img/studio_threads.mp4" type="video/mp4">
+</video>
 
 ## Edit Thread State
 
 The LangGraph Studio UI contains features for editing thread state. Explore these features in the right-hand pane. Select the `Edit` icon, modify the desired state, and then select `Fork` to invoke the assistant with the updated state.
 
-The following GIF shows how to edit a thread in the studio:
+The following video shows how to edit a thread in the studio:
 
-![Using LangGraph Studio](./img/studio_forks.gif)
+<video controls allowfullscreen="true" poster="../img/studio_forks_poster.png">
+    <source src="../img/studio_forks.mp4" type="video/mp4">
+</video>

docs/docs/cloud/index.md

@@ -12,7 +12,11 @@
 !!! warning "Under Construction"
     LangGraph Cloud documentation is under construction. Contents may change until general availability.
 
-![GIF](./how-tos/img/studio_input.gif)
+
+<video controls preload="auto" allowfullscreen="true" poster="how-tos/img/studio_forks_poster.png">
+    <source src="how-tos/img/studio_forks.mp4" type="video/mp4">
+</video>
+
 
 ## Overview
 

docs/docs/cloud/reference/cli.md

@@ -13,7 +13,7 @@ The LangGraph CLI requires a JSON configuration file with the following keys:
 | Key | Description |
 | --- | ----------- |
 | `dependencies` | **Required**. Array of dependencies for LangGraph Cloud API server. Dependencies can be one of the following: (1) `"."`, which will look for local Python packages, (2) `pyproject.toml`, `setup.py` or `requirements.txt` in the app directory `"./local_package"`, or (3) a package name. |
-| `graphs` | **Required**. Mapping from graph ID to path where the compiled graph is defined. Example: `./your_package/your_file.py:variable`, where `variable` is an instance of `langgraph.graph.graph.CompiledGraph`. |
+| `graphs` | **Required**. Mapping from graph ID to path where the compiled graph or a function that makes a graph is defined. Example: <ul><li>`./your_package/your_file.py:variable`, where `variable` is an instance of `langgraph.graph.state.CompiledStateGraph`</li><li>`./your_package/your_file.py:make_graph`, where `make_graph` is a function that takes a config dictionary (`langchain_core.runnables.RunnableConfig`) and creates an instance of `langgraph.graph.state.StateGraph` / `langgraph.graph.state.CompiledStateGraph`.</li></ul> |
 | `env` | Path to `.env` file or a mapping from environment variable to its value. |
 | `python_version` | `3.11` or `3.12`. Defaults to `3.11`. |
 | `pip_config_file`| Path to `pip` config file. |
@@ -49,7 +49,7 @@ Example:
         "."
     ],
     "graphs": {
-        "my_graph_id": "./your_package/your_file.py:variable"
+        "my_graph_id": "./your_package/your_file.py:make_graph"
     },
     "env": {
         "OPENAI_API_KEY": "secret-key"

docs/docs/concepts/index.md

@@ -20,7 +20,7 @@ Low Level Concepts
 - [State](low_level.md#state)
     - [Schema](low_level.md#schema)
     - [Reducers](low_level.md#reducers)
-    - [MessageState](low_level.md#messagestate)
+    - [MessageState](low_level.md#working-with-messages-in-graph-state)
 - [Nodes](low_level.md#nodes)
     - [`START` node](low_level.md#start-node)
     - [`END` node](low_level.md#end-node)

docs/docs/concepts/low_level.md

@@ -49,6 +49,7 @@ The main documented way to specify the schema of a graph is by using `TypedDict`
 By default, the graph will have the same input and output schemas. If you want to change this, you can also specify explicit input and output schemas directly. This is useful when you have a lot of keys, and some are explicitly for input and others for output. See the [notebook here](../how-tos/input_output_schema.ipynb) for how to use.
 
 By default, all nodes in the graph will share the same state. This means that they will read and write to the same state channels. It is possible to have nodes write to private state channels inside the graph for internal node communication - see [this notebook](../how-tos/pass_private_state.ipynb) for how to do that.
+
 ### Reducers
 
 Reducers are key to understanding how updates from nodes are applied to the `State`. Each key in the `State` has its own independent reducer function. If no reducer function is explicitly specified then it is assumed that all updates to that key should override it. Let's take a look at a few examples to understand them better.
@@ -78,22 +79,44 @@ class State(TypedDict):
 
 In this example, we've used the `Annotated` type to specify a reducer function (`operator.add`) for the second key (`bar`). Note that the first key remains unchanged. Let's assume the input to the graph is `{"foo": 1, "bar": ["hi"]}`. Let's then assume the first `Node` returns `{"foo": 2}`. This is treated as an update to the state. Notice that the `Node` does not need to return the whole `State` schema - just an update. After applying this update, the `State` would then be `{"foo": 2, "bar": ["hi"]}`. If the second node returns `{"bar": ["bye"]}` then the `State` would then be `{"foo": 2, "bar": ["hi", "bye"]}`. Notice here that the `bar` key is updated by adding the two lists together.
 
-### MessageState
+### Working with Messages in Graph State
 
-`MessageState` is one of the few opinionated components in LangGraph. `MessageState` is a special state designed to make it easy to use a list of messages as a key in your state. Specifically, `MessageState` is defined as:
+#### Why use messages?
+
+Most modern LLM providers have a chat model interface that accepts a list of messages as input. LangChain's [`ChatModel`](https://python.langchain.com/v0.2/docs/concepts/#chat-models) in particular accepts a list of `Message` objects as inputs. These messages come in a variety of forms such as `HumanMessage` (user input) or `AIMessage` (LLM response). To read more about what message objects are, please refer to [this](https://python.langchain.com/v0.2/docs/concepts/#messages) conceptual guide.
+
+#### Using Messages in your Graph
+
+In many cases, it is helpful to store prior conversation history as a list of messages in your graph state. To do so, we can add a key (channel) to the graph state that stores a list of `Message` objects and annotate it with a reducer function (see `messages` key in the example below). The reducer function is vital to telling the graph how to update the list of `Message` objects in the state with each state update (for example, when a node sends an update). If you don't specify a reducer, every state update will overwrite the list of messages with the most recently provided value. If you wanted to simply append messages to the existing list, you could use `operator.add` as a reducer.
+
+However, you might also want to manually update messages in your graph state (e.g. human-in-the-loop). If you were to use `operator.add`, the manual state updates you send to the graph would be appended to the existing list of messages, instead of updating existing messages. To avoid that, you need a reducer that can keep track of message IDs and overwrite existing messages, if updated. To achieve this, you can use the prebuilt `add_messages` function. For brand new messages, it will simply append to existing list, but it will also handle the updates for existing messages correctly.
+
+#### Serialization
+
+In addition to keeping track of message IDs, the `add_messages` function will also try to deserialize messages into LangChain `Message` objects whenever a state update is received on the `messages` channel. See more information on LangChain serialization/deserialization [here](https://python.langchain.com/v0.2/docs/how_to/serialization/). This allows sending graph inputs / state updates in the following format:
+
+```python
+# this is supported
+{"messages": [HumanMessage(content="message")]}
+
+# and this is also supported
+{"messages": [{"type": "human", "content": "message"}]}
+```
+
+Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as it's reducer function.
 
 ```python
 from langchain_core.messages import AnyMessage
 from langgraph.graph.message import add_messages
 from typing import Annotated, TypedDict
 
-class MessagesState(TypedDict):
+class GraphState(TypedDict):
     messages: Annotated[list[AnyMessage], add_messages]
 ```
 
-What this is doing is creating a `TypedDict` with a single key: `messages`. This is a list of `Message` objects, with `add_messages` as a reducer. `add_messages` basically adds messages to the existing list (it also does some nice extra things, like convert from OpenAI message format to the standard LangChain message format, handle updates based on message IDs, etc).
+#### MessagesState
 
-We often see a list of messages being a key component of state, so this prebuilt state is intended to make it easy to use messages. Typically, there is more state to track than just messages, so we see people subclass this state and add more fields, like:
+Since having a list of messages in your state is so common, there exists a prebuilt state called `MessagesState` which makes it easy to use messages. `MessagesState` is defined with a single `messages` key which is a list of `AnyMessage` objects and uses the `add_messages` reducer. Typically, there is more state to track than just messages, so we see people subclass this state and add more fields, like:
 
 ```python
 from langgraph.graph import MessagesState

docs/docs/how-tos/index.md

@@ -61,6 +61,13 @@ These guides show how to use different streaming modes.
 - [How to pass graph state to tools](pass-run-time-values-to-tools.ipynb)
 - [How to pass config to tools](pass-config-to-tools.ipynb)
 
+## State Management
+
+- [Use Pydantic model as state](state-model.ipynb)
+- [Use a context object in state](state-context-key.ipynb)
+- [Have a separate input and output schema](input_output_schema.ipynb)
+- [Pass private state between nodes inside the graph](pass_private_state.ipynb)
+
 ## Other
 
 - [How to run graph asynchronously](async.ipynb)

docs/mkdocs.yml

@@ -192,6 +192,7 @@ nav:
           - Deployment:
               - Setup App: "cloud/deployment/setup.md"
               - Setup App (pyproject.toml): "cloud/deployment/setup_pyproject.md"
+              - Rebuild Graph at Runtime: "cloud/deployment/graph_rebuild.md"
               - Test App Locally: "cloud/deployment/test_locally.md"
               - Deploy to Cloud: "cloud/deployment/cloud.md"
               - Self-Host: "cloud/deployment/self_hosted.md"

examples/cloud_examples/langgraph_to_langgraph_cloud.ipynb

@@ -50,7 +50,7 @@
    "metadata": {},
    "outputs": [
     {
-     "name": "stdin",
+     "name": "stdout",
      "output_type": "stream",
      "text": [
       "OPENAI_API_KEY:  ········\n"
@@ -991,7 +991,7 @@
    "id": "08996d90-a3ff-4655-9763-1dd4971344d4",
    "metadata": {},
    "source": [
-    "### With LangGraph Clound"
+    "### With LangGraph Cloud"
    ]
   },
   {

examples/introduction.ipynb

@@ -84,8 +84,9 @@
    ]
   },
   {
+   "attachments": {},
    "cell_type": "markdown",
-   "id": "ef7bcad1-1274-4b7c-a2e9-365180ef3a31",
+   "id": "9c374e41-f9b7-439e-a520-6d8c853c5220",
    "metadata": {},
    "source": [
     "## Part 1: Build a Basic Chatbot\n",
@@ -120,13 +121,24 @@
     "graph_builder = StateGraph(State)"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "31c755cd-8994-4867-bdff-96a55d7beae7",
+   "metadata": {},
+   "source": [
+    "<div class=\"admonition tip\">\n",
+    "    <p class=\"admonition-title\">Note</p>\n",
+    "    <p>\n",
+    "    The first thing you do when you define a graph is define the <code>State</code> of the graph. The <code>State</code> consists of the schema of the graph as well as reducer functions which specify how to apply updates to the state. In our example <code>State</code> is a <code>TypedDict</code> with a single key: <code>messages</code>. The <code>messages</code> key is annotated with the <a href=\"https://langchain-ai.github.io/langgraph/reference/graphs/?h=add+messages#add_messages\"><code>add_messages</code></a> reducer function, which tells LangGraph to append new messages to the existing list, rather than overwriting it. State keys without an annotation will be overwritten by each update, storing the most recent value. Check out <a href=\"https://langchain-ai.github.io/langgraph/reference/graphs/?h=add+messages#add_messages\">this conceptual guide</a> to learn more about state, reducers and other low-level concepts.\n",
+    "    </p>\n",
+    "</div>"
+   ]
+  },
   {
    "cell_type": "markdown",
    "id": "4137feed-746e-4c72-a34a-f7a699ad5dcf",
    "metadata": {},
    "source": [
-    "**Notice** that we've defined our `State` as a TypedDict with a single key: `messages`. The `messages` key is annotated with the [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/?h=add+messages#add_messages) function, which tells LangGraph to append new messages to the existing list, rather than overwriting it.\n",
-    "\n",
     "So now our graph knows two things:\n",
     "\n",
     "1. Every `node` we define will receive the current `State` as input and return a value that updates that state.\n",
@@ -3056,9 +3068,9 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
+   "display_name": "langgraph",
    "language": "python",
-   "name": "python3"
+   "name": "langgraph"
   },
   "language_info": {
    "codemirror_mode": {
@@ -3070,7 +3082,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.11.1"
+   "version": "3.11.9"
   }
  },
  "nbformat": 4,

examples/subgraph.ipynb

@@ -11,7 +11,7 @@
    "source": [
     "# How to create subgraphs\n",
     "\n",
-    "For more complex systems, sub-graphs are a useful design principle. Sub-graphs allow you to create and manage different states in different parts of your graph. This allows you build things like [multi-agent teams](./multi_agent/hierarchical_agent_teams.ipynb), where each team can track its own separate state.\n",
+    "For more complex systems, sub-graphs are a useful design principle. Sub-graphs allow you to create and manage different states in different parts of your graph. This allows you build things like [multi-agent teams](https://langchain-ai.github.io/langgraph/tutorials/multi_agent/hierarchical_agent_teams/), where each team can track its own separate state.\n",
     "\n",
     "![Screenshot 2024-07-11 at 1.01.28 PM.png](attachment:71516aef-9c00-4730-a676-a54e90cb6472.png)"
    ]

examples/tutorials/tool-calling-agent-local.ipynb

@@ -15,6 +15,7 @@
     "\n",
     "```\n",
     "ollama pull llama3-groq-tool-use\n",
+    "ollama pull llama3.1\n",
     "```\n",
     "\n",
     "And also, we'll use the Ollama partner package.\n",
@@ -39,35 +40,39 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 6,
    "id": "120c1da8-e45e-4ffa-9ac1-a536026c7e1c",
    "metadata": {},
-   "outputs": [],
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "\n",
+      "\u001b[1m[\u001b[0m\u001b[34;49mnotice\u001b[0m\u001b[1;39;49m]\u001b[0m\u001b[39;49m A new release of pip is available: \u001b[0m\u001b[31;49m24.0\u001b[0m\u001b[39;49m -> \u001b[0m\u001b[32;49m24.1.2\u001b[0m\n",
+      "\u001b[1m[\u001b[0m\u001b[34;49mnotice\u001b[0m\u001b[1;39;49m]\u001b[0m\u001b[39;49m To update, run: \u001b[0m\u001b[32;49mpip install --upgrade pip\u001b[0m\n",
+      "Note: you may need to restart the kernel to use updated packages.\n"
+     ]
+    }
+   ],
    "source": [
     "%pip install -qU langchain-ollama"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": 8,
    "id": "32c0504b-007a-4af6-9976-c7294ed26b73",
    "metadata": {},
-   "outputs": [
-    {
-     "name": "stderr",
-     "output_type": "stream",
-     "text": [
-      "USER_AGENT environment variable not set, consider setting it to identify your requests.\n"
-     ]
-    }
-   ],
+   "outputs": [],
    "source": [
     "# /// LLM ///\n",
     "\n",
     "from langchain_ollama import ChatOllama\n",
     "\n",
     "llm = ChatOllama(\n",
-    "    model=\"llama3-groq-tool-use\",\n",
+    "    # model=\"llama3-groq-tool-use\",\n",
+    "    model=\"llama3.1\",\n",
     "    temperature=0,\n",
     ")\n",
     "\n",
@@ -129,14 +134,13 @@
     "        for d in web_results\n",
     "    ]\n",
     "\n",
-    "\n",
     "# Tool list\n",
     "tools = [retrieve_documents, web_search]"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": 9,
    "id": "30052f47-2b5d-46f5-9873-eb716145cda1",
    "metadata": {},
    "outputs": [],
@@ -148,11 +152,9 @@
     "from langgraph.graph.message import AnyMessage, add_messages\n",
     "from typing_extensions import TypedDict\n",
     "\n",
-    "\n",
     "class State(TypedDict):\n",
     "    messages: Annotated[list[AnyMessage], add_messages]\n",
     "\n",
-    "\n",
     "class Assistant:\n",
     "    def __init__(self, runnable: Runnable):\n",
     "        \"\"\"\n",
@@ -209,7 +211,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 3,
+   "execution_count": 10,
    "id": "40504a0b-8a99-4420-a6bf-561c62e893d1",
    "metadata": {},
    "outputs": [
@@ -282,7 +284,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 4,
+   "execution_count": 11,
    "id": "43c633d5-e7a7-4b7c-8dc7-760a3b032e95",
    "metadata": {},
    "outputs": [],
@@ -301,9 +303,19 @@
     "response = predict_react_agent_answer(example)"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "bf82fa52-9e6c-4f37-94ae-91450dac602e",
+   "metadata": {},
+   "source": [
+    "See trace with llama3.1 here:\n",
+    "\n",
+    "https://smith.langchain.com/public/44d0c7dd-a756-47ad-8025-ee7ae6469ecb/r"
+   ]
+  },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 13,
    "id": "cd74a0b3-be40-46cd-97bf-ef9676878289",
    "metadata": {},
    "outputs": [],
@@ -311,6 +323,24 @@
     "example = {\"input\": \"Get me information about the current weather in SF.\"}\n",
     "response = predict_react_agent_answer(example)"
    ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "8cac91bf-c975-44a2-a9fd-99706fee5735",
+   "metadata": {},
+   "source": [
+    "See trace with llama3.1 here:\n",
+    "\n",
+    "https://smith.langchain.com/public/7a4938e3-f94f-4e04-a162-bf592fba4643/r"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "74b813cb-18ed-42d8-b313-6ee56ded4bcc",
+   "metadata": {},
+   "outputs": [],
+   "source": []
   }
  ],
  "metadata": {

examples/visualization.ipynb

@@ -268,7 +268,7 @@
    ],
    "source": [
     "from IPython.display import Image, display\n",
-    "from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeColors\n",
+    "from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles\n",
     "\n",
     "display(\n",
     "    Image(\n",
@@ -340,7 +340,7 @@
     "    Image(\n",
     "        app.get_graph().draw_mermaid_png(\n",
     "            curve_style=CurveStyle.LINEAR,\n",
-    "            node_colors=NodeColors(start=\"#ffdfba\", end=\"#baffc9\", other=\"#fad7de\"),\n",
+    "            node_colors=NodeStyles(first=\"#ffdfba\", last=\"#baffc9\", default=\"#fad7de\"),\n",
     "            wrap_label_n_words=9,\n",
     "            output_file_path=None,\n",
     "            draw_method=MermaidDrawMethod.PYPPETEER,\n",

libs/langgraph/README.md

@@ -3,7 +3,6 @@
 ![Version](https://img.shields.io/pypi/v/langgraph)
 [![Downloads](https://static.pepy.tech/badge/langgraph/month)](https://pepy.tech/project/langgraph)
 [![Open Issues](https://img.shields.io/github/issues-raw/langchain-ai/langgraph)](https://github.com/langchain-ai/langgraph/issues)
-[![](https://dcbadge.vercel.app/api/server/6adMQxSpJS?compact=true&style=flat)](https://discord.com/channels/1038097195422978059/1170024642245832774)
 [![Docs](https://img.shields.io/badge/docs-latest-blue)](https://langchain-ai.github.io/langgraph/)
 
 ⚡ Building language agents as graphs ⚡

libs/langgraph/langgraph/channels/manager.py

@@ -44,19 +44,22 @@ async def AsyncChannelsManager(
 
 def create_checkpoint(
     checkpoint: Checkpoint,
-    channels: Mapping[str, BaseChannel],
+    channels: Optional[Mapping[str, BaseChannel]],
     step: int,
     *,
     id: Optional[str] = None,
 ) -> Checkpoint:
     """Create a checkpoint for the given channels."""
     ts = datetime.now(timezone.utc).isoformat()
-    values: dict[str, Any] = {}
-    for k, v in channels.items():
-        try:
-            values[k] = v.checkpoint()
-        except EmptyChannelError:
-            pass
+    if channels is None:
+        values = checkpoint["channel_values"]
+    else:
+        values: dict[str, Any] = {}
+        for k, v in channels.items():
+            try:
+                values[k] = v.checkpoint()
+            except EmptyChannelError:
+                pass
     return Checkpoint(
         v=1,
         ts=ts,

libs/langgraph/langgraph/graph/state.py

@@ -452,7 +452,11 @@ class CompiledStateGraph(CompiledGraph):
     def get_input_schema(
         self, config: Optional[RunnableConfig] = None
     ) -> type[BaseModel]:
-        if isclass(self.builder.input) and issubclass(self.builder.input, BaseModel):
+        from pydantic import BaseModel as BaseModelP
+
+        if isclass(self.builder.input) and issubclass(
+            self.builder.input, (BaseModel, BaseModelP)
+        ):
             return self.builder.input
         else:
             keys = list(self.builder.schemas[self.builder.input].keys())
@@ -475,7 +479,11 @@ class CompiledStateGraph(CompiledGraph):
     def get_output_schema(
         self, config: Optional[RunnableConfig] = None
     ) -> type[BaseModel]:
-        if isclass(self.builder.input) and issubclass(self.builder.output, BaseModel):
+        from pydantic import BaseModel as BaseModelP
+
+        if isclass(self.builder.input) and issubclass(
+            self.builder.output, (BaseModel, BaseModelP)
+        ):
             return self.builder.output
 
         return super().get_output_schema(config)
@@ -497,7 +505,7 @@ class CompiledStateGraph(CompiledGraph):
                 return SKIP_WRITE
             elif isinstance(input, dict):
                 return input.get(key, SKIP_WRITE)
-            elif get_type_hints(type(input)).get(key):
+            elif get_type_hints(type(input)):
                 value = getattr(input, key, SKIP_WRITE)
                 return value if value is not None else SKIP_WRITE
             else:
@@ -602,7 +610,7 @@ class CompiledStateGraph(CompiledGraph):
                 if branch.then and branch.then != END:
                     writes.append(
                         ChannelWriteEntry(
-                            f"branch:{start}:{name}:then",
+                            f"branch:{start}:{name}::then",
                             WaitForNames(
                                 {p.node if isinstance(p, Send) else p for p in filtered}
                             ),
@@ -622,12 +630,12 @@ class CompiledStateGraph(CompiledGraph):
         for end in ends:
             if end != END:
                 channel_name = f"branch:{start}:{name}:{end}"
-                self.channels[channel_name] = EphemeralValue(Any)
+                self.channels[channel_name] = EphemeralValue(Any, guard=False)
                 self.nodes[end].triggers.append(channel_name)
 
         # attach then subscriber
         if branch.then and branch.then != END:
-            channel_name = f"branch:{start}:{name}:then"
+            channel_name = f"branch:{start}:{name}::then"
             self.channels[channel_name] = DynamicBarrierValue(str)
             self.nodes[branch.then].triggers.append(channel_name)
             for end in ends:

libs/langgraph/langgraph/pregel/__init__.py

@@ -504,7 +504,7 @@ class Pregel(
     def update_state(
         self,
         config: RunnableConfig,
-        values: dict[str, Any] | Any,
+        values: Optional[Union[dict[str, Any], Any]],
         as_node: Optional[str] = None,
     ) -> RunnableConfig:
         """Update the state of the graph with the given values, as if they came from
@@ -517,8 +517,28 @@ class Pregel(
         # get last checkpoint
         saved = self.checkpointer.get_tuple(config)
         checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
+        step = saved.metadata.get("step", -1) if saved else -1
+        # merge configurable fields with previous checkpoint config
+        checkpoint_config = config
+        if saved:
+            checkpoint_config = {
+                "configurable": {
+                    **config.get("configurable", {}),
+                    **saved.config["configurable"],
+                }
+            }
         # find last node that updated the state, if not provided
-        if as_node is None and not any(
+        if values is None and as_node is None:
+            return self.checkpointer.put(
+                checkpoint_config,
+                create_checkpoint(checkpoint, None, step),
+                {
+                    "source": "update",
+                    "step": step,
+                    "writes": {},
+                },
+            )
+        elif as_node is None and not any(
             v for vv in checkpoint["versions_seen"].values() for v in vv.values()
         ):
             if (
@@ -577,24 +597,13 @@ class Pregel(
             apply_writes(
                 checkpoint, channels, [task], self.checkpointer.get_next_version
             )
-            step = saved.metadata.get("step", -2) + 1 if saved else -1
-
-            # merge configurable fields with previous checkpoint config
-            checkpoint_config = config
-            if saved:
-                checkpoint_config = {
-                    "configurable": {
-                        **config.get("configurable", {}),
-                        **saved.config["configurable"],
-                    }
-                }
 
             return self.checkpointer.put(
                 checkpoint_config,
-                create_checkpoint(checkpoint, channels, step),
+                create_checkpoint(checkpoint, channels, step + 1),
                 {
                     "source": "update",
-                    "step": step,
+                    "step": step + 1,
                     "writes": {as_node: values},
                 },
             )

libs/langgraph/poetry.lock

@@ -1760,13 +1760,13 @@ langchain-core = ">=0.2.2rc1,<0.3"
 
 [[package]]
 name = "langchain-core"
-version = "0.2.22"
+version = "0.2.25"
 description = "Building applications with LLMs through composability"
 optional = false
 python-versions = "<4.0,>=3.8.1"
 files = [
-    {file = "langchain_core-0.2.22-py3-none-any.whl", hash = "sha256:7731a86440c0958b3186c003fb9b26b2d5a682a6344bda7bfb9174e2898f8b43"},
-    {file = "langchain_core-0.2.22.tar.gz", hash = "sha256:582d6f929a43b830139444e4124123cd415331ad62f25757b1406252958cdcac"},
+    {file = "langchain_core-0.2.25-py3-none-any.whl", hash = "sha256:03d61b2a7f4b5f98df248c1b1f0ccd95c9d5ef2269e174133724365cd2a7ee1e"},
+    {file = "langchain_core-0.2.25.tar.gz", hash = "sha256:e64106a7d0e37e4d35b767f79e6c62b56e825f08f9e8cc4368bcea9955257a7e"},
 ]
 
 [package.dependencies]

libs/langgraph/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph"
-version = "0.1.14"
+version = "0.1.17"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 license = "MIT"

libs/langgraph/tests/test_pregel.py

@@ -549,172 +549,380 @@ def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
     assert step == 2
 
 
-def test_invoke_two_processes_in_out_interrupt(mocker: MockerFixture) -> None:
-    add_one = mocker.Mock(side_effect=lambda x: x + 1)
-    one = Channel.subscribe_to("input") | add_one | Channel.write_to("inbox")
-    two = Channel.subscribe_to("inbox") | add_one | Channel.write_to("output")
+@pytest.mark.parametrize(
+    "checkpointer",
+    [
+        MemorySaverAssertImmutable(),
+        SqliteSaver.from_conn_string(":memory:"),
+    ],
+    ids=[
+        "memory",
+        "sqlite",
+    ],
+)
+def test_invoke_two_processes_in_out_interrupt(
+    checkpointer: BaseCheckpointSaver, mocker: MockerFixture
+) -> None:
+    try:
+        add_one = mocker.Mock(side_effect=lambda x: x + 1)
+        one = Channel.subscribe_to("input") | add_one | Channel.write_to("inbox")
+        two = Channel.subscribe_to("inbox") | add_one | Channel.write_to("output")
 
-    memory = MemorySaverAssertImmutable()
-    app = Pregel(
-        nodes={"one": one, "two": two},
-        channels={
-            "inbox": LastValue(int),
-            "output": LastValue(int),
-            "input": LastValue(int),
-        },
-        input_channels="input",
-        output_channels="output",
-        checkpointer=memory,
-        interrupt_after_nodes=["one"],
-    )
-
-    # start execution, stop at inbox
-    assert app.invoke(2, {"configurable": {"thread_id": 1}}) is None
-
-    # inbox == 3
-    checkpoint = memory.get({"configurable": {"thread_id": 1}})
-    assert checkpoint is not None
-    assert checkpoint["channel_values"]["inbox"] == 3
-
-    # resume execution, finish
-    assert app.invoke(None, {"configurable": {"thread_id": 1}}) == 4
-
-    # start execution again, stop at inbox
-    assert app.invoke(20, {"configurable": {"thread_id": 1}}) is None
-
-    # inbox == 21
-    checkpoint = memory.get({"configurable": {"thread_id": 1}})
-    assert checkpoint is not None
-    assert checkpoint["channel_values"]["inbox"] == 21
-
-    # send a new value in, interrupting the previous execution
-    assert app.invoke(3, {"configurable": {"thread_id": 1}}) is None
-    assert app.invoke(None, {"configurable": {"thread_id": 1}}) == 5
-
-    # start execution again, stopping at inbox
-    assert app.invoke(20, {"configurable": {"thread_id": 2}}) is None
-
-    # inbox == 21
-    snapshot = app.get_state({"configurable": {"thread_id": 2}})
-    assert snapshot.values["inbox"] == 21
-    assert snapshot.next == ("two",)
-
-    # update the state, resume
-    app.update_state({"configurable": {"thread_id": 2}}, 25, as_node="one")
-    assert app.invoke(None, {"configurable": {"thread_id": 2}}) == 26
-
-    # no pending tasks
-    snapshot = app.get_state({"configurable": {"thread_id": 2}})
-    assert snapshot.next == ()
-
-    # list history
-    thread1 = {"configurable": {"thread_id": 1}}
-    assert [c for c in app.get_state_history(thread1)] == [
-        StateSnapshot(
-            values={"inbox": 4, "output": 5, "input": 3},
-            next=(),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
+        app = Pregel(
+            nodes={"one": one, "two": two},
+            channels={
+                "inbox": LastValue(int),
+                "output": LastValue(int),
+                "input": LastValue(int),
             },
-            metadata={"source": "loop", "step": 6, "writes": 5},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][1].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 4, "output": 4, "input": 3},
-            next=("two",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 5, "writes": None},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][2].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 21, "output": 4, "input": 3},
-            next=("one",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "input", "step": 4, "writes": 3},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][3].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 21, "output": 4, "input": 20},
-            next=("two",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 3, "writes": None},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][4].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 3, "output": 4, "input": 20},
-            next=("one",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "input", "step": 2, "writes": 20},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][5].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 3, "output": 4, "input": 2},
-            next=(),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 1, "writes": 4},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][6].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 3, "input": 2},
-            next=("two",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 0, "writes": None},
-            created_at=AnyStr(),
-            parent_config=[*app.checkpointer.list(thread1)][7].config,
-        ),
-        StateSnapshot(
-            values={"input": 2},
-            next=("one",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "input", "step": -1, "writes": 2},
-            created_at=AnyStr(),
-            parent_config=None,
-        ),
-    ]
+            input_channels="input",
+            output_channels="output",
+            checkpointer=checkpointer,
+            interrupt_after_nodes=["one"],
+        )
+        thread1 = {"configurable": {"thread_id": "1"}}
+        thread2 = {"configurable": {"thread_id": "2"}}
+
+        # start execution, stop at inbox
+        assert app.invoke(2, thread1) is None
+
+        # inbox == 3
+        checkpoint = checkpointer.get(thread1)
+        assert checkpoint is not None
+        assert checkpoint["channel_values"]["inbox"] == 3
+
+        # resume execution, finish
+        assert app.invoke(None, thread1) == 4
+
+        # start execution again, stop at inbox
+        assert app.invoke(20, thread1) is None
+
+        # inbox == 21
+        checkpoint = checkpointer.get(thread1)
+        assert checkpoint is not None
+        assert checkpoint["channel_values"]["inbox"] == 21
+
+        # send a new value in, interrupting the previous execution
+        assert app.invoke(3, thread1) is None
+        assert app.invoke(None, thread1) == 5
+
+        # start execution again, stopping at inbox
+        assert app.invoke(20, thread2) is None
+
+        # inbox == 21
+        snapshot = app.get_state(thread2)
+        assert snapshot.values["inbox"] == 21
+        assert snapshot.next == ("two",)
+
+        # update the state, resume
+        app.update_state(thread2, 25, as_node="one")
+        assert app.invoke(None, thread2) == 26
+
+        # no pending tasks
+        snapshot = app.get_state(thread2)
+        assert snapshot.next == ()
+
+        # list history
+        history = [c for c in app.get_state_history(thread1)]
+        assert history == [
+            StateSnapshot(
+                values={"inbox": 4, "output": 5, "input": 3},
+                next=(),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 6, "writes": 5},
+                created_at=AnyStr(),
+                parent_config=history[1].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 4, "output": 4, "input": 3},
+                next=("two",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 5, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[2].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 21, "output": 4, "input": 3},
+                next=("one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": 4, "writes": 3},
+                created_at=AnyStr(),
+                parent_config=history[3].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 21, "output": 4, "input": 20},
+                next=("two",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 3, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[4].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 3, "output": 4, "input": 20},
+                next=("one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": 2, "writes": 20},
+                created_at=AnyStr(),
+                parent_config=history[5].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 3, "output": 4, "input": 2},
+                next=(),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 1, "writes": 4},
+                created_at=AnyStr(),
+                parent_config=history[6].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 3, "input": 2},
+                next=("two",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 0, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[7].config,
+            ),
+            StateSnapshot(
+                values={"input": 2},
+                next=("one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": -1, "writes": 2},
+                created_at=AnyStr(),
+                parent_config=None,
+            ),
+        ]
+
+        # forking from any previous checkpoint w/out forking should do nothing
+        assert [
+            c for c in app.stream(None, history[0].config, stream_mode="updates")
+        ] == []
+        assert [
+            c for c in app.stream(None, history[1].config, stream_mode="updates")
+        ] == []
+        assert [
+            c for c in app.stream(None, history[2].config, stream_mode="updates")
+        ] == []
+
+        # forking and re-running from any prev checkpoint should re-run nodes
+        fork_config = app.update_state(history[0].config, None)
+        assert [c for c in app.stream(None, fork_config, stream_mode="updates")] == []
+
+        fork_config = app.update_state(history[1].config, None)
+        assert [c for c in app.stream(None, fork_config, stream_mode="updates")] == [
+            {"two": {"output": 5}}
+        ]
+
+        fork_config = app.update_state(history[2].config, None)
+        assert [c for c in app.stream(None, fork_config, stream_mode="updates")] == [
+            {"one": {"inbox": 4}}
+        ]
+    finally:
+        if hasattr(checkpointer, "__exit__"):
+            checkpointer.__exit__(None, None, None)
+
+
+@pytest.mark.parametrize(
+    "checkpointer",
+    [
+        MemorySaverAssertImmutable(),
+        SqliteSaver.from_conn_string(":memory:"),
+    ],
+    ids=[
+        "memory",
+        "sqlite",
+    ],
+)
+def test_fork_always_re_runs_nodes(
+    checkpointer: BaseCheckpointSaver, mocker: MockerFixture
+) -> None:
+    try:
+        add_one = mocker.Mock(side_effect=lambda _: 1)
+
+        builder = StateGraph(Annotated[int, operator.add])
+        builder.add_node("add_one", add_one)
+        builder.add_edge(START, "add_one")
+        builder.add_conditional_edges(
+            "add_one", lambda cnt: "add_one" if cnt < 6 else END
+        )
+        graph = builder.compile(checkpointer=checkpointer)
+
+        thread1 = {"configurable": {"thread_id": "1"}}
+
+        # start execution, stop at inbox
+        assert [*graph.stream(1, thread1, stream_mode=["values", "updates"])] == [
+            ("values", 1),
+            ("updates", {"add_one": 1}),
+            ("values", 2),
+            ("updates", {"add_one": 1}),
+            ("values", 3),
+            ("updates", {"add_one": 1}),
+            ("values", 4),
+            ("updates", {"add_one": 1}),
+            ("values", 5),
+            ("updates", {"add_one": 1}),
+            ("values", 6),
+        ]
+
+        # list history
+        history = [c for c in graph.get_state_history(thread1)]
+        assert history == [
+            StateSnapshot(
+                values=6,
+                next=(),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 5, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[1].config,
+            ),
+            StateSnapshot(
+                values=5,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 4, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[2].config,
+            ),
+            StateSnapshot(
+                values=4,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 3, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[3].config,
+            ),
+            StateSnapshot(
+                values=3,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 2, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[4].config,
+            ),
+            StateSnapshot(
+                values=2,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 1, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[5].config,
+            ),
+            StateSnapshot(
+                values=1,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 0, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[6].config,
+            ),
+            StateSnapshot(
+                values=0,
+                next=("__start__",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": -1, "writes": 1},
+                created_at=AnyStr(),
+                parent_config=None,
+            ),
+        ]
+
+        # forking from any previous checkpoint w/out forking should do nothing
+        assert [
+            c for c in graph.stream(None, history[0].config, stream_mode="updates")
+        ] == []
+        assert [
+            c for c in graph.stream(None, history[1].config, stream_mode="updates")
+        ] == []
+
+        # forking and re-running from any prev checkpoint should re-run nodes
+        fork_config = graph.update_state(history[0].config, None)
+        assert [c for c in graph.stream(None, fork_config, stream_mode="updates")] == []
+
+        fork_config = graph.update_state(history[1].config, None)
+        assert [c for c in graph.stream(None, fork_config, stream_mode="updates")] == [
+            {"add_one": 1}
+        ]
+
+        fork_config = graph.update_state(history[2].config, None)
+        assert [c for c in graph.stream(None, fork_config, stream_mode="updates")] == [
+            {"add_one": 1},
+            {"add_one": 1},
+        ]
+    finally:
+        if hasattr(checkpointer, "__exit__"):
+            checkpointer.__exit__(None, None, None)
 
 
 def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
@@ -1134,15 +1342,15 @@ def test_cond_edge_after_send() -> None:
             setattr(self, "__name__", name)
 
         def __call__(self, state):
-            return state + [self.name]
+            return [self.name]
 
     def send_for_fun(state):
-        return [Send("2", state)]
+        return [Send("2", state), Send("2", state)]
 
     def route_to_three(state) -> Literal["3"]:
         return "3"
 
-    builder = StateGraph(list)
+    builder = StateGraph(Annotated[list, operator.add])
     builder.add_node(Node("1"))
     builder.add_node(Node("2"))
     builder.add_node(Node("3"))
@@ -1150,7 +1358,7 @@ def test_cond_edge_after_send() -> None:
     builder.add_conditional_edges("1", send_for_fun)
     builder.add_conditional_edges("2", route_to_three)
     graph = builder.compile()
-    assert graph.invoke(["0"]) == ["0", "1", "2", "3"]
+    assert graph.invoke(["0"]) == ["0", "1", "2", "2", "3"]
 
 
 async def test_checkpointer_null_pending_writes() -> None:
@@ -6536,10 +6744,10 @@ def test_branch_then(snapshot: SnapshotAssertion) -> None:
                 "timestamp": AnyStr(),
                 "step": 3,
                 "payload": {
-                    "id": "ceada3c5-5f25-59e4-9ea5-544599ce1d2f",
+                    "id": "9b590c54-15ef-54b1-83a7-140d27b0bc52",
                     "name": "finish",
                     "input": {"my_key": "value prepared slow", "market": "DE"},
-                    "triggers": ["branch:prepare:condition:then"],
+                    "triggers": ["branch:prepare:condition::then"],
                 },
             },
             {
@@ -6547,7 +6755,7 @@ def test_branch_then(snapshot: SnapshotAssertion) -> None:
                 "timestamp": AnyStr(),
                 "step": 3,
                 "payload": {
-                    "id": "ceada3c5-5f25-59e4-9ea5-544599ce1d2f",
+                    "id": "9b590c54-15ef-54b1-83a7-140d27b0bc52",
                     "name": "finish",
                     "result": [("my_key", " finished")],
                 },
@@ -6798,7 +7006,7 @@ def test_branch_then(snapshot: SnapshotAssertion) -> None:
             created_at=AnyStr(),
             metadata={
                 "source": "update",
-                "step": -1,
+                "step": 0,
                 "writes": {START: {"my_key": "key", "market": "DE"}},
             },
         )
@@ -6815,7 +7023,7 @@ def test_branch_then(snapshot: SnapshotAssertion) -> None:
             created_at=tool_two.checkpointer.get_tuple(thread3).checkpoint["ts"],
             metadata={
                 "source": "loop",
-                "step": 0,
+                "step": 1,
                 "writes": {"prepare": {"my_key": " prepared"}},
             },
             parent_config=uconfig,
@@ -6832,7 +7040,7 @@ def test_branch_then(snapshot: SnapshotAssertion) -> None:
             created_at=tool_two.checkpointer.get_tuple(thread3).checkpoint["ts"],
             metadata={
                 "source": "loop",
-                "step": 2,
+                "step": 3,
                 "writes": {"finish": {"my_key": " finished"}},
             },
             parent_config=[*tool_two.checkpointer.list(thread3, limit=2)][-1].config,
@@ -7048,7 +7256,7 @@ def test_in_one_fan_out_state_graph_waiting_edge_via_branch(
     ]
 
 
-def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class(
+def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic1(
     snapshot: SnapshotAssertion,
 ) -> None:
     from langchain_core.pydantic_v1 import BaseModel, ValidationError
@@ -7062,8 +7270,12 @@ def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class(
             y = [t[1] for t in y]
         return sorted(operator.add(x, y))
 
+    class InnerObject(BaseModel):
+        yo: int
+
     class State(BaseModel):
         query: str
+        inner: InnerObject
         answer: Optional[str] = None
         docs: Annotated[list[str], sorted_add]
 
@@ -7112,17 +7324,20 @@ def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class(
     app = workflow.compile()
 
     assert app.get_graph().draw_mermaid(with_styles=False) == snapshot
+    assert app.get_input_schema().schema() == snapshot
+    assert app.get_output_schema().schema() == snapshot
 
     with pytest.raises(ValidationError):
         app.invoke({"query": {}})
 
-    assert app.invoke({"query": "what is weather in sf"}) == {
+    assert app.invoke({"query": "what is weather in sf", "inner": {"yo": 1}}) == {
         "query": "analyzed: query: what is weather in sf",
         "docs": ["doc1", "doc2", "doc3", "doc4"],
         "answer": "doc1,doc2,doc3,doc4",
+        "inner": {"yo": 1},
     }
 
-    assert [*app.stream({"query": "what is weather in sf"})] == [
+    assert [*app.stream({"query": "what is weather in sf", "inner": {"yo": 1}})] == [
         {"rewrite_query": {"query": "query: what is weather in sf"}},
         {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}},
         {"retriever_two": {"docs": ["doc3", "doc4"]}},
@@ -7137,7 +7352,122 @@ def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class(
     config = {"configurable": {"thread_id": "1"}}
 
     assert [
-        c for c in app_w_interrupt.stream({"query": "what is weather in sf"}, config)
+        c
+        for c in app_w_interrupt.stream(
+            {"query": "what is weather in sf", "inner": {"yo": 1}}, config
+        )
+    ] == [
+        {"rewrite_query": {"query": "query: what is weather in sf"}},
+        {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}},
+        {"retriever_two": {"docs": ["doc3", "doc4"]}},
+        {"retriever_one": {"docs": ["doc1", "doc2"]}},
+    ]
+
+    assert [c for c in app_w_interrupt.stream(None, config)] == [
+        {"qa": {"answer": "doc1,doc2,doc3,doc4"}},
+    ]
+
+
+def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class_pydantic2(
+    snapshot: SnapshotAssertion,
+) -> None:
+    from pydantic import BaseModel, ValidationError
+
+    def sorted_add(
+        x: list[str], y: Union[list[str], list[tuple[str, str]]]
+    ) -> list[str]:
+        if isinstance(y[0], tuple):
+            for rem, _ in y:
+                x.remove(rem)
+            y = [t[1] for t in y]
+        return sorted(operator.add(x, y))
+
+    class InnerObject(BaseModel):
+        yo: int
+
+    class State(BaseModel):
+        query: str
+        inner: InnerObject
+        answer: Optional[str] = None
+        docs: Annotated[list[str], sorted_add]
+
+    class StateUpdate(BaseModel):
+        query: Optional[str] = None
+        answer: Optional[str] = None
+        docs: Optional[list[str]] = None
+
+    def rewrite_query(data: State) -> State:
+        return {"query": f"query: {data.query}"}
+
+    def analyzer_one(data: State) -> State:
+        return StateUpdate(query=f"analyzed: {data.query}")
+
+    def retriever_one(data: State) -> State:
+        return {"docs": ["doc1", "doc2"]}
+
+    def retriever_two(data: State) -> State:
+        time.sleep(0.1)
+        return {"docs": ["doc3", "doc4"]}
+
+    def qa(data: State) -> State:
+        return {"answer": ",".join(data.docs)}
+
+    def decider(data: State) -> str:
+        assert isinstance(data, State)
+        return "retriever_two"
+
+    workflow = StateGraph(State)
+
+    workflow.add_node("rewrite_query", rewrite_query)
+    workflow.add_node("analyzer_one", analyzer_one)
+    workflow.add_node("retriever_one", retriever_one)
+    workflow.add_node("retriever_two", retriever_two)
+    workflow.add_node("qa", qa)
+
+    workflow.set_entry_point("rewrite_query")
+    workflow.add_edge("rewrite_query", "analyzer_one")
+    workflow.add_edge("analyzer_one", "retriever_one")
+    workflow.add_conditional_edges(
+        "rewrite_query", decider, {"retriever_two": "retriever_two"}
+    )
+    workflow.add_edge(["retriever_one", "retriever_two"], "qa")
+    workflow.set_finish_point("qa")
+
+    app = workflow.compile()
+
+    assert app.get_graph().draw_mermaid(with_styles=False) == snapshot
+    assert app.get_input_schema().schema() == snapshot
+    assert app.get_output_schema().schema() == snapshot
+
+    with pytest.raises(ValidationError):
+        app.invoke({"query": {}})
+
+    assert app.invoke({"query": "what is weather in sf", "inner": {"yo": 1}}) == {
+        "query": "analyzed: query: what is weather in sf",
+        "docs": ["doc1", "doc2", "doc3", "doc4"],
+        "answer": "doc1,doc2,doc3,doc4",
+        "inner": {"yo": 1},
+    }
+
+    assert [*app.stream({"query": "what is weather in sf", "inner": {"yo": 1}})] == [
+        {"rewrite_query": {"query": "query: what is weather in sf"}},
+        {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}},
+        {"retriever_two": {"docs": ["doc3", "doc4"]}},
+        {"retriever_one": {"docs": ["doc1", "doc2"]}},
+        {"qa": {"answer": "doc1,doc2,doc3,doc4"}},
+    ]
+
+    app_w_interrupt = workflow.compile(
+        checkpointer=MemorySaverAssertImmutable(),
+        interrupt_after=["retriever_one"],
+    )
+    config = {"configurable": {"thread_id": "1"}}
+
+    assert [
+        c
+        for c in app_w_interrupt.stream(
+            {"query": "what is weather in sf", "inner": {"yo": 1}}, config
+        )
     ] == [
         {"rewrite_query": {"query": "query: what is weather in sf"}},
         {"analyzer_one": {"query": "analyzed: query: what is weather in sf"}},
@@ -8975,3 +9305,130 @@ def test_remove_message_from_node():
     output = app.invoke([HumanMessage(content="Hi")])
     assert len(output) == 2
     assert output[-1].content == "How can I help you?"
+
+
+def test_xray_lance(snapshot: SnapshotAssertion):
+    from langchain_core.messages import AnyMessage, HumanMessage
+    from langchain_core.pydantic_v1 import BaseModel, Field
+
+    class Analyst(BaseModel):
+        affiliation: str = Field(
+            description="Primary affiliation of the investment analyst.",
+        )
+        name: str = Field(
+            description="Name of the investment analyst.",
+            pattern=r"^[a-zA-Z0-9_-]{1,64}$",
+        )
+        role: str = Field(
+            description="Role of the investment analyst in the context of the topic.",
+        )
+        description: str = Field(
+            description="Description of the investment analyst focus, concerns, and motives.",
+        )
+
+        @property
+        def persona(self) -> str:
+            return f"Name: {self.name}\nRole: {self.role}\nAffiliation: {self.affiliation}\nDescription: {self.description}\n"
+
+    class Perspectives(BaseModel):
+        analysts: List[Analyst] = Field(
+            description="Comprehensive list of investment analysts with their roles and affiliations.",
+        )
+
+    class Section(BaseModel):
+        section_title: str = Field(..., title="Title of the section")
+        context: str = Field(
+            ..., title="Provide a clear summary of the focus area that you researched."
+        )
+        findings: str = Field(
+            ...,
+            title="Give a clear and detailed overview of your findings based upon the expert interview.",
+        )
+        thesis: str = Field(
+            ...,
+            title="Give a clear and specific investment thesis based upon these findings.",
+        )
+
+    class InterviewState(TypedDict):
+        messages: Annotated[List[AnyMessage], add_messages]
+        analyst: Analyst
+        section: Section
+
+    class ResearchGraphState(TypedDict):
+        analysts: List[Analyst]
+        topic: str
+        max_analysts: int
+        sections: List[Section]
+        interviews: Annotated[list, operator.add]
+
+    # Conditional edge
+    def route_messages(state):
+        return "ask_question"
+
+    def generate_question(state):
+        return ...
+
+    def generate_answer(state):
+        return ...
+
+    # Add nodes and edges
+    interview_builder = StateGraph(InterviewState)
+    interview_builder.add_node("ask_question", generate_question)
+    interview_builder.add_node("answer_question", generate_answer)
+
+    # Flow
+    interview_builder.add_edge(START, "ask_question")
+    interview_builder.add_edge("ask_question", "answer_question")
+    interview_builder.add_conditional_edges("answer_question", route_messages)
+
+    # Set up memory
+    memory = MemorySaver()
+
+    # Interview
+    interview_graph = interview_builder.compile(checkpointer=memory).with_config(
+        run_name="Conduct Interviews"
+    )
+
+    # View
+    assert interview_graph.get_graph().to_json() == snapshot
+
+    def run_all_interviews(state: ResearchGraphState):
+        """Edge to run the interview sub-graph using Send"""
+        return [
+            Send(
+                "conduct_interview",
+                {
+                    "analyst": Analyst(),
+                    "messages": [
+                        HumanMessage(
+                            content="So you said you were writing an article on ...?"
+                        )
+                    ],
+                },
+            )
+            for s in state["analysts"]
+        ]
+
+    def generate_sections(state: ResearchGraphState):
+        return ...
+
+    def generate_analysts(state: ResearchGraphState):
+        return ...
+
+    builder = StateGraph(ResearchGraphState)
+    builder.add_node("generate_analysts", generate_analysts)
+    builder.add_node("conduct_interview", interview_builder.compile())
+    builder.add_node("generate_sections", generate_sections)
+
+    builder.add_edge(START, "generate_analysts")
+    builder.add_conditional_edges(
+        "generate_analysts", run_all_interviews, ["conduct_interview"]
+    )
+    builder.add_edge("conduct_interview", "generate_sections")
+    builder.add_edge("generate_sections", END)
+
+    graph = builder.compile()
+
+    # View
+    assert graph.get_graph().to_json() == snapshot
+    assert graph.get_graph(xray=1).to_json() == snapshot

libs/langgraph/tests/test_pregel_async.py

@@ -665,172 +665,364 @@ async def test_invoke_two_processes_in_out(mocker: MockerFixture) -> None:
     assert step == 2
 
 
-async def test_invoke_two_processes_in_out_interrupt(mocker: MockerFixture) -> None:
-    add_one = mocker.Mock(side_effect=lambda x: x + 1)
-    one = Channel.subscribe_to("input") | add_one | Channel.write_to("inbox")
-    two = Channel.subscribe_to("inbox") | add_one | Channel.write_to("output")
+@pytest.mark.parametrize(
+    "checkpointer",
+    [
+        MemorySaverAssertImmutable(),
+        AsyncSqliteSaver.from_conn_string(":memory:"),
+    ],
+    ids=[
+        "memory",
+        "sqlite",
+    ],
+)
+async def test_invoke_two_processes_in_out_interrupt(
+    checkpointer: BaseCheckpointSaver, mocker: MockerFixture
+) -> None:
+    try:
+        add_one = mocker.Mock(side_effect=lambda x: x + 1)
+        one = Channel.subscribe_to("input") | add_one | Channel.write_to("inbox")
+        two = Channel.subscribe_to("inbox") | add_one | Channel.write_to("output")
 
-    memory = MemorySaverAssertImmutable()
-    app = Pregel(
-        nodes={"one": one, "two": two},
-        channels={
-            "inbox": LastValue(int),
-            "output": LastValue(int),
-            "input": LastValue(int),
-        },
-        input_channels="input",
-        output_channels="output",
-        checkpointer=memory,
-        interrupt_after_nodes=["one"],
-    )
-
-    # start execution, stop at inbox
-    assert await app.ainvoke(2, {"configurable": {"thread_id": 1}}) is None
-
-    # inbox == 3
-    checkpoint = await memory.aget({"configurable": {"thread_id": 1}})
-    assert checkpoint is not None
-    assert checkpoint["channel_values"]["inbox"] == 3
-
-    # resume execution, finish
-    assert await app.ainvoke(None, {"configurable": {"thread_id": 1}}) == 4
-
-    # start execution again, stop at inbox
-    assert await app.ainvoke(20, {"configurable": {"thread_id": 1}}) is None
-
-    # inbox == 21
-    checkpoint = await memory.aget({"configurable": {"thread_id": 1}})
-    assert checkpoint is not None
-    assert checkpoint["channel_values"]["inbox"] == 21
-
-    # send a new value in, interrupting the previous execution
-    assert await app.ainvoke(3, {"configurable": {"thread_id": 1}}) is None
-    assert await app.ainvoke(None, {"configurable": {"thread_id": 1}}) == 5
-
-    # start execution again, stopping at inbox
-    assert await app.ainvoke(20, {"configurable": {"thread_id": 2}}) is None
-
-    # inbox == 21
-    snapshot = await app.aget_state({"configurable": {"thread_id": 2}})
-    assert snapshot.values["inbox"] == 21
-    assert snapshot.next == ("two",)
-
-    # update the state, resume
-    await app.aupdate_state({"configurable": {"thread_id": 2}}, 25, as_node="one")
-    assert await app.ainvoke(None, {"configurable": {"thread_id": 2}}) == 26
-
-    # no pending tasks
-    snapshot = await app.aget_state({"configurable": {"thread_id": 2}})
-    assert snapshot.next == ()
-
-    # list history
-    thread1 = {"configurable": {"thread_id": 1}}
-    assert [c async for c in app.aget_state_history(thread1)] == [
-        StateSnapshot(
-            values={"inbox": 4, "output": 5, "input": 3},
-            next=(),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
+        app = Pregel(
+            nodes={"one": one, "two": two},
+            channels={
+                "inbox": LastValue(int),
+                "output": LastValue(int),
+                "input": LastValue(int),
             },
-            metadata={"source": "loop", "step": 6, "writes": 5},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][1].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 4, "output": 4, "input": 3},
-            next=("two",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 5, "writes": None},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][2].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 21, "output": 4, "input": 3},
-            next=("one",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "input", "step": 4, "writes": 3},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][3].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 21, "output": 4, "input": 20},
-            next=("two",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 3, "writes": None},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][4].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 3, "output": 4, "input": 20},
-            next=("one",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "input", "step": 2, "writes": 20},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][5].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 3, "output": 4, "input": 2},
-            next=(),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 1, "writes": 4},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][6].config,
-        ),
-        StateSnapshot(
-            values={"inbox": 3, "input": 2},
-            next=("two",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "loop", "step": 0, "writes": None},
-            created_at=AnyStr(),
-            parent_config=[c async for c in app.checkpointer.alist(thread1)][7].config,
-        ),
-        StateSnapshot(
-            values={"input": 2},
-            next=("one",),
-            config={
-                "configurable": {
-                    "thread_id": 1,
-                    "thread_ts": AnyStr(),
-                }
-            },
-            metadata={"source": "input", "step": -1, "writes": 2},
-            created_at=AnyStr(),
-            parent_config=None,
-        ),
-    ]
+            input_channels="input",
+            output_channels="output",
+            checkpointer=checkpointer,
+            interrupt_after_nodes=["one"],
+        )
+        thread1 = {"configurable": {"thread_id": "1"}}
+        thread2 = {"configurable": {"thread_id": "2"}}
+
+        # start execution, stop at inbox
+        assert await app.ainvoke(2, thread1) is None
+
+        # inbox == 3
+        checkpoint = await checkpointer.aget(thread1)
+        assert checkpoint is not None
+        assert checkpoint["channel_values"]["inbox"] == 3
+
+        # resume execution, finish
+        assert await app.ainvoke(None, thread1) == 4
+
+        # start execution again, stop at inbox
+        assert await app.ainvoke(20, thread1) is None
+
+        # inbox == 21
+        checkpoint = await checkpointer.aget(thread1)
+        assert checkpoint is not None
+        assert checkpoint["channel_values"]["inbox"] == 21
+
+        # send a new value in, interrupting the previous execution
+        assert await app.ainvoke(3, thread1) is None
+        assert await app.ainvoke(None, thread1) == 5
+
+        # start execution again, stopping at inbox
+        assert await app.ainvoke(20, thread2) is None
+
+        # inbox == 21
+        snapshot = await app.aget_state(thread2)
+        assert snapshot.values["inbox"] == 21
+        assert snapshot.next == ("two",)
+
+        # update the state, resume
+        await app.aupdate_state(thread2, 25, as_node="one")
+        assert await app.ainvoke(None, thread2) == 26
+
+        # no pending tasks
+        snapshot = await app.aget_state(thread2)
+        assert snapshot.next == ()
+
+        # list history
+        history = [c async for c in app.aget_state_history(thread1)]
+        assert history == [
+            StateSnapshot(
+                values={"inbox": 4, "output": 5, "input": 3},
+                next=(),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 6, "writes": 5},
+                created_at=AnyStr(),
+                parent_config=history[1].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 4, "output": 4, "input": 3},
+                next=("two",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 5, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[2].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 21, "output": 4, "input": 3},
+                next=("one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": 4, "writes": 3},
+                created_at=AnyStr(),
+                parent_config=history[3].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 21, "output": 4, "input": 20},
+                next=("two",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 3, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[4].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 3, "output": 4, "input": 20},
+                next=("one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": 2, "writes": 20},
+                created_at=AnyStr(),
+                parent_config=history[5].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 3, "output": 4, "input": 2},
+                next=(),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 1, "writes": 4},
+                created_at=AnyStr(),
+                parent_config=history[6].config,
+            ),
+            StateSnapshot(
+                values={"inbox": 3, "input": 2},
+                next=("two",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 0, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[7].config,
+            ),
+            StateSnapshot(
+                values={"input": 2},
+                next=("one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": -1, "writes": 2},
+                created_at=AnyStr(),
+                parent_config=None,
+            ),
+        ]
+    finally:
+        if hasattr(checkpointer, "__aexit__"):
+            await checkpointer.__aexit__(None, None, None)
+
+
+@pytest.mark.parametrize(
+    "checkpointer",
+    [
+        MemorySaverAssertImmutable(),
+        AsyncSqliteSaver.from_conn_string(":memory:"),
+    ],
+    ids=[
+        "memory",
+        "sqlite",
+    ],
+)
+async def test_fork_always_re_runs_nodes(
+    checkpointer: BaseCheckpointSaver, mocker: MockerFixture
+) -> None:
+    try:
+        add_one = mocker.Mock(side_effect=lambda _: 1)
+
+        builder = StateGraph(Annotated[int, operator.add])
+        builder.add_node("add_one", add_one)
+        builder.add_edge(START, "add_one")
+        builder.add_conditional_edges(
+            "add_one", lambda cnt: "add_one" if cnt < 6 else END
+        )
+        graph = builder.compile(checkpointer=checkpointer)
+
+        thread1 = {"configurable": {"thread_id": "1"}}
+
+        # start execution, stop at inbox
+        assert [
+            c
+            async for c in graph.astream(1, thread1, stream_mode=["values", "updates"])
+        ] == [
+            ("values", 1),
+            ("updates", {"add_one": 1}),
+            ("values", 2),
+            ("updates", {"add_one": 1}),
+            ("values", 3),
+            ("updates", {"add_one": 1}),
+            ("values", 4),
+            ("updates", {"add_one": 1}),
+            ("values", 5),
+            ("updates", {"add_one": 1}),
+            ("values", 6),
+        ]
+
+        # list history
+        history = [c async for c in graph.aget_state_history(thread1)]
+        assert history == [
+            StateSnapshot(
+                values=6,
+                next=(),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 5, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[1].config,
+            ),
+            StateSnapshot(
+                values=5,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 4, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[2].config,
+            ),
+            StateSnapshot(
+                values=4,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 3, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[3].config,
+            ),
+            StateSnapshot(
+                values=3,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 2, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[4].config,
+            ),
+            StateSnapshot(
+                values=2,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 1, "writes": {"add_one": 1}},
+                created_at=AnyStr(),
+                parent_config=history[5].config,
+            ),
+            StateSnapshot(
+                values=1,
+                next=("add_one",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "loop", "step": 0, "writes": None},
+                created_at=AnyStr(),
+                parent_config=history[6].config,
+            ),
+            StateSnapshot(
+                values=0,
+                next=("__start__",),
+                config={
+                    "configurable": {
+                        "thread_id": "1",
+                        "thread_ts": AnyStr(),
+                    }
+                },
+                metadata={"source": "input", "step": -1, "writes": 1},
+                created_at=AnyStr(),
+                parent_config=None,
+            ),
+        ]
+
+        # forking from any previous checkpoint w/out forking should do nothing
+        assert [
+            c
+            async for c in graph.astream(None, history[0].config, stream_mode="updates")
+        ] == []
+        assert [
+            c
+            async for c in graph.astream(None, history[1].config, stream_mode="updates")
+        ] == []
+
+        # forking and re-running from any prev checkpoint should re-run nodes
+        fork_config = await graph.aupdate_state(history[0].config, None)
+        assert [
+            c async for c in graph.astream(None, fork_config, stream_mode="updates")
+        ] == []
+
+        fork_config = await graph.aupdate_state(history[1].config, None)
+        assert [
+            c async for c in graph.astream(None, fork_config, stream_mode="updates")
+        ] == [{"add_one": 1}]
+
+        fork_config = await graph.aupdate_state(history[2].config, None)
+        assert [
+            c async for c in graph.astream(None, fork_config, stream_mode="updates")
+        ] == [
+            {"add_one": 1},
+            {"add_one": 1},
+        ]
+    finally:
+        if hasattr(checkpointer, "__aexit__"):
+            await checkpointer.__aexit__(None, None, None)
 
 
 async def test_invoke_two_processes_in_dict_out(mocker: MockerFixture) -> None:
@@ -1257,15 +1449,15 @@ async def test_cond_edge_after_send() -> None:
             setattr(self, "__name__", name)
 
         async def __call__(self, state):
-            return state + [self.name]
+            return [self.name]
 
     async def send_for_fun(state):
-        return [Send("2", state)]
+        return [Send("2", state), Send("2", state)]
 
     async def route_to_three(state) -> Literal["3"]:
         return "3"
 
-    builder = StateGraph(list)
+    builder = StateGraph(Annotated[list, operator.add])
     builder.add_node(Node("1"))
     builder.add_node(Node("2"))
     builder.add_node(Node("3"))
@@ -1274,7 +1466,7 @@ async def test_cond_edge_after_send() -> None:
     builder.add_conditional_edges("2", route_to_three)
     graph = builder.compile()
 
-    assert await graph.ainvoke(["0"]) == ["0", "1", "2", "3"]
+    assert await graph.ainvoke(["0"]) == ["0", "1", "2", "2", "3"]
 
 
 async def test_invoke_checkpoint_aiosqlite(mocker: MockerFixture) -> None:
@@ -5119,10 +5311,10 @@ async def test_branch_then() -> None:
                 "timestamp": AnyStr(),
                 "step": 3,
                 "payload": {
-                    "id": "ceada3c5-5f25-59e4-9ea5-544599ce1d2f",
+                    "id": "9b590c54-15ef-54b1-83a7-140d27b0bc52",
                     "name": "finish",
                     "input": {"my_key": "value prepared slow", "market": "DE"},
-                    "triggers": ["branch:prepare:condition:then"],
+                    "triggers": ["branch:prepare:condition::then"],
                 },
             },
             {
@@ -5130,7 +5322,7 @@ async def test_branch_then() -> None:
                 "timestamp": AnyStr(),
                 "step": 3,
                 "payload": {
-                    "id": "ceada3c5-5f25-59e4-9ea5-544599ce1d2f",
+                    "id": "9b590c54-15ef-54b1-83a7-140d27b0bc52",
                     "name": "finish",
                     "result": [("my_key", " finished")],
                 },