diff --git a/examples/foundational/07n-interruptible-google-http.py b/examples/foundational/07n-interruptible-google-http.py index 2ef65e474..4a0382990 100644 --- a/examples/foundational/07n-interruptible-google-http.py +++ b/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 = [ diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index 73dd49e78..28b61c151 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/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 = [ diff --git a/examples/foundational/07s-interruptible-google-audio-in.py b/examples/foundational/07s-interruptible-google-audio-in.py index 67772e40d..90bff6062 100644 --- a/examples/foundational/07s-interruptible-google-audio-in.py +++ b/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( diff --git a/examples/foundational/49-thinking-functions.py b/examples/foundational/49-thinking-functions.py new file mode 100644 index 000000000..2b96e304d --- /dev/null +++ b/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() diff --git a/examples/foundational/49-thinking.py b/examples/foundational/49-thinking.py new file mode 100644 index 000000000..512163be4 --- /dev/null +++ b/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() diff --git a/src/pipecat/adapters/services/anthropic_adapter.py b/src/pipecat/adapters/services/anthropic_adapter.py index 75fa5899d..e111b34df 100644 --- a/src/pipecat/adapters/services/anthropic_adapter.py +++ b/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. diff --git a/src/pipecat/adapters/services/gemini_adapter.py b/src/pipecat/adapters/services/gemini_adapter.py index a4f70b1fa..fd91b818f 100644 --- a/src/pipecat/adapters/services/gemini_adapter.py +++ b/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 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 9cb969f28..2c3b802ec 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 69fc649ce..04f9c4ae9 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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) diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py index 93e0c37b4..0dae1ad6a 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/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 diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index a5c67e90e..662975056 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/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") diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 840b473b2..114df5f28 100644 --- a/src/pipecat/services/google/llm.py +++ b/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: diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index 6e5355263..71e31d9b6 100644 --- a/src/pipecat/services/llm_service.py +++ b/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, )