From e0da491fe66f71bf83b158c6112f855b82ff1ef6 Mon Sep 17 00:00:00 2001 From: David Asamu Date: Thu, 17 Apr 2025 21:38:41 +0100 Subject: [PATCH 01/12] docs: Add docs for REDIS_KEY_PREFIX env var --- docs/docs/cloud/reference/env_var.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/docs/cloud/reference/env_var.md b/docs/docs/cloud/reference/env_var.md index e15157cac..e82778248 100644 --- a/docs/docs/cloud/reference/env_var.md +++ b/docs/docs/cloud/reference/env_var.md @@ -89,3 +89,12 @@ 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 `''`. From f94eeabebde25cb290a26c215428fa1bd4b1242d Mon Sep 17 00:00:00 2001 From: David Asamu Date: Thu, 17 Apr 2025 23:20:53 +0100 Subject: [PATCH 02/12] add doc for REDIS_CLUSTER env var --- docs/docs/cloud/reference/env_var.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/docs/docs/cloud/reference/env_var.md b/docs/docs/cloud/reference/env_var.md index e82778248..099ee8b9d 100644 --- a/docs/docs/cloud/reference/env_var.md +++ b/docs/docs/cloud/reference/env_var.md @@ -98,3 +98,12 @@ Specify `REDIS_URI_CUSTOM` to use a custom Redis instance. The value of `REDIS_U 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 for Self-Hosted Data Plane and Self-Hosted Control Plane" + Redis Cluster mode is 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. + +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`. From 29e9ee2d7b7a32518ff04d4a814f7188b0452950 Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Tue, 22 Apr 2025 21:25:00 -0400 Subject: [PATCH 03/12] docs(agents): use list of messages format (#4378) --- docs/docs/agents/agents.md | 16 ++++++---- docs/docs/agents/context.md | 6 ++-- docs/docs/agents/human-in-the-loop.md | 4 +-- docs/docs/agents/mcp.md | 8 +++-- docs/docs/agents/memory.md | 10 +++--- docs/docs/agents/multi-agent.md | 44 +++++++++++++++++++++------ docs/docs/agents/run_agents.md | 12 ++++---- docs/docs/agents/streaming.md | 16 +++++----- docs/docs/agents/tools.md | 24 +++++++++++---- 9 files changed, 92 insertions(+), 48 deletions(-) diff --git a/docs/docs/agents/agents.md b/docs/docs/agents/agents.md index d58762118..c58895114 100644 --- a/docs/docs/agents/agents.md +++ b/docs/docs/agents/agents.md @@ -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"] diff --git a/docs/docs/agents/context.md b/docs/docs/agents/context.md index 9247fc5b6..c3854b6dc 100644 --- a/docs/docs/agents/context.md +++ b/docs/docs/agents/context.md @@ -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"}} ) diff --git a/docs/docs/agents/human-in-the-loop.md b/docs/docs/agents/human-in-the-loop.md index 4af4a6b13..a59e55f65 100644 --- a/docs/docs/agents/human-in-the-loop.md +++ b/docs/docs/agents/human-in-the-loop.md @@ -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 ): diff --git a/docs/docs/agents/mcp.md b/docs/docs/agents/mcp.md index fd520cf4e..22417ceb4 100644 --- a/docs/docs/agents/mcp.md +++ b/docs/docs/agents/mcp.md @@ -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 diff --git a/docs/docs/agents/memory.md b/docs/docs/agents/memory.md index 17a3d1ab4..ab90550c9 100644 --- a/docs/docs/agents/memory.md +++ b/docs/docs/agents/memory.md @@ -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)! ) diff --git a/docs/docs/agents/multi-agent.md b/docs/docs/agents/multi-agent.md index 7b838831b..5494f42fc 100644 --- a/docs/docs/agents/multi-agent.md +++ b/docs/docs/agents/multi-agent.md @@ -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") ``` diff --git a/docs/docs/agents/run_agents.md b/docs/docs/agents/run_agents.md index 1d139a19e..1dc3d6217 100644 --- a/docs/docs/agents/run_agents.md +++ b/docs/docs/agents/run_agents.md @@ -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.") diff --git a/docs/docs/agents/streaming.md b/docs/docs/agents/streaming.md index b92f3f5a4..1491c7b38 100644 --- a/docs/docs/agents/streaming.md +++ b/docs/docs/agents/streaming.md @@ -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"] ): diff --git a/docs/docs/agents/tools.md b/docs/docs/agents/tools.md index 06aa9f049..436958cb7 100644 --- a/docs/docs/agents/tools.md +++ b/docs/docs/agents/tools.md @@ -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]. From 05f21a384bcaf3844f5da79f1efb494c0a1be8a7 Mon Sep 17 00:00:00 2001 From: William FH <13333726+hinthornw@users.noreply.github.com> Date: Wed, 23 Apr 2025 07:48:01 -0700 Subject: [PATCH 04/12] Add image arg to up command (#4385) Using this argument, you can get more customization since you can do `langgraph build` or directly `docker build` your image and then re-use the `langgraph up --image my-image` and have it also spin up redis & postgres for you. Easier then writing your own compose file --- libs/cli/langgraph_cli/cli.py | 14 +++ libs/cli/langgraph_cli/config.py | 26 ++++-- libs/cli/langgraph_cli/docker.py | 6 ++ libs/cli/pyproject.toml | 2 +- libs/cli/tests/unit_tests/cli/test_cli.py | 104 ++++++++++++++++++++++ 5 files changed, 143 insertions(+), 9 deletions(-) diff --git a/libs/cli/langgraph_cli/cli.py b/libs/cli/langgraph_cli/cli.py index 14b9e6741..7ec9a3746 100644 --- a/libs/cli/langgraph_cli/cli.py +++ b/libs/cli/langgraph_cli/cli.py @@ -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 diff --git a/libs/cli/langgraph_cli/config.py b/libs/cli/langgraph_cli/config.py index 826ec3b89..efbe55aa2 100644 --- a/libs/cli/langgraph_cli/config.py +++ b/libs/cli/langgraph_cli/config.py @@ -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 diff --git a/libs/cli/langgraph_cli/docker.py b/libs/cli/langgraph_cli/docker.py index dc6ac245e..e6b633216 100644 --- a/libs/cli/langgraph_cli/docker.py +++ b/libs/cli/langgraph_cli/docker.py @@ -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 diff --git a/libs/cli/pyproject.toml b/libs/cli/pyproject.toml index ec8b82f58..7860bf2ec 100644 --- a/libs/cli/pyproject.toml +++ b/libs/cli/pyproject.toml @@ -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" diff --git a/libs/cli/tests/unit_tests/cli/test_cli.py b/libs/cli/tests/unit_tests/cli/test_cli.py index 8762448e6..e81b9e291 100644 --- a/libs/cli/tests/unit_tests/cli/test_cli.py +++ b/libs/cli/tests/unit_tests/cli/test_cli.py @@ -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() From 24f04fba3700b0b6f22d237807268b9dc68f112d Mon Sep 17 00:00:00 2001 From: Asamu David Date: Wed, 23 Apr 2025 15:52:07 +0100 Subject: [PATCH 05/12] Update REDIS_CLUSTER env doc.md --- docs/docs/cloud/reference/env_var.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/docs/cloud/reference/env_var.md b/docs/docs/cloud/reference/env_var.md index 099ee8b9d..47812b98b 100644 --- a/docs/docs/cloud/reference/env_var.md +++ b/docs/docs/cloud/reference/env_var.md @@ -101,8 +101,8 @@ Defaults to `''`. ## `REDIS_CLUSTER` -!!! info "Only for Self-Hosted Data Plane and Self-Hosted Control Plane" - Redis Cluster mode is 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. +!!! 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. From 364c572f0fd808b00417ce82206d0e44613b0f1e Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Wed, 23 Apr 2025 11:46:57 -0400 Subject: [PATCH 06/12] langgraph: fix messages streaming for list of Commands (#4379) Fixes #4372 --- libs/langgraph/langgraph/pregel/messages.py | 72 +++++++++++---------- libs/langgraph/tests/test_pregel.py | 15 ++++- 2 files changed, 53 insertions(+), 34 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/messages.py b/libs/langgraph/langgraph/pregel/messages.py index fa9ace1e1..de6425309 100644 --- a/libs/langgraph/langgraph/pregel/messages.py +++ b/libs/langgraph/langgraph/pregel/messages.py @@ -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, diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index ce6474ca1..480d9dd33 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -6925,9 +6925,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() ) @@ -6959,6 +6962,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:"), + }, + ), ] From 0147790937995d6bc87e750bad0bd7e9afa983df Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Wed, 23 Apr 2025 11:51:29 -0400 Subject: [PATCH 07/12] langgraph: release 0.3.32 (#4386) --- libs/langgraph/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index f3e2a9b1c..6c47446f2 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langgraph" -version = "0.3.31" +version = "0.3.32" description = "Building stateful, multi-actor applications with LLMs" authors = [] license = "MIT" From 6bca615e6a926ffc2cf97b9d4da497eb61ce08c9 Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Wed, 23 Apr 2025 14:04:54 -0400 Subject: [PATCH 08/12] langgraph: allow nested lists/dicts of primitives in RemoteGraph config (#4387) --- libs/langgraph/langgraph/pregel/remote.py | 36 ++++++++- libs/langgraph/tests/test_remote_graph.py | 98 +++++++++++++++++++++++ 2 files changed, 130 insertions(+), 4 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index ba7cdb2c6..09d5f90a8 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -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( diff --git a/libs/langgraph/tests/test_remote_graph.py b/libs/langgraph/tests/test_remote_graph.py index 878f5e6f1..b30ec3a05 100644 --- a/libs/langgraph/tests/test_remote_graph.py +++ b/libs/langgraph/tests/test_remote_graph.py @@ -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"}} + } From 6bcab08f55cfd485a2c5d3c62a902291a2111809 Mon Sep 17 00:00:00 2001 From: Vadym Barda Date: Wed, 23 Apr 2025 16:28:12 -0400 Subject: [PATCH 09/12] langgraph: release 0.3.33 (#4388) --- libs/langgraph/pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index 6c47446f2..6ac7a0957 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langgraph" -version = "0.3.32" +version = "0.3.33" description = "Building stateful, multi-actor applications with LLMs" authors = [] license = "MIT" From f42fc971e39257526f8e2a989b2d13bb1e8e4d9a Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Wed, 23 Apr 2025 14:16:35 -0700 Subject: [PATCH 10/12] fix double interrupt raise with task path bool flag --- libs/langgraph/langgraph/pregel/algo.py | 18 +++++----- libs/langgraph/langgraph/pregel/loop.py | 6 ++++ libs/langgraph/tests/test_large_cases.py | 36 +++++++++---------- .../langgraph/tests/test_large_cases_async.py | 16 ++++----- libs/langgraph/tests/test_pregel_async.py | 20 +++++------ 5 files changed, 52 insertions(+), 44 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 94a5ade90..f931c3eb1 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -542,11 +542,12 @@ def prepare_single_task( str(task_path[2]), ) task_checkpoint_ns = f"{checkpoint_ns}:{task_id}" + 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 +576,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 +599,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 +638,12 @@ def prepare_single_task( logger.warning(f"Ignoring invalid PUSH task path {task_path}") return task_checkpoint_ns = f"{checkpoint_ns}:{task_id}" + 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 +680,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 +710,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]) diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index c88362df6..4a2a69f77 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -903,12 +903,18 @@ class PregelLoop(LoopProtocol): def _output_writes( self, task_id: str, writes: Sequence[tuple[str, Any]], *, cached: bool = False ) -> None: + print(f"output writes {task_id}, {writes}") if task := self.tasks.get(task_id): if task.config is not None and TAG_HIDDEN in task.config.get( "tags", EMPTY_SEQ ): 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 self._emit( "updates", lambda: iter( diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index 3c0e7da87..eb3406e1c 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -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={ @@ -5916,7 +5916,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( @@ -5970,7 +5970,7 @@ def test_copy_checkpoint( PregelTask( id=AnyStr(), name="tool_one", - path=("__pregel_push", 0), + path=("__pregel_push", 0, False), ), PregelTask( AnyStr(), @@ -7485,7 +7485,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, @@ -7494,7 +7494,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, @@ -7540,7 +7540,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, @@ -7549,7 +7549,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, @@ -9543,7 +9543,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, @@ -9703,7 +9703,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, @@ -9794,7 +9794,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, @@ -10013,7 +10013,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, diff --git a/libs/langgraph/tests/test_large_cases_async.py b/libs/langgraph/tests/test_large_cases_async.py index 6dbf19fb2..3062682f7 100644 --- a/libs/langgraph/tests/test_large_cases_async.py +++ b/libs/langgraph/tests/test_large_cases_async.py @@ -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, diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index 1f6b25574..802845e46 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -992,7 +992,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, @@ -1044,7 +1044,7 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None: PregelTask( AnyStr(), "tool_one", - (PUSH, 0), + (PUSH, 0, False), result=None, ), PregelTask( @@ -2952,7 +2952,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, @@ -2961,7 +2961,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, @@ -3007,7 +3007,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, @@ -3016,7 +3016,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, @@ -3295,7 +3295,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, @@ -3453,7 +3453,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, @@ -3544,7 +3544,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, @@ -3761,7 +3761,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, From 37b2956758586c98dd58df006ecf3831e9da5ebb Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Wed, 23 Apr 2025 14:24:57 -0700 Subject: [PATCH 11/12] helpful comments + test --- libs/langgraph/langgraph/pregel/algo.py | 4 ++ libs/langgraph/langgraph/pregel/loop.py | 1 - libs/langgraph/tests/test_pregel_async.py | 61 +++++++++++++++++++++++ 3 files changed, 65 insertions(+), 1 deletion(-) diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index f931c3eb1..19cd9d1ff 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -542,6 +542,8 @@ 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, @@ -638,6 +640,8 @@ 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, diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 4a2a69f77..3e3849fc7 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -903,7 +903,6 @@ class PregelLoop(LoopProtocol): def _output_writes( self, task_id: str, writes: Sequence[tuple[str, Any]], *, cached: bool = False ) -> None: - print(f"output writes {task_id}, {writes}") if task := self.tasks.get(task_id): if task.config is not None and TAG_HIDDEN in task.config.get( "tags", EMPTY_SEQ diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index 802845e46..58f481f4a 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -8190,6 +8190,67 @@ async def test_handles_multiple_interrupts_from_tasks() -> None: assert result[1] == "Added Will!" +@NEEDS_CONTEXTVARS +async def test_tasks_in_interrupts_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: From 67075bfb7fc18589612771dd2dee5e1b118033a3 Mon Sep 17 00:00:00 2001 From: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com> Date: Wed, 23 Apr 2025 15:50:22 -0700 Subject: [PATCH 12/12] Update libs/langgraph/tests/test_pregel_async.py --- libs/langgraph/tests/test_pregel_async.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index 58f481f4a..af3b16dfa 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -8191,7 +8191,7 @@ async def test_handles_multiple_interrupts_from_tasks() -> None: @NEEDS_CONTEXTVARS -async def test_tasks_in_interrupts_surfaced_once() -> None: +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}?")