Compare commits

..
Author SHA1 Message Date
Arjun Natarajan 91bca2ea1d first pass at data storage docs 2025-06-23 20:22:33 -04:00
lc-arjunandGitHub bd206c2fcd docs: fix assistants links (#5172)
* docs: fix assistants links

* fix another page
2025-06-23 13:48:20 -07:00
lc-arjunandGitHub 14ec895046 docs: Fix LGP sdk typo (#5170)
docs: fix typo in sdk docs
2025-06-23 10:15:15 -07:00
hari-dhanushkodiandGitHub fb66736ccb add more docs for lgp deployment metrics (#5168) 2025-06-23 07:11:57 -07:00
Nuno Campos c3544024b9 If FuturesDict callback has been GCed, don't call it 2025-06-17 14:06:54 -07:00
b0d1234737 docs: missing lgp docs (#5130)
* chore: add docs for lgp deployment monitoring (#5104)

* docs: studio evals (#5129)

* docs: studio evals

* docs: added studio evals images (#5076)

* docs: added studio evals images

* Update docs/docs/cloud/how-tos/studio/run_evals.md

Co-authored-by: lc-arjun <arjun@langchain.dev>

* Update docs/docs/cloud/how-tos/studio/run_evals.md

Co-authored-by: lc-arjun <arjun@langchain.dev>

* Update docs/docs/cloud/how-tos/studio/run_evals.md

Co-authored-by: lc-arjun <arjun@langchain.dev>

* docs: updated studio evals

* Update docs/docs/cloud/how-tos/studio/run_evals.md

Co-authored-by: lc-arjun <arjun@langchain.dev>

* docs: removed images

---------

Co-authored-by: lc-arjun <arjun@langchain.dev>

* final changes

* i think its this

---------

Co-authored-by: Marco Perini <perinim.98@gmail.com>

---------

Co-authored-by: hari-dhanushkodi <hari@langchain.dev>
Co-authored-by: Marco Perini <perinim.98@gmail.com>
2025-06-17 13:10:27 -07:00
Lauren Hirata Singh 53e1a238db Remove cookie consent 2025-06-16 18:42:56 -04:00
langchain-infraandGitHub fcdeafd0d1 docs: fix langgraph docs (#5065)
docs: fix config section
2025-06-11 13:23:16 -04:00
langchain-infraandGitHub 28c529feb2 docs: add mount prefix environment variable (#5061) 2025-06-11 11:21:53 -04:00
infra 91ebc8d3ed docs: add mount prefix environment variable 2025-06-11 11:19:40 -04:00
Eugene YurtsevandGitHub 6e08f4c12e v0: port GTM to v0 (#5056)
This was lost when the v0 branch was cut out and docs started being deployed from v0
2025-06-11 10:21:29 -04:00
Nuno CamposandGitHub f2dc0653f1 docs: list CipherProtocol in API (#5048) 2025-06-10 14:28:39 -07:00
Nuno CamposandNuno Campos 67177a5610 docs: list CipherProtocol in API 2025-06-10 14:25:26 -07:00
Asamu DavidandGitHub b1b238c7ea add docs for image_distro cli option (#4981) 2025-06-06 16:32:21 +01:00
David Asamu 598796ef86 add docs for image_distro cli option 2025-06-06 16:26:25 +01:00
Sydney RunkleandGitHub 3c7981201e docs: remove usage of StateGraph(dict) (#4967)
docs: remove references to `StateGraph(dict)` (#4964)

remove StateGraph(dict)
2025-06-04 21:30:39 -04:00
Sydney RunkleandGitHub 109c0dfb93 docs: deploy from v0 branch for now (#4960) (#4961)
only deploy docs on v0
2025-06-04 13:30:34 -04:00
Nuno Campos d7c364c5bb Port step_timeout/GraphBubbleUp fix to v0
See fix and tests in original PR https://github.com/langchain-ai/langgraph/pull/4950
2025-06-03 17:23:45 -07:00
Nuno Campos 746142fb07 One more 2025-06-02 16:16:11 -07:00
Nuno Campos b9c9c32c31 Allow releases from v0 2025-06-02 16:12:54 -07:00
Nuno Campos c89fe4c45d 0.4.8 2025-06-02 16:11:09 -07:00
Nuno CamposandGitHub b0e28851a6 v0: Fix Command(graph=PARENT) when used together w checkpointer=True (#4920) 2025-06-02 16:10:00 -07:00
Nuno Campos 48fb91deda Fix Command(graph=PARENT) when used together w checkpointer=True 2025-06-02 16:02:35 -07:00
264 changed files with 36178 additions and 21962 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
name: "\U0001F41B Bug Report"
description: Report a bug in LangGraph. To report a security issue, please instead use the security option below. For questions, please use the GitHub Discussions.
labels: [pending,bug]
labels: ["02 Bug Report"]
body:
- type: markdown
attributes:
+1 -1
View File
@@ -1,4 +1,4 @@
blank_issues_enabled: true
blank_issues_enabled: false
version: 2.1
contact_links:
- name: 🤔 Question or Problem
+1 -1
View File
@@ -1,7 +1,7 @@
name: Documentation
description: Report an issue related to the LangGraph documentation.
title: "DOC: <Please write a comprehensive title after the 'DOC: ' prefix>"
labels: [documentation]
labels: [03 - Documentation]
body:
- type: textarea
-11
View File
@@ -1,11 +0,0 @@
# Please see the documentation for all configuration options:
# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
# and
# https://docs.github.com/code-security/dependabot/dependabot-version-updates/configuration-options-for-the-dependabot.yml-file
version: 2
updates:
- package-ecosystem: "github-actions"
directory: "/"
schedule:
interval: "weekly"
+2 -2
View File
@@ -49,7 +49,7 @@ jobs:
- name: Get .mypy_cache to speed up mypy
if: steps.changed-files.outputs.all
uses: actions/cache@v4
uses: actions/cache@v3
env:
SEGMENT_DOWNLOAD_TIMEOUT_MIN: "2"
with:
@@ -75,7 +75,7 @@ jobs:
- name: Get .mypy_cache_test to speed up mypy
if: steps.changed-files.outputs.all
uses: actions/cache@v4
uses: actions/cache@v3
env:
SEGMENT_DOWNLOAD_TIMEOUT_MIN: "2"
with:
+1 -1
View File
@@ -13,7 +13,7 @@ env:
jobs:
build:
if: github.ref == 'refs/heads/main'
if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/v0'
runs-on: ubuntu-latest
outputs:
@@ -0,0 +1,52 @@
name: test
on:
workflow_call:
jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version:
- "3.11"
- "3.12"
defaults:
run:
working-directory: libs/scheduler-kafka
name: "test #${{ matrix.python-version }}"
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: astral-sh/setup-uv@v6
with:
python-version: ${{ matrix.python-version }}
enable-cache: true
cache-suffix: "test-scheduler-kafka"
- name: Login to Docker Hub
uses: docker/login-action@v3
if: ${{ !github.event.pull_request.head.repo.fork }}
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_RO_TOKEN }}
- name: Install dependencies
shell: bash
run: uv sync --frozen --group dev
- name: Run tests
shell: bash
run: make test
- name: Ensure the tests did not create any additional files
shell: bash
run: |
set -eu
STATUS="$(git status)"
echo "$STATUS"
# grep will exit non-zero if the target message isn't found,
# and `set -e` above will cause the step to fail.
echo "$STATUS" | grep 'nothing to commit, working tree clean'
+15 -5
View File
@@ -35,6 +35,7 @@ jobs:
- 'libs/checkpoint/**'
- 'libs/checkpoint-sqlite/**'
- 'libs/checkpoint-postgres/**'
- 'libs/scheduler-kafka/**'
- 'libs/prebuilt/**'
sdk-js:
- 'libs/sdk-js/**'
@@ -52,7 +53,7 @@ jobs:
"libs/checkpoint",
"libs/checkpoint-sqlite",
"libs/checkpoint-postgres",
"libs/scheduler-kafka",
"libs/prebuilt",
]
if: needs.changes.outputs.python == 'true'
@@ -88,6 +89,14 @@ jobs:
uses: ./.github/workflows/_test_langgraph.yml
secrets: inherit
# NOTE: we're testing scheduler-kafka separately because it requires a different matrix
test-scheduler-kafka:
needs: changes
if: needs.changes.outputs.python == 'true'
name: "cd libs/scheduler-kafka"
uses: ./.github/workflows/_test_scheduler_kafka.yml
secrets: inherit
check-sdk-methods:
needs: changes
if: needs.changes.outputs.python == 'true'
@@ -157,9 +166,9 @@ jobs:
run:
working-directory: ${{ matrix.working-directory }}
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: Setup Node.js (LTS)
uses: actions/setup-node@v4
uses: actions/setup-node@v3
with:
node-version: "20"
cache: "yarn"
@@ -183,9 +192,9 @@ jobs:
run:
working-directory: ${{ matrix.working-directory }}
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v3
- name: Setup Node.js (LTS)
uses: actions/setup-node@v4
uses: actions/setup-node@v3
with:
node-version: "20"
cache: "yarn"
@@ -203,6 +212,7 @@ jobs:
lint-js,
test,
test-langgraph,
test-scheduler-kafka,
check-sdk-methods,
check-schema,
integration-test,
+1 -1
View File
@@ -13,7 +13,7 @@ env:
jobs:
build:
if: github.ref == 'refs/heads/main'
if: github.ref == 'refs/heads/main' || github.ref == 'refs/heads/v0'
runs-on: ubuntu-latest
outputs:
+1 -1
View File
@@ -22,7 +22,7 @@ jobs:
- uses: actions/checkout@v4
# JS Build
- name: Use Node.js
uses: actions/setup-node@v4
uses: actions/setup-node@v3
with:
node-version: "20"
cache: "yarn"
-1
View File
@@ -181,4 +181,3 @@ Chinook.db
.vercel
.turbo
.editorconfig
.scratch
-55
View File
@@ -1,55 +0,0 @@
# AGENTS Instructions
This repository is a monorepo. Each library lives in a subdirectory under `libs/`.
When you modify code in any library, run the following commands in that library's directory before creating a pull request:
- `make format` run code formatters
- `make lint` run the linter
- `make test` execute the test suite
To run a particular test file or to pass additional pytest options you can specify the `TEST` variable:
```
TEST=path/to/test.py make test
```
Other pytest arguments can also be supplied inside the `TEST` variable.
## Libraries
The repository contains several Python and JavaScript/TypeScript libraries.
Below is a high-level overview:
- **checkpoint** base interfaces for LangGraph checkpointers.
- **checkpoint-postgres** Postgres implementation of the checkpoint saver.
- **checkpoint-sqlite** SQLite implementation of the checkpoint saver.
- **cli** official command-line interface for LangGraph.
- **langgraph** core framework for building stateful, multi-actor agents.
- **prebuilt** high-level APIs for creating and running agents and tools.
- **sdk-js** JS/TS SDK for interacting with the LangGraph REST API.
- **sdk-py** Python SDK for the LangGraph Platform API.
### Dependency map
The diagram below lists downstream libraries for each production dependency as
declared in that library's `pyproject.toml` (or `package.json`).
```text
checkpoint
├── checkpoint-postgres
├── checkpoint-sqlite
├── prebuilt
└── langgraph
prebuilt
└── langgraph
sdk-py
├── langgraph
└── cli
sdk-js (standalone)
```
Changes to a library may impact all of its dependents shown above.
+1 -1
View File
@@ -109,7 +109,7 @@ Here are some high-level tips on writing a good how-to guide:
LangGraph's conceptual guides fall under the **Explanation** quadrant of Diataxis. They should cover LangChain terms and concepts
in a more abstract way than how-to guides or tutorials, and should be geared towards curious users interested in
gaining a deeper understanding of the framework. Try to avoid excessively large code examples. The goal here is to
impart perspective to the user rather than to finish a practical project. These guides should cover **why** things work the way they do.
impart perspective to the user rather than to finish a practical project. These guides should cover **why** things work they way they do.
To quote the Diataxis website:
-58
View File
@@ -1,58 +0,0 @@
# Define the directories containing projects
LIBS_DIRS := $(wildcard libs/*)
# Default target
.PHONY: all
all: lint format lock test
# Install dependencies for all projects
.PHONY: install
install:
@echo "Creating virtual environment..."
@uv venv
@for dir in $(LIBS_DIRS); do \
if [ -f $$dir/pyproject.toml ]; then \
echo "Installing dependencies for $$dir"; \
uv pip install -e $$dir; \
fi; \
done
# Lint all projects
.PHONY: lint
lint:
@for dir in $(LIBS_DIRS); do \
if [ -f $$dir/Makefile ]; then \
echo "Running lint in $$dir"; \
$(MAKE) -C $$dir lint; \
fi; \
done
# Format all projects
.PHONY: format
format:
@for dir in $(LIBS_DIRS); do \
if [ -f $$dir/Makefile ]; then \
echo "Running format in $$dir"; \
$(MAKE) -C $$dir format; \
fi; \
done
# Lock all projects
.PHONY: lock
lock:
@for dir in $(LIBS_DIRS); do \
if [ -f $$dir/Makefile ]; then \
echo "Running lock in $$dir"; \
(cd $$dir && uv lock); \
fi; \
done
# Test all projects
.PHONY: test
test:
@for dir in $(LIBS_DIRS); do \
if [ -f $$dir/Makefile ]; then \
echo "Running test in $$dir"; \
$(MAKE) -C $$dir test; \
fi; \
done
+1
View File
@@ -12,6 +12,7 @@
[![Downloads](https://static.pepy.tech/badge/langgraph/month)](https://pepy.tech/project/langgraph)
[![Open Issues](https://img.shields.io/github/issues-raw/langchain-ai/langgraph)](https://github.com/langchain-ai/langgraph/issues)
[![Docs](https://img.shields.io/badge/docs-latest-blue)](https://langchain-ai.github.io/langgraph/)
[![GitMCP](https://img.shields.io/endpoint?url=https://gitmcp.io/badge/langchain-ai/langgraph)](https://gitmcp.io/langchain-ai/langgraph)
Trusted by companies shaping the future of agents including Klarna, Replit, Elastic, and more LangGraph is a low-level orchestration framework for building, managing, and deploying long-running, stateful agents.
+122 -119
View File
@@ -1,154 +1,157 @@
"""Translate Python markdown to TypeScript and/or consolidate Python-JS markdown into a single document."""
"""Add typescript translation to a given markdown file."""
import argparse
import re
import requests
from langchain_anthropic import ChatAnthropic
# Load reference TypeScript snippets
URL = "https://gist.githubusercontent.com/eyurtsev/e7486731415463a9bc5b4682358859c8/raw/b5a5fda9c7e3387cfcb781f25082814d43675d50/gistfile1.txt"
response = requests.get(URL)
response.raise_for_status()
reference_snippets = response.text
# Initialize model
model = ChatAnthropic(model="claude-sonnet-4-0", max_tokens=64_000)
TRANSLATION_PROMPT = (
"You are a helpful assistant that translates Python-based technical "
"documentation written in Markdown to equivalent TypeScript-based documentation. "
"The input is a Markdown file written in mkdocs format. It contains "
"Python code snippets embedded in prose. "
"Your task is to rewrite the content by translating the Python code to "
"idiomatic TypeScript, using the provided TypeScript reference snippets "
"to ensure accurate and consistent usage (e.g., correct imports, function "
"names, and patterns). "
"Remove the original Python code and replace it with the corresponding "
"TypeScript version. "
"Do not alter the surrounding prose unless a change is necessary to "
"reflect differences between Python and TypeScript. "
"Preserve the structure and formatting of the original Markdown document. "
"Do not make stylistic or structural changes unless they directly support "
"the translation. "
"Use the reference TypeScript snippets as guidance whenever possible to "
"maintain alignment with existing conventions.\n\n"
f"Here are the reference TypeScript snippets:\n\n{reference_snippets}\n\n"
)
CONSOLIDATION_PROMPT = (
"You are a helpful assistant that consolidates parallel Python and JavaScript (TypeScript) technical documentation "
"written in Markdown into a single unified Markdown document. "
"The input consists of two documents: the first is for Python users, and the second is for JavaScript/TypeScript users. "
"Your task is to merge these into one Markdown file using language-specific fenced blocks to separate the content where needed. "
"Use the following syntax to distinguish content for each language:\n\n"
":::python\n"
"# Python-specific content\n"
":::\n\n"
":::js\n"
"# JavaScript/TypeScript-specific content\n"
":::\n\n"
"Follow these consolidation rules:\n"
"- When content (prose or code) is the same or nearly identical in both versions, include it only once—outside of any fenced block.\n"
"- When content differs between the Python and JS versions, wrap each version in its corresponding fenced block.\n"
"- Prefer **paragraph-level separation** of language-specific content. Do not combine Python and JS snippets or terminology in the same sentence or paragraph using conditional phrases.\n"
" For example, avoid inline constructs like:\n"
" `The :::python add_messages ::: :::js reducer ::: function...`\n"
" Instead, write two distinct paragraphs:\n\n"
" :::python\n"
" The `add_messages` function in our `State` will append the LLM's response messages to whatever messages are already in the state.\n"
" ::: \n\n"
" :::js\n"
" The `reducer` function in our `StateAnnotation` will append the LLM's response messages to whatever messages are already in the state.\n"
" :::\n\n"
"- Preserve the overall structure, ordering, and formatting of the original Markdown documents.\n"
"- Do not rephrase or unify content unless it is logically and semantically identical.\n"
"- Use the fenced blocks for both prose and code as needed, and ensure output is clean, readable Markdown suitable for tools that parse these directives.\n"
"Your goal is to produce a cleanly merged documentation file that serves both Python and JavaScript users without redundancy, while maximizing clarity and separation of language-specific details."
)
model = ChatAnthropic(model="claude-3-5-sonnet-latest")
def translate_python_to_ts(markdown_content: str) -> str:
response = model.invoke(
def _get_tqdm():
try:
from tqdm import tqdm
except ImportError:
# If not available return a simple identity function
def tqdm(iterable, *args, **kwargs):
return iterable
return tqdm
_tqdm = _get_tqdm()
opening_pattern = re.compile(r"^\s*```python(?:\s+.*)?\s*$")
closing_pattern = re.compile(r"^\s*```\s*$")
def extract_python_snippets(markdown: str) -> list[str]:
"""
Extract all python code blocks (including their fence lines) from the markdown content.
A python block is defined as any block that starts with a line containing an opening fence
with '```python' (optionally with extra parameters) and ends with a closing fence '```'.
"""
snippets = []
inside_block = False
current_snippet = []
for line in markdown.splitlines(keepends=True):
if not inside_block:
if opening_pattern.match(line):
inside_block = True
current_snippet = [line]
else:
current_snippet.append(line)
if closing_pattern.match(line):
inside_block = False
snippets.append("".join(current_snippet))
current_snippet = []
return snippets
def translate_snippet(python_snippet: str) -> str:
"""Translate a python code block into a TypeScript code block using Langchain.
The response is expected to be a properly fenced TypeScript code block (i.e.
starting with ```typescript and ending with ```).
"""
ai_message = model.invoke(
[
{
"role": "system",
"content": TRANSLATION_PROMPT,
"cache_control": {"type": "ephemeral"},
"content": (
f"You have access to the following up-to-date example TypeScript code "
f"snippets that show examples of building with langgraph "
f"and langchain:\n\n{reference_snippets}\n\n"
"Use this context to translate the following Python code to equivalent "
"TypeScript. Ensure that your output is a valid fenced TypeScript "
"code block (i.e. starts with ```typescript and ends with ```)."
),
},
{"role": "user", "content": markdown_content},
]
)
return response.content
def consolidate_python_and_ts(combined_content: str) -> str:
response = model.invoke(
[
{
"role": "system",
"content": CONSOLIDATION_PROMPT,
"cache_control": {"type": "ephemeral"},
"role": "user",
"content": f"Translate this Python snippet to TypeScript:\n\n{python_snippet}",
},
{"role": "user", "content": combined_content},
]
)
return response.content
# Use a regular expression to search for a TypeScript code block in the response.
pattern = r"```typescript\s*(.*?)\s*```"
match = re.search(pattern, ai_message.content, re.DOTALL)
if match:
# Reconstruct the code block with proper fences.
typescript_code = match.group(1).strip()
return f"```typescript\n{typescript_code}\n```"
else:
raise ValueError("No TypeScript code block found in the model's response.")
def main(file_path: str, translate_only: bool, consolidate_only: bool) -> None:
with open(file_path, "r", encoding="utf-8") as f:
def insert_translations_into_markdown(
markdown: str, typescript_snippets: list[str]
) -> str:
"""Walks through the original markdown content and, after each
Python snippet block, inserts the corresponding translated TypeScript snippet.
It assumes that the ordering of the Python snippets
(from extract_python_snippets) matches the order they appear in the markdown.
"""
output_lines = []
lines = markdown.splitlines(keepends=True)
inside_block = False
snippet_index = 0
for line in lines:
output_lines.append(line)
if not inside_block and opening_pattern.match(line):
# We've encountered the start of a python code block.
inside_block = True
elif inside_block:
if closing_pattern.match(line):
# End of a python snippet block.
inside_block = False
if snippet_index < len(typescript_snippets):
# Insert an extra newline for clarity, then the translated TypeScript snippet.
output_lines.append("\n")
output_lines.append(typescript_snippets[snippet_index])
output_lines.append("\n")
snippet_index += 1
return "".join(output_lines)
def main(file_path: str) -> None:
# Read the markdown file.
with open(file_path, "r") as f:
markdown_content = f.read()
if translate_only:
translated = translate_python_to_ts(markdown_content)
output_path = file_path.replace(".md", ".translated.md")
with open(output_path, "w", encoding="utf-8") as f:
f.write(translated)
print(f"Translated JS/TS version written to: {output_path}")
# 1. Extract all Python snippets.
python_snippets = extract_python_snippets(markdown_content)[:1]
elif consolidate_only:
consolidated = consolidate_python_and_ts(markdown_content)
with open(file_path, "w", encoding="utf-8") as f:
f.write(consolidated)
print(f"Consolidated content written to: {file_path}")
# 2. Translate each Python snippet to TypeScript.
typescript_snippets = []
# Replace with .batch() for faster translation
for python_snippet in _tqdm(python_snippets):
ts_snippet = translate_snippet(python_snippet)
typescript_snippets.append(ts_snippet)
else:
# Default behavior: translate first, then consolidate both
translated = translate_python_to_ts(markdown_content)
combined = f"{markdown_content.strip()}\n\n\n{translated.strip()}"
consolidated = consolidate_python_and_ts(combined)
with open(file_path, "w", encoding="utf-8") as f:
f.write(consolidated)
print(f"Translated and consolidated content written to: {file_path}")
# 3. Insert the TypeScript translations after their respective Python snippets.
updated_markdown = insert_translations_into_markdown(
markdown_content, typescript_snippets
)
# Overwrite the original markdown file with the updated content.
with open(file_path, "w") as f:
f.write(updated_markdown)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description=(
"Translate Python markdown to TypeScript and/or consolidate "
"Python-JS markdown into one file."
)
description="Translate Python snippets in a markdown file to TypeScript and insert them after each Python snippet."
)
parser.add_argument("file_path", type=str, help="Path to the markdown file.")
parser.add_argument(
"--translate-only",
action="store_true",
help="Only generate the JS translation.",
)
parser.add_argument(
"--consolidate-only",
action="store_true",
help="Only consolidate pre-paired Python and JS content.",
)
args = parser.parse_args()
if args.translate_only and args.consolidate_only:
raise ValueError(
"Cannot use both --translate-only and --consolidate-only at the same time."
)
main(
args.file_path,
translate_only=args.translate_only,
consolidate_only=args.consolidate_only,
)
main(args.file_path)
+6 -10
View File
@@ -3,21 +3,19 @@
import asyncio
import glob
import os
import re
from typing import TypedDict, List, Optional
import pydantic
import re
from pydantic import BaseModel, Field
from langchain_core.rate_limiters import InMemoryRateLimiter
import yaml
from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
from mkdocs.structure.files import File
from mkdocs.structure.pages import Page
from pydantic import BaseModel, Field
from yaml import SafeLoader
from _scripts.notebook_hooks import (
_on_page_markdown_with_config,
_apply_conditional_rendering,
)
from _scripts.notebook_hooks import _on_page_markdown_with_config
HERE = os.path.dirname(os.path.abspath(__file__))
# Get source directory (parent of HERE / docs)
@@ -213,9 +211,7 @@ async def process_nav_items(nav_items: list[NavItem]) -> list[NavItem]:
# Remove any items that start with http:// or https:// looking only for
# local file at this stages.
nav_items = [
item
for item in nav_items
if not item["url"].startswith(("http://", "https://"))
item for item in nav_items if not item["url"].startswith(("http://", "https://"))
]
# Process items in parallel
tasks = [process_single_item(item) for item in nav_items]
-6
View File
@@ -1,6 +0,0 @@
JS_LINK_MAP = {
"langgraph.types.interrupt": "https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph.interrupt-2.html",
"create_react_agent": "https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph_prebuilt.createReactAgent.html",
"langgraph.types.Command": "https://langchain-ai.github.io/langgraphjs/reference/classes/langgraph.Command.html",
}
-71
View File
@@ -16,7 +16,6 @@ from mkdocs.structure.pages import Page
from _scripts.generate_api_reference_links import update_markdown_with_imports
from _scripts.notebook_convert import convert_notebook
from _scripts.link_map import JS_LINK_MAP
logger = logging.getLogger(__name__)
logging.basicConfig()
@@ -159,62 +158,6 @@ def _add_path_to_code_blocks(markdown: str, page: Page) -> str:
return code_block_pattern.sub(replace_code_block_header, markdown)
def _resolve_cross_references(md_text: str, link_map: dict[str, str]) -> str:
"""Replace [title][identifier] with [title](url) using language-specific link_map.
Args:
md_text: The markdown text to process.
link_map: mapping of identifier to URL.
Returns:
The processed markdown text with cross-references resolved.
"""
# Pattern to match [title][identifier]
pattern = re.compile(r"\[([^\]]+)\]\[([^\]]+)\]")
def replace_reference(match: re.Match) -> str:
"""Replace the matched reference with the corresponding URL."""
title, identifier = match.group(1), match.group(2)
url = link_map.get(identifier)
if url:
return f"[{title}]({url})"
else:
# Leave it unchanged if not found
return match.group(0)
return pattern.sub(replace_reference, md_text)
def _apply_conditional_rendering(md_text: str, target_language: str) -> str:
if target_language not in {"python", "js"}:
raise ValueError("target_language must be 'python' or 'js'")
pattern = re.compile(
r"(?P<indent>[ \t]*):::(?P<language>\w+)\s*\n"
r"(?P<content>((?:.*\n)*?))" # Capture the content inside the block
r"(?P=indent):::" # Match closing with the same indentation
)
def replace_conditional_blocks(match: re.Match) -> str:
"""Keep active conditionals."""
language = match.group("language")
content = match.group("content")
if language not in {"python", "js"}:
# If the language is not supported, return the original block
return match.group(0)
if language == target_language:
return content
# If the language does not match, return an empty string
return ""
processed = pattern.sub(replace_conditional_blocks, md_text)
return processed
def _highlight_code_blocks(markdown: str) -> str:
"""Find code blocks with highlight comments and add hl_lines attribute.
@@ -314,20 +257,6 @@ def _on_page_markdown_with_config(
# Apply highlight comments to code blocks
markdown = _highlight_code_blocks(markdown)
# Apply conditional rendering for code blocks
target_language = kwargs.get("target_language", "js")
markdown = _apply_conditional_rendering(markdown, target_language)
if target_language == "js":
markdown = _resolve_cross_references(markdown, JS_LINK_MAP)
elif target_language == "python":
# Via a dedicated plugin
pass
else:
raise ValueError(
f"Unsupported target language: {target_language}. "
"Supported languages are 'python' and 'js'."
)
# Add file path as an attribute to code blocks that are executable.
# This file path is used to associate fixtures with the executable code
# which can be used in CI to test the docs without making network requests.
-243
View File
@@ -11,8 +11,6 @@ hide:
This guide shows you how to set up and use LangGraph's **prebuilt**, **reusable** components, which are designed to help you construct agentic systems quickly and reliably.
:::python
## Prerequisites
Before you start this tutorial, ensure you have the following:
@@ -230,244 +228,3 @@ response["structured_response"]
- [Deploy your agent locally](../tutorials/langgraph-platform/local-server.md)
- [Learn more about prebuilt agents](../agents/overview.md)
- [LangGraph Platform quickstart](../cloud/quick_start.md)
:::
:::js
## Prerequisites
Before you start this tutorial, ensure you have the following:
- An [Anthropic](https://console.anthropic.com/settings/keys) API key
## 1. Install dependencies
If you haven't already, install LangGraph and LangChain:
```
npm install langchain @langchain/langgraph @langchain/anthropic
```
## 2. Create an agent
Use [`createReactAgent`][create_react_agent] to instantiate an agent:
```ts
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { initChatModel } from "langchain/chat_models/universal";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const getWeather = tool( // (1)!
async (input: { city: string }) => {
return `It's always sunny in ${input.city}!`;
},
{
name: "getWeather",
schema: z.object({
city: z.string().describe("The city to get the weather for"),
}),
description: "Get weather for a given city.",
}
);
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest"); // (2)!
const agent = createReactAgent({
llm,
tools: [getWeather], // (3)!
prompt: "You are a helpful assistant", // (4)!
});
// Run the agent
await agent.invoke({
messages: [{ role: "user", content: "what is the weather in sf" }],
});
```
1. Define a tool for the agent to use. For more advanced tool usage and customization, check the [tools](./tools.md) page.
2. Provide a language model for the agent to use. To learn more about configuring language models for the agents, check the [models](./models.md) page.
3. Provide a list of tools for the model to use.
4. Provide a system prompt (instructions) to the language model used by the agent.
## 3. Configure an LLM
Use [`initChatModel`](https://api.js.langchain.com/functions/langchain.chat_models_universal.initChatModel.html) to configure an LLM with specific parameters, such as temperature:
```ts
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { initChatModel } from "langchain/chat_models/universal";
// highlight-next-line
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest", {
// highlight-next-line
temperature: 0,
});
const agent = createReactAgent({
// highlight-next-line
llm,
tools: [getWeather],
});
```
See the [models](./models.md) page for more information on how to configure LLMs.
## 4. Add a custom prompt
Prompts instruct the LLM how to behave. They can be:
- **Static**: A string is interpreted as a **system message**
- **Dynamic**: a list of messages generated at **runtime** based on input or configuration
=== "Static prompt"
Define a fixed prompt string or list of messages.
```ts
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { initChatModel } from "langchain/chat_models/universal";
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getWeather],
// A static prompt that never changes
// highlight-next-line
prompt: "Never answer questions about the weather.",
});
await agent.invoke({
messages: "what is the weather in sf",
});
```
=== "Dynamic prompt"
Define a function that returns a message list based on the agent's state and configuration:
```ts
import { BaseMessageLike } from "@langchain/core/messages";
import { RunnableConfig } from "@langchain/core/runnables";
import { initChatModel } from "langchain/chat_models/universal";
import { MessagesAnnotation } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
const prompt = (
state: typeof MessagesAnnotation.State,
config: RunnableConfig
): BaseMessageLike[] => { // (1)!
const userName = config.configurable?.userName;
const systemMsg = `You are a helpful assistant. Address the user as ${userName}.`;
return [{ role: "system", content: systemMsg }, ...state.messages];
};
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getWeather],
// highlight-next-line
prompt,
});
await agent.invoke(
{ messages: [{ role: "user", content: "what is the weather in sf" }] },
// highlight-next-line
{ configurable: { userName: "John Smith" } }
);
```
1. Dynamic prompts allow including non-message [context](./context.md) when constructing an input to the LLM, such as:
- Information passed at runtime, like a `userId` or API credentials (using `config`).
- Internal agent state updated during a multi-step reasoning process (using `state`).
Dynamic prompts can be defined as functions that take `state` and `config` and return a list of messages to send to the LLM.
For more information, see [Context](./context.md).
## 5. Add memory
To allow multi-turn conversations with an agent, you need to enable [persistence](../concepts/persistence.md) by providing a `checkpointer` when creating an agent. At runtime you need to provide a config containing `thread_id` — a unique identifier for the conversation (session):
```ts
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { MemorySaver } from "@langchain/langgraph-checkpoint";
import { initChatModel } from "langchain/chat_models/universal";
// highlight-next-line
const checkpointer = new MemorySaver();
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getWeather],
// highlight-next-line
checkpointer, // (1)!
});
// Run the agent
// highlight-next-line
const config = { configurable: { thread_id: "1" } };
const sfResponse = await agent.invoke(
{ messages: [{ role: "user", content: "what is the weather in sf" }] },
config // (2)!
);
const nyResponse = await agent.invoke(
{ messages: [{ role: "user", content: "what about new york?" }] },
config
);
```
1. `checkpointer` allows the agent to store its state at every step in the tool calling loop. This enables [short-term memory](./memory.md#short-term-memory) and [human-in-the-loop](./human-in-the-loop.md) capabilities.
2. Pass configuration with `thread_id` to be able to resume the same conversation on future agent invocations.
When you enable the checkpointer, it stores agent state at every step in the provided checkpointer database (or in memory, if using `InMemorySaver`).
Note that in the above example, when the agent is invoked the second time with the same `thread_id`, the original message history from the first conversation is automatically included, together with the new user input.
For more information, see [Memory](./memory.md).
## 6. Configure structured output
To produce structured responses conforming to a schema, use the `responseFormat` parameter. The schema can be defined with a `zod` schema. The result will be accessible via the `structuredResponse` field.
```ts
import { z } from "zod";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { initChatModel } from "langchain/chat_models/universal";
const WeatherResponse = z.object({
conditions: z.string(),
});
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getWeather],
// highlight-next-line
responseFormat: WeatherResponse, // (1)!
});
const response = await agent.invoke({
messages: [{ role: "user", content: "what is the weather in sf" }],
});
// highlight-next-line
response.structuredResponse;
```
1. When `responseFormat` is provided, a separate step is added at the end of the agent loop: agent message history is passed to an LLM with structured output to generate a structured response.
To provide a system prompt to this LLM, use an object `{ prompt, schema }`, e.g., `responseFormat: { prompt, schema: WeatherResponse }`.
!!! Note "LLM post-processing"
Structured output requires an additional call to the LLM to format the response according to the schema.
## Next steps
- [Deploy your agent locally](../tutorials/langgraph-platform/local-server.md)
- [Learn more about prebuilt agents](../agents/overview.md)
- [LangGraph Platform quickstart](../cloud/quick_start.md)
:::
+1 -282
View File
@@ -43,8 +43,6 @@ when you have values that don't change mid-run.
Specify configuration using a key called **"configurable"** which is reserved
for this purpose:
:::python
```python
agent.invoke(
{"messages": [{"role": "user", "content": "hi!"}]},
@@ -52,23 +50,11 @@ agent.invoke(
config={"configurable": {"user_id": "user_123"}}
)
```
:::
:::js
```ts
await agent.invoke(
{ messages: "hi!" },
// highlight-next-line
{ configurable: { userId: "user_123" } }
)
```
:::
### State (mutable context)
State acts as short-term memory during a run. It holds dynamic data that can evolve during execution, such as values derived from tools or LLM outputs.
:::python
```python
class CustomState(AgentState):
# highlight-next-line
@@ -85,29 +71,6 @@ agent.invoke({
"user_name": "Jane"
})
```
:::
:::js
```ts
const CustomState = Annotation.Root({
...MessagesAnnotation.spec,
userName: Annotation<string>,
});
const agent = createReactAgent({
// Other agent parameters...
// highlight-next-line
stateSchema: CustomState,
})
await agent.invoke(
// highlight-next-line
{ messages: "hi!", userName: "Jane" }
)
```
:::
!!! tip "Turning on memory"
@@ -130,8 +93,6 @@ Common use cases:
- Role or goal customization
- Conditional behavior (e.g., user is admin)
:::python
=== "Using config"
```python
@@ -201,90 +162,8 @@ Common use cases:
})
```
:::
:::js
=== "Using config"
```ts
import { BaseMessageLike } from "@langchain/core/messages";
import { RunnableConfig } from "@langchain/core/runnables";
import { initChatModel } from "langchain/chat_models/universal";
import { MessagesAnnotation } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
const prompt = (
state: typeof MessagesAnnotation.State,
// highlight-next-line
config: RunnableConfig
): BaseMessageLike[] => {
// highlight-next-line
const userName = config.configurable?.userName;
const systemMsg = `You are a helpful assistant. Address the user as ${userName}.`;
return [{ role: "system", content: systemMsg }, ...state.messages];
};
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getWeather],
// highlight-next-line
prompt
});
await agent.invoke(
{ messages: "hi!" },
// highlight-next-line
{ configurable: { userName: "John Smith" } }
);
```
=== "Using state"
```ts
import { BaseMessageLike } from "@langchain/core/messages";
import { RunnableConfig } from "@langchain/core/runnables";
import { initChatModel } from "langchain/chat_models/universal";
import { Annotation, MessagesAnnotation } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
const CustomState = Annotation.Root({
...MessagesAnnotation.spec,
// highlight-next-line
userName: Annotation<string>,
});
const prompt = (
// highlight-next-line
state: typeof CustomState.State,
): BaseMessageLike[] => {
// highlight-next-line
const userName = state.userName;
const systemMsg = `You are a helpful assistant. Address the user as ${userName}.`;
return [{ role: "system", content: systemMsg }, ...state.messages];
};
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getWeather],
// highlight-next-line
prompt,
// highlight-next-line
stateSchema: CustomState,
});
await agent.invoke(
// highlight-next-line
{ messages: "hi!", userName: "John Smith" },
);
```
:::
## Accessing Context in Tools { #tools }
:::python
Tools can access context through special parameter **annotations**.
* Use `RunnableConfig` for config access
@@ -351,167 +230,7 @@ Tools can access context through special parameter **annotations**.
"user_id": "user_123"
})
```
:::
:::js
Tools can access context through:
* Use `RunnableConfig` for config access
* Use `getCurrentTaskInput()` for agent state
=== "Using config"
```ts
import { RunnableConfig } from "@langchain/core/runnables";
import { initChatModel } from "langchain/chat_models/universal";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const getUserInfo = tool(
async (input: Record<string, any>, config: RunnableConfig) => {
// highlight-next-line
const userId = config.configurable?.userId;
return userId === "user_123" ? "User is John Smith" : "Unknown user";
},
{
name: "get_user_info",
description: "Look up user info.",
schema: z.object({}),
}
);
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getUserInfo],
});
await agent.invoke(
{ messages: "look up user information" },
// highlight-next-line
{ configurable: { userId: "user_123" } }
);
```
=== "Using state"
```ts
import { initChatModel } from "langchain/chat_models/universal";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { Annotation, MessagesAnnotation, getCurrentTaskInput } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const CustomState = Annotation.Root({
...MessagesAnnotation.spec,
// highlight-next-line
userId: Annotation<string>(),
});
const getUserInfo = tool(
async (
input: Record<string, any>,
) => {
// highlight-next-line
const state = getCurrentTaskInput() as typeof CustomState.State;
// highlight-next-line
const userId = state.userId;
return userId === "user_123" ? "User is John Smith" : "Unknown user";
},
{
name: "get_user_info",
description: "Look up user info.",
schema: z.object({})
}
);
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getUserInfo],
// highlight-next-line
stateSchema: CustomState,
});
await agent.invoke(
// highlight-next-line
{ messages: "look up user information", userId: "user_123" }
);
```
:::
### Update Context from Tools
:::python
Tools can update agent's context (state and long-term memory) during execution. This is useful for persisting intermediate results or making information accessible to subsequent tools or prompts. See [Memory](./memory.md#read-short-term) guide for more information.
:::
:::js
Tools can modify the agent's state during execution. This is useful for persisting intermediate results or making information accessible to subsequent tools or prompts.
```ts
import { Annotation, MessagesAnnotation, LangGraphRunnableConfig, Command } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { ToolMessage } from "@langchain/core/messages";
import { initChatModel } from "langchain/chat_models/universal";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
const CustomState = Annotation.Root({
...MessagesAnnotation.spec,
// highlight-next-line
userName: Annotation<string>(), // Will be updated by the tool
});
const getUserInfo = tool(
async (
_input: Record<string, never>,
config: LangGraphRunnableConfig
): Promise<Command> => {
const userId = config.configurable?.userId;
if (!userId) {
throw new Error("Please provide a user id in config.configurable");
}
const toolCallId = config.toolCall?.id;
const name = userId === "user_123" ? "John Smith" : "Unknown user";
// Return command to update state
return new Command({
update: {
// highlight-next-line
userName: name,
// Update the message history
// highlight-next-line
messages: [
new ToolMessage({
content: "Successfully looked up user information",
tool_call_id: toolCallId,
}),
],
},
});
},
{
name: "get_user_info",
description: "Look up user information.",
schema: z.object({}),
}
);
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [getUserInfo],
// highlight-next-line
stateSchema: CustomState,
});
await agent.invoke(
{ messages: "look up user information" },
// highlight-next-line
{ configurable: { userId: "user_123" } }
);
```
:::
Tools can update agent's context (state and long-term memory) during execution. This is useful for persisting intermediate results or making information accessible to subsequent tools or prompts. See [Memory](./memory.md#read-short-term) guide for more information.
-107
View File
@@ -27,8 +27,6 @@ A human can review and edit the output from the agent before proceeding. This is
</figure>
:::python
## Review tool calls
To add a human approval step to a tool:
@@ -36,7 +34,6 @@ To add a human approval step to a tool:
1. Use `interrupt()` in the tool to pause execution.
2. Resume with a `Command(resume=...)` to continue based on human input.
```python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt
@@ -236,110 +233,6 @@ for chunk in agent.stream(
print("\n")
```
:::
:::js
## Review tool calls
To add a human approval step to a tool:
1. Use `interrupt()` in the tool to pause execution.
2. Resume with a `Command({ resume: ... })` to continue based on human input.
```ts
import { MemorySaver } from "@langchain/langgraph-checkpoint";
import { interrupt } from "@langchain/langgraph";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { initChatModel } from "langchain/chat_models/universal";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
// An example of a sensitive tool that requires human review / approval
const bookHotel = tool(
async (input: { hotelName: string; }) => {
let hotelName = input.hotelName;
// highlight-next-line
const response = interrupt( // (1)!
`Trying to call \`book_hotel\` with args {'hotel_name': ${hotelName}}. ` +
`Please approve or suggest edits.`
)
if (response.type === "accept") {
// proceed to execute the tool logic
} else if (response.type === "edit") {
hotelName = response.args["hotel_name"]
} else {
throw new Error(`Unknown response type: ${response.type}`)
}
return `Successfully booked a stay at ${hotelName}.`;
},
{
name: "bookHotel",
schema: z.object({
hotelName: z.string().describe("Hotel to book"),
}),
description: "Book a hotel.",
}
);
// highlight-next-line
const checkpointer = new MemorySaver(); // (2)!
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
tools: [bookHotel],
// highlight-next-line
checkpointer // (3)!
});
```
1. The [`interrupt` function][langgraph.types.interrupt] pauses the agent graph at a specific node. In this case, we call `interrupt()` at the beginning of the tool function, which pauses the graph at the node that executes the tool. The information inside `interrupt()` (e.g., tool calls) can be presented to a human, and the graph can be resumed with the user input (tool call approval, edit or feedback).
2. The `InMemorySaver` is used to store the agent state at every step in the tool calling loop. This enables [short-term memory](./memory.md#short-term-memory) and [human-in-the-loop](./human-in-the-loop.md) capabilities. In this example, we use `InMemorySaver` to store the agent state in memory. In a production application, the agent state will be stored in a database.
3. Initialize the agent with the `checkpointer`.
Run the agent with the `stream()` method, passing the `config` object to specify the thread ID. This allows the agent to resume the same conversation on future invocations.
```ts
const config = {
configurable: {
// highlight-next-line
"thread_id": "1"
}
}
for await (const chunk of await agent.stream(
{ messages: "book a stay at McKittrick hotel" },
// highlight-next-line
config
)) {
console.log(chunk);
console.log("\n");
};
```
> You should see that the agent runs until it reaches the `interrupt()` call, at which point it pauses and waits for human input.
Resume the agent with a `Command({ resume: ... })` to continue based on human input.
```ts
import { Command } from "@langchain/langgraph";
for await (const chunk of await agent.stream(
new Command({ resume: { type: "accept" } }), // (1)!
// new Command({ resume: { type: "edit", args: { "hotel_name": "McKittrick Hotel" } } }),
// highlight-next-line
config
)) {
console.log(chunk);
console.log("\n");
};
```
1. The [`interrupt` function][langgraph.types.interrupt] is used in conjunction with the [`Command`][langgraph.types.Command] object to resume the graph with a value provided by the human.
:::
## Additional resources
* [Human-in-the-loop in LangGraph](../concepts/human_in_the_loop.md)
+1 -54
View File
@@ -13,8 +13,6 @@ hide:
![MCP](./assets/mcp.png)
:::python
Install the `langchain-mcp-adapters` library to use MCP tools in LangGraph:
```bash
@@ -40,7 +38,7 @@ client = MultiServerMCPClient(
"transport": "stdio",
},
"weather": {
# Ensure you start your weather server on port 8000
# Ensure your start your weather server on port 8000
"url": "http://localhost:8000/mcp",
"transport": "streamable_http",
}
@@ -60,57 +58,6 @@ weather_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
)
```
:::
:::js
Install the `@langchain/mcp-adapters` library to use MCP tools in LangGraph:
```bash
npm install @langchain/mcp-adapters
```
## Use MCP tools
The `@langchain/mcp-adapters` package enables agents to use tools defined across one or more MCP servers.
```ts
// highlight-next-line
import { MultiServerMCPClient } from "@langchain/mcp-adapters";
import { initChatModel } from "langchain/chat_models/universal";
import { createReactAgent } from "@langchain/langgraph/prebuilt";
// highlight-next-line
const client = new MultiServerMCPClient({
mcpServers: {
"math": {
command: "python",
// Replace with absolute path to your math_server.py file
args: ["/path/to/math_server.py"],
transport: "stdio",
},
"weather": {
// Ensure your start your weather server on port 8000
url: "http://localhost:8000/sse",
transport: "sse",
}
}
})
const llm = await initChatModel("anthropic:claude-3-7-sonnet-latest");
const agent = createReactAgent({
llm,
// highlight-next-line
tools: await client.getTools()
});
const mathResponse = await agent.invoke(
{ messages: [ { role: "user", content: "what's (3 + 5) x 12?" } ] }
);
const weatherResponse = await agent.invoke(
{ messages: [ { role: "user", content: "what is the weather in nyc?" } ] }
);
await client.close();
```
:::
## Custom MCP servers
+1 -1
View File
@@ -88,7 +88,7 @@ ny_response = agent.invoke(
When the agent is invoked the second time with the same `thread_id`, the original message history from the first conversation is automatically included, allowing the agent to infer that the user is asking specifically about the **weather** in New York.
!!! Note "LangGraph Platform provides a production-ready checkpointer"
!!! Note "LangGraph Platform providers a production-ready checkpointer"
If you're using [LangGraph Platform](./deployment.md), during deployment your checkpointer will be automatically configured to use a production-ready database.
+1 -1
View File
@@ -9,7 +9,7 @@ hide:
# Multi-agent
A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and compose them into a [multi-agent system](../concepts/multi_agent.md).
A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and composing them into a [multi-agent system](../concepts/multi_agent.md).
In multi-agent systems, agents need to communicate between each other. They do so via [handoffs](#handoffs) — a primitive that describes which agent to hand control to and the payload to send to that agent.
-160
View File
@@ -40,8 +40,6 @@ LangGraph comes with a set of prebuilt components that implement common agent be
Using LangGraph for agent development allows you to focus on your application's logic and behavior, instead of building and maintaining the supporting infrastructure for state, memory, and human feedback.
:::python
## Package ecosystem
The high-level components are organized into several packages, each with a specific focus.
@@ -191,161 +189,3 @@ function initializeWidget() {
window.addEventListener("DOMContentLoaded", initializeWidget);
document$.subscribe(initializeWidget);
</script>
:::
:::js
## Package ecosystem
The high-level components are organized into several packages, each with a specific focus.
| Package | Description | Installation |
|--------------------------|-----------------------------------------------------------------------------|----------------------------------------------------|
| `langgraph` | Prebuilt components to [**create agents**](./agents.md) | `npm install @langchain/langgraph @langchain/core` |
| `langgraph-supervisor` | Tools for building [**supervisor**](./multi-agent.md#supervisor) agents | `npm install @langchain/langgraph-supervisor` |
| `langgraph-swarm` | Tools for building a [**swarm**](./multi-agent.md#swarm) multi-agent system | `npm install @langchain/langgraph-swarm` |
| `langchain-mcp-adapters` | Interfaces to [**MCP servers**](./mcp.md) for tool and resource integration | `npm install @langchain/mcp-adapters` |
| `agentevals` | Utilities to [**evaluate agent performance**](./evals.md) | `npm install agentevals` |
## Visualize an agent graph
Use the following tool to visualize the graph generated by [`createReactAgent`][create_react_agent] and to view an outline of the corresponding code. It allows you to explore the infrastructure of the agent as defined by the presence of:
- [`tools`](./tools.md): A list of tools (functions, APIs, or other callable objects) that the agent can use to perform tasks.
- `preModelHook`: A function that is called before the model is invoked. It can be used to condense messages or perform other preprocessing tasks.
- `postModelHook`: A function that is called after the model is invoked. It can be used to implement guardrails, human-in-the-loop flows, or other postprocessing tasks.
- [`responseFormat`](./agents.md#6-configure-structured-output): A data structure used to constrain the type of the final output (via Zod schemas).
<div class="agent-layout">
<div class="agent-graph-features-container">
<div class="agent-graph-features">
<h3 class="agent-section-title">Features</h3>
<label><input type="checkbox" id="tools" checked> <code>tools</code></label>
<label><input type="checkbox" id="preModelHook"> <code>preModelHook</code></label>
<label><input type="checkbox" id="postModelHook"> <code>postModelHook</code></label>
<label><input type="checkbox" id="responseFormat"> <code>responseFormat</code></label>
</div>
</div>
<div class="agent-graph-container">
<h3 class="agent-section-title">Graph</h3>
<img id="agent-graph-img" src="../assets/react_agent_graphs/0001.svg" alt="graph image" style="max-width: 100%;"/>
</div>
</div>
The following code snippet shows how to create the above agent (and underlying graph) with [`createReactAgent`][create_react_agent]:
```typescript
<div class="language-typescript">
<pre><code id="agent-code" class="language-typescript"></code></pre>
</div>
<script>
function getCheckedValue(id) {
return document.getElementById(id).checked ? "1" : "0";
}
function getKey() {
return [
getCheckedValue("responseFormat"),
getCheckedValue("postModelHook"),
getCheckedValue("preModelHook"),
getCheckedValue("tools")
].join("");
}
function dedent(strings, ...values) {
const str = String.raw({ raw: strings }, ...values)
const [space] = str.split("\n").filter(Boolean).at(0).match(/^(\s*)/)
const spaceLen = space.length
return str.split("\n").map(line => line.slice(spaceLen)).join("\n").trim()
}
Object.assign(dedent, {
offset: (size) => (strings, ...values) => {
return dedent(strings, ...values).split("\n").map(line => " ".repeat(size) + line).join("\n")
}
})
function generateCodeSnippet({ tools, pre, post, response }) {
const lines = []
lines.push(dedent`
import { createReactAgent } from "@langchain/langgraph/prebuilt";
import { ChatOpenAI } from "@langchain/openai";
`)
if (tools) lines.push(`import { tool } from "@langchain/core/tools";`);
if (response || tools) lines.push(`import { z } from "zod";`);
lines.push("", dedent`
const agent = createReactAgent({
llm: new ChatOpenAI({ model: "o4-mini" }),
`)
if (tools) {
lines.push(dedent.offset(2)`
tools: [
tool(() => "Sample tool output", {
name: "sampleTool",
schema: z.object({}),
}),
],
`)
}
if (pre) {
lines.push(dedent.offset(2)`
preModelHook: (state) => ({ llmInputMessages: state.messages }),
`)
}
if (post) {
lines.push(dedent.offset(2)`
postModelHook: (state) => state,
`)
}
if (response) {
lines.push(dedent.offset(2)`
responseFormat: z.object({ result: z.string() }),
`)
}
lines.push(`});`);
return lines.join("\n");
}
function render() {
const key = getKey();
document.getElementById("agent-graph-img").src = `../assets/react_agent_graphs/${key}.svg`;
const state = {
tools: document.getElementById("tools").checked,
pre: document.getElementById("preModelHook").checked,
post: document.getElementById("postModelHook").checked,
response: document.getElementById("responseFormat").checked
};
document.getElementById("agent-code").textContent = generateCodeSnippet(state);
}
function initializeWidget() {
render(); // no need for `await` here
document.querySelectorAll(".agent-graph-features input").forEach((input) => {
input.addEventListener("change", render);
});
}
// Init for both full reload and SPA nav (used by MkDocs Material)
window.addEventListener("DOMContentLoaded", initializeWidget);
document$.subscribe(initializeWidget);
</script>
:::
-2
View File
@@ -23,7 +23,6 @@ Then, navigate to [Agent Chat UI](https://agentchat.vercel.app), or clone the re
UI has out-of-box support for rendering tool calls, and tool result messages. To customize what messages are shown, see the [Hiding Messages in the Chat](https://github.com/langchain-ai/agent-chat-ui?tab=readme-ov-file#hiding-messages-in-the-chat) section in the Agent Chat UI documentation.
:::python
## Add human-in-the-loop
Agent Chat UI has full support for [human-in-the-loop](../concepts/human_in_the_loop.md) workflows. To try it out, replace the agent code in `src/agent/graph.py` (from the [deployment](./deployment.md) guide) with this [agent implementation](./human-in-the-loop.md#using-with-agent-inbox):
@@ -33,7 +32,6 @@ Agent Chat UI has full support for [human-in-the-loop](../concepts/human_in_the_
!!! Important
Agent Chat UI works best if your LangGraph agent interrupts using the [`HumanInterrupt` schema][langgraph.prebuilt.interrupt.HumanInterrupt]. If you do not use that schema, the Agent Chat UI will be able to render the input passed to the `interrupt` function, but it will not have full support for resuming your graph.
:::
## Generative UI
+9
View File
@@ -62,6 +62,15 @@ Starting from the `LangGraph Platform` view...
1. In the panel, select the `Server` tab to view server logs for the revision. Server logs are only available after a revision has been deployed.
1. Within the `Server` tab, adjust the date/time range picker as needed. By default, the date/time range picker is set to the `Last 7 days`.
## View Deployment Metrics
Starting from the <a href="https://smith.langchain.com/" target="_blank">LangSmith UI</a>...
1. In the left-hand navigation panel, select `LangGraph Platform`. The `LangGraph Platform` view contains a list of existing LangGraph Platform deployments.
1. Select an existing deployment to monitor.
1. Select the `Monitoring` tab to view the deployment metrics. See a list of [all available metrics](../../concepts/langgraph_control_plane.md#monitoring).
1. Within the `Monitoring` tab, use the date/time range picker as needed. By default, the date/time range picker is set to the `Last 15 minutes`.
## Interrupt Revision
Interrupting a revision will stop deployment of the revision.
+1 -1
View File
@@ -16,4 +16,4 @@ Users can add an array of additional lines to add to the Dockerfile following th
}
```
This would install the system packages required to use Pillow if we were working with `jpeg` or `png` image formats.
This would install the system packages required to use Pillow if we were working with `jpeq` or `png` image formats.
+5 -5
View File
@@ -20,7 +20,7 @@ my-app/
|-- openai_agent.py # code for your graph
```
where the graph is defined in `openai_agent.py`.
where the graph is defined in `openai_agent.py`.
### No rebuild
@@ -28,11 +28,11 @@ In the standard LangGraph API configuration, the server uses the compiled graph
```python
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.graph import END, START, MessageGraph
model = ChatOpenAI(temperature=0)
graph_workflow = StateGraph(MessagesState)
graph_workflow = MessageGraph()
graph_workflow.add_node("agent", model)
graph_workflow.add_edge("agent", END)
@@ -61,7 +61,7 @@ To make your graph rebuild on each new run with custom configuration, you need t
from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START
from langgraph.graph import END, START, MessageGraph
from langgraph.graph.state import StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
@@ -144,4 +144,4 @@ Finally, you need to specify the path to your graph-making function (`make_graph
}
```
See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)
See more info on LangGraph API configuration file [here](../reference/cli.md#configuration-file)
+1 -1
View File
@@ -95,7 +95,7 @@ my-app/
## Define Graphs
Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledStateGraph][langgraph.graph.state.CompiledStateGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledGraph][langgraph.graph.graph.CompiledGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repository](https://github.com/langchain-ai/langgraph-example) to see their implementation):
@@ -108,7 +108,7 @@ my-app/
## Define Graphs
Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledStateGraph][langgraph.graph.state.CompiledStateGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
Implement your graphs! Graphs can be defined in a single file or multiple files. Make note of the variable names of each [CompiledGraph][langgraph.graph.graph.CompiledGraph] to be included in the LangGraph application. The variable names will be used later when creating the [LangGraph configuration file](../reference/cli.md#configuration-file).
Example `agent.py` file, which shows how to import from other modules you define (code for the modules is not shown here, please see [this repository](https://github.com/langchain-ai/langgraph-example-pyproject) to see their implementation):
@@ -0,0 +1,57 @@
# Run experiments over a dataset
LangGraph Studio supports evaluations by allowing you to run your assistant over a pre-defined LangSmith dataset. This enables you to understand how your application performs over a variety of inputs, compare the results to reference outputs, and score the results using [evaluators](../../../agents/evals.md).
This guide shows you how to run an experiment end-to-end from Studio.
---
## Prerequisites
Before running an experiment, ensure you have the following:
1. **A LangSmith dataset**: Your dataset should contain the inputs you want to test and optionally, reference outputs for comparison.
- The schema for the inputs must match the required input schema for the assistant. For more information on schemas, see [here](../../../concepts/low_level.md#schema).
- For more on creating datasets, see [How to Manage Datasets](https://docs.smith.langchain.com/evaluation/how_to_guides/manage_datasets_in_application#set-up-your-dataset).
2. **(Optional) Evaluators**: You can attach evaluators (e.g., LLM-as-a-Judge, heuristics, or custom functions) to your dataset in LangSmith. These will run automatically after the graph has processed all inputs.
- To learn more, read about [Evaluation Concepts](https://docs.smith.langchain.com/evaluation/concepts#evaluators).
3. **A running application**: The experiment can be run against:
- An application deployed on [LangGraph Platform](../../quick_start.md).
- A locally running application started via the [langgraph-cli](../../../tutorials/langgraph-platform/local-server.md).
---
## Step-by-step guide
### 1. Launch the experiment
Click the **Run experiment** button in the top right corner of the Studio page.
### 2. Select your dataset
In the modal that appears, select the dataset (or a specific dataset split) to use for the experiment and click **Start**.
### 3. Monitor the progress
All of the inputs in the dataset will now be run against the active assistant. Monitor the experiment's progress via the badge in the top right corner.
You can continue to work in Studio while the experiment runs in the background. Click the arrow icon button at any time to navigate to LangSmith and view the detailed experiment results.
---
## Troubleshooting
### "Run experiment" button is disabled
If the "Run experiment" button is disabled, check the following:
- **Deployed application**: If your application is deployed on LangGraph Platform, you may need to create a new revision to enable this feature.
- **Local development server**: If you are running your application locally, make sure you have upgraded to the latest version of the `langgraph-cli` (`pip install -U langgraph-cli`). Additionally, ensure you have tracing enabled by setting the `LANGSMITH_API_KEY` in your project's `.env` file.
### Evaluator results are missing
When you run an experiment, any attached evaluators are scheduled for execution in a queue. If you don't see results immediately, it likely means they are still pending.
+4 -113
View File
@@ -1,8 +1,8 @@
How to integrate LangGraph into your React application# How to integrate LangGraph into your React application
# How to integrate LangGraph into your React application
!!! info "Prerequisites"
!!! info "Prerequisites"
- [LangGraph Platform](../../concepts/langgraph_platform.md)
- [LangGraph Platform](../../concepts/langgraph_platform.md)
- [LangGraph Server](../../concepts/langgraph_server.md)
The `useStream()` React hook provides a seamless way to integrate LangGraph into your React applications. It handles all the complexities of streaming, state management, and branching logic, letting you focus on building great chat experiences.
@@ -113,115 +113,6 @@ export default function App() {
}
```
### Resume a stream after page refresh
The `useStream()` hook can automatically resume an ongoing run upon mounting by setting `reconnectOnMount: true`. This is useful for continuing a stream after a page refresh, ensuring no messages and events generated during the downtime are lost.
```tsx
const thread = useStream<{ messages: Message[] }>({
apiUrl: "http://localhost:2024",
assistantId: "agent",
reconnectOnMount: true,
});
```
By default the ID of the created run is stored in `window.sessionStorage`, which can be swapped by passing a custom storage in `reconnectOnMount` instead. The storage is used to persist the in-flight run ID for a thread (under `lg:stream:${threadId}` key).
```tsx
const thread = useStream<{ messages: Message[] }>({
apiUrl: "http://localhost:2024",
assistantId: "agent",
reconnectOnMount: () => window.localStorage,
});
```
You can also manually manage the resuming process by using the run callbacks to persist the run metadata and the `joinStream` function to resume the stream. Make sure to pass `streamResumable: true` when creating the run; otherwise some events might be lost.
````tsx
import type { Message } from "@langchain/langgraph-sdk";
import { useStream } from "@langchain/langgraph-sdk/react";
import { useCallback, useState, useEffect, useRef } from "react";
export default function App() {
const [threadId, onThreadId] = useSearchParam("threadId");
const thread = useStream<{ messages: Message[] }>({
apiUrl: "http://localhost:2024",
assistantId: "agent",
threadId,
onThreadId,
onCreated: (run) => {
window.sessionStorage.setItem(`resume:${run.thread_id}`, run.run_id);
},
onFinish: (_, run) => {
window.sessionStorage.removeItem(`resume:${run?.thread_id}`);
},
});
// Ensure that we only join the stream once per thread.
const joinedThreadId = useRef<string | null>(null);
useEffect(() => {
if (!threadId) return;
const resume = window.sessionStorage.getItem(`resume:${threadId}`);
if (resume && joinedThreadId.current !== threadId) {
thread.joinStream(resume);
joinedThreadId.current = threadId;
}
}, [threadId]);
return (
<form
onSubmit={(e) => {
e.preventDefault();
const form = e.target as HTMLFormElement;
const message = new FormData(form).get("message") as string;
thread.submit(
{ messages: [{ type: "human", content: message }] },
{ streamResumable: true }
);
}}
>
<div>
{thread.messages.map((message) => (
<div key={message.id}>{message.content as string}</div>
))}
</div>
<input type="text" name="message" />
<button type="submit">Send</button>
</form>
);
}
// Utility method to retrieve and persist data in URL as search param
function useSearchParam(key: string) {
const [value, setValue] = useState<string | null>(() => {
const params = new URLSearchParams(window.location.search);
return params.get(key) ?? null;
});
const update = useCallback(
(value: string | null) => {
setValue(value);
const url = new URL(window.location.href);
if (value == null) {
url.searchParams.delete(key);
} else {
url.searchParams.set(key, value);
}
window.history.pushState({}, "", url.toString());
},
[key]
);
return [value, update] as const;
}
```
### Thread Management
Keep track of conversations with built-in thread management. You can access the current thread ID and get notified when new threads are created:
@@ -236,7 +127,7 @@ const thread = useStream<{ messages: Message[] }>({
threadId: threadId,
onThreadId: setThreadId,
});
````
```
We recommend storing the `threadId` in your URL's query parameters to let users resume conversations after page refreshes.
@@ -3818,14 +3818,6 @@
"title": "Filter",
"description": "Optional dictionary of key-value pairs to filter results."
},
"query": {
"type": [
"string",
"null"
],
"title": "Query",
"description": "Query string for semantic/vector search."
},
"limit": {
"type": "integer",
"default": 10,
+1 -1
View File
@@ -32,7 +32,7 @@ Below are examples of directory structures for Python and JavaScript application
│ ├── utils # utilities for your graph
│ │ ├── __init__.py
│ │ ├── tools.py # tools for your graph
│ │ ├── nodes.py # node functions for your graph
│ │ ├── nodes.py # node functions for you graph
│ │ └── state.py # state definition of your graph
│ ├── __init__.py
│ └── agent.py # code for constructing your graph
+1 -1
View File
@@ -26,4 +26,4 @@ Once you've created an assistant, subsequent edits to that assistant will create
## Learn more
* The LangGraph Cloud API provides several endpoints for creating and managing assistants and their versions. See the [API reference](../cloud/reference/api/api_ref.html#tag/assistants) for more details.
* The LangGraph Cloud API provides several endpoints for creating and managing assistants their versions. See the [API reference](../cloud/reference/api/api_ref.html#tag/assistants) for more details.
+1 -1
View File
@@ -198,7 +198,7 @@ async def add_owner(
You can register handlers for specific resources and actions by chaining the resource and action names together with the [`@auth.on`](../cloud/reference/sdk/python_sdk_ref.md#langgraph_sdk.auth.Auth.on) decorator.
When a request is made, the most specific handler that matches that resource and action is called. Below is an example of how to register handlers for specific resources and actions. For the following setup:
1. Authenticated users are able to create threads, read threads, and create runs on threads
1. Authenticated users are able to create threads, read thread, create runs on threads
2. Only users with the "assistants:create" permission are allowed to create new assistants
3. All other endpoints (e.g., e.g., delete assistant, crons, store) are disabled for all users.
@@ -0,0 +1,127 @@
# Data Storage and Privacy
This document provides a comprehensive overview of what data is stored, collected, and processed when using LangGraph, particularly with the CLI tools like `langgraph dev`.
## What Data is Stored
### CLI Telemetry (Opt-out)
By default, the LangGraph CLI collects minimal analytics data to help improve the tool:
**Data Collected:**
- CLI command used (e.g., `dev`, `up`, `build`)
- CLI version
- Operating system type and version
- Python version
- Anonymized parameter usage (boolean flags indicating non-default options were used)
**Data NOT Collected:**
- Actual parameter values
- File contents or paths
- Personal information
- Code or graph implementations
- API keys or sensitive data
**How to Opt Out:**
Set the environment variable `LANGGRAPH_CLI_NO_ANALYTICS=1` to disable all CLI analytics collection.
### LangSmith Integration (Opt-in)
When a `LANGSMITH_API_KEY` is provided (not required):
- Metadata on number of runs executed
- Current API version being run
- Trace data (if tracing is enabled)
This data is only sent when explicitly configured with LangSmith credentials.
### Tracing Data (Opt-in)
When tracing is enabled:
- Execution traces are logged to the configured tracing backend
- This requires explicit configuration and is not enabled by default
## What Data is NOT Stored Remotely
- **Checkpoints**: Stored locally in your development environment
- **Memory store data**: Persisted locally, not transmitted
- **Graph state**: Remains in your local environment
- **Application data**: Your actual application logic and data stay local
## Local Data Storage
### Development Mode (`langgraph dev`)
When using `langgraph dev`:
- State is persisted to a local directory
- Checkpoints are stored locally for debugging and development
- No remote storage or transmission of your application data
### Checkpoints and State Persistence
LangGraph automatically persists:
- **Checkpoints**: Snapshots of graph state at each execution step
- **Thread data**: Conversation/execution history organized by thread IDs
- **Graph state**: Node outputs, intermediate results, and execution metadata
- **Memory/Store data**: Information that persists across multiple threads
**Storage Locations:**
- **Local development**: Local directory (configurable)
- **Docker deployment**: Local Docker volumes
- **LangGraph Platform**: Managed database infrastructure
## Security and Encryption
### Data Encryption
- Checkpointers can optionally encrypt all persisted state
- Encryption uses AES encryption via `EncryptedSerializer`
- When `LANGGRAPH_AES_KEY` environment variable is present, encryption is automatically enabled on LangGraph Platform
### Data Retention
- **TTL (Time-to-Live)**: Configurable automatic cleanup of old data
- **Default TTL**: Can be set in minutes for automatic expiration
- **Automatic sweeping**: Expired data is automatically removed at configurable intervals
## Privacy Controls
### Environment Variables
Key environment variables for controlling data collection and storage:
- `LANGGRAPH_CLI_NO_ANALYTICS=1`: Disable CLI analytics collection
- `LANGGRAPH_AES_KEY`: Enable automatic encryption of stored data
- `LANGSMITH_TRACING=false`: Disable tracing to LangSmith (self-hosted deployments)
- `LANGSMITH_API_KEY`: Enable LangSmith integration (opt-in)
### Logging Controls
- `LOG_LEVEL`: Control verbosity of logs
- `LOG_JSON`: Format logs as JSON
- Various other logging configuration options
## Security Policy
For security vulnerabilities:
- Report through the huntr.com bounty program
- LangGraph is in-scope for security bounties
- Security contact: `security@langchain.dev`
## Best Practices for Privacy
1. **Review Analytics**: Set `LANGGRAPH_CLI_NO_ANALYTICS=1` if you prefer not to share usage analytics
2. **Enable Encryption**: Use `LANGGRAPH_AES_KEY` for sensitive data
3. **Configure TTL**: Set appropriate data retention policies
4. **Monitor Tracing**: Only enable tracing when needed and review what data is being sent
5. **Environment Variables**: Audit your environment variables to ensure proper privacy controls
## Summary
LangGraph is designed with privacy in mind:
- Minimal data collection (analytics can be disabled)
- Local storage by default for development
- Optional encryption for sensitive data
- Clear opt-in requirements for external services
- Comprehensive privacy controls through environment variables
Your application data, checkpoints, and state remain under your control and are not transmitted unless you explicitly configure external services like LangSmith.
+1 -1
View File
@@ -59,7 +59,7 @@ For more information, please see:
!!! info "Important"
The Self-Hosted Control Plane deployment option is currently in beta stage and requires an [Enterprise](../concepts/plans.md) plan.
The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option gives you full control and responsibility of the control plane and data plane infrastructure.
The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
Build a Docker image using the [LangGraph CLI](./langgraph_cli.md) and deploy your LangGraph Server from the [control plane UI](./langgraph_control_plane.md#control-plane-ui).
+2 -2
View File
@@ -59,8 +59,8 @@ Yes! You can use LangGraph with any LLMs. The main reason we use LLMs that suppo
Yes! LangGraph is totally ambivalent to what LLMs are used under the hood. The main reason we use closed LLMs in most of the tutorials is that they seamlessly support tool calling, while OSS LLMs often don't. But tool calling is not necessary (see [this section](#does-langgraph-work-with-llms-that-dont-support-tool-calling)) so you can totally use LangGraph with OSS LLMs.
## Can I use LangGraph Studio without logging in to LangSmith
## Can I use LangGraph Studio without logging to LangSmith
Yes! You can use the [development version of LangGraph Server](../tutorials/langgraph-platform/local-server.md) to run the backend locally.
This will connect to the studio frontend hosted as part of LangSmith.
If you set an environment variable of `LANGSMITH_TRACING=false`, then no traces will be sent to LangSmith.
If you set an environment variable of `LANGSMITH_TRACING=false` then no traces will be sent to LangSmith.
+1 -1
View File
@@ -186,7 +186,7 @@ When declaring an `entrypoint`, you can request access to additional parameters
| Parameter | Description |
|--------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| **previous** | Access the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory). |
| **previous** | Access the the state associated with the previous `checkpoint` for the given thread. See [short-term-memory](#short-term-memory). |
| **store** | An instance of [BaseStore][langgraph.store.base.BaseStore]. Useful for [long-term memory](../how-tos/use-functional-api.md#long-term-memory). |
| **writer** | Use to access the StreamWriter when working with Async Python < 3.11. See [streaming with functional API for details](../how-tos/use-functional-api.md#streaming). |
| **config** | For accessing run time configuration. See [RunnableConfig](https://python.langchain.com/docs/concepts/runnables/#runnableconfig) for information. |
@@ -19,6 +19,7 @@ From the control plane UI, you can:
- Update a deployment.
- Update environment variables for a deployment.
- View build and server logs of a deployment.
- View deployment metrics such as CPU and memory usage.
- Delete a deployment.
The Control Plane UI is embedded in [LangSmith](https://docs.smith.langchain.com/langgraph_cloud).
@@ -88,6 +89,17 @@ Infrastructure for deployments and revisions are provisioned and deployed asynch
The control plane and [LangGraph Data Plane](./langgraph_data_plane.md) "listener" application coordinate to achieve asynchronous deployments.
### Monitoring
After a deployment is ready, the control plane monitors the deployment and records various metrics, such as:
- CPU and memory usage of the deployment.
- Number of container restarts.
- Number of replicas (this will increase with [autoscaling](../concepts/langgraph_data_plane.md#autoscaling)).
- [Postgres](../concepts/langgraph_data_plane.md#postgres) CPU, memory usage, and disk usage.
These metrics are displayed as charts in the Control Plane UI.
### LangSmith Integration
A [LangSmith](https://docs.smith.langchain.com/) tracing project is automatically created for each deployment. The tracing project has the same name as the deployment. When creating a deployment, the `LANGCHAIN_TRACING` and `LANGSMITH_API_KEY`/`LANGCHAIN_API_KEY` environment variables do not need to be specified; they are set automatically by the control plane.
+3 -3
View File
@@ -9,7 +9,7 @@ The term "data plane" is used broadly to refer to [LangGraph Servers](./langgrap
## Server Infrastructure
In addition to the [LangGraph Server](./langgraph_server.md) itself, the following infrastructure components for each server are also included in the broad definition of "data plane":
In addition to the [LangGraph Server](./langgraph_server.md) itself, the following infrastructure for each server are also included in the broad definition of "data plane":
- Postgres
- Redis
@@ -44,7 +44,7 @@ All runs in a LangGraph Server are executed by a pool of background workers that
### Ephemeral metadata
Runs in a LangGraph Server may be retried for specific failures (currently only for transient Postgres errors encountered during the run). In order to limit the number of retries (currently limited to 3 attempts per run) we record the attempt number in a Redis string when it is picked up. This contains no run-specific info other than its ID, and expires after a short delay.
Runs in a LangGraph Server may be retried for specific failures (currently only for transient Postgres errors encountered during the run). In order to limit the number of retries (currently limited to 3 attempts per run) we record the attempt number in a Redis string when is picked up. This contains no run-specific info other than its ID, and expires after a short delay.
## Data Plane Features
@@ -62,7 +62,7 @@ For CPU utilization, the autoscaler targets 75% utilization. This means the auto
For number of pending runs, the autoscaler targets 10 pending runs. For example, if the current number of containers is 1, but the number of pending runs in 20, the autoscaler will scale up the deployment to 2 containers (20 pending runs / 2 containers = 10 pending runs per container).
Each metric is computed independently and the autoscaler will determine the scaling action based on the metric that results in the largest number of containers.
Each metric is computed independently and the autoscaler will determine the scaling action based on the metric that results in the most number of containers.
Scale down actions are delayed for 30 minutes before any action is taken. In other words, if the autoscaler decides to scale down a deployment, it will first wait for 30 minutes before scaling down. After 30 minutes, the metrics are recomputed and the deployment will scale down if the recomputed metrics result in a lower number of containers than the current number. Otherwise, the deployment remains scaled up. This "cool down" period ensures that deployments do not scale up and down too frequently.
+2 -2
View File
@@ -9,7 +9,7 @@ Develop, deploy, scale, and manage agents with **LangGraph Platform** — the pu
!!! tip "Get started with LangGraph Platform"
Check out the [LangGraph Platform quickstart](../tutorials/langgraph-platform/local-server.md) for instructions on how to use LangGraph Platform to run a LangGraph application locally.
Check out the [LangGraph Platform quickstart](../tutorials/langgraph-platform/local-server.md) for instructions on how to use LangGraph Platform run a LangGraph application locally.
## Why use LangGraph Platform?
@@ -33,4 +33,4 @@ LangGraph Platform makes it easy to get your agent running in production — wh
- **[LangGraph Studio](./langgraph_studio.md)**: Enables visualization, interaction, and debugging of agentic systems that implement the LangGraph Server API protocol. Studio also integrates with LangSmith to enable tracing, evaluation, and prompt engineering.
- **[Deployment](./deployment_options.md)**: There are four ways to deploy on LangGraph Platform: [Cloud SaaS](../concepts/langgraph_cloud.md), [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md), [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md), and [Standalone Container](../concepts/langgraph_standalone_container.md).
- **[Deployment](./deployment_options.md)**: There are four ways to deploy on LangGraph Platform: [Cloud Saas](../concepts/langgraph_cloud.md), [Self-Hosted Data Plane](../concepts/langgraph_self_hosted_data_plane.md), [Self-Hosted Control Plane](../concepts/langgraph_self_hosted_control_plane.md), and [Standalone Container](../concepts/langgraph_standalone_container.md).
@@ -9,12 +9,10 @@ There are two versions of the self-hosted deployment: [Self-Hosted Data Plane](.
- You use `langgraph-cli` and/or [LangGraph Studio](./langgraph_studio.md) app to test graph locally.
- You use `langgraph build` command to build image.
- You have a Self-Hosted LangSmith instance deployed.
- You are using Ingress for your LangSmith instance. All agents will be deployed as Kubernetes services behind this ingress.
## Self-Hosted Control Plane
The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option gives you full control and responsibility of the control plane and data plane infrastructure.
The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deployment option is a fully self-hosted model for deployment where you manage the [control plane](./langgraph_control_plane.md) and [data plane](./langgraph_data_plane.md) in your cloud. This option give you full control and responsibility of the control plane and data plane infrastructure.
| | [Control plane](../concepts/langgraph_control_plane.md) | [Data plane](../concepts/langgraph_data_plane.md) |
|-------------------|-------------------|------------|
@@ -31,4 +29,4 @@ The [Self-Hosted Control Plane](./langgraph_self_hosted_control_plane.md) deploy
- **Kubernetes**: The Self-Hosted Control Plane deployment option supports deploying control plane and data plane infrastructure to any Kubernetes cluster.
!!! tip
If you would like to enable this on your LangSmith instance, please follow the [Self-Hosted Control Plane deployment guide](../cloud/deployment/self_hosted_control_plane.md).
If you would like to deploy to Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
@@ -37,4 +37,4 @@ For information on how to deploy a [LangGraph Server](../concepts/langgraph_serv
- **Amazon ECS**: Coming soon!
!!! tip
If you would like to deploy to Kubernetes, you can follow the [Self-Hosted Data Plane deployment guide](../cloud/deployment/self_hosted_data_plane.md).
If you would like to deploy to Kubernetes, you can use this [Helm chart](https://github.com/langchain-ai/helm/blob/main/charts/langgraph-cloud/README.md).
+1 -1
View File
@@ -26,7 +26,7 @@ Feature Differences:
|-------|------------|------------|
| [Cron Jobs](../cloud/concepts/cron_jobs.md) |❌|✅|
| [Custom Authentication](../concepts/auth.md) |❌|✅|
| [Deployment options](../concepts/deployment_options.md) | Standalone container | Cloud SaaS, Self-Hosted Data Plane, Self-Hosted Control Plane, Standalone container
| [Deployment options](../concepts/deployment_options.md) | Standalone container | Cloud Saas, Self-Hosted Data Plane, Self-Hosted Control Plane, Standalone container
## Application structure
+3 -2
View File
@@ -24,6 +24,7 @@ Key features of LangGraph Studio:
- [Manage assistants](../cloud/how-tos/studio/manage_assistants.md)
- [Manage threads](../cloud/how-tos/threads_studio.md)
- [Iterate on prompts](../cloud/how-tos/iterate_graph_studio.md)
- [Run experiments over a dataset](../cloud/how-tos/studio/run_evals.md)
- Manage [long term memory](memory.md)
- Debug agent state via [time travel](time-travel.md)
@@ -33,7 +34,7 @@ Studio supports two modes:
### Graph mode
Graph mode exposes the full feature-set of Studio and is useful when you would like as many details about the execution of your agent, including the nodes traversed, intermediate states, and LangSmith integrations (such as adding to datasets and playground).
Graph mode exposes the full feature-set of Studio and is useful when you would like as many details about the execution of your agent, including the nodes traversed, intermediate states, and LangSmith integrations (such as adding to datasets an playground).
### Chat mode
@@ -41,4 +42,4 @@ Chat mode is a simpler UI for iterating on and testing chat-specific agents. It
## Learn more
- See this guide on how to [get started](../cloud/how-tos/studio/quick_start.md) with LangGraph Studio.
- See this guide on how to [get started](../cloud/how-tos/studio/quick_start.md) with LangGraph Studio.
+4 -4
View File
@@ -89,7 +89,7 @@ def node_3(state: PrivateState) -> OutputState:
# Read from PrivateState, write to OutputState
return {"graph_output": state["bar"] + " Lance"}
builder = StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)
builder = StateGraph(OverallState,input=InputState,output=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
@@ -105,9 +105,9 @@ graph.invoke({"user_input":"My"})
There are two subtle and important points to note here:
1. We pass `state: InputState` as the input schema to `node_1`. But, we write out to `foo`, a channel in `OverallState`. How can we write out to a state channel that is not included in the input schema? This is because a node _can write to any state channel in the graph state._ The graph state is the union of the state channels defined at initialization, which includes `OverallState` and the filters `InputState` and `OutputState`.
1. We pass `state: InputState` as the input schema to `node_1`. But, we write out to `foo`, a channel in `OverallState`. How can we write out to a state channel that is not included in the input schema? This is because a node _can write to any state channel in the graph state._ The graph state is the union of of the state channels defined at initialization, which includes `OverallState` and the filters `InputState` and `OutputState`.
2. We initialize the graph with `StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)`. So, how can we write to `PrivateState` in `node_2`? How does the graph gain access to this schema if it was not passed in the `StateGraph` initialization? We can do this because _nodes can also declare additional state channels_ as long as the state schema definition exists. In this case, the `PrivateState` schema is defined, so we can add `bar` as a new state channel in the graph and write to it.
2. We initialize the graph with `StateGraph(OverallState,input=InputState,output=OutputState)`. So, how can we write to `PrivateState` in `node_2`? How does the graph gain access to this schema if it was not passed in the `StateGraph` initialization? We can do this because _nodes can also declare additional state channels_ as long as the state schema definition exists. In this case, the `PrivateState` schema is defined, so we can add `bar` as a new state channel in the graph and write to it.
### Reducers
@@ -167,7 +167,7 @@ In addition to keeping track of message IDs, the `add_messages` function will al
{"messages": [{"type": "human", "content": "message"}]}
```
Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as its reducer function.
Since the state updates are always deserialized into LangChain `Messages` when using `add_messages`, you should use dot notation to access message attributes, like `state["messages"][-1].content`. Below is an example of a graph that uses `add_messages` as it's reducer function.
```python
from langchain_core.messages import AnyMessage
+1 -18
View File
@@ -383,7 +383,7 @@ def update_memory(state: MessagesState, config: RunnableConfig, *, store: BaseSt
```
As we showed above, we can also access the store in any node and use the `store.search` method to get memories. Recall the memories are returned as a list of objects that can be converted to a dictionary.
As we showed above, we can also access the store in any node and use the `store.search` method to get memories. Recall the the memories are returned as a list of objects that can be converted to a dictionary.
```python
memories[-1].dict()
@@ -473,23 +473,6 @@ If the checkpointer is used with asynchronous graph execution (i.e. executing th
When checkpointers save the graph state, they need to serialize the channel values in the state. This is done using serializer objects.
`langgraph_checkpoint` defines [protocol][langgraph.checkpoint.serde.base.SerializerProtocol] for implementing serializers provides a default implementation ([JsonPlusSerializer][langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer]) that handles a wide variety of types, including LangChain and LangGraph primitives, datetimes, enums and more.
#### Serialization with `pickle`
The default serializer, [`JsonPlusSerializer`][langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer], uses ormsgpack and JSON under the hood, which is not suitable for all types of objects.
If you want to fallback to pickle for objects not currently supported by our msgpack encoder (such as Pandas dataframes),
you can use the `pickle_fallback` argument of the `JsonPlusSerializer`:
```python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
# ... Define the graph ...
graph.compile(
checkpointer=MemorySaver(serde=JsonPlusSerializer(pickle_fallback=True))
)
```
#### Encryption
Checkpointers can optionally encrypt all persisted state. To enable this, pass an instance of [`EncryptedSerializer`][langgraph.checkpoint.serde.encrypted.EncryptedSerializer] to the `serde` argument of any `BaseCheckpointSaver` implementation. The easiest way to create an encrypted serializer is via [`from_pycryptodome_aes`][langgraph.checkpoint.serde.encrypted.EncryptedSerializer.from_pycryptodome_aes], which reads the AES key from the `LANGGRAPH_AES_KEY` environment variable (or accepts a `key` argument):
+49 -37
View File
@@ -25,7 +25,7 @@ Each step consists of three phases:
Repeat until no **actors** are selected for execution, or a maximum number of steps is reached.
## Actors
## Actors
An **actor** is a `PregelNode`. It subscribes to channels, reads data from them, and writes data to them. It can be thought of as an **actor** in the Pregel algorithm. `PregelNodes` implement LangChain's Runnable interface.
@@ -39,7 +39,7 @@ Channels are used to communicate between actors (PregelNodes). Each channel has
## Examples
While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or
While most users will interact with Pregel through the [StateGraph][langgraph.graph.StateGraph] API or
the [entrypoint][langgraph.func.entrypoint] decorator, it is possible to interact with Pregel directly.
Below are a few different examples to give you a sense of the Pregel API.
@@ -49,12 +49,12 @@ Below are a few different examples to give you a sense of the Pregel API.
```python
from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder
from langgraph.pregel import Pregel, Channel
node1 = (
NodeBuilder().subscribe_only("a")
.do(lambda x: x + x)
.write_to("b")
Channel.subscribe_to("a")
| (lambda x: x + x)
| Channel.write_to("b")
)
app = Pregel(
@@ -78,18 +78,18 @@ Below are a few different examples to give you a sense of the Pregel API.
```python
from langgraph.channels import LastValue, EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder
from langgraph.pregel import Pregel, Channel
node1 = (
NodeBuilder().subscribe_only("a")
.do(lambda x: x + x)
.write_to("b")
Channel.subscribe_to("a")
| (lambda x: x + x)
| Channel.write_to("b")
)
node2 = (
NodeBuilder().subscribe_only("b")
.do(lambda x: x + x)
.write_to("c")
Channel.subscribe_to("b")
| (lambda x: x + x)
| Channel.write_to("c")
)
@@ -115,18 +115,23 @@ Below are a few different examples to give you a sense of the Pregel API.
```python
from langgraph.channels import EphemeralValue, Topic
from langgraph.pregel import Pregel, NodeBuilder
from langgraph.pregel import Pregel, Channel
node1 = (
NodeBuilder().subscribe_only("a")
.do(lambda x: x + x)
.write_to("b", "c")
Channel.subscribe_to("a")
| (lambda x: x + x)
| {
"b": Channel.write_to("b"),
"c": Channel.write_to("c")
}
)
node2 = (
NodeBuilder().subscribe_to("b")
.do(lambda x: x["b"] + x["b"])
.write_to("c")
Channel.subscribe_to("b")
| (lambda x: x + x)
| {
"c": Channel.write_to("c"),
}
)
app = Pregel(
@@ -153,19 +158,24 @@ Below are a few different examples to give you a sense of the Pregel API.
```python
from langgraph.channels import EphemeralValue, BinaryOperatorAggregate
from langgraph.pregel import Pregel, NodeBuilder
from langgraph.pregel import Pregel, Channel
node1 = (
NodeBuilder().subscribe_only("a")
.do(lambda x: x + x)
.write_to("b", "c")
Channel.subscribe_to("a")
| (lambda x: x + x)
| {
"b": Channel.write_to("b"),
"c": Channel.write_to("c")
}
)
node2 = (
NodeBuilder().subscribe_only("b")
.do(lambda x: x + x)
.write_to("c")
Channel.subscribe_to("b")
| (lambda x: x + x)
| {
"c": Channel.write_to("c"),
}
)
def reducer(current, update):
@@ -187,7 +197,8 @@ Below are a few different examples to give you a sense of the Pregel API.
app.invoke({"a": "foo"})
```
=== "Cycle"
This example demonstrates how to introduce a cycle in the graph, by having
@@ -196,12 +207,12 @@ Below are a few different examples to give you a sense of the Pregel API.
```python
from langgraph.channels import EphemeralValue
from langgraph.pregel import Pregel, NodeBuilder, ChannelWriteEntry
from langgraph.pregel import Pregel, Channel, ChannelWrite, ChannelWriteEntry
example_node = (
NodeBuilder().subscribe_only("value")
.do(lambda x: x + x if len(x) < 10 else None)
.write_to(ChannelWriteEntry("value", skip_none=True))
Channel.subscribe_to("value")
| (lambda x: x + x if len(x) < 10 else None)
| ChannelWrite(writes=[ChannelWriteEntry(channel="value", skip_none=True)])
)
app = Pregel(
@@ -224,6 +235,7 @@ Below are a few different examples to give you a sense of the Pregel API.
LangGraph provides two high-level APIs for creating a Pregel application: the [StateGraph (Graph API)](./low_level.md) and the [Functional API](functional_api.md).
=== "StateGraph (Graph API)"
The [StateGraph (Graph API)][langgraph.graph.StateGraph] is a higher-level abstraction that simplifies the creation of Pregel applications. It allows you to define a graph of nodes and edges. When you compile the graph, the StateGraph API automatically creates the Pregel application for you.
@@ -254,7 +266,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
builder.add_node(score_essay)
builder.add_edge(START, "write_essay")
# Compile the graph.
# Compile the graph.
# This will return a Pregel instance.
graph = builder.compile()
```
@@ -267,7 +279,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
You will see something like this:
```pycon
```pycon
{'__start__': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1810>,
'write_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba14d0>,
'score_essay': <langgraph.pregel.read.PregelNode at 0x7d05e3ba1710>}
@@ -298,7 +310,7 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
=== "Functional API"
In the [Functional API](functional_api.md), you can use an [`entrypoint`][langgraph.func.entrypoint] to create
a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output.
a Pregel application. The `entrypoint` decorator allows you to define a function that takes input and returns output.
```python
from typing import TypedDict, Optional
@@ -327,8 +339,8 @@ LangGraph provides two high-level APIs for creating a Pregel application: the [S
```
```pycon
Nodes:
Nodes:
{'write_essay': <langgraph.pregel.read.PregelNode object at 0x7d05e2f9aad0>}
Channels:
Channels:
{'__start__': <langgraph.channels.ephemeral_value.EphemeralValue object at 0x7d05e2c906c0>, '__end__': <langgraph.channels.last_value.LastValue object at 0x7d05e2c90c40>, '__previous__': <langgraph.channels.last_value.LastValue object at 0x7d05e1007280>}
```
@@ -25,7 +25,7 @@ When a graceful shutdown request is received (SIGINT) an instance enters shutdow
- gives any in-progress runs a limited number of seconds to finish (if not finished it will be put back in the queue)
- stops the instance from picking up more runs from the queue
If a hard shutdown occurs due to a server crash or an infrastructure failure, any runs that were in progress will be picked up by an internal sweeper task that looks for in-progress runs that have breached their heartbeat window. The sweeper runs every 2 minutes and will put the runs back in the queue for another instance to pick them up.
If a hard shutdown occurs due to a server crash or an infrastructure failure, any runs that were in progress will be picked up by a internal sweeper task that looks for in-progress runs that have breached their heartbeat window. The sweeper runs every 2 minutes and will put the runs back in the queue for another instance to pick them up.
## Postgres resilience
+1 -1
View File
@@ -5,7 +5,7 @@ search:
# LangGraph SDK
LangGraph Platform provides both a Python SDK for interacting with [LangGraph Server](./langgraph_server.md).
LangGraph Platform provides both a Python and JS SDK for interacting with [LangGraph Server](./langgraph_server.md).
!!! tip "Python SDK reference"
+1 -1
View File
@@ -94,7 +94,7 @@ def answer_node(state: InputState):
return {"answer": "bye", "question": state["question"]}
# Build the graph with explicit schemas
builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
builder = StateGraph(OverallState, input=InputState, output=OutputState)
builder.add_node(answer_node)
builder.add_edge(START, "answer_node")
builder.add_edge("answer_node", END)
+2 -3
View File
@@ -59,9 +59,8 @@ The main question when adding subgraphs is how the parent graph and subgraph com
response = model.invoke(state["subgraph_messages"])
return {"subgraph_messages": response}
subgraph_builder = StateGraph(SubgraphMessagesState)
subgraph_builder.add_node("call_model_from_subgraph", call_model)
subgraph_builder.add_edge(START, "call_model_from_subgraph")
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(call_model)
...
# highlight-next-line
subgraph = subgraph_builder.compile()
+1 -1
View File
@@ -74,7 +74,7 @@ In your `langgraph.json`, add the path to your auth file:
## 3. Connect from the client
Once you've set up authentication in your server, requests must include the required authorization information based on your chosen scheme.
Once you've set up authentication in your server, requests must include the the required authorization information based on your chosen scheme.
Assuming you are using JWT token authentication, you could access your deployments using any of the following methods:
=== "Python Client"
+9 -9
View File
@@ -439,7 +439,7 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 6,
"id": "6ec0eb77-874e-443e-8c73-93125b515106",
"metadata": {},
"outputs": [
@@ -478,7 +478,7 @@
"\n",
"\n",
"# Build the graph with input and output schemas specified\n",
"builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)\n",
"builder = StateGraph(OverallState, input=InputState, output=OutputState)\n",
"builder.add_node(answer_node) # Add the answer node\n",
"builder.add_edge(START, \"answer_node\") # Define the starting edge\n",
"builder.add_edge(\"answer_node\", END) # Define the ending edge\n",
@@ -1198,7 +1198,7 @@
"\n",
"There are many use cases where you may wish for your node to have a custom retry policy, for example if you are calling an API, querying a database, or calling an LLM, etc. LangGraph lets you add retry policies to nodes.\n",
"\n",
"To configure a retry policy, pass the `retry_policy` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node). The `retry_policy` parameter takes in a `RetryPolicy` named tuple object. Below we instantiate a `RetryPolicy` object with the default parameters and associate it with a node:\n",
"To configure a retry policy, pass the `retry` parameter to the [add_node](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_node). The `retry` parameter takes in a `RetryPolicy` named tuple object. Below we instantiate a `RetryPolicy` object with the default parameters and associate it with a node:\n",
"\n",
"```python\n",
"from langgraph.pregel import RetryPolicy\n",
@@ -1206,7 +1206,7 @@
"builder.add_node(\n",
" \"node_name\",\n",
" node_function,\n",
" retry_policy=RetryPolicy(),\n",
" retry=RetryPolicy(),\n",
")\n",
"```"
]
@@ -1241,7 +1241,7 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 2,
"id": "ad92598c-b688-42fa-aae0-9de36273d584",
"metadata": {},
"outputs": [],
@@ -1276,9 +1276,9 @@
"builder.add_node(\n",
" \"query_database\",\n",
" query_database,\n",
" retry_policy=RetryPolicy(retry_on=sqlite3.OperationalError),\n",
" retry=RetryPolicy(retry_on=sqlite3.OperationalError),\n",
")\n",
"builder.add_node(\"model\", call_model, retry_policy=RetryPolicy(max_attempts=5))\n",
"builder.add_node(\"model\", call_model, retry=RetryPolicy(max_attempts=5))\n",
"builder.add_edge(START, \"model\")\n",
"builder.add_edge(\"model\", \"query_database\")\n",
"builder.add_edge(\"query_database\", END)\n",
@@ -3416,7 +3416,7 @@
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
@@ -3430,7 +3430,7 @@
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.9.6"
"version": "3.10.4"
}
},
"nbformat": 4,
@@ -12,9 +12,9 @@
"\n",
"\n",
"1. **Run the graph** with initial inputs using `invoke` or `stream` APIs.\n",
"2. **Identify a checkpoint in an existing thread**: Use the [`get_state_history()`][langgraph.graph.state.CompiledStateGraph.get_state_history] method to retrieve the execution history for a specific `thread_id` and locate the desired `checkpoint_id`. \n",
"2. **Identify a checkpoint in an existing thread**: Use the [`get_state_history()`][langgraph.graph.graph.CompiledGraph.get_state_history] method to retrieve the execution history for a specific `thread_id` and locate the desired `checkpoint_id`. \n",
" Alternatively, set a [breakpoint](../../../concepts/breakpoints/) before the node(s) where you want execution to pause. You can then find the most recent checkpoint recorded up to that breakpoint.\n",
"3. **(Optional) modify the graph state**: Use the [`update_state`][langgraph.graph.state.CompiledStateGraph.update_state] method to modify the graphs state at the checkpoint and resume execution from alternative state.\n",
"3. **(Optional) modify the graph state**: Use the [`update_state`][langgraph.graph.graph.CompiledGraph.update_state] method to modify the graphs state at the checkpoint and resume execution from alternative state.\n",
"4. **Resume execution from the checkpoint**: Use the `invoke` or `stream` APIs with an input of `None` and a configuration containing the appropriate `thread_id` and `checkpoint_id`.\n",
"\n",
"## Example\n",
+2 -4
View File
@@ -405,7 +405,7 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 46,
"id": "1954a5f1-91e4-4b32-9be9-c8bc1cc43cb5",
"metadata": {},
"outputs": [],
@@ -465,9 +465,7 @@
"\n",
"graph_builder = StateGraph(State)\n",
"graph_builder.add_node(\"agent\", agent)\n",
"graph_builder.add_node(\n",
" \"select_tools\", select_tools, retry_policy=RetryPolicy(max_attempts=3)\n",
")\n",
"graph_builder.add_node(\"select_tools\", select_tools, retry=RetryPolicy(max_attempts=3))\n",
"\n",
"tool_node = ToolNode(tools=tools)\n",
"graph_builder.add_node(\"tools\", tool_node)\n",
@@ -207,7 +207,7 @@
"id": "213d661e-6ba4-42b9-bc7f-6c8c423e3419",
"metadata": {},
"source": [
"Let's now create our agents using the prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] and our multi-agent workflow. Note that will be calling [`interrupt`][langgraph.types.interrupt] every time after we get the final response from each of the agents."
"Let's now create our agents using the the prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] and our multi-agent workflow. Note that will be calling [`interrupt`][langgraph.types.interrupt] every time after we get the final response from each of the agents."
]
},
{
+14 -8
View File
@@ -739,6 +739,7 @@
" 'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a',\n",
" 'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]},\n",
" 'pending_sends': []\n",
" },\n",
" metadata={\n",
" 'source': 'loop',\n",
@@ -855,7 +856,7 @@
" 'id': '1f029ca3-1f5b-6704-8004-820c16b69a5a', \n",
" 'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000006.0.3205149138784782', 'branch:to:call_model': '00000000000000000000000000000006.0.14611156755133758'}, \n",
" 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}},\n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]},\n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\"), AIMessage(content='Your name is Bob.')]}, 'pending_sends': []\n",
" },\n",
" metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'}, \n",
" parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}}, \n",
@@ -869,7 +870,8 @@
" 'id': '1f029ca3-1790-6b0a-8003-baf965b6a38f', \n",
" 'channel_versions': {'__start__': '00000000000000000000000000000005.0.5290678567601859', 'messages': '00000000000000000000000000000005.0.7935064215293443', 'branch:to:call_model': '00000000000000000000000000000005.0.1410174088651449'}, \n",
" 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000004.0.5736472536395331'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\")], 'branch:to:call_model': None}\n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content=\"what's my name?\")], 'branch:to:call_model': None}, \n",
" 'pending_sends': []\n",
" }, \n",
" metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'}, \n",
" parent_config={...}, \n",
@@ -883,7 +885,8 @@
" 'id': '1f029ca3-1790-616e-8002-9e021694a0cd', \n",
" 'channel_versions': {'__start__': '00000000000000000000000000000004.0.5736472536395331', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, \n",
" 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
" 'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}, 'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\n",
" 'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}, 'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, \n",
" 'pending_sends': []\n",
" }, \n",
" metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': \"what's my name?\"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'}, \n",
" parent_config={...}, \n",
@@ -897,7 +900,8 @@
" 'id': '1f029ca3-178d-6f54-8001-d7b180db0c89', \n",
" 'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000003.0.7056767754077798', 'branch:to:call_model': '00000000000000000000000000000003.0.22059023329132854'}, \n",
" 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, 'call_model': {'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}}, \n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}\n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, \n",
" 'pending_sends': []\n",
" }, \n",
" metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'}, \n",
" parent_config={...}, \n",
@@ -911,7 +915,8 @@
" 'id': '1f029ca3-0874-6612-8000-339f2abc83b1', \n",
" 'channel_versions': {'__start__': '00000000000000000000000000000002.0.18673090920108737', 'messages': '00000000000000000000000000000002.0.30296526818059655', 'branch:to:call_model': '00000000000000000000000000000002.0.9300422176788571'}, \n",
" 'versions_seen': {'__input__': {}, '__start__': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}}, \n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\")], 'branch:to:call_model': None}\n",
" 'channel_values': {'messages': [HumanMessage(content=\"hi! I'm bob\")], 'branch:to:call_model': None}, \n",
" 'pending_sends': []\n",
" }, \n",
" metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'}, \n",
" parent_config={...}, \n",
@@ -925,7 +930,8 @@
" 'id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565', \n",
" 'channel_versions': {'__start__': '00000000000000000000000000000001.0.7040775356287469'}, \n",
" 'versions_seen': {'__input__': {}}, \n",
" 'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}\n",
" 'channel_values': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}, \n",
" 'pending_sends': []\n",
" }, \n",
" metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': \"hi! I'm bob\"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'}, \n",
" parent_config=None, \n",
@@ -1107,10 +1113,10 @@
"source": [
"### Use in production\n",
"\n",
"In production, you would want to use a store backed by a database:\n",
"In production, you would want to use a checkpointer backed by a database:\n",
"\n",
"```python\n",
"from langgraph.store.postgres import PostgresStore\n",
"from langgraph.checkpoint.postgres import PostgresSaver\n",
"\n",
"DB_URI = \"postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable\"\n",
"# highlight-next-line\n",
+1 -1
View File
@@ -321,7 +321,7 @@ attempts = 0
# The default RetryPolicy is optimized for retrying specific network errors.
retry_policy = RetryPolicy(retry_on=ValueError)
@task(retry_policy=retry_policy)
@task(retry=retry_policy)
def get_info():
global attempts
attempts += 1
+35
View File
@@ -36,6 +36,41 @@
- aget_subgraphs
- with_config
::: langgraph.graph.graph.Graph
options:
show_if_no_docstring: true
show_root_heading: true
show_root_full_path: false
members:
- add_node
- add_edge
- add_conditional_edges
- compile
::: langgraph.graph.graph.CompiledGraph
options:
show_if_no_docstring: true
show_root_heading: true
show_root_full_path: false
members:
- stream
- astream
- invoke
- ainvoke
- get_state
- aget_state
- get_state_history
- aget_state_history
- update_state
- aupdate_state
- bulk_update_state
- abulk_update_state
- get_graph
- aget_graph
- get_subgraphs
- aget_subgraphs
- with_config
::: langgraph.graph.message
options:
members:
+1 -1
View File
@@ -22,7 +22,7 @@ Welcome to the LangGraph reference docs! These pages detail the core interfaces
## LangGraph
The core APIs for the LangGraph open source library.
The core APIs for the LangGraph opens source library.
- [Graphs](graphs.md): Main graph abstraction and usage.
- [Functional API](func.md): Functional programming interface for graphs.
-16
View File
@@ -1,21 +1,5 @@
# Pregel
::: langgraph.pregel.NodeBuilder
options:
show_if_no_docstring: true
show_root_heading: true
show_root_full_path: false
members:
- subscribe_only
- subscribe_to
- read_from
- do
- write_to
- meta
- retry
- cache
- build
::: langgraph.pregel.Pregel
options:
show_if_no_docstring: true
@@ -580,7 +580,9 @@
" ]\n",
")\n",
"\n",
"evaluator = prompt | ChatOpenAI(model=\"gpt-4o\").with_structured_output(RedTeamingResult)\n",
"evaluator = prompt | ChatOpenAI(model=\"gpt-4-turbo-preview\").with_structured_output(\n",
" RedTeamingResult, method=\"function_calling\"\n",
")\n",
"\n",
"\n",
"def did_resist(run, example):\n",
@@ -833,7 +833,7 @@
"@tool\n",
"def book_excursion(recommendation_id: int) -> str:\n",
" \"\"\"\n",
" Book an excursion by its recommendation ID.\n",
" Book a excursion by its recommendation ID.\n",
"\n",
" Args:\n",
" recommendation_id (int): The ID of the trip recommendation to book.\n",
+3 -3
View File
@@ -89,7 +89,7 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 2,
"id": "baf669a0-04ee-492d-80d8-8fcb658ed128",
"metadata": {},
"outputs": [],
@@ -313,8 +313,8 @@
"\n",
" builder.add_edge(\"finalizer\", END)\n",
"\n",
" # These functions let the step be used in a\n",
" # StateGraph with 'messages' as the key.\n",
" # These functions let the step be used in a MessageGraph\n",
" # or a StateGraph with 'messages' as the key.\n",
" def encode(x: Union[Sequence[AnyMessage], PromptValue]) -> dict:\n",
" \"\"\"Ensure the input is the correct format.\"\"\"\n",
" if isinstance(x, PromptValue):\n",
@@ -1,6 +1,6 @@
# Build a basic chatbot
In this tutorial, you will build a basic chatbot. This chatbot is the basis for the following series of tutorials where you will progressively add more sophisticated capabilities, and be introduced to key LangGraph concepts along the way. Let's dive in! 🌟
In this tutorial, you will build a basic chatbot. This chatbot is the basis for the following series of tutorials where you will progressively add more sophisticated capabilities, and be introduced to key LangGraph concepts along the way. Lets dive in! 🌟
## Prerequisites
@@ -13,17 +13,9 @@ tool-calling features, such as [OpenAI](https://platform.openai.com/api-keys),
Install the required packages:
:::python
```bash
pip install -U langgraph langsmith
```
:::
:::js
```bash
npm install @langchain/langgraph @langchain/core langsmith
```
:::
!!! tip
@@ -35,13 +27,12 @@ Now you can create a basic chatbot using LangGraph. This chatbot will respond di
Start by creating a `StateGraph`. A `StateGraph` object defines the structure of our chatbot as a "state machine". We'll add `nodes` to represent the llm and functions our chatbot can call and `edges` to specify how the bot should transition between these functions.
:::python
```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
@@ -54,53 +45,24 @@ class State(TypedDict):
graph_builder = StateGraph(State)
```
:::
:::js
```typescript
import { Annotation } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";
import { StateGraph, START, END } from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
// Messages have the type "BaseMessage[]". The messagesStateReducer function
// defines how this state key should be updated
// (in this case, it appends messages to the list, rather than overwriting them)
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
});
const graphBuilder = new StateGraph(StateAnnotation);
```
:::
Our graph can now handle two key tasks:
1. Each `node` can receive the current `State` as input and output an update to the state.
2. Updates to `messages` will be appended to the existing list rather than overwriting it, thanks to the prebuilt function used with the annotation.
2. Updates to `messages` will be appended to the existing list rather than overwriting it, thanks to the prebuilt [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/?h=add+messages#add_messages) function used with the `Annotated` syntax.
------
!!! tip "Concept"
When defining a graph, the first step is to define its `State`. The `State` includes the graph's schema and [reducer functions](https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers) that handle state updates. Keys without a reducer annotation will overwrite previous values. To learn more about state, reducers, and related concepts, see [LangGraph reference docs](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages).
:::python
In our example, `State` is a `TypedDict` with one key: `messages`. The [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages) reducer function is used to append new messages to the list instead of overwriting it.
:::
:::js
In our example, `StateAnnotation` defines a state with one key: `messages`. The reducer function is used to append new messages to the list instead of overwriting it.
:::
When defining a graph, the first step is to define its `State`. The `State` includes the graph's schema and [reducer functions](https://langchain-ai.github.io/langgraph/concepts/low_level/#reducers) that handle state updates. In our example, `State` is a `TypedDict` with one key: `messages`. The [`add_messages`](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages) reducer function is used to append new messages to the list instead of overwriting it. Keys without a reducer annotation will overwrite previous values. To learn more about state, reducers, and related concepts, see [LangGraph reference docs](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages).
## 3. Add a node
Next, add a "`chatbot`" node. **Nodes** represent units of work and are typically regular functions.
Next, add a "`chatbot`" node. **Nodes** represent units of work and are typically regular Python functions.
Let's first select a chat model:
:::python
{!snippets/chat_model_tabs.md!}
<!---
@@ -110,21 +72,10 @@ from langchain.chat_models import init_chat_model
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
-->
:::
:::js
```typescript
import { ChatAnthropic } from "@langchain/anthropic";
const llm = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
});
```
:::
We can now incorporate the chat model into a simple node:
:::python
```python
def chatbot(state: State):
@@ -136,87 +87,32 @@ def chatbot(state: State):
# the node is used.
graph_builder.add_node("chatbot", chatbot)
```
:::
:::js
```typescript
const chatbot = async (state: typeof StateAnnotation.State) => {
return { messages: [await llm.invoke(state.messages)] };
};
// The first argument is the unique node name
// The second argument is the function or object that will be called whenever
// the node is used.
graphBuilder.addNode("chatbot", chatbot);
```
:::
**Notice** how the `chatbot` node function takes the current `State` as input and returns a dictionary containing an updated `messages` list under the key "messages". This is the basic pattern for all LangGraph node functions.
:::python
The `add_messages` function in our `State` will append the LLM's response messages to whatever messages are already in the state.
:::
:::js
The reducer function in our `StateAnnotation` will append the LLM's response messages to whatever messages are already in the state.
:::
## 4. Add an `entry` point
Add an `entry` point to tell the graph **where to start its work** each time it is run:
:::python
```python
graph_builder.add_edge(START, "chatbot")
```
:::
:::js
```typescript
graphBuilder.addEdge(START, "chatbot");
```
:::
## 5. Add an `exit` point
Add an `exit` point to indicate **where the graph should finish execution**. This is helpful for more complex flows, but even in a simple graph like this, adding an end node improves clarity.
:::python
```python
graph_builder.add_edge("chatbot", END)
```
:::
:::js
```typescript
graphBuilder.addEdge("chatbot", END);
```
:::
This tells the graph to terminate after running the chatbot node.
## 6. Compile the graph
## 5. Compile the graph
Before running the graph, we'll need to compile it. We can do so by calling `compile()`
on the graph builder. This creates a `CompiledGraph` we can invoke on our state.
:::python
```python
graph = graph_builder.compile()
```
:::
:::js
```typescript
const graph = graphBuilder.compile();
```
:::
## 7. Visualize the graph (optional)
## 6. Visualize the graph (optional)
You can visualize the graph using the `get_graph` method and one of the "draw" methods, like `draw_ascii` or `draw_png`. The `draw` methods each require additional dependencies.
:::python
```python
from IPython.display import Image, display
@@ -226,31 +122,14 @@ except Exception:
# This requires some extra dependencies and is optional
pass
```
:::
:::js
```typescript
import * as tslab from "tslab";
try {
const drawableGraph = graph.getGraph();
const image = await drawableGraph.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();
await tslab.display.png(new Uint8Array(arrayBuffer));
} catch (error) {
// This requires some extra dependencies and is optional
console.log("Graph visualization not available");
}
```
:::
![basic chatbot diagram](basic-chatbot.png)
## 8. Run the chatbot
## 7. Run the chatbot
Now run the chatbot!
:::python
!!! tip
You can exit the chat loop at any time by typing `quit`, `exit`, or `q`.
@@ -281,48 +160,18 @@ while True:
Assistant: LangGraph is a library designed to help build stateful multi-agent applications using language models. It provides tools for creating workflows and state machines to coordinate multiple AI agents or language model interactions. LangGraph is built on top of LangChain, leveraging its components while adding graph-based coordination capabilities. It's particularly useful for developing more complex, stateful AI applications that go beyond simple query-response interactions.
Goodbye!
```
:::
:::js
```typescript
import { HumanMessage } from "@langchain/core/messages";
async function streamGraphUpdates(userInput: string) {
const stream = await graph.stream({
messages: [new HumanMessage(userInput)]
});
for await (const event of stream) {
for (const value of Object.values(event)) {
console.log("Assistant:", value.messages[value.messages.length - 1].content);
}
}
}
// Example usage
const userInput = "What do you know about LangGraph?";
console.log("User:", userInput);
await streamGraphUpdates(userInput);
```
```
User: What do you know about LangGraph?
Assistant: LangGraph is a library designed to help build stateful multi-agent applications using language models. It provides tools for creating workflows and state machines to coordinate multiple AI agents or language model interactions. LangGraph is built on top of LangChain, leveraging its components while adding graph-based coordination capabilities. It's particularly useful for developing more complex, stateful AI applications that go beyond simple query-response interactions.
```
:::
**Congratulations!** You've built your first chatbot using LangGraph. This bot can engage in basic conversation by taking user input and generating responses using an LLM. You can inspect a [LangSmith Trace](https://smith.langchain.com/public/7527e308-9502-4894-b347-f34385740d5a/r) for the call above.
Below is the full code for this tutorial:
:::python
```python
from typing import Annotated
from langchain.chat_models import init_chat_model
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
@@ -345,44 +194,11 @@ def chatbot(state: State):
# the node is used.
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()
```
:::
:::js
```typescript
import { Annotation } from "@langchain/langgraph";
import { BaseMessage, HumanMessage } from "@langchain/core/messages";
import { StateGraph, START, END } from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
});
const graphBuilder = new StateGraph(StateAnnotation);
const llm = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
});
const chatbot = async (state: typeof StateAnnotation.State) => {
return { messages: [await llm.invoke(state.messages)] };
};
// The first argument is the unique node name
// The second argument is the function or object that will be called whenever
// the node is used.
graphBuilder.addNode("chatbot", chatbot);
graphBuilder.addEdge(START, "chatbot");
graphBuilder.addEdge("chatbot", END);
const graph = graphBuilder.compile();
```
:::
## Next steps
You may have noticed that the bot's knowledge is limited to what's in its training data. In the next part, we'll [add a web search tool](./2-add-tools.md) to expand the bot's knowledge and make it more capable.
You may have noticed that the bot's knowledge is limited to what's in its training data. In the next part, we'll [add a web search tool](./2-add-tools.md) to expand the bot's knowledge and make it more capable.
+2 -314
View File
@@ -1,6 +1,6 @@
# Add tools
To handle queries that your chatbot can't answer "from memory", integrate a web search tool. The chatbot can use this tool to find relevant information and provide better responses.
To handle queries you chatbot can't answer "from memory", integrate a web search tool. The chatbot can use this tool to find relevant information and provide better responses.
!!! note
@@ -8,39 +8,19 @@ To handle queries that your chatbot can't answer "from memory", integrate a web
## Prerequisites
:::python
Before you start this tutorial, ensure you have the following:
- An API key for the [Tavily Search Engine](https://python.langchain.com/docs/integrations/tools/tavily_search/).
:::
:::js
Before you start this tutorial, ensure you have the following:
- An API key for the [Tavily Search Engine](https://js.langchain.com/docs/integrations/tools/tavily_search/).
:::
## 1. Install the search engine
:::python
Install the requirements to use the [Tavily Search Engine](https://python.langchain.com/docs/integrations/tools/tavily_search/):
```bash
pip install -U langchain-tavily
```
:::
:::js
Install the requirements to use the [Tavily Search Engine](https://js.langchain.com/docs/integrations/tools/tavily_search/):
```bash
npm install @langchain/community
```
:::
## 2. Configure your environment
:::python
Configure your environment with your search engine API key:
```bash
@@ -50,21 +30,11 @@ _set_env("TAVILY_API_KEY")
```
TAVILY_API_KEY: ········
```
:::
:::js
Configure your environment with your search engine API key:
```typescript
process.env.TAVILY_API_KEY = "tvly-...";
```
:::
## 3. Define the tool
Define the web search tool:
:::python
```python
from langchain_tavily import TavilySearch
@@ -72,21 +42,9 @@ tool = TavilySearch(max_results=2)
tools = [tool]
tool.invoke("What's a 'node' in LangGraph?")
```
:::
:::js
```typescript
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
const tool = new TavilySearchResults({ maxResults: 2 });
const tools = [tool];
await tool.invoke("What's a 'node' in LangGraph?");
```
:::
The results are page summaries our chat bot can use to answer questions:
:::python
```
{'query': "What's a 'node' in LangGraph?",
'follow_up_questions': None,
@@ -104,17 +62,9 @@ The results are page summaries our chat bot can use to answer questions:
'raw_content': None}],
'response_time': 1.38}
```
:::
:::js
```
'[{"title":"Introduction to LangGraph: A Beginner\'s Guide - Medium","url":"https://medium.com/@cplog/introduction-to-langgraph-a-beginners-guide-14f9be027141","content":"Stateful Graph: LangGraph revolves around the concept of a stateful graph, where each node in the graph represents a step in your computation, and the graph maintains a state that is passed around and updated as the computation progresses. LangGraph supports conditional edges, allowing you to dynamically determine the next node to execute based on the current state of the graph. We define nodes for classifying the input, handling greetings, and handling search queries. def classify_input_node(state): LangGraph is a versatile tool for building complex, stateful applications with LLMs. By understanding its core concepts and working through simple examples, beginners can start to leverage its power for their projects. Remember to pay attention to state management, conditional edges, and ensuring there are no dead-end nodes in your graph.","score":0.7065353,"raw_content":null},{"title":"LangGraph Tutorial: What Is LangGraph and How to Use It?","url":"https://www.datacamp.com/tutorial/langgraph-tutorial","content":"LangGraph is a library within the LangChain ecosystem that provides a framework for defining, coordinating, and executing multiple LLM agents (or chains) in a structured and efficient manner. By managing the flow of data and the sequence of operations, LangGraph allows developers to focus on the high-level logic of their applications rather than the intricacies of agent coordination. Whether you need a chatbot that can handle various types of user requests or a multi-agent system that performs complex tasks, LangGraph provides the tools to build exactly what you need. LangGraph significantly simplifies the development of complex LLM applications by providing a structured framework for managing state and coordinating agent interactions.","score":0.5008063,"raw_content":null}]'
```
:::
## 4. Define the graph
:::python
For the `StateGraph` you created in the [first tutorial](./1-build-basic-chatbot.md), add `bind_tools` on the LLM. This lets the LLM know the correct JSON format to use if it wants to use the search engine.
Let's first select our LLM:
@@ -153,52 +103,9 @@ def chatbot(state: State):
graph_builder.add_node("chatbot", chatbot)
```
:::
:::js
For the `StateGraph` you created in the [first tutorial](./1-build-basic-chatbot.md), add `bindTools` on the LLM. This lets the LLM know the correct JSON format to use if it wants to use the search engine.
Let's first select our LLM:
```typescript
import { ChatOpenAI } from "@langchain/openai";
const llm = new ChatOpenAI({
model: "gpt-4o",
temperature: 0,
});
```
We can now incorporate it into a `StateGraph`:
```typescript hl_lines="15"
import { Annotation } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
});
import { StateGraph, START, END } from "@langchain/langgraph";
const graphBuilder = new StateGraph(StateAnnotation);
// Modification: tell the LLM which tools it can call
const llmWithTools = llm.bindTools(tools);
const chatbot = async (state: typeof StateAnnotation.State) => {
return { messages: [await llmWithTools.invoke(state.messages)] };
};
graphBuilder.addNode("chatbot", chatbot);
```
:::
## 5. Create a function to run the tools
:::python
Now, create a function to run the tools if they are called. Do this by adding the tools to a new node called`BasicToolNode` that checks the most recent message in the state and calls tools if the message contains `tool_calls`. It relies on the LLM's `tool_calling` support, which is available in Anthropic, OpenAI, Google Gemini, and a number of other LLM providers.
```python
@@ -236,50 +143,6 @@ class BasicToolNode:
tool_node = BasicToolNode(tools=[tool])
graph_builder.add_node("tools", tool_node)
```
:::
:::js
Now, create a function to run the tools if they are called. Do this by adding the tools to a new node called `BasicToolNode` that checks the most recent message in the state and calls tools if the message contains `tool_calls`. It relies on the LLM's `tool_calling` support, which is available in Anthropic, OpenAI, Google Gemini, and a number of other LLM providers.
```typescript
import { ToolMessage } from "@langchain/core/messages";
class BasicToolNode {
private toolsByName: Record<string, any>;
constructor(tools: any[]) {
this.toolsByName = {};
for (const tool of tools) {
this.toolsByName[tool.name] = tool;
}
}
async __call__(inputs: Record<string, any>): Promise<{ messages: ToolMessage[] }> {
const messages = inputs.messages || [];
if (messages.length === 0) {
throw new Error("No message found in input");
}
const message = messages[messages.length - 1];
const outputs: ToolMessage[] = [];
for (const toolCall of message.tool_calls || []) {
const toolResult = await this.toolsByName[toolCall.name].invoke(toolCall.args);
outputs.push(
new ToolMessage({
content: JSON.stringify(toolResult),
name: toolCall.name,
tool_call_id: toolCall.id,
})
);
}
return { messages: outputs };
}
}
const toolNode = new BasicToolNode([tool]);
graphBuilder.addNode("tools", async (state) => toolNode.__call__(state));
```
:::
!!! note
@@ -291,7 +154,6 @@ With the tool node added, now you can define the `conditional_edges`.
**Edges** route the control flow from one node to the next. **Conditional edges** start from a single node and usually contain "if" statements to route to different nodes depending on the current graph state. These functions receive the current graph `state` and return a string or list of strings indicating which node(s) to call next.
:::python
Next, define a router function called `route_tools` that checks for `tool_calls` in the chatbot's output. Provide this function to the graph by calling `add_conditional_edges`, which tells the graph that whenever the `chatbot` node completes to check this function to see where to go next.
The condition will route to `tools` if tool calls are present and `END` if not. Because the condition can return `END`, you do not need to explicitly set a `finish_point` this time.
@@ -332,51 +194,6 @@ graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()
```
:::
:::js
Next, define a router function called `routeTools` that checks for `tool_calls` in the chatbot's output. Provide this function to the graph by calling `addConditionalEdges`, which tells the graph that whenever the `chatbot` node completes to check this function to see where to go next.
The condition will route to `tools` if tool calls are present and `END` if not. Because the condition can return `END`, you do not need to explicitly set a `finish_point` this time.
```typescript
import { AIMessage } from "@langchain/core/messages";
const routeTools = (state: typeof StateAnnotation.State) => {
/**
* Use in the conditional_edge to route to the ToolNode if the last message
* has tool calls. Otherwise, route to the end.
*/
const messages = state.messages;
const lastMessage = messages[messages.length - 1] as AIMessage;
if (lastMessage.tool_calls && lastMessage.tool_calls.length > 0) {
return "tools";
}
return END;
};
// The `routeTools` function returns "tools" if the chatbot asks to use a tool, and "END" if
// it is fine directly responding. This conditional routing defines the main agent loop.
graphBuilder.addConditionalEdges(
"chatbot",
routeTools,
// The following dictionary lets you tell the graph to interpret the condition's outputs as a specific node
// It defaults to the identity function, but if you
// want to use a node named something else apart from "tools",
// You can update the value of the dictionary to something else
// e.g., "tools": "my_tools"
{
tools: "tools",
[END]: END,
}
);
// Any time a tool is called, we return to the chatbot to decide the next step
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
const graph = graphBuilder.compile();
```
:::
!!! note
@@ -384,7 +201,6 @@ const graph = graphBuilder.compile();
## 7. Visualize the graph (optional)
:::python
You can visualize the graph using the `get_graph` method and one of the "draw" methods, like `draw_ascii` or `draw_png`. The `draw` methods each require additional dependencies.
```python
@@ -396,26 +212,6 @@ except Exception:
# This requires some extra dependencies and is optional
pass
```
:::
:::js
You can visualize the graph using the `getGraph` method and one of the "draw" methods, like `drawAscii` or `drawMermaidPng`. The `draw` methods each require additional dependencies.
```typescript
import * as tslab from "tslab";
try {
const representation = graph.getGraph();
const image = await representation.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();
await tslab.display.png(new Uint8Array(arrayBuffer));
} catch (error) {
// This requires some extra dependencies and is optional
console.log("Graph visualization not available");
}
```
:::
![chatbot-with-tools-diagram](chatbot-with-tools.png)
@@ -423,7 +219,6 @@ try {
Now you can ask the chatbot questions outside its training data:
:::python
```python
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
@@ -479,71 +274,11 @@ LangGraph appears to be a significant tool in the evolving landscape of LLM-base
Goodbye!
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
:::
:::js
```typescript
import { HumanMessage } from "@langchain/core/messages";
const streamGraphUpdates = async (userInput: string) => {
const stream = await graph.stream(
{ messages: [new HumanMessage(userInput)] },
{ streamMode: "values" }
);
for await (const event of stream) {
const messages = event.messages;
const lastMessage = messages[messages.length - 1];
console.log("Assistant:", lastMessage.content);
}
};
// Example usage
const userInput = "What do you know about LangGraph?";
console.log("User:", userInput);
await streamGraphUpdates(userInput);
```
```
Assistant: I'll search for information about LangGraph to provide you with accurate details.
Assistant: [{"title": "Introduction to LangGraph: A Beginner's Guide - Medium", "url": "https://medium.com/@cplog/introduction-to-langgraph-a-beginners-guide-14f9be027141", "content": "Stateful Graph: LangGraph revolves around the concept of a stateful graph, where each node in the graph represents a step in your computation, and the graph maintains a state that is passed around and updated as the computation progresses. LangGraph supports conditional edges, allowing you to dynamically determine the next node to execute based on the current state of the graph. We define nodes for classifying the input, handling greetings, and handling search queries. def classify_input_node(state): LangGraph is a versatile tool for building complex, stateful applications with LLMs. By understanding its core concepts and working through simple examples, beginners can start to leverage its power for their projects. Remember to pay attention to state management, conditional edges, and ensuring there are no dead-end nodes in your graph.", "score": 0.7065353, "raw_content": null}, {"title": "LangGraph Tutorial: What Is LangGraph and How to Use It?", "url": "https://www.datacamp.com/tutorial/langgraph-tutorial", "content": "LangGraph is a library within the LangChain ecosystem that provides a framework for defining, coordinating, and executing multiple LLM agents or chains in a structured and efficient manner. By managing the flow of data and the sequence of operations, LangGraph allows developers to focus on the high-level logic of their applications rather than the intricacies of agent coordination. Whether you need a chatbot that can handle various types of user requests or a multi-agent system that performs complex tasks, LangGraph provides the tools to build exactly what you need. LangGraph significantly simplifies the development of complex LLM applications by providing a structured framework for managing state and coordinating agent interactions.", "score": 0.5008063, "raw_content": null}]
Assistant: Based on the search results, I can provide you with comprehensive information about LangGraph:
## What is LangGraph?
LangGraph is a library within the LangChain ecosystem designed for building stateful, multi-actor applications with Large Language Models (LLMs). It provides a framework for defining, coordinating, and executing multiple LLM agents or chains in a structured and efficient manner.
## Key Features:
1. **Stateful Graph Architecture**: LangGraph revolves around the concept of a stateful graph where each node represents a step in your computation, and the graph maintains state that is passed around and updated as the computation progresses.
2. **Conditional Edges**: It supports conditional edges, allowing you to dynamically determine the next node to execute based on the current state of the graph.
3. **Multi-Agent Coordination**: LangGraph manages the flow of data and sequence of operations, allowing developers to focus on high-level logic rather than the intricacies of agent coordination.
## Use Cases:
- Building conversational agents
- Creating chatbots that can handle various types of user requests
- Developing multi-agent systems that perform complex tasks
- Complex task automation
- Custom LLM-backed experiences
## Benefits:
- **Simplified Development**: LangGraph significantly simplifies the development of complex LLM applications by providing a structured framework for managing state and coordinating agent interactions.
- **Flexibility**: It's a versatile tool for building complex, stateful applications with LLMs.
- **Focus on Logic**: Developers can focus on the high-level logic of their applications rather than coordination details.
LangGraph is particularly valuable for projects that require sophisticated AI workflows with multiple steps, decision points, and state management across different components.
```
:::
## 9. Use prebuilts
For ease of use, adjust your code to replace the following with LangGraph prebuilt components. These have built in functionality like parallel API execution.
:::python
- `BasicToolNode` is replaced with the prebuilt [ToolNode](https://langchain-ai.github.io/langgraph/reference/prebuilt/#toolnode)
- `route_tools` is replaced with the prebuilt [tools_condition](https://langchain-ai.github.io/langgraph/reference/prebuilt/#tools_condition)
@@ -587,56 +322,9 @@ graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()
```
:::
:::js
- `BasicToolNode` is replaced with the prebuilt [ToolNode](https://langchain-ai.github.io/langgraph/reference/prebuilt/#toolnode)
- `routeTools` is replaced with the prebuilt [tools_condition](https://langchain-ai.github.io/langgraph/reference/prebuilt/#tools_condition)
```typescript hl_lines="25 30"
import { Annotation } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, START, END } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
});
const graphBuilder = new StateGraph(StateAnnotation);
const tool = new TavilySearchResults({ maxResults: 2 });
const tools = [tool];
const llm = new ChatOpenAI({ model: "gpt-4o", temperature: 0 });
const llmWithTools = llm.bindTools(tools);
const chatbot = async (state: typeof StateAnnotation.State) => {
return { messages: [await llmWithTools.invoke(state.messages)] };
};
graphBuilder.addNode("chatbot", chatbot);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
graphBuilder.addConditionalEdges(
"chatbot",
toolsCondition,
);
// Any time a tool is called, we return to the chatbot to decide the next step
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
const graph = graphBuilder.compile();
```
:::
**Congratulations!** You've created a conversational agent in LangGraph that can use a search engine to retrieve updated information when needed. Now it can handle a wider range of user queries. To inspect all the steps your agent just took, check out this [LangSmith trace](https://smith.langchain.com/public/4fbd7636-25af-4638-9587-5a02fdbb0172/r).
## Next steps
The chatbot cannot remember past interactions on its own, which limits its ability to have coherent, multi-turn conversations. In the next part, you will [add **memory**](./3-add-memory.md) to address this.
The chatbot cannot remember past interactions on its own, which limits its ability to have coherent, multi-turn conversations. In the next part, you will [add **memory**](./3-add-memory.md) to address this.
+1 -236
View File
@@ -14,21 +14,11 @@ We will see later that **checkpointing** is _much_ more powerful than simple cha
Create a `MemorySaver` checkpointer:
:::python
``` python
from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
```
:::
:::js
```typescript
import { MemorySaver } from "@langchain/langgraph";
const memory = new MemorySaver();
```
:::
This is in-memory checkpointer, which is convenient for the tutorial. However, in a production application, you would likely change this to use `SqliteSaver` or `PostgresSaver` and connect a database.
@@ -36,7 +26,6 @@ This is in-memory checkpointer, which is convenient for the tutorial. However, i
Compile the graph with the provided checkpointer, which will checkpoint the `State` as the graph works through each node:
:::python
``` python
graph = graph_builder.compile(checkpointer=memory)
```
@@ -50,27 +39,6 @@ except Exception:
# This requires some extra dependencies and is optional
pass
```
:::
:::js
```typescript
const graph = graphBuilder.compile({ checkpointer: memory });
```
```typescript
import * as tslab from "tslab";
try {
const representation = graph.getGraph();
const image = await representation.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();
await tslab.display.png(new Uint8Array(arrayBuffer));
} catch (e) {
// This requires some extra dependencies and is optional
}
```
:::
## 3. Interact with your chatbot
@@ -78,21 +46,12 @@ Now you can interact with your bot!
1. Pick a thread to use as the key for this conversation.
:::python
```python
config = {"configurable": {"thread_id": "1"}}
```
:::
:::js
```typescript
const config = { configurable: { thread_id: "1" } };
```
:::
2. Call your chatbot:
:::python
```python
user_input = "Hi there! My name is Will."
@@ -105,24 +64,6 @@ Now you can interact with your bot!
for event in events:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
const userInput = "Hi there! My name is Will.";
// The config is the **second positional argument** to stream() or invoke()!
const events = await graph.stream(
{ messages: [{ role: "user", content: userInput }] },
{ ...config, streamMode: "values" }
);
for await (const event of events) {
const messages = event.messages;
console.log(messages[messages.length - 1]);
}
```
:::
```
================================ Human Message =================================
@@ -133,23 +74,14 @@ Now you can interact with your bot!
Hello Will! It's nice to meet you. How can I assist you today? Is there anything specific you'd like to know or discuss?
```
:::python
!!! note
The config was provided as the **second positional argument** when calling our graph. It importantly is _not_ nested within the graph inputs (`{'messages': []}`).
:::
:::js
!!! note
The config was provided as the **second positional argument** when calling our graph. It importantly is _not_ nested within the graph inputs (`{ messages: [] }`).
:::
## 4. Ask a follow up question
Ask a follow up question:
:::python
```python
user_input = "Remember my name?"
@@ -162,24 +94,6 @@ events = graph.stream(
for event in events:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
const userInput2 = "Remember my name?";
// The config is the **second positional argument** to stream() or invoke()!
const events2 = await graph.stream(
{ messages: [{ role: "user", content: userInput2 }] },
{ ...config, streamMode: "values" }
);
for await (const event of events2) {
const messages = event.messages;
console.log(messages[messages.length - 1]);
}
```
:::
```
================================ Human Message =================================
@@ -194,7 +108,6 @@ Of course, I remember your name, Will. I always try to pay attention to importan
Don't believe me? Try this using a different config.
:::python
```python
# The only difference is we change the `thread_id` here to "2" instead of "1"
events = graph.stream(
@@ -206,23 +119,6 @@ events = graph.stream(
for event in events:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
// The only difference is we change the `thread_id` here to "2" instead of "1"
const events3 = await graph.stream(
{ messages: [{ role: "user", content: userInput2 }] },
// highlight-next-line
{ configurable: { thread_id: "2" }, streamMode: "values" }
);
for await (const event of events3) {
const messages = event.messages;
console.log(messages[messages.length - 1]);
}
```
:::
```
================================ Human Message =================================
@@ -237,15 +133,8 @@ I apologize, but I don't have any previous context or memory of your name. As an
## 5. Inspect the state
:::python
By now, we have made a few checkpoints across two different threads. But what goes into a checkpoint? To inspect a graph's `state` for a given config at any time, call `get_state(config)`.
:::
:::js
By now, we have made a few checkpoints across two different threads. But what goes into a checkpoint? To inspect a graph's `state` for a given config at any time, call `getState(config)`.
:::
:::python
```python
snapshot = graph.get_state(config)
snapshot
@@ -258,75 +147,6 @@ StateSnapshot(values={'messages': [HumanMessage(content='Hi there! My name is Wi
```
snapshot.next # (since the graph ended this turn, `next` is empty. If you fetch a state from within a graph invocation, next tells which node will execute next)
```
:::
:::js
```typescript
const snapshot = await graph.getState(config);
console.log(snapshot);
```
```
StateSnapshot {
values: {
messages: [
HumanMessage {
content: 'Hi there! My name is Will.',
id: '8c1ca919-c553-4ebf-95d4-b59a2d61e078'
},
AIMessage {
content: "Hello Will! It's nice to meet you. How can I assist you today? Is there anything specific you'd like to know or discuss?",
id: 'run-58587b77-8c82-41e6-8a90-d62c444a261d-0'
},
HumanMessage {
content: 'Remember my name?',
id: 'daba7df6-ad75-4d6b-8057-745881cea1ca'
},
AIMessage {
content: "Of course, I remember your name, Will. I always try to pay attention to important details that users share with me. Is there anything else you'd like to talk about or any questions you have? I'm here to help with a wide range of topics or tasks.",
id: 'run-ffeaae5c-4d2d-4ddb-bd59-5d5cbf2a5af8-0'
}
]
},
next: [],
config: {
configurable: {
thread_id: '1',
checkpoint_ns: '',
checkpoint_id: '1ef7d06e-93e0-6acc-8004-f2ac846575d2'
}
},
metadata: {
source: 'loop',
writes: {
chatbot: {
messages: [
AIMessage {
content: "Of course, I remember your name, Will. I always try to pay attention to important details that users share with me. Is there anything else you'd like to talk about or any questions you have? I'm here to help with a wide range of topics or tasks.",
id: 'run-ffeaae5c-4d2d-4ddb-bd59-5d5cbf2a5af8-0'
}
]
}
},
step: 4,
parents: {}
},
createdAt: '2024-09-27T19:30:10.820758+00:00',
parentConfig: {
configurable: {
thread_id: '1',
checkpoint_ns: '',
checkpoint_id: '1ef7d06e-859f-6206-8003-e1bd3c264b8f'
}
},
tasks: []
}
```
```typescript
console.log(snapshot.next); // (since the graph ended this turn, `next` is empty. If you fetch a state from within a graph invocation, next tells which node will execute next)
```
:::
The snapshot above contains the current state values, corresponding config, and the `next` node to process. In our case, the graph has reached an `END` state, so `next` is empty.
@@ -337,25 +157,14 @@ Check out the code snippet below to review the graph from this tutorial:
{!snippets/chat_model_tabs.md!}
<!---
:::python
```python
from langchain.chat_models import init_chat_model
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
:::
:::js
```typescript
import { ChatOpenAI } from "@langchain/openai";
const llm = new ChatOpenAI({ model: "gpt-4" });
```
:::
-->
:::python
```python hl_lines="36 37"
```python
from typing import Annotated
from langchain.chat_models import init_chat_model
@@ -394,50 +203,6 @@ graph_builder.set_entry_point("chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript hl_lines="36 37"
import { Annotation } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { BaseMessage } from "@langchain/core/messages";
import { MemorySaver, StateGraph } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
});
const graphBuilder = new StateGraph(StateAnnotation);
const tool = new TavilySearchResults({ maxResults: 2 });
const tools = [tool];
const llm = new ChatOpenAI({ model: "gpt-4" });
const llmWithTools = llm.bindTools(tools);
function chatbot(state: typeof StateAnnotation.State) {
return { messages: [llmWithTools.invoke(state.messages)] };
}
graphBuilder.addNode("chatbot", chatbot);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
graphBuilder.addConditionalEdges(
"chatbot",
toolsCondition,
);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge("__start__", "chatbot");
const memory = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer: memory });
```
:::
## Next steps
@@ -14,7 +14,6 @@ Starting with the existing code from the [Add memory to the chatbot](./3-add-mem
Let's first select a chat model:
:::python
{!snippets/chat_model_tabs.md!}
<!---
@@ -24,21 +23,9 @@ from langchain.chat_models import init_chat_model
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
-->
:::
:::js
```typescript
import { ChatAnthropic } from "@langchain/anthropic";
const llm = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
});
```
:::
We can now incorporate it into our `StateGraph` with an additional tool:
:::python
``` python hl_lines="12 19 20 21 22 23"
from typing import Annotated
@@ -88,60 +75,6 @@ graph_builder.add_conditional_edges(
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")
```
:::
:::js
```typescript hl_lines="12 19 20 21 22 23"
import { tool } from "@langchain/core/tools";
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { z } from "zod";
import { MemorySaver } from "@langchain/langgraph";
import { StateGraph, START, END, MessagesAnnotation } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { interrupt, Command } from "@langchain/langgraph";
const humanAssistance = tool(async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
}, {
name: "human_assistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human")
})
});
const searchTool = new TavilySearchResults({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = llm.bindTools(tools);
const chatbot = async (state: typeof MessagesAnnotation.State) => {
const message = await llmWithTools.invoke(state.messages);
// Because we will be interrupting during tool execution,
// we disable parallel tool calling to avoid repeating any
// tool invocations when we resume.
if (message.tool_calls && message.tool_calls.length > 1) {
throw new Error("Multiple tool calls not supported for this example");
}
return { messages: [message] };
};
const graphBuilder = new StateGraph(MessagesAnnotation)
.addNode("chatbot", chatbot);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
graphBuilder.addConditionalEdges(
"chatbot",
toolsCondition,
);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
```
:::
!!! tip
@@ -151,27 +84,16 @@ graphBuilder.addEdge(START, "chatbot");
We compile the graph with a checkpointer, as before:
:::python
```python
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
const memory = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer: memory });
```
:::
## 3. Visualize the graph (optional)
Visualizing the graph, you get the same layout as before just with the added tool!
:::python
``` python
from IPython.display import Image, display
@@ -181,19 +103,6 @@ except Exception:
# This requires some extra dependencies and is optional
pass
```
:::
:::js
```typescript
import * as tslab from "tslab";
const drawableGraph = graph.getGraph();
const image = await drawableGraph.drawMermaidPng();
const arrayBuffer = await image.arrayBuffer();
await tslab.display.png(new Uint8Array(arrayBuffer));
```
:::
![chatbot-with-tools-diagram](chatbot-with-tools.png)
@@ -201,7 +110,6 @@ await tslab.display.png(new Uint8Array(arrayBuffer));
Now, prompt the chatbot with a question that will engage the new `human_assistance` tool:
:::python
```python
user_input = "I need some expert guidance for building an AI agent. Could you request assistance for me?"
config = {"configurable": {"thread_id": "1"}}
@@ -229,49 +137,9 @@ Tool Calls:
Args:
query: A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?
```
:::
:::js
```typescript
const userInput = "I need some expert guidance for building an AI agent. Could you request assistance for me?";
const config = { configurable: { thread_id: "1" }, streamMode: "values" as const };
const events = graph.stream(
{ messages: [{ role: "user", content: userInput }] },
config,
);
for await (const event of events) {
if (event.messages) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage.getType()} Message =================================`);
console.log(lastMessage.content);
if (lastMessage.tool_calls?.length) {
console.log("Tool Calls:");
lastMessage.tool_calls.forEach((call) => {
console.log(` ${call.name} (${call.id})`);
console.log(` Args: ${JSON.stringify(call.args)}`);
});
}
}
}
```
```
================================ Human Message =================================
I need some expert guidance for building an AI agent. Could you request assistance for me?
================================== Ai Message ==================================
I'd be happy to request expert assistance for you regarding building an AI agent. Let me use the human assistance function to get you some expert guidance.
Tool Calls:
human_assistance (toolu_01ABUqneqnuHNuo1vhfDFQCW)
Args: {"query":"A user is requesting expert guidance for building an AI agent. Could you please provide some expert advice or resources on this topic?"}
```
:::
The chatbot generated a tool call, but then execution has been interrupted. If you inspect the graph state, you see that it stopped at the tools node:
:::python
```python
snapshot = graph.get_state(config)
snapshot.next
@@ -280,20 +148,7 @@ snapshot.next
```
('tools',)
```
:::
:::js
```typescript
const snapshot = await graph.getState(config);
console.log(snapshot.next);
```
```
['tools']
```
:::
:::python
!!! info Additional information
Take a closer look at the `human_assistance` tool:
@@ -307,34 +162,11 @@ console.log(snapshot.next);
```
Similar to Python's built-in `input()` function, calling `interrupt` inside the tool will pause execution. Progress is persisted based on the [checkpointer](../../concepts/persistence.md#checkpointer-libraries); so if it is persisting with Postgres, it can resume at any time as long as the database is alive. In this example, it is persisting with the in-memory checkpointer and can resume any time if the Python kernel is running.
:::
:::js
!!! info Additional information
Take a closer look at the `human_assistance` tool:
```typescript
const humanAssistance = tool(async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
}, {
name: "human_assistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human")
})
});
```
Similar to Python's built-in `input()` function, calling `interrupt` inside the tool will pause execution. Progress is persisted based on the [checkpointer](../../concepts/persistence.md#checkpointer-libraries); so if it is persisting with Postgres, it can resume at any time as long as the database is alive. In this example, it is persisting with the in-memory checkpointer and can resume any time if the JavaScript runtime is running.
:::
## 5. Resume execution
To resume execution, pass a [`Command`](../../concepts/low_level.md#command) object containing data expected by the tool. The format of this data can be customized based on needs. For this example, use a dict with a key `"data"`:
:::python
``` python
human_response = (
"We, the experts are here to help! We'd recommend you check out LangGraph to build your agent."
@@ -382,47 +214,6 @@ LangGraph is likely a framework or library designed specifically for creating AI
If you'd like more specific information about LangGraph or have any questions about this recommendation, please feel free to ask, and I can request further assistance from the experts.
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
:::
:::js
```typescript
const humanResponse =
"We, the experts are here to help! We'd recommend you check out LangGraph to build your agent." +
" It's much more reliable and extensible than simple autonomous agents.";
const humanCommand = new Command({ resume: { data: humanResponse } });
const resumeEvents = graph.stream(humanCommand, config);
for await (const event of resumeEvents) {
if (event.messages) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage.getType()} Message =================================`);
console.log(lastMessage.content);
}
}
```
```
================================== Ai Message ==================================
I'd be happy to request expert assistance for you regarding building an AI agent. Let me use the human assistance function to get you some expert guidance.
================================= Tool Message =================================
We, the experts are here to help! We'd recommend you check out LangGraph to build your agent. It's much more reliable and extensible than simple autonomous agents.
================================== Ai Message ==================================
Thank you for your patience. I've received some expert advice regarding your request for guidance on building an AI agent. Here's what the experts have suggested:
The experts recommend that you look into LangGraph for building your AI agent. They mention that LangGraph is a more reliable and extensible option compared to simple autonomous agents.
LangGraph is likely a framework or library designed specifically for creating AI agents with advanced capabilities. Here are a few points to consider based on this recommendation:
1. Reliability: The experts emphasize that LangGraph is more reliable than simpler autonomous agent approaches. This could mean it has better stability, error handling, or consistent performance.
2. Extensibility: LangGraph is described as more extensible, which suggests that it probably offers a flexible architecture that allows you to easily add new features or modify existing ones as your agent's requirements evolve.
3. Advanced capabilities: Given that it's recommended over "simple autonomous agents," LangGraph likely provides more sophisticated tools and techniques for building complex AI agents.
...
```
:::
The input has been received and processed as a tool message. Review this call's [LangSmith trace](https://smith.langchain.com/public/9f0f87e3-56a7-4dde-9c76-b71675624e91/r) to see the exact work that was done in the above call. Notice that the state is loaded in the first step so that our chatbot can continue where it left off.
@@ -430,7 +221,6 @@ The input has been received and processed as a tool message. Review this call's
Check out the code snippet below to review the graph from this tutorial:
:::python
{!snippets/chat_model_tabs.md!}
```python
@@ -481,64 +271,6 @@ graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
import { tool } from "@langchain/core/tools";
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { z } from "zod";
import { ChatAnthropic } from "@langchain/anthropic";
import { MemorySaver } from "@langchain/langgraph";
import { StateGraph, START, END, MessagesAnnotation } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { interrupt, Command } from "@langchain/langgraph";
const llm = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
});
const humanAssistance = tool(async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
}, {
name: "human_assistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human")
})
});
const searchTool = new TavilySearchResults({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = llm.bindTools(tools);
const chatbot = async (state: typeof MessagesAnnotation.State) => {
const message = await llmWithTools.invoke(state.messages);
if (message.tool_calls && message.tool_calls.length > 1) {
throw new Error("Multiple tool calls not supported for this example");
}
return { messages: [message] };
};
const graphBuilder = new StateGraph(MessagesAnnotation)
.addNode("chatbot", chatbot);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
graphBuilder.addConditionalEdges(
"chatbot",
toolsCondition,
);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
const memory = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer: memory });
```
:::
## Next steps
@@ -10,7 +10,6 @@ In this tutorial, you will add additional fields to the state to define complex
Update the chatbot to research the birthday of an entity by adding `name` and `birthday` keys to the state:
:::python
```python
from typing import Annotated
@@ -26,30 +25,11 @@ class State(TypedDict):
# highlight-next-line
birthday: str
```
:::
:::js
```typescript
import { Annotation } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
// highlight-next-line
name: Annotation<string>,
// highlight-next-line
birthday: Annotation<string>,
});
```
:::
Adding this information to the state makes it easily accessible by other graph nodes (like a downstream node that stores or processes the information), as well as the graph's persistence layer.
## 2. Update the state inside the tool
:::python
Now, populate the state keys inside of the `human_assistance` tool. This allows a human to review the information before it is stored in the state. Use [`Command`](../../concepts/low_level.md#using-inside-tools) to issue a state update from inside the tool.
``` python
@@ -95,73 +75,11 @@ def human_assistance(
# We return a Command object in the tool to update our state.
return Command(update=state_update)
```
:::
:::js
Now, populate the state keys inside of the `humanAssistance` tool. This allows a human to review the information before it is stored in the state. Use [`Command`](../../concepts/low_level.md#using-inside-tools) to issue a state update from inside the tool.
```typescript
import { tool } from "@langchain/core/tools";
import { ToolMessage } from "@langchain/core/messages";
import { z } from "zod";
import { Command, interrupt } from "@langchain/langgraph";
const humanAssistance = tool(async (input, config) => {
const { name, birthday } = input;
// Note that because we are generating a ToolMessage for a state update, we
// generally require the ID of the corresponding tool call. We can access this
// from the tool's config when it's called by a model.
const toolCallId = config?.toolCall?.id;
const humanResponse = interrupt({
question: "Is this correct?",
name: name,
birthday: birthday,
});
let verifiedName, verifiedBirthday, response;
// If the information is correct, update the state as-is.
if (humanResponse?.correct?.toLowerCase().startsWith("y")) {
verifiedName = name;
verifiedBirthday = birthday;
response = "Correct";
} else {
// Otherwise, receive information from the human reviewer.
verifiedName = humanResponse?.name || name;
verifiedBirthday = humanResponse?.birthday || birthday;
response = `Made a correction: ${JSON.stringify(humanResponse)}`;
}
// This time we explicitly update the state with a ToolMessage inside
// the tool.
const stateUpdate = {
name: verifiedName,
birthday: verifiedBirthday,
messages: [new ToolMessage({
content: response,
tool_call_id: toolCallId!
})],
};
// We return a Command object in the tool to update our state.
return new Command({ update: stateUpdate });
}, {
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
name: z.string(),
birthday: z.string(),
}),
});
```
:::
The rest of the graph stays the same.
## 3. Prompt the chatbot
:::python
Prompt the chatbot to look up the "birthday" of the LangGraph library and direct the chatbot to reach out to the `human_assistance` tool once it has the required information. By setting `name` and `birthday` in the arguments for the tool, you force the chatbot to generate proposals for these fields.
```python
@@ -180,30 +98,6 @@ for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
```
:::
:::js
Prompt the chatbot to look up the "birthday" of the LangGraph library and direct the chatbot to reach out to the `humanAssistance` tool once it has the required information. By setting `name` and `birthday` in the arguments for the tool, you force the chatbot to generate proposals for these fields.
```typescript
const userInput = "Can you look up when LangGraph was released? " +
"When you have the answer, use the humanAssistance tool for review.";
const config = { configurable: { thread_id: "1" } };
const events = graph.stream(
{ messages: [{ role: "user", content: userInput }] },
{ ...config, streamMode: "values" }
);
for await (const event of events) {
if (event.messages) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage._getType()} Message =================================`);
console.log(lastMessage.content);
}
}
```
:::
```
================================ Human Message =================================
@@ -236,7 +130,6 @@ We've hit the `interrupt` in the `human_assistance` tool again.
## 4. Add human assistance
:::python
The chatbot failed to identify the correct date, so supply it with information:
```python
@@ -252,32 +145,6 @@ for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
```
:::
:::js
The chatbot failed to identify the correct date, so supply it with information:
```typescript
import { Command } from "@langchain/langgraph";
const humanCommand = new Command({
resume: {
name: "LangGraph",
birthday: "Jan 17, 2024",
},
});
const resumeEvents = graph.stream(humanCommand, { ...config, streamMode: "values" });
for await (const event of resumeEvents) {
if (event.messages) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage._getType()} Message =================================`);
console.log(lastMessage.content);
}
}
```
:::
```
================================== Ai Message ==================================
@@ -308,25 +175,11 @@ It's worth noting that LangGraph had been in development and use for some time b
Note that these fields are now reflected in the state:
:::python
```python
snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
```
:::
:::js
```typescript
const snapshot = await graph.getState(config);
const relevantState = {
name: snapshot.values.name,
birthday: snapshot.values.birthday
};
console.log(relevantState);
```
:::
```
{'name': 'LangGraph', 'birthday': 'Jan 17, 2024'}
@@ -336,21 +189,11 @@ This makes them easily accessible to downstream nodes (e.g., a node that further
## 5. Manually update the state
:::python
LangGraph gives a high degree of control over the application state. For instance, at any point (including when interrupted), you can manually override a key using `graph.update_state`:
``` python
graph.update_state(config, {"name": "LangGraph (library)"})
```
:::
:::js
LangGraph gives a high degree of control over the application state. For instance, at any point (including when interrupted), you can manually override a key using `graph.updateState`:
```typescript
await graph.updateState(config, { name: "LangGraph (library)" });
```
:::
```
{'configurable': {'thread_id': '1',
@@ -360,7 +203,6 @@ await graph.updateState(config, { name: "LangGraph (library)" });
## 6. View the new value
:::python
If you call `graph.get_state`, you can see the new value is reflected:
``` python
@@ -368,21 +210,6 @@ snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
```
:::
:::js
If you call `graph.getState`, you can see the new value is reflected:
```typescript
const updatedSnapshot = await graph.getState(config);
const updatedState = {
name: updatedSnapshot.values.name,
birthday: updatedSnapshot.values.birthday
};
console.log(updatedState);
```
:::
```
{'name': 'LangGraph (library)', 'birthday': 'Jan 17, 2024'}
@@ -404,7 +231,6 @@ llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
-->
:::python
```python
from typing import Annotated
@@ -478,106 +304,8 @@ graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { tool } from "@langchain/core/tools";
import { ToolMessage, BaseMessage } from "@langchain/core/messages";
import { z } from "zod";
import { MemorySaver } from "@langchain/langgraph";
import { StateGraph, START, Annotation } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { Command, interrupt } from "@langchain/langgraph";
const llm = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
});
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
name: Annotation<string>,
birthday: Annotation<string>,
});
const humanAssistance = tool(async (input, config) => {
const { name, birthday } = input;
const toolCallId = config?.toolCall?.id;
const humanResponse = interrupt({
question: "Is this correct?",
name: name,
birthday: birthday,
});
let verifiedName, verifiedBirthday, response;
if (humanResponse?.correct?.toLowerCase().startsWith("y")) {
verifiedName = name;
verifiedBirthday = birthday;
response = "Correct";
} else {
verifiedName = humanResponse?.name || name;
verifiedBirthday = humanResponse?.birthday || birthday;
response = `Made a correction: ${JSON.stringify(humanResponse)}`;
}
const stateUpdate = {
name: verifiedName,
birthday: verifiedBirthday,
messages: [new ToolMessage({
content: response,
tool_call_id: toolCallId!
})],
};
return new Command({ update: stateUpdate });
}, {
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
name: z.string(),
birthday: z.string(),
}),
});
const searchTool = new TavilySearchResults({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = llm.bindTools(tools);
const chatbot = async (state: typeof StateAnnotation.State) => {
const message = await llmWithTools.invoke(state.messages);
return { messages: [message] };
};
const shouldContinue = (state: typeof StateAnnotation.State) => {
const lastMessage = state.messages[state.messages.length - 1];
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
return "tools";
}
return "__end__";
};
const graphBuilder = new StateGraph(StateAnnotation);
graphBuilder.addNode("chatbot", chatbot);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
graphBuilder.addConditionalEdges("chatbot", shouldContinue);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
const memory = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer: memory });
```
:::
## Next steps
There's one more concept to review before finishing the LangGraph basics tutorials: connecting `checkpointing` and `state updates` to [time travel](./6-time-travel.md).
There's one more concept to review before finishing the LangGraph basics tutorials: connecting `checkpointing` and `state updates` to [time travel](./6-time-travel.md).
@@ -12,35 +12,18 @@ You can create these types of experiences using LangGraph's built-in **time trav
## 1. Rewind your graph
:::python
Rewind your graph by fetching a checkpoint using the graph's `get_state_history` method. You can then resume execution at this previous point in time.
:::
:::js
Rewind your graph by fetching a checkpoint using the graph's `getStateHistory` method. You can then resume execution at this previous point in time.
:::
{!snippets/chat_model_tabs.md!}
<!---
:::python
```python
from langchain.chat_models import init_chat_model
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
:::
:::js
```typescript
import { initChatModel } from "langchain/chat_models/init";
const llm = initChatModel("anthropic:claude-3-5-sonnet-latest");
```
:::
-->
:::python
```python
from typing import Annotated
@@ -80,62 +63,11 @@ graph_builder.add_edge(START, "chatbot")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
import { TavilySearchResults } from "@langchain/community/tools/tavily_search";
import { ChatAnthropic } from "@langchain/anthropic";
import { BaseMessage } from "@langchain/core/messages";
import { Annotation, StateGraph, START, END } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { messagesStateReducer } from "@langchain/langgraph";
const StateAnnotation = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: messagesStateReducer,
}),
});
const graphBuilder = new StateGraph(StateAnnotation);
const tool = new TavilySearchResults({ maxResults: 2 });
const tools = [tool];
const llm = new ChatAnthropic({ model: "claude-3-5-sonnet-latest" });
const llmWithTools = llm.bindTools(tools);
const chatbot = async (state: typeof StateAnnotation.State) => {
return { messages: [await llmWithTools.invoke(state.messages)] };
};
graphBuilder.addNode("chatbot", chatbot);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
const toolsCondition = (state: typeof StateAnnotation.State) => {
const lastMessage = state.messages[state.messages.length - 1];
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
return "tools";
}
return END;
};
graphBuilder.addConditionalEdges("chatbot", toolsCondition);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
const memory = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer: memory });
```
:::
## 2. Add steps
Add steps to your graph. Every step will be checkpointed in its state history:
:::python
``` python
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
@@ -157,42 +89,6 @@ for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
const config = { configurable: { thread_id: "1" } };
const events = await graph.stream(
{
messages: [
{
role: "user",
content: (
"I'm learning LangGraph. " +
"Could you do some research on it for me?"
),
},
],
},
{ ...config, streamMode: "values" }
);
for await (const event of events) {
if ("messages" in event) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage._getType()} Message =================================`);
console.log(lastMessage.content);
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
console.log("Tool Calls:");
for (const toolCall of lastMessage.tool_calls) {
console.log(` ${toolCall.name} (${toolCall.id})`);
console.log(` Args: ${JSON.stringify(toolCall.args)}`);
}
}
}
}
```
:::
```
================================ Human Message =================================
@@ -227,7 +123,6 @@ Is there any specific aspect of LangGraph you'd like to know more about? I'd be
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
:::python
```python
events = graph.stream(
{
@@ -248,41 +143,6 @@ for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
const events2 = await graph.stream(
{
messages: [
{
role: "user",
content: (
"Ya that's helpful. Maybe I'll " +
"build an autonomous agent with it!"
),
},
],
},
{ ...config, streamMode: "values" }
);
for await (const event of events2) {
if ("messages" in event) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage._getType()} Message =================================`);
console.log(lastMessage.content);
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
console.log("Tool Calls:");
for (const toolCall of lastMessage.tool_calls) {
console.log(` ${toolCall.name} (${toolCall.id})`);
console.log(` Args: ${JSON.stringify(toolCall.args)}`);
}
}
}
}
```
:::
```
================================ Human Message =================================
@@ -299,7 +159,7 @@ Tool Calls:
================================= Tool Message =================================
Name: tavily_search_results_json
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a user's question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a users question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
================================== Ai Message ==================================
Great idea! Building an autonomous agent with LangGraph is definitely an exciting project. Based on the latest information I've found, here are some insights and tips for building autonomous agents with LangGraph:
@@ -321,7 +181,6 @@ Output is truncated. View as a scrollable element or open in a text editor. Adju
Now that you have added steps to the chatbot, you can `replay` the full state history to see everything that occurred.
:::python
``` python
to_replay = None
for state in graph.get_state_history(config):
@@ -331,24 +190,7 @@ for state in graph.get_state_history(config):
# We are somewhat arbitrarily selecting a specific state based on the number of chat messages in the state.
to_replay = state
```
:::
:::js
```typescript
let toReplay = null;
const stateHistory = await graph.getStateHistory(config);
for await (const state of stateHistory) {
console.log("Num Messages: ", state.values.messages.length, "Next: ", state.next);
console.log("-".repeat(80));
if (state.values.messages.length === 6) {
// We are somewhat arbitrarily selecting a specific state based on the number of chat messages in the state.
toReplay = state;
}
}
```
:::
:::python
```
Num Messages: 8 Next: ()
--------------------------------------------------------------------------------
@@ -371,32 +213,6 @@ Num Messages: 1 Next: ('chatbot',)
Num Messages: 0 Next: ('__start__',)
--------------------------------------------------------------------------------
```
:::
:::js
```
Num Messages: 8 Next: []
--------------------------------------------------------------------------------
Num Messages: 7 Next: ["chatbot"]
--------------------------------------------------------------------------------
Num Messages: 6 Next: ["tools"]
--------------------------------------------------------------------------------
Num Messages: 5 Next: ["chatbot"]
--------------------------------------------------------------------------------
Num Messages: 4 Next: ["__start__"]
--------------------------------------------------------------------------------
Num Messages: 4 Next: []
--------------------------------------------------------------------------------
Num Messages: 3 Next: ["chatbot"]
--------------------------------------------------------------------------------
Num Messages: 2 Next: ["tools"]
--------------------------------------------------------------------------------
Num Messages: 1 Next: ["chatbot"]
--------------------------------------------------------------------------------
Num Messages: 0 Next: ["__start__"]
--------------------------------------------------------------------------------
```
:::
Checkpoints are saved for every step of the graph. This __spans invocations__ so you can rewind across a full thread's history.
@@ -404,74 +220,27 @@ Checkpoints are saved for every step of the graph. This __spans invocations__ so
Resume from the `to_replay` state, which is after the `chatbot` node in the second graph invocation. Resuming from this point will call the **action** node next.
:::python
```python
print(to_replay.next)
print(to_replay.config)
```
:::
:::js
```typescript
console.log(toReplay.next);
console.log(toReplay.config);
```
:::
:::python
```
('tools',)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1efd43e3-0c1f-6c4e-8006-891877d65740'}}
```
:::
:::js
```
["tools"]
{
"configurable": {
"thread_id": "1",
"checkpoint_ns": "",
"checkpoint_id": "1efd43e3-0c1f-6c4e-8006-891877d65740"
}
}
```
:::
## 4. Load a state from a moment-in-time
The checkpoint's `to_replay.config` contains a `checkpoint_id` timestamp. Providing this `checkpoint_id` value tells LangGraph's checkpointer to **load** the state from that moment in time.
:::python
``` python
# The `checkpoint_id` in the `to_replay.config` corresponds to a state we've persisted to our checkpointer.
for event in graph.stream(None, to_replay.config, stream_mode="values"):
if "messages" in event:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
// The `checkpoint_id` in the `toReplay.config` corresponds to a state we've persisted to our checkpointer.
const timeTravel = await graph.stream(null, { ...toReplay.config, streamMode: "values" });
for await (const event of timeTravel) {
if ("messages" in event) {
const lastMessage = event.messages[event.messages.length - 1];
console.log(`================================ ${lastMessage._getType()} Message =================================`);
console.log(lastMessage.content);
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
console.log("Tool Calls:");
for (const toolCall of lastMessage.tool_calls) {
console.log(` ${toolCall.name} (${toolCall.id})`);
console.log(` Args: ${JSON.stringify(toolCall.args)}`);
}
}
}
}
```
:::
```
================================== Ai Message ==================================
@@ -485,7 +254,7 @@ Tool Calls:
================================= Tool Message =================================
Name: tavily_search_results_json
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a user's question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a users question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
================================== Ai Message ==================================
Great idea! Building an autonomous agent with LangGraph is indeed an excellent way to apply and deepen your understanding of the technology. Based on the search results, I can provide you with some insights and resources to help you get started:
@@ -471,7 +471,7 @@
"\n",
"_get_pass(\"TAVILY_API_KEY\")\n",
"\n",
"calculate = get_math_tool(ChatOpenAI(model=\"gpt-4o\"))\n",
"calculate = get_math_tool(ChatOpenAI(model=\"gpt-4-turbo-preview\"))\n",
"search = TavilySearchResults(\n",
" max_results=1,\n",
" description='tavily_search_results_json(query=\"the search query\") - a search engine.',\n",
@@ -540,11 +540,11 @@
"name": "stdout",
"output_type": "stream",
"text": [
"================================\u001B[1m System Message \u001B[0m================================\n",
"================================\u001b[1m System Message \u001b[0m================================\n",
"\n",
"Given a user query, create a plan to solve it with the utmost parallelizability. Each plan should comprise an action from the following \u001B[33;1m\u001B[1;3m{num_tools}\u001B[0m types:\n",
"\u001B[33;1m\u001B[1;3m{tool_descriptions}\u001B[0m\n",
"\u001B[33;1m\u001B[1;3m{num_tools}\u001B[0m. join(): Collects and combines results from prior actions.\n",
"Given a user query, create a plan to solve it with the utmost parallelizability. Each plan should comprise an action from the following \u001b[33;1m\u001b[1;3m{num_tools}\u001b[0m types:\n",
"\u001b[33;1m\u001b[1;3m{tool_descriptions}\u001b[0m\n",
"\u001b[33;1m\u001b[1;3m{num_tools}\u001b[0m. join(): Collects and combines results from prior actions.\n",
"\n",
" - An LLM agent is called upon invoking join() to either finalize the user query or wait until the plans are executed.\n",
" - join should always be the last action in the plan, and will be called in two scenarios:\n",
@@ -561,11 +561,11 @@
" - Only use the provided action types. If a query cannot be addressed using these, invoke the join action for the next steps.\n",
" - Never introduce new actions other than the ones provided.\n",
"\n",
"=============================\u001B[1m Messages Placeholder \u001B[0m=============================\n",
"=============================\u001b[1m Messages Placeholder \u001b[0m=============================\n",
"\n",
"\u001B[33;1m\u001B[1;3m{messages}\u001B[0m\n",
"\u001b[33;1m\u001b[1;3m{messages}\u001b[0m\n",
"\n",
"================================\u001B[1m System Message \u001B[0m================================\n",
"================================\u001b[1m System Message \u001b[0m================================\n",
"\n",
"Remember, ONLY respond with the task list in the correct format! E.g.:\n",
"idx. tool(arg_name=args)\n",
@@ -1030,7 +1030,7 @@
"joiner_prompt = hub.pull(\"wfh/llm-compiler-joiner\").partial(\n",
" examples=\"\"\n",
") # You can optionally add examples\n",
"llm = ChatOpenAI(model=\"gpt-4o\")\n",
"llm = ChatOpenAI(model=\"gpt-4-turbo-preview\")\n",
"\n",
"runnable = joiner_prompt | llm.with_structured_output(\n",
" JoinOutputs, method=\"function_calling\"\n",
File diff suppressed because one or more lines are too long
@@ -135,6 +135,7 @@
"metadata": {},
"outputs": [],
"source": [
"from langchain import hub\n",
"from langchain_openai import ChatOpenAI\n",
"\n",
"from langgraph.prebuilt import create_react_agent\n",
@@ -90,11 +90,7 @@
"id": "9ac1c2cd-81fb-40eb-8ba1-e9197800cba6",
"metadata": {},
"source": [
"## Create Index\n",
"\n",
"Set up a vector database using OpenAI Embeddings and the Chroma vector database. \n",
"Input URLs of blog posts related to agents, prompt engineering, and large language models (LLMs). \n",
"Generate vector indices for use in Retrieval-Augmented Generation (RAG)."
"## Create Index"
]
},
{
@@ -163,21 +159,6 @@
"</div>"
]
},
{
"cell_type": "markdown",
"id": "6cdd5ac0-fa18-4ee9-8051-062a0c56268f",
"metadata": {},
"source": [
"### Router for Query Analysis\n",
"\n",
"Lets start with Routing. First, assign the query analysis to the LLM.\n",
"\n",
"Create a RouteQuery data model and specify it in a structured format for the LLM. The decision for routing should be embedded in the prompt. You need to clearly define which parts of the document should be directed to RAG based on the topic.\n",
"\n",
"While you could automate this process by having the LLM summarize the RAG documents again, its more cost-effective to manually manage this when dealing with large documents, as automation could become expensive.\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": 4,
@@ -238,18 +219,6 @@
"print(question_router.invoke({\"question\": \"What are the types of agent memory?\"}))"
]
},
{
"cell_type": "markdown",
"id": "cb248c94-0b0c-4d86-8565-32aa8d7424e4",
"metadata": {},
"source": [
"### Retrieval Grader\n",
"\n",
"After performing retrieval, evaluate the results. Although you initially decided to use RAG based on the query, the retrieved documents might not be satisfactory. Assess whether the retrieved documents are sufficiently relevant to the query.\n",
"\n",
"For this, rely on the LLM to evaluate the relevance, providing a binary yes or no decision."
]
},
{
"cell_type": "code",
"execution_count": 5,
@@ -340,17 +309,6 @@
"print(generation)"
]
},
{
"cell_type": "markdown",
"id": "cb0ab54a-4a4f-45fa-b1c5-cea1bf4c59d5",
"metadata": {},
"source": [
"### Hallucination Grader\n",
"\n",
"Verify if the LLM produced any hallucinations by comparing its output to the retrieved facts. \n",
"Provide the LLMs evaluation in a binary yes or no format.\n"
]
},
{
"cell_type": "code",
"execution_count": 7,
@@ -399,16 +357,6 @@
"hallucination_grader.invoke({\"documents\": docs, \"generation\": generation})"
]
},
{
"cell_type": "markdown",
"id": "4f58502a-c25f-4d80-a402-5583b0cd3e41",
"metadata": {},
"source": [
"### Answer Grader\n",
"\n",
"Evaluate the answer finally."
]
},
{
"cell_type": "code",
"execution_count": 8,
@@ -457,18 +405,6 @@
"answer_grader.invoke({\"question\": question, \"generation\": generation})"
]
},
{
"cell_type": "markdown",
"id": "af77946c-2646-4039-86b0-e2fde1ab7459",
"metadata": {},
"source": [
"### Question Rewriting\n",
"\n",
"The original question from user was directly used in RAG. \n",
"However, the users question might not be in a form suitable for RAG. \n",
"To improve retrieval, rephrase the question to ensure it aligns better with vector similarity search."
]
},
{
"cell_type": "code",
"execution_count": 9,
@@ -514,9 +450,7 @@
"id": "d07c0b31-b919-4498-869f-9673125c2473",
"metadata": {},
"source": [
"## Web Search Tool\n",
"\n",
"Use Tavily Search tool to get information from the web."
"## Web Search Tool"
]
},
{
@@ -582,13 +516,11 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 12,
"id": "b76b5ec3-0720-443d-85b1-c0e79659ca0a",
"metadata": {},
"outputs": [],
"source": [
"from pprint import pprint\n",
"\n",
"from langchain.schema import Document\n",
"\n",
"\n",
@@ -864,7 +796,7 @@
},
{
"cell_type": "code",
"execution_count": null,
"execution_count": 14,
"id": "29acc541-d726-4b75-84d1-a215845fe88a",
"metadata": {},
"outputs": [
@@ -891,6 +823,8 @@
}
],
"source": [
"from pprint import pprint\n",
"\n",
"# Run\n",
"inputs = {\n",
" \"question\": \"What player at the Bears expected to draft first in the 2024 NFL draft?\"\n",
+1 -1
View File
@@ -185,7 +185,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeDocuments)\n",
"\n",
"# Prompt\n",
@@ -389,7 +389,7 @@
"text": [
"{'generate': {'messages': [AIMessage(content='Title: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nAntoine de Saint-Exupéry\\'s \"The Little Prince\" is a classic novella that has captured the hearts of millions since its publication in 1943. While it might be easy to dismiss this work as a children\\'s story, its profound themes and timeless message make it a relevant and topical piece in modern life. This essay will explore the allegorical nature of \"The Little Prince\" and discuss how its message can be applied to the complexities of the modern world.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\n\"The Little Prince\" is an allegorical tale that explores various aspects of the human condition through its whimsical characters and situations. The Little Prince himself represents innocence, curiosity, and the importance of human connection. As the story unfolds, readers encounter different characters that symbolize various aspects of adult life, such as vanity, materialism, and authority. These representations allow the story to transcend age and culture, making it relatable to a wide range of readers, even in the modern context.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe Little Prince\\'s message is centered around the importance of looking beyond superficial appearances and forming meaningful connections with others. In a world increasingly dominated by technology and social media, where surface-level interactions are commonplace, this message is more relevant than ever. The Little Prince encourages readers to cherish and nurture genuine relationships, reminding us that true happiness and fulfillment come from understanding and empathizing with others.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\n\"The Little Prince\" also offers a critique of modern society, highlighting the dangers of materialism, consumerism, and the pursuit of power. These themes resonate strongly in today\\'s world, where wealth inequality and environmental degradation are pressing issues. The story serves as a reminder that the pursuit of material possessions and status often comes at the expense of our own happiness and the well-being of our planet.\\n\\nConclusion:\\nIn conclusion, \"The Little Prince\" remains a topical and relevant work in modern life due to its allegorical nature, timeless message, and critique of modern society. Its exploration of human connections, materialism, and the pursuit of power offers valuable insights for readers of all ages. By embracing the story\\'s wisdom, we can better navigate the complexities of the modern world and foster a more compassionate, sustainable, and interconnected society.', response_metadata={'token_usage': {'prompt_tokens': 72, 'total_tokens': 632, 'completion_tokens': 560}, 'model_name': 'accounts/fireworks/models/mixtral-8x7b-instruct', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None}, id='run-b39a25ab-24f6-42d0-96c2-0f74c3ecc8f7-0', usage_metadata={'input_tokens': 72, 'output_tokens': 560, 'total_tokens': 632})]}}\n",
"---\n",
"{'reflect': {'messages': [HumanMessage(content='Essay Critique and Recommendations:\\n\\nTitle: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nThe introduction effectively sets the stage for the essay by providing background information on \"The Little Prince\" and its relevance in modern life. However, consider adding a hook to engage the reader\\'s attention and create a stronger first impression.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\nThis paragraph provides a clear explanation of the allegorical nature of \"The Little Prince.\" To enhance this section, consider offering specific examples from the text to illustrate how the characters and situations symbolize various aspects of adult life. This will strengthen your analysis and make it more engaging for the reader.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe relevance of the Little Prince\\'s message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince\\'s wisdom and its relevance to contemporary issues.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\nThis paragraph effectively highlights the story\\'s critique of modern society. To deepen your analysis, explore how themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince\\'s message that could help address these issues.\\n\\nConclusion:\\nThe conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story\\'s message and its implications for their own lives.\\n\\nRecommendations:\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay.\\n5. Revise and edit the essay for clarity, coherence, and grammar.')]}}\n",
"{'reflect': {'messages': [HumanMessage(content='Essay Critique and Recommendations:\\n\\nTitle: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nThe introduction effectively sets the stage for the essay by providing background information on \"The Little Prince\" and its relevance in modern life. However, consider adding a hook to engage the reader\\'s attention and create a stronger first impression.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\nThis paragraph provides a clear explanation of the allegorical nature of \"The Little Prince.\" To enhance this section, consider offering specific examples from the text to illustrate how the characters and situations symbolize various aspects of adult life. This will strengthen your analysis and make it more engaging for the reader.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe relevance of the Little Prince\\'s message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince\\'s wisdom and its relevance to contemporary issues.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\nThis paragraph effectively highlights the story\\'s critique of modern society. To deepen your analysis, explore how the themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince\\'s message that could help address these issues.\\n\\nConclusion:\\nThe conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story\\'s message and its implications for their own lives.\\n\\nRecommendations:\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay.\\n5. Revise and edit the essay for clarity, coherence, and grammar.')]}}\n",
"---\n",
"{'generate': {'messages': [AIMessage(content='Title: The Little Prince: A Topical Allegory for Modern Life\\n\\nIntroduction:\\nIn Antoine de Saint-Exupéry\\'s classic novella \"The Little Prince,\" a young boy embarks on a journey through the universe, meeting various characters that symbolize different aspects of adult life. This timeless tale, published in 1943, remains incredibly relevant in today\\'s modern world. Its allegorical nature, thought-provoking message, and critique of modern society offer invaluable insights for readers of all ages. This essay will explore the allegory of \"The Little Prince,\" analyze the relevance of its message, and discuss its critique of modern society, demonstrating its topicality in contemporary life.\\n\\nBody Paragraph 1 - The Allegory of the Little Prince:\\n\"The Little Prince\" is an allegorical tale that uses whimsical characters and situations to explore various aspects of the human condition. For instance, the king represents authority without substance, while the businessman embodies the futility of materialism. The fox, conversely, symbolizes the importance of forming genuine connections and nurturing meaningful relationships. These allegorical representations allow the story to transcend age and culture, making it relatable to a wide range of readers, even in the modern context.\\n\\nBody Paragraph 2 - The Relevance of the Little Prince\\'s Message:\\nThe Little Prince\\'s message is centered around the importance of looking beyond superficial appearances and forming meaningful connections with others. In a world increasingly dominated by technology and social media, where surface-level interactions are commonplace, this message is more relevant than ever. Neglecting this message can lead to feelings of isolation, loneliness, and dissatisfaction. By embracing the story\\'s wisdom, we can prioritize genuine relationships, fostering a more compassionate and interconnected society.\\n\\nBody Paragraph 3 - The Critique of Modern Society:\\n\"The Little Prince\" offers a critique of modern society, highlighting the dangers of materialism, consumerism, and the pursuit of power. These themes resonate strongly in today\\'s world, where wealth inequality and environmental degradation are pressing issues. The story serves as a reminder that the pursuit of material possessions and status often comes at the expense of our own happiness and the well-being of our planet. To address these challenges, we must reevaluate our priorities, focusing on sustainability, empathy, and the cultivation of meaningful relationships.\\n\\nConclusion:\\nIn conclusion, \"The Little Prince\" remains a topical and relevant work in modern life due to its allegorical nature, timeless message, and critique of modern society. Its exploration of human connections, materialism, and the pursuit of power offers valuable insights for readers of all ages. By embracing the story\\'s wisdom, we can better navigate the complexities of the modern world and foster a more compassionate, sustainable, and interconnected society. As the Little Prince so eloquently states, \"What is essential is invisible to the eye,\" reminding us that true happiness and fulfillment come from understanding and empathizing with others.\\n\\nExpanded Essay Recommendations:\\n\\n1. Expand the essay to approximately 1,200-1,500 words to allow for a more in-depth analysis.\\n2. Incorporate specific examples and quotes from \"The Little Prince\" to support your arguments and engage the reader. For instance, use quotes like, \"You become responsible, forever, for what you have tamed,\" to emphasize the importance of forming genuine connections.\\n3. Ensure that each body paragraph contains a clear thesis statement, supporting evidence, and analysis.\\n4. Consider discussing counterarguments or potential criticisms of the Little Prince\\'s message to add depth and complexity to your essay. For example, explore the idea that the pursuit of material possessions can provide a sense of security and comfort.\\n5. Revise and edit the essay for clarity, coherence, and grammar. Ensure that transitions between paragraphs are smooth and that your arguments flow logically.', response_metadata={'token_usage': {'prompt_tokens': 1168, 'total_tokens': 2044, 'completion_tokens': 876}, 'model_name': 'accounts/fireworks/models/mixtral-8x7b-instruct', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None}, id='run-9bfc9ff2-3186-43f5-8b75-498d532d8d1a-0', usage_metadata={'input_tokens': 1168, 'output_tokens': 876, 'total_tokens': 2044})]}}\n",
"---\n",
@@ -478,7 +478,7 @@
"The relevance of the Little Prince's message is well-articulated in this paragraph. To further strengthen your argument, consider discussing the consequences of ignoring this message in the context of modern society. This will help emphasize the importance of the Little Prince's wisdom and its relevance to contemporary issues.\n",
"\n",
"Body Paragraph 3 - The Critique of Modern Society:\n",
"This paragraph effectively highlights the story's critique of modern society. To deepen your analysis, explore how themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince's message that could help address these issues.\n",
"This paragraph effectively highlights the story's critique of modern society. To deepen your analysis, explore how the themes of materialism, consumerism, and the pursuit of power interconnect and contribute to the challenges faced by modern society. Additionally, consider discussing potential solutions or actions inspired by the Little Prince's message that could help address these issues.\n",
"\n",
"Conclusion:\n",
"The conclusion effectively summarizes the main points of the essay and emphasizes the relevance of \"The Little Prince\" in modern life. To further enhance this section, consider incorporating a thought-provoking question or statement that encourages readers to reflect on the story's message and its implications for their own lives.\n",
+1 -1
View File
@@ -1758,7 +1758,7 @@
"id": "4eb67198-c84f-458b-8baf-783d7246dddc",
"metadata": {},
"source": [
"Let's let the agent try again. Call `stream` with `None` to just use the inputs loaded from the memory. We will skip our human review for the next few attempts\n",
"Let's let the agent try again. Call `stream` with `None` to just use the inputs loaded from the memory. We will skip our human review for the next few attempats\n",
"to see if it can correct itself."
]
},
+2 -10
View File
@@ -179,10 +179,12 @@ nav:
- cloud/how-tos/studio/manage_assistants.md
- cloud/how-tos/threads_studio.md
- cloud/how-tos/iterate_graph_studio.md
- cloud/how-tos/studio/run_evals.md
- cloud/how-tos/clone_traces_studio.md
- cloud/how-tos/datasets_studio.md
- LangGraph SDK: concepts/sdk.md
- Data management:
- Data storage & privacy: concepts/data_storage_and_privacy.md
- Add semantic search: cloud/deployment/semantic_search.md
- Add TTLs: how-tos/ttl/configure_ttl.md
- Authentication & access control:
@@ -364,16 +366,6 @@ markdown_extensions:
hooks:
- _scripts/notebook_hooks.py
extra:
consent:
title: Cookie consent
actions:
- accept
- reject
description: >-
We use cookies to recognize your repeated visits and preferences, as well
as to measure the effectiveness of our documentation and whether users
find what they're searching for. <strong>Clicking "Accept" makes our
documentation better. Thank you!</strong> ❤️
social:
- icon: fontawesome/brands/js
link: https://langchain-ai.github.io/langgraphjs/
@@ -1,22 +0,0 @@
from _scripts.notebook_hooks import _apply_conditional_rendering
CONDITIONAL_RENDERING = """
above
:::js
js-content
:::
between
:::python
python-content
:::
below
"""
def test_conditional_rendering() -> None:
"""Test logic for conditional rendering of content."""
output = _apply_conditional_rendering(CONDITIONAL_RENDERING, "js")
assert output.strip() == "above\njs-content\n\nbetween\n\nbelow"
output = _apply_conditional_rendering(CONDITIONAL_RENDERING, "python")
assert output.strip() == "above\n\nbetween\npython-content\n\nbelow"
Generated
+3059 -3060
View File
File diff suppressed because it is too large Load Diff
+4 -4
View File
@@ -184,7 +184,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_router = llm.with_structured_output(RouteQuery)\n",
"\n",
"# Prompt\n",
@@ -235,7 +235,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeDocuments)\n",
"\n",
"# Prompt\n",
@@ -328,7 +328,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeHallucinations)\n",
"\n",
"# Prompt\n",
@@ -376,7 +376,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeAnswer)\n",
"\n",
"# Prompt\n",
+4 -4
View File
@@ -200,11 +200,11 @@
"output_type": "stream",
"text": [
"********************Prompt[rlm/rag-prompt]********************\n",
"================================\u001B[1m Human Message \u001B[0m=================================\n",
"================================\u001b[1m Human Message \u001b[0m=================================\n",
"\n",
"You are an assistant for question-answering tasks. Use the following pieces of retrieved context to answer the question. If you don't know the answer, just say that you don't know. Use three sentences maximum and keep the answer concise.\n",
"Question: \u001B[33;1m\u001B[1;3m{question}\u001B[0m \n",
"Context: \u001B[33;1m\u001B[1;3m{context}\u001B[0m \n",
"Question: \u001b[33;1m\u001b[1;3m{question}\u001b[0m \n",
"Context: \u001b[33;1m\u001b[1;3m{context}\u001b[0m \n",
"Answer:\n"
]
}
@@ -244,7 +244,7 @@
" binary_score: str = Field(description=\"Relevance score 'yes' or 'no'\")\n",
"\n",
" # LLM\n",
" model = ChatOpenAI(temperature=0, model=\"gpt-4o\", streaming=True)\n",
" model = ChatOpenAI(temperature=0, model=\"gpt-4-0125-preview\", streaming=True)\n",
"\n",
" # LLM with tool and validation\n",
" llm_with_tool = model.with_structured_output(grade)\n",
+1 -1
View File
@@ -171,7 +171,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeDocuments)\n",
"\n",
"# Prompt\n",
+3 -3
View File
@@ -191,7 +191,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeDocuments)\n",
"\n",
"# Prompt\n",
@@ -284,7 +284,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeHallucinations)\n",
"\n",
"# Prompt\n",
@@ -332,7 +332,7 @@
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"llm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeAnswer)\n",
"\n",
"# Prompt\n",
@@ -33,9 +33,7 @@
"id": "a384cc48-0425-4e8f-aafc-cfb8e56025c9",
"metadata": {},
"outputs": [],
"source": [
"%pip install -qU langchain-pinecone langchain-openai langchainhub langgraph"
]
"source": ["%pip install -qU langchain-pinecone langchain-openai langchainhub langgraph"]
},
{
"cell_type": "markdown",
@@ -53,9 +51,7 @@
"id": "ccc3dae5-1df6-48ca-af8a-50f0e6128876",
"metadata": {},
"outputs": [],
"source": [
"import os\n\nos.environ[\"LANGCHAIN_TRACING_V2\"] = \"true\"\nos.environ[\"LANGCHAIN_ENDPOINT\"] = \"https://api.smith.langchain.com\"\nos.environ[\"LANGCHAIN_API_KEY\"] = \"<your-api-key>\""
]
"source": ["import os\n\nos.environ[\"LANGCHAIN_TRACING_V2\"] = \"true\"\nos.environ[\"LANGCHAIN_ENDPOINT\"] = \"https://api.smith.langchain.com\"\nos.environ[\"LANGCHAIN_API_KEY\"] = \"<your-api-key>\""]
},
{
"cell_type": "code",
@@ -63,9 +59,7 @@
"id": "88637820",
"metadata": {},
"outputs": [],
"source": [
"import os\n\nos.environ[\"LANGCHAIN_PROJECT\"] = \"pinecone-devconnect\""
]
"source": ["import os\n\nos.environ[\"LANGCHAIN_PROJECT\"] = \"pinecone-devconnect\""]
},
{
"cell_type": "markdown",
@@ -83,9 +77,7 @@
"id": "565a6d44-2c9f-4fff-b1ec-eea05df9350d",
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import OpenAIEmbeddings\nfrom langchain_pinecone import PineconeVectorStore\n\n# use pinecone movies database\n\n# Add to vectorDB\nvectorstore = PineconeVectorStore(\n embedding=OpenAIEmbeddings(),\n index_name=\"sample-movies\",\n text_key=\"summary\",\n)\nretriever = vectorstore.as_retriever()"
]
"source": ["from langchain_openai import OpenAIEmbeddings\nfrom langchain_pinecone import PineconeVectorStore\n\n# use pinecone movies database\n\n# Add to vectorDB\nvectorstore = PineconeVectorStore(\n embedding=OpenAIEmbeddings(),\n index_name=\"sample-movies\",\n text_key=\"summary\",\n)\nretriever = vectorstore.as_retriever()"]
},
{
"cell_type": "code",
@@ -112,9 +104,7 @@
]
}
],
"source": [
"docs = retriever.invoke(\"James Cameron\")\nfor doc in docs:\n print(\"# \" + doc.metadata[\"title\"])\n print(doc.page_content)\n print()"
]
"source": ["docs = retriever.invoke(\"James Cameron\")\nfor doc in docs:\n print(\"# \" + doc.metadata[\"title\"])\n print(doc.page_content)\n print()"]
},
{
"cell_type": "markdown",
@@ -130,32 +120,7 @@
"id": "1fafad21-60cc-483e-92a3-6a7edb1838e3",
"metadata": {},
"outputs": [],
"source": [
"### Retrieval Grader\n",
"\n",
"from langchain import hub\n",
"from langchain_core.pydantic_v1 import BaseModel, Field\n",
"from langchain_openai import ChatOpenAI\n",
"\n",
"\n",
"# Data model\n",
"class GradeDocuments(BaseModel):\n",
" \"\"\"Binary score for relevance check on retrieved documents.\"\"\"\n",
"\n",
" binary_score: str = Field(\n",
" description=\"Documents are relevant to the question, 'yes' or 'no'\"\n",
" )\n",
"\n",
"\n",
"# https://smith.langchain.com/hub/efriis/self-rag-retrieval-grader\n",
"grade_prompt = hub.pull(\"efriis/self-rag-retrieval-grader\")\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeDocuments)\n",
"\n",
"retrieval_grader = grade_prompt | structured_llm_grader"
]
"source": ["### Retrieval Grader\n\nfrom langchain import hub\nfrom langchain_core.pydantic_v1 import BaseModel, Field\nfrom langchain_openai import ChatOpenAI\n\n\n# Data model\nclass GradeDocuments(BaseModel):\n \"\"\"Binary score for relevance check on retrieved documents.\"\"\"\n\n binary_score: str = Field(\n description=\"Documents are relevant to the question, 'yes' or 'no'\"\n )\n\n\n# https://smith.langchain.com/hub/efriis/self-rag-retrieval-grader\ngrade_prompt = hub.pull(\"efriis/self-rag-retrieval-grader\")\n\n# LLM with function call\nllm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\nstructured_llm_grader = llm.with_structured_output(GradeDocuments)\n\nretrieval_grader = grade_prompt | structured_llm_grader"]
},
{
"cell_type": "code",
@@ -172,9 +137,7 @@
]
}
],
"source": [
"# Test the retrieval grader\nquestion = \"movies starring jason momoa\"\ndocs = retriever.invoke(question)\ndoc_txt = docs[0].page_content\nprint(doc_txt)\nprint(retrieval_grader.invoke({\"question\": question, \"document\": doc_txt}))"
]
"source": ["# Test the retrieval grader\nquestion = \"movies starring jason momoa\"\ndocs = retriever.invoke(question)\ndoc_txt = docs[0].page_content\nprint(doc_txt)\nprint(retrieval_grader.invoke({\"question\": question, \"document\": doc_txt}))"]
},
{
"cell_type": "markdown",
@@ -200,9 +163,7 @@
]
}
],
"source": [
"### Generate\n\nfrom langchain import hub\nfrom langchain_core.output_parsers import StrOutputParser\n\n# Prompt\nprompt = hub.pull(\"rlm/rag-prompt\")\n\n# LLM\nllm = ChatOpenAI(model_name=\"gpt-3.5-turbo\", temperature=0)\n\n# Chain\nrag_chain = prompt | llm | StrOutputParser()\n\n# Run\ngeneration = rag_chain.invoke({\"context\": docs, \"question\": question})\nprint(generation)"
]
"source": ["### Generate\n\nfrom langchain import hub\nfrom langchain_core.output_parsers import StrOutputParser\n\n# Prompt\nprompt = hub.pull(\"rlm/rag-prompt\")\n\n# LLM\nllm = ChatOpenAI(model_name=\"gpt-3.5-turbo\", temperature=0)\n\n# Chain\nrag_chain = prompt | llm | StrOutputParser()\n\n# Run\ngeneration = rag_chain.invoke({\"context\": docs, \"question\": question})\nprint(generation)"]
},
{
"cell_type": "code",
@@ -228,30 +189,7 @@
"output_type": "execute_result"
}
],
"source": [
"### Hallucination Grader\n",
"\n",
"\n",
"# Data model\n",
"class GradeHallucinations(BaseModel):\n",
" \"\"\"Binary score for hallucination present in generation answer.\"\"\"\n",
"\n",
" binary_score: str = Field(\n",
" description=\"Answer is grounded in the facts, 'yes' or 'no'\"\n",
" )\n",
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeHallucinations)\n",
"\n",
"# https://smith.langchain.com/hub/efriis/self-rag-hallucination-grader\n",
"hallucination_prompt = hub.pull(\"efriis/self-rag-hallucination-grader\")\n",
"\n",
"hallucination_grader = hallucination_prompt | structured_llm_grader\n",
"print(generation)\n",
"hallucination_grader.invoke({\"documents\": docs, \"generation\": generation})"
]
"source": ["### Hallucination Grader\n\n\n# Data model\nclass GradeHallucinations(BaseModel):\n \"\"\"Binary score for hallucination present in generation answer.\"\"\"\n\n binary_score: str = Field(\n description=\"Answer is grounded in the facts, 'yes' or 'no'\"\n )\n\n\n# LLM with function call\nllm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\nstructured_llm_grader = llm.with_structured_output(GradeHallucinations)\n\n# https://smith.langchain.com/hub/efriis/self-rag-hallucination-grader\nhallucination_prompt = hub.pull(\"efriis/self-rag-hallucination-grader\")\n\nhallucination_grader = hallucination_prompt | structured_llm_grader\nprint(generation)\nhallucination_grader.invoke({\"documents\": docs, \"generation\": generation})"]
},
{
"cell_type": "code",
@@ -278,31 +216,7 @@
"output_type": "execute_result"
}
],
"source": [
"### Answer Grader\n",
"\n",
"\n",
"# Data model\n",
"class GradeAnswer(BaseModel):\n",
" \"\"\"Binary score to assess answer addresses question.\"\"\"\n",
"\n",
" binary_score: str = Field(\n",
" description=\"Answer addresses the question, 'yes' or 'no'\"\n",
" )\n",
"\n",
"\n",
"# LLM with function call\n",
"llm = ChatOpenAI(model=\"gpt-4o-mini\", temperature=0)\n",
"structured_llm_grader = llm.with_structured_output(GradeAnswer)\n",
"\n",
"# Prompt\n",
"answer_prompt = hub.pull(\"efriis/self-rag-answer-grader\")\n",
"\n",
"answer_grader = answer_prompt | structured_llm_grader\n",
"print(question)\n",
"print(generation)\n",
"answer_grader.invoke({\"question\": question, \"generation\": generation})"
]
"source": ["### Answer Grader\n\n\n# Data model\nclass GradeAnswer(BaseModel):\n \"\"\"Binary score to assess answer addresses question.\"\"\"\n\n binary_score: str = Field(\n description=\"Answer addresses the question, 'yes' or 'no'\"\n )\n\n\n# LLM with function call\nllm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\nstructured_llm_grader = llm.with_structured_output(GradeAnswer)\n\n# Prompt\nanswer_prompt = hub.pull(\"efriis/self-rag-answer-grader\")\n\nanswer_grader = answer_prompt | structured_llm_grader\nprint(question)\nprint(generation)\nanswer_grader.invoke({\"question\": question, \"generation\": generation})"]
},
{
"cell_type": "code",
@@ -328,9 +242,7 @@
"output_type": "execute_result"
}
],
"source": [
"### Question Re-writer\n\n# LLM\nllm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n\n# Prompt\nre_write_prompt = hub.pull(\"efriis/self-rag-question-rewriter\")\n\nquestion_rewriter = re_write_prompt | llm | StrOutputParser()\nprint(question)\nquestion_rewriter.invoke({\"question\": question})"
]
"source": ["### Question Re-writer\n\n# LLM\nllm = ChatOpenAI(model=\"gpt-3.5-turbo-0125\", temperature=0)\n\n# Prompt\nre_write_prompt = hub.pull(\"efriis/self-rag-question-rewriter\")\n\nquestion_rewriter = re_write_prompt | llm | StrOutputParser()\nprint(question)\nquestion_rewriter.invoke({\"question\": question})"]
},
{
"cell_type": "markdown",
@@ -350,9 +262,7 @@
"id": "f1617e9e-66a8-4c1a-a1fe-cc936284c085",
"metadata": {},
"outputs": [],
"source": [
"from typing import List\n\nfrom typing_extensions import TypedDict\n\n\nclass GraphState(TypedDict):\n \"\"\"\n Represents the state of our graph.\n\n Attributes:\n question: question\n generation: LLM generation\n documents: list of documents\n \"\"\"\n\n question: str\n generation: str\n documents: List[str]"
]
"source": ["from typing import List\n\nfrom typing_extensions import TypedDict\n\n\nclass GraphState(TypedDict):\n \"\"\"\n Represents the state of our graph.\n\n Attributes:\n question: question\n generation: LLM generation\n documents: list of documents\n \"\"\"\n\n question: str\n generation: str\n documents: List[str]"]
},
{
"cell_type": "code",
@@ -360,9 +270,7 @@
"id": "add509d8-6682-4127-8d95-13dd37d79702",
"metadata": {},
"outputs": [],
"source": [
"### Nodes\n\n\ndef retrieve(state):\n \"\"\"\n Retrieve documents\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): New key added to state, documents, that contains retrieved documents\n \"\"\"\n print(\"---RETRIEVE---\")\n question = state[\"question\"]\n\n # Retrieval\n documents = retriever.invoke(question)\n return {\"documents\": documents, \"question\": question}\n\n\ndef generate(state):\n \"\"\"\n Generate answer\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): New key added to state, generation, that contains LLM generation\n \"\"\"\n print(\"---GENERATE---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n\n # RAG generation\n generation = rag_chain.invoke({\"context\": documents, \"question\": question})\n return {\"documents\": documents, \"question\": question, \"generation\": generation}\n\n\ndef grade_documents(state):\n \"\"\"\n Determines whether the retrieved documents are relevant to the question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): Updates documents key with only filtered relevant documents\n \"\"\"\n\n print(\"---CHECK DOCUMENT RELEVANCE TO QUESTION---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n\n # Score each doc\n filtered_docs = []\n for d in documents:\n score = retrieval_grader.invoke(\n {\"question\": question, \"document\": d.page_content}\n )\n grade = score.binary_score\n if grade == \"yes\":\n print(\"---GRADE: DOCUMENT RELEVANT---\")\n filtered_docs.append(d)\n else:\n print(\"---GRADE: DOCUMENT NOT RELEVANT---\")\n continue\n return {\"documents\": filtered_docs, \"question\": question}\n\n\ndef transform_query(state):\n \"\"\"\n Transform the query to produce a better question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): Updates question key with a re-phrased question\n \"\"\"\n\n print(\"---TRANSFORM QUERY---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n\n # Re-write question\n better_question = question_rewriter.invoke({\"question\": question})\n return {\"documents\": documents, \"question\": better_question}"
]
"source": ["### Nodes\n\n\ndef retrieve(state):\n \"\"\"\n Retrieve documents\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): New key added to state, documents, that contains retrieved documents\n \"\"\"\n print(\"---RETRIEVE---\")\n question = state[\"question\"]\n\n # Retrieval\n documents = retriever.invoke(question)\n return {\"documents\": documents, \"question\": question}\n\n\ndef generate(state):\n \"\"\"\n Generate answer\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): New key added to state, generation, that contains LLM generation\n \"\"\"\n print(\"---GENERATE---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n\n # RAG generation\n generation = rag_chain.invoke({\"context\": documents, \"question\": question})\n return {\"documents\": documents, \"question\": question, \"generation\": generation}\n\n\ndef grade_documents(state):\n \"\"\"\n Determines whether the retrieved documents are relevant to the question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): Updates documents key with only filtered relevant documents\n \"\"\"\n\n print(\"---CHECK DOCUMENT RELEVANCE TO QUESTION---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n\n # Score each doc\n filtered_docs = []\n for d in documents:\n score = retrieval_grader.invoke(\n {\"question\": question, \"document\": d.page_content}\n )\n grade = score.binary_score\n if grade == \"yes\":\n print(\"---GRADE: DOCUMENT RELEVANT---\")\n filtered_docs.append(d)\n else:\n print(\"---GRADE: DOCUMENT NOT RELEVANT---\")\n continue\n return {\"documents\": filtered_docs, \"question\": question}\n\n\ndef transform_query(state):\n \"\"\"\n Transform the query to produce a better question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n state (dict): Updates question key with a re-phrased question\n \"\"\"\n\n print(\"---TRANSFORM QUERY---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n\n # Re-write question\n better_question = question_rewriter.invoke({\"question\": question})\n return {\"documents\": documents, \"question\": better_question}"]
},
{
"cell_type": "code",
@@ -370,9 +278,7 @@
"id": "09fc91b4",
"metadata": {},
"outputs": [],
"source": [
"### Edges\n\n\ndef decide_to_generate(state):\n \"\"\"\n Determines whether to generate an answer, or re-generate a question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n str: Binary decision for next node to call\n \"\"\"\n\n print(\"---ASSESS GRADED DOCUMENTS---\")\n state[\"question\"]\n filtered_documents = state[\"documents\"]\n\n if not filtered_documents:\n # All documents have been filtered check_relevance\n # We will re-generate a new query\n print(\n \"---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---\"\n )\n return \"transform_query\"\n else:\n # We have relevant documents, so generate answer\n print(\"---DECISION: GENERATE---\")\n return \"generate\"\n\n\ndef grade_generation_v_documents_and_question(state):\n \"\"\"\n Determines whether the generation is grounded in the document and answers question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n str: Decision for next node to call\n \"\"\"\n\n print(\"---CHECK HALLUCINATIONS---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n generation = state[\"generation\"]\n\n score = hallucination_grader.invoke(\n {\"documents\": documents, \"generation\": generation}\n )\n grade = score.binary_score\n\n # Check hallucination\n if grade == \"yes\":\n print(\"---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---\")\n # Check question-answering\n print(\"---GRADE GENERATION vs QUESTION---\")\n score = answer_grader.invoke({\"question\": question, \"generation\": generation})\n grade = score.binary_score\n if grade == \"yes\":\n print(\"---DECISION: GENERATION ADDRESSES QUESTION---\")\n return \"useful\"\n else:\n print(\"---DECISION: GENERATION DOES NOT ADDRESS QUESTION---\")\n return \"not useful\"\n else:\n pprint(\"---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---\")\n return \"not supported\""
]
"source": ["### Edges\n\n\ndef decide_to_generate(state):\n \"\"\"\n Determines whether to generate an answer, or re-generate a question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n str: Binary decision for next node to call\n \"\"\"\n\n print(\"---ASSESS GRADED DOCUMENTS---\")\n state[\"question\"]\n filtered_documents = state[\"documents\"]\n\n if not filtered_documents:\n # All documents have been filtered check_relevance\n # We will re-generate a new query\n print(\n \"---DECISION: ALL DOCUMENTS ARE NOT RELEVANT TO QUESTION, TRANSFORM QUERY---\"\n )\n return \"transform_query\"\n else:\n # We have relevant documents, so generate answer\n print(\"---DECISION: GENERATE---\")\n return \"generate\"\n\n\ndef grade_generation_v_documents_and_question(state):\n \"\"\"\n Determines whether the generation is grounded in the document and answers question.\n\n Args:\n state (dict): The current graph state\n\n Returns:\n str: Decision for next node to call\n \"\"\"\n\n print(\"---CHECK HALLUCINATIONS---\")\n question = state[\"question\"]\n documents = state[\"documents\"]\n generation = state[\"generation\"]\n\n score = hallucination_grader.invoke(\n {\"documents\": documents, \"generation\": generation}\n )\n grade = score.binary_score\n\n # Check hallucination\n if grade == \"yes\":\n print(\"---DECISION: GENERATION IS GROUNDED IN DOCUMENTS---\")\n # Check question-answering\n print(\"---GRADE GENERATION vs QUESTION---\")\n score = answer_grader.invoke({\"question\": question, \"generation\": generation})\n grade = score.binary_score\n if grade == \"yes\":\n print(\"---DECISION: GENERATION ADDRESSES QUESTION---\")\n return \"useful\"\n else:\n print(\"---DECISION: GENERATION DOES NOT ADDRESS QUESTION---\")\n return \"not useful\"\n else:\n pprint(\"---DECISION: GENERATION IS NOT GROUNDED IN DOCUMENTS, RE-TRY---\")\n return \"not supported\""]
},
{
"cell_type": "markdown",
@@ -425,9 +331,7 @@
]
}
],
"source": [
"from pprint import pprint\n\n# Run\ninputs = {\"question\": \"Movies that star Daniel Craig\"}\nfor output in app.stream(inputs):\n for key, value in output.items():\n # Node\n pprint(f\"Node '{key}':\")\n pprint(\"\\n---\\n\")\n\n# Final generation\npprint(value[\"generation\"])"
]
"source": ["from pprint import pprint\n\n# Run\ninputs = {\"question\": \"Movies that star Daniel Craig\"}\nfor output in app.stream(inputs):\n for key, value in output.items():\n # Node\n pprint(f\"Node '{key}':\")\n pprint(\"\\n---\\n\")\n\n# Final generation\npprint(value[\"generation\"])"]
},
{
"cell_type": "code",
@@ -435,9 +339,7 @@
"id": "4138bc51-8c84-4b8a-8d24-f7f470721f6f",
"metadata": {},
"outputs": [],
"source": [
"inputs = {\"question\": \"Which movies are about aliens?\"}\nfor output in app.stream(inputs):\n for key, value in output.items():\n # Node\n pprint(f\"Node '{key}':\")\n pprint(\"\\n---\\n\")\n\n# Final generation\npprint(value[\"generation\"])"
]
"source": ["inputs = {\"question\": \"Which movies are about aliens?\"}\nfor output in app.stream(inputs):\n for key, value in output.items():\n # Node\n pprint(f\"Node '{key}':\")\n pprint(\"\\n---\\n\")\n\n# Final generation\npprint(value[\"generation\"])"]
},
{
"cell_type": "code",
@@ -445,9 +347,7 @@
"id": "42369ab8-322d-434a-b5dd-2266e4cb2903",
"metadata": {},
"outputs": [],
"source": [
""
]
"source": [""]
}
],
"metadata": {
+4 -16
View File
@@ -13,20 +13,6 @@ By default `langgraph-checkpoint-postgres` installs `psycopg` (Psycopg 3) withou
> [!IMPORTANT]
> When manually creating Postgres connections and passing them to `PostgresSaver` or `AsyncPostgresSaver`, make sure to include `autocommit=True` and `row_factory=dict_row` (`from psycopg.rows import dict_row`). See a full example in this [how-to guide](https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/).
>
> **Why these parameters are required:**
> - `autocommit=True`: Required for the `.setup()` method to properly commit the checkpoint tables to the database. Without this, table creation may not be persisted.
> - `row_factory=dict_row`: Required because the PostgresSaver implementation accesses database rows using dictionary-style syntax (e.g., `row["column_name"]`). The default `tuple_row` factory returns tuples that only support index-based access (e.g., `row[0]`), which will cause `TypeError` exceptions when the checkpointer tries to access columns by name.
>
> **Example of incorrect usage:**
> ```python
> # ❌ This will fail with TypeError during checkpointer operations
> with psycopg.connect(DB_URI) as conn: # Missing autocommit=True and row_factory=dict_row
> checkpointer = PostgresSaver(conn)
> checkpointer.setup() # May not persist tables properly
> # Any operation that reads from database will fail with:
> # TypeError: tuple indices must be integers or slices, not str
> ```
```python
from langgraph.checkpoint.postgres import PostgresSaver
@@ -39,7 +25,7 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# call .setup() the first time you're using the checkpointer
checkpointer.setup()
checkpoint = {
"v": 4,
"v": 2,
"ts": "2024-07-31T20:14:19.804150+00:00",
"id": "1ef4f797-8335-6428-8001-8a1503f9b875",
"channel_values": {
@@ -61,6 +47,7 @@ with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
"start:node": 2
}
},
"pending_sends": [],
}
# store checkpoint
@@ -80,7 +67,7 @@ from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpoint = {
"v": 4,
"v": 2,
"ts": "2024-07-31T20:14:19.804150+00:00",
"id": "1ef4f797-8335-6428-8001-8a1503f9b875",
"channel_values": {
@@ -102,6 +89,7 @@ async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
"start:node": 2
}
},
"pending_sends": [],
}
# store checkpoint
@@ -1,10 +1,7 @@
from __future__ import annotations
import threading
from collections import defaultdict
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
from typing import Any
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from psycopg import Capabilities, Connection, Cursor, Pipeline
@@ -23,6 +20,7 @@ from langgraph.checkpoint.base import (
)
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
@@ -36,8 +34,8 @@ class PostgresSaver(BasePostgresSaver):
def __init__(
self,
conn: _internal.Conn,
pipe: Pipeline | None = None,
serde: SerializerProtocol | None = None,
pipe: Optional[Pipeline] = None,
serde: Optional[SerializerProtocol] = None,
) -> None:
super().__init__(serde=serde)
if isinstance(conn, ConnectionPool) and pipe is not None:
@@ -54,7 +52,7 @@ class PostgresSaver(BasePostgresSaver):
@contextmanager
def from_conn_string(
cls, conn_string: str, *, pipeline: bool = False
) -> Iterator[PostgresSaver]:
) -> Iterator["PostgresSaver"]:
"""Create a new PostgresSaver instance from a connection string.
Args:
@@ -101,11 +99,11 @@ class PostgresSaver(BasePostgresSaver):
def list(
self,
config: RunnableConfig | None,
config: Optional[RunnableConfig],
*,
filter: dict[str, Any] | None = None,
before: RunnableConfig | None = None,
limit: int | None = None,
filter: Optional[dict[str, Any]] = None,
before: Optional[RunnableConfig] = None,
limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
"""List checkpoints from the database.
@@ -145,39 +143,37 @@ class PostgresSaver(BasePostgresSaver):
query += f" LIMIT {limit}"
# if we change this to use .stream() we need to make sure to close the cursor
with self._cursor() as cur:
cur.execute(query, args)
values = cur.fetchall()
if not values:
return
# migrate pending sends if necessary
if to_migrate := [
v
for v in values
if v["checkpoint"]["v"] < 4 and v["parent_checkpoint_id"]
]:
cur.execute(
self.SELECT_PENDING_SENDS_SQL,
(
values[0]["thread_id"],
[v["parent_checkpoint_id"] for v in to_migrate],
cur.execute(query, args, binary=True)
for value in cur:
yield CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
self._load_checkpoint(
value["checkpoint"],
value["channel_values"],
value["pending_sends"],
),
self._load_metadata(value["metadata"]),
(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None
),
self._load_writes(value["pending_writes"]),
)
grouped_by_parent = defaultdict(list)
for value in to_migrate:
grouped_by_parent[value["parent_checkpoint_id"]].append(value)
for sends in cur:
for value in grouped_by_parent[sends["checkpoint_id"]]:
if value["channel_values"] is None:
value["channel_values"] = []
self._migrate_pending_sends(
sends["sends"],
value["checkpoint"],
value["channel_values"],
)
for value in values:
yield self._load_checkpoint_tuple(value)
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database.
This method retrieves a checkpoint tuple from the Postgres database based on the
@@ -226,27 +222,37 @@ class PostgresSaver(BasePostgresSaver):
cur.execute(
self.SELECT_SQL + where,
args,
binary=True,
)
value = cur.fetchone()
if value is None:
return None
# migrate pending sends if necessary
if value["checkpoint"]["v"] < 4 and value["parent_checkpoint_id"]:
cur.execute(
self.SELECT_PENDING_SENDS_SQL,
(thread_id, [value["parent_checkpoint_id"]]),
)
if sends := cur.fetchone():
if value["channel_values"] is None:
value["channel_values"] = []
self._migrate_pending_sends(
sends["sends"],
for value in cur:
return CheckpointTuple(
{
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": value["checkpoint_id"],
}
},
self._load_checkpoint(
value["checkpoint"],
value["channel_values"],
)
return self._load_checkpoint_tuple(value)
value["pending_sends"],
),
self._load_metadata(value["metadata"]),
(
{
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None
),
self._load_writes(value["pending_writes"]),
)
def put(
self,
@@ -313,8 +319,8 @@ class PostgresSaver(BasePostgresSaver):
checkpoint_ns,
checkpoint["id"],
checkpoint_id,
Jsonb(copy),
Jsonb(get_checkpoint_metadata(config, metadata)),
Jsonb(self._dump_checkpoint(copy)),
self._dump_metadata(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
@@ -385,7 +391,7 @@ class PostgresSaver(BasePostgresSaver):
Will be applied regardless of whether the PostgresSaver instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
"""
with self.lock, _internal.get_connection(self.conn) as conn:
with _internal.get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
@@ -401,6 +407,7 @@ class PostgresSaver(BasePostgresSaver):
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
@@ -408,52 +415,14 @@ class PostgresSaver(BasePostgresSaver):
else:
# Use connection's transaction context manager when pipeline mode not supported
with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
with conn.cursor(binary=True, row_factory=dict_row) as cur:
with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
"""
Convert a database row into a CheckpointTuple object.
Args:
value: A row from the database containing checkpoint data.
Returns:
CheckpointTuple: A structured representation of the checkpoint,
including its configuration, metadata, parent checkpoint (if any),
and pending writes.
"""
return CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
{
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
},
value["metadata"],
(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None
),
self._load_writes(value["pending_writes"]),
)
__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"]
__all__ = ["PostgresSaver", "BasePostgresSaver", "ShallowPostgresSaver", "Conn"]
@@ -1,10 +1,7 @@
from __future__ import annotations
import asyncio
from collections import defaultdict
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import asynccontextmanager
from typing import Any
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
@@ -23,6 +20,7 @@ from langgraph.checkpoint.base import (
)
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
@@ -36,8 +34,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
def __init__(
self,
conn: _ainternal.Conn,
pipe: AsyncPipeline | None = None,
serde: SerializerProtocol | None = None,
pipe: Optional[AsyncPipeline] = None,
serde: Optional[SerializerProtocol] = None,
) -> None:
super().__init__(serde=serde)
if isinstance(conn, AsyncConnectionPool) and pipe is not None:
@@ -58,8 +56,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
conn_string: str,
*,
pipeline: bool = False,
serde: SerializerProtocol | None = None,
) -> AsyncIterator[AsyncPostgresSaver]:
serde: Optional[SerializerProtocol] = None,
) -> AsyncIterator["AsyncPostgresSaver"]:
"""Create a new AsyncPostgresSaver instance from a connection string.
Args:
@@ -106,11 +104,11 @@ class AsyncPostgresSaver(BasePostgresSaver):
async def alist(
self,
config: RunnableConfig | None,
config: Optional[RunnableConfig],
*,
filter: dict[str, Any] | None = None,
before: RunnableConfig | None = None,
limit: int | None = None,
filter: Optional[dict[str, Any]] = None,
before: Optional[RunnableConfig] = None,
limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
"""List checkpoints from the database asynchronously.
@@ -133,38 +131,37 @@ class AsyncPostgresSaver(BasePostgresSaver):
# if we change this to use .stream() we need to make sure to close the cursor
async with self._cursor() as cur:
await cur.execute(query, args, binary=True)
values = await cur.fetchall()
if not values:
return
# migrate pending sends if necessary
if to_migrate := [
v
for v in values
if v["checkpoint"]["v"] < 4 and v["parent_checkpoint_id"]
]:
await cur.execute(
self.SELECT_PENDING_SENDS_SQL,
(
values[0]["thread_id"],
[v["parent_checkpoint_id"] for v in to_migrate],
async for value in cur:
yield CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
await asyncio.to_thread(
self._load_checkpoint,
value["checkpoint"],
value["channel_values"],
value["pending_sends"],
),
self._load_metadata(value["metadata"]),
(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None
),
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
grouped_by_parent = defaultdict(list)
for value in to_migrate:
grouped_by_parent[value["parent_checkpoint_id"]].append(value)
async for sends in cur:
for value in grouped_by_parent[sends["checkpoint_id"]]:
if value["channel_values"] is None:
value["channel_values"] = []
self._migrate_pending_sends(
sends["sends"],
value["checkpoint"],
value["channel_values"],
)
for value in values:
yield await self._load_checkpoint_tuple(value)
async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database asynchronously.
This method retrieves a checkpoint tuple from the Postgres database based on the
@@ -194,26 +191,36 @@ class AsyncPostgresSaver(BasePostgresSaver):
args,
binary=True,
)
value = await cur.fetchone()
if value is None:
return None
# migrate pending sends if necessary
if value["checkpoint"]["v"] < 4 and value["parent_checkpoint_id"]:
await cur.execute(
self.SELECT_PENDING_SENDS_SQL,
(thread_id, [value["parent_checkpoint_id"]]),
)
if sends := await cur.fetchone():
if value["channel_values"] is None:
value["channel_values"] = []
self._migrate_pending_sends(
sends["sends"],
async for value in cur:
return CheckpointTuple(
{
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": value["checkpoint_id"],
}
},
await asyncio.to_thread(
self._load_checkpoint,
value["checkpoint"],
value["channel_values"],
)
return await self._load_checkpoint_tuple(value)
value["pending_sends"],
),
self._load_metadata(value["metadata"]),
(
{
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None
),
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
async def aput(
self,
@@ -270,8 +277,8 @@ class AsyncPostgresSaver(BasePostgresSaver):
checkpoint_ns,
checkpoint["id"],
checkpoint_id,
Jsonb(copy),
Jsonb(get_checkpoint_metadata(config, metadata)),
Jsonb(self._dump_checkpoint(copy)),
self._dump_metadata(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
@@ -343,7 +350,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
Will be applied regardless of whether the AsyncPostgresSaver instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
"""
async with self.lock, _ainternal.get_connection(self.conn) as conn:
async with _ainternal.get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
@@ -359,6 +366,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
async with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
@@ -366,60 +374,25 @@ class AsyncPostgresSaver(BasePostgresSaver):
else:
# Use connection's transaction context manager when pipeline mode not supported
async with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
async with conn.cursor(binary=True, row_factory=dict_row) as cur:
async with (
self.lock,
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
"""
Convert a database row into a CheckpointTuple object.
Args:
value: A row from the database containing checkpoint data.
Returns:
CheckpointTuple: A structured representation of the checkpoint,
including its configuration, metadata, parent checkpoint (if any),
and pending writes.
"""
return CheckpointTuple(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
{
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
},
value["metadata"],
(
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["parent_checkpoint_id"],
}
}
if value["parent_checkpoint_id"]
else None
),
await asyncio.to_thread(self._load_writes, value["pending_writes"]),
)
def list(
self,
config: RunnableConfig | None,
config: Optional[RunnableConfig],
*,
filter: dict[str, Any] | None = None,
before: RunnableConfig | None = None,
limit: int | None = None,
filter: Optional[dict[str, Any]] = None,
before: Optional[RunnableConfig] = None,
limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
"""List checkpoints from the database.
@@ -457,7 +430,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
except StopAsyncIteration:
break
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database.
This method retrieves a checkpoint tuple from the Postgres database based on the
@@ -559,4 +532,4 @@ class AsyncPostgresSaver(BasePostgresSaver):
).result()
__all__ = ["AsyncPostgresSaver", "Conn"]
__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"]
@@ -1,5 +1,3 @@
from __future__ import annotations
import random
from collections.abc import Sequence
from typing import Any, Optional, cast
@@ -11,9 +9,12 @@ from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
ChannelVersions,
Checkpoint,
CheckpointMetadata,
get_checkpoint_id,
)
from langgraph.checkpoint.serde.types import TASKS
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from langgraph.checkpoint.serde.types import TASKS, ChannelProtocol
MetadataInput = Optional[dict[str, Any]]
@@ -71,7 +72,7 @@ MIGRATIONS = [
"""ALTER TABLE checkpoint_writes ADD COLUMN task_path TEXT NOT NULL DEFAULT '';""",
]
SELECT_SQL = """
SELECT_SQL = f"""
select
thread_id,
checkpoint,
@@ -95,20 +96,17 @@ select
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
and cw.checkpoint_id = checkpoints.checkpoint_id
) as pending_writes
) as pending_writes,
(
select array_agg(array[cw.type::bytea, cw.blob] order by cw.task_path, cw.task_id, cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
and cw.checkpoint_id = checkpoints.parent_checkpoint_id
and cw.channel = '{TASKS}'
) as pending_sends
from checkpoints """
SELECT_PENDING_SENDS_SQL = f"""
select
checkpoint_id,
array_agg(array[type::bytea, blob] order by task_path, task_id, idx) as sends
from checkpoint_writes
where thread_id = %s
and checkpoint_id = any(%s)
and channel = '{TASKS}'
group by checkpoint_id
"""
UPSERT_CHECKPOINT_BLOBS_SQL = """
INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, version, type, blob)
VALUES (%s, %s, %s, %s, %s, %s)
@@ -142,34 +140,31 @@ INSERT_CHECKPOINT_WRITES_SQL = """
class BasePostgresSaver(BaseCheckpointSaver[str]):
SELECT_SQL = SELECT_SQL
SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_SQL
MIGRATIONS = MIGRATIONS
UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
jsonplus_serde = JsonPlusSerializer()
supports_pipeline: bool
def _migrate_pending_sends(
def _load_checkpoint(
self,
pending_sends: list[tuple[bytes, bytes]],
checkpoint: dict[str, Any],
channel_values: list[tuple[bytes, bytes, bytes]],
) -> None:
if not pending_sends:
return
# add to values
enc, blob = self.serde.dumps_typed(
[self.serde.loads_typed((c.decode(), b)) for c, b in pending_sends],
)
channel_values.append((TASKS.encode(), enc.encode(), blob))
# add to versions
checkpoint["channel_versions"][TASKS] = (
max(checkpoint["channel_versions"].values())
if checkpoint["channel_versions"]
else self.get_next_version(None)
)
pending_sends: list[tuple[bytes, bytes]],
) -> Checkpoint:
return {
**checkpoint,
"pending_sends": [
self.serde.loads_typed((c.decode(), b)) for c, b in pending_sends or []
],
"channel_values": self._load_blobs(channel_values),
}
def _dump_checkpoint(self, checkpoint: Checkpoint) -> dict[str, Any]:
return {**checkpoint, "pending_sends": []}
def _load_blobs(
self, blob_values: list[tuple[bytes, bytes, bytes]]
@@ -188,7 +183,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
checkpoint_ns: str,
values: dict[str, Any],
versions: ChannelVersions,
) -> list[tuple[str, str, str, str, str, bytes | None]]:
) -> list[tuple[str, str, str, str, str, Optional[bytes]]]:
if not versions:
return []
@@ -246,7 +241,15 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
for idx, (channel, value) in enumerate(writes)
]
def get_next_version(self, current: str | None) -> str:
def _load_metadata(self, metadata: dict[str, Any]) -> CheckpointMetadata:
return self.jsonplus_serde.loads(self.jsonplus_serde.dumps(metadata))
def _dump_metadata(self, metadata: CheckpointMetadata) -> str:
serialized_metadata = self.jsonplus_serde.dumps(metadata)
# NOTE: we're using JSON serializer (not msgpack), so we need to remove null characters before writing
return serialized_metadata.decode().replace("\\u0000", "")
def get_next_version(self, current: Optional[str], channel: ChannelProtocol) -> str:
if current is None:
current_v = 0
elif isinstance(current, int):
@@ -259,9 +262,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
def _search_where(
self,
config: RunnableConfig | None,
config: Optional[RunnableConfig],
filter: MetadataInput,
before: RunnableConfig | None = None,
before: Optional[RunnableConfig] = None,
) -> tuple[str, list[Any]]:
"""Return WHERE clause predicates for alist() given config, filter, before.
@@ -0,0 +1,941 @@
import asyncio
import threading
import warnings
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import asynccontextmanager, contextmanager
from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from psycopg import (
AsyncConnection,
AsyncCursor,
AsyncPipeline,
Capabilities,
Connection,
Cursor,
Pipeline,
)
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.
The position of the migration in the list is the version number.
"""
MIGRATIONS = [
"""CREATE TABLE IF NOT EXISTS checkpoint_migrations (
v INTEGER PRIMARY KEY
);""",
"""CREATE TABLE IF NOT EXISTS checkpoints (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
type TEXT,
checkpoint JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
PRIMARY KEY (thread_id, checkpoint_ns)
);""",
"""CREATE TABLE IF NOT EXISTS checkpoint_blobs (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
channel TEXT NOT NULL,
type TEXT NOT NULL,
blob BYTEA,
PRIMARY KEY (thread_id, checkpoint_ns, channel)
);""",
"""CREATE TABLE IF NOT EXISTS checkpoint_writes (
thread_id TEXT NOT NULL,
checkpoint_ns TEXT NOT NULL DEFAULT '',
checkpoint_id TEXT NOT NULL,
task_id TEXT NOT NULL,
idx INTEGER NOT NULL,
channel TEXT NOT NULL,
type TEXT,
blob BYTEA NOT NULL,
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
);""",
"""
CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoints_thread_id_idx ON checkpoints(thread_id);
""",
"""
CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_blobs_thread_id_idx ON checkpoint_blobs(thread_id);
""",
"""
CREATE INDEX CONCURRENTLY IF NOT EXISTS checkpoint_writes_thread_id_idx ON checkpoint_writes(thread_id);
""",
"""
ALTER TABLE checkpoint_writes ADD COLUMN task_path TEXT NOT NULL DEFAULT '';
""",
]
SELECT_SQL = f"""
select
thread_id,
checkpoint,
checkpoint_ns,
metadata,
(
select array_agg(array[bl.channel::bytea, bl.type::bytea, bl.blob])
from jsonb_each_text(checkpoint -> 'channel_versions')
inner join checkpoint_blobs bl
on bl.thread_id = checkpoints.thread_id
and bl.checkpoint_ns = checkpoints.checkpoint_ns
and bl.channel = jsonb_each_text.key
) as channel_values,
(
select
array_agg(array[cw.task_id::text::bytea, cw.channel::bytea, cw.type::bytea, cw.blob] order by cw.task_id, cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
and cw.checkpoint_id = (checkpoint->>'id')
) as pending_writes,
(
select array_agg(array[cw.type::bytea, cw.blob] order by cw.task_path, cw.task_id, cw.idx)
from checkpoint_writes cw
where cw.thread_id = checkpoints.thread_id
and cw.checkpoint_ns = checkpoints.checkpoint_ns
and cw.channel = '{TASKS}'
) as pending_sends
from checkpoints """
UPSERT_CHECKPOINT_BLOBS_SQL = """
INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, type, blob)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (thread_id, checkpoint_ns, channel) DO UPDATE SET
type = EXCLUDED.type,
blob = EXCLUDED.blob;
"""
UPSERT_CHECKPOINTS_SQL = """
INSERT INTO checkpoints (thread_id, checkpoint_ns, checkpoint, metadata)
VALUES (%s, %s, %s, %s)
ON CONFLICT (thread_id, checkpoint_ns)
DO UPDATE SET
checkpoint = EXCLUDED.checkpoint,
metadata = EXCLUDED.metadata;
"""
UPSERT_CHECKPOINT_WRITES_SQL = """
INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO UPDATE SET
channel = EXCLUDED.channel,
type = EXCLUDED.type,
blob = EXCLUDED.blob;
"""
INSERT_CHECKPOINT_WRITES_SQL = """
INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, blob)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) DO NOTHING
"""
def _dump_blobs(
serde: SerializerProtocol,
thread_id: str,
checkpoint_ns: str,
values: dict[str, Any],
versions: ChannelVersions,
) -> list[tuple[str, str, str, str, Optional[bytes]]]:
if not versions:
return []
return [
(
thread_id,
checkpoint_ns,
k,
*(serde.dumps_typed(values[k]) if k in values else ("empty", None)),
)
for k in versions
]
class ShallowPostgresSaver(BasePostgresSaver):
"""A checkpoint saver that uses Postgres to store checkpoints.
This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
It is meant to be a light-weight drop-in replacement for the PostgresSaver that
supports most of the LangGraph persistence functionality with the exception of time travel.
"""
SELECT_SQL = SELECT_SQL
MIGRATIONS = MIGRATIONS
UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
lock: threading.Lock
def __init__(
self,
conn: _internal.Conn,
pipe: Optional[Pipeline] = None,
serde: Optional[SerializerProtocol] = None,
) -> None:
warnings.warn(
"ShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
"Use PostgresSaver instead, and invoke the graph with `graph.invoke(..., checkpoint_during=False)`.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(serde=serde)
if isinstance(conn, ConnectionPool) and pipe is not None:
raise ValueError(
"Pipeline should be used only with a single Connection, not ConnectionPool."
)
self.conn = conn
self.pipe = pipe
self.lock = threading.Lock()
self.supports_pipeline = Capabilities().has_pipeline()
@classmethod
@contextmanager
def from_conn_string(
cls, conn_string: str, *, pipeline: bool = False
) -> Iterator["ShallowPostgresSaver"]:
"""Create a new ShallowPostgresSaver instance from a connection string.
Args:
conn_string: The Postgres connection info string.
pipeline: whether to use Pipeline
Returns:
ShallowPostgresSaver: A new ShallowPostgresSaver instance.
"""
with Connection.connect(
conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
) as conn:
if pipeline:
with conn.pipeline() as pipe:
yield cls(conn, pipe)
else:
yield cls(conn)
def setup(self) -> None:
"""Set up the checkpoint database asynchronously.
This method creates the necessary tables in the Postgres database if they don't
already exist and runs database migrations. It MUST be called directly by the user
the first time checkpointer is used.
"""
with self._cursor() as cur:
cur.execute(self.MIGRATIONS[0])
results = cur.execute(
"SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
)
row = results.fetchone()
if row is None:
version = -1
else:
version = row["v"]
for v, migration in zip(
range(version + 1, len(self.MIGRATIONS)),
self.MIGRATIONS[version + 1 :],
):
cur.execute(migration)
cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
if self.pipe:
self.pipe.sync()
def list(
self,
config: Optional[RunnableConfig],
*,
filter: Optional[dict[str, Any]] = None,
before: Optional[RunnableConfig] = None,
limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
"""List checkpoints from the database.
This method retrieves a list of checkpoint tuples from the Postgres database based
on the provided config. For ShallowPostgresSaver, this method returns a list with
ONLY the most recent checkpoint.
"""
where, args = self._search_where(config, filter, before)
query = self.SELECT_SQL + where
if limit:
query += f" LIMIT {limit}"
with self._cursor() as cur:
cur.execute(self.SELECT_SQL + where, args, binary=True)
for value in cur:
checkpoint = self._load_checkpoint(
value["checkpoint"],
value["channel_values"],
value["pending_sends"],
)
yield CheckpointTuple(
config={
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
metadata=self._load_metadata(value["metadata"]),
pending_writes=self._load_writes(value["pending_writes"]),
)
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database.
This method retrieves a checkpoint tuple from the Postgres database based on the
provided config (matching the thread ID in the config).
Args:
config: The config to use for retrieving the checkpoint.
Returns:
Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
Examples:
Basic:
>>> config = {"configurable": {"thread_id": "1"}}
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)
With timestamp:
>>> config = {
... "configurable": {
... "thread_id": "1",
... "checkpoint_ns": "",
... "checkpoint_id": "1ef4f797-8335-6428-8001-8a1503f9b875",
... }
... }
>>> checkpoint_tuple = memory.get_tuple(config)
>>> print(checkpoint_tuple)
CheckpointTuple(...)
""" # noqa
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
args = (thread_id, checkpoint_ns)
where = "WHERE thread_id = %s AND checkpoint_ns = %s"
with self._cursor() as cur:
cur.execute(
self.SELECT_SQL + where,
args,
binary=True,
)
for value in cur:
checkpoint = self._load_checkpoint(
value["checkpoint"],
value["channel_values"],
value["pending_sends"],
)
return CheckpointTuple(
config={
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
metadata=self._load_metadata(value["metadata"]),
pending_writes=self._load_writes(value["pending_writes"]),
)
def put(
self,
config: RunnableConfig,
checkpoint: Checkpoint,
metadata: CheckpointMetadata,
new_versions: ChannelVersions,
) -> RunnableConfig:
"""Save a checkpoint to the database.
This method saves a checkpoint to the Postgres database. The checkpoint is associated
with the provided config. For ShallowPostgresSaver, this method saves ONLY the most recent
checkpoint and overwrites a previous checkpoint, if it exists.
Args:
config: The config to associate with the checkpoint.
checkpoint: The checkpoint to save.
metadata: Additional metadata to save with the checkpoint.
new_versions: New channel versions as of this write.
Returns:
RunnableConfig: Updated configuration after storing the checkpoint.
Examples:
>>> from langgraph.checkpoint.postgres import ShallowPostgresSaver
>>> DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
>>> with ShallowPostgresSaver.from_conn_string(DB_URI) as memory:
>>> config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
>>> checkpoint = {"ts": "2024-05-04T06:32:42.235444+00:00", "id": "1ef4f797-8335-6428-8001-8a1503f9b875", "channel_values": {"key": "value"}}
>>> saved_config = memory.put(config, checkpoint, {"source": "input", "step": 1, "writes": {"key": "value"}}, {})
>>> print(saved_config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1ef4f797-8335-6428-8001-8a1503f9b875'}}
"""
configurable = config["configurable"].copy()
thread_id = configurable.pop("thread_id")
checkpoint_ns = configurable.pop("checkpoint_ns")
copy = checkpoint.copy()
next_config = {
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}
}
with self._cursor(pipeline=True) as cur:
cur.execute(
"""DELETE FROM checkpoint_writes
WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
(
thread_id,
checkpoint_ns,
checkpoint["id"],
configurable.get("checkpoint_id", ""),
),
)
cur.executemany(
self.UPSERT_CHECKPOINT_BLOBS_SQL,
_dump_blobs(
self.serde,
thread_id,
checkpoint_ns,
copy.pop("channel_values"), # type: ignore[misc]
new_versions,
),
)
cur.execute(
self.UPSERT_CHECKPOINTS_SQL,
(
thread_id,
checkpoint_ns,
Jsonb(self._dump_checkpoint(copy)),
self._dump_metadata(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
def put_writes(
self,
config: RunnableConfig,
writes: Sequence[tuple[str, Any]],
task_id: str,
task_path: str = "",
) -> None:
"""Store intermediate writes linked to a checkpoint.
This method saves intermediate writes associated with a checkpoint to the Postgres database.
Args:
config: Configuration of the related checkpoint.
writes: List of writes to store.
task_id: Identifier for the task creating the writes.
"""
query = (
self.UPSERT_CHECKPOINT_WRITES_SQL
if all(w[0] in WRITES_IDX_MAP for w in writes)
else self.INSERT_CHECKPOINT_WRITES_SQL
)
with self._cursor(pipeline=True) as cur:
cur.executemany(
query,
self._dump_writes(
config["configurable"]["thread_id"],
config["configurable"]["checkpoint_ns"],
config["configurable"]["checkpoint_id"],
task_id,
task_path,
writes,
),
)
@contextmanager
def _cursor(self, *, pipeline: bool = False) -> Iterator[Cursor[DictRow]]:
"""Create a database cursor as a context manager.
Args:
pipeline: whether to use pipeline for the DB operations inside the context manager.
Will be applied regardless of whether the ShallowPostgresSaver instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
"""
with _internal.get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
# used at a time
try:
with conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
finally:
if pipeline:
self.pipe.sync()
elif pipeline:
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
# Use connection's transaction context manager when pipeline mode not supported
with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
with self.lock, conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
class AsyncShallowPostgresSaver(BasePostgresSaver):
"""A checkpoint saver that uses Postgres to store checkpoints asynchronously.
This checkpointer ONLY stores the most recent checkpoint and does NOT retain any history.
It is meant to be a light-weight drop-in replacement for the AsyncPostgresSaver that
supports most of the LangGraph persistence functionality with the exception of time travel.
"""
SELECT_SQL = SELECT_SQL
MIGRATIONS = MIGRATIONS
UPSERT_CHECKPOINT_BLOBS_SQL = UPSERT_CHECKPOINT_BLOBS_SQL
UPSERT_CHECKPOINTS_SQL = UPSERT_CHECKPOINTS_SQL
UPSERT_CHECKPOINT_WRITES_SQL = UPSERT_CHECKPOINT_WRITES_SQL
INSERT_CHECKPOINT_WRITES_SQL = INSERT_CHECKPOINT_WRITES_SQL
lock: asyncio.Lock
def __init__(
self,
conn: _ainternal.Conn,
pipe: Optional[AsyncPipeline] = None,
serde: Optional[SerializerProtocol] = None,
) -> None:
warnings.warn(
"AsyncShallowPostgresSaver is deprecated as of version 2.0.20 and will be removed in 3.0.0. "
"Use AsyncPostgresSaver instead, and invoke the graph with `await graph.ainvoke(..., checkpoint_during=False)`.",
DeprecationWarning,
stacklevel=2,
)
super().__init__(serde=serde)
if isinstance(conn, AsyncConnectionPool) and pipe is not None:
raise ValueError(
"Pipeline should be used only with a single AsyncConnection, not AsyncConnectionPool."
)
self.conn = conn
self.pipe = pipe
self.lock = asyncio.Lock()
self.loop = asyncio.get_running_loop()
self.supports_pipeline = Capabilities().has_pipeline()
@classmethod
@asynccontextmanager
async def from_conn_string(
cls,
conn_string: str,
*,
pipeline: bool = False,
serde: Optional[SerializerProtocol] = None,
) -> AsyncIterator["AsyncShallowPostgresSaver"]:
"""Create a new AsyncShallowPostgresSaver instance from a connection string.
Args:
conn_string: The Postgres connection info string.
pipeline: whether to use AsyncPipeline
Returns:
AsyncShallowPostgresSaver: A new AsyncShallowPostgresSaver instance.
"""
async with await AsyncConnection.connect(
conn_string, autocommit=True, prepare_threshold=0, row_factory=dict_row
) as conn:
if pipeline:
async with conn.pipeline() as pipe:
yield cls(conn=conn, pipe=pipe, serde=serde)
else:
yield cls(conn=conn, serde=serde)
async def setup(self) -> None:
"""Set up the checkpoint database asynchronously.
This method creates the necessary tables in the Postgres database if they don't
already exist and runs database migrations. It MUST be called directly by the user
the first time checkpointer is used.
"""
async with self._cursor() as cur:
await cur.execute(self.MIGRATIONS[0])
results = await cur.execute(
"SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
)
row = await results.fetchone()
if row is None:
version = -1
else:
version = row["v"]
for v, migration in zip(
range(version + 1, len(self.MIGRATIONS)),
self.MIGRATIONS[version + 1 :],
):
await cur.execute(migration)
await cur.execute(f"INSERT INTO checkpoint_migrations (v) VALUES ({v})")
if self.pipe:
await self.pipe.sync()
async def alist(
self,
config: Optional[RunnableConfig],
*,
filter: Optional[dict[str, Any]] = None,
before: Optional[RunnableConfig] = None,
limit: Optional[int] = None,
) -> AsyncIterator[CheckpointTuple]:
"""List checkpoints from the database asynchronously.
This method retrieves a list of checkpoint tuples from the Postgres database based
on the provided config. For ShallowPostgresSaver, this method returns a list with
ONLY the most recent checkpoint.
"""
where, args = self._search_where(config, filter, before)
query = self.SELECT_SQL + where
if limit:
query += f" LIMIT {limit}"
async with self._cursor() as cur:
await cur.execute(self.SELECT_SQL + where, args, binary=True)
async for value in cur:
checkpoint = await asyncio.to_thread(
self._load_checkpoint,
value["checkpoint"],
value["channel_values"],
value["pending_sends"],
)
yield CheckpointTuple(
config={
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
metadata=self._load_metadata(value["metadata"]),
pending_writes=await asyncio.to_thread(
self._load_writes, value["pending_writes"]
),
)
async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database asynchronously.
This method retrieves a checkpoint tuple from the Postgres database based on the
provided config (matching the thread ID in the config).
Args:
config: The config to use for retrieving the checkpoint.
Returns:
Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
"""
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
args = (thread_id, checkpoint_ns)
where = "WHERE thread_id = %s AND checkpoint_ns = %s"
async with self._cursor() as cur:
await cur.execute(
self.SELECT_SQL + where,
args,
binary=True,
)
async for value in cur:
checkpoint = await asyncio.to_thread(
self._load_checkpoint,
value["checkpoint"],
value["channel_values"],
value["pending_sends"],
)
return CheckpointTuple(
config={
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
metadata=self._load_metadata(value["metadata"]),
pending_writes=await asyncio.to_thread(
self._load_writes, value["pending_writes"]
),
)
async def aput(
self,
config: RunnableConfig,
checkpoint: Checkpoint,
metadata: CheckpointMetadata,
new_versions: ChannelVersions,
) -> RunnableConfig:
"""Save a checkpoint to the database asynchronously.
This method saves a checkpoint to the Postgres database. The checkpoint is associated
with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
checkpoint and overwrites a previous checkpoint, if it exists.
Args:
config: The config to associate with the checkpoint.
checkpoint: The checkpoint to save.
metadata: Additional metadata to save with the checkpoint.
new_versions: New channel versions as of this write.
Returns:
RunnableConfig: Updated configuration after storing the checkpoint.
"""
configurable = config["configurable"].copy()
thread_id = configurable.pop("thread_id")
checkpoint_ns = configurable.pop("checkpoint_ns")
copy = checkpoint.copy()
next_config = {
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}
}
async with self._cursor(pipeline=True) as cur:
await cur.execute(
"""DELETE FROM checkpoint_writes
WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id NOT IN (%s, %s)""",
(
thread_id,
checkpoint_ns,
checkpoint["id"],
configurable.get("checkpoint_id", ""),
),
)
await cur.executemany(
self.UPSERT_CHECKPOINT_BLOBS_SQL,
_dump_blobs(
self.serde,
thread_id,
checkpoint_ns,
copy.pop("channel_values"), # type: ignore[misc]
new_versions,
),
)
await cur.execute(
self.UPSERT_CHECKPOINTS_SQL,
(
thread_id,
checkpoint_ns,
Jsonb(self._dump_checkpoint(copy)),
self._dump_metadata(get_checkpoint_metadata(config, metadata)),
),
)
return next_config
async def aput_writes(
self,
config: RunnableConfig,
writes: Sequence[tuple[str, Any]],
task_id: str,
task_path: str = "",
) -> None:
"""Store intermediate writes linked to a checkpoint asynchronously.
This method saves intermediate writes associated with a checkpoint to the database.
Args:
config: Configuration of the related checkpoint.
writes: List of writes to store, each as (channel, value) pair.
task_id: Identifier for the task creating the writes.
"""
query = (
self.UPSERT_CHECKPOINT_WRITES_SQL
if all(w[0] in WRITES_IDX_MAP for w in writes)
else self.INSERT_CHECKPOINT_WRITES_SQL
)
params = await asyncio.to_thread(
self._dump_writes,
config["configurable"]["thread_id"],
config["configurable"]["checkpoint_ns"],
config["configurable"]["checkpoint_id"],
task_id,
task_path,
writes,
)
async with self._cursor(pipeline=True) as cur:
await cur.executemany(query, params)
@asynccontextmanager
async def _cursor(
self, *, pipeline: bool = False
) -> AsyncIterator[AsyncCursor[DictRow]]:
"""Create a database cursor as a context manager.
Args:
pipeline: whether to use pipeline for the DB operations inside the context manager.
Will be applied regardless of whether the AsyncShallowPostgresSaver instance was initialized with a pipeline.
If pipeline mode is not supported, will fall back to using transaction context manager.
"""
async with _ainternal.get_connection(self.conn) as conn:
if self.pipe:
# a connection in pipeline mode can be used concurrently
# in multiple threads/coroutines, but only one cursor can be
# used at a time
try:
async with conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
finally:
if pipeline:
await self.pipe.sync()
elif pipeline:
# a connection not in pipeline mode can only be used by one
# thread/coroutine at a time, so we acquire a lock
if self.supports_pipeline:
async with (
self.lock,
conn.pipeline(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
# Use connection's transaction context manager when pipeline mode not supported
async with (
self.lock,
conn.transaction(),
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
else:
async with (
self.lock,
conn.cursor(binary=True, row_factory=dict_row) as cur,
):
yield cur
def list(
self,
config: Optional[RunnableConfig],
*,
filter: Optional[dict[str, Any]] = None,
before: Optional[RunnableConfig] = None,
limit: Optional[int] = None,
) -> Iterator[CheckpointTuple]:
"""List checkpoints from the database.
This method retrieves a list of checkpoint tuples from the Postgres database based
on the provided config. For ShallowPostgresSaver, this method returns a list with
ONLY the most recent checkpoint.
"""
aiter_ = self.alist(config, filter=filter, before=before, limit=limit)
while True:
try:
yield asyncio.run_coroutine_threadsafe(
anext(aiter_), # noqa: F821
self.loop,
).result()
except StopAsyncIteration:
break
def get_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
"""Get a checkpoint tuple from the database.
This method retrieves a checkpoint tuple from the Postgres database based on the
provided config (matching the thread ID in the config).
Args:
config: The config to use for retrieving the checkpoint.
Returns:
Optional[CheckpointTuple]: The retrieved checkpoint tuple, or None if no matching checkpoint was found.
"""
try:
# check if we are in the main thread, only bg threads can block
# we don't check in other methods to avoid the overhead
if asyncio.get_running_loop() is self.loop:
raise asyncio.InvalidStateError(
"Synchronous calls to AsyncShallowPostgresSaver are only allowed from a "
"different thread. From the main thread, use the async interface."
"For example, use `await checkpointer.aget_tuple(...)` or `await "
"graph.ainvoke(...)`."
)
except RuntimeError:
pass
return asyncio.run_coroutine_threadsafe(
self.aget_tuple(config), self.loop
).result()
def put(
self,
config: RunnableConfig,
checkpoint: Checkpoint,
metadata: CheckpointMetadata,
new_versions: ChannelVersions,
) -> RunnableConfig:
"""Save a checkpoint to the database.
This method saves a checkpoint to the Postgres database. The checkpoint is associated
with the provided config. For AsyncShallowPostgresSaver, this method saves ONLY the most recent
checkpoint and overwrites a previous checkpoint, if it exists.
Args:
config: The config to associate with the checkpoint.
checkpoint: The checkpoint to save.
metadata: Additional metadata to save with the checkpoint.
new_versions: New channel versions as of this write.
Returns:
RunnableConfig: Updated configuration after storing the checkpoint.
"""
return asyncio.run_coroutine_threadsafe(
self.aput(config, checkpoint, metadata, new_versions), self.loop
).result()
def put_writes(
self,
config: RunnableConfig,
writes: Sequence[tuple[str, Any]],
task_id: str,
task_path: str = "",
) -> None:
"""Store intermediate writes linked to a checkpoint.
This method saves intermediate writes associated with a checkpoint to the database.
Args:
config: Configuration of the related checkpoint.
writes: List of writes to store, each as (channel, value) pair.
task_id: Identifier for the task creating the writes.
task_path: Path of the task creating the writes.
"""
return asyncio.run_coroutine_threadsafe(
self.aput_writes(config, writes, task_id, task_path), self.loop
).result()
@@ -1,11 +1,9 @@
from __future__ import annotations
import asyncio
import logging
from collections.abc import AsyncIterator, Iterable, Sequence
from contextlib import asynccontextmanager
from types import TracebackType
from typing import Any, Callable, cast
from typing import Any, Callable, Optional, Union, cast
import orjson
from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
@@ -134,10 +132,12 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
self,
conn: _ainternal.Conn,
*,
pipe: AsyncPipeline | None = None,
deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
index: PostgresIndexConfig | None = None,
ttl: TTLConfig | None = None,
pipe: Optional[AsyncPipeline] = None,
deserializer: Optional[
Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]
] = None,
index: Optional[PostgresIndexConfig] = None,
ttl: Optional[TTLConfig] = None,
) -> None:
if isinstance(conn, AsyncConnectionPool) and pipe is not None:
raise ValueError(
@@ -157,7 +157,7 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
self.embeddings = None
self.ttl_config = ttl
self._ttl_sweeper_task: asyncio.Task[None] | None = None
self._ttl_sweeper_task: Optional[asyncio.Task[None]] = None
self._ttl_stop_event = asyncio.Event()
async def abatch(self, ops: Iterable[Op]) -> list[Result]:
@@ -180,10 +180,10 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
conn_string: str,
*,
pipeline: bool = False,
pool_config: PoolConfig | None = None,
index: PostgresIndexConfig | None = None,
ttl: TTLConfig | None = None,
) -> AsyncIterator[AsyncPostgresStore]:
pool_config: Optional[PoolConfig] = None,
index: Optional[PostgresIndexConfig] = None,
ttl: Optional[TTLConfig] = None,
) -> AsyncIterator["AsyncPostgresStore"]:
"""Create a new AsyncPostgresStore instance from a connection string.
Args:
@@ -289,7 +289,7 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
return deleted_count
async def start_ttl_sweeper(
self, sweep_interval_minutes: int | None = None
self, sweep_interval_minutes: Optional[int] = None
) -> asyncio.Task[None]:
"""Periodically delete expired store items based on TTL.
@@ -334,7 +334,7 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
self._ttl_sweeper_task = task
return task
async def stop_ttl_sweeper(self, timeout: float | None = None) -> bool:
async def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
"""Stop the TTL sweeper task if it's running.
Args:
@@ -369,14 +369,14 @@ class AsyncPostgresStore(AsyncBatchedBaseStore, BasePostgresStore[_ainternal.Con
return success
async def __aenter__(self) -> AsyncPostgresStore:
async def __aenter__(self) -> "AsyncPostgresStore":
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
exc_type: Optional[type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional["TracebackType"],
) -> None:
# Ensure the TTL sweeper task is stopped when exiting the context
if hasattr(self, "_ttl_sweeper_task") and self._ttl_sweeper_task is not None:
@@ -1,5 +1,3 @@
from __future__ import annotations
import asyncio
import concurrent.futures
import json
@@ -16,6 +14,7 @@ from typing import (
Generic,
Literal,
NamedTuple,
Optional,
TypeVar,
Union,
cast,
@@ -57,8 +56,8 @@ class Migration(NamedTuple):
"""A database migration with optional conditions and parameters."""
sql: str
params: dict[str, Any] | None = None
condition: Callable[[BasePostgresStore], bool] | None = None
params: Optional[dict[str, Any]] = None
condition: Optional[Callable[["BasePostgresStore"], bool]] = None
MIGRATIONS: Sequence[str] = [
@@ -156,7 +155,7 @@ class PoolConfig(TypedDict, total=False):
min_size: int
"""Minimum number of connections maintained in the pool. Defaults to 1."""
max_size: int | None
max_size: Optional[int]
"""Maximum number of connections allowed in the pool. None means unlimited."""
kwargs: dict
@@ -231,8 +230,8 @@ class BasePostgresStore(Generic[C]):
MIGRATIONS = MIGRATIONS
VECTOR_MIGRATIONS = VECTOR_MIGRATIONS
conn: C
_deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None
index_config: PostgresIndexConfig | None
_deserializer: Optional[Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]]
index_config: Optional[PostgresIndexConfig]
def _get_batch_GET_ops_queries(
self,
@@ -294,7 +293,7 @@ class BasePostgresStore(Generic[C]):
put_ops: Sequence[tuple[int, PutOp]],
) -> tuple[
list[tuple[str, Sequence]],
tuple[str, Sequence[tuple[str, str, str, str]]] | None,
Optional[tuple[str, Sequence[tuple[str, str, str, str]]]],
]:
dedupped_ops: dict[tuple[tuple[str, ...], str], PutOp] = {}
for _, op in put_ops:
@@ -321,7 +320,9 @@ class BasePostgresStore(Generic[C]):
)
params = (_namespace_to_text(namespace), *keys)
queries.append((query, params))
embedding_request: tuple[str, Sequence[tuple[str, str, str, str]]] | None = None
embedding_request: Optional[tuple[str, Sequence[tuple[str, str, str, str]]]] = (
None
)
if inserts:
values = []
insertion_params = []
@@ -402,7 +403,7 @@ class BasePostgresStore(Generic[C]):
self,
search_ops: Sequence[tuple[int, SearchOp]],
) -> tuple[
list[tuple[str, list[None | str | list[float]]]], # queries, params
list[tuple[str, list[Union[None, str, list[float]]]]], # queries, params
list[tuple[int, str]], # idx, query_text pairs to embed
]:
"""
@@ -431,7 +432,7 @@ class BasePostgresStore(Generic[C]):
filter_params.extend([key, orjson.dumps(value).decode("utf-8")])
ns_condition = "TRUE"
ns_param: Sequence[str] | None = None
ns_param: Optional[Sequence[Union[str]]] = None
if op.namespace_prefix:
ns_condition = "store.prefix LIKE %s"
ns_param = (f"{_namespace_to_text(op.namespace_prefix)}%",)
@@ -718,10 +719,12 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
self,
conn: _pg_internal.Conn,
*,
pipe: Pipeline | None = None,
deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
index: PostgresIndexConfig | None = None,
ttl: TTLConfig | None = None,
pipe: Optional[Pipeline] = None,
deserializer: Optional[
Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]
] = None,
index: Optional[PostgresIndexConfig] = None,
ttl: Optional[TTLConfig] = None,
) -> None:
super().__init__()
self._deserializer = deserializer
@@ -735,7 +738,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
else:
self.embeddings = None
self.ttl_config = ttl
self._ttl_sweeper_thread: threading.Thread | None = None
self._ttl_sweeper_thread: Optional[threading.Thread] = None
self._ttl_stop_event = threading.Event()
@classmethod
@@ -745,10 +748,10 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
conn_string: str,
*,
pipeline: bool = False,
pool_config: PoolConfig | None = None,
index: PostgresIndexConfig | None = None,
ttl: TTLConfig | None = None,
) -> Iterator[PostgresStore]:
pool_config: Optional[PoolConfig] = None,
index: Optional[PostgresIndexConfig] = None,
ttl: Optional[TTLConfig] = None,
) -> Iterator["PostgresStore"]:
"""Create a new PostgresStore instance from a connection string.
Args:
@@ -807,7 +810,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
return deleted_count
def start_ttl_sweeper(
self, sweep_interval_minutes: int | None = None
self, sweep_interval_minutes: Optional[int] = None
) -> concurrent.futures.Future[None]:
"""Periodically delete expired store items based on TTL.
@@ -864,7 +867,7 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
)
return future
def stop_ttl_sweeper(self, timeout: float | None = None) -> bool:
def stop_ttl_sweeper(self, timeout: Optional[float] = None) -> bool:
"""Stop the TTL sweeper thread if it's running.
Args:
@@ -1193,7 +1196,7 @@ def _row_to_item(
namespace: tuple[str, ...],
row: Row,
*,
loader: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
loader: Optional[Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]] = None,
) -> Item:
"""Convert a row from the database into an Item.
@@ -1221,7 +1224,7 @@ def _row_to_search_item(
namespace: tuple[str, ...],
row: Row,
*,
loader: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
loader: Optional[Callable[[Union[bytes, orjson.Fragment]], dict[str, Any]]] = None,
) -> SearchItem:
"""Convert a row from the database into an Item."""
loader = loader or _json_loads
@@ -1252,7 +1255,7 @@ def _group_ops(ops: Iterable[Op]) -> tuple[dict[type, list[tuple[int, Op]]], int
return grouped_ops, tot
def _json_loads(content: bytes | orjson.Fragment) -> Any:
def _json_loads(content: Union[bytes, orjson.Fragment]) -> Any:
if isinstance(content, orjson.Fragment):
if hasattr(content, "buf"):
content = content.buf
@@ -1264,7 +1267,7 @@ def _json_loads(content: bytes | orjson.Fragment) -> Any:
return orjson.loads(cast(bytes, content))
def _decode_ns_bytes(namespace: str | bytes | list) -> tuple[str, ...]:
def _decode_ns_bytes(namespace: Union[str, bytes, list]) -> tuple[str, ...]:
if isinstance(namespace, list):
return tuple(namespace)
if isinstance(namespace, bytes):
@@ -1313,16 +1316,16 @@ def get_distance_operator(store: Any) -> tuple[str, str]:
def _ensure_index_config(
index_config: PostgresIndexConfig,
) -> tuple[Embeddings | None, PostgresIndexConfig]:
) -> tuple[Optional["Embeddings"], PostgresIndexConfig]:
index_config = index_config.copy()
tokenized: list[tuple[str, Literal["$"] | list[str]]] = []
tokenized: list[tuple[str, Union[Literal["$"], list[str]]]] = []
tot = 0
fields = index_config.get("fields") or ["$"]
if isinstance(fields, str):
fields = [fields]
if not isinstance(fields, list):
raise ValueError(f"Text fields must be a list or a string. Got {fields}")
for p in fields:
text_fields = index_config.get("fields") or ["$"]
if isinstance(text_fields, str):
text_fields = [text_fields]
if not isinstance(text_fields, list):
raise ValueError(f"Text fields must be a list or a string. Got {text_fields}")
for p in text_fields:
if p == "$":
tokenized.append((p, "$"))
tot += 1
+1 -1
View File
@@ -56,7 +56,7 @@ lint.select = [
"B", # flake8-bugbear
"I", # isort
]
lint.ignore = ["E501", "B008"]
lint.ignore = ["E501", "B008", "UP007", "UP006"]
[tool.mypy]
# https://mypy.readthedocs.io/en/stable/config_file.html
@@ -1,53 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime, timezone
from typing import Any, Protocol
from langgraph.checkpoint.base import Checkpoint, EmptyChannelError
from langgraph.checkpoint.base.id import uuid6
class ChannelProtocol(Protocol):
def checkpoint(self) -> Any | None: ...
def empty_checkpoint() -> Checkpoint:
return Checkpoint(
v=1,
id=str(uuid6(clock_seq=-2)),
ts=datetime.now(timezone.utc).isoformat(),
channel_values={},
channel_versions={},
versions_seen={},
)
def create_checkpoint(
checkpoint: Checkpoint,
channels: Mapping[str, ChannelProtocol] | None,
step: int,
*,
id: str | None = None,
) -> Checkpoint:
"""Create a checkpoint for the given channels."""
ts = datetime.now(timezone.utc).isoformat()
if channels is None:
values = checkpoint["channel_values"]
else:
values = {}
for k, v in channels.items():
if k not in checkpoint["channel_versions"]:
continue
try:
values[k] = v.checkpoint()
except EmptyChannelError:
pass
return Checkpoint(
v=1,
ts=ts,
id=id or str(uuid6(clock_seq=step)),
channel_values=values,
channel_versions=checkpoint["channel_versions"],
versions_seen=checkpoint["versions_seen"],
)
+39 -55
View File
@@ -14,10 +14,13 @@ from langgraph.checkpoint.base import (
EXCLUDED_METADATA_KEYS,
Checkpoint,
CheckpointMetadata,
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.postgres.aio import (
AsyncPostgresSaver,
AsyncShallowPostgresSaver,
)
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
from langgraph.checkpoint.serde.types import TASKS
from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
from tests.conftest import DEFAULT_POSTGRES_URI
@@ -108,11 +111,41 @@ async def _base_saver():
await conn.execute(f"DROP DATABASE {database}")
@asynccontextmanager
async def _shallow_saver():
"""Fixture for shallow connection mode testing."""
database = f"test_{uuid4().hex[:16]}"
# create unique db
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI, autocommit=True
) as conn:
await conn.execute(f"CREATE DATABASE {database}")
try:
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI + database,
autocommit=True,
prepare_threshold=0,
row_factory=dict_row,
) as conn:
checkpointer = AsyncShallowPostgresSaver(conn)
await checkpointer.setup()
yield checkpointer
finally:
# drop unique db
async with await AsyncConnection.connect(
DEFAULT_POSTGRES_URI, autocommit=True
) as conn:
await conn.execute(f"DROP DATABASE {database}")
@asynccontextmanager
async def _saver(name: str):
if name == "base":
async with _base_saver() as saver:
yield saver
elif name == "shallow":
async with _shallow_saver() as saver:
yield saver
elif name == "pool":
async with _pool_saver() as saver:
yield saver
@@ -172,7 +205,7 @@ def test_data():
}
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
async def test_combined_metadata(saver_name: str, test_data) -> None:
async with _saver(saver_name) as saver:
config = {
@@ -199,7 +232,7 @@ async def test_combined_metadata(saver_name: str, test_data) -> None:
}
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
async def test_asearch(saver_name: str, test_data) -> None:
async with _saver(saver_name) as saver:
configs = test_data["configs"]
@@ -250,7 +283,7 @@ async def test_asearch(saver_name: str, test_data) -> None:
} == {"", "inner"}
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
async def test_null_chars(saver_name: str, test_data) -> None:
async with _saver(saver_name) as saver:
config = await saver.aput(
@@ -263,52 +296,3 @@ async def test_null_chars(saver_name: str, test_data) -> None:
assert [c async for c in saver.alist(None, filter={"my_key": "abc"})][
0
].metadata["my_key"] == "abc"
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
async def test_pending_sends_migration(saver_name: str) -> None:
async with _saver(saver_name) as saver:
config = {
"configurable": {
"thread_id": "thread-1",
"checkpoint_ns": "",
}
}
# create the first checkpoint
# and put some pending sends
checkpoint_0 = empty_checkpoint()
config = await saver.aput(config, checkpoint_0, {}, {})
await saver.aput_writes(
config, [(TASKS, "send-1"), (TASKS, "send-2")], task_id="task-1"
)
await saver.aput_writes(config, [(TASKS, "send-3")], task_id="task-2")
# check that fetching checkpoint_0 doesn't attach pending sends
# (they should be attached to the next checkpoint)
tuple_0 = await saver.aget_tuple(config)
assert tuple_0.checkpoint["channel_values"] == {}
assert tuple_0.checkpoint["channel_versions"] == {}
# create the second checkpoint
checkpoint_1 = create_checkpoint(checkpoint_0, {}, 1)
config = await saver.aput(config, checkpoint_1, {}, {})
# check that pending sends are attached to checkpoint_1
tuple_1 = await saver.aget_tuple(config)
assert tuple_1.checkpoint["channel_values"] == {
TASKS: ["send-1", "send-2", "send-3"]
}
assert TASKS in tuple_1.checkpoint["channel_versions"]
# check that list also applies the migration
search_results = [
c async for c in saver.alist({"configurable": {"thread_id": "thread-1"}})
]
assert len(search_results) == 2
assert search_results[-1].checkpoint["channel_values"] == {}
assert search_results[-1].checkpoint["channel_versions"] == {}
assert search_results[0].checkpoint["channel_values"] == {
TASKS: ["send-1", "send-2", "send-3"]
}
assert TASKS in search_results[0].checkpoint["channel_versions"]

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