Interruption Handling

1
# Basic interruption setup
2
(
3
    bridge.on(UserStoppedSpeaking)
4
    .interrupt_on(UserStartedSpeaking)
5
    .stream(node.generate)
6
    .broadcast()
7
)
8
9
10
# With custom interrupt handler
11
async def handle_interrupt(message):
12
    logger.info("User interrupted, stopping generation")
13
    # Perform any needed cleanup
14
    await cleanup_resources()
15
16
17
(
18
    bridge.on(UserStoppedSpeaking)
19
    .interrupt_on(UserStartedSpeaking, handler=handle_interrupt)
20
    .stream(node.generate)
21
    .broadcast()
22
)

1
class InterruptAwareNode(ReasoningNode):
2
    def on_interrupt_generate(self, message):
3
        """Called when generation is interrupted."""
4
        logger.info("Generation interrupted by user")
5
        # Cleanup streaming resources, cancel API calls, etc.
6
7
    async def process_context(self, context):
8
        try:
9
            # Stream response chunks
10
            async for chunk in self.llm_client.generate_stream(
11
                messages
12
            ):
13
                yield AgentResponse(content=chunk.text)
14
15
        except asyncio.CancelledError:
16
            # Handle cancellation gracefully
17
            logger.info(
18
                "Generation cancelled due to user interruption"
19
            )
20
            # Perform any final cleanup
21
            raise  # Re-raise to complete cancellation
22
23
24
# Connect interrupt handler in route
25
(
26
    bridge.on(UserStoppedSpeaking)
27
    .interrupt_on(
28
        UserStartedSpeaking, handler=node.on_interrupt_generate
29
    )
30
    .stream(node.generate)
31
    .broadcast()
32
)

1
# Only allow interruption after agent has spoken for a minimum time
2
def can_interrupt(message):
3
    return time.time() - generation_start_time > MIN_SPEAK_TIME
4
5
6
(
7
    bridge.on(UserStoppedSpeaking)
8
    .interrupt_on(UserStartedSpeaking, condition=can_interrupt)
9
    .stream(node.generate)
10
    .broadcast()
11
)

1
# Interrupt on multiple event types
2
(
3
    bridge.on(UserStoppedSpeaking)
4
    .interrupt_on([UserStartedSpeaking, EmergencyStop, AgentHandoff])
5
    .stream(node.generate)
6
    .broadcast()
7
)

1
class StatefulInterruptNode(ReasoningNode):
2
    def __init__(self):
3
        super().__init__()
4
        self.generation_state = None
5
6
    def on_interrupt_generate(self, message):
7
        # Save state for potential recovery
8
        self.generation_state = {
9
            "interrupted_at": time.time(),
10
            "partial_response": self.current_response,
11
            "context_snapshot": self.context.copy(),
12
        }
13
        logger.info("Saved generation state for recovery")
14
15
    async def process_context(self, context):
16
        try:
17
            # Check if resuming from interruption
18
            if self.generation_state:
19
                logger.info("Resuming from previous interruption")
20
                # Potentially use saved state
21
                self.generation_state = None
22
23
            # Normal generation flow
24
            async for chunk in self.generate_response(context):
25
                yield AgentResponse(content=chunk)
26
27
        except asyncio.CancelledError:
28
            logger.info("Generation cancelled")
29
            raise