langgraph: release 0.3.30 (#4267 )

langgraph: support streaming messages from Command.update (#4250 )
simplify
2026-08-19 22:25:44 +02:00 · 2025-04-14 17:05:10 -04:00 · 2025-04-14 12:25:20 -07:00 · 2025-04-14 14:53:08 -04:00 · 2025-04-14 11:47:22 -07:00 · 2025-04-14 14:12:40 -04:00
33 changed files with 717 additions and 376 deletions
@@ -1,4 +1,4 @@
-# How to Deploy to Cloud SaaS
+# How to Deploy to Cloud SaaS (Beta)

 Before deploying, review the [conceptual guide for the Cloud SaaS](../../concepts/langgraph_cloud.md) deployment option.

@@ -1,4 +1,4 @@
-# How to Deploy Self-Hosted Control Plane
+# How to Deploy Self-Hosted Control Plane (Beta)

 Before deploying, review the [conceptual guide for the Self-Hosted Control Plane](../../concepts/langgraph_self_hosted_control_plane.md) deployment option.

@@ -1,4 +1,4 @@
-# How to Deploy Self-Hosted Data Plane
+# How to Deploy Self-Hosted Data Plane (Beta)

 Before deploying, review the [conceptual guide for the Self-Hosted Data Plane](../../concepts/langgraph_self_hosted_data_plane.md) deployment option.

@@ -1,6 +1,6 @@
 # Environment Variables

-The LangGraph Cloud Server supports specific environment variables for configuring a deployment.
+The LangGraph Server supports specific environment variables for configuring a deployment.

 ## `BG_JOB_ISOLATED_LOOPS`

@@ -32,7 +32,7 @@ See <a href="https://docs.smith.langchain.com/how_to_guides/tracing/sample_trace

 ## `LANGGRAPH_AUTH_TYPE`

-Type of authentication for the LangGraph Cloud Server deployment. Valid values: `langsmith`, `noop`.
+Type of authentication for the LangGraph Server deployment. Valid values: `langsmith`, `noop`.

 For deployments to LangGraph Cloud, this environment variable is set automatically. For local development or deployments where authentication is handled externally (e.g. self-hosted), set this environment variable to `noop`.

@@ -44,19 +44,27 @@ Set this environment variable to have a BYOC deployment send traces to a self-ho

 `SELF_HOSTED_LANGSMITH_HOSTNAME` is the hostname of the self-hosted LangSmith instance. It must be accessible to the BYOC deployment. `LANGSMITH_API_KEY` is a LangSmith API generated from the self-hosted LangSmith instance.

+## `LANGSMITH_TRACING`
+
+!!! info "Only for Self-Hosted Data Plane, Self-Hosted Control Plane, and Standalone Container"
+    Disabling LangSmith tracing is only available for [Self-Hosted Data Plane](../../concepts/langgraph_self_hosted_data_plane.md), [Self-Hosted Control Plane](../../concepts/langgraph_self_hosted_control_plane.md), and [Standalone Container](../../concepts/langgraph_standalone_container.md) deployments.
+
+Set `LANGSMITH_TRACING` to `false` to disable tracing to LangSmith.
+
 ## `LOG_LEVEL`

 Configure [log level](https://docs.python.org/3/library/logging.html#logging-levels). Defaults to `INFO`.

 ## `N_JOBS_PER_WORKER`

-Number of jobs per worker for the LangGraph Cloud task queue. Defaults to `10`.
+Number of jobs per worker for the LangGraph Server task queue. Defaults to `10`.

 ## `POSTGRES_URI_CUSTOM`

-For [Bring Your Own Cloud (BYOC)](../../concepts/bring_your_own_cloud.md) deployments only.
+!!! info "Only for Self-Hosted Data Plane and Self-Hosted Control Plane"
+    Custom Postgres instances are only available for [Self-Hosted Data Plane](../../concepts/langgraph_self_hosted_data_plane.md) and [Self-Hosted Control Plane](../../concepts/langgraph_self_hosted_control_plane.md) deployments.

-Specify `POSTGRES_URI_CUSTOM` to use an externally managed Postgres instance. The value of `POSTGRES_URI_CUSTOM` must be a valid [Postgres connection URI](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING-URIS).
+Specify `POSTGRES_URI_CUSTOM` to use a custom Postgres instance. The value of `POSTGRES_URI_CUSTOM` must be a valid [Postgres connection URI](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING-URIS).

 Postgres:

@@ -73,11 +81,11 @@ Control Plane Functionality:

 Database Connectivity:

- The externally managed Postgres instance must be accessible by the LangGraph Server service in the ECS cluster. The BYOC user is responsible for ensuring connectivity.
- For example, if an AWS RDS Postgres instance is provisioned, it can be provisioned in the same VPC (`langgraph-cloud-vpc`) as the ECS cluster with the `langgraph-cloud-service-sg` security group to ensure connectivity.
+- The custom Postgres instance must be accessible by the LangGraph Server. The user is responsible for ensuring connectivity.

 ## `REDIS_URI_CUSTOM`

-For [Bring Your Own Cloud (BYOC)](../../concepts/bring_your_own_cloud.md) deployments only.
+!!! info "Only for Self-Hosted Data Plane and Self-Hosted Control Plane"
+    Custom Redis instances are only available for [Self-Hosted Data Plane](../../concepts/langgraph_self_hosted_data_plane.md) and [Self-Hosted Control Plane](../../concepts/langgraph_self_hosted_control_plane.md) deployments.

-Specify `REDIS_URI_CUSTOM` to use an externally managed Redis instance. The value of `REDIS_URI_CUSTOM` must be a valid [Redis connection URI](https://redis-py.readthedocs.io/en/stable/connections.html#redis.Redis.from_url).
+Specify `REDIS_URI_CUSTOM` to use a custom Redis instance. The value of `REDIS_URI_CUSTOM` must be a valid [Redis connection URI](https://redis-py.readthedocs.io/en/stable/connections.html#redis.Redis.from_url).
@@ -10,11 +10,11 @@

 There are 4 main options for deploying with the LangGraph Platform:

-1. **[Cloud SaaS](#cloud-saas)**: Available for **Plus** and **Enterprise** plans.
+1. **<a href="#cloud-saas">Cloud SaaS<sup>(Beta)</sup></a>**: Available for **Plus** and **Enterprise** plans.

-1. **[Self-Hosted Data Plane](#self-hosted-data-plane)**: Available for the **Enterprise** plan.
+1. **<a href="#self-hosted-data-plane">Self-Hosted Data Plane<sup>(Beta)</sup></a>**: Available for the **Enterprise** plan.

-1. **[Self-Hosted Control Plane](#self-hosted-control-plane)**: Available for the **Enterprise** plan.
+1. **<a href="#self-hosted-control-plane">Self-Hosted Control Plane<sup>(Beta)</sup></a>**: Available for the **Enterprise** plan.

 1. **[Standalone Container](#standalone-container)**: Available for all plans.

@@ -76,7 +76,7 @@ The LangGraph Platform comprises several components that work together to suppor

 ### Deployment Options

- [Cloud SaaS](../concepts/langgraph_cloud.md): Connect to your GitHub repositories and deploy LangGraph Servers to LangChain's cloud. We manage everything.
- [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md): Create deployments from the [Control Plane UI](../concepts/langgraph_control_plane.md#control-plane-ui) and deploy LangGraph Servers to your cloud. We manage the [control plane](../concepts/langgraph_control_plane.md), you manage the deployments.
- [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md#control-plane-ui): Create deployments from a self-hosted [Control Plane UI](../concepts/langgraph_control_plane.md) and deploy LangGraph Servers to your cloud. You manage everything.
+- <a href="./langgraph_cloud/">Cloud SaaS<sup>(Beta)</sup></a>: Connect to your GitHub repositories and deploy LangGraph Servers to LangChain's cloud. We manage everything.
+- <a href="./langgraph_self_hosted_data_plane/">Self-Hosted Data Plane<sup>(Beta)</sup></a>: Create deployments from the [Control Plane UI](../concepts/langgraph_control_plane.md#control-plane-ui) and deploy LangGraph Servers to your cloud. We manage the [control plane](../concepts/langgraph_control_plane.md), you manage the deployments.
+- <a href="./langgraph_self_hosted_control_plane/">Self-Hosted Control Plane<sup>(Beta)</sup></a>: Create deployments from a self-hosted [Control Plane UI](../concepts/langgraph_control_plane.md#control-plane-ui) and deploy LangGraph Servers to your cloud. You manage everything.
 - [Standalone Container](../concepts/langgraph_standalone_container.md): Deploy LangGraph Server Docker images however you like.
@@ -1,4 +1,4 @@
-# Cloud SaaS
+# Cloud SaaS (Beta)

 To deploy a [LangGraph Server](../concepts/langgraph_server.md), follow the how-to guide for [how to deploy to Cloud SaaS](../cloud/deployment/cloud.md).

@@ -58,7 +58,7 @@ In the future, the autoscaling implementation may evolve to accommodate other me
 ### Static IP Addresses

 !!! info "Only for Cloud SaaS"
-    Static IP addresses are only available for [Cloud SaaS](../concepts/langgraph_cloud.md).
+    Static IP addresses are only available for [Cloud SaaS](../concepts/langgraph_cloud.md) deployments.

 All traffic from deployments created after January 6th 2025 will come through a NAT gateway. This NAT gateway will have several static IP addresses depending on the data region. Refer to the table below for the list of static IP addresses:

@@ -72,3 +72,46 @@ All traffic from deployments created after January 6th 2025 will come through a
 | 34.169.88.30   | 34.91.238.184  |
 | 34.19.93.202   | 35.204.101.241 |
 | 34.19.34.50    | 35.204.48.32   |
+
+### Custom Postgres
+
+!!! info "Only for Self-Hosted Data Plane and Self-Hosted Control Plane"
+    Custom Postgres instances are only available for [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md) and [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md) deployments.
+
+A custom Postgres instance can be used instead of the [one automatically created by the control plane](./langgraph_control_plane.md#database-provisioning). Specify the [`POSTGRES_URI_CUSTOM`](../cloud/reference/env_var.md#postgres_uri_custom) environment variable to use a custom Postgres instance.
+
+Multiple deployments can share the same Postgres instance. For example, for `Deployment A`, `POSTGRES_URI_CUSTOM` can be set to `postgres://<user>:<password>@/<database_name_1>?host=<hostname_1>` and for `Deployment B`, `POSTGRES_URI_CUSTOM` can be set to `postgres://<user>:<password>@/<database_name_2>?host=<hostname_1>`. `<database_name_1>` and `database_name_2` are different databases within the same instance, but `<hostname_1>` is shared. **The same database cannot be used for separate deployments**.
+
+### Custom Redis
+
+!!! info "Only for Self-Hosted Data Plane and Self-Hosted Control Plane"
+    Custom Redis instances are only available for [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md) and [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md) deployments.
+
+A custom Redis instance can be used instead of the one automatically created by the control plane. Specify the [REDIS_URI_CUSTOM](../cloud/reference/env_var.md#redis_uri_custom) environment variable to use a custom Redis instance.
+
+
+Multiple deployments can share the same Redis instance. For example, for `Deployment A`, `REDIS_URI_CUSTOM` can be set to `redis://<hostname_1>:<port>/1` and for `Deployment B`, `REDIS_URI_CUSTOM` can be set to `redis://<hostname_1>:<port>/2`. `1` and `2` are different database numbers within the same instance, but `<hostname_1>` is shared. **The same database number cannot be used for separate deployments**.
+
+### LangSmith Tracing
+
+LangGraph Server is automatically configured to send traces to LangSmith. See the table below for details with respect to each deployment option.
+
+| Cloud SaaS | Self-Hosted Data Plane | Self-Hosted Control Plane | Standalone Container |
+|------------|------------------------|---------------------------|----------------------|
+| Required<br><br>Trace to LangSmith SaaS. | Optional<br><br>Disable tracing or trace to LangSmith SaaS. | Optional<br><br>Disable tracing or trace to Self-Hosted LangSmith. | Optional<br><br>Disable tracing, trace to LangSmith SaaS, or trace to Self-Hosted LangSmith. |
+
+### Telemetry
+
+LangGraph Server is automatically configured to report telemetry metadata for billing purposes. See the table below for details with respect to each deployment option.
+
+| Cloud SaaS | Self-Hosted Data Plane | Self-Hosted Control Plane | Standalone Container |
+|------------|------------------------|---------------------------|----------------------|
+| Telemetry sent to LangSmith SaaS. | Telemetry sent to LangSmith SaaS. | Self-reported usage (audit) for air-gapped license key.<br><br>Telemetry sent to LangSmith SaaS for LangGraph Platform License Key. | Self-reported usage (audit) for air-gapped license key.<br><br>Telemetry sent to LangSmith SaaS for LangGraph Platform License Key. |
+
+### Licensing
+
+LangGraph Server is automatically configured to perform license key validation. See the table below for details with respect to each deployment option.
+
+| Cloud SaaS | Self-Hosted Data Plane | Self-Hosted Control Plane | Standalone Container |
+|------------|------------------------|---------------------------|----------------------|
+| LangSmith API Key validated against LangSmith SaaS. | LangSmith API Key validated against LangSmith SaaS. | Air-gapped license key or LangGraph Platform License Key validated against LangSmith SaaS. | Air-gapped license key or LangGraph Platform License Key validated against LangSmith SaaS. |
@@ -5,6 +5,10 @@ search:

 # LangGraph Platform

+Watch this 4-minute overview of LangGraph Platform to see how it helps you build, deploy, and evaluate agentic applications.
+
+<iframe width="560" height="315" src="https://www.youtube.com/embed/pfAQxBS5z88?si=XGS6Chydn6lhSO1S" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" referrerpolicy="strict-origin-when-cross-origin" allowfullscreen></iframe>
+
 ## Overview

 LangGraph Platform is a commercial solution for deploying agentic applications to production, built on the open-source [LangGraph framework](./high_level.md).
@@ -1,4 +1,4 @@
-# Self-Hosted Control Plane
+# Self-Hosted Control Plane (Beta)

 To deploy a [LangGraph Server](../concepts/langgraph_server.md), follow the how-to guide for [how to deploy the Self-Hosted Control Plane](../cloud/deployment/self_hosted_control_plane.md).

@@ -1,4 +1,4 @@
-# Self-Hosted Data Plane
+# Self-Hosted Data Plane (Beta)

 To deploy a [LangGraph Server](../concepts/langgraph_server.md), follow the how-to guide for [how to deploy the Self-Hosted Data Plane](../cloud/deployment/self_hosted_data_plane.md).

@@ -4,7 +4,7 @@

 ## How we use Postgres

-Postgres is the persistence layer for all user and run data in LGP. This stores both checkpoints (see more info [here](./persistence.md)) as well as the server resources (threads, runs, assistants and crons).
+Postgres is the persistence layer for all user, run, and long-term memory data in LGP. This stores both checkpoints (see more info [here](./persistence.md)), server resources (threads, runs, assistants and crons), as well as items saved in the long-term memory store (see more info [here](./persistence.md#memory-store)).

 ## How we use Redis

@@ -17,9 +17,9 @@ Get started deploying your LangGraph applications locally or on the cloud with

 ## Deployment Options

- [Cloud SaaS](../concepts/langgraph_cloud.md): Connect to your GitHub repositories and deploy LangGraph Servers to LangChain's cloud. We manage everything.
- [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md): Create deployments from the [Control Plane UI](../concepts/langgraph_control_plane.md#control-plane-ui) and deploy LangGraph Servers to your cloud. We manage the [control plane](../concepts/langgraph_control_plane.md), you manage the deployments.
- [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md#control-plane-ui): Create deployments from a self-hosted [Control Plane UI](../concepts/langgraph_control_plane.md) and deploy LangGraph Servers to your cloud. You manage everything.
+- <a href="../../concepts/langgraph_cloud/">Cloud SaaS<sup>(Beta)</sup></a>: Connect to your GitHub repositories and deploy LangGraph Servers to LangChain's cloud. We manage everything.
+- <a href="../../concepts/langgraph_self_hosted_data_plane/">Self-Hosted Data Plane<sup>(Beta)</sup></a>: Create deployments from the [Control Plane UI](../concepts/langgraph_control_plane.md#control-plane-ui) and deploy LangGraph Servers to your cloud. We manage the [control plane](../concepts/langgraph_control_plane.md), you manage the deployments.
+- <a href="../../concepts/langgraph_self_hosted_control_plane/">Self-Hosted Control Plane<sup>(Beta)</sup></a>: Create deployments from a self-hosted [Control Plane UI](../concepts/langgraph_control_plane.md#control-plane-ui) and deploy LangGraph Servers to your cloud. You manage everything.
 - [Standalone Container](../concepts/langgraph_standalone_container.md): Deploy LangGraph Server Docker images however you like.

 A quick comparison...
@@ -213,8 +213,8 @@ nav:
                - how-tos/ttl/configure_ttl.md
            - Authentication & Access Control:
                - Authentication & Access Control: how-tos#authentication-access-control
-                - cloud/how-tos/auth/custom_auth_new.md
-                - cloud/how-tos/auth/openapi_security_new.md
+                - how-tos/auth/custom_auth.md
+                - how-tos/auth/openapi_security.md
            - Assistants:
                - Assistants: how-tos#assistants
                - cloud/how-tos/configuration_cloud.md
@@ -256,6 +256,11 @@ nav:
                - cloud/how-tos/webhooks.md
            - Cron Jobs:
                - cloud/how-tos/cron_jobs.md
+            - Modifying the API:
+                - Modifying the API: how-tos#modifying-the-api
+                - how-tos/http/custom_lifespan.md
+                - how-tos/http/custom_middleware.md
+                - how-tos/http/custom_routes.md
            - LangGraph Studio:
                - LangGraph Studio: how-tos#langgraph-studio
                - cloud/how-tos/test_deployment.md
@@ -265,6 +270,7 @@ nav:
                - cloud/how-tos/datasets_studio.md
                - cloud/how-tos/iterate_graph_studio.md
                - cloud/how-tos/clone_traces_studio.md
+                - how-tos/local-studio.md
      - Concepts:
        - concepts/index.md
        - LangGraph:
@@ -273,8 +279,9 @@ nav:
          - concepts/low_level.md
          - concepts/agentic_concepts.md
          - concepts/multi_agent.md
-          - concepts/breakpoints
+          - concepts/breakpoints.md
          - concepts/human_in_the_loop.md
+          - concepts/v0-human-in-the-loop.md
          - concepts/time-travel.md
          - concepts/persistence.md
          - concepts/memory.md
@@ -287,7 +294,10 @@ nav:
          - High Level:
            - High Level: concepts#high-level
            - concepts/langgraph_platform.md
+            - concepts/platform_architecture.md
+            - concepts/scalability_and_resilience.md
            - concepts/deployment_options.md
+            - concepts/bring_your_own_cloud.md
            - concepts/plans.md
            - concepts/template_applications.md
          - Components:
@@ -311,6 +321,7 @@ nav:
            - concepts/langgraph_self_hosted_data_plane.md
            - concepts/langgraph_self_hosted_control_plane.md
            - concepts/langgraph_standalone_container.md
+            - concepts/self_hosted.md
      - Tutorials:
        - tutorials/index.md
        - Quick Start:
@@ -387,6 +398,7 @@ nav:
      - LangGraph Academy Course: https://academy.langchain.com/courses/intro-to-langgraph

  - API reference:
+    - reference/index.md
    - Library:
      - Graphs: reference/graphs.md
      - Checkpointing: reference/checkpoints.md
@@ -497,13 +509,8 @@ extra:
            Thanks for your feedback! Please help us improve this page by adding to the discussion below.
 validation:
  # https://www.mkdocs.org/user-guide/configuration/
-  # We're `ignoring` nav.omitted_files because we are going to rely
-  # on files being properly links to from the index pages of:
-  # - tutorials
-  # - concepts
-  # - how-tos
-  # - reference
-  omitted_files: ignore
+  # We are still raising for omitted files because they determine the breadcrumbs for pages.
+  omitted_files: warn 
  absolute_links: warn
  unrecognized_links: warn
  # TODO: figure out how to enable 'warn' for this
@@ -106,6 +106,7 @@ def fanout_to_subgraph_sync() -> StateGraph:
 if __name__ == "__main__":
    import asyncio
    import random
+    import time

    import uvloop

@@ -123,4 +124,7 @@ if __name__ == "__main__":
        len([c async for c in graph.astream(input, config=config)])

    uvloop.install()
+    start = time.time()
    asyncio.run(run())
+    end = time.time()
+    print(f"Time taken: {end - start:.4f} seconds")
@@ -20,7 +20,7 @@ from typing import (
 from langgraph.channels.ephemeral_value import EphemeralValue
 from langgraph.channels.last_value import LastValue
 from langgraph.checkpoint.base import BaseCheckpointSaver
-from langgraph.constants import END, PREVIOUS, START, TAG_HIDDEN
+from langgraph.constants import END, PREVIOUS, START
 from langgraph.pregel import Pregel
 from langgraph.pregel.call import (
    P,
@@ -434,8 +434,7 @@ class entrypoint:
                            [
                                ChannelWriteEntry(END, mapper=_pluck_return_value),
                                ChannelWriteEntry(PREVIOUS, mapper=_pluck_save_value),
-                            ],
-                            tags=[TAG_HIDDEN],
+                            ]
                        )
                    ],
                )
@@ -1,4 +1,3 @@
-import asyncio
 from inspect import (
    isfunction,
    ismethod,
@@ -178,7 +177,7 @@ class Branch(NamedTuple):
        ],
    ) -> Runnable:
        if reader:
-            value = await asyncio.to_thread(reader, config)
+            value = reader(config)
            # passthrough additional keys from node to branch
            # only doable when using dict states
            if (
@@ -366,16 +366,14 @@ class CompiledGraph(Pregel):
        self.nodes[key] = (
            PregelNode(channels=[], triggers=[], metadata=node.metadata)
            | node.runnable
-            | ChannelWrite([ChannelWriteEntry(key)], tags=[TAG_HIDDEN])
+            | ChannelWrite([ChannelWriteEntry(key)])
        )
        cast(list[str], self.stream_channels).append(key)

    def attach_edge(self, start: str, end: str) -> None:
        if end == END:
            # publish to end channel
-            self.nodes[start].writers.append(
-                ChannelWrite([ChannelWriteEntry(END)], tags=[TAG_HIDDEN])
-            )
+            self.nodes[start].writers.append(ChannelWrite([ChannelWriteEntry(END)]))
        else:
            # subscribe to start channel
            self.nodes[end].triggers.append(start)
@@ -393,10 +391,7 @@ class CompiledGraph(Pregel):
                )
                for p in packets
            ]
-            return ChannelWrite(
-                cast(Sequence[Union[ChannelWriteEntry, Send]], writes),
-                tags=[TAG_HIDDEN],
-            )
+            return ChannelWrite(cast(Sequence[Union[ChannelWriteEntry, Send]], writes))

        # add hidden start node
        if start == START and start not in self.nodes:
@@ -43,8 +43,8 @@ from langgraph.constants import (
    MISSING,
    NS_END,
    NS_SEP,
-    SELF,
    TAG_HIDDEN,
+    TASKS,
 )
 from langgraph.errors import (
    ErrorCode,
@@ -77,9 +77,9 @@ from langgraph.pregel.write import (
 )
 from langgraph.store.base import BaseStore
 from langgraph.types import All, Checkpointer, Command, RetryPolicy
-from langgraph.utils.fields import get_field_default
+from langgraph.utils.fields import get_field_default, get_update_as_tuples
 from langgraph.utils.pydantic import create_model
-from langgraph.utils.runnable import RunnableCallable, RunnableLike, coerce_to_runnable
+from langgraph.utils.runnable import RunnableLike, coerce_to_runnable

 logger = logging.getLogger(__name__)

@@ -639,6 +639,7 @@ class StateGraph(Graph):

        compiled = CompiledStateGraph(
            builder=self,
+            schema_to_mapper={},
            config_type=self.config_schema,
            input_model=(
                self.input
@@ -670,10 +671,6 @@ class StateGraph(Graph):
        for key, node in self.nodes.items():
            compiled.attach_node(key, node)

-        compiled.attach_branch(START, SELF, CONTROL_BRANCH, with_reader=False)
-        for key, node in self.nodes.items():
-            compiled.attach_branch(key, SELF, CONTROL_BRANCH, with_reader=False)
-
        for start, end in self.edges:
            compiled.attach_edge(start, end)

@@ -689,6 +686,16 @@ class StateGraph(Graph):

 class CompiledStateGraph(CompiledGraph):
    builder: StateGraph
+    schema_to_mapper: dict[Type[Any], Optional[Callable[[Any], Any]]]
+
+    def __init__(
+        self,
+        *,
+        schema_to_mapper: dict[Type[Any], Optional[Callable[[Any], Any]]],
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.schema_to_mapper = schema_to_mapper

    def get_input_schema(
        self, config: Optional[RunnableConfig] = None
@@ -724,28 +731,6 @@ class CompiledStateGraph(CompiledGraph):
                if is_writable_managed_value(v)
            ]

-        def _get_root(input: Any) -> Optional[Sequence[tuple[str, Any]]]:
-            if isinstance(input, Command):
-                if input.graph == Command.PARENT:
-                    return ()
-                return input._update_as_tuples()
-            elif (
-                isinstance(input, (list, tuple))
-                and input
-                and any(isinstance(i, Command) for i in input)
-            ):
-                updates: list[tuple[str, Any]] = []
-                for i in input:
-                    if isinstance(i, Command):
-                        if i.graph == Command.PARENT:
-                            continue
-                        updates.extend(i._update_as_tuples())
-                    else:
-                        updates.append(("__root__", i))
-                return updates
-            elif input is not None:
-                return [("__root__", input)]
-
        def _get_updates(
            input: Union[None, dict, Any],
        ) -> Optional[Sequence[tuple[str, Any]]]:
@@ -776,32 +761,7 @@ class CompiledStateGraph(CompiledGraph):
                        updates.extend(_get_updates(i) or ())
                return updates
            elif (t := type(input)) and get_type_hints(t):
-                # Pydantic v2
-                if isinstance(input, BaseModelV1):
-                    keep: Optional[set[str]] = input.__fields_set__
-                    defaults = {k: v.default for k, v in t.__fields__.items()}
-                elif isinstance(input, BaseModel):
-                    keep = input.model_fields_set
-                    defaults = {k: v.default for k, v in input.model_fields.items()}
-                # Pydantic v1
-                else:
-                    keep = None
-                    defaults = {}
-
-                # NOTE: This behavior for Pydantic is somewhat inelegant,
-                # but we keep around for backwards compatibility
-                # if input is a Pydantic model, only update values
-                # that are different from the default values or in the keep set
-                return [
-                    (k, value)
-                    for k in output_keys
-                    if (value := getattr(input, k, MISSING)) is not MISSING
-                    and (
-                        value is not None
-                        or defaults.get(k, MISSING) is not None
-                        or (keep is not None and k in keep)
-                    )
-                ]
+                return get_update_as_tuples(input, output_keys)
            else:
                msg = create_error_message(
                    message=f"Expected dict, got {input}",
@@ -814,6 +774,7 @@ class CompiledStateGraph(CompiledGraph):
            ChannelWriteTupleEntry(
                mapper=_get_root if output_keys == ["__root__"] else _get_updates
            ),
+            ChannelWriteTupleEntry(mapper=_control_branch),
        )

        # add node and output channel
@@ -822,12 +783,21 @@ class CompiledStateGraph(CompiledGraph):
                tags=[TAG_HIDDEN],
                triggers=[START],
                channels=[START],
-                writers=[ChannelWrite(write_entries, tags=[TAG_HIDDEN])],
+                writers=[ChannelWrite(write_entries)],
            )
        elif node is not None:
            input_schema = node.input if node else self.builder.schema
            input_values = {k: k for k in self.builder.schemas[input_schema]}
            is_single_input = len(input_values) == 1 and "__root__" in input_values
+            if input_schema in self.schema_to_mapper:
+                mapper = self.schema_to_mapper[input_schema]
+            else:
+                mapper = _pick_mapper(
+                    list(input_values),
+                    input_schema,
+                    self.builder.type_hints[input_schema],
+                )
+                self.schema_to_mapper[input_schema] = mapper

            branch_channel = CHANNEL_BRANCH_TO.format(key)
            self.channels[branch_channel] = EphemeralValue(Any, guard=False)
@@ -836,13 +806,9 @@ class CompiledStateGraph(CompiledGraph):
                # read state keys and managed values
                channels=(list(input_values) if is_single_input else input_values),
                # coerce state dict to schema class (eg. pydantic model)
-                mapper=_pick_mapper(
-                    list(input_values),
-                    input_schema,
-                    self.builder.type_hints[input_schema],
-                ),
+                mapper=mapper,
                # publish to state keys
-                writers=[ChannelWrite(write_entries, tags=[TAG_HIDDEN])],
+                writers=[ChannelWrite(write_entries)],
                metadata=node.metadata,
                retry_policy=node.retry_policy,
                bound=node.runnable,
@@ -868,9 +834,7 @@ class CompiledStateGraph(CompiledGraph):
            # publish to channel
            for start in starts:
                self.nodes[start].writers.append(
-                    ChannelWrite(
-                        (ChannelWriteEntry(channel_name, start),), tags=[TAG_HIDDEN]
-                    )
+                    ChannelWrite((ChannelWriteEntry(channel_name, start),))
                )

    def attach_branch(
@@ -901,19 +865,33 @@ class CompiledStateGraph(CompiledGraph):
                    config, cast(Sequence[Union[Send, ChannelWriteEntry]], writes)
                )

-        schema = branch.input_schema or (
-            self.builder.nodes[start].input
-            if start in self.builder.nodes
-            else self.builder.schema
-        )
+        if with_reader:
+            # get schema
+            schema = branch.input_schema or (
+                self.builder.nodes[start].input
+                if start in self.builder.nodes
+                else self.builder.schema
+            )
+            channels = list(self.builder.schemas[schema])
+            # get mapper
+            if schema in self.schema_to_mapper:
+                mapper = self.schema_to_mapper[schema]
+            else:
+                mapper = _pick_mapper(channels, schema, self.builder.type_hints[schema])
+                self.schema_to_mapper[schema] = mapper
+            # create reader
+            reader: Optional[Callable[[RunnableConfig], Any]] = partial(
+                ChannelRead.do_read,
+                select=channels[0] if channels == ["__root__"] else channels,
+                fresh=True,
+                # coerce state dict to schema class (eg. pydantic model)
+                mapper=mapper,
+            )
+        else:
+            reader = None

        # attach branch publisher
-        self.nodes[start].writers.append(
-            branch.run(
-                branch_writer,
-                _get_state_reader(self.builder, schema) if with_reader else None,
-            )
-        )
+        self.nodes[start].writers.append(branch.run(branch_writer, reader))

        # attach then subscriber
        if branch.then and branch.then != END:
@@ -928,9 +906,7 @@ class CompiledStateGraph(CompiledGraph):
            for end in ends:
                if end != END:
                    self.nodes[end].writers.append(
-                        ChannelWrite(
-                            [ChannelWriteEntry(channel_name, end)], tags=[TAG_HIDDEN]
-                        )
+                        ChannelWrite((ChannelWriteEntry(channel_name, end),))
                    )

    def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
@@ -1038,20 +1014,6 @@ class CompiledStateGraph(CompiledGraph):
                        seen[INTERRUPT].pop(k, MISSING)


-def _get_state_reader(
-    builder: StateGraph, schema: Type[Any]
-) -> Callable[[RunnableConfig], Any]:
-    state_keys = list(builder.channels)
-    select = list(builder.schemas[schema])
-    return partial(
-        ChannelRead.do_read,
-        select=select[0] if select == ["__root__"] else select,
-        fresh=True,
-        # coerce state dict to schema class (eg. pydantic model)
-        mapper=_pick_mapper(state_keys, schema, builder.type_hints[schema]),
-    )
-
-
 def _pick_mapper(
    state_keys: Sequence[str], schema: Type[Any], type_hints: Optional[dict[str, Any]]
 ) -> Optional[Callable[[Any], Any]]:
@@ -1069,9 +1031,9 @@ def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]:
    return schema(**input)


-def _control_branch(value: Any) -> Sequence[Union[str, Send]]:
+def _control_branch(value: Any) -> Sequence[tuple[str, Any]]:
    if isinstance(value, Send):
-        return [value]
+        return ((TASKS, value),)
    commands: list[Command] = []
    if isinstance(value, Command):
        commands.append(value)
@@ -1079,51 +1041,45 @@ def _control_branch(value: Any) -> Sequence[Union[str, Send]]:
        for cmd in value:
            if isinstance(cmd, Command):
                commands.append(cmd)
-    rtn: list[Union[str, Send]] = []
+    rtn: list[tuple[str, Any]] = []
    for command in commands:
        if command.graph == Command.PARENT:
            raise ParentCommand(command)
        if isinstance(command.goto, Send):
-            rtn.append(command.goto)
+            rtn.append((TASKS, command.goto))
        elif isinstance(command.goto, str):
-            rtn.append(command.goto)
+            rtn.append((CHANNEL_BRANCH_TO.format(command.goto), None))
        else:
-            rtn.extend(command.goto)
+            rtn.extend(
+                (TASKS, go)
+                if isinstance(go, Send)
+                else (CHANNEL_BRANCH_TO.format(go), None)
+                for go in command.goto
+            )
    return rtn


-async def _acontrol_branch(value: Any) -> Sequence[Union[str, Send]]:
-    if isinstance(value, Send):
-        return [value]
-    commands: list[Command] = []
-    if isinstance(value, Command):
-        commands.append(value)
-    elif isinstance(value, (list, tuple)):
-        for cmd in value:
-            if isinstance(cmd, Command):
-                commands.append(cmd)
-    rtn: list[Union[str, Send]] = []
-    for command in commands:
-        if command.graph == Command.PARENT:
-            raise ParentCommand(command)
-        if isinstance(command.goto, Send):
-            rtn.append(command.goto)
-        elif isinstance(command.goto, str):
-            rtn.append(command.goto)
-        else:
-            rtn.extend(command.goto)
-    return rtn
-
-
-CONTROL_BRANCH_PATH = RunnableCallable(
-    _control_branch,
-    _acontrol_branch,
-    tags=[TAG_HIDDEN],
-    trace=False,
-    recurse=False,
-    func_accepts_config=False,
-)
-CONTROL_BRANCH = Branch(CONTROL_BRANCH_PATH, None)
+def _get_root(input: Any) -> Optional[Sequence[tuple[str, Any]]]:
+    if isinstance(input, Command):
+        if input.graph == Command.PARENT:
+            return ()
+        return input._update_as_tuples()
+    elif (
+        isinstance(input, (list, tuple))
+        and input
+        and any(isinstance(i, Command) for i in input)
+    ):
+        updates: list[tuple[str, Any]] = []
+        for i in input:
+            if isinstance(i, Command):
+                if i.graph == Command.PARENT:
+                    continue
+                updates.extend(i._update_as_tuples())
+            else:
+                updates.append(("__root__", i))
+        return updates
+    elif input is not None:
+        return [("__root__", input)]


 def _get_channels(
@@ -2550,11 +2550,12 @@ class Pregel(PregelProtocol):
        do_stream = (
            next(
                (
-                    cast(_StreamingCallbackHandler, h)
+                    True
                    for h in run_manager.handlers
                    if isinstance(h, _StreamingCallbackHandler)
+                    and not isinstance(h, StreamMessagesHandler)
                ),
-                None,
+                False,
            )
            if _StreamingCallbackHandler is not None
            else False
@@ -2624,7 +2625,7 @@ class Pregel(PregelProtocol):
                    ),
                    put_writes=weakref.WeakMethod(loop.put_writes),
                    schedule_task=weakref.WeakMethod(loop.accept_push),
-                    use_astream=do_stream is not None,
+                    use_astream=do_stream,
                    node_finished=config[CONF].get(CONFIG_KEY_NODE_FINISHED),
                )
                # enable subgraph streaming
@@ -3,13 +3,13 @@ import itertools
 import sys
 import threading
 from collections import defaultdict, deque
+from copy import copy
 from functools import partial
 from hashlib import sha1
 from typing import (
    Any,
    Callable,
    Iterable,
-    Iterator,
    Literal,
    Mapping,
    NamedTuple,
@@ -49,6 +49,7 @@ from langgraph.constants import (
    EMPTY_SEQ,
    ERROR,
    INTERRUPT,
+    MISSING,
    NO_WRITES,
    NS_END,
    NS_SEP,
@@ -63,12 +64,12 @@ from langgraph.constants import (
    TASKS,
    Send,
 )
-from langgraph.errors import EmptyChannelError, InvalidUpdateError
+from langgraph.errors import InvalidUpdateError
 from langgraph.managed.base import ManagedValueMapping
 from langgraph.pregel.call import get_runnable_for_task
-from langgraph.pregel.io import read_channel, read_channels
+from langgraph.pregel.io import read_channels
 from langgraph.pregel.log import logger
-from langgraph.pregel.read import PregelNode
+from langgraph.pregel.read import INPUT_CACHE_KEY_TYPE, PregelNode
 from langgraph.store.base import BaseStore
 from langgraph.types import (
    All,
@@ -423,6 +424,7 @@ def prepare_next_tasks(
        are the tasks themselves. This is the union of all PUSH tasks (Sends)
        and PULL tasks (nodes triggered by edges).
    """
+    input_cache: dict[INPUT_CACHE_KEY_TYPE, Any] = {}
    checkpoint_id_bytes = binascii.unhexlify(checkpoint["id"].replace("-", ""))
    null_version = checkpoint_null_version(checkpoint)
    tasks: list[Union[PregelTask, PregelExecutableTask]] = []
@@ -444,6 +446,7 @@ def prepare_next_tasks(
            store=store,
            checkpointer=checkpointer,
            manager=manager,
+            input_cache=input_cache,
        ):
            tasks.append(task)

@@ -486,6 +489,7 @@ def prepare_next_tasks(
            store=store,
            checkpointer=checkpointer,
            manager=manager,
+            input_cache=input_cache,
        ):
            tasks.append(task)
    return {t.id: t for t in tasks}
@@ -511,6 +515,7 @@ def prepare_single_task(
    store: Optional[BaseStore] = None,
    checkpointer: Optional[BaseCheckpointSaver] = None,
    manager: Union[None, ParentRunManager, AsyncParentRunManager] = None,
+    input_cache: Optional[dict[INPUT_CACHE_KEY_TYPE, Any]] = None,
 ) -> Union[None, PregelTask, PregelExecutableTask]:
    """Prepares a single task for the next Pregel step, given a task path, which
    uniquely identifies a PUSH or PULL task within the graph."""
@@ -729,11 +734,15 @@ def prepare_single_task(
        ):
            triggers = tuple(sorted(proc.triggers))
            try:
-                val = next(
-                    _proc_input(proc, managed, channels, for_execution=for_execution)
+                val = _proc_input(
+                    proc,
+                    managed,
+                    channels,
+                    for_execution=for_execution,
+                    input_cache=input_cache,
                )
-            except StopIteration:
-                return
+                if val is MISSING:
+                    return
            except Exception as exc:
                if SUPPORTS_EXC_NOTES:
                    exc.add_note(
@@ -926,34 +935,32 @@ def _proc_input(
    channels: Mapping[str, BaseChannel],
    *,
    for_execution: bool,
-) -> Iterator[Any]:
+    input_cache: Optional[dict[INPUT_CACHE_KEY_TYPE, Any]],
+) -> Any:
    """Prepare input for a PULL task, based on the process's channels and triggers."""
+    # if in cache return shallow copy
+    if input_cache is not None and proc.input_cache_key in input_cache:
+        return copy(input_cache[proc.input_cache_key])
    # If all trigger channels subscribed by this process are not empty
    # then invoke the process with the values of all non-empty channels
    if isinstance(proc.channels, dict):
-        try:
-            val: dict[str, Any] = {}
-            for k, chan in proc.channels.items():
-                if chan in proc.triggers:
-                    val[k] = read_channel(channels, chan, catch=False)
-                elif chan in channels:
-                    try:
-                        val[k] = read_channel(channels, chan, catch=False)
-                    except EmptyChannelError:
-                        continue
-                else:
-                    val[k] = managed[k]()
-        except EmptyChannelError:
-            return
+        val: dict[str, Any] = {}
+        for k, chan in proc.channels.items():
+            if chan in channels:
+                if channels[chan].is_available():
+                    val[k] = channels[chan].get()
+            else:
+                val[k] = managed[k]()
    elif isinstance(proc.channels, list):
        for chan in proc.channels:
-            try:
-                val = read_channel(channels, chan, catch=False)
-                break
-            except EmptyChannelError:
-                pass
+            if chan in channels:
+                if channels[chan].is_available():
+                    val = channels[chan].get()
+                    break
+            else:
+                val[k] = managed[k]()
        else:
-            return
+            return MISSING
    else:
        raise RuntimeError(
            "Invalid channels type, expected list or dict, got {proc.channels}"
@@ -963,7 +970,11 @@ def _proc_input(
    if for_execution and proc.mapper is not None:
        val = proc.mapper(val)

-    yield val
+    # Cache the input value
+    if input_cache is not None:
+        input_cache[proc.input_cache_key] = val
+
+    return val


 def _uuid5_str(namespace: bytes, *parts: str) -> str:
@@ -10,7 +10,7 @@ from typing import Any, Callable, Generator, Generic, Optional, Sequence, TypeVa
 from langchain_core.runnables import Runnable
 from typing_extensions import ParamSpec

-from langgraph.constants import CONF, CONFIG_KEY_CALL, RETURN, TAG_HIDDEN
+from langgraph.constants import CONF, CONFIG_KEY_CALL, RETURN
 from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
 from langgraph.types import RetryPolicy
 from langgraph.utils.config import get_config
@@ -197,7 +197,7 @@ def get_runnable_for_task(func: Callable[..., Any]) -> RunnableSeq:
            )
        seq = RunnableSeq(
            run,
-            ChannelWrite([ChannelWriteEntry(RETURN)], tags=[TAG_HIDDEN]),
+            ChannelWrite([ChannelWriteEntry(RETURN)]),
            name=name,
            trace_inputs=functools.partial(
                _explode_args_trace_inputs, inspect.signature(func)
@@ -18,7 +18,7 @@ from langchain_core.messages import BaseMessage
 from langchain_core.outputs import ChatGenerationChunk, LLMResult

 from langgraph.constants import NS_SEP, TAG_HIDDEN, TAG_NOSTREAM
-from langgraph.types import StreamChunk
+from langgraph.types import Command, StreamChunk

 try:
    from langchain_core.tracers._streaming import _StreamingCallbackHandler
@@ -153,6 +153,17 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
        **kwargs: Any,
    ) -> Any:
        if meta := self.metadata.pop(run_id, None):
+            if isinstance(response, Command):
+                response = response.update
+
+            if isinstance(response, Sequence) and any(
+                isinstance(value, Command) for value in response
+            ):
+                response = [
+                    value.update if isinstance(value, Command) else value
+                    for value in response
+                ]
+
            if isinstance(response, BaseMessage):
                self._emit(meta, response, dedupe=True)
            elif isinstance(response, Sequence):
@@ -30,6 +30,7 @@ from langgraph.utils.config import merge_configs
 from langgraph.utils.runnable import RunnableCallable, RunnableSeq

 READ_TYPE = Callable[[Union[str, Sequence[str]], bool], Union[Any, dict[str, Any]]]
+INPUT_CACHE_KEY_TYPE = tuple[Callable[..., Any], tuple[str, ...]]


 class ChannelRead(RunnableCallable):
@@ -67,6 +68,7 @@ class ChannelRead(RunnableCallable):
            afunc=self._aread,
            tags=tags,
            name=None,
+            trace=False,
            func_accepts_config=True,
        )
        self.fresh = fresh
@@ -231,6 +233,17 @@ class PregelNode(Runnable):
        else:
            return self.bound

+    @cached_property
+    def input_cache_key(self) -> INPUT_CACHE_KEY_TYPE:
+        """Get a cache key for the input to the node.
+        This is used to avoid calculating the same input multiple times."""
+        return (
+            self.mapper,
+            tuple(f"{key}:{value}" for key, value in self.channels.items())
+            if isinstance(self.channels, dict)
+            else tuple(self.channels),
+        )
+
    def join(self, channels: Sequence[str]) -> PregelNode:
        assert isinstance(channels, list) or isinstance(
            channels, tuple
@@ -54,14 +54,14 @@ class ChannelWrite(RunnableCallable):
        self,
        writes: Sequence[Union[ChannelWriteEntry, ChannelWriteTupleEntry, Send]],
        *,
-        tags: Optional[Sequence[str]] = None,
+        tags: Optional[Sequence[str]] = None,  # ignored
        require_at_least_one_of: Optional[Sequence[str]] = None,  # ignored
    ):
        super().__init__(
            func=self._write,
            afunc=self._awrite,
            name=None,
-            tags=tags,
+            trace=False,
            func_accepts_config=True,
        )
        self.writes = cast(
@@ -152,6 +152,8 @@ class ChannelWrite(RunnableCallable):
                tuples.append((w.channel, value))
            else:
                raise ValueError(f"Invalid write entry: {w}")
+        # if we want to persist writes found before hitting a ParentCommand
+        # can move this to a finally block
        write: TYPE_SEND = config[CONF][CONFIG_KEY_SEND]
        write(tuples)

@@ -23,6 +23,7 @@ from langchain_core.runnables import Runnable, RunnableConfig
 from typing_extensions import Self

 from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
+from langgraph.utils.fields import get_update_as_tuples

 if TYPE_CHECKING:
    from langgraph.pregel.protocol import PregelProtocol
@@ -318,7 +319,7 @@ class Command(Generic[N], ToolOutputMixin):
        ):
            return self.update
        elif hints := get_type_hints(type(self.update)):
-            return [(k, getattr(self.update, k)) for k in hints]
+            return get_update_as_tuples(self.update, tuple(hints.keys()))
        elif self.update is not None:
            return [("__root__", self.update)]
        else:
@@ -1,8 +1,14 @@
 import dataclasses
-from typing import Any, Generator, Optional, Type, Union, get_type_hints
+from typing import Any, Generator, Optional, Sequence, Type, Union, get_type_hints

+from pydantic import BaseModel
+from pydantic.v1 import BaseModel as BaseModelV1
 from typing_extensions import Annotated, NotRequired, ReadOnly, Required, get_origin

+# NOTE: this is redefined here separately from langgraph.constants
+# to avoid a circular import
+MISSING = object()
+

 def _is_optional_type(type_: Any) -> bool:
    """Check if a type is Optional."""
@@ -147,3 +153,33 @@ def get_enhanced_type_hints(
            pass

        yield name, typ, default, description
+
+
+def get_update_as_tuples(input: Any, keys: Sequence[str]) -> list[tuple[str, Any]]:
+    """Get Pydantic state update as a list of (key, value) tuples."""
+    # Pydantic v1
+    if isinstance(input, BaseModelV1):
+        keep: Optional[set[str]] = input.__fields_set__
+        defaults = {k: v.default for k, v in input.__fields__.items()}
+    # Pydantic v2
+    elif isinstance(input, BaseModel):
+        keep = input.model_fields_set
+        defaults = {k: v.default for k, v in input.model_fields.items()}
+    else:
+        keep = None
+        defaults = {}
+
+    # NOTE: This behavior for Pydantic is somewhat inelegant,
+    # but we keep around for backwards compatibility
+    # if input is a Pydantic model, only update values
+    # that are different from the default values or in the keep set
+    return [
+        (k, value)
+        for k in keys
+        if (value := getattr(input, k, MISSING)) is not MISSING
+        and (
+            value is not None
+            or defaults.get(k, MISSING) is not None
+            or (keep is not None and k in keep)
+        )
+    ]
@@ -36,6 +36,7 @@ from langchain_core.runnables.config import (
    var_child_runnable_config,
 )
 from langchain_core.runnables.utils import Input, Output
+from langchain_core.tracers.langchain import LangChainTracer
 from typing_extensions import TypeGuard

 from langgraph.constants import (
@@ -60,58 +61,34 @@ except ImportError:


 def _set_config_context(
-    config: RunnableConfig,
-) -> tuple[Token[Optional[RunnableConfig]], Optional[dict[str, Any]]]:
+    config: RunnableConfig, run: Any = None
+) -> Token[Optional[RunnableConfig]]:
    """Set the child Runnable config + tracing context.

    Args:
        config (RunnableConfig): The config to set.
    """
-    from langchain_core.tracers.langchain import LangChainTracer
-
    config_token = var_child_runnable_config.set(config)
-    current_context = None
-    if (
-        (callbacks := config.get("callbacks"))
-        and (
-            parent_run_id := getattr(callbacks, "parent_run_id", None)
-        )  # Is callback manager
-        and (
-            tracer := next(
-                (
-                    handler
-                    for handler in getattr(callbacks, "handlers", [])
-                    if isinstance(handler, LangChainTracer)
-                ),
-                None,
-            )
-        )
-        and (run := tracer.run_map.get(str(parent_run_id)))
-    ):
-        from langsmith.run_helpers import _set_tracing_context, get_tracing_context
+    if run is not None:
+        from langsmith.run_helpers import _set_tracing_context

-        current_context = get_tracing_context()
        _set_tracing_context({"parent": run})
-    return config_token, current_context
+    return config_token


-@contextmanager
-def set_config_context(config: RunnableConfig) -> Generator[Context, None, None]:
+def _unset_config_context(
+    token: Token[Optional[RunnableConfig]], run: Any = None
+) -> None:
    """Set the child Runnable config + tracing context.

    Args:
        config (RunnableConfig): The config to set.
    """
-    from langsmith.run_helpers import _set_tracing_context
+    var_child_runnable_config.reset(token)
+    if run is not None:
+        from langsmith.run_helpers import _set_tracing_context

-    ctx = copy_context()
-    config_token, _ = ctx.run(_set_config_context, config)
-    try:
-        yield ctx
-    finally:
-        ctx.run(var_child_runnable_config.reset, config_token)
-        ctx.run(
-            _set_tracing_context,
+        _set_tracing_context(
            {
                "parent": None,
                "project_name": None,
@@ -119,10 +96,27 @@ def set_config_context(config: RunnableConfig) -> Generator[Context, None, None]
                "metadata": None,
                "enabled": None,
                "client": None,
-            },
+            }
        )


+@contextmanager
+def set_config_context(
+    config: RunnableConfig, run: Any = None
+) -> Generator[Context, None, None]:
+    """Set the child Runnable config + tracing context.
+
+    Args:
+        config (RunnableConfig): The config to set.
+    """
+    ctx = copy_context()
+    config_token = ctx.run(_set_config_context, config, run)
+    try:
+        yield ctx
+    finally:
+        ctx.run(_unset_config_context, config_token, run)
+
+
 # Before Python 3.11 native StrEnum is not available
 class StrEnum(str, enum.Enum):
    """A string enum."""
@@ -363,7 +357,15 @@ class RunnableCallable(Runnable):
            )
            try:
                child_config = patch_config(config, callbacks=run_manager.get_child())
-                with set_config_context(child_config) as context:
+                # get the run
+                for h in run_manager.handlers:
+                    if isinstance(h, LangChainTracer):
+                        run = h.run_map.get(str(run_manager.run_id))
+                        break
+                else:
+                    run = None
+                # run in context
+                with set_config_context(child_config, run) as context:
                    ret = context.run(self.func, *args, **kwargs)
            except BaseException as e:
                run_manager.on_chain_error(e)
@@ -371,9 +373,8 @@ class RunnableCallable(Runnable):
            else:
                run_manager.on_chain_end(ret)
        else:
-            with set_config_context(config) as context:
-                ret = context.run(self.func, *args, **kwargs)
-        if isinstance(ret, Runnable) and self.recurse:
+            ret = self.func(*args, **kwargs)
+        if self.recurse and isinstance(ret, Runnable):
            return ret.invoke(input, config)
        return ret

@@ -417,25 +418,26 @@ class RunnableCallable(Runnable):
            )
            try:
                child_config = patch_config(config, callbacks=run_manager.get_child())
-                with set_config_context(child_config) as context:
-                    coro = cast(Coroutine[None, None, Any], self.afunc(*args, **kwargs))
-                    if ASYNCIO_ACCEPTS_CONTEXT:
-                        ret = await asyncio.create_task(coro, context=context)
+                coro = cast(Coroutine[None, None, Any], self.afunc(*args, **kwargs))
+                if ASYNCIO_ACCEPTS_CONTEXT:
+                    for h in run_manager.handlers:
+                        if isinstance(h, LangChainTracer):
+                            run = h.run_map.get(str(run_manager.run_id))
+                            break
                    else:
-                        ret = await coro
+                        run = None
+                    with set_config_context(child_config, run) as context:
+                        ret = await asyncio.create_task(coro, context=context)
+                else:
+                    ret = await coro
            except BaseException as e:
                await run_manager.on_chain_error(e)
                raise
            else:
                await run_manager.on_chain_end(ret)
        else:
-            with set_config_context(config) as context:
-                if ASYNCIO_ACCEPTS_CONTEXT:
-                    coro = cast(Coroutine[None, None, Any], self.afunc(*args, **kwargs))
-                    ret = await asyncio.create_task(coro, context=context)
-                else:
-                    ret = await self.afunc(*args, **kwargs)
-        if isinstance(ret, Runnable) and self.recurse:
+            ret = await self.afunc(*args, **kwargs)
+        if self.recurse and isinstance(ret, Runnable):
            return await ret.ainvoke(input, config)
        return ret

@@ -598,7 +600,6 @@ class RunnableSeq(Runnable):
            name=config.get("run_name") or self.get_name(),
            run_id=config.pop("run_id", None),
        )
-
        # invoke all steps in sequence
        try:
            for i, step in enumerate(self.steps):
@@ -606,8 +607,19 @@ class RunnableSeq(Runnable):
                config = patch_config(
                    config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
                )
+                # 1st step is the actual node,
+                # others are writers which don't need to be run in context
                if i == 0:
-                    input = step.invoke(input, config, **kwargs)
+                    # get the run object
+                    for h in run_manager.handlers:
+                        if isinstance(h, LangChainTracer):
+                            run = h.run_map.get(str(run_manager.run_id))
+                            break
+                    else:
+                        run = None
+                    # run in context
+                    with set_config_context(config, run) as context:
+                        input = context.run(step.invoke, input, config, **kwargs)
                else:
                    input = step.invoke(input, config)
        # finish the root run
@@ -643,8 +655,24 @@ class RunnableSeq(Runnable):
                config = patch_config(
                    config, callbacks=run_manager.get_child(f"seq:step:{i + 1}")
                )
+                # 1st step is the actual node,
+                # others are writers which don't need to be run in context
                if i == 0:
-                    input = await step.ainvoke(input, config, **kwargs)
+                    if ASYNCIO_ACCEPTS_CONTEXT:
+                        # get the run object
+                        for h in run_manager.handlers:
+                            if isinstance(h, LangChainTracer):
+                                run = h.run_map.get(str(run_manager.run_id))
+                                break
+                        else:
+                            run = None
+                        # run in context
+                        with set_config_context(config, run) as context:
+                            input = await asyncio.create_task(
+                                step.ainvoke(input, config, **kwargs), context=context
+                            )
+                    else:
+                        input = await step.ainvoke(input, config, **kwargs)
                else:
                    input = await step.ainvoke(input, config)
        # finish the root run
@@ -672,53 +700,48 @@ class RunnableSeq(Runnable):
            name=config.get("run_name") or self.get_name(),
            run_id=config.pop("run_id", None),
        )
-
-        try:
-            # stream the last steps
-            # transform the input stream of each step with the next
-            # steps that don't natively support transforming an input stream will
-            # buffer input in memory until all available, and then start emitting output
-            for idx, step in enumerate(self.steps):
-                config = patch_config(
-                    config,
-                    callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
-                )
-                if idx == 0:
-                    iterator = step.stream(input, config, **kwargs)
-                else:
-                    iterator = step.transform(iterator, config)
-            if _StreamingCallbackHandler is not None and (
-                stream_handler := next(
-                    (
-                        cast(_StreamingCallbackHandler, h)
-                        for h in run_manager.handlers
-                        if isinstance(h, _StreamingCallbackHandler)
-                    ),
-                    None,
-                )
-            ):
-                # populates streamed_output in astream_log() output if needed
-                iterator = stream_handler.tap_output_iter(run_manager.run_id, iterator)
-            output: Any = None
-            add_supported = False
-            for chunk in iterator:
-                yield chunk
-                # collect final output
-                if output is None:
-                    output = chunk
-                elif add_supported:
-                    try:
-                        output = output + chunk
-                    except TypeError:
-                        output = chunk
-                        add_supported = False
-                else:
-                    output = chunk
-        except BaseException as e:
-            run_manager.on_chain_error(e)
-            raise
+        # get the run object
+        for h in run_manager.handlers:
+            if isinstance(h, LangChainTracer):
+                run = h.run_map.get(str(run_manager.run_id))
+                break
        else:
-            run_manager.on_chain_end(output)
+            run = None
+        # create first step config
+        config = patch_config(
+            config,
+            callbacks=run_manager.get_child(f"seq:step:{1}"),
+        )
+        # run all in context
+        with set_config_context(config, run) as context:
+            try:
+                # stream the last steps
+                # transform the input stream of each step with the next
+                # steps that don't natively support transforming an input stream will
+                # buffer input in memory until all available, and then start emitting output
+                for idx, step in enumerate(self.steps):
+                    if idx == 0:
+                        iterator = step.stream(input, config, **kwargs)
+                    else:
+                        config = patch_config(
+                            config,
+                            callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
+                        )
+                        iterator = step.transform(iterator, config)
+                # populates streamed_output in astream_log() output if needed
+                if _StreamingCallbackHandler is not None:
+                    for h in run_manager.handlers:
+                        if isinstance(h, _StreamingCallbackHandler):
+                            iterator = h.tap_output_iter(run_manager.run_id, iterator)
+                # consume into final output
+                output = context.run(_consume_iter, iterator)
+                # sequence doesn't emit output, yield to mark as generator
+                yield
+            except BaseException as e:
+                run_manager.on_chain_error(e)
+                raise
+            else:
+                run_manager.on_chain_end(output)

    async def astream(
        self,
@@ -737,53 +760,121 @@ class RunnableSeq(Runnable):
            name=config.get("run_name") or self.get_name(),
            run_id=config.pop("run_id", None),
        )
-
-        try:
-            async with AsyncExitStack() as stack:
-                # stream the last steps
-                # transform the input stream of each step with the next
-                # steps that don't natively support transforming an input stream will
-                # buffer input in memory until all available, and then start emitting output
-                for idx, step in enumerate(self.steps):
-                    config = patch_config(
-                        config,
-                        callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
-                    )
-                    if idx == 0:
-                        aiterator = step.astream(input, config, **kwargs)
-                    else:
-                        aiterator = step.atransform(aiterator, config)
-                    if hasattr(aiterator, "aclose"):
-                        stack.push_async_callback(aiterator.aclose)
-                if _StreamingCallbackHandler is not None and (
-                    stream_handler := next(
-                        (
-                            cast(_StreamingCallbackHandler, h)
-                            for h in run_manager.handlers
-                            if isinstance(h, _StreamingCallbackHandler)
-                        ),
-                        None,
-                    )
-                ):
-                    # populates streamed_output in astream_log() output if needed
-                    aiterator = stream_handler.tap_output_aiter(
-                        run_manager.run_id, aiterator
-                    )
-                output: Any = None
-                add_supported = False
-                async for chunk in aiterator:
-                    yield chunk
-                    # collect final output
-                    if add_supported:
-                        try:
-                            output = output + chunk
-                        except TypeError:
-                            output = chunk
-                            add_supported = False
-                    else:
-                        output = chunk
-        except BaseException as e:
-            await run_manager.on_chain_error(e)
-            raise
+        # stream the last steps
+        # transform the input stream of each step with the next
+        # steps that don't natively support transforming an input stream will
+        # buffer input in memory until all available, and then start emitting output
+        if ASYNCIO_ACCEPTS_CONTEXT:
+            # get the run object
+            for h in run_manager.handlers:
+                if isinstance(h, LangChainTracer):
+                    run = h.run_map.get(str(run_manager.run_id))
+                    break
+            else:
+                run = None
+            # create first step config
+            config = patch_config(
+                config,
+                callbacks=run_manager.get_child(f"seq:step:{1}"),
+            )
+            # run all in context
+            with set_config_context(config, run) as context:
+                try:
+                    async with AsyncExitStack() as stack:
+                        for idx, step in enumerate(self.steps):
+                            if idx == 0:
+                                aiterator = step.astream(input, config, **kwargs)
+                            else:
+                                config = patch_config(
+                                    config,
+                                    callbacks=run_manager.get_child(
+                                        f"seq:step:{idx + 1}"
+                                    ),
+                                )
+                                aiterator = step.atransform(aiterator, config)
+                            if hasattr(aiterator, "aclose"):
+                                stack.push_async_callback(aiterator.aclose)
+                        # populates streamed_output in astream_log() output if needed
+                        if _StreamingCallbackHandler is not None:
+                            for h in run_manager.handlers:
+                                if isinstance(h, _StreamingCallbackHandler):
+                                    aiterator = h.tap_output_aiter(
+                                        run_manager.run_id, aiterator
+                                    )
+                        # consume into final output
+                        output = await asyncio.create_task(
+                            _consume_aiter(aiterator), context=context
+                        )
+                        # sequence doesn't emit output, yield to mark as generator
+                        yield
+                except BaseException as e:
+                    await run_manager.on_chain_error(e)
+                    raise
+                else:
+                    await run_manager.on_chain_end(output)
        else:
-            await run_manager.on_chain_end(output)
+            try:
+                async with AsyncExitStack() as stack:
+                    for idx, step in enumerate(self.steps):
+                        config = patch_config(
+                            config,
+                            callbacks=run_manager.get_child(f"seq:step:{idx + 1}"),
+                        )
+                        if idx == 0:
+                            aiterator = step.astream(input, config, **kwargs)
+                        else:
+                            aiterator = step.atransform(aiterator, config)
+                        if hasattr(aiterator, "aclose"):
+                            stack.push_async_callback(aiterator.aclose)
+                    # populates streamed_output in astream_log() output if needed
+                    if _StreamingCallbackHandler is not None:
+                        for h in run_manager.handlers:
+                            if isinstance(h, _StreamingCallbackHandler):
+                                aiterator = h.tap_output_aiter(
+                                    run_manager.run_id, aiterator
+                                )
+                    # consume into final output
+                    output = await _consume_aiter(aiterator)
+                    # sequence doesn't emit output, yield to mark as generator
+                    yield
+            except BaseException as e:
+                await run_manager.on_chain_error(e)
+                raise
+            else:
+                await run_manager.on_chain_end(output)
+
+
+def _consume_iter(it: Iterator[Any]) -> Any:
+    """Consume an iterator."""
+    output: Any = None
+    add_supported = False
+    for chunk in it:
+        # collect final output
+        if output is None:
+            output = chunk
+        elif add_supported:
+            try:
+                output = output + chunk
+            except TypeError:
+                output = chunk
+                add_supported = False
+        else:
+            output = chunk
+    return output
+
+
+async def _consume_aiter(it: AsyncIterator[Any]) -> Any:
+    """Consume an async iterator."""
+    output: Any = None
+    add_supported = False
+    async for chunk in it:
+        # collect final output
+        if add_supported:
+            try:
+                output = output + chunk
+            except TypeError:
+                output = chunk
+                add_supported = False
+        else:
+            output = chunk
+    return output
@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "langgraph"
-version = "0.3.28"
+version = "0.3.30"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 license = "MIT"
@@ -4660,7 +4660,7 @@ def test_root_graph(
                    content="result for query",
                    name="search_api",
                    tool_call_id="tool_call123",
-                    id="00000000-0000-4000-8000-000000000040",
+                    id="00000000-0000-4000-8000-000000000024",
                )
            ]
        },
@@ -4683,7 +4683,7 @@ def test_root_graph(
                    content="result for another",
                    name="search_api",
                    tool_call_id="tool_call456",
-                    id="00000000-0000-4000-8000-000000000049",
+                    id="00000000-0000-4000-8000-000000000030",
                )
            ]
        },
@@ -5387,7 +5387,7 @@ def test_root_graph(
        "__root__": [
            HumanMessage(
                content="what is weather in sf",
-                id="00000000-0000-4000-8000-000000000083",
+                id="00000000-0000-4000-8000-000000000051",
            ),
            AIMessage(
                content="",
@@ -5407,7 +5407,7 @@ def test_root_graph(
            ),
            AIMessage(content="answer", id="ai2"),
            AIMessage(
-                content="an extra message", id="00000000-0000-4000-8000-000000000107"
+                content="an extra message", id="00000000-0000-4000-8000-000000000066"
            ),
            HumanMessage(content="what is weather in la"),
        ],
@@ -7168,6 +7168,52 @@ def test_tags_stream_mode_messages() -> None:
    ]


+def test_stream_mode_messages_command() -> None:
+    from langchain_core.messages import HumanMessage
+
+    def my_node(state):
+        return {"messages": HumanMessage(content="foo")}
+
+    def my_other_node(state):
+        return Command(update={"messages": HumanMessage(content="bar")})
+
+    graph = (
+        StateGraph(MessagesState)
+        .add_sequence([my_node, my_other_node])
+        .add_edge(START, "my_node")
+        .compile()
+    )
+    assert list(
+        graph.stream(
+            {
+                "messages": [],
+            },
+            stream_mode="messages",
+        )
+    ) == [
+        (
+            _AnyIdHumanMessage(content="foo"),
+            {
+                "langgraph_step": 1,
+                "langgraph_node": "my_node",
+                "langgraph_triggers": ("branch:to:my_node",),
+                "langgraph_path": ("__pregel_pull", "my_node"),
+                "langgraph_checkpoint_ns": AnyStr("my_node:"),
+            },
+        ),
+        (
+            _AnyIdHumanMessage(content="bar"),
+            {
+                "langgraph_step": 2,
+                "langgraph_node": "my_other_node",
+                "langgraph_triggers": ("branch:to:my_other_node",),
+                "langgraph_path": ("__pregel_pull", "my_other_node"),
+                "langgraph_checkpoint_ns": AnyStr("my_other_node:"),
+            },
+        ),
+    ]
+
+
 def test_node_destinations() -> None:
    class State(TypedDict):
        foo: Annotated[str, operator.add]
@@ -7246,6 +7292,39 @@ def test_pydantic_none_state_update() -> None:
    assert graph.invoke({"foo": ""}) == {"foo": None}


+def test_pydantic_state_update_command() -> None:
+    from pydantic import BaseModel
+
+    class State(BaseModel):
+        foo: Optional[str]
+
+    def node_a(state: State) -> State:
+        return Command(update=State(foo=None))
+
+    graph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()
+    assert graph.invoke({"foo": ""}) == {"foo": None}
+
+    class State(BaseModel):
+        foo: Optional[str] = None
+        bar: Optional[str] = None
+
+    def node_a(state: State):
+        return State(foo="foo")
+
+    def node_b(state: State):
+        return Command(update=State(bar="bar"))
+
+    builder = StateGraph(State)
+    builder.add_node(node_a)
+    builder.add_node(node_b)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_b", END)
+    graph = builder.compile()
+
+    assert graph.invoke(State()) == {"foo": "foo", "bar": "bar"}
+
+
 def test_pydantic_state_mutation() -> None:
    from pydantic import BaseModel, Field

@@ -7280,6 +7359,40 @@ def test_pydantic_state_mutation() -> None:
    assert graph.invoke({"outer": 1}) == {"outer": 10, "inner": Inner(a=5)}


+def test_pydantic_state_mutation_command() -> None:
+    from pydantic import BaseModel, Field
+
+    class Inner(BaseModel):
+        a: int = 0
+
+    class State(BaseModel):
+        inner: Inner = Inner()
+        outer: int = 0
+
+    def my_node(state: State) -> State:
+        state.inner.a = 5
+        state.outer = 10
+        return Command(update=state)
+
+    graph = StateGraph(State).add_node(my_node).add_edge(START, "my_node").compile()
+
+    assert graph.invoke({"outer": 1}) == {"outer": 10, "inner": Inner(a=5)}
+
+    # test w/ default_factory
+    class State(BaseModel):
+        inner: Inner = Field(default_factory=Inner)
+        outer: int = 0
+
+    def my_node(state: State) -> State:
+        state.inner.a = 5
+        state.outer = 10
+        return Command(update=state)
+
+    graph = StateGraph(State).add_node(my_node).add_edge(START, "my_node").compile()
+
+    assert graph.invoke({"outer": 1}) == {"outer": 10, "inner": Inner(a=5)}
+
+
 def test_get_stream_writer() -> None:
    class State(TypedDict):
        foo: str
@@ -7894,6 +7894,53 @@ async def test_tags_stream_mode_messages() -> None:
    ]


+async def test_stream_mode_messages_command() -> None:
+    from langchain_core.messages import HumanMessage
+
+    async def my_node(state):
+        return {"messages": HumanMessage(content="foo")}
+
+    async def my_other_node(state):
+        return Command(update={"messages": HumanMessage(content="bar")})
+
+    graph = (
+        StateGraph(MessagesState)
+        .add_sequence([my_node, my_other_node])
+        .add_edge(START, "my_node")
+        .compile()
+    )
+    assert [
+        c
+        async for c in graph.astream(
+            {
+                "messages": [],
+            },
+            stream_mode="messages",
+        )
+    ] == [
+        (
+            _AnyIdHumanMessage(content="foo"),
+            {
+                "langgraph_step": 1,
+                "langgraph_node": "my_node",
+                "langgraph_triggers": ("branch:to:my_node",),
+                "langgraph_path": ("__pregel_pull", "my_node"),
+                "langgraph_checkpoint_ns": AnyStr("my_node:"),
+            },
+        ),
+        (
+            _AnyIdHumanMessage(content="bar"),
+            {
+                "langgraph_step": 2,
+                "langgraph_node": "my_other_node",
+                "langgraph_triggers": ("branch:to:my_other_node",),
+                "langgraph_path": ("__pregel_pull", "my_other_node"),
+                "langgraph_checkpoint_ns": AnyStr("my_other_node:"),
+            },
+        ),
+    ]
+
+
 async def test_stream_messages_dedupe_inputs() -> None:
    from langchain_core.messages import AIMessage
Author	SHA1	Message	Date
Vadym BardaandGitHub	73b3535c4d	langgraph: release 0.3.30 (#4267 )	2025-04-14 17:05:10 -04:00
Nuno CamposandGitHub	3e0629c56c	langgraph: support streaming messages from Command.update (#4250 )	2025-04-14 12:25:20 -07:00
vbarda	04dd69b1cd	simplify	2025-04-14 14:53:08 -04:00
Nuno CamposandGitHub	ff22eb6495	langgraph: handle pydantic updates consistently in Command (#4255 ) Fixes https://github.com/langchain-ai/langgraph/issues/3950	2025-04-14 11:47:22 -07:00
vbarda	07ca03ff15	lower depth	2025-04-14 14:12:40 -04:00
William FHandGitHub	6eea15ec3b	Add store in platform (#4266 )	2025-04-14 17:52:14 +00:00
vbarda	0e111b2f44	3.9	2025-04-14 13:23:26 -04:00
vbarda	b526fe0a4b	set max recursion depth	2025-04-14 13:21:00 -04:00
vbarda	062bf4d717	Merge branch 'vb/fix-command-messages' of github.com:langchain-ai/langgraph into vb/fix-command-messages	2025-04-14 13:20:48 -04:00
vbarda	173f4f6ccf	Merge branch 'main' into vb/fix-command-messages	2025-04-14 13:13:49 -04:00
Vadym BardaandGitHub	9a45a5b0f2	Merge branch 'main' into vb/pydantic-command	2025-04-14 13:06:22 -04:00
vbarda	2c557e9e46	move to fields	2025-04-14 13:00:03 -04:00
William FHandGitHub	6c34e599ab	Re-warn for omitted nav (#4265 )	2025-04-14 09:56:41 -07:00
vbarda	d4224a7abb	Merge branch 'main' into vb/pydantic-command	2025-04-14 12:56:22 -04:00
Andrew NguonlyandGitHub	c700dab97c	docs: Add docs for `LANGSMITH_TRACING` env var (#4257 )	2025-04-13 15:33:44 -07:00
vbarda	704b78b8fe	tests	2025-04-12 10:45:10 -04:00
vbarda	2ed453debe	factor out util	2025-04-12 10:34:02 -04:00
Nuno Campos	62b2580ad5	0.3.29	2025-04-11 16:21:09 -07:00
Nuno Campos	dfbf0ddbcb	Don't run branch reader in bg thread	2025-04-11 16:20:45 -07:00
Nuno CamposandGitHub	41bb20ee5e	Reduce perf impact of set_context (#4256 ) - call it less often - find the run from the run manager at callsite	2025-04-11 15:32:42 -07:00
Nuno Campos	560d6a1f65	Reduce perf impact of set_context - call it less often - find the run from the run manager at callsite	2025-04-11 15:01:17 -07:00
vbarda	dc6fa9ed30	langgraph: handle pydantic updates consistently in Command	2025-04-11 17:51:56 -04:00
Andrew NguonlyandGitHub	a9be75f745	docs: Add Data Plane features sections for custom Postgres/Redis, tracing, telemetry, and licensing (#4254 )	2025-04-11 14:48:20 -07:00
Nuno CamposandGitHub	20e3469296	Merge branch 'main' into vb/fix-command-messages	2025-04-11 14:27:26 -07:00
Eugene Yurtsev GitHub Catherine	233cca1357	Update langgraph_platform.md (#4251 ) Co-authored-by: Catherine <catherine@langchain.dev>	2025-04-11 15:29:11 -04:00
Andrew NguonlyandGitHub	d1ac0a0e13	docs: Add alpha and beta labels for respective LangGraph Platform deployment options (#4249 ) ### Summary Examples: ![image](https://github.com/user-attachments/assets/2a36a262-5373-498d-9907-19d5447fbb6a) ![image](https://github.com/user-attachments/assets/70671e08-34b6-40ed-964d-9d195ea8308d) ![image](https://github.com/user-attachments/assets/fcb877a6-475c-47a4-b8af-91cbdc00f89b)	2025-04-11 12:01:43 -07:00
vbarda	5071a6cd97	langgraph: support streaming messages from Command.update	2025-04-11 14:05:13 -04:00
Nuno CamposandGitHub	72d7b23638	Use tuple entry for control branch (#4248 )	2025-04-11 10:29:44 -07:00
Nuno Campos	64aa1e6cd8	Use tuple entry for control branch	2025-04-11 09:53:54 -07:00
Nuno CamposandGitHub	d6f2f0c90d	Simplify path for control branch attached to every node (#4247 ) - attached to every node to handle command/send return values - used to be a full blown conditional edge, can be simpler by doing all of it in a single function	2025-04-11 09:44:18 -07:00
Nuno Campos	5a7edead8c	Lint	2025-04-11 09:20:10 -07:00
Nuno Campos	8ff5c43cf0	Avoid creating contexts for control branches	2025-04-11 09:10:58 -07:00
Nuno CamposandGitHub	0eb32a4251	Avoid validating node input more than once per superstep (#4242 )	2025-04-11 09:06:07 -07:00
Nuno Campos	3d12a2df59	Simplify path for control branch attached to every node - attached to every node to handle command/send return values - used to be a full blown conditional edge, can be simpler by doing all of it in a single function	2025-04-11 09:00:16 -07:00
Nuno Campos	04d3c9d30f	Use cache in attach_branch too	2025-04-11 08:38:14 -07:00
Nuno Campos	4d1b3370df	Lint	2025-04-10 17:38:25 -07:00
Nuno Campos	bf5017f6e0	Lint	2025-04-10 17:29:36 -07:00
Nuno Campos	64086aa814	Avoid validating node input more than once per superstep	2025-04-10 17:28:00 -07:00