Add additional functionality related to "thinking", for Google and Anthropic LLMs.

Thinking, sometimes called "extended thinking" or "reasoning", is an LLM process where the model takes some additional time before giving an answer. It's useful for complex tasks that may require some level of planning and structured, step-by-step reasoning. The model can output its thoughts (or thought summaries, depending on the model) in addition to the answer. The thoughts are usually pretty granular and not really suitable for being spoken out loud in a conversation, but can be useful for logging or prompt debugging.

Here's what's added:

1. New typed input parameters for Google and Anthropic LLMs that control the models' thinking behavior (like how much thinking to do, and whether to output thoughts or thought summaries).
2. New frames for representing thoughts output by LLMs.
3. A generic mechanism for associating extra LLM-specific data with a function call in context, used specifically to support Google's function-call-related "thought signatures", which are necessary to ensure thinking continuity between function calls in a chain (where the model thinks, makes a function call, thinks some more, etc.)
4. A generic mechanism for recording LLM thoughts to context, used specifically to support Anthropic, whose thought signatures are expected to appear alongside the text of the thoughts within assistant context messages.
5. An expansion of `TranscriptProcessor` to process LLM thoughts in addition to user and assistant utterances.

examples/foundational/07n-interruptible-google-http.py

@@ -75,8 +75,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm = GoogleLLMService(
         api_key=os.getenv("GOOGLE_API_KEY"),
         model="gemini-2.5-flash",
-        # turn on thinking if you want it
-        # params=GoogleLLMService.InputParams(extra={"thinking_config": {"thinking_budget": 4096}}),)
+        # force a certain amount of thinking if you want it
+        # params=GoogleLLMService.InputParams(
+        #     thinking=GoogleLLMService.ThinkingConfig(thinking_budget=4096)
+        # ),
     )
 
     messages = [

examples/foundational/07n-interruptible-google.py

@@ -75,8 +75,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm = GoogleLLMService(
         api_key=os.getenv("GOOGLE_API_KEY"),
         model="gemini-2.5-flash",
-        # turn on thinking if you want it
-        # params=GoogleLLMService.InputParams(extra={"thinking_config": {"thinking_budget": 4096}}),)
+        # force a certain amount of thinking if you want it
+        # params=GoogleLLMService.InputParams(
+        #     thinking=GoogleLLMService.ThinkingConfig(thinking_budget=4096)
+        # ),
     )
 
     messages = [

examples/foundational/07s-interruptible-google-audio-in.py

@@ -224,8 +224,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     llm = GoogleLLMService(
         api_key=os.getenv("GOOGLE_API_KEY"),
         model="gemini-2.5-flash",
-        # turn on thinking if you want it
-        # params=GoogleLLMService.InputParams(extra={"thinking_config": {"thinking_budget": 4096}}),
+        # force a certain amount of thinking if you want it
+        # params=GoogleLLMService.InputParams(
+        #     thinking=GoogleLLMService.ThinkingConfig(thinking_budget=4096)
+        # ),
     )
 
     tts = GoogleTTSService(

examples/foundational/49-thinking-functions.py

@@ -0,0 +1,222 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+import random
+import sys
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.processors.transcript_processor import TranscriptProcessor
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.anthropic.llm import AnthropicLLMService
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def check_flight_status(params: FunctionCallParams, flight_number: str):
+    """Check the status of a flight. Returns status (e.g., "on time", "delayed") and departure time.
+
+    Args:
+        flight_number (str): The flight number, e.g. "AA100".
+    """
+    await params.result_callback({"status": "delayed", "departure_time": "14:30"})
+
+
+async def book_taxi(params: FunctionCallParams, time: str):
+    """Book a taxi for a given time. Returns status (e.g., "done").
+
+    Args:
+        time (str): The time to book the taxi for, e.g. "15:00".
+    """
+    await params.result_callback({"status": "done"})
+
+
+# LLM provider constants
+LLM_ANTHROPIC = "anthropic"
+LLM_GOOGLE = "google"
+LLM_DEFAULT = LLM_GOOGLE
+
+# 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(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+}
+
+
+async def run_bot(
+    transport: BaseTransport, runner_args: RunnerArguments, llm_provider: str = LLM_DEFAULT
+):
+    logger.info(f"Starting bot with {llm_provider.capitalize()} LLM")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    if llm_provider == LLM_ANTHROPIC:
+        llm = AnthropicLLMService(
+            api_key=os.getenv("ANTHROPIC_API_KEY"),
+            params=AnthropicLLMService.InputParams(
+                thinking=AnthropicLLMService.ThinkingConfig(type="enabled", budget_tokens=2048)
+            ),
+        )
+    elif llm_provider == LLM_GOOGLE:
+        llm = GoogleLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            params=GoogleLLMService.InputParams(
+                thinking=GoogleLLMService.ThinkingConfig(
+                    thinking_budget=-1,  # Dynamic thinking
+                    include_thoughts=True,
+                )
+            ),
+        )
+    else:
+        raise ValueError(f"Unsupported LLM provider: {llm_provider}")
+
+    llm.register_direct_function(check_flight_status)
+    llm.register_direct_function(book_taxi)
+
+    tools = ToolsSchema(standard_tools=[check_flight_status, book_taxi])
+
+    transcript = TranscriptProcessor()
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = LLMContext(messages, tools)
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            transcript.user(),  # User transcripts
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            transcript.thought(),  # Thought transcripts
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            transcript.assistant(),  # Assistant transcripts
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    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.
+        # This example comes from Gemini docs.
+        messages.append(
+            {
+                "role": "user",
+                "content": "Check the status of flight AA100 and book me a taxi 2 hours beforehand if the flight is delayed.",
+            }
+        )
+        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()
+
+    @transcript.event_handler("on_transcript_update")
+    async def on_transcript_update(processor, frame):
+        for msg in frame.messages:
+            if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)):
+                timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+                role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role
+                logger.info(f"Transcript: {timestamp}{role}: {msg.content}")
+
+    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."""
+    # Get llm_provider from module attribute set in __main__
+    llm_provider = getattr(sys.modules[__name__], "llm_provider", LLM_DEFAULT)
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args, llm_provider)
+
+
+if __name__ == "__main__":
+    # Parse custom arguments before calling runner main()
+    parser = argparse.ArgumentParser(description="Thinking LLM Bot")
+    parser.add_argument(
+        "--llm",
+        type=str,
+        choices=[LLM_ANTHROPIC, LLM_GOOGLE],
+        default=LLM_DEFAULT,
+        help=f"LLM provider to use (default: {LLM_DEFAULT})",
+    )
+    # Parse only known args to allow runner's main() to handle its own args
+    args, remaining = parser.parse_known_args()
+
+    # Store the llm_provider in sys.modules for bot() function to access
+    sys.modules[__name__].llm_provider = args.llm
+
+    # Restore sys.argv with remaining args for runner's main()
+    sys.argv[1:] = remaining
+
+    from pipecat.runner.run import main
+
+    main()

examples/foundational/49-thinking.py

@@ -0,0 +1,198 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+import random
+import sys
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import LLMRunFrame, ThoughtTranscriptionMessage, TranscriptionMessage
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.processors.transcript_processor import TranscriptProcessor
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.anthropic.llm import AnthropicLLMService
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.google.llm import GoogleLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+# LLM provider constants
+LLM_ANTHROPIC = "anthropic"
+LLM_GOOGLE = "google"
+LLM_DEFAULT = LLM_GOOGLE
+
+# 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(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+        turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()),
+    ),
+}
+
+
+async def run_bot(
+    transport: BaseTransport, runner_args: RunnerArguments, llm_provider: str = LLM_DEFAULT
+):
+    logger.info(f"Starting bot with {llm_provider.capitalize()} LLM")
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    if llm_provider == LLM_ANTHROPIC:
+        llm = AnthropicLLMService(
+            api_key=os.getenv("ANTHROPIC_API_KEY"),
+            params=AnthropicLLMService.InputParams(
+                thinking=AnthropicLLMService.ThinkingConfig(type="enabled", budget_tokens=2048)
+            ),
+        )
+    elif llm_provider == LLM_GOOGLE:
+        llm = GoogleLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            params=GoogleLLMService.InputParams(
+                thinking=GoogleLLMService.ThinkingConfig(
+                    thinking_budget=-1,  # Dynamic thinking
+                    include_thoughts=True,
+                )
+            ),
+        )
+    else:
+        raise ValueError(f"Unsupported LLM provider: {llm_provider}")
+
+    transcript = TranscriptProcessor()
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = LLMContext(messages)
+    context_aggregator = LLMContextAggregatorPair(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            transcript.user(),  # User transcripts
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            transcript.thought(),  # Thought transcripts
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            transcript.assistant(),  # Assistant transcripts
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    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")
+        # Choose a random prompt to demonstrate thinking capabilities.
+        # These prompts were chosen from Google and Anthropic docs.
+        thinking_prompt_1 = "Analogize photosynthesis and growing up."
+        thinking_prompt_2 = "Compare and contrast electric cars and hybrid cars."
+        thinking_prompt_3 = "Are there an infinite number of prime numbers such that n mod 4 == 3?"
+        selected_prompt = random.choice([thinking_prompt_1, thinking_prompt_2, thinking_prompt_3])
+
+        # Kick off the conversation.
+        messages.append({"role": "user", "content": selected_prompt})
+        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()
+
+    # Register event handler for transcript updates
+    @transcript.event_handler("on_transcript_update")
+    async def on_transcript_update(processor, frame):
+        for msg in frame.messages:
+            if isinstance(msg, (ThoughtTranscriptionMessage, TranscriptionMessage)):
+                timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+                role = "THOUGHT" if isinstance(msg, ThoughtTranscriptionMessage) else msg.role
+                logger.info(f"Transcript: {timestamp}{role}: {msg.content}")
+
+    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."""
+    # Get llm_provider from module attribute set in __main__
+    llm_provider = getattr(sys.modules[__name__], "llm_provider", LLM_DEFAULT)
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args, llm_provider)
+
+
+if __name__ == "__main__":
+    # Parse custom arguments before calling runner main()
+    parser = argparse.ArgumentParser(description="Thinking LLM Bot")
+    parser.add_argument(
+        "--llm",
+        type=str,
+        choices=[LLM_ANTHROPIC, LLM_GOOGLE],
+        default=LLM_DEFAULT,
+        help=f"LLM provider to use (default: {LLM_DEFAULT})",
+    )
+    # Parse only known args to allow runner's main() to handle its own args
+    args, remaining = parser.parse_known_args()
+
+    # Store the llm_provider in sys.modules for bot() function to access
+    sys.modules[__name__].llm_provider = args.llm
+
+    # Restore sys.argv with remaining args for runner's main()
+    sys.argv[1:] = remaining
+
+    from pipecat.runner.run import main
+
+    main()

src/pipecat/adapters/services/anthropic_adapter.py

@@ -165,9 +165,43 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
 
     def _from_universal_context_message(self, message: LLMContextMessage) -> MessageParam:
         if isinstance(message, LLMSpecificMessage):
-            return copy.deepcopy(message.message)
+            return self._from_anthropic_specific_message(message)
         return self._from_standard_message(message)
 
+    def _from_anthropic_specific_message(self, message: LLMSpecificMessage) -> MessageParam:
+        """Convert LLMSpecificMessage to Anthropic format.
+
+        Assumes that we already know the message is intended for Anthropic.
+
+        Args:
+            message: Message in LLMSpecificMessage format.
+        """
+        # Handle special case of thought messages.
+        # These can be converted to standalone "assistant" messages; later
+        # these thinking messages will be properly merged into the assistant
+        # response messages before the context is sent to Anthropic for the
+        # next turn.
+        if (
+            isinstance(message.message, dict)
+            and message.message.get("type") == "thought"
+            and (text := message.message.get("text"))
+            and isinstance(metadata := message.message.get("metadata"), dict)
+            and (signature := metadata.get("signature"))
+        ):
+            return {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "thinking",
+                        "thinking": text,
+                        "signature": signature,
+                    }
+                ],
+            }
+
+        # Fallback to assumption that the message is already in Anthropic format
+        return copy.deepcopy(message.message)
+
     def _from_standard_message(self, message: LLMStandardMessage) -> MessageParam:
         """Convert standard universal context message to Anthropic format.
 

src/pipecat/adapters/services/gemini_adapter.py

@@ -167,6 +167,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
     class MessageConversionResult:
         """Result of converting a single universal context message to Google format.
 
+        # TODO: content could be other things, like {"tool_call_extra": ...}, for example. All bets are off when it's LLMSpecificMessage.
         Either content (a Google Content object) or a system instruction string
         is guaranteed to be set.
 
@@ -219,6 +220,20 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                     tool_call_id_to_name_mapping=tool_call_id_to_name_mapping,
                 ),
             )
+
+            # If we found a function-call-related thought_signature, modify the
+            # corresponding function call message to include it
+            if (
+                isinstance(result.content, dict)
+                and result.content.get("type") == "tool_call_extra"
+                and isinstance(data := result.content.get("data"), dict)
+                and (thought_signature := data.get("thought_signature"))
+            ):
+                self._apply_function_call_thought_signature_to_messages(
+                    thought_signature, result.content.get("tool_call_id"), messages
+                )
+                continue
+
             # Each result is either a Content or a system instruction
             if result.content:
                 messages.append(result.content)
@@ -410,3 +425,32 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
             content=Content(role=role, parts=parts),
             tool_call_id_to_name_mapping=tool_call_id_to_name_mapping,
         )
+
+    def _apply_function_call_thought_signature_to_messages(
+        self, thought_signature: bytes, tool_call_id: str, messages: List[Content]
+    ) -> None:
+        """Apply tool_call_extra metadata to the corresponding function call message.
+
+        Args:
+            thought_signature: The thought signature bytes to apply.
+            tool_call_id: ID of the tool call message to find and modify.
+            messages: List of Content messages to search through.
+        """
+        # Search backwards through messages to find the matching function call
+        for message in reversed(messages):
+            if not isinstance(message, Content) or not message.parts:
+                continue
+            # Find the specific part with the matching function call
+            for part in message.parts:
+                if (
+                    hasattr(part, "function_call")
+                    and part.function_call
+                    and part.function_call.id == tool_call_id
+                ):
+                    part.thought_signature = thought_signature
+                    break
+            else:
+                # Continue outer loop if inner loop didn't break
+                continue
+            # Break outer loop if inner loop broke (found match)
+            break

src/pipecat/frames/frames.py

@@ -512,6 +512,14 @@ class TranscriptionMessage:
     timestamp: Optional[str] = None
 
 
+@dataclass
+class ThoughtTranscriptionMessage:
+    """An LLM thought message in a conversation transcript."""
+
+    content: str
+    timestamp: Optional[str] = None
+
+
 @dataclass
 class TranscriptionUpdateFrame(DataFrame):
     """Frame containing new messages added to conversation transcript.
@@ -556,7 +564,7 @@ class TranscriptionUpdateFrame(DataFrame):
         messages: List of new transcript messages that were added.
     """
 
-    messages: List[TranscriptionMessage]
+    messages: List[TranscriptionMessage | ThoughtTranscriptionMessage]
 
     def __str__(self):
         pts = format_pts(self.pts)
@@ -577,6 +585,73 @@ class LLMContextFrame(Frame):
     context: "LLMContext"
 
 
+@dataclass
+class LLMThoughtStartFrame(ControlFrame):
+    """Frame indicating the start of an LLM thought.
+
+    Parameters:
+        append_to_context: Whether the thought should be appended to the LLM context.
+            If it is appended, the `llm` field is required, since it will be
+            appended as an `LLMSpecificMessage`.
+        llm: Optional identifier of the LLM provider for LLM-specific handling.
+            Only required if `append_to_context` is True.
+    """
+
+    append_to_context: bool = False
+    llm: Optional[str] = None
+
+    def __post_init__(self):
+        super().__post_init__()
+        if self.append_to_context and self.llm is None:
+            raise ValueError("When append_to_context is True, llm must be set")
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return (
+            f"{self.name}(pts: {pts}, append_to_context: {self.append_to_context}, llm: {self.llm})"
+        )
+
+
+@dataclass
+class LLMThoughtTextFrame(DataFrame):
+    """Frame containing the text (or text chunk) of an LLM thought.
+
+    Note that despite this containing text, it is a DataFrame and not a
+    TextFrame, to avoid most typical text processing, such as TTS.
+
+    Parameters:
+        text: The text (or text chunk) of the thought.
+    """
+
+    text: str
+    includes_inter_frame_spaces: bool = field(init=False)
+
+    def __post_init__(self):
+        super().__post_init__()
+        # Assume that thought text chunks include all necessary spaces
+        self.includes_inter_frame_spaces = True
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, thought text: {self.text})"
+
+
+@dataclass
+class LLMThoughtEndFrame(ControlFrame):
+    """Frame indicating the end of an LLM thought.
+
+    Parameters:
+        thought_metadata: Optional metadata associated with the thought,
+            e.g. thought signature.
+    """
+
+    thought_metadata: Optional[Dict[str, Any]] = None
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, metadata: {self.thought_metadata})"
+
+
 @dataclass
 class LLMMessagesFrame(DataFrame):
     """Frame containing LLM messages for chat completion.
@@ -1119,12 +1194,16 @@ class FunctionCallFromLLM:
         tool_call_id: A unique identifier for the function call.
         arguments: The arguments to pass to the function.
         context: The LLM context when the function call was made.
+        llm_specific_extra: Optional extra data specific to particular LLMs, e.g.:
+            {"google": {"thought_signature": ...}}
+            Uses the LLM adapter's ID for LLM-specific messages as the key.
     """
 
     function_name: str
     tool_call_id: str
     arguments: Mapping[str, Any]
     context: Any
+    llm_specific_extra: Optional[Dict[str, Any]] = None
 
 
 @dataclass
@@ -1662,6 +1741,9 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame):
         function_name: Name of the function being executed.
         tool_call_id: Unique identifier for this function call.
         arguments: Arguments passed to the function.
+        llm_specific_extra: Optional extra data specific to particular LLMs, e.g.:
+            {"google": {"thought_signature": ...}}
+            Uses the LLM adapter's ID for LLM-specific messages as the key.
         cancel_on_interruption: Whether to cancel this call if interrupted.
 
     """
@@ -1669,6 +1751,7 @@ class FunctionCallInProgressFrame(ControlFrame, UninterruptibleFrame):
     function_name: str
     tool_call_id: str
     arguments: Any
+    llm_specific_extra: Optional[Dict[str, Any]] = None
     cancel_on_interruption: bool = False
 
 

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -47,6 +47,9 @@ from pipecat.frames.frames import (
     LLMRunFrame,
     LLMSetToolChoiceFrame,
     LLMSetToolsFrame,
+    LLMThoughtEndFrame,
+    LLMThoughtStartFrame,
+    LLMThoughtTextFrame,
     SpeechControlParamsFrame,
     StartFrame,
     TextFrame,
@@ -592,6 +595,10 @@ class LLMAssistantAggregator(LLMContextAggregator):
         self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
         self._context_updated_tasks: Set[asyncio.Task] = set()
 
+        self._thought_aggregation_enabled = False
+        self._thought_llm: str = ""
+        self._thought_aggregation: List[TextPartForConcatenation] = []
+
     @property
     def has_function_calls_in_progress(self) -> bool:
         """Check if there are any function calls currently in progress.
@@ -601,6 +608,17 @@ class LLMAssistantAggregator(LLMContextAggregator):
         """
         return bool(self._function_calls_in_progress)
 
+    async def reset(self):
+        """Reset the aggregation state."""
+        await super().reset()
+        await self._reset_thought_aggregation()  # Just to be safe
+
+    async def _reset_thought_aggregation(self):
+        """Reset the thought aggregation state."""
+        self._thought_aggregation_enabled = False
+        self._thought_llm = ""
+        self._thought_aggregation = []
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames for assistant response aggregation and function call management.
 
@@ -619,6 +637,12 @@ class LLMAssistantAggregator(LLMContextAggregator):
             await self._handle_llm_end(frame)
         elif isinstance(frame, TextFrame):
             await self._handle_text(frame)
+        elif isinstance(frame, LLMThoughtStartFrame):
+            await self._handle_thought_start(frame)
+        elif isinstance(frame, LLMThoughtTextFrame):
+            await self._handle_thought_text(frame)
+        elif isinstance(frame, LLMThoughtEndFrame):
+            await self._handle_thought_end(frame)
         elif isinstance(frame, LLMRunFrame):
             await self._handle_llm_run(frame)
         elif isinstance(frame, LLMMessagesAppendFrame):
@@ -716,6 +740,24 @@ class LLMAssistantAggregator(LLMContextAggregator):
             }
         )
 
+        # If there's LLM-specific extra data associated with this function call
+        # add it to the context as an adjacent LLM-specific message. The
+        # LLM-specific adapter can then use this extra data as needed, for
+        # example by merging it into the tool call message. This is how Google's
+        # "thought_signature" makes it into the tool call message.
+        if frame.llm_specific_extra:
+            for key, value in frame.llm_specific_extra.items():
+                self._context.add_message(
+                    LLMSpecificMessage(
+                        llm=key,
+                        message={
+                            "type": "tool_call_extra",
+                            "data": value,
+                            "tool_call_id": frame.tool_call_id,
+                        },
+                    )
+                )
+
         self._function_calls_in_progress[frame.tool_call_id] = frame
 
     async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
@@ -824,6 +866,47 @@ class LLMAssistantAggregator(LLMContextAggregator):
             )
         )
 
+    async def _handle_thought_start(self, frame: LLMThoughtStartFrame):
+        if not self._started:
+            return
+
+        await self._reset_thought_aggregation()
+        self._thought_aggregation_enabled = frame.append_to_context
+        self._thought_llm = frame.llm
+
+    async def _handle_thought_text(self, frame: LLMThoughtTextFrame):
+        if not self._started or not self._thought_aggregation_enabled:
+            return
+
+        # Make sure we really have text (spaces count, too!)
+        if len(frame.text) == 0:
+            return
+
+        self._thought_aggregation.append(
+            TextPartForConcatenation(
+                frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
+            )
+        )
+
+    async def _handle_thought_end(self, frame: LLMThoughtEndFrame):
+        if not self._started or not self._thought_aggregation_enabled:
+            return
+
+        thought = concatenate_aggregated_text(self._thought_aggregation)
+        llm = self._thought_llm
+        await self._reset_thought_aggregation()
+
+        self._context.add_message(
+            LLMSpecificMessage(
+                llm=llm,
+                message={
+                    "type": "thought",
+                    "text": thought,
+                    "metadata": frame.thought_metadata,
+                },
+            )
+        )
+
     def _context_updated_task_finished(self, task: asyncio.Task):
         self._context_updated_tasks.discard(task)
 

src/pipecat/processors/transcript_processor.py

@@ -20,6 +20,10 @@ from pipecat.frames.frames import (
     EndFrame,
     Frame,
     InterruptionFrame,
+    LLMThoughtEndFrame,
+    LLMThoughtStartFrame,
+    LLMThoughtTextFrame,
+    ThoughtTranscriptionMessage,
     TranscriptionFrame,
     TranscriptionMessage,
     TranscriptionUpdateFrame,
@@ -202,10 +206,113 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
             await self.push_frame(frame, direction)
 
 
+class ThoughtTranscriptProcessor(BaseTranscriptProcessor):
+    """Processes LLM thought frames into timestamped thought messages.
+
+    This processor aggregates LLM thought text frames into complete thoughts
+    and emits them as thought transcript messages. Thoughts are completed when:
+
+    - A thought ends (LLMThoughtEndFrame)
+    - The bot is interrupted (InterruptionFrame)
+    - The pipeline ends (EndFrame)
+    """
+
+    def __init__(self, **kwargs):
+        """Initialize processor with thought aggregation state.
+
+        Args:
+            **kwargs: Additional arguments passed to parent class.
+        """
+        super().__init__(**kwargs)
+        self._current_thought_parts: List[TextPartForConcatenation] = []
+        self._thought_start_time: Optional[str] = None
+        self._thought_active = False
+
+    async def _emit_aggregated_thought(self):
+        """Aggregates and emits thought text fragments as a thought transcript message.
+
+        This method aggregates thought fragments that may arrive in multiple
+        LLMThoughtTextFrame instances and emits them as a single ThoughtTranscriptionMessage.
+        """
+        if self._current_thought_parts and self._thought_start_time:
+            content = concatenate_aggregated_text(self._current_thought_parts)
+            if content:
+                logger.trace(f"Emitting aggregated thought message: {content}")
+                message = ThoughtTranscriptionMessage(
+                    content=content,
+                    timestamp=self._thought_start_time,
+                )
+                await self._emit_update([message])
+            else:
+                logger.trace("No thought content to emit after stripping whitespace")
+
+            # Reset aggregation state
+            self._current_thought_parts = []
+            self._thought_start_time = None
+            self._thought_active = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames into thought transcript messages.
+
+        Handles different frame types:
+
+        - LLMThoughtStartFrame: Begins aggregating a new thought
+        - LLMThoughtTextFrame: Aggregates text for current thought
+        - LLMThoughtEndFrame: Completes current thought
+        - InterruptionFrame: Completes current thought due to interruption
+        - EndFrame: Completes current thought at pipeline end
+        - CancelFrame: Completes current thought due to cancellation
+
+        Args:
+            frame: Input frame to process.
+            direction: Frame processing direction.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, (InterruptionFrame, CancelFrame)):
+            # Push frame first otherwise our emitted transcription update frame
+            # might get cleaned up.
+            await self.push_frame(frame, direction)
+            # Emit accumulated thought with interruptions
+            if self._thought_active:
+                await self._emit_aggregated_thought()
+        elif isinstance(frame, LLMThoughtStartFrame):
+            # Start a new thought
+            self._thought_active = True
+            self._thought_start_time = time_now_iso8601()
+            self._current_thought_parts = []
+            # Push frame.
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, LLMThoughtTextFrame):
+            # Aggregate thought text if we have an active thought
+            if self._thought_active:
+                self._current_thought_parts.append(
+                    TextPartForConcatenation(
+                        frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces
+                    )
+                )
+            # Push frame.
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, LLMThoughtEndFrame):
+            # Emit accumulated thought when thought ends
+            if self._thought_active:
+                await self._emit_aggregated_thought()
+            # Push frame.
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, EndFrame):
+            # Emit accumulated thought at pipeline end if still active
+            if self._thought_active:
+                await self._emit_aggregated_thought()
+            # Push frame.
+            await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
+
+
 class TranscriptProcessor:
     """Factory for creating and managing transcript processors.
 
-    Provides unified access to user and assistant transcript processors
+    Provides unified access to user, assistant, and thought transcript processors
     with shared event handling.
 
     Example::
@@ -219,9 +326,10 @@ class TranscriptProcessor:
                 transcript.user(),              # User transcripts
                 context_aggregator.user(),
                 llm,
+                transcript.thought(),           # Thought transcripts
                 tts,
                 transport.output(),
-                transcript.assistant_tts(),     # Assistant transcripts
+                transcript.assistant(),         # Assistant transcripts
                 context_aggregator.assistant(),
             ]
         )
@@ -235,6 +343,7 @@ class TranscriptProcessor:
         """Initialize factory."""
         self._user_processor = None
         self._assistant_processor = None
+        self._thought_processor = None
         self._event_handlers = {}
 
     def user(self, **kwargs) -> UserTranscriptProcessor:
@@ -277,6 +386,26 @@ class TranscriptProcessor:
 
         return self._assistant_processor
 
+    def thought(self, **kwargs) -> ThoughtTranscriptProcessor:
+        """Get the thought transcript processor.
+
+        Args:
+            **kwargs: Arguments specific to ThoughtTranscriptProcessor.
+
+        Returns:
+            The thought transcript processor instance.
+        """
+        if self._thought_processor is None:
+            self._thought_processor = ThoughtTranscriptProcessor(**kwargs)
+            # Apply any registered event handlers
+            for event_name, handler in self._event_handlers.items():
+
+                @self._thought_processor.event_handler(event_name)
+                async def thought_handler(processor, frame):
+                    return await handler(processor, frame)
+
+        return self._thought_processor
+
     def event_handler(self, event_name: str):
         """Register event handler for both processors.
 
@@ -303,6 +432,12 @@ class TranscriptProcessor:
                 async def assistant_handler(processor, frame):
                     return await handler(processor, frame)
 
+            if self._thought_processor:
+
+                @self._thought_processor.event_handler(event_name)
+                async def thought_handler(processor, frame):
+                    return await handler(processor, frame)
+
             return handler
 
         return decorator

src/pipecat/services/anthropic/llm.py

@@ -17,7 +17,7 @@ import io
 import json
 import re
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Dict, List, Literal, Optional, Union
 
 import httpx
 from loguru import logger
@@ -40,6 +40,9 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMTextFrame,
+    LLMThoughtEndFrame,
+    LLMThoughtStartFrame,
+    LLMThoughtTextFrame,
     LLMUpdateSettingsFrame,
     UserImageRawFrame,
 )
@@ -110,6 +113,24 @@ class AnthropicLLMService(LLMService):
     # Overriding the default adapter to use the Anthropic one.
     adapter_class = AnthropicLLMAdapter
 
+    class ThinkingConfig(BaseModel):
+        """Configuration for extended thinking.
+
+        Parameters:
+            type: Type of thinking mode (currently only "enabled" or "disabled").
+            budget_tokens: Maximum number of tokens for thinking.
+                With today's models, the minimum is 1024.
+                Only allowed if type is "enabled".
+        """
+
+        # Why `| str` here? To not break compatibility in case Anthropic adds
+        # more types in the future.
+        type: Literal["enabled", "disabled"] | str
+
+        # Why not enforce minimnum of 1024 here? To not break compatibility in
+        # case Anthropic changes this requirement in the future.
+        budget_tokens: int
+
     class InputParams(BaseModel):
         """Input parameters for Anthropic model inference.
 
@@ -124,6 +145,10 @@ class AnthropicLLMService(LLMService):
             temperature: Sampling temperature between 0.0 and 1.0.
             top_k: Top-k sampling parameter.
             top_p: Top-p sampling parameter between 0.0 and 1.0.
+            thinking: Extended thinking configuration.
+                Enabling extended thinking causes the model to spend more time "thinking" before responding.
+                It also causes this service to emit LLMThinking*Frames during response generation.
+                Extended thinking is disabled by default.
             extra: Additional parameters to pass to the API.
         """
 
@@ -133,6 +158,9 @@ class AnthropicLLMService(LLMService):
         temperature: Optional[float] = Field(default_factory=lambda: NOT_GIVEN, ge=0.0, le=1.0)
         top_k: Optional[int] = Field(default_factory=lambda: NOT_GIVEN, ge=0)
         top_p: Optional[float] = Field(default_factory=lambda: NOT_GIVEN, ge=0.0, le=1.0)
+        thinking: Optional["AnthropicLLMService.ThinkingConfig"] = Field(
+            default_factory=lambda: NOT_GIVEN
+        )
         extra: Optional[Dict[str, Any]] = Field(default_factory=dict)
 
         def model_post_init(self, __context):
@@ -191,6 +219,7 @@ class AnthropicLLMService(LLMService):
             "temperature": params.temperature,
             "top_k": params.top_k,
             "top_p": params.top_p,
+            "thinking": params.thinking,
             "extra": params.extra if isinstance(params.extra, dict) else {},
         }
 
@@ -354,12 +383,21 @@ class AnthropicLLMService(LLMService):
                 "top_p": self._settings["top_p"],
             }
 
+            # Add thinking parameter if set
+            if self._settings["thinking"]:
+                params["thinking"] = self._settings["thinking"].model_dump(exclude_unset=True)
+
             # Messages, system, tools
             params.update(params_from_context)
 
             params.update(self._settings["extra"])
 
-            response = await self._create_message_stream(self._client.messages.create, params)
+            # "Interleaved thinking" needed to allow thinking between sequences
+            # of function calls, when extended thinking is enabled.
+            # Note that this requires us to use `client.beta`, below.
+            params.update({"betas": ["interleaved-thinking-2025-05-14"]})
+
+            response = await self._create_message_stream(self._client.beta.messages.create, params)
 
             await self.stop_ttfb_metrics()
 
@@ -380,10 +418,25 @@ class AnthropicLLMService(LLMService):
                         completion_tokens_estimate += self._estimate_tokens(
                             event.delta.partial_json
                         )
+                    elif hasattr(event.delta, "thinking"):
+                        await self.push_frame(LLMThoughtTextFrame(text=event.delta.thinking))
+                    elif hasattr(event.delta, "signature"):
+                        await self.push_frame(
+                            LLMThoughtEndFrame(
+                                thought_metadata={"signature": event.delta.signature}
+                            )
+                        )
                 elif event.type == "content_block_start":
                     if event.content_block.type == "tool_use":
                         tool_use_block = event.content_block
                         json_accumulator = ""
+                    elif event.content_block.type == "thinking":
+                        await self.push_frame(
+                            LLMThoughtStartFrame(
+                                append_to_context=True,
+                                llm=self.get_llm_adapter().id_for_llm_specific_messages,
+                            )
+                        )
                 elif (
                     event.type == "message_delta"
                     and hasattr(event.delta, "stop_reason")

src/pipecat/services/google/llm.py

@@ -16,7 +16,7 @@ import json
 import os
 import uuid
 from dataclasses import dataclass
-from typing import Any, AsyncIterator, Dict, List, Optional
+from typing import Any, AsyncIterator, Dict, List, Literal, Optional
 
 from loguru import logger
 from PIL import Image
@@ -34,6 +34,9 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMTextFrame,
+    LLMThoughtEndFrame,
+    LLMThoughtStartFrame,
+    LLMThoughtTextFrame,
     LLMUpdateSettingsFrame,
     OutputImageRawFrame,
     UserImageRawFrame,
@@ -665,6 +668,34 @@ class GoogleLLMService(LLMService):
     # Overriding the default adapter to use the Gemini one.
     adapter_class = GeminiLLMAdapter
 
+    class ThinkingConfig(BaseModel):
+        """Configuration for controlling the model's internal "thinking" process used before generating a response.
+
+        Gemini 2.5 and 3 series models have this thinking process.
+
+        Parameters:
+            thinking_level: Thinking level for Gemini 3 Pro. Can be "low" or "high".
+                If not provided, Gemini 3 Pro defaults to "high".
+                Note: Gemini 2.5 series should use thinking_budget instead.
+            thinking_budget: Token budget for thinking, for Gemini 2.5 series.
+                -1 for dynamic thinking (model decides), 0 to disable thinking,
+                or a specific token count (e.g., 128-32768 for 2.5 Pro).
+                If not provided, most models today default to dynamic thinking.
+                See https://ai.google.dev/gemini-api/docs/thinking#set-budget
+                for default values and allowed ranges.
+                Note: Gemini 3 Pro should use thinking_level instead.
+            include_thoughts: Whether to include thought summaries in the response.
+                Today's models default to not including thoughts (False).
+        """
+
+        thinking_budget: Optional[int] = Field(default=None)
+
+        # Why `| str` here? To not break compatibility in case Google adds more
+        # levels in the future.
+        thinking_level: Optional[Literal["low", "high"] | str] = Field(default=None)
+
+        include_thoughts: Optional[bool] = Field(default=None)
+
     class InputParams(BaseModel):
         """Input parameters for Google AI models.
 
@@ -673,6 +704,12 @@ class GoogleLLMService(LLMService):
             temperature: Sampling temperature between 0.0 and 2.0.
             top_k: Top-k sampling parameter.
             top_p: Top-p sampling parameter between 0.0 and 1.0.
+            thinking: Thinking configuration with thinking_budget, thinking_level, and include_thoughts.
+                Used to control the model's internal "thinking" process used before generating a response.
+                Gemini 2.5 series models use thinking_budget; Gemini 3 models use thinking_level.
+                If this is not provided, Pipecat disables thinking for all
+                models where that's possible (the 2.5 series, except 2.5 Pro),
+                to reduce latency.
             extra: Additional parameters as a dictionary.
         """
 
@@ -680,6 +717,7 @@ class GoogleLLMService(LLMService):
         temperature: Optional[float] = Field(default=None, ge=0.0, le=2.0)
         top_k: Optional[int] = Field(default=None, ge=0)
         top_p: Optional[float] = Field(default=None, ge=0.0, le=1.0)
+        thinking: Optional["GoogleLLMService.ThinkingConfig"] = Field(default=None)
         extra: Optional[Dict[str, Any]] = Field(default_factory=dict)
 
     def __init__(
@@ -720,6 +758,7 @@ class GoogleLLMService(LLMService):
             "temperature": params.temperature,
             "top_k": params.top_k,
             "top_p": params.top_p,
+            "thinking": params.thinking,
             "extra": params.extra if isinstance(params.extra, dict) else {},
         }
         self._tools = tools
@@ -830,6 +869,12 @@ class GoogleLLMService(LLMService):
             if v is not None
         }
 
+        # Add thinking parameters if configured
+        if self._settings["thinking"]:
+            generation_params["thinking_config"] = self._settings["thinking"].model_dump(
+                exclude_unset=True
+            )
+
         if self._settings["extra"]:
             generation_params.update(self._settings["extra"])
 
@@ -918,9 +963,17 @@ class GoogleLLMService(LLMService):
                 for candidate in chunk.candidates:
                     if candidate.content and candidate.content.parts:
                         for part in candidate.content.parts:
-                            if not part.thought and part.text:
-                                search_result += part.text
-                                await self.push_frame(LLMTextFrame(part.text))
+                            if part.text:
+                                if part.thought:
+                                    # Gemini emits fully-formed thoughts rather
+                                    # than chunks so bracket each thought in
+                                    # start/end
+                                    await self.push_frame(LLMThoughtStartFrame())
+                                    await self.push_frame(LLMThoughtTextFrame(part.text))
+                                    await self.push_frame(LLMThoughtEndFrame())
+                                else:
+                                    search_result += part.text
+                                    await self.push_frame(LLMTextFrame(part.text))
                             elif part.function_call:
                                 function_call = part.function_call
                                 id = function_call.id or str(uuid.uuid4())
@@ -931,6 +984,13 @@ class GoogleLLMService(LLMService):
                                         tool_call_id=id,
                                         function_name=function_call.name,
                                         arguments=function_call.args or {},
+                                        llm_specific_extra={
+                                            self.get_llm_adapter().id_for_llm_specific_messages: {
+                                                "thought_signature": part.thought_signature
+                                            }
+                                        }
+                                        if part.thought_signature
+                                        else None,
                                     )
                                 )
                             elif part.inline_data and part.inline_data.data:

src/pipecat/services/llm_service.py

@@ -127,6 +127,9 @@ class FunctionCallRunnerItem:
         tool_call_id: A unique identifier for the function call.
         arguments: The arguments for the function.
         context: The LLM context.
+        llm_specific_extra: Optional extra data specific to particular LLMs, e.g.:
+            {"google": {"thought_signature": ...}}
+            Uses the LLM adapter's ID for LLM-specific messages as the key.
         run_llm: Optional flag to control LLM execution after function call.
     """
 
@@ -135,6 +138,7 @@ class FunctionCallRunnerItem:
     tool_call_id: str
     arguments: Mapping[str, Any]
     context: OpenAILLMContext | LLMContext
+    llm_specific_extra: Optional[Dict[str, Any]] = None
     run_llm: Optional[bool] = None
 
 
@@ -456,6 +460,7 @@ class LLMService(AIService):
                     tool_call_id=function_call.tool_call_id,
                     arguments=function_call.arguments,
                     context=function_call.context,
+                    llm_specific_extra=function_call.llm_specific_extra,
                 )
             )
 
@@ -580,6 +585,7 @@ class LLMService(AIService):
             function_name=runner_item.function_name,
             tool_call_id=runner_item.tool_call_id,
             arguments=runner_item.arguments,
+            llm_specific_extra=runner_item.llm_specific_extra,
             cancel_on_interruption=item.cancel_on_interruption,
         )