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

Merge pull request #1394 from pipecat-ai/aleix/function-calls-as-tasks

Browse Source 
function calls as tasks
This commit is contained in:
Aleix Conchillo Flaqué
2025-03-20 09:34:37 -07:00
committed by GitHub
parent fc06306efd e0c3f6ad83
commit c51291190b
40 changed files with 729 additions and 1081 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGELOG.md
View File

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- When registering a function call it is now possible to indicate if you want
the function call to be cancelled if there's a user interruption via
`cancel_on_interruption` (defaults to False). This is now possible because
function calls are executed concurrently.
- Added support for detecting idle pipelines. By default, if no activity has
been detected during 5 minutes, the `PipelineTask` will be automatically
cancelled. It is possible to override this behavior by passing
@@ -40,6 +45,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
being aggregated. A text aggregator can be passed via `text_aggregator` to the
TTS service.
- Added new `sample_rate` constructor parameter to `TavusVideoService` to allow
changing the output sample rate.
- Added new `UltravoxSTTService`.
(see https://github.com/fixie-ai/ultravox)
@@ -120,6 +128,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Changed
- Function calls are now executed in tasks. This means that the pipeline will
not be blocked while the function call is being executed.
- ⚠️ `PipelineTask` will now be automatically cancelled if no bot activity is
happening in the pipeline. There are a few settings to configure this
behavior, see `PipelineTask` documentation for more details.
@@ -140,6 +151,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Deprecated
- Passing a `start_callback` to `LLMService.register_function()` is now
deprecated, simply move the code from the start callback to the function call.
- `TTSService` parameter `text_filter` is now deprecated, use `text_filters`
instead which is now a list. This allows passing multiple filters that will be
executed in order.
@@ -162,6 +176,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Fixed
- Fixed an assistant aggregator issue that could cause assistant text to be
split into multiple chunks during function calls.
- Fixed an assistant aggregator issue that was causing assistant text to not be
added to the context during function calls. This could lead to duplications.
- Fixed a `SegmentedSTTService` issue that was causing audio to be sent
prematurely to the STT service. Instead of analyzing the volume in this
service we rely on VAD events which use both VAD and volume.
@@ -1978,7 +1998,7 @@ async def on_connected(processor):
completed. If a task is never ran `has_finished()` will return False.
- `PipelineRunner` now supports SIGTERM. If received, the runner will be
canceled.
cancelled.
### Fixed

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

@@ -30,13 +30,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -62,9 +57,10 @@ async def main():
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -39,7 +39,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
question = arguments["question"]
await llm.request_image_frame(user_id=video_participant_id, text_content=question)
await llm.request_image_frame(
user_id=video_participant_id,
function_name=function_name,
tool_call_id=tool_call_id,
text_content=question,
)
async def main():

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -66,9 +61,9 @@ async def main():
api_key=os.getenv("TOGETHER_API_KEY"),
model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
)
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -39,7 +39,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}")
question = arguments["question"]
await llm.request_image_frame(user_id=video_participant_id, text_content=question)
await llm.request_image_frame(
user_id=video_participant_id,
function_name=function_name,
tool_call_id=tool_call_id,
text_content=question,
)
async def main():
@@ -141,7 +146,7 @@ indicate you should use the get_image tool are:
await transport.capture_participant_transcription(participant["id"])
await transport.capture_participant_video(video_participant_id, framerate=0)
# Kick off the conversation.
await tts.say("Hi! Ask me about the weather in San Francisco.")
await task.queue_frames([context_aggregator.user().get_context_frame()])
runner = PipelineRunner()

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

@@ -33,13 +33,8 @@ logger.add(sys.stderr, level="DEBUG")
video_participant_id = None
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 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.")
@@ -47,7 +42,12 @@ async def get_weather(function_name, tool_call_id, arguments, llm, context, resu
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
logger.debug(f"!!! IN get_image {video_participant_id}, {arguments}")
question = arguments["question"]
await llm.request_image_frame(user_id=video_participant_id, text_content=question)
await llm.request_image_frame(
user_id=video_participant_id,
function_name=function_name,
tool_call_id=tool_call_id,
text_content=question,
)
async def main():
@@ -72,7 +72,7 @@ async def main():
)
llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
llm.register_function("get_weather", get_weather, start_fetch_weather)
llm.register_function("get_weather", get_weather)
llm.register_function("get_image", get_image)
weather_function = FunctionSchema(

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -65,9 +60,9 @@ async def main():
)
llm = GroqLLMService(api_key=os.getenv("GROQ_API_KEY"), model="llama-3.3-70b-versatile")
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -16,7 +16,6 @@ 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
@@ -31,12 +30,6 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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"})
@@ -63,9 +56,9 @@ async def main():
)
llm = GrokLLMService(api_key=os.getenv("GROK_API_KEY"))
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -67,9 +62,9 @@ async def main():
endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
model=os.getenv("AZURE_CHATGPT_MODEL"),
)
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -64,11 +59,11 @@ async def main():
llm = FireworksLLMService(
api_key=os.getenv("FIREWORKS_API_KEY"),
model="accounts/fireworks/models/firefunction-v2",
model="accounts/fireworks/models/llama-v3p1-405b-instruct",
)
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -66,9 +61,9 @@ async def main():
llm = NimLLMService(
api_key=os.getenv("NVIDIA_API_KEY"), model="meta/llama-3.3-70b-instruct"
)
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -63,9 +58,9 @@ async def main():
)
llm = CerebrasLLMService(api_key=os.getenv("CEREBRAS_API_KEY"), model="llama-3.3-70b")
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -63,9 +58,9 @@ async def main():
)
llm = DeepSeekLLMService(api_key=os.getenv("DEEPSEEK_API_KEY"), model="deepseek-chat")
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -67,9 +62,9 @@ async def main():
llm = OpenRouterLLMService(
api_key=os.getenv("OPENROUTER_API_KEY"), model="openai/gpt-4o-2024-11-20"
)
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -63,11 +58,9 @@ async def main():
)
llm = GoogleLLMOpenAIBetaService(api_key=os.getenv("GEMINI_API_KEY"))
# Register a function_name of None to get all functions
# You can aslo register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(
"get_current_weather", fetch_weather_from_api, start_callback=start_fetch_weather
)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -31,13 +31,8 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -68,11 +63,9 @@ async def main():
project_id="<google-project-id>",
)
)
# Register a function_name of None to get all functions
# You can aslo register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(
"get_current_weather", fetch_weather_from_api, start_callback=start_fetch_weather
)
llm.register_function("get_current_weather", fetch_weather_from_api)
weather_function = FunctionSchema(
name="get_current_weather",

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

@@ -54,7 +54,12 @@ async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context
async def get_image(function_name, tool_call_id, arguments, llm, context, result_callback):
question = arguments["question"]
await llm.request_image_frame(user_id=video_participant_id, text_content=question)
await llm.request_image_frame(
user_id=video_participant_id,
function_name=function_name,
tool_call_id=tool_call_id,
text_content=question,
)
async def get_saved_conversation_filenames(

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

@@ -199,13 +199,8 @@ class OutputGate(FrameProcessor):
break
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -239,9 +234,9 @@ async def main():
# This is the regular LLM.
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
# Register a function_name of None to get all functions
# You can also register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
tools = [
ChatCompletionToolParam(

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

@@ -403,13 +403,8 @@ class OutputGate(FrameProcessor):
break
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 llm.push_frame(TTSSpeakFrame("Let me check on that."))
await result_callback({"conditions": "nice", "temperature": "75"})
@@ -451,7 +446,7 @@ async def main():
)
# Register a function_name of None to get all functions
# sent to the same callback with an additional function_name parameter.
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
tools = [
ChatCompletionToolParam(

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

@@ -30,10 +30,6 @@ logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
async def start_fetch_weather(function_name, llm, context):
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):
# Add a delay to test interruption during function calls
logger.info("Weather API call starting...")
@@ -72,7 +68,7 @@ async def main():
tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
llm.register_function(None, fetch_weather_from_api, start_callback=start_fetch_weather)
llm.register_function("get_current_weather", fetch_weather_from_api)
tools = [
ChatCompletionToolParam(

1
examples/moondream-chatbot/bot.py
View File

@@ -23,7 +23,6 @@ from pipecat.frames.frames import (
OutputImageRawFrame,
SpriteFrame,
TextFrame,
UserImageRawFrame,
UserImageRequestFrame,
)
from pipecat.pipeline.parallel_pipeline import ParallelPipeline

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

@@ -142,7 +142,9 @@ class IntakeProcessor:
]
)
async def start_prescriptions(self, function_name, llm, context):
async def list_prescriptions(
self, function_name, tool_call_id, args, llm, context, result_callback
):
print(f"!!! doing start prescriptions")
# Move on to allergies
context.set_tools(
@@ -182,9 +184,12 @@ class IntakeProcessor:
print(f"!!! about to await llm process frame in start prescrpitions")
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
print(f"!!! past await process frame in start prescriptions")
await self.save_data(args, result_callback)
async def start_allergies(self, function_name, llm, context):
print("!!! doing start allergies")
async def list_allergies(
self, function_name, tool_call_id, args, llm, context, result_callback
):
print("!!! doing list allergies")
# Move on to conditions
context.set_tools(
[
@@ -221,8 +226,11 @@ class IntakeProcessor:
}
)
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await self.save_data(args, result_callback)
async def start_conditions(self, function_name, llm, context):
async def list_conditions(
self, function_name, tool_call_id, args, llm, context, result_callback
):
print("!!! doing start conditions")
# Move on to visit reasons
context.set_tools(
@@ -260,8 +268,11 @@ class IntakeProcessor:
}
)
await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM)
await self.save_data(args, result_callback)
async def start_visit_reasons(self, function_name, llm, context):
async def list_visit_reasons(
self, function_name, tool_call_id, args, llm, context, result_callback
):
print("!!! doing start visit reasons")
# move to finish call
context.set_tools([])
@@ -269,8 +280,9 @@ class IntakeProcessor:
{"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)
async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback):
async def save_data(self, args, result_callback):
logger.info(f"!!! Saving data: {args}")
# Since this is supposed to be "async", returning None from the callback
# will prevent adding anything to context or re-prompting
@@ -319,18 +331,10 @@ async def main():
intake = IntakeProcessor(context)
llm.register_function("verify_birthday", intake.verify_birthday)
llm.register_function(
"list_prescriptions", intake.save_data, start_callback=intake.start_prescriptions
)
llm.register_function(
"list_allergies", intake.save_data, start_callback=intake.start_allergies
)
llm.register_function(
"list_conditions", intake.save_data, start_callback=intake.start_conditions
)
llm.register_function(
"list_visit_reasons", intake.save_data, start_callback=intake.start_visit_reasons
)
llm.register_function("list_prescriptions", intake.list_prescriptions)
llm.register_function("list_allergies", intake.list_allergies)
llm.register_function("list_conditions", intake.list_conditions)
llm.register_function("list_visit_reasons", intake.list_visit_reasons)
fl = FrameLogger("LLM Output")

22
src/pipecat/frames/frames.py
View File

@@ -634,6 +634,15 @@ class FunctionCallInProgressFrame(SystemFrame):
function_name: str
tool_call_id: str
arguments: str
cancel_on_interruption: bool
@dataclass
class FunctionCallCancelFrame(SystemFrame):
"""A frame to signal a function call has been cancelled."""
function_name: str
tool_call_id: str
@dataclass
@@ -653,13 +662,19 @@ class TransportMessageUrgentFrame(SystemFrame):
@dataclass
class UserImageRequestFrame(SystemFrame):
"""A frame user to request an image from the given user."""
"""A frame to request an image from the given user. The frame might be
generated by a function call in which case the corresponding fields will be
properly set.
"""
user_id: str
context: Optional[Any] = None
function_name: Optional[str] = None
tool_call_id: Optional[str] = None
def __str__(self):
return f"{self.name}, user: {self.user_id}"
return f"{self.name}(user: {self.user_id}, function: {self.function_name}, request: {self.tool_call_id})"
@dataclass
@@ -689,10 +704,11 @@ class UserImageRawFrame(InputImageRawFrame):
"""An image associated to a user."""
user_id: str
request: Optional[UserImageRequestFrame] = None
def __str__(self):
pts = format_pts(self.pts)
return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})"
return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format}, request: {self.request})"
@dataclass

2
src/pipecat/pipeline/task.py
View File

@@ -409,7 +409,7 @@ class PipelineTask(BaseTask):
async def _process_push_queue(self):
"""This is the task that runs the pipeline for the first time by sending
a StartFrame and by pushing any other frames queued by the user. It runs
until the tasks is canceled or stopped (e.g. with an EndFrame).
until the tasks is cancelled or stopped (e.g. with an EndFrame).
"""
self._clock.start()

244
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -7,14 +7,20 @@
import asyncio
import time
from abc import abstractmethod
from typing import List
from typing import Dict, List
from loguru import logger
from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
CancelFrame,
EmulateUserStartedSpeakingFrame,
EmulateUserStoppedSpeakingFrame,
EndFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
@@ -23,10 +29,12 @@ from pipecat.frames.frames import (
LLMMessagesUpdateFrame,
LLMSetToolsFrame,
LLMTextFrame,
OpenAILLMContextAssistantTimestampFrame,
StartFrame,
StartInterruptionFrame,
TextFrame,
TranscriptionFrame,
UserImageRawFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
@@ -35,6 +43,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContextFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.time import time_now_iso8601
class LLMFullResponseAggregator(FrameProcessor):
@@ -139,68 +148,20 @@ class BaseLLMResponseAggregator(FrameProcessor):
pass
@abstractmethod
async def push_aggregation(self):
async def handle_aggregation(self, aggregation: str):
"""Adds the given aggregation to the aggregator. The aggregator can use
a simple list of message or a context. It doesn't not push any frames.
"""
pass
class LLMResponseAggregator(BaseLLMResponseAggregator):
"""This is a base LLM aggregator that uses a simple list of messages to
store the conversation. It pushes `LLMMessagesFrame` as an aggregation
frame.
"""
def __init__(
self,
*,
messages: List[dict],
role: str = "user",
**kwargs,
):
super().__init__(**kwargs)
self._messages = messages
self._role = role
self._aggregation = ""
self.reset()
@property
def messages(self) -> List[dict]:
return self._messages
@property
def role(self) -> str:
return self._role
def add_messages(self, messages):
self._messages.extend(messages)
def set_messages(self, messages):
self.reset()
self._messages.clear()
self._messages.extend(messages)
def set_tools(self, tools):
pass
def reset(self):
self._aggregation = ""
@abstractmethod
async def push_aggregation(self):
if len(self._aggregation) > 0:
self._messages.append({"role": self._role, "content": self._aggregation})
"""Pushes the current aggregation. For example, iN the case of context
aggregation this might push a new context frame.
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
frame = LLMMessagesFrame(self._messages)
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
"""
pass
class LLMContextResponseAggregator(BaseLLMResponseAggregator):
@@ -247,20 +208,6 @@ class LLMContextResponseAggregator(BaseLLMResponseAggregator):
def reset(self):
self._aggregation = ""
async def push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self.role, "content": self._aggregation})
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
class LLMUserContextAggregator(LLMContextResponseAggregator):
"""This is a user LLM aggregator that uses an LLM context to store the
@@ -290,12 +237,13 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
self._aggregation_event = asyncio.Event()
self._aggregation_task = None
self.reset()
def reset(self):
super().reset()
self._seen_interim_results = False
async def handle_aggregation(self, aggregation: str):
self._context.add_message({"role": self.role, "content": self._aggregation})
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
@@ -331,6 +279,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
else:
await self.push_frame(frame, direction)
async def push_aggregation(self):
if len(self._aggregation) > 0:
await self.handle_aggregation(self._aggregation)
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
self._aggregation = ""
frame = OpenAILLMContextFrame(self._context)
await self.push_frame(frame)
# Reset our accumulator state.
self.reset()
async def _start(self, frame: StartFrame):
self._create_aggregation_task()
@@ -424,17 +386,29 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
super().__init__(context=context, role="assistant", **kwargs)
self._expect_stripped_words = expect_stripped_words
self._started = False
self._started = 0
self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {}
self.reset()
async def handle_aggregation(self, aggregation: str):
self._context.add_message({"role": "assistant", "content": aggregation})
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
pass
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
pass
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
pass
async def handle_user_image_frame(self, frame: UserImageRawFrame):
pass
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
await self.push_aggregation()
# Reset anyways
self.reset()
await self._handle_interruptions(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
@@ -448,14 +422,116 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
self.set_messages(frame.messages)
elif isinstance(frame, LLMSetToolsFrame):
self.set_tools(frame.tools)
elif isinstance(frame, FunctionCallInProgressFrame):
await self._handle_function_call_in_progress(frame)
elif isinstance(frame, FunctionCallResultFrame):
await self._handle_function_call_result(frame)
elif isinstance(frame, FunctionCallCancelFrame):
await self._handle_function_call_cancel(frame)
elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id:
await self._handle_user_image_frame(frame)
elif isinstance(frame, BotStoppedSpeakingFrame):
await self.push_aggregation()
else:
await self.push_frame(frame, direction)
async def push_aggregation(self):
if not self._aggregation:
return
aggregation = self._aggregation.strip()
self.reset()
if aggregation:
await self.handle_aggregation(aggregation)
# Push context frame
await self.push_context_frame()
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_frame(timestamp_frame)
async def _handle_interruptions(self, frame: StartInterruptionFrame):
await self.push_aggregation()
self._started = 0
self.reset()
async def _handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
logger.debug(
f"{self} FunctionCallInProgressFrame: [{frame.function_name}:{frame.tool_call_id}]"
)
await self.handle_function_call_in_progress(frame)
self._function_calls_in_progress[frame.tool_call_id] = frame
async def _handle_function_call_result(self, frame: FunctionCallResultFrame):
logger.debug(
f"{self} FunctionCallResultFrame: [{frame.function_name}:{frame.tool_call_id}]"
)
if frame.tool_call_id not in self._function_calls_in_progress:
logger.warning(
f"FunctionCallResultFrame tool_call_id [{frame.tool_call_id}] is not running"
)
return
del self._function_calls_in_progress[frame.tool_call_id]
properties = frame.properties
await self.handle_function_call_result(frame)
# Run inference if the function call result requires it.
if frame.result:
run_llm = False
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
run_llm = properties.run_llm
else:
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Emit the on_context_updated callback once the function call
# result is added to the context
if properties and properties.on_context_updated:
await properties.on_context_updated()
async def _handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
logger.debug(
f"{self} FunctionCallCancelFrame: [{frame.function_name}:{frame.tool_call_id}]"
)
if frame.tool_call_id not in self._function_calls_in_progress:
return
if self._function_calls_in_progress[frame.tool_call_id].cancel_on_interruption:
await self.handle_function_call_cancel(frame)
del self._function_calls_in_progress[frame.tool_call_id]
async def _handle_user_image_frame(self, frame: UserImageRawFrame):
logger.debug(
f"{self} UserImageRawFrame: [{frame.request.function_name}:{frame.request.tool_call_id}]"
)
if frame.request.tool_call_id not in self._function_calls_in_progress:
logger.warning(
f"UserImageRawFrame tool_call_id [{frame.request.tool_call_id}] is not running"
)
return
del self._function_calls_in_progress[frame.request.tool_call_id]
await self.handle_user_image_frame(frame)
await self.push_aggregation()
await self.push_context_frame(FrameDirection.UPSTREAM)
async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
self._started = True
self._started += 1
async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
self._started = False
self._started -= 1
await self.push_aggregation()
async def _handle_text(self, frame: TextFrame):
@@ -474,7 +550,7 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
async def push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self.role, "content": self._aggregation})
await self.handle_aggregation(self._aggregation)
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.
@@ -493,7 +569,7 @@ class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
async def push_aggregation(self):
if len(self._aggregation) > 0:
self._context.add_message({"role": self.role, "content": self._aggregation})
await self.handle_aggregation(self._aggregation)
# Reset the aggregation. Reset it before pushing it down, otherwise
# if the tasks gets cancelled we won't be able to clear things up.

66
src/pipecat/processors/aggregators/openai_llm_context.py
View File

@@ -9,9 +9,8 @@ import copy
import io
import json
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, List, Optional
from typing import Any, List, Optional
from loguru import logger
from openai._types import NOT_GIVEN, NotGiven
from openai.types.chat import (
ChatCompletionMessageParam,
@@ -22,12 +21,7 @@ from PIL import Image
from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import (
AudioRawFrame,
Frame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
)
from pipecat.frames.frames import AudioRawFrame, Frame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
# JSON custom encoder to handle bytes arrays so that we can log contexts
@@ -52,7 +46,6 @@ class OpenAILLMContext:
self._messages: List[ChatCompletionMessageParam] = messages if messages else []
self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
self._tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = tools
self._user_image_request_context = {}
self._llm_adapter: Optional[BaseLLMAdapter] = None
def get_llm_adapter(self) -> Optional[BaseLLMAdapter]:
@@ -187,61 +180,6 @@ class OpenAILLMContext:
# todo: implement for OpenAI models and others
pass
async def call_function(
self,
f: Callable[
[str, str, Any, FrameProcessor, "OpenAILLMContext", Callable[[Any], Awaitable[None]]],
Awaitable[None],
],
*,
function_name: str,
tool_call_id: str,
arguments: str,
llm: FrameProcessor,
run_llm: bool = True,
) -> None:
logger.info(f"Calling function {function_name} with arguments {arguments}")
# Push a SystemFrame downstream. This frame will let our assistant context aggregator
# know that we are in the middle of a function call. Some contexts/aggregators may
# not need this. But some definitely do (Anthropic, for example).
# Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
progress_frame_downstream = FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
)
progress_frame_upstream = FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
)
# Push frame both downstream and upstream
await llm.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM)
await llm.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):
result_frame_downstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
result=result,
properties=properties,
)
result_frame_upstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
result=result,
properties=properties,
)
await llm.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
await llm.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback)
def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
# RIFF chunk descriptor
header = bytearray()

219
src/pipecat/services/ai_services.py
View File

@@ -8,7 +8,8 @@ import asyncio
import io
import wave
from abc import abstractmethod
from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Tuple, Type
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Set, Tuple, Type
from loguru import logger
@@ -22,6 +23,9 @@ from pipecat.frames.frames import (
EndFrame,
ErrorFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
StartFrame,
@@ -138,6 +142,13 @@ class AIService(FrameProcessor):
await self.push_frame(f)
@dataclass
class FunctionEntry:
function_name: Optional[str]
callback: Any # TODO(aleix): add proper typing.
cancel_on_interruption: bool
class LLMService(AIService):
"""This class is a no-op but serves as a base class for LLM services."""
@@ -147,38 +158,74 @@ class LLMService(AIService):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._callbacks = {}
self._functions = {}
self._start_callbacks = {}
self._adapter = self.adapter_class()
self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set()
self._register_event_handler("on_completion_timeout")
def get_llm_adapter(self) -> BaseLLMAdapter:
return self._adapter
def create_context_aggregator(
self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
) -> Any:
pass
self._register_event_handler("on_completion_timeout")
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
# TODO-CB: callback function type
def register_function(self, function_name: Optional[str], callback, start_callback=None):
if isinstance(frame, StartInterruptionFrame):
await self._handle_interruptions(frame)
async def _handle_interruptions(self, frame: StartInterruptionFrame):
for function_name, entry in self._functions.items():
if entry.cancel_on_interruption:
await self._cancel_function_call(function_name)
def register_function(
self,
function_name: Optional[str],
callback: 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._callbacks[function_name] = callback
# QUESTION FOR CB: maybe this isn't needed anymore?
self._functions[function_name] = FunctionEntry(
function_name=function_name,
callback=callback,
cancel_on_interruption=cancel_on_interruption,
)
# Start callbacks are now deprecated.
if start_callback:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'start_callback' is deprecated, just put your code on top of the actual function call instead.",
DeprecationWarning,
)
self._start_callbacks[function_name] = start_callback
def unregister_function(self, function_name: Optional[str]):
del self._callbacks[function_name]
del self._functions[function_name]
if self._start_callbacks[function_name]:
del self._start_callbacks[function_name]
def has_function(self, function_name: str):
if None in self._callbacks.keys():
if None in self._functions.keys():
return True
return function_name in self._callbacks.keys()
return function_name in self._functions.keys()
async def call_function(
self,
@@ -188,36 +235,144 @@ class LLMService(AIService):
function_name: str,
arguments: str,
run_llm: bool = True,
) -> None:
f = None
if function_name in self._callbacks.keys():
f = self._callbacks[function_name]
elif None in self._callbacks.keys():
f = self._callbacks[None]
else:
return None
await self.call_start_function(context, function_name)
await context.call_function(
f,
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
llm=self,
run_llm=run_llm,
):
if not function_name in self._functions.keys() and not None in self._functions.keys():
return
task = self.create_task(
self._run_function_call(context, tool_call_id, function_name, arguments, run_llm)
)
# QUESTION FOR CB: maybe this isn't needed anymore?
self._function_call_tasks.add((task, tool_call_id, function_name))
task.add_done_callback(self._function_call_task_finished)
async def call_start_function(self, context: OpenAILLMContext, function_name: str):
if function_name in self._start_callbacks.keys():
await self._start_callbacks[function_name](function_name, self, context)
elif None in self._start_callbacks.keys():
return await self._start_callbacks[None](function_name, self, context)
async def request_image_frame(self, user_id: str, *, text_content: Optional[str] = None):
async def request_image_frame(
self,
user_id: str,
*,
function_name: Optional[str] = None,
tool_call_id: Optional[str] = None,
text_content: Optional[str] = None,
):
await self.push_frame(
UserImageRequestFrame(user_id=user_id, context=text_content), FrameDirection.UPSTREAM
UserImageRequestFrame(
user_id=user_id,
function_name=function_name,
tool_call_id=tool_call_id,
context=text_content,
),
FrameDirection.UPSTREAM,
)
async def _run_function_call(
self,
context: OpenAILLMContext,
tool_call_id: str,
function_name: str,
arguments: str,
run_llm: bool = True,
):
if function_name in self._functions.keys():
entry = self._functions[function_name]
elif None in self._functions.keys():
entry = self._functions[None]
else:
return
logger.debug(
f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}"
)
# NOTE(aleix): This needs to be removed after we remove the deprecation.
await self.call_start_function(context, function_name)
# Push a SystemFrame downstream. This frame will let our assistant context aggregator
# know that we are in the middle of a function call. Some contexts/aggregators may
# not need this. But some definitely do (Anthropic, for example).
# Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
progress_frame_downstream = FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
cancel_on_interruption=entry.cancel_on_interruption,
)
progress_frame_upstream = FunctionCallInProgressFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
cancel_on_interruption=entry.cancel_on_interruption,
)
# Push frame both downstream and upstream
await self.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM)
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):
result_frame_downstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
result=result,
properties=properties,
)
result_frame_upstream = FunctionCallResultFrame(
function_name=function_name,
tool_call_id=tool_call_id,
arguments=arguments,
result=result,
properties=properties,
)
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
)
async def _cancel_function_call(self, function_name: str):
cancelled_tasks = set()
for task, tool_call_id, name in self._function_call_tasks:
if name == function_name:
# We remove the callback because we are going to cancel the task
# now, otherwise we will be removing it from the set while we
# are iterating.
task.remove_done_callback(self._function_call_task_finished)
logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...")
await self.cancel_task(task)
frame = FunctionCallCancelFrame(
function_name=function_name, tool_call_id=tool_call_id
)
await self.push_frame(frame)
logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
cancelled_tasks.add(task)
# Remove all cancelled tasks from our set.
for task in cancelled_tasks:
self._function_call_task_finished(task)
def _function_call_task_finished(self, task: asyncio.Task):
tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None)
if tuple_to_remove:
self._function_call_tasks.discard(tuple_to_remove)
# The task is finished so this should exit immediately. We need to
# do this because otherwise the task manager would have a dangling
# task if we don't remove it.
asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop())
class TTSService(AIService):
def __init__(
@@ -366,12 +521,14 @@ class TTSService(AIService):
else:
await self.push_frame(frame, direction)
elif isinstance(frame, TTSSpeakFrame):
# Store if we were processing text or not so we can set it back.
processing_text = self._processing_text
await self._push_tts_frames(frame.text)
# We pause processing incoming frames because we are sending data to
# the TTS. We pause to avoid audio overlapping.
await self._maybe_pause_frame_processing()
await self.flush_audio()
self._processing_text = False
self._processing_text = processing_text
elif isinstance(frame, TTSUpdateSettingsFrame):
await self._update_settings(frame.settings)
elif isinstance(frame, BotStoppedSpeakingFrame):

212
src/pipecat/services/anthropic.py
View File

@@ -21,19 +21,16 @@ from pydantic import BaseModel, Field
from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
from pipecat.frames.frames import (
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallResultProperties,
LLMEnablePromptCachingFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMTextFrame,
LLMUpdateSettingsFrame,
OpenAILLMContextAssistantTimestampFrame,
StartInterruptionFrame,
UserImageRawFrame,
UserImageRequestFrame,
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
@@ -47,7 +44,6 @@ from pipecat.processors.aggregators.openai_llm_context import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_services import LLMService
from pipecat.utils.time import time_now_iso8601
try:
from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven
@@ -60,13 +56,6 @@ except ModuleNotFoundError as e:
raise Exception(f"Missing module: {e}")
# internal use only -- todo: refactor
@dataclass
class AnthropicImageMessageFrame(Frame):
user_image_raw_frame: UserImageRawFrame
text: Optional[str] = None
@dataclass
class AnthropicContextAggregatorPair:
_user: "AnthropicUserContextAggregator"
@@ -683,42 +672,7 @@ class AnthropicLLMContext(OpenAILLMContext):
class AnthropicUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
super().__init__(context=context, **kwargs)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# Our parent method has already called push_frame(). So we can't interrupt the
# flow here and we don't need to call push_frame() ourselves. Possibly something
# to talk through (tagging @aleix). At some point we might need to refactor these
# context aggregators.
try:
if isinstance(frame, UserImageRequestFrame):
# The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
# that frame so we can use it when we assemble the image message in the assistant
# context aggregator.
if frame.context:
if isinstance(frame.context, str):
self._context._user_image_request_context[frame.user_id] = frame.context
else:
logger.error(
f"Unexpected UserImageRequestFrame context type: {type(frame.context)}"
)
del self._context._user_image_request_context[frame.user_id]
else:
if frame.user_id in self._context._user_image_request_context:
del self._context._user_image_request_context[frame.user_id]
elif isinstance(frame, UserImageRawFrame):
# Push a new AnthropicImageMessageFrame with the text context we cached
# downstream to be handled by our assistant context aggregator. This is
# necessary so that we add the message to the context in the right order.
text = self._context._user_image_request_context.get(frame.user_id) or ""
if text:
del self._context._user_image_request_context[frame.user_id]
frame = AnthropicImageMessageFrame(user_image_raw_frame=frame, text=text)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
pass
#
@@ -732,112 +686,64 @@ class AnthropicUserContextAggregator(LLMUserContextAggregator):
class AnthropicAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, context: OpenAILLMContext | AnthropicLLMContext, **kwargs):
super().__init__(context=context, **kwargs)
self._function_call_in_progress = None
self._function_call_result = None
self._pending_image_frame_message = None
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
assistant_message = {"role": "assistant", "content": []}
assistant_message["content"].append(
{
"type": "tool_use",
"id": frame.tool_call_id,
"name": frame.function_name,
"input": frame.arguments,
}
)
self._context.add_message(assistant_message)
self._context.add_message(
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": frame.tool_call_id,
"content": "IN_PROGRESS",
}
],
}
)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_call_in_progress = None
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
self._function_call_in_progress = frame
elif isinstance(frame, FunctionCallResultFrame):
if (
self._function_call_in_progress
and self._function_call_in_progress.tool_call_id == frame.tool_call_id
):
self._function_call_in_progress = None
self._function_call_result = frame
await self.push_aggregation()
else:
logger.warning(
"FunctionCallResultFrame tool_call_id != InProgressFrame tool_call_id"
)
self._function_call_in_progress = None
self._function_call_result = None
elif isinstance(frame, AnthropicImageMessageFrame):
self._pending_image_frame_message = frame
await self.push_aggregation()
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
if frame.result:
result = json.dumps(frame.result)
await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
else:
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "COMPLETED"
)
async def push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
return
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
async def _update_function_call_result(
self, function_name: str, tool_call_id: str, result: str
):
for message in self._context.messages:
if message["role"] == "user":
for content in message["content"]:
if (
isinstance(content, dict)
and content["type"] == "tool_result"
and content["tool_use_id"] == tool_call_id
):
content["content"] = result
aggregation = self._aggregation.strip()
self.reset()
try:
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
self._function_call_result = None
if frame.result:
assistant_message = {"role": "assistant", "content": []}
assistant_message["content"].append(
{
"type": "tool_use",
"id": frame.tool_call_id,
"name": frame.function_name,
"input": frame.arguments,
}
)
self._context.add_message(assistant_message)
self._context.add_message(
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": frame.tool_call_id,
"content": json.dumps(frame.result),
}
],
}
)
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
run_llm = properties.run_llm
else:
# Default behavior
run_llm = True
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text,
)
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
# Push context frame
await self.push_context_frame()
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_frame(timestamp_frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
async def handle_user_image_frame(self, frame: UserImageRawFrame):
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
self._context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.request.context,
)

9
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -39,6 +39,7 @@ from pipecat.frames.frames import (
TTSStartedFrame,
TTSStoppedFrame,
TTSTextFrame,
UserImageRawFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
@@ -118,10 +119,10 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
async def push_aggregation(self):
# We don't want to store any images in the context. Revisit this later when the API evolves.
self._pending_image_frame_message = None
await super().push_aggregation()
async def handle_user_image_frame(self, frame: UserImageRawFrame):
# We don't want to store any images in the context. Revisit this later
# when the API evolves.
pass
@dataclass

147
src/pipecat/services/google/google.py
View File

@@ -10,6 +10,7 @@ import io
import json
import os
import time
import uuid
from google.api_core.exceptions import DeadlineExceeded
from openai import AsyncStream
@@ -33,20 +34,22 @@ from pipecat.frames.frames import (
EndFrame,
ErrorFrame,
Frame,
FunctionCallResultProperties,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
InterimTranscriptionFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMTextFrame,
LLMUpdateSettingsFrame,
OpenAILLMContextAssistantTimestampFrame,
StartFrame,
TranscriptionFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
URLImageRawFrame,
UserImageRawFrame,
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
@@ -565,91 +568,76 @@ class GoogleUserContextAggregator(OpenAIUserContextAggregator):
class GoogleAssistantContextAggregator(OpenAIAssistantContextAggregator):
async def push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
return
async def handle_aggregation(self, aggregation: str):
self._context.add_message(glm.Content(role="model", parts=[glm.Part(text=aggregation)]))
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
aggregation = self._aggregation.strip()
self.reset()
try:
if aggregation:
self._context.add_message(
glm.Content(role="model", parts=[glm.Part(text=aggregation)])
)
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
self._function_call_result = None
if frame.result:
logger.debug(f"FunctionCallResultFrame result: {frame.arguments}")
self._context.add_message(
glm.Content(
role="model",
parts=[
glm.Part(
function_call=glm.FunctionCall(
name=frame.function_name, args=frame.arguments
)
)
],
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
self._context.add_message(
glm.Content(
role="model",
parts=[
glm.Part(
function_call=glm.FunctionCall(
id=frame.tool_call_id, name=frame.function_name, args=frame.arguments
)
)
response = frame.result
if isinstance(response, str):
response = {"response": response}
self._context.add_message(
glm.Content(
role="user",
parts=[
glm.Part(
function_response=glm.FunctionResponse(
name=frame.function_name, response=response
)
)
],
],
)
)
self._context.add_message(
glm.Content(
role="user",
parts=[
glm.Part(
function_response=glm.FunctionResponse(
id=frame.tool_call_id,
name=frame.function_name,
response={"response": "IN_PROGRESS"},
)
)
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
run_llm = properties.run_llm
else:
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
],
)
)
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text,
)
run_llm = True
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
if frame.result:
if not isinstance(frame.result, str):
return
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
response = {"response": frame.result}
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, response
)
else:
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "COMPLETED"
)
# Push context frame
await self.push_context_frame()
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_frame(timestamp_frame)
async def _update_function_call_result(
self, function_name: str, tool_call_id: str, result: Any
):
for message in self._context.messages:
if message.role == "user":
for part in message.parts:
if part.function_response and part.function_response.id == tool_call_id:
part.function_response.response = {"response": result}
except Exception as e:
logger.exception(f"Error processing frame: {e}")
async def handle_user_image_frame(self, frame: UserImageRawFrame):
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
self._context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.request.context,
)
@dataclass
@@ -1071,7 +1059,7 @@ class GoogleLLMService(LLMService):
args = type(c.function_call).to_dict(c.function_call).get("args", {})
await self.call_function(
context=context,
tool_call_id="what_should_this_be",
tool_call_id=str(uuid.uuid4()),
function_name=c.function_call.name,
arguments=args,
)
@@ -1335,9 +1323,12 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService):
class GoogleVertexLLMService(OpenAILLMService):
"""Implements inference with Google's AI models via Vertex AI while maintaining OpenAI API compatibility.
"""Implements inference with Google's AI models via Vertex AI while
maintaining OpenAI API compatibility.
Reference:
https://cloud.google.com/vertex-ai/generative-ai/docs/multimodal/call-vertex-using-openai-library
"""
class InputParams(OpenAILLMService.InputParams):

85
src/pipecat/services/grok.py
View File

@@ -25,94 +25,15 @@ from pipecat.services.openai import (
)
class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator):
"""Custom assistant context aggregator for Grok that handles empty content requirement."""
async def push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
return
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
aggregation = self._aggregation.strip()
self.reset()
try:
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
self._function_call_result = None
if frame.result:
# Grok requires an empty content field for function calls
self._context.add_message(
{
"role": "assistant",
"content": "", # Required by Grok
"tool_calls": [
{
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments),
},
"type": "function",
}
],
}
)
self._context.add_message(
{
"role": "tool",
"content": json.dumps(frame.result),
"tool_call_id": frame.tool_call_id,
}
)
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
run_llm = properties.run_llm
else:
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text,
)
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
await self.push_context_frame()
except Exception as e:
logger.error(f"Error processing frame: {e}")
@dataclass
class GrokContextAggregatorPair:
_user: "OpenAIUserContextAggregator"
_assistant: "GrokAssistantContextAggregator"
_assistant: "OpenAIAssistantContextAggregator"
def user(self) -> "OpenAIUserContextAggregator":
return self._user
def assistant(self) -> "GrokAssistantContextAggregator":
def assistant(self) -> "OpenAIAssistantContextAggregator":
return self._assistant
@@ -235,5 +156,5 @@ class GrokLLMService(OpenAILLMService):
context.set_llm_adapter(self.get_llm_adapter())
user = OpenAIUserContextAggregator(context, **user_kwargs)
assistant = GrokAssistantContextAggregator(context, **assistant_kwargs)
assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs)
return GrokContextAggregatorPair(_user=user, _assistant=assistant)

207
src/pipecat/services/openai.py
View File

@@ -27,23 +27,20 @@ from pydantic import BaseModel, Field
from pipecat.frames.frames import (
ErrorFrame,
Frame,
FunctionCallCancelFrame,
FunctionCallInProgressFrame,
FunctionCallResultFrame,
FunctionCallResultProperties,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMMessagesFrame,
LLMTextFrame,
LLMUpdateSettingsFrame,
OpenAILLMContextAssistantTimestampFrame,
StartFrame,
StartInterruptionFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
URLImageRawFrame,
UserImageRawFrame,
UserImageRequestFrame,
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
@@ -63,7 +60,6 @@ from pipecat.services.ai_services import (
)
from pipecat.services.base_whisper import BaseWhisperSTTService, Transcription
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
ValidVoice = Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"]
@@ -558,156 +554,67 @@ class OpenAITTSService(TTSService):
logger.exception(f"{self} error generating TTS: {e}")
# internal use only -- todo: refactor
@dataclass
class OpenAIImageMessageFrame(Frame):
user_image_raw_frame: UserImageRawFrame
text: Optional[str] = None
class OpenAIUserContextAggregator(LLMUserContextAggregator):
def __init__(self, context: OpenAILLMContext, **kwargs):
super().__init__(context=context, **kwargs)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# Our parent method has already called push_frame(). So we can't interrupt the
# flow here and we don't need to call push_frame() ourselves.
try:
if isinstance(frame, UserImageRequestFrame):
# The LLM sends a UserImageRequestFrame upstream. Cache any context provided with
# that frame so we can use it when we assemble the image message in the assistant
# context aggregator.
if frame.context:
if isinstance(frame.context, str):
self._context._user_image_request_context[frame.user_id] = frame.context
else:
logger.error(
f"Unexpected UserImageRequestFrame context type: {type(frame.context)}"
)
del self._context._user_image_request_context[frame.user_id]
else:
if frame.user_id in self._context._user_image_request_context:
del self._context._user_image_request_context[frame.user_id]
elif isinstance(frame, UserImageRawFrame):
# Push a new OpenAIImageMessageFrame with the text context we cached
# downstream to be handled by our assistant context aggregator. This is
# necessary so that we add the message to the context in the right order.
text = self._context._user_image_request_context.get(frame.user_id) or ""
if text:
del self._context._user_image_request_context[frame.user_id]
frame = OpenAIImageMessageFrame(user_image_raw_frame=frame, text=text)
await self.push_frame(frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
pass
class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
def __init__(self, context: OpenAILLMContext, **kwargs):
super().__init__(context=context, **kwargs)
self._function_calls_in_progress = {}
self._function_call_result = None
self._pending_image_frame_message = None
async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame):
self._context.add_message(
{
"role": "assistant",
"tool_calls": [
{
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments),
},
"type": "function",
}
],
}
)
self._context.add_message(
{
"role": "tool",
"content": "IN_PROGRESS",
"tool_call_id": frame.tool_call_id,
}
)
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
# See note above about not calling push_frame() here.
if isinstance(frame, StartInterruptionFrame):
self._function_calls_in_progress.clear()
self._function_call_finished = None
elif isinstance(frame, FunctionCallInProgressFrame):
logger.debug(f"FunctionCallInProgressFrame: {frame}")
self._function_calls_in_progress[frame.tool_call_id] = frame
elif isinstance(frame, FunctionCallResultFrame):
logger.debug(f"FunctionCallResultFrame: {frame}")
if frame.tool_call_id in self._function_calls_in_progress:
del self._function_calls_in_progress[frame.tool_call_id]
self._function_call_result = frame
# TODO-CB: Kwin wants us to refactor this out of here but I REFUSE
await self.push_aggregation()
else:
logger.warning(
"FunctionCallResultFrame tool_call_id does not match any function call in progress"
)
self._function_call_result = None
elif isinstance(frame, OpenAIImageMessageFrame):
self._pending_image_frame_message = frame
await self.push_aggregation()
async def handle_function_call_result(self, frame: FunctionCallResultFrame):
if frame.result:
result = json.dumps(frame.result)
await self._update_function_call_result(frame.function_name, frame.tool_call_id, result)
else:
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "COMPLETED"
)
async def push_aggregation(self):
if not (
self._aggregation or self._function_call_result or self._pending_image_frame_message
):
return
async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame):
await self._update_function_call_result(
frame.function_name, frame.tool_call_id, "CANCELLED"
)
run_llm = False
properties: Optional[FunctionCallResultProperties] = None
async def _update_function_call_result(
self, function_name: str, tool_call_id: str, result: str
):
for message in self._context.messages:
if (
message["role"] == "tool"
and message["tool_call_id"]
and message["tool_call_id"] == tool_call_id
):
message["content"] = result
aggregation = self._aggregation.strip()
self.reset()
try:
if aggregation:
self._context.add_message({"role": "assistant", "content": aggregation})
if self._function_call_result:
frame = self._function_call_result
properties = frame.properties
self._function_call_result = None
if frame.result:
self._context.add_message(
{
"role": "assistant",
"tool_calls": [
{
"id": frame.tool_call_id,
"function": {
"name": frame.function_name,
"arguments": json.dumps(frame.arguments),
},
"type": "function",
}
],
}
)
self._context.add_message(
{
"role": "tool",
"content": json.dumps(frame.result),
"tool_call_id": frame.tool_call_id,
}
)
if properties and properties.run_llm is not None:
# If the tool call result has a run_llm property, use it
run_llm = properties.run_llm
else:
# Default behavior is to run the LLM if there are no function calls in progress
run_llm = not bool(self._function_calls_in_progress)
if self._pending_image_frame_message:
frame = self._pending_image_frame_message
self._pending_image_frame_message = None
self._context.add_image_frame_message(
format=frame.user_image_raw_frame.format,
size=frame.user_image_raw_frame.size,
image=frame.user_image_raw_frame.image,
text=frame.text,
)
run_llm = True
if run_llm:
await self.push_context_frame(FrameDirection.UPSTREAM)
# Emit the on_context_updated callback once the function call result is added to the context
if properties and properties.on_context_updated is not None:
await properties.on_context_updated()
# Push context frame
await self.push_context_frame()
# Push timestamp frame with current time
timestamp_frame = OpenAILLMContextAssistantTimestampFrame(timestamp=time_now_iso8601())
await self.push_frame(timestamp_frame)
except Exception as e:
logger.error(f"Error processing frame: {e}")
async def handle_user_image_frame(self, frame: UserImageRawFrame):
await self._update_function_call_result(
frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
)
self._context.add_image_frame_message(
format=frame.format,
size=frame.size,
image=frame.image,
text=frame.request.context,
)

42
src/pipecat/tests/utils.py
View File

@@ -6,7 +6,7 @@
import asyncio
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Dict, Sequence, Tuple
from typing import Any, Awaitable, Callable, Dict, Optional, Sequence, Tuple
from pipecat.frames.frames import (
EndFrame,
@@ -80,8 +80,8 @@ async def run_test(
processor: FrameProcessor,
*,
frames_to_send: Sequence[Frame],
expected_down_frames: Sequence[type],
expected_up_frames: Sequence[type] = [],
expected_down_frames: Optional[Sequence[type]] = None,
expected_up_frames: Optional[Sequence[type]] = None,
ignore_start: bool = True,
start_metadata: Dict[str, Any] = {},
send_end_frame: bool = True,
@@ -126,33 +126,35 @@ async def run_test(
# Down frames
#
received_down_frames: Sequence[Frame] = []
while not received_down.empty():
frame = await received_down.get()
if not isinstance(frame, EndFrame) or not send_end_frame:
received_down_frames.append(frame)
if expected_down_frames is not None:
while not received_down.empty():
frame = await received_down.get()
if not isinstance(frame, EndFrame) or not send_end_frame:
received_down_frames.append(frame)
print("received DOWN frames =", received_down_frames)
print("expected DOWN frames =", expected_down_frames)
print("received DOWN frames =", received_down_frames)
print("expected DOWN frames =", expected_down_frames)
assert len(received_down_frames) == len(expected_down_frames)
assert len(received_down_frames) == len(expected_down_frames)
for real, expected in zip(received_down_frames, expected_down_frames):
assert isinstance(real, expected)
for real, expected in zip(received_down_frames, expected_down_frames):
assert isinstance(real, expected)
#
# Up frames
#
received_up_frames: Sequence[Frame] = []
while not received_up.empty():
frame = await received_up.get()
received_up_frames.append(frame)
if expected_up_frames is not None:
while not received_up.empty():
frame = await received_up.get()
received_up_frames.append(frame)
print("received UP frames =", received_up_frames)
print("expected UP frames =", expected_up_frames)
print("received UP frames =", received_up_frames)
print("expected UP frames =", expected_up_frames)
assert len(received_up_frames) == len(expected_up_frames)
assert len(received_up_frames) == len(expected_up_frames)
for real, expected in zip(received_up_frames, expected_up_frames):
assert isinstance(real, expected)
for real, expected in zip(received_up_frames, expected_up_frames):
assert isinstance(real, expected)
return (received_down_frames, received_up_frames)

21
src/pipecat/transports/services/daily.py
View File

@@ -898,7 +898,7 @@ class DailyInputTransport(BaseInputTransport):
await super().process_frame(frame, direction)
if isinstance(frame, UserImageRequestFrame):
await self.request_participant_image(frame.user_id)
await self.request_participant_image(frame)
#
# Frames
@@ -935,16 +935,16 @@ class DailyInputTransport(BaseInputTransport):
self._video_renderers[participant_id] = {
"framerate": framerate,
"timestamp": 0,
"render_next_frame": False,
"render_next_frame": [],
}
await self._client.capture_participant_video(
participant_id, self._on_participant_video_frame, framerate, video_source, color_format
)
async def request_participant_image(self, participant_id: str):
if participant_id in self._video_renderers:
self._video_renderers[participant_id]["render_next_frame"] = True
async def request_participant_image(self, frame: UserImageRequestFrame):
if frame.user_id in self._video_renderers:
self._video_renderers[frame.user_id]["render_next_frame"].append(frame)
async def _on_participant_video_frame(self, participant_id: str, buffer, size, format):
render_frame = False
@@ -953,17 +953,24 @@ class DailyInputTransport(BaseInputTransport):
prev_time = self._video_renderers[participant_id]["timestamp"]
framerate = self._video_renderers[participant_id]["framerate"]
# Some times we render frames because of a request.
request_frame = None
if framerate > 0:
next_time = prev_time + 1 / framerate
render_frame = (next_time - curr_time) < 0.1
elif self._video_renderers[participant_id]["render_next_frame"]:
self._video_renderers[participant_id]["render_next_frame"] = False
request_frame = self._video_renderers[participant_id]["render_next_frame"].pop(0)
render_frame = True
if render_frame:
frame = UserImageRawFrame(
user_id=participant_id, image=buffer, size=size, format=format
user_id=participant_id,
request=request_frame,
image=buffer,
size=size,
format=format,
)
await self.push_frame(frame)
self._video_renderers[participant_id]["timestamp"] = curr_time

48
src/pipecat/utils/test_frame_processor.py
View File

@@ -1,48 +0,0 @@
from typing import List
from pipecat.processors.frame_processor import FrameProcessor
class TestException(Exception):
pass
class TestFrameProcessor(FrameProcessor):
__test__ = False # Prevents pytest from collecting this class as a test
def __init__(self, test_frames):
self.test_frames = test_frames
self._list_counter = 0
super().__init__()
async def process_frame(self, frame, direction):
await super().process_frame(frame, direction)
if not self.test_frames[
0
]: # then we've run out of required frames but the generator is still going?
raise TestException(f"Oops, got an extra frame, {frame}")
if isinstance(self.test_frames[0], List):
# We need to consume frames until we see the next frame type after this
next_frame = self.test_frames[1]
if isinstance(frame, next_frame):
# we're done iterating the list I guess
print(f"TestFrameProcessor got expected list exit frame: {frame}")
# pop twice to get rid of the list, as well as the next frame
self.test_frames.pop(0)
self.test_frames.pop(0)
self.list_counter = 0
else:
fl = self.test_frames[0]
fl_el = fl[self._list_counter % len(fl)]
if isinstance(frame, fl_el):
print(f"TestFrameProcessor got expected list frame: {frame}")
self._list_counter += 1
else:
raise TestException(f"Inside a list, expected {fl_el} but got {frame}")
else:
if not isinstance(frame, self.test_frames[0]):
raise TestException(f"Expected {self.test_frames[0]}, but got {frame}")
print(f"TestFrameProcessor got expected frame: {frame}")
self.test_frames.pop(0)

33
tests/integration/integration_azure_llm.py
View File

@@ -1,33 +0,0 @@
import asyncio
import os
import unittest
from openai.types.chat import (
ChatCompletionSystemMessageParam,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.services.azure import AzureLLMService
if __name__ == "__main__":
@unittest.skip("Skip azure integration test")
async def test_chat():
llm = AzureLLMService(
api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
model=os.getenv("AZURE_CHATGPT_MODEL"),
)
context = OpenAILLMContext()
message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
content="Please tell the world hello.", name="system", role="system"
)
context.add_message(message)
frame = OpenAILLMContextFrame(context)
async for s in llm.process_frame(frame):
print(s)
asyncio.run(test_chat())

28
tests/integration/integration_ollama_llm.py
View File

@@ -1,28 +0,0 @@
import asyncio
import unittest
from openai.types.chat import (
ChatCompletionSystemMessageParam,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
)
from pipecat.services.ollama import OLLamaLLMService
if __name__ == "__main__":
@unittest.skip("Skip azure integration test")
async def test_chat():
llm = OLLamaLLMService()
context = OpenAILLMContext()
message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
content="Please tell the world hello.", name="system", role="system"
)
context.add_message(message)
frame = OpenAILLMContextFrame(context)
async for s in llm.process_frame(frame):
print(s)
asyncio.run(test_chat())

128
tests/integration/integration_openai_llm.py
View File

@@ -1,128 +0,0 @@
import asyncio
import json
import os
from typing import List
from openai.types.chat import (
ChatCompletionSystemMessageParam,
ChatCompletionToolParam,
ChatCompletionUserMessageParam,
)
from pipecat.frames.frames import LLMFullResponseEndFrame, LLMFullResponseStartFrame, TextFrame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
from pipecat.utils.test_frame_processor import TestFrameProcessor
tools = [
ChatCompletionToolParam(
type="function",
function={
"name": "get_current_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"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 users location.",
},
},
"required": ["location", "format"],
},
},
)
]
if __name__ == "__main__":
async def test_simple_functions():
async def get_weather_from_api(llm, args):
return json.dumps({"conditions": "nice", "temperature": "75"})
api_key = os.getenv("OPENAI_API_KEY")
llm = OpenAILLMService(
api_key=api_key or "",
model="gpt-4-1106-preview",
)
llm.register_function("get_current_weather", get_weather_from_api)
t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame])
llm.link(t)
context = OpenAILLMContext(tools=tools)
system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
content="Ask the user to ask for a weather report", name="system", role="system"
)
user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam(
content="Could you tell me the weather for Boulder, Colorado",
name="user",
role="user",
)
context.add_message(system_message)
context.add_message(user_message)
frame = OpenAILLMContextFrame(context)
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
async def test_advanced_functions():
async def get_weather_from_api(llm, args):
return [
{
"role": "system",
"content": "The user has asked for live weather. Respond by telling them we don't currently support live weather for that area, but it's coming soon.",
}
]
api_key = os.getenv("OPENAI_API_KEY")
llm = OpenAILLMService(
api_key=api_key or "",
model="gpt-4-1106-preview",
)
llm.register_function("get_current_weather", get_weather_from_api)
t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame])
llm.link(t)
context = OpenAILLMContext(tools=tools)
system_message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
content="Ask the user to ask for a weather report", name="system", role="system"
)
user_message: ChatCompletionUserMessageParam = ChatCompletionUserMessageParam(
content="Could you tell me the weather for Boulder, Colorado",
name="user",
role="user",
)
context.add_message(system_message)
context.add_message(user_message)
frame = OpenAILLMContextFrame(context)
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
async def test_chat():
api_key = os.getenv("OPENAI_API_KEY")
t = TestFrameProcessor([LLMFullResponseStartFrame, TextFrame, LLMFullResponseEndFrame])
llm = OpenAILLMService(
api_key=api_key or "",
model="gpt-4o",
)
llm.link(t)
context = OpenAILLMContext()
message: ChatCompletionSystemMessageParam = ChatCompletionSystemMessageParam(
content="Please tell the world hello.", name="system", role="system"
)
context.add_message(message)
frame = OpenAILLMContextFrame(context)
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
async def run_tests():
await test_simple_functions()
await test_advanced_functions()
await test_chat()
asyncio.run(run_tests())

27
tests/integration/test_integration_unified_function_calling.py
View File

@@ -1,3 +1,9 @@
#
# Copyright (c) 2024-2025 Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from unittest.mock import AsyncMock
@@ -6,17 +12,12 @@ from dotenv import load_dotenv
from pipecat.adapters.schemas.function_schema import FunctionSchema
from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.frames.frames import (
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
LLMTextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.pipeline.pipeline import Pipeline
from pipecat.services.ai_services import LLMService
from pipecat.services.anthropic import AnthropicLLMService
from pipecat.services.google import GoogleLLMService
from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
from pipecat.utils.test_frame_processor import TestFrameProcessor
from pipecat.tests.utils import run_test
load_dotenv(override=True)
@@ -47,8 +48,6 @@ async def _test_llm_function_calling(llm: LLMService):
mock_fetch_weather = AsyncMock()
llm.register_function(None, mock_fetch_weather)
t = TestFrameProcessor([LLMFullResponseStartFrame, LLMTextFrame, LLMFullResponseEndFrame])
llm.link(t)
messages = [
{
@@ -61,10 +60,14 @@ async def _test_llm_function_calling(llm: LLMService):
# This is done by default inside the create_context_aggregator
context.set_llm_adapter(llm.get_llm_adapter())
frame = OpenAILLMContextFrame(context)
pipeline = Pipeline([llm])
# This will fail if an exception is raised
await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
frames_to_send = [OpenAILLMContextFrame(context)]
await run_test(
pipeline,
frames_to_send=frames_to_send,
expected_down_frames=None,
)
# Assert that the mock function was called
mock_fetch_weather.assert_called_once()

3
tests/test_context_aggregators.py
View File

@@ -418,7 +418,7 @@ class BaseTestUserContextAggregator:
class BaseTestAssistantContextAggreagator:
CONTEXT_CLASS = None # To be set in subclasses
AGGREGATOR_CLASS = None # To be set in subclasses
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame]
EXPECTED_CONTEXT_FRAMES = None # To be set in subclasses
def check_message_content(self, context: OpenAILLMContext, index: int, content: str):
assert context.messages[index]["content"] == content
@@ -577,6 +577,7 @@ class TestLLMAssistantContextAggregator(
):
CONTEXT_CLASS = OpenAILLMContext
AGGREGATOR_CLASS = LLMAssistantContextAggregator
EXPECTED_CONTEXT_FRAMES = [OpenAILLMContextFrame, OpenAILLMContextAssistantTimestampFrame]
#