Compare commits

..
Author SHA1 Message Date
Vadym BardaandGitHub 1907646bd4 sdk-py: release 0.1.28 (#1417) 2024-08-21 17:10:18 -04:00
Vadym BardaandGitHub 3a65f83ae1 sdk-py: allow passing custom headers (#1416) 2024-08-21 21:05:03 +00:00
Nuno CamposandGitHub 57811a6bfd Merge pull request #1414 from langchain-ai/nc/21aug/combine-managers
Combine channel and managed values manager
2024-08-21 13:49:15 -07:00
Nuno Campos e76f4cc434 Combine channel and managed values manager 2024-08-21 13:30:18 -07:00
Nuno Campos 14ec51601c lib0.2.8 2024-08-21 12:09:11 -07:00
Nuno CamposandGitHub 14976d4c56 Interrupts shouldn't be retried (#1413)
* Interrupts shouldn't be retried

* Add async test

* Lint
2024-08-21 12:08:58 -07:00
Nuno Campos 2f41b2891b lib0.2.7 2024-08-21 11:33:54 -07:00
Nuno CamposandGitHub c857a77dd1 Merge pull request #1352 from langchain-ai/nc/14aug/scoped-value
Add SharedValue
2024-08-21 11:33:18 -07:00
Nuno Campos 5fb2c2c6f8 Lint 2024-08-21 11:27:04 -07:00
Nuno Campos 4250ff92b8 Fix 2024-08-21 11:25:37 -07:00
Nuno Campos 7fd4a9ed30 Lint 2024-08-21 10:48:52 -07:00
Nuno Campos 2106b5e4a6 Lint 2024-08-21 10:48:04 -07:00
Nuno Campos 1af0367b34 Add error message 2024-08-21 09:41:53 -07:00
Nuno Campos f037a2e9cb Update docstring 2024-08-21 09:39:58 -07:00
Nuno Campos aa1a6be160 Rename more 2024-08-21 09:36:21 -07:00
Nuno Campos c9e6ee6da7 Rename 2024-08-21 09:34:12 -07:00
Nuno Campos bd7b9cca21 WIP 2024-08-21 09:30:47 -07:00
Nuno Campos b228fc1a9b Add serde 2024-08-21 09:30:34 -07:00
Nuno Campos c3794f1fd3 Use batched async kv inside loop 2024-08-21 09:30:22 -07:00
Nuno Campos 4e1db854f6 Add async test 2024-08-21 09:30:22 -07:00
Nuno Campos 630d9c79ed Split out async batch to sep file 2024-08-21 09:30:22 -07:00
Nuno Campos 8f8f3849fc Lint 2024-08-21 09:30:22 -07:00
Nuno Campos 656f89e16a Lint 2024-08-21 09:30:22 -07:00
Nuno Campos 9b90a24d94 Lint 2024-08-21 09:30:22 -07:00
Nuno Campos 77d7deb033 Rename 2024-08-21 09:30:22 -07:00
Nuno Campos 7ca37afc74 Fix 2024-08-21 09:30:21 -07:00
Nuno Campos 2bc0e2df42 Fix 2024-08-21 09:30:21 -07:00
Nuno Campos 1dc09dc45f Remove warning on write to managed channel 2024-08-21 09:30:21 -07:00
Nuno Campos 1cc02825ea Add ScopedValue
- state shared between threads
2024-08-21 09:30:21 -07:00
Nuno Campos d1b7a787ee lib0.2.6 2024-08-21 09:22:27 -07:00
Nuno CamposandGitHub f8b053a286 Enable raising Interrupt from inside a node, add list of current interrupts to get_state (#1354)
* Enable raising Interrupt from inside a node, add list of current interrupts to get_state

* Lint

* Allow multiple interrupt values in exception

* Better typings

* Fix some tests

* Fix up

* Lint

* Add test

* Lint

* WIP stay=True

* Fix tests for get_state

* Remove ids

* Fix step count

* 999

* less fun

* Undo

* Update debug interface

* Remove ability to pass multiple values

* Undo
2024-08-21 16:19:15 +00:00
David DuongandGitHub ebf060675d feat(js-sdk): add config and parent_config to thread state schema (#1411)
* feat(js-sdk): add config and parent_config to thread state schema

* Bump to 0.0.5
2024-08-21 17:00:10 +02:00
Nuno Campos 3d33b575d6 0.2.5 2024-08-20 16:59:04 -07:00
Vadym BardaandGitHub 314488b5c5 ci: run full test suite on all changes (#1397) 2024-08-20 08:47:12 -04:00
Nuno CamposandGitHub 6e4dc26890 Merge pull request #1396 from langchain-ai/nc/19aug/fix-args
postgres: Fix args to put_writes
2024-08-19 15:57:44 -07:00
Nuno Campos 9cfb4b01c8 postgres: Fix args to put_writes 2024-08-19 15:54:49 -07:00
Nuno CamposandGitHub e7d2621a05 Merge pull request #1395 from langchain-ai/nc/19aug/put-writes-idempotent
checkpoint-*: In put_writes clear any previously saved writes for this task_id
2024-08-19 15:48:40 -07:00
Nuno Campos c0c534cbe3 Add pipeline arg 2024-08-19 15:44:22 -07:00
Nuno Campos 7aaeedd7ba checkpoint-*: In put_writes clear any previously saved writes for this task_id
- Previously implementations were clearing only tasks if the index matched a previously saved one, which isn't enough to guarantee idempotency
2024-08-19 15:40:04 -07:00
Nuno CamposandGitHub e15d56c2f2 Merge pull request #1390 from langchain-ai/nc/19aug/add-xray-test
Add xray test for issue #593
2024-08-19 12:09:51 -07:00
Nuno Campos 8a7c6b4fa7 Add xray test 2024-08-19 12:05:36 -07:00
Nuno CamposandGitHub 2e073473dc Merge pull request #1389 from langchain-ai/nc/19aug/fix-ready-signal
cli: Fix ready signal
2024-08-19 08:26:14 -07:00
Nuno Campos 31dcd15927 cli: Fix ready signal 2024-08-19 07:52:00 -07:00
Nuno CamposandGitHub fc95028738 Merge pull request #1370 from langchain-ai/nc/16aug/checkpoint-errors
Checkpoint errors in pending_writes, expose under StateSnapshot.tasks
2024-08-16 13:15:40 -07:00
Nuno Campos e916cab08f Checkpoint errors in pending_writes, expose under StateSnapshot.tasks
- Save errors produced by tasks, under pending_writes
- Re-work logic to cancel other tasks when one fails, ready to change for interrupt exception
- Update serializer to handle exceptions
- Update get_state/get_state_history with new return value property "tasks" which contains a richer description of the next tasks, currently with id, name and error (if already ran and errored)
2024-08-16 13:11:27 -07:00
Vadym BardaandGitHub e4f1f90127 update pyproject in monorepo to reflect consistent min python version (#1369) 2024-08-16 15:41:56 +00:00
Nuno Campos a16a33806c lib0.2.4 2024-08-15 15:37:51 -07:00
Nuno CamposandGitHub 007f419bdc Merge pull request #1363 from Glavin001/patch-1
Include exception traceback in run_with_retry logging
2024-08-15 15:37:01 -07:00
Isaac FranciscoandGitHub 5bee9e92b5 [docs]: add studio desktop callout (#1361) 2024-08-15 22:35:04 +00:00
Nuno Campos 6a68581a9f Lint 2024-08-15 15:32:48 -07:00
Nuno Campos 78781b0a90 Use exc_info 2024-08-15 15:23:54 -07:00
Glavin WiechertandGitHub d15c5a46de Include exception traceback in run_with_retry logging 2024-08-15 18:29:05 -03:00
Isaac FranciscoandGitHub ce37fd3a04 [docs]: studio faqs (#1207)
* draft

* spelling

* rephrases

* vadym comments
2024-08-15 11:56:00 -07:00
Isaac FranciscoandGitHub 64636ee978 [docs]: add auth and small changes (#1356)
* wip

* m

* nit
2024-08-15 18:25:53 +00:00
Isaac FranciscoandGitHub 388a9643a5 [docs]: copying threads (#1350)
* wip

* wip

* wip

* vadym comments
2024-08-15 11:19:18 -07:00
Nuno CamposandGitHub c9f4b5aede Merge pull request #1360 from langchain-ai/nc/14auug/debug-checkpoint-full
Add missing attributes to checkpoint event in debug output
2024-08-15 10:26:27 -07:00
Nuno Campos c701fef634 Lint 2024-08-15 09:40:53 -07:00
Nuno Campos 2c8fc3c9c6 Add missing attributes to checkpoint event in debug output 2024-08-15 09:30:29 -07:00
65 changed files with 2361 additions and 534 deletions
-9
View File
@@ -24,13 +24,7 @@ jobs:
name: "test #${{ matrix.python-version }}"
steps:
- uses: actions/checkout@v4
- name: Get changed files
id: changed-files
uses: Ana06/get-changed-files@v2.2.0
with:
filter: "${{ inputs.working-directory }}/**"
- name: Set up Python ${{ matrix.python-version }} + Poetry ${{ env.POETRY_VERSION }}
if: steps.changed-files.outputs.all
uses: "./.github/actions/poetry_setup"
with:
python-version: ${{ matrix.python-version }}
@@ -39,20 +33,17 @@ jobs:
cache-key: core
- name: Install dependencies
if: steps.changed-files.outputs.all
shell: bash
working-directory: ${{ inputs.working-directory }}
run: poetry install --with dev
- name: Run core tests
if: steps.changed-files.outputs.all
shell: bash
working-directory: ${{ inputs.working-directory }}
run: |
make test
- name: Ensure the tests did not create any additional files
if: steps.changed-files.outputs.all
shell: bash
working-directory: ${{ inputs.working-directory }}
run: |
@@ -38,6 +38,46 @@ Ready!
We can now interact with the API server using the LangGraph SDK. First, we need to start our client, select our assistant (in this case a graph we called "agent", make sure to select the proper assistant you wish to test).
You can either initialize by passing authentication or by setting an environment variable.
#### Initialize with authentication
=== "Python"
```python
from langgraph_sdk import get_client
# only pass the url argument to get_client() if you changed the default port when calling langgraph up
client = get_client(url=<DEPLOYMENT_URL>,api_key=<LANGCHAIN_API_KEY>)
assistant_id = "agent"
thread = await client.threads.create()
```
=== "Javascript"
```js
import { Client } from "@langchain/langgraph-sdk";
// only set the apiUrl if you changed the default port when calling langgraph up
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <LANGCHAIN_API_KEY> });
const assistantId = "agent"
const thread = await client.threads.create();
```
=== "CURL"
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json'
--header 'x-api-key: <LANGCHAIN_API_KEY>'
```
#### Initialize with environment variables
If you have a `LANGCHAIN_API_KEY` set in your environment, you do not need to explicitly pass authentication to the client
=== "Python"
```python
@@ -60,6 +100,14 @@ We can now interact with the API server using the LangGraph SDK. First, we need
const thread = await client.threads.create();
```
=== "CURL"
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json'
```
Now we can invoke our graph to ensure it is working. Make sure to change the input to match the proper schema for your graph.
=== "Python"
@@ -96,4 +144,39 @@ Now we can invoke our graph to ensure it is working. Make sure to change the inp
}
```
=== "CURL"
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
\"assistant_id\": \"agent\",
\"input\": {\"messages\": [{\"role\": \"human\", \"content\": \"what's the weather in sf\"}]},
\"stream_mode\": [
\"events\"
]
}" | \
sed 's/\r$//' | \
awk '
/^event:/ {
if (data_content != "") {
print data_content "\n"
}
sub(/^event: /, "Receiving event of type: ", $0)
printf "%s...\n", $0
data_content = ""
}
/^data:/ {
sub(/^data: /, "", $0)
data_content = $0
}
END {
if (data_content != "") {
print data_content "\n"
}
}
'
```
If your graph works correctly, you should see your graph output displayed in the console. Of course, there are many more ways you might need to test your graph, for a full list of commands you can send with the SDK, see the [Python](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/python_sdk_ref/) and [JS/TS](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/js_ts_sdk_ref/) references.
+65
View File
@@ -0,0 +1,65 @@
# Studio FAQs
## Why is my project failing to start?
There are a few reasons that your project might fail to start, here are some of the most common ones.
### Docker issues
LangGraph Studio requires Docker Desktop version 4.24 or higher. Please make sure you have a version of Docker installed that satisfies that requirement and also make sure you have the Docker Desktop app up and running before trying to use LangGraph Studio. In addition, make sure you have docker-compose updated to version 2.22.0 or higher.
### Configuration or environment issues
Another reason your project might fail to start is because your configuration file is defined incorrectly, or you are missing required environment variables.
## How does interrupt work?
When you select the `Interrupts` dropdown and select a node to interrupt the graph will pause execution before and after (unless the node goes straight to `END`) that node has run. This means that you will be able to both edit the state before the node is ran and the state after the node has ran. This is intended to allow developers more fine-grained control over the behavior of a node and make it easier to observe how the node is behaving. You will not be able to edit the state after the node has ran if the node is the final node in the graph.
## How do I reload the app?
If you would like to reload the app, don't use Command+R as you might normally do. Instead, close and reopen the app for a full refresh.
## How does automatic rebuilding work?
One of the key features of LangGraph Studio is that it automatically rebuilds your image when you change the source code. This allows for a super fast development and testing cycle which makes it easy to iterate on your graph. There are two different ways that LangGraph rebuilds your image: either by editing the image or completely rebuilding it.
### Rebuilds from source code changes
If you modified the source code only (no configuration or dependency changes!) then the image does not require a full rebuild, and LangGraph Studio will only update the relevant parts. The UI status in the bottom left will switch from `Online` to `Stopping` temporarily while the image gets edited. The logs will be shown as this process is happening, and after the image has been edited the status will change back to `Online` and you will be able to run your graph with the modified code!
### Rebuilds from configuration or dependency changes
If you edit your graph configuration file (`langgraph.json`) or the dependencies (either `pyproject.toml` or `requirements.txt`) then the entire image will be rebuilt. This will cause the UI to switch away from the graph view and start showing the logs of the new image building process. This can take a minute or two, and once it is done your updated image will be ready to use!
## Why is my graph taking so long to startup?
The LangGraph Studio interacts with a local LangGraph API server. To stay aligned with ongoing updates, the LangGraph API requires regular rebuilding. As a result, you may occasionally experience slight delays when starting up your project.
## Why are extra edges showing up in my graph?
If you don't define your conditional edges carefully, you might notice extra edges appearing in your graph. This is because without proper definition, LangGraph Studio assumes the conditional edge could access all other nodes. In order for this to not be the case, you need to be explicit about how you define the nodes the conditional edge routes to. There are two ways you can do this:
### Solution 1: Include a path map
The first way to solve this is to add path maps to your conditional edges. A path map is just a dictionary that maps the possible outputs of your router function with the names of the nodes that each output corresponds to. The path map is passed as the third argument to the `add_conditional_edges` function like so:
```python
graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})
```
In this case, the routing function returns either True or False, which map to `node_b` and `node_c` respectively.
### Solution 2: Update the typing of the router
Instead of passing a path map, you can also be explicit about the typing of your routing function by specifying the nodes it can map to using the `Literal` python definition. Here is an example of how to define a routing function in that way:
```python
def routing_function(state: GraphState) -> Literal["node_b","node_c"]:
if state['some_condition'] == True:
return "node_a"
else:
return "node_b"
```
@@ -12,7 +12,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -22,7 +22,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = agent;
const thread = await client.threads.create();
```
@@ -32,10 +32,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
## Find idle threads
+132
View File
@@ -0,0 +1,132 @@
# Copying Threads
You may wish to copy (i.e. "fork") an existing thread in order to keep the existing thread's history and create independent runs that do not affect the original thread. This guide shows how you can do that.
## Setup
This code assumes you already have a thread to copy. You can read about what a thread is [here](https://langchain-ai.github.io/langgraph/cloud/concepts/api/#threads) and learn how to stream a run on a thread in [these how-to guides](https://langchain-ai.github.io/langgraph/cloud/how-tos/#streaming).
### SDK initialization
First, we need to setup our client so that we can communicate with our hosted graph:
=== "Python"
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
assistant_id = "agent"
thread = await client.threads.create()
```
=== "Javascript"
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const assistantId = agent;
const thread = await client.threads.create();
```
=== "CURL"
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
```
## Copying a thread
The code below assumes that a thread you'd like to copy already exists.
Copying a thread will create a new thread with the same history as the existing thread, and then allow you to continue executing runs.
### Create copy
=== "Python"
```python
copied_thread = await client.threads.copy(<THREAD_ID>)
```
=== "Javascript"
```js
let copiedThread = await client.threads.copy(<THREAD_ID>);
```
=== "CURL"
```bash
curl --request POST --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/copy \
--header 'Content-Type: application/json'
```
### Verify copy
We can verify that the history from the prior thread did indeed copy over correctly:
=== "Python"
```python
def remove_thread_id(d):
if 'metadata' in d and 'thread_id' in d['metadata']:
del d['metadata']['thread_id']
return d
original_thread_history = list(map(remove_thread_id,await client.threads.get_history(<THREAD_ID>)))
copied_thread_history = list(map(remove_thread_id,await client.threads.get_history(copied_thread['thread_id'])))
# Compare the two histories
assert original_thread_history == copied_thread_history
# if we made it here the assertion passed!
print("The histories are the same.")
```
=== "Javascript"
```js
function removeThreadId(d) {
if (d.metadata && d.metadata.thread_id) {
delete d.metadata.thread_id;
}
return d;
}
// Assuming `client.threads.getHistory(threadId)` is an async function that returns a list of dicts
async function compareThreadHistories(threadId, copiedThreadId) {
const originalThreadHistory = (await client.threads.getHistory(threadId)).map(removeThreadId);
const copiedThreadHistory = (await client.threads.getHistory(copiedThreadId)).map(removeThreadId);
// Compare the two histories
console.assert(JSON.stringify(originalThreadHistory) === JSON.stringify(copiedThreadHistory))
// if we made it here the assertion passed!
console.log("The histories are the same.");
}
// Example usage
compareThreadHistories(<THREAD_ID>, copiedThread.thread_id);
```
=== "CURL"
```bash
if diff <(
curl --request GET --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/history | jq -S 'map(del(.metadata.thread_id))'
) <(
curl --request GET --url <DEPLOYMENT_URL>/threads/<COPIED_THREAD_ID>/history | jq -S 'map(del(.metadata.thread_id))'
) >/dev/null; then
echo "The histories are the same."
else
echo "The histories are different."
fi
```
Output:
The histories are the same.
@@ -31,7 +31,7 @@ Then, let's import our required packages and instantiate our client, assistant,
from langchain_core.messages import convert_to_messages
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -42,7 +42,7 @@ Then, let's import our required packages and instantiate our client, assistant,
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
@@ -21,7 +21,7 @@ In this how-to we use a simple ReAct style hosted graph (you can see the full co
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -31,7 +31,7 @@ In this how-to we use a simple ReAct style hosted graph (you can see the full co
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent"
const thread = await client.threads.create();
```
@@ -41,10 +41,7 @@ In this how-to we use a simple ReAct style hosted graph (you can see the full co
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
## Adding a breakpoint
@@ -17,7 +17,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -27,7 +27,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
@@ -37,10 +37,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
## Editing state
@@ -28,7 +28,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -38,7 +38,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
@@ -14,7 +14,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -24,7 +24,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = agent;
const thread = await client.threads.create();
```
@@ -34,10 +34,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
## Replay a state
@@ -24,7 +24,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -34,7 +34,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
@@ -44,10 +44,7 @@ First, we need to setup our client so that we can communicate with our hosted gr
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
## Waiting for user input
+2 -1
View File
@@ -73,4 +73,5 @@ Other guides that may prove helpful!
- [How to configure agents](cloud_examples/configuration_cloud.ipynb)
- [How to convert LangGraph calls to LangGraph cloud calls](cloud_examples/langgraph_to_langgraph_cloud.ipynb)
- [How to integrate webhooks](cloud_examples/webhooks.ipynb)
- [How to check status of your threads](./check_thread_status.md)
- [How to copy threads](./copy_threads.md)
- [How to check status of your threads](./check_thread_status.md)
@@ -29,7 +29,7 @@ Now, let's import our required packages and instantiate our client, assistant, a
from langchain_core.messages import convert_to_messages
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -39,7 +39,7 @@ Now, let's import our required packages and instantiate our client, assistant, a
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
+2 -2
View File
@@ -28,7 +28,7 @@ Now, let's import our required packages and instantiate our client, assistant, a
from langchain_core.messages import convert_to_messages
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -38,7 +38,7 @@ Now, let's import our required packages and instantiate our client, assistant, a
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
@@ -30,7 +30,7 @@ Now, let's import our required packages and instantiate our client, assistant, a
from langchain_core.messages import convert_to_messages
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
```
@@ -40,7 +40,7 @@ Now, let's import our required packages and instantiate our client, assistant, a
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantId = "agent";
const thread = await client.threads.create();
```
+2 -2
View File
@@ -9,7 +9,7 @@ First let's set up our client and thread:
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
# create thread
thread = await client.threads.create()
print(thread)
@@ -20,7 +20,7 @@ First let's set up our client and thread:
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// create thread
const thread = await client.threads.create();
console.log(thread)
+3 -6
View File
@@ -6,7 +6,7 @@ This guide covers how to stream events from your graph (`stream_mode="events"`).
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
# create thread
thread = await client.threads.create()
print(thread)
@@ -17,7 +17,7 @@ This guide covers how to stream events from your graph (`stream_mode="events"`).
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// create thread
const thread = await client.threads.create();
console.log(thread)
@@ -28,10 +28,7 @@ This guide covers how to stream events from your graph (`stream_mode="events"`).
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
Output:
+3 -6
View File
@@ -41,7 +41,7 @@ First let's set up our client and thread:
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
# create thread
thread = await client.threads.create()
print(thread)
@@ -52,7 +52,7 @@ First let's set up our client and thread:
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// create thread
const thread = await client.threads.create();
console.log(thread)
@@ -63,10 +63,7 @@ First let's set up our client and thread:
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
Output:
+3 -6
View File
@@ -9,7 +9,7 @@ First let's set up our client and thread:
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
# create thread
thread = await client.threads.create()
print(thread)
@@ -20,7 +20,7 @@ First let's set up our client and thread:
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// create thread
const thread = await client.threads.create();
console.log(thread)
@@ -31,10 +31,7 @@ First let's set up our client and thread:
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
Output:
+3 -6
View File
@@ -16,7 +16,7 @@ First let's set up our client and thread:
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
# create thread
thread = await client.threads.create()
print(thread)
@@ -27,7 +27,7 @@ First let's set up our client and thread:
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl:"<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// create thread
const thread = await client.threads.create();
console.log(thread)
@@ -38,10 +38,7 @@ First let's set up our client and thread:
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
Output:
+3 -6
View File
@@ -16,7 +16,7 @@ First let's set up our client and thread:
```python
from langgraph_sdk import get_client
client = get_client(url="<DEPLOYMENT_URL>")
client = get_client(url=<DEPLOYMENT_URL>)
# create thread
thread = await client.threads.create()
print(thread)
@@ -27,7 +27,7 @@ First let's set up our client and thread:
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: "<DEPLOYMENT_URL>" });
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
// create thread
const thread = await client.threads.create();
console.log(thread)
@@ -38,10 +38,7 @@ First let's set up our client and thread:
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{
"metadata": {}
}'
--header 'Content-Type: application/json'
```
Output:
Binary file not shown.

After

Width:  |  Height:  |  Size: 884 KiB

+2
View File
@@ -23,6 +23,8 @@ The LangGraph Cloud API exposes functionality of your LangGraph application thro
LangGraph Cloud is seamlessly integrated with [LangSmith](https://www.langchain.com/langsmith) and is accessible from within the LangSmith UI.
LangGraph Cloud applications can be tested and debugged using the [LangGraph Studio Desktop](https://github.com/langchain-ai/langgraph-studio).
## Key Features
The LangGraph Cloud API supports key LangGraph features in addition to new functionality for enabling complex, agentic workflows.
+10
View File
@@ -66,6 +66,16 @@ Now that we have set everything up on our local file system, we are ready to hos
## Test the graph build locally
### Using LangGraph Studio Desktop (recommended)
![LangGraph Studio Desktop](./img/graph_video_poster.png)
Testing your graph locally is easy with LangGraph Studio Desktop. LangGraph Studio offers a new way to develop LLM applications by providing a specialized agent IDE that enables visualization, interaction, and debugging of complex agentic applications
With visual graphs and the ability to edit state, you can better understand agent workflows and iterate faster. LangGraph Studio integrates with [LangSmith](https://smith.langchain.com) so you can collaborate with teammates to debug failure modes.
### Using the LangGraph CLI
Before deploying to the cloud, we probably want to test the building of our graph locally. This is useful to make sure we have configured our [CLI configuration file][langgraph.json] correctly and our graph runs.
In order to do this we can first install the LangGraph CLI
+3
View File
@@ -231,6 +231,7 @@ nav:
- Configure Agents: "cloud/how-tos/cloud_examples/configuration_cloud.ipynb"
- Convert LangGraph calls to LangGraph Cloud calls: "cloud/how-tos/cloud_examples/langgraph_to_langgraph_cloud.ipynb"
- Integrate Webhooks: 'cloud/how-tos/cloud_examples/webhooks.ipynb'
- Copy Threads: 'cloud/how-tos/copy_threads.md'
- Check Status of Threads: "cloud/how-tos/check_thread_status.md"
- Conceptual Guides:
- API Concepts: "cloud/concepts/api.md"
@@ -242,6 +243,8 @@ nav:
- JS/TS: "cloud/reference/sdk/js_ts_sdk_ref.md"
- CLI: "cloud/reference/cli.md"
- Environment Variables: "cloud/reference/env_var.md"
- FAQ:
- Studio: "cloud/faq/studio.md"
markdown_extensions:
- abbr
@@ -316,7 +316,17 @@ class PostgresSaver(BasePostgresSaver):
writes (List[Tuple[str, Any]]): List of writes to store.
task_id (str): Identifier for the task creating the writes.
"""
with self._cursor() as cur:
with self._cursor(pipeline=True) as cur:
cur.execute(
self.DELETE_WRITES_SQL,
(
config["configurable"]["thread_id"],
config["configurable"]["checkpoint_ns"],
config["configurable"]["checkpoint_id"],
task_id,
len(writes),
),
)
cur.executemany(
self.UPSERT_CHECKPOINT_WRITES_SQL,
self._dump_writes(
@@ -272,7 +272,17 @@ class AsyncPostgresSaver(BasePostgresSaver):
writes (Sequence[Tuple[str, Any]]): List of writes to store, each as (channel, value) pair.
task_id (str): Identifier for the task creating the writes.
"""
async with self._cursor() as cur:
async with self._cursor(pipeline=True) as cur:
await cur.execute(
self.DELETE_WRITES_SQL,
(
config["configurable"]["thread_id"],
config["configurable"]["checkpoint_ns"],
config["configurable"]["checkpoint_id"],
task_id,
len(writes),
),
)
await cur.executemany(
self.UPSERT_CHECKPOINT_WRITES_SQL,
await asyncio.to_thread(
@@ -105,6 +105,15 @@ UPSERT_CHECKPOINT_WRITES_SQL = """
ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
"""
DELETE_WRITES_SQL = """
DELETE FROM checkpoint_writes
WHERE thread_id = %s
AND checkpoint_ns = %s
AND checkpoint_id = %s
AND task_id = %s
AND idx >= %s
"""
class BasePostgresSaver(BaseCheckpointSaver):
SELECT_SQL = SELECT_SQL
@@ -112,6 +121,8 @@ class BasePostgresSaver(BaseCheckpointSaver):
UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
DELETE_WRITES_SQL = DELETE_WRITES_SQL
jsonplus_serde = JsonPlusSerializer()
def _load_checkpoint(self, checkpoint: dict[str, Any]) -> Checkpoint:
@@ -424,6 +424,16 @@ class SqliteSaver(BaseCheckpointSaver):
task_id (str): Identifier for the task creating the writes.
"""
with self.lock, self.cursor() as cur:
cur.execute(
"DELETE FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? AND task_id = ? AND idx >= ?",
(
str(config["configurable"]["thread_id"]),
str(config["configurable"]["checkpoint_ns"]),
str(config["configurable"]["checkpoint_id"]),
task_id,
len(writes),
),
)
cur.executemany(
"INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
[
@@ -432,19 +432,29 @@ class AsyncSqliteSaver(BaseCheckpointSaver):
task_id (str): Identifier for the task creating the writes.
"""
await self.setup()
async with self.conn.executemany(
"INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
[
async with self.conn.cursor() as cur:
await cur.execute(
"DELETE FROM writes WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? AND task_id = ? AND idx >= ?",
(
str(config["configurable"]["thread_id"]),
str(config["configurable"]["checkpoint_ns"]),
str(config["configurable"]["checkpoint_id"]),
task_id,
idx,
channel,
*self.serde.dumps_typed(value),
)
for idx, (channel, value) in enumerate(writes)
],
):
await self.conn.commit()
len(writes),
),
)
await cur.executemany(
"INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
[
(
str(config["configurable"]["thread_id"]),
str(config["configurable"]["checkpoint_ns"]),
str(config["configurable"]["checkpoint_id"]),
task_id,
idx,
channel,
*self.serde.dumps_typed(value),
)
for idx, (channel, value) in enumerate(writes)
],
)
@@ -288,6 +288,7 @@ class MemorySaver(
checkpoint_ns = config["configurable"]["checkpoint_ns"]
checkpoint_id = config["configurable"]["checkpoint_id"]
key = (thread_id, checkpoint_ns, checkpoint_id)
self.writes[key] = [w for w in self.writes[key] if w[0] != task_id]
self.writes[key].extend(
[(task_id, c, self.serde.dumps_typed(v)) for c, v in writes]
)
@@ -110,6 +110,8 @@ class JsonPlusSerializer(SerializerProtocol):
return self._encode_constructor_args(
obj.__class__, method="fromhex", args=[obj.hex()]
)
elif isinstance(obj, BaseException):
return self._encode_constructor_args(obj.__class__, args=obj.args)
else:
raise TypeError(
f"Object of type {obj.__class__.__name__} is not JSON serializable"
@@ -131,9 +133,16 @@ class JsonPlusSerializer(SerializerProtocol):
# Instantiate class
if value["method"] is not None:
method = getattr(cls, value["method"])
return method(*value["args"], **value["kwargs"])
else:
return cls(*value["args"], **value["kwargs"])
method = cls
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):
return None
+1 -1
View File
@@ -224,7 +224,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
def on_stdout(line: str):
if "unpacking to docker.io" in line:
set("Starting...")
elif "GET /ok" in line:
elif "Application startup complete" in line:
debugger_origin = (
f"http://localhost:{debugger_port}"
if debugger_port
@@ -1,39 +0,0 @@
from contextlib import AsyncExitStack, ExitStack, asynccontextmanager, contextmanager
from typing import AsyncGenerator, Generator, Mapping
from langchain_core.runnables import RunnableConfig
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import Checkpoint
@contextmanager
def ChannelsManager(
channels: Mapping[str, BaseChannel],
checkpoint: Checkpoint,
config: RunnableConfig,
) -> Generator[Mapping[str, BaseChannel], None, None]:
"""Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
with ExitStack() as stack:
yield {
k: stack.enter_context(
v.from_checkpoint(checkpoint["channel_values"].get(k), config)
)
for k, v in channels.items()
}
@asynccontextmanager
async def AsyncChannelsManager(
channels: Mapping[str, BaseChannel],
checkpoint: Checkpoint,
config: RunnableConfig,
) -> AsyncGenerator[Mapping[str, BaseChannel], None]:
"""Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
async with AsyncExitStack() as stack:
yield {
k: await stack.enter_async_context(
v.afrom_checkpoint(checkpoint["channel_values"].get(k), config)
)
for k, v in channels.items()
}
+14 -1
View File
@@ -1,19 +1,26 @@
from typing import Any
from dataclasses import dataclass
from typing import Any, Literal
INPUT = "__input__"
CONFIG_KEY_SEND = "__pregel_send"
CONFIG_KEY_READ = "__pregel_read"
CONFIG_KEY_CHECKPOINTER = "__pregel_checkpointer"
CONFIG_KEY_STORE = "__pregel_store"
CONFIG_KEY_RESUMING = "__pregel_resuming"
CONFIG_KEY_TASK_ID = "__pregel_task_id"
INTERRUPT = "__interrupt__"
ERROR = "__error__"
TASKS = "__pregel_tasks"
RESERVED = {
INTERRUPT,
ERROR,
TASKS,
CONFIG_KEY_SEND,
CONFIG_KEY_READ,
CONFIG_KEY_CHECKPOINTER,
CONFIG_KEY_STORE,
CONFIG_KEY_RESUMING,
CONFIG_KEY_TASK_ID,
INPUT,
}
TAG_HIDDEN = "langsmith:hidden"
@@ -91,3 +98,9 @@ class Send:
and self.node == value.node
and self.arg == value.arg
)
@dataclass
class Interrupt:
when: Literal["before", "during", "after"]
value: Any = None
+13 -1
View File
@@ -1,4 +1,7 @@
from typing import Any
from langgraph.checkpoint.base import EmptyChannelError
from langgraph.constants import Interrupt
class GraphRecursionError(RecursionError):
@@ -29,7 +32,15 @@ class InvalidUpdateError(Exception):
class GraphInterrupt(Exception):
"""Raised when a subgraph is interrupted."""
pass
def __init__(self, interrupts: list[Interrupt]) -> None:
super().__init__(interrupts)
class NodeInterrupt(GraphInterrupt):
"""Raised by a node to interrupt execution."""
def __init__(self, value: Any) -> None:
super().__init__([Interrupt("during", value)])
class EmptyInputError(Exception):
@@ -42,6 +53,7 @@ __all__ = [
"GraphRecursionError",
"InvalidUpdateError",
"GraphInterrupt",
"NodeInterrupt",
"EmptyInputError",
"EmptyChannelError",
]
+39 -17
View File
@@ -40,10 +40,18 @@ from langgraph.graph.graph import (
Graph,
Send,
)
from langgraph.managed.base import ManagedValue, is_managed_value
from langgraph.managed.base import (
ChannelKeyPlaceholder,
ChannelTypePlaceholder,
ConfiguredManagedValue,
ManagedValueSpec,
is_managed_value,
is_writable_managed_value,
)
from langgraph.pregel.read import ChannelRead, PregelNode
from langgraph.pregel.types import All, RetryPolicy
from langgraph.pregel.write import SKIP_WRITE, ChannelWrite, ChannelWriteEntry
from langgraph.store.base import BaseStore
from langgraph.utils import RunnableCallable, coerce_to_runnable
logger = logging.getLogger(__name__)
@@ -121,8 +129,8 @@ class StateGraph(Graph):
nodes: dict[str, StateNodeSpec]
channels: dict[str, BaseChannel]
managed: dict[str, Type[ManagedValue]]
schemas: dict[Type[Any], dict[str, Union[BaseChannel, Type[ManagedValue]]]]
managed: dict[str, ManagedValueSpec]
schemas: dict[Type[Any], dict[str, Union[BaseChannel, ManagedValueSpec]]]
def __init__(
self,
@@ -374,6 +382,8 @@ class StateGraph(Graph):
def compile(
self,
checkpointer: Optional[BaseCheckpointSaver] = None,
*,
store: Optional[BaseStore] = None,
interrupt_before: Optional[Union[All, Sequence[str]]] = None,
interrupt_after: Optional[Union[All, Sequence[str]]] = None,
debug: bool = False,
@@ -432,7 +442,11 @@ class StateGraph(Graph):
builder=self,
config_type=self.config_schema,
nodes={},
channels={**self.channels, START: EphemeralValue(self.input)},
channels={
**self.channels,
**self.managed,
START: EphemeralValue(self.input),
},
input_channels=START,
stream_mode="updates",
output_channels=output_channels,
@@ -442,6 +456,7 @@ class StateGraph(Graph):
interrupt_after_nodes=interrupt_after,
auto_validate=False,
debug=debug,
store=store,
)
compiled.attach_node(START, None)
@@ -486,7 +501,7 @@ class CompiledStateGraph(CompiledGraph):
**{
k: (self.channels[k].UpdateType, None)
for k in self.builder.schemas[self.builder.input]
if k in self.channels
if isinstance(self.channels[k], BaseChannel)
and not isinstance(self.channels[k], Context)
},
)
@@ -511,7 +526,11 @@ class CompiledStateGraph(CompiledGraph):
if not isinstance(v, Context) and not is_managed_value(v)
]
else:
output_keys = list(self.builder.channels)
output_keys = list(self.builder.channels) + [
k
for k, v in self.builder.managed.items()
if is_writable_managed_value(v)
]
def _get_state_key(
input: Union[None, dict, Any], config: RunnableConfig, *, key: str
@@ -557,10 +576,7 @@ class CompiledStateGraph(CompiledGraph):
)
else:
input_schema = node.input if node else self.builder.schema
input_values = {
k: v if is_managed_value(v) else k
for k, v in self.builder.schemas[input_schema].items()
}
input_values = {k: k for k in self.builder.schemas[input_schema]}
is_single_input = len(input_values) == 1 and "__root__" in input_values
self.channels[key] = EphemeralValue(Any, guard=False)
@@ -679,12 +695,12 @@ def _coerce_state(schema: Type[Any], input: dict[str, Any]) -> dict[str, Any]:
def _get_channels(
schema: Type[dict],
) -> tuple[dict[str, BaseChannel], dict[str, Type[ManagedValue]]]:
) -> tuple[dict[str, BaseChannel], dict[str, ManagedValueSpec]]:
if not hasattr(schema, "__annotations__"):
return {"__root__": _get_channel(schema, allow_managed=False)}, {}
return {"__root__": _get_channel("__root__", schema, allow_managed=False)}, {}
all_keys = {
name: _get_channel(typ)
name: _get_channel(name, typ)
for name, typ in get_type_hints(schema, include_extras=True).items()
if name != "__slots__"
}
@@ -695,9 +711,9 @@ def _get_channels(
def _get_channel(
annotation: Any, *, allow_managed: bool = True
) -> Union[BaseChannel, Type[ManagedValue]]:
if manager := _is_field_managed_value(annotation):
name: str, annotation: Any, *, allow_managed: bool = True
) -> Union[BaseChannel, ManagedValueSpec]:
if manager := _is_field_managed_value(name, annotation):
if allow_managed:
return manager
else:
@@ -736,12 +752,18 @@ def _is_field_binop(typ: Type[Any]) -> Optional[BinaryOperatorAggregate]:
return None
def _is_field_managed_value(typ: Type[Any]) -> Optional[Type[ManagedValue]]:
def _is_field_managed_value(name: str, typ: Type[Any]) -> Optional[ManagedValueSpec]:
if hasattr(typ, "__metadata__"):
meta = typ.__metadata__
if len(meta) >= 1:
decoration = get_origin(meta[-1]) or meta[-1]
if is_managed_value(decoration):
if isinstance(decoration, ConfiguredManagedValue):
for k, v in decoration.kwargs.items():
if v is ChannelKeyPlaceholder:
decoration.kwargs[k] = name
if v is ChannelTypePlaceholder:
decoration.kwargs[k] = typ.__origin__
return decoration
return None
+36 -56
View File
@@ -1,14 +1,13 @@
import asyncio
from abc import ABC, abstractmethod
from contextlib import AsyncExitStack, ExitStack, asynccontextmanager, contextmanager
from contextlib import asynccontextmanager, contextmanager
from inspect import isclass
from typing import (
TYPE_CHECKING,
Any,
AsyncGenerator,
Generator,
AsyncIterator,
Generic,
Iterator,
NamedTuple,
Sequence,
Type,
TypeVar,
Union,
@@ -17,10 +16,8 @@ from typing import (
from langchain_core.runnables import RunnableConfig
from typing_extensions import Self, TypeGuard
if TYPE_CHECKING:
from langgraph.pregel.types import PregelTaskDescription
V = TypeVar("V")
U = TypeVar("U")
class ManagedValue(ABC, Generic[V]):
@@ -29,9 +26,7 @@ class ManagedValue(ABC, Generic[V]):
@classmethod
@contextmanager
def enter(
cls, config: RunnableConfig, **kwargs: Any
) -> Generator[Self, None, None]:
def enter(cls, config: RunnableConfig, **kwargs: Any) -> Iterator[Self]:
try:
value = cls(config, **kwargs)
yield value
@@ -45,9 +40,7 @@ class ManagedValue(ABC, Generic[V]):
@classmethod
@asynccontextmanager
async def aenter(
cls, config: RunnableConfig, **kwargs: Any
) -> AsyncGenerator[Self, None]:
async def aenter(cls, config: RunnableConfig, **kwargs: Any) -> AsyncIterator[Self]:
try:
value = cls(config, **kwargs)
yield value
@@ -60,7 +53,17 @@ class ManagedValue(ABC, Generic[V]):
pass
@abstractmethod
def __call__(self, step: int, task: "PregelTaskDescription") -> V:
def __call__(self, step: int) -> V:
...
class WritableManagedValue(Generic[V, U], ManagedValue[V], ABC):
@abstractmethod
def update(self, writes: Sequence[U]) -> None:
...
@abstractmethod
async def aupdate(self, writes: Sequence[U]) -> None:
...
@@ -80,46 +83,23 @@ def is_managed_value(value: Any) -> TypeGuard[ManagedValueSpec]:
)
@contextmanager
def ManagedValuesManager(
values: dict[str, ManagedValueSpec],
config: RunnableConfig,
) -> Generator[ManagedValueMapping, None, None]:
if values:
with ExitStack() as stack:
yield {
key: stack.enter_context(
value.cls.enter(config, **value.kwargs)
if isinstance(value, ConfiguredManagedValue)
else value.enter(config)
)
for key, value in values.items()
}
else:
yield {}
def is_readonly_managed_value(value: Any) -> TypeGuard[Type[ManagedValue]]:
return (
isclass(value)
and issubclass(value, ManagedValue)
and not issubclass(value, WritableManagedValue)
) or (
isinstance(value, ConfiguredManagedValue)
and not issubclass(value.cls, WritableManagedValue)
)
@asynccontextmanager
async def AsyncManagedValuesManager(
values: dict[str, ManagedValueSpec],
config: RunnableConfig,
) -> AsyncGenerator[ManagedValueMapping, None]:
if values:
async with AsyncExitStack() as stack:
# create enter tasks with reference to spec
tasks = {
asyncio.create_task(
stack.enter_async_context(
value.cls.aenter(config, **value.kwargs)
if isinstance(value, ConfiguredManagedValue)
else value.aenter(config)
)
): key
for key, value in values.items()
}
# wait for all enter tasks
done, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
# build mapping from spec to result
yield {tasks[task]: task.result() for task in done}
else:
yield {}
def is_writable_managed_value(value: Any) -> TypeGuard[Type[WritableManagedValue]]:
return (isclass(value) and issubclass(value, WritableManagedValue)) or (
isinstance(value, ConfiguredManagedValue)
and issubclass(value.cls, WritableManagedValue)
)
ChannelKeyPlaceholder = object()
ChannelTypePlaceholder = object()
@@ -1,11 +1,10 @@
from typing import Annotated
from langgraph.managed.base import ManagedValue
from langgraph.pregel.types import PregelExecutableTask
class IsLastStepManager(ManagedValue[bool]):
def __call__(self, step: int, task: PregelExecutableTask) -> bool:
def __call__(self, step: int) -> bool:
return step == self.config["recursion_limit"] - 1
@@ -0,0 +1,126 @@
import collections.abc
from contextlib import asynccontextmanager, contextmanager
from typing import (
Any,
AsyncIterator,
Iterator,
Optional,
Sequence,
Type,
)
from langchain_core.runnables import RunnableConfig
from typing_extensions import NotRequired, Required, Self
from langgraph.constants import CONFIG_KEY_STORE
from langgraph.errors import InvalidUpdateError
from langgraph.managed.base import (
ChannelKeyPlaceholder,
ChannelTypePlaceholder,
ConfiguredManagedValue,
WritableManagedValue,
)
from langgraph.store.base import BaseStore
V = dict[str, Any]
Value = dict[str, V]
Update = dict[str, Optional[V]]
# Adapted from typing_extensions
def _strip_extras(t):
"""Strips Annotated, Required and NotRequired from a given type."""
if hasattr(t, "__origin__"):
return _strip_extras(t.__origin__)
if hasattr(t, "__origin__") and t.__origin__ in (Required, NotRequired):
return _strip_extras(t.__args__[0])
return t
class SharedValue(WritableManagedValue[Value, Update]):
@staticmethod
def on(scope: str) -> ConfiguredManagedValue:
return ConfiguredManagedValue(
SharedValue,
{
"scope": scope,
"key": ChannelKeyPlaceholder,
"typ": ChannelTypePlaceholder,
},
)
@classmethod
@contextmanager
def enter(cls, config: RunnableConfig, **kwargs: Any) -> Iterator[Self]:
with super().enter(config, **kwargs) as value:
if value.store is not None:
saved = value.store.list([value.ns])
value.value = saved[value.ns] or {}
yield value
@classmethod
@asynccontextmanager
async def aenter(cls, config: RunnableConfig, **kwargs: Any) -> AsyncIterator[Self]:
async with super().aenter(config, **kwargs) as value:
if value.store is not None:
saved = await value.store.alist([value.ns])
value.value = saved[value.ns] or {}
yield value
def __init__(
self, config: RunnableConfig, *, typ: Type[Any], scope: str, key: str
) -> None:
if typ := _strip_extras(typ):
if typ not in (
dict,
collections.abc.Mapping,
collections.abc.MutableMapping,
):
raise ValueError("SharedValue must be a dict")
self.scope = scope
self.config = config
self.value: Value = {}
self.store: BaseStore = config["configurable"].get(CONFIG_KEY_STORE)
if self.store is None:
self.ns: Optional[str] = None
elif scope_value := config["configurable"].get(self.scope):
self.ns = f"scoped:{scope}:{key}:{scope_value}"
else:
raise ValueError(
f"Scope {scope} for shared state key not in config.configurable"
)
def __call__(self, step: int) -> Value:
return self.value.copy()
def _process_update(
self, values: Sequence[Update]
) -> list[tuple[str, str, Optional[dict[str, Any]]]]:
writes = []
for vv in values:
for k, v in vv.items():
if v is None:
if k in self.value:
self.value[k] = None
writes.append((self.ns, k, None))
elif not isinstance(v, dict):
raise InvalidUpdateError("Received a non-dict value")
else:
self.value[k] = v
writes.append((self.ns, k, v))
return writes
def update(self, values: Sequence[Update]) -> None:
if self.store is None:
self._process_update(values)
else:
return self.store.put(self._process_update(values))
async def aupdate(self, writes: Sequence[Update]) -> None:
if self.store is None:
self._process_update(writes)
else:
return await self.store.aput(self._process_update(writes))
+194 -107
View File
@@ -52,11 +52,6 @@ from langgraph.channels.base import (
BaseChannel,
)
from langgraph.channels.context import Context
from langgraph.channels.last_value import LastValue
from langgraph.channels.manager import (
AsyncChannelsManager,
ChannelsManager,
)
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
copy_checkpoint,
@@ -68,31 +63,31 @@ from langgraph.constants import (
CONFIG_KEY_READ,
CONFIG_KEY_RESUMING,
CONFIG_KEY_SEND,
ERROR,
INTERRUPT,
Interrupt,
)
from langgraph.errors import GraphRecursionError, InvalidUpdateError
from langgraph.managed.base import (
AsyncManagedValuesManager,
ManagedValuesManager,
ManagedValueSpec,
is_managed_value,
)
from langgraph.errors import GraphInterrupt, GraphRecursionError, InvalidUpdateError
from langgraph.managed.base import ManagedValueSpec
from langgraph.pregel.algo import (
apply_writes,
local_read,
prepare_next_tasks,
should_interrupt,
)
from langgraph.pregel.debug import (
map_debug_task_results,
print_step_checkpoint,
print_step_tasks,
print_step_writes,
tasks_w_writes,
)
from langgraph.pregel.io import (
map_output_updates,
read_channels,
)
from langgraph.pregel.loop import AsyncPregelLoop, SyncPregelLoop
from langgraph.pregel.manager import AsyncChannelsManager, ChannelsManager
from langgraph.pregel.read import PregelNode
from langgraph.pregel.retry import RetryPolicy, arun_with_retry, run_with_retry
from langgraph.pregel.types import (
@@ -104,6 +99,7 @@ from langgraph.pregel.types import (
from langgraph.pregel.utils import get_new_channel_versions
from langgraph.pregel.validate import validate_graph, validate_keys
from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
from langgraph.store.base import BaseStore
WriteValue = Union[
Runnable[Input, Output],
@@ -192,7 +188,9 @@ class Pregel(
):
nodes: Mapping[str, PregelNode]
channels: Mapping[str, BaseChannel] = Field(default_factory=dict)
channels: Mapping[str, Union[BaseChannel, ManagedValueSpec]] = Field(
default_factory=dict
)
auto_validate: bool = True
@@ -219,6 +217,9 @@ class Pregel(
checkpointer: Optional[BaseCheckpointSaver] = None
"""Checkpointer used to save and load graph state. Defaults to None."""
store: Optional[BaseStore] = None
"""Memory store to use for SharedValues. Defaults to None."""
retry_policy: Optional[RetryPolicy] = None
"""Retry policy to use when running tasks. Set to None to disable."""
@@ -342,16 +343,6 @@ class Pregel(
k for k in self.channels if not isinstance(self.channels[k], Context)
]
@property
def managed_values_dict(self) -> dict[str, ManagedValueSpec]:
return {
k: v
for node in self.nodes.values()
if isinstance(node.channels, dict)
for k, v in node.channels.items()
if is_managed_value(v)
}
def get_state(self, config: RunnableConfig) -> StateSnapshot:
"""Get the current state of the graph."""
if not self.checkpointer:
@@ -360,25 +351,20 @@ class Pregel(
saved = self.checkpointer.get_tuple(config)
checkpoint = saved.checkpoint if saved else empty_checkpoint()
config = saved.config if saved else config
with ChannelsManager(
{
k: LastValue(None) if isinstance(c, Context) else c
for k, c in self.channels.items()
},
checkpoint,
config,
) as channels, ManagedValuesManager(
self.managed_values_dict, ensure_config(config)
) as managed:
with ChannelsManager(self.channels, checkpoint, config, skip_context=True) as (
channels,
managed,
):
next_tasks = prepare_next_tasks(
checkpoint,
self.nodes,
channels,
managed,
config,
-1,
saved.metadata.get("step", -1) + 1 if saved else -1,
for_execution=False,
)
return StateSnapshot(
read_channels(channels, self.stream_channels_asis),
tuple(t.name for t in next_tasks),
@@ -386,6 +372,7 @@ class Pregel(
saved.metadata if saved else None,
saved.checkpoint["ts"] if saved else None,
saved.parent_config if saved else None,
tasks_w_writes(next_tasks, saved.pending_writes),
)
async def aget_state(self, config: RunnableConfig) -> StateSnapshot:
@@ -398,22 +385,15 @@ class Pregel(
config = saved.config if saved else config
async with AsyncChannelsManager(
{
k: LastValue(None) if isinstance(c, Context) else c
for k, c in self.channels.items()
},
checkpoint,
config,
) as channels, AsyncManagedValuesManager(
self.managed_values_dict, ensure_config(config)
) as managed:
self.channels, checkpoint, config, skip_context=True
) as (channels, managed):
next_tasks = prepare_next_tasks(
checkpoint,
self.nodes,
channels,
managed,
config,
-1,
saved.metadata.get("step", -1) + 1 if saved else -1,
for_execution=False,
)
return StateSnapshot(
@@ -423,6 +403,7 @@ class Pregel(
saved.metadata if saved else None,
saved.checkpoint["ts"] if saved else None,
saved.parent_config if saved else None,
tasks_w_writes(next_tasks, saved.pending_writes),
)
def get_state_history(
@@ -441,26 +422,23 @@ class Pregel(
and signature(self.checkpointer.list).parameters.get("filter") is None
):
raise ValueError("Checkpointer does not support filtering")
for config, checkpoint, metadata, parent_config, _ in self.checkpointer.list(
config, before=before, limit=limit, filter=filter
):
for (
config,
checkpoint,
metadata,
parent_config,
pending_writes,
) in self.checkpointer.list(config, before=before, limit=limit, filter=filter):
with ChannelsManager(
{
k: LastValue(None) if isinstance(c, Context) else c
for k, c in self.channels.items()
},
checkpoint,
config,
) as channels, ManagedValuesManager(
self.managed_values_dict, ensure_config(config)
) as managed:
self.channels, checkpoint, config, skip_context=True
) as (channels, managed):
next_tasks = prepare_next_tasks(
checkpoint,
self.nodes,
channels,
managed,
config,
-1,
metadata.get("step", -1) + 1,
for_execution=False,
)
yield StateSnapshot(
@@ -470,6 +448,7 @@ class Pregel(
metadata,
checkpoint["ts"],
parent_config,
tasks_w_writes(next_tasks, pending_writes),
)
async def aget_state_history(
@@ -493,25 +472,18 @@ class Pregel(
checkpoint,
metadata,
parent_config,
_,
pending_writes,
) in self.checkpointer.alist(config, before=before, limit=limit, filter=filter):
async with AsyncChannelsManager(
{
k: LastValue(None) if isinstance(c, Context) else c
for k, c in self.channels.items()
},
checkpoint,
config,
) as channels, AsyncManagedValuesManager(
self.managed_values_dict, ensure_config(config)
) as managed:
self.channels, checkpoint, config, skip_context=True
) as (channels, managed):
next_tasks = prepare_next_tasks(
checkpoint,
self.nodes,
channels,
managed,
config,
-1,
metadata.get("step", -1) + 1,
for_execution=False,
)
yield StateSnapshot(
@@ -521,6 +493,7 @@ class Pregel(
metadata,
checkpoint["ts"],
parent_config,
tasks_w_writes(next_tasks, pending_writes),
)
def update_state(
@@ -540,7 +513,7 @@ class Pregel(
saved = self.checkpointer.get_tuple(config)
checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
checkpoint_previous_versions = (
saved.checkpoint["channel_versions"] if saved else {}
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
@@ -566,7 +539,7 @@ class Pregel(
create_checkpoint(checkpoint, None, step),
{
"source": "update",
"step": step,
"step": step + 1,
"writes": {},
},
{},
@@ -596,7 +569,10 @@ class Pregel(
if as_node not in self.nodes:
raise InvalidUpdateError(f"Node {as_node} does not exist")
# update channels
with ChannelsManager(self.channels, checkpoint, config) as channels:
with ChannelsManager(self.channels, checkpoint, config) as (
channels,
managed,
):
# create task to run all writers of the chosen node
writers = self.nodes[as_node].get_writers()
if not writers:
@@ -627,22 +603,46 @@ class Pregel(
),
)
# apply to checkpoint and save
apply_writes(
assert not apply_writes(
checkpoint, channels, [task], self.checkpointer.get_next_version
)
new_versions = get_new_channel_versions(
checkpoint_previous_versions, checkpoint["channel_versions"]
)
), "Can't write to SharedValues from update_state"
checkpoint = create_checkpoint(checkpoint, channels, step + 1)
# check interrupt before
if tasks := should_interrupt(
checkpoint,
self.interrupt_before_nodes,
prepare_next_tasks(
checkpoint,
self.nodes,
channels,
managed,
config,
step + 2,
for_execution=False,
),
):
for t in tasks:
self.checkpointer.put_writes(
{
"configurable": {
**checkpoint_config["configurable"],
"checkpoint_id": checkpoint["id"],
}
},
[(INTERRUPT, Interrupt("before"))],
t.id,
)
return self.checkpointer.put(
checkpoint_config,
create_checkpoint(checkpoint, channels, step + 1),
checkpoint,
{
"source": "update",
"step": step + 1,
"writes": {as_node: values},
},
new_versions,
get_new_channel_versions(
checkpoint_previous_versions, checkpoint["channel_versions"]
),
)
async def aupdate_state(
@@ -658,7 +658,7 @@ class Pregel(
saved = await self.checkpointer.aget_tuple(config)
checkpoint = copy_checkpoint(saved.checkpoint) if saved else empty_checkpoint()
checkpoint_previous_versions = (
saved.checkpoint["channel_versions"] if saved else {}
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
@@ -684,7 +684,7 @@ class Pregel(
create_checkpoint(checkpoint, None, step),
{
"source": "update",
"step": step,
"step": step + 1,
"writes": {},
},
{},
@@ -712,7 +712,10 @@ class Pregel(
if as_node not in self.nodes:
raise InvalidUpdateError(f"Node {as_node} does not exist")
# update channels, acting as the chosen node
async with AsyncChannelsManager(self.channels, checkpoint, config) as channels:
async with AsyncChannelsManager(self.channels, checkpoint, config) as (
channels,
managed,
):
# create task to run all writers of the chosen node
writers = self.nodes[as_node].get_writers()
if not writers:
@@ -743,22 +746,50 @@ class Pregel(
),
)
# apply to checkpoint and save
apply_writes(
assert not apply_writes(
checkpoint, channels, [task], self.checkpointer.get_next_version
)
new_versions = get_new_channel_versions(
checkpoint_previous_versions, checkpoint["channel_versions"]
)
), "Can't write to SharedValues from update_state"
checkpoint = create_checkpoint(checkpoint, channels, step + 1)
# check interrupt before
if tasks := should_interrupt(
checkpoint,
self.interrupt_before_nodes,
prepare_next_tasks(
checkpoint,
self.nodes,
channels,
managed,
config,
step + 2,
for_execution=False,
),
):
await asyncio.gather(
*(
self.checkpointer.aput_writes(
{
"configurable": {
**checkpoint_config["configurable"],
"checkpoint_id": checkpoint["id"],
}
},
[(INTERRUPT, Interrupt("before"))],
t.id,
)
for t in tasks
)
)
return await self.checkpointer.aput(
checkpoint_config,
create_checkpoint(checkpoint, channels, step + 1),
checkpoint,
{
"source": "update",
"step": step + 1,
"writes": {as_node: values},
},
new_versions,
get_new_channel_versions(
checkpoint_previous_versions, checkpoint["channel_versions"]
),
)
def _defaults(
@@ -921,7 +952,13 @@ class Pregel(
)
with SyncPregelLoop(
input, config=config, checkpointer=checkpointer, graph=self
input,
config=config,
store=self.store,
checkpointer=checkpointer,
graph=self,
nodes=self.nodes,
specs=self.channels,
) as loop:
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates
@@ -935,7 +972,7 @@ class Pregel(
manager=run_manager,
):
# debug flag
if self.debug:
if debug:
print_step_checkpoint(
loop.checkpoint_metadata,
loop.channels,
@@ -965,6 +1002,7 @@ class Pregel(
for task in loop.tasks
if not task.writes
}
all_futures = futures.copy()
end_time = (
self.step_timeout + time.monotonic()
if self.step_timeout
@@ -986,10 +1024,14 @@ class Pregel(
break # timed out
for fut in done:
task = futures.pop(fut)
if fut.exception() is not None:
# we got an exception, break out of while loop
# exception will be handled in panic_or_proceed
futures.clear()
if exc := _exception(fut):
# save error to checkpointer
if isinstance(exc, GraphInterrupt):
loop.put_writes(
task.id, [(INTERRUPT, i) for i in exc.args[0]]
)
else:
loop.put_writes(task.id, [(ERROR, exc)])
else:
# save task writes to checkpointer
loop.put_writes(task.id, task.writes)
@@ -1013,9 +1055,11 @@ class Pregel(
else:
# remove references to loop vars
del fut, task
if _should_stop_others(done):
break
# panic on failure or timeout
_panic_or_proceed(done, inflight, loop.step)
_panic_or_proceed(all_futures, loop.step)
# don't keep futures around in memory longer than needed
del done, inflight, futures
# debug flag
@@ -1164,7 +1208,13 @@ class Pregel(
debug=debug,
)
async with AsyncPregelLoop(
input, config=config, checkpointer=checkpointer, graph=self
input,
config=config,
store=self.store,
checkpointer=checkpointer,
graph=self,
nodes=self.nodes,
specs=self.channels,
) as loop:
aioloop = asyncio.get_event_loop()
# Similarly to Bulk Synchronous Parallel / Pregel model
@@ -1179,7 +1229,7 @@ class Pregel(
manager=run_manager,
):
# debug flag
if self.debug:
if debug:
print_step_checkpoint(
loop.checkpoint_metadata,
loop.channels,
@@ -1212,6 +1262,7 @@ class Pregel(
for task in loop.tasks
if not task.writes
}
all_futures = futures.copy()
end_time = (
self.step_timeout + aioloop.time()
if self.step_timeout
@@ -1231,10 +1282,14 @@ class Pregel(
break # timed out
for fut in done:
task = futures.pop(fut)
if fut.exception() is not None:
# we got an exception, break out of while loop
# exception will be handled in panic_or_proceed
futures.clear()
if exc := _exception(fut):
# save error to checkpointer
if isinstance(exc, GraphInterrupt):
loop.put_writes(
task.id, [(INTERRUPT, i) for i in exc.args[0]]
)
else:
loop.put_writes(task.id, [(ERROR, exc)])
else:
# save task writes to checkpointer
loop.put_writes(task.id, task.writes)
@@ -1260,9 +1315,11 @@ class Pregel(
else:
# remove references to loop vars
del fut, task
if _should_stop_others(done):
break
# panic on failure or timeout
_panic_or_proceed(done, inflight, loop.step, asyncio.TimeoutError)
_panic_or_proceed(all_futures, loop.step, asyncio.TimeoutError)
# don't keep futures around in memory longer than needed
del done, inflight, futures
# debug flag
@@ -1403,15 +1460,45 @@ class Pregel(
return chunks
def _panic_or_proceed(
def _should_stop_others(
done: Union[set[concurrent.futures.Future[Any]], set[asyncio.Task[Any]]],
inflight: Union[set[concurrent.futures.Future[Any]], set[asyncio.Task[Any]]],
) -> bool:
for fut in done:
if fut.cancelled():
return True
if exc := fut.exception():
return not isinstance(exc, GraphInterrupt)
else:
return False
def _exception(
fut: Union[concurrent.futures.Future[Any], asyncio.Task[Any]],
) -> Optional[BaseException]:
if fut.cancelled():
if isinstance(fut, asyncio.Task):
return asyncio.CancelledError()
else:
return concurrent.futures.CancelledError()
else:
return fut.exception()
def _panic_or_proceed(
futs: Union[set[concurrent.futures.Future[Any]], set[asyncio.Task[Any]]],
step: int,
timeout_exc_cls: Type[Exception] = TimeoutError,
) -> None:
done: set[Union[concurrent.futures.Future[Any], asyncio.Task[Any]]] = set()
inflight: set[Union[concurrent.futures.Future[Any], asyncio.Task[Any]]] = set()
for fut in futs:
if fut.done():
done.add(fut)
else:
inflight.add(fut)
while done:
# if any task failed
if exc := done.pop().exception():
if exc := _exception(done.pop()):
# cancel all pending tasks
while inflight:
inflight.pop().cancel()
+85 -71
View File
@@ -25,7 +25,6 @@ from langchain_core.runnables.config import (
from langgraph.channels.base import BaseChannel
from langgraph.channels.context import Context
from langgraph.channels.manager import ChannelsManager
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
@@ -38,6 +37,7 @@ from langgraph.constants import (
CONFIG_KEY_READ,
CONFIG_KEY_RESUMING,
CONFIG_KEY_SEND,
CONFIG_KEY_TASK_ID,
INTERRUPT,
RESERVED,
TAG_HIDDEN,
@@ -45,11 +45,12 @@ from langgraph.constants import (
Send,
)
from langgraph.errors import EmptyChannelError, InvalidUpdateError
from langgraph.managed.base import ManagedValueMapping, is_managed_value
from langgraph.managed.base import ManagedValueMapping
from langgraph.pregel.io import read_channel, read_channels
from langgraph.pregel.log import logger
from langgraph.pregel.manager import ChannelsManager
from langgraph.pregel.read import PregelNode
from langgraph.pregel.types import All, PregelExecutableTask, PregelTaskDescription
from langgraph.pregel.types import All, PregelExecutableTask, PregelTask
class WritesProtocol(Protocol):
@@ -68,26 +69,28 @@ def should_interrupt(
checkpoint: Checkpoint,
interrupt_nodes: Union[All, Sequence[str]],
tasks: list[PregelExecutableTask],
) -> bool:
) -> list[PregelExecutableTask]:
version_type = type(next(iter(checkpoint["channel_versions"].values()), None))
null_version = version_type()
seen = checkpoint["versions_seen"].get(INTERRUPT, {})
# interrupt if any channel has been updated since last interrupt
any_updates_since_prev_interrupt = any(
version > seen.get(chan, null_version)
for chan, version in checkpoint["channel_versions"].items()
)
# and any triggered node is in interrupt_nodes list
return (
# interrupt if any channel has been updated since last interrupt
any(
version > seen.get(chan, null_version)
for chan, version in checkpoint["channel_versions"].items()
)
# and any triggered node is in interrupt_nodes list
and any(
task.name
[
task
for task in tasks
if (
(not task.config or TAG_HIDDEN not in task.config.get("tags"))
if interrupt_nodes == "*"
else task.name in interrupt_nodes
)
)
]
if any_updates_since_prev_interrupt
else []
)
@@ -102,11 +105,10 @@ def local_read(
if fresh:
new_checkpoint = create_checkpoint(copy_checkpoint(checkpoint), channels, -1)
context_channels = {k: v for k, v in channels.items() if isinstance(v, Context)}
with ChannelsManager(
{k: v for k, v in channels.items() if k not in context_channels},
new_checkpoint,
config,
) as channels:
with ChannelsManager(channels, new_checkpoint, config, skip_context=True) as (
channels,
_,
):
all_channels = {**channels, **context_channels}
apply_writes(new_checkpoint, all_channels, [task], None)
return read_channels(all_channels, select)
@@ -118,6 +120,7 @@ def local_write(
commit: Callable[[Sequence[tuple[str, Any]]], None],
processes: Mapping[str, PregelNode],
channels: Mapping[str, BaseChannel],
managed: ManagedValueMapping,
writes: Sequence[tuple[str, Any]],
) -> None:
for chan, value in writes:
@@ -128,7 +131,7 @@ def local_write(
)
if value.node not in processes:
raise InvalidUpdateError(f"Invalid node name {value.node} in packet")
elif chan not in channels:
elif chan not in channels and chan not in managed:
logger.warning(f"Skipping write for channel '{chan}' which has no readers")
commit(writes)
@@ -142,7 +145,7 @@ def apply_writes(
channels: Mapping[str, BaseChannel],
tasks: Sequence[WritesProtocol],
get_next_version: Optional[Callable[[int, BaseChannel], int]],
) -> None:
) -> dict[str, list[Any]]:
# update seen versions
for task in tasks:
checkpoint["versions_seen"].setdefault(task.name, {}).update(
@@ -158,6 +161,7 @@ def apply_writes(
max_version = max(checkpoint["channel_versions"].values())
else:
max_version = None
# Consume all channels that were read
for chan in {
chan for task in tasks for chan in task.triggers if chan not in RESERVED
@@ -174,12 +178,15 @@ def apply_writes(
# Group writes by channel
pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
pending_writes_by_managed: dict[str, list[Any]] = defaultdict(list)
for task in tasks:
for chan, val in task.writes:
if chan == TASKS:
checkpoint["pending_sends"].append(val)
else:
elif chan in channels:
pending_writes_by_channel[chan].append(val)
else:
pending_writes_by_managed[chan].append(val)
# Find the highest version of all channels
if checkpoint["channel_versions"]:
@@ -211,6 +218,9 @@ def apply_writes(
max_version, channels[chan]
)
# Return managed values writes to be applied externally
return pending_writes_by_managed
@overload
def prepare_next_tasks(
@@ -225,7 +235,7 @@ def prepare_next_tasks(
is_resuming: bool = False,
checkpointer: Literal[None] = None,
manager: Literal[None] = None,
) -> list[PregelTaskDescription]:
) -> list[PregelTask]:
...
@@ -258,9 +268,9 @@ def prepare_next_tasks(
is_resuming: bool = False,
checkpointer: Optional[BaseCheckpointSaver] = None,
manager: Union[None, ParentRunManager, AsyncParentRunManager] = None,
) -> Union[list[PregelTaskDescription], list[PregelExecutableTask]]:
) -> Union[list[PregelTask], list[PregelExecutableTask]]:
parent_ns = config.get("configurable", {}).get("checkpoint_ns", "")
tasks: Union[list[PregelTaskDescription], list[PregelExecutableTask]] = []
tasks: Union[list[PregelTask], list[PregelExecutableTask]] = []
# Consume pending packets
for packet in checkpoint["pending_sends"]:
if not isinstance(packet, Send):
@@ -269,24 +279,25 @@ def prepare_next_tasks(
if packet.node not in processes:
logger.warn(f"Ignoring unknown node name {packet.node} in pending sends")
continue
# create task id
triggers = [TASKS]
metadata = {
"langgraph_step": step,
"langgraph_node": packet.node,
"langgraph_triggers": triggers,
"langgraph_task_idx": len(tasks),
}
checkpoint_ns = (
f"{parent_ns}{CHECKPOINT_NAMESPACE_SEPARATOR}{packet.node}"
if parent_ns
else packet.node
)
task_id = str(
uuid5(UUID(checkpoint["id"]), json.dumps((checkpoint_ns, metadata)))
)
if for_execution:
proc = processes[packet.node]
if node := proc.get_node():
triggers = [TASKS]
metadata = {
"langgraph_step": step,
"langgraph_node": packet.node,
"langgraph_triggers": triggers,
"langgraph_task_idx": len(tasks),
}
checkpoint_ns = (
f"{parent_ns}{CHECKPOINT_NAMESPACE_SEPARATOR}{packet.node}"
if parent_ns
else packet.node
)
task_id = str(
uuid5(UUID(checkpoint["id"]), json.dumps((checkpoint_ns, metadata)))
)
writes = deque()
tasks.append(
PregelExecutableTask(
@@ -307,9 +318,14 @@ def prepare_next_tasks(
else None
),
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write, writes.extend, processes, channels
local_write,
writes.extend,
processes,
channels,
managed,
),
CONFIG_KEY_READ: partial(
local_read,
@@ -328,7 +344,7 @@ def prepare_next_tasks(
)
)
else:
tasks.append(PregelTaskDescription(packet.node))
tasks.append(PregelTask(task_id, packet.node))
# Check if any processes should be run in next step
# If so, prepare the values to be passed to them
version_type = type(next(iter(checkpoint["channel_versions"].values()), None))
@@ -356,26 +372,27 @@ def prepare_next_tasks(
except StopIteration:
continue
# create task id
metadata = {
"langgraph_step": step,
"langgraph_node": name,
"langgraph_triggers": triggers,
"langgraph_task_idx": len(tasks),
}
checkpoint_ns = (
f"{parent_ns}{CHECKPOINT_NAMESPACE_SEPARATOR}{name}"
if parent_ns
else name
)
task_id = str(
uuid5(
UUID(checkpoint["id"]),
json.dumps((checkpoint_ns, metadata)),
)
)
if for_execution:
if node := proc.get_node():
metadata = {
"langgraph_step": step,
"langgraph_node": name,
"langgraph_triggers": triggers,
"langgraph_task_idx": len(tasks),
}
checkpoint_ns = (
f"{parent_ns}{CHECKPOINT_NAMESPACE_SEPARATOR}{name}"
if parent_ns
else name
)
task_id = str(
uuid5(
UUID(checkpoint["id"]),
json.dumps((checkpoint_ns, metadata)),
)
)
writes = deque()
tasks.append(
PregelExecutableTask(
@@ -396,9 +413,14 @@ def prepare_next_tasks(
else None
),
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write, writes.extend, processes, channels
local_write,
writes.extend,
processes,
channels,
managed,
),
CONFIG_KEY_READ: partial(
local_read,
@@ -424,7 +446,7 @@ def prepare_next_tasks(
)
)
else:
tasks.append(PregelTaskDescription(name))
tasks.append(PregelTask(task_id, name))
return tasks
@@ -447,18 +469,10 @@ def _proc_input(
chan,
catch=chan not in proc.triggers,
)
if chan in channels
else managed[k](step)
for k, chan in proc.channels.items()
if isinstance(chan, str)
}
managed_values = {}
for key, chan in proc.channels.items():
if is_managed_value(chan):
managed_values[key] = managed[key](
step, PregelTaskDescription(name)
)
val.update(managed_values)
except EmptyChannelError:
return
elif isinstance(proc.channels, list):
+60 -6
View File
@@ -9,10 +9,10 @@ from langchain_core.runnables.config import RunnableConfig
from langchain_core.utils.input import get_bolded_text, get_colored_text
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import CheckpointMetadata
from langgraph.constants import TAG_HIDDEN
from langgraph.checkpoint.base import Checkpoint, CheckpointMetadata, PendingWrite
from langgraph.constants import ERROR, INTERRUPT, TAG_HIDDEN
from langgraph.pregel.io import read_channels
from langgraph.pregel.types import PregelExecutableTask
from langgraph.pregel.types import PregelExecutableTask, PregelTask
class TaskPayload(TypedDict):
@@ -28,10 +28,19 @@ class TaskResultPayload(TypedDict):
result: list[tuple[str, Any]]
class CheckpointTask(TypedDict):
id: str
name: str
error: Optional[str]
interrupts: list[dict]
class CheckpointPayload(TypedDict):
config: Optional[RunnableConfig]
metadata: CheckpointMetadata
values: dict[str, Any]
next: list[str]
tasks: list[CheckpointTask]
class DebugOutputBase(TypedDict):
@@ -118,16 +127,33 @@ def map_debug_checkpoint(
channels: Mapping[str, BaseChannel],
stream_channels: Union[str, Sequence[str]],
metadata: CheckpointMetadata,
checkpoint: Checkpoint,
tasks: list[PregelExecutableTask],
pending_writes: list[PendingWrite],
) -> Iterator[DebugOutputCheckpoint]:
ts = datetime.now(timezone.utc).isoformat()
yield {
"type": "checkpoint",
"timestamp": ts,
"timestamp": checkpoint["ts"],
"step": step,
"payload": {
"config": config,
"values": read_channels(channels, stream_channels),
"metadata": metadata,
"next": [t.name for t in tasks],
"tasks": [
{
"id": t.id,
"name": t.name,
"error": t.error,
}
if t.error
else {
"id": t.id,
"name": t.name,
"interrupts": t.interrupts,
}
for t in tasks_w_writes(tasks, pending_writes)
],
},
}
@@ -166,10 +192,38 @@ def print_step_writes(
def print_step_checkpoint(
step: int, channels: Mapping[str, BaseChannel], whitelist: Sequence[str]
metadata: CheckpointMetadata,
channels: Mapping[str, BaseChannel],
whitelist: Sequence[str],
) -> None:
step = metadata["step"]
print(
f"{get_colored_text(f'[{step}:checkpoint]', color='blue')} "
+ get_bolded_text(f"State at the end of step {step}:\n")
+ pformat(read_channels(channels, whitelist), depth=3)
)
def tasks_w_writes(
tasks: list[PregelExecutableTask],
pending_writes: Optional[list[PendingWrite]],
) -> tuple[PregelTask, ...]:
pending_writes = pending_writes or []
return tuple(
PregelTask(
task.id,
task.name,
next(
(
exc
for tid, n, exc in pending_writes
if tid == task.id and n == ERROR
),
None,
),
tuple(
v for tid, n, v in pending_writes if tid == task.id and n == INTERRUPT
),
)
for task in tasks
)
+106 -46
View File
@@ -18,6 +18,7 @@ from typing import (
Type,
TypeVar,
Union,
cast,
)
from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
@@ -25,10 +26,6 @@ from langchain_core.runnables import RunnableConfig
from typing_extensions import Self
from langgraph.channels.base import BaseChannel
from langgraph.channels.manager import (
AsyncChannelsManager,
ChannelsManager,
)
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
@@ -39,12 +36,19 @@ from langgraph.checkpoint.base import (
create_checkpoint,
empty_checkpoint,
)
from langgraph.constants import CONFIG_KEY_READ, CONFIG_KEY_RESUMING, INPUT, INTERRUPT
from langgraph.constants import (
CONFIG_KEY_READ,
CONFIG_KEY_RESUMING,
ERROR,
INPUT,
INTERRUPT,
Interrupt,
)
from langgraph.errors import EmptyInputError, GraphInterrupt
from langgraph.managed.base import (
AsyncManagedValuesManager,
ManagedValueMapping,
ManagedValuesManager,
ManagedValueSpec,
WritableManagedValue,
)
from langgraph.pregel.algo import (
PregelTaskWrites,
@@ -60,8 +64,12 @@ from langgraph.pregel.executor import (
Submit,
)
from langgraph.pregel.io import map_input, map_output_updates, map_output_values, single
from langgraph.pregel.manager import AsyncChannelsManager, ChannelsManager
from langgraph.pregel.read import PregelNode
from langgraph.pregel.types import PregelExecutableTask
from langgraph.pregel.utils import get_new_channel_versions
from langgraph.store.base import BaseStore
from langgraph.store.batch import AsyncBatchedStore
if TYPE_CHECKING:
from langgraph.pregel import Pregel
@@ -76,7 +84,12 @@ EMPTY_SEQ = ()
class PregelLoop:
input: Optional[Any]
config: RunnableConfig
store: Optional[BaseStore]
checkpointer: Optional[BaseCheckpointSaver]
nodes: Mapping[str, PregelNode]
specs: Mapping[str, Union[BaseChannel, ManagedValueSpec]]
is_nested: bool
checkpointer_get_next_version: Callable[[Optional[V]], V]
checkpointer_put_writes: Optional[
Callable[[RunnableConfig, Sequence[tuple[str, Any]], str], Any]
@@ -93,7 +106,7 @@ class PregelLoop:
]
]
graph: "Pregel"
store: Optional[BaseStore]
submit: Submit
channels: Mapping[str, BaseChannel]
managed: ManagedValueMapping
@@ -111,7 +124,6 @@ class PregelLoop:
]
tasks: Sequence[PregelExecutableTask]
stream: deque[Tuple[str, Any]]
is_nested: bool
# public
@@ -120,16 +132,20 @@ class PregelLoop:
input: Optional[Any],
*,
config: RunnableConfig,
store: Optional[BaseStore],
checkpointer: Optional[BaseCheckpointSaver],
nodes: Mapping[str, PregelNode],
specs: Mapping[str, Union[BaseChannel, ManagedValueSpec]],
graph: "Pregel",
) -> None:
self.stream = deque()
self.input = input
self.config = config
self.store = store
self.checkpointer = checkpointer
self.graph = graph
# TODO if managed values no longer needs graph we can replace with
# managed_specs, channel_specs
self.nodes = nodes
self.specs = specs
self.is_nested = CONFIG_KEY_READ in self.config.get("configurable", {})
def mark_tasks_scheduled(self, tasks: Sequence[PregelExecutableTask]) -> None:
@@ -175,12 +191,15 @@ class PregelLoop:
elif all(task.writes for task in self.tasks):
writes = [w for t in self.tasks for w in t.writes]
# all tasks have finished
apply_writes(
mv_writes = apply_writes(
self.checkpoint,
self.channels,
self.tasks,
self.checkpointer_get_next_version,
)
# apply writes to managed values
for key, values in mv_writes.items():
self._update_mv(key, values)
# produce values output
self.stream.extend(
("values", v)
@@ -200,10 +219,13 @@ class PregelLoop:
}
)
# after execution, check if we should interrupt
if should_interrupt(self.checkpoint, interrupt_after, self.tasks):
if tasks := should_interrupt(self.checkpoint, interrupt_after, self.tasks):
self.status = "interrupt_after"
interrupts = [(t.id, Interrupt("after")) for t in tasks]
for tid, interrupt in interrupts:
self.put_writes(tid, [(INTERRUPT, interrupt)])
if self.is_nested:
raise GraphInterrupt(self)
raise GraphInterrupt([i[1] for i in interrupts])
else:
return False
else:
@@ -217,7 +239,7 @@ class PregelLoop:
# prepare next tasks
self.tasks = prepare_next_tasks(
self.checkpoint,
self.graph.nodes,
self.nodes,
self.channels,
self.managed,
self.config,
@@ -228,6 +250,22 @@ class PregelLoop:
is_resuming=self.input is INPUT_RESUMING,
)
# produce debug output
if self._checkpointer_put_after_previous is not None:
self.stream.extend(
("debug", v)
for v in map_debug_checkpoint(
self.step - 1, # printing checkpoint for previous step
self.checkpoint_config,
self.channels,
self.graph.stream_channels_asis,
self.checkpoint_metadata,
self.checkpoint,
self.tasks,
self.checkpoint_pending_writes,
)
)
# if no more tasks, we're done
if not self.tasks:
self.status = "done"
@@ -236,6 +274,8 @@ class PregelLoop:
# if there are pending writes from a previous loop, apply them
if self.checkpoint_pending_writes:
for tid, k, v in self.checkpoint_pending_writes:
if k in (ERROR, INTERRUPT):
continue
if task := next((t for t in self.tasks if t.id == tid), None):
task.writes.append((k, v))
@@ -249,10 +289,13 @@ class PregelLoop:
)
# before execution, check if we should interrupt
if should_interrupt(self.checkpoint, interrupt_before, self.tasks):
if tasks := should_interrupt(self.checkpoint, interrupt_before, self.tasks):
self.status = "interrupt_before"
interrupts = [(t.id, Interrupt("before")) for t in tasks]
for tid, interrupt in interrupts:
self.put_writes(tid, [(INTERRUPT, interrupt)])
if self.is_nested:
raise GraphInterrupt()
raise GraphInterrupt([i[1] for i in interrupts])
else:
return False
@@ -284,7 +327,7 @@ class PregelLoop:
# discard any unfinished tasks from previous checkpoint
discard_tasks = prepare_next_tasks(
self.checkpoint,
self.graph.nodes,
self.nodes,
self.channels,
self.managed,
self.config,
@@ -293,12 +336,12 @@ class PregelLoop:
manager=None,
)
# apply input writes
apply_writes(
assert not apply_writes(
self.checkpoint,
self.channels,
discard_tasks + [PregelTaskWrites(INPUT, input_writes, [])],
self.checkpointer_get_next_version,
)
), "Can't write to SharedValues in graph input"
# save input checkpoint
self._put_checkpoint({"source": "input", "writes": self.input})
else:
@@ -361,27 +404,19 @@ class PregelLoop:
"checkpoint_id": self.checkpoint["id"],
},
}
# produce debug output
self.stream.extend(
("debug", v)
for v in map_debug_checkpoint(
self.step,
self.checkpoint_config,
self.channels,
self.graph.stream_channels_asis,
self.checkpoint_metadata,
)
)
# increment step
self.step += 1
def _update_mv(self, key: str, values: Sequence[Any]) -> None:
raise NotImplementedError
def _suppress_interrupt(
self,
exc_type: Optional[Type[BaseException]],
exc_value: Optional[BaseException],
traceback: Optional[TracebackType],
) -> Optional[bool]:
if exc_type is GraphInterrupt and not self.is_nested:
if isinstance(exc_value, GraphInterrupt) and not self.is_nested:
return True
@@ -391,12 +426,22 @@ class SyncPregelLoop(PregelLoop, ContextManager):
input: Optional[Any],
*,
config: RunnableConfig,
store: Optional[BaseStore],
checkpointer: Optional[BaseCheckpointSaver],
nodes: Mapping[str, PregelNode],
specs: Mapping[str, Union[BaseChannel, ManagedValueSpec]],
graph: "Pregel",
) -> None:
super().__init__(input, config=config, checkpointer=checkpointer, graph=graph)
super().__init__(
input,
config=config,
checkpointer=checkpointer,
graph=graph,
store=store,
nodes=nodes,
specs=specs,
)
self.stack = ExitStack()
self.stack.push(self._suppress_interrupt)
if checkpointer:
self.checkpointer_get_next_version = checkpointer.get_next_version
self.checkpointer_put_writes = checkpointer.put_writes
@@ -419,6 +464,9 @@ class SyncPregelLoop(PregelLoop, ContextManager):
finally:
self.checkpointer.put(config, checkpoint, metadata, new_versions)
def _update_mv(self, key: str, values: Sequence[Any]) -> None:
return self.submit(cast(WritableManagedValue, self.managed[key]).update, values)
# context manager
def __enter__(self) -> Self:
@@ -438,12 +486,10 @@ class SyncPregelLoop(PregelLoop, ContextManager):
self.checkpoint_pending_writes = saved.pending_writes or []
self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
self.channels = self.stack.enter_context(
ChannelsManager(self.graph.channels, self.checkpoint, self.config)
)
self.managed = self.stack.enter_context(
ManagedValuesManager(self.graph.managed_values_dict, self.config)
self.channels, self.managed = self.stack.enter_context(
ChannelsManager(self.specs, self.checkpoint, self.config, self.store)
)
self.stack.push(self._suppress_interrupt)
self.status = "pending"
self.step = self.checkpoint_metadata["step"] + 1
self.stop = self.step + self.config["recursion_limit"] + 1
@@ -468,12 +514,23 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
input: Optional[Any],
*,
config: RunnableConfig,
store: Optional[BaseStore],
checkpointer: Optional[BaseCheckpointSaver],
nodes: Mapping[str, PregelNode],
specs: Mapping[str, Union[BaseChannel, ManagedValueSpec]],
graph: "Pregel",
) -> None:
super().__init__(input, config=config, checkpointer=checkpointer, graph=graph)
super().__init__(
input,
config=config,
checkpointer=checkpointer,
graph=graph,
store=store,
nodes=nodes,
specs=specs,
)
self.store = AsyncBatchedStore(self.store) if self.store else None
self.stack = AsyncExitStack()
self.stack.push(self._suppress_interrupt)
if checkpointer:
self.checkpointer_get_next_version = checkpointer.get_next_version
self.checkpointer_put_writes = checkpointer.aput_writes
@@ -496,6 +553,11 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
finally:
await self.checkpointer.aput(config, checkpoint, metadata, new_versions)
def _update_mv(self, key: str, values: Sequence[Any]) -> None:
return self.submit(
cast(WritableManagedValue, self.managed[key]).aupdate, values
)
# context manager
async def __aenter__(self) -> Self:
@@ -517,12 +579,10 @@ class AsyncPregelLoop(PregelLoop, AsyncContextManager):
self.checkpoint_pending_writes = saved.pending_writes or []
self.submit = await self.stack.enter_async_context(AsyncBackgroundExecutor())
self.channels = await self.stack.enter_async_context(
AsyncChannelsManager(self.graph.channels, self.checkpoint, self.config)
)
self.managed = await self.stack.enter_async_context(
AsyncManagedValuesManager(self.graph.managed_values_dict, self.config)
self.channels, self.managed = await self.stack.enter_async_context(
AsyncChannelsManager(self.specs, self.checkpoint, self.config, self.store)
)
self.stack.push(self._suppress_interrupt)
self.status = "pending"
self.step = self.checkpoint_metadata["step"] + 1
self.stop = self.step + self.config["recursion_limit"] + 1
+104
View File
@@ -0,0 +1,104 @@
import asyncio
from contextlib import AsyncExitStack, ExitStack, asynccontextmanager, contextmanager
from typing import AsyncIterator, Iterator, Mapping, Optional, Union
from langchain_core.runnables import RunnableConfig, patch_config
from langgraph.channels.base import BaseChannel
from langgraph.channels.context import Context
from langgraph.channels.last_value import LastValue
from langgraph.checkpoint.base import Checkpoint
from langgraph.constants import CONFIG_KEY_STORE
from langgraph.managed.base import (
ConfiguredManagedValue,
ManagedValueMapping,
ManagedValueSpec,
)
from langgraph.store.base import BaseStore
@contextmanager
def ChannelsManager(
specs: Mapping[str, Union[BaseChannel, ManagedValueSpec]],
checkpoint: Checkpoint,
config: RunnableConfig,
store: Optional[BaseStore] = None,
*,
skip_context: bool = False,
) -> Iterator[tuple[Mapping[str, BaseChannel], ManagedValueMapping]]:
"""Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
config_for_managed = patch_config(config, configurable={CONFIG_KEY_STORE: store})
channel_specs: Mapping[str, BaseChannel] = {}
managed_specs: Mapping[str, ManagedValueSpec] = {}
for k, v in specs.items():
if skip_context and isinstance(v, Context):
channel_specs[k] = LastValue(None)
elif isinstance(v, BaseChannel):
channel_specs[k] = v
else:
managed_specs[k] = v
with ExitStack() as stack:
yield (
{
k: stack.enter_context(
v.from_checkpoint(checkpoint["channel_values"].get(k), config)
)
for k, v in channel_specs.items()
},
{
key: stack.enter_context(
value.cls.enter(config_for_managed, **value.kwargs)
if isinstance(value, ConfiguredManagedValue)
else value.enter(config_for_managed)
)
for key, value in managed_specs.items()
},
)
@asynccontextmanager
async def AsyncChannelsManager(
specs: Mapping[str, Union[BaseChannel, ManagedValueSpec]],
checkpoint: Checkpoint,
config: RunnableConfig,
store: Optional[BaseStore] = None,
*,
skip_context: bool = False,
) -> AsyncIterator[Mapping[str, BaseChannel]]:
"""Manage channels for the lifetime of a Pregel invocation (multiple steps)."""
config_for_managed = patch_config(config, configurable={CONFIG_KEY_STORE: store})
channel_specs: Mapping[str, BaseChannel] = {}
managed_specs: Mapping[str, ManagedValueSpec] = {}
for k, v in specs.items():
if skip_context and isinstance(v, Context):
channel_specs[k] = LastValue(None)
elif isinstance(v, BaseChannel):
channel_specs[k] = v
else:
managed_specs[k] = v
async with AsyncExitStack() as stack:
# managed: create enter tasks with reference to spec, await them
if tasks := {
asyncio.create_task(
stack.enter_async_context(
value.cls.aenter(config_for_managed, **value.kwargs)
if isinstance(value, ConfiguredManagedValue)
else value.aenter(config_for_managed)
)
): key
for key, value in managed_specs.items()
}:
done, _ = await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
else:
done = set()
yield (
# channels: enter each channel with checkpoint
{
k: await stack.enter_async_context(
v.afrom_checkpoint(checkpoint["channel_values"].get(k), config)
)
for k, v in channel_specs.items()
},
# managed: build mapping from spec to result
{tasks[task]: task.result() for task in done},
)
+1 -2
View File
@@ -15,7 +15,6 @@ from langchain_core.runnables.config import merge_configs
from langchain_core.runnables.utils import ConfigurableFieldSpec
from langgraph.constants import CONFIG_KEY_READ
from langgraph.managed.base import ManagedValueSpec
from langgraph.pregel.retry import RetryPolicy
from langgraph.pregel.write import ChannelWrite
from langgraph.utils import RunnableCallable
@@ -101,7 +100,7 @@ DEFAULT_BOUND: RunnablePassthrough = RunnablePassthrough()
class PregelNode(RunnableBindingBase):
channels: Union[list[str], Mapping[str, Union[str, ManagedValueSpec]]]
channels: Union[list[str], Mapping[str, str]]
triggers: list[str] = Field(default_factory=list)
+11 -2
View File
@@ -4,6 +4,7 @@ import random
import time
from typing import Optional
from langgraph.errors import GraphInterrupt
from langgraph.pregel.types import PregelExecutableTask, RetryPolicy
logger = logging.getLogger(__name__)
@@ -25,6 +26,9 @@ def run_with_retry(
task.proc.invoke(task.input, task.config)
# if successful, end
break
except GraphInterrupt:
# if interrupted, end
raise
except Exception as exc:
if retry_policy is None:
raise
@@ -49,7 +53,8 @@ def run_with_retry(
)
# log the retry
logger.info(
f"Retrying task {task.name} after {interval:.2f} seconds (attempt {attempts}) after {exc.__class__.__name__} {exc}"
f"Retrying task {task.name} after {interval:.2f} seconds (attempt {attempts}) after {exc.__class__.__name__} {exc}",
exc_info=exc,
)
@@ -74,6 +79,9 @@ async def arun_with_retry(
await task.proc.ainvoke(task.input, task.config)
# if successful, end
break
except GraphInterrupt:
# if interrupted, end
raise
except Exception as exc:
if retry_policy is None:
raise
@@ -98,5 +106,6 @@ async def arun_with_retry(
)
# log the retry
logger.info(
f"Retrying task {task.name} after {interval:.2f} seconds (attempt {attempts}) after {exc.__class__.__name__} {exc}"
f"Retrying task {task.name} after {interval:.2f} seconds (attempt {attempts}) after {exc.__class__.__name__} {exc}",
exc_info=exc,
)
+12 -4
View File
@@ -4,6 +4,7 @@ from typing import Any, Callable, Literal, NamedTuple, Optional, Type, Union
from langchain_core.runnables import Runnable, RunnableConfig
from langgraph.checkpoint.base import CheckpointMetadata
from langgraph.constants import Interrupt
def default_retry_on(exc: Exception) -> bool:
@@ -56,8 +57,11 @@ 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 PregelTaskDescription(NamedTuple):
class PregelTask(NamedTuple):
id: str
name: str
error: Optional[Exception] = None
interrupts: tuple[Interrupt, ...] = ()
class PregelExecutableTask(NamedTuple):
@@ -72,18 +76,22 @@ class PregelExecutableTask(NamedTuple):
class StateSnapshot(NamedTuple):
"""Snapshot of the state of the graph at the beginning of a step."""
values: Union[dict[str, Any], Any]
"""Current values of channels"""
next: tuple[str]
"""Nodes to execute in the next step, if any"""
next: tuple[str, ...]
"""The name of the node to execute in each task for this step."""
config: RunnableConfig
"""Config used to fetch this snapshot"""
metadata: Optional[CheckpointMetadata]
"""Metadata associated with this snapshot"""
created_at: Optional[str]
"""Timestamp of snapshot creation"""
parent_config: Optional[RunnableConfig] = None
parent_config: Optional[RunnableConfig]
"""Config used to fetch the parent snapshot, if any"""
tasks: tuple[PregelTask, ...]
"""Tasks to execute in this step. If already attempted, may contain an error."""
All = Literal["*"]
+21
View File
@@ -0,0 +1,21 @@
from typing import Any, List, Optional
V = dict[str, Any]
class BaseStore:
def list(self, prefixes: List[str]) -> dict[str, dict[str, V]]:
# list[namespace] -> dict[namespace, list[value]]
raise NotImplementedError
def put(self, writes: List[tuple[str, str, Optional[V]]]) -> None:
# list[(namespace, key, value | none)] -> None
raise NotImplementedError
async def alist(self, prefixes: List[str]) -> dict[str, dict[str, V]]:
# list[namespace] -> dict[namespace, list[value]]
raise NotImplementedError
async def aput(self, writes: List[tuple[str, str, Optional[V]]]) -> None:
# list[(namespace, key, value | none)] -> None
raise NotImplementedError
+65
View File
@@ -0,0 +1,65 @@
import asyncio
from typing import NamedTuple, Optional, Union
from langgraph.store.base import BaseStore, V
class ListOp(NamedTuple):
prefixes: list[str]
class PutOp(NamedTuple):
writes: list[tuple[str, str, Optional[V]]]
class AsyncBatchedStore(BaseStore):
def __init__(self, store: BaseStore) -> None:
self.store = store
self.aqueue: dict[asyncio.Future, Union[ListOp, PutOp]] = {}
self.task = asyncio.create_task(_run(self.aqueue, self.store))
def __del__(self) -> None:
self.task.cancel()
async def alist(self, prefixes: list[str]) -> dict[str, dict[str, V]]:
fut = asyncio.get_running_loop().create_future()
self.aqueue[fut] = ListOp(prefixes)
return await fut
async def aput(self, writes: list[tuple[str, str, Optional[V]]]) -> None:
fut = asyncio.get_running_loop().create_future()
self.aqueue[fut] = PutOp(writes)
return await fut
async def _run(
aqueue: dict[asyncio.Future, Union[ListOp, PutOp]], store: BaseStore
) -> None:
while True:
await asyncio.sleep(0)
if not aqueue:
continue
# this could use a lock, if we want thread safety
taken = aqueue.copy()
aqueue.clear()
# action each operation
lists = {f: o for f, o in taken.items() if isinstance(o, ListOp)}
if lists:
try:
results = await store.alist(
[p for op in lists.values() for p in op.prefixes]
)
for fut, op in lists.items():
fut.set_result({k: results.get(k) for k in op.prefixes})
except Exception as e:
for fut in lists:
fut.set_exception(e)
puts = {f: o for f, o in taken.items() if isinstance(o, PutOp)}
if puts:
try:
await store.aput([w for op in puts.values() for w in op.writes])
for fut in puts:
fut.set_result(None)
except Exception as e:
for fut in puts:
fut.set_exception(e)
+25
View File
@@ -0,0 +1,25 @@
from collections import defaultdict
from typing import List, Optional
from langgraph.store.base import BaseStore, V
class MemoryStore(BaseStore):
def __init__(self) -> None:
self.data: dict[str, dict[str, V]] = defaultdict(dict)
def list(self, prefixes: List[str]) -> dict[str, dict[str, V]]:
return {prefix: self.data[prefix] for prefix in prefixes}
async def alist(self, prefixes: List[str]) -> dict[str, dict[str, V]]:
return self.list(prefixes)
def put(self, writes: List[tuple[str, str, Optional[V]]]) -> None:
for namespace, key, value in writes:
if value is None:
self.data[namespace].pop(key, None)
else:
self.data[namespace][key] = value
async def aput(self, writes: List[tuple[str, str, Optional[V]]]) -> None:
return self.put(writes)
+2 -2
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph"
version = "0.2.3"
version = "0.2.8"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
license = "MIT"
@@ -65,7 +65,7 @@ omit = ["tests/*"]
[tool.pytest-watcher]
now = true
delay = 0.1
runner_args = ["--ff", "-v", "-n", "auto", "--dist", "worksteal", "--snapshot-update", "--tb", "short"]
runner_args = ["--ff", "-v", "-x", "-n", "auto", "--dist", "worksteal", "--snapshot-update", "--tb", "short"]
patterns = ["*.py"]
[build-system]
@@ -682,6 +682,24 @@
'''
# ---
# name: test_dynamic_interrupt
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
tool_two_slow(tool_two_slow)
tool_two_fast(tool_two_fast)
__end__([__end__]):::last
__start__ -.-> tool_two_slow;
tool_two_slow --> __end__;
__start__ -.-> tool_two_fast;
tool_two_fast --> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_in_one_fan_out_state_graph_waiting_edge
'''
graph TD;
@@ -1356,6 +1374,33 @@
# 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_xray_issue
'''
%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
__start__([__start__]):::first
p_one(p_one)
p_two___start__(__start__)
p_two_c_one(c_one)
p_two_c_two(c_two)
p_two___end__(__end__)
__end__([__end__]):::last
subgraph p_two
p_two___start__ --> p_two_c_one;
p_two_c_two --> p_two_c_one;
p_two_c_one -. &nbsp0&nbsp .-> p_two_c_two;
p_two_c_one -. &nbsp1&nbsp .-> p_two___end__;
end
__start__ --> p_one;
p_two___end__ --> p_one;
p_one -. &nbsp0&nbsp .-> p_two___start__;
p_one -. &nbsp1&nbsp .-> __end__;
classDef default fill:#f2f0ff,line-height:1.2
classDef first fill-opacity:0
classDef last fill:#bfb6fc
'''
# ---
# name: test_xray_lance
dict({
'edges': list([
+18
View File
@@ -4,3 +4,21 @@ class AnyStr(str):
def __eq__(self, other: object) -> bool:
return isinstance(other, str)
def __hash__(self) -> int:
return hash(str(self))
class ExceptionLike:
def __init__(self, exc: Exception) -> None:
self.exc = exc
def __eq__(self, value: object) -> bool:
return (
isinstance(value, Exception)
and self.exc.__class__ == value.__class__
and str(self.exc) == str(value)
)
def __hash__(self) -> int:
return hash((self.exc.__class__, str(self.exc)))
+2 -5
View File
@@ -1,7 +1,6 @@
from langgraph.channels.manager import ChannelsManager
from langgraph.checkpoint.base import empty_checkpoint
from langgraph.managed.base import ManagedValuesManager
from langgraph.pregel.algo import prepare_next_tasks
from langgraph.pregel.manager import ChannelsManager
def test_prepare_next_tasks() -> None:
@@ -9,9 +8,7 @@ def test_prepare_next_tasks() -> None:
processes = {}
checkpoint = empty_checkpoint()
with ManagedValuesManager({}, config) as managed, ChannelsManager(
{}, checkpoint, config
) as channels:
with ChannelsManager({}, checkpoint, config) as (channels, managed):
assert (
prepare_next_tasks(
checkpoint, processes, channels, managed, config, 0, for_execution=False
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+35
View File
@@ -0,0 +1,35 @@
import asyncio
from typing import Any, Optional
from pytest_mock import MockerFixture
from langgraph.store.base import BaseStore
from langgraph.store.batch import AsyncBatchedStore
async def test_async_batch_store(mocker: MockerFixture) -> None:
aget = mocker.stub()
alist = mocker.stub()
class MockStore(BaseStore):
async def aget(
self, pairs: list[tuple[str, str]]
) -> dict[tuple[str, str], Optional[dict[str, Any]]]:
aget(pairs)
return {pair: 1 for pair in pairs}
async def alist(self, prefixes: list[str]) -> dict[str, dict[str, Any]]:
alist(prefixes)
return {prefix: {prefix: 1} for prefix in prefixes}
store = AsyncBatchedStore(MockStore())
# concurrent calls are batched
results = await asyncio.gather(
store.alist(["a", "b"]),
store.alist(["c", "d"]),
)
assert results == [{"a": {"a": 1}, "b": {"b": 1}}, {"c": {"c": 1}, "d": {"d": 1}}]
assert [c.args for c in alist.call_args_list] == [
(["a", "b", "c", "d"],),
]
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@langchain/langgraph-sdk",
"version": "0.0.4",
"version": "0.0.5",
"description": "Client library for interacting with the LangGraph API",
"type": "module",
"packageManager": "yarn@1.22.19",
+3
View File
@@ -101,6 +101,9 @@ export interface ThreadState<ValuesType = DefaultValues> {
metadata: Metadata;
created_at: Optional<string>;
parent_checkpoint_id: Optional<string>;
config: Config;
parent_config?: Config;
}
export interface Run {
+30 -8
View File
@@ -40,8 +40,14 @@ from langgraph_sdk.schema import (
logger = logging.getLogger(__name__)
RESERVED_HEADERS = ("x-api-key",)
def get_client(
*, url: Optional[str] = None, api_key: Optional[str] = None
*,
url: Optional[str] = None,
api_key: Optional[str] = None,
headers: Optional[dict[str, str]] = None,
) -> LangGraphClient:
"""Get a LangGraphClient instance.
@@ -53,6 +59,7 @@ def get_client(
2. LANGGRAPH_API_KEY
3. LANGSMITH_API_KEY
4. LANGCHAIN_API_KEY
headers: Optional custom headers
"""
transport: Optional[httpx.AsyncBaseTransport] = None
if url is None:
@@ -65,17 +72,12 @@ def get_client(
url = "http://localhost:8123"
if transport is None:
transport = httpx.AsyncHTTPTransport(retries=5)
headers = {
"User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}",
}
api_key = _get_api_key(api_key)
if api_key:
headers["x-api-key"] = api_key
client = httpx.AsyncClient(
base_url=url,
transport=transport,
timeout=httpx.Timeout(connect=5, read=60, write=60, pool=5),
headers=headers,
headers=_get_headers(api_key, headers),
)
return LangGraphClient(client)
@@ -1695,3 +1697,23 @@ def _get_api_key(api_key: Optional[str] = None) -> Optional[str]:
if env := os.getenv(f"{prefix}_API_KEY"):
return env.strip().strip('"').strip("'")
return None # type: ignore
def _get_headers(
api_key: Optional[str], custom_headers: Optional[dict[str, str]]
) -> dict[str, str]:
"""Combine api_key and custom user-provided headers."""
custom_headers = custom_headers or {}
for header in RESERVED_HEADERS:
if header in custom_headers:
raise ValueError(f"Cannot set reserved header '{header}'")
headers = {
"User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}",
**custom_headers,
}
api_key = _get_api_key(api_key)
if api_key:
headers["x-api-key"] = api_key
return headers
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-sdk"
version = "0.1.27"
version = "0.1.28"
description = "SDK for interacting with LangGraph API"
authors = []
license = "MIT"
Generated
+165 -20
View File
@@ -152,6 +152,27 @@ webencodings = "*"
[package.extras]
css = ["tinycss2 (>=1.1.0,<1.3)"]
[[package]]
name = "cachecontrol"
version = "0.14.0"
description = "httplib2 caching for requests"
optional = false
python-versions = ">=3.7"
files = [
{file = "cachecontrol-0.14.0-py3-none-any.whl", hash = "sha256:f5bf3f0620c38db2e5122c0726bdebb0d16869de966ea6a2befe92470b740ea0"},
{file = "cachecontrol-0.14.0.tar.gz", hash = "sha256:7db1195b41c81f8274a7bbd97c956f44e8348265a1bc7641c37dfebc39f0c938"},
]
[package.dependencies]
filelock = {version = ">=3.8.0", optional = true, markers = "extra == \"filecache\""}
msgpack = ">=0.5.2,<2.0.0"
requests = ">=2.16.0"
[package.extras]
dev = ["CacheControl[filecache,redis]", "black", "build", "cherrypy", "furo", "mypy", "pytest", "pytest-cov", "sphinx", "sphinx-copybutton", "tox", "types-redis", "types-requests"]
filecache = ["filelock (>=3.8.0)"]
redis = ["redis (>=2.10.5)"]
[[package]]
name = "cairocffi"
version = "1.7.1"
@@ -533,6 +554,22 @@ files = [
[package.extras]
devel = ["colorama", "json-spec", "jsonschema", "pylint", "pytest", "pytest-benchmark", "pytest-cache", "validictory"]
[[package]]
name = "filelock"
version = "3.15.4"
description = "A platform independent file lock."
optional = false
python-versions = ">=3.8"
files = [
{file = "filelock-3.15.4-py3-none-any.whl", hash = "sha256:6ca1fffae96225dab4c6eaf1c4f4f28cd2568d3ec2a44e15a08520504de468e7"},
{file = "filelock-3.15.4.tar.gz", hash = "sha256:2207938cbc1844345cb01a5a95524dae30f0ce089eba5b00378295a17e3e90cb"},
]
[package.extras]
docs = ["furo (>=2023.9.10)", "sphinx (>=7.2.6)", "sphinx-autodoc-typehints (>=1.25.2)"]
testing = ["covdefaults (>=2.3)", "coverage (>=7.3.2)", "diff-cover (>=8.0.1)", "pytest (>=7.4.3)", "pytest-asyncio (>=0.21)", "pytest-cov (>=4.1)", "pytest-mock (>=3.12)", "pytest-timeout (>=2.2)", "virtualenv (>=20.26.2)"]
typing = ["typing-extensions (>=4.8)"]
[[package]]
name = "ghp-import"
version = "2.1.0"
@@ -684,6 +721,25 @@ files = [
{file = "idna-3.7.tar.gz", hash = "sha256:028ff3aadf0609c1fd278d8ea3089299412a7a8b9bd005dd08b9f8285bcb5cfc"},
]
[[package]]
name = "importlib-metadata"
version = "8.2.0"
description = "Read metadata from Python packages"
optional = false
python-versions = ">=3.8"
files = [
{file = "importlib_metadata-8.2.0-py3-none-any.whl", hash = "sha256:11901fa0c2f97919b288679932bb64febaeacf289d18ac84dd68cb2e74213369"},
{file = "importlib_metadata-8.2.0.tar.gz", hash = "sha256:72e8d4399996132204f9a16dcc751af254a48f8d1b20b9ff0f98d4a8f901e73d"},
]
[package.dependencies]
zipp = ">=0.5"
[package.extras]
doc = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-lint"]
perf = ["ipython"]
test = ["flufl.flake8", "importlib-resources (>=1.3)", "jaraco.test (>=5.4)", "packaging", "pyfakefs", "pytest (>=6,!=8.1.*)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-mypy", "pytest-perf (>=0.9.2)", "pytest-ruff (>=0.2.1)"]
[[package]]
name = "ipykernel"
version = "6.29.4"
@@ -719,13 +775,13 @@ test = ["flaky", "ipyparallel", "pre-commit", "pytest (>=7.0)", "pytest-asyncio
[[package]]
name = "ipython"
version = "8.25.0"
version = "8.18.1"
description = "IPython: Productive Interactive Computing"
optional = false
python-versions = ">=3.10"
python-versions = ">=3.9"
files = [
{file = "ipython-8.25.0-py3-none-any.whl", hash = "sha256:53eee7ad44df903a06655871cbab66d156a051fd86f3ec6750470ac9604ac1ab"},
{file = "ipython-8.25.0.tar.gz", hash = "sha256:c6ed726a140b6e725b911528f80439c534fac915246af3efc39440a6b0f9d716"},
{file = "ipython-8.18.1-py3-none-any.whl", hash = "sha256:e8267419d72d81955ec1177f8a29aaa90ac80ad647499201119e2f05e99aa397"},
{file = "ipython-8.18.1.tar.gz", hash = "sha256:ca6f079bb33457c66e233e4580ebfc4128855b4cf6370dddd73842a9563e8a27"},
]
[package.dependencies]
@@ -734,26 +790,25 @@ decorator = "*"
exceptiongroup = {version = "*", markers = "python_version < \"3.11\""}
jedi = ">=0.16"
matplotlib-inline = "*"
pexpect = {version = ">4.3", markers = "sys_platform != \"win32\" and sys_platform != \"emscripten\""}
pexpect = {version = ">4.3", markers = "sys_platform != \"win32\""}
prompt-toolkit = ">=3.0.41,<3.1.0"
pygments = ">=2.4.0"
stack-data = "*"
traitlets = ">=5.13.0"
typing-extensions = {version = ">=4.6", markers = "python_version < \"3.12\""}
traitlets = ">=5"
typing-extensions = {version = "*", markers = "python_version < \"3.10\""}
[package.extras]
all = ["ipython[black,doc,kernel,matplotlib,nbconvert,nbformat,notebook,parallel,qtconsole]", "ipython[test,test-extra]"]
all = ["black", "curio", "docrepr", "exceptiongroup", "ipykernel", "ipyparallel", "ipywidgets", "matplotlib", "matplotlib (!=3.2.0)", "nbconvert", "nbformat", "notebook", "numpy (>=1.22)", "pandas", "pickleshare", "pytest (<7)", "pytest (<7.1)", "pytest-asyncio (<0.22)", "qtconsole", "setuptools (>=18.5)", "sphinx (>=1.3)", "sphinx-rtd-theme", "stack-data", "testpath", "trio", "typing-extensions"]
black = ["black"]
doc = ["docrepr", "exceptiongroup", "intersphinx-registry", "ipykernel", "ipython[test]", "matplotlib", "setuptools (>=18.5)", "sphinx (>=1.3)", "sphinx-rtd-theme", "sphinxcontrib-jquery", "tomli", "typing-extensions"]
doc = ["docrepr", "exceptiongroup", "ipykernel", "matplotlib", "pickleshare", "pytest (<7)", "pytest (<7.1)", "pytest-asyncio (<0.22)", "setuptools (>=18.5)", "sphinx (>=1.3)", "sphinx-rtd-theme", "stack-data", "testpath", "typing-extensions"]
kernel = ["ipykernel"]
matplotlib = ["matplotlib"]
nbconvert = ["nbconvert"]
nbformat = ["nbformat"]
notebook = ["ipywidgets", "notebook"]
parallel = ["ipyparallel"]
qtconsole = ["qtconsole"]
test = ["pickleshare", "pytest", "pytest-asyncio (<0.22)", "testpath"]
test-extra = ["curio", "ipython[test]", "matplotlib (!=3.2.0)", "nbformat", "numpy (>=1.23)", "pandas", "trio"]
test = ["pickleshare", "pytest (<7.1)", "pytest-asyncio (<0.22)", "testpath"]
test-extra = ["curio", "matplotlib (!=3.2.0)", "nbformat", "numpy (>=1.22)", "pandas", "pickleshare", "pytest (<7.1)", "pytest-asyncio (<0.22)", "testpath", "trio"]
[[package]]
name = "jedi"
@@ -873,6 +928,7 @@ files = [
]
[package.dependencies]
importlib-metadata = {version = ">=4.8.3", markers = "python_version < \"3.10\""}
jupyter-core = ">=4.12,<5.0.dev0 || >=5.1.dev0"
python-dateutil = ">=2.8.2"
pyzmq = ">=23.0"
@@ -968,7 +1024,7 @@ typing-extensions = ">=4.7"
[[package]]
name = "langgraph"
version = "0.2.2"
version = "0.2.3"
description = "Building stateful, multi-actor applications with LLMs"
optional = false
python-versions = ">=3.9.0,<4.0"
@@ -1001,7 +1057,7 @@ url = "libs/checkpoint"
[[package]]
name = "langgraph-checkpoint-postgres"
version = "1.0.1"
version = "1.0.3"
description = "Library with a Postgres implementation of LangGraph checkpoint saver."
optional = false
python-versions = "^3.9.0,<4.0"
@@ -1079,6 +1135,9 @@ files = [
{file = "Markdown-3.6.tar.gz", hash = "sha256:ed4f41f6daecbeeb96e576ce414c41d2d876daa9a16cb35fa8ed8c2ddfad0224"},
]
[package.dependencies]
importlib-metadata = {version = ">=4.4", markers = "python_version < \"3.10\""}
[package.extras]
docs = ["mdx-gh-links (>=0.2)", "mkdocs (>=1.5)", "mkdocs-gen-files", "mkdocs-literate-nav", "mkdocs-nature (>=0.6)", "mkdocs-section-index", "mkdocstrings[python]"]
testing = ["coverage", "pyyaml"]
@@ -1288,6 +1347,7 @@ files = [
click = ">=7.0"
colorama = {version = ">=0.4", markers = "platform_system == \"Windows\""}
ghp-import = ">=1.0"
importlib-metadata = {version = ">=4.4", markers = "python_version < \"3.10\""}
jinja2 = ">=2.11.1"
markdown = ">=3.3.6"
markupsafe = ">=2.0.1"
@@ -1331,6 +1391,7 @@ files = [
]
[package.dependencies]
importlib-metadata = {version = ">=4.3", markers = "python_version < \"3.10\""}
mergedeep = ">=1.3.4"
platformdirs = ">=2.2.0"
pyyaml = ">=5.1"
@@ -1449,16 +1510,17 @@ test = ["autoflake", "black", "isort", "pytest"]
[[package]]
name = "mkdocs-rss-plugin"
version = "1.13.1"
version = "1.15.0"
description = "MkDocs plugin which generates a static RSS feed using git log and page.meta."
optional = false
python-versions = "<4,>=3.10"
python-versions = "<4,>=3.8"
files = [
{file = "mkdocs_rss_plugin-1.13.1-py2.py3-none-any.whl", hash = "sha256:665af194d0a48a1a88f3753c3cee78fee8e00b269dd61af1fe79f3555457b7a4"},
{file = "mkdocs_rss_plugin-1.13.1.tar.gz", hash = "sha256:1c12b1a7449dcff686b8fd991648a88bb8735e5013242b10ff4f375b90a93803"},
{file = "mkdocs_rss_plugin-1.15.0-py2.py3-none-any.whl", hash = "sha256:7308ac13f0976c0479db5a62cb7ef9b10fdd74b6521e459bb66a13e2cfe69d4b"},
{file = "mkdocs_rss_plugin-1.15.0.tar.gz", hash = "sha256:92995ed6c77b2ae1f5f2913e62282c27e50c35d618c4291b5b939e50badd7504"},
]
[package.dependencies]
cachecontrol = {version = ">=0.14,<1", extras = ["filecache"]}
GitPython = ">=3.1,<3.2"
mkdocs = ">=1.5,<2"
requests = ">=2.31,<3"
@@ -1482,6 +1544,7 @@ files = [
[package.dependencies]
click = ">=7.0"
importlib-metadata = {version = ">=4.6", markers = "python_version < \"3.10\""}
Jinja2 = ">=2.11.1"
Markdown = ">=3.3"
MarkupSafe = ">=1.1"
@@ -1489,6 +1552,7 @@ mkdocs = ">=1.4"
mkdocs-autorefs = ">=0.3.1"
platformdirs = ">=2.2.0"
pymdown-extensions = ">=6.3"
typing-extensions = {version = ">=4.1", markers = "python_version < \"3.10\""}
[package.extras]
crystal = ["mkdocstrings-crystal (>=0.3.4)"]
@@ -1510,6 +1574,71 @@ files = [
griffe = ">=0.47"
mkdocstrings = ">=0.25"
[[package]]
name = "msgpack"
version = "1.0.8"
description = "MessagePack serializer"
optional = false
python-versions = ">=3.8"
files = [
{file = "msgpack-1.0.8-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:505fe3d03856ac7d215dbe005414bc28505d26f0c128906037e66d98c4e95868"},
{file = "msgpack-1.0.8-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:e6b7842518a63a9f17107eb176320960ec095a8ee3b4420b5f688e24bf50c53c"},
{file = "msgpack-1.0.8-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:376081f471a2ef24828b83a641a02c575d6103a3ad7fd7dade5486cad10ea659"},
{file = "msgpack-1.0.8-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5e390971d082dba073c05dbd56322427d3280b7cc8b53484c9377adfbae67dc2"},
{file = "msgpack-1.0.8-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:00e073efcba9ea99db5acef3959efa45b52bc67b61b00823d2a1a6944bf45982"},
{file = "msgpack-1.0.8-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:82d92c773fbc6942a7a8b520d22c11cfc8fd83bba86116bfcf962c2f5c2ecdaa"},
{file = "msgpack-1.0.8-cp310-cp310-musllinux_1_1_aarch64.whl", hash = "sha256:9ee32dcb8e531adae1f1ca568822e9b3a738369b3b686d1477cbc643c4a9c128"},
{file = "msgpack-1.0.8-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:e3aa7e51d738e0ec0afbed661261513b38b3014754c9459508399baf14ae0c9d"},
{file = "msgpack-1.0.8-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:69284049d07fce531c17404fcba2bb1df472bc2dcdac642ae71a2d079d950653"},
{file = "msgpack-1.0.8-cp310-cp310-win32.whl", hash = "sha256:13577ec9e247f8741c84d06b9ece5f654920d8365a4b636ce0e44f15e07ec693"},
{file = "msgpack-1.0.8-cp310-cp310-win_amd64.whl", hash = "sha256:e532dbd6ddfe13946de050d7474e3f5fb6ec774fbb1a188aaf469b08cf04189a"},
{file = "msgpack-1.0.8-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:9517004e21664f2b5a5fd6333b0731b9cf0817403a941b393d89a2f1dc2bd836"},
{file = "msgpack-1.0.8-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:d16a786905034e7e34098634b184a7d81f91d4c3d246edc6bd7aefb2fd8ea6ad"},
{file = "msgpack-1.0.8-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:e2872993e209f7ed04d963e4b4fbae72d034844ec66bc4ca403329db2074377b"},
{file = "msgpack-1.0.8-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:5c330eace3dd100bdb54b5653b966de7f51c26ec4a7d4e87132d9b4f738220ba"},
{file = "msgpack-1.0.8-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:83b5c044f3eff2a6534768ccfd50425939e7a8b5cf9a7261c385de1e20dcfc85"},
{file = "msgpack-1.0.8-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1876b0b653a808fcd50123b953af170c535027bf1d053b59790eebb0aeb38950"},
{file = "msgpack-1.0.8-cp311-cp311-musllinux_1_1_aarch64.whl", hash = "sha256:dfe1f0f0ed5785c187144c46a292b8c34c1295c01da12e10ccddfc16def4448a"},
{file = "msgpack-1.0.8-cp311-cp311-musllinux_1_1_i686.whl", hash = "sha256:3528807cbbb7f315bb81959d5961855e7ba52aa60a3097151cb21956fbc7502b"},
{file = "msgpack-1.0.8-cp311-cp311-musllinux_1_1_x86_64.whl", hash = "sha256:e2f879ab92ce502a1e65fce390eab619774dda6a6ff719718069ac94084098ce"},
{file = "msgpack-1.0.8-cp311-cp311-win32.whl", hash = "sha256:26ee97a8261e6e35885c2ecd2fd4a6d38252246f94a2aec23665a4e66d066305"},
{file = "msgpack-1.0.8-cp311-cp311-win_amd64.whl", hash = "sha256:eadb9f826c138e6cf3c49d6f8de88225a3c0ab181a9b4ba792e006e5292d150e"},
{file = "msgpack-1.0.8-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:114be227f5213ef8b215c22dde19532f5da9652e56e8ce969bf0a26d7c419fee"},
{file = "msgpack-1.0.8-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:d661dc4785affa9d0edfdd1e59ec056a58b3dbb9f196fa43587f3ddac654ac7b"},
{file = "msgpack-1.0.8-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:d56fd9f1f1cdc8227d7b7918f55091349741904d9520c65f0139a9755952c9e8"},
{file = "msgpack-1.0.8-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:0726c282d188e204281ebd8de31724b7d749adebc086873a59efb8cf7ae27df3"},
{file = "msgpack-1.0.8-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:8db8e423192303ed77cff4dce3a4b88dbfaf43979d280181558af5e2c3c71afc"},
{file = "msgpack-1.0.8-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:99881222f4a8c2f641f25703963a5cefb076adffd959e0558dc9f803a52d6a58"},
{file = "msgpack-1.0.8-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:b5505774ea2a73a86ea176e8a9a4a7c8bf5d521050f0f6f8426afe798689243f"},
{file = "msgpack-1.0.8-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:ef254a06bcea461e65ff0373d8a0dd1ed3aa004af48839f002a0c994a6f72d04"},
{file = "msgpack-1.0.8-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:e1dd7839443592d00e96db831eddb4111a2a81a46b028f0facd60a09ebbdd543"},
{file = "msgpack-1.0.8-cp312-cp312-win32.whl", hash = "sha256:64d0fcd436c5683fdd7c907eeae5e2cbb5eb872fafbc03a43609d7941840995c"},
{file = "msgpack-1.0.8-cp312-cp312-win_amd64.whl", hash = "sha256:74398a4cf19de42e1498368c36eed45d9528f5fd0155241e82c4082b7e16cffd"},
{file = "msgpack-1.0.8-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:0ceea77719d45c839fd73abcb190b8390412a890df2f83fb8cf49b2a4b5c2f40"},
{file = "msgpack-1.0.8-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:1ab0bbcd4d1f7b6991ee7c753655b481c50084294218de69365f8f1970d4c151"},
{file = "msgpack-1.0.8-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:1cce488457370ffd1f953846f82323cb6b2ad2190987cd4d70b2713e17268d24"},
{file = "msgpack-1.0.8-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:3923a1778f7e5ef31865893fdca12a8d7dc03a44b33e2a5f3295416314c09f5d"},
{file = "msgpack-1.0.8-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a22e47578b30a3e199ab067a4d43d790249b3c0587d9a771921f86250c8435db"},
{file = "msgpack-1.0.8-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:bd739c9251d01e0279ce729e37b39d49a08c0420d3fee7f2a4968c0576678f77"},
{file = "msgpack-1.0.8-cp38-cp38-musllinux_1_1_aarch64.whl", hash = "sha256:d3420522057ebab1728b21ad473aa950026d07cb09da41103f8e597dfbfaeb13"},
{file = "msgpack-1.0.8-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:5845fdf5e5d5b78a49b826fcdc0eb2e2aa7191980e3d2cfd2a30303a74f212e2"},
{file = "msgpack-1.0.8-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:6a0e76621f6e1f908ae52860bdcb58e1ca85231a9b0545e64509c931dd34275a"},
{file = "msgpack-1.0.8-cp38-cp38-win32.whl", hash = "sha256:374a8e88ddab84b9ada695d255679fb99c53513c0a51778796fcf0944d6c789c"},
{file = "msgpack-1.0.8-cp38-cp38-win_amd64.whl", hash = "sha256:f3709997b228685fe53e8c433e2df9f0cdb5f4542bd5114ed17ac3c0129b0480"},
{file = "msgpack-1.0.8-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:f51bab98d52739c50c56658cc303f190785f9a2cd97b823357e7aeae54c8f68a"},
{file = "msgpack-1.0.8-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:73ee792784d48aa338bba28063e19a27e8d989344f34aad14ea6e1b9bd83f596"},
{file = "msgpack-1.0.8-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:f9904e24646570539a8950400602d66d2b2c492b9010ea7e965025cb71d0c86d"},
{file = "msgpack-1.0.8-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e75753aeda0ddc4c28dce4c32ba2f6ec30b1b02f6c0b14e547841ba5b24f753f"},
{file = "msgpack-1.0.8-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5dbf059fb4b7c240c873c1245ee112505be27497e90f7c6591261c7d3c3a8228"},
{file = "msgpack-1.0.8-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:4916727e31c28be8beaf11cf117d6f6f188dcc36daae4e851fee88646f5b6b18"},
{file = "msgpack-1.0.8-cp39-cp39-musllinux_1_1_aarch64.whl", hash = "sha256:7938111ed1358f536daf311be244f34df7bf3cdedb3ed883787aca97778b28d8"},
{file = "msgpack-1.0.8-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:493c5c5e44b06d6c9268ce21b302c9ca055c1fd3484c25ba41d34476c76ee746"},
{file = "msgpack-1.0.8-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:5fbb160554e319f7b22ecf530a80a3ff496d38e8e07ae763b9e82fadfe96f273"},
{file = "msgpack-1.0.8-cp39-cp39-win32.whl", hash = "sha256:f9af38a89b6a5c04b7d18c492c8ccf2aee7048aff1ce8437c4683bb5a1df893d"},
{file = "msgpack-1.0.8-cp39-cp39-win_amd64.whl", hash = "sha256:ed59dd52075f8fc91da6053b12e8c89e37aa043f8986efd89e61fae69dc1b011"},
{file = "msgpack-1.0.8.tar.gz", hash = "sha256:95c02b0e27e706e48d0e5426d1710ca78e0f0628d6e89d5b5a5b91a5f12274f3"},
]
[[package]]
name = "nbclient"
version = "0.10.0"
@@ -1547,6 +1676,7 @@ files = [
beautifulsoup4 = "*"
bleach = "!=5.0.0"
defusedxml = "*"
importlib-metadata = {version = ">=3.6", markers = "python_version < \"3.10\""}
jinja2 = ">=3.0"
jupyter-core = ">=4.7"
jupyterlab-pygments = "*"
@@ -2826,7 +2956,22 @@ files = [
{file = "webencodings-0.5.1.tar.gz", hash = "sha256:b36a1c245f2d304965eb4e0a82848379241dc04b865afcc4aab16748587e1923"},
]
[[package]]
name = "zipp"
version = "3.20.0"
description = "Backport of pathlib-compatible object wrapper for zip files"
optional = false
python-versions = ">=3.8"
files = [
{file = "zipp-3.20.0-py3-none-any.whl", hash = "sha256:58da6168be89f0be59beb194da1250516fdaa062ccebd30127ac65d30045e10d"},
{file = "zipp-3.20.0.tar.gz", hash = "sha256:0145e43d89664cfe1a2e533adc75adafed82fe2da404b4bbb6b026c0157bdb31"},
]
[package.extras]
doc = ["furo", "jaraco.packaging (>=9.3)", "jaraco.tidelift (>=1.4)", "rst.linker (>=1.9)", "sphinx (>=3.5)", "sphinx-lint"]
test = ["big-O", "importlib-resources", "jaraco.functools", "jaraco.itertools", "jaraco.test", "more-itertools", "pytest (>=6,!=8.1.*)", "pytest-checkdocs (>=2.4)", "pytest-cov", "pytest-enabler (>=2.2)", "pytest-ignore-flaky", "pytest-mypy", "pytest-ruff (>=0.2.1)"]
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "8bbefefef5e786ed86783a98ff0809cfaaad26d59b2d83623c269587a3bf5f7a"
python-versions = "^3.9"
content-hash = "d98361cff93b5d06ac80fd076e73bcf3b0883fb3ad8e66bc3ca0cfc6b955c94c"
+1 -1
View File
@@ -7,7 +7,7 @@ license = "MIT"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.10"
python = "^3.9"
[tool.poetry.group.docs.dependencies]
langgraph = { path = "libs/langgraph/", develop = true }