interrupts - graph - agent based

[pgrayy]

Description

Allow users to raise interrupts from an agent node in Graph. This is a follow up to #1478 and an iteration on #1350.

Usage

from strands import Agent, tool
from strands.interrupt import Interrupt
from strands.multiagent import GraphBuilder, Status
from strands.types.tools import ToolContext

@tool(context=True)
def weather_tool(tool_context: ToolContext) -> str:
    response = tool_context.interrupt("weather_interrupt", reason="need weather")
    return response

weather_agent = Agent(name="weather")

builder = GraphBuilder()
builder.add_node(weather_agent, "weather_agent")
graph = builder.build()

multiagent_result = graph("What is the weather?")
while multiagent_result.status == Status.INTERRUPTED:
    responses = []
    for interrupt in multiagent_result.interrupts:
        if interrupt.name == "weather_interrupt":
            response = input(f"{interrupt.reason}: ")
            responses.append(
                {
                    "interruptResponse": {
                         "interruptId": interrupt.id,
                         "response": response,
                    },
                },
             )

    multiagent_result = graph(responses)

print(multiagent_result.results)

Follow Up

Support raising an interrupt from a multi-agent node. There are some special considerations with session management to get this working properly.

Related Issues

#204

Documentation PR

Will update https://strandsagents.com/latest/documentation/docs/user-guide/concepts/interrupts/#multi-agents in follow up. Note, swarm interrupt docs have been added but not yet released.

Type of Change

New feature

Testing

How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

• I ran hatch run prepare: Wrote new unit tests
• I ran hatch test tests_integ/interrupts/multiagent: Wrote new integ tests

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

❌ Patch coverage is 94.11765% with 1 line in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/strands/multiagent/graph.py	94.11%	0 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!

[pgrayy]

Agents are not allowed to use session managers in a graph execution. Consequently, we need to store some agent state in the graph interrupt state to help persist the interrupt workflow between shutdowns. Note, this is the same behavior we have in Swarm (src).

[pgrayy]

We check if executor is an Agent instance right now because MultiAgentBase executors may not have the same context. We will figure out how to handle that case in a separate (and final) PR for graph interrupt support.

[pgrayy]

Here is where we restore the agent node state upon resuming. We extract it from the graph interrupt state.

[pgrayy]

Here I am making the switch to interrupt on an agent node rather than through a hook. This is what we do for the swarm integ test. It is more comprehensive as it not only tests persistence of the graph interrupt state, but also the interrupt state of the agent node. Hook interrupts only require graph interrupt state persistence.

[dbschmigelski]

Can this be AgentBase now?

[pgrayy]

Not yet as class Agent does not yet derive from AgentBase (src).

[dbschmigelski]

Can you explain how interrupts work in graph at a higher level?

I understand how this works for an interrupt that is raised in one Agent node by that Agent node.

But what happens if two nodes are executing and one of them raises an interrupt?

[pgrayy]

If two nodes are executing in parallel and one interrupts, the other node will be allowed finished. Once done, the call stack returns back to _execute_graph. Here, before preparing the next batch of nodes to execute, we check to see if an interrupt has been activated (any time any node interrupts, we immediately set graph state to INTERRUPTED). If we have an interrupt, we store the one completed node in the interrupt state context. Context about the interrupted node has already been stored in state as part of the _execute_node call.

Upon resuming, we unpack the interrupted node and the one already completed from interrupt state (src). Within _execute_nodes_parallel, we filter the batch down to just the interrupted node (src). This means the completed node does not get executed again. However, we have the reference so that we can identify its dependent nodes to execute next after we finish resuming the interrupted node.

For the interrupted node, we pass the user interrupt responses into node.executor.stream_async (src). The actual task that the agent node is meant to execute is already stored in the agent node message history.

From here, things proceed as normal.

[pgrayy]

For additional context, we explicitly test parallel node interrupts in the integ tests presented further down.