Compare commits

..
Author SHA1 Message Date
William Fu-Hinthorn 8dba54fa0d Add windows CLI testing 2024-11-20 17:34:16 -08:00
Kangxu LiuandGitHub 5247952b31 [CLI] Fix relative path issue on Windows (#2480) 2024-11-20 17:25:43 -08:00
76 changed files with 4816 additions and 6609 deletions
+60 -2
View File
@@ -14,7 +14,7 @@ jobs:
python-version:
- "3.10"
- "3.11"
name: "CLI integration test"
name: "CLI integration test (Linux)"
defaults:
run:
working-directory: libs/cli
@@ -71,4 +71,62 @@ jobs:
working-directory: libs/cli/js-examples
run: |
langgraph build -t langgraph-test-e
windows-build:
runs-on: windows-latest
name: "CLI integration test (Windows)"
defaults:
run:
working-directory: libs/cli
shell: bash
steps:
- uses: actions/checkout@v4
- name: Get changed files
id: changed-files
uses: Ana06/get-changed-files@v2.3.0
with:
filter: "libs/cli/**"
- name: Set up Python 3.11 + Poetry ${{ env.POETRY_VERSION }}
if: steps.changed-files.outputs.all
uses: "./.github/actions/poetry_setup"
with:
python-version: "3.11"
poetry-version: ${{ env.POETRY_VERSION }}
cache-key: integration-test-cli-windows
- name: Setup env
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples
run: cp .env.example .env
- name: Install cli globally
if: steps.changed-files.outputs.all
run: pip install -e .
- name: Build and test service A
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples
run: |
langgraph build -t langgraph-test-a --base-image "langchain/langgraph-trial"
cp .env.example .envg
timeout 60 python ../../../.github/scripts/run_langgraph_cli_test.py -c langgraph.json -t langgraph-test-a
- name: Build and test service B
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graphs
run: |
langgraph build -t langgraph-test-b --base-image "langchain/langgraph-trial"
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-b
- name: Build and test service C
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graphs_reqs_a
run: |
langgraph build -t langgraph-test-c --base-image "langchain/langgraph-trial"
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-c
- name: Build and test service D
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graphs_reqs_b
run: |
langgraph build -t langgraph-test-d --base-image "langchain/langgraph-trial"
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-d
- name: Build JS service
if: steps.changed-files.outputs.all
working-directory: libs/cli/js-examples
run: |
langgraph build -t langgraph-test-e
-2
View File
@@ -88,7 +88,6 @@ jobs:
--check-links-ignore "https://(api|web|docs)\.smith\.langchain\.com/.*" \
--check-links-ignore "https://x.com/.*" \
--check-links-ignore "https://github\.com/.*" \
--check-links-ignore "http://localhost:8123/.*" \
--check-links-ignore "/.*\.(ipynb|html)$" \
--check-links-ignore "https://python\.langchain\.com/.*" \
--check-links-ignore "https://openai\.com/.*" \
@@ -105,7 +104,6 @@ jobs:
echo "Running link check on HTML files matching changed notebook files..."
poetry run pytest -v \
--check-links-ignore "https://(api|web|docs)\.smith\.langchain\.com/.*" \
--check-links-ignore "http://localhost:8123/.*" \
--check-links-ignore "https://x.com/.*" \
--check-links-ignore "https://github\.com/.*" \
--check-links-ignore "/.*\.(ipynb|html)$" \
+1 -1
View File
@@ -238,7 +238,7 @@ final_state["messages"][-1].content
* [How-to Guides](https://langchain-ai.github.io/langgraph/how-tos/): Accomplish specific things within LangGraph, from streaming, to adding memory & persistence, to common design patterns (branching, subgraphs, etc.), these are the place to go if you want to copy and run a specific code snippet.
* [Conceptual Guides](https://langchain-ai.github.io/langgraph/concepts/high_level/): In-depth explanations of the key concepts and principles behind LangGraph, such as nodes, edges, state and more.
* [API Reference](https://langchain-ai.github.io/langgraph/reference/graphs/): Review important classes and methods, simple examples of how to use the graph and checkpointing APIs, higher-level prebuilt components and more.
* [LangGraph Platform](https://langchain-ai.github.io/langgraph/concepts/#langgraph-platform): LangGraph Platform is a commercial solution for deploying agentic applications in production, built on the open-source LangGraph framework.
* [Cloud (beta)](https://langchain-ai.github.io/langgraph/cloud/): With one click, deploy LangGraph applications to LangGraph Cloud.
## Contributing
+3 -3
View File
@@ -8,9 +8,9 @@ If you want to learn how to build an agent like this from scratch, take a look a
This tutorial will use:
- Anthropic for the LLM - sign up and get an API key [here](https://console.anthropic.com/).
- Tavily for the search engine - sign up and get an API key [here](https://app.tavily.com/).
- LangSmith for hosting - sign up and get an API key [here](https://smith.langchain.com/).
- Anthropic for the LLM - sign up and get an API key [here](https://console.anthropic.com/)
- Tavily for the search engine - sign up and get an API key [here](https://app.tavily.com/)
- LangSmith for hosting - sign up and get an API key [here](https://smith.langchain.com/)
## Create and configure your app
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,19 @@
<!doctype html>
<html>
<head>
<title>Open Assistants API Specification</title>
<meta charset="utf-8" />
<meta
name="viewport"
content="width=device-width, initial-scale=1" />
</head>
<body>
<script id="api-reference" data-url="./open_agent_api.json"></script>
<script>
var configuration = {}
document.getElementById('api-reference').dataset.configuration =
JSON.stringify(configuration)
</script>
<script src="https://cdn.jsdelivr.net/npm/@scalar/api-reference"></script>
</body>
</html>
+6 -15
View File
@@ -1557,11 +1557,8 @@
"200": {
"description": "Success",
"content": {
"text/event-stream": {
"schema": {
"type": "string",
"description": "The server will send a stream of events in SSE format.\n\n**Example event**:\n\nid: 1\n\nevent: message\n\ndata: {}"
}
"application/json": {
"schema": {}
}
}
},
@@ -1908,11 +1905,8 @@
"200": {
"description": "Success",
"content": {
"text/event-stream": {
"schema": {
"type": "string",
"description": "The server will send a stream of events in SSE format.\n\n**Example event**:\n\nid: 1\n\nevent: message\n\ndata: {}"
}
"application/json": {
"schema": {}
}
}
},
@@ -2149,11 +2143,8 @@
"200": {
"description": "Success",
"content": {
"text/event-stream": {
"schema": {
"type": "string",
"description": "The server will send a stream of events in SSE format.\n\n**Example event**:\n\nid: 1\n\nevent: message\n\ndata: {}"
}
"application/json": {
"schema": {}
}
}
},
+1 -1
View File
@@ -30,7 +30,7 @@ The conceptual guide does not cover step-by-step instructions or specific implem
- [Streaming](streaming.md): Streaming is crucial for enhancing the responsiveness of applications built on LLMs. By displaying output progressively, even before a complete response is ready, streaming significantly improves user experience (UX), particularly when dealing with the latency of LLMs.
- [FAQ](faq.md): Frequently asked questions about LangGraph.
## LangGraph Platform
## LangGraph Platform
LangGraph Platform is a commercial solution for deploying agentic applications in production, built on the open-source LangGraph framework.
-8
View File
@@ -14,13 +14,6 @@ A **deployment** is an instance of a LangGraph API. A single deployment can have
See the [how-to guide](../cloud/deployment/cloud.md#create-new-deployment) for creating a new deployment.
## Resource Allocation
| **Deployment Type** | **CPU** | **Memory** | **Scaling** |
|---------------------|---------|------------|---------------------|
| Development | 1 CPU | 1 GB | Up to 1 container |
| Production | 1 CPU | 2 GB | Up to 10 containers |
## Revision
A revision is an iteration of a [deployment](#deployment). When a new deployment is created, an initial revision is automatically created. To deploy new code changes or update environment variable configurations for a deployment, a new revision must be created. When a revision is created, a new container image is built automatically.
@@ -40,7 +33,6 @@ A high-level diagram of a Cloud SaaS deployment.
![diagram](img/langgraph_cloud_architecture.png)
## Related
- [Deployment Options](./deployment_options.md)
+1 -5
View File
@@ -7,7 +7,7 @@
## Versions
There are two versions of the self-hosted deployment: [Self-Hosted Enterprise](./deployment_options.md#self-hosted-enterprise) and [Self-Hosted Lite](./deployment_options.md#self-hosted-lite).
There are two versions of the self hosted deployment: [Self-Hosted Enterprise](./deployment_options.md#self-hosted-enterprise) and [Self-Hosted Lite](./deployment_options.md#self-hosted-lite).
### Self-Hosted Lite
@@ -34,10 +34,6 @@ To use the Self-Hosted Enterprise version, you must acquire a license key that y
For step-by-step instructions, see [How to set up a self-hosted deployment of LangGraph](../how-tos/deploy-self-hosted.md).
## Helm Chart
If you would like to deploy LangGraph Cloud on Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
## Related
- [How to set up a self-hosted deployment of LangGraph](../how-tos/deploy-self-hosted.md).
-4
View File
@@ -17,10 +17,6 @@ You will need to do the following:
2. Build a docker image with the [LangGraph Server](../concepts/langgraph_server.md) using the [LangGraph CLI](../concepts/langgraph_cli.md).
3. Deploy a web server that will run the docker image and pass in the necessary environment variables.
## Helm Chart
If you would like to deploy LangGraph Cloud on Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
## Environment Variables
You will eventually need to pass in the following environment variables to the LangGraph Deploy server:
+12 -10
View File
@@ -6,23 +6,25 @@ title: Tutorials
# Tutorials
New to LangGraph or LLM app development? Read this material to get up and running building your first applications.
Welcome to the LangGraph Tutorials! These notebooks introduce LangGraph through building various language agents and applications.
## Get Started 🚀 {#quick-start}
## Quick Start
- [LangGraph Quickstart](introduction.ipynb): Build a chatbot that can use tools and keep track of conversation history. Add human-in-the-loop capabilities and explore how time-travel works.
- [LangGraph Server Quickstart](langgraph-platform/local-server.md): Launch a LangGraph server locally and interact with it using the REST API and LangGraph Studio Web UI.
- [LangGraph Cloud QuickStart](../cloud/quick_start.md): Deploy a LangGraph app using LangGraph Cloud.
Learn the basics of LangGraph through a comprehensive quick start in which you will build an agent from scratch.
## Use cases 🛠️
- [Quick Start](introduction.ipynb): In this tutorial, you will build a support chatbot using LangGraph.
- [LangGraph Cloud Quick Start](../cloud/quick_start.md): In this tutorial, you will build and deploy an agent to LangGraph Cloud.
Explore practical implementations tailored for specific scenarios:
## Use cases
Learn from example implementations of graphs designed for specific scenarios and that implement common design patterns.
### Chatbots
- [Customer Support](customer-support/customer-support.ipynb): Build a multi-functional support bot for flights, hotels, and car rentals.
- [Prompt Generation from User Requirements](chatbots/information-gather-prompting.ipynb): Build an information gathering chatbot.
- [Code Assistant](code_assistant/langgraph_code_assistant.ipynb): Build a code analysis and generation assistant.
- [Customer Support](customer-support/customer-support.ipynb): Build a customer support chatbot to manage flights, hotel reservations, car rentals, and other tasks
- [Prompt Generation from User Requirements](chatbots/information-gather-prompting.ipynb): Build an information gathering chatbot
- [Code Assistant](code_assistant/langgraph_code_assistant.ipynb): Build a code analysis and generation assistant
### RAG
+40 -47
View File
@@ -5,21 +5,21 @@
"id": "4a1aae78-88a6-4133-b905-7e46c8e3772f",
"metadata": {},
"source": [
"# 🚀 LangGraph Quick Start\n",
"# LangGraph Quick Start\n",
"\n",
"In this tutorial, we will build a support chatbot in LangGraph that can:\n",
"In this comprehensive quick start, we will build a support chatbot in LangGraph that can:\n",
"\n",
"✅ **Answer common questions** by searching the web \n",
"✅ **Maintain conversation state** across calls \n",
"✅ **Route complex queries** to a human for review \n",
"✅ **Use custom state** to control its behavior \n",
"✅ **Rewind and explore** alternative conversation paths \n",
"- Answer common questions by searching the web\n",
"- Maintain conversation state across calls\n",
"- Route complex queries to a human for review\n",
"- Use custom state to control its behavior\n",
"- Rewind and explore alternative conversation paths\n",
"\n",
"We'll start with a **basic chatbot** and progressively add more sophisticated capabilities, introducing key LangGraph concepts along the way. Lets dive in! 🌟\n",
"We'll start with a basic chatbot and progressively add more sophisticated capabilities, introducing key LangGraph concepts along the way.\n",
"\n",
"## Setup\n",
"\n",
"First, install the required packages and configure your environment:"
"First, install the required packages:"
]
},
{
@@ -33,6 +33,14 @@
"%pip install -U langgraph langsmith langchain_anthropic"
]
},
{
"cell_type": "markdown",
"id": "a6d1e870-1bc0-4d44-86c0-96681ccf6113",
"metadata": {},
"source": [
"Next, set your API keys:"
]
},
{
"cell_type": "code",
"execution_count": 2,
@@ -112,24 +120,27 @@
]
},
{
"attachments": {},
"cell_type": "markdown",
"id": "c08c41da-0855-49d3-9a3d-b7eb94413367",
"id": "31c755cd-8994-4867-bdff-96a55d7beae7",
"metadata": {},
"source": [
"Our graph can now handle two key tasks:\n",
"\n",
"1. Each `node` can receive the current `State` as input and output an update to the state.\n",
"2. Updates to `messages` will be appended to the existing list rather than overwriting it, thanks to the prebuilt [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/?h=add+messages#add_messages) function used with the `Annotated` syntax.\n",
"\n",
"------\n",
"\n",
"!!! tip \"Concept\"\n",
"\n",
" When defining a graph, the first step is to define its `State`. The `State` includes the graph's schema and [reducer functions](https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers) that handle state updates. In our example, `State` is a `TypedDict` with one key: `messages`. The [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages) reducer function is used to append new messages to the list instead of overwriting it. Keys without a reducer annotation will overwrite previous values. Learn more about state, reducers, and related concepts in [this guide](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages).\n",
"\n",
"---------\n",
"<div class=\"admonition tip\">\n",
" <p class=\"admonition-title\">Note</p>\n",
" <p>\n",
" The first thing you do when you define a graph is define the <code>State</code> of the graph. The <code>State</code> consists of the schema of the graph as well as <a href=\"https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers\">reducer functions</a> which specify how to apply updates to the state. In our example <code>State</code> is a <code>TypedDict</code> with a single key: <code>messages</code>. The <code>messages</code> key is annotated with the <a href=\"https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages\"><code>add_messages</code></a> reducer function, which tells LangGraph to append new messages to the existing list, rather than overwriting it. State keys without an annotation will be overwritten by each update, storing the most recent value. Check out <a href=\"https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages\">this conceptual guide</a> to learn more about state, reducers and other low-level concepts.\n",
" </p>\n",
"</div>"
]
},
{
"cell_type": "markdown",
"id": "4137feed-746e-4c72-a34a-f7a699ad5dcf",
"metadata": {},
"source": [
"So now our graph knows two things:\n",
"\n",
"1. Every `node` we define will receive the current `State` as input and return a value that updates that state.\n",
"2. `messages` will be _appended_ to the current list, rather than directly overwritten. This is communicated via the prebuilt [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/?h=add+messages#add_messages) function in the `Annotated` syntax.\n",
"\n",
"Next, add a \"`chatbot`\" node. Nodes represent units of work. They are typically regular python functions."
]
@@ -354,7 +365,7 @@
"id": "f22c5d4a-3134-413c-81fe-dd9752fbeb66",
"metadata": {},
"source": [
"## Part 2: 🛠️ Enhancing the Chatbot with Tools\n",
"## Part 2: Enhancing the Chatbot with Tools\n",
"\n",
"To handle queries our chatbot can't answer \"from memory\", we'll integrate a web search tool. Our bot can use this tool to find relevant information and provide better responses.\n",
"\n",
@@ -2035,7 +2046,7 @@
"\n",
"So far, we've relied on a simple state (it's just a list of messages!). You can go far with this simple state, but if you want to define complex behavior without relying on the message list, you can add additional fields to the state. In this section, we will extend our chat bot with a new node to illustrate this.\n",
"\n",
"In the examples above, we involved a human deterministically: the graph __always__ interrupted whenever a tool was invoked. Suppose we wanted our chat bot to have the choice of relying on a human.\n",
"In the examples above, we involved a human deterministically: the graph __always__ interrupted whenever an tool was invoked. Suppose we wanted our chat bot to have the choice of relying on a human.\n",
"\n",
"One way to do this is to create a passthrough \"human\" node, before which the graph will always stop. We will only execute this node if the LLM invokes a \"human\" tool. For our convenience, we will include an \"ask_human\" flag in our graph state that we will flip if the LLM calls this tool.\n",
"\n",
@@ -3125,29 +3136,11 @@
"id": "e584d57f-5aad-4507-815f-0b2e4b64b791",
"metadata": {},
"source": [
"## Next Steps\n",
"## Conclusion\n",
"\n",
"Take your journey further by exploring deployment and advanced features:\n",
"Congrats! You've completed the intro tutorial and built a chat bot in LangGraph that supports tool calling, persistent memory, human-in-the-loop interactivity, and even time-travel!\n",
"\n",
"### Server Quickstart\n",
"\n",
"- **[LangGraph Server Quickstart](../langgraph-platform/local-server)**: Launch a LangGraph server locally and interact with it using the REST API and LangGraph Studio Web UI.\n",
"\n",
"### LangGraph Cloud\n",
"\n",
"- **[LangGraph Cloud QuickStart](../../cloud/quick_start)**: Deploy your LangGraph app using LangGraph Cloud.\n",
"\n",
"### LangGraph Framework\n",
"\n",
"- **[LangGraph Concepts](../../concepts)**: Learn the foundational concepts of LangGraph. \n",
"- **[LangGraph How-to Guides](../../how-tos)**: Guides for common tasks with LangGraph.\n",
"\n",
"### LangGraph Platform\n",
"\n",
"Expand your knowledge with these resources:\n",
"\n",
"- **[LangGraph Platform Concepts](../../concepts#langgraph-platform)**: Understand the foundational concepts of the LangGraph Platform. \n",
"- **[LangGraph Platform How-to Guides](../../how-tos#langgraph-platform)**: Guides for common tasks with LangGraph Platform. "
"The [LangGraph documentation](https://langchain-ai.github.io/langgraph/) is a great resource for diving deeper into the library's capabilities."
]
}
],
@@ -3167,7 +3160,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.4"
"version": "3.11.9"
}
},
"nbformat": 4,
@@ -1,253 +0,0 @@
# Quick Start: Launch Local LangGraph Server
This is a quick start guide to help you get a LangGraph app up and running locally.
!!! info "Requirements"
- Python >= 3.11
- [LangGraph CLI](https://langchain-ai.github.io/langgraph/cloud/reference/cli/): Requires langchain-cli[inmem] >= 0.1.58
## Install the LangGraph CLI
```bash
pip install "langgraph-cli[inmem]==0.1.58" python-dotenv
```
## 🌱 Create a LangGraph App
Create a new app from the `react-agent` template. This template is a simple agent that can be flexibly extended to many tools.
=== "Python Server"
```shell
langgraph new path/to/your/app --template react-agent-python
```
=== "Node Server"
```shell
langgraph new path/to/your/app --template react-agent-js
```
!!! tip "Additional Templates"
If you use `langgraph new` without specifying a template, you will be presented with an interactive menu that will allow you to choose from a list of available templates.
## Install Dependencies
In the root of your new LangGraph app, install the dependencies:
```shell
pip install .
```
## Create a `.env` file
You will find a `.env.example` in the root of your new LangGraph app. Create
a `.env` file in the root of your new LangGraph app and copy the contents of the `.env.example` file into it, filling in the necessary API keys:
```bash
LANGSMITH_API_KEY=lsv2...
TAVILY_API_KEY=tvly-...
ANTHROPIC_API_KEY=sk-
OPENAI_API_KEY=sk-...
```
<details><summary>Get API Keys</summary>
<ul>
<li> <b>LANGSMITH_API_KEY</b>: Go to the <a href="https://smith.langchain.com/settings">LangSmith Settings page</a>. Then clck <b>Create API Key</b>.
</li>
<li>
<b>ANTHROPIC_API_KEY</b>: Get an API key from <a href="https://console.anthropic.com/">Anthropic</a>.
</li>
<li>
<b>OPENAI_API_KEY</b>: Get an API key from <a href="https://openai.com/">OpenAI</a>.
</li>
<li>
<b>TAVILY_API_KEY</b>: Get an API key on the <a href="https://app.tavily.com/">Tavily website</a>.
</li>
</ul>
</details>
## 🚀 Launch LangGraph Server
```shell
langgraph dev
```
This will start up the LangGraph API server locally. If this runs successfully, you should see something like:
> Ready!
>
> - API: [http://localhost:8123](http://localhost:8123/)
>
> - Docs: http://localhost:8123/docs
>
> - LangGraph Studio Web UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:8123
!!! note "In-Memory Mode"
The `langgraph dev` command starts LangGraph Server in an in-memory mode. This mode is suitable for development and testing purposes. For production use, you should deploy LangGraph Server with access to a persistent storage backend.
If you want to test your application with a persistent storage backend, you can use the `langgraph up` command instead of `langgraph dev`. You will
need to have `docker` installed on your machine to use this command.
## LangGraph Studio Web UI
Test your graph in the LangGraph Studio Web UI by visiting the URL provided in the output of the `langgraph up` command.
> - LangGraph Studio Web UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:8123
!!! warning "Safari Compatibility"
Currently, LangGraph Studio Web does not support Safari when running a server locally.
## Test the API
=== "Python SDK (Async)"
**Install the LangGraph Python SDK**
```shell
pip install langgraph-sdk
```
**Send a message to the assistant (threadless run)**
```python
from langgraph_sdk import get_client
client = get_client(url="http://localhost:8123")
async for chunk in client.runs.stream(
None, # Threadless run
"agent", # Name of assistant. Defined in langgraph.json.
input={
"messages": [{
"role": "human",
"content": "What is LangGraph?",
}],
},
stream_mode="updates",
):
print(f"Receiving new event of type: {chunk.event}...")
print(chunk.data)
print("\n\n")
```
=== "Python SDK (Sync)"
**Install the LangGraph Python SDK**
```shell
pip install langgraph-sdk
```
**Send a message to the assistant (threadless run)**
```python
from langgraph_sdk import get_sync_client
client = get_sync_client(url="http://localhost:8123")
for chunk in client.runs.stream(
None, # Threadless run
"agent", # Name of assistant. Defined in langgraph.json.
input={
"messages": [{
"role": "human",
"content": "What is LangGraph?",
}],
},
stream_mode="updates",
):
print(f"Receiving new event of type: {chunk.event}...")
print(chunk.data)
print("\n\n")
```
=== "Javascript SDK"
**Install the LangGraph JS SDK**
```shell
npm install @langchain/langgraph-sdk
```
**Send a message to the assistant (threadless run)**
```js
const { Client } = await import("@langchain/langgraph-sdk");
// only set the apiUrl if you changed the default port when calling langgraph up
const client = new Client({ apiUrl: "http://localhost:8123"});
const streamResponse = client.runs.stream(
null, // Threadless run
"agent", // Assistant ID
{
input: {
"messages": [
{ "role": "user", "content": "What is LangGraph?"}
]
},
streamMode: "messages",
}
);
for await (const chunk of streamResponse) {
console.log(`Receiving new event of type: ${chunk.event}...`);
console.log(JSON.stringify(chunk.data));
console.log("\n\n");
}
```
=== "Rest API"
```bash
curl -s --request POST \
--url "http://localhost:8123/runs/stream" \
--header 'Content-Type: application/json' \
--data "{
\"assistant_id\": \"agent\",
\"input\": {
\"messages\": [
{
\"role\": \"human\",
\"content\": \"What is LangGraph?\"
}
]
},
\"stream_mode\": \"updates\"
}"
```
!!! tip "Auth"
If you're connecting to a remote server, you will need to provide a LangSmith
API Key for authorization. Please see the API Reference for the clients
for more information.
## Next Steps
Now that you have a LangGraph app running locally, take your journey further by exploring deployment and advanced features:
### 🌐 Deploy to LangGraph Cloud
- **[LangGraph Cloud QuickStart](../../cloud/quick_start.md)**: Deploy your LangGraph app using LangGraph Cloud.
### 📚 Learn More about LangGraph Platform
Expand your knowledge with these resources:
- **[LangGraph Platform Concepts](../../concepts/index.md#langgraph-platform)**: Understand the foundational concepts of the LangGraph Platform.
- **[LangGraph Platform How-to Guides](../../how-tos/index.md#langgraph-platform)**: Discover step-by-step guides to build and deploy applications.
### 🛠️ Developer References
Access detailed documentation for development and API usage:
- **[LangGraph Server API Reference](../../cloud/reference/api/api_ref.html)**: Explore the LangGraph Server API documentation.
- **[Python SDK Reference](../../cloud/reference/sdk/python_sdk_ref.md)**: Explore the Python SDK API Reference.
- **[JS/TS SDK Reference](../../cloud/reference/sdk/js_ts_sdk_ref.md)**: Explore the Python SDK API Reference.
@@ -934,7 +934,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.4"
"version": "3.11.9"
}
},
"nbformat": 4,
+1 -2
View File
@@ -94,7 +94,6 @@ nav:
- Quick Start:
- Quick Start: tutorials#quick-start
- tutorials/introduction.ipynb
- tutorials/langgraph-platform/local-server.md
- cloud/quick_start.md
- Chatbots:
- Chatbots: tutorials#chatbots
@@ -439,4 +438,4 @@ validation:
# and those anchors are not available in the actual doc
anchors: info
# this is needed to handle headers with anchors for nav
not_found: info
not_found: info
@@ -23,7 +23,6 @@ from langgraph.store.base import (
Op,
PutOp,
Result,
SearchItem,
SearchOp,
)
@@ -284,7 +283,7 @@ class DuckDBStore(BaseStore, BaseDuckDBStore[duckdb.DuckDBPyConnection]):
for cur, idx in cursors:
rows = cur.fetchall()
items = [_row_to_search_item(_convert_ns(row[0]), row) for row in rows]
items = [_row_to_item(_convert_ns(row[0]), row) for row in rows]
results[idx] = items
def _batch_list_namespaces_ops(
@@ -377,22 +376,6 @@ def _row_to_item(
)
def _row_to_search_item(
namespace: tuple[str, ...],
row: tuple,
) -> SearchItem:
"""Convert a row from the database into an SearchItem."""
# TODO: Add support for search
_, key, val, created_at, updated_at = row
return SearchItem(
value=val if isinstance(val, dict) else json.loads(val),
key=key,
namespace=namespace,
created_at=created_at,
updated_at=updated_at,
)
def _group_ops(ops: Iterable[Op]) -> tuple[dict[type, list[tuple[int, Op]]], int]:
grouped_ops: dict[type, list[tuple[int, Op]]] = defaultdict(list)
tot = 0
+1 -5
View File
@@ -5,11 +5,7 @@
######################
start-postgres:
POSTGRES_VERSION=${POSTGRES_VERSION:-16} docker compose -f tests/compose-postgres.yml up -V --force-recreate --wait || ( \
echo "Failed to start PostgreSQL, printing logs..."; \
docker compose -f tests/compose-postgres.yml logs; \
exit 1 \
)
POSTGRES_VERSION=${POSTGRES_VERSION:-16} docker compose -f tests/compose-postgres.yml up -V --force-recreate --wait
stop-postgres:
docker compose -f tests/compose-postgres.yml down
@@ -1,7 +1,6 @@
import threading
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
from typing import Any, Optional
from typing import Any, Iterator, Optional, Sequence, Union
from langchain_core.runnables import RunnableConfig
from psycopg import Capabilities, Connection, Cursor, Pipeline
@@ -18,11 +17,21 @@ from langgraph.checkpoint.base import (
CheckpointTuple,
get_checkpoint_id,
)
from langgraph.checkpoint.postgres import _internal
from langgraph.checkpoint.postgres.base import BasePostgresSaver
from langgraph.checkpoint.serde.base import SerializerProtocol
Conn = _internal.Conn # For backward compatibility
Conn = Union[Connection[DictRow], ConnectionPool[Connection[DictRow]]]
@contextmanager
def _get_connection(conn: Conn) -> Iterator[Connection[DictRow]]:
if isinstance(conn, Connection):
yield conn
elif isinstance(conn, ConnectionPool):
with conn.connection() as conn:
yield conn
else:
raise TypeError(f"Invalid connection type: {type(conn)}")
class PostgresSaver(BasePostgresSaver):
@@ -30,7 +39,7 @@ class PostgresSaver(BasePostgresSaver):
def __init__(
self,
conn: _internal.Conn,
conn: Conn,
pipe: Optional[Pipeline] = None,
serde: Optional[SerializerProtocol] = None,
) -> None:
@@ -64,9 +73,9 @@ class PostgresSaver(BasePostgresSaver):
) as conn:
if pipeline:
with conn.pipeline() as pipe:
yield cls(conn, pipe)
yield PostgresSaver(conn, pipe)
else:
yield cls(conn)
yield PostgresSaver(conn)
def setup(self) -> None:
"""Set up the checkpoint database asynchronously.
@@ -364,7 +373,7 @@ class PostgresSaver(BasePostgresSaver):
Will be applied regardless of whether the PostgresSaver instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
"""
with _internal.get_connection(self.conn) as conn:
with _get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
@@ -379,23 +388,19 @@ class PostgresSaver(BasePostgresSaver):
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
with self.lock, conn.pipeline(), conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
else:
# Use connection's transaction context manager when pipeline mode not supported
with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
with self.lock, conn.transaction(), conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
else:
with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"]
__all__ = ["PostgresSaver", "Conn"]
@@ -1,24 +0,0 @@
"""Shared async utility functions for the Postgres checkpoint & storage classes."""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Union
from psycopg import AsyncConnection
from psycopg.rows import DictRow
from psycopg_pool import AsyncConnectionPool
Conn = Union[AsyncConnection[DictRow], AsyncConnectionPool[AsyncConnection[DictRow]]]
@asynccontextmanager
async def get_connection(
conn: Conn,
) -> AsyncIterator[AsyncConnection[DictRow]]:
if isinstance(conn, AsyncConnection):
yield conn
elif isinstance(conn, AsyncConnectionPool):
async with conn.connection() as conn:
yield conn
else:
raise TypeError(f"Invalid connection type: {type(conn)}")
@@ -1,22 +0,0 @@
"""Shared utility functions for the Postgres checkpoint & storage classes."""
from collections.abc import Iterator
from contextlib import contextmanager
from typing import Union
from psycopg import Connection
from psycopg.rows import DictRow
from psycopg_pool import ConnectionPool
Conn = Union[Connection[DictRow], ConnectionPool[Connection[DictRow]]]
@contextmanager
def get_connection(conn: Conn) -> Iterator[Connection[DictRow]]:
if isinstance(conn, Connection):
yield conn
elif isinstance(conn, ConnectionPool):
with conn.connection() as conn:
yield conn
else:
raise TypeError(f"Invalid connection type: {type(conn)}")
@@ -1,7 +1,6 @@
import asyncio
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import asynccontextmanager
from typing import Any, Optional
from typing import Any, AsyncIterator, Iterator, Optional, Sequence, Union
from langchain_core.runnables import RunnableConfig
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
@@ -18,11 +17,23 @@ from langgraph.checkpoint.base import (
CheckpointTuple,
get_checkpoint_id,
)
from langgraph.checkpoint.postgres import _ainternal
from langgraph.checkpoint.postgres.base import BasePostgresSaver
from langgraph.checkpoint.serde.base import SerializerProtocol
Conn = _ainternal.Conn # For backward compatibility
Conn = Union[AsyncConnection[DictRow], AsyncConnectionPool[AsyncConnection[DictRow]]]
@asynccontextmanager
async def _get_connection(
conn: Conn,
) -> AsyncIterator[AsyncConnection[DictRow]]:
if isinstance(conn, AsyncConnection):
yield conn
elif isinstance(conn, AsyncConnectionPool):
async with conn.connection() as conn:
yield conn
else:
raise TypeError(f"Invalid connection type: {type(conn)}")
class AsyncPostgresSaver(BasePostgresSaver):
@@ -30,7 +41,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
def __init__(
self,
conn: _ainternal.Conn,
conn: Conn,
pipe: Optional[AsyncPipeline] = None,
serde: Optional[SerializerProtocol] = None,
) -> None:
@@ -69,9 +80,9 @@ class AsyncPostgresSaver(BasePostgresSaver):
) as conn:
if pipeline:
async with conn.pipeline() as pipe:
yield cls(conn=conn, pipe=pipe, serde=serde)
yield AsyncPostgresSaver(conn=conn, pipe=pipe, serde=serde)
else:
yield cls(conn=conn, serde=serde)
yield AsyncPostgresSaver(conn=conn, serde=serde)
async def setup(self) -> None:
"""Set up the checkpoint database asynchronously.
@@ -146,17 +157,15 @@ class AsyncPostgresSaver(BasePostgresSaver):
value["pending_sends"],
),
self._load_metadata(value["metadata"]),
(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
if value["parent_checkpoint_id"]
else None
),
}
if value["parent_checkpoint_id"]
else None,
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
@@ -207,17 +216,15 @@ class AsyncPostgresSaver(BasePostgresSaver):
value["pending_sends"],
),
self._load_metadata(value["metadata"]),
(
{
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": value["parent_checkpoint_id"],
}
{
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": value["parent_checkpoint_id"],
}
if value["parent_checkpoint_id"]
else None
),
}
if value["parent_checkpoint_id"]
else None,
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
@@ -324,7 +331,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
Will be applied regardless of whether the AsyncPostgresSaver instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
"""
async with _ainternal.get_connection(self.conn) as conn:
async with _get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
@@ -339,25 +346,20 @@ class AsyncPostgresSaver(BasePostgresSaver):
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
async with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
async with self.lock, conn.pipeline(), conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
else:
# Use connection's transaction context manager when pipeline mode not supported
async with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
async with self.lock, conn.transaction(), conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
else:
async with (
self.lock,
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
async with self.lock, conn.cursor(
binary=True, row_factory=dict_row
) as cur:
yield cur
def list(
@@ -386,7 +388,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
while True:
try:
yield asyncio.run_coroutine_threadsafe(
anext(aiter_), # noqa: F821
anext(aiter_),
self.loop,
).result()
except StopAsyncIteration:
@@ -465,6 +467,3 @@ class AsyncPostgresSaver(BasePostgresSaver):
return asyncio.run_coroutine_threadsafe(
self.aput_writes(config, writes, task_id), self.loop
).result()
__all__ = ["AsyncPostgresSaver", "Conn"]
@@ -1,6 +1,5 @@
import random
from collections.abc import Sequence
from typing import Any, Optional, cast
from typing import Any, List, Optional, Sequence, Tuple, cast
from langchain_core.runnables import RunnableConfig
from psycopg.types.json import Jsonb
@@ -250,7 +249,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
config: Optional[RunnableConfig],
filter: MetadataInput,
before: Optional[RunnableConfig] = None,
) -> tuple[str, list[Any]]:
) -> Tuple[str, List[Any]]:
"""Return WHERE clause predicates for alist() given config, filter, before.
This method returns a tuple of a string and a tuple of values. The string
@@ -1,241 +1,110 @@
import asyncio
import logging
from collections.abc import AsyncIterator, Iterable, Sequence
from contextlib import asynccontextmanager
from typing import Any, Callable, Optional, Union, cast
from typing import (
Any,
AsyncIterator,
Callable,
Iterable,
Optional,
Sequence,
Union,
cast,
)
import orjson
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
from psycopg import AsyncConnection, AsyncCursor
from psycopg.errors import UndefinedTable
from psycopg.rows import DictRow, dict_row
from psycopg_pool import AsyncConnectionPool
from psycopg.rows import dict_row
from langgraph.checkpoint.postgres import _ainternal
from langgraph.store.base import (
GetOp,
ListNamespacesOp,
Op,
PutOp,
Result,
SearchOp,
)
from langgraph.store.base import GetOp, ListNamespacesOp, Op, PutOp, Result, SearchOp
from langgraph.store.base.batch import AsyncBatchedBaseStore
from langgraph.store.postgres.base import (
_PLACEHOLDER,
BasePostgresStore,
PoolConfig,
PostgresIndexConfig,
Row,
_decode_ns_bytes,
_ensure_index_config,
_group_ops,
_row_to_item,
_row_to_search_item,
)
logger = logging.getLogger(__name__)
class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Conn]):
__slots__ = (
"_deserializer",
"pipe",
"lock",
"supports_pipeline",
"index_config",
"embeddings",
)
class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[AsyncConnection]):
__slots__ = ("_deserializer",)
def __init__(
self,
conn: _ainternal.Conn,
conn: AsyncConnection[Any],
*,
pipe: Optional[AsyncPipeline] = None,
deserializer: Optional[
Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]
] = None,
index: Optional[PostgresIndexConfig] = None,
) -> None:
if isinstance(conn, AsyncConnectionPool) and pipe is not None:
raise ValueError(
"Pipeline should be used only with a single AsyncConnection, not AsyncConnectionPool."
)
super().__init__()
self._deserializer = deserializer
self.conn = conn
self.pipe = pipe
self.lock = asyncio.Lock()
self.loop = asyncio.get_running_loop()
self.supports_pipeline = Capabilities().has_pipeline()
self.index_config = index
if self.index_config:
self.embeddings, self.index_config = _ensure_index_config(self.index_config)
else:
self.embeddings = None
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
grouped_ops, num_ops = _group_ops(ops)
results: list[Result] = [None] * num_ops
async with _ainternal.get_connection(self.conn) as conn:
if self.pipe:
async with self.pipe:
await self._execute_batch(grouped_ops, results, conn)
else:
await self._execute_batch(grouped_ops, results, conn)
async with self.conn.pipeline():
tasks = []
if GetOp in grouped_ops:
tasks.append(
self._batch_get_ops(
cast(Sequence[tuple[int, GetOp]], grouped_ops[GetOp]), results
)
)
if PutOp in grouped_ops:
tasks.append(
self._batch_put_ops(
cast(Sequence[tuple[int, PutOp]], grouped_ops[PutOp])
)
)
if SearchOp in grouped_ops:
tasks.append(
self._batch_search_ops(
cast(Sequence[tuple[int, SearchOp]], grouped_ops[SearchOp]),
results,
)
)
if ListNamespacesOp in grouped_ops:
tasks.append(
self._batch_list_namespaces_ops(
cast(
Sequence[tuple[int, ListNamespacesOp]],
grouped_ops[ListNamespacesOp],
),
results,
)
)
await asyncio.gather(*tasks)
return results
def batch(self, ops: Iterable[Op]) -> list[Result]:
return asyncio.run_coroutine_threadsafe(self.abatch(ops), self.loop).result()
@classmethod
@asynccontextmanager
async def from_conn_string(
cls,
conn_string: str,
*,
pipeline: bool = False,
pool_config: Optional[PoolConfig] = None,
index: Optional[PostgresIndexConfig] = None,
) -> AsyncIterator["AsyncPostgresStore"]:
"""Create a new AsyncPostgresStore instance from a connection string.
Args:
conn_string (str): The Postgres connection info string.
pipeline (bool): Whether to use AsyncPipeline (only for single connections)
pool_config (Optional[PoolConfig]): Configuration for the connection pool.
If provided, will create a connection pool and use it instead of a single connection.
This overrides the `pipeline` argument.
index (Optional[PostgresIndexConfig]): The embedding config.
Returns:
AsyncPostgresStore: A new AsyncPostgresStore instance.
"""
if pool_config is not None:
pc = pool_config.copy()
async with cast(
AsyncConnectionPool[AsyncConnection[DictRow]],
AsyncConnectionPool(
conn_string,
min_size=pc.pop("min_size", 1),
max_size=pc.pop("max_size", None),
kwargs={
"autocommit": True,
"prepare_threshold": 0,
"row_factory": dict_row,
**(pc.pop("kwargs", None) or {}),
},
**cast(dict, pc),
),
) as pool:
yield cls(conn=pool, index=index)
else:
async with await AsyncConnection.connect(
conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
) as conn:
if pipeline:
async with conn.pipeline() as pipe:
yield cls(conn=conn, pipe=pipe, index=index)
else:
yield cls(conn=conn, index=index)
async def setup(self) -> None:
"""Set up the store database asynchronously.
This method creates the necessary tables in the Postgres database if they don't
already exist and runs database migrations. It MUST be called directly by the user
the first time the store is used.
"""
async def _get_version(cur: AsyncCursor[DictRow], table: str) -> int:
try:
await cur.execute(f"SELECT v FROM {table} ORDER BY v DESC LIMIT 1")
row = await cur.fetchone()
if row is None:
version = -1
else:
version = row["v"]
except UndefinedTable:
version = -1
await cur.execute(
f"""
CREATE TABLE IF NOT EXISTS {table} (
v INTEGER PRIMARY KEY
)
"""
)
return version
async with self._cursor() as cur:
version = await _get_version(cur, table="store_migrations")
for v, sql in enumerate(self.MIGRATIONS[version + 1 :], start=version + 1):
await cur.execute(sql)
await cur.execute("INSERT INTO store_migrations (v) VALUES (%s)", (v,))
if self.index_config:
version = await _get_version(cur, table="vector_migrations")
for v, migration in enumerate(
self.VECTOR_MIGRATIONS[version + 1 :], start=version + 1
):
sql = migration.sql
if migration.params:
params = {
k: v(self) if v is not None and callable(v) else v
for k, v in migration.params.items()
}
sql = sql % params
await cur.execute(sql)
await cur.execute(
"INSERT INTO vector_migrations (v) VALUES (%s)", (v,)
)
async def _execute_batch(
self,
grouped_ops: dict,
results: list[Result],
conn: AsyncConnection[DictRow],
) -> None:
async with self._cursor(pipeline=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,
)
async def _batch_get_ops(
self,
get_ops: Sequence[tuple[int, GetOp]],
results: list[Result],
cur: AsyncCursor[DictRow],
) -> None:
cursors = []
for query, params, namespace, items in self._get_batch_GET_ops_queries(get_ops):
cur = self.conn.cursor(binary=True)
await cur.execute(query, params)
cursors.append((cur, namespace, items))
for cur, namespace, items in cursors:
rows = cast(list[Row], await cur.fetchall())
key_to_row = {row["key"]: row for row in rows}
for idx, key in items:
@@ -250,59 +119,29 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
async def _batch_put_ops(
self,
put_ops: Sequence[tuple[int, PutOp]],
cur: AsyncCursor[DictRow],
) -> None:
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 EmbeddingConfig when initializing the {self.__class__.__name__}."
)
query, txt_params = embedding_request
vectors = await self.embeddings.aembed_documents(
[param[-1] for param in txt_params]
)
queries.append(
(
query,
[
p
for (ns, k, pathname, _), vector in zip(txt_params, vectors)
for p in (ns, k, pathname, vector)
],
)
)
queries = self._get_batch_PUT_queries(put_ops)
for query, params in queries:
cur = self.conn.cursor(binary=True)
await cur.execute(query, params)
async def _batch_search_ops(
self,
search_ops: Sequence[tuple[int, SearchOp]],
results: list[Result],
cur: AsyncCursor[DictRow],
) -> None:
queries, embedding_requests = self._prepare_batch_search_queries(search_ops)
queries = self._get_batch_search_queries(search_ops)
cursors: list[tuple[AsyncCursor[Any], int]] = []
if embedding_requests and self.embeddings:
vectors = await self.embeddings.aembed_documents(
[query for _, query in embedding_requests]
)
for (idx, _), vector in zip(embedding_requests, vectors):
_paramslist = queries[idx][1]
for i in range(len(_paramslist)):
if _paramslist[i] is _PLACEHOLDER:
_paramslist[i] = vector
for (idx, _), (query, params) in zip(search_ops, queries):
for (query, params), (idx, _) in zip(queries, search_ops):
cur = self.conn.cursor(binary=True)
await cur.execute(query, params)
cursors.append((cur, idx))
for cur, idx in cursors:
rows = cast(list[Row], await cur.fetchall())
items = [
_row_to_search_item(
_row_to_item(
_decode_ns_bytes(row["prefix"]), row, loader=self._deserializer
)
for row in rows
@@ -313,57 +152,67 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
self,
list_ops: Sequence[tuple[int, ListNamespacesOp]],
results: list[Result],
cur: AsyncCursor[DictRow],
) -> None:
queries = self._get_batch_list_namespaces_queries(list_ops)
cursors: list[tuple[AsyncCursor[Any], int]] = []
for (query, params), (idx, _) in zip(queries, list_ops):
cur = self.conn.cursor(binary=True)
await cur.execute(query, params)
cursors.append((cur, idx))
for cur, idx in cursors:
rows = cast(list[dict], await cur.fetchall())
namespaces = [_decode_ns_bytes(row["truncated_prefix"]) for row in rows]
results[idx] = namespaces
@classmethod
@asynccontextmanager
async def _cursor(
self, *, pipeline: bool = False
) -> AsyncIterator[AsyncCursor[DictRow]]:
"""Create a database cursor as a context manager.
async def from_conn_string(
cls,
conn_string: str,
) -> AsyncIterator["AsyncPostgresStore"]:
"""Create a new AsyncPostgresStore instance from a connection string.
Args:
pipeline: whether to use pipeline for the DB operations inside the context manager.
Will be applied regardless of whether the PostgresStore instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
conn_string (str): The Postgres connection info string.
Returns:
AsyncPostgresStore: A new AsyncPostgresStore instance.
"""
async with _ainternal.get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
# used at a time
try:
async with conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
finally:
if pipeline:
await self.pipe.sync()
elif pipeline:
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
async with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
async with await AsyncConnection.connect(
conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
) as conn:
yield cls(conn=conn)
async def setup(self) -> None:
"""Set up the store database asynchronously.
This method creates the necessary tables in the Postgres database if they don't
already exist and runs database migrations. It MUST be called directly by the user
the first time the store is used.
"""
async with self.conn.cursor() as cur:
try:
await cur.execute(
"SELECT v FROM store_migrations ORDER BY v DESC LIMIT 1"
)
row = cast(dict, await cur.fetchone())
if row is None:
version = -1
else:
async with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
async with (
self.lock,
conn.cursor(binary=True) as cur,
):
yield cur
version = row["v"]
except UndefinedTable:
version = -1
# Create store_migrations table if it doesn't exist
await cur.execute(
"""
CREATE TABLE IF NOT EXISTS store_migrations (
v INTEGER PRIMARY KEY
)
"""
)
for v, migration in enumerate(
self.MIGRATIONS[version + 1 :], start=version + 1
):
await cur.execute(migration)
await cur.execute("INSERT INTO store_migrations (v) VALUES (%s)", (v,))
File diff suppressed because it is too large Load Diff
+426 -529
View File
File diff suppressed because it is too large Load Diff
+4 -4
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint-postgres"
version = "2.0.6"
version = "2.0.3"
description = "Library with a Postgres implementation of LangGraph checkpoint saver."
authors = []
license = "MIT"
@@ -10,10 +10,10 @@ packages = [{ include = "langgraph" }]
[tool.poetry.dependencies]
python = "^3.9.0,<4.0"
langgraph-checkpoint = "^2.0.7"
langgraph-checkpoint = "^2.0.2"
orjson = ">=3.10.1"
psycopg = "^3.2.0"
psycopg-pool = "^3.2.0"
psycopg = "^3.0.0"
psycopg-pool = "^3.0.0"
[tool.poetry.group.dev.dependencies]
ruff = "^0.6.2"
@@ -1,13 +1,12 @@
services:
postgres-test:
image: pgvector/pgvector:pg${POSTGRES_VERSION:-16}
image: postgres:${POSTGRES_VERSION:-16}
ports:
- "5441:5432"
environment:
POSTGRES_DB: postgres
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
command: ["postgres", "-c", "shared_preload_libraries=vector"]
healthcheck:
test: pg_isready -U postgres
start_period: 10s
+1 -15
View File
@@ -1,12 +1,10 @@
from collections.abc import AsyncIterator
from typing import AsyncIterator
import pytest
from psycopg import AsyncConnection
from psycopg.errors import UndefinedTable
from psycopg.rows import DictRow, dict_row
from tests.embed_test_utils import CharacterEmbeddings
DEFAULT_URI = "postgres://postgres:postgres@localhost:5441/postgres?sslmode=disable"
@@ -26,18 +24,6 @@ async def clear_test_db(conn: AsyncConnection[DictRow]) -> None:
await conn.execute("DELETE FROM checkpoint_blobs")
await conn.execute("DELETE FROM checkpoint_writes")
await conn.execute("DELETE FROM checkpoint_migrations")
except UndefinedTable:
pass
try:
await conn.execute("DELETE FROM store_migrations")
await conn.execute("DELETE FROM store")
except UndefinedTable:
pass
@pytest.fixture
def fake_embeddings() -> CharacterEmbeddings:
return CharacterEmbeddings(dims=500)
VECTOR_TYPES = ["vector", "halfvec"]
@@ -1,55 +0,0 @@
"""Embedding utilities for testing."""
import math
import random
from collections import Counter, defaultdict
from typing import Any
from langchain_core.embeddings import Embeddings
class CharacterEmbeddings(Embeddings):
"""Simple character-frequency based embeddings using random projections."""
def __init__(self, dims: int = 50, seed: int = 42):
"""Initialize with embedding dimensions and random seed."""
self._rng = random.Random(seed)
self.dims = dims
# Create projection vector for each character lazily
self._char_projections: defaultdict[str, list[float]] = defaultdict(
lambda: [
self._rng.gauss(0, 1 / math.sqrt(self.dims)) for _ in range(self.dims)
]
)
def _embed_one(self, text: str) -> list[float]:
"""Embed a single text."""
counts = Counter(text)
total = sum(counts.values())
if total == 0:
return [0.0] * self.dims
embedding = [0.0] * self.dims
for char, count in counts.items():
weight = count / total
char_proj = self._char_projections[char]
for i, proj in enumerate(char_proj):
embedding[i] += weight * proj
norm = math.sqrt(sum(x * x for x in embedding))
if norm > 0:
embedding = [x / norm for x in embedding]
return embedding
def embed_documents(self, texts: list[str]) -> list[list[float]]:
"""Embed a list of documents."""
return [self._embed_one(text) for text in texts]
def embed_query(self, text: str) -> list[float]:
"""Embed a query string."""
return self._embed_one(text)
def __eq__(self, other: Any) -> bool:
return isinstance(other, CharacterEmbeddings) and self.dims == other.dims
+1 -1
View File
@@ -1,6 +1,7 @@
from typing import Any
import pytest
from conftest import DEFAULT_URI # type: ignore
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
@@ -10,7 +11,6 @@ from langgraph.checkpoint.base import (
empty_checkpoint,
)
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from tests.conftest import DEFAULT_URI
class TestAsyncPostgresSaver:
+377 -355
View File
@@ -1,84 +1,114 @@
# type: ignore
import itertools
import sys
import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any, Optional
from datetime import datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock
import pytest
from langchain_core.embeddings import Embeddings
from psycopg import AsyncConnection
from conftest import DEFAULT_URI # type: ignore
from langgraph.store.base import GetOp, Item, ListNamespacesOp, PutOp, SearchOp
from langgraph.store.postgres import AsyncPostgresStore
from tests.conftest import (
DEFAULT_URI,
VECTOR_TYPES,
CharacterEmbeddings,
)
@pytest.fixture(scope="function", params=["default", "pipe", "pool"])
async def store(request) -> AsyncIterator[AsyncPostgresStore]:
if sys.version_info < (3, 10):
pytest.skip("Async Postgres tests require Python 3.10+")
class MockAsyncCursor:
def __init__(self, fetch_result: Any) -> None:
self.fetch_result = fetch_result
self.execute = AsyncMock()
self.fetchall = AsyncMock(return_value=self.fetch_result)
database = f"test_{uuid.uuid4().hex[:16]}"
uri_parts = DEFAULT_URI.split("/")
uri_base = "/".join(uri_parts[:-1])
query_params = ""
if "?" in uri_parts[-1]:
db_name, query_params = uri_parts[-1].split("?", 1)
query_params = "?" + query_params
conn_string = f"{uri_base}/{database}{query_params}"
admin_conn_string = DEFAULT_URI
class MockAsyncConnection:
def __init__(self) -> None:
self.cursor = MagicMock()
self.pipeline = MagicMock(
return_value=AsyncMock(__aenter__=AsyncMock(), __aexit__=AsyncMock())
)
async with await AsyncConnection.connect(
admin_conn_string, autocommit=True
) as conn:
await conn.execute(f"CREATE DATABASE {database}")
try:
async with AsyncPostgresStore.from_conn_string(conn_string) as store:
await store.setup()
if request.param == "pipe":
async with AsyncPostgresStore.from_conn_string(
conn_string, pipeline=True
) as store:
yield store
elif request.param == "pool":
async with AsyncPostgresStore.from_conn_string(
conn_string, pool_config={"min_size": 1, "max_size": 10}
) as store:
yield store
else: # default
async with AsyncPostgresStore.from_conn_string(conn_string) as store:
yield store
finally:
async with await AsyncConnection.connect(
admin_conn_string, autocommit=True
) as conn:
await conn.execute(f"DROP DATABASE {database}")
@pytest.fixture
def mock_connection() -> MockAsyncConnection:
return MockAsyncConnection()
@pytest.fixture
async def store(mock_connection: MockAsyncConnection) -> AsyncPostgresStore:
return AsyncPostgresStore(mock_connection)
async def test_abatch_order(store: AsyncPostgresStore) -> None:
# Setup test data
await store.aput(("test", "foo"), "key1", {"data": "value1"})
await store.aput(("test", "bar"), "key2", {"data": "value2"})
mock_connection = store.conn
mock_get_cursor = MockAsyncCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
{
"key": "key2",
"value": '{"data": "value2"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.bar",
},
]
)
mock_search_cursor = MockAsyncCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
]
)
mock_list_namespaces_cursor = MockAsyncCursor(
[
{"truncated_prefix": b"\x01test"},
]
)
failures = []
def cursor_side_effect(binary: bool = False) -> Any:
cursor = MagicMock()
async def execute_side_effect(query: str, *params: Any) -> None:
# My super sophisticated database.
if "SELECT prefix, key," in query:
cursor.fetchall = mock_search_cursor.fetchall
elif "SELECT DISTINCT ON (truncated_prefix)" in query:
cursor.fetchall = mock_list_namespaces_cursor.fetchall
elif "WHERE prefix = %s AND key" in query:
cursor.fetchall = mock_get_cursor.fetchall
elif "INSERT INTO " in query:
pass
else:
e = ValueError(f"Unmatched query: {query}")
failures.append(e)
raise e
cursor.execute = AsyncMock(side_effect=execute_side_effect)
return cursor
mock_connection.cursor.side_effect = cursor_side_effect # type: ignore
ops = [
GetOp(namespace=("test", "foo"), key="key1"),
PutOp(namespace=("test", "bar"), key="key2", value={"data": "value2"}),
GetOp(namespace=("test",), key="key1"),
PutOp(namespace=("test",), 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(ops)
assert not failures
assert len(results) == 5
assert isinstance(results[0], Item)
assert isinstance(results[0].value, dict)
@@ -88,29 +118,27 @@ async def test_abatch_order(store: AsyncPostgresStore) -> None:
assert isinstance(results[2], list)
assert len(results[2]) == 1
assert isinstance(results[3], list)
assert ("test", "foo") in results[3] and ("test", "bar") in results[3]
assert results[3] == [("test",)]
assert results[4] is None
ops_reordered = [
SearchOp(namespace_prefix=("test",), filter=None, limit=5, offset=0),
GetOp(namespace=("test", "bar"), key="key2"),
GetOp(namespace=("test",), 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"),
GetOp(namespace=("test",), key="key1"),
]
results_reordered = await store.abatch(ops_reordered)
assert not failures
assert len(results_reordered) == 5
assert isinstance(results_reordered[0], list)
assert len(results_reordered[0]) == 2
assert len(results_reordered[0]) == 1
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 ("test", "foo") in results_reordered[2] and (
"test",
"bar",
) in results_reordered[2]
assert results_reordered[2] == [("test",)]
assert results_reordered[3] is None
assert isinstance(results_reordered[4], Item)
assert results_reordered[4].value == {"data": "value1"}
@@ -118,9 +146,26 @@ async def test_abatch_order(store: AsyncPostgresStore) -> None:
async def test_batch_get_ops(store: AsyncPostgresStore) -> None:
# Setup test data
await store.aput(("test",), "key1", {"data": "value1"})
await store.aput(("test",), "key2", {"data": "value2"})
mock_connection = store.conn
mock_cursor = MockAsyncCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
{
"key": "key2",
"value": '{"data": "value2"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.bar",
},
]
)
mock_connection.cursor.return_value = mock_cursor
ops = [
GetOp(namespace=("test",), key="key1"),
@@ -139,6 +184,10 @@ async def test_batch_get_ops(store: AsyncPostgresStore) -> None:
async def test_batch_put_ops(store: AsyncPostgresStore) -> None:
mock_connection = store.conn
mock_cursor = MockAsyncCursor([])
mock_connection.cursor.return_value = mock_cursor
ops = [
PutOp(namespace=("test",), key="key1", value={"data": "value1"}),
PutOp(namespace=("test",), key="key2", value={"data": "value2"}),
@@ -149,16 +198,30 @@ async def test_batch_put_ops(store: AsyncPostgresStore) -> None:
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
assert mock_cursor.execute.call_count == 2
async def test_batch_search_ops(store: AsyncPostgresStore) -> None:
# Setup test data
await store.aput(("test", "foo"), "key1", {"data": "value1"})
await store.aput(("test", "bar"), "key2", {"data": "value2"})
mock_connection = store.conn
mock_cursor = MockAsyncCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
{
"key": "key2",
"value": '{"data": "value2"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.bar",
},
]
)
mock_connection.cursor.return_value = mock_cursor
ops = [
SearchOp(
@@ -170,338 +233,297 @@ async def test_batch_search_ops(store: AsyncPostgresStore) -> None:
results = await store.abatch(ops)
assert len(results) == 2
assert len(results[0]) == 1 # Filtered results
assert len(results[1]) == 2 # All results
assert len(results[0]) == 2
assert len(results[1]) == 2
async def test_batch_list_namespaces_ops(store: AsyncPostgresStore) -> None:
# Setup test data
await store.aput(("test", "namespace1"), "key1", {"data": "value1"})
await store.aput(("test", "namespace2"), "key2", {"data": "value2"})
mock_connection = store.conn
mock_cursor = MockAsyncCursor(
[
{"truncated_prefix": b"\x01test.namespace1"},
{"truncated_prefix": b"\x01test.namespace2"},
]
)
mock_connection.cursor.return_value = mock_cursor
ops = [ListNamespacesOp(match_conditions=None, max_depth=None, limit=10, offset=0)]
results = await store.abatch(ops)
assert len(results) == 1
assert len(results[0]) == 2
assert ("test", "namespace1") in results[0]
assert ("test", "namespace2") in results[0]
assert results[0] == [("test", "namespace1"), ("test", "namespace2")]
@asynccontextmanager
async def _create_vector_store(
vector_type: str,
distance_type: str,
fake_embeddings: CharacterEmbeddings,
text_fields: Optional[list[str]] = None,
) -> AsyncIterator[AsyncPostgresStore]:
"""Create a store with vector search enabled."""
if sys.version_info < (3, 10):
pytest.skip("Async Postgres tests require Python 3.10+")
# The following use the actual DB connection
database = f"test_{uuid.uuid4().hex[:16]}"
uri_parts = DEFAULT_URI.split("/")
uri_base = "/".join(uri_parts[:-1])
query_params = ""
if "?" in uri_parts[-1]:
db_name, query_params = uri_parts[-1].split("?", 1)
query_params = "?" + query_params
conn_string = f"{uri_base}/{database}{query_params}"
admin_conn_string = DEFAULT_URI
index_config = {
"dims": fake_embeddings.dims,
"embed": fake_embeddings,
"ann_index_config": {
"vector_type": vector_type,
},
"distance_type": distance_type,
"text_fields": text_fields,
}
async with await AsyncConnection.connect(
admin_conn_string, autocommit=True
) as conn:
await conn.execute(f"CREATE DATABASE {database}")
try:
async with AsyncPostgresStore.from_conn_string(
conn_string,
index=index_config,
) as store:
class TestAsyncPostgresStore:
@pytest.fixture(autouse=True)
async def setup(self) -> None:
async with AsyncPostgresStore.from_conn_string(DEFAULT_URI) as store:
await store.setup()
yield store
finally:
async with await AsyncConnection.connect(
admin_conn_string, autocommit=True
) as conn:
await conn.execute(f"DROP DATABASE {database}")
async def test_basic_store_ops(self) -> None:
async with AsyncPostgresStore.from_conn_string(DEFAULT_URI) as store:
namespace = ("test", "documents")
item_id = "doc1"
item_value = {"title": "Test Document", "content": "Hello, World!"}
@pytest.fixture(
scope="function",
params=[
(vector_type, distance_type)
for vector_type in VECTOR_TYPES
for distance_type in (
["hamming"] if vector_type == "bit" else ["l2", "inner_product", "cosine"]
)
],
ids=lambda p: f"{p[0]}_{p[1]}",
)
async def vector_store(
request,
fake_embeddings: CharacterEmbeddings,
) -> AsyncIterator[AsyncPostgresStore]:
"""Create a store with vector search enabled."""
vector_type, distance_type = request.param
async with _create_vector_store(
vector_type, distance_type, fake_embeddings
) as store:
yield store
await store.aput(namespace, item_id, item_value)
item = await store.aget(namespace, item_id)
assert item
assert item.namespace == namespace
assert item.key == item_id
assert item.value == item_value
async def test_vector_store_initialization(
vector_store: AsyncPostgresStore, fake_embeddings: CharacterEmbeddings
) -> None:
"""Test store initialization with embedding config."""
assert vector_store.index_config is not None
assert vector_store.index_config["dims"] == fake_embeddings.dims
if isinstance(vector_store.index_config["embed"], Embeddings):
assert vector_store.index_config["embed"] == fake_embeddings
updated_value = {
"title": "Updated Test Document",
"content": "Hello, LangGraph!",
}
await store.aput(namespace, item_id, updated_value)
updated_item = await store.aget(namespace, item_id)
assert updated_item.value == updated_value
assert updated_item.updated_at > item.updated_at
different_namespace = ("test", "other_documents")
item_in_different_namespace = await store.aget(different_namespace, item_id)
assert item_in_different_namespace is None
async def test_vector_insert_with_auto_embedding(
vector_store: AsyncPostgresStore,
) -> None:
"""Test inserting items that get auto-embedded."""
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"}),
]
new_item_id = "doc2"
new_item_value = {"title": "Another Document", "content": "Greetings!"}
await store.aput(namespace, new_item_id, new_item_value)
for key, value in docs:
await vector_store.aput(("test",), key, value)
search_results = await store.asearch(["test"], limit=10)
items = search_results
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)
results = await vector_store.asearch(("test",), query="long text")
assert len(results) > 0
namespaces = await store.alist_namespaces(prefix=["test"])
assert ("test", "documents") in namespaces
doc_order = [r.key for r in results]
assert "doc2" in doc_order
assert "doc3" in doc_order
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
async def test_vector_update_with_embedding(vector_store: AsyncPostgresStore) -> None:
"""Test that updating items properly updates their embeddings."""
await vector_store.aput(("test",), "doc1", {"text": "zany zebra Xerxes"})
await vector_store.aput(("test",), "doc2", {"text": "something about dogs"})
await vector_store.aput(("test",), "doc3", {"text": "text about birds"})
empty_search_results = await store.asearch(["test"], limit=10)
assert len(empty_search_results) == 0
results_initial = await vector_store.asearch(("test",), query="Zany Xerxes")
assert len(results_initial) > 0
assert results_initial[0].key == "doc1"
initial_score = results_initial[0].score
async def test_list_namespaces(self) -> None:
async with AsyncPostgresStore.from_conn_string(DEFAULT_URI) 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),
]
await vector_store.aput(("test",), "doc1", {"text": "new text about dogs"})
for namespace in test_namespaces:
await store.aput(namespace, "dummy", {"content": "dummy"})
results_after = await vector_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 < initial_score
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])
results_new = await vector_store.asearch(("test",), query="new text about dogs")
for r in results_new:
if r.key == "doc1":
assert r.score > after_score
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]
)
# Don't index this one
await vector_store.aput(
("test",), "doc4", {"text": "new text about dogs"}, index=False
)
results_new = await vector_store.asearch(
("test",), query="new text about dogs", limit=3
)
assert not any(r.key == "doc4" for r in results_new)
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)
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
)
async def test_vector_search_with_filters(vector_store: AsyncPostgresStore) -> None:
"""Test combining vector search with filters."""
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}),
]
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)
for key, value in docs:
await vector_store.aput(("test",), key, value)
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,
)
results = await vector_store.asearch(
("test",), query="apple", filter={"color": "red"}
)
assert len(results) == 2
assert results[0].key == "doc1"
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(tuple(res) for res in max_depth_result))
== len(max_depth_result)
== 5
)
results = await vector_store.asearch(
("test",), query="car", filter={"color": "red"}
)
assert len(results) == 2
assert results[0].key == "doc2"
limit_result = await store.alist_namespaces(prefix=[test_pref], limit=3)
assert len(limit_result) == 3
results = await vector_store.asearch(
("test",), query="bbbbluuu", filter={"score": {"$gt": 3.2}}
)
assert len(results) == 3
assert results[0].key == "doc4"
offset_result = await store.alist_namespaces(prefix=[test_pref], offset=3)
assert len(offset_result) == len(test_namespaces) - 3
results = await vector_store.asearch(
("test",), query="apple", filter={"score": {"$gte": 4.0}, "color": "green"}
)
assert len(results) == 1
assert results[0].key == "doc3"
empty_prefix_result = await store.alist_namespaces(prefix=[test_pref])
assert len(empty_prefix_result) == len(test_namespaces)
assert set(tuple(ns) for ns in empty_prefix_result) == set(
tuple(ns) for ns in test_namespaces
)
for namespace in test_namespaces:
await store.adelete(namespace, "dummy")
async def test_vector_search_pagination(vector_store: AsyncPostgresStore) -> None:
"""Test pagination with vector search."""
for i in range(5):
await vector_store.aput(
("test",), f"doc{i}", {"text": f"test document number {i}"}
)
async def test_search(self):
async with AsyncPostgresStore.from_conn_string(DEFAULT_URI) as store:
test_namespaces = [
("test_search", "documents", "user1"),
("test_search", "documents", "user2"),
("test_search", "reports", "department1"),
("test_search", "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"]},
]
empty = await store.asearch(
(
"scoped",
"assistant_id",
"shared",
"6c5356f6-63ab-4158-868d-cd9fd14c736e",
),
limit=10,
offset=0,
)
assert len(empty) == 0
results_page1 = await vector_store.asearch(("test",), query="test", limit=2)
results_page2 = await vector_store.asearch(
("test",), query="test", limit=2, offset=2
)
for namespace, item in zip(test_namespaces, test_items):
await store.aput(namespace, f"item_{namespace[-1]}", item)
assert len(results_page1) == 2
assert len(results_page2) == 2
assert results_page1[0].key != results_page2[0].key
docs_result = await store.asearch(["test_search", "documents"])
assert len(docs_result) == 2
assert all([item.namespace[1] == "documents" for item in docs_result]), [
item.namespace for item in docs_result
]
all_results = await vector_store.asearch(("test",), query="test", limit=10)
assert len(all_results) == 5
reports_result = await store.asearch(["test_search", "reports"])
assert len(reports_result) == 2
assert all(item.namespace[1] == "reports" for item in reports_result)
limited_result = await store.asearch(["test_search"], limit=2)
assert len(limited_result) == 2
offset_result = await store.asearch(["test_search"])
assert len(offset_result) == 4
async def test_vector_search_edge_cases(vector_store: AsyncPostgresStore) -> None:
"""Test edge cases in vector search."""
await vector_store.aput(("test",), "doc1", {"text": "test document"})
offset_result = await store.asearch(["test_search"], offset=2)
assert len(offset_result) == 2
assert all(item not in limited_result for item in offset_result)
perfect_match = await vector_store.asearch(("test",), query="text test document")
perfect_score = perfect_match[0].score
john_doe_result = await store.asearch(
["test_search"], filter={"author": "John Doe"}
)
assert len(john_doe_result) == 2
assert all(item.value["author"] == "John Doe" for item in john_doe_result)
results = await vector_store.asearch(("test",), query="")
assert len(results) == 1
assert results[0].score is None
draft_result = await store.asearch(
["test_search"], filter={"tags": ["draft"]}
)
assert len(draft_result) == 2
assert all("draft" in item.value["tags"] for item in draft_result)
results = await vector_store.asearch(("test",), query=None)
assert len(results) == 1
assert results[0].score is None
page1 = await store.asearch(["test_search"], limit=2, offset=0)
page2 = await store.asearch(["test_search"], limit=2, offset=2)
all_items = page1 + page2
assert len(all_items) == 4
assert len(set(item.key for item in all_items)) == 4
empty = await store.asearch(
(
"scoped",
"assistant_id",
"shared",
"again",
"maybe",
"some-long",
"6be5cb0e-2eb4-42e6-bb6b-fba3c269db25",
),
limit=10,
offset=0,
)
assert len(empty) == 0
long_query = "foo " * 100
results = await vector_store.asearch(("test",), query=long_query)
assert len(results) == 1
assert results[0].score < perfect_score
# Test with a namespace beginning with a number (like a UUID)
uuid_namespace = (str(uuid.uuid4()), "documents")
uuid_item_id = "uuid_doc"
uuid_item_value = {
"title": "UUID Document",
"content": "This document has a UUID namespace.",
}
special_query = "test!@#$%^&*()"
results = await vector_store.asearch(("test",), query=special_query)
assert len(results) == 1
assert results[0].score < perfect_score
# Insert the item with the UUID namespace
await store.aput(uuid_namespace, uuid_item_id, uuid_item_value)
# Retrieve the item to verify it was stored correctly
retrieved_item = await store.aget(uuid_namespace, uuid_item_id)
assert retrieved_item is not None
assert retrieved_item.namespace == uuid_namespace
assert retrieved_item.key == uuid_item_id
assert retrieved_item.value == uuid_item_value
@pytest.mark.parametrize(
"vector_type,distance_type",
[
*itertools.product(["vector", "halfvec"], ["cosine", "inner_product", "l2"]),
],
)
async def test_embed_with_path(
request: Any,
fake_embeddings: CharacterEmbeddings,
vector_type: str,
distance_type: str,
) -> None:
"""Test vector search with specific text fields in Postgres store."""
async with _create_vector_store(
vector_type,
distance_type,
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)
# Search for the item using the UUID namespace
search_result = await store.asearch([uuid_namespace[0]])
assert len(search_result) == 1
assert search_result[0].key == uuid_item_id
assert search_result[0].value == uuid_item_value
# 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
ascore = results[0].score
bscore = results[1].score
assert ascore == pytest.approx(bscore, abs=1e-3)
# Clean up: delete the item with the UUID namespace
await store.adelete(uuid_namespace, uuid_item_id)
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
assert ascore == pytest.approx(results[0].score, abs=1e-3)
# Verify the item was deleted
deleted_item = await store.aget(uuid_namespace, uuid_item_id)
assert deleted_item is None
# 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 < ascore
assert results[1].score < ascore
@pytest.mark.parametrize(
"vector_type,distance_type",
[
*itertools.product(["vector", "halfvec"], ["cosine", "inner_product", "l2"]),
],
)
async def test_search_sorting(
request: Any,
fake_embeddings: CharacterEmbeddings,
vector_type: str,
distance_type: str,
) -> None:
"""Test operation-level field configuration for vector search."""
async with _create_vector_store(
vector_type,
distance_type,
fake_embeddings,
text_fields=["key1"], # Default fields that won't match our test data
) as store:
amatch = {
"key1": "mmm",
}
await store.aput(("test", "M"), "M", amatch)
N = 100
for i in range(N):
await store.aput(("test", "A"), f"A{i}", {"key1": "no"})
for i in range(N):
await store.aput(("test", "Z"), f"Z{i}", {"key1": "no"})
results = await store.asearch(("test",), query="mmm", limit=10)
assert len(results) == 10
assert len(set(r.key for r in results)) == 10
assert results[0].key == "M"
assert results[0].score > results[1].score
for namespace in test_namespaces:
await store.adelete(namespace, f"item_{namespace[-1]}")
+312 -542
View File
@@ -1,125 +1,174 @@
# type: ignore
from contextlib import contextmanager
from typing import Any, Optional
from uuid import uuid4
import uuid
from datetime import datetime
from typing import Any
from unittest.mock import MagicMock
import pytest
from langchain_core.embeddings import Embeddings
from psycopg import Connection
from conftest import DEFAULT_URI # type: ignore
from langgraph.store.base import (
GetOp,
Item,
ListNamespacesOp,
MatchCondition,
PutOp,
SearchOp,
)
from langgraph.store.base import GetOp, Item, ListNamespacesOp, PutOp, SearchOp
from langgraph.store.postgres import PostgresStore
from tests.conftest import (
DEFAULT_URI,
VECTOR_TYPES,
CharacterEmbeddings,
)
@pytest.fixture(scope="function", params=["default", "pipe", "pool"])
def store(request) -> PostgresStore:
database = f"test_{uuid4().hex[:16]}"
uri_parts = DEFAULT_URI.split("/")
uri_base = "/".join(uri_parts[:-1])
query_params = ""
if "?" in uri_parts[-1]:
db_name, query_params = uri_parts[-1].split("?", 1)
query_params = "?" + query_params
class MockCursor:
def __init__(self, fetch_result: Any) -> None:
self.fetch_result = fetch_result
self.execute = MagicMock()
self.fetchall = MagicMock(return_value=self.fetch_result)
conn_string = f"{uri_base}/{database}{query_params}"
admin_conn_string = DEFAULT_URI
with Connection.connect(admin_conn_string, autocommit=True) as conn:
conn.execute(f"CREATE DATABASE {database}")
try:
with PostgresStore.from_conn_string(conn_string) as store:
store.setup()
class MockConnection:
def __init__(self) -> None:
self.cursor = MagicMock()
self.pipeline = MagicMock()
if request.param == "pipe":
with PostgresStore.from_conn_string(conn_string, pipeline=True) as store:
yield store
elif request.param == "pool":
with PostgresStore.from_conn_string(
conn_string, pool_config={"min_size": 1, "max_size": 10}
) as store:
yield store
else: # default
with PostgresStore.from_conn_string(conn_string) as store:
yield store
finally:
with Connection.connect(admin_conn_string, autocommit=True) as conn:
conn.execute(f"DROP DATABASE {database}")
@pytest.fixture
def mock_connection() -> MockConnection:
return MockConnection()
@pytest.fixture
def store(mock_connection: MockConnection) -> PostgresStore:
return PostgresStore(mock_connection)
def test_batch_order(store: PostgresStore) -> None:
# Setup test data
store.put(("test", "foo"), "key1", {"data": "value1"})
store.put(("test", "bar"), "key2", {"data": "value2"})
mock_connection = store.conn
mock_get_cursor = MockCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
{
"key": "key2",
"value": '{"data": "value2"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.bar",
},
]
)
mock_search_cursor = MockCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
]
)
mock_list_namespaces_cursor = MockCursor(
[
{"truncated_prefix": b"\x01test"},
]
)
failures = []
def cursor_side_effect(binary: bool = False) -> Any:
cursor = MagicMock()
def execute_side_effect(query: str, *params: Any) -> None:
# My super sophisticated database.
if "SELECT prefix, key, value" in query:
cursor.fetchall = mock_search_cursor.fetchall
elif "SELECT DISTINCT ON (truncated_prefix)" in query:
cursor.fetchall = mock_list_namespaces_cursor.fetchall
elif "WHERE prefix = %s AND key" in query:
cursor.fetchall = mock_get_cursor.fetchall
elif "INSERT INTO " in query:
pass
else:
e = ValueError(f"Unmatched query: {query}")
failures.append(e)
raise e
cursor.execute = MagicMock(side_effect=execute_side_effect)
return cursor
mock_connection.cursor.side_effect = cursor_side_effect
ops = [
GetOp(namespace=("test", "foo"), key="key1"),
PutOp(namespace=("test", "bar"), key="key2", value={"data": "value2"}),
GetOp(namespace=("test",), key="key1"),
PutOp(namespace=("test",), 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 = store.batch(ops)
assert not failures
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 results[1] is None
assert isinstance(results[2], list)
assert len(results[2]) == 1
assert isinstance(results[3], list)
assert len(results[3]) > 0 # Should contain at least our test namespaces
assert results[4] is None # Non-existent key returns None
assert results[3] == [("test",)]
assert results[4] is None
# Test reordered operations
ops_reordered = [
SearchOp(namespace_prefix=("test",), filter=None, limit=5, offset=0),
GetOp(namespace=("test", "bar"), key="key2"),
GetOp(namespace=("test",), 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"),
GetOp(namespace=("test",), key="key1"),
]
results_reordered = store.batch(ops_reordered)
assert not failures
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 len(results_reordered[0]) == 1
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 results_reordered[2] == [("test",)]
assert results_reordered[3] is None
assert isinstance(results_reordered[4], Item)
assert results_reordered[4].value == {"data": "value1"}
assert results_reordered[4].key == "key1"
def test_batch_get_ops(store: PostgresStore) -> None:
# Setup test data
store.put(("test",), "key1", {"data": "value1"})
store.put(("test",), "key2", {"data": "value2"})
mock_connection = store.conn
mock_cursor = MockCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
{
"key": "key2",
"value": '{"data": "value2"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.bar",
},
]
)
mock_connection.cursor.return_value = mock_cursor
ops = [
GetOp(namespace=("test",), key="key1"),
GetOp(namespace=("test",), key="key2"),
GetOp(namespace=("test",), key="key3"), # Non-existent key
GetOp(namespace=("test",), key="key3"),
]
results = store.batch(ops)
@@ -133,90 +182,75 @@ def test_batch_get_ops(store: PostgresStore) -> None:
def test_batch_put_ops(store: PostgresStore) -> None:
mock_connection = store.conn
mock_cursor = MockCursor([])
mock_connection.cursor.return_value = mock_cursor
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
PutOp(namespace=("test",), key="key3", value=None),
]
results = store.batch(ops)
assert len(results) == 3
assert all(result is None for result in results)
# Verify the puts worked
item1 = store.get(("test",), "key1")
item2 = store.get(("test",), "key2")
item3 = store.get(("test",), "key3")
assert item1 and item1.value == {"data": "value1"}
assert item2 and item2.value == {"data": "value2"}
assert item3 is None
assert mock_cursor.execute.call_count == 2
def test_batch_search_ops(store: PostgresStore) -> None:
# Setup test data
test_data = [
(("test", "foo"), "key1", {"data": "value1", "tag": "a"}),
(("test", "bar"), "key2", {"data": "value2", "tag": "a"}),
(("test", "baz"), "key3", {"data": "value3", "tag": "b"}),
]
for namespace, key, value in test_data:
store.put(namespace, key, value)
mock_connection = store.conn
mock_cursor = MockCursor(
[
{
"key": "key1",
"value": '{"data": "value1"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.foo",
},
{
"key": "key2",
"value": '{"data": "value2"}',
"created_at": datetime.now(),
"updated_at": datetime.now(),
"prefix": "test.bar",
},
]
)
mock_connection.cursor.return_value = mock_cursor
ops = [
SearchOp(namespace_prefix=("test",), filter={"tag": "a"}, limit=10, offset=0),
SearchOp(namespace_prefix=("test",), filter=None, limit=2, offset=0),
SearchOp(namespace_prefix=("test", "foo"), filter=None, limit=10, offset=0),
SearchOp(
namespace_prefix=("test",), filter={"data": "value1"}, limit=10, offset=0
),
SearchOp(namespace_prefix=("test",), filter=None, limit=5, offset=0),
]
results = store.batch(ops)
assert len(results) == 3
# First search should find items with tag "a"
assert len(results) == 2
assert len(results[0]) == 2
assert all(item.value["tag"] == "a" for item in results[0])
# Second search should return first 2 items
assert len(results[1]) == 2
# Third search should only find items in test/foo namespace
assert len(results[2]) == 1
assert results[2][0].namespace == ("test", "foo")
def test_batch_list_namespaces_ops(store: PostgresStore) -> None:
# Setup test data with various namespaces
test_data = [
(("test", "documents", "public"), "doc1", {"content": "public doc"}),
(("test", "documents", "private"), "doc2", {"content": "private doc"}),
(("test", "images", "public"), "img1", {"content": "public image"}),
(("prod", "documents", "public"), "doc3", {"content": "prod doc"}),
]
for namespace, key, value in test_data:
store.put(namespace, key, value)
mock_connection = store.conn
mock_cursor = MockCursor(
[
{"truncated_prefix": b"\x01test.namespace1"},
{"truncated_prefix": b"\x01test.namespace2"},
]
)
mock_connection.cursor.return_value = mock_cursor
ops = [
ListNamespacesOp(match_conditions=None, max_depth=None, limit=10, offset=0),
ListNamespacesOp(match_conditions=None, max_depth=2, limit=10, offset=0),
ListNamespacesOp(
match_conditions=[MatchCondition("suffix", "public")],
max_depth=None,
limit=10,
offset=0,
),
]
ops = [ListNamespacesOp(match_conditions=None, max_depth=None, limit=10, offset=0)]
results = store.batch(ops)
assert len(results) == 3
# First operation should list all namespaces
assert len(results[0]) == len(test_data)
# Second operation should only return namespaces up to depth 2
assert all(len(ns) <= 2 for ns in results[1])
# Third operation should only return namespaces ending with "public"
assert all(ns[-1] == "public" for ns in results[2])
assert len(results) == 1
assert results[0] == [("test", "namespace1"), ("test", "namespace2")]
class TestPostgresStore:
@@ -239,459 +273,195 @@ class TestPostgresStore:
assert item.key == item_id
assert item.value == item_value
# Test update
updated_value = {"title": "Updated Document", "content": "Hello, Updated!"}
updated_value = {
"title": "Updated Test Document",
"content": "Hello, LangGraph!",
}
store.put(namespace, item_id, updated_value)
updated_item = store.get(namespace, item_id)
assert updated_item.value == updated_value
assert updated_item.updated_at > item.updated_at
# Test get from non-existent namespace
different_namespace = ("test", "other_documents")
item_in_different_namespace = store.get(different_namespace, item_id)
assert item_in_different_namespace is None
# Test delete
new_item_id = "doc2"
new_item_value = {"title": "Another Document", "content": "Greetings!"}
store.put(namespace, new_item_id, new_item_value)
search_results = store.search(["test"], limit=10)
items = search_results
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 = store.list_namespaces(prefix=["test"])
assert ("test", "documents") in namespaces
store.delete(namespace, item_id)
store.delete(namespace, new_item_id)
deleted_item = store.get(namespace, item_id)
assert deleted_item is None
deleted_item = store.get(namespace, new_item_id)
assert deleted_item is None
empty_search_results = store.search(["test"], limit=10)
assert len(empty_search_results) == 0
def test_list_namespaces(self) -> None:
with PostgresStore.from_conn_string(DEFAULT_URI) as store:
# Create test data with various namespaces
test_pref = str(uuid.uuid4())
test_namespaces = [
("test", "documents", "public"),
("test", "documents", "private"),
("test", "images", "public"),
("test", "images", "private"),
("prod", "documents", "public"),
("prod", "documents", "private"),
(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),
]
# Insert test data
for namespace in test_namespaces:
store.put(namespace, "dummy", {"content": "dummy"})
# Test listing with various filters
all_namespaces = store.list_namespaces()
assert len(all_namespaces) == len(test_namespaces)
prefix_result = store.list_namespaces(prefix=[test_pref, "test"])
assert len(prefix_result) == 4
assert all([ns[1] == "test" for ns in prefix_result])
# Test prefix filtering
test_prefix_namespaces = store.list_namespaces(prefix=["test"])
assert len(test_prefix_namespaces) == 4
assert all(ns[0] == "test" for ns in test_prefix_namespaces)
specific_prefix_result = store.list_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
public_namespaces = store.list_namespaces(suffix=["public"])
assert len(public_namespaces) == 3
assert all(ns[-1] == "public" for ns in public_namespaces)
suffix_result = store.list_namespaces(suffix=["public", test_pref])
assert len(suffix_result) == 4
assert all(ns[-2] == "public" for ns in suffix_result)
# Test max depth
depth_2_namespaces = store.list_namespaces(max_depth=2)
assert all(len(ns) <= 2 for ns in depth_2_namespaces)
prefix_suffix_result = store.list_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 pagination
paginated_namespaces = store.list_namespaces(limit=3)
assert len(paginated_namespaces) == 3
wildcard_prefix_result = store.list_namespaces(
prefix=[test_pref, "*", "documents"]
)
assert len(wildcard_prefix_result) == 5
assert all(ns[2] == "documents" for ns in wildcard_prefix_result)
wildcard_suffix_result = store.list_namespaces(
suffix=["*", "public", test_pref]
)
assert len(wildcard_suffix_result) == 4
assert all(ns[-2] == "public" for ns in wildcard_suffix_result)
wildcard_single = store.list_namespaces(
suffix=["some", "*", "public", test_pref]
)
assert len(wildcard_single) == 1
assert wildcard_single[0] == (
test_pref,
"prod",
"documents",
"some",
"nesting",
"public",
test_pref,
)
max_depth_result = store.list_namespaces(max_depth=3)
assert all([len(ns) <= 3 for ns in max_depth_result])
max_depth_result = store.list_namespaces(
max_depth=4, prefix=[test_pref, "*", "documents"]
)
assert (
len(set(tuple(res) for res in max_depth_result))
== len(max_depth_result)
== 5
)
limit_result = store.list_namespaces(prefix=[test_pref], limit=3)
assert len(limit_result) == 3
offset_result = store.list_namespaces(prefix=[test_pref], offset=3)
assert len(offset_result) == len(test_namespaces) - 3
empty_prefix_result = store.list_namespaces(prefix=[test_pref])
assert len(empty_prefix_result) == len(test_namespaces)
assert set(tuple(ns) for ns in empty_prefix_result) == set(
tuple(ns) for ns in test_namespaces
)
# Cleanup
for namespace in test_namespaces:
store.delete(namespace, "dummy")
def test_search(self) -> None:
def test_search(self):
with PostgresStore.from_conn_string(DEFAULT_URI) as store:
# Create test data
test_data = [
(
("test", "docs"),
"doc1",
{"title": "First Doc", "author": "Alice", "tags": ["important"]},
),
(
("test", "docs"),
"doc2",
{"title": "Second Doc", "author": "Bob", "tags": ["draft"]},
),
(
("test", "images"),
"img1",
{"title": "Image 1", "author": "Alice", "tags": ["final"]},
),
test_namespaces = [
("test_search", "documents", "user1"),
("test_search", "documents", "user2"),
("test_search", "reports", "department1"),
("test_search", "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"]},
]
for namespace, key, value in test_data:
store.put(namespace, key, value)
for namespace, item in zip(test_namespaces, test_items):
store.put(namespace, f"item_{namespace[-1]}", item)
# Test basic search
all_items = store.search(["test"])
assert len(all_items) == 3
docs_result = store.search(["test_search", "documents"])
assert len(docs_result) == 2
assert all(
[item.namespace[1] == "documents" for item in docs_result]
), docs_result
# Test namespace filtering
docs_items = store.search(["test", "docs"])
assert len(docs_items) == 2
assert all(item.namespace == ("test", "docs") for item in docs_items)
reports_result = store.search(["test_search", "reports"])
assert len(reports_result) == 2
assert all(item.namespace[1] == "reports" for item in reports_result)
# Test value filtering
alice_items = store.search(["test"], filter={"author": "Alice"})
assert len(alice_items) == 2
assert all(item.value["author"] == "Alice" for item in alice_items)
limited_result = store.search(["test_search"], limit=2)
assert len(limited_result) == 2
offset_result = store.search(["test_search"])
assert len(offset_result) == 4
# Test pagination
paginated_items = store.search(["test"], limit=2)
assert len(paginated_items) == 2
offset_result = store.search(["test_search"], offset=2)
assert len(offset_result) == 2
assert all(item not in limited_result for item in offset_result)
offset_items = store.search(["test"], offset=2)
assert len(offset_items) == 1
john_doe_result = store.search(
["test_search"], filter={"author": "John Doe"}
)
assert len(john_doe_result) == 2
assert all(item.value["author"] == "John Doe" for item in john_doe_result)
# Cleanup
for namespace, key, _ in test_data:
store.delete(namespace, key)
draft_result = store.search(["test_search"], filter={"tags": ["draft"]})
assert len(draft_result) == 2
assert all("draft" in item.value["tags"] for item in draft_result)
page1 = store.search(["test_search"], limit=2, offset=0)
page2 = store.search(["test_search"], limit=2, offset=2)
all_items = page1 + page2
assert len(all_items) == 4
assert len(set(item.key for item in all_items)) == 4
@contextmanager
def _create_vector_store(
vector_type: str,
distance_type: str,
fake_embeddings: Embeddings,
text_fields: Optional[list[str]] = None,
) -> PostgresStore:
"""Create a store with vector search enabled."""
database = f"test_{uuid4().hex[:16]}"
uri_parts = DEFAULT_URI.split("/")
uri_base = "/".join(uri_parts[:-1])
query_params = ""
if "?" in uri_parts[-1]:
db_name, query_params = uri_parts[-1].split("?", 1)
query_params = "?" + query_params
conn_string = f"{uri_base}/{database}{query_params}"
admin_conn_string = DEFAULT_URI
index_config = {
"dims": fake_embeddings.dims,
"embed": fake_embeddings,
"ann_index_config": {
"vector_type": vector_type,
},
"distance_type": distance_type,
"text_fields": text_fields,
}
with Connection.connect(admin_conn_string, autocommit=True) as conn:
conn.execute(f"CREATE DATABASE {database}")
try:
with PostgresStore.from_conn_string(
conn_string,
index=index_config,
) as store:
store.setup()
yield store
finally:
with Connection.connect(admin_conn_string, autocommit=True) as conn:
conn.execute(f"DROP DATABASE {database}")
@pytest.fixture(
scope="function",
params=[
(vector_type, distance_type)
for vector_type in VECTOR_TYPES
for distance_type in (
["hamming"] if vector_type == "bit" else ["l2", "inner_product", "cosine"]
)
],
ids=lambda p: f"{p[0]}_{p[1]}",
)
def vector_store(
request,
fake_embeddings: Embeddings,
) -> PostgresStore:
"""Create a store with vector search enabled."""
vector_type, distance_type = request.param
with _create_vector_store(vector_type, distance_type, fake_embeddings) as store:
yield store
def test_vector_store_initialization(
vector_store: PostgresStore, fake_embeddings: CharacterEmbeddings
) -> None:
"""Test store initialization with embedding config."""
# Store should be initialized with embedding config
assert vector_store.index_config is not None
assert vector_store.index_config["dims"] == fake_embeddings.dims
assert vector_store.index_config["embed"] == fake_embeddings
def test_vector_insert_with_auto_embedding(vector_store: PostgresStore) -> None:
"""Test inserting items that get auto-embedded."""
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:
vector_store.put(("test",), key, value)
results = vector_store.search(("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
def test_vector_update_with_embedding(vector_store: PostgresStore) -> None:
"""Test that updating items properly updates their embeddings."""
vector_store.put(("test",), "doc1", {"text": "zany zebra Xerxes"})
vector_store.put(("test",), "doc2", {"text": "something about dogs"})
vector_store.put(("test",), "doc3", {"text": "text about birds"})
results_initial = vector_store.search(("test",), query="Zany Xerxes")
assert len(results_initial) > 0
assert results_initial[0].key == "doc1"
initial_score = results_initial[0].score
vector_store.put(("test",), "doc1", {"text": "new text about dogs"})
results_after = vector_store.search(("test",), query="Zany Xerxes")
after_score = next((r.score for r in results_after if r.key == "doc1"), 0.0)
assert after_score < initial_score
results_new = vector_store.search(("test",), query="new text about dogs")
for r in results_new:
if r.key == "doc1":
assert r.score > after_score
# Don't index this one
vector_store.put(("test",), "doc4", {"text": "new text about dogs"}, index=False)
results_new = vector_store.search(("test",), query="new text about dogs", limit=3)
assert not any(r.key == "doc4" for r in results_new)
def test_vector_search_with_filters(vector_store: PostgresStore) -> None:
"""Test combining vector search with filters."""
# Insert test documents
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:
vector_store.put(("test",), key, value)
results = vector_store.search(("test",), query="apple", filter={"color": "red"})
assert len(results) == 2
assert results[0].key == "doc1"
results = vector_store.search(("test",), query="car", filter={"color": "red"})
assert len(results) == 2
assert results[0].key == "doc2"
results = vector_store.search(
("test",), query="bbbbluuu", filter={"score": {"$gt": 3.2}}
)
assert len(results) == 3
assert results[0].key == "doc4"
# Multiple filters
results = vector_store.search(
("test",), query="apple", filter={"score": {"$gte": 4.0}, "color": "green"}
)
assert len(results) == 1
assert results[0].key == "doc3"
def test_vector_search_pagination(vector_store: PostgresStore) -> None:
"""Test pagination with vector search."""
# Insert multiple similar documents
for i in range(5):
vector_store.put(("test",), f"doc{i}", {"text": f"test document number {i}"})
# Test with different page sizes
results_page1 = vector_store.search(("test",), query="test", limit=2)
results_page2 = vector_store.search(("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
# Get all results
all_results = vector_store.search(("test",), query="test", limit=10)
assert len(all_results) == 5
def test_vector_search_edge_cases(vector_store: PostgresStore) -> None:
"""Test edge cases in vector search."""
vector_store.put(("test",), "doc1", {"text": "test document"})
results = vector_store.search(("test",), query="")
assert len(results) == 1
results = vector_store.search(("test",), query=None)
assert len(results) == 1
long_query = "test " * 100
results = vector_store.search(("test",), query=long_query)
assert len(results) == 1
special_query = "test!@#$%^&*()"
results = vector_store.search(("test",), query=special_query)
assert len(results) == 1
@pytest.mark.parametrize(
"vector_type,distance_type",
[
("vector", "cosine"),
("vector", "inner_product"),
("halfvec", "cosine"),
("halfvec", "inner_product"),
],
)
def test_embed_with_path_sync(
request: Any,
fake_embeddings: CharacterEmbeddings,
vector_type: str,
distance_type: str,
) -> None:
"""Test vector search with specific text fields in Postgres store."""
with _create_vector_store(
vector_type,
distance_type,
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",
}
store.put(("test",), "doc1", doc1)
store.put(("test",), "doc2", doc2)
# doc2.key3 and doc1.key1 both would have the highest score
results = store.search(("test",), query="xxx")
assert len(results) == 2
assert results[0].key != results[1].key
ascore = results[0].score
bscore = results[1].score
assert ascore == pytest.approx(bscore, abs=1e-3)
# ~Only match doc2
results = store.search(("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
assert ascore == pytest.approx(results[0].score, abs=1e-3)
# ~Only match doc1
results = store.search(("test",), query="zzz")
assert len(results) == 2
assert results[0].key != results[1].key
assert results[0].key == "doc1"
assert results[0].score > results[1].score
assert ascore == pytest.approx(results[0].score, abs=1e-3)
# Un-indexed - will have low results for both. Not zero (because we're projecting)
# but less than the above.
results = store.search(("test",), query="www")
assert len(results) == 2
assert results[0].key != results[1].key
assert results[0].score < ascore
assert results[1].score < ascore
@pytest.mark.parametrize(
"vector_type,distance_type",
[
("vector", "cosine"),
("vector", "inner_product"),
("halfvec", "cosine"),
("halfvec", "inner_product"),
],
)
def test_embed_with_path_operation_config(
request: Any,
fake_embeddings: CharacterEmbeddings,
vector_type: str,
distance_type: str,
) -> None:
"""Test operation-level field configuration for vector search."""
with _create_vector_store(
vector_type,
distance_type,
fake_embeddings,
text_fields=["key17"], # Default fields that won't match our test data
) as store:
doc3 = {
"key0": "aaa",
"key1": "bbb",
"key2": "ccc",
"key3": "ddd",
}
doc4 = {
"key0": "eee",
"key1": "bbb", # Same as doc3.key1
"key2": "fff",
"key3": "ggg",
}
store.put(("test",), "doc3", doc3, index=["key0", "key1"])
store.put(("test",), "doc4", doc4, index=["key1", "key3"])
results = store.search(("test",), query="aaa")
assert len(results) == 2
assert results[0].key == "doc3"
assert len(set(r.key for r in results)) == 2
assert results[0].score > results[1].score
results = store.search(("test",), query="ggg")
assert len(results) == 2
assert results[0].key == "doc4"
assert results[0].score > results[1].score
results = store.search(("test",), query="bbb")
assert len(results) == 2
assert results[0].key != results[1].key
assert results[0].score == pytest.approx(results[1].score, abs=1e-3)
results = store.search(("test",), query="ccc")
assert len(results) == 2
assert all(
r.score < 0.9 for r in results
) # Unindexed field should have low scores
# Test index=False behavior
doc5 = {
"key0": "hhh",
"key1": "iii",
}
store.put(("test",), "doc5", doc5, index=False)
results = store.search(("test",))
assert len(results) == 3
assert all(r.score is None for r in results)
assert any(r.key == "doc5" for r in results)
results = store.search(("test",), query="hhh")
# TODO: We don't currently fill in additional results if there are not enough
# returned during vector search.
# assert len(results) == 3
# doc5_result = next(r for r in results if r.key == "doc5")
# assert doc5_result.score is None
for namespace in test_namespaces:
store.delete(namespace, f"item_{namespace[-1]}")
+1 -1
View File
@@ -1,6 +1,7 @@
from typing import Any
import pytest
from conftest import DEFAULT_URI # type: ignore
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
@@ -10,7 +11,6 @@ from langgraph.checkpoint.base import (
empty_checkpoint,
)
from langgraph.checkpoint.postgres import PostgresSaver
from tests.conftest import DEFAULT_URI
class TestPostgresSaver:
+2 -4
View File
@@ -4,13 +4,11 @@
# TESTING AND COVERAGE
######################
TEST ?= .
test:
poetry run pytest $(TEST)
poetry run pytest tests
test_watch:
poetry run ptw $(TEST)
poetry run ptw .
######################
# LINTING AND FORMATTING
+105 -548
View File
@@ -1,27 +1,12 @@
"""Base classes and types for persistent key-value stores.
Stores provide long-term memory that persists across threads and conversations.
Supports hierarchical namespaces, key-value storage, and optional vector search.
Core types:
- BaseStore: Store interface with sync/async operations
- Item: Stored key-value pairs with metadata
- Op: Get/Put/Search/List operations
Stores enable persistence and memory that can be shared across threads,
scoped to user IDs, assistant IDs, or other arbitrary namespaces.
"""
from abc import ABC, abstractmethod
from datetime import datetime
from typing import Any, Iterable, Literal, NamedTuple, Optional, TypedDict, Union, cast
from langchain_core.embeddings import Embeddings
from langgraph.store.base.embed import (
AEmbeddingsFunc,
EmbeddingsFunc,
ensure_embeddings,
get_text_at_path,
tokenize_path,
)
from typing import Any, Iterable, Literal, NamedTuple, Optional, Union, cast
class Item:
@@ -88,415 +73,112 @@ class Item:
}
class SearchItem(Item):
"""Represents a result item with additional response metadata."""
__slots__ = ("score",)
def __init__(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
created_at: datetime,
updated_at: datetime,
score: Optional[float] = None,
) -> None:
"""Initialize a result item.
Args:
namespace: Hierarchical path to the item.
key: Unique identifier within the namespace.
value: The stored value.
created_at: When the item was first created.
updated_at: When the item was last updated.
score: Relevance/similarity score if from a ranked operation.
"""
super().__init__(
value=value,
key=key,
namespace=namespace,
created_at=created_at,
updated_at=updated_at,
)
self.score = score
def dict(self) -> dict:
result = super().dict()
result["score"] = self.score
return result
class GetOp(NamedTuple):
"""Operation to retrieve a specific item by its namespace and key.
This operation allows precise retrieval of stored items using their full path
(namespace) and unique identifier (key) combination.
??? example "Examples"
Basic item retrieval:
```python
GetOp(namespace=("users", "profiles"), key="user123")
GetOp(namespace=("cache", "embeddings"), key="doc456")
```
"""
"""Operation to retrieve an item by namespace and key."""
namespace: tuple[str, ...]
"""Hierarchical path that uniquely identifies the item's location.
??? example "Examples"
```python
("users",) # Root level users namespace
("users", "profiles") # Profiles within users namespace
```
"""
"""Hierarchical path for the item."""
key: str
"""Unique identifier for the item within its specific namespace.
??? example "Examples"
```python
"user123" # For a user profile
"doc456" # For a document
```
"""
"""Unique identifier within the namespace."""
class SearchOp(NamedTuple):
"""Operation to search for items within a specified namespace hierarchy.
This operation supports both structured filtering and natural language search
within a given namespace prefix. It provides pagination through limit and offset
parameters.
Note:
Natural language search support depends on your store implementation.
??? example "Examples"
Search with filters and pagination:
```python
SearchOp(
namespace_prefix=("documents",),
filter={"type": "report", "status": "active"},
limit=5,
offset=10
)
```
Natural language search:
```python
SearchOp(
namespace_prefix=("users", "content"),
query="technical documentation about APIs",
limit=20
)
```
"""
"""Operation to search for items within a namespace prefix."""
namespace_prefix: tuple[str, ...]
"""Hierarchical path prefix defining the search scope.
??? example "Examples"
```python
() # Search entire store
("documents",) # Search all documents
("users", "content") # Search within user content
```
"""
"""Hierarchical path prefix to search within."""
filter: Optional[dict[str, Any]] = None
"""Key-value pairs for filtering results based on exact matches or comparison operators.
The filter supports both exact matches and operator-based comparisons.
Supported Operators:
- $eq: Equal to (same as direct value comparison)
- $ne: Not equal to
- $gt: Greater than
- $gte: Greater than or equal to
- $lt: Less than
- $lte: Less than or equal to
??? example "Examples"
Simple exact match:
```python
{"status": "active"}
```
Comparison operators:
```python
{"score": {"$gt": 4.99}} # Score greater than 4.99
```
Multiple conditions:
```python
{
"score": {"$gte": 3.0},
"color": "red"
}
```
Note:
Comparison operator support depends on your store implementation.
"""
"""Key-value pairs to filter results."""
limit: int = 10
"""Maximum number of items to return in the search results."""
"""Maximum number of items to return."""
offset: int = 0
"""Number of matching items to skip for pagination."""
"""Number of items to skip before returning results."""
query: Optional[str] = None
"""Natural language search query for semantic search capabilities.
??? example "Examples"
- "technical documentation about REST APIs"
- "machine learning papers from 2023"
class PutOp(NamedTuple):
"""Operation to store, update, or delete an item."""
namespace: tuple[str, ...]
"""Hierarchical path for the item.
Represented as a tuple of strings, allowing for nested categorization.
For example: ("documents", "user123")
"""
key: str
"""Unique identifier for the document.
Should be distinct within its namespace.
"""
value: Optional[dict[str, Any]]
"""Data to be stored, or None to delete the item.
Schema:
- Should be a dictionary where:
- Keys are strings representing field names
- Values can be of any serializable type
- If None, it indicates that the item should be deleted
"""
# Type representing a namespace path that can include wildcards
NamespacePath = tuple[Union[str, Literal["*"]], ...]
"""A tuple representing a namespace path that can include wildcards.
NameSpacePath = tuple[Union[str, Literal["*"]], ...]
Examples:
("users",) # Exact users namespace
("documents", "*") # Any sub-namespace under documents
("cache", "*", "v1") # Any cache category with v1 version
"""
# Type for specifying how to match namespaces
NamespaceMatchType = Literal["prefix", "suffix"]
"""Specifies how to match namespace paths.
Values:
"prefix": Match from the start of the namespace
"suffix": Match from the end of the namespace
"""
class MatchCondition(NamedTuple):
"""Represents a pattern for matching namespaces in the store.
This class combines a match type (prefix or suffix) with a namespace path
pattern that can include wildcards to flexibly match different namespace
hierarchies.
??? example "Examples"
Prefix matching:
```python
MatchCondition(match_type="prefix", path=("users", "profiles"))
```
Suffix matching with wildcard:
```python
MatchCondition(match_type="suffix", path=("cache", "*"))
```
Simple suffix matching:
```python
MatchCondition(match_type="suffix", path=("v1",))
```
"""
"""Represents a single match condition."""
match_type: NamespaceMatchType
"""Type of namespace matching to perform."""
path: NamespacePath
"""Namespace path pattern that can include wildcards."""
path: NameSpacePath
class ListNamespacesOp(NamedTuple):
"""Operation to list and filter namespaces in the store.
This operation allows exploring the organization of data, finding specific
collections, and navigating the namespace hierarchy.
??? example "Examples"
List all namespaces under the "documents" path:
```python
ListNamespacesOp(
match_conditions=(MatchCondition(match_type="prefix", path=("documents",)),),
max_depth=2
)
```
List all namespaces that end with "v1":
```python
ListNamespacesOp(
match_conditions=(MatchCondition(match_type="suffix", path=("v1",)),),
limit=50
)
```
"""
"""Operation to list namespaces with optional match conditions."""
match_conditions: Optional[tuple[MatchCondition, ...]] = None
"""Optional conditions for filtering namespaces.
??? example "Examples"
All user namespaces:
```python
(MatchCondition(match_type="prefix", path=("users",)),)
```
All namespaces that start with "docs" and end with "draft":
```python
(
MatchCondition(match_type="prefix", path=("docs",)),
MatchCondition(match_type="suffix", path=("draft",))
)
```
"""
"""A tuple of match conditions to apply to namespaces."""
max_depth: Optional[int] = None
"""Maximum depth of namespace hierarchy to return.
Note:
Namespaces deeper than this level will be truncated.
"""
"""Return namespaces up to this depth in the hierarchy."""
limit: int = 100
"""Maximum number of namespaces to return."""
offset: int = 0
"""Number of namespaces to skip for pagination."""
class PutOp(NamedTuple):
"""Operation to store, update, or delete an item in the store.
This class represents a single operation to modify the store's contents,
whether adding new items, updating existing ones, or removing them.
"""
namespace: tuple[str, ...]
"""Hierarchical path that identifies the location of the item.
The namespace acts as a folder-like structure to organize items.
Each element in the tuple represents one level in the hierarchy.
??? example "Examples"
Root level documents
```python
("documents",)
```
User-specific documents
```python
("documents", "user123")
```
Nested cache structure
```python
("cache", "embeddings", "v1")
```
"""
key: str
"""Unique identifier for the item within its namespace.
The key must be unique within the specific namespace to avoid conflicts.
Together with the namespace, it forms a complete path to the item.
Example:
If namespace is ("documents", "user123") and key is "report1",
the full path would effectively be "documents/user123/report1"
"""
value: Optional[dict[str, Any]]
"""The data to store, or None to mark the item for deletion.
The value must be a dictionary with string keys and JSON-serializable values.
Setting this to None signals that the item should be deleted.
Example:
{
"field1": "string value",
"field2": 123,
"nested": {"can": "contain", "any": "serializable data"}
}
"""
index: Optional[Union[Literal[False], list[str]]] = None # type: ignore[assignment]
"""Controls how the item's fields are indexed for search operations.
Indexing configuration determines how the item can be found through search:
- None (default): Uses the store's default indexing configuration (if provided)
- False: Disables indexing for this item
- list[str]: Specifies which json path fields to index for search
The item remains accessible through direct get() operations regardless of indexing.
When indexed, fields can be searched using natural language queries through
vector similarity search (if supported by the store implementation).
Path Syntax:
- Simple field access: "field"
- Nested fields: "parent.child.grandchild"
- Array indexing:
- Specific index: "array[0]"
- Last element: "array[-1]"
- All elements (each individually): "array[*]"
??? example "Examples"
- None - Use store defaults
- False - Don't index this item
- list[str] - List of fields to index
```python
[
"metadata.title", # Nested field access
"chapters[*].content", # Index content from all chapters as separate vectors
"authors[0].name", # First author's name
"revisions[-1].changes", # Most recent revision's changes
"sections[*].paragraphs[*].text", # All text from all paragraphs in all sections
"metadata.tags[*]", # All tags in metadata
]
```
"""
"""Number of namespaces to skip before returning results."""
Op = Union[GetOp, SearchOp, PutOp, ListNamespacesOp]
Result = Union[Item, list[Item], list[SearchItem], list[tuple[str, ...]], None]
Result = Union[Item, list[Item], list[tuple[str, ...]], None]
class InvalidNamespaceError(ValueError):
"""Provided namespace is invalid."""
class IndexConfig(TypedDict, total=False):
"""Configuration for indexing documents for semantic search in the store."""
dims: int
"""Number of dimensions in the embedding vectors.
Common embedding models have the following dimensions:
- OpenAI text-embedding-3-large: 256, 1024, or 3072
- OpenAI text-embedding-3-small: 512 or 1536
- OpenAI text-embedding-ada-002: 1536
- Cohere embed-english-v3.0: 1024
- Cohere embed-english-light-v3.0: 384
- Cohere embed-multilingual-v3.0: 1024
- Cohere embed-multilingual-light-v3.0: 384
"""
embed: Union[Embeddings, EmbeddingsFunc, AEmbeddingsFunc]
"""Optional function to generate embeddings from text."""
fields: Optional[list[str]]
"""Fields to extract text from for embedding generation.
Defaults to the root ["$"], which embeds the json object as a whole.
"""
def _validate_namespace(namespace: tuple[str, ...]) -> None:
if not namespace:
raise InvalidNamespaceError("Namespace cannot be empty.")
for label in namespace:
if not isinstance(label, str):
raise InvalidNamespaceError(
f"Invalid namespace label '{label}' found in {namespace}. Namespace labels"
f" must be strings, but got {type(label).__name__}."
)
if "." in label:
raise InvalidNamespaceError(
f"Invalid namespace label '{label}' found in {namespace}. Namespace labels cannot contain periods ('.')."
)
elif not label:
raise InvalidNamespaceError(
f"Namespace labels cannot be empty strings. Got {label} in {namespace}"
)
if namespace[0] == "langgraph":
raise InvalidNamespaceError(
f'Root label for namespace cannot be "langgraph". Got: {namespace}'
)
class BaseStore(ABC):
@@ -549,16 +231,14 @@ class BaseStore(ABC):
namespace_prefix: tuple[str, ...],
/,
*,
query: Optional[str] = None,
filter: Optional[dict[str, Any]] = None,
limit: int = 10,
offset: int = 0,
) -> list[SearchItem]:
) -> list[Item]:
"""Search for items within a namespace prefix.
Args:
namespace_prefix: Hierarchical path prefix to search within.
query: Optional query for natural language search.
filter: Key-value pairs to filter results.
limit: Maximum number of items to return.
offset: Number of items to skip before returning results.
@@ -566,54 +246,18 @@ class BaseStore(ABC):
Returns:
List of items matching the search criteria.
"""
return self.batch([SearchOp(namespace_prefix, filter, limit, offset, query)])[0]
return self.batch([SearchOp(namespace_prefix, filter, limit, offset)])[0]
def put(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
index: Optional[Union[Literal[False], list[str]]] = None,
) -> None:
"""Store or update an item in the store.
def put(self, namespace: tuple[str, ...], key: str, value: dict[str, Any]) -> None:
"""Store or update an item.
Args:
namespace: Hierarchical path for the item, represented as a tuple of strings.
Example: ("documents", "user123")
key: Unique identifier within the namespace. Together with namespace forms
the complete path to the item.
value: Dictionary containing the item's data. Must contain string keys
and JSON-serializable values.
index: Controls how the item's fields are indexed for search:
- None (default): Use store's default indexing configuration
- False: Disable indexing for this item
- list[str]: List of field paths to index, supporting:
- Nested fields: "metadata.title"
- Array access: "chapters[*].content" (each indexed separately)
- Specific indices: "authors[0].name"
Note:
Indexing capabilities depend on your store implementation.
Some implementations may support only a subset of indexing features.
??? example "Examples"
Simple storage without special indexing (respects store defaults)
```python
store.put(("docs",), "report", {"title": "Annual Report"})
```
Index specific fields for search
```python
store.put(("docs",), "report", {"title": "Annual Report"}, index=["title"])
```
Do not index for semantic search
```python
store.put(("docs",), "report", {"title": "Annual Report"}, index=False)
```
namespace: Hierarchical path for the item.
key: Unique identifier within the namespace.
value: Dictionary containing the item's data.
"""
_validate_namespace(namespace)
self.batch([PutOp(namespace, key, value, index=index)])
self.batch([PutOp(namespace, key, value)])
def delete(self, namespace: tuple[str, ...], key: str) -> None:
"""Delete an item.
@@ -627,8 +271,8 @@ class BaseStore(ABC):
def list_namespaces(
self,
*,
prefix: Optional[NamespacePath] = None,
suffix: Optional[NamespacePath] = None,
prefix: Optional[NameSpacePath] = None,
suffix: Optional[NameSpacePath] = None,
max_depth: Optional[int] = None,
limit: int = 100,
offset: int = 0,
@@ -642,7 +286,7 @@ class BaseStore(ABC):
prefix (Optional[Tuple[str, ...]]): Filter namespaces that start with this path.
suffix (Optional[Tuple[str, ...]]): Filter namespaces that end with this path.
max_depth (Optional[int]): Return namespaces up to this depth in the hierarchy.
Namespaces deeper than this level will be truncated.
Namespaces deeper than this level will be truncated to this depth.
limit (int): Maximum number of namespaces to return (default 100).
offset (int): Number of namespaces to skip for pagination (default 0).
@@ -650,18 +294,16 @@ class BaseStore(ABC):
List[Tuple[str, ...]]: A list of namespace tuples that match the criteria.
Each tuple represents a full namespace path up to `max_depth`.
??? example "Examples":
Examples:
Setting max_depth=3. Given the namespaces:
```python
# Example if you have the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
store.list_namespaces(prefix=("a", "b"), max_depth=3)
# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
```
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
store.list_namespaces(prefix=("a", "b"), max_depth=3)
# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
"""
match_conditions = []
if prefix:
@@ -694,16 +336,14 @@ class BaseStore(ABC):
namespace_prefix: tuple[str, ...],
/,
*,
query: Optional[str] = None,
filter: Optional[dict[str, Any]] = None,
limit: int = 10,
offset: int = 0,
) -> list[SearchItem]:
) -> list[Item]:
"""Asynchronously search for items within a namespace prefix.
Args:
namespace_prefix: Hierarchical path prefix to search within.
query: Optional query for natural language search.
filter: Key-value pairs to filter results.
limit: Maximum number of items to return.
offset: Number of items to skip before returning results.
@@ -711,61 +351,22 @@ class BaseStore(ABC):
Returns:
List of items matching the search criteria.
"""
return (
await self.abatch(
[SearchOp(namespace_prefix, filter, limit, offset, query)]
)
)[0]
return (await self.abatch([SearchOp(namespace_prefix, filter, limit, offset)]))[
0
]
async def aput(
self,
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
index: Optional[Union[Literal[False], list[str]]] = None,
self, namespace: tuple[str, ...], key: str, value: dict[str, Any]
) -> None:
"""Asynchronously store or update an item in the store.
"""Asynchronously store or update an item.
Args:
namespace: Hierarchical path for the item, represented as a tuple of strings.
Example: ("documents", "user123")
key: Unique identifier within the namespace. Together with namespace forms
the complete path to the item.
value: Dictionary containing the item's data. Must contain string keys
and JSON-serializable values.
index: Controls how the item's fields are indexed for search:
- None (default): Use store's default indexing configuration
- False: Disable indexing for this item
- list[str]: List of field paths to index, supporting:
- Nested fields: "metadata.title"
- Array access: "chapters[*].content" (each indexed separately)
- Specific indices: "authors[0].name"
Note:
Indexing capabilities depend on your store implementation.
Some implementations may support only a subset of indexing features.
??? example "Examples"
Simple storage without special indexing:
```python
await store.aput(("docs",), "report", {"title": "Annual Report"})
```
Index specific fields for search:
```python
await store.aput(
("docs",),
"report",
{
"title": "Q4 Report",
"chapters": [{"content": "..."}, {"content": "..."}]
},
index=["title", "chapters[*].content"]
)
```
namespace: Hierarchical path for the item.
key: Unique identifier within the namespace.
value: Dictionary containing the item's data.
"""
_validate_namespace(namespace)
await self.abatch([PutOp(namespace, key, value, index=index)])
await self.abatch([PutOp(namespace, key, value)])
async def adelete(self, namespace: tuple[str, ...], key: str) -> None:
"""Asynchronously delete an item.
@@ -779,8 +380,8 @@ class BaseStore(ABC):
async def alist_namespaces(
self,
*,
prefix: Optional[NamespacePath] = None,
suffix: Optional[NamespacePath] = None,
prefix: Optional[NameSpacePath] = None,
suffix: Optional[NameSpacePath] = None,
max_depth: Optional[int] = None,
limit: int = 100,
offset: int = 0,
@@ -802,19 +403,16 @@ class BaseStore(ABC):
List[Tuple[str, ...]]: A list of namespace tuples that match the criteria.
Each tuple represents a full namespace path up to `max_depth`.
??? example "Examples"
Setting max_depth=3 with existing namespaces:
```python
# Given the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
Examples:
await store.alist_namespaces(prefix=("a", "b"), max_depth=3)
# Returns: [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
```
Setting max_depth=3. Given the namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
await store.alist_namespaces(prefix=("a", "b"), max_depth=3)
# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
"""
match_conditions = []
if prefix:
@@ -829,44 +427,3 @@ class BaseStore(ABC):
offset=offset,
)
return (await self.abatch([op]))[0]
def _validate_namespace(namespace: tuple[str, ...]) -> None:
if not namespace:
raise InvalidNamespaceError("Namespace cannot be empty.")
for label in namespace:
if not isinstance(label, str):
raise InvalidNamespaceError(
f"Invalid namespace label '{label}' found in {namespace}. Namespace labels"
f" must be strings, but got {type(label).__name__}."
)
if "." in label:
raise InvalidNamespaceError(
f"Invalid namespace label '{label}' found in {namespace}. Namespace labels cannot contain periods ('.')."
)
elif not label:
raise InvalidNamespaceError(
f"Namespace labels cannot be empty strings. Got {label} in {namespace}"
)
if namespace[0] == "langgraph":
raise InvalidNamespaceError(
f'Root label for namespace cannot be "langgraph". Got: {namespace}'
)
__all__ = [
"BaseStore",
"Item",
"Op",
"PutOp",
"GetOp",
"SearchOp",
"ListNamespacesOp",
"MatchCondition",
"NamespacePath",
"NamespaceMatchType",
"Embeddings",
"ensure_embeddings",
"tokenize_path",
"get_text_at_path",
]
+5 -84
View File
@@ -1,17 +1,13 @@
import asyncio
import weakref
from typing import Any, Literal, Optional, Union
from typing import Any, Optional
from langgraph.store.base import (
BaseStore,
GetOp,
Item,
ListNamespacesOp,
MatchCondition,
NamespacePath,
Op,
PutOp,
SearchItem,
SearchOp,
_validate_namespace,
)
@@ -44,13 +40,12 @@ class AsyncBatchedBaseStore(BaseStore):
namespace_prefix: tuple[str, ...],
/,
*,
query: Optional[str] = None,
filter: Optional[dict[str, Any]] = None,
limit: int = 10,
offset: int = 0,
) -> list[SearchItem]:
) -> list[Item]:
fut = self._loop.create_future()
self._aqueue[fut] = SearchOp(namespace_prefix, filter, limit, offset, query)
self._aqueue[fut] = SearchOp(namespace_prefix, filter, limit, offset)
return await fut
async def aput(
@@ -58,11 +53,10 @@ class AsyncBatchedBaseStore(BaseStore):
namespace: tuple[str, ...],
key: str,
value: dict[str, Any],
index: Optional[Union[Literal[False], list[str]]] = None,
) -> None:
_validate_namespace(namespace)
fut = self._loop.create_future()
self._aqueue[fut] = PutOp(namespace, key, value, index)
self._aqueue[fut] = PutOp(namespace, key, value)
return await fut
async def adelete(
@@ -74,74 +68,6 @@ class AsyncBatchedBaseStore(BaseStore):
self._aqueue[fut] = PutOp(namespace, key, None)
return await fut
async def alist_namespaces(
self,
*,
prefix: Optional[NamespacePath] = None,
suffix: Optional[NamespacePath] = None,
max_depth: Optional[int] = None,
limit: int = 100,
offset: int = 0,
) -> list[tuple[str, ...]]:
fut = self._loop.create_future()
match_conditions = []
if prefix:
match_conditions.append(MatchCondition(match_type="prefix", path=prefix))
if suffix:
match_conditions.append(MatchCondition(match_type="suffix", path=suffix))
op = ListNamespacesOp(
match_conditions=tuple(match_conditions),
max_depth=max_depth,
limit=limit,
offset=offset,
)
self._aqueue[fut] = op
return await fut
def _dedupe_ops(values: list[Op]) -> tuple[Optional[list[int]], list[Op]]:
"""Dedupe operations while preserving order for results.
Args:
values: List of operations to dedupe
Returns:
Tuple of (listen indices, deduped operations)
where listen indices map deduped operation results back to original positions
"""
if len(values) <= 1:
return None, list(values)
dedupped: list[Op] = []
listen: list[int] = []
puts: dict[tuple[tuple[str, ...], str], int] = {}
for op in values:
if isinstance(op, (GetOp, SearchOp, ListNamespacesOp)):
try:
listen.append(dedupped.index(op))
except ValueError:
listen.append(len(dedupped))
dedupped.append(op)
elif isinstance(op, PutOp):
putkey = (op.namespace, op.key)
if putkey in puts:
# Overwrite previous put
ix = puts[putkey]
dedupped[ix] = op
listen.append(ix)
else:
puts[putkey] = len(dedupped)
listen.append(len(dedupped))
dedupped.append(op)
else: # Any new ops will be treated regularly
listen.append(len(dedupped))
dedupped.append(op)
return listen, dedupped
async def _run(
aqueue: dict[asyncio.Future, Op], store: weakref.ReferenceType[BaseStore]
@@ -155,12 +81,7 @@ async def _run(
taken = aqueue.copy()
# action each operation
try:
values = list(taken.values())
listen, dedupped = _dedupe_ops(values)
results = await s.abatch(dedupped)
if listen is not None:
results = [results[ix] for ix in listen]
results = await s.abatch(taken.values())
# set the results of each operation
for fut, result in zip(taken, results):
fut.set_result(result)
@@ -1,380 +0,0 @@
"""Utilities for working with embedding functions and LangChain's Embeddings interface.
This module provides tools to wrap arbitrary embedding functions (both sync and async)
into LangChain's Embeddings interface. This enables using custom embedding functions
with LangChain-compatible tools while maintaining support for both synchronous and
asynchronous operations.
"""
import asyncio
import json
from typing import Any, Awaitable, Callable, Optional, Sequence, Union
from langchain_core.embeddings import Embeddings
EmbeddingsFunc = Callable[[Sequence[str]], list[list[float]]]
"""Type for synchronous embedding functions.
The function should take a sequence of strings and return a list of embeddings,
where each embedding is a list of floats. The dimensionality of the embeddings
should be consistent for all inputs.
"""
AEmbeddingsFunc = Callable[[Sequence[str]], Awaitable[list[list[float]]]]
"""Type for asynchronous embedding functions.
Similar to EmbeddingsFunc, but returns an awaitable that resolves to the embeddings.
"""
def ensure_embeddings(
embed: Union[Embeddings, EmbeddingsFunc, AEmbeddingsFunc, None],
) -> Embeddings:
"""Ensure that an embedding function conforms to LangChain's Embeddings interface.
This function wraps arbitrary embedding functions to make them compatible with
LangChain's Embeddings interface. It handles both synchronous and asynchronous
functions.
Args:
embed: Either an existing Embeddings instance, or a function that converts
text to embeddings. If the function is async, it will be used for both
sync and async operations.
Returns:
An Embeddings instance that wraps the provided function(s).
??? example "Examples"
Wrap a synchronous embedding function:
```python
def my_embed_fn(texts):
return [[0.1, 0.2] for _ in texts]
embeddings = ensure_embeddings(my_embed_fn)
result = embeddings.embed_query("hello") # Returns [0.1, 0.2]
```
Wrap an asynchronous embedding function:
```python
async def my_async_fn(texts):
return [[0.1, 0.2] for _ in texts]
embeddings = ensure_embeddings(my_async_fn)
result = await embeddings.aembed_query("hello") # Returns [0.1, 0.2]
```
"""
if embed is None:
raise ValueError("embed must be provided")
if isinstance(embed, Embeddings):
return embed
return EmbeddingsLambda(embed)
class EmbeddingsLambda(Embeddings):
"""Wrapper to convert embedding functions into LangChain's Embeddings interface.
This class allows arbitrary embedding functions to be used with LangChain-compatible
tools. It supports both synchronous and asynchronous operations, and can handle:
1. A synchronous function for sync operations (async operations will use sync function)
2. An async function for both sync/async operations (sync operations will raise an error)
The embedding functions should convert text into fixed-dimensional vectors that
capture the semantic meaning of the text.
Args:
func: Function that converts text to embeddings. Can be sync or async.
If async, it will be used for async operations, but sync operations
will raise an error. If sync, it will be used for both sync and async operations.
??? example "Examples"
With a sync function:
```python
def my_embed_fn(texts):
# Return 2D embeddings for each text
return [[0.1, 0.2] for _ in texts]
embeddings = EmbeddingsLambda(my_embed_fn)
result = embeddings.embed_query("hello") # Returns [0.1, 0.2]
await embeddings.aembed_query("hello") # Also returns [0.1, 0.2]
```
With an async function:
```python
async def my_async_fn(texts):
return [[0.1, 0.2] for _ in texts]
embeddings = EmbeddingsLambda(my_async_fn)
await embeddings.aembed_query("hello") # Returns [0.1, 0.2]
# Note: embed_query() would raise an error
```
"""
def __init__(
self,
func: Union[EmbeddingsFunc, AEmbeddingsFunc],
) -> None:
if func is None:
raise ValueError("func must be provided")
if _is_async_callable(func):
self.afunc = func
else:
self.func = func
def embed_documents(self, texts: list[str]) -> list[list[float]]:
"""Embed a list of texts into vectors.
Args:
texts: list of texts to convert to embeddings.
Returns:
list of embeddings, one per input text. Each embedding is a list of floats.
Raises:
ValueError: If the instance was initialized with only an async function.
"""
func = getattr(self, "func", None)
if func is None:
raise ValueError(
"EmbeddingsLambda was initialized with an async function but no sync function. "
"Use aembed_documents for async operation or provide a sync function."
)
return func(texts)
def embed_query(self, text: str) -> list[float]:
"""Embed a single piece of text.
Args:
text: Text to convert to an embedding.
Returns:
Embedding vector as a list of floats.
Note:
This is equivalent to calling embed_documents with a single text
and taking the first result.
"""
return self.embed_documents([text])[0]
async def aembed_documents(self, texts: list[str]) -> list[list[float]]:
"""Asynchronously embed a list of texts into vectors.
Args:
texts: list of texts to convert to embeddings.
Returns:
list of embeddings, one per input text. Each embedding is a list of floats.
Note:
If no async function was provided, this falls back to the sync implementation.
"""
afunc = getattr(self, "afunc", None)
if afunc is None:
return await super().aembed_documents(texts)
return await afunc(texts)
async def aembed_query(self, text: str) -> list[float]:
"""Asynchronously embed a single piece of text.
Args:
text: Text to convert to an embedding.
Returns:
Embedding vector as a list of floats.
Note:
This is equivalent to calling aembed_documents with a single text
and taking the first result.
"""
afunc = getattr(self, "afunc", None)
if afunc is None:
return await super().aembed_query(text)
return (await afunc([text]))[0]
def get_text_at_path(obj: Any, path: Union[str, list[str]]) -> list[str]:
"""Extract text from an object using a path expression or pre-tokenized path.
Args:
obj: The object to extract text from
path: Either a path string or pre-tokenized path list.
!!! info "Path types handled"
- Simple paths: "field1.field2"
- Array indexing: "[0]", "[*]", "[-1]"
- Wildcards: "*"
- Multi-field selection: "{field1,field2}"
- Nested paths in multi-field: "{field1,nested.field2}"
"""
if not path or path == "$":
return [json.dumps(obj, sort_keys=True)]
tokens = tokenize_path(path) if isinstance(path, str) else path
def _extract_from_obj(obj: Any, tokens: list[str], pos: int) -> list[str]:
if pos >= len(tokens):
if isinstance(obj, (str, int, float, bool)):
return [str(obj)]
elif obj is None:
return []
elif isinstance(obj, (list, dict)):
return [json.dumps(obj, sort_keys=True)]
return []
token = tokens[pos]
results = []
if token.startswith("[") and token.endswith("]"):
if not isinstance(obj, list):
return []
index = token[1:-1]
if index == "*":
for item in obj:
results.extend(_extract_from_obj(item, tokens, pos + 1))
else:
try:
idx = int(index)
if idx < 0:
idx = len(obj) + idx
if 0 <= idx < len(obj):
results.extend(_extract_from_obj(obj[idx], tokens, pos + 1))
except (ValueError, IndexError):
return []
elif token.startswith("{") and token.endswith("}"):
if not isinstance(obj, dict):
return []
fields = [f.strip() for f in token[1:-1].split(",")]
for field in fields:
nested_tokens = tokenize_path(field)
if nested_tokens:
current_obj: Optional[dict] = obj
for nested_token in nested_tokens:
if (
isinstance(current_obj, dict)
and nested_token in current_obj
):
current_obj = current_obj[nested_token]
else:
current_obj = None
break
if current_obj is not None:
if isinstance(current_obj, (str, int, float, bool)):
results.append(str(current_obj))
elif isinstance(current_obj, (list, dict)):
results.append(json.dumps(current_obj, sort_keys=True))
# Handle wildcard
elif token == "*":
if isinstance(obj, dict):
for value in obj.values():
results.extend(_extract_from_obj(value, tokens, pos + 1))
elif isinstance(obj, list):
for item in obj:
results.extend(_extract_from_obj(item, tokens, pos + 1))
# Handle regular field
else:
if isinstance(obj, dict) and token in obj:
results.extend(_extract_from_obj(obj[token], tokens, pos + 1))
return results
return _extract_from_obj(obj, tokens, 0)
# Private utility functions
def tokenize_path(path: str) -> list[str]:
"""Tokenize a path into components.
!!! info "Types handled"
- Simple paths: "field1.field2"
- Array indexing: "[0]", "[*]", "[-1]"
- Wildcards: "*"
- Multi-field selection: "{field1,field2}"
"""
if not path:
return []
tokens = []
current: list[str] = []
i = 0
while i < len(path):
char = path[i]
if char == "[": # Handle array index
if current:
tokens.append("".join(current))
current = []
bracket_count = 1
index_chars = ["["]
i += 1
while i < len(path) and bracket_count > 0:
if path[i] == "[":
bracket_count += 1
elif path[i] == "]":
bracket_count -= 1
index_chars.append(path[i])
i += 1
tokens.append("".join(index_chars))
continue
elif char == "{": # Handle multi-field selection
if current:
tokens.append("".join(current))
current = []
brace_count = 1
field_chars = ["{"]
i += 1
while i < len(path) and brace_count > 0:
if path[i] == "{":
brace_count += 1
elif path[i] == "}":
brace_count -= 1
field_chars.append(path[i])
i += 1
tokens.append("".join(field_chars))
continue
elif char == ".": # Handle regular field
if current:
tokens.append("".join(current))
current = []
else:
current.append(char)
i += 1
if current:
tokens.append("".join(current))
return tokens
def _is_async_callable(
func: Any,
) -> bool:
"""Check if a function is async.
This includes both async def functions and classes with async __call__ methods.
Args:
func: Function or callable object to check.
Returns:
True if the function is async, False otherwise.
"""
return (
asyncio.iscoroutinefunction(func)
or hasattr(func, "__call__") # noqa: B004
and asyncio.iscoroutinefunction(func.__call__)
)
__all__ = [
"ensure_embeddings",
"EmbeddingsFunc",
"AEmbeddingsFunc",
]
@@ -1,379 +1,79 @@
"""In-memory key-value store.
A lightweight store implementation using Python dictionaries. Supports basic
key-value operations and vector search when configured with embeddings.
Examples:
Basic key-value storage:
store = InMemoryStore()
store.put(("users", "123"), "prefs", {"theme": "dark"})
item = store.get(("users", "123"), "prefs")
Vector search with embeddings:
from langchain_openai import OpenAIEmbeddings
store = InMemoryStore(index={
"dims": 1536,
"embed": OpenAIEmbeddings(model="text-embedding-3-small"),
})
# Store documents
store.put(("docs",), "doc1", {"text": "Python tutorial"})
store.put(("docs",), "doc2", {"text": "TypeScript guide"})
# Search by similarity
results = store.search(("docs",), query="python programming")
Note:
For production use cases requiring persistence, use a database-backed store instead.
"""
import asyncio
import concurrent.futures as cf
import functools
import logging
from collections import defaultdict
from datetime import datetime, timezone
from importlib import util
from typing import Any, Iterable, Optional
from langchain_core.embeddings import Embeddings
from typing import Iterable
from langgraph.store.base import (
BaseStore,
GetOp,
IndexConfig,
Item,
ListNamespacesOp,
MatchCondition,
Op,
PutOp,
Result,
SearchItem,
SearchOp,
ensure_embeddings,
get_text_at_path,
tokenize_path,
)
logger = logging.getLogger(__name__)
class InMemoryStore(BaseStore):
"""In-memory dictionary-backed store with optional vector search.
"""A KV store backed by an in-memory python dictionary.
Examples:
Basic key-value storage:
store = InMemoryStore()
store.put(("users", "123"), "prefs", {"theme": "dark"})
item = store.get(("users", "123"), "prefs")
Vector search with embeddings:
from langchain_openai import OpenAIEmbeddings
store = InMemoryStore(index={
"dims": 1536,
"embed": OpenAIEmbeddings(model="text-embedding-3-small"),
})
# Store documents
store.put(("docs",), "doc1", {"text": "Python tutorial"})
store.put(("docs",), "doc2", {"text": "TypeScript guide"})
# Search by similarity
results = store.search(("docs",), query="python programming")
Warning:
This store keeps all data in memory. Data is lost when the process exits.
For persistence, use a database-backed store like PostgresStore.
Tip:
For vector search, install numpy for better performance:
```bash
pip install numpy
```
Useful for testing/experimentation and lightweight PoC's.
For actual persistence, use a Store backed by a proper database.
"""
__slots__ = (
"_data",
"_vectors",
"index_config",
"embeddings",
)
__slots__ = ("_data",)
def __init__(self, *, index: Optional[IndexConfig] = None) -> None:
# Both _data and _vectors are wrapped in the In-memory API
# Do not change their names
def __init__(self) -> None:
self._data: dict[tuple[str, ...], dict[str, Item]] = defaultdict(dict)
# [ns][key][path]
self._vectors: dict[tuple[str, ...], dict[str, dict[str, list[float]]]] = (
defaultdict(lambda: defaultdict(dict))
)
self.index_config = index
if self.index_config:
self.index_config = self.index_config.copy()
self.embeddings: Optional[Embeddings] = ensure_embeddings(
self.index_config.get("embed"),
)
self.index_config["__tokenized_fields"] = [
(p, tokenize_path(p)) if p != "$" else (p, p)
for p in (self.index_config.get("fields") or ["$"])
]
else:
self.index_config = None
self.embeddings = None
def batch(self, ops: Iterable[Op]) -> list[Result]:
# The batch/abatch methods are treated as internal.
# Users should access via put/search/get/list_namespaces/etc.
results, put_ops, search_ops = self._prepare_ops(ops)
if search_ops:
queryinmem_store = self._embed_search_queries(search_ops)
self._batch_search(search_ops, queryinmem_store, results)
to_embed = self._extract_texts(put_ops)
if to_embed and self.index_config and self.embeddings:
embeddings = self.embeddings.embed_documents(list(to_embed))
self._insertinmem_store(to_embed, embeddings)
self._apply_put_ops(put_ops)
return results
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
# The batch/abatch methods are treated as internal.
# Users should access via put/search/get/list_namespaces/etc.
results, put_ops, search_ops = self._prepare_ops(ops)
if search_ops:
queryinmem_store = await self._aembed_search_queries(search_ops)
self._batch_search(search_ops, queryinmem_store, results)
to_embed = self._extract_texts(put_ops)
if to_embed and self.index_config and self.embeddings:
embeddings = await self.embeddings.aembed_documents(list(to_embed))
self._insertinmem_store(to_embed, embeddings)
self._apply_put_ops(put_ops)
return results
# Helpers
def _filter_items(self, op: SearchOp) -> list[tuple[Item, list[list[float]]]]:
"""Filter items by namespace and filter function, return items with their embeddings."""
namespace_prefix = op.namespace_prefix
def filter_func(item: Item) -> bool:
if not op.filter:
return True
return all(
_compare_values(item.value.get(key), filter_value)
for key, filter_value in op.filter.items()
)
filtered = []
for namespace in self._data:
if not (
namespace[: len(namespace_prefix)] == namespace_prefix
if len(namespace) >= len(namespace_prefix)
else False
):
continue
for key, item in self._data[namespace].items():
if filter_func(item):
if op.query and (embeddings := self._vectors[namespace].get(key)):
filtered.append((item, list(embeddings.values())))
else:
filtered.append((item, []))
return filtered
def _embed_search_queries(
self,
search_ops: dict[int, tuple[SearchOp, list[tuple[Item, list[list[float]]]]]],
) -> dict[str, list[float]]:
queryinmem_store = {}
if self.index_config and self.embeddings and search_ops:
queries = {op.query for (op, _) in search_ops.values() if op.query}
if queries:
with cf.ThreadPoolExecutor() as executor:
futures = {
q: executor.submit(self.embeddings.embed_query, q)
for q in list(queries)
}
for query, future in futures.items():
queryinmem_store[query] = future.result()
return queryinmem_store
async def _aembed_search_queries(
self,
search_ops: dict[int, tuple[SearchOp, list[tuple[Item, list[list[float]]]]]],
) -> dict[str, list[float]]:
queryinmem_store = {}
if self.index_config and self.embeddings and search_ops:
queries = {op.query for (op, _) in search_ops.values() if op.query}
if queries:
coros = [self.embeddings.aembed_query(q) for q in list(queries)]
results = await asyncio.gather(*coros)
queryinmem_store = dict(zip(queries, results))
return queryinmem_store
def _batch_search(
self,
ops: dict[int, tuple[SearchOp, list[tuple[Item, list[list[float]]]]]],
queryinmem_store: dict[str, list[float]],
results: list[Result],
) -> None:
"""Perform batch similarity search for multiple queries."""
for i, (op, candidates) in ops.items():
if not candidates:
results[i] = []
continue
if op.query and queryinmem_store:
query_embedding = queryinmem_store[op.query]
flat_items, flat_vectors = [], []
scoreless = []
for item, vectors in candidates:
for vector in vectors:
flat_items.append(item)
flat_vectors.append(vector)
if not vectors:
scoreless.append(item)
scores = _cosine_similarity(query_embedding, flat_vectors)
sorted_results = sorted(
zip(scores, flat_items), key=lambda x: x[0], reverse=True
)
# max pooling
seen: set[tuple[tuple[str, ...], str]] = set()
kept: list[tuple[Optional[float], Item]] = []
for score, item in sorted_results:
key = (item.namespace, item.key)
if key in seen:
continue
ix = len(seen)
seen.add(key)
if ix >= op.offset + op.limit:
break
if ix < op.offset:
continue
kept.append((score, item))
if scoreless and len(kept) < op.limit:
# Corner case: if we request more items than what we have embedded,
# fill the rest with non-scored items
kept.extend(
(None, item) for item in scoreless[: op.limit - len(kept)]
)
results[i] = [
SearchItem(
namespace=item.namespace,
key=item.key,
value=item.value,
created_at=item.created_at,
updated_at=item.updated_at,
score=float(score) if score is not None else None,
)
for score, item in kept
]
else:
results[i] = [
SearchItem(
namespace=item.namespace,
key=item.key,
value=item.value,
created_at=item.created_at,
updated_at=item.updated_at,
)
for (item, _) in candidates[op.offset : op.offset + op.limit]
]
def _prepare_ops(
self, ops: Iterable[Op]
) -> tuple[
list[Result],
dict[tuple[tuple[str, ...], str], PutOp],
dict[int, tuple[SearchOp, list[tuple[Item, list[list[float]]]]]],
]:
results: list[Result] = []
put_ops: dict[tuple[tuple[str, ...], str], PutOp] = {}
search_ops: dict[
int, tuple[SearchOp, list[tuple[Item, list[list[float]]]]]
] = {}
for i, op in enumerate(ops):
for op in ops:
if isinstance(op, GetOp):
item = self._data[op.namespace].get(op.key)
results.append(item)
elif isinstance(op, SearchOp):
search_ops[i] = (op, self._filter_items(op))
candidates = [
item
for namespace, items in self._data.items()
if (
namespace[: len(op.namespace_prefix)] == op.namespace_prefix
if len(namespace) >= len(op.namespace_prefix)
else False
)
for item in items.values()
]
if op.filter:
candidates = [
item
for item in candidates
if item.value.items() >= op.filter.items()
]
results.append(candidates[op.offset : op.offset + op.limit])
elif isinstance(op, PutOp):
if op.value is None:
self._data[op.namespace].pop(op.key, None)
elif op.key in self._data[op.namespace]:
self._data[op.namespace][op.key].value = op.value
self._data[op.namespace][op.key].updated_at = datetime.now(
timezone.utc
)
else:
self._data[op.namespace][op.key] = Item(
value=op.value,
key=op.key,
namespace=op.namespace,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
results.append(None)
elif isinstance(op, ListNamespacesOp):
results.append(self._handle_list_namespaces(op))
elif isinstance(op, PutOp):
put_ops[(op.namespace, op.key)] = op
results.append(None)
else:
raise ValueError(f"Unknown operation type: {type(op)}")
return results
return results, put_ops, search_ops
def _apply_put_ops(self, put_ops: dict[tuple[tuple[str, ...], str], PutOp]) -> None:
for (namespace, key), op in put_ops.items():
if op.value is None:
self._data[namespace].pop(key, None)
self._vectors[namespace].pop(key, None)
else:
self._data[namespace][key] = Item(
value=op.value,
key=key,
namespace=namespace,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
def _extract_texts(
self, put_ops: dict[tuple[tuple[str, ...], str], PutOp]
) -> dict[str, list[tuple[tuple[str, ...], str, str]]]:
if put_ops and self.index_config and self.embeddings:
to_embed = defaultdict(list)
for op in put_ops.values():
if op.value is not None and op.index is not False:
if op.index is None:
paths = self.index_config["__tokenized_fields"]
else:
paths = [(ix, tokenize_path(ix)) for ix in op.index]
for path, field in paths:
texts = get_text_at_path(op.value, field)
if texts:
if len(texts) > 1:
for i, text in enumerate(texts):
to_embed[text].append(
(op.namespace, op.key, f"{path}.{i}")
)
else:
to_embed[texts[0]].append((op.namespace, op.key, path))
return to_embed
return {}
def _insertinmem_store(
self,
to_embed: dict[str, list[tuple[tuple[str, ...], str, str]]],
embeddings: list[list[float]],
) -> None:
indices = [index for indices in to_embed.values() for index in indices]
if len(indices) != len(embeddings):
raise ValueError(
f"Number of embeddings ({len(embeddings)}) does not"
f" match number of indices ({len(indices)})"
)
for embedding, (ns, key, path) in zip(embeddings, indices):
self._vectors[ns][key][path] = embedding
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
return self.batch(ops)
def _handle_list_namespaces(self, op: ListNamespacesOp) -> list[tuple[str, ...]]:
all_namespaces = list(
@@ -394,52 +94,7 @@ class InMemoryStore(BaseStore):
return namespaces[op.offset : op.offset + op.limit]
@functools.lru_cache(maxsize=1)
def _check_numpy() -> bool:
if bool(util.find_spec("numpy")):
return True
logger.warning(
"NumPy not found in the current Python environment. "
"The InMemoryStore will use a pure Python implementation for vector operations, "
"which may significantly impact performance, especially for large datasets or frequent searches. "
"For optimal speed and efficiency, consider installing NumPy: "
"pip install numpy"
)
return False
def _cosine_similarity(X: list[float], Y: list[list[float]]) -> list[float]:
"""
Compute cosine similarity between a vector X and a matrix Y.
Lazy import numpy for efficiency.
"""
if _check_numpy():
import numpy as np # type: ignore
X_arr = np.array(X) if not isinstance(X, np.ndarray) else X
Y_arr = np.array(Y) if not isinstance(Y, np.ndarray) else Y
X_norm = np.linalg.norm(X_arr)
Y_norm = np.linalg.norm(Y_arr, axis=1)
# Avoid division by zero
mask = Y_norm != 0
similarities = np.zeros_like(Y_norm)
similarities[mask] = np.dot(Y_arr[mask], X_arr) / (Y_norm[mask] * X_norm)
return similarities.tolist()
similarities = []
for y in Y:
dot_product = sum(a * b for a, b in zip(X, y))
norm1 = sum(a * a for a in X) ** 0.5
norm2 = sum(a * a for a in y) ** 0.5
similarity = dot_product / (norm1 * norm2) if norm1 > 0 and norm2 > 0 else 0.0
similarities.append(similarity)
return similarities
def _does_match(match_condition: MatchCondition, key: tuple[str, ...]) -> bool:
"""Whether a namespace key matches a match condition."""
match_type = match_condition.match_type
path = match_condition.path
@@ -462,44 +117,3 @@ def _does_match(match_condition: MatchCondition, key: tuple[str, ...]) -> bool:
return True
else:
raise ValueError(f"Unsupported match type: {match_type}")
def _compare_values(item_value: Any, filter_value: Any) -> bool:
"""Compare values in a JSONB-like way, handling nested objects."""
if isinstance(filter_value, dict):
if any(k.startswith("$") for k in filter_value):
return all(
_apply_operator(item_value, op_key, op_value)
for op_key, op_value in filter_value.items()
)
if not isinstance(item_value, dict):
return False
return all(
_compare_values(item_value.get(k), v) for k, v in filter_value.items()
)
elif isinstance(filter_value, (list, tuple)):
return (
isinstance(item_value, (list, tuple))
and len(item_value) == len(filter_value)
and all(_compare_values(iv, fv) for iv, fv in zip(item_value, filter_value))
)
else:
return item_value == filter_value
def _apply_operator(value: Any, operator: str, op_value: Any) -> bool:
"""Apply a comparison operator, matching PostgreSQL's JSONB behavior."""
if operator == "$eq":
return value == op_value
elif operator == "$gt":
return float(value) > float(op_value)
elif operator == "$gte":
return float(value) >= float(op_value)
elif operator == "$lt":
return float(value) < float(op_value)
elif operator == "$lte":
return float(value) <= float(op_value)
elif operator == "$ne":
return value != op_value
else:
raise ValueError(f"Unsupported operator: {operator}")
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-checkpoint"
version = "2.0.7"
version = "2.0.5"
description = "Library with base interfaces for LangGraph checkpoint savers."
authors = []
license = "MIT"
-55
View File
@@ -1,55 +0,0 @@
"""Embedding utilities for testing."""
import math
import random
from collections import Counter, defaultdict
from typing import Any
from langchain_core.embeddings import Embeddings
class CharacterEmbeddings(Embeddings):
"""Simple character-frequency based embeddings using random projections."""
def __init__(self, dims: int = 50, seed: int = 42):
"""Initialize with embedding dimensions and random seed."""
self._rng = random.Random(seed)
self.dims = dims
# Create projection vector for each character lazily
self._char_projections: defaultdict[str, list[float]] = defaultdict(
lambda: [
self._rng.gauss(0, 1 / math.sqrt(self.dims)) for _ in range(self.dims)
]
)
def _embed_one(self, text: str) -> list[float]:
"""Embed a single text."""
counts = Counter(text)
total = sum(counts.values())
if total == 0:
return [0.0] * self.dims
embedding = [0.0] * self.dims
for char, count in counts.items():
weight = count / total
char_proj = self._char_projections[char]
for i, proj in enumerate(char_proj):
embedding[i] += weight * proj
norm = math.sqrt(sum(x * x for x in embedding))
if norm > 0:
embedding = [x / norm for x in embedding]
return embedding
def embed_documents(self, texts: list[str]) -> list[list[float]]:
"""Embed a list of documents."""
return [self._embed_one(text) for text in texts]
def embed_query(self, text: str) -> list[float]:
"""Embed a query string."""
return self._embed_one(text)
def __eq__(self, other: Any) -> bool:
return isinstance(other, CharacterEmbeddings) and self.dims == other.dims
+15 -608
View File
@@ -1,104 +1,13 @@
# mypy: disable-error-code="operator"
import asyncio
import json
from datetime import datetime
from typing import Any, Iterable
from typing import Iterable
import pytest
from pytest_mock import MockerFixture
from langgraph.store.base import (
GetOp,
InvalidNamespaceError,
Item,
Op,
PutOp,
Result,
get_text_at_path,
)
from langgraph.store.base import GetOp, InvalidNamespaceError, Item, Op, PutOp, Result
from langgraph.store.base.batch import AsyncBatchedBaseStore
from langgraph.store.memory import InMemoryStore
from tests.embed_test_utils import CharacterEmbeddings
class MockAsyncBatchedStore(AsyncBatchedBaseStore):
def __init__(self, **kwargs: Any) -> None:
super().__init__()
self._store = InMemoryStore(**kwargs)
def batch(self, ops: Iterable[Op]) -> list[Result]:
return self._store.batch(ops)
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
return self._store.batch(ops)
def test_get_text_at_path() -> None:
nested_data = {
"name": "test",
"info": {
"age": 25,
"tags": ["a", "b", "c"],
"metadata": {"created": "2024-01-01", "updated": "2024-01-02"},
},
"items": [
{"id": 1, "value": "first", "tags": ["x", "y"]},
{"id": 2, "value": "second", "tags": ["y", "z"]},
{"id": 3, "value": "third", "tags": ["z", "w"]},
],
"empty": None,
"zeros": [0, 0.0, "0"],
"empty_list": [],
"empty_dict": {},
}
assert get_text_at_path(nested_data, "$") == [
json.dumps(nested_data, sort_keys=True)
]
assert get_text_at_path(nested_data, "name") == ["test"]
assert get_text_at_path(nested_data, "info.age") == ["25"]
assert get_text_at_path(nested_data, "info.metadata.created") == ["2024-01-01"]
assert get_text_at_path(nested_data, "items[0].value") == ["first"]
assert get_text_at_path(nested_data, "items[-1].value") == ["third"]
assert get_text_at_path(nested_data, "items[1].tags[0]") == ["y"]
values = get_text_at_path(nested_data, "items[*].value")
assert set(values) == {"first", "second", "third"}
metadata_dates = get_text_at_path(nested_data, "info.metadata.*")
assert set(metadata_dates) == {"2024-01-01", "2024-01-02"}
name_and_age = get_text_at_path(nested_data, "{name,info.age}")
assert set(name_and_age) == {"test", "25"}
item_fields = get_text_at_path(nested_data, "items[*].{id,value}")
assert set(item_fields) == {"1", "2", "3", "first", "second", "third"}
all_tags = get_text_at_path(nested_data, "items[*].tags[*]")
assert set(all_tags) == {"x", "y", "z", "w"}
assert get_text_at_path(None, "any.path") == []
assert get_text_at_path({}, "any.path") == []
assert get_text_at_path(nested_data, "") == [
json.dumps(nested_data, sort_keys=True)
]
assert get_text_at_path(nested_data, "nonexistent") == []
assert get_text_at_path(nested_data, "items[99].value") == []
assert get_text_at_path(nested_data, "items[*].nonexistent") == []
assert get_text_at_path(nested_data, "empty") == []
assert get_text_at_path(nested_data, "empty_list") == ["[]"]
assert get_text_at_path(nested_data, "empty_dict") == ["{}"]
zeros = get_text_at_path(nested_data, "zeros[*]")
assert set(zeros) == {"0", "0.0"}
assert get_text_at_path(nested_data, "items[].value") == []
assert get_text_at_path(nested_data, "items[abc].value") == []
assert get_text_at_path(nested_data, "{unclosed") == []
assert get_text_at_path(nested_data, "nested[{invalid}]") == []
async def test_async_batch_store(mocker: MockerFixture) -> None:
@@ -383,14 +292,12 @@ async def test_cannot_put_empty_namespace() -> None:
await store.aput(("foo", "langgraph", "foo"), "bar", doc)
assert (await store.aget(("foo", "langgraph", "foo"), "bar")).value == doc # type: ignore[union-attr]
assert (await store.asearch(("foo", "langgraph", "foo"), query="bar"))[
0
].value == doc
assert (await store.asearch(("foo", "langgraph", "foo")))[0].value == doc
await store.adelete(("foo", "langgraph", "foo"), "bar")
assert (await store.aget(("foo", "langgraph", "foo"), "bar")) is None
store.put(("foo", "langgraph", "foo"), "bar", doc)
assert store.get(("foo", "langgraph", "foo"), "bar").value == doc # type: ignore[union-attr]
assert store.search(("foo", "langgraph", "foo"), query="bar")[0].value == doc
assert store.search(("foo", "langgraph", "foo"))[0].value == doc
store.delete(("foo", "langgraph", "foo"), "bar")
assert store.get(("foo", "langgraph", "foo"), "bar") is None
@@ -406,6 +313,17 @@ async def test_cannot_put_empty_namespace() -> None:
store.delete(("langgraph", "foo"), "bar")
assert store.get(("langgraph", "foo"), "bar") is None
class MockAsyncBatchedStore(AsyncBatchedBaseStore):
def __init__(self) -> None:
super().__init__()
self._store = InMemoryStore()
def batch(self, ops: Iterable[Op]) -> list[Result]:
return self._store.batch(ops)
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
return self._store.batch(ops)
async_store = MockAsyncBatchedStore()
doc = {"foo": "bar"}
@@ -426,9 +344,6 @@ async def test_cannot_put_empty_namespace() -> None:
assert val is not None
assert val.value == doc
assert (await async_store.asearch(("foo", "langgraph", "foo")))[0].value == doc
assert (await async_store.asearch(("foo", "langgraph", "foo"), query="bar"))[
0
].value == doc
await async_store.adelete(("foo", "langgraph", "foo"), "bar")
assert (await async_store.aget(("foo", "langgraph", "foo"), "bar")) is None
@@ -439,511 +354,3 @@ async def test_cannot_put_empty_namespace() -> None:
assert (await async_store.asearch(("valid", "namespace")))[0].value == doc
await async_store.adelete(("valid", "namespace"), "key")
assert (await async_store.aget(("valid", "namespace"), "key")) is None
async def test_async_batch_store_deduplication(mocker: MockerFixture) -> None:
abatch = mocker.spy(InMemoryStore, "batch")
store = MockAsyncBatchedStore()
same_doc = {"value": "same"}
diff_doc = {"value": "different"}
await asyncio.gather(
store.aput(namespace=("test",), key="same", value=same_doc),
store.aput(namespace=("test",), key="different", value=diff_doc),
)
abatch.reset_mock()
results = await asyncio.gather(
store.aget(namespace=("test",), key="same"),
store.aget(namespace=("test",), key="same"),
store.aget(namespace=("test",), key="different"),
)
assert len(results) == 3
assert results[0] == results[1]
assert results[0] != results[2]
assert results[0].value == same_doc # type: ignore
assert results[2].value == diff_doc # type: ignore
assert len(abatch.call_args_list) == 1
ops = list(abatch.call_args_list[0].args[1])
assert len(ops) == 2
assert GetOp(("test",), "same") in ops
assert GetOp(("test",), "different") in ops
abatch.reset_mock()
doc1 = {"value": 1}
doc2 = {"value": 2}
results = await asyncio.gather(
store.aput(namespace=("test",), key="key", value=doc1),
store.aput(namespace=("test",), key="key", value=doc2),
)
assert len(abatch.call_args_list) == 1
ops = list(abatch.call_args_list[0].args[1])
assert len(ops) == 1
assert ops[0] == PutOp(("test",), "key", doc2)
assert len(results) == 2
assert all(result is None for result in results)
result = await store.aget(namespace=("test",), key="key")
assert result is not None
assert result.value == doc2
abatch.reset_mock()
results = await asyncio.gather(
store.asearch(("test",), filter={"value": 2}),
store.asearch(("test",), filter={"value": 2}),
)
assert len(abatch.call_args_list) == 1
ops = list(abatch.call_args_list[0].args[1])
assert len(ops) == 1
assert len(results) == 2
assert results[0] == results[1]
assert len(results[0]) == 1
assert results[0][0].value == doc2
abatch.reset_mock()
@pytest.fixture
def fake_embeddings() -> CharacterEmbeddings:
return CharacterEmbeddings(dims=500)
def test_vector_store_initialization(fake_embeddings: CharacterEmbeddings) -> None:
"""Test store initialization with embedding config."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
assert store.index_config is not None
assert store.index_config["dims"] == fake_embeddings.dims
assert store.index_config["embed"] == fake_embeddings
def test_vector_insert_with_auto_embedding(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test inserting items that get auto-embedded."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
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:
store.put(("test",), key, value)
results = store.search(("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_async_vector_insert_with_auto_embedding(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test inserting items that get auto-embedded using async methods."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
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
def test_vector_update_with_embedding(fake_embeddings: CharacterEmbeddings) -> None:
"""Test that updating items properly updates their embeddings."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
store.put(("test",), "doc1", {"text": "zany zebra Xerxes"})
store.put(("test",), "doc2", {"text": "something about dogs"})
store.put(("test",), "doc3", {"text": "text about birds"})
results_initial = store.search(("test",), query="Zany Xerxes")
assert len(results_initial) > 0
assert results_initial[0].key == "doc1"
initial_score = results_initial[0].score
assert initial_score is not None
store.put(("test",), "doc1", {"text": "new text about dogs"})
results_after = store.search(("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
assert after_score < initial_score
results_new = store.search(("test",), query="new text about dogs")
for r in results_new:
if r.key == "doc1":
assert r.score > after_score
# Don't index this one
store.put(("test",), "doc4", {"text": "new text about dogs"}, index=False)
results_new = store.search(("test",), query="new text about dogs", limit=3)
assert not any(r.key == "doc4" for r in results_new)
async def test_async_vector_update_with_embedding(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test that updating items properly updates their embeddings using async methods."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
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].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
assert 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
assert 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)
def test_vector_search_with_filters(fake_embeddings: CharacterEmbeddings) -> None:
"""Test combining vector search with filters."""
inmem_store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
# Insert test documents
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:
inmem_store.put(("test",), key, value)
results = inmem_store.search(("test",), query="apple", filter={"color": "red"})
assert len(results) == 2
assert results[0].key == "doc1"
results = inmem_store.search(("test",), query="car", filter={"color": "red"})
assert len(results) == 2
assert results[0].key == "doc2"
results = inmem_store.search(
("test",), query="bbbbluuu", filter={"score": {"$gt": 3.2}}
)
assert len(results) == 3
assert results[0].key == "doc4"
# Multiple filters
results = inmem_store.search(
("test",), query="apple", filter={"score": {"$gte": 4.0}, "color": "green"}
)
assert len(results) == 1
assert results[0].key == "doc3"
async def test_async_vector_search_with_filters(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test combining vector search with filters using async methods."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
# Insert test documents
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)
results = await store.asearch(("test",), query="apple", filter={"color": "red"})
assert len(results) == 2
assert results[0].key == "doc1"
results = await store.asearch(("test",), query="car", filter={"color": "red"})
assert len(results) == 2
assert results[0].key == "doc2"
results = await store.asearch(
("test",), query="bbbbluuu", filter={"score": {"$gt": 3.2}}
)
assert len(results) == 3
assert results[0].key == "doc4"
# Multiple filters
results = await store.asearch(
("test",), query="apple", filter={"score": {"$gte": 4.0}, "color": "green"}
)
assert len(results) == 1
assert results[0].key == "doc3"
async def test_async_batched_vector_search_concurrent(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test concurrent vector search operations using async batched store."""
store = MockAsyncBatchedStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
colors = ["red", "blue", "green", "yellow", "purple"]
items = ["apple", "car", "house", "book", "phone"]
scores = [3.0, 3.5, 4.0, 4.5, 5.0]
docs = []
for i in range(50):
color = colors[i % len(colors)]
item = items[i % len(items)]
score = scores[i % len(scores)]
docs.append(
(
f"doc{i}",
{"text": f"{color} {item}", "color": color, "score": score, "index": i},
)
)
coros = [
*[store.aput(("test",), key, value) for key, value in docs],
*[store.adelete(("test",), key) for key, value in docs],
*[store.aput(("test",), key, value) for key, value in docs],
]
await asyncio.gather(*coros)
# Prepare multiple search queries with different filters
search_queries: list[tuple[str, dict[str, Any]]] = [
("apple", {"color": "red"}),
("car", {"color": "blue"}),
("house", {"color": "green"}),
("phone", {"score": {"$gt": 4.99}}),
("book", {"score": {"$lte": 3.5}}),
("apple", {"score": {"$gte": 3.0}, "color": "red"}),
("car", {"score": {"$lt": 5.1}, "color": "blue"}),
("house", {"index": {"$gt": 25}}),
("phone", {"index": {"$lte": 10}}),
]
all_results = await asyncio.gather(
*[
store.asearch(("test",), query=query, filter=filter_)
for query, filter_ in search_queries
]
)
for results, (query, filter_) in zip(all_results, search_queries):
assert len(results) > 0, f"No results for query '{query}' with filter {filter_}"
for result in results:
if "color" in filter_:
assert result.value["color"] == filter_["color"]
if "score" in filter_:
score = result.value["score"]
for op, value in filter_["score"].items():
if op == "$gt":
assert score > value
elif op == "$gte":
assert score >= value
elif op == "$lt":
assert score < value
elif op == "$lte":
assert score <= value
if "index" in filter_:
index = result.value["index"]
for op, value in filter_["index"].items():
if op == "$gt":
assert index > value
elif op == "$gte":
assert index >= value
elif op == "$lt":
assert index < value
elif op == "$lte":
assert index <= value
def test_vector_search_pagination(fake_embeddings: CharacterEmbeddings) -> None:
"""Test pagination with vector search."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
for i in range(5):
store.put(("test",), f"doc{i}", {"text": f"test document number {i}"})
results_page1 = store.search(("test",), query="test", limit=2)
results_page2 = store.search(("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 = store.search(("test",), query="test", limit=10)
assert len(all_results) == 5
async def test_async_vector_search_pagination(
fake_embeddings: CharacterEmbeddings,
) -> None:
"""Test pagination with vector search using async methods."""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
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_embed_with_path(fake_embeddings: CharacterEmbeddings) -> None:
# Test store-level field configuration
store = InMemoryStore(
index={
"dims": fake_embeddings.dims,
"embed": fake_embeddings,
# Key 2 isn't included. Don't index it.
"fields": ["key0", "key1", "key3"],
}
)
# 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
ascore = results[0].score
bscore = results[1].score
assert ascore == bscore
assert ascore is not None and bscore is not None
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 is not None and results[0].score > results[1].score
assert ascore == pytest.approx(results[0].score, abs=1e-5)
# 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 < ascore
assert results[1].score < ascore
# Test operation-level field configuration
store_no_defaults = InMemoryStore(
index={
"dims": fake_embeddings.dims,
"embed": fake_embeddings,
"fields": ["key17"],
}
)
doc3 = {
"key0": "aaa",
"key1": "bbb",
"key2": "ccc",
"key3": "ddd",
}
doc4 = {
"key0": "eee",
"key1": "bbb", # Same as doc3.key1
"key2": "fff",
"key3": "ggg",
}
await store_no_defaults.aput(("test",), "doc3", doc3, index=["key0", "key1"])
await store_no_defaults.aput(("test",), "doc4", doc4, index=["key1", "key3"])
results = await store_no_defaults.asearch(("test",), query="aaa")
assert len(results) == 2
assert results[0].key == "doc3"
assert results[0].score is not None and results[0].score > results[1].score
results = await store_no_defaults.asearch(("test",), query="ggg")
assert len(results) == 2
assert results[0].key == "doc4"
assert results[0].score is not None and results[0].score > results[1].score
results = await store_no_defaults.asearch(("test",), query="bbb")
assert len(results) == 2
assert results[0].key != results[1].key
assert results[0].score == results[1].score
results = await store_no_defaults.asearch(("test",), query="ccc")
assert len(results) == 2
assert all(r.score < ascore for r in results)
doc5 = {
"key0": "hhh",
"key1": "iii",
}
await store_no_defaults.aput(("test",), "doc5", doc5, index=False)
results = await store_no_defaults.asearch(("test",), query="hhh")
assert len(results) == 3
doc5_result = next(r for r in results if r.key == "doc5")
assert doc5_result.score is None
+3 -3
View File
@@ -1299,9 +1299,9 @@ create-jest@^29.7.0:
prompts "^2.0.1"
cross-spawn@^7.0.2, cross-spawn@^7.0.3:
version "7.0.6"
resolved "https://registry.yarnpkg.com/cross-spawn/-/cross-spawn-7.0.6.tgz#8a58fe78f00dcd70c370451759dfbfaf03e8ee9f"
integrity sha512-uV2QOWP2nWzsy2aMp8aRibhi9dlzF5Hgh5SHaB9OiTGEyDTiJJyx0uy51QXdyWbtAHNua4XJzUKca3OzKUd3vA==
version "7.0.3"
resolved "https://registry.yarnpkg.com/cross-spawn/-/cross-spawn-7.0.3.tgz#f73a85b9d5d41d045551c177e2882d4ac85728a6"
integrity sha512-iRDPJKUPVEND7dHPO8rkbOnPpyDygcDFtWjpeWNCgy8WP2rXcxXL8TskReQl6OrB2G7+UJrags1q15Fudc7G6w==
dependencies:
path-key "^3.1.0"
shebang-command "^2.0.0"
+14 -32
View File
@@ -1,4 +1,3 @@
import os
import pathlib
import shutil
import sys
@@ -190,6 +189,7 @@ def up(
click.secho(
"""For local dev, requires env var LANGSMITH_API_KEY with access to LangGraph Cloud closed beta.
For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KEY.""",
fg="red",
)
with Runner() as runner, Progress(message="Pulling...") as set:
capabilities = langgraph_cli.docker.check_capabilities(runner)
@@ -511,6 +511,19 @@ def dockerfile(save_path: str, config: pathlib.Path, add_docker_compose: bool) -
)
@click.argument("path", required=False)
@click.option(
"--template",
type=str,
help=TEMPLATE_HELP_STRING,
)
@cli.command("new", help="🌱 Create a new LangGraph project from a template.")
@log_command
def new(path: Optional[str], template: Optional[str]) -> None:
"""Create a new LangGraph project from a template."""
return create_new(path, template)
@click.option(
"--host",
default="127.0.0.1",
@@ -550,12 +563,6 @@ def dockerfile(save_path: str, config: pathlib.Path, add_docker_compose: bool) -
type=int,
help="Enable remote debugging by listening on specified port. Requires debugpy to be installed",
)
@click.option(
"--wait-for-client",
is_flag=True,
help="Wait for a debugger client to connect to the debug port before starting the server",
default=False,
)
@cli.command(
"dev",
help="🏃‍♀️‍➡️ Run LangGraph API server in development mode with hot reloading and debugging support",
@@ -569,7 +576,6 @@ def dev(
n_jobs_per_worker: Optional[int],
no_browser: bool,
debug_port: Optional[int],
wait_for_client: bool,
):
"""CLI entrypoint for running the LangGraph API server."""
try:
@@ -593,16 +599,8 @@ def dev(
) from None
config_json = langgraph_cli.config.validate_config_file(config)
cwd = os.getcwd()
sys.path.append(cwd)
dependencies = config_json.get("dependencies", [])
for dep in dependencies:
dep_path = pathlib.Path(cwd) / dep
if dep_path.is_dir() and dep_path.exists():
sys.path.append(str(dep_path))
graphs = config_json.get("graphs", {})
run_server(
host,
port,
@@ -611,25 +609,9 @@ def dev(
n_jobs_per_worker=n_jobs_per_worker,
open_browser=not no_browser,
debug_port=debug_port,
env=config_json.get("env"),
store=config_json.get("store"),
wait_for_client=wait_for_client,
)
@click.argument("path", required=False)
@click.option(
"--template",
type=str,
help=TEMPLATE_HELP_STRING,
)
@cli.command("new", help="🌱 Create a new LangGraph project from a template.")
@log_command
def new(path: Optional[str], template: Optional[str]) -> None:
"""Create a new LangGraph project from a template."""
return create_new(path, template)
def prepare_args_and_stdin(
*,
capabilities: DockerCapabilities,
+87 -131
View File
@@ -1,5 +1,4 @@
import json
import os
import pathlib
import textwrap
from typing import NamedTuple, Optional, TypedDict, Union
@@ -10,44 +9,7 @@ MIN_NODE_VERSION = "20"
MIN_PYTHON_VERSION = "3.11"
class IndexConfig(TypedDict, total=False):
"""Configuration for indexing documents for semantic search in the store."""
dims: int
"""Number of dimensions in the embedding vectors.
Common embedding models have the following dimensions:
- OpenAI text-embedding-3-large: 256, 1024, or 3072
- OpenAI text-embedding-3-small: 512 or 1536
- OpenAI text-embedding-ada-002: 1536
- Cohere embed-english-v3.0: 1024
- Cohere embed-english-light-v3.0: 384
- Cohere embed-multilingual-v3.0: 1024
- Cohere embed-multilingual-light-v3.0: 384
"""
embed: str
"""Optional model (string) to generate embeddings from text or path to model or function.
Examples:
- "openai:text-embedding-3-large"
- "cohere:embed-multilingual-v3.0"
- "src/app.py:embeddings
"""
fields: Optional[list[str]]
"""Fields to extract text from for embedding generation.
Defaults to the root ["$"], which embeds the json object as a whole.
"""
class StoreConfig(TypedDict, total=False):
embed: Optional[IndexConfig]
"""Configuration for vector embeddings in store."""
class Config(TypedDict, total=False):
class Config(TypedDict):
python_version: str
node_version: Optional[str]
pip_config_file: Optional[str]
@@ -55,7 +17,6 @@ class Config(TypedDict, total=False):
dependencies: list[str]
graphs: dict[str, str]
env: Union[dict[str, str], str]
store: Optional[StoreConfig]
def _parse_version(version_str: str) -> tuple[int, int]:
@@ -87,7 +48,6 @@ def validate_config(config: Config) -> Config:
"dockerfile_lines": config.get("dockerfile_lines", []),
"graphs": config.get("graphs", {}),
"env": config.get("env", {}),
"store": config.get("store"),
}
if config.get("node_version")
else {
@@ -97,7 +57,6 @@ def validate_config(config: Config) -> Config:
"dependencies": config.get("dependencies", []),
"graphs": config.get("graphs", {}),
"env": config.get("env", {}),
"store": config.get("store"),
}
)
@@ -232,66 +191,76 @@ def _assemble_local_deps(config_path: pathlib.Path, config: Config) -> LocalDeps
resolved = config_path.parent / local_dep
# validate local dependency
if not resolved.exists():
raise FileNotFoundError(f"Could not find local dependency: {resolved}")
elif not resolved.is_dir():
raise NotADirectoryError(
f"Local dependency must be a directory: {resolved}"
)
elif not resolved.is_relative_to(config_path.parent):
raise ValueError(
f"Local dependency '{resolved}' must be a subdirectory of '{config_path.parent}'"
)
# if it's installable, add it to local_pkgs
# otherwise, add it to faux_pkgs, and create a pyproject.toml
files = os.listdir(resolved)
if "pyproject.toml" in files:
real_pkgs[resolved] = local_dep
if local_dep == ".":
working_dir = f"/deps/{resolved.name}"
elif "setup.py" in files:
real_pkgs[resolved] = local_dep
if local_dep == ".":
working_dir = f"/deps/{resolved.name}"
else:
if any(file == "__init__.py" for file in files):
# flat layout
if "-" in resolved.name:
raise ValueError(
f"Package name '{resolved.name}' contains a hyphen. "
"Rename the directory to use it as flat-layout package."
)
check_reserved(resolved.name, local_dep)
container_path = f"/deps/__outer_{resolved.name}/{resolved.name}"
else:
# src layout
container_path = f"/deps/__outer_{resolved.name}/src"
for file in files:
rfile = resolved / file
if (
rfile.is_dir()
and file != "__pycache__"
and not file.startswith(".")
):
try:
for subfile in os.listdir(rfile):
if subfile.endswith(".py"):
check_reserved(file, local_dep)
break
except PermissionError:
pass
faux_pkgs[resolved] = (local_dep, container_path)
if local_dep == ".":
working_dir = container_path
if "requirements.txt" in files:
rfile = resolved / "requirements.txt"
pip_reqs.append(
(
rfile.relative_to(config_path.parent),
f"{container_path}/requirements.txt",
)
try:
if not resolved.exists():
raise FileNotFoundError(f"Could not find local dependency: {resolved}")
elif not resolved.is_dir():
raise NotADirectoryError(
f"Local dependency must be a directory: {resolved}"
)
elif not resolved.is_relative_to(config_path.parent):
raise ValueError(
f"Local dependency '{resolved}' must be a subdirectory of '{config_path.parent}'"
)
# if it's installable, add it to local_pkgs
# otherwise, add it to faux_pkgs, and create a pyproject.toml
try:
files = list(resolved.iterdir())
file_names = [f.name for f in files]
except (PermissionError, OSError) as e:
raise click.UsageError(
f"Cannot access directory {resolved}: {str(e)}"
) from None
if "pyproject.toml" in file_names:
real_pkgs[resolved] = local_dep
if local_dep == ".":
working_dir = f"/deps/{resolved.name}"
elif "setup.py" in file_names:
real_pkgs[resolved] = local_dep
if local_dep == ".":
working_dir = f"/deps/{resolved.name}"
else:
if any(file == "__init__.py" for file in file_names):
# flat layout
if "-" in resolved.name:
raise ValueError(
f"Package name '{resolved.name}' contains a hyphen. "
"Rename the directory to use it as flat-layout package."
)
check_reserved(resolved.name, local_dep)
container_path = f"/deps/__outer_{resolved.name}/{resolved.name}"
else:
# src layout
container_path = f"/deps/__outer_{resolved.name}/src"
for file in files:
if (
file.is_dir()
and file.name != "__pycache__"
and not file.name.startswith(".")
):
try:
subfiles = list(file.iterdir())
if any(f.name.endswith(".py") for f in subfiles):
check_reserved(file.name, local_dep)
except (PermissionError, OSError):
continue
faux_pkgs[resolved] = (local_dep, container_path)
if local_dep == ".":
working_dir = container_path
if "requirements.txt" in file_names:
rfile = resolved / "requirements.txt"
pip_reqs.append(
(
pathlib.PurePosixPath(
rfile.relative_to(config_path.parent)
),
f"{container_path}/requirements.txt",
)
)
except (PermissionError, OSError) as e:
raise click.UsageError(f"Cannot access path {resolved}: {str(e)}") from e
return LocalDeps(pip_reqs, real_pkgs, faux_pkgs, working_dir)
@@ -315,12 +284,12 @@ def _update_graph_paths(
else:
for path in local_deps.real_pkgs:
if resolved.is_relative_to(path):
module_str = f"/deps/{path.name}/{resolved.relative_to(path)}"
module_str = f"/deps/{path.name}/{pathlib.PurePosixPath(resolved.relative_to(path))}"
break
else:
for faux_pkg, (_, destpath) in local_deps.faux_pkgs.items():
if resolved.is_relative_to(faux_pkg):
module_str = f"{destpath}/{resolved.relative_to(faux_pkg)}"
module_str = f"{destpath}/{pathlib.PurePosixPath(resolved.relative_to(faux_pkg))}"
break
else:
raise ValueError(
@@ -353,17 +322,17 @@ def python_config_to_docker(config_path: pathlib.Path, config: Config, base_imag
pip_pkgs_str = f"RUN {pip_install} {' '.join(pypi_deps)}" if pypi_deps else ""
if local_deps.pip_reqs:
pip_reqs_str = os.linesep.join(
pip_reqs_str = "\n".join(
f"ADD {reqpath} {destpath}" for reqpath, destpath in local_deps.pip_reqs
)
pip_reqs_str += f'{os.linesep}RUN {pip_install} {" ".join("-r " + r for _,r in local_deps.pip_reqs)}'
pip_reqs_str += (
f'\nRUN {pip_install} {" ".join("-r " + r for _,r in local_deps.pip_reqs)}'
)
else:
pip_reqs_str = ""
# https://setuptools.pypa.io/en/latest/userguide/datafiles.html#package-data
# https://til.simonwillison.net/python/pyproject
faux_pkgs_str = f"{os.linesep}{os.linesep}".join(
faux_pkgs_str = "\n\n".join(
f"""ADD {relpath} {destpath}
RUN set -ex && \\
for line in '[project]' \\
@@ -375,12 +344,12 @@ RUN set -ex && \\
done"""
for fullpath, (relpath, destpath) in local_deps.faux_pkgs.items()
)
local_pkgs_str = os.linesep.join(
local_pkgs_str = "\n".join(
f"ADD {relpath} /deps/{fullpath.name}"
for fullpath, relpath in local_deps.real_pkgs.items()
)
installs = f"{os.linesep}{os.linesep}".join(
installs = "\n\n".join(
filter(
None,
[
@@ -392,25 +361,19 @@ RUN set -ex && \\
],
)
)
store_config = config.get("store")
env_additional_config = (
""
if not store_config
else f"""
ENV LANGGRAPH_STORE='{json.dumps(store_config)}'
"""
)
_workdir = f"WORKDIR {local_deps.working_dir}" if local_deps.working_dir else ""
dockerfile_lines = "\n".join(config["dockerfile_lines"])
return f"""FROM {base_image}:{config['python_version']}
{os.linesep.join(config["dockerfile_lines"])}
{dockerfile_lines}
{installs}
RUN {pip_install} -e /deps/*
{env_additional_config}
ENV LANGSERVE_GRAPHS='{json.dumps(config["graphs"])}'
{f"WORKDIR {local_deps.working_dir}" if local_deps.working_dir else ""}"""
{_workdir}"""
def node_config_to_docker(config_path: pathlib.Path, config: Config, base_image: str):
@@ -437,22 +400,15 @@ def node_config_to_docker(config_path: pathlib.Path, config: Config, base_image:
install_cmd = "npm ci"
else:
install_cmd = "npm i"
store_config = config.get("store")
env_additional_config = (
""
if not store_config
else f"""
ENV LANGGRAPH_STORE='{json.dumps(store_config)}'
"""
)
dockerfile_lines = "\n".join(config["dockerfile_lines"])
return f"""FROM {base_image}:{config['node_version']}
{os.linesep.join(config["dockerfile_lines"])}
{dockerfile_lines}
ADD . {faux_path}
RUN cd {faux_path} && {install_cmd}
{env_additional_config}
ENV LANGSERVE_GRAPHS='{json.dumps(config["graphs"])}'
WORKDIR {faux_path}
+659 -645
View File
File diff suppressed because it is too large Load Diff
+5 -6
View File
@@ -1,12 +1,12 @@
[tool.poetry]
name = "langgraph-cli"
version = "0.1.61"
version = "0.1.56"
description = "CLI for interacting with LangGraph API"
authors = []
license = "MIT"
readme = "README.md"
repository = "https://www.github.com/langchain-ai/langgraph"
packages = [{ include = "langgraph_cli" }]
packages = [{include = "langgraph_cli"}]
[tool.poetry.scripts]
langgraph = "langgraph_cli.cli:cli"
@@ -14,8 +14,7 @@ langgraph = "langgraph_cli.cli:cli"
[tool.poetry.dependencies]
python = "^3.9.0,<4.0"
click = "^8.1.7"
langgraph-api = { version = ">=0.0.6,<0.1.0", optional = true, python = ">=3.11,<4.0" }
python-dotenv = { version = ">=0.8.0", optional = true }
langgraph-api-inmem = { version = ">=0.0.3,<0.1.0", optional = true }
[tool.poetry.group.dev.dependencies]
ruff = "^0.6.2"
@@ -27,7 +26,7 @@ pytest-watch = "^4.2.0"
mypy = "^1.10.0"
[tool.poetry.extras]
inmem = ["langgraph-api", "python-dotenv"]
inmem = ["langgraph-api-inmem"]
[tool.pytest.ini_options]
# --strict-markers will raise errors on unknown marks.
@@ -57,4 +56,4 @@ lint.select = [
# isort
"I",
]
lint.ignore = ["E501", "B008"]
lint.ignore = [ "E501", "B008" ]
-2
View File
@@ -30,7 +30,6 @@ def test_validate_config():
"pip_config_file": None,
"dockerfile_lines": [],
"env": {},
"store": None,
**expected_config,
}
actual_config = validate_config(expected_config)
@@ -47,7 +46,6 @@ def test_validate_config():
"agent": "./agent.py:graph",
},
"env": env,
"store": None,
}
actual_config = validate_config(expected_config)
assert actual_config == expected_config
+1 -6
View File
@@ -48,13 +48,8 @@ test:
make stop-postgres; \
exit $$EXIT_CODE
WORKERS ?= auto
XDIST_ARGS := $(if $(WORKERS),-n $(WORKERS) --dist worksteal,)
MAXFAIL ?=
MAXFAIL_ARGS := $(if $(MAXFAIL),--maxfail $(MAXFAIL),)
test_watch:
make start-postgres && poetry run ptw . -- --ff -vv -x $(XDIST_ARGS) $(MAXFAIL_ARGS) --snapshot-update --tb short $(TEST); \
make start-postgres && poetry run ptw . -- --ff -vv -x -n auto --dist worksteal --snapshot-update --tb short $(TEST); \
EXIT_CODE=$$?; \
make stop-postgres; \
exit $$EXIT_CODE
+1 -1
View File
@@ -238,7 +238,7 @@ final_state["messages"][-1].content
* [How-to Guides](https://langchain-ai.github.io/langgraph/how-tos/): Accomplish specific things within LangGraph, from streaming, to adding memory & persistence, to common design patterns (branching, subgraphs, etc.), these are the place to go if you want to copy and run a specific code snippet.
* [Conceptual Guides](https://langchain-ai.github.io/langgraph/concepts/high_level/): In-depth explanations of the key concepts and principles behind LangGraph, such as nodes, edges, state and more.
* [API Reference](https://langchain-ai.github.io/langgraph/reference/graphs/): Review important classes and methods, simple examples of how to use the graph and checkpointing APIs, higher-level prebuilt components and more.
* [LangGraph Platform](https://langchain-ai.github.io/langgraph/concepts/#langgraph-platform): LangGraph Platform is a commercial solution for deploying agentic applications in production, built on the open-source LangGraph framework.
* [Cloud (beta)](https://langchain-ai.github.io/langgraph/cloud/): With one click, deploy LangGraph applications to LangGraph Cloud.
## Contributing
+3 -14
View File
@@ -2,7 +2,7 @@ from enum import Enum
from typing import Any, Sequence
from langgraph.checkpoint.base import EmptyChannelError # noqa: F401
from langgraph.types import Command, Interrupt
from langgraph.types import Interrupt
# EmptyChannelError re-exported for backwards compatibility
@@ -58,11 +58,7 @@ class InvalidUpdateError(Exception):
pass
class GraphBubbleUp(Exception):
pass
class GraphInterrupt(GraphBubbleUp):
class GraphInterrupt(Exception):
"""Raised when a subgraph is interrupted, suppressed by the root graph.
Never raised directly, or surfaced to the user."""
@@ -77,20 +73,13 @@ class NodeInterrupt(GraphInterrupt):
super().__init__([Interrupt(value=value)])
class GraphDelegate(GraphBubbleUp):
class GraphDelegate(Exception):
"""Raised when a graph is delegated (for distributed mode)."""
def __init__(self, *args: dict[str, Any]) -> None:
super().__init__(*args)
class ParentCommand(GraphBubbleUp):
args: tuple[Command]
def __init__(self, command: Command) -> None:
super().__init__(command)
class EmptyInputError(Exception):
"""Raised when graph receives an empty input."""
+2 -18
View File
@@ -37,12 +37,7 @@ from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue
from langgraph.channels.named_barrier_value import NamedBarrierValue
from langgraph.constants import EMPTY_SEQ, NS_END, NS_SEP, SELF, TAG_HIDDEN
from langgraph.errors import (
ErrorCode,
InvalidUpdateError,
ParentCommand,
create_error_message,
)
from langgraph.errors import ErrorCode, InvalidUpdateError, create_error_message
from langgraph.graph.graph import END, START, Branch, CompiledGraph, Graph, Send
from langgraph.managed.base import (
ChannelKeyPlaceholder,
@@ -628,27 +623,20 @@ class CompiledStateGraph(CompiledGraph):
def _get_root(input: Any) -> Any:
if isinstance(input, Command):
if input.graph == Command.PARENT:
return SKIP_WRITE
return input.update
else:
return input
# to avoid name collision below
node_key = key
def _get_state_key(input: Union[None, dict, Any], *, key: str) -> Any:
if input is None:
return SKIP_WRITE
elif isinstance(input, dict):
if all(k not in output_keys for k in input):
raise InvalidUpdateError(
f"Expected node {node_key} to update at least one of {output_keys}, got {input}"
f"Expected node {key} to update at least one of {output_keys}, got {input}"
)
return input.get(key, SKIP_WRITE)
elif isinstance(input, Command):
if input.graph == Command.PARENT:
return SKIP_WRITE
return _get_state_key(input.update, key=key)
elif get_type_hints(type(input)):
value = getattr(input, key, SKIP_WRITE)
@@ -831,8 +819,6 @@ def _control_branch(value: Any) -> Sequence[Union[str, Send]]:
return [value]
if not isinstance(value, GraphCommand):
return EMPTY_SEQ
if value.graph == Command.PARENT:
raise ParentCommand(value)
rtn: list[Union[str, Send]] = []
if isinstance(value.goto, str):
rtn.append(value.goto)
@@ -850,8 +836,6 @@ async def _acontrol_branch(value: Any) -> Sequence[Union[str, Send]]:
return [value]
if not isinstance(value, GraphCommand):
return EMPTY_SEQ
if value.graph == Command.PARENT:
raise ParentCommand(value)
rtn: list[Union[str, Send]] = []
if isinstance(value.goto, str):
rtn.append(value.goto)
@@ -212,7 +212,6 @@ def create_react_agent(
Args:
model: The `LangChain` chat model that supports tool calling.
tools: A list of tools, a ToolExecutor, or a ToolNode instance.
If an empty list is provided, the agent will consist of a single LLM node without tool calling.
state_schema: An optional state schema that defines graph state.
Must have `messages` and `is_last_step` keys.
Defaults to `AgentState` that defines those two keys.
@@ -541,11 +540,20 @@ def create_react_agent(
# get the tool functions wrapped in a tool class from the ToolNode
tool_classes = list(tool_node.tools_by_name.values())
tool_calling_enabled = len(tool_classes) > 0
if _should_bind_tools(model, tool_classes) and tool_calling_enabled:
if _should_bind_tools(model, tool_classes):
model = cast(BaseChatModel, model).bind_tools(tool_classes)
# Define the function that determines whether to continue or not
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
messages = 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__"
# Otherwise if there is, we continue
else:
return "tools"
# we're passing store here for validation
preprocessor = _get_model_preprocessing_runnable(
state_modifier, messages_modifier, store
@@ -627,30 +635,6 @@ def create_react_agent(
# We return a list, because this will get added to the existing list
return {"messages": [response]}
if not tool_calling_enabled:
# Define a new graph
workflow = StateGraph(state_schema or AgentState)
workflow.add_node("agent", RunnableCallable(call_model, acall_model))
workflow.set_entry_point("agent")
return workflow.compile(
checkpointer=checkpointer,
store=store,
interrupt_before=interrupt_before,
interrupt_after=interrupt_after,
debug=debug,
)
# Define the function that determines whether to continue or not
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
messages = 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__"
# Otherwise if there is, we continue
else:
return "tools"
# Define a new graph
workflow = StateGraph(state_schema or AgentState)
@@ -37,7 +37,7 @@ from langchain_core.tools import tool as create_tool
from langchain_core.tools.base import get_all_basemodel_annotations
from typing_extensions import Annotated, get_args, get_origin
from langgraph.errors import GraphBubbleUp
from langgraph.errors import GraphInterrupt
from langgraph.store.base import BaseStore
from langgraph.utils.runnable import RunnableCallable
@@ -275,7 +275,7 @@ class ToolNode(RunnableCallable):
# (2) a NodeInterrupt is raised inside a graph node for a graph called as a tool
# (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph called as a tool
# (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture)
except GraphBubbleUp as e:
except GraphInterrupt as e:
raise e
except Exception as e:
if isinstance(self.handle_tool_errors, tuple):
@@ -316,7 +316,7 @@ class ToolNode(RunnableCallable):
# (2) a NodeInterrupt is raised inside a graph node for a graph called as a tool
# (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph called as a tool
# (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture)
except GraphBubbleUp as e:
except GraphInterrupt as e:
raise e
except Exception as e:
if isinstance(self.handle_tool_errors, tuple):
-10
View File
@@ -1,4 +1,3 @@
import sys
from collections import defaultdict, deque
from functools import partial
from hashlib import sha1
@@ -67,7 +66,6 @@ from langgraph.types import All, LoopProtocol, PregelExecutableTask, PregelTask
from langgraph.utils.config import merge_configs, patch_config
GetNextVersion = Callable[[Optional[V], BaseChannel], V]
SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)
class WritesProtocol(Protocol):
@@ -604,7 +602,6 @@ def prepare_single_task(
None,
task_id,
task_path,
writers=proc.flat_writers,
)
else:
@@ -636,12 +633,6 @@ def prepare_single_task(
)
except StopIteration:
return
except Exception as exc:
if SUPPORTS_EXC_NOTES:
exc.add_note(
f"Before task with name '{name}' and path '{task_path[:3]}'"
)
raise
# create task id
checkpoint_ns = f"{parent_ns}{NS_SEP}{name}" if parent_ns else name
@@ -729,7 +720,6 @@ def prepare_single_task(
None,
task_id,
task_path,
writers=proc.flat_writers,
)
else:
return PregelTask(task_id, name, task_path)
+3 -3
View File
@@ -20,7 +20,7 @@ from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.config import get_executor_for_config
from typing_extensions import ParamSpec
from langgraph.errors import GraphBubbleUp
from langgraph.errors import GraphInterrupt
P = ParamSpec("P")
T = TypeVar("T")
@@ -68,7 +68,7 @@ class BackgroundExecutor(ContextManager):
def done(self, task: concurrent.futures.Future) -> None:
try:
task.result()
except GraphBubbleUp:
except GraphInterrupt:
# This exception is an interruption signal, not an error
# so we don't want to re-raise it on exit
self.tasks.pop(task)
@@ -155,7 +155,7 @@ class AsyncBackgroundExecutor(AsyncContextManager):
if exc := task.exception():
# This exception is an interruption signal, not an error
# so we don't want to re-raise it on exit
if isinstance(exc, GraphBubbleUp):
if isinstance(exc, GraphInterrupt):
self.tasks.pop(task)
else:
self.tasks.pop(task)
-3
View File
@@ -15,7 +15,6 @@ from langgraph.constants import (
TAG_HIDDEN,
TASKS,
)
from langgraph.errors import InvalidUpdateError
from langgraph.pregel.log import logger
from langgraph.types import Command, PregelExecutableTask, Send
@@ -69,8 +68,6 @@ def map_command(
cmd: Command,
) -> Iterator[tuple[str, str, Any]]:
"""Map input chunk to a sequence of pending writes in the form (channel, value)."""
if cmd.graph == Command.PARENT:
raise InvalidUpdateError("There is not parent graph")
if cmd.send:
if isinstance(cmd.send, (tuple, list)):
sends = cmd.send
+4 -40
View File
@@ -1,9 +1,7 @@
import asyncio
import logging
import random
import sys
import time
from dataclasses import replace
from functools import partial
from typing import Any, Callable, Optional, Sequence
@@ -12,14 +10,12 @@ from langgraph.constants import (
CONFIG_KEY_CHECKPOINT_NS,
CONFIG_KEY_RESUMING,
CONFIG_KEY_SEND,
NS_SEP,
)
from langgraph.errors import _SEEN_CHECKPOINT_NS, GraphBubbleUp, ParentCommand
from langgraph.types import Command, PregelExecutableTask, RetryPolicy
from langgraph.errors import _SEEN_CHECKPOINT_NS, GraphInterrupt
from langgraph.types import PregelExecutableTask, RetryPolicy
from langgraph.utils.config import patch_configurable
logger = logging.getLogger(__name__)
SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)
def run_with_retry(
@@ -44,26 +40,10 @@ def run_with_retry(
task.proc.invoke(task.input, config)
# if successful, end
break
except ParentCommand as exc:
ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
cmd = exc.args[0]
if cmd.graph == ns:
# this command is for the current graph, handle it
for w in task.writers:
w.invoke(cmd, config)
break
elif cmd.graph == Command.PARENT:
# this command is for the parent graph, assign it to the parent
parent_ns = NS_SEP.join(ns.split(NS_SEP)[:-1])
exc.args = (replace(cmd, graph=parent_ns),)
# bubble up
raise
except GraphBubbleUp:
except GraphInterrupt:
# if interrupted, end
raise
except Exception as exc:
if SUPPORTS_EXC_NOTES:
exc.add_note(f"During task with name '{task.name}' and id '{task.id}'")
if retry_policy is None:
raise
# increment attempts
@@ -138,26 +118,10 @@ async def arun_with_retry(
await task.proc.ainvoke(task.input, config)
# if successful, end
break
except ParentCommand as exc:
ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
cmd = exc.args[0]
if cmd.graph == ns:
# this command is for the current graph, handle it
for w in task.writers:
w.invoke(cmd, config)
break
elif cmd.graph == Command.PARENT:
# this command is for the parent graph, assign it to the parent
parent_ns = NS_SEP.join(ns.split(NS_SEP)[:-1])
exc.args = (replace(cmd, graph=parent_ns),)
# bubble up
raise
except GraphBubbleUp:
except GraphInterrupt:
# if interrupted, end
raise
except Exception as exc:
if SUPPORTS_EXC_NOTES:
exc.add_note(f"During task with name '{task.name}' and id '{task.id}'")
if retry_policy is None:
raise
# increment attempts
+3 -3
View File
@@ -23,7 +23,7 @@ from langgraph.constants import (
PUSH,
TAG_HIDDEN,
)
from langgraph.errors import GraphBubbleUp, GraphInterrupt
from langgraph.errors import GraphDelegate, GraphInterrupt
from langgraph.pregel.executor import Submit
from langgraph.pregel.retry import arun_with_retry, run_with_retry
from langgraph.types import PregelExecutableTask, RetryPolicy
@@ -298,7 +298,7 @@ class PregelRunner:
# save interrupt to checkpointer
if interrupts := [(INTERRUPT, i) for i in exception.args[0]]:
self.put_writes(task.id, interrupts)
elif isinstance(exception, GraphBubbleUp):
elif isinstance(exception, GraphDelegate):
raise exception
else:
# save error to checkpointer
@@ -324,7 +324,7 @@ def _should_stop_others(
if fut.cancelled():
return True
if exc := fut.exception():
return not isinstance(exc, GraphBubbleUp)
return not isinstance(exc, GraphInterrupt)
else:
return False
-5
View File
@@ -5,7 +5,6 @@ from typing import (
TYPE_CHECKING,
Any,
Callable,
ClassVar,
Generic,
Hashable,
Literal,
@@ -141,7 +140,6 @@ class PregelExecutableTask(NamedTuple):
id: str
path: tuple[Union[str, int, tuple], ...]
scheduled: bool = False
writers: Sequence[Runnable] = ()
class StateSnapshot(NamedTuple):
@@ -241,7 +239,6 @@ N = TypeVar("N", bound=Hashable)
class Command(Generic[N]):
"""One or more commands to update the graph's state and send messages to nodes."""
graph: Optional[str] = None
update: Optional[dict[str, Any]] = None
send: Union[Send, Sequence[Send]] = ()
resume: Optional[Union[Any, dict[str, Any]]] = None
@@ -255,8 +252,6 @@ class Command(Generic[N]):
)
return f"Command({contents})"
PARENT: ClassVar[Literal["__parent__"]] = "__parent__"
StreamChunk = tuple[tuple[str, ...], str, Any]
+13 -13
View File
@@ -1,4 +1,4 @@
# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.
[[package]]
name = "aiosqlite"
@@ -1253,13 +1253,13 @@ test = ["jupyter-server (>=2.0.0)", "pytest (>=7.0)", "pytest-jupyter[server] (>
[[package]]
name = "jupyterlab"
version = "4.2.5"
version = "4.3.0"
description = "JupyterLab computational environment"
optional = false
python-versions = ">=3.8"
files = [
{file = "jupyterlab-4.2.5-py3-none-any.whl", hash = "sha256:73b6e0775d41a9fee7ee756c80f58a6bed4040869ccc21411dc559818874d321"},
{file = "jupyterlab-4.2.5.tar.gz", hash = "sha256:ae7f3a1b8cb88b4f55009ce79fa7c06f99d70cd63601ee4aa91815d054f46f75"},
{file = "jupyterlab-4.3.0-py3-none-any.whl", hash = "sha256:f67e1095ad61ae04349024f0b40345062ab108a0c6998d9810fec6a3c1a70cd5"},
{file = "jupyterlab-4.3.0.tar.gz", hash = "sha256:7c6835cbf8df0af0ec8a39332e85ff11693fb9a468205343b4fc0bfbc74817e5"},
]
[package.dependencies]
@@ -1280,9 +1280,9 @@ tornado = ">=6.2.0"
traitlets = "*"
[package.extras]
dev = ["build", "bump2version", "coverage", "hatch", "pre-commit", "pytest-cov", "ruff (==0.3.5)"]
docs = ["jsx-lexer", "myst-parser", "pydata-sphinx-theme (>=0.13.0)", "pytest", "pytest-check-links", "pytest-jupyter", "sphinx (>=1.8,<7.3.0)", "sphinx-copybutton"]
docs-screenshots = ["altair (==5.3.0)", "ipython (==8.16.1)", "ipywidgets (==8.1.2)", "jupyterlab-geojson (==3.4.0)", "jupyterlab-language-pack-zh-cn (==4.1.post2)", "matplotlib (==3.8.3)", "nbconvert (>=7.0.0)", "pandas (==2.2.1)", "scipy (==1.12.0)", "vega-datasets (==0.9.0)"]
dev = ["build", "bump2version", "coverage", "hatch", "pre-commit", "pytest-cov", "ruff (==0.6.9)"]
docs = ["jsx-lexer", "myst-parser", "pydata-sphinx-theme (>=0.13.0)", "pytest", "pytest-check-links", "pytest-jupyter", "sphinx (>=1.8,<8.1.0)", "sphinx-copybutton"]
docs-screenshots = ["altair (==5.4.1)", "ipython (==8.16.1)", "ipywidgets (==8.1.5)", "jupyterlab-geojson (==3.4.0)", "jupyterlab-language-pack-zh-cn (==4.2.post3)", "matplotlib (==3.9.2)", "nbconvert (>=7.0.0)", "pandas (==2.2.3)", "scipy (==1.14.1)", "vega-datasets (==0.9.0)"]
test = ["coverage", "pytest (>=7.0)", "pytest-check-links (>=0.7)", "pytest-console-scripts", "pytest-cov", "pytest-jupyter (>=0.5.3)", "pytest-timeout", "pytest-tornasync", "requests", "requests-cache", "virtualenv"]
upgrade-extension = ["copier (>=9,<10)", "jinja2-time (<0.3)", "pydantic (<3.0)", "pyyaml-include (<3.0)", "tomli-w (<2.0)"]
@@ -1792,26 +1792,26 @@ files = [
[[package]]
name = "notebook"
version = "7.2.2"
version = "7.0.7"
description = "Jupyter Notebook - A web-based notebook environment for interactive computing"
optional = false
python-versions = ">=3.8"
files = [
{file = "notebook-7.2.2-py3-none-any.whl", hash = "sha256:c89264081f671bc02eec0ed470a627ed791b9156cad9285226b31611d3e9fe1c"},
{file = "notebook-7.2.2.tar.gz", hash = "sha256:2ef07d4220421623ad3fe88118d687bc0450055570cdd160814a59cf3a1c516e"},
{file = "notebook-7.0.7-py3-none-any.whl", hash = "sha256:289b606d7e173f75a18beb1406ef411b43f97f7a9c55ba03efa3622905a62346"},
{file = "notebook-7.0.7.tar.gz", hash = "sha256:3bcff00c17b3ac142ef5f436d50637d936b274cfa0b41f6ac0175363de9b4e09"},
]
[package.dependencies]
jupyter-server = ">=2.4.0,<3"
jupyterlab = ">=4.2.0,<4.3"
jupyterlab-server = ">=2.27.1,<3"
jupyterlab = ">=4.0.2,<5"
jupyterlab-server = ">=2.22.1,<3"
notebook-shim = ">=0.2,<0.3"
tornado = ">=6.2.0"
[package.extras]
dev = ["hatch", "pre-commit"]
docs = ["myst-parser", "nbsphinx", "pydata-sphinx-theme", "sphinx (>=1.3.6)", "sphinxcontrib-github-alt", "sphinxcontrib-spelling"]
test = ["importlib-resources (>=5.0)", "ipykernel", "jupyter-server[test] (>=2.4.0,<3)", "jupyterlab-server[test] (>=2.27.1,<3)", "nbval", "pytest (>=7.0)", "pytest-console-scripts", "pytest-timeout", "pytest-tornasync", "requests"]
test = ["importlib-resources (>=5.0)", "ipykernel", "jupyter-server[test] (>=2.4.0,<3)", "jupyterlab-server[test] (>=2.22.1,<3)", "nbval", "pytest (>=7.0)", "pytest-console-scripts", "pytest-timeout", "pytest-tornasync", "requests"]
[[package]]
name = "notebook-shim"
+2 -107
View File
@@ -272,54 +272,6 @@ async def _store_postgres_aio():
await conn.execute(f"DROP DATABASE {database}")
@asynccontextmanager
async def _store_postgres_aio_pipe():
if sys.version_info < (3, 10):
pytest.skip("Async Postgres tests require Python 3.10+")
database = f"test_{uuid4().hex[:16]}"
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI, autocommit=True
) as conn:
await conn.execute(f"CREATE DATABASE {database}")
try:
async with AsyncPostgresStore.from_conn_string(
DEFAULT_POSTGRES_URI + database
) as store:
await store.setup() # Run in its own transaction
async with AsyncPostgresStore.from_conn_string(
DEFAULT_POSTGRES_URI + database, pipeline=True
) as store:
yield store
finally:
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI, autocommit=True
) as conn:
await conn.execute(f"DROP DATABASE {database}")
@asynccontextmanager
async def _store_postgres_aio_pool():
if sys.version_info < (3, 10):
pytest.skip("Async Postgres tests require Python 3.10+")
database = f"test_{uuid4().hex[:16]}"
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI, autocommit=True
) as conn:
await conn.execute(f"CREATE DATABASE {database}")
try:
async with AsyncPostgresStore.from_conn_string(
DEFAULT_POSTGRES_URI + database,
pool_config={"max_size": 10},
) as store:
await store.setup()
yield store
finally:
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI, autocommit=True
) as conn:
await conn.execute(f"DROP DATABASE {database}")
@asynccontextmanager
async def _store_duckdb_aio():
async with AsyncDuckDBStore.from_conn_string(":memory:") as store:
@@ -344,45 +296,6 @@ def store_postgres():
conn.execute(f"DROP DATABASE {database}")
@pytest.fixture(scope="function")
def store_postgres_pipe():
database = f"test_{uuid4().hex[:16]}"
# create unique db
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
conn.execute(f"CREATE DATABASE {database}")
try:
# yield store
with PostgresStore.from_conn_string(DEFAULT_POSTGRES_URI + database) as store:
store.setup() # Run in its own transaction
with PostgresStore.from_conn_string(
DEFAULT_POSTGRES_URI + database, pipeline=True
) as store:
yield store
finally:
# drop unique db
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
conn.execute(f"DROP DATABASE {database}")
@pytest.fixture(scope="function")
def store_postgres_pool():
database = f"test_{uuid4().hex[:16]}"
# create unique db
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
conn.execute(f"CREATE DATABASE {database}")
try:
# yield store
with PostgresStore.from_conn_string(
DEFAULT_POSTGRES_URI + database, pool_config={"max_size": 10}
) as store:
store.setup()
yield store
finally:
# drop unique db
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
conn.execute(f"DROP DATABASE {database}")
@pytest.fixture(scope="function")
def store_duckdb():
with DuckDBStore.from_conn_string(":memory:") as store:
@@ -404,12 +317,6 @@ async def awith_store(store_name: Optional[str]) -> AsyncIterator[BaseStore]:
elif store_name == "postgres_aio":
async with _store_postgres_aio() as store:
yield store
elif store_name == "postgres_aio_pipe":
async with _store_postgres_aio_pipe() as store:
yield store
elif store_name == "postgres_aio_pool":
async with _store_postgres_aio_pool() as store:
yield store
elif store_name == "duckdb_aio":
async with _store_duckdb_aio() as store:
yield store
@@ -435,17 +342,5 @@ ALL_CHECKPOINTERS_ASYNC_PLUS_NONE = [
*ALL_CHECKPOINTERS_ASYNC,
None,
]
ALL_STORES_SYNC = [
"in_memory",
"postgres",
"postgres_pipe",
"postgres_pool",
"duckdb",
]
ALL_STORES_ASYNC = [
"in_memory",
"postgres_aio",
"postgres_aio_pipe",
"postgres_aio_pool",
"duckdb_aio",
]
ALL_STORES_SYNC = ["in_memory", "postgres", "duckdb"]
ALL_STORES_ASYNC = ["in_memory", "postgres_aio", "duckdb_aio"]
-3
View File
@@ -102,9 +102,6 @@ class FakeToolCallingModel(BaseChatModel):
tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
**kwargs: Any,
) -> Runnable[LanguageModelInput, BaseMessage]:
if len(tools) == 0:
raise ValueError("Must provide at least one tool")
tool_dicts = []
for tool in tools:
if not isinstance(tool, BaseTool):
+44 -137
View File
@@ -1,6 +1,5 @@
import enum
import json
import logging
import operator
import re
import time
@@ -68,9 +67,16 @@ from langgraph.errors import InvalidUpdateError, MultipleSubgraphsError, NodeInt
from langgraph.graph import END, Graph, GraphCommand, StateGraph
from langgraph.graph.message import MessageGraph, MessagesState, add_messages
from langgraph.managed.shared_value import SharedValue
from langgraph.prebuilt.chat_agent_executor import create_tool_calling_executor
from langgraph.prebuilt.chat_agent_executor import (
create_tool_calling_executor,
)
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.pregel import Channel, GraphRecursionError, Pregel, StateSnapshot
from langgraph.pregel import (
Channel,
GraphRecursionError,
Pregel,
StateSnapshot,
)
from langgraph.pregel.retry import RetryPolicy
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
@@ -98,8 +104,6 @@ from tests.messages import (
_AnyIdToolMessage,
)
logger = logging.getLogger(__name__)
# define these objects to avoid importing langchain_core.agents
# and therefore avoid relying on core Pydantic version
@@ -6624,7 +6628,11 @@ def test_message_graph(
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
)
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import tool
@@ -13929,75 +13937,50 @@ def test_store_injected(
doc_id = str(uuid.uuid4())
doc = {"some-key": "this-is-a-val"}
uid = uuid.uuid4().hex
namespace = (f"foo-{uid}", "bar")
thread_1 = str(uuid.uuid4())
thread_2 = str(uuid.uuid4())
class Node:
def __init__(self, i: Optional[int] = None):
self.i = i
def __call__(self, inputs: State, config: RunnableConfig, store: BaseStore):
assert isinstance(store, BaseStore)
store.put(
namespace
if self.i is not None
and config["configurable"]["thread_id"] in (thread_1, thread_2)
else (f"foo_{self.i}", "bar"),
doc_id,
{
**doc,
"from_thread": config["configurable"]["thread_id"],
"some_val": inputs["count"],
},
)
return {"count": 1}
def node(input: State, config: RunnableConfig, store: BaseStore):
assert isinstance(store, BaseStore)
store.put(
("foo", "bar"),
doc_id,
{
**doc,
"from_thread": config["configurable"]["thread_id"],
"some_val": input["count"],
},
)
return {"count": 1}
builder = StateGraph(State)
builder.add_node("node", Node())
builder.add_node("node", node)
builder.add_edge("__start__", "node")
N = 500
M = 1
if "duckdb" in store_name:
logger.warning(
"DuckDB store implementation has a known issue that does not"
" support concurrent writes, so we're reducing the test scope"
)
N = M = 1
for i in range(N):
builder.add_node(f"node_{i}", Node(i))
builder.add_edge("__start__", f"node_{i}")
graph = builder.compile(store=the_store, checkpointer=checkpointer)
results = graph.batch(
[{"count": 0}] * M,
([{"configurable": {"thread_id": str(uuid.uuid4())}}] * (M - 1))
+ [{"configurable": {"thread_id": thread_1}}],
)
result = results[-1]
assert result == {"count": N + 1}
returned_doc = the_store.get(namespace, doc_id).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": 0}
assert len(the_store.search(namespace)) == 1
# Check results after another turn of the same thread
thread_1 = str(uuid.uuid4())
result = graph.invoke({"count": 0}, {"configurable": {"thread_id": thread_1}})
assert result == {"count": (N + 1) * 2}
returned_doc = the_store.get(namespace, doc_id).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": N + 1}
assert len(the_store.search(namespace)) == 1
assert result == {"count": 1}
returned_doc = the_store.get(("foo", "bar"), doc_id).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": 0}
assert len(the_store.search(("foo", "bar"))) == 1
# Check update on existing thread
result = graph.invoke({"count": 0}, {"configurable": {"thread_id": thread_1}})
assert result == {"count": 2}
returned_doc = the_store.get(("foo", "bar"), doc_id).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": 1}
assert len(the_store.search(("foo", "bar"))) == 1
thread_2 = str(uuid.uuid4())
result = graph.invoke({"count": 0}, {"configurable": {"thread_id": thread_2}})
assert result == {"count": N + 1}
returned_doc = the_store.get(namespace, doc_id).value
assert result == {"count": 1}
returned_doc = the_store.get(("foo", "bar"), doc_id).value
assert returned_doc == {
**doc,
"from_thread": thread_2,
"some_val": 0,
} # Overwrites the whole doc
assert len(the_store.search(namespace)) == 1 # still overwriting the same one
assert len(the_store.search(("foo", "bar"))) == 1 # still overwriting the same one
def test_enum_node_names():
@@ -14395,79 +14378,3 @@ def test_runnable_passthrough_node_graph() -> None:
graph = graph_builder.compile()
assert graph.get_graph(xray=True).to_json() == graph.get_graph(xray=False).to_json()
@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_SYNC)
def test_parent_command(request: pytest.FixtureRequest, checkpointer_name: str) -> None:
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
@tool(return_direct=True)
def get_user_name() -> GraphCommand:
"""Retrieve user name"""
return GraphCommand(update={"user_name": "Meow"}, graph=GraphCommand.PARENT)
subgraph_builder = StateGraph(MessagesState)
subgraph_builder.add_node("tool", get_user_name)
subgraph_builder.add_edge(START, "tool")
subgraph = subgraph_builder.compile()
class CustomParentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# this key is not available to the child graph
user_name: str
builder = StateGraph(CustomParentState)
builder.add_node("alice", subgraph)
builder.add_edge(START, "alice")
checkpointer = request.getfixturevalue(f"checkpointer_{checkpointer_name}")
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
assert graph.invoke({"messages": [("user", "get user name")]}, config) == {
"messages": [
_AnyIdHumanMessage(
content="get user name", additional_kwargs={}, response_metadata={}
),
],
"user_name": "Meow",
}
assert graph.get_state(config) == StateSnapshot(
values={
"messages": [
_AnyIdHumanMessage(
content="get user name", additional_kwargs={}, response_metadata={}
),
],
"user_name": "Meow",
},
next=(),
config={
"configurable": {
"thread_id": "1",
"checkpoint_ns": "",
"checkpoint_id": AnyStr(),
}
},
metadata={
"source": "loop",
"writes": {
"alice": {
"user_name": "Meow",
}
},
"thread_id": "1",
"step": 1,
"parents": {},
},
created_at=AnyStr(),
parent_config={
"configurable": {
"thread_id": "1",
"checkpoint_ns": "",
"checkpoint_id": AnyStr(),
}
},
tasks=(),
)
+32 -144
View File
@@ -1,5 +1,4 @@
import asyncio
import logging
import operator
import random
import re
@@ -101,8 +100,6 @@ from tests.messages import (
_AnyIdToolMessage,
)
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.anyio
@@ -12275,89 +12272,60 @@ async def test_store_injected_async(checkpointer_name: str, store_name: str) ->
doc_id = str(uuid.uuid4())
doc = {"some-key": "this-is-a-val"}
uid = uuid.uuid4().hex
namespace = (f"foo-{uid}", "bar")
thread_1 = str(uuid.uuid4())
thread_2 = str(uuid.uuid4())
class Node:
def __init__(self, i: Optional[int] = None):
self.i = i
async def __call__(
self, inputs: State, config: RunnableConfig, store: BaseStore
):
assert isinstance(store, BaseStore)
await store.aput(
namespace
if self.i is not None
and config["configurable"]["thread_id"] in (thread_1, thread_2)
else (f"foo_{self.i}", "bar"),
doc_id,
{
**doc,
"from_thread": config["configurable"]["thread_id"],
"some_val": inputs["count"],
},
)
return {"count": 1}
async def node(input: State, config: RunnableConfig, store: BaseStore):
assert isinstance(store, BaseStore)
await store.aput(
("foo", "bar"),
doc_id,
{
**doc,
"from_thread": config["configurable"]["thread_id"],
"some_val": input["count"],
},
)
return {"count": 1}
builder = StateGraph(State)
builder.add_node("node", Node())
builder.add_node("node", node)
builder.add_edge("__start__", "node")
N = 500
M = 1
if "duckdb" in store_name:
logger.warning(
"DuckDB store implementation has a known issue that does not"
" support concurrent writes, so we're reducing the test scope"
)
N = M = 1
for i in range(N):
builder.add_node(f"node_{i}", Node(i))
builder.add_edge("__start__", f"node_{i}")
async with awith_checkpointer(checkpointer_name) as checkpointer, awith_store(
store_name
) as the_store:
graph = builder.compile(store=the_store, checkpointer=checkpointer)
# Test batch operations with multiple threads
results = await graph.abatch(
[{"count": 0}] * M,
([{"configurable": {"thread_id": str(uuid.uuid4())}}] * (M - 1))
+ [{"configurable": {"thread_id": thread_1}}],
)
result = results[-1]
assert result == {"count": N + 1}
returned_doc = (await the_store.aget(namespace, doc_id)).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": 0}
assert len((await the_store.asearch(namespace))) == 1
# Check results after another turn of the same thread
thread_1 = str(uuid.uuid4())
result = await graph.ainvoke(
{"count": 0}, {"configurable": {"thread_id": thread_1}}
)
assert result == {"count": (N + 1) * 2}
returned_doc = (await the_store.aget(namespace, doc_id)).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": N + 1}
assert len((await the_store.asearch(namespace))) == 1
assert result == {"count": 1}
returned_doc = (await the_store.aget(("foo", "bar"), doc_id)).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": 0}
assert len((await the_store.asearch(("foo", "bar")))) == 1
# Check update on existing thread
result = await graph.ainvoke(
{"count": 0}, {"configurable": {"thread_id": thread_1}}
)
assert result == {"count": 2}
returned_doc = (await the_store.aget(("foo", "bar"), doc_id)).value
assert returned_doc == {**doc, "from_thread": thread_1, "some_val": 1}
assert len((await the_store.asearch(("foo", "bar")))) == 1
thread_2 = str(uuid.uuid4())
# Test with a different thread
result = await graph.ainvoke(
{"count": 0}, {"configurable": {"thread_id": thread_2}}
)
assert result == {"count": N + 1}
returned_doc = (await the_store.aget(namespace, doc_id)).value
assert result == {"count": 1}
returned_doc = (await the_store.aget(("foo", "bar"), doc_id)).value
assert returned_doc == {
**doc,
"from_thread": thread_2,
"some_val": 0,
} # Overwrites the whole doc
assert (
len((await the_store.asearch(namespace))) == 1
len((await the_store.asearch(("foo", "bar")))) == 1
) # still overwriting the same one
@@ -12597,83 +12565,3 @@ async def test_debug_nested_subgraphs():
assert stream_task["interrupts"] == history_task.interrupts
assert stream_task.get("error") == history_task.error
assert stream_task.get("state") == history_task.state
@pytest.mark.parametrize("checkpointer_name", ALL_CHECKPOINTERS_ASYNC)
async def test_parent_command(checkpointer_name: str) -> None:
from langchain_core.messages import BaseMessage
from langchain_core.tools import tool
@tool(return_direct=True)
def get_user_name() -> GraphCommand:
"""Retrieve user name"""
return GraphCommand(update={"user_name": "Meow"}, graph=GraphCommand.PARENT)
subgraph_builder = StateGraph(MessagesState)
subgraph_builder.add_node("tool", get_user_name)
subgraph_builder.add_edge(START, "tool")
subgraph = subgraph_builder.compile()
class CustomParentState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
# this key is not available to the child graph
user_name: str
builder = StateGraph(CustomParentState)
builder.add_node("alice", subgraph)
builder.add_edge(START, "alice")
async with awith_checkpointer(checkpointer_name) as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
assert await graph.ainvoke(
{"messages": [("user", "get user name")]}, config
) == {
"messages": [
_AnyIdHumanMessage(
content="get user name", additional_kwargs={}, response_metadata={}
),
],
"user_name": "Meow",
}
assert await graph.aget_state(config) == StateSnapshot(
values={
"messages": [
_AnyIdHumanMessage(
content="get user name",
additional_kwargs={},
response_metadata={},
),
],
"user_name": "Meow",
},
next=(),
config={
"configurable": {
"thread_id": "1",
"checkpoint_ns": "",
"checkpoint_id": AnyStr(),
}
},
metadata={
"source": "loop",
"writes": {
"alice": {
"user_name": "Meow",
}
},
"thread_id": "1",
"step": 1,
"parents": {},
},
created_at=AnyStr(),
parent_config={
"configurable": {
"thread_id": "1",
"checkpoint_ns": "",
"checkpoint_id": AnyStr(),
}
},
tasks=(),
)
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@langchain/langgraph-sdk",
"version": "0.0.29",
"version": "0.0.26",
"description": "Client library for interacting with the LangGraph API",
"type": "module",
"packageManager": "yarn@1.22.19",
+1 -2
View File
@@ -33,7 +33,6 @@ import {
OnConflictBehavior,
} from "./types.js";
import { mergeSignals } from "./utils/signals.js";
import { getEnvironmentVariable } from "./utils/env.js";
/**
* Get the API key from the environment.
@@ -54,7 +53,7 @@ export function getApiKey(apiKey?: string): string | undefined {
const prefixes = ["LANGGRAPH", "LANGSMITH", "LANGCHAIN"];
for (const prefix of prefixes) {
const envKey = getEnvironmentVariable(`${prefix}_API_KEY`);
const envKey = process.env[`${prefix}_API_KEY`];
if (envKey) {
// Remove surrounding quotes
return envKey.trim().replace(/^["']|["']$/g, "");
-1
View File
@@ -15,7 +15,6 @@ export type {
ThreadStatus,
Cron,
Checkpoint,
Interrupt,
} from "./schema.js";
export type { OnConflictBehavior, Command } from "./types.js";
+6 -14
View File
@@ -137,16 +137,6 @@ export interface AssistantGraph {
}>;
}
/**
* An interrupt thrown inside a thread.
*/
export interface Interrupt {
value: unknown;
when: "during";
resumable: boolean;
ns?: string[];
}
export interface Thread<ValuesType = DefaultValues> {
/** The ID of the thread. */
thread_id: string;
@@ -165,9 +155,6 @@ export interface Thread<ValuesType = DefaultValues> {
/** The current state of the thread. */
values: ValuesType;
/** Interrupts which were thrown in this thread */
interrupts: Record<string, Array<Interrupt>>;
}
export interface Cron {
@@ -223,7 +210,12 @@ export interface ThreadTask {
name: string;
result?: unknown;
error: Optional<string>;
interrupts: Array<Interrupt>;
interrupts: Array<{
value: unknown;
when: "during";
resumable: boolean;
ns?: string[];
}>;
checkpoint: Optional<Checkpoint>;
state: Optional<ThreadState>;
}
-1
View File
@@ -4,7 +4,6 @@ import PQueueMod from "p-queue";
const STATUS_NO_RETRY = [
400, // Bad Request
401, // Unauthorized
402, // Payment required
403, // Forbidden
404, // Not Found
405, // Method Not Allowed
-11
View File
@@ -1,11 +0,0 @@
export function getEnvironmentVariable(name: string): string | undefined {
// Certain setups (Deno, frontend) will throw an error if you try to access environment variables
try {
return typeof process !== "undefined"
? // eslint-disable-next-line no-process-env
process.env?.[name]
: undefined;
} catch (e) {
return undefined;
}
}
+8 -24
View File
@@ -18,7 +18,6 @@ from typing import (
Dict,
Iterator,
List,
Literal,
Optional,
Sequence,
Union,
@@ -1947,7 +1946,7 @@ class CronClient:
Example Usage:
cron_run = client.crons.create(
cron_run = await client.crons.create(
assistant_id="agent",
schedule="27 15 * * *",
input={"messages": [{"role": "user", "content": "hello!"}]},
@@ -2071,12 +2070,7 @@ class StoreClient:
self.http = http
async def put_item(
self,
namespace: Sequence[str],
/,
key: str,
value: dict[str, Any],
index: Optional[Union[Literal[False], list[str]]] = None,
self, namespace: Sequence[str], /, key: str, value: dict[str, Any]
) -> None:
"""Store or update an item.
@@ -2084,7 +2078,6 @@ class StoreClient:
namespace: A list of strings representing the namespace path.
key: The unique identifier for the item within the namespace.
value: A dictionary containing the item's data.
index: Controls search indexing - None (use defaults), False (disable), or list of field paths to index.
Returns:
None
@@ -2102,7 +2095,11 @@ class StoreClient:
raise ValueError(
f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')."
)
payload = {"namespace": namespace, "key": key, "value": value, "index": index}
payload = {
"namespace": namespace,
"key": key,
"value": value,
}
await self.http.put("/store/items", json=payload)
async def get_item(self, namespace: Sequence[str], /, key: str) -> Item:
@@ -2170,7 +2167,6 @@ class StoreClient:
filter: Optional[dict[str, Any]] = None,
limit: int = 10,
offset: int = 0,
query: Optional[str] = None,
) -> SearchItemsResponse:
"""Search for items within a namespace prefix.
@@ -2179,7 +2175,6 @@ class StoreClient:
filter: Optional dictionary of key-value pairs to filter results.
limit: Maximum number of items to return (default is 10).
offset: Number of items to skip before returning results (default is 0).
query: Optional query for natural language search.
Returns:
List[Item]: A list of items matching the search criteria.
@@ -2217,7 +2212,6 @@ class StoreClient:
"filter": filter,
"limit": limit,
"offset": offset,
"query": query,
}
return await self.http.post("/store/items/search", json=_provided_vals(payload))
@@ -4160,12 +4154,7 @@ class SyncStoreClient:
self.http = http
def put_item(
self,
namespace: Sequence[str],
/,
key: str,
value: dict[str, Any],
index: Optional[Union[Literal[False], list[str]]] = None,
self, namespace: Sequence[str], /, key: str, value: dict[str, Any]
) -> None:
"""Store or update an item.
@@ -4173,7 +4162,6 @@ class SyncStoreClient:
namespace: A list of strings representing the namespace path.
key: The unique identifier for the item within the namespace.
value: A dictionary containing the item's data.
index: Controls search indexing - None (use defaults), False (disable), or list of field paths to index.
Returns:
None
@@ -4195,7 +4183,6 @@ class SyncStoreClient:
"namespace": namespace,
"key": key,
"value": value,
"index": index,
}
self.http.put("/store/items", json=payload)
@@ -4263,7 +4250,6 @@ class SyncStoreClient:
filter: Optional[dict[str, Any]] = None,
limit: int = 10,
offset: int = 0,
query: Optional[str] = None,
) -> SearchItemsResponse:
"""Search for items within a namespace prefix.
@@ -4272,7 +4258,6 @@ class SyncStoreClient:
filter: Optional dictionary of key-value pairs to filter results.
limit: Maximum number of items to return (default is 10).
offset: Number of items to skip before returning results (default is 0).
query: Optional query for natural language search.
Returns:
List[Item]: A list of items matching the search criteria.
@@ -4310,7 +4295,6 @@ class SyncStoreClient:
"filter": filter,
"limit": limit,
"offset": offset,
"query": query,
}
return self.http.post("/store/items/search", json=_provided_vals(payload))
+1 -12
View File
@@ -325,21 +325,10 @@ class ListNamespaceResponse(TypedDict):
"""A list of namespace paths, where each path is a list of strings."""
class SearchItem(Item, total=False):
"""Item with an optional relevance score from search operations.
Attributes:
score (Optional[float]): Relevance/similarity score. Included when
searching a compatible store with a natural language query.
"""
score: Optional[float]
class SearchItemsResponse(TypedDict):
"""Response structure for searching items."""
items: list[SearchItem]
items: list[Item]
"""A list of items matching the search criteria."""
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph-sdk"
version = "0.1.40"
version = "0.1.36"
description = "SDK for interacting with LangGraph API"
authors = []
license = "MIT"
Generated
+100 -100
View File
@@ -13,108 +13,108 @@ files = [
[[package]]
name = "aiohttp"
version = "3.10.11"
version = "3.10.6"
description = "Async http client/server framework (asyncio)"
optional = false
python-versions = ">=3.8"
files = [
{file = "aiohttp-3.10.11-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:5077b1a5f40ffa3ba1f40d537d3bec4383988ee51fbba6b74aa8fb1bc466599e"},
{file = "aiohttp-3.10.11-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:8d6a14a4d93b5b3c2891fca94fa9d41b2322a68194422bef0dd5ec1e57d7d298"},
{file = "aiohttp-3.10.11-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:ffbfde2443696345e23a3c597049b1dd43049bb65337837574205e7368472177"},
{file = "aiohttp-3.10.11-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:20b3d9e416774d41813bc02fdc0663379c01817b0874b932b81c7f777f67b217"},
{file = "aiohttp-3.10.11-cp310-cp310-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:2b943011b45ee6bf74b22245c6faab736363678e910504dd7531a58c76c9015a"},
{file = "aiohttp-3.10.11-cp310-cp310-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:48bc1d924490f0d0b3658fe5c4b081a4d56ebb58af80a6729d4bd13ea569797a"},
{file = "aiohttp-3.10.11-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e12eb3f4b1f72aaaf6acd27d045753b18101524f72ae071ae1c91c1cd44ef115"},
{file = "aiohttp-3.10.11-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:f14ebc419a568c2eff3c1ed35f634435c24ead2fe19c07426af41e7adb68713a"},
{file = "aiohttp-3.10.11-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:72b191cdf35a518bfc7ca87d770d30941decc5aaf897ec8b484eb5cc8c7706f3"},
{file = "aiohttp-3.10.11-cp310-cp310-musllinux_1_2_i686.whl", hash = "sha256:5ab2328a61fdc86424ee540d0aeb8b73bbcad7351fb7cf7a6546fc0bcffa0038"},
{file = "aiohttp-3.10.11-cp310-cp310-musllinux_1_2_ppc64le.whl", hash = "sha256:aa93063d4af05c49276cf14e419550a3f45258b6b9d1f16403e777f1addf4519"},
{file = "aiohttp-3.10.11-cp310-cp310-musllinux_1_2_s390x.whl", hash = "sha256:30283f9d0ce420363c24c5c2421e71a738a2155f10adbb1a11a4d4d6d2715cfc"},
{file = "aiohttp-3.10.11-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:e5358addc8044ee49143c546d2182c15b4ac3a60be01c3209374ace05af5733d"},
{file = "aiohttp-3.10.11-cp310-cp310-win32.whl", hash = "sha256:e1ffa713d3ea7cdcd4aea9cddccab41edf6882fa9552940344c44e59652e1120"},
{file = "aiohttp-3.10.11-cp310-cp310-win_amd64.whl", hash = "sha256:778cbd01f18ff78b5dd23c77eb82987ee4ba23408cbed233009fd570dda7e674"},
{file = "aiohttp-3.10.11-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:80ff08556c7f59a7972b1e8919f62e9c069c33566a6d28586771711e0eea4f07"},
{file = "aiohttp-3.10.11-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:2c8f96e9ee19f04c4914e4e7a42a60861066d3e1abf05c726f38d9d0a466e695"},
{file = "aiohttp-3.10.11-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:fb8601394d537da9221947b5d6e62b064c9a43e88a1ecd7414d21a1a6fba9c24"},
{file = "aiohttp-3.10.11-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:2ea224cf7bc2d8856d6971cea73b1d50c9c51d36971faf1abc169a0d5f85a382"},
{file = "aiohttp-3.10.11-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:db9503f79e12d5d80b3efd4d01312853565c05367493379df76d2674af881caa"},
{file = "aiohttp-3.10.11-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:0f449a50cc33f0384f633894d8d3cd020e3ccef81879c6e6245c3c375c448625"},
{file = "aiohttp-3.10.11-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:82052be3e6d9e0c123499127782a01a2b224b8af8c62ab46b3f6197035ad94e9"},
{file = "aiohttp-3.10.11-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:20063c7acf1eec550c8eb098deb5ed9e1bb0521613b03bb93644b810986027ac"},
{file = "aiohttp-3.10.11-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:489cced07a4c11488f47aab1f00d0c572506883f877af100a38f1fedaa884c3a"},
{file = "aiohttp-3.10.11-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:ea9b3bab329aeaa603ed3bf605f1e2a6f36496ad7e0e1aa42025f368ee2dc07b"},
{file = "aiohttp-3.10.11-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:ca117819d8ad113413016cb29774b3f6d99ad23c220069789fc050267b786c16"},
{file = "aiohttp-3.10.11-cp311-cp311-musllinux_1_2_s390x.whl", hash = "sha256:2dfb612dcbe70fb7cdcf3499e8d483079b89749c857a8f6e80263b021745c730"},
{file = "aiohttp-3.10.11-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:f9b615d3da0d60e7d53c62e22b4fd1c70f4ae5993a44687b011ea3a2e49051b8"},
{file = "aiohttp-3.10.11-cp311-cp311-win32.whl", hash = "sha256:29103f9099b6068bbdf44d6a3d090e0a0b2be6d3c9f16a070dd9d0d910ec08f9"},
{file = "aiohttp-3.10.11-cp311-cp311-win_amd64.whl", hash = "sha256:236b28ceb79532da85d59aa9b9bf873b364e27a0acb2ceaba475dc61cffb6f3f"},
{file = "aiohttp-3.10.11-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:7480519f70e32bfb101d71fb9a1f330fbd291655a4c1c922232a48c458c52710"},
{file = "aiohttp-3.10.11-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:f65267266c9aeb2287a6622ee2bb39490292552f9fbf851baabc04c9f84e048d"},
{file = "aiohttp-3.10.11-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:7400a93d629a0608dc1d6c55f1e3d6e07f7375745aaa8bd7f085571e4d1cee97"},
{file = "aiohttp-3.10.11-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f34b97e4b11b8d4eb2c3a4f975be626cc8af99ff479da7de49ac2c6d02d35725"},
{file = "aiohttp-3.10.11-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:1e7b825da878464a252ccff2958838f9caa82f32a8dbc334eb9b34a026e2c636"},
{file = "aiohttp-3.10.11-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:f9f92a344c50b9667827da308473005f34767b6a2a60d9acff56ae94f895f385"},
{file = "aiohttp-3.10.11-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bc6f1ab987a27b83c5268a17218463c2ec08dbb754195113867a27b166cd6087"},
{file = "aiohttp-3.10.11-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:1dc0f4ca54842173d03322793ebcf2c8cc2d34ae91cc762478e295d8e361e03f"},
{file = "aiohttp-3.10.11-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:7ce6a51469bfaacff146e59e7fb61c9c23006495d11cc24c514a455032bcfa03"},
{file = "aiohttp-3.10.11-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:aad3cd91d484d065ede16f3cf15408254e2469e3f613b241a1db552c5eb7ab7d"},
{file = "aiohttp-3.10.11-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:f4df4b8ca97f658c880fb4b90b1d1ec528315d4030af1ec763247ebfd33d8b9a"},
{file = "aiohttp-3.10.11-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:2e4e18a0a2d03531edbc06c366954e40a3f8d2a88d2b936bbe78a0c75a3aab3e"},
{file = "aiohttp-3.10.11-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:6ce66780fa1a20e45bc753cda2a149daa6dbf1561fc1289fa0c308391c7bc0a4"},
{file = "aiohttp-3.10.11-cp312-cp312-win32.whl", hash = "sha256:a919c8957695ea4c0e7a3e8d16494e3477b86f33067478f43106921c2fef15bb"},
{file = "aiohttp-3.10.11-cp312-cp312-win_amd64.whl", hash = "sha256:b5e29706e6389a2283a91611c91bf24f218962717c8f3b4e528ef529d112ee27"},
{file = "aiohttp-3.10.11-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:703938e22434d7d14ec22f9f310559331f455018389222eed132808cd8f44127"},
{file = "aiohttp-3.10.11-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:9bc50b63648840854e00084c2b43035a62e033cb9b06d8c22b409d56eb098413"},
{file = "aiohttp-3.10.11-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:5f0463bf8b0754bc744e1feb61590706823795041e63edf30118a6f0bf577461"},
{file = "aiohttp-3.10.11-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:f6c6dec398ac5a87cb3a407b068e1106b20ef001c344e34154616183fe684288"},
{file = "aiohttp-3.10.11-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:bcaf2d79104d53d4dcf934f7ce76d3d155302d07dae24dff6c9fffd217568067"},
{file = "aiohttp-3.10.11-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:25fd5470922091b5a9aeeb7e75be609e16b4fba81cdeaf12981393fb240dd10e"},
{file = "aiohttp-3.10.11-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:bbde2ca67230923a42161b1f408c3992ae6e0be782dca0c44cb3206bf330dee1"},
{file = "aiohttp-3.10.11-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:249c8ff8d26a8b41a0f12f9df804e7c685ca35a207e2410adbd3e924217b9006"},
{file = "aiohttp-3.10.11-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:878ca6a931ee8c486a8f7b432b65431d095c522cbeb34892bee5be97b3481d0f"},
{file = "aiohttp-3.10.11-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:8663f7777ce775f0413324be0d96d9730959b2ca73d9b7e2c2c90539139cbdd6"},
{file = "aiohttp-3.10.11-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:6cd3f10b01f0c31481fba8d302b61603a2acb37b9d30e1d14e0f5a58b7b18a31"},
{file = "aiohttp-3.10.11-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:4e8d8aad9402d3aa02fdc5ca2fe68bcb9fdfe1f77b40b10410a94c7f408b664d"},
{file = "aiohttp-3.10.11-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:38e3c4f80196b4f6c3a85d134a534a56f52da9cb8d8e7af1b79a32eefee73a00"},
{file = "aiohttp-3.10.11-cp313-cp313-win32.whl", hash = "sha256:fc31820cfc3b2863c6e95e14fcf815dc7afe52480b4dc03393c4873bb5599f71"},
{file = "aiohttp-3.10.11-cp313-cp313-win_amd64.whl", hash = "sha256:4996ff1345704ffdd6d75fb06ed175938c133425af616142e7187f28dc75f14e"},
{file = "aiohttp-3.10.11-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:74baf1a7d948b3d640badeac333af581a367ab916b37e44cf90a0334157cdfd2"},
{file = "aiohttp-3.10.11-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:473aebc3b871646e1940c05268d451f2543a1d209f47035b594b9d4e91ce8339"},
{file = "aiohttp-3.10.11-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:c2f746a6968c54ab2186574e15c3f14f3e7f67aef12b761e043b33b89c5b5f95"},
{file = "aiohttp-3.10.11-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d110cabad8360ffa0dec8f6ec60e43286e9d251e77db4763a87dcfe55b4adb92"},
{file = "aiohttp-3.10.11-cp38-cp38-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:e0099c7d5d7afff4202a0c670e5b723f7718810000b4abcbc96b064129e64bc7"},
{file = "aiohttp-3.10.11-cp38-cp38-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:0316e624b754dbbf8c872b62fe6dcb395ef20c70e59890dfa0de9eafccd2849d"},
{file = "aiohttp-3.10.11-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5a5f7ab8baf13314e6b2485965cbacb94afff1e93466ac4d06a47a81c50f9cca"},
{file = "aiohttp-3.10.11-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c891011e76041e6508cbfc469dd1a8ea09bc24e87e4c204e05f150c4c455a5fa"},
{file = "aiohttp-3.10.11-cp38-cp38-musllinux_1_2_aarch64.whl", hash = "sha256:9208299251370ee815473270c52cd3f7069ee9ed348d941d574d1457d2c73e8b"},
{file = "aiohttp-3.10.11-cp38-cp38-musllinux_1_2_i686.whl", hash = "sha256:459f0f32c8356e8125f45eeff0ecf2b1cb6db1551304972702f34cd9e6c44658"},
{file = "aiohttp-3.10.11-cp38-cp38-musllinux_1_2_ppc64le.whl", hash = "sha256:14cdc8c1810bbd4b4b9f142eeee23cda528ae4e57ea0923551a9af4820980e39"},
{file = "aiohttp-3.10.11-cp38-cp38-musllinux_1_2_s390x.whl", hash = "sha256:971aa438a29701d4b34e4943e91b5e984c3ae6ccbf80dd9efaffb01bd0b243a9"},
{file = "aiohttp-3.10.11-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:9a309c5de392dfe0f32ee57fa43ed8fc6ddf9985425e84bd51ed66bb16bce3a7"},
{file = "aiohttp-3.10.11-cp38-cp38-win32.whl", hash = "sha256:9ec1628180241d906a0840b38f162a3215114b14541f1a8711c368a8739a9be4"},
{file = "aiohttp-3.10.11-cp38-cp38-win_amd64.whl", hash = "sha256:9c6e0ffd52c929f985c7258f83185d17c76d4275ad22e90aa29f38e211aacbec"},
{file = "aiohttp-3.10.11-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:cdc493a2e5d8dc79b2df5bec9558425bcd39aff59fc949810cbd0832e294b106"},
{file = "aiohttp-3.10.11-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:b3e70f24e7d0405be2348da9d5a7836936bf3a9b4fd210f8c37e8d48bc32eca6"},
{file = "aiohttp-3.10.11-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:968b8fb2a5eee2770eda9c7b5581587ef9b96fbdf8dcabc6b446d35ccc69df01"},
{file = "aiohttp-3.10.11-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:deef4362af9493d1382ef86732ee2e4cbc0d7c005947bd54ad1a9a16dd59298e"},
{file = "aiohttp-3.10.11-cp39-cp39-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:686b03196976e327412a1b094f4120778c7c4b9cff9bce8d2fdfeca386b89829"},
{file = "aiohttp-3.10.11-cp39-cp39-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:3bf6d027d9d1d34e1c2e1645f18a6498c98d634f8e373395221121f1c258ace8"},
{file = "aiohttp-3.10.11-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:099fd126bf960f96d34a760e747a629c27fb3634da5d05c7ef4d35ef4ea519fc"},
{file = "aiohttp-3.10.11-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c73c4d3dae0b4644bc21e3de546530531d6cdc88659cdeb6579cd627d3c206aa"},
{file = "aiohttp-3.10.11-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:0c5580f3c51eea91559db3facd45d72e7ec970b04528b4709b1f9c2555bd6d0b"},
{file = "aiohttp-3.10.11-cp39-cp39-musllinux_1_2_i686.whl", hash = "sha256:fdf6429f0caabfd8a30c4e2eaecb547b3c340e4730ebfe25139779b9815ba138"},
{file = "aiohttp-3.10.11-cp39-cp39-musllinux_1_2_ppc64le.whl", hash = "sha256:d97187de3c276263db3564bb9d9fad9e15b51ea10a371ffa5947a5ba93ad6777"},
{file = "aiohttp-3.10.11-cp39-cp39-musllinux_1_2_s390x.whl", hash = "sha256:0acafb350cfb2eba70eb5d271f55e08bd4502ec35e964e18ad3e7d34d71f7261"},
{file = "aiohttp-3.10.11-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:c13ed0c779911c7998a58e7848954bd4d63df3e3575f591e321b19a2aec8df9f"},
{file = "aiohttp-3.10.11-cp39-cp39-win32.whl", hash = "sha256:22b7c540c55909140f63ab4f54ec2c20d2635c0289cdd8006da46f3327f971b9"},
{file = "aiohttp-3.10.11-cp39-cp39-win_amd64.whl", hash = "sha256:7b26b1551e481012575dab8e3727b16fe7dd27eb2711d2e63ced7368756268fb"},
{file = "aiohttp-3.10.11.tar.gz", hash = "sha256:9dc2b8f3dcab2e39e0fa309c8da50c3b55e6f34ab25f1a71d3288f24924d33a7"},
{file = "aiohttp-3.10.6-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:682836fc672972cc3101cc9e30d49c5f7e8f1d010478d46119fe725a4545acfd"},
{file = "aiohttp-3.10.6-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:289fa8a20018d0d5aa9e4b35d899bd51bcb80f0d5f365d9a23e30dac3b79159b"},
{file = "aiohttp-3.10.6-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:8617c96a20dd57e7e9d398ff9d04f3d11c4d28b1767273a5b1a018ada5a654d3"},
{file = "aiohttp-3.10.6-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:bdbeff1b062751c2a2a55b171f7050fb7073633c699299d042e962aacdbe1a07"},
{file = "aiohttp-3.10.6-cp310-cp310-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:7ea35d849cdd4a9268f910bff4497baebbc1aa3f2f625fd8ccd9ac99c860c621"},
{file = "aiohttp-3.10.6-cp310-cp310-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:473961b3252f3b949bb84873d6e268fb6d8aa0ccc6eb7404fa58c76a326bb8e1"},
{file = "aiohttp-3.10.6-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3d2665c5df629eb2f981dab244c01bfa6cdc185f4ffa026639286c4d56fafb54"},
{file = "aiohttp-3.10.6-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:25d92f794f1332f656e3765841fc2b7ad5c26c3f3d01e8949eeb3495691cf9f4"},
{file = "aiohttp-3.10.6-cp310-cp310-musllinux_1_2_aarch64.whl", hash = "sha256:9bd6b2033993d5ae80883bb29b83fb2b432270bbe067c2f53cc73bb57c46065f"},
{file = "aiohttp-3.10.6-cp310-cp310-musllinux_1_2_i686.whl", hash = "sha256:d7f408c43f5e75ea1edc152fb375e8f46ef916f545fb66d4aebcbcfad05e2796"},
{file = "aiohttp-3.10.6-cp310-cp310-musllinux_1_2_ppc64le.whl", hash = "sha256:cf8b8560aa965f87bf9c13bf9fed7025993a155ca0ce8422da74bf46d18c2f5f"},
{file = "aiohttp-3.10.6-cp310-cp310-musllinux_1_2_s390x.whl", hash = "sha256:14477c4e52e2f17437b99893fd220ffe7d7ee41df5ebf931a92b8ca82e6fd094"},
{file = "aiohttp-3.10.6-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:fb138fbf9f53928e779650f5ed26d0ea1ed8b2cab67f0ea5d63afa09fdc07593"},
{file = "aiohttp-3.10.6-cp310-cp310-win32.whl", hash = "sha256:9843d683b8756971797be171ead21511d2215a2d6e3c899c6e3107fbbe826791"},
{file = "aiohttp-3.10.6-cp310-cp310-win_amd64.whl", hash = "sha256:f8b8e49fe02f744d38352daca1dbef462c3874900bd8166516f6ea8e82b5aacf"},
{file = "aiohttp-3.10.6-cp311-cp311-macosx_10_9_universal2.whl", hash = "sha256:f52e54fd776ad0da1006708762213b079b154644db54bcfc62f06eaa5b896402"},
{file = "aiohttp-3.10.6-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:995ab1a238fd0d19dc65f2d222e5eb064e409665c6426a3e51d5101c1979ee84"},
{file = "aiohttp-3.10.6-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:0749c4d5a08a802dd66ecdf59b2df4d76b900004017468a7bb736c3b5a3dd902"},
{file = "aiohttp-3.10.6-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:e05b39158f2af0e2438cc2075cfc271f4ace0c3cc4a81ec95b27a0432e161951"},
{file = "aiohttp-3.10.6-cp311-cp311-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:a9f196c970db2dcde4f24317e06615363349dc357cf4d7a3b0716c20ac6d7bcd"},
{file = "aiohttp-3.10.6-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:47647c8af04a70e07a2462931b0eba63146a13affa697afb4ecbab9d03a480ce"},
{file = "aiohttp-3.10.6-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:669c0efe7e99f6d94d63274c06344bd0e9c8daf184ce5602a29bc39e00a18720"},
{file = "aiohttp-3.10.6-cp311-cp311-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c9721cdd83a994225352ca84cd537760d41a9da3c0eacb3ff534747ab8fba6d0"},
{file = "aiohttp-3.10.6-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:0b82c8ebed66ce182893e7c0b6b60ba2ace45b1df104feb52380edae266a4850"},
{file = "aiohttp-3.10.6-cp311-cp311-musllinux_1_2_i686.whl", hash = "sha256:b169f8e755e541b72e714b89a831b315bbe70db44e33fead28516c9e13d5f931"},
{file = "aiohttp-3.10.6-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:0be3115753baf8b4153e64f9aa7bf6c0c64af57979aa900c31f496301b374570"},
{file = "aiohttp-3.10.6-cp311-cp311-musllinux_1_2_s390x.whl", hash = "sha256:e1f80cd17d81a404b6e70ef22bfe1870bafc511728397634ad5f5efc8698df56"},
{file = "aiohttp-3.10.6-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:6419728b08fb6380c66a470d2319cafcec554c81780e2114b7e150329b9a9a7f"},
{file = "aiohttp-3.10.6-cp311-cp311-win32.whl", hash = "sha256:bd294dcdc1afdc510bb51d35444003f14e327572877d016d576ac3b9a5888a27"},
{file = "aiohttp-3.10.6-cp311-cp311-win_amd64.whl", hash = "sha256:bf861da9a43d282d6dd9dcd64c23a0fccf2c5aa5cd7c32024513c8c79fb69de3"},
{file = "aiohttp-3.10.6-cp312-cp312-macosx_10_9_universal2.whl", hash = "sha256:2708baccdc62f4b1251e59c2aac725936a900081f079b88843dabcab0feeeb27"},
{file = "aiohttp-3.10.6-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:7475da7a5e2ccf1a1c86c8fee241e277f4874c96564d06f726d8df8e77683ef7"},
{file = "aiohttp-3.10.6-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:02108326574ff60267b7b35b17ac5c0bbd0008ccb942ce4c48b657bb90f0b8aa"},
{file = "aiohttp-3.10.6-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:029a019627b37fa9eac5c75cc54a6bb722c4ebbf5a54d8c8c0fb4dd8facf2702"},
{file = "aiohttp-3.10.6-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:8a637d387db6fdad95e293fab5433b775fd104ae6348d2388beaaa60d08b38c4"},
{file = "aiohttp-3.10.6-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:dc1a16f3fc1944c61290d33c88dc3f09ba62d159b284c38c5331868425aca426"},
{file = "aiohttp-3.10.6-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:81b292f37969f9cc54f4643f0be7dacabf3612b3b4a65413661cf6c350226787"},
{file = "aiohttp-3.10.6-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:0754690a3a26e819173a34093798c155bafb21c3c640bff13be1afa1e9d421f9"},
{file = "aiohttp-3.10.6-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:164ecd32e65467d86843dbb121a6666c3deb23b460e3f8aefdcaacae79eb718a"},
{file = "aiohttp-3.10.6-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:438c5863feb761f7ca3270d48c292c334814459f61cc12bab5ba5b702d7c9e56"},
{file = "aiohttp-3.10.6-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:ba18573bb1de1063d222f41de64a0d3741223982dcea863b3f74646faf618ec7"},
{file = "aiohttp-3.10.6-cp312-cp312-musllinux_1_2_s390x.whl", hash = "sha256:c82a94ddec996413a905f622f3da02c4359952aab8d817c01cf9915419525e95"},
{file = "aiohttp-3.10.6-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:92351aa5363fc3c1f872ca763f86730ced32b01607f0c9662b1fa711087968d0"},
{file = "aiohttp-3.10.6-cp312-cp312-win32.whl", hash = "sha256:3e15e33bfc73fa97c228f72e05e8795e163a693fd5323549f49367c76a6e5883"},
{file = "aiohttp-3.10.6-cp312-cp312-win_amd64.whl", hash = "sha256:fe517113fe4d35d9072b826c3e147d63c5f808ca8167d450b4f96c520c8a1d8d"},
{file = "aiohttp-3.10.6-cp313-cp313-macosx_10_13_universal2.whl", hash = "sha256:482f74057ea13d387a7549d7a7ecb60e45146d15f3e58a2d93a0ad2d5a8457cd"},
{file = "aiohttp-3.10.6-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:03fa40d1450ee5196e843315ddf74a51afc7e83d489dbfc380eecefea74158b1"},
{file = "aiohttp-3.10.6-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:1e52e59ed5f4cc3a3acfe2a610f8891f216f486de54d95d6600a2c9ba1581f4d"},
{file = "aiohttp-3.10.6-cp313-cp313-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d2b3935a22c9e41a8000d90588bed96cf395ef572dbb409be44c6219c61d900d"},
{file = "aiohttp-3.10.6-cp313-cp313-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:4bef1480ee50f75abcfcb4b11c12de1005968ca9d0172aec4a5057ba9f2b644f"},
{file = "aiohttp-3.10.6-cp313-cp313-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:671745ea7db19693ce867359d503772177f0b20fa8f6ee1e74e00449f4c4151d"},
{file = "aiohttp-3.10.6-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6b50b367308ca8c12e0b50cba5773bc9abe64c428d3fd2bbf5cd25aab37c77bf"},
{file = "aiohttp-3.10.6-cp313-cp313-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:6a504d7cdb431a777d05a124fd0b21efb94498efa743103ea01b1e3136d2e4fb"},
{file = "aiohttp-3.10.6-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:66bc81361131763660b969132a22edce2c4d184978ba39614e8f8f95db5c95f8"},
{file = "aiohttp-3.10.6-cp313-cp313-musllinux_1_2_i686.whl", hash = "sha256:27cf19a38506e2e9f12fc17e55f118f04897b0a78537055d93a9de4bf3022e3d"},
{file = "aiohttp-3.10.6-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:3468b39f977a11271517c6925b226720e148311039a380cc9117b1e2258a721f"},
{file = "aiohttp-3.10.6-cp313-cp313-musllinux_1_2_s390x.whl", hash = "sha256:9d26da22a793dfd424be1050712a70c0afd96345245c29aced1e35dbace03413"},
{file = "aiohttp-3.10.6-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:844d48ff9173d0b941abed8b2ea6a412f82b56d9ab1edb918c74000c15839362"},
{file = "aiohttp-3.10.6-cp313-cp313-win32.whl", hash = "sha256:2dd56e3c43660ed3bea67fd4c5025f1ac1f9ecf6f0b991a6e5efe2e678c490c5"},
{file = "aiohttp-3.10.6-cp313-cp313-win_amd64.whl", hash = "sha256:c91781d969fbced1993537f45efe1213bd6fccb4b37bfae2a026e20d6fbed206"},
{file = "aiohttp-3.10.6-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:4407a80bca3e694f2d2a523058e20e1f9f98a416619e04f6dc09dc910352ac8b"},
{file = "aiohttp-3.10.6-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:1cb045ec5961f51af3e2c08cd6fe523f07cc6e345033adee711c49b7b91bb954"},
{file = "aiohttp-3.10.6-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:4fabdcdc781a36b8fd7b2ca9dea8172f29a99e11d00ca0f83ffeb50958da84a1"},
{file = "aiohttp-3.10.6-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:79a9f42efcc2681790595ab3d03c0e52d01edc23a0973ea09f0dc8d295e12b8e"},
{file = "aiohttp-3.10.6-cp38-cp38-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:cca776a440795db437d82c07455761c85bbcf3956221c3c23b8c93176c278ce7"},
{file = "aiohttp-3.10.6-cp38-cp38-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:5582de171f0898139cf51dd9fcdc79b848e28d9abd68e837f0803fc9f30807b1"},
{file = "aiohttp-3.10.6-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:370e2d47575c53c817ee42a18acc34aad8da4dbdaac0a6c836d58878955f1477"},
{file = "aiohttp-3.10.6-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:444d1704e2af6b30766debed9be8a795958029e552fe77551355badb1944012c"},
{file = "aiohttp-3.10.6-cp38-cp38-musllinux_1_2_aarch64.whl", hash = "sha256:40271a2a375812967401c9ca8077de9368e09a43a964f4dce0ff603301ec9358"},
{file = "aiohttp-3.10.6-cp38-cp38-musllinux_1_2_i686.whl", hash = "sha256:f3af26f86863fad12e25395805bb0babbd49d512806af91ec9708a272b696248"},
{file = "aiohttp-3.10.6-cp38-cp38-musllinux_1_2_ppc64le.whl", hash = "sha256:4752df44df48fd42b80f51d6a97553b482cda1274d9dc5df214a3a1aa5d8f018"},
{file = "aiohttp-3.10.6-cp38-cp38-musllinux_1_2_s390x.whl", hash = "sha256:2cd5290ab66cfca2f90045db2cc6434c1f4f9fbf97c9f1c316e785033782e7d2"},
{file = "aiohttp-3.10.6-cp38-cp38-musllinux_1_2_x86_64.whl", hash = "sha256:3427031064b0d5c95647e6369c4aa3c556402f324a3e18107cb09517abe5f962"},
{file = "aiohttp-3.10.6-cp38-cp38-win32.whl", hash = "sha256:614fc21e86adc28e4165a6391f851a6da6e9cbd7bb232d0df7718b453a89ee98"},
{file = "aiohttp-3.10.6-cp38-cp38-win_amd64.whl", hash = "sha256:58c5d7318a136a3874c78717dd6de57519bc64f6363c5827c2b1cb775bea71dd"},
{file = "aiohttp-3.10.6-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:5db26bbca8e7968c4c977a0c640e0b9ce7224e1f4dcafa57870dc6ee28e27de6"},
{file = "aiohttp-3.10.6-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:3fb4216e3ec0dbc01db5ba802f02ed78ad8f07121be54eb9e918448cc3f61b7c"},
{file = "aiohttp-3.10.6-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:a976ef488f26e224079deb3d424f29144c6d5ba4ded313198169a8af8f47fb82"},
{file = "aiohttp-3.10.6-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6a86610174de8a85a920e956e2d4f9945e7da89f29a00e95ac62a4a414c4ef4e"},
{file = "aiohttp-3.10.6-cp39-cp39-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:217791c6a399cc4f2e6577bb44344cba1f5714a2aebf6a0bea04cfa956658284"},
{file = "aiohttp-3.10.6-cp39-cp39-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:ba3662d41abe2eab0eeec7ee56f33ef4e0b34858f38abf24377687f9e1fb00a5"},
{file = "aiohttp-3.10.6-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:d4dfa5ad4bce9ca30a76117fbaa1c1decf41ebb6c18a4e098df44298941566f9"},
{file = "aiohttp-3.10.6-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e0009258e97502936d3bd5bf2ced15769629097d0abb81e6495fba1047824fe0"},
{file = "aiohttp-3.10.6-cp39-cp39-musllinux_1_2_aarch64.whl", hash = "sha256:0a75d5c9fb4f06c41d029ae70ad943c3a844c40c0a769d12be4b99b04f473d3d"},
{file = "aiohttp-3.10.6-cp39-cp39-musllinux_1_2_i686.whl", hash = "sha256:8198b7c002aae2b40b2d16bfe724b9a90bcbc9b78b2566fc96131ef4e382574d"},
{file = "aiohttp-3.10.6-cp39-cp39-musllinux_1_2_ppc64le.whl", hash = "sha256:4611db8c907f90fe86be112efdc2398cd7b4c8eeded5a4f0314b70fdea8feab0"},
{file = "aiohttp-3.10.6-cp39-cp39-musllinux_1_2_s390x.whl", hash = "sha256:ff99ae06eef85c7a565854826114ced72765832ee16c7e3e766c5e4c5b98d20e"},
{file = "aiohttp-3.10.6-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:7641920bdcc7cd2d3ddfb8bb9133a6c9536b09dbd49490b79e125180b2d25b93"},
{file = "aiohttp-3.10.6-cp39-cp39-win32.whl", hash = "sha256:e2e7d5591ea868d5ec82b90bbeb366a198715672841d46281b623e23079593db"},
{file = "aiohttp-3.10.6-cp39-cp39-win_amd64.whl", hash = "sha256:b504c08c45623bf5c7ca41be380156d925f00199b3970efd758aef4a77645feb"},
{file = "aiohttp-3.10.6.tar.gz", hash = "sha256:d2578ef941be0c2ba58f6f421a703527d08427237ed45ecb091fed6f83305336"},
]
[package.dependencies]
aiohappyeyeballs = ">=2.3.0"
aiosignal = ">=1.1.2"
async-timeout = {version = ">=4.0,<6.0", markers = "python_version < \"3.11\""}
async-timeout = {version = ">=4.0,<5.0", markers = "python_version < \"3.11\""}
attrs = ">=17.3.0"
frozenlist = ">=1.1.1"
multidict = ">=4.5,<7.0"
@@ -1292,18 +1292,18 @@ tests = ["asttokens (>=2.1.0)", "coverage", "coverage-enable-subprocess", "ipyth
[[package]]
name = "fastapi"
version = "0.115.5"
version = "0.115.0"
description = "FastAPI framework, high performance, easy to learn, fast to code, ready for production"
optional = false
python-versions = ">=3.8"
files = [
{file = "fastapi-0.115.5-py3-none-any.whl", hash = "sha256:596b95adbe1474da47049e802f9a65ab2ffa9c2b07e7efee70eb8a66c9f2f796"},
{file = "fastapi-0.115.5.tar.gz", hash = "sha256:0e7a4d0dc0d01c68df21887cce0945e72d3c48b9f4f79dfe7a7d53aa08fbb289"},
{file = "fastapi-0.115.0-py3-none-any.whl", hash = "sha256:17ea427674467486e997206a5ab25760f6b09e069f099b96f5b55a32fb6f1631"},
{file = "fastapi-0.115.0.tar.gz", hash = "sha256:f93b4ca3529a8ebc6fc3fcf710e5efa8de3df9b41570958abf1d97d843138004"},
]
[package.dependencies]
pydantic = ">=1.7.4,<1.8 || >1.8,<1.8.1 || >1.8.1,<2.0.0 || >2.0.0,<2.0.1 || >2.0.1,<2.1.0 || >2.1.0,<3.0.0"
starlette = ">=0.40.0,<0.42.0"
starlette = ">=0.37.2,<0.39.0"
typing-extensions = ">=4.8.0"
[package.extras]
@@ -6446,13 +6446,13 @@ tests = ["cython", "littleutils", "pygments", "pytest", "typeguard"]
[[package]]
name = "starlette"
version = "0.40.0"
version = "0.38.6"
description = "The little ASGI library that shines."
optional = false
python-versions = ">=3.8"
files = [
{file = "starlette-0.40.0-py3-none-any.whl", hash = "sha256:c494a22fae73805376ea6bf88439783ecfba9aac88a43911b48c653437e784c4"},
{file = "starlette-0.40.0.tar.gz", hash = "sha256:1a3139688fb298ce5e2d661d37046a66ad996ce94be4d4983be019a23a04ea35"},
{file = "starlette-0.38.6-py3-none-any.whl", hash = "sha256:4517a1409e2e73ee4951214ba012052b9e16f60e90d73cfb06192c19203bbb05"},
{file = "starlette-0.38.6.tar.gz", hash = "sha256:863a1588f5574e70a821dadefb41e4881ea451a47a3cd1b4df359d4ffefe5ead"},
]
[package.dependencies]