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

Merge branch 'main' into wfh/optional_types

Browse Source
This commit is contained in:
William Fu-Hinthorn
2024-08-30 16:40:56 -07:00
parent 55263c640f 7d6a302858
commit 85e827ae85
33 changed files with 2240 additions and 6925 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/customer-support/customer-support.ipynb
+42 -38
View File
@@ -99,46 +99,50 @@
" # Backup - we will use this to \"reset\" our DB in each section\n",
" shutil.copy(local_file, backup_file)\n",
"# Convert the flights to present time for our tutorial\n",
"conn = sqlite3.connect(local_file)\n",
"cursor = conn.cursor()\n",
"def update_dates(file):\n",
" shutil.copy(backup_file, file)\n",
" conn = sqlite3.connect(file)\n",
" cursor = conn.cursor()\n",
"\n",
"tables = pd.read_sql(\n",
" \"SELECT name FROM sqlite_master WHERE type='table';\", conn\n",
").name.tolist()\n",
"tdf = {}\n",
"for t in tables:\n",
" tdf[t] = pd.read_sql(f\"SELECT * from {t}\", conn)\n",
" tables = pd.read_sql(\n",
" \"SELECT name FROM sqlite_master WHERE type='table';\", conn\n",
" ).name.tolist()\n",
" tdf = {}\n",
" for t in tables:\n",
" tdf[t] = pd.read_sql(f\"SELECT * from {t}\", conn)\n",
"\n",
"example_time = pd.to_datetime(\n",
" tdf[\"flights\"][\"actual_departure\"].replace(\"\\\\N\", pd.NaT)\n",
").max()\n",
"current_time = pd.to_datetime(\"now\").tz_localize(example_time.tz)\n",
"time_diff = current_time - example_time\n",
" example_time = pd.to_datetime(\n",
" tdf[\"flights\"][\"actual_departure\"].replace(\"\\\\N\", pd.NaT)\n",
" ).max()\n",
" current_time = pd.to_datetime(\"now\").tz_localize(example_time.tz)\n",
" time_diff = current_time - example_time\n",
"\n",
"tdf[\"bookings\"][\"book_date\"] = (\n",
" pd.to_datetime(tdf[\"bookings\"][\"book_date\"].replace(\"\\\\N\", pd.NaT), utc=True)\n",
" + time_diff\n",
")\n",
"\n",
"datetime_columns = [\n",
" \"scheduled_departure\",\n",
" \"scheduled_arrival\",\n",
" \"actual_departure\",\n",
" \"actual_arrival\",\n",
"]\n",
"for column in datetime_columns:\n",
" tdf[\"flights\"][column] = (\n",
" pd.to_datetime(tdf[\"flights\"][column].replace(\"\\\\N\", pd.NaT)) + time_diff\n",
" tdf[\"bookings\"][\"book_date\"] = (\n",
" pd.to_datetime(tdf[\"bookings\"][\"book_date\"].replace(\"\\\\N\", pd.NaT), utc=True)\n",
" + time_diff\n",
" )\n",
"\n",
"for table_name, df in tdf.items():\n",
" df.to_sql(table_name, conn, if_exists=\"replace\", index=False)\n",
"del df\n",
"del tdf\n",
"conn.commit()\n",
"conn.close()\n",
" datetime_columns = [\n",
" \"scheduled_departure\",\n",
" \"scheduled_arrival\",\n",
" \"actual_departure\",\n",
" \"actual_arrival\",\n",
" ]\n",
" for column in datetime_columns:\n",
" tdf[\"flights\"][column] = (\n",
" pd.to_datetime(tdf[\"flights\"][column].replace(\"\\\\N\", pd.NaT)) + time_diff\n",
" )\n",
"\n",
"db = local_file # We'll be using this local file as our DB in this tutorial"
" for table_name, df in tdf.items():\n",
" df.to_sql(table_name, conn, if_exists=\"replace\", index=False)\n",
" del df\n",
" del tdf\n",
" conn.commit()\n",
" conn.close()\n",
"\n",
" return file\n",
"\n",
"db = update_dates(local_file)"
]
},
{
@@ -1750,7 +1754,7 @@
"]\n",
"\n",
"# Update with the backup file so we can restart from the original place in each section\n",
"shutil.copy(backup_file, db)\n",
"db = update_dates(db)\n",
"thread_id = str(uuid.uuid4())\n",
"\n",
"config = {\n",
@@ -2304,7 +2308,7 @@
"import uuid\n",
"\n",
"# Update with the backup file so we can restart from the original place in each section\n",
"shutil.copy(backup_file, db)\n",
"db = update_dates(db)\n",
"thread_id = str(uuid.uuid4())\n",
"\n",
"config = {\n",
@@ -2908,7 +2912,7 @@
"import uuid\n",
"\n",
"# Update with the backup file so we can restart from the original place in each section\n",
"shutil.copy(backup_file, db)\n",
"db = update_dates(db)\n",
"thread_id = str(uuid.uuid4())\n",
"\n",
"config = {\n",
@@ -4330,7 +4334,7 @@
"import uuid\n",
"\n",
"# Update with the backup file so we can restart from the original place in each section\n",
"shutil.copy(backup_file, db)\n",
"db = update_dates(db)\n",
"thread_id = str(uuid.uuid4())\n",
"\n",
"config = {\n",
examples/rag/langgraph_agentic_rag.ipynb
+1 -1
View File
@@ -126,7 +126,7 @@
"source": [
"## Agent state\n",
" \n",
"We will defined a graph.\n",
"We will define a graph.\n",
"\n",
"A `state` object that it passes around to each node.\n",
"\n",
examples/subgraph.ipynb
+648 -682
View File
File diff suppressed because one or more lines are too long
libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py
+2 -2
View File
@@ -76,14 +76,14 @@ select
) as channel_values,
(
select
array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob])
array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob] order by cw.task_id, cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
and cw.checkpoint_id = checkpoints.checkpoint_id
) as pending_writes,
(
select array_agg(array[cw.type::bytea, cw.blob])
select array_agg(array[cw.type::bytea, cw.blob] order by cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
libs/checkpoint-postgres/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint-postgres"
version = "1.0.4"
version = "1.0.5"
description = "Library with a Postgres implementation of LangGraph checkpoint saver."
authors = []
license = "MIT"
libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py
+2 -2
View File
@@ -244,7 +244,7 @@ class SqliteSaver(BaseCheckpointSaver):
}
# find any pending writes
cur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(
str(config["configurable"]["thread_id"]),
checkpoint_ns,
@@ -331,7 +331,7 @@ class SqliteSaver(BaseCheckpointSaver):
metadata,
) in cur:
wcur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(thread_id, checkpoint_ns, checkpoint_id),
)
yield CheckpointTuple(
libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py
+2 -2
View File
@@ -278,7 +278,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver):
}
# find any pending writes
await cur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(
str(config["configurable"]["thread_id"]),
checkpoint_ns,
@@ -348,7 +348,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver):
metadata,
) in cur:
await wcur.execute(
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ?",
"SELECT task_id, channel, type, value FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? ORDER BY task_id, idx",
(thread_id, checkpoint_ns, checkpoint_id),
)
yield CheckpointTuple(
libs/checkpoint-sqlite/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint-sqlite"
version = "1.0.0"
version = "1.0.1"
description = "Library with a SQLite implementation of LangGraph checkpoint saver."
authors = []
license = "MIT"
libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py
+25 -16
View File
@@ -52,11 +52,11 @@ class JsonPlusSerializer(SerializerProtocol):
return obj.to_json()
elif hasattr(obj, "model_dump") and callable(obj.model_dump):
return self._encode_constructor_args(
obj.__class__, method="model_construct", kwargs=obj.model_dump()
obj.__class__, method=[None, "model_construct"], kwargs=obj.model_dump()
)
elif hasattr(obj, "dict") and callable(obj.dict):
return self._encode_constructor_args(
obj.__class__, method="construct", kwargs=obj.dict()
obj.__class__, method=[None, "construct"], kwargs=obj.dict()
)
elif isinstance(obj, pathlib.Path):
return self._encode_constructor_args(pathlib.Path, args=obj.parts)
@@ -136,21 +136,30 @@ class JsonPlusSerializer(SerializerProtocol):
# Import class
cls = getattr(mod, name)
# Instantiate class
if value["method"] is not None:
method = getattr(cls, value["method"])
if isinstance(value["method"], str):
methods = [getattr(cls, value["method"])]
elif isinstance(value["method"], list):
methods = [
cls if method is None else getattr(cls, method)
for method in value["method"]
]
else:
method = cls
if isclass(method) and issubclass(method, BaseException):
return None
if value["args"] and value["kwargs"]:
return method(*value["args"], **value["kwargs"])
elif value["args"]:
return method(*value["args"])
elif value["kwargs"]:
return method(**value["kwargs"])
else:
return method()
except (ImportError, AttributeError, TypeError):
methods = [cls]
for method in methods:
try:
if isclass(method) and issubclass(method, BaseException):
return None
if value["args"] and value["kwargs"]:
return method(*value["args"], **value["kwargs"])
elif value["args"]:
return method(*value["args"])
elif value["kwargs"]:
return method(**value["kwargs"])
else:
return method()
except Exception:
continue
except Exception:
return None
return LC_REVIVER(value)
libs/checkpoint/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint"
version = "1.0.6"
version = "1.0.8"
description = "Library with base interfaces for LangGraph checkpoint savers."
authors = []
license = "MIT"
libs/checkpoint/tests/test_jsonplus.py
+1 -1
View File
@@ -122,7 +122,7 @@ def test_serde_jsonplus() -> None:
assert dumped == (
"json",
b"""{"path": {"lc": 2, "type": "constructor", "id": ["pathlib", "Path"], "method": null, "args": ["foo", "bar"], "kwargs": {}}, "re": {"lc": 2, "type": "constructor", "id": ["re", "compile"], "method": null, "args": ["foo", 48], "kwargs": {}}, "decimal": {"lc": 2, "type": "constructor", "id": ["decimal", "Decimal"], "method": null, "args": ["1.10101"], "kwargs": {}}, "ip4": {"lc": 2, "type": "constructor", "id": ["ipaddress", "IPv4Address"], "method": null, "args": ["192.168.0.1"], "kwargs": {}}, "deque": {"lc": 2, "type": "constructor", "id": ["collections", "deque"], "method": null, "args": [[1, 2, 3]], "kwargs": {}}, "tzn": {"lc": 2, "type": "constructor", "id": ["zoneinfo", "ZoneInfo"], "method": null, "args": ["America/New_York"], "kwargs": {}}, "date": {"lc": 2, "type": "constructor", "id": ["datetime", "date"], "method": null, "args": [2024, 4, 19], "kwargs": {}}, "time": {"lc": 2, "type": "constructor", "id": ["datetime", "time"], "method": null, "args": [23, 4, 57, 51022, {"lc": 2, "type": "constructor", "id": ["datetime", "timezone"], "method": null, "args": [{"lc": 2, "type": "constructor", "id": ["datetime", "timedelta"], "method": null, "args": [0, 86340, 0], "kwargs": {}}], "kwargs": {}}], "kwargs": {"fold": 0}}, "uid": {"lc": 2, "type": "constructor", "id": ["uuid", "UUID"], "method": null, "args": ["00000000000000000000000000000001"], "kwargs": {}}, "timestamp": {"lc": 2, "type": "constructor", "id": ["datetime", "datetime"], "method": "fromisoformat", "args": ["2024-04-19T23:04:57.051022+23:59"], "kwargs": {}}, "my_slotted_class": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclassWSlots"], "method": null, "args": [], "kwargs": {"foo": "bar", "bar": 2}}, "my_dataclass": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclass"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_enum": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyEnum"], "method": null, "args": ["foo"], "kwargs": {}}, "my_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyPydantic"], "method": "model_construct", "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_funny_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyFunnyPydantic"], "method": "construct", "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "person": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "Person"], "method": null, "args": [], "kwargs": {"name": "foo"}}, "a_bool": true, "a_none": null, "a_str": "foo", "a_str_nuc": "foo\\u0000", "a_str_uc": "foo \xe2\x9b\xb0\xef\xb8\x8f", "a_str_ucuc": "foo \xe2\x9b\xb0\xef\xb8\x8f\\u0000", "a_str_ucucuc": "foo \\\\u26f0\\\\ufe0f", "text": ["Hello", "Python", "Surrogate", "Example", "String", "With", "Surrogates", "Embedded", "In", "The", "Text", "\xe6\x94\xb6\xe8\x8a\xb1\xf0\x9f\x99\x84\xc2\xb7\xe5\x88\xb0"], "an_int": 1, "a_float": 1.1, "runnable_map": {"lc": 1, "type": "constructor", "id": ["langchain", "schema", "runnable", "RunnableParallel"], "kwargs": {"steps__": {}}, "name": "RunnableParallel<>", "graph": {"nodes": [{"id": 0, "type": "schema", "data": "Parallel<>Input"}, {"id": 1, "type": "schema", "data": "Parallel<>Output"}], "edges": []}}}""",
b"""{"path": {"lc": 2, "type": "constructor", "id": ["pathlib", "Path"], "method": null, "args": ["foo", "bar"], "kwargs": {}}, "re": {"lc": 2, "type": "constructor", "id": ["re", "compile"], "method": null, "args": ["foo", 48], "kwargs": {}}, "decimal": {"lc": 2, "type": "constructor", "id": ["decimal", "Decimal"], "method": null, "args": ["1.10101"], "kwargs": {}}, "ip4": {"lc": 2, "type": "constructor", "id": ["ipaddress", "IPv4Address"], "method": null, "args": ["192.168.0.1"], "kwargs": {}}, "deque": {"lc": 2, "type": "constructor", "id": ["collections", "deque"], "method": null, "args": [[1, 2, 3]], "kwargs": {}}, "tzn": {"lc": 2, "type": "constructor", "id": ["zoneinfo", "ZoneInfo"], "method": null, "args": ["America/New_York"], "kwargs": {}}, "date": {"lc": 2, "type": "constructor", "id": ["datetime", "date"], "method": null, "args": [2024, 4, 19], "kwargs": {}}, "time": {"lc": 2, "type": "constructor", "id": ["datetime", "time"], "method": null, "args": [23, 4, 57, 51022, {"lc": 2, "type": "constructor", "id": ["datetime", "timezone"], "method": null, "args": [{"lc": 2, "type": "constructor", "id": ["datetime", "timedelta"], "method": null, "args": [0, 86340, 0], "kwargs": {}}], "kwargs": {}}], "kwargs": {"fold": 0}}, "uid": {"lc": 2, "type": "constructor", "id": ["uuid", "UUID"], "method": null, "args": ["00000000000000000000000000000001"], "kwargs": {}}, "timestamp": {"lc": 2, "type": "constructor", "id": ["datetime", "datetime"], "method": "fromisoformat", "args": ["2024-04-19T23:04:57.051022+23:59"], "kwargs": {}}, "my_slotted_class": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclassWSlots"], "method": null, "args": [], "kwargs": {"foo": "bar", "bar": 2}}, "my_dataclass": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyDataclass"], "method": null, "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_enum": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyEnum"], "method": null, "args": ["foo"], "kwargs": {}}, "my_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyPydantic"], "method": [null, "model_construct"], "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "my_funny_pydantic": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "MyFunnyPydantic"], "method": [null, "construct"], "args": [], "kwargs": {"foo": "foo", "bar": 1}}, "person": {"lc": 2, "type": "constructor", "id": ["tests", "test_jsonplus", "Person"], "method": null, "args": [], "kwargs": {"name": "foo"}}, "a_bool": true, "a_none": null, "a_str": "foo", "a_str_nuc": "foo\\u0000", "a_str_uc": "foo \xe2\x9b\xb0\xef\xb8\x8f", "a_str_ucuc": "foo \xe2\x9b\xb0\xef\xb8\x8f\\u0000", "a_str_ucucuc": "foo \\\\u26f0\\\\ufe0f", "text": ["Hello", "Python", "Surrogate", "Example", "String", "With", "Surrogates", "Embedded", "In", "The", "Text", "\xe6\x94\xb6\xe8\x8a\xb1\xf0\x9f\x99\x84\xc2\xb7\xe5\x88\xb0"], "an_int": 1, "a_float": 1.1, "runnable_map": {"lc": 1, "type": "constructor", "id": ["langchain", "schema", "runnable", "RunnableParallel"], "kwargs": {"steps__": {}}, "name": "RunnableParallel<>", "graph": {"nodes": [{"id": 0, "type": "schema", "data": "Parallel<>Input"}, {"id": 1, "type": "schema", "data": "Parallel<>Output"}], "edges": []}}}""",
)
assert serde.loads_typed(dumped) == {
libs/cli/examples/poetry.lock
Generated
+4 -2080
View File
File diff suppressed because it is too large Load Diff
libs/cli/examples/pyproject.toml
-1
View File
@@ -9,7 +9,6 @@ package-mode = false
[tool.poetry.dependencies]
python = "^3.9.0,<4.0"
jupyter = "^1.0.0"
langgraph-cli = {path = "../../cli", develop = true}
langgraph-sdk = {path = "../../sdk-py", develop = true}
libs/langgraph/Makefile
+3 -5
View File
@@ -18,20 +18,18 @@ start-postgres:
docker compose -f tests/compose-postgres.yml up -V --force-recreate --wait
stop-postgres:
docker compose -f tests/compose-postgres.yml down
docker compose -f tests/compose-postgres.yml down -v
TEST_PATH ?= .
test:
make start-postgres; \
poetry run pytest $(TEST_PATH); \
make start-postgres && poetry run pytest $(TEST_PATH); \
EXIT_CODE=$$?; \
make stop-postgres; \
exit $$EXIT_CODE
test_watch:
make start-postgres; \
poetry run ptw . -- --ff -v -x -n auto --dist worksteal --snapshot-update --tb short $(TEST_PATH); \
make start-postgres && poetry run ptw . -- --ff -v -x -n auto --dist worksteal --snapshot-update --tb short $(TEST_PATH); \
EXIT_CODE=$$?; \
make stop-postgres; \
exit $$EXIT_CODE
libs/langgraph/langgraph/graph/graph.py
+6 -2
View File
@@ -484,6 +484,10 @@ class CompiledGraph(Pregel):
START: graph.add_node(self.get_input_schema(config), START)
}
end_nodes: dict[str, DrawableNode] = {}
if xray:
subgraphs = dict(self.get_subgraphs())
else:
subgraphs = {}
def add_edge(
start: str, end: str, label: Optional[str] = None, conditional: bool = False
@@ -503,11 +507,11 @@ class CompiledGraph(Pregel):
metadata["__interrupt"] = "after"
if xray:
subgraph = (
node.get_graph(
subgraphs[key].get_graph(
config=config,
xray=xray - 1 if isinstance(xray, int) and xray > 0 else xray,
)
if isinstance(node, CompiledGraph)
if key in subgraphs
else node.get_graph(config=config)
)
subgraph.trim_first_node()
libs/langgraph/langgraph/pregel/__init__.py
+111 -62
View File
@@ -28,6 +28,7 @@ from langchain_core.load.dump import dumpd
from langchain_core.pydantic_v1 import BaseModel, Field, root_validator
from langchain_core.runnables import (
Runnable,
RunnableLambda,
RunnableSequence,
RunnableSerializable,
)
@@ -43,6 +44,7 @@ from langchain_core.runnables.config import (
from langchain_core.runnables.utils import (
ConfigurableFieldSpec,
create_model,
get_function_nonlocals,
get_unique_config_specs,
)
from langchain_core.tracers._streaming import _StreamingCallbackHandler
@@ -64,6 +66,7 @@ from langgraph.constants import (
CONFIG_KEY_RESUMING,
CONFIG_KEY_SEND,
CONFIG_KEY_STREAM,
CONFIG_KEY_TASK_ID,
ERROR,
INTERRUPT,
NS_END,
@@ -77,6 +80,7 @@ from langgraph.pregel.algo import (
local_write,
prepare_next_tasks,
)
from langgraph.pregel.config import patch_checkpoint_map, patch_configurable
from langgraph.pregel.debug import (
print_step_checkpoint,
print_step_tasks,
@@ -100,6 +104,7 @@ from langgraph.pregel.utils import (
from langgraph.pregel.validate import validate_graph, validate_keys
from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
from langgraph.store.base import BaseStore
from langgraph.utils import RunnableCallable
WriteValue = Union[
Runnable[Input, Output],
@@ -354,13 +359,26 @@ class Pregel(
for name, node in self.nodes.items():
# find the subgraph, if any
graph: Optional[Pregel] = None
if isinstance(node.bound, Pregel):
graph = node.bound
elif isinstance(node.bound, RunnableSequence):
for runnable in node.bound.steps:
if isinstance(runnable, Pregel):
graph = runnable
break
candidates = [node.bound]
for candidate in candidates:
if isinstance(candidate, Pregel):
graph = candidate
break
elif isinstance(candidate, RunnableSequence):
candidates.extend(candidate.steps)
elif isinstance(candidate, RunnableLambda):
candidates.extend(candidate.deps)
elif isinstance(candidate, RunnableCallable):
if candidate.func is not None:
candidates.extend(
nl.__self__ if hasattr(nl, "__self__") else nl
for nl in get_function_nonlocals(candidate.func)
)
if candidate.afunc is not None:
candidates.extend(
nl.__self__ if hasattr(nl, "__self__") else nl
for nl in get_function_nonlocals(candidate.afunc)
)
# if found, yield recursively
if graph:
yield name, graph
@@ -442,7 +460,7 @@ class Pregel(
return StateSnapshot(
read_channels(channels, self.stream_channels_asis),
tuple(t.name for t in next_tasks),
saved.config,
patch_checkpoint_map(saved.config, saved.metadata),
saved.metadata,
saved.checkpoint["ts"],
saved.parent_config,
@@ -518,7 +536,7 @@ class Pregel(
return StateSnapshot(
read_channels(channels, self.stream_channels_asis),
tuple(t.name for t in next_tasks),
saved.config,
patch_checkpoint_map(saved.config, saved.metadata),
saved.metadata,
saved.checkpoint["ts"],
saved.parent_config,
@@ -546,12 +564,9 @@ class Pregel(
for name, pregel in self.get_subgraphs(recurse=True):
if name == recast_checkpoint_ns:
return pregel.get_state(
{
"configurable": {
**config["configurable"],
CONFIG_KEY_CHECKPOINTER: checkpointer,
}
},
patch_configurable(
config, {CONFIG_KEY_CHECKPOINTER: checkpointer}
),
subgraphs=subgraphs,
)
else:
@@ -584,12 +599,9 @@ class Pregel(
async for name, pregel in self.aget_subgraphs(recurse=True):
if name == recast_checkpoint_ns:
return await pregel.aget_state(
{
"configurable": {
**config["configurable"],
CONFIG_KEY_CHECKPOINTER: checkpointer,
}
},
patch_configurable(
config, {CONFIG_KEY_CHECKPOINTER: checkpointer}
),
subgraphs=subgraphs,
)
else:
@@ -627,12 +639,9 @@ class Pregel(
for name, pregel in self.get_subgraphs(recurse=True):
if name == recast_checkpoint_ns:
yield from pregel.get_state_history(
{
"configurable": {
**config["configurable"],
CONFIG_KEY_CHECKPOINTER: checkpointer,
}
},
patch_configurable(
config, {CONFIG_KEY_CHECKPOINTER: checkpointer}
),
filter=filter,
before=before,
limit=limit,
@@ -678,12 +687,9 @@ class Pregel(
async for name, pregel in self.aget_subgraphs(recurse=True):
if name == recast_checkpoint_ns:
async for state in pregel.aget_state_history(
{
"configurable": {
**config["configurable"],
CONFIG_KEY_CHECKPOINTER: checkpointer,
}
},
patch_configurable(
config, {CONFIG_KEY_CHECKPOINTER: checkpointer}
),
filter=filter,
before=before,
limit=limit,
@@ -717,36 +723,51 @@ class Pregel(
node `as_node`. If `as_node` is not provided, it will be set to the last node
that updated the state, if not ambiguous.
"""
if not self.checkpointer:
checkpointer: Optional[BaseCheckpointSaver] = config["configurable"].get(
CONFIG_KEY_CHECKPOINTER, self.checkpointer
)
if not checkpointer:
raise ValueError("No checkpointer set")
# delegate to subgraph
if (
checkpoint_ns := config["configurable"].get("checkpoint_ns", "")
) and CONFIG_KEY_CHECKPOINTER not in config["configurable"]:
# remove task_ids from checkpoint_ns
recast_checkpoint_ns = NS_SEP.join(
part.split(NS_END)[0] for part in checkpoint_ns.split(NS_SEP)
)
# find the subgraph with the matching name
for name, pregel in self.get_subgraphs(recurse=True):
if name == recast_checkpoint_ns:
return pregel.update_state(
patch_configurable(
config, {CONFIG_KEY_CHECKPOINTER: checkpointer}
),
values,
as_node,
)
else:
raise ValueError(f"Subgraph {recast_checkpoint_ns} not found")
# get last checkpoint
config = merge_configs(self.config, config) if self.config else config
saved = self.checkpointer.get_tuple(config)
saved = checkpointer.get_tuple(config)
checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
checkpoint_previous_versions = (
saved.checkpoint["channel_versions"].copy() if saved else {}
)
step = saved.metadata.get("step", -1) if saved else -1
# merge configurable fields with previous checkpoint config
checkpoint_config = {
**config,
"configurable": {
**config["configurable"],
# TODO: add proper support for updating nested subgraph state
"checkpoint_ns": "",
},
}
checkpoint_config = patch_configurable(
config,
{"checkpoint_ns": config["configurable"].get("checkpoint_ns", "")},
)
if saved:
checkpoint_config = {
"configurable": {
**config.get("configurable", {}),
**saved.config["configurable"],
}
}
checkpoint_config = patch_configurable(config, saved.config["configurable"])
# find last node that updated the state, if not provided
if values is None and as_node is None:
return self.checkpointer.put(
next_config = checkpointer.put(
checkpoint_config,
create_checkpoint(checkpoint, None, step),
{
@@ -757,6 +778,7 @@ class Pregel(
},
{},
)
return patch_checkpoint_map(next_config, saved.metadata if saved else None)
elif as_node is None and not any(
v for vv in checkpoint["versions_seen"].values() for v in vv.values()
):
@@ -798,6 +820,7 @@ class Pregel(
None,
[INTERRUPT],
None,
None,
str(uuid5(UUID(checkpoint["id"]), INTERRUPT)),
)
# execute task
@@ -830,13 +853,13 @@ class Pregel(
)
# save task writes
if saved:
self.checkpointer.put_writes(checkpoint_config, task.writes, task.id)
checkpointer.put_writes(checkpoint_config, task.writes, task.id)
# apply to checkpoint and save
assert not apply_writes(
checkpoint, channels, [task], self.checkpointer.get_next_version
checkpoint, channels, [task], checkpointer.get_next_version
), "Can't write to SharedValues from update_state"
checkpoint = create_checkpoint(checkpoint, channels, step + 1)
return self.checkpointer.put(
next_config = checkpointer.put(
checkpoint_config,
checkpoint,
{
@@ -849,6 +872,7 @@ class Pregel(
checkpoint_previous_versions, checkpoint["channel_versions"]
),
)
return patch_checkpoint_map(next_config, saved.metadata if saved else None)
async def aupdate_state(
self,
@@ -856,12 +880,36 @@ class Pregel(
values: dict[str, Any] | Any,
as_node: Optional[str] = None,
) -> RunnableConfig:
if not self.checkpointer:
checkpointer: Optional[BaseCheckpointSaver] = config["configurable"].get(
CONFIG_KEY_CHECKPOINTER, self.checkpointer
)
if not checkpointer:
raise ValueError("No checkpointer set")
# delegate to subgraph
if (
checkpoint_ns := config["configurable"].get("checkpoint_ns", "")
) and CONFIG_KEY_CHECKPOINTER not in config["configurable"]:
# remove task_ids from checkpoint_ns
recast_checkpoint_ns = NS_SEP.join(
part.split(NS_END)[0] for part in checkpoint_ns.split(NS_SEP)
)
# find the subgraph with the matching name
async for name, pregel in self.aget_subgraphs(recurse=True):
if name == recast_checkpoint_ns:
return await pregel.aupdate_state(
patch_configurable(
config, {CONFIG_KEY_CHECKPOINTER: checkpointer}
),
values,
as_node,
)
else:
raise ValueError(f"Subgraph {recast_checkpoint_ns} not found")
# get last checkpoint
config = merge_configs(self.config, config) if self.config else config
saved = await self.checkpointer.aget_tuple(config)
saved = await checkpointer.aget_tuple(config)
checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
checkpoint_previous_versions = (
saved.checkpoint["channel_versions"].copy() if saved else {}
@@ -885,7 +933,7 @@ class Pregel(
}
# find last node that updated the state, if not provided
if values is None and as_node is None:
return await self.checkpointer.aput(
next_config = await checkpointer.aput(
checkpoint_config,
create_checkpoint(checkpoint, None, step),
{
@@ -896,6 +944,7 @@ class Pregel(
},
{},
)
return patch_checkpoint_map(next_config, saved.metadata if saved else None)
elif as_node is None and not saved:
if (
isinstance(self.input_channels, str)
@@ -935,6 +984,7 @@ class Pregel(
None,
[INTERRUPT],
None,
None,
str(uuid5(UUID(checkpoint["id"]), INTERRUPT)),
)
# execute task
@@ -967,15 +1017,13 @@ class Pregel(
)
# save task writes
if saved:
await self.checkpointer.aput_writes(
checkpoint_config, task.writes, task.id
)
await checkpointer.aput_writes(checkpoint_config, task.writes, task.id)
# apply to checkpoint and save
assert not apply_writes(
checkpoint, channels, [task], self.checkpointer.get_next_version
checkpoint, channels, [task], checkpointer.get_next_version
), "Can't write to SharedValues from update_state"
checkpoint = create_checkpoint(checkpoint, channels, step + 1)
return await self.checkpointer.aput(
next_config = await checkpointer.aput(
checkpoint_config,
checkpoint,
{
@@ -988,6 +1036,7 @@ class Pregel(
checkpoint_previous_versions, checkpoint["channel_versions"]
),
)
return patch_checkpoint_map(next_config, saved.metadata if saved else None)
def _defaults(
self,
@@ -1016,7 +1065,7 @@ class Pregel(
stream_mode = stream_mode if stream_mode is not None else self.stream_mode
if not isinstance(stream_mode, list):
stream_mode = [stream_mode]
if CONFIG_KEY_READ in config.get("configurable", {}):
if CONFIG_KEY_TASK_ID in config.get("configurable", {}):
# if being called as a node in another graph, always use values mode
stream_mode = ["values"]
if CONFIG_KEY_CHECKPOINTER in config.get("configurable", {}):
libs/langgraph/langgraph/pregel/algo.py
+18 -14
View File
@@ -303,6 +303,7 @@ def prepare_next_tasks(
if node := proc.get_node():
managed.replace_runtime_placeholders(step, packet.arg)
writes = deque()
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
tasks.append(
PregelExecutableTask(
packet.node,
@@ -351,11 +352,12 @@ def prepare_next_tasks(
},
CONFIG_KEY_RESUMING: is_resuming,
"checkpoint_id": None,
"checkpoint_ns": f"{checkpoint_ns}:{task_id}",
"checkpoint_ns": task_checkpoint_ns,
},
),
triggers,
proc.retry_policy,
None,
task_id,
)
)
@@ -382,7 +384,7 @@ def prepare_next_tasks(
try:
val = next(
_proc_input(
step, name, proc, managed, channels, for_execution=for_execution
step, proc, managed, channels, for_execution=for_execution
)
)
except StopIteration:
@@ -406,6 +408,7 @@ def prepare_next_tasks(
if for_execution:
if node := proc.get_node():
writes = deque()
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
tasks.append(
PregelExecutableTask(
name,
@@ -455,11 +458,12 @@ def prepare_next_tasks(
parent_ns: checkpoint["id"],
},
CONFIG_KEY_RESUMING: is_resuming,
"checkpoint_ns": f"{checkpoint_ns}:{task_id}",
"checkpoint_ns": task_checkpoint_ns,
},
),
triggers,
proc.retry_policy,
None,
task_id,
)
)
@@ -470,7 +474,6 @@ def prepare_next_tasks(
def _proc_input(
step: int,
name: str,
proc: PregelNode,
managed: ManagedValueMapping,
channels: Mapping[str, BaseChannel],
@@ -481,16 +484,17 @@ def _proc_input(
# then invoke the process with the values of all non-empty channels
if isinstance(proc.channels, dict):
try:
val: dict = {
k: read_channel(
channels,
chan,
catch=chan not in proc.triggers,
)
if chan in channels
else managed[k](step)
for k, chan in proc.channels.items()
}
val: dict[str, Any] = {}
for k, chan in proc.channels.items():
if chan in proc.triggers:
val[k] = read_channel(channels, chan, catch=False)
elif chan in channels:
try:
val[k] = read_channel(channels, chan, catch=False)
except EmptyChannelError:
continue
else:
val[k] = managed[k](step)
except EmptyChannelError:
return
elif isinstance(proc.channels, list):
libs/langgraph/langgraph/pregel/config.py
+34
View File
@@ -0,0 +1,34 @@
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import CheckpointMetadata
from langgraph.constants import CONFIG_KEY_CHECKPOINT_MAP
def patch_configurable(
config: Optional[RunnableConfig], patch: dict[str, Any]
) -> RunnableConfig:
if config is None:
return {"configurable": patch}
else:
return {**config, "configurable": {**config["configurable"], **patch}}
def patch_checkpoint_map(
config: RunnableConfig, metadata: Optional[CheckpointMetadata]
) -> RunnableConfig:
if parents := (metadata.get("parents") if metadata else None):
return patch_configurable(
config,
{
CONFIG_KEY_CHECKPOINT_MAP: {
**parents,
config["configurable"]["checkpoint_ns"]: config["configurable"][
"checkpoint_id"
],
},
},
)
else:
return config
libs/langgraph/langgraph/pregel/debug.py
+16 -12
View File
@@ -78,11 +78,11 @@ def map_debug_tasks(
step: int, tasks: list[PregelExecutableTask]
) -> Iterator[DebugOutputTask]:
ts = datetime.now(timezone.utc).isoformat()
for name, input, _, _, config, triggers, _, _ in tasks:
if config is not None and TAG_HIDDEN in config.get("tags", []):
for task in tasks:
if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
continue
metadata = config["metadata"].copy()
metadata = task.config["metadata"].copy()
metadata.pop("checkpoint_id", None)
yield {
@@ -90,10 +90,12 @@ def map_debug_tasks(
"timestamp": ts,
"step": step,
"payload": {
"id": str(uuid5(TASK_NAMESPACE, json.dumps((name, step, metadata)))),
"name": name,
"input": input,
"triggers": triggers,
"id": str(
uuid5(TASK_NAMESPACE, json.dumps((task.name, step, metadata)))
),
"name": task.name,
"input": task.input,
"triggers": task.triggers,
},
}
@@ -107,11 +109,11 @@ def map_debug_task_results(
[stream_keys] if isinstance(stream_keys, str) else stream_keys
)
ts = datetime.now(timezone.utc).isoformat()
for (name, _, _, _, config, _, _, _), writes in tasks:
if config is not None and TAG_HIDDEN in config.get("tags", []):
for task, writes in tasks:
if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
continue
metadata = config["metadata"].copy()
metadata = task.config["metadata"].copy()
metadata.pop("checkpoint_id", None)
# TODO: make task IDs deterministic in tests and reuse task IDs for payload ID
@@ -120,8 +122,10 @@ def map_debug_task_results(
"timestamp": ts,
"step": step,
"payload": {
"id": str(uuid5(TASK_NAMESPACE, json.dumps((name, step, metadata)))),
"name": name,
"id": str(
uuid5(TASK_NAMESPACE, json.dumps((task.name, step, metadata)))
),
"name": task.name,
"error": next((w[1] for w in writes if w[0] == ERROR), None),
"result": [w for w in writes if w[0] in stream_channels_list],
"interrupts": [asdict(w[1]) for w in writes if w[0] == INTERRUPT],
libs/langgraph/langgraph/pregel/io.py
+8 -4
View File
@@ -1,4 +1,4 @@
from typing import Any, Iterator, Mapping, Optional, Sequence, TypeVar, Union
from typing import Any, Iterator, Literal, Mapping, Optional, Sequence, TypeVar, Union
from langchain_core.runnables.utils import AddableDict
@@ -73,15 +73,19 @@ class AddableValuesDict(AddableDict):
def map_output_values(
output_channels: Union[str, Sequence[str]],
pending_writes: Sequence[tuple[str, Any]],
pending_writes: Union[Literal[True], Sequence[tuple[str, Any]]],
channels: Mapping[str, BaseChannel],
) -> Iterator[Union[dict[str, Any], Any]]:
"""Map pending writes (a sequence of tuples (channel, value)) to output chunk."""
if isinstance(output_channels, str):
if any(chan == output_channels for chan, _ in pending_writes):
if pending_writes is True or any(
chan == output_channels for chan, _ in pending_writes
):
yield read_channel(channels, output_channels)
else:
if {c for c, _ in pending_writes if c in output_channels}:
if pending_writes is True or {
c for c, _ in pending_writes if c in output_channels
}:
yield AddableValuesDict(read_channels(channels, output_channels))
libs/langgraph/langgraph/pregel/loop.py
+35 -6
View File
@@ -40,9 +40,9 @@ from langgraph.checkpoint.base import (
)
from langgraph.constants import (
CONFIG_KEY_CHECKPOINT_MAP,
CONFIG_KEY_READ,
CONFIG_KEY_RESUMING,
CONFIG_KEY_STREAM,
CONFIG_KEY_TASK_ID,
ERROR,
INPUT,
INTERRUPT,
@@ -60,6 +60,7 @@ from langgraph.pregel.algo import (
prepare_next_tasks,
should_interrupt,
)
from langgraph.pregel.config import patch_configurable
from langgraph.pregel.debug import (
map_debug_checkpoint,
map_debug_task_results,
@@ -121,6 +122,7 @@ class PregelLoop:
output_keys: Union[str, Sequence[str]]
stream_keys: Union[str, Sequence[str]]
is_nested: bool
skip_done_tasks: bool
checkpointer_get_next_version: Callable[[Optional[V]], V]
checkpointer_put_writes: Optional[
@@ -178,11 +180,31 @@ class PregelLoop:
self.specs = specs
self.output_keys = output_keys
self.stream_keys = stream_keys
self.is_nested = CONFIG_KEY_READ in self.config.get("configurable", {})
self.is_nested = CONFIG_KEY_TASK_ID in self.config.get("configurable", {})
self.skip_done_tasks = "checkpoint_id" not in config["configurable"]
if CONFIG_KEY_STREAM in config["configurable"]:
self.stream = DuplexStream(
self.stream, config["configurable"][CONFIG_KEY_STREAM]
)
if not self.is_nested and config["configurable"].get("checkpoint_ns"):
self.config = patch_configurable(
config, {"checkpoint_ns": "", "checkpoint_id": None}
)
if (
CONFIG_KEY_CHECKPOINT_MAP in self.config["configurable"]
and self.config["configurable"].get("checkpoint_ns")
in self.config["configurable"][CONFIG_KEY_CHECKPOINT_MAP]
):
self.checkpoint_config = patch_configurable(
self.config,
{
"checkpoint_id": config["configurable"][CONFIG_KEY_CHECKPOINT_MAP][
self.config["configurable"]["checkpoint_ns"]
]
},
)
else:
self.checkpoint_config = config
def put_writes(self, task_id: str, writes: Sequence[tuple[str, Any]]) -> None:
"""Put writes for a task, to be read by the next tick."""
@@ -320,7 +342,7 @@ class PregelLoop:
return False
# if there are pending writes from a previous loop, apply them
if self.checkpoint_pending_writes:
if self.skip_done_tasks and self.checkpoint_pending_writes:
for tid, k, v in self.checkpoint_pending_writes:
if k in (ERROR, INTERRUPT):
continue
@@ -374,6 +396,11 @@ class PregelLoop:
if k in self.checkpoint["channel_versions"]:
version = self.checkpoint["channel_versions"][k]
self.checkpoint["versions_seen"][INTERRUPT][k] = version
# produce values output
self.stream.extend(
(self.config["configurable"].get("checkpoint_ns", ""), "values", v)
for v in map_output_values(self.output_keys, True, self.channels)
)
# map inputs to channel updates
elif input_writes := deque(map_input(input_keys, self.input)):
# discard any unfinished tasks from previous checkpoint
@@ -395,7 +422,7 @@ class PregelLoop:
self.checkpointer_get_next_version,
), "Can't write to SharedValues in graph input"
# save input checkpoint
self._put_checkpoint({"source": "input", "writes": self.input})
self._put_checkpoint({"source": "input", "writes": dict(input_writes)})
else:
raise EmptyInputError(f"Received no input for {input_keys}")
# done with input
@@ -524,7 +551,9 @@ class SyncPregelLoop(PregelLoop, ContextManager):
def __enter__(self) -> Self:
saved = (
self.checkpointer.get_tuple(self.config) if self.checkpointer else None
self.checkpointer.get_tuple(self.checkpoint_config)
if self.checkpointer
else None
) or CheckpointTuple(self.config, empty_checkpoint(), {"step": -2}, None, [])
self.checkpoint_config = {
**self.config,
@@ -616,7 +645,7 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
async def __aenter__(self) -> Self:
saved = (
await self.checkpointer.aget_tuple(self.config)
await self.checkpointer.aget_tuple(self.checkpoint_config)
if self.checkpointer
else None
) or CheckpointTuple(self.config, empty_checkpoint(), {"step": -2}, None, [])
libs/langgraph/langgraph/pregel/types.py
+7
View File
@@ -57,6 +57,12 @@ class RetryPolicy(NamedTuple):
"""List of exception classes that should trigger a retry, or a callable that returns True for exceptions that should trigger a retry."""
class CachePolicy(NamedTuple):
"""Configuration for caching nodes."""
pass
class PregelTask(NamedTuple):
id: str
name: str
@@ -73,6 +79,7 @@ class PregelExecutableTask(NamedTuple):
config: RunnableConfig
triggers: list[str]
retry_policy: Optional[RetryPolicy]
cache_policy: Optional[CachePolicy]
id: str
libs/langgraph/langgraph/utils.py
+2
View File
@@ -201,6 +201,8 @@ def _is_optional_type(type_: Any) -> bool:
return any(
arg is type(None) or _is_optional_type(arg) for arg in type_.__args__
)
if origin is Annotated:
return _is_optional_type(type_.__args__[0])
return origin is None
if hasattr(type_, "__bound__") and type_.__bound__ is not None:
return _is_optional_type(type_.__bound__)
libs/langgraph/poetry.lock
Generated
+8 -8
View File
@@ -1,4 +1,4 @@
# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
[[package]]
name = "aiohappyeyeballs"
@@ -1634,13 +1634,13 @@ test = ["jupyter-server (>=2.0.0)", "pytest (>=7.0)", "pytest-jupyter[server] (>
[[package]]
name = "jupyterlab"
version = "4.2.2"
version = "4.2.5"
description = "JupyterLab computational environment"
optional = false
python-versions = ">=3.8"
files = [
{file = "jupyterlab-4.2.2-py3-none-any.whl", hash = "sha256:59ee9b839f43308c3dfd55d72d1f1a299ed42a7f91f2d1afe9c12a783f9e525f"},
{file = "jupyterlab-4.2.2.tar.gz", hash = "sha256:a534b6a25719a92a40d514fb133a9fe8f0d9981b0bbce5d8a5fcaa33344a3038"},
{file = "jupyterlab-4.2.5-py3-none-any.whl", hash = "sha256:73b6e0775d41a9fee7ee756c80f58a6bed4040869ccc21411dc559818874d321"},
{file = "jupyterlab-4.2.5.tar.gz", hash = "sha256:ae7f3a1b8cb88b4f55009ce79fa7c06f99d70cd63601ee4aa91815d054f46f75"},
]
[package.dependencies]
@@ -1665,7 +1665,7 @@ dev = ["build", "bump2version", "coverage", "hatch", "pre-commit", "pytest-cov",
docs = ["jsx-lexer", "myst-parser", "pydata-sphinx-theme (>=0.13.0)", "pytest", "pytest-check-links", "pytest-jupyter", "sphinx (>=1.8,<7.3.0)", "sphinx-copybutton"]
docs-screenshots = ["altair (==5.3.0)", "ipython (==8.16.1)", "ipywidgets (==8.1.2)", "jupyterlab-geojson (==3.4.0)", "jupyterlab-language-pack-zh-cn (==4.1.post2)", "matplotlib (==3.8.3)", "nbconvert (>=7.0.0)", "pandas (==2.2.1)", "scipy (==1.12.0)", "vega-datasets (==0.9.0)"]
test = ["coverage", "pytest (>=7.0)", "pytest-check-links (>=0.7)", "pytest-console-scripts", "pytest-cov", "pytest-jupyter (>=0.5.3)", "pytest-timeout", "pytest-tornasync", "requests", "requests-cache", "virtualenv"]
upgrade-extension = ["copier (>=8,<10)", "jinja2-time (<0.3)", "pydantic (<2.0)", "pyyaml-include (<2.0)", "tomli-w (<2.0)"]
upgrade-extension = ["copier (>=9,<10)", "jinja2-time (<0.3)", "pydantic (<3.0)", "pyyaml-include (<3.0)", "tomli-w (<2.0)"]
[[package]]
name = "jupyterlab-pygments"
@@ -2243,13 +2243,13 @@ files = [
[[package]]
name = "notebook"
version = "7.2.1"
version = "7.2.2"
description = "Jupyter Notebook - A web-based notebook environment for interactive computing"
optional = false
python-versions = ">=3.8"
files = [
{file = "notebook-7.2.1-py3-none-any.whl", hash = "sha256:f45489a3995746f2195a137e0773e2130960b51c9ac3ce257dbc2705aab3a6ca"},
{file = "notebook-7.2.1.tar.gz", hash = "sha256:4287b6da59740b32173d01d641f763d292f49c30e7a51b89c46ba8473126341e"},
{file = "notebook-7.2.2-py3-none-any.whl", hash = "sha256:c89264081f671bc02eec0ed470a627ed791b9156cad9285226b31611d3e9fe1c"},
{file = "notebook-7.2.2.tar.gz", hash = "sha256:2ef07d4220421623ad3fe88118d687bc0450055570cdd160814a59cf3a1c516e"},
]
[package.dependencies]
libs/langgraph/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph"
version = "0.2.14"
version = "0.2.15"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
license = "MIT"
libs/langgraph/tests/__snapshots__/test_pregel.ambr
+125
View File
@@ -5099,6 +5099,131 @@
# name: test_state_graph_w_config_inherited_state_keys.2
'{"title": "LangGraphOutput", "type": "object", "properties": {"input": {"title": "Input", "type": "string"}, "agent_outcome": {"title": "Agent Outcome", "anyOf": [{"$ref": "#/definitions/AgentAction"}, {"$ref": "#/definitions/AgentFinish"}]}, "intermediate_steps": {"title": "Intermediate Steps", "type": "array", "items": {"type": "array", "minItems": 2, "maxItems": 2, "items": [{"$ref": "#/definitions/AgentAction"}, {"type": "string"}]}}}, "definitions": {"AgentAction": {"title": "AgentAction", "description": "Represents a request to execute an action by an agent.\\n\\nThe action consists of the name of the tool to execute and the input to pass\\nto the tool. The log is used to pass along extra information about the action.", "type": "object", "properties": {"tool": {"title": "Tool", "type": "string"}, "tool_input": {"title": "Tool Input", "anyOf": [{"type": "string"}, {"type": "object"}]}, "log": {"title": "Log", "type": "string"}, "type": {"title": "Type", "default": "AgentAction", "enum": ["AgentAction"], "type": "string"}}, "required": ["tool", "tool_input", "log"]}, "AgentFinish": {"title": "AgentFinish", "description": "Final return value of an ActionAgent.\\n\\nAgents return an AgentFinish when they have reached a stopping condition.", "type": "object", "properties": {"return_values": {"title": "Return Values", "type": "object"}, "log": {"title": "Log", "type": "string"}, "type": {"title": "Type", "default": "AgentFinish", "enum": ["AgentFinish"], "type": "string"}}, "required": ["return_values", "log"]}}}'
# ---
# name: test_weather_subgraph[memory]
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
router_node(router_node)
normal_llm_node(normal_llm_node)
weather_graph_model_node(model_node)
weather_graph_weather_node(weather_node<hr/><small><em>__interrupt = before</em></small>)
__end__([__end__]):::last
subgraph weather_graph
weather_graph_model_node --> weather_graph_weather_node;
end
__start__ --> router_node;
normal_llm_node --> __end__;
weather_graph_weather_node --> __end__;
router_node -.-> normal_llm_node;
router_node -.-> weather_graph_model_node;
router_node -.-> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_weather_subgraph[postgres]
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
router_node(router_node)
normal_llm_node(normal_llm_node)
weather_graph_model_node(model_node)
weather_graph_weather_node(weather_node<hr/><small><em>__interrupt = before</em></small>)
__end__([__end__]):::last
subgraph weather_graph
weather_graph_model_node --> weather_graph_weather_node;
end
__start__ --> router_node;
normal_llm_node --> __end__;
weather_graph_weather_node --> __end__;
router_node -.-> normal_llm_node;
router_node -.-> weather_graph_model_node;
router_node -.-> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_weather_subgraph[postgres_pipe]
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
router_node(router_node)
normal_llm_node(normal_llm_node)
weather_graph_model_node(model_node)
weather_graph_weather_node(weather_node<hr/><small><em>__interrupt = before</em></small>)
__end__([__end__]):::last
subgraph weather_graph
weather_graph_model_node --> weather_graph_weather_node;
end
__start__ --> router_node;
normal_llm_node --> __end__;
weather_graph_weather_node --> __end__;
router_node -.-> normal_llm_node;
router_node -.-> weather_graph_model_node;
router_node -.-> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_weather_subgraph[postgres_pool]
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
router_node(router_node)
normal_llm_node(normal_llm_node)
weather_graph_model_node(model_node)
weather_graph_weather_node(weather_node<hr/><small><em>__interrupt = before</em></small>)
__end__([__end__]):::last
subgraph weather_graph
weather_graph_model_node --> weather_graph_weather_node;
end
__start__ --> router_node;
normal_llm_node --> __end__;
weather_graph_weather_node --> __end__;
router_node -.-> normal_llm_node;
router_node -.-> weather_graph_model_node;
router_node -.-> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_weather_subgraph[sqlite]
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
router_node(router_node)
normal_llm_node(normal_llm_node)
weather_graph_model_node(model_node)
weather_graph_weather_node(weather_node<hr/><small><em>__interrupt = before</em></small>)
__end__([__end__]):::last
subgraph weather_graph
weather_graph_model_node --> weather_graph_weather_node;
end
__start__ --> router_node;
normal_llm_node --> __end__;
weather_graph_weather_node --> __end__;
router_node -.-> normal_llm_node;
router_node -.-> weather_graph_model_node;
router_node -.-> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_xray_issue
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
libs/langgraph/tests/any_str.py
+9 -4
View File
@@ -1,16 +1,21 @@
from typing import Any, Sequence
import re
from typing import Any, Sequence, Union
class AnyStr(str):
def __init__(self, prefix: str = "") -> None:
def __init__(self, prefix: Union[str, re.Pattern] = "") -> None:
super().__init__()
self.prefix = prefix
def __eq__(self, other: object) -> bool:
return isinstance(other, str) and other.startswith(self.prefix)
return isinstance(other, str) and (
other.startswith(self.prefix)
if isinstance(self.prefix, str)
else self.prefix.match(other)
)
def __hash__(self) -> int:
return hash(str(self))
return hash((str(self), self.prefix))
class AnyDict(dict):
libs/langgraph/tests/test_pregel.py
+703 -1942
View File
File diff suppressed because it is too large Load Diff
libs/langgraph/tests/test_pregel_async.py
+403 -2019
View File
File diff suppressed because it is too large Load Diff
libs/langgraph/tests/test_state.py
+2 -2
View File
@@ -1,10 +1,10 @@
from typing import Annotated as Annotated2
from typing import Any, NotRequired, Optional, Required
from typing import Any, Optional
import pytest
from langchain_core.runnables import RunnableConfig
from pydantic.v1 import BaseModel
from typing_extensions import Annotated, TypedDict
from typing_extensions import Annotated, NotRequired, Required, TypedDict
from langgraph.graph.state import StateGraph, _warn_invalid_state_schema
libs/sdk-py/langgraph_sdk/client.py
+12 -9
View File
@@ -27,7 +27,7 @@ from langgraph_sdk.schema import (
Cron,
DisconnectMode,
GraphSchema,
Metadata,
Json,
MultitaskStrategy,
OnCompletionBehavior,
OnConflictBehavior,
@@ -414,7 +414,7 @@ class AssistantsClient:
graph_id: Optional[str],
config: Optional[Config] = None,
*,
metadata: Metadata = None,
metadata: Json = None,
assistant_id: Optional[str] = None,
if_exists: Optional[OnConflictBehavior] = None,
) -> Assistant:
@@ -462,7 +462,7 @@ class AssistantsClient:
*,
graph_id: Optional[str] = None,
config: Optional[Config] = None,
metadata: Metadata = None,
metadata: Json = None,
) -> Assistant:
"""Update an assistant.
@@ -524,7 +524,7 @@ class AssistantsClient:
async def search(
self,
*,
metadata: Metadata = None,
metadata: Json = None,
graph_id: Optional[str] = None,
limit: int = 10,
offset: int = 0,
@@ -600,7 +600,7 @@ class ThreadsClient:
async def create(
self,
*,
metadata: Metadata = None,
metadata: Json = None,
thread_id: Optional[str] = None,
if_exists: Optional[OnConflictBehavior] = None,
) -> Thread:
@@ -675,7 +675,8 @@ class ThreadsClient:
async def search(
self,
*,
metadata: Metadata = None,
metadata: Json = None,
values: Json = None,
status: Optional[ThreadStatus] = None,
limit: int = 10,
offset: int = 0,
@@ -708,6 +709,8 @@ class ThreadsClient:
}
if metadata:
payload["metadata"] = metadata
if values:
payload["values"] = values
if status:
payload["status"] = status
return await self.http.post(
@@ -1447,8 +1450,8 @@ class RunsClient:
json=None,
)
async def join(self, thread_id: str, run_id: str) -> None:
"""Block until a run is done.
async def join(self, thread_id: str, run_id: str) -> dict:
"""Block until a run is done. Returns the final state of the thread.
Args:
thread_id: The thread ID to join.
@@ -1459,7 +1462,7 @@ class RunsClient:
Example Usage:
await client.runs.join(
result =await client.runs.join(
thread_id="thread_id_to_join",
run_id="run_id_to_join"
)
libs/sdk-py/langgraph_sdk/schema.py
+6 -6
View File
@@ -1,7 +1,7 @@
from datetime import datetime
from typing import Any, Literal, Optional, Sequence, TypedDict, Union
Metadata = Optional[dict[str, Any]]
Json = Optional[dict[str, Any]]
RunStatus = Literal["pending", "running", "error", "success", "timeout", "interrupted"]
@@ -70,7 +70,7 @@ class Assistant(TypedDict):
"""The time the assistant was created."""
updated_at: datetime
"""The last time the assistant was updated."""
metadata: Metadata
metadata: Json
"""The assistant metadata."""
@@ -81,11 +81,11 @@ class Thread(TypedDict):
"""The time the thread was created."""
updated_at: datetime
"""The last time the thread was updated."""
metadata: Metadata
metadata: Json
"""The thread metadata."""
status: ThreadStatus
"""The status of the thread, one of 'idle', 'busy', 'interrupted'."""
values: dict
values: Json
"""The current state of the thread."""
@@ -97,7 +97,7 @@ class ThreadState(TypedDict):
received."""
checkpoint_id: str
"""The ID of the checkpoint."""
metadata: Metadata
metadata: Json
"""Metadata for this state"""
created_at: Optional[str]
"""Timestamp of state creation"""
@@ -118,7 +118,7 @@ class Run(TypedDict):
"""The last time the run was updated."""
status: RunStatus
"""The status of the run. One of 'pending', 'running', "error", 'success', "timeout", "interrupted"."""
metadata: Metadata
metadata: Json
"""The run metadata."""
multitask_strategy: MultitaskStrategy
"""Strategy to handle concurrent runs on the same thread."""
libs/sdk-py/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-sdk"
version = "0.1.29"
version = "0.1.30"
description = "SDK for interacting with LangGraph API"
authors = []
license = "MIT"