interrupts - swarm

[pgrayy]

Description

Adding interrupt (HIL) support to Swarm. Will add support to Graph in a separate PR.

Note, session managed interrupts are supported out of the box with this PR. For a usage example, see tests_integ/interrupts/multiagent/test_session.py under "Files changed".

Usage

Note, these are the same interfaces available for single agent interrupts (docs).

BeforeNodeCallEvent Hook

import json

from strands import Agent
from strands.experimental.hooks.multiagent import BeforeNodeCallEvent
from strands.hooks import HookProvider
from strands.interrupt import Interrupt
from strands.multiagent import Swarm
from strands.multiagent.base import Status


class ApprovalHook(HookProvider):
    def register_hooks(self, registry):
        registry.add_callback(BeforeNodeCallEvent, self.approve)

    def approve(self, event):
        if event.node_id == "info":
            return

        response = event.interrupt("my_interrupt", reason=f"{event.node_id} needs approval")
        if response != "APPROVE":
            event.cancel_node = "node rejected"


info_agent = Agent(name="info")
weather_agent = Agent(name="weather")
swarm = Swarm([info_agent, weather_agent], hooks=[ApprovalHook()])

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

    multiagent_result = swarm(responses)


weather_result = multiagent_result.results["weather"]
weather_message = json.dumps(weather_result.result.message).lower()
print(weather_message)

Within Agent

import json

from strands import Agent, tool
from strands.interrupt import Interrupt
from strands.multiagent import Swarm
from strands.multiagent.base import Status


@tool(context=True)
def user_name_tool(tool_context: ToolContext) -> str:
    response = tool_context.interrupt("my_interrupt", reason="need user name")
    return response


user_agent = Agent(name="user", tools=[user_name_tool])
swarm = Swarm([user_agent])

multiagent_result = swarm("Who is the current user?")
while multiagent_result.status == Status.INTERRUPTED:
    response = []
    for interrupt in multiagent_result.interrupts:
        if interrupt.name == "my_interrupt":
            response = input(f"{interrupt.reason}: ")
            responses.append(
                {
                    "interruptResponse": {
                         "interruptId": interrupt.id,
                         "response": response,
                    },
                },
             )

    multiagent_result = swarm(responses)


user_result = multiagent_result.results["user"]
user_message = json.dumps(user_result.result.message).lower()
print(user_message)

Related Issues

#204

Documentation PR

After closing out PR, will add usage instructions to https://strandsagents.com/latest/documentation/docs/user-guide/concepts/interrupts/. Will also document the new streamed MultiAgentNodeInterruptEvent in https://strandsagents.com/latest/documentation/docs/user-guide/concepts/streaming/.

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/*.py: 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 80.68182% with 17 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/strands/multiagent/swarm.py	83.05%	5 Missing and 5 partials ⚠️
src/strands/multiagent/base.py	40.00%	6 Missing ⚠️
src/strands/multiagent/graph.py	50.00%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

[afarntrog]

From the PR description:

while multiagent_result.status != Status.INTERRUPTED:

Shouldn't this be == instead of !=? As we would want to handle interrupts only when the status is interrupted? What am i missing?

[pgrayy]

From the PR description:

while multiagent_result.status != Status.INTERRUPTED:

Shouldn't this be == instead of !=? As we would want to handle interrupts only when the status is interrupted? What am i missing?

That was a typo. I ran the examples from https://strandsagents.com/latest/documentation/docs/user-guide/concepts/interrupts/ (but for multi agents). I tried to simplify the loop though and copied the if condition without updating to ==. Just updated the description.