docs: Convert notebooks (#5322)

* docs: Convert subgraphs notebook

* graph api conversion

* fix examples

* add

* fixes

* fix links

* fix links

* be gone!

* multi-agent conversion

* fix link

* fix links

* fix link
This commit is contained in:
Lauren Hirata Singh
2025-07-02 23:28:42 +00:00
committed by GitHub
parent 22e09d2739
commit 543e4c4e7e
25 changed files with 3022 additions and 4039 deletions
+19 -19
View File
@@ -34,20 +34,20 @@ REDIRECT_MAP = {
"how-tos/streaming-from-final-node.ipynb": "how-tos/streaming-specific-nodes.ipynb",
"how-tos/streaming-events-from-within-tools-without-langchain.ipynb": "how-tos/streaming-events-from-within-tools.ipynb#example-without-langchain",
# graph-api
"how-tos/state-reducers.ipynb": "how-tos/graph-api#define-and-update-state",
"how-tos/sequence.ipynb": "how-tos/graph-api#create-a-sequence-of-steps",
"how-tos/branching.ipynb": "how-tos/graph-api#create-branches",
"how-tos/recursion-limit.ipynb": "how-tos/graph-api#create-and-control-loops",
"how-tos/visualization.ipynb": "how-tos/graph-api#visualize-your-graph",
"how-tos/input_output_schema.ipynb": "how-tos/graph-api#define-input-and-output-schemas",
"how-tos/pass_private_state.ipynb": "how-tos/graph-api#pass-private-state-between-nodes",
"how-tos/state-model.ipynb": "how-tos/graph-api#use-pydantic-models-for-graph-state",
"how-tos/map-reduce.ipynb": "how-tos/graph-api/#map-reduce-and-the-send-api",
"how-tos/command.ipynb": "how-tos/graph-api/#combine-control-flow-and-state-updates-with-command",
"how-tos/configuration.ipynb": "how-tos/graph-api/#add-runtime-configuration",
"how-tos/node-retries.ipynb": "how-tos/graph-api/#add-retry-policies",
"how-tos/return-when-recursion-limit-hits.ipynb": "how-tos/graph-api/#impose-a-recursion-limit",
"how-tos/async.ipynb": "how-tos/graph-api/#async",
"how-tos/state-reducers.ipynb": "how-tos/graph-api.md#define-and-update-state",
"how-tos/sequence.ipynb": "how-tos/graph-api.md#create-a-sequence-of-steps",
"how-tos/branching.ipynb": "how-tos/graph-api.md#create-branches",
"how-tos/recursion-limit.ipynb": "how-tos/graph-api.md#create-and-control-loops",
"how-tos/visualization.ipynb": "how-tos/graph-api.md#visualize-your-graph",
"how-tos/input_output_schema.ipynb": "how-tos/graph-api.md#define-input-and-output-schemas",
"how-tos/pass_private_state.ipynb": "how-tos/graph-api.md#pass-private-state-between-nodes",
"how-tos/state-model.ipynb": "how-tos/graph-api.md#use-pydantic-models-for-graph-state",
"how-tos/map-reduce.ipynb": "how-tos/graph-api.md#map-reduce-and-the-send-api",
"how-tos/command.ipynb": "how-tos/graph-api.md#combine-control-flow-and-state-updates-with-command",
"how-tos/configuration.ipynb": "how-tos/graph-api.md#add-runtime-configuration",
"how-tos/node-retries.ipynb": "how-tos/graph-api.md#add-retry-policies",
"how-tos/return-when-recursion-limit-hits.ipynb": "how-tos/graph-api.md#impose-a-recursion-limit",
"how-tos/async.ipynb": "how-tos/graph-api.md#async",
# memory how-tos
"how-tos/memory/manage-conversation-history.ipynb": "how-tos/memory/add-memory.md",
"how-tos/memory/delete-messages.ipynb": "how-tos/memory/add-memory.md#delete-messages",
@@ -55,8 +55,8 @@ REDIRECT_MAP = {
"how-tos/memory.ipynb": "how-tos/memory/add-memory.md",
"agents/memory.ipynb": "how-tos/memory/add-memory.md",
# subgraph how-tos
"how-tos/subgraph-transform-state.ipynb": "how-tos/subgraph.ipynb#different-state-schemas",
"how-tos/subgraphs-manage-state.ipynb": "how-tos/subgraph.ipynb#add-persistence",
"how-tos/subgraph-transform-state.ipynb": "how-tos/subgraph.md#different-state-schemas",
"how-tos/subgraphs-manage-state.ipynb": "how-tos/subgraph.md#add-persistence",
# persistence how-tos
"how-tos/persistence_postgres.ipynb": "how-tos/memory/add-memory.md#use-in-production",
"how-tos/persistence_mongodb.ipynb": "how-tos/memory/add-memory.md#use-in-production",
@@ -73,9 +73,9 @@ REDIRECT_MAP = {
"how-tos/pass-run-time-values-to-tools.ipynb": "how-tos/tool-calling.ipynb#read-state",
"how-tos/update-state-from-tools.ipynb": "how-tos/tool-calling.ipynb#update-state",
# multi-agent how-tos
"how-tos/agent-handoffs.ipynb": "how-tos/multi_agent.ipynb#handoffs",
"how-tos/multi-agent-network.ipynb": "how-tos/multi_agent.ipynb#use-in-a-multi-agent-system",
"how-tos/multi-agent-multi-turn-convo.ipynb": "how-tos/multi_agent.ipynb#multi-turn-conversation",
"how-tos/agent-handoffs.ipynb": "how-tos/multi_agent.md#handoffs",
"how-tos/multi-agent-network.ipynb": "how-tos/multi_agent.md#use-in-a-multi-agent-system",
"how-tos/multi-agent-multi-turn-convo.ipynb": "how-tos/multi_agent.md#multi-turn-conversation",
# cloud redirects
"cloud/index.md": "index.md",
"cloud/how-tos/index.md": "concepts/langgraph_platform",
+2 -2
View File
@@ -97,7 +97,7 @@ Parallel processing is vital for efficient multi-agent systems and complex tasks
- Implementation of map-reduce-like operations
- Efficient handling of independent subtasks
For practical implementation, see our [map-reduce tutorial](../how-tos/graph-api.ipynb#map-reduce-and-the-send-api)
For practical implementation, see our [map-reduce tutorial](../how-tos/graph-api.md#map-reduce-and-the-send-api)
### Subgraphs
@@ -107,7 +107,7 @@ For practical implementation, see our [map-reduce tutorial](../how-tos/graph-api
- Hierarchical organization of agent teams
- Controlled communication between agents and the main system
Subgraphs communicate with the parent graph through overlapping keys in the state schema. This enables flexible, modular agent design. For implementation details, refer to our [subgraph how-to guide](../how-tos/subgraph.ipynb).
Subgraphs communicate with the parent graph through overlapping keys in the state schema. This enables flexible, modular agent design. For implementation details, refer to our [subgraph how-to guide](../how-tos/subgraph.md).
### Reflection
+10 -10
View File
@@ -45,9 +45,9 @@ The first thing you do when you define a graph is define the `State` of the grap
### Schema
The main documented way to specify the schema of a graph is by using `TypedDict`. However, we also support [using a Pydantic BaseModel](../how-tos/graph-api.ipynb#use-pydantic-models-for-graph-state) as your graph state to add **default values** and additional data validation.
The main documented way to specify the schema of a graph is by using `TypedDict`. However, we also support [using a Pydantic BaseModel](../how-tos/graph-api.md#use-pydantic-models-for-graph-state) as your graph state to add **default values** and additional data validation.
By default, the graph will have the same input and output schemas. If you want to change this, you can also specify explicit input and output schemas directly. This is useful when you have a lot of keys, and some are explicitly for input and others for output. See the [guide here](../how-tos/graph-api.ipynb#define-input-and-output-schemas) for how to use.
By default, the graph will have the same input and output schemas. If you want to change this, you can also specify explicit input and output schemas directly. This is useful when you have a lot of keys, and some are explicitly for input and others for output. See the [guide here](../how-tos/graph-api.md#define-input-and-output-schemas) for how to use.
#### Multiple schemas
@@ -56,9 +56,9 @@ Typically, all graph nodes communicate with a single schema. This means that the
- Internal nodes can pass information that is not required in the graph's input / output.
- We may also want to use different input / output schemas for the graph. The output might, for example, only contain a single relevant output key.
It is possible to have nodes write to private state channels inside the graph for internal node communication. We can simply define a private schema, `PrivateState`. See [this guide](../how-tos/graph-api.ipynb#pass-private-state-between-nodes) for more detail.
It is possible to have nodes write to private state channels inside the graph for internal node communication. We can simply define a private schema, `PrivateState`. See [this guide](../how-tos/graph-api.md#pass-private-state-between-nodes) for more detail.
It is also possible to define explicit input and output schemas for a graph. In these cases, we define an "internal" schema that contains _all_ keys relevant to graph operations. But, we also define `input` and `output` schemas that are sub-sets of the "internal" schema to constrain the input and output of the graph. See [this guide](../how-tos/graph-api.ipynb#define-input-and-output-schemas) for more detail.
It is also possible to define explicit input and output schemas for a graph. In these cases, we define an "internal" schema that contains _all_ keys relevant to graph operations. But, we also define `input` and `output` schemas that are sub-sets of the "internal" schema to constrain the input and output of the graph. See [this guide](../how-tos/graph-api.md#define-input-and-output-schemas) for more detail.
Let's look at an example:
@@ -406,7 +406,7 @@ def my_node(state: State) -> Command[Literal["my_other_node"]]:
When returning `Command` in your node functions, you must add return type annotations with the list of node names the node is routing to, e.g. `Command[Literal["my_other_node"]]`. This is necessary for the graph rendering and tells LangGraph that `my_node` can navigate to `my_other_node`.
Check out this [how-to guide](../how-tos/graph-api.ipynb#combine-control-flow-and-state-updates-with-command) for an end-to-end example of how to use `Command`.
Check out this [how-to guide](../how-tos/graph-api.md#combine-control-flow-and-state-updates-with-command) for an end-to-end example of how to use `Command`.
### When should I use Command instead of conditional edges?
@@ -433,17 +433,17 @@ def my_node(state: State) -> Command[Literal["other_subgraph"]]:
!!! important "State updates with `Command.PARENT`"
When you send updates from a subgraph node to a parent graph node for a key that's shared by both parent and subgraph [state schemas](#schema), you **must** define a [reducer](#reducers) for the key you're updating in the parent graph state. See this [example](../how-tos/graph-api.ipynb#navigate-to-a-node-in-a-parent-graph).
When you send updates from a subgraph node to a parent graph node for a key that's shared by both parent and subgraph [state schemas](#schema), you **must** define a [reducer](#reducers) for the key you're updating in the parent graph state. See this [example](../how-tos/graph-api.md#navigate-to-a-node-in-a-parent-graph).
This is particularly useful when implementing [multi-agent handoffs](./multi_agent.md#handoffs).
Check out [this guide](../how-tos/graph-api.ipynb#navigate-to-a-node-in-a-parent-graph) for detail.
Check out [this guide](../how-tos/graph-api.md#navigate-to-a-node-in-a-parent-graph) for detail.
### Using inside tools
A common use case is updating graph state from inside a tool. For example, in a customer support application you might want to look up customer information based on their account number or ID in the beginning of the conversation.
Refer to [this guide](../how-tos/graph-api.ipynb#use-inside-tools) for detail.
Refer to [this guide](../how-tos/graph-api.md#use-inside-tools) for detail.
### Human-in-the-loop
@@ -489,7 +489,7 @@ def node_a(state, config):
...
```
See [this guide](../how-tos/graph-api.ipynb#add-runtime-configuration) for a full breakdown on configuration.
See [this guide](../how-tos/graph-api.md#add-runtime-configuration) for a full breakdown on configuration.
### Recursion Limit
@@ -503,4 +503,4 @@ Read [this how-to](https://langchain-ai.github.io/langgraph/how-tos/recursion-li
## Visualization
It's often nice to be able to visualize graphs, especially as they get more complex. LangGraph comes with several built-in ways to visualize graphs. See [this how-to guide](../how-tos/graph-api.ipynb#visualize-your-graph) for more info.
It's often nice to be able to visualize graphs, especially as they get more complex. LangGraph comes with several built-in ways to visualize graphs. See [this how-to guide](../how-tos/graph-api.md#visualize-your-graph) for more info.
+1 -1
View File
@@ -53,5 +53,5 @@ sequenceDiagram
LangGraph -->> ClientApp: 12. Return resources / tool output
```
For more information, see [MCP endpoint in LangGraph Server](../concepts/server-mcp.md#use-mcp-tools-in-your-deployment).
For more information, see [MCP endpoint in LangGraph Server](../concepts/server-mcp.md#use-user-scoped-mcp-tools-in-your-deployment).
+3 -3
View File
@@ -166,7 +166,7 @@ network = builder.compile()
### Supervisor
In this architecture, we define agents as nodes and add a supervisor node (LLM) that decides which agent nodes should be called next. We use [`Command`](./low_level.md#command) to route execution to the appropriate agent node based on supervisor's decision. This architecture also lends itself well to running multiple agents in parallel or using [map-reduce](../how-tos/graph-api.ipynb#map-reduce-and-the-send-api) pattern.
In this architecture, we define agents as nodes and add a supervisor node (LLM) that decides which agent nodes should be called next. We use [`Command`](./low_level.md#command) to route execution to the appropriate agent node based on supervisor's decision. This architecture also lends itself well to running multiple agents in parallel or using [map-reduce](../how-tos/graph-api.md#map-reduce-and-the-send-api) pattern.
```python
from typing import Literal
@@ -414,5 +414,5 @@ There are two high-level approaches to achieve that:
An agent might need to have a different state schema from the rest of the agents. For example, a search agent might only need to keep track of queries and retrieved documents. There are two ways to achieve this in LangGraph:
- Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, its important to [add input / output transformations](../how-tos/subgraph.ipynb#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
- Define agent node functions with a [private input state schema](../how-tos/graph-api.ipynb/#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
- Define [subgraph](./subgraphs.md) agents with a separate state schema. If there are no shared state keys (channels) between the subgraph and the parent graph, its important to [add input / output transformations](../how-tos/subgraph.md#different-state-schemas) so that the parent graph knows how to communicate with the subgraphs.
- Define agent node functions with a [private input state schema](../how-tos/graph-api.md/#pass-private-state-between-nodes) that is distinct from the overall graph state schema. This allows passing information that is only needed for executing that particular agent.
+1 -1
View File
@@ -198,7 +198,7 @@ print(graph.invoke({"question": "hi"}))
For more details, see the [low-level concepts guide](https://langchain-ai.github.io/langgraph/concepts/low_level/#state).
## Use User-Scoped MCP tools in your deployment
## Use user-scoped MCP tools in your deployment
!!! tip "Prerequisites"
+2 -2
View File
@@ -12,7 +12,7 @@ Some reasons for using subgraphs are:
The main question when adding subgraphs is how the parent graph and subgraph communicate, i.e. how they pass the [state](./low_level.md#state) between each other during the graph execution. There are two scenarios:
* parent and subgraph have **shared state keys** in their state [schemas](./low_level.md#state). In this case, you can [include the subgraph as a node in the parent graph](../how-tos/subgraph.ipynb#shared-state-schemas)
* parent and subgraph have **shared state keys** in their state [schemas](./low_level.md#state). In this case, you can [include the subgraph as a node in the parent graph](../how-tos/subgraph.md#shared-state-schemas)
```python
from langgraph.graph import StateGraph, MessagesState, START
@@ -40,7 +40,7 @@ The main question when adding subgraphs is how the parent graph and subgraph com
graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
```
* parent graph and subgraph have **different schemas** (no shared state keys in their state [schemas](./low_level.md#state)). In this case, you have to [call the subgraph from inside a node in the parent graph](../how-tos/subgraph.ipynb#different-state-schemas): this is useful when the parent graph and the subgraph have different state schemas and you need to transform state before or after calling the subgraph
* parent graph and subgraph have **different schemas** (no shared state keys in their state [schemas](./low_level.md#state)). In this case, you have to [call the subgraph from inside a node in the parent graph](../how-tos/subgraph.md#different-state-schemas): this is useful when the parent graph and the subgraph have different state schemas and you need to transform state before or after calling the subgraph
```python
from typing_extensions import TypedDict, Annotated
Binary file not shown.

After

Width:  |  Height:  |  Size: 3.9 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 147 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.0 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 9.5 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 9.9 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.2 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 11 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.5 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 15 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 7.7 KiB

File diff suppressed because one or more lines are too long
File diff suppressed because it is too large Load Diff
+580
View File
@@ -0,0 +1,580 @@
# Build multi-agent systems
A single agent might struggle if it needs to specialize in multiple domains or manage many tools. To tackle this, you can break your agent into smaller, independent agents and composing them into a [multi-agent system](../concepts/multi_agent.md).
In multi-agent systems, agents need to communicate between each other. They do so via [handoffs](#handoffs) — a primitive that describes which agent to hand control to and the payload to send to that agent.
This guide covers the following:
* implementing [handoffs](#handoffs) between agents
* using handoffs and the prebuilt [agent](../agents/agents.md) to [build a custom multi-agent system](#build-a-multi-agent-system)
To get started with building multi-agent systems, check out LangGraph [prebuilt implementations](#prebuilt-implementations) of two of the most popular multi-agent architectures — [supervisor](../agents/multi-agent.md#supervisor) and [swarm](../agents/multi-agent.md#swarm).
## Handoffs
To set up communication between the agents in a multi-agent system you can use [**handoffs**](../concepts/multi_agent.md#handoffs) — a pattern where one agent *hands off* control to another. Handoffs allow you to specify:
- **destination**: target agent to navigate to (e.g., name of the LangGraph node to go to)
- **payload**: information to pass to that agent (e.g., state update)
### Create handoffs
To implement handoffs, you can return `Command` objects from your agent nodes or tools:
```python
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command
def create_handoff_tool(*, agent_name: str, description: str | None = None):
name = f"transfer_to_{agent_name}"
description = description or f"Transfer to {agent_name}"
@tool(name, description=description)
def handoff_tool(
# highlight-next-line
state: Annotated[MessagesState, InjectedState], # (1)!
# highlight-next-line
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
tool_message = {
"role": "tool",
"content": f"Successfully transferred to {agent_name}",
"name": name,
"tool_call_id": tool_call_id,
}
return Command( # (2)!
# highlight-next-line
goto=agent_name, # (3)!
# highlight-next-line
update={"messages": state["messages"] + [tool_message]}, # (4)!
# highlight-next-line
graph=Command.PARENT, # (5)!
)
return handoff_tool
```
1. Access the [state](../concepts/low_level.md#state) of the agent that is calling the handoff tool using the [InjectedState][langgraph.prebuilt.InjectedState] annotation.
2. The `Command` primitive allows specifying a state update and a node transition as a single operation, making it useful for implementing handoffs.
3. Name of the agent or node to hand off to.
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
5. Indicate to LangGraph that we need to navigate to agent node in a **parent** multi-agent graph.
!!! tip
If you want to use tools that return `Command`, you can either use prebuilt [`create_react_agent`][langgraph.prebuilt.chat_agent_executor.create_react_agent] / [`ToolNode`][langgraph.prebuilt.tool_node.ToolNode] components, or implement your own tool-executing node that collects `Command` objects returned by the tools and returns a list of them, e.g.:
```python
def call_tools(state):
...
commands = [tools_by_name[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
return commands
```
!!! Important
This handoff implementation assumes that:
- each agent receives overall message history (across all agents) in the multi-agent system as its input. If you want more control over agent inputs, see [this section](#control-agent-inputs)
- each agent outputs its internal messages history to the overall message history of the multi-agent system. If you want more control over **how agent outputs are added**, wrap the agent in a separate node function:
```python
def call_hotel_assistant(state):
# return agent's final response,
# excluding inner monologue
response = hotel_assistant.invoke(state)
# highlight-next-line
return {"messages": response["messages"][-1]}
```
### Control agent inputs
You can use the [`Send()`][langgraph.types.Send] primitive to directly send data to the worker agents during the handoff. For example, you can request that the calling agent populate a task description for the next agent:
```python
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import StateGraph, START, MessagesState
# highlight-next-line
from langgraph.types import Command, Send
def create_task_description_handoff_tool(
*, agent_name: str, description: str | None = None
):
name = f"transfer_to_{agent_name}"
description = description or f"Ask {agent_name} for help."
@tool(name, description=description)
def handoff_tool(
# this is populated by the calling agent
task_description: Annotated[
str,
"Description of what the next agent should do, including all of the relevant context.",
],
# these parameters are ignored by the LLM
state: Annotated[MessagesState, InjectedState],
) -> Command:
task_description_message = {"role": "user", "content": task_description}
agent_input = {**state, "messages": [task_description_message]}
return Command(
# highlight-next-line
goto=[Send(agent_name, agent_input)],
graph=Command.PARENT,
)
return handoff_tool
```
See the multi-agent [supervisor](../tutorials/multi_agent/agent_supervisor.ipynb#4-create-delegation-tasks) example for a full example of using [`Send()`][langgraph.types.Send] in handoffs.
## Build a multi-agent system
You can use handoffs in any agents built with LangGraph. We recommend using the prebuilt [agent](../agents/overview.md) or [`ToolNode`](./tool-calling.md#toolnode), as they natively support handoffs tools returning `Command`. Below is an example of how you can implement a multi-agent system for booking travel using handoffs:
```python
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, MessagesState
def create_handoff_tool(*, agent_name: str, description: str | None = None):
# same implementation as above
...
return Command(...)
# Handoffs
transfer_to_hotel_assistant = create_handoff_tool(agent_name="hotel_assistant")
transfer_to_flight_assistant = create_handoff_tool(agent_name="flight_assistant")
# Define agents
flight_assistant = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
# highlight-next-line
tools=[..., transfer_to_hotel_assistant],
# highlight-next-line
name="flight_assistant"
)
hotel_assistant = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
# highlight-next-line
tools=[..., transfer_to_flight_assistant],
# highlight-next-line
name="hotel_assistant"
)
# Define multi-agent graph
multi_agent_graph = (
StateGraph(MessagesState)
# highlight-next-line
.add_node(flight_assistant)
# highlight-next-line
.add_node(hotel_assistant)
.add_edge(START, "flight_assistant")
.compile()
)
```
??? example "Full example: Multi-agent system for booking travel"
```python
from typing import Annotated
from langchain_core.messages import convert_to_messages
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.types import Command
# We'll use `pretty_print_messages` helper to render the streamed agent outputs nicely
def pretty_print_message(message, indent=False):
pretty_message = message.pretty_repr(html=True)
if not indent:
print(pretty_message)
return
indented = "\n".join("\t" + c for c in pretty_message.split("\n"))
print(indented)
def pretty_print_messages(update, last_message=False):
is_subgraph = False
if isinstance(update, tuple):
ns, update = update
# skip parent graph updates in the printouts
if len(ns) == 0:
return
graph_id = ns[-1].split(":")[0]
print(f"Update from subgraph {graph_id}:")
print("\n")
is_subgraph = True
for node_name, node_update in update.items():
update_label = f"Update from node {node_name}:"
if is_subgraph:
update_label = "\t" + update_label
print(update_label)
print("\n")
messages = convert_to_messages(node_update["messages"])
if last_message:
messages = messages[-1:]
for m in messages:
pretty_print_message(m, indent=is_subgraph)
print("\n")
def create_handoff_tool(*, agent_name: str, description: str | None = None):
name = f"transfer_to_{agent_name}"
description = description or f"Transfer to {agent_name}"
@tool(name, description=description)
def handoff_tool(
# highlight-next-line
state: Annotated[MessagesState, InjectedState], # (1)!
# highlight-next-line
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
tool_message = {
"role": "tool",
"content": f"Successfully transferred to {agent_name}",
"name": name,
"tool_call_id": tool_call_id,
}
return Command( # (2)!
# highlight-next-line
goto=agent_name, # (3)!
# highlight-next-line
update={"messages": state["messages"] + [tool_message]}, # (4)!
# highlight-next-line
graph=Command.PARENT, # (5)!
)
return handoff_tool
# Handoffs
transfer_to_hotel_assistant = create_handoff_tool(
agent_name="hotel_assistant",
description="Transfer user to the hotel-booking assistant.",
)
transfer_to_flight_assistant = create_handoff_tool(
agent_name="flight_assistant",
description="Transfer user to the flight-booking assistant.",
)
# Simple agent tools
def book_hotel(hotel_name: str):
"""Book a hotel"""
return f"Successfully booked a stay at {hotel_name}."
def book_flight(from_airport: str, to_airport: str):
"""Book a flight"""
return f"Successfully booked a flight from {from_airport} to {to_airport}."
# Define agents
flight_assistant = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
# highlight-next-line
tools=[book_flight, transfer_to_hotel_assistant],
prompt="You are a flight booking assistant",
# highlight-next-line
name="flight_assistant"
)
hotel_assistant = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
# highlight-next-line
tools=[book_hotel, transfer_to_flight_assistant],
prompt="You are a hotel booking assistant",
# highlight-next-line
name="hotel_assistant"
)
# Define multi-agent graph
multi_agent_graph = (
StateGraph(MessagesState)
.add_node(flight_assistant)
.add_node(hotel_assistant)
.add_edge(START, "flight_assistant")
.compile()
)
# Run the multi-agent graph
for chunk in multi_agent_graph.stream(
{
"messages": [
{
"role": "user",
"content": "book a flight from BOS to JFK and a stay at McKittrick Hotel"
}
]
},
# highlight-next-line
subgraphs=True
):
pretty_print_messages(chunk)
```
1. Access agent's state
2. The `Command` primitive allows specifying a state update and a node transition as a single operation, making it useful for implementing handoffs.
3. Name of the agent or node to hand off to.
4. Take the agent's messages and **add** them to the parent's **state** as part of the handoff. The next agent will see the parent state.
5. Indicate to LangGraph that we need to navigate to agent node in a **parent** multi-agent graph.
## Multi-turn conversation
Users might want to engage in a *multi-turn conversation* with one or more agents. To build a system that can handle this, you can create a node that uses an [`interrupt`][langgraph.types.interrupt] to collect user input and routes back to the **active** agent.
The agents can then be implemented as nodes in a graph that executes agent steps and determines the next action:
1. **Wait for user input** to continue the conversation, or
2. **Route to another agent** (or back to itself, such as in a loop) via a [handoff](#handoffs)
```python
def human(state) -> Command[Literal["agent", "another_agent"]]:
"""A node for collecting user input."""
user_input = interrupt(value="Ready for user input.")
# Determine the active agent.
active_agent = ...
...
return Command(
update={
"messages": [{
"role": "human",
"content": user_input,
}]
},
goto=active_agent
)
def agent(state) -> Command[Literal["agent", "another_agent", "human"]]:
# The condition for routing/halting can be anything, e.g. LLM tool call / structured output, etc.
goto = get_next_agent(...) # 'agent' / 'another_agent'
if goto:
return Command(goto=goto, update={"my_state_key": "my_state_value"})
else:
return Command(goto="human") # Go to human node
```
??? example "Full example: multi-agent system for travel recommendations"
In this example, we will build a team of travel assistant agents that can communicate with each other via handoffs.
We will create 2 agents:
* travel_advisor: can help with travel destination recommendations. Can ask hotel_advisor for help.
* hotel_advisor: can help with hotel recommendations. Can ask travel_advisor for help.
```python
from langchain_anthropic import ChatAnthropic
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.prebuilt import create_react_agent, InjectedState
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver
model = ChatAnthropic(model="claude-3-5-sonnet-latest")
class MultiAgentState(MessagesState):
last_active_agent: str
# Define travel advisor tools and ReAct agent
travel_advisor_tools = [
get_travel_recommendations,
make_handoff_tool(agent_name="hotel_advisor"),
]
travel_advisor = create_react_agent(
model,
travel_advisor_tools,
prompt=(
"You are a general travel expert that can recommend travel destinations (e.g. countries, cities, etc). "
"If you need hotel recommendations, ask 'hotel_advisor' for help. "
"You MUST include human-readable response before transferring to another agent."
),
)
def call_travel_advisor(
state: MultiAgentState,
) -> Command[Literal["hotel_advisor", "human"]]:
# You can also add additional logic like changing the input to the agent / output from the agent, etc.
# NOTE: we're invoking the ReAct agent with the full history of messages in the state
response = travel_advisor.invoke(state)
update = {**response, "last_active_agent": "travel_advisor"}
return Command(update=update, goto="human")
# Define hotel advisor tools and ReAct agent
hotel_advisor_tools = [
get_hotel_recommendations,
make_handoff_tool(agent_name="travel_advisor"),
]
hotel_advisor = create_react_agent(
model,
hotel_advisor_tools,
prompt=(
"You are a hotel expert that can provide hotel recommendations for a given destination. "
"If you need help picking travel destinations, ask 'travel_advisor' for help."
"You MUST include human-readable response before transferring to another agent."
),
)
def call_hotel_advisor(
state: MultiAgentState,
) -> Command[Literal["travel_advisor", "human"]]:
response = hotel_advisor.invoke(state)
update = {**response, "last_active_agent": "hotel_advisor"}
return Command(update=update, goto="human")
def human_node(
state: MultiAgentState, config
) -> Command[Literal["hotel_advisor", "travel_advisor", "human"]]:
"""A node for collecting user input."""
user_input = interrupt(value="Ready for user input.")
active_agent = state["last_active_agent"]
return Command(
update={
"messages": [
{
"role": "human",
"content": user_input,
}
]
},
goto=active_agent,
)
builder = StateGraph(MultiAgentState)
builder.add_node("travel_advisor", call_travel_advisor)
builder.add_node("hotel_advisor", call_hotel_advisor)
# This adds a node to collect human input, which will route
# back to the active agent.
builder.add_node("human", human_node)
# We'll always start with a general travel advisor.
builder.add_edge(START, "travel_advisor")
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)
```
Let's test a multi turn conversation with this application.
```python
import uuid
thread_config = {"configurable": {"thread_id": str(uuid.uuid4())}}
inputs = [
# 1st round of conversation,
{
"messages": [
{"role": "user", "content": "i wanna go somewhere warm in the caribbean"}
]
},
# Since we're using `interrupt`, we'll need to resume using the Command primitive.
# 2nd round of conversation,
Command(
resume="could you recommend a nice hotel in one of the areas and tell me which area it is."
),
# 3rd round of conversation,
Command(
resume="i like the first one. could you recommend something to do near the hotel?"
),
]
for idx, user_input in enumerate(inputs):
print()
print(f"--- Conversation Turn {idx + 1} ---")
print()
print(f"User: {user_input}")
print()
for update in graph.stream(
user_input,
config=thread_config,
stream_mode="updates",
):
for node_id, value in update.items():
if isinstance(value, dict) and value.get("messages", []):
last_message = value["messages"][-1]
if isinstance(last_message, dict) or last_message.type != "ai":
continue
print(f"{node_id}: {last_message.content}")
```
```
--- Conversation Turn 1 ---
User: {'messages': [{'role': 'user', 'content': 'i wanna go somewhere warm in the caribbean'}]}
travel_advisor: Based on the recommendations, Aruba would be an excellent choice for your Caribbean getaway! Aruba is known as "One Happy Island" and offers:
- Year-round warm weather with consistent temperatures around 82°F (28°C)
- Beautiful white sand beaches like Eagle Beach and Palm Beach
- Clear turquoise waters perfect for swimming and snorkeling
- Minimal rainfall and location outside the hurricane belt
- A blend of Caribbean and Dutch culture
- Great dining options and nightlife
- Various water sports and activities
Would you like me to get some specific hotel recommendations in Aruba for your stay? I can transfer you to our hotel advisor who can help with accommodations.
--- Conversation Turn 2 ---
User: Command(resume='could you recommend a nice hotel in one of the areas and tell me which area it is.')
hotel_advisor: Based on the recommendations, I can suggest two excellent options:
1. The Ritz-Carlton, Aruba - Located in Palm Beach
- This luxury resort is situated in the vibrant Palm Beach area
- Known for its exceptional service and amenities
- Perfect if you want to be close to dining, shopping, and entertainment
- Features multiple restaurants, a casino, and a world-class spa
- Located on a pristine stretch of Palm Beach
2. Bucuti & Tara Beach Resort - Located in Eagle Beach
- An adults-only boutique resort on Eagle Beach
- Known for being more intimate and peaceful
- Award-winning for its sustainability practices
- Perfect for a romantic getaway or peaceful vacation
- Located on one of the most beautiful beaches in the Caribbean
Would you like more specific information about either of these properties or their locations?
--- Conversation Turn 3 ---
User: Command(resume='i like the first one. could you recommend something to do near the hotel?')
travel_advisor: Near the Ritz-Carlton in Palm Beach, here are some highly recommended activities:
1. Visit the Palm Beach Plaza Mall - Just a short walk from the hotel, featuring shopping, dining, and entertainment
2. Try your luck at the Stellaris Casino - It's right in the Ritz-Carlton
3. Take a sunset sailing cruise - Many depart from the nearby pier
4. Visit the California Lighthouse - A scenic landmark just north of Palm Beach
5. Enjoy water sports at Palm Beach:
- Jet skiing
- Parasailing
- Snorkeling
- Stand-up paddleboarding
Would you like more specific information about any of these activities or would you like to know about other options in the area?
```
## Prebuilt implementations
LangGraph comes with prebuilt implementations of two of the most popular multi-agent architectures:
- [supervisor](../agents/multi-agent.md#supervisor) — individual agents are coordinated by a central supervisor agent. The supervisor controls all communication flow and task delegation, making decisions about which agent to invoke based on the current context and task requirements. You can use [`langgraph-supervisor`](https://github.com/langchain-ai/langgraph-supervisor-py) library to create a supervisor multi-agent systems.
- [swarm](../agents/multi-agent.md#supervisor) — agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent. You can use [`langgraph-swarm`](https://github.com/langchain-ai/langgraph-swarm-py) library to create a swarm multi-agent systems.
-559
View File
@@ -1,559 +0,0 @@
{
"cells": [
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"# Use subgraphs\n",
"\n",
"This guide explains the mechanics of using [subgraphs](../../concepts/subgraphs). A common application of subgraphs is to build [multi-agent](../../concepts/multi_agent) systems.\n",
"\n",
"When adding subgraphs, you need to define how the parent graph and the subgraph communicate:\n",
"\n",
"* [Shared state schemas](#shared-state-schemas) — parent and subgraph have **shared state keys** in their state [schemas](../../concepts/low_level#state)\n",
"* [Different state schemas](#different-state-schemas) — **no shared state keys** in parent and subgraph [schemas](../../concepts/low_level#state)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Setup"
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"%%capture --no-stderr\n",
"%pip install -U langgraph"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"<div class=\"admonition tip\">\n",
" <p class=\"admonition-title\">Set up <a href=\"https://smith.langchain.com\">LangSmith</a> for LangGraph development</p>\n",
" <p style=\"padding-top: 5px;\">\n",
" 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 — read more about how to get started <a href=\"https://docs.smith.langchain.com\">here</a>. \n",
" </p>\n",
"</div>"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Shared state schemas\n",
"\n",
"A common case is for the parent graph and subgraph to communicate over a shared state key (channel) in the [schema](../../concepts/low_level#state). For example, in [multi-agent](../../concepts/multi_agent) systems, the agents often communicate over a shared [messages](https://langchain-ai.github.io/langgraph/concepts/low_level/#why-use-messages) key.\n",
"\n",
"If your subgraph shares state keys with the parent graph, you can follow these steps to add it to your graph:\n",
"\n",
"1. Define the subgraph workflow (`subgraph_builder` in the example below) and compile it\n",
"2. Pass compiled subgraph to the `.add_node` method when defining the parent graph workflow\n",
"\n",
"```python\n",
"from typing_extensions import TypedDict\n",
"from langgraph.graph.state import StateGraph, START\n",
"\n",
"class State(TypedDict):\n",
" foo: str\n",
"\n",
"# Subgraph\n",
"\n",
"def subgraph_node_1(state: State):\n",
" return {\"foo\": \"hi! \" + state[\"foo\"]}\n",
"\n",
"subgraph_builder = StateGraph(State)\n",
"subgraph_builder.add_node(subgraph_node_1)\n",
"subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
"# highlight-next-line\n",
"subgraph = subgraph_builder.compile()\n",
"\n",
"# Parent graph\n",
"\n",
"builder = StateGraph(State)\n",
"# highlight-next-line\n",
"builder.add_node(\"node_1\", subgraph)\n",
"builder.add_edge(START, \"node_1\")\n",
"graph = builder.compile()\n",
"```\n",
"\n",
"??? example \"Full example: shared state schemas\"\n",
"\n",
" ```python\n",
" from typing_extensions import TypedDict\n",
" from langgraph.graph.state import StateGraph, START\n",
"\n",
" # Define subgraph\n",
" class SubgraphState(TypedDict):\n",
" foo: str # (1)! \n",
" bar: str # (2)!\n",
" \n",
" def subgraph_node_1(state: SubgraphState):\n",
" return {\"bar\": \"bar\"}\n",
" \n",
" def subgraph_node_2(state: SubgraphState):\n",
" # note that this node is using a state key ('bar') that is only available in the subgraph\n",
" # and is sending update on the shared state key ('foo')\n",
" return {\"foo\": state[\"foo\"] + state[\"bar\"]}\n",
" \n",
" subgraph_builder = StateGraph(SubgraphState)\n",
" subgraph_builder.add_node(subgraph_node_1)\n",
" subgraph_builder.add_node(subgraph_node_2)\n",
" subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
" subgraph_builder.add_edge(\"subgraph_node_1\", \"subgraph_node_2\")\n",
" subgraph = subgraph_builder.compile()\n",
" \n",
" # Define parent graph\n",
" class ParentState(TypedDict):\n",
" foo: str\n",
" \n",
" def node_1(state: ParentState):\n",
" return {\"foo\": \"hi! \" + state[\"foo\"]}\n",
" \n",
" builder = StateGraph(ParentState)\n",
" builder.add_node(\"node_1\", node_1)\n",
" # highlight-next-line\n",
" builder.add_node(\"node_2\", subgraph)\n",
" builder.add_edge(START, \"node_1\")\n",
" builder.add_edge(\"node_1\", \"node_2\")\n",
" graph = builder.compile()\n",
" \n",
" for chunk in graph.stream({\"foo\": \"foo\"}):\n",
" print(chunk)\n",
" ```\n",
"\n",
" 1. This key is shared with the parent graph state\n",
" 2. This key is private to the `SubgraphState` and is not visible to the parent graph\n",
" \n",
" ```\n",
" {'node_1': {'foo': 'hi! foo'}}\n",
" {'node_2': {'foo': 'hi! foobar'}}\n",
" ```\n",
"\n",
" ```"
]
},
{
"attachments": {},
"cell_type": "markdown",
"metadata": {},
"source": [
"## Different state schemas\n",
"\n",
"For more complex systems you might want to define subgraphs that have a **completely different schema** from the parent graph (no shared keys). For example, you might want to keep a private message history for each of the agents in a [multi-agent](../concepts/multi_agent.md) system.\n",
"\n",
"If that's the case for your application, you need to define a node **function that invokes the subgraph**. This function needs to transform the input (parent) state to the subgraph state before invoking the subgraph, and transform the results back to the parent state before returning the state update from the node.\n",
"\n",
"```python\n",
"from typing_extensions import TypedDict\n",
"from langgraph.graph.state import StateGraph, START\n",
"\n",
"class SubgraphState(TypedDict):\n",
" bar: str\n",
"\n",
"# Subgraph\n",
"\n",
"def subgraph_node_1(state: SubgraphState):\n",
" return {\"bar\": \"hi! \" + state[\"bar\"]}\n",
"\n",
"subgraph_builder = StateGraph(SubgraphState)\n",
"subgraph_builder.add_node(subgraph_node_1)\n",
"subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
"# highlight-next-line\n",
"subgraph = subgraph_builder.compile()\n",
"\n",
"# Parent graph\n",
"\n",
"class State(TypedDict):\n",
" foo: str\n",
"\n",
"def call_subgraph(state: State):\n",
" # highlight-next-line\n",
" subgraph_output = subgraph.invoke({\"bar\": state[\"foo\"]}) # (1)!\n",
" # highlight-next-line\n",
" return {\"foo\": subgraph_output[\"bar\"]} # (2)!\n",
"\n",
"builder = StateGraph(State)\n",
"# highlight-next-line\n",
"builder.add_node(\"node_1\", call_subgraph)\n",
"builder.add_edge(START, \"node_1\")\n",
"graph = builder.compile()\n",
"```\n",
"\n",
"1. Transform the state to the subgraph state\n",
"2. Transform response back to the parent state\n",
"\n",
"??? example \"Full example: different state schemas\"\n",
"\n",
" ```python\n",
" from typing_extensions import TypedDict\n",
" from langgraph.graph.state import StateGraph, START\n",
"\n",
" # Define subgraph\n",
" class SubgraphState(TypedDict):\n",
" # note that none of these keys are shared with the parent graph state\n",
" bar: str\n",
" baz: str\n",
" \n",
" def subgraph_node_1(state: SubgraphState):\n",
" return {\"baz\": \"baz\"}\n",
" \n",
" def subgraph_node_2(state: SubgraphState):\n",
" return {\"bar\": state[\"bar\"] + state[\"baz\"]}\n",
" \n",
" subgraph_builder = StateGraph(SubgraphState)\n",
" subgraph_builder.add_node(subgraph_node_1)\n",
" subgraph_builder.add_node(subgraph_node_2)\n",
" subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
" subgraph_builder.add_edge(\"subgraph_node_1\", \"subgraph_node_2\")\n",
" subgraph = subgraph_builder.compile()\n",
" \n",
" # Define parent graph\n",
" class ParentState(TypedDict):\n",
" foo: str\n",
" \n",
" def node_1(state: ParentState):\n",
" return {\"foo\": \"hi! \" + state[\"foo\"]}\n",
" \n",
" def node_2(state: ParentState):\n",
" # highlight-next-line\n",
" response = subgraph.invoke({\"bar\": state[\"foo\"]}) # (1)!\n",
" # highlight-next-line\n",
" return {\"foo\": response[\"bar\"]} # (2)!\n",
" \n",
" \n",
" builder = StateGraph(ParentState)\n",
" builder.add_node(\"node_1\", node_1)\n",
" # highlight-next-line\n",
" builder.add_node(\"node_2\", node_2)\n",
" builder.add_edge(START, \"node_1\")\n",
" builder.add_edge(\"node_1\", \"node_2\")\n",
" graph = builder.compile()\n",
" \n",
" for chunk in graph.stream({\"foo\": \"foo\"}, subgraphs=True):\n",
" print(chunk)\n",
" ```\n",
"\n",
" 1. Transform the state to the subgraph state\n",
" 2. Transform response back to the parent state\n",
"\n",
" ```\n",
" ((), {'node_1': {'foo': 'hi! foo'}})\n",
" (('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'subgraph_node_1': {'baz': 'baz'}})\n",
" (('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'subgraph_node_2': {'bar': 'hi! foobaz'}})\n",
" ((), {'node_2': {'foo': 'hi! foobaz'}})\n",
" ```\n",
"\n",
"??? example \"Full example: different state schemas (two levels of subgraphs)\"\n",
"\n",
" This is an example with two levels of subgraphs: parent -> child -> grandchild.\n",
"\n",
" ```python\n",
" # Grandchild graph\n",
" from typing_extensions import TypedDict\n",
" from langgraph.graph.state import StateGraph, START, END\n",
" \n",
" class GrandChildState(TypedDict):\n",
" my_grandchild_key: str\n",
" \n",
" def grandchild_1(state: GrandChildState) -> GrandChildState:\n",
" # NOTE: child or parent keys will not be accessible here\n",
" return {\"my_grandchild_key\": state[\"my_grandchild_key\"] + \", how are you\"}\n",
" \n",
" \n",
" grandchild = StateGraph(GrandChildState)\n",
" grandchild.add_node(\"grandchild_1\", grandchild_1)\n",
" \n",
" grandchild.add_edge(START, \"grandchild_1\")\n",
" grandchild.add_edge(\"grandchild_1\", END)\n",
" \n",
" grandchild_graph = grandchild.compile()\n",
" \n",
" # Child graph\n",
" class ChildState(TypedDict):\n",
" my_child_key: str\n",
" \n",
" def call_grandchild_graph(state: ChildState) -> ChildState:\n",
" # NOTE: parent or grandchild keys won't be accessible here\n",
" grandchild_graph_input = {\"my_grandchild_key\": state[\"my_child_key\"]} # (1)!\n",
" # highlight-next-line\n",
" grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)\n",
" return {\"my_child_key\": grandchild_graph_output[\"my_grandchild_key\"] + \" today?\"} # (2)!\n",
" \n",
" child = StateGraph(ChildState)\n",
" # highlight-next-line\n",
" child.add_node(\"child_1\", call_grandchild_graph) # (3)!\n",
" child.add_edge(START, \"child_1\")\n",
" child.add_edge(\"child_1\", END)\n",
" child_graph = child.compile()\n",
" \n",
" # Parent graph\n",
" class ParentState(TypedDict):\n",
" my_key: str\n",
" \n",
" def parent_1(state: ParentState) -> ParentState:\n",
" # NOTE: child or grandchild keys won't be accessible here\n",
" return {\"my_key\": \"hi \" + state[\"my_key\"]}\n",
" \n",
" def parent_2(state: ParentState) -> ParentState:\n",
" return {\"my_key\": state[\"my_key\"] + \" bye!\"}\n",
" \n",
" def call_child_graph(state: ParentState) -> ParentState:\n",
" child_graph_input = {\"my_child_key\": state[\"my_key\"]} # (4)!\n",
" # highlight-next-line\n",
" child_graph_output = child_graph.invoke(child_graph_input)\n",
" return {\"my_key\": child_graph_output[\"my_child_key\"]} # (5)!\n",
" \n",
" parent = StateGraph(ParentState)\n",
" parent.add_node(\"parent_1\", parent_1)\n",
" # highlight-next-line\n",
" parent.add_node(\"child\", call_child_graph) # (6)!\n",
" parent.add_node(\"parent_2\", parent_2)\n",
" \n",
" parent.add_edge(START, \"parent_1\")\n",
" parent.add_edge(\"parent_1\", \"child\")\n",
" parent.add_edge(\"child\", \"parent_2\")\n",
" parent.add_edge(\"parent_2\", END)\n",
" \n",
" parent_graph = parent.compile()\n",
" \n",
" for chunk in parent_graph.stream({\"my_key\": \"Bob\"}, subgraphs=True):\n",
" print(chunk)\n",
" ```\n",
"\n",
" 1. We're transforming the state from the child state channels (`my_child_key`) to the child state channels (`my_grandchild_key`)\n",
" 2. We're transforming the state from the grandchild state channels (`my_grandchild_key`) back to the child state channels (`my_child_key`)\n",
" 3. We're passing a function here instead of just compiled graph (`grandchild_graph`)\n",
" 4. We're transforming the state from the parent state channels (`my_key`) to the child state channels (`my_child_key`)\n",
" 5. We're transforming the state from the child state channels (`my_child_key`) back to the parent state channels (`my_key`)\n",
" 6. We're passing a function here instead of just a compiled graph (`child_graph`)\n",
"\n",
" ```\n",
" ((), {'parent_1': {'my_key': 'hi Bob'}})\n",
" (('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child_1:781bb3b1-3971-84ce-810b-acf819a03f9c'), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})\n",
" (('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b',), {'child_1': {'my_child_key': 'hi Bob, how are you today?'}})\n",
" ((), {'child': {'my_key': 'hi Bob, how are you today?'}})\n",
" ((), {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}})\n",
" ```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Add persistence \n",
"\n",
"You only need to **provide the checkpointer when compiling the parent graph**. LangGraph will automatically propagate the checkpointer to the child subgraphs.\n",
"\n",
"```python\n",
"from langgraph.graph import START, StateGraph\n",
"from langgraph.checkpoint.memory import InMemorySaver\n",
"from typing_extensions import TypedDict\n",
"\n",
"class State(TypedDict):\n",
" foo: str\n",
"\n",
"# Subgraph\n",
"\n",
"def subgraph_node_1(state: State):\n",
" return {\"foo\": state[\"foo\"] + \"bar\"}\n",
"\n",
"subgraph_builder = StateGraph(State)\n",
"subgraph_builder.add_node(subgraph_node_1)\n",
"subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
"# highlight-next-line\n",
"subgraph = subgraph_builder.compile()\n",
"\n",
"# Parent graph\n",
"\n",
"builder = StateGraph(State)\n",
"# highlight-next-line\n",
"builder.add_node(\"node_1\", subgraph)\n",
"builder.add_edge(START, \"node_1\")\n",
"\n",
"checkpointer = InMemorySaver()\n",
"# highlight-next-line\n",
"graph = builder.compile(checkpointer=checkpointer)\n",
"``` \n",
"\n",
"If you want the subgraph to **have its own memory**, you can compile it `with checkpointer=True`. This is useful in [multi-agent](../../concepts/multi_agent) systems, if you want agents to keep track of their internal message histories:\n",
"\n",
"```python\n",
"subgraph_builder = StateGraph(...)\n",
"# highlight-next-line\n",
"subgraph = subgraph_builder.compile(checkpointer=True)\n",
"```"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## View subgraph state\n",
"\n",
"When you enable [persistence](../persistence), you can [inspect the graph state](../persistence#manage-checkpoints) (checkpoint) via `graph.get_state(config)`. To view the subgraph state, you can use `graph.get_state(config, subgraphs=True)`.\n",
"\n",
"!!! important \"Available **only** when interrupted\"\n",
"\n",
" Subgraph state can only be viewed **when the subgraph is interrupted**. Once you resume the graph, you won't be able to access the subgraph state.\n",
"\n",
"??? example \"View interrupted subgraph state\"\n",
"\n",
" ```python\n",
" from langgraph.graph import START, StateGraph\n",
" from langgraph.checkpoint.memory import InMemorySaver\n",
" from langgraph.types import interrupt, Command\n",
" from typing_extensions import TypedDict\n",
" \n",
" class State(TypedDict):\n",
" foo: str\n",
" \n",
" # Subgraph\n",
" \n",
" def subgraph_node_1(state: State):\n",
" # highlight-next-line\n",
" value = interrupt(\"Provide value:\")\n",
" return {\"foo\": state[\"foo\"] + value}\n",
" \n",
" subgraph_builder = StateGraph(State)\n",
" subgraph_builder.add_node(subgraph_node_1)\n",
" subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
" \n",
" subgraph = subgraph_builder.compile()\n",
" \n",
" # Parent graph\n",
" \n",
" builder = StateGraph(State)\n",
" # highlight-next-line\n",
" builder.add_node(\"node_1\", subgraph)\n",
" builder.add_edge(START, \"node_1\")\n",
" \n",
" checkpointer = InMemorySaver()\n",
" # highlight-next-line\n",
" graph = builder.compile(checkpointer=checkpointer)\n",
" \n",
" config = {\"configurable\": {\"thread_id\": \"1\"}}\n",
" \n",
" graph.invoke({\"foo\": \"\"}, config)\n",
" parent_state = graph.get_state(config)\n",
" # highlight-next-line\n",
" subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state # (1)!\n",
" \n",
" # resume the subgraph\n",
" graph.invoke(Command(resume=\"bar\"), config)\n",
" ```\n",
" \n",
" 1. This will be available only when the subgraph is interrupted. Once you resume the graph, you won't be able to access the subgraph state."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Stream subgraph outputs\n",
"\n",
"To include outputs from [subgraphs](../concepts/low_level.md#subgraphs) in the streamed outputs, you can set `subgraphs=True` in the `.stream()` method of the parent graph. This will stream outputs from both the parent graph and any subgraphs.\n",
"\n",
"```python\n",
"for chunk in graph.stream(\n",
" {\"foo\": \"foo\"},\n",
" # highlight-next-line\n",
" subgraphs=True, # (1)!\n",
" stream_mode=\"updates\",\n",
"):\n",
" print(chunk)\n",
"```\n",
"\n",
"1. Set `subgraphs=True` to stream outputs from subgraphs.\n",
"\n",
"??? example \"Stream from subgraphs\"\n",
"\n",
" ```python\n",
" from typing_extensions import TypedDict\n",
" from langgraph.graph.state import StateGraph, START\n",
"\n",
" # Define subgraph\n",
" class SubgraphState(TypedDict):\n",
" foo: str\n",
" bar: str\n",
" \n",
" def subgraph_node_1(state: SubgraphState):\n",
" return {\"bar\": \"bar\"}\n",
" \n",
" def subgraph_node_2(state: SubgraphState):\n",
" # note that this node is using a state key ('bar') that is only available in the subgraph\n",
" # and is sending update on the shared state key ('foo')\n",
" return {\"foo\": state[\"foo\"] + state[\"bar\"]}\n",
" \n",
" subgraph_builder = StateGraph(SubgraphState)\n",
" subgraph_builder.add_node(subgraph_node_1)\n",
" subgraph_builder.add_node(subgraph_node_2)\n",
" subgraph_builder.add_edge(START, \"subgraph_node_1\")\n",
" subgraph_builder.add_edge(\"subgraph_node_1\", \"subgraph_node_2\")\n",
" subgraph = subgraph_builder.compile()\n",
" \n",
" # Define parent graph\n",
" class ParentState(TypedDict):\n",
" foo: str\n",
" \n",
" def node_1(state: ParentState):\n",
" return {\"foo\": \"hi! \" + state[\"foo\"]}\n",
" \n",
" builder = StateGraph(ParentState)\n",
" builder.add_node(\"node_1\", node_1)\n",
" # highlight-next-line\n",
" builder.add_node(\"node_2\", subgraph)\n",
" builder.add_edge(START, \"node_1\")\n",
" builder.add_edge(\"node_1\", \"node_2\")\n",
" graph = builder.compile()\n",
"\n",
" for chunk in graph.stream(\n",
" {\"foo\": \"foo\"},\n",
" stream_mode=\"updates\",\n",
" # highlight-next-line\n",
" subgraphs=True, # (1)!\n",
" ):\n",
" print(chunk)\n",
" ```\n",
" \n",
" 1. Set `subgraphs=True` to stream outputs from subgraphs.\n",
"\n",
" ```\n",
" ((), {'node_1': {'foo': 'hi! foo'}})\n",
" (('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_1': {'bar': 'bar'}})\n",
" (('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_2': {'foo': 'hi! foobar'}})\n",
" ((), {'node_2': {'foo': 'hi! foobar'}})\n",
" ```"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.3"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
+453
View File
@@ -0,0 +1,453 @@
# Use subgraphs
This guide explains the mechanics of using [subgraphs](../concepts/subgraphs.md). A common application of subgraphs is to build [multi-agent](../concepts/multi_agent.md) systems.
When adding subgraphs, you need to define how the parent graph and the subgraph communicate:
* [Shared state schemas](#shared-state-schemas) — parent and subgraph have **shared state keys** in their state [schemas](../concepts/low_level.md#state)
* [Different state schemas](#different-state-schemas) — **no shared state keys** in parent and subgraph [schemas](../concepts/low_level.md#state)
## Setup
```bash
pip install -U langgraph
```
!!! tip "Set up LangSmith for LangGraph development"
Sign up for [LangSmith](https://smith.langchain.com) to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph — read more about how to get started [here](https://docs.smith.langchain.com).
## Shared state schemas
A common case is for the parent graph and subgraph to communicate over a shared state key (channel) in the [schema](../concepts/low_level.md#state). For example, in [multi-agent](../concepts/multi_agent.md) systems, the agents often communicate over a shared [messages](https://langchain-ai.github.io/langgraph/concepts/low_level.md#why-use-messages) key.
If your subgraph shares state keys with the parent graph, you can follow these steps to add it to your graph:
1. Define the subgraph workflow (`subgraph_builder` in the example below) and compile it
2. Pass compiled subgraph to the `.add_node` method when defining the parent graph workflow
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
class State(TypedDict):
foo: str
# Subgraph
def subgraph_node_1(state: State):
return {"foo": "hi! " + state["foo"]}
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# Parent graph
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()
```
??? example "Full example: shared state schemas"
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
# Define subgraph
class SubgraphState(TypedDict):
foo: str # (1)!
bar: str # (2)!
def subgraph_node_1(state: SubgraphState):
return {"bar": "bar"}
def subgraph_node_2(state: SubgraphState):
# note that this node is using a state key ('bar') that is only available in the subgraph
# and is sending update on the shared state key ('foo')
return {"foo": state["foo"] + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# Define parent graph
class ParentState(TypedDict):
foo: str
def node_1(state: ParentState):
return {"foo": "hi! " + state["foo"]}
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()
for chunk in graph.stream({"foo": "foo"}):
print(chunk)
```
1. This key is shared with the parent graph state
2. This key is private to the `SubgraphState` and is not visible to the parent graph
```
{'node_1': {'foo': 'hi! foo'}}
{'node_2': {'foo': 'hi! foobar'}}
```
## Different state schemas
For more complex systems you might want to define subgraphs that have a **completely different schema** from the parent graph (no shared keys). For example, you might want to keep a private message history for each of the agents in a [multi-agent](../concepts/multi_agent.md) system.
If that's the case for your application, you need to define a node **function that invokes the subgraph**. This function needs to transform the input (parent) state to the subgraph state before invoking the subgraph, and transform the results back to the parent state before returning the state update from the node.
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
class SubgraphState(TypedDict):
bar: str
# Subgraph
def subgraph_node_1(state: SubgraphState):
return {"bar": "hi! " + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# Parent graph
class State(TypedDict):
foo: str
def call_subgraph(state: State):
subgraph_output = subgraph.invoke({"bar": state["foo"]}) # (1)!
return {"foo": subgraph_output["bar"]} # (2)!
builder = StateGraph(State)
builder.add_node("node_1", call_subgraph)
builder.add_edge(START, "node_1")
graph = builder.compile()
```
1. Transform the state to the subgraph state
2. Transform response back to the parent state
??? example "Full example: different state schemas"
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
# Define subgraph
class SubgraphState(TypedDict):
# note that none of these keys are shared with the parent graph state
bar: str
baz: str
def subgraph_node_1(state: SubgraphState):
return {"baz": "baz"}
def subgraph_node_2(state: SubgraphState):
return {"bar": state["bar"] + state["baz"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# Define parent graph
class ParentState(TypedDict):
foo: str
def node_1(state: ParentState):
return {"foo": "hi! " + state["foo"]}
def node_2(state: ParentState):
response = subgraph.invoke({"bar": state["foo"]}) # (1)!
return {"foo": response["bar"]} # (2)!
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()
for chunk in graph.stream({"foo": "foo"}, subgraphs=True):
print(chunk)
```
1. Transform the state to the subgraph state
2. Transform response back to the parent state
```
((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})
(('node_2:9c36dd0f-151a-cb42-cbad-fa2f851f9ab7',), {'grandchild_2': {'bar': 'hi! foobaz'}})
((), {'node_2': {'foo': 'hi! foobaz'}})
```
??? example "Full example: different state schemas (two levels of subgraphs)"
This is an example with two levels of subgraphs: parent -> child -> grandchild.
```python
# Grandchild graph
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START, END
class GrandChildState(TypedDict):
my_grandchild_key: str
def grandchild_1(state: GrandChildState) -> GrandChildState:
# NOTE: child or parent keys will not be accessible here
return {"my_grandchild_key": state["my_grandchild_key"] + ", how are you"}
grandchild = StateGraph(GrandChildState)
grandchild.add_node("grandchild_1", grandchild_1)
grandchild.add_edge(START, "grandchild_1")
grandchild.add_edge("grandchild_1", END)
grandchild_graph = grandchild.compile()
# Child graph
class ChildState(TypedDict):
my_child_key: str
def call_grandchild_graph(state: ChildState) -> ChildState:
# NOTE: parent or grandchild keys won't be accessible here
grandchild_graph_input = {"my_grandchild_key": state["my_child_key"]} # (1)!
grandchild_graph_output = grandchild_graph.invoke(grandchild_graph_input)
return {"my_child_key": grandchild_graph_output["my_grandchild_key"] + " today?"} # (2)!
child = StateGraph(ChildState)
child.add_node("child_1", call_grandchild_graph) # (3)!
child.add_edge(START, "child_1")
child.add_edge("child_1", END)
child_graph = child.compile()
# Parent graph
class ParentState(TypedDict):
my_key: str
def parent_1(state: ParentState) -> ParentState:
# NOTE: child or grandchild keys won't be accessible here
return {"my_key": "hi " + state["my_key"]}
def parent_2(state: ParentState) -> ParentState:
return {"my_key": state["my_key"] + " bye!"}
def call_child_graph(state: ParentState) -> ParentState:
child_graph_input = {"my_child_key": state["my_key"]} # (4)!
child_graph_output = child_graph.invoke(child_graph_input)
return {"my_key": child_graph_output["my_child_key"]} # (5)!
parent = StateGraph(ParentState)
parent.add_node("parent_1", parent_1)
parent.add_node("child", call_child_graph) # (6)!
parent.add_node("parent_2", parent_2)
parent.add_edge(START, "parent_1")
parent.add_edge("parent_1", "child")
parent.add_edge("child", "parent_2")
parent.add_edge("parent_2", END)
parent_graph = parent.compile()
for chunk in parent_graph.stream({"my_key": "Bob"}, subgraphs=True):
print(chunk)
```
1. We're transforming the state from the child state channels (`my_child_key`) to the child state channels (`my_grandchild_key`)
2. We're transforming the state from the grandchild state channels (`my_grandchild_key`) back to the child state channels (`my_child_key`)
3. We're passing a function here instead of just compiled graph (`grandchild_graph`)
4. We're transforming the state from the parent state channels (`my_key`) to the child state channels (`my_child_key`)
5. We're transforming the state from the child state channels (`my_child_key`) back to the parent state channels (`my_key`)
6. We're passing a function here instead of just a compiled graph (`child_graph`)
```
((), {'parent_1': {'my_key': 'hi Bob'}})
(('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b', 'child_1:781bb3b1-3971-84ce-810b-acf819a03f9c'), {'grandchild_1': {'my_grandchild_key': 'hi Bob, how are you'}})
(('child:2e26e9ce-602f-862c-aa66-1ea5a4655e3b',), {'child_1': {'my_child_key': 'hi Bob, how are you today?'}})
((), {'child': {'my_key': 'hi Bob, how are you today?'}})
((), {'parent_2': {'my_key': 'hi Bob, how are you today? bye!'}})
```
## Add persistence
You only need to **provide the checkpointer when compiling the parent graph**. LangGraph will automatically propagate the checkpointer to the child subgraphs.
```python
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict
class State(TypedDict):
foo: str
# Subgraph
def subgraph_node_1(state: State):
return {"foo": state["foo"] + "bar"}
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# Parent graph
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
```
If you want the subgraph to **have its own memory**, you can compile it `with checkpointer=True`. This is useful in [multi-agent](../concepts/multi_agent.md) systems, if you want agents to keep track of their internal message histories:
```python
subgraph_builder = StateGraph(...)
subgraph = subgraph_builder.compile(checkpointer=True)
```
## View subgraph state
When you enable [persistence](../concepts/persistence.md), you can [inspect the graph state](../concepts/persistence.md#checkpoints) (checkpoint) via `graph.get_state(config)`. To view the subgraph state, you can use `graph.get_state(config, subgraphs=True)`.
!!! important "Available **only** when interrupted"
Subgraph state can only be viewed **when the subgraph is interrupted**. Once you resume the graph, you won't be able to access the subgraph state.
??? example "View interrupted subgraph state"
```python
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from typing_extensions import TypedDict
class State(TypedDict):
foo: str
# Subgraph
def subgraph_node_1(state: State):
value = interrupt("Provide value:")
return {"foo": state["foo"] + value}
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# Parent graph
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"foo": ""}, config)
parent_state = graph.get_state(config)
subgraph_state = graph.get_state(config, subgraphs=True).tasks[0].state # (1)!
# resume the subgraph
graph.invoke(Command(resume="bar"), config)
```
1. This will be available only when the subgraph is interrupted. Once you resume the graph, you won't be able to access the subgraph state.
## Stream subgraph outputs
To include outputs from subgraphs in the streamed outputs, you can set `subgraphs=True` in the `.stream()` method of the parent graph. This will stream outputs from both the parent graph and any subgraphs.
```python
for chunk in graph.stream(
{"foo": "foo"},
subgraphs=True, # (1)!
stream_mode="updates",
):
print(chunk)
```
1. Set `subgraphs=True` to stream outputs from subgraphs.
??? example "Stream from subgraphs"
```python
from typing_extensions import TypedDict
from langgraph.graph.state import StateGraph, START
# Define subgraph
class SubgraphState(TypedDict):
foo: str
bar: str
def subgraph_node_1(state: SubgraphState):
return {"bar": "bar"}
def subgraph_node_2(state: SubgraphState):
# note that this node is using a state key ('bar') that is only available in the subgraph
# and is sending update on the shared state key ('foo')
return {"foo": state["foo"] + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# Define parent graph
class ParentState(TypedDict):
foo: str
def node_1(state: ParentState):
return {"foo": "hi! " + state["foo"]}
builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()
for chunk in graph.stream(
{"foo": "foo"},
stream_mode="updates",
subgraphs=True, # (1)!
):
print(chunk)
```
1. Set `subgraphs=True` to stream outputs from subgraphs.
```
((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_1': {'bar': 'bar'}})
(('node_2:e58e5673-a661-ebb0-70d4-e298a7fc28b7',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
((), {'node_2': {'foo': 'hi! foobar'}})
+3 -3
View File
@@ -112,7 +112,7 @@ nav:
- LangGraph APIs:
- Graph API:
- Overview: concepts/low_level.md
- Use the Graph API: how-tos/graph-api.ipynb
- Use the Graph API: how-tos/graph-api.md
- Functional API:
- Overview: concepts/functional_api.md
- Use the Functional API: how-tos/use-functional-api.md
@@ -150,11 +150,11 @@ nav:
- Use Server API: cloud/how-tos/human_in_the_loop_time_travel.md
- Subgraphs:
- Overview: concepts/subgraphs.md
- Use subgraphs: how-tos/subgraph.ipynb
- Use subgraphs: how-tos/subgraph.md
- Multi-agent:
- Overview: concepts/multi_agent.md
- Prebuilt implementation: agents/multi-agent.md
- Custom implementation: how-tos/multi_agent.ipynb
- Custom implementation: how-tos/multi_agent.md
- MCP:
- Overview: concepts/mcp.md
- Use MCP: agents/mcp.md
Generated
+1 -1
View File
@@ -2590,7 +2590,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "0.5.0"
version = "0.5.1"
source = { editable = "../libs/langgraph" }
dependencies = [
{ name = "langchain-core" },
+70
View File
@@ -0,0 +1,70 @@
#!/usr/bin/env python3
"""
Script to extract images from the graph-api.ipynb notebook and save them to assets folder.
"""
import json
import base64
import os
from pathlib import Path
def extract_images_from_notebook(notebook_path, assets_dir):
"""Extract images from notebook and save them to assets directory."""
# Read the notebook
with open(notebook_path, 'r') as f:
notebook = json.load(f)
# Create assets directory if it doesn't exist
os.makedirs(assets_dir, exist_ok=True)
image_count = 0
# Process each cell
for cell_idx, cell in enumerate(notebook['cells']):
if cell['cell_type'] == 'code':
# Check if this cell contains draw_mermaid_png
source = ''.join(cell.get('source', []))
if 'draw_mermaid_png' in source:
print(f"Found draw_mermaid_png in cell {cell_idx}")
# Check for outputs with images
if 'outputs' in cell:
for output_idx, output in enumerate(cell['outputs']):
if output.get('output_type') == 'display_data':
data = output.get('data', {})
# Check for PNG data
if 'image/png' in data:
png_data = data['image/png']
# Decode base64 data
try:
image_bytes = base64.b64decode(png_data)
# Generate filename
image_count += 1
filename = f"graph_api_image_{image_count}.png"
filepath = os.path.join(assets_dir, filename)
# Save the image
with open(filepath, 'wb') as img_file:
img_file.write(image_bytes)
print(f"Saved image: {filepath}")
except Exception as e:
print(f"Error decoding image {image_count}: {e}")
print(f"Extracted {image_count} images to {assets_dir}")
return image_count
if __name__ == "__main__":
notebook_path = "docs/docs/how-tos/graph-api.ipynb"
assets_dir = "docs/docs/how-tos/assets"
if os.path.exists(notebook_path):
count = extract_images_from_notebook(notebook_path, assets_dir)
print(f"Successfully extracted {count} images")
else:
print(f"Notebook not found: {notebook_path}")