mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-22 09:35:07 +02:00
Merge branch 'main' into eugene/document_interrupt
This commit is contained in:
Eugene Yurtsev
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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 []
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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, {})
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user