diff --git a/docs/docs/concepts/langgraph_cloud.md b/docs/docs/concepts/langgraph_cloud.md index 11e5c8afc..6cd6f4b8b 100644 --- a/docs/docs/concepts/langgraph_cloud.md +++ b/docs/docs/concepts/langgraph_cloud.md @@ -21,6 +21,18 @@ See the [how-to guide](../cloud/deployment/cloud.md#create-new-deployment) for c | Development | 1 CPU | 1 GB | Up to 1 container | | Production | 1 CPU | 2 GB | Up to 10 containers | +## Autoscaling +`Production` type deployments automatically scale up to 10 containers. Scaling is based on the current request load for a single container. Specifically, the autoscaling implementation scales the deployment so that each container is processing about 10 concurrent requests. For example... + +- If the deployment is processing 20 concurrent requests, the deployment will scale up from 1 container to 2 containers (20 requests / 2 containers = 10 requests per container). +- If a deployment of 2 containers is processing 10 requests, the deployment will scale down from 2 containers to 1 container (10 requests / 1 container = 10 requests per container). + +10 concurrent requests per container is the target threshold. However, 10 concurrent requests per container is not a hard limit. The number of concurrent requests can exceed 10 if there is a sudden burst of requests. + +Scale down actions are delayed for 30 minutes before any action is taken. In other words, if the autoscaling implementation decides to scale down a deployment, it will first wait for 30 minutes before scaling down. After 30 minutes, the concurrency metric is recomputed and the deployment will scale down if the concurrency metric has met the target threshold. Otherwise, the deployment remains scaled up. This "cool down" period ensures that deployments do not scale up and down too frequently. + +In the future, the autoscaling implementation may evolve to accommodate other metrics such as background run queue size. + ## Revision A revision is an iteration of a [deployment](#deployment). When a new deployment is created, an initial revision is automatically created. To deploy new code changes or update environment variable configurations for a deployment, a new revision must be created. When a revision is created, a new container image is built automatically. diff --git a/docs/docs/concepts/low_level.md b/docs/docs/concepts/low_level.md index a2a36534b..522017bcc 100644 --- a/docs/docs/concepts/low_level.md +++ b/docs/docs/concepts/low_level.md @@ -353,6 +353,12 @@ def my_node(state: State) -> Command[Literal["my_other_node"]]: Check out this [how-to guide](../how-tos/command.ipynb) for an end-to-end example of how to use `Command`. +### When should I use Command instead of conditional edges? + +Use `Command` when you need to **both** update the graph state **and** route to a different node. For example, when implementing [multi-agent handoffs](./multi_agent.md#handoffs) where it's important to route to a different agent and pass some information to that agent. + +Use [conditional edges](#conditional-edges) to route between nodes conditionally without updating the state. + ### Using inside tools A common use case is updating graph state from inside a tool. For example, in a customer support application you might want to look up customer information based on their account number or ID in the beginning of the conversation. To update the graph state from the tool, you can return `Command(update={"my_custom_key": "foo", "messages": [...]})` from the tool: diff --git a/docs/docs/concepts/multi_agent.md b/docs/docs/concepts/multi_agent.md index 6979c249c..b0f4c3f4a 100644 --- a/docs/docs/concepts/multi_agent.md +++ b/docs/docs/concepts/multi_agent.md @@ -153,7 +153,7 @@ network = builder.compile() ### Supervisor -In this architecture, we define agents as nodes and add a supervisor node (LLM) that decides which agent nodes should be called next. We use [conditional edges](./low_level.md#conditional-edges) to route execution to the appropriate agent node based on supervisor's decision. This architecture also lends itself well to running multiple agents in parallel or using [map-reduce](../how-tos/map-reduce.ipynb) pattern. +In this architecture, we define agents as nodes and add a supervisor node (LLM) that decides which agent nodes should be called next. We use [`Command`](./low_level.md#command) to route execution to the appropriate agent node based on supervisor's decision. This architecture also lends itself well to running multiple agents in parallel or using [map-reduce](../how-tos/map-reduce.ipynb) pattern. ```python from typing import Literal diff --git a/docs/docs/how-tos/multi-agent-network.ipynb b/docs/docs/how-tos/multi-agent-network.ipynb index 4dde1ac79..3c8461438 100644 --- a/docs/docs/how-tos/multi-agent-network.ipynb +++ b/docs/docs/how-tos/multi-agent-network.ipynb @@ -254,7 +254,7 @@ "name": "stdout", "output_type": "stream", "text": [ - "{'travel_advisor': {'messages': {'role': 'ai', 'content': 'The Caribbean is a fantastic choice for warm, sunny weather and beautiful beaches. Here are a few destinations you might consider:\\n\\n1. **Jamaica**: Known for its vibrant culture, reggae music, and stunning beaches like Negril and Montego Bay.\\n\\n2. **Bahamas**: With over 700 islands, the Bahamas offers clear turquoise waters and beautiful sandy beaches, perfect for relaxation and water sports.\\n\\n3. **Dominican Republic**: Known for its resorts, beaches, and golfing. Punta Cana and Puerto Plata are popular destinations.\\n\\n4. **Barbados**: Offers beautiful beaches and a rich history, with plenty of activities and festivals.\\n\\n5. **Puerto Rico**: A mix of Spanish heritage and modern resorts, with opportunities for hiking in El Yunque National Forest and enjoying the vibrant nightlife in San Juan.\\n\\n6. **Aruba**: Known for its dry climate and sunny days, with beautiful beaches and activities like snorkeling and diving.\\n\\nEach of these destinations has its own unique charm and appeal. If you need specific sightseeing or hotel recommendations, let me know!', 'name': 'travel_advisor'}}}\n", + "{'travel_advisor': {'messages': {'role': 'ai', 'content': 'The Caribbean offers many warm destinations perfect for a relaxing getaway. Consider visiting Jamaica for its beautiful beaches and vibrant culture, the Bahamas for its stunning islands and clear waters, or the Dominican Republic for its all-inclusive resorts and rich history. Let me know if you need more information on sightseeing or hotel recommendations!', 'name': 'travel_advisor'}}}\n", "\n", "\n" ] @@ -286,10 +286,13 @@ "name": "stdout", "output_type": "stream", "text": [ - "{'travel_advisor': {'messages': {'role': 'ai', 'content': \"I recommend visiting Barbados for a warm Caribbean getaway. It's known for its beautiful beaches, vibrant culture, and friendly locals. Let me gather some sightseeing and hotel recommendations for you.\", 'name': 'travel_advisor'}}}\n", + "{'travel_advisor': {'messages': {'role': 'ai', 'content': 'I recommend visiting Jamaica, a beautiful Caribbean island known for its warm climate, stunning beaches, and vibrant culture.', 'name': 'travel_advisor'}}}\n", "\n", "\n", - "{'sightseeing_advisor': {'messages': {'role': 'ai', 'content': \"Barbados is a fantastic destination to experience the warmth of the Caribbean. Here are some things to do and places to stay:\\n\\n### Sightseeing Recommendations:\\n1. **Harrison's Cave**: Explore this stunning limestone cave with its impressive stalactites and stalagmites. \\n2. **Bathsheba Beach**: Known for its unique rock formations and surf-friendly waves, it's perfect for a day of relaxation and exploration.\\n3. **St. Nicholas Abbey**: Visit this historic plantation house, distillery, and museum for a glimpse into the island's colonial past.\\n4. **Animal Flower Cave**: Located at the northern tip of Barbados, this sea cave offers incredible views and natural rock pools.\\n5. **Oistins Fish Fry**: Experience local culture and cuisine with fresh seafood, music, and dancing every Friday night.\\n\\n### Hotel Recommendations:\\n1. **Sandy Lane**: A luxurious resort offering world-class amenities, golf courses, and a private beach.\\n2. **The Crane Resort**: Known for its historic charm and stunning ocean views, it provides a unique blend of luxury and culture.\\n3. **Sea Breeze Beach House**: Offers an all-inclusive experience with multiple dining options and beachfront access.\\n4. **The House by Elegant Hotels**: A boutique, adults-only hotel perfect for a romantic getaway with personalized service and beachfront location.\\n\\nEnjoy your trip to Barbados!\", 'name': 'sightseeing_advisor'}}}\n", + "{'sightseeing_advisor': {'messages': {'role': 'ai', 'content': \"Jamaica is a fantastic choice for a warm Caribbean getaway. Here are some top things to do while you're there:\\n\\n1. **Dunn's River Falls**: Located near Ocho Rios, this is one of Jamaica's most famous waterfalls. You can climb the falls, swim in the refreshing pools, or simply enjoy the beautiful surroundings.\\n\\n2. **Seven Mile Beach**: Located in Negril, this is one of the most beautiful beaches in the Caribbean. It's perfect for sunbathing, swimming, and enjoying water sports.\\n\\n3. **Bob Marley Museum**: Situated in Kingston, this museum is dedicated to the life and legacy of the reggae legend Bob Marley and is a must-visit for music lovers.\\n\\n4. **Blue Mountains**: Go hiking or take a tour to explore the Blue Mountains, where you can enjoy breathtaking views and taste some of the world's best coffee.\\n\\n5. **Luminous Lagoon**: Experience the natural wonder of the Luminous Lagoon in Falmouth, where the water glows at night due to bioluminescent microorganisms.\\n\\nFor hotel recommendations, I suggest checking with a hotel advisor for the best options that suit your budget and preferences.\", 'name': 'sightseeing_advisor'}}}\n", + "\n", + "\n", + "{'hotel_advisor': {'messages': {'role': 'ai', 'content': 'For hotel recommendations in Jamaica, here are a few options across different areas: \\n\\n1. **Sandals Montego Bay** (Montego Bay): A luxurious all-inclusive resort ideal for couples, offering beautiful beachfront views and a variety of dining options.\\n\\n2. **Half Moon Resort** (Montego Bay): A family-friendly resort with a private beach, golf course, and various activities for all ages.\\n\\n3. **Jamaica Inn** (Ocho Rios): A charming boutique hotel known for its excellent service and tranquil atmosphere.\\n\\n4. **The Caves** (Negril): A unique and romantic cliff-side resort offering stunning ocean views and intimate dining experiences.\\n\\n5. **Trident Hotel** (Port Antonio): A luxurious and contemporary hotel offering privacy, elegance, and beautiful views of the Caribbean Sea.\\n\\nThese options cater to different tastes and budgets, ensuring a comfortable and enjoyable stay in Jamaica.', 'name': 'hotel_advisor'}}}\n", "\n", "\n" ] @@ -558,7 +561,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.12.3" + "version": "3.11.9" } }, "nbformat": 4, diff --git a/docs/docs/how-tos/update-state-from-tools.ipynb b/docs/docs/how-tos/update-state-from-tools.ipynb index 41147688e..6e916ca2a 100644 --- a/docs/docs/how-tos/update-state-from-tools.ipynb +++ b/docs/docs/how-tos/update-state-from-tools.ipynb @@ -31,7 +31,7 @@ " # update the state keys\n", " \"user_info\": user_info,\n", " # update the message history\n", - " \"messages\": [ToolMessage(\"Successfully looked up user information\", tool_call_id=\"\")]\n", + " \"messages\": [ToolMessage(\"Successfully looked up user information\", tool_call_id=tool_call_id)]\n", " }\n", " )\n", "```\n", diff --git a/libs/checkpoint/langgraph/checkpoint/memory/__init__.py b/libs/checkpoint/langgraph/checkpoint/memory/__init__.py index e30c082c7..cb6b7b852 100644 --- a/libs/checkpoint/langgraph/checkpoint/memory/__init__.py +++ b/libs/checkpoint/langgraph/checkpoint/memory/__init__.py @@ -1,4 +1,3 @@ -import asyncio import logging import os import pickle @@ -6,7 +5,6 @@ import random import shutil from collections import defaultdict from contextlib import AbstractAsyncContextManager, AbstractContextManager, ExitStack -from functools import partial from types import TracebackType from typing import Any, AsyncIterator, Dict, Iterator, Optional, Sequence, Tuple, Type @@ -395,9 +393,7 @@ class MemorySaver( Returns: Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found. """ - return await asyncio.get_running_loop().run_in_executor( - None, self.get_tuple, config - ) + return self.get_tuple(config) async def alist( self, @@ -418,24 +414,8 @@ class MemorySaver( Yields: AsyncIterator[CheckpointTuple]: An asynchronous iterator of checkpoint tuples. """ - loop = asyncio.get_running_loop() - iter = await loop.run_in_executor( - None, - partial( - self.list, - before=before, - limit=limit, - filter=filter, - ), - config, - ) - while True: - # handling StopIteration exception inside coroutine won't work - # as expected, so using next() with default value to break the loop - if item := await loop.run_in_executor(None, next, iter, None): - yield item - else: - break + for item in self.list(config, filter=filter, before=before, limit=limit): + yield item async def aput( self, @@ -455,9 +435,7 @@ class MemorySaver( Returns: RunnableConfig: The updated config containing the saved checkpoint's timestamp. """ - return await asyncio.get_running_loop().run_in_executor( - None, self.put, config, checkpoint, metadata, new_versions - ) + return self.put(config, checkpoint, metadata, new_versions) async def aput_writes( self, @@ -474,10 +452,9 @@ class MemorySaver( config (RunnableConfig): The config to associate with the writes. writes (List[Tuple[str, Any]]): The writes to save, each as a (channel, value) pair. task_id (str): Identifier for the task creating the writes. + return self.put_writes(config, writes, task_id) """ - return await asyncio.get_running_loop().run_in_executor( - None, self.put_writes, config, writes, task_id - ) + return self.put_writes(config, writes, task_id) def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str: if current is None: diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index c416d5f6a..7a5614f91 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -559,6 +559,7 @@ class StateGraph(Graph): for key, node in self.nodes.items(): compiled.attach_node(key, node) + compiled.attach_branch(START, SELF, CONTROL_BRANCH, with_reader=False) for key, node in self.nodes.items(): compiled.attach_branch(key, SELF, CONTROL_BRANCH, with_reader=False) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index e714afe21..ba4533830 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -18,7 +18,6 @@ from typing import ( Type, Union, cast, - get_type_hints, overload, ) from uuid import UUID, uuid5 @@ -117,6 +116,7 @@ from langgraph.utils.config import ( patch_config, patch_configurable, ) +from langgraph.utils.fields import get_enhanced_type_hints from langgraph.utils.pydantic import create_model from langgraph.utils.queue import AsyncQueue, SyncQueue # type: ignore[attr-defined] @@ -319,8 +319,15 @@ class Pregel(PregelProtocol): ) + ( [ - ConfigurableFieldSpec(id=name, annotation=typ) - for name, typ in get_type_hints(self.config_type).items() + ConfigurableFieldSpec( + id=name, + annotation=typ, + default=default, + description=description, + ) + for name, typ, default, description in get_enhanced_type_hints( + self.config_type + ) ] if self.config_type is not None else [] diff --git a/libs/langgraph/langgraph/pregel/io.py b/libs/langgraph/langgraph/pregel/io.py index f2df972d8..d54f8b8b2 100644 --- a/libs/langgraph/langgraph/pregel/io.py +++ b/libs/langgraph/langgraph/pregel/io.py @@ -14,6 +14,8 @@ from langgraph.constants import ( PUSH, RESUME, RETURN, + SELF, + START, TAG_HIDDEN, TASKS, ) @@ -79,12 +81,14 @@ def map_command( else: sends = [cmd.goto] for send in sends: - if not isinstance(send, Send): + if isinstance(send, Send): + yield (NULL_TASK_ID, PUSH if FF_SEND_V2 else TASKS, send) + elif isinstance(send, str): + yield (NULL_TASK_ID, f"branch:{START}:{SELF}:{send}", START) + else: raise TypeError( - f"In Command.goto, expected Send, got {type(send).__name__}" + f"In Command.goto, expected Send/str, got {type(send).__name__}" ) - yield (NULL_TASK_ID, PUSH if FF_SEND_V2 else TASKS, send) - # TODO handle goto str for state graph if cmd.resume: if isinstance(cmd.resume, dict) and all(is_task_id(k) for k in cmd.resume): for tid, resume in cmd.resume.items(): diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index e96276259..cf0716e7c 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -311,7 +311,9 @@ class PregelLoop(LoopProtocol): ) -> Optional[PregelExecutableTask]: """Accept a PUSH from a task, potentially returning a new task to start.""" # don't start if we should interrupt *after* the original task - if should_interrupt(self.checkpoint, self.interrupt_after, [task]): + if self.interrupt_after and should_interrupt( + self.checkpoint, self.interrupt_after, [task] + ): self.to_interrupt.append(task) return if pushed := cast( @@ -333,7 +335,9 @@ class PregelLoop(LoopProtocol): ), ): # don't start if we should interrupt *before* the new task - if should_interrupt(self.checkpoint, self.interrupt_before, [pushed]): + if self.interrupt_before and should_interrupt( + self.checkpoint, self.interrupt_before, [pushed] + ): self.to_interrupt.append(pushed) return # produce debug output @@ -409,7 +413,7 @@ class PregelLoop(LoopProtocol): } ) # after execution, check if we should interrupt - if should_interrupt( + if self.interrupt_after and should_interrupt( self.checkpoint, self.interrupt_after, self.tasks.values() ): self.status = "interrupt_after" @@ -422,18 +426,6 @@ class PregelLoop(LoopProtocol): self.status = "out_of_steps" return False - # apply NULL writes - if null_writes := [ - w[1:] for w in self.checkpoint_pending_writes if w[0] == NULL_TASK_ID - ]: - mv_writes = apply_writes( - self.checkpoint, - self.channels, - [PregelTaskWrites((), INPUT, null_writes, [])], - self.checkpointer_get_next_version, - ) - for key, values in mv_writes.items(): - self._update_mv(key, values) # prepare next tasks self.tasks = prepare_next_tasks( self.checkpoint, @@ -493,7 +485,7 @@ class PregelLoop(LoopProtocol): return self.tick(input_keys=input_keys) # before execution, check if we should interrupt - if should_interrupt( + if self.interrupt_before and should_interrupt( self.checkpoint, self.interrupt_before, self.tasks.values() ): self.status = "interrupt_before" @@ -552,6 +544,18 @@ class PregelLoop(LoopProtocol): # save writes for tid, ws in writes.items(): self.put_writes(tid, ws) + # apply NULL writes + if null_writes := [ + w[1:] for w in self.checkpoint_pending_writes if w[0] == NULL_TASK_ID + ]: + mv_writes = apply_writes( + self.checkpoint, + self.channels, + [PregelTaskWrites((), INPUT, null_writes, [])], + self.checkpointer_get_next_version, + ) + for key, values in mv_writes.items(): + self._update_mv(key, values) # proceed past previous checkpoint if is_resuming: self.checkpoint["versions_seen"].setdefault(INTERRUPT, {}) diff --git a/libs/langgraph/langgraph/utils/fields.py b/libs/langgraph/langgraph/utils/fields.py index f4786cb34..009e5aee1 100644 --- a/libs/langgraph/langgraph/utils/fields.py +++ b/libs/langgraph/langgraph/utils/fields.py @@ -1,5 +1,5 @@ import dataclasses -from typing import Any, Optional, Type, Union +from typing import Any, Generator, Optional, Type, Union, get_type_hints from typing_extensions import Annotated, NotRequired, ReadOnly, Required, get_origin @@ -106,3 +106,44 @@ def get_field_default(name: str, type_: Any, schema: Type[Any]) -> Any: if _is_optional_type(type_): return None return ... + + +def get_enhanced_type_hints( + type: Type[Any], +) -> Generator[tuple[str, Any, Any, Optional[str]], None, None]: + """Attempt to extract default values and descriptions from provided type, used for config schema.""" + for name, typ in get_type_hints(type).items(): + default = None + description = None + + # Pydantic models + try: + if hasattr(type, "__fields__") and name in type.__fields__: + field = type.__fields__[name] + + if hasattr(field, "description") and field.description is not None: + description = field.description + + if hasattr(field, "default") and field.default is not None: + default = field.default + if ( + hasattr(default, "__class__") + and getattr(default.__class__, "__name__", "") + == "PydanticUndefinedType" + ): + default = None + + except (AttributeError, KeyError, TypeError): + pass + + # TypedDict, dataclass + try: + if hasattr(type, "__dict__"): + type_dict = getattr(type, "__dict__") + + if name in type_dict: + default = type_dict[name] + except (AttributeError, KeyError, TypeError): + pass + + yield name, typ, default, description diff --git a/libs/langgraph/langgraph/utils/runnable.py b/libs/langgraph/langgraph/utils/runnable.py index ccebba862..7cd6a85b9 100644 --- a/libs/langgraph/langgraph/utils/runnable.py +++ b/libs/langgraph/langgraph/utils/runnable.py @@ -404,12 +404,10 @@ class RunnableSeq(Runnable): config = patch_config( config, callbacks=run_manager.get_child(f"seq:step:{i+1}") ) - context = copy_context() - context.run(_set_config_context, config) if i == 0: - input = context.run(step.invoke, input, config, **kwargs) + input = step.invoke(input, config, **kwargs) else: - input = context.run(step.invoke, input, config) + input = step.invoke(input, config) # finish the root run except BaseException as e: run_manager.on_chain_error(e) @@ -443,16 +441,10 @@ class RunnableSeq(Runnable): config = patch_config( config, callbacks=run_manager.get_child(f"seq:step:{i+1}") ) - context = copy_context() - context.run(_set_config_context, config) if i == 0: - coro = step.ainvoke(input, config, **kwargs) + input = await step.ainvoke(input, config, **kwargs) else: - coro = step.ainvoke(input, config) - if ASYNCIO_ACCEPTS_CONTEXT: - input = await asyncio.create_task(coro, context=context) - else: - input = await asyncio.create_task(coro) + input = await step.ainvoke(input, config) # finish the root run except BaseException as e: await run_manager.on_chain_error(e) diff --git a/libs/langgraph/pyproject.toml b/libs/langgraph/pyproject.toml index cf29b2375..a5fa31c74 100644 --- a/libs/langgraph/pyproject.toml +++ b/libs/langgraph/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "langgraph" -version = "0.2.57" +version = "0.2.58" description = "Building stateful, multi-actor applications with LLMs" authors = [] license = "MIT" diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 91d7907d0..ad48108a9 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -7602,7 +7602,7 @@ def test_root_graph( content="result for query", name="search_api", tool_call_id="tool_call123", - id="00000000-0000-4000-8000-000000000033", + id="00000000-0000-4000-8000-000000000037", ) ] }, @@ -7625,7 +7625,7 @@ def test_root_graph( content="result for another", name="search_api", tool_call_id="tool_call456", - id="00000000-0000-4000-8000-000000000041", + id="00000000-0000-4000-8000-000000000045", ) ] }, @@ -8235,7 +8235,7 @@ def test_root_graph( "__root__": [ HumanMessage( content="what is weather in sf", - id="00000000-0000-4000-8000-000000000070", + id="00000000-0000-4000-8000-000000000078", ), AIMessage( content="", @@ -8255,7 +8255,7 @@ def test_root_graph( ), AIMessage(content="answer", id="ai2"), AIMessage( - content="an extra message", id="00000000-0000-4000-8000-000000000092" + content="an extra message", id="00000000-0000-4000-8000-000000000100" ), HumanMessage(content="what is weather in la"), ], @@ -12901,7 +12901,7 @@ def test_send_to_nested_graphs( metadata={ "step": 1, "source": "loop", - "writes": {"edit": None}, + "writes": None, "parents": {"": AnyStr()}, "thread_id": "1", "checkpoint_ns": AnyStr("generate_joke:"), @@ -12946,7 +12946,7 @@ def test_send_to_nested_graphs( metadata={ "step": 1, "source": "loop", - "writes": {"edit": None}, + "writes": None, "parents": {"": AnyStr()}, "thread_id": "1", "checkpoint_ns": AnyStr("generate_joke:"), @@ -14906,9 +14906,14 @@ def test_dict_mixed_return() -> None: assert graph.invoke({"foo": ""}) == {"foo": "ab"} -def test_command_with_static_breakpoints() -> None: +@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC) +def test_command_with_static_breakpoints( + request: pytest.FixtureRequest, checkpointer_name: str +) -> None: """Test that we can use Command to resume and update with static breakpoints.""" + checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}") + class State(TypedDict): """The graph state.""" @@ -14930,17 +14935,110 @@ def test_command_with_static_breakpoints() -> None: builder.add_edge(START, "node1") builder.add_edge("node1", "node2") - # A checkpointer must be enabled for interrupts to work! - checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node1"]) - - config = { - "configurable": { - "thread_id": uuid.uuid4(), - } - } + config = {"configurable": {"thread_id": str(uuid.uuid4())}} # Start the graph and interrupt at the first node graph.invoke({"foo": "abc"}, config) result = graph.invoke(Command(resume="node1"), config) assert result == {"foo": "abc|node-1|node-2"} + + +@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC) +def test_multistep_plan(request: pytest.FixtureRequest, checkpointer_name: str): + from langchain_core.messages import AnyMessage + + checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}") + + class State(TypedDict, total=False): + plan: list[Union[str, list[str]]] + messages: Annotated[list[AnyMessage], add_messages] + + def planner(state: State): + if state.get("plan") is None: + # create plan somehow + plan = ["step1", ["step2", "step3"], "step4"] + # pick the first step to execute next + first_step, *plan = plan + # put the rest of plan in state + return Command(goto=first_step, update={"plan": plan}) + elif state["plan"]: + # go to the next step of the plan + next_step, *next_plan = state["plan"] + return Command(goto=next_step, update={"plan": next_plan}) + else: + # the end of the plan + pass + + def step1(state: State): + return Command(goto="planner", update={"messages": [("human", "step1")]}) + + def step2(state: State): + return Command(goto="planner", update={"messages": [("human", "step2")]}) + + def step3(state: State): + return Command(goto="planner", update={"messages": [("human", "step3")]}) + + def step4(state: State): + return Command(goto="planner", update={"messages": [("human", "step4")]}) + + builder = StateGraph(State) + builder.add_node(planner) + builder.add_node(step1) + builder.add_node(step2) + builder.add_node(step3) + builder.add_node(step4) + builder.add_edge(START, "planner") + graph = builder.compile(checkpointer=checkpointer) + + config = {"configurable": {"thread_id": "1"}} + + assert graph.invoke({"messages": [("human", "start")]}, config) == { + "messages": [ + _AnyIdHumanMessage(content="start"), + _AnyIdHumanMessage(content="step1"), + _AnyIdHumanMessage(content="step2"), + _AnyIdHumanMessage(content="step3"), + _AnyIdHumanMessage(content="step4"), + ], + "plan": [], + } + + +@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC) +def test_command_goto_with_static_breakpoints( + request: pytest.FixtureRequest, checkpointer_name: str +) -> None: + """Use Command goto with static breakpoints.""" + + checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}") + + class State(TypedDict): + """The graph state.""" + + foo: Annotated[str, operator.add] + + def node1(state: State): + return { + "foo": "|node-1", + } + + def node2(state: State): + return { + "foo": "|node-2", + } + + builder = StateGraph(State) + builder.add_node("node1", node1) + builder.add_node("node2", node2) + builder.add_edge(START, "node1") + builder.add_edge("node1", "node2") + + graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node1"]) + + config = {"configurable": {"thread_id": str(uuid.uuid4())}} + + # Start the graph and interrupt at the first node + graph.invoke({"foo": "abc"}, config) + result = graph.invoke(Command(goto=["node2"]), config) + assert result == {"foo": "abc|node-1|node-2|node-2"} diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index b3317e5ae..cde54ae7c 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -13199,3 +13199,135 @@ async def test_interrupt_loop(checkpointer_name: str): ] == [ {"node": {"age": 19}}, ] + + +@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC) +async def test_command_with_static_breakpoints(checkpointer_name: str) -> None: + """Test that we can use Command to resume and update with static breakpoints.""" + + class State(TypedDict): + """The graph state.""" + + foo: str + + def node1(state: State): + return { + "foo": state["foo"] + "|node-1", + } + + def node2(state: State): + return { + "foo": state["foo"] + "|node-2", + } + + builder = StateGraph(State) + builder.add_node("node1", node1) + builder.add_node("node2", node2) + builder.add_edge(START, "node1") + builder.add_edge("node1", "node2") + + async with awith_checkpointer(checkpointer_name) as checkpointer: + graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node1"]) + config = {"configurable": {"thread_id": str(uuid.uuid4())}} + + # Start the graph and interrupt at the first node + await graph.ainvoke({"foo": "abc"}, config) + result = await graph.ainvoke(Command(update={"foo": "def"}), config) + assert result == {"foo": "def|node-1|node-2"} + + +@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC) +async def test_multistep_plan(checkpointer_name: str): + from langchain_core.messages import AnyMessage + + class State(TypedDict, total=False): + plan: list[Union[str, list[str]]] + messages: Annotated[list[AnyMessage], add_messages] + + def planner(state: State): + if state.get("plan") is None: + # create plan somehow + plan = ["step1", ["step2", "step3"], "step4"] + # pick the first step to execute next + first_step, *plan = plan + # put the rest of plan in state + return Command(goto=first_step, update={"plan": plan}) + elif state["plan"]: + # go to the next step of the plan + next_step, *next_plan = state["plan"] + return Command(goto=next_step, update={"plan": next_plan}) + else: + # the end of the plan + pass + + def step1(state: State): + return Command(goto="planner", update={"messages": [("human", "step1")]}) + + def step2(state: State): + return Command(goto="planner", update={"messages": [("human", "step2")]}) + + def step3(state: State): + return Command(goto="planner", update={"messages": [("human", "step3")]}) + + def step4(state: State): + return Command(goto="planner", update={"messages": [("human", "step4")]}) + + builder = StateGraph(State) + builder.add_node(planner) + builder.add_node(step1) + builder.add_node(step2) + builder.add_node(step3) + builder.add_node(step4) + builder.add_edge(START, "planner") + + async with awith_checkpointer(checkpointer_name) as checkpointer: + graph = builder.compile(checkpointer=checkpointer) + + config = {"configurable": {"thread_id": "1"}} + + assert await graph.ainvoke({"messages": [("human", "start")]}, config) == { + "messages": [ + _AnyIdHumanMessage(content="start"), + _AnyIdHumanMessage(content="step1"), + _AnyIdHumanMessage(content="step2"), + _AnyIdHumanMessage(content="step3"), + _AnyIdHumanMessage(content="step4"), + ], + "plan": [], + } + + +@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC) +async def test_command_goto_with_static_breakpoints(checkpointer_name: str) -> None: + """Use Command goto with static breakpoints.""" + + class State(TypedDict): + """The graph state.""" + + foo: Annotated[str, operator.add] + + def node1(state: State): + return { + "foo": "|node-1", + } + + def node2(state: State): + return { + "foo": "|node-2", + } + + builder = StateGraph(State) + builder.add_node("node1", node1) + builder.add_node("node2", node2) + builder.add_edge(START, "node1") + builder.add_edge("node1", "node2") + + async with awith_checkpointer(checkpointer_name) as checkpointer: + graph = builder.compile(checkpointer=checkpointer, interrupt_before=["node1"]) + + config = {"configurable": {"thread_id": str(uuid.uuid4())}} + + # Start the graph and interrupt at the first node + await graph.ainvoke({"foo": "abc"}, config) + result = await graph.ainvoke(Command(goto=["node2"]), config) + assert result == {"foo": "abc|node-1|node-2|node-2"} diff --git a/libs/langgraph/tests/test_utils.py b/libs/langgraph/tests/test_utils.py index e8ea94fff..616f1a78f 100644 --- a/libs/langgraph/tests/test_utils.py +++ b/libs/langgraph/tests/test_utils.py @@ -21,7 +21,11 @@ from typing_extensions import Annotated, NotRequired, Required from langgraph.graph import END, StateGraph from langgraph.graph.graph import CompiledGraph -from langgraph.utils.fields import _is_optional_type, get_field_default +from langgraph.utils.fields import ( + _is_optional_type, + get_enhanced_type_hints, + get_field_default, +) from langgraph.utils.runnable import is_async_callable, is_async_generator pytestmark = pytest.mark.anyio @@ -227,3 +231,57 @@ def test_is_required(): assert get_field_default("val_12", gcannos["val_12"], MyGrandChildDict) is None assert get_field_default("val_9", gcannos["val_9"], MyGrandChildDict) is None assert get_field_default("val_13", gcannos["val_13"], MyGrandChildDict) == ... + + +def test_enhanced_type_hints() -> None: + from dataclasses import dataclass + from typing import Annotated + + from pydantic import BaseModel, Field + + class MyTypedDict(TypedDict): + val_1: str + val_2: int = 42 + val_3: str = "default" + + hints = list(get_enhanced_type_hints(MyTypedDict)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, None) + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", None) + + @dataclass + class MyDataclass: + val_1: str + val_2: int = 42 + val_3: str = "default" + + hints = list(get_enhanced_type_hints(MyDataclass)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, None) + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", None) + + class MyPydanticModel(BaseModel): + val_1: str + val_2: int = 42 + val_3: str = Field(default="default", description="A description") + + hints = list(get_enhanced_type_hints(MyPydanticModel)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, None) + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", "A description") + + class MyPydanticModelWithAnnotated(BaseModel): + val_1: Annotated[str, Field(description="A description")] + val_2: Annotated[int, Field(default=42)] + val_3: Annotated[ + str, Field(default="default", description="Another description") + ] + + hints = list(get_enhanced_type_hints(MyPydanticModelWithAnnotated)) + assert len(hints) == 3 + assert hints[0] == ("val_1", str, None, "A description") + assert hints[1] == ("val_2", int, 42, None) + assert hints[2] == ("val_3", str, "default", "Another description") diff --git a/libs/sdk-js/package.json b/libs/sdk-js/package.json index 02628e6dc..8e3e061b2 100644 --- a/libs/sdk-js/package.json +++ b/libs/sdk-js/package.json @@ -1,6 +1,6 @@ { "name": "@langchain/langgraph-sdk", - "version": "0.0.31", + "version": "0.0.32", "description": "Client library for interacting with the LangGraph API", "type": "module", "packageManager": "yarn@1.22.19",