Compare commits

..
142 changed files with 4312 additions and 7306 deletions
+1 -4
View File
@@ -1,9 +1,6 @@
blank_issues_enabled: false
version: 2.1
contact_links:
- name: Documentation
url: https://github.com/langchain-ai/docs/issues/new?template=langgraph.yml
about: Report an issue related to the LangGraph documentation
- name: LangChain Forum
url: https://forum.langchain.com/
about: General community discussions and support
about: General community discussions, support, and feature requests
+19
View File
@@ -0,0 +1,19 @@
name: Documentation
description: Report an issue related to the LangGraph documentation.
title: "DOC: <Please write a comprehensive title after the 'DOC: ' prefix>"
labels: [documentation]
body:
- type: textarea
attributes:
label: "Issue with current documentation:"
description: >
Please make sure to leave a reference to the document/code you're
referring to.
- type: textarea
attributes:
label: "Idea or request for content:"
description: >
Please describe as clearly as possible what topics you think are missing
from the current documentation.
+41 -38
View File
@@ -14,19 +14,6 @@ jobs:
python-version:
- "3.10"
- "3.11"
example:
- name: A
workdir: libs/cli/examples
tag: langgraph-test-a
- name: B
workdir: libs/cli/examples/graphs
tag: langgraph-test-b
- name: C
workdir: libs/cli/examples/graphs_reqs_a
tag: langgraph-test-c
- name: D
workdir: libs/cli/examples/graphs_reqs_b
tag: langgraph-test-d
name: "CLI integration test"
defaults:
run:
@@ -46,24 +33,54 @@ jobs:
enable-cache: true
cache-suffix: "cli-integration-test"
ignore-nothing-to-cache: true
- name: Setup env
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples
run: cat .env.example > .env
- name: Install cli globally
if: steps.changed-files.outputs.all
run: pip install -e .
- name: Build and test service ${{ matrix.example.name }}
- name: Build and test service A
if: steps.changed-files.outputs.all
working-directory: ${{ matrix.example.workdir }}
working-directory: libs/cli/examples
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
run: |
# Build the image for this example
langgraph build -t ${{ matrix.example.tag }}
# Prepare environment file from local or parent example directory
if [ -f .env.example ]; then cp .env.example .env; elif [ -f ../.env.example ]; then cp ../.env.example .env && cp ../.env.example ../.env; fi
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env; if [ -f ../.env ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> ../.env; fi; fi
# Run the integration test using the built tag
# Compute repo root to reference the shared script robustly
REPO_ROOT=$(git rev-parse --show-toplevel)
timeout 60 python "$REPO_ROOT/.github/scripts/run_langgraph_cli_test.py" -t ${{ matrix.example.tag }}
# The build-arg isn't used; just testing that we accept other args
langgraph build -t langgraph-test-a
cp .env.example .env
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env; fi
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
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
run: |
langgraph build -t langgraph-test-b
cp ../.env.example .env
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env; fi
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
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
run: |
langgraph build -t langgraph-test-c
cp ../.env.example .env
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env; fi
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
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
run: |
langgraph build -t langgraph-test-d
cp ../.env.example .env
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env; fi
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-d
- name: Build JS service
if: steps.changed-files.outputs.all
@@ -94,17 +111,3 @@ jobs:
cp ../.env.example .env
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> .env; fi
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-h
LANGGRAPH_VERSION=$(docker run --rm --entrypoint "" langgraph-test-h python -c "import sys; from importlib.metadata import version; v = version('langgraph'); print(v);")
if [ "$LANGGRAPH_VERSION" != "1.0.0a2" ]; then
exit 1
fi
LANGCHAIN_OPENAI_VERSION=$(docker run --rm --entrypoint "" langgraph-test-h python -c "import sys; from importlib.metadata import version; v = version('langchain-openai'); print(v);")
if [ "$LANGCHAIN_OPENAI_VERSION" != "0.3.0" ]; then
exit 1
fi
- name: Build and test prerelease reqs fail service
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graph_prerelease_reqs_fail
run: |
langgraph build -t langgraph-test-i || [ $? -eq 1 ]
+1 -7
View File
@@ -7,12 +7,6 @@ on:
required: true
type: string
description: "From which folder this pipeline executes"
release-branch:
required: false
type: string
default: "main"
env:
PYTHON_VERSION: "3.10"
@@ -22,7 +16,7 @@ permissions:
jobs:
build:
if: github.ref == 'refs/heads/${{ inputs.release-branch }}'
if: github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
outputs:
+1 -1
View File
@@ -57,7 +57,7 @@ jobs:
echo EOF
} >> "$GITHUB_OUTPUT"
- name: Annotation
uses: actions/github-script@v8
uses: actions/github-script@v7
with:
script: |
const file = JSON.parse(`${{ steps.files.outputs.added_modified_renamed }}`)[0]
+11 -45
View File
@@ -16,7 +16,7 @@ env:
jobs:
build:
if: github.ref == 'refs/heads/0.6'
if: github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
outputs:
@@ -79,10 +79,6 @@ jobs:
echo short-pkg-name="$SHORT_PKG_NAME" >> $GITHUB_OUTPUT
echo version="$VERSION" >> $GITHUB_OUTPUT
echo tag="$TAG" >> $GITHUB_OUTPUT
echo "Pkg-name: $PKG_NAME"
echo "Short-pkg-name: $SHORT_PKG_NAME"
echo "Version: $VERSION"
echo "Tag: $TAG"
release-notes:
needs:
@@ -97,7 +93,7 @@ jobs:
path: langgraph
sparse-checkout: | # this only grabs files for relevant dir
${{ inputs.working-directory }}
ref: "0.6" #this scopes to just 0.6 branch
ref: main # this scopes to just master branch
fetch-depth: 0 # this fetches entire commit history
- name: Check Tags
id: check-tags
@@ -109,48 +105,19 @@ jobs:
VERSION: ${{ needs.build.outputs.version }}
TAG: ${{ needs.build.outputs.tag }}
run: |
# 1) Parse major/minor and enforce we're on the 0.6 series
MAJOR="${VERSION%%.*}"
REST="${VERSION#*.}"
MINOR="${REST%%.*}"
# Minor/major numeric guard: fail if >= 0.7.* (or any non-0 major)
if [ -z "$MAJOR" ] || [ -z "$MINOR" ]; then
echo "Could not parse VERSION '$VERSION' to MAJOR.MINOR"
exit 1
fi
if [ "$MAJOR" -ne 0 ] || [ "$MINOR" -ne 6 ]; then
echo "Refusing to release VERSION '$VERSION' from 0.6 maintenance branch (got $MAJOR.$MINOR)"
exit 1
fi
# 2) Build a regex that only matches 0.6.* tags, handling both tag shapes
# - Plain: ^0\.6\.\d+((a|b|rc)\d+)?$
# - With prefix: ^<SHORT>==0\.6\.\d+((a|b|rc)\d+)?$
SERIES_REGEX="0\\.6"
if [ -z "$SHORT_PKG_NAME" ]; then
REGEX="^${SERIES_REGEX}\\.\\d+((a|b|rc)\\d+)?$"
if [ -z $SHORT_PKG_NAME ]; then
REGEX="^\\d+\\.\\d+\\.\\d+((a|b|rc)\\d+)?\$"
else
# SHORT_PKG_NAME is already pre-sanitized by your step (e.g., no spaces).
REGEX="^${SHORT_PKG_NAME}==${SERIES_REGEX}\\.\\d+((a|b|rc)\\d+)?$"
REGEX="^$SHORT_PKG_NAME==\\d+\\.\\d+\\.\\d+((a|b|rc)\\d+)?\$"
fi
echo "Tag match regex: $REGEX"
# 3) Find the most recent tag in this series
PREV_TAG="$(git tag --sort=-creatordate | grep -P "$REGEX" | head -1 || true)"
echo "Previous tag in series: ${PREV_TAG:-<none>}"
# 4) If computed TAG equals the previous tag, theres nothing new to release
if [ "$TAG" = "$PREV_TAG" ] && [ -n "$TAG" ]; then
echo "No new version to release for 0.6.x (TAG matches previous)."
echo $REGEX
PREV_TAG=$(git tag --sort=-creatordate | grep -P $REGEX | head -1 || echo "")
echo $PREV_TAG
if [ "$TAG" == "$PREV_TAG" ]; then
echo "No new version to release"
exit 1
fi
# 5) Surface prev-tag for later steps
echo "prev-tag=$PREV_TAG" >> "$GITHUB_OUTPUT"
echo prev-tag="$PREV_TAG" >> $GITHUB_OUTPUT
- name: Generate release body
id: generate-release-body
working-directory: langgraph
@@ -182,7 +149,6 @@ jobs:
uses: ./.github/workflows/_test_release.yml
with:
working-directory: ${{ inputs.working-directory }}
release-branch: "0.6"
secrets: inherit
pre-release-checks:
+2 -2
View File
@@ -33,8 +33,8 @@ jobs:
uses: peter-evans/create-pull-request@v7
with:
token: ${{ secrets.GITHUB_TOKEN }}
commit-message: "chore(deps): upgrade dependencies with `uv lock --upgrade`"
title: "chore(deps): upgrade dependencies with `uv lock --upgrade`"
commit-message: "chore[deps]: upgrade dependencies with `uv lock --upgrade`"
title: "chore[deps]: upgrade dependencies with `uv lock --upgrade`"
body: |
This PR updates the dependencies in all Python packages using `uv lock --upgrade`.
+3 -3
View File
@@ -277,9 +277,9 @@ def my_function(arg1: int, arg2: str) -> float:
Examples:
This is a section for examples of how to use the function.
```python
my_function(1, "hello")
\```
.. code-block:: python
my_function(1, "hello")
Args:
arg1: This is a description of arg1. We do not need to specify the type since
+14 -117
View File
@@ -1,126 +1,24 @@
# LangGraph Documentation
# Setup
For more information on contributing to our documentation, see the [Contributing Guide](../CONTRIBUTING.md).
## Structure
The primary documentation is located in the `docs/` directory. This directory contains both the source files for the main documentation as well as the API reference doc build process.
### Main Documentation
Main documentation files are located in `docs/docs/` and are written in Markdown format. The site uses [**MkDocs**](https://www.mkdocs.org/) with the [Material theme](https://squidfunk.github.io/mkdocs-material/) and includes:
- **Concepts**: Core LangGraph concepts and explanations
- **Tutorials**: Step-by-step learning guides
- **How-tos**: Task-focused guides for specific use cases
- **Examples**: Real-world applications and use cases
- **Jupyter Notebooks**: Interactive tutorials that are automatically converted to markdown
### API Reference
API reference documentation is defined in `docs/docs/reference/`. Each `.md` file outlines the "template" that each page is built from. Reference content is automatically generated from docstrings in the codebase using the **mkdocstrings** plugin. Once generated, the content is plugged into the corresponding markdown file where it is referenced by using manual directives to specify which classes and/or functions are documented:
```markdown
::: langgraph.graph.state.StateGraph
options:
show_if_no_docstring: true
show_root_heading: true
show_root_full_path: false
members:
- add_node
- add_edge
- add_conditional_edges
- add_sequence
- compile
```
## Build Process
Docs are built following these steps:
1. **Content Processing:**
- `_scripts/notebook_hooks.py` - Main processing pipeline that:
- Converts how-tos/tutorial Jupyter notebooks to markdown using `notebook_convert.py`
- Adds automatic API reference links to code blocks using `generate_api_reference_links.py`
- Handles conditional rendering for Python/JS versions
- Processes highlight comments and custom syntax
2. **API Reference Generation:**
- **mkdocstrings** plugin extracts docstrings from Python source code
- Manual `::: module.Class` directives in reference pages (`/docs/docs/*`) specify what to document
- Cross-references are automatically generated between docs and API
3. **Site Generation:**
- **MkDocs** processes all markdown files and generates static HTML
- Custom hooks handle redirects and inject additional functionality
4. **Deployment:**
- Site is deployed with Vercel
- `make build-docs` generates production build (also usable for local testing)
- Automatic redirects handle URL changes between versions
### Local Development
For local development, use the Makefile targets:
To setup requirements for building docs you can run:
```bash
uv sync --group test
```
## Serving documentation locally
To run the documentation server locally you can run:
```bash
# Serve docs locally with hot reloading
make serve-docs
# Clean build for production testing
make build-docs
# Serve with clean build
make serve-clean-docs
```
The `serve-docs` command:
- Watches source files for changes
- Includes dirty builds for faster iteration
- Serves on [http://127.0.0.1:8000/langgraph/](http://127.0.0.1:8000/langgraph/)
## Standards
**Docstring Format:**
The API reference uses **Google-style docstrings** with Markdown markup. The `mkdocstrings` plugin processes these to generate documentation.
**Required format:**
```python
def example_function(param1: str, param2: int = 5) -> bool:
"""Brief description of the function.
Longer description can go here. Use Markdown syntax for
rich formatting like **bold** and *italic*.
Args:
param1: Description of the first parameter.
param2: Description of the second parameter with default value.
Returns:
Description of the return value.
Raises:
ValueError: When param1 is empty.
TypeError: When param2 is not an integer.
!!! warning
This function is experimental and may change.
!!! version-added "Added in version 0.2.0"
"""
```
**Special Markers:**
- **MkDocs admonitions**: `!!! warning`, `!!! note`, `!!! version-added`
- **Code blocks**: Standard markdown ``` syntax
- **Cross-references**: Automatic linking via `generate_api_reference_links.py`
This will start the documentation server on [http://127.0.0.1:8000/langgraph/](http://127.0.0.1:8000/langgraph/).
## Execute notebooks
If you would like to automatically execute all of the notebooks, to mimic the "Run notebooks" GitHub action, you can run:
If you would like to automatically execute all of the notebooks, to mimic the "Run notebooks" GHA, you can run:
```bash
python _scripts/prepare_notebooks_for_ci.py
@@ -135,9 +33,8 @@ python _scripts/prepare_notebooks_for_ci.py --comment-install-cells
```
`prepare_notebooks_for_ci.py` script will add VCR cassette context manager for each cell in the notebook, so that:
- when the notebook is run for the first time, cells with network requests will be recorded to a VCR cassette file
- when the notebook is run subsequently, the cells with network requests will be replayed from the cassettes
* when the notebook is run for the first time, cells with network requests will be recorded to a VCR cassette file
* when the notebook is run subsequently, the cells with network requests will be replayed from the cassettes
## Adding new notebooks
+2 -14
View File
@@ -1,5 +1,3 @@
"""Generate API reference links for imports in Python code blocks within markdown files."""
import ast
import importlib
import logging
@@ -72,18 +70,8 @@ MANUAL_API_REFERENCES_LANGGRAPH = [
([], "langgraph.checkpoint.postgres.aio", "AsyncPostgresSaver", "checkpoints"),
([], "langgraph.checkpoint.postgres", "PostgresSaver", "checkpoints"),
# other prebuilts
(
["langgraph_supervisor"],
"langgraph_supervisor.supervisor",
"create_supervisor",
"supervisor",
),
(
["langgraph_supervisor"],
"langgraph_supervisor.handoff",
"create_handoff_tool",
"supervisor",
),
(["langgraph_supervisor"], "langgraph_supervisor.supervisor", "create_supervisor", "supervisor"),
(["langgraph_supervisor"], "langgraph_supervisor.handoff", "create_handoff_tool", "supervisor"),
([], "langgraph_supervisor.handoff", "create_forward_message_tool", "supervisor"),
(["langgraph_swarm"], "langgraph_swarm.swarm", "create_swarm", "swarm"),
(["langgraph_swarm"], "langgraph_swarm.swarm", "add_active_agent_router", "swarm"),
@@ -2108,9 +2108,9 @@ __metadata:
linkType: hard
"hono@npm:^4.5.4":
version: 4.9.7
resolution: "hono@npm:4.9.7"
checksum: 10c0/089184660a9211ea216ab95bafa45260e371651cb019db49828064b7982b0ae61cc3c4715324bfeb9037aa2460c39ffa2c91d84ad0c8d500fa77cbcc7fc07a8f
version: 4.9.6
resolution: "hono@npm:4.9.6"
checksum: 10c0/182a144eb3b9e05bd9e43d15af15c93f60d3d747fef6c6904b9993e9db8129ea7fadf6190331d6f76b1bf6dd2b2c3b13efea105236f541ef411397e30475422d
languageName: node
linkType: hard
-2
View File
@@ -1,5 +1,3 @@
"""Convert Jupyter notebooks to markdown with custom processing."""
import ast
import os
import re
+7 -7
View File
@@ -144,13 +144,13 @@ REDIRECT_MAP = {
"concepts/langgraph_cli.md": "https://docs.langchain.com/langgraph-platform/langgraph-cli",
"concepts/langgraph_studio.md": "https://docs.langchain.com/langgraph-platform/langgraph-studio",
"cloud/how-tos/studio/quick_start.md": "https://docs.langchain.com/langgraph-platform/quick-start-studio",
"cloud/how-tos/invoke_studio.md": "https://docs.langchain.com/langgraph-platform/use-studio#run-application",
"cloud/how-tos/studio/manage_assistants.md": "https://docs.langchain.com/langgraph-platform/use-studio#manage-assistants",
"cloud/how-tos/threads_studio.md": "https://docs.langchain.com/langgraph-platform/use-studio#manage-threads",
"cloud/how-tos/iterate_graph_studio.md": "https://docs.langchain.com/langgraph-platform/observability-studio#iterate-on-prompts",
"cloud/how-tos/studio/run_evals.md": "https://docs.langchain.com/langgraph-platform/observability-studio#run-experiments-over-a-dataset",
"cloud/how-tos/clone_traces_studio.md": "https://docs.langchain.com/langgraph-platform/observability-studio#debug-langsmith-traces",
"cloud/how-tos/datasets_studio.md": "https://docs.langchain.com/langgraph-platform/observability-studio#add-node-to-dataset",
"cloud/how-tos/invoke_studio.md": "https://docs.langchain.com/langgraph-platform/invoke-studio",
"cloud/how-tos/studio/manage_assistants.md": "https://docs.langchain.com/langgraph-platform/manage-assistants-studio",
"cloud/how-tos/threads_studio.md": "https://docs.langchain.com/langgraph-platform/threads-studio",
"cloud/how-tos/iterate_graph_studio.md": "https://docs.langchain.com/langgraph-platform/iterate-graph-studio",
"cloud/how-tos/studio/run_evals.md": "https://docs.langchain.com/langgraph-platform/run-evals-studio",
"cloud/how-tos/clone_traces_studio.md": "https://docs.langchain.com/langgraph-platform/clone-traces-studio",
"cloud/how-tos/datasets_studio.md": "https://docs.langchain.com/langgraph-platform/datasets-studio",
"concepts/sdk.md": "https://docs.langchain.com/langgraph-platform/sdk",
"concepts/plans.md": "https://docs.langchain.com/langgraph-platform/plans",
"concepts/application_structure.md": "https://docs.langchain.com/langgraph-platform/application-structure",
+1 -1
View File
@@ -33,7 +33,7 @@ LangGraph provides three ways to manage context, which combines the mutability a
**Static runtime context** represents immutable data like user metadata, tools, and database connections that are passed to an application at the start of a run via the `context` argument to `invoke`/`stream`. This data does not change during execution.
!!! version-added "Added in version 0.6.0: `context` replaces `config['configurable']`"
!!! version-added "New in LangGraph v0.6: `context` replaces `config['configurable']`"
Runtime context is now passed to the `context` argument of `invoke`/`stream`,
which replaces the previous pattern of passing application configuration to `config['configurable']`.
+1 -5
View File
@@ -211,7 +211,7 @@ output = agent.invoke(
print(output["messages"][-1].text())
```
!!! version-added "Added in version 0.6.0"
!!! version-added "New in LangGraph v0.6"
:::
@@ -351,13 +351,11 @@ If your desired LLM isn't officially supported by LangChain, consider these opti
:::python
1. **Implement a custom LangChain chat model**: Create a model conforming to the [LangChain chat model interface](https://python.langchain.com/docs/how_to/custom_chat_model/). This enables full compatibility with LangGraph's agents and workflows but requires understanding of the LangChain framework.
:::
:::js
1. **Implement a custom LangChain chat model**: Create a model conforming to the [LangChain chat model interface](https://js.langchain.com/docs/how_to/custom_chat/). This enables full compatibility with LangGraph's agents and workflows but requires understanding of the LangChain framework.
:::
2. **Direct invocation with custom streaming**: Use your model directly by [adding custom streaming logic](../how-tos/streaming.md#use-with-any-llm) with `StreamWriter`.
@@ -373,7 +371,6 @@ If your desired LLM isn't officially supported by LangChain, consider these opti
- [Force model to call a specific tool](https://python.langchain.com/docs/how_to/tool_choice/)
- [All chat model how-to guides](https://python.langchain.com/docs/how_to/#chat-models)
- [Chat model integrations](https://python.langchain.com/docs/integrations/chat/)
:::
:::js
@@ -384,5 +381,4 @@ If your desired LLM isn't officially supported by LangChain, consider these opti
- [Force model to call a specific tool](https://js.langchain.com/docs/how_to/tool_choice/)
- [All chat model how-to guides](https://js.langchain.com/docs/how_to/#chat-models)
- [Chat model integrations](https://js.langchain.com/docs/integrations/chat/)
:::
+1 -1
View File
@@ -244,7 +244,7 @@ output = agent.invoke(
print(output["messages"][-1].text())
```
!!! version-added "Added in version 0.6.0"
!!! version-added "New in langgraph>=0.6"
:::
+2 -2
View File
@@ -291,7 +291,7 @@ j=d.createElement(s),dl=l!='dataLayer'?'&l='+l:'';j.async=true;j.src=
}
.md-banner {
background-color: #FFAE42;
background-color: #CFC9FA;
color: #000000;
}
@@ -360,5 +360,5 @@ j=d.createElement(s),dl=l!='dataLayer'?'&l='+l:'';j.async=true;j.src=
{% endblock %}
{% block announce %}
These docs will be deprecated and removed with the release of LangGraph v1.0 in October 2025. <a href="https://docs.langchain.com/oss/python/langgraph/overview" target="_blank">Visit the v1.0 alpha docs</a>
Our new LangChain Academy Course Deep Research with LangGraph is now live! <a href="https://academy.langchain.com/courses/deep-research-with-langgraph/?utm_medium=internal&utm_source=docs&utm_campaign=q3-2025_deep-research-course_co" target="_blank">Enroll for free</a>.
{% endblock %}
+4 -4
View File
@@ -7,14 +7,14 @@ name = "langgraph-docs"
version = "0.0.1"
description = "LangGraph docs"
authors = []
requires-python = ">=3.11.0,<4.0.0"
requires-python = "~=3.11"
readme = "README.md"
license = "MIT"
dependencies = [
"aiohappyeyeballs==2.4.3",
"hub>=3.0.1,<4.0.0",
"xxhash>=3.5.0,<4.0.0",
"black>=25.1.0,<26.0.0",
"hub>=3.0.1,<4",
"xxhash>=3.5.0,<4",
"black>=25.1.0,<26",
]
[dependency-groups]
Generated
+4 -5
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 2
requires-python = ">=3.11, <4"
resolution-markers = [
"python_full_version >= '3.13' and platform_python_implementation != 'PyPy'",
@@ -2337,7 +2337,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.6.7"
version = "0.6.2"
source = { editable = "../libs/langgraph" }
dependencies = [
{ name = "langchain-core" },
@@ -2380,7 +2380,6 @@ dev = [
{ name = "pytest-repeat" },
{ name = "pytest-watcher" },
{ name = "pytest-xdist", extras = ["psutil"] },
{ name = "redis" },
{ name = "ruff" },
{ name = "syrupy" },
{ name = "types-requests" },
@@ -2414,7 +2413,6 @@ dev = [
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
{ name = "pytest-watcher" },
{ name = "redis" },
{ name = "ruff" },
]
@@ -2645,7 +2643,7 @@ test = [
[[package]]
name = "langgraph-prebuilt"
version = "0.6.4"
version = "0.6.2"
source = { editable = "../libs/prebuilt" }
dependencies = [
{ name = "langchain-core" },
@@ -2676,6 +2674,7 @@ dev = [
[[package]]
name = "langgraph-sdk"
version = "0.2.0"
source = { editable = "../libs/sdk-py" }
dependencies = [
{ name = "httpx" },
+1 -1
View File
@@ -5,7 +5,7 @@
"id": "18526f23",
"metadata": {},
"source": [
"This file has been moved to https://github.com/langchain-ai/langgraph/blob/main/docs/docs/how-tos/memory/add-memory.md"
"This file has been moved to https://github.com/langchain-ai/langgraph/blob/main/docs/docs/how-tos/persistence_postgres.ipynb"
]
}
],
@@ -7,6 +7,11 @@ from contextlib import contextmanager
from typing import Any
from langchain_core.runnables import RunnableConfig
from psycopg import Capabilities, Connection, Cursor, Pipeline
from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
ChannelVersions,
@@ -14,17 +19,12 @@ from langgraph.checkpoint.base import (
CheckpointMetadata,
CheckpointTuple,
get_checkpoint_id,
get_serializable_checkpoint_metadata,
get_checkpoint_metadata,
)
from langgraph.checkpoint.serde.base import SerializerProtocol
from psycopg import Capabilities, Connection, Cursor, Pipeline
from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import _internal
from langgraph.checkpoint.postgres.base import BasePostgresSaver
from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
from langgraph.checkpoint.serde.base import SerializerProtocol
Conn = _internal.Conn # For backward compatibility
@@ -325,7 +325,7 @@ class PostgresSaver(BasePostgresSaver):
checkpoint["id"],
checkpoint_id,
Jsonb(copy),
Jsonb(get_serializable_checkpoint_metadata(config, metadata)),
Jsonb(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
@@ -450,7 +450,7 @@ class PostgresSaver(BasePostgresSaver):
{
**value["checkpoint"],
"channel_values": {
**(value["checkpoint"].get("channel_values") or {}),
**value["checkpoint"].get("channel_values"),
**self._load_blobs(value["channel_values"]),
},
},
@@ -7,6 +7,11 @@ from contextlib import asynccontextmanager
from typing import Any
from langchain_core.runnables import RunnableConfig
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
ChannelVersions,
@@ -14,17 +19,12 @@ from langgraph.checkpoint.base import (
CheckpointMetadata,
CheckpointTuple,
get_checkpoint_id,
get_serializable_checkpoint_metadata,
get_checkpoint_metadata,
)
from langgraph.checkpoint.serde.base import SerializerProtocol
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres import _ainternal
from langgraph.checkpoint.postgres.base import BasePostgresSaver
from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
from langgraph.checkpoint.serde.base import SerializerProtocol
Conn = _ainternal.Conn # For backward compatibility
@@ -283,7 +283,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
checkpoint["id"],
checkpoint_id,
Jsonb(copy),
Jsonb(get_serializable_checkpoint_metadata(config, metadata)),
Jsonb(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
@@ -409,7 +409,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
{
**value["checkpoint"],
"channel_values": {
**(value["checkpoint"].get("channel_values") or {}),
**value["checkpoint"].get("channel_values"),
**self._load_blobs(value["channel_values"]),
},
},
@@ -1,12 +1,12 @@
from __future__ import annotations
import random
import warnings
from collections.abc import Sequence
from importlib.metadata import version as get_version
from typing import Any, Optional, cast
from langchain_core.runnables import RunnableConfig
from psycopg.types.json import Jsonb
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
@@ -14,22 +14,9 @@ from langgraph.checkpoint.base import (
get_checkpoint_id,
)
from langgraph.checkpoint.serde.types import TASKS
from psycopg.types.json import Jsonb
MetadataInput = Optional[dict[str, Any]]
try:
major, minor = get_version("langgraph").split(".")[:2]
if int(major) == 0 and int(minor) < 5:
warnings.warn(
"You're using incompatible versions of langgraph and checkpoint-postgres. Please upgrade langgraph to avoid unexpected behavior.",
DeprecationWarning,
stacklevel=2,
)
except Exception:
# skip version check if running from source
pass
"""
To add a new migration, add a new string to the MIGRATIONS list.
The position of the migration in the list is the version number.
@@ -6,16 +6,6 @@ from contextlib import asynccontextmanager, contextmanager
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
ChannelVersions,
Checkpoint,
CheckpointMetadata,
CheckpointTuple,
get_serializable_checkpoint_metadata,
)
from langgraph.checkpoint.serde.base import SerializerProtocol
from langgraph.checkpoint.serde.types import TASKS
from psycopg import (
AsyncConnection,
AsyncCursor,
@@ -29,8 +19,18 @@ from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import AsyncConnectionPool, ConnectionPool
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
ChannelVersions,
Checkpoint,
CheckpointMetadata,
CheckpointTuple,
get_checkpoint_metadata,
)
from langgraph.checkpoint.postgres import _ainternal, _internal
from langgraph.checkpoint.postgres.base import BasePostgresSaver
from langgraph.checkpoint.serde.base import SerializerProtocol
from langgraph.checkpoint.serde.types import TASKS
"""
To add a new migration, add a new string to the MIGRATIONS list.
@@ -441,7 +441,7 @@ class ShallowPostgresSaver(BasePostgresSaver):
thread_id,
checkpoint_ns,
Jsonb(copy),
Jsonb(get_serializable_checkpoint_metadata(config, metadata)),
Jsonb(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
@@ -774,7 +774,7 @@ class AsyncShallowPostgresSaver(BasePostgresSaver):
thread_id,
checkpoint_ns,
Jsonb(copy),
Jsonb(get_serializable_checkpoint_metadata(config, metadata)),
Jsonb(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
@@ -1,4 +1,4 @@
from langgraph.store.postgres.aio import AsyncPostgresStore
from langgraph.store.postgres.base import PoolConfig, PostgresStore
from langgraph.store.postgres.base import PostgresStore
__all__ = ["AsyncPostgresStore", "PoolConfig", "PostgresStore"]
__all__ = ["AsyncPostgresStore", "PostgresStore"]
@@ -8,6 +8,11 @@ from types import TracebackType
from typing import Any, Callable, cast
import orjson
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
from psycopg.rows import DictRow, dict_row
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres import _ainternal
from langgraph.store.base import (
GetOp,
ListNamespacesOp,
@@ -17,11 +22,6 @@ from langgraph.store.base import (
SearchOp,
)
from langgraph.store.base.batch import AsyncBatchedBaseStore
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
from psycopg.rows import DictRow, dict_row
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres import _ainternal
from langgraph.store.postgres.base import (
PLACEHOLDER,
BasePostgresStore,
@@ -22,6 +22,14 @@ from typing import (
)
import orjson
from psycopg import Capabilities, Connection, Cursor, Pipeline
from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool
from typing_extensions import TypedDict
from langgraph.checkpoint.postgres import _ainternal as _ainternal
from langgraph.checkpoint.postgres import _internal as _pg_internal
from langgraph.store.base import (
BaseStore,
GetOp,
@@ -38,14 +46,6 @@ from langgraph.store.base import (
get_text_at_path,
tokenize_path,
)
from psycopg import Capabilities, Connection, Cursor, Pipeline
from psycopg.rows import DictRow, dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool
from typing_extensions import TypedDict
from langgraph.checkpoint.postgres import _ainternal as _ainternal
from langgraph.checkpoint.postgres import _internal as _pg_internal
if TYPE_CHECKING:
from langchain_core.embeddings import Embeddings
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-checkpoint-postgres"
version = "2.0.25"
version = "2.0.23"
description = "Library with a Postgres implementation of LangGraph checkpoint saver."
authors = []
requires-python = ">=3.9"
@@ -12,7 +12,7 @@ readme = "README.md"
license = "MIT"
license-files = ['LICENSE']
dependencies = [
"langgraph-checkpoint>=2.1.2,<3.0.0",
"langgraph-checkpoint>=2.0.21,<3.0.0",
"orjson>=3.10.1",
"psycopg>=3.2.0",
"psycopg-pool>=3.2.0",
+9 -36
View File
@@ -6,6 +6,10 @@ from uuid import uuid4
import pytest
from langchain_core.runnables import RunnableConfig
from psycopg import AsyncConnection
from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.base import (
EXCLUDED_METADATA_KEYS,
Checkpoint,
@@ -13,15 +17,11 @@ from langgraph.checkpoint.base import (
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.serde.types import TASKS
from psycopg import AsyncConnection
from psycopg.rows import dict_row
from psycopg_pool import AsyncConnectionPool
from langgraph.checkpoint.postgres.aio import (
AsyncPostgresSaver,
AsyncShallowPostgresSaver,
)
from langgraph.checkpoint.serde.types import TASKS
from tests.conftest import DEFAULT_POSTGRES_URI
@@ -187,11 +187,13 @@ def test_data():
metadata_1: CheckpointMetadata = {
"source": "input",
"step": 2,
"writes": {},
"score": 1,
}
metadata_2: CheckpointMetadata = {
"source": "loop",
"step": 1,
"writes": {"foo": "bar"},
"score": None,
}
metadata_3: CheckpointMetadata = {}
@@ -218,6 +220,7 @@ async def test_combined_metadata(saver_name: str, test_data) -> None:
metadata: CheckpointMetadata = {
"source": "loop",
"step": 1,
"writes": {"foo": "bar"},
"score": None,
}
await saver.aput(config, chkpnt, metadata, {})
@@ -243,6 +246,7 @@ async def test_asearch(saver_name: str, test_data) -> None:
query_1 = {"source": "input"} # search by 1 key
query_2 = {
"step": 1,
"writes": {"foo": "bar"},
} # search by multiple keys
query_3: dict[str, Any] = {} # search by no keys, return all checkpoints
query_4 = {"source": "update", "step": 1} # no match
@@ -340,34 +344,3 @@ async def test_pending_sends_migration(saver_name: str) -> None:
TASKS: ["send-1", "send-2", "send-3"]
}
assert TASKS in search_results[0].checkpoint["channel_versions"]
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
async def test_get_checkpoint_no_channel_values(
monkeypatch, saver_name: str, test_data
) -> None:
"""Backwards compatibility test that verifies a checkpoint with no channel_values key can be retrieved without throwing an error."""
async with _saver(saver_name) as saver:
config = {
"configurable": {
"thread_id": "thread-2",
"checkpoint_ns": "",
"__super_private_key": "super_private_value",
},
"metadata": {"run_id": "my_run_id"},
}
chkpnt: Checkpoint = create_checkpoint(empty_checkpoint(), {}, 1)
await saver.aput(config, chkpnt, {}, {})
load_checkpoint_tuple = saver._load_checkpoint_tuple
def patched_load_checkpoint_tuple(value):
value["checkpoint"].pop("channel_values", None)
return load_checkpoint_tuple(value)
monkeypatch.setattr(
saver, "_load_checkpoint_tuple", patched_load_checkpoint_tuple
)
checkpoint = await saver.aget_tuple(config)
assert checkpoint.checkpoint["channel_values"] == {}
@@ -12,6 +12,8 @@ from typing import Any
import pytest
from langchain_core.embeddings import Embeddings
from psycopg import AsyncConnection
from langgraph.store.base import (
GetOp,
Item,
@@ -19,8 +21,6 @@ from langgraph.store.base import (
PutOp,
SearchOp,
)
from psycopg import AsyncConnection
from langgraph.store.postgres import AsyncPostgresStore
from tests.conftest import (
DEFAULT_URI,
+8 -3
View File
@@ -9,6 +9,8 @@ from uuid import uuid4
import pytest
from langchain_core.embeddings import Embeddings
from psycopg import Connection
from langgraph.store.base import (
GetOp,
Item,
@@ -17,8 +19,6 @@ from langgraph.store.base import (
PutOp,
SearchOp,
)
from psycopg import Connection
from langgraph.store.postgres import PostgresStore
from tests.conftest import (
DEFAULT_URI,
@@ -879,7 +879,12 @@ def test_non_ascii(
distance_type: str,
) -> None:
"""Test support for non-ascii characters"""
with _create_vector_store(vector_type, distance_type, fake_embeddings) as store:
with _create_vector_store(
vector_type,
distance_type,
fake_embeddings
) as store:
store.put(("user_123", "memories"), "1", {"text": "这是中文"}) # Chinese
store.put(
("user_123", "memories"), "2", {"text": "これは日本語です"}
+9 -35
View File
@@ -7,6 +7,10 @@ from uuid import uuid4
import pytest
from langchain_core.runnables import RunnableConfig
from psycopg import Connection
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.base import (
EXCLUDED_METADATA_KEYS,
Checkpoint,
@@ -14,12 +18,8 @@ from langgraph.checkpoint.base import (
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.serde.types import TASKS
from psycopg import Connection
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool
from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver
from langgraph.checkpoint.serde.types import TASKS
from tests.conftest import DEFAULT_POSTGRES_URI
@@ -169,11 +169,13 @@ def test_data():
metadata_1: CheckpointMetadata = {
"source": "input",
"step": 2,
"writes": {},
"score": 1,
}
metadata_2: CheckpointMetadata = {
"source": "loop",
"step": 1,
"writes": {"foo": "bar"},
"score": None,
}
metadata_3: CheckpointMetadata = {}
@@ -200,6 +202,7 @@ def test_combined_metadata(saver_name: str, test_data) -> None:
metadata: CheckpointMetadata = {
"source": "loop",
"step": 1,
"writes": {"foo": "bar"},
"score": None,
}
saver.put(config, chkpnt, metadata, {})
@@ -225,6 +228,7 @@ def test_search(saver_name: str, test_data) -> None:
query_1 = {"source": "input"} # search by 1 key
query_2 = {
"step": 1,
"writes": {"foo": "bar"},
} # search by multiple keys
query_3: dict[str, Any] = {} # search by no keys, return all checkpoints
query_4 = {"source": "update", "step": 1} # no match
@@ -328,33 +332,3 @@ def test_pending_sends_migration(saver_name: str) -> None:
TASKS: ["send-1", "send-2", "send-3"]
}
assert TASKS in search_results[0].checkpoint["channel_versions"]
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
def test_get_checkpoint_no_channel_values(
monkeypatch, saver_name: str, test_data
) -> None:
"""Backwards compatibility test that verifies a checkpoint with no channel_values key can be retrieved without throwing an error."""
with _saver(saver_name) as saver:
config = {
"configurable": {
"thread_id": "thread-2",
"checkpoint_ns": "",
"__super_private_key": "super_private_value",
},
}
chkpnt: Checkpoint = create_checkpoint(empty_checkpoint(), {}, 1)
saver.put(config, chkpnt, {}, {})
load_checkpoint_tuple = saver._load_checkpoint_tuple
def patched_load_checkpoint_tuple(value):
value["checkpoint"].pop("channel_values", None)
return load_checkpoint_tuple(value)
monkeypatch.setattr(
saver, "_load_checkpoint_tuple", patched_load_checkpoint_tuple
)
checkpoint = saver.get_tuple(config)
assert checkpoint.checkpoint["channel_values"] == {}
+500 -485
View File
File diff suppressed because it is too large Load Diff
@@ -8,6 +8,7 @@ from contextlib import closing, contextmanager
from typing import Any, cast
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
@@ -20,7 +21,6 @@ from langgraph.checkpoint.base import (
get_checkpoint_metadata,
)
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from langgraph.checkpoint.sqlite.utils import search_where
_AIO_ERROR_MSG = (
@@ -8,6 +8,7 @@ from typing import Any, Callable, TypeVar, cast
import aiosqlite
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
@@ -20,7 +21,6 @@ from langgraph.checkpoint.base import (
get_checkpoint_metadata,
)
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from langgraph.checkpoint.sqlite.utils import search_where
T = TypeVar("T", bound=Callable)
@@ -5,6 +5,7 @@ from collections.abc import Sequence
from typing import Any
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import get_checkpoint_id
@@ -11,6 +11,7 @@ from typing import Any, Callable, cast
import aiosqlite
import orjson
import sqlite_vec # type: ignore[import-untyped]
from langgraph.store.base import (
GetOp,
ListNamespacesOp,
@@ -21,7 +22,6 @@ from langgraph.store.base import (
TTLConfig,
)
from langgraph.store.base.batch import AsyncBatchedBaseStore
from langgraph.store.sqlite.base import (
_PLACEHOLDER,
BaseSqliteStore,
@@ -507,9 +507,7 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
results: List to store results in.
cur: Database cursor.
"""
prepared_queries, embedding_requests = self._prepare_batch_search_queries(
search_ops
)
queries, embedding_requests = self._prepare_batch_search_queries(search_ops)
# Setup dot_product function if it doesn't exist
if embedding_requests and self.embeddings:
@@ -517,60 +515,23 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
[query for _, query in embedding_requests]
)
for (embed_req_idx, _), embedding in zip(embedding_requests, vectors):
# Find the corresponding query in prepared_queries
# The embed_req_idx is the original index in search_ops, which should map to prepared_queries
if embed_req_idx < len(prepared_queries):
_params_list: list = prepared_queries[embed_req_idx][1]
for i, param in enumerate(_params_list):
if param is _PLACEHOLDER:
_params_list[i] = sqlite_vec.serialize_float32(embedding)
else:
logger.warning(
f"Embedding request index {embed_req_idx} out of bounds for prepared_queries."
)
for (idx, _), embedding in zip(embedding_requests, vectors):
_params_list: list = queries[idx][1]
for i, param in enumerate(_params_list):
if param is _PLACEHOLDER:
_params_list[i] = sqlite_vec.serialize_float32(embedding)
for (original_op_idx, _), (query, params, needs_refresh) in zip(
search_ops, prepared_queries
):
for (idx, _), (query, params) in zip(search_ops, queries):
await cur.execute(query, params)
rows = await cur.fetchall()
if needs_refresh and rows and self.ttl_config:
keys_to_refresh = []
for row_data in rows:
# Assuming row_data[0] is prefix (text), row_data[1] is key (text)
# These are raw text values directly from the DB.
keys_to_refresh.append((row_data[0], row_data[1]))
if keys_to_refresh:
updates_by_prefix = defaultdict(list)
for prefix_text, key_text in keys_to_refresh:
updates_by_prefix[prefix_text].append(key_text)
for prefix_text, key_list in updates_by_prefix.items():
placeholders = ",".join(["?"] * len(key_list))
update_query = f"""
UPDATE store
SET expires_at = DATETIME(CURRENT_TIMESTAMP, '+' || ttl_minutes || ' minutes')
WHERE prefix = ? AND key IN ({placeholders}) AND ttl_minutes IS NOT NULL
"""
update_params = (prefix_text, *key_list)
try:
await cur.execute(update_query, update_params)
except Exception as e:
logger.error(
f"Error during TTL refresh update for search: {e}"
)
# Process rows into items
if "score" in query: # Vector search query
if "score" in query:
items = [
_row_to_search_item(
_decode_ns_text(row[0]), # prefix
_decode_ns_text(row[0]),
{
"key": row[1], # key
"value": row[2], # value
"key": row[1],
"value": row[2],
"created_at": row[3],
"updated_at": row[4],
"expires_at": row[5] if len(row) > 5 else None,
@@ -584,10 +545,10 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
else: # Regular search query
items = [
_row_to_search_item(
_decode_ns_text(row[0]), # prefix
_decode_ns_text(row[0]),
{
"key": row[1], # key
"value": row[2], # value
"key": row[1],
"value": row[2],
"created_at": row[3],
"updated_at": row[4],
"expires_at": row[5] if len(row) > 5 else None,
@@ -598,7 +559,7 @@ class AsyncSqliteStore(AsyncBatchedBaseStore, BaseSqliteStore):
for row in rows
]
results[original_op_idx] = items
results[idx] = items
async def _batch_list_namespaces_ops(
self,
@@ -13,6 +13,7 @@ from typing import Any, Callable, Literal, NamedTuple, cast
import orjson
import sqlite_vec # type: ignore[import-untyped]
from langgraph.store.base import (
BaseStore,
GetOp,
@@ -371,15 +372,13 @@ class BaseSqliteStore:
def _prepare_batch_search_queries(
self, search_ops: Sequence[tuple[int, SearchOp]]
) -> tuple[
list[
tuple[str, list[None | str | list[float]], bool]
], # queries, params, needs_refresh
list[tuple[str, list[None | str | list[float]]]], # queries, params
list[tuple[int, str]], # idx, query_text pairs to embed
]:
"""
Build per-SearchOp SQL queries (with optional TTL refresh flag) plus embedding requests.
Build per-SearchOp SQL queries (with optional TTL refresh) plus embedding requests.
Returns:
- queries: list of (SQL, param_list, needs_ttl_refresh_flag)
- queries: list of (SQL, param_list)
- embedding_requests: list of (original_index_in_search_ops, text_query)
"""
queries = []
@@ -520,18 +519,30 @@ class BaseSqliteStore:
logger.debug(f"Search query: {base_query}")
logger.debug(f"Search params: {params}")
# Determine if TTL refresh is needed
needs_ttl_refresh = bool(
# Handle TTL refresh if requested
if (
op.refresh_ttl
and self.ttl_config
and self.ttl_config.get("refresh_on_read", False)
)
):
final_sql = f"""
WITH search_results AS (
{base_query}
),
updated AS (
UPDATE store
SET expires_at = DATETIME(CURRENT_TIMESTAMP, '+' || ttl_minutes || ' minutes')
WHERE (prefix, key) IN (SELECT prefix, key FROM search_results)
AND ttl_minutes IS NOT NULL
)
SELECT * FROM search_results
"""
final_params = params[:] # copy params
else:
final_sql = base_query
final_params = params
# The base_query is now the final_sql, and we pass the refresh flag
final_sql = base_query
final_params = params
queries.append((final_sql, final_params, needs_ttl_refresh))
queries.append((final_sql, final_params))
return queries, embedding_requests
@@ -1320,9 +1331,7 @@ class SqliteStore(BaseSqliteStore, BaseStore):
results: list[Result],
cur: sqlite3.Cursor,
) -> None:
prepared_queries, embedding_requests = self._prepare_batch_search_queries(
search_ops
)
queries, embedding_requests = self._prepare_batch_search_queries(search_ops)
# Setup similarity functions if they don't exist
if embedding_requests and self.embeddings:
@@ -1332,48 +1341,16 @@ class SqliteStore(BaseSqliteStore, BaseStore):
)
# Replace placeholders with actual embeddings
for (embed_req_idx, _), embedding in zip(embedding_requests, embeddings):
if embed_req_idx < len(prepared_queries):
_params_list: list = prepared_queries[embed_req_idx][1]
for i, param in enumerate(_params_list):
if param is _PLACEHOLDER:
_params_list[i] = sqlite_vec.serialize_float32(embedding)
else:
logger.warning(
f"Embedding request index {embed_req_idx} out of bounds for prepared_queries."
)
for (idx, _), embedding in zip(embedding_requests, embeddings):
_params_list: list = queries[idx][1]
for i, param in enumerate(_params_list):
if param is _PLACEHOLDER:
_params_list[i] = sqlite_vec.serialize_float32(embedding)
for (original_op_idx, _), (query, params, needs_refresh) in zip(
search_ops, prepared_queries
):
for (idx, _), (query, params) in zip(search_ops, queries):
cur.execute(query, params)
rows = cur.fetchall()
if needs_refresh and rows and self.ttl_config:
keys_to_refresh = []
for row_data in rows:
keys_to_refresh.append((row_data[0], row_data[1]))
if keys_to_refresh:
updates_by_prefix = defaultdict(list)
for prefix_text, key_text in keys_to_refresh:
updates_by_prefix[prefix_text].append(key_text)
for prefix_text, key_list in updates_by_prefix.items():
placeholders = ",".join(["?"] * len(key_list))
update_query = f"""
UPDATE store
SET expires_at = DATETIME(CURRENT_TIMESTAMP, '+' || ttl_minutes || ' minutes')
WHERE prefix = ? AND key IN ({placeholders}) AND ttl_minutes IS NOT NULL
"""
update_params = (prefix_text, *key_list)
try:
cur.execute(update_query, update_params)
except Exception as e:
logger.error(
f"Error during TTL refresh update for search: {e}"
)
if "score" in query: # Vector search query
items = [
_row_to_search_item(
@@ -1408,7 +1385,7 @@ class SqliteStore(BaseSqliteStore, BaseStore):
for row in rows
]
results[original_op_idx] = items
results[idx] = items
def _batch_list_namespaces_ops(
self,
@@ -2,13 +2,13 @@ from typing import Any
import pytest
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
Checkpoint,
CheckpointMetadata,
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
@@ -8,6 +8,7 @@ from contextlib import asynccontextmanager
from typing import Optional, Union, cast
import pytest
from langgraph.store.base import (
GetOp,
Item,
@@ -15,7 +16,6 @@ from langgraph.store.base import (
PutOp,
SearchOp,
)
from langgraph.store.sqlite import AsyncSqliteStore
from langgraph.store.sqlite.base import SqliteIndexConfig
from tests.test_store import CharacterEmbeddings
+2 -12
View File
@@ -2,13 +2,13 @@ from typing import Any, cast
import pytest
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
Checkpoint,
CheckpointMetadata,
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.sqlite.utils import _metadata_predicate, search_where
@@ -116,17 +116,7 @@ class TestSqliteSaver:
search_results_5[1].config["configurable"]["checkpoint_ns"],
} == {"", "inner"}
# search with before param
search_results_6 = list(saver.list(None, before=search_results_5[1].config))
assert len(search_results_6) == 1
assert search_results_6[0].config["configurable"]["thread_id"] == "thread-1"
# search with limit param
search_results_7 = list(
saver.list({"configurable": {"thread_id": "thread-2"}}, limit=1)
)
assert len(search_results_7) == 1
assert search_results_7[0].config["configurable"]["thread_id"] == "thread-2"
# TODO: test before and limit params
def test_search_where(self) -> None:
# call method / assertions
+1 -1
View File
@@ -9,6 +9,7 @@ from typing import Any, Literal, Optional, Union, cast
import pytest
from langchain_core.embeddings import Embeddings
from langgraph.store.base import (
GetOp,
Item,
@@ -17,7 +18,6 @@ from langgraph.store.base import (
PutOp,
SearchOp,
)
from langgraph.store.sqlite import SqliteStore
from langgraph.store.sqlite.base import SqliteIndexConfig
+2 -76
View File
@@ -7,7 +7,6 @@ import time
from collections.abc import Generator
import pytest
from langgraph.store.base import TTLConfig
from langgraph.store.sqlite import SqliteStore
from langgraph.store.sqlite.aio import AsyncSqliteStore
@@ -94,13 +93,9 @@ def test_ttl_sweeper(temp_db_file: str) -> None:
ttl_seconds = 2
ttl_minutes = ttl_seconds / 60
ttl_config: TTLConfig = {
"default_ttl": ttl_minutes,
"sweep_interval_minutes": ttl_minutes / 2,
}
with SqliteStore.from_conn_string(
temp_db_file,
ttl=ttl_config,
ttl={"default_ttl": ttl_minutes, "sweep_interval_minutes": ttl_minutes / 2},
) as store:
store.setup()
@@ -303,14 +298,9 @@ async def test_async_ttl_sweeper(temp_db_file: str) -> None:
ttl_seconds = 2
ttl_minutes = ttl_seconds / 60
ttl_config: TTLConfig = {
"default_ttl": ttl_minutes,
"sweep_interval_minutes": ttl_minutes / 2,
}
async with AsyncSqliteStore.from_conn_string(
temp_db_file,
ttl=ttl_config,
ttl={"default_ttl": ttl_minutes, "sweep_interval_minutes": ttl_minutes / 2},
) as store:
await store.setup()
@@ -363,67 +353,3 @@ async def test_async_search_with_ttl(temp_db_file: str) -> None:
# Search after expiration
results = await store.asearch(("test",), filter={"value": "apple"})
assert len(results) == 0
@pytest.mark.asyncio
@pytest.mark.flaky(retries=3)
async def test_async_asearch_refresh_ttl(temp_db_file: str) -> None:
"""Test TTL refresh on asearch with async API."""
ttl_seconds = 4.0 # Increased TTL for less sensitivity to timing
ttl_minutes = ttl_seconds / 60.0
async with AsyncSqliteStore.from_conn_string(
temp_db_file, ttl={"default_ttl": ttl_minutes, "refresh_on_read": True}
) as store:
await store.setup()
namespace = ("docs", "user1")
# t=0: items put, expire at t=4.0s
await store.aput(namespace, "item1", {"text": "content1", "id": 1})
await store.aput(namespace, "item2", {"text": "content2", "id": 2})
# t=3.0s: (after sleep ttl_seconds * 0.75 = 3s)
await asyncio.sleep(ttl_seconds * 0.75)
# Perform asearch with refresh_ttl=True for item1.
# item1's TTL should be refreshed. New expiry: t=3.0s + 4.0s = t=7.0s.
# item2's TTL is not affected. Expires at t=4.0s.
searched_items = await store.asearch(
namespace, filter={"id": 1}, refresh_ttl=True
)
assert len(searched_items) == 1
assert searched_items[0].key == "item1"
# t=5.0s: (after sleep ttl_seconds * 0.5 = 2s more. Total elapsed: 3s + 2s = 5s)
await asyncio.sleep(ttl_seconds * 0.5)
# At this point:
# - item1 (refreshed by asearch) should expire at t=7.0s. Should be ALIVE.
# - item2 (original TTL) should have expired at t=4.0s. Should be GONE after sweep.
await store.sweep_ttl()
# Check item1 (should exist due to asearch refresh)
item1_check1 = await store.aget(namespace, "item1", refresh_ttl=False)
assert item1_check1 is not None, (
"Item1 should exist after asearch refresh and first sweep"
)
assert item1_check1.value["text"] == "content1"
# Check item2 (should be gone)
item2_check1 = await store.aget(namespace, "item2", refresh_ttl=False)
assert item2_check1 is None, (
"Item2 should be gone after its original TTL expired"
)
# t=7.5s: (after sleep ttl_seconds * 0.625 = 2.5s more. Total elapsed: 5s + 2.5s = 7.5s)
await asyncio.sleep(ttl_seconds * 0.625)
# At this point:
# - item1 (refreshed by asearch, expired at t=7.0s) should be GONE after sweep.
await store.sweep_ttl()
# Check item1 again (should be gone now)
item1_final_check = await store.aget(namespace, "item1", refresh_ttl=False)
assert item1_final_check is None, (
"Item1 should be gone after its refreshed TTL expired"
)
+440 -426
View File
File diff suppressed because it is too large Load Diff
@@ -404,16 +404,6 @@ def get_checkpoint_metadata(
return metadata
def get_serializable_checkpoint_metadata(
config: RunnableConfig, metadata: CheckpointMetadata
) -> CheckpointMetadata:
"""Get checkpoint metadata in a backwards-compatible manner."""
checkpoint_metadata = get_checkpoint_metadata(config, metadata)
if "writes" in checkpoint_metadata:
checkpoint_metadata.pop("writes")
return checkpoint_metadata
"""
Mapping from error type to error index.
Regular writes just map to their index in the list of writes being saved.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-checkpoint"
version = "2.1.2"
version = "2.1.1"
description = "Library with base interfaces for LangGraph checkpoint savers."
authors = []
requires-python = ">=3.9"
+1 -1
View File
@@ -950,8 +950,8 @@ async def test_embed_with_path(fake_embeddings: CharacterEmbeddings) -> None:
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
assert ascore == pytest.approx(bscore, abs=1e-5)
results = await store.asearch(("test",), query="uuu")
assert len(results) == 2
+544 -553
View File
File diff suppressed because it is too large Load Diff
@@ -1,6 +1,7 @@
from collections.abc import Sequence
from typing import Annotated, Literal, TypedDict
from langchain.chat_models import init_chat_model
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
@@ -9,8 +10,10 @@ from langgraph.prebuilt import ToolNode
tools = [TavilySearchResults(max_results=1)]
model_anth = init_chat_model("claude-3-7-sonnet-20250219", model_provider="anthropic")
model_oai = ChatOpenAI(temperature=0)
model_anth = model_anth.bind_tools(tools)
model_oai = model_oai.bind_tools(tools)
@@ -32,7 +35,10 @@ def should_continue(state):
# Define the function that calls the model
def call_model(state, config):
model = model_oai
if config["configurable"].get("model", "anthropic") == "anthropic":
model = model_anth
else:
model = model_oai
messages = state["messages"]
response = model.invoke(messages)
# We return a list, because this will get added to the existing list
@@ -1,9 +0,0 @@
[project]
name = "graph-prerelease-reqs-additional-deps"
version = "0.1.0"
description = "Test for prerelease stuff"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"langgraph==0.6.0"
]
@@ -1,9 +0,0 @@
[project]
name = "graph-prerelease-reqs-zuper-deps"
version = "0.1.0"
description = "Test for prerelease stuff"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"langchain-openai==0.3.0"
]
@@ -1,9 +1,7 @@
{
"python_version": "3.12",
"dependencies": [
".",
"./deps/additional_deps",
"./deps/zuper_deps"
"."
],
"graphs": {
"agent": "./agent.py:graph"
@@ -1,14 +0,0 @@
[project]
name = "graph-prerelease-reqs"
version = "0.1.0"
description = "Test for prerelease stuff"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"langchain-openai==1.0.0a2",
"langgraph==1.0.0a2",
"langchain_community>=0.3.0",
]
[tool.uv]
prerelease = "allow"
@@ -0,0 +1,6 @@
requests
langchain_anthropic
langchain_openai
langchain_community
langchain
langgraph==1.0.0a2
@@ -1,89 +0,0 @@
from collections.abc import Sequence
from typing import Annotated, Literal, TypedDict
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, add_messages
from langgraph.prebuilt import ToolNode
tools = [TavilySearchResults(max_results=1)]
model_oai = ChatOpenAI(temperature=0)
model_oai = model_oai.bind_tools(tools)
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
# Define the function that determines whether to continue or not
def should_continue(state):
messages = state["messages"]
last_message = messages[-1]
# If there are no tool calls, then we finish
if not last_message.tool_calls:
return "end"
# Otherwise if there is, we continue
else:
return "continue"
# Define the function that calls the model
def call_model(state, config):
model = model_oai
messages = state["messages"]
response = model.invoke(messages)
# We return a list, because this will get added to the existing list
return {"messages": [response]}
# Define the function to execute tools
tool_node = ToolNode(tools)
class ContextSchema(TypedDict):
model: Literal["anthropic", "openai"]
# Define a new graph
workflow = StateGraph(AgentState, context_schema=ContextSchema)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agent`.
# This means these are the edges taken after the `agent` node is called.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
# Finally we pass in a mapping.
# The keys are strings, and the values are other nodes.
# END is a special node marking that the graph should finish.
# What will happen is we will call `should_continue`, and then the output of that
# will be matched against the keys in this mapping.
# Based on which one it matches, that node will then be called.
{
# If `tools`, then we call the tool node.
"continue": "action",
# Otherwise we finish.
"end": END,
},
)
# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("action", "agent")
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
graph = workflow.compile()
@@ -1,11 +0,0 @@
{
"python_version": "3.12",
"dependencies": [
"."
],
"graphs": {
"agent": "./agent.py:graph"
},
"env": "../.env"
}
@@ -1,11 +0,0 @@
[project]
name = "graph-prerelease-reqs"
version = "0.1.0"
description = "Test for prerelease stuff"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
"langchain-openai==1.0.0a2",
"langgraph==1.0.0a2",
"langchain_community>=0.3.0",
]
-1
View File
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"python_version": "3.12",
"dependencies": [
"langchain_community",
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"dependencies": [
"."
],
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"dependencies": [
"."
],
-1
View File
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"node_version": "20",
"graphs": {
"agent": "./src/agent/graph.ts:graph"
+1 -1
View File
@@ -1 +1 @@
__version__ = "0.4.3"
__version__ = "0.4.2"
+3 -18
View File
@@ -271,7 +271,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
f"""Ready!
- API: http://localhost:{port}
- Docs: http://localhost:{port}/docs
- LangSmith Debugger: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
- LangGraph Studio: {debugger_origin}/studio/?baseUrl={debugger_base_url_query}
"""
)
sys.stdout.flush()
@@ -652,17 +652,11 @@ def dockerfile(
help="Wait for a debugger client to connect to the debug port before starting the server",
default=False,
)
@click.option(
"--debugger-url",
type=str,
default=None,
help="URL of the LangSmith Debugger instance to connect to. Defaults to https://smith.langchain.com",
)
@click.option(
"--studio-url",
type=str,
default=None,
help="(Deprecated: use --debugger-url instead) URL of the LangSmith Debugger instance to connect to.",
help="URL of the LangGraph Studio instance to connect to. Defaults to https://smith.langchain.com",
)
@click.option(
"--allow-blocking",
@@ -698,21 +692,12 @@ def dev(
no_browser: bool,
debug_port: Optional[int],
wait_for_client: bool,
debugger_url: Optional[str],
studio_url: Optional[str],
allow_blocking: bool,
tunnel: bool,
server_log_level: str,
):
"""CLI entrypoint for running the LangGraph API server."""
if studio_url is not None:
click.secho(
"Warning: --studio-url is deprecated and will be removed in a future version. "
"Please use --debugger-url instead.",
fg="yellow",
)
if debugger_url is None:
debugger_url = studio_url
try:
from langgraph_api.cli import run_server # type: ignore
except ImportError:
@@ -776,7 +761,7 @@ def dev(
http=config_json.get("http"),
ui=config_json.get("ui"),
ui_config=config_json.get("ui_config"),
studio_url=debugger_url,
studio_url=studio_url,
allow_blocking=allow_blocking,
tunnel=tunnel,
server_level=server_log_level,
+6 -40
View File
@@ -18,7 +18,6 @@ DEFAULT_IMAGE_DISTRO = "debian"
Distros = Literal["debian", "wolfi", "bullseye", "bookworm"]
MiddlewareOrders = Literal["auth_first", "middleware_first"]
class TTLConfig(TypedDict, total=False):
@@ -360,25 +359,6 @@ class HttpConfig(TypedDict, total=False):
agent's behavior or permissions on a request's headers."""
logging_headers: Optional[ConfigurableHeaderConfig]
"""Optional. Defines which headers are excluded from logging."""
middleware_order: Optional[MiddlewareOrders]
"""Optional. Defines the order in which to apply server customizations.
Choices:
- "auth_first": Authentication hooks (custom or default) are evaluated
before custom middleware.
- "middleware_first": Custom middleware is evaluated
before authentication hooks (custom or default).
Default is `middleware_first`.
"""
enable_custom_route_auth: bool
"""Optional. If True, authentication is enabled for custom routes,
not just the routes that are protected by default.
(Routes protected by default include /assistants, /threads, and /runs).
Default is False. This flag only affects authentication behavior
if `app` is provided and contains custom routes.
"""
class Config(TypedDict, total=False):
@@ -1276,22 +1256,16 @@ def python_config_to_docker(
else:
pip_installer = "pip"
if pip_installer == "uv":
install_cmd = "uv pip install --system"
install_cmd = "uv pip install --system --prerelease=allow"
elif pip_installer == "pip":
install_cmd = "pip install"
else:
raise ValueError(f"Invalid pip_installer: {pip_installer}")
# configure pip
local_reqs_pip_install = f"PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir -c /api/constraints.txt"
global_reqs_pip_install = f"PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir -c /api/constraints.txt"
pip_install = f"PYTHONDONTWRITEBYTECODE=1 {install_cmd} --no-cache-dir -c /api/constraints.txt"
if config.get("pip_config_file"):
local_reqs_pip_install = (
f"PIP_CONFIG_FILE=/pipconfig.txt {local_reqs_pip_install}"
)
global_reqs_pip_install = (
f"PIP_CONFIG_FILE=/pipconfig.txt {global_reqs_pip_install}"
)
pip_install = f"PIP_CONFIG_FILE=/pipconfig.txt {pip_install}"
pip_config_file_str = (
f"ADD {config['pip_config_file']} /pipconfig.txt"
if config.get("pip_config_file")
@@ -1308,9 +1282,7 @@ def python_config_to_docker(
# Rewrite HTTP app path, so it points to the correct location in the Docker container
_update_http_app_path(config_path, config, local_deps)
pip_pkgs_str = (
f"RUN {local_reqs_pip_install} {' '.join(pypi_deps)}" if pypi_deps else ""
)
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(
(
@@ -1320,7 +1292,7 @@ def python_config_to_docker(
)
for reqpath, destpath in local_deps.pip_reqs
)
pip_reqs_str += f"{os.linesep}RUN {local_reqs_pip_install} {' '.join('-r ' + r for _, r 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"""# -- Installing local requirements --
{pip_reqs_str}
# -- End of local requirements install --"""
@@ -1430,13 +1402,7 @@ ADD {relpath} /deps/{name}
installs,
"",
"# -- Installing all local dependencies --",
f"""RUN for dep in /deps/*; do \
echo "Installing $dep"; \
if [ -d "$dep" ]; then \
echo "Installing $dep"; \
(cd "$dep" && {global_reqs_pip_install} .); \
fi; \
done""",
f"RUN {pip_install} -e /deps/*",
"# -- End of local dependencies install --",
os.linesep.join(env_vars),
"",
+1 -3
View File
@@ -40,9 +40,7 @@ def _parse_version(version: str) -> Version:
patch = "0"
else:
major, minor, patch = parts
return Version(
int(major.lstrip("v")), int(minor), int(patch.split("-")[0].split("+")[0])
)
return Version(int(major.lstrip("v")), int(minor), int(patch.split("-")[0]))
def check_capabilities(runner) -> DockerCapabilities:
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"dependencies": [".", "../../libs/shared", "../../libs/common"],
"graphs": {
"agent": "./src/agent/graph.py:graph"
-18
View File
@@ -577,10 +577,6 @@
"type": "boolean",
"description": "Optional. If True, /threads routes are removed.\n\nDefault is False.\n"
},
"enable_custom_route_auth": {
"type": "boolean",
"description": "Optional. If True, authentication is enabled for custom routes,\nnot just the routes that are protected by default.\n(Routes protected by default include /assistants, /threads, and /runs).\n\nDefault is False. This flag only affects authentication behavior\nif `app` is provided and contains custom routes.\n"
},
"logging_headers": {
"anyOf": [
{
@@ -591,20 +587,6 @@
}
],
"description": "Optional. Defines which headers are excluded from logging."
},
"middleware_order": {
"anyOf": [
{
"enum": [
"auth_first",
"middleware_first"
]
},
{
"type": "null"
}
],
"description": "Optional. Defines the order in which to apply server customizations.\n"
}
},
"required": []
-18
View File
@@ -577,10 +577,6 @@
"type": "boolean",
"description": "Optional. If True, /threads routes are removed.\n\nDefault is False.\n"
},
"enable_custom_route_auth": {
"type": "boolean",
"description": "Optional. If True, authentication is enabled for custom routes,\nnot just the routes that are protected by default.\n(Routes protected by default include /assistants, /threads, and /runs).\n\nDefault is False. This flag only affects authentication behavior\nif `app` is provided and contains custom routes.\n"
},
"logging_headers": {
"anyOf": [
{
@@ -591,20 +587,6 @@
}
],
"description": "Optional. Defines which headers are excluded from logging."
},
"middleware_order": {
"anyOf": [
{
"enum": [
"auth_first",
"middleware_first"
]
},
{
"type": "null"
}
],
"description": "Optional. Defines the order in which to apply server customizations.\n"
}
},
"required": []
+2 -2
View File
@@ -15,7 +15,7 @@ from langgraph_cli.docker import DEFAULT_POSTGRES_URI, DockerCapabilities, Versi
from langgraph_cli.util import clean_empty_lines
FORMATTED_CLEANUP_LINES = _get_pip_cleanup_lines(
install_cmd="uv pip install --system",
install_cmd="uv pip install --system --prerelease=allow",
to_uninstall=("pip", "setuptools", "wheel"),
pip_installer="uv",
)
@@ -149,7 +149,7 @@ services:
COPY --from=cli_1 . /deps/cli_1
# -- End of local package ../../.. --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "agent.py:graph"}}'
{textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), " ")}
+21 -21
View File
@@ -20,7 +20,7 @@ from langgraph_cli.config import (
from langgraph_cli.util import clean_empty_lines
FORMATTED_CLEANUP_LINES = _get_pip_cleanup_lines(
install_cmd="uv pip install --system",
install_cmd="uv pip install --system --prerelease=allow",
to_uninstall=("pip", "setuptools", "wheel"),
pip_installer="uv",
)
@@ -422,7 +422,7 @@ def test_config_to_docker_simple():
FROM langchain/langgraph-api:3.11
# -- Installing local requirements --
COPY --from=outer-requirements.txt requirements.txt /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt -r /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -r /deps/outer-graphs_reqs_a/graphs_reqs_a/requirements.txt
# -- End of local requirements install --
# -- Adding local package ../../examples --
COPY --from=examples . /deps/examples
@@ -456,7 +456,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency graphs_reqs_a --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGGRAPH_HTTP='{{"app": "/deps/examples/my_app.py:app"}}'
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
@@ -512,7 +512,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}'
"""
@@ -559,7 +559,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}'
"""
@@ -621,7 +621,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency graphs --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-graphs/src/agent.py:graph"}}'
{FORMATTED_CLEANUP_LINES}\
@@ -657,7 +657,7 @@ dependencies = ["langchain"]"""
ADD . /deps/unit_tests
# -- End of local package . --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{"agent": "/deps/unit_tests/graphs/agent.py:graph"}'
"""
@@ -689,7 +689,7 @@ def test_config_to_docker_end_to_end():
ARG meow
ARG foo
ADD pipconfig.txt /pipconfig.txt
RUN PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt langchain langchain_openai
RUN PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt langchain langchain_openai
# -- Adding non-package dependency graphs --
ADD ./graphs/ /deps/outer-graphs/src
RUN set -ex && \\
@@ -705,7 +705,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency graphs --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-graphs/src/agent.py:graph"}}'
{FORMATTED_CLEANUP_LINES}"""
@@ -811,7 +811,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGGRAPH_UI='{{"agent": "./graphs/agent.ui.jsx"}}'
ENV LANGGRAPH_UI_CONFIG='{{"shared": ["nuqs"]}}'
@@ -857,7 +857,7 @@ RUN set -ex && \\
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"python": "/deps/outer-unit_tests/unit_tests/multiplatform/python.py:graph", "js": "/deps/outer-unit_tests/unit_tests/multiplatform/js.mts:graph"}}'
# -- Installing JS dependencies --
@@ -887,7 +887,7 @@ def test_config_to_docker_pip_installer():
docker_auto, _ = config_to_docker(
PATH_TO_CONFIG, config_auto, "langchain/langgraph-api:0.2.47"
)
assert "uv pip install --system " in docker_auto
assert "uv pip install --system --prerelease=allow" in docker_auto
assert "rm /usr/bin/uv /usr/bin/uvx" in docker_auto
# Test explicit pip setting
@@ -895,7 +895,7 @@ def test_config_to_docker_pip_installer():
docker_pip, _ = config_to_docker(
PATH_TO_CONFIG, config_pip, "langchain/langgraph-api:0.2.47"
)
assert "uv pip install --system " not in docker_pip
assert "uv pip install --system --prerelease=allow" not in docker_pip
assert "pip install" in docker_pip
assert "rm /usr/bin/uv" not in docker_pip
@@ -904,7 +904,7 @@ def test_config_to_docker_pip_installer():
docker_uv, _ = config_to_docker(
PATH_TO_CONFIG, config_uv, "langchain/langgraph-api:0.2.47"
)
assert "uv pip install --system " in docker_uv
assert "uv pip install --system --prerelease=allow" in docker_uv
assert "rm /usr/bin/uv /usr/bin/uvx" in docker_uv
# Test auto behavior with older image (should use pip)
@@ -914,7 +914,7 @@ def test_config_to_docker_pip_installer():
docker_auto_old, _ = config_to_docker(
PATH_TO_CONFIG, config_auto_old, "langchain/langgraph-api:0.2.46"
)
assert "uv pip install --system " not in docker_auto_old
assert "uv pip install --system --prerelease=allow" not in docker_auto_old
assert "pip install" in docker_auto_old
assert "rm /usr/bin/uv" not in docker_auto_old
@@ -923,7 +923,7 @@ def test_config_to_docker_pip_installer():
docker_default, _ = config_to_docker(
PATH_TO_CONFIG, config_default, "langchain/langgraph-api:0.2.47"
)
assert "uv pip install --system " in docker_default
assert "uv pip install --system --prerelease=allow" in docker_default
def test_config_retain_build_tools():
@@ -998,7 +998,7 @@ def test_config_to_compose_simple_config():
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
{textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), " ")}
@@ -1039,7 +1039,7 @@ def test_config_to_compose_env_vars():
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
{textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), " ")}
@@ -1084,7 +1084,7 @@ def test_config_to_compose_env_file():
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
{textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), " ")}
@@ -1122,7 +1122,7 @@ def test_config_to_compose_watch():
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
{textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), " ")}
@@ -1169,7 +1169,7 @@ def test_config_to_compose_end_to_end():
done
# -- End of non-package dependency unit_tests --
# -- Installing all local dependencies --
RUN for dep in /deps/*; do echo "Installing $dep"; if [ -d "$dep" ]; then echo "Installing $dep"; (cd "$dep" && PYTHONDONTWRITEBYTECODE=1 uv pip install --system --no-cache-dir -c /api/constraints.txt .); fi; done
RUN PYTHONDONTWRITEBYTECODE=1 uv pip install --system --prerelease=allow --no-cache-dir -c /api/constraints.txt -e /deps/*
# -- End of local dependencies install --
ENV LANGSERVE_GRAPHS='{{"agent": "/deps/outer-unit_tests/unit_tests/agent.py:graph"}}'
{textwrap.indent(textwrap.dedent(FORMATTED_CLEANUP_LINES), " ")}
-22
View File
@@ -1,10 +1,7 @@
import pytest
from langgraph_cli.docker import (
DEFAULT_POSTGRES_URI,
DockerCapabilities,
Version,
_parse_version,
compose,
)
from langgraph_cli.util import clean_empty_lines
@@ -366,22 +363,3 @@ services:
REDIS_URI: redis://langgraph-redis:6379
POSTGRES_URI: {DEFAULT_POSTGRES_URI}"""
assert clean_empty_lines(actual_compose_str) == expected_compose_str
@pytest.mark.parametrize(
"input_str,expected",
[
("1.2.3", Version(1, 2, 3)),
("v1.2.3", Version(1, 2, 3)),
("1.2.3-alpha", Version(1, 2, 3)),
("1.2.3+1", Version(1, 2, 3)),
("1.2.3-alpha+build", Version(1, 2, 3)),
("1.2", Version(1, 2, 0)),
("1", Version(1, 0, 0)),
("v28.1.1+1", Version(28, 1, 1)),
("2.0.0-beta.1+exp.sha.5114f85", Version(2, 0, 0)),
("v3.4.5-rc1+build.123", Version(3, 4, 5)),
],
)
def test_parse_version_w_edge_cases(input_str, expected):
assert _parse_version(input_str) == expected
-188
View File
@@ -1,188 +0,0 @@
from unittest.mock import patch
from langgraph_cli.util import clean_empty_lines, warn_non_wolfi_distro
def test_clean_empty_lines():
"""Test clean_empty_lines function."""
# Test with empty lines
input_str = "line1\n\nline2\n\nline3"
result = clean_empty_lines(input_str)
assert result == "line1\nline2\nline3"
# Test with no empty lines
input_str = "line1\nline2\nline3"
result = clean_empty_lines(input_str)
assert result == "line1\nline2\nline3"
# Test with only empty lines
input_str = "\n\n\n"
result = clean_empty_lines(input_str)
assert result == ""
# Test empty string
input_str = ""
result = clean_empty_lines(input_str)
assert result == ""
def test_warn_non_wolfi_distro_with_debian(capsys):
"""Test that warning is shown when image_distro is 'debian'."""
config = {"image_distro": "debian"}
warn_non_wolfi_distro(config)
captured = capsys.readouterr()
assert (
"⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security."
in captured.out
)
assert (
"Wolfi is a security-oriented, minimal Linux distribution designed for containers."
in captured.out
)
assert (
'To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.'
in captured.out
)
def test_warn_non_wolfi_distro_with_default_debian(capsys):
"""Test that warning is shown when image_distro is missing (defaults to debian)."""
config = {} # No image_distro key, should default to debian
warn_non_wolfi_distro(config)
captured = capsys.readouterr()
assert (
"⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security."
in captured.out
)
assert (
"Wolfi is a security-oriented, minimal Linux distribution designed for containers."
in captured.out
)
assert (
'To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.'
in captured.out
)
def test_warn_non_wolfi_distro_with_wolfi(capsys):
"""Test that no warning is shown when image_distro is 'wolfi'."""
config = {"image_distro": "wolfi"}
warn_non_wolfi_distro(config)
captured = capsys.readouterr()
assert captured.out == "" # No output should be generated
def test_warn_non_wolfi_distro_with_other_distro(capsys):
"""Test that warning is shown when image_distro is something other than 'wolfi'."""
config = {"image_distro": "ubuntu"}
warn_non_wolfi_distro(config)
captured = capsys.readouterr()
assert (
"⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security."
in captured.out
)
assert (
"Wolfi is a security-oriented, minimal Linux distribution designed for containers."
in captured.out
)
assert (
'To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.'
in captured.out
)
def test_warn_non_wolfi_distro_output_formatting():
"""Test that the warning output is properly formatted with colors and empty line."""
config = {"image_distro": "debian"}
with patch("click.secho") as mock_secho:
warn_non_wolfi_distro(config)
# Verify click.secho was called with the correct parameters
expected_calls = [
(
(
"⚠️ Security Recommendation: Consider switching to Wolfi Linux for enhanced security.",
),
{"fg": "yellow", "bold": True},
),
(
(
" Wolfi is a security-oriented, minimal Linux distribution designed for containers.",
),
{"fg": "yellow"},
),
(
(
' To switch, add \'"image_distro": "wolfi"\' to your langgraph.json config file.',
),
{"fg": "yellow"},
),
(
("",), # Empty line
{},
),
]
assert mock_secho.call_count == 4
for i, (expected_args, expected_kwargs) in enumerate(expected_calls):
actual_call = mock_secho.call_args_list[i]
assert actual_call.args == expected_args
assert actual_call.kwargs == expected_kwargs
def test_warn_non_wolfi_distro_various_configs(capsys):
"""Test warn_non_wolfi_distro with various config scenarios."""
test_cases = [
# (config, should_warn, description)
({"image_distro": "debian"}, True, "explicit debian"),
({"image_distro": "wolfi"}, False, "explicit wolfi"),
({}, True, "missing image_distro (defaults to debian)"),
({"image_distro": "alpine"}, True, "other distro"),
({"image_distro": "ubuntu"}, True, "ubuntu distro"),
({"other_config": "value"}, True, "unrelated config keys"),
]
for config, should_warn, description in test_cases:
# Clear any previous output
capsys.readouterr()
warn_non_wolfi_distro(config)
captured = capsys.readouterr()
if should_warn:
assert "⚠️ Security Recommendation" in captured.out, (
f"Should warn for {description}"
)
assert "Wolfi" in captured.out, f"Should mention Wolfi for {description}"
else:
assert captured.out == "", f"Should not warn for {description}"
def test_warn_non_wolfi_distro_return_value():
"""Test that warn_non_wolfi_distro returns None."""
config = {"image_distro": "debian"}
result = warn_non_wolfi_distro(config)
assert result is None
config = {"image_distro": "wolfi"}
result = warn_non_wolfi_distro(config)
assert result is None
def test_warn_non_wolfi_distro_does_not_modify_config():
"""Test that warn_non_wolfi_distro does not modify the input config."""
original_config = {"image_distro": "debian", "other_key": "value"}
config_copy = original_config.copy()
warn_non_wolfi_distro(config_copy)
assert config_copy == original_config # Config should remain unchanged
+400 -617
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -76,7 +76,7 @@ test:
test_parallel:
make start-services &&\
make start-dev-server &&\
uv run pytest -n auto --dist worksteal $(TEST) --lf --snapshot-update; \
uv run pytest -n auto --dist worksteal $(TEST); \
EXIT_CODE=$$?; \
make stop-services; \
make stop-dev-server; \
+1 -1
View File
@@ -2,7 +2,6 @@ import random
from uuid import uuid4
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from pyperf._runner import Runner
from uvloop import new_event_loop
@@ -12,6 +11,7 @@ from bench.react_agent import react_agent
from bench.sequential import create_sequential
from bench.wide_dict import wide_dict
from bench.wide_state import wide_state
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
from langgraph.pregel import Pregel
@@ -114,6 +114,7 @@ if __name__ == "__main__":
import time
import uvloop
from langgraph.checkpoint.memory import InMemorySaver
graph = fanout_to_subgraph().compile(checkpointer=InMemorySaver())
+1
View File
@@ -303,6 +303,7 @@ if __name__ == "__main__":
import asyncio
import uvloop
from langgraph.checkpoint.memory import InMemorySaver
graph = pydantic_state(1000).compile(checkpointer=InMemorySaver())
+2 -1
View File
@@ -8,9 +8,9 @@ from langchain_core.language_models.fake_chat_models import (
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from langchain_core.tools import StructuredTool
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.prebuilt.chat_agent_executor import create_react_agent
from langgraph.pregel import Pregel
@@ -67,6 +67,7 @@ if __name__ == "__main__":
import asyncio
import uvloop
from langgraph.checkpoint.memory import InMemorySaver
graph = react_agent(100, checkpointer=InMemorySaver())
+1
View File
@@ -129,6 +129,7 @@ if __name__ == "__main__":
import asyncio
import uvloop
from langgraph.checkpoint.memory import InMemorySaver
graph = wide_dict(1000).compile(checkpointer=InMemorySaver())
+1
View File
@@ -139,6 +139,7 @@ if __name__ == "__main__":
import asyncio
import uvloop
from langgraph.checkpoint.memory import InMemorySaver
graph = wide_state(1000).compile(checkpointer=InMemorySaver())
@@ -17,7 +17,6 @@ from langchain_core.runnables.config import (
COPIABLE_KEYS,
var_child_runnable_config,
)
from langgraph.checkpoint.base import CheckpointMetadata
from langgraph._internal._constants import (
CONF,
@@ -27,6 +26,7 @@ from langgraph._internal._constants import (
NS_END,
NS_SEP,
)
from langgraph.checkpoint.base import CheckpointMetadata
DEFAULT_RECURSION_LIMIT = int(getenv("LANGGRAPH_DEFAULT_RECURSION_LIMIT", "25"))
@@ -41,7 +41,6 @@ from langchain_core.runnables.config import (
)
from langchain_core.runnables.utils import Input, Output
from langchain_core.tracers.langchain import LangChainTracer
from langgraph.store.base import BaseStore
from typing_extensions import TypeGuard
from langgraph._internal._config import (
@@ -55,6 +54,7 @@ from langgraph._internal._constants import (
CONFIG_KEY_RUNTIME,
)
from langgraph._internal._typing import MISSING
from langgraph.store.base import BaseStore
from langgraph.types import StreamWriter
try:
@@ -345,7 +345,7 @@ class RunnableCallable(Runnable):
args = (input,)
kwargs = {**self.kwargs, **kwargs}
runtime = config.get(CONF, {}).get(CONFIG_KEY_RUNTIME)
runtime = config[CONF].get(CONFIG_KEY_RUNTIME)
for kw, (runtime_key, default) in self.func_accepts.items():
# If the kwarg is already set, use the set value
@@ -417,7 +417,7 @@ class RunnableCallable(Runnable):
args = (input,)
kwargs = {**self.kwargs, **kwargs}
runtime = config.get(CONF, {}).get(CONFIG_KEY_RUNTIME)
runtime = config[CONF].get(CONFIG_KEY_RUNTIME)
for kw, (runtime_key, default) in self.func_accepts.items():
# If the kwarg has already been set, use the set value
+4 -16
View File
@@ -4,9 +4,9 @@ from typing import Any
from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.config import var_child_runnable_config
from langgraph.store.base import BaseStore
from langgraph._internal._constants import CONF, CONFIG_KEY_RUNTIME
from langgraph.store.base import BaseStore
from langgraph.types import StreamWriter
@@ -66,17 +66,14 @@ def get_store() -> BaseStore:
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})
class State(TypedDict):
foo: int
def my_node(state: State):
my_store = get_store()
stored_value = my_store.get(("values",), "foo").value["bar"]
return {"foo": stored_value + 1}
graph = (
StateGraph(State)
.add_node(my_node)
@@ -88,7 +85,7 @@ def get_store() -> BaseStore:
```
```pycon
{"foo": 3}
{'foo': 3}
```
Example: Using with functional API
@@ -100,19 +97,16 @@ def get_store() -> BaseStore:
store = InMemoryStore()
store.put(("values",), "foo", {"bar": 2})
@task
def my_task(value: int):
my_store = get_store()
stored_value = my_store.get(("values",), "foo").value["bar"]
return stored_value + 1
@entrypoint(store=store)
def workflow(value: int):
return my_task(value).result()
workflow.invoke(1)
```
@@ -140,17 +134,14 @@ def get_stream_writer() -> StreamWriter:
from langgraph.graph import StateGraph, START
from langgraph.config import get_stream_writer
class State(TypedDict):
foo: int
def my_node(state: State):
my_stream_writer = get_stream_writer()
my_stream_writer({"custom_data": "Hello!"})
return {"foo": state["foo"] + 1}
graph = (
StateGraph(State)
.add_node(my_node)
@@ -163,7 +154,7 @@ def get_stream_writer() -> StreamWriter:
```
```pycon
{"custom_data": "Hello!"}
{'custom_data': 'Hello!'}
```
Example: Using with functional API
@@ -171,25 +162,22 @@ def get_stream_writer() -> StreamWriter:
from langgraph.func import entrypoint, task
from langgraph.config import get_stream_writer
@task
def my_task(value: int):
my_stream_writer = get_stream_writer()
my_stream_writer({"custom_data": "Hello!"})
return value + 1
@entrypoint(store=store)
def workflow(value: int):
return my_task(value).result()
for chunk in workflow.stream(1, stream_mode="custom"):
print(chunk)
```
```pycon
{"custom_data": "Hello!"}
{'custom_data': 'Hello!'}
```
"""
runtime = get_config()[CONF][CONFIG_KEY_RUNTIME]
+2 -2
View File
@@ -5,10 +5,10 @@ from enum import Enum
from typing import Any
from warnings import warn
# EmptyChannelError is re-exported from langgraph.channels.base
from langgraph.checkpoint.base import EmptyChannelError # noqa: F401
from typing_extensions import deprecated
# EmptyChannelError is re-exported from langgraph.channels.base
from langgraph.checkpoint.base import EmptyChannelError # noqa: F401
from langgraph.types import Command, Interrupt
from langgraph.warnings import LangGraphDeprecatedSinceV10
+39 -38
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
import asyncio
import concurrent.futures
import functools
import inspect
import warnings
@@ -16,15 +18,14 @@ from typing import (
overload,
)
from langgraph.cache.base import BaseCache
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.store.base import BaseStore
from typing_extensions import Unpack
from langgraph._internal._constants import CACHE_NS_WRITES, PREVIOUS
from langgraph._internal._typing import MISSING, DeprecatedKwargs
from langgraph.cache.base import BaseCache
from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.constants import END, START
from langgraph.pregel import Pregel
from langgraph.pregel._call import (
@@ -37,6 +38,7 @@ from langgraph.pregel._call import (
)
from langgraph.pregel._read import PregelNode
from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
from langgraph.store.base import BaseStore
from langgraph.types import _DC_KWARGS, CachePolicy, RetryPolicy, StreamMode
from langgraph.typing import ContextT
from langgraph.warnings import LangGraphDeprecatedSinceV05, LangGraphDeprecatedSinceV10
@@ -47,7 +49,7 @@ __all__ = ("task", "entrypoint")
class _TaskFunction(Generic[P, T]):
def __init__(
self,
func: Callable[P, Awaitable[T]] | Callable[P, T],
func: Callable[P, T],
*,
retry_policy: Sequence[RetryPolicy],
cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
@@ -58,7 +60,7 @@ class _TaskFunction(Generic[P, T]):
# handle class methods
# NOTE: we're modifying the instance method to avoid modifying
# the original class method in case it's shared across multiple tasks
instance_method = functools.partial(func.__func__, func.__self__) # type: ignore [union-attr]
instance_method = functools.partial(func.__func__, func.__self__) # type: ignore [attr-defined]
instance_method.__name__ = name # type: ignore [attr-defined]
func = instance_method
else:
@@ -93,26 +95,32 @@ class _TaskFunction(Generic[P, T]):
@overload
def task(
__func_or_none__: None = None,
*,
name: str | None = None,
retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Callable[
[Callable[P, Awaitable[T]] | Callable[P, T]],
_TaskFunction[P, T],
]: ...
) -> Callable[[Callable[P, T]], _TaskFunction[P, T]]: ...
@overload
def task(__func_or_none__: Callable[P, Awaitable[T]]) -> _TaskFunction[P, T]: ...
def task(
*,
name: str | None = None,
retry_policy: RetryPolicy | Sequence[RetryPolicy] | None = None,
cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> Callable[[Callable[P, Awaitable[T]]], _TaskFunction[P, T]]: ...
@overload
def task(__func_or_none__: Callable[P, T]) -> _TaskFunction[P, T]: ...
@overload
def task(__func_or_none__: Callable[P, Awaitable[T]]) -> _TaskFunction[P, T]: ...
def task(
__func_or_none__: Callable[P, Awaitable[T]] | Callable[P, T] | None = None,
*,
@@ -121,7 +129,8 @@ def task(
cache_policy: CachePolicy[Callable[P, str | bytes]] | None = None,
**kwargs: Unpack[DeprecatedKwargs],
) -> (
Callable[[Callable[P, Awaitable[T]] | Callable[P, T]], _TaskFunction[P, T]]
Callable[[Callable[P, T]], _TaskFunction[P, T]]
| Callable[[Callable[P, Awaitable[T]]], _TaskFunction[P, T]]
| _TaskFunction[P, T]
):
"""Define a LangGraph task using the `task` decorator.
@@ -150,19 +159,16 @@ def task(
```python
from langgraph.func import entrypoint, task
@task
def add_one(a: int) -> int:
return a + 1
@entrypoint()
def add_one(numbers: list[int]) -> list[int]:
futures = [add_one(n) for n in numbers]
results = [f.result() for f in futures]
return results
# Call the entrypoint
add_one.invoke([1, 2, 3]) # Returns [2, 3, 4]
```
@@ -172,18 +178,15 @@ def task(
import asyncio
from langgraph.func import entrypoint, task
@task
async def add_one(a: int) -> int:
return a + 1
@entrypoint()
async def add_one(numbers: list[int]) -> list[int]:
futures = [add_one(n) for n in numbers]
return asyncio.gather(*futures)
# Call the entrypoint
await add_one.ainvoke([1, 2, 3]) # Returns [2, 3, 4]
```
@@ -207,7 +210,7 @@ def task(
def decorator(
func: Callable[P, Awaitable[T]] | Callable[P, T],
) -> Callable[P, SyncAsyncFuture[T]]:
) -> Callable[P, concurrent.futures.Future[T]] | Callable[P, asyncio.Future[T]]:
return _TaskFunction(
func, retry_policy=retry_policies, cache_policy=cache_policy, name=name
)
@@ -348,13 +351,15 @@ class entrypoint(Generic[ContextT]):
from langgraph.func import entrypoint
@entrypoint(checkpointer=InMemorySaver())
def my_workflow(input_data: str, previous: Optional[str] = None) -> str:
return "world"
config = {"configurable": {"thread_id": "some_thread"}}
config = {
"configurable": {
"thread_id": "some_thread"
}
}
my_workflow.invoke("hello", config)
```
@@ -371,21 +376,19 @@ class entrypoint(Generic[ContextT]):
from langgraph.func import entrypoint
@entrypoint(checkpointer=InMemorySaver())
def my_workflow(
number: int,
*,
previous: Any = None,
) -> entrypoint.final[int, int]:
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
previous = previous or 0
# This will return the previous value to the caller, saving
# 2 * number to the checkpoint, which will be used in the next invocation
# for the `previous` parameter.
return entrypoint.final(value=previous, save=2 * number)
config = {"configurable": {"thread_id": "some_thread"}}
config = {
"configurable": {
"thread_id": "some_thread"
}
}
my_workflow.invoke(3, config) # 0 (previous was None)
my_workflow.invoke(1, config) # 6 (previous was 3 * 2 from the previous invocation)
@@ -440,21 +443,19 @@ class entrypoint(Generic[ContextT]):
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.func import entrypoint
@entrypoint(checkpointer=InMemorySaver())
def my_workflow(
number: int,
*,
previous: Any = None,
) -> entrypoint.final[int, int]:
def my_workflow(number: int, *, previous: Any = None) -> entrypoint.final[int, int]:
previous = previous or 0
# This will return the previous value to the caller, saving
# 2 * number to the checkpoint, which will be used in the next invocation
# for the `previous` parameter.
return entrypoint.final(value=previous, save=2 * number)
config = {"configurable": {"thread_id": "1"}}
config = {
"configurable": {
"thread_id": "1"
}
}
my_workflow.invoke(3, config) # 0 (previous was None)
my_workflow.invoke(1, config) # 6 (previous was 3 * 2 from the previous invocation)
+1 -1
View File
@@ -6,11 +6,11 @@ from dataclasses import dataclass
from typing import Any, Generic, Protocol, Union
from langchain_core.runnables import Runnable, RunnableConfig
from langgraph.store.base import BaseStore
from typing_extensions import TypeAlias
from langgraph._internal._typing import EMPTY_SEQ
from langgraph.runtime import Runtime
from langgraph.store.base import BaseStore
from langgraph.types import CachePolicy, RetryPolicy, StreamWriter
from langgraph.typing import ContextT, NodeInputT, NodeInputT_contra
+20 -28
View File
@@ -92,7 +92,6 @@ def add_messages(
Example:
```python title="Basic usage"
from langchain_core.messages import AIMessage, HumanMessage
msgs1 = [HumanMessage(content="Hello", id="1")]
msgs2 = [AIMessage(content="Hi there!", id="2")]
add_messages(msgs1, msgs2)
@@ -111,11 +110,9 @@ def add_messages(
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
class State(TypedDict):
messages: Annotated[list, add_messages]
builder = StateGraph(State)
builder.add_node("chatbot", lambda state: {"messages": [("assistant", "Hello")]})
builder.set_entry_point("chatbot")
@@ -130,35 +127,30 @@ def add_messages(
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, add_messages
class State(TypedDict):
messages: Annotated[list, add_messages(format="langchain-openai")]
messages: Annotated[list, add_messages(format='langchain-openai')]
def chatbot_node(state: State) -> list:
return {
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Here's an image:",
"cache_control": {"type": "ephemeral"},
return {"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "Here's an image:",
"cache_control": {"type": "ephemeral"},
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": "1234",
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": "1234",
},
},
],
},
]
}
},
]
},
]}
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
+7 -20
View File
@@ -24,9 +24,6 @@ from typing import (
)
from langchain_core.runnables import Runnable, RunnableConfig
from langgraph.cache.base import BaseCache
from langgraph.checkpoint.base import Checkpoint
from langgraph.store.base import BaseStore
from pydantic import BaseModel, TypeAdapter
from typing_extensions import NotRequired, Required, Self, Unpack, is_typeddict
@@ -44,6 +41,7 @@ from langgraph._internal._fields import (
from langgraph._internal._pydantic import create_model
from langgraph._internal._runnable import coerce_to_runnable
from langgraph._internal._typing import EMPTY_SEQ, MISSING, DeprecatedKwargs
from langgraph.cache.base import BaseCache
from langgraph.channels.base import BaseChannel
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.ephemeral_value import EphemeralValue
@@ -52,6 +50,7 @@ from langgraph.channels.named_barrier_value import (
NamedBarrierValue,
NamedBarrierValueAfterFinish,
)
from langgraph.checkpoint.base import Checkpoint
from langgraph.constants import END, START, TAG_HIDDEN
from langgraph.errors import (
ErrorCode,
@@ -72,6 +71,7 @@ from langgraph.pregel._write import (
ChannelWriteEntry,
ChannelWriteTupleEntry,
)
from langgraph.store.base import BaseStore
from langgraph.types import (
All,
CachePolicy,
@@ -141,31 +141,25 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
def reducer(a: list, b: int | None) -> list:
if b is not None:
return a + [b]
return a
class State(TypedDict):
x: Annotated[list, reducer]
class Context(TypedDict):
r: float
graph = StateGraph(state_schema=State, context_schema=Context)
def node(state: State, runtime: Runtime[Context]) -> dict:
r = runtime.context.get("r", 1.0)
x = state["x"][-1]
next_value = x * r * (1 - x)
return {"x": next_value}
graph.add_node("A", node)
graph.set_entry_point("A")
graph.set_finish_point("A")
@@ -391,15 +385,12 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
from langchain_core.runnables import RunnableConfig
from langgraph.graph import START, StateGraph
class State(TypedDict):
x: int
def my_node(state: State, config: RunnableConfig) -> State:
return {"x": state["x"] + 1}
builder = StateGraph(State)
builder.add_node(my_node) # node name will be 'my_node'
builder.add_edge(START, "my_node")
@@ -1369,14 +1360,10 @@ def _get_channel(
def _is_field_channel(typ: type[Any]) -> BaseChannel | None:
if hasattr(typ, "__metadata__"):
meta = typ.__metadata__
# Search through all annotated medata to find channel annotations
for item in meta:
if isinstance(item, BaseChannel):
return item
elif isclass(item) and issubclass(item, BaseChannel):
# ex, Annotated[int, EphemeralValue, SomeOtherAnnotation]
# would return EphemeralValue(int)
return item(typ.__origin__ if hasattr(typ, "__origin__") else typ)
if len(meta) >= 1 and isinstance(meta[-1], BaseChannel):
return meta[-1]
elif len(meta) >= 1 and isclass(meta[-1]) and issubclass(meta[-1], BaseChannel):
return meta[-1](typ.__origin__ if hasattr(typ, "__origin__") else typ)
return None
+10 -9
View File
@@ -82,19 +82,18 @@ def push_ui_message(
message: Optional message object to associate with the UI message.
state_key: Key in the graph state where the UI messages are stored.
Defaults to "ui".
merge: Whether to merge props with existing UI message (True) or replace
them (False). Defaults to False.
Returns:
The created UI message.
Example:
```python
.. code-block:: python
push_ui_message(
name="component-name",
props={"content": "Hello world"},
)
```
"""
from langgraph._internal._constants import CONFIG_KEY_SEND
@@ -145,9 +144,10 @@ def delete_ui_message(id: str, *, state_key: str = "ui") -> RemoveUIMessage:
The remove UI message.
Example:
```python
.. code-block:: python
delete_ui_message("message-123")
```
"""
from langgraph._internal._constants import CONFIG_KEY_SEND
@@ -181,12 +181,13 @@ def ui_message_reducer(
Combined list of UI messages with removals applied.
Example:
```python
.. code-block:: python
messages = ui_message_reducer(
[{"type": "ui", "id": "1", "name": "Chat", "props": {}}],
{"type": "remove-ui", "id": "1"},
{"type": "remove-ui", "id": "1"}
)
```
"""
if not isinstance(left, list):
+12 -18
View File
@@ -23,14 +23,6 @@ from typing import (
from langchain_core.callbacks import Callbacks
from langchain_core.callbacks.manager import AsyncParentRunManager, ParentRunManager
from langchain_core.runnables.config import RunnableConfig
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
ChannelVersions,
Checkpoint,
PendingWrite,
V,
)
from langgraph.store.base import BaseStore
from xxhash import xxh3_128_hexdigest
from langgraph._internal._config import merge_configs, patch_config
@@ -65,6 +57,13 @@ from langgraph._internal._scratchpad import PregelScratchpad
from langgraph._internal._typing import EMPTY_SEQ, MISSING
from langgraph.channels.base import BaseChannel
from langgraph.channels.topic import Topic
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
ChannelVersions,
Checkpoint,
PendingWrite,
V,
)
from langgraph.constants import TAG_HIDDEN
from langgraph.managed.base import ManagedValueMapping
from langgraph.pregel._call import get_runnable_for_task, identifier
@@ -72,6 +71,7 @@ from langgraph.pregel._io import read_channels
from langgraph.pregel._log import logger
from langgraph.pregel._read import INPUT_CACHE_KEY_TYPE, PregelNode
from langgraph.runtime import DEFAULT_RUNTIME, Runtime
from langgraph.store.base import BaseStore
from langgraph.types import (
All,
CacheKey,
@@ -715,17 +715,13 @@ def prepare_single_task(
runtime = runtime.override(
store=store, previous=checkpoint["channel_values"].get(PREVIOUS, None)
)
additional_config: RunnableConfig = {
"metadata": metadata,
"tags": proc.tags,
}
return PregelExecutableTask(
packet.node,
packet.arg,
proc_node,
writes,
patch_config(
merge_configs(config, additional_config),
merge_configs(config, {"metadata": metadata, "tags": proc.tags}),
run_name=packet.node,
callbacks=(
manager.get_child(f"graph:step:{step}") if manager else None
@@ -860,17 +856,15 @@ def prepare_single_task(
previous=checkpoint["channel_values"].get(PREVIOUS, None),
store=store,
)
additional_config = {
"metadata": metadata,
"tags": proc.tags,
}
return PregelExecutableTask(
name,
val,
node,
writes,
patch_config(
merge_configs(config, additional_config),
merge_configs(
config, {"metadata": metadata, "tags": proc.tags}
),
run_name=name,
callbacks=(
manager.get_child(f"graph:step:{step}")
+2 -2
View File
@@ -7,7 +7,7 @@ import functools
import inspect
import sys
import types
from collections.abc import Awaitable, Generator, Sequence
from collections.abc import Generator, Sequence
from typing import Any, Callable, Generic, TypeVar, cast
from langchain_core.runnables import Runnable
@@ -251,7 +251,7 @@ class SyncAsyncFuture(Generic[T], concurrent.futures.Future[T]):
def call(
func: Callable[P, Awaitable[T]] | Callable[P, T],
func: Callable[P, T],
*args: Any,
retry_policy: Sequence[RetryPolicy] | None = None,
cache_policy: CachePolicy | None = None,
@@ -3,11 +3,10 @@ from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime, timezone
from langgraph.checkpoint.base import Checkpoint
from langgraph.checkpoint.base.id import uuid6
from langgraph._internal._typing import MISSING
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import Checkpoint
from langgraph.checkpoint.base.id import uuid6
from langgraph.managed.base import ManagedValueMapping, ManagedValueSpec
LATEST_VERSION = 4
+19 -52
View File
@@ -2,15 +2,14 @@ from __future__ import annotations
from collections import defaultdict
from collections.abc import Mapping, Sequence
from typing import Any, NamedTuple, cast
from typing import Any, cast
from langchain_core.runnables.config import RunnableConfig
from langchain_core.runnables.graph import Graph, Node
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph._internal._constants import CONF, CONFIG_KEY_SEND, INPUT
from langgraph.channels.base import BaseChannel
from langgraph.channels.last_value import LastValueAfterFinish
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.constants import END, START
from langgraph.managed.base import ManagedValueSpec
from langgraph.pregel._algo import (
@@ -26,19 +25,6 @@ from langgraph.pregel._write import ChannelWrite
from langgraph.types import All, Checkpointer
class Edge(NamedTuple):
source: str
target: str
conditional: bool
data: str | None
class TriggerEdge(NamedTuple):
source: str
conditional: bool
data: str | None
def draw_graph(
config: RunnableConfig,
*,
@@ -63,7 +49,7 @@ def draw_graph(
The graph for this Pregel instance.
"""
# (src, dest, is_conditional, label)
edges: set[Edge] = set()
edges: set[tuple[str, str, bool, str | None]] = set()
step = -1
checkpoint = empty_checkpoint()
@@ -77,9 +63,8 @@ def draw_graph(
checkpoint,
)
static_seen: set[Any] = set()
sources: dict[str, set[TriggerEdge]] = {}
step_sources: dict[str, set[TriggerEdge]] = {}
static_declared_writes: dict[str, set[TriggerEdge]] = defaultdict(set)
sources: dict[str, set[tuple[str, bool, str | None]]] = {}
step_sources: dict[str, set[tuple[str, bool, str | None]]] = {}
# remove node mappers
nodes = {
k: v.copy(update={"mapper": None}) if v.mapper is not None else v
@@ -138,36 +123,32 @@ def draw_graph(
# END writes are not written, but become edges directly
for t in writes:
if t[0] == END:
edges.add(Edge(task.name, t[0], True, t[2]))
edges.add((task.name, t[0], True, t[2]))
writes = [t for t in writes if t[0] != END]
conditionals.update(
{(task.name, t[0], t[1] or None): t[2] for t in writes}
)
# record static writes for edge creation
for t in writes:
static_declared_writes[task.name].add(
TriggerEdge(t[0], True, t[2])
)
task.config[CONF][CONFIG_KEY_SEND]([t[:2] for t in writes])
# collect sources
step_sources = {}
for task in tasks.values():
task_edges = {
TriggerEdge(
step_sources = {
task.name: {
(
w[0],
(task.name, w[0], w[1] or None) in conditionals,
conditionals.get((task.name, w[0], w[1] or None)),
)
for w in task.writes
}
task_edges |= static_declared_writes.get(task.name, set())
step_sources[task.name] = task_edges
for task in tasks.values()
}
sources.update(step_sources)
# invert triggers
trigger_to_sources: dict[str, set[TriggerEdge]] = defaultdict(set)
trigger_to_sources: dict[str, set[tuple[str, bool, str | None]]] = defaultdict(
set
)
for src, triggers in sources.items():
for trigger, cond, label in triggers:
trigger_to_sources[trigger].add(TriggerEdge(src, cond, label))
trigger_to_sources[trigger].add((src, cond, label))
# apply writes
updated_channels = apply_writes(
checkpoint, channels, tasks.values(), get_next_version, trigger_to_nodes
@@ -189,39 +170,26 @@ def draw_graph(
trigger_to_nodes=trigger_to_nodes,
updated_channels=updated_channels,
)
# collect deferred nodes
deferred_nodes: set[str] = set()
edges_to_deferred_nodes: set[Edge] = set()
for channel, item in channels.items():
if isinstance(item, LastValueAfterFinish):
deferred_node = channel.split(":", 2)[-1]
deferred_nodes.add(deferred_node)
# collect edges
for task in tasks.values():
added = False
for trigger in task.triggers:
for src, cond, label in sorted(trigger_to_sources[trigger]):
# record edge to be reviewed later
if task.name in deferred_nodes:
edges_to_deferred_nodes.add(Edge(src, task.name, cond, label))
edges.add(Edge(src, task.name, cond, label))
edges.add((src, task.name, cond, label))
# if the edge is from this step, skip adding the implicit edges
if (trigger, cond, label) in step_sources.get(src, set()):
added = True
else:
sources[src].discard(TriggerEdge(trigger, cond, label))
sources[src].discard((trigger, cond, label))
# if no edges from this step, add implicit edges from all previous tasks
if not added:
for src in step_sources:
edges.add(Edge(src, task.name, True, None))
edges.add((src, task.name, True, None))
# assemble the graph
graph = Graph()
# add nodes
for name, node in nodes.items():
metadata = dict(node.metadata or {})
if name in deferred_nodes:
metadata["defer"] = True
if name in interrupt_before_nodes and name in interrupt_after_nodes:
metadata["__interrupt"] = "before,after"
elif name in interrupt_before_nodes:
@@ -247,11 +215,10 @@ def draw_graph(
termini = {d for _, d, _, _ in edges if d != END}.difference(
s for s, _, _, _ in edges
)
end_edge_exists = any(d == END for _, d, _, _ in edges)
if termini:
for src in sorted(termini):
add_edge(graph, src, END)
elif len(step_sources) == 1 and not end_edge_exists:
elif len(step_sources) == 1:
for src in sorted(step_sources):
add_edge(graph, src, END, conditional=True)
# replace subgraphs
+22 -64
View File
@@ -25,17 +25,6 @@ from typing import (
from langchain_core.callbacks import AsyncParentRunManager, ParentRunManager
from langchain_core.runnables import RunnableConfig
from langgraph.cache.base import BaseCache
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
ChannelVersions,
Checkpoint,
CheckpointMetadata,
CheckpointTuple,
PendingWrite,
)
from langgraph.store.base import BaseStore
from typing_extensions import ParamSpec, Self
from langgraph._internal._config import patch_configurable
@@ -61,7 +50,17 @@ from langgraph._internal._constants import (
)
from langgraph._internal._scratchpad import PregelScratchpad
from langgraph._internal._typing import EMPTY_SEQ, MISSING
from langgraph.cache.base import BaseCache
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
ChannelVersions,
Checkpoint,
CheckpointMetadata,
CheckpointTuple,
PendingWrite,
)
from langgraph.constants import TAG_HIDDEN
from langgraph.errors import (
EmptyInputError,
@@ -109,6 +108,7 @@ from langgraph.pregel.debug import (
map_debug_tasks,
)
from langgraph.pregel.protocol import StreamChunk, StreamProtocol
from langgraph.store.base import BaseStore
from langgraph.types import (
All,
CachePolicy,
@@ -568,36 +568,6 @@ class PregelLoop:
if task := tasks.get(tid):
task.writes.append((k, v))
def _pending_interrupts(self) -> set[str]:
"""Return the set of interrupt ids that are pending without corresponding resume values."""
# mapping of task ids to interrupt ids
pending_interrupts: dict[str, str] = {}
# set of resume task ids
pending_resumes: set[str] = set()
for task_id, write_type, value in self.checkpoint_pending_writes:
if write_type == INTERRUPT:
# interrupts is always a list, but there should only be one element
pending_interrupts[task_id] = value[0].id
elif write_type == RESUME:
pending_resumes.add(task_id)
resumed_interrupt_ids = {
pending_interrupts[task_id]
for task_id in pending_resumes
if task_id in pending_interrupts
}
# Keep only interrupts whose interrupt_id is not resumed
hanging_interrupts: set[str] = {
interrupt_id
for interrupt_id in pending_interrupts.values()
if interrupt_id not in resumed_interrupt_ids
}
return hanging_interrupts
def _first(
self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
) -> set[str] | None:
@@ -620,24 +590,16 @@ class PregelLoop:
# map command to writes
if isinstance(self.input, Command):
if (resume := self.input.resume) is not None:
if not self.checkpointer:
raise RuntimeError(
"Cannot use Command(resume=...) without checkpointer"
)
if resume_is_map := (
isinstance(resume, dict)
and all(is_xxh3_128_hexdigest(k) for k in resume)
):
self.config[CONF][CONFIG_KEY_RESUME_MAP] = resume
else:
if len(self._pending_interrupts()) > 1:
raise RuntimeError(
"When there are multiple pending interrupts, you must specify the interrupt id when resuming. "
"Docs: https://docs.langchain.com/oss/python/langgraph/add-human-in-the-loop#resume-multiple-interrupts-with-one-invocation."
)
if resume_is_map := (
(resume := self.input.resume) is not None
and isinstance(resume, dict)
and all(is_xxh3_128_hexdigest(k) for k in resume)
):
self.config[CONF][CONFIG_KEY_RESUME_MAP] = self.input.resume
if resume is not None and not self.checkpointer:
raise RuntimeError(
"Cannot use Command(resume=...) without checkpointer"
)
writes: defaultdict[str, list[tuple[str, Any]]] = defaultdict(list)
# group writes by task ID
for tid, c, v in map_command(cmd=self.input):
@@ -904,11 +866,7 @@ class PregelLoop:
)
}
]
stream_modes = self.stream.modes if self.stream else []
if "updates" in stream_modes:
self._emit("updates", lambda: iter(interrupts))
elif "values" in stream_modes:
self._emit("values", lambda: iter(interrupts))
self._emit("updates", lambda: iter(interrupts))
elif writes[0][0] != ERROR:
self._emit(
"updates",
+4 -8
View File
@@ -231,10 +231,9 @@ class PregelNode:
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> Any:
self_config: RunnableConfig = {"metadata": self.metadata, "tags": self.tags}
return self.bound.invoke(
input,
merge_configs(self_config, config),
merge_configs({"metadata": self.metadata, "tags": self.tags}, config),
**kwargs,
)
@@ -244,10 +243,9 @@ class PregelNode:
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> Any:
self_config: RunnableConfig = {"metadata": self.metadata, "tags": self.tags}
return await self.bound.ainvoke(
input,
merge_configs(self_config, config),
merge_configs({"metadata": self.metadata, "tags": self.tags}, config),
**kwargs,
)
@@ -257,10 +255,9 @@ class PregelNode:
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> Iterator[Any]:
self_config: RunnableConfig = {"metadata": self.metadata, "tags": self.tags}
yield from self.bound.stream(
input,
merge_configs(self_config, config),
merge_configs({"metadata": self.metadata, "tags": self.tags}, config),
**kwargs,
)
@@ -270,10 +267,9 @@ class PregelNode:
config: RunnableConfig | None = None,
**kwargs: Any | None,
) -> AsyncIterator[Any]:
self_config: RunnableConfig = {"metadata": self.metadata, "tags": self.tags}
async for item in self.bound.astream(
input,
merge_configs(self_config, config),
merge_configs({"metadata": self.metadata, "tags": self.tags}, config),
**kwargs,
):
yield item
+1 -1
View File
@@ -7,10 +7,10 @@ import textwrap
from typing import Any, Callable
from langchain_core.runnables import Runnable, RunnableLambda, RunnableSequence
from langgraph.checkpoint.base import ChannelVersions
from typing_extensions import override
from langgraph._internal._runnable import RunnableCallable, RunnableSeq
from langgraph.checkpoint.base import ChannelVersions
from langgraph.pregel.protocol import PregelProtocol
+46 -82
View File
@@ -6,7 +6,6 @@ from typing import Any
from uuid import UUID
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite
from typing_extensions import TypedDict
from langgraph._internal._config import patch_checkpoint_map
@@ -21,6 +20,7 @@ from langgraph._internal._constants import (
)
from langgraph._internal._typing import MISSING
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite
from langgraph.constants import TAG_HIDDEN
from langgraph.pregel._io import read_channels
from langgraph.types import PregelExecutableTask, PregelTask, StateSnapshot
@@ -40,7 +40,7 @@ class TaskResultPayload(TypedDict):
name: str
error: str | None
interrupts: list[dict]
result: dict[str, Any]
result: list[tuple[str, Any]]
class CheckpointTask(TypedDict):
@@ -48,7 +48,7 @@ class CheckpointTask(TypedDict):
name: str
error: str | None
interrupts: list[dict]
state: StateSnapshot | RunnableConfig | None
state: RunnableConfig | None
class CheckpointPayload(TypedDict):
@@ -77,38 +77,6 @@ def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPaylo
}
def is_multiple_channel_write(value: Any) -> bool:
"""Return True if the payload already wraps multiple writes from the same channel."""
return (
isinstance(value, dict)
and "$writes" in value
and isinstance(value["$writes"], list)
)
def map_task_result_writes(writes: Sequence[tuple[str, Any]]) -> dict[str, Any]:
"""Folds task writes into a result dict and aggregates multiple writes to the same channel.
If the channel contains a single write, we record the write in the result dict as `{channel: write}`
If the channel contains multiple writes, we record the writes in the result dict as `{channel: {'$writes': [write1, write2, ...]}}`"""
result: dict[str, Any] = {}
for channel, value in writes:
existing = result.get(channel)
if existing is not None:
channel_writes = (
existing["$writes"]
if is_multiple_channel_write(existing)
else [existing]
)
channel_writes.append(value)
result[channel] = {"$writes": channel_writes}
else:
result[channel] = value
return result
def map_debug_task_results(
task_tup: tuple[PregelExecutableTask, Sequence[tuple[str, Any]]],
stream_keys: str | Sequence[str],
@@ -122,9 +90,7 @@ def map_debug_task_results(
"id": task.id,
"name": task.name,
"error": next((w[1] for w in writes if w[0] == ERROR), None),
"result": map_task_result_writes(
[w for w in writes if w[0] in stream_channels_list or w[0] == RETURN]
),
"result": [w for w in writes if w[0] in stream_channels_list or w[0] == RETURN],
"interrupts": [
asdict(v)
for w in writes
@@ -230,56 +196,54 @@ def tasks_w_writes(
),
MISSING,
)
task_error = next(
(exc for tid, n, exc in pending_writes if tid == task.id and n == ERROR),
None,
)
task_interrupts = tuple(
v
for tid, n, vv in pending_writes
if tid == task.id and n == INTERRUPT
for v in (vv if isinstance(vv, Sequence) else [vv])
)
task_writes = [
(chan, val)
for tid, chan, val in pending_writes
if tid == task.id and chan not in (ERROR, INTERRUPT, RETURN)
]
if rtn is not MISSING:
task_result = rtn
elif isinstance(output_keys, str):
# unwrap single channel writes to just the write value
filtered_writes = [
(chan, val) for chan, val in task_writes if chan == output_keys
]
mapped_writes = map_task_result_writes(filtered_writes)
task_result = mapped_writes.get(str(output_keys)) if mapped_writes else None
else:
if isinstance(output_keys, str):
output_keys = [output_keys]
# map task result writes to the desired output channels
# repeateed writes to the same channel are aggregated into: {'$writes': [write1, write2, ...]}
filtered_writes = [
(chan, val) for chan, val in task_writes if chan in output_keys
]
mapped_writes = map_task_result_writes(filtered_writes)
task_result = mapped_writes if filtered_writes else {}
has_writes = rtn is not MISSING or any(
w[0] == task.id and w[1] not in (ERROR, INTERRUPT) for w in pending_writes
)
out.append(
PregelTask(
task.id,
task.name,
task.path,
task_error,
task_interrupts,
next(
(
exc
for tid, n, exc in pending_writes
if tid == task.id and n == ERROR
),
None,
),
tuple(
v
for tid, n, vv in pending_writes
if tid == task.id and n == INTERRUPT
for v in (vv if isinstance(vv, Sequence) else [vv])
),
states.get(task.id) if states else None,
task_result if has_writes else None,
(
rtn
if rtn is not MISSING
else next(
(
val
for tid, chan, val in pending_writes
if tid == task.id and chan == output_keys
),
None,
)
if isinstance(output_keys, str)
else {
chan: val
for tid, chan, val in pending_writes
if tid == task.id
and (
chan == output_keys
if isinstance(output_keys, str)
else chan in output_keys
)
}
)
if any(
w[0] == task.id and w[1] not in (ERROR, INTERRUPT)
for w in pending_writes
)
else None,
)
)
return tuple(out)
+60 -119
View File
@@ -3,24 +3,15 @@ from __future__ import annotations
import asyncio
import concurrent
import concurrent.futures
import contextlib
import queue
import warnings
import weakref
from collections import defaultdict, deque
from collections.abc import AsyncIterator, Awaitable, Iterator, Mapping, Sequence
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
from dataclasses import is_dataclass
from functools import partial
from inspect import isclass
from typing import (
Any,
Callable,
Generic,
Optional,
Union,
cast,
get_type_hints,
)
from typing import Any, Callable, Generic, Optional, Union, cast, get_type_hints
from uuid import UUID, uuid5
from langchain_core.globals import get_debug
@@ -34,13 +25,6 @@ from langchain_core.runnables.config import (
get_callback_manager_for_config,
)
from langchain_core.runnables.graph import Graph
from langgraph.cache.base import BaseCache
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
CheckpointTuple,
)
from langgraph.store.base import BaseStore
from pydantic import BaseModel, TypeAdapter
from typing_extensions import Self, Unpack, deprecated, is_typeddict
@@ -89,8 +73,14 @@ from langgraph._internal._runnable import (
coerce_to_runnable,
)
from langgraph._internal._typing import MISSING, DeprecatedKwargs
from langgraph.cache.base import BaseCache
from langgraph.channels.base import BaseChannel
from langgraph.channels.topic import Topic
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
CheckpointTuple,
)
from langgraph.config import get_config
from langgraph.constants import END
from langgraph.errors import (
@@ -127,6 +117,7 @@ from langgraph.pregel._write import ChannelWrite, ChannelWriteEntry
from langgraph.pregel.debug import get_bolded_text, get_colored_text, tasks_w_writes
from langgraph.pregel.protocol import PregelProtocol, StreamChunk, StreamProtocol
from langgraph.runtime import DEFAULT_RUNTIME, Runtime
from langgraph.store.base import BaseStore
from langgraph.types import (
All,
CachePolicy,
@@ -386,7 +377,7 @@ class Pregel(
However, for **advanced** use cases, Pregel can be used directly. If you're
not sure whether you need to use Pregel directly, then the answer is probably no
- you should use the Graph API or Functional API instead. These are higher-level
you should use the Graph API or Functional API instead. These are higher-level
interfaces that will compile down to Pregel under the hood.
Here are some examples to give you a sense of how it works:
@@ -488,7 +479,7 @@ class Pregel(
```
```pycon
{"c": ["foofoo", "foofoofoofoo"]}
{'c': ['foofoo', 'foofoofoofoo']}
```
Example: Using a BinaryOperatorAggregate channel
@@ -516,7 +507,6 @@ class Pregel(
else:
return update
app = Pregel(
nodes={"node1": node1, "node2": node2},
channels={
@@ -525,7 +515,7 @@ class Pregel(
"c": BinaryOperatorAggregate(str, operator=reducer),
},
input_channels=["a"],
output_channels=["c"],
output_channels=["c"]
)
app.invoke({"a": "foo"})
@@ -545,8 +535,7 @@ class Pregel(
from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry
example_node = (
NodeBuilder()
.subscribe_only("value")
NodeBuilder().subscribe_only("value")
.do(lambda x: x + x if len(x) < 10 else None)
.write_to(ChannelWriteEntry(channel="value", skip_none=True))
)
@@ -557,7 +546,7 @@ class Pregel(
"value": EphemeralValue(str),
},
input_channels=["value"],
output_channels=["value"],
output_channels=["value"]
)
app.invoke({"value": "a"})
@@ -1695,10 +1684,12 @@ class Pregel(
return patch_checkpoint_map(next_config, saved.metadata)
# task ids can be provided in the StateUpdate, but if not,
# we use the task id generated by prepare_next_tasks
node_to_task_ids: dict[str, deque[str]] = defaultdict(deque)
if saved is not None and saved.pending_writes is not None:
# apply pending writes, if not on specific checkpoint
if (
CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
and saved is not None
and saved.pending_writes
):
# tasks for this checkpoint
next_tasks = prepare_next_tasks(
checkpoint,
@@ -1714,10 +1705,6 @@ class Pregel(
checkpointer=checkpointer,
manager=None,
)
# collect task ids to reuse so we can properly attach task results
for t in next_tasks.values():
node_to_task_ids[t.name].append(t.id)
# apply null writes
if null_writes := [
w[1:] for w in saved.pending_writes or [] if w[0] == NULL_TASK_ID
@@ -1799,14 +1786,8 @@ class Pregel(
raise InvalidUpdateError(f"Node {as_node} has no writers")
writes: deque[tuple[str, Any]] = deque()
task = PregelTaskWrites((), as_node, writes, [INTERRUPT])
# get the task ids that were prepared for this node
# if a task id was provided in the StateUpdate, we use it
# otherwise, we use the next available task id
prepared_task_ids = node_to_task_ids.get(as_node, deque())
task_id = provided_task_id or (
prepared_task_ids.popleft()
if prepared_task_ids
else str(uuid5(UUID(checkpoint["id"]), INTERRUPT))
task_id = provided_task_id or str(
uuid5(UUID(checkpoint["id"]), INTERRUPT)
)
run_tasks.append(task)
run_task_ids.append(task_id)
@@ -2159,11 +2140,12 @@ class Pregel(
return patch_checkpoint_map(
next_config, saved.metadata if saved else None
)
# task ids can be provided in the StateUpdate, but if not,
# we use the task id generated by prepare_next_tasks
node_to_task_ids: dict[str, deque[str]] = defaultdict(deque)
if saved is not None and saved.pending_writes is not None:
# apply pending writes, if not on specific checkpoint
if (
CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
and saved is not None
and saved.pending_writes
):
# tasks for this checkpoint
next_tasks = prepare_next_tasks(
checkpoint,
@@ -2179,10 +2161,6 @@ class Pregel(
checkpointer=checkpointer,
manager=None,
)
# collect task ids to reuse so we can properly attach task results
for t in next_tasks.values():
node_to_task_ids[t.name].append(t.id)
# apply null writes
if null_writes := [
w[1:] for w in saved.pending_writes or [] if w[0] == NULL_TASK_ID
@@ -2259,14 +2237,8 @@ class Pregel(
raise InvalidUpdateError(f"Node {as_node} has no writers")
writes: deque[tuple[str, Any]] = deque()
task = PregelTaskWrites((), as_node, writes, [INTERRUPT])
# get the task ids that were prepared for this node
# if a task id was provided in the StateUpdate, we use it
# otherwise, we use the next available task id
prepared_task_ids = node_to_task_ids.get(as_node, deque())
task_id = provided_task_id or (
prepared_task_ids.popleft()
if prepared_task_ids
else str(uuid5(UUID(checkpoint["id"]), INTERRUPT))
task_id = provided_task_id or str(
uuid5(UUID(checkpoint["id"]), INTERRUPT)
)
run_tasks.append(task)
run_task_ids.append(task_id)
@@ -2462,7 +2434,7 @@ class Pregel(
input: The input to the graph.
config: The configuration to use for the run.
context: The static context to use for the run.
!!! version-added "Added in version 0.6.0"
!!! version-added "Added in version 0.6.0."
stream_mode: The mode to stream output, defaults to `self.stream_mode`.
Options are:
@@ -2640,7 +2612,6 @@ class Pregel(
if subgraphs:
loop.config[CONF][CONFIG_KEY_STREAM] = loop.stream
# enable concurrent streaming
get_waiter: Callable[[], concurrent.futures.Future[None]] | None = None
if (
self.stream_eager
or subgraphs
@@ -2663,6 +2634,8 @@ class Pregel(
else:
return waiter
else:
get_waiter = None # type: ignore[assignment]
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates.
# Channel updates from step N are only visible in step N+1
@@ -2728,7 +2701,7 @@ class Pregel(
input: The input to the graph.
config: The configuration to use for the run.
context: The static context to use for the run.
!!! version-added "Added in version 0.6.0"
!!! version-added "Added in version 0.6.0."
stream_mode: The mode to stream output, defaults to `self.stream_mode`.
Options are:
@@ -2943,77 +2916,45 @@ class Pregel(
stream_put, stream_modes
)
# enable concurrent streaming
get_waiter: Callable[[], asyncio.Task[None]] | None = None
_cleanup_waiter: Callable[[], Awaitable[None]] | None = None
if (
self.stream_eager
or subgraphs
or "messages" in stream_modes
or "custom" in stream_modes
):
# Keep a single waiter task alive; ensure cleanup on exit.
waiter: asyncio.Task[None] | None = None
def get_waiter() -> asyncio.Task[None]:
nonlocal waiter
if waiter is None or waiter.done():
waiter = aioloop.create_task(stream.wait())
def _clear(t: asyncio.Task[None]) -> None:
nonlocal waiter
if waiter is t:
waiter = None
waiter.add_done_callback(_clear)
return waiter
async def _cleanup_waiter() -> None:
"""Wake pending waiter and/or cancel+await to avoid pending tasks."""
nonlocal waiter
# Try to wake via semaphore like SyncPregelLoop
with contextlib.suppress(Exception):
if hasattr(stream, "_count"):
stream._count.release()
t = waiter
waiter = None
if t is not None and not t.done():
t.cancel()
with contextlib.suppress(asyncio.CancelledError):
await t
return aioloop.create_task(stream.wait())
else:
get_waiter = None # type: ignore[assignment]
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates
# channel updates from step N are only visible in step N+1
# channels are guaranteed to be immutable for the duration of the step,
# with channel updates applied only at the transition between steps
try:
while loop.tick():
for task in await loop.amatch_cached_writes():
loop.output_writes(task.id, task.writes, cached=True)
async for _ in runner.atick(
[t for t in loop.tasks.values() if not t.writes],
timeout=self.step_timeout,
get_waiter=get_waiter,
schedule_task=loop.aaccept_push,
while loop.tick():
for task in await loop.amatch_cached_writes():
loop.output_writes(task.id, task.writes, cached=True)
async for _ in runner.atick(
[t for t in loop.tasks.values() if not t.writes],
timeout=self.step_timeout,
get_waiter=get_waiter,
schedule_task=loop.aaccept_push,
):
# emit output
for o in _output(
stream_mode,
print_mode,
subgraphs,
stream.get_nowait,
asyncio.QueueEmpty,
):
# emit output
for o in _output(
stream_mode,
print_mode,
subgraphs,
stream.get_nowait,
asyncio.QueueEmpty,
):
yield o
loop.after_tick()
# wait for checkpoint
if durability_ == "sync":
await cast(asyncio.Future, loop._put_checkpoint_fut)
finally:
# ensure waiter doesn't remain pending on cancel/shutdown
if _cleanup_waiter is not None:
await _cleanup_waiter()
yield o
loop.after_tick()
# wait for checkpoint
if durability_ == "sync":
await cast(asyncio.Future, loop._put_checkpoint_fut)
# emit output
for o in _output(
stream_mode,
@@ -3060,7 +3001,7 @@ class Pregel(
input: The input data for the graph. It can be a dictionary or any other type.
config: Optional. The configuration for the graph run.
context: The static context to use for the run.
!!! version-added "Added in version 0.6.0"
!!! version-added "Added in version 0.6.0."
stream_mode: Optional[str]. The stream mode for the graph run. Default is "values".
print_mode: Accepts the same values as `stream_mode`, but only prints the output to the console, for debugging purposes. Does not affect the output of the graph in any way.
output_keys: Optional. The output keys to retrieve from the graph run.
@@ -3145,7 +3086,7 @@ class Pregel(
input: The input data for the computation. It can be a dictionary or any other type.
config: Optional. The configuration for the computation.
context: The static context to use for the run.
!!! version-added "Added in version 0.6.0"
!!! version-added "Added in version 0.6.0."
stream_mode: Optional. The stream mode for the computation. Default is "values".
print_mode: Accepts the same values as `stream_mode`, but only prints the output to the console, for debugging purposes. Does not affect the output of the graph in any way.
output_keys: Optional. The output keys to include in the result. Default is None.
+2 -8
View File
@@ -1,6 +1,5 @@
from __future__ import annotations
import logging
from collections.abc import AsyncIterator, Iterator, Sequence
from dataclasses import asdict
from typing import (
@@ -8,7 +7,6 @@ from typing import (
Literal,
cast,
)
from uuid import UUID
import langsmith as ls
from langchain_core.runnables import RunnableConfig
@@ -21,7 +19,6 @@ from langchain_core.runnables.graph import (
from langchain_core.runnables.graph import (
Node as DrawableNode,
)
from langgraph.checkpoint.base import CheckpointMetadata
from langgraph_sdk.client import (
LangGraphClient,
SyncLangGraphClient,
@@ -52,6 +49,7 @@ from langgraph._internal._constants import (
INTERRUPT,
NS_SEP,
)
from langgraph.checkpoint.base import CheckpointMetadata
from langgraph.errors import GraphInterrupt, ParentCommand
from langgraph.pregel.protocol import PregelProtocol, StreamProtocol
from langgraph.types import (
@@ -63,8 +61,6 @@ from langgraph.types import (
StreamMode,
)
logger = logging.getLogger(__name__)
__all__ = ("RemoteGraph", "RemoteException")
_CONF_DROPLIST = frozenset(
@@ -79,7 +75,7 @@ _CONF_DROPLIST = frozenset(
def _sanitize_config_value(v: Any) -> Any:
"""Recursively sanitize a config value to ensure it contains only primitives."""
if isinstance(v, (str, int, float, bool, UUID)):
if isinstance(v, (str, int, float, bool)):
return v
elif isinstance(v, dict):
sanitized_dict = {}
@@ -954,7 +950,6 @@ class RemoteGraph(PregelProtocol):
try:
return chunk
except UnboundLocalError:
logger.warning("No events received from remote graph")
return None
async def ainvoke(
@@ -995,7 +990,6 @@ class RemoteGraph(PregelProtocol):
try:
return chunk
except UnboundLocalError:
logger.warning("No events received from remote graph")
return None

Some files were not shown because too many files have changed in this diff Show More