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

[breaking]: Improve interrupt behavior when stream_mode='values' (#4374)

Browse Source 
This PR does a few things:
1. Surfaces interrupts when `stream_mode='values'` (particularly
relevant for `invoke`, where this is the default behavior)
2. Adds an `interrupt_id` property to the `Interrupt` dataclass so that
interrupts can effectively be mapped to resumes
3. Minor docs updates to reflect the new pattern (no need for a special
section on interrupts with `invoke` and `ainvoke`)

* In a different PR (the one with the multiple resume values), as it's
more relevant there: add an `interrupts` property to `StateSnapshot` so
that `interrupts` can easily be iterated over if users are attempting to
map interrupts to resumes.

I **don't** recommend we release this until we have multi-resumes
working.

## Example

We have the following setup where we're sending multiple prompts to the
child graph, which uses `interrupt`:

```py
def child_graph(state):
    human_input = interrupt(state["prompt"])

    return {
        "human_inputs": [human_input],
    }
```

<img width="142" alt="Screenshot 2025-04-23 at 10 01 12 AM"
src="https://github.com/user-attachments/assets/c6238bf1-54ad-4e48-ab0b-60a0bfc18485"
/>

Old behavior:

```py
initial_input = {"prompts": ["a", "b"]}

print(parent_graph.invoke(input=initial_input,config=thread_config,stream_mode="values"))
#> {'prompts': ['a', 'b'], 'human_inputs': []}

print(parent_graph.invoke(Command(resume="hello 1"),config=thread_config,stream_mode="values"))
#> {'prompts': ['a', 'b'], 'human_inputs': ['hello 1']}

print(parent_graph.invoke(Command(resume="hello 2"),config=thread_config,stream_mode="values"))
#> {'prompts': ['a', 'b'], 'human_inputs': ['hello 1', 'hello 2']}
```

New behavior:

```py
initial_input = {"prompts": ["a", "b"]}

print(parent_graph.invoke(input=initial_input,config=thread_config,stream_mode="values"))
"""
{
  "prompts": ["a", "b"],
  "human_inputs": [],
  "__interrupt__": [
    Interrupt(
      value="a",
      resumable=True,
      ns=["child_graph:38d43a18-a5e7-8ab2-ca83-9d80f6e9ca83"]
    ),
    Interrupt(
      value="b",
      resumable=True,
      ns=["child_graph:dad810e8-738e-9f90-41cd-30c0091eb79b"]
    )
  ]
}
"""

print(parent_graph.invoke(Command(resume="hello 1"),config=thread_config,stream_mode="values"))
"""
{
  "prompts": ["a", "b"],
  "human_inputs": ["hello 1"],
  "__interrupt__": [
    Interrupt(
      value="b",
      resumable=True,
      ns=["child_graph:dad810e8-738e-9f90-41cd-30c0091eb79b"]
    )
  ]
}
"""

print(parent_graph.invoke(Command(resume="hello 2"),config=thread_config,stream_mode="values"))
#> {'prompts': ['a', 'b'], 'human_inputs': ['hello 1', 'hello 2']}
```
This commit is contained in:
Sydney Runkle
2025-04-24 08:21:28 -07:00
committed by GitHub
parent 6263c2f710 b81c21f311
commit 4eb124e83d
8 changed files with 272 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs/concepts/human_in_the_loop.md
-33
View File
@@ -409,39 +409,6 @@ The `Command` primitive provides several options to control and modify the graph
By leveraging `Command`, you can resume graph execution, handle user inputs, and dynamically adjust the graph's state.
## Using with `invoke` and `ainvoke`
When you use `stream` or `astream` to run the graph, you will receive an `Interrupt` event that let you know the `interrupt` was triggered.
`invoke` and `ainvoke` do not return the interrupt information. To access this information, you must use the [get_state](../reference/graphs.md#langgraph.graph.graph.CompiledGraph.get_state) method to retrieve the graph state after calling `invoke` or `ainvoke`.
```python
# Run the graph up to the interrupt
result = graph.invoke(inputs, thread_config)
# Get the graph state to get interrupt information.
state = graph.get_state(thread_config)
# Print the state values
print(state.values)
# Print the pending tasks
print(state.tasks)
# Resume the graph with the user's input.
graph.invoke(Command(resume={"age": "25"}), thread_config)
```
```pycon
{'foo': 'bar'} # State values
(
PregelTask(
id='5d8ffc92-8011-0c9b-8b59-9d3545b7e553',
name='node_foo',
path=('__pregel_pull', 'node_foo'),
error=None,
interrupts=(Interrupt(value='value_in_interrupt', resumable=True, ns=['node_foo:5d8ffc92-8011-0c9b-8b59-9d3545b7e553'], when='during'),), state=None,
result=None
),
) # Pending tasks. interrupts
```
## How does resuming from an interrupt work?
!!! warning
libs/langgraph/langgraph/pregel/__init__.py
+41 -12
View File
@@ -103,6 +103,7 @@ from langgraph.store.base import BaseStore
from langgraph.types import (
All,
Checkpointer,
Interrupt,
LoopProtocol,
StateSnapshot,
StateUpdate,
@@ -2191,7 +2192,7 @@ class Pregel(PregelProtocol):
stream_mode: The mode to stream output, defaults to self.stream_mode.
Options are:
- `"values"`: Emit all values in the state after each step.
- `"values"`: Emit all values in the state after each step, including interrupts.
When used with functional API, values are emitted once at the end of the workflow.
- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step.
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
@@ -2478,7 +2479,7 @@ class Pregel(PregelProtocol):
stream_mode: The mode to stream output, defaults to self.stream_mode.
Options are:
- `"values"`: Emit all values in the state after each step.
- `"values"`: Emit all values in the state after each step, including interrupts.
When used with functional API, values are emitted once at the end of the workflow.
- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step.
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
@@ -2788,10 +2789,11 @@ class Pregel(PregelProtocol):
If stream_mode is not "values", it returns a list of output chunks.
"""
output_keys = output_keys if output_keys is not None else self.output_channels
if stream_mode == "values":
latest: dict[str, Any] | Any = None
else:
chunks = []
latest: Union[dict[str, Any], Any] = None
chunks: list[Union[dict[str, Any], Any]] = []
interrupts: list[Interrupt] = []
for chunk in self.stream(
input,
config,
@@ -2804,10 +2806,23 @@ class Pregel(PregelProtocol):
**kwargs,
):
if stream_mode == "values":
latest = chunk
if (
isinstance(chunk, dict)
and (ints := chunk.get(INTERRUPT)) is not None
):
interrupts.extend(ints)
else:
latest = chunk
else:
chunks.append(chunk)
if stream_mode == "values":
if interrupts:
return (
{**latest, INTERRUPT: interrupts}
if isinstance(latest, dict)
else {INTERRUPT: interrupts}
)
return latest
else:
return chunks
@@ -2843,10 +2858,11 @@ class Pregel(PregelProtocol):
"""
output_keys = output_keys if output_keys is not None else self.output_channels
if stream_mode == "values":
latest: dict[str, Any] | Any = None
else:
chunks = []
latest: Union[dict[str, Any], Any] = None
chunks: list[Union[dict[str, Any], Any]] = []
interrupts: list[Interrupt] = []
async for chunk in self.astream(
input,
config,
@@ -2859,10 +2875,23 @@ class Pregel(PregelProtocol):
**kwargs,
):
if stream_mode == "values":
latest = chunk
if (
isinstance(chunk, dict)
and (ints := chunk.get(INTERRUPT)) is not None
):
interrupts.extend(ints)
else:
latest = chunk
else:
chunks.append(chunk)
if stream_mode == "values":
if interrupts:
return (
{**latest, INTERRUPT: interrupts}
if isinstance(latest, dict)
else {INTERRUPT: interrupts}
)
return latest
else:
return chunks
libs/langgraph/langgraph/pregel/loop.py
+15 -17
View File
@@ -914,23 +914,21 @@ class PregelLoop(LoopProtocol):
# we don't emit the interrupt as it'll be emitted by the parent
if task.path[0] == PUSH and task.path[-1] is True:
return
self._emit(
"updates",
lambda: iter(
[
{
INTERRUPT: tuple(
v
for w in writes
if w[0] == INTERRUPT
for v in (
w[1] if isinstance(w[1], Sequence) else (w[1],)
)
)
}
]
),
)
interrupts = [
{
INTERRUPT: tuple(
v
for w in writes
if w[0] == INTERRUPT
for v in (w[1] if isinstance(w[1], Sequence) else (w[1],))
)
}
]
stream_modes = self.stream.modes if self.stream else []
if "updates" in stream_modes:
self._emit("updates", lambda: iter(interrupts))
elif "values" in stream_modes:
self._emit("values", lambda: iter(interrupts))
elif writes[0][0] != ERROR:
self._emit(
"updates",
libs/langgraph/langgraph/types.py
+9 -1
View File
@@ -19,6 +19,7 @@ from typing import (
from langchain_core.runnables import Runnable, RunnableConfig
from typing_extensions import Self
from xxhash import xxh3_128_hexdigest
from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
from langgraph.utils.fields import get_update_as_tuples
@@ -48,7 +49,7 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver]
StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
"""How the stream method should emit outputs.
- `"values"`: Emit all values in the state after each step.
- `"values"`: Emit all values in the state after each step, including interrupts.
When used with functional API, values are emitted once at the end of the workflow.
- `"updates"`: Emit only the node or task names and updates returned by the nodes or tasks after each step.
If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
@@ -141,6 +142,13 @@ class Interrupt:
ns: Optional[Sequence[str]] = None
when: Literal["during"] = dataclasses.field(default="during", repr=False)
@property
def interrupt_id(self) -> str:
"""Generate a unique ID for the interrupt based on its namespace."""
if self.ns is None:
return "placeholder-id"
return xxh3_128_hexdigest("".join(self.ns).encode())
class StateUpdate(NamedTuple):
values: Optional[dict[str, Any]]
libs/langgraph/tests/test_large_cases.py
+35 -9
View File
@@ -5666,6 +5666,9 @@ def test_dynamic_interrupt(
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -5712,6 +5715,9 @@ def test_dynamic_interrupt(
assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
"my_key": "value ⛰️",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
if "shallow" not in checkpointer_name:
@@ -5840,6 +5846,9 @@ def test_copy_checkpoint(
) == {
"my_key": "value one",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -5890,6 +5899,9 @@ def test_copy_checkpoint(
assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
"my_key": "value ⛰️ one",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
if "shallow" not in checkpointer_name:
assert [c.metadata for c in tool_two.checkpointer.list(thread1)] == [
@@ -6040,6 +6052,13 @@ def test_dynamic_interrupt_subgraph(
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:"), AnyStr("do:")],
)
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -6086,6 +6105,13 @@ def test_dynamic_interrupt_subgraph(
assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == {
"my_key": "value ⛰️",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:"), AnyStr("do:")],
)
],
}
if "shallow" not in checkpointer_name:
@@ -7318,15 +7344,15 @@ def test_send_dedupe_on_resume(
graph = builder.compile(checkpointer=checkpointer)
thread1 = {"configurable": {"thread_id": "1"}}
assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == [
"0",
"1",
"3.1",
"2|Command(goto=Send(node='2', arg=3))",
"2|Command(goto=Send(node='flaky', arg=4))",
"3",
"2|3",
]
assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == {
"__interrupt__": [
Interrupt(
value="Bahh",
resumable=False,
ns=None,
),
],
}
assert builder.nodes["2"].runnable.func.ticks == 3
assert builder.nodes["flaky"].runnable.func.ticks == 1
# check state
libs/langgraph/tests/test_pregel.py
+63 -6
View File
@@ -3703,7 +3703,16 @@ def test_subgraph_checkpoint_true_interrupt(
assert graph.invoke(
{"foo": "foo"}, config, checkpoint_during=checkpoint_during
) == {"foo": "hi! foo"}
) == {
"foo": "hi! foo",
"__interrupt__": [
Interrupt(
value="Provide baz value",
resumable=True,
ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")],
)
],
}
assert graph.get_state(config, subgraphs=True).tasks[0].state.values == {
"bar": "hi! foo"
}
@@ -5413,7 +5422,15 @@ def test_interrupt_functional(
config = {"configurable": {"thread_id": "1"}}
# First run, interrupted at bar
graph.invoke({"a": ""}, config)
assert graph.invoke({"a": ""}, config) == {
"__interrupt__": [
Interrupt(
value="Provide value for bar:",
resumable=True,
ns=[AnyStr("graph:")],
)
]
}
# Resume with an answer
res = graph.invoke(Command(resume="bar"), config)
assert res == {"a": "foobar", "b": "bar"}
@@ -5444,7 +5461,15 @@ def test_interrupt_task_functional(
config = {"configurable": {"thread_id": "1"}}
# First run, interrupted at bar
assert not graph.invoke({"a": ""}, config)
assert graph.invoke({"a": ""}, config) == {
"__interrupt__": [
Interrupt(
value="Provide value for bar:",
resumable=True,
ns=[AnyStr("graph:"), AnyStr("bar:")],
),
]
}
# Resume with an answer
res = graph.invoke(Command(resume="bar"), config)
assert res == {"a": "foobar"}
@@ -5460,9 +5485,17 @@ def test_interrupt_task_functional(
return baz_result
# First run, interrupted at bar
assert not graph.invoke({"a": ""}, config)
assert graph.invoke({"a": ""}, config) == {
"__interrupt__": [
Interrupt(
value="Provide value for bar:",
resumable=True,
ns=[AnyStr("graph:"), AnyStr("bar:")],
),
]
}
# Provide resumes
assert not graph.invoke(Command(resume="bar"), config)
graph.invoke(Command(resume="bar"), config)
assert graph.invoke(Command(resume="baz"), config) == {"a": "foobarbaz"}
@@ -7347,10 +7380,27 @@ def test_interrupt_subgraph_reenter_checkpointer_true(
)
config = {"configurable": {"thread_id": "1"}}
assert parent.invoke({"foo": "", "counter": 0}, config) == {"foo": "", "counter": 0}
assert parent.invoke({"foo": "", "counter": 0}, config) == {
"foo": "",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
assert parent.invoke(Command(resume="bar"), config) == {
"foo": "subgraph_2",
"counter": 1,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
assert parent.invoke(Command(resume="qux"), config) == {
"foo": "subgraph_2|parent",
@@ -7375,6 +7425,13 @@ def test_interrupt_subgraph_reenter_checkpointer_true(
assert parent.invoke({"foo": "meow", "counter": 0}, config) == {
"foo": "meow",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
# confirm that we preserve the state values from the previous invocation
assert bar_values == [None, "barbaz", "quxbaz"]
libs/langgraph/tests/test_pregel_async.py
+105 -15
View File
@@ -531,6 +531,9 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -713,6 +716,13 @@ async def test_dynamic_interrupt_subgraph(checkpointer_name: str) -> None:
) == {
"my_key": "value",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:"), AnyStr("do:")],
)
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -903,6 +913,9 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
) == {
"my_key": "value one",
"market": "DE",
"__interrupt__": [
Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")])
],
}
assert tool_two_node_count == 1, "interrupts aren't retried"
assert len(tracer.runs) == 1
@@ -964,6 +977,13 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
) == {
"my_key": "value ⛰️ one",
"market": "DE",
"__interrupt__": [
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
)
],
}
if "shallow" not in checkpointer_name:
@@ -1108,13 +1128,29 @@ async def test_node_not_cancelled_on_other_node_interrupted(
# writes from "awhile" are applied to last chunk
assert await graph.ainvoke({"hello": "world"}, thread) == {
"hello": "world again"
"hello": "world again",
"__interrupt__": [
Interrupt(
value="I am bad",
resumable=True,
ns=[AnyStr("bad:")],
)
],
}
assert not inner_task_cancelled
assert awhiles == 1
assert await graph.ainvoke(None, thread, debug=True) == {"hello": "world again"}
assert await graph.ainvoke(None, thread, debug=True) == {
"hello": "world again",
"__interrupt__": [
Interrupt(
value="I am bad",
resumable=True,
ns=[AnyStr("bad:")],
)
],
}
assert not inner_task_cancelled
assert awhiles == 1
@@ -2795,15 +2831,15 @@ async def test_send_dedupe_on_resume(
thread1 = {"configurable": {"thread_id": "1"}}
assert await graph.ainvoke(
["0"], thread1, checkpoint_during=checkpoint_during
) == [
"0",
"1",
"3.1",
"2|Command(goto=Send(node='2', arg=3))",
"2|Command(goto=Send(node='flaky', arg=4))",
"3",
"2|3",
]
) == {
"__interrupt__": [
Interrupt(
value="Bahh",
resumable=False,
ns=None,
),
],
}
assert builder.nodes["2"].runnable.func.ticks == 3
assert builder.nodes["flaky"].runnable.func.ticks == 1
# resume execution
@@ -5555,7 +5591,16 @@ async def test_subgraph_checkpoint_true_interrupt(
assert await graph.ainvoke(
{"foo": "foo"}, config, checkpoint_during=checkpoint_during
) == {"foo": "hi! foo"}
) == {
"foo": "hi! foo",
"__interrupt__": [
Interrupt(
value="Provide baz value",
resumable=True,
ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")],
)
],
}
assert (await graph.aget_state(config, subgraphs=True)).tasks[
0
].state.values == {"bar": "hi! foo"}
@@ -6828,7 +6873,15 @@ async def test_interrupt_task_functional(checkpointer_name: str) -> None:
config = {"configurable": {"thread_id": "1"}}
# First run, interrupted at bar
await graph.ainvoke({"a": ""}, config)
assert await graph.ainvoke({"a": ""}, config) == {
"__interrupt__": [
Interrupt(
value="Provide value for bar:",
resumable=True,
ns=[AnyStr("graph:"), AnyStr("bar:")],
),
]
}
# Resume with an answer
res = await graph.ainvoke(Command(resume="bar"), config)
assert res == {"a": "foobar"}
@@ -8104,10 +8157,24 @@ async def test_interrupt_subgraph_reenter_checkpointer_true(
assert await parent.ainvoke({"foo": "", "counter": 0}, config) == {
"foo": "",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
assert await parent.ainvoke(Command(resume="bar"), config) == {
"foo": "subgraph_2",
"counter": 1,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2")],
)
],
}
assert await parent.ainvoke(Command(resume="qux"), config) == {
"foo": "subgraph_2|parent",
@@ -8132,6 +8199,13 @@ async def test_interrupt_subgraph_reenter_checkpointer_true(
assert await parent.ainvoke({"foo": "meow", "counter": 0}, config) == {
"foo": "meow",
"counter": 0,
"__interrupt__": [
Interrupt(
value="Provide value",
resumable=True,
ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")],
)
],
}
# confirm that we preserve the state values from the previous invocation
assert bar_values == [None, "barbaz", "quxbaz"]
@@ -8160,7 +8234,15 @@ async def test_handles_multiple_interrupts_from_tasks() -> None:
config = {"configurable": {"thread_id": "1"}}
result = await program.ainvoke("this is ignored", config=config)
assert result is None
assert result == {
"__interrupt__": [
Interrupt(
value="Hey do you want to add James?",
resumable=True,
ns=[AnyStr("program:"), AnyStr("add_participant:")],
),
]
}
state = await program.aget_state(config=config)
assert len(state.tasks[0].interrupts) == 1
@@ -8172,7 +8254,15 @@ async def test_handles_multiple_interrupts_from_tasks() -> None:
assert task_interrupt.value == "Hey do you want to add James?"
result = await program.ainvoke(Command(resume=True), config=config)
assert result is None
assert result == {
"__interrupt__": [
Interrupt(
value="Hey do you want to add Will?",
resumable=True,
ns=[AnyStr("program:"), AnyStr("add_participant:")],
),
]
}
state = await program.aget_state(config=config)
assert len(state.tasks[0].interrupts) == 1
libs/prebuilt/tests/test_react_agent.py
+4 -2
View File
@@ -672,7 +672,8 @@ def test_react_agent_parallel_tool_calls(
for event in agent.stream(
{"messages": [("user", query)]}, config, stream_mode="values"
):
message_types.append([message.type for message in event["messages"]])
if messages := event.get("messages"):
message_types.append([m.type for m in messages])
if version == "v1":
assert message_types == [
@@ -691,7 +692,8 @@ def test_react_agent_parallel_tool_calls(
for event in agent.stream(
Command(resume={"data": "Hello"}), config, stream_mode="values"
):
message_types.append([message.type for message in event["messages"]])
if messages := event.get("messages"):
message_types.append([m.type for m in messages])
assert message_types == [
["human", "ai"],