mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-06 17:57:49 +02:00
Merge branch 'main' into vb/pydantic-command
This commit is contained in:
vbarda
@@ -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).
|
||||
|
||||
|
||||
@@ -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...
|
||||
|
||||
@@ -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,
|
||||
@@ -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 (
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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]]]:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -68,6 +68,7 @@ class ChannelRead(RunnableCallable):
|
||||
afunc=self._aread,
|
||||
tags=tags,
|
||||
name=None,
|
||||
trace=False,
|
||||
func_accepts_config=True,
|
||||
)
|
||||
self.fresh = fresh
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"),
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user