diff --git a/examples/foundational/07m-interruptible-aws-strands.py b/examples/foundational/07m-interruptible-aws-strands.py index 584153ba2..eca2947fc 100644 --- a/examples/foundational/07m-interruptible-aws-strands.py +++ b/examples/foundational/07m-interruptible-aws-strands.py @@ -12,11 +12,11 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer 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 from pipecat.processors.aggregators.llm_response import ( LLMAssistantContextAggregator, LLMUserContextAggregator, ) +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frameworks.strands_agents import StrandsAgentsProcessor from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport @@ -58,16 +58,18 @@ transport_params = { ), } + 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": @@ -76,14 +78,14 @@ def build_agent(model_id: str, max_tokens: int): 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." + system_prompt="You are a helpful assistant that can check the weather in a given location.", ) return agent @@ -102,10 +104,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # Create Strands agent processor try: - agent = build_agent( - model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0", - max_tokens=8000 - ) + 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: @@ -120,15 +119,17 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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 - ]) + 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, diff --git a/src/pipecat/processors/frameworks/strands_agents.py b/src/pipecat/processors/frameworks/strands_agents.py index 5c2011b43..05aef0f17 100644 --- a/src/pipecat/processors/frameworks/strands_agents.py +++ b/src/pipecat/processors/frameworks/strands_agents.py @@ -15,6 +15,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMTextFrame, ) +from pipecat.metrics.metrics import LLMTokenUsage from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -84,29 +85,79 @@ class StrandsAgentsProcessor(FrameProcessor): 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: + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + await self.start_ttfb_metrics() + ttfb_tracking = True + if self.graph: # Graph does not stream; await full result then emit assistant text graph_result = await self.graph.invoke_async(text) + if ttfb_tracking: + await self.stop_ttfb_metrics() + ttfb_tracking = False try: node_result = graph_result.results[self.graph_exit_node] + logger.debug(f"Node result: {node_result}") for agent_result in node_result.get_agent_results(): + # Push to TTS service 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"]))) + # Update usage metrics + await self._report_usage_metrics( + agent_result.metrics.accumulated_usage.get('inputTokens', 0), + agent_result.metrics.accumulated_usage.get('outputTokens', 0), + agent_result.metrics.accumulated_usage.get('totalTokens', 0) + ) 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): + # Push to TTS service if isinstance(event, dict) and "data" in event: await self.push_frame(LLMTextFrame(str(event["data"]))) + if ttfb_tracking: + await self.stop_ttfb_metrics() + ttfb_tracking = False + + # Update usage metrics + if isinstance(event, dict) and "event" in event and "metadata" in event['event']: + if 'usage' in event['event']['metadata']: + usage = event['event']['metadata']['usage'] + await self._report_usage_metrics(usage.get('inputTokens', 0), usage.get('outputTokens', 0), usage.get('totalTokens', 0)) 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 + if ttfb_tracking: + await self.stop_ttfb_metrics() + ttfb_tracking = False + await self.stop_processing_metrics() + await self.push_frame(LLMFullResponseEndFrame()) + + def can_generate_metrics(self) -> bool: + """Check if this service can generate performance metrics. + + Returns: + True as this service supports metrics generation. + """ + return True + + async def _report_usage_metrics( + self, + prompt_tokens: int, + completion_tokens: int, + total_tokens: int + ): + tokens = LLMTokenUsage( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=total_tokens + ) + await self.start_llm_usage_metrics(tokens) \ No newline at end of file