Compare commits

..
Author SHA1 Message Date
Eugene Yurtsev 2eb3b316c9 x 2025-08-05 14:06:48 -04:00
Eugene Yurtsev 1f11173c30 x 2025-08-05 14:00:55 -04:00
Eugene Yurtsev 517031d5ab x 2025-08-05 13:59:27 -04:00
Eugene Yurtsev 0ce981a0e6 Test with multiple interrupts 2025-08-05 13:57:06 -04:00
Eugene Yurtsev 2bcb423afa x 2025-08-05 13:50:43 -04:00
215 changed files with 5174 additions and 16705 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.
+2 -7
View File
@@ -1,15 +1,10 @@
import ast
import os
from itertools import filterfalse
from typing import Dict, List, Tuple
from typing import List, Tuple
ROOT_PATH = os.path.abspath(os.path.join(__file__, "..", "..", ".."))
CLIENT_PATH = os.path.join(ROOT_PATH, "libs", "sdk-py", "langgraph_sdk", "client.py")
ASYNC_TO_SYNC_METHOD_MAP: Dict[str, str] = {
"aclose": "close",
"__aenter__": "__enter__",
"__aexit__": "__exit__",
}
def get_class_methods(node: ast.ClassDef) -> List[str]:
@@ -27,7 +22,7 @@ def find_classes(tree: ast.AST) -> List[Tuple[str, List[str]]]:
def compare_sync_async_methods(sync_methods: List[str], async_methods: List[str]) -> List[str]:
sync_set = set(sync_methods)
async_set = {ASYNC_TO_SYNC_METHOD_MAP.get(async_method, async_method) for async_method in async_methods}
async_set = set(async_methods)
missing_in_sync = list(async_set - sync_set)
missing_in_async = list(sync_set - async_set)
return missing_in_sync + missing_in_async
+86 -124
View File
@@ -1,145 +1,107 @@
import asyncio
import json
import os
import pathlib
import sys
import time
from urllib import request, error
import langgraph_cli
import langgraph_cli.config
import langgraph_cli.docker
from langgraph_cli.cli import prepare_args_and_stdin
from langgraph_cli.constants import DEFAULT_PORT
import langgraph_cli.config
from langgraph_cli.exec import Runner, subp_exec
from langgraph_cli.progress import Progress
from langgraph_cli.constants import DEFAULT_PORT
def test(config: pathlib.Path, port: int, tag: str, verbose: bool):
"""Spin up API with Postgres/Redis via docker compose and wait until ready."""
def test(
config: pathlib.Path,
port: int,
tag: str,
verbose: bool,
):
with Runner() as runner, Progress(message="Pulling...") as set:
# Detect docker/compose capabilities
# check docker available
capabilities = langgraph_cli.docker.check_capabilities(runner)
# Validate config and prepare compose stdin/args using built image
# open config
config_json = langgraph_cli.config.validate_config_file(config)
args, stdin = prepare_args_and_stdin(
capabilities=capabilities,
config_path=config,
config=config_json,
docker_compose=None,
port=port,
watch=False,
debugger_port=None,
debugger_base_url=f"http://127.0.0.1:{port}",
postgres_uri=None,
api_version=None,
image=tag,
base_image=None,
)
# Compose up with wait (implies detach), similar to `langgraph up --wait`
args_up = [*args, "up", "--remove-orphans", "--wait"]
set("Running...")
args = [
"run",
"--rm",
"-p",
f"{port}:8000",
]
if isinstance(config_json["env"], str):
args.extend(
[
"--env-file",
str(config.parent / config_json["env"]),
]
)
else:
for k, v in config_json["env"].items():
args.extend(
[
"-e",
f"{k}={v}",
]
)
if capabilities.healthcheck_start_interval:
args.extend(
[
"--health-interval",
"5s",
"--health-retries",
"1",
"--health-start-period",
"10s",
"--health-start-interval",
"1s",
]
)
else:
args.extend(
[
"--health-interval",
"5s",
"--health-retries",
"2",
]
)
compose_cmd = ["docker", "compose"]
if capabilities.compose_type == "standalone":
compose_cmd = ["docker-compose"]
_task = None
def on_stdout(line: str):
nonlocal _task
if "GET /ok" in line or "Uvicorn running on" in line:
set("")
sys.stdout.write(
f"""Ready!
- API: http://localhost:{port}
"""
)
sys.stdout.flush()
_task.cancel()
return True
return False
async def subp_exec_task(*args, **kwargs):
nonlocal _task
_task = asyncio.create_task(subp_exec(*args, **kwargs))
await _task
set("Starting...")
try:
runner.run(
subp_exec(
*compose_cmd,
*args_up,
input=stdin,
subp_exec_task(
"docker",
*args,
tag,
verbose=verbose,
on_stdout=on_stdout,
)
)
except Exception as e: # noqa: BLE001
# On failure, show diagnostics then ensure clean teardown
sys.stderr.write(f"docker compose up failed: {e}\n")
try:
sys.stderr.write("\n== docker compose ps ==\n")
runner.run(subp_exec(*compose_cmd, *args, "ps", input=stdin, verbose=False))
except Exception:
pass
try:
sys.stderr.write("\n== docker compose logs (api) ==\n")
runner.run(
subp_exec(
*compose_cmd,
*args,
"logs",
"langgraph-api",
input=stdin,
verbose=False,
)
)
except Exception:
pass
finally:
try:
runner.run(
subp_exec(
*compose_cmd,
*args,
"down",
"-v",
"--remove-orphans",
input=stdin,
verbose=False,
)
)
finally:
raise
set("")
base_url = f"http://localhost:{port}"
ok_url = f"{base_url}/ok"
print(f"Waiting for {ok_url} to respond with 200...")
deadline = time.time() + 30
last_err: Exception | None = None
while time.time() < deadline:
try:
with request.urlopen(ok_url, timeout=2) as resp:
if resp.status == 200:
sys.stdout.write(
f"""Ready!\n- API: {base_url}\n- /ok: 200 OK\n"""
)
sys.stdout.flush()
break
else:
last_err = RuntimeError(f"Unexpected status: {resp.status}")
print(f"Unexpected status: {resp.status}")
except error.URLError as e:
last_err = e
except Exception as e: # noqa: BLE001
last_err = e
time.sleep(0.5)
else:
# Bring stack down before raising
args_down = [*args, "down", "-v", "--remove-orphans"]
try:
runner.run(
subp_exec(
*compose_cmd,
*args_down,
input=stdin,
verbose=verbose,
)
)
finally:
raise SystemExit(
f"/ok did not return 202 within timeout. Last error: {last_err}"
)
# Clean up: bring compose stack down to free ports for next test
args_down = [*args, "down", "-v", "--remove-orphans"]
runner.run(
subp_exec(
*compose_cmd,
*args_down,
input=stdin,
verbose=verbose,
)
)
except asyncio.CancelledError:
pass
if __name__ == "__main__":
@@ -148,6 +110,6 @@ if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("-t", "--tag", type=str)
parser.add_argument("-c", "--config", type=str, default="./langgraph.json")
parser.add_argument("-p", "--port", type=int, default=DEFAULT_PORT)
parser.add_argument("-p", "--port", default=DEFAULT_PORT)
args = parser.parse_args()
test(pathlib.Path(args.config), args.port, args.tag, verbose=True)
+29 -65
View File
@@ -14,25 +14,12 @@ 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:
working-directory: libs/cli
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Get changed files
id: changed-files
uses: Ana06/get-changed-files@v2.3.0
@@ -46,65 +33,42 @@ 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 }}
env:
LANGSMITH_API_KEY: ${{ secrets.LANGSMITH_API_KEY }}
working-directory: libs/cli/examples
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 --base-image "langchain/langgraph-trial"
cp .env.example .envg
timeout 60 python ../../../.github/scripts/run_langgraph_cli_test.py -c langgraph.json -t langgraph-test-a
- name: Build and test service B
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graphs
run: |
langgraph build -t langgraph-test-b --base-image "langchain/langgraph-trial"
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-b
- name: Build and test service C
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graphs_reqs_a
run: |
langgraph build -t langgraph-test-c --base-image "langchain/langgraph-trial"
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-c
- name: Build and test service D
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graphs_reqs_b
run: |
langgraph build -t langgraph-test-d --base-image "langchain/langgraph-trial"
timeout 60 python ../../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-d
- name: Build JS service
if: steps.changed-files.outputs.all
working-directory: libs/cli/js-examples
run: |
langgraph build -t langgraph-test-e
- name: Build JS monorepo service
if: steps.changed-files.outputs.all
working-directory: libs/cli/js-monorepo-example
run: |
langgraph build -t langgraph-test-f -c apps/agent/langgraph.json --build-command "yarn run turbo build" --install-command "yarn install"
- name: Build Python monorepo service
if: steps.changed-files.outputs.all
working-directory: libs/cli/python-monorepo-example
run: |
langgraph build -t langgraph-test-g -c apps/agent/langgraph.json
cp apps/agent/.env.example apps/agent/.env
if [ -n "${{ secrets.LANGSMITH_API_KEY }}" ]; then echo "LANGSMITH_API_KEY=${{ secrets.LANGSMITH_API_KEY }}" >> apps/agent/.env; fi
timeout 60 python ../../../.github/scripts/run_langgraph_cli_test.py -t langgraph-test-g -c apps/agent/langgraph.json
- name: Build and test prerelease reqs service
if: steps.changed-files.outputs.all
working-directory: libs/cli/examples/graph_prerelease_reqs
run: |
langgraph build -t langgraph-test-h
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 -1
View File
@@ -31,7 +31,7 @@ jobs:
- "3.12"
name: "lint #${{ matrix.python-version }}"
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Get changed files
id: changed-files
uses: Ana06/get-changed-files@v2.3.0
+1 -1
View File
@@ -25,7 +25,7 @@ jobs:
name: "test #${{ matrix.python-version }}"
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: astral-sh/setup-uv@v6
with:
+1 -1
View File
@@ -23,7 +23,7 @@ jobs:
working-directory: libs/langgraph
name: "test #${{ matrix.python-version }}"
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: astral-sh/setup-uv@v6
with:
+3 -3
View File
@@ -24,7 +24,7 @@ jobs:
version: ${{ steps.check-version.outputs.version }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python $${ env.PYTHON_VERSION }}
uses: astral-sh/setup-uv@v6
@@ -75,9 +75,9 @@ jobs:
id-token: write
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- uses: actions/download-artifact@v5
- uses: actions/download-artifact@v4
with:
name: test-dist
path: ${{ inputs.working-directory }}/dist/
+1 -1
View File
@@ -17,7 +17,7 @@ jobs:
run:
working-directory: libs/langgraph
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- run: SHA=$(git rev-parse HEAD) && echo "SHA=$SHA" >> $GITHUB_ENV
- name: Set up Python 3.11
uses: astral-sh/setup-uv@v6
+2 -2
View File
@@ -15,7 +15,7 @@ jobs:
run:
working-directory: libs/langgraph
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- id: files
name: Get changed files
uses: Ana06/get-changed-files@v2.3.0
@@ -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]
+4 -5
View File
@@ -27,7 +27,7 @@ jobs:
python: ${{ steps.filter.outputs.python }}
deps: ${{ steps.filter.outputs.deps }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- uses: dorny/paths-filter@v3
id: filter
with:
@@ -78,7 +78,6 @@ jobs:
"libs/checkpoint-sqlite",
"libs/checkpoint-postgres",
"libs/prebuilt",
"libs/sdk-py",
]
if: needs.changes.outputs.python == 'true' || needs.changes.outputs.deps == 'true'
uses: ./.github/workflows/_test.yml
@@ -100,9 +99,9 @@ jobs:
name: "Check SDK methods matching"
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v6
uses: actions/setup-python@v5
with:
python-version: "3.11"
- name: Run check_sdk_methods script
@@ -118,7 +117,7 @@ jobs:
python-version:
- "3.11"
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: astral-sh/setup-uv@v6
with:
+1 -1
View File
@@ -21,7 +21,7 @@
steps:
- name: Checkout
uses: actions/checkout@v5
uses: actions/checkout@v4
- name: Install Dependencies
run: |
+3 -3
View File
@@ -28,7 +28,7 @@ jobs:
outputs:
changed-files: ${{ steps.changed-files.outputs.added_modified }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Get changed files
id: changed-files
uses: Ana06/get-changed-files@v2.3.0
@@ -41,7 +41,7 @@ jobs:
env:
GITHUB_TOKEN: ${{ secrets.MKDOCS_GITHUB_TOKEN }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
with:
fetch-depth: 0
@@ -140,7 +140,7 @@ jobs:
- name: Upload Pages Artifact
# if: github.ref == 'refs/heads/main'
uses: actions/upload-pages-artifact@v4
uses: actions/upload-pages-artifact@v3
with:
path: ./docs/site/
+2 -2
View File
@@ -19,7 +19,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v5
uses: actions/checkout@v4
with:
fetch-depth: 0
@@ -36,7 +36,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v5
uses: actions/checkout@v4
with:
fetch-depth: 1
+1 -2
View File
@@ -12,7 +12,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Validate PR Title
uses: amannn/action-semantic-pull-request@v6
uses: amannn/action-semantic-pull-request@v5
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
with:
@@ -40,7 +40,6 @@ jobs:
sdk-py
docs
ci
deps
requireScope: false
ignoreLabels: |
ignore-lint-pr-title
+8 -14
View File
@@ -26,7 +26,7 @@ jobs:
tag: ${{ steps.check-version.outputs.tag }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python
uses: astral-sh/setup-uv@v6
@@ -62,13 +62,7 @@ jobs:
working-directory: ${{ inputs.working-directory }}
run: |
PKG_NAME=$(grep -m 1 "^name = " pyproject.toml | cut -d '"' -f 2)
if grep -q 'dynamic.*=.*\[.*"version".*\]' pyproject.toml; then
# handle dynamic versioning
DIR_NAME=$(echo "$PKG_NAME" | tr '-' '_')
VERSION=$(grep -m 1 '^__version__' "${DIR_NAME}/__init__.py" | cut -d '"' -f 2)
else
VERSION=$(grep -m 1 "^version = " pyproject.toml | cut -d '"' -f 2)
fi
VERSION=$(grep -m 1 "^version = " pyproject.toml | cut -d '"' -f 2)
SHORT_PKG_NAME="$(echo "$PKG_NAME" | sed -e 's/langgraph//g' -e 's/-//g')"
if [ -z $SHORT_PKG_NAME ]; then
TAG="$VERSION"
@@ -87,7 +81,7 @@ jobs:
outputs:
release-body: ${{ steps.generate-release-body.outputs.release-body }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
with:
repository: langchain-ai/langgraph
path: langgraph
@@ -158,7 +152,7 @@ jobs:
- test-pypi-publish
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
# We explicitly *don't* set up caching here. This ensures our tests are
# maximally sensitive to catching breakage.
@@ -261,7 +255,7 @@ jobs:
working-directory: ${{ inputs.working-directory }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python
uses: astral-sh/setup-uv@v6
@@ -270,7 +264,7 @@ jobs:
enable-cache: true
cache-suffix: "release"
- uses: actions/download-artifact@v5
- uses: actions/download-artifact@v4
with:
name: dist
path: ${{ inputs.working-directory }}/dist/
@@ -302,7 +296,7 @@ jobs:
working-directory: ${{ inputs.working-directory }}
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python
uses: astral-sh/setup-uv@v6
@@ -311,7 +305,7 @@ jobs:
enable-cache: true
cache-suffix: "release"
- uses: actions/download-artifact@v5
- uses: actions/download-artifact@v4
with:
name: dist
path: ${{ inputs.working-directory }}/dist/
+1 -1
View File
@@ -28,7 +28,7 @@ jobs:
- "latest"
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up Python + Poetry
uses: astral-sh/setup-uv@v6
with:
+3 -3
View File
@@ -16,7 +16,7 @@ jobs:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
- uses: actions/checkout@v4
- name: Set up uv
uses: astral-sh/setup-uv@v6
@@ -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`.
+1 -1
View File
@@ -71,7 +71,7 @@ While LangGraph can be used standalone, it also integrates seamlessly with any L
## Additional resources
- [Guides](https://langchain-ai.github.io/langgraph/guides/): Quick, actionable code snippets for topics such as streaming, adding memory & persistence, and design patterns (e.g. branching, subgraphs, etc.).
- [Guides](https://langchain-ai.github.io/langgraph/how-tos/): Quick, actionable code snippets for topics such as streaming, adding memory & persistence, and design patterns (e.g. branching, subgraphs, etc.).
- [Reference](https://langchain-ai.github.io/langgraph/reference/graphs/): Detailed reference on core classes, methods, how to use the graph and checkpointing APIs, and higher-level prebuilt components.
- [Examples](https://langchain-ai.github.io/langgraph/examples/): Guided examples on getting started with LangGraph.
- [LangChain Forum](https://forum.langchain.com/): Connect with the community and share all of your technical questions, ideas, and feedback.
+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.8.9
resolution: "hono@npm:4.8.9"
checksum: 10c0/385539d1787fdc747bc869ef0e5ccc9f39cbe40289b94f23eecfc82c6ca440f059704647cd6381a5066d2cf7baa43ab25184c78d44af4c5c98a5c5b07670059e
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
+2 -2
View File
@@ -190,11 +190,11 @@ REDIRECT_MAP = {
"concepts/langgraph_cloud.md": "https://docs.langchain.com/langgraph-platform/cloud",
"concepts/langgraph_self_hosted_data_plane.md": "https://docs.langchain.com/langgraph-platform/hybrid",
"concepts/langgraph_self_hosted_control_plane.md": "https://docs.langchain.com/langgraph-platform/self-hosted",
"concepts/langgraph_standalone_container.md": "https://docs.langchain.com/langgraph-platform/self-hosted#standalone-server",
"concepts/langgraph_standalone_container.md": "https://docs.langchain.com/langgraph-platform/self-hosted#data-plane-only",
"cloud/deployment/cloud.md": "https://docs.langchain.com/langgraph-platform/cloud",
"cloud/deployment/self_hosted_data_plane.md": "https://docs.langchain.com/langgraph-platform/deploy-hybrid",
"cloud/deployment/self_hosted_control_plane.md": "https://docs.langchain.com/langgraph-platform/deploy-self-hosted-full-platform",
"cloud/deployment/standalone_container.md": "https://docs.langchain.com/langgraph-platform/deploy-standalone-server",
"cloud/deployment/standalone_container.md": "https://docs.langchain.com/langgraph-platform/deploy-data-plane-only",
"concepts/server-mcp.md": "https://docs.langchain.com/langgraph-platform/server-mcp",
"cloud/how-tos/human_in_the_loop_time_travel.md": "https://docs.langchain.com/langgraph-platform/human-in-the-loop-time-travel",
"cloud/how-tos/add-human-in-the-loop.md": "https://docs.langchain.com/langgraph-platform/add-human-in-the-loop",
+1 -1
View File
@@ -90,7 +90,7 @@ graph.invoke( # (1)!
from langgraph.runtime import Runtime
# highlight-next-line
def node(state: State, runtime: Runtime[ContextSchema]):
def node(state: State, config: Runtime[ContextSchema]):
user_name = runtime.context.user_name
...
```
+8 -9
View File
@@ -367,13 +367,13 @@ To implement handoffs with `createReactAgent`, you need to:
3. Define a parent graph that contains individual agents as nodes:
```typescript
import { StateGraph, MessagesZodState } from "@langchain/langgraph";
const multiAgentGraph = new StateGraph(MessagesZodState)
.addNode("flight_assistant", flightAssistant)
.addNode("hotel_assistant", hotelAssistant)
// ...
```
```typescript
import { StateGraph, MessagesZodState } from "@langchain/langgraph";
const multiAgentGraph = new StateGraph(MessagesZodState)
.addNode("flight_assistant", flightAssistant)
.addNode("hotel_assistant", hotelAssistant)
// ...
```
:::
@@ -619,8 +619,7 @@ for await (const chunk of multiAgentGraph.stream({
3. Name of the agent or node to hand off to.
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
5. Indicate to LangGraph that we need to navigate to agent node in a **parent** multi-agent graph.
:::
:::
!!! Note
+2 -2
View File
@@ -99,8 +99,8 @@ Starting from the `LangGraph Platform` view...
1. In the top-right corner, select the gear icon (`Deployment Settings`).
1. Update the `Git Branch` to the desired branch.
1. Check/uncheck checkbox to `Automatically update deployment on push to branch`.
1. Branch creation/deletion and tag creation/deletion events will not trigger an update. Only pushes to an existing branch will trigger an update.
1. Pushes in quick succession to a branch will queue subsequent updates. Once a build completes, the most recent commit will begin building and the other queued builds will be skipped.
1. Branch creation/deletion and tag creation/deletion events will not trigger an update. Only pushes to an existing branch will trigger an update.
1. Pushes in quick succession to a branch will not trigger subsequent updates. In the future, this functionality may be changed/improved.
## Add or Remove GitHub Repositories
File diff suppressed because it is too large Load Diff
+3 -3
View File
@@ -483,19 +483,19 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
RUN PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 pip install --no-cache-dir -c /api/constraints.txt langchain_community langchain_anthropic langchain_openai wikipedia scikit-learn
ADD ./graphs /deps/outer-graphs/src
ADD ./graphs /deps/__outer_graphs/src
RUN set -ex && \
for line in '[project]' \
'name = "graphs"' \
'version = "0.1"' \
'[tool.setuptools.package-data]' \
'"*" = ["**/*"]'; do \
echo "$line" >> /deps/outer-graphs/pyproject.toml; \
echo "$line" >> /deps/__outer_graphs/pyproject.toml; \
done
RUN PIP_CONFIG_FILE=/pipconfig.txt PYTHONDONTWRITEBYTECODE=1 pip install --no-cache-dir -c /api/constraints.txt -e /deps/*
ENV LANGSERVE_GRAPHS='{"agent": "/deps/outer-graphs/src/agent.py:graph", "storm": "/deps/outer-graphs/src/storm.py:graph"}'
ENV LANGSERVE_GRAPHS='{"agent": "/deps/__outer_graphs/src/agent.py:graph", "storm": "/deps/__outer_graphs/src/storm.py:graph"}'
```
???+ note "Updating your langgraph.json file"
+1 -1
View File
@@ -1040,7 +1040,7 @@ def node_a(state: State, runtime: Runtime[ContextSchema]):
...
```
See [this guide](../how-tos/graph-api.md#add-runtime-configuration) for a full breakdown on configuration.
See [this guide](../how-tos/graph-api.ipynb#add-runtime-configuration) for a full breakdown on configuration.
:::
:::js
-7
View File
@@ -6,14 +6,7 @@
Install the `langchain-mcp-adapters` library to use MCP tools in LangGraph:
:::python
```bash
pip install langchain-mcp-adapters
```
:::
:::js
```bash
npm install @langchain/mcp-adapters
```
:::
+2 -2
View File
@@ -134,7 +134,7 @@ def update_instructions(state: State, store: BaseStore):
namespace = ("instructions",)
current_instructions = store.search(namespace)[0]
# Memory logic
prompt = prompt_template.format(instructions=current_instructions.value["instructions"], conversation=state["messages"])
prompt = prompt_template.format(instructions=instructions.value["instructions"], conversation=state["messages"])
output = llm.invoke(prompt)
new_instructions = output['new_instructions']
store.put(("agent_instructions",), "agent_a", {"instructions": new_instructions})
@@ -278,4 +278,4 @@ const items = await store.search(
```
:::
For more information about the memory store, see the [Persistence](persistence.md#memory-store) guide.
For more information about the memory store, see the [Persistence](persistence.md#memory-store) guide.
+2 -2
View File
@@ -897,5 +897,5 @@ There are two high-level approaches to achieve that:
An agent might need to have a different state schema from the rest of the agents. For example, a search agent might only need to keep track of queries and retrieved documents. There are two ways to achieve this in LangGraph:
- Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, it's important to [add input / output transformations](../how-tos/subgraph.md#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
- Define agent node functions with a [private input state schema](../how-tos/graph-api.md#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
- Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, it's important to [add input / output transformations](../how-tos/subgraph.ipynb#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
+1 -2
View File
@@ -315,8 +315,7 @@ In our example, the output of `get_state_history` will look like this:
tasks=(),
),
StateSnapshot(
values={'foo': 'a', 'bar': ['a']},
next=('node_b',),
values={'foo': 'a', 'bar': ['a']}, next=('node_b',),
config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef663ba-28f9-6ec4-8001-31981c2c39f8'}},
metadata={'source': 'loop', 'writes': {'node_a': {'foo': 'a', 'bar': ['a']}}, 'step': 1},
created_at='2024-08-29T19:19:38.819946+00:00',
+1 -1
View File
@@ -12,7 +12,7 @@ There are three different plans for using it.
- **Developer**: All [LangSmith](https://smith.langchain.com/) users have access to this plan. You can sign up for this plan simply by creating a LangSmith account. This gives you access to the [local deployment](./deployment_options.md#free-deployment) option.
- **Plus**: All [LangSmith](https://smith.langchain.com/) users with a [Plus account](https://docs.smith.langchain.com/administration/pricing) have access to this plan. You can sign up for this plan simply by upgrading your LangSmith account to the Plus plan type. This gives you access to the [Cloud](./deployment_options.md#cloud-saas) deployment option.
- **Enterprise**: This is separate from LangSmith plans. You can sign up for this plan by [contacting our sales team](https://www.langchain.com/contact-sales). This gives you access to all [deployment options](./deployment_options.md).
- **Enterprise**: This is separate from LangSmith plans. You can sign up for this plan by contacting sales@langchain.dev. This gives you access to all [deployment options](./deployment_options.md).
## Plan Details
File diff suppressed because it is too large Load Diff
@@ -366,8 +366,8 @@ result = graph.invoke(
# Resume with mapping of interrupt IDs to values
resume_map = {
i.id: f"edited text for {i.value['text_to_revise']}"
for i in graph.get_state(config).interrupts
i.interrupt_id: f"human input for prompt {i.value}"
for i in parent.get_state(thread_config).interrupts
}
print(graph.invoke(Command(resume=resume_map), config=config))
# > {'text_1': 'edited text for original text 1', 'text_2': 'edited text for original text 2'}
+11 -684
View File
@@ -22,7 +22,6 @@ To set up communication between the agents in a multi-agent system you can use [
To implement handoffs, you can return `Command` objects from your agent nodes or tools:
:::python
```python
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
@@ -74,109 +73,25 @@ def create_handoff_tool(*, agent_name: str, description: str | None = None):
commands = [tools_by_name[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
return commands
```
:::
:::js
```typescript
import { tool } from "@langchain/core/tools";
import { Command, MessagesZodState } from "@langchain/langgraph";
import { z } from "zod";
function createHandoffTool({
agentName,
description,
}: {
agentName: string;
description?: string;
}) {
const name = `transfer_to_${agentName}`;
const toolDescription = description || `Transfer to ${agentName}`;
return tool(
async (_, config) => {
// (1)!
const state = config.state;
const toolCallId = config.toolCall.id;
const toolMessage = {
role: "tool" as const,
content: `Successfully transferred to ${agentName}`,
name: name,
tool_call_id: toolCallId,
};
return new Command({
// (3)!
goto: agentName,
// (4)!
update: { messages: [...state.messages, toolMessage] },
// (5)!
graph: Command.PARENT,
});
},
{
name,
description: toolDescription,
schema: z.object({}),
}
);
}
```
1. Access the [state](../concepts/low_level.md#state) of the agent that is calling the handoff tool through the `config` parameter.
2. The `Command` primitive allows specifying a state update and a node transition as a single operation, making it useful for implementing handoffs.
3. Name of the agent or node to hand off to.
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
5. Indicate to LangGraph that we need to navigate to agent node in a **parent** multi-agent graph.
!!! tip
If you want to use tools that return `Command`, you can either use prebuilt @[`create_react_agent`][create_react_agent] / @[`ToolNode`][ToolNode] components, or implement your own tool-executing node that collects `Command` objects returned by the tools and returns a list of them, e.g.:
```typescript
const callTools = async (state) => {
// ...
const commands = await Promise.all(
toolCalls.map(toolCall => toolsByName[toolCall.name].invoke(toolCall))
);
return commands;
};
```
:::
!!! Important
This handoff implementation assumes that:
- each agent receives overall message history (across all agents) in the multi-agent system as its input. If you want more control over agent inputs, see [this section](#control-agent-inputs)
- each agent outputs its internal messages history to the overall message history of the multi-agent system. If you want more control over **how agent outputs are added**, wrap the agent in a separate node function:
- each agent receives overall message history (across all agents) in the multi-agent system as its input. If you want more control over agent inputs, see [this section](#control-agent-inputs)
- each agent outputs its internal messages history to the overall message history of the multi-agent system. If you want more control over **how agent outputs are added**, wrap the agent in a separate node function:
:::python
```python
def call_hotel_assistant(state):
# return agent's final response,
# excluding inner monologue
response = hotel_assistant.invoke(state)
# highlight-next-line
return {"messages": response["messages"][-1]}
```
:::
:::js
```typescript
const callHotelAssistant = async (state) => {
// return agent's final response,
// excluding inner monologue
const response = await hotelAssistant.invoke(state);
// highlight-next-line
return { messages: [response.messages.at(-1)] };
};
```
:::
```python
def call_hotel_assistant(state):
# return agent's final response,
# excluding inner monologue
response = hotel_assistant.invoke(state)
# highlight-next-line
return {"messages": response["messages"][-1]}
```
### Control agent inputs
:::python
You can use the @[`Send()`][Send] primitive to directly send data to the worker agents during the handoff. For example, you can request that the calling agent populate a task description for the next agent:
```python
@@ -214,63 +129,6 @@ def create_task_description_handoff_tool(
return handoff_tool
```
:::
:::js
You can use the @[`Send()`][Send] primitive to directly send data to the worker agents during the handoff. For example, you can request that the calling agent populate a task description for the next agent:
```typescript
import { tool } from "@langchain/core/tools";
import { Command, Send, MessagesZodState } from "@langchain/langgraph";
import { z } from "zod";
function createTaskDescriptionHandoffTool({
agentName,
description,
}: {
agentName: string;
description?: string;
}) {
const name = `transfer_to_${agentName}`;
const toolDescription = description || `Ask ${agentName} for help.`;
return tool(
async (
{ taskDescription },
config
) => {
const state = config.state;
const taskDescriptionMessage = {
role: "user" as const,
content: taskDescription,
};
const agentInput = {
...state,
messages: [taskDescriptionMessage],
};
return new Command({
// highlight-next-line
goto: [new Send(agentName, agentInput)],
graph: Command.PARENT,
});
},
{
name,
description: toolDescription,
schema: z.object({
taskDescription: z
.string()
.describe(
"Description of what the next agent should do, including all of the relevant context."
),
}),
}
);
}
```
:::
See the multi-agent [supervisor](../tutorials/multi_agent/agent_supervisor.md#4-create-delegation-tasks) example for a full example of using @[`Send()`][Send] in handoffs.
@@ -278,7 +136,6 @@ See the multi-agent [supervisor](../tutorials/multi_agent/agent_supervisor.md#4-
You can use handoffs in any agents built with LangGraph. We recommend using the prebuilt [agent](../agents/overview.md) or [`ToolNode`](./tool-calling.md#toolnode), as they natively support handoffs tools returning `Command`. Below is an example of how you can implement a multi-agent system for booking travel using handoffs:
:::python
```python
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, MessagesState
@@ -319,65 +176,9 @@ multi_agent_graph = (
.compile()
)
```
:::
:::js
```typescript
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { StateGraph, START, MessagesZodState } from "@langchain/langgraph";
import { z } from "zod";
function createHandoffTool({
agentName,
description,
}: {
agentName: string;
description?: string;
}) {
// same implementation as above
// ...
return new Command(/* ... */);
}
// Handoffs
const transferToHotelAssistant = createHandoffTool({
agentName: "hotel_assistant",
});
const transferToFlightAssistant = createHandoffTool({
agentName: "flight_assistant",
});
// Define agents
const flightAssistant = createReactAgent({
llm: model,
// highlight-next-line
tools: [/* ... */, transferToHotelAssistant],
// highlight-next-line
name: "flight_assistant",
});
const hotelAssistant = createReactAgent({
llm: model,
// highlight-next-line
tools: [/* ... */, transferToFlightAssistant],
// highlight-next-line
name: "hotel_assistant",
});
// Define multi-agent graph
const multiAgentGraph = new StateGraph(MessagesZodState)
// highlight-next-line
.addNode("flight_assistant", flightAssistant)
// highlight-next-line
.addNode("hotel_assistant", hotelAssistant)
.addEdge(START, "flight_assistant")
.compile();
```
:::
??? example "Full example: Multi-agent system for booking travel"
:::python
```python
from typing import Annotated
from langchain_core.messages import convert_to_messages
@@ -522,183 +323,6 @@ const multiAgentGraph = new StateGraph(MessagesZodState)
3. Name of the agent or node to hand off to.
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
5. Indicate to LangGraph that we need to navigate to agent node in a **parent** multi-agent graph.
:::
:::js
```typescript
import { tool } from "@langchain/core/tools";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { StateGraph, START, MessagesZodState, Command } from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";
import { isBaseMessage } from "@langchain/core/messages";
import { z } from "zod";
// We'll use a helper to render the streamed agent outputs nicely
const prettyPrintMessages = (update: Record<string, any>) => {
// Handle tuple case with namespace
if (Array.isArray(update)) {
const [ns, updateData] = update;
// Skip parent graph updates in the printouts
if (ns.length === 0) {
return;
}
const graphId = ns[ns.length - 1].split(":")[0];
console.log(`Update from subgraph ${graphId}:\n`);
update = updateData;
}
for (const [nodeName, updateValue] of Object.entries(update)) {
console.log(`Update from node ${nodeName}:\n`);
const messages = updateValue.messages || [];
for (const message of messages) {
if (isBaseMessage(message)) {
const textContent =
typeof message.content === "string"
? message.content
: JSON.stringify(message.content);
console.log(`${message.getType()}: ${textContent}`);
}
}
console.log("\n");
}
};
function createHandoffTool({
agentName,
description,
}: {
agentName: string;
description?: string;
}) {
const name = `transfer_to_${agentName}`;
const toolDescription = description || `Transfer to ${agentName}`;
return tool(
async (_, config) => {
// highlight-next-line
const state = config.state; // (1)!
const toolCallId = config.toolCall.id;
const toolMessage = {
role: "tool" as const,
content: `Successfully transferred to ${agentName}`,
name: name,
tool_call_id: toolCallId,
};
return new Command({
// highlight-next-line
goto: agentName, // (3)!
// highlight-next-line
update: { messages: [...state.messages, toolMessage] }, // (4)!
// highlight-next-line
graph: Command.PARENT, // (5)!
});
},
{
name,
description: toolDescription,
schema: z.object({}),
}
);
}
// Handoffs
const transferToHotelAssistant = createHandoffTool({
agentName: "hotel_assistant",
description: "Transfer user to the hotel-booking assistant.",
});
const transferToFlightAssistant = createHandoffTool({
agentName: "flight_assistant",
description: "Transfer user to the flight-booking assistant.",
});
// Simple agent tools
const bookHotel = tool(
async ({ hotelName }) => {
return `Successfully booked a stay at ${hotelName}.`;
},
{
name: "book_hotel",
description: "Book a hotel",
schema: z.object({
hotelName: z.string(),
}),
}
);
const bookFlight = tool(
async ({ fromAirport, toAirport }) => {
return `Successfully booked a flight from ${fromAirport} to ${toAirport}.`;
},
{
name: "book_flight",
description: "Book a flight",
schema: z.object({
fromAirport: z.string(),
toAirport: z.string(),
}),
}
);
const model = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
});
// Define agents
const flightAssistant = createReactAgent({
llm: model,
// highlight-next-line
tools: [bookFlight, transferToHotelAssistant],
prompt: "You are a flight booking assistant",
// highlight-next-line
name: "flight_assistant",
});
const hotelAssistant = createReactAgent({
llm: model,
// highlight-next-line
tools: [bookHotel, transferToFlightAssistant],
prompt: "You are a hotel booking assistant",
// highlight-next-line
name: "hotel_assistant",
});
// Define multi-agent graph
const multiAgentGraph = new StateGraph(MessagesZodState)
.addNode("flight_assistant", flightAssistant)
.addNode("hotel_assistant", hotelAssistant)
.addEdge(START, "flight_assistant")
.compile();
// Run the multi-agent graph
const stream = await multiAgentGraph.stream(
{
messages: [
{
role: "user",
content: "book a flight from BOS to JFK and a stay at McKittrick Hotel",
},
],
},
// highlight-next-line
{ subgraphs: true }
);
for await (const chunk of stream) {
prettyPrintMessages(chunk);
}
```
1. Access agent's state
2. The `Command` primitive allows specifying a state update and a node transition as a single operation, making it useful for implementing handoffs.
3. Name of the agent or node to hand off to.
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
5. Indicate to LangGraph that we need to navigate to agent node in a **parent** multi-agent graph.
:::
## Multi-turn conversation
@@ -709,7 +333,6 @@ The agents can then be implemented as nodes in a graph that executes agent steps
1. **Wait for user input** to continue the conversation, or
2. **Route to another agent** (or back to itself, such as in a loop) via a [handoff](#handoffs)
:::python
```python
def human(state) -> Command[Literal["agent", "another_agent"]]:
"""A node for collecting user input."""
@@ -737,44 +360,6 @@ def agent(state) -> Command[Literal["agent", "another_agent", "human"]]:
else:
return Command(goto="human") # Go to human node
```
:::
:::js
```typescript
import { interrupt, Command } from "@langchain/langgraph";
function human(state: MessagesState): Command {
const userInput: string = interrupt("Ready for user input.");
// Determine the active agent
const activeAgent = /* ... */;
return new Command({
update: {
messages: [{
role: "human",
content: userInput,
}]
},
goto: activeAgent,
});
}
function agent(state: MessagesState): Command {
// The condition for routing/halting can be anything, e.g. LLM tool call / structured output, etc.
const goto = getNextAgent(/* ... */); // 'agent' / 'anotherAgent'
if (goto) {
return new Command({
goto,
update: { myStateKey: "myStateValue" }
});
}
return new Command({ goto: "human" });
}
```
:::
??? example "Full example: multi-agent system for travel recommendations"
@@ -785,7 +370,6 @@ function agent(state: MessagesState): Command {
* travel_advisor: can help with travel destination recommendations. Can ask hotel_advisor for help.
* hotel_advisor: can help with hotel recommendations. Can ask travel_advisor for help.
:::python
```python
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, StateGraph, START
@@ -987,267 +571,10 @@ function agent(state: MessagesState): Command {
Would you like more specific information about any of these activities or would you like to know about other options in the area?
```
:::
:::js
```typescript
import { ChatAnthropic } from "@langchain/anthropic";
import { StateGraph, START, MessagesZodState, Command, interrupt, MemorySaver } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const model = new ChatAnthropic({ model: "claude-3-5-sonnet-latest" });
const MultiAgentState = MessagesZodState.extend({
lastActiveAgent: z.string().optional(),
});
// Define travel advisor tools
const getTravelRecommendations = tool(
async () => {
// Placeholder implementation
return "Based on current trends, I recommend visiting Japan, Portugal, or New Zealand.";
},
{
name: "get_travel_recommendations",
description: "Get current travel destination recommendations",
schema: z.object({}),
}
);
const makeHandoffTool = (agentName: string) => {
return tool(
async (_, config) => {
const state = config.state;
const toolCallId = config.toolCall.id;
const toolMessage = {
role: "tool" as const,
content: `Successfully transferred to ${agentName}`,
name: `transfer_to_${agentName}`,
tool_call_id: toolCallId,
};
return new Command({
goto: agentName,
update: { messages: [...state.messages, toolMessage] },
graph: Command.PARENT,
});
},
{
name: `transfer_to_${agentName}`,
description: `Transfer to ${agentName}`,
schema: z.object({}),
}
);
};
const travelAdvisorTools = [
getTravelRecommendations,
makeHandoffTool("hotel_advisor"),
];
const travelAdvisor = createReactAgent({
llm: model,
tools: travelAdvisorTools,
prompt: [
"You are a general travel expert that can recommend travel destinations (e.g. countries, cities, etc). ",
"If you need hotel recommendations, ask 'hotel_advisor' for help. ",
"You MUST include human-readable response before transferring to another agent."
].join("")
});
const callTravelAdvisor = async (
state: z.infer<typeof MultiAgentState>
): Promise<Command> => {
const response = await travelAdvisor.invoke(state);
const update = { ...response, lastActiveAgent: "travel_advisor" };
return new Command({ update, goto: "human" });
};
// Define hotel advisor tools
const getHotelRecommendations = tool(
async () => {
// Placeholder implementation
return "I recommend the Ritz-Carlton for luxury stays or boutique hotels for unique experiences.";
},
{
name: "get_hotel_recommendations",
description: "Get hotel recommendations for destinations",
schema: z.object({}),
}
);
const hotelAdvisorTools = [
getHotelRecommendations,
makeHandoffTool("travel_advisor"),
];
const hotelAdvisor = createReactAgent({
llm: model,
tools: hotelAdvisorTools,
prompt: [
"You are a hotel expert that can provide hotel recommendations for a given destination. ",
"If you need help picking travel destinations, ask 'travel_advisor' for help.",
"You MUST include human-readable response before transferring to another agent."
].join("")
});
const callHotelAdvisor = async (
state: z.infer<typeof MultiAgentState>
): Promise<Command> => {
const response = await hotelAdvisor.invoke(state);
const update = { ...response, lastActiveAgent: "hotel_advisor" };
return new Command({ update, goto: "human" });
};
const humanNode = async (
state: z.infer<typeof MultiAgentState>
): Promise<Command> => {
const userInput: string = interrupt("Ready for user input.");
const activeAgent = state.lastActiveAgent || "travel_advisor";
return new Command({
update: {
messages: [
{
role: "human",
content: userInput,
}
]
},
goto: activeAgent,
});
};
const builder = new StateGraph(MultiAgentState)
.addNode("travel_advisor", callTravelAdvisor)
.addNode("hotel_advisor", callHotelAdvisor)
.addNode("human", humanNode)
.addEdge(START, "travel_advisor");
const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });
```
Let's test a multi turn conversation with this application.
```typescript
import { v4 as uuidv4 } from "uuid";
import { Command } from "@langchain/langgraph";
const threadConfig = { configurable: { thread_id: uuidv4() } };
const inputs = [
// 1st round of conversation
{
messages: [
{ role: "user", content: "i wanna go somewhere warm in the caribbean" }
]
},
// Since we're using `interrupt`, we'll need to resume using the Command primitive.
// 2nd round of conversation
new Command({
resume: "could you recommend a nice hotel in one of the areas and tell me which area it is."
}),
// 3rd round of conversation
new Command({
resume: "i like the first one. could you recommend something to do near the hotel?"
}),
];
for (const [idx, userInput] of inputs.entries()) {
console.log();
console.log(`--- Conversation Turn ${idx + 1} ---`);
console.log();
console.log(`User: ${JSON.stringify(userInput)}`);
console.log();
for await (const update of await graph.stream(
userInput,
{ ...threadConfig, streamMode: "updates" }
)) {
for (const [nodeId, value] of Object.entries(update)) {
if (value?.messages?.length) {
const lastMessage = value.messages.at(-1);
if (lastMessage?.getType?.() === "ai") {
console.log(`${nodeId}: ${lastMessage.content}`);
}
}
}
}
}
```
```
--- Conversation Turn 1 ---
User: {"messages":[{"role":"user","content":"i wanna go somewhere warm in the caribbean"}]}
travel_advisor: Based on the recommendations, Aruba would be an excellent choice for your Caribbean getaway! Aruba is known as "One Happy Island" and offers:
- Year-round warm weather with consistent temperatures around 82°F (28°C)
- Beautiful white sand beaches like Eagle Beach and Palm Beach
- Clear turquoise waters perfect for swimming and snorkeling
- Minimal rainfall and location outside the hurricane belt
- A blend of Caribbean and Dutch culture
- Great dining options and nightlife
- Various water sports and activities
Would you like me to get some specific hotel recommendations in Aruba for your stay? I can transfer you to our hotel advisor who can help with accommodations.
--- Conversation Turn 2 ---
User: Command { resume: 'could you recommend a nice hotel in one of the areas and tell me which area it is.' }
hotel_advisor: Based on the recommendations, I can suggest two excellent options:
1. The Ritz-Carlton, Aruba - Located in Palm Beach
- This luxury resort is situated in the vibrant Palm Beach area
- Known for its exceptional service and amenities
- Perfect if you want to be close to dining, shopping, and entertainment
- Features multiple restaurants, a casino, and a world-class spa
- Located on a pristine stretch of Palm Beach
2. Bucuti & Tara Beach Resort - Located in Eagle Beach
- An adults-only boutique resort on Eagle Beach
- Known for being more intimate and peaceful
- Award-winning for its sustainability practices
- Perfect for a romantic getaway or peaceful vacation
- Located on one of the most beautiful beaches in the Caribbean
Would you like more specific information about either of these properties or their locations?
--- Conversation Turn 3 ---
User: Command { resume: 'i like the first one. could you recommend something to do near the hotel?' }
travel_advisor: Near the Ritz-Carlton in Palm Beach, here are some highly recommended activities:
1. Visit the Palm Beach Plaza Mall - Just a short walk from the hotel, featuring shopping, dining, and entertainment
2. Try your luck at the Stellaris Casino - It's right in the Ritz-Carlton
3. Take a sunset sailing cruise - Many depart from the nearby pier
4. Visit the California Lighthouse - A scenic landmark just north of Palm Beach
5. Enjoy water sports at Palm Beach:
- Jet skiing
- Parasailing
- Snorkeling
- Stand-up paddleboarding
Would you like more specific information about any of these activities or would you like to know about other options in the area?
```
:::
## Prebuilt implementations
LangGraph comes with prebuilt implementations of two of the most popular multi-agent architectures:
:::python
- [supervisor](../agents/multi-agent.md#supervisor) — individual agents are coordinated by a central supervisor agent. The supervisor controls all communication flow and task delegation, making decisions about which agent to invoke based on the current context and task requirements. You can use [`langgraph-supervisor`](https://github.com/langchain-ai/langgraph-supervisor-py) library to create a supervisor multi-agent systems.
- [swarm](../agents/multi-agent.md#supervisor) — agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent. You can use [`langgraph-swarm`](https://github.com/langchain-ai/langgraph-swarm-py) library to create a swarm multi-agent systems.
:::
:::js
- [supervisor](../agents/multi-agent.md#supervisor) — individual agents are coordinated by a central supervisor agent. The supervisor controls all communication flow and task delegation, making decisions about which agent to invoke based on the current context and task requirements. You can use [`langgraph-supervisor`](https://github.com/langchain-ai/langgraph-supervisor-js) library to create a supervisor multi-agent systems.
- [swarm](../agents/multi-agent.md#supervisor) — agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent. You can use [`langgraph-swarm`](https://github.com/langchain-ai/langgraph-swarm-js) library to create a swarm multi-agent systems.
:::
- [swarm](../agents/multi-agent.md#supervisor) — agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent. You can use [`langgraph-swarm`](https://github.com/langchain-ai/langgraph-swarm-py) library to create a swarm multi-agent systems.
+8 -465
View File
@@ -9,20 +9,11 @@ When adding subgraphs, you need to define how the parent graph and the subgraph
## Setup
:::python
```bash
pip install -U langgraph
```
:::
:::js
```bash
npm install @langchain/langgraph
```
:::
!!! tip "Set up LangSmith for LangGraph development"
Sign up for [LangSmith](https://smith.langchain.com) to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph — read more about how to get started [here](https://docs.smith.langchain.com).
## Shared state schemas
@@ -31,7 +22,6 @@ A common case is for the parent graph and subgraph to communicate over a shared
If your subgraph shares state keys with the parent graph, you can follow these steps to add it to your graph:
:::python
1. Define the subgraph workflow (`subgraph_builder` in the example below) and compile it
2. Pass compiled subgraph to the `.add_node` method when defining the parent graph workflow
@@ -59,41 +49,9 @@ builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()
```
:::
:::js
1. Define the subgraph workflow (`subgraphBuilder` in the example below) and compile it
2. Pass compiled subgraph to the `.addNode` method when defining the parent graph workflow
```typescript
import { StateGraph, START } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({
foo: z.string(),
});
// Subgraph
const subgraphBuilder = new StateGraph(State)
.addNode("subgraphNode1", (state) => {
return { foo: "hi! " + state.foo };
})
.addEdge(START, "subgraphNode1");
const subgraph = subgraphBuilder.compile();
// Parent graph
const builder = new StateGraph(State)
.addNode("node1", subgraph)
.addEdge(START, "node1");
const graph = builder.compile();
```
:::
??? example "Full example: shared state schemas"
:::python
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
@@ -143,61 +101,6 @@ const graph = builder.compile();
{'node_1': {'foo': 'hi! foo'}}
{'node_2': {'foo': 'hi! foobar'}}
```
:::
:::js
```typescript
import { StateGraph, START } from "@langchain/langgraph";
import { z } from "zod";
// Define subgraph
const SubgraphState = z.object({
foo: z.string(), // (1)!
bar: z.string(), // (2)!
});
const subgraphBuilder = new StateGraph(SubgraphState)
.addNode("subgraphNode1", (state) => {
return { bar: "bar" };
})
.addNode("subgraphNode2", (state) => {
// note that this node is using a state key ('bar') that is only available in the subgraph
// and is sending update on the shared state key ('foo')
return { foo: state.foo + state.bar };
})
.addEdge(START, "subgraphNode1")
.addEdge("subgraphNode1", "subgraphNode2");
const subgraph = subgraphBuilder.compile();
// Define parent graph
const ParentState = z.object({
foo: z.string(),
});
const builder = new StateGraph(ParentState)
.addNode("node1", (state) => {
return { foo: "hi! " + state.foo };
})
.addNode("node2", subgraph)
.addEdge(START, "node1")
.addEdge("node1", "node2");
const graph = builder.compile();
for await (const chunk of await graph.stream({ foo: "foo" })) {
console.log(chunk);
}
```
3. This key is shared with the parent graph state
4. This key is private to the `SubgraphState` and is not visible to the parent graph
```
{ node1: { foo: 'hi! foo' } }
{ node2: { foo: 'hi! foobar' } }
```
:::
## Different state schemas
@@ -205,7 +108,6 @@ For more complex systems you might want to define subgraphs that have a **comple
If that's the case for your application, you need to define a node **function that invokes the subgraph**. This function needs to transform the input (parent) state to the subgraph state before invoking the subgraph, and transform the results back to the parent state before returning the state update from the node.
:::python
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
@@ -240,48 +142,9 @@ graph = builder.compile()
1. Transform the state to the subgraph state
2. Transform response back to the parent state
:::
:::js
```typescript
import { StateGraph, START } from "@langchain/langgraph";
import { z } from "zod";
const SubgraphState = z.object({
bar: z.string(),
});
// Subgraph
const subgraphBuilder = new StateGraph(SubgraphState)
.addNode("subgraphNode1", (state) => {
return { bar: "hi! " + state.bar };
})
.addEdge(START, "subgraphNode1");
const subgraph = subgraphBuilder.compile();
// Parent graph
const State = z.object({
foo: z.string(),
});
const builder = new StateGraph(State)
.addNode("node1", async (state) => {
const subgraphOutput = await subgraph.invoke({ bar: state.foo }); // (1)!
return { foo: subgraphOutput.bar }; // (2)!
})
.addEdge(START, "node1");
const graph = builder.compile();
```
1. Transform the state to the subgraph state
2. Transform response back to the parent state
:::
??? example "Full example: different state schemas"
:::python
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
@@ -337,74 +200,11 @@ const graph = builder.compile();
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})
```
:::
:::js
```typescript
import { StateGraph, START } from "@langchain/langgraph";
import { z } from "zod";
// Define subgraph
const SubgraphState = z.object({
// note that none of these keys are shared with the parent graph state
bar: z.string(),
baz: z.string(),
});
const subgraphBuilder = new StateGraph(SubgraphState)
.addNode("subgraphNode1", (state) => {
return { baz: "baz" };
})
.addNode("subgraphNode2", (state) => {
return { bar: state.bar + state.baz };
})
.addEdge(START, "subgraphNode1")
.addEdge("subgraphNode1", "subgraphNode2");
const subgraph = subgraphBuilder.compile();
// Define parent graph
const ParentState = z.object({
foo: z.string(),
});
const builder = new StateGraph(ParentState)
.addNode("node1", (state) => {
return { foo: "hi! " + state.foo };
})
.addNode("node2", async (state) => {
const response = await subgraph.invoke({ bar: state.foo }); // (1)!
return { foo: response.bar }; // (2)!
})
.addEdge(START, "node1")
.addEdge("node1", "node2");
const graph = builder.compile();
for await (const chunk of await graph.stream(
{ foo: "foo" },
{ subgraphs: true }
)) {
console.log(chunk);
}
```
3. Transform the state to the subgraph state
4. Transform response back to the parent state
```
[[], { node1: { foo: 'hi! foo' } }]
[['node2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7'], { subgraphNode1: { baz: 'baz' } }]
[['node2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7'], { subgraphNode2: { bar: 'hi! foobaz' } }]
[[], { node2: { foo: 'hi! foobaz' } }]
```
:::
??? example "Full example: different state schemas (two levels of subgraphs)"
This is an example with two levels of subgraphs: parent -> child -> grandchild.
:::python
```python
# Grandchild graph
from typing_extensions import TypedDict
@@ -488,102 +288,14 @@ const graph = builder.compile();
((), {'child': {'my_key': 'hi Bob, how are you today?'}})
((), {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}})
```
:::
:::js
```typescript
import { StateGraph, START, END } from "@langchain/langgraph";
import { z } from "zod";
// Grandchild graph
const GrandChildState = z.object({
myGrandchildKey: z.string(),
});
const grandchild = new StateGraph(GrandChildState)
.addNode("grandchild1", (state) => {
// NOTE: child or parent keys will not be accessible here
return { myGrandchildKey: state.myGrandchildKey + ", how are you" };
})
.addEdge(START, "grandchild1")
.addEdge("grandchild1", END);
const grandchildGraph = grandchild.compile();
// Child graph
const ChildState = z.object({
myChildKey: z.string(),
});
const child = new StateGraph(ChildState)
.addNode("child1", async (state) => {
// NOTE: parent or grandchild keys won't be accessible here
const grandchildGraphInput = { myGrandchildKey: state.myChildKey }; // (1)!
const grandchildGraphOutput = await grandchildGraph.invoke(grandchildGraphInput);
return { myChildKey: grandchildGraphOutput.myGrandchildKey + " today?" }; // (2)!
}) // (3)!
.addEdge(START, "child1")
.addEdge("child1", END);
const childGraph = child.compile();
// Parent graph
const ParentState = z.object({
myKey: z.string(),
});
const parent = new StateGraph(ParentState)
.addNode("parent1", (state) => {
// NOTE: child or grandchild keys won't be accessible here
return { myKey: "hi " + state.myKey };
})
.addNode("child", async (state) => {
const childGraphInput = { myChildKey: state.myKey }; // (4)!
const childGraphOutput = await childGraph.invoke(childGraphInput);
return { myKey: childGraphOutput.myChildKey }; // (5)!
}) // (6)!
.addNode("parent2", (state) => {
return { myKey: state.myKey + " bye!" };
})
.addEdge(START, "parent1")
.addEdge("parent1", "child")
.addEdge("child", "parent2")
.addEdge("parent2", END);
const parentGraph = parent.compile();
for await (const chunk of await parentGraph.stream(
{ myKey: "Bob" },
{ subgraphs: true }
)) {
console.log(chunk);
}
```
7. We're transforming the state from the child state channels (`myChildKey`) to the grandchild state channels (`myGrandchildKey`)
8. We're transforming the state from the grandchild state channels (`myGrandchildKey`) back to the child state channels (`myChildKey`)
9. We're passing a function here instead of just compiled graph (`grandchildGraph`)
10. We're transforming the state from the parent state channels (`myKey`) to the child state channels (`myChildKey`)
11. We're transforming the state from the child state channels (`myChildKey`) back to the parent state channels (`myKey`)
12. We're passing a function here instead of just a compiled graph (`childGraph`)
```
[[], { parent1: { myKey: 'hi Bob' } }]
[['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child1:781bb3b1-3971-84ce-810b-acf819a03f9c'], { grandchild1: { myGrandchildKey: 'hi Bob, how are you' } }]
[['child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b'], { child1: { myChildKey: 'hi Bob, how are you today?' } }]
[[], { child: { myKey: 'hi Bob, how are you today?' } }]
[[], { parent2: { myKey: 'hi Bob, how are you today? bye!' } }]
```
:::
## Add persistence
You only need to **provide the checkpointer when compiling the parent graph**. LangGraph will automatically propagate the checkpointer to the child subgraphs.
:::python
```python
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict
class State(TypedDict):
@@ -605,66 +317,20 @@ builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = MemorySaver()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
```
:::
:::js
```typescript
import { StateGraph, START, MemorySaver } from "@langchain/langgraph";
import { z } from "zod";
If you want the subgraph to **have its own memory**, you can compile it `with checkpointer=True`. This is useful in [multi-agent](../concepts/multi_agent.md) systems, if you want agents to keep track of their internal message histories:
const State = z.object({
foo: z.string(),
});
// Subgraph
const subgraphBuilder = new StateGraph(State)
.addNode("subgraphNode1", (state) => {
return { foo: state.foo + "bar" };
})
.addEdge(START, "subgraphNode1");
const subgraph = subgraphBuilder.compile();
// Parent graph
const builder = new StateGraph(State)
.addNode("node1", subgraph)
.addEdge(START, "node1");
const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });
```
:::
If you want the subgraph to **have its own memory**, you can compile it with the appropriate checkpointer option. This is useful in [multi-agent](../concepts/multi_agent.md) systems, if you want agents to keep track of their internal message histories:
:::python
```python
subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)
```
:::
:::js
```typescript
const subgraphBuilder = new StateGraph(...)
const subgraph = subgraphBuilder.compile({ checkpointer: true });
```
:::
## View subgraph state
When you enable [persistence](../concepts/persistence.md), you can [inspect the graph state](../concepts/persistence.md#checkpoints) (checkpoint) via the appropriate method. To view the subgraph state, you can use the subgraphs option.
:::python
You can inspect the graph state via `graph.get_state(config)`. To view the subgraph state, you can use `graph.get_state(config, subgraphs=True)`.
:::
:::js
You can inspect the graph state via `graph.getState(config)`. To view the subgraph state, you can use `graph.getState(config, { subgraphs: true })`.
:::
When you enable [persistence](../concepts/persistence.md), you can [inspect the graph state](../concepts/persistence.md#checkpoints) (checkpoint) via `graph.get_state(config)`. To view the subgraph state, you can use `graph.get_state(config, subgraphs=True)`.
!!! important "Available **only** when interrupted"
@@ -672,10 +338,9 @@ You can inspect the graph state via `graph.getState(config)`. To view the subgra
??? example "View interrupted subgraph state"
:::python
```python
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict
@@ -700,7 +365,7 @@ You can inspect the graph state via `graph.getState(config)`. To view the subgra
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = MemorySaver()
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
@@ -714,53 +379,11 @@ You can inspect the graph state via `graph.getState(config)`. To view the subgra
```
1. This will be available only when the subgraph is interrupted. Once you resume the graph, you won't be able to access the subgraph state.
:::
:::js
```typescript
import { StateGraph, START, MemorySaver, interrupt, Command } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({
foo: z.string(),
});
// Subgraph
const subgraphBuilder = new StateGraph(State)
.addNode("subgraphNode1", (state) => {
const value = interrupt("Provide value:");
return { foo: state.foo + value };
})
.addEdge(START, "subgraphNode1");
const subgraph = subgraphBuilder.compile();
// Parent graph
const builder = new StateGraph(State)
.addNode("node1", subgraph)
.addEdge(START, "node1");
const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });
const config = { configurable: { thread_id: "1" } };
await graph.invoke({ foo: "" }, config);
const parentState = await graph.getState(config);
const subgraphState = (await graph.getState(config, { subgraphs: true })).tasks[0].state; // (1)!
// resume the subgraph
await graph.invoke(new Command({ resume: "bar" }), config);
```
2. This will be available only when the subgraph is interrupted. Once you resume the graph, you won't be able to access the subgraph state.
:::
## Stream subgraph outputs
To include outputs from subgraphs in the streamed outputs, you can set the subgraphs option in the stream method of the parent graph. This will stream outputs from both the parent graph and any subgraphs.
To include outputs from subgraphs in the streamed outputs, you can set `subgraphs=True` in the `.stream()` method of the parent graph. This will stream outputs from both the parent graph and any subgraphs.
:::python
```python
for chunk in graph.stream(
{"foo": "foo"},
@@ -771,27 +394,9 @@ for chunk in graph.stream(
```
1. Set `subgraphs=True` to stream outputs from subgraphs.
:::
:::js
```typescript
for await (const chunk of await graph.stream(
{ foo: "foo" },
{
subgraphs: true, // (1)!
streamMode: "updates",
}
)) {
console.log(chunk);
}
```
1. Set `subgraphs: true` to stream outputs from subgraphs.
:::
??? example "Stream from subgraphs"
:::python
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
@@ -845,66 +450,4 @@ for await (const chunk of await graph.stream(
(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_1': {'bar': 'bar'}})
(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
((), {'node_2': {'foo': 'hi! foobar'}})
```
:::
:::js
```typescript
import { StateGraph, START } from "@langchain/langgraph";
import { z } from "zod";
// Define subgraph
const SubgraphState = z.object({
foo: z.string(),
bar: z.string(),
});
const subgraphBuilder = new StateGraph(SubgraphState)
.addNode("subgraphNode1", (state) => {
return { bar: "bar" };
})
.addNode("subgraphNode2", (state) => {
// note that this node is using a state key ('bar') that is only available in the subgraph
// and is sending update on the shared state key ('foo')
return { foo: state.foo + state.bar };
})
.addEdge(START, "subgraphNode1")
.addEdge("subgraphNode1", "subgraphNode2");
const subgraph = subgraphBuilder.compile();
// Define parent graph
const ParentState = z.object({
foo: z.string(),
});
const builder = new StateGraph(ParentState)
.addNode("node1", (state) => {
return { foo: "hi! " + state.foo };
})
.addNode("node2", subgraph)
.addEdge(START, "node1")
.addEdge("node1", "node2");
const graph = builder.compile();
for await (const chunk of await graph.stream(
{ foo: "foo" },
{
streamMode: "updates",
subgraphs: true, // (1)!
}
)) {
console.log(chunk);
}
```
2. Set `subgraphs: true` to stream outputs from subgraphs.
```
[[], { node1: { foo: 'hi! foo' } }]
[['node2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'], { subgraphNode1: { bar: 'bar' } }]
[['node2:e58e5673-a661-ebb0-70d4-e298a7fc28b7'], { subgraphNode2: { foo: 'hi! foobar' } }]
[[], { node2: { foo: 'hi! foobar' } }]
```
:::
+1 -1
View File
@@ -1948,7 +1948,7 @@ const llmWithTools = llm.bindTools(tools);
# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: MessagesState) -> Literal["Action", END]:
def should_continue(state: MessagesState) -> Literal["environment", END]:
"""Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
messages = state["messages"]
+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>
<strong>LangGraph Platform docs have moved!</strong> Find the LangGraph Platform docs at the new <a href="https://docs.langchain.com/langgraph-platform" target="_blank">LangChain Docs</a> site.
{% 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"
]
}
],
+1 -3
View File
@@ -707,9 +707,7 @@
" \"\"\"\n",
" Find all tool calls in the messages returned\n",
" \"\"\"\n",
" tool_calls = [\n",
" tc[\"name\"] for m in messages[\"messages\"] for tc in getattr(m, \"tool_calls\", [])\n",
" ]\n",
" tool_calls = [tc['name'] for m in messages['messages'] for tc in getattr(m, 'tool_calls', [])]\n",
" return tool_calls\n",
"\n",
"\n",
@@ -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,
@@ -16,15 +21,10 @@ from langgraph.checkpoint.base import (
get_checkpoint_id,
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
@@ -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,
@@ -16,15 +21,10 @@ from langgraph.checkpoint.base import (
get_checkpoint_id,
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
@@ -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"]),
},
},
@@ -5,6 +5,8 @@ from collections.abc import Sequence
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,
@@ -12,7 +14,6 @@ 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]]
@@ -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_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.
@@ -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
+5 -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
@@ -344,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,
+2 -40
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,
@@ -861,41 +861,3 @@ def test_store_ttl(store):
# Now has been (TTL_SECONDS-2)*2 > TTL_SECONDS + TTL_SECONDS/2
res = store.search(ns, query="bar", refresh_ttl=False)
assert len(res) == 0
@pytest.mark.parametrize(
"vector_type,distance_type",
[
("vector", "cosine"),
("vector", "inner_product"),
("halfvec", "cosine"),
("halfvec", "inner_product"),
],
)
def test_non_ascii(
request: Any,
fake_embeddings: CharacterEmbeddings,
vector_type: str,
distance_type: str,
) -> None:
"""Test support for non-ascii characters"""
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": "これは日本語です"}
) # Japanese
store.put(("user_123", "memories"), "3", {"text": "이건 한국어야"}) # Korean
store.put(("user_123", "memories"), "4", {"text": "Это русский"}) # Russian
store.put(("user_123", "memories"), "5", {"text": "यह रूसी है"}) # Hindi
result1 = store.search(("user_123", "memories"), query="这是中文")
result2 = store.search(("user_123", "memories"), query="これは日本語です")
result3 = store.search(("user_123", "memories"), query="이건 한국어야")
result4 = store.search(("user_123", "memories"), query="Это русский")
result5 = store.search(("user_123", "memories"), query="यह रूसी है")
assert result1[0].key == "1"
assert result2[0].key == "2"
assert result3[0].key == "3"
assert result4[0].key == "4"
assert result5[0].key == "5"
+5 -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
@@ -332,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"] == {}
+498 -484
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 -29
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
@@ -1067,31 +1067,3 @@ def test_sql_injection_vulnerability(store: SqliteStore) -> None:
with pytest.raises(ValueError, match="Invalid filter key"):
store.search(("docs",), filter={malicious_key: "dummy"})
@pytest.mark.parametrize("distance_type", VECTOR_TYPES)
def test_non_ascii(
fake_embeddings: CharacterEmbeddings,
distance_type: str,
) -> None:
"""Test support for non-ascii characters"""
with create_vector_store(fake_embeddings, distance_type=distance_type) as store:
store.put(("user_123", "memories"), "1", {"text": "这是中文"}) # Chinese
store.put(
("user_123", "memories"), "2", {"text": "これは日本語です"}
) # Japanese
store.put(("user_123", "memories"), "3", {"text": "이건 한국어야"}) # Korean
store.put(("user_123", "memories"), "4", {"text": "Это русский"}) # Russian
store.put(("user_123", "memories"), "5", {"text": "यह रूसी है"}) # Hindi
result1 = store.search(("user_123", "memories"), query="这是中文")
result2 = store.search(("user_123", "memories"), query="これは日本語です")
result3 = store.search(("user_123", "memories"), query="이건 한국어야")
result4 = store.search(("user_123", "memories"), query="Это русский")
result5 = store.search(("user_123", "memories"), query="यह रूसी है")
assert result1[0].key == "1"
assert result2[0].key == "2"
assert result3[0].key == "3"
assert result4[0].key == "4"
assert result5[0].key == "5"
+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"
)
+439 -426
View File
File diff suppressed because it is too large Load Diff
-144
View File
@@ -1,144 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Any
from langgraph.cache.base import BaseCache, FullKey, Namespace, ValueT
from langgraph.checkpoint.serde.base import SerializerProtocol
class RedisCache(BaseCache[ValueT]):
"""Redis-based cache implementation with TTL support."""
def __init__(
self,
redis: Any,
*,
serde: SerializerProtocol | None = None,
prefix: str = "langgraph:cache:",
) -> None:
"""Initialize the cache with a Redis client.
Args:
redis: Redis client instance (sync or async)
serde: Serializer to use for values
prefix: Key prefix for all cached values
"""
super().__init__(serde=serde)
self.redis = redis
self.prefix = prefix
def _make_key(self, ns: Namespace, key: str) -> str:
"""Create a Redis key from namespace and key."""
ns_str = ":".join(ns) if ns else ""
return f"{self.prefix}{ns_str}:{key}" if ns_str else f"{self.prefix}{key}"
def _parse_key(self, redis_key: str) -> tuple[Namespace, str]:
"""Parse a Redis key back to namespace and key."""
if not redis_key.startswith(self.prefix):
raise ValueError(
f"Key {redis_key} does not start with prefix {self.prefix}"
)
remaining = redis_key[len(self.prefix) :]
if ":" in remaining:
parts = remaining.split(":")
key = parts[-1]
ns_parts = parts[:-1]
return (tuple(ns_parts), key)
else:
return (tuple(), remaining)
def get(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]:
"""Get the cached values for the given keys."""
if not keys:
return {}
# Build Redis keys
redis_keys = [self._make_key(ns, key) for ns, key in keys]
# Get values from Redis using MGET
try:
raw_values = self.redis.mget(redis_keys)
except Exception:
# If Redis is unavailable, return empty dict
return {}
values: dict[FullKey, ValueT] = {}
for i, raw_value in enumerate(raw_values):
if raw_value is not None:
try:
# Deserialize the value
encoding, data = raw_value.split(b":", 1)
values[keys[i]] = self.serde.loads_typed((encoding.decode(), data))
except Exception:
# Skip corrupted entries
continue
return values
async def aget(self, keys: Sequence[FullKey]) -> dict[FullKey, ValueT]:
"""Asynchronously get the cached values for the given keys."""
return self.get(keys)
def set(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None:
"""Set the cached values for the given keys and TTLs."""
if not mapping:
return
# Use pipeline for efficient batch operations
pipe = self.redis.pipeline()
for (ns, key), (value, ttl) in mapping.items():
redis_key = self._make_key(ns, key)
encoding, data = self.serde.dumps_typed(value)
# Store as "encoding:data" format
serialized_value = f"{encoding}:".encode() + data
if ttl is not None:
pipe.setex(redis_key, ttl, serialized_value)
else:
pipe.set(redis_key, serialized_value)
try:
pipe.execute()
except Exception:
# Silently fail if Redis is unavailable
pass
async def aset(self, mapping: Mapping[FullKey, tuple[ValueT, int | None]]) -> None:
"""Asynchronously set the cached values for the given keys and TTLs."""
self.set(mapping)
def clear(self, namespaces: Sequence[Namespace] | None = None) -> None:
"""Delete the cached values for the given namespaces.
If no namespaces are provided, clear all cached values."""
try:
if namespaces is None:
# Clear all keys with our prefix
pattern = f"{self.prefix}*"
keys = self.redis.keys(pattern)
if keys:
self.redis.delete(*keys)
else:
# Clear specific namespaces
keys_to_delete = []
for ns in namespaces:
ns_str = ":".join(ns) if ns else ""
pattern = (
f"{self.prefix}{ns_str}:*" if ns_str else f"{self.prefix}*"
)
keys = self.redis.keys(pattern)
keys_to_delete.extend(keys)
if keys_to_delete:
self.redis.delete(*keys_to_delete)
except Exception:
# Silently fail if Redis is unavailable
pass
async def aclear(self, namespaces: Sequence[Namespace] | None = None) -> None:
"""Asynchronously delete the cached values for the given namespaces.
If no namespaces are provided, clear all cached values."""
self.clear(namespaces)
@@ -81,9 +81,6 @@ class Checkpoint(TypedDict):
This keeps track of the versions of the channels that each node has seen.
Used to determine which nodes to execute next.
"""
updated_channels: list[str] | None
"""The channels that were updated in this checkpoint.
"""
def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
@@ -95,7 +92,6 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
channel_versions=checkpoint["channel_versions"].copy(),
versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
pending_sends=checkpoint.get("pending_sends", []).copy(),
updated_channels=checkpoint.get("updated_channels", None),
)
@@ -441,7 +437,6 @@ def empty_checkpoint() -> Checkpoint:
channel_versions={},
versions_seen={},
pending_sends=[],
updated_channels=None,
)
@@ -475,5 +470,4 @@ def create_checkpoint(
channel_versions=checkpoint["channel_versions"],
versions_seen=checkpoint["versions_seen"],
pending_sends=checkpoint.get("pending_sends", []),
updated_channels=None,
)
+7 -14
View File
@@ -64,21 +64,14 @@ class AsyncBatchedBaseStore(BaseStore):
super().__init__()
self._loop = asyncio.get_running_loop()
self._aqueue: asyncio.Queue[tuple[asyncio.Future, Op]] = asyncio.Queue()
self._task: asyncio.Task | None = None
self._ensure_task()
self._task = self._loop.create_task(_run(self._aqueue, weakref.ref(self)))
def __del__(self) -> None:
try:
if self._task:
self._task.cancel()
self._task.cancel()
except RuntimeError:
pass
def _ensure_task(self) -> None:
"""Ensure the background processing loop is running."""
if self._task is None or self._task.done():
self._task = self._loop.create_task(_run(self._aqueue, weakref.ref(self)))
async def aget(
self,
namespace: tuple[str, ...],
@@ -86,7 +79,7 @@ class AsyncBatchedBaseStore(BaseStore):
*,
refresh_ttl: bool | None = None,
) -> Item | None:
self._ensure_task()
assert not self._task.done()
fut = self._loop.create_future()
self._aqueue.put_nowait(
(
@@ -111,7 +104,7 @@ class AsyncBatchedBaseStore(BaseStore):
offset: int = 0,
refresh_ttl: bool | None = None,
) -> list[SearchItem]:
self._ensure_task()
assert not self._task.done()
fut = self._loop.create_future()
self._aqueue.put_nowait(
(
@@ -137,7 +130,7 @@ class AsyncBatchedBaseStore(BaseStore):
*,
ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None:
self._ensure_task()
assert not self._task.done()
_validate_namespace(namespace)
fut = self._loop.create_future()
self._aqueue.put_nowait(
@@ -155,7 +148,7 @@ class AsyncBatchedBaseStore(BaseStore):
namespace: tuple[str, ...],
key: str,
) -> None:
self._ensure_task()
assert not self._task.done()
fut = self._loop.create_future()
self._aqueue.put_nowait((fut, PutOp(namespace, key, None)))
return await fut
@@ -169,7 +162,7 @@ class AsyncBatchedBaseStore(BaseStore):
limit: int = 100,
offset: int = 0,
) -> list[tuple[str, ...]]:
self._ensure_task()
assert not self._task.done()
fut = self._loop.create_future()
match_conditions = []
if prefix:
@@ -238,7 +238,7 @@ def get_text_at_path(obj: Any, path: str | list[str]) -> list[str]:
- Nested paths in multi-field: "{field1,nested.field2}"
"""
if not path or path == "$":
return [json.dumps(obj, sort_keys=True, ensure_ascii=False)]
return [json.dumps(obj, sort_keys=True)]
tokens = tokenize_path(path) if isinstance(path, str) else path
@@ -249,7 +249,7 @@ def get_text_at_path(obj: Any, path: str | list[str]) -> list[str]:
elif obj is None:
return []
elif isinstance(obj, (list, dict)):
return [json.dumps(obj, sort_keys=True, ensure_ascii=False)]
return [json.dumps(obj, sort_keys=True)]
return []
token = tokens[pos]
@@ -295,11 +295,7 @@ def get_text_at_path(obj: Any, path: str | list[str]) -> list[str]:
if isinstance(current_obj, (str, int, float, bool)):
results.append(str(current_obj))
elif isinstance(current_obj, (list, dict)):
results.append(
json.dumps(
current_obj, sort_keys=True, ensure_ascii=False
)
)
results.append(json.dumps(current_obj, sort_keys=True))
# Handle wildcard
elif token == "*":
-1
View File
@@ -32,7 +32,6 @@ dev = [
"numpy",
"pandas",
"pandas-stubs>=2.2.2.240807",
"redis",
]
[tool.hatch.build.targets.wheel]
-316
View File
@@ -1,316 +0,0 @@
"""Unit tests for Redis cache implementation."""
import time
import pytest
import redis
from langgraph.cache.base import FullKey
from langgraph.cache.redis import RedisCache
class TestRedisCache:
@pytest.fixture(autouse=True)
def setup(self) -> None:
"""Set up test Redis client and cache."""
self.client = redis.Redis(
host="localhost", port=6379, db=0, decode_responses=False
)
try:
self.client.ping()
except redis.ConnectionError:
pytest.skip("Redis server not available")
self.cache: RedisCache = RedisCache(self.client, prefix="test:cache:")
# Clean up before each test
self.client.flushdb()
def teardown_method(self) -> None:
"""Clean up after each test."""
try:
self.client.flushdb()
except Exception:
pass
def test_basic_set_and_get(self) -> None:
"""Test basic set and get operations."""
keys: list[FullKey] = [(("graph", "node"), "key1")]
values = {keys[0]: ({"result": 42}, None)}
# Set value
self.cache.set(values)
# Get value
result = self.cache.get(keys)
assert len(result) == 1
assert result[keys[0]] == {"result": 42}
def test_batch_operations(self) -> None:
"""Test batch set and get operations."""
keys: list[FullKey] = [
(("graph", "node1"), "key1"),
(("graph", "node2"), "key2"),
(("other", "node"), "key3"),
]
values = {
keys[0]: ({"result": 1}, None),
keys[1]: ({"result": 2}, 60), # With TTL
keys[2]: ({"result": 3}, None),
}
# Set values
self.cache.set(values)
# Get all values
result = self.cache.get(keys)
assert len(result) == 3
assert result[keys[0]] == {"result": 1}
assert result[keys[1]] == {"result": 2}
assert result[keys[2]] == {"result": 3}
def test_ttl_behavior(self) -> None:
"""Test TTL (time-to-live) functionality."""
key: FullKey = (("graph", "node"), "ttl_key")
values = {key: ({"data": "expires_soon"}, 1)} # 1 second TTL
# Set with TTL
self.cache.set(values)
# Should be available immediately
result = self.cache.get([key])
assert len(result) == 1
assert result[key] == {"data": "expires_soon"}
# Wait for expiration
time.sleep(1.1)
# Should be expired
result = self.cache.get([key])
assert len(result) == 0
def test_namespace_isolation(self) -> None:
"""Test that different namespaces are isolated."""
key1: FullKey = (("graph1", "node"), "same_key")
key2: FullKey = (("graph2", "node"), "same_key")
values = {key1: ({"graph": 1}, None), key2: ({"graph": 2}, None)}
self.cache.set(values)
result = self.cache.get([key1, key2])
assert result[key1] == {"graph": 1}
assert result[key2] == {"graph": 2}
def test_clear_all(self) -> None:
"""Test clearing all cached values."""
keys: list[FullKey] = [
(("graph", "node1"), "key1"),
(("graph", "node2"), "key2"),
]
values = {keys[0]: ({"result": 1}, None), keys[1]: ({"result": 2}, None)}
self.cache.set(values)
# Verify data exists
result = self.cache.get(keys)
assert len(result) == 2
# Clear all
self.cache.clear()
# Verify data is gone
result = self.cache.get(keys)
assert len(result) == 0
def test_clear_by_namespace(self) -> None:
"""Test clearing cached values by namespace."""
keys: list[FullKey] = [
(("graph1", "node"), "key1"),
(("graph2", "node"), "key2"),
(("graph1", "other"), "key3"),
]
values = {
keys[0]: ({"result": 1}, None),
keys[1]: ({"result": 2}, None),
keys[2]: ({"result": 3}, None),
}
self.cache.set(values)
# Clear only graph1 namespace
self.cache.clear([("graph1", "node"), ("graph1", "other")])
# graph1 should be cleared, graph2 should remain
result = self.cache.get(keys)
assert len(result) == 1
assert result[keys[1]] == {"result": 2}
def test_empty_operations(self) -> None:
"""Test behavior with empty keys/values."""
# Empty get
result = self.cache.get([])
assert result == {}
# Empty set
self.cache.set({}) # Should not raise error
def test_nonexistent_keys(self) -> None:
"""Test getting keys that don't exist."""
keys: list[FullKey] = [(("graph", "node"), "nonexistent")]
result = self.cache.get(keys)
assert len(result) == 0
@pytest.mark.asyncio
async def test_async_operations(self) -> None:
"""Test async set and get operations with sync Redis client."""
# Create sync Redis client and cache (like main integration tests)
client = redis.Redis(host="localhost", port=6379, db=1, decode_responses=False)
try:
client.ping()
except Exception:
pytest.skip("Redis not available")
cache: RedisCache = RedisCache(client, prefix="test:async:")
keys: list[FullKey] = [(("graph", "node"), "async_key")]
values = {keys[0]: ({"async": True}, None)}
# Async set (delegates to sync)
await cache.aset(values)
# Async get (delegates to sync)
result = await cache.aget(keys)
assert len(result) == 1
assert result[keys[0]] == {"async": True}
# Cleanup
client.flushdb()
@pytest.mark.asyncio
async def test_async_clear(self) -> None:
"""Test async clear operations with sync Redis client."""
# Create sync Redis client and cache (like main integration tests)
client = redis.Redis(host="localhost", port=6379, db=1, decode_responses=False)
try:
client.ping()
except Exception:
pytest.skip("Redis not available")
cache: RedisCache = RedisCache(client, prefix="test:async:")
keys: list[FullKey] = [(("graph", "node"), "key")]
values = {keys[0]: ({"data": "test"}, None)}
await cache.aset(values)
# Verify data exists
result = await cache.aget(keys)
assert len(result) == 1
# Clear all (delegates to sync)
await cache.aclear()
# Verify data is gone
result = await cache.aget(keys)
assert len(result) == 0
# Cleanup
client.flushdb()
def test_redis_unavailable_get(self) -> None:
"""Test behavior when Redis is unavailable during get operations."""
# Create cache with non-existent Redis server
bad_client = redis.Redis(
host="nonexistent", port=9999, socket_connect_timeout=0.1
)
cache: RedisCache = RedisCache(bad_client, prefix="test:cache:")
keys: list[FullKey] = [(("graph", "node"), "key")]
result = cache.get(keys)
# Should return empty dict when Redis unavailable
assert result == {}
def test_redis_unavailable_set(self) -> None:
"""Test behavior when Redis is unavailable during set operations."""
# Create cache with non-existent Redis server
bad_client = redis.Redis(
host="nonexistent", port=9999, socket_connect_timeout=0.1
)
cache: RedisCache = RedisCache(bad_client, prefix="test:cache:")
keys: list[FullKey] = [(("graph", "node"), "key")]
values = {keys[0]: ({"data": "test"}, None)}
# Should not raise exception when Redis unavailable
cache.set(values) # Should silently fail
@pytest.mark.asyncio
async def test_redis_unavailable_async(self) -> None:
"""Test async behavior when Redis is unavailable."""
# Create sync cache with non-existent Redis server (like main integration tests)
bad_client = redis.Redis(
host="nonexistent", port=9999, socket_connect_timeout=0.1
)
cache: RedisCache = RedisCache(bad_client, prefix="test:cache:")
keys: list[FullKey] = [(("graph", "node"), "key")]
values = {keys[0]: ({"data": "test"}, None)}
# Should return empty dict for get (delegates to sync)
result = await cache.aget(keys)
assert result == {}
# Should not raise exception for set (delegates to sync)
await cache.aset(values) # Should silently fail
def test_corrupted_data_handling(self) -> None:
"""Test handling of corrupted data in Redis."""
# Set some valid data first
keys: list[FullKey] = [(("graph", "node"), "valid_key")]
values = {keys[0]: ({"data": "valid"}, None)}
self.cache.set(values)
# Manually insert corrupted data
corrupted_key = self.cache._make_key(("graph", "node"), "corrupted_key")
self.client.set(corrupted_key, b"invalid:data:format:too:many:colons")
# Should skip corrupted entry and return only valid ones
all_keys: list[FullKey] = [keys[0], (("graph", "node"), "corrupted_key")]
result = self.cache.get(all_keys)
assert len(result) == 1
assert result[keys[0]] == {"data": "valid"}
def test_key_parsing_edge_cases(self) -> None:
"""Test key parsing with edge cases."""
# Test empty namespace
key1: FullKey = ((), "empty_ns")
values = {key1: ({"data": "empty_ns"}, None)}
self.cache.set(values)
result = self.cache.get([key1])
assert result[key1] == {"data": "empty_ns"}
# Test namespace with special characters
key2: FullKey = (
("graph:with:colons", "node-with-dashes"),
"key_with_underscores",
)
values = {key2: ({"data": "special_chars"}, None)}
self.cache.set(values)
result = self.cache.get([key2])
assert result[key2] == {"data": "special_chars"}
def test_large_data_serialization(self) -> None:
"""Test handling of large data objects."""
# Create a large data structure
large_data = {"large_list": list(range(1000)), "nested": {"data": "x" * 1000}}
key: FullKey = (("graph", "node"), "large_key")
values = {key: (large_data, None)}
self.cache.set(values)
result = self.cache.get([key])
assert len(result) == 1
assert result[key] == large_data
+1 -61
View File
@@ -34,42 +34,6 @@ class MockAsyncBatchedStore(AsyncBatchedBaseStore):
return self._store.batch(ops)
async def test_async_batch_store_resilience() -> None:
"""Test that AsyncBatchedBaseStore recovers gracefully from task cancellation."""
doc = {"foo": "bar"}
async_store = MockAsyncBatchedStore()
await async_store.aput(("foo", "langgraph", "foo"), "bar", doc)
# Store the original task reference
original_task = async_store._task
assert original_task is not None
assert not original_task.done()
# Cancel the background task
original_task.cancel()
await asyncio.sleep(0.01)
assert original_task.cancelled()
# Perform a new operation - this should trigger _ensure_task() to create a new task
result = await async_store.asearch(("foo", "langgraph", "foo"))
assert len(result) > 0
assert result[0].value == doc
# Verify a new task was created
new_task = async_store._task
assert new_task is not None
assert new_task is not original_task
assert not new_task.done()
# Test that operations continue to work with the new task
doc2 = {"baz": "qux"}
await async_store.aput(("test", "namespace"), "key", doc2)
result2 = await async_store.aget(("test", "namespace"), "key")
assert result2 is not None
assert result2.value == doc2
def test_get_text_at_path() -> None:
nested_data = {
"name": "test",
@@ -950,8 +914,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
@@ -1021,27 +985,3 @@ async def test_embed_with_path(fake_embeddings: CharacterEmbeddings) -> None:
assert len(results) == 3
doc5_result = next(r for r in results if r.key == "doc5")
assert doc5_result.score is None
def test_non_ascii(fake_embeddings: CharacterEmbeddings) -> None:
"""Test support for non-ascii characters"""
store = InMemoryStore(
index={"dims": fake_embeddings.dims, "embed": fake_embeddings}
)
store.put(("user_123", "memories"), "1", {"text": "这是中文"}) # Chinese
store.put(("user_123", "memories"), "2", {"text": "これは日本語です"}) # Japanese
store.put(("user_123", "memories"), "3", {"text": "이건 한국어야"}) # Korean
store.put(("user_123", "memories"), "4", {"text": "Это русский"}) # Russian
store.put(("user_123", "memories"), "5", {"text": "यह रूसी है"}) # Hindi
result1 = store.search(("user_123", "memories"), query="这是中文")
result2 = store.search(("user_123", "memories"), query="これは日本語です")
result3 = store.search(("user_123", "memories"), query="이건 한국어야")
result4 = store.search(("user_123", "memories"), query="Это русский")
result5 = store.search(("user_123", "memories"), query="यह रूसी है")
assert result1[0].key == "1"
assert result2[0].key == "2"
assert result3[0].key == "3"
assert result4[0].key == "4"
assert result5[0].key == "5"
+544 -576
View File
File diff suppressed because it is too large Load Diff
+1 -2
View File
@@ -4,9 +4,8 @@
# TESTING AND COVERAGE
######################
TEST?= "tests/unit_tests"
test:
uv run pytest $(TEST)
uv run pytest tests/unit_tests
test-integration:
uv run pytest tests/integration_tests
+7
View File
@@ -1,3 +1,10 @@
OPENAI_API_KEY=placeholder
ANTHROPIC_API_KEY=placeholder
TAVILY_API_KEY=placeholder
LANGCHAIN_TRACING_V2=false
LANGCHAIN_ENDPOINT=placeholder
LANGCHAIN_API_KEY=placeholder
LANGCHAIN_PROJECT=placeholder
LANGGRAPH_AUTH_TYPE=noop
LANGSMITH_AUTH_ENDPOINT=placeholder
LANGSMITH_TENANT_ID=placeholder
@@ -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,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,13 +0,0 @@
{
"python_version": "3.12",
"dependencies": [
".",
"./deps/additional_deps",
"./deps/zuper_deps"
],
"graphs": {
"agent": "./agent.py:graph"
},
"env": "../.env"
}
@@ -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"
@@ -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",
]
+7 -8
View File
@@ -7,7 +7,6 @@ from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from langgraph.runtime import Runtime
tools = [TavilySearchResults(max_results=1)]
@@ -18,10 +17,6 @@ model_anth = model_anth.bind_tools(tools)
model_oai = model_oai.bind_tools(tools)
class AgentContext(TypedDict):
model: Literal["anthropic", "openai"]
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
@@ -39,8 +34,8 @@ def should_continue(state):
# Define the function that calls the model
def call_model(state, runtime: Runtime[AgentContext]):
if runtime.context.get("model", "anthropic") == "anthropic":
def call_model(state, config):
if config["configurable"].get("model", "anthropic") == "anthropic":
model = model_anth
else:
model = model_oai
@@ -54,8 +49,12 @@ def call_model(state, runtime: Runtime[AgentContext]):
tool_node = ToolNode(tools)
class ContextSchema(TypedDict):
model: Literal["anthropic", "openai"]
# Define a new graph
workflow = StateGraph(AgentState, context_schema=AgentContext)
workflow = StateGraph(AgentState, context_schema=ContextSchema)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
-1
View File
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"python_version": "3.12",
"dependencies": [
"langchain_community",
@@ -1,6 +1,6 @@
from collections.abc import Sequence
from pathlib import Path
from typing import Annotated, Literal, TypedDict
from typing import Annotated, TypedDict
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
@@ -8,7 +8,6 @@ from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from langgraph.runtime import Runtime
tools = [TavilySearchResults(max_results=1)]
@@ -22,10 +21,6 @@ prompt = open(Path(__file__).parent.parent / "prompt.txt").read()
subprompt = open(Path(__file__).parent / "subprompt.txt").read()
class AgentContext(TypedDict):
model: Literal["anthropic", "openai"]
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
@@ -43,8 +38,8 @@ def should_continue(state):
# Define the function that calls the model
def call_model(state, runtime: Runtime[AgentContext]):
if runtime.context.get("model", "anthropic") == "anthropic":
def call_model(state, config):
if config["configurable"].get("model", "anthropic") == "anthropic":
model = model_anth
else:
model = model_oai
@@ -57,8 +52,9 @@ def call_model(state, runtime: Runtime[AgentContext]):
# Define the function to execute tools
tool_node = ToolNode(tools)
# Define a new graph
workflow = StateGraph(AgentState, context_schema=AgentContext)
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"dependencies": [
"."
],
@@ -1,6 +1,6 @@
from collections.abc import Sequence
from pathlib import Path
from typing import Annotated, Literal, TypedDict
from typing import Annotated, TypedDict
from langchain_anthropic import ChatAnthropic
from langchain_community.tools.tavily_search import TavilySearchResults
@@ -8,7 +8,6 @@ from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from langgraph.runtime import Runtime
tools = [TavilySearchResults(max_results=1)]
@@ -22,10 +21,6 @@ prompt = open(Path(__file__).parent.parent / "prompt.txt").read()
subprompt = open(Path(__file__).parent / "subprompt.txt").read()
class AgentContext(TypedDict):
model: Literal["anthropic", "openai"]
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], add_messages]
@@ -43,8 +38,8 @@ def should_continue(state):
# Define the function that calls the model
def call_model(state, runtime: Runtime[AgentContext]):
if runtime.context.get("model", "anthropic") == "anthropic":
def call_model(state, config):
if config["configurable"].get("model", "anthropic") == "anthropic":
model = model_anth
else:
model = model_oai
@@ -59,7 +54,7 @@ tool_node = ToolNode(tools)
# Define a new graph
workflow = StateGraph(AgentState, context_schema=AgentContext)
workflow = StateGraph(AgentState)
# Define the two nodes we will cycle between
workflow.add_node("agent", call_model)
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"dependencies": [
"."
],
+8 -1
View File
@@ -163,7 +163,14 @@ def generate_schema():
# Add enum constraint for python_version
if "python_version" in python_schema["properties"]:
python_schema["properties"]["python_version"]["enum"] = ["3.11", "3.12", "3.13"]
python_schema["properties"]["python_version"]["enum"] = ["3.11", "3.12"]
# Add enum constraint for image_distro
if "image_distro" in python_schema["properties"]:
python_schema["properties"]["image_distro"]["anyOf"] = [
{"type": "string", "enum": ["debian", "wolfi"]},
{"type": "null"},
]
# Create Node.js schema with node_version
node_schema = {
-1
View File
@@ -1,5 +1,4 @@
{
"$schema": "https://langgra.ph/schema.json",
"node_version": "20",
"graphs": {
"agent": "./src/agent/graph.ts:graph"
@@ -1,62 +0,0 @@
module.exports = {
extends: [
"eslint:recommended",
"prettier",
"plugin:@typescript-eslint/recommended",
],
parserOptions: {
ecmaVersion: 12,
parser: "@typescript-eslint/parser",
project: "./tsconfig.json",
sourceType: "module",
},
plugins: ["import", "@typescript-eslint", "no-instanceof"],
ignorePatterns: [
".eslintrc.cjs",
"scripts",
"src/utils/lodash/*",
"node_modules",
"dist",
"dist-cjs",
"*.js",
"*.cjs",
"*.d.ts",
],
rules: {
"no-process-env": 2,
"no-instanceof/no-instanceof": 2,
"@typescript-eslint/explicit-module-boundary-types": 0,
"@typescript-eslint/no-empty-function": 0,
"@typescript-eslint/no-shadow": 0,
"@typescript-eslint/no-empty-interface": 0,
"@typescript-eslint/no-use-before-define": ["error", "nofunc"],
"@typescript-eslint/no-unused-vars": ["warn", { args: "none" }],
"@typescript-eslint/no-floating-promises": "error",
"@typescript-eslint/no-misused-promises": "error",
camelcase: 0,
"class-methods-use-this": 0,
"import/extensions": [2, "ignorePackages"],
"import/no-extraneous-dependencies": [
"error",
{ devDependencies: ["**/*.test.ts"] },
],
"import/no-unresolved": 0,
"import/prefer-default-export": 0,
"keyword-spacing": "error",
"max-classes-per-file": 0,
"max-len": 0,
"no-await-in-loop": 0,
"no-bitwise": 0,
"no-console": 0,
"no-restricted-syntax": 0,
"no-shadow": 0,
"no-continue": 0,
"no-underscore-dangle": 0,
"no-use-before-define": 0,
"no-useless-constructor": 0,
"no-return-await": 0,
"consistent-return": 0,
"no-else-return": 0,
"new-cap": ["error", { properties: false, capIsNew: false }],
},
};
@@ -1,7 +0,0 @@
{
"node_version": "20",
"graphs": {
"agent": "./src/graph.ts:graph"
},
"env": "../../.env"
}
@@ -1,18 +0,0 @@
{
"name": "@js-monorepo-example/agent",
"version": "0.0.1",
"type": "module",
"main": "src/graph.ts",
"scripts": {
"build": "tsc",
"clean": "rm -rf dist"
},
"dependencies": {
"@js-monorepo-example/shared": "*",
"@langchain/core": "^0.3.2",
"@langchain/langgraph": "^0.2.5"
},
"devDependencies": {
"typescript": "^5.3.3"
}
}
@@ -1,47 +0,0 @@
/**
* Simple LangGraph.js example for monorepo testing
*/
import { StateGraph } from "@langchain/langgraph";
import { RunnableConfig } from "@langchain/core/runnables";
import { StateAnnotation } from "./state.js";
import { getGreeting } from "@js-monorepo-example/shared";
/**
* Simple node that uses the shared library
*/
const callModel = async (
state: typeof StateAnnotation.State,
_config: RunnableConfig,
): Promise<typeof StateAnnotation.Update> => {
// Use functions from the shared library
const greeting = getGreeting();
return {
messages: [
{
role: "assistant",
content: `${greeting}`,
},
],
};
};
/**
* Simple routing function
*/
export const route = (
state: typeof StateAnnotation.State,
): "__end__" | "callModel" => {
if (state.messages.length > 0) {
return "__end__";
}
return "callModel";
};
// Create the graph
const builder = new StateGraph(StateAnnotation)
.addNode("callModel", callModel)
.addEdge("__start__", "callModel")
.addConditionalEdges("callModel", route);
export const graph = builder.compile();

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