WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-17 21:25:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into fix-examples-link-readme

Browse Source
This commit is contained in:
Lauren Hirata Singh
2025-05-23 22:08:32 -04:00
committed by GitHub
parent 6fc21046cd 6633173918
commit a69860baa6
27 changed files with 3766 additions and 3635 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs/agents/tools.md
+15 -1
View File
@@ -280,7 +280,21 @@ LangGraph allows access to short-term and long-term memory from tools. See [Memo
## Prebuilt tools
LangChain supports a wide range of prebuilt tool integrations for interacting with APIs, databases, file systems, web data, and more. These tools extend the functionality of agents and enable rapid development.
You can use prebuilt tools from model providers by passing a dictionary with tool specs to the `tools` parameter of `create_react_agent`. For example, to use the `web_search_preview` tool from OpenAI:
```python
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(
model="openai:gpt-4o-mini",
tools=[{"type": "web_search_preview"}]
)
response = agent.invoke(
{"messages": ["What was a positive news story from today?"]}
)
```
Additionally, LangChain supports a wide range of prebuilt tool integrations for interacting with APIs, databases, file systems, web data, and more. These tools extend the functionality of agents and enable rapid development.
You can browse the full list of available integrations in the [LangChain integrations directory](https://python.langchain.com/docs/integrations/tools/).
docs/docs/cloud/deployment/self_hosted_control_plane.md
+2 -2
View File
@@ -2,8 +2,8 @@
Before deploying, review the [conceptual guide for the Self-Hosted Control Plane](../../concepts/langgraph_self_hosted_control_plane.md) deployment option.
!!! important "Beta"
The Self-Hosted Control Plane deployment option is currently in beta stage.
!!! info "Important"
The Self-Hosted Control Plane deployment option is currently in beta stage and requires an [Enterprise](../../concepts/plans.md) plan.
## Prerequisites
docs/docs/cloud/deployment/self_hosted_data_plane.md
+2 -2
View File
@@ -2,8 +2,8 @@
Before deploying, review the [conceptual guide for the Self-Hosted Data Plane](../../concepts/langgraph_self_hosted_data_plane.md) deployment option.
!!! important "Beta"
The Self-Hosted Data Plane deployment option is currently in beta stage.
!!! info "Important"
The Self-Hosted Data Plane deployment option is currently in beta stage and requires an [Enterprise](../../concepts/plans.md) plan.
## Prerequisites
docs/docs/concepts/deployment_options.md
+4 -4
View File
@@ -40,8 +40,8 @@ For more information, please see:
## Self-Hosted Data Plane
!!! important "Beta"
The Self-Hosted Data Plane deployment option is currently in beta stage.
!!! info "Important"
The Self-Hosted Data Plane deployment option is currently in beta stage and requires an [Enterprise](../concepts/plans.md) plan.
The [Self-Hosted Data Plane](./langgraph_self_hosted_data_plane.md) deployment option is a "hybrid" model for deployment where we manage the [control plane](./langgraph_control_plane.md) in our cloud and you manage the [data plane](./langgraph_data_plane.md) in your cloud. This option provides a way to securely manage your data plane infrastructure, while offloading control plane management to us.
@@ -56,8 +56,8 @@ For more information, please see:
## Self-Hosted Control Plane
!!! important "Beta"
The Self-Hosted Control Plane deployment option is currently in beta stage.
!!! info "Important"
The Self-Hosted Control Plane deployment option is currently in beta stage and requires an [Enterprise](../concepts/plans.md) plan.
The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
docs/docs/concepts/langgraph_control_plane.md
+11 -6
View File
@@ -47,17 +47,22 @@ This section describes various features of the control plane.
For simplicity, the control plane offers two deployment types with different resource allocations: `Development` and `Production`.
| **Deployment Type** | **CPU** | **Memory** | **Scaling** |
|---------------------|---------|------------|---------------------|
| Development | 1 CPU | 1 GB | Up to 1 container |
| Production | 2 CPU | 2 GB | Up to 10 containers |
| **Deployment Type** | **CPU/Memory** | **Scaling** | **Database** |
|---------------------|-----------------|---------------------|----------------------------------------------------------------------------------|
| Development | 1 CPU, 1 GB RAM | Up to 1 container | 10 GB disk, no backups |
| Production | 2 CPU, 2 GB RAM | Up to 10 containers | Autoscaling disk, automatic backups, highly available (multi-zone configuration) |
CPU and memory resources are per container.
!!! info "For [Cloud SaaS](../concepts/langgraph_cloud.md)"
!!! warning "Immutable Deployment Type"
Once a deployment is created, the deployment type cannot be changed.
!!! info "Resource Customization"
For `Production` type deployments, resources can be manually increased on a case-by-case basis depending on use case and capacity constraints. Contact support@langchain.dev to request an increase in resources.
!!! info
For `Development` types deployments, database disk size can be manually increased on a case-by-case basis depending on use case and capacity constraints. For most use cases, [TTLs](../how-tos/ttl/configure_ttl.md) should be configured to manage disk usage. Contact support@langchain.dev to request an increase in resources.
Resources 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 can be fully customized.
### Database Provisioning
docs/docs/concepts/langgraph_self_hosted_control_plane.md
+2 -2
View File
@@ -2,8 +2,8 @@
There are two versions of the self-hosted deployment: [Self-Hosted Data Plane](./deployment_options.md#self-hosted-data-plane) and [Self-Hosted Control Plane](./deployment_options.md#self-hosted-control-plane).
!!! important "Beta"
The Self-Hosted Control Plane deployment option is currently in beta stage.
!!! info "Important"
The Self-Hosted Control Plane deployment option is currently in beta stage and requires an [Enterprise](../../concepts/plans.md) plan.
## Requirements
docs/docs/concepts/langgraph_self_hosted_data_plane.md
+2 -2
View File
@@ -7,8 +7,8 @@ search:
There are two versions of the self-hosted deployment: [Self-Hosted Data Plane](./deployment_options.md#self-hosted-data-plane) and [Self-Hosted Control Plane](./deployment_options.md#self-hosted-control-plane).
!!! important "Beta"
The Self-Hosted Data Plane deployment option is currently in beta stage.
!!! info "Important"
The Self-Hosted Data Plane deployment option is currently in beta stage and requires an [Enterprise](../../concepts/plans.md) plan.
## Requirements
docs/docs/how-tos/graph-api.ipynb
+1 -1
View File
@@ -2235,7 +2235,7 @@
" if termination_condition(state):\n",
" return END\n",
" else:\n",
" return \"a\"\n",
" return \"b\"\n",
"\n",
"builder.add_edge(START, \"a\")\n",
"builder.add_conditional_edges(\"a\", route)\n",
docs/uv.lock
Generated
+3 -3
View File
@@ -2891,7 +2891,7 @@ test = [
[[package]]
name = "langgraph-prebuilt"
version = "0.1.8"
version = "0.2.1"
source = { editable = "../libs/prebuilt" }
dependencies = [
{ name = "langchain-core" },
@@ -2948,8 +2948,8 @@ dev = [
[[package]]
name = "langgraph-supervisor"
version = "0.0.21"
source = { git = "https://github.com/langchain-ai/langgraph-supervisor-py#6367bebd5462ac899e7def931ac6ab9cc6a9b070" }
version = "0.0.25"
source = { git = "https://github.com/langchain-ai/langgraph-supervisor-py#79380b5c21d3170e2d20dc6c55149ee057a306b1" }
dependencies = [
{ name = "langchain-core" },
{ name = "langgraph" },
libs/langgraph/langgraph/graph/message.py
+10 -1
View File
@@ -23,6 +23,7 @@ from langchain_core.messages import (
)
from typing_extensions import TypedDict
from langgraph.constants import CONF, CONFIG_KEY_SEND
from langgraph.graph.state import StateGraph
Messages = Union[list[MessageLikeRepresentation], MessageLikeRepresentation]
@@ -298,8 +299,13 @@ def _format_messages(messages: Sequence[BaseMessage]) -> list[BaseMessage]:
def push_message(
message: Union[MessageLikeRepresentation, BaseMessageChunk],
*,
state_key: Optional[str] = "messages",
) -> AnyMessage:
"""Write a message manually to the `messages` / `messages-tuple` stream mode."""
"""Write a message manually to the `messages` / `messages-tuple` stream mode.
Will automatically write to the channel specified in the `state_key` unless `state_key` is `None`.
"""
from langchain_core.callbacks.base import (
BaseCallbackHandler,
@@ -334,4 +340,7 @@ def push_message(
)
stream_handler._emit(message_meta, message, dedupe=False)
if state_key:
config[CONF][CONFIG_KEY_SEND]([(state_key, message)])
return message
libs/langgraph/langgraph/graph/ui.py
+3 -2
View File
@@ -54,7 +54,7 @@ def push_ui_message(
id: Optional[str] = None,
metadata: Optional[dict[str, Any]] = None,
message: Optional[AnyMessage] = None,
state_key: str = "ui",
state_key: Optional[str] = "ui",
merge: bool = False,
) -> UIMessage:
"""Push a new UI message to update the UI state.
@@ -111,7 +111,8 @@ def push_ui_message(
}
writer(evt)
config[CONF][CONFIG_KEY_SEND]([(state_key, evt)])
if state_key:
config[CONF][CONFIG_KEY_SEND]([(state_key, evt)])
return evt
libs/langgraph/langgraph/pregel/__init__.py
+6 -4
View File
@@ -2214,12 +2214,14 @@ class Pregel(PregelProtocol):
validate_keys(output_keys, self.channels)
interrupt_before = interrupt_before or self.interrupt_before_nodes
interrupt_after = interrupt_after or self.interrupt_after_nodes
stream_mode = stream_mode if stream_mode is not None else self.stream_mode
if stream_mode is None and CONFIG_KEY_TASK_ID in config.get(CONF, {}):
# if being called as a node in another graph, default to values mode
# but don't overwrite stream_mode arg if provided
stream_mode = ["values"]
elif stream_mode is None:
stream_mode = self.stream_mode
if not isinstance(stream_mode, list):
stream_mode = [stream_mode]
if CONFIG_KEY_TASK_ID in config.get(CONF, {}):
# if being called as a node in another graph, always use values mode
stream_mode = ["values"]
if self.checkpointer is False:
checkpointer: BaseCheckpointSaver | None = None
elif CONFIG_KEY_CHECKPOINTER in config.get(CONF, {}):
libs/langgraph/langgraph/pregel/loop.py
-1
View File
@@ -135,7 +135,6 @@ P = ParamSpec("P")
INPUT_DONE = object()
INPUT_RESUMING = object()
INPUT_SHOULD_VALIDATE = object()
SPECIAL_CHANNELS = (ERROR, INTERRUPT, SCHEDULED)
WritesT = Sequence[tuple[str, Any]]
libs/langgraph/langgraph/pregel/runner.py
+19 -17
View File
@@ -56,6 +56,10 @@ EXCLUDED_FRAME_FNAMES = (
"concurrent/futures/_base.py",
)
SKIP_RERAISE_SET: weakref.WeakSet[Union[concurrent.futures.Future, asyncio.Future]] = (
weakref.WeakSet()
)
class FuturesDict(Generic[F, E], dict[F, Optional[PregelExecutableTask]]):
event: E
@@ -165,7 +169,6 @@ class PregelRunner:
futures=weakref.ref(futures),
schedule_task=schedule_task,
submit=self.submit,
reraise=reraise,
),
},
)
@@ -207,7 +210,6 @@ class PregelRunner:
futures=weakref.ref(futures),
schedule_task=schedule_task,
submit=self.submit,
reraise=reraise,
),
},
__reraise_on_exit__=reraise,
@@ -302,7 +304,6 @@ class PregelRunner:
futures=weakref.ref(futures),
schedule_task=schedule_task,
submit=self.submit,
reraise=reraise,
loop=loop,
),
},
@@ -349,7 +350,6 @@ class PregelRunner:
futures=weakref.ref(futures),
schedule_task=schedule_task,
submit=self.submit,
reraise=reraise,
loop=loop,
),
},
@@ -434,7 +434,8 @@ class PregelRunner:
raise exception
else:
# save error to checkpointer
self.put_writes()(task.id, [(ERROR, exception)]) # type: ignore[misc]
task.writes.append((ERROR, exception))
self.put_writes()(task.id, task.writes) # type: ignore[misc]
else:
if self.node_finished and (
task.config is None or TAG_HIDDEN not in task.config.get("tags", [])
@@ -456,7 +457,7 @@ def _should_stop_others(
if fut.cancelled():
continue
elif exc := fut.exception():
if not isinstance(exc, GraphBubbleUp):
if not isinstance(exc, GraphBubbleUp) and fut not in SKIP_RERAISE_SET:
return True
return False
@@ -494,7 +495,8 @@ def _panic_or_proceed(
interrupts: list[GraphInterrupt] = []
while done:
# if any task failed
if exc := _exception(done.pop()):
fut = done.pop()
if exc := _exception(fut):
# cancel all pending tasks
while inflight:
inflight.pop().cancel()
@@ -503,7 +505,7 @@ def _panic_or_proceed(
if isinstance(exc, GraphInterrupt):
# collect interrupts
interrupts.append(exc)
else:
elif fut not in SKIP_RERAISE_SET:
raise exc
# raise combined interrupts
if interrupts:
@@ -530,7 +532,6 @@ def _call(
[PregelExecutableTask, int, Optional[Call]], Optional[PregelExecutableTask]
],
submit: weakref.ref[Submit],
reraise: bool,
) -> concurrent.futures.Future[Any]:
if asyncio.iscoroutinefunction(func):
raise RuntimeError("In an sync context async tasks cannot be called")
@@ -582,14 +583,16 @@ def _call(
callbacks=callbacks,
schedule_task=schedule_task,
submit=submit,
reraise=reraise,
),
},
__reraise_on_exit__=reraise,
__reraise_on_exit__=False,
# starting a new task in the next tick ensures
# updates from this tick are committed/streamed first
__next_tick__=True,
)
# exceptions for call() tasks are raised into the parent task
# so we should not re-raise at the end of the tick
SKIP_RERAISE_SET.add(fut)
futures()[fut] = next_task # type: ignore[index]
fut = cast(Union[asyncio.Future, concurrent.futures.Future], fut)
# return a chained future to ensure commit() callback is called
@@ -613,7 +616,6 @@ def _acall(
],
submit: weakref.ref[Submit],
loop: asyncio.AbstractEventLoop,
reraise: bool = False,
stream: bool = False,
) -> Union[asyncio.Future[Any], concurrent.futures.Future[Any]]:
# return a chained future to ensure commit() callback is called
@@ -643,7 +645,6 @@ def _acall(
schedule_task=schedule_task,
submit=submit,
loop=loop,
reraise=reraise,
stream=stream,
),
loop,
@@ -669,7 +670,6 @@ async def _acall_impl(
],
submit: weakref.ref[Submit],
loop: asyncio.AbstractEventLoop,
reraise: bool = False,
stream: bool = False,
) -> None:
try:
@@ -726,17 +726,19 @@ async def _acall_impl(
schedule_task=schedule_task,
submit=submit,
loop=loop,
reraise=reraise,
),
},
__name__=task().name, # type: ignore[union-attr]
__name__=next_task.name,
__cancel_on_exit__=True,
__reraise_on_exit__=reraise,
__reraise_on_exit__=False,
# starting a new task in the next tick ensures
# updates from this tick are committed/streamed first
__next_tick__=True,
),
)
# exceptions for call() tasks are raised into the parent task
# so we should not re-raise at the end of the tick
SKIP_RERAISE_SET.add(fut)
futures()[fut] = next_task # type: ignore[index]
if fut is not None:
chain_future(fut, destination)
libs/langgraph/pyproject.toml
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph"
version = "0.4.5"
version = "0.4.7"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
requires-python = ">=3.9"
@@ -15,7 +15,7 @@ dependencies = [
"langchain-core>=0.1",
"langgraph-checkpoint>=2.0.26",
"langgraph-sdk>=0.1.42",
"langgraph-prebuilt>=0.1.8",
"langgraph-prebuilt>=0.2.0",
"xxhash>=3.5.0",
"pydantic>=2.7.4",
]
libs/langgraph/tests/test_messages_state.py
+3 -7
View File
@@ -342,13 +342,9 @@ def test_push_messages_in_graph():
with pytest.raises(ValueError, match="Message ID is required"):
push_message(AIMessage(content="No ID"))
return {
"messages": [
push_message(AIMessage(content="First", id="1")),
push_message(HumanMessage(content="Second", id="2")),
push_message(AIMessage(content="Third", id="3")),
]
}
push_message(AIMessage(content="First", id="1"))
push_message(HumanMessage(content="Second", id="2"))
push_message(AIMessage(content="Third", id="3"))
builder = StateGraph(MessagesState)
builder.add_node(chat)
libs/langgraph/tests/test_pregel.py
+34 -1
View File
@@ -6873,7 +6873,7 @@ def test_sync_streaming_with_functional_api() -> None:
should be greater than the time delay between the two tasks.
"""
time_delay = 0.01
time_delay = 0.05
@task()
def slow() -> dict:
@@ -8769,3 +8769,36 @@ def test_get_graph_root_channel(snapshot: SnapshotAssertion) -> None:
assert json.dumps(graph.get_graph().to_json(), indent=2) == snapshot
assert graph.get_graph().draw_mermaid(with_styles=False) == snapshot
def test_imp_exception(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
@task()
def my_task(number: int):
time.sleep(0.1)
return number * 2
@task()
def task_with_exception(number: int):
time.sleep(0.1)
raise Exception("This is a test exception")
@entrypoint(checkpointer=sync_checkpointer)
def my_workflow(number: int):
my_task(number).result()
try:
task_with_exception(number).result()
except Exception as e:
print(f"Exception caught: {e}")
my_task(number).result()
return "done"
thread1 = {"configurable": {"thread_id": "1"}}
assert my_workflow.invoke(1, thread1) == "done"
assert [c for c in my_workflow.stream(1, thread1)] == [
{"my_task": 2},
{"my_task": 2},
{"my_workflow": "done"},
]
libs/langgraph/tests/test_pregel_async.py
+298
View File
@@ -9148,3 +9148,301 @@ async def test_draw_invalid():
{"source": "nothing", "target": "__end__"},
],
}
@NEEDS_CONTEXTVARS
async def test_imp_exception(
async_checkpointer: BaseCheckpointSaver,
) -> None:
@task()
async def my_task(number: int):
await asyncio.sleep(0.1)
return number * 2
@task()
async def task_with_exception(number: int):
await asyncio.sleep(0.1)
raise Exception("This is a test exception")
@entrypoint(checkpointer=async_checkpointer)
async def my_workflow(number: int):
await my_task(number)
try:
await task_with_exception(number)
except Exception as e:
print(f"Exception caught: {e}")
await my_task(number)
return "done"
thread1 = {"configurable": {"thread_id": "1"}}
assert await my_workflow.ainvoke(1, thread1) == "done"
assert [c async for c in my_workflow.astream(1, thread1)] == [
{"my_task": 2},
{"my_task": 2},
{"my_workflow": "done"},
]
assert [c async for c in my_workflow.astream_events(1, thread1)] == [
{
"event": "on_chain_start",
"data": {"input": 1},
"name": "LangGraph",
"tags": [],
"run_id": AnyStr(),
"metadata": {"thread_id": "1"},
"parent_ids": [],
},
{
"event": "on_chain_start",
"data": {"input": 1},
"name": "my_workflow",
"tags": ["graph:step:4"],
"run_id": AnyStr(),
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_workflow",
"langgraph_triggers": ("__start__",),
"langgraph_path": ("__pregel_pull", "my_workflow"),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [AnyStr()],
},
{
"event": "on_chain_start",
"data": {"input": {"number": 1}},
"name": "my_task",
"tags": ["seq:step:1"],
"run_id": AnyStr(),
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_stream",
"run_id": AnyStr(),
"name": "my_task",
"tags": ["seq:step:1"],
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"data": {"chunk": 2},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_end",
"data": {"output": 2, "input": {"number": 1}},
"run_id": AnyStr(),
"name": "my_task",
"tags": ["seq:step:1"],
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_stream",
"run_id": AnyStr(),
"name": "LangGraph",
"tags": [],
"metadata": {"thread_id": "1"},
"data": {"chunk": {"my_task": 2}},
"parent_ids": [],
},
{
"event": "on_chain_start",
"data": {"input": {"number": 1}},
"name": "task_with_exception",
"tags": ["seq:step:1"],
"run_id": AnyStr(),
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_start",
"data": {"input": {"number": 1}},
"name": "my_task",
"tags": ["seq:step:1"],
"run_id": AnyStr(),
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_stream",
"run_id": AnyStr(),
"name": "my_task",
"tags": ["seq:step:1"],
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"data": {"chunk": 2},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_end",
"data": {"output": 2, "input": {"number": 1}},
"run_id": AnyStr(),
"name": "my_task",
"tags": ["seq:step:1"],
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_task",
"langgraph_triggers": ("__pregel_push",),
"langgraph_path": (
"__pregel_push",
("__pregel_pull", "my_workflow"),
2,
True,
),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [
AnyStr(),
AnyStr(),
],
},
{
"event": "on_chain_stream",
"run_id": AnyStr(),
"name": "my_workflow",
"tags": ["graph:step:4"],
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_workflow",
"langgraph_triggers": ("__start__",),
"langgraph_path": ("__pregel_pull", "my_workflow"),
"langgraph_checkpoint_ns": AnyStr(),
},
"data": {"chunk": "done"},
"parent_ids": [AnyStr()],
},
{
"event": "on_chain_stream",
"run_id": AnyStr(),
"name": "LangGraph",
"tags": [],
"metadata": {"thread_id": "1"},
"data": {"chunk": {"my_task": 2}},
"parent_ids": [],
},
{
"event": "on_chain_end",
"data": {"output": "done", "input": 1},
"run_id": AnyStr(),
"name": "my_workflow",
"tags": ["graph:step:4"],
"metadata": {
"thread_id": "1",
"langgraph_step": 4,
"langgraph_node": "my_workflow",
"langgraph_triggers": ("__start__",),
"langgraph_path": ("__pregel_pull", "my_workflow"),
"langgraph_checkpoint_ns": AnyStr(),
},
"parent_ids": [AnyStr()],
},
{
"event": "on_chain_stream",
"run_id": AnyStr(),
"name": "LangGraph",
"tags": [],
"metadata": {"thread_id": "1"},
"data": {"chunk": {"my_workflow": "done"}},
"parent_ids": [],
},
{
"event": "on_chain_end",
"data": {"output": "done"},
"run_id": AnyStr(),
"name": "LangGraph",
"tags": [],
"metadata": {"thread_id": "1"},
"parent_ids": [],
},
]
libs/langgraph/uv.lock
Generated
+1567 -1566
View File
File diff suppressed because it is too large Load Diff
libs/prebuilt/langgraph/prebuilt/chat_agent_executor.py
+10 -5
View File
@@ -240,7 +240,7 @@ def _validate_chat_history(
def create_react_agent(
model: Union[str, LanguageModelLike],
tools: Union[Sequence[Union[BaseTool, Callable]], ToolNode],
tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
*,
prompt: Optional[Prompt] = None,
response_format: Optional[
@@ -420,12 +420,13 @@ def create_react_agent(
else AgentState
)
llm_builtin_tools: list[dict] = []
if isinstance(tools, ToolNode):
tool_classes = list(tools.tools_by_name.values())
tool_node = tools
else:
tool_node = ToolNode(tools)
# get the tool functions wrapped in a tool class from the ToolNode
llm_builtin_tools = [t for t in tools if isinstance(t, dict)]
tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
tool_classes = list(tool_node.tools_by_name.values())
if isinstance(model, str):
@@ -442,8 +443,12 @@ def create_react_agent(
tool_calling_enabled = len(tool_classes) > 0
if _should_bind_tools(model, tool_classes) and tool_calling_enabled:
model = cast(BaseChatModel, model).bind_tools(tool_classes)
if (
_should_bind_tools(model, tool_classes)
and len(tool_classes) > 0
or (len(llm_builtin_tools) > 0)
):
model = cast(BaseChatModel, model).bind_tools(tool_classes + llm_builtin_tools) # type: ignore[operator]
model_runnable = _get_prompt_runnable(prompt) | model
libs/prebuilt/langgraph/prebuilt/interrupt.py
+1 -162
View File
@@ -1,12 +1,7 @@
from copy import deepcopy
from typing import Any, Literal, Optional, Union, cast
from typing import Literal, Optional, Union
from langchain_core.messages import ToolCall, ToolMessage
from typing_extensions import TypedDict
from langgraph.types import Command, interrupt
from langgraph.utils.runnable import RunnableCallable
class HumanInterruptConfig(TypedDict):
"""Configuration that defines what actions are allowed for a human interrupt.
@@ -93,159 +88,3 @@ class HumanResponse(TypedDict):
type: Literal["accept", "ignore", "response", "edit"]
args: Union[None, str, ActionRequest]
class InterruptToolNode(RunnableCallable):
"""Prebuilt post model hook node used to enable common patterns for tool interrupts.
For any tools with specified policies, an interrupt will be raised when the LLM returns
a tool call for said tool. The interrupt policy will be used to determine what sort of resume logic is allowed.
Any of the following resume patterns are supported:
* accept: the tool call is executed as planned
* edit: the args for the tool call are edited and then the tool call is executed
* response: text response/feedback is fed back into the LLM
* ignore: the current tool call is ignored / skipped
Args:
**interrupt_policy: a mapping of tool names to [`HumanInterruptConfig`][prebuilt.interrupt.HumanInterruptConfig] dictionaries
specifying which interrupt patterns to enable for said tool.
Example:
```python
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt.interrupt import HumanInterruptConfig, InterruptToolNode
from langgraph.types import Command
def book_hotel(hotel_name: str) -> str:
'''Book a room at the provided hotel.'''
# Some hotel API calls, a sensitive / expensive operation
return f"Booked a hotel at {hotel_name}."
agent = create_react_agent(
"openai:gpt-4.1",
tools=[book_hotel],
prompt="You are a hotel booking assistant.",
post_model_hook=InterruptToolNode(
book_hotel=HumanInterruptConfig(
allow_accept=True,
allow_edit=True,
allow_ignore=True,
allow_respond=True,
)
),
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": 1}}
response = agent.invoke(
{"messages": [{"role": "user", "content": "please book a hotel at the hilton inn in boston."}]},
config=config,
)
response = agent.invoke(Command(resume={"type": "accept"}), config=config)
```
"""
def __init__(self, **interrupt_policy: HumanInterruptConfig):
super().__init__(self._func, self._afunc)
self.interrupt_policy = interrupt_policy
def _interrupt(
self,
tool_call: ToolCall,
interrupt_config: HumanInterruptConfig,
) -> Union[ToolCall, ToolMessage]:
"""Interrupt before a tool call and ask for human input."""
call_id = tool_call["id"]
tool_name = tool_call["name"]
request = HumanInterrupt(
action_request=ActionRequest(
action=tool_name,
args=tool_call["args"],
),
config=interrupt_config,
description=f"Please review tool call for `{tool_name}` before execution.",
)
response = interrupt([request])
# resume provided by agent inbox as a list
response = response[0] if isinstance(response, list) else response
try:
response_type = response.get("type")
except AttributeError:
raise TypeError(
f"Unexpected resume value: {response}."
f"Expected a dict with `'type'` key."
)
if response_type == "accept" and interrupt_config["allow_accept"]:
return tool_call
elif response_type == "edit" and interrupt_config["allow_edit"]:
return ToolCall(
args=cast(ActionRequest, response)["args"]["args"],
name=tool_name,
id=call_id,
type="tool_call",
)
elif response_type == "response" and interrupt_config["allow_respond"]:
return ToolMessage(
content=cast(str, response["args"]),
name=tool_name,
tool_call_id=call_id,
status="error",
)
elif response_type == "ignore" and interrupt_config["allow_ignore"]:
return ToolMessage(
content=f"User ignored the tool call for `{tool_name}` with id {call_id}",
name=tool_name,
tool_call_id=call_id,
status="success",
)
allowed_types = [
type_name
for type_name, is_allowed in {
"accept": interrupt_config["allow_accept"],
"edit": interrupt_config["allow_edit"],
"response": interrupt_config["allow_respond"],
"ignore": interrupt_config["allow_ignore"],
}.items()
if is_allowed
]
raise ValueError(
f"Unexpected human response: {response}. "
f"Expected one with `'type'` in {allowed_types} based on {tool_name}'s interrupt configuration."
)
def _func(self, input: dict[str, Any]) -> Command:
ai_msg = input["messages"][-1]
tool_calls: list[ToolCall] = deepcopy(ai_msg.tool_calls) or []
tool_messages: list[ToolMessage] = []
for idx, tool_call in enumerate(tool_calls):
if interrupt_config := self.interrupt_policy.get(tool_call["name"]):
interrupt_result = self._interrupt(
tool_call=tool_call, interrupt_config=interrupt_config
)
if isinstance(interrupt_result, ToolMessage):
tool_messages.append(interrupt_result)
else:
tool_calls[idx] = interrupt_result
updated_ai_msg = ai_msg.copy(update={"tool_calls": tool_calls})
# conditional routing logic for post_model_hook will direct to the tools node
# or agent node depending on if there are pending tool calls
return {"messages": [updated_ai_msg, *tool_messages]}
async def _afunc(self, input: dict[str, Any]) -> Command:
return self._func(input)
libs/prebuilt/pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-prebuilt"
version = "0.1.8"
version = "0.2.1"
description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
authors = []
requires-python = ">=3.9"
libs/prebuilt/tests/test_interrupt_tool_node.py
-191
View File
@@ -1,191 +0,0 @@
import pytest
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.interrupt import HumanInterruptConfig, InterruptToolNode
from langgraph.types import Command
from tests.model import FakeToolCallingModel
def hello_tool(name: str) -> str:
"""Return a greeting for the provided person."""
return f"Hello, {name}!"
post_model_hook = InterruptToolNode(
hello_tool=HumanInterruptConfig(
allow_accept=True,
allow_edit=True,
allow_ignore=True,
allow_respond=True,
)
)
default_model = FakeToolCallingModel(
tool_calls=[
[
{
"name": "hello_tool",
"args": {"name": "lady gaga"},
"id": "some-random-id",
}
]
]
)
def test_interrupt_surfaced(
request: pytest.FixtureRequest,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
agent = create_react_agent(
default_model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
result = agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
interrupt_data = result["__interrupt__"]
assert interrupt_data[0].value == [
{
"action_request": {"action": "hello_tool", "args": {"name": "lady gaga"}},
"config": {
"allow_accept": True,
"allow_edit": True,
"allow_ignore": True,
"allow_respond": True,
},
"description": "Please review tool call for `hello_tool` before execution.",
}
]
response = agent.invoke(Command(resume={"type": "accept"}), config=config)
tool_message: ToolMessage = response["messages"][-2]
assert tool_message.content == "Hello, lady gaga!"
assert tool_message.name == "hello_tool"
@pytest.mark.parametrize(
"resume, expected_content",
[
({"type": "accept"}, "Hello, lady gaga!"),
(
{"type": "ignore"},
"User ignored the tool call for `hello_tool` with id some-random-id",
),
(
{
"type": "edit",
"args": {"action": "hello_tool", "args": {"name": "bruno mars"}},
},
"Hello, bruno mars!",
),
],
)
def test_interrupt_resume_variants(
request: pytest.FixtureRequest,
sync_checkpointer: BaseCheckpointSaver,
resume: dict,
expected_content: str,
) -> None:
agent = create_react_agent(
default_model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
response = agent.invoke(Command(resume=resume), config=config)
tool_message: ToolMessage = response["messages"][-2]
assert tool_message.name == "hello_tool"
assert tool_message.content == expected_content
if resume["type"] == "edit":
ai_msg = response["messages"][-1]
assert ai_msg.tool_calls == [
{
"name": "hello_tool",
"args": {"name": "lady gaga"},
"id": "some-random-id",
"type": "tool_call",
}
]
def test_resume_with_response(
request: pytest.FixtureRequest,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
model = FakeToolCallingModel(
tool_calls=[
[
{
"name": "hello_tool",
"args": {"name": "lady gaga"},
"id": "some-random-id",
}
],
[
{
"name": "hello_tool",
"args": {"name": "bruno mars"},
"id": "some-random-id-2",
}
],
]
)
agent = create_react_agent(
model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
# Provide user response
agent.invoke(
Command(
resume={
"type": "response",
"args": "actually, please say hello to bruno mars",
}
),
config=config,
)
# Accept the updated call
response = agent.invoke(Command(resume={"type": "accept"}), config=config)
assert len(response["messages"]) == 6
tool_message: ToolMessage = response["messages"][-2]
assert tool_message.name == "hello_tool"
assert tool_message.content == "Hello, bruno mars!"
def test_resume_with_type_not_allowed(sync_checkpointer: BaseCheckpointSaver) -> None:
agent = create_react_agent(
default_model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
with pytest.raises(ValueError) as exc_info:
agent.invoke(Command(resume={"type": "not-allowed"}), config=config)
assert (
str(exc_info.value)
== "Unexpected human response: {'type': 'not-allowed'}. Expected one with `'type'` in ['accept', 'edit', 'response', 'ignore'] based on hello_tool's interrupt configuration."
)
libs/prebuilt/uv.lock
Generated
+742 -743
View File
File diff suppressed because it is too large Load Diff
libs/scheduler-kafka/uv.lock
Generated
+837 -838
View File
File diff suppressed because it is too large Load Diff
libs/sdk-js/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@langchain/langgraph-sdk",
"version": "0.0.77",
"version": "0.0.78",
"description": "Client library for interacting with the LangGraph API",
"type": "module",
"packageManager": "yarn@1.22.19",
libs/sdk-js/src/react/stream.tsx
+190 -70
View File
@@ -457,6 +457,11 @@ export interface UseStreamOptions<
*/
onFinish?: (state: ThreadState<StateType>) => void;
/**
* Callback that is called when a new stream is created.
*/
onCreated?: (run: { run_id: string; thread_id: string }) => void;
/**
* Callback that is called when an update event is received.
*/
@@ -502,6 +507,15 @@ export interface UseStreamOptions<
* Callback that is called when the thread ID is updated (ie when a new thread is created).
*/
onThreadId?: (threadId: string) => void;
/** Will reconnect the stream on mount */
reconnectOnMount?: boolean | (() => RunMetadataStorage);
}
interface RunMetadataStorage {
getItem(key: `lg:stream:${string}`): string | null;
setItem(key: `lg:stream:${string}`, value: string): void;
removeItem(key: `lg:stream:${string}`): void;
}
export interface UseStream<
@@ -590,6 +604,11 @@ export interface UseStream<
* The ID of the assistant to use.
*/
assistantId: string;
/**
* Join an active stream.
*/
joinStream: (runId: string) => Promise<void>;
}
type ConfigWithConfigurable<ConfigurableType extends Record<string, unknown>> =
@@ -619,6 +638,7 @@ interface SubmitOptions<
* @default false
*/
streamSubgraphs?: boolean;
streamResumable?: boolean;
}
export function useStream<
@@ -647,7 +667,17 @@ export function useStream<
| ErrorStreamEvent
| FeedbackStreamEvent;
let { assistantId, messagesKey, onError, onFinish } = options;
let { assistantId, messagesKey, onCreated, onError, onFinish } = options;
const reconnectOnMountRef = useRef(options.reconnectOnMount);
const runMetadataStorage = useMemo(() => {
if (typeof window === "undefined") return null;
const storage = reconnectOnMountRef.current;
if (storage === true) return window.sessionStorage;
if (typeof storage === "function") return storage();
return null;
}, []);
messagesKey ??= "messages";
const client = useMemo(
@@ -722,6 +752,7 @@ export function useStream<
// TODO: this should be done on the server to avoid pagination
// TODO: should we permit adapter? SWR / React Query?
// TODO: make this only when branching is expected
const history = useThreadHistory<StateType>(
threadId,
client,
@@ -800,15 +831,23 @@ export function useStream<
);
})();
const stop = useCallback(() => {
const stop = () => {
if (abortRef.current != null) abortRef.current.abort();
abortRef.current = null;
}, []);
const submit = async (
values: UpdateType | null | undefined,
submitOptions?: SubmitOptions<StateType, ConfigurableType>,
) => {
if (runMetadataStorage && threadId) {
const runId = runMetadataStorage.getItem(`lg:stream:${threadId}`);
if (runId) client.runs.cancel(threadId, runId);
runMetadataStorage.removeItem(`lg:stream:${threadId}`);
}
};
async function consumeStream(
action: (signal: AbortSignal) => Promise<{
onSuccess: () => Promise<ThreadState<StateType>[]>;
stream: AsyncGenerator<EventStreamEvent>;
}>,
) {
try {
setIsLoading(true);
setStreamError(undefined);
@@ -816,69 +855,10 @@ export function useStream<
submittingRef.current = true;
abortRef.current = new AbortController();
// Unbranch things
const newPath = submitOptions?.checkpoint?.checkpoint_id
? branchByCheckpoint[submitOptions?.checkpoint?.checkpoint_id]?.branch
: undefined;
if (newPath != null) setBranch(newPath ?? "");
// Assumption: we're setting the initial value
// Used for instant feedback
setStreamValues(() => {
const values = { ...historyValues };
if (submitOptions?.optimisticValues != null) {
return {
...values,
...(typeof submitOptions.optimisticValues === "function"
? submitOptions.optimisticValues(values)
: submitOptions.optimisticValues),
};
}
return values;
});
let usableThreadId = threadId;
if (!usableThreadId) {
const thread = await client.threads.create();
onThreadId(thread.thread_id);
usableThreadId = thread.thread_id;
}
const streamMode = unique([
...(submitOptions?.streamMode ?? []),
...trackStreamModeRef.current,
...callbackStreamMode,
]);
const checkpoint =
submitOptions?.checkpoint ?? threadHead?.checkpoint ?? undefined;
// @ts-expect-error
if (checkpoint != null) delete checkpoint.thread_id;
const run = client.runs.stream(usableThreadId, assistantId, {
input: values as Record<string, unknown>,
config: submitOptions?.config,
command: submitOptions?.command,
interruptBefore: submitOptions?.interruptBefore,
interruptAfter: submitOptions?.interruptAfter,
metadata: submitOptions?.metadata,
multitaskStrategy: submitOptions?.multitaskStrategy,
onCompletion: submitOptions?.onCompletion,
onDisconnect: submitOptions?.onDisconnect ?? "cancel",
signal: abortRef.current.signal,
checkpoint,
streamMode,
streamSubgraphs: submitOptions?.streamSubgraphs,
}) as AsyncGenerator<EventStreamEvent>;
const run = await action(abortRef.current.signal);
let streamError: StreamError | undefined;
for await (const { event, data } of run) {
for await (const { event, data } of run.stream) {
if (event === "error") {
streamError = new StreamError(data);
break;
@@ -930,9 +910,9 @@ export function useStream<
}
// TODO: stream created checkpoints to avoid an unnecessary network request
const result = await history.mutate(usableThreadId);
setStreamValues(null);
const result = await run.onSuccess();
setStreamValues(null);
if (streamError != null) throw streamError;
const lastHead = result.at(0);
@@ -956,8 +936,146 @@ export function useStream<
submittingRef.current = false;
abortRef.current = null;
}
}
const joinStream = async (runId: string, lastEventId?: string) => {
lastEventId ??= "-1";
if (!threadId) return;
await consumeStream(async (signal: AbortSignal) => {
const stream = client.runs.joinStream(threadId, runId, {
signal,
lastEventId,
}) as AsyncGenerator<EventStreamEvent>;
return {
onSuccess: () => {
runMetadataStorage?.removeItem(`lg:stream:${threadId}`);
return history.mutate(threadId);
},
stream,
};
});
};
const submit = async (
values: UpdateType | null | undefined,
submitOptions?: SubmitOptions<StateType, ConfigurableType>,
) => {
await consumeStream(async (signal: AbortSignal) => {
// Unbranch things
const newPath = submitOptions?.checkpoint?.checkpoint_id
? branchByCheckpoint[submitOptions?.checkpoint?.checkpoint_id]?.branch
: undefined;
if (newPath != null) setBranch(newPath ?? "");
// Assumption: we're setting the initial value
// Used for instant feedback
setStreamValues(() => {
const values = { ...historyValues };
if (submitOptions?.optimisticValues != null) {
return {
...values,
...(typeof submitOptions.optimisticValues === "function"
? submitOptions.optimisticValues(values)
: submitOptions.optimisticValues),
};
}
return values;
});
let usableThreadId = threadId;
if (!usableThreadId) {
const thread = await client.threads.create();
onThreadId(thread.thread_id);
usableThreadId = thread.thread_id;
}
const streamMode = unique([
...(submitOptions?.streamMode ?? []),
...trackStreamModeRef.current,
...callbackStreamMode,
]);
const checkpoint =
submitOptions?.checkpoint ?? threadHead?.checkpoint ?? undefined;
// @ts-expect-error
if (checkpoint != null) delete checkpoint.thread_id;
let rejoinKey: `lg:stream:${string}` | undefined;
const stream = client.runs.stream(usableThreadId, assistantId, {
input: values as Record<string, unknown>,
config: submitOptions?.config,
command: submitOptions?.command,
interruptBefore: submitOptions?.interruptBefore,
interruptAfter: submitOptions?.interruptAfter,
metadata: submitOptions?.metadata,
multitaskStrategy: submitOptions?.multitaskStrategy,
onCompletion: submitOptions?.onCompletion,
onDisconnect:
submitOptions?.onDisconnect ??
(runMetadataStorage ? "continue" : "cancel"),
signal,
checkpoint,
streamMode,
streamSubgraphs: submitOptions?.streamSubgraphs,
streamResumable: submitOptions?.streamResumable ?? !!runMetadataStorage,
onRunCreated(params) {
const runParams = {
run_id: params.run_id,
thread_id: params.thread_id ?? usableThreadId,
};
if (runMetadataStorage) {
rejoinKey = `lg:stream:${runParams.thread_id}`;
runMetadataStorage.setItem(rejoinKey, runParams.run_id);
}
onCreated?.(runParams);
},
}) as AsyncGenerator<EventStreamEvent>;
return {
stream,
onSuccess: () => {
if (rejoinKey) runMetadataStorage?.removeItem(rejoinKey);
return history.mutate(usableThreadId);
},
};
});
};
const reconnectKey = useMemo(() => {
if (!runMetadataStorage || isLoading) return undefined;
if (typeof window === "undefined") return undefined;
const runId = runMetadataStorage?.getItem(`lg:stream:${threadId}`);
if (!runId) return undefined;
return { runId, threadId };
}, [runMetadataStorage, isLoading, threadId]);
const shouldReconnect = !!runMetadataStorage;
const reconnectRef = useRef({ threadId, shouldReconnect });
const joinStreamRef = useRef<typeof joinStream>(joinStream);
joinStreamRef.current = joinStream;
useEffect(() => {
// reset shouldReconnect when switching threads
if (reconnectRef.current.threadId !== threadId) {
reconnectRef.current = { threadId, shouldReconnect };
}
}, [threadId, shouldReconnect]);
useEffect(() => {
if (reconnectKey && reconnectRef.current.shouldReconnect) {
reconnectRef.current.shouldReconnect = false;
joinStreamRef.current?.(reconnectKey.runId);
}
}, [reconnectKey]);
const error = streamError ?? historyError;
const values = streamValues ?? historyValues;
@@ -976,6 +1094,8 @@ export function useStream<
stop,
submit,
joinStream,
branch,
setBranch,