pass int instead of float to

Introduce support for extracting and processing grounding metadata from GoogleLLMService.

CHANGELOG.md

@@ -12,6 +12,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - It is now possible to specify the period of the `PipelineTask` heartbeat
   frames with `heartbeats_period_secs`.
 
+- Added `DailyMeetingTokenProperties` and `DailyMeetingTokenParams` Pydantic models 
+  for meeting token creation in `get_token` method of `DailyRESTHelper`.
+
+- Added `enable_recording` and `geo` parameters to `DailyRoomProperties`.
+
+- Added `RecordingsBucketConfig` to `DailyRoomProperties` to upload recordings to a custom AWS bucket.
+
 ### Changed
 
 - Modified `TranscriptProcessor` to use TTS text frames for more accurate assistant
@@ -31,6 +38,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Fixed an issue where `OpenAIRealtimeBetaLLMService` function calling resulted
   in an error.
 
+- Fixed an issue in `AudioBufferProcessor` where the last audio buffer was not
+  being processed, in cases where the `_user_audio_buffer` was smaller than the
+  buffer size.
+
 ### Performance
 
 - Replaced audio resampling library `resampy` with `soxr`. Resampling a 2:21s

examples/canonical-metrics/runner.py

@@ -53,4 +53,3 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
     token = await daily_rest_helper.get_token(url, expiry_time)
 
     return (url, token)
-    return (url, token)

examples/foundational/31-gemini-grounding-metadata.py

@@ -0,0 +1,130 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import Frame
+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.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.services.google import GoogleLLMService, LLMSearchResponseFrame
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+sys.path.append(str(Path(__file__).parent.parent))
+from runner import configure
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+# Function handlers for the LLM
+search_tool = {"google_search_retrieval": {}}
+tools = [search_tool]
+
+system_instruction = """
+You are an expert at providing the most recent news from any place. Your responses will be converted to audio, so avoid using special characters or overly complex formatting. 
+
+Always use the google search API to retrieve the latest news. You must also use it to check which day is today.
+
+You can:
+- Use the Google search API to check the current date.
+- Provide the most recent and relevant news from any place by using the google search API.
+- Answer any questions the user may have, ensuring your responses are accurate and concise.
+
+Start each interaction by asking the user about which place they would like to know the information.
+"""
+
+
+class LLMSearchLoggerProcessor(FrameProcessor):
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMSearchResponseFrame):
+            print(f"LLMSearchLoggerProcessor: {frame}")
+
+        await self.push_frame(frame)
+
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Latest news!",
+            DailyParams(
+                audio_out_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True,
+            ),
+        )
+
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        # Initialize the Gemini Multimodal Live model
+        llm = GoogleLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            system_instruction=system_instruction,
+            tools=tools,
+        )
+
+        context = OpenAILLMContext(
+            [
+                {
+                    "role": "user",
+                    "content": "Start by greeting the user warmly, introducing yourself, and mentioning the current day. Be friendly and engaging to set a positive tone for the interaction.",
+                }
+            ],
+        )
+        context_aggregator = llm.create_context_aggregator(context)
+
+        llm_search_logger = LLMSearchLoggerProcessor()
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                stt,
+                context_aggregator.user(),
+                llm,
+                llm_search_logger,
+                tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        runner = PipelineRunner()
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -6,6 +6,7 @@
 
 from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio
 from pipecat.frames.frames import (
+    EndFrame,
     Frame,
     InputAudioRawFrame,
     OutputAudioRawFrame,
@@ -86,6 +87,9 @@ class AudioBufferProcessor(FrameProcessor):
         if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size:
             await self._call_on_audio_data_handler()
 
+        if isinstance(frame, EndFrame):
+            await self._call_on_audio_data_handler()
+
         await self.push_frame(frame, direction)
 
     async def _call_on_audio_data_handler(self):

src/pipecat/services/cartesia.py

@@ -88,7 +88,7 @@ class CartesiaTTSService(WordTTSService, WebsocketService):
         voice_id: str,
         cartesia_version: str = "2024-06-10",
         url: str = "wss://api.cartesia.ai/tts/websocket",
-        model: str = "sonic-english",
+        model: str = "sonic",
         sample_rate: int = 24000,
         encoding: str = "pcm_s16le",
         container: str = "raw",
@@ -329,7 +329,7 @@ class CartesiaHttpTTSService(TTSService):
         *,
         api_key: str,
         voice_id: str,
-        model: str = "sonic-english",
+        model: str = "sonic",
         base_url: str = "https://api.cartesia.ai",
         sample_rate: int = 24000,
         encoding: str = "pcm_s16le",

src/pipecat/services/google/__init__.py

@@ -0,0 +1,2 @@
+from .frames import LLMSearchResponseFrame
+from .google import *

src/pipecat/services/google/frames.py

@@ -0,0 +1,33 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+from pipecat.frames.frames import DataFrame
+
+
+@dataclass
+class LLMSearchResult:
+    text: str
+    confidence: Optional[float] = None
+
+
+@dataclass
+class LLMSearchOrigin:
+    site_uri: Optional[str] = None
+    site_title: Optional[str] = None
+    results: List[LLMSearchResult] = field(default_factory=list)
+
+
+@dataclass
+class LLMSearchResponseFrame(DataFrame):
+    search_result: Optional[str] = None
+    rendered_content: Optional[str] = None
+    origins: List[LLMSearchOrigin] = field(default_factory=list)
+
+    def __str__(self):
+        return f"LLMSearchResponseFrame(search_result={self.search_result}, origins={self.origins})"

src/pipecat/services/google/google.py

@@ -38,6 +38,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import LLMService, TTSService
+from pipecat.services.google.frames import LLMSearchResponseFrame
 from pipecat.services.openai import (
     OpenAIAssistantContextAggregator,
     OpenAIUserContextAggregator,
@@ -639,6 +640,9 @@ class GoogleLLMService(LLMService):
         completion_tokens = 0
         total_tokens = 0
 
+        grounding_metadata = None
+        search_result = ""
+
         try:
             logger.debug(
                 # f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}"
@@ -698,6 +702,7 @@ class GoogleLLMService(LLMService):
                 try:
                     for c in chunk.parts:
                         if c.text:
+                            search_result += c.text
                             await self.push_frame(LLMTextFrame(c.text))
                         elif c.function_call:
                             logger.debug(f"!!! Function call: {c.function_call}")
@@ -708,6 +713,63 @@ class GoogleLLMService(LLMService):
                                 function_name=c.function_call.name,
                                 arguments=args,
                             )
+                    # Handle grounding metadata
+                    # It seems only the last chunk that we receive may contain this information
+                    # If the response doesn't include groundingMetadata, this means the response wasn't grounded.
+                    if chunk.candidates:
+                        for candidate in chunk.candidates:
+                            # logger.debug(f"candidate received: {candidate}")
+                            # Extract grounding metadata
+                            grounding_metadata = (
+                                {
+                                    "rendered_content": getattr(
+                                        getattr(candidate, "grounding_metadata", None),
+                                        "search_entry_point",
+                                        None,
+                                    ).rendered_content
+                                    if hasattr(
+                                        getattr(candidate, "grounding_metadata", None),
+                                        "search_entry_point",
+                                    )
+                                    else None,
+                                    "origins": [
+                                        {
+                                            "site_uri": getattr(grounding_chunk.web, "uri", None),
+                                            "site_title": getattr(
+                                                grounding_chunk.web, "title", None
+                                            ),
+                                            "results": [
+                                                {
+                                                    "text": getattr(
+                                                        grounding_support.segment, "text", ""
+                                                    ),
+                                                    "confidence": getattr(
+                                                        grounding_support, "confidence_scores", None
+                                                    ),
+                                                }
+                                                for grounding_support in getattr(
+                                                    getattr(candidate, "grounding_metadata", None),
+                                                    "grounding_supports",
+                                                    [],
+                                                )
+                                                if index
+                                                in getattr(
+                                                    grounding_support, "grounding_chunk_indices", []
+                                                )
+                                            ],
+                                        }
+                                        for index, grounding_chunk in enumerate(
+                                            getattr(
+                                                getattr(candidate, "grounding_metadata", None),
+                                                "grounding_chunks",
+                                                [],
+                                            )
+                                        )
+                                    ],
+                                }
+                                if getattr(candidate, "grounding_metadata", None)
+                                else None
+                            )
                 except Exception as e:
                     # Google LLMs seem to flag safety issues a lot!
                     if chunk.candidates[0].finish_reason == 3:
@@ -720,6 +782,14 @@ class GoogleLLMService(LLMService):
         except Exception as e:
             logger.exception(f"{self} exception: {e}")
         finally:
+            if grounding_metadata is not None and isinstance(grounding_metadata, dict):
+                llm_search_frame = LLMSearchResponseFrame(
+                    search_result=search_result,
+                    origins=grounding_metadata["origins"],
+                    rendered_content=grounding_metadata["rendered_content"],
+                )
+                await self.push_frame(llm_search_frame)
+
             await self.start_llm_usage_metrics(
                 LLMTokenUsage(
                     prompt_tokens=prompt_tokens,

src/pipecat/transports/services/helpers/daily_rest.py

@@ -33,6 +33,19 @@ class DailyRoomSipParams(BaseModel):
     num_endpoints: int = 1
 
 
+class RecordingsBucketConfig(BaseModel):
+    """Configuration for storing Daily recordings in a custom S3 bucket.
+
+    Refer to the Daily API documentation for more information:
+    https://docs.daily.co/guides/products/live-streaming-recording/storing-recordings-in-a-custom-s3-bucket
+    """
+
+    bucket_name: str
+    bucket_region: str
+    assume_role_arn: str
+    allow_api_access: bool = False
+
+
 class DailyRoomProperties(BaseModel, extra="allow"):
     """Properties for configuring a Daily room.
 
@@ -43,6 +56,8 @@ class DailyRoomProperties(BaseModel, extra="allow"):
         enable_emoji_reactions: Whether emoji reactions are enabled
         eject_at_room_exp: Whether to remove participants when room expires
         enable_dialout: Whether SIP dial-out is enabled
+        enable_recording: Recording settings ('cloud', 'local', 'raw-tracks')
+        geo: Geographic region for room
         max_participants: Maximum number of participants allowed in the room
         sip: SIP configuration parameters
         sip_uri: SIP URI information returned by Daily
@@ -57,7 +72,10 @@ class DailyRoomProperties(BaseModel, extra="allow"):
     enable_emoji_reactions: bool = False
     eject_at_room_exp: bool = True
     enable_dialout: Optional[bool] = None
+    enable_recording: Optional[Literal["cloud", "local", "raw-tracks"]] = None
+    geo: Optional[str] = None
     max_participants: Optional[int] = None
+    recordings_bucket: Optional[RecordingsBucketConfig] = None
     sip: Optional[DailyRoomSipParams] = None
     sip_uri: Optional[dict] = None
     start_video_off: bool = False
@@ -111,6 +129,84 @@ class DailyRoomObject(BaseModel):
     config: DailyRoomProperties
 
 
+class DailyMeetingTokenProperties(BaseModel):
+    """Properties for configuring a Daily meeting token.
+
+    Refer to the Daily API documentation for more information:
+    https://docs.daily.co/reference/rest-api/meeting-tokens/create-meeting-token#properties
+    """
+
+    room_name: Optional[str] = Field(
+        default=None,
+        description="The room for which this token is valid. If not set, the token is valid for all rooms in your domain. You should always set room_name if using this token to control meeting access.",
+    )
+
+    eject_at_token_exp: Optional[bool] = Field(
+        default=None,
+        description="If `true`, the user will be ejected from the room when the token expires. Defaults to `false`.",
+    )
+    eject_after_elapsed: Optional[int] = Field(
+        default=None,
+        description="The number of seconds after which the user will be ejected from the room. If not provided, the user will not be ejected based on elapsed time.",
+    )
+
+    nbf: Optional[int] = Field(
+        default=None,
+        description="Not before. This is a unix timestamp (seconds since the epoch.) Users cannot join a meeting in with this token before this time.",
+    )
+
+    exp: Optional[int] = Field(
+        default=None,
+        description="Expiration time (unix timestamp in seconds). We strongly recommend setting this value for security. If not set, the token will not expire. Refer docs for more info.",
+    )
+    is_owner: Optional[bool] = Field(
+        default=None,
+        description="If `true`, the token will grant owner privileges in the room. Defaults to `false`.",
+    )
+    user_name: Optional[str] = Field(
+        default=None,
+        description="The name of the user. This will be added to the token payload.",
+    )
+    user_id: Optional[str] = Field(
+        default=None,
+        description="A unique identifier for the user. This will be added to the token payload.",
+    )
+    enable_screenshare: Optional[bool] = Field(
+        default=None,
+        description="If `true`, the user will be able to share their screen. Defaults to `true`.",
+    )
+    start_video_off: Optional[bool] = Field(
+        default=None,
+        description="If `true`, the user's video will be turned off when they join the room. Defaults to `false`.",
+    )
+    start_audio_off: Optional[bool] = Field(
+        default=None,
+        description="If `true`, the user's audio will be turned off when they join the room. Defaults to `false`.",
+    )
+    enable_recording: Optional[Literal["cloud", "local", "raw-tracks"]] = Field(
+        default=None,
+        description="Recording settings for the token. Must be one of `cloud`, `local` or `raw-tracks`.",
+    )
+    enable_prejoin_ui: Optional[bool] = Field(
+        default=None,
+        description="If `true`, the user will see the prejoin UI before joining the room.",
+    )
+    start_cloud_recording: Optional[bool] = Field(
+        default=None,
+        description="Start cloud recording when the user joins the room. This can be used to always record and archive meetings, for example in a customer support context.",
+    )
+
+
+class DailyMeetingTokenParams(BaseModel):
+    """Parameters for creating a Daily meeting token.
+
+    Refer to the Daily API documentation for more information:
+    https://docs.daily.co/reference/rest-api/meeting-tokens/create-meeting-token#body-params
+    """
+
+    properties: DailyMeetingTokenProperties = Field(default_factory=DailyMeetingTokenProperties)
+
+
 class DailyRESTHelper:
     """Helper class for interacting with Daily's REST API.
 
@@ -129,6 +225,7 @@ class DailyRESTHelper:
         daily_api_url: str = "https://api.daily.co/v1",
         aiohttp_session: aiohttp.ClientSession,
     ):
+        """Initialize the Daily REST helper."""
         self.daily_api_key = daily_api_key
         self.daily_api_url = daily_api_url
         self.aiohttp_session = aiohttp_session
@@ -169,7 +266,7 @@ class DailyRESTHelper:
             Exception: If room creation fails or response is invalid
         """
         headers = {"Authorization": f"Bearer {self.daily_api_key}"}
-        json = {**params.model_dump(exclude_none=True)}
+        json = params.model_dump(exclude_none=True)
         async with self.aiohttp_session.post(
             f"{self.daily_api_url}/rooms", headers=headers, json=json
         ) as r:
@@ -187,7 +284,11 @@ class DailyRESTHelper:
         return room
 
     async def get_token(
-        self, room_url: str, expiry_time: float = 60 * 60, owner: bool = True
+        self,
+        room_url: str,
+        expiry_time: float = 60 * 60,
+        owner: bool = True,
+        params: Optional[DailyMeetingTokenParams] = None,
     ) -> str:
         """Generate a meeting token for user to join a Daily room.
 
@@ -195,6 +296,7 @@ class DailyRESTHelper:
             room_url: Daily room URL
             expiry_time: Token validity duration in seconds (default: 1 hour)
             owner: Whether token has owner privileges
+            params: Parameters for creating a Daily meeting token
 
         Returns:
             str: Meeting token
@@ -207,12 +309,23 @@ class DailyRESTHelper:
                 "No Daily room specified. You must specify a Daily room in order a token to be generated."
             )
 
-        expiration: float = time.time() + expiry_time
+        expiration: int = int(time.time() + expiry_time)
 
         room_name = self.get_name_from_url(room_url)
 
         headers = {"Authorization": f"Bearer {self.daily_api_key}"}
-        json = {"properties": {"room_name": room_name, "is_owner": owner, "exp": expiration}}
+
+        if params is None:
+            params = DailyMeetingTokenParams(
+                **{"properties": {"room_name": room_name, "is_owner": owner, "exp": expiration}}
+            )
+        else:
+            params.properties.room_name = room_name
+            params.properties.exp = int(expiration)
+            params.properties.is_owner = owner
+
+        json = params.model_dump(exclude_none=True)
+
         async with self.aiohttp_session.post(
             f"{self.daily_api_url}/meeting-tokens", headers=headers, json=json
         ) as r: