From cbdbdee4c05091e98f27ffed9b0e2876c364d322 Mon Sep 17 00:00:00 2001 From: Adithya Suresh Date: Thu, 4 Sep 2025 16:58:58 +1000 Subject: [PATCH] Initial StrandsAgentsProcessor implementation --- .../07m-interruptible-aws-strands.py | 169 ++++++++++++++++++ .../processors/frameworks/strands_agents.py | 112 ++++++++++++ 2 files changed, 281 insertions(+) create mode 100644 examples/foundational/07m-interruptible-aws-strands.py create mode 100644 src/pipecat/processors/frameworks/strands_agents.py diff --git a/examples/foundational/07m-interruptible-aws-strands.py b/examples/foundational/07m-interruptible-aws-strands.py new file mode 100644 index 000000000..82d7192c2 --- /dev/null +++ b/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() diff --git a/src/pipecat/processors/frameworks/strands_agents.py b/src/pipecat/processors/frameworks/strands_agents.py new file mode 100644 index 000000000..5c2011b43 --- /dev/null +++ b/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()) \ No newline at end of file