217f03b9cc
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.
223 lines
8.1 KiB
Python
223 lines
8.1 KiB
Python
#
|
||
# 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()
|