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

Merge pull request #1667 from pipecat-ai/aleix/function-call-single-parameter

Browse Source 
function call single parameter
This commit is contained in:
Aleix Conchillo Flaqué
2025-04-25 13:53:10 -07:00
committed by GitHub
parent 1d863ee7de 9cad6dfce9
commit 029ef4f8c2
39 changed files with 370 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGELOG.md
View File

@@ -25,6 +25,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- Function calls now receive a single parameter `FunctionCallParams` instead of
`(function_name, tool_call_id, args, llm, context, result_callback)` which is
now deprecated.
- Changed the user aggregator timeout for late transcriptions from 1.0s to 0.5s
(`LLMUserAggregatorParams.aggregation_timeout`). Sometimes, the STT services
might give us more than one transcription which could come after the user
@@ -52,6 +56,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Deprecated
- Function calls with parameters `(function_name, tool_call_id, args, llm,
context, result_callback)` are deprectated, use a single `FunctionCallParams`
parameter instead.
- `TransportParams.camera_*` parameters are now deprecated, use
`TransportParams.video_*` instead.

7
examples/foundational/14-function-calling.py
View File

@@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14a-function-calling-anthropic.py
View File

@@ -20,6 +20,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -27,9 +28,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
location = arguments["location"]
await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_weather(params: FunctionCallParams):
location = params.arguments["location"]
await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

21
examples/foundational/14b-function-calling-anthropic-video.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -29,23 +30,23 @@ load_dotenv(override=True)
# Global variable to store the peer connection ID
webrtc_peer_id = None
webrtc_peer_id = ""
async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
location = arguments["location"]
await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_weather(params: FunctionCallParams):
location = params.arguments["location"]
await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
question = arguments["question"]
async def get_image(params: FunctionCallParams):
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={webrtc_peer_id}, question={question}")
# Request the image frame
await llm.request_image_frame(
await params.llm.request_image_frame(
user_id=webrtc_peer_id,
function_name=function_name,
tool_call_id=tool_call_id,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
text_content=question,
)
@@ -53,7 +54,7 @@ async def get_image(function_name, tool_call_id, arguments, llm, context, result
await asyncio.sleep(0.5)
# Return a result to complete the function call
await result_callback(
await params.result_callback(
f"I've captured an image from your camera and I'm analyzing what you asked about: {question}"
)

7
examples/foundational/14c-function-calling-together.py
View File

@@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.together.llm import TogetherLLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

21
examples/foundational/14d-function-calling-video.py
View File

@@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -29,23 +30,23 @@ load_dotenv(override=True)
# Global variable to store the peer connection ID
webrtc_peer_id = None
webrtc_peer_id = ""
async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
location = arguments["location"]
await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_weather(params: FunctionCallParams):
location = params.arguments["location"]
await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
question = arguments["question"]
async def get_image(params: FunctionCallParams):
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={webrtc_peer_id}, question={question}")
# Request the image frame
await llm.request_image_frame(
await params.llm.request_image_frame(
user_id=webrtc_peer_id,
function_name=function_name,
tool_call_id=tool_call_id,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
text_content=question,
)
@@ -53,7 +54,7 @@ async def get_image(function_name, tool_call_id, arguments, llm, context, result
await asyncio.sleep(0.5)
# Return a result to complete the function call
await result_callback(
await params.result_callback(
f"I've captured an image from your camera and I'm analyzing what you asked about: {question}"
)

23
examples/foundational/14e-function-calling-gemini.py
View File

@@ -22,6 +22,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
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 TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -30,24 +31,24 @@ load_dotenv(override=True)
# Global variable to store the peer connection ID
webrtc_peer_id = None
webrtc_peer_id = ""
async def get_weather(function_name, tool_call_id, arguments, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
location = arguments["location"]
await result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_weather(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
location = params.arguments["location"]
await params.result_callback(f"The weather in {location} is currently 72 degrees and sunny.")
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
question = arguments["question"]
async def get_image(params: FunctionCallParams):
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={webrtc_peer_id}, question={question}")
# Request the image frame
await llm.request_image_frame(
await params.llm.request_image_frame(
user_id=webrtc_peer_id,
function_name=function_name,
tool_call_id=tool_call_id,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
text_content=question,
)
@@ -55,7 +56,7 @@ async def get_image(function_name, tool_call_id, arguments, llm, context, result
await asyncio.sleep(0.5)
# Return a result to complete the function call
await result_callback(
await params.result_callback(
f"I've captured an image from your camera and I'm analyzing what you asked about: {question}"
)

7
examples/foundational/14f-function-calling-groq.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.groq.llm import GroqLLMService
from pipecat.services.groq.stt import GroqSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

5
examples/foundational/14g-function-calling-grok.py
View File

@@ -20,6 +20,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.grok.llm import GrokLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -27,8 +28,8 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14h-function-calling-azure.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.azure.llm import AzureLLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14i-function-calling-fireworks.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.fireworks.llm import FireworksLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14j-function-calling-nim.py
View File

@@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.nim.llm import NimLLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14k-function-calling-cerebras.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.cerebras.llm import CerebrasLLMService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14l-function-calling-deepseek.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.deepseek.llm import DeepSeekLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14m-function-calling-openrouter.py
View File

@@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.azure.tts import AzureTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openrouter.llm import OpenRouterLLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14o-function-calling-gemini-openai-format.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.google.llm_openai import GoogleLLMOpenAIBetaService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14p-function-calling-gemini-vertex-ai.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.google.llm_vertex import GoogleVertexLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/14q-function-calling-qwen.py
View File

@@ -20,6 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.qwen.llm import QwenLLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -28,9 +29,9 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/15-switch-voices.py
View File

@@ -20,6 +20,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -31,10 +32,10 @@ load_dotenv(override=True)
current_voice = "News Lady"
async def switch_voice(function_name, tool_call_id, args, llm, context, result_callback):
async def switch_voice(params: FunctionCallParams):
global current_voice
current_voice = args["voice"]
await result_callback(
current_voice = params.arguments["voice"]
await params.result_callback(
{
"voice": f"You are now using your {current_voice} voice. Your responses should now be as if you were a {current_voice}."
}

9
examples/foundational/15a-switch-languages.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -32,10 +33,12 @@ load_dotenv(override=True)
current_language = "English"
async def switch_language(function_name, tool_call_id, args, llm, context, result_callback):
async def switch_language(params: FunctionCallParams):
global current_language
current_language = args["language"]
await result_callback({"voice": f"Your answers from now on should be in {current_language}."})
current_language = params.arguments["language"]
await params.result_callback(
{"voice": f"Your answers from now on should be in {current_language}."}
)
async def english_filter(frame) -> bool:

9
examples/foundational/19-openai-realtime-beta.py
View File

@@ -19,6 +19,7 @@ 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.llm_service import FunctionCallParams
from pipecat.services.openai_realtime_beta import (
InputAudioNoiseReduction,
InputAudioTranscription,
@@ -33,13 +34,13 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)

9
examples/foundational/19a-azure-realtime-beta.py
View File

@@ -19,6 +19,7 @@ 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.llm_service import FunctionCallParams
from pipecat.services.openai_realtime_beta import (
AzureRealtimeBetaLLMService,
InputAudioTranscription,
@@ -31,13 +32,13 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)

37
examples/foundational/20a-persistent-context-openai.py
View File

@@ -23,6 +23,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -35,21 +36,19 @@ BASE_FILENAME = "/tmp/pipecat_conversation_"
tts = None
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_saved_conversation_filenames(
function_name, tool_call_id, args, llm, context, result_callback
):
async def get_saved_conversation_filenames(params: FunctionCallParams):
# Construct the full pattern including the BASE_FILENAME
full_pattern = f"{BASE_FILENAME}*.json"
@@ -57,37 +56,39 @@ async def get_saved_conversation_filenames(
matching_files = glob.glob(full_pattern)
logger.debug(f"matching files: {matching_files}")
await result_callback({"filenames": matching_files})
await params.result_callback({"filenames": matching_files})
async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def save_conversation(params: FunctionCallParams):
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}")
logger.debug(
f"writing conversation to {filename}\n{json.dumps(params.context.messages, indent=4)}"
)
try:
with open(filename, "w") as file:
messages = context.get_messages_for_persistent_storage()
messages = params.context.get_messages_for_persistent_storage()
# remove the last message, which is the instruction we just gave to save the conversation
messages.pop()
json.dump(messages, file, indent=2)
await result_callback({"success": True})
await params.result_callback({"success": True})
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def load_conversation(params: FunctionCallParams):
global tts
filename = args["filename"]
filename = params.arguments["filename"]
logger.debug(f"loading conversation from {filename}")
try:
with open(filename, "r") as file:
context.set_messages(json.load(file))
params.context.set_messages(json.load(file))
logger.debug(
f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}"
f"loaded conversation from {filename}\n{json.dumps(params.context.messages, indent=4)}"
)
await tts.say("Ok, I've loaded that conversation.")
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
messages = [

52
examples/foundational/20b-persistent-context-openai-realtime.py
View File

@@ -23,6 +23,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
)
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai_realtime_beta import (
InputAudioTranscription,
OpenAIRealtimeBetaLLMService,
@@ -38,21 +39,19 @@ load_dotenv(override=True)
BASE_FILENAME = "/tmp/pipecat_conversation_"
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_saved_conversation_filenames(
function_name, tool_call_id, args, llm, context, result_callback
):
async def get_saved_conversation_filenames(params: FunctionCallParams):
# Construct the full pattern including the BASE_FILENAME
full_pattern = f"{BASE_FILENAME}*.json"
@@ -60,48 +59,37 @@ async def get_saved_conversation_filenames(
matching_files = glob.glob(full_pattern)
logger.debug(f"matching files: {matching_files}")
await result_callback({"filenames": matching_files})
await params.result_callback({"filenames": matching_files})
# async def get_saved_conversation_filenames(
# function_name, tool_call_id, args, llm, context, result_callback
# ):
# pattern = re.compile(re.escape(BASE_FILENAME) + "\\d{8}_\\d{6}\\.json$")
# matching_files = []
# for filename in os.listdir("."):
# if pattern.match(filename):
# matching_files.append(filename)
# await result_callback({"filenames": matching_files})
async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def save_conversation(params: FunctionCallParams):
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}")
logger.debug(
f"writing conversation to {filename}\n{json.dumps(params.context.messages, indent=4)}"
)
try:
with open(filename, "w") as file:
messages = context.get_messages_for_persistent_storage()
messages = params.context.get_messages_for_persistent_storage()
# remove the last message, which is the instruction we just gave to save the conversation
messages.pop()
json.dump(messages, file, indent=2)
await result_callback({"success": True})
await params.result_callback({"success": True})
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def load_conversation(params: FunctionCallParams):
async def _reset():
filename = args["filename"]
filename = params.arguments["filename"]
logger.debug(f"loading conversation from {filename}")
try:
with open(filename, "r") as file:
context.set_messages(json.load(file))
await llm.reset_conversation()
await llm._create_response()
params.context.set_messages(json.load(file))
await params.llm.reset_conversation()
await params.llm._create_response()
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
asyncio.create_task(_reset())

37
examples/foundational/20c-persistent-context-anthropic.py
View File

@@ -24,6 +24,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -35,21 +36,19 @@ BASE_FILENAME = "/tmp/pipecat_conversation_"
tts = None
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_saved_conversation_filenames(
function_name, tool_call_id, args, llm, context, result_callback
):
async def get_saved_conversation_filenames(params: FunctionCallParams):
# Construct the full pattern including the BASE_FILENAME
full_pattern = f"{BASE_FILENAME}*.json"
@@ -57,38 +56,40 @@ async def get_saved_conversation_filenames(
matching_files = glob.glob(full_pattern)
logger.debug(f"matching files: {matching_files}")
await result_callback({"filenames": matching_files})
await params.result_callback({"filenames": matching_files})
async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def save_conversation(params: FunctionCallParams):
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(f"writing conversation to {filename}\n{json.dumps(context.messages, indent=4)}")
logger.debug(
f"writing conversation to {filename}\n{json.dumps(params.context.messages, indent=4)}"
)
try:
with open(filename, "w") as file:
# todo: extract 'system' into the first message in the list
messages = context.get_messages_for_persistent_storage()
messages = params.context.get_messages_for_persistent_storage()
# remove the last message, which is the instruction we just gave to save the conversation
messages.pop()
json.dump(messages, file, indent=2)
await result_callback({"success": True})
await params.result_callback({"success": True})
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def load_conversation(params: FunctionCallParams):
global tts
filename = args["filename"]
filename = params.arguments["filename"]
logger.debug(f"loading conversation from {filename}")
try:
with open(filename, "r") as file:
context.set_messages(json.load(file))
params.context.set_messages(json.load(file))
logger.debug(
f"loaded conversation from {filename}\n{json.dumps(context.messages, indent=4)}"
f"loaded conversation from {filename}\n{json.dumps(params.context.messages, indent=4)}"
)
await tts.say("Ok, I've loaded that conversation.")
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
# Test message munging ...

47
examples/foundational/20d-persistent-context-gemini.py
View File

@@ -24,6 +24,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
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 TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -36,37 +37,35 @@ video_participant_id = None
BASE_FILENAME = "/tmp/pipecat_conversation_"
tts = None
webrtc_peer_id = None
webrtc_peer_id = ""
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
question = arguments["question"]
async def get_image(params: FunctionCallParams):
question = params.arguments["question"]
logger.debug(f"Requesting image with user_id={webrtc_peer_id}, question={question}")
# Request the image frame
await llm.request_image_frame(
await params.llm.request_image_frame(
user_id=webrtc_peer_id,
function_name=function_name,
tool_call_id=tool_call_id,
function_name=params.function_name,
tool_call_id=params.tool_call_id,
text_content=question,
)
async def get_saved_conversation_filenames(
function_name, tool_call_id, args, llm, context, result_callback
):
async def get_saved_conversation_filenames(params: FunctionCallParams):
# Construct the full pattern including the BASE_FILENAME
full_pattern = f"{BASE_FILENAME}*.json"
@@ -74,43 +73,43 @@ async def get_saved_conversation_filenames(
matching_files = glob.glob(full_pattern)
logger.debug(f"matching files: {matching_files}")
await result_callback({"filenames": matching_files})
await params.result_callback({"filenames": matching_files})
async def save_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def save_conversation(params: FunctionCallParams):
timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
filename = f"{BASE_FILENAME}{timestamp}.json"
logger.debug(
f"writing conversation to {filename}\n{json.dumps(context.get_messages_for_logging(), indent=4)}"
f"writing conversation to {filename}\n{json.dumps(params.context.get_messages_for_logging(), indent=4)}"
)
try:
with open(filename, "w") as file:
# todo: extract 'system' into the first message in the list
messages = context.get_messages_for_persistent_storage()
messages = params.context.get_messages_for_persistent_storage()
# remove the last message (the instruction to save the context)
messages.pop()
json.dump(messages, file, indent=2)
await result_callback({"success": True})
await params.result_callback({"success": True})
except Exception as e:
logger.debug(f"error saving conversation: {e}")
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
async def load_conversation(params: FunctionCallParams):
global tts
filename = args["filename"]
filename = params.arguments["filename"]
logger.debug(f"loading conversation from {filename}")
try:
with open(filename, "r") as file:
context.set_messages(json.load(file))
await result_callback(
params.context.set_messages(json.load(file))
await params.result_callback(
{
"success": True,
"message": "The most recent conversation has been loaded. Awaiting further instructions.",
}
)
except Exception as e:
await result_callback({"success": False, "error": str(e)})
await params.result_callback({"success": False, "error": str(e)})
# Test message munging ...

7
examples/foundational/22b-natural-conversation-proposal.py
View File

@@ -44,6 +44,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.processors.user_idle_processor import UserIdleProcessor
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier
@@ -196,9 +197,9 @@ class OutputGate(FrameProcessor):
break
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

7
examples/foundational/22c-natural-conversation-mixed-llms.py
View File

@@ -45,6 +45,7 @@ from pipecat.processors.user_idle_processor import UserIdleProcessor
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.sync.base_notifier import BaseNotifier
from pipecat.sync.event_notifier import EventNotifier
@@ -400,9 +401,9 @@ class OutputGate(FrameProcessor):
break
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
await llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
async def fetch_weather_from_api(params: FunctionCallParams):
await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

5
examples/foundational/24-stt-mute-filter.py
View File

@@ -21,6 +21,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.stt_mute_filter import STTMuteConfig, STTMuteFilter, STTMuteStrategy
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.deepgram.tts import DeepgramTTSService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -29,12 +30,12 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
async def fetch_weather_from_api(params: FunctionCallParams):
# Add a delay to test interruption during function calls
logger.info("Weather API call starting...")
await asyncio.sleep(5) # 5-second delay
logger.info("Weather API call completed")
await result_callback({"conditions": "nice", "temperature": "75"})
await params.result_callback({"conditions": "nice", "temperature": "75"})
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

9
examples/foundational/26b-gemini-multimodal-live-function-calling.py
View File

@@ -20,6 +20,7 @@ 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.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -27,13 +28,13 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
temperature = 75 if args["format"] == "fahrenheit" else 24
await result_callback(
async def fetch_weather_from_api(params: FunctionCallParams):
temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
await params.result_callback(
{
"conditions": "nice",
"temperature": temperature,
"format": args["format"],
"format": params.arguments["format"],
"timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
}
)

13
examples/foundational/33-gemini-rag.py
View File

@@ -64,6 +64,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
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 TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -107,11 +108,9 @@ Here is the knowledge base you have access to:
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
async def query_knowledge_base(
function_name, tool_call_id, arguments, llm, context, result_callback
):
async def query_knowledge_base(params: FunctionCallParams):
"""Query the knowledge base for the answer to the question."""
logger.info(f"Querying knowledge base for question: {arguments['question']}")
logger.info(f"Querying knowledge base for question: {params.arguments['question']}")
client = genai.GenerativeModel(
model_name=RAG_MODEL,
system_instruction=RAG_PROMPT,
@@ -122,11 +121,11 @@ async def query_knowledge_base(
)
# for our case, the first two messages are the instructions and the user message
# so we remove them.
conversation_turns = context.messages[2:]
conversation_turns = params.context.messages[2:]
# convert to standard messages
messages = []
for turn in conversation_turns:
messages.extend(context.to_standard_messages(turn))
messages.extend(params.context.to_standard_messages(turn))
def _is_tool_call(turn):
if turn.get("role", None) == "tool":
@@ -150,7 +149,7 @@ async def query_knowledge_base(
end = time.perf_counter()
logger.info(f"Time taken: {end - start:.2f} seconds")
logger.info(response.text)
await result_callback(response.text)
await params.result_callback(response.text)
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

5
examples/foundational/36-user-email-gathering.py
View File

@@ -18,6 +18,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.rime.tts import RimeHttpTTSService
from pipecat.transports.base_transport import TransportParams
@@ -27,8 +28,8 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
load_dotenv(override=True)
async def store_user_emails(function_name, tool_call_id, args, llm, context, result_callback):
print(f"User emails: {args}")
async def store_user_emails(params: FunctionCallParams):
print(f"User emails: {params.arguments}")
async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):

69
examples/patient-intake/bot.py
View File

@@ -23,6 +23,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFr
from pipecat.processors.frame_processor import FrameDirection
from pipecat.processors.logger import FrameLogger
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMContext, OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -84,11 +85,9 @@ class IntakeProcessor:
]
)
async def verify_birthday(
self, function_name, tool_call_id, args, llm, context, result_callback
):
if args["birthday"] == "1983-01-01":
context.set_tools(
async def verify_birthday(self, params: FunctionCallParams):
if params.arguments["birthday"] == "1983-01-01":
params.context.set_tools(
[
{
"type": "function",
@@ -124,7 +123,7 @@ class IntakeProcessor:
# await llm.push_frame(sounds["ding2.wav"], FrameDirection.DOWNSTREAM)
# We don't need the function call in the context, so just return a new
# system message and let the framework re-prompt
await result_callback(
await params.result_callback(
[
{
"role": "system",
@@ -134,7 +133,7 @@ class IntakeProcessor:
)
else:
# The user provided an incorrect birthday; ask them to try again
await result_callback(
await params.result_callback(
[
{
"role": "system",
@@ -143,12 +142,10 @@ class IntakeProcessor:
]
)
async def list_prescriptions(
self, function_name, tool_call_id, args, llm, context, result_callback
):
async def list_prescriptions(self, params: FunctionCallParams):
print(f"!!! doing start prescriptions")
# Move on to allergies
context.set_tools(
params.context.set_tools(
[
{
"type": "function",
@@ -176,23 +173,23 @@ class IntakeProcessor:
}
]
)
context.add_message(
params.context.add_message(
{
"role": "system",
"content": "Next, ask the user if they have any allergies. Once they have listed their allergies or confirmed they don't have any, call the list_allergies function.",
}
)
print(f"!!! about to await llm process frame in start prescrpitions")
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await params.llm.queue_frame(
OpenAILLMContextFrame(params.context), FrameDirection.DOWNSTREAM
)
print(f"!!! past await process frame in start prescriptions")
await self.save_data(args, result_callback)
await self.save_data(params.arguments, params.result_callback)
async def list_allergies(
self, function_name, tool_call_id, args, llm, context, result_callback
):
async def list_allergies(self, params: FunctionCallParams):
print("!!! doing list allergies")
# Move on to conditions
context.set_tools(
params.context.set_tools(
[
{
"type": "function",
@@ -220,21 +217,21 @@ class IntakeProcessor:
},
]
)
context.add_message(
params.context.add_message(
{
"role": "system",
"content": "Now ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function.",
}
)
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await self.save_data(args, result_callback)
await params.llm.queue_frame(
OpenAILLMContextFrame(params.context), FrameDirection.DOWNSTREAM
)
await self.save_data(params.arguments, params.result_callback)
async def list_conditions(
self, function_name, tool_call_id, args, llm, context, result_callback
):
async def list_conditions(self, params: FunctionCallParams):
print("!!! doing start conditions")
# Move on to visit reasons
context.set_tools(
params.context.set_tools(
[
{
"type": "function",
@@ -262,26 +259,28 @@ class IntakeProcessor:
}
]
)
context.add_message(
params.context.add_message(
{
"role": "system",
"content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function.",
}
)
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await self.save_data(args, result_callback)
await params.llm.queue_frame(
OpenAILLMContextFrame(params.context), FrameDirection.DOWNSTREAM
)
await self.save_data(params.arguments, params.result_callback)
async def list_visit_reasons(
self, function_name, tool_call_id, args, llm, context, result_callback
):
async def list_visit_reasons(self, params: FunctionCallParams):
print("!!! doing start visit reasons")
# move to finish call
context.set_tools([])
context.add_message(
params.context.set_tools([])
params.context.add_message(
{"role": "system", "content": "Now, thank the user and end the conversation."}
)
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await self.save_data(args, result_callback)
await params.llm.queue_frame(
OpenAILLMContextFrame(params.context), FrameDirection.DOWNSTREAM
)
await self.save_data(params.arguments, params.result_callback)
async def save_data(self, args, result_callback):
logger.info(f"!!! Saving data: {args}")

24
examples/phone-chatbot/call_transfer.py
View File

@@ -29,7 +29,7 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.filters.function_filter import FunctionFilter
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallParams, LLMService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyDialinSettings, DailyParams, DailyTransport
@@ -220,12 +220,7 @@ async def main(
async def terminate_call(
task: PipelineTask, # Pipeline task reference
function_name,
tool_call_id,
args,
llm: LLMService,
context: OpenAILLMContext,
result_callback,
params: FunctionCallParams,
):
"""Function the bot can call to terminate the call."""
# Create a message to add
@@ -237,16 +232,9 @@ async def main(
await task.queue_frames([LLMMessagesFrame(messages)])
# Then end the call
await llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
async def dial_operator(
function_name: str,
tool_call_id: str,
args: dict,
llm: LLMService,
context: dict,
result_callback: callable,
):
async def dial_operator(params: FunctionCallParams):
"""Function the bot can call to dial an operator."""
dialout_setting = session_manager.call_flow_state.get_current_dialout_setting()
if call_config_manager.get_transfer_mode() == "dialout":
@@ -306,9 +294,7 @@ async def main(
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
# Register functions with the LLM
llm.register_function(
"terminate_call", lambda *args, **kwargs: terminate_call(task, *args, **kwargs)
)
llm.register_function("terminate_call", lambda params: terminate_call(task, params))
llm.register_function("dial_operator", dial_operator)
# Initialize LLM context and aggregator

8
examples/phone-chatbot/simple_dialin.py
View File

@@ -22,7 +22,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyDialinSettings, DailyParams, DailyTransport
@@ -99,16 +99,14 @@ async def main(
# ------------ FUNCTION DEFINITIONS ------------
async def terminate_call(
function_name, tool_call_id, args, llm: LLMService, context, result_callback
):
async def terminate_call(params: FunctionCallParams):
"""Function the bot can call to terminate the call upon completion of a voicemail message."""
if session_manager:
# Mark that the call was terminated by the bot
session_manager.call_flow_state.set_call_terminated()
# Then end the call
await llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
# Define function schemas for tools
terminate_call_function = FunctionSchema(

8
examples/phone-chatbot/simple_dialout.py
View File

@@ -22,7 +22,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.llm_service import LLMService
from pipecat.services.llm_service import FunctionCallParams
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -77,11 +77,9 @@ async def main(
# ------------ FUNCTION DEFINITIONS ------------
async def terminate_call(
function_name, tool_call_id, args, llm: LLMService, context, result_callback
):
async def terminate_call(params: FunctionCallParams):
"""Function the bot can call to terminate the call upon completion of a voicemail message."""
await llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
# Define function schemas for tools
terminate_call_function = FunctionSchema(

37
examples/phone-chatbot/voicemail_detection.py
View File

@@ -32,7 +32,7 @@ from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.google.google import GoogleLLMContext
from pipecat.services.google.llm import GoogleLLMService
from pipecat.services.llm_service import LLMService # Base LLM service class
from pipecat.services.llm_service import FunctionCallParams
from pipecat.transports.services.daily import (
DailyParams,
DailyTransport,
@@ -98,35 +98,19 @@ class FunctionHandlers:
self.session_manager = session_manager
self.prompt = None # Can be set externally
async def voicemail_response(
self,
function_name,
tool_call_id,
args,
llm: LLMService,
context,
result_callback,
):
async def voicemail_response(self, params: FunctionCallParams):
"""Function the bot can call to leave a voicemail message."""
message = """You are Chatbot leaving a voicemail message. Say EXACTLY this message and then terminate the call:
'Hello, this is a message for Pipecat example user. This is Chatbot. Please call back on 123-456-7891. Thank you.'"""
await result_callback(message)
await params.result_callback(message)
async def human_conversation(
self,
function_name,
tool_call_id,
args,
llm: LLMService,
context,
result_callback,
):
async def human_conversation(self, params: FunctionCallParams):
"""Function called when bot detects it's talking to a human."""
# Update state to indicate human was detected
self.session_manager.call_flow_state.set_human_detected()
await llm.push_frame(StopTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.push_frame(StopTaskFrame(), FrameDirection.UPSTREAM)
# ------------ MAIN FUNCTION ------------
@@ -182,12 +166,7 @@ async def main(
# ------------ FUNCTION DEFINITIONS ------------
async def terminate_call(
function_name,
tool_call_id,
args,
llm: LLMService,
context,
result_callback,
params: FunctionCallParams,
session_manager=None,
):
"""Function the bot can call to terminate the call."""
@@ -195,7 +174,7 @@ async def main(
# Set call terminated flag in the session manager
session_manager.call_flow_state.set_call_terminated()
await llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
await params.llm.queue_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
# ------------ VOICEMAIL DETECTION PHASE SETUP ------------
@@ -275,7 +254,7 @@ async def main(
"switch_to_human_conversation", handlers.human_conversation
)
voicemail_detection_llm.register_function(
"terminate_call", functools.partial(terminate_call, session_manager=session_manager)
"terminate_call", lambda params: terminate_call(params, session_manager)
)
# Set up audio collector for handling audio input

20
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -251,7 +251,7 @@ class RTVIBotReady(BaseModel):
class RTVILLMFunctionCallMessageData(BaseModel):
function_name: str
tool_call_id: str
args: dict
arguments: Mapping[str, Any]
class RTVILLMFunctionCallMessage(BaseModel):
@@ -662,13 +662,12 @@ class RTVIProcessor(FrameProcessor):
self,
function_name: str,
tool_call_id: str,
arguments: dict,
llm: FrameProcessor,
context: OpenAILLMContext,
result_callback,
arguments: Mapping[str, Any],
):
fn = RTVILLMFunctionCallMessageData(
function_name=function_name, tool_call_id=tool_call_id, args=arguments
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
)
message = RTVILLMFunctionCallMessage(data=fn)
await self._push_transport_message(message, exclude_none=False)
@@ -676,6 +675,15 @@ class RTVIProcessor(FrameProcessor):
async def handle_function_call_start(
self, function_name: str, llm: FrameProcessor, context: OpenAILLMContext
):
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Function `RTVIProcessor.handle_function_call_start()` is deprecated, use `RTVIProcessor.handle_function_call()` instead.",
DeprecationWarning,
)
fn = RTVILLMFunctionCallStartMessageData(function_name=function_name)
message = RTVILLMFunctionCallStartMessage(data=fn)
await self._push_transport_message(message, exclude_none=False)

94
src/pipecat/services/llm_service.py
View File

@@ -5,8 +5,9 @@
#
import asyncio
import inspect
from dataclasses import dataclass
from typing import Any, Optional, Set, Tuple, Type
from typing import Any, Awaitable, Callable, Mapping, Optional, Protocol, Set, Tuple, Type
from loguru import logger
@@ -17,6 +18,7 @@ from pipecat.frames.frames import (
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallResultProperties,
StartInterruptionFrame,
UserImageRequestFrame,
)
@@ -28,14 +30,55 @@ from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_service import AIService
# Type alias for a callable that handles LLM function calls.
FunctionCallHandler = Callable[["FunctionCallParams"], Awaitable[None]]
# Type alias for a callback function that handles the result of an LLM function call.
class FunctionCallResultCallback(Protocol):
async def __call__(
self, result: Any, *, properties: Optional[FunctionCallResultProperties] = None
) -> None: ...
@dataclass
class FunctionEntry:
class FunctionCallEntry:
"""Represents an internal entry for a function call.
Attributes:
function_name (Optional[str]): The name of the function.
handler (FunctionCallHandler): The handler for processing function call parameters.
cancel_on_interruption (bool): Flag indicating whether to cancel the call on interruption.
"""
function_name: Optional[str]
callback: Any # TODO(aleix): add proper typing.
handler: FunctionCallHandler
cancel_on_interruption: bool
@dataclass
class FunctionCallParams:
"""Parameters for a function call.
Attributes:
function_name (str): The name of the function being called.
arguments (Mapping[str, Any]): The arguments for the function.
tool_call_id (str): A unique identifier for the function call.
llm (LLMService): The LLMService instance being used.
context (OpenAILLMContext): The LLM context.
result_callback (FunctionCallResultCallback): Callback to handle the result of the function call.
"""
function_name: str
tool_call_id: str
arguments: Mapping[str, Any]
llm: "LLMService"
context: OpenAILLMContext
result_callback: FunctionCallResultCallback
class LLMService(AIService):
"""This class is a no-op but serves as a base class for LLM services."""
@@ -78,16 +121,16 @@ class LLMService(AIService):
def register_function(
self,
function_name: Optional[str],
callback: Any,
handler: Any,
start_callback=None,
*,
cancel_on_interruption: bool = False,
):
# Registering a function with the function_name set to None will run that callback
# for all functions
self._functions[function_name] = FunctionEntry(
# Registering a function with the function_name set to None will run
# that handler for all functions
self._functions[function_name] = FunctionCallEntry(
function_name=function_name,
callback=callback,
handler=handler,
cancel_on_interruption=cancel_on_interruption,
)
@@ -120,7 +163,7 @@ class LLMService(AIService):
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: str,
arguments: Mapping[str, Any],
run_llm: bool = True,
):
if not function_name in self._functions.keys() and not None in self._functions.keys():
@@ -163,7 +206,7 @@ class LLMService(AIService):
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: str,
arguments: Mapping[str, Any],
run_llm: bool = True,
):
if function_name in self._functions.keys():
@@ -202,7 +245,9 @@ class LLMService(AIService):
await self.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
# Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
async def function_call_result_callback(result, *, properties=None):
async def function_call_result_callback(
result: Any, *, properties: Optional[FunctionCallResultProperties] = None
):
result_frame_downstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
@@ -221,9 +266,30 @@ class LLMService(AIService):
await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
await entry.callback(
function_name, tool_call_id, arguments, self, context, function_call_result_callback
)
signature = inspect.signature(entry.handler)
if len(signature.parameters) > 1:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Function calls with parameters `(function_name, tool_call_id, arguments, llm, context, result_callback)` are deprecated, use a single `FunctionCallParams` parameter instead.",
DeprecationWarning,
)
await entry.handler(
function_name, tool_call_id, arguments, self, context, function_call_result_callback
)
else:
params = FunctionCallParams(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
llm=self,
context=context,
result_callback=function_call_result_callback,
)
await entry.handler(params)
async def _cancel_function_call(self, function_name: str):
cancelled_tasks = set()