diff --git a/docs/docs/cloud/deployment/cloud.md b/docs/docs/cloud/deployment/cloud.md index c53d516a6..638d095ba 100644 --- a/docs/docs/cloud/deployment/cloud.md +++ b/docs/docs/cloud/deployment/cloud.md @@ -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. diff --git a/docs/docs/cloud/deployment/self_hosted_control_plane.md b/docs/docs/cloud/deployment/self_hosted_control_plane.md index f72de8bd2..c9e8265c1 100644 --- a/docs/docs/cloud/deployment/self_hosted_control_plane.md +++ b/docs/docs/cloud/deployment/self_hosted_control_plane.md @@ -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. diff --git a/docs/docs/cloud/deployment/self_hosted_data_plane.md b/docs/docs/cloud/deployment/self_hosted_data_plane.md index 652842e49..c9512eb10 100644 --- a/docs/docs/cloud/deployment/self_hosted_data_plane.md +++ b/docs/docs/cloud/deployment/self_hosted_data_plane.md @@ -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. diff --git a/docs/docs/cloud/reference/env_var.md b/docs/docs/cloud/reference/env_var.md index e4d4de0f2..e15157cac 100644 --- a/docs/docs/cloud/reference/env_var.md +++ b/docs/docs/cloud/reference/env_var.md @@ -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 Cloud SaaS(Beta)**: Available for **Plus** and **Enterprise** plans. -1. **[Self-Hosted Data Plane](#self-hosted-data-plane)**: Available for the **Enterprise** plan. +1. **Self-Hosted Data Plane(Beta)**: Available for the **Enterprise** plan. -1. **[Self-Hosted Control Plane](#self-hosted-control-plane)**: Available for the **Enterprise** plan. +1. **Self-Hosted Control Plane(Beta)**: Available for the **Enterprise** plan. 1. **[Standalone Container](#standalone-container)**: Available for all plans. diff --git a/docs/docs/concepts/index.md b/docs/docs/concepts/index.md index 9fd2f2ad2..d4cc38333 100644 --- a/docs/docs/concepts/index.md +++ b/docs/docs/concepts/index.md @@ -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. +- Cloud SaaS(Beta): Connect to your GitHub repositories and deploy LangGraph Servers to LangChain's cloud. We manage everything. +- Self-Hosted Data Plane(Beta): 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(Beta): 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. diff --git a/docs/docs/concepts/langgraph_cloud.md b/docs/docs/concepts/langgraph_cloud.md index 557b39e60..1e083dca8 100644 --- a/docs/docs/concepts/langgraph_cloud.md +++ b/docs/docs/concepts/langgraph_cloud.md @@ -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). diff --git a/docs/docs/concepts/langgraph_data_plane.md b/docs/docs/concepts/langgraph_data_plane.md index ecb154a10..fa2368387 100644 --- a/docs/docs/concepts/langgraph_data_plane.md +++ b/docs/docs/concepts/langgraph_data_plane.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://:@/?host=` and for `Deployment B`, `POSTGRES_URI_CUSTOM` can be set to `postgres://:@/?host=`. `` and `database_name_2` are different databases within the same instance, but `` 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://:/1` and for `Deployment B`, `REDIS_URI_CUSTOM` can be set to `redis://:/2`. `1` and `2` are different database numbers within the same instance, but `` 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

Trace to LangSmith SaaS. | Optional

Disable tracing or trace to LangSmith SaaS. | Optional

Disable tracing or trace to Self-Hosted LangSmith. | Optional

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.

Telemetry sent to LangSmith SaaS for LangGraph Platform License Key. | Self-reported usage (audit) for air-gapped license key.

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. | diff --git a/docs/docs/concepts/langgraph_platform.md b/docs/docs/concepts/langgraph_platform.md index bde8917d7..77d3f7c06 100644 --- a/docs/docs/concepts/langgraph_platform.md +++ b/docs/docs/concepts/langgraph_platform.md @@ -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. + + + ## Overview LangGraph Platform is a commercial solution for deploying agentic applications to production, built on the open-source [LangGraph framework](./high_level.md). diff --git a/docs/docs/concepts/langgraph_self_hosted_control_plane.md b/docs/docs/concepts/langgraph_self_hosted_control_plane.md index 9ab2c2549..718556dc7 100644 --- a/docs/docs/concepts/langgraph_self_hosted_control_plane.md +++ b/docs/docs/concepts/langgraph_self_hosted_control_plane.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). diff --git a/docs/docs/concepts/langgraph_self_hosted_data_plane.md b/docs/docs/concepts/langgraph_self_hosted_data_plane.md index bc4c65c0c..51f0114e1 100644 --- a/docs/docs/concepts/langgraph_self_hosted_data_plane.md +++ b/docs/docs/concepts/langgraph_self_hosted_data_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). diff --git a/docs/docs/tutorials/deployment.md b/docs/docs/tutorials/deployment.md index 021d52aad..d2c7567e7 100644 --- a/docs/docs/tutorials/deployment.md +++ b/docs/docs/tutorials/deployment.md @@ -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. +- Cloud SaaS(Beta): Connect to your GitHub repositories and deploy LangGraph Servers to LangChain's cloud. We manage everything. +- Self-Hosted Data Plane(Beta): 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(Beta): 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... diff --git a/libs/langgraph/bench/fanout_to_subgraph.py b/libs/langgraph/bench/fanout_to_subgraph.py index 612d0fbf2..fc8c7133b 100644 --- a/libs/langgraph/bench/fanout_to_subgraph.py +++ b/libs/langgraph/bench/fanout_to_subgraph.py @@ -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") diff --git a/libs/langgraph/langgraph/func/__init__.py b/libs/langgraph/langgraph/func/__init__.py index 8c734127c..92d8419d9 100644 --- a/libs/langgraph/langgraph/func/__init__.py +++ b/libs/langgraph/langgraph/func/__init__.py @@ -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], + ] ) ], ) diff --git a/libs/langgraph/langgraph/graph/branch.py b/libs/langgraph/langgraph/graph/branch.py index fd2039f20..33a2aca1e 100644 --- a/libs/langgraph/langgraph/graph/branch.py +++ b/libs/langgraph/langgraph/graph/branch.py @@ -1,4 +1,3 @@ -import asyncio from inspect import ( isfunction, ismethod, @@ -138,7 +137,6 @@ class Branch(NamedTuple): reader=reader, name=None, trace=False, - set_context=False, func_accepts_config=True, ) ) @@ -179,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 ( diff --git a/libs/langgraph/langgraph/graph/graph.py b/libs/langgraph/langgraph/graph/graph.py index d8bde68ec..fa28243fb 100644 --- a/libs/langgraph/langgraph/graph/graph.py +++ b/libs/langgraph/langgraph/graph/graph.py @@ -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: diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index 85acf73b3..1a58e023f 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -44,6 +44,7 @@ from langgraph.constants import ( NS_END, NS_SEP, TAG_HIDDEN, + TASKS, ) from langgraph.errors import ( ErrorCode, @@ -78,7 +79,7 @@ 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.pydantic import create_model, get_update_as_tuples -from langgraph.utils.runnable import RunnableCallable, RunnableLike, coerce_to_runnable +from langgraph.utils.runnable import RunnableLike, coerce_to_runnable logger = logging.getLogger(__name__) @@ -670,10 +671,6 @@ class StateGraph(Graph): for key, node in self.nodes.items(): compiled.attach_node(key, node) - compiled.nodes[START].writers.append(CONTROL_BRANCH_PATH) - for key in self.nodes: - compiled.nodes[key].writers.append(CONTROL_BRANCH_PATH) - for start, end in self.edges: compiled.attach_edge(start, end) @@ -777,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 @@ -785,7 +783,7 @@ 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 @@ -810,7 +808,7 @@ class CompiledStateGraph(CompiledGraph): # coerce state dict to schema class (eg. pydantic model) 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, @@ -836,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( @@ -910,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: @@ -1037,10 +1031,9 @@ def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]: return schema(**input) -def _control_branch(value: Any, config: RunnableConfig) -> Any: +def _control_branch(value: Any) -> Sequence[tuple[str, Any]]: if isinstance(value, Send): - ChannelWrite.do_write(config, (value,)) - return value + return ((TASKS, value),) commands: list[Command] = [] if isinstance(value, Command): commands.append(value) @@ -1048,66 +1041,22 @@ def _control_branch(value: Any, config: RunnableConfig) -> Any: for cmd in value: if isinstance(cmd, Command): commands.append(cmd) - rtn: list[Union[ChannelWriteEntry, 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(ChannelWriteEntry(CHANNEL_BRANCH_TO.format(command.goto), None)) + rtn.append((CHANNEL_BRANCH_TO.format(command.goto), None)) else: rtn.extend( - go + (TASKS, go) if isinstance(go, Send) - else ChannelWriteEntry(CHANNEL_BRANCH_TO.format(go), None) + else (CHANNEL_BRANCH_TO.format(go), None) for go in command.goto ) - if rtn: - ChannelWrite.do_write(config, rtn) - return value - - -async def _acontrol_branch(value: Any, config: RunnableConfig) -> Any: - if isinstance(value, Send): - ChannelWrite.do_write(config, (value,)) - 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[ChannelWriteEntry, 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(ChannelWriteEntry(CHANNEL_BRANCH_TO.format(command.goto), None)) - else: - rtn.extend( - go - if isinstance(go, Send) - else ChannelWriteEntry(CHANNEL_BRANCH_TO.format(go), None) - for go in command.goto - ) - if rtn: - ChannelWrite.do_write(config, rtn) - return value - - -CONTROL_BRANCH_PATH = RunnableCallable( - _control_branch, - _acontrol_branch, - tags=[TAG_HIDDEN], - trace=False, - recurse=False, - set_context=False, - func_accepts_config=True, -) + return rtn def _get_root(input: Any) -> Optional[Sequence[tuple[str, Any]]]: diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index c5d335d75..902317e20 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -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 diff --git a/libs/langgraph/langgraph/pregel/call.py b/libs/langgraph/langgraph/pregel/call.py index e1856aa80..a16fa36fd 100644 --- a/libs/langgraph/langgraph/pregel/call.py +++ b/libs/langgraph/langgraph/pregel/call.py @@ -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) diff --git a/libs/langgraph/langgraph/pregel/read.py b/libs/langgraph/langgraph/pregel/read.py index 5b345e811..e57620ec7 100644 --- a/libs/langgraph/langgraph/pregel/read.py +++ b/libs/langgraph/langgraph/pregel/read.py @@ -68,6 +68,7 @@ class ChannelRead(RunnableCallable): afunc=self._aread, tags=tags, name=None, + trace=False, func_accepts_config=True, ) self.fresh = fresh diff --git a/libs/langgraph/langgraph/pregel/write.py b/libs/langgraph/langgraph/pregel/write.py index 05fe5a388..234c1f5d7 100644 --- a/libs/langgraph/langgraph/pregel/write.py +++ b/libs/langgraph/langgraph/pregel/write.py @@ -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) diff --git a/libs/langgraph/langgraph/utils/runnable.py b/libs/langgraph/langgraph/utils/runnable.py index 2736cc7d5..c4ba527d6 100644 --- a/libs/langgraph/langgraph/utils/runnable.py +++ b/libs/langgraph/langgraph/utils/runnable.py @@ -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.""" @@ -254,7 +248,6 @@ class RunnableCallable(Runnable): tags: Optional[Sequence[str]] = None, trace: bool = True, recurse: bool = True, - set_context: bool = True, explode_args: bool = False, func_accepts_config: Optional[bool] = None, **kwargs: Any, @@ -278,7 +271,6 @@ class RunnableCallable(Runnable): self.kwargs = kwargs self.trace = trace self.recurse = recurse - self.set_context = set_context self.explode_args = explode_args # check signature if func is None and afunc is None: @@ -365,19 +357,21 @@ class RunnableCallable(Runnable): ) try: child_config = patch_config(config, callbacks=run_manager.get_child()) - if self.set_context: - with set_config_context(child_config) as context: - ret = context.run(self.func, *args, **kwargs) + # 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: - ret = self.func(*args, **kwargs) + 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) raise else: run_manager.on_chain_end(ret) - elif self.set_context: - with set_config_context(config) as context: - ret = context.run(self.func, *args, **kwargs) else: ret = self.func(*args, **kwargs) if self.recurse and isinstance(ret, Runnable): @@ -425,8 +419,14 @@ class RunnableCallable(Runnable): try: child_config = patch_config(config, callbacks=run_manager.get_child()) coro = cast(Coroutine[None, None, Any], self.afunc(*args, **kwargs)) - if ASYNCIO_ACCEPTS_CONTEXT and self.set_context: - with set_config_context(child_config) as context: + 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: + run = None + with set_config_context(child_config, run) as context: ret = await asyncio.create_task(coro, context=context) else: ret = await coro @@ -435,10 +435,6 @@ class RunnableCallable(Runnable): raise else: await run_manager.on_chain_end(ret) - elif ASYNCIO_ACCEPTS_CONTEXT and self.set_context: - with set_config_context(config) as 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 self.recurse and isinstance(ret, Runnable): @@ -604,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): @@ -612,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 @@ -649,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 @@ -678,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, @@ -743,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 diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 75d349bd9..8adf1d9cf 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langgraph" -version = "0.3.28" +version = "0.3.29" description = "Building stateful, multi-actor applications with LLMs" authors = [] license = "MIT" diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index f5be91cee..7bba55ff5 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -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"), ],