Initial StrandsAgentsProcessor implementation

examples/foundational/07m-interruptible-aws-strands.py

@@ -0,0 +1,169 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    LLMAssistantContextAggregator,
+    LLMUserContextAggregator,
+)
+from pipecat.processors.frameworks.strands_agents import StrandsAgentsProcessor
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.aws.stt import AWSTranscribeSTTService
+from pipecat.services.aws.tts import AWSPollyTTSService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+# Strands agent setup
+try:
+    from strands import Agent, tool
+    from strands.models import BedrockModel
+except ImportError:
+    logger.warning("Strands not installed. Please install with: pip install strands-agents")
+    Agent = None
+    BedrockModel = None
+
+load_dotenv(override=True)
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+def build_agent(model_id: str, max_tokens: int):
+    """Create and configure a Strands agent for NAB customer service coaching.
+    
+    Args:
+        model_id: The AWS Bedrock model ID to use
+        max_tokens: Maximum tokens for the model
+        
+    Returns:
+        Configured Strands Agent
+    """
+    @tool
+    def check_weather(location: str) -> str:
+        if location.lower() == "san francisco":
+            return "The weather in San Francisco is sunny and 30 degrees."
+        elif location.lower() == "sydney":
+            return "The weather in Sydney is cloudy and 20 degrees."
+        else:
+            return "I'm not sure about the weather in that location."
+    
+    agent = Agent(
+        model=BedrockModel(
+            model_id=model_id,
+            max_tokens=max_tokens,
+        ),
+        tools=[check_weather],
+        system_prompt="You are a helpful assistant that can check the weather in a given location."
+    )
+
+    return agent
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = AWSTranscribeSTTService()
+
+    tts = AWSPollyTTSService(
+        region="us-west-2",  # only specific regions support generative TTS
+        voice_id="Joanna",
+        params=AWSPollyTTSService.InputParams(engine="generative", rate="1.1"),
+    )
+
+    # Create Strands agent processor
+    try:
+        agent = build_agent(
+            model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0", 
+            max_tokens=8000
+        )
+        llm = StrandsAgentsProcessor(agent=agent)
+        logger.info("Successfully created Strands agent for NAB customer service coaching")
+    except Exception as e:
+        logger.error(f"Failed to create Strands agent: {e}")
+        raise ValueError(
+            "Unable to create Strands processor. Please ensure you have properly "
+            "installed strands-agents and configured your AWS credentials."
+        )
+
+    # Setup context aggregators for message handling
+    context = OpenAILLMContext()
+    tma_in = LLMUserContextAggregator(context=context)
+    tma_out = LLMAssistantContextAggregator(context=context)
+
+    pipeline = Pipeline([
+        transport.input(),      # Transport user input
+        stt,                   # Speech-to-text
+        tma_in,               # User context aggregator
+        llm,                  # Strands Agents processor
+        tts,                  # Text-to-speech
+        transport.output(),   # Transport bot output
+        tma_out              # Assistant context aggregator
+    ])
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "user", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.run(task)
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()

src/pipecat/processors/frameworks/strands_agents.py

@@ -0,0 +1,112 @@
+"""
+Strands Agent integration for Pipecat.
+
+This module provides integration with Strands Agents for handling conversational AI
+interactions. It supports both single agent and multi-agent graphs.
+"""
+
+from typing import Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    Frame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMTextFrame,
+)
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+try:
+    from strands import Agent
+    from strands.multiagent.graph import Graph
+except ModuleNotFoundError as e:
+    logger.exception("In order to use Strands Agents, you need to `pip install strands-agents`.")
+    raise Exception(f"Missing module: {e}")
+
+
+class StrandsAgentsProcessor(FrameProcessor):
+    """Processor that integrates Strands Agents with Pipecat's frame pipeline.
+
+    This processor takes LLM message frames, extracts the latest user message,
+    and processes it through either a single Strands Agent or a multi-agent Graph.
+    The response is streamed back as text frames with appropriate response markers.
+
+    Supports both single agent streaming and graph-based multi-agent workflows.
+    """
+
+    def __init__(
+        self,
+        agent: Optional[Agent] = None,
+        graph: Optional[Graph] = None,
+        graph_exit_node: Optional[str] = None,
+    ):
+        """Initialize the Strands Agents processor.
+
+        Args:
+            agent: The Strands Agent to use for single-agent processing.
+            graph: The Strands multi-agent Graph to use for graph-based processing.
+            graph_exit_node: The exit node name when using graph-based processing.
+
+        Raises:
+            AssertionError: If neither agent nor graph is provided, or if graph is
+                          provided without a graph_exit_node.
+        """
+        super().__init__()
+        self.agent = agent
+        self.graph = graph
+        self.graph_exit_node = graph_exit_node
+
+        assert self.agent or self.graph, "Either agent or graph must be provided"
+
+        if self.graph:
+            assert self.graph_exit_node, "graph_exit_node must be provided if graph is provided"
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process incoming frames and handle LLM message frames.
+
+        Args:
+            frame: The incoming frame to process.
+            direction: The direction of frame flow in the pipeline.
+        """
+        await super().process_frame(frame, direction)
+        if isinstance(frame, OpenAILLMContextFrame):
+            text = frame.context.messages[-1]["content"]
+            await self._ainvoke(str(text).strip())
+        else:
+            await self.push_frame(frame, direction)
+
+    async def _ainvoke(self, text: str):
+        """Invoke the Strands agent with the provided text and stream results as Pipecat frames.
+
+        Args:
+            text: The user input text to process through the agent or graph.
+        """
+        logger.debug(f"Invoking Strands agent with: {text}")
+        await self.push_frame(LLMFullResponseStartFrame())
+        try:
+            if self.graph:
+                # Graph does not stream; await full result then emit assistant text
+                graph_result = await self.graph.invoke_async(text)
+                try:
+                    node_result = graph_result.results[self.graph_exit_node]
+                    for agent_result in node_result.get_agent_results():
+                        message = getattr(agent_result, "message", None)
+                        if isinstance(message, dict) and "content" in message:
+                            for block in message["content"]:
+                                if isinstance(block, dict) and "text" in block:
+                                    await self.push_frame(LLMTextFrame(str(block["text"])))
+                except Exception as parse_err:
+                    logger.warning(f"Failed to extract messages from GraphResult: {parse_err}")
+            else:
+                # Agent supports streaming events via async iterator
+                async for event in self.agent.stream_async(text):
+                    if isinstance(event, dict) and "data" in event:
+                        await self.push_frame(LLMTextFrame(str(event["data"])))
+        except GeneratorExit:
+            logger.warning(f"{self} generator was closed prematurely")
+        except Exception as e:
+            logger.exception(f"{self} an unknown error occurred: {e}")
+        finally:
+            await self.push_frame(LLMFullResponseEndFrame())