WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-30 21:45:08 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into bagatur/optional_conditional_edge_mapping

Browse Source
This commit is contained in:
Bagatur
2024-01-26 18:40:53 -08:00
parent 9cbc67fecc 4969f7f62a
commit d99debe61c
3 changed files with 311 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/streaming-tokens.ipynb
+43 -112
View File
@@ -198,40 +198,6 @@
"model = model.bind_functions(functions)"
]
},
{
"cell_type": "markdown",
"id": "8e8b9211-93d0-4ad5-aa7a-9c09099c53ff",
"metadata": {},
"source": [
"## Define the agent state\n",
"\n",
"The main type of graph in `langgraph` is the `StatefulGraph`.\n",
"This graph is parameterized by a state object that it passes around to each node.\n",
"Each node then returns operations to update that state.\n",
"These operations can either SET specific attributes on the state (e.g. overwrite the existing values) or ADD to the existing attribute.\n",
"Whether to set or add is denoted by annotating the state object you construct the graph with.\n",
"\n",
"For this example, the state we will track will just be a list of messages.\n",
"We want each node to just add messages to that list.\n",
"Therefore, we will use a `TypedDict` with one key (`messages`) and annotate it so that the `messages` attribute is always added to.\n"
]
},
{
"cell_type": "code",
"execution_count": 5,
"id": "ea793afa-2eab-4901-910d-6eed90cd6564",
"metadata": {},
"outputs": [],
"source": [
"from typing import TypedDict, Annotated, Sequence\n",
"import operator\n",
"from langchain_core.messages import BaseMessage\n",
"\n",
"\n",
"class AgentState(TypedDict):\n",
" messages: Annotated[Sequence[BaseMessage], operator.add]"
]
},
{
"cell_type": "markdown",
"id": "e03c5094-9297-4d19-a04e-3eedc75cefb4",
@@ -265,7 +231,7 @@
},
{
"cell_type": "code",
"execution_count": 6,
"execution_count": 5,
"id": "3b541bb9-900c-40d0-964d-7b5dfee30667",
"metadata": {},
"outputs": [],
@@ -275,8 +241,7 @@
"from langchain_core.messages import FunctionMessage\n",
"\n",
"# Define the function that determines whether to continue or not\n",
"def should_continue(state):\n",
" messages = state['messages']\n",
"def should_continue(messages):\n",
" last_message = messages[-1]\n",
" # If there is no function call, then we finish\n",
" if \"function_call\" not in last_message.additional_kwargs:\n",
@@ -286,15 +251,13 @@
" return \"continue\"\n",
"\n",
"# Define the function that calls the model\n",
"async def call_model(state):\n",
" messages = state['messages']\n",
"async def call_model(messages):\n",
" response = await model.ainvoke(messages)\n",
" # We return a list, because this will get added to the existing list\n",
" return {\"messages\": [response]}\n",
" return response\n",
"\n",
"# Define the function to execute tools\n",
"async def call_tool(state):\n",
" messages = state['messages']\n",
"async def call_tool(messages):\n",
" # Based on the continue condition\n",
" # we know the last message involves a function call\n",
" last_message = messages[-1]\n",
@@ -308,7 +271,7 @@
" # We use the response to create a FunctionMessage\n",
" function_message = FunctionMessage(content=str(response), name=action.tool)\n",
" # We return a list, because this will get added to the existing list\n",
" return {\"messages\": [function_message]}"
" return function_message"
]
},
{
@@ -323,14 +286,14 @@
},
{
"cell_type": "code",
"execution_count": 7,
"execution_count": 6,
"id": "813ae66c-3b58-4283-a02a-36da72a2ab90",
"metadata": {},
"outputs": [],
"source": [
"from langgraph.graph import StateGraph, END\n",
"from langgraph.graph import MessageGraph, END\n",
"# Define a new graph\n",
"workflow = StateGraph(AgentState)\n",
"workflow = MessageGraph()\n",
"\n",
"# Define the two nodes we will cycle between\n",
"workflow.add_node(\"agent\", call_model)\n",
@@ -385,84 +348,52 @@
},
{
"cell_type": "code",
"execution_count": 10,
"execution_count": 7,
"id": "cfd140f0-a5a6-4697-8115-322242f197b5",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"/Users/harrisonchase/workplace/langchain/libs/core/langchain_core/_api/beta_decorator.py:86: LangChainBetaWarning: This API is in beta and may change in the future.\n",
" warn_beta(\n"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"content='' additional_kwargs={'function_call': {'arguments': '', 'name': 'tavily_search_results_json'}}\n",
"content='' additional_kwargs={'function_call': {'arguments': '{\\n', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': ' ', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': ' \"', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': 'query', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': '\":', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': ' \"', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': 'weather', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': ' in', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': ' San', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': ' Francisco', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': '\"\\n', 'name': ''}}\n",
"content='' additional_kwargs={'function_call': {'arguments': '}', 'name': ''}}\n",
"content=''\n",
"content=''\n",
"content='I'\n",
"content=\"'m\"\n",
"content=' sorry'\n",
"content=','\n",
"content=' but'\n",
"content=' I'\n",
"content=' couldn'\n",
"content=\"'t\"\n",
"content=' find'\n",
"content=' the'\n",
"content=' current'\n",
"content=' weather'\n",
"content=' in'\n",
"content=' San'\n",
"content=' Francisco'\n",
"content='.'\n",
"content=' However'\n",
"content=','\n",
"content=' you'\n",
"content=' can'\n",
"content=' check'\n",
"content=' the'\n",
"content=' weather'\n",
"content=' forecast'\n",
"content=' for'\n",
"content=' San'\n",
"content=' Francisco'\n",
"content=' on'\n",
"content=' websites'\n",
"content=' like'\n",
"content=' Weather'\n",
"content='.com'\n",
"content=' or'\n",
"content=' Acc'\n",
"content='u'\n",
"content='Weather'\n",
"content='.'\n",
"content=''\n"
"--\n",
"Starting tool: tavily_search_results_json with inputs: {'query': 'weather in San Francisco'}\n",
"Done tool: tavily_search_results_json\n",
"Tool output was: [{'url': 'https://www.whereandwhen.net/when/north-america/california/san-francisco-ca/january/', 'content': 'Best time to go to San Francisco? Weather in San Francisco in january 2024 How was the weather last january? Here is the day by day recorded weather in San Francisco in january 2023: Seasonal average climate and temperature of San Francisco in january 8% 46% 29% 12% 8% Evolution of daily average temperature and precipitation in San Francisco in januaryWeather in San Francisco in january 2024. The weather in San Francisco in january comes from statistical datas on the past years. You can view the weather statistics the entire month, but also by using the tabs for the beginning, the middle and the end of the month. ... 23-01-2023 47°F to 61°F. 24-01-2023 43°F to 58°F. 25-01-2023 47°F to ...'}]\n",
"--\n",
"I|'m| sorry|,| but| I| couldn|'t| find| the| current| weather| in| San| Francisco|.| However|,| you| can| check| the| weather| in| San| Francisco| for| the| month| of| January| on| this| website|:| [|San| Francisco| Weather| in| January|](|https|://|www|.where|and|when|.net|/|when|/n|orth|-|amer|ica|/cal|ifornia|/s|an|-fr|anc|isco|-ca|/j|an|uary|/|).|"
]
}
],
"source": [
"from langchain_core.messages import HumanMessage\n",
"inputs = {\"messages\": [HumanMessage(content=\"what is the weather in sf\")]}\n",
"async for output in app.astream_log(inputs, include_types=[\"llm\"]):\n",
" # astream_log() yields the requested logs (here LLMs) in JSONPatch format\n",
" for op in output.ops:\n",
" if op[\"path\"] == \"/streamed_output/-\":\n",
" # this is the output from .stream()\n",
" ...\n",
" elif op[\"path\"].startswith(\"/logs/\") and op[\"path\"].endswith(\n",
" \"/streamed_output/-\"\n",
" ):\n",
" # because we chose to only include LLMs, these are LLM tokens\n",
" print(op[\"value\"])"
"inputs = [HumanMessage(content=\"what is the weather in sf\")]\n",
"async for event in app.astream_events(inputs, version=\"v1\"):\n",
" kind = event[\"event\"]\n",
" if kind == \"on_chat_model_stream\":\n",
" content = event[\"data\"][\"chunk\"].content\n",
" if content:\n",
" # Empty content in the context of OpenAI means\n",
" # that the model is asking for a tool to be invoked.\n",
" # So we only print non-empty content\n",
" print(content, end=\"|\")\n",
" elif kind == \"on_tool_start\":\n",
" print(\"--\")\n",
" print(\n",
" f\"Starting tool: {event['name']} with inputs: {event['data'].get('input')}\"\n",
" )\n",
" elif kind == \"on_tool_end\":\n",
" print(f\"Done tool: {event['name']}\")\n",
" print(f\"Tool output was: {event['data'].get('output')}\")\n",
" print(\"--\")"
]
},
{
langgraph/pregel/__init__.py
+267 -229
View File
@@ -253,127 +253,144 @@ class Pregel(
input_keys: Optional[Union[str, Sequence[str]]] = None,
output_keys: Optional[Union[str, Sequence[str]]] = None,
) -> Iterator[Union[dict[str, Any], Any]]:
if config["recursion_limit"] < 1:
raise ValueError("recursion_limit must be at least 1")
# assign defaults
if output_keys is None:
output_keys = [chan for chan in self.channels if chan not in self.hidden]
else:
validate_keys(output_keys, self.channels)
if input_keys is None:
input_keys = self.input
else:
validate_keys(input_keys, self.channels)
# copy nodes to ignore mutations during execution
processes = {**self.nodes}
# get checkpoint from saver, or create an empty one
checkpoint = self.checkpointer.get(config) if self.checkpointer else None
checkpoint = checkpoint or empty_checkpoint()
# create channels from checkpoint
with ChannelsManager(
self.channels, checkpoint
) as channels, get_executor_for_config(config) as executor:
# map inputs to channel updates
_apply_writes(
checkpoint,
channels,
deque(w for c in input for w in map_input(input_keys, c)),
config,
0,
)
read = partial(_read_channel, channels)
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates
# channel updates from step N are only visible in step N+1
# channels are guaranteed to be immutable for the duration of the step,
# with channel updates applied only at the transition between steps
for step in range(config["recursion_limit"] + 1):
next_tasks = _prepare_next_tasks(checkpoint, processes, channels)
# if no more tasks, we're done
if not next_tasks:
break
elif step == config["recursion_limit"]:
raise GraphRecursionError(
f"Recursion limit of {config['recursion_limit']} reached"
"without hitting a stop condition. You can increase the limit"
"by setting the `recursion_limit` config key."
)
if self.debug:
print_step_start(step, next_tasks)
# collect all writes to channels, without applying them yet
pending_writes = deque[tuple[str, Any]]()
# prepare tasks with config
tasks_w_config = [
(
proc,
input,
patch_config(
config,
run_name=name,
callbacks=run_manager.get_child(f"graph:step:{step}"),
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: pending_writes.extend,
CONFIG_KEY_READ: read,
},
),
)
for proc, input, name in next_tasks
try:
if config["recursion_limit"] < 1:
raise ValueError("recursion_limit must be at least 1")
# assign defaults
if output_keys is None:
output_keys = [
chan for chan in self.channels if chan not in self.hidden
]
# execute tasks, and wait for one to fail or all to finish.
# each task is independent from all other concurrent tasks
done, inflight = concurrent.futures.wait(
[
executor.submit(proc.invoke, input, config)
for proc, input, config in tasks_w_config
],
return_when=concurrent.futures.FIRST_EXCEPTION,
timeout=self.step_timeout,
else:
validate_keys(output_keys, self.channels)
if input_keys is None:
input_keys = self.input
else:
validate_keys(input_keys, self.channels)
# copy nodes to ignore mutations during execution
processes = {**self.nodes}
# get checkpoint from saver, or create an empty one
checkpoint = self.checkpointer.get(config) if self.checkpointer else None
checkpoint = checkpoint or empty_checkpoint()
# create channels from checkpoint
with ChannelsManager(
self.channels, checkpoint
) as channels, get_executor_for_config(config) as executor:
# map inputs to channel updates
_apply_writes(
checkpoint,
channels,
deque(w for c in input for w in map_input(input_keys, c)),
config,
0,
)
# interrupt on failure or timeout
_interrupt_or_proceed(done, inflight, step)
read = partial(_read_channel, channels)
# apply writes to channels
_apply_writes(checkpoint, channels, pending_writes, config, step + 1)
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates
# channel updates from step N are only visible in step N+1
# channels are guaranteed to be immutable for the duration of the step,
# with channel updates applied only at the transition between steps
for step in range(config["recursion_limit"] + 1):
next_tasks = _prepare_next_tasks(checkpoint, processes, channels)
if self.debug:
print_checkpoint(step, channels)
# if no more tasks, we're done
if not next_tasks:
break
elif step == config["recursion_limit"]:
raise GraphRecursionError(
f"Recursion limit of {config['recursion_limit']} reached"
"without hitting a stop condition. You can increase the limit"
"by setting the `recursion_limit` config key."
)
# yield current value and checkpoint view
if step_output := map_output(output_keys, pending_writes, channels):
yield step_output
# we can detect updates when output is multiple channels (ie. dict)
if not isinstance(output_keys, str):
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, step_output)
if self.debug:
print_step_start(step, next_tasks)
# save end of step checkpoint
# collect all writes to channels, without applying them yet
pending_writes = deque[tuple[str, Any]]()
# prepare tasks with config
tasks_w_config = [
(
proc,
input,
patch_config(
config,
run_name=name,
callbacks=run_manager.get_child(f"graph:step:{step}"),
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: pending_writes.extend,
CONFIG_KEY_READ: read,
},
),
)
for proc, input, name in next_tasks
]
futures = [
executor.submit(proc.invoke, input, config)
for proc, input, config in tasks_w_config
]
# execute tasks, and wait for one to fail or all to finish.
# each task is independent from all other concurrent tasks
done, inflight = concurrent.futures.wait(
futures,
return_when=concurrent.futures.FIRST_EXCEPTION,
timeout=self.step_timeout,
)
# interrupt on failure or timeout
_interrupt_or_proceed(done, inflight, step)
# apply writes to channels
_apply_writes(
checkpoint, channels, pending_writes, config, step + 1
)
if self.debug:
print_checkpoint(step, channels)
# yield current value and checkpoint view
if step_output := map_output(output_keys, pending_writes, channels):
yield step_output
# we can detect updates when output is multiple channels (ie. dict)
if not isinstance(output_keys, str):
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, step_output)
# save end of step checkpoint
if (
self.checkpointer is not None
and self.checkpointer.at == CheckpointAt.END_OF_STEP
):
checkpoint = create_checkpoint(checkpoint, channels)
self.checkpointer.put(config, checkpoint)
# interrupt if any channel written to is in interrupt list
if any(
chan for chan, _ in pending_writes if chan in self.interrupt
):
break
# save end of run checkpoint
if (
self.checkpointer is not None
and self.checkpointer.at == CheckpointAt.END_OF_STEP
and self.checkpointer.at == CheckpointAt.END_OF_RUN
):
checkpoint = create_checkpoint(checkpoint, channels)
self.checkpointer.put(config, checkpoint)
# interrupt if any channel written to is in interrupt list
if any(chan for chan, _ in pending_writes if chan in self.interrupt):
break
# save end of run checkpoint
if (
self.checkpointer is not None
and self.checkpointer.at == CheckpointAt.END_OF_RUN
):
checkpoint = create_checkpoint(checkpoint, channels)
self.checkpointer.put(config, checkpoint)
finally:
# cancel any pending tasks when generator is interrupted
try:
futures
except NameError:
return
for task in futures:
task.cancel()
async def _atransform(
self,
@@ -384,139 +401,160 @@ class Pregel(
input_keys: Optional[Union[str, Sequence[str]]] = None,
output_keys: Optional[Union[str, Sequence[str]]] = None,
) -> AsyncIterator[Union[dict[str, Any], Any]]:
if config["recursion_limit"] < 1:
raise ValueError("recursion_limit must be at least 1")
# if running from astream_log() run each proc with streaming
do_stream = next(
(
h
for h in run_manager.handlers
if isinstance(h, LogStreamCallbackHandler)
),
None,
)
# assign defaults
if output_keys is None:
output_keys = [chan for chan in self.channels if chan not in self.hidden]
else:
validate_keys(output_keys, self.channels)
if input_keys is None:
input_keys = self.input
else:
validate_keys(input_keys, self.channels)
# copy nodes to ignore mutations during execution
processes = {**self.nodes}
# get checkpoint from saver, or create an empty one
checkpoint = await self.checkpointer.aget(config) if self.checkpointer else None
checkpoint = checkpoint or empty_checkpoint()
# create channels from checkpoint
async with AsyncChannelsManager(self.channels, checkpoint) as channels:
# map inputs to channel updates
_apply_writes(
checkpoint,
channels,
deque([w async for c in input for w in map_input(input_keys, c)]),
config,
0,
try:
if config["recursion_limit"] < 1:
raise ValueError("recursion_limit must be at least 1")
# if running from astream_log() run each proc with streaming
do_stream = next(
(
h
for h in run_manager.handlers
if isinstance(h, LogStreamCallbackHandler)
),
None,
)
read = partial(_read_channel, channels)
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates
# channel updates from step N are only visible in step N+1,
# channels are guaranteed to be immutable for the duration of the step,
# channel updates being applied only at the transition between steps
for step in range(config["recursion_limit"] + 1):
next_tasks = _prepare_next_tasks(checkpoint, processes, channels)
# if no more tasks, we're done
if not next_tasks:
break
elif step == config["recursion_limit"]:
raise GraphRecursionError(
f"Recursion limit of {config['recursion_limit']} reached"
"without hitting a stop condition. You can increase the limit"
"by setting the `recursion_limit` config key."
)
if self.debug:
print_step_start(step, next_tasks)
# collect all writes to channels, without applying them yet
pending_writes = deque[tuple[str, Any]]()
# prepare tasks with config
tasks_w_config = [
(
proc,
input,
patch_config(
config,
run_name=name,
callbacks=run_manager.get_child(f"graph:step:{step}"),
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: pending_writes.extend,
CONFIG_KEY_READ: read,
},
),
)
for proc, input, name in next_tasks
# assign defaults
if output_keys is None:
output_keys = [
chan for chan in self.channels if chan not in self.hidden
]
# execute tasks, and wait for one to fail or all to finish.
# each task is independent from all other concurrent tasks
done, inflight = await asyncio.wait(
[
asyncio.create_task(_aconsume(proc.astream(input, config)))
for proc, input, config in tasks_w_config
]
if do_stream
else [
asyncio.create_task(proc.ainvoke(input, config))
for proc, input, config in tasks_w_config
],
return_when=asyncio.FIRST_EXCEPTION,
timeout=self.step_timeout,
else:
validate_keys(output_keys, self.channels)
if input_keys is None:
input_keys = self.input
else:
validate_keys(input_keys, self.channels)
# copy nodes to ignore mutations during execution
processes = {**self.nodes}
# get checkpoint from saver, or create an empty one
checkpoint = (
await self.checkpointer.aget(config) if self.checkpointer else None
)
checkpoint = checkpoint or empty_checkpoint()
# create channels from checkpoint
async with AsyncChannelsManager(self.channels, checkpoint) as channels:
# map inputs to channel updates
_apply_writes(
checkpoint,
channels,
deque([w async for c in input for w in map_input(input_keys, c)]),
config,
0,
)
# interrupt on failure or timeout
_interrupt_or_proceed(done, inflight, step)
read = partial(_read_channel, channels)
# apply writes to channels
_apply_writes(checkpoint, channels, pending_writes, config, step + 1)
# Similarly to Bulk Synchronous Parallel / Pregel model
# computation proceeds in steps, while there are channel updates
# channel updates from step N are only visible in step N+1,
# channels are guaranteed to be immutable for the duration of the step,
# channel updates being applied only at the transition between steps
for step in range(config["recursion_limit"] + 1):
next_tasks = _prepare_next_tasks(checkpoint, processes, channels)
if self.debug:
print_checkpoint(step, channels)
# if no more tasks, we're done
if not next_tasks:
break
elif step == config["recursion_limit"]:
raise GraphRecursionError(
f"Recursion limit of {config['recursion_limit']} reached"
"without hitting a stop condition. You can increase the limit"
"by setting the `recursion_limit` config key."
)
# yield current value and checkpoint view
if step_output := map_output(output_keys, pending_writes, channels):
yield step_output
# we can detect updates when output is multiple channels (ie. dict)
if not isinstance(output_keys, str):
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, step_output)
if self.debug:
print_step_start(step, next_tasks)
# save end of step checkpoint
# collect all writes to channels, without applying them yet
pending_writes = deque[tuple[str, Any]]()
# prepare tasks with config
tasks_w_config = [
(
proc,
input,
patch_config(
config,
run_name=name,
callbacks=run_manager.get_child(f"graph:step:{step}"),
configurable={
# deque.extend is thread-safe
CONFIG_KEY_SEND: pending_writes.extend,
CONFIG_KEY_READ: read,
},
),
)
for proc, input, name in next_tasks
]
futures = (
[
asyncio.create_task(_aconsume(proc.astream(input, config)))
for proc, input, config in tasks_w_config
]
if do_stream
else [
asyncio.create_task(proc.ainvoke(input, config))
for proc, input, config in tasks_w_config
]
)
# execute tasks, and wait for one to fail or all to finish.
# each task is independent from all other concurrent tasks
done, inflight = await asyncio.wait(
futures,
return_when=asyncio.FIRST_EXCEPTION,
timeout=self.step_timeout,
)
# interrupt on failure or timeout
_interrupt_or_proceed(done, inflight, step)
# apply writes to channels
_apply_writes(
checkpoint, channels, pending_writes, config, step + 1
)
if self.debug:
print_checkpoint(step, channels)
# yield current value and checkpoint view
if step_output := map_output(output_keys, pending_writes, channels):
yield step_output
# we can detect updates when output is multiple channels (ie. dict)
if not isinstance(output_keys, str):
# if view was updated, apply writes to channels
_apply_writes_from_view(checkpoint, channels, step_output)
# save end of step checkpoint
if (
self.checkpointer is not None
and self.checkpointer.at == CheckpointAt.END_OF_STEP
):
checkpoint = create_checkpoint(checkpoint, channels)
await self.checkpointer.aput(config, checkpoint)
# interrupt if any channel written to is in interrupt list
if any(
chan for chan, _ in pending_writes if chan in self.interrupt
):
break
# save end of run checkpoint
if (
self.checkpointer is not None
and self.checkpointer.at == CheckpointAt.END_OF_STEP
and self.checkpointer.at == CheckpointAt.END_OF_RUN
):
checkpoint = create_checkpoint(checkpoint, channels)
await self.checkpointer.aput(config, checkpoint)
# interrupt if any channel written to is in interrupt list
if any(chan for chan, _ in pending_writes if chan in self.interrupt):
break
# save end of run checkpoint
if (
self.checkpointer is not None
and self.checkpointer.at == CheckpointAt.END_OF_RUN
):
checkpoint = create_checkpoint(checkpoint, channels)
await self.checkpointer.aput(config, checkpoint)
finally:
# cancel any pending tasks when generator is interrupted
try:
futures
except NameError:
return
for task in futures:
task.cancel()
def invoke(
self,
pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[tool.poetry]
name = "langgraph"
version = "0.0.17"
version = "0.0.19"
description = "langgraph"
authors = []
license = "LangGraph License"