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

Merge branch 'main' into sr/better-interrupts

Browse Source
This commit is contained in:
Sydney Runkle
2025-04-23 15:58:57 -07:00
committed by GitHub
parent 8fe7c8101f 0ff913c50b
commit f0fb8d9187
25 changed files with 553 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs/agents/agents.md
+10 -6
View File
@@ -29,7 +29,9 @@ agent = create_react_agent(
)
# Run the agent
agent.invoke({"messages": "what is the weather in sf"})
agent.invoke(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)
```
1. Define a tool for the agent to use. Tools can be defined as vanilla Python functions. For more advanced tool usage and customization, check the [tools](./tools.md) page.
@@ -85,7 +87,7 @@ agent = create_react_agent(
)
agent.invoke(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)
```
@@ -113,7 +115,7 @@ agent = create_react_agent(
)
agent.invoke(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
config={"configurable": {"user_name": "John Smith"}}
)
@@ -150,12 +152,12 @@ agent = create_react_agent(
# highlight-next-line
config = {"configurable": {"thread_id": "1"}}
sf_response = agent.invoke(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
config # (2)!
)
ny_response = agent.invoke(
{"messages": "what about new york?"},
{"messages": [{"role": "user", "content": "what about new york?"}]},
# highlight-next-line
config
)
@@ -189,7 +191,9 @@ agent = create_react_agent(
response_format=WeatherResponse # (1)!
)
response = agent.invoke({"messages": "what is the weather in sf"})
response = agent.invoke(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]}
)
# highlight-next-line
response["structured_response"]
docs/docs/agents/context.md
+3 -3
View File
@@ -36,7 +36,7 @@ for this purpose:
```python
agent.invoke(
{"messages": "hi!"},
{"messages": [{"role": "user", "content": "hi!"}]},
# highlight-next-line
config={"configurable": {"user_id": "user_123"}}
)
@@ -183,7 +183,7 @@ Tools can access context through special parameter **annotations**.
)
agent.invoke(
{"messages": "look up user information"},
{"messages": [{"role": "user", "content": "look up user information"}]},
# highlight-next-line
config={"configurable": {"user_id": "user_123"}}
)
@@ -278,7 +278,7 @@ agent = create_react_agent(
)
agent.invoke(
{"messages": "greet the user"},
{"messages": [{"role": "user", "content": "greet the user"}]},
# highlight-next-line
config={"configurable": {"user_id": "user_123"}}
)
docs/docs/agents/human-in-the-loop.md
+2 -2
View File
@@ -70,7 +70,7 @@ config = {
}
for chunk in agent.stream(
{"messages": "book a stay at McKittrick hotel"},
{"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
# highlight-next-line
config
):
@@ -194,7 +194,7 @@ config = {"configurable": {"thread_id": "1"}}
# Run the agent
for chunk in agent.stream(
{"messages": "book a stay at McKittrick hotel"},
{"messages": [{"role": "user", "content": "book a stay at McKittrick hotel"}]},
# highlight-next-line
config
):
docs/docs/agents/mcp.md
+6 -2
View File
@@ -40,8 +40,12 @@ async with MultiServerMCPClient(
# highlight-next-line
client.get_tools()
)
math_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
weather_response = await agent.ainvoke({"messages": "what is the weather in nyc?"})
math_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
)
weather_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
)
```
## Custom MCP servers
docs/docs/agents/memory.md
+5 -5
View File
@@ -59,14 +59,14 @@ config = {
}
sf_response = agent.invoke(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
config
)
# Continue the conversation using the same thread_id
ny_response = agent.invoke(
{"messages": "what about new york?"},
{"messages": [{"role": "user", "content": "what about new york?"}]},
# highlight-next-line
config # (4)!
)
@@ -188,7 +188,7 @@ agent = create_react_agent(
# Run the agent
agent.invoke(
{"messages": "look up user information"},
{"messages": [{"role": "user", "content": "look up user information"}]},
# highlight-next-line
config={"configurable": {"user_id": "user_123"}}
)
@@ -206,7 +206,7 @@ agent.invoke(
### Writing
```python title="Example of a tool that updates user information"
from typing import TypedDict
from typing_extensions import TypedDict
from langgraph.config import get_store
from langgraph.prebuilt import create_react_agent
@@ -236,7 +236,7 @@ agent = create_react_agent(
# Run the agent
agent.invoke(
{"messages": "My name is John Smith"},
{"messages": [{"role": "user", "content": "My name is John Smith"}]},
# highlight-next-line
config={"configurable": {"user_id": "user_123"}} # (6)!
)
docs/docs/agents/multi-agent.md
+34 -10
View File
@@ -53,12 +53,22 @@ hotel_assistant = create_react_agent(
supervisor = create_supervisor(
agents=[flight_assistant, hotel_assistant],
model=ChatOpenAI(model="gpt-4o"),
prompt="You manage a hotel booking assistant and a flight booking assistant. Assign work to them."
prompt=(
"You manage a hotel booking assistant and a"
"flight booking assistant. Assign work to them."
)
).compile()
for chunk in supervisor.stream({
"messages": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}):
for chunk in supervisor.stream(
{
"messages": [
{
"role": "user",
"content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}
]
}
):
print(chunk)
print("\n")
```
@@ -110,9 +120,16 @@ swarm = create_swarm(
default_active_agent="flight_assistant"
).compile()
for chunk in supervisor.stream({
"messages": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}):
for chunk in swarm.stream(
{
"messages": [
{
"role": "user",
"content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}
]
}
):
print(chunk)
print("\n")
```
@@ -253,9 +270,16 @@ multi_agent_graph = (
)
# Run the multi-agent graph
for chunk in multi_agent_graph.stream({
"messages": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}):
for chunk in multi_agent_graph.stream(
{
"messages": [
{
"role": "user",
"content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}
]
}
):
print(chunk)
print("\n")
```
docs/docs/agents/run_agents.md
+6 -6
View File
@@ -18,7 +18,7 @@ Agents can be executed in two primary modes:
agent = create_react_agent(...)
# highlight-next-line
response = agent.invoke({"messages": "what is the weather in sf"})
response = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})
```
=== "Async invocation"
@@ -27,7 +27,7 @@ Agents can be executed in two primary modes:
agent = create_react_agent(...)
# highlight-next-line
response = await agent.ainvoke({"messages": "what is the weather in sf"})
response = await agent.ainvoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})
```
## Inputs and outputs
@@ -82,7 +82,7 @@ Streaming is available in both sync and async modes:
```python
for chunk in agent.stream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="updates"
):
print(chunk)
@@ -92,7 +92,7 @@ Streaming is available in both sync and async modes:
```python
async for chunk in agent.astream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="updates"
):
print(chunk)
@@ -122,7 +122,7 @@ To control agent execution and avoid infinite loops, set a recursion limit. This
try:
response = agent.invoke(
{"messages": "what's the weather in sf"},
{"messages": [{"role": "user", "content": "what's the weather in sf"}]},
# highlight-next-line
{"recursion_limit": recursion_limit},
)
@@ -148,7 +148,7 @@ To control agent execution and avoid infinite loops, set a recursion limit. This
try:
response = agent_with_recursion_limit.invoke(
{"messages": "what's the weather in sf"},
{"messages": [{"role": "user", "content": "what's the weather in sf"}]},
)
except GraphRecursionError:
print("Agent stopped due to max iterations.")
docs/docs/agents/streaming.md
+8 -8
View File
@@ -35,7 +35,7 @@ For example, if you have an agent that calls a tool once, you should see the fol
)
# highlight-next-line
for chunk in agent.stream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode="updates"
):
@@ -52,7 +52,7 @@ For example, if you have an agent that calls a tool once, you should see the fol
)
# highlight-next-line
async for chunk in agent.astream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode="updates"
):
@@ -73,7 +73,7 @@ To stream tokens as they are produced by the LLM, use `stream_mode="messages"`:
)
# highlight-next-line
for token, metadata in agent.stream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode="messages"
):
@@ -91,7 +91,7 @@ To stream tokens as they are produced by the LLM, use `stream_mode="messages"`:
)
# highlight-next-line
async for token, metadata in agent.astream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode="messages"
):
@@ -125,7 +125,7 @@ To stream updates from tools as they are executed, you can use [get_stream_write
)
for chunk in agent.stream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode="custom"
):
@@ -154,7 +154,7 @@ To stream updates from tools as they are executed, you can use [get_stream_write
)
async for chunk in agent.astream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode="custom"
):
@@ -178,7 +178,7 @@ You can specify multiple streaming modes by passing stream mode as a list: `stre
)
for stream_mode, chunk in agent.stream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode=["updates", "messages", "custom"]
):
@@ -195,7 +195,7 @@ You can specify multiple streaming modes by passing stream mode as a list: `stre
)
async for stream_mode, chunk in agent.astream(
{"messages": "what is the weather in sf"},
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
# highlight-next-line
stream_mode=["updates", "messages", "custom"]
):
docs/docs/agents/tools.md
+18 -6
View File
@@ -116,7 +116,9 @@ agent = create_react_agent(
tools=tools
)
agent.invoke({"messages": "what's 3 + 5 and 4 * 7? make both calculations in parallel"})
agent.invoke(
{"messages": [{"role": "user", "content": "what's 3 + 5 and 4 * 7?"}]}
)
```
## Return tool results directly
@@ -137,7 +139,9 @@ agent = create_react_agent(
tools=[add]
)
agent.invoke({"messages": "what's 3 + 5?"})
agent.invoke(
{"messages": [{"role": "user", "content": "what's 3 + 5?"}]}
)
```
## Force tool use
@@ -161,7 +165,9 @@ agent = create_react_agent(
tools=tools
)
agent.invoke({"messages": "Hi, I am Bob"})
agent.invoke(
{"messages": [{"role": "user", "content": "Hi, I am Bob"}]}
)
```
!!! Warning "Avoid infinite loops"
@@ -191,7 +197,9 @@ By default, the agent will catch all exceptions raised during tool calls and wil
model="anthropic:claude-3-7-sonnet-latest",
tools=[multiply]
)
agent.invoke({"messages": "what's 42 x 7?"})
agent.invoke(
{"messages": [{"role": "user", "content": "what's 42 x 7?"}]}
)
```
=== "Disable error handling"
@@ -215,7 +223,9 @@ By default, the agent will catch all exceptions raised during tool calls and wil
model="anthropic:claude-3-7-sonnet-latest",
tools=tool_node
)
agent_no_error_handling.invoke({"messages": "what's 42 x 7?"})
agent_no_error_handling.invoke(
{"messages": [{"role": "user", "content": "what's 42 x 7?"}]}
)
```
1. This disables error handling (enabled by default). See all available strategies in the [API reference][langgraph.prebuilt.tool_node.ToolNode].
@@ -243,7 +253,9 @@ By default, the agent will catch all exceptions raised during tool calls and wil
model="anthropic:claude-3-7-sonnet-latest",
tools=tool_node
)
agent_custom_error_handling.invoke({"messages": "what's 42 x 7?"})
agent_custom_error_handling.invoke(
{"messages": [{"role": "user", "content": "what's 42 x 7?"}]}
)
```
1. This provides a custom message to send to the LLM in case of an exception. See all available strategies in the [API reference][langgraph.prebuilt.tool_node.ToolNode].
docs/docs/cloud/reference/env_var.md
+18
View File
@@ -97,3 +97,21 @@ Database Connectivity:
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 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).
## `REDIS_KEY_PREFIX`
!!! info "Available in API Server version 0.1.9+"
This environment variable is supported in API Server version 0.1.9 and above.
Specify a prefix for Redis keys. This allows multiple LangGraph Server instances to share the same Redis instance by using different key prefixes.
Defaults to `''`.
## `REDIS_CLUSTER`
!!! info "Only Allowed in Self-Hosted Deployments"
Redis Cluster mode is only available in Self-Hosted Deployment models, LangGraph Cloud SaaS will provision a redis instance for you by default.
Set `REDIS_CLUSTER` to `True` to enable Redis Cluster mode. When enabled, the system will connect to Redis using cluster mode. This is useful when connecting to a Redis Cluster deployment.
Defaults to `False`.
libs/cli/langgraph_cli/cli.py
+14
View File
@@ -168,6 +168,13 @@ def cli():
@OPT_DEBUGGER_BASE_URL
@OPT_WATCH
@OPT_POSTGRES_URI
@click.option(
"--image",
type=str,
default=None,
help="Docker image to use for the langgraph-api service. If specified, skips building and uses this image directly."
" Useful if you want to test against an image already built using `langgraph build`.",
)
@click.option(
"--wait",
is_flag=True,
@@ -187,6 +194,7 @@ def up(
debugger_port: Optional[int],
debugger_base_url: Optional[str],
postgres_uri: Optional[str],
image: Optional[str],
):
click.secho("Starting LangGraph API server...", fg="green")
click.secho(
@@ -207,6 +215,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
debugger_port=debugger_port,
debugger_base_url=debugger_base_url,
postgres_uri=postgres_uri,
image=image,
)
# add up + options
args.extend(["up", "--remove-orphans"])
@@ -692,6 +701,7 @@ def prepare_args_and_stdin(
debugger_port: Optional[int] = None,
debugger_base_url: Optional[str] = None,
postgres_uri: Optional[str] = None,
image: Optional[str] = None,
) -> Tuple[List[str], str]:
assert config_path.exists(), f"Config file not found: {config_path}"
# prepare args
@@ -701,6 +711,7 @@ def prepare_args_and_stdin(
debugger_port=debugger_port,
debugger_base_url=debugger_base_url,
postgres_uri=postgres_uri,
image=image, # Pass image to compose YAML generator
)
args = [
"--project-directory",
@@ -716,6 +727,7 @@ def prepare_args_and_stdin(
config,
watch=watch,
base_image=langgraph_cli.config.default_base_image(config),
image=image,
)
return args, stdin
@@ -733,6 +745,7 @@ def prepare(
debugger_port: Optional[int] = None,
debugger_base_url: Optional[str] = None,
postgres_uri: Optional[str] = None,
image: Optional[str] = None,
) -> Tuple[List[str], str]:
"""Prepare the arguments and stdin for running the LangGraph API server."""
config_json = langgraph_cli.config.validate_config_file(config_path)
@@ -757,5 +770,6 @@ def prepare(
debugger_port=debugger_port,
debugger_base_url=debugger_base_url or f"http://127.0.0.1:{port}",
postgres_uri=postgres_uri,
image=image,
)
return args, stdin
libs/cli/langgraph_cli/config.py
+18 -8
View File
@@ -1288,6 +1288,7 @@ def config_to_compose(
config_path: pathlib.Path,
config: Config,
base_image: Optional[str] = None,
image: Optional[str] = None,
watch: bool = False,
) -> str:
base_image = base_image or default_base_image(config)
@@ -1314,19 +1315,28 @@ def config_to_compose(
"""
else:
watch_str = ""
if image:
return f"""
{textwrap.indent(env_vars_str, " ")}
{env_file_str}
{watch_str}
"""
dockerfile, additional_contexts = config_to_docker(config_path, config, base_image)
else:
dockerfile, additional_contexts = config_to_docker(
config_path, config, base_image
)
additional_contexts_str = "\n".join(
f" - {name}: {path}"
for name, path in additional_contexts.items()
)
if additional_contexts_str:
additional_contexts_str = f"""
additional_contexts_str = "\n".join(
f" - {name}: {path}"
for name, path in additional_contexts.items()
)
if additional_contexts_str:
additional_contexts_str = f"""
additional_contexts:
{additional_contexts_str}"""
return f"""
return f"""
{textwrap.indent(env_vars_str, " ")}
{env_file_str}
pull_policy: build
libs/cli/langgraph_cli/docker.py
+6
View File
@@ -143,6 +143,8 @@ def compose_as_dict(
debugger_base_url: Optional[str] = None,
# postgres://user:password@host:port/database?option=value
postgres_uri: Optional[str] = None,
# If you are running against an already-built image, you can pass it here
image: Optional[str] = None,
) -> dict:
"""Create a docker compose file as a dictionary in YML style."""
if postgres_uri is None:
@@ -211,6 +213,8 @@ def compose_as_dict(
"POSTGRES_URI": postgres_uri,
},
}
if image:
services["langgraph-api"]["image"] = image
# If Postgres is included, add it to the dependencies of langgraph-api
if include_db:
@@ -244,6 +248,7 @@ def compose(
debugger_base_url: Optional[str] = None,
# postgres://user:password@host:port/database?option=value
postgres_uri: Optional[str] = None,
image: Optional[str] = None,
) -> str:
"""Create a docker compose file as a string."""
compose_content = compose_as_dict(
@@ -252,6 +257,7 @@ def compose(
debugger_port=debugger_port,
debugger_base_url=debugger_base_url,
postgres_uri=postgres_uri,
image=image,
)
compose_str = dict_to_yaml(compose_content)
return compose_str
libs/cli/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-cli"
version = "0.2.6"
version = "0.2.7"
description = "CLI for interacting with LangGraph API"
authors = []
license = "MIT"
libs/cli/tests/unit_tests/cli/test_cli.py
+104
View File
@@ -160,6 +160,110 @@ services:
assert clean_empty_lines(actual_stdin) == expected_stdin
def test_prepare_args_and_stdin_with_image() -> None:
# this basically serves as an end-to-end test for using config and docker helpers
config_path = pathlib.Path(__file__).parent / "langgraph.json"
config = validate_config(
Config(dependencies=[".", "../../.."], graphs={"agent": "agent.py:graph"})
)
port = 8000
debugger_port = 8001
debugger_graph_url = f"http://127.0.0.1:{port}"
actual_args, actual_stdin = prepare_args_and_stdin(
capabilities=DEFAULT_DOCKER_CAPABILITIES,
config_path=config_path,
config=config,
docker_compose=pathlib.Path("custom-docker-compose.yml"),
port=port,
debugger_port=debugger_port,
debugger_base_url=debugger_graph_url,
watch=True,
image="my-cool-image",
)
expected_args = [
"--project-directory",
str(pathlib.Path(__file__).parent.absolute()),
"-f",
"custom-docker-compose.yml",
"-f",
"-",
]
expected_stdin = f"""volumes:
langgraph-data:
driver: local
services:
langgraph-redis:
image: redis:6
healthcheck:
test: redis-cli ping
interval: 5s
timeout: 1s
retries: 5
langgraph-postgres:
image: pgvector/pgvector:pg16
ports:
- "5433:5432"
environment:
POSTGRES_DB: postgres
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
command:
- postgres
- -c
- shared_preload_libraries=vector
volumes:
- langgraph-data:/var/lib/postgresql/data
healthcheck:
test: pg_isready -U postgres
start_period: 10s
timeout: 1s
retries: 5
interval: 60s
start_interval: 1s
langgraph-debugger:
image: langchain/langgraph-debugger
restart: on-failure
depends_on:
langgraph-postgres:
condition: service_healthy
ports:
- "{debugger_port}:3968"
environment:
VITE_STUDIO_LOCAL_GRAPH_URL: {debugger_graph_url}
langgraph-api:
ports:
- "8000:8000"
depends_on:
langgraph-redis:
condition: service_healthy
langgraph-postgres:
condition: service_healthy
environment:
REDIS_URI: redis://langgraph-redis:6379
POSTGRES_URI: {DEFAULT_POSTGRES_URI}
image: my-cool-image
healthcheck:
test: python /api/healthcheck.py
interval: 60s
start_interval: 1s
start_period: 10s
develop:
watch:
- path: langgraph.json
action: rebuild
- path: .
action: rebuild
- path: ../../..
action: rebuild\
"""
assert actual_args == expected_args
assert clean_empty_lines(actual_stdin) == expected_stdin
def test_version_option() -> None:
"""Test the --version option of the CLI."""
runner = CliRunner()
libs/langgraph/langgraph/pregel/algo.py
+14 -8
View File
@@ -542,11 +542,14 @@ def prepare_single_task(
str(task_path[2]),
)
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
# we append True to the task path to indicate that a call is being
# made, so we should not return interrupts from this task (responsibility lies with the parent)
task_path = (*task_path[:3], True)
metadata = {
"langgraph_step": step,
"langgraph_node": name,
"langgraph_triggers": triggers,
"langgraph_path": task_path[:3],
"langgraph_path": task_path,
"langgraph_checkpoint_ns": task_checkpoint_ns,
}
if task_id_checksum is not None:
@@ -575,7 +578,7 @@ def prepare_single_task(
local_read,
channels,
managed,
PregelTaskWrites(task_path[:3], name, writes, triggers),
PregelTaskWrites(task_path, name, writes, triggers),
),
CONFIG_KEY_STORE: (store or configurable.get(CONFIG_KEY_STORE)),
CONFIG_KEY_CHECKPOINTER: (
@@ -598,10 +601,10 @@ def prepare_single_task(
call.retry,
None,
task_id,
task_path[:3],
task_path,
)
else:
return PregelTask(task_id, name, task_path[:3])
return PregelTask(task_id, name, task_path)
elif task_path[0] == PUSH:
if len(task_path) == 2:
# SEND tasks, executed in superstep n+1
@@ -637,11 +640,14 @@ def prepare_single_task(
logger.warning(f"Ignoring invalid PUSH task path {task_path}")
return
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
# we append False to the task path to indicate that a call is not being made
# so we should return interrupts from this task
task_path = (*task_path[:3], False)
metadata = {
"langgraph_step": step,
"langgraph_node": packet.node,
"langgraph_triggers": triggers,
"langgraph_path": task_path[:3],
"langgraph_path": task_path,
"langgraph_checkpoint_ns": task_checkpoint_ns,
}
if task_id_checksum is not None:
@@ -678,7 +684,7 @@ def prepare_single_task(
channels,
managed,
PregelTaskWrites(
task_path[:3], packet.node, writes, triggers
task_path, packet.node, writes, triggers
),
),
CONFIG_KEY_STORE: (
@@ -708,12 +714,12 @@ def prepare_single_task(
proc.retry_policy,
None,
task_id,
task_path[:3],
task_path,
writers=proc.flat_writers,
subgraphs=proc.subgraphs,
)
else:
return PregelTask(task_id, packet.node, task_path[:3])
return PregelTask(task_id, packet.node, task_path)
elif task_path[0] == PULL:
# (PULL, node name)
name = cast(str, task_path[1])
libs/langgraph/langgraph/pregel/loop.py
+5
View File
@@ -909,6 +909,11 @@ class PregelLoop(LoopProtocol):
):
return
if writes[0][0] == INTERRUPT:
# in loop.py we append a bool to the PUSH task paths to indicate
# whether or not a call was present (that was popped). If so,
# we don't emit the interrupt as it'll be emitted by the parent
if task.path[0] == PUSH and task.path[-1] is True:
return
interrupts = [
{
INTERRUPT: tuple(
libs/langgraph/langgraph/pregel/messages.py
+39 -33
View File
@@ -46,6 +46,34 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
self.seen.add(message.id)
self.stream((meta[0], "messages", (message, meta[1])))
def _find_and_emit_messages(self, meta: Meta, response: Any) -> None:
if isinstance(response, BaseMessage):
self._emit(meta, response, dedupe=True)
elif isinstance(response, Sequence):
for value in response:
if isinstance(value, BaseMessage):
self._emit(meta, value, dedupe=True)
elif isinstance(response, dict):
for value in response.values():
if isinstance(value, BaseMessage):
self._emit(meta, value, dedupe=True)
elif isinstance(value, Sequence):
for item in value:
if isinstance(item, BaseMessage):
self._emit(meta, item, dedupe=True)
elif hasattr(response, "__dir__") and callable(response.__dir__):
for key in dir(response):
try:
value = getattr(response, key)
if isinstance(value, BaseMessage):
self._emit(meta, value, dedupe=True)
elif isinstance(value, Sequence):
for item in value:
if isinstance(item, BaseMessage):
self._emit(meta, item, dedupe=True)
except AttributeError:
pass
def tap_output_aiter(
self, run_id: UUID, output: AsyncIterator[T]
) -> AsyncIterator[T]:
@@ -149,43 +177,21 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
**kwargs: Any,
) -> Any:
if meta := self.metadata.pop(run_id, None):
# Handle Command node updates
if isinstance(response, Command):
response = response.update
if isinstance(response, Sequence) and any(
self._find_and_emit_messages(meta, response.update)
# Handle list of Command updates
elif isinstance(response, Sequence) and any(
isinstance(value, Command) for value in response
):
response = [
value.update if isinstance(value, Command) else value
for value in response
]
if isinstance(response, BaseMessage):
self._emit(meta, response, dedupe=True)
elif isinstance(response, Sequence):
for value in response:
if isinstance(value, BaseMessage):
self._emit(meta, value, dedupe=True)
elif isinstance(response, dict):
for value in response.values():
if isinstance(value, BaseMessage):
self._emit(meta, value, dedupe=True)
elif isinstance(value, Sequence):
for item in value:
if isinstance(item, BaseMessage):
self._emit(meta, item, dedupe=True)
elif hasattr(response, "__dir__") and callable(response.__dir__):
for key in dir(response):
try:
value = getattr(response, key)
if isinstance(value, BaseMessage):
self._emit(meta, value, dedupe=True)
elif isinstance(value, Sequence):
for item in value:
if isinstance(item, BaseMessage):
self._emit(meta, item, dedupe=True)
except AttributeError:
pass
if isinstance(value, Command):
self._find_and_emit_messages(meta, value.update)
else:
self._find_and_emit_messages(meta, value)
# Handle basic updates / streaming
else:
self._find_and_emit_messages(meta, response)
def on_chain_error(
self,
libs/langgraph/langgraph/pregel/remote.py
+32 -4
View File
@@ -54,6 +54,28 @@ CONF_DROPLIST = frozenset(
)
def sanitize_config_value(v: Any) -> Any:
"""Recursively sanitize a config value to ensure it contains only primitives."""
if isinstance(v, (str, int, float, bool)):
return v
elif isinstance(v, dict):
sanitized_dict = {}
for k, val in v.items():
if isinstance(k, str):
sanitized_value = sanitize_config_value(val)
if sanitized_value is not None:
sanitized_dict[k] = sanitized_value
return sanitized_dict
elif isinstance(v, (list, tuple)):
sanitized_list = []
for item in v:
sanitized_item = sanitize_config_value(item)
if sanitized_item is not None:
sanitized_list.append(sanitized_item)
return sanitized_list
return None
class RemoteException(Exception):
"""Exception raised when an error occurs in the remote graph."""
@@ -303,20 +325,26 @@ class RemoteGraph(PregelProtocol):
sanitized["recursion_limit"] = config["recursion_limit"]
if "tags" in config:
sanitized["tags"] = [tag for tag in config["tags"] if isinstance(tag, str)]
if "metadata" in config:
sanitized["metadata"] = {}
for k, v in config["metadata"].items():
if isinstance(k, str) and isinstance(v, (str, int, float, bool)):
sanitized["metadata"][k] = v
if (
isinstance(k, str)
and (sanitized_value := sanitize_config_value(v)) is not None
):
sanitized["metadata"][k] = sanitized_value
if "configurable" in config:
sanitized["configurable"] = {}
for k, v in config["configurable"].items():
if (
isinstance(k, str)
and k not in CONF_DROPLIST
and isinstance(v, (str, int, float, bool))
and (sanitized_value := sanitize_config_value(v)) is not None
):
sanitized["configurable"][k] = v
sanitized["configurable"][k] = sanitized_value
return sanitized
def get_state(
libs/langgraph/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph"
version = "0.3.31"
version = "0.3.33"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
license = "MIT"
libs/langgraph/tests/test_large_cases.py
+18 -18
View File
@@ -3034,7 +3034,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -3098,7 +3098,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -3209,8 +3209,8 @@ def test_state_graph_packets(
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config={
@@ -3367,7 +3367,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -3431,7 +3431,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=app_w_interrupt.checkpointer.get_tuple(config).config,
created_at=AnyStr(),
@@ -3536,8 +3536,8 @@ def test_state_graph_packets(
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config={
@@ -5928,7 +5928,7 @@ def test_copy_checkpoint(
PregelTask(
id=AnyStr(),
name="tool_one",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
result={"my_key": " one"},
),
PregelTask(
@@ -5982,7 +5982,7 @@ def test_copy_checkpoint(
PregelTask(
id=AnyStr(),
name="tool_one",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
),
PregelTask(
AnyStr(),
@@ -7511,7 +7511,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -7520,7 +7520,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="flaky",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),),
state=None,
@@ -7566,7 +7566,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -7575,7 +7575,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(),
state=None,
@@ -9569,7 +9569,7 @@ def test_send_react_interrupt(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -9729,7 +9729,7 @@ def test_send_react_interrupt(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -9820,7 +9820,7 @@ def test_send_react_interrupt(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -10039,7 +10039,7 @@ def test_send_react_interrupt_control(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
libs/langgraph/tests/test_large_cases_async.py
+8 -8
View File
@@ -2757,7 +2757,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -2822,7 +2822,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
@@ -2929,8 +2929,8 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config=tup.config,
@@ -3074,7 +3074,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
@@ -3135,7 +3135,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
@@ -3242,8 +3242,8 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config=tup.config,
libs/langgraph/tests/test_pregel.py
+14 -1
View File
@@ -6968,9 +6968,12 @@ def test_stream_mode_messages_command() -> None:
def my_other_node(state):
return Command(update={"messages": HumanMessage(content="bar")})
def my_last_node(state):
return [Command(update={"messages": HumanMessage(content="baz")})]
graph = (
StateGraph(MessagesState)
.add_sequence([my_node, my_other_node])
.add_sequence([my_node, my_other_node, my_last_node])
.add_edge(START, "my_node")
.compile()
)
@@ -7002,6 +7005,16 @@ def test_stream_mode_messages_command() -> None:
"langgraph_checkpoint_ns": AnyStr("my_other_node:"),
},
),
(
_AnyIdHumanMessage(content="baz"),
{
"langgraph_step": 3,
"langgraph_node": "my_last_node",
"langgraph_triggers": ("branch:to:my_last_node",),
"langgraph_path": ("__pregel_pull", "my_last_node"),
"langgraph_checkpoint_ns": AnyStr("my_last_node:"),
},
),
]
libs/langgraph/tests/test_pregel_async.py
+71 -10
View File
@@ -1012,7 +1012,7 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
PregelTask(
AnyStr(),
name="tool_one",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -1064,7 +1064,7 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
PregelTask(
AnyStr(),
"tool_one",
(PUSH, 0),
(PUSH, 0, False),
result=None,
),
PregelTask(
@@ -2988,7 +2988,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -2997,7 +2997,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="flaky",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),),
state=None,
@@ -3043,7 +3043,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3052,7 +3052,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(),
state=None,
@@ -3331,7 +3331,7 @@ async def test_send_react_interrupt(checkpointer_name: str) -> None:
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3489,7 +3489,7 @@ async def test_send_react_interrupt(checkpointer_name: str) -> None:
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3580,7 +3580,7 @@ async def test_send_react_interrupt(checkpointer_name: str) -> None:
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3797,7 +3797,7 @@ async def test_send_react_interrupt_control(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -8282,6 +8282,67 @@ async def test_handles_multiple_interrupts_from_tasks() -> None:
assert result[1] == "Added Will!"
@NEEDS_CONTEXTVARS
async def test_interrupts_in_tasks_surfaced_once() -> None:
@task
async def add_participant(name: str) -> str:
feedback = interrupt(f"Hey do you want to add {name}?")
if feedback is False:
return f"The user changed their mind and doesn't want to add {name}!"
if feedback is True:
return f"Added {name}!"
raise ValueError("Invalid feedback")
@entrypoint(checkpointer=MemorySaver())
async def program(_state: Any) -> list[str]:
first = await add_participant("James")
second = await add_participant("Will")
return [first, second]
config = {"configurable": {"thread_id": "1"}}
interrupts = [
e
async for e in program.astream("this is ignored", config=config)
if "__interrupt__" in e
]
assert len(interrupts) == 1
state = await program.aget_state(config=config)
assert len(state.tasks[0].interrupts) == 1
task_interrupt = state.tasks[0].interrupts[0]
assert task_interrupt.resumable is True
assert len(task_interrupt.ns) == 2
assert task_interrupt.ns[0].startswith("program:")
assert task_interrupt.ns[1].startswith("add_participant:")
assert task_interrupt.value == "Hey do you want to add James?"
interrupts = [
e
async for e in program.astream(Command(resume=True), config=config)
if "__interrupt__" in e
]
assert len(interrupts) == 1
state = await program.aget_state(config=config)
assert len(state.tasks[0].interrupts) == 1
task_interrupt = state.tasks[0].interrupts[0]
assert task_interrupt.resumable is True
assert len(task_interrupt.ns) == 2
assert task_interrupt.ns[0].startswith("program:")
assert task_interrupt.ns[1].startswith("add_participant:")
assert task_interrupt.value == "Hey do you want to add Will?"
result = await program.ainvoke(Command(resume=True), config=config)
assert result is not None
assert len(result) == 2
assert result[0] == "Added James!"
assert result[1] == "Added Will!"
async def test_pregel_loop_refcount():
gc.collect()
try:
libs/langgraph/tests/test_remote_graph.py
+98
View File
@@ -1,6 +1,7 @@
from unittest.mock import AsyncMock, MagicMock
import pytest
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.graph import (
Edge as DrawableEdge,
)
@@ -861,3 +862,100 @@ async def test_langgraph_cloud_integration():
remote_pregel.graph_id = "fe096781-5601-53d2-b2f6-0d3403f7e9ca" # must be UUID
graph = await remote_pregel.aget_graph(xray=True)
print("graph:", graph)
def test_sanitize_config():
# Create a test instance
remote = RemoteGraph("test-graph")
# Test 1: Basic config with primitives
basic_config: RunnableConfig = {
"recursion_limit": 10,
"tags": ["tag1", "tag2"],
"metadata": {"str_key": "value", "int_key": 42, "bool_key": True},
"configurable": {"param1": "value1", "param2": 123},
}
sanitized = remote._sanitize_config(basic_config)
assert sanitized["recursion_limit"] == 10
assert sanitized["tags"] == ["tag1", "tag2"]
assert sanitized["metadata"] == {
"str_key": "value",
"int_key": 42,
"bool_key": True,
}
assert sanitized["configurable"] == {"param1": "value1", "param2": 123}
# Test 2: Config with non-string tags and complex metadata
complex_config: RunnableConfig = {
"tags": ["tag1", 123, {"obj": "tag"}, "tag2"], # Only string tags should remain
"metadata": {
"nested": {
"key": "value",
"num": 42,
"invalid": lambda x: x,
}, # Last item should be removed
"list": [1, 2, "three"],
"invalid": lambda x: x, # Should be removed
"tuple": (1, 2, 3), # Should be converted to list
},
}
sanitized = remote._sanitize_config(complex_config)
assert sanitized["tags"] == ["tag1", "tag2"]
assert sanitized["metadata"] == {
"nested": {"key": "value", "num": 42},
"list": [1, 2, "three"],
"tuple": [1, 2, 3],
}
assert "invalid" not in sanitized["metadata"]
# Test 3: Config with configurable fields that should be dropped
config_with_drops: RunnableConfig = {
"configurable": {
"normal_param": "value",
"checkpoint_map": {"key": "value"}, # Should be dropped
"checkpoint_id": "123", # Should be dropped
"checkpoint_ns": "ns", # Should be dropped
}
}
sanitized = remote._sanitize_config(config_with_drops)
assert sanitized["configurable"] == {"normal_param": "value"}
assert "checkpoint_map" not in sanitized["configurable"]
assert "checkpoint_id" not in sanitized["configurable"]
assert "checkpoint_ns" not in sanitized["configurable"]
# Test 4: Empty config
empty_config: RunnableConfig = {}
sanitized = remote._sanitize_config(empty_config)
assert sanitized == {}
# Test 5: Config with non-string keys in configurable
invalid_keys_config: RunnableConfig = {
"configurable": {
"valid": "value",
123: "invalid", # Should be dropped
("tuple", "key"): "invalid", # Should be dropped
}
}
sanitized = remote._sanitize_config(invalid_keys_config)
assert sanitized["configurable"] == {"valid": "value"}
# Test 6: Deeply nested structures
nested_config: RunnableConfig = {
"metadata": {
"level1": {
"level2": {
"level3": {
"str": "value",
"list": [1, [2, [3]]],
"dict": {"a": {"b": {"c": "d"}}},
}
}
}
}
}
sanitized = remote._sanitize_config(nested_config)
assert sanitized["metadata"]["level1"]["level2"]["level3"]["str"] == "value"
assert sanitized["metadata"]["level1"]["level2"]["level3"]["list"] == [1, [2, [3]]]
assert sanitized["metadata"]["level1"]["level2"]["level3"]["dict"] == {
"a": {"b": {"c": "d"}}
}