This commit is contained in:
Eugene Yurtsev
2024-12-11 10:58:58 -05:00
parent f6c44ec154
commit 82fa597e84
@@ -1,621 +0,0 @@
{
"cells": [
{
"cell_type": "markdown",
"id": "0ff51c4b-5d5c-4b6e-8478-0ea489adb689",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"# How to use interrupt for human-in-the-loop?\n",
"\n",
"!!! tip \"Prerequisites\"\n",
"\n",
" This guide assumes familiarity with the following concepts:\n",
"\n",
" * [Breakpoints](../../../concepts/breakpoints)\n",
" * [Interrupt](../../../concepts/low_level#interrupt)\n",
" * [Command](../../../concepts/low_level#command)\n",
" \n",
"\n",
"An `interrupt` is a convenient way to support human-in-the-loop workflows.\n",
"\n",
"To use an `interrupt`, you must enable a checkpointer, as the feature relies on persisting the graph state.\n",
"\n",
"An `interrupt` can be used within a node to pause execution and wait for input, as shown in this example:\n",
"\n",
"```python\n",
"async def some_node(state: State):\n",
" ...\n",
" # Surface any value as part of the interrupt\n",
" value = {\"question\": \"how old are you?\"} \n",
" answer = interrupt(value)\n",
" ...\n",
"```\n",
"\n",
"Graph execution will pause when the interrupt function is called. To resume execution, pass a `Command` with the desired resume value:\n",
"\n",
"```python\n",
"for chunk in graph.stream(Command(resume=some_value), config={\"configurable\": {\"thread_id\": ...}}):\n",
" ...\n",
"```\n",
"\n",
"Remember that graph execution always restarts at the beginning of the node. Be cautious of side effects, such as API calls that mutate data, as these may inadvertently be triggered multiple times.\n",
"\n",
"When a node contains multiple interrupt calls, LangGraph maintains a list of resume values scoped to the specific task executing the node. When resuming, execution always starts at the beginning of the node, and for each interrupt encountered, LangGraph checks whether a corresponding value exists in the task's list. Matching is strictly index-based, making the order of interrupt calls within the node critical. Users should avoid logic that dynamically removes, adds, or reorders interrupt calls between executions, as this can lead to mismatched indices. Such patterns often involve unconventional state mutations, such as altering state via `Command(resume=..., update=SOME_STATE_MUTATION)` or relying on global variables to modify the node's structure."
]
},
{
"cell_type": "markdown",
"id": "17ecd33b-7879-47a8-9ad3-0b4ce1fdc0c9",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"## Setup\n",
"\n",
"First we need to install the required packages:"
]
},
{
"cell_type": "code",
"execution_count": 14,
"id": "cbc003a6-b45f-4526-bcf1-963d951797ae",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [],
"source": [
"%%capture --no-stderr\n",
"%pip install --quiet -U langgraph"
]
},
{
"cell_type": "markdown",
"id": "7091bfdc-2754-4b3c-83b9-20cfb1d2be66",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"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>"
]
},
{
"cell_type": "markdown",
"id": "8aa9eb6a-d54f-40e9-a618-0ad7290d15dc",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"## Basic usage of interrupt and Command\n",
"\n",
"Here is an example that shows how to use `interrupt` to interrupt the execution of a graph, and then resume the execution using the `Command` primitive."
]
},
{
"cell_type": "code",
"execution_count": 10,
"id": "ce902820-2739-402f-a784-867a44a3997c",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 1 # of times\n",
"{'__interrupt__': (Interrupt(value='what is your age?', resumable=True, ns=['node:62e598fa-8653-9d6d-2046-a70203020e37'], when='during'),)}\n"
]
}
],
"source": [
"import uuid\n",
"import operator\n",
"from typing import TypedDict, Annotated, Optional\n",
"\n",
"from langgraph.graph import StateGraph\n",
"from langgraph.constants import START, INTERRUPT\n",
"from langgraph.types import interrupt, Command\n",
"from langgraph.checkpoint.memory import MemorySaver\n",
"\n",
"\n",
"class State(TypedDict):\n",
" \"\"\"The graph state.\"\"\"\n",
"\n",
" foo: str\n",
" human_value: Optional[str]\n",
" \"\"\"Human value will be updated using an interrupt.\"\"\"\n",
"\n",
"\n",
"counter = 0\n",
"\n",
"\n",
"def node(state: State):\n",
" global counter\n",
" counter += 1\n",
" print(f\"> Entered the node: {counter} # of times\")\n",
" answer = interrupt(\n",
" # This value will be sent to the client\n",
" # as part of the interrupt information.\n",
" \"what is your age?\"\n",
" )\n",
" print(f\"> Received an input from the interrupt: {answer}\")\n",
" return {\"human_value\": answer}\n",
"\n",
"\n",
"builder = StateGraph(State)\n",
"builder.add_node(\"node\", node)\n",
"builder.add_edge(START, \"node\")\n",
"\n",
"# A checkpointer must be enabled for interrupts to work!\n",
"checkpointer = MemorySaver()\n",
"graph = builder.compile(checkpointer=checkpointer)\n",
"\n",
"config = {\n",
" \"configurable\": {\n",
" \"thread_id\": uuid.uuid4(),\n",
" }\n",
"}\n",
"\n",
"for chunk in graph.stream({\"foo\": \"abc\"}, config):\n",
" print(chunk)"
]
},
{
"cell_type": "markdown",
"id": "8df4319f-f7e8-40dc-8943-91fa3fad31a0",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"Let's resume graph execution from the given node:"
]
},
{
"cell_type": "code",
"execution_count": 11,
"id": "c2ea37d8-4b92-442d-85e1-212e20123907",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 2 # of times\n",
"> Received an input from the interrupt: some input from a human!!!\n",
"{'node': {'human_value': 'some input from a human!!!'}}\n"
]
}
],
"source": [
"command = Command(resume=\"some input from a human!!!\")\n",
"\n",
"for chunk in graph.stream(Command(resume=\"some input from a human!!!\"), config):\n",
" print(chunk)"
]
},
{
"cell_type": "markdown",
"id": "0036a06d-c875-461b-917d-1cd48c233e7d",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"!!! important \"Graph execution resumes at the start of a node\"\n",
"\n",
" Graph execution resumes from the start of the **node** where the interrupt was raised rather than from the line where the `interrupt` was raised.\n",
"\n",
" As a result, you should see that the node was entered 2 times rather than once!\n",
"\n",
" Exercise care if your code has side-effects like making mutable API calls between consecutive interrupts!"
]
},
{
"cell_type": "markdown",
"id": "9b40603f-5d01-41d7-afc7-2ead455a6803",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"## Using multiple interrupts calls within a single node\n",
"\n",
"In some situations, you may need to use interrupt more than once within a single node. A common use case is performing runtime validation on the value supplied through `Command(resume=value)`."
]
},
{
"cell_type": "code",
"execution_count": 12,
"id": "30b5821e-8bb8-4076-b477-60f39ea65445",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 1 # of times\n",
"{'__interrupt__': (Interrupt(value='What is your age?', resumable=True, ns=['node:ed4d470f-5753-d7f3-eec6-4435f5f93f72'], when='during'),)}\n"
]
}
],
"source": [
"import uuid\n",
"import operator\n",
"from typing import TypedDict, Annotated, Optional, Literal\n",
"\n",
"from langgraph.graph import StateGraph\n",
"from langgraph.constants import START\n",
"from langgraph.types import interrupt, Command\n",
"from langgraph.checkpoint.memory import MemorySaver\n",
"\n",
"\n",
"class State(TypedDict):\n",
" \"\"\"The graph state.\"\"\"\n",
"\n",
" foo: str\n",
" human_value: Optional[str]\n",
" \"\"\"Human value will be updated using an interrupt.\"\"\"\n",
"\n",
"\n",
"counter = 0\n",
"\n",
"\n",
"def node(state: State):\n",
" global counter\n",
" counter += 1\n",
" print(f\"> Entered the node: {counter} # of times\")\n",
"\n",
" answer = None\n",
" question = \"What is your age?\"\n",
"\n",
" while answer is None:\n",
" answer = interrupt(question)\n",
"\n",
" if not isinstance(answer, int) or answer < 0:\n",
" question = f\"'{answer} is not a valid age. What is your age?\"\n",
" answer = None\n",
" continue\n",
" else:\n",
" break\n",
"\n",
" return {\"human_value\": f\"The human is {answer} years old.\"}\n",
"\n",
"\n",
"builder = StateGraph(State)\n",
"builder.add_node(\"node\", node)\n",
"builder.add_edge(START, \"node\")\n",
"\n",
"# A checkpointer must be enabled for interrupts to work!\n",
"checkpointer = MemorySaver()\n",
"graph = builder.compile(checkpointer=checkpointer)\n",
"\n",
"config = {\n",
" \"configurable\": {\n",
" \"thread_id\": uuid.uuid4(),\n",
" }\n",
"}\n",
"\n",
"for chunk in graph.stream({\"foo\": \"abc\"}, config):\n",
" print(chunk)"
]
},
{
"cell_type": "markdown",
"id": "0d95194e-e9f2-4abf-83dc-c29b5f1f6f2d",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"Let's resume with a bad input"
]
},
{
"cell_type": "code",
"execution_count": 13,
"id": "a79e330d-f849-44ea-af37-221efcffe1a3",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 2 # of times\n",
"{'__interrupt__': (Interrupt(value=\"'-20 is not a valid age. What is your age?\", resumable=True, ns=['node:ed4d470f-5753-d7f3-eec6-4435f5f93f72'], when='during'),)}\n"
]
}
],
"source": [
"bad_input = -20 # Negative number!\n",
"for chunk in graph.stream(Command(resume=bad_input), config):\n",
" print(chunk)"
]
},
{
"cell_type": "code",
"execution_count": 14,
"id": "170b0cde-d94b-4aca-bfea-70f9927f4288",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 3 # of times\n",
"{'__interrupt__': (Interrupt(value=\"'{'foo': 'bar'} is not a valid age. What is your age?\", resumable=True, ns=['node:ed4d470f-5753-d7f3-eec6-4435f5f93f72'], when='during'),)}\n"
]
}
],
"source": [
"bad_input = {\"foo\": \"bar\"} # Not a number!\n",
"for chunk in graph.stream(Command(resume=bad_input), config):\n",
" print(chunk)"
]
},
{
"cell_type": "code",
"execution_count": 15,
"id": "c9d01c77-7a9c-43aa-bbfa-c969b1d3e3f2",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 4 # of times\n",
"{'node': {'human_value': 'The human is 25 years old.'}}\n"
]
}
],
"source": [
"ok_input = 25\n",
"for chunk in graph.stream(Command(resume=ok_input), config):\n",
" print(chunk)"
]
},
{
"cell_type": "markdown",
"id": "c7181eea-a0a8-43df-8ce6-b2172ea6cb21",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"source": [
"## Usage with invoke / ainvoke\n",
"\n",
"If you're using `invoke` and/or `ainvoke`, you will need to explicitly access the state of the graph using `graph.get_state(config)` to determine if there was an interrupt and if so what value it was associated with."
]
},
{
"cell_type": "code",
"execution_count": 21,
"id": "c9ae2c5f-c81d-4188-9f7e-f3f90e675e12",
"metadata": {
"editable": true,
"slideshow": {
"slide_type": ""
},
"tags": []
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 1 # of times\n"
]
}
],
"source": [
"import uuid\n",
"import operator\n",
"from typing import TypedDict, Annotated, Optional\n",
"\n",
"from langgraph.graph import StateGraph\n",
"from langgraph.constants import START, INTERRUPT\n",
"from langgraph.types import interrupt, Command\n",
"from langgraph.checkpoint.memory import MemorySaver\n",
"\n",
"\n",
"class State(TypedDict):\n",
" \"\"\"The graph state.\"\"\"\n",
"\n",
" foo: str\n",
" human_value: Optional[str]\n",
" \"\"\"Human value will be updated using an interrupt.\"\"\"\n",
"\n",
"\n",
"counter = 0\n",
"\n",
"\n",
"def node(state: State):\n",
" global counter\n",
" counter += 1\n",
" print(f\"> Entered the node: {counter} # of times\")\n",
" answer = interrupt(\n",
" # This value will be sent to the client\n",
" # as part of the interrupt information.\n",
" \"what is your age?\"\n",
" )\n",
" print(f\"> Received an input from the interrupt: {answer}\")\n",
" return {\"human_value\": answer}\n",
"\n",
"\n",
"builder = StateGraph(State)\n",
"builder.add_node(\"node\", node)\n",
"builder.add_edge(START, \"node\")\n",
"\n",
"# A checkpointer must be enabled for interrupts to work!\n",
"checkpointer = MemorySaver()\n",
"graph = builder.compile(checkpointer=checkpointer)\n",
"\n",
"config = {\n",
" \"configurable\": {\n",
" \"thread_id\": uuid.uuid4(),\n",
" }\n",
"}\n",
"\n",
"for event in graph.invoke({\"foo\": \"abc\"}, config):\n",
" print"
]
},
{
"cell_type": "code",
"execution_count": 22,
"id": "f15b73f1-259b-4045-b633-2686873ef1f6",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"('node',)\n",
"\n",
"PregelTask(id='57efb8b4-1170-b872-647e-60de07598d61', name='node', path=('__pregel_pull', 'node'), error=None, interrupts=(Interrupt(value='what is your age?', resumable=True, ns=['node:57efb8b4-1170-b872-647e-60de07598d61'], when='during'),), state=None, result=None)\n",
"\n",
"(Interrupt(value='what is your age?', resumable=True, ns=['node:57efb8b4-1170-b872-647e-60de07598d61'], when='during'),)\n"
]
}
],
"source": [
"state = graph.get_state(config)\n",
"\n",
"print(state.next)\n",
"print()\n",
"print(state.tasks[0])\n",
"print()\n",
"print(state.tasks[0].interrupts)"
]
},
{
"cell_type": "markdown",
"id": "113a746b-c9b7-4efa-ad6c-516f85b9cf5f",
"metadata": {},
"source": [
"Let's resume now:"
]
},
{
"cell_type": "code",
"execution_count": 23,
"id": "69a46d8f-ebe6-4ee0-84c9-30ce81d576c9",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"> Entered the node: 2 # of times\n",
"> Received an input from the interrupt: 25\n"
]
},
{
"data": {
"text/plain": [
"{'foo': 'abc', 'human_value': 25}"
]
},
"execution_count": 23,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"ok_input = 25\n",
"graph.invoke(Command(resume=ok_input), config)"
]
}
],
"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.11.4"
}
},
"nbformat": 4,
"nbformat_minor": 5
}