Added usage metrics

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(
+        agent = build_agent(model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0", max_tokens=8000)
-            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([
+    pipeline = Pipeline(
-        transport.input(),      # Transport user input
+        [
-        stt,                   # Speech-to-text
+            transport.input(),  # Transport user input
-        tma_in,               # User context aggregator
+            stt,  # Speech-to-text
-        llm,                  # Strands Agents processor
+            tma_in,  # User context aggregator
-        tts,                  # Text-to-speech
+            llm,  # Strands Agents processor
-        transport.output(),   # Transport bot output
+            tts,  # Text-to-speech
-        tma_out              # Assistant context aggregator
+            transport.output(),  # Transport bot output
-    ])
+            tma_out,  # Assistant context aggregator
+        ]
+    )
 
     task = PipelineTask(
         pipeline,

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())
+            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)