From cedb310a122d5c3ac971d3b1c5b5777e1f15f97b Mon Sep 17 00:00:00 2001 From: William Fu-Hinthorn <13333726+hinthornw@users.noreply.github.com> Date: Wed, 7 Jan 2026 10:32:35 -0800 Subject: [PATCH] feat: topology arg for langgraph up --- libs/cli/langgraph_cli/cli.py | 41 ++++++++ libs/cli/langgraph_cli/config.py | 55 ++++++++++- libs/cli/langgraph_cli/docker.py | 50 +++++++++- libs/cli/tests/unit_tests/cli/test_cli.py | 2 + libs/cli/tests/unit_tests/test_config.py | 5 + libs/cli/tests/unit_tests/test_docker.py | 114 +++++++++++++++++++++- 6 files changed, 258 insertions(+), 9 deletions(-) diff --git a/libs/cli/langgraph_cli/cli.py b/libs/cli/langgraph_cli/cli.py index af78f9000..ac75cfbfa 100644 --- a/libs/cli/langgraph_cli/cli.py +++ b/libs/cli/langgraph_cli/cli.py @@ -158,6 +158,25 @@ OPT_API_VERSION = click.option( help="API server version to use for the base image. If unspecified, the latest version will be used.", ) +OPT_TOPOLOGY = click.option( + "--topology", + type=click.Choice(["single", "split"]), + default="split", + show_default=True, + help="""Deployment topology for the LangGraph services. + + \b + - 'single': Run API server and queue worker in the same container (legacy behavior) + - 'split': Run API server and queue worker in separate containers (recommended) + Other topologies (e.g., 'distributed') coming soon. + + \b + Example: + langgraph up --topology single + \b + """, +) + @click.group() @click.version_option(version=__version__, prog_name="LangGraph CLI") @@ -176,6 +195,7 @@ def cli(): @OPT_WATCH @OPT_POSTGRES_URI @OPT_API_VERSION +@OPT_TOPOLOGY @click.option( "--image", type=str, @@ -210,6 +230,7 @@ def up( debugger_base_url: str | None, postgres_uri: str | None, api_version: str | None, + topology: str, image: str | None, base_image: str | None, ): @@ -233,6 +254,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE debugger_base_url=debugger_base_url, postgres_uri=postgres_uri, api_version=api_version, + topology=topology, image=image, base_image=base_image, ) @@ -557,6 +579,7 @@ def dockerfile( capabilities, port=8123, base_image=base_image, + topology="split", ) # Add .env file to the docker-compose.yml for the langgraph-api service compose_dict["services"]["langgraph-api"]["env_file"] = [".env"] @@ -570,6 +593,17 @@ def dockerfile( compose_dict["services"]["langgraph-api"]["build"]["args"] = { "BASE_IMAGE": base_image } + # Also configure the worker service with the same build context + if "langgraph-worker" in compose_dict["services"]: + compose_dict["services"]["langgraph-worker"]["env_file"] = [".env"] + compose_dict["services"]["langgraph-worker"]["build"] = { + "context": ".", + "dockerfile": save_path.name, + } + if base_image: + compose_dict["services"]["langgraph-worker"]["build"]["args"] = { + "BASE_IMAGE": base_image + } f.write(langgraph_cli.docker.dict_to_yaml(compose_dict)) secho("✅ Created: docker-compose.yml", fg="green") @@ -793,6 +827,8 @@ def prepare_args_and_stdin( debugger_base_url: str | None = None, postgres_uri: str | None = None, api_version: str | None = None, + # Deployment topology: "single" (combined) or "split" (separate API/worker) + topology: str = "split", # Like "my-tag" (if you already built it locally) image: str | None = None, # Like "langchain/langgraphjs-api" or "langchain/langgraph-api @@ -809,6 +845,7 @@ def prepare_args_and_stdin( image=image, # Pass image to compose YAML generator base_image=base_image, api_version=api_version, + topology=topology, ) args = [ "--project-directory", @@ -826,6 +863,8 @@ def prepare_args_and_stdin( base_image=langgraph_cli.config.default_base_image(config), api_version=api_version, image=image, + topology=topology, + postgres_uri=postgres_uri, ) return args, stdin @@ -844,6 +883,7 @@ def prepare( debugger_base_url: str | None = None, postgres_uri: str | None = None, api_version: str | None = None, + topology: str = "split", image: str | None = None, base_image: str | None = None, ) -> tuple[list[str], str]: @@ -872,6 +912,7 @@ def prepare( debugger_base_url=debugger_base_url or f"http://127.0.0.1:{port}", postgres_uri=postgres_uri, api_version=api_version, + topology=topology, image=image, base_image=base_image, ) diff --git a/libs/cli/langgraph_cli/config.py b/libs/cli/langgraph_cli/config.py index 393f54f6d..32dacde45 100644 --- a/libs/cli/langgraph_cli/config.py +++ b/libs/cli/langgraph_cli/config.py @@ -1242,6 +1242,8 @@ def config_to_compose( api_version: str | None = None, image: str | None = None, watch: bool = False, + topology: str = "split", + postgres_uri: str | None = None, ) -> str: base_image = base_image or default_base_image(config) @@ -1268,11 +1270,20 @@ def config_to_compose( else: watch_str = "" if image: - return f""" + api_config = f""" {textwrap.indent(env_vars_str, " ")} {env_file_str} {watch_str} """ + # For split topology with pre-built image, worker also needs env vars + if topology == "split": + worker_config = f""" + langgraph-worker: +{textwrap.indent(env_vars_str, " ")} + {env_file_str} +""" + return api_config + worker_config + return api_config else: dockerfile, additional_contexts = config_to_docker( @@ -1292,7 +1303,7 @@ def config_to_compose( additional_contexts: {additional_contexts_str}""" - return f""" + api_config = f""" {textwrap.indent(env_vars_str, " ")} {env_file_str} pull_policy: build @@ -1302,3 +1313,43 @@ def config_to_compose( {textwrap.indent(dockerfile, " ")} {watch_str} """ + # For split topology, add complete worker service definition with build config + # (worker service is NOT added in docker.py when building from Dockerfile) + if topology == "split": + # Import here to avoid circular dependency + from langgraph_cli.docker import DEFAULT_POSTGRES_URI + + # Determine if using internal postgres (include_db) based on postgres_uri + include_db = postgres_uri is None + effective_postgres_uri = postgres_uri or DEFAULT_POSTGRES_URI + + # Build depends_on based on whether we have internal postgres + if include_db: + worker_depends_on = """depends_on: + langgraph-redis: + condition: service_healthy + langgraph-postgres: + condition: service_healthy""" + else: + worker_depends_on = """depends_on: + langgraph-redis: + condition: service_healthy""" + + worker_config = f""" + langgraph-worker: + {worker_depends_on} + environment: + REDIS_URI: redis://langgraph-redis:6379 + POSTGRES_URI: {effective_postgres_uri} +{textwrap.indent(env_vars_str, " ")} + {env_file_str} + entrypoint: + - /storage/queue_entrypoint.sh + pull_policy: build + build: + context: .{additional_contexts_str} + dockerfile_inline: | +{textwrap.indent(dockerfile, " ")} +""" + return api_config + worker_config + return api_config diff --git a/libs/cli/langgraph_cli/docker.py b/libs/cli/langgraph_cli/docker.py index 2f55f3484..3e0cdf84d 100644 --- a/libs/cli/langgraph_cli/docker.py +++ b/libs/cli/langgraph_cli/docker.py @@ -20,6 +20,7 @@ class Version(NamedTuple): DockerComposeType = Literal["plugin", "standalone"] +TopologyType = Literal["single", "split"] class DockerCapabilities(NamedTuple): @@ -149,6 +150,8 @@ def compose_as_dict( base_image: str | None = None, # API version of the base image api_version: str | None = None, + # Deployment topology: "single" (combined) or "split" (separate API/worker) + topology: TopologyType = "split", ) -> dict: """Create a docker compose file as a dictionary in YML style.""" if postgres_uri is None: @@ -207,15 +210,20 @@ def compose_as_dict( )["langgraph-debugger"] # Add langgraph-api service + api_environment = { + "REDIS_URI": "redis://langgraph-redis:6379", + "POSTGRES_URI": postgres_uri, + } + # In split mode, disable queue processing in the API server + if topology == "split": + api_environment["N_JOBS_PER_WORKER"] = "0" + services["langgraph-api"] = { "ports": [f'"{port}:8000"'], "depends_on": { "langgraph-redis": {"condition": "service_healthy"}, }, - "environment": { - "REDIS_URI": "redis://langgraph-redis:6379", - "POSTGRES_URI": postgres_uri, - }, + "environment": api_environment, } if image: services["langgraph-api"]["image"] = image @@ -235,6 +243,37 @@ def compose_as_dict( "start_period": "10s", } + # Add langgraph-worker service in split mode ONLY when using pre-built image + # When building from Dockerfile, the worker service is added in config_to_compose + # to avoid duplicate service key issues with YAML concatenation + if topology == "split" and image: + services["langgraph-worker"] = { + "depends_on": { + "langgraph-redis": {"condition": "service_healthy"}, + }, + "environment": { + "REDIS_URI": "redis://langgraph-redis:6379", + "POSTGRES_URI": postgres_uri, + }, + "entrypoint": ["/storage/queue_entrypoint.sh"], + "image": image, + } + + # If Postgres is included, add it to the dependencies of langgraph-worker + if include_db: + services["langgraph-worker"]["depends_on"]["langgraph-postgres"] = { + "condition": "service_healthy" + } + + # Additional healthcheck for langgraph-worker if supported + if capabilities.healthcheck_start_interval: + services["langgraph-worker"]["healthcheck"] = { + "test": "python /api/healthcheck.py", + "interval": "60s", + "start_interval": "1s", + "start_period": "10s", + } + # Final compose dictionary with volumes included if needed compose_dict = {} if include_db: @@ -255,6 +294,8 @@ def compose( image: str | None = None, base_image: str | None = None, api_version: str | None = None, + # Deployment topology: "single" (combined) or "split" (separate API/worker) + topology: TopologyType = "split", ) -> str: """Create a docker compose file as a string.""" compose_content = compose_as_dict( @@ -266,6 +307,7 @@ def compose( image=image, base_image=base_image, api_version=api_version, + topology=topology, ) compose_str = dict_to_yaml(compose_content) return compose_str diff --git a/libs/cli/tests/unit_tests/cli/test_cli.py b/libs/cli/tests/unit_tests/cli/test_cli.py index 58ef68343..0ddf9f3fc 100644 --- a/libs/cli/tests/unit_tests/cli/test_cli.py +++ b/libs/cli/tests/unit_tests/cli/test_cli.py @@ -66,6 +66,7 @@ def test_prepare_args_and_stdin() -> None: debugger_port=debugger_port, debugger_base_url=debugger_graph_url, watch=True, + topology="single", ) expected_args = [ @@ -189,6 +190,7 @@ def test_prepare_args_and_stdin_with_image() -> None: debugger_base_url=debugger_graph_url, watch=True, image="my-cool-image", + topology="single", ) expected_args = [ diff --git a/libs/cli/tests/unit_tests/test_config.py b/libs/cli/tests/unit_tests/test_config.py index 566ba272f..058fae44a 100644 --- a/libs/cli/tests/unit_tests/test_config.py +++ b/libs/cli/tests/unit_tests/test_config.py @@ -1147,6 +1147,7 @@ def test_config_to_compose_simple_config(): PATH_TO_CONFIG, validate_config({"dependencies": ["."], "graphs": graphs}), "langchain/langgraph-api", + topology="single", ) assert ( clean_empty_lines(actual_compose_stdin).strip() @@ -1195,6 +1196,7 @@ def test_config_to_compose_env_vars(): } ), "langchain/langgraph-api-custom", + topology="single", ) assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin @@ -1233,6 +1235,7 @@ def test_config_to_compose_env_file(): PATH_TO_CONFIG, validate_config({"dependencies": ["."], "graphs": graphs, "env": ".env"}), "langchain/langgraph-api", + topology="single", ) assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin @@ -1279,6 +1282,7 @@ def test_config_to_compose_watch(): validate_config({"dependencies": ["."], "graphs": graphs}), "langchain/langgraph-api", watch=True, + topology="single", ) assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin @@ -1326,6 +1330,7 @@ def test_config_to_compose_end_to_end(): validate_config({"dependencies": ["."], "graphs": graphs, "env": ".env"}), "langchain/langgraph-api", watch=True, + topology="single", ) assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin diff --git a/libs/cli/tests/unit_tests/test_docker.py b/libs/cli/tests/unit_tests/test_docker.py index b16d0df14..b0fd79d26 100644 --- a/libs/cli/tests/unit_tests/test_docker.py +++ b/libs/cli/tests/unit_tests/test_docker.py @@ -20,7 +20,10 @@ def test_compose_with_no_debugger_and_custom_db(): port = 8123 custom_postgres_uri = "custom_postgres_uri" actual_compose_str = compose( - DEFAULT_DOCKER_CAPABILITIES, port=port, postgres_uri=custom_postgres_uri + DEFAULT_DOCKER_CAPABILITIES, + port=port, + postgres_uri=custom_postgres_uri, + topology="single", ) expected_compose_str = f"""services: langgraph-redis: @@ -49,6 +52,7 @@ def test_compose_with_no_debugger_and_custom_db_with_healthcheck(): DEFAULT_DOCKER_CAPABILITIES._replace(healthcheck_start_interval=True), port=port, postgres_uri=custom_postgres_uri, + topology="single", ) expected_compose_str = f"""services: langgraph-redis: @@ -82,6 +86,7 @@ def test_compose_with_debugger_and_custom_db(): DEFAULT_DOCKER_CAPABILITIES, port=port, postgres_uri=custom_postgres_uri, + topology="single", ) expected_compose_str = f"""services: langgraph-redis: @@ -105,7 +110,9 @@ def test_compose_with_debugger_and_custom_db(): def test_compose_with_debugger_and_default_db(): port = 8123 - actual_compose_str = compose(DEFAULT_DOCKER_CAPABILITIES, port=port) + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES, port=port, topology="single" + ) expected_compose_str = f"""volumes: langgraph-data: driver: local @@ -157,7 +164,10 @@ def test_compose_with_api_version(): api_version = "0.2.74" actual_compose_str = compose( - DEFAULT_DOCKER_CAPABILITIES, port=port, api_version=api_version + DEFAULT_DOCKER_CAPABILITIES, + port=port, + api_version=api_version, + topology="single", ) # The compose function should generate a compose file that doesn't directly @@ -219,6 +229,7 @@ def test_compose_with_api_version_and_base_image(): port=port, api_version=api_version, base_image=base_image, + topology="single", ) # Similar to the previous test - the compose function doesn't directly embed @@ -280,6 +291,7 @@ def test_compose_with_api_version_and_custom_postgres(): port=port, api_version=api_version, postgres_uri=custom_postgres_uri, + topology="single", ) expected_compose_str = f"""services: @@ -313,6 +325,7 @@ def test_compose_with_api_version_and_debugger(): port=port, api_version=api_version, debugger_port=debugger_port, + topology="single", ) expected_compose_str = f"""volumes: @@ -368,6 +381,101 @@ services: assert clean_empty_lines(actual_compose_str) == expected_compose_str +def test_compose_with_single_topology(): + """Test compose function with single topology (legacy behavior).""" + port = 8123 + custom_postgres_uri = "custom_postgres_uri" + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES, + port=port, + postgres_uri=custom_postgres_uri, + topology="single", + ) + # Assert langgraph-worker service is NOT present + assert "langgraph-worker" not in actual_compose_str + assert "/storage/queue_entrypoint.sh" not in actual_compose_str + # Assert N_JOBS_PER_WORKER is NOT set (queue runs in API container) + assert "N_JOBS_PER_WORKER" not in actual_compose_str + + +def test_compose_with_split_topology(): + """Test compose function with split topology when using pre-built image.""" + port = 8123 + custom_postgres_uri = "custom_postgres_uri" + image = "my-custom-image:latest" + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES, + port=port, + postgres_uri=custom_postgres_uri, + topology="split", + image=image, + ) + # Assert langgraph-worker service is present when using image + assert "langgraph-worker" in actual_compose_str + assert "/storage/queue_entrypoint.sh" in actual_compose_str + # Assert N_JOBS_PER_WORKER is set to 0 on API service + assert "N_JOBS_PER_WORKER" in actual_compose_str + + +def test_compose_split_topology_is_default(): + """Test that split topology is the default (N_JOBS_PER_WORKER set on api).""" + port = 8123 + custom_postgres_uri = "custom_postgres_uri" + # Call without specifying topology + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES, + port=port, + postgres_uri=custom_postgres_uri, + ) + # In split mode without image, worker is NOT in compose (added in config_to_compose) + # But N_JOBS_PER_WORKER should be set on api service + assert "N_JOBS_PER_WORKER" in actual_compose_str + + +def test_compose_split_topology_with_image(): + """Test that worker service uses the same image as api in split mode.""" + port = 8123 + image = "my-custom-image:latest" + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES, + port=port, + image=image, + topology="split", + ) + # Both services should reference the same image + assert actual_compose_str.count(f"image: {image}") == 2 + + +def test_compose_split_topology_with_postgres(): + """Test split topology includes postgres service (worker added when image provided).""" + port = 8123 + image = "my-custom-image:latest" + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES, + port=port, + topology="split", + image=image, + ) + # Worker should be present when image is provided + assert "langgraph-worker" in actual_compose_str + # The compose should include langgraph-postgres service + assert "langgraph-postgres" in actual_compose_str + + +def test_compose_split_topology_with_healthcheck(): + """Test split topology with healthcheck capabilities when using image.""" + port = 8123 + image = "my-custom-image:latest" + actual_compose_str = compose( + DEFAULT_DOCKER_CAPABILITIES._replace(healthcheck_start_interval=True), + port=port, + topology="split", + image=image, + ) + # Both api and worker should have healthchecks + assert actual_compose_str.count("python /api/healthcheck.py") == 2 + + @pytest.mark.parametrize( "input_str,expected", [