Compare commits

..
Author SHA1 Message Date
Eugene Yurtsev 55880c9813 x 2025-07-27 16:41:25 -04:00
Eugene Yurtsev 22af613437 x 2025-07-27 16:24:58 -04:00
Eugene Yurtsev a254978893 x 2025-07-25 17:06:13 -04:00
29 changed files with 562 additions and 3190 deletions
+9
View File
@@ -35,7 +35,16 @@ jobs:
with:
filter: "docs/docs/**"
# TODO: Uncomment this to run on PRs
# run-changed-notebooks:
# needs: get-changed-files
# uses: ./.github/workflows/run_notebooks.yml
# secrets: inherit
# with:
# changed-files: ${{ needs.get-changed-files.outputs.changed-files }}
deploy:
# needs: run-changed-notebooks
runs-on: ubuntu-latest
timeout-minutes: 10 # Job will be cancelled if it runs for more than 10 minutes
env:
-1
View File
@@ -39,7 +39,6 @@ jobs:
scheduler-kafka
sdk-py
docs
ci
requireScope: false
ignoreLabels: |
ignore-lint-pr-title
+1 -3
View File
@@ -137,9 +137,7 @@ jobs:
needs:
- build
- release-notes
permissions:
contents: read
id-token: write
permissions: write-all
uses: ./.github/workflows/_test_release.yml
with:
working-directory: ${{ inputs.working-directory }}
+2 -2
View File
@@ -15,8 +15,8 @@ build-prebuilt:
fi
uv run python -m _scripts.third_party_page.create_third_party_page stats.yml docs/agents/prebuilt.md --language python
build-docs: build-typedoc build-prebuilt
TARGET_LANGUAGE=python uv run python -m mkdocs build --clean -f mkdocs.yml --strict
build-docs: build-prebuilt
uv run python -m mkdocs build --clean -f mkdocs.yml --strict
llms-text:
uv run python -m _scripts.generate_llms_text docs/llms-full.txt
+5 -10
View File
@@ -310,12 +310,6 @@ def _highlight_code_blocks(markdown: str) -> str:
return markdown
TARGET_LANGUAGE = os.environ.get("TARGET_LANGUAGE", "python")
if TARGET_LANGUAGE not in {"python", "js"}:
raise ValueError(f"TARGET_LANGUAGE must be 'python' or 'js', got {TARGET_LANGUAGE}")
def _on_page_markdown_with_config(
markdown: str,
page: Page,
@@ -338,15 +332,16 @@ def _on_page_markdown_with_config(
markdown = _highlight_code_blocks(markdown)
# Apply conditional rendering for code blocks
markdown = _apply_conditional_rendering(markdown, TARGET_LANGUAGE)
if TARGET_LANGUAGE == "js":
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":
elif target_language == "python":
# Via a dedicated plugin
pass
else:
raise ValueError(
f"Unsupported target language: {TARGET_LANGUAGE}. "
f"Unsupported target language: {target_language}. "
"Supported languages are 'python' and 'js'."
)
+6 -13
View File
@@ -8,7 +8,7 @@ Context includes *any* data outside the message list that can shape behavior. Th
- Internal state updated during a multi-step reasoning process.
- Persistent memory or facts from previous interactions.
LangGraph provides **three** primary ways to manage context:
LangGraph provides **three** primary ways to supply context:
| Type | Description | Mutable? | Lifetime |
|------------------------------------------------------------------------------|-----------------------------------------------|----------|-------------------------|
@@ -18,21 +18,14 @@ LangGraph provides **three** primary ways to manage context:
### Runtime Context
Runtime context is for immutable data like user metadata, tools, db connections, etc. Use this when you have values that don't change mid-run.
!!! note "`config['configurable']` -> `runtime.context`"
!!! version-added "New in LangGraph v0.6: `Runtime.context` replaces `config['configurable']`"
In LangGraph < v1.0, static runtime context was passed via the `config['configurable']` key, paired with a `config_schema` argument
to `StateGraph` or `Pregel`. This is now deprecated and will be removed in v2.0.
The `Runtime` object is recommended to access static context and runtime-specific information like the store and stream writer.
As of LangGraph v1.0, the Runtime object is recommended to access static context and runtime-specific information like the store and stream writer.
!!! note
Runtime context refers to local context: data and dependencies your code needs to run. It does not refer to:
* The LLM context, which is the data passed into the LLM's prompt.
* The "context window", which is the maximum number of tokens that can be passed to the LLM.
You likely want to use the local context to optimize the LLM's context window. For example, you
could use a user id to fetch a user's name and information from a database to populate the context window with relevant memories.
Runtime context is for immutable data like user metadata or API keys. Use this when you have values that don't change mid-run.
Specify static context via the `context` argument to `invoke` / `stream`, which is reserved for this purpose:
@@ -4,22 +4,6 @@
---
## v0.2.109 (2025-07-28)
- Fixed an issue where missing config schema occurred when `config_type` was not set.
## v0.2.108 (2025-07-28)
- Added compatibility for langgraph v0.6, including new context API support and a migration to enhance context handling in assistant operations.
## v0.2.107 (2025-07-27)
- Implemented caching for authentication processes to improve performance.
- Merged count and select queries to improve database query efficiency.
## v0.2.106 (2025-07-27)
- Log whether run uses resumable streams.
## v0.2.105 (2025-07-27)
- Added a `/heapdump` endpoint to capture and save JS process heap data.
## v0.2.103 (2025-07-25)
- Corrected the metadata endpoint to ensure accurate data retrieval.
+1 -1
View File
@@ -28,7 +28,7 @@ To review, edit, and approve tool calls in an agent or workflow, [use LangGraph'
There are two ways to pause a graph:
- [Dynamic interrupts](../how-tos/human_in_the_loop/add-human-in-the-loop.md#pause-using-interrupt): Use `interrupt` to pause a graph from inside a specific node, based on the current state of the graph.
- [Static interrupts](../how-tos/human_in_the_loop/add-human-in-the-loop.md#debug-with-interrupts): Use `interrupt_before` and `interrupt_after` to pause the graph at pre-defined points, either before or after a node executes.
- [Static interrupts](../how-tos/human_in_the_loop/add-human-in-the-loop.md#debug-with-interrupts): Use `interrupt_before` and `interrupt_after` to pause the graph at defined points, either before or after a node executes.
<figure markdown="1">
![image](./img/breakpoints.png){: style="max-height:400px"}
Binary file not shown.

Before

Width:  |  Height:  |  Size: 9.2 KiB

@@ -128,7 +128,7 @@ print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)!
!!! tip "New in 0.4.0"
`__interrupt__` is a special key that will be returned when running the graph if the graph is interrupted. Support for `__interrupt__` in `invoke` and `ainvoke` has been added in version 0.4.0. If you're on an older version, you will only see `__interrupt__` in the result if you use `stream` or `astream`. You can also use `graph.get_state(thread_id)` to get the interrupt value(s).
`__interrupt__` is a special key that will be returned when running the graph if the graph is interrupted. Support for `__interrupt__` in `invoke` and `ainvoke` has been added in version 0.4.0. If you're on an older version, you will only see `__interrupt__` in the result if you use `stream` or `astream`. You can also use `graph.get_state(thread_id)` to get the interrupt value.
!!! warning
@@ -145,67 +145,19 @@ To resume execution, use the [`Command`][langgraph.types.Command] primitive, whi
graph.invoke(Command(resume={"age": "25"}), thread_config)
```
## Resuming Multiple interrupts
### Resume multiple interrupts with one invocation
When nodes with interrupt conditions are run in parallel, it's possible to have multiple interrupts in the task queue.
For example, the following graph has two nodes run in parallel that require human input:
<figure markdown="1">
![image](../assets/human_in_loop_parallel.png){: style="max-height:400px"}
</figure>
Once your graph has been interrupted and is stalled, you can resume all the interrupts at once with `Command.resume`, passing a dictionary mapping of interrupt ids to resume values.
If you have multiple interrupts in the task queue, you can use `Command.resume` with a dictionary mapping of interrupt ids to resume with a single `invoke` / `stream` call.
For example, once your graph has been interrupted (multiple times, theoretically) and is stalled:
```python
from typing import TypedDict
import uuid
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command
class State(TypedDict):
text_1: str
text_2: str
def human_node_1(state: State):
value = interrupt({"text_to_revise": state["text_1"]})
return {"text_1": value}
def human_node_2(state: State):
value = interrupt({"text_to_revise": state["text_2"]})
return {"text_2": value}
graph_builder = StateGraph(State)
graph_builder.add_node("human_node_1", human_node_1)
graph_builder.add_node("human_node_2", human_node_2)
# Add both nodes in parallel from START
graph_builder.add_edge(START, "human_node_1")
graph_builder.add_edge(START, "human_node_2")
checkpointer = InMemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
thread_id = str(uuid.uuid4())
config: RunnableConfig = {"configurable": {"thread_id": thread_id}}
result = graph.invoke(
{"text_1": "original text 1", "text_2": "original text 2"}, config=config
)
# Resume with mapping of interrupt IDs to values
resume_map = {
i.id: f"edited text for {i.value['text_to_revise']}"
for i in result["__interrupt__"]
i.id: f"human input for prompt {i.value}"
for i in parent.get_state(thread_config).interrupts
}
print(graph.invoke(Command(resume=resume_map), config=config))
# > {'text_1': 'edited text for original text 1', 'text_2': 'edited text for original text 2'}
parent_graph.invoke(Command(resume=resume_map), config=thread_config)
```
## Common patterns
@@ -1075,7 +1027,7 @@ def node_in_parent_graph(state: State):
{'parent_node': {'state_counter': 1}}
```
### Using multiple interrupts in a single node
### Using multiple interrupts
Using multiple interrupts within a **single** node can be helpful for patterns like [validating human input](#validate-human-input). However, using multiple interrupts in the same node can lead to unexpected behavior if not handled carefully.
@@ -1,6 +1,6 @@
# Build a basic chatbot
In this tutorial, you will build a basic chatbot. This chatbot is the basis for the following series of tutorials where you will progressively add more sophisticated capabilities, and be introduced to key LangGraph concepts along the way. Let's dive in! 🌟
In this tutorial, you will build a basic chatbot. This chatbot is the basis for the following series of tutorials where you will progressively add more sophisticated capabilities, and be introduced to key LangGraph concepts along the way. Lets dive in! 🌟
## Prerequisites
@@ -13,45 +13,13 @@ tool-calling features, such as [OpenAI](https://platform.openai.com/api-keys),
Install the required packages:
:::python
```bash
pip install -U langgraph langsmith
```
:::
:::js
=== "npm"
```bash
npm install @langchain/langgraph @langchain/core zod
```
=== "yarn"
```bash
yarn add @langchain/langgraph @langchain/core zod
```
=== "pnpm"
```bash
pnpm add @langchain/langgraph @langchain/core zod
```
=== "bun"
```bash
bun add @langchain/langgraph @langchain/core zod
```
:::
!!! tip
Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph. For more information on how to get started, see [LangSmith docs](https://docs.smith.langchain.com).
Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph. For more information on how to get started, see [LangSmith docs](https://docs.smith.langchain.com).
Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph. For more information on how to get started, see [LangSmith docs](https://docs.smith.langchain.com).
## 2. Create a `StateGraph`
@@ -59,8 +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
@@ -80,43 +46,24 @@ class State(TypedDict):
graph_builder = StateGraph(State)
```
:::
:::js
```typescript
import { StateGraph, MessagesZodState, START } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const graph = new StateGraph(State).compile();
```
:::
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 reducer function.
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. In our example, `State` is a schema with one key: `messages`. The 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).
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:
{!snippets/chat_model_tabs.md!}
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
<!---
```python
@@ -126,26 +73,9 @@ llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
-->
:::
:::js
```typescript
import { ChatOpenAI } from "@langchain/openai";
// or import { ChatAnthropic } from "@langchain/anthropic";
const llm = new ChatOpenAI({
model: "gpt-4o",
temperature: 0,
});
```
:::
We can now incorporate the chat model into a simple node:
:::python
```python
def chatbot(state: State):
@@ -158,133 +88,38 @@ def chatbot(state: State):
graph_builder.add_node("chatbot", chatbot)
```
:::
:::js
```typescript hl_lines="7-9"
import { StateGraph, MessagesZodState, START } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const graph = new StateGraph(State)
.addNode("chatbot", async (state: z.infer<typeof State>) => {
return { messages: [await llm.invoke(state.messages)] };
})
.compile();
```
:::
**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 `addMessages` function used within `MessagesZodState` 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 hl_lines="10"
import { StateGraph, MessagesZodState, START } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const graph = new StateGraph(State)
.addNode("chatbot", async (state: z.infer<typeof State>) => {
return { messages: [await llm.invoke(state.messages)] };
})
.addEdge(START, "chatbot")
.compile();
```
:::
## 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 hl_lines="11"
import { StateGraph, MessagesZodState, START, END } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const graph = new StateGraph(State)
.addNode("chatbot", async (state: z.infer<typeof State>) => {
return { messages: [await llm.invoke(state.messages)] };
})
.addEdge(START, "chatbot")
.addEdge("chatbot", END)
.compile();
```
:::
This tells the graph to terminate after running the chatbot node.
## 6. Compile the graph
Before running the graph, we'll need to compile it. We can do so by calling `compile()`
on the graph builder. This creates a `CompiledGraph` we can invoke on our state.
:::python
on the graph builder. This creates a `CompiledStateGraph` we can invoke on our state.
```python
graph = graph_builder.compile()
```
:::
:::js
```typescript hl_lines="12"
import { StateGraph, MessagesZodState, START, END } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const graph = new StateGraph(State)
.addNode("chatbot", async (state: z.infer<typeof State>) => {
return { messages: [await llm.invoke(state.messages)] };
})
.addEdge(START, "chatbot")
.addEdge("chatbot", END)
.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
@@ -297,35 +132,17 @@ except Exception:
pass
```
:::
:::js
You can visualize the graph using the `getGraph` method and render the graph with the `drawMermaidPng` method.
```typescript
import * as fs from "node:fs/promises";
const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());
await fs.writeFile("basic-chatbot.png", imageBuffer);
```
:::
![basic chatbot diagram](basic-chatbot.png)
## 8. Run the chatbot
Now run the chatbot!
Now run the chatbot!
!!! tip
You can exit the chat loop at any time by typing `quit`, `exit`, or `q`.
:::python
```python
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
@@ -348,90 +165,15 @@ while True:
break
```
:::
:::js
```typescript
import { HumanMessage } from "@langchain/core/messages";
async function streamGraphUpdates(userInput: string) {
const stream = await graph.stream({
messages: [new HumanMessage(userInput)],
});
import * as readline from "node:readline/promises";
import { StateGraph, MessagesZodState, START, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";
const llm = new ChatOpenAI({ model: "gpt-4o-mini" });
const State = z.object({ messages: MessagesZodState.shape.messages });
const graph = new StateGraph(State)
.addNode("chatbot", async (state: z.infer<typeof State>) => {
return { messages: [await llm.invoke(state.messages)] };
})
.addEdge(START, "chatbot")
.addEdge("chatbot", END)
.compile();
async function generateText(content: string) {
const stream = await graph.stream(
{ messages: [{ type: "human", content }] },
{ streamMode: "values" }
);
for await (const event of stream) {
for (const value of Object.values(event)) {
console.log(
"Assistant:",
value.messages[value.messages.length - 1].content
);
const lastMessage = event.messages.at(-1);
if (lastMessage?.getType() === "ai") {
console.log(`Assistant: ${lastMessage.text}`);
}
}
}
const prompt = readline.createInterface({
input: process.stdin,
output: process.stdout,
});
while (true) {
const human = await prompt.question("User: ");
if (["quit", "exit", "q"].includes(human.trim())) break;
await generateText(human || "What do you know about LangGraph?");
}
prompt.close();
```
:::
```
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.
```
:::python
```
Goodbye!
```
:::
**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.
:::python
Below is the full code for this tutorial:
:::python
```python
from typing import Annotated
@@ -465,44 +207,8 @@ graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()
```
:::
:::js
```typescript
import { Annotation } from "@langchain/langgraph";
import { StateGraph, START, END } from "@langchain/langgraph";
import { BaseMessage, HumanMessage } from "@langchain/core/messages";
import { ChatOpenAI } from "@langchain/openai";
const State = Annotation.Root({
messages: Annotation<BaseMessage[]>({
reducer: (x, y) => x.concat(y),
}),
});
const graphBuilder = new StateGraph(State);
const llm = new ChatOpenAI({
model: "gpt-4o",
temperature: 0,
});
const chatbot = async (state: typeof State.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.
+12 -410
View File
@@ -10,65 +10,24 @@ To handle queries that your chatbot can't answer "from memory", integrate a web
Before you start this tutorial, ensure you have the following:
:::python
- An API key for the [Tavily Search Engine](https://python.langchain.com/docs/integrations/tools/tavily_search/).
:::
:::js
- 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://docs.tavily.com/):
=== "npm"
```bash
npm install @langchain/tavily
```
=== "yarn"
```bash
yarn add @langchain/tavily
```
=== "pnpm"
```bash
pnpm add @langchain/tavily
```
=== "bun"
```bash
bun add @langchain/tavily
```
:::
## 2. Configure your environment
Configure your environment with your search engine API key:
:::python
```python
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
```bash
_set_env("TAVILY_API_KEY")
```
@@ -76,22 +35,10 @@ _set_env("TAVILY_API_KEY")
os.environ["TAVILY_API_KEY"]: "········"
```
:::
:::js
```typescript
process.env.TAVILY_API_KEY = "tvly-...";
```
:::
## 3. Define the tool
Define the web search tool:
:::python
```python
from langchain_tavily import TavilySearch
@@ -100,25 +47,8 @@ tools = [tool]
tool.invoke("What's a 'node' in LangGraph?")
```
:::
:::js
```typescript
import { TavilySearch } from "@langchain/tavily";
const tool = new TavilySearch({ maxResults: 2 });
const tools = [tool];
await tool.invoke({ query: "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,
@@ -137,51 +67,13 @@ The results are page summaries our chat bot can use to answer questions:
'response_time': 1.38}
```
:::
:::js
```json
{
"query": "What's a 'node' in LangGraph?",
"follow_up_questions": null,
"answer": null,
"images": [],
"results": [
{
"url": "https://blog.langchain.dev/langgraph/",
"title": "LangGraph - LangChain Blog",
"content": "TL;DR: LangGraph is module built on top of LangChain to better enable creation of cyclical graphs, often needed for agent runtimes. This state is updated by nodes in the graph, which return operations to attributes of this state (in the form of a key-value store). After adding nodes, you can then add edges to create the graph. An example of this may be in the basic agent runtime, where we always want the model to be called after we call a tool. The state of this graph by default contains concepts that should be familiar to you if you've used LangChain agents: `input`, `chat_history`, `intermediate_steps` (and `agent_outcome` to represent the most recent agent outcome)",
"score": 0.7407191,
"raw_content": null
},
{
"url": "https://medium.com/@cplog/introduction-to-langgraph-a-beginners-guide-14f9be027141",
"title": "Introduction to LangGraph: A Beginner's Guide - Medium",
"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. Image 10: Introduction to AI Agent with LangChain and LangGraph: A Beginners Guide Image 18: How to build LLM Agent with LangGraph — StateGraph and Reducer Image 20: Simplest Graphs using LangGraph Framework Image 24: Building a ReAct Agent with Langgraph: A Step-by-Step Guide Image 28: Building an Agentic RAG with LangGraph: A Step-by-Step Guide",
"score": 0.65279555,
"raw_content": null
}
],
"response_time": 1.34
}
```
:::
## 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.
:::
:::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:
{!snippets/chat_model_tabs.md!}
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
<!---
```python
@@ -191,22 +83,8 @@ 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 a `StateGraph`:
:::python
```python hl_lines="15"
from typing import Annotated
@@ -230,31 +108,9 @@ def chatbot(state: State):
graph_builder.add_node("chatbot", chatbot)
```
:::
:::js
```typescript hl_lines="7-8"
import { StateGraph, MessagesZodState } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const chatbot = async (state: z.infer<typeof State>) => {
// Modification: tell the LLM which tools it can call
const llmWithTools = llm.bindTools(tools);
return { messages: [await llmWithTools.invoke(state.messages)] };
};
```
:::
## 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.
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
import json
@@ -296,80 +152,16 @@ graph_builder.add_node("tools", tool_node)
If you do not want to build this yourself in the future, you can use LangGraph's prebuilt [ToolNode](https://langchain-ai.github.io/langgraph/reference/agents/#langgraph.prebuilt.tool_node.ToolNode).
:::
:::js
Now, create a function to run the tools if they are called. Do this by adding the tools to a new node called `"tools"` 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 type { StructuredToolInterface } from "@langchain/core/tools";
import { isAIMessage, ToolMessage } from "@langchain/core/messages";
function createToolNode(tools: StructuredToolInterface[]) {
const toolByName: Record<string, StructuredToolInterface> = {};
for (const tool of tools) {
toolByName[tool.name] = tool;
}
return async (inputs: z.infer<typeof State>) => {
const { messages } = inputs;
if (!messages || messages.length === 0) {
throw new Error("No message found in input");
}
const message = messages.at(-1);
if (!message || !isAIMessage(message) || !message.tool_calls) {
throw new Error("Last message is not an AI message with tool calls");
}
const outputs: ToolMessage[] = [];
for (const toolCall of message.tool_calls) {
if (!toolCall.id) throw new Error("Tool call ID is required");
const tool = toolByName[toolCall.name];
if (!tool) throw new Error(`Tool ${toolCall.name} not found`);
const result = await tool.invoke(toolCall.args);
outputs.push(
new ToolMessage({
content: JSON.stringify(result),
name: toolCall.name,
tool_call_id: toolCall.id,
})
);
}
return { messages: outputs };
};
}
```
!!! note
If you do not want to build this yourself in the future, you can use LangGraph's prebuilt [ToolNode](https://langchain-ai.github.io/langgraphjs/reference/classes/langgraph_prebuilt.ToolNode.html).
:::
## 6. Define the `conditional_edges`
With the tool node added, now you can define the `conditional_edges`.
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.
:::
:::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.
:::
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.
:::python
```python
def route_tools(
state: State,
@@ -409,61 +201,10 @@ graph = graph_builder.compile()
!!! note
You can replace this with the prebuilt [tools_condition](https://langchain-ai.github.io/langgraph/reference/prebuilt/#tools_condition) to be more concise.
:::
:::js
```typescript
import { END, START } from "@langchain/langgraph";
const routeTools = (state: z.infer<typeof State>) => {
/**
* Use as conditional edge to route to the ToolNode if the last message
* has tool calls.
*/
const lastMessage = state.messages.at(-1);
if (
lastMessage &&
isAIMessage(lastMessage) &&
lastMessage.tool_calls?.length
) {
return "tools";
}
/** Otherwise, route to the end. */
return END;
};
const graph = new StateGraph(State)
.addNode("chatbot", chatbot)
// 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.
.addNode("tools", createToolNode(tools))
// Start the graph with the chatbot
.addEdge(START, "chatbot")
// The `routeTools` function returns "tools" if the chatbot asks to use a tool, and "END" if
// it is fine directly responding.
.addConditionalEdges("chatbot", routeTools, ["tools", END])
// Any time a tool is called, we need to return to the chatbot
.addEdge("tools", "chatbot")
.compile();
```
!!! note
You can replace this with the prebuilt [toolsCondition](https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph_prebuilt.toolsCondition.html) to be more concise.
:::
You can replace this with the prebuilt [tools_condition](https://langchain-ai.github.io/langgraph/reference/prebuilt/#tools_condition) to be more concise.
## 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
@@ -476,31 +217,12 @@ except Exception:
pass
```
:::
:::js
You can visualize the graph using the `getGraph` method and render the graph with the `drawMermaidPng` method.
```typescript
import * as fs from "node:fs/promises";
const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());
await fs.writeFile("chatbot-with-tools.png", imageBuffer);
```
:::
![chatbot-with-tools-diagram](chatbot-with-tools.png)
## 8. Ask the bot questions
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}]}):
@@ -523,7 +245,7 @@ while True:
break
```
```
```
Assistant: [{'text': "To provide you with accurate and up-to-date information about LangGraph, I'll need to search for the latest details. Let me do that for you.", 'type': 'text'}, {'id': 'toolu_01Q588CszHaSvvP2MxRq9zRD', 'input': {'query': 'LangGraph AI tool information'}, 'name': 'tavily_search_results_json', 'type': 'tool_use'}]
Assistant: [{"url": "https://www.langchain.com/langgraph", "content": "LangGraph sets the foundation for how we can build and scale AI workloads \u2014 from conversational agents, complex task automation, to custom LLM-backed experiences that 'just work'. The next chapter in building complex production-ready features with LLMs is agentic, and with LangGraph and LangSmith, LangChain delivers an out-of-the-box solution ..."}, {"url": "https://github.com/langchain-ai/langgraph", "content": "Overview. LangGraph is a library for building stateful, multi-actor applications with LLMs, used to create agent and multi-agent workflows. Compared to other LLM frameworks, it offers these core benefits: cycles, controllability, and persistence. LangGraph allows you to define flows that involve cycles, essential for most agentic architectures ..."}]
Assistant: Based on the search results, I can provide you with information about LangGraph:
@@ -554,99 +276,18 @@ Assistant: Based on the search results, I can provide you with information about
LangGraph appears to be a significant tool in the evolving landscape of LLM-based application development, offering developers new ways to create more complex, stateful, and interactive AI systems.
Goodbye!
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
:::
:::js
```typescript
import readline from "node:readline/promises";
const prompt = readline.createInterface({
input: process.stdin,
output: process.stdout,
});
async function generateText(content: string) {
const stream = await graph.stream(
{ messages: [{ type: "human", content }] },
{ streamMode: "values" }
);
for await (const event of stream) {
const lastMessage = event.messages.at(-1);
if (lastMessage?.getType() === "ai" || lastMessage?.getType() === "tool") {
console.log(`Assistant: ${lastMessage?.text}`);
}
}
}
while (true) {
const human = await prompt.question("User: ");
if (["quit", "exit", "q"].includes(human.trim())) break;
await generateText(human || "What do you know about LangGraph?");
}
prompt.close();
```
```
User: What do you know about LangGraph?
Assistant: I'll search for the latest information about LangGraph for you.
Assistant: [{"title":"Introduction to LangGraph: A Beginner's Guide - Medium","url":"https://medium.com/@cplog/introduction-to-langgraph-a-beginners-guide-14f9be027141","content":"..."}]
Assistant: Based on the search results, I can provide you with information about LangGraph:
LangGraph is a library within the LangChain ecosystem designed for building stateful, multi-actor applications with Large Language Models (LLMs). Here are the key aspects:
**Core Purpose:**
- LangGraph is specifically designed for creating agent and multi-agent workflows
- It provides a framework for defining, coordinating, and executing multiple LLM agents in a structured manner
**Key Features:**
1. **Stateful Graph Architecture**: LangGraph revolves around a stateful graph where each node represents a step in 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. **Cycles**: Unlike other LLM frameworks, LangGraph allows you to define flows that involve cycles, which is essential for most agentic architectures
4. **Controllability**: It offers enhanced control over the application flow
5. **Persistence**: The library provides ways to maintain state and persistence in LLM-based applications
**Use Cases:**
- Conversational agents
- Complex task automation
- Custom LLM-backed experiences
- Multi-agent systems that perform complex tasks
**Benefits:**
LangGraph allows developers to focus on the high-level logic of their applications rather than the intricacies of agent coordination, making it easier to build complex, production-ready features with LLMs.
This makes LangGraph a significant tool in the evolving landscape of LLM-based application development.
```
:::
## 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)
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
<!---
```python
from langchain.chat_models import init_chat_model
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
-->
```python hl_lines="25 30"
from typing import Annotated
@@ -686,46 +327,7 @@ graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()
```
:::
:::js
- `createToolNode` is replaced with the prebuilt [ToolNode](https://langchain-ai.github.io/langgraphjs/reference/classes/langgraph_prebuilt.ToolNode.html)
- `routeTools` is replaced with the prebuilt [toolsCondition](https://langchain-ai.github.io/langgraphjs/reference/functions/langgraph_prebuilt.toolsCondition.html)
```typescript
import { TavilySearch } from "@langchain/tavily";
import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, START, MessagesZodState, END } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const tools = [new TavilySearch({ maxResults: 2 })];
const llm = new ChatOpenAI({ model: "gpt-4o-mini" }).bindTools(tools);
const graph = new StateGraph(State)
.addNode("chatbot", async (state) => ({
messages: [await llm.invoke(state.messages)],
}))
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.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.
:::python
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).
:::
**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
+14 -253
View File
@@ -2,7 +2,7 @@
The chatbot can now [use tools](./2-add-tools.md) to answer user questions, but it does not remember the context of previous interactions. This limits its ability to have coherent, multi-turn conversations.
LangGraph solves this problem through **persistent checkpointing**. If you provide a `checkpointer` when compiling the graph and a `thread_id` when calling your graph, LangGraph automatically saves the state after each step. When you invoke the graph again using the same `thread_id`, the graph loads its saved state, allowing the chatbot to pick up where it left off.
LangGraph solves this problem through **persistent checkpointing**. If you provide a `checkpointer` when compiling the graph and a `thread_id` when calling your graph, LangGraph automatically saves the state after each step. When you invoke the graph again using the same `thread_id`, the graph loads its saved state, allowing the chatbot to pick up where it left off.
We will see later that **checkpointing** is _much_ more powerful than simple chat memory - it lets you save and resume complex state at any time for error recovery, human-in-the-loop workflows, time travel interactions, and more. But first, let's add checkpointing to enable multi-turn conversations.
@@ -14,79 +14,43 @@ We will see later that **checkpointing** is _much_ more powerful than simple cha
Create a `InMemorySaver` checkpointer:
:::python
```python
``` python
from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
```
:::
:::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.
## 2. Compile the graph
Compile the graph with the provided checkpointer, which will checkpoint the `State` as the graph works through each node:
:::python
```python
``` python
graph = graph_builder.compile(checkpointer=memory)
```
:::
``` python
from IPython.display import Image, display
:::js
```typescript hl_lines="7"
const graph = new StateGraph(State)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.compile({ checkpointer: memory });
try:
display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
# This requires some extra dependencies and is optional
pass
```
:::
## 3. Interact with your chatbot
Now you can interact with your bot!
1. Pick a thread to use as the key for this conversation.
:::python
1. Pick a thread to use as the key for this conversation.
```python
config = {"configurable": {"thread_id": "1"}}
```
:::
:::js
```typescript
const config = { configurable: { thread_id: "1" } };
```
:::
2. Call your chatbot:
:::python
2. Call your chatbot:
```python
user_input = "Hi there! My name is Will."
@@ -110,45 +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?
```
!!! note
!!! 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
```typescript
const userInput = "Hi there! My name is Will.";
const events = await graph.stream(
{ messages: [{ type: "human", content: userInput }] },
{ configurable: { thread_id: "1" }, streamMode: "values" }
);
for await (const event of events) {
const lastMessage = event.messages.at(-1);
console.log(`${lastMessage?.getType()}: ${lastMessage?.text}`);
}
```
```
human: Hi there! My name is Will.
ai: 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?
```
!!! note
The config was provided as the **second parameter** 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?"
@@ -171,37 +104,10 @@ Remember my name?
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.
```
:::
:::js
```typescript
const userInput2 = "Remember my name?";
const events2 = await graph.stream(
{ messages: [{ type: "human", content: userInput2 }] },
{ configurable: { thread_id: "1" }, streamMode: "values" }
);
for await (const event of events2) {
const lastMessage = event.messages.at(-1);
console.log(`${lastMessage?.getType()}: ${lastMessage?.text}`);
}
```
```
human: Remember my name?
ai: Yes, your name is Will. How can I help you today?
```
:::
**Notice** that we aren't using an external list for memory: it's all handled by the checkpointer! You can inspect the full execution in this [LangSmith trace](https://smith.langchain.com/public/29ba22b5-6d40-4fbe-8d27-b369e3329c84/r) to see what's going on.
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(
@@ -223,36 +129,10 @@ Remember my name?
I apologize, but I don't have any previous context or memory of your name. As an AI assistant, I don't retain information from past conversations. Each interaction starts fresh. Could you please tell me your name so I can address you properly in this conversation?
```
:::
:::js
```typescript hl_lines="3-4"
const events3 = await graph.stream(
{ messages: [{ type: "human", content: userInput2 }] },
// The only difference is we change the `thread_id` here to "2" instead of "1"
{ configurable: { thread_id: "2" }, streamMode: "values" }
);
for await (const event of events3) {
const lastMessage = event.messages.at(-1);
console.log(`${lastMessage?.getType()}: ${lastMessage?.text}`);
}
```
```
human: Remember my name?
ai: I don't have the ability to remember personal information about users between interactions. However, I'm here to help you with any questions or topics you want to discuss!
```
:::
**Notice** that the **only** change we've made is to modify the `thread_id` in the config. See this call's [LangSmith trace](https://smith.langchain.com/public/51a62351-2f0a-4058-91cc-9996c5561428/r) for comparison.
## 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)`.
```python
@@ -268,94 +148,12 @@ 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
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)`.
```typescript
await graph.getState({ configurable: { thread_id: "1" } });
```
```typescript
{
values: {
messages: [
HumanMessage {
"id": "32fabcef-b3b8-481f-8bcb-fd83399a5f8d",
"content": "Hi there! My name is Will.",
"additional_kwargs": {},
"response_metadata": {}
},
AIMessage {
"id": "chatcmpl-BrPbTsCJbVqBvXWySlYoTJvM75Kv8",
"content": "Hello Will! How can I assist you today?",
"additional_kwargs": {},
"response_metadata": {},
"tool_calls": [],
"invalid_tool_calls": []
},
HumanMessage {
"id": "561c3aad-f8fc-4fac-94a6-54269a220856",
"content": "Remember my name?",
"additional_kwargs": {},
"response_metadata": {}
},
AIMessage {
"id": "chatcmpl-BrPbU4BhhsUikGbW37hYuF5vvnnE2",
"content": "Yes, I remember your name, Will! How can I help you today?",
"additional_kwargs": {},
"response_metadata": {},
"tool_calls": [],
"invalid_tool_calls": []
}
]
},
next: [],
tasks: [],
metadata: {
source: 'loop',
step: 4,
parents: {},
thread_id: '1'
},
config: {
configurable: {
thread_id: '1',
checkpoint_id: '1f05cccc-9bb6-6270-8004-1d2108bcec77',
checkpoint_ns: ''
}
},
createdAt: '2025-07-09T13:58:27.607Z',
parentConfig: {
configurable: {
thread_id: '1',
checkpoint_ns: '',
checkpoint_id: '1f05cccc-78fa-68d0-8003-ffb01a76b599'
}
}
}
```
```typescript
import * as assert from "node:assert";
// 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)
assert.deepEqual(snapshot.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.
**Congratulations!** Your chatbot can now maintain conversation state across sessions thanks to LangGraph's checkpointing system. This opens up exciting possibilities for more natural, contextual interactions. LangGraph's checkpointing even handles **arbitrarily complex graph states**, which is much more expressive and powerful than simple chat memory.
Check out the code snippet below to review the graph from this tutorial:
:::python
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
<!---
@@ -406,43 +204,6 @@ memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript hl_lines="16 26"
import { END, MessagesZodState, START } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { TavilySearch } from "@langchain/tavily";
import { MemorySaver } from "@langchain/langgraph";
import { StateGraph } from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { z } from "zod";
const State = z.object({
messages: MessagesZodState.shape.messages,
});
const tools = [new TavilySearch({ maxResults: 2 })];
const llm = new ChatOpenAI({ model: "gpt-4o-mini" }).bindTools(tools);
// highlight-next-line
const memory = new MemorySaver();
const graph = new StateGraph(State)
.addNode("chatbot", async (state) => ({
messages: [await llm.invoke(state.messages)],
}))
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
// highlight-next-line
.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.
@@ -2,15 +2,7 @@
Agents can be unreliable and may need human input to successfully accomplish tasks. Similarly, for some actions, you may want to require human approval before running to ensure that everything is running as intended.
LangGraph's [persistence](../../concepts/persistence.md) layer supports **human-in-the-loop** workflows, allowing execution to pause and resume based on user feedback. The primary interface to this functionality is the [`interrupt`](../../how-tos/human_in_the_loop/add-human-in-the-loop.md) function. Calling `interrupt` inside a node will pause execution. Execution can be resumed, together with new input from a human, by passing in a [Command](../../concepts/low_level.md#command).
:::python
`interrupt` is ergonomically similar to Python's built-in `input()`, [with some caveats](../../how-tos/human_in_the_loop/add-human-in-the-loop.md).
:::
:::js
`interrupt` is ergonomically similar to Node.js's built-in `readline.question()` function, [with some caveats](../../how-tos/human_in_the_loop/add-human-in-the-loop.md).
:::
LangGraph's [persistence](../../concepts/persistence.md) layer supports **human-in-the-loop** workflows, allowing execution to pause and resume based on user feedback. The primary interface to this functionality is the [`interrupt`](../../how-tos/human_in_the_loop/add-human-in-the-loop.md) function. Calling `interrupt` inside a node will pause execution. Execution can be resumed, together with new input from a human, by passing in a [Command](../../concepts/low_level.md#command). `interrupt` is ergonomically similar to Python's built-in `input()`, [with some caveats](../../how-tos/human_in_the_loop/add-human-in-the-loop.md).
!!! note
@@ -22,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
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
<!---
@@ -33,24 +24,9 @@ llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")
```
-->
:::
:::js
```typescript
// Add your API key here
process.env.ANTHROPIC_API_KEY = "YOUR_API_KEY";
```
:::
We can now incorporate it into our `StateGraph` with an additional tool:
:::python
````python hl_lines="12 19 20 21 22 23"
```python hl_lines="12 19 20 21 22 23"
``` python hl_lines="12 19 20 21 22 23"
from typing import Annotated
from langchain_tavily import TavilySearch
@@ -98,103 +74,7 @@ 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 { TavilySearchResults } from "@langchain/community/tools/tavily_search";
```typescript hl_lines="1 7-19"
import { interrupt, MessagesZodState } from "@langchain/langgraph";
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearch } from "@langchain/tavily";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { MemorySaver } from "@langchain/langgraph";
import {
StateGraph,
START,
END,
MessagesAnnotation,
} from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { ChatAnthropic } from "@langchain/anthropic";
import { Command, interrupt } from "@langchain/langgraph";
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human"),
}),
}
);
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human"),
}),
}
);
const searchTool = new TavilySearch({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
}).bindTools(tools);
async function chatbot(state: z.infer<typeof MessagesZodState>) {
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 with interrupts");
}
return { messages: [message] };
}
const graphBuilder = new StateGraph(MessagesAnnotation).addNode(
"chatbot",
chatbot
);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
const shouldContinue = (state: typeof MessagesAnnotation.State) => {
const messages = state.messages;
const lastMessage = messages[messages.length - 1];
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
return "tools";
}
return END;
};
graphBuilder.addConditionalEdges("chatbot", shouldContinue);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
````
:::
```
!!! tip
@@ -204,39 +84,17 @@ graphBuilder.addEdge(START, "chatbot");
We compile the graph with a checkpointer, as before:
:::python
```python
memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
const memory = new MemorySaver();
const graph = new StateGraph(MessagesZodState)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.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
``` python
from IPython.display import Image, display
try:
@@ -246,30 +104,12 @@ except Exception:
pass
```
:::
:::js
```typescript
import * as fs from "node:fs/promises";
const drawableGraph = await graph.getGraphAsync();
const image = await drawableGraph.drawMermaidPng();
const imageBuffer = new Uint8Array(await image.arrayBuffer());
await fs.writeFile("chatbot-with-tools.png", imageBuffer);
```
:::
![chatbot-with-tools-diagram](chatbot-with-tools.png)
## 4. Prompt the chatbot
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"}}
@@ -298,60 +138,8 @@ Tool Calls:
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 = await graph.stream(
{ messages: [{ role: "user", content: userInput }] },
{ configurable: { thread_id: "1" }, streamMode: "values" }
);
for await (const event of events) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(`[${lastMessage?.getType()}]: ${lastMessage?.text}`);
if (
lastMessage &&
isAIMessage(lastMessage) &&
lastMessage.tool_calls?.length
) {
console.log("Tool calls:", lastMessage.tool_calls);
}
}
}
```
```
[human]: I need some expert guidance for building an AI agent. Could you request assistance for me?
[ai]: I'll help you request human assistance for guidance on building an AI agent.
Tool calls: [
{
name: 'humanAssistance',
args: {
query: 'I would like expert guidance on building an AI agent. Could you please provide assistance with this topic?'
},
id: 'toolu_01Bpxc8rFVMhSaRosS6b85Ts',
type: 'tool_call'
}
]
```
:::
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
@@ -361,28 +149,8 @@ snapshot.next
('tools',)
```
:::
:::js
```typescript
const snapshot = await graph.getState({ configurable: { thread_id: "1" } });
snapshot.next;
```
```json
["tools"]
```
['tools']
````
:::
!!! info Additional information
:::python
Take a closer look at the `human_assistance` tool:
```python
@@ -394,40 +162,12 @@ 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
Take a closer look at the `humanAssistance` tool:
```typescript hl_lines="3"
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human"),
}),
},
);
```
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.
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
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."
" It's much more reliable and extensible than simple autonomous agents."
@@ -439,7 +179,7 @@ events = graph.stream(human_command, config, stream_mode="values")
for event in events:
if "messages" in event:
event["messages"][-1].pretty_print()
````
```
```
================================== Ai Message ==================================
@@ -475,58 +215,13 @@ If you'd like more specific information about LangGraph or have any questions ab
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
:::
:::js
For this example, use an object with a key `"data"`:
```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 = await graph.stream(humanCommand, {
configurable: { thread_id: "1" },
streamMode: "values",
});
for await (const event of resumeEvents) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(`[${lastMessage?.getType()}]: ${lastMessage?.text}`);
}
}
```
```
[tool]: 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]: 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.
**Congratulations!** You've used an `interrupt` to add human-in-the-loop execution to your chatbot, allowing for human oversight and intervention when needed. This opens up the potential UIs you can create with your AI systems. Since you have already added a **checkpointer**, as long as the underlying persistence layer is running, the graph can be paused **indefinitely** and resumed at any time as if nothing had happened.
Check out the code snippet below to review the graph from this tutorial:
:::python
{!snippets/chat_model_tabs.md!}
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
```python
from typing import Annotated
@@ -577,94 +272,6 @@ memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
import {
interrupt,
MessagesZodState,
StateGraph,
MemorySaver,
START,
END,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { isAIMessage } from "@langchain/core/messages";
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearch } from "@langchain/tavily";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const humanAssistance = tool(
async ({ query }) => {
const humanResponse = interrupt({ query });
return humanResponse.data;
},
{
name: "humanAssistance",
description: "Request assistance from a human.",
schema: z.object({
query: z.string().describe("Human readable question for the human"),
}),
}
);
const searchTool = new TavilySearch({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
}).bindTools(tools);
const chatbot = async (state: z.infer<typeof MessagesZodState>) => {
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 with interrupts");
}
return { messages: message };
};
const graphBuilder = new StateGraph(MessagesAnnotation).addNode(
"chatbot",
chatbot
);
const toolNode = new ToolNode(tools);
graphBuilder.addNode("tools", toolNode);
const shouldContinue = (state: typeof MessagesAnnotation.State) => {
const messages = state.messages;
const lastMessage = messages[messages.length - 1];
if ("tool_calls" in lastMessage && lastMessage.tool_calls?.length) {
return "tools";
}
return END;
};
graphBuilder.addConditionalEdges("chatbot", shouldContinue);
graphBuilder.addEdge("tools", "chatbot");
graphBuilder.addEdge(START, "chatbot");
const memory = new MemorySaver();
const graph = new StateGraph(MessagesZodState)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.compile({ checkpointer: memory });
```
:::
## Next steps
So far, the tutorial examples have relied on a simple state with one entry: a list of messages. You can go far with this simple state, but if you want to define complex behavior without relying on the message list, you can [add additional fields to the state](./5-customize-state.md).
So far, the tutorial examples have relied on a simple state with one entry: a list of messages. You can go far with this simple state, but if you want to define complex behavior without relying on the message list, you can [add additional fields to the state](./5-customize-state.md).
@@ -10,8 +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
@@ -28,34 +26,13 @@ class State(TypedDict):
birthday: str
```
:::
:::js
```typescript
import { MessagesZodState } from "@langchain/langgraph";
import { z } from "zod";
const State = z.object({
messages: MessagesZodState.shape.messages,
// highlight-next-line
name: z.string(),
// highlight-next-line
birthday: z.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
``` python
from langchain_core.messages import ToolMessage
from langchain_core.tools import InjectedToolCallId, tool
@@ -99,78 +76,10 @@ def human_assistance(
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 { Command, interrupt } from "@langchain/langgraph";
const humanAssistance = tool(
async (input, config) => {
// Note that because we are generating a ToolMessage for a state update,
// we generally require the ID of the corresponding tool call.
// This is available in the tool's config.
const toolCallId = config?.toolCall?.id as string | undefined;
if (!toolCallId) throw new Error("Tool call ID is required");
const humanResponse = await interrupt({
question: "Is this correct?",
name: input.name,
birthday: input.birthday,
});
// We explicitly update the state with a ToolMessage inside the tool.
const stateUpdate = (() => {
// If the information is correct, update the state as-is.
if (humanResponse.correct?.toLowerCase().startsWith("y")) {
return {
name: input.name,
birthday: input.birthday,
messages: [
new ToolMessage({ content: "Correct", tool_call_id: toolCallId }),
],
};
}
// Otherwise, receive information from the human reviewer.
return {
name: humanResponse.name || input.name,
birthday: humanResponse.birthday || input.birthday,
messages: [
new ToolMessage({
content: `Made a correction: ${JSON.stringify(humanResponse)}`,
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().describe("The name of the entity"),
birthday: z.string().describe("The birthday/release date of the entity"),
}),
}
);
```
:::
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
@@ -190,51 +99,6 @@ for event in events:
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
import { isAIMessage } from "@langchain/core/messages";
const userInput =
"Can you look up when LangGraph was released? " +
"When you have the answer, use the humanAssistance tool for review.";
const events = await graph.stream(
{ messages: [{ role: "user", content: userInput }] },
{ configurable: { thread_id: "1" }, streamMode: "values" }
);
for await (const event of events) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(
"=".repeat(32),
`${lastMessage?.getType()} Message`,
"=".repeat(32)
);
console.log(lastMessage?.text);
if (
lastMessage &&
isAIMessage(lastMessage) &&
lastMessage.tool_calls?.length
) {
console.log("Tool Calls:");
for (const call of lastMessage.tool_calls) {
console.log(` ${call.name} (${call.id})`);
console.log(` Args: ${JSON.stringify(call.args)}`);
}
}
}
}
```
:::
```
================================ Human Message =================================
@@ -262,20 +126,12 @@ Tool Calls:
birthday: 2023-01-01
```
:::python
We've hit the `interrupt` in the `human_assistance` tool again.
:::
:::js
We've hit the `interrupt` in the `humanAssistance` tool again.
:::
## 4. Add human assistance
The chatbot failed to identify the correct date, so supply it with information:
:::python
```python
human_command = Command(
resume={
@@ -290,53 +146,6 @@ for event in events:
event["messages"][-1].pretty_print()
```
:::
:::js
```typescript
import { Command } from "@langchain/langgraph";
const humanCommand = new Command({
resume: {
name: "LangGraph",
birthday: "Jan 17, 2024",
},
});
const resumeEvents = await graph.stream(humanCommand, {
configurable: { thread_id: "1" },
streamMode: "values",
});
for await (const event of resumeEvents) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(
"=".repeat(32),
`${lastMessage?.getType()} Message`,
"=".repeat(32)
);
console.log(lastMessage?.text);
if (
lastMessage &&
isAIMessage(lastMessage) &&
lastMessage.tool_calls?.length
) {
console.log("Tool Calls:");
for (const call of lastMessage.tool_calls) {
console.log(` ${call.name} (${call.id})`);
console.log(` Args: ${JSON.stringify(call.args)}`);
}
}
}
}
```
:::
```
================================== Ai Message ==================================
@@ -366,8 +175,6 @@ 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)
@@ -378,34 +185,13 @@ snapshot = graph.get_state(config)
{'name': 'LangGraph', 'birthday': 'Jan 17, 2024'}
```
:::
:::js
```typescript
const snapshot = await graph.getState(config);
const relevantState = Object.fromEntries(
Object.entries(snapshot.values).filter(([k]) =>
["name", "birthday"].includes(k)
)
);
```
```
{ name: 'LangGraph', birthday: 'Jan 17, 2024' }
```
:::
This makes them easily accessible to downstream nodes (e.g., a node that further processes or stores the information).
## 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
``` python
graph.update_state(config, {"name": "LangGraph (library)"})
```
@@ -415,36 +201,11 @@ graph.update_state(config, {"name": "LangGraph (library)"})
'checkpoint_id': '1efd4ec5-cf69-6352-8006-9278f1730162'}}
```
:::
:::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(
{ configurable: { thread_id: "1" } },
{ name: "LangGraph (library)" }
);
```
```typescript
{
configurable: {
thread_id: '1',
checkpoint_ns: '',
checkpoint_id: '1efd4ec5-cf69-6352-8006-9278f1730162'
}
}
```
:::
## 6. View the new value
:::python
If you call `graph.get_state`, you can see the new value is reflected:
```python
``` python
snapshot = graph.get_state(config)
{k: v for k, v in snapshot.values.items() if k in ("name", "birthday")}
@@ -454,35 +215,12 @@ snapshot = graph.get_state(config)
{'name': 'LangGraph (library)', 'birthday': 'Jan 17, 2024'}
```
:::
:::js
If you call `graph.getState`, you can see the new value is reflected:
```typescript
const updatedSnapshot = await graph.getState(config);
const updatedRelevantState = Object.fromEntries(
Object.entries(updatedSnapshot.values).filter(([k]) =>
["name", "birthday"].includes(k)
)
);
```
```typescript
{ name: 'LangGraph (library)', birthday: 'Jan 17, 2024' }
```
:::
Manual state updates will [generate a trace](https://smith.langchain.com/public/7ebb7827-378d-49fe-9f6c-5df0e90086c8/r) in LangSmith. If desired, they can also be used to [control human-in-the-loop workflows](../../how-tos/human_in_the_loop/add-human-in-the-loop.md). Use of the `interrupt` function is generally recommended instead, as it allows data to be transmitted in a human-in-the-loop interaction independently of state updates.
**Congratulations!** You've added custom keys to the state to facilitate a more complex workflow, and learned how to generate state updates from inside tools.
Check out the code snippet below to review the graph from this tutorial:
:::python
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
<!---
@@ -567,111 +305,7 @@ memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
import {
Command,
interrupt,
MessagesZodState,
MemorySaver,
StateGraph,
END,
START,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { ChatAnthropic } from "@langchain/anthropic";
import { TavilySearch } from "@langchain/tavily";
import { ToolMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
const State = z.object({
messages: MessagesZodState.shape.messages,
name: z.string(),
birthday: z.string(),
});
const humanAssistance = tool(
async (input, config) => {
// Note that because we are generating a ToolMessage for a state update, we
// generally require the ID of the corresponding tool call. This is available
// in the tool's config.
const toolCallId = config?.toolCall?.id as string | undefined;
if (!toolCallId) throw new Error("Tool call ID is required");
const humanResponse = await interrupt({
question: "Is this correct?",
name: input.name,
birthday: input.birthday,
});
// We explicitly update the state with a ToolMessage inside the tool.
const stateUpdate = (() => {
// If the information is correct, update the state as-is.
if (humanResponse.correct?.toLowerCase().startsWith("y")) {
return {
name: input.name,
birthday: input.birthday,
messages: [
new ToolMessage({ content: "Correct", tool_call_id: toolCallId }),
],
};
}
// Otherwise, receive information from the human reviewer.
return {
name: humanResponse.name || input.name,
birthday: humanResponse.birthday || input.birthday,
messages: [
new ToolMessage({
content: `Made a correction: ${JSON.stringify(humanResponse)}`,
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().describe("The name of the entity"),
birthday: z.string().describe("The birthday/release date of the entity"),
}),
}
);
const searchTool = new TavilySearch({ maxResults: 2 });
const tools = [searchTool, humanAssistance];
const llmWithTools = new ChatAnthropic({
model: "claude-3-5-sonnet-latest",
}).bindTools(tools);
const memory = new MemorySaver();
const chatbot = async (state: z.infer<typeof State>) => {
const message = await llmWithTools.invoke(state.messages);
return { messages: message };
};
const graph = new StateGraph(State)
.addNode("chatbot", chatbot)
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.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).
+17 -341
View File
@@ -4,7 +4,7 @@ In a typical chatbot workflow, the user interacts with the bot one or more times
What if you want a user to be able to start from a previous response and explore a different outcome? Or what if you want users to be able to rewind your chatbot's work to fix mistakes or try a different strategy, something that is common in applications like autonomous software engineers?
You can create these types of experiences using LangGraph's built-in **time travel** functionality.
You can create these types of experiences using LangGraph's built-in **time travel** functionality.
!!! note
@@ -12,15 +12,7 @@ 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.
:::
:::python
{% include-markdown "../../../snippets/chat_model_tabs.md" %}
@@ -72,49 +64,11 @@ memory = InMemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
:::
:::js
```typescript
import {
StateGraph,
START,
END,
MessagesZodState,
MemorySaver,
} from "@langchain/langgraph";
import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { TavilySearch } from "@langchain/tavily";
import { ChatOpenAI } from "@langchain/openai";
import { z } from "zod";
const State = z.object({ messages: MessagesZodState.shape.messages });
const tools = [new TavilySearch({ maxResults: 2 })];
const llmWithTools = new ChatOpenAI({ model: "gpt-4o-mini" }).bindTools(tools);
const memory = new MemorySaver();
const graph = new StateGraph(State)
.addNode("chatbot", async (state) => ({
messages: [await llmWithTools.invoke(state.messages)],
}))
.addNode("tools", new ToolNode(tools))
.addConditionalEdges("chatbot", toolsCondition, ["tools", END])
.addEdge("tools", "chatbot")
.addEdge(START, "chatbot")
.compile({ checkpointer: memory });
```
:::
## 2. Add steps
Add steps to your graph. Every step will be checkpointed in its state history:
:::python
```python
``` python
config = {"configurable": {"thread_id": "1"}}
events = graph.stream(
{
@@ -205,7 +159,7 @@ Tool Calls:
================================= Tool Message =================================
Name: tavily_search_results_json
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a user's question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a users question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
================================== Ai Message ==================================
Great idea! Building an autonomous agent with LangGraph is definitely an exciting project. Based on the latest information I've found, here are some insights and tips for building autonomous agents with LangGraph:
@@ -223,140 +177,11 @@ Building an autonomous agent is an iterative process, so be prepared to refine a
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
:::
:::js
```typescript
import { randomUUID } from "node:crypto";
const threadId = randomUUID();
let iter = 0;
for (const userInput of [
"I'm learning LangGraph. Could you do some research on it for me?",
"Ya that's helpful. Maybe I'll build an autonomous agent with it!",
]) {
iter += 1;
console.log(`\n--- Conversation Turn ${iter} ---\n`);
const events = await graph.stream(
{ messages: [{ role: "user", content: userInput }] },
{ configurable: { thread_id: threadId }, streamMode: "values" }
);
for await (const event of events) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(
"=".repeat(32),
`${lastMessage?.getType()} Message`,
"=".repeat(32)
);
console.log(lastMessage?.text);
}
}
}
```
```
--- Conversation Turn 1 ---
================================ human Message ================================
I'm learning LangGraph.js. Could you do some research on it for me?
================================ ai Message ================================
I'll search for information about LangGraph.js for you.
================================ tool Message ================================
{
"query": "LangGraph.js framework TypeScript langchain what is it tutorial guide",
"follow_up_questions": null,
"answer": null,
"images": [],
"results": [
{
"url": "https://techcommunity.microsoft.com/blog/educatordeveloperblog/an-absolute-beginners-guide-to-langgraph-js/4212496",
"title": "An Absolute Beginner's Guide to LangGraph.js",
"content": "(...)",
"score": 0.79369855,
"raw_content": null
},
{
"url": "https://langchain-ai.github.io/langgraphjs/",
"title": "LangGraph.js",
"content": "(...)",
"score": 0.78154784,
"raw_content": null
}
],
"response_time": 2.37
}
================================ ai Message ================================
Let me provide you with an overview of LangGraph.js based on the search results:
LangGraph.js is a JavaScript/TypeScript library that's part of the LangChain ecosystem, specifically designed for creating and managing complex LLM (Large Language Model) based workflows. Here are the key points about LangGraph.js:
1. Purpose:
- It's a low-level orchestration framework for building controllable agents
- Particularly useful for creating agentic workflows where LLMs decide the course of action based on current state
- Helps model workflows as graphs with nodes and edges
(...)
--- Conversation Turn 2 ---
================================ human Message ================================
Ya that's helpful. Maybe I'll build an autonomous agent with it!
================================ ai Message ================================
Let me search for specific information about building autonomous agents with LangGraph.js.
================================ tool Message ================================
{
"query": "how to build autonomous agents with LangGraph.js examples tutorial react agent",
"follow_up_questions": null,
"answer": null,
"images": [],
"results": [
{
"url": "https://ai.google.dev/gemini-api/docs/langgraph-example",
"title": "ReAct agent from scratch with Gemini 2.5 and LangGraph",
"content": "(...)",
"score": 0.7602419,
"raw_content": null
},
{
"url": "https://www.youtube.com/watch?v=ZfjaIshGkmk",
"title": "Build Autonomous AI Agents with ReAct and LangGraph Tools",
"content": "(...)",
"score": 0.7471924,
"raw_content": null
}
],
"response_time": 1.98
}
================================ ai Message ================================
Based on the search results, I can provide you with a practical overview of how to build an autonomous agent with LangGraph.js. Here's what you need to know:
1. Basic Structure for Building an Agent:
- LangGraph.js provides a ReAct (Reason + Act) pattern implementation
- The basic components include:
- State management for conversation history
- Nodes for different actions
- Edges for decision-making flow
- Tools for specific functionalities
(...)
```
:::
## 3. Replay the full state history
Now that you have added steps to the chatbot, you can `replay` the full state history to see everything that occurred.
:::python
```python
``` python
to_replay = None
for state in graph.get_state_history(config):
print("Num Messages: ", len(state.values["messages"]), "Next: ", state.next)
@@ -389,61 +214,10 @@ Num Messages: 0 Next: ('__start__',)
--------------------------------------------------------------------------------
```
:::
:::js
```typescript
import type { StateSnapshot } from "@langchain/langgraph";
let toReplay: StateSnapshot | undefined;
for await (const state of graph.getStateHistory({
configurable: { thread_id: threadId },
})) {
console.log(
`Num Messages: ${state.values.messages.length}, Next: ${JSON.stringify(
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;
}
}
```
```
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.
Checkpoints are saved for every step of the graph. This __spans invocations__ so you can rewind across a full thread's history.
## Resume from a checkpoint
:::python
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
@@ -456,37 +230,12 @@ print(to_replay.config)
{'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1efd43e3-0c1f-6c4e-8006-891877d65740'}}
```
:::
:::js
Resume from the `toReplay` state, which is after the `chatbot` node in one of the graph invocations. Resuming from this point will call the next scheduled node.
```typescript
console.log(toReplay.next);
console.log(toReplay.config);
```
```
["tools"]
{
configurable: {
thread_id: "007708b8-ea9b-4ff7-a7ad-3843364dbf75",
checkpoint_ns: "",
checkpoint_id: "1efd43e3-0c1f-6c4e-8006-891877d65740"
}
}
```
:::
## 4. Load a state from a moment-in-time
:::python
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:
@@ -505,16 +254,19 @@ Tool Calls:
================================= Tool Message =================================
Name: tavily_search_results_json
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a user's question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
[{"url": "https://towardsdatascience.com/building-autonomous-multi-tool-agents-with-gemini-2-0-and-langgraph-ad3d7bd5e79d", "content": "Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph | by Youness Mansar | Jan, 2025 | Towards Data Science Building Autonomous Multi-Tool Agents with Gemini 2.0 and LangGraph A practical tutorial with full code examples for building and running multi-tool agents Towards Data Science LLMs are remarkable — they can memorize vast amounts of information, answer general knowledge questions, write code, generate stories, and even fix your grammar. In this tutorial, we are going to build a simple LLM agent that is equipped with four tools that it can use to answer a users question. This Agent will have the following specifications: Follow Published in Towards Data Science --------------------------------- Your home for data science and AI. Follow Follow Follow"}, {"url": "https://github.com/anmolaman20/Tools_and_Agents", "content": "GitHub - anmolaman20/Tools_and_Agents: This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository provides resources for building AI agents using Langchain and Langgraph. This repository serves as a comprehensive guide for building AI-powered agents using Langchain and Langgraph. It provides hands-on examples, practical tutorials, and resources for developers and AI enthusiasts to master building intelligent systems and workflows. AI Agent Development: Gain insights into creating intelligent systems that think, reason, and adapt in real time. This repository is ideal for AI practitioners, developers exploring language models, or anyone interested in building intelligent systems. This repository provides resources for building AI agents using Langchain and Langgraph."}]
================================== Ai Message ==================================
Great idea! Building an autonomous agent with LangGraph is definitely an exciting project. Based on the latest information I've found, here are some insights and tips for building autonomous agents with LangGraph:
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. Multi-Tool Agents: LangGraph is particularly well-suited for creating autonomous agents that can use multiple tools. This allows your agent to have a diverse set of capabilities and choose the right tool for each task.
1. Multi-Tool Agents:
LangGraph is well-suited for building autonomous agents that can use multiple tools. This allows your agent to have a variety of capabilities and choose the appropriate tool based on the task at hand.
2. Integration with Large Language Models (LLMs): You can combine LangGraph with powerful LLMs like Gemini 2.0 to create more intelligent and capable agents. The LLM can serve as the "brain" of your agent, making decisions and generating responses.
2. Integration with Large Language Models (LLMs):
There's a tutorial that specifically mentions using Gemini 2.0 (Google's LLM) with LangGraph to build autonomous agents. This suggests that LangGraph can be integrated with various LLMs, giving you flexibility in choosing the language model that best fits your needs.
3. Workflow Management: LangGraph excels at managing complex, multi-step AI workflows. This is crucial for autonomous agents that need to break down tasks into smaller steps and execute them in the right order.
3. Practical Tutorials:
There are tutorials available that provide full code examples for building and running multi-tool agents. These can be invaluable as you start your project, giving you a concrete starting point and demonstrating best practices.
...
Remember, building an autonomous agent is an iterative process. Start simple and gradually increase complexity as you become more comfortable with LangGraph and its capabilities.
@@ -523,83 +275,7 @@ Would you like more information on any specific aspect of building your autonomo
Output is truncated. View as a scrollable element or open in a text editor. Adjust cell output settings...
```
The graph resumed execution from the `tools` node. You can tell this is the case since the first value printed above is the response from our search engine tool.
:::
:::js
The checkpoint's `toReplay.config` contains a `checkpoint_id` timestamp. Providing this `checkpoint_id` value tells LangGraph's checkpointer to **load** the state from that moment in time.
```typescript
// The `checkpoint_id` in the `toReplay.config` corresponds to a state we've persisted to our checkpointer.
for await (const event of await graph.stream(null, {
...toReplay?.config,
streamMode: "values",
})) {
if ("messages" in event) {
const lastMessage = event.messages.at(-1);
console.log(
"=".repeat(32),
`${lastMessage?.getType()} Message`,
"=".repeat(32)
);
console.log(lastMessage?.text);
}
}
```
```
================================ ai Message ================================
Let me search for specific information about building autonomous agents with LangGraph.js.
================================ tool Message ================================
{
"query": "how to build autonomous agents with LangGraph.js examples tutorial",
"follow_up_questions": null,
"answer": null,
"images": [],
"results": [
{
"url": "https://www.mongodb.com/developer/languages/typescript/build-javascript-ai-agent-langgraphjs-mongodb/",
"title": "Build a JavaScript AI Agent With LangGraph.js and MongoDB",
"content": "(...)",
"score": 0.7672197,
"raw_content": null
},
{
"url": "https://medium.com/@lorevanoudenhove/how-to-build-ai-agents-with-langgraph-a-step-by-step-guide-5d84d9c7e832",
"title": "How to Build AI Agents with LangGraph: A Step-by-Step Guide",
"content": "(...)",
"score": 0.7407191,
"raw_content": null
}
],
"response_time": 0.82
}
================================ ai Message ================================
Based on the search results, I can share some practical information about building autonomous agents with LangGraph.js. Here are some concrete examples and approaches:
1. Example HR Assistant Agent:
- Can handle HR-related queries using employee information
- Features include:
- Starting and continuing conversations
- Looking up information using vector search
- Persisting conversation state using checkpoints
- Managing threaded conversations
2. Energy Savings Calculator Agent:
- Functions as a lead generation tool for solar panel sales
- Capabilities include:
- Calculating potential energy savings
- Handling multi-step conversations
- Processing user inputs for personalized estimates
- Managing conversation state
(...)
```
The graph resumed execution from the `tools` node. You can tell this is the case since the first value printed above is the response from our search engine tool.
:::
The graph resumed execution from the `action` node. You can tell this is the case since the first value printed above is the response from our search engine tool.
**Congratulations!** You've now used time-travel checkpoint traversal in LangGraph. Being able to rewind and explore alternative paths opens up a world of possibilities for debugging, experimentation, and interactive applications.
@@ -609,4 +285,4 @@ Take your LangGraph journey further by exploring deployment and advanced feature
- **[LangGraph Server quickstart](../../tutorials/langgraph-platform/local-server.md)**: Launch a LangGraph server locally and interact with it using the REST API and LangGraph Studio Web UI.
- **[LangGraph Platform quickstart](../../cloud/quick_start.md)**: Deploy your LangGraph app using LangGraph Platform.
- **[LangGraph Platform concepts](../../concepts/langgraph_platform.md)**: Understand the foundational concepts of the LangGraph Platform.
- **[LangGraph Platform concepts](../../concepts/langgraph_platform.md)**: Understand the foundational concepts of the LangGraph Platform.
+3 -112
View File
@@ -152,14 +152,6 @@ base64-js@^1.5.1:
resolved "https://registry.yarnpkg.com/base64-js/-/base64-js-1.5.1.tgz#1b1b440160a5bf7ad40b650f095963481903930a"
integrity sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA==
call-bind-apply-helpers@^1.0.1, call-bind-apply-helpers@^1.0.2:
version "1.0.2"
resolved "https://registry.yarnpkg.com/call-bind-apply-helpers/-/call-bind-apply-helpers-1.0.2.tgz#4b5428c222be985d79c3d82657479dbe0b59b2d6"
integrity sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ==
dependencies:
es-errors "^1.3.0"
function-bind "^1.1.2"
camelcase@6:
version "6.3.0"
resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-6.3.0.tgz#5685b95eb209ac9c0c177467778c9c84df58ba9a"
@@ -209,42 +201,6 @@ delayed-stream@~1.0.0:
resolved "https://registry.yarnpkg.com/delayed-stream/-/delayed-stream-1.0.0.tgz#df3ae199acadfb7d440aaae0b29e2272b24ec619"
integrity sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==
dunder-proto@^1.0.1:
version "1.0.1"
resolved "https://registry.yarnpkg.com/dunder-proto/-/dunder-proto-1.0.1.tgz#d7ae667e1dc83482f8b70fd0f6eefc50da30f58a"
integrity sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==
dependencies:
call-bind-apply-helpers "^1.0.1"
es-errors "^1.3.0"
gopd "^1.2.0"
es-define-property@^1.0.1:
version "1.0.1"
resolved "https://registry.yarnpkg.com/es-define-property/-/es-define-property-1.0.1.tgz#983eb2f9a6724e9303f61addf011c72e09e0b0fa"
integrity sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g==
es-errors@^1.3.0:
version "1.3.0"
resolved "https://registry.yarnpkg.com/es-errors/-/es-errors-1.3.0.tgz#05f75a25dab98e4fb1dcd5e1472c0546d5057c8f"
integrity sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw==
es-object-atoms@^1.0.0, es-object-atoms@^1.1.1:
version "1.1.1"
resolved "https://registry.yarnpkg.com/es-object-atoms/-/es-object-atoms-1.1.1.tgz#1c4f2c4837327597ce69d2ca190a7fdd172338c1"
integrity sha512-FGgH2h8zKNim9ljj7dankFPcICIK9Cp5bm+c2gQSYePhpaG5+esrLODihIorn+Pe6FGJzWhXQotPv73jTaldXA==
dependencies:
es-errors "^1.3.0"
es-set-tostringtag@^2.1.0:
version "2.1.0"
resolved "https://registry.yarnpkg.com/es-set-tostringtag/-/es-set-tostringtag-2.1.0.tgz#f31dbbe0c183b00a6d26eb6325c810c0fd18bd4d"
integrity sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==
dependencies:
es-errors "^1.3.0"
get-intrinsic "^1.2.6"
has-tostringtag "^1.0.2"
hasown "^2.0.2"
event-lite@^0.1.1:
version "0.1.3"
resolved "https://registry.yarnpkg.com/event-lite/-/event-lite-0.1.3.tgz#3dfe01144e808ac46448f0c19b4ab68e403a901d"
@@ -266,14 +222,12 @@ form-data-encoder@1.7.2:
integrity sha512-qfqtYan3rxrnCk1VYaA4H+Ms9xdpPqvLZa6xmMgFvhO32x7/3J/ExcTd6qpxM0vH2GdMI+poehyBZvqfMTto8A==
form-data@^4.0.0:
version "4.0.4"
resolved "https://registry.yarnpkg.com/form-data/-/form-data-4.0.4.tgz#784cdcce0669a9d68e94d11ac4eea98088edd2c4"
integrity sha512-KrGhL9Q4zjj0kiUt5OO4Mr/A/jlI2jDYs5eHBpYHPcBEVSiipAvn2Ko2HnPe20rmcuuvMHNdZFp+4IlGTMF0Ow==
version "4.0.1"
resolved "https://registry.yarnpkg.com/form-data/-/form-data-4.0.1.tgz#ba1076daaaa5bfd7e99c1a6cb02aa0a5cff90d48"
integrity sha512-tzN8e4TX8+kkxGPK8D5u0FNmjPUjw3lwC9lSLxxoB/+GtsJG91CO8bSWy73APlgAZzZbXEYZJuxjkHH2w+Ezhw==
dependencies:
asynckit "^0.4.0"
combined-stream "^1.0.8"
es-set-tostringtag "^2.1.0"
hasown "^2.0.2"
mime-types "^2.1.12"
formdata-node@^4.3.2:
@@ -284,69 +238,11 @@ formdata-node@^4.3.2:
node-domexception "1.0.0"
web-streams-polyfill "4.0.0-beta.3"
function-bind@^1.1.2:
version "1.1.2"
resolved "https://registry.yarnpkg.com/function-bind/-/function-bind-1.1.2.tgz#2c02d864d97f3ea6c8830c464cbd11ab6eab7a1c"
integrity sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA==
get-intrinsic@^1.2.6:
version "1.3.0"
resolved "https://registry.yarnpkg.com/get-intrinsic/-/get-intrinsic-1.3.0.tgz#743f0e3b6964a93a5491ed1bffaae054d7f98d01"
integrity sha512-9fSjSaos/fRIVIp+xSJlE6lfwhES7LNtKaCBIamHsjr2na1BiABJPo0mOjjz8GJDURarmCPGqaiVg5mfjb98CQ==
dependencies:
call-bind-apply-helpers "^1.0.2"
es-define-property "^1.0.1"
es-errors "^1.3.0"
es-object-atoms "^1.1.1"
function-bind "^1.1.2"
get-proto "^1.0.1"
gopd "^1.2.0"
has-symbols "^1.1.0"
hasown "^2.0.2"
math-intrinsics "^1.1.0"
get-proto@^1.0.1:
version "1.0.1"
resolved "https://registry.yarnpkg.com/get-proto/-/get-proto-1.0.1.tgz#150b3f2743869ef3e851ec0c49d15b1d14d00ee1"
integrity sha512-sTSfBjoXBp89JvIKIefqw7U2CCebsc74kiY6awiGogKtoSGbgjYE/G/+l9sF3MWFPNc9IcoOC4ODfKHfxFmp0g==
dependencies:
dunder-proto "^1.0.1"
es-object-atoms "^1.0.0"
gopd@^1.2.0:
version "1.2.0"
resolved "https://registry.yarnpkg.com/gopd/-/gopd-1.2.0.tgz#89f56b8217bdbc8802bd299df6d7f1081d7e51a1"
integrity sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==
has-flag@^4.0.0:
version "4.0.0"
resolved "https://registry.yarnpkg.com/has-flag/-/has-flag-4.0.0.tgz#944771fd9c81c81265c4d6941860da06bb59479b"
integrity sha512-EykJT/Q1KjTWctppgIAgfSO0tKVuZUjhgMr17kqTumMl6Afv3EISleU7qZUzoXDFTAHTDC4NOoG/ZxU3EvlMPQ==
has-symbols@^1.0.3, has-symbols@^1.1.0:
version "1.1.0"
resolved "https://registry.yarnpkg.com/has-symbols/-/has-symbols-1.1.0.tgz#fc9c6a783a084951d0b971fe1018de813707a338"
integrity sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ==
has-tostringtag@^1.0.2:
version "1.0.2"
resolved "https://registry.yarnpkg.com/has-tostringtag/-/has-tostringtag-1.0.2.tgz#2cdc42d40bef2e5b4eeab7c01a73c54ce7ab5abc"
integrity sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==
dependencies:
has-symbols "^1.0.3"
hasown@^2.0.2:
version "2.0.2"
resolved "https://registry.yarnpkg.com/hasown/-/hasown-2.0.2.tgz#003eaf91be7adc372e84ec59dc37252cedb80003"
integrity sha512-0hJU9SCPvmMzIBdZFqNPXWa6dqh7WdH0cII9y+CyS8rG3nL48Bclra9HmKhVVUHyPWNH5Y7xDwAB7bfgSjkUMQ==
dependencies:
function-bind "^1.1.2"
he@^1.2.0:
version "1.2.0"
resolved "https://registry.yarnpkg.com/he/-/he-1.2.0.tgz#84ae65fa7eafb165fddb61566ae14baf05664f0f"
integrity sha512-F/1DnUGPopORZi0ni+CvrCgHQ5FyEAHRLSApuYWMmrbSwoN2Mn/7k+Gl38gJnR7yyDZk6WLXwiGod1JOWNDKGw==
humanize-ms@^1.2.1:
version "1.2.1"
resolved "https://registry.yarnpkg.com/humanize-ms/-/humanize-ms-1.2.1.tgz#c46e3159a293f6b896da29316d8b6fe8bb79bbed"
@@ -399,11 +295,6 @@ json-stringify-safe@^5.0.1:
semver "^7.6.3"
uuid "^10.0.0"
math-intrinsics@^1.1.0:
version "1.1.0"
resolved "https://registry.yarnpkg.com/math-intrinsics/-/math-intrinsics-1.1.0.tgz#a0dd74be81e2aa5c2f27e65ce283605ee4e2b7f9"
integrity sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==
mime-db@1.52.0:
version "1.52.0"
resolved "https://registry.yarnpkg.com/mime-db/-/mime-db-1.52.0.tgz#bbabcdc02859f4987301c856e3387ce5ec43bf70"
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph-checkpoint-sqlite"
version = "2.0.11"
version = "2.0.10"
description = "Library with a SQLite implementation of LangGraph checkpoint saver."
authors = []
requires-python = ">=3.9"
+1 -1
View File
@@ -346,7 +346,7 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.11"
version = "2.0.10"
source = { editable = "." }
dependencies = [
{ name = "aiosqlite" },
@@ -1 +0,0 @@
"""Legacy utilities module, to be removed in v1."""
-4
View File
@@ -1,4 +0,0 @@
"""Backwards compat imports for config utilities, to be removed in v1."""
from langgraph._internal._config import ensure_config, patch_configurable # noqa: F401
from langgraph.config import get_config, get_store # noqa: F401
@@ -1,3 +0,0 @@
"""Backwards compat imports for runnable utilities, to be removed in v1."""
from langgraph._internal._runnable import RunnableCallable, RunnableLike # noqa: F401
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph"
version = "0.6.0"
version = "0.6.0a1"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
requires-python = ">=3.9"
View File
+2
View File
@@ -0,0 +1,2 @@
# import for backwards compatibility
from langgraph._internal._runnable import RunnableCallable, RunnableSeq # noqa: F401
+2 -2
View File
@@ -1192,7 +1192,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.6.0"
version = "0.6.0a1"
source = { editable = "." }
dependencies = [
{ name = "langchain-core" },
@@ -1364,7 +1364,7 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.11"
version = "2.0.10"
source = { editable = "../checkpoint-sqlite" }
dependencies = [
{ name = "aiosqlite" },
@@ -1,7 +1,6 @@
import inspect
from typing import (
Any,
Awaitable,
Callable,
Literal,
Optional,
@@ -45,10 +44,8 @@ from langgraph.graph.state import CompiledStateGraph
from langgraph.managed import IsLastStep, RemainingSteps
from langgraph.prebuilt._internal import ToolCallWithContext
from langgraph.prebuilt.tool_node import ToolNode
from langgraph.runtime import Runtime
from langgraph.store.base import BaseStore
from langgraph.types import Checkpointer, Send
from langgraph.typing import ContextT
from langgraph.warnings import LangGraphDeprecatedSinceV10
StructuredResponse = Union[dict, BaseModel]
@@ -248,13 +245,437 @@ def _validate_chat_history(
raise ValueError(error_message)
class _AgentBuilder:
"""Internal builder class for constructing React agents with intuitive method-to-node mapping."""
def __init__(
self,
model: Union[str, LanguageModelLike],
tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
*,
prompt: Optional[Prompt] = None,
response_format: Optional[
Union[StructuredResponseSchema, tuple[str, StructuredResponseSchema]]
] = None,
pre_model_hook: Optional[RunnableLike] = None,
post_model_hook: Optional[RunnableLike] = None,
state_schema: Optional[StateSchemaType] = None,
context_schema: Optional[Type[Any]] = None,
version: Literal["v1", "v2"] = "v2",
name: Optional[str] = None,
):
# Store all parameters
self.model = model
self.tools = tools
self.prompt = prompt
self.response_format = response_format
self.pre_model_hook = pre_model_hook
self.post_model_hook = post_model_hook
self.state_schema = state_schema
self.context_schema = context_schema
self.version = version
self.name = name
# Setup tools
if isinstance(self.tools, ToolNode):
self._tool_classes = list(self.tools.tools_by_name.values())
self._tool_node = self.tools
else:
self._llm_builtin_tools = [t for t in self.tools if isinstance(t, dict)]
self._tool_node = ToolNode(
[t for t in self.tools if not isinstance(t, dict)]
)
self._tool_classes = list(self._tool_node.tools_by_name.values())
self._should_return_direct: set[str] = {
t.name for t in self._tool_classes if t.return_direct
}
# Setup state schema
if self.state_schema is not None:
required_keys = {"messages", "remaining_steps"}
if self.response_format is not None:
required_keys.add("structured_response")
schema_keys = set(get_type_hints(self.state_schema))
if missing_keys := required_keys - schema_keys:
raise ValueError(
f"Missing required key(s) {missing_keys} in state_schema"
)
self._final_state_schema = self.state_schema
else:
self._final_state_schema = (
AgentStateWithStructuredResponse
if self.response_format is not None
else AgentState
)
# Setup model
model = self.model
# Convert string models
if isinstance(model, str):
try:
from langchain.chat_models import init_chat_model # type: ignore[import-not-found]
except ImportError:
raise ImportError(
"Please install langchain (`pip install langchain`) to use '<provider>:<model>' string syntax for `model` parameter."
)
model = cast(BaseChatModel, init_chat_model(model))
# Bind tools if needed
if (
_should_bind_tools(
model, self._tool_classes, num_builtin=len(self._llm_builtin_tools)
)
and len(self._tool_classes + self._llm_builtin_tools) > 0
):
model = cast(BaseChatModel, model).bind_tools(
self._tool_classes + self._llm_builtin_tools
) # type: ignore[operator]
self._model_runnable = _get_prompt_runnable(self.prompt) | model
def create_model_node(self) -> RunnableCallable:
"""Create the 'agent' node that calls the LLM."""
def _get_model_input_state(state: StateSchema) -> StateSchema:
if self.pre_model_hook is not None:
messages: Optional[Sequence[BaseMessage]] = (
_get_state_value(state, "llm_input_messages")
) or _get_state_value(state, "messages")
error_msg: str = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
else:
messages = _get_state_value(state, "messages")
error_msg = f"Expected input to call_model to have 'messages' key, but got {state}"
if messages is None:
raise ValueError(error_msg)
_validate_chat_history(messages)
if isinstance(self._final_state_schema, type) and issubclass(
self._final_state_schema, BaseModel
):
state.messages = messages # type: ignore
else:
state["messages"] = messages # type: ignore
return state
def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool:
has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
all_tools_return_direct = (
all(
call["name"] in self._should_return_direct
for call in response.tool_calls
)
if isinstance(response, AIMessage)
else False
)
remaining_steps = _get_state_value(state, "remaining_steps", None)
is_last_step = _get_state_value(state, "is_last_step", False)
return (
(remaining_steps is None and is_last_step and has_tool_calls)
or (
remaining_steps is not None
and remaining_steps < 1
and all_tools_return_direct
)
or (
remaining_steps is not None
and remaining_steps < 2
and has_tool_calls
)
)
def call_model(state: StateSchema, config: RunnableConfig) -> StateSchema:
state = _get_model_input_state(state)
response = cast(AIMessage, self._model_runnable.invoke(state, config)) # type: ignore[union-attr]
response.name = self.name
if _are_more_steps_needed(state, response):
return {
"messages": [
AIMessage(
id=response.id,
content="Sorry, need more steps to process this request.",
)
]
}
return {"messages": [response]}
async def acall_model(
state: StateSchema, config: RunnableConfig
) -> StateSchema:
state = _get_model_input_state(state)
response = cast(
AIMessage, await self._model_runnable.ainvoke(state, config)
) # type: ignore[union-attr]
response.name = self.name
if _are_more_steps_needed(state, response):
return {
"messages": [
AIMessage(
id=response.id,
content="Sorry, need more steps to process this request.",
)
]
}
return {"messages": [response]}
# Determine input schema
input_schema = self._final_state_schema
if self.pre_model_hook is not None:
if isinstance(self._final_state_schema, type) and issubclass(
self._final_state_schema, BaseModel
):
from pydantic import create_model
input_schema = create_model(
"CallModelInputSchema",
llm_input_messages=(list[AnyMessage], ...),
__base__=self._final_state_schema,
)
else:
class CallModelInputSchema(self._final_state_schema): # type: ignore
llm_input_messages: list[AnyMessage]
input_schema = CallModelInputSchema
return RunnableCallable(call_model, acall_model, input_schema=input_schema)
def create_structured_response_node(self) -> Optional[RunnableCallable]:
"""Create the 'generate_structured_response' node if configured."""
if self.response_format is None:
return None
def generate_structured_response(
state: StateSchema, config: RunnableConfig
) -> StateSchema:
messages = _get_state_value(state, "messages")
structured_response_schema = self.response_format
if isinstance(self.response_format, tuple):
system_prompt, structured_response_schema = self.response_format
messages = [SystemMessage(content=system_prompt)] + list(messages)
model_with_structured_output = _get_model(
self._model_runnable
).with_structured_output( # type: ignore[arg-type]
cast(StructuredResponseSchema, structured_response_schema)
)
response = model_with_structured_output.invoke(messages, config)
return {"structured_response": response}
async def agenerate_structured_response(
state: StateSchema, config: RunnableConfig
) -> StateSchema:
messages = _get_state_value(state, "messages")
structured_response_schema = self.response_format
if isinstance(self.response_format, tuple):
system_prompt, structured_response_schema = self.response_format
messages = [SystemMessage(content=system_prompt)] + list(messages)
model_with_structured_output = _get_model(
self._model_runnable
).with_structured_output( # type: ignore[arg-type]
cast(StructuredResponseSchema, structured_response_schema)
)
response = await model_with_structured_output.ainvoke(messages, config)
return {"structured_response": response}
return RunnableCallable(
generate_structured_response, agenerate_structured_response
)
def create_model_router(self) -> Callable[[StateSchema], Union[str, list[Send]]]:
"""Create routing function for model node conditional edges."""
def should_continue(state: StateSchema) -> Union[str, list[Send]]:
messages = _get_state_value(state, "messages")
last_message = messages[-1]
if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
if self.post_model_hook is not None:
return "post_model_hook"
elif self.response_format is not None:
return "generate_structured_response"
else:
return END
else:
if self.version == "v1":
return "tools"
elif self.version == "v2":
if self.post_model_hook is not None:
return "post_model_hook"
return [
Send(
"tools",
ToolCallWithContext(
__type="tool_call_with_context",
tool_call=tool_call,
state=state,
),
)
for tool_call in last_message.tool_calls
]
return should_continue
def post_model_hook_router(self, state: StateSchema) -> Union[str, list[Send]]:
"""Route to the next node after post_model_hook."""
messages = _get_state_value(state, "messages")
tool_messages = [m.tool_call_id for m in messages if isinstance(m, ToolMessage)]
last_ai_message = next(
m for m in reversed(messages) if isinstance(m, AIMessage)
)
pending_tool_calls = [
c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
]
if pending_tool_calls:
return [
Send(
"tools",
ToolCallWithContext(
__type="tool_call_with_context",
tool_call=tool_call,
state=state,
),
)
for tool_call in pending_tool_calls
]
elif isinstance(messages[-1], ToolMessage):
return self._get_entry_point()
elif self.response_format is not None:
return "generate_structured_response"
else:
return END
def create_tools_router(self) -> Optional[Callable[[StateSchema], str]]:
"""Create routing function for tools node conditional edges."""
if not self._should_return_direct:
return None
def route_tool_responses(state: StateSchema) -> str:
messages = _get_state_value(state, "messages")
for m in reversed(messages):
if not isinstance(m, ToolMessage):
break
if m.name in self._should_return_direct:
return END
if isinstance(m, AIMessage) and m.tool_calls:
if any(
call["name"] in self._should_return_direct for call in m.tool_calls
):
return END
return self._get_entry_point()
return route_tool_responses
def _get_entry_point(self) -> str:
"""Get the workflow entry point."""
return "pre_model_hook" if self.pre_model_hook else "agent"
def _has_tools(self) -> bool:
"""Check if agent has tools enabled."""
return len(self._tool_classes) > 0
def _get_model_edges(self) -> list[str]:
"""Get possible edge destinations from model node."""
edges = []
# If post_model_hook exists, we don't add edges here - we use direct edge instead
if not self.post_model_hook:
if self._has_tools():
edges.append("tools")
if self.response_format:
edges.append("generate_structured_response")
if not self._has_tools() and not self.response_format:
edges.append(END)
return edges
def _get_post_model_hook_edges(self) -> list[str]:
"""Get possible edge destinations from post_model_hook node."""
edges = [self._get_entry_point()]
if self._has_tools():
edges.append("tools")
if self.response_format:
edges.append("generate_structured_response")
else:
edges.append(END)
return edges
def build(self) -> StateGraph:
"""Build the agent workflow graph (uncompiled)."""
# Create workflow
workflow = StateGraph(
state_schema=self._final_state_schema, # type: ignore[arg-type]
context_schema=self.context_schema,
)
# Add nodes
# Always add model node (named 'agent' for backwards compatibility)
workflow.add_node("agent", self.create_model_node())
# Add tools node if needed
if self._has_tools():
workflow.add_node("tools", self._tool_node)
# Add hook nodes if configured
if self.pre_model_hook:
workflow.add_node("pre_model_hook", self.pre_model_hook) # type: ignore[arg-type]
if self.post_model_hook:
workflow.add_node("post_model_hook", self.post_model_hook) # type: ignore[arg-type]
# Add structured response node if configured
structured_node = self.create_structured_response_node()
if structured_node:
workflow.add_node("generate_structured_response", structured_node)
# Add edges
entry_point = self._get_entry_point()
workflow.set_entry_point(entry_point)
# Pre-model hook edge
if self.pre_model_hook:
workflow.add_edge("pre_model_hook", "agent")
# Model node edges
if self.post_model_hook:
# Direct edge from model node to post_model_hook when post_model_hook exists
workflow.add_edge("agent", "post_model_hook")
# Post-model hook conditional edges
post_hook_edges = self._get_post_model_hook_edges()
workflow.add_conditional_edges(
"post_model_hook", self.post_model_hook_router, path_map=post_hook_edges
) # type: ignore[arg-type]
else:
# Conditional edges from model node when no post_model_hook
model_router = self.create_model_router()
model_edges = self._get_model_edges()
workflow.add_conditional_edges("agent", model_router, path_map=model_edges) # type: ignore[arg-type]
# Tools edges
if self._has_tools():
tools_router = self.create_tools_router()
if tools_router:
workflow.add_conditional_edges(
"tools", tools_router, path_map=[entry_point, END]
)
else:
workflow.add_edge("tools", entry_point)
return workflow
def create_react_agent(
model: Union[
str,
LanguageModelLike,
Callable[[StateSchema, Runtime[ContextT]], BaseChatModel],
Callable[[StateSchema, Runtime[ContextT]], Awaitable[BaseChatModel]],
],
model: Union[str, LanguageModelLike],
tools: Union[Sequence[Union[BaseTool, Callable, dict[str, Any]]], ToolNode],
*,
prompt: Optional[Prompt] = None,
@@ -279,43 +700,7 @@ def create_react_agent(
For more details on using `create_react_agent`, visit [Agents](https://langchain-ai.github.io/langgraph/agents/overview/) documentation.
Args:
model: The language model for the agent. Supports static and dynamic
model selection.
- **Static model**: A chat model instance (e.g., `ChatOpenAI()`) or
string identifier (e.g., `"openai:gpt-4"`)
- **Dynamic model**: A callable with signature
`(state, runtime) -> BaseChatModel` that returns different models
based on runtime context
Dynamic functions receive graph state and runtime, enabling
context-dependent model selection. Must return a `BaseChatModel`
instance. For tool calling, bind tools using `.bind_tools()`.
Bound tools must be a subset of the `tools` parameter.
Dynamic model example:
```python
from dataclasses import dataclass
@dataclass
class ModelContext:
model_name: str = "gpt-3.5-turbo"
# Instantiate models globally
gpt4_model = ChatOpenAI(model="gpt-4")
gpt35_model = ChatOpenAI(model="gpt-3.5-turbo")
def select_model(state: AgentState, runtime: Runtime[ModelContext]) -> ChatOpenAI:
model_name = runtime.context.model_name
model = gpt4_model if model_name == "gpt-4" else gpt35_model
return model.bind_tools(tools)
```
!!! note "Dynamic Model Requirements"
Ensure returned models have appropriate tools bound via
`.bind_tools()` and support required functionality. Bound tools
must be a subset of those specified in the `tools` parameter.
model: The `LangChain` chat model that supports tool calling.
tools: A list of tools or a ToolNode instance.
If an empty list is provided, the agent will consist of a single LLM node without tool calling.
prompt: An optional prompt for the LLM. Can take a few different forms:
@@ -455,6 +840,7 @@ def create_react_agent(
print(chunk)
```
"""
# Handle deprecated config_schema parameter
if (
config_schema := deprecated_kwargs.pop("config_schema", MISSING)
) is not MISSING:
@@ -466,471 +852,29 @@ def create_react_agent(
if context_schema is not None:
context_schema = config_schema
# Validate version
if version not in ("v1", "v2"):
raise ValueError(
f"Invalid version {version}. Supported versions are 'v1' and 'v2'."
)
if state_schema is not None:
required_keys = {"messages", "remaining_steps"}
if response_format is not None:
required_keys.add("structured_response")
schema_keys = set(get_type_hints(state_schema))
if missing_keys := required_keys - set(schema_keys):
raise ValueError(f"Missing required key(s) {missing_keys} in state_schema")
if state_schema is None:
state_schema = (
AgentStateWithStructuredResponse
if response_format is not None
else AgentState
)
llm_builtin_tools: list[dict] = []
if isinstance(tools, ToolNode):
tool_classes = list(tools.tools_by_name.values())
tool_node = tools
else:
llm_builtin_tools = [t for t in tools if isinstance(t, dict)]
tool_node = ToolNode([t for t in tools if not isinstance(t, dict)])
tool_classes = list(tool_node.tools_by_name.values())
is_dynamic_model = not isinstance(model, (str, Runnable)) and callable(model)
is_async_dynamic_model = is_dynamic_model and inspect.iscoroutinefunction(model)
tool_calling_enabled = len(tool_classes) > 0
if not is_dynamic_model:
if isinstance(model, str):
try:
from langchain.chat_models import ( # type: ignore[import-not-found]
init_chat_model,
)
except ImportError:
raise ImportError(
"Please install langchain (`pip install langchain`) to "
"use '<provider>:<model>' string syntax for `model` parameter."
)
model = cast(BaseChatModel, init_chat_model(model))
if (
_should_bind_tools(model, tool_classes, num_builtin=len(llm_builtin_tools)) # type: ignore[arg-type]
and len(tool_classes + llm_builtin_tools) > 0
):
model = cast(BaseChatModel, model).bind_tools(
tool_classes + llm_builtin_tools # type: ignore[operator]
)
static_model: Optional[Runnable] = _get_prompt_runnable(prompt) | model # type: ignore[operator]
else:
# For dynamic models, we'll create the runnable at runtime
static_model = None
# If any of the tools are configured to return_directly after running,
# our graph needs to check if these were called
should_return_direct = {t.name for t in tool_classes if t.return_direct}
def _resolve_model(
state: StateSchema, runtime: Runtime[ContextT]
) -> LanguageModelLike:
"""Resolve the model to use, handling both static and dynamic models."""
if is_dynamic_model:
return _get_prompt_runnable(prompt) | model(state, runtime) # type: ignore[operator]
else:
return static_model
async def _aresolve_model(
state: StateSchema, runtime: Runtime[ContextT]
) -> LanguageModelLike:
"""Async resolve the model to use, handling both static and dynamic models."""
if is_async_dynamic_model:
resolved_model = await model(state, runtime) # type: ignore[misc,operator]
return _get_prompt_runnable(prompt) | resolved_model
elif is_dynamic_model:
return _get_prompt_runnable(prompt) | model(state, runtime) # type: ignore[operator]
else:
return static_model
def _are_more_steps_needed(state: StateSchema, response: BaseMessage) -> bool:
has_tool_calls = isinstance(response, AIMessage) and response.tool_calls
all_tools_return_direct = (
all(call["name"] in should_return_direct for call in response.tool_calls)
if isinstance(response, AIMessage)
else False
)
remaining_steps = _get_state_value(state, "remaining_steps", None)
is_last_step = _get_state_value(state, "is_last_step", False)
return (
(remaining_steps is None and is_last_step and has_tool_calls)
or (
remaining_steps is not None
and remaining_steps < 1
and all_tools_return_direct
)
or (remaining_steps is not None and remaining_steps < 2 and has_tool_calls)
)
def _get_model_input_state(state: StateSchema) -> StateSchema:
if pre_model_hook is not None:
messages = (
_get_state_value(state, "llm_input_messages")
) or _get_state_value(state, "messages")
error_msg = f"Expected input to call_model to have 'llm_input_messages' or 'messages' key, but got {state}"
else:
messages = _get_state_value(state, "messages")
error_msg = (
f"Expected input to call_model to have 'messages' key, but got {state}"
)
if messages is None:
raise ValueError(error_msg)
_validate_chat_history(messages)
# we're passing messages under `messages` key, as this is expected by the prompt
if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
state.messages = messages # type: ignore
else:
state["messages"] = messages # type: ignore
return state
# Define the function that calls the model
def call_model(
state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
) -> StateSchema:
if is_async_dynamic_model:
msg = (
"Async model callable provided but agent invoked synchronously. "
"Use agent.ainvoke() or agent.astream(), or "
"provide a sync model callable."
)
raise RuntimeError(msg)
model_input = _get_model_input_state(state)
if is_dynamic_model:
# Resolve dynamic model at runtime and apply prompt
dynamic_model = _resolve_model(state, runtime)
response = cast(AIMessage, dynamic_model.invoke(model_input, config)) # type: ignore[arg-type]
else:
response = cast(AIMessage, static_model.invoke(model_input, config)) # type: ignore[union-attr]
# add agent name to the AIMessage
response.name = name
if _are_more_steps_needed(state, response):
return {
"messages": [
AIMessage(
id=response.id,
content="Sorry, need more steps to process this request.",
)
]
}
# We return a list, because this will get added to the existing list
return {"messages": [response]}
async def acall_model(
state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
) -> StateSchema:
model_input = _get_model_input_state(state)
if is_dynamic_model:
# Resolve dynamic model at runtime and apply prompt
# (supports both sync and async)
dynamic_model = await _aresolve_model(state, runtime)
response = cast(AIMessage, await dynamic_model.ainvoke(model_input, config)) # type: ignore[arg-type]
else:
response = cast(AIMessage, await static_model.ainvoke(model_input, config)) # type: ignore[union-attr]
# add agent name to the AIMessage
response.name = name
if _are_more_steps_needed(state, response):
return {
"messages": [
AIMessage(
id=response.id,
content="Sorry, need more steps to process this request.",
)
]
}
# We return a list, because this will get added to the existing list
return {"messages": [response]}
input_schema: StateSchemaType
if pre_model_hook is not None:
# Dynamically create a schema that inherits from state_schema and adds 'llm_input_messages'
if isinstance(state_schema, type) and issubclass(state_schema, BaseModel):
# For Pydantic schemas
from pydantic import create_model
input_schema = create_model(
"CallModelInputSchema",
llm_input_messages=(list[AnyMessage], ...),
__base__=state_schema,
)
else:
# For TypedDict schemas
class CallModelInputSchema(state_schema): # type: ignore
llm_input_messages: list[AnyMessage]
input_schema = CallModelInputSchema
else:
input_schema = state_schema
def generate_structured_response(
state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
) -> StateSchema:
if is_async_dynamic_model:
msg = (
"Async model callable provided but agent invoked synchronously. "
"Use agent.ainvoke() or agent.astream(), or provide a sync model callable."
)
raise RuntimeError(msg)
messages = _get_state_value(state, "messages")
structured_response_schema = response_format
if isinstance(response_format, tuple):
system_prompt, structured_response_schema = response_format
messages = [SystemMessage(content=system_prompt)] + list(messages)
resolved_model = _resolve_model(state, runtime)
model_with_structured_output = _get_model(
resolved_model
).with_structured_output(
cast(StructuredResponseSchema, structured_response_schema)
)
response = model_with_structured_output.invoke(messages, config)
return {"structured_response": response}
async def agenerate_structured_response(
state: StateSchema, runtime: Runtime[ContextT], config: RunnableConfig
) -> StateSchema:
messages = _get_state_value(state, "messages")
structured_response_schema = response_format
if isinstance(response_format, tuple):
system_prompt, structured_response_schema = response_format
messages = [SystemMessage(content=system_prompt)] + list(messages)
resolved_model = await _aresolve_model(state, runtime)
model_with_structured_output = _get_model(
resolved_model
).with_structured_output(
cast(StructuredResponseSchema, structured_response_schema)
)
response = await model_with_structured_output.ainvoke(messages, config)
return {"structured_response": response}
if not tool_calling_enabled:
# Define a new graph
workflow = StateGraph(state_schema=state_schema, context_schema=context_schema)
workflow.add_node(
"agent",
RunnableCallable(call_model, acall_model),
input_schema=input_schema,
)
if pre_model_hook is not None:
workflow.add_node("pre_model_hook", pre_model_hook) # type: ignore[arg-type]
workflow.add_edge("pre_model_hook", "agent")
entrypoint = "pre_model_hook"
else:
entrypoint = "agent"
workflow.set_entry_point(entrypoint)
if post_model_hook is not None:
workflow.add_node("post_model_hook", post_model_hook) # type: ignore[arg-type]
workflow.add_edge("agent", "post_model_hook")
if response_format is not None:
workflow.add_node(
"generate_structured_response",
RunnableCallable(
generate_structured_response,
agenerate_structured_response,
),
)
if post_model_hook is not None:
workflow.add_edge("post_model_hook", "generate_structured_response")
else:
workflow.add_edge("agent", "generate_structured_response")
return workflow.compile(
checkpointer=checkpointer,
store=store,
interrupt_before=interrupt_before,
interrupt_after=interrupt_after,
debug=debug,
name=name,
)
# Define the function that determines whether to continue or not
def should_continue(state: StateSchema) -> Union[str, list[Send]]:
messages = _get_state_value(state, "messages")
last_message = messages[-1]
# If there is no function call, then we finish
if not isinstance(last_message, AIMessage) or not last_message.tool_calls:
if post_model_hook is not None:
return "post_model_hook"
elif response_format is not None:
return "generate_structured_response"
else:
return END
# Otherwise if there is, we continue
else:
if version == "v1":
return "tools"
elif version == "v2":
if post_model_hook is not None:
return "post_model_hook"
return [
Send(
"tools",
ToolCallWithContext(
__type="tool_call_with_context",
tool_call=tool_call,
state=state,
),
)
for tool_call in last_message.tool_calls
]
# Define a new graph
workflow = StateGraph(
state_schema=state_schema or AgentState, context_schema=context_schema
# Build the graph using the internal builder
builder = _AgentBuilder(
model=model,
tools=tools,
prompt=prompt,
response_format=response_format,
pre_model_hook=pre_model_hook,
post_model_hook=post_model_hook,
state_schema=state_schema,
context_schema=context_schema,
version=version,
name=name,
)
# Define the two nodes we will cycle between
workflow.add_node(
"agent",
RunnableCallable(call_model, acall_model),
input_schema=input_schema,
)
workflow.add_node("tools", tool_node)
workflow = builder.build()
# Optionally add a pre-model hook node that will be called
# every time before the "agent" (LLM-calling node)
if pre_model_hook is not None:
workflow.add_node("pre_model_hook", pre_model_hook) # type: ignore[arg-type]
workflow.add_edge("pre_model_hook", "agent")
entrypoint = "pre_model_hook"
else:
entrypoint = "agent"
# Set the entrypoint as `agent`
# This means that this node is the first one called
workflow.set_entry_point(entrypoint)
agent_paths = []
post_model_hook_paths = [entrypoint, "tools"]
# Add a post model hook node if post_model_hook is provided
if post_model_hook is not None:
workflow.add_node("post_model_hook", post_model_hook) # type: ignore[arg-type]
agent_paths.append("post_model_hook")
workflow.add_edge("agent", "post_model_hook")
else:
agent_paths.append("tools")
# Add a structured output node if response_format is provided
if response_format is not None:
workflow.add_node(
"generate_structured_response",
RunnableCallable(
generate_structured_response,
agenerate_structured_response,
),
)
if post_model_hook is not None:
post_model_hook_paths.append("generate_structured_response")
else:
agent_paths.append("generate_structured_response")
else:
if post_model_hook is not None:
post_model_hook_paths.append(END)
else:
agent_paths.append(END)
if post_model_hook is not None:
def post_model_hook_router(state: StateSchema) -> Union[str, list[Send]]:
"""Route to the next node after post_model_hook.
Routes to one of:
* "tools": if there are pending tool calls without a corresponding message.
* "generate_structured_response": if no pending tool calls exist and response_format is specified.
* END: if no pending tool calls exist and no response_format is specified.
"""
messages = _get_state_value(state, "messages")
tool_messages = [
m.tool_call_id for m in messages if isinstance(m, ToolMessage)
]
last_ai_message = next(
m for m in reversed(messages) if isinstance(m, AIMessage)
)
pending_tool_calls = [
c for c in last_ai_message.tool_calls if c["id"] not in tool_messages
]
if pending_tool_calls:
return [
Send(
"tools",
ToolCallWithContext(
__type="tool_call_with_context",
tool_call=tool_call,
state=state,
),
)
for tool_call in pending_tool_calls
]
elif isinstance(messages[-1], ToolMessage):
return entrypoint
elif response_format is not None:
return "generate_structured_response"
else:
return END
workflow.add_conditional_edges(
"post_model_hook",
post_model_hook_router, # type: ignore[arg-type]
path_map=post_model_hook_paths,
)
workflow.add_conditional_edges(
"agent",
should_continue, # type: ignore[arg-type]
path_map=agent_paths,
)
def route_tool_responses(state: StateSchema) -> str:
for m in reversed(_get_state_value(state, "messages")):
if not isinstance(m, ToolMessage):
break
if m.name in should_return_direct:
return END
# handle a case of parallel tool calls where
# the tool w/ `return_direct` was executed in a different `Send`
if isinstance(m, AIMessage) and m.tool_calls:
if any(call["name"] in should_return_direct for call in m.tool_calls):
return END
return entrypoint
if should_return_direct:
workflow.add_conditional_edges(
"tools", route_tool_responses, path_map=[entrypoint, END]
)
else:
workflow.add_edge("tools", entrypoint)
# Finally, we compile it!
# This compiles it into a LangChain Runnable,
# meaning you can use it as you would any other runnable
# Compile and return the graph
return workflow.compile(
checkpointer=checkpointer,
store=store,
+1 -374
View File
@@ -13,12 +13,10 @@ from typing import (
)
import pytest
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import (
AIMessage,
AnyMessage,
HumanMessage,
MessageLikeRepresentation,
RemoveMessage,
SystemMessage,
ToolCall,
@@ -54,7 +52,6 @@ from langgraph.prebuilt.tool_node import (
_get_state_args,
_infer_handled_types,
)
from langgraph.runtime import Runtime
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.types import Command, Interrupt, interrupt
@@ -1095,7 +1092,7 @@ def test_inspect_react() -> None:
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_react_with_subgraph_tools(
sync_checkpointer: BaseCheckpointSaver, version: Literal["v1", "v2"]
sync_checkpointer: BaseCheckpointSaver, version: str
) -> None:
class State(TypedDict):
a: int
@@ -1370,376 +1367,6 @@ def test_get_model() -> None:
_get_model(RunnableLambda(lambda message: message))
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_basic(version: str) -> None:
"""Test basic dynamic model functionality."""
def dynamic_model(state, runtime: Runtime):
# Return different models based on state
if "urgent" in state["messages"][-1].content:
return FakeToolCallingModel(tool_calls=[])
else:
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(dynamic_model, [], version=version)
result = agent.invoke({"messages": [HumanMessage("hello")]})
assert len(result["messages"]) == 2
assert result["messages"][-1].content == "hello"
result = agent.invoke({"messages": [HumanMessage("urgent help")]})
assert len(result["messages"]) == 2
assert result["messages"][-1].content == "urgent help"
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_with_tools(version: Literal["v1", "v2"]) -> None:
"""Test dynamic model with tool calling."""
@dec_tool
def basic_tool(x: int) -> str:
"""Basic tool."""
return f"basic: {x}"
@dec_tool
def advanced_tool(x: int) -> str:
"""Advanced tool."""
return f"advanced: {x}"
def dynamic_model(state: dict, runtime: Runtime) -> BaseChatModel:
# Return model with different behaviors based on message content
if "advanced" in state["messages"][-1].content:
return FakeToolCallingModel(
tool_calls=[
[{"args": {"x": 1}, "id": "1", "name": "advanced_tool"}],
[],
]
)
else:
return FakeToolCallingModel(
tool_calls=[[{"args": {"x": 1}, "id": "1", "name": "basic_tool"}], []]
)
agent = create_react_agent(
dynamic_model, [basic_tool, advanced_tool], version=version
)
# Test basic tool usage
result = agent.invoke({"messages": [HumanMessage("basic request")]})
assert len(result["messages"]) == 3
tool_message = result["messages"][-1]
assert tool_message.content == "basic: 1"
assert tool_message.name == "basic_tool"
# Test advanced tool usage
result = agent.invoke({"messages": [HumanMessage("advanced request")]})
assert len(result["messages"]) == 3
tool_message = result["messages"][-1]
assert tool_message.content == "advanced: 1"
assert tool_message.name == "advanced_tool"
@dataclasses.dataclass
class Context:
user_id: str
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_with_context(version: str) -> None:
"""Test dynamic model using config parameters."""
def dynamic_model(state, runtime: Runtime[Context]):
# Use context to determine model behavior
user_id = runtime.context.user_id
if user_id == "user_premium":
return FakeToolCallingModel(tool_calls=[])
else:
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(
dynamic_model, [], context_schema=Context, version=version
)
# Test with basic user
result = agent.invoke(
{"messages": [HumanMessage("hello")]},
context=Context(user_id="user_basic"),
)
assert len(result["messages"]) == 2
# Test with premium user
result = agent.invoke(
{"messages": [HumanMessage("hello")]},
context=Context(user_id="user_premium"),
)
assert len(result["messages"]) == 2
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_with_state_schema(version: Literal["v1", "v2"]) -> None:
"""Test dynamic model with custom state schema."""
class CustomDynamicState(AgentState):
model_preference: str = "default"
def dynamic_model(state: CustomDynamicState, runtime: Runtime) -> BaseChatModel:
# Use custom state field to determine model
if state.get("model_preference") == "advanced":
return FakeToolCallingModel(tool_calls=[])
else:
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(
dynamic_model, [], state_schema=CustomDynamicState, version=version
)
result = agent.invoke(
{"messages": [HumanMessage("hello")], "model_preference": "advanced"}
)
assert len(result["messages"]) == 2
assert result["model_preference"] == "advanced"
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_with_prompt(version: Literal["v1", "v2"]) -> None:
"""Test dynamic model with different prompt types."""
def dynamic_model(state: AgentState, runtime: Runtime) -> BaseChatModel:
return FakeToolCallingModel(tool_calls=[])
# Test with string prompt
agent = create_react_agent(dynamic_model, [], prompt="system_msg", version=version)
result = agent.invoke({"messages": [HumanMessage("human_msg")]})
assert result["messages"][-1].content == "system_msg-human_msg"
# Test with callable prompt
def dynamic_prompt(state: AgentState) -> list[MessageLikeRepresentation]:
"""Generate a dynamic system message based on state."""
return [{"role": "system", "content": "system_msg"}] + list(state["messages"])
agent = create_react_agent(
dynamic_model, [], prompt=dynamic_prompt, version=version
)
result = agent.invoke({"messages": [HumanMessage("human_msg")]})
assert result["messages"][-1].content == "system_msg-human_msg"
async def test_dynamic_model_async() -> None:
"""Test dynamic model with async operations."""
def dynamic_model(state: AgentState, runtime: Runtime) -> BaseChatModel:
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(dynamic_model, [])
result = await agent.ainvoke({"messages": [HumanMessage("hello async")]})
assert len(result["messages"]) == 2
assert result["messages"][-1].content == "hello async"
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_with_structured_response(version: str) -> None:
"""Test dynamic model with structured response format."""
class TestResponse(BaseModel):
message: str
confidence: float
def dynamic_model(state, runtime: Runtime):
expected_response = TestResponse(message="dynamic response", confidence=0.9)
return FakeToolCallingModel(
tool_calls=[], structured_response=expected_response
)
agent = create_react_agent(
dynamic_model, [], response_format=TestResponse, version=version
)
result = agent.invoke({"messages": [HumanMessage("hello")]})
assert "structured_response" in result
assert result["structured_response"].message == "dynamic response"
assert result["structured_response"].confidence == 0.9
def test_dynamic_model_with_checkpointer(sync_checkpointer):
"""Test dynamic model with checkpointer."""
call_count = 0
def dynamic_model(state: AgentState, runtime: Runtime) -> BaseChatModel:
nonlocal call_count
call_count += 1
return FakeToolCallingModel(
tool_calls=[],
# Incrementing the call count as it is used to assign an id
# to the AIMessage.
# The default reducer semantics are to overwrite an existing message
# with the new one if the id matches.
index=call_count,
)
agent = create_react_agent(dynamic_model, [], checkpointer=sync_checkpointer)
config = {"configurable": {"thread_id": "test_dynamic"}}
# First call
result1 = agent.invoke({"messages": [HumanMessage("hello")]}, config)
assert len(result1["messages"]) == 2 # Human + AI message
# Second call - should load from checkpoint
result2 = agent.invoke({"messages": [HumanMessage("world")]}, config)
assert len(result2["messages"]) == 4
# Dynamic model should be called each time
assert call_count >= 2
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_state_dependent_tools(version: Literal["v1", "v2"]) -> None:
"""Test dynamic model that changes available tools based on state."""
@dec_tool
def tool_a(x: int) -> str:
"""Tool A."""
return f"A: {x}"
@dec_tool
def tool_b(x: int) -> str:
"""Tool B."""
return f"B: {x}"
def dynamic_model(state, runtime: Runtime):
# Switch tools based on message history
if any("use_b" in msg.content for msg in state["messages"]):
return FakeToolCallingModel(
tool_calls=[[{"args": {"x": 2}, "id": "1", "name": "tool_b"}], []]
)
else:
return FakeToolCallingModel(
tool_calls=[[{"args": {"x": 1}, "id": "1", "name": "tool_a"}], []]
)
agent = create_react_agent(dynamic_model, [tool_a, tool_b], version=version)
# Ask to use tool B
result = agent.invoke({"messages": [HumanMessage("use_b please")]})
last_message = result["messages"][-1]
assert isinstance(last_message, ToolMessage)
assert last_message.content == "B: 2"
# Ask to use tool A
result = agent.invoke({"messages": [HumanMessage("hello")]})
last_message = result["messages"][-1]
assert isinstance(last_message, ToolMessage)
assert last_message.content == "A: 1"
@pytest.mark.parametrize("version", REACT_TOOL_CALL_VERSIONS)
def test_dynamic_model_error_handling(version: Literal["v1", "v2"]) -> None:
"""Test error handling in dynamic model."""
def failing_dynamic_model(state, runtime: Runtime):
if "fail" in state["messages"][-1].content:
raise ValueError("Dynamic model failed")
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(failing_dynamic_model, [], version=version)
# Normal operation should work
result = agent.invoke({"messages": [HumanMessage("hello")]})
assert len(result["messages"]) == 2
# Should propagate the error
with pytest.raises(ValueError, match="Dynamic model failed"):
agent.invoke({"messages": [HumanMessage("fail now")]})
def test_dynamic_model_vs_static_model_behavior():
"""Test that dynamic and static models produce equivalent results when configured the same."""
# Static model
static_model = FakeToolCallingModel(tool_calls=[])
static_agent = create_react_agent(static_model, [])
# Dynamic model returning the same model
def dynamic_model(state, runtime: Runtime):
return FakeToolCallingModel(tool_calls=[])
dynamic_agent = create_react_agent(dynamic_model, [])
input_msg = {"messages": [HumanMessage("test message")]}
static_result = static_agent.invoke(input_msg)
dynamic_result = dynamic_agent.invoke(input_msg)
# Results should be equivalent (content-wise, IDs may differ)
assert len(static_result["messages"]) == len(dynamic_result["messages"])
assert static_result["messages"][0].content == dynamic_result["messages"][0].content
assert static_result["messages"][1].content == dynamic_result["messages"][1].content
def test_dynamic_model_receives_correct_state():
"""Test that the dynamic model function receives the correct state, not the model input."""
received_states = []
class CustomAgentState(AgentState):
custom_field: str
def dynamic_model(state, runtime: Runtime) -> BaseChatModel:
# Capture the state that's passed to the dynamic model function
received_states.append(state)
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(dynamic_model, [], state_schema=CustomAgentState)
# Test with initial state
input_state = {"messages": [HumanMessage("hello")], "custom_field": "test_value"}
agent.invoke(input_state)
# The dynamic model function should receive the original state, not the processed model input
assert len(received_states) == 1
received_state = received_states[0]
# Should have the custom field from original state
assert "custom_field" in received_state
assert received_state["custom_field"] == "test_value"
# Should have the original messages
assert len(received_state["messages"]) == 1
assert received_state["messages"][0].content == "hello"
async def test_dynamic_model_receives_correct_state_async():
"""Test that the async dynamic model function receives the correct state, not the model input."""
received_states = []
class CustomAgentStateAsync(AgentState):
custom_field: str
def dynamic_model(state, runtime: Runtime):
# Capture the state that's passed to the dynamic model function
received_states.append(state)
return FakeToolCallingModel(tool_calls=[])
agent = create_react_agent(dynamic_model, [], state_schema=CustomAgentStateAsync)
# Test with initial state
input_state = {
"messages": [HumanMessage("hello async")],
"custom_field": "test_value_async",
}
await agent.ainvoke(input_state)
# The dynamic model function should receive the original state, not the processed model input
assert len(received_states) == 1
received_state = received_states[0]
# Should have the custom field from original state
assert "custom_field" in received_state
assert received_state["custom_field"] == "test_value_async"
# Should have the original messages
assert len(received_state["messages"]) == 1
assert received_state["messages"][0].content == "hello async"
def test_pre_model_hook() -> None:
model = FakeToolCallingModel(tool_calls=[])
+2 -2
View File
@@ -316,7 +316,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.6.0"
version = "0.6.0a1"
source = { editable = "../langgraph" }
dependencies = [
{ name = "langchain-core" },
@@ -430,7 +430,7 @@ dev = [
[[package]]
name = "langgraph-checkpoint-sqlite"
version = "2.0.11"
version = "2.0.10"
source = { editable = "../checkpoint-sqlite" }
dependencies = [
{ name = "aiosqlite" },