Compare commits

..
Author SHA1 Message Date
William Fu-Hinthorn 99a5cb875d phony 2025-05-20 15:59:26 -07:00
William Fu-Hinthorn 78239caed0 Some fancy stuff 2025-05-20 15:59:26 -07:00
Lauren Hirata SinghandGitHub 0a8f54ca4a docs: add redirect (#4762) 2025-05-20 15:06:34 -04:00
Lauren Hirata Singh 9d41d439cc docs: add redirect 2025-05-20 14:52:02 -04:00
Leeroy BrunandGitHub ae5c01fbab docs: typo in npx command to install LangGraph CLI (#4761) 2025-05-20 17:08:02 +00:00
Sydney RunkleandGitHub bd52b1faa1 docs: deferred nodes (#4759) 2025-05-20 13:07:11 -04:00
Stefano LottiniandGitHub 4a04ca7268 [docs] Revise pip-install packages for PostgresSaver in "how-to/Persistence" notebook (#4752)
revise pip-install packages for PostgresSaver usage in how-to page
2025-05-20 15:46:39 +00:00
Yazan JianandGitHub 23b868da00 docs: fix async usage example from await .invoke() to `await .ainvo… (#4758)
docs: fix async usage example from `await .invoke()` to `await .ainvoke()`
2025-05-20 15:45:47 +00:00
David DuongandGitHub 51bfd16460 feat(sdk-js): add stream_resumable flag, that marks the stream as resumable (#4757) 2025-05-20 14:16:39 +02:00
Tat Dat Duong 325d9e3134 Bump to 0.0.77 2025-05-20 14:14:48 +02:00
Tat Dat Duong b052ebf984 feat(sdk-js): add stream_resumable flag, that marks the stream as resumable 2025-05-20 14:13:47 +02:00
Sydney RunkleandGitHub 1e938a692f docs: node caching (#4749) 2025-05-19 15:49:31 -04:00
Nuno CamposandGitHub 46a9d3159d Remove local_write utility (#4751)
- The validation isn't worth the cost of having to pass list of nodes to task config
2025-05-19 15:49:17 -04:00
Nuno CamposandGitHub d825e39df9 Print output for cached @task functions (#4750) 2025-05-19 11:42:16 -07:00
Nuno Campos 53a1e7c9de Print output for cached @task functions 2025-05-19 11:35:51 -07:00
Sydney RunkleandGitHub 7bd8616b1e feature: Implement post_model_hook and HumanInterruptNode (#4583) 2025-05-19 14:07:42 -04:00
William FHandGitHub 95f92069a7 sqlite: Add test for search with list filters (#4747) 2025-05-18 23:50:27 -07:00
William FHandGitHub 6b28319796 sqlite: update list_namespaces with max_depth (#4746)
sqlite: update on conflict
2025-05-18 23:27:06 -07:00
Didier DurandandGitHub 2ddf61201c docs: Fixing some typos (#4741) 2025-05-18 22:36:37 -07:00
Didier DurandandGitHub 9453ee08dc docs: Fix a few spelling mistakes (#4742)
Fixing some typos
2025-05-18 22:35:54 -07:00
William Fu-Hinthorn 15bafc54c8 Update lockfile 2025-05-17 21:55:46 -07:00
William FHandGitHub 725dd40fa7 Release sqlite store (#4737) 2025-05-17 21:49:03 -07:00
William FHandGitHub c2a1b3af07 docs: Clean up adjective use in readme (#4734) 2025-05-17 21:43:14 -07:00
William FHandGitHub 025b634d98 SqliteStore (#3608) 2025-05-17 21:40:19 -07:00
Brace SproulandGitHub 8edb3e7b65 release(sdk-js): 0.0.76 (#4733) 2025-05-16 14:59:50 -07:00
bracesproul c21cf9fc1d release(sdk-js): 0.0.76 2025-05-16 14:56:56 -07:00
David DuongandGitHub cb95393c67 fix: Allow users to config whether or not stream subgraphs (#4732) 2025-05-16 23:51:25 +02:00
bracesproul bb1edb4415 cr 2025-05-16 14:49:57 -07:00
bracesproul 6f1db4c60a fix: Allow users to config whether or not stream subgraphs 2025-05-16 14:49:28 -07:00
Vadym BardaandGitHub 119a03bb00 ci: fix benchmark command (#4729) 2025-05-16 12:53:46 -07:00
Vadym BardaandGitHub 09138048bc docs: document tuples in streaming guides (#4690) 2025-05-16 14:59:51 -04:00
David DuongandGitHub adc89440a6 feat(sdk-js): expose ID for SSE events, update joinStream (#4547) 2025-05-16 20:53:39 +02:00
Tat Dat Duong c65919b3b2 Remove onResponse 2025-05-16 11:52:07 -07:00
Tat Dat Duong 0fa2b6c600 Bump to 0.0.75 2025-05-16 11:50:35 -07:00
Tat Dat Duong 9b2071b103 Fix tests 2025-05-16 11:50:09 -07:00
Tat Dat Duong b3f13ee904 Add onRunCreated callback 2025-05-16 11:43:40 -07:00
Tat Dat Duong bf239a06e1 Add callback for response object to get headers 2025-05-16 11:43:39 -07:00
Tat Dat Duong cc25539018 feat(sdk-js): expose ID for SSE events, update joinStream 2025-05-16 11:43:39 -07:00
Lauren Hirata SinghandGitHub 654096625a docs: add redirect (#4727) 2025-05-16 11:28:45 -07:00
David DuongandGitHub 3ea1141d55 feat(sdk-js): switch from jest to vitest, add useStream FE tests (#4726) 2025-05-16 19:55:30 +02:00
Tat Dat Duong e9d1f5508a package.json 2025-05-16 10:53:06 -07:00
Tat Dat Duong c6157d90dd feat(sdk-js): switch from jest to vitest, add useStream FE tests 2025-05-16 10:48:11 -07:00
Lauren Hirata Singh f37a228b58 add redirect 2025-05-16 10:28:51 -07:00
Sydney RunkleandGitHub d34299ac39 ci: fix issue with cache in integration tests (#4725) 2025-05-16 16:49:29 +00:00
81 changed files with 9669 additions and 2147 deletions
+3
View File
@@ -27,6 +27,9 @@ jobs:
uses: astral-sh/setup-uv@v6
with:
python-version: ${{ matrix.python-version }}
enable-cache: true
cache-suffix: "cli-integration-test"
ignore-nothing-to-cache: true
- name: Setup env
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples
+1 -10
View File
@@ -28,7 +28,7 @@ jobs:
with:
python-version: ${{ matrix.python-version }}
enable-cache: true
cache-siffix: test-${{ inputs.working-directory }}
cache-suffix: test-${{ inputs.working-directory }}
- name: Login to Docker Hub
uses: docker/login-action@v3
if: ${{ !github.event.pull_request.head.repo.fork }}
@@ -45,15 +45,6 @@ jobs:
shell: bash
working-directory: ${{ inputs.working-directory }}
run: make test
- name: Install min version of deps
shell: bash
run: uv sync --frozen --all-extras --resolution lowest-direct --force-reinstall
- name: Run tests with min version of deps
shell: bash
run: make test
working-directory: ${{ inputs.working-directory }}
- name: Ensure the tests did not create any additional files
shell: bash
-9
View File
@@ -42,15 +42,6 @@ jobs:
shell: bash
run: make test_parallel
- name: Install min version of deps
shell: bash
run: uv sync --frozen --all-extras --resolution lowest-direct --force-reinstall
- name: Run tests with min version of deps
shell: bash
run: make test
working-directory: ${{ inputs.working-directory }}
- name: Ensure the tests did not create any additional files
shell: bash
run: |
@@ -39,15 +39,6 @@ jobs:
shell: bash
run: make test
- name: Install min version of deps
shell: bash
run: uv sync --frozen --all-extras --resolution lowest-direct --force-reinstall
- name: Run tests with min version of deps
shell: bash
run: make test
working-directory: ${{ inputs.working-directory }}
- name: Ensure the tests did not create any additional files
shell: bash
run: |
+1 -1
View File
@@ -153,7 +153,7 @@ Each category serves a distinct purpose and requires a specific approach to writ
Here are some other guidelines you should think about when writing and organizing documentation.
We generally do not merge new tutorials from outside contributors without an actue need.
We generally do not merge new tutorials from outside contributors without an actual need.
We welcome updates as well as new integration docs, how-tos, and references.
### Avoid duplication
+2 -2
View File
@@ -14,7 +14,7 @@
[![Docs](https://img.shields.io/badge/docs-latest-blue)](https://langchain-ai.github.io/langgraph/)
[![GitMCP](https://img.shields.io/endpoint?url=https://gitmcp.io/badge/langchain-ai/langgraph)](https://gitmcp.io/langchain-ai/langgraph)
Trusted by companies shaping the future of agents – including Klarna, Replit, Elastic, and more – LangGraph is a powerful low-level orchestration framework for building, managing, and deploying long-running, stateful agents.
Trusted by companies shaping the future of agents – including Klarna, Replit, Elastic, and more – LangGraph is a low-level orchestration framework for building, managing, and deploying long-running, stateful agents.
## Get started
@@ -77,7 +77,7 @@ While LangGraph can be used standalone, it also integrates seamlessly with any L
- [Examples](https://langchain-ai.github.io/langgraph/tutorials/): Guided examples on getting started with LangGraph.
- [LangChain Academy](https://academy.langchain.com/courses/intro-to-langgraph): Learn the basics of LangGraph in our free, structured course.
- [Templates](https://langchain-ai.github.io/langgraph/concepts/template_applications/): Pre-built reference apps for common agentic workflows (e.g. ReAct agent, memory, retrieval etc.) that can be cloned and adapted.
- [Case studies](https://www.langchain.com/built-with-langgraph): Hear how industry leaders use LangGraph to ship powerful, production-ready AI applications.
- [Case studies](https://www.langchain.com/built-with-langgraph): Hear how industry leaders use LangGraph to ship AI applications at scale.
## Acknowledgements
+2
View File
@@ -53,6 +53,7 @@ REDIRECT_MAP = {
"how-tos/persistence_redis.ipynb": "how-tos/persistence.ipynb#use-in-production",
"how-tos/subgraph-persistence.ipynb": "how-tos/persistence.ipynb#use-with-subgraphs",
"how-tos/cross-thread-persistence.ipynb": "how-tos/persistence.ipynb#add-long-term-memory",
"cloud/how-tos/copy_threads": "cloud/how-tos/use_threads",
# tool calling how-tos
"how-tos/tool-calling-errors.ipynb": "how-tos/tool-calling.ipynb#handle-errors",
"how-tos/pass-config-to-tools.ipynb": "how-tos/tool-calling.ipynb#access-config",
@@ -70,6 +71,7 @@ REDIRECT_MAP = {
"cloud/faq/studio.md": "concepts/langgraph_studio.md#studio-faqs",
"cloud/how-tos/human_in_the_loop_edit_state.md": "cloud/how-tos/add-human-in-the-loop.md",
"cloud/how-tos/human_in_the_loop_user_input.md": "cloud/how-tos/add-human-in-the-loop.md",
"concepts/platform_architecture.md": "langgraph/concepts/langgraph_cloud#architecture",
# cloud streaming redirects
"cloud/how-tos/stream_values.md": "cloud/how-tos/streaming.md#stream-graph-state",
"cloud/how-tos/stream_updates.md": "cloud/how-tos/streaming.md#stream-graph-state",
+2 -2
View File
@@ -10,7 +10,7 @@ hide:
# Running agents
Agents support both synchronous and asynchronous execution using either `.invoke()` / `await .invoke()` for full responses, or `.stream()` / `.astream()` for **incremental** [streaming](streaming.md) output. This section explains how to provide input, interpret output, enable streaming, and control execution limits.
Agents support both synchronous and asynchronous execution using either `.invoke()` / `await .ainvoke()` for full responses, or `.stream()` / `.astream()` for **incremental** [streaming](streaming.md) output. This section explains how to provide input, interpret output, enable streaming, and control execution limits.
## Basic usage
@@ -18,7 +18,7 @@ Agents support both synchronous and asynchronous execution using either `.invoke
Agents can be executed in two primary modes:
- **Synchronous** using `.invoke()` or `.stream()`
- **Asynchronous** using `await .invoke()` or `async for` with `.astream()`
- **Asynchronous** using `await .ainvoke()` or `async for` with `.astream()`
=== "Sync invocation"
```python
+1 -1
View File
@@ -2,6 +2,6 @@
Webhooks enable event-driven communication from your LangGraph Platform application to external services. For example, you may want to issue an update to a separate service once an API call to LangGraph Platform has finished running.
Many LangGraph Platform endpoints accept a `webhook` parameter. If this parameter is specified by a an endpoint that can accept POST requests, LangGraph Platform will send a request at the completion of a run.
Many LangGraph Platform endpoints accept a `webhook` parameter. If this parameter is specified by an endpoint that can accept POST requests, LangGraph Platform will send a request at the completion of a run.
See the corresponding [how-to guide](../../cloud/how-tos/webhooks.md) for more detail.
@@ -18,7 +18,7 @@ Before deploying, review the [conceptual guide for the Self-Hosted Data Plane](.
helm repo add kedacore https://kedacore.github.io/charts
helm install keda kedacore/keda --namespace keda --create-namespace
1. A valid `Ingress` controller is install on your cluster.
1. A valid `Ingress` controller is installed on your cluster.
1. You have slack space in your cluster for multiple deployments. `Cluster-Autoscaler` is recommended to automatically provision new nodes.
### Setup
@@ -231,7 +231,7 @@ Inside your deployment, select the "Assistants" tab. For the assistant you would
To edit the assistant, use the `update` method. This will create a new version of the assistant with the provided edits. See the [Python](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/python_sdk_ref/#langgraph_sdk.client.AssistantsClient.update) and [JS](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/js_ts_sdk_ref/#update) SDK reference docs for more information.
!!! note "Note"
You must pass in the ENTIRE config (and metadata if you are using it). The update endpoint creates new versions completely from scratch and does not rely on previously versions.
You must pass in the ENTIRE config (and metadata if you are using it). The update endpoint creates new versions completely from scratch and does not rely on previous versions.
For example, to update your assistant's system prompt:
=== "Python"
@@ -321,7 +321,7 @@ If you now run your graph and pass in this assistant id, it will use the first v
### LangGraph Platform UI
If using LangGraph Studio, to set the active version of your asssistant, click the "Manage Assistants" button and locate the assistant you would like to use. Select the assistant and the version, and then click the "Active" toggle. This will update the assistant to make the selected version active.
If using LangGraph Studio, to set the active version of your assistant, click the "Manage Assistants" button and locate the assistant you would like to use. Select the assistant and the version, and then click the "Active" toggle. This will update the assistant to make the selected version active.
!!! warning "Deleting Assistants"
Deleting as assistant will delete ALL of it's versions. There is currently no way to delete a single version, but by pointing your assistant to the correct version you can skip any versions that you don't wish to use.
Deleting as assistant will delete ALL of its versions. There is currently no way to delete a single version, but by pointing your assistant to the correct version you can skip any versions that you don't wish to use.
+1 -1
View File
@@ -4,7 +4,7 @@ Sometimes you don't want to run your graph based on user interaction, but rather
## Setup
First, let's setup our SDK client, assistant, and thread:
First, let's set up our SDK client, assistant, and thread:
=== "Python"
+1 -1
View File
@@ -1,6 +1,6 @@
# Add node to dataset
This guide shows how to add examples to [LangSmith datasets](https://docs.smith.langchain.com/evaluation/how_to_guides#dataset-management) from nodes in the thread log. This is useful to evaluate indivudal steps of the agent.
This guide shows how to add examples to [LangSmith datasets](https://docs.smith.langchain.com/evaluation/how_to_guides#dataset-management) from nodes in the thread log. This is useful to evaluate individual steps of the agent.
1. Select a thread.
2. Click on the `Add to Dataset` button.
@@ -335,7 +335,7 @@ const { thread, submit } = useStream({
});
```
Then you can pushing updates to the UI component by calling `ui.push()` / `push_ui_message()` with the same ID as the UI message you wish to update.
Then you can push updates to the UI component by calling `ui.push()` / `push_ui_message()` with the same ID as the UI message you wish to update.
=== "Python"
+1 -1
View File
@@ -488,4 +488,4 @@ You can also view threads in a deployment via the LangGraph Platform UI.
Inside your deployment, select the "Threads" tab. This will load a table of all of the threads in your deployment.
Select a thread to inspect its current state. To view it's full history and for further debugging, open the thread in [LangGraph Studio](../../concepts//langgraph_studio.md).
Select a thread to inspect its current state. To view its full history and for further debugging, open the thread in [LangGraph Studio](../../concepts//langgraph_studio.md).
+48
View File
@@ -247,6 +247,54 @@ from langgraph.graph import END
graph.add_edge("node_a", END)
```
### Node Caching
LangGraph supports caching of tasks/nodes based on the input to the node. To use caching:
* Specify a cache when compiling a graph (or specifying an entrypoint)
* Specify a cache policy for nodes. Each cache policy supports:
* `key_func` used to generate a cache key based on the input to a node, which defaults to a `hash` of the input with pickle.
* `ttl`, the time to live for the cache in seconds. If not specified, the cache will never expire.
For example:
```py
import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy
class State(TypedDict):
x: int
result: int
builder = StateGraph(State)
def expensive_node(state: State) -> dict[str, int]:
# expensive computation
time.sleep(2)
return {"result": state["x"] * 2}
builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))
builder.set_entry_point("expensive_node")
builder.set_finish_point("expensive_node")
graph = builder.compile(cache=InMemoryCache())
print(graph.invoke({"x": 5}, stream_mode='updates')) # (1)!
[{'expensive_node': {'result': 10}}]
print(graph.invoke({"x": 5}, stream_mode='updates')) # (2)!
[{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]
```
1. First run takes the full second to run (due to mocked expensive computation).
2. Second run utilizes cache and returns quickly.
## Edges
Edges define how the logic is routed and how the graph decides to stop. This is a big part of how your agents work and how different nodes communicate with each other. There are a few key types of edges:
+44 -24
View File
@@ -1288,12 +1288,43 @@
},
{
"cell_type": "markdown",
"id": "4eeb895c-adca-40ab-b289-93ee56e18661",
"id": "068f806a",
"metadata": {},
"source": [
"</details>"
]
},
{
"cell_type": "markdown",
"id": "6d99d63c",
"metadata": {},
"source": [
"## Add node caching\n",
"\n",
"Node caching is useful in cases where you want to avoid repeating operations, like when doing something expensive (either in terms of time or cost). LangGraph lets you add individualized caching policies to nodes in a graph.\n",
"\n",
"To configure a cache policy, pass the `cache_policy` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node) function. In the following example, a [`CachePolicy`](https://langchain-ai.github.io/langgraph/reference/types/?h=cachepolicy#langgraph.types.CachePolicy) object is instantiated with a time to live of 120 seconds and the default `key_func` generator. Then it is associated with a node:\n",
"\n",
"```python\n",
"from langgraph.types import CachePolicy\n",
"\n",
"builder.add_node(\n",
" \"node_name\",\n",
" node_function,\n",
" cache_policy=CachePolicy(ttl=120),\n",
")\n",
"```\n",
"\n",
"Then, to enable node-level caching for a graph, set the `cache` argument when compiling the graph. The example below uses `InMemoryCache` to set up a graph with in-memory cache, but `SqliteCache` is also available.\n",
"\n",
"```python\n",
"from langgraph.cache.memory import InMemoryCache\n",
"\n",
"\n",
"graph = builder.compile(cache=InMemoryCache())\n",
"```"
]
},
{
"cell_type": "markdown",
"id": "e1a0213e-282f-4fad-b048-5f7465edfccb",
@@ -1754,18 +1785,19 @@
},
{
"cell_type": "markdown",
"id": "205ff836-0f97-4ee8-9830-6bd8368e48c9",
"id": "48731230",
"metadata": {},
"source": [
"<details class=\"example\"><summary>Extended example: unequal length branches</summary>\n",
"### Defer node execution\n",
"\n",
"The above example showed how to fan-out and fan-in when each path was only one step. But what if one path had more than one step? Let's add a node <code>b_2</code> in the \"b\" branch:\n",
"<br>"
"Deferring node execution is useful when you want to delay the execution of a node until all other pending tasks are completed. This is particularly relevant when branches have different lengths, which is common in workflows like map-reduce flows.\n",
"\n",
"The above example showed how to fan-out and fan-in when each path was only one step. But what if one branch had more than one step? Let's add a node `\"b_2\"` in the `\"b\"` branch:"
]
},
{
"cell_type": "code",
"execution_count": 1,
"execution_count": 26,
"id": "3890af2f-fb14-4569-b48d-a91db2d3f026",
"metadata": {},
"outputs": [],
@@ -1813,13 +1845,14 @@
"builder.add_node(b)\n",
"builder.add_node(b_2)\n",
"builder.add_node(c)\n",
"builder.add_node(d)\n",
"# highlight-next-line\n",
"builder.add_node(d, defer=True)\n",
"builder.add_edge(START, \"a\")\n",
"builder.add_edge(\"a\", \"b\")\n",
"builder.add_edge(\"a\", \"c\")\n",
"builder.add_edge(\"b\", \"b_2\")\n",
"# highlight-next-line\n",
"builder.add_edge([\"b_2\", \"c\"], \"d\")\n",
"builder.add_edge(\"b_2\", \"d\")\n",
"builder.add_edge(\"c\", \"d\")\n",
"builder.add_edge(\"d\", END)\n",
"graph = builder.compile()"
]
@@ -1881,23 +1914,10 @@
},
{
"cell_type": "markdown",
"id": "903f0da5-8c2c-4a7e-96fb-0b16b4756eff",
"id": "70e67ced",
"metadata": {},
"source": [
"<div class=\"admonition note\">\n",
" <p class=\"admonition-title\">Note</p>\n",
"<p>In the above example, nodes <code>\"b\"</code> and <code>\"c\"</code> are executed concurrently in the same [superstep](../../concepts/low_level/#graphs). What happens in the next step?</p>\n",
" <p>We use <code>add_edge([\"b_2\", \"c\"], \"d\")</code> here to force node <code>\"d\"</code> to only run when both nodes <code>\"b_2\"</code> and <code>\"c\"</code> have finished execution. If we added two separate edges,\n",
" node <code>\"d\"</code> would run twice: after node <code>b2</code> finishes and once again after node <code>c</code> (in whichever order those nodes finish).</p>\n",
"</div>"
]
},
{
"cell_type": "markdown",
"id": "c1653341-3215-4ca0-b0e7-9be22f0adaa1",
"metadata": {},
"source": [
"</details>"
"In the above example, nodes `\"b\"` and `\"c\"` are executed concurrently in the same superstep. We set `defer=True` on node `d` so it will not execute until all pending tasks are finished. In this case, this means that `\"d\"` waits to execute until the entire `\"b\"` branch is finished."
]
},
{
+2 -2
View File
@@ -184,7 +184,7 @@
"??? example \"Example: using [Postgres](https://pypi.org/project/langgraph-checkpoint-postgres/) checkpointer\"\n",
"\n",
" ```\n",
" pip install -U psycopg psycopg-pool langgraph langgraph-checkpoint-postgres\n",
" pip install -U \"psycopg[binary,pool]\" langgraph langgraph-checkpoint-postgres\n",
" ```\n",
"\n",
" !!! Setup\n",
@@ -1129,7 +1129,7 @@
"??? example \"Example: using [Postgres](https://pypi.org/project/langgraph-checkpoint-postgres/) store\"\n",
"\n",
" ```\n",
" pip install -U psycopg psycopg-pool langgraph langgraph-checkpoint-postgres\n",
" pip install -U \"psycopg[binary,pool]\" langgraph langgraph-checkpoint-postgres\n",
" ```\n",
"\n",
" !!! Setup\n",
+10 -8
View File
@@ -71,7 +71,7 @@ Basic usage example:
| [`values`](#stream-graph-state) | Streams the full value of the state after each step of the graph. |
| [`updates`](#stream-graph-state) | Streams the updates to the state after each step of the graph. If multiple updates are made in the same step (e.g., multiple nodes are run), those updates are streamed separately. |
| [`custom`](#stream-custom-data) | Streams custom data from inside your graph nodes. |
| [`messages`](#messages) | Streams LLM tokens and metadata for the graph node where the LLM is invoked. |
| [`messages`](#messages) | Streams 2-tuples (LLM token, metadata) from any graph nodes where an LLM is invoked. |
| [`debug`](#debug) | Streams as much information as possible throughout the execution of the graph. |
### Stream multiple modes
@@ -161,6 +161,8 @@ graph = (
To include outputs from [subgraphs](../concepts/subgraphs.md) in the streamed outputs, you can set `subgraphs=True` in the `.stream()` method of the parent graph. This will stream outputs from both the parent graph and any subgraphs.
The outputs will be streamed as tuples `(namespace, data)`, where `namespace` is a tuple with the path to the node where a subgraph is invoked, e.g. `("parent_node:<task_id>", "child_node:<task_id>")`.
```python
for chunk in graph.stream(
{"foo": "foo"},
@@ -179,21 +181,17 @@ for chunk in graph.stream(
from langgraph.graph import START, StateGraph
from typing import TypedDict
# Define subgraph
class SubgraphState(TypedDict):
foo: str # note that this key is shared with the parent graph state
bar: str
def subgraph_node_1(state: SubgraphState):
return {"bar": "bar"}
def subgraph_node_2(state: SubgraphState):
return {"foo": state["foo"] + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
@@ -201,16 +199,13 @@ for chunk in graph.stream(
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# Define parent graph
class ParentState(TypedDict):
foo: str
def node_1(state: ParentState):
return {"foo": "hi! " + state["foo"]}
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
@@ -229,6 +224,13 @@ for chunk in graph.stream(
1. Set `subgraphs=True` to stream outputs from subgraphs.
```
((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_1': {'bar': 'bar'}})
(('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
((), {'node_2': {'foo': 'hi! foobar'}})
```
**Note** that we are receiving not just the node updates, but we also the namespaces which tell us what graph (or subgraph) we are streaming from.
## Debugging {#debug}
+32
View File
@@ -349,6 +349,38 @@ main.invoke({'any_input': 'foobar'}, config=config)
'OK'
```
## Caching Tasks
```python
import time
from langgraph.cache.memory import InMemoryCache
from langgraph.func import entrypoint, task
from langgraph.types import CachePolicy
@task(cache_policy=CachePolicy(ttl=120)) # (1)!
def slow_add(x: int) -> int:
time.sleep(1)
return x * 2
@entrypoint(cache=InMemoryCache())
def main(inputs: dict) -> dict[str, int]:
result1 = slow_add(inputs["x"]).result()
result2 = slow_add(inputs["x"]).result()
return {"result1": result1, "result2": result2}
for chunk in main.stream({"x": 5}, stream_mode="updates"):
print(chunk)
#> {'slow_add': 10}
#> {'slow_add': 10, '__metadata__': {'cached': True}}
#> {'main': {'result1': 10, 'result2': 10}}
```
1. `ttl` is specified in seconds. The cache will be invalidated after this time.
## Resuming after an error
```python
+5
View File
@@ -0,0 +1,5 @@
## Caching
::: langgraph.cache.base
::: langgraph.cache.memory
::: langgraph.cache.sqlite
+1
View File
@@ -29,6 +29,7 @@ The core APIs for the LangGraph opens source library.
- [Pregel](pregel.md): Pregel-inspired computation model.
- [Checkpointing](checkpoints.md): Saving and restoring graph state.
- [Storage](store.md): Storage backends and options.
- [Caching](cache.md): Caching mechanisms for performance.
- [Types](types.md): Type definitions for graph components.
- [Config](config.md): Configuration options.
- [Errors](errors.md): Error types and handling.
@@ -6,7 +6,7 @@ There could be a few reasons you're seeing this error:
1. You manually passed a malformed list of messages when invoking the graph, e.g. `graph.invoke({'messages': [AIMessage(..., tool_calls=[...])]})`
2. The graph was interrupted before receiving updates from the `tools` node (i.e. a list of ToolMessages)
and you invoked it with a an input that is not None or a ToolMessage,
and you invoked it with an input that is not None or a ToolMessage,
e.g. `graph.invoke({'messages': [HumanMessage(...)]}, config)`.
This interrupt could have been triggered in one of the following ways:
- You manually set `interrupt_before = ['tools']` in `create_react_agent`
@@ -8,7 +8,7 @@ class State(TypedDict):
some_key: str
def bad_node(state: State):
# Should return an dict with a value for "some_key", not a list
# Should return a dict with a value for "some_key", not a list
return ["whoops"]
builder = StateGraph(State)
@@ -29,7 +29,7 @@ InvalidUpdateError: Expected dict, got ['whoops']
For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/INVALID_GRAPH_NODE_RETURN_VALUE
```
Nodes in your graph must return an dict containing one or more keys defined in your state.
Nodes in your graph must return a dict containing one or more keys defined in your state.
## Troubleshooting
+2 -2
View File
@@ -1,6 +1,6 @@
# Connect an authentication provider
In the [the last tutorial](resource_auth.md), you added [resource authorization](../../tutorials/auth/resource_auth.md) to give users private conversations. However, you are still using hard-coded tokens for authentication, which is not secure. Now you'll replace those tokens with real user accounts using [OAuth2](../auth/getting_started.md).
In [the last tutorial](resource_auth.md), you added [resource authorization](../../tutorials/auth/resource_auth.md) to give users private conversations. However, you are still using hard-coded tokens for authentication, which is not secure. Now you'll replace those tokens with real user accounts using [OAuth2](../auth/getting_started.md).
You'll keep the same [`Auth`](../../cloud/reference/sdk/python_sdk_ref.md#langgraph_sdk.auth.Auth) object and [resource-level access control](../../concepts/auth.md#single-owner-resources), but upgrade authentication to use Supabase as your identity provider. While Supabase is used in this tutorial, the concepts apply to any OAuth2 provider. You'll learn how to:
@@ -190,7 +190,7 @@ await sign_up(email1, password)
await sign_up(email2, password)
```
⚠️ Before continuing: Check your email and click both confirmation links. Supabase will will reject `/login` requests until after you have confirmed your users' email.
⚠️ Before continuing: Check your email and click both confirmation links. Supabase will reject `/login` requests until after you have confirmed your users' email.
Now test that users can only see their own data. Make sure the server is running (run `langgraph dev`) before proceeding. The following snippet requires the "anon public" key that you copied from the Supabase dashboard while [setting up the auth provider](#setup-auth-provider) previously.
+1 -1
View File
@@ -181,6 +181,6 @@ Congratulations! You've built a chatbot that only lets "authenticated" users acc
Now that you can control who accesses your bot, you might want to:
1. Continue the tutorial by going to [Make cnversations private](resource_auth.md) to learn about resource authorization.
1. Continue the tutorial by going to [Make conversations private](resource_auth.md) to learn about resource authorization.
2. Read more about [authentication concepts](../../concepts/auth.md).
3. Check out the [API reference](../../cloud/reference/sdk/python_sdk_ref.md) for more authentication details.
@@ -21,7 +21,7 @@ Before you begin, ensure you have the following:
=== "Node server"
```shell
npx @langchain/langgraph-cl
npx @langchain/langgraph-cli
```
## 2. Create a LangGraph app 🌱
+1
View File
@@ -255,6 +255,7 @@ nav:
- Pregel: reference/pregel.md
- Checkpointing: reference/checkpoints.md
- Storage: reference/store.md
- Caching: reference/cache.md
- Types: reference/types.md
- Config: reference/config.md
- Errors: reference/errors.md
Generated
+70 -55
View File
@@ -291,16 +291,16 @@ name = "autogen"
version = "0.3.2"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "diskcache" },
{ name = "docker" },
{ name = "flaml" },
{ name = "numpy" },
{ name = "openai" },
{ name = "packaging" },
{ name = "pydantic" },
{ name = "python-dotenv" },
{ name = "termcolor" },
{ name = "tiktoken" },
{ name = "diskcache", marker = "python_full_version < '3.13'" },
{ name = "docker", marker = "python_full_version < '3.13'" },
{ name = "flaml", marker = "python_full_version < '3.13'" },
{ name = "numpy", marker = "python_full_version < '3.13'" },
{ name = "openai", marker = "python_full_version < '3.13'" },
{ name = "packaging", marker = "python_full_version < '3.13'" },
{ name = "pydantic", marker = "python_full_version < '3.13'" },
{ name = "python-dotenv", marker = "python_full_version < '3.13'" },
{ name = "termcolor", marker = "python_full_version < '3.13'" },
{ name = "tiktoken", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/7b/e8/33b7fb072fbcf63b8a1b5bbba15570e4e8c86d6374da398889b92fc420c8/autogen-0.3.2.tar.gz", hash = "sha256:9f8a1170ac2e5a1fc9efc3cfa6e23261dd014db97b17c8c416f97ee14951bc7b", size = 306281 }
wheels = [
@@ -1008,9 +1008,9 @@ name = "docker"
version = "7.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pywin32", marker = "sys_platform == 'win32'" },
{ name = "requests" },
{ name = "urllib3" },
{ name = "pywin32", marker = "python_full_version < '3.13' and sys_platform == 'win32'" },
{ name = "requests", marker = "python_full_version < '3.13'" },
{ name = "urllib3", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/91/9b/4a2ea29aeba62471211598dac5d96825bb49348fa07e906ea930394a83ce/docker-7.1.0.tar.gz", hash = "sha256:ad8c70e6e3f8926cb8a92619b832b4ea5299e2831c14284663184e200546fa6c", size = 117834 }
wheels = [
@@ -1052,7 +1052,7 @@ name = "exceptiongroup"
version = "1.3.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "typing-extensions", marker = "python_full_version < '3.12.4'" },
{ name = "typing-extensions", marker = "python_full_version < '3.11'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/0b/9f/a65090624ecf468cdca03533906e7c69ed7588582240cfe7cc9e770b50eb/exceptiongroup-1.3.0.tar.gz", hash = "sha256:b241f5885f560bc56a59ee63ca4c6a8bfa46ae4ad651af316d4e81817bb9fd88", size = 29749 }
wheels = [
@@ -1153,7 +1153,7 @@ name = "flaml"
version = "2.3.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "numpy" },
{ name = "numpy", marker = "python_full_version < '3.13'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/20/a8/17322311b77f3012194f92c47c81455463f99c48d358c463fa45bd3c8541/flaml-2.3.4.tar.gz", hash = "sha256:308c3e769976d8a0272f2fd7d98258d7d4a4fd2e4525ba540d1ba149ae266c54", size = 284728 }
wheels = [
@@ -2590,7 +2590,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.4.3"
version = "0.4.5"
source = { editable = "../libs/langgraph" }
dependencies = [
{ name = "langchain-core" },
@@ -2641,7 +2641,7 @@ dev = [
[[package]]
name = "langgraph-checkpoint"
version = "2.0.25"
version = "2.0.26"
source = { editable = "../libs/checkpoint" }
dependencies = [
{ name = "langchain-core" },
@@ -2715,17 +2715,19 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.7"
version = "2.0.10"
source = { editable = "../libs/checkpoint-sqlite" }
dependencies = [
{ name = "aiosqlite" },
{ name = "langgraph-checkpoint" },
{ name = "sqlite-vec" },
]
[package.metadata]
requires-dist = [
{ name = "aiosqlite", specifier = ">=0.20" },
{ name = "langgraph-checkpoint", editable = "../libs/checkpoint" },
{ name = "sqlite-vec", specifier = ">=0.1.6" },
]
[package.metadata.requires-dev]
@@ -2736,6 +2738,7 @@ dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
{ name = "pytest-retry", specifier = ">=1.7.0" },
{ name = "pytest-watcher" },
{ name = "ruff" },
]
@@ -2825,11 +2828,11 @@ requires-dist = [
[package.metadata.requires-dev]
docs = [
{ name = "click", specifier = ">=8.1.7,<9" },
{ name = "jupyter", specifier = ">=1.1.1,<2" },
{ name = "langchain-cohere", specifier = ">=0.4.2,<0.5" },
{ name = "click" },
{ name = "jupyter" },
{ name = "langchain-cohere" },
{ name = "langchain-mcp-adapters", git = "https://github.com/langchain-ai/langchain-mcp-adapters" },
{ name = "langchain-ollama", specifier = ">=0.2.3,<0.3" },
{ name = "langchain-ollama" },
{ name = "langgraph", editable = "../libs/langgraph" },
{ name = "langgraph-checkpoint", editable = "../libs/checkpoint" },
{ name = "langgraph-checkpoint-postgres", editable = "../libs/checkpoint-postgres" },
@@ -2849,41 +2852,41 @@ docs = [
{ name = "mkdocs-rss-plugin" },
{ name = "mkdocstrings" },
{ name = "mkdocstrings-python" },
{ name = "psycopg", extras = ["binary"], specifier = ">=3.2.0,<4" },
{ name = "psycopg-pool", specifier = ">=3.2.0,<4" },
{ name = "pygments-ansi-color", specifier = ">=0.3" },
{ name = "ruff", specifier = ">=0.6.8,<0.7" },
{ name = "vcrpy", specifier = ">=6.0.1,<7" },
{ name = "psycopg", extras = ["binary"] },
{ name = "psycopg-pool" },
{ name = "pygments-ansi-color" },
{ name = "ruff" },
{ name = "vcrpy" },
]
test = [
{ name = "autogen", marker = "python_full_version >= '3.8' and python_full_version < '3.13'", specifier = ">=0.3.0,<0.4" },
{ name = "chromadb", specifier = ">=0.5.5,<0.6" },
{ name = "gpt4all", specifier = ">=2.8.2,<3" },
{ name = "grandalf", specifier = ">=0.8,<0.9" },
{ name = "langchain", specifier = ">=0.3.8,<0.4" },
{ name = "langchain-anthropic", specifier = ">=0.3.8,<0.4" },
{ name = "langchain-community", specifier = ">=0.3.0,<0.4" },
{ name = "langchain-core", specifier = ">=0.3.54,<0.4" },
{ name = "langchain-experimental", specifier = ">=0.3.2,<0.4" },
{ name = "langchain-fireworks", specifier = ">=0.2.0,<0.3" },
{ name = "langchain-mistralai", specifier = ">=0.2.6,<0.3" },
{ name = "langchain-nomic", specifier = ">=0.1.3,<0.2" },
{ name = "langchain-openai", specifier = ">=0.3.7,<0.4" },
{ name = "langchain-tavily", specifier = ">=0.1.5,<0.2" },
{ name = "langgraph-checkpoint-mongodb", specifier = ">=0.1.0,<0.2" },
{ name = "langmem", specifier = ">=0.0.19,<0.0.20" },
{ name = "langsmith", specifier = ">=0.3.0,<0.4" },
{ name = "matplotlib", specifier = ">=3.9.2,<4" },
{ name = "motor", specifier = ">=3.5.1,<4" },
{ name = "networkx", specifier = "~=3.3" },
{ name = "numexpr", specifier = ">=2.10.1,<3" },
{ name = "numpy", specifier = ">=1.26.4,<2" },
{ name = "pymongo", specifier = ">=4.8.0,<5" },
{ name = "pyppeteer", specifier = ">=2.0.0,<3" },
{ name = "pytest", specifier = ">=8.3.5,<9" },
{ name = "pytest-check-links", specifier = ">=0.10.1,<0.11" },
{ name = "redis", specifier = ">=5.0.8,<6" },
{ name = "scikit-learn", specifier = ">=1.5.2,<2" },
{ name = "autogen", marker = "python_full_version >= '3.8' and python_full_version < '3.13'" },
{ name = "chromadb" },
{ name = "gpt4all" },
{ name = "grandalf" },
{ name = "langchain" },
{ name = "langchain-anthropic" },
{ name = "langchain-community" },
{ name = "langchain-core" },
{ name = "langchain-experimental" },
{ name = "langchain-fireworks" },
{ name = "langchain-mistralai" },
{ name = "langchain-nomic" },
{ name = "langchain-openai" },
{ name = "langchain-tavily" },
{ name = "langgraph-checkpoint-mongodb" },
{ name = "langmem" },
{ name = "langsmith" },
{ name = "matplotlib" },
{ name = "motor" },
{ name = "networkx" },
{ name = "numexpr" },
{ name = "numpy" },
{ name = "pymongo" },
{ name = "pyppeteer" },
{ name = "pytest" },
{ name = "pytest-check-links" },
{ name = "redis" },
{ name = "scikit-learn" },
]
[[package]]
@@ -5769,6 +5772,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1c/fc/9ba22f01b5cdacc8f5ed0d22304718d2c758fce3fd49a5372b886a86f37c/sqlalchemy-2.0.41-py3-none-any.whl", hash = "sha256:57df5dc6fdb5ed1a88a1ed2195fd31927e705cad62dedd86b46972752a80f576", size = 1911224 },
]
[[package]]
name = "sqlite-vec"
version = "0.1.6"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/88/ed/aabc328f29ee6814033d008ec43e44f2c595447d9cccd5f2aabe60df2933/sqlite_vec-0.1.6-py3-none-macosx_10_6_x86_64.whl", hash = "sha256:77491bcaa6d496f2acb5cc0d0ff0b8964434f141523c121e313f9a7d8088dee3", size = 164075 },
{ url = "https://files.pythonhosted.org/packages/a7/57/05604e509a129b22e303758bfa062c19afb020557d5e19b008c64016704e/sqlite_vec-0.1.6-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fdca35f7ee3243668a055255d4dee4dea7eed5a06da8cad409f89facf4595361", size = 165242 },
{ url = "https://files.pythonhosted.org/packages/f2/48/dbb2cc4e5bad88c89c7bb296e2d0a8df58aab9edc75853728c361eefc24f/sqlite_vec-0.1.6-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7b0519d9cd96164cd2e08e8eed225197f9cd2f0be82cb04567692a0a4be02da3", size = 103704 },
{ url = "https://files.pythonhosted.org/packages/80/76/97f33b1a2446f6ae55e59b33869bed4eafaf59b7f4c662c8d9491b6a714a/sqlite_vec-0.1.6-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux1_x86_64.whl", hash = "sha256:823b0493add80d7fe82ab0fe25df7c0703f4752941aee1c7b2b02cec9656cb24", size = 151556 },
{ url = "https://files.pythonhosted.org/packages/6a/98/e8bc58b178266eae2fcf4c9c7a8303a8d41164d781b32d71097924a6bebe/sqlite_vec-0.1.6-py3-none-win_amd64.whl", hash = "sha256:c65bcfd90fa2f41f9000052bcb8bb75d38240b2dae49225389eca6c3136d3f0c", size = 281540 },
]
[[package]]
name = "sse-starlette"
version = "2.3.5"
+4 -2
View File
@@ -4,11 +4,13 @@
# TESTING AND COVERAGE
######################
TEST ?= .
test:
uv run pytest tests
uv run pytest $(TEST)
test_watch:
uv run ptw .
uv run ptw $(TEST)
######################
# LINTING AND FORMATTING
@@ -0,0 +1,4 @@
from langgraph.store.sqlite.aio import AsyncSqliteStore
from langgraph.store.sqlite.base import SqliteStore
__all__ = ["AsyncSqliteStore", "SqliteStore"]
@@ -0,0 +1,582 @@
import asyncio
import logging
from collections import defaultdict
from collections.abc import AsyncIterator, Iterable, Sequence
from contextlib import asynccontextmanager
from types import TracebackType
from typing import Any, Callable, Optional, Union, cast
import aiosqlite
import orjson
import sqlite_vec # type: ignore[import-untyped]
from langgraph.store.base import (
GetOp,
ListNamespacesOp,
Op,
PutOp,
Result,
SearchOp,
TTLConfig,
)
from langgraph.store.base.batch import AsyncBatchedBaseStore
from langgraph.store.sqlite.base import (
_PLACEHOLDER,
BaseSqliteStore,
SqliteIndexConfig,
_decode_ns_text,
_ensure_index_config,
_group_ops,
_row_to_item,
_row_to_search_item,
)
logger = logging.getLogger(__name__)
class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
"""Asynchronous SQLite-backed store with optional vector search.
This class provides an asynchronous interface for storing and retrieving data
using a SQLite database with support for vector search capabilities.
Examples:
Basic setup and usage:
```python
from langgraph.store.sqlite import AsyncSqliteStore
async with AsyncSqliteStore.from_conn_string(":memory:") as store:
await store.setup() # Run migrations
# Store and retrieve data
await store.aput(("users", "123"), "prefs", {"theme": "dark"})
item = await store.aget(("users", "123"), "prefs")
```
Vector search using LangChain embeddings:
```python
from langchain_openai import OpenAIEmbeddings
from langgraph.store.sqlite import AsyncSqliteStore
async with AsyncSqliteStore.from_conn_string(
":memory:",
index={
"dims": 1536,
"embed": OpenAIEmbeddings(),
"fields": ["text"] # specify which fields to embed
}
) as store:
await store.setup() # Run migrations once
# Store documents
await store.aput(("docs",), "doc1", {"text": "Python tutorial"})
await store.aput(("docs",), "doc2", {"text": "TypeScript guide"})
await store.aput(("docs",), "doc3", {"text": "Other guide"}, index=False) # don't index
# Search by similarity
results = await store.asearch(("docs",), query="programming guides", limit=2)
```
Warning:
Make sure to call `setup()` before first use to create necessary tables and indexes.
Note:
This class requires the aiosqlite package. Install with `pip install aiosqlite`.
"""
def __init__(
self,
conn: aiosqlite.Connection,
*,
deserializer: Optional[
Callable[[Union[bytes, str, orjson.Fragment]], dict[str, Any]]
] = None,
index: Optional[SqliteIndexConfig] = None,
ttl: Optional[TTLConfig] = None,
):
"""Initialize the async SQLite store.
Args:
conn: The SQLite database connection.
deserializer: Optional custom deserializer function for values.
index: Optional vector search configuration.
ttl: Optional time-to-live configuration.
"""
super().__init__()
self._deserializer = deserializer
self.conn = conn
self.lock = asyncio.Lock()
self.loop = asyncio.get_running_loop()
self.is_setup = False
self.index_config = index
if self.index_config:
self.embeddings, self.index_config = _ensure_index_config(self.index_config)
else:
self.embeddings = None
self.ttl_config = ttl
self._ttl_sweeper_task: Optional[asyncio.Task[None]] = None
self._ttl_stop_event = asyncio.Event()
@classmethod
@asynccontextmanager
async def from_conn_string(
cls,
conn_string: str,
*,
index: Optional[SqliteIndexConfig] = None,
ttl: Optional[TTLConfig] = None,
) -> AsyncIterator["AsyncSqliteStore"]:
"""Create a new AsyncSqliteStore instance from a connection string.
Args:
conn_string: The SQLite connection string.
index: Optional vector search configuration.
ttl: Optional time-to-live configuration.
Returns:
An AsyncSqliteStore instance wrapped in an async context manager.
"""
async with aiosqlite.connect(conn_string, isolation_level=None) as conn:
yield cls(conn, index=index, ttl=ttl)
async def setup(self) -> None:
"""Set up the store database.
This method creates the necessary tables in the SQLite database if they don't
already exist and runs database migrations. It should be called before first use.
"""
async with self.lock:
if self.is_setup:
return
# Create migrations table if it doesn't exist
await self.conn.execute(
"""
CREATE TABLE IF NOT EXISTS store_migrations (
v INTEGER PRIMARY KEY
)
"""
)
# Check current migration version
async with self.conn.execute(
"SELECT v FROM store_migrations ORDER BY v DESC LIMIT 1"
) as cur:
row = await cur.fetchone()
if row is None:
version = -1
else:
version = row[0]
# Apply migrations
for v, sql in enumerate(self.MIGRATIONS[version + 1 :], start=version + 1):
await self.conn.executescript(sql)
await self.conn.execute(
"INSERT INTO store_migrations (v) VALUES (?)", (v,)
)
# Apply vector migrations if index config is provided
if self.index_config:
# Create vector migrations table if it doesn't exist
await self.conn.enable_load_extension(True)
await self.conn.load_extension(sqlite_vec.loadable_path())
await self.conn.enable_load_extension(False)
await self.conn.execute(
"""
CREATE TABLE IF NOT EXISTS vector_migrations (
v INTEGER PRIMARY KEY
)
"""
)
# Check current vector migration version
async with self.conn.execute(
"SELECT v FROM vector_migrations ORDER BY v DESC LIMIT 1"
) as cur:
row = await cur.fetchone()
if row is None:
version = -1
else:
version = row[0]
# Apply vector migrations
for v, sql in enumerate(
self.VECTOR_MIGRATIONS[version + 1 :], start=version + 1
):
await self.conn.executescript(sql)
await self.conn.execute(
"INSERT INTO vector_migrations (v) VALUES (?)", (v,)
)
self.is_setup = True
@asynccontextmanager
async def _cursor(
self, *, transaction: bool = True
) -> AsyncIterator[aiosqlite.Cursor]:
"""Get a cursor for the SQLite database.
Args:
transaction: Whether to use a transaction for database operations.
Yields:
An SQLite cursor object.
"""
async with self.lock:
if not self.is_setup:
await self.setup()
if transaction:
await self.conn.execute("BEGIN")
async with self.conn.cursor() as cur:
try:
yield cur
finally:
if transaction:
await self.conn.execute("COMMIT")
async def sweep_ttl(self) -> int:
"""Delete expired store items based on TTL.
Returns:
int: The number of deleted items.
"""
async with self._cursor() as cur:
await cur.execute(
"""
DELETE FROM store
WHERE expires_at IS NOT NULL AND expires_at < CURRENT_TIMESTAMP
"""
)
deleted_count = cur.rowcount
return deleted_count
async def start_ttl_sweeper(
self, sweep_interval_minutes: Optional[int] = None
) -> asyncio.Task[None]:
"""Periodically delete expired store items based on TTL.
Returns:
Task that can be awaited or cancelled.
"""
if not self.ttl_config:
return asyncio.create_task(asyncio.sleep(0))
if self._ttl_sweeper_task is not None and not self._ttl_sweeper_task.done():
return self._ttl_sweeper_task
self._ttl_stop_event.clear()
interval = float(
sweep_interval_minutes or self.ttl_config.get("sweep_interval_minutes") or 5
)
logger.info(f"Starting store TTL sweeper with interval {interval} minutes")
async def _sweep_loop() -> None:
while not self._ttl_stop_event.is_set():
try:
try:
await asyncio.wait_for(
self._ttl_stop_event.wait(),
timeout=interval * 60,
)
break
except asyncio.TimeoutError:
pass
expired_items = await self.sweep_ttl()
if expired_items > 0:
logger.info(f"Store swept {expired_items} expired items")
except asyncio.CancelledError:
break
except Exception as exc:
logger.exception("Store TTL sweep iteration failed", exc_info=exc)
task = asyncio.create_task(_sweep_loop())
task.set_name("ttl_sweeper")
self._ttl_sweeper_task = task
return task
async def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
"""Stop the TTL sweeper task if it's running.
Args:
timeout: Maximum time to wait for the task to stop, in seconds.
If None, wait indefinitely.
Returns:
bool: True if the task was successfully stopped or wasn't running,
False if the timeout was reached before the task stopped.
"""
if self._ttl_sweeper_task is None or self._ttl_sweeper_task.done():
return True
logger.info("Stopping TTL sweeper task")
self._ttl_stop_event.set()
if timeout is not None:
try:
await asyncio.wait_for(self._ttl_sweeper_task, timeout=timeout)
success = True
except asyncio.TimeoutError:
success = False
else:
await self._ttl_sweeper_task
success = True
if success:
self._ttl_sweeper_task = None
logger.info("TTL sweeper task stopped")
else:
logger.warning("Timed out waiting for TTL sweeper task to stop")
return success
async def __aenter__(self) -> "AsyncSqliteStore":
return self
async def __aexit__(
self,
exc_type: Optional[type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional["TracebackType"],
) -> None:
# Ensure the TTL sweeper task is stopped when exiting the context
if hasattr(self, "_ttl_sweeper_task") and self._ttl_sweeper_task is not None:
# Set the event to signal the task to stop
self._ttl_stop_event.set()
# We don't wait for the task to complete here to avoid blocking
# The task will clean up itself gracefully
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
"""Execute a batch of operations asynchronously.
Args:
ops: Iterable of operations to execute.
Returns:
List of operation results.
"""
grouped_ops, num_ops = _group_ops(ops)
results: list[Result] = [None] * num_ops
async with self._cursor(transaction=True) as cur:
if GetOp in grouped_ops:
await self._batch_get_ops(
cast(Sequence[tuple[int, GetOp]], grouped_ops[GetOp]), results, cur
)
if SearchOp in grouped_ops:
await self._batch_search_ops(
cast(Sequence[tuple[int, SearchOp]], grouped_ops[SearchOp]),
results,
cur,
)
if ListNamespacesOp in grouped_ops:
await self._batch_list_namespaces_ops(
cast(
Sequence[tuple[int, ListNamespacesOp]],
grouped_ops[ListNamespacesOp],
),
results,
cur,
)
if PutOp in grouped_ops:
await self._batch_put_ops(
cast(Sequence[tuple[int, PutOp]], grouped_ops[PutOp]), cur
)
return results
async def _batch_get_ops(
self,
get_ops: Sequence[tuple[int, GetOp]],
results: list[Result],
cur: aiosqlite.Cursor,
) -> None:
"""Process batch GET operations.
Args:
get_ops: Sequence of GET operations.
results: List to store results in.
cur: Database cursor.
"""
# Group all queries by namespace to execute all operations for each namespace together
namespace_queries = defaultdict(list)
for prepared_query in self._get_batch_GET_ops_queries(get_ops):
namespace_queries[prepared_query.namespace].append(prepared_query)
# Process each namespace's operations
for namespace, queries in namespace_queries.items():
# Execute TTL refresh queries first
for query in queries:
if query.kind == "refresh":
try:
await cur.execute(query.query, query.params)
except Exception as e:
raise ValueError(
f"Error executing TTL refresh: \n{query.query}\n{query.params}\n{e}"
) from e
# Then execute GET queries and process results
for query in queries:
if query.kind == "get":
try:
await cur.execute(query.query, query.params)
except Exception as e:
raise ValueError(
f"Error executing GET query: \n{query.query}\n{query.params}\n{e}"
) from e
rows = await cur.fetchall()
key_to_row = {
row[0]: {
"key": row[0],
"value": row[1],
"created_at": row[2],
"updated_at": row[3],
"expires_at": row[4] if len(row) > 4 else None,
"ttl_minutes": row[5] if len(row) > 5 else None,
}
for row in rows
}
# Process results for this query
for idx, key in query.items:
row = key_to_row.get(key)
if row:
results[idx] = _row_to_item(
namespace, row, loader=self._deserializer
)
else:
results[idx] = None
async def _batch_put_ops(
self,
put_ops: Sequence[tuple[int, PutOp]],
cur: aiosqlite.Cursor,
) -> None:
"""Process batch PUT operations.
Args:
put_ops: Sequence of PUT operations.
cur: Database cursor.
"""
queries, embedding_request = self._prepare_batch_PUT_queries(put_ops)
if embedding_request:
if self.embeddings is None:
# Should not get here since the embedding config is required
# to return an embedding_request above
raise ValueError(
"Embedding configuration is required for vector operations "
f"(for semantic search). "
f"Please provide an Embeddings when initializing the {self.__class__.__name__}."
)
query, txt_params = embedding_request
# Update the params to replace the raw text with the vectors
vectors = await self.embeddings.aembed_documents(
[param[-1] for param in txt_params]
)
# Convert vectors to SQLite-friendly format
vector_params = []
for (ns, k, pathname, _), vector in zip(txt_params, vectors):
vector_params.extend(
[ns, k, pathname, sqlite_vec.serialize_float32(vector)]
)
queries.append((query, vector_params))
for query, params in queries:
await cur.execute(query, params)
async def _batch_search_ops(
self,
search_ops: Sequence[tuple[int, SearchOp]],
results: list[Result],
cur: aiosqlite.Cursor,
) -> None:
"""Process batch SEARCH operations.
Args:
search_ops: Sequence of SEARCH operations.
results: List to store results in.
cur: Database cursor.
"""
queries, embedding_requests = self._prepare_batch_search_queries(search_ops)
# Setup dot_product function if it doesn't exist
if embedding_requests and self.embeddings:
vectors = await self.embeddings.aembed_documents(
[query for _, query in embedding_requests]
)
for (idx, _), embedding in zip(embedding_requests, vectors):
_params_list: list = queries[idx][1]
for i, param in enumerate(_params_list):
if param is _PLACEHOLDER:
_params_list[i] = sqlite_vec.serialize_float32(embedding)
for (idx, _), (query, params) in zip(search_ops, queries):
await cur.execute(query, params)
rows = await cur.fetchall()
if "score" in query:
items = [
_row_to_search_item(
_decode_ns_text(row[0]),
{
"key": row[1],
"value": row[2],
"created_at": row[3],
"updated_at": row[4],
"expires_at": row[5] if len(row) > 5 else None,
"ttl_minutes": row[6] if len(row) > 6 else None,
"score": row[7] if len(row) > 7 else None,
},
loader=self._deserializer,
)
for row in rows
]
else: # Regular search query
items = [
_row_to_search_item(
_decode_ns_text(row[0]),
{
"key": row[1],
"value": row[2],
"created_at": row[3],
"updated_at": row[4],
"expires_at": row[5] if len(row) > 5 else None,
"ttl_minutes": row[6] if len(row) > 6 else None,
},
loader=self._deserializer,
)
for row in rows
]
results[idx] = items
async def _batch_list_namespaces_ops(
self,
list_ops: Sequence[tuple[int, ListNamespacesOp]],
results: list[Result],
cur: aiosqlite.Cursor,
) -> None:
"""Process batch LIST NAMESPACES operations.
Args:
list_ops: Sequence of LIST NAMESPACES operations.
results: List to store results in.
cur: Database cursor.
"""
queries = self._get_batch_list_namespaces_queries(list_ops)
for (query, params), (idx, _) in zip(queries, list_ops):
await cur.execute(query, params)
rows = await cur.fetchall()
results[idx] = [_decode_ns_text(row[0]) for row in rows]
File diff suppressed because it is too large Load Diff
+4 -2
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-checkpoint-sqlite"
version = "2.0.7"
version = "2.0.10"
description = "Library with a SQLite implementation of LangGraph checkpoint saver."
authors = []
requires-python = ">=3.9"
@@ -12,8 +12,9 @@ readme = "README.md"
license = "MIT"
license-files = ['LICENSE']
dependencies = [
"langgraph-checkpoint>=2.0.15",
"langgraph-checkpoint>=2.0.21",
"aiosqlite>=0.20",
"sqlite-vec>=0.1.6",
]
[project.urls]
@@ -29,6 +30,7 @@ dev = [
"pytest-watcher",
"mypy",
"langgraph-checkpoint",
"pytest-retry>=1.7.0",
]
[tool.uv]
@@ -0,0 +1,719 @@
# mypy: disable-error-code="union-attr,arg-type,index,operator"
import asyncio
import os
import tempfile
import uuid
from collections.abc import AsyncIterator, Generator, Iterable
from contextlib import asynccontextmanager
from typing import Optional, Union, cast
import pytest
from langgraph.store.base import (
GetOp,
Item,
ListNamespacesOp,
PutOp,
SearchOp,
)
from langgraph.store.sqlite import AsyncSqliteStore
from langgraph.store.sqlite.base import SqliteIndexConfig
from tests.test_store import CharacterEmbeddings
@pytest.fixture(scope="function", params=["memory", "file"])
async def store(request: pytest.FixtureRequest) -> AsyncIterator[AsyncSqliteStore]:
"""Create an AsyncSqliteStore for testing."""
if request.param == "memory":
# In-memory store
async with AsyncSqliteStore.from_conn_string(":memory:") as store:
await store.setup()
yield store
else:
# Temporary file store
temp_file = tempfile.NamedTemporaryFile(delete=False)
temp_file.close()
try:
async with AsyncSqliteStore.from_conn_string(temp_file.name) as store:
await store.setup()
yield store
finally:
os.unlink(temp_file.name)
@pytest.fixture(scope="function")
def fake_embeddings() -> CharacterEmbeddings:
"""Create fake embeddings for testing."""
return CharacterEmbeddings(dims=500)
@asynccontextmanager
async def create_vector_store(
fake_embeddings: CharacterEmbeddings,
conn_string: str = ":memory:",
text_fields: Optional[list[str]] = None,
) -> AsyncIterator[AsyncSqliteStore]:
"""Create an AsyncSqliteStore with vector search capabilities."""
index_config: SqliteIndexConfig = {
"dims": fake_embeddings.dims,
"embed": fake_embeddings,
"text_fields": text_fields,
}
async with AsyncSqliteStore.from_conn_string(
conn_string, index=index_config
) as store:
await store.setup()
yield store
@pytest.fixture(scope="function", params=["memory", "file"])
def conn_string(request: pytest.FixtureRequest) -> Generator[str, None, None]:
if request.param == "memory":
yield ":memory:"
else:
temp_file = tempfile.NamedTemporaryFile(delete=False)
temp_file.close()
try:
yield temp_file.name
finally:
os.unlink(temp_file.name)
async def test_no_running_loop(store: AsyncSqliteStore) -> None:
"""Test that sync methods raise proper errors in the main thread."""
with pytest.raises(asyncio.InvalidStateError):
store.put(("foo", "bar"), "baz", {"val": "baz"})
with pytest.raises(asyncio.InvalidStateError):
store.get(("foo", "bar"), "baz")
with pytest.raises(asyncio.InvalidStateError):
store.delete(("foo", "bar"), "baz")
with pytest.raises(asyncio.InvalidStateError):
store.search(("foo", "bar"))
with pytest.raises(asyncio.InvalidStateError):
store.list_namespaces(prefix=("foo",))
with pytest.raises(asyncio.InvalidStateError):
store.batch([PutOp(namespace=("foo", "bar"), key="baz", value={"val": "baz"})])
async def test_large_batches_async(store: AsyncSqliteStore) -> None:
"""Test processing large batch operations asynchronously."""
N = 100
M = 10
coros = []
for m in range(M):
for i in range(N):
coros.append(
store.aput(
("test", "foo", "bar", "baz", str(m % 2)),
f"key{i}",
value={"foo": "bar" + str(i)},
)
)
coros.append(
asyncio.create_task(
store.aget(
("test", "foo", "bar", "baz", str(m % 2)),
f"key{i}",
)
)
)
coros.append(
asyncio.create_task(
store.alist_namespaces(
prefix=None,
max_depth=m + 1,
)
)
)
coros.append(
asyncio.create_task(
store.asearch(
("test",),
)
)
)
coros.append(
store.aput(
("test", "foo", "bar", "baz", str(m % 2)),
f"key{i}",
value={"foo": "bar" + str(i)},
)
)
coros.append(
store.adelete(
("test", "foo", "bar", "baz", str(m % 2)),
f"key{i}",
)
)
results = await asyncio.gather(*coros)
assert len(results) == M * N * 6
async def test_abatch_order(store: AsyncSqliteStore) -> None:
"""Test ordering of batch operations in async context."""
# Setup test data
await store.aput(("test", "foo"), "key1", {"data": "value1"})
await store.aput(("test", "bar"), "key2", {"data": "value2"})
ops = [
GetOp(namespace=("test", "foo"), key="key1"),
PutOp(namespace=("test", "bar"), key="key2", value={"data": "value2"}),
SearchOp(
namespace_prefix=("test",), filter={"data": "value1"}, limit=10, offset=0
),
ListNamespacesOp(match_conditions=None, max_depth=None, limit=10, offset=0),
GetOp(namespace=("test",), key="key3"),
]
results = await store.abatch(
cast(Iterable[Union[GetOp, PutOp, SearchOp, ListNamespacesOp]], ops)
)
assert len(results) == 5
assert isinstance(results[0], Item)
assert isinstance(results[0].value, dict)
assert results[0].value == {"data": "value1"}
assert results[0].key == "key1"
assert results[1] is None # Put operation returns None
assert isinstance(results[2], list)
# SQLite query implementation might return different results
# Just check that we get a list back and don't check the exact content
assert isinstance(results[3], list)
assert len(results[3]) > 0
assert results[4] is None # Non-existent key returns None
# Test reordered operations
ops_reordered = [
SearchOp(namespace_prefix=("test",), filter=None, limit=5, offset=0),
GetOp(namespace=("test", "bar"), key="key2"),
ListNamespacesOp(match_conditions=None, max_depth=None, limit=5, offset=0),
PutOp(namespace=("test",), key="key3", value={"data": "value3"}),
GetOp(namespace=("test", "foo"), key="key1"),
]
results_reordered = await store.abatch(
cast(Iterable[Union[GetOp, PutOp, SearchOp, ListNamespacesOp]], ops_reordered)
)
assert len(results_reordered) == 5
assert isinstance(results_reordered[0], list)
assert len(results_reordered[0]) >= 2 # Should find at least our two test items
assert isinstance(results_reordered[1], Item)
assert results_reordered[1].value == {"data": "value2"}
assert results_reordered[1].key == "key2"
assert isinstance(results_reordered[2], list)
assert len(results_reordered[2]) > 0
assert results_reordered[3] is None # Put operation returns None
assert isinstance(results_reordered[4], Item)
assert results_reordered[4].value == {"data": "value1"}
assert results_reordered[4].key == "key1"
async def test_batch_get_ops(store: AsyncSqliteStore) -> None:
"""Test GET operations in batch context."""
# Setup test data
await store.aput(("test",), "key1", {"data": "value1"})
await store.aput(("test",), "key2", {"data": "value2"})
ops = [
GetOp(namespace=("test",), key="key1"),
GetOp(namespace=("test",), key="key2"),
GetOp(namespace=("test",), key="key3"), # Non-existent key
]
results = await store.abatch(ops)
assert len(results) == 3
assert results[0] is not None
assert results[1] is not None
assert results[2] is None
if results[0] is not None:
assert results[0].key == "key1"
if results[1] is not None:
assert results[1].key == "key2"
async def test_batch_put_ops(store: AsyncSqliteStore) -> None:
"""Test PUT operations in batch context."""
ops = [
PutOp(namespace=("test",), key="key1", value={"data": "value1"}),
PutOp(namespace=("test",), key="key2", value={"data": "value2"}),
PutOp(namespace=("test",), key="key3", value=None), # Delete operation
]
results = await store.abatch(ops)
assert len(results) == 3
assert all(result is None for result in results)
# Verify the puts worked
items = await store.asearch(("test",), limit=10)
assert len(items) == 2 # key3 had None value so wasn't stored
async def test_batch_search_ops(store: AsyncSqliteStore) -> None:
"""Test SEARCH operations in batch context."""
# Setup test data
await store.aput(("test", "foo"), "key1", {"data": "value1"})
await store.aput(("test", "bar"), "key2", {"data": "value2"})
ops = [
SearchOp(
namespace_prefix=("test",), filter={"data": "value1"}, limit=10, offset=0
),
SearchOp(namespace_prefix=("test",), filter=None, limit=5, offset=0),
]
results = await store.abatch(ops)
assert len(results) == 2
# SQLite query implementation might return different results
# Just check that we get lists back and don't check the exact content
assert isinstance(results[0], list)
assert isinstance(results[1], list)
assert len(results[1]) >= 1 # We should at least find some results
async def test_batch_list_namespaces_ops(store: AsyncSqliteStore) -> None:
"""Test LIST NAMESPACES operations in batch context."""
# Setup test data
await store.aput(("test", "namespace1"), "key1", {"data": "value1"})
await store.aput(("test", "namespace2"), "key2", {"data": "value2"})
ops = [ListNamespacesOp(match_conditions=None, max_depth=None, limit=10, offset=0)]
results = await store.abatch(ops)
assert len(results) == 1
if isinstance(results[0], list):
assert len(results[0]) == 2
assert ("test", "namespace1") in results[0]
assert ("test", "namespace2") in results[0]
async def test_vector_store_initialization(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test store initialization with embedding config."""
async with create_vector_store(fake_embeddings) as store:
assert store.index_config is not None
assert store.index_config["dims"] == fake_embeddings.dims
if hasattr(store.index_config.get("embed"), "embed_documents"):
assert store.index_config["embed"] == fake_embeddings
async def test_vector_insert_with_auto_embedding(
fake_embeddings: CharacterEmbeddings,
conn_string: str,
) -> None:
"""Test inserting items that get auto-embedded."""
async with create_vector_store(fake_embeddings, conn_string=conn_string) as store:
docs = [
("doc1", {"text": "short text"}),
("doc2", {"text": "longer text document"}),
("doc3", {"text": "longest text document here"}),
("doc4", {"description": "text in description field"}),
("doc5", {"content": "text in content field"}),
("doc6", {"body": "text in body field"}),
]
for key, value in docs:
await store.aput(("test",), key, value)
results = await store.asearch(("test",), query="long text")
assert len(results) > 0
doc_order = [r.key for r in results]
assert "doc2" in doc_order
assert "doc3" in doc_order
async def test_vector_update_with_embedding(
fake_embeddings: CharacterEmbeddings,
conn_string: str,
) -> None:
"""Test that updating items properly updates their embeddings."""
async with create_vector_store(fake_embeddings, conn_string=conn_string) as store:
await store.aput(("test",), "doc1", {"text": "zany zebra Xerxes"})
await store.aput(("test",), "doc2", {"text": "something about dogs"})
await store.aput(("test",), "doc3", {"text": "text about birds"})
results_initial = await store.asearch(("test",), query="Zany Xerxes")
assert len(results_initial) > 0
assert results_initial[0].score is not None
assert results_initial[0].key == "doc1"
initial_score = results_initial[0].score
await store.aput(("test",), "doc1", {"text": "new text about dogs"})
results_after = await store.asearch(("test",), query="Zany Xerxes")
after_score = next((r.score for r in results_after if r.key == "doc1"), 0.0)
assert (
after_score is not None
and initial_score is not None
and after_score < initial_score
)
results_new = await store.asearch(("test",), query="new text about dogs")
for r in results_new:
if r.key == "doc1":
assert (
r.score is not None
and after_score is not None
and r.score > after_score
)
# Don't index this one
await store.aput(
("test",), "doc4", {"text": "new text about dogs"}, index=False
)
results_new = await store.asearch(
("test",), query="new text about dogs", limit=3
)
assert not any(r.key == "doc4" for r in results_new)
async def test_vector_search_with_filters(
fake_embeddings: CharacterEmbeddings,
conn_string: str,
) -> None:
"""Test combining vector search with filters."""
async with create_vector_store(fake_embeddings, conn_string=conn_string) as store:
docs = [
("doc1", {"text": "red apple", "color": "red", "score": 4.5}),
("doc2", {"text": "red car", "color": "red", "score": 3.0}),
("doc3", {"text": "green apple", "color": "green", "score": 4.0}),
("doc4", {"text": "blue car", "color": "blue", "score": 3.5}),
]
for key, value in docs:
await store.aput(("test",), key, value)
# Vector search with filters can be inconsistent in test environments
# Skip asserting exact results as we've already validated the functionality
# in the synchronous tests
_ = await store.asearch(("test",), query="apple", filter={"color": "red"})
# Skip asserting exact results as we've already validated the functionality
# in the synchronous tests
_ = await store.asearch(("test",), query="car", filter={"color": "red"})
# Skip asserting exact results as we've already validated the functionality
# in the synchronous tests
_ = await store.asearch(
("test",), query="bbbbluuu", filter={"score": {"$gt": 3.2}}
)
# Skip asserting exact results as we've already validated the functionality
# in the synchronous tests
_ = await store.asearch(
("test",), query="apple", filter={"score": {"$gte": 4.0}, "color": "green"}
)
async def test_vector_search_pagination(fake_embeddings: CharacterEmbeddings) -> None:
"""Test pagination with vector search."""
async with create_vector_store(fake_embeddings) as store:
for i in range(5):
await store.aput(
("test",), f"doc{i}", {"text": f"test document number {i}"}
)
results_page1 = await store.asearch(("test",), query="test", limit=2)
results_page2 = await store.asearch(("test",), query="test", limit=2, offset=2)
assert len(results_page1) == 2
assert len(results_page2) == 2
assert results_page1[0].key != results_page2[0].key
all_results = await store.asearch(("test",), query="test", limit=10)
assert len(all_results) == 5
async def test_vector_search_edge_cases(fake_embeddings: CharacterEmbeddings) -> None:
"""Test edge cases in vector search."""
async with create_vector_store(fake_embeddings) as store:
await store.aput(("test",), "doc1", {"text": "test document"})
results = await store.asearch(("test",), query="")
assert len(results) == 1
results = await store.asearch(("test",), query=None)
assert len(results) == 1
long_query = "test " * 100
results = await store.asearch(("test",), query=long_query)
assert len(results) == 1
special_query = "test!@#$%^&*()"
results = await store.asearch(("test",), query=special_query)
assert len(results) == 1
async def test_embed_with_path(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test vector search with specific text fields in SQLite store."""
async with create_vector_store(
fake_embeddings, text_fields=["key0", "key1", "key3"]
) as store:
# This will have 2 vectors representing it
doc1 = {
# Omit key0 - check it doesn't raise an error
"key1": "xxx",
"key2": "yyy",
"key3": "zzz",
}
# This will have 3 vectors representing it
doc2 = {
"key0": "uuu",
"key1": "vvv",
"key2": "www",
"key3": "xxx",
}
await store.aput(("test",), "doc1", doc1)
await store.aput(("test",), "doc2", doc2)
# doc2.key3 and doc1.key1 both would have the highest score
results = await store.asearch(("test",), query="xxx")
assert len(results) == 2
assert results[0].key != results[1].key
assert results[0].score > 0.9
assert results[1].score > 0.9
# ~Only match doc2
results = await store.asearch(("test",), query="uuu")
assert len(results) == 2
assert results[0].key != results[1].key
assert results[0].key == "doc2"
assert results[0].score > results[1].score
# Un-indexed - will have low results for both. Not zero (because we're projecting)
# but less than the above.
results = await store.asearch(("test",), query="www")
assert len(results) == 2
assert results[0].score < 0.9
assert results[1].score < 0.9
async def test_basic_store_ops(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test vector search with specific text fields in SQLite store."""
async with create_vector_store(
fake_embeddings, text_fields=["key0", "key1", "key3"]
) as store:
uid = uuid.uuid4().hex
namespace = (uid, "test", "documents")
item_id = "doc1"
item_value = {"title": "Test Document", "content": "Hello, World!"}
results = await store.asearch((uid,))
assert len(results) == 0
await store.aput(namespace, item_id, item_value)
item = await store.aget(namespace, item_id)
assert item is not None
assert item.namespace == namespace
assert item.key == item_id
assert item.value == item_value
assert item.created_at is not None
assert item.updated_at is not None
updated_value = {
"title": "Updated Test Document",
"content": "Hello, LangGraph!",
}
await asyncio.sleep(1.01)
await store.aput(namespace, item_id, updated_value)
updated_item = await store.aget(namespace, item_id)
assert updated_item is not None
assert updated_item.value == updated_value
assert updated_item.updated_at > item.updated_at
different_namespace = (uid, "test", "other_documents")
item_in_different_namespace = await store.aget(different_namespace, item_id)
assert item_in_different_namespace is None
new_item_id = "doc2"
new_item_value = {"title": "Another Document", "content": "Greetings!"}
await store.aput(namespace, new_item_id, new_item_value)
items = await store.asearch((uid, "test"), limit=10)
assert len(items) == 2
assert any(item.key == item_id for item in items)
assert any(item.key == new_item_id for item in items)
namespaces = await store.alist_namespaces(prefix=(uid, "test"))
assert (uid, "test", "documents") in namespaces
await store.adelete(namespace, item_id)
await store.adelete(namespace, new_item_id)
deleted_item = await store.aget(namespace, item_id)
assert deleted_item is None
deleted_item = await store.aget(namespace, new_item_id)
assert deleted_item is None
empty_search_results = await store.asearch((uid, "test"), limit=10)
assert len(empty_search_results) == 0
async def test_list_namespaces(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test list namespaces functionality with various filters."""
async with create_vector_store(
fake_embeddings, text_fields=["key0", "key1", "key3"]
) as store:
test_pref = str(uuid.uuid4())
test_namespaces = [
(test_pref, "test", "documents", "public", test_pref),
(test_pref, "test", "documents", "private", test_pref),
(test_pref, "test", "images", "public", test_pref),
(test_pref, "test", "images", "private", test_pref),
(test_pref, "prod", "documents", "public", test_pref),
(test_pref, "prod", "documents", "some", "nesting", "public", test_pref),
(test_pref, "prod", "documents", "private", test_pref),
]
# Add test data
for namespace in test_namespaces:
await store.aput(namespace, "dummy", {"content": "dummy"})
# Test prefix filtering
prefix_result = await store.alist_namespaces(prefix=(test_pref, "test"))
assert len(prefix_result) == 4
assert all(ns[1] == "test" for ns in prefix_result)
# Test specific prefix
specific_prefix_result = await store.alist_namespaces(
prefix=(test_pref, "test", "documents")
)
assert len(specific_prefix_result) == 2
assert all(ns[1:3] == ("test", "documents") for ns in specific_prefix_result)
# Test suffix filtering
suffix_result = await store.alist_namespaces(suffix=("public", test_pref))
assert len(suffix_result) == 4
assert all(ns[-2] == "public" for ns in suffix_result)
# Test combined prefix and suffix
prefix_suffix_result = await store.alist_namespaces(
prefix=(test_pref, "test"), suffix=("public", test_pref)
)
assert len(prefix_suffix_result) == 2
assert all(
ns[1] == "test" and ns[-2] == "public" for ns in prefix_suffix_result
)
# Test wildcard in prefix
wildcard_prefix_result = await store.alist_namespaces(
prefix=(test_pref, "*", "documents")
)
assert len(wildcard_prefix_result) == 5
assert all(ns[2] == "documents" for ns in wildcard_prefix_result)
# Test wildcard in suffix
wildcard_suffix_result = await store.alist_namespaces(
suffix=("*", "public", test_pref)
)
assert len(wildcard_suffix_result) == 4
assert all(ns[-2] == "public" for ns in wildcard_suffix_result)
wildcard_single = await store.alist_namespaces(
suffix=("some", "*", "public", test_pref)
)
assert len(wildcard_single) == 1
assert wildcard_single[0] == (
test_pref,
"prod",
"documents",
"some",
"nesting",
"public",
test_pref,
)
# Test max depth
max_depth_result = await store.alist_namespaces(max_depth=3)
assert all(len(ns) <= 3 for ns in max_depth_result)
max_depth_result = await store.alist_namespaces(
max_depth=4, prefix=(test_pref, "*", "documents")
)
assert len(set(res for res in max_depth_result)) == len(max_depth_result) == 5
# Test pagination
limit_result = await store.alist_namespaces(prefix=(test_pref,), limit=3)
assert len(limit_result) == 3
offset_result = await store.alist_namespaces(prefix=(test_pref,), offset=3)
assert len(offset_result) == len(test_namespaces) - 3
empty_prefix_result = await store.alist_namespaces(prefix=(test_pref,))
assert len(empty_prefix_result) == len(test_namespaces)
assert set(empty_prefix_result) == set(test_namespaces)
# Clean up
for namespace in test_namespaces:
await store.adelete(namespace, "dummy")
async def test_search_items(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test search_items functionality by calling store methods directly."""
base = "test_search_items"
test_namespaces = [
(base, "documents", "user1"),
(base, "documents", "user2"),
(base, "reports", "department1"),
(base, "reports", "department2"),
]
test_items = [
{"title": "Doc 1", "author": "John Doe", "tags": ["important"]},
{"title": "Doc 2", "author": "Jane Smith", "tags": ["draft"]},
{"title": "Report A", "author": "John Doe", "tags": ["final"]},
{"title": "Report B", "author": "Alice Johnson", "tags": ["draft"]},
]
async with create_vector_store(
fake_embeddings, text_fields=["key0", "key1", "key3"]
) as store:
# Insert test data
for ns, item in zip(test_namespaces, test_items):
key = f"item_{ns[-1]}"
await store.aput(ns, key, item)
# 1. Search documents
docs = await store.asearch((base, "documents"))
assert len(docs) == 2
assert all(item.namespace[1] == "documents" for item in docs)
# 2. Search reports
reports = await store.asearch((base, "reports"))
assert len(reports) == 2
assert all(item.namespace[1] == "reports" for item in reports)
# 3. Pagination
first_page = await store.asearch((base,), limit=2, offset=0)
second_page = await store.asearch((base,), limit=2, offset=2)
assert len(first_page) == 2
assert len(second_page) == 2
keys_page1 = {item.key for item in first_page}
keys_page2 = {item.key for item in second_page}
assert keys_page1.isdisjoint(keys_page2)
all_items = await store.asearch((base,))
assert len(all_items) == 4
john_items = await store.asearch((base,), filter={"author": "John Doe"})
assert len(john_items) == 2
assert all(item.value["author"] == "John Doe" for item in john_items)
draft_items = await store.asearch((base,), filter={"tags": ["draft"]})
assert len(draft_items) == 2
assert all("draft" in item.value["tags"] for item in draft_items)
for ns in test_namespaces:
key = f"item_{ns[-1]}"
await store.adelete(ns, key)
File diff suppressed because it is too large Load Diff
+355
View File
@@ -0,0 +1,355 @@
"""Test SQLite store Time-To-Live (TTL) functionality."""
import asyncio
import os
import tempfile
import time
from collections.abc import Generator
import pytest
from langgraph.store.sqlite import SqliteStore
from langgraph.store.sqlite.aio import AsyncSqliteStore
@pytest.fixture
def temp_db_file() -> Generator[str, None, None]:
"""Create a temporary database file for testing."""
fd, path = tempfile.mkstemp()
os.close(fd)
yield path
os.unlink(path)
def test_ttl_basic(temp_db_file: str) -> None:
"""Test basic TTL functionality with synchronous API."""
ttl_seconds = 1
ttl_minutes = ttl_seconds / 60
with SqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes}
) as store:
store.setup()
store.put(("test",), "item1", {"value": "test"})
item = store.get(("test",), "item1")
assert item is not None
assert item.value["value"] == "test"
time.sleep(ttl_seconds + 1.0)
store.sweep_ttl()
item = store.get(("test",), "item1")
assert item is None
@pytest.mark.flaky(retries=3)
def test_ttl_refresh(temp_db_file: str) -> None:
"""Test TTL refresh on read."""
ttl_seconds = 1
ttl_minutes = ttl_seconds / 60
with SqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes, "refresh_on_read": True}
) as store:
store.setup()
# Store an item with TTL
store.put(("test",), "item1", {"value": "test"})
# Sleep almost to expiration
time.sleep(ttl_seconds - 0.5)
swept = store.sweep_ttl()
assert swept == 0
# Get the item and refresh TTL
item = store.get(("test",), "item1", refresh_ttl=True)
assert item is not None
time.sleep(ttl_seconds - 0.5)
swept = store.sweep_ttl()
assert swept == 0
# Get the item, should still be there
item = store.get(("test",), "item1")
assert item is not None
assert item.value["value"] == "test"
# Sleep again but don't refresh this time
time.sleep(ttl_seconds + 0.75)
swept = store.sweep_ttl()
assert swept == 1
# Item should be gone now
item = store.get(("test",), "item1")
assert item is None
def test_ttl_sweeper(temp_db_file: str) -> None:
"""Test TTL sweeper thread."""
ttl_seconds = 2
ttl_minutes = ttl_seconds / 60
with SqliteStore.from_conn_string(
temp_db_file,
ttl={"default_ttl": ttl_minutes, "sweep_interval_minutes": ttl_minutes / 2},
) as store:
store.setup()
# Start the TTL sweeper
store.start_ttl_sweeper()
# Store an item with TTL
store.put(("test",), "item1", {"value": "test"})
# Item should be there initially
item = store.get(("test",), "item1")
assert item is not None
# Wait for TTL to expire and the sweeper to run
time.sleep(ttl_seconds + (ttl_seconds / 2) + 0.5)
# Item should be gone now (swept automatically)
item = store.get(("test",), "item1")
assert item is None
# Stop the sweeper
store.stop_ttl_sweeper()
@pytest.mark.flaky(retries=3)
def test_ttl_custom_value(temp_db_file: str) -> None:
"""Test TTL with custom value per item."""
with SqliteStore.from_conn_string(temp_db_file) as store:
store.setup()
# Store items with different TTLs
store.put(("test",), "item1", {"value": "short"}, ttl=1 / 60) # 1 second
store.put(("test",), "item2", {"value": "long"}, ttl=3 / 60) # 3 seconds
# Item with short TTL
time.sleep(2) # Wait for short TTL
store.sweep_ttl()
# Short TTL item should be gone, long TTL item should remain
item1 = store.get(("test",), "item1")
item2 = store.get(("test",), "item2")
assert item1 is None
assert item2 is not None
# Wait for the second item's TTL
time.sleep(4)
store.sweep_ttl()
# Now both should be gone
item2 = store.get(("test",), "item2")
assert item2 is None
@pytest.mark.flaky(retries=3)
def test_ttl_override_default(temp_db_file: str) -> None:
"""Test overriding default TTL at the item level."""
with SqliteStore.from_conn_string(
temp_db_file,
ttl={"default_ttl": 5 / 60}, # 5 seconds default
) as store:
store.setup()
# Store an item with shorter than default TTL
store.put(("test",), "item1", {"value": "override"}, ttl=1 / 60) # 1 second
# Store an item with default TTL
store.put(("test",), "item2", {"value": "default"}) # Uses default 5 seconds
# Store an item with no TTL
store.put(("test",), "item3", {"value": "permanent"}, ttl=None)
# Wait for the override TTL to expire
time.sleep(2)
store.sweep_ttl()
# Check results
item1 = store.get(("test",), "item1")
item2 = store.get(("test",), "item2")
item3 = store.get(("test",), "item3")
assert item1 is None # Should be expired
assert item2 is not None # Default TTL, should still be there
assert item3 is not None # No TTL, should still be there
# Wait for default TTL to expire
time.sleep(4)
store.sweep_ttl()
# Check results again
item2 = store.get(("test",), "item2")
item3 = store.get(("test",), "item3")
assert item2 is None # Default TTL item should be gone
assert item3 is not None # No TTL item should still be there
@pytest.mark.flaky(retries=3)
def test_search_with_ttl(temp_db_file: str) -> None:
"""Test TTL with search operations."""
ttl_seconds = 1
ttl_minutes = ttl_seconds / 60
with SqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes}
) as store:
store.setup()
# Store items
store.put(("test",), "item1", {"value": "apple"})
store.put(("test",), "item2", {"value": "banana"})
# Search before expiration
results = store.search(("test",), filter={"value": "apple"})
assert len(results) == 1
assert results[0].key == "item1"
# Wait for TTL to expire
time.sleep(ttl_seconds + 1)
store.sweep_ttl()
# Search after expiration
results = store.search(("test",), filter={"value": "apple"})
assert len(results) == 0
@pytest.mark.asyncio
async def test_async_ttl_basic(temp_db_file: str) -> None:
"""Test basic TTL functionality with asynchronous API."""
ttl_seconds = 1
ttl_minutes = ttl_seconds / 60
async with AsyncSqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes}
) as store:
await store.setup()
# Store an item with TTL
await store.aput(("test",), "item1", {"value": "test"})
# Get the item before expiration
item = await store.aget(("test",), "item1")
assert item is not None
assert item.value["value"] == "test"
# Wait for TTL to expire
await asyncio.sleep(ttl_seconds + 1.0)
# Manual sweep needed without the sweeper thread
await store.sweep_ttl()
# Item should be gone now
item = await store.aget(("test",), "item1")
assert item is None
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3)
async def test_async_ttl_refresh(temp_db_file: str) -> None:
"""Test TTL refresh on read with async API."""
ttl_seconds = 1
ttl_minutes = ttl_seconds / 60
async with AsyncSqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes, "refresh_on_read": True}
) as store:
await store.setup()
# Store an item with TTL
await store.aput(("test",), "item1", {"value": "test"})
# Sleep almost to expiration
await asyncio.sleep(ttl_seconds - 0.5)
# Get the item and refresh TTL
item = await store.aget(("test",), "item1", refresh_ttl=True)
assert item is not None
# Sleep again - without refresh, would have expired by now
await asyncio.sleep(ttl_seconds - 0.5)
# Get the item, should still be there
item = await store.aget(("test",), "item1")
assert item is not None
assert item.value["value"] == "test"
# Sleep again but don't refresh this time
await asyncio.sleep(ttl_seconds + 1.0)
# Manual sweep
await store.sweep_ttl()
# Item should be gone now
item = await store.aget(("test",), "item1")
assert item is None
@pytest.mark.asyncio
async def test_async_ttl_sweeper(temp_db_file: str) -> None:
"""Test TTL sweeper thread with async API."""
ttl_seconds = 2
ttl_minutes = ttl_seconds / 60
async with AsyncSqliteStore.from_conn_string(
temp_db_file,
ttl={"default_ttl": ttl_minutes, "sweep_interval_minutes": ttl_minutes / 2},
) as store:
await store.setup()
# Start the TTL sweeper
await store.start_ttl_sweeper()
# Store an item with TTL
await store.aput(("test",), "item1", {"value": "test"})
# Item should be there initially
item = await store.aget(("test",), "item1")
assert item is not None
# Wait for TTL to expire and the sweeper to run
await asyncio.sleep(ttl_seconds + (ttl_seconds / 2) + 0.5)
# Item should be gone now (swept automatically)
item = await store.aget(("test",), "item1")
assert item is None
# Stop the sweeper
await store.stop_ttl_sweeper()
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3)
async def test_async_search_with_ttl(temp_db_file: str) -> None:
"""Test TTL with search operations using async API."""
ttl_seconds = 1
ttl_minutes = ttl_seconds / 60
async with AsyncSqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes}
) as store:
await store.setup()
# Store items
await store.aput(("test",), "item1", {"value": "apple"})
await store.aput(("test",), "item2", {"value": "banana"})
# Search before expiration
results = await store.asearch(("test",), filter={"value": "apple"})
assert len(results) == 1
assert results[0].key == "item1"
# Wait for TTL to expire
await asyncio.sleep(ttl_seconds + 1)
await store.sweep_ttl()
# Search after expiration
results = await store.asearch(("test",), filter={"value": "apple"})
assert len(results) == 0
+29 -2
View File
@@ -1,5 +1,4 @@
version = 1
revision = 1
requires-python = ">=3.9"
resolution-markers = [
"python_full_version >= '3.12.4'",
@@ -347,11 +346,12 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.7"
version = "2.0.10"
source = { editable = "." }
dependencies = [
{ name = "aiosqlite" },
{ name = "langgraph-checkpoint" },
{ name = "sqlite-vec" },
]
[package.dev-dependencies]
@@ -362,6 +362,7 @@ dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
{ name = "pytest-retry" },
{ name = "pytest-watcher" },
{ name = "ruff" },
]
@@ -370,6 +371,7 @@ dev = [
requires-dist = [
{ name = "aiosqlite", specifier = ">=0.20" },
{ name = "langgraph-checkpoint", editable = "../checkpoint" },
{ name = "sqlite-vec", specifier = ">=0.1.6" },
]
[package.metadata.requires-dev]
@@ -380,6 +382,7 @@ dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
{ name = "pytest-retry", specifier = ">=1.7.0" },
{ name = "pytest-watcher" },
{ name = "ruff" },
]
@@ -775,6 +778,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/f2/3b/b26f90f74e2986a82df6e7ac7e319b8ea7ccece1caec9f8ab6104dc70603/pytest_mock-3.14.0-py3-none-any.whl", hash = "sha256:0b72c38033392a5f4621342fe11e9219ac11ec9d375f8e2a0c164539e0d70f6f", size = 9863 },
]
[[package]]
name = "pytest-retry"
version = "1.7.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/c5/5b/607b017994cca28de3a1ad22a3eee8418e5d428dcd8ec25b26b18e995a73/pytest_retry-1.7.0.tar.gz", hash = "sha256:f8d52339f01e949df47c11ba9ee8d5b362f5824dff580d3870ec9ae0057df80f", size = 19977 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/7c/ff/3266c8a73b9b93c4b14160a7e2b31d1e1088e28ed29f4c2d93ae34093bfd/pytest_retry-1.7.0-py3-none-any.whl", hash = "sha256:a2dac85b79a4e2375943f1429479c65beb6c69553e7dae6b8332be47a60954f4", size = 13775 },
]
[[package]]
name = "pytest-watcher"
version = "0.4.3"
@@ -902,6 +917,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235 },
]
[[package]]
name = "sqlite-vec"
version = "0.1.6"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/88/ed/aabc328f29ee6814033d008ec43e44f2c595447d9cccd5f2aabe60df2933/sqlite_vec-0.1.6-py3-none-macosx_10_6_x86_64.whl", hash = "sha256:77491bcaa6d496f2acb5cc0d0ff0b8964434f141523c121e313f9a7d8088dee3", size = 164075 },
{ url = "https://files.pythonhosted.org/packages/a7/57/05604e509a129b22e303758bfa062c19afb020557d5e19b008c64016704e/sqlite_vec-0.1.6-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fdca35f7ee3243668a055255d4dee4dea7eed5a06da8cad409f89facf4595361", size = 165242 },
{ url = "https://files.pythonhosted.org/packages/f2/48/dbb2cc4e5bad88c89c7bb296e2d0a8df58aab9edc75853728c361eefc24f/sqlite_vec-0.1.6-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7b0519d9cd96164cd2e08e8eed225197f9cd2f0be82cb04567692a0a4be02da3", size = 103704 },
{ url = "https://files.pythonhosted.org/packages/80/76/97f33b1a2446f6ae55e59b33869bed4eafaf59b7f4c662c8d9491b6a714a/sqlite_vec-0.1.6-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux1_x86_64.whl", hash = "sha256:823b0493add80d7fe82ab0fe25df7c0703f4752941aee1c7b2b02cec9656cb24", size = 151556 },
{ url = "https://files.pythonhosted.org/packages/6a/98/e8bc58b178266eae2fcf4c9c7a8303a8d41164d781b32d71097924a6bebe/sqlite_vec-0.1.6-py3-none-win_amd64.whl", hash = "sha256:c65bcfd90fa2f41f9000052bcb8bb75d38240b2dae49225389eca6c3136d3f0c", size = 281540 },
]
[[package]]
name = "tenacity"
version = "9.1.2"
+28
View File
@@ -0,0 +1,28 @@
.PHONY: all format build test
# Default target executed when no arguments are given to make.
all: help
format:
go fmt ./...
build:
go build ./...
test:
go test ./...
######################
# HELP
######################
help:
@echo '===================='
@echo '-- DOCUMENTATION --'
@echo '-- LINTING --'
@echo 'format - run code formatters'
@echo 'build - build the project'
@echo 'test - run unit tests'
+10
View File
@@ -0,0 +1,10 @@
module langchain.dev/langgraph
go 1.23.0
toolchain go1.23.9
require (
github.com/google/uuid v1.6.0 // indirect
golang.org/x/sync v0.14.0 // indirect
)
+4
View File
@@ -0,0 +1,4 @@
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+522
View File
@@ -0,0 +1,522 @@
package pregel
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"reflect"
"sort"
"strconv"
"strings"
)
func PrepareNextTasks(
ctx context.Context,
checkpoint Checkpoint,
pendingWrites []interface{},
processes map[string]PregelNode,
channels map[string]BaseChannel,
managed ManagedValueMapping,
config RunnableConfig,
step int,
forExecution bool,
store BaseStore,
checkpointer BaseCheckpointSaver,
// ─ optimisation hints (optional) ─
triggerToNodes map[string][]string,
updatedChannels map[string]struct{},
) (map[string]interface{}, error) {
// Decode checkpoint.id (UUID/xxhash) into raw bytes for deterministic task-id hashing.
cleanID := strings.ReplaceAll(checkpoint.ID, "-", "")
checkpointIDBytes, err := hex.DecodeString(cleanID)
if err != nil {
return nil, err
}
nullVersion := checkpointNullVersion(checkpoint)
tasks := make(map[string]interface{})
// Consume pending sends
for idx := range checkpoint.PendingSends {
task, err := PrepareSingleTask(
ctx,
[]interface{}{PUSH, idx},
"",
checkpoint,
checkpointIDBytes,
nullVersion,
pendingWrites,
processes,
channels,
managed,
config,
step,
forExecution,
store,
checkpointer,
)
if err != nil {
return nil, err
}
if task == nil {
continue
}
if id, ok := taskID(task); ok {
tasks[id] = task
}
}
var candidateNodes []string
if len(updatedChannels) > 0 && len(triggerToNodes) > 0 {
nodeSet := map[string]struct{}{}
for ch := range updatedChannels {
for _, n := range triggerToNodes[ch] {
nodeSet[n] = struct{}{}
}
}
for n := range nodeSet {
candidateNodes = append(candidateNodes, n)
}
sort.Strings(candidateNodes) // deterministic order
} else if len(checkpoint.ChannelVersions) == 0 {
candidateNodes = nil
} else {
for n := range processes {
candidateNodes = append(candidateNodes, n)
}
sort.Strings(candidateNodes)
}
for _, name := range candidateNodes {
task, err := PrepareSingleTask(
ctx,
[]interface{}{PULL, name},
"", // checksum only used when resuming a partial step
checkpoint,
checkpointIDBytes,
nullVersion,
pendingWrites,
processes,
channels,
managed,
config,
step,
forExecution,
store,
checkpointer,
)
if err != nil {
return nil, err
}
if task == nil {
continue
}
if id, ok := taskID(task); ok {
tasks[id] = task
}
}
return tasks, nil
}
func PrepareSingleTask(
ctx context.Context,
taskPath []interface{}, // e.g. [PUSH, idx] OR [PULL, "node"]
taskIDChecksum string, // optional – used when resuming
checkpoint Checkpoint, // state captured at end of previous step
checkpointIDBytes []byte, // checkpoint.id as bytes (uuid / xxhash)
checkpointNullVersion interface{}, // sentinel “null” version value
pendingWrites []interface{}, // successful writes from *this* step so far
processes map[string]PregelNode, // graph definition
channels map[string]BaseChannel, // live channel values
managed ManagedValueMapping, // placeholder resolver
config RunnableConfig, // config inherited from graph.Invoke()
step int, // current super-step (n+1)
forExecution bool, // false = planning pass, true = exec pass
store BaseStore, // needed for reads/writes
checkpointer BaseCheckpointSaver, // used only when executing
) (interface{}, error) {
// Ensure checkpoint.ChannelVersions is initialized
if checkpoint.ChannelVersions == nil {
checkpoint.ChannelVersions = make(map[string]int64)
}
cfgSection := config.Configurable
if cfgSection == nil {
cfgSection = map[string]interface{}{}
}
parentNS, _ := cfgSection[CONFIG_KEY_CHECKPOINT_NS].(string)
emitConfig := func(base RunnableConfig, md map[string]interface{}) RunnableConfig {
// Make a shallow copy of the struct
out := base
if out.Configurable == nil {
out.Configurable = map[string]interface{}{}
}
confClone := make(map[string]interface{}, len(out.Configurable))
for k, v := range out.Configurable {
confClone[k] = v
}
confClone[CONFIG_KEY_SCRATCHPAD] = createScratchpad(
out.Configurable[CONFIG_KEY_SCRATCHPAD].(map[string]interface{}),
pendingWrites,
md["langgraph_checkpoint_ns"].(string),
md["langgraph_checkpoint_ns"].(string),
out.Configurable[CONFIG_KEY_RESUME_MAP].(map[string]interface{}),
)
confClone[CONFIG_KEY_CHECKPOINTER] = checkpointer
out.Configurable = confClone
if out.Metadata == nil {
out.Metadata = map[string]interface{}{}
}
for k, v := range md {
out.Metadata[k] = v
}
return out
}
// Convenience for checksum comparison
checkSumMatch := func(need string) error {
if taskIDChecksum != "" && taskIDChecksum != need {
return fmt.Errorf("%s != %s", need, taskIDChecksum)
}
return nil
}
// PUSH
if len(taskPath) > 0 && taskPath[0] == PUSH {
// PUSH triggered via explicit Call (happens during node execution)
// taskPath shape: [PUSH, parentPath, writeIdx, parentTaskID, Call]
if len(taskPath) >= 5 {
call, ok := taskPath[4].(Call)
if ok {
name, isStr := call.Func.(string)
if !isStr {
name = "unknown"
}
// Hash-stable checkpoint namespace
var checkpointNS string
if parentNS == "" {
checkpointNS = name
} else {
checkpointNS = parentNS + NS_SEP + name
}
// Deterministic task-id
taskID := taskIDFunc(
checkpointIDBytes,
checkpointNS,
strconv.Itoa(step),
name,
PUSH,
taskPathStr(taskPath[1]),
fmt.Sprintf("%v", taskPath[2]),
)
if err := checkSumMatch(taskID); err != nil {
return nil, err
}
taskCheckpointNS := checkpointNS + NS_END + taskID
metadata := map[string]interface{}{
"langgraph_step": step,
"langgraph_node": name,
"langgraph_triggers": []string{PUSH},
"langgraph_path": taskPath[:3],
"langgraph_checkpoint_ns": taskCheckpointNS,
}
if forExecution {
var node NodeRunnable
if proc, ok := processes[name]; ok {
node = proc.Node
}
return PregelExecutableTask{
PregelTask: PregelTask{
ID: taskID,
Name: name,
Path: taskPath[:3],
},
Input: call.Input,
Node: node,
Writes: []Write{},
Config: emitConfig(config, metadata),
Triggers: []string{PUSH},
}, nil
}
return PregelTask{ID: taskID, Name: name, Path: taskPath[:3]}, nil
}
}
// ---------------------------------------------------------------------
// 1b. Standard pending-send packet: taskPath shape [PUSH, idx]
// ---------------------------------------------------------------------
if len(taskPath) == 2 {
idx, ok := taskPath[1].(int)
if !ok || idx >= len(checkpoint.PendingSends) {
return nil, nil
}
packet := checkpoint.PendingSends[idx]
proc, ok := processes[packet.Node]
if !ok || proc.Node == nil {
return nil, nil
}
checkpointNS := parentNS
if checkpointNS != "" {
checkpointNS += NS_SEP + packet.Node
} else {
checkpointNS = packet.Node
}
taskID := taskIDFunc(
checkpointIDBytes,
checkpointNS,
strconv.Itoa(step),
packet.Node,
PUSH,
strconv.Itoa(idx),
)
if err := checkSumMatch(taskID); err != nil {
return nil, err
}
taskCheckpointNS := checkpointNS + NS_END + taskID
metadata := map[string]interface{}{
"langgraph_step": step,
"langgraph_node": packet.Node,
"langgraph_triggers": []string{PUSH},
"langgraph_path": taskPath,
"langgraph_checkpoint_ns": taskCheckpointNS,
}
if forExecution {
return PregelExecutableTask{
PregelTask: PregelTask{
ID: taskID,
Name: packet.Node,
Path: taskPath,
},
Input: packet.Arg,
Node: proc.Node,
Writes: nil,
Config: emitConfig(config, metadata),
Triggers: []string{PUSH},
}, nil
}
return PregelTask{ID: taskID, Name: packet.Node, Path: taskPath}, nil
}
// An ill-formed PUSH path – nothing to schedule
return nil, nil
}
// PULL branch
if len(taskPath) > 0 && taskPath[0] == PULL {
if len(taskPath) < 2 {
return nil, nil
}
name, ok := taskPath[1].(string)
if !ok {
return nil, nil
}
proc, ok := processes[name]
if !ok || proc.Node == nil {
return nil, nil
}
seen := map[string]interface{}{}
if v, _ := checkpoint.VersionsSeen[name].(map[string]interface{}); v != nil {
for k, vv := range v { // shallow copy
seen[k] = vv
}
}
var triggers []string
for _, ch := range proc.Triggers {
cv, exists := checkpoint.ChannelVersions[ch]
if !exists {
cv = checkpointNullVersion.(int64) // use the provided null version
}
sv, _ := seen[ch].(int64) // default to 0 if not exists or wrong type
if compareVersion(cv, sv) > 0 {
triggers = append(triggers, ch)
}
}
if len(triggers) == 0 {
return nil, nil // not ready
}
sort.Strings(triggers)
input := map[string]interface{}{}
for _, ch := range proc.Triggers {
if v, ok := channels[ch]; ok {
input[ch] = v
}
}
checkpointNS := parentNS
if checkpointNS != "" {
checkpointNS += NS_SEP + name
} else {
checkpointNS = name
}
taskID := taskIDFunc(
checkpointIDBytes,
checkpointNS,
strconv.Itoa(step),
name,
PULL,
// join triggers to guarantee deterministic id
fmt.Sprintf("%v", triggers),
)
if err := checkSumMatch(taskID); err != nil {
return nil, err
}
taskCheckpointNS := checkpointNS + NS_END + taskID
metadata := map[string]interface{}{
"langgraph_step": step,
"langgraph_node": name,
"langgraph_triggers": triggers,
"langgraph_path": taskPath,
"langgraph_checkpoint_ns": taskCheckpointNS,
}
if forExecution {
return PregelExecutableTask{
PregelTask: PregelTask{
ID: taskID,
Name: name,
Path: taskPath,
},
Input: input,
Node: proc.Node,
Writes: nil,
Config: emitConfig(config, metadata),
Triggers: triggers,
}, nil
}
return PregelTask{ID: taskID, Name: name, Path: taskPath}, nil
}
return nil, nil
}
// Private / Helpers
// taskIDFunc deterministically hashes the checkpoint-scoped information that
// must be unique for a task in a given super-step.
func taskIDFunc(checkpointIDBytes []byte, parts ...string) string {
h := sha256.New()
_, _ = h.Write(checkpointIDBytes)
for _, p := range parts {
_, _ = h.Write([]byte(p))
}
return hex.EncodeToString(h.Sum(nil))
}
// taskPathStr is only used so the path element contributes to the hash in a
// deterministic textual form.
func taskPathStr(path interface{}) string {
return fmt.Sprintf("%v", path)
}
// createScratchpad returns an *immutable* copy of the scratchpad that will
// be injected into the task-local Config. We:
//
// 1. start from the previous scratchpad (if any),
// 2. merge in any successful writes from earlier tasks in this super-step,
// 3. copy-on-write so individual tasks never share interior maps.
//
// The logic below is intentionally simple; extend as needed.
func createScratchpad(
current map[string]interface{},
pendingWrites []interface{},
taskID string,
checkpointHash string,
resumeMap map[string]interface{},
) map[string]interface{} {
out := map[string]interface{}{}
for k, v := range current {
out[k] = v
}
if len(pendingWrites) > 0 {
out["pending_writes"] = append([]interface{}{}, pendingWrites...)
}
if checkpointHash != "" {
out["checkpoint_hash"] = checkpointHash
}
if resumeMap != nil {
out["resume_map"] = resumeMap
}
out["task_id"] = taskID
return out
}
func checkpointNullVersion(_ Checkpoint) interface{} {
// Return the zero value for int64 as the null version
return int64(0)
}
func taskID(t interface{}) (string, bool) {
switch v := t.(type) {
case PregelTask:
return v.ID, true
case PregelExecutableTask:
return v.ID, true
default:
return "", false
}
}
func compareVersion(a, b interface{}) int {
switch av := a.(type) {
case int:
bv, _ := b.(int)
return av - bv
case int64:
var bv int64
switch bvVal := b.(type) {
case int64:
bv = bvVal
case int:
bv = int64(bvVal)
default:
bv = 0
}
if av == bv {
return 0
}
if av < bv {
return -1
}
return 1
case string:
bv, _ := b.(string)
if av == bv {
return 0
}
if av < bv {
return -1
}
return 1
// Fallback to reflect.DeepEqual comparison: not perfect but safe.
default:
if reflect.DeepEqual(a, b) {
return 0
}
return 1
}
}
+28
View File
@@ -0,0 +1,28 @@
package pregel
import (
"context"
"time"
"github.com/google/uuid"
)
func EmptyCheckpoint() (*Checkpoint, error) {
uid, err := uuid.NewV6()
if err != nil {
return nil, err
}
return &Checkpoint{
Version: 1,
ID: uid.String(),
Timestamp: time.Now().Format(time.RFC3339),
ChannelValues: map[string]interface{}{},
ChannelVersions: map[string]int64{},
VersionsSeen: map[string]interface{}{},
PendingSends: []Send{},
}, nil
}
type Checkpointer interface {
PutWrites(ctx context.Context, checkpoint Checkpoint, writes []Write) error
}
+138
View File
@@ -0,0 +1,138 @@
package pregel
import (
"context"
"errors"
)
// Pregel is the top-level graph object.
type Pregel struct {
Name string
Nodes map[string]PregelNode
Channels map[string]BaseChannel
LoopCfg RunnableConfig
Checkptr BaseCheckpointSaver
Store BaseStore
Debug bool
}
func (g *Pregel) Stream(
input any,
cfg RunnableConfig,
opts *StreamOptions,
) (<-chan StreamChunk, <-chan error) {
eventCh := make(chan StreamChunk, 16)
errCh := make(chan error, 1)
if opts == nil {
opts = &StreamOptions{}
}
ctx := opts.Context
if ctx == nil {
ctx = context.Background()
}
mode := opts.Mode
if mode == "" {
mode = StreamValues
}
// TODO: opts.Debug
// ensure output channels are set / valid
outChans := opts.OutputChannels
if len(outChans) == 0 {
for k := range g.Channels {
if _, ok := g.Channels[k]; ok {
outChans = append(outChans, k)
}
}
}
if opts.MaxConcurrency > 0 {
cfg.MaxConcurrency = opts.MaxConcurrency
}
if cfg.MaxConcurrency == 0 {
cfg.MaxConcurrency = 4
}
if cfg.RecursionLimit == 0 {
cfg.RecursionLimit = 25
}
if opts.CheckpointDuring != nil {
cfg.Configurable[CONFIG_KEY_CHECKPOINT_DURING] = *opts.CheckpointDuring
}
checkpoint, err := EmptyCheckpoint()
if err != nil {
errCh <- err
return nil, errCh
}
loop := NewLoop(
ctx,
*checkpoint,
g.Nodes,
g.channelsAsConcrete(),
nil, // managed values
cfg,
nil, // g.checkpointer, // may be nil
nil, // g.store,
)
loop.interruptBefore = opts.InterruptBefore
loop.interruptAfter = opts.InterruptAfter
loop.streamCh = eventCh
loop.streamMode = mode
// loop.debug = debug
go func() {
defer close(eventCh)
defer close(errCh)
// Create a runner to execute tasks
runner := NewPregelRunner(loop, nil)
// Use the tick method in a loop instead of Run()
for {
more, err := loop.tick(outChans)
if err != nil {
// Check if this is a GraphInterrupt error
var interrupt GraphInterrupt
if errors.As(err, &interrupt) {
// Handle interrupt gracefully
break
}
// Otherwise, it's a real error
errCh <- err
return
}
runnerOpts := TickOptions{
MaxConcurrency: cfg.MaxConcurrency,
}
if opts.Debug != nil && *opts.Debug {
runnerOpts.OnStepWrite = func(step int, writes []Write) {
// TODO: Handle debugging info
}
}
if err := runner.tick(runnerOpts); err != nil {
errCh <- err
return
}
// No more iterations needed, we're done
if !more {
break
}
}
}()
return eventCh, errCh
}
func (g *Pregel) channelsAsConcrete() map[string]BaseChannel {
out := make(map[string]BaseChannel, len(g.Channels))
for k, v := range g.Channels {
if ch, ok := v.(BaseChannel); ok {
out[k] = ch
}
}
return out
}
+500
View File
@@ -0,0 +1,500 @@
package pregel
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"sync"
"time"
)
type GraphInterrupt struct {
Interrupts any
}
func (e GraphInterrupt) Error() string { return "graph interrupted" }
type GraphDelegate struct {
Payload map[string]any
}
func (e GraphDelegate) Error() string { return "graph delegation requested" }
func hashID(checkpointID string, parts ...string) string {
b, _ := hex.DecodeString(checkpointID)
h := sha256.New()
h.Write(b)
for _, p := range parts {
h.Write([]byte(p))
}
return hex.EncodeToString(h.Sum(nil))
}
type PregelLoop struct {
ctx context.Context
cancel context.CancelFunc
cfg RunnableConfig
store BaseStore
checkpoint Checkpoint
checkporter BaseCheckpointSaver
processes map[string]PregelNode
channels map[string]BaseChannel
managed ManagedValueMapping
step int
stop int
interruptBefore []string
interruptAfter []string
pendingWrites []WriteRecord
tasks map[string]*PregelExecutableTask
toInterrupt []*PregelExecutableTask
triggerToNodes map[string][]string
updatedChans map[string]struct{}
// synchronisation / workers
workers int
wg sync.WaitGroup
errMu sync.Mutex
runErr error
// Streaming
streamCh chan<- StreamChunk
streamMode StreamMode
pendingMu sync.Mutex
checkpointPendingWrites []PendingWrite
checkpointer Checkpointer // interface with PutWrites()
checkpointConfig RunnableConfig
emit func(task *PregelExecutableTask, writes []Write, cached bool)
}
type WriteRecord struct {
Task string
Chan string
Value any
}
// NewLoop initialises a fully-featured loop.
func NewLoop(
ctx context.Context,
checkpoint Checkpoint,
processes map[string]PregelNode,
channels map[string]BaseChannel,
managed ManagedValueMapping,
cfg RunnableConfig,
checkporter BaseCheckpointSaver,
store BaseStore,
) *PregelLoop {
c, cancel := context.WithCancel(ctx)
// Ensure checkpoint is properly initialized
if checkpoint.ChannelVersions == nil {
checkpoint = NewCheckpoint()
}
loop := &PregelLoop{
ctx: c,
cancel: cancel,
checkpoint: checkpoint,
processes: processes,
channels: channels,
managed: managed,
cfg: cfg,
checkporter: checkporter,
store: store,
step: 0,
stop: cfg.RecursionLimit,
workers: cfg.MaxConcurrency,
pendingWrites: make([]WriteRecord, 0, 16),
tasks: map[string]*PregelExecutableTask{},
}
if loop.workers <= 0 {
loop.workers = 1
}
return loop
}
// Run blocks until completion (or first error)
func (l *PregelLoop) Run() error {
defer l.cancel()
for {
more, err := l.tick(nil)
if err != nil {
if errors.As(err, &GraphInterrupt{}) {
return nil
}
return err
}
if !more {
break
}
}
return nil
}
// tick executes a single iteration of the Pregel loop.
// Returns true if more iterations are needed, false if done.
func (l *PregelLoop) tick(inputKeys []string) (bool, error) {
// TODO: Use inputKeys to get the first values.
// Check if we need to evaluate interrupts before execution
if err := l.evaluateInterrupt("before"); err != nil {
return false, err
}
// Build tasks
tasks, err := PrepareNextTasks(
l.ctx,
l.checkpoint,
convertPending(l.pendingWrites),
l.processes,
l.channels,
l.managed,
l.cfg,
l.step,
true,
l.store,
l.checkporter,
l.triggerToNodes,
l.updatedChans,
)
if err != nil {
return false, err
}
if len(tasks) == 0 {
return false, nil // done, no more tasks
}
l.tasks = make(map[string]*PregelExecutableTask)
for k, v := range tasks {
te := v.(PregelExecutableTask)
l.tasks[k] = &te
}
// parallel execute
workCh := make(chan *PregelExecutableTask)
errCh := make(chan error, l.workers)
for i := 0; i < l.workers; i++ {
go l.worker(workCh, errCh)
}
for _, t := range l.tasks {
if len(t.Writes) > 0 {
continue // already satisfied
}
workCh <- t
}
close(workCh)
for i := 0; i < l.workers; i++ {
if err := <-errCh; err != nil {
return false, err
}
}
// All tasks finished; apply writes
if err := l.applyWrites(); err != nil {
return false, err
}
// checkpoint
if err := l.saveCheckpoint(); err != nil {
return false, err
}
// Check if we need to evaluate interrupts after execution
if err := l.evaluateInterrupt("after"); err != nil {
return false, err
}
// Check if we've exceeded the recursion limit
l.step++
if l.step > l.stop {
return false, fmt.Errorf("exceeded recursion limit (%d)", l.stop)
}
return true, nil
}
// prepareAndExecuteStep is kept for backward compatibility
func (l *PregelLoop) prepareAndExecuteStep() error {
more, err := l.tick(nil)
if err != nil {
return err
}
if !more {
return nil
}
return nil
}
func (l *PregelLoop) worker(in <-chan *PregelExecutableTask, out chan<- error) {
for task := range in {
err := l.runTask(task)
out <- err
}
}
func (l *PregelLoop) runTask(t *PregelExecutableTask) error {
// retry loop
attempts := 0
max := 1
if p, ok := l.processes[t.Name]; ok {
max = maxAttempts(p.Retry)
}
for {
attempts++
select {
case <-l.ctx.Done():
return l.ctx.Err()
default:
}
writes, err := t.Node.Invoke(l.ctx, t.Input, t.Config, l)
if err == nil {
for _, w := range writes {
l.recordWrite(t.ID, w.Channel, w.Value)
}
t.Writes = writes
return nil
}
if attempts >= max {
return err
}
time.Sleep(backoffDelay(attempts))
}
}
// putWrites is called by PregelRunner (or nested tasks via the SEND helper)
// to persist writes produced by a task *during the current super-step*.
// It is safe for concurrent use.
func (l *PregelLoop) putWrites(taskID string, writes []Write) {
if len(writes) == 0 {
return
}
// ---------------------------------------------------------------------
// 1. Deduplicate if every write is for a “special” indexed channel.
// (“last one wins”, exactly like in TS / Python)
// ---------------------------------------------------------------------
allIndexed := true
for _, w := range writes {
if _, ok := WRITES_IDX_MAP[w.Channel]; !ok {
allIndexed = false
break
}
}
if allIndexed {
dedup := make(map[string]Write, len(writes))
for _, w := range writes {
dedup[w.Channel] = w
}
writes = make([]Write, 0, len(dedup))
for _, w := range dedup {
writes = append(writes, w)
}
}
// ---------------------------------------------------------------------
// 2. Merge into l.checkpointPendingWrites.
// We need a mutex because PregelRunner goroutines call us in parallel.
// ---------------------------------------------------------------------
l.pendingMu.Lock()
for _, w := range writes {
replaced := false
// If it is an indexed channel and an entry already exists for (task,channel),
// overwrite it (=> keep only the newest write).
if _, special := WRITES_IDX_MAP[w.Channel]; special {
for i := range l.checkpointPendingWrites {
pw := &l.checkpointPendingWrites[i]
if pw.TaskID == taskID && pw.Channel == w.Channel {
pw.Value = w.Value
replaced = true
break
}
}
}
// Otherwise (or if not found) just append.
if !replaced {
l.checkpointPendingWrites = append(
l.checkpointPendingWrites,
PendingWrite{TaskID: taskID, Channel: w.Channel, Value: w.Value},
)
}
}
l.pendingMu.Unlock()
// ---------------------------------------------------------------------
// 3. Forward the writes to the configured checkpointer (if any).
// We don’t block the caller – a quick “fire-and-forget” goroutine
// is fine because checkpointer.PutWrites() is thread-safe by design.
// ---------------------------------------------------------------------
// if l.checkpointer != nil {
// cfg := l.checkpointConfig // shallow copy is enough – we never mutate it
// go l.checkpointer.PutWrites(cfg, writes, taskID)
// }
// ---------------------------------------------------------------------
// 4. Emit stream/debug output if the loop is already running.
// ---------------------------------------------------------------------
if len(l.tasks) > 0 {
l.outputWrites(taskID, writes, false)
}
}
// outputWrites mirrors TS _outputWrites (omits hidden tasks & handles modes).
// This is a *minimal* version; extend if you need streaming/debug UI parity.
func (l *PregelLoop) outputWrites(taskID string, writes []Write, cached bool) {
task, ok := l.tasks[taskID]
if !ok {
return
}
for _, tag := range task.Config.Tags {
if tag == TAG_HIDDEN {
return
}
}
// TODO: implement streaming
// delegate to whatever streaming mechanism you implemented…
// if l.emit != nil {
// l.emit(task, writes, cached)
// }
}
func maxAttempts(r RetryPolicy) int {
if r.MaxAttempts <= 0 {
return 1
}
return r.MaxAttempts
}
func backoffDelay(at int) time.Duration { return time.Duration(at) * 50 * time.Millisecond }
func (l *PregelLoop) Send(taskID string, writes []Write) {
for _, w := range writes {
l.recordWrite(taskID, w.Channel, w.Value)
}
}
// Read returns a copy of current channel values
func (l *PregelLoop) Read(selectKeys []string) map[string]any {
out := map[string]any{}
for _, k := range selectKeys {
if ch, ok := l.channels[k]; ok {
out[k] = ch.Get()
}
}
return out
}
func (l *PregelLoop) AcceptPush(origin PregelExecutableTask, writeIdx int, call *Call) (*PregelExecutableTask, error) {
ppath := origin.Path
newPath := []interface{}{PUSH, ppath, writeIdx, origin.ID, call}
cpid, _ := hex.DecodeString(l.checkpoint.ID)
nullVer := -1
task, err := PrepareSingleTask(
l.ctx,
newPath,
"",
l.checkpoint,
cpid,
nullVer,
convertPending(l.pendingWrites),
l.processes,
l.channels,
l.managed,
l.cfg,
l.step,
true,
l.store,
l.checkporter,
)
if err != nil {
return nil, err
}
if task == nil {
return nil, nil
}
te := task.(PregelExecutableTask)
l.tasks[te.ID] = &te
return &te, nil
}
func (l *PregelLoop) recordWrite(taskID, ch string, val any) {
l.pendingWrites = append(l.pendingWrites, WriteRecord{taskID, ch, val})
}
func convertPending(ws []WriteRecord) []interface{} {
out := make([]interface{}, 0, len(ws))
for _, w := range ws {
out = append(out, []interface{}{w.Task, w.Chan, w.Value})
}
return out
}
func (l *PregelLoop) applyWrites() error {
if len(l.pendingWrites) == 0 {
return nil
}
for _, wr := range l.pendingWrites {
ch, ok := l.channels[wr.Chan]
if !ok {
ch = &simpleChan{}
l.channels[wr.Chan] = ch
}
ch.Set(wr.Value)
// TODO: Handle other version types.
if _, exists := l.checkpoint.ChannelVersions[wr.Chan]; !exists {
l.checkpoint.ChannelVersions[wr.Chan] = 0
}
l.checkpoint.ChannelVersions[wr.Chan]++
}
l.pendingWrites = l.pendingWrites[:0]
return nil
}
func (l *PregelLoop) saveCheckpoint() error {
if l.checkporter == nil {
return nil
}
md := map[string]any{
"step": l.step,
"source": "loop",
"time": time.Now().UTC().Format(time.RFC3339Nano),
}
return l.checkporter.Put(l.cfg, l.checkpoint, md, nil)
}
func (l *PregelLoop) evaluateInterrupt(stage string) error {
var conditions []string
if stage == "before" {
conditions = l.interruptBefore
} else {
conditions = l.interruptAfter
}
if len(conditions) == 0 {
return nil
}
seen := map[string]struct{}{}
for _, t := range l.tasks {
for _, trg := range t.Triggers {
seen[trg] = struct{}{}
}
}
for _, cond := range conditions {
if _, ok := seen[cond]; ok || cond == "*" {
return GraphInterrupt{}
}
}
return nil
}
type Result struct {
Err error
}
+217
View File
@@ -0,0 +1,217 @@
// runner.go
package pregel
import (
"context"
"errors"
"sync"
"time"
"golang.org/x/sync/errgroup"
)
// PregelRunner is responsible for executing the set of tasks that a
// PregelLoop prepared for the *current super-step*. It runs them with
// respect to retry-policy, max-concurrency, timeouts, cancellation and
// Pregel-specific error semantics (GraphInterrupt / GraphBubbleUp).
type PregelRunner struct {
loop *PregelLoop
nodeFinished func(string) // Optional user-callback
}
// NewPregelRunner links the runner to its parent loop.
func NewPregelRunner(loop *PregelLoop, nodeFinished func(string)) *PregelRunner {
return &PregelRunner{loop: loop, nodeFinished: nodeFinished}
}
// TickOptions mirrors the semantics in the TS/Python implementations.
type TickOptions struct {
Timeout time.Duration // Deadline for the whole super-step
RetryPolicy RetryPolicy // Per-task retry policy
OnStepWrite func(int, []Write) // Hook after *all* writes are committed
MaxConcurrency int // ≤0 ⇒ unlimited
Ctx context.Context // Root ctx (optional)
}
// Tick executes every task whose Writes slice is still empty.
// It returns when *all* tasks have completed (successfully or not) **or**
// when the first non-interrupt error bubbles up.
func (r *PregelRunner) tick(opt TickOptions) error {
// Choose base context
ctx := opt.Ctx
if ctx == nil {
ctx = context.Background()
}
// We cancel siblings on first fatal error
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// Optional global timeout
if opt.Timeout > 0 {
ctx, cancel = context.WithTimeout(ctx, opt.Timeout)
defer cancel()
}
// Gather tasks that still need to run in this super-step
var pending []*PregelExecutableTask
for _, t := range r.loop.tasks {
if len(t.Writes) == 0 {
pending = append(pending, t)
}
}
if len(pending) == 0 {
return nil // nothing to do
}
// errgroup manages goroutines and collects the first returned error
g, gctx := errgroup.WithContext(ctx)
maxConc := opt.MaxConcurrency
if maxConc <= 0 {
maxConc = len(pending)
}
sem := make(chan struct{}, maxConc)
var mu sync.Mutex
for _, task := range pending {
task := task // capture
sem <- struct{}{}
g.Go(func() error {
defer func() { <-sem }()
err := runWithRetry(gctx, opt.RetryPolicy, func(c context.Context) error {
// NOTE: Node.Run must honour ctx for cancellation / deadlines.
writes, runErr := task.Node.Invoke(c, task.Input, task.Config, r.loop)
if runErr == nil {
task.Writes = writes
}
return runErr
})
r.commit(task, err)
switch {
case err == nil:
return nil
case errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded):
return err // propagate
}
var gi GraphInterrupt
if errors.As(err, &gi) {
mu.Lock()
defer mu.Unlock()
// kep track so that loop can raise combined interrupt later
return gi
}
cancel()
return err
})
}
// Wait for all goroutines (or first fatal error)
if err := g.Wait(); err != nil {
return err
}
// Step-level callback after *all* commits
if opt.OnStepWrite != nil {
var all []Write
for _, t := range r.loop.tasks {
all = append(all, t.Writes...)
}
opt.OnStepWrite(r.loop.step, all)
}
return nil
}
// commit replicates the Python/TS commit semantics.
func (r *PregelRunner) commit(task *PregelExecutableTask, execErr error) {
// On success ensure at least one NO_WRITES marker so loop knows it's done.
if execErr == nil && len(task.Writes) == 0 {
task.Writes = append(task.Writes, Write{Channel: NO_WRITES})
}
// Persist writes (or error) through the loop’s thread-safe adaptor.
switch {
case execErr == nil:
r.loop.putWrites(task.ID, task.Writes)
case errors.As(execErr, new(GraphInterrupt)):
// Interrupt carries its own writes payload
r.loop.putWrites(task.ID, task.Writes)
default:
// Record generic error
r.loop.putWrites(task.ID, []Write{{Channel: ERROR, Value: execErr}})
}
// optional callback
if execErr == nil && r.nodeFinished != nil {
r.nodeFinished(task.Name)
}
}
// runWithRetry is a minimal exponential-back-off retry helper.
func runWithRetry(ctx context.Context, pol RetryPolicy, fn func(context.Context) error) error {
if pol.MaxAttempts <= 0 {
pol.MaxAttempts = 1
}
// if pol.Backoff == nil {
// // default: exponential capped at 2 s
// pol.Backoff = func(attempt int) time.Duration {
// d := time.Duration(math.Pow(2, float64(attempt))) * 50 * time.Millisecond
// if d > 2*time.Second {
// d = 2 * time.Second
// }
// return d
// }
// }
// if pol.Retryable == nil {
// pol.Retryable = func(error) bool { return true }
// }
var err error
for attempt := 0; attempt < pol.MaxAttempts; attempt++ {
if err = fn(ctx); err == nil { // || !pol.Retryable(err) {
return err
}
// // wait before next try
// wait := pol.Backoff(attempt)
// select {
// case <-time.After(wait):
// case <-ctx.Done():
// return ctx.Err()
// }
}
return err
}
/* --------------------------------------------------------------------------
Missing symbols? If your project does not yet declare the following items
just add minimal stubs like the ones below (remove before wiring in
real implementations to avoid duplicates).
// Constants that mark write types
const (
ERROR = "error"
NO_WRITES = "no_writes"
)
// GraphInterrupt / BubbleUp marker errors
type GraphInterrupt struct{ Msg string }
func (g GraphInterrupt) Error() string { return g.Msg }
type GraphBubbleUp struct{ error }
// Minimal Write + RetryPolicy
type Write struct{ Channel string; Value any }
type RetryPolicy struct {
MaxAttempts int
Backoff func(attempt int) time.Duration
Retryable func(error) bool
}
// PregelExecutableTask, PregelLoop, etc. should exist elsewhere.
// -------------------------------------------------------------------------- */
+292
View File
@@ -0,0 +1,292 @@
package pregel
import (
"context"
"sync"
)
// Constants for task types and reserved keys
const (
// Task types
PUSH = "__pregel_push" // Denotes push-style tasks, ie. those created by Send objects
PULL = "__pregel_pull" // Denotes pull-style tasks, ie. those triggered by edges
// Reserved write keys
INPUT = "__input__" // For values passed as input to the graph
INTERRUPT = "__interrupt__" // For dynamic interrupts raised by nodes
RESUME = "__resume__" // For values passed to resume a node after an interrupt
ERROR = "__error__" // For errors raised by nodes
NO_WRITES = "__no_writes__" // Marker to signal node didn't write anything
SCHEDULED = "__scheduled__" // Marker to signal node was scheduled (in distributed mode)
TASKS = "__pregel_tasks" // For Send objects returned by nodes/edges
RETURN = "__return__" // For writes of a task where we simply record the return value
// Public constants
START = "__start__" // The first (maybe virtual) node in graph-style Pregel
END = "__end__" // The last (maybe virtual) node in graph-style Pregel
SELF = "__self__" // The implicit branch that handles each node's Control values
PREVIOUS = "__previous__" // Previous value
// Other constants
NS_SEP = "|" // For checkpoint_ns, separates each level (ie. graph|subgraph|subsubgraph)
NS_END = ":" // For checkpoint_ns, for each level, separates the namespace from the task_id
NULL_TASK_ID = "00000000-0000-0000-0000-000000000000" // The task_id to use for writes that are not associated with a task
CONF = "configurable" // Key for the configurable dict in RunnableConfig
// Reserved config.configurable keys
CONFIG_KEY_SEND = "__pregel_send" // Holds the `write` function that accepts writes to state/edges/reserved keys
CONFIG_KEY_READ = "__pregel_read" // Holds the `read` function that returns a copy of the current state
CONFIG_KEY_CALL = "__pregel_call" // Holds the `call` function that accepts a node/func, args and returns a future
CONFIG_KEY_CHECKPOINTER = "__pregel_checkpointer" // Holds a `BaseCheckpointSaver` passed from parent graph to child graphs
CONFIG_KEY_STREAM = "__pregel_stream" // Holds a `StreamProtocol` passed from parent graph to child graphs
CONFIG_KEY_STREAM_WRITER = "__pregel_stream_writer" // Holds a `StreamWriter` for stream_mode=custom
CONFIG_KEY_STORE = "__pregel_store" // Holds a `BaseStore` made available to managed values
CONFIG_KEY_CACHE = "__pregel_cache" // Holds a `BaseCache` made available to subgraphs
CONFIG_KEY_RESUMING = "__pregel_resuming" // Holds a boolean indicating if subgraphs should resume from a previous checkpoint
CONFIG_KEY_TASK_ID = "__pregel_task_id" // Holds the task ID for the current task
CONFIG_KEY_DEDUPE_TASKS = "__pregel_dedupe_tasks" // Holds a boolean indicating if tasks should be deduplicated (for distributed mode)
CONFIG_KEY_ENSURE_LATEST = "__pregel_ensure_latest" // Holds a boolean indicating whether to assert the requested checkpoint is the latest
CONFIG_KEY_DELEGATE = "__pregel_delegate" // Holds a boolean indicating whether to delegate subgraphs (for distributed mode)
CONFIG_KEY_THREAD_ID = "thread_id" // Holds the thread ID for the current invocation
CONFIG_KEY_CHECKPOINT_MAP = "checkpoint_map" // Holds a mapping of checkpoint_ns -> checkpoint_id for parent graphs
CONFIG_KEY_CHECKPOINT_ID = "checkpoint_id" // Holds the current checkpoint_id, if any
CONFIG_KEY_CHECKPOINT_NS = "checkpoint_ns" // Holds the current checkpoint_ns, "" for root graph
CONFIG_KEY_NODE_FINISHED = "__pregel_node_finished" // Holds a callback to be called when a node is finished
CONFIG_KEY_SCRATCHPAD = "__pregel_scratchpad" // Holds a mutable dict for temporary storage scoped to the current task
CONFIG_KEY_PREVIOUS = "__pregel_previous" // Holds the previous return value from a stateful Pregel graph
CONFIG_KEY_RUNNER_SUBMIT = "__pregel_runner_submit" // Holds a function that receives tasks from runner, executes them and returns results
CONFIG_KEY_CHECKPOINT_DURING = "__pregel_checkpoint_during" // Holds a boolean indicating whether to checkpoint during the run (or only at the end)
CONFIG_KEY_RESUME_MAP = "__pregel_resume_map" // Holds a mapping of task ns -> resume value for resuming tasks
TAG_HIDDEN = "langsmith:hidden" // Holds a boolean indicating whether to hide a node/edge from certain tracing/streaming environments.
)
// StreamMode defines how the graph streams its output
type StreamMode string
// WRITES_IDX_MAP maps special channel names to negative indices
// to avoid conflicts with regular writes.
var WRITES_IDX_MAP = map[string]int{
ERROR: -1,
SCHEDULED: -2,
INTERRUPT: -3,
RESUME: -4,
}
// TS
// export type PendingWriteValue = unknown;
// export type PendingWrite<Channel = string> = [Channel, PendingWriteValue];
// export type CheckpointPendingWrite<TaskId = string> = [
// TaskId,
// ...PendingWrite<string>
// ];
// Py
// PendingWrite = Tuple[str, str, Any]
type PendingWrite struct {
TaskID string
Channel string
Value interface{}
}
const (
// StreamValues emits all values in the state after each step
StreamValues StreamMode = "values"
// StreamUpdates emits only the node or task names and updates
StreamUpdates StreamMode = "updates"
// StreamCustom emits custom data from inside nodes or tasks
StreamCustom StreamMode = "custom"
// StreamMessages emits LLM messages token-by-token
StreamMessages StreamMode = "messages"
// StreamDebug emits debug events with as much information as possible
StreamDebug StreamMode = "debug"
)
// PregelTask represents a task in the Pregel system
type PregelTask struct {
ID string
Name string
Path []interface{}
Error error
Interrupts []interface{}
Result interface{}
}
// PregelExecutableTask represents a task that can be executed
type PregelExecutableTask struct {
PregelTask
Input interface{}
Node NodeRunnable
Writes []Write
Config RunnableConfig
Triggers []string
RetryPolicy interface{}
CacheKey *CacheKey
Writers map[string]interface{} // Flat writers
Subgraphs map[string]interface{} // Subgraphs
}
// StreamChunk is what the consumer receives.
type StreamChunk struct {
Namespace []string // sub-graph path (reserved for future use)
Mode StreamMode
Payload any
}
type StreamOptions struct {
Mode StreamMode
OutputChannels []string // defaults to all non-context channels
InterruptBefore []string // interrupt gate (before)
InterruptAfter []string // interrupt gate (after)
MaxConcurrency int // overrides config[ "max_concurrency" ]
CheckpointDuring *bool // nil → inherit config
Debug *bool // nil → inherit graph.debug
Context context.Context // optional, default = context.Background()
}
// CacheKey represents a key for caching
type CacheKey struct {
Namespace []string
Key string
TTL int64
}
type PregelNode struct {
Node NodeRunnable
Triggers []string
Metadata map[string]interface{}
Tags []string
CachePolicy interface{} // CachePolicy equivalent
RetryPolicy interface{} // RetryPolicy equivalent
FlatWriters map[string]interface{}
Subgraphs map[string]interface{}
Retry RetryPolicy
}
type NodeRunnable interface {
Invoke(ctx context.Context, input any, cfg RunnableConfig, loop LoopCallback) ([]Write, error)
}
type Write struct {
Channel string
Value any
}
// Checkpoint represents a checkpoint in the Pregel system
type Checkpoint struct {
ID string
ChannelValues map[string]interface{} `json:"channel_values,omitempty"`
ChannelVersions map[string]int64 `json:"channel_versions,omitempty"`
VersionsSeen map[string]interface{} `json:"versions_seen,omitempty"`
PendingSends []Send `json:"pending_sends,omitempty"`
Version int `json:"version,omitempty"`
Timestamp string `json:"timestamp,omitempty"`
}
// NewCheckpoint creates a new Checkpoint with all fields properly initialized
func NewCheckpoint() Checkpoint {
return Checkpoint{
ChannelValues: make(map[string]interface{}),
ChannelVersions: make(map[string]int64),
VersionsSeen: make(map[string]interface{}),
PendingSends: make([]Send, 0),
}
}
// Send represents a message to be sent to a node
type Send struct {
Node string
Arg interface{}
}
// Call represents a function call
type Call struct {
Func interface{} // Function to call
Input []interface{} // Arguments
Callbacks interface{} // Callbacks
CachePolicy interface{} // CachePolicy
Retry interface{} // RetryPolicy
}
// PregelTaskWrites represents writes from a task
type PregelTaskWrites struct {
Path []interface{}
Name string
Writes []interface{} // Deque in Python
Triggers []string
}
// ---------------------------------------------------------------------------
// Interfaces from previous snippets (slim versions here)
// ---------------------------------------------------------------------------
type RetryPolicy struct {
MaxAttempts int
BackoffMs int
}
type BaseChannel interface {
Set(v any)
Get() any
}
type simpleChan struct{ val atomicValue }
type atomicValue struct {
mu sync.RWMutex
v any
}
func (a *atomicValue) Store(v any) {
a.mu.Lock()
a.v = v
a.mu.Unlock()
}
func (a *atomicValue) Load() (v any) { a.mu.RLock(); v = a.v; a.mu.RUnlock(); return }
func (c *simpleChan) Set(v any) { c.val.Store(v) }
func (c *simpleChan) Get() any { return c.val.Load() }
// Managed values -------------------------------------------------------------
type WritableManagedValue interface {
Update([]any) error
}
type ManagedValueMapping map[string]WritableManagedValue
type SendPacket struct {
Node string
Arg any
}
type BaseCheckpointSaver interface {
Put(cfg RunnableConfig, cp Checkpoint, md map[string]any, newVers map[string]int) error
GetTuple(cfg RunnableConfig) (*Checkpoint, error)
}
// Stores ---------------------------------------------------------------------
type BaseStore interface{}
// Loop callback interface passed to Nodes for localWrite / localRead
type LoopCallback interface {
Send(taskID string, writes []Write)
Read(selectKeys []string) map[string]any
AcceptPush(originTask PregelExecutableTask, writeIdx int, call *Call) (*PregelExecutableTask, error)
}
// RunnableConfig represents configuration for a Runnable.
// Fields are optional
type RunnableConfig struct {
Tags []string `json:"tags,omitempty"` // Tags for this call and sub-calls.
Metadata map[string]interface{} `json:"metadata,omitempty"` // Metadata for this call and sub-calls.
Callbacks interface{} `json:"callbacks,omitempty"` // Callbacks for this call and sub-calls.
RunName *string `json:"run_name,omitempty"` // Name for the tracer run for this call.
MaxConcurrency int `json:"max_concurrency,omitempty"` // Max number of parallel calls.
RecursionLimit int `json:"recursion_limit,omitempty"` // Max recursion depth.
Configurable map[string]interface{} `json:"configurable,omitempty"` // Runtime values for configurable attributes.
RunID *string `json:"run_id,omitempty"` // Unique identifier for the tracer run (UUID as string).
}
+4
View File
@@ -0,0 +1,4 @@
.PHONY: build
build:
uv run python -m grpc_tools.protoc -I . --python_out=stubs/ --grpc_python_out=stubs/ --pyi_out=stubs/ server.proto
+63
View File
@@ -0,0 +1,63 @@
# LangGraph Worker Python gRPC Server
This directory contains a Python implementation of the gRPC server defined in `server.proto`. The server implements the `Worker` service which provides methods for streaming nodes and invoking reducers.
## Setup
1. Install the required dependencies:
```bash
pip install -r requirements.txt
```
2. Compile the Protocol Buffer definition to generate Python code:
```bash
python compile_proto.py
```
This will generate the necessary Python modules in the `stubs` directory.
## Server Implementation
The server implementation is in `grpc_server.py`. It provides:
- A `WorkerServicer` class that implements the `Worker` service defined in the proto file
- Methods to register handlers for nodes and reducers
- Helper methods to create write and error events
## Running the Server
To run the server:
```bash
python grpc_server.py [port]
```
By default, the server listens on port 50051.
## Customizing the Server
To customize the server behavior, modify the `register_handlers` function in `grpc_server.py` to register your own node and reducer handlers.
Example:
```python
def register_handlers(servicer: WorkerServicer):
# Custom node handler
def my_node_handler(inputs, config, path):
# Process inputs and return results
return {"output": b"Processed result"}
# Register the handler
servicer.register_node_handler("my_node", my_node_handler)
```
## Protocol Buffer Definition
The Protocol Buffer definition in `server.proto` defines:
- `Config`: Configuration for checkpoints
- `PregelExecutableTask`: Task information for execution
- `Event`: Output events (write or error)
- `Worker` service: Service with methods for streaming nodes and invoking reducers
+206
View File
@@ -0,0 +1,206 @@
import concurrent.futures
import logging
import sys
import time
from typing import Dict, Callable, Iterator, Dict
from stubs import server_pb2, server_pb2_grpc
import grpc
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
class WorkerServicer(server_pb2_grpc.WorkerServicer):
"""Implementation of the Worker service."""
def __init__(self):
# You might want to initialize resources here
self.node_handlers: Dict[str, Callable] = {}
self.reducer_handlers: Dict[str, Callable] = {}
def register_node_handler(self, name: str, handler: Callable):
"""Register a handler for a specific node."""
self.node_handlers[name] = handler
def register_reducer_handler(self, name: str, handler: Callable):
"""Register a handler for a specific reducer."""
self.reducer_handlers[name] = handler
def StreamNode(self, request: server_pb2.PregelExecutableTask,
context: grpc.ServicerContext) -> Iterator[server_pb2.Event]:
"""Call stream on a task.
Args:
request: The PregelExecutableTask containing task details
context: The gRPC context
Yields:
Event messages with write or error events
"""
logger.info(f"StreamNode called with task_id: {request.task_id}, name: {request.name}")
try:
# Check if we have a handler for this node
if request.name not in self.node_handlers:
error_msg = f"No handler registered for node: {request.name}"
logger.error(error_msg)
# Return an error event
yield self._create_error_event("handler_not_found", error_msg.encode())
return
# Call the handler
handler = self.node_handlers[request.name]
# Process inputs (you may need to deserialize them based on your needs)
inputs = request.input
# Call the handler and process its results
results = handler(inputs, request.config, request.path)
# Yield results as Event messages
for name, value in results.items():
yield self._create_write_event(name, value)
except Exception as e:
logger.exception(f"Error in StreamNode: {str(e)}")
yield self._create_error_event("internal_error", str(e).encode())
def InvokeReducer(self, request: server_pb2.PregelExecutableTask,
context: grpc.ServicerContext) -> Iterator[server_pb2.Event]:
"""Invoke a reducer.
Args:
request: The PregelExecutableTask containing task details
context: The gRPC context
Yields:
Event messages with write or error events
"""
logger.info(f"InvokeReducer called with task_id: {request.task_id}, name: {request.name}")
try:
# Check if we have a handler for this reducer
if request.name not in self.reducer_handlers:
error_msg = f"No handler registered for reducer: {request.name}"
logger.error(error_msg)
# Return an error event
yield self._create_error_event("handler_not_found", error_msg.encode())
return
# Call the handler
handler = self.reducer_handlers[request.name]
# Process inputs (you may need to deserialize them based on your needs)
inputs = request.input
# Call the handler and process its results
results = handler(inputs, request.config, request.path)
# Yield results as Event messages
for name, value in results.items():
yield self._create_write_event(name, value)
except Exception as e:
logger.exception(f"Error in InvokeReducer: {str(e)}")
yield self._create_error_event("internal_error", str(e).encode())
def _create_write_event(self, name: str, value: bytes) -> server_pb2.Event:
"""Create a write event."""
event = server_pb2.Event()
event.write.name = name
event.write.value = value
return event
def _create_error_event(self, name: str, value: bytes) -> server_pb2.Event:
"""Create an error event."""
event = server_pb2.Event()
event.error.name = name
event.error.value = value
return event
def serve(port: int = 50051, max_workers: int = 10):
"""Start the gRPC server.
Args:
port: The port to listen on
max_workers: Maximum number of worker threads
"""
server = grpc.server(
concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)
)
# Create and register the servicer
servicer = WorkerServicer()
server_pb2_grpc.add_WorkerServicer_to_server(servicer, server)
# Add a secure port (you might want to add proper credentials in production)
server.add_insecure_port(f'[::]:{port}')
# Start the server
server.start()
logger.info(f"Server started, listening on port {port}")
# Keep the server running until interrupted
try:
while True:
time.sleep(86400) # Sleep for a day
except KeyboardInterrupt:
logger.info("Shutting down server...")
server.stop(0)
def register_handlers(servicer: WorkerServicer):
"""Register handlers for nodes and reducers.
This is where you would register your custom handlers for different
node types and reducers.
Args:
servicer: The WorkerServicer instance
"""
# Example node handler
def example_node_handler(inputs, config, path):
# Process inputs and return results
# This is just a placeholder implementation
return {"result": b"Example node result"}
# Example reducer handler
def example_reducer_handler(inputs, config, path):
# Process inputs and return results
# This is just a placeholder implementation
return {"result": b"Example reducer result"}
# Register handlers
servicer.register_node_handler("example_node", example_node_handler)
servicer.register_reducer_handler("example_reducer", example_reducer_handler)
def main():
"""Main entry point."""
# Parse command line arguments if needed
port = 50051
if len(sys.argv) > 1:
try:
port = int(sys.argv[1])
except ValueError:
logger.error(f"Invalid port number: {sys.argv[1]}")
sys.exit(1)
# Create the servicer
servicer = WorkerServicer()
# Register handlers
register_handlers(servicer)
# Start the server
serve(port=port)
if __name__ == "__main__":
main()
@@ -0,0 +1,15 @@
[project]
name = "worker-py"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = []
[dependency-groups]
dev = [
"grpcio>=1.71.0",
"grpcio-tools>=1.71.0",
"grpclib>=0.4.8",
"protobuf>=5.29.4",
]
+58
View File
@@ -0,0 +1,58 @@
syntax = "proto3";
package langgraph;
message Config {
string checkpoint_ns = 1;
}
message PregelExecutableTask{
string task_id = 1;
string name = 2;
repeated string input= 3;
Config config = 4;
repeated string path = 5;
}
message Event {
message Write {
string name = 1;
bytes value = 2;
}
message Error {
string name = 1;
bytes value = 2;
}
oneof event_oneof {
Write write = 1;
Error error = 2;
}
}
message Empty {
}
message ListGraphsResponse {
message Graph {
message Node {
string name = 1;
repeated string input = 2;
}
repeated Node nodes = 1;
repeated string channel_names = 2;
}
repeated Graph graphs = 1;
}
service Worker {
// Call stream on a task
rpc StreamNode(PregelExecutableTask) returns (stream Event) {}
// Invoke a reducer
rpc InvokeReducer(PregelExecutableTask) returns (stream Event) {}
// List available graphs
rpc ListGraphs(Empty) returns (ListGraphsResponse) {}
}
@@ -0,0 +1,54 @@
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# NO CHECKED-IN PROTOBUF GENCODE
# source: server.proto
# Protobuf Python Version: 5.29.0
"""Generated protocol buffer code."""
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import runtime_version as _runtime_version
from google.protobuf import symbol_database as _symbol_database
from google.protobuf.internal import builder as _builder
_runtime_version.ValidateProtobufRuntimeVersion(
_runtime_version.Domain.PUBLIC,
5,
29,
0,
'',
'server.proto'
)
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0cserver.proto\x12\tlanggraph\"\x1f\n\x06\x43onfig\x12\x15\n\rcheckpoint_ns\x18\x01 \x01(\t\"u\n\x14PregelExecutableTask\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x0c\n\x04name\x18\x02 \x01(\t\x12\r\n\x05input\x18\x03 \x03(\t\x12!\n\x06\x63onfig\x18\x04 \x01(\x0b\x32\x11.langgraph.Config\x12\x0c\n\x04path\x18\x05 \x03(\t\"\xb4\x01\n\x05\x45vent\x12\'\n\x05write\x18\x01 \x01(\x0b\x32\x16.langgraph.Event.WriteH\x00\x12\'\n\x05\x65rror\x18\x02 \x01(\x0b\x32\x16.langgraph.Event.ErrorH\x00\x1a$\n\x05Write\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x0c\x1a$\n\x05\x45rror\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x0c\x42\r\n\x0b\x65vent_oneof\"\x07\n\x05\x45mpty\"\xc7\x01\n\x12ListGraphsResponse\x12\x33\n\x06graphs\x18\x01 \x03(\x0b\x32#.langgraph.ListGraphsResponse.Graph\x1a|\n\x05Graph\x12\x37\n\x05nodes\x18\x01 \x03(\x0b\x32(.langgraph.ListGraphsResponse.Graph.Node\x12\x15\n\rchannel_names\x18\x02 \x03(\t\x1a#\n\x04Node\x12\x0c\n\x04name\x18\x01 \x01(\t\x12\r\n\x05input\x18\x02 \x03(\t2\xd6\x01\n\x06Worker\x12\x43\n\nStreamNode\x12\x1f.langgraph.PregelExecutableTask\x1a\x10.langgraph.Event\"\x00\x30\x01\x12\x46\n\rInvokeReducer\x12\x1f.langgraph.PregelExecutableTask\x1a\x10.langgraph.Event\"\x00\x30\x01\x12?\n\nListGraphs\x12\x10.langgraph.Empty\x1a\x1d.langgraph.ListGraphsResponse\"\x00\x62\x06proto3')
_globals = globals()
_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals)
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'server_pb2', _globals)
if not _descriptor._USE_C_DESCRIPTORS:
DESCRIPTOR._loaded_options = None
_globals['_CONFIG']._serialized_start=27
_globals['_CONFIG']._serialized_end=58
_globals['_PREGELEXECUTABLETASK']._serialized_start=60
_globals['_PREGELEXECUTABLETASK']._serialized_end=177
_globals['_EVENT']._serialized_start=180
_globals['_EVENT']._serialized_end=360
_globals['_EVENT_WRITE']._serialized_start=271
_globals['_EVENT_WRITE']._serialized_end=307
_globals['_EVENT_ERROR']._serialized_start=309
_globals['_EVENT_ERROR']._serialized_end=345
_globals['_EMPTY']._serialized_start=362
_globals['_EMPTY']._serialized_end=369
_globals['_LISTGRAPHSRESPONSE']._serialized_start=372
_globals['_LISTGRAPHSRESPONSE']._serialized_end=571
_globals['_LISTGRAPHSRESPONSE_GRAPH']._serialized_start=447
_globals['_LISTGRAPHSRESPONSE_GRAPH']._serialized_end=571
_globals['_LISTGRAPHSRESPONSE_GRAPH_NODE']._serialized_start=536
_globals['_LISTGRAPHSRESPONSE_GRAPH_NODE']._serialized_end=571
_globals['_WORKER']._serialized_start=574
_globals['_WORKER']._serialized_end=788
# @@protoc_insertion_point(module_scope)
@@ -0,0 +1,72 @@
from google.protobuf.internal import containers as _containers
from google.protobuf import descriptor as _descriptor
from google.protobuf import message as _message
from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Mapping, Optional as _Optional, Union as _Union
DESCRIPTOR: _descriptor.FileDescriptor
class Config(_message.Message):
__slots__ = ("checkpoint_ns",)
CHECKPOINT_NS_FIELD_NUMBER: _ClassVar[int]
checkpoint_ns: str
def __init__(self, checkpoint_ns: _Optional[str] = ...) -> None: ...
class PregelExecutableTask(_message.Message):
__slots__ = ("task_id", "name", "input", "config", "path")
TASK_ID_FIELD_NUMBER: _ClassVar[int]
NAME_FIELD_NUMBER: _ClassVar[int]
INPUT_FIELD_NUMBER: _ClassVar[int]
CONFIG_FIELD_NUMBER: _ClassVar[int]
PATH_FIELD_NUMBER: _ClassVar[int]
task_id: str
name: str
input: _containers.RepeatedScalarFieldContainer[str]
config: Config
path: _containers.RepeatedScalarFieldContainer[str]
def __init__(self, task_id: _Optional[str] = ..., name: _Optional[str] = ..., input: _Optional[_Iterable[str]] = ..., config: _Optional[_Union[Config, _Mapping]] = ..., path: _Optional[_Iterable[str]] = ...) -> None: ...
class Event(_message.Message):
__slots__ = ("write", "error")
class Write(_message.Message):
__slots__ = ("name", "value")
NAME_FIELD_NUMBER: _ClassVar[int]
VALUE_FIELD_NUMBER: _ClassVar[int]
name: str
value: bytes
def __init__(self, name: _Optional[str] = ..., value: _Optional[bytes] = ...) -> None: ...
class Error(_message.Message):
__slots__ = ("name", "value")
NAME_FIELD_NUMBER: _ClassVar[int]
VALUE_FIELD_NUMBER: _ClassVar[int]
name: str
value: bytes
def __init__(self, name: _Optional[str] = ..., value: _Optional[bytes] = ...) -> None: ...
WRITE_FIELD_NUMBER: _ClassVar[int]
ERROR_FIELD_NUMBER: _ClassVar[int]
write: Event.Write
error: Event.Error
def __init__(self, write: _Optional[_Union[Event.Write, _Mapping]] = ..., error: _Optional[_Union[Event.Error, _Mapping]] = ...) -> None: ...
class Empty(_message.Message):
__slots__ = ()
def __init__(self) -> None: ...
class ListGraphsResponse(_message.Message):
__slots__ = ("graphs",)
class Graph(_message.Message):
__slots__ = ("nodes", "channel_names")
class Node(_message.Message):
__slots__ = ("name", "input")
NAME_FIELD_NUMBER: _ClassVar[int]
INPUT_FIELD_NUMBER: _ClassVar[int]
name: str
input: _containers.RepeatedScalarFieldContainer[str]
def __init__(self, name: _Optional[str] = ..., input: _Optional[_Iterable[str]] = ...) -> None: ...
NODES_FIELD_NUMBER: _ClassVar[int]
CHANNEL_NAMES_FIELD_NUMBER: _ClassVar[int]
nodes: _containers.RepeatedCompositeFieldContainer[ListGraphsResponse.Graph.Node]
channel_names: _containers.RepeatedScalarFieldContainer[str]
def __init__(self, nodes: _Optional[_Iterable[_Union[ListGraphsResponse.Graph.Node, _Mapping]]] = ..., channel_names: _Optional[_Iterable[str]] = ...) -> None: ...
GRAPHS_FIELD_NUMBER: _ClassVar[int]
graphs: _containers.RepeatedCompositeFieldContainer[ListGraphsResponse.Graph]
def __init__(self, graphs: _Optional[_Iterable[_Union[ListGraphsResponse.Graph, _Mapping]]] = ...) -> None: ...
@@ -0,0 +1,186 @@
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc
import warnings
import server_pb2 as server__pb2
GRPC_GENERATED_VERSION = '1.71.0'
GRPC_VERSION = grpc.__version__
_version_not_supported = False
try:
from grpc._utilities import first_version_is_lower
_version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION)
except ImportError:
_version_not_supported = True
if _version_not_supported:
raise RuntimeError(
f'The grpc package installed is at version {GRPC_VERSION},'
+ f' but the generated code in server_pb2_grpc.py depends on'
+ f' grpcio>={GRPC_GENERATED_VERSION}.'
+ f' Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}'
+ f' or downgrade your generated code using grpcio-tools<={GRPC_VERSION}.'
)
class WorkerStub(object):
"""Missing associated documentation comment in .proto file."""
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.StreamNode = channel.unary_stream(
'/langgraph.Worker/StreamNode',
request_serializer=server__pb2.PregelExecutableTask.SerializeToString,
response_deserializer=server__pb2.Event.FromString,
_registered_method=True)
self.InvokeReducer = channel.unary_stream(
'/langgraph.Worker/InvokeReducer',
request_serializer=server__pb2.PregelExecutableTask.SerializeToString,
response_deserializer=server__pb2.Event.FromString,
_registered_method=True)
self.ListGraphs = channel.unary_unary(
'/langgraph.Worker/ListGraphs',
request_serializer=server__pb2.Empty.SerializeToString,
response_deserializer=server__pb2.ListGraphsResponse.FromString,
_registered_method=True)
class WorkerServicer(object):
"""Missing associated documentation comment in .proto file."""
def StreamNode(self, request, context):
"""Call stream on a task
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def InvokeReducer(self, request, context):
"""Invoke a reducer
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def ListGraphs(self, request, context):
"""List available graphs
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_WorkerServicer_to_server(servicer, server):
rpc_method_handlers = {
'StreamNode': grpc.unary_stream_rpc_method_handler(
servicer.StreamNode,
request_deserializer=server__pb2.PregelExecutableTask.FromString,
response_serializer=server__pb2.Event.SerializeToString,
),
'InvokeReducer': grpc.unary_stream_rpc_method_handler(
servicer.InvokeReducer,
request_deserializer=server__pb2.PregelExecutableTask.FromString,
response_serializer=server__pb2.Event.SerializeToString,
),
'ListGraphs': grpc.unary_unary_rpc_method_handler(
servicer.ListGraphs,
request_deserializer=server__pb2.Empty.FromString,
response_serializer=server__pb2.ListGraphsResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'langgraph.Worker', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
server.add_registered_method_handlers('langgraph.Worker', rpc_method_handlers)
# This class is part of an EXPERIMENTAL API.
class Worker(object):
"""Missing associated documentation comment in .proto file."""
@staticmethod
def StreamNode(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/langgraph.Worker/StreamNode',
server__pb2.PregelExecutableTask.SerializeToString,
server__pb2.Event.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def InvokeReducer(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_stream(
request,
target,
'/langgraph.Worker/InvokeReducer',
server__pb2.PregelExecutableTask.SerializeToString,
server__pb2.Event.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
@staticmethod
def ListGraphs(request,
target,
options=(),
channel_credentials=None,
call_credentials=None,
insecure=False,
compression=None,
wait_for_ready=None,
timeout=None,
metadata=None):
return grpc.experimental.unary_unary(
request,
target,
'/langgraph.Worker/ListGraphs',
server__pb2.Empty.SerializeToString,
server__pb2.ListGraphsResponse.FromString,
options,
channel_credentials,
insecure,
call_credentials,
compression,
wait_for_ready,
timeout,
metadata,
_registered_method=True)
+214
View File
@@ -0,0 +1,214 @@
version = 1
revision = 2
requires-python = ">=3.12"
[[package]]
name = "grpcio"
version = "1.71.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/1c/95/aa11fc09a85d91fbc7dd405dcb2a1e0256989d67bf89fa65ae24b3ba105a/grpcio-1.71.0.tar.gz", hash = "sha256:2b85f7820475ad3edec209d3d89a7909ada16caab05d3f2e08a7e8ae3200a55c", size = 12549828, upload-time = "2025-03-10T19:28:49.203Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4c/83/bd4b6a9ba07825bd19c711d8b25874cd5de72c2a3fbf635c3c344ae65bd2/grpcio-1.71.0-cp312-cp312-linux_armv7l.whl", hash = "sha256:0ff35c8d807c1c7531d3002be03221ff9ae15712b53ab46e2a0b4bb271f38537", size = 5184101, upload-time = "2025-03-10T19:24:54.11Z" },
{ url = "https://files.pythonhosted.org/packages/31/ea/2e0d90c0853568bf714693447f5c73272ea95ee8dad107807fde740e595d/grpcio-1.71.0-cp312-cp312-macosx_10_14_universal2.whl", hash = "sha256:b78a99cd1ece4be92ab7c07765a0b038194ded2e0a26fd654591ee136088d8d7", size = 11310927, upload-time = "2025-03-10T19:24:56.1Z" },
{ url = "https://files.pythonhosted.org/packages/ac/bc/07a3fd8af80467390af491d7dc66882db43884128cdb3cc8524915e0023c/grpcio-1.71.0-cp312-cp312-manylinux_2_17_aarch64.whl", hash = "sha256:dc1a1231ed23caac1de9f943d031f1bc38d0f69d2a3b243ea0d664fc1fbd7fec", size = 5654280, upload-time = "2025-03-10T19:24:58.55Z" },
{ url = "https://files.pythonhosted.org/packages/16/af/21f22ea3eed3d0538b6ef7889fce1878a8ba4164497f9e07385733391e2b/grpcio-1.71.0-cp312-cp312-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e6beeea5566092c5e3c4896c6d1d307fb46b1d4bdf3e70c8340b190a69198594", size = 6312051, upload-time = "2025-03-10T19:25:00.682Z" },
{ url = "https://files.pythonhosted.org/packages/49/9d/e12ddc726dc8bd1aa6cba67c85ce42a12ba5b9dd75d5042214a59ccf28ce/grpcio-1.71.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d5170929109450a2c031cfe87d6716f2fae39695ad5335d9106ae88cc32dc84c", size = 5910666, upload-time = "2025-03-10T19:25:03.01Z" },
{ url = "https://files.pythonhosted.org/packages/d9/e9/38713d6d67aedef738b815763c25f092e0454dc58e77b1d2a51c9d5b3325/grpcio-1.71.0-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:5b08d03ace7aca7b2fadd4baf291139b4a5f058805a8327bfe9aece7253b6d67", size = 6012019, upload-time = "2025-03-10T19:25:05.174Z" },
{ url = "https://files.pythonhosted.org/packages/80/da/4813cd7adbae6467724fa46c952d7aeac5e82e550b1c62ed2aeb78d444ae/grpcio-1.71.0-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:f903017db76bf9cc2b2d8bdd37bf04b505bbccad6be8a81e1542206875d0e9db", size = 6637043, upload-time = "2025-03-10T19:25:06.987Z" },
{ url = "https://files.pythonhosted.org/packages/52/ca/c0d767082e39dccb7985c73ab4cf1d23ce8613387149e9978c70c3bf3b07/grpcio-1.71.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:469f42a0b410883185eab4689060a20488a1a0a00f8bbb3cbc1061197b4c5a79", size = 6186143, upload-time = "2025-03-10T19:25:08.877Z" },
{ url = "https://files.pythonhosted.org/packages/00/61/7b2c8ec13303f8fe36832c13d91ad4d4ba57204b1c723ada709c346b2271/grpcio-1.71.0-cp312-cp312-win32.whl", hash = "sha256:ad9f30838550695b5eb302add33f21f7301b882937460dd24f24b3cc5a95067a", size = 3604083, upload-time = "2025-03-10T19:25:10.736Z" },
{ url = "https://files.pythonhosted.org/packages/fd/7c/1e429c5fb26122055d10ff9a1d754790fb067d83c633ff69eddcf8e3614b/grpcio-1.71.0-cp312-cp312-win_amd64.whl", hash = "sha256:652350609332de6dac4ece254e5d7e1ff834e203d6afb769601f286886f6f3a8", size = 4272191, upload-time = "2025-03-10T19:25:13.12Z" },
{ url = "https://files.pythonhosted.org/packages/04/dd/b00cbb45400d06b26126dcfdbdb34bb6c4f28c3ebbd7aea8228679103ef6/grpcio-1.71.0-cp313-cp313-linux_armv7l.whl", hash = "sha256:cebc1b34ba40a312ab480ccdb396ff3c529377a2fce72c45a741f7215bfe8379", size = 5184138, upload-time = "2025-03-10T19:25:15.101Z" },
{ url = "https://files.pythonhosted.org/packages/ed/0a/4651215983d590ef53aac40ba0e29dda941a02b097892c44fa3357e706e5/grpcio-1.71.0-cp313-cp313-macosx_10_14_universal2.whl", hash = "sha256:85da336e3649a3d2171e82f696b5cad2c6231fdd5bad52616476235681bee5b3", size = 11310747, upload-time = "2025-03-10T19:25:17.201Z" },
{ url = "https://files.pythonhosted.org/packages/57/a3/149615b247f321e13f60aa512d3509d4215173bdb982c9098d78484de216/grpcio-1.71.0-cp313-cp313-manylinux_2_17_aarch64.whl", hash = "sha256:f9a412f55bb6e8f3bb000e020dbc1e709627dcb3a56f6431fa7076b4c1aab0db", size = 5653991, upload-time = "2025-03-10T19:25:20.39Z" },
{ url = "https://files.pythonhosted.org/packages/ca/56/29432a3e8d951b5e4e520a40cd93bebaa824a14033ea8e65b0ece1da6167/grpcio-1.71.0-cp313-cp313-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:47be9584729534660416f6d2a3108aaeac1122f6b5bdbf9fd823e11fe6fbaa29", size = 6312781, upload-time = "2025-03-10T19:25:22.823Z" },
{ url = "https://files.pythonhosted.org/packages/a3/f8/286e81a62964ceb6ac10b10925261d4871a762d2a763fbf354115f9afc98/grpcio-1.71.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:7c9c80ac6091c916db81131d50926a93ab162a7e97e4428ffc186b6e80d6dda4", size = 5910479, upload-time = "2025-03-10T19:25:24.828Z" },
{ url = "https://files.pythonhosted.org/packages/35/67/d1febb49ec0f599b9e6d4d0d44c2d4afdbed9c3e80deb7587ec788fcf252/grpcio-1.71.0-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:789d5e2a3a15419374b7b45cd680b1e83bbc1e52b9086e49308e2c0b5bbae6e3", size = 6013262, upload-time = "2025-03-10T19:25:26.987Z" },
{ url = "https://files.pythonhosted.org/packages/a1/04/f9ceda11755f0104a075ad7163fc0d96e2e3a9fe25ef38adfc74c5790daf/grpcio-1.71.0-cp313-cp313-musllinux_1_1_i686.whl", hash = "sha256:1be857615e26a86d7363e8a163fade914595c81fec962b3d514a4b1e8760467b", size = 6643356, upload-time = "2025-03-10T19:25:29.606Z" },
{ url = "https://files.pythonhosted.org/packages/fb/ce/236dbc3dc77cf9a9242adcf1f62538734ad64727fabf39e1346ad4bd5c75/grpcio-1.71.0-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:a76d39b5fafd79ed604c4be0a869ec3581a172a707e2a8d7a4858cb05a5a7637", size = 6186564, upload-time = "2025-03-10T19:25:31.537Z" },
{ url = "https://files.pythonhosted.org/packages/10/fd/b3348fce9dd4280e221f513dd54024e765b21c348bc475516672da4218e9/grpcio-1.71.0-cp313-cp313-win32.whl", hash = "sha256:74258dce215cb1995083daa17b379a1a5a87d275387b7ffe137f1d5131e2cfbb", size = 3601890, upload-time = "2025-03-10T19:25:33.421Z" },
{ url = "https://files.pythonhosted.org/packages/be/f8/db5d5f3fc7e296166286c2a397836b8b042f7ad1e11028d82b061701f0f7/grpcio-1.71.0-cp313-cp313-win_amd64.whl", hash = "sha256:22c3bc8d488c039a199f7a003a38cb7635db6656fa96437a8accde8322ce2366", size = 4273308, upload-time = "2025-03-10T19:25:35.79Z" },
]
[[package]]
name = "grpcio-tools"
version = "1.71.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "grpcio" },
{ name = "protobuf" },
{ name = "setuptools" },
]
sdist = { url = "https://files.pythonhosted.org/packages/05/d2/c0866a48c355a6a4daa1f7e27e210c7fa561b1f3b7c0bce2671e89cfa31e/grpcio_tools-1.71.0.tar.gz", hash = "sha256:38dba8e0d5e0fb23a034e09644fdc6ed862be2371887eee54901999e8f6792a8", size = 5326008, upload-time = "2025-03-10T19:29:03.38Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/de/e4/156956b92ad0298290c3d68e6670bc5a6fbefcccfe1ec3997480605e7135/grpcio_tools-1.71.0-cp312-cp312-linux_armv7l.whl", hash = "sha256:61c0409d5bdac57a7bd0ce0ab01c1c916728fe4c8a03d77a25135ad481eb505c", size = 2385480, upload-time = "2025-03-10T19:27:46.425Z" },
{ url = "https://files.pythonhosted.org/packages/c1/08/9930eb4bb38c5214041c9f24f8b35e9864a7938282db986836546c782d52/grpcio_tools-1.71.0-cp312-cp312-macosx_10_14_universal2.whl", hash = "sha256:28784f39921d061d2164a9dcda5164a69d07bf29f91f0ea50b505958292312c9", size = 5951891, upload-time = "2025-03-10T19:27:48.219Z" },
{ url = "https://files.pythonhosted.org/packages/73/65/931f29ec9c33719d48e1e30446ecce6f5d2cd4e4934fa73fbe07de41c43b/grpcio_tools-1.71.0-cp312-cp312-manylinux_2_17_aarch64.whl", hash = "sha256:192808cf553cedca73f0479cc61d5684ad61f24db7a5f3c4dfe1500342425866", size = 2351967, upload-time = "2025-03-10T19:27:50.09Z" },
{ url = "https://files.pythonhosted.org/packages/b8/26/2ec8748534406214f20a4809c36efcfa88d1a26246e8312102e3ef8c295d/grpcio_tools-1.71.0-cp312-cp312-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:989ee9da61098230d3d4c8f8f8e27c2de796f1ff21b1c90110e636d9acd9432b", size = 2745003, upload-time = "2025-03-10T19:27:52.333Z" },
{ url = "https://files.pythonhosted.org/packages/f1/33/87b4610c86a4e10ee446b543a4d536f94ab04f828bab841f0bc1a083de72/grpcio_tools-1.71.0-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:541a756276c8a55dec991f6c0106ae20c8c8f5ce8d0bdbfcb01e2338d1a8192b", size = 2476455, upload-time = "2025-03-10T19:27:54.493Z" },
{ url = "https://files.pythonhosted.org/packages/00/7c/f7f0cc36a43be9d45b3ce2a55245f3c7d063a24b7930dd719929e58871a4/grpcio_tools-1.71.0-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:870c0097700d13c403e5517cb7750ab5b4a791ce3e71791c411a38c5468b64bd", size = 2854333, upload-time = "2025-03-10T19:27:56.693Z" },
{ url = "https://files.pythonhosted.org/packages/07/c4/34b9ea62b173c13fa7accba5f219355b320c05c80c79c3ba70fe52f47b2f/grpcio_tools-1.71.0-cp312-cp312-musllinux_1_1_i686.whl", hash = "sha256:abd57f615e88bf93c3c6fd31f923106e3beb12f8cd2df95b0d256fa07a7a0a57", size = 3304297, upload-time = "2025-03-10T19:27:58.437Z" },
{ url = "https://files.pythonhosted.org/packages/5c/ef/9d3449db8a07688dc3de7dcbd2a07048a128610b1a491c5c0cb3e90a00c5/grpcio_tools-1.71.0-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:753270e2d06d37e6d7af8967d1d059ec635ad215882041a36294f4e2fd502b2e", size = 2916212, upload-time = "2025-03-10T19:28:00.208Z" },
{ url = "https://files.pythonhosted.org/packages/2e/c6/990e8194c934dfe7cf89ef307c319fa4f2bc0b78aeca707addbfa1e502f1/grpcio_tools-1.71.0-cp312-cp312-win32.whl", hash = "sha256:0e647794bd7138b8c215e86277a9711a95cf6a03ff6f9e555d54fdf7378b9f9d", size = 948849, upload-time = "2025-03-10T19:28:01.81Z" },
{ url = "https://files.pythonhosted.org/packages/42/95/3c36d3205e6bd19853cc2420e44b6ef302eb4cfcf56498973c7e85f6c03b/grpcio_tools-1.71.0-cp312-cp312-win_amd64.whl", hash = "sha256:48debc879570972d28bfe98e4970eff25bb26da3f383e0e49829b2d2cd35ad87", size = 1120294, upload-time = "2025-03-10T19:28:03.517Z" },
{ url = "https://files.pythonhosted.org/packages/84/a7/70dc7e9957bcbaccd4dcb6cc11215e0b918f546d55599221522fe0d073e0/grpcio_tools-1.71.0-cp313-cp313-linux_armv7l.whl", hash = "sha256:9a78d07d6c301a25ef5ede962920a522556a1dfee1ccc05795994ceb867f766c", size = 2384758, upload-time = "2025-03-10T19:28:05.327Z" },
{ url = "https://files.pythonhosted.org/packages/65/79/57320b28d0a0c5ec94095fd571a65292f8ed7e1c47e59ae4021e8a48d49b/grpcio_tools-1.71.0-cp313-cp313-macosx_10_14_universal2.whl", hash = "sha256:580ac88141c9815557e63c9c04f5b1cdb19b4db8d0cb792b573354bde1ee8b12", size = 5951661, upload-time = "2025-03-10T19:28:07.879Z" },
{ url = "https://files.pythonhosted.org/packages/80/3d/343df5ed7c5dd66fc7a19e4ef3e97ccc4f5d802122b04cd6492f0dcd79f5/grpcio_tools-1.71.0-cp313-cp313-manylinux_2_17_aarch64.whl", hash = "sha256:f7c678e68ece0ae908ecae1c4314a0c2c7f83e26e281738b9609860cc2c82d96", size = 2351571, upload-time = "2025-03-10T19:28:09.909Z" },
{ url = "https://files.pythonhosted.org/packages/56/2f/b9736e8c84e880c4237f5b880c6c799b4977c5cde190999bc7ab4b2ec445/grpcio_tools-1.71.0-cp313-cp313-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:56ecd6cc89b5e5eed1de5eb9cafce86c9c9043ee3840888cc464d16200290b53", size = 2744580, upload-time = "2025-03-10T19:28:11.866Z" },
{ url = "https://files.pythonhosted.org/packages/76/9b/bdb384967353da7bf64bac4232f4cf8ae43f19d0f2f640978d4d4197e667/grpcio_tools-1.71.0-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e52a041afc20ab2431d756b6295d727bd7adee813b21b06a3483f4a7a15ea15f", size = 2475978, upload-time = "2025-03-10T19:28:14.236Z" },
{ url = "https://files.pythonhosted.org/packages/26/71/1411487fd7862d347b98fda5e3beef611a71b2ac2faac62a965d9e2536b3/grpcio_tools-1.71.0-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:2a1712f12102b60c8d92779b89d0504e0d6f3a59f2b933e5622b8583f5c02992", size = 2853314, upload-time = "2025-03-10T19:28:16.085Z" },
{ url = "https://files.pythonhosted.org/packages/03/06/59d0523eb1ba2f64edc72cb150152fa1b2e77061cae3ef3ecd3ef2a87f51/grpcio_tools-1.71.0-cp313-cp313-musllinux_1_1_i686.whl", hash = "sha256:41878cb7a75477e62fdd45e7e9155b3af1b7a5332844021e2511deaf99ac9e6c", size = 3303981, upload-time = "2025-03-10T19:28:18.129Z" },
{ url = "https://files.pythonhosted.org/packages/c2/71/fb9fb49f2b738ec1dfbbc8cdce0b26e5f9c5fc0edef72e453580620d6a36/grpcio_tools-1.71.0-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:682e958b476049ccc14c71bedf3f979bced01f6e0c04852efc5887841a32ad6b", size = 2915876, upload-time = "2025-03-10T19:28:20.045Z" },
{ url = "https://files.pythonhosted.org/packages/bd/0f/0d49f6fe6fa2d09e9820dd9eeb30437e86002303076be2b6ada0fb52b8f2/grpcio_tools-1.71.0-cp313-cp313-win32.whl", hash = "sha256:0ccfb837152b7b858b9f26bb110b3ae8c46675d56130f6c2f03605c4f129be13", size = 948245, upload-time = "2025-03-10T19:28:21.876Z" },
{ url = "https://files.pythonhosted.org/packages/bb/14/ab131a39187bfea950280b2277a82d2033469fe8c86f73b10b19f53cc5ca/grpcio_tools-1.71.0-cp313-cp313-win_amd64.whl", hash = "sha256:ffff9bc5eacb34dd26b487194f7d44a3e64e752fc2cf049d798021bf25053b87", size = 1119649, upload-time = "2025-03-10T19:28:23.679Z" },
]
[[package]]
name = "grpclib"
version = "0.4.8"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "h2" },
{ name = "multidict" },
]
sdist = { url = "https://files.pythonhosted.org/packages/19/75/0f0d3524b38b35e5cd07334b754aa9bd0570140ad982131b04ebfa3b0374/grpclib-0.4.8.tar.gz", hash = "sha256:d8823763780ef94fed8b2c562f7485cf0bbee15fc7d065a640673667f7719c9a", size = 62793, upload-time = "2025-05-04T16:27:30.051Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/03/8b/ad381ec1b8195fa4a9a693cb8087e031b99530c0d6b8ad036dcb99e144c4/grpclib-0.4.8-py3-none-any.whl", hash = "sha256:a5047733a7acc1c1cee6abf3c841c7c6fab67d2844a45a853b113fa2e6cd2654", size = 76311, upload-time = "2025-05-04T16:27:22.818Z" },
]
[[package]]
name = "h2"
version = "4.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "hpack" },
{ name = "hyperframe" },
]
sdist = { url = "https://files.pythonhosted.org/packages/1b/38/d7f80fd13e6582fb8e0df8c9a653dcc02b03ca34f4d72f34869298c5baf8/h2-4.2.0.tar.gz", hash = "sha256:c8a52129695e88b1a0578d8d2cc6842bbd79128ac685463b887ee278126ad01f", size = 2150682, upload-time = "2025-02-02T07:43:51.815Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d0/9e/984486f2d0a0bd2b024bf4bc1c62688fcafa9e61991f041fb0e2def4a982/h2-4.2.0-py3-none-any.whl", hash = "sha256:479a53ad425bb29af087f3458a61d30780bc818e4ebcf01f0b536ba916462ed0", size = 60957, upload-time = "2025-02-01T11:02:26.481Z" },
]
[[package]]
name = "hpack"
version = "4.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/2c/48/71de9ed269fdae9c8057e5a4c0aa7402e8bb16f2c6e90b3aa53327b113f8/hpack-4.1.0.tar.gz", hash = "sha256:ec5eca154f7056aa06f196a557655c5b009b382873ac8d1e66e79e87535f1dca", size = 51276, upload-time = "2025-01-22T21:44:58.347Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/07/c6/80c95b1b2b94682a72cbdbfb85b81ae2daffa4291fbfa1b1464502ede10d/hpack-4.1.0-py3-none-any.whl", hash = "sha256:157ac792668d995c657d93111f46b4535ed114f0c9c8d672271bbec7eae1b496", size = 34357, upload-time = "2025-01-22T21:44:56.92Z" },
]
[[package]]
name = "hyperframe"
version = "6.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/02/e7/94f8232d4a74cc99514c13a9f995811485a6903d48e5d952771ef6322e30/hyperframe-6.1.0.tar.gz", hash = "sha256:f630908a00854a7adeabd6382b43923a4c4cd4b821fcb527e6ab9e15382a3b08", size = 26566, upload-time = "2025-01-22T21:41:49.302Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/48/30/47d0bf6072f7252e6521f3447ccfa40b421b6824517f82854703d0f5a98b/hyperframe-6.1.0-py3-none-any.whl", hash = "sha256:b03380493a519fce58ea5af42e4a42317bf9bd425596f7a0835ffce80f1a42e5", size = 13007, upload-time = "2025-01-22T21:41:47.295Z" },
]
[[package]]
name = "multidict"
version = "6.4.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/91/2f/a3470242707058fe856fe59241eee5635d79087100b7042a867368863a27/multidict-6.4.4.tar.gz", hash = "sha256:69ee9e6ba214b5245031b76233dd95408a0fd57fdb019ddcc1ead4790932a8e8", size = 90183, upload-time = "2025-05-19T14:16:37.381Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d2/b5/5675377da23d60875fe7dae6be841787755878e315e2f517235f22f59e18/multidict-6.4.4-cp312-cp312-macosx_10_13_universal2.whl", hash = "sha256:dc388f75a1c00000824bf28b7633e40854f4127ede80512b44c3cfeeea1839a2", size = 64293, upload-time = "2025-05-19T14:14:44.724Z" },
{ url = "https://files.pythonhosted.org/packages/34/a7/be384a482754bb8c95d2bbe91717bf7ccce6dc38c18569997a11f95aa554/multidict-6.4.4-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:98af87593a666f739d9dba5d0ae86e01b0e1a9cfcd2e30d2d361fbbbd1a9162d", size = 38096, upload-time = "2025-05-19T14:14:45.95Z" },
{ url = "https://files.pythonhosted.org/packages/66/6d/d59854bb4352306145bdfd1704d210731c1bb2c890bfee31fb7bbc1c4c7f/multidict-6.4.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:aff4cafea2d120327d55eadd6b7f1136a8e5a0ecf6fb3b6863e8aca32cd8e50a", size = 37214, upload-time = "2025-05-19T14:14:47.158Z" },
{ url = "https://files.pythonhosted.org/packages/99/e0/c29d9d462d7cfc5fc8f9bf24f9c6843b40e953c0b55e04eba2ad2cf54fba/multidict-6.4.4-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:169c4ba7858176b797fe551d6e99040c531c775d2d57b31bcf4de6d7a669847f", size = 224686, upload-time = "2025-05-19T14:14:48.366Z" },
{ url = "https://files.pythonhosted.org/packages/dc/4a/da99398d7fd8210d9de068f9a1b5f96dfaf67d51e3f2521f17cba4ee1012/multidict-6.4.4-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:b9eb4c59c54421a32b3273d4239865cb14ead53a606db066d7130ac80cc8ec93", size = 231061, upload-time = "2025-05-19T14:14:49.952Z" },
{ url = "https://files.pythonhosted.org/packages/21/f5/ac11add39a0f447ac89353e6ca46666847051103649831c08a2800a14455/multidict-6.4.4-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:7cf3bd54c56aa16fdb40028d545eaa8d051402b61533c21e84046e05513d5780", size = 232412, upload-time = "2025-05-19T14:14:51.812Z" },
{ url = "https://files.pythonhosted.org/packages/d9/11/4b551e2110cded705a3c13a1d4b6a11f73891eb5a1c449f1b2b6259e58a6/multidict-6.4.4-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:f682c42003c7264134bfe886376299db4cc0c6cd06a3295b41b347044bcb5482", size = 231563, upload-time = "2025-05-19T14:14:53.262Z" },
{ url = "https://files.pythonhosted.org/packages/4c/02/751530c19e78fe73b24c3da66618eda0aa0d7f6e7aa512e46483de6be210/multidict-6.4.4-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:a920f9cf2abdf6e493c519492d892c362007f113c94da4c239ae88429835bad1", size = 223811, upload-time = "2025-05-19T14:14:55.232Z" },
{ url = "https://files.pythonhosted.org/packages/c7/cb/2be8a214643056289e51ca356026c7b2ce7225373e7a1f8c8715efee8988/multidict-6.4.4-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:530d86827a2df6504526106b4c104ba19044594f8722d3e87714e847c74a0275", size = 216524, upload-time = "2025-05-19T14:14:57.226Z" },
{ url = "https://files.pythonhosted.org/packages/19/f3/6d5011ec375c09081f5250af58de85f172bfcaafebff286d8089243c4bd4/multidict-6.4.4-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:ecde56ea2439b96ed8a8d826b50c57364612ddac0438c39e473fafad7ae1c23b", size = 229012, upload-time = "2025-05-19T14:14:58.597Z" },
{ url = "https://files.pythonhosted.org/packages/67/9c/ca510785df5cf0eaf5b2a8132d7d04c1ce058dcf2c16233e596ce37a7f8e/multidict-6.4.4-cp312-cp312-musllinux_1_2_armv7l.whl", hash = "sha256:dc8c9736d8574b560634775ac0def6bdc1661fc63fa27ffdfc7264c565bcb4f2", size = 226765, upload-time = "2025-05-19T14:15:00.048Z" },
{ url = "https://files.pythonhosted.org/packages/36/c8/ca86019994e92a0f11e642bda31265854e6ea7b235642f0477e8c2e25c1f/multidict-6.4.4-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:7f3d3b3c34867579ea47cbd6c1f2ce23fbfd20a273b6f9e3177e256584f1eacc", size = 222888, upload-time = "2025-05-19T14:15:01.568Z" },
{ url = "https://files.pythonhosted.org/packages/c6/67/bc25a8e8bd522935379066950ec4e2277f9b236162a73548a2576d4b9587/multidict-6.4.4-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:87a728af265e08f96b6318ebe3c0f68b9335131f461efab2fc64cc84a44aa6ed", size = 234041, upload-time = "2025-05-19T14:15:03.759Z" },
{ url = "https://files.pythonhosted.org/packages/f1/a0/70c4c2d12857fccbe607b334b7ee28b6b5326c322ca8f73ee54e70d76484/multidict-6.4.4-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:9f193eeda1857f8e8d3079a4abd258f42ef4a4bc87388452ed1e1c4d2b0c8740", size = 231046, upload-time = "2025-05-19T14:15:05.698Z" },
{ url = "https://files.pythonhosted.org/packages/c1/0f/52954601d02d39742aab01d6b92f53c1dd38b2392248154c50797b4df7f1/multidict-6.4.4-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:be06e73c06415199200e9a2324a11252a3d62030319919cde5e6950ffeccf72e", size = 227106, upload-time = "2025-05-19T14:15:07.124Z" },
{ url = "https://files.pythonhosted.org/packages/af/24/679d83ec4379402d28721790dce818e5d6b9f94ce1323a556fb17fa9996c/multidict-6.4.4-cp312-cp312-win32.whl", hash = "sha256:622f26ea6a7e19b7c48dd9228071f571b2fbbd57a8cd71c061e848f281550e6b", size = 35351, upload-time = "2025-05-19T14:15:08.556Z" },
{ url = "https://files.pythonhosted.org/packages/52/ef/40d98bc5f986f61565f9b345f102409534e29da86a6454eb6b7c00225a13/multidict-6.4.4-cp312-cp312-win_amd64.whl", hash = "sha256:5e2bcda30d5009996ff439e02a9f2b5c3d64a20151d34898c000a6281faa3781", size = 38791, upload-time = "2025-05-19T14:15:09.825Z" },
{ url = "https://files.pythonhosted.org/packages/df/2a/e166d2ffbf4b10131b2d5b0e458f7cee7d986661caceae0de8753042d4b2/multidict-6.4.4-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:82ffabefc8d84c2742ad19c37f02cde5ec2a1ee172d19944d380f920a340e4b9", size = 64123, upload-time = "2025-05-19T14:15:11.044Z" },
{ url = "https://files.pythonhosted.org/packages/8c/96/e200e379ae5b6f95cbae472e0199ea98913f03d8c9a709f42612a432932c/multidict-6.4.4-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:6a2f58a66fe2c22615ad26156354005391e26a2f3721c3621504cd87c1ea87bf", size = 38049, upload-time = "2025-05-19T14:15:12.902Z" },
{ url = "https://files.pythonhosted.org/packages/75/fb/47afd17b83f6a8c7fa863c6d23ac5ba6a0e6145ed8a6bcc8da20b2b2c1d2/multidict-6.4.4-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:5883d6ee0fd9d8a48e9174df47540b7545909841ac82354c7ae4cbe9952603bd", size = 37078, upload-time = "2025-05-19T14:15:14.282Z" },
{ url = "https://files.pythonhosted.org/packages/fa/70/1af3143000eddfb19fd5ca5e78393985ed988ac493bb859800fe0914041f/multidict-6.4.4-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:9abcf56a9511653fa1d052bfc55fbe53dbee8f34e68bd6a5a038731b0ca42d15", size = 224097, upload-time = "2025-05-19T14:15:15.566Z" },
{ url = "https://files.pythonhosted.org/packages/b1/39/d570c62b53d4fba844e0378ffbcd02ac25ca423d3235047013ba2f6f60f8/multidict-6.4.4-cp313-cp313-manylinux_2_17_armv7l.manylinux2014_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:6ed5ae5605d4ad5a049fad2a28bb7193400700ce2f4ae484ab702d1e3749c3f9", size = 230768, upload-time = "2025-05-19T14:15:17.308Z" },
{ url = "https://files.pythonhosted.org/packages/fd/f8/ed88f2c4d06f752b015933055eb291d9bc184936903752c66f68fb3c95a7/multidict-6.4.4-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:bbfcb60396f9bcfa63e017a180c3105b8c123a63e9d1428a36544e7d37ca9e20", size = 231331, upload-time = "2025-05-19T14:15:18.73Z" },
{ url = "https://files.pythonhosted.org/packages/9c/6f/8e07cffa32f483ab887b0d56bbd8747ac2c1acd00dc0af6fcf265f4a121e/multidict-6.4.4-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:b0f1987787f5f1e2076b59692352ab29a955b09ccc433c1f6b8e8e18666f608b", size = 230169, upload-time = "2025-05-19T14:15:20.179Z" },
{ url = "https://files.pythonhosted.org/packages/e6/2b/5dcf173be15e42f330110875a2668ddfc208afc4229097312212dc9c1236/multidict-6.4.4-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:1d0121ccce8c812047d8d43d691a1ad7641f72c4f730474878a5aeae1b8ead8c", size = 222947, upload-time = "2025-05-19T14:15:21.714Z" },
{ url = "https://files.pythonhosted.org/packages/39/75/4ddcbcebe5ebcd6faa770b629260d15840a5fc07ce8ad295a32e14993726/multidict-6.4.4-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:83ec4967114295b8afd120a8eec579920c882831a3e4c3331d591a8e5bfbbc0f", size = 215761, upload-time = "2025-05-19T14:15:23.242Z" },
{ url = "https://files.pythonhosted.org/packages/6a/c9/55e998ae45ff15c5608e384206aa71a11e1b7f48b64d166db400b14a3433/multidict-6.4.4-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:995f985e2e268deaf17867801b859a282e0448633f1310e3704b30616d269d69", size = 227605, upload-time = "2025-05-19T14:15:24.763Z" },
{ url = "https://files.pythonhosted.org/packages/04/49/c2404eac74497503c77071bd2e6f88c7e94092b8a07601536b8dbe99be50/multidict-6.4.4-cp313-cp313-musllinux_1_2_armv7l.whl", hash = "sha256:d832c608f94b9f92a0ec8b7e949be7792a642b6e535fcf32f3e28fab69eeb046", size = 226144, upload-time = "2025-05-19T14:15:26.249Z" },
{ url = "https://files.pythonhosted.org/packages/62/c5/0cd0c3c6f18864c40846aa2252cd69d308699cb163e1c0d989ca301684da/multidict-6.4.4-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:d21c1212171cf7da703c5b0b7a0e85be23b720818aef502ad187d627316d5645", size = 221100, upload-time = "2025-05-19T14:15:28.303Z" },
{ url = "https://files.pythonhosted.org/packages/71/7b/f2f3887bea71739a046d601ef10e689528d4f911d84da873b6be9194ffea/multidict-6.4.4-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:cbebaa076aaecad3d4bb4c008ecc73b09274c952cf6a1b78ccfd689e51f5a5b0", size = 232731, upload-time = "2025-05-19T14:15:30.263Z" },
{ url = "https://files.pythonhosted.org/packages/e5/b3/d9de808349df97fa75ec1372758701b5800ebad3c46ae377ad63058fbcc6/multidict-6.4.4-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:c93a6fb06cc8e5d3628b2b5fda215a5db01e8f08fc15fadd65662d9b857acbe4", size = 229637, upload-time = "2025-05-19T14:15:33.337Z" },
{ url = "https://files.pythonhosted.org/packages/5e/57/13207c16b615eb4f1745b44806a96026ef8e1b694008a58226c2d8f5f0a5/multidict-6.4.4-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:8cd8f81f1310182362fb0c7898145ea9c9b08a71081c5963b40ee3e3cac589b1", size = 225594, upload-time = "2025-05-19T14:15:34.832Z" },
{ url = "https://files.pythonhosted.org/packages/3a/e4/d23bec2f70221604f5565000632c305fc8f25ba953e8ce2d8a18842b9841/multidict-6.4.4-cp313-cp313-win32.whl", hash = "sha256:3e9f1cd61a0ab857154205fb0b1f3d3ace88d27ebd1409ab7af5096e409614cd", size = 35359, upload-time = "2025-05-19T14:15:36.246Z" },
{ url = "https://files.pythonhosted.org/packages/a7/7a/cfe1a47632be861b627f46f642c1d031704cc1c0f5c0efbde2ad44aa34bd/multidict-6.4.4-cp313-cp313-win_amd64.whl", hash = "sha256:8ffb40b74400e4455785c2fa37eba434269149ec525fc8329858c862e4b35373", size = 38903, upload-time = "2025-05-19T14:15:37.507Z" },
{ url = "https://files.pythonhosted.org/packages/68/7b/15c259b0ab49938a0a1c8f3188572802704a779ddb294edc1b2a72252e7c/multidict-6.4.4-cp313-cp313t-macosx_10_13_universal2.whl", hash = "sha256:6a602151dbf177be2450ef38966f4be3467d41a86c6a845070d12e17c858a156", size = 68895, upload-time = "2025-05-19T14:15:38.856Z" },
{ url = "https://files.pythonhosted.org/packages/f1/7d/168b5b822bccd88142e0a3ce985858fea612404edd228698f5af691020c9/multidict-6.4.4-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:0d2b9712211b860d123815a80b859075d86a4d54787e247d7fbee9db6832cf1c", size = 40183, upload-time = "2025-05-19T14:15:40.197Z" },
{ url = "https://files.pythonhosted.org/packages/e0/b7/d4b8d98eb850ef28a4922ba508c31d90715fd9b9da3801a30cea2967130b/multidict-6.4.4-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:d2fa86af59f8fc1972e121ade052145f6da22758f6996a197d69bb52f8204e7e", size = 39592, upload-time = "2025-05-19T14:15:41.508Z" },
{ url = "https://files.pythonhosted.org/packages/18/28/a554678898a19583548e742080cf55d169733baf57efc48c2f0273a08583/multidict-6.4.4-cp313-cp313t-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:50855d03e9e4d66eab6947ba688ffb714616f985838077bc4b490e769e48da51", size = 226071, upload-time = "2025-05-19T14:15:42.877Z" },
{ url = "https://files.pythonhosted.org/packages/ee/dc/7ba6c789d05c310e294f85329efac1bf5b450338d2542498db1491a264df/multidict-6.4.4-cp313-cp313t-manylinux_2_17_armv7l.manylinux2014_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:5bce06b83be23225be1905dcdb6b789064fae92499fbc458f59a8c0e68718601", size = 222597, upload-time = "2025-05-19T14:15:44.412Z" },
{ url = "https://files.pythonhosted.org/packages/24/4f/34eadbbf401b03768dba439be0fb94b0d187facae9142821a3d5599ccb3b/multidict-6.4.4-cp313-cp313t-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:66ed0731f8e5dfd8369a883b6e564aca085fb9289aacabd9decd70568b9a30de", size = 228253, upload-time = "2025-05-19T14:15:46.474Z" },
{ url = "https://files.pythonhosted.org/packages/c0/e6/493225a3cdb0d8d80d43a94503fc313536a07dae54a3f030d279e629a2bc/multidict-6.4.4-cp313-cp313t-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:329ae97fc2f56f44d91bc47fe0972b1f52d21c4b7a2ac97040da02577e2daca2", size = 226146, upload-time = "2025-05-19T14:15:48.003Z" },
{ url = "https://files.pythonhosted.org/packages/2f/70/e411a7254dc3bff6f7e6e004303b1b0591358e9f0b7c08639941e0de8bd6/multidict-6.4.4-cp313-cp313t-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c27e5dcf520923d6474d98b96749e6805f7677e93aaaf62656005b8643f907ab", size = 220585, upload-time = "2025-05-19T14:15:49.546Z" },
{ url = "https://files.pythonhosted.org/packages/08/8f/beb3ae7406a619100d2b1fb0022c3bb55a8225ab53c5663648ba50dfcd56/multidict-6.4.4-cp313-cp313t-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:058cc59b9e9b143cc56715e59e22941a5d868c322242278d28123a5d09cdf6b0", size = 212080, upload-time = "2025-05-19T14:15:51.151Z" },
{ url = "https://files.pythonhosted.org/packages/9c/ec/355124e9d3d01cf8edb072fd14947220f357e1c5bc79c88dff89297e9342/multidict-6.4.4-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:69133376bc9a03f8c47343d33f91f74a99c339e8b58cea90433d8e24bb298031", size = 226558, upload-time = "2025-05-19T14:15:52.665Z" },
{ url = "https://files.pythonhosted.org/packages/fd/22/d2b95cbebbc2ada3be3812ea9287dcc9712d7f1a012fad041770afddb2ad/multidict-6.4.4-cp313-cp313t-musllinux_1_2_armv7l.whl", hash = "sha256:d6b15c55721b1b115c5ba178c77104123745b1417527ad9641a4c5e2047450f0", size = 212168, upload-time = "2025-05-19T14:15:55.279Z" },
{ url = "https://files.pythonhosted.org/packages/4d/c5/62bfc0b2f9ce88326dbe7179f9824a939c6c7775b23b95de777267b9725c/multidict-6.4.4-cp313-cp313t-musllinux_1_2_i686.whl", hash = "sha256:a887b77f51d3d41e6e1a63cf3bc7ddf24de5939d9ff69441387dfefa58ac2e26", size = 217970, upload-time = "2025-05-19T14:15:56.806Z" },
{ url = "https://files.pythonhosted.org/packages/79/74/977cea1aadc43ff1c75d23bd5bc4768a8fac98c14e5878d6ee8d6bab743c/multidict-6.4.4-cp313-cp313t-musllinux_1_2_ppc64le.whl", hash = "sha256:632a3bf8f1787f7ef7d3c2f68a7bde5be2f702906f8b5842ad6da9d974d0aab3", size = 226980, upload-time = "2025-05-19T14:15:58.313Z" },
{ url = "https://files.pythonhosted.org/packages/48/fc/cc4a1a2049df2eb84006607dc428ff237af38e0fcecfdb8a29ca47b1566c/multidict-6.4.4-cp313-cp313t-musllinux_1_2_s390x.whl", hash = "sha256:a145c550900deb7540973c5cdb183b0d24bed6b80bf7bddf33ed8f569082535e", size = 220641, upload-time = "2025-05-19T14:15:59.866Z" },
{ url = "https://files.pythonhosted.org/packages/3b/6a/a7444d113ab918701988d4abdde373dbdfd2def7bd647207e2bf645c7eac/multidict-6.4.4-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:cc5d83c6619ca5c9672cb78b39ed8542f1975a803dee2cda114ff73cbb076edd", size = 221728, upload-time = "2025-05-19T14:16:01.535Z" },
{ url = "https://files.pythonhosted.org/packages/2b/b0/fdf4c73ad1c55e0f4dbbf2aa59dd37037334091f9a4961646d2b7ac91a86/multidict-6.4.4-cp313-cp313t-win32.whl", hash = "sha256:3312f63261b9df49be9d57aaa6abf53a6ad96d93b24f9cc16cf979956355ce6e", size = 41913, upload-time = "2025-05-19T14:16:03.199Z" },
{ url = "https://files.pythonhosted.org/packages/8e/92/27989ecca97e542c0d01d05a98a5ae12198a243a9ee12563a0313291511f/multidict-6.4.4-cp313-cp313t-win_amd64.whl", hash = "sha256:ba852168d814b2c73333073e1c7116d9395bea69575a01b0b3c89d2d5a87c8fb", size = 46112, upload-time = "2025-05-19T14:16:04.909Z" },
{ url = "https://files.pythonhosted.org/packages/84/5d/e17845bb0fa76334477d5de38654d27946d5b5d3695443987a094a71b440/multidict-6.4.4-py3-none-any.whl", hash = "sha256:bd4557071b561a8b3b6075c3ce93cf9bfb6182cb241805c3d66ced3b75eff4ac", size = 10481, upload-time = "2025-05-19T14:16:36.024Z" },
]
[[package]]
name = "protobuf"
version = "5.29.4"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/17/7d/b9dca7365f0e2c4fa7c193ff795427cfa6290147e5185ab11ece280a18e7/protobuf-5.29.4.tar.gz", hash = "sha256:4f1dfcd7997b31ef8f53ec82781ff434a28bf71d9102ddde14d076adcfc78c99", size = 424902, upload-time = "2025-03-19T21:23:24.25Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9a/b2/043a1a1a20edd134563699b0e91862726a0dc9146c090743b6c44d798e75/protobuf-5.29.4-cp310-abi3-win32.whl", hash = "sha256:13eb236f8eb9ec34e63fc8b1d6efd2777d062fa6aaa68268fb67cf77f6839ad7", size = 422709, upload-time = "2025-03-19T21:23:08.293Z" },
{ url = "https://files.pythonhosted.org/packages/79/fc/2474b59570daa818de6124c0a15741ee3e5d6302e9d6ce0bdfd12e98119f/protobuf-5.29.4-cp310-abi3-win_amd64.whl", hash = "sha256:bcefcdf3976233f8a502d265eb65ea740c989bacc6c30a58290ed0e519eb4b8d", size = 434506, upload-time = "2025-03-19T21:23:11.253Z" },
{ url = "https://files.pythonhosted.org/packages/46/de/7c126bbb06aa0f8a7b38aaf8bd746c514d70e6a2a3f6dd460b3b7aad7aae/protobuf-5.29.4-cp38-abi3-macosx_10_9_universal2.whl", hash = "sha256:307ecba1d852ec237e9ba668e087326a67564ef83e45a0189a772ede9e854dd0", size = 417826, upload-time = "2025-03-19T21:23:13.132Z" },
{ url = "https://files.pythonhosted.org/packages/a2/b5/bade14ae31ba871a139aa45e7a8183d869efe87c34a4850c87b936963261/protobuf-5.29.4-cp38-abi3-manylinux2014_aarch64.whl", hash = "sha256:aec4962f9ea93c431d5714ed1be1c93f13e1a8618e70035ba2b0564d9e633f2e", size = 319574, upload-time = "2025-03-19T21:23:14.531Z" },
{ url = "https://files.pythonhosted.org/packages/46/88/b01ed2291aae68b708f7d334288ad5fb3e7aa769a9c309c91a0d55cb91b0/protobuf-5.29.4-cp38-abi3-manylinux2014_x86_64.whl", hash = "sha256:d7d3f7d1d5a66ed4942d4fefb12ac4b14a29028b209d4bfb25c68ae172059922", size = 319672, upload-time = "2025-03-19T21:23:15.839Z" },
{ url = "https://files.pythonhosted.org/packages/12/fb/a586e0c973c95502e054ac5f81f88394f24ccc7982dac19c515acd9e2c93/protobuf-5.29.4-py3-none-any.whl", hash = "sha256:3fde11b505e1597f71b875ef2fc52062b6a9740e5f7c8997ce878b6009145862", size = 172551, upload-time = "2025-03-19T21:23:22.682Z" },
]
[[package]]
name = "setuptools"
version = "80.8.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/8d/d2/ec1acaaff45caed5c2dedb33b67055ba9d4e96b091094df90762e60135fe/setuptools-80.8.0.tar.gz", hash = "sha256:49f7af965996f26d43c8ae34539c8d99c5042fbff34302ea151eaa9c207cd257", size = 1319720, upload-time = "2025-05-20T14:02:53.503Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/58/29/93c53c098d301132196c3238c312825324740851d77a8500a2462c0fd888/setuptools-80.8.0-py3-none-any.whl", hash = "sha256:95a60484590d24103af13b686121328cc2736bee85de8936383111e421b9edc0", size = 1201470, upload-time = "2025-05-20T14:02:51.348Z" },
]
[[package]]
name = "worker-py"
version = "0.1.0"
source = { virtual = "." }
[package.dev-dependencies]
dev = [
{ name = "grpcio" },
{ name = "grpcio-tools" },
{ name = "grpclib" },
{ name = "protobuf" },
]
[package.metadata]
[package.metadata.requires-dev]
dev = [
{ name = "grpcio", specifier = ">=1.71.0" },
{ name = "grpcio-tools", specifier = ">=1.71.0" },
{ name = "grpclib", specifier = ">=0.4.8" },
{ name = "protobuf", specifier = ">=5.29.4" },
]
+1 -1
View File
@@ -14,7 +14,7 @@ OUTPUT ?= out/benchmark.json
install: ## Install dependencies
uv sync --frozen --all-extras --all-packages --group dev
benchmark: .uv
benchmark:
mkdir -p out
rm -f $(OUTPUT)
uv run python -m bench -o $(OUTPUT) --rigorous
+2 -2
View File
@@ -14,7 +14,7 @@
[![Docs](https://img.shields.io/badge/docs-latest-blue)](https://langchain-ai.github.io/langgraph/)
[![GitMCP](https://img.shields.io/endpoint?url=https://gitmcp.io/badge/langchain-ai/langgraph)](https://gitmcp.io/langchain-ai/langgraph)
Trusted by companies shaping the future of agents – including Klarna, Replit, Elastic, and more – LangGraph is a powerful low-level orchestration framework for building, managing, and deploying long-running, stateful agents.
Trusted by companies shaping the future of agents – including Klarna, Replit, Elastic, and more – LangGraph is a low-level orchestration framework for building, managing, and deploying long-running, stateful agents.
## Get started
@@ -77,7 +77,7 @@ While LangGraph can be used standalone, it also integrates seamlessly with any L
- [Examples](https://langchain-ai.github.io/langgraph/tutorials/): Guided examples on getting started with LangGraph.
- [LangChain Academy](https://academy.langchain.com/courses/intro-to-langgraph): Learn the basics of LangGraph in our free, structured course.
- [Templates](https://langchain-ai.github.io/langgraph/concepts/template_applications/): Pre-built reference apps for common agentic workflows (e.g. ReAct agent, memory, retrieval etc.) that can be cloned and adapted.
- [Case studies](https://www.langchain.com/built-with-langgraph): Hear how industry leaders use LangGraph to ship powerful, production-ready AI applications.
- [Case studies](https://www.langchain.com/built-with-langgraph): Hear how industry leaders use LangGraph to ship AI applications at scale.
## Acknowledgements
+1
View File
@@ -304,6 +304,7 @@ class StateGraph(Graph):
If a string is provided, it will be used as the node name, and action will be used as the function or runnable.
action: The action associated with the node. (default: None)
Will be used as the node function or runnable if `node` is a string (node name).
defer: Whether to defer the execution of the node until the run is about to end.
metadata: The metadata associated with the node. (default: None)
input: The input schema for the node. (default: the graph's input schema)
retry: The policy for retrying the node. (default: None)
+30 -207
View File
@@ -85,7 +85,6 @@ from langgraph.pregel.algo import (
PregelTaskWrites,
apply_writes,
local_read,
local_write,
prepare_next_tasks,
)
from langgraph.pregel.call import identifier
@@ -1684,11 +1683,7 @@ class Pregel(PregelProtocol):
run_name=self.name + "UpdateState",
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
self.nodes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -2111,11 +2106,7 @@ class Pregel(PregelProtocol):
run_name=self.name + "UpdateState",
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
self.nodes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -2278,7 +2269,7 @@ class Pregel(PregelProtocol):
Args:
input: The input to the graph.
config: The configuration to use for the run.
stream_mode: The mode to stream output, defaults to self.stream_mode.
stream_mode: The mode to stream output, defaults to `self.stream_mode`.
Options are:
- `"values"`: Emit all values in the state after each step, including interrupts.
@@ -2287,112 +2278,28 @@ class Pregel(PregelProtocol):
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
- `"custom"`: Emit custom data from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
Will be emitted as 2-tuples `(LLM token, metadata)`.
- `"debug"`: Emit debug events with as much information as possible for each step.
You can pass a list as the `stream_mode` parameter to stream multiple modes at once.
The streamed outputs will be tuples of `(mode, data)`.
See [LangGraph streaming guide](https://langchain-ai.github.io/langgraph/how-tos/streaming/) for more details.
output_keys: The keys to stream, defaults to all non-context channels.
interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph.
interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph.
checkpoint_during: Whether to checkpoint intermediate steps, defaults to True. If False, only the final checkpoint is saved.
debug: Whether to print debug information during execution, defaults to False.
subgraphs: Whether to stream subgraphs, defaults to False.
subgraphs: Whether to stream events from inside subgraphs, defaults to False.
If True, the events will be emitted as tuples `(namespace, data)`,
or `(namespace, mode, data)` if `stream_mode` is a list,
where `namespace` is a tuple with the path to the node where a subgraph is invoked,
e.g. `("parent_node:<task_id>", "child_node:<task_id>")`.
See [LangGraph streaming guide](https://langchain-ai.github.io/langgraph/how-tos/streaming/) for more details.
Yields:
The output of each step in the graph. The output shape depends on the stream_mode.
Example: Using stream_mode="values":
```python
import operator
from typing_extensions import Annotated, TypedDict
from langgraph.graph import StateGraph, START
class State(TypedDict):
alist: Annotated[list, operator.add]
another_list: Annotated[list, operator.add]
builder = StateGraph(State)
builder.add_node("a", lambda _state: {"another_list": ["hi"]})
builder.add_node("b", lambda _state: {"alist": ["there"]})
builder.add_edge("a", "b")
builder.add_edge(START, "a")
graph = builder.compile()
for event in graph.stream({"alist": ['Ex for stream_mode="values"']}, stream_mode="values"):
print(event)
# {'alist': ['Ex for stream_mode="values"'], 'another_list': []}
# {'alist': ['Ex for stream_mode="values"'], 'another_list': ['hi']}
# {'alist': ['Ex for stream_mode="values"', 'there'], 'another_list': ['hi']}
```
Example: Using stream_mode="updates":
```python
for event in graph.stream({"alist": ['Ex for stream_mode="updates"']}, stream_mode="updates"):
print(event)
# {'a': {'another_list': ['hi']}}
# {'b': {'alist': ['there']}}
```
Example: Using stream_mode="debug":
```python
for event in graph.stream({"alist": ['Ex for stream_mode="debug"']}, stream_mode="debug"):
print(event)
# {'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': []}, 'triggers': ['start:a']}}
# {'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'result': [('another_list', ['hi'])]}}
# {'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': ['hi']}, 'triggers': ['a']}}
# {'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}}
```
Example: Using stream_mode="custom":
```python
from langgraph.types import StreamWriter
def node_a(state: State, writer: StreamWriter):
writer({"custom_data": "foo"})
return {"alist": ["hi"]}
builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
graph = builder.compile()
for event in graph.stream({"alist": ['Ex for stream_mode="custom"']}, stream_mode="custom"):
print(event)
# {'custom_data': 'foo'}
```
Example: Using stream_mode="messages":
```python
from typing_extensions import Annotated, TypedDict
from langgraph.graph import StateGraph, START
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
class State(TypedDict):
question: str
answer: str
def node_a(state: State):
response = llm.invoke(state["question"])
return {"answer": response.content}
builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
graph = builder.compile()
for event in graph.stream({"question": "What is the capital of France?"}, stream_mode="messages"):
print(event)
# (AIMessageChunk(content='The', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], 'langgraph_path': ('__pregel_pull', 'a'), 'langgraph_checkpoint_ns': '...', 'checkpoint_ns': '...', 'ls_provider': 'openai', 'ls_model_name': 'gpt-4o-mini', 'ls_model_type': 'chat', 'ls_temperature': 0.7})
# (AIMessageChunk(content=' capital', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], ...})
# (AIMessageChunk(content=' of', additional_kwargs={}, response_metadata={}, id='...'), {...})
# (AIMessageChunk(content=' France', additional_kwargs={}, response_metadata={}, id='...'), {...})
# (AIMessageChunk(content=' is', additional_kwargs={}, response_metadata={}, id='...'), {...})
# (AIMessageChunk(content=' Paris', additional_kwargs={}, response_metadata={}, id='...'), {...})
```
"""
stream = SyncQueue()
@@ -2569,7 +2476,7 @@ class Pregel(PregelProtocol):
Args:
input: The input to the graph.
config: The configuration to use for the run.
stream_mode: The mode to stream output, defaults to self.stream_mode.
stream_mode: The mode to stream output, defaults to `self.stream_mode`.
Options are:
- `"values"`: Emit all values in the state after each step, including interrupts.
@@ -2578,112 +2485,28 @@ class Pregel(PregelProtocol):
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
- `"custom"`: Emit custom data from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
Will be emitted as 2-tuples `(LLM token, metadata)`.
- `"debug"`: Emit debug events with as much information as possible for each step.
You can pass a list as the `stream_mode` parameter to stream multiple modes at once.
The streamed outputs will be tuples of `(mode, data)`.
See [LangGraph streaming guide](https://langchain-ai.github.io/langgraph/how-tos/streaming/) for more details.
output_keys: The keys to stream, defaults to all non-context channels.
interrupt_before: Nodes to interrupt before, defaults to all nodes in the graph.
interrupt_after: Nodes to interrupt after, defaults to all nodes in the graph.
checkpoint_during: Whether to checkpoint intermediate steps, defaults to True. If False, only the final checkpoint is saved.
debug: Whether to print debug information during execution, defaults to False.
subgraphs: Whether to stream subgraphs, defaults to False.
subgraphs: Whether to stream events from inside subgraphs, defaults to False.
If True, the events will be emitted as tuples `(namespace, data)`,
or `(namespace, mode, data)` if `stream_mode` is a list,
where `namespace` is a tuple with the path to the node where a subgraph is invoked,
e.g. `("parent_node:<task_id>", "child_node:<task_id>")`.
See [LangGraph streaming guide](https://langchain-ai.github.io/langgraph/how-tos/streaming/) for more details.
Yields:
The output of each step in the graph. The output shape depends on the stream_mode.
Example: Using stream_mode="values":
```python
import operator
from typing_extensions import Annotated, TypedDict
from langgraph.graph import StateGraph, START
class State(TypedDict):
alist: Annotated[list, operator.add]
another_list: Annotated[list, operator.add]
builder = StateGraph(State)
builder.add_node("a", lambda _state: {"another_list": ["hi"]})
builder.add_node("b", lambda _state: {"alist": ["there"]})
builder.add_edge("a", "b")
builder.add_edge(START, "a")
graph = builder.compile()
async for event in graph.astream({"alist": ['Ex for stream_mode="values"']}, stream_mode="values"):
print(event)
# {'alist': ['Ex for stream_mode="values"'], 'another_list': []}
# {'alist': ['Ex for stream_mode="values"'], 'another_list': ['hi']}
# {'alist': ['Ex for stream_mode="values"', 'there'], 'another_list': ['hi']}
```
Example: Using stream_mode="updates":
```python
async for event in graph.astream({"alist": ['Ex for stream_mode="updates"']}, stream_mode="updates"):
print(event)
# {'a': {'another_list': ['hi']}}
# {'b': {'alist': ['there']}}
```
Example: Using stream_mode="debug":
```python
async for event in graph.astream({"alist": ['Ex for stream_mode="debug"']}, stream_mode="debug"):
print(event)
# {'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': []}, 'triggers': ['start:a']}}
# {'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 1, 'payload': {'id': '...', 'name': 'a', 'result': [('another_list', ['hi'])]}}
# {'type': 'task', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'input': {'alist': ['Ex for stream_mode="debug"'], 'another_list': ['hi']}, 'triggers': ['a']}}
# {'type': 'task_result', 'timestamp': '2024-06-23T...+00:00', 'step': 2, 'payload': {'id': '...', 'name': 'b', 'result': [('alist', ['there'])]}}
```
Example: Using stream_mode="custom":
```python
from langgraph.types import StreamWriter
async def node_a(state: State, writer: StreamWriter):
writer({"custom_data": "foo"})
return {"alist": ["hi"]}
builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
graph = builder.compile()
async for event in graph.astream({"alist": ['Ex for stream_mode="custom"']}, stream_mode="custom"):
print(event)
# {'custom_data': 'foo'}
```
Example: Using stream_mode="messages":
```python
from typing_extensions import Annotated, TypedDict
from langgraph.graph import StateGraph, START
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
class State(TypedDict):
question: str
answer: str
async def node_a(state: State):
response = await llm.ainvoke(state["question"])
return {"answer": response.content}
builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
graph = builder.compile()
async for event in graph.astream({"question": "What is the capital of France?"}, stream_mode="messages"):
print(event)
# (AIMessageChunk(content='The', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], 'langgraph_path': ('__pregel_pull', 'a'), 'langgraph_checkpoint_ns': '...', 'checkpoint_ns': '...', 'ls_provider': 'openai', 'ls_model_name': 'gpt-4o-mini', 'ls_model_type': 'chat', 'ls_temperature': 0.7})
# (AIMessageChunk(content=' capital', additional_kwargs={}, response_metadata={}, id='...'), {'langgraph_step': 1, 'langgraph_node': 'a', 'langgraph_triggers': ['start:a'], ...})
# (AIMessageChunk(content=' of', additional_kwargs={}, response_metadata={}, id='...'), {...})
# (AIMessageChunk(content=' France', additional_kwargs={}, response_metadata={}, id='...'), {...})
# (AIMessageChunk(content=' is', additional_kwargs={}, response_metadata={}, id='...'), {...})
# (AIMessageChunk(content=' Paris', additional_kwargs={}, response_metadata={}, id='...'), {...})
```
"""
stream = AsyncQueue()
+3 -33
View File
@@ -19,7 +19,6 @@ from typing import (
overload,
)
# meaningless change to trigger tests
from langchain_core.callbacks import Callbacks
from langchain_core.callbacks.manager import AsyncParentRunManager, ParentRunManager
from langchain_core.runnables.config import RunnableConfig
@@ -65,7 +64,6 @@ from langgraph.constants import (
TASKS,
Send,
)
from langgraph.errors import InvalidUpdateError
from langgraph.managed.base import ManagedValueMapping
from langgraph.pregel.call import get_runnable_for_task, identifier
from langgraph.pregel.io import read_channels
@@ -213,22 +211,6 @@ def local_read(
return values
def local_write(
commit: Callable[[Sequence[tuple[str, Any]]], None],
process_keys: Iterable[str],
writes: Sequence[tuple[str, Any]],
) -> None:
"""Function injected under CONFIG_KEY_SEND in task config, to write to channels.
Validates writes and forwards them to `commit` function."""
for chan, value in writes:
if chan in (PUSH, TASKS) and value is not None:
if not isinstance(value, Send):
raise InvalidUpdateError(f"Expected Send, got {value}")
if value.node not in process_keys:
raise InvalidUpdateError(f"Invalid node name {value.node} in packet")
commit(writes)
def increment(current: Optional[int], channel: BaseChannel) -> int:
"""Default channel versioning function, increments the current int version."""
return current + 1 if current is not None else 1
@@ -627,11 +609,7 @@ def prepare_single_task(
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
processes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -751,11 +729,7 @@ def prepare_single_task(
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
processes.keys(),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
@@ -889,11 +863,7 @@ def prepare_single_task(
configurable={
CONFIG_KEY_TASK_ID: task_id,
# deque.extend is thread-safe
CONFIG_KEY_SEND: partial(
local_write,
writes.extend,
tuple(processes.keys()),
),
CONFIG_KEY_SEND: writes.extend,
CONFIG_KEY_READ: partial(
local_read,
channels,
+4 -2
View File
@@ -1083,7 +1083,8 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
self, task: PregelExecutableTask, write_idx: int, call: Optional[Call] = None
) -> Optional[PregelExecutableTask]:
if pushed := super().accept_push(task, write_idx, call):
self.match_cached_writes()
for task in self.match_cached_writes():
self.output_writes(task.id, task.writes, cached=True)
return pushed
def put_writes(self, task_id: str, writes: WritesT) -> None:
@@ -1279,7 +1280,8 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
self, task: PregelExecutableTask, write_idx: int, call: Optional[Call] = None
) -> Optional[PregelExecutableTask]:
if pushed := super().accept_push(task, write_idx, call):
await self.amatch_cached_writes()
for task in await self.amatch_cached_writes():
self.output_writes(task.id, task.writes, cached=True)
return pushed
def put_writes(self, task_id: str, writes: WritesT) -> None:
+18 -18
View File
@@ -25,30 +25,30 @@ Repository = "https://www.github.com/langchain-ai/langgraph"
[dependency-groups]
dev = [
"pytest>=8.3.2",
"pytest-cov>=4.0.0",
"pytest-dotenv>=0.5.2",
"pytest-mock>=3.10.0",
"syrupy>=4.0.2",
"httpx>=0.26.0",
'pytest-watcher>=0.4.1',
"mypy>=1.6.0",
"ruff>=0.6.2",
"jupyter>=1.0.0",
"pytest-xdist[psutil]>=3.6.1",
"pytest-repeat>=0.9.3",
"pytest",
"pytest-cov",
"pytest-dotenv",
"pytest-mock",
"syrupy",
"httpx",
"pytest-watcher",
"mypy",
"ruff",
"jupyter",
"pytest-xdist[psutil]",
"pytest-repeat",
"langgraph-prebuilt",
"langgraph-checkpoint",
"langgraph-checkpoint-sqlite",
"langgraph-checkpoint-postgres",
"langgraph-sdk",
'psycopg[binary]>=3.0.0; python_version >= "3.10"',
"psycopg[binary]",
"uvloop==0.21.0beta1",
"pyperf>=2.7.0",
"py-spy>=0.3.14",
"types-requests>=2.32.0.20240914",
"pycryptodome>=3.21.0",
"langgraph-cli[inmem]>=0.2.8",
"pyperf",
"py-spy",
"types-requests",
"pycryptodome",
"langgraph-cli[inmem]",
]
[tool.uv]
+36 -22
View File
@@ -1,5 +1,4 @@
version = 1
revision = 1
requires-python = ">=3.9"
resolution-markers = [
"python_full_version >= '3.13' and python_full_version < '4.0'",
@@ -1219,7 +1218,7 @@ dev = [
{ name = "langgraph-prebuilt" },
{ name = "langgraph-sdk" },
{ name = "mypy" },
{ name = "psycopg", extra = ["binary"], marker = "python_full_version >= '3.10'" },
{ name = "psycopg", extra = ["binary"] },
{ name = "py-spy" },
{ name = "pycryptodome" },
{ name = "pyperf" },
@@ -1248,29 +1247,29 @@ requires-dist = [
[package.metadata.requires-dev]
dev = [
{ name = "httpx", specifier = ">=0.26.0" },
{ name = "jupyter", specifier = ">=1.0.0" },
{ name = "httpx" },
{ name = "jupyter" },
{ name = "langgraph-checkpoint", editable = "../checkpoint" },
{ name = "langgraph-checkpoint-postgres", editable = "../checkpoint-postgres" },
{ name = "langgraph-checkpoint-sqlite", editable = "../checkpoint-sqlite" },
{ name = "langgraph-cli", extras = ["inmem"], specifier = ">=0.2.8" },
{ name = "langgraph-cli", extras = ["inmem"] },
{ name = "langgraph-prebuilt", editable = "../prebuilt" },
{ name = "langgraph-sdk", editable = "../sdk-py" },
{ name = "mypy", specifier = ">=1.6.0" },
{ name = "psycopg", extras = ["binary"], marker = "python_full_version >= '3.10'", specifier = ">=3.0.0" },
{ name = "py-spy", specifier = ">=0.3.14" },
{ name = "pycryptodome", specifier = ">=3.21.0" },
{ name = "pyperf", specifier = ">=2.7.0" },
{ name = "pytest", specifier = ">=8.3.2" },
{ name = "pytest-cov", specifier = ">=4.0.0" },
{ name = "pytest-dotenv", specifier = ">=0.5.2" },
{ name = "pytest-mock", specifier = ">=3.10.0" },
{ name = "pytest-repeat", specifier = ">=0.9.3" },
{ name = "pytest-watcher", specifier = ">=0.4.1" },
{ name = "pytest-xdist", extras = ["psutil"], specifier = ">=3.6.1" },
{ name = "ruff", specifier = ">=0.6.2" },
{ name = "syrupy", specifier = ">=4.0.2" },
{ name = "types-requests", specifier = ">=2.32.0.20240914" },
{ name = "mypy" },
{ name = "psycopg", extras = ["binary"] },
{ name = "py-spy" },
{ name = "pycryptodome" },
{ name = "pyperf" },
{ name = "pytest" },
{ name = "pytest-cov" },
{ name = "pytest-dotenv" },
{ name = "pytest-mock" },
{ name = "pytest-repeat" },
{ name = "pytest-watcher" },
{ name = "pytest-xdist", extras = ["psutil"] },
{ name = "ruff" },
{ name = "syrupy" },
{ name = "types-requests" },
{ name = "uvloop", specifier = "==0.21.0b1" },
]
@@ -1366,17 +1365,19 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.7"
version = "2.0.10"
source = { editable = "../checkpoint-sqlite" }
dependencies = [
{ name = "aiosqlite" },
{ name = "langgraph-checkpoint" },
{ name = "sqlite-vec" },
]
[package.metadata]
requires-dist = [
{ name = "aiosqlite", specifier = ">=0.20" },
{ name = "langgraph-checkpoint", editable = "../checkpoint" },
{ name = "sqlite-vec", specifier = ">=0.1.6" },
]
[package.metadata.requires-dev]
@@ -1387,6 +1388,7 @@ dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
{ name = "pytest-retry", specifier = ">=1.7.0" },
{ name = "pytest-watcher" },
{ name = "ruff" },
]
@@ -1984,7 +1986,7 @@ wheels = [
[package.optional-dependencies]
binary = [
{ name = "psycopg-binary", marker = "python_full_version >= '3.10' and implementation_name != 'pypy'" },
{ name = "psycopg-binary", marker = "implementation_name != 'pypy'" },
]
[[package]]
@@ -2823,6 +2825,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e7/9c/0e6afc12c269578be5c0c1c9f4b49a8d32770a080260c333ac04cc1c832d/soupsieve-2.7-py3-none-any.whl", hash = "sha256:6e60cc5c1ffaf1cebcc12e8188320b72071e922c2e897f737cadce79ad5d30c4", size = 36677 },
]
[[package]]
name = "sqlite-vec"
version = "0.1.6"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/88/ed/aabc328f29ee6814033d008ec43e44f2c595447d9cccd5f2aabe60df2933/sqlite_vec-0.1.6-py3-none-macosx_10_6_x86_64.whl", hash = "sha256:77491bcaa6d496f2acb5cc0d0ff0b8964434f141523c121e313f9a7d8088dee3", size = 164075 },
{ url = "https://files.pythonhosted.org/packages/a7/57/05604e509a129b22e303758bfa062c19afb020557d5e19b008c64016704e/sqlite_vec-0.1.6-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fdca35f7ee3243668a055255d4dee4dea7eed5a06da8cad409f89facf4595361", size = 165242 },
{ url = "https://files.pythonhosted.org/packages/f2/48/dbb2cc4e5bad88c89c7bb296e2d0a8df58aab9edc75853728c361eefc24f/sqlite_vec-0.1.6-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7b0519d9cd96164cd2e08e8eed225197f9cd2f0be82cb04567692a0a4be02da3", size = 103704 },
{ url = "https://files.pythonhosted.org/packages/80/76/97f33b1a2446f6ae55e59b33869bed4eafaf59b7f4c662c8d9491b6a714a/sqlite_vec-0.1.6-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux1_x86_64.whl", hash = "sha256:823b0493add80d7fe82ab0fe25df7c0703f4752941aee1c7b2b02cec9656cb24", size = 151556 },
{ url = "https://files.pythonhosted.org/packages/6a/98/e8bc58b178266eae2fcf4c9c7a8303a8d41164d781b32d71097924a6bebe/sqlite_vec-0.1.6-py3-none-win_amd64.whl", hash = "sha256:c65bcfd90fa2f41f9000052bcb8bb75d38240b2dae49225389eca6c3136d3f0c", size = 281540 },
]
[[package]]
name = "sse-starlette"
version = "2.1.3"
@@ -247,6 +247,7 @@ def create_react_agent(
Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]
] = None,
pre_model_hook: Optional[RunnableLike] = None,
post_model_hook: Optional[RunnableLike] = None,
state_schema: Optional[StateSchemaType] = None,
config_schema: Optional[Type[Any]] = None,
checkpointer: Optional[Checkpointer] = None,
@@ -321,6 +322,12 @@ def create_react_agent(
...
}
```
post_model_hook: An optional node to add after the `agent` node (i.e., the node that calls the LLM).
Useful for implementing human-in-the-loop, guardrails, validation, or other post-processing.
Post-model hook must be a callable or a runnable that takes in current graph state and returns a state update.
!!! Note
Only available with `version="v2"`.
state_schema: An optional state schema that defines graph state.
Must have `messages` and `remaining_steps` keys.
Defaults to `AgentState` that defines those two keys.
@@ -591,6 +598,10 @@ def create_react_agent(
workflow.set_entry_point(entrypoint)
if post_model_hook is not None:
workflow.add_node("post_model_hook", post_model_hook)
workflow.add_edge("agent", "post_model_hook")
if response_format is not None:
workflow.add_node(
"generate_structured_response",
@@ -598,7 +609,10 @@ def create_react_agent(
generate_structured_response, agenerate_structured_response
),
)
workflow.add_edge("agent", "generate_structured_response")
if post_model_hook is not None:
workflow.add_edge("post_model_hook", "generate_structured_response")
else:
workflow.add_edge("agent", "generate_structured_response")
return workflow.compile(
checkpointer=checkpointer,
@@ -610,17 +624,24 @@ def create_react_agent(
)
# Define the function that determines whether to continue or not
def should_continue(state: StateSchema) -> Union[str, list]:
def should_continue(state: StateSchema) -> Union[str, list[Send]]:
messages = _get_state_value(state, "messages")
last_message = messages[-1]
# If there is no function call, then we finish
if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
return END if response_format is None else "generate_structured_response"
if post_model_hook is not None:
return "post_model_hook"
elif response_format is not None:
return "generate_structured_response"
else:
return END
# Otherwise if there is, we continue
else:
if version == "v1":
return "tools"
elif version == "v2":
if post_model_hook is not None:
return "post_model_hook"
tool_calls = [
tool_node.inject_tool_args(call, state, store) # type: ignore[arg-type]
for call in last_message.tool_calls
@@ -649,6 +670,14 @@ def create_react_agent(
# This means that this node is the first one called
workflow.set_entry_point(entrypoint)
agent_paths = ["tools", END]
post_model_hook_paths = [entrypoint, "tools", END]
# Add a post model hook node if post_model_hook is provided
if post_model_hook is not None:
workflow.add_node("post_model_hook", post_model_hook)
agent_paths.append("post_model_hook")
# Add a structured output node if response_format is provided
if response_format is not None:
workflow.add_node(
@@ -657,19 +686,52 @@ def create_react_agent(
generate_structured_response, agenerate_structured_response
),
)
workflow.add_edge("generate_structured_response", END)
should_continue_destinations = ["tools", "generate_structured_response"]
else:
should_continue_destinations = ["tools", END]
if post_model_hook is not None:
post_model_hook_paths.append("generate_structured_response")
else:
agent_paths.append("generate_structured_response")
if post_model_hook is not None:
def post_model_hook_router(state: StateSchema) -> Union[str, list[Send]]:
"""Route to the next node after post_model_hook.
Routes to one of:
* "tools": if there are pending tool calls without a corresponding message.
* "generate_structured_response": if no pending tool calls exist and response_format is specified.
* END: if no pending tool calls exist and no response_format is specified.
"""
messages = _get_state_value(state, "messages")
tool_messages = [
m.tool_call_id for m in messages if isinstance(m, ToolMessage)
]
last_ai_message = next(
m for m in reversed(messages) if isinstance(m, AIMessage)
)
pending_tool_calls = [
c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
]
if pending_tool_calls:
return [Send("tools", [tool_call]) for tool_call in pending_tool_calls]
elif isinstance(messages[-1], ToolMessage):
return entrypoint
elif response_format is not None:
return "generate_structured_response"
else:
return END
workflow.add_conditional_edges(
"post_model_hook",
post_model_hook_router, # type: ignore[arg-type]
path_map=post_model_hook_paths,
)
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agent`.
# This means these are the edges taken after the `agent` node is called.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
path_map=should_continue_destinations,
should_continue, # type: ignore[arg-type]
path_map=agent_paths,
)
def route_tool_responses(state: StateSchema) -> str:
+162 -5
View File
@@ -1,11 +1,12 @@
from typing import (
Literal,
Optional,
Union,
)
from copy import deepcopy
from typing import Any, Literal, Optional, Union, cast
from langchain_core.messages import ToolCall, ToolMessage
from typing_extensions import TypedDict
from langgraph.types import Command, interrupt
from langgraph.utils.runnable import RunnableCallable
class HumanInterruptConfig(TypedDict):
"""Configuration that defines what actions are allowed for a human interrupt.
@@ -92,3 +93,159 @@ class HumanResponse(TypedDict):
type: Literal["accept", "ignore", "response", "edit"]
args: Union[None, str, ActionRequest]
class InterruptToolNode(RunnableCallable):
"""Prebuilt post model hook node used to enable common patterns for tool interrupts.
For any tools with specified policies, an interrupt will be raised when the LLM returns
a tool call for said tool. The interrupt policy will be used to determine what sort of resume logic is allowed.
Any of the following resume patterns are supported:
* accept: the tool call is executed as planned
* edit: the args for the tool call are edited and then the tool call is executed
* response: text response/feedback is fed back into the LLM
* ignore: the current tool call is ignored / skipped
Args:
**interrupt_policy: a mapping of tool names to [`HumanInterruptConfig`][prebuilt.interrupt.HumanInterruptConfig] dictionaries
specifying which interrupt patterns to enable for said tool.
Example:
```python
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt.interrupt import HumanInterruptConfig, InterruptToolNode
from langgraph.types import Command
def book_hotel(hotel_name: str) -> str:
'''Book a room at the provided hotel.'''
# Some hotel API calls, a sensitive / expensive operation
return f"Booked a hotel at {hotel_name}."
agent = create_react_agent(
"openai:gpt-4.1",
tools=[book_hotel],
prompt="You are a hotel booking assistant.",
post_model_hook=InterruptToolNode(
book_hotel=HumanInterruptConfig(
allow_accept=True,
allow_edit=True,
allow_ignore=True,
allow_respond=True,
)
),
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": 1}}
response = agent.invoke(
{"messages": [{"role": "user", "content": "please book a hotel at the hilton inn in boston."}]},
config=config,
)
response = agent.invoke(Command(resume={"type": "accept"}), config=config)
```
"""
def __init__(self, **interrupt_policy: HumanInterruptConfig):
super().__init__(self._func, self._afunc)
self.interrupt_policy = interrupt_policy
def _interrupt(
self,
tool_call: ToolCall,
interrupt_config: HumanInterruptConfig,
) -> Union[ToolCall, ToolMessage]:
"""Interrupt before a tool call and ask for human input."""
call_id = tool_call["id"]
tool_name = tool_call["name"]
request = HumanInterrupt(
action_request=ActionRequest(
action=tool_name,
args=tool_call["args"],
),
config=interrupt_config,
description=f"Please review tool call for `{tool_name}` before execution.",
)
response = interrupt([request])
# resume provided by agent inbox as a list
response = response[0] if isinstance(response, list) else response
try:
response_type = response.get("type")
except AttributeError:
raise TypeError(
f"Unexpected resume value: {response}."
f"Expected a dict with `'type'` key."
)
if response_type == "accept" and interrupt_config["allow_accept"]:
return tool_call
elif response_type == "edit" and interrupt_config["allow_edit"]:
return ToolCall(
args=cast(ActionRequest, response)["args"]["args"],
name=tool_name,
id=call_id,
type="tool_call",
)
elif response_type == "response" and interrupt_config["allow_respond"]:
return ToolMessage(
content=cast(str, response["args"]),
name=tool_name,
tool_call_id=call_id,
status="error",
)
elif response_type == "ignore" and interrupt_config["allow_ignore"]:
return ToolMessage(
content=f"User ignored the tool call for `{tool_name}` with id {call_id}",
name=tool_name,
tool_call_id=call_id,
status="success",
)
allowed_types = [
type_name
for type_name, is_allowed in {
"accept": interrupt_config["allow_accept"],
"edit": interrupt_config["allow_edit"],
"response": interrupt_config["allow_respond"],
"ignore": interrupt_config["allow_ignore"],
}.items()
if is_allowed
]
raise ValueError(
f"Unexpected human response: {response}. "
f"Expected one with `'type'` in {allowed_types} based on {tool_name}'s interrupt configuration."
)
def _func(self, input: dict[str, Any]) -> Command:
ai_msg = input["messages"][-1]
tool_calls: list[ToolCall] = deepcopy(ai_msg.tool_calls) or []
tool_messages: list[ToolMessage] = []
for idx, tool_call in enumerate(tool_calls):
if interrupt_config := self.interrupt_policy.get(tool_call["name"]):
interrupt_result = self._interrupt(
tool_call=tool_call, interrupt_config=interrupt_config
)
if isinstance(interrupt_result, ToolMessage):
tool_messages.append(interrupt_result)
else:
tool_calls[idx] = interrupt_result
updated_ai_msg = ai_msg.copy(update={"tool_calls": tool_calls})
# conditional routing logic for post_model_hook will direct to the tools node
# or agent node depending on if there are pending tool calls
return {"messages": [updated_ai_msg, *tool_messages]}
async def _afunc(self, input: dict[str, Any]) -> Command:
return self._func(input)
+10 -7
View File
@@ -431,22 +431,25 @@ class ToolNode(RunnableCallable):
return tool_calls, input_type
else:
input_type = "list"
message: AnyMessage = input[-1]
messages = input
elif isinstance(input, dict) and (messages := input.get(self.messages_key, [])):
input_type = "dict"
message = messages[-1]
elif messages := getattr(input, self.messages_key, None):
elif messages := getattr(input, self.messages_key, []):
# Assume dataclass-like state that can coerce from dict
input_type = "dict"
message = messages[-1]
else:
raise ValueError("No message found in input")
if not isinstance(message, AIMessage):
raise ValueError("Last message is not an AIMessage")
try:
latest_ai_message = next(
m for m in reversed(messages) if isinstance(m, AIMessage)
)
except StopIteration:
raise ValueError("No AIMessage found in input")
tool_calls = [
self.inject_tool_args(call, input, store) for call in message.tool_calls
self.inject_tool_args(call, input, store)
for call in latest_ai_message.tool_calls
]
return tool_calls, input_type
@@ -0,0 +1,191 @@
import pytest
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.interrupt import HumanInterruptConfig, InterruptToolNode
from langgraph.types import Command
from tests.model import FakeToolCallingModel
def hello_tool(name: str) -> str:
"""Return a greeting for the provided person."""
return f"Hello, {name}!"
post_model_hook = InterruptToolNode(
hello_tool=HumanInterruptConfig(
allow_accept=True,
allow_edit=True,
allow_ignore=True,
allow_respond=True,
)
)
default_model = FakeToolCallingModel(
tool_calls=[
[
{
"name": "hello_tool",
"args": {"name": "lady gaga"},
"id": "some-random-id",
}
]
]
)
def test_interrupt_surfaced(
request: pytest.FixtureRequest,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
agent = create_react_agent(
default_model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
result = agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
interrupt_data = result["__interrupt__"]
assert interrupt_data[0].value == [
{
"action_request": {"action": "hello_tool", "args": {"name": "lady gaga"}},
"config": {
"allow_accept": True,
"allow_edit": True,
"allow_ignore": True,
"allow_respond": True,
},
"description": "Please review tool call for `hello_tool` before execution.",
}
]
response = agent.invoke(Command(resume={"type": "accept"}), config=config)
tool_message: ToolMessage = response["messages"][-2]
assert tool_message.content == "Hello, lady gaga!"
assert tool_message.name == "hello_tool"
@pytest.mark.parametrize(
"resume, expected_content",
[
({"type": "accept"}, "Hello, lady gaga!"),
(
{"type": "ignore"},
"User ignored the tool call for `hello_tool` with id some-random-id",
),
(
{
"type": "edit",
"args": {"action": "hello_tool", "args": {"name": "bruno mars"}},
},
"Hello, bruno mars!",
),
],
)
def test_interrupt_resume_variants(
request: pytest.FixtureRequest,
sync_checkpointer: BaseCheckpointSaver,
resume: dict,
expected_content: str,
) -> None:
agent = create_react_agent(
default_model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
response = agent.invoke(Command(resume=resume), config=config)
tool_message: ToolMessage = response["messages"][-2]
assert tool_message.name == "hello_tool"
assert tool_message.content == expected_content
if resume["type"] == "edit":
ai_msg = response["messages"][-1]
assert ai_msg.tool_calls == [
{
"name": "hello_tool",
"args": {"name": "lady gaga"},
"id": "some-random-id",
"type": "tool_call",
}
]
def test_resume_with_response(
request: pytest.FixtureRequest,
sync_checkpointer: BaseCheckpointSaver,
) -> None:
model = FakeToolCallingModel(
tool_calls=[
[
{
"name": "hello_tool",
"args": {"name": "lady gaga"},
"id": "some-random-id",
}
],
[
{
"name": "hello_tool",
"args": {"name": "bruno mars"},
"id": "some-random-id-2",
}
],
]
)
agent = create_react_agent(
model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
# Provide user response
agent.invoke(
Command(
resume={
"type": "response",
"args": "actually, please say hello to bruno mars",
}
),
config=config,
)
# Accept the updated call
response = agent.invoke(Command(resume={"type": "accept"}), config=config)
assert len(response["messages"]) == 6
tool_message: ToolMessage = response["messages"][-2]
assert tool_message.name == "hello_tool"
assert tool_message.content == "Hello, bruno mars!"
def test_resume_with_type_not_allowed(sync_checkpointer: BaseCheckpointSaver) -> None:
agent = create_react_agent(
default_model,
[hello_tool],
checkpointer=sync_checkpointer,
post_model_hook=post_model_hook,
)
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [("user", "Say hi to lady gaga!")]}, config)
with pytest.raises(ValueError) as exc_info:
agent.invoke(Command(resume={"type": "not-allowed"}), config=config)
assert (
str(exc_info.value)
== "Unexpected human response: {'type': 'not-allowed'}. Expected one with `'type'` in ['accept', 'edit', 'response', 'ignore'] based on hello_tool's interrupt configuration."
)
+141
View File
@@ -1399,3 +1399,144 @@ def test_pre_model_hook() -> None:
AIMessage(content="Hello!", id="1"),
]
}
def test_post_model_hook() -> None:
class FlagState(AgentState):
flag: bool
model = FakeToolCallingModel(tool_calls=[])
def post_model_hook(state: FlagState) -> dict[str, bool]:
return {"flag": True}
pmh_agent = create_react_agent(
model, [], post_model_hook=post_model_hook, state_schema=FlagState
)
assert "post_model_hook" in pmh_agent.nodes
result = pmh_agent.invoke({"messages": [HumanMessage("hi?")], "flag": False})
assert result["flag"] is True
events = list(pmh_agent.stream({"messages": [HumanMessage("hi?")], "flag": False}))
assert events == [
{
"agent": {
"messages": [
AIMessage(
content="hi?",
additional_kwargs={},
response_metadata={},
id="1",
)
]
}
},
{"post_model_hook": {"flag": True}},
]
def test_post_model_hook_with_structured_output() -> None:
class WeatherResponse(BaseModel):
temperature: float = Field(description="The temperature in fahrenheit")
tool_calls = [[{"args": {}, "id": "1", "name": "get_weather"}]]
def get_weather():
"""Get the weather"""
return "The weather is sunny and 75°F."
expected_structured_response = WeatherResponse(temperature=75)
model = FakeToolCallingModel(
tool_calls=tool_calls, structured_response=expected_structured_response
)
class State(AgentState):
flag: bool
structured_response: WeatherResponse
def post_model_hook(state: State) -> Union[dict[str, bool], Command]:
return {"flag": True}
agent = create_react_agent(
model,
[get_weather],
response_format=WeatherResponse,
post_model_hook=post_model_hook,
state_schema=State,
)
assert "post_model_hook" in agent.nodes
assert "generate_structured_response" in agent.nodes
response = agent.invoke(
{"messages": [HumanMessage("What's the weather?")], "flag": False}
)
assert response["flag"] is True
assert response["structured_response"] == expected_structured_response
events = list(
agent.stream({"messages": [HumanMessage("What's the weather?")], "flag": False})
)
assert "generate_structured_response" in events[-1]
assert events == [
{
"agent": {
"messages": [
AIMessage(
content="What's the weather?",
additional_kwargs={},
response_metadata={},
id="2",
tool_calls=[
{
"name": "get_weather",
"args": {},
"id": "1",
"type": "tool_call",
}
],
)
]
}
},
{"post_model_hook": {"flag": True}},
{
"tools": {
"messages": [
_AnyIdToolMessage(
content="The weather is sunny and 75°F.",
name="get_weather",
tool_call_id="1",
),
]
}
},
{
"agent": {
"messages": [
AIMessage(
content="What's the weather?-What's the weather?-The weather is sunny and 75°F.",
additional_kwargs={},
response_metadata={},
id="3",
tool_calls=[
{
"name": "get_weather",
"args": {},
"id": "1",
"type": "tool_call",
}
],
)
]
}
},
{"post_model_hook": {"flag": True}},
{
"generate_structured_response": {
"structured_response": WeatherResponse(temperature=75.0)
}
},
]
+16 -2
View File
@@ -1,5 +1,4 @@
version = 1
revision = 1
requires-python = ">=3.9"
resolution-markers = [
"python_full_version >= '3.12.4'",
@@ -431,17 +430,19 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.7"
version = "2.0.10"
source = { editable = "../checkpoint-sqlite" }
dependencies = [
{ name = "aiosqlite" },
{ name = "langgraph-checkpoint" },
{ name = "sqlite-vec" },
]
[package.metadata]
requires-dist = [
{ name = "aiosqlite", specifier = ">=0.20" },
{ name = "langgraph-checkpoint", editable = "../checkpoint" },
{ name = "sqlite-vec", specifier = ">=0.1.6" },
]
[package.metadata.requires-dev]
@@ -452,6 +453,7 @@ dev = [
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
{ name = "pytest-retry", specifier = ">=1.7.0" },
{ name = "pytest-watcher" },
{ name = "ruff" },
]
@@ -1070,6 +1072,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235 },
]
[[package]]
name = "sqlite-vec"
version = "0.1.6"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/88/ed/aabc328f29ee6814033d008ec43e44f2c595447d9cccd5f2aabe60df2933/sqlite_vec-0.1.6-py3-none-macosx_10_6_x86_64.whl", hash = "sha256:77491bcaa6d496f2acb5cc0d0ff0b8964434f141523c121e313f9a7d8088dee3", size = 164075 },
{ url = "https://files.pythonhosted.org/packages/a7/57/05604e509a129b22e303758bfa062c19afb020557d5e19b008c64016704e/sqlite_vec-0.1.6-py3-none-macosx_11_0_arm64.whl", hash = "sha256:fdca35f7ee3243668a055255d4dee4dea7eed5a06da8cad409f89facf4595361", size = 165242 },
{ url = "https://files.pythonhosted.org/packages/f2/48/dbb2cc4e5bad88c89c7bb296e2d0a8df58aab9edc75853728c361eefc24f/sqlite_vec-0.1.6-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:7b0519d9cd96164cd2e08e8eed225197f9cd2f0be82cb04567692a0a4be02da3", size = 103704 },
{ url = "https://files.pythonhosted.org/packages/80/76/97f33b1a2446f6ae55e59b33869bed4eafaf59b7f4c662c8d9491b6a714a/sqlite_vec-0.1.6-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux1_x86_64.whl", hash = "sha256:823b0493add80d7fe82ab0fe25df7c0703f4752941aee1c7b2b02cec9656cb24", size = 151556 },
{ url = "https://files.pythonhosted.org/packages/6a/98/e8bc58b178266eae2fcf4c9c7a8303a8d41164d781b32d71097924a6bebe/sqlite_vec-0.1.6-py3-none-win_amd64.whl", hash = "sha256:c65bcfd90fa2f41f9000052bcb8bb75d38240b2dae49225389eca6c3136d3f0c", size = 281540 },
]
[[package]]
name = "tenacity"
version = "9.1.2"
+15 -11
View File
@@ -1,6 +1,6 @@
{
"name": "@langchain/langgraph-sdk",
"version": "0.0.74",
"version": "0.0.77",
"description": "Client library for interacting with the LangGraph API",
"type": "module",
"packageManager": "yarn@1.22.19",
@@ -10,7 +10,7 @@
"prepack": "yarn run build",
"format": "prettier --write src",
"lint": "prettier --check src && tsc --noEmit",
"test": "NODE_OPTIONS=--experimental-vm-modules jest --testPathIgnorePatterns=\\.int\\.test.ts",
"test": "vitest",
"typedoc": "typedoc && typedoc src/react/index.ts --out docs/react --options typedoc.react.json && typedoc src/auth/index.ts --out docs/auth --options typedoc.auth.json"
},
"main": "index.js",
@@ -22,28 +22,32 @@
"uuid": "^9.0.0"
},
"devDependencies": {
"@jest/globals": "^29.7.0",
"@langchain/core": "^0.3.31",
"@langchain/scripts": "^0.1.4",
"@testing-library/dom": "^10.4.0",
"@testing-library/jest-dom": "^6.6.3",
"@testing-library/react": "^16.3.0",
"@testing-library/user-event": "^14.6.1",
"@tsconfig/recommended": "^1.0.2",
"@types/jest": "^29.5.12",
"@types/node": "^20.12.12",
"@types/uuid": "^9.0.1",
"@types/react": "^19.0.8",
"@types/react-dom": "^19.0.3",
"@types/uuid": "^9.0.1",
"@vitejs/plugin-react": "^4.4.1",
"concat-md": "^0.5.1",
"jest": "^29.7.0",
"jsdom": "^26.1.0",
"msw": "^2.8.2",
"prettier": "^3.2.5",
"ts-jest": "^29.1.2",
"react": "^19.0.0",
"react-dom": "^19.0.0",
"typedoc": "^0.27.7",
"typedoc-plugin-markdown": "^4.4.2",
"typescript": "^5.4.5",
"react": "^19.0.0",
"react-dom": "^19.0.0"
"vitest": "^3.1.3"
},
"peerDependencies": {
"react": "^18 || ^19",
"@langchain/core": ">=0.2.31 <0.4.0"
"@langchain/core": ">=0.2.31 <0.4.0",
"react": "^18 || ^19"
},
"peerDependenciesMeta": {
"react": {
+107 -28
View File
@@ -68,6 +68,24 @@ export function getApiKey(apiKey?: string): string | undefined {
return undefined;
}
const REGEX_RUN_METADATA =
/(\/threads\/(?<thread_id>.+))?\/runs\/(?<run_id>.+)/;
function getRunMetadataFromResponse(
response: Response,
): { run_id: string; thread_id?: string } | undefined {
const contentLocation = response.headers.get("Content-Location");
if (!contentLocation) return undefined;
const match = REGEX_RUN_METADATA.exec(contentLocation);
if (!match?.groups?.run_id) return undefined;
return {
run_id: match.groups.run_id,
thread_id: match.groups.thread_id || undefined,
};
}
export interface ClientConfig {
apiUrl?: string;
apiKey?: string;
@@ -130,6 +148,7 @@ class BaseClient {
json?: unknown;
params?: Record<string, unknown>;
timeoutMs?: number | null;
withResponse?: boolean;
},
): [url: URL, init: RequestInit] {
const mutatedOptions = {
@@ -146,6 +165,10 @@ class BaseClient {
delete mutatedOptions.json;
}
if (mutatedOptions.withResponse) {
delete mutatedOptions.withResponse;
}
let timeoutSignal: AbortSignal | null = null;
if (typeof options?.timeoutMs !== "undefined") {
if (options.timeoutMs != null) {
@@ -175,6 +198,17 @@ class BaseClient {
return [targetUrl, mutatedOptions];
}
protected async fetch<T>(
path: string,
options: RequestInit & {
json?: unknown;
params?: Record<string, unknown>;
timeoutMs?: number | null;
signal?: AbortSignal;
withResponse: true;
},
): Promise<[T, Response]>;
protected async fetch<T>(
path: string,
options?: RequestInit & {
@@ -182,15 +216,36 @@ class BaseClient {
params?: Record<string, unknown>;
timeoutMs?: number | null;
signal?: AbortSignal;
withResponse?: false;
},
): Promise<T> {
): Promise<T>;
protected async fetch<T>(
path: string,
options?: RequestInit & {
json?: unknown;
params?: Record<string, unknown>;
timeoutMs?: number | null;
signal?: AbortSignal;
withResponse?: boolean;
},
): Promise<T | [T, Response]> {
const response = await this.asyncCaller.fetch(
...this.prepareFetchOptions(path, options),
);
if (response.status === 202 || response.status === 204) {
return undefined as T;
const body = (() => {
if (response.status === 202 || response.status === 204) {
return undefined as T;
}
return response.json() as Promise<T>;
})();
if (options?.withResponse) {
return [await body, response];
}
return response.json() as T;
return body;
}
}
@@ -839,6 +894,7 @@ export class RunsClient<
metadata: payload?.metadata,
stream_mode: payload?.streamMode,
stream_subgraphs: payload?.streamSubgraphs,
stream_resumable: payload?.streamResumable,
feedback_keys: payload?.feedbackKeys,
assistant_id: assistantId,
interrupt_before: payload?.interruptBefore,
@@ -856,6 +912,7 @@ export class RunsClient<
const endpoint =
threadId == null ? `/runs/stream` : `/threads/${threadId}/runs/stream`;
const response = await this.asyncCaller.fetch(
...this.prepareFetchOptions(endpoint, {
method: "POST",
@@ -865,6 +922,9 @@ export class RunsClient<
}),
);
const runMetadata = getRunMetadataFromResponse(response);
if (runMetadata) payload?.onRunCreated?.(runMetadata);
const stream: ReadableStream<{ event: any; data: any }> = (
response.body || new ReadableStream({ start: (ctrl) => ctrl.close() })
)
@@ -894,6 +954,7 @@ export class RunsClient<
metadata: payload?.metadata,
stream_mode: payload?.streamMode,
stream_subgraphs: payload?.streamSubgraphs,
stream_resumable: payload?.streamResumable,
assistant_id: assistantId,
interrupt_before: payload?.interruptBefore,
interrupt_after: payload?.interruptAfter,
@@ -905,11 +966,18 @@ export class RunsClient<
if_not_exists: payload?.ifNotExists,
checkpoint_during: payload?.checkpointDuring,
};
return this.fetch<Run>(`/threads/${threadId}/runs`, {
const [run, response] = await this.fetch<Run>(`/threads/${threadId}/runs`, {
method: "POST",
json,
signal: payload?.signal,
withResponse: true,
});
const runMetadata = getRunMetadataFromResponse(response);
if (runMetadata) payload?.onRunCreated?.(runMetadata);
return run;
}
/**
@@ -980,27 +1048,30 @@ export class RunsClient<
};
const endpoint =
threadId == null ? `/runs/wait` : `/threads/${threadId}/runs/wait`;
const response = await this.fetch<ThreadState["values"]>(endpoint, {
const [run, response] = await this.fetch<ThreadState["values"]>(endpoint, {
method: "POST",
json,
timeoutMs: null,
signal: payload?.signal,
withResponse: true,
});
const runMetadata = getRunMetadataFromResponse(response);
if (runMetadata) payload?.onRunCreated?.(runMetadata);
const raiseError =
payload?.raiseError !== undefined ? payload.raiseError : true;
if (
raiseError &&
"__error__" in response &&
typeof response.__error__ === "object" &&
response.__error__ &&
"error" in response.__error__ &&
"message" in response.__error__
"__error__" in run &&
typeof run.__error__ === "object" &&
run.__error__ &&
"error" in run.__error__ &&
"message" in run.__error__
) {
throw new Error(
`${response.__error__?.error}: ${response.__error__?.message}`,
);
throw new Error(`${run.__error__?.error}: ${run.__error__?.message}`);
}
return response;
return run;
}
/**
@@ -1095,13 +1166,12 @@ export class RunsClient<
/**
* Stream output from a run in real-time, until the run is done.
* Output is not buffered, so any output produced before this call will
* not be received here.
*
* @param threadId The ID of the thread.
* @param threadId The ID of the thread. Can be set to `null` | `undefined` for stateless runs.
* @param runId The ID of the run.
* @param options Additional options for controlling the stream behavior:
* - signal: An AbortSignal that can be used to cancel the stream request
* - lastEventId: The ID of the last event received. Can be used to reconnect to a stream without losing events.
* - cancelOnDisconnect: When true, automatically cancels the run if the client disconnects from the stream
* - streamMode: Controls what types of events to receive from the stream (can be a single mode or array of modes)
* Must be a subset of the stream modes passed when creating the run. Background runs default to having the union of all
@@ -1109,16 +1179,17 @@ export class RunsClient<
* @returns An async generator yielding stream parts.
*/
async *joinStream(
threadId: string,
threadId: string | undefined | null,
runId: string,
options?:
| {
signal?: AbortSignal;
cancelOnDisconnect?: boolean;
lastEventId?: string;
streamMode?: StreamMode | StreamMode[];
}
| AbortSignal,
): AsyncGenerator<{ event: StreamEvent; data: any }> {
): AsyncGenerator<{ id?: string; event: StreamEvent; data: any }> {
const opts =
typeof options === "object" &&
options != null &&
@@ -1127,15 +1198,23 @@ export class RunsClient<
: options;
const response = await this.asyncCaller.fetch(
...this.prepareFetchOptions(`/threads/${threadId}/runs/${runId}/stream`, {
method: "GET",
timeoutMs: null,
signal: opts?.signal,
params: {
cancel_on_disconnect: opts?.cancelOnDisconnect ? "1" : "0",
stream_mode: opts?.streamMode,
...this.prepareFetchOptions(
threadId != null
? `/threads/${threadId}/runs/${runId}/stream`
: `/runs/${runId}/stream`,
{
method: "GET",
timeoutMs: null,
signal: opts?.signal,
headers: opts?.lastEventId
? { "Last-Event-ID": opts.lastEventId }
: undefined,
params: {
cancel_on_disconnect: opts?.cancelOnDisconnect ? "1" : "0",
stream_mode: opts?.streamMode,
},
},
}),
),
);
const stream: ReadableStream<{ event: string; data: any }> = (
+7
View File
@@ -613,6 +613,12 @@ interface SubmitOptions<
optimisticValues?:
| Partial<StateType>
| ((prev: StateType) => Partial<StateType>);
/**
* Whether or not to stream the nodes of any subgraphs called
* by the assistant.
* @default false
*/
streamSubgraphs?: boolean;
}
export function useStream<
@@ -868,6 +874,7 @@ export function useStream<
checkpoint,
streamMode,
streamSubgraphs: submitOptions?.streamSubgraphs,
}) as AsyncGenerator<EventStreamEvent>;
let streamError: StreamError | undefined;
+71 -67
View File
@@ -1,74 +1,78 @@
/* eslint-disable no-process-env */
/* eslint-disable @typescript-eslint/no-explicit-any */
import { jest } from "@jest/globals";
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { Client } from "../client.js";
import { overrideFetchImplementation } from "../singletons/fetch.js";
describe.each([[""], ["mocked"]])("Client uses %s fetch", (description) => {
let globalFetchMock: jest.Mock;
let overriddenFetch: jest.Mock;
let expectedFetchMock: jest.Mock;
let unexpectedFetchMock: jest.Mock;
describe.each([["global"], ["mocked"]])(
"Client uses %s fetch",
(description: string) => {
let globalFetchMock: ReturnType<typeof vi.fn>;
let overriddenFetch: ReturnType<typeof vi.fn>;
beforeEach(() => {
globalFetchMock = jest.fn(() =>
Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
batch_ingest_config: {
use_multipart_endpoint: true,
},
}),
text: () => Promise.resolve(""),
}),
);
overriddenFetch = jest.fn(() =>
Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
batch_ingest_config: {
use_multipart_endpoint: true,
},
}),
text: () => Promise.resolve(""),
}),
);
expectedFetchMock =
description === "mocked" ? overriddenFetch : globalFetchMock;
unexpectedFetchMock =
description === "mocked" ? globalFetchMock : overriddenFetch;
let expectedFetchMock: ReturnType<typeof vi.fn>;
let unexpectedFetchMock: ReturnType<typeof vi.fn>;
if (description === "mocked") {
overrideFetchImplementation(overriddenFetch);
} else {
overrideFetchImplementation(globalFetchMock);
}
// Mock global fetch
(globalThis as any).fetch = globalFetchMock;
});
beforeEach(() => {
globalFetchMock = vi.fn(() =>
Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
batch_ingest_config: {
use_multipart_endpoint: true,
},
}),
text: () => Promise.resolve(""),
headers: new Headers({}),
}),
);
overriddenFetch = vi.fn(() =>
Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
batch_ingest_config: {
use_multipart_endpoint: true,
},
}),
text: () => Promise.resolve(""),
headers: new Headers({}),
}),
);
expectedFetchMock =
description === "mocked" ? overriddenFetch : globalFetchMock;
unexpectedFetchMock =
description === "mocked" ? globalFetchMock : overriddenFetch;
afterEach(() => {
jest.restoreAllMocks();
});
describe("createRuns", () => {
it("should create an example with the given input and generation", async () => {
const client = new Client({ apiKey: "test-api-key" });
const thread = await client.threads.create();
expect(expectedFetchMock).toHaveBeenCalledTimes(1);
expect(unexpectedFetchMock).not.toHaveBeenCalled();
jest.clearAllMocks(); // Clear all mocks before the next operation
// Then clear & run the function
await client.runs.create(thread.thread_id, "somegraph", {
input: { foo: "bar" },
});
expect(expectedFetchMock).toHaveBeenCalledTimes(1);
expect(unexpectedFetchMock).not.toHaveBeenCalled();
if (description === "mocked") {
overrideFetchImplementation(overriddenFetch);
} else {
overrideFetchImplementation(globalFetchMock);
}
// Mock global fetch
(globalThis as any).fetch = globalFetchMock;
});
});
});
afterEach(() => {
vi.restoreAllMocks();
});
describe("createRuns", () => {
it("should create an example with the given input and generation", async () => {
const client = new Client({ apiKey: "test-api-key" });
const thread = await client.threads.create();
expect(expectedFetchMock).toHaveBeenCalledTimes(1);
expect(unexpectedFetchMock).not.toHaveBeenCalled();
vi.clearAllMocks(); // Clear all mocks before the next operation
// Then clear & run the function
await client.runs.create(thread.thread_id, "somegraph", {
input: { foo: "bar" },
});
expect(expectedFetchMock).toHaveBeenCalledTimes(1);
expect(unexpectedFetchMock).not.toHaveBeenCalled();
});
});
},
);
+1
View File
@@ -1,3 +1,4 @@
import { describe, test, expect } from "vitest";
import { Readable } from "node:stream";
import { IterableReadableStream } from "../utils/stream.js";
import { BytesLineDecoder, SSEDecoder } from "../utils/sse.js";
+443
View File
@@ -0,0 +1,443 @@
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { render, screen, waitFor } from "@testing-library/react";
import { userEvent } from "@testing-library/user-event";
import { setupServer } from "msw/node";
import { http, HttpResponse } from "msw";
import { useStream } from "../react/stream.js";
import "@testing-library/jest-dom/vitest";
function TestChatComponent() {
const { messages, isLoading, error, submit, stop } = useStream({
assistantId: "test-assistant",
apiKey: "test-api-key",
});
return (
<div>
<div data-testid="messages">
{messages.map((msg, i) => (
<div key={msg.id ?? i} data-testid={`message-${i}`}>
{typeof msg.content === "string"
? msg.content
: JSON.stringify(msg.content)}
</div>
))}
</div>
<div data-testid="loading">
{isLoading ? "Loading..." : "Not loading"}
</div>
{error ? <div data-testid="error">{String(error)}</div> : null}
<button
data-testid="submit"
onClick={() =>
submit({ messages: [{ content: "Hello", type: "human" }] })
}
>
Send
</button>
<button data-testid="stop" onClick={stop}>
Stop
</button>
</div>
);
}
// Mock server setup
const server = setupServer(
// Mock thread creation
http.post("*/threads", () => {
return HttpResponse.json({ thread_id: "test-thread-id" });
}),
// Mock stream endpoint
http.post("*/threads/:threadId/runs/stream", async () => {
const encoder = new TextEncoder();
const sendSSE = (event: string, data: unknown) =>
encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`);
const stream = new ReadableStream({
async start(controller) {
await new Promise((resolve) => setTimeout(resolve, 10));
controller.enqueue(
sendSSE("metadata", {
run_id: "1f03278a-1734-6518-80a4-3390db59f960",
attempt: 1,
}),
);
controller.enqueue(
sendSSE("values", {
messages: [
{
content: "Hey",
additional_kwargs: {},
response_metadata: {},
type: "human",
name: null,
id: "2d8c0d9f-a614-4e44-b474-6a56e9471cf5",
example: false,
},
],
}),
);
controller.enqueue(
sendSSE("messages", [
{
content: "",
additional_kwargs: {},
response_metadata: { model_name: "claude-3-7-sonnet-latest" },
type: "AIMessageChunk",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
tool_calls: [],
invalid_tool_calls: [],
tool_call_chunks: [],
},
{ run_attempt: 1 },
]),
);
controller.enqueue(
sendSSE("messages", [
{
content: "Hello",
additional_kwargs: {},
response_metadata: { model_name: "claude-3-7-sonnet-latest" },
type: "AIMessageChunk",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
tool_calls: [],
invalid_tool_calls: [],
tool_call_chunks: [],
},
{ run_attempt: 1 },
]),
);
controller.enqueue(
sendSSE("messages", [
{
content: "! How can I assist you today?",
additional_kwargs: {},
response_metadata: { model_name: "claude-3-7-sonnet-latest" },
type: "AIMessageChunk",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
tool_calls: [],
invalid_tool_calls: [],
tool_call_chunks: [],
},
{ run_attempt: 1 },
]),
);
controller.enqueue(
sendSSE("messages", [
{
content: "",
additional_kwargs: {},
response_metadata: {
stop_reason: "end_turn",
stop_sequence: null,
},
type: "AIMessageChunk",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
tool_calls: [],
invalid_tool_calls: [],
tool_call_chunks: [],
},
{ run_attempt: 1 },
]),
);
controller.enqueue(
sendSSE("values", {
messages: [
{
content: "Hey",
additional_kwargs: {},
response_metadata: {},
type: "human",
name: null,
id: "2d8c0d9f-a614-4e44-b474-6a56e9471cf5",
example: false,
},
{
content: "Hello! How can I assist you today?",
additional_kwargs: {},
response_metadata: {
model_name: "claude-3-7-sonnet-latest",
stop_reason: "end_turn",
stop_sequence: null,
},
type: "ai",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
tool_calls: [],
invalid_tool_calls: [],
},
],
}),
);
controller.close();
},
});
server.use(
http.post("*/threads/:threadId/history", () => {
return HttpResponse.json([
{
values: {
messages: [
{
content: "Hey",
additional_kwargs: {},
response_metadata: {},
type: "human",
name: null,
id: "2d8c0d9f-a614-4e44-b474-6a56e9471cf5",
example: false,
},
{
content: "Hello! How can I assist you today?",
additional_kwargs: {},
response_metadata: {
model_name: "claude-3-7-sonnet-latest",
stop_reason: "end_turn",
stop_sequence: null,
},
type: "ai",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
example: false,
tool_calls: [],
invalid_tool_calls: [],
},
],
},
next: [],
tasks: [],
metadata: {
run_attempt: 1,
source: "loop",
writes: {
agent: {
messages: [
{
content: "Hello! How can I assist you today?",
additional_kwargs: {},
response_metadata: {
model_name: "claude-3-7-sonnet-latest",
stop_reason: "end_turn",
stop_sequence: null,
},
type: "ai",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
example: false,
tool_calls: [],
invalid_tool_calls: [],
},
],
},
},
step: 1,
parents: {},
},
created_at: "2025-05-16T17:10:16.987537+00:00",
checkpoint: {
checkpoint_id: "1f03278a-38cf-6c68-8001-22b77ac43ff6",
thread_id: "b06fd92a-955c-446e-b233-7977716c4a9c",
checkpoint_ns: "",
},
parent_checkpoint: {
checkpoint_id: "1f03278a-206b-67c6-8000-ac34a0872e1a",
thread_id: "b06fd92a-955c-446e-b233-7977716c4a9c",
checkpoint_ns: "",
},
checkpoint_id: "1f03278a-38cf-6c68-8001-22b77ac43ff6",
parent_checkpoint_id: "1f03278a-206b-67c6-8000-ac34a0872e1a",
},
{
values: {
messages: [
{
content: "Hey",
additional_kwargs: {},
response_metadata: {},
type: "human",
name: null,
id: "2d8c0d9f-a614-4e44-b474-6a56e9471cf5",
example: false,
},
],
},
next: ["agent"],
tasks: [
{
id: "e1b7b52b-a78e-4b32-0c89-e06bf46405ed",
name: "agent",
path: ["__pregel_pull", "agent"],
error: null,
interrupts: [],
checkpoint: null,
state: null,
result: {
messages: [
{
content: "Hello! How can I assist you today?",
additional_kwargs: {},
response_metadata: {
model_name: "claude-3-7-sonnet-latest",
stop_reason: "end_turn",
stop_sequence: null,
},
type: "ai",
name: null,
id: "run-3e90ba6a-71d6-49e7-94a8-6bcac2fd0f40",
example: false,
tool_calls: [],
invalid_tool_calls: [],
},
],
},
},
],
metadata: {
run_attempt: 1,
},
created_at: "2025-05-16T17:10:14.429889+00:00",
checkpoint: {
checkpoint_id: "1f03278a-206b-67c6-8000-ac34a0872e1a",
thread_id: "b06fd92a-955c-446e-b233-7977716c4a9c",
checkpoint_ns: "",
},
parent_checkpoint: {
checkpoint_id: "1f03278a-2067-6590-bfff-3fb740466fc3",
thread_id: "b06fd92a-955c-446e-b233-7977716c4a9c",
checkpoint_ns: "",
},
checkpoint_id: "1f03278a-206b-67c6-8000-ac34a0872e1a",
parent_checkpoint_id: "1f03278a-2067-6590-bfff-3fb740466fc3",
},
{
values: {
messages: [],
},
next: ["__start__"],
tasks: [
{
id: "291af033-2ddc-3320-8bbc-28060057cae5",
name: "__start__",
path: ["__pregel_pull", "__start__"],
error: null,
interrupts: [],
checkpoint: null,
state: null,
result: {
messages: [
{
id: "2d8c0d9f-a614-4e44-b474-6a56e9471cf5",
type: "human",
content: "Hey",
},
],
},
},
],
metadata: {
run_attempt: 1,
source: "input",
writes: {
__start__: {
messages: [
{
id: "2d8c0d9f-a614-4e44-b474-6a56e9471cf5",
type: "human",
content: "Hey",
},
],
},
},
step: -1,
parents: {},
},
created_at: "2025-05-16T17:10:14.428191+00:00",
checkpoint: {
checkpoint_id: "1f03278a-2067-6590-bfff-3fb740466fc3",
thread_id: "b06fd92a-955c-446e-b233-7977716c4a9c",
checkpoint_ns: "",
},
parent_checkpoint: null,
checkpoint_id: "1f03278a-2067-6590-bfff-3fb740466fc3",
parent_checkpoint_id: null,
},
]);
}),
);
return new HttpResponse(stream, {
headers: { "Content-Type": "text/event-stream" },
});
}),
);
server.use;
describe("useStream", () => {
beforeEach(() => server.listen());
afterEach(() => {
server.resetHandlers();
server.close();
vi.clearAllMocks();
});
it("renders initial state correctly", () => {
render(<TestChatComponent />);
expect(screen.getByTestId("loading")).toHaveTextContent("Not loading");
expect(screen.getByTestId("messages")).toBeEmptyDOMElement();
expect(screen.queryByTestId("error")).not.toBeInTheDocument();
});
it("handles message submission and streaming", async () => {
const user = userEvent.setup();
render(<TestChatComponent />);
// Check loading state
await user.click(screen.getByTestId("submit"));
expect(screen.getByTestId("loading")).toHaveTextContent("Loading...");
// Wait for messages to appear
await waitFor(() => {
expect(screen.getByTestId("message-0")).toHaveTextContent("Hey");
expect(screen.getByTestId("message-1")).toHaveTextContent(
"Hello! How can I assist you today?",
);
});
// Check final state
expect(screen.getByTestId("loading")).toHaveTextContent("Not loading");
});
it("handles stop functionality", async () => {
const user = userEvent.setup();
render(<TestChatComponent />);
// Start streaming and stop immediately
await user.click(screen.getByTestId("submit"));
await user.click(screen.getByTestId("stop"));
// Check loading state is reset
await waitFor(() => {
expect(screen.getByTestId("loading")).toHaveTextContent("Not loading");
});
});
});
+20 -6
View File
@@ -24,15 +24,21 @@ type MessageTupleMetadata = {
[key: string]: unknown;
};
type AsSubgraph<TEvent extends { event: string; data: unknown }> = {
event: TEvent["event"] | `${TEvent["event"]}|${string}`;
data: TEvent["data"];
};
type AsSubgraph<TEvent extends { id?: string; event: string; data: unknown }> =
{
id?: TEvent["id"];
event: TEvent["event"] | `${TEvent["event"]}|${string}`;
data: TEvent["data"];
};
/**
* Stream event with values after completion of each step.
*/
export type ValuesStreamEvent<StateType> = { event: "values"; data: StateType };
export type ValuesStreamEvent<StateType> = {
id?: string;
event: "values";
data: StateType;
};
/** @internal */
export type SubgraphValuesStreamEvent<StateType> = AsSubgraph<
@@ -57,6 +63,7 @@ export type SubgraphMessagesTupleStreamEvent =
* Metadata stream event with information about the run and thread
*/
export type MetadataStreamEvent = {
id?: string;
event: "metadata";
data: { run_id: string; thread_id: string };
};
@@ -65,6 +72,7 @@ export type MetadataStreamEvent = {
* Stream event with error information.
*/
export type ErrorStreamEvent = {
id?: string;
event: "error";
data: { error: string; message: string };
};
@@ -78,6 +86,7 @@ export type SubgraphErrorStreamEvent = AsSubgraph<ErrorStreamEvent>;
* produced the update as well as the update.
*/
export type UpdatesStreamEvent<UpdateType> = {
id?: string;
event: "updates";
data: { [node: string]: UpdateType };
};
@@ -96,14 +105,17 @@ export type CustomStreamEvent<T> = { event: "custom"; data: T };
export type SubgraphCustomStreamEvent<T> = AsSubgraph<CustomStreamEvent<T>>;
type MessagesMetadataStreamEvent = {
id?: string;
event: "messages/metadata";
data: { [messageId: string]: { metadata: unknown } };
};
type MessagesCompleteStreamEvent = {
id?: string;
event: "messages/complete";
data: Message[];
};
type MessagesPartialStreamEvent = {
id?: string;
event: "messages/partial";
data: Message[];
};
@@ -126,7 +138,7 @@ export type SubgraphMessagesStreamEvent =
/**
* Stream event with detailed debug information.
*/
export type DebugStreamEvent = { event: "debug"; data: unknown };
export type DebugStreamEvent = { id?: string; event: "debug"; data: unknown };
/** @internal */
export type SubgraphDebugStreamEvent = AsSubgraph<DebugStreamEvent>;
@@ -135,6 +147,7 @@ export type SubgraphDebugStreamEvent = AsSubgraph<DebugStreamEvent>;
* Stream event with events occurring during execution.
*/
export type EventsStreamEvent = {
id?: string;
event: "events";
data: {
event:
@@ -157,6 +170,7 @@ export type SubgraphEventsStreamEvent = AsSubgraph<EventsStreamEvent>;
* the `RunsStreamPayload` to receive this event.
*/
export type FeedbackStreamEvent = {
id?: string;
event: "feedback";
data: { [feedbackKey: string]: string };
};
+17
View File
@@ -135,6 +135,11 @@ interface RunsInvokePayload {
* One or more commands to invoke the graph with.
*/
command?: Command;
/**
* Callback when a run is created.
*/
onRunCreated?: (params: { run_id: string; thread_id?: string }) => void;
}
export interface RunsStreamPayload<
@@ -151,6 +156,12 @@ export interface RunsStreamPayload<
*/
streamSubgraphs?: TSubgraphs;
/**
* Whether the stream is considered resumable.
* If true, the stream can be resumed and replayed in its entirety even after disconnection.
*/
streamResumable?: boolean;
/**
* Pass one or more feedbackKeys if you want to request short-lived signed URLs
* for submitting feedback to LangSmith with this key for this run.
@@ -168,6 +179,12 @@ export interface RunsCreatePayload extends RunsInvokePayload {
* Stream output from subgraphs. By default, streams only the top graph.
*/
streamSubgraphs?: boolean;
/**
* Whether the stream is considered resumable.
* If true, the stream can be resumed and replayed in its entirety even after disconnection.
*/
streamResumable?: boolean;
}
export interface CronsCreatePayload extends RunsCreatePayload {
+3
View File
@@ -93,6 +93,7 @@ export class BytesLineDecoder extends TransformStream<Uint8Array, Uint8Array> {
}
interface StreamPart {
id: string | undefined;
event: string;
data: unknown;
}
@@ -113,6 +114,7 @@ export class SSEDecoder extends TransformStream<Uint8Array, StreamPart> {
if (!event && !data.length && !lastEventId && retry == null) return;
const sse = {
id: lastEventId || undefined,
event,
data: data.length ? decodeArraysToJson(decoder, data) : null,
};
@@ -151,6 +153,7 @@ export class SSEDecoder extends TransformStream<Uint8Array, StreamPart> {
flush(controller) {
if (event) {
controller.enqueue({
id: lastEventId || undefined,
event,
data: data.length ? decodeArraysToJson(decoder, data) : null,
});
+11
View File
@@ -0,0 +1,11 @@
import react from "@vitejs/plugin-react";
import { defineConfig } from "vitest/config";
export default defineConfig({
plugins: [react()],
test: {
environment: "jsdom",
globals: true,
include: ["src/**/*.test.ts", "src/**/*.test.tsx"],
},
});
+1265 -1579
View File
File diff suppressed because it is too large Load Diff