wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removing the extra examples inside the unified-format-function-calling folder

Browse Source
This commit is contained in:
Filipi Fuchter
2025-03-06 12:00:22 -03:00
parent 21443b9a08
commit 55b0797fd5
18 changed files with 0 additions and 745 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
examples/unified-format-function-calling/base_function_calling.py
View File

@@ -1,134 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
import sys
import aiohttp
from dotenv import load_dotenv
from loguru import logger
from runner import configure
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.ai_services import LLMService
from pipecat.services.cartesia import CartesiaTTSService
from pipecat.transports.services.daily import DailyParams, DailyTransport
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
load_dotenv(override=True)
async def start_fetch_weather(function_name, llm, context):
"""Push a frame to the LLM; this is handy when the LLM response might take a while."""
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await result_callback({"conditions": "nice", "temperature": "75"})
class WeatherBot:
"""Generic base class for setting up and running an LLM-powered bot."""
def __init__(self, llm: LLMService):
"""Initialize the base handler with a specific LLM."""
self.llm = llm
async def run(self):
"""Set up and start the processing pipeline."""
async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport(
room_url,
token,
"Respond bot",
DailyParams(
audio_out_enabled=True,
transcription_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady
)
# Register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
self.llm.register_function(
None, fetch_weather_from_api, start_callback=start_fetch_weather
)
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location"],
)
tools = ToolsSchema(standard_tools=[weather_function])
messages = [
{
"role": "system",
"content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation.",
},
{"role": "user", "content": " Start the conversation by introducing yourself."},
]
context = OpenAILLMContext(messages, tools)
context_aggregator = self.llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
self.llm,
tts,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
enable_metrics=True,
enable_usage_metrics=True,
report_only_initial_ttfb=True,
),
)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
await task.queue_frames([context_aggregator.user().get_context_frame()])
runner = PipelineRunner()
await runner.run(task)

126
examples/unified-format-function-calling/multimodal_base_function_calling.py
View File

@@ -1,126 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import sys
from typing import List
import aiohttp
from dotenv import load_dotenv
from loguru import logger
from runner import configure
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.ai_services import LLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
load_dotenv(override=True)
async def start_fetch_weather(function_name, llm, context):
"""Push a frame to the LLM; this is handy when the LLM response might take a while."""
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await result_callback({"conditions": "nice", "temperature": "75"})
class MultimodalWeatherBot:
"""Generic base class for setting up and running an LLM-powered bot."""
def __init__(self, llm: LLMService):
"""Initialize the base handler with a specific LLM."""
self.llm = llm
@staticmethod
def tools() -> ToolsSchema:
weather_function = FunctionSchema(
name="get_current_weather",
description="Get the current weather",
properties={
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"format": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "The temperature unit to use. Infer this from the user's location.",
},
},
required=["location"],
)
return ToolsSchema(standard_tools=[weather_function])
async def run(self):
"""Set up and start the processing pipeline."""
async with aiohttp.ClientSession() as session:
(room_url, token) = await configure(session)
transport = DailyTransport(
room_url,
token,
"Respond bot",
DailyParams(
audio_out_enabled=True,
vad_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
vad_audio_passthrough=True,
),
)
# Register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
self.llm.register_function(
None, fetch_weather_from_api, start_callback=start_fetch_weather
)
messages = [
{
"role": "system",
"content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation.",
},
{"role": "user", "content": " Start the conversation by introducing yourself."},
]
context = OpenAILLMContext(messages, MultimodalWeatherBot.tools())
context_aggregator = self.llm.create_context_aggregator(context)
pipeline = Pipeline(
[
transport.input(),
context_aggregator.user(),
self.llm,
transport.output(),
context_aggregator.assistant(),
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
allow_interruptions=True,
),
)
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):
await transport.capture_participant_transcription(participant["id"])
await task.queue_frames([context_aggregator.user().get_context_frame()])
runner = PipelineRunner()
await runner.run(task)

64
examples/unified-format-function-calling/runner.py
View File

@@ -1,64 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import os
from typing import Optional
import aiohttp
from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
async def configure(aiohttp_session: aiohttp.ClientSession):
(url, token, _) = await configure_with_args(aiohttp_session)
return (url, token)
async def configure_with_args(
aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
):
if not parser:
parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
parser.add_argument(
"-u", "--url", type=str, required=False, help="URL of the Daily room to join"
)
parser.add_argument(
"-k",
"--apikey",
type=str,
required=False,
help="Daily API Key (needed to create an owner token for the room)",
)
args, unknown = parser.parse_known_args()
url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
key = args.apikey or os.getenv("DAILY_API_KEY")
if not url:
raise Exception(
"No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
)
if not key:
raise Exception(
"No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
)
daily_rest_helper = DailyRESTHelper(
daily_api_key=key,
daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
aiohttp_session=aiohttp_session,
)
# Create a meeting token for the given room with an expiration 1 hour in
# the future.
expiry_time: float = 60 * 60
token = await daily_rest_helper.get_token(url, expiry_time)
return (url, token, args)

29
examples/unified-format-function-calling/standard-function-calling-anthropic.py
View File

@@ -1,29 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.anthropic import AnthropicLLMService
load_dotenv(override=True)
class AnthropicWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = AnthropicLLMService(
api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620"
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(AnthropicWeatherBot().run())

31
examples/unified-format-function-calling/standard-function-calling-azure.py
View File

@@ -1,31 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.azure import AzureLLMService
load_dotenv(override=True)
class AzureWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = AzureLLMService(
api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
model=os.getenv("AZURE_CHATGPT_MODEL"),
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(AzureWeatherBot().run())

27
examples/unified-format-function-calling/standard-function-calling-cerebras.py
View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.cerebras import CerebrasLLMService
load_dotenv(override=True)
class CerebrasWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = CerebrasLLMService(api_key=os.getenv("CEREBRAS_API_KEY"), model="llama-3.3-70b")
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(CerebrasWeatherBot().run())

27
examples/unified-format-function-calling/standard-function-calling-deepseek.py
View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.deepseek import DeepSeekLLMService
load_dotenv(override=True)
class DeepSeekWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = DeepSeekLLMService(api_key=os.getenv("DEEPSEEK_API_KEY"), model="deepseek-chat")
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(DeepSeekWeatherBot().run())

29
examples/unified-format-function-calling/standard-function-calling-fireworks.py
View File

@@ -1,29 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.fireworks import FireworksLLMService
load_dotenv(override=True)
class FireworksWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = FireworksLLMService(
api_key=os.getenv("FIREWORKS_API_KEY"),
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(FireworksWeatherBot().run())

38
examples/unified-format-function-calling/standard-function-calling-gemini-multimodal.py
View File

@@ -1,38 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from dotenv import load_dotenv
from multimodal_base_function_calling import MultimodalWeatherBot
from pipecat.adapters.schemas.tools_schema import AdapterType
from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
load_dotenv(override=True)
class GeminiMultimodalWeatherBot(MultimodalWeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
search_tool = {"google_search": {}}
tools_def = MultimodalWeatherBot.tools()
tools_def.custom_tools = {AdapterType.GEMINI: [search_tool]}
llm = GeminiMultimodalLiveLLMService(
api_key=os.getenv("GOOGLE_API_KEY"),
voice_id="Puck",
transcribe_user_audio=True,
transcribe_model_audio=True,
tools=tools_def,
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(GeminiMultimodalWeatherBot().run())

27
examples/unified-format-function-calling/standard-function-calling-gemini.py
View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.google import GoogleLLMService
load_dotenv(override=True)
class GeminiWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(GeminiWeatherBot().run())

27
examples/unified-format-function-calling/standard-function-calling-grok.py
View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.grok import GrokLLMService
load_dotenv(override=True)
class GrokWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = GrokLLMService(api_key=os.getenv("GROK_API_KEY"))
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(GrokWeatherBot().run())

27
examples/unified-format-function-calling/standard-function-calling-groq.py
View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.groq import GroqLLMService
load_dotenv(override=True)
class GroqWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = GroqLLMService(api_key=os.getenv("GROQ_API_KEY"), model="llama-3.3-70b-versatile")
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(GroqWeatherBot().run())

29
examples/unified-format-function-calling/standard-function-calling-nim.py
View File

@@ -1,29 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.nim import NimLLMService
load_dotenv(override=True)
class NimWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = NimLLMService(
api_key=os.getenv("NVIDIA_API_KEY"), model="meta/llama-3.3-70b-instruct"
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(NimWeatherBot().run())

43
examples/unified-format-function-calling/standard-function-calling-openai-realtime.py
View File

@@ -1,43 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from dotenv import load_dotenv
from multimodal_base_function_calling import MultimodalWeatherBot
from pipecat.services.openai_realtime_beta import (
InputAudioTranscription,
OpenAIRealtimeBetaLLMService,
SessionProperties,
TurnDetection,
)
load_dotenv(override=True)
class OpenAiRealTimeWeatherBot(MultimodalWeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
session_properties = SessionProperties(
input_audio_transcription=InputAudioTranscription(),
# Set openai TurnDetection parameters. Not setting this at all will turn it
# on by default
turn_detection=TurnDetection(silence_duration_ms=1000),
)
llm = OpenAIRealtimeBetaLLMService(
api_key=os.getenv("OPENAI_API_KEY"),
session_properties=session_properties,
start_audio_paused=False,
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(OpenAiRealTimeWeatherBot().run())

27
examples/unified-format-function-calling/standard-function-calling-openai.py
View File

@@ -1,27 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.openai import OpenAILLMService
load_dotenv(override=True)
class OpenAiWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(OpenAiWeatherBot().run())

29
examples/unified-format-function-calling/standard-function-calling-openrouter.py
View File

@@ -1,29 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.openrouter import OpenRouterLLMService
load_dotenv(override=True)
class OpenRouterWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = OpenRouterLLMService(
api_key=os.getenv("OPENROUTER_API_KEY"), model="openai/gpt-4o-2024-11-20"
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(OpenRouterWeatherBot().run())

30
examples/unified-format-function-calling/standard-function-calling-together.py
View File

@@ -1,30 +0,0 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
from base_function_calling import WeatherBot
from dotenv import load_dotenv
from pipecat.services.together import TogetherLLMService
load_dotenv(override=True)
class TogetherWeatherBot(WeatherBot):
"""Main class defining the LLM and passing it to the base handler."""
def __init__(self):
llm = TogetherLLMService(
api_key=os.getenv("TOGETHER_API_KEY"),
model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
)
super().__init__(llm)
if __name__ == "__main__":
asyncio.run(TogetherWeatherBot().run())

1
src/pipecat/services/azure.py
View File

@@ -579,7 +579,6 @@ class AzureTTSService(AzureBaseTTSService):
async def flush_audio(self):
logger.trace(f"{self}: flushing audio")
# TODO: check what we need to implement here ?
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"{self}: Generating TTS [{text}]")