Compare commits

..
Author SHA1 Message Date
Sydney Runkle c8a9aec9d8 fix signature 2025-06-24 09:08:08 -04:00
Sydney Runkle e05cef7732 support invokable signature 2025-06-24 08:31:58 -04:00
Vedant PanchalandGitHub ac328c3fd8 [fix] snippet for Agents as a Tool (#5181)
snippet had a missing decorator that might lead the user to confuse whether it is really a tool
2025-06-24 11:44:32 +00:00
Nuno CamposandGitHub ebd7977936 Revert change to default value of checkpoint_during arg (#5177) 2025-06-23 18:12:19 -07:00
Nuno Campos a6381c32b0 Revert change to default value of checkpoint_during arg 2025-06-23 18:05:46 -07:00
Nuno CamposandGitHub 294d346650 Fix bug where Command(update=) could be ignored if there was a 2nd interrupt after it (#5175) 2025-06-23 17:35:36 -07:00
Nuno Campos 866c8009dc Fix bug where Command(update=) could be ignored if there was a 2nd interrupt after it
- writes from the null task (ie. from outside tasks) should be accummulated across invocations
2025-06-23 17:23:53 -07:00
Rauf ParchievandGitHub 73bed2cf7c Update workflows.md (#5116) 2025-06-23 23:54:45 +00:00
Eugene YurtsevandGitHub 4cffe58065 docs: fix tab syntax errors and admonition syntax errors (#5091) 2025-06-23 19:53:54 -04:00
nikhildigdeandGitHub e73964a971 Update custom_routes.md (#5110)
Updated the statement about app.py
2025-06-23 23:53:25 +00:00
William Fu-Hinthorn de91f21f6b Update cli config doc on pip_installer 2025-06-23 16:40:07 -07:00
David DuongandGitHub 41eed326b8 Disable values update on use stream in interrupt events (#5041) 2025-06-24 00:08:14 +02:00
Mason DaughertyandGitHub 946d23213d docs: add Homebrew install option to CLI docs (#5160)
Add Homebrew install option to CLI docs
2025-06-23 11:33:12 -04:00
hari-dhanushkodiandGitHub c78197a583 add more docs for lgp deployment metrics (#5151) 2025-06-23 07:12:06 -07:00
Lauren Hirata SinghandGitHub 50756207ee docs: fix a typo 'prebuit' to 'prebuilt' in notebook_hooks.py (#5154) 2025-06-23 09:44:45 -04:00
OfirTeneJunoandGitHub 903cec0cfa Merge branch 'main' into interrupt-use-stream-values 2025-06-23 13:25:35 +03:00
foie0222 f63952595d fix: a typo prebuit to prebuilt 2025-06-21 17:49:40 +09:00
Andrew NguonlyandGitHub 4a252bd03a docs: Remove note for LANGSMITH_TRACING environment variable (#5147)
Remove note for LANGSMITH_TRACING.
2025-06-20 13:36:29 -07:00
Eugene YurtsevandGitHub f0f329d9e1 docs: Add conditional js/python rendering (#5128)
# Overview

Adding conditional rendering logic to co-locate js and python documentation.

* `:::` conditional syntax can be used to switch between python only or js only content.
* Contains simple unit tests for `:::`
* PR adds set up for a way to implement a context switch between languages, but it will not be enabled until JS content is merged in.
* Contains a script that can add javascript documentation 

Implementation of: https://github.com/langchain-ai/langgraph/pull/5118

## Example

Example of conditional rendering / compilation.


```markdown

### Config (static context)

Config is for immutable data like user metadata or API keys. Use
when you have values that don't change mid-run.

Specify configuration using a key called **"configurable"** which is reserved
for this purpose:


:::python

This content will only be rendered for the python site.
:::

:::js
this content will only be rendered for the js / ts site.
:::

```
2025-06-20 16:20:16 -04:00
OfirTeneJunoandGitHub 77306c5142 Merge branch 'main' into interrupt-use-stream-values 2025-06-19 14:06:24 +03:00
Nuno CamposandGitHub eba18c3213 Reduce extraneous keys in checkpoint.metadata (#5133) 2025-06-17 17:46:45 -07:00
Nuno Campos a1c856c088 Reduce extraneous keys in checkpoint.metadata
- Leave it up to each checkpointer implementation to decide whether to merge in configurable/metadata (previously PregelLoop would do some of this always)
- Never copy over internal langgraph keys into checkpoint.metadata (these are redundant/misleading to include)
2025-06-17 17:40:18 -07:00
Josh RogersandGitHub 92010f84ec Adding disable_mcp field to cli docs (#5132) 2025-06-17 16:22:50 -07:00
Nuno CamposandGitHub 3b8f3f9de3 If FuturesDict callback has been GCed, don't call it (#5131) 2025-06-17 14:03:16 -07:00
Nuno Campos 6dcff8a839 If FuturesDict callback has been GCed, don't call it 2025-06-17 13:47:47 -07:00
1309243b29 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>
2025-06-17 12:32:04 -07:00
Nuno Campos 771c6150a4 langgraph 0.5.0rc1 2025-06-16 17:52:13 -07:00
Nuno Campos edfb65fd3a langgraph-prebuilt 0.5.0rc0 2025-06-16 17:47:21 -07:00
Lauren Hirata SinghandGitHub 0f92470e49 docs: Remove cookie consent (#5123) 2025-06-16 18:41:55 -04:00
Nuno Campos dfcaf97c73 langgraph 0.5.0rc0 2025-06-16 15:17:56 -07:00
Nuno Campos 63a0028372 langgraph-checkpoint 2.1.0 2025-06-16 14:58:50 -07:00
Nuno CamposandGitHub 1134017d07 Preparation for 0.5 release: langgraph-checkpoint (#5124)
Prepare langgraph-checkpoint for 0.5

- Given we have no upper bound on langgraph-checkpoint dep need to undo all changes in langgraph-checkpoint that might break previous versions of langgraph
2025-06-16 21:57:11 +00:00
Lauren Hirata Singh 33feba4877 Remove cookie consent 2025-06-16 16:33:32 -04:00
Nuno CamposandGitHub 4fec8e9dec Preparation for 0.5 release (#5121) 2025-06-16 13:14:25 -07:00
Nuno Campos c137169325 Preparation for 0.5 release
- Update deprecation warnings to mention 0.5, no 1.0
- Add back type hint support for Runnable arg to add_node
2025-06-16 13:07:52 -07:00
Nuno CamposandGitHub 1e2672e63d Restore shallow checkpointer (#5105) 2025-06-16 11:23:27 -07:00
Nuno CamposandGitHub 06803ab683 Add migration for pending_sends (#5106) 2025-06-16 11:23:17 -07:00
hari-dhanushkodiandGitHub 3488ee47e0 chore: add docs for lgp deployment monitoring (#5104) 2025-06-16 10:21:42 -07:00
Nuno CamposandGitHub 289bdd0cea Introduce "tasks" and "checkpoints" stream modes (#5117) 2025-06-16 10:14:18 -07:00
Nuno Campos 417103066b Lint 2025-06-16 09:29:03 -07:00
Nuno Campos 25a59447c1 Introduce "tasks" and "checkpoints" stream modes
- These are split out of "debug" stream mode, which is now an alias for ["tasks", "checkpoints"]
2025-06-16 08:47:45 -07:00
Nuno Campos 21906d2b7b Add migration for pending_sends
- Checkpoints saved on older versions of langgraph will be compatible with langgraph 0.5 and 1.0
2025-06-13 17:42:14 -07:00
Nuno Campos 0cad7019cb Restore shallow checkpointer
- This should definitely be removed soon, but let's give people more time to update
2025-06-13 17:37:40 -07:00
Nuno CamposandGitHub 7e735672bf Restore compatibility with custom checkpointer classes created in prior versions (#5103) 2025-06-13 16:36:36 -07:00
Nuno Campos 5498893780 Restore compatibility with custom checkpointer classes created in prior versions
- Ensure existing custom checkpointer classes are compatible with new langgraph-checkpoint release
2025-06-13 16:29:55 -07:00
Nuno CamposandGitHub e80f47aa01 Revert removals of APIs that were slated for removal in 1.0 (#5101) 2025-06-13 16:09:20 -07:00
Nuno Campos a0b2f742a3 Revert "Remove UntrackedValue channel"
This reverts commit 05f3904d09.
2025-06-13 15:36:47 -07:00
William FHandGitHub b7973d65db fix: Update lockfile (#5102) 2025-06-13 14:53:40 -07:00
Nuno Campos 3fa3a586b5 Revert "Remove MessageGraph (#4875)"
This reverts commit a5e6223569.
2025-06-13 14:21:05 -07:00
William FHandGitHub 053b606b46 cli: 0.3.3 (#5100) 2025-06-13 13:15:26 -07:00
OfirTeneJunoandGitHub 666279a241 Merge branch 'main' into interrupt-use-stream-values 2025-06-11 09:53:25 +03:00
OfirTeneJuno 2172bc89ed remove yarn 2025-06-10 17:37:27 +03:00
OfirTeneJuno 4138ef9c43 Change to continue 2025-06-10 17:33:52 +03:00
OfirTeneJuno 0ff181b7ce Disable values update on use stream in interrupt events 2025-06-10 17:21:17 +03:00
89 changed files with 4825 additions and 3392 deletions
+6 -12
View File
@@ -15,8 +15,8 @@ from mkdocs.structure.files import Files, File
from mkdocs.structure.pages import Page
from _scripts.generate_api_reference_links import update_markdown_with_imports
from _scripts.link_map import JS_LINK_MAP
from _scripts.notebook_convert import convert_notebook
from _scripts.link_map import JS_LINK_MAP
logger = logging.getLogger(__name__)
logging.basicConfig()
@@ -87,7 +87,7 @@ REDIRECT_MAP = {
"cloud/how-tos/stream_events.md": "cloud/how-tos/streaming.md#stream-events",
"cloud/how-tos/stream_debug.md": "cloud/how-tos/streaming.md#debug",
"cloud/how-tos/stream_multiple.md": "cloud/how-tos/streaming.md#stream-multiple-modes",
# prebuit redirects
# prebuilt redirects
"how-tos/create-react-agent.ipynb": "agents/agents.md#basic-configuration",
"how-tos/create-react-agent-memory.ipynb": "agents/memory.md",
"how-tos/create-react-agent-system-prompt.ipynb": "agents/context.md#prompts",
@@ -187,7 +187,7 @@ def _resolve_cross_references(md_text: str, link_map: dict[str, str]) -> str:
def _apply_conditional_rendering(md_text: str, target_language: str) -> str:
if target_language not in {"python", "js", "switcher"}:
if target_language not in {"python", "js"}:
raise ValueError("target_language must be 'python' or 'js'")
pattern = re.compile(
@@ -201,16 +201,10 @@ def _apply_conditional_rendering(md_text: str, target_language: str) -> str:
language = match.group("language")
content = match.group("content")
if language not in {"python", "js", "switcher"}:
if language not in {"python", "js"}:
# If the language is not supported, return the original block
return match.group(0)
if target_language == "switcher":
# Both Python and JavaScript blocks are wrapped in a tag that
# allows the user to switch between them.
standardized_language = "javascript" if language == "js" else "python"
return f'<div class="lang-{standardized_language}">\n' + content + "\n</div>"
if language == target_language:
return content
@@ -321,8 +315,8 @@ def _on_page_markdown_with_config(
markdown = _highlight_code_blocks(markdown)
# Apply conditional rendering for code blocks
target_language = kwargs.get("target_language", "js")
markdown = _apply_conditional_rendering(markdown, "switcher")
target_language = kwargs.get("target_language", "python")
markdown = _apply_conditional_rendering(markdown, target_language)
if target_language == "js":
markdown = _resolve_cross_references(markdown, JS_LINK_MAP)
elif target_language == "python":
+1 -1
View File
@@ -233,4 +233,4 @@ Tools can access context through special parameter **annotations**.
### Update Context from Tools
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.
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.
+1 -1
View File
@@ -106,4 +106,4 @@ if __name__ == "__main__":
## Additional resources
- [MCP documentation](https://modelcontextprotocol.io/introduction)
- [MCP Transport documentation](https://modelcontextprotocol.io/docs/concepts/transports)
- [MCP Transport documentation](https://modelcontextprotocol.io/docs/concepts/transports)
+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.
+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)
@@ -212,6 +212,7 @@ We have now created an assistant called "Open AI Assistant" that has `model_name
Output:
```
Receiving event of type: metadata
{'run_id': '1ef6746e-5893-67b1-978a-0f1cd4060e16'}
@@ -219,6 +220,7 @@ Output:
Receiving event of type: updates
{'agent': {'messages': [{'content': 'I was created by OpenAI, a research organization focused on developing and advancing artificial intelligence technology.', 'additional_kwargs': {}, 'response_metadata': {'finish_reason': 'stop', 'model_name': 'gpt-4o-2024-05-13', 'system_fingerprint': 'fp_157b3831f5'}, 'type': 'ai', 'name': None, 'id': 'run-e1a6b25c-8416-41f2-9981-f9cfe043f414', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}}
```
### LangGraph Platform UI
@@ -231,9 +233,11 @@ Inside your deployment, select the "Assistants" tab. For the assistant you would
To edit the assistant, use the `update` method. This will create a new version of the assistant with the provided edits. See the [Python](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/python_sdk_ref/#langgraph_sdk.client.AssistantsClient.update) and [JS](https://langchain-ai.github.io/langgraph/cloud/reference/sdk/js_ts_sdk_ref/#update) SDK reference docs for more information.
!!! note "Note"
You must pass in the ENTIRE config (and metadata if you are using it). The update endpoint creates new versions completely from scratch and does not rely on previous versions.
You must pass in the ENTIRE config (and metadata if you are using it). The update endpoint creates new versions completely from scratch and does not rely on previous versions.
For example, to update your assistant's system prompt:
=== "Python"
```python
@@ -247,5 +247,7 @@ Verify that the original, interrupted run was interrupted
Output:
```
'interrupted'
```
+14 -7
View File
@@ -73,9 +73,11 @@ langgraph dev --debug-port 5678
Then attach your preferred debugger:
=== "VS Code"
Add this configuration to `launch.json`:
`json
{
Add this configuration to `launch.json`:
```json
{
"name": "Attach to LangGraph",
"type": "debugpy",
"request": "attach",
@@ -83,11 +85,16 @@ Add this configuration to `launch.json`:
"host": "0.0.0.0",
"port": 5678
}
}
`
Specify the port number you chose in the previous step.
}
```
=== "PyCharm" 1. Go to Run → Edit Configurations 2. Click + and select "Python Debug Server" 3. Set IDE host name: `localhost` 4. Set port: `5678` (or the port number you chose in the previous step) 5. Click "OK" and start debugging
=== "PyCharm"
1. Go to Run → Edit Configurations
2. Click + and select "Python Debug Server"
3. Set IDE host name: `localhost`
4. Set port: `5678` (or the port number you chose in the previous step)
5. Click "OK" and start debugging
## Troubleshooting
@@ -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.
+74 -68
View File
@@ -8,15 +8,15 @@ Currently, the SDK does not provide built-in support for defining webhook endpoi
The following API endpoints accept a `webhook` parameter:
| Operation | HTTP Method | Endpoint |
|-----------|------------|----------|
| Create Run | `POST` | `/thread/{thread_id}/runs` |
| Create Thread Cron | `POST` | `/thread/{thread_id}/runs/crons` |
| Stream Run | `POST` | `/thread/{thread_id}/runs/stream` |
| Wait Run | `POST` | `/thread/{thread_id}/runs/wait` |
| Create Cron | `POST` | `/runs/crons` |
| Stream Run Stateless | `POST` | `/runs/stream` |
| Wait Run Stateless | `POST` | `/runs/wait` |
| Operation | HTTP Method | Endpoint |
|----------------------|-------------|-----------------------------------|
| Create Run | `POST` | `/thread/{thread_id}/runs` |
| Create Thread Cron | `POST` | `/thread/{thread_id}/runs/crons` |
| Stream Run | `POST` | `/thread/{thread_id}/runs/stream` |
| Wait Run | `POST` | `/thread/{thread_id}/runs/wait` |
| Create Cron | `POST` | `/runs/crons` |
| Stream Run Stateless | `POST` | `/runs/stream` |
| Wait Run Stateless | `POST` | `/runs/wait` |
In this guide, we’ll show how to trigger a webhook after streaming a run.
@@ -25,36 +25,39 @@ In this guide, we’ll show how to trigger a webhook after streaming a run.
Before making API calls, set up your assistant and thread.
=== "Python"
```python
from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
print(thread)
```
```python
from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)
assistant_id = "agent"
thread = await client.threads.create()
print(thread)
```
=== "JavaScript"
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantID = "agent";
const thread = await client.threads.create();
console.log(thread);
```
```js
import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });
const assistantID = "agent";
const thread = await client.threads.create();
console.log(thread);
```
=== "CURL"
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/assistants/search \
--header 'Content-Type: application/json' \
--data '{ "limit": 10, "offset": 0 }' | jq -c 'map(select(.config == null or .config == {})) | .[0]' && \
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'
```
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/assistants/search \
--header 'Content-Type: application/json' \
--data '{ "limit": 10, "offset": 0 }' | jq -c 'map(select(.config == null or .config == {})) | .[0]' && \
curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'
```
Example response:
@@ -77,48 +80,51 @@ To use a webhook, specify the `webhook` parameter in your API request. When the
For example, if your server listens for webhook events at `https://my-server.app/my-webhook-endpoint`, include this in your request:
=== "Python"
```python
input = { "messages": [{ "role": "user", "content": "Hello!" }] }
async for chunk in client.runs.stream(
thread_id=thread["thread_id"],
assistant_id=assistant_id,
input=input,
stream_mode="events",
webhook="https://my-server.app/my-webhook-endpoint"
):
pass
```
```python
input = { "messages": [{ "role": "user", "content": "Hello!" }] }
async for chunk in client.runs.stream(
thread_id=thread["thread_id"],
assistant_id=assistant_id,
input=input,
stream_mode="events",
webhook="https://my-server.app/my-webhook-endpoint"
):
pass
```
=== "JavaScript"
```js
const input = { messages: [{ role: "human", content: "Hello!" }] };
const streamResponse = client.runs.stream(
thread["thread_id"],
assistantID,
{
input: input,
webhook: "https://my-server.app/my-webhook-endpoint"
}
);
```js
const input = { messages: [{ role: "human", content: "Hello!" }] };
for await (const chunk of streamResponse) {
// Handle stream output
}
```
const streamResponse = client.runs.stream(
thread["thread_id"],
assistantID,
{
input: input,
webhook: "https://my-server.app/my-webhook-endpoint"
}
);
for await (const chunk of streamResponse) {
// Handle stream output
}
```
=== "CURL"
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data '{
"assistant_id": <ASSISTANT_ID>,
"input": {"messages": [{"role": "user", "content": "Hello!"}]},
"webhook": "https://my-server.app/my-webhook-endpoint"
}'
```
```bash
curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data '{
"assistant_id": <ASSISTANT_ID>,
"input": {"messages": [{"role": "user", "content": "Hello!"}]},
"webhook": "https://my-server.app/my-webhook-endpoint"
}'
```
## Webhook payload
+7 -6
View File
@@ -50,9 +50,10 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
| <span style="white-space: nowrap;">`python_version`</span> | `3.11`, `3.12`, or `3.13`. Defaults to `3.11`. |
| <span style="white-space: nowrap;">`node_version`</span> | Specify `node_version: 20` to use LangGraph.js. |
| <span style="white-space: nowrap;">`pip_config_file`</span> | Path to `pip` config file. |
| <span style="white-space: nowrap;">`pip_installer`</span> | _(Added in v0.3)_ Optional. Python package installer selector. It can be set to `"auto"`, `"pip"`, or `"uv"`. From version&nbsp;0.3 onward the default strategy is to run `uv pip`, which typically delivers faster builds while remaining a drop-in replacement. In the uncommon situation where `uv` cannot handle your dependency graph or the structure of your `pyproject.toml`, specify `"pip"` here to revert to the earlier behaviour. |
| <span style="white-space: nowrap;">`dockerfile_lines`</span> | Array of additional lines to add to Dockerfile following the import from parent image. |
| <span style="white-space: nowrap;">`checkpointer`</span> | Configuration for the checkpointer. Contains a `ttl` field which is an object with the following keys: <ul><li>`strategy`: How to handle expired checkpoints (e.g., `"delete"`).</li><li>`sweep_interval_minutes`: How often to check for expired checkpoints (integer).</li><li>`default_ttl`: Default time-to-live for checkpoints in **minutes** (integer). Defines how long checkpoints are kept before the specified strategy is applied.</li></ul> |
| <span style="white-space: nowrap;">`http`</span> | HTTP server configuration with the following fields: <ul><li>`app`: Path to custom Starlette/FastAPI app (e.g., `"./src/agent/webapp.py:app"`). See [custom routes guide](../../how-tos/http/custom_routes.md).</li><li>`disable_assistants`: Disable `/assistants` routes</li><li>`disable_threads`: Disable `/threads` routes</li><li>`disable_runs`: Disable `/runs` routes</li><li>`disable_store`: Disable `/store` routes</li><li>`disable_meta`: Disable `/ok`, `/info`, `/metrics`, and `/docs` routes</li><li>`cors`: CORS configuration with fields for `allow_origins`, `allow_methods`, `allow_headers`, etc.</li><li>`configurable_headers`: Define which request headers to exclude or include as a run's configurable values.</li></ul> |
| <span style="white-space: nowrap;">`http`</span> | HTTP server configuration with the following fields: <ul><li>`app`: Path to custom Starlette/FastAPI app (e.g., `"./src/agent/webapp.py:app"`). See [custom routes guide](../../how-tos/http/custom_routes.md).</li><li>`disable_assistants`: Disable `/assistants` routes</li><li>`disable_threads`: Disable `/threads` routes</li><li>`disable_runs`: Disable `/runs` routes</li><li>`disable_store`: Disable `/store` routes</li><li>`disable_meta`: Disable `/ok`, `/info`, `/metrics`, and `/docs` routes</li><li>`disable_mcp`: Disable `/mcp` routes</li><li>`cors`: CORS configuration with fields for `allow_origins`, `allow_methods`, `allow_headers`, etc.</li><li>`configurable_headers`: Define which request headers to exclude or include as a run's configurable values.</li></ul> |
=== "JS"
@@ -128,7 +129,7 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
- `cohere:embed-english-v3.0`: 1024
- `cohere:embed-english-light-v3.0`: 384
- `cohere:embed-multilingual-v3.0`: 1024
- `cohere:embed-multilingual-light-v3.0`: 384
- `cohere:embed-multilingual-light-v3.0`: 384
#### Semantic search with a custom embedding function
@@ -361,8 +362,8 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
**Options**
| Option | Default | Description |
| -------------------- | ---------------- | ---------------------------------------------------------------------------------------------------------------------------- |
| Option | Default | Description |
| -------------------- | ---------------- | --------------------------------------------------------------------------------------------------------------- |
| `--platform TEXT` | | Target platform(s) to build the Docker image for. Example: `langgraph build --platform linux/amd64,linux/arm64` |
| `-t, --tag TEXT` | | **Required**. Tag for the Docker image. Example: `langgraph build -t my-image` |
| `--pull / --no-pull` | `--pull` | Build with latest remote Docker image. Use `--no-pull` for running the LangGraph Platform API server with locally built images. |
@@ -381,8 +382,8 @@ The LangGraph CLI requires a JSON configuration file that follows this [schema](
**Options**
| Option | Default | Description |
| -------------------- | ---------------- | ---------------------------------------------------------------------------------------------------------------------------- |
| Option | Default | Description |
| -------------------- | ---------------- | --------------------------------------------------------------------------------------------------------------- |
| `--platform TEXT` | | Target platform(s) to build the Docker image for. Example: `langgraph build --platform linux/amd64,linux/arm64` |
| `-t, --tag TEXT` | | **Required**. Tag for the Docker image. Example: `langgraph build -t my-image` |
| `--no-pull` | | Use locally built images. Defaults to `false` to build with latest remote Docker image. |
+2 -3
View File
@@ -50,11 +50,10 @@ Set this environment variable to have a deployment send traces to a self-hosted
## `LANGSMITH_TRACING`
!!! info "Only for Self-Hosted Data Plane, Self-Hosted Control Plane, and Standalone Container"
Disabling LangSmith tracing is only available for [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) deployments.
Set `LANGSMITH_TRACING` to `false` to disable tracing to LangSmith.
Defaults to `true`.
## `LOG_LEVEL`
Configure [log level](https://docs.python.org/3/library/logging.html#logging-levels). Defaults to `INFO`.
+6 -1
View File
@@ -9,13 +9,18 @@ search:
## Installation
The LangGraph CLI can be installed via pip:
The LangGraph CLI can be installed via pip or [Homebrew](https://brew.sh/):
=== "pip"
```bash
pip install langgraph-cli
```
=== "Homebrew"
```bash
brew install langgraph-cli
```
## Commands
LangGraph CLI provides the following core functionality:
@@ -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.
+2 -1
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)
@@ -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.
+2 -1
View File
@@ -87,6 +87,7 @@ One of the most common agent types is a [tool-calling agent](../agents/overview.
```python
from langchain_core.tools import tool
@tool
def transfer_to_bob():
"""Transfer to bob."""
return Command(
@@ -414,4 +415,4 @@ There are two high-level approaches to achieve that:
An agent might need to have a different state schema from the rest of the agents. For example, a search agent might only need to keep track of queries and retrieved documents. There are two ways to achieve this in LangGraph:
- Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, it’s important to [add input / output transformations](../how-tos/subgraph.ipynb#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb/#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb/#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
+2 -2
View File
@@ -34,7 +34,7 @@ def read_root():
## Configure `langgraph.json`
Add the following to your `langgraph.json` configuration file. Make sure the path points to the `app.py` file you created above.
Add the following to your `langgraph.json` configuration file. Make sure the path points to the FastAPI application instance `app` in the `webapp.py` file you created above.
```json
{
@@ -71,4 +71,4 @@ You can deploy this app as-is to LangGraph Platform or to your self-hosted platf
## Next steps
Now that you've added a custom route to your deployment, you can use this same technique to further customize how your server behaves, such as defining custom [custom middleware](./custom_middleware.md) and [custom lifespan events](./custom_lifespan.md).
Now that you've added a custom route to your deployment, you can use this same technique to further customize how your server behaves, such as defining custom [custom middleware](./custom_middleware.md) and [custom lifespan events](./custom_lifespan.md).
+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. Let’s 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,7 +27,6 @@ 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
@@ -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,63 +87,26 @@ 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
@@ -200,23 +114,14 @@ This tells the graph to terminate after running the chatbot node.
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)
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 +131,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
Now run the chatbot!
:::python
!!! tip
You can exit the chat loop at any time by typing `quit`, `exit`, or `q`.
@@ -281,41 +169,11 @@ 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
@@ -348,41 +206,9 @@ 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.
+1 -313
View File
@@ -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,24 +157,13 @@ 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"
from typing import Annotated
@@ -394,51 +203,7 @@ 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
In the next tutorial, you will [add human-in-the-loop to the chatbot](./4-human-in-the-loop.md) to handle situations where it may need guidance or verification before proceeding.
In the next tutorial, you will [add human-in-the-loop to the chatbot](./4-human-in-the-loop.md) to handle situations where it may need guidance or verification before proceeding.
@@ -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 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."}]
================================== 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 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."}]
================================== 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:
+1 -1
View File
@@ -648,7 +648,7 @@ With orchestrator-worker, an orchestrator breaks down a task and delegates each
Because orchestrator-worker workflows are common, LangGraph **has the `Send` API to support this**. It lets you dynamically create worker nodes and send each one a specific input. Each worker has its own state, and all worker outputs are written to a *shared state key* that is accessible to the orchestrator graph. This gives the orchestrator access to all worker output and allows it to synthesize them into a final output. As you can see below, we iterate over a list of sections and `Send` each to a worker node. See further documentation [here](https://langchain-ai.github.io/langgraph/how-tos/map-reduce/) and [here](https://langchain-ai.github.io/langgraph/concepts/low_level/#send).
```python
from langgraph.constants import Send
from langgraph.types import Send
# Graph state
+2 -13
View File
@@ -179,6 +179,7 @@ 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
@@ -364,16 +365,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/
@@ -398,6 +389,4 @@ extra_css:
- stylesheets/logos.css
- stylesheets/sticky_navigation.css
- stylesheets/agent_graph_widget.css
- language-switcher.css
extra_javascript:
- language-switcher.js
-70
View File
@@ -1,70 +0,0 @@
{#-
This file was automatically generated - do not edit
-#}
{% set class = "md-header" %}
{% if "navigation.tabs.sticky" in features %}
{% set class = class ~ " md-header--shadow md-header--lifted" %}
{% elif "navigation.tabs" not in features %}
{% set class = class ~ " md-header--shadow" %}
{% endif %}
<header class="{{ class }}" data-md-component="header">
<nav class="md-header__inner md-grid" aria-label="{{ lang.t('header') }}">
<a href="{{ config.extra.homepage | d(nav.homepage.url, true) | url }}" title="{{ config.site_name | e }}" class="md-header__button md-logo" aria-label="{{ config.site_name }}" data-md-component="logo">
{% include "partials/logo.html" %}
</a>
<label class="md-header__button md-icon" for="__drawer">
{% set icon = config.theme.icon.menu or "material/menu" %}
{% include ".icons/" ~ icon ~ ".svg" %}
</label>
<div class="md-header__title" data-md-component="header-title">
<div class="md-header__ellipsis">
<div class="md-header__topic">
<span class="md-ellipsis">
{{ config.site_name }}
</span>
</div>
<div class="md-header__topic" data-md-component="header-topic">
<span class="md-ellipsis">
{% if page.meta and page.meta.title %}
{{ page.meta.title }}
{% else %}
{{ page.title }}
{% endif %}
</span>
</div>
</div>
</div>
{% if config.theme.palette %}
{% if not config.theme.palette is mapping %}
{% include "partials/palette.html" %}
{% endif %}
{% endif %}
{% if not config.theme.palette is mapping %}
{% include "partials/javascripts/palette.html" %}
{% endif %}
{% if config.extra.alternate %}
{% include "partials/alternate.html" %}
{% endif %}
{% if "material/search" in config.plugins %}
{% set search = config.plugins["material/search"] | attr("config") %}
{% if search.enabled %}
<label class="md-header__button md-icon" for="__search">
{% set icon = config.theme.icon.search or "material/magnify" %}
{% include ".icons/" ~ icon ~ ".svg" %}
</label>
{% include "partials/search.html" %}
{% endif %}
{% endif %}
{% if config.repo_url %}
<div class="md-header__source">
{% include "partials/source.html" %}
</div>
{% endif %}
{% include "partials/language-toggle.html" %}
</nav>
{% if "navigation.tabs.sticky" in features %}
{% if "navigation.tabs" in features %}
{% include "partials/tabs.html" %}
{% endif %}
{% endif %}
</header>
@@ -23,6 +23,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
@@ -456,4 +457,4 @@ class PostgresSaver(BasePostgresSaver):
)
__all__ = ["PostgresSaver", "BasePostgresSaver", "Conn"]
__all__ = ["PostgresSaver", "BasePostgresSaver", "ShallowPostgresSaver", "Conn"]
@@ -23,6 +23,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
@@ -559,4 +560,4 @@ class AsyncPostgresSaver(BasePostgresSaver):
).result()
__all__ = ["AsyncPostgresSaver", "Conn"]
__all__ = ["AsyncPostgresSaver", "AsyncShallowPostgresSaver", "Conn"]
@@ -168,7 +168,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
checkpoint["channel_versions"][TASKS] = (
max(checkpoint["channel_versions"].values())
if checkpoint["channel_versions"]
else self.get_next_version(None)
else self.get_next_version(None, None)
)
def _load_blobs(
@@ -246,7 +246,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
for idx, (channel, value) in enumerate(writes)
]
def get_next_version(self, current: str | None) -> str:
def get_next_version(self, current: str | None, channel: None) -> str:
if current is None:
current_v = 0
elif isinstance(current, int):
@@ -0,0 +1,959 @@
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: Checkpoint = {
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
"pending_sends": [
self.serde.loads_typed((t.decode(), v))
for t, v in value["pending_sends"]
]
if value["pending_sends"]
else [],
}
yield CheckpointTuple(
config={
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
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: Checkpoint = {
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
"pending_sends": [
self.serde.loads_typed((t.decode(), v))
for t, v in value["pending_sends"]
]
if value["pending_sends"]
else [],
}
return CheckpointTuple(
config={
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
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(copy),
Jsonb(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: Checkpoint = {
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
"pending_sends": [
self.serde.loads_typed((t.decode(), v))
for t, v in value["pending_sends"]
]
if value["pending_sends"]
else [],
}
yield CheckpointTuple(
config={
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
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: Checkpoint = {
**value["checkpoint"],
"channel_values": self._load_blobs(value["channel_values"]),
"pending_sends": [
self.serde.loads_typed((t.decode(), v))
for t, v in value["pending_sends"]
]
if value["pending_sends"]
else [],
}
return CheckpointTuple(
config={
"configurable": {
"thread_id": thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint["id"],
}
},
checkpoint=checkpoint,
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(copy),
Jsonb(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,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 -6
View File
@@ -14,10 +14,14 @@ 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 +112,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 +206,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 = {
@@ -194,12 +228,11 @@ async def test_combined_metadata(saver_name: str, test_data) -> None:
checkpoint = await saver.aget_tuple(config)
assert checkpoint.metadata == {
**metadata,
"thread_id": "thread-2",
"run_id": "my_run_id",
}
@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(
+32 -6
View File
@@ -15,10 +15,11 @@ from langgraph.checkpoint.base import (
EXCLUDED_METADATA_KEYS,
Checkpoint,
CheckpointMetadata,
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.postgres import PostgresSaver, ShallowPostgresSaver
from langgraph.checkpoint.serde.types import TASKS
from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
from tests.conftest import DEFAULT_POSTGRES_URI
@@ -97,11 +98,37 @@ def _base_saver():
conn.execute(f"DROP DATABASE {database}")
@contextmanager
def _shallow_saver():
"""Fixture for regular connection mode testing with a shallow checkpointer."""
database = f"test_{uuid4().hex[:16]}"
# create unique db
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
conn.execute(f"CREATE DATABASE {database}")
try:
with Connection.connect(
DEFAULT_POSTGRES_URI + database,
autocommit=True,
prepare_threshold=0,
row_factory=dict_row,
) as conn:
checkpointer = ShallowPostgresSaver(conn)
checkpointer.setup()
yield checkpointer
finally:
# drop unique db
with Connection.connect(DEFAULT_POSTGRES_URI, autocommit=True) as conn:
conn.execute(f"DROP DATABASE {database}")
@contextmanager
def _saver(name: str):
if name == "base":
with _base_saver() as saver:
yield saver
elif name == "shallow":
with _shallow_saver() as saver:
yield saver
elif name == "pool":
with _pool_saver() as saver:
yield saver
@@ -161,7 +188,7 @@ def test_data():
}
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
def test_combined_metadata(saver_name: str, test_data) -> None:
with _saver(saver_name) as saver:
config = {
@@ -183,12 +210,11 @@ def test_combined_metadata(saver_name: str, test_data) -> None:
checkpoint = saver.get_tuple(config)
assert checkpoint.metadata == {
**metadata,
"thread_id": "thread-2",
"run_id": "my_run_id",
}
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
def test_search(saver_name: str, test_data) -> None:
with _saver(saver_name) as saver:
configs = test_data["configs"]
@@ -237,7 +263,7 @@ def test_search(saver_name: str, test_data) -> None:
} == {"", "inner"}
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe", "shallow"])
def test_null_chars(saver_name: str, test_data) -> None:
with _saver(saver_name) as saver:
config = saver.put(
+2 -1
View File
@@ -308,7 +308,7 @@ wheels = [
[[package]]
name = "langgraph-checkpoint"
version = "2.0.26"
version = "2.1.0"
source = { editable = "../checkpoint" }
dependencies = [
{ name = "langchain-core" },
@@ -328,6 +328,7 @@ dev = [
{ name = "mypy" },
{ name = "numpy" },
{ name = "pandas" },
{ name = "pandas-stubs", specifier = ">=2.2.2.240807" },
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
@@ -536,7 +536,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
"""
raise NotImplementedError(_AIO_ERROR_MSG)
def get_next_version(self, current: str | None) -> str:
def get_next_version(self, current: str | None, channel: None) -> str:
"""Generate the next version ID for a channel.
This method creates a new version identifier for a channel based on its current version.
@@ -591,7 +591,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
)
await self.conn.commit()
def get_next_version(self, current: str | None) -> str:
def get_next_version(self, current: str | None, channel: None) -> str:
"""Generate the next version ID for a channel.
This method creates a new version identifier for a channel based on its current version.
@@ -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"],
)
+4 -11
View File
@@ -6,9 +6,10 @@ from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
Checkpoint,
CheckpointMetadata,
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
class TestAsyncSqliteSaver:
@@ -70,7 +71,6 @@ class TestAsyncSqliteSaver:
checkpoint = await saver.aget_tuple(config)
assert checkpoint is not None and checkpoint.metadata == {
**self.metadata_2,
"thread_id": "thread-2",
"run_id": "my_run_id",
}
@@ -91,18 +91,11 @@ class TestAsyncSqliteSaver:
search_results_1 = [c async for c in saver.alist(None, filter=query_1)]
assert len(search_results_1) == 1
assert search_results_1[0].metadata == {
"thread_id": "thread-1",
"thread_ts": "1",
**self.metadata_1,
}
assert search_results_1[0].metadata == self.metadata_1
search_results_2 = [c async for c in saver.alist(None, filter=query_2)]
assert len(search_results_2) == 1
assert search_results_2[0].metadata == {
"thread_id": "thread-2",
**self.metadata_2,
}
assert search_results_2[0].metadata == self.metadata_2
search_results_3 = [c async for c in saver.alist(None, filter=query_3)]
assert len(search_results_3) == 3
+4 -11
View File
@@ -6,10 +6,11 @@ from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
Checkpoint,
CheckpointMetadata,
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.sqlite.utils import _metadata_predicate, search_where
from tests.checkpoint_utils import create_checkpoint, empty_checkpoint
class TestSqliteSaver:
@@ -71,7 +72,6 @@ class TestSqliteSaver:
checkpoint = saver.get_tuple(config)
assert checkpoint is not None and checkpoint.metadata == {
**self.metadata_2,
"thread_id": "thread-2",
"run_id": "my_run_id",
}
@@ -94,18 +94,11 @@ class TestSqliteSaver:
search_results_1 = list(saver.list(None, filter=query_1))
assert len(search_results_1) == 1
assert search_results_1[0].metadata == {
"thread_id": "thread-1",
"thread_ts": "1",
**self.metadata_1,
}
assert search_results_1[0].metadata == self.metadata_1
search_results_2 = list(saver.list(None, filter=query_2))
assert len(search_results_2) == 1
assert search_results_2[0].metadata == {
"thread_id": "thread-2",
**self.metadata_2,
}
assert search_results_2[0].metadata == self.metadata_2
search_results_3 = list(saver.list(None, filter=query_3))
assert len(search_results_3) == 3
+2 -1
View File
@@ -320,7 +320,7 @@ wheels = [
[[package]]
name = "langgraph-checkpoint"
version = "2.0.26"
version = "2.1.0"
source = { editable = "../checkpoint" }
dependencies = [
{ name = "langchain-core" },
@@ -340,6 +340,7 @@ dev = [
{ name = "mypy" },
{ name = "numpy" },
{ name = "pandas" },
{ name = "pandas-stubs", specifier = ">=2.2.2.240807" },
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
@@ -1,6 +1,6 @@
from __future__ import annotations
from collections.abc import AsyncIterator, Iterator, Sequence
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
from typing import ( # noqa: UP035
Any,
Generic,
@@ -13,6 +13,7 @@ from typing import ( # noqa: UP035
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base.id import uuid6
from langgraph.checkpoint.serde.base import SerializerProtocol, maybe_add_typed_methods
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from langgraph.checkpoint.serde.types import (
@@ -20,6 +21,7 @@ from langgraph.checkpoint.serde.types import (
INTERRUPT,
RESUME,
SCHEDULED,
ChannelProtocol,
)
V = TypeVar("V", int, float, str)
@@ -89,6 +91,7 @@ def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
channel_values=checkpoint["channel_values"].copy(),
channel_versions=checkpoint["channel_versions"].copy(),
versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
pending_sends=checkpoint.get("pending_sends", []).copy(),
)
@@ -125,6 +128,15 @@ class BaseCheckpointSaver(Generic[V]):
) -> None:
self.serde = maybe_add_typed_methods(serde or self.serde)
@property
def config_specs(self) -> list:
"""Define the configuration options for the checkpoint saver.
Returns:
list: List of configuration field specs.
"""
return []
def get(self, config: RunnableConfig) -> Checkpoint | None:
"""Fetch a checkpoint using the given configuration.
@@ -334,7 +346,7 @@ class BaseCheckpointSaver(Generic[V]):
"""
raise NotImplementedError
def get_next_version(self, current: V | None) -> V:
def get_next_version(self, current: V | None, channel: None) -> V:
"""Generate the next version ID for a channel.
Default is to use integer versions, incrementing by 1. If you override, you can use str/int/float versions,
@@ -342,6 +354,7 @@ class BaseCheckpointSaver(Generic[V]):
Args:
current: The current version identifier (int, float, or str).
channel: Deprecated argument, kept for backwards compatibility.
Returns:
V: The next version identifier, which must be increasing.
@@ -379,11 +392,10 @@ def get_checkpoint_metadata(
for obj in (config.get("metadata"), config.get("configurable")):
if not obj:
continue
for key in obj:
for key, v in obj.items():
if key in metadata or key in EXCLUDED_METADATA_KEYS or key.startswith("__"):
continue
v = obj[key]
if isinstance(v, str):
elif isinstance(v, str):
metadata[key] = v.replace("\u0000", "")
elif isinstance(v, (int, bool, float)):
metadata[key] = v
@@ -400,7 +412,65 @@ Each Checkpointer implementation should use this mapping in put_writes.
WRITES_IDX_MAP = {ERROR: -1, SCHEDULED: -2, INTERRUPT: -3, RESUME: -4}
EXCLUDED_METADATA_KEYS = {
"thread_id",
"thread_ts",
"checkpoint_id",
"checkpoint_ns",
"checkpoint_map",
"langgraph_step",
"langgraph_node",
"langgraph_triggers",
"langgraph_path",
"langgraph_checkpoint_ns",
}
# --- below are deprecated utilities used by past versions of LangGraph ---
LATEST_VERSION = 2
def empty_checkpoint() -> Checkpoint:
from datetime import datetime, timezone
return Checkpoint(
v=LATEST_VERSION,
id=str(uuid6(clock_seq=-2)),
ts=datetime.now(timezone.utc).isoformat(),
channel_values={},
channel_versions={},
versions_seen={},
pending_sends=[],
)
def create_checkpoint(
checkpoint: Checkpoint,
channels: Mapping[str, ChannelProtocol] | None,
step: int,
*,
id: str | None = None,
) -> Checkpoint:
"""Create a checkpoint for the given channels."""
from datetime import datetime, timezone
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=LATEST_VERSION,
ts=ts,
id=id or str(uuid6(clock_seq=step)),
channel_values=values,
channel_versions=checkpoint["channel_versions"],
versions_seen=checkpoint["versions_seen"],
pending_sends=checkpoint.get("pending_sends", []),
)
@@ -512,7 +512,7 @@ class InMemorySaver(
"""
return self.delete_thread(thread_id)
def get_next_version(self, current: str | None) -> str:
def get_next_version(self, current: str | None, channel: None) -> str:
if current is None:
current_v = 0
elif isinstance(current, int):
@@ -1,4 +1,13 @@
from typing import Any, Protocol, TypeVar, runtime_checkable
from collections.abc import Sequence
from typing import (
Any,
Optional,
Protocol,
TypeVar,
runtime_checkable,
)
from typing_extensions import Self
ERROR = "__error__"
SCHEDULED = "__scheduled__"
@@ -11,6 +20,25 @@ Update = TypeVar("Update", contravariant=True)
C = TypeVar("C")
class ChannelProtocol(Protocol[Value, Update, C]):
# Mirrors langgraph.channels.base.BaseChannel
@property
def ValueType(self) -> Any: ...
@property
def UpdateType(self) -> Any: ...
def checkpoint(self) -> Optional[C]: ...
def from_checkpoint(self, checkpoint: Optional[C]) -> Self: ...
def update(self, values: Sequence[Update]) -> bool: ...
def get(self) -> Value: ...
def consume(self) -> bool: ...
@runtime_checkable
class SendProtocol(Protocol):
# Mirrors langgraph.constants.Send
+2 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-checkpoint"
version = "2.0.26"
version = "2.1.0"
description = "Library with base interfaces for LangGraph checkpoint savers."
authors = []
requires-python = ">=3.9"
@@ -31,6 +31,7 @@ dev = [
"dataclasses-json",
"numpy",
"pandas",
"pandas-stubs>=2.2.2.240807",
]
[tool.hatch.build.targets.wheel]
-53
View File
@@ -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"],
)
+5 -22
View File
@@ -6,12 +6,10 @@ from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
Checkpoint,
CheckpointMetadata,
)
from langgraph.checkpoint.memory import InMemorySaver
from tests.checkpoint_utils import (
create_checkpoint,
empty_checkpoint,
)
from langgraph.checkpoint.memory import InMemorySaver
class TestMemorySaver:
@@ -77,7 +75,6 @@ class TestMemorySaver:
assert checkpoint is not None
assert checkpoint.metadata == {
**self.metadata_2,
"thread_id": "thread-2",
"run_id": "my_run_id",
}
@@ -114,18 +111,11 @@ class TestMemorySaver:
search_results_1 = list(self.memory_saver.list(None, filter=query_1))
assert len(search_results_1) == 1
assert search_results_1[0].metadata == {
"thread_id": "thread-1",
"thread_ts": "1",
**self.metadata_1,
}
assert search_results_1[0].metadata == self.metadata_1
search_results_2 = list(self.memory_saver.list(None, filter=query_2))
assert len(search_results_2) == 1
assert search_results_2[0].metadata == {
"thread_id": "thread-2",
**self.metadata_2,
}
assert search_results_2[0].metadata == self.metadata_2
search_results_3 = list(self.memory_saver.list(None, filter=query_3))
assert len(search_results_3) == 3
@@ -180,20 +170,13 @@ class TestMemorySaver:
c async for c in self.memory_saver.alist(None, filter=query_1)
]
assert len(search_results_1) == 1
assert search_results_1[0].metadata == {
"thread_id": "thread-1",
"thread_ts": "1",
**self.metadata_1,
}
assert search_results_1[0].metadata == self.metadata_1
search_results_2 = [
c async for c in self.memory_saver.alist(None, filter=query_2)
]
assert len(search_results_2) == 1
assert search_results_2[0].metadata == {
"thread_id": "thread-2",
**self.metadata_2,
}
assert search_results_2[0].metadata == self.metadata_2
search_results_3 = [
c async for c in self.memory_saver.alist(None, filter=query_3)
+896 -848
View File
File diff suppressed because it is too large Load Diff
+5
View File
@@ -330,6 +330,11 @@ class HttpConfig(TypedDict, total=False):
disable_store: bool
"""Optional. If True, /store routes are removed, disabling direct store interactions via HTTP.
Default is False.
"""
disable_mcp: bool
"""Optional. If True, /mcp routes are removed, disabling the MCP server.
Default is False.
"""
disable_meta: bool
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-cli"
version = "0.3.2"
version = "0.3.3"
description = "CLI for interacting with LangGraph API"
authors = []
requires-python = ">=3.9"
+4
View File
@@ -499,6 +499,10 @@
"type": "boolean",
"description": "Optional. If True, /assistants routes are removed from the server.\n\nDefault is False (meaning /assistants is enabled).\n"
},
"disable_mcp": {
"type": "boolean",
"description": "Optional. If True, /mcp routes are removed, disabling the MCP server.\n\nDefault is False.\n"
},
"disable_meta": {
"type": "boolean",
"description": "Optional. If True, all meta endpoints (/ok, /info, /metrics, /docs) are disabled.\n\nDefault is False.\n"
+4
View File
@@ -499,6 +499,10 @@
"type": "boolean",
"description": "Optional. If True, /assistants routes are removed from the server.\n\nDefault is False (meaning /assistants is enabled).\n"
},
"disable_mcp": {
"type": "boolean",
"description": "Optional. If True, /mcp routes are removed, disabling the MCP server.\n\nDefault is False.\n"
},
"disable_meta": {
"type": "boolean",
"description": "Optional. If True, all meta endpoints (/ok, /info, /metrics, /docs) are disabled.\n\nDefault is False.\n"
+1 -1
View File
@@ -501,7 +501,7 @@ wheels = [
[[package]]
name = "langgraph-cli"
version = "0.3.2"
version = "0.3.3"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -1,14 +1,15 @@
from langgraph.channels.any_value import AnyValue
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue, LastValueAfterFinish
from langgraph.channels.last_value import LastValue
from langgraph.channels.topic import Topic
from langgraph.channels.untracked_value import UntrackedValue
__all__ = [
"LastValue",
"LastValueAfterFinish",
"Topic",
"BinaryOperatorAggregate",
"UntrackedValue",
"EphemeralValue",
"AnyValue",
]
@@ -0,0 +1,66 @@
from collections.abc import Sequence
from typing import Generic
from typing_extensions import Self
from langgraph.channels.base import BaseChannel, Value
from langgraph.constants import MISSING
from langgraph.errors import EmptyChannelError, InvalidUpdateError
class UntrackedValue(Generic[Value], BaseChannel[Value, Value, Value]):
"""Stores the last value received, never checkpointed."""
__slots__ = ("value", "guard")
def __init__(self, typ: type[Value], guard: bool = True) -> None:
super().__init__(typ)
self.guard = guard
self.value = MISSING
def __eq__(self, value: object) -> bool:
return isinstance(value, UntrackedValue) and value.guard == self.guard
@property
def ValueType(self) -> type[Value]:
"""The type of the value stored in the channel."""
return self.typ
@property
def UpdateType(self) -> type[Value]:
"""The type of the update received by the channel."""
return self.typ
def copy(self) -> Self:
"""Return a copy of the channel."""
empty = self.__class__(self.typ, self.guard)
empty.key = self.key
empty.value = self.value
return empty
def checkpoint(self) -> Value:
return MISSING
def from_checkpoint(self, checkpoint: Value) -> Self:
empty = self.__class__(self.typ, self.guard)
empty.key = self.key
return empty
def update(self, values: Sequence[Value]) -> bool:
if len(values) == 0:
return False
if len(values) != 1 and self.guard:
raise InvalidUpdateError(
f"At key '{self.key}': UntrackedValue(guard=True) can receive only one value per step. Use guard=False if you want to store any one of multiple values."
)
self.value = values[-1]
return True
def get(self) -> Value:
if self.value is MISSING:
raise EmptyChannelError()
return self.value
def is_available(self) -> bool:
return self.value is not MISSING
+3 -3
View File
@@ -38,7 +38,7 @@ from langgraph.pregel.read import PregelNode
from langgraph.pregel.write import ChannelWrite, ChannelWriteEntry
from langgraph.store.base import BaseStore
from langgraph.types import _DC_KWARGS, CachePolicy, RetryPolicy, StreamMode
from langgraph.warnings import LangGraphDeprecatedSinceV10
from langgraph.warnings import LangGraphDeprecatedSinceV05
class TaskFunction(Generic[P, T]):
@@ -179,7 +179,7 @@ def task(
if (retry := kwargs.get("retry", UNSET)) is not UNSET:
warnings.warn(
"`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
category=LangGraphDeprecatedSinceV10,
category=LangGraphDeprecatedSinceV05,
)
if retry_policy is None:
retry_policy = retry # type: ignore[assignment]
@@ -383,7 +383,7 @@ class entrypoint:
if (retry := kwargs.get("retry", UNSET)) is not UNSET:
warnings.warn(
"`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
category=LangGraphDeprecatedSinceV10,
category=LangGraphDeprecatedSinceV05,
)
if retry_policy is None:
retry_policy = retry # type: ignore[assignment]
+2 -1
View File
@@ -1,11 +1,12 @@
from langgraph.constants import END, START
from langgraph.graph.message import MessagesState, add_messages
from langgraph.graph.message import MessageGraph, MessagesState, add_messages
from langgraph.graph.state import StateGraph
__all__ = [
"END",
"START",
"StateGraph",
"MessageGraph",
"add_messages",
"MessagesState",
]
+52
View File
@@ -25,6 +25,7 @@ from langchain_core.messages import (
from typing_extensions import TypedDict
from langgraph.constants import CONF, CONFIG_KEY_SEND
from langgraph.graph.state import StateGraph
Messages = Union[list[MessageLikeRepresentation], MessageLikeRepresentation]
@@ -226,6 +227,57 @@ def add_messages(
return merged
class MessageGraph(StateGraph):
"""A StateGraph where every node receives a list of messages as input and returns one or more messages as output.
MessageGraph is a subclass of StateGraph whose entire state is a single, append-only* list of messages.
Each node in a MessageGraph takes a list of messages as input and returns zero or more
messages as output. The `add_messages` function is used to merge the output messages from each node
into the existing list of messages in the graph's state.
Examples:
```pycon
>>> from langgraph.graph.message import MessageGraph
...
>>> builder = MessageGraph()
>>> builder.add_node("chatbot", lambda state: [("assistant", "Hello!")])
>>> builder.set_entry_point("chatbot")
>>> builder.set_finish_point("chatbot")
>>> builder.compile().invoke([("user", "Hi there.")])
[HumanMessage(content="Hi there.", id='...'), AIMessage(content="Hello!", id='...')]
```
```pycon
>>> from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
>>> from langgraph.graph.message import MessageGraph
...
>>> builder = MessageGraph()
>>> builder.add_node(
... "chatbot",
... lambda state: [
... AIMessage(
... content="Hello!",
... tool_calls=[{"name": "search", "id": "123", "args": {"query": "X"}}],
... )
... ],
... )
>>> builder.add_node(
... "search", lambda state: [ToolMessage(content="Searching...", tool_call_id="123")]
... )
>>> builder.set_entry_point("chatbot")
>>> builder.add_edge("chatbot", "search")
>>> builder.set_finish_point("search")
>>> builder.compile().invoke([HumanMessage(content="Hi there. Can you search for X?")])
{'messages': [HumanMessage(content="Hi there. Can you search for X?", id='b8b7d8f4-7f4d-4f4d-9c1d-f8b8d8f4d9c1'),
AIMessage(content="Hello!", id='f4d9c1d8-8d8f-4d9c-b8b7-d8f4f4d9c1d8'),
ToolMessage(content="Searching...", id='d8f4f4d9-c1d8-4f4d-b8b7-d8f4f4d9c1d8', tool_call_id="123")]}
```
"""
def __init__(self) -> None:
super().__init__(Annotated[list[AnyMessage], add_messages]) # type: ignore[arg-type]
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
+17 -6
View File
@@ -86,7 +86,7 @@ from langgraph.utils.fields import (
)
from langgraph.utils.pydantic import create_model
from langgraph.utils.runnable import coerce_to_runnable
from langgraph.warnings import LangGraphDeprecatedSinceV10
from langgraph.warnings import LangGraphDeprecatedSinceV05
logger = logging.getLogger(__name__)
@@ -148,6 +148,15 @@ class _NodeWithConfigWriterStore(Protocol[StateT_contra]):
) -> Any: ...
class _Invokable(Protocol[StateT_contra]):
def invoke(
self,
input: StateT_contra,
config: RunnableConfig | None = None,
**kwargs: Any,
) -> Any: ...
# TODO: we probably don't want to explicitly support the config / store signatures once
# we move to adding a context arg. Maybe what we do is we add support for kwargs with param spec
# this is purely for typing purposes though, so can easily change in the coming weeks.
@@ -160,6 +169,7 @@ StateNode: TypeAlias = Union[
_NodeWithConfigWriter[StateT_contra],
_NodeWithConfigStore[StateT_contra],
_NodeWithConfigWriterStore[StateT_contra],
_Invokable[StateT_contra],
]
@@ -261,7 +271,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
if (input_ := kwargs.get("input", UNSET)) is not UNSET:
warnings.warn(
"`input` is deprecated and will be removed. Please use `input_schema` instead.",
category=LangGraphDeprecatedSinceV10,
category=LangGraphDeprecatedSinceV05,
stacklevel=2,
)
if input_schema is None:
@@ -270,7 +280,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
if (output := kwargs.get("output", UNSET)) is not UNSET:
warnings.warn(
"`output` is deprecated and will be removed. Please use `output_schema` instead.",
category=LangGraphDeprecatedSinceV10,
category=LangGraphDeprecatedSinceV05,
stacklevel=2,
)
if output_schema is None:
@@ -436,7 +446,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
if (retry := kwargs.get("retry", UNSET)) is not UNSET:
warnings.warn(
"`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
category=LangGraphDeprecatedSinceV10,
category=LangGraphDeprecatedSinceV05,
)
if retry_policy is None:
retry_policy = retry # type: ignore[assignment]
@@ -444,7 +454,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
if (input_ := kwargs.get("input", UNSET)) is not UNSET:
warnings.warn(
"`input` is deprecated and will be removed. Please use `input_schema` instead.",
category=LangGraphDeprecatedSinceV10,
category=LangGraphDeprecatedSinceV05,
)
if input_schema is None:
input_schema = cast(Union[type[InputT], None], input_)
@@ -535,7 +545,7 @@ class StateGraph(Generic[StateT, InputT, OutputT]):
if input_schema is not None:
self._add_schema(input_schema)
self.nodes[node] = StateNodeSpec(
coerce_to_runnable(action, name=node, trace=False), # type: ignore
coerce_to_runnable(action, name=node, trace=False), # type: ignore[arg-type]
metadata,
input=input_schema or self.state_schema,
retry_policy=retry_policy,
@@ -1101,6 +1111,7 @@ class CompiledStateGraph(
def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
"""Migrate a checkpoint to new channel layout."""
super()._migrate_checkpoint(checkpoint)
values = checkpoint["channel_values"]
versions = checkpoint["channel_versions"]
+17 -19
View File
@@ -32,7 +32,6 @@ from langgraph.checkpoint.base import (
BaseCheckpointSaver,
Checkpoint,
CheckpointTuple,
copy_checkpoint,
)
from langgraph.config import get_config
from langgraph.constants import (
@@ -79,6 +78,7 @@ from langgraph.pregel.algo import (
from langgraph.pregel.call import identifier
from langgraph.pregel.checkpoint import (
channels_from_checkpoint,
copy_checkpoint,
create_checkpoint,
empty_checkpoint,
)
@@ -908,7 +908,12 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
"""Migrate a saved checkpoint to new channel layout."""
pass
if checkpoint["v"] < 4 and checkpoint.get("pending_sends"):
pending_sends: list[Send] = checkpoint.pop("pending_sends")
checkpoint["channel_values"][TASKS] = pending_sends
checkpoint["channel_versions"][TASKS] = max(
checkpoint["channel_versions"].values()
)
def _prepare_state_snapshot(
self,
@@ -1410,10 +1415,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
)
},
)
checkpoint_metadata = config["metadata"]
if saved:
checkpoint_config = patch_configurable(config, saved.config[CONF])
checkpoint_metadata = {**saved.metadata, **checkpoint_metadata}
channels, managed = channels_from_checkpoint(
self.channels,
checkpoint,
@@ -1478,7 +1481,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
create_checkpoint(checkpoint, None, step),
{
**checkpoint_metadata,
"source": "update",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1501,7 +1503,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
next_checkpoint,
{
**checkpoint_metadata,
"source": "update",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1538,9 +1539,11 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
create_checkpoint(checkpoint, channels, next_step),
{
**checkpoint_metadata,
"source": "input",
"step": next_step,
"parents": saved.metadata.get("parents", {})
if saved
else {},
},
get_new_channel_versions(
checkpoint_previous_versions,
@@ -1576,7 +1579,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
saved.parent_config or saved.config if saved else checkpoint_config,
next_checkpoint,
{
**checkpoint_metadata,
"source": "fork",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1738,7 +1740,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
checkpoint,
{
**checkpoint_metadata,
"source": "update",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1832,10 +1833,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
)
},
)
checkpoint_metadata = config["metadata"]
if saved:
checkpoint_config = patch_configurable(config, saved.config[CONF])
checkpoint_metadata = {**saved.metadata, **checkpoint_metadata}
channels, managed = channels_from_checkpoint(
self.channels,
checkpoint,
@@ -1898,7 +1897,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
create_checkpoint(checkpoint, None, step),
{
**checkpoint_metadata,
"source": "update",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1921,7 +1919,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
next_checkpoint,
{
**checkpoint_metadata,
"source": "update",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -1958,9 +1955,11 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
create_checkpoint(checkpoint, channels, next_step),
{
**checkpoint_metadata,
"source": "input",
"step": next_step,
"parents": saved.metadata.get("parents", {})
if saved
else {},
},
get_new_channel_versions(
checkpoint_previous_versions,
@@ -1996,7 +1995,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
saved.parent_config or saved.config if saved else checkpoint_config,
next_checkpoint,
{
**checkpoint_metadata,
"source": "fork",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -2156,7 +2154,6 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
checkpoint_config,
checkpoint,
{
**checkpoint_metadata,
"source": "update",
"step": step + 1,
"parents": saved.metadata.get("parents", {}) if saved else {},
@@ -2298,7 +2295,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
- `"custom"`: Emit custom data from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
Will be emitted as 2-tuples `(LLM token, metadata)`.
- `"debug"`: Emit debug events with as much information as possible for each step.
- `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state().
- `"tasks"`: Emit events when tasks start and finish, including their results and errors.
You can pass a list as the `stream_mode` parameter to stream multiple modes at once.
The streamed outputs will be tuples of `(mode, data)`.
@@ -2414,7 +2412,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
debug=debug,
checkpoint_during=checkpoint_during
if checkpoint_during is not None
else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, False),
else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True),
trigger_to_nodes=self.trigger_to_nodes,
migrate_checkpoint=self._migrate_checkpoint,
retry_policy=self.retry_policy,
@@ -2658,7 +2656,7 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
debug=debug,
checkpoint_during=checkpoint_during
if checkpoint_during is not None
else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, False),
else config[CONF].get(CONFIG_KEY_CHECKPOINT_DURING, True),
trigger_to_nodes=self.trigger_to_nodes,
migrate_checkpoint=self._migrate_checkpoint,
retry_policy=self.retry_policy,
+4 -3
View File
@@ -83,7 +83,7 @@ from langgraph.types import (
)
from langgraph.utils.config import merge_configs, patch_config
GetNextVersion = Callable[[Optional[V]], V]
GetNextVersion = Callable[[Optional[V], None], V]
SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)
@@ -214,7 +214,7 @@ def local_read(
return values
def increment(current: int | None) -> int:
def increment(current: int | None, channel: None) -> int:
"""Default channel versioning function, increments the current int version."""
return current + 1 if current is not None else 1
@@ -265,7 +265,8 @@ def apply_writes(
next_version = get_next_version(
max(checkpoint["channel_versions"].values())
if checkpoint["channel_versions"]
else None
else None,
None,
)
# Consume all channels that were read
@@ -71,3 +71,14 @@ def channels_from_checkpoint(
},
managed_specs,
)
def copy_checkpoint(checkpoint: Checkpoint) -> Checkpoint:
return Checkpoint(
v=checkpoint["v"],
ts=checkpoint["ts"],
id=checkpoint["id"],
channel_values=checkpoint["channel_values"].copy(),
channel_versions=checkpoint["channel_versions"].copy(),
versions_seen={k: v.copy() for k, v in checkpoint["versions_seen"].items()},
)
+48 -101
View File
@@ -3,13 +3,8 @@ from __future__ import annotations
from collections import defaultdict
from collections.abc import Iterable, Iterator, Mapping, Sequence
from dataclasses import asdict
from datetime import datetime, timezone
from pprint import pformat
from typing import (
Any,
Literal,
Union,
)
from typing import Any
from uuid import UUID
from langchain_core.runnables.config import RunnableConfig
@@ -17,7 +12,7 @@ from langchain_core.utils.input import get_bolded_text, get_colored_text
from typing_extensions import TypedDict
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import Checkpoint, CheckpointMetadata, PendingWrite
from langgraph.checkpoint.base import CheckpointMetadata, PendingWrite
from langgraph.constants import (
CONF,
CONFIG_KEY_CHECKPOINT_NS,
@@ -66,82 +61,43 @@ class CheckpointPayload(TypedDict):
tasks: list[CheckpointTask]
class DebugOutputBase(TypedDict):
timestamp: str
step: int
class DebugOutputTask(DebugOutputBase):
type: Literal["task"]
payload: TaskPayload
class DebugOutputTaskResult(DebugOutputBase):
type: Literal["task_result"]
payload: TaskResultPayload
class DebugOutputCheckpoint(DebugOutputBase):
type: Literal["checkpoint"]
payload: CheckpointPayload
DebugOutput = Union[DebugOutputTask, DebugOutputTaskResult, DebugOutputCheckpoint]
TASK_NAMESPACE = UUID("6ba7b831-9dad-11d1-80b4-00c04fd430c8")
def map_debug_tasks(
step: int, tasks: Iterable[PregelExecutableTask]
) -> Iterator[DebugOutputTask]:
def map_debug_tasks(tasks: Iterable[PregelExecutableTask]) -> Iterator[TaskPayload]:
"""Produce "task" events for stream_mode=debug."""
ts = datetime.now(timezone.utc).isoformat()
for task in tasks:
if task.config is not None and TAG_HIDDEN in task.config.get("tags", []):
continue
yield {
"type": "task",
"timestamp": ts,
"step": step,
"payload": {
"id": task.id,
"name": task.name,
"input": task.input,
"triggers": task.triggers,
},
"id": task.id,
"name": task.name,
"input": task.input,
"triggers": task.triggers,
}
def map_debug_task_results(
step: int,
task_tup: tuple[PregelExecutableTask, Sequence[tuple[str, Any]]],
stream_keys: str | Sequence[str],
) -> Iterator[DebugOutputTaskResult]:
) -> Iterator[TaskResultPayload]:
"""Produce "task_result" events for stream_mode=debug."""
stream_channels_list = (
[stream_keys] if isinstance(stream_keys, str) else stream_keys
)
task, writes = task_tup
yield {
"type": "task_result",
"timestamp": datetime.now(timezone.utc).isoformat(),
"step": step,
"payload": {
"id": task.id,
"name": task.name,
"error": next((w[1] for w in writes if w[0] == ERROR), None),
"result": [
w for w in writes if w[0] in stream_channels_list or w[0] == RETURN
],
"interrupts": [
asdict(v)
for w in writes
if w[0] == INTERRUPT
for v in (w[1] if isinstance(w[1], Sequence) else [w[1]])
],
},
"id": task.id,
"name": task.name,
"error": next((w[1] for w in writes if w[0] == ERROR), None),
"result": [w for w in writes if w[0] in stream_channels_list or w[0] == RETURN],
"interrupts": [
asdict(v)
for w in writes
if w[0] == INTERRUPT
for v in (w[1] if isinstance(w[1], Sequence) else [w[1]])
],
}
@@ -159,17 +115,15 @@ def rm_pregel_keys(config: RunnableConfig | None) -> RunnableConfig | None:
def map_debug_checkpoint(
step: int,
config: RunnableConfig,
channels: Mapping[str, BaseChannel],
stream_channels: str | Sequence[str],
metadata: CheckpointMetadata,
checkpoint: Checkpoint,
tasks: Iterable[PregelExecutableTask],
pending_writes: list[PendingWrite],
parent_config: RunnableConfig | None,
output_keys: str | Sequence[str],
) -> Iterator[DebugOutputCheckpoint]:
) -> Iterator[CheckpointPayload]:
"""Produce "checkpoint" events for stream_mode=debug."""
parent_ns = config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
@@ -193,42 +147,35 @@ def map_debug_checkpoint(
}
yield {
"type": "checkpoint",
"timestamp": checkpoint["ts"],
"step": step,
"payload": {
"config": rm_pregel_keys(patch_checkpoint_map(config, metadata)),
"parent_config": rm_pregel_keys(
patch_checkpoint_map(parent_config, metadata)
),
"values": read_channels(channels, stream_channels),
"metadata": metadata,
"next": [t.name for t in tasks],
"tasks": [
{
"id": t.id,
"name": t.name,
"error": t.error,
"state": t.state,
}
if t.error
else {
"id": t.id,
"name": t.name,
"result": t.result,
"interrupts": tuple(asdict(i) for i in t.interrupts),
"state": t.state,
}
if t.result
else {
"id": t.id,
"name": t.name,
"interrupts": tuple(asdict(i) for i in t.interrupts),
"state": t.state,
}
for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys)
],
},
"config": rm_pregel_keys(patch_checkpoint_map(config, metadata)),
"parent_config": rm_pregel_keys(patch_checkpoint_map(parent_config, metadata)),
"values": read_channels(channels, stream_channels),
"metadata": metadata,
"next": [t.name for t in tasks],
"tasks": [
{
"id": t.id,
"name": t.name,
"error": t.error,
"state": t.state,
}
if t.error
else {
"id": t.id,
"name": t.name,
"result": t.result,
"interrupts": tuple(asdict(i) for i in t.interrupts),
"state": t.state,
}
if t.result
else {
"id": t.id,
"name": t.name,
"interrupts": tuple(asdict(i) for i in t.interrupts),
"state": t.state,
}
for t in tasks_w_writes(tasks, pending_writes, task_states, output_keys)
],
}
+50 -23
View File
@@ -11,6 +11,7 @@ from contextlib import (
AsyncExitStack,
ExitStack,
)
from datetime import datetime, timezone
from inspect import signature
from types import TracebackType
from typing import (
@@ -29,7 +30,6 @@ from typing_extensions import ParamSpec, Self
from langgraph.cache.base import BaseCache
from langgraph.channels.base import BaseChannel
from langgraph.checkpoint.base import (
EXCLUDED_METADATA_KEYS,
WRITES_IDX_MAP,
BaseCheckpointSaver,
ChannelVersions,
@@ -37,7 +37,6 @@ from langgraph.checkpoint.base import (
CheckpointMetadata,
CheckpointTuple,
PendingWrite,
copy_checkpoint,
)
from langgraph.constants import (
CONF,
@@ -84,6 +83,7 @@ from langgraph.pregel.algo import (
)
from langgraph.pregel.checkpoint import (
channels_from_checkpoint,
copy_checkpoint,
create_checkpoint,
empty_checkpoint,
)
@@ -118,6 +118,7 @@ from langgraph.types import (
PregelScratchpad,
RetryPolicy,
StreamChunk,
StreamMode,
StreamProtocol,
)
from langgraph.utils.config import patch_configurable
@@ -312,10 +313,22 @@ class PregelLoop:
# deduplicate writes to special channels, last write wins
if all(w[0] in WRITES_IDX_MAP for w in writes):
writes = list({w[0]: w for w in writes}.values())
# remove existing writes for this task
self.checkpoint_pending_writes = [
w for w in self.checkpoint_pending_writes if w[0] != task_id
]
if task_id == NULL_TASK_ID:
# writes for the null task are accumulated
self.checkpoint_pending_writes = [
w
for w in self.checkpoint_pending_writes
if w[0] != task_id or w[1] not in WRITES_IDX_MAP
]
writes_to_save: WritesT = [
w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
] + list(writes)
else:
# remove existing writes for this task
self.checkpoint_pending_writes = [
w for w in self.checkpoint_pending_writes if w[0] != task_id
]
writes_to_save = writes
# save writes
self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
if self.checkpoint_during and self.checkpointer_put_writes is not None:
@@ -336,7 +349,7 @@ class PregelLoop:
self.submit(
self.checkpointer_put_writes,
config,
writes,
writes_to_save,
task_id,
task_path_str(task.path) if task else "",
)
@@ -344,7 +357,7 @@ class PregelLoop:
self.submit(
self.checkpointer_put_writes,
config,
writes,
writes_to_save,
task_id,
)
# output writes
@@ -421,7 +434,7 @@ class PregelLoop:
),
):
# produce debug output
self._emit("debug", map_debug_tasks, self.step, [pushed])
self._emit("tasks", map_debug_tasks, [pushed])
# debug flag
if self.debug:
print_step_tasks(self.step, [pushed])
@@ -471,9 +484,8 @@ class PregelLoop:
# produce debug output
if self._checkpointer_put_after_previous is not None:
self._emit(
"debug",
"checkpoints",
map_debug_checkpoint,
self.step - 1, # printing checkpoint for previous step
{
**self.checkpoint_config,
CONF: {
@@ -484,7 +496,6 @@ class PregelLoop:
self.channels,
self.stream_keys,
self.checkpoint_metadata,
self.checkpoint,
self.tasks.values(),
self.checkpoint_pending_writes,
self.prev_checkpoint_config,
@@ -508,7 +519,7 @@ class PregelLoop:
raise GraphInterrupt()
# produce debug output
self._emit("debug", map_debug_tasks, self.step, self.tasks.values())
self._emit("tasks", map_debug_tasks, self.tasks.values())
# debug flag
if self.debug:
@@ -721,11 +732,6 @@ class PregelLoop:
)
# bail if no checkpointer
if do_checkpoint and self._checkpointer_put_after_previous is not None:
for k, v in self.config["metadata"].items():
if k in EXCLUDED_METADATA_KEYS:
continue
metadata.setdefault(k, v) # type: ignore
self.prev_checkpoint_config = (
self.checkpoint_config
if CONFIG_KEY_CHECKPOINT_ID in self.checkpoint_config[CONF]
@@ -833,17 +839,39 @@ class PregelLoop:
def _emit(
self,
mode: str,
mode: StreamMode,
values: Callable[P, Iterator[Any]],
*args: P.args,
**kwargs: P.kwargs,
) -> None:
if self.stream is None:
return
if mode not in self.stream.modes:
debug_remap = mode in ("checkpoints", "tasks") and "debug" in self.stream.modes
if mode not in self.stream.modes and not debug_remap:
return
for v in values(*args, **kwargs):
self.stream((self.checkpoint_ns, mode, v))
if mode in self.stream.modes:
self.stream((self.checkpoint_ns, mode, v))
# "debug" mode is "checkpoints" or "tasks" with a wrapper dict
if debug_remap:
self.stream(
(
self.checkpoint_ns,
"debug",
{
"step": self.step - 1
if mode == "checkpoints"
else self.step,
"timestamp": datetime.now(timezone.utc).isoformat(),
"type": "checkpoint"
if mode == "checkpoints"
else "task_result"
if "result" in v
else "task",
"payload": v,
},
)
)
def output_writes(
self, task_id: str, writes: WritesT, *, cached: bool = False
@@ -884,9 +912,8 @@ class PregelLoop:
)
if not cached:
self._emit(
"debug",
"tasks",
map_debug_task_results,
self.step,
(task, writes),
self.stream_keys,
)
+2 -1
View File
@@ -104,7 +104,8 @@ class FuturesDict(Generic[F, E], dict[F, Optional[PregelExecutableTask]]):
fut: F,
) -> None:
try:
self.callback()(task, _exception(fut)) # type: ignore[misc]
if cb := self.callback():
cb(task, _exception(fut))
finally:
with self.lock:
self.done.add(fut)
+6 -2
View File
@@ -46,7 +46,9 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver]
- False disables checkpointing, even if the parent graph has a checkpointer.
- None inherits checkpointer from the parent graph."""
StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
StreamMode = Literal[
"values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"
]
"""How the stream method should emit outputs.
- `"values"`: Emit all values in the state after each step, including interrupts.
@@ -55,7 +57,9 @@ StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
- `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`.
- `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
- `"debug"`: Emit debug events with as much information as possible for each step.
- `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state().
- `"tasks"`: Emit events when tasks start and finish, including their results and errors.
- `"debug"`: Emit "checlkpoints" and "tasks" events, for debugging purposes.
"""
StreamWriter = Callable[[Any], None]
+3 -9
View File
@@ -1,7 +1,5 @@
from __future__ import annotations
from typing import Union
from typing_extensions import TypeVar
from langgraph._typing import StateLike
@@ -19,12 +17,8 @@ InputT = TypeVar("InputT", bound=StateLike, default=StateT)
Defaults to `StateT`.
"""
ResolvedInputT = TypeVar("ResolvedInputT", bound=StateLike)
"""Type variable used to represent the resolved input to a state graph.
OutputT = TypeVar("OutputT", bound=StateLike, default=StateT)
"""Type variable used to represent the output of a state graph.
No default.
Defaults to `StateT`.
"""
OutputT = TypeVar("OutputT", bound=Union[StateLike, None], default=StateT)
"""Type variable used to represent the output of a state graph."""
+3 -3
View File
@@ -41,8 +41,8 @@ class LangGraphDeprecationWarning(DeprecationWarning):
return message
class LangGraphDeprecatedSinceV10(LangGraphDeprecationWarning):
"""A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v1.0.0"""
class LangGraphDeprecatedSinceV05(LangGraphDeprecationWarning):
"""A specific `LangGraphDeprecationWarning` subclass defining functionality deprecated since LangGraph v0.5.0"""
def __init__(self, message: str, *args: object) -> None:
super().__init__(message, *args, since=(1, 0), expected_removal=(2, 0))
super().__init__(message, *args, since=(0, 5), expected_removal=(2, 0))
+3 -3
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph"
version = "0.4.7"
version = "0.5.0rc1"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
requires-python = ">=3.9"
@@ -13,9 +13,9 @@ license = "MIT"
license-files = ['LICENSE']
dependencies = [
"langchain-core>=0.1",
"langgraph-checkpoint>=2.0.26",
"langgraph-checkpoint>=2.1.0",
"langgraph-sdk>=0.1.42",
"langgraph-prebuilt>=0.2.0",
"langgraph-prebuilt>=0.5.0rc0",
"xxhash>=3.5.0",
"pydantic>=2.7.4",
]
+5
View File
@@ -12,6 +12,7 @@ from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.store.base import BaseStore
from tests.conftest_checkpointer import (
_checkpointer_memory,
_checkpointer_memory_migrate_sends,
_checkpointer_postgres,
_checkpointer_postgres_aio,
_checkpointer_postgres_aio_pipe,
@@ -125,6 +126,7 @@ async def async_store(request: pytest.FixtureRequest) -> AsyncIterator[BaseStore
if NO_DOCKER
else [
"memory",
"memory_migrate_sends",
"sqlite",
"sqlite_aes",
"postgres",
@@ -139,6 +141,9 @@ def sync_checkpointer(
if checkpointer_name == "memory":
with _checkpointer_memory() as checkpointer:
yield checkpointer
elif checkpointer_name == "memory_migrate_sends":
with _checkpointer_memory_migrate_sends() as checkpointer:
yield checkpointer
elif checkpointer_name == "sqlite":
with _checkpointer_sqlite() as checkpointer:
yield checkpointer
+10 -1
View File
@@ -14,7 +14,10 @@ from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
pytest.register_assert_rewrite("tests.memory_assert")
from tests.memory_assert import MemorySaverAssertImmutable # noqa: E402
from tests.memory_assert import ( # noqa: E402
MemorySaverAssertImmutable,
MemorySaverNeedsPendingSendsMigration,
)
DEFAULT_POSTGRES_URI = "postgres://postgres:postgres@localhost:5442/"
@@ -24,6 +27,11 @@ def _checkpointer_memory():
yield MemorySaverAssertImmutable()
@contextmanager
def _checkpointer_memory_migrate_sends():
yield MemorySaverNeedsPendingSendsMigration()
@contextmanager
def _checkpointer_sqlite():
with SqliteSaver.from_conn_string(":memory:") as checkpointer:
@@ -187,6 +195,7 @@ async def _checkpointer_postgres_aio_pool():
__all__ = [
"_checkpointer_memory",
"_checkpointer_memory_migrate_sends",
"_checkpointer_sqlite",
"_checkpointer_sqlite_aes",
"_checkpointer_postgres",
+24
View File
@@ -7,6 +7,7 @@ from typing import Any, Optional
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
ChannelVersions,
Checkpoint,
CheckpointMetadata,
@@ -14,6 +15,7 @@ from langgraph.checkpoint.base import (
SerializerProtocol,
)
from langgraph.checkpoint.memory import InMemorySaver, PersistentDict
from langgraph.constants import TASKS
class NoopSerializer(SerializerProtocol):
@@ -24,6 +26,28 @@ class NoopSerializer(SerializerProtocol):
return "type", obj
class MemorySaverNeedsPendingSendsMigration(BaseCheckpointSaver):
def __init__(self) -> None:
self.saver = InMemorySaver()
def __getattribute__(self, name):
if name in ("saver", "__class__", "get_tuple"):
return object.__getattribute__(self, name)
return getattr(self.saver, name)
def get_tuple(self, config):
if tup := self.saver.get_tuple(config):
if tup.checkpoint["v"] == 4 and tup.checkpoint["channel_values"].get(TASKS):
tup.checkpoint["v"] = 3
tup.checkpoint["pending_sends"] = tup.checkpoint["channel_values"].pop(
TASKS
)
tup.checkpoint["channel_versions"].pop(TASKS)
for seen in tup.checkpoint["versions_seen"].values():
seen.pop(TASKS, None)
return tup
class MemorySaverAssertImmutable(InMemorySaver):
storage_for_copies: defaultdict[str, dict[str, dict[str, Checkpoint]]]
@@ -7,12 +7,9 @@ from typing import Annotated, Literal, Optional, Union
import pytest
from typing_extensions import TypedDict
from langgraph.checkpoint.base import (
BaseCheckpointSaver,
CheckpointTuple,
copy_checkpoint,
)
from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple
from langgraph.graph.state import StateGraph
from langgraph.pregel.checkpoint import copy_checkpoint
from langgraph.types import Command, Interrupt, PregelTask, StateSnapshot, interrupt
from langgraph.utils.config import patch_configurable
from tests.any_int import AnyInt
@@ -46,7 +43,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
"source": "loop",
"step": 4,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config={
@@ -76,7 +72,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
"source": "loop",
"step": 3,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config={
@@ -134,7 +129,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
"source": "loop",
"step": 2,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config={
@@ -171,7 +165,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
"source": "loop",
"step": 1,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config={
@@ -221,7 +214,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
"source": "loop",
"step": 0,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config={
@@ -260,7 +252,6 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]:
"source": "input",
"step": -1,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -346,7 +337,6 @@ SAVED_CHECKPOINTS = {
"source": "loop",
"step": 4,
"parents": {},
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -407,7 +397,6 @@ SAVED_CHECKPOINTS = {
"source": "loop",
"step": 3,
"parents": {},
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -483,7 +472,6 @@ SAVED_CHECKPOINTS = {
"source": "loop",
"step": 2,
"parents": {},
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -535,7 +523,6 @@ SAVED_CHECKPOINTS = {
"source": "loop",
"step": 1,
"parents": {},
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -590,7 +577,6 @@ SAVED_CHECKPOINTS = {
"source": "loop",
"step": 0,
"parents": {},
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -639,7 +625,6 @@ SAVED_CHECKPOINTS = {
"source": "input",
"step": -1,
"parents": {},
"thread_id": "1",
},
parent_config=None,
pending_writes=[
@@ -723,7 +708,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 4,
"parents": {},
},
@@ -785,7 +769,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 3,
"parents": {},
},
@@ -864,7 +847,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 2,
"parents": {},
},
@@ -920,7 +902,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 1,
"parents": {},
},
@@ -980,7 +961,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 0,
"parents": {},
},
@@ -1029,7 +1009,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "input",
"thread_id": "1",
"step": -1,
"parents": {},
},
@@ -1115,7 +1094,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 4,
"parents": {},
},
@@ -1177,7 +1155,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 3,
"parents": {},
},
@@ -1256,7 +1233,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 2,
"parents": {},
},
@@ -1312,7 +1288,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 1,
"parents": {},
},
@@ -1372,7 +1347,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 0,
"parents": {},
},
@@ -1421,7 +1395,6 @@ SAVED_CHECKPOINTS = {
},
metadata={
"source": "input",
"thread_id": "1",
"step": -1,
"parents": {},
},
+7 -7
View File
@@ -4,7 +4,7 @@ from typing_extensions import TypedDict
from langgraph.func import entrypoint, task
from langgraph.graph import StateGraph
from langgraph.types import RetryPolicy
from langgraph.warnings import LangGraphDeprecatedSinceV10
from langgraph.warnings import LangGraphDeprecatedSinceV05
class PlainState(TypedDict): ...
@@ -14,7 +14,7 @@ def test_add_node_retry_arg() -> None:
builder = StateGraph(PlainState)
with pytest.warns(
LangGraphDeprecatedSinceV10,
LangGraphDeprecatedSinceV05,
match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
):
builder.add_node("test_node", lambda state: state, retry=RetryPolicy()) # type: ignore[arg-type]
@@ -22,7 +22,7 @@ def test_add_node_retry_arg() -> None:
def test_task_retry_arg() -> None:
with pytest.warns(
LangGraphDeprecatedSinceV10,
LangGraphDeprecatedSinceV05,
match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
):
@@ -33,7 +33,7 @@ def test_task_retry_arg() -> None:
def test_entrypoint_retry_arg() -> None:
with pytest.warns(
LangGraphDeprecatedSinceV10,
LangGraphDeprecatedSinceV05,
match="`retry` is deprecated and will be removed. Please use `retry_policy` instead.",
):
@@ -44,7 +44,7 @@ def test_entrypoint_retry_arg() -> None:
def test_state_graph_input_schema() -> None:
with pytest.warns(
LangGraphDeprecatedSinceV10,
LangGraphDeprecatedSinceV05,
match="`input` is deprecated and will be removed. Please use `input_schema` instead.",
):
StateGraph(PlainState, input=PlainState) # type: ignore[arg-type]
@@ -52,7 +52,7 @@ def test_state_graph_input_schema() -> None:
def test_state_graph_output_schema() -> None:
with pytest.warns(
LangGraphDeprecatedSinceV10,
LangGraphDeprecatedSinceV05,
match="`output` is deprecated and will be removed. Please use `output_schema` instead.",
):
StateGraph(PlainState, output=PlainState) # type: ignore[arg-type]
@@ -62,7 +62,7 @@ def test_add_node_input_schema() -> None:
builder = StateGraph(PlainState)
with pytest.warns(
LangGraphDeprecatedSinceV10,
LangGraphDeprecatedSinceV05,
match="`input` is deprecated and will be removed. Please use `input_schema` instead.",
):
builder.add_node("test_node", lambda state: state, input=PlainState) # type: ignore[arg-type]
File diff suppressed because it is too large Load Diff
+443 -131
View File
@@ -16,13 +16,14 @@ from langchain_core.runnables import RunnableConfig, RunnablePick
from pytest_mock import MockerFixture
from typing_extensions import TypedDict
from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue
from langgraph.channels.untracked_value import UntrackedValue
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.constants import END, PULL, PUSH, START
from langgraph.graph.message import add_messages
from langgraph.graph.message import MessageGraph, add_messages
from langgraph.graph.state import StateGraph
from langgraph.prebuilt.chat_agent_executor import create_react_agent
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.pregel import NodeBuilder, Pregel
from langgraph.types import PregelTask, Send, StateSnapshot, StreamWriter
from tests.any_int import AnyInt
@@ -118,7 +119,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "loop",
"step": 6,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[1].config,
@@ -139,7 +139,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "loop",
"step": 5,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[2].config,
@@ -160,7 +159,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "input",
"step": 4,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[3].config,
@@ -181,7 +179,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "loop",
"step": 3,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[4].config,
@@ -202,7 +199,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "input",
"step": 2,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[5].config,
@@ -223,7 +219,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[6].config,
@@ -244,7 +239,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[7].config,
@@ -265,7 +259,6 @@ async def test_invoke_two_processes_in_out_interrupt(
"parents": {},
"source": "input",
"step": -1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -341,7 +334,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "loop",
"step": 5,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[1].config,
@@ -362,7 +354,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "loop",
"step": 4,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[2].config,
@@ -383,7 +374,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "loop",
"step": 3,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[3].config,
@@ -404,7 +394,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "loop",
"step": 2,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[4].config,
@@ -425,7 +414,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[5].config,
@@ -446,7 +434,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=history[6].config,
@@ -467,7 +454,6 @@ async def test_fork_always_re_runs_nodes(
"parents": {},
"source": "input",
"step": -1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -499,7 +485,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
from langchain_core.tools import tool
class AgentState(TypedDict):
input: Annotated[str, EphemeralValue]
input: Annotated[str, UntrackedValue]
agent_outcome: Optional[Union[AgentAction, AgentFinish]]
intermediate_steps: Annotated[list[tuple[AgentAction, str]], operator.add]
@@ -574,6 +560,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
app = workflow.compile()
assert await app.ainvoke({"input": "what is weather in sf"}) == {
"input": "what is weather in sf",
"intermediate_steps": [
[
AgentAction(
@@ -696,7 +683,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
assert [
c
async for c in app_w_interrupt.astream(
{"input": "what is weather in sf"}, config
{"input": "what is weather in sf"}, config, checkpoint_during=False
)
] == [
{
@@ -734,7 +721,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
parent_config=None,
interrupts=(),
@@ -774,7 +760,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
"parents": {},
"source": "update",
"step": 2,
"thread_id": "1",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -852,7 +837,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
"parents": {},
"source": "update",
"step": 5,
"thread_id": "1",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -874,7 +858,7 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
assert [
c
async for c in app_w_interrupt.astream(
{"input": "what is weather in sf"}, config
{"input": "what is weather in sf"}, config, checkpoint_during=False
)
] == [
{
@@ -910,7 +894,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "2",
},
parent_config=None,
interrupts=(),
@@ -950,7 +933,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
"parents": {},
"source": "update",
"step": 2,
"thread_id": "2",
},
parent_config=[
c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)
@@ -1026,7 +1008,6 @@ async def test_conditional_graph_state(async_checkpointer: BaseCheckpointSaver)
"parents": {},
"source": "update",
"step": 5,
"thread_id": "2",
},
parent_config=[
c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)
@@ -1593,7 +1574,9 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
assert [
c
async for c in app_w_interrupt.astream(
{"messages": HumanMessage(content="what is weather in sf")}, config
{"messages": HumanMessage(content="what is weather in sf")},
config,
checkpoint_during=False,
)
] == [
{
@@ -1645,7 +1628,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
parent_config=None,
interrupts=(),
@@ -1683,7 +1665,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "update",
"step": 2,
"thread_id": "1",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -1776,7 +1757,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "loop",
"step": 4,
"thread_id": "1",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -1824,7 +1804,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "update",
"step": 5,
"thread_id": "1",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -1846,7 +1825,9 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
assert [
c
async for c in app_w_interrupt.astream(
{"messages": HumanMessage(content="what is weather in sf")}, config
{"messages": HumanMessage(content="what is weather in sf")},
config,
checkpoint_during=False,
)
] == [
{
@@ -1892,7 +1873,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "2",
},
parent_config=None,
interrupts=(),
@@ -1930,7 +1910,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "update",
"step": 2,
"thread_id": "2",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -2023,7 +2002,6 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "loop",
"step": 4,
"thread_id": "2",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -2071,7 +2049,411 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
"parents": {},
"source": "update",
"step": 5,
"thread_id": "2",
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
-1
].config
),
interrupts=(),
)
async def test_message_graph(async_checkpointer: BaseCheckpointSaver) -> None:
from langchain_core.language_models.fake_chat_models import (
FakeMessagesListChatModel,
)
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.tools import tool
class FakeFuntionChatModel(FakeMessagesListChatModel):
def bind_functions(self, functions: list):
return self
@tool()
def search_api(query: str) -> str:
"""Searches the API for the query."""
return f"result for {query}"
tools = [search_api]
model = FakeFuntionChatModel(
responses=[
AIMessage(
content="",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "query"},
}
],
id="ai1",
),
AIMessage(
content="",
tool_calls=[
{
"id": "tool_call456",
"name": "search_api",
"args": {"query": "another"},
}
],
id="ai2",
),
AIMessage(content="answer", id="ai3"),
]
)
# Define the function that determines whether to continue or not
def should_continue(messages):
last_message = messages[-1]
# If there is no function call, then we finish
if not last_message.tool_calls:
return "end"
# Otherwise if there is, we continue
else:
return "continue"
# Define a new graph
workflow = MessageGraph()
# Define the two nodes we will cycle between
workflow.add_node("agent", model)
workflow.add_node("tools", ToolNode(tools))
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point("agent")
# We now add a conditional edge
workflow.add_conditional_edges(
# First, we define the start node. We use `agent`.
# This means these are the edges taken after the `agent` node is called.
"agent",
# Next, we pass in the function that will determine which node is called next.
should_continue,
# Finally we pass in a mapping.
# The keys are strings, and the values are other nodes.
# END is a special node marking that the graph should finish.
# What will happen is we will call `should_continue`, and then the output of that
# will be matched against the keys in this mapping.
# Based on which one it matches, that node will then be called.
{
# If `tools`, then we call the tool node.
"continue": "tools",
# Otherwise we finish.
"end": END,
},
)
# We now add a normal edge from `tools` to `agent`.
# This means that after `tools` is called, `agent` node is called next.
workflow.add_edge("tools", "agent")
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
app = workflow.compile()
assert await app.ainvoke(HumanMessage(content="what is weather in sf")) == [
_AnyIdHumanMessage(
content="what is weather in sf",
),
AIMessage(
content="",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "query"},
}
],
id="ai1", # respects ids passed in
),
_AnyIdToolMessage(
content="result for query",
name="search_api",
tool_call_id="tool_call123",
),
AIMessage(
content="",
tool_calls=[
{
"id": "tool_call456",
"name": "search_api",
"args": {"query": "another"},
}
],
id="ai2",
),
_AnyIdToolMessage(
content="result for another",
name="search_api",
tool_call_id="tool_call456",
),
AIMessage(content="answer", id="ai3"),
]
assert [
c async for c in app.astream([HumanMessage(content="what is weather in sf")])
] == [
{
"agent": AIMessage(
content="",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "query"},
}
],
id="ai1",
)
},
{
"tools": [
_AnyIdToolMessage(
content="result for query",
name="search_api",
tool_call_id="tool_call123",
)
]
},
{
"agent": AIMessage(
content="",
tool_calls=[
{
"id": "tool_call456",
"name": "search_api",
"args": {"query": "another"},
}
],
id="ai2",
)
},
{
"tools": [
_AnyIdToolMessage(
content="result for another",
name="search_api",
tool_call_id="tool_call456",
)
]
},
{"agent": AIMessage(content="answer", id="ai3")},
]
app_w_interrupt = workflow.compile(
checkpointer=async_checkpointer,
interrupt_after=["agent"],
)
config = {"configurable": {"thread_id": "1"}}
assert [
c
async for c in app_w_interrupt.astream(
HumanMessage(content="what is weather in sf"),
config,
checkpoint_during=False,
)
] == [
{
"agent": AIMessage(
content="",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "query"},
}
],
id="ai1",
)
},
{"__interrupt__": ()},
]
tup = await app_w_interrupt.checkpointer.aget_tuple(config)
assert await app_w_interrupt.aget_state(config) == StateSnapshot(
values=[
_AnyIdHumanMessage(content="what is weather in sf"),
AIMessage(
content="",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "query"},
}
],
id="ai1",
),
],
tasks=(PregelTask(AnyStr(), "tools", (PULL, "tools")),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
metadata={
"parents": {},
"source": "loop",
"step": 1,
},
parent_config=None,
interrupts=(),
)
# modify ai message
last_message = (await app_w_interrupt.aget_state(config)).values[-1]
last_message.tool_calls[0]["args"] = {"query": "a different query"}
await app_w_interrupt.aupdate_state(config, last_message)
# message was replaced instead of appended
tup = await app_w_interrupt.checkpointer.aget_tuple(config)
assert await app_w_interrupt.aget_state(config) == StateSnapshot(
values=[
_AnyIdHumanMessage(content="what is weather in sf"),
AIMessage(
content="",
id="ai1",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "a different query"},
}
],
),
],
tasks=(PregelTask(AnyStr(), "tools", (PULL, "tools")),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
metadata={
"parents": {},
"source": "update",
"step": 2,
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
-1
].config
),
interrupts=(),
)
assert [c async for c in app_w_interrupt.astream(None, config)] == [
{
"tools": [
_AnyIdToolMessage(
content="result for a different query",
name="search_api",
tool_call_id="tool_call123",
)
]
},
{
"agent": AIMessage(
content="",
tool_calls=[
{
"id": "tool_call456",
"name": "search_api",
"args": {"query": "another"},
}
],
id="ai2",
)
},
{"__interrupt__": ()},
]
tup = await app_w_interrupt.checkpointer.aget_tuple(config)
assert await app_w_interrupt.aget_state(config) == StateSnapshot(
values=[
_AnyIdHumanMessage(content="what is weather in sf"),
AIMessage(
content="",
id="ai1",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "a different query"},
}
],
),
_AnyIdToolMessage(
content="result for a different query",
name="search_api",
tool_call_id="tool_call123",
),
AIMessage(
content="",
tool_calls=[
{
"id": "tool_call456",
"name": "search_api",
"args": {"query": "another"},
}
],
id="ai2",
),
],
tasks=(PregelTask(AnyStr(), "tools", (PULL, "tools")),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
metadata={
"parents": {},
"source": "loop",
"step": 4,
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
-1
].config
),
interrupts=(),
)
await app_w_interrupt.aupdate_state(
config,
AIMessage(content="answer", id="ai2"),
)
# replaces message even if object identity is different, as long as id is the same
tup = await app_w_interrupt.checkpointer.aget_tuple(config)
assert await app_w_interrupt.aget_state(config) == StateSnapshot(
values=[
_AnyIdHumanMessage(content="what is weather in sf"),
AIMessage(
content="",
id="ai1",
tool_calls=[
{
"id": "tool_call123",
"name": "search_api",
"args": {"query": "a different query"},
}
],
),
_AnyIdToolMessage(
content="result for a different query",
name="search_api",
tool_call_id="tool_call123",
),
AIMessage(content="answer", id="ai2"),
],
tasks=(),
next=(),
config=tup.config,
created_at=tup.checkpoint["ts"],
metadata={
"parents": {},
"source": "update",
"step": 5,
},
parent_config=(
[c async for c in app_w_interrupt.checkpointer.alist(config, limit=2)][
@@ -2083,6 +2465,9 @@ async def test_state_graph_packets(async_checkpointer: BaseCheckpointSaver) -> N
async def test_in_one_fan_out_out_one_graph_state() -> None:
def sorted_add(x: list[str], y: list[str]) -> list[str]:
return sorted(operator.add(x, y))
class State(TypedDict, total=False):
query: str
answer: str
@@ -2354,7 +2739,7 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
app = graph.compile(checkpointer=async_checkpointer)
config = {"configurable": {"thread_id": "1"}}
await app.ainvoke({"my_key": "my value"}, config, debug=True)
await app.ainvoke({"my_key": "my value"}, config, checkpoint_during=False)
# test state w/ nested subgraph state (right after interrupt)
# first get_state without subgraph state
expected = StateSnapshot(
@@ -2379,7 +2764,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -2423,12 +2807,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
},
"source": "loop",
"step": 1,
"thread_id": "1",
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:inner"],
"langgraph_checkpoint_ns": AnyStr("inner:"),
},
created_at=AnyStr(),
parent_config=None,
@@ -2448,7 +2826,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -2482,12 +2859,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
"source": "loop",
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "1",
"langgraph_node": "inner",
"langgraph_path": [PULL, "inner"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:inner"],
"langgraph_checkpoint_ns": AnyStr("inner:"),
},
created_at=AnyStr(),
parent_config=None,
@@ -2499,7 +2870,7 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
assert child_history == expected_child_history
# resume
await app.ainvoke(None, config, debug=True)
await app.ainvoke(None, config, checkpoint_during=False)
# test state w/ nested subgraph state (after resuming from interrupt)
assert await app.aget_state(config) == StateSnapshot(
values={"my_key": "hi my value here and there and back again"},
@@ -2516,7 +2887,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
"parents": {},
"source": "loop",
"step": 3,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
@@ -2548,7 +2918,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
"parents": {},
"source": "loop",
"step": 3,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
@@ -2590,7 +2959,6 @@ async def test_nested_graph_state(async_checkpointer: BaseCheckpointSaver) -> No
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -2659,7 +3027,10 @@ async def test_doubly_nested_graph_state(
# test invoke w/ nested interrupt
config = {"configurable": {"thread_id": "1"}}
assert [
c async for c in app.astream({"my_key": "my value"}, config, subgraphs=True)
c
async for c in app.astream(
{"my_key": "my value"}, config, subgraphs=True, checkpoint_during=False
)
] == [
((), {"parent_1": {"my_key": "hi my value"}}),
(
@@ -2697,7 +3068,6 @@ async def test_doubly_nested_graph_state(
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -2734,15 +3104,9 @@ async def test_doubly_nested_graph_state(
}
},
metadata={
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child",
"langgraph_path": ["__pregel_pull", "child"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:child"],
"parents": {"": AnyStr()},
"source": "loop",
"step": 0,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -2782,14 +3146,6 @@ async def test_doubly_nested_graph_state(
),
"source": "loop",
"step": 1,
"thread_id": "1",
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [PULL, AnyStr("child_1")],
"langgraph_step": 1,
"langgraph_triggers": [
"branch:to:child_1",
],
},
created_at=AnyStr(),
parent_config=None,
@@ -2845,17 +3201,6 @@ async def test_doubly_nested_graph_state(
),
"source": "loop",
"step": 1,
"thread_id": "1",
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
PULL,
AnyStr("child_1"),
],
"langgraph_step": 1,
"langgraph_triggers": [
"branch:to:child_1",
],
},
created_at=AnyStr(),
parent_config=None,
@@ -2878,14 +3223,6 @@ async def test_doubly_nested_graph_state(
"parents": {"": AnyStr()},
"source": "loop",
"step": 0,
"thread_id": "1",
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
"langgraph_triggers": [
"branch:to:child",
],
"langgraph_checkpoint_ns": AnyStr("child:"),
},
created_at=AnyStr(),
parent_config=None,
@@ -2905,14 +3242,18 @@ async def test_doubly_nested_graph_state(
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
interrupts=(),
)
# resume
assert [c async for c in app.astream(None, config, subgraphs=True)] == [
assert [
c
async for c in app.astream(
None, config, subgraphs=True, checkpoint_during=False
)
] == [
(
(AnyStr("child:"), AnyStr("child_1:")),
{"grandchild_2": {"my_key": "hi my value here and there"}},
@@ -2943,7 +3284,6 @@ async def test_doubly_nested_graph_state(
"parents": {},
"source": "loop",
"step": 3,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
@@ -2977,7 +3317,6 @@ async def test_doubly_nested_graph_state(
"parents": {},
"source": "loop",
"step": 3,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config={
@@ -3016,7 +3355,6 @@ async def test_doubly_nested_graph_state(
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -3045,12 +3383,6 @@ async def test_doubly_nested_graph_state(
"source": "loop",
"step": 0,
"parents": {"": AnyStr()},
"thread_id": "1",
"langgraph_node": "child",
"langgraph_path": [PULL, AnyStr("child")],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:child"],
"langgraph_checkpoint_ns": AnyStr("child:"),
},
created_at=AnyStr(),
parent_config=None,
@@ -3102,17 +3434,6 @@ async def test_doubly_nested_graph_state(
AnyStr("child:"): AnyStr(),
}
),
"thread_id": "1",
"langgraph_checkpoint_ns": AnyStr("child:"),
"langgraph_node": "child_1",
"langgraph_path": [
PULL,
AnyStr("child_1"),
],
"langgraph_step": 1,
"langgraph_triggers": [
"branch:to:child_1",
],
},
created_at=AnyStr(),
parent_config=None,
@@ -3336,7 +3657,11 @@ async def test_weather_subgraph(
assert [
c
async for c in graph.astream(
inputs, config=config, stream_mode="updates", subgraphs=True
inputs,
config=config,
stream_mode="updates",
subgraphs=True,
checkpoint_during=False,
)
] == [
((), {"router_node": {"route": "weather"}}),
@@ -3363,7 +3688,6 @@ async def test_weather_subgraph(
"source": "loop",
"step": 1,
"parents": {},
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=None,
@@ -3422,7 +3746,11 @@ async def test_weather_subgraph(
assert [
c
async for c in graph.astream(
inputs, config=config, stream_mode="updates", subgraphs=True
inputs,
config=config,
stream_mode="updates",
subgraphs=True,
checkpoint_during=False,
)
] == [
((), {"router_node": {"route": "weather"}}),
@@ -3447,7 +3775,6 @@ async def test_weather_subgraph(
"source": "loop",
"step": 1,
"parents": {},
"thread_id": "14",
},
created_at=AnyStr(),
parent_config=None,
@@ -3481,12 +3808,6 @@ async def test_weather_subgraph(
"source": "loop",
"step": 1,
"parents": {"": AnyStr()},
"thread_id": "14",
"langgraph_node": "weather_graph",
"langgraph_path": [PULL, "weather_graph"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:weather_graph"],
"langgraph_checkpoint_ns": AnyStr("weather_graph:"),
},
created_at=AnyStr(),
parent_config=None,
@@ -3526,7 +3847,6 @@ async def test_weather_subgraph(
"source": "loop",
"step": 1,
"parents": {},
"thread_id": "14",
},
created_at=AnyStr(),
parent_config=None,
@@ -3562,14 +3882,6 @@ async def test_weather_subgraph(
"step": 2,
"source": "update",
"parents": {"": AnyStr()},
"thread_id": "14",
"checkpoint_id": AnyStr(),
"checkpoint_ns": AnyStr("weather_graph:"),
"langgraph_node": "weather_graph",
"langgraph_path": [PULL, "weather_graph"],
"langgraph_step": 2,
"langgraph_triggers": ["branch:to:weather_graph"],
"langgraph_checkpoint_ns": AnyStr("weather_graph:"),
},
created_at=AnyStr(),
parent_config=(
+28 -35
View File
@@ -46,7 +46,7 @@ from langgraph.constants import CONFIG_KEY_NODE_FINISHED, ERROR, PULL, START
from langgraph.errors import InvalidUpdateError, ParentCommand
from langgraph.func import entrypoint, task
from langgraph.graph import END, StateGraph
from langgraph.graph.message import MessagesState, add_messages
from langgraph.graph.message import MessageGraph, MessagesState, add_messages
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.pregel import (
GraphRecursionError,
@@ -159,7 +159,7 @@ def test_checkpoint_errors() -> None:
raise ValueError("Faulty put_writes")
class FaultyVersionCheckpointer(InMemorySaver):
def get_next_version(self, current: Optional[int]) -> int:
def get_next_version(self, current: Optional[int], channel: None) -> int:
raise ValueError("Faulty get_next_version")
def logic(inp: str) -> str:
@@ -904,7 +904,6 @@ def test_pending_writes_resume(
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
}
# get_state with checkpoint_id should not apply any pending writes
state = graph.get_state(state.config)
@@ -994,7 +993,6 @@ def test_pending_writes_resume(
"parents": {},
"step": 1,
"source": "loop",
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -1042,7 +1040,6 @@ def test_pending_writes_resume(
"parents": {},
"step": 0,
"source": "loop",
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -1094,7 +1091,6 @@ def test_pending_writes_resume(
"parents": {},
"step": -1,
"source": "input",
"thread_id": "1",
},
parent_config=None,
pending_writes=UnsortedSequence(
@@ -2058,7 +2054,6 @@ def test_in_one_fan_out_state_graph_waiting_edge(
"parents": {},
"source": "update",
"step": 4,
"thread_id": "2",
},
parent_config=expected_parent_config,
interrupts=(),
@@ -2328,7 +2323,6 @@ def test_in_one_fan_out_state_graph_defer_node(
"parents": {},
"source": "update",
"step": 4,
"thread_id": "2",
},
parent_config=expected_parent_config,
interrupts=(),
@@ -3930,7 +3924,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
# assert that checkpoint metadata contains the run's configurable fields
chkpnt_metadata_1 = sync_checkpointer.get_tuple(config).metadata
assert chkpnt_metadata_1["thread_id"] == "1"
assert chkpnt_metadata_1["test_config_1"] == "foo"
assert chkpnt_metadata_1["test_config_2"] == "bar"
@@ -3939,7 +3932,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
# on how the graph is constructed.
chkpnt_tuples_1 = sync_checkpointer.list(config)
for chkpnt_tuple in chkpnt_tuples_1:
assert chkpnt_tuple.metadata["thread_id"] == "1"
assert chkpnt_tuple.metadata["test_config_1"] == "foo"
assert chkpnt_tuple.metadata["test_config_2"] == "bar"
@@ -3959,7 +3951,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
# assert that checkpoint metadata contains the run's configurable fields
chkpnt_metadata_2 = sync_checkpointer.get_tuple(config).metadata
assert chkpnt_metadata_2["thread_id"] == "2"
assert chkpnt_metadata_2["test_config_3"] == "foo"
assert chkpnt_metadata_2["test_config_4"] == "bar"
@@ -3977,7 +3968,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
# assert that checkpoint metadata contains the run's configurable fields
chkpnt_metadata_3 = sync_checkpointer.get_tuple(config).metadata
assert chkpnt_metadata_3["thread_id"] == "2"
assert chkpnt_metadata_3["test_config_3"] == "foo"
assert chkpnt_metadata_3["test_config_4"] == "bar"
@@ -3986,7 +3976,6 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
# on how the graph is constructed.
chkpnt_tuples_2 = sync_checkpointer.list(config)
for chkpnt_tuple in chkpnt_tuples_2:
assert chkpnt_tuple.metadata["thread_id"] == "2"
assert chkpnt_tuple.metadata["test_config_3"] == "foo"
assert chkpnt_tuple.metadata["test_config_4"] == "bar"
@@ -3994,14 +3983,9 @@ def test_checkpoint_metadata(sync_checkpointer: BaseCheckpointSaver) -> None:
def test_remove_message_via_state_update(
sync_checkpointer: BaseCheckpointSaver,
) -> None:
from langchain_core.messages import (
AIMessage,
AnyMessage,
HumanMessage,
RemoveMessage,
)
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
workflow = StateGraph(Annotated[list[AnyMessage], add_messages])
workflow = MessageGraph()
workflow.add_node(
"chatbot",
lambda state: [
@@ -4032,14 +4016,9 @@ def test_remove_message_via_state_update(
def test_remove_message_from_node():
from langchain_core.messages import (
AIMessage,
AnyMessage,
HumanMessage,
RemoveMessage,
)
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
workflow = StateGraph(Annotated[list[AnyMessage], add_messages])
workflow = MessageGraph()
workflow.add_node(
"chatbot",
lambda state: [
@@ -4832,7 +4811,9 @@ def test_parent_command(
config = {"configurable": {"thread_id": "1"}}
assert graph.invoke({"messages": [("user", "get user name")]}, config) == {
assert graph.invoke(
{"messages": [("user", "get user name")]}, config, checkpoint_during=False
) == {
"messages": [
_AnyIdHumanMessage(
content="get user name", additional_kwargs={}, response_metadata={}
@@ -4859,7 +4840,6 @@ def test_parent_command(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 1,
"parents": {},
},
@@ -4931,7 +4911,7 @@ def test_interrupt_multiple(sync_checkpointer: BaseCheckpointSaver):
assert [
event
for event in graph.stream(
Command(resume="answer 1", update={"my_key": "foofoo"}), thread1
Command(resume="answer 1", update={"my_key": " foofoo "}), thread1
)
] == [
{
@@ -4946,8 +4926,14 @@ def test_interrupt_multiple(sync_checkpointer: BaseCheckpointSaver):
}
]
assert [event for event in graph.stream(Command(resume="answer 2"), thread1)] == [
{"node": {"my_key": "answer 1 answer 2"}},
assert [
event
for event in graph.stream(
Command(resume="answer 2"), thread1, stream_mode="values"
)
] == [
{"my_key": "DE foofoo "},
{"my_key": "DE foofoo answer 1 answer 2"},
]
@@ -5555,7 +5541,10 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
configurable = {"configurable": {"thread_id": uuid.uuid4()}}
assert [
chunk for chunk in graph.stream({"a": 5}, configurable, stream_mode="debug")
chunk
for chunk in graph.stream(
{"a": 5}, configurable, stream_mode="debug", checkpoint_during=False
)
] == [
{
"payload": {
@@ -5657,7 +5646,12 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
]
assert [
c
for c in graph.stream(Command(resume="123"), configurable, stream_mode="debug")
for c in graph.stream(
Command(resume="123"),
configurable,
stream_mode="debug",
checkpoint_during=False,
)
] == [
{
"payload": {
@@ -5672,7 +5666,6 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
"parents": {},
"source": "input",
"step": -1,
"thread_id": AnyStr(),
},
"next": [
"graph",
+31 -45
View File
@@ -103,7 +103,7 @@ async def test_checkpoint_errors() -> None:
raise ValueError("Faulty put_writes")
class FaultyVersionCheckpointer(InMemorySaver):
def get_next_version(self, current: Optional[int]) -> int:
def get_next_version(self, current: Optional[int], channel: None) -> int:
raise ValueError("Faulty get_next_version")
def logic(inp: str) -> str:
@@ -274,7 +274,9 @@ async def test_checkpoint_put_after_cancellation() -> None:
thread1 = {"configurable": {"thread_id": "1"}}
# start the task
t = asyncio.create_task(graph.ainvoke({"hello": "world"}, thread1))
t = asyncio.create_task(
graph.ainvoke({"hello": "world"}, thread1, checkpoint_during=False)
)
# cancel after 0.2 seconds
await asyncio.sleep(0.2)
t.cancel()
@@ -340,7 +342,7 @@ async def test_checkpoint_put_after_cancellation_stream_anext() -> None:
thread1 = {"configurable": {"thread_id": "1"}}
# start the task
s = graph.astream({"hello": "world"}, thread1)
s = graph.astream({"hello": "world"}, thread1, checkpoint_during=False)
t = asyncio.create_task(s.__anext__())
# cancel after 0.2 seconds
await asyncio.sleep(0.2)
@@ -408,7 +410,11 @@ async def test_checkpoint_put_after_cancellation_stream_events_anext() -> None:
# start the task
s = graph.astream_events(
{"hello": "world"}, thread1, version="v2", include_names=["LangGraph"]
{"hello": "world"},
thread1,
version="v2",
include_names=["LangGraph"],
checkpoint_during=False,
)
# skip first event (happens right away)
await s.__anext__()
@@ -595,7 +601,9 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
# stop when about to enter node
assert [
c
async for c in tool_two.astream({"my_key": "value ⛰️", "market": "DE"}, thread1)
async for c in tool_two.astream(
{"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False
)
] == [
{
"__interrupt__": (
@@ -612,7 +620,6 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
]
tup = await tool_two.checkpointer.aget_tuple(thread1)
@@ -639,7 +646,6 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
parent_config=None,
interrupts=(
@@ -665,7 +671,6 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non
"parents": {},
"source": "update",
"step": 1,
"thread_id": "1",
},
parent_config=(
[c async for c in tool_two.checkpointer.alist(thread1, limit=2)][-1].config
@@ -768,7 +773,9 @@ async def test_dynamic_interrupt_subgraph(
# stop when about to enter node
assert [
c
async for c in tool_two.astream({"my_key": "value ⛰️", "market": "DE"}, thread1)
async for c in tool_two.astream(
{"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False
)
] == [
{
"__interrupt__": (
@@ -785,7 +792,6 @@ async def test_dynamic_interrupt_subgraph(
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
]
tup = await tool_two.checkpointer.aget_tuple(thread1)
@@ -818,7 +824,6 @@ async def test_dynamic_interrupt_subgraph(
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
parent_config=None,
interrupts=(
@@ -844,7 +849,6 @@ async def test_dynamic_interrupt_subgraph(
"parents": {},
"source": "update",
"step": 1,
"thread_id": "1",
},
parent_config=(
[c async for c in tool_two.checkpointer.alist(thread1root, limit=2)][
@@ -946,7 +950,9 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
# flow: interrupt -> clear tasks
thread1 = {"configurable": {"thread_id": "1"}}
# stop when about to enter node
assert await tool_two.ainvoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
assert await tool_two.ainvoke(
{"my_key": "value ⛰️", "market": "DE"}, thread1, checkpoint_during=False
) == {
"my_key": "value ⛰️ one",
"market": "DE",
"__interrupt__": [
@@ -963,7 +969,6 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
]
@@ -1000,7 +1005,6 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
},
parent_config=None,
interrupts=(
@@ -1039,7 +1043,6 @@ async def test_copy_checkpoint(async_checkpointer: BaseCheckpointSaver) -> None:
"parents": {},
"source": "fork",
"step": 1,
"thread_id": "1",
},
parent_config=(
[c async for c in tool_two.checkpointer.alist(thread1, limit=2)][-1].config
@@ -1217,7 +1220,6 @@ async def test_cancel_graph_astream(async_checkpointer: BaseCheckpointSaver) ->
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
}
@@ -1292,7 +1294,6 @@ async def test_cancel_graph_astream_events_v2(
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "2",
}
@@ -1815,7 +1816,6 @@ async def test_pending_writes_resume(
"parents": {},
"source": "loop",
"step": 0,
"thread_id": "1",
}
# get_state with checkpoint_id should not apply any pending writes
state = await graph.aget_state(state.config)
@@ -1905,7 +1905,6 @@ async def test_pending_writes_resume(
"parents": {},
"step": 1,
"source": "loop",
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -1953,7 +1952,6 @@ async def test_pending_writes_resume(
"parents": {},
"step": 0,
"source": "loop",
"thread_id": "1",
},
parent_config={
"configurable": {
@@ -2001,7 +1999,6 @@ async def test_pending_writes_resume(
"parents": {},
"step": -1,
"source": "input",
"thread_id": "1",
},
parent_config=None,
pending_writes=UnsortedSequence(
@@ -2601,7 +2598,6 @@ async def test_send_dedupe_on_resume(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 4,
"parents": {},
},
@@ -2637,7 +2633,6 @@ async def test_send_dedupe_on_resume(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 3,
"parents": {},
},
@@ -2680,7 +2675,6 @@ async def test_send_dedupe_on_resume(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 2,
"parents": {},
},
@@ -2735,7 +2729,6 @@ async def test_send_dedupe_on_resume(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 1,
"parents": {},
},
@@ -2790,7 +2783,6 @@ async def test_send_dedupe_on_resume(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 0,
"parents": {},
},
@@ -2827,7 +2819,6 @@ async def test_send_dedupe_on_resume(
},
metadata={
"source": "input",
"thread_id": "1",
"step": -1,
"parents": {},
},
@@ -2957,7 +2948,9 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
foo_called = 0
graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"])
thread1 = {"configurable": {"thread_id": "2"}}
assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == {
assert await graph.ainvoke(
{"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False
) == {
"messages": [
_AnyIdHumanMessage(content="hello"),
_AnyIdAIMessage(
@@ -3006,7 +2999,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
"step": 1,
"source": "loop",
"parents": {},
"thread_id": "2",
},
created_at=AnyStr(),
parent_config=None,
@@ -3052,7 +3044,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
"step": 2,
"source": "update",
"parents": {},
"thread_id": "2",
},
created_at=AnyStr(),
parent_config=(
@@ -3081,7 +3072,9 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
foo_called = 0
graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"])
thread1 = {"configurable": {"thread_id": "3"}}
assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == {
assert await graph.ainvoke(
{"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False
) == {
"messages": [
_AnyIdHumanMessage(content="hello"),
_AnyIdAIMessage(
@@ -3130,7 +3123,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
"step": 1,
"source": "loop",
"parents": {},
"thread_id": "3",
},
created_at=AnyStr(),
parent_config=None,
@@ -3197,7 +3189,6 @@ async def test_send_react_interrupt(async_checkpointer: BaseCheckpointSaver) ->
"step": 2,
"source": "update",
"parents": {},
"thread_id": "3",
},
created_at=AnyStr(),
parent_config=(
@@ -3346,7 +3337,9 @@ async def test_send_react_interrupt_control(
foo_called = 0
graph = builder.compile(checkpointer=async_checkpointer, interrupt_before=["foo"])
thread1 = {"configurable": {"thread_id": "2"}}
assert await graph.ainvoke({"messages": [HumanMessage("hello")]}, thread1) == {
assert await graph.ainvoke(
{"messages": [HumanMessage("hello")]}, thread1, checkpoint_during=False
) == {
"messages": [
_AnyIdHumanMessage(content="hello"),
_AnyIdAIMessage(
@@ -3395,7 +3388,6 @@ async def test_send_react_interrupt_control(
"step": 1,
"source": "loop",
"parents": {},
"thread_id": "2",
},
created_at=AnyStr(),
parent_config=None,
@@ -3441,7 +3433,6 @@ async def test_send_react_interrupt_control(
"step": 2,
"source": "update",
"parents": {},
"thread_id": "2",
},
created_at=AnyStr(),
parent_config=(
@@ -4387,7 +4378,6 @@ async def test_in_one_fan_out_state_graph_waiting_edge_custom_state_class(
"parents": {},
"source": "loop",
"step": 4,
"thread_id": "1",
},
created_at=AnyStr(),
parent_config=(
@@ -5646,7 +5636,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
# assert that checkpoint metadata contains the run's configurable fields
chkpnt_metadata_1 = (await async_checkpointer.aget_tuple(config)).metadata
assert chkpnt_metadata_1["thread_id"] == "1"
assert chkpnt_metadata_1["test_config_1"] == "foo"
assert chkpnt_metadata_1["test_config_2"] == "bar"
@@ -5655,7 +5644,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
# on how the graph is constructed.
chkpnt_tuples_1 = async_checkpointer.alist(config)
async for chkpnt_tuple in chkpnt_tuples_1:
assert chkpnt_tuple.metadata["thread_id"] == "1"
assert chkpnt_tuple.metadata["test_config_1"] == "foo"
assert chkpnt_tuple.metadata["test_config_2"] == "bar"
@@ -5675,7 +5663,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
# assert that checkpoint metadata contains the run's configurable fields
chkpnt_metadata_2 = (await async_checkpointer.aget_tuple(config)).metadata
assert chkpnt_metadata_2["thread_id"] == "2"
assert chkpnt_metadata_2["test_config_3"] == "foo"
assert chkpnt_metadata_2["test_config_4"] == "bar"
@@ -5693,7 +5680,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
# assert that checkpoint metadata contains the run's configurable fields
chkpnt_metadata_3 = (await async_checkpointer.aget_tuple(config)).metadata
assert chkpnt_metadata_3["thread_id"] == "2"
assert chkpnt_metadata_3["test_config_3"] == "foo"
assert chkpnt_metadata_3["test_config_4"] == "bar"
@@ -5702,7 +5688,6 @@ async def test_checkpoint_metadata(async_checkpointer: BaseCheckpointSaver) -> N
# on how the graph is constructed.
chkpnt_tuples_2 = async_checkpointer.alist(config)
async for chkpnt_tuple in chkpnt_tuples_2:
assert chkpnt_tuple.metadata["thread_id"] == "2"
assert chkpnt_tuple.metadata["test_config_3"] == "foo"
assert chkpnt_tuple.metadata["test_config_4"] == "bar"
@@ -6110,7 +6095,9 @@ async def test_parent_command(
config = {"configurable": {"thread_id": "1"}}
assert await graph.ainvoke({"messages": [("user", "get user name")]}, config) == {
assert await graph.ainvoke(
{"messages": [("user", "get user name")]}, config, checkpoint_during=False
) == {
"messages": [
_AnyIdHumanMessage(
content="get user name", additional_kwargs={}, response_metadata={}
@@ -6139,7 +6126,6 @@ async def test_parent_command(
},
metadata={
"source": "loop",
"thread_id": "1",
"step": 1,
"parents": {},
},
+19 -1
View File
@@ -1,6 +1,6 @@
from dataclasses import dataclass
from operator import add
from typing import Annotated, Any
from typing import Annotated, Any, Union
from langchain_core.runnables import RunnableConfig
from pydantic import BaseModel
@@ -103,3 +103,21 @@ def test_input_state_specified() -> None:
new_graph.invoke({"something": 1})
new_graph.invoke({"something": 2, "info": ["hello", "world"]}) # type: ignore[arg-type]
def test_invokeable_node_signature() -> None:
class State(TypedDict):
info: Annotated[list[str], add]
graph_builder = StateGraph(State)
class RunnableIsh:
def invoke(
self,
input: State,
config: Union[RunnableConfig, None] = None,
**kwargs: Any,
) -> dict[str, str]:
return {}
graph_builder.add_node("runnable", RunnableIsh())
+4 -3
View File
@@ -1201,7 +1201,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.4.7"
version = "0.5.0rc1"
source = { editable = "." }
dependencies = [
{ name = "langchain-core" },
@@ -1310,7 +1310,7 @@ wheels = [
[[package]]
name = "langgraph-checkpoint"
version = "2.0.26"
version = "2.1.0"
source = { editable = "../checkpoint" }
dependencies = [
{ name = "langchain-core" },
@@ -1330,6 +1330,7 @@ dev = [
{ name = "mypy" },
{ name = "numpy" },
{ name = "pandas" },
{ name = "pandas-stubs", specifier = ">=2.2.2.240807" },
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
@@ -1422,7 +1423,7 @@ inmem = [
[[package]]
name = "langgraph-prebuilt"
version = "0.2.2"
version = "0.5.0rc0"
source = { editable = "../prebuilt" }
dependencies = [
{ name = "langchain-core" },
@@ -591,7 +591,7 @@ def create_react_agent(
workflow = StateGraph(state_schema, config_schema=config_schema)
workflow.add_node(
"agent",
RunnableCallable(call_model, acall_model), # type: ignore[call-overload]
RunnableCallable(call_model, acall_model),
input_schema=input_schema,
)
if pre_model_hook is not None:
@@ -610,7 +610,7 @@ def create_react_agent(
if response_format is not None:
workflow.add_node(
"generate_structured_response",
RunnableCallable( # type: ignore[call-overload]
RunnableCallable(
generate_structured_response,
agenerate_structured_response,
),
@@ -660,10 +660,10 @@ def create_react_agent(
# Define the two nodes we will cycle between
workflow.add_node(
"agent",
RunnableCallable(call_model, acall_model), # type: ignore[call-overload]
RunnableCallable(call_model, acall_model),
input_schema=input_schema,
)
workflow.add_node("tools", tool_node) # type: ignore[call-overload]
workflow.add_node("tools", tool_node)
# Optionally add a pre-model hook node that will be called
# every time before the "agent" (LLM-calling node)
@@ -693,7 +693,7 @@ def create_react_agent(
if response_format is not None:
workflow.add_node(
"generate_structured_response",
RunnableCallable( # type: ignore[call-overload]
RunnableCallable(
generate_structured_response,
agenerate_structured_response,
),
@@ -629,7 +629,7 @@ def tools_condition(
Args:
state: The state to check for
tool calls. Must have a list of messages or have the
tool calls. Must have a list of messages (MessageGraph) or have the
"messages" key (StateGraph).
Returns:
@@ -2,7 +2,7 @@
in a langchain graph. It applies a pydantic schema to tool_calls in the models' outputs,
and returns a ToolMessage with the validated content. If the schema is not valid, it
returns a ToolMessage with the error message. The ValidationNode can be used in a
StateGraph with a "messages" key. If multiple tool calls are
StateGraph with a "messages" key or in a MessageGraph. If multiple tool calls are
requested, they will be run in parallel.
"""
@@ -49,7 +49,7 @@ def _default_format_error(
class ValidationNode(RunnableCallable):
"""A node that validates all tools requests from the last AIMessage.
It can be used in StateGraph with a "messages" key.
It can be used either in StateGraph with a "messages" key or in MessageGraph.
!!! note
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-prebuilt"
version = "0.2.2"
version = "0.5.0rc0"
description = "Library with high-level APIs for creating and executing LangGraph agents and tools."
authors = []
requires-python = ">=3.9"
@@ -12,7 +12,7 @@ readme = "README.md"
license = "MIT"
license-files = ['LICENSE']
dependencies = [
"langgraph-checkpoint>=2.0.10",
"langgraph-checkpoint>=2.1.0",
"langchain-core>=0.3.22",
]
+1 -1
View File
@@ -13,9 +13,9 @@ from langgraph.checkpoint.base import (
CheckpointMetadata,
CheckpointTuple,
SerializerProtocol,
copy_checkpoint,
)
from langgraph.checkpoint.memory import InMemorySaver, PersistentDict
from langgraph.pregel.checkpoint import copy_checkpoint
class NoopSerializer(SerializerProtocol):
-2
View File
@@ -91,7 +91,6 @@ def test_no_prompt(sync_checkpointer: BaseCheckpointSaver, version: str) -> None
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "123",
}
assert saved.pending_writes == []
@@ -118,7 +117,6 @@ async def test_no_prompt_async(async_checkpointer: BaseCheckpointSaver) -> None:
"parents": {},
"source": "loop",
"step": 1,
"thread_id": "123",
}
assert saved.pending_writes == []
+4 -3
View File
@@ -320,7 +320,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.4.7"
version = "0.5.0rc1"
source = { editable = "../langgraph" }
dependencies = [
{ name = "langchain-core" },
@@ -371,7 +371,7 @@ dev = [
[[package]]
name = "langgraph-checkpoint"
version = "2.0.26"
version = "2.1.0"
source = { editable = "../checkpoint" }
dependencies = [
{ name = "langchain-core" },
@@ -391,6 +391,7 @@ dev = [
{ name = "mypy" },
{ name = "numpy" },
{ name = "pandas" },
{ name = "pandas-stubs", specifier = ">=2.2.2.240807" },
{ name = "pytest" },
{ name = "pytest-asyncio" },
{ name = "pytest-mock" },
@@ -463,7 +464,7 @@ dev = [
[[package]]
name = "langgraph-prebuilt"
version = "0.2.2"
version = "0.5.0rc0"
source = { editable = "." }
dependencies = [
{ name = "langchain-core" },
+7 -1
View File
@@ -894,7 +894,13 @@ export function useStream<
if (event === "events") options.onLangChainEvent?.(data);
if (event === "debug") options.onDebugEvent?.(data);
if (event === "values") setStreamValues(data);
if (event === "values") {
if ("__interrupt__" in data) {
// don't update values on interrupt values event
continue;
}
setStreamValues(data);
}
if (event === "messages") {
const [serialized] = data;
+11 -1
View File
@@ -36,7 +36,15 @@ Represents the status of a thread:
"""
StreamMode = Literal[
"values", "messages", "updates", "events", "debug", "custom", "messages-tuple"
"values",
"messages",
"updates",
"events",
"tasks",
"checkpoints",
"debug",
"custom",
"messages-tuple",
]
"""
Defines the mode of streaming:
@@ -44,6 +52,8 @@ Defines the mode of streaming:
- "messages": Stream complete messages.
- "updates": Stream updates to the state.
- "events": Stream events occurring during execution.
- "checkpoints": Stream checkpoints as they are created.
- "tasks": Stream task start and finish events.
- "debug": Stream detailed debug information.
- "custom": Stream custom events.
"""